Whamcloud - gitweb
fc9802ca271e8d22f52906180db9579509540894
[fs/lustre-release.git] / lustre / lod / lod_object.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,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  * GNU General Public License version 2 for more details.  A copy is
14  * included in the COPYING file that accompanied this code.
15  *
16  * You should have received a copy of the GNU General Public License
17  * along with this program; if not, write to the Free Software
18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright  2009 Sun Microsystems, Inc. All rights reserved
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2012, 2014, Intel Corporation.
27  */
28 /*
29  * lustre/lod/lod_object.c
30  *
31  * This file contains implementations of methods for the OSD API
32  * for the Logical Object Device (LOD) layer, which provides a virtual
33  * local OSD object interface to the MDD layer, and abstracts the
34  * addressing of local (OSD) and remote (OSP) objects. The API is
35  * described in the file lustre/include/dt_object.h and in
36  * lustre/doc/osd-api.txt.
37  *
38  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
39  */
40
41 #define DEBUG_SUBSYSTEM S_MDS
42
43 #include <obd.h>
44 #include <obd_class.h>
45 #include <lustre_ver.h>
46 #include <obd_support.h>
47 #include <lprocfs_status.h>
48
49 #include <lustre_fid.h>
50 #include <lustre_param.h>
51 #include <lustre_fid.h>
52 #include <lustre_lmv.h>
53 #include <md_object.h>
54 #include <lustre_linkea.h>
55
56 #include "lod_internal.h"
57
58 static const char dot[] = ".";
59 static const char dotdot[] = "..";
60
61 static const struct dt_body_operations lod_body_lnk_ops;
62
63 /**
64  * Implementation of dt_index_operations::dio_lookup
65  *
66  * Used with regular (non-striped) objects.
67  *
68  * \see dt_index_operations::dio_lookup() in the API description for details.
69  */
70 static int lod_index_lookup(const struct lu_env *env, struct dt_object *dt,
71                             struct dt_rec *rec, const struct dt_key *key,
72                             struct lustre_capa *capa)
73 {
74         struct dt_object *next = dt_object_child(dt);
75         return next->do_index_ops->dio_lookup(env, next, rec, key, capa);
76 }
77
78 /**
79  * Implementation of dt_index_operations::dio_declare_insert.
80  *
81  * Used with regular (non-striped) objects.
82  *
83  * \see dt_index_operations::dio_declare_insert() in the API description
84  * for details.
85  */
86 static int lod_declare_index_insert(const struct lu_env *env,
87                                     struct dt_object *dt,
88                                     const struct dt_rec *rec,
89                                     const struct dt_key *key,
90                                     struct thandle *handle)
91 {
92         return dt_declare_insert(env, dt_object_child(dt), rec, key, handle);
93 }
94
95 /**
96  * Implementation of dt_index_operations::dio_insert.
97  *
98  * Used with regular (non-striped) objects
99  *
100  * \see dt_index_operations::dio_insert() in the API description for details.
101  */
102 static int lod_index_insert(const struct lu_env *env,
103                             struct dt_object *dt,
104                             const struct dt_rec *rec,
105                             const struct dt_key *key,
106                             struct thandle *th,
107                             struct lustre_capa *capa,
108                             int ign)
109 {
110         return dt_insert(env, dt_object_child(dt), rec, key, th, capa, ign);
111 }
112
113 /**
114  * Implementation of dt_index_operations::dio_declare_delete.
115  *
116  * Used with regular (non-striped) objects.
117  *
118  * \see dt_index_operations::dio_declare_delete() in the API description
119  * for details.
120  */
121 static int lod_declare_index_delete(const struct lu_env *env,
122                                     struct dt_object *dt,
123                                     const struct dt_key *key,
124                                     struct thandle *th)
125 {
126         return dt_declare_delete(env, dt_object_child(dt), key, th);
127 }
128
129 /**
130  * Implementation of dt_index_operations::dio_delete.
131  *
132  * Used with regular (non-striped) objects.
133  *
134  * \see dt_index_operations::dio_delete() in the API description for details.
135  */
136 static int lod_index_delete(const struct lu_env *env,
137                             struct dt_object *dt,
138                             const struct dt_key *key,
139                             struct thandle *th,
140                             struct lustre_capa *capa)
141 {
142         return dt_delete(env, dt_object_child(dt), key, th, capa);
143 }
144
145 /**
146  * Implementation of dt_it_ops::init.
147  *
148  * Used with regular (non-striped) objects.
149  *
150  * \see dt_it_ops::init() in the API description for details.
151  */
152 static struct dt_it *lod_it_init(const struct lu_env *env,
153                                  struct dt_object *dt, __u32 attr,
154                                  struct lustre_capa *capa)
155 {
156         struct dt_object        *next = dt_object_child(dt);
157         struct lod_it           *it = &lod_env_info(env)->lti_it;
158         struct dt_it            *it_next;
159
160
161         it_next = next->do_index_ops->dio_it.init(env, next, attr, capa);
162         if (IS_ERR(it_next))
163                 return it_next;
164
165         /* currently we do not use more than one iterator per thread
166          * so we store it in thread info. if at some point we need
167          * more active iterators in a single thread, we can allocate
168          * additional ones */
169         LASSERT(it->lit_obj == NULL);
170
171         it->lit_it = it_next;
172         it->lit_obj = next;
173
174         return (struct dt_it *)it;
175 }
176
177 #define LOD_CHECK_IT(env, it)                                   \
178 do {                                                            \
179         LASSERT((it)->lit_obj != NULL);                         \
180         LASSERT((it)->lit_it != NULL);                          \
181 } while (0)
182
183 /**
184  * Implementation of dt_index_operations::dio_it.fini.
185  *
186  * Used with regular (non-striped) objects.
187  *
188  * \see dt_index_operations::dio_it.fini() in the API description for details.
189  */
190 static void lod_it_fini(const struct lu_env *env, struct dt_it *di)
191 {
192         struct lod_it *it = (struct lod_it *)di;
193
194         LOD_CHECK_IT(env, it);
195         it->lit_obj->do_index_ops->dio_it.fini(env, it->lit_it);
196
197         /* the iterator not in use any more */
198         it->lit_obj = NULL;
199         it->lit_it = NULL;
200 }
201
202 /**
203  * Implementation of dt_it_ops::get.
204  *
205  * Used with regular (non-striped) objects.
206  *
207  * \see dt_it_ops::get() in the API description for details.
208  */
209 static int lod_it_get(const struct lu_env *env, struct dt_it *di,
210                       const struct dt_key *key)
211 {
212         const struct lod_it *it = (const struct lod_it *)di;
213
214         LOD_CHECK_IT(env, it);
215         return it->lit_obj->do_index_ops->dio_it.get(env, it->lit_it, key);
216 }
217
218 /**
219  * Implementation of dt_it_ops::put.
220  *
221  * Used with regular (non-striped) objects.
222  *
223  * \see dt_it_ops::put() in the API description for details.
224  */
225 static void lod_it_put(const struct lu_env *env, struct dt_it *di)
226 {
227         struct lod_it *it = (struct lod_it *)di;
228
229         LOD_CHECK_IT(env, it);
230         return it->lit_obj->do_index_ops->dio_it.put(env, it->lit_it);
231 }
232
233 /**
234  * Implementation of dt_it_ops::next.
235  *
236  * Used with regular (non-striped) objects
237  *
238  * \see dt_it_ops::next() in the API description for details.
239  */
240 static int lod_it_next(const struct lu_env *env, struct dt_it *di)
241 {
242         struct lod_it *it = (struct lod_it *)di;
243
244         LOD_CHECK_IT(env, it);
245         return it->lit_obj->do_index_ops->dio_it.next(env, it->lit_it);
246 }
247
248 /**
249  * Implementation of dt_it_ops::key.
250  *
251  * Used with regular (non-striped) objects.
252  *
253  * \see dt_it_ops::key() in the API description for details.
254  */
255 static struct dt_key *lod_it_key(const struct lu_env *env,
256                                  const struct dt_it *di)
257 {
258         const struct lod_it *it = (const struct lod_it *)di;
259
260         LOD_CHECK_IT(env, it);
261         return it->lit_obj->do_index_ops->dio_it.key(env, it->lit_it);
262 }
263
264 /**
265  * Implementation of dt_it_ops::key_size.
266  *
267  * Used with regular (non-striped) objects.
268  *
269  * \see dt_it_ops::key_size() in the API description for details.
270  */
271 static int lod_it_key_size(const struct lu_env *env, const struct dt_it *di)
272 {
273         struct lod_it *it = (struct lod_it *)di;
274
275         LOD_CHECK_IT(env, it);
276         return it->lit_obj->do_index_ops->dio_it.key_size(env, it->lit_it);
277 }
278
279 /**
280  * Implementation of dt_it_ops::rec.
281  *
282  * Used with regular (non-striped) objects.
283  *
284  * \see dt_it_ops::rec() in the API description for details.
285  */
286 static int lod_it_rec(const struct lu_env *env, const struct dt_it *di,
287                       struct dt_rec *rec, __u32 attr)
288 {
289         const struct lod_it *it = (const struct lod_it *)di;
290
291         LOD_CHECK_IT(env, it);
292         return it->lit_obj->do_index_ops->dio_it.rec(env, it->lit_it, rec,
293                                                      attr);
294 }
295
296 /**
297  * Implementation of dt_it_ops::rec_size.
298  *
299  * Used with regular (non-striped) objects.
300  *
301  * \see dt_it_ops::rec_size() in the API description for details.
302  */
303 static int lod_it_rec_size(const struct lu_env *env, const struct dt_it *di,
304                            __u32 attr)
305 {
306         const struct lod_it *it = (const struct lod_it *)di;
307
308         LOD_CHECK_IT(env, it);
309         return it->lit_obj->do_index_ops->dio_it.rec_size(env, it->lit_it,
310                                                           attr);
311 }
312
313 /**
314  * Implementation of dt_it_ops::store.
315  *
316  * Used with regular (non-striped) objects.
317  *
318  * \see dt_it_ops::store() in the API description for details.
319  */
320 static __u64 lod_it_store(const struct lu_env *env, const struct dt_it *di)
321 {
322         const struct lod_it *it = (const struct lod_it *)di;
323
324         LOD_CHECK_IT(env, it);
325         return it->lit_obj->do_index_ops->dio_it.store(env, it->lit_it);
326 }
327
328 /**
329  * Implementation of dt_it_ops::load.
330  *
331  * Used with regular (non-striped) objects.
332  *
333  * \see dt_it_ops::load() in the API description for details.
334  */
335 static int lod_it_load(const struct lu_env *env, const struct dt_it *di,
336                        __u64 hash)
337 {
338         const struct lod_it *it = (const struct lod_it *)di;
339
340         LOD_CHECK_IT(env, it);
341         return it->lit_obj->do_index_ops->dio_it.load(env, it->lit_it, hash);
342 }
343
344 /**
345  * Implementation of dt_it_ops::key_rec.
346  *
347  * Used with regular (non-striped) objects.
348  *
349  * \see dt_it_ops::rec() in the API description for details.
350  */
351 static int lod_it_key_rec(const struct lu_env *env, const struct dt_it *di,
352                           void *key_rec)
353 {
354         const struct lod_it *it = (const struct lod_it *)di;
355
356         LOD_CHECK_IT(env, it);
357         return it->lit_obj->do_index_ops->dio_it.key_rec(env, it->lit_it,
358                                                          key_rec);
359 }
360
361 static struct dt_index_operations lod_index_ops = {
362         .dio_lookup             = lod_index_lookup,
363         .dio_declare_insert     = lod_declare_index_insert,
364         .dio_insert             = lod_index_insert,
365         .dio_declare_delete     = lod_declare_index_delete,
366         .dio_delete             = lod_index_delete,
367         .dio_it = {
368                 .init           = lod_it_init,
369                 .fini           = lod_it_fini,
370                 .get            = lod_it_get,
371                 .put            = lod_it_put,
372                 .next           = lod_it_next,
373                 .key            = lod_it_key,
374                 .key_size       = lod_it_key_size,
375                 .rec            = lod_it_rec,
376                 .rec_size       = lod_it_rec_size,
377                 .store          = lod_it_store,
378                 .load           = lod_it_load,
379                 .key_rec        = lod_it_key_rec,
380         }
381 };
382
383 /**
384  * Implementation of dt_it_ops::init.
385  *
386  * Used with striped objects. Internally just initializes the iterator
387  * on the first stripe.
388  *
389  * \see dt_it_ops::init() in the API description for details.
390  */
391 static struct dt_it *lod_striped_it_init(const struct lu_env *env,
392                                          struct dt_object *dt, __u32 attr,
393                                          struct lustre_capa *capa)
394 {
395         struct lod_object       *lo = lod_dt_obj(dt);
396         struct dt_object        *next;
397         struct lod_it           *it = &lod_env_info(env)->lti_it;
398         struct dt_it            *it_next;
399         ENTRY;
400
401         LASSERT(lo->ldo_stripenr > 0);
402         next = lo->ldo_stripe[0];
403         LASSERT(next != NULL);
404         LASSERT(next->do_index_ops != NULL);
405
406         it_next = next->do_index_ops->dio_it.init(env, next, attr, capa);
407         if (IS_ERR(it_next))
408                 return it_next;
409
410         /* currently we do not use more than one iterator per thread
411          * so we store it in thread info. if at some point we need
412          * more active iterators in a single thread, we can allocate
413          * additional ones */
414         LASSERT(it->lit_obj == NULL);
415
416         it->lit_stripe_index = 0;
417         it->lit_attr = attr;
418         it->lit_it = it_next;
419         it->lit_obj = dt;
420
421         return (struct dt_it *)it;
422 }
423
424 #define LOD_CHECK_STRIPED_IT(env, it, lo)                       \
425 do {                                                            \
426         LASSERT((it)->lit_obj != NULL);                         \
427         LASSERT((it)->lit_it != NULL);                          \
428         LASSERT((lo)->ldo_stripenr > 0);                        \
429         LASSERT((it)->lit_stripe_index < (lo)->ldo_stripenr);   \
430 } while (0)
431
432 /**
433  * Implementation of dt_it_ops::fini.
434  *
435  * Used with striped objects.
436  *
437  * \see dt_it_ops::fini() in the API description for details.
438  */
439 static void lod_striped_it_fini(const struct lu_env *env, struct dt_it *di)
440 {
441         struct lod_it           *it = (struct lod_it *)di;
442         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
443         struct dt_object        *next;
444
445         LOD_CHECK_STRIPED_IT(env, it, lo);
446
447         next = lo->ldo_stripe[it->lit_stripe_index];
448         LASSERT(next != NULL);
449         LASSERT(next->do_index_ops != NULL);
450
451         next->do_index_ops->dio_it.fini(env, it->lit_it);
452
453         /* the iterator not in use any more */
454         it->lit_obj = NULL;
455         it->lit_it = NULL;
456         it->lit_stripe_index = 0;
457 }
458
459 /**
460  * Implementation of dt_it_ops::get.
461  *
462  * Right now it's not used widely, only to reset the iterator to the
463  * initial position. It should be possible to implement a full version
464  * which chooses a correct stripe to be able to position with any key.
465  *
466  * \see dt_it_ops::get() in the API description for details.
467  */
468 static int lod_striped_it_get(const struct lu_env *env, struct dt_it *di,
469                               const struct dt_key *key)
470 {
471         const struct lod_it     *it = (const struct lod_it *)di;
472         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
473         struct dt_object        *next;
474         ENTRY;
475
476         LOD_CHECK_STRIPED_IT(env, it, lo);
477
478         next = lo->ldo_stripe[it->lit_stripe_index];
479         LASSERT(next != NULL);
480         LASSERT(next->do_index_ops != NULL);
481
482         return next->do_index_ops->dio_it.get(env, it->lit_it, key);
483 }
484
485 /**
486  * Implementation of dt_it_ops::put.
487  *
488  * Used with striped objects.
489  *
490  * \see dt_it_ops::put() in the API description for details.
491  */
492 static void lod_striped_it_put(const struct lu_env *env, struct dt_it *di)
493 {
494         struct lod_it           *it = (struct lod_it *)di;
495         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
496         struct dt_object        *next;
497
498         LOD_CHECK_STRIPED_IT(env, it, lo);
499
500         next = lo->ldo_stripe[it->lit_stripe_index];
501         LASSERT(next != NULL);
502         LASSERT(next->do_index_ops != NULL);
503
504         return next->do_index_ops->dio_it.put(env, it->lit_it);
505 }
506
507 /**
508  * Implementation of dt_it_ops::next.
509  *
510  * Used with striped objects. When the end of the current stripe is
511  * reached, the method takes the next stripe's iterator.
512  *
513  * \see dt_it_ops::next() in the API description for details.
514  */
515 static int lod_striped_it_next(const struct lu_env *env, struct dt_it *di)
516 {
517         struct lod_it           *it = (struct lod_it *)di;
518         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
519         struct dt_object        *next;
520         struct dt_it            *it_next;
521         int                     rc;
522         ENTRY;
523
524         LOD_CHECK_STRIPED_IT(env, it, lo);
525
526         next = lo->ldo_stripe[it->lit_stripe_index];
527         LASSERT(next != NULL);
528         LASSERT(next->do_index_ops != NULL);
529 again:
530         rc = next->do_index_ops->dio_it.next(env, it->lit_it);
531         if (rc < 0)
532                 RETURN(rc);
533
534         if (rc == 0 && it->lit_stripe_index == 0)
535                 RETURN(rc);
536
537         if (rc == 0 && it->lit_stripe_index > 0) {
538                 struct lu_dirent *ent;
539
540                 ent = (struct lu_dirent *)lod_env_info(env)->lti_key;
541
542                 rc = next->do_index_ops->dio_it.rec(env, it->lit_it,
543                                                     (struct dt_rec *)ent,
544                                                     it->lit_attr);
545                 if (rc != 0)
546                         RETURN(rc);
547
548                 /* skip . and .. for slave stripe */
549                 if ((strncmp(ent->lde_name, ".",
550                              le16_to_cpu(ent->lde_namelen)) == 0 &&
551                      le16_to_cpu(ent->lde_namelen) == 1) ||
552                     (strncmp(ent->lde_name, "..",
553                              le16_to_cpu(ent->lde_namelen)) == 0 &&
554                      le16_to_cpu(ent->lde_namelen) == 2))
555                         goto again;
556
557                 RETURN(rc);
558         }
559
560         /* go to next stripe */
561         if (it->lit_stripe_index + 1 >= lo->ldo_stripenr)
562                 RETURN(1);
563
564         it->lit_stripe_index++;
565
566         next->do_index_ops->dio_it.put(env, it->lit_it);
567         next->do_index_ops->dio_it.fini(env, it->lit_it);
568
569         rc = next->do_ops->do_index_try(env, next, &dt_directory_features);
570         if (rc != 0)
571                 RETURN(rc);
572
573         next = lo->ldo_stripe[it->lit_stripe_index];
574         LASSERT(next != NULL);
575         LASSERT(next->do_index_ops != NULL);
576
577         it_next = next->do_index_ops->dio_it.init(env, next, it->lit_attr,
578                                                   BYPASS_CAPA);
579         if (!IS_ERR(it_next)) {
580                 it->lit_it = it_next;
581                 goto again;
582         } else {
583                 rc = PTR_ERR(it_next);
584         }
585
586         RETURN(rc);
587 }
588
589 /**
590  * Implementation of dt_it_ops::key.
591  *
592  * Used with striped objects.
593  *
594  * \see dt_it_ops::key() in the API description for details.
595  */
596 static struct dt_key *lod_striped_it_key(const struct lu_env *env,
597                                          const struct dt_it *di)
598 {
599         const struct lod_it     *it = (const struct lod_it *)di;
600         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
601         struct dt_object        *next;
602
603         LOD_CHECK_STRIPED_IT(env, it, lo);
604
605         next = lo->ldo_stripe[it->lit_stripe_index];
606         LASSERT(next != NULL);
607         LASSERT(next->do_index_ops != NULL);
608
609         return next->do_index_ops->dio_it.key(env, it->lit_it);
610 }
611
612 /**
613  * Implementation of dt_it_ops::key_size.
614  *
615  * Used with striped objects.
616  *
617  * \see dt_it_ops::size() in the API description for details.
618  */
619 static int lod_striped_it_key_size(const struct lu_env *env,
620                                    const struct dt_it *di)
621 {
622         struct lod_it           *it = (struct lod_it *)di;
623         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
624         struct dt_object        *next;
625
626         LOD_CHECK_STRIPED_IT(env, it, lo);
627
628         next = lo->ldo_stripe[it->lit_stripe_index];
629         LASSERT(next != NULL);
630         LASSERT(next->do_index_ops != NULL);
631
632         return next->do_index_ops->dio_it.key_size(env, it->lit_it);
633 }
634
635 /**
636  * Implementation of dt_it_ops::rec.
637  *
638  * Used with striped objects.
639  *
640  * \see dt_it_ops::rec() in the API description for details.
641  */
642 static int lod_striped_it_rec(const struct lu_env *env, const struct dt_it *di,
643                               struct dt_rec *rec, __u32 attr)
644 {
645         const struct lod_it     *it = (const struct lod_it *)di;
646         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
647         struct dt_object        *next;
648
649         LOD_CHECK_STRIPED_IT(env, it, lo);
650
651         next = lo->ldo_stripe[it->lit_stripe_index];
652         LASSERT(next != NULL);
653         LASSERT(next->do_index_ops != NULL);
654
655         return next->do_index_ops->dio_it.rec(env, it->lit_it, rec, attr);
656 }
657
658 /**
659  * Implementation of dt_it_ops::rec_size.
660  *
661  * Used with striped objects.
662  *
663  * \see dt_it_ops::rec_size() in the API description for details.
664  */
665 static int lod_striped_it_rec_size(const struct lu_env *env,
666                                    const struct dt_it *di, __u32 attr)
667 {
668         struct lod_it           *it = (struct lod_it *)di;
669         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
670         struct dt_object        *next;
671
672         LOD_CHECK_STRIPED_IT(env, it, lo);
673
674         next = lo->ldo_stripe[it->lit_stripe_index];
675         LASSERT(next != NULL);
676         LASSERT(next->do_index_ops != NULL);
677
678         return next->do_index_ops->dio_it.rec_size(env, it->lit_it, attr);
679 }
680
681 /**
682  * Implementation of dt_it_ops::store.
683  *
684  * Used with striped objects.
685  *
686  * \see dt_it_ops::store() in the API description for details.
687  */
688 static __u64 lod_striped_it_store(const struct lu_env *env,
689                                   const struct dt_it *di)
690 {
691         const struct lod_it     *it = (const struct lod_it *)di;
692         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
693         struct dt_object        *next;
694
695         LOD_CHECK_STRIPED_IT(env, it, lo);
696
697         next = lo->ldo_stripe[it->lit_stripe_index];
698         LASSERT(next != NULL);
699         LASSERT(next->do_index_ops != NULL);
700
701         return next->do_index_ops->dio_it.store(env, it->lit_it);
702 }
703
704 /**
705  * Implementation of dt_it_ops::load.
706  *
707  * Used with striped objects.
708  *
709  * \see dt_it_ops::load() in the API description for details.
710  */
711 static int lod_striped_it_load(const struct lu_env *env,
712                                const struct dt_it *di, __u64 hash)
713 {
714         const struct lod_it     *it = (const struct lod_it *)di;
715         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
716         struct dt_object        *next;
717
718         LOD_CHECK_STRIPED_IT(env, it, lo);
719
720         next = lo->ldo_stripe[it->lit_stripe_index];
721         LASSERT(next != NULL);
722         LASSERT(next->do_index_ops != NULL);
723
724         return next->do_index_ops->dio_it.load(env, it->lit_it, hash);
725 }
726
727 static struct dt_index_operations lod_striped_index_ops = {
728         .dio_lookup             = lod_index_lookup,
729         .dio_declare_insert     = lod_declare_index_insert,
730         .dio_insert             = lod_index_insert,
731         .dio_declare_delete     = lod_declare_index_delete,
732         .dio_delete             = lod_index_delete,
733         .dio_it = {
734                 .init           = lod_striped_it_init,
735                 .fini           = lod_striped_it_fini,
736                 .get            = lod_striped_it_get,
737                 .put            = lod_striped_it_put,
738                 .next           = lod_striped_it_next,
739                 .key            = lod_striped_it_key,
740                 .key_size       = lod_striped_it_key_size,
741                 .rec            = lod_striped_it_rec,
742                 .rec_size       = lod_striped_it_rec_size,
743                 .store          = lod_striped_it_store,
744                 .load           = lod_striped_it_load,
745         }
746 };
747
748 /**
749  * Append the FID for each shard of the striped directory after the
750  * given LMV EA header.
751  *
752  * To simplify striped directory and the consistency verification,
753  * we only store the LMV EA header on disk, for both master object
754  * and slave objects. When someone wants to know the whole LMV EA,
755  * such as client readdir(), we can build the entrie LMV EA on the
756  * MDT side (in RAM) via iterating the sub-directory entries that
757  * are contained in the master object of the stripe directory.
758  *
759  * For the master object of the striped directroy, the valid name
760  * for each shard is composed of the ${shard_FID}:${shard_idx}.
761  *
762  * There may be holes in the LMV EA if some shards' name entries
763  * are corrupted or lost.
764  *
765  * \param[in] env       pointer to the thread context
766  * \param[in] lo        pointer to the master object of the striped directory
767  * \param[in] buf       pointer to the lu_buf which will hold the LMV EA
768  * \param[in] resize    whether re-allocate the buffer if it is not big enough
769  *
770  * \retval              positive size of the LMV EA
771  * \retval              0 for nothing to be loaded
772  * \retval              negative error number on failure
773  */
774 int lod_load_lmv_shards(const struct lu_env *env, struct lod_object *lo,
775                         struct lu_buf *buf, bool resize)
776 {
777         struct lu_dirent        *ent    =
778                         (struct lu_dirent *)lod_env_info(env)->lti_key;
779         struct lod_device       *lod    = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
780         struct dt_object        *obj    = dt_object_child(&lo->ldo_obj);
781         struct lmv_mds_md_v1    *lmv1   = buf->lb_buf;
782         struct dt_it            *it;
783         const struct dt_it_ops  *iops;
784         __u32                    stripes;
785         __u32                    magic  = le32_to_cpu(lmv1->lmv_magic);
786         size_t                   lmv1_size;
787         int                      rc;
788         ENTRY;
789
790         /* If it is not a striped directory, then load nothing. */
791         if (magic != LMV_MAGIC_V1)
792                 RETURN(0);
793
794         /* If it is in migration (or failure), then load nothing. */
795         if (le32_to_cpu(lmv1->lmv_hash_type) & LMV_HASH_FLAG_MIGRATION)
796                 RETURN(0);
797
798         stripes = le32_to_cpu(lmv1->lmv_stripe_count);
799         if (stripes < 1)
800                 RETURN(0);
801
802         rc = lmv_mds_md_size(stripes, magic);
803         if (rc < 0)
804                 RETURN(rc);
805         lmv1_size = rc;
806         if (buf->lb_len < lmv1_size) {
807                 struct lu_buf tbuf;
808
809                 if (!resize)
810                         RETURN(-ERANGE);
811
812                 tbuf = *buf;
813                 buf->lb_buf = NULL;
814                 buf->lb_len = 0;
815                 lu_buf_alloc(buf, lmv1_size);
816                 lmv1 = buf->lb_buf;
817                 if (lmv1 == NULL)
818                         RETURN(-ENOMEM);
819
820                 memcpy(buf->lb_buf, tbuf.lb_buf, tbuf.lb_len);
821         }
822
823         if (unlikely(!dt_try_as_dir(env, obj)))
824                 RETURN(-ENOTDIR);
825
826         memset(&lmv1->lmv_stripe_fids[0], 0, stripes * sizeof(struct lu_fid));
827         iops = &obj->do_index_ops->dio_it;
828         it = iops->init(env, obj, LUDA_64BITHASH, BYPASS_CAPA);
829         if (IS_ERR(it))
830                 RETURN(PTR_ERR(it));
831
832         rc = iops->load(env, it, 0);
833         if (rc == 0)
834                 rc = iops->next(env, it);
835         else if (rc > 0)
836                 rc = 0;
837
838         while (rc == 0) {
839                 char             name[FID_LEN + 2] = "";
840                 struct lu_fid    fid;
841                 __u32            index;
842                 int              len;
843
844                 rc = iops->rec(env, it, (struct dt_rec *)ent, LUDA_64BITHASH);
845                 if (rc != 0)
846                         break;
847
848                 rc = -EIO;
849
850                 fid_le_to_cpu(&fid, &ent->lde_fid);
851                 ent->lde_namelen = le16_to_cpu(ent->lde_namelen);
852                 if (ent->lde_name[0] == '.') {
853                         if (ent->lde_namelen == 1)
854                                 goto next;
855
856                         if (ent->lde_namelen == 2 && ent->lde_name[1] == '.')
857                                 goto next;
858                 }
859
860                 len = snprintf(name, FID_LEN + 1, DFID":", PFID(&ent->lde_fid));
861                 /* The ent->lde_name is composed of ${FID}:${index} */
862                 if (ent->lde_namelen < len + 1 ||
863                     memcmp(ent->lde_name, name, len) != 0) {
864                         CDEBUG(lod->lod_lmv_failout ? D_ERROR : D_INFO,
865                                "%s: invalid shard name %.*s with the FID "DFID
866                                " for the striped directory "DFID", %s\n",
867                                lod2obd(lod)->obd_name, ent->lde_namelen,
868                                ent->lde_name, PFID(&fid),
869                                PFID(lu_object_fid(&obj->do_lu)),
870                                lod->lod_lmv_failout ? "failout" : "skip");
871
872                         if (lod->lod_lmv_failout)
873                                 break;
874
875                         goto next;
876                 }
877
878                 index = 0;
879                 do {
880                         if (ent->lde_name[len] < '0' ||
881                             ent->lde_name[len] > '9') {
882                                 CDEBUG(lod->lod_lmv_failout ? D_ERROR : D_INFO,
883                                        "%s: invalid shard name %.*s with the "
884                                        "FID "DFID" for the striped directory "
885                                        DFID", %s\n",
886                                        lod2obd(lod)->obd_name, ent->lde_namelen,
887                                        ent->lde_name, PFID(&fid),
888                                        PFID(lu_object_fid(&obj->do_lu)),
889                                        lod->lod_lmv_failout ?
890                                        "failout" : "skip");
891
892                                 if (lod->lod_lmv_failout)
893                                         break;
894
895                                 goto next;
896                         }
897
898                         index = index * 10 + ent->lde_name[len++] - '0';
899                 } while (len < ent->lde_namelen);
900
901                 if (len == ent->lde_namelen) {
902                         /* Out of LMV EA range. */
903                         if (index >= stripes) {
904                                 CERROR("%s: the shard %.*s for the striped "
905                                        "directory "DFID" is out of the known "
906                                        "LMV EA range [0 - %u], failout\n",
907                                        lod2obd(lod)->obd_name, ent->lde_namelen,
908                                        ent->lde_name,
909                                        PFID(lu_object_fid(&obj->do_lu)),
910                                        stripes - 1);
911
912                                 break;
913                         }
914
915                         /* The slot has been occupied. */
916                         if (!fid_is_zero(&lmv1->lmv_stripe_fids[index])) {
917                                 struct lu_fid fid0;
918
919                                 fid_le_to_cpu(&fid0,
920                                         &lmv1->lmv_stripe_fids[index]);
921                                 CERROR("%s: both the shard "DFID" and "DFID
922                                        " for the striped directory "DFID
923                                        " claim the same LMV EA slot at the "
924                                        "index %d, failout\n",
925                                        lod2obd(lod)->obd_name,
926                                        PFID(&fid0), PFID(&fid),
927                                        PFID(lu_object_fid(&obj->do_lu)), index);
928
929                                 break;
930                         }
931
932                         /* stored as LE mode */
933                         lmv1->lmv_stripe_fids[index] = ent->lde_fid;
934
935 next:
936                         rc = iops->next(env, it);
937                 }
938         }
939
940         iops->put(env, it);
941         iops->fini(env, it);
942
943         RETURN(rc > 0 ? lmv_mds_md_size(stripes, magic) : rc);
944 }
945
946 /**
947  * Implementation of dt_object_operations::do_index_try.
948  *
949  * \see dt_object_operations::do_index_try() in the API description for details.
950  */
951 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
952                          const struct dt_index_features *feat)
953 {
954         struct lod_object       *lo = lod_dt_obj(dt);
955         struct dt_object        *next = dt_object_child(dt);
956         int                     rc;
957         ENTRY;
958
959         LASSERT(next->do_ops);
960         LASSERT(next->do_ops->do_index_try);
961
962         rc = lod_load_striping_locked(env, lo);
963         if (rc != 0)
964                 RETURN(rc);
965
966         rc = next->do_ops->do_index_try(env, next, feat);
967         if (rc != 0)
968                 RETURN(rc);
969
970         if (lo->ldo_stripenr > 0) {
971                 int i;
972
973                 for (i = 0; i < lo->ldo_stripenr; i++) {
974                         if (dt_object_exists(lo->ldo_stripe[i]) == 0)
975                                 continue;
976                         rc = lo->ldo_stripe[i]->do_ops->do_index_try(env,
977                                                 lo->ldo_stripe[i], feat);
978                         if (rc != 0)
979                                 RETURN(rc);
980                 }
981                 dt->do_index_ops = &lod_striped_index_ops;
982         } else {
983                 dt->do_index_ops = &lod_index_ops;
984         }
985
986         RETURN(rc);
987 }
988
989 /**
990  * Implementation of dt_object_operations::do_read_lock.
991  *
992  * \see dt_object_operations::do_read_lock() in the API description for details.
993  */
994 static void lod_object_read_lock(const struct lu_env *env,
995                                  struct dt_object *dt, unsigned role)
996 {
997         dt_read_lock(env, dt_object_child(dt), role);
998 }
999
1000 /**
1001  * Implementation of dt_object_operations::do_write_lock.
1002  *
1003  * \see dt_object_operations::do_write_lock() in the API description for
1004  * details.
1005  */
1006 static void lod_object_write_lock(const struct lu_env *env,
1007                                   struct dt_object *dt, unsigned role)
1008 {
1009         dt_write_lock(env, dt_object_child(dt), role);
1010 }
1011
1012 /**
1013  * Implementation of dt_object_operations::do_read_unlock.
1014  *
1015  * \see dt_object_operations::do_read_unlock() in the API description for
1016  * details.
1017  */
1018 static void lod_object_read_unlock(const struct lu_env *env,
1019                                    struct dt_object *dt)
1020 {
1021         dt_read_unlock(env, dt_object_child(dt));
1022 }
1023
1024 /**
1025  * Implementation of dt_object_operations::do_write_unlock.
1026  *
1027  * \see dt_object_operations::do_write_unlock() in the API description for
1028  * details.
1029  */
1030 static void lod_object_write_unlock(const struct lu_env *env,
1031                                     struct dt_object *dt)
1032 {
1033         dt_write_unlock(env, dt_object_child(dt));
1034 }
1035
1036 /**
1037  * Implementation of dt_object_operations::do_write_locked.
1038  *
1039  * \see dt_object_operations::do_write_locked() in the API description for
1040  * details.
1041  */
1042 static int lod_object_write_locked(const struct lu_env *env,
1043                                    struct dt_object *dt)
1044 {
1045         return dt_write_locked(env, dt_object_child(dt));
1046 }
1047
1048 /**
1049  * Implementation of dt_object_operations::do_attr_get.
1050  *
1051  * \see dt_object_operations::do_attr_get() in the API description for details.
1052  */
1053 static int lod_attr_get(const struct lu_env *env,
1054                         struct dt_object *dt,
1055                         struct lu_attr *attr,
1056                         struct lustre_capa *capa)
1057 {
1058         /* Note: for striped directory, client will merge attributes
1059          * from all of the sub-stripes see lmv_merge_attr(), and there
1060          * no MDD logic depend on directory nlink/size/time, so we can
1061          * always use master inode nlink and size for now. */
1062         return dt_attr_get(env, dt_object_child(dt), attr, capa);
1063 }
1064
1065 /**
1066  * Mark all of the striped directory sub-stripes dead.
1067  *
1068  * When a striped object is a subject to removal, we have
1069  * to mark all the stripes to prevent further access to
1070  * them (e.g. create a new file in those). So we mark
1071  * all the stripes with LMV_HASH_FLAG_DEAD. The function
1072  * can be used to declare the changes and to apply them.
1073  * If the object isn't striped, then just return success.
1074  *
1075  * \param[in] env       execution environment
1076  * \param[in] dt        the striped object
1077  * \param[in] handle    transaction handle
1078  * \param[in] declare   whether to declare the change or apply
1079  *
1080  * \retval              0 on success
1081  * \retval              negative if failed
1082  **/
1083 static int lod_mark_dead_object(const struct lu_env *env,
1084                                 struct dt_object *dt,
1085                                 struct thandle *handle,
1086                                 bool declare)
1087 {
1088         struct lod_object       *lo = lod_dt_obj(dt);
1089         struct lmv_mds_md_v1    *lmv;
1090         __u32                   dead_hash_type;
1091         int                     rc;
1092         int                     i;
1093
1094         ENTRY;
1095
1096         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
1097                 RETURN(0);
1098
1099         rc = lod_load_striping_locked(env, lo);
1100         if (rc != 0)
1101                 RETURN(rc);
1102
1103         if (lo->ldo_stripenr == 0)
1104                 RETURN(0);
1105
1106         rc = lod_get_lmv_ea(env, lo);
1107         if (rc <= 0)
1108                 RETURN(rc);
1109
1110         lmv = lod_env_info(env)->lti_ea_store;
1111         lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_STRIPE);
1112         dead_hash_type = le32_to_cpu(lmv->lmv_hash_type) | LMV_HASH_FLAG_DEAD;
1113         lmv->lmv_hash_type = cpu_to_le32(dead_hash_type);
1114         for (i = 0; i < lo->ldo_stripenr; i++) {
1115                 struct lu_buf buf;
1116
1117                 lmv->lmv_master_mdt_index = i;
1118                 buf.lb_buf = lmv;
1119                 buf.lb_len = sizeof(*lmv);
1120                 if (declare) {
1121                         rc = dt_declare_xattr_set(env, lo->ldo_stripe[i], &buf,
1122                                                   XATTR_NAME_LMV,
1123                                                   LU_XATTR_REPLACE, handle);
1124                 } else {
1125                         rc = dt_xattr_set(env, lo->ldo_stripe[i], &buf,
1126                                           XATTR_NAME_LMV, LU_XATTR_REPLACE,
1127                                           handle, BYPASS_CAPA);
1128                 }
1129                 if (rc != 0)
1130                         break;
1131         }
1132
1133         RETURN(rc);
1134 }
1135
1136 /**
1137  * Implementation of dt_object_operations::do_declare_attr_set.
1138  *
1139  * If the object is striped, then apply the changes to all the stripes.
1140  *
1141  * \see dt_object_operations::do_declare_attr_set() in the API description
1142  * for details.
1143  */
1144 static int lod_declare_attr_set(const struct lu_env *env,
1145                                 struct dt_object *dt,
1146                                 const struct lu_attr *attr,
1147                                 struct thandle *handle)
1148 {
1149         struct dt_object  *next = dt_object_child(dt);
1150         struct lod_object *lo = lod_dt_obj(dt);
1151         int                rc, i;
1152         ENTRY;
1153
1154         /* Set dead object on all other stripes */
1155         if (attr->la_valid & LA_FLAGS && !(attr->la_valid & ~LA_FLAGS) &&
1156             attr->la_flags & LUSTRE_SLAVE_DEAD_FL) {
1157                 rc = lod_mark_dead_object(env, dt, handle, true);
1158                 RETURN(rc);
1159         }
1160
1161         /*
1162          * declare setattr on the local object
1163          */
1164         rc = dt_declare_attr_set(env, next, attr, handle);
1165         if (rc)
1166                 RETURN(rc);
1167
1168         /* osp_declare_attr_set() ignores all attributes other than
1169          * UID, GID, and size, and osp_attr_set() ignores all but UID
1170          * and GID.  Declaration of size attr setting happens through
1171          * lod_declare_init_size(), and not through this function.
1172          * Therefore we need not load striping unless ownership is
1173          * changing.  This should save memory and (we hope) speed up
1174          * rename(). */
1175         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1176                 if (!(attr->la_valid & (LA_UID | LA_GID)))
1177                         RETURN(rc);
1178
1179                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
1180                         RETURN(0);
1181         } else {
1182                 if (!(attr->la_valid & (LA_UID | LA_GID | LA_MODE |
1183                                         LA_ATIME | LA_MTIME | LA_CTIME)))
1184                         RETURN(rc);
1185         }
1186         /*
1187          * load striping information, notice we don't do this when object
1188          * is being initialized as we don't need this information till
1189          * few specific cases like destroy, chown
1190          */
1191         rc = lod_load_striping(env, lo);
1192         if (rc)
1193                 RETURN(rc);
1194
1195         if (lo->ldo_stripenr == 0)
1196                 RETURN(0);
1197
1198         /*
1199          * if object is striped declare changes on the stripes
1200          */
1201         LASSERT(lo->ldo_stripe);
1202         for (i = 0; i < lo->ldo_stripenr; i++) {
1203                 if (likely(lo->ldo_stripe[i] != NULL)) {
1204                         rc = dt_declare_attr_set(env, lo->ldo_stripe[i], attr,
1205                                                  handle);
1206                         if (rc != 0) {
1207                                 CERROR("failed declaration: %d\n", rc);
1208                                 break;
1209                         }
1210                 }
1211         }
1212
1213         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE) &&
1214             dt_object_exists(next) != 0 &&
1215             dt_object_remote(next) == 0)
1216                 dt_declare_xattr_del(env, next, XATTR_NAME_LOV, handle);
1217
1218         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE) &&
1219             dt_object_exists(next) &&
1220             dt_object_remote(next) == 0 && S_ISREG(attr->la_mode)) {
1221                 struct lod_thread_info *info = lod_env_info(env);
1222                 struct lu_buf *buf = &info->lti_buf;
1223
1224                 buf->lb_buf = info->lti_ea_store;
1225                 buf->lb_len = info->lti_ea_store_size;
1226                 dt_declare_xattr_set(env, next, buf, XATTR_NAME_LOV,
1227                                      LU_XATTR_REPLACE, handle);
1228         }
1229
1230         RETURN(rc);
1231 }
1232
1233 /**
1234  * Implementation of dt_object_operations::do_attr_set.
1235  *
1236  * If the object is striped, then apply the changes to all or subset of
1237  * the stripes depending on the object type and specific attributes.
1238  *
1239  * \see dt_object_operations::do_attr_set() in the API description for details.
1240  */
1241 static int lod_attr_set(const struct lu_env *env,
1242                         struct dt_object *dt,
1243                         const struct lu_attr *attr,
1244                         struct thandle *handle,
1245                         struct lustre_capa *capa)
1246 {
1247         struct dt_object        *next = dt_object_child(dt);
1248         struct lod_object       *lo = lod_dt_obj(dt);
1249         int                     rc, i;
1250         ENTRY;
1251
1252         /* Set dead object on all other stripes */
1253         if (attr->la_valid & LA_FLAGS && !(attr->la_valid & ~LA_FLAGS) &&
1254             attr->la_flags & LUSTRE_SLAVE_DEAD_FL) {
1255                 rc = lod_mark_dead_object(env, dt, handle, false);
1256                 RETURN(rc);
1257         }
1258
1259         /*
1260          * apply changes to the local object
1261          */
1262         rc = dt_attr_set(env, next, attr, handle, capa);
1263         if (rc)
1264                 RETURN(rc);
1265
1266         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1267                 if (!(attr->la_valid & (LA_UID | LA_GID)))
1268                         RETURN(rc);
1269
1270                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
1271                         RETURN(0);
1272         } else {
1273                 if (!(attr->la_valid & (LA_UID | LA_GID | LA_MODE |
1274                                         LA_ATIME | LA_MTIME | LA_CTIME)))
1275                         RETURN(rc);
1276         }
1277
1278         if (lo->ldo_stripenr == 0)
1279                 RETURN(0);
1280
1281         /*
1282          * if object is striped, apply changes to all the stripes
1283          */
1284         LASSERT(lo->ldo_stripe);
1285         for (i = 0; i < lo->ldo_stripenr; i++) {
1286                 if (unlikely(lo->ldo_stripe[i] == NULL))
1287                         continue;
1288                 if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
1289                     (dt_object_exists(lo->ldo_stripe[i]) == 0))
1290                         continue;
1291
1292                 rc = dt_attr_set(env, lo->ldo_stripe[i], attr, handle, capa);
1293                 if (rc != 0) {
1294                         CERROR("failed declaration: %d\n", rc);
1295                         break;
1296                 }
1297         }
1298
1299         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE) &&
1300             dt_object_exists(next) != 0 &&
1301             dt_object_remote(next) == 0)
1302                 dt_xattr_del(env, next, XATTR_NAME_LOV, handle, BYPASS_CAPA);
1303
1304         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE) &&
1305             dt_object_exists(next) &&
1306             dt_object_remote(next) == 0 && S_ISREG(attr->la_mode)) {
1307                 struct lod_thread_info *info = lod_env_info(env);
1308                 struct lu_buf *buf = &info->lti_buf;
1309                 struct ost_id *oi = &info->lti_ostid;
1310                 struct lu_fid *fid = &info->lti_fid;
1311                 struct lov_mds_md_v1 *lmm;
1312                 struct lov_ost_data_v1 *objs;
1313                 __u32 magic;
1314                 int rc1;
1315
1316                 rc1 = lod_get_lov_ea(env, lo);
1317                 if (rc1  <= 0)
1318                         RETURN(rc);
1319
1320                 buf->lb_buf = info->lti_ea_store;
1321                 buf->lb_len = info->lti_ea_store_size;
1322                 lmm = info->lti_ea_store;
1323                 magic = le32_to_cpu(lmm->lmm_magic);
1324                 if (magic == LOV_MAGIC_V1)
1325                         objs = &(lmm->lmm_objects[0]);
1326                 else
1327                         objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
1328                 ostid_le_to_cpu(&objs->l_ost_oi, oi);
1329                 ostid_to_fid(fid, oi, le32_to_cpu(objs->l_ost_idx));
1330                 fid->f_oid--;
1331                 fid_to_ostid(fid, oi);
1332                 ostid_cpu_to_le(oi, &objs->l_ost_oi);
1333                 dt_xattr_set(env, next, buf, XATTR_NAME_LOV,
1334                              LU_XATTR_REPLACE, handle, BYPASS_CAPA);
1335         }
1336
1337         RETURN(rc);
1338 }
1339
1340 /**
1341  * Implementation of dt_object_operations::do_xattr_get.
1342  *
1343  * If LOV EA is requested from the root object and it's not
1344  * found, then return default striping for the filesystem.
1345  *
1346  * \see dt_object_operations::do_xattr_get() in the API description for details.
1347  */
1348 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
1349                          struct lu_buf *buf, const char *name,
1350                          struct lustre_capa *capa)
1351 {
1352         struct lod_thread_info  *info = lod_env_info(env);
1353         struct lod_device       *dev = lu2lod_dev(dt->do_lu.lo_dev);
1354         int                      rc, is_root;
1355         ENTRY;
1356
1357         rc = dt_xattr_get(env, dt_object_child(dt), buf, name, capa);
1358         if (strcmp(name, XATTR_NAME_LMV) == 0) {
1359                 struct lmv_mds_md_v1    *lmv1;
1360                 int                      rc1 = 0;
1361
1362                 if (rc > (typeof(rc))sizeof(*lmv1))
1363                         RETURN(rc);
1364
1365                 if (rc < (typeof(rc))sizeof(*lmv1))
1366                         RETURN(rc = rc > 0 ? -EINVAL : rc);
1367
1368                 if (buf->lb_buf == NULL || buf->lb_len == 0) {
1369                         CLASSERT(sizeof(*lmv1) <= sizeof(info->lti_key));
1370
1371                         info->lti_buf.lb_buf = info->lti_key;
1372                         info->lti_buf.lb_len = sizeof(*lmv1);
1373                         rc = dt_xattr_get(env, dt_object_child(dt),
1374                                           &info->lti_buf, name, capa);
1375                         if (unlikely(rc != sizeof(*lmv1)))
1376                                 RETURN(rc = rc > 0 ? -EINVAL : rc);
1377
1378                         lmv1 = info->lti_buf.lb_buf;
1379                         /* The on-disk LMV EA only contains header, but the
1380                          * returned LMV EA size should contain the space for
1381                          * the FIDs of all shards of the striped directory. */
1382                         if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_V1)
1383                                 rc = lmv_mds_md_size(
1384                                         le32_to_cpu(lmv1->lmv_stripe_count),
1385                                         LMV_MAGIC_V1);
1386                 } else {
1387                         rc1 = lod_load_lmv_shards(env, lod_dt_obj(dt),
1388                                                   buf, false);
1389                 }
1390
1391                 RETURN(rc = rc1 != 0 ? rc1 : rc);
1392         }
1393
1394         if (rc != -ENODATA || !S_ISDIR(dt->do_lu.lo_header->loh_attr & S_IFMT))
1395                 RETURN(rc);
1396
1397         /*
1398          * lod returns default striping on the real root of the device
1399          * this is like the root stores default striping for the whole
1400          * filesystem. historically we've been using a different approach
1401          * and store it in the config.
1402          */
1403         dt_root_get(env, dev->lod_child, &info->lti_fid);
1404         is_root = lu_fid_eq(&info->lti_fid, lu_object_fid(&dt->do_lu));
1405
1406         if (is_root && strcmp(XATTR_NAME_LOV, name) == 0) {
1407                 struct lov_user_md *lum = buf->lb_buf;
1408                 struct lov_desc    *desc = &dev->lod_desc;
1409
1410                 if (buf->lb_buf == NULL) {
1411                         rc = sizeof(*lum);
1412                 } else if (buf->lb_len >= sizeof(*lum)) {
1413                         lum->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V1);
1414                         lmm_oi_set_seq(&lum->lmm_oi, FID_SEQ_LOV_DEFAULT);
1415                         lmm_oi_set_id(&lum->lmm_oi, 0);
1416                         lmm_oi_cpu_to_le(&lum->lmm_oi, &lum->lmm_oi);
1417                         lum->lmm_pattern = cpu_to_le32(desc->ld_pattern);
1418                         lum->lmm_stripe_size = cpu_to_le32(
1419                                                 desc->ld_default_stripe_size);
1420                         lum->lmm_stripe_count = cpu_to_le16(
1421                                                 desc->ld_default_stripe_count);
1422                         lum->lmm_stripe_offset = cpu_to_le16(
1423                                                 desc->ld_default_stripe_offset);
1424                         rc = sizeof(*lum);
1425                 } else {
1426                         rc = -ERANGE;
1427                 }
1428         }
1429
1430         RETURN(rc);
1431 }
1432
1433 /**
1434  * Verify LVM EA.
1435  *
1436  * Checks that the magic of the stripe is sane.
1437  *
1438  * \param[in] lod       lod device
1439  * \param[in] lum       a buffer storing LMV EA to verify
1440  *
1441  * \retval              0 if the EA is sane
1442  * \retval              negative otherwise
1443  */
1444 static int lod_verify_md_striping(struct lod_device *lod,
1445                                   const struct lmv_user_md_v1 *lum)
1446 {
1447         if (unlikely(le32_to_cpu(lum->lum_magic) != LMV_USER_MAGIC)) {
1448                 CERROR("%s: invalid lmv_user_md: magic = %x, "
1449                        "stripe_offset = %d, stripe_count = %u: rc = %d\n",
1450                        lod2obd(lod)->obd_name, le32_to_cpu(lum->lum_magic),
1451                        (int)le32_to_cpu(lum->lum_stripe_offset),
1452                        le32_to_cpu(lum->lum_stripe_count), -EINVAL);
1453                 return -EINVAL;
1454         }
1455
1456         return 0;
1457 }
1458
1459 /**
1460  * Initialize LMV EA for a slave.
1461  *
1462  * Initialize slave's LMV EA from the master's LMV EA.
1463  *
1464  * \param[in] master_lmv        a buffer containing master's EA
1465  * \param[out] slave_lmv        a buffer where slave's EA will be stored
1466  *
1467  */
1468 static void lod_prep_slave_lmv_md(struct lmv_mds_md_v1 *slave_lmv,
1469                                   const struct lmv_mds_md_v1 *master_lmv)
1470 {
1471         *slave_lmv = *master_lmv;
1472         slave_lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_STRIPE);
1473 }
1474
1475 /**
1476  * Generate LMV EA.
1477  *
1478  * Generate LMV EA from the object passed as \a dt. The object must have
1479  * the stripes created and initialized.
1480  *
1481  * \param[in] env       execution environment
1482  * \param[in] dt        object
1483  * \param[out] lmv_buf  buffer storing generated LMV EA
1484  *
1485  * \retval              0 on success
1486  * \retval              negative if failed
1487  */
1488 static int lod_prep_lmv_md(const struct lu_env *env, struct dt_object *dt,
1489                            struct lu_buf *lmv_buf)
1490 {
1491         struct lod_thread_info  *info = lod_env_info(env);
1492         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1493         struct lod_object       *lo = lod_dt_obj(dt);
1494         struct lmv_mds_md_v1    *lmm1;
1495         int                     stripe_count;
1496         int                     type = LU_SEQ_RANGE_ANY;
1497         int                     rc;
1498         __u32                   mdtidx;
1499         ENTRY;
1500
1501         LASSERT(lo->ldo_dir_striped != 0);
1502         LASSERT(lo->ldo_stripenr > 0);
1503         stripe_count = lo->ldo_stripenr;
1504         /* Only store the LMV EA heahder on the disk. */
1505         if (info->lti_ea_store_size < sizeof(*lmm1)) {
1506                 rc = lod_ea_store_resize(info, sizeof(*lmm1));
1507                 if (rc != 0)
1508                         RETURN(rc);
1509         } else {
1510                 memset(info->lti_ea_store, 0, sizeof(*lmm1));
1511         }
1512
1513         lmm1 = (struct lmv_mds_md_v1 *)info->lti_ea_store;
1514         lmm1->lmv_magic = cpu_to_le32(LMV_MAGIC);
1515         lmm1->lmv_stripe_count = cpu_to_le32(stripe_count);
1516         lmm1->lmv_hash_type = cpu_to_le32(lo->ldo_dir_hash_type);
1517         rc = lod_fld_lookup(env, lod, lu_object_fid(&dt->do_lu),
1518                             &mdtidx, &type);
1519         if (rc != 0)
1520                 RETURN(rc);
1521
1522         lmm1->lmv_master_mdt_index = cpu_to_le32(mdtidx);
1523         lmv_buf->lb_buf = info->lti_ea_store;
1524         lmv_buf->lb_len = sizeof(*lmm1);
1525
1526         RETURN(rc);
1527 }
1528
1529 /**
1530  * Create in-core represenation for a striped directory.
1531  *
1532  * Parse the buffer containing LMV EA and instantiate LU objects
1533  * representing the stripe objects. The pointers to the objects are
1534  * stored in ldo_stripe field of \a lo. This function is used when
1535  * we need to access an already created object (i.e. load from a disk).
1536  *
1537  * \param[in] env       execution environment
1538  * \param[in] lo        lod object
1539  * \param[in] buf       buffer containing LMV EA
1540  *
1541  * \retval              0 on success
1542  * \retval              negative if failed
1543  */
1544 int lod_parse_dir_striping(const struct lu_env *env, struct lod_object *lo,
1545                            const struct lu_buf *buf)
1546 {
1547         struct lod_thread_info  *info = lod_env_info(env);
1548         struct lod_device       *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1549         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
1550         struct dt_object        **stripe;
1551         union lmv_mds_md        *lmm = buf->lb_buf;
1552         struct lmv_mds_md_v1    *lmv1 = &lmm->lmv_md_v1;
1553         struct lu_fid           *fid = &info->lti_fid;
1554         unsigned int            i;
1555         int                     rc = 0;
1556         ENTRY;
1557
1558         if (le32_to_cpu(lmv1->lmv_hash_type) & LMV_HASH_FLAG_MIGRATION)
1559                 RETURN(0);
1560
1561         if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_STRIPE) {
1562                 lo->ldo_dir_slave_stripe = 1;
1563                 RETURN(0);
1564         }
1565
1566         if (le32_to_cpu(lmv1->lmv_magic) != LMV_MAGIC_V1)
1567                 RETURN(-EINVAL);
1568
1569         if (le32_to_cpu(lmv1->lmv_stripe_count) < 1)
1570                 RETURN(0);
1571
1572         LASSERT(lo->ldo_stripe == NULL);
1573         OBD_ALLOC(stripe, sizeof(stripe[0]) *
1574                   (le32_to_cpu(lmv1->lmv_stripe_count)));
1575         if (stripe == NULL)
1576                 RETURN(-ENOMEM);
1577
1578         for (i = 0; i < le32_to_cpu(lmv1->lmv_stripe_count); i++) {
1579                 struct dt_device        *tgt_dt;
1580                 struct dt_object        *dto;
1581                 int                     type = LU_SEQ_RANGE_ANY;
1582                 __u32                   idx;
1583
1584                 fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[i]);
1585                 if (!fid_is_sane(fid))
1586                         GOTO(out, rc = -ESTALE);
1587
1588                 rc = lod_fld_lookup(env, lod, fid, &idx, &type);
1589                 if (rc != 0)
1590                         GOTO(out, rc);
1591
1592                 if (idx == lod2lu_dev(lod)->ld_site->ld_seq_site->ss_node_id) {
1593                         tgt_dt = lod->lod_child;
1594                 } else {
1595                         struct lod_tgt_desc     *tgt;
1596
1597                         tgt = LTD_TGT(ltd, idx);
1598                         if (tgt == NULL)
1599                                 GOTO(out, rc = -ESTALE);
1600                         tgt_dt = tgt->ltd_tgt;
1601                 }
1602
1603                 dto = dt_locate_at(env, tgt_dt, fid,
1604                                   lo->ldo_obj.do_lu.lo_dev->ld_site->ls_top_dev,
1605                                   NULL);
1606                 if (IS_ERR(dto))
1607                         GOTO(out, rc = PTR_ERR(dto));
1608
1609                 stripe[i] = dto;
1610         }
1611 out:
1612         lo->ldo_stripe = stripe;
1613         lo->ldo_stripenr = le32_to_cpu(lmv1->lmv_stripe_count);
1614         lo->ldo_stripes_allocated = le32_to_cpu(lmv1->lmv_stripe_count);
1615         if (rc != 0)
1616                 lod_object_free_striping(env, lo);
1617
1618         RETURN(rc);
1619 }
1620
1621 /**
1622  * Create a striped directory.
1623  *
1624  * Create a striped directory with a given stripe pattern on the specified MDTs.
1625  * A striped directory is represented as a regular directory - an index listing
1626  * all the stripes. The stripes point back to the master object with ".." and
1627  * LinkEA. The master object gets LMV EA which identifies it as a striped
1628  * directory. The function allocates FIDs for all the stripes.
1629  *
1630  * \param[in] env       execution environment
1631  * \param[in] dt        object
1632  * \param[in] attr      attributes to initialize the objects with
1633  * \param[in] lum       a pattern specifying the number of stripes and
1634  *                      MDT to start from
1635  * \param[in] dof       type of objects to be created
1636  * \param[in] th        transaction handle
1637  *
1638  * \retval              0 on success
1639  * \retval              negative if failed
1640  */
1641 static int lod_prep_md_striped_create(const struct lu_env *env,
1642                                       struct dt_object *dt,
1643                                       struct lu_attr *attr,
1644                                       const struct lmv_user_md_v1 *lum,
1645                                       struct dt_object_format *dof,
1646                                       struct thandle *th)
1647 {
1648         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1649         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
1650         struct lod_object       *lo = lod_dt_obj(dt);
1651         struct lod_thread_info  *info = lod_env_info(env);
1652         struct dt_object        **stripe;
1653         struct lu_buf           lmv_buf;
1654         struct lu_buf           slave_lmv_buf;
1655         struct lmv_mds_md_v1    *lmm;
1656         struct lmv_mds_md_v1    *slave_lmm = NULL;
1657         struct dt_insert_rec    *rec = &info->lti_dt_rec;
1658         __u32                   stripe_count;
1659         int                     *idx_array;
1660         int                     rc = 0;
1661         __u32                   i;
1662         __u32                   j;
1663         ENTRY;
1664
1665         /* The lum has been verifed in lod_verify_md_striping */
1666         LASSERT(le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC);
1667         LASSERT(le32_to_cpu(lum->lum_stripe_count) > 0);
1668
1669         stripe_count = le32_to_cpu(lum->lum_stripe_count);
1670
1671         /* shrink the stripe_count to the avaible MDT count */
1672         if (stripe_count > lod->lod_remote_mdt_count + 1)
1673                 stripe_count = lod->lod_remote_mdt_count + 1;
1674
1675         OBD_ALLOC(stripe, sizeof(stripe[0]) * stripe_count);
1676         if (stripe == NULL)
1677                 RETURN(-ENOMEM);
1678
1679         OBD_ALLOC(idx_array, sizeof(idx_array[0]) * stripe_count);
1680         if (idx_array == NULL)
1681                 GOTO(out_free, rc = -ENOMEM);
1682
1683         for (i = 0; i < stripe_count; i++) {
1684                 struct lod_tgt_desc     *tgt = NULL;
1685                 struct dt_object        *dto;
1686                 struct lu_fid           fid = { 0 };
1687                 int                     idx;
1688                 struct lu_object_conf   conf = { 0 };
1689                 struct dt_device        *tgt_dt = NULL;
1690
1691                 if (i == 0) {
1692                         /* Right now, master stripe and master object are
1693                          * on the same MDT */
1694                         idx = le32_to_cpu(lum->lum_stripe_offset);
1695                         rc = obd_fid_alloc(env, lod->lod_child_exp, &fid,
1696                                            NULL);
1697                         if (rc < 0)
1698                                 GOTO(out_put, rc);
1699                         tgt_dt = lod->lod_child;
1700                         goto next;
1701                 }
1702
1703                 idx = (idx_array[i - 1] + 1) % (lod->lod_remote_mdt_count + 1);
1704
1705                 for (j = 0; j < lod->lod_remote_mdt_count;
1706                      j++, idx = (idx + 1) % (lod->lod_remote_mdt_count + 1)) {
1707                         bool already_allocated = false;
1708                         __u32 k;
1709
1710                         CDEBUG(D_INFO, "try idx %d, mdt cnt %u,"
1711                                " allocated %u, last allocated %d\n", idx,
1712                                lod->lod_remote_mdt_count, i, idx_array[i - 1]);
1713
1714                         /* Find next available target */
1715                         if (!cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx))
1716                                 continue;
1717
1718                         /* check whether the idx already exists
1719                          * in current allocated array */
1720                         for (k = 0; k < i; k++) {
1721                                 if (idx_array[k] == idx) {
1722                                         already_allocated = true;
1723                                         break;
1724                                 }
1725                         }
1726
1727                         if (already_allocated)
1728                                 continue;
1729
1730                         /* check the status of the OSP */
1731                         tgt = LTD_TGT(ltd, idx);
1732                         if (tgt == NULL)
1733                                 continue;
1734
1735                         tgt_dt = tgt->ltd_tgt;
1736                         rc = dt_statfs(env, tgt_dt, NULL);
1737                         if (rc) {
1738                                 /* this OSP doesn't feel well */
1739                                 rc = 0;
1740                                 continue;
1741                         }
1742
1743                         rc = obd_fid_alloc(env, tgt->ltd_exp, &fid, NULL);
1744                         if (rc < 0) {
1745                                 rc = 0;
1746                                 continue;
1747                         }
1748
1749                         break;
1750                 }
1751
1752                 /* Can not allocate more stripes */
1753                 if (j == lod->lod_remote_mdt_count) {
1754                         CDEBUG(D_INFO, "%s: require stripes %u only get %d\n",
1755                                lod2obd(lod)->obd_name, stripe_count, i - 1);
1756                         break;
1757                 }
1758
1759                 CDEBUG(D_INFO, "idx %d, mdt cnt %u,"
1760                        " allocated %u, last allocated %d\n", idx,
1761                        lod->lod_remote_mdt_count, i, idx_array[i - 1]);
1762
1763 next:
1764                 /* tgt_dt and fid must be ready after search avaible OSP
1765                  * in the above loop */
1766                 LASSERT(tgt_dt != NULL);
1767                 LASSERT(fid_is_sane(&fid));
1768                 conf.loc_flags = LOC_F_NEW;
1769                 dto = dt_locate_at(env, tgt_dt, &fid,
1770                                    dt->do_lu.lo_dev->ld_site->ls_top_dev,
1771                                    &conf);
1772                 if (IS_ERR(dto))
1773                         GOTO(out_put, rc = PTR_ERR(dto));
1774                 stripe[i] = dto;
1775                 idx_array[i] = idx;
1776         }
1777
1778         lo->ldo_dir_striped = 1;
1779         lo->ldo_stripe = stripe;
1780         lo->ldo_stripenr = i;
1781         lo->ldo_stripes_allocated = stripe_count;
1782
1783         if (lo->ldo_stripenr == 0)
1784                 GOTO(out_put, rc = -ENOSPC);
1785
1786         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
1787         if (rc != 0)
1788                 GOTO(out_put, rc);
1789         lmm = lmv_buf.lb_buf;
1790
1791         OBD_ALLOC_PTR(slave_lmm);
1792         if (slave_lmm == NULL)
1793                 GOTO(out_put, rc = -ENOMEM);
1794
1795         lod_prep_slave_lmv_md(slave_lmm, lmm);
1796         slave_lmv_buf.lb_buf = slave_lmm;
1797         slave_lmv_buf.lb_len = sizeof(*slave_lmm);
1798
1799         if (!dt_try_as_dir(env, dt_object_child(dt)))
1800                 GOTO(out_put, rc = -EINVAL);
1801
1802         rec->rec_type = S_IFDIR;
1803         for (i = 0; i < lo->ldo_stripenr; i++) {
1804                 struct dt_object        *dto            = stripe[i];
1805                 char                    *stripe_name    = info->lti_key;
1806                 struct lu_name          *sname;
1807                 struct linkea_data       ldata          = { NULL };
1808                 struct lu_buf            linkea_buf;
1809
1810                 rc = dt_declare_create(env, dto, attr, NULL, dof, th);
1811                 if (rc != 0)
1812                         GOTO(out_put, rc);
1813
1814                 if (!dt_try_as_dir(env, dto))
1815                         GOTO(out_put, rc = -EINVAL);
1816
1817                 rc = dt_declare_ref_add(env, dto, th);
1818                 if (rc != 0)
1819                         GOTO(out_put, rc);
1820
1821                 rec->rec_fid = lu_object_fid(&dto->do_lu);
1822                 rc = dt_declare_insert(env, dto, (const struct dt_rec *)rec,
1823                                        (const struct dt_key *)dot, th);
1824                 if (rc != 0)
1825                         GOTO(out_put, rc);
1826
1827                 /* master stripe FID will be put to .. */
1828                 rec->rec_fid = lu_object_fid(&dt->do_lu);
1829                 rc = dt_declare_insert(env, dto, (const struct dt_rec *)rec,
1830                                        (const struct dt_key *)dotdot, th);
1831                 if (rc != 0)
1832                         GOTO(out_put, rc);
1833
1834                 /* probably nothing to inherite */
1835                 if (lo->ldo_def_striping_set &&
1836                     !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
1837                                          lo->ldo_def_stripenr,
1838                                          lo->ldo_def_stripe_offset,
1839                                          lo->ldo_pool)) {
1840                         struct lov_user_md_v3   *v3;
1841
1842                         /* sigh, lti_ea_store has been used for lmv_buf,
1843                          * so we have to allocate buffer for default
1844                          * stripe EA */
1845                         OBD_ALLOC_PTR(v3);
1846                         if (v3 == NULL)
1847                                 GOTO(out_put, rc = -ENOMEM);
1848
1849                         memset(v3, 0, sizeof(*v3));
1850                         v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
1851                         v3->lmm_stripe_count =
1852                                 cpu_to_le16(lo->ldo_def_stripenr);
1853                         v3->lmm_stripe_offset =
1854                                 cpu_to_le16(lo->ldo_def_stripe_offset);
1855                         v3->lmm_stripe_size =
1856                                 cpu_to_le32(lo->ldo_def_stripe_size);
1857                         if (lo->ldo_pool != NULL)
1858                                 strlcpy(v3->lmm_pool_name, lo->ldo_pool,
1859                                         sizeof(v3->lmm_pool_name));
1860
1861                         info->lti_buf.lb_buf = v3;
1862                         info->lti_buf.lb_len = sizeof(*v3);
1863                         rc = dt_declare_xattr_set(env, dto,
1864                                                   &info->lti_buf,
1865                                                   XATTR_NAME_LOV,
1866                                                   0, th);
1867                         OBD_FREE_PTR(v3);
1868                         if (rc != 0)
1869                                 GOTO(out_put, rc);
1870                 }
1871
1872                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
1873                     cfs_fail_val != i) {
1874                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
1875                             cfs_fail_val == i)
1876                                 slave_lmm->lmv_master_mdt_index =
1877                                                         cpu_to_le32(i + 1);
1878                         else
1879                                 slave_lmm->lmv_master_mdt_index =
1880                                                         cpu_to_le32(i);
1881                         rc = dt_declare_xattr_set(env, dto, &slave_lmv_buf,
1882                                                   XATTR_NAME_LMV, 0, th);
1883                         if (rc != 0)
1884                                 GOTO(out_put, rc);
1885                 }
1886
1887                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
1888                     cfs_fail_val == i)
1889                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1890                                 PFID(lu_object_fid(&dto->do_lu)), i + 1);
1891                 else
1892                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1893                                 PFID(lu_object_fid(&dto->do_lu)), i);
1894
1895                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
1896                 rc = linkea_data_new(&ldata, &info->lti_linkea_buf);
1897                 if (rc != 0)
1898                         GOTO(out_put, rc);
1899
1900                 rc = linkea_add_buf(&ldata, sname, lu_object_fid(&dt->do_lu));
1901                 if (rc != 0)
1902                         GOTO(out_put, rc);
1903
1904                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
1905                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
1906                 rc = dt_declare_xattr_set(env, dto, &linkea_buf,
1907                                           XATTR_NAME_LINK, 0, th);
1908                 if (rc != 0)
1909                         GOTO(out_put, rc);
1910
1911                 rec->rec_fid = lu_object_fid(&dto->do_lu);
1912                 rc = dt_declare_insert(env, dt_object_child(dt),
1913                                        (const struct dt_rec *)rec,
1914                                        (const struct dt_key *)stripe_name, th);
1915                 if (rc != 0)
1916                         GOTO(out_put, rc);
1917
1918                 rc = dt_declare_ref_add(env, dt_object_child(dt), th);
1919                 if (rc != 0)
1920                         GOTO(out_put, rc);
1921         }
1922
1923         rc = dt_declare_xattr_set(env, dt_object_child(dt), &lmv_buf,
1924                                   XATTR_NAME_LMV, 0, th);
1925         if (rc != 0)
1926                 GOTO(out_put, rc);
1927
1928 out_put:
1929         if (rc < 0) {
1930                 for (i = 0; i < stripe_count; i++)
1931                         if (stripe[i] != NULL)
1932                                 lu_object_put(env, &stripe[i]->do_lu);
1933                 OBD_FREE(stripe, sizeof(stripe[0]) * stripe_count);
1934                 lo->ldo_stripenr = 0;
1935                 lo->ldo_stripes_allocated = 0;
1936                 lo->ldo_stripe = NULL;
1937         }
1938
1939 out_free:
1940         if (idx_array != NULL)
1941                 OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count);
1942         if (slave_lmm != NULL)
1943                 OBD_FREE_PTR(slave_lmm);
1944
1945         RETURN(rc);
1946 }
1947
1948 /**
1949  * Declare create striped md object.
1950  *
1951  * The function declares intention to create a striped directory. This is a
1952  * wrapper for lod_prep_md_striped_create(). The only additional functionality
1953  * is to verify pattern \a lum_buf is good. Check that function for the details.
1954  *
1955  * \param[in] env       execution environment
1956  * \param[in] dt        object
1957  * \param[in] attr      attributes to initialize the objects with
1958  * \param[in] lum_buf   a pattern specifying the number of stripes and
1959  *                      MDT to start from
1960  * \param[in] dof       type of objects to be created
1961  * \param[in] th        transaction handle
1962  *
1963  * \retval              0 on success
1964  * \retval              negative if failed
1965  *
1966  */
1967 static int lod_declare_xattr_set_lmv(const struct lu_env *env,
1968                                      struct dt_object *dt,
1969                                      struct lu_attr *attr,
1970                                      const struct lu_buf *lum_buf,
1971                                      struct dt_object_format *dof,
1972                                      struct thandle *th)
1973 {
1974         struct lod_object       *lo = lod_dt_obj(dt);
1975         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1976         struct lmv_user_md_v1   *lum;
1977         int                     rc;
1978         ENTRY;
1979
1980         lum = lum_buf->lb_buf;
1981         LASSERT(lum != NULL);
1982
1983         CDEBUG(D_INFO, "lum magic = %x count = %u offset = %d\n",
1984                le32_to_cpu(lum->lum_magic), le32_to_cpu(lum->lum_stripe_count),
1985                (int)le32_to_cpu(lum->lum_stripe_offset));
1986
1987         if (le32_to_cpu(lum->lum_stripe_count) == 0)
1988                 GOTO(out, rc = 0);
1989
1990         rc = lod_verify_md_striping(lod, lum);
1991         if (rc != 0)
1992                 GOTO(out, rc);
1993
1994         /* prepare dir striped objects */
1995         rc = lod_prep_md_striped_create(env, dt, attr, lum, dof, th);
1996         if (rc != 0) {
1997                 /* failed to create striping, let's reset
1998                  * config so that others don't get confused */
1999                 lod_object_free_striping(env, lo);
2000                 GOTO(out, rc);
2001         }
2002 out:
2003         RETURN(rc);
2004 }
2005
2006
2007 /**
2008  * Implementation of dt_object_operations::do_declare_xattr_set.
2009  *
2010  * Used with regular (non-striped) objects. Basically it
2011  * initializes the striping information and applies the
2012  * change to all the stripes.
2013  *
2014  * \see dt_object_operations::do_declare_xattr_set() in the API description
2015  * for details.
2016  */
2017 static int lod_dir_declare_xattr_set(const struct lu_env *env,
2018                                      struct dt_object *dt,
2019                                      const struct lu_buf *buf,
2020                                      const char *name, int fl,
2021                                      struct thandle *th)
2022 {
2023         struct dt_object        *next = dt_object_child(dt);
2024         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2025         struct lod_object       *lo = lod_dt_obj(dt);
2026         int                     i;
2027         int                     rc;
2028         ENTRY;
2029
2030         if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
2031                 struct lmv_user_md_v1 *lum;
2032
2033                 LASSERT(buf != NULL && buf->lb_buf != NULL);
2034                 lum = buf->lb_buf;
2035                 rc = lod_verify_md_striping(d, lum);
2036                 if (rc != 0)
2037                         RETURN(rc);
2038         }
2039
2040         rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
2041         if (rc != 0)
2042                 RETURN(rc);
2043
2044         /* set xattr to each stripes, if needed */
2045         rc = lod_load_striping(env, lo);
2046         if (rc != 0)
2047                 RETURN(rc);
2048
2049         /* Note: Do not set LinkEA on sub-stripes, otherwise
2050          * it will confuse the fid2path process(see mdt_path_current()).
2051          * The linkEA between master and sub-stripes is set in
2052          * lod_xattr_set_lmv(). */
2053         if (lo->ldo_stripenr == 0 || strcmp(name, XATTR_NAME_LINK) == 0)
2054                 RETURN(0);
2055
2056         for (i = 0; i < lo->ldo_stripenr; i++) {
2057                 LASSERT(lo->ldo_stripe[i]);
2058                 rc = dt_declare_xattr_set(env, lo->ldo_stripe[i], buf,
2059                                           name, fl, th);
2060                 if (rc != 0)
2061                         break;
2062         }
2063
2064         RETURN(rc);
2065 }
2066
2067 /**
2068  * Implementation of dt_object_operations::do_declare_xattr_set.
2069  *
2070  * \see dt_object_operations::do_declare_xattr_set() in the API description
2071  * for details.
2072  *
2073  * the extension to the API:
2074  *   - declaring LOVEA requests striping creation
2075  *   - LU_XATTR_REPLACE means layout swap
2076  */
2077 static int lod_declare_xattr_set(const struct lu_env *env,
2078                                  struct dt_object *dt,
2079                                  const struct lu_buf *buf,
2080                                  const char *name, int fl,
2081                                  struct thandle *th)
2082 {
2083         struct dt_object *next = dt_object_child(dt);
2084         struct lu_attr   *attr = &lod_env_info(env)->lti_attr;
2085         __u32             mode;
2086         int               rc;
2087         ENTRY;
2088
2089         /*
2090          * allow to declare predefined striping on a new (!mode) object
2091          * which is supposed to be replay of regular file creation
2092          * (when LOV setting is declared)
2093          * LU_XATTR_REPLACE is set to indicate a layout swap
2094          */
2095         mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
2096         if ((S_ISREG(mode) || mode == 0) && strcmp(name, XATTR_NAME_LOV) == 0 &&
2097              !(fl & LU_XATTR_REPLACE)) {
2098                 /*
2099                  * this is a request to manipulate object's striping
2100                  */
2101                 if (dt_object_exists(dt)) {
2102                         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
2103                         if (rc)
2104                                 RETURN(rc);
2105                 } else {
2106                         memset(attr, 0, sizeof(*attr));
2107                         attr->la_valid = LA_TYPE | LA_MODE;
2108                         attr->la_mode = S_IFREG;
2109                 }
2110                 rc = lod_declare_striped_object(env, dt, attr, buf, th);
2111         } else if (S_ISDIR(mode)) {
2112                 rc = lod_dir_declare_xattr_set(env, dt, buf, name, fl, th);
2113         } else {
2114                 rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
2115         }
2116
2117         RETURN(rc);
2118 }
2119
2120 /**
2121  * Resets cached default striping in the object.
2122  *
2123  * \param[in] lo        object
2124  */
2125 static void lod_lov_stripe_cache_clear(struct lod_object *lo)
2126 {
2127         lo->ldo_def_striping_set = 0;
2128         lo->ldo_def_striping_cached = 0;
2129         lod_object_set_pool(lo, NULL);
2130         lo->ldo_def_stripe_size = 0;
2131         lo->ldo_def_stripenr = 0;
2132         if (lo->ldo_dir_stripe != NULL)
2133                 lo->ldo_dir_def_striping_cached = 0;
2134 }
2135
2136 /**
2137  * Apply xattr changes to the object.
2138  *
2139  * Applies xattr changes to the object and the stripes if the latter exist.
2140  *
2141  * \param[in] env       execution environment
2142  * \param[in] dt        object
2143  * \param[in] buf       buffer pointing to the new value of xattr
2144  * \param[in] name      name of xattr
2145  * \param[in] fl        flags
2146  * \param[in] th        transaction handle
2147  * \param[in] capa      not used currently
2148  *
2149  * \retval              0 on success
2150  * \retval              negative if failed
2151  */
2152 static int lod_xattr_set_internal(const struct lu_env *env,
2153                                   struct dt_object *dt,
2154                                   const struct lu_buf *buf,
2155                                   const char *name, int fl, struct thandle *th,
2156                                   struct lustre_capa *capa)
2157 {
2158         struct dt_object        *next = dt_object_child(dt);
2159         struct lod_object       *lo = lod_dt_obj(dt);
2160         int                     rc;
2161         int                     i;
2162         ENTRY;
2163
2164         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
2165         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
2166                 RETURN(rc);
2167
2168         /* Note: Do not set LinkEA on sub-stripes, otherwise
2169          * it will confuse the fid2path process(see mdt_path_current()).
2170          * The linkEA between master and sub-stripes is set in
2171          * lod_xattr_set_lmv(). */
2172         if (lo->ldo_stripenr == 0 || strcmp(name, XATTR_NAME_LINK) == 0)
2173                 RETURN(0);
2174
2175         for (i = 0; i < lo->ldo_stripenr; i++) {
2176                 LASSERT(lo->ldo_stripe[i]);
2177                 rc = dt_xattr_set(env, lo->ldo_stripe[i], buf, name, fl, th,
2178                                   capa);
2179                 if (rc != 0)
2180                         break;
2181         }
2182
2183         RETURN(rc);
2184 }
2185
2186 /**
2187  * Delete an extended attribute.
2188  *
2189  * Deletes specified xattr from the object and the stripes if the latter exist.
2190  *
2191  * \param[in] env       execution environment
2192  * \param[in] dt        object
2193  * \param[in] name      name of xattr
2194  * \param[in] th        transaction handle
2195  * \param[in] capa      not used currently
2196  *
2197  * \retval              0 on success
2198  * \retval              negative if failed
2199  */
2200 static int lod_xattr_del_internal(const struct lu_env *env,
2201                                   struct dt_object *dt,
2202                                   const char *name, struct thandle *th,
2203                                   struct lustre_capa *capa)
2204 {
2205         struct dt_object        *next = dt_object_child(dt);
2206         struct lod_object       *lo = lod_dt_obj(dt);
2207         int                     rc;
2208         int                     i;
2209         ENTRY;
2210
2211         rc = dt_xattr_del(env, next, name, th, capa);
2212         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
2213                 RETURN(rc);
2214
2215         if (lo->ldo_stripenr == 0)
2216                 RETURN(rc);
2217
2218         for (i = 0; i < lo->ldo_stripenr; i++) {
2219                 LASSERT(lo->ldo_stripe[i]);
2220                 rc = dt_xattr_del(env, lo->ldo_stripe[i], name, th,
2221                                   capa);
2222                 if (rc != 0)
2223                         break;
2224         }
2225
2226         RETURN(rc);
2227 }
2228
2229 /**
2230  * Set default striping on a directory.
2231  *
2232  * Sets specified striping on a directory object unless it matches the default
2233  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
2234  * EA. This striping will be used when regular file is being created in this
2235  * directory.
2236  *
2237  * \param[in] env       execution environment
2238  * \param[in] dt        the striped object
2239  * \param[in] buf       buffer with the striping
2240  * \param[in] name      name of EA
2241  * \param[in] fl        xattr flag (see OSD API description)
2242  * \param[in] th        transaction handle
2243  * \param[in] capa      not used
2244  *
2245  * \retval              0 on success
2246  * \retval              negative if failed
2247  */
2248 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
2249                                     struct dt_object *dt,
2250                                     const struct lu_buf *buf,
2251                                     const char *name, int fl,
2252                                     struct thandle *th,
2253                                     struct lustre_capa *capa)
2254 {
2255         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2256         struct lod_object       *l = lod_dt_obj(dt);
2257         struct lov_user_md_v1   *lum;
2258         struct lov_user_md_v3   *v3 = NULL;
2259         const char              *pool_name = NULL;
2260         int                      rc;
2261         ENTRY;
2262
2263         /* If it is striped dir, we should clear the stripe cache for
2264          * slave stripe as well, but there are no effective way to
2265          * notify the LOD on the slave MDT, so we do not cache stripe
2266          * information for slave stripe for now. XXX*/
2267         lod_lov_stripe_cache_clear(l);
2268         LASSERT(buf != NULL && buf->lb_buf != NULL);
2269         lum = buf->lb_buf;
2270
2271         rc = lod_verify_striping(d, buf, false);
2272         if (rc)
2273                 RETURN(rc);
2274
2275         if (lum->lmm_magic == LOV_USER_MAGIC_V3) {
2276                 v3 = buf->lb_buf;
2277                 if (v3->lmm_pool_name[0] != '\0')
2278                         pool_name = v3->lmm_pool_name;
2279         }
2280
2281         /* if { size, offset, count } = { 0, -1, 0 } and no pool
2282          * (i.e. all default values specified) then delete default
2283          * striping from dir. */
2284         CDEBUG(D_OTHER,
2285                 "set default striping: sz %u # %u offset %d %s %s\n",
2286                 (unsigned)lum->lmm_stripe_size,
2287                 (unsigned)lum->lmm_stripe_count,
2288                 (int)lum->lmm_stripe_offset,
2289                 v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
2290
2291         if (LOVEA_DELETE_VALUES(lum->lmm_stripe_size, lum->lmm_stripe_count,
2292                                 lum->lmm_stripe_offset, pool_name)) {
2293                 rc = lod_xattr_del_internal(env, dt, name, th, capa);
2294                 if (rc == -ENODATA)
2295                         rc = 0;
2296         } else {
2297                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th, capa);
2298         }
2299
2300         RETURN(rc);
2301 }
2302
2303 /**
2304  * Set default striping on a directory object.
2305  *
2306  * Sets specified striping on a directory object unless it matches the default
2307  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
2308  * EA. This striping will be used when a new directory is being created in the
2309  * directory.
2310  *
2311  * \param[in] env       execution environment
2312  * \param[in] dt        the striped object
2313  * \param[in] buf       buffer with the striping
2314  * \param[in] name      name of EA
2315  * \param[in] fl        xattr flag (see OSD API description)
2316  * \param[in] th        transaction handle
2317  * \param[in] capa      not used
2318  *
2319  * \retval              0 on success
2320  * \retval              negative if failed
2321  */
2322 static int lod_xattr_set_default_lmv_on_dir(const struct lu_env *env,
2323                                             struct dt_object *dt,
2324                                             const struct lu_buf *buf,
2325                                             const char *name, int fl,
2326                                             struct thandle *th,
2327                                             struct lustre_capa *capa)
2328 {
2329         struct lod_object       *l = lod_dt_obj(dt);
2330         struct lmv_user_md_v1   *lum;
2331         int                      rc;
2332         ENTRY;
2333
2334         LASSERT(buf != NULL && buf->lb_buf != NULL);
2335         lum = buf->lb_buf;
2336
2337         CDEBUG(D_OTHER, "set default stripe_count # %u stripe_offset %d\n",
2338               le32_to_cpu(lum->lum_stripe_count),
2339               (int)le32_to_cpu(lum->lum_stripe_offset));
2340
2341         if (LMVEA_DELETE_VALUES((le32_to_cpu(lum->lum_stripe_count)),
2342                                  le32_to_cpu(lum->lum_stripe_offset)) &&
2343                                 le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC) {
2344                 rc = lod_xattr_del_internal(env, dt, name, th, capa);
2345                 if (rc == -ENODATA)
2346                         rc = 0;
2347         } else {
2348                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th, capa);
2349                 if (rc != 0)
2350                         RETURN(rc);
2351         }
2352
2353         /* Update default stripe cache */
2354         if (l->ldo_dir_stripe == NULL) {
2355                 OBD_ALLOC_PTR(l->ldo_dir_stripe);
2356                 if (l->ldo_dir_stripe == NULL)
2357                         RETURN(-ENOMEM);
2358         }
2359
2360         l->ldo_dir_def_striping_cached = 0;
2361         RETURN(rc);
2362 }
2363
2364 /**
2365  * Turn directory into a striped directory.
2366  *
2367  * During replay the client sends the striping created before MDT
2368  * failure, then the layer above LOD sends this defined striping
2369  * using ->do_xattr_set(), so LOD uses this method to replay creation
2370  * of the stripes. Notice the original information for the striping
2371  * (#stripes, FIDs, etc) was transfered in declare path.
2372  *
2373  * \param[in] env       execution environment
2374  * \param[in] dt        the striped object
2375  * \param[in] buf       not used currently
2376  * \param[in] name      not used currently
2377  * \param[in] fl        xattr flag (see OSD API description)
2378  * \param[in] th        transaction handle
2379  * \param[in] capa      not used
2380  *
2381  * \retval              0 on success
2382  * \retval              negative if failed
2383  */
2384 static int lod_xattr_set_lmv(const struct lu_env *env, struct dt_object *dt,
2385                              const struct lu_buf *buf, const char *name,
2386                              int fl, struct thandle *th,
2387                              struct lustre_capa *capa)
2388 {
2389         struct lod_object       *lo = lod_dt_obj(dt);
2390         struct lod_thread_info  *info = lod_env_info(env);
2391         struct lu_attr          *attr = &info->lti_attr;
2392         struct dt_object_format *dof = &info->lti_format;
2393         struct lu_buf           lmv_buf;
2394         struct lu_buf           slave_lmv_buf;
2395         struct lmv_mds_md_v1    *lmm;
2396         struct lmv_mds_md_v1    *slave_lmm = NULL;
2397         struct dt_insert_rec    *rec = &info->lti_dt_rec;
2398         int                     i;
2399         int                     rc;
2400         ENTRY;
2401
2402         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
2403                 RETURN(-ENOTDIR);
2404
2405         /* The stripes are supposed to be allocated in declare phase,
2406          * if there are no stripes being allocated, it will skip */
2407         if (lo->ldo_stripenr == 0)
2408                 RETURN(0);
2409
2410         rc = dt_attr_get(env, dt_object_child(dt), attr, BYPASS_CAPA);
2411         if (rc != 0)
2412                 RETURN(rc);
2413
2414         attr->la_valid = LA_ATIME | LA_MTIME | LA_CTIME |
2415                          LA_MODE | LA_UID | LA_GID | LA_TYPE;
2416         dof->dof_type = DFT_DIR;
2417
2418         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
2419         if (rc != 0)
2420                 RETURN(rc);
2421         lmm = lmv_buf.lb_buf;
2422
2423         OBD_ALLOC_PTR(slave_lmm);
2424         if (slave_lmm == NULL)
2425                 RETURN(-ENOMEM);
2426
2427         lod_prep_slave_lmv_md(slave_lmm, lmm);
2428         slave_lmv_buf.lb_buf = slave_lmm;
2429         slave_lmv_buf.lb_len = sizeof(*slave_lmm);
2430
2431         rec->rec_type = S_IFDIR;
2432         for (i = 0; i < lo->ldo_stripenr; i++) {
2433                 struct dt_object        *dto;
2434                 char                    *stripe_name    = info->lti_key;
2435                 struct lu_name          *sname;
2436                 struct linkea_data       ldata          = { NULL };
2437                 struct lu_buf            linkea_buf;
2438
2439                 dto = lo->ldo_stripe[i];
2440                 dt_write_lock(env, dto, MOR_TGT_CHILD);
2441                 rc = dt_create(env, dto, attr, NULL, dof, th);
2442                 if (rc != 0) {
2443                         dt_write_unlock(env, dto);
2444                         RETURN(rc);
2445                 }
2446
2447                 rc = dt_ref_add(env, dto, th);
2448                 dt_write_unlock(env, dto);
2449                 if (rc != 0)
2450                         RETURN(rc);
2451
2452                 rec->rec_fid = lu_object_fid(&dto->do_lu);
2453                 rc = dt_insert(env, dto, (const struct dt_rec *)rec,
2454                                (const struct dt_key *)dot, th, capa, 0);
2455                 if (rc != 0)
2456                         RETURN(rc);
2457
2458                 rec->rec_fid = lu_object_fid(&dt->do_lu);
2459                 rc = dt_insert(env, dto, (struct dt_rec *)rec,
2460                                (const struct dt_key *)dotdot, th, capa, 0);
2461                 if (rc != 0)
2462                         RETURN(rc);
2463
2464                 if (lo->ldo_def_striping_set &&
2465                     !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
2466                                          lo->ldo_def_stripenr,
2467                                          lo->ldo_def_stripe_offset,
2468                                          lo->ldo_pool)) {
2469                         struct lov_user_md_v3   *v3;
2470
2471                         /* sigh, lti_ea_store has been used for lmv_buf,
2472                          * so we have to allocate buffer for default
2473                          * stripe EA */
2474                         OBD_ALLOC_PTR(v3);
2475                         if (v3 == NULL)
2476                                 GOTO(out, rc);
2477
2478                         memset(v3, 0, sizeof(*v3));
2479                         v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
2480                         v3->lmm_stripe_count =
2481                                 cpu_to_le16(lo->ldo_def_stripenr);
2482                         v3->lmm_stripe_offset =
2483                                 cpu_to_le16(lo->ldo_def_stripe_offset);
2484                         v3->lmm_stripe_size =
2485                                 cpu_to_le32(lo->ldo_def_stripe_size);
2486                         if (lo->ldo_pool != NULL)
2487                                 strlcpy(v3->lmm_pool_name, lo->ldo_pool,
2488                                         sizeof(v3->lmm_pool_name));
2489
2490                         info->lti_buf.lb_buf = v3;
2491                         info->lti_buf.lb_len = sizeof(*v3);
2492                         rc = dt_xattr_set(env, dto, &info->lti_buf,
2493                                           XATTR_NAME_LOV, 0, th, capa);
2494                         OBD_FREE_PTR(v3);
2495                         if (rc != 0)
2496                                 GOTO(out, rc);
2497                 }
2498
2499                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
2500                     cfs_fail_val != i) {
2501                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
2502                             cfs_fail_val == i)
2503                                 slave_lmm->lmv_master_mdt_index =
2504                                                         cpu_to_le32(i + 1);
2505                         else
2506                                 slave_lmm->lmv_master_mdt_index =
2507                                                         cpu_to_le32(i);
2508                         rc = dt_xattr_set(env, dto, &slave_lmv_buf,
2509                                           XATTR_NAME_LMV, fl, th, capa);
2510                         if (rc != 0)
2511                                 GOTO(out, rc);
2512                 }
2513
2514                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
2515                     cfs_fail_val == i)
2516                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
2517                                  PFID(lu_object_fid(&dto->do_lu)), i + 1);
2518                 else
2519                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
2520                                  PFID(lu_object_fid(&dto->do_lu)), i);
2521
2522                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
2523                 rc = linkea_data_new(&ldata, &info->lti_linkea_buf);
2524                 if (rc != 0)
2525                         GOTO(out, rc);
2526
2527                 rc = linkea_add_buf(&ldata, sname, lu_object_fid(&dt->do_lu));
2528                 if (rc != 0)
2529                         GOTO(out, rc);
2530
2531                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
2532                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
2533                 rc = dt_xattr_set(env, dto, &linkea_buf, XATTR_NAME_LINK,
2534                                   0, th, BYPASS_CAPA);
2535                 if (rc != 0)
2536                         GOTO(out, rc);
2537
2538                 rec->rec_fid = lu_object_fid(&dto->do_lu);
2539                 rc = dt_insert(env, dt_object_child(dt),
2540                                (const struct dt_rec *)rec,
2541                                (const struct dt_key *)stripe_name, th, capa, 0);
2542                 if (rc != 0)
2543                         GOTO(out, rc);
2544
2545                 rc = dt_ref_add(env, dt_object_child(dt), th);
2546                 if (rc != 0)
2547                         GOTO(out, rc);
2548         }
2549
2550         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MASTER_LMV))
2551                 rc = dt_xattr_set(env, dt_object_child(dt), &lmv_buf,
2552                                   XATTR_NAME_LMV, fl, th, capa);
2553
2554 out:
2555         if (slave_lmm != NULL)
2556                 OBD_FREE_PTR(slave_lmm);
2557
2558         RETURN(rc);
2559 }
2560
2561 /**
2562  * Helper function to declare/execute creation of a striped directory
2563  *
2564  * Called in declare/create object path, prepare striping for a directory
2565  * and prepare defaults data striping for the objects to be created in
2566  * that directory. Notice the function calls "declaration" or "execution"
2567  * methods depending on \a declare param. This is a consequence of the
2568  * current approach while we don't have natural distributed transactions:
2569  * we basically execute non-local updates in the declare phase. So, the
2570  * arguments for the both phases are the same and this is the reason for
2571  * this function to exist.
2572  *
2573  * \param[in] env       execution environment
2574  * \param[in] dt        object
2575  * \param[in] attr      attributes the stripes will be created with
2576  * \param[in] dof       format of stripes (see OSD API description)
2577  * \param[in] th        transaction handle
2578  * \param[in] declare   where to call "declare" or "execute" methods
2579  *
2580  * \retval              0 on success
2581  * \retval              negative if failed
2582  */
2583 static int lod_dir_striping_create_internal(const struct lu_env *env,
2584                                             struct dt_object *dt,
2585                                             struct lu_attr *attr,
2586                                             struct dt_object_format *dof,
2587                                             struct thandle *th,
2588                                             bool declare)
2589 {
2590         struct lod_thread_info  *info = lod_env_info(env);
2591         struct lod_object       *lo = lod_dt_obj(dt);
2592         int                     rc;
2593         ENTRY;
2594
2595         if (!LMVEA_DELETE_VALUES(lo->ldo_stripenr,
2596                                  lo->ldo_dir_stripe_offset)) {
2597                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
2598                 int stripe_count = lo->ldo_stripenr;
2599
2600                 if (info->lti_ea_store_size < sizeof(*v1)) {
2601                         rc = lod_ea_store_resize(info, sizeof(*v1));
2602                         if (rc != 0)
2603                                 RETURN(rc);
2604                         v1 = info->lti_ea_store;
2605                 }
2606
2607                 memset(v1, 0, sizeof(*v1));
2608                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
2609                 v1->lum_stripe_count = cpu_to_le32(stripe_count);
2610                 v1->lum_stripe_offset =
2611                                 cpu_to_le32(lo->ldo_dir_stripe_offset);
2612
2613                 info->lti_buf.lb_buf = v1;
2614                 info->lti_buf.lb_len = sizeof(*v1);
2615
2616                 if (declare)
2617                         rc = lod_declare_xattr_set_lmv(env, dt, attr,
2618                                                        &info->lti_buf, dof, th);
2619                 else
2620                         rc = lod_xattr_set_lmv(env, dt, &info->lti_buf,
2621                                                XATTR_NAME_LMV, 0, th,
2622                                                BYPASS_CAPA);
2623                 if (rc != 0)
2624                         RETURN(rc);
2625         }
2626
2627         /* Transfer default LMV striping from the parent */
2628         if (lo->ldo_dir_def_striping_set &&
2629             !LMVEA_DELETE_VALUES(lo->ldo_dir_def_stripenr,
2630                                  lo->ldo_dir_def_stripe_offset)) {
2631                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
2632                 int def_stripe_count = lo->ldo_dir_def_stripenr;
2633
2634                 if (info->lti_ea_store_size < sizeof(*v1)) {
2635                         rc = lod_ea_store_resize(info, sizeof(*v1));
2636                         if (rc != 0)
2637                                 RETURN(rc);
2638                         v1 = info->lti_ea_store;
2639                 }
2640
2641                 memset(v1, 0, sizeof(*v1));
2642                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
2643                 v1->lum_stripe_count = cpu_to_le32(def_stripe_count);
2644                 v1->lum_stripe_offset =
2645                                 cpu_to_le32(lo->ldo_dir_def_stripe_offset);
2646                 v1->lum_hash_type =
2647                                 cpu_to_le32(lo->ldo_dir_def_hash_type);
2648
2649                 info->lti_buf.lb_buf = v1;
2650                 info->lti_buf.lb_len = sizeof(*v1);
2651                 if (declare)
2652                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
2653                                                        XATTR_NAME_DEFAULT_LMV,
2654                                                        0, th);
2655                 else
2656                         rc = lod_xattr_set_default_lmv_on_dir(env, dt,
2657                                                   &info->lti_buf,
2658                                                   XATTR_NAME_DEFAULT_LMV, 0,
2659                                                   th, BYPASS_CAPA);
2660                 if (rc != 0)
2661                         RETURN(rc);
2662         }
2663
2664         /* Transfer default LOV striping from the parent */
2665         if (lo->ldo_def_striping_set &&
2666             !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
2667                                  lo->ldo_def_stripenr,
2668                                  lo->ldo_def_stripe_offset,
2669                                  lo->ldo_pool)) {
2670                 struct lov_user_md_v3 *v3 = info->lti_ea_store;
2671
2672                 if (info->lti_ea_store_size < sizeof(*v3)) {
2673                         rc = lod_ea_store_resize(info, sizeof(*v3));
2674                         if (rc != 0)
2675                                 RETURN(rc);
2676                         v3 = info->lti_ea_store;
2677                 }
2678
2679                 memset(v3, 0, sizeof(*v3));
2680                 v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
2681                 v3->lmm_stripe_count = cpu_to_le16(lo->ldo_def_stripenr);
2682                 v3->lmm_stripe_offset = cpu_to_le16(lo->ldo_def_stripe_offset);
2683                 v3->lmm_stripe_size = cpu_to_le32(lo->ldo_def_stripe_size);
2684                 if (lo->ldo_pool != NULL)
2685                         strlcpy(v3->lmm_pool_name, lo->ldo_pool,
2686                                 sizeof(v3->lmm_pool_name));
2687
2688                 info->lti_buf.lb_buf = v3;
2689                 info->lti_buf.lb_len = sizeof(*v3);
2690
2691                 if (declare)
2692                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
2693                                                        XATTR_NAME_LOV, 0, th);
2694                 else
2695                         rc = lod_xattr_set_lov_on_dir(env, dt, &info->lti_buf,
2696                                                       XATTR_NAME_LOV, 0, th,
2697                                                       BYPASS_CAPA);
2698                 if (rc != 0)
2699                         RETURN(rc);
2700         }
2701
2702         RETURN(0);
2703 }
2704
2705 static int lod_declare_dir_striping_create(const struct lu_env *env,
2706                                            struct dt_object *dt,
2707                                            struct lu_attr *attr,
2708                                            struct dt_object_format *dof,
2709                                            struct thandle *th)
2710 {
2711         return lod_dir_striping_create_internal(env, dt, attr, dof, th, true);
2712 }
2713
2714 static int lod_dir_striping_create(const struct lu_env *env,
2715                                    struct dt_object *dt,
2716                                    struct lu_attr *attr,
2717                                    struct dt_object_format *dof,
2718                                    struct thandle *th)
2719 {
2720         struct lod_object *lo = lod_dt_obj(dt);
2721         int rc;
2722
2723         rc = lod_dir_striping_create_internal(env, dt, attr, dof, th, false);
2724         if (rc == 0)
2725                 lo->ldo_striping_cached = 1;
2726
2727         return rc;
2728 }
2729
2730 /**
2731  * Implementation of dt_object_operations::do_xattr_set.
2732  *
2733  * Sets specified extended attribute on the object. Three types of EAs are
2734  * special:
2735  *   LOV EA - stores striping for a regular file or default striping (when set
2736  *            on a directory)
2737  *   LMV EA - stores a marker for the striped directories
2738  *   DMV EA - stores default directory striping
2739  *
2740  * When striping is applied to a non-striped existing object (this is called
2741  * late striping), then LOD notices the caller wants to turn the object into a
2742  * striped one. The stripe objects are created and appropriate EA is set:
2743  * LOV EA storing all the stripes directly or LMV EA storing just a small header
2744  * with striping configuration.
2745  *
2746  * \see dt_object_operations::do_xattr_set() in the API description for details.
2747  */
2748 static int lod_xattr_set(const struct lu_env *env,
2749                          struct dt_object *dt, const struct lu_buf *buf,
2750                          const char *name, int fl, struct thandle *th,
2751                          struct lustre_capa *capa)
2752 {
2753         struct dt_object        *next = dt_object_child(dt);
2754         int                      rc;
2755         ENTRY;
2756
2757         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
2758             strcmp(name, XATTR_NAME_LMV) == 0) {
2759                 struct lmv_mds_md_v1 *lmm = buf->lb_buf;
2760
2761                 if (lmm != NULL && le32_to_cpu(lmm->lmv_hash_type) &
2762                                                 LMV_HASH_FLAG_MIGRATION)
2763                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
2764                 else
2765                         rc = lod_dir_striping_create(env, dt, NULL, NULL, th);
2766
2767                 RETURN(rc);
2768         }
2769
2770         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
2771             strcmp(name, XATTR_NAME_LOV) == 0) {
2772                 /* default LOVEA */
2773                 rc = lod_xattr_set_lov_on_dir(env, dt, buf, name, fl, th, capa);
2774                 RETURN(rc);
2775         } else if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
2776                    strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
2777                 /* default LMVEA */
2778                 rc = lod_xattr_set_default_lmv_on_dir(env, dt, buf, name, fl,
2779                                                       th, capa);
2780                 RETURN(rc);
2781         } else if (S_ISREG(dt->do_lu.lo_header->loh_attr) &&
2782                    !strcmp(name, XATTR_NAME_LOV)) {
2783                 /* in case of lov EA swap, just set it
2784                  * if not, it is a replay so check striping match what we
2785                  * already have during req replay, declare_xattr_set()
2786                  * defines striping, then create() does the work
2787                 */
2788                 if (fl & LU_XATTR_REPLACE) {
2789                         /* free stripes, then update disk */
2790                         lod_object_free_striping(env, lod_dt_obj(dt));
2791                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
2792                 } else {
2793                         rc = lod_striping_create(env, dt, NULL, NULL, th);
2794                 }
2795                 RETURN(rc);
2796         }
2797
2798         /* then all other xattr */
2799         rc = lod_xattr_set_internal(env, dt, buf, name, fl, th, capa);
2800
2801         RETURN(rc);
2802 }
2803
2804 /**
2805  * Implementation of dt_object_operations::do_declare_xattr_del.
2806  *
2807  * \see dt_object_operations::do_declare_xattr_del() in the API description
2808  * for details.
2809  */
2810 static int lod_declare_xattr_del(const struct lu_env *env,
2811                                  struct dt_object *dt, const char *name,
2812                                  struct thandle *th)
2813 {
2814         return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
2815 }
2816
2817 /**
2818  * Implementation of dt_object_operations::do_xattr_del.
2819  *
2820  * If EA storing a regular striping is being deleted, then release
2821  * all the references to the stripe objects in core.
2822  *
2823  * \see dt_object_operations::do_xattr_del() in the API description for details.
2824  */
2825 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
2826                          const char *name, struct thandle *th,
2827                          struct lustre_capa *capa)
2828 {
2829         if (!strcmp(name, XATTR_NAME_LOV))
2830                 lod_object_free_striping(env, lod_dt_obj(dt));
2831         return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
2832 }
2833
2834 /**
2835  * Implementation of dt_object_operations::do_xattr_list.
2836  *
2837  * \see dt_object_operations::do_xattr_list() in the API description
2838  * for details.
2839  */
2840 static int lod_xattr_list(const struct lu_env *env,
2841                           struct dt_object *dt, const struct lu_buf *buf,
2842                           struct lustre_capa *capa)
2843 {
2844         return dt_xattr_list(env, dt_object_child(dt), buf, capa);
2845 }
2846
2847 /**
2848  * Initialize a pool the object belongs to.
2849  *
2850  * When a striped object is being created, striping configuration
2851  * may demand the stripes are allocated on a limited set of the
2852  * targets. These limited sets are known as "pools". So we copy
2853  * a pool name into the object and later actual creation methods
2854  * (like lod_object_create()) will use this information to allocate
2855  * the stripes properly.
2856  *
2857  * \param[in] o         object
2858  * \param[in] pool      pool name
2859  */
2860 int lod_object_set_pool(struct lod_object *o, char *pool)
2861 {
2862         int len;
2863
2864         if (o->ldo_pool) {
2865                 len = strlen(o->ldo_pool);
2866                 OBD_FREE(o->ldo_pool, len + 1);
2867                 o->ldo_pool = NULL;
2868         }
2869         if (pool) {
2870                 len = strlen(pool);
2871                 OBD_ALLOC(o->ldo_pool, len + 1);
2872                 if (o->ldo_pool == NULL)
2873                         return -ENOMEM;
2874                 strcpy(o->ldo_pool, pool);
2875         }
2876         return 0;
2877 }
2878
2879 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
2880 {
2881         return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
2882 }
2883
2884
2885 /**
2886  * Cache default regular striping in the object.
2887  *
2888  * To improve performance of striped regular object creation we cache
2889  * default LOV striping (if it exists) in the parent directory object.
2890  *
2891  * \param[in] env               execution environment
2892  * \param[in] lp                object
2893  *
2894  * \retval              0 on success
2895  * \retval              negative if failed
2896  */
2897 static int lod_cache_parent_lov_striping(const struct lu_env *env,
2898                                          struct lod_object *lp)
2899 {
2900         struct lod_thread_info  *info = lod_env_info(env);
2901         struct lov_user_md_v1   *v1 = NULL;
2902         struct lov_user_md_v3   *v3 = NULL;
2903         int                      rc;
2904         ENTRY;
2905
2906         /* called from MDD without parent being write locked,
2907          * lock it here */
2908         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
2909         rc = lod_get_lov_ea(env, lp);
2910         if (rc < 0)
2911                 GOTO(unlock, rc);
2912
2913         if (rc < (typeof(rc))sizeof(struct lov_user_md)) {
2914                 /* don't lookup for non-existing or invalid striping */
2915                 lp->ldo_def_striping_set = 0;
2916                 lp->ldo_def_striping_cached = 1;
2917                 lp->ldo_def_stripe_size = 0;
2918                 lp->ldo_def_stripenr = 0;
2919                 lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1);
2920                 GOTO(unlock, rc = 0);
2921         }
2922
2923         rc = 0;
2924         v1 = info->lti_ea_store;
2925         if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1)) {
2926                 lustre_swab_lov_user_md_v1(v1);
2927         } else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3)) {
2928                 v3 = (struct lov_user_md_v3 *)v1;
2929                 lustre_swab_lov_user_md_v3(v3);
2930         }
2931
2932         if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1)
2933                 GOTO(unlock, rc = 0);
2934
2935         if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
2936                 GOTO(unlock, rc = 0);
2937
2938         CDEBUG(D_INFO, DFID" stripe_count=%d stripe_size=%d stripe_offset=%d\n",
2939                PFID(lu_object_fid(&lp->ldo_obj.do_lu)),
2940                (int)v1->lmm_stripe_count,
2941                (int)v1->lmm_stripe_size, (int)v1->lmm_stripe_offset);
2942
2943         lp->ldo_def_stripenr = v1->lmm_stripe_count;
2944         lp->ldo_def_stripe_size = v1->lmm_stripe_size;
2945         lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
2946         lp->ldo_def_striping_cached = 1;
2947         lp->ldo_def_striping_set = 1;
2948         if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
2949                 /* XXX: sanity check here */
2950                 v3 = (struct lov_user_md_v3 *) v1;
2951                 if (v3->lmm_pool_name[0])
2952                         lod_object_set_pool(lp, v3->lmm_pool_name);
2953         }
2954         EXIT;
2955 unlock:
2956         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
2957         return rc;
2958 }
2959
2960
2961 /**
2962  * Cache default directory striping in the object.
2963  *
2964  * To improve performance of striped directory creation we cache default
2965  * directory striping (if it exists) in the parent directory object.
2966  *
2967  * \param[in] env               execution environment
2968  * \param[in] lp                object
2969  *
2970  * \retval              0 on success
2971  * \retval              negative if failed
2972  */
2973 static int lod_cache_parent_lmv_striping(const struct lu_env *env,
2974                                          struct lod_object *lp)
2975 {
2976         struct lod_thread_info  *info = lod_env_info(env);
2977         struct lmv_user_md_v1   *v1 = NULL;
2978         int                      rc;
2979         ENTRY;
2980
2981         /* called from MDD without parent being write locked,
2982          * lock it here */
2983         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
2984         rc = lod_get_default_lmv_ea(env, lp);
2985         if (rc < 0)
2986                 GOTO(unlock, rc);
2987
2988         if (rc < (typeof(rc))sizeof(struct lmv_user_md)) {
2989                 /* don't lookup for non-existing or invalid striping */
2990                 lp->ldo_dir_def_striping_set = 0;
2991                 lp->ldo_dir_def_striping_cached = 1;
2992                 lp->ldo_dir_def_stripenr = 0;
2993                 lp->ldo_dir_def_stripe_offset =
2994                                         (typeof(v1->lum_stripe_offset))(-1);
2995                 lp->ldo_dir_def_hash_type = LMV_HASH_TYPE_FNV_1A_64;
2996                 GOTO(unlock, rc = 0);
2997         }
2998
2999         rc = 0;
3000         v1 = info->lti_ea_store;
3001
3002         lp->ldo_dir_def_stripenr = le32_to_cpu(v1->lum_stripe_count);
3003         lp->ldo_dir_def_stripe_offset = le32_to_cpu(v1->lum_stripe_offset);
3004         lp->ldo_dir_def_hash_type = le32_to_cpu(v1->lum_hash_type);
3005         lp->ldo_dir_def_striping_set = 1;
3006         lp->ldo_dir_def_striping_cached = 1;
3007
3008         EXIT;
3009 unlock:
3010         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
3011         return rc;
3012 }
3013
3014 /**
3015  * Cache default striping in the object.
3016  *
3017  * To improve performance of striped object creation we cache default striping
3018  * (if it exists) in the parent directory object. We always cache default
3019  * striping for the regular files (stored in LOV EA) and we cache default
3020  * striping for the directories if requested by \a child_mode (when a new
3021  * directory is being created).
3022  *
3023  * \param[in] env               execution environment
3024  * \param[in] lp                object
3025  * \param[in] child_mode        new object's mode
3026  *
3027  * \retval              0 on success
3028  * \retval              negative if failed
3029  */
3030 static int lod_cache_parent_striping(const struct lu_env *env,
3031                                      struct lod_object *lp,
3032                                      umode_t child_mode)
3033 {
3034         int rc = 0;
3035         ENTRY;
3036
3037         if (!lp->ldo_def_striping_cached) {
3038                 /* we haven't tried to get default striping for
3039                  * the directory yet, let's cache it in the object */
3040                 rc = lod_cache_parent_lov_striping(env, lp);
3041                 if (rc != 0)
3042                         RETURN(rc);
3043         }
3044
3045         /* If the parent is on the remote MDT, we should always
3046          * try to refresh the default stripeEA cache, because we
3047          * do not cache default striping information for remote
3048          * object. */
3049         if (S_ISDIR(child_mode) && (!lp->ldo_dir_def_striping_cached ||
3050                                     dt_object_remote(&lp->ldo_obj)))
3051                 rc = lod_cache_parent_lmv_striping(env, lp);
3052
3053         RETURN(rc);
3054 }
3055
3056 /**
3057  * Implementation of dt_object_operations::do_ah_init.
3058  *
3059  * This method is used to make a decision on the striping configuration for the
3060  * object being created. It can be taken from the \a parent object if it exists,
3061  * or filesystem's default. The resulting configuration (number of stripes,
3062  * stripe size/offset, pool name, etc) is stored in the object itself and will
3063  * be used by the methods like ->doo_declare_create().
3064  *
3065  * \see dt_object_operations::do_ah_init() in the API description for details.
3066  */
3067 static void lod_ah_init(const struct lu_env *env,
3068                         struct dt_allocation_hint *ah,
3069                         struct dt_object *parent,
3070                         struct dt_object *child,
3071                         umode_t child_mode)
3072 {
3073         struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
3074         struct dt_object  *nextp = NULL;
3075         struct dt_object  *nextc;
3076         struct lod_object *lp = NULL;
3077         struct lod_object *lc;
3078         struct lov_desc   *desc;
3079         int               rc;
3080         ENTRY;
3081
3082         LASSERT(child);
3083
3084         if (likely(parent)) {
3085                 nextp = dt_object_child(parent);
3086                 lp = lod_dt_obj(parent);
3087                 rc = lod_load_striping(env, lp);
3088                 if (rc != 0)
3089                         return;
3090         }
3091
3092         nextc = dt_object_child(child);
3093         lc = lod_dt_obj(child);
3094
3095         LASSERT(lc->ldo_stripenr == 0);
3096         LASSERT(lc->ldo_stripe == NULL);
3097
3098         /*
3099          * local object may want some hints
3100          * in case of late striping creation, ->ah_init()
3101          * can be called with local object existing
3102          */
3103         if (!dt_object_exists(nextc) || dt_object_remote(nextc)) {
3104                 struct dt_object *obj;
3105
3106                 obj = (nextp != NULL && dt_object_remote(nextp)) ? NULL : nextp;
3107                 nextc->do_ops->do_ah_init(env, ah, obj, nextc, child_mode);
3108         }
3109
3110         if (S_ISDIR(child_mode)) {
3111                 if (lc->ldo_dir_stripe == NULL) {
3112                         OBD_ALLOC_PTR(lc->ldo_dir_stripe);
3113                         if (lc->ldo_dir_stripe == NULL)
3114                                 return;
3115                 }
3116
3117                 LASSERT(lp != NULL);
3118                 if (lp->ldo_dir_stripe == NULL) {
3119                         OBD_ALLOC_PTR(lp->ldo_dir_stripe);
3120                         if (lp->ldo_dir_stripe == NULL)
3121                                 return;
3122                 }
3123
3124                 rc = lod_cache_parent_striping(env, lp, child_mode);
3125                 if (rc != 0)
3126                         return;
3127
3128                 /* transfer defaults to new directory */
3129                 if (lp->ldo_def_striping_set) {
3130                         if (lp->ldo_pool)
3131                                 lod_object_set_pool(lc, lp->ldo_pool);
3132                         lc->ldo_def_stripenr = lp->ldo_def_stripenr;
3133                         lc->ldo_def_stripe_size = lp->ldo_def_stripe_size;
3134                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
3135                         lc->ldo_def_striping_set = 1;
3136                         lc->ldo_def_striping_cached = 1;
3137                         CDEBUG(D_OTHER, "inherite EA sz:%d off:%d nr:%d\n",
3138                                (int)lc->ldo_def_stripe_size,
3139                                (int)lc->ldo_def_stripe_offset,
3140                                (int)lc->ldo_def_stripenr);
3141                 }
3142
3143                 /* transfer dir defaults to new directory */
3144                 if (lp->ldo_dir_def_striping_set) {
3145                         lc->ldo_dir_def_stripenr = lp->ldo_dir_def_stripenr;
3146                         lc->ldo_dir_def_stripe_offset =
3147                                                   lp->ldo_dir_def_stripe_offset;
3148                         lc->ldo_dir_def_hash_type =
3149                                                   lp->ldo_dir_def_hash_type;
3150                         lc->ldo_dir_def_striping_set = 1;
3151                         lc->ldo_dir_def_striping_cached = 1;
3152                         CDEBUG(D_INFO, "inherit default EA nr:%d off:%d t%u\n",
3153                                (int)lc->ldo_dir_def_stripenr,
3154                                (int)lc->ldo_dir_def_stripe_offset,
3155                                lc->ldo_dir_def_hash_type);
3156                 }
3157
3158                 /* It should always honour the specified stripes */
3159                 if (ah->dah_eadata != NULL && ah->dah_eadata_len != 0) {
3160                         const struct lmv_user_md_v1 *lum1 = ah->dah_eadata;
3161
3162                         rc = lod_verify_md_striping(d, lum1);
3163                         if (rc == 0 &&
3164                                 le32_to_cpu(lum1->lum_stripe_count) > 1) {
3165                                 /* Directory will be striped only if
3166                                  * stripe_count > 1 */
3167                                 lc->ldo_stripenr =
3168                                         le32_to_cpu(lum1->lum_stripe_count);
3169                                 lc->ldo_dir_stripe_offset =
3170                                         le32_to_cpu(lum1->lum_stripe_offset);
3171                                 lc->ldo_dir_hash_type =
3172                                         le32_to_cpu(lum1->lum_hash_type);
3173                                 CDEBUG(D_INFO, "set stripe EA nr:%hu off:%d\n",
3174                                        lc->ldo_stripenr,
3175                                        (int)lc->ldo_dir_stripe_offset);
3176                         }
3177                 /* then check whether there is default stripes from parent */
3178                 } else if (lp->ldo_dir_def_striping_set) {
3179                         /* If there are default dir stripe from parent */
3180                         lc->ldo_stripenr = lp->ldo_dir_def_stripenr;
3181                         lc->ldo_dir_stripe_offset =
3182                                         lp->ldo_dir_def_stripe_offset;
3183                         lc->ldo_dir_hash_type =
3184                                         lp->ldo_dir_def_hash_type;
3185                         CDEBUG(D_INFO, "inherit EA nr:%hu off:%d\n",
3186                                lc->ldo_stripenr,
3187                                (int)lc->ldo_dir_stripe_offset);
3188                 } else {
3189                         /* set default stripe for this directory */
3190                         lc->ldo_stripenr = 0;
3191                         lc->ldo_dir_stripe_offset = -1;
3192                 }
3193
3194                 CDEBUG(D_INFO, "final striping count:%hu, offset:%d\n",
3195                        lc->ldo_stripenr, (int)lc->ldo_dir_stripe_offset);
3196
3197                 goto out;
3198         }
3199
3200         /*
3201          * if object is going to be striped over OSTs, transfer default
3202          * striping information to the child, so that we can use it
3203          * during declaration and creation
3204          */
3205         if (!lod_object_will_be_striped(S_ISREG(child_mode),
3206                                         lu_object_fid(&child->do_lu)))
3207                 goto out;
3208         /*
3209          * try from the parent
3210          */
3211         if (likely(parent)) {
3212                 lod_cache_parent_striping(env, lp, child_mode);
3213
3214                 lc->ldo_def_stripe_offset = LOV_OFFSET_DEFAULT;
3215
3216                 if (lp->ldo_def_striping_set) {
3217                         if (lp->ldo_pool)
3218                                 lod_object_set_pool(lc, lp->ldo_pool);
3219                         lc->ldo_stripenr = lp->ldo_def_stripenr;
3220                         lc->ldo_stripe_size = lp->ldo_def_stripe_size;
3221                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
3222                         CDEBUG(D_OTHER, "striping from parent: #%d, sz %d %s\n",
3223                                lc->ldo_stripenr, lc->ldo_stripe_size,
3224                                lp->ldo_pool ? lp->ldo_pool : "");
3225                 }
3226         }
3227
3228         /*
3229          * if the parent doesn't provide with specific pattern, grab fs-wide one
3230          */
3231         desc = &d->lod_desc;
3232         if (lc->ldo_stripenr == 0)
3233                 lc->ldo_stripenr = desc->ld_default_stripe_count;
3234         if (lc->ldo_stripe_size == 0)
3235                 lc->ldo_stripe_size = desc->ld_default_stripe_size;
3236         CDEBUG(D_OTHER, "final striping: # %d stripes, sz %d from %s\n",
3237                lc->ldo_stripenr, lc->ldo_stripe_size,
3238                lc->ldo_pool ? lc->ldo_pool : "");
3239
3240 out:
3241         /* we do not cache stripe information for slave stripe, see
3242          * lod_xattr_set_lov_on_dir */
3243         if (lp != NULL && lp->ldo_dir_slave_stripe)
3244                 lod_lov_stripe_cache_clear(lp);
3245
3246         EXIT;
3247 }
3248
3249 #define ll_do_div64(aaa,bbb)    do_div((aaa), (bbb))
3250 /**
3251  * Size initialization on late striping.
3252  *
3253  * Propagate the size of a truncated object to a deferred striping.
3254  * This function handles a special case when truncate was done on a
3255  * non-striped object and now while the striping is being created
3256  * we can't lose that size, so we have to propagate it to the stripes
3257  * being created.
3258  *
3259  * \param[in] env       execution environment
3260  * \param[in] dt        object
3261  * \param[in] th        transaction handle
3262  *
3263  * \retval              0 on success
3264  * \retval              negative if failed
3265  */
3266 static int lod_declare_init_size(const struct lu_env *env,
3267                                  struct dt_object *dt, struct thandle *th)
3268 {
3269         struct dt_object   *next = dt_object_child(dt);
3270         struct lod_object  *lo = lod_dt_obj(dt);
3271         struct lu_attr     *attr = &lod_env_info(env)->lti_attr;
3272         uint64_t            size, offs;
3273         int                 rc, stripe;
3274         ENTRY;
3275
3276         /* XXX: we support the simplest (RAID0) striping so far */
3277         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
3278         LASSERT(lo->ldo_stripe_size > 0);
3279
3280         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
3281         LASSERT(attr->la_valid & LA_SIZE);
3282         if (rc)
3283                 RETURN(rc);
3284
3285         size = attr->la_size;
3286         if (size == 0)
3287                 RETURN(0);
3288
3289         /* ll_do_div64(a, b) returns a % b, and a = a / b */
3290         ll_do_div64(size, (__u64) lo->ldo_stripe_size);
3291         stripe = ll_do_div64(size, (__u64) lo->ldo_stripenr);
3292
3293         size = size * lo->ldo_stripe_size;
3294         offs = attr->la_size;
3295         size += ll_do_div64(offs, lo->ldo_stripe_size);
3296
3297         attr->la_valid = LA_SIZE;
3298         attr->la_size = size;
3299
3300         rc = dt_declare_attr_set(env, lo->ldo_stripe[stripe], attr, th);
3301
3302         RETURN(rc);
3303 }
3304
3305 /**
3306  * Declare creation of striped object.
3307  *
3308  * The function declares creation stripes for a regular object. The function
3309  * also declares whether the stripes will be created with non-zero size if
3310  * previously size was set non-zero on the master object. If object \a dt is
3311  * not local, then only fully defined striping can be applied in \a lovea.
3312  * Otherwise \a lovea can be in the form of pattern, see lod_qos_parse_config()
3313  * for the details.
3314  *
3315  * \param[in] env       execution environment
3316  * \param[in] dt        object
3317  * \param[in] attr      attributes the stripes will be created with
3318  * \param[in] lovea     a buffer containing striping description
3319  * \param[in] th        transaction handle
3320  *
3321  * \retval              0 on success
3322  * \retval              negative if failed
3323  */
3324 int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt,
3325                                struct lu_attr *attr,
3326                                const struct lu_buf *lovea, struct thandle *th)
3327 {
3328         struct lod_thread_info  *info = lod_env_info(env);
3329         struct dt_object        *next = dt_object_child(dt);
3330         struct lod_object       *lo = lod_dt_obj(dt);
3331         int                      rc;
3332         ENTRY;
3333
3334         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) {
3335                 /* failed to create striping, let's reset
3336                  * config so that others don't get confused */
3337                 lod_object_free_striping(env, lo);
3338                 GOTO(out, rc = -ENOMEM);
3339         }
3340
3341         if (!dt_object_remote(next)) {
3342                 /* choose OST and generate appropriate objects */
3343                 rc = lod_qos_prep_create(env, lo, attr, lovea, th);
3344                 if (rc) {
3345                         /* failed to create striping, let's reset
3346                          * config so that others don't get confused */
3347                         lod_object_free_striping(env, lo);
3348                         GOTO(out, rc);
3349                 }
3350
3351                 /*
3352                  * declare storage for striping data
3353                  */
3354                 info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
3355                                 lo->ldo_pool ?  LOV_MAGIC_V3 : LOV_MAGIC_V1);
3356         } else {
3357                 /* LOD can not choose OST objects for remote objects, i.e.
3358                  * stripes must be ready before that. Right now, it can only
3359                  * happen during migrate, i.e. migrate process needs to create
3360                  * remote regular file (mdd_migrate_create), then the migrate
3361                  * process will provide stripeEA. */
3362                 LASSERT(lovea != NULL);
3363                 info->lti_buf = *lovea;
3364         }
3365
3366         rc = dt_declare_xattr_set(env, next, &info->lti_buf,
3367                                   XATTR_NAME_LOV, 0, th);
3368         if (rc)
3369                 GOTO(out, rc);
3370
3371         /*
3372          * if striping is created with local object's size > 0,
3373          * we have to propagate this size to specific object
3374          * the case is possible only when local object was created previously
3375          */
3376         if (dt_object_exists(next))
3377                 rc = lod_declare_init_size(env, dt, th);
3378
3379 out:
3380         RETURN(rc);
3381 }
3382
3383 /**
3384  * Implementation of dt_object_operations::do_declare_create.
3385  *
3386  * The method declares creation of a new object. If the object will be striped,
3387  * then helper functions are called to find FIDs for the stripes, declare
3388  * creation of the stripes and declare initialization of the striping
3389  * information to be stored in the master object.
3390  *
3391  * \see dt_object_operations::do_declare_create() in the API description
3392  * for details.
3393  */
3394 static int lod_declare_object_create(const struct lu_env *env,
3395                                      struct dt_object *dt,
3396                                      struct lu_attr *attr,
3397                                      struct dt_allocation_hint *hint,
3398                                      struct dt_object_format *dof,
3399                                      struct thandle *th)
3400 {
3401         struct dt_object   *next = dt_object_child(dt);
3402         struct lod_object  *lo = lod_dt_obj(dt);
3403         int                 rc;
3404         ENTRY;
3405
3406         LASSERT(dof);
3407         LASSERT(attr);
3408         LASSERT(th);
3409
3410         /*
3411          * first of all, we declare creation of local object
3412          */
3413         rc = dt_declare_create(env, next, attr, hint, dof, th);
3414         if (rc)
3415                 GOTO(out, rc);
3416
3417         if (dof->dof_type == DFT_SYM)
3418                 dt->do_body_ops = &lod_body_lnk_ops;
3419
3420         /*
3421          * it's lod_ah_init() who has decided the object will striped
3422          */
3423         if (dof->dof_type == DFT_REGULAR) {
3424                 /* callers don't want stripes */
3425                 /* XXX: all tricky interactions with ->ah_make_hint() decided
3426                  * to use striping, then ->declare_create() behaving differently
3427                  * should be cleaned */
3428                 if (dof->u.dof_reg.striped == 0)
3429                         lo->ldo_stripenr = 0;
3430                 if (lo->ldo_stripenr > 0)
3431                         rc = lod_declare_striped_object(env, dt, attr,
3432                                                         NULL, th);
3433         } else if (dof->dof_type == DFT_DIR) {
3434                 struct seq_server_site *ss;
3435
3436                 ss = lu_site2seq(dt->do_lu.lo_dev->ld_site);
3437
3438                 /* If the parent has default stripeEA, and client
3439                  * did not find it before sending create request,
3440                  * then MDT will return -EREMOTE, and client will
3441                  * retrieve the default stripeEA and re-create the
3442                  * sub directory.
3443                  *
3444                  * Note: if dah_eadata != NULL, it means creating the
3445                  * striped directory with specified stripeEA, then it
3446                  * should ignore the default stripeEA */
3447                 if ((hint == NULL || hint->dah_eadata == NULL) &&
3448                     lo->ldo_dir_stripe_offset != -1 &&
3449                     lo->ldo_dir_stripe_offset != ss->ss_node_id)
3450                         GOTO(out, rc = -EREMOTE);
3451
3452                 /* Orphan object (like migrating object) does not have
3453                  * lod_dir_stripe, see lod_ah_init */
3454                 if (lo->ldo_dir_stripe != NULL)
3455                         rc = lod_declare_dir_striping_create(env, dt, attr,
3456                                                              dof, th);
3457         }
3458 out:
3459         RETURN(rc);
3460 }
3461
3462 /**
3463  * Creation of a striped regular object.
3464  *
3465  * The function is called to create the stripe objects for a regular
3466  * striped file. This can happen at the initial object creation or
3467  * when the caller asks LOD to do so using ->do_xattr_set() method
3468  * (so called late striping). Notice all the information are already
3469  * prepared in the form of the list of objects (ldo_stripe field).
3470  * This is done during declare phase.
3471  *
3472  * \param[in] env       execution environment
3473  * \param[in] dt        object
3474  * \param[in] attr      attributes the stripes will be created with
3475  * \param[in] dof       format of stripes (see OSD API description)
3476  * \param[in] th        transaction handle
3477  *
3478  * \retval              0 on success
3479  * \retval              negative if failed
3480  */
3481 int lod_striping_create(const struct lu_env *env, struct dt_object *dt,
3482                         struct lu_attr *attr, struct dt_object_format *dof,
3483                         struct thandle *th)
3484 {
3485         struct lod_object *lo = lod_dt_obj(dt);
3486         int                rc = 0, i;
3487         ENTRY;
3488
3489         LASSERT(lo->ldo_striping_cached == 0);
3490
3491         /* create all underlying objects */
3492         for (i = 0; i < lo->ldo_stripenr; i++) {
3493                 LASSERT(lo->ldo_stripe[i]);
3494                 rc = dt_create(env, lo->ldo_stripe[i], attr, NULL, dof, th);
3495
3496                 if (rc)
3497                         break;
3498         }
3499
3500         if (rc == 0) {
3501                 rc = lod_generate_and_set_lovea(env, lo, th);
3502                 if (rc == 0)
3503                         lo->ldo_striping_cached = 1;
3504         }
3505
3506         RETURN(rc);
3507 }
3508
3509 /**
3510  * Implementation of dt_object_operations::do_create.
3511  *
3512  * If any of preceeding methods (like ->do_declare_create(),
3513  * ->do_ah_init(), etc) chose to create a striped object,
3514  * then this method will create the master and the stripes.
3515  *
3516  * \see dt_object_operations::do_create() in the API description for details.
3517  */
3518 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
3519                              struct lu_attr *attr,
3520                              struct dt_allocation_hint *hint,
3521                              struct dt_object_format *dof, struct thandle *th)
3522 {
3523         struct dt_object   *next = dt_object_child(dt);
3524         struct lod_object  *lo = lod_dt_obj(dt);
3525         int                 rc;
3526         ENTRY;
3527
3528         /* create local object */
3529         rc = dt_create(env, next, attr, hint, dof, th);
3530         if (rc != 0)
3531                 RETURN(rc);
3532
3533         if (S_ISREG(dt->do_lu.lo_header->loh_attr) &&
3534             lo->ldo_stripe && dof->u.dof_reg.striped != 0)
3535                 rc = lod_striping_create(env, dt, attr, dof, th);
3536
3537         RETURN(rc);
3538 }
3539
3540 /**
3541  * Implementation of dt_object_operations::do_declare_destroy.
3542  *
3543  * If the object is a striped directory, then the function declares reference
3544  * removal from the master object (this is an index) to the stripes and declares
3545  * destroy of all the stripes. In all the cases, it declares an intention to
3546  * destroy the object itself.
3547  *
3548  * \see dt_object_operations::do_declare_destroy() in the API description
3549  * for details.
3550  */
3551 static int lod_declare_object_destroy(const struct lu_env *env,
3552                                       struct dt_object *dt,
3553                                       struct thandle *th)
3554 {
3555         struct dt_object   *next = dt_object_child(dt);
3556         struct lod_object  *lo = lod_dt_obj(dt);
3557         struct lod_thread_info *info = lod_env_info(env);
3558         char               *stripe_name = info->lti_key;
3559         int                 rc, i;
3560         ENTRY;
3561
3562         /*
3563          * load striping information, notice we don't do this when object
3564          * is being initialized as we don't need this information till
3565          * few specific cases like destroy, chown
3566          */
3567         rc = lod_load_striping(env, lo);
3568         if (rc)
3569                 RETURN(rc);
3570
3571         /* declare destroy for all underlying objects */
3572         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
3573                 rc = next->do_ops->do_index_try(env, next,
3574                                                 &dt_directory_features);
3575                 if (rc != 0)
3576                         RETURN(rc);
3577
3578                 for (i = 0; i < lo->ldo_stripenr; i++) {
3579                         rc = dt_declare_ref_del(env, next, th);
3580                         if (rc != 0)
3581                                 RETURN(rc);
3582                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
3583                                 PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)),
3584                                 i);
3585                         rc = dt_declare_delete(env, next,
3586                                         (const struct dt_key *)stripe_name, th);
3587                         if (rc != 0)
3588                                 RETURN(rc);
3589                 }
3590         }
3591         /*
3592          * we declare destroy for the local object
3593          */
3594         rc = dt_declare_destroy(env, next, th);
3595         if (rc)
3596                 RETURN(rc);
3597
3598         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ))
3599                 RETURN(0);
3600
3601         /* declare destroy all striped objects */
3602         for (i = 0; i < lo->ldo_stripenr; i++) {
3603                 if (likely(lo->ldo_stripe[i] != NULL)) {
3604                         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
3605                                 rc = dt_declare_ref_del(env, lo->ldo_stripe[i],
3606                                                         th);
3607                                 if (rc != 0)
3608                                         RETURN(rc);
3609                         }
3610
3611                         rc = dt_declare_destroy(env, lo->ldo_stripe[i], th);
3612                         if (rc != 0)
3613                                 break;
3614                 }
3615         }
3616
3617         RETURN(rc);
3618 }
3619
3620 /**
3621  * Implementation of dt_object_operations::do_destroy.
3622  *
3623  * If the object is a striped directory, then the function removes references
3624  * from the master object (this is an index) to the stripes and destroys all
3625  * the stripes. In all the cases, the function destroys the object itself.
3626  *
3627  * \see dt_object_operations::do_destroy() in the API description for details.
3628  */
3629 static int lod_object_destroy(const struct lu_env *env,
3630                 struct dt_object *dt, struct thandle *th)
3631 {
3632         struct dt_object  *next = dt_object_child(dt);
3633         struct lod_object *lo = lod_dt_obj(dt);
3634         struct lod_thread_info *info = lod_env_info(env);
3635         char               *stripe_name = info->lti_key;
3636         unsigned int       i;
3637         int                rc;
3638         ENTRY;
3639
3640         /* destroy sub-stripe of master object */
3641         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
3642                 rc = next->do_ops->do_index_try(env, next,
3643                                                 &dt_directory_features);
3644                 if (rc != 0)
3645                         RETURN(rc);
3646
3647                 for (i = 0; i < lo->ldo_stripenr; i++) {
3648                         rc = dt_ref_del(env, next, th);
3649                         if (rc != 0)
3650                                 RETURN(rc);
3651
3652                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
3653                                 PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)),
3654                                 i);
3655
3656                         CDEBUG(D_INFO, DFID" delete stripe %s "DFID"\n",
3657                                PFID(lu_object_fid(&dt->do_lu)), stripe_name,
3658                                PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)));
3659
3660                         rc = dt_delete(env, next,
3661                                        (const struct dt_key *)stripe_name,
3662                                        th, BYPASS_CAPA);
3663                         if (rc != 0)
3664                                 RETURN(rc);
3665                 }
3666         }
3667         rc = dt_destroy(env, next, th);
3668         if (rc != 0)
3669                 RETURN(rc);
3670
3671         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ))
3672                 RETURN(0);
3673
3674         /* destroy&nb