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