Whamcloud - gitweb
1bc3ccdde1fb86ff76d07e882cb7d5a9157c3c8f
[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 <lustre/lustre_idl.h>
46 #include "ofd_internal.h"
47 #include <obd_cksum.h>
48 #include <lustre_quota.h>
49
50 static int ofd_export_stats_init(struct ofd_device *ofd,
51                                  struct obd_export *exp, void *client_nid)
52 {
53         struct obd_device       *obd = ofd_obd(ofd);
54         struct nid_stat         *stats;
55         int                      num_stats;
56         int                      rc, newnid = 0;
57
58         ENTRY;
59
60         if (obd_uuid_equals(&exp->exp_client_uuid, &obd->obd_uuid))
61                 /* Self-export gets no proc entry */
62                 RETURN(0);
63
64         rc = lprocfs_exp_setup(exp, client_nid, &newnid);
65         if (rc) {
66                 /* Mask error for already created
67                  * /proc entries */
68                 if (rc == -EALREADY)
69                         rc = 0;
70                 RETURN(rc);
71         }
72
73         if (newnid == 0)
74                 RETURN(0);
75
76         stats = exp->exp_nid_stats;
77         LASSERT(stats != NULL);
78
79         num_stats = (sizeof(*obd->obd_type->typ_dt_ops) / sizeof(void *)) +
80                      LPROC_OFD_LAST - 1;
81
82         stats->nid_stats = lprocfs_alloc_stats(num_stats,
83                                                LPROCFS_STATS_FLAG_NOPERCPU);
84         if (stats->nid_stats == NULL)
85                 return -ENOMEM;
86
87         lprocfs_init_ops_stats(LPROC_OFD_LAST, stats->nid_stats);
88         lprocfs_counter_init(stats->nid_stats, LPROC_OFD_READ_BYTES,
89                              LPROCFS_CNTR_AVGMINMAX, "read_bytes", "bytes");
90         lprocfs_counter_init(stats->nid_stats, LPROC_OFD_WRITE_BYTES,
91                              LPROCFS_CNTR_AVGMINMAX, "write_bytes", "bytes");
92
93         rc = lprocfs_register_stats(stats->nid_proc, "stats",
94                                     stats->nid_stats);
95         if (rc)
96                 GOTO(clean, rc);
97
98         rc = lprocfs_nid_ldlm_stats_init(stats);
99         if (rc) {
100                 lprocfs_free_stats(&stats->nid_stats);
101                 GOTO(clean, rc);
102         }
103
104         RETURN(0);
105 clean:
106         return rc;
107 }
108
109 static int ofd_parse_connect_data(const struct lu_env *env,
110                                   struct obd_export *exp,
111                                   struct obd_connect_data *data)
112 {
113         struct ofd_device                *ofd = ofd_exp(exp);
114         struct filter_export_data        *fed = &exp->exp_filter_data;
115
116         if (!data)
117                 RETURN(0);
118
119         CDEBUG(D_RPCTRACE, "%s: cli %s/%p ocd_connect_flags: "LPX64
120                " ocd_version: %x ocd_grant: %d ocd_index: %u"
121                " ocd_group %u\n",
122                exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp,
123                data->ocd_connect_flags, data->ocd_version,
124                data->ocd_grant, data->ocd_index, data->ocd_group);
125
126         if (fed->fed_group != 0 && fed->fed_group != data->ocd_group) {
127                 CWARN("!!! This export (nid %s) used object group %d "
128                       "earlier; now it's trying to use group %d!  This could "
129                       "be a bug in the MDS. Please report to "
130                       "http://bugs.whamcloud.com/\n",
131                       obd_export_nid2str(exp), fed->fed_group,
132                       data->ocd_group);
133                 RETURN(-EPROTO);
134         }
135         fed->fed_group = data->ocd_group;
136
137         data->ocd_connect_flags &= OST_CONNECT_SUPPORTED;
138         exp->exp_connect_data = *data;
139         data->ocd_version = LUSTRE_VERSION_CODE;
140
141         /* Kindly make sure the SKIP_ORPHAN flag is from MDS. */
142         if (data->ocd_connect_flags & OBD_CONNECT_MDS)
143                 CDEBUG(D_HA, "%s: Received MDS connection for group %u\n",
144                        exp->exp_obd->obd_name, data->ocd_group);
145         else if (data->ocd_connect_flags & OBD_CONNECT_SKIP_ORPHAN)
146                 RETURN(-EPROTO);
147
148         if (ofd_grant_param_supp(exp)) {
149                 exp->exp_filter_data.fed_pagesize = data->ocd_blocksize;
150                 /* ocd_{blocksize,inodespace} are log2 values */
151                 data->ocd_blocksize  = ofd->ofd_blockbits;
152                 data->ocd_inodespace = ofd->ofd_dt_conf.ddp_inodespace;
153                 /* ocd_grant_extent is in 1K blocks */
154                 data->ocd_grant_extent = ofd->ofd_dt_conf.ddp_grant_frag >> 10;
155         }
156
157         if (exp_connect_flags(exp) & OBD_CONNECT_GRANT)
158                 data->ocd_grant = ofd_grant_connect(env, exp, data->ocd_grant);
159
160         if (data->ocd_connect_flags & OBD_CONNECT_INDEX) {
161                 struct lr_server_data *lsd = &ofd->ofd_lut.lut_lsd;
162                 int                    index = lsd->lsd_osd_index;
163
164                 if (index != data->ocd_index) {
165                         LCONSOLE_ERROR_MSG(0x136, "Connection from %s to index"
166                                            " %u doesn't match actual OST index"
167                                            " %u in last_rcvd file, bad "
168                                            "configuration?\n",
169                                            obd_export_nid2str(exp), index,
170                                            data->ocd_index);
171                         RETURN(-EBADF);
172                 }
173                 if (!(lsd->lsd_feature_compat & OBD_COMPAT_OST)) {
174                         /* this will only happen on the first connect */
175                         lsd->lsd_feature_compat |= OBD_COMPAT_OST;
176                         /* sync is not needed here as lut_client_add will
177                          * set exp_need_sync flag */
178                         tgt_server_data_update(env, &ofd->ofd_lut, 0);
179                 }
180         }
181         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_SIZE)) {
182                 data->ocd_brw_size = 65536;
183         } else if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) {
184                 data->ocd_brw_size = min(data->ocd_brw_size,
185                                          (__u32)DT_MAX_BRW_SIZE);
186                 if (data->ocd_brw_size == 0) {
187                         CERROR("%s: cli %s/%p ocd_connect_flags: "LPX64
188                                " ocd_version: %x ocd_grant: %d ocd_index: %u "
189                                "ocd_brw_size is unexpectedly zero, "
190                                "network data corruption?"
191                                "Refusing connection of this client\n",
192                                exp->exp_obd->obd_name,
193                                exp->exp_client_uuid.uuid,
194                                exp, data->ocd_connect_flags, data->ocd_version,
195                                data->ocd_grant, data->ocd_index);
196                         RETURN(-EPROTO);
197                 }
198         }
199
200         if (data->ocd_connect_flags & OBD_CONNECT_CKSUM) {
201                 __u32 cksum_types = data->ocd_cksum_types;
202
203                 /* The client set in ocd_cksum_types the checksum types it
204                  * supports. We have to mask off the algorithms that we don't
205                  * support */
206                 data->ocd_cksum_types &= cksum_types_supported_server();
207
208                 if (unlikely(data->ocd_cksum_types == 0)) {
209                         CERROR("%s: Connect with checksum support but no "
210                                "ocd_cksum_types is set\n",
211                                exp->exp_obd->obd_name);
212                         RETURN(-EPROTO);
213                 }
214
215                 CDEBUG(D_RPCTRACE, "%s: cli %s supports cksum type %x, return "
216                        "%x\n", exp->exp_obd->obd_name, obd_export_nid2str(exp),
217                        cksum_types, data->ocd_cksum_types);
218         } else {
219                 /* This client does not support OBD_CONNECT_CKSUM
220                  * fall back to CRC32 */
221                 CDEBUG(D_RPCTRACE, "%s: cli %s does not support "
222                        "OBD_CONNECT_CKSUM, CRC32 will be used\n",
223                        exp->exp_obd->obd_name, obd_export_nid2str(exp));
224         }
225
226         if (data->ocd_connect_flags & OBD_CONNECT_MAXBYTES)
227                 data->ocd_maxbytes = ofd->ofd_dt_conf.ddp_maxbytes;
228
229         if (OCD_HAS_FLAG(data, PINGLESS)) {
230                 if (ptlrpc_pinger_suppress_pings()) {
231                         spin_lock(&exp->exp_obd->obd_dev_lock);
232                         list_del_init(&exp->exp_obd_chain_timed);
233                         spin_unlock(&exp->exp_obd->obd_dev_lock);
234                 } else {
235                         data->ocd_connect_flags &= ~OBD_CONNECT_PINGLESS;
236                 }
237         }
238
239         RETURN(0);
240 }
241
242 static int ofd_obd_reconnect(const struct lu_env *env, struct obd_export *exp,
243                              struct obd_device *obd, struct obd_uuid *cluuid,
244                              struct obd_connect_data *data, void *localdata)
245 {
246         struct ofd_device       *ofd;
247         int                      rc;
248
249         ENTRY;
250
251         if (exp == NULL || obd == NULL || cluuid == NULL)
252                 RETURN(-EINVAL);
253
254         ofd = ofd_dev(obd->obd_lu_dev);
255
256         rc = lu_env_refill((struct lu_env *)env);
257         if (rc != 0) {
258                 CERROR("Failure to refill session: '%d'\n", rc);
259                 RETURN(rc);
260         }
261
262         ofd_info_init(env, exp);
263         rc = ofd_parse_connect_data(env, exp, data);
264         if (rc == 0)
265                 ofd_export_stats_init(ofd, exp, localdata);
266
267         RETURN(rc);
268 }
269
270 static int ofd_obd_connect(const struct lu_env *env, struct obd_export **_exp,
271                            struct obd_device *obd, struct obd_uuid *cluuid,
272                            struct obd_connect_data *data, void *localdata)
273 {
274         struct obd_export       *exp;
275         struct ofd_device       *ofd;
276         struct lustre_handle     conn = { 0 };
277         int                      rc;
278         ENTRY;
279
280         if (_exp == NULL || obd == NULL || cluuid == NULL)
281                 RETURN(-EINVAL);
282
283         ofd = ofd_dev(obd->obd_lu_dev);
284
285         rc = class_connect(&conn, obd, cluuid);
286         if (rc)
287                 RETURN(rc);
288
289         exp = class_conn2export(&conn);
290         LASSERT(exp != NULL);
291
292         rc = lu_env_refill((struct lu_env *)env);
293         if (rc != 0) {
294                 CERROR("Failure to refill session: '%d'\n", rc);
295                 GOTO(out, rc);
296         }
297
298         ofd_info_init(env, exp);
299
300         rc = ofd_parse_connect_data(env, exp, data);
301         if (rc)
302                 GOTO(out, rc);
303
304         if (obd->obd_replayable) {
305                 struct tg_export_data *ted = &exp->exp_target_data;
306
307                 memcpy(ted->ted_lcd->lcd_uuid, cluuid,
308                        sizeof(ted->ted_lcd->lcd_uuid));
309                 rc = tgt_client_new(env, exp);
310                 if (rc != 0)
311                         GOTO(out, rc);
312                 ofd_export_stats_init(ofd, exp, localdata);
313         }
314
315         CDEBUG(D_HA, "%s: get connection from MDS %d\n", obd->obd_name,
316                data->ocd_group);
317
318 out:
319         if (rc != 0) {
320                 class_disconnect(exp);
321                 *_exp = NULL;
322         } else {
323                 *_exp = exp;
324         }
325         RETURN(rc);
326 }
327
328 static int ofd_obd_disconnect(struct obd_export *exp)
329 {
330         struct ofd_device       *ofd = ofd_dev(exp->exp_obd->obd_lu_dev);
331         struct lu_env            env;
332         int                      rc;
333
334         ENTRY;
335
336         LASSERT(exp);
337         class_export_get(exp);
338
339         if (!(exp->exp_flags & OBD_OPT_FORCE))
340                 ofd_grant_sanity_check(ofd_obd(ofd), __FUNCTION__);
341
342         rc = server_disconnect_export(exp);
343
344         ofd_grant_discard(exp);
345
346         rc = lu_env_init(&env, LCT_DT_THREAD);
347         if (rc)
348                 RETURN(rc);
349
350         /* Do not erase record for recoverable client. */
351         if (exp->exp_obd->obd_replayable &&
352             (!exp->exp_obd->obd_fail || exp->exp_failed))
353                 tgt_client_del(&env, exp);
354         lu_env_fini(&env);
355
356         class_export_put(exp);
357         RETURN(rc);
358 }
359
360 static int ofd_init_export(struct obd_export *exp)
361 {
362         int rc;
363
364         spin_lock_init(&exp->exp_filter_data.fed_lock);
365         CFS_INIT_LIST_HEAD(&exp->exp_filter_data.fed_mod_list);
366         spin_lock(&exp->exp_lock);
367         exp->exp_connecting = 1;
368         spin_unlock(&exp->exp_lock);
369
370         /* self-export doesn't need client data and ldlm initialization */
371         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
372                                      &exp->exp_client_uuid)))
373                 return 0;
374
375         rc = tgt_client_alloc(exp);
376         if (rc == 0)
377                 ldlm_init_export(exp);
378         if (rc)
379                 CERROR("%s: Can't initialize export: rc %d\n",
380                        exp->exp_obd->obd_name, rc);
381         return rc;
382 }
383
384 static int ofd_destroy_export(struct obd_export *exp)
385 {
386         struct ofd_device *ofd = ofd_dev(exp->exp_obd->obd_lu_dev);
387
388         if (exp->exp_filter_data.fed_pending)
389                 CERROR("%s: cli %s/%p has %lu pending on destroyed export"
390                        "\n", exp->exp_obd->obd_name, exp->exp_client_uuid.uuid,
391                        exp, exp->exp_filter_data.fed_pending);
392
393         target_destroy_export(exp);
394
395         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
396                                      &exp->exp_client_uuid)))
397                 return 0;
398
399         ldlm_destroy_export(exp);
400         tgt_client_free(exp);
401
402         ofd_fmd_cleanup(exp);
403
404         /*
405          * discard grants once we're sure no more
406          * interaction with the client is possible
407          */
408         ofd_grant_discard(exp);
409         ofd_fmd_cleanup(exp);
410
411         if (exp_connect_flags(exp) & OBD_CONNECT_GRANT_SHRINK) {
412                 if (ofd->ofd_tot_granted_clients > 0)
413                         ofd->ofd_tot_granted_clients --;
414         }
415
416         if (!(exp->exp_flags & OBD_OPT_FORCE))
417                 ofd_grant_sanity_check(exp->exp_obd, __FUNCTION__);
418
419         LASSERT(cfs_list_empty(&exp->exp_filter_data.fed_mod_list));
420         return 0;
421 }
422
423 int ofd_postrecov(const struct lu_env *env, struct ofd_device *ofd)
424 {
425         struct lu_device *ldev = &ofd->ofd_dt_dev.dd_lu_dev;
426
427         CDEBUG(D_HA, "%s: recovery is over\n", ofd_obd(ofd)->obd_name);
428         return ldev->ld_ops->ldo_recovery_complete(env, ldev);
429 }
430
431 int ofd_obd_postrecov(struct obd_device *obd)
432 {
433         struct lu_env            env;
434         struct lu_device        *ldev = obd->obd_lu_dev;
435         int                      rc;
436
437         ENTRY;
438
439         rc = lu_env_init(&env, LCT_DT_THREAD);
440         if (rc)
441                 RETURN(rc);
442         ofd_info_init(&env, obd->obd_self_export);
443
444         rc = ofd_postrecov(&env, ofd_dev(ldev));
445
446         lu_env_fini(&env);
447         RETURN(rc);
448 }
449
450 static int ofd_adapt_sptlrpc_conf(const struct lu_env *env,
451                                   struct obd_device *obd, int initial)
452 {
453         struct filter_obd       *fo = &obd->u.filter;
454         struct sptlrpc_rule_set  tmp_rset;
455         int                      rc;
456
457         sptlrpc_rule_set_init(&tmp_rset);
458         rc = sptlrpc_conf_target_get_rules(obd, &tmp_rset, initial);
459         if (rc) {
460                 CERROR("%s: failed get sptlrpc rules: rc = %d\n",
461                        obd->obd_name, rc);
462                 return rc;
463         }
464
465         sptlrpc_target_update_exp_flavor(obd, &tmp_rset);
466
467         write_lock(&fo->fo_sptlrpc_lock);
468         sptlrpc_rule_set_free(&fo->fo_sptlrpc_rset);
469         fo->fo_sptlrpc_rset = tmp_rset;
470         write_unlock(&fo->fo_sptlrpc_lock);
471
472         return 0;
473 }
474
475 static int ofd_set_mds_conn(struct obd_export *exp, void *val)
476 {
477         int rc = 0;
478
479         ENTRY;
480
481         LCONSOLE_WARN("%s: received MDS connection from %s\n",
482                       exp->exp_obd->obd_name, obd_export_nid2str(exp));
483         RETURN(rc);
484 }
485
486 static int ofd_set_info_async(const struct lu_env *env, struct obd_export *exp,
487                               __u32 keylen, void *key, __u32 vallen, void *val,
488                               struct ptlrpc_request_set *set)
489 {
490         struct ofd_device       *ofd;
491         int                      rc = 0;
492
493         ENTRY;
494
495         if (exp->exp_obd == NULL) {
496                 CDEBUG(D_IOCTL, "invalid export %p\n", exp);
497                 RETURN(-EINVAL);
498         }
499
500         ofd = ofd_exp(exp);
501
502         if (KEY_IS(KEY_CAPA_KEY)) {
503                 rc = ofd_update_capa_key(ofd, val);
504                 if (rc)
505                         CERROR("%s: update capability key failed: rc = %d\n",
506                                exp->exp_obd->obd_name, rc);
507         } else if (KEY_IS(KEY_SPTLRPC_CONF)) {
508                 ofd_adapt_sptlrpc_conf(env, exp->exp_obd, 0);
509         } else if (KEY_IS(KEY_MDS_CONN)) {
510                 rc = ofd_set_mds_conn(exp, val);
511         } else if (KEY_IS(KEY_GRANT_SHRINK)) {
512                 struct ost_body *body = val;
513
514                 ofd_info_init(env, exp);
515                 /** handle grant shrink, similar to a read request */
516                 ofd_grant_prepare_read(env, exp, &body->oa);
517         } else {
518                 CERROR("%s: Unsupported key %s\n",
519                        exp->exp_obd->obd_name, (char*)key);
520                 rc = -EOPNOTSUPP;
521         }
522
523         RETURN(rc);
524 }
525
526 static int ofd_get_info(const struct lu_env *env, struct obd_export *exp,
527                         __u32 keylen, void *key, __u32 *vallen, void *val,
528                         struct lov_stripe_md *lsm)
529 {
530         struct ofd_device       *ofd;
531         int                      rc = 0;
532
533         ENTRY;
534
535         if (exp->exp_obd == NULL) {
536                 CDEBUG(D_IOCTL, "invalid client export %p\n", exp);
537                 RETURN(-EINVAL);
538         }
539
540         ofd = ofd_exp(exp);
541
542         if (KEY_IS(KEY_BLOCKSIZE)) {
543                 __u32 *blocksize = val;
544                 if (blocksize) {
545                         if (*vallen < sizeof(*blocksize))
546                                 RETURN(-EOVERFLOW);
547                         *blocksize = 1 << ofd->ofd_dt_conf.ddp_block_shift;
548                 }
549                 *vallen = sizeof(*blocksize);
550         } else if (KEY_IS(KEY_BLOCKSIZE_BITS)) {
551                 __u32 *blocksize_bits = val;
552                 if (blocksize_bits) {
553                         if (*vallen < sizeof(*blocksize_bits))
554                                 RETURN(-EOVERFLOW);
555                         *blocksize_bits = ofd->ofd_dt_conf.ddp_block_shift;
556                 }
557                 *vallen = sizeof(*blocksize_bits);
558         } else if (KEY_IS(KEY_LAST_ID)) {
559                 obd_id *last_id = val;
560                 struct ofd_seq *oseq;
561
562                 if (val == NULL) {
563                         *vallen = sizeof(obd_id);
564                         RETURN(0);
565                 }
566                 ofd_info_init(env, exp);
567                 oseq = ofd_seq_load(env, ofd,
568                                     (obd_seq)exp->exp_filter_data.fed_group);
569                 LASSERT(!IS_ERR(oseq));
570                 if (last_id) {
571                         if (*vallen < sizeof(*last_id)) {
572                                 ofd_seq_put(env, oseq);
573                                 RETURN(-EOVERFLOW);
574                         }
575                         *last_id = ofd_seq_last_oid(oseq);
576                 }
577                 ofd_seq_put(env, oseq);
578                 *vallen = sizeof(*last_id);
579         } else if (KEY_IS(KEY_FIEMAP)) {
580                 struct ofd_thread_info          *info;
581                 struct ofd_device               *ofd = ofd_exp(exp);
582                 struct ofd_object               *fo;
583                 struct ll_fiemap_info_key       *fm_key = key;
584
585                 if (val == NULL) {
586                         *vallen = fiemap_count_to_size(
587                                                fm_key->fiemap.fm_extent_count);
588                         RETURN(0);
589                 }
590
591                 info = ofd_info_init(env, exp);
592                 ostid_to_fid(&info->fti_fid, &fm_key->oa.o_oi, 0);
593                 CDEBUG(D_INODE, "get FIEMAP of object "DFID"\n",
594                        PFID(&info->fti_fid));
595
596                 fo = ofd_object_find(env, ofd, &info->fti_fid);
597                 if (IS_ERR(fo)) {
598                         CERROR("%s: error finding object "DFID"\n",
599                                exp->exp_obd->obd_name, PFID(&info->fti_fid));
600                         rc = PTR_ERR(fo);
601                 } else {
602                         struct ll_user_fiemap *fiemap = val;
603
604                         ofd_read_lock(env, fo);
605                         if (ofd_object_exists(fo)) {
606                                 *fiemap = fm_key->fiemap;
607                                 rc = dt_fiemap_get(env,
608                                                    ofd_object_child(fo),
609                                                    fiemap);
610                         } else {
611                                 rc = -ENOENT;
612                         }
613                         ofd_read_unlock(env, fo);
614                         ofd_object_put(env, fo);
615                 }
616         } else if (KEY_IS(KEY_SYNC_LOCK_CANCEL)) {
617                 *((__u32 *) val) = ofd->ofd_sync_lock_cancel;
618                 *vallen = sizeof(__u32);
619         } else if (KEY_IS(KEY_LAST_FID)) {
620                 struct lu_env           env;
621                 struct ofd_device       *ofd = ofd_exp(exp);
622                 struct ofd_seq          *oseq;
623                 struct lu_fid           *fid = val;
624                 int                     rc;
625
626                 if (fid == NULL) {
627                         *vallen = sizeof(struct lu_fid);
628                         RETURN(0);
629                 }
630
631                 if (*vallen < sizeof(*fid))
632                         RETURN(-EOVERFLOW);
633
634                 rc = lu_env_init(&env, LCT_DT_THREAD);
635                 if (rc != 0)
636                         RETURN(rc);
637                 ofd_info_init(&env, exp);
638
639                 fid_le_to_cpu(fid, fid);
640
641                 oseq = ofd_seq_load(&env, ofd, ostid_seq((struct ost_id *)fid));
642                 if (IS_ERR(oseq))
643                         GOTO(out_fini, rc = PTR_ERR(oseq));
644
645                 ostid_to_fid(fid, &oseq->os_oi,
646                              ofd->ofd_lut.lut_lsd.lsd_osd_index);
647
648                 CDEBUG(D_HA, "%s: LAST FID is "DFID"\n", ofd_name(ofd),
649                        PFID(fid));
650                 *vallen = sizeof(*fid);
651                 ofd_seq_put(&env, oseq);
652 out_fini:
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         ostid_to_fid(&info->fti_fid, &oinfo->oi_oa->o_oi, 0);
833         ost_fid_build_resid(&info->fti_fid, &info->fti_resid);
834
835         rc = ofd_auth_capa(exp, &info->fti_fid, ostid_seq(&oa->o_oi),
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         ostid_to_fid(&info->fti_fid, &oinfo->oi_oa->o_oi, 0);
919         ost_fid_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, ostid_seq(&oinfo->oi_oa->o_oi),
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         ost_fid_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                 ostid_set_seq_mdt0(&oa->o_oi);
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
1056         CDEBUG(D_HA, "%s: Destroy object "DOSTID" count %d\n", ofd_name(ofd),
1057                POSTID(&oa->o_oi), count);
1058         while (count > 0) {
1059                 int lrc;
1060
1061                 ostid_to_fid(&info->fti_fid, &oa->o_oi, 0);
1062                 lrc = ofd_destroy_by_fid(env, ofd, &info->fti_fid, 0);
1063                 if (lrc == -ENOENT) {
1064                         CDEBUG(D_INODE,
1065                                "%s: destroying non-existent object "DFID"\n",
1066                                ofd_obd(ofd)->obd_name, PFID(&info->fti_fid));
1067                         /* rewrite rc with -ENOENT only if it is 0 */
1068                         if (rc == 0)
1069                                 rc = lrc;
1070                 } else if (lrc != 0) {
1071                         CERROR("%s: error destroying object "DFID": %d\n",
1072                                ofd_obd(ofd)->obd_name, PFID(&info->fti_fid),
1073                                rc);
1074                         rc = lrc;
1075                 }
1076                 count--;
1077                 ostid_inc_id(&oa->o_oi);
1078         }
1079
1080         /* if we have transaction then there were some deletions, we don't
1081          * need to return ENOENT in that case because it will not wait
1082          * for commit of these deletions. The ENOENT must be returned only
1083          * if there were no transations.
1084          */
1085         if (rc == -ENOENT) {
1086                 if (info->fti_transno != 0)
1087                         rc = 0;
1088         } else if (rc != 0) {
1089                 /*
1090                  * If we have at least one transaction then llog record
1091                  * on server will be removed upon commit, so for rc != 0
1092                  * we return no transno and llog record will be reprocessed.
1093                  */
1094                 info->fti_transno = 0;
1095         }
1096         ofd_info2oti(info, oti);
1097         RETURN(rc);
1098 }
1099
1100 static int ofd_orphans_destroy(const struct lu_env *env,
1101                                struct obd_export *exp, struct ofd_device *ofd,
1102                                struct obdo *oa)
1103 {
1104         struct ofd_thread_info  *info = ofd_info(env);
1105         obd_id                   last;
1106         int                      skip_orphan;
1107         int                      rc = 0;
1108         struct ost_id            oi = oa->o_oi;
1109         __u64                    end_id = ostid_id(&oa->o_oi);
1110         struct ofd_seq          *oseq;
1111
1112         ENTRY;
1113
1114         oseq = ofd_seq_get(ofd, ostid_seq(&oa->o_oi));
1115         if (oseq == NULL) {
1116                 CERROR("%s: Can not find seq for "DOSTID"\n",
1117                        ofd_name(ofd), POSTID(&oa->o_oi));
1118                 RETURN(-EINVAL);
1119         }
1120
1121         LASSERT(exp != NULL);
1122         skip_orphan = !!(exp_connect_flags(exp) & OBD_CONNECT_SKIP_ORPHAN);
1123
1124         last = ofd_seq_last_oid(oseq);
1125         LCONSOLE_INFO("%s: deleting orphan objects from "LPX64":"LPU64
1126                       " to "LPU64"\n", ofd_name(ofd), ostid_seq(&oa->o_oi),
1127                       end_id + 1, last);
1128
1129         for (ostid_set_id(&oi, last); ostid_id(&oi) > end_id;
1130                           ostid_dec_id(&oi)) {
1131                 ostid_to_fid(&info->fti_fid, &oi, 0);
1132                 rc = ofd_destroy_by_fid(env, ofd, &info->fti_fid, 1);
1133                 if (rc && rc != -ENOENT) /* this is pretty fatal... */
1134                         CEMERG("error destroying precreated id "DOSTID": %d\n",
1135                                POSTID(&oi), rc);
1136                 if (!skip_orphan) {
1137                         ofd_seq_last_oid_set(oseq, ostid_id(&oi) - 1);
1138                         /* update last_id on disk periodically so that if we
1139                          * restart * we don't need to re-scan all of the just
1140                          * deleted objects. */
1141                         if ((ostid_id(&oi) & 511) == 0)
1142                                 ofd_seq_last_oid_write(env, ofd, oseq);
1143                 }
1144         }
1145         CDEBUG(D_HA, "%s: after destroy: set last_objids"DOSTID"\n",
1146                ofd_obd(ofd)->obd_name, POSTID(&oa->o_oi));
1147         if (!skip_orphan) {
1148                 rc = ofd_seq_last_oid_write(env, ofd, oseq);
1149         } else {
1150                 /* don't reuse orphan object, return last used objid */
1151                 ostid_set_id(&oa->o_oi, last);
1152                 rc = 0;
1153         }
1154         ofd_seq_put(env, oseq);
1155         RETURN(rc);
1156 }
1157
1158 int ofd_create(const struct lu_env *env, struct obd_export *exp,
1159                struct obdo *oa, struct lov_stripe_md **ea,
1160                struct obd_trans_info *oti)
1161 {
1162         struct ofd_device       *ofd = ofd_exp(exp);
1163         struct ofd_thread_info  *info;
1164         obd_seq                 seq = ostid_seq(&oa->o_oi);
1165         struct ofd_seq          *oseq;
1166         int                     rc = 0, diff;
1167         int                     sync_trans = 0;
1168
1169         ENTRY;
1170
1171         info = ofd_info_init(env, exp);
1172         ofd_oti2info(info, oti);
1173
1174         LASSERT(ostid_seq(&oa->o_oi) >= FID_SEQ_OST_MDT0);
1175         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
1176
1177         CDEBUG(D_INFO, "ofd_create("DOSTID")\n", POSTID(&oa->o_oi));
1178
1179         oseq = ofd_seq_load(env, ofd, seq);
1180         if (IS_ERR(oseq)) {
1181                 CERROR("%s: Can't find FID Sequence "LPX64": rc = %ld\n",
1182                        ofd_name(ofd), seq, PTR_ERR(oseq));
1183                 RETURN(-EINVAL);
1184         }
1185
1186         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
1187             (oa->o_flags & OBD_FL_RECREATE_OBJS)) {
1188                 if (!ofd_obd(ofd)->obd_recovering ||
1189                     ostid_id(&oa->o_oi) > ofd_seq_last_oid(oseq)) {
1190                         CERROR("recreate objid "DOSTID" > last id "LPU64"\n",
1191                                POSTID(&oa->o_oi), ofd_seq_last_oid(oseq));
1192                         GOTO(out_nolock, rc = -EINVAL);
1193                 }
1194                 /* do nothing because we create objects during first write */
1195                 GOTO(out_nolock, rc = 0);
1196         }
1197         /* former ofd_handle_precreate */
1198         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
1199             (oa->o_flags & OBD_FL_DELORPHAN)) {
1200                 /* destroy orphans */
1201                 if (oti->oti_conn_cnt < exp->exp_conn_cnt) {
1202                         CERROR("%s: dropping old orphan cleanup request\n",
1203                                ofd_name(ofd));
1204                         GOTO(out_nolock, rc = 0);
1205                 }
1206                 /* This causes inflight precreates to abort and drop lock */
1207                 oseq->os_destroys_in_progress = 1;
1208                 mutex_lock(&oseq->os_create_lock);
1209                 if (!oseq->os_destroys_in_progress) {
1210                         CERROR("%s:["LPU64"] destroys_in_progress already"
1211                                " cleared\n", exp->exp_obd->obd_name,
1212                                ostid_seq(&oa->o_oi));
1213                         GOTO(out, rc = 0);
1214                 }
1215                 diff = ostid_id(&oa->o_oi) - ofd_seq_last_oid(oseq);
1216                 CDEBUG(D_HA, "ofd_last_id() = "LPU64" -> diff = %d\n",
1217                         ofd_seq_last_oid(oseq), diff);
1218                 if (-diff > OST_MAX_PRECREATE) {
1219                         /* FIXME: should reset precreate_next_id on MDS */
1220                         rc = 0;
1221                 } else if (diff < 0) {
1222                         rc = ofd_orphans_destroy(env, exp, ofd, oa);
1223                         oseq->os_destroys_in_progress = 0;
1224                 } else {
1225                         /* XXX: Used by MDS for the first time! */
1226                         oseq->os_destroys_in_progress = 0;
1227                 }
1228         } else {
1229                 mutex_lock(&oseq->os_create_lock);
1230                 if (oti->oti_conn_cnt < exp->exp_conn_cnt) {
1231                         CERROR("%s: dropping old precreate request\n",
1232                                 ofd_obd(ofd)->obd_name);
1233                         GOTO(out, rc = 0);
1234                 }
1235                 /* only precreate if seq is 0, IDIF or normal and also o_id
1236                  * must be specfied */
1237                 if ((!fid_seq_is_mdt(ostid_seq(&oa->o_oi)) &&
1238                      !fid_seq_is_norm(ostid_seq(&oa->o_oi)) &&
1239                      !fid_seq_is_idif(ostid_seq(&oa->o_oi))) ||
1240                                 ostid_id(&oa->o_oi) == 0) {
1241                         diff = 1; /* shouldn't we create this right now? */
1242                 } else {
1243                         diff = ostid_id(&oa->o_oi) - ofd_seq_last_oid(oseq);
1244                         /* Do sync create if the seq is about to used up */
1245                         if (fid_seq_is_idif(ostid_seq(&oa->o_oi)) ||
1246                             fid_seq_is_mdt0(ostid_seq(&oa->o_oi))) {
1247                                 if (unlikely(ostid_id(&oa->o_oi) >= IDIF_MAX_OID - 1))
1248                                         sync_trans = 1;
1249                         } else if (fid_seq_is_norm(ostid_seq(&oa->o_oi))) {
1250                                 if (unlikely(ostid_id(&oa->o_oi) >=
1251                                              LUSTRE_DATA_SEQ_MAX_WIDTH - 1))
1252                                         sync_trans = 1;
1253                         } else {
1254                                 CERROR("%s : invalid o_seq "DOSTID": rc = %d\n",
1255                                        ofd_name(ofd), POSTID(&oa->o_oi), -EINVAL);
1256                                 GOTO(out, rc = -EINVAL);
1257                         }
1258                 }
1259         }
1260         if (diff > 0) {
1261                 cfs_time_t       enough_time = cfs_time_shift(DISK_TIMEOUT);
1262                 obd_id           next_id;
1263                 int              created = 0;
1264                 int              count;
1265
1266                 if (!(oa->o_valid & OBD_MD_FLFLAGS) ||
1267                     !(oa->o_flags & OBD_FL_DELORPHAN)) {
1268                         /* don't enforce grant during orphan recovery */
1269                         rc = ofd_grant_create(env,
1270                                               ofd_obd(ofd)->obd_self_export,
1271                                               &diff);
1272                         if (rc) {
1273                                 CDEBUG(D_HA, "%s: failed to acquire grant "
1274                                        "space for precreate (%d): rc = %d\n",
1275                                        ofd_name(ofd), diff, rc);
1276                                 diff = 0;
1277                         }
1278                 }
1279
1280                 while (diff > 0) {
1281                         next_id = ofd_seq_last_oid(oseq) + 1;
1282                         count = ofd_precreate_batch(ofd, diff);
1283
1284                         CDEBUG(D_HA, "%s: reserve %d objects in group "LPX64
1285                                " at "LPU64"\n", ofd_obd(ofd)->obd_name,
1286                                count, ostid_seq(&oa->o_oi), next_id);
1287
1288                         if (cfs_time_after(jiffies, enough_time)) {
1289                                 LCONSOLE_WARN("%s: Slow creates, %d/%d objects"
1290                                               " created at a rate of %d/s\n",
1291                                               ofd_obd(ofd)->obd_name,
1292                                               created, diff + created,
1293                                               created / DISK_TIMEOUT);
1294                                 break;
1295                         }
1296
1297                         rc = ofd_precreate_objects(env, ofd, next_id,
1298                                                    oseq, count, sync_trans);
1299                         if (rc > 0) {
1300                                 created += rc;
1301                                 diff -= rc;
1302                         } else if (rc < 0) {
1303                                 break;
1304                         }
1305                 }
1306                 if (created > 0)
1307                         /* some objects got created, we can return
1308                          * them, even if last creation failed */
1309                         rc = 0;
1310                 else
1311                         CERROR("%s: unable to precreate: rc = %d\n",
1312                                ofd_name(ofd), rc);
1313
1314                 ostid_set_id(&oa->o_oi, ofd_seq_last_oid(oseq));
1315                 oa->o_valid |= OBD_MD_FLID | OBD_MD_FLGROUP;
1316
1317                 if (!(oa->o_valid & OBD_MD_FLFLAGS) ||
1318                     !(oa->o_flags & OBD_FL_DELORPHAN))
1319                         ofd_grant_commit(env, ofd_obd(ofd)->obd_self_export,
1320                                          rc);
1321         }
1322
1323         ofd_info2oti(info, oti);
1324 out:
1325         mutex_unlock(&oseq->os_create_lock);
1326 out_nolock:
1327         if (rc == 0 && ea != NULL) {
1328                 struct lov_stripe_md *lsm = *ea;
1329
1330                 lsm->lsm_oi = oa->o_oi;
1331         }
1332         ofd_seq_put(env, oseq);
1333         RETURN(rc);
1334 }
1335
1336 int ofd_getattr(const struct lu_env *env, struct obd_export *exp,
1337                 struct obd_info *oinfo)
1338 {
1339         struct ofd_device       *ofd = ofd_exp(exp);
1340         struct ofd_thread_info  *info;
1341         struct ofd_object       *fo;
1342         __u64                    curr_version;
1343         int                      rc = 0;
1344
1345         ENTRY;
1346
1347         info = ofd_info_init(env, exp);
1348
1349         ostid_to_fid(&info->fti_fid, &oinfo->oi_oa->o_oi, 0);
1350         rc = ofd_auth_capa(exp, &info->fti_fid, ostid_seq(&oinfo->oi_oa->o_oi),
1351                            oinfo_capa(oinfo), CAPA_OPC_META_READ);
1352         if (rc)
1353                 GOTO(out, rc);
1354
1355         fo = ofd_object_find(env, ofd, &info->fti_fid);
1356         if (IS_ERR(fo))
1357                 GOTO(out, rc = PTR_ERR(fo));
1358         LASSERT(fo != NULL);
1359         rc = ofd_attr_get(env, fo, &info->fti_attr);
1360         oinfo->oi_oa->o_valid = OBD_MD_FLID;
1361         if (rc == 0)
1362                 obdo_from_la(oinfo->oi_oa, &info->fti_attr,
1363                              OFD_VALID_FLAGS | LA_UID | LA_GID);
1364
1365         /* Store object version in reply */
1366         curr_version = dt_version_get(env, ofd_object_child(fo));
1367         if ((__s64)curr_version != -EOPNOTSUPP) {
1368                 oinfo->oi_oa->o_valid |= OBD_MD_FLDATAVERSION;
1369                 oinfo->oi_oa->o_data_version = curr_version;
1370         }
1371         ofd_object_put(env, fo);
1372 out:
1373         RETURN(rc);
1374 }
1375
1376 static int ofd_sync(const struct lu_env *env, struct obd_export *exp,
1377                     struct obd_info *oinfo, obd_size start, obd_size end,
1378                     struct ptlrpc_request_set *set)
1379 {
1380         struct ofd_device       *ofd = ofd_exp(exp);
1381         struct ofd_thread_info  *info;
1382         struct ofd_object       *fo;
1383         int                      rc = 0;
1384
1385         ENTRY;
1386
1387         /* if no objid is specified, it means "sync whole filesystem" */
1388         if (oinfo->oi_oa == NULL || !(oinfo->oi_oa->o_valid & OBD_MD_FLID)) {
1389                 rc = dt_sync(env, ofd->ofd_osd);
1390                 GOTO(out, rc);
1391         }
1392
1393         info = ofd_info_init(env, exp);
1394         ostid_to_fid(&info->fti_fid, &oinfo->oi_oa->o_oi, 0);
1395
1396         rc = ofd_auth_capa(exp, &info->fti_fid, ostid_seq(&oinfo->oi_oa->o_oi),
1397                            oinfo_capa(oinfo), CAPA_OPC_OSS_TRUNC);
1398         if (rc)
1399                 GOTO(out, rc);
1400
1401         fo = ofd_object_find(env, ofd, &info->fti_fid);
1402         if (IS_ERR(fo)) {
1403                 CERROR("%s: error finding object "DFID": rc = %ld\n",
1404                        exp->exp_obd->obd_name, PFID(&info->fti_fid),
1405                        PTR_ERR(fo));
1406                 GOTO(out, rc = PTR_ERR(fo));
1407         }
1408
1409         if (!ofd_object_exists(fo))
1410                 GOTO(put, rc = -ENOENT);
1411
1412         if (dt_version_get(env, ofd_object_child(fo)) >
1413             ofd_obd(ofd)->obd_last_committed) {
1414                 rc = dt_object_sync(env, ofd_object_child(fo));
1415                 if (rc)
1416                         GOTO(put, rc);
1417         }
1418
1419         oinfo->oi_oa->o_valid = OBD_MD_FLID;
1420         rc = ofd_attr_get(env, fo, &info->fti_attr);
1421         obdo_from_la(oinfo->oi_oa, &info->fti_attr, OFD_VALID_FLAGS);
1422
1423         ofd_counter_incr(exp, LPROC_OFD_STATS_SYNC, oinfo->oi_jobid, 1);
1424         EXIT;
1425 put:
1426         ofd_object_put(env, fo);
1427 out:
1428         return rc;
1429 }
1430
1431 static int ofd_ioc_get_obj_version(const struct lu_env *env,
1432                                    struct ofd_device *ofd, void *karg)
1433 {
1434         struct obd_ioctl_data *data = karg;
1435         struct lu_fid          fid;
1436         struct ofd_object     *fo;
1437         dt_obj_version_t       version;
1438         int                    rc = 0;
1439
1440         ENTRY;
1441
1442         if (data->ioc_inlbuf2 == NULL || data->ioc_inllen2 != sizeof(version))
1443                 GOTO(out, rc = -EINVAL);
1444
1445         if (data->ioc_inlbuf1 != NULL && data->ioc_inllen1 == sizeof(fid)) {
1446                 fid = *(struct lu_fid *)data->ioc_inlbuf1;
1447         } else if (data->ioc_inlbuf3 != NULL &&
1448                    data->ioc_inllen3 == sizeof(__u64) &&
1449                    data->ioc_inlbuf4 != NULL &&
1450                    data->ioc_inllen4 == sizeof(__u64)) {
1451                 struct ost_id ostid;
1452
1453                 ostid_set_seq(&ostid, *(__u64 *)data->ioc_inlbuf4);
1454                 ostid_set_id(&ostid, *(__u64 *)data->ioc_inlbuf3);
1455                 rc = ostid_to_fid(&fid, &ostid, 0);
1456                 if (rc != 0)
1457                         GOTO(out, rc);
1458         } else {
1459                 GOTO(out, rc = -EINVAL);
1460         }
1461
1462         if (!fid_is_sane(&fid))
1463                 GOTO(out, rc = -EINVAL);
1464
1465         fo = ofd_object_find(env, ofd, &fid);
1466         if (IS_ERR(fo))
1467                 GOTO(out, rc = PTR_ERR(fo));
1468
1469         if (!ofd_object_exists(fo))
1470                 GOTO(out_fo, rc = -ENOENT);
1471
1472         if (lu_object_remote(&fo->ofo_obj.do_lu))
1473                 GOTO(out_fo, rc = -EREMOTE);
1474
1475         version = dt_version_get(env, ofd_object_child(fo));
1476         if (version == 0)
1477                 GOTO(out_fo, rc = -EIO);
1478
1479         *(dt_obj_version_t *)data->ioc_inlbuf2 = version;
1480
1481         EXIT;
1482 out_fo:
1483         ofd_object_put(env, fo);
1484 out:
1485         return rc;
1486 }
1487
1488 int ofd_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1489                   void *karg, void *uarg)
1490 {
1491         struct lu_env            env;
1492         struct ofd_device       *ofd = ofd_exp(exp);
1493         struct obd_device       *obd = ofd_obd(ofd);
1494         int                      rc;
1495
1496         ENTRY;
1497
1498         CDEBUG(D_IOCTL, "handling ioctl cmd %#x\n", cmd);
1499         rc = lu_env_init(&env, LCT_DT_THREAD);
1500         if (rc)
1501                 RETURN(rc);
1502
1503         switch (cmd) {
1504         case OBD_IOC_ABORT_RECOVERY:
1505                 CERROR("%s: aborting recovery\n", obd->obd_name);
1506                 target_stop_recovery_thread(obd);
1507                 break;
1508         case OBD_IOC_SYNC:
1509                 CDEBUG(D_RPCTRACE, "syncing ost %s\n", obd->obd_name);
1510                 rc = dt_sync(&env, ofd->ofd_osd);
1511                 break;
1512         case OBD_IOC_SET_READONLY:
1513                 rc = dt_sync(&env, ofd->ofd_osd);
1514                 if (rc == 0)
1515                         rc = dt_ro(&env, ofd->ofd_osd);
1516                 break;
1517         case OBD_IOC_GET_OBJ_VERSION:
1518                 rc = ofd_ioc_get_obj_version(&env, ofd, karg);
1519                 break;
1520         default:
1521                 CERROR("%s: not supported cmd = %d\n", obd->obd_name, cmd);
1522                 rc = -ENOTTY;
1523         }
1524
1525         lu_env_fini(&env);
1526         RETURN(rc);
1527 }
1528
1529 static int ofd_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
1530 {
1531         int rc = 0;
1532
1533         ENTRY;
1534
1535         switch(stage) {
1536         case OBD_CLEANUP_EARLY:
1537                 break;
1538         case OBD_CLEANUP_EXPORTS:
1539                 target_cleanup_recovery(obd);
1540                 break;
1541         }
1542         RETURN(rc);
1543 }
1544
1545 static int ofd_ping(const struct lu_env *env, struct obd_export *exp)
1546 {
1547         ofd_fmd_expire(exp);
1548         return 0;
1549 }
1550
1551 static int ofd_health_check(const struct lu_env *nul, struct obd_device *obd)
1552 {
1553         struct ofd_device       *ofd = ofd_dev(obd->obd_lu_dev);
1554         struct ofd_thread_info  *info;
1555         struct lu_env            env;
1556 #ifdef USE_HEALTH_CHECK_WRITE
1557         struct thandle          *th;
1558 #endif
1559         int                      rc = 0;
1560
1561         /* obd_proc_read_health pass NULL env, we need real one */
1562         rc = lu_env_init(&env, LCT_DT_THREAD);
1563         if (rc)
1564                 RETURN(rc);
1565
1566         info = ofd_info_init(&env, NULL);
1567         rc = dt_statfs(&env, ofd->ofd_osd, &info->fti_u.osfs);
1568         if (unlikely(rc))
1569                 GOTO(out, rc);
1570
1571         if (info->fti_u.osfs.os_state == OS_STATE_READONLY)
1572                 GOTO(out, rc = -EROFS);
1573
1574 #ifdef USE_HEALTH_CHECK_WRITE
1575         OBD_ALLOC(info->fti_buf.lb_buf, CFS_PAGE_SIZE);
1576         if (info->fti_buf.lb_buf == NULL)
1577                 GOTO(out, rc = -ENOMEM);
1578
1579         info->fti_buf.lb_len = CFS_PAGE_SIZE;
1580         info->fti_off = 0;
1581
1582         th = dt_trans_create(&env, ofd->ofd_osd);
1583         if (IS_ERR(th))
1584                 GOTO(out, rc = PTR_ERR(th));
1585
1586         rc = dt_declare_record_write(&env, ofd->ofd_health_check_file,
1587                                      info->fti_buf.lb_len, info->fti_off, th);
1588         if (rc == 0) {
1589                 th->th_sync = 1; /* sync IO is needed */
1590                 rc = dt_trans_start_local(&env, ofd->ofd_osd, th);
1591                 if (rc == 0)
1592                         rc = dt_record_write(&env, ofd->ofd_health_check_file,
1593                                              &info->fti_buf, &info->fti_off,
1594                                              th);
1595         }
1596         dt_trans_stop(&env, ofd->ofd_osd, th);
1597
1598         OBD_FREE(info->fti_buf.lb_buf, CFS_PAGE_SIZE);
1599
1600         CDEBUG(D_INFO, "write 1 page synchronously for checking io rc %d\n",rc);
1601 #endif
1602 out:
1603         lu_env_fini(&env);
1604         return !!rc;
1605 }
1606
1607 /*
1608  * Handle quota control requests to consult current usage/limit.
1609  *
1610  * \param obd - is the obd device associated with the ofd
1611  * \param exp - is the client's export
1612  * \param oqctl - is the obd_quotactl request to be processed
1613  */
1614 static int ofd_quotactl(struct obd_device *obd, struct obd_export *exp,
1615                         struct obd_quotactl *oqctl)
1616 {
1617         struct ofd_device  *ofd = ofd_dev(obd->obd_lu_dev);
1618         struct lu_env       env;
1619         int                 rc;
1620         ENTRY;
1621
1622         /* report success for quota on/off for interoperability with current MDT
1623          * stack */
1624         if (oqctl->qc_cmd == Q_QUOTAON || oqctl->qc_cmd == Q_QUOTAOFF)
1625                 RETURN(0);
1626
1627         rc = lu_env_init(&env, LCT_DT_THREAD);
1628         if (rc)
1629                 RETURN(rc);
1630
1631         rc = lquotactl_slv(&env, ofd->ofd_osd, oqctl);
1632         lu_env_fini(&env);
1633
1634         RETURN(rc);
1635 }
1636
1637 struct obd_ops ofd_obd_ops = {
1638         .o_owner                = THIS_MODULE,
1639         .o_connect              = ofd_obd_connect,
1640         .o_reconnect            = ofd_obd_reconnect,
1641         .o_disconnect           = ofd_obd_disconnect,
1642         .o_set_info_async       = ofd_set_info_async,
1643         .o_get_info             = ofd_get_info,
1644         .o_create               = ofd_create,
1645         .o_statfs               = ofd_statfs,
1646         .o_setattr              = ofd_setattr,
1647         .o_preprw               = ofd_preprw,
1648         .o_commitrw             = ofd_commitrw,
1649         .o_destroy              = ofd_destroy,
1650         .o_init_export          = ofd_init_export,
1651         .o_destroy_export       = ofd_destroy_export,
1652         .o_postrecov            = ofd_obd_postrecov,
1653         .o_punch                = ofd_punch,
1654         .o_getattr              = ofd_getattr,
1655         .o_sync                 = ofd_sync,
1656         .o_iocontrol            = ofd_iocontrol,
1657         .o_precleanup           = ofd_precleanup,
1658         .o_ping                 = ofd_ping,
1659         .o_health_check         = ofd_health_check,
1660         .o_quotactl             = ofd_quotactl,
1661 };