Whamcloud - gitweb
791466c2711376052b043890a99d2d1f910e0ba4
[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         struct lod_object       *lo = lod_dt_obj(dt);
2815         int                     rc;
2816         int                     i;
2817         ENTRY;
2818
2819         rc = dt_declare_xattr_del(env, dt_object_child(dt), name, th);
2820         if (rc != 0)
2821                 RETURN(rc);
2822
2823         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
2824                 RETURN(0);
2825
2826         /* set xattr to each stripes, if needed */
2827         rc = lod_load_striping(env, lo);
2828         if (rc != 0)
2829                 RETURN(rc);
2830
2831         if (lo->ldo_stripenr == 0)
2832                 RETURN(0);
2833
2834         for (i = 0; i < lo->ldo_stripenr; i++) {
2835                 LASSERT(lo->ldo_stripe[i]);
2836                 rc = dt_declare_xattr_del(env, lo->ldo_stripe[i], name, th);
2837                 if (rc != 0)
2838                         break;
2839         }
2840
2841         RETURN(rc);
2842 }
2843
2844 /**
2845  * Implementation of dt_object_operations::do_xattr_del.
2846  *
2847  * If EA storing a regular striping is being deleted, then release
2848  * all the references to the stripe objects in core.
2849  *
2850  * \see dt_object_operations::do_xattr_del() in the API description for details.
2851  */
2852 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
2853                          const char *name, struct thandle *th,
2854                          struct lustre_capa *capa)
2855 {
2856         struct dt_object        *next = dt_object_child(dt);
2857         struct lod_object       *lo = lod_dt_obj(dt);
2858         int                     rc;
2859         int                     i;
2860         ENTRY;
2861
2862         if (!strcmp(name, XATTR_NAME_LOV))
2863                 lod_object_free_striping(env, lod_dt_obj(dt));
2864
2865         rc = dt_xattr_del(env, next, name, th, capa);
2866         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
2867                 RETURN(rc);
2868
2869         if (lo->ldo_stripenr == 0)
2870                 RETURN(0);
2871
2872         for (i = 0; i < lo->ldo_stripenr; i++) {
2873                 LASSERT(lo->ldo_stripe[i]);
2874                 rc = dt_xattr_del(env, lo->ldo_stripe[i], name, th, capa);
2875                 if (rc != 0)
2876                         break;
2877         }
2878
2879         RETURN(rc);
2880 }
2881
2882 /**
2883  * Implementation of dt_object_operations::do_xattr_list.
2884  *
2885  * \see dt_object_operations::do_xattr_list() in the API description
2886  * for details.
2887  */
2888 static int lod_xattr_list(const struct lu_env *env,
2889                           struct dt_object *dt, const struct lu_buf *buf,
2890                           struct lustre_capa *capa)
2891 {
2892         return dt_xattr_list(env, dt_object_child(dt), buf, capa);
2893 }
2894
2895 /**
2896  * Initialize a pool the object belongs to.
2897  *
2898  * When a striped object is being created, striping configuration
2899  * may demand the stripes are allocated on a limited set of the
2900  * targets. These limited sets are known as "pools". So we copy
2901  * a pool name into the object and later actual creation methods
2902  * (like lod_object_create()) will use this information to allocate
2903  * the stripes properly.
2904  *
2905  * \param[in] o         object
2906  * \param[in] pool      pool name
2907  */
2908 int lod_object_set_pool(struct lod_object *o, char *pool)
2909 {
2910         int len;
2911
2912         if (o->ldo_pool) {
2913                 len = strlen(o->ldo_pool);
2914                 OBD_FREE(o->ldo_pool, len + 1);
2915                 o->ldo_pool = NULL;
2916         }
2917         if (pool) {
2918                 len = strlen(pool);
2919                 OBD_ALLOC(o->ldo_pool, len + 1);
2920                 if (o->ldo_pool == NULL)
2921                         return -ENOMEM;
2922                 strcpy(o->ldo_pool, pool);
2923         }
2924         return 0;
2925 }
2926
2927 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
2928 {
2929         return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
2930 }
2931
2932
2933 /**
2934  * Cache default regular striping in the object.
2935  *
2936  * To improve performance of striped regular object creation we cache
2937  * default LOV striping (if it exists) in the parent directory object.
2938  *
2939  * \param[in] env               execution environment
2940  * \param[in] lp                object
2941  *
2942  * \retval              0 on success
2943  * \retval              negative if failed
2944  */
2945 static int lod_cache_parent_lov_striping(const struct lu_env *env,
2946                                          struct lod_object *lp)
2947 {
2948         struct lod_thread_info  *info = lod_env_info(env);
2949         struct lov_user_md_v1   *v1 = NULL;
2950         struct lov_user_md_v3   *v3 = NULL;
2951         int                      rc;
2952         ENTRY;
2953
2954         /* called from MDD without parent being write locked,
2955          * lock it here */
2956         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
2957         rc = lod_get_lov_ea(env, lp);
2958         if (rc < 0)
2959                 GOTO(unlock, rc);
2960
2961         if (rc < (typeof(rc))sizeof(struct lov_user_md)) {
2962                 /* don't lookup for non-existing or invalid striping */
2963                 lp->ldo_def_striping_set = 0;
2964                 lp->ldo_def_striping_cached = 1;
2965                 lp->ldo_def_stripe_size = 0;
2966                 lp->ldo_def_stripenr = 0;
2967                 lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1);
2968                 GOTO(unlock, rc = 0);
2969         }
2970
2971         rc = 0;
2972         v1 = info->lti_ea_store;
2973         if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1)) {
2974                 lustre_swab_lov_user_md_v1(v1);
2975         } else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3)) {
2976                 v3 = (struct lov_user_md_v3 *)v1;
2977                 lustre_swab_lov_user_md_v3(v3);
2978         }
2979
2980         if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1)
2981                 GOTO(unlock, rc = 0);
2982
2983         if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
2984                 GOTO(unlock, rc = 0);
2985
2986         CDEBUG(D_INFO, DFID" stripe_count=%d stripe_size=%d stripe_offset=%d\n",
2987                PFID(lu_object_fid(&lp->ldo_obj.do_lu)),
2988                (int)v1->lmm_stripe_count,
2989                (int)v1->lmm_stripe_size, (int)v1->lmm_stripe_offset);
2990
2991         lp->ldo_def_stripenr = v1->lmm_stripe_count;
2992         lp->ldo_def_stripe_size = v1->lmm_stripe_size;
2993         lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
2994         lp->ldo_def_striping_cached = 1;
2995         lp->ldo_def_striping_set = 1;
2996         if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
2997                 /* XXX: sanity check here */
2998                 v3 = (struct lov_user_md_v3 *) v1;
2999                 if (v3->lmm_pool_name[0])
3000                         lod_object_set_pool(lp, v3->lmm_pool_name);
3001         }
3002         EXIT;
3003 unlock:
3004         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
3005         return rc;
3006 }
3007
3008
3009 /**
3010  * Cache default directory striping in the object.
3011  *
3012  * To improve performance of striped directory creation we cache default
3013  * directory striping (if it exists) in the parent directory object.
3014  *
3015  * \param[in] env               execution environment
3016  * \param[in] lp                object
3017  *
3018  * \retval              0 on success
3019  * \retval              negative if failed
3020  */
3021 static int lod_cache_parent_lmv_striping(const struct lu_env *env,
3022                                          struct lod_object *lp)
3023 {
3024         struct lod_thread_info  *info = lod_env_info(env);
3025         struct lmv_user_md_v1   *v1 = NULL;
3026         int                      rc;
3027         ENTRY;
3028
3029         /* called from MDD without parent being write locked,
3030          * lock it here */
3031         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
3032         rc = lod_get_default_lmv_ea(env, lp);
3033         if (rc < 0)
3034                 GOTO(unlock, rc);
3035
3036         if (rc < (typeof(rc))sizeof(struct lmv_user_md)) {
3037                 /* don't lookup for non-existing or invalid striping */
3038                 lp->ldo_dir_def_striping_set = 0;
3039                 lp->ldo_dir_def_striping_cached = 1;
3040                 lp->ldo_dir_def_stripenr = 0;
3041                 lp->ldo_dir_def_stripe_offset =
3042                                         (typeof(v1->lum_stripe_offset))(-1);
3043                 lp->ldo_dir_def_hash_type = LMV_HASH_TYPE_FNV_1A_64;
3044                 GOTO(unlock, rc = 0);
3045         }
3046
3047         rc = 0;
3048         v1 = info->lti_ea_store;
3049
3050         lp->ldo_dir_def_stripenr = le32_to_cpu(v1->lum_stripe_count);
3051         lp->ldo_dir_def_stripe_offset = le32_to_cpu(v1->lum_stripe_offset);
3052         lp->ldo_dir_def_hash_type = le32_to_cpu(v1->lum_hash_type);
3053         lp->ldo_dir_def_striping_set = 1;
3054         lp->ldo_dir_def_striping_cached = 1;
3055
3056         EXIT;
3057 unlock:
3058         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
3059         return rc;
3060 }
3061
3062 /**
3063  * Cache default striping in the object.
3064  *
3065  * To improve performance of striped object creation we cache default striping
3066  * (if it exists) in the parent directory object. We always cache default
3067  * striping for the regular files (stored in LOV EA) and we cache default
3068  * striping for the directories if requested by \a child_mode (when a new
3069  * directory is being created).
3070  *
3071  * \param[in] env               execution environment
3072  * \param[in] lp                object
3073  * \param[in] child_mode        new object's mode
3074  *
3075  * \retval              0 on success
3076  * \retval              negative if failed
3077  */
3078 static int lod_cache_parent_striping(const struct lu_env *env,
3079                                      struct lod_object *lp,
3080                                      umode_t child_mode)
3081 {
3082         int rc = 0;
3083         ENTRY;
3084
3085         if (!lp->ldo_def_striping_cached) {
3086                 /* we haven't tried to get default striping for
3087                  * the directory yet, let's cache it in the object */
3088                 rc = lod_cache_parent_lov_striping(env, lp);
3089                 if (rc != 0)
3090                         RETURN(rc);
3091         }
3092
3093         /* If the parent is on the remote MDT, we should always
3094          * try to refresh the default stripeEA cache, because we
3095          * do not cache default striping information for remote
3096          * object. */
3097         if (S_ISDIR(child_mode) && (!lp->ldo_dir_def_striping_cached ||
3098                                     dt_object_remote(&lp->ldo_obj)))
3099                 rc = lod_cache_parent_lmv_striping(env, lp);
3100
3101         RETURN(rc);
3102 }
3103
3104 /**
3105  * Implementation of dt_object_operations::do_ah_init.
3106  *
3107  * This method is used to make a decision on the striping configuration for the
3108  * object being created. It can be taken from the \a parent object if it exists,
3109  * or filesystem's default. The resulting configuration (number of stripes,
3110  * stripe size/offset, pool name, etc) is stored in the object itself and will
3111  * be used by the methods like ->doo_declare_create().
3112  *
3113  * \see dt_object_operations::do_ah_init() in the API description for details.
3114  */
3115 static void lod_ah_init(const struct lu_env *env,
3116                         struct dt_allocation_hint *ah,
3117                         struct dt_object *parent,
3118                         struct dt_object *child,
3119                         umode_t child_mode)
3120 {
3121         struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
3122         struct dt_object  *nextp = NULL;
3123         struct dt_object  *nextc;
3124         struct lod_object *lp = NULL;
3125         struct lod_object *lc;
3126         struct lov_desc   *desc;
3127         int               rc;
3128         ENTRY;
3129
3130         LASSERT(child);
3131
3132         if (likely(parent)) {
3133                 nextp = dt_object_child(parent);
3134                 lp = lod_dt_obj(parent);
3135                 rc = lod_load_striping(env, lp);
3136                 if (rc != 0)
3137                         return;
3138         }
3139
3140         nextc = dt_object_child(child);
3141         lc = lod_dt_obj(child);
3142
3143         LASSERT(lc->ldo_stripenr == 0);
3144         LASSERT(lc->ldo_stripe == NULL);
3145
3146         /*
3147          * local object may want some hints
3148          * in case of late striping creation, ->ah_init()
3149          * can be called with local object existing
3150          */
3151         if (!dt_object_exists(nextc) || dt_object_remote(nextc)) {
3152                 struct dt_object *obj;
3153
3154                 obj = (nextp != NULL && dt_object_remote(nextp)) ? NULL : nextp;
3155                 nextc->do_ops->do_ah_init(env, ah, obj, nextc, child_mode);
3156         }
3157
3158         if (S_ISDIR(child_mode)) {
3159                 if (lc->ldo_dir_stripe == NULL) {
3160                         OBD_ALLOC_PTR(lc->ldo_dir_stripe);
3161                         if (lc->ldo_dir_stripe == NULL)
3162                                 return;
3163                 }
3164
3165                 LASSERT(lp != NULL);
3166                 if (lp->ldo_dir_stripe == NULL) {
3167                         OBD_ALLOC_PTR(lp->ldo_dir_stripe);
3168                         if (lp->ldo_dir_stripe == NULL)
3169                                 return;
3170                 }
3171
3172                 rc = lod_cache_parent_striping(env, lp, child_mode);
3173                 if (rc != 0)
3174                         return;
3175
3176                 /* transfer defaults to new directory */
3177                 if (lp->ldo_def_striping_set) {
3178                         if (lp->ldo_pool)
3179                                 lod_object_set_pool(lc, lp->ldo_pool);
3180                         lc->ldo_def_stripenr = lp->ldo_def_stripenr;
3181                         lc->ldo_def_stripe_size = lp->ldo_def_stripe_size;
3182                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
3183                         lc->ldo_def_striping_set = 1;
3184                         lc->ldo_def_striping_cached = 1;
3185                         CDEBUG(D_OTHER, "inherite EA sz:%d off:%d nr:%d\n",
3186                                (int)lc->ldo_def_stripe_size,
3187                                (int)lc->ldo_def_stripe_offset,
3188                                (int)lc->ldo_def_stripenr);
3189                 }
3190
3191                 /* transfer dir defaults to new directory */
3192                 if (lp->ldo_dir_def_striping_set) {
3193                         lc->ldo_dir_def_stripenr = lp->ldo_dir_def_stripenr;
3194                         lc->ldo_dir_def_stripe_offset =
3195                                                   lp->ldo_dir_def_stripe_offset;
3196                         lc->ldo_dir_def_hash_type =
3197                                                   lp->ldo_dir_def_hash_type;
3198                         lc->ldo_dir_def_striping_set = 1;
3199                         lc->ldo_dir_def_striping_cached = 1;
3200                         CDEBUG(D_INFO, "inherit default EA nr:%d off:%d t%u\n",
3201                                (int)lc->ldo_dir_def_stripenr,
3202                                (int)lc->ldo_dir_def_stripe_offset,
3203                                lc->ldo_dir_def_hash_type);
3204                 }
3205
3206                 /* It should always honour the specified stripes */
3207                 if (ah->dah_eadata != NULL && ah->dah_eadata_len != 0) {
3208                         const struct lmv_user_md_v1 *lum1 = ah->dah_eadata;
3209
3210                         rc = lod_verify_md_striping(d, lum1);
3211                         if (rc == 0 &&
3212                                 le32_to_cpu(lum1->lum_stripe_count) > 1) {
3213                                 /* Directory will be striped only if
3214                                  * stripe_count > 1 */
3215                                 lc->ldo_stripenr =
3216                                         le32_to_cpu(lum1->lum_stripe_count);
3217                                 lc->ldo_dir_stripe_offset =
3218                                         le32_to_cpu(lum1->lum_stripe_offset);
3219                                 lc->ldo_dir_hash_type =
3220                                         le32_to_cpu(lum1->lum_hash_type);
3221                                 CDEBUG(D_INFO, "set stripe EA nr:%hu off:%d\n",
3222                                        lc->ldo_stripenr,
3223                                        (int)lc->ldo_dir_stripe_offset);
3224                         }
3225                 /* then check whether there is default stripes from parent */
3226                 } else if (lp->ldo_dir_def_striping_set) {
3227                         /* If there are default dir stripe from parent */
3228                         lc->ldo_stripenr = lp->ldo_dir_def_stripenr;
3229                         lc->ldo_dir_stripe_offset =
3230                                         lp->ldo_dir_def_stripe_offset;
3231                         lc->ldo_dir_hash_type =
3232                                         lp->ldo_dir_def_hash_type;
3233                         CDEBUG(D_INFO, "inherit EA nr:%hu off:%d\n",
3234                                lc->ldo_stripenr,
3235                                (int)lc->ldo_dir_stripe_offset);
3236                 } else {
3237                         /* set default stripe for this directory */
3238                         lc->ldo_stripenr = 0;
3239                         lc->ldo_dir_stripe_offset = -1;
3240                 }
3241
3242                 CDEBUG(D_INFO, "final striping count:%hu, offset:%d\n",
3243                        lc->ldo_stripenr, (int)lc->ldo_dir_stripe_offset);
3244
3245                 goto out;
3246         }
3247
3248         /*
3249          * if object is going to be striped over OSTs, transfer default
3250          * striping information to the child, so that we can use it
3251          * during declaration and creation
3252          */
3253         if (!lod_object_will_be_striped(S_ISREG(child_mode),
3254                                         lu_object_fid(&child->do_lu)))
3255                 goto out;
3256         /*
3257          * try from the parent
3258          */
3259         if (likely(parent)) {
3260                 lod_cache_parent_striping(env, lp, child_mode);
3261
3262                 lc->ldo_def_stripe_offset = LOV_OFFSET_DEFAULT;
3263
3264                 if (lp->ldo_def_striping_set) {
3265                         if (lp->ldo_pool)
3266                                 lod_object_set_pool(lc, lp->ldo_pool);
3267                         lc->ldo_stripenr = lp->ldo_def_stripenr;
3268                         lc->ldo_stripe_size = lp->ldo_def_stripe_size;
3269                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
3270                         CDEBUG(D_OTHER, "striping from parent: #%d, sz %d %s\n",
3271                                lc->ldo_stripenr, lc->ldo_stripe_size,
3272                                lp->ldo_pool ? lp->ldo_pool : "");
3273                 }
3274         }
3275
3276         /*
3277          * if the parent doesn't provide with specific pattern, grab fs-wide one
3278          */
3279         desc = &d->lod_desc;
3280         if (lc->ldo_stripenr == 0)
3281                 lc->ldo_stripenr = desc->ld_default_stripe_count;
3282         if (lc->ldo_stripe_size == 0)
3283                 lc->ldo_stripe_size = desc->ld_default_stripe_size;
3284         CDEBUG(D_OTHER, "final striping: # %d stripes, sz %d from %s\n",
3285                lc->ldo_stripenr, lc->ldo_stripe_size,
3286                lc->ldo_pool ? lc->ldo_pool : "");
3287
3288 out:
3289         /* we do not cache stripe information for slave stripe, see
3290          * lod_xattr_set_lov_on_dir */
3291         if (lp != NULL && lp->ldo_dir_slave_stripe)
3292                 lod_lov_stripe_cache_clear(lp);
3293
3294         EXIT;
3295 }
3296
3297 #define ll_do_div64(aaa,bbb)    do_div((aaa), (bbb))
3298 /**
3299  * Size initialization on late striping.
3300  *
3301  * Propagate the size of a truncated object to a deferred striping.
3302  * This function handles a special case when truncate was done on a
3303  * non-striped object and now while the striping is being created
3304  * we can't lose that size, so we have to propagate it to the stripes
3305  * being created.
3306  *
3307  * \param[in] env       execution environment
3308  * \param[in] dt        object
3309  * \param[in] th        transaction handle
3310  *
3311  * \retval              0 on success
3312  * \retval              negative if failed
3313  */
3314 static int lod_declare_init_size(const struct lu_env *env,
3315                                  struct dt_object *dt, struct thandle *th)
3316 {
3317         struct dt_object   *next = dt_object_child(dt);
3318         struct lod_object  *lo = lod_dt_obj(dt);
3319         struct lu_attr     *attr = &lod_env_info(env)->lti_attr;
3320         uint64_t            size, offs;
3321         int                 rc, stripe;
3322         ENTRY;
3323
3324         /* XXX: we support the simplest (RAID0) striping so far */
3325         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
3326         LASSERT(lo->ldo_stripe_size > 0);
3327
3328         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
3329         LASSERT(attr->la_valid & LA_SIZE);
3330         if (rc)
3331                 RETURN(rc);
3332
3333         size = attr->la_size;
3334         if (size == 0)
3335                 RETURN(0);
3336
3337         /* ll_do_div64(a, b) returns a % b, and a = a / b */
3338         ll_do_div64(size, (__u64) lo->ldo_stripe_size);
3339         stripe = ll_do_div64(size, (__u64) lo->ldo_stripenr);
3340
3341         size = size * lo->ldo_stripe_size;
3342         offs = attr->la_size;
3343         size += ll_do_div64(offs, lo->ldo_stripe_size);
3344
3345         attr->la_valid = LA_SIZE;
3346         attr->la_size = size;
3347
3348         rc = dt_declare_attr_set(env, lo->ldo_stripe[stripe], attr, th);
3349
3350         RETURN(rc);
3351 }
3352
3353 /**
3354  * Declare creation of striped object.
3355  *
3356  * The function declares creation stripes for a regular object. The function
3357  * also declares whether the stripes will be created with non-zero size if
3358  * previously size was set non-zero on the master object. If object \a dt is
3359  * not local, then only fully defined striping can be applied in \a lovea.
3360  * Otherwise \a lovea can be in the form of pattern, see lod_qos_parse_config()
3361  * for the details.
3362  *
3363  * \param[in] env       execution environment
3364  * \param[in] dt        object
3365  * \param[in] attr      attributes the stripes will be created with
3366  * \param[in] lovea     a buffer containing striping description
3367  * \param[in] th        transaction handle
3368  *
3369  * \retval              0 on success
3370  * \retval              negative if failed
3371  */
3372 int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt,
3373                                struct lu_attr *attr,
3374                                const struct lu_buf *lovea, struct thandle *th)
3375 {
3376         struct lod_thread_info  *info = lod_env_info(env);
3377         struct dt_object        *next = dt_object_child(dt);
3378         struct lod_object       *lo = lod_dt_obj(dt);
3379         int                      rc;
3380         ENTRY;
3381
3382         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) {
3383                 /* failed to create striping, let's reset
3384                  * config so that others don't get confused */
3385                 lod_object_free_striping(env, lo);
3386                 GOTO(out, rc = -ENOMEM);
3387         }
3388
3389         if (!dt_object_remote(next)) {
3390                 /* choose OST and generate appropriate objects */
3391                 rc = lod_qos_prep_create(env, lo, attr, lovea, th);
3392                 if (rc) {
3393                         /* failed to create striping, let's reset
3394                          * config so that others don't get confused */
3395                         lod_object_free_striping(env, lo);
3396                         GOTO(out, rc);
3397                 }
3398
3399                 /*
3400                  * declare storage for striping data
3401                  */
3402                 info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
3403                                 lo->ldo_pool ?  LOV_MAGIC_V3 : LOV_MAGIC_V1);
3404         } else {
3405                 /* LOD can not choose OST objects for remote objects, i.e.
3406                  * stripes must be ready before that. Right now, it can only
3407                  * happen during migrate, i.e. migrate process needs to create
3408                  * remote regular file (mdd_migrate_create), then the migrate
3409                  * process will provide stripeEA. */
3410                 LASSERT(lovea != NULL);
3411                 info->lti_buf = *lovea;
3412         }
3413
3414         rc = dt_declare_xattr_set(env, next, &info->lti_buf,
3415                                   XATTR_NAME_LOV, 0, th);
3416         if (rc)
3417                 GOTO(out, rc);
3418
3419         /*
3420          * if striping is created with local object's size > 0,
3421          * we have to propagate this size to specific object
3422          * the case is possible only when local object was created previously
3423          */
3424         if (dt_object_exists(next))
3425                 rc = lod_declare_init_size(env, dt, th);
3426
3427 out:
3428         RETURN(rc);
3429 }
3430
3431 /**
3432  * Implementation of dt_object_operations::do_declare_create.
3433  *
3434  * The method declares creation of a new object. If the object will be striped,
3435  * then helper functions are called to find FIDs for the stripes, declare
3436  * creation of the stripes and declare initialization of the striping
3437  * information to be stored in the master object.
3438  *
3439  * \see dt_object_operations::do_declare_create() in the API description
3440  * for details.
3441  */
3442 static int lod_declare_object_create(const struct lu_env *env,
3443                                      struct dt_object *dt,
3444                                      struct lu_attr *attr,
3445                                      struct dt_allocation_hint *hint,
3446                                      struct dt_object_format *dof,
3447                                      struct thandle *th)
3448 {
3449         struct dt_object   *next = dt_object_child(dt);
3450         struct lod_object  *lo = lod_dt_obj(dt);
3451         int                 rc;
3452         ENTRY;
3453
3454         LASSERT(dof);
3455         LASSERT(attr);
3456         LASSERT(th);
3457
3458         /*
3459          * first of all, we declare creation of local object
3460          */
3461         rc = dt_declare_create(env, next, attr, hint, dof, th);
3462         if (rc)
3463                 GOTO(out, rc);
3464
3465         if (dof->dof_type == DFT_SYM)
3466                 dt->do_body_ops = &lod_body_lnk_ops;
3467
3468         /*
3469          * it's lod_ah_init() who has decided the object will striped
3470          */
3471         if (dof->dof_type == DFT_REGULAR) {
3472                 /* callers don't want stripes */
3473                 /* XXX: all tricky interactions with ->ah_make_hint() decided
3474                  * to use striping, then ->declare_create() behaving differently
3475                  * should be cleaned */
3476                 if (dof->u.dof_reg.striped == 0)
3477                         lo->ldo_stripenr = 0;
3478                 if (lo->ldo_stripenr > 0)
3479                         rc = lod_declare_striped_object(env, dt, attr,
3480                                                         NULL, th);
3481         } else if (dof->dof_type == DFT_DIR) {
3482                 struct seq_server_site *ss;
3483
3484                 ss = lu_site2seq(dt->do_lu.lo_dev->ld_site);
3485
3486                 /* If the parent has default stripeEA, and client
3487                  * did not find it before sending create request,
3488                  * then MDT will return -EREMOTE, and client will
3489                  * retrieve the default stripeEA and re-create the
3490                  * sub directory.
3491                  *
3492                  * Note: if dah_eadata != NULL, it means creating the
3493                  * striped directory with specified stripeEA, then it
3494                  * should ignore the default stripeEA */
3495                 if ((hint == NULL || hint->dah_eadata == NULL) &&
3496                     lo->ldo_dir_stripe_offset != -1 &&
3497                     lo->ldo_dir_stripe_offset != ss->ss_node_id)
3498                         GOTO(out, rc = -EREMOTE);
3499
3500                 /* Orphan object (like migrating object) does not have
3501                  * lod_dir_stripe, see lod_ah_init */
3502                 if (lo->ldo_dir_stripe != NULL)
3503                         rc = lod_declare_dir_striping_create(env, dt, attr,
3504                                                              dof, th);
3505         }
3506 out:
3507         RETURN(rc);
3508 }
3509
3510 /**
3511  * Creation of a striped regular object.
3512  *
3513  * The function is called to create the stripe objects for a regular
3514  * striped file. This can happen at the initial object creation or
3515  * when the caller asks LOD to do so using ->do_xattr_set() method
3516  * (so called late striping). Notice all the information are already
3517  * prepared in the form of the list of objects (ldo_stripe field).
3518  * This is done during declare phase.
3519  *
3520  * \param[in] env       execution environment
3521  * \param[in] dt        object
3522  * \param[in] attr      attributes the stripes will be created with
3523  * \param[in] dof       format of stripes (see OSD API description)
3524  * \param[in] th        transaction handle
3525  *
3526  * \retval              0 on success
3527  * \retval              negative if failed
3528  */
3529 int lod_striping_create(const struct lu_env *env, struct dt_object *dt,
3530                         struct lu_attr *attr, struct dt_object_format *dof,
3531                         struct thandle *th)
3532 {
3533         struct lod_object *lo = lod_dt_obj(dt);
3534         int                rc = 0, i;
3535         ENTRY;
3536
3537         LASSERT(lo->ldo_striping_cached == 0);
3538
3539         /* create all underlying objects */
3540         for (i = 0; i < lo->ldo_stripenr; i++) {
3541                 LASSERT(lo->ldo_stripe[i]);
3542                 rc = dt_create(env, lo->ldo_stripe[i], attr, NULL, dof, th);
3543
3544                 if (rc)
3545                         break;
3546         }
3547
3548         if (rc == 0) {
3549                 rc = lod_generate_and_set_lovea(env, lo, th);
3550                 if (rc == 0)
3551                         lo->ldo_striping_cached = 1;
3552         }
3553
3554         RETURN(rc);
3555 }
3556
3557 /**
3558  * Implementation of dt_object_operations::do_create.
3559  *
3560  * If any of preceeding methods (like ->do_declare_create(),
3561  * ->do_ah_init(), etc) chose to create a striped object,
3562  * then this method will create the master and the stripes.
3563  *
3564  * \see dt_object_operations::do_create() in the API description for details.
3565  */
3566 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
3567                              struct lu_attr *attr,
3568                              struct dt_allocation_hint *hint,
3569                              struct dt_object_format *dof, struct thandle *th)
3570 {
3571         struct dt_object   *next = dt_object_child(dt);
3572         struct lod_object  *lo = lod_dt_obj(dt);
3573         int                 rc;
3574         ENTRY;
3575
3576         /* create local object */
3577         rc = dt_create(env, next, attr, hint, dof, th);
3578         if (rc != 0)
3579                 RETURN(rc);
3580
3581         if (S_ISREG(dt->do_lu.lo_header->loh_attr) &&
3582             lo->ldo_stripe && dof->u.dof_reg.striped != 0)
3583                 rc = lod_striping_create(env, dt, attr, dof, th);
3584
3585         RETURN(rc);
3586 }
3587
3588 /**
3589  * Implementation of dt_object_operations::do_declare_destroy.
3590  *
3591  * If the object is a striped directory, then the function declares reference
3592  * removal from the master object (this is an index) to the stripes and declares
3593  * destroy of all the stripes. In all the cases, it declares an intention to
3594  * destroy the object itself.
3595  *
3596  * \see dt_object_operations::do_declare_destroy() in the API description
3597  * for details.
3598  */
3599 static int lod_declare_object_destroy(const struct lu_env *env,
3600                                       struct dt_object *dt,
3601                                       struct thandle *th)
3602 {
3603         struct dt_object   *next = dt_object_child(dt);
3604         struct lod_object  *lo = lod_dt_obj(dt);
3605         struct lod_thread_info *info = lod_env_info(env);
3606         char               *stripe_name = info->lti_key;
3607         int                 rc, i;
3608         ENTRY;
3609
3610         /*
3611          * load striping information, notice we don't do this when object
3612          * is being initialized as we don't need this information till
3613          * few specific cases like destroy, chown
3614          */
3615         rc = lod_load_striping(env, lo);
3616         if (rc)
3617                 RETURN(rc);
3618
3619         /* declare destroy for all underlying objects */
3620         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
3621                 rc = next->do_ops->do_index_try(env, next,
3622                                                 &dt_directory_features);
3623                 if (rc != 0)
3624                         RETURN(rc);
3625
3626                 for (i = 0; i < lo->ldo_stripenr; i++) {
3627                         rc = dt_declare_ref_del(env, next, th);
3628                         if (rc != 0)
3629                                 RETURN(rc);
3630                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
3631                                 PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)),
3632                                 i);
3633                         rc = dt_declare_delete(env, next,
3634                                         (const struct dt_key *)stripe_name, th);
3635                         if (rc != 0)
3636                                 RETURN(rc);
3637                 }
3638         }
3639         /*
3640          * we declare destroy for the local object
3641          */
3642         rc = dt_declare_destroy(env, next, th);
3643         if (rc)
3644                 RETURN(rc);
3645
3646         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ))
3647                 RETURN(0);
3648
3649         /* declare destroy all striped objects */
3650         for (i = 0; i < lo->ldo_stripenr; i++) {
3651                 if (likely(lo->ldo_stripe[i] != NULL)) {
3652                         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
3653                                 rc = dt_declare_ref_del(env, lo->ldo_stripe[i],
3654                                                         th);
3655                                 if (rc != 0)
3656                                         RETURN(rc);
3657                         }
3658
3659                         rc = dt_declare_destroy(env, lo->ldo_stripe[i], th);
3660                         if (rc != 0)
3661                                 break;
3662                 }
3663         }
3664
3665         RETURN(rc);
3666 }
3667
3668 /**
3669  * Implementation of dt_object_operations::do_destroy.
3670  *
3671  * If the object is a striped directory, then the function removes references
3672  * from the master object (this is an index) to the stripes and destroys all
3673  * the stripes. In all the cases, the function destroys the object itself.
3674  *
3675  * \see dt_object_operations::do_destroy() in the API description for details.
3676  */
3677 static int lod_object_destroy(const struct lu_env *env,
3678                 struct dt_object *dt, struct thandle *th)
3679 {
3680         struct dt_object  *next = dt_object_child(dt);
3681         struct&