Whamcloud - gitweb
450e7f7fa0f6f0b70516dc1bacb5a1ff0cad112c
[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 and the number of the stripes are 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         int     rc = 0;
1448         ENTRY;
1449
1450         if (unlikely(le32_to_cpu(lum->lum_magic) != LMV_USER_MAGIC))
1451                 GOTO(out, rc = -EINVAL);
1452
1453         if (unlikely(le32_to_cpu(lum->lum_stripe_count) == 0))
1454                 GOTO(out, rc = -EINVAL);
1455 out:
1456         if (rc != 0)
1457                 CERROR("%s: invalid lmv_user_md: magic = %x, "
1458                        "stripe_offset = %d, stripe_count = %u: rc = %d\n",
1459                        lod2obd(lod)->obd_name, le32_to_cpu(lum->lum_magic),
1460                        (int)le32_to_cpu(lum->lum_stripe_offset),
1461                        le32_to_cpu(lum->lum_stripe_count), rc);
1462         return rc;
1463 }
1464
1465 /**
1466  * Initialize LMV EA for a slave.
1467  *
1468  * Initialize slave's LMV EA from the master's LMV EA.
1469  *
1470  * \param[in] master_lmv        a buffer containing master's EA
1471  * \param[out] slave_lmv        a buffer where slave's EA will be stored
1472  *
1473  */
1474 static void lod_prep_slave_lmv_md(struct lmv_mds_md_v1 *slave_lmv,
1475                                   const struct lmv_mds_md_v1 *master_lmv)
1476 {
1477         *slave_lmv = *master_lmv;
1478         slave_lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_STRIPE);
1479 }
1480
1481 /**
1482  * Generate LMV EA.
1483  *
1484  * Generate LMV EA from the object passed as \a dt. The object must have
1485  * the stripes created and initialized.
1486  *
1487  * \param[in] env       execution environment
1488  * \param[in] dt        object
1489  * \param[out] lmv_buf  buffer storing generated LMV EA
1490  *
1491  * \retval              0 on success
1492  * \retval              negative if failed
1493  */
1494 static int lod_prep_lmv_md(const struct lu_env *env, struct dt_object *dt,
1495                            struct lu_buf *lmv_buf)
1496 {
1497         struct lod_thread_info  *info = lod_env_info(env);
1498         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1499         struct lod_object       *lo = lod_dt_obj(dt);
1500         struct lmv_mds_md_v1    *lmm1;
1501         int                     stripe_count;
1502         int                     type = LU_SEQ_RANGE_ANY;
1503         int                     rc;
1504         __u32                   mdtidx;
1505         ENTRY;
1506
1507         LASSERT(lo->ldo_dir_striped != 0);
1508         LASSERT(lo->ldo_stripenr > 0);
1509         stripe_count = lo->ldo_stripenr;
1510         /* Only store the LMV EA heahder on the disk. */
1511         if (info->lti_ea_store_size < sizeof(*lmm1)) {
1512                 rc = lod_ea_store_resize(info, sizeof(*lmm1));
1513                 if (rc != 0)
1514                         RETURN(rc);
1515         } else {
1516                 memset(info->lti_ea_store, 0, sizeof(*lmm1));
1517         }
1518
1519         lmm1 = (struct lmv_mds_md_v1 *)info->lti_ea_store;
1520         lmm1->lmv_magic = cpu_to_le32(LMV_MAGIC);
1521         lmm1->lmv_stripe_count = cpu_to_le32(stripe_count);
1522         lmm1->lmv_hash_type = cpu_to_le32(lo->ldo_dir_hash_type);
1523         rc = lod_fld_lookup(env, lod, lu_object_fid(&dt->do_lu),
1524                             &mdtidx, &type);
1525         if (rc != 0)
1526                 RETURN(rc);
1527
1528         lmm1->lmv_master_mdt_index = cpu_to_le32(mdtidx);
1529         lmv_buf->lb_buf = info->lti_ea_store;
1530         lmv_buf->lb_len = sizeof(*lmm1);
1531         lo->ldo_dir_striping_cached = 1;
1532
1533         RETURN(rc);
1534 }
1535
1536 /**
1537  * Create in-core represenation for a striped directory.
1538  *
1539  * Parse the buffer containing LMV EA and instantiate LU objects
1540  * representing the stripe objects. The pointers to the objects are
1541  * stored in ldo_stripe field of \a lo. This function is used when
1542  * we need to access an already created object (i.e. load from a disk).
1543  *
1544  * \param[in] env       execution environment
1545  * \param[in] lo        lod object
1546  * \param[in] buf       buffer containing LMV EA
1547  *
1548  * \retval              0 on success
1549  * \retval              negative if failed
1550  */
1551 int lod_parse_dir_striping(const struct lu_env *env, struct lod_object *lo,
1552                            const struct lu_buf *buf)
1553 {
1554         struct lod_thread_info  *info = lod_env_info(env);
1555         struct lod_device       *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1556         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
1557         struct dt_object        **stripe;
1558         union lmv_mds_md        *lmm = buf->lb_buf;
1559         struct lmv_mds_md_v1    *lmv1 = &lmm->lmv_md_v1;
1560         struct lu_fid           *fid = &info->lti_fid;
1561         unsigned int            i;
1562         int                     rc = 0;
1563         ENTRY;
1564
1565         if (le32_to_cpu(lmv1->lmv_hash_type) & LMV_HASH_FLAG_MIGRATION)
1566                 RETURN(0);
1567
1568         if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_STRIPE) {
1569                 lo->ldo_dir_slave_stripe = 1;
1570                 RETURN(0);
1571         }
1572
1573         if (le32_to_cpu(lmv1->lmv_magic) != LMV_MAGIC_V1)
1574                 RETURN(-EINVAL);
1575
1576         if (le32_to_cpu(lmv1->lmv_stripe_count) < 1)
1577                 RETURN(0);
1578
1579         LASSERT(lo->ldo_stripe == NULL);
1580         OBD_ALLOC(stripe, sizeof(stripe[0]) *
1581                   (le32_to_cpu(lmv1->lmv_stripe_count)));
1582         if (stripe == NULL)
1583                 RETURN(-ENOMEM);
1584
1585         for (i = 0; i < le32_to_cpu(lmv1->lmv_stripe_count); i++) {
1586                 struct dt_device        *tgt_dt;
1587                 struct dt_object        *dto;
1588                 int                     type = LU_SEQ_RANGE_ANY;
1589                 __u32                   idx;
1590
1591                 fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[i]);
1592                 if (!fid_is_sane(fid))
1593                         GOTO(out, rc = -ESTALE);
1594
1595                 rc = lod_fld_lookup(env, lod, fid, &idx, &type);
1596                 if (rc != 0)
1597                         GOTO(out, rc);
1598
1599                 if (idx == lod2lu_dev(lod)->ld_site->ld_seq_site->ss_node_id) {
1600                         tgt_dt = lod->lod_child;
1601                 } else {
1602                         struct lod_tgt_desc     *tgt;
1603
1604                         tgt = LTD_TGT(ltd, idx);
1605                         if (tgt == NULL)
1606                                 GOTO(out, rc = -ESTALE);
1607                         tgt_dt = tgt->ltd_tgt;
1608                 }
1609
1610                 dto = dt_locate_at(env, tgt_dt, fid,
1611                                   lo->ldo_obj.do_lu.lo_dev->ld_site->ls_top_dev,
1612                                   NULL);
1613                 if (IS_ERR(dto))
1614                         GOTO(out, rc = PTR_ERR(dto));
1615
1616                 stripe[i] = dto;
1617         }
1618 out:
1619         lo->ldo_stripe = stripe;
1620         lo->ldo_stripenr = le32_to_cpu(lmv1->lmv_stripe_count);
1621         lo->ldo_stripes_allocated = le32_to_cpu(lmv1->lmv_stripe_count);
1622         if (rc != 0)
1623                 lod_object_free_striping(env, lo);
1624
1625         RETURN(rc);
1626 }
1627
1628 /**
1629  * Create a striped directory.
1630  *
1631  * Create a striped directory with a given stripe pattern on the specified MDTs.
1632  * A striped directory is represented as a regular directory - an index listing
1633  * all the stripes. The stripes point back to the master object with ".." and
1634  * LinkEA. The master object gets LMV EA which identifies it as a striped
1635  * directory. The function allocates FIDs for all the stripes.
1636  *
1637  * \param[in] env       execution environment
1638  * \param[in] dt        object
1639  * \param[in] attr      attributes to initialize the objects with
1640  * \param[in] lum       a pattern specifying the number of stripes and
1641  *                      MDT to start from
1642  * \param[in] dof       type of objects to be created
1643  * \param[in] th        transaction handle
1644  *
1645  * \retval              0 on success
1646  * \retval              negative if failed
1647  */
1648 static int lod_prep_md_striped_create(const struct lu_env *env,
1649                                       struct dt_object *dt,
1650                                       struct lu_attr *attr,
1651                                       const struct lmv_user_md_v1 *lum,
1652                                       struct dt_object_format *dof,
1653                                       struct thandle *th)
1654 {
1655         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1656         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
1657         struct lod_object       *lo = lod_dt_obj(dt);
1658         struct lod_thread_info  *info = lod_env_info(env);
1659         struct dt_object        **stripe;
1660         struct lu_buf           lmv_buf;
1661         struct lu_buf           slave_lmv_buf;
1662         struct lmv_mds_md_v1    *lmm;
1663         struct lmv_mds_md_v1    *slave_lmm = NULL;
1664         struct dt_insert_rec    *rec = &info->lti_dt_rec;
1665         __u32                   stripe_count;
1666         int                     *idx_array;
1667         int                     rc = 0;
1668         __u32                   i;
1669         __u32                   j;
1670         ENTRY;
1671
1672         /* The lum has been verifed in lod_verify_md_striping */
1673         LASSERT(le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC);
1674         LASSERT(le32_to_cpu(lum->lum_stripe_count) > 0);
1675
1676         stripe_count = le32_to_cpu(lum->lum_stripe_count);
1677
1678         /* shrink the stripe_count to the avaible MDT count */
1679         if (stripe_count > lod->lod_remote_mdt_count + 1)
1680                 stripe_count = lod->lod_remote_mdt_count + 1;
1681
1682         OBD_ALLOC(stripe, sizeof(stripe[0]) * stripe_count);
1683         if (stripe == NULL)
1684                 RETURN(-ENOMEM);
1685
1686         OBD_ALLOC(idx_array, sizeof(idx_array[0]) * stripe_count);
1687         if (idx_array == NULL)
1688                 GOTO(out_free, rc = -ENOMEM);
1689
1690         for (i = 0; i < stripe_count; i++) {
1691                 struct lod_tgt_desc     *tgt = NULL;
1692                 struct dt_object        *dto;
1693                 struct lu_fid           fid = { 0 };
1694                 int                     idx;
1695                 struct lu_object_conf   conf = { 0 };
1696                 struct dt_device        *tgt_dt = NULL;
1697
1698                 if (i == 0) {
1699                         /* Right now, master stripe and master object are
1700                          * on the same MDT */
1701                         idx = le32_to_cpu(lum->lum_stripe_offset);
1702                         rc = obd_fid_alloc(env, lod->lod_child_exp, &fid,
1703                                            NULL);
1704                         if (rc < 0)
1705                                 GOTO(out_put, rc);
1706                         tgt_dt = lod->lod_child;
1707                         goto next;
1708                 }
1709
1710                 idx = (idx_array[i - 1] + 1) % (lod->lod_remote_mdt_count + 1);
1711
1712                 for (j = 0; j < lod->lod_remote_mdt_count;
1713                      j++, idx = (idx + 1) % (lod->lod_remote_mdt_count + 1)) {
1714                         bool already_allocated = false;
1715                         __u32 k;
1716
1717                         CDEBUG(D_INFO, "try idx %d, mdt cnt %u,"
1718                                " allocated %u, last allocated %d\n", idx,
1719                                lod->lod_remote_mdt_count, i, idx_array[i - 1]);
1720
1721                         /* Find next available target */
1722                         if (!cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx))
1723                                 continue;
1724
1725                         /* check whether the idx already exists
1726                          * in current allocated array */
1727                         for (k = 0; k < i; k++) {
1728                                 if (idx_array[k] == idx) {
1729                                         already_allocated = true;
1730                                         break;
1731                                 }
1732                         }
1733
1734                         if (already_allocated)
1735                                 continue;
1736
1737                         /* check the status of the OSP */
1738                         tgt = LTD_TGT(ltd, idx);
1739                         if (tgt == NULL)
1740                                 continue;
1741
1742                         tgt_dt = tgt->ltd_tgt;
1743                         rc = dt_statfs(env, tgt_dt, NULL);
1744                         if (rc) {
1745                                 /* this OSP doesn't feel well */
1746                                 rc = 0;
1747                                 continue;
1748                         }
1749
1750                         rc = obd_fid_alloc(env, tgt->ltd_exp, &fid, NULL);
1751                         if (rc < 0) {
1752                                 rc = 0;
1753                                 continue;
1754                         }
1755
1756                         break;
1757                 }
1758
1759                 /* Can not allocate more stripes */
1760                 if (j == lod->lod_remote_mdt_count) {
1761                         CDEBUG(D_INFO, "%s: require stripes %u only get %d\n",
1762                                lod2obd(lod)->obd_name, stripe_count, i - 1);
1763                         break;
1764                 }
1765
1766                 CDEBUG(D_INFO, "idx %d, mdt cnt %u,"
1767                        " allocated %u, last allocated %d\n", idx,
1768                        lod->lod_remote_mdt_count, i, idx_array[i - 1]);
1769
1770 next:
1771                 /* tgt_dt and fid must be ready after search avaible OSP
1772                  * in the above loop */
1773                 LASSERT(tgt_dt != NULL);
1774                 LASSERT(fid_is_sane(&fid));
1775                 conf.loc_flags = LOC_F_NEW;
1776                 dto = dt_locate_at(env, tgt_dt, &fid,
1777                                    dt->do_lu.lo_dev->ld_site->ls_top_dev,
1778                                    &conf);
1779                 if (IS_ERR(dto))
1780                         GOTO(out_put, rc = PTR_ERR(dto));
1781                 stripe[i] = dto;
1782                 idx_array[i] = idx;
1783         }
1784
1785         lo->ldo_dir_striped = 1;
1786         lo->ldo_stripe = stripe;
1787         lo->ldo_stripenr = i;
1788         lo->ldo_stripes_allocated = stripe_count;
1789
1790         if (lo->ldo_stripenr == 0)
1791                 GOTO(out_put, rc = -ENOSPC);
1792
1793         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
1794         if (rc != 0)
1795                 GOTO(out_put, rc);
1796         lmm = lmv_buf.lb_buf;
1797
1798         OBD_ALLOC_PTR(slave_lmm);
1799         if (slave_lmm == NULL)
1800                 GOTO(out_put, rc = -ENOMEM);
1801
1802         lod_prep_slave_lmv_md(slave_lmm, lmm);
1803         slave_lmv_buf.lb_buf = slave_lmm;
1804         slave_lmv_buf.lb_len = sizeof(*slave_lmm);
1805
1806         if (!dt_try_as_dir(env, dt_object_child(dt)))
1807                 GOTO(out_put, rc = -EINVAL);
1808
1809         rec->rec_type = S_IFDIR;
1810         for (i = 0; i < lo->ldo_stripenr; i++) {
1811                 struct dt_object        *dto            = stripe[i];
1812                 char                    *stripe_name    = info->lti_key;
1813                 struct lu_name          *sname;
1814                 struct linkea_data       ldata          = { 0 };
1815                 struct lu_buf            linkea_buf;
1816
1817                 rc = dt_declare_create(env, dto, attr, NULL, dof, th);
1818                 if (rc != 0)
1819                         GOTO(out_put, rc);
1820
1821                 if (!dt_try_as_dir(env, dto))
1822                         GOTO(out_put, rc = -EINVAL);
1823
1824                 rec->rec_fid = lu_object_fid(&dto->do_lu);
1825                 rc = dt_declare_insert(env, dto, (const struct dt_rec *)rec,
1826                                        (const struct dt_key *)dot, th);
1827                 if (rc != 0)
1828                         GOTO(out_put, rc);
1829
1830                 /* master stripe FID will be put to .. */
1831                 rec->rec_fid = lu_object_fid(&dt->do_lu);
1832                 rc = dt_declare_insert(env, dto, (const struct dt_rec *)rec,
1833                                        (const struct dt_key *)dotdot, th);
1834                 if (rc != 0)
1835                         GOTO(out_put, rc);
1836
1837                 /* probably nothing to inherite */
1838                 if (lo->ldo_striping_cached &&
1839                     !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
1840                                          lo->ldo_def_stripenr,
1841                                          lo->ldo_def_stripe_offset,
1842                                          lo->ldo_pool)) {
1843                         struct lov_user_md_v3   *v3;
1844
1845                         /* sigh, lti_ea_store has been used for lmv_buf,
1846                          * so we have to allocate buffer for default
1847                          * stripe EA */
1848                         OBD_ALLOC_PTR(v3);
1849                         if (v3 == NULL)
1850                                 GOTO(out_put, rc = -ENOMEM);
1851
1852                         memset(v3, 0, sizeof(*v3));
1853                         v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
1854                         v3->lmm_stripe_count =
1855                                 cpu_to_le16(lo->ldo_def_stripenr);
1856                         v3->lmm_stripe_offset =
1857                                 cpu_to_le16(lo->ldo_def_stripe_offset);
1858                         v3->lmm_stripe_size =
1859                                 cpu_to_le32(lo->ldo_def_stripe_size);
1860                         if (lo->ldo_pool != NULL)
1861                                 strlcpy(v3->lmm_pool_name, lo->ldo_pool,
1862                                         sizeof(v3->lmm_pool_name));
1863
1864                         info->lti_buf.lb_buf = v3;
1865                         info->lti_buf.lb_len = sizeof(*v3);
1866                         rc = dt_declare_xattr_set(env, dto,
1867                                                   &info->lti_buf,
1868                                                   XATTR_NAME_LOV,
1869                                                   0, th);
1870                         OBD_FREE_PTR(v3);
1871                         if (rc != 0)
1872                                 GOTO(out_put, rc);
1873                 }
1874
1875                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
1876                     cfs_fail_val != i) {
1877                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
1878                             cfs_fail_val == i)
1879                                 slave_lmm->lmv_master_mdt_index =
1880                                                         cpu_to_le32(i + 1);
1881                         else
1882                                 slave_lmm->lmv_master_mdt_index =
1883                                                         cpu_to_le32(i);
1884                         rc = dt_declare_xattr_set(env, dto, &slave_lmv_buf,
1885                                                   XATTR_NAME_LMV, 0, th);
1886                         if (rc != 0)
1887                                 GOTO(out_put, rc);
1888                 }
1889
1890                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
1891                     cfs_fail_val == i)
1892                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1893                                 PFID(lu_object_fid(&dto->do_lu)), i + 1);
1894                 else
1895                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1896                                 PFID(lu_object_fid(&dto->do_lu)), i);
1897
1898                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
1899                 rc = linkea_data_new(&ldata, &info->lti_linkea_buf);
1900                 if (rc != 0)
1901                         GOTO(out_put, rc);
1902
1903                 rc = linkea_add_buf(&ldata, sname, lu_object_fid(&dt->do_lu));
1904                 if (rc != 0)
1905                         GOTO(out_put, rc);
1906
1907                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
1908                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
1909                 rc = dt_declare_xattr_set(env, dto, &linkea_buf,
1910                                           XATTR_NAME_LINK, 0, th);
1911                 if (rc != 0)
1912                         GOTO(out_put, rc);
1913
1914                 rec->rec_fid = lu_object_fid(&dto->do_lu);
1915                 rc = dt_declare_insert(env, dt_object_child(dt),
1916                                        (const struct dt_rec *)rec,
1917                                        (const struct dt_key *)stripe_name, th);
1918                 if (rc != 0)
1919                         GOTO(out_put, rc);
1920
1921                 rc = dt_declare_ref_add(env, dt_object_child(dt), th);
1922                 if (rc != 0)
1923                         GOTO(out_put, rc);
1924         }
1925
1926         rc = dt_declare_xattr_set(env, dt_object_child(dt), &lmv_buf,
1927                                   XATTR_NAME_LMV, 0, th);
1928         if (rc != 0)
1929                 GOTO(out_put, rc);
1930
1931 out_put:
1932         if (rc < 0) {
1933                 for (i = 0; i < stripe_count; i++)
1934                         if (stripe[i] != NULL)
1935                                 lu_object_put(env, &stripe[i]->do_lu);
1936                 OBD_FREE(stripe, sizeof(stripe[0]) * stripe_count);
1937                 lo->ldo_stripenr = 0;
1938                 lo->ldo_stripes_allocated = 0;
1939                 lo->ldo_stripe = NULL;
1940         }
1941
1942 out_free:
1943         if (idx_array != NULL)
1944                 OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count);
1945         if (slave_lmm != NULL)
1946                 OBD_FREE_PTR(slave_lmm);
1947
1948         RETURN(rc);
1949 }
1950
1951 /**
1952  * Declare create striped md object.
1953  *
1954  * The function declares intention to create a striped directory. This is a
1955  * wrapper for lod_prep_md_striped_create(). The only additional functionality
1956  * is to verify pattern \a lum_buf is good. Check that function for the details.
1957  *
1958  * \param[in] env       execution environment
1959  * \param[in] dt        object
1960  * \param[in] attr      attributes to initialize the objects with
1961  * \param[in] lum_buf   a pattern specifying the number of stripes and
1962  *                      MDT to start from
1963  * \param[in] dof       type of objects to be created
1964  * \param[in] th        transaction handle
1965  *
1966  * \retval              0 on success
1967  * \retval              negative if failed
1968  *
1969  */
1970 static int lod_declare_xattr_set_lmv(const struct lu_env *env,
1971                                      struct dt_object *dt,
1972                                      struct lu_attr *attr,
1973                                      const struct lu_buf *lum_buf,
1974                                      struct dt_object_format *dof,
1975                                      struct thandle *th)
1976 {
1977         struct lod_object       *lo = lod_dt_obj(dt);
1978         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1979         struct lmv_user_md_v1   *lum;
1980         int                     rc;
1981         ENTRY;
1982
1983         lum = lum_buf->lb_buf;
1984         LASSERT(lum != NULL);
1985
1986         CDEBUG(D_INFO, "lum magic = %x count = %u offset = %d\n",
1987                le32_to_cpu(lum->lum_magic), le32_to_cpu(lum->lum_stripe_count),
1988                (int)le32_to_cpu(lum->lum_stripe_offset));
1989
1990         if (le32_to_cpu(lum->lum_stripe_count) == 0)
1991                 GOTO(out, rc = 0);
1992
1993         rc = lod_verify_md_striping(lod, lum);
1994         if (rc != 0)
1995                 GOTO(out, rc);
1996
1997         /* prepare dir striped objects */
1998         rc = lod_prep_md_striped_create(env, dt, attr, lum, dof, th);
1999         if (rc != 0) {
2000                 /* failed to create striping, let's reset
2001                  * config so that others don't get confused */
2002                 lod_object_free_striping(env, lo);
2003                 GOTO(out, rc);
2004         }
2005 out:
2006         RETURN(rc);
2007 }
2008
2009
2010 /**
2011  * Implementation of dt_object_operations::do_declare_xattr_set.
2012  *
2013  * Used with regular (non-striped) objects. Basically it
2014  * initializes the striping information and applies the
2015  * change to all the stripes.
2016  *
2017  * \see dt_object_operations::do_declare_xattr_set() in the API description
2018  * for details.
2019  */
2020 static int lod_dir_declare_xattr_set(const struct lu_env *env,
2021                                      struct dt_object *dt,
2022                                      const struct lu_buf *buf,
2023                                      const char *name, int fl,
2024                                      struct thandle *th)
2025 {
2026         struct dt_object        *next = dt_object_child(dt);
2027         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2028         struct lod_object       *lo = lod_dt_obj(dt);
2029         int                     i;
2030         int                     rc;
2031         ENTRY;
2032
2033         if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
2034                 struct lmv_user_md_v1 *lum;
2035
2036                 LASSERT(buf != NULL && buf->lb_buf != NULL);
2037                 lum = buf->lb_buf;
2038                 rc = lod_verify_md_striping(d, lum);
2039                 if (rc != 0)
2040                         RETURN(rc);
2041         }
2042
2043         rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
2044         if (rc != 0)
2045                 RETURN(rc);
2046
2047         /* set xattr to each stripes, if needed */
2048         rc = lod_load_striping(env, lo);
2049         if (rc != 0)
2050                 RETURN(rc);
2051
2052         /* Note: Do not set LinkEA on sub-stripes, otherwise
2053          * it will confuse the fid2path process(see mdt_path_current()).
2054          * The linkEA between master and sub-stripes is set in
2055          * lod_xattr_set_lmv(). */
2056         if (lo->ldo_stripenr == 0 || strcmp(name, XATTR_NAME_LINK) == 0)
2057                 RETURN(0);
2058
2059         for (i = 0; i < lo->ldo_stripenr; i++) {
2060                 LASSERT(lo->ldo_stripe[i]);
2061                 rc = dt_declare_xattr_set(env, lo->ldo_stripe[i], buf,
2062                                           name, fl, th);
2063                 if (rc != 0)
2064                         break;
2065         }
2066
2067         RETURN(rc);
2068 }
2069
2070 /**
2071  * Implementation of dt_object_operations::do_declare_xattr_set.
2072  *
2073  * \see dt_object_operations::do_declare_xattr_set() in the API description
2074  * for details.
2075  *
2076  * the extension to the API:
2077  *   - declaring LOVEA requests striping creation
2078  *   - LU_XATTR_REPLACE means layout swap
2079  */
2080 static int lod_declare_xattr_set(const struct lu_env *env,
2081                                  struct dt_object *dt,
2082                                  const struct lu_buf *buf,
2083                                  const char *name, int fl,
2084                                  struct thandle *th)
2085 {
2086         struct dt_object *next = dt_object_child(dt);
2087         struct lu_attr   *attr = &lod_env_info(env)->lti_attr;
2088         __u32             mode;
2089         int               rc;
2090         ENTRY;
2091
2092         /*
2093          * allow to declare predefined striping on a new (!mode) object
2094          * which is supposed to be replay of regular file creation
2095          * (when LOV setting is declared)
2096          * LU_XATTR_REPLACE is set to indicate a layout swap
2097          */
2098         mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
2099         if ((S_ISREG(mode) || mode == 0) && strcmp(name, XATTR_NAME_LOV) == 0 &&
2100              !(fl & LU_XATTR_REPLACE)) {
2101                 /*
2102                  * this is a request to manipulate object's striping
2103                  */
2104                 if (dt_object_exists(dt)) {
2105                         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
2106                         if (rc)
2107                                 RETURN(rc);
2108                 } else {
2109                         memset(attr, 0, sizeof(*attr));
2110                         attr->la_valid = LA_TYPE | LA_MODE;
2111                         attr->la_mode = S_IFREG;
2112                 }
2113                 rc = lod_declare_striped_object(env, dt, attr, buf, th);
2114         } else if (S_ISDIR(mode)) {
2115                 rc = lod_dir_declare_xattr_set(env, dt, buf, name, fl, th);
2116         } else {
2117                 rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
2118         }
2119
2120         RETURN(rc);
2121 }
2122
2123 /**
2124  * Resets cached default striping in the object.
2125  *
2126  * \param[in] lo        object
2127  */
2128 static void lod_lov_stripe_cache_clear(struct lod_object *lo)
2129 {
2130         lo->ldo_striping_cached = 0;
2131         lo->ldo_def_striping_set = 0;
2132         lod_object_set_pool(lo, NULL);
2133         lo->ldo_def_stripe_size = 0;
2134         lo->ldo_def_stripenr = 0;
2135         if (lo->ldo_dir_stripe != NULL)
2136                 lo->ldo_dir_striping_cached = 0;
2137 }
2138
2139 /**
2140  * Apply xattr changes to the object.
2141  *
2142  * Applies xattr changes to the object and the stripes if the latter exist.
2143  *
2144  * \param[in] env       execution environment
2145  * \param[in] dt        object
2146  * \param[in] buf       buffer pointing to the new value of xattr
2147  * \param[in] name      name of xattr
2148  * \param[in] fl        flags
2149  * \param[in] th        transaction handle
2150  * \param[in] capa      not used currently
2151  *
2152  * \retval              0 on success
2153  * \retval              negative if failed
2154  */
2155 static int lod_xattr_set_internal(const struct lu_env *env,
2156                                   struct dt_object *dt,
2157                                   const struct lu_buf *buf,
2158                                   const char *name, int fl, struct thandle *th,
2159                                   struct lustre_capa *capa)
2160 {
2161         struct dt_object        *next = dt_object_child(dt);
2162         struct lod_object       *lo = lod_dt_obj(dt);
2163         int                     rc;
2164         int                     i;
2165         ENTRY;
2166
2167         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
2168         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
2169                 RETURN(rc);
2170
2171         /* Note: Do not set LinkEA on sub-stripes, otherwise
2172          * it will confuse the fid2path process(see mdt_path_current()).
2173          * The linkEA between master and sub-stripes is set in
2174          * lod_xattr_set_lmv(). */
2175         if (lo->ldo_stripenr == 0 || strcmp(name, XATTR_NAME_LINK) == 0)
2176                 RETURN(0);
2177
2178         for (i = 0; i < lo->ldo_stripenr; i++) {
2179                 LASSERT(lo->ldo_stripe[i]);
2180                 rc = dt_xattr_set(env, lo->ldo_stripe[i], buf, name, fl, th,
2181                                   capa);
2182                 if (rc != 0)
2183                         break;
2184         }
2185
2186         RETURN(rc);
2187 }
2188
2189 /**
2190  * Delete an extended attribute.
2191  *
2192  * Deletes specified xattr from the object and the stripes if the latter exist.
2193  *
2194  * \param[in] env       execution environment
2195  * \param[in] dt        object
2196  * \param[in] name      name of xattr
2197  * \param[in] th        transaction handle
2198  * \param[in] capa      not used currently
2199  *
2200  * \retval              0 on success
2201  * \retval              negative if failed
2202  */
2203 static int lod_xattr_del_internal(const struct lu_env *env,
2204                                   struct dt_object *dt,
2205                                   const char *name, struct thandle *th,
2206                                   struct lustre_capa *capa)
2207 {
2208         struct dt_object        *next = dt_object_child(dt);
2209         struct lod_object       *lo = lod_dt_obj(dt);
2210         int                     rc;
2211         int                     i;
2212         ENTRY;
2213
2214         rc = dt_xattr_del(env, next, name, th, capa);
2215         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
2216                 RETURN(rc);
2217
2218         if (lo->ldo_stripenr == 0)
2219                 RETURN(rc);
2220
2221         for (i = 0; i < lo->ldo_stripenr; i++) {
2222                 LASSERT(lo->ldo_stripe[i]);
2223                 rc = dt_xattr_del(env, lo->ldo_stripe[i], name, th,
2224                                   capa);
2225                 if (rc != 0)
2226                         break;
2227         }
2228
2229         RETURN(rc);
2230 }
2231
2232 /**
2233  * Set default striping on a directory.
2234  *
2235  * Sets specified striping on a directory object unless it matches the default
2236  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
2237  * EA. This striping will be used when regular file is being created in this
2238  * directory.
2239  *
2240  * \param[in] env       execution environment
2241  * \param[in] dt        the striped object
2242  * \param[in] buf       buffer with the striping
2243  * \param[in] name      name of EA
2244  * \param[in] fl        xattr flag (see OSD API description)
2245  * \param[in] th        transaction handle
2246  * \param[in] capa      not used
2247  *
2248  * \retval              0 on success
2249  * \retval              negative if failed
2250  */
2251 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
2252                                     struct dt_object *dt,
2253                                     const struct lu_buf *buf,
2254                                     const char *name, int fl,
2255                                     struct thandle *th,
2256                                     struct lustre_capa *capa)
2257 {
2258         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2259         struct lod_object       *l = lod_dt_obj(dt);
2260         struct lov_user_md_v1   *lum;
2261         struct lov_user_md_v3   *v3 = NULL;
2262         const char              *pool_name = NULL;
2263         int                      rc;
2264         ENTRY;
2265
2266         /* If it is striped dir, we should clear the stripe cache for
2267          * slave stripe as well, but there are no effective way to
2268          * notify the LOD on the slave MDT, so we do not cache stripe
2269          * information for slave stripe for now. XXX*/
2270         lod_lov_stripe_cache_clear(l);
2271         LASSERT(buf != NULL && buf->lb_buf != NULL);
2272         lum = buf->lb_buf;
2273
2274         rc = lod_verify_striping(d, buf, false);
2275         if (rc)
2276                 RETURN(rc);
2277
2278         if (lum->lmm_magic == LOV_USER_MAGIC_V3) {
2279                 v3 = buf->lb_buf;
2280                 if (v3->lmm_pool_name[0] != '\0')
2281                         pool_name = v3->lmm_pool_name;
2282         }
2283
2284         /* if { size, offset, count } = { 0, -1, 0 } and no pool
2285          * (i.e. all default values specified) then delete default
2286          * striping from dir. */
2287         CDEBUG(D_OTHER,
2288                 "set default striping: sz %u # %u offset %d %s %s\n",
2289                 (unsigned)lum->lmm_stripe_size,
2290                 (unsigned)lum->lmm_stripe_count,
2291                 (int)lum->lmm_stripe_offset,
2292                 v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
2293
2294         if (LOVEA_DELETE_VALUES(lum->lmm_stripe_size, lum->lmm_stripe_count,
2295                                 lum->lmm_stripe_offset, pool_name)) {
2296                 rc = lod_xattr_del_internal(env, dt, name, th, capa);
2297                 if (rc == -ENODATA)
2298                         rc = 0;
2299         } else {
2300                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th, capa);
2301         }
2302
2303         RETURN(rc);
2304 }
2305
2306 /**
2307  * Set default striping on a directory object.
2308  *
2309  * Sets specified striping on a directory object unless it matches the default
2310  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
2311  * EA. This striping will be used when a new directory is being created in the
2312  * directory.
2313  *
2314  * \param[in] env       execution environment
2315  * \param[in] dt        the striped object
2316  * \param[in] buf       buffer with the striping
2317  * \param[in] name      name of EA
2318  * \param[in] fl        xattr flag (see OSD API description)
2319  * \param[in] th        transaction handle
2320  * \param[in] capa      not used
2321  *
2322  * \retval              0 on success
2323  * \retval              negative if failed
2324  */
2325 static int lod_xattr_set_default_lmv_on_dir(const struct lu_env *env,
2326                                             struct dt_object *dt,
2327                                             const struct lu_buf *buf,
2328                                             const char *name, int fl,
2329                                             struct thandle *th,
2330                                             struct lustre_capa *capa)
2331 {
2332         struct lod_object       *l = lod_dt_obj(dt);
2333         struct lmv_user_md_v1   *lum;
2334         int                      rc;
2335         ENTRY;
2336
2337         LASSERT(buf != NULL && buf->lb_buf != NULL);
2338         lum = buf->lb_buf;
2339
2340         CDEBUG(D_OTHER, "set default stripe_count # %u stripe_offset %d\n",
2341               le32_to_cpu(lum->lum_stripe_count),
2342               (int)le32_to_cpu(lum->lum_stripe_offset));
2343
2344         if (LMVEA_DELETE_VALUES((le32_to_cpu(lum->lum_stripe_count)),
2345                                  le32_to_cpu(lum->lum_stripe_offset)) &&
2346                                 le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC) {
2347                 rc = lod_xattr_del_internal(env, dt, name, th, capa);
2348                 if (rc == -ENODATA)
2349                         rc = 0;
2350         } else {
2351                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th, capa);
2352                 if (rc != 0)
2353                         RETURN(rc);
2354         }
2355
2356         /* Update default stripe cache */
2357         if (l->ldo_dir_stripe == NULL) {
2358                 OBD_ALLOC_PTR(l->ldo_dir_stripe);
2359                 if (l->ldo_dir_stripe == NULL)
2360                         RETURN(-ENOMEM);
2361         }
2362
2363         l->ldo_dir_striping_cached = 0;
2364         l->ldo_dir_def_striping_set = 1;
2365         l->ldo_dir_def_stripenr = le32_to_cpu(lum->lum_stripe_count);
2366
2367         RETURN(rc);
2368 }
2369
2370 /**
2371  * Turn directory into a striped directory.
2372  *
2373  * During replay the client sends the striping created before MDT
2374  * failure, then the layer above LOD sends this defined striping
2375  * using ->do_xattr_set(), so LOD uses this method to replay creation
2376  * of the stripes. Notice the original information for the striping
2377  * (#stripes, FIDs, etc) was transfered in declare path.
2378  *
2379  * \param[in] env       execution environment
2380  * \param[in] dt        the striped object
2381  * \param[in] buf       not used currently
2382  * \param[in] name      not used currently
2383  * \param[in] fl        xattr flag (see OSD API description)
2384  * \param[in] th        transaction handle
2385  * \param[in] capa      not used
2386  *
2387  * \retval              0 on success
2388  * \retval              negative if failed
2389  */
2390 static int lod_xattr_set_lmv(const struct lu_env *env, struct dt_object *dt,
2391                              const struct lu_buf *buf, const char *name,
2392                              int fl, struct thandle *th,
2393                              struct lustre_capa *capa)
2394 {
2395         struct lod_object       *lo = lod_dt_obj(dt);
2396         struct lod_thread_info  *info = lod_env_info(env);
2397         struct lu_attr          *attr = &info->lti_attr;
2398         struct dt_object_format *dof = &info->lti_format;
2399         struct lu_buf           lmv_buf;
2400         struct lu_buf           slave_lmv_buf;
2401         struct lmv_mds_md_v1    *lmm;
2402         struct lmv_mds_md_v1    *slave_lmm = NULL;
2403         struct dt_insert_rec    *rec = &info->lti_dt_rec;
2404         int                     i;
2405         int                     rc;
2406         ENTRY;
2407
2408         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
2409                 RETURN(-ENOTDIR);
2410
2411         /* The stripes are supposed to be allocated in declare phase,
2412          * if there are no stripes being allocated, it will skip */
2413         if (lo->ldo_stripenr == 0)
2414                 RETURN(0);
2415
2416         rc = dt_attr_get(env, dt_object_child(dt), attr, BYPASS_CAPA);
2417         if (rc != 0)
2418                 RETURN(rc);
2419
2420         attr->la_valid = LA_TYPE | LA_MODE;
2421         dof->dof_type = DFT_DIR;
2422
2423         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
2424         if (rc != 0)
2425                 RETURN(rc);
2426         lmm = lmv_buf.lb_buf;
2427
2428         OBD_ALLOC_PTR(slave_lmm);
2429         if (slave_lmm == NULL)
2430                 RETURN(-ENOMEM);
2431
2432         lod_prep_slave_lmv_md(slave_lmm, lmm);
2433         slave_lmv_buf.lb_buf = slave_lmm;
2434         slave_lmv_buf.lb_len = sizeof(*slave_lmm);
2435
2436         rec->rec_type = S_IFDIR;
2437         for (i = 0; i < lo->ldo_stripenr; i++) {
2438                 struct dt_object        *dto;
2439                 char                    *stripe_name    = info->lti_key;
2440                 struct lu_name          *sname;
2441                 struct linkea_data       ldata          = { 0 };
2442                 struct lu_buf            linkea_buf;
2443
2444                 dto = lo->ldo_stripe[i];
2445                 dt_write_lock(env, dto, MOR_TGT_CHILD);
2446                 rc = dt_create(env, dto, attr, NULL, dof, th);
2447                 dt_write_unlock(env, dto);
2448                 if (rc != 0)
2449                         RETURN(rc);
2450
2451                 rec->rec_fid = lu_object_fid(&dto->do_lu);
2452                 rc = dt_insert(env, dto, (const struct dt_rec *)rec,
2453                                (const struct dt_key *)dot, th, capa, 0);
2454                 if (rc != 0)
2455                         RETURN(rc);
2456
2457                 rec->rec_fid = lu_object_fid(&dt->do_lu);
2458                 rc = dt_insert(env, dto, (struct dt_rec *)rec,
2459                                (const struct dt_key *)dotdot, th, capa, 0);
2460                 if (rc != 0)
2461                         RETURN(rc);
2462
2463                 if (lo->ldo_striping_cached &&
2464                     !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
2465                                          lo->ldo_def_stripenr,
2466                                          lo->ldo_def_stripe_offset,
2467                                          lo->ldo_pool)) {
2468                         struct lov_user_md_v3   *v3;
2469
2470                         /* sigh, lti_ea_store has been used for lmv_buf,
2471                          * so we have to allocate buffer for default
2472                          * stripe EA */
2473                         OBD_ALLOC_PTR(v3);
2474                         if (v3 == NULL)
2475                                 GOTO(out, rc);
2476
2477                         memset(v3, 0, sizeof(*v3));
2478                         v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
2479                         v3->lmm_stripe_count =
2480                                 cpu_to_le16(lo->ldo_def_stripenr);
2481                         v3->lmm_stripe_offset =
2482                                 cpu_to_le16(lo->ldo_def_stripe_offset);
2483                         v3->lmm_stripe_size =
2484                                 cpu_to_le32(lo->ldo_def_stripe_size);
2485                         if (lo->ldo_pool != NULL)
2486                                 strlcpy(v3->lmm_pool_name, lo->ldo_pool,
2487                                         sizeof(v3->lmm_pool_name));
2488
2489                         info->lti_buf.lb_buf = v3;
2490                         info->lti_buf.lb_len = sizeof(*v3);
2491                         rc = dt_xattr_set(env, dto, &info->lti_buf,
2492                                           XATTR_NAME_LOV, 0, th, capa);
2493                         OBD_FREE_PTR(v3);
2494                         if (rc != 0)
2495                                 GOTO(out, rc);
2496                 }
2497
2498                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
2499                     cfs_fail_val != i) {
2500                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
2501                             cfs_fail_val == i)
2502                                 slave_lmm->lmv_master_mdt_index =
2503                                                         cpu_to_le32(i + 1);
2504                         else
2505                                 slave_lmm->lmv_master_mdt_index =
2506                                                         cpu_to_le32(i);
2507                         rc = dt_xattr_set(env, dto, &slave_lmv_buf,
2508                                           XATTR_NAME_LMV, fl, th, capa);
2509                         if (rc != 0)
2510                                 GOTO(out, rc);
2511                 }
2512
2513                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
2514                     cfs_fail_val == i)
2515                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
2516                                  PFID(lu_object_fid(&dto->do_lu)), i + 1);
2517                 else
2518                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
2519                                  PFID(lu_object_fid(&dto->do_lu)), i);
2520
2521                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
2522                 rc = linkea_data_new(&ldata, &info->lti_linkea_buf);
2523                 if (rc != 0)
2524                         GOTO(out, rc);
2525
2526                 rc = linkea_add_buf(&ldata, sname, lu_object_fid(&dt->do_lu));
2527                 if (rc != 0)
2528                         GOTO(out, rc);
2529
2530                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
2531                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
2532                 rc = dt_xattr_set(env, dto, &linkea_buf, XATTR_NAME_LINK,
2533                                   0, th, BYPASS_CAPA);
2534                 if (rc != 0)
2535                         GOTO(out, rc);
2536
2537                 rec->rec_fid = lu_object_fid(&dto->do_lu);
2538                 rc = dt_insert(env, dt_object_child(dt),
2539                                (const struct dt_rec *)rec,
2540                                (const struct dt_key *)stripe_name, th, capa, 0);
2541                 if (rc != 0)
2542                         GOTO(out, rc);
2543
2544                 rc = dt_ref_add(env, dt_object_child(dt), th);
2545                 if (rc != 0)
2546                         GOTO(out, rc);
2547         }
2548
2549         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MASTER_LMV))
2550                 rc = dt_xattr_set(env, dt_object_child(dt), &lmv_buf,
2551                                   XATTR_NAME_LMV, fl, th, capa);
2552
2553 out:
2554         if (slave_lmm != NULL)
2555                 OBD_FREE_PTR(slave_lmm);
2556
2557         RETURN(rc);
2558 }
2559
2560 /**
2561  * Helper function to declare/execute creation of a striped directory
2562  *
2563  * Called in declare/create object path, prepare striping for a directory
2564  * and prepare defaults data striping for the objects to be created in
2565  * that directory. Notice the function calls "declaration" or "execution"
2566  * methods depending on \a declare param. This is a consequence of the
2567  * current approach while we don't have natural distributed transactions:
2568  * we basically execute non-local updates in the declare phase. So, the
2569  * arguments for the both phases are the same and this is the reason for
2570  * this function to exist.
2571  *
2572  * \param[in] env       execution environment
2573  * \param[in] dt        object
2574  * \param[in] attr      attributes the stripes will be created with
2575  * \param[in] dof       format of stripes (see OSD API description)
2576  * \param[in] th        transaction handle
2577  * \param[in] declare   where to call "declare" or "execute" methods
2578  *
2579  * \retval              0 on success
2580  * \retval              negative if failed
2581  */
2582 static int lod_dir_striping_create_internal(const struct lu_env *env,
2583                                             struct dt_object *dt,
2584                                             struct lu_attr *attr,
2585                                             struct dt_object_format *dof,
2586                                             struct thandle *th,
2587                                             bool declare)
2588 {
2589         struct lod_thread_info  *info = lod_env_info(env);
2590         struct lod_object       *lo = lod_dt_obj(dt);
2591         int                     rc;
2592         ENTRY;
2593
2594         if (!LMVEA_DELETE_VALUES(lo->ldo_stripenr,
2595                                  lo->ldo_dir_stripe_offset)) {
2596                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
2597                 int stripe_count = lo->ldo_stripenr;
2598
2599                 if (info->lti_ea_store_size < sizeof(*v1)) {
2600                         rc = lod_ea_store_resize(info, sizeof(*v1));
2601                         if (rc != 0)
2602                                 RETURN(rc);
2603                         v1 = info->lti_ea_store;
2604                 }
2605
2606                 memset(v1, 0, sizeof(*v1));
2607                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
2608                 v1->lum_stripe_count = cpu_to_le32(stripe_count);
2609                 v1->lum_stripe_offset =
2610                                 cpu_to_le32(lo->ldo_dir_stripe_offset);
2611
2612                 info->lti_buf.lb_buf = v1;
2613                 info->lti_buf.lb_len = sizeof(*v1);
2614
2615                 if (declare)
2616                         rc = lod_declare_xattr_set_lmv(env, dt, attr,
2617                                                        &info->lti_buf, dof, th);
2618                 else
2619                         rc = lod_xattr_set_lmv(env, dt, &info->lti_buf,
2620                                                XATTR_NAME_LMV, 0, th,
2621                                                BYPASS_CAPA);
2622                 if (rc != 0)
2623                         RETURN(rc);
2624         }
2625
2626         /* Transfer default LMV striping from the parent */
2627         if (lo->ldo_dir_striping_cached &&
2628             !LMVEA_DELETE_VALUES(lo->ldo_dir_def_stripenr,
2629                                  lo->ldo_dir_def_stripe_offset)) {
2630                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
2631                 int def_stripe_count = lo->ldo_dir_def_stripenr;
2632
2633                 if (info->lti_ea_store_size < sizeof(*v1)) {
2634                         rc = lod_ea_store_resize(info, sizeof(*v1));
2635                         if (rc != 0)
2636                                 RETURN(rc);
2637                         v1 = info->lti_ea_store;
2638                 }
2639
2640                 memset(v1, 0, sizeof(*v1));
2641                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
2642                 v1->lum_stripe_count = cpu_to_le32(def_stripe_count);
2643                 v1->lum_stripe_offset =
2644                                 cpu_to_le32(lo->ldo_dir_def_stripe_offset);
2645                 v1->lum_hash_type =
2646                                 cpu_to_le32(lo->ldo_dir_def_hash_type);
2647
2648                 info->lti_buf.lb_buf = v1;
2649                 info->lti_buf.lb_len = sizeof(*v1);
2650                 if (declare)
2651                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
2652                                                        XATTR_NAME_DEFAULT_LMV,
2653                                                        0, th);
2654                 else
2655                         rc = lod_xattr_set_default_lmv_on_dir(env, dt,
2656                                                   &info->lti_buf,
2657                                                   XATTR_NAME_DEFAULT_LMV, 0,
2658                                                   th, BYPASS_CAPA);
2659                 if (rc != 0)
2660                         RETURN(rc);
2661         }
2662
2663         /* Transfer default LOV striping from the parent */
2664         if (lo->ldo_striping_cached &&
2665             !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
2666                                  lo->ldo_def_stripenr,
2667                                  lo->ldo_def_stripe_offset,
2668                                  lo->ldo_pool)) {
2669                 struct lov_user_md_v3 *v3 = info->lti_ea_store;
2670
2671                 if (info->lti_ea_store_size < sizeof(*v3)) {
2672                         rc = lod_ea_store_resize(info, sizeof(*v3));
2673                         if (rc != 0)
2674                                 RETURN(rc);
2675                         v3 = info->lti_ea_store;
2676                 }
2677
2678                 memset(v3, 0, sizeof(*v3));
2679                 v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
2680                 v3->lmm_stripe_count = cpu_to_le16(lo->ldo_def_stripenr);
2681                 v3->lmm_stripe_offset = cpu_to_le16(lo->ldo_def_stripe_offset);
2682                 v3->lmm_stripe_size = cpu_to_le32(lo->ldo_def_stripe_size);
2683                 if (lo->ldo_pool != NULL)
2684                         strlcpy(v3->lmm_pool_name, lo->ldo_pool,
2685                                 sizeof(v3->lmm_pool_name));
2686
2687                 info->lti_buf.lb_buf = v3;
2688                 info->lti_buf.lb_len = sizeof(*v3);
2689
2690                 if (declare)
2691                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
2692                                                        XATTR_NAME_LOV, 0, th);
2693                 else
2694                         rc = lod_xattr_set_lov_on_dir(env, dt, &info->lti_buf,
2695                                                       XATTR_NAME_LOV, 0, th,
2696                                                       BYPASS_CAPA);
2697                 if (rc != 0)
2698                         RETURN(rc);
2699         }
2700
2701         RETURN(0);
2702 }
2703
2704 static int lod_declare_dir_striping_create(const struct lu_env *env,
2705                                            struct dt_object *dt,
2706                                            struct lu_attr *attr,
2707                                            struct dt_object_format *dof,
2708                                            struct thandle *th)
2709 {
2710         return lod_dir_striping_create_internal(env, dt, attr, dof, th, true);
2711 }
2712
2713 static int lod_dir_striping_create(const struct lu_env *env,
2714                                    struct dt_object *dt,
2715                                    struct lu_attr *attr,
2716                                    struct dt_object_format *dof,
2717                                    struct thandle *th)
2718 {
2719         return lod_dir_striping_create_internal(env, dt, attr, dof, th, false);
2720 }
2721
2722 /**
2723  * Implementation of dt_object_operations::do_xattr_set.
2724  *
2725  * Sets specified extended attribute on the object. Three types of EAs are
2726  * special:
2727  *   LOV EA - stores striping for a regular file or default striping (when set
2728  *            on a directory)
2729  *   LMV EA - stores a marker for the striped directories
2730  *   DMV EA - stores default directory striping
2731  *
2732  * When striping is applied to a non-striped existing object (this is called
2733  * late striping), then LOD notices the caller wants to turn the object into a
2734  * striped one. The stripe objects are created and appropriate EA is set:
2735  * LOV EA storing all the stripes directly or LMV EA storing just a small header
2736  * with striping configuration.
2737  *
2738  * \see dt_object_operations::do_xattr_set() in the API description for details.
2739  */
2740 static int lod_xattr_set(const struct lu_env *env,
2741                          struct dt_object *dt, const struct lu_buf *buf,
2742                          const char *name, int fl, struct thandle *th,
2743                          struct lustre_capa *capa)
2744 {
2745         struct dt_object        *next = dt_object_child(dt);
2746         int                      rc;
2747         ENTRY;
2748
2749         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
2750             strcmp(name, XATTR_NAME_LMV) == 0) {
2751                 struct lmv_mds_md_v1 *lmm = buf->lb_buf;
2752
2753                 if (lmm != NULL && le32_to_cpu(lmm->lmv_hash_type) &
2754                                                 LMV_HASH_FLAG_MIGRATION)
2755                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
2756                 else
2757                         rc = lod_dir_striping_create(env, dt, NULL, NULL, th);
2758
2759                 RETURN(rc);
2760         }
2761
2762         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
2763             strcmp(name, XATTR_NAME_LOV) == 0) {
2764                 /* default LOVEA */
2765                 rc = lod_xattr_set_lov_on_dir(env, dt, buf, name, fl, th, capa);
2766                 RETURN(rc);
2767         } else if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
2768                    strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
2769                 /* default LMVEA */
2770                 rc = lod_xattr_set_default_lmv_on_dir(env, dt, buf, name, fl,
2771                                                       th, capa);
2772                 RETURN(rc);
2773         } else if (S_ISREG(dt->do_lu.lo_header->loh_attr) &&
2774                    !strcmp(name, XATTR_NAME_LOV)) {
2775                 /* in case of lov EA swap, just set it
2776                  * if not, it is a replay so check striping match what we
2777                  * already have during req replay, declare_xattr_set()
2778                  * defines striping, then create() does the work
2779                 */
2780                 if (fl & LU_XATTR_REPLACE) {
2781                         /* free stripes, then update disk */
2782                         lod_object_free_striping(env, lod_dt_obj(dt));
2783                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
2784                 } else {
2785                         rc = lod_striping_create(env, dt, NULL, NULL, th);
2786                 }
2787                 RETURN(rc);
2788         }
2789
2790         /* then all other xattr */
2791         rc = lod_xattr_set_internal(env, dt, buf, name, fl, th, capa);
2792
2793         RETURN(rc);
2794 }
2795
2796 /**
2797  * Implementation of dt_object_operations::do_declare_xattr_del.
2798  *
2799  * \see dt_object_operations::do_declare_xattr_del() in the API description
2800  * for details.
2801  */
2802 static int lod_declare_xattr_del(const struct lu_env *env,
2803                                  struct dt_object *dt, const char *name,
2804                                  struct thandle *th)
2805 {
2806         return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
2807 }
2808
2809 /**
2810  * Implementation of dt_object_operations::do_xattr_del.
2811  *
2812  * If EA storing a regular striping is being deleted, then release
2813  * all the references to the stripe objects in core.
2814  *
2815  * \see dt_object_operations::do_xattr_del() in the API description for details.
2816  */
2817 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
2818                          const char *name, struct thandle *th,
2819                          struct lustre_capa *capa)
2820 {
2821         if (!strcmp(name, XATTR_NAME_LOV))
2822                 lod_object_free_striping(env, lod_dt_obj(dt));
2823         return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
2824 }
2825
2826 /**
2827  * Implementation of dt_object_operations::do_xattr_list.
2828  *
2829  * \see dt_object_operations::do_xattr_list() in the API description
2830  * for details.
2831  */
2832 static int lod_xattr_list(const struct lu_env *env,
2833                           struct dt_object *dt, const struct lu_buf *buf,
2834                           struct lustre_capa *capa)
2835 {
2836         return dt_xattr_list(env, dt_object_child(dt), buf, capa);
2837 }
2838
2839 /**
2840  * Initialize a pool the object belongs to.
2841  *
2842  * When a striped object is being created, striping configuration
2843  * may demand the stripes are allocated on a limited set of the
2844  * targets. These limited sets are known as "pools". So we copy
2845  * a pool name into the object and later actual creation methods
2846  * (like lod_object_create()) will use this information to allocate
2847  * the stripes properly.
2848  *
2849  * \param[in] o         object
2850  * \param[in] pool      pool name
2851  */
2852 int lod_object_set_pool(struct lod_object *o, char *pool)
2853 {
2854         int len;
2855
2856         if (o->ldo_pool) {
2857                 len = strlen(o->ldo_pool);
2858                 OBD_FREE(o->ldo_pool, len + 1);
2859                 o->ldo_pool = NULL;
2860         }
2861         if (pool) {
2862                 len = strlen(pool);
2863                 OBD_ALLOC(o->ldo_pool, len + 1);
2864                 if (o->ldo_pool == NULL)
2865                         return -ENOMEM;
2866                 strcpy(o->ldo_pool, pool);
2867         }
2868         return 0;
2869 }
2870
2871 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
2872 {
2873         return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
2874 }
2875
2876
2877 /**
2878  * Cache default regular striping in the object.
2879  *
2880  * To improve performance of striped regular object creation we cache
2881  * default LOV striping (if it exists) in the parent directory object.
2882  *
2883  * \param[in] env               execution environment
2884  * \param[in] lp                object
2885  *
2886  * \retval              0 on success
2887  * \retval              negative if failed
2888  */
2889 static int lod_cache_parent_lov_striping(const struct lu_env *env,
2890                                          struct lod_object *lp)
2891 {
2892         struct lod_thread_info  *info = lod_env_info(env);
2893         struct lov_user_md_v1   *v1 = NULL;
2894         struct lov_user_md_v3   *v3 = NULL;
2895         int                      rc;
2896         ENTRY;
2897
2898         /* called from MDD without parent being write locked,
2899          * lock it here */
2900         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
2901         rc = lod_get_lov_ea(env, lp);
2902         if (rc < 0)
2903                 GOTO(unlock, rc);
2904
2905         if (rc < (typeof(rc))sizeof(struct lov_user_md)) {
2906                 /* don't lookup for non-existing or invalid striping */
2907                 lp->ldo_def_striping_set = 0;
2908                 lp->ldo_striping_cached = 1;
2909                 lp->ldo_def_stripe_size = 0;
2910                 lp->ldo_def_stripenr = 0;
2911                 lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1);
2912                 GOTO(unlock, rc = 0);
2913         }
2914
2915         rc = 0;
2916         v1 = info->lti_ea_store;
2917         if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1)) {
2918                 lustre_swab_lov_user_md_v1(v1);
2919         } else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3)) {
2920                 v3 = (struct lov_user_md_v3 *)v1;
2921                 lustre_swab_lov_user_md_v3(v3);
2922         }
2923
2924         if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1)
2925                 GOTO(unlock, rc = 0);
2926
2927         if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
2928                 GOTO(unlock, rc = 0);
2929
2930         CDEBUG(D_INFO, DFID" stripe_count=%d stripe_size=%d stripe_offset=%d\n",
2931                PFID(lu_object_fid(&lp->ldo_obj.do_lu)),
2932                (int)v1->lmm_stripe_count,
2933                (int)v1->lmm_stripe_size, (int)v1->lmm_stripe_offset);
2934
2935         lp->ldo_def_stripenr = v1->lmm_stripe_count;
2936         lp->ldo_def_stripe_size = v1->lmm_stripe_size;
2937         lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
2938         lp->ldo_striping_cached = 1;
2939         lp->ldo_def_striping_set = 1;
2940         if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
2941                 /* XXX: sanity check here */
2942                 v3 = (struct lov_user_md_v3 *) v1;
2943                 if (v3->lmm_pool_name[0])
2944                         lod_object_set_pool(lp, v3->lmm_pool_name);
2945         }
2946         EXIT;
2947 unlock:
2948         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
2949         return rc;
2950 }
2951
2952
2953 /**
2954  * Cache default directory striping in the object.
2955  *
2956  * To improve performance of striped directory creation we cache default
2957  * directory striping (if it exists) in the parent directory object.
2958  *
2959  * \param[in] env               execution environment
2960  * \param[in] lp                object
2961  *
2962  * \retval              0 on success
2963  * \retval              negative if failed
2964  */
2965 static int lod_cache_parent_lmv_striping(const struct lu_env *env,
2966                                          struct lod_object *lp)
2967 {
2968         struct lod_thread_info  *info = lod_env_info(env);
2969         struct lmv_user_md_v1   *v1 = NULL;
2970         int                      rc;
2971         ENTRY;
2972
2973         /* called from MDD without parent being write locked,
2974          * lock it here */
2975         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
2976         rc = lod_get_default_lmv_ea(env, lp);
2977         if (rc < 0)
2978                 GOTO(unlock, rc);
2979
2980         if (rc < (typeof(rc))sizeof(struct lmv_user_md)) {
2981                 /* don't lookup for non-existing or invalid striping */
2982                 lp->ldo_dir_def_striping_set = 0;
2983                 lp->ldo_dir_striping_cached = 1;
2984                 lp->ldo_dir_def_stripenr = 0;
2985                 lp->ldo_dir_def_stripe_offset =
2986                                         (typeof(v1->lum_stripe_offset))(-1);
2987                 lp->ldo_dir_def_hash_type = LMV_HASH_TYPE_FNV_1A_64;
2988                 GOTO(unlock, rc = 0);
2989         }
2990
2991         rc = 0;
2992         v1 = info->lti_ea_store;
2993
2994         lp->ldo_dir_def_stripenr = le32_to_cpu(v1->lum_stripe_count);
2995         lp->ldo_dir_def_stripe_offset = le32_to_cpu(v1->lum_stripe_offset);
2996         lp->ldo_dir_def_hash_type = le32_to_cpu(v1->lum_hash_type);
2997         lp->ldo_dir_def_striping_set = 1;
2998         lp->ldo_dir_striping_cached = 1;
2999
3000         EXIT;
3001 unlock:
3002         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
3003         return rc;
3004 }
3005
3006 /**
3007  * Cache default striping in the object.
3008  *
3009  * To improve performance of striped object creation we cache default striping
3010  * (if it exists) in the parent directory object. We always cache default
3011  * striping for the regular files (stored in LOV EA) and we cache default
3012  * striping for the directories if requested by \a child_mode (when a new
3013  * directory is being created).
3014  *
3015  * \param[in] env               execution environment
3016  * \param[in] lp                object
3017  * \param[in] child_mode        new object's mode
3018  *
3019  * \retval              0 on success
3020  * \retval              negative if failed
3021  */
3022 static int lod_cache_parent_striping(const struct lu_env *env,
3023                                      struct lod_object *lp,
3024                                      umode_t child_mode)
3025 {
3026         int rc = 0;
3027         ENTRY;
3028
3029         rc = lod_load_striping(env, lp);
3030         if (rc != 0)
3031                 RETURN(rc);
3032
3033         if (!lp->ldo_striping_cached) {
3034                 /* we haven't tried to get default striping for
3035                  * the directory yet, let's cache it in the object */
3036                 rc = lod_cache_parent_lov_striping(env, lp);
3037                 if (rc != 0)
3038                         RETURN(rc);
3039         }
3040
3041         if (S_ISDIR(child_mode) && !lp->ldo_dir_striping_cached)
3042                 rc = lod_cache_parent_lmv_striping(env, lp);
3043
3044         RETURN(rc);
3045 }
3046
3047 /**
3048  * Implementation of dt_object_operations::do_ah_init.
3049  *
3050  * This method is used to make a decision on the striping configuration for the
3051  * object being created. It can be taken from the \a parent object if it exists,
3052  * or filesystem's default. The resulting configuration (number of stripes,
3053  * stripe size/offset, pool name, etc) is stored in the object itself and will
3054  * be used by the methods like ->doo_declare_create().
3055  *
3056  * \see dt_object_operations::do_ah_init() in the API description for details.
3057  */
3058 static void lod_ah_init(const struct lu_env *env,
3059                         struct dt_allocation_hint *ah,
3060                         struct dt_object *parent,
3061                         struct dt_object *child,
3062                         umode_t child_mode)
3063 {
3064         struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
3065         struct dt_object  *nextp = NULL;
3066         struct dt_object  *nextc;
3067         struct lod_object *lp = NULL;
3068         struct lod_object *lc;
3069         struct lov_desc   *desc;
3070         int               rc;
3071         ENTRY;
3072
3073         LASSERT(child);
3074
3075         if (likely(parent)) {
3076                 nextp = dt_object_child(parent);
3077                 lp = lod_dt_obj(parent);
3078                 rc = lod_load_striping(env, lp);
3079                 if (rc != 0)
3080                         return;
3081         }
3082
3083         nextc = dt_object_child(child);
3084         lc = lod_dt_obj(child);
3085
3086         LASSERT(lc->ldo_stripenr == 0);
3087         LASSERT(lc->ldo_stripe == NULL);
3088
3089         /*
3090          * local object may want some hints
3091          * in case of late striping creation, ->ah_init()
3092          * can be called with local object existing
3093          */
3094         if (!dt_object_exists(nextc) || dt_object_remote(nextc)) {
3095                 struct dt_object *obj;
3096
3097                 obj = (nextp != NULL && dt_object_remote(nextp)) ? NULL : nextp;
3098                 nextc->do_ops->do_ah_init(env, ah, obj, nextc, child_mode);
3099         }
3100
3101         if (S_ISDIR(child_mode)) {
3102                 if (lc->ldo_dir_stripe == NULL) {
3103                         OBD_ALLOC_PTR(lc->ldo_dir_stripe);
3104                         if (lc->ldo_dir_stripe == NULL)
3105                                 return;
3106                 }
3107
3108                 if (lp->ldo_dir_stripe == NULL) {
3109                         OBD_ALLOC_PTR(lp->ldo_dir_stripe);
3110                         if (lp->ldo_dir_stripe == NULL)
3111                                 return;
3112                 }
3113
3114                 rc = lod_cache_parent_striping(env, lp, child_mode);
3115                 if (rc != 0)
3116                         return;
3117
3118                 /* transfer defaults to new directory */
3119                 if (lp->ldo_striping_cached) {
3120                         if (lp->ldo_pool)
3121                                 lod_object_set_pool(lc, lp->ldo_pool);
3122                         lc->ldo_def_stripenr = lp->ldo_def_stripenr;
3123                         lc->ldo_def_stripe_size = lp->ldo_def_stripe_size;
3124                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
3125                         lc->ldo_striping_cached = 1;
3126                         lc->ldo_def_striping_set = 1;
3127                         CDEBUG(D_OTHER, "inherite EA sz:%d off:%d nr:%d\n",
3128                                (int)lc->ldo_def_stripe_size,
3129                                (int)lc->ldo_def_stripe_offset,
3130                                (int)lc->ldo_def_stripenr);
3131                 }
3132
3133                 /* transfer dir defaults to new directory */
3134                 if (lp->ldo_dir_striping_cached) {
3135                         lc->ldo_dir_def_stripenr = lp->ldo_dir_def_stripenr;
3136                         lc->ldo_dir_def_stripe_offset =
3137                                                   lp->ldo_dir_def_stripe_offset;
3138                         lc->ldo_dir_def_hash_type =
3139                                                   lp->ldo_dir_def_hash_type;
3140                         lc->ldo_dir_striping_cached = 1;
3141                         lc->ldo_dir_def_striping_set = 1;
3142                         CDEBUG(D_INFO, "inherit default EA nr:%d off:%d t%u\n",
3143                                (int)lc->ldo_dir_def_stripenr,
3144                                (int)lc->ldo_dir_def_stripe_offset,
3145                                lc->ldo_dir_def_hash_type);
3146                 }
3147
3148                 /* It should always honour the specified stripes */
3149                 if (ah->dah_eadata != NULL && ah->dah_eadata_len != 0) {
3150                         const struct lmv_user_md_v1 *lum1 = ah->dah_eadata;
3151
3152                         rc = lod_verify_md_striping(d, lum1);
3153                         if (rc == 0 &&
3154                                 le32_to_cpu(lum1->lum_stripe_count) > 1) {
3155                                 /* Directory will be striped only if
3156                                  * stripe_count > 1 */
3157                                 lc->ldo_stripenr =
3158                                         le32_to_cpu(lum1->lum_stripe_count);
3159                                 lc->ldo_dir_stripe_offset =
3160                                         le32_to_cpu(lum1->lum_stripe_offset);
3161                                 lc->ldo_dir_hash_type =
3162                                         le32_to_cpu(lum1->lum_hash_type);
3163                                 CDEBUG(D_INFO, "set stripe EA nr:%hu off:%d\n",
3164                                        lc->ldo_stripenr,
3165                                        (int)lc->ldo_dir_stripe_offset);
3166                         }
3167                 /* then check whether there is default stripes from parent */
3168                 } else if (lp->ldo_dir_def_striping_set) {
3169                         /* If there are default dir stripe from parent */
3170                         lc->ldo_stripenr = lp->ldo_dir_def_stripenr;
3171                         lc->ldo_dir_stripe_offset =
3172                                         lp->ldo_dir_def_stripe_offset;
3173                         lc->ldo_dir_hash_type =
3174                                         lp->ldo_dir_def_hash_type;
3175                         CDEBUG(D_INFO, "inherit EA nr:%hu off:%d\n",
3176                                lc->ldo_stripenr,
3177                                (int)lc->ldo_dir_stripe_offset);
3178                 } else {
3179                         /* set default stripe for this directory */
3180                         lc->ldo_stripenr = 0;
3181                         lc->ldo_dir_stripe_offset = -1;
3182                 }
3183
3184                 CDEBUG(D_INFO, "final striping count:%hu, offset:%d\n",
3185                        lc->ldo_stripenr, (int)lc->ldo_dir_stripe_offset);
3186
3187                 goto out;
3188         }
3189
3190         /*
3191          * if object is going to be striped over OSTs, transfer default
3192          * striping information to the child, so that we can use it
3193          * during declaration and creation
3194          */
3195         if (!lod_object_will_be_striped(S_ISREG(child_mode),
3196                                         lu_object_fid(&child->do_lu)))
3197                 goto out;
3198         /*
3199          * try from the parent
3200          */
3201         if (likely(parent)) {
3202                 lod_cache_parent_striping(env, lp, child_mode);
3203
3204                 lc->ldo_def_stripe_offset = LOV_OFFSET_DEFAULT;
3205
3206                 if (lp->ldo_def_striping_set) {
3207                         if (lp->ldo_pool)
3208                                 lod_object_set_pool(lc, lp->ldo_pool);
3209                         lc->ldo_stripenr = lp->ldo_def_stripenr;
3210                         lc->ldo_stripe_size = lp->ldo_def_stripe_size;
3211                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
3212                         CDEBUG(D_OTHER, "striping from parent: #%d, sz %d %s\n",
3213                                lc->ldo_stripenr, lc->ldo_stripe_size,
3214                                lp->ldo_pool ? lp->ldo_pool : "");
3215                 }
3216         }
3217
3218         /*
3219          * if the parent doesn't provide with specific pattern, grab fs-wide one
3220          */
3221         desc = &d->lod_desc;
3222         if (lc->ldo_stripenr == 0)
3223                 lc->ldo_stripenr = desc->ld_default_stripe_count;
3224         if (lc->ldo_stripe_size == 0)
3225                 lc->ldo_stripe_size = desc->ld_default_stripe_size;
3226         CDEBUG(D_OTHER, "final striping: # %d stripes, sz %d from %s\n",
3227                lc->ldo_stripenr, lc->ldo_stripe_size,
3228                lc->ldo_pool ? lc->ldo_pool : "");
3229
3230 out:
3231         /* we do not cache stripe information for slave stripe, see
3232          * lod_xattr_set_lov_on_dir */
3233         if (lp != NULL && lp->ldo_dir_slave_stripe)
3234                 lod_lov_stripe_cache_clear(lp);
3235
3236         EXIT;
3237 }
3238
3239 #define ll_do_div64(aaa,bbb)    do_div((aaa), (bbb))
3240 /**
3241  * Size initialization on late striping.
3242  *
3243  * Propagate the size of a truncated object to a deferred striping.
3244  * This function handles a special case when truncate was done on a
3245  * non-striped object and now while the striping is being created
3246  * we can't lose that size, so we have to propagate it to the stripes
3247  * being created.
3248  *
3249  * \param[in] env       execution environment
3250  * \param[in] dt        object
3251  * \param[in] th        transaction handle
3252  *
3253  * \retval              0 on success
3254  * \retval              negative if failed
3255  */
3256 static int lod_declare_init_size(const struct lu_env *env,
3257                                  struct dt_object *dt, struct thandle *th)
3258 {
3259         struct dt_object   *next = dt_object_child(dt);
3260         struct lod_object  *lo = lod_dt_obj(dt);
3261         struct lu_attr     *attr = &lod_env_info(env)->lti_attr;
3262         uint64_t            size, offs;
3263         int                 rc, stripe;
3264         ENTRY;
3265
3266         /* XXX: we support the simplest (RAID0) striping so far */
3267         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
3268         LASSERT(lo->ldo_stripe_size > 0);
3269
3270         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
3271         LASSERT(attr->la_valid & LA_SIZE);
3272         if (rc)
3273                 RETURN(rc);
3274
3275         size = attr->la_size;
3276         if (size == 0)
3277                 RETURN(0);
3278
3279         /* ll_do_div64(a, b) returns a % b, and a = a / b */
3280         ll_do_div64(size, (__u64) lo->ldo_stripe_size);
3281         stripe = ll_do_div64(size, (__u64) lo->ldo_stripenr);
3282
3283         size = size * lo->ldo_stripe_size;
3284         offs = attr->la_size;
3285         size += ll_do_div64(offs, lo->ldo_stripe_size);
3286
3287         attr->la_valid = LA_SIZE;
3288         attr->la_size = size;
3289
3290         rc = dt_declare_attr_set(env, lo->ldo_stripe[stripe], attr, th);
3291
3292         RETURN(rc);
3293 }
3294
3295 /**
3296  * Declare creation of striped object.
3297  *
3298  * The function declares creation stripes for a regular object. The function
3299  * also declares whether the stripes will be created with non-zero size if
3300  * previously size was set non-zero on the master object. If object \a dt is
3301  * not local, then only fully defined striping can be applied in \a lovea.
3302  * Otherwise \a lovea can be in the form of pattern, see lod_qos_parse_config()
3303  * for the details.
3304  *
3305  * \param[in] env       execution environment
3306  * \param[in] dt        object
3307  * \param[in] attr      attributes the stripes will be created with
3308  * \param[in] lovea     a buffer containing striping description
3309  * \param[in] th        transaction handle
3310  *
3311  * \retval              0 on success
3312  * \retval              negative if failed
3313  */
3314 int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt,
3315                                struct lu_attr *attr,
3316                                const struct lu_buf *lovea, struct thandle *th)
3317 {
3318         struct lod_thread_info  *info = lod_env_info(env);
3319         struct dt_object        *next = dt_object_child(dt);
3320         struct lod_object       *lo = lod_dt_obj(dt);
3321         int                      rc;
3322         ENTRY;
3323
3324         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) {
3325                 /* failed to create striping, let's reset
3326                  * config so that others don't get confused */
3327                 lod_object_free_striping(env, lo);
3328                 GOTO(out, rc = -ENOMEM);
3329         }
3330
3331         if (!dt_object_remote(next)) {
3332                 /* choose OST and generate appropriate objects */
3333                 rc = lod_qos_prep_create(env, lo, attr, lovea, th);
3334                 if (rc) {
3335                         /* failed to create striping, let's reset
3336                          * config so that others don't get confused */
3337                         lod_object_free_striping(env, lo);
3338                         GOTO(out, rc);
3339                 }
3340
3341                 /*
3342                  * declare storage for striping data
3343                  */
3344                 info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
3345                                 lo->ldo_pool ?  LOV_MAGIC_V3 : LOV_MAGIC_V1);
3346         } else {
3347                 /* LOD can not choose OST objects for remote objects, i.e.
3348                  * stripes must be ready before that. Right now, it can only
3349                  * happen during migrate, i.e. migrate process needs to create
3350                  * remote regular file (mdd_migrate_create), then the migrate
3351                  * process will provide stripeEA. */
3352                 LASSERT(lovea != NULL);
3353                 info->lti_buf = *lovea;
3354         }
3355
3356         rc = dt_declare_xattr_set(env, next, &info->lti_buf,
3357                                   XATTR_NAME_LOV, 0, th);
3358         if (rc)
3359                 GOTO(out, rc);
3360
3361         /*
3362          * if striping is created with local object's size > 0,
3363          * we have to propagate this size to specific object
3364          * the case is possible only when local object was created previously
3365          */
3366         if (dt_object_exists(next))
3367                 rc = lod_declare_init_size(env, dt, th);
3368
3369 out:
3370         RETURN(rc);
3371 }
3372
3373 /**
3374  * Implementation of dt_object_operations::do_declare_create.
3375  *
3376  * The method declares creation of a new object. If the object will be striped,
3377  * then helper functions are called to find FIDs for the stripes, declare
3378  * creation of the stripes and declare initialization of the striping
3379  * information to be stored in the master object.
3380  *
3381  * \see dt_object_operations::do_declare_create() in the API description
3382  * for details.
3383  */
3384 static int lod_declare_object_create(const struct lu_env *env,
3385                                      struct dt_object *dt,
3386                                      struct lu_attr *attr,
3387                                      struct dt_allocation_hint *hint,
3388                                      struct dt_object_format *dof,
3389                                      struct thandle *th)
3390 {
3391         struct dt_object   *next = dt_object_child(dt);
3392         struct lod_object  *lo = lod_dt_obj(dt);
3393         int                 rc;
3394         ENTRY;
3395
3396         LASSERT(dof);
3397         LASSERT(attr);
3398         LASSERT(th);
3399
3400         /*
3401          * first of all, we declare creation of local object
3402          */
3403         rc = dt_declare_create(env, next, attr, hint, dof, th);
3404         if (rc)
3405                 GOTO(out, rc);
3406
3407         if (dof->dof_type == DFT_SYM)
3408                 dt->do_body_ops = &lod_body_lnk_ops;
3409
3410         /*
3411          * it's lod_ah_init() who has decided the object will striped
3412          */
3413         if (dof->dof_type == DFT_REGULAR) {
3414                 /* callers don't want stripes */
3415                 /* XXX: all tricky interactions with ->ah_make_hint() decided
3416                  * to use striping, then ->declare_create() behaving differently
3417                  * should be cleaned */
3418                 if (dof->u.dof_reg.striped == 0)
3419                         lo->ldo_stripenr = 0;
3420                 if (lo->ldo_stripenr > 0)
3421                         rc = lod_declare_striped_object(env, dt, attr,
3422                                                         NULL, th);
3423         } else if (dof->dof_type == DFT_DIR) {
3424                 /* Orphan object (like migrating object) does not have
3425                  * lod_dir_stripe, see lod_ah_init */
3426                 if (lo->ldo_dir_stripe != NULL)
3427                         rc = lod_declare_dir_striping_create(env, dt, attr,
3428                                                              dof, th);
3429         }
3430 out:
3431         RETURN(rc);
3432 }
3433
3434 /**
3435  * Creation of a striped regular object.
3436  *
3437  * The function is called to create the stripe objects for a regular
3438  * striped file. This can happen at the initial object creation or
3439  * when the caller asks LOD to do so using ->do_xattr_set() method
3440  * (so called late striping). Notice all the information are already
3441  * prepared in the form of the list of objects (ldo_stripe field).
3442  * This is done during declare phase.
3443  *
3444  * \param[in] env       execution environment
3445  * \param[in] dt        object
3446  * \param[in] attr      attributes the stripes will be created with
3447  * \param[in] dof       format of stripes (see OSD API description)
3448  * \param[in] th        transaction handle
3449  *
3450  * \retval              0 on success
3451  * \retval              negative if failed
3452  */
3453 int lod_striping_create(const struct lu_env *env, struct dt_object *dt,
3454                         struct lu_attr *attr, struct dt_object_format *dof,
3455                         struct thandle *th)
3456 {
3457         struct lod_object *lo = lod_dt_obj(dt);
3458         int                rc = 0, i;
3459         ENTRY;
3460
3461         LASSERT(lo->ldo_striping_cached == 0);
3462
3463         /* create all underlying objects */
3464         for (i = 0; i < lo->ldo_stripenr; i++) {
3465                 LASSERT(lo->ldo_stripe[i]);
3466                 rc = dt_create(env, lo->ldo_stripe[i], attr, NULL, dof, th);
3467
3468                 if (rc)
3469                         break;
3470         }
3471         if (rc == 0)
3472                 rc = lod_generate_and_set_lovea(env, lo, th);
3473
3474         RETURN(rc);
3475 }
3476
3477 /**
3478  * Implementation of dt_object_operations::do_create.
3479  *
3480  * If any of preceeding methods (like ->do_declare_create(),
3481  * ->do_ah_init(), etc) chose to create a striped object,
3482  * then this method will create the master and the stripes.
3483  *
3484  * \see dt_object_operations::do_create() in the API description for details.
3485  */
3486 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
3487                              struct lu_attr *attr,
3488                              struct dt_allocation_hint *hint,
3489                              struct dt_object_format *dof, struct thandle *th)
3490 {
3491         struct dt_object   *next = dt_object_child(dt);
3492         struct lod_object  *lo = lod_dt_obj(dt);
3493         int                 rc;
3494         ENTRY;
3495
3496         /* create local object */
3497         rc = dt_create(env, next, attr, hint, dof, th);
3498         if (rc != 0)
3499                 RETURN(rc);
3500
3501         if (S_ISREG(dt->do_lu.lo_header->loh_attr) &&
3502             lo->ldo_stripe && dof->u.dof_reg.striped != 0)
3503                 rc = lod_striping_create(env, dt, attr, dof, th);
3504
3505         RETURN(rc);
3506 }
3507
3508 /**
3509  * Implementation of dt_object_operations::do_declare_destroy.
3510  *
3511  * If the object is a striped directory, then the function declares reference
3512  * removal from the master object (this is an index) to the stripes and declares
3513  * destroy of all the stripes. In all the cases, it declares an intention to
3514  * destroy the object itself.
3515  *
3516  * \see dt_object_operations::do_declare_destroy() in the API description
3517  * for details.
3518  */
3519 static int lod_declare_object_destroy(const struct lu_env *env,
3520                                       struct dt_object *dt,
3521                                       struct thandle *th)
3522 {
3523         struct dt_object   *next = dt_object_child(dt);
3524         struct lod_object  *lo = lod_dt_obj(dt);
3525         struct lod_thread_info *info = lod_env_info(env);
3526         char               *stripe_name = info->lti_key;
3527         int                 rc, i;
3528         ENTRY;
3529
3530         /*
3531          * load striping information, notice we don't do this when object
3532          * is being initialized as we don't need this information till
3533          * few specific cases like destroy, chown
3534          */
3535         rc = lod_load_striping(env, lo);
3536         if (rc)
3537                 RETURN(rc);
3538
3539         /* declare destroy for all underlying objects */
3540         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
3541                 rc = next->do_ops->do_index_try(env, next,
3542                                                 &dt_directory_features);
3543                 if (rc != 0)
3544                         RETURN(rc);
3545
3546                 for (i = 0; i < lo->ldo_stripenr; i++) {
3547                         rc = dt_declare_ref_del(env, next, th);
3548                         if (rc != 0)
3549                                 RETURN(rc);
3550                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
3551                                 PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)),
3552                                 i);
3553                         rc = dt_declare_delete(env, next,
3554                                         (const struct dt_key *)stripe_name, th);
3555                         if (rc != 0)
3556                                 RETURN(rc);
3557                 }
3558         }
3559         /*
3560          * we declare destroy for the local object
3561          */
3562         rc = dt_declare_destroy(env, next, th);
3563         if (rc)
3564                 RETURN(rc);
3565
3566         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ))
3567                 RETURN(0);
3568
3569         /* declare destroy all striped objects */
3570         for (i = 0; i < lo->ldo_stripenr; i++) {
3571                 if (likely(lo->ldo_stripe[i] != NULL)) {
3572                         rc = dt_declare_destroy(env, lo->ldo_stripe[i], th);
3573                         if (rc != 0)
3574                                 break;
3575                 }
3576         }
3577
3578         RETURN(rc);
3579 }
3580
3581 /**
3582  * Implementation of dt_object_operations::do_destroy.
3583  *
3584  * If the object is a striped directory, then the function removes references
3585  * from the master object (this is an index) to the stripes and destroys all
3586  * the stripes. In all the cases, the function destroys the object itself.
3587  *
3588  * \see dt_object_operations::do_destroy() in the API description for details.
3589  */
3590 static int lod_object_destroy(const struct lu_env *env,
3591                 struct dt_object *dt, struct thandle *th)
3592 {
3593         struct dt_object  *next = dt_object_child(dt);
3594         struct lod_object *lo = lod_dt_obj(dt);
3595         struct lod_thread_info *info = lod_env_info(env);
3596         char               *stripe_name = info->lti_key;
3597         unsigned int       i;
3598         int                rc;
3599         ENTRY;
3600
3601         /* destroy sub-stripe of master object */
3602         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
3603                 rc = next->do_ops->do_index_try(env, next,
3604                                                 &dt_directory_features);
3605                 if (rc != 0)
3606                         RETURN(rc);
3607
3608                 for (i = 0; i < lo->ldo_stripenr; i++) {
3609                         rc = dt_ref_del(env, next, th);
3610                         if (rc != 0)
3611                                 RETURN(rc);
3612
3613                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
3614                                 PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)),
3615                                 i);
3616
3617                         CDEBUG(D_INFO, DFID" delete stripe %s "DFID"\n",
3618                                PFID(lu_object_fid(&dt->do_lu)), stripe_name,
3619                                PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)));
3620
3621                         rc = dt_delete(env, next,
3622                                        (const struct dt_key *)stripe_name,
3623                                        th, BYPASS_CAPA);
3624                         if (rc != 0)
3625                                 RETURN(rc);
3626                 }
3627         }
3628         rc = dt_destroy(env, next, th);
3629         if (rc != 0)
3630                 RETURN(rc);
3631
3632         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ))
3633                 RETURN(0);
3634
3635         /* destroy all striped objects */
3636         for (i = 0; i < lo->ldo_stripenr; i++) {
3637                 if (likely(lo->ldo_stripe[i] != NULL) &&
3638                     (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SPEOBJ) ||
3639                      i == cfs_fail_val)) {
3640                         rc = dt_destroy(env, lo->ldo_stripe[i], th);
3641                         if (rc != 0)
3642                                 break;
3643                 }
3644         }
3645
3646         RETURN(rc);
3647 }
3648
3649 /**
3650  * Implementation of dt_object_operations::do_declare_ref_add.
3651  *
3652  * \see dt_object_operations::do_declare_ref_add() in the API description
3653  * for details.
3654  */
3655 static int lod_declare_ref_add(const struct lu_env *env,
3656                                struct dt_object *dt, struct thandle *th)
3657 {
3658         return dt_declare_ref_add(env, dt_object_child(dt), th);
3659 }
3660
3661 /**
3662  * Implementation of dt_object_operations::do_ref_add.
3663  *
3664  * \see dt_object_operations::do_ref_add() in the API description for details.
3665  */
3666 static int lod_ref_add(const struct lu_env *env,
3667                        struct dt_object *dt, struct thandle *th)
3668 {
3669         return dt_ref_add(env, dt_object_child(dt), th);
3670 }
3671
3672 /**
3673  * Implementation of dt_object_operations::do_declare_ref_del.
3674  *
3675  * \see dt_object_operations::do_declare_ref_del() in the API description
3676  * for details.
3677  */
3678 static int lod_declare_ref_del(const struct lu_env *env,
3679                                struct dt_object *dt, struct thandle *th)
3680 {
3681         return dt_declare_ref_del(env, dt_object_child(dt), th);
3682 }
3683
3684 /**
3685  * Implementation of dt_object_operations::do_ref_del
3686  *
3687  * \see dt_object_operations::do_ref_del() in the API description for details.