Whamcloud - gitweb
LU-1866 lfsck: enhance otable-based iteration
[fs/lustre-release.git] / lustre / ofd / ofd_obd.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2012, Intel Corporation.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lustre/ofd/ofd_obd.c
37  *
38  * Author: Andreas Dilger <adilger@whamcloud.com>
39  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
40  * Author: Mike Pershin <tappro@whamcloud.com>
41  */
42
43 #define DEBUG_SUBSYSTEM S_FILTER
44
45 #include "ofd_internal.h"
46 #include <obd_cksum.h>
47 #include <lustre_quota.h>
48
49 static int ofd_export_stats_init(struct ofd_device *ofd,
50                                  struct obd_export *exp, void *client_nid)
51 {
52         struct obd_device       *obd = ofd_obd(ofd);
53         struct nid_stat         *stats;
54         int                      num_stats, i;
55         int                      rc, newnid = 0;
56
57         ENTRY;
58
59         if (obd_uuid_equals(&exp->exp_client_uuid, &obd->obd_uuid))
60                 /* Self-export gets no proc entry */
61                 RETURN(0);
62
63         rc = lprocfs_exp_setup(exp, client_nid, &newnid);
64         if (rc) {
65                 /* Mask error for already created
66                  * /proc entries */
67                 if (rc == -EALREADY)
68                         rc = 0;
69                 RETURN(rc);
70         }
71
72         if (newnid == 0)
73                 RETURN(0);
74
75         stats = exp->exp_nid_stats;
76         LASSERT(stats != NULL);
77
78         OBD_ALLOC(stats->nid_brw_stats, sizeof(struct brw_stats));
79         if (stats->nid_brw_stats == NULL)
80                 GOTO(clean, rc = -ENOMEM);
81
82         for (i = 0; i < BRW_LAST; i++)
83                 spin_lock_init(&stats->nid_brw_stats->hist[i].oh_lock);
84
85         rc = lprocfs_seq_create(stats->nid_proc, "brw_stats", 0644,
86                                 &ofd_per_nid_stats_fops, stats);
87         if (rc)
88                 CWARN("Error adding the brw_stats file\n");
89
90         num_stats = (sizeof(*obd->obd_type->typ_dt_ops) / sizeof(void *)) +
91                      LPROC_OFD_LAST - 1;
92
93         stats->nid_stats = lprocfs_alloc_stats(num_stats,
94                                                LPROCFS_STATS_FLAG_NOPERCPU);
95         if (stats->nid_stats == NULL)
96                 return -ENOMEM;
97
98         lprocfs_init_ops_stats(LPROC_OFD_LAST, stats->nid_stats);
99         lprocfs_counter_init(stats->nid_stats, LPROC_OFD_READ_BYTES,
100                              LPROCFS_CNTR_AVGMINMAX, "read_bytes", "bytes");
101         lprocfs_counter_init(stats->nid_stats, LPROC_OFD_WRITE_BYTES,
102                              LPROCFS_CNTR_AVGMINMAX, "write_bytes", "bytes");
103
104         rc = lprocfs_register_stats(stats->nid_proc, "stats",
105                                     stats->nid_stats);
106         if (rc)
107                 GOTO(clean, rc);
108
109         rc = lprocfs_nid_ldlm_stats_init(stats);
110         if (rc) {
111                 lprocfs_free_stats(&stats->nid_stats);
112                 GOTO(clean, rc);
113         }
114
115         RETURN(0);
116 clean:
117         return rc;
118 }
119
120 static int ofd_parse_connect_data(const struct lu_env *env,
121                                   struct obd_export *exp,
122                                   struct obd_connect_data *data)
123 {
124         struct ofd_device                *ofd = ofd_exp(exp);
125         struct filter_export_data        *fed = &exp->exp_filter_data;
126
127         if (!data)
128                 RETURN(0);
129
130         CDEBUG(D_RPCTRACE, "%s: cli %s/%p ocd_connect_flags: "LPX64
131                " ocd_version: %x ocd_grant: %d ocd_index: %u"
132                " ocd_group %u\n",
133                exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp,
134                data->ocd_connect_flags, data->ocd_version,
135                data->ocd_grant, data->ocd_index, data->ocd_group);
136
137         if (fed->fed_group != 0 && fed->fed_group != data->ocd_group) {
138                 CWARN("!!! This export (nid %s) used object group %d "
139                       "earlier; now it's trying to use group %d!  This could "
140                       "be a bug in the MDS. Please report to "
141                       "http://bugs.whamcloud.com/\n",
142                       obd_export_nid2str(exp), fed->fed_group,
143                       data->ocd_group);
144                 RETURN(-EPROTO);
145         }
146         fed->fed_group = data->ocd_group;
147
148         data->ocd_connect_flags &= OST_CONNECT_SUPPORTED;
149         exp->exp_connect_data = *data;
150         data->ocd_version = LUSTRE_VERSION_CODE;
151
152         /* Kindly make sure the SKIP_ORPHAN flag is from MDS. */
153         if (data->ocd_connect_flags & OBD_CONNECT_MDS)
154                 CDEBUG(D_HA, "%s: Received MDS connection for group %u\n",
155                        exp->exp_obd->obd_name, data->ocd_group);
156         else if (data->ocd_connect_flags & OBD_CONNECT_SKIP_ORPHAN)
157                 RETURN(-EPROTO);
158
159         if (ofd_grant_param_supp(exp)) {
160                 exp->exp_filter_data.fed_pagesize = data->ocd_blocksize;
161                 /* ocd_{blocksize,inodespace} are log2 values */
162                 data->ocd_blocksize  = ofd->ofd_blockbits;
163                 data->ocd_inodespace = ofd->ofd_dt_conf.ddp_inodespace;
164                 /* ocd_grant_extent is in 1K blocks */
165                 data->ocd_grant_extent = ofd->ofd_dt_conf.ddp_grant_frag >> 10;
166         }
167
168         if (exp_connect_flags(exp) & OBD_CONNECT_GRANT)
169                 data->ocd_grant = ofd_grant_connect(env, exp, data->ocd_grant);
170
171         if (data->ocd_connect_flags & OBD_CONNECT_INDEX) {
172                 struct lr_server_data *lsd = &ofd->ofd_lut.lut_lsd;
173                 int                    index = lsd->lsd_osd_index;
174
175                 if (index != data->ocd_index) {
176                         LCONSOLE_ERROR_MSG(0x136, "Connection from %s to index"
177                                            " %u doesn't match actual OST index"
178                                            " %u in last_rcvd file, bad "
179                                            "configuration?\n",
180                                            obd_export_nid2str(exp), index,
181                                            data->ocd_index);
182                         RETURN(-EBADF);
183                 }
184                 if (!(lsd->lsd_feature_compat & OBD_COMPAT_OST)) {
185                         /* this will only happen on the first connect */
186                         lsd->lsd_feature_compat |= OBD_COMPAT_OST;
187                         /* sync is not needed here as lut_client_add will
188                          * set exp_need_sync flag */
189                         tgt_server_data_update(env, &ofd->ofd_lut, 0);
190                 }
191         }
192         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_SIZE)) {
193                 data->ocd_brw_size = 65536;
194         } else if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) {
195                 data->ocd_brw_size = min(data->ocd_brw_size,
196                                          (__u32)DT_MAX_BRW_SIZE);
197                 if (data->ocd_brw_size == 0) {
198                         CERROR("%s: cli %s/%p ocd_connect_flags: "LPX64
199                                " ocd_version: %x ocd_grant: %d ocd_index: %u "
200                                "ocd_brw_size is unexpectedly zero, "
201                                "network data corruption?"
202                                "Refusing connection of this client\n",
203                                exp->exp_obd->obd_name,
204                                exp->exp_client_uuid.uuid,
205                                exp, data->ocd_connect_flags, data->ocd_version,
206                                data->ocd_grant, data->ocd_index);
207                         RETURN(-EPROTO);
208                 }
209         }
210
211         if (data->ocd_connect_flags & OBD_CONNECT_CKSUM) {
212                 __u32 cksum_types = data->ocd_cksum_types;
213
214                 /* The client set in ocd_cksum_types the checksum types it
215                  * supports. We have to mask off the algorithms that we don't
216                  * support */
217                 data->ocd_cksum_types &= cksum_types_supported_server();
218
219                 if (unlikely(data->ocd_cksum_types == 0)) {
220                         CERROR("%s: Connect with checksum support but no "
221                                "ocd_cksum_types is set\n",
222                                exp->exp_obd->obd_name);
223                         RETURN(-EPROTO);
224                 }
225
226                 CDEBUG(D_RPCTRACE, "%s: cli %s supports cksum type %x, return "
227                        "%x\n", exp->exp_obd->obd_name, obd_export_nid2str(exp),
228                        cksum_types, data->ocd_cksum_types);
229         } else {
230                 /* This client does not support OBD_CONNECT_CKSUM
231                  * fall back to CRC32 */
232                 CDEBUG(D_RPCTRACE, "%s: cli %s does not support "
233                        "OBD_CONNECT_CKSUM, CRC32 will be used\n",
234                        exp->exp_obd->obd_name, obd_export_nid2str(exp));
235         }
236
237         if (data->ocd_connect_flags & OBD_CONNECT_MAXBYTES)
238                 data->ocd_maxbytes = ofd->ofd_dt_conf.ddp_maxbytes;
239
240         RETURN(0);
241 }
242
243 static int ofd_obd_reconnect(const struct lu_env *env, struct obd_export *exp,
244                              struct obd_device *obd, struct obd_uuid *cluuid,
245                              struct obd_connect_data *data, void *localdata)
246 {
247         struct ofd_device       *ofd;
248         int                      rc;
249
250         ENTRY;
251
252         if (exp == NULL || obd == NULL || cluuid == NULL)
253                 RETURN(-EINVAL);
254
255         ofd = ofd_dev(obd->obd_lu_dev);
256
257         rc = lu_env_refill((struct lu_env *)env);
258         if (rc != 0) {
259                 CERROR("Failure to refill session: '%d'\n", rc);
260                 RETURN(rc);
261         }
262
263         ofd_info_init(env, exp);
264         rc = ofd_parse_connect_data(env, exp, data);
265         if (rc == 0)
266                 ofd_export_stats_init(ofd, exp, localdata);
267
268         RETURN(rc);
269 }
270
271 static int ofd_obd_connect(const struct lu_env *env, struct obd_export **_exp,
272                            struct obd_device *obd, struct obd_uuid *cluuid,
273                            struct obd_connect_data *data, void *localdata)
274 {
275         struct obd_export       *exp;
276         struct ofd_device       *ofd;
277         struct lustre_handle     conn = { 0 };
278         int                      rc;
279         ENTRY;
280
281         if (_exp == NULL || obd == NULL || cluuid == NULL)
282                 RETURN(-EINVAL);
283
284         ofd = ofd_dev(obd->obd_lu_dev);
285
286         rc = class_connect(&conn, obd, cluuid);
287         if (rc)
288                 RETURN(rc);
289
290         exp = class_conn2export(&conn);
291         LASSERT(exp != NULL);
292
293         rc = lu_env_refill((struct lu_env *)env);
294         if (rc != 0) {
295                 CERROR("Failure to refill session: '%d'\n", rc);
296                 GOTO(out, rc);
297         }
298
299         ofd_info_init(env, exp);
300
301         rc = ofd_parse_connect_data(env, exp, data);
302         if (rc)
303                 GOTO(out, rc);
304
305         if (obd->obd_replayable) {
306                 struct tg_export_data *ted = &exp->exp_target_data;
307
308                 memcpy(ted->ted_lcd->lcd_uuid, cluuid,
309                        sizeof(ted->ted_lcd->lcd_uuid));
310                 rc = tgt_client_new(env, exp);
311                 if (rc != 0)
312                         GOTO(out, rc);
313                 ofd_export_stats_init(ofd, exp, localdata);
314         }
315
316         CDEBUG(D_HA, "%s: get connection from MDS %d\n", obd->obd_name,
317                data->ocd_group);
318
319 out:
320         if (rc != 0) {
321                 class_disconnect(exp);
322                 *_exp = NULL;
323         } else {
324                 *_exp = exp;
325         }
326         RETURN(rc);
327 }
328
329 static int ofd_obd_disconnect(struct obd_export *exp)
330 {
331         struct ofd_device       *ofd = ofd_dev(exp->exp_obd->obd_lu_dev);
332         struct lu_env            env;
333         int                      rc;
334
335         ENTRY;
336
337         LASSERT(exp);
338         class_export_get(exp);
339
340         if (!(exp->exp_flags & OBD_OPT_FORCE))
341                 ofd_grant_sanity_check(ofd_obd(ofd), __FUNCTION__);
342
343         rc = server_disconnect_export(exp);
344
345         ofd_grant_discard(exp);
346
347         rc = lu_env_init(&env, LCT_DT_THREAD);
348         if (rc)
349                 RETURN(rc);
350
351         /* Do not erase record for recoverable client. */
352         if (exp->exp_obd->obd_replayable &&
353             (!exp->exp_obd->obd_fail || exp->exp_failed))
354                 tgt_client_del(&env, exp);
355         lu_env_fini(&env);
356
357         class_export_put(exp);
358         RETURN(rc);
359 }
360
361 static int ofd_init_export(struct obd_export *exp)
362 {
363         int rc;
364
365         spin_lock_init(&exp->exp_filter_data.fed_lock);
366         CFS_INIT_LIST_HEAD(&exp->exp_filter_data.fed_mod_list);
367         spin_lock(&exp->exp_lock);
368         exp->exp_connecting = 1;
369         spin_unlock(&exp->exp_lock);
370
371         /* self-export doesn't need client data and ldlm initialization */
372         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
373                                      &exp->exp_client_uuid)))
374                 return 0;
375
376         rc = tgt_client_alloc(exp);
377         if (rc == 0)
378                 ldlm_init_export(exp);
379         if (rc)
380                 CERROR("%s: Can't initialize export: rc %d\n",
381                        exp->exp_obd->obd_name, rc);
382         return rc;
383 }
384
385 static int ofd_destroy_export(struct obd_export *exp)
386 {
387         struct ofd_device *ofd = ofd_dev(exp->exp_obd->obd_lu_dev);
388
389         if (exp->exp_filter_data.fed_pending)
390                 CERROR("%s: cli %s/%p has %lu pending on destroyed export"
391                        "\n", exp->exp_obd->obd_name, exp->exp_client_uuid.uuid,
392                        exp, exp->exp_filter_data.fed_pending);
393
394         target_destroy_export(exp);
395
396         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
397                                      &exp->exp_client_uuid)))
398                 return 0;
399
400         ldlm_destroy_export(exp);
401         tgt_client_free(exp);
402
403         ofd_fmd_cleanup(exp);
404
405         /*
406          * discard grants once we're sure no more
407          * interaction with the client is possible
408          */
409         ofd_grant_discard(exp);
410         ofd_fmd_cleanup(exp);
411
412         if (exp_connect_flags(exp) & OBD_CONNECT_GRANT_SHRINK) {
413                 if (ofd->ofd_tot_granted_clients > 0)
414                         ofd->ofd_tot_granted_clients --;
415         }
416
417         if (!(exp->exp_flags & OBD_OPT_FORCE))
418                 ofd_grant_sanity_check(exp->exp_obd, __FUNCTION__);
419
420         LASSERT(cfs_list_empty(&exp->exp_filter_data.fed_mod_list));
421         return 0;
422 }
423
424 int ofd_postrecov(const struct lu_env *env, struct ofd_device *ofd)
425 {
426         struct lu_device *ldev = &ofd->ofd_dt_dev.dd_lu_dev;
427
428         CDEBUG(D_HA, "%s: recovery is over\n", ofd_obd(ofd)->obd_name);
429         return ldev->ld_ops->ldo_recovery_complete(env, ldev);
430 }
431
432 int ofd_obd_postrecov(struct obd_device *obd)
433 {
434         struct lu_env            env;
435         struct lu_device        *ldev = obd->obd_lu_dev;
436         int                      rc;
437
438         ENTRY;
439
440         rc = lu_env_init(&env, LCT_DT_THREAD);
441         if (rc)
442                 RETURN(rc);
443         ofd_info_init(&env, obd->obd_self_export);
444
445         rc = ofd_postrecov(&env, ofd_dev(ldev));
446
447         lu_env_fini(&env);
448         RETURN(rc);
449 }
450
451 static int ofd_adapt_sptlrpc_conf(const struct lu_env *env,
452                                   struct obd_device *obd, int initial)
453 {
454         struct filter_obd       *fo = &obd->u.filter;
455         struct sptlrpc_rule_set  tmp_rset;
456         int                      rc;
457
458         sptlrpc_rule_set_init(&tmp_rset);
459         rc = sptlrpc_conf_target_get_rules(obd, &tmp_rset, initial);
460         if (rc) {
461                 CERROR("%s: failed get sptlrpc rules: rc = %d\n",
462                        obd->obd_name, rc);
463                 return rc;
464         }
465
466         sptlrpc_target_update_exp_flavor(obd, &tmp_rset);
467
468         write_lock(&fo->fo_sptlrpc_lock);
469         sptlrpc_rule_set_free(&fo->fo_sptlrpc_rset);
470         fo->fo_sptlrpc_rset = tmp_rset;
471         write_unlock(&fo->fo_sptlrpc_lock);
472
473         return 0;
474 }
475
476 static int ofd_set_mds_conn(struct obd_export *exp, void *val)
477 {
478         int rc = 0;
479
480         ENTRY;
481
482         LCONSOLE_WARN("%s: received MDS connection from %s\n",
483                       exp->exp_obd->obd_name, obd_export_nid2str(exp));
484         RETURN(rc);
485 }
486
487 static int ofd_set_info_async(const struct lu_env *env, struct obd_export *exp,
488                               __u32 keylen, void *key, __u32 vallen, void *val,
489                               struct ptlrpc_request_set *set)
490 {
491         struct ofd_device       *ofd;
492         int                      rc = 0;
493
494         ENTRY;
495
496         if (exp->exp_obd == NULL) {
497                 CDEBUG(D_IOCTL, "invalid export %p\n", exp);
498                 RETURN(-EINVAL);
499         }
500
501         ofd = ofd_exp(exp);
502
503         if (KEY_IS(KEY_CAPA_KEY)) {
504                 rc = ofd_update_capa_key(ofd, val);
505                 if (rc)
506                         CERROR("%s: update capability key failed: rc = %d\n",
507                                exp->exp_obd->obd_name, rc);
508         } else if (KEY_IS(KEY_SPTLRPC_CONF)) {
509                 ofd_adapt_sptlrpc_conf(env, exp->exp_obd, 0);
510         } else if (KEY_IS(KEY_MDS_CONN)) {
511                 rc = ofd_set_mds_conn(exp, val);
512         } else if (KEY_IS(KEY_GRANT_SHRINK)) {
513                 struct ost_body *body = val;
514
515                 ofd_info_init(env, exp);
516                 /** handle grant shrink, similar to a read request */
517                 ofd_grant_prepare_read(env, exp, &body->oa);
518         } else {
519                 CERROR("%s: Unsupported key %s\n",
520                        exp->exp_obd->obd_name, (char*)key);
521                 rc = -EOPNOTSUPP;
522         }
523
524         RETURN(rc);
525 }
526
527 static int ofd_get_info(const struct lu_env *env, struct obd_export *exp,
528                         __u32 keylen, void *key, __u32 *vallen, void *val,
529                         struct lov_stripe_md *lsm)
530 {
531         struct ofd_device       *ofd;
532         int                      rc = 0;
533
534         ENTRY;
535
536         if (exp->exp_obd == NULL) {
537                 CDEBUG(D_IOCTL, "invalid client export %p\n", exp);
538                 RETURN(-EINVAL);
539         }
540
541         ofd = ofd_exp(exp);
542
543         if (KEY_IS(KEY_BLOCKSIZE)) {
544                 __u32 *blocksize = val;
545                 if (blocksize) {
546                         if (*vallen < sizeof(*blocksize))
547                                 RETURN(-EOVERFLOW);
548                         *blocksize = 1 << ofd->ofd_dt_conf.ddp_block_shift;
549                 }
550                 *vallen = sizeof(*blocksize);
551         } else if (KEY_IS(KEY_BLOCKSIZE_BITS)) {
552                 __u32 *blocksize_bits = val;
553                 if (blocksize_bits) {
554                         if (*vallen < sizeof(*blocksize_bits))
555                                 RETURN(-EOVERFLOW);
556                         *blocksize_bits = ofd->ofd_dt_conf.ddp_block_shift;
557                 }
558                 *vallen = sizeof(*blocksize_bits);
559         } else if (KEY_IS(KEY_LAST_ID)) {
560                 obd_id *last_id = val;
561                 struct ofd_seq *oseq;
562
563                 if (val == NULL) {
564                         *vallen = sizeof(obd_id);
565                         RETURN(0);
566                 }
567                 ofd_info_init(env, exp);
568                 oseq = ofd_seq_load(env, ofd,
569                                     (obd_seq)exp->exp_filter_data.fed_group);
570                 LASSERT(oseq != NULL);
571                 if (last_id) {
572                         if (*vallen < sizeof(*last_id)) {
573                                 ofd_seq_put(env, oseq);
574                                 RETURN(-EOVERFLOW);
575                         }
576                         *last_id = ofd_seq_last_oid(oseq);
577                 }
578                 ofd_seq_put(env, oseq);
579                 *vallen = sizeof(*last_id);
580         } else if (KEY_IS(KEY_FIEMAP)) {
581                 struct ofd_thread_info          *info;
582                 struct ofd_device               *ofd = ofd_exp(exp);
583                 struct ofd_object               *fo;
584                 struct ll_fiemap_info_key       *fm_key = key;
585
586                 if (val == NULL) {
587                         *vallen = fiemap_count_to_size(
588                                                fm_key->fiemap.fm_extent_count);
589                         RETURN(0);
590                 }
591
592                 info = ofd_info_init(env, exp);
593
594                 fid_ostid_unpack(&info->fti_fid, &fm_key->oa.o_oi, 0);
595
596                 CDEBUG(D_INODE, "get FIEMAP of object "DFID"\n",
597                        PFID(&info->fti_fid));
598
599                 fo = ofd_object_find(env, ofd, &info->fti_fid);
600                 if (IS_ERR(fo)) {
601                         CERROR("%s: error finding object "DFID"\n",
602                                exp->exp_obd->obd_name, PFID(&info->fti_fid));
603                         rc = PTR_ERR(fo);
604                 } else {
605                         struct ll_user_fiemap *fiemap = val;
606
607                         ofd_read_lock(env, fo);
608                         if (ofd_object_exists(fo)) {
609                                 *fiemap = fm_key->fiemap;
610                                 rc = dt_fiemap_get(env,
611                                                    ofd_object_child(fo),
612                                                    fiemap);
613                         } else {
614                                 rc = -ENOENT;
615                         }
616                         ofd_read_unlock(env, fo);
617                         ofd_object_put(env, fo);
618                 }
619         } else if (KEY_IS(KEY_SYNC_LOCK_CANCEL)) {
620                 *((__u32 *) val) = ofd->ofd_sync_lock_cancel;
621                 *vallen = sizeof(__u32);
622         } else if (KEY_IS(KEY_LAST_FID)) {
623                 struct lu_env      env;
624                 struct ofd_device *ofd = ofd_exp(exp);
625                 struct ofd_seq    *oseq;
626                 struct lu_fid     *last_fid = val;
627                 int             rc;
628
629                 if (last_fid == NULL) {
630                         *vallen = sizeof(struct lu_fid);
631                         RETURN(0);
632                 }
633
634                 if (*vallen < sizeof(*last_fid))
635                         RETURN(-EOVERFLOW);
636
637                 rc = lu_env_init(&env, LCT_DT_THREAD);
638                 if (rc != 0)
639                         RETURN(rc);
640                 ofd_info_init(&env, exp);
641                 fid_le_to_cpu(last_fid, last_fid);
642                 oseq = ofd_seq_load(&env, ofd, fid_seq(last_fid));
643                 if (IS_ERR(oseq))
644                         GOTO(out_fid, rc = PTR_ERR(oseq));
645
646                 last_fid->f_seq = oseq->os_seq;
647                 last_fid->f_oid = oseq->os_last_oid;
648                 fid_cpu_to_le(last_fid, last_fid);
649
650                 *vallen = sizeof(*last_fid);
651                 ofd_seq_put(&env, oseq);
652 out_fid:
653                 lu_env_fini(&env);
654         } else {
655                 CERROR("Not supported key %s\n", (char*)key);
656                 rc = -EOPNOTSUPP;
657         }
658
659         RETURN(rc);
660 }
661
662 /** helper function for statfs, also used by grant code */
663 int ofd_statfs_internal(const struct lu_env *env, struct ofd_device *ofd,
664                         struct obd_statfs *osfs, __u64 max_age, int *from_cache)
665 {
666         int rc;
667
668         spin_lock(&ofd->ofd_osfs_lock);
669         if (cfs_time_before_64(ofd->ofd_osfs_age, max_age) || max_age == 0) {
670                 obd_size unstable;
671
672                 /* statfs data are too old, get up-to-date one.
673                  * we must be cautious here since multiple threads might be
674                  * willing to update statfs data concurrently and we must
675                  * grant that cached statfs data are always consistent */
676
677                 if (ofd->ofd_statfs_inflight == 0)
678                         /* clear inflight counter if no users, although it would
679                          * take a while to overflow this 64-bit counter ... */
680                         ofd->ofd_osfs_inflight = 0;
681                 /* notify ofd_grant_commit() that we want to track writes
682                  * completed as of now */
683                 ofd->ofd_statfs_inflight++;
684                 /* record value of inflight counter before running statfs to
685                  * compute the diff once statfs is completed */
686                 unstable = ofd->ofd_osfs_inflight;
687                 spin_unlock(&ofd->ofd_osfs_lock);
688
689                 /* statfs can sleep ... hopefully not for too long since we can
690                  * call it fairly often as space fills up */
691                 rc = dt_statfs(env, ofd->ofd_osd, osfs);
692                 if (unlikely(rc))
693                         return rc;
694
695                 spin_lock(&ofd->ofd_grant_lock);
696                 spin_lock(&ofd->ofd_osfs_lock);
697                 /* calculate how much space was written while we released the
698                  * ofd_osfs_lock */
699                 unstable = ofd->ofd_osfs_inflight - unstable;
700                 ofd->ofd_osfs_unstable = 0;
701                 if (unstable) {
702                         /* some writes completed while we were running statfs
703                          * w/o the ofd_osfs_lock. Those ones got added to
704                          * the cached statfs data that we are about to crunch.
705                          * Take them into account in the new statfs data */
706                         osfs->os_bavail -= min_t(obd_size, osfs->os_bavail,
707                                                unstable >> ofd->ofd_blockbits);
708                         /* However, we don't really know if those writes got
709                          * accounted in the statfs call, so tell
710                          * ofd_grant_space_left() there is some uncertainty
711                          * on the accounting of those writes.
712                          * The purpose is to prevent spurious error messages in
713                          * ofd_grant_space_left() since those writes might be
714                          * accounted twice. */
715                         ofd->ofd_osfs_unstable += unstable;
716                 }
717                 /* similarly, there is some uncertainty on write requests
718                  * between prepare & commit */
719                 ofd->ofd_osfs_unstable += ofd->ofd_tot_pending;
720                 spin_unlock(&ofd->ofd_grant_lock);
721
722                 /* finally udpate cached statfs data */
723                 ofd->ofd_osfs = *osfs;
724                 ofd->ofd_osfs_age = cfs_time_current_64();
725
726                 ofd->ofd_statfs_inflight--; /* stop tracking */
727                 if (ofd->ofd_statfs_inflight == 0)
728                         ofd->ofd_osfs_inflight = 0;
729                 spin_unlock(&ofd->ofd_osfs_lock);
730
731                 if (from_cache)
732                         *from_cache = 0;
733         } else {
734                 /* use cached statfs data */
735                 *osfs = ofd->ofd_osfs;
736                 spin_unlock(&ofd->ofd_osfs_lock);
737                 if (from_cache)
738                         *from_cache = 1;
739         }
740         return 0;
741 }
742
743 static int ofd_statfs(const struct lu_env *env,  struct obd_export *exp,
744                       struct obd_statfs *osfs, __u64 max_age, __u32 flags)
745 {
746         struct obd_device       *obd = class_exp2obd(exp);
747         struct ofd_device       *ofd = ofd_dev(exp->exp_obd->obd_lu_dev);
748         int                      rc;
749
750         ENTRY;
751
752         rc = ofd_statfs_internal(env, ofd, osfs, max_age, NULL);
753         if (unlikely(rc))
754                 GOTO(out, rc);
755
756         /* at least try to account for cached pages.  its still racy and
757          * might be under-reporting if clients haven't announced their
758          * caches with brw recently */
759
760         CDEBUG(D_SUPER | D_CACHE, "blocks cached "LPU64" granted "LPU64
761                " pending "LPU64" free "LPU64" avail "LPU64"\n",
762                ofd->ofd_tot_dirty, ofd->ofd_tot_granted, ofd->ofd_tot_pending,
763                osfs->os_bfree << ofd->ofd_blockbits,
764                osfs->os_bavail << ofd->ofd_blockbits);
765
766         osfs->os_bavail -= min_t(obd_size, osfs->os_bavail,
767                                  ((ofd->ofd_tot_dirty + ofd->ofd_tot_pending +
768                                    osfs->os_bsize - 1) >> ofd->ofd_blockbits));
769
770         /* The QoS code on the MDS does not care about space reserved for
771          * precreate, so take it out. */
772         if (exp_connect_flags(exp) & OBD_CONNECT_MDS) {
773                 struct filter_export_data *fed;
774
775                 fed = &obd->obd_self_export->exp_filter_data;
776                 osfs->os_bavail -= min_t(obd_size, osfs->os_bavail,
777                                          fed->fed_grant >> ofd->ofd_blockbits);
778         }
779
780         ofd_grant_sanity_check(obd, __FUNCTION__);
781         CDEBUG(D_CACHE, LPU64" blocks: "LPU64" free, "LPU64" avail; "
782                LPU64" objects: "LPU64" free; state %x\n",
783                osfs->os_blocks, osfs->os_bfree, osfs->os_bavail,
784                osfs->os_files, osfs->os_ffree, osfs->os_state);
785
786         if (OBD_FAIL_CHECK_VALUE(OBD_FAIL_OST_ENOINO,
787                                  ofd->ofd_lut.lut_lsd.lsd_osd_index))
788                 osfs->os_ffree = 0;
789
790         /* OS_STATE_READONLY can be set by OSD already */
791         if (ofd->ofd_raid_degraded)
792                 osfs->os_state |= OS_STATE_DEGRADED;
793
794         if (obd->obd_self_export != exp && ofd_grant_compat(exp, ofd)) {
795                 /* clients which don't support OBD_CONNECT_GRANT_PARAM
796                  * should not see a block size > page size, otherwise
797                  * cl_lost_grant goes mad. Therefore, we emulate a 4KB (=2^12)
798                  * block size which is the biggest block size known to work
799                  * with all client's page size. */
800                 osfs->os_blocks <<= ofd->ofd_blockbits - COMPAT_BSIZE_SHIFT;
801                 osfs->os_bfree  <<= ofd->ofd_blockbits - COMPAT_BSIZE_SHIFT;
802                 osfs->os_bavail <<= ofd->ofd_blockbits - COMPAT_BSIZE_SHIFT;
803                 osfs->os_bsize    = 1 << COMPAT_BSIZE_SHIFT;
804         }
805
806         if (OBD_FAIL_CHECK_VALUE(OBD_FAIL_OST_ENOSPC,
807                                  ofd->ofd_lut.lut_lsd.lsd_osd_index))
808                 osfs->os_bfree = osfs->os_bavail = 2;
809
810         EXIT;
811 out:
812         return rc;
813 }
814
815 int ofd_setattr(const struct lu_env *env, struct obd_export *exp,
816                 struct obd_info *oinfo, struct obd_trans_info *oti)
817 {
818         struct ofd_thread_info  *info;
819         struct ofd_device       *ofd = ofd_exp(exp);
820         struct ldlm_namespace   *ns = ofd->ofd_namespace;
821         struct ldlm_resource    *res;
822         struct ofd_object       *fo;
823         struct obdo             *oa = oinfo->oi_oa;
824         struct filter_fid       *ff = NULL;
825         int                      rc = 0;
826
827         ENTRY;
828
829         info = ofd_info_init(env, exp);
830         ofd_oti2info(info, oti);
831
832         fid_ostid_unpack(&info->fti_fid, &oinfo->oi_oa->o_oi, 0);
833         ofd_build_resid(&info->fti_fid, &info->fti_resid);
834
835         rc = ofd_auth_capa(exp, &info->fti_fid, oa->o_seq,
836                            oinfo_capa(oinfo), CAPA_OPC_META_WRITE);
837         if (rc)
838                 GOTO(out, rc);
839
840         /* This would be very bad - accidentally truncating a file when
841          * changing the time or similar - bug 12203. */
842         if (oinfo->oi_oa->o_valid & OBD_MD_FLSIZE &&
843             oinfo->oi_policy.l_extent.end != OBD_OBJECT_EOF) {
844                 static char mdsinum[48];
845
846                 if (oinfo->oi_oa->o_valid & OBD_MD_FLFID)
847                         snprintf(mdsinum, sizeof(mdsinum) - 1,
848                                  "of parent "DFID, oinfo->oi_oa->o_parent_seq,
849                                  oinfo->oi_oa->o_parent_oid, 0);
850                 else
851                         mdsinum[0] = '\0';
852
853                 CERROR("%s: setattr from %s trying to truncate object "DFID
854                        " %s\n", exp->exp_obd->obd_name,
855                        obd_export_nid2str(exp), PFID(&info->fti_fid), mdsinum);
856                 GOTO(out, rc = -EPERM);
857         }
858
859         fo = ofd_object_find(env, ofd, &info->fti_fid);
860         if (IS_ERR(fo)) {
861                 CERROR("%s: can't find object "DFID"\n",
862                        exp->exp_obd->obd_name, PFID(&info->fti_fid));
863                 GOTO(out, rc = PTR_ERR(fo));
864         }
865
866         la_from_obdo(&info->fti_attr, oinfo->oi_oa, oinfo->oi_oa->o_valid);
867         info->fti_attr.la_valid &= ~LA_TYPE;
868
869         if (oa->o_valid & OBD_MD_FLFID) {
870                 ff = &info->fti_mds_fid;
871                 ofd_prepare_fidea(ff, oa);
872         }
873
874         /* setting objects attributes (including owner/group) */
875         rc = ofd_attr_set(env, fo, &info->fti_attr, ff);
876         if (rc)
877                 GOTO(out_unlock, rc);
878
879         res = ldlm_resource_get(ns, NULL, &info->fti_resid, LDLM_EXTENT, 0);
880         if (res != NULL) {
881                 ldlm_res_lvbo_update(res, NULL, 0);
882                 ldlm_resource_putref(res);
883         }
884
885         oinfo->oi_oa->o_valid = OBD_MD_FLID;
886
887         /* Quota release needs uid/gid info */
888         rc = ofd_attr_get(env, fo, &info->fti_attr);
889         obdo_from_la(oinfo->oi_oa, &info->fti_attr,
890                      OFD_VALID_FLAGS | LA_UID | LA_GID);
891         ofd_info2oti(info, oti);
892
893         ofd_counter_incr(exp, LPROC_OFD_STATS_SETATTR, oti->oti_jobid, 1);
894         EXIT;
895 out_unlock:
896         ofd_object_put(env, fo);
897 out:
898         return rc;
899 }
900
901 static int ofd_punch(const struct lu_env *env, struct obd_export *exp,
902                      struct obd_info *oinfo, struct obd_trans_info *oti,
903                      struct ptlrpc_request_set *rqset)
904 {
905         struct ofd_thread_info  *info;
906         struct ofd_device       *ofd = ofd_exp(exp);
907         struct ldlm_namespace   *ns = ofd->ofd_namespace;
908         struct ldlm_resource    *res;
909         struct ofd_object       *fo;
910         struct filter_fid       *ff = NULL;
911         int                      rc = 0;
912
913         ENTRY;
914
915         info = ofd_info_init(env, exp);
916         ofd_oti2info(info, oti);
917
918         fid_ostid_unpack(&info->fti_fid, &oinfo->oi_oa->o_oi, 0);
919         ofd_build_resid(&info->fti_fid, &info->fti_resid);
920
921         CDEBUG(D_INODE, "calling punch for object "DFID", valid = "LPX64
922                ", start = "LPD64", end = "LPD64"\n", PFID(&info->fti_fid),
923                oinfo->oi_oa->o_valid, oinfo->oi_policy.l_extent.start,
924                oinfo->oi_policy.l_extent.end);
925
926         rc = ofd_auth_capa(exp, &info->fti_fid, oinfo->oi_oa->o_seq,
927                            oinfo_capa(oinfo), CAPA_OPC_OSS_TRUNC);
928         if (rc)
929                 GOTO(out_env, rc);
930
931         fo = ofd_object_find(env, ofd, &info->fti_fid);
932         if (IS_ERR(fo)) {
933                 CERROR("%s: error finding object "DFID": rc = %ld\n",
934                        exp->exp_obd->obd_name, PFID(&info->fti_fid),
935                        PTR_ERR(fo));
936                 GOTO(out_env, rc = PTR_ERR(fo));
937         }
938
939         LASSERT(oinfo->oi_policy.l_extent.end == OBD_OBJECT_EOF);
940         if (oinfo->oi_policy.l_extent.end == OBD_OBJECT_EOF) {
941                 /* Truncate case */
942                 oinfo->oi_oa->o_size = oinfo->oi_policy.l_extent.start;
943         } else if (oinfo->oi_policy.l_extent.end >= oinfo->oi_oa->o_size) {
944                 oinfo->oi_oa->o_size = oinfo->oi_policy.l_extent.end;
945         }
946
947         la_from_obdo(&info->fti_attr, oinfo->oi_oa,
948                      OBD_MD_FLMTIME | OBD_MD_FLATIME | OBD_MD_FLCTIME);
949         info->fti_attr.la_valid &= ~LA_TYPE;
950         info->fti_attr.la_size = oinfo->oi_policy.l_extent.start;
951         info->fti_attr.la_valid |= LA_SIZE;
952
953         if (oinfo->oi_oa->o_valid & OBD_MD_FLFID) {
954                 ff = &info->fti_mds_fid;
955                 ofd_prepare_fidea(ff, oinfo->oi_oa);
956         }
957
958         rc = ofd_object_punch(env, fo, oinfo->oi_policy.l_extent.start,
959                               oinfo->oi_policy.l_extent.end, &info->fti_attr,
960                               ff);
961         if (rc)
962                 GOTO(out, rc);
963
964         res = ldlm_resource_get(ns, NULL, &info->fti_resid, LDLM_EXTENT, 0);
965         if (res != NULL) {
966                 ldlm_res_lvbo_update(res, NULL, 0);
967                 ldlm_resource_putref(res);
968         }
969
970         oinfo->oi_oa->o_valid = OBD_MD_FLID;
971         /* Quota release needs uid/gid info */
972         rc = ofd_attr_get(env, fo, &info->fti_attr);
973         obdo_from_la(oinfo->oi_oa, &info->fti_attr,
974                      OFD_VALID_FLAGS | LA_UID | LA_GID);
975         ofd_info2oti(info, oti);
976
977         ofd_counter_incr(exp, LPROC_OFD_STATS_PUNCH, oti->oti_jobid, 1);
978         EXIT;
979 out:
980         ofd_object_put(env, fo);
981 out_env:
982         return rc;
983 }
984
985 static int ofd_destroy_by_fid(const struct lu_env *env,
986                               struct ofd_device *ofd,
987                               const struct lu_fid *fid, int orphan)
988 {
989         struct ofd_thread_info  *info = ofd_info(env);
990         struct lustre_handle     lockh;
991         __u64                    flags = LDLM_AST_DISCARD_DATA, rc = 0;
992         ldlm_policy_data_t       policy = {
993                                         .l_extent = { 0, OBD_OBJECT_EOF }
994                                  };
995         struct ofd_object       *fo;
996
997         ENTRY;
998
999         fo = ofd_object_find(env, ofd, fid);
1000         if (IS_ERR(fo))
1001                 RETURN(PTR_ERR(fo));
1002
1003         /* Tell the clients that the object is gone now and that they should
1004          * throw away any cached pages. */
1005         ofd_build_resid(fid, &info->fti_resid);
1006         rc = ldlm_cli_enqueue_local(ofd->ofd_namespace, &info->fti_resid,
1007                                     LDLM_EXTENT, &policy, LCK_PW, &flags,
1008                                     ldlm_blocking_ast, ldlm_completion_ast,
1009                                     NULL, NULL, 0, LVB_T_NONE, NULL, &lockh);
1010
1011         /* We only care about the side-effects, just drop the lock. */
1012         if (rc == ELDLM_OK)
1013                 ldlm_lock_decref(&lockh, LCK_PW);
1014
1015         LASSERT(fo != NULL);
1016
1017         rc = ofd_object_destroy(env, fo, orphan);
1018
1019         ofd_object_put(env, fo);
1020         RETURN(rc);
1021 }
1022
1023 int ofd_destroy(const struct lu_env *env, struct obd_export *exp,
1024                 struct obdo *oa, struct lov_stripe_md *md,
1025                 struct obd_trans_info *oti, struct obd_export *md_exp,
1026                 void *capa)
1027 {
1028         struct ofd_device       *ofd = ofd_exp(exp);
1029         struct ofd_thread_info  *info;
1030         obd_count                count;
1031         int                      rc = 0;
1032
1033         ENTRY;
1034
1035         info = ofd_info_init(env, exp);
1036         ofd_oti2info(info, oti);
1037
1038         if (!(oa->o_valid & OBD_MD_FLGROUP))
1039                 oa->o_seq = 0;
1040
1041         /* check that o_misc makes sense */
1042         if (oa->o_valid & OBD_MD_FLOBJCOUNT)
1043                 count = oa->o_misc;
1044         else
1045                 count = 1; /* default case - single destroy */
1046
1047         /**
1048          * There can be sequence of objects to destroy. Therefore this request
1049          * may have multiple transaction involved in. It is OK, we need only
1050          * the highest used transno to be reported back in reply but not for
1051          * replays, they must report their transno
1052          */
1053         if (info->fti_transno == 0) /* not replay */
1054                 info->fti_mult_trans = 1;
1055         while (count > 0) {
1056                 int lrc;
1057
1058                 fid_ostid_unpack(&info->fti_fid, &oa->o_oi, 0);
1059                 lrc = ofd_destroy_by_fid(env, ofd, &info->fti_fid, 0);
1060                 if (lrc == -ENOENT) {
1061                         CDEBUG(D_INODE,
1062                                "%s: destroying non-existent object "DFID"\n",
1063                                ofd_obd(ofd)->obd_name, PFID(&info->fti_fid));
1064                         /* rewrite rc with -ENOENT only if it is 0 */
1065                         if (rc == 0)
1066                                 rc = lrc;
1067                 } else if (lrc != 0) {
1068                         CERROR("%s: error destroying object "DFID": %d\n",
1069                                ofd_obd(ofd)->obd_name, PFID(&info->fti_fid),
1070                                rc);
1071                         rc = lrc;
1072                 }
1073                 count--;
1074                 oa->o_id++;
1075         }
1076
1077         /* if we have transaction then there were some deletions, we don't
1078          * need to return ENOENT in that case because it will not wait
1079          * for commit of these deletions. The ENOENT must be returned only
1080          * if there were no transations.
1081          */
1082         if (rc == -ENOENT) {
1083                 if (info->fti_transno != 0)
1084                         rc = 0;
1085         } else if (rc != 0) {
1086                 /*
1087                  * If we have at least one transaction then llog record
1088                  * on server will be removed upon commit, so for rc != 0
1089                  * we return no transno and llog record will be reprocessed.
1090                  */
1091                 info->fti_transno = 0;
1092         }
1093         ofd_info2oti(info, oti);
1094         RETURN(rc);
1095 }
1096
1097 static int ofd_orphans_destroy(const struct lu_env *env,
1098                                struct obd_export *exp, struct ofd_device *ofd,
1099                                struct obdo *oa)
1100 {
1101         struct ofd_thread_info  *info = ofd_info(env);
1102         obd_id                   last;
1103         int                      skip_orphan;
1104         int                      rc = 0;
1105         struct ost_id            oi = oa->o_oi;
1106         struct ofd_seq          *oseq;
1107
1108         ENTRY;
1109
1110         oseq = ofd_seq_get(ofd, oa->o_seq);
1111         if (oseq == NULL) {
1112                 CERROR("%s: Can not find seq for "LPU64":"LPU64"\n",
1113                        ofd_name(ofd), oa->o_seq, oa->o_id);
1114                 RETURN(-EINVAL);
1115         }
1116
1117         LASSERT(exp != NULL);
1118         skip_orphan = !!(exp_connect_flags(exp) & OBD_CONNECT_SKIP_ORPHAN);
1119
1120         last = ofd_seq_last_oid(oseq);
1121         LCONSOLE_INFO("%s: deleting orphan objects from "LPX64":"LPU64
1122                       " to "LPU64"\n", ofd_name(ofd), oa->o_seq,
1123                       oa->o_id + 1, last);
1124
1125         for (oi.oi_id = last; oi.oi_id > oa->o_id; oi.oi_id--) {
1126                 fid_ostid_unpack(&info->fti_fid, &oi, 0);
1127                 rc = ofd_destroy_by_fid(env, ofd, &info->fti_fid, 1);
1128                 if (rc && rc != -ENOENT) /* this is pretty fatal... */
1129                         CEMERG("error destroying precreated id "LPU64": %d\n",
1130                                oi.oi_id, rc);
1131                 if (!skip_orphan) {
1132                         ofd_seq_last_oid_set(oseq, oi.oi_id - 1);
1133                         /* update last_id on disk periodically so that if we
1134                          * restart * we don't need to re-scan all of the just
1135                          * deleted objects. */
1136                         if ((oi.oi_id & 511) == 0)
1137                                 ofd_seq_last_oid_write(env, ofd, oseq);
1138                 }
1139         }
1140         CDEBUG(D_HA, "%s: after destroy: set last_objids["LPU64"] = "LPU64"\n",
1141                ofd_obd(ofd)->obd_name, oa->o_seq, oa->o_id);
1142         if (!skip_orphan) {
1143                 rc = ofd_seq_last_oid_write(env, ofd, oseq);
1144         } else {
1145                 /* don't reuse orphan object, return last used objid */
1146                 oa->o_id = last;
1147                 rc = 0;
1148         }
1149         ofd_seq_put(env, oseq);
1150         RETURN(rc);
1151 }
1152
1153 int ofd_create(const struct lu_env *env, struct obd_export *exp,
1154                struct obdo *oa, struct lov_stripe_md **ea,
1155                struct obd_trans_info *oti)
1156 {
1157         struct ofd_device       *ofd = ofd_exp(exp);
1158         struct ofd_thread_info  *info;
1159         obd_seq                 seq = oa->o_seq;
1160         struct ofd_seq          *oseq;
1161         int                     rc = 0, diff;
1162         int                     sync_trans = 0;
1163
1164         ENTRY;
1165
1166         info = ofd_info_init(env, exp);
1167         ofd_oti2info(info, oti);
1168
1169         LASSERT(oa->o_seq >= FID_SEQ_OST_MDT0);
1170         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
1171
1172         CDEBUG(D_INFO, "ofd_create(oa->o_seq="LPU64",oa->o_id="LPU64")\n",
1173                seq, oa->o_id);
1174
1175         oseq = ofd_seq_load(env, ofd, seq);
1176         if (oseq == NULL) {
1177                 CERROR("%s: Can't find oseq "LPX64"\n", ofd_name(ofd), seq);
1178                 RETURN(-EINVAL);
1179         }
1180
1181         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
1182             (oa->o_flags & OBD_FL_RECREATE_OBJS)) {
1183                 if (!ofd_obd(ofd)->obd_recovering ||
1184                     oa->o_id > ofd_seq_last_oid(oseq)) {
1185                         CERROR("recreate objid "LPU64" > last id "LPU64"\n",
1186                                         oa->o_id, ofd_seq_last_oid(oseq));
1187                         GOTO(out_nolock, rc = -EINVAL);
1188                 }
1189                 /* do nothing because we create objects during first write */
1190                 GOTO(out_nolock, rc = 0);
1191         }
1192         /* former ofd_handle_precreate */
1193         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
1194             (oa->o_flags & OBD_FL_DELORPHAN)) {
1195                 /* destroy orphans */
1196                 if (oti->oti_conn_cnt < exp->exp_conn_cnt) {
1197                         CERROR("%s: dropping old orphan cleanup request\n",
1198                                ofd_name(ofd));
1199                         GOTO(out_nolock, rc = 0);
1200                 }
1201                 /* This causes inflight precreates to abort and drop lock */
1202                 oseq->os_destroys_in_progress = 1;
1203                 mutex_lock(&oseq->os_create_lock);
1204                 if (!oseq->os_destroys_in_progress) {
1205                         CERROR("%s:["LPU64"] destroys_in_progress already cleared\n",
1206                                exp->exp_obd->obd_name, oa->o_seq);
1207                         GOTO(out, rc = 0);
1208                 }
1209                 diff = oa->o_id - ofd_seq_last_oid(oseq);
1210                 CDEBUG(D_HA, "ofd_last_id() = "LPU64" -> diff = %d\n",
1211                         ofd_seq_last_oid(oseq), diff);
1212                 if (-diff > OST_MAX_PRECREATE) {
1213                         /* FIXME: should reset precreate_next_id on MDS */
1214                         rc = 0;
1215                 } else if (diff < 0) {
1216                         rc = ofd_orphans_destroy(env, exp, ofd, oa);
1217                         oseq->os_destroys_in_progress = 0;
1218                 } else {
1219                         /* XXX: Used by MDS for the first time! */
1220                         oseq->os_destroys_in_progress = 0;
1221                 }
1222         } else {
1223                 mutex_lock(&oseq->os_create_lock);
1224                 if (oti->oti_conn_cnt < exp->exp_conn_cnt) {
1225                         CERROR("%s: dropping old precreate request\n",
1226                                 ofd_obd(ofd)->obd_name);
1227                         GOTO(out, rc = 0);
1228                 }
1229                 /* only precreate if seq is 0, IDIF or normal and also o_id
1230                  * must be specfied */
1231                 if ((!fid_seq_is_mdt(oa->o_seq) &&
1232                      !fid_seq_is_norm(oa->o_seq) &&
1233                      !fid_seq_is_idif(oa->o_seq)) || oa->o_id == 0) {
1234                         diff = 1; /* shouldn't we create this right now? */
1235                 } else {
1236                         diff = oa->o_id - ofd_seq_last_oid(oseq);
1237                         /* Do sync create if the seq is about to used up */
1238                         if (fid_seq_is_idif(oa->o_seq) ||
1239                             fid_seq_is_mdt0(oa->o_seq)) {
1240                                 if (unlikely(oa->o_id >= IDIF_MAX_OID - 1))
1241                                         sync_trans = 1;
1242                         } else if (fid_seq_is_norm(oa->o_seq)) {
1243                                 if (unlikely(oa->o_id >=
1244                                              LUSTRE_DATA_SEQ_MAX_WIDTH - 1))
1245                                         sync_trans = 1;
1246                         } else {
1247                                 CERROR("%s : invalid o_seq "LPX64": rc = %d\n",
1248                                        ofd_name(ofd), oa->o_seq, -EINVAL);
1249                                 GOTO(out, rc = -EINVAL);
1250                         }
1251                 }
1252         }
1253         if (diff > 0) {
1254                 cfs_time_t       enough_time = cfs_time_shift(DISK_TIMEOUT);
1255                 obd_id           next_id;
1256                 int              created = 0;
1257                 int              count;
1258
1259                 if (!(oa->o_valid & OBD_MD_FLFLAGS) ||
1260                     !(oa->o_flags & OBD_FL_DELORPHAN)) {
1261                         /* don't enforce grant during orphan recovery */
1262                         rc = ofd_grant_create(env,
1263                                               ofd_obd(ofd)->obd_self_export,
1264                                               &diff);
1265                         if (rc) {
1266                                 CDEBUG(D_HA, "%s: failed to acquire grant space"
1267                                        "for precreate (%d)\n",
1268                                        ofd_name(ofd), diff);
1269                                 diff = 0;
1270                         }
1271                 }
1272
1273                 while (diff > 0) {
1274                         next_id = ofd_seq_last_oid(oseq) + 1;
1275                         count = ofd_precreate_batch(ofd, diff);
1276
1277                         CDEBUG(D_HA, "%s: reserve %d objects in group "LPU64
1278                                " at "LPU64"\n", ofd_obd(ofd)->obd_name,
1279                                count, oa->o_seq, next_id);
1280
1281                         if (cfs_time_after(jiffies, enough_time)) {
1282                                 LCONSOLE_WARN("%s: Slow creates, %d/%d objects"
1283                                               " created at a rate of %d/s\n",
1284                                               ofd_obd(ofd)->obd_name,
1285                                               created, diff + created,
1286                                               created / DISK_TIMEOUT);
1287                                 break;
1288                         }
1289
1290                         rc = ofd_precreate_objects(env, ofd, next_id,
1291                                                    oseq, count, sync_trans);
1292                         if (rc > 0) {
1293                                 created += rc;
1294                                 diff -= rc;
1295                         } else if (rc < 0) {
1296                                 break;
1297                         }
1298                 }
1299                 if (created > 0)
1300                         /* some objects got created, we can return
1301                          * them, even if last creation failed */
1302                         rc = 0;
1303                 else
1304                         CERROR("unable to precreate: %d\n", rc);
1305
1306                 oa->o_id = ofd_seq_last_oid(oseq);
1307                 oa->o_valid |= OBD_MD_FLID | OBD_MD_FLGROUP;
1308
1309                 if (!(oa->o_valid & OBD_MD_FLFLAGS) ||
1310                     !(oa->o_flags & OBD_FL_DELORPHAN))
1311                         ofd_grant_commit(env, ofd_obd(ofd)->obd_self_export,
1312                                          rc);
1313         }
1314
1315         ofd_info2oti(info, oti);
1316 out:
1317         mutex_unlock(&oseq->os_create_lock);
1318 out_nolock:
1319         if (rc == 0 && ea != NULL) {
1320                 struct lov_stripe_md *lsm = *ea;
1321
1322                 lsm->lsm_object_id = oa->o_id;
1323         }
1324         ofd_seq_put(env, oseq);
1325         RETURN(rc);
1326 }
1327
1328 int ofd_getattr(const struct lu_env *env, struct obd_export *exp,
1329                 struct obd_info *oinfo)
1330 {
1331         struct ofd_device       *ofd = ofd_exp(exp);
1332         struct ofd_thread_info  *info;
1333         struct ofd_object       *fo;
1334         __u64                    curr_version;
1335         int                      rc = 0;
1336
1337         ENTRY;
1338
1339         info = ofd_info_init(env, exp);
1340
1341         fid_ostid_unpack(&info->fti_fid, &oinfo->oi_oa->o_oi, 0);
1342         rc = ofd_auth_capa(exp, &info->fti_fid, oinfo->oi_oa->o_seq,
1343                            oinfo_capa(oinfo), CAPA_OPC_META_READ);
1344         if (rc)
1345                 GOTO(out, rc);
1346
1347         fo = ofd_object_find(env, ofd, &info->fti_fid);
1348         if (IS_ERR(fo))
1349                 GOTO(out, rc = PTR_ERR(fo));
1350         LASSERT(fo != NULL);
1351         rc = ofd_attr_get(env, fo, &info->fti_attr);
1352         oinfo->oi_oa->o_valid = OBD_MD_FLID;
1353         if (rc == 0)
1354                 obdo_from_la(oinfo->oi_oa, &info->fti_attr,
1355                              OFD_VALID_FLAGS | LA_UID | LA_GID);
1356
1357         /* Store object version in reply */
1358         curr_version = dt_version_get(env, ofd_object_child(fo));
1359         if ((__s64)curr_version != -EOPNOTSUPP) {
1360                 oinfo->oi_oa->o_valid |= OBD_MD_FLDATAVERSION;
1361                 oinfo->oi_oa->o_data_version = curr_version;
1362         }
1363         ofd_object_put(env, fo);
1364 out:
1365         RETURN(rc);
1366 }
1367
1368 static int ofd_sync(const struct lu_env *env, struct obd_export *exp,
1369                     struct obd_info *oinfo, obd_size start, obd_size end,
1370                     struct ptlrpc_request_set *set)
1371 {
1372         struct ofd_device       *ofd = ofd_exp(exp);
1373         struct ofd_thread_info  *info;
1374         struct ofd_object       *fo;
1375         int                      rc = 0;
1376
1377         ENTRY;
1378
1379         /* if no objid is specified, it means "sync whole filesystem" */
1380         if (oinfo->oi_oa == NULL || !(oinfo->oi_oa->o_valid & OBD_MD_FLID)) {
1381                 rc = dt_sync(env, ofd->ofd_osd);
1382                 GOTO(out, rc);
1383         }
1384
1385         info = ofd_info_init(env, exp);
1386         fid_ostid_unpack(&info->fti_fid, &oinfo->oi_oa->o_oi, 0);
1387
1388         rc = ofd_auth_capa(exp, &info->fti_fid, oinfo->oi_oa->o_seq,
1389                            oinfo_capa(oinfo), CAPA_OPC_OSS_TRUNC);
1390         if (rc)
1391                 GOTO(out, rc);
1392
1393         fo = ofd_object_find(env, ofd, &info->fti_fid);
1394         if (IS_ERR(fo)) {
1395                 CERROR("%s: error finding object "DFID": rc = %ld\n",
1396                        exp->exp_obd->obd_name, PFID(&info->fti_fid),
1397                        PTR_ERR(fo));
1398                 GOTO(out, rc = PTR_ERR(fo));
1399         }
1400
1401         if (!ofd_object_exists(fo))
1402                 GOTO(put, rc = -ENOENT);
1403
1404         if (dt_version_get(env, ofd_object_child(fo)) >
1405             ofd_obd(ofd)->obd_last_committed) {
1406                 rc = dt_object_sync(env, ofd_object_child(fo));
1407                 if (rc)
1408                         GOTO(put, rc);
1409         }
1410
1411         oinfo->oi_oa->o_valid = OBD_MD_FLID;
1412         rc = ofd_attr_get(env, fo, &info->fti_attr);
1413         obdo_from_la(oinfo->oi_oa, &info->fti_attr, OFD_VALID_FLAGS);
1414
1415         ofd_counter_incr(exp, LPROC_OFD_STATS_SYNC, oinfo->oi_jobid, 1);
1416         EXIT;
1417 put:
1418         ofd_object_put(env, fo);
1419 out:
1420         return rc;
1421 }
1422
1423 int ofd_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1424                   void *karg, void *uarg)
1425 {
1426         struct lu_env            env;
1427         struct ofd_device       *ofd = ofd_exp(exp);
1428         struct obd_device       *obd = ofd_obd(ofd);
1429         int                      rc;
1430
1431         ENTRY;
1432
1433         CDEBUG(D_IOCTL, "handling ioctl cmd %#x\n", cmd);
1434         rc = lu_env_init(&env, LCT_LOCAL);
1435         if (rc)
1436                 RETURN(rc);
1437
1438         switch (cmd) {
1439         case OBD_IOC_ABORT_RECOVERY:
1440                 CERROR("%s: aborting recovery\n", obd->obd_name);
1441                 target_stop_recovery_thread(obd);
1442                 break;
1443         case OBD_IOC_SYNC:
1444                 CDEBUG(D_RPCTRACE, "syncing ost %s\n", obd->obd_name);
1445                 rc = dt_sync(&env, ofd->ofd_osd);
1446                 break;
1447         case OBD_IOC_SET_READONLY:
1448                 rc = dt_sync(&env, ofd->ofd_osd);
1449                 if (rc == 0)
1450                         rc = dt_ro(&env, ofd->ofd_osd);
1451                 break;
1452         default:
1453                 CERROR("%s: not supported cmd = %d\n", obd->obd_name, cmd);
1454                 rc = -ENOTTY;
1455         }
1456
1457         lu_env_fini(&env);
1458         RETURN(rc);
1459 }
1460
1461 static int ofd_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
1462 {
1463         int rc = 0;
1464
1465         ENTRY;
1466
1467         switch(stage) {
1468         case OBD_CLEANUP_EARLY:
1469                 break;
1470         case OBD_CLEANUP_EXPORTS:
1471                 target_cleanup_recovery(obd);
1472                 break;
1473         }
1474         RETURN(rc);
1475 }
1476
1477 static int ofd_ping(const struct lu_env *env, struct obd_export *exp)
1478 {
1479         ofd_fmd_expire(exp);
1480         return 0;
1481 }
1482
1483 static int ofd_health_check(const struct lu_env *nul, struct obd_device *obd)
1484 {
1485         struct ofd_device       *ofd = ofd_dev(obd->obd_lu_dev);
1486         struct ofd_thread_info  *info;
1487         struct lu_env            env;
1488 #ifdef USE_HEALTH_CHECK_WRITE
1489         struct thandle          *th;
1490 #endif
1491         int                      rc = 0;
1492
1493         /* obd_proc_read_health pass NULL env, we need real one */
1494         rc = lu_env_init(&env, LCT_DT_THREAD);
1495         if (rc)
1496                 RETURN(rc);
1497
1498         info = ofd_info_init(&env, NULL);
1499         rc = dt_statfs(&env, ofd->ofd_osd, &info->fti_u.osfs);
1500         if (unlikely(rc))
1501                 GOTO(out, rc);
1502
1503         if (info->fti_u.osfs.os_state == OS_STATE_READONLY)
1504                 GOTO(out, rc = -EROFS);
1505
1506 #ifdef USE_HEALTH_CHECK_WRITE
1507         OBD_ALLOC(info->fti_buf.lb_buf, CFS_PAGE_SIZE);
1508         if (info->fti_buf.lb_buf == NULL)
1509                 GOTO(out, rc = -ENOMEM);
1510
1511         info->fti_buf.lb_len = CFS_PAGE_SIZE;
1512         info->fti_off = 0;
1513
1514         th = dt_trans_create(&env, ofd->ofd_osd);
1515         if (IS_ERR(th))
1516                 GOTO(out, rc = PTR_ERR(th));
1517
1518         rc = dt_declare_record_write(&env, ofd->ofd_health_check_file,
1519                                      info->fti_buf.lb_len, info->fti_off, th);
1520         if (rc == 0) {
1521                 th->th_sync = 1; /* sync IO is needed */
1522                 rc = dt_trans_start_local(&env, ofd->ofd_osd, th);
1523                 if (rc == 0)
1524                         rc = dt_record_write(&env, ofd->ofd_health_check_file,
1525                                              &info->fti_buf, &info->fti_off,
1526                                              th);
1527         }
1528         dt_trans_stop(&env, ofd->ofd_osd, th);
1529
1530         OBD_FREE(info->fti_buf.lb_buf, CFS_PAGE_SIZE);
1531
1532         CDEBUG(D_INFO, "write 1 page synchronously for checking io rc %d\n",rc);
1533 #endif
1534 out:
1535         lu_env_fini(&env);
1536         return !!rc;
1537 }
1538
1539 /*
1540  * Handle quota control requests to consult current usage/limit.
1541  *
1542  * \param obd - is the obd device associated with the ofd
1543  * \param exp - is the client's export
1544  * \param oqctl - is the obd_quotactl request to be processed
1545  */
1546 static int ofd_quotactl(struct obd_device *obd, struct obd_export *exp,
1547                         struct obd_quotactl *oqctl)
1548 {
1549         struct ofd_device  *ofd = ofd_dev(obd->obd_lu_dev);
1550         struct lu_env       env;
1551         int                 rc;
1552         ENTRY;
1553
1554         /* report success for quota on/off for interoperability with current MDT
1555          * stack */
1556         if (oqctl->qc_cmd == Q_QUOTAON || oqctl->qc_cmd == Q_QUOTAOFF)
1557                 RETURN(0);
1558
1559         rc = lu_env_init(&env, LCT_DT_THREAD);
1560         if (rc)
1561                 RETURN(rc);
1562
1563         rc = lquotactl_slv(&env, ofd->ofd_osd, oqctl);
1564         lu_env_fini(&env);
1565
1566         RETURN(rc);
1567 }
1568
1569 struct obd_ops ofd_obd_ops = {
1570         .o_owner                = THIS_MODULE,
1571         .o_connect              = ofd_obd_connect,
1572         .o_reconnect            = ofd_obd_reconnect,
1573         .o_disconnect           = ofd_obd_disconnect,
1574         .o_set_info_async       = ofd_set_info_async,
1575         .o_get_info             = ofd_get_info,
1576         .o_create               = ofd_create,
1577         .o_statfs               = ofd_statfs,
1578         .o_setattr              = ofd_setattr,
1579         .o_preprw               = ofd_preprw,
1580         .o_commitrw             = ofd_commitrw,
1581         .o_destroy              = ofd_destroy,
1582         .o_init_export          = ofd_init_export,
1583         .o_destroy_export       = ofd_destroy_export,
1584         .o_postrecov            = ofd_obd_postrecov,
1585         .o_punch                = ofd_punch,
1586         .o_getattr              = ofd_getattr,
1587         .o_sync                 = ofd_sync,
1588         .o_iocontrol            = ofd_iocontrol,
1589         .o_precleanup           = ofd_precleanup,
1590         .o_ping                 = ofd_ping,
1591         .o_health_check         = ofd_health_check,
1592         .o_quotactl             = ofd_quotactl,
1593 };