Whamcloud - gitweb
24516ed9848d0b862f10914d66778fcf55858ed9
[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,
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                 /* has stripe but not inited yet, this component has been
1065                  * declared to be created, but hasn't created yet.
1066                  */
1067                 if (!lod_comp_inited(lod_comp))
1068                         continue;
1069
1070                 if (data->locd_comp_skip_cb &&
1071                     data->locd_comp_skip_cb(env, lo, i, data))
1072                         continue;
1073
1074                 LASSERT(lod_comp->llc_stripe_count > 0);
1075                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
1076                         struct dt_object *dt = lod_comp->llc_stripe[j];
1077
1078                         if (dt == NULL)
1079                                 continue;
1080                         rc = data->locd_stripe_cb(env, lo, dt, th, i, j, data);
1081                         if (rc != 0)
1082                                 RETURN(rc);
1083                 }
1084         }
1085         RETURN(0);
1086 }
1087
1088 static bool lod_obj_attr_set_comp_skip_cb(const struct lu_env *env,
1089                 struct lod_object *lo, int comp_idx,
1090                 struct lod_obj_stripe_cb_data *data)
1091 {
1092         struct lod_layout_component *lod_comp = &lo->ldo_comp_entries[comp_idx];
1093         bool skipped = false;
1094
1095         if (!(data->locd_attr->la_valid & LA_LAYOUT_VERSION))
1096                 return skipped;
1097
1098         switch (lo->ldo_flr_state) {
1099         case LCM_FL_WRITE_PENDING: {
1100                 int i;
1101
1102                 /* skip stale components */
1103                 if (lod_comp->llc_flags & LCME_FL_STALE) {
1104                         skipped = true;
1105                         break;
1106                 }
1107
1108                 /* skip valid and overlapping components, therefore any
1109                  * attempts to write overlapped components will never succeed
1110                  * because client will get EINPROGRESS. */
1111                 for (i = 0; i < lo->ldo_comp_cnt; i++) {
1112                         if (i == comp_idx)
1113                                 continue;
1114
1115                         if (lo->ldo_comp_entries[i].llc_flags & LCME_FL_STALE)
1116                                 continue;
1117
1118                         if (lu_extent_is_overlapped(&lod_comp->llc_extent,
1119                                         &lo->ldo_comp_entries[i].llc_extent)) {
1120                                 skipped = true;
1121                                 break;
1122                         }
1123                 }
1124                 break;
1125         }
1126         default:
1127                 LASSERTF(0, "impossible: %d\n", lo->ldo_flr_state);
1128         case LCM_FL_SYNC_PENDING:
1129                 break;
1130         }
1131
1132         CDEBUG(D_LAYOUT, DFID": %s to set component %x to version: %u\n",
1133                PFID(lu_object_fid(&lo->ldo_obj.do_lu)),
1134                skipped ? "skipped" : "chose", lod_comp->llc_id,
1135                data->locd_attr->la_layout_version);
1136
1137         return skipped;
1138 }
1139
1140 static inline int
1141 lod_obj_stripe_attr_set_cb(const struct lu_env *env, struct lod_object *lo,
1142                            struct dt_object *dt, struct thandle *th,
1143                            int comp_idx, int stripe_idx,
1144                            struct lod_obj_stripe_cb_data *data)
1145 {
1146         if (data->locd_declare)
1147                 return lod_sub_declare_attr_set(env, dt, data->locd_attr, th);
1148
1149         if (data->locd_attr->la_valid & LA_LAYOUT_VERSION) {
1150                 CDEBUG(D_LAYOUT, DFID": set layout version: %u, comp_idx: %d\n",
1151                        PFID(lu_object_fid(&dt->do_lu)),
1152                        data->locd_attr->la_layout_version, comp_idx);
1153         }
1154
1155         return lod_sub_attr_set(env, dt, data->locd_attr, th);
1156 }
1157
1158 /**
1159  * Implementation of dt_object_operations::do_declare_attr_set.
1160  *
1161  * If the object is striped, then apply the changes to all the stripes.
1162  *
1163  * \see dt_object_operations::do_declare_attr_set() in the API description
1164  * for details.
1165  */
1166 static int lod_declare_attr_set(const struct lu_env *env,
1167                                 struct dt_object *dt,
1168                                 const struct lu_attr *attr,
1169                                 struct thandle *th)
1170 {
1171         struct dt_object  *next = dt_object_child(dt);
1172         struct lod_object *lo = lod_dt_obj(dt);
1173         int                rc, i;
1174         ENTRY;
1175
1176         /*
1177          * declare setattr on the local object
1178          */
1179         rc = lod_sub_declare_attr_set(env, next, attr, th);
1180         if (rc)
1181                 RETURN(rc);
1182
1183         /* osp_declare_attr_set() ignores all attributes other than
1184          * UID, GID, PROJID, and size, and osp_attr_set() ignores all
1185          * but UID, GID and PROJID. Declaration of size attr setting
1186          * happens through lod_declare_init_size(), and not through
1187          * this function. Therefore we need not load striping unless
1188          * ownership is changing.  This should save memory and (we hope)
1189          * speed up rename().
1190          */
1191         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1192                 if (!(attr->la_valid & LA_REMOTE_ATTR_SET))
1193                         RETURN(rc);
1194
1195                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
1196                         RETURN(0);
1197         } else {
1198                 if (!(attr->la_valid & (LA_UID | LA_GID | LA_PROJID | LA_MODE |
1199                                         LA_ATIME | LA_MTIME | LA_CTIME |
1200                                         LA_FLAGS)))
1201                         RETURN(rc);
1202         }
1203         /*
1204          * load striping information, notice we don't do this when object
1205          * is being initialized as we don't need this information till
1206          * few specific cases like destroy, chown
1207          */
1208         rc = lod_load_striping(env, lo);
1209         if (rc)
1210                 RETURN(rc);
1211
1212         if (!lod_obj_is_striped(dt))
1213                 RETURN(0);
1214
1215         /*
1216          * if object is striped declare changes on the stripes
1217          */
1218         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1219                 LASSERT(lo->ldo_stripe);
1220                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
1221                         if (lo->ldo_stripe[i] == NULL)
1222                                 continue;
1223                         rc = lod_sub_declare_attr_set(env, lo->ldo_stripe[i],
1224                                                       attr, th);
1225                         if (rc != 0)
1226                                 RETURN(rc);
1227                 }
1228         } else {
1229                 struct lod_obj_stripe_cb_data data = { { 0 } };
1230
1231                 data.locd_attr = attr;
1232                 data.locd_declare = true;
1233                 data.locd_stripe_cb = lod_obj_stripe_attr_set_cb;
1234                 rc = lod_obj_for_each_stripe(env, lo, th, &data);
1235         }
1236
1237         if (rc)
1238                 RETURN(rc);
1239
1240         if (!dt_object_exists(next) || dt_object_remote(next) ||
1241             !S_ISREG(attr->la_mode))
1242                 RETURN(0);
1243
1244         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE)) {
1245                 rc = lod_sub_declare_xattr_del(env, next, XATTR_NAME_LOV, th);
1246                 RETURN(rc);
1247         }
1248
1249         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE) ||
1250             OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PFL_RANGE)) {
1251                 struct lod_thread_info *info = lod_env_info(env);
1252                 struct lu_buf *buf = &info->lti_buf;
1253
1254                 buf->lb_buf = info->lti_ea_store;
1255                 buf->lb_len = info->lti_ea_store_size;
1256                 rc = lod_sub_declare_xattr_set(env, next, buf, XATTR_NAME_LOV,
1257                                                LU_XATTR_REPLACE, th);
1258         }
1259
1260         RETURN(rc);
1261 }
1262
1263 /**
1264  * Implementation of dt_object_operations::do_attr_set.
1265  *
1266  * If the object is striped, then apply the changes to all or subset of
1267  * the stripes depending on the object type and specific attributes.
1268  *
1269  * \see dt_object_operations::do_attr_set() in the API description for details.
1270  */
1271 static int lod_attr_set(const struct lu_env *env,
1272                         struct dt_object *dt,
1273                         const struct lu_attr *attr,
1274                         struct thandle *th)
1275 {
1276         struct dt_object        *next = dt_object_child(dt);
1277         struct lod_object       *lo = lod_dt_obj(dt);
1278         int                     rc, i;
1279         ENTRY;
1280
1281         /*
1282          * apply changes to the local object
1283          */
1284         rc = lod_sub_attr_set(env, next, attr, th);
1285         if (rc)
1286                 RETURN(rc);
1287
1288         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1289                 if (!(attr->la_valid & LA_REMOTE_ATTR_SET))
1290                         RETURN(rc);
1291
1292                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
1293                         RETURN(0);
1294         } else {
1295                 if (!(attr->la_valid & (LA_UID | LA_GID | LA_MODE | LA_PROJID |
1296                                         LA_ATIME | LA_MTIME | LA_CTIME |
1297                                         LA_FLAGS)))
1298                         RETURN(rc);
1299         }
1300
1301         /* FIXME: a tricky case in the code path of mdd_layout_change():
1302          * the in-memory striping information has been freed in lod_xattr_set()
1303          * due to layout change. It has to load stripe here again. It only
1304          * changes flags of layout so declare_attr_set() is still accurate */
1305         rc = lod_load_striping_locked(env, lo);
1306         if (rc)
1307                 RETURN(rc);
1308
1309         if (!lod_obj_is_striped(dt))
1310                 RETURN(0);
1311
1312         /*
1313          * if object is striped, apply changes to all the stripes
1314          */
1315         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1316                 LASSERT(lo->ldo_stripe);
1317                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
1318                         if (unlikely(lo->ldo_stripe[i] == NULL))
1319                                 continue;
1320
1321                         if ((dt_object_exists(lo->ldo_stripe[i]) == 0))
1322                                 continue;
1323
1324                         rc = lod_sub_attr_set(env, lo->ldo_stripe[i], attr, th);
1325                         if (rc != 0)
1326                                 break;
1327                 }
1328         } else {
1329                 struct lod_obj_stripe_cb_data data = { { 0 } };
1330
1331                 data.locd_attr = attr;
1332                 data.locd_declare = false;
1333                 data.locd_comp_skip_cb = lod_obj_attr_set_comp_skip_cb;
1334                 data.locd_stripe_cb = lod_obj_stripe_attr_set_cb;
1335                 rc = lod_obj_for_each_stripe(env, lo, th, &data);
1336         }
1337
1338         if (rc)
1339                 RETURN(rc);
1340
1341         if (!dt_object_exists(next) || dt_object_remote(next) ||
1342             !S_ISREG(attr->la_mode))
1343                 RETURN(0);
1344
1345         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE)) {
1346                 rc = lod_sub_xattr_del(env, next, XATTR_NAME_LOV, th);
1347                 RETURN(rc);
1348         }
1349
1350         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE)) {
1351                 struct lod_thread_info *info = lod_env_info(env);
1352                 struct lu_buf *buf = &info->lti_buf;
1353                 struct ost_id *oi = &info->lti_ostid;
1354                 struct lu_fid *fid = &info->lti_fid;
1355                 struct lov_mds_md_v1 *lmm;
1356                 struct lov_ost_data_v1 *objs;
1357                 __u32 magic;
1358
1359                 rc = lod_get_lov_ea(env, lo);
1360                 if (rc <= 0)
1361                         RETURN(rc);
1362
1363                 buf->lb_buf = info->lti_ea_store;
1364                 buf->lb_len = info->lti_ea_store_size;
1365                 lmm = info->lti_ea_store;
1366                 magic = le32_to_cpu(lmm->lmm_magic);
1367                 if (magic == LOV_MAGIC_COMP_V1) {
1368                         struct lov_comp_md_v1 *lcm = buf->lb_buf;
1369                         struct lov_comp_md_entry_v1 *lcme =
1370                                                 &lcm->lcm_entries[0];
1371
1372                         lmm = buf->lb_buf + le32_to_cpu(lcme->lcme_offset);
1373                         magic = le32_to_cpu(lmm->lmm_magic);
1374                 }
1375
1376                 if (magic == LOV_MAGIC_V1)
1377                         objs = &(lmm->lmm_objects[0]);
1378                 else
1379                         objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
1380                 ostid_le_to_cpu(&objs->l_ost_oi, oi);
1381                 ostid_to_fid(fid, oi, le32_to_cpu(objs->l_ost_idx));
1382                 fid->f_oid--;
1383                 fid_to_ostid(fid, oi);
1384                 ostid_cpu_to_le(oi, &objs->l_ost_oi);
1385
1386                 rc = lod_sub_xattr_set(env, next, buf, XATTR_NAME_LOV,
1387                                        LU_XATTR_REPLACE, th);
1388         } else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PFL_RANGE)) {
1389                 struct lod_thread_info *info = lod_env_info(env);
1390                 struct lu_buf *buf = &info->lti_buf;
1391                 struct lov_comp_md_v1 *lcm;
1392                 struct lov_comp_md_entry_v1 *lcme;
1393
1394                 rc = lod_get_lov_ea(env, lo);
1395                 if (rc <= 0)
1396                         RETURN(rc);
1397
1398                 buf->lb_buf = info->lti_ea_store;
1399                 buf->lb_len = info->lti_ea_store_size;
1400                 lcm = buf->lb_buf;
1401                 if (le32_to_cpu(lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
1402                         RETURN(-EINVAL);
1403
1404                 le32_add_cpu(&lcm->lcm_layout_gen, 1);
1405                 lcme = &lcm->lcm_entries[0];
1406                 le64_add_cpu(&lcme->lcme_extent.e_start, 1);
1407                 le64_add_cpu(&lcme->lcme_extent.e_end, -1);
1408
1409                 rc = lod_sub_xattr_set(env, next, buf, XATTR_NAME_LOV,
1410                                        LU_XATTR_REPLACE, th);
1411         }
1412
1413         RETURN(rc);
1414 }
1415
1416 /**
1417  * Implementation of dt_object_operations::do_xattr_get.
1418  *
1419  * If LOV EA is requested from the root object and it's not
1420  * found, then return default striping for the filesystem.
1421  *
1422  * \see dt_object_operations::do_xattr_get() in the API description for details.
1423  */
1424 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
1425                          struct lu_buf *buf, const char *name)
1426 {
1427         struct lod_thread_info *info = lod_env_info(env);
1428         struct lod_device *dev = lu2lod_dev(dt->do_lu.lo_dev);
1429         int is_root;
1430         int rc;
1431         ENTRY;
1432
1433         rc = dt_xattr_get(env, dt_object_child(dt), buf, name);
1434         if (strcmp(name, XATTR_NAME_LMV) == 0) {
1435                 struct lmv_mds_md_v1    *lmv1;
1436                 int                      rc1 = 0;
1437
1438                 if (rc > (typeof(rc))sizeof(*lmv1))
1439                         RETURN(rc);
1440
1441                 if (rc < (typeof(rc))sizeof(*lmv1))
1442                         RETURN(rc = rc > 0 ? -EINVAL : rc);
1443
1444                 if (buf->lb_buf == NULL || buf->lb_len == 0) {
1445                         CLASSERT(sizeof(*lmv1) <= sizeof(info->lti_key));
1446
1447                         info->lti_buf.lb_buf = info->lti_key;
1448                         info->lti_buf.lb_len = sizeof(*lmv1);
1449                         rc = dt_xattr_get(env, dt_object_child(dt),
1450                                           &info->lti_buf, name);
1451                         if (unlikely(rc != sizeof(*lmv1)))
1452                                 RETURN(rc = rc > 0 ? -EINVAL : rc);
1453
1454                         lmv1 = info->lti_buf.lb_buf;
1455                         /* The on-disk LMV EA only contains header, but the
1456                          * returned LMV EA size should contain the space for
1457                          * the FIDs of all shards of the striped directory. */
1458                         if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_V1)
1459                                 rc = lmv_mds_md_size(
1460                                         le32_to_cpu(lmv1->lmv_stripe_count),
1461                                         LMV_MAGIC_V1);
1462                 } else {
1463                         rc1 = lod_load_lmv_shards(env, lod_dt_obj(dt),
1464                                                   buf, false);
1465                 }
1466
1467                 RETURN(rc = rc1 != 0 ? rc1 : rc);
1468         }
1469
1470         if (rc != -ENODATA || !S_ISDIR(dt->do_lu.lo_header->loh_attr & S_IFMT))
1471                 RETURN(rc);
1472
1473         /*
1474          * XXX: Only used by lfsck
1475          *
1476          * lod returns default striping on the real root of the device
1477          * this is like the root stores default striping for the whole
1478          * filesystem. historically we've been using a different approach
1479          * and store it in the config.
1480          */
1481         dt_root_get(env, dev->lod_child, &info->lti_fid);
1482         is_root = lu_fid_eq(&info->lti_fid, lu_object_fid(&dt->do_lu));
1483
1484         if (is_root && strcmp(XATTR_NAME_LOV, name) == 0) {
1485                 struct lov_user_md *lum = buf->lb_buf;
1486                 struct lov_desc    *desc = &dev->lod_desc;
1487
1488                 if (buf->lb_buf == NULL) {
1489                         rc = sizeof(*lum);
1490                 } else if (buf->lb_len >= sizeof(*lum)) {
1491                         lum->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V1);
1492                         lmm_oi_set_seq(&lum->lmm_oi, FID_SEQ_LOV_DEFAULT);
1493                         lmm_oi_set_id(&lum->lmm_oi, 0);
1494                         lmm_oi_cpu_to_le(&lum->lmm_oi, &lum->lmm_oi);
1495                         lum->lmm_pattern = cpu_to_le32(desc->ld_pattern);
1496                         lum->lmm_stripe_size = cpu_to_le32(
1497                                                 desc->ld_default_stripe_size);
1498                         lum->lmm_stripe_count = cpu_to_le16(
1499                                                 desc->ld_default_stripe_count);
1500                         lum->lmm_stripe_offset = cpu_to_le16(
1501                                                 desc->ld_default_stripe_offset);
1502                         rc = sizeof(*lum);
1503                 } else {
1504                         rc = -ERANGE;
1505                 }
1506         }
1507
1508         RETURN(rc);
1509 }
1510
1511 /**
1512  * Verify LVM EA.
1513  *
1514  * Checks that the magic of the stripe is sane.
1515  *
1516  * \param[in] lod       lod device
1517  * \param[in] lum       a buffer storing LMV EA to verify
1518  *
1519  * \retval              0 if the EA is sane
1520  * \retval              negative otherwise
1521  */
1522 static int lod_verify_md_striping(struct lod_device *lod,
1523                                   const struct lmv_user_md_v1 *lum)
1524 {
1525         if (unlikely(le32_to_cpu(lum->lum_magic) != LMV_USER_MAGIC)) {
1526                 CERROR("%s: invalid lmv_user_md: magic = %x, "
1527                        "stripe_offset = %d, stripe_count = %u: rc = %d\n",
1528                        lod2obd(lod)->obd_name, le32_to_cpu(lum->lum_magic),
1529                        (int)le32_to_cpu(lum->lum_stripe_offset),
1530                        le32_to_cpu(lum->lum_stripe_count), -EINVAL);
1531                 return -EINVAL;
1532         }
1533
1534         return 0;
1535 }
1536
1537 /**
1538  * Initialize LMV EA for a slave.
1539  *
1540  * Initialize slave's LMV EA from the master's LMV EA.
1541  *
1542  * \param[in] master_lmv        a buffer containing master's EA
1543  * \param[out] slave_lmv        a buffer where slave's EA will be stored
1544  *
1545  */
1546 static void lod_prep_slave_lmv_md(struct lmv_mds_md_v1 *slave_lmv,
1547                                   const struct lmv_mds_md_v1 *master_lmv)
1548 {
1549         *slave_lmv = *master_lmv;
1550         slave_lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_STRIPE);
1551 }
1552
1553 /**
1554  * Generate LMV EA.
1555  *
1556  * Generate LMV EA from the object passed as \a dt. The object must have
1557  * the stripes created and initialized.
1558  *
1559  * \param[in] env       execution environment
1560  * \param[in] dt        object
1561  * \param[out] lmv_buf  buffer storing generated LMV EA
1562  *
1563  * \retval              0 on success
1564  * \retval              negative if failed
1565  */
1566 static int lod_prep_lmv_md(const struct lu_env *env, struct dt_object *dt,
1567                            struct lu_buf *lmv_buf)
1568 {
1569         struct lod_thread_info  *info = lod_env_info(env);
1570         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1571         struct lod_object       *lo = lod_dt_obj(dt);
1572         struct lmv_mds_md_v1    *lmm1;
1573         int                     stripe_count;
1574         int                     type = LU_SEQ_RANGE_ANY;
1575         int                     rc;
1576         __u32                   mdtidx;
1577         ENTRY;
1578
1579         LASSERT(lo->ldo_dir_striped != 0);
1580         LASSERT(lo->ldo_dir_stripe_count > 0);
1581         stripe_count = lo->ldo_dir_stripe_count;
1582         /* Only store the LMV EA heahder on the disk. */
1583         if (info->lti_ea_store_size < sizeof(*lmm1)) {
1584                 rc = lod_ea_store_resize(info, sizeof(*lmm1));
1585                 if (rc != 0)
1586                         RETURN(rc);
1587         } else {
1588                 memset(info->lti_ea_store, 0, sizeof(*lmm1));
1589         }
1590
1591         lmm1 = (struct lmv_mds_md_v1 *)info->lti_ea_store;
1592         lmm1->lmv_magic = cpu_to_le32(LMV_MAGIC);
1593         lmm1->lmv_stripe_count = cpu_to_le32(stripe_count);
1594         lmm1->lmv_hash_type = cpu_to_le32(lo->ldo_dir_hash_type);
1595         rc = lod_fld_lookup(env, lod, lu_object_fid(&dt->do_lu),
1596                             &mdtidx, &type);
1597         if (rc != 0)
1598                 RETURN(rc);
1599
1600         lmm1->lmv_master_mdt_index = cpu_to_le32(mdtidx);
1601         lmv_buf->lb_buf = info->lti_ea_store;
1602         lmv_buf->lb_len = sizeof(*lmm1);
1603
1604         RETURN(rc);
1605 }
1606
1607 /**
1608  * Create in-core represenation for a striped directory.
1609  *
1610  * Parse the buffer containing LMV EA and instantiate LU objects
1611  * representing the stripe objects. The pointers to the objects are
1612  * stored in ldo_stripe field of \a lo. This function is used when
1613  * we need to access an already created object (i.e. load from a disk).
1614  *
1615  * \param[in] env       execution environment
1616  * \param[in] lo        lod object
1617  * \param[in] buf       buffer containing LMV EA
1618  *
1619  * \retval              0 on success
1620  * \retval              negative if failed
1621  */
1622 int lod_parse_dir_striping(const struct lu_env *env, struct lod_object *lo,
1623                            const struct lu_buf *buf)
1624 {
1625         struct lod_thread_info  *info = lod_env_info(env);
1626         struct lod_device       *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1627         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
1628         struct dt_object        **stripe;
1629         union lmv_mds_md        *lmm = buf->lb_buf;
1630         struct lmv_mds_md_v1    *lmv1 = &lmm->lmv_md_v1;
1631         struct lu_fid           *fid = &info->lti_fid;
1632         unsigned int            i;
1633         int                     rc = 0;
1634         ENTRY;
1635
1636         if (le32_to_cpu(lmv1->lmv_hash_type) & LMV_HASH_FLAG_MIGRATION)
1637                 RETURN(0);
1638
1639         if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_STRIPE) {
1640                 lo->ldo_dir_slave_stripe = 1;
1641                 RETURN(0);
1642         }
1643
1644         if (le32_to_cpu(lmv1->lmv_magic) != LMV_MAGIC_V1)
1645                 RETURN(-EINVAL);
1646
1647         if (le32_to_cpu(lmv1->lmv_stripe_count) < 1)
1648                 RETURN(0);
1649
1650         LASSERT(lo->ldo_stripe == NULL);
1651         OBD_ALLOC(stripe, sizeof(stripe[0]) *
1652                   (le32_to_cpu(lmv1->lmv_stripe_count)));
1653         if (stripe == NULL)
1654                 RETURN(-ENOMEM);
1655
1656         for (i = 0; i < le32_to_cpu(lmv1->lmv_stripe_count); i++) {
1657                 struct dt_device        *tgt_dt;
1658                 struct dt_object        *dto;
1659                 int                     type = LU_SEQ_RANGE_ANY;
1660                 __u32                   idx;
1661
1662                 fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[i]);
1663                 if (!fid_is_sane(fid))
1664                         GOTO(out, rc = -ESTALE);
1665
1666                 rc = lod_fld_lookup(env, lod, fid, &idx, &type);
1667                 if (rc != 0)
1668                         GOTO(out, rc);
1669
1670                 if (idx == lod2lu_dev(lod)->ld_site->ld_seq_site->ss_node_id) {
1671                         tgt_dt = lod->lod_child;
1672                 } else {
1673                         struct lod_tgt_desc     *tgt;
1674
1675                         tgt = LTD_TGT(ltd, idx);
1676                         if (tgt == NULL)
1677                                 GOTO(out, rc = -ESTALE);
1678                         tgt_dt = tgt->ltd_tgt;
1679                 }
1680
1681                 dto = dt_locate_at(env, tgt_dt, fid,
1682                                   lo->ldo_obj.do_lu.lo_dev->ld_site->ls_top_dev,
1683                                   NULL);
1684                 if (IS_ERR(dto))
1685                         GOTO(out, rc = PTR_ERR(dto));
1686
1687                 stripe[i] = dto;
1688         }
1689 out:
1690         lo->ldo_stripe = stripe;
1691         lo->ldo_dir_stripe_count = le32_to_cpu(lmv1->lmv_stripe_count);
1692         lo->ldo_dir_stripes_allocated = le32_to_cpu(lmv1->lmv_stripe_count);
1693         if (rc != 0)
1694                 lod_object_free_striping(env, lo);
1695
1696         RETURN(rc);
1697 }
1698
1699 /**
1700  * Declare create a striped directory.
1701  *
1702  * Declare creating a striped directory with a given stripe pattern on the
1703  * specified MDTs. A striped directory is represented as a regular directory
1704  * - an index listing all the stripes. The stripes point back to the master
1705  * object with ".." and LinkEA. The master object gets LMV EA which
1706  * identifies it as a striped directory. The function allocates FIDs
1707  * for all stripes.
1708  *
1709  * \param[in] env       execution environment
1710  * \param[in] dt        object
1711  * \param[in] attr      attributes to initialize the objects with
1712  * \param[in] dof       type of objects to be created
1713  * \param[in] th        transaction handle
1714  *
1715  * \retval              0 on success
1716  * \retval              negative if failed
1717  */
1718 static int lod_dir_declare_create_stripes(const struct lu_env *env,
1719                                           struct dt_object *dt,
1720                                           struct lu_attr *attr,
1721                                           struct dt_object_format *dof,
1722                                           struct thandle *th)
1723 {
1724         struct lod_thread_info  *info = lod_env_info(env);
1725         struct lu_buf           lmv_buf;
1726         struct lu_buf           slave_lmv_buf;
1727         struct lmv_mds_md_v1    *lmm;
1728         struct lmv_mds_md_v1    *slave_lmm = NULL;
1729         struct dt_insert_rec    *rec = &info->lti_dt_rec;
1730         struct lod_object       *lo = lod_dt_obj(dt);
1731         int                     rc;
1732         __u32                   i;
1733         ENTRY;
1734
1735         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
1736         if (rc != 0)
1737                 GOTO(out, rc);
1738         lmm = lmv_buf.lb_buf;
1739
1740         OBD_ALLOC_PTR(slave_lmm);
1741         if (slave_lmm == NULL)
1742                 GOTO(out, rc = -ENOMEM);
1743
1744         lod_prep_slave_lmv_md(slave_lmm, lmm);
1745         slave_lmv_buf.lb_buf = slave_lmm;
1746         slave_lmv_buf.lb_len = sizeof(*slave_lmm);
1747
1748         if (!dt_try_as_dir(env, dt_object_child(dt)))
1749                 GOTO(out, rc = -EINVAL);
1750
1751         rec->rec_type = S_IFDIR;
1752         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
1753                 struct dt_object        *dto = lo->ldo_stripe[i];
1754                 char                    *stripe_name = info->lti_key;
1755                 struct lu_name          *sname;
1756                 struct linkea_data       ldata          = { NULL };
1757                 struct lu_buf           linkea_buf;
1758
1759                 rc = lod_sub_declare_create(env, dto, attr, NULL, dof, th);
1760                 if (rc != 0)
1761                         GOTO(out, rc);
1762
1763                 if (!dt_try_as_dir(env, dto))
1764                         GOTO(out, rc = -EINVAL);
1765
1766                 rc = lod_sub_declare_ref_add(env, dto, th);
1767                 if (rc != 0)
1768                         GOTO(out, rc);
1769
1770                 rec->rec_fid = lu_object_fid(&dto->do_lu);
1771                 rc = lod_sub_declare_insert(env, dto,
1772                                             (const struct dt_rec *)rec,
1773                                             (const struct dt_key *)dot, th);
1774                 if (rc != 0)
1775                         GOTO(out, rc);
1776
1777                 /* master stripe FID will be put to .. */
1778                 rec->rec_fid = lu_object_fid(&dt->do_lu);
1779                 rc = lod_sub_declare_insert(env, dto,
1780                                             (const struct dt_rec *)rec,
1781                                             (const struct dt_key *)dotdot, th);
1782                 if (rc != 0)
1783                         GOTO(out, rc);
1784
1785                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
1786                     cfs_fail_val != i) {
1787                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
1788                             cfs_fail_val == i)
1789                                 slave_lmm->lmv_master_mdt_index =
1790                                                         cpu_to_le32(i + 1);
1791                         else
1792                                 slave_lmm->lmv_master_mdt_index =
1793                                                         cpu_to_le32(i);
1794                         rc = lod_sub_declare_xattr_set(env, dto, &slave_lmv_buf,
1795                                                        XATTR_NAME_LMV, 0, th);
1796                         if (rc != 0)
1797                                 GOTO(out, rc);
1798                 }
1799
1800                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
1801                     cfs_fail_val == i)
1802                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1803                                 PFID(lu_object_fid(&dto->do_lu)), i + 1);
1804                 else
1805                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1806                                 PFID(lu_object_fid(&dto->do_lu)), i);
1807
1808                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
1809                 rc = linkea_links_new(&ldata, &info->lti_linkea_buf,
1810                                       sname, lu_object_fid(&dt->do_lu));
1811                 if (rc != 0)
1812                         GOTO(out, rc);
1813
1814                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
1815                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
1816                 rc = lod_sub_declare_xattr_set(env, dto, &linkea_buf,
1817                                                XATTR_NAME_LINK, 0, th);
1818                 if (rc != 0)
1819                         GOTO(out, rc);
1820
1821                 rec->rec_fid = lu_object_fid(&dto->do_lu);
1822                 rc = lod_sub_declare_insert(env, dt_object_child(dt),
1823                                             (const struct dt_rec *)rec,
1824                                             (const struct dt_key *)stripe_name,
1825                                             th);
1826                 if (rc != 0)
1827                         GOTO(out, rc);
1828
1829                 rc = lod_sub_declare_ref_add(env, dt_object_child(dt), th);
1830                 if (rc != 0)
1831                         GOTO(out, rc);
1832         }
1833
1834         rc = lod_sub_declare_xattr_set(env, dt_object_child(dt),
1835                                        &lmv_buf, XATTR_NAME_LMV, 0, th);
1836         if (rc != 0)
1837                 GOTO(out, rc);
1838 out:
1839         if (slave_lmm != NULL)
1840                 OBD_FREE_PTR(slave_lmm);
1841
1842         RETURN(rc);
1843 }
1844
1845 static int lod_prep_md_striped_create(const struct lu_env *env,
1846                                       struct dt_object *dt,
1847                                       struct lu_attr *attr,
1848                                       const struct lmv_user_md_v1 *lum,
1849                                       struct dt_object_format *dof,
1850                                       struct thandle *th)
1851 {
1852         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1853         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
1854         struct lod_object       *lo = lod_dt_obj(dt);
1855         struct dt_object        **stripe;
1856         __u32                   stripe_count;
1857         int                     *idx_array;
1858         __u32                   master_index;
1859         int                     rc = 0;
1860         __u32                   i;
1861         __u32                   j;
1862         ENTRY;
1863
1864         /* The lum has been verifed in lod_verify_md_striping */
1865         LASSERT(le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC);
1866         LASSERT(le32_to_cpu(lum->lum_stripe_count) > 0);
1867
1868         stripe_count = le32_to_cpu(lum->lum_stripe_count);
1869
1870         OBD_ALLOC(idx_array, sizeof(idx_array[0]) * stripe_count);
1871         if (idx_array == NULL)
1872                 RETURN(-ENOMEM);
1873
1874         OBD_ALLOC(stripe, sizeof(stripe[0]) * stripe_count);
1875         if (stripe == NULL)
1876                 GOTO(out_free, rc = -ENOMEM);
1877
1878         /* Start index must be the master MDT */
1879         master_index = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id;
1880         idx_array[0] = master_index;
1881         for (i = 0; i < stripe_count; i++) {
1882                 struct lod_tgt_desc     *tgt = NULL;
1883                 struct dt_object        *dto;
1884                 struct lu_fid           fid = { 0 };
1885                 int                     idx;
1886                 struct lu_object_conf   conf = { 0 };
1887                 struct dt_device        *tgt_dt = NULL;
1888
1889                 /* Try to find next avaible target */
1890                 idx = idx_array[i];
1891                 for (j = 0; j < lod->lod_remote_mdt_count;
1892                      j++, idx = (idx + 1) % (lod->lod_remote_mdt_count + 1)) {
1893                         bool already_allocated = false;
1894                         __u32 k;
1895
1896                         CDEBUG(D_INFO, "try idx %d, mdt cnt %u, allocated %u\n",
1897                                idx, lod->lod_remote_mdt_count + 1, i);
1898
1899                         if (likely(!OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE))) {
1900                                 /* check whether the idx already exists
1901                                  * in current allocated array */
1902                                 for (k = 0; k < i; k++) {
1903                                         if (idx_array[k] == idx) {
1904                                                 already_allocated = true;
1905                                                 break;
1906                                         }
1907                                 }
1908
1909                                 if (already_allocated)
1910                                         continue;
1911                         }
1912
1913                         /* Sigh, this index is not in the bitmap, let's check
1914                          * next available target */
1915                         if (!cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx) &&
1916                             idx != master_index)
1917                                 continue;
1918
1919                         if (idx == master_index) {
1920                                 /* Allocate the FID locally */
1921                                 rc = obd_fid_alloc(env, lod->lod_child_exp,
1922                                                    &fid, NULL);
1923                                 if (rc < 0)
1924                                         GOTO(out_put, rc);
1925                                 tgt_dt = lod->lod_child;
1926                                 break;
1927                         }
1928
1929                         /* check the status of the OSP */
1930                         tgt = LTD_TGT(ltd, idx);
1931                         if (tgt == NULL)
1932                                 continue;
1933
1934                         tgt_dt = tgt->ltd_tgt;
1935                         rc = dt_statfs(env, tgt_dt, NULL);
1936                         if (rc) {
1937                                 /* this OSP doesn't feel well */
1938                                 rc = 0;
1939                                 continue;
1940                         }
1941
1942                         rc = obd_fid_alloc(env, tgt->ltd_exp, &fid, NULL);
1943                         if (rc < 0) {
1944                                 rc = 0;
1945                                 continue;
1946                         }
1947
1948                         break;
1949                 }
1950
1951                 /* Can not allocate more stripes */
1952                 if (j == lod->lod_remote_mdt_count) {
1953                         CDEBUG(D_INFO, "%s: require stripes %u only get %d\n",
1954                                lod2obd(lod)->obd_name, stripe_count, i);
1955                         break;
1956                 }
1957
1958                 CDEBUG(D_INFO, "Get idx %d, for stripe %d "DFID"\n",
1959                        idx, i, PFID(&fid));
1960                 idx_array[i] = idx;
1961                 /* Set the start index for next stripe allocation */
1962                 if (i < stripe_count - 1)
1963                         idx_array[i + 1] = (idx + 1) %
1964                                            (lod->lod_remote_mdt_count + 1);
1965                 /* tgt_dt and fid must be ready after search avaible OSP
1966                  * in the above loop */
1967                 LASSERT(tgt_dt != NULL);
1968                 LASSERT(fid_is_sane(&fid));
1969                 conf.loc_flags = LOC_F_NEW;
1970                 dto = dt_locate_at(env, tgt_dt, &fid,
1971                                    dt->do_lu.lo_dev->ld_site->ls_top_dev,
1972                                    &conf);
1973                 if (IS_ERR(dto))
1974                         GOTO(out_put, rc = PTR_ERR(dto));
1975                 stripe[i] = dto;
1976         }
1977
1978         lo->ldo_dir_stripe_loaded = 1;
1979         lo->ldo_dir_striped = 1;
1980         lo->ldo_stripe = stripe;
1981         lo->ldo_dir_stripe_count = i;
1982         lo->ldo_dir_stripes_allocated = stripe_count;
1983
1984         if (lo->ldo_dir_stripe_count == 0)
1985                 GOTO(out_put, rc = -ENOSPC);
1986
1987         rc = lod_dir_declare_create_stripes(env, dt, attr, dof, th);
1988         if (rc != 0)
1989                 GOTO(out_put, rc);
1990
1991 out_put:
1992         if (rc < 0) {
1993                 for (i = 0; i < stripe_count; i++)
1994                         if (stripe[i] != NULL)
1995                                 dt_object_put(env, stripe[i]);
1996                 OBD_FREE(stripe, sizeof(stripe[0]) * stripe_count);
1997                 lo->ldo_dir_stripe_count = 0;
1998                 lo->ldo_dir_stripes_allocated = 0;
1999                 lo->ldo_stripe = NULL;
2000         }
2001
2002 out_free:
2003         OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count);
2004
2005         RETURN(rc);
2006 }
2007
2008 /**
2009  * Declare create striped md object.
2010  *
2011  * The function declares intention to create a striped directory. This is a
2012  * wrapper for lod_prep_md_striped_create(). The only additional functionality
2013  * is to verify pattern \a lum_buf is good. Check that function for the details.
2014  *
2015  * \param[in] env       execution environment
2016  * \param[in] dt        object
2017  * \param[in] attr      attributes to initialize the objects with
2018  * \param[in] lum_buf   a pattern specifying the number of stripes and
2019  *                      MDT to start from
2020  * \param[in] dof       type of objects to be created
2021  * \param[in] th        transaction handle
2022  *
2023  * \retval              0 on success
2024  * \retval              negative if failed
2025  *
2026  */
2027 static int lod_declare_xattr_set_lmv(const struct lu_env *env,
2028                                      struct dt_object *dt,
2029                                      struct lu_attr *attr,
2030                                      const struct lu_buf *lum_buf,
2031                                      struct dt_object_format *dof,
2032                                      struct thandle *th)
2033 {
2034         struct lod_object       *lo = lod_dt_obj(dt);
2035         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
2036         struct lmv_user_md_v1   *lum;
2037         int                     rc;
2038         ENTRY;
2039
2040         lum = lum_buf->lb_buf;
2041         LASSERT(lum != NULL);
2042
2043         CDEBUG(D_INFO, "lum magic = %x count = %u offset = %d\n",
2044                le32_to_cpu(lum->lum_magic), le32_to_cpu(lum->lum_stripe_count),
2045                (int)le32_to_cpu(lum->lum_stripe_offset));
2046
2047         if (le32_to_cpu(lum->lum_stripe_count) == 0)
2048                 GOTO(out, rc = 0);
2049
2050         rc = lod_verify_md_striping(lod, lum);
2051         if (rc != 0)
2052                 GOTO(out, rc);
2053
2054         /* prepare dir striped objects */
2055         rc = lod_prep_md_striped_create(env, dt, attr, lum, dof, th);
2056         if (rc != 0) {
2057                 /* failed to create striping, let's reset
2058                  * config so that others don't get confused */
2059                 lod_object_free_striping(env, lo);
2060                 GOTO(out, rc);
2061         }
2062 out:
2063         RETURN(rc);
2064 }
2065
2066 /**
2067  * Implementation of dt_object_operations::do_declare_xattr_set.
2068  *
2069  * Used with regular (non-striped) objects. Basically it
2070  * initializes the striping information and applies the
2071  * change to all the stripes.
2072  *
2073  * \see dt_object_operations::do_declare_xattr_set() in the API description
2074  * for details.
2075  */
2076 static int lod_dir_declare_xattr_set(const struct lu_env *env,
2077                                      struct dt_object *dt,
2078                                      const struct lu_buf *buf,
2079                                      const char *name, int fl,
2080                                      struct thandle *th)
2081 {
2082         struct dt_object        *next = dt_object_child(dt);
2083         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2084         struct lod_object       *lo = lod_dt_obj(dt);
2085         int                     i;
2086         int                     rc;
2087         ENTRY;
2088
2089         if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
2090                 struct lmv_user_md_v1 *lum;
2091
2092                 LASSERT(buf != NULL && buf->lb_buf != NULL);
2093                 lum = buf->lb_buf;
2094                 rc = lod_verify_md_striping(d, lum);
2095                 if (rc != 0)
2096                         RETURN(rc);
2097         } else if (strcmp(name, XATTR_NAME_LOV) == 0) {
2098                 rc = lod_verify_striping(d, lo, buf, false);
2099                 if (rc != 0)
2100                         RETURN(rc);
2101         }
2102
2103         rc = lod_sub_declare_xattr_set(env, next, buf, name, fl, th);
2104         if (rc != 0)
2105                 RETURN(rc);
2106
2107         /* Note: Do not set LinkEA on sub-stripes, otherwise
2108          * it will confuse the fid2path process(see mdt_path_current()).
2109          * The linkEA between master and sub-stripes is set in
2110          * lod_xattr_set_lmv(). */
2111         if (strcmp(name, XATTR_NAME_LINK) == 0)
2112                 RETURN(0);
2113
2114         /* set xattr to each stripes, if needed */
2115         rc = lod_load_striping(env, lo);
2116         if (rc != 0)
2117                 RETURN(rc);
2118
2119         if (lo->ldo_dir_stripe_count == 0)
2120                 RETURN(0);
2121
2122         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
2123                 LASSERT(lo->ldo_stripe[i]);
2124
2125                 rc = lod_sub_declare_xattr_set(env, lo->ldo_stripe[i],
2126                                                buf, name, fl, th);
2127                 if (rc != 0)
2128                         break;
2129         }
2130
2131         RETURN(rc);
2132 }
2133
2134 static int
2135 lod_obj_stripe_replace_parent_fid_cb(const struct lu_env *env,
2136                                      struct lod_object *lo,
2137                                      struct dt_object *dt, struct thandle *th,
2138                                      int comp_idx, int stripe_idx,
2139                                      struct lod_obj_stripe_cb_data *data)
2140 {
2141         struct lod_thread_info *info = lod_env_info(env);
2142         struct filter_fid *ff = &info->lti_ff;
2143         struct lu_buf *buf = &info->lti_buf;
2144         int rc;
2145
2146         buf->lb_buf = ff;
2147         buf->lb_len = sizeof(*ff);
2148         rc = dt_xattr_get(env, dt, buf, XATTR_NAME_FID);
2149         if (rc == -ENODATA)
2150                 return 0;
2151
2152         if (rc < 0)
2153                 return rc;
2154
2155         ff->ff_parent = *lu_object_fid(&lo->ldo_obj.do_lu);
2156         ff->ff_parent.f_ver = stripe_idx;
2157         fid_cpu_to_le(&ff->ff_parent, &ff->ff_parent);
2158         if (data->locd_declare)
2159                 rc = lod_sub_declare_xattr_set(env, dt, buf, XATTR_NAME_FID,
2160                                                LU_XATTR_REPLACE, th);
2161         else
2162                 rc = lod_sub_xattr_set(env, dt, buf, XATTR_NAME_FID,
2163                                        LU_XATTR_REPLACE, th);
2164
2165         return rc;
2166 }
2167
2168 /**
2169  * Reset parent FID on OST object
2170  *
2171  * Replace parent FID with @dt object FID, which is only called during migration
2172  * to reset the parent FID after the MDT object is migrated to the new MDT, i.e.
2173  * the FID is changed.
2174  *
2175  * \param[in] env execution environment
2176  * \param[in] dt dt_object whose stripes's parent FID will be reset
2177  * \parem[in] th thandle
2178  * \param[in] declare if it is declare
2179  *
2180  * \retval      0 if reset succeeds
2181  * \retval      negative errno if reset fails
2182  */
2183 static int lod_replace_parent_fid(const struct lu_env *env,
2184                                   struct dt_object *dt,
2185                                   struct thandle *th, bool declare)
2186 {
2187         struct lod_object *lo = lod_dt_obj(dt);
2188         struct lod_thread_info  *info = lod_env_info(env);
2189         struct lu_buf *buf = &info->lti_buf;
2190         struct filter_fid *ff;
2191         struct lod_obj_stripe_cb_data data = { { 0 } };
2192         int rc;
2193         ENTRY;
2194
2195         LASSERT(S_ISREG(dt->do_lu.lo_header->loh_attr));
2196
2197         /* set xattr to each stripes, if needed */
2198         rc = lod_load_striping(env, lo);
2199         if (rc != 0)
2200                 RETURN(rc);
2201
2202         if (!lod_obj_is_striped(dt))
2203                 RETURN(0);
2204
2205         if (info->lti_ea_store_size < sizeof(*ff)) {
2206                 rc = lod_ea_store_resize(info, sizeof(*ff));
2207                 if (rc != 0)
2208                         RETURN(rc);
2209         }
2210
2211         buf->lb_buf = info->lti_ea_store;
2212         buf->lb_len = info->lti_ea_store_size;
2213
2214         data.locd_declare = declare;
2215         data.locd_stripe_cb = lod_obj_stripe_replace_parent_fid_cb;
2216         rc = lod_obj_for_each_stripe(env, lo, th, &data);
2217
2218         RETURN(rc);
2219 }
2220
2221 inline __u16 lod_comp_entry_stripe_count(struct lod_object *lo,
2222                                          struct lod_layout_component *entry,
2223                                          bool is_dir)
2224 {
2225         struct lod_device *lod = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
2226
2227         if (is_dir)
2228                 return  0;
2229         else if (lod_comp_inited(entry))
2230                 return entry->llc_stripe_count;
2231         else if ((__u16)-1 == entry->llc_stripe_count)
2232                 return lod->lod_desc.ld_tgt_count;
2233         else
2234                 return lod_get_stripe_count(lod, lo, entry->llc_stripe_count);
2235 }
2236
2237 static int lod_comp_md_size(struct lod_object *lo, bool is_dir)
2238 {
2239         int magic, size = 0, i;
2240         struct lod_layout_component *comp_entries;
2241         __u16 comp_cnt;
2242         bool is_composite;
2243
2244         if (is_dir) {
2245                 comp_cnt = lo->ldo_def_striping->lds_def_comp_cnt;
2246                 comp_entries = lo->ldo_def_striping->lds_def_comp_entries;
2247                 is_composite =
2248                         lo->ldo_def_striping->lds_def_striping_is_composite;
2249         } else {
2250                 comp_cnt = lo->ldo_comp_cnt;
2251                 comp_entries = lo->ldo_comp_entries;
2252                 is_composite = lo->ldo_is_composite;
2253         }
2254
2255
2256         LASSERT(comp_cnt != 0 && comp_entries != NULL);
2257         if (is_composite) {
2258                 size = sizeof(struct lov_comp_md_v1) +
2259                        sizeof(struct lov_comp_md_entry_v1) * comp_cnt;
2260                 LASSERT(size % sizeof(__u64) == 0);
2261         }
2262
2263         for (i = 0; i < comp_cnt; i++) {
2264                 __u16 stripe_count;
2265
2266                 magic = comp_entries[i].llc_pool ? LOV_MAGIC_V3 : LOV_MAGIC_V1;
2267                 stripe_count = lod_comp_entry_stripe_count(lo, &comp_entries[i],
2268                                                            is_dir);
2269                 if (!is_dir && is_composite)
2270                         lod_comp_shrink_stripe_count(&comp_entries[i],
2271                                                      &stripe_count);
2272
2273                 size += lov_user_md_size(stripe_count, magic);
2274                 LASSERT(size % sizeof(__u64) == 0);
2275         }
2276         return size;
2277 }
2278
2279 /**
2280  * Declare component add. The xattr name is XATTR_LUSTRE_LOV.add, and
2281  * the xattr value is binary lov_comp_md_v1 which contains component(s)
2282  * to be added.
2283   *
2284  * \param[in] env       execution environment
2285  * \param[in] dt        dt_object to add components on
2286  * \param[in] buf       buffer contains components to be added
2287  * \parem[in] th        thandle
2288  *
2289  * \retval      0 on success
2290  * \retval      negative errno on failure
2291  */
2292 static int lod_declare_layout_add(const struct lu_env *env,
2293                                   struct dt_object *dt,
2294                                   const struct lu_buf *buf,
2295                                   struct thandle *th)
2296 {
2297         struct lod_thread_info  *info = lod_env_info(env);
2298         struct lod_layout_component *comp_array, *lod_comp, *old_array;
2299         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2300         struct dt_object *next = dt_object_child(dt);
2301         struct lov_desc         *desc = &d->lod_desc;
2302         struct lod_object       *lo = lod_dt_obj(dt);
2303         struct lov_user_md_v3   *v3;
2304         struct lov_comp_md_v1   *comp_v1 = buf->lb_buf;
2305         __u32   magic;
2306         int     i, rc, array_cnt, old_array_cnt;
2307         ENTRY;
2308
2309         LASSERT(lo->ldo_is_composite);
2310
2311         if (lo->ldo_flr_state != LCM_FL_NOT_FLR)
2312                 RETURN(-EBUSY);
2313
2314         rc = lod_verify_striping(d, lo, buf, false);
2315         if (rc != 0)
2316                 RETURN(rc);
2317
2318         magic = comp_v1->lcm_magic;
2319         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
2320                 lustre_swab_lov_comp_md_v1(comp_v1);
2321                 magic = comp_v1->lcm_magic;
2322         }
2323
2324         if (magic != LOV_USER_MAGIC_COMP_V1)
2325                 RETURN(-EINVAL);
2326
2327         array_cnt = lo->ldo_comp_cnt + comp_v1->lcm_entry_count;
2328         OBD_ALLOC(comp_array, sizeof(*comp_array) * array_cnt);
2329         if (comp_array == NULL)
2330                 RETURN(-ENOMEM);
2331
2332         memcpy(comp_array, lo->ldo_comp_entries,
2333                sizeof(*comp_array) * lo->ldo_comp_cnt);
2334
2335         for (i = 0; i < comp_v1->lcm_entry_count; i++) {
2336                 struct lov_user_md_v1 *v1;
2337                 struct lu_extent *ext;
2338
2339                 v1 = (struct lov_user_md *)((char *)comp_v1 +
2340                                 comp_v1->lcm_entries[i].lcme_offset);
2341                 ext = &comp_v1->lcm_entries[i].lcme_extent;
2342
2343                 lod_comp = &comp_array[lo->ldo_comp_cnt + i];
2344                 lod_comp->llc_extent.e_start = ext->e_start;
2345                 lod_comp->llc_extent.e_end = ext->e_end;
2346                 lod_comp->llc_stripe_offset = v1->lmm_stripe_offset;
2347                 lod_comp->llc_flags = comp_v1->lcm_entries[i].lcme_flags;
2348
2349                 lod_comp->llc_stripe_count = v1->lmm_stripe_count;
2350                 if (!lod_comp->llc_stripe_count ||
2351                     lod_comp->llc_stripe_count == (__u16)-1)
2352                         lod_comp->llc_stripe_count =
2353                                 desc->ld_default_stripe_count;
2354                 lod_comp->llc_stripe_size = v1->lmm_stripe_size;
2355                 if (!lod_comp->llc_stripe_size)
2356                         lod_comp->llc_stripe_size =
2357                                 desc->ld_default_stripe_size;
2358
2359                 if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
2360                         v3 = (struct lov_user_md_v3 *) v1;
2361                         if (v3->lmm_pool_name[0] != '\0') {
2362                                 rc = lod_set_pool(&lod_comp->llc_pool,
2363                                                   v3->lmm_pool_name);
2364                                 if (rc)
2365                                         GOTO(error, rc);
2366                         }
2367                 }
2368         }
2369
2370         old_array = lo->ldo_comp_entries;
2371         old_array_cnt = lo->ldo_comp_cnt;
2372
2373         lo->ldo_comp_entries = comp_array;
2374         lo->ldo_comp_cnt = array_cnt;
2375
2376         /* No need to increase layout generation here, it will be increased
2377          * later when generating component ID for the new components */
2378
2379         info->lti_buf.lb_len = lod_comp_md_size(lo, false);
2380         rc = lod_sub_declare_xattr_set(env, next, &info->lti_buf,
2381                                               XATTR_NAME_LOV, 0, th);
2382         if (rc) {
2383                 lo->ldo_comp_entries = old_array;
2384                 lo->ldo_comp_cnt = old_array_cnt;
2385                 GOTO(error, rc);
2386         }
2387
2388         OBD_FREE(old_array, sizeof(*lod_comp) * old_array_cnt);
2389
2390         LASSERT(lo->ldo_mirror_count == 1);
2391         lo->ldo_mirrors[0].lme_end = array_cnt - 1;
2392
2393         RETURN(0);
2394
2395 error:
2396         for (i = lo->ldo_comp_cnt; i < array_cnt; i++) {
2397                 lod_comp = &comp_array[i];
2398                 if (lod_comp->llc_pool != NULL) {
2399                         OBD_FREE(lod_comp->llc_pool,
2400                                  strlen(lod_comp->llc_pool) + 1);
2401                         lod_comp->llc_pool = NULL;
2402                 }
2403         }
2404         OBD_FREE(comp_array, sizeof(*comp_array) * array_cnt);
2405         RETURN(rc);
2406 }
2407
2408 /**
2409  * Declare component set. The xattr is name XATTR_LUSTRE_LOV.set.$field,
2410  * the '$field' can only be 'flags' now. The xattr value is binary
2411  * lov_comp_md_v1 which contains the component ID(s) and the value of
2412  * the field to be modified.
2413  *
2414  * \param[in] env       execution environment
2415  * \param[in] dt        dt_object to be modified
2416  * \param[in] op        operation string, like "set.flags"
2417  * \param[in] buf       buffer contains components to be set
2418  * \parem[in] th        thandle
2419  *
2420  * \retval      0 on success
2421  * \retval      negative errno on failure
2422  */
2423 static int lod_declare_layout_set(const struct lu_env *env,
2424                                   struct dt_object *dt,
2425                                   char *op, const struct lu_buf *buf,
2426                                   struct thandle *th)
2427 {
2428         struct lod_layout_component     *lod_comp;
2429         struct lod_thread_info  *info = lod_env_info(env);
2430         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2431         struct lod_object       *lo = lod_dt_obj(dt);
2432         struct lov_comp_md_v1   *comp_v1 = buf->lb_buf;
2433         __u32   magic, id;
2434         int     i, j, rc;
2435         bool    changed = false;
2436         ENTRY;
2437
2438         if (strcmp(op, "set.flags") != 0) {
2439                 CDEBUG(D_LAYOUT, "%s: operation (%s) not supported.\n",
2440                        lod2obd(d)->obd_name, op);
2441                 RETURN(-ENOTSUPP);
2442         }
2443
2444         magic = comp_v1->lcm_magic;
2445         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
2446                 lustre_swab_lov_comp_md_v1(comp_v1);
2447                 magic = comp_v1->lcm_magic;
2448         }
2449
2450         if (magic != LOV_USER_MAGIC_COMP_V1)
2451                 RETURN(-EINVAL);
2452
2453         if (comp_v1->lcm_entry_count == 0) {
2454                 CDEBUG(D_LAYOUT, "%s: entry count is zero.\n",
2455                        lod2obd(d)->obd_name);
2456                 RETURN(-EINVAL);
2457         }
2458
2459         for (i = 0; i < comp_v1->lcm_entry_count; i++) {
2460                 id = comp_v1->lcm_entries[i].lcme_id;
2461
2462                 for (j = 0; j < lo->ldo_comp_cnt; j++) {
2463                         lod_comp = &lo->ldo_comp_entries[j];
2464                         if (id == lod_comp->llc_id || id == LCME_ID_ALL) {
2465                                 lod_comp->llc_flags =
2466                                         comp_v1->lcm_entries[i].lcme_flags;
2467                                 changed = true;
2468                         }
2469                 }
2470         }
2471
2472         if (!changed) {
2473                 CDEBUG(D_LAYOUT, "%s: requested component(s) not found.\n",
2474                        lod2obd(d)->obd_name);
2475                 RETURN(-EINVAL);
2476         }
2477
2478         lod_obj_inc_layout_gen(lo);
2479
2480         info->lti_buf.lb_len = lod_comp_md_size(lo, false);
2481         rc = lod_sub_declare_xattr_set(env, dt, &info->lti_buf,
2482                                        XATTR_NAME_LOV, 0, th);
2483         RETURN(rc);
2484 }
2485
2486 /**
2487  * Declare component deletion. The xattr name is XATTR_LUSTRE_LOV.del,
2488  * and the xattr value is a unique component ID or a special lcme_id.
2489  *
2490  * \param[in] env       execution environment
2491  * \param[in] dt        dt_object to be operated on
2492  * \param[in] buf       buffer contains component ID or lcme_id
2493  * \parem[in] th        thandle
2494  *
2495  * \retval      0 on success
2496  * \retval      negative errno on failure
2497  */
2498 static int lod_declare_layout_del(const struct lu_env *env,
2499                                   struct dt_object *dt,
2500                                   const struct lu_buf *buf,
2501                                   struct thandle *th)
2502 {
2503         struct lod_thread_info  *info = lod_env_info(env);
2504         struct dt_object *next = dt_object_child(dt);
2505         struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
2506         struct lod_object *lo = lod_dt_obj(dt);
2507         struct lu_attr *attr = &lod_env_info(env)->lti_attr;
2508         struct lov_comp_md_v1 *comp_v1 = buf->lb_buf;
2509         __u32 magic, id, flags, neg_flags = 0;
2510         int rc, i, j, left;
2511         ENTRY;
2512
2513         LASSERT(lo->ldo_is_composite);
2514
2515         if (lo->ldo_flr_state != LCM_FL_NOT_FLR)
2516                 RETURN(-EBUSY);
2517
2518         magic = comp_v1->lcm_magic;
2519         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
2520                 lustre_swab_lov_comp_md_v1(comp_v1);
2521                 magic = comp_v1->lcm_magic;
2522         }
2523
2524         if (magic != LOV_USER_MAGIC_COMP_V1)
2525                 RETURN(-EINVAL);
2526
2527         id = comp_v1->lcm_entries[0].lcme_id;
2528         flags = comp_v1->lcm_entries[0].lcme_flags;
2529
2530         if (id > LCME_ID_MAX || (flags & ~LCME_KNOWN_FLAGS)) {
2531                 CDEBUG(D_LAYOUT, "%s: invalid component id %#x, flags %#x\n",
2532                        lod2obd(d)->obd_name, id, flags);
2533                 RETURN(-EINVAL);
2534         }
2535
2536         if (id != LCME_ID_INVAL && flags != 0) {
2537                 CDEBUG(D_LAYOUT, "%s: specified both id and flags.\n",
2538                        lod2obd(d)->obd_name);
2539                 RETURN(-EINVAL);
2540         }
2541
2542         if (flags & LCME_FL_NEG) {
2543                 neg_flags = flags & ~LCME_FL_NEG;
2544                 flags = 0;
2545         }
2546
2547         left = lo->ldo_comp_cnt;
2548         if (left <= 0)
2549                 RETURN(-EINVAL);
2550
2551         for (i = (lo->ldo_comp_cnt - 1); i >= 0; i--) {
2552                 struct lod_layout_component *lod_comp;
2553
2554                 lod_comp = &lo->ldo_comp_entries[i];
2555
2556                 if (id != LCME_ID_INVAL && id != lod_comp->llc_id)
2557                         continue;
2558                 else if (flags && !(flags & lod_comp->llc_flags))
2559                         continue;
2560                 else if (neg_flags && (neg_flags & lod_comp->llc_flags))
2561                         continue;
2562
2563                 if (left != (i + 1)) {
2564                         CDEBUG(D_LAYOUT, "%s: this deletion will create "
2565                                "a hole.\n", lod2obd(d)->obd_name);
2566                         RETURN(-EINVAL);
2567                 }
2568                 left--;
2569
2570                 /* Mark the component as deleted */
2571                 lod_comp->llc_id = LCME_ID_INVAL;
2572
2573                 /* Not instantiated component */
2574                 if (lod_comp->llc_stripe == NULL)
2575                         continue;
2576
2577                 LASSERT(lod_comp->llc_stripe_count > 0);
2578                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
2579                         struct dt_object *obj = lod_comp->llc_stripe[j];
2580
2581                         if (obj == NULL)
2582                                 continue;
2583                         rc = lod_sub_declare_destroy(env, obj, th);
2584                         if (rc)
2585                                 RETURN(rc);
2586                 }
2587         }
2588
2589         LASSERTF(left >= 0, "left = %d\n", left);
2590         if (left == lo->ldo_comp_cnt) {
2591                 CDEBUG(D_LAYOUT, "%s: requested component id:%#x not found\n",
2592                        lod2obd(d)->obd_name, id);
2593                 RETURN(-EINVAL);
2594         }
2595
2596         memset(attr, 0, sizeof(*attr));
2597         attr->la_valid = LA_SIZE;
2598         rc = lod_sub_declare_attr_set(env, next, attr, th);
2599         if (rc)
2600                 RETURN(rc);
2601
2602         if (left > 0) {
2603                 info->lti_buf.lb_len = lod_comp_md_size(lo, false);
2604                 rc = lod_sub_declare_xattr_set(env, next, &info->lti_buf,
2605                                                XATTR_NAME_LOV, 0, th);
2606         } else {
2607                 rc = lod_sub_declare_xattr_del(env, next, XATTR_NAME_LOV, th);
2608         }
2609
2610         RETURN(rc);
2611 }
2612
2613 /**
2614  * Declare layout add/set/del operations issued by special xattr names:
2615  *
2616  * XATTR_LUSTRE_LOV.add         add component(s) to existing file
2617  * XATTR_LUSTRE_LOV.del         delete component(s) from existing file
2618  * XATTR_LUSTRE_LOV.set.$field  set specified field of certain component(s)
2619  *
2620  * \param[in] env       execution environment
2621  * \param[in] dt        object
2622  * \param[in] name      name of xattr
2623  * \param[in] buf       lu_buf contains xattr value
2624  * \param[in] th        transaction handle
2625  *
2626  * \retval              0 on success
2627  * \retval              negative if failed
2628  */
2629 static int lod_declare_modify_layout(const struct lu_env *env,
2630                                      struct dt_object *dt,
2631                                      const char *name,
2632                                      const struct lu_buf *buf,
2633                                      struct thandle *th)
2634 {
2635         struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
2636         struct lod_object *lo = lod_dt_obj(dt);
2637         struct dt_object *next = dt_object_child(&lo->ldo_obj);
2638         char *op;
2639         int rc, len = strlen(XATTR_LUSTRE_LOV);
2640         ENTRY;
2641
2642         LASSERT(dt_object_exists(dt));
2643
2644         if (strlen(name) <= len || name[len] != '.') {
2645                 CDEBUG(D_LAYOUT, "%s: invalid xattr name: %s\n",
2646                        lod2obd(d)->obd_name, name);
2647                 RETURN(-EINVAL);
2648         }
2649         len++;
2650
2651         dt_write_lock(env, next, 0);
2652         rc = lod_load_striping_locked(env, lo);
2653         if (rc)
2654                 GOTO(unlock, rc);
2655
2656         /* the layout to be modified must be a composite layout */
2657         if (!lo->ldo_is_composite) {
2658                 CDEBUG(D_LAYOUT, "%s: object "DFID" isn't a composite file.\n",
2659                        lod2obd(d)->obd_name, PFID(lu_object_fid(&dt->do_lu)));
2660                 GOTO(unlock, rc = -EINVAL);
2661         }
2662
2663         op = (char *)name + len;
2664         if (strcmp(op, "add") == 0) {
2665                 rc = lod_declare_layout_add(env, dt, buf, th);
2666         } else if (strcmp(op, "del") == 0) {
2667                 rc = lod_declare_layout_del(env, dt, buf, th);
2668         } else if (strncmp(op, "set", strlen("set")) == 0) {
2669                 rc = lod_declare_layout_set(env, dt, op, buf, th);
2670         } else  {
2671                 CDEBUG(D_LAYOUT, "%s: unsupported xattr name:%s\n",
2672                        lod2obd(d)->obd_name, name);
2673                 GOTO(unlock, rc = -ENOTSUPP);
2674         }
2675 unlock:
2676         if (rc)
2677                 lod_object_free_striping(env, lo);
2678         dt_write_unlock(env, next);
2679
2680         RETURN(rc);
2681 }
2682
2683 /**
2684  * Merge layouts to form a mirrored file.
2685  */
2686 static int lod_declare_layout_merge(const struct lu_env *env,
2687                 struct dt_object *dt, const struct lu_buf *mbuf,
2688                 struct thandle *th)
2689 {
2690         struct lod_thread_info  *info = lod_env_info(env);
2691         struct lu_buf           *buf = &info->lti_buf;
2692         struct lod_object       *lo = lod_dt_obj(dt);
2693         struct lov_comp_md_v1   *lcm;
2694         struct lov_comp_md_v1   *cur_lcm;
2695         struct lov_comp_md_v1   *merge_lcm;
2696         struct lov_comp_md_entry_v1     *lcme;
2697         size_t size = 0;
2698         size_t offset;
2699         __u16 cur_entry_count;
2700         __u16 merge_entry_count;
2701         __u32 id = 0;
2702         __u16 mirror_id = 0;
2703         __u32 mirror_count;
2704         int     rc, i;
2705         ENTRY;
2706
2707         merge_lcm = mbuf->lb_buf;
2708         if (mbuf->lb_len < sizeof(*merge_lcm))
2709                 RETURN(-EINVAL);
2710
2711         /* must be an existing layout from disk */
2712         if (le32_to_cpu(merge_lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
2713                 RETURN(-EINVAL);
2714
2715         merge_entry_count = le16_to_cpu(merge_lcm->lcm_entry_count);
2716
2717         /* do not allow to merge two mirrored files */
2718         if (le16_to_cpu(merge_lcm->lcm_mirror_count))
2719                 RETURN(-EBUSY);
2720
2721         /* verify the target buffer */
2722         rc = lod_get_lov_ea(env, lo);
2723         if (rc <= 0)
2724                 RETURN(rc ? : -ENODATA);
2725
2726         cur_lcm = info->lti_ea_store;
2727         if (le32_to_cpu(cur_lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
2728                 RETURN(-EINVAL);
2729
2730         cur_entry_count = le16_to_cpu(cur_lcm->lcm_entry_count);
2731
2732         /* 'lcm_mirror_count + 1' is the current # of mirrors the file has */
2733         mirror_count = le16_to_cpu(cur_lcm->lcm_mirror_count) + 1;
2734         if (mirror_count + 1 > LUSTRE_MIRROR_COUNT_MAX)
2735                 RETURN(-ERANGE);
2736
2737         /* size of new layout */
2738         size = le32_to_cpu(cur_lcm->lcm_size) +
2739                le32_to_cpu(merge_lcm->lcm_size) - sizeof(*cur_lcm);
2740
2741         memset(buf, 0, sizeof(*buf));
2742         lu_buf_alloc(buf, size);
2743         if (buf->lb_buf == NULL)
2744                 RETURN(-ENOMEM);
2745
2746         lcm = buf->lb_buf;
2747         memcpy(lcm, cur_lcm, sizeof(*lcm) + cur_entry_count * sizeof(*lcme));
2748
2749         offset = sizeof(*lcm) +
2750                  sizeof(*lcme) * (cur_entry_count + merge_entry_count);
2751         for (i = 0; i < cur_entry_count; i++) {
2752                 struct lov_comp_md_entry_v1 *cur_lcme;
2753
2754                 lcme = &lcm->lcm_entries[i];
2755                 cur_lcme = &cur_lcm->lcm_entries[i];
2756
2757                 lcme->lcme_offset = cpu_to_le32(offset);
2758                 memcpy((char *)lcm + offset,
2759                        (char *)cur_lcm + le32_to_cpu(cur_lcme->lcme_offset),
2760                        le32_to_cpu(lcme->lcme_size));
2761
2762                 offset += le32_to_cpu(lcme->lcme_size);
2763
2764                 if (mirror_count == 1) {
2765                         /* new mirrored file, create new mirror ID */
2766                         id = pflr_id(1, i + 1);
2767                         lcme->lcme_id = cpu_to_le32(id);
2768                 }
2769
2770                 id = MAX(le32_to_cpu(lcme->lcme_id), id);
2771         }
2772
2773         mirror_id = mirror_id_of(id) + 1;
2774         for (i = 0; i < merge_entry_count; i++) {
2775                 struct lov_comp_md_entry_v1 *merge_lcme;
2776
2777                 merge_lcme = &merge_lcm->lcm_entries[i];
2778                 lcme = &lcm->lcm_entries[cur_entry_count + i];
2779
2780                 *lcme = *merge_lcme;
2781                 lcme->lcme_offset = cpu_to_le32(offset);
2782
2783                 id = pflr_id(mirror_id, i + 1);
2784                 lcme->lcme_id = cpu_to_le32(id);
2785
2786                 memcpy((char *)lcm + offset,
2787                        (char *)merge_lcm + le32_to_cpu(merge_lcme->lcme_offset),
2788                        le32_to_cpu(lcme->lcme_size));
2789
2790                 offset += le32_to_cpu(lcme->lcme_size);
2791         }
2792
2793         /* fixup layout information */
2794         lod_obj_inc_layout_gen(lo);
2795         lcm->lcm_layout_gen = cpu_to_le32(lo->ldo_layout_gen);
2796         lcm->lcm_size = cpu_to_le32(size);
2797         lcm->lcm_entry_count = cpu_to_le16(cur_entry_count + merge_entry_count);
2798         lcm->lcm_mirror_count = cpu_to_le16(mirror_count);
2799         if ((le16_to_cpu(lcm->lcm_flags) & LCM_FL_FLR_MASK) == LCM_FL_NOT_FLR)
2800                 lcm->lcm_flags = cpu_to_le32(LCM_FL_RDONLY);
2801
2802         LASSERT(dt_write_locked(env, dt_object_child(dt)));
2803         lod_object_free_striping(env, lo);
2804         rc = lod_parse_striping(env, lo, buf);
2805         if (rc)
2806                 GOTO(out, rc);
2807
2808         rc = lod_sub_declare_xattr_set(env, dt_object_child(dt), buf,
2809                                         XATTR_NAME_LOV, LU_XATTR_REPLACE, th);
2810
2811 out:
2812         lu_buf_free(buf);
2813         RETURN(rc);
2814 }
2815
2816 /**
2817  * Implementation of dt_object_operations::do_declare_xattr_set.
2818  *
2819  * \see dt_object_operations::do_declare_xattr_set() in the API description
2820  * for details.
2821  *
2822  * the extension to the API:
2823  *   - declaring LOVEA requests striping creation
2824  *   - LU_XATTR_REPLACE means layout swap
2825  */
2826 static int lod_declare_xattr_set(const struct lu_env *env,
2827                                  struct dt_object *dt,
2828                                  const struct lu_buf *buf,
2829                                  const char *name, int fl,
2830                                  struct thandle *th)
2831 {
2832         struct dt_object *next = dt_object_child(dt);
2833         struct lu_attr   *attr = &lod_env_info(env)->lti_attr;
2834         __u32             mode;
2835         int               rc;
2836         ENTRY;
2837
2838         mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
2839         if ((S_ISREG(mode) || mode == 0) &&
2840             !(fl & (LU_XATTR_REPLACE | LU_XATTR_MERGE)) &&
2841             (strcmp(name, XATTR_NAME_LOV) == 0 ||
2842              strcmp(name, XATTR_LUSTRE_LOV) == 0)) {
2843                 /*
2844                  * this is a request to create object's striping.
2845                  *
2846                  * allow to declare predefined striping on a new (!mode) object
2847                  * which is supposed to be replay of regular file creation
2848                  * (when LOV setting is declared)
2849                  *
2850                  * LU_XATTR_REPLACE is set to indicate a layout swap
2851                  */
2852                 if (dt_object_exists(dt)) {
2853                         rc = dt_attr_get(env, next, attr);
2854                         if (rc)
2855                                 RETURN(rc);
2856                 } else {
2857                         memset(attr, 0, sizeof(*attr));
2858                         attr->la_valid = LA_TYPE | LA_MODE;
2859                         attr->la_mode = S_IFREG;
2860                 }
2861                 rc = lod_declare_striped_create(env, dt, attr, buf, th);
2862         } else if (fl & LU_XATTR_MERGE) {
2863                 LASSERT(strcmp(name, XATTR_NAME_LOV) == 0 ||
2864                         strcmp(name, XATTR_LUSTRE_LOV) == 0);
2865                 rc = lod_declare_layout_merge(env, dt, buf, th);
2866         } else if (S_ISREG(mode) &&
2867                    strlen(name) > strlen(XATTR_LUSTRE_LOV) + 1 &&
2868                    strncmp(name, XATTR_LUSTRE_LOV,
2869                            strlen(XATTR_LUSTRE_LOV)) == 0) {
2870                 /*
2871                  * this is a request to modify object's striping.
2872                  * add/set/del component(s).
2873                  */
2874                 if (!dt_object_exists(dt))
2875                         RETURN(-ENOENT);
2876
2877                 rc = lod_declare_modify_layout(env, dt, name, buf, th);
2878         } else if (S_ISDIR(mode)) {
2879                 rc = lod_dir_declare_xattr_set(env, dt, buf, name, fl, th);
2880         } else if (strcmp(name, XATTR_NAME_FID) == 0) {
2881                 rc = lod_replace_parent_fid(env, dt, th, true);
2882         } else {
2883                 rc = lod_sub_declare_xattr_set(env, next, buf, name, fl, th);
2884         }
2885
2886         RETURN(rc);
2887 }
2888
2889 /**
2890  * Apply xattr changes to the object.
2891  *
2892  * Applies xattr changes to the object and the stripes if the latter exist.
2893  *
2894  * \param[in] env       execution environment
2895  * \param[in] dt        object
2896  * \param[in] buf       buffer pointing to the new value of xattr
2897  * \param[in] name      name of xattr
2898  * \param[in] fl        flags
2899  * \param[in] th        transaction handle
2900  *
2901  * \retval              0 on success
2902  * \retval              negative if failed
2903  */
2904 static int lod_xattr_set_internal(const struct lu_env *env,
2905                                   struct dt_object *dt,
2906                                   const struct lu_buf *buf,
2907                                   const char *name, int fl,
2908                                   struct thandle *th)
2909 {
2910         struct dt_object        *next = dt_object_child(dt);
2911         struct lod_object       *lo = lod_dt_obj(dt);
2912         int                     rc;
2913         int                     i;
2914         ENTRY;
2915
2916         rc = lod_sub_xattr_set(env, next, buf, name, fl, th);
2917         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
2918                 RETURN(rc);
2919
2920         /* Note: Do not set LinkEA on sub-stripes, otherwise
2921          * it will confuse the fid2path process(see mdt_path_current()).
2922          * The linkEA between master and sub-stripes is set in
2923          * lod_xattr_set_lmv(). */
2924         if (lo->ldo_dir_stripe_count == 0 || strcmp(name, XATTR_NAME_LINK) == 0)
2925                 RETURN(0);
2926
2927         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
2928                 LASSERT(lo->ldo_stripe[i]);
2929
2930                 rc = lod_sub_xattr_set(env, lo->ldo_stripe[i], buf, name,
2931                                        fl, th);
2932                 if (rc != 0)
2933                         break;
2934         }
2935
2936         RETURN(rc);
2937 }
2938
2939 /**
2940  * Delete an extended attribute.
2941  *
2942  * Deletes specified xattr from the object and the stripes if the latter exist.
2943  *
2944  * \param[in] env       execution environment
2945  * \param[in] dt        object
2946  * \param[in] name      name of xattr
2947  * \param[in] th        transaction handle
2948  *
2949  * \retval              0 on success
2950  * \retval              negative if failed
2951  */
2952 static int lod_xattr_del_internal(const struct lu_env *env,
2953                                   struct dt_object *dt,
2954                                   const char *name, struct thandle *th)
2955 {
2956         struct dt_object        *next = dt_object_child(dt);
2957         struct lod_object       *lo = lod_dt_obj(dt);
2958         int                     rc;
2959         int                     i;
2960         ENTRY;
2961
2962         rc = lod_sub_xattr_del(env, next, name, th);
2963         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
2964                 RETURN(rc);
2965
2966         if (lo->ldo_dir_stripe_count == 0)
2967                 RETURN(rc);
2968
2969         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
2970                 LASSERT(lo->ldo_stripe[i]);
2971
2972                 rc = lod_sub_xattr_del(env, lo->ldo_stripe[i], name, th);
2973                 if (rc != 0)
2974                         break;
2975         }
2976
2977         RETURN(rc);
2978 }
2979
2980 /**
2981  * Set default striping on a directory.
2982  *
2983  * Sets specified striping on a directory object unless it matches the default
2984  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
2985  * EA. This striping will be used when regular file is being created in this
2986  * directory.
2987  *
2988  * \param[in] env       execution environment
2989  * \param[in] dt        the striped object
2990  * \param[in] buf       buffer with the striping
2991  * \param[in] name      name of EA
2992  * \param[in] fl        xattr flag (see OSD API description)
2993  * \param[in] th        transaction handle
2994  *
2995  * \retval              0 on success
2996  * \retval              negative if failed
2997  */
2998 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
2999                                     struct dt_object *dt,
3000                                     const struct lu_buf *buf,
3001                                     const char *name, int fl,
3002                                     struct thandle *th)
3003 {
3004         struct lov_user_md_v1   *lum;
3005         struct lov_user_md_v3   *v3 = NULL;
3006         const char              *pool_name = NULL;
3007         int                      rc;
3008         bool                     is_del;
3009         ENTRY;
3010
3011         LASSERT(buf != NULL && buf->lb_buf != NULL);
3012         lum = buf->lb_buf;
3013
3014         switch (lum->lmm_magic) {
3015         case LOV_USER_MAGIC_V3:
3016                 v3 = buf->lb_buf;
3017                 if (v3->lmm_pool_name[0] != '\0')
3018                         pool_name = v3->lmm_pool_name;
3019                 /* fall through */
3020         case LOV_USER_MAGIC_V1:
3021                 /* if { size, offset, count } = { 0, -1, 0 } and no pool
3022                  * (i.e. all default values specified) then delete default
3023                  * striping from dir. */
3024                 CDEBUG(D_LAYOUT,
3025                        "set default striping: sz %u # %u offset %d %s %s\n",
3026                        (unsigned)lum->lmm_stripe_size,
3027                        (unsigned)lum->lmm_stripe_count,
3028                        (int)lum->lmm_stripe_offset,
3029                        v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
3030
3031                 is_del = LOVEA_DELETE_VALUES(lum->lmm_stripe_size,
3032                                              lum->lmm_stripe_count,
3033                                              lum->lmm_stripe_offset,
3034                                              pool_name);
3035                 break;
3036         case LOV_USER_MAGIC_COMP_V1:
3037                 is_del = false;
3038                 break;
3039         default:
3040                 CERROR("Invalid magic %x\n", lum->lmm_magic);
3041                 RETURN(-EINVAL);
3042         }
3043
3044         if (is_del) {
3045                 rc = lod_xattr_del_internal(env, dt, name, th);
3046                 if (rc == -ENODATA)
3047                         rc = 0;
3048         } else {
3049                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th);
3050         }
3051
3052         RETURN(rc);
3053 }
3054
3055 /**
3056  * Set default striping on a directory object.
3057  *
3058  * Sets specified striping on a directory object unless it matches the default
3059  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
3060  * EA. This striping will be used when a new directory is being created in the
3061  * directory.
3062  *
3063  * \param[in] env       execution environment
3064  * \param[in] dt        the striped object
3065  * \param[in] buf       buffer with the striping
3066  * \param[in] name      name of EA
3067  * \param[in] fl        xattr flag (see OSD API description)
3068  * \param[in] th        transaction handle
3069  *
3070  * \retval              0 on success
3071  * \retval              negative if failed
3072  */
3073 static int lod_xattr_set_default_lmv_on_dir(const struct lu_env *env,
3074                                             struct dt_object *dt,
3075                                             const struct lu_buf *buf,
3076                                             const char *name, int fl,
3077                                             struct thandle *th)
3078 {
3079         struct lmv_user_md_v1   *lum;
3080         int                      rc;
3081         ENTRY;
3082
3083         LASSERT(buf != NULL && buf->lb_buf != NULL);
3084         lum = buf->lb_buf;
3085
3086         CDEBUG(D_OTHER, "set default stripe_count # %u stripe_offset %d\n",
3087               le32_to_cpu(lum->lum_stripe_count),
3088               (int)le32_to_cpu(lum->lum_stripe_offset));
3089
3090         if (LMVEA_DELETE_VALUES((le32_to_cpu(lum->lum_stripe_count)),
3091                                  le32_to_cpu(lum->lum_stripe_offset)) &&
3092                                 le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC) {
3093                 rc = lod_xattr_del_internal(env, dt, name, th);
3094                 if (rc == -ENODATA)
3095                         rc = 0;
3096         } else {
3097                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th);
3098                 if (rc != 0)
3099                         RETURN(rc);
3100         }
3101
3102         RETURN(rc);
3103 }
3104
3105 /**
3106  * Turn directory into a striped directory.
3107  *
3108  * During replay the client sends the striping created before MDT
3109  * failure, then the layer above LOD sends this defined striping
3110  * using ->do_xattr_set(), so LOD uses this method to replay creation
3111  * of the stripes. Notice the original information for the striping
3112  * (#stripes, FIDs, etc) was transferred in declare path.
3113  *
3114  * \param[in] env       execution environment
3115  * \param[in] dt        the striped object
3116  * \param[in] buf       not used currently
3117  * \param[in] name      not used currently
3118  * \param[in] fl        xattr flag (see OSD API description)
3119  * \param[in] th        transaction handle
3120  *
3121  * \retval              0 on success
3122  * \retval              negative if failed
3123  */
3124 static int lod_xattr_set_lmv(const struct lu_env *env, struct dt_object *dt,
3125                              const struct lu_buf *buf, const char *name,
3126                              int fl, struct thandle *th)
3127 {
3128         struct lod_object       *lo = lod_dt_obj(dt);
3129         struct lod_thread_info  *info = lod_env_info(env);
3130         struct lu_attr          *attr = &info->lti_attr;
3131         struct dt_object_format *dof = &info->lti_format;
3132         struct lu_buf           lmv_buf;
3133         struct lu_buf           slave_lmv_buf;
3134         struct lmv_mds_md_v1    *lmm;
3135         struct lmv_mds_md_v1    *slave_lmm = NULL;
3136         struct dt_insert_rec    *rec = &info->lti_dt_rec;
3137         int                     i;
3138         int                     rc;
3139         ENTRY;
3140
3141         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
3142                 RETURN(-ENOTDIR);
3143
3144         /* The stripes are supposed to be allocated in declare phase,
3145          * if there are no stripes being allocated, it will skip */
3146         if (lo->ldo_dir_stripe_count == 0)
3147                 RETURN(0);
3148
3149         rc = dt_attr_get(env, dt_object_child(dt), attr);
3150         if (rc != 0)
3151                 RETURN(rc);
3152
3153         attr->la_valid = LA_ATIME | LA_MTIME | LA_CTIME |
3154                          LA_MODE | LA_UID | LA_GID | LA_TYPE | LA_PROJID;
3155         dof->dof_type = DFT_DIR;
3156
3157         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
3158         if (rc != 0)
3159                 RETURN(rc);
3160         lmm = lmv_buf.lb_buf;
3161
3162         OBD_ALLOC_PTR(slave_lmm);
3163         if (slave_lmm == NULL)
3164                 RETURN(-ENOMEM);
3165
3166         lod_prep_slave_lmv_md(slave_lmm, lmm);
3167         slave_lmv_buf.lb_buf = slave_lmm;
3168         slave_lmv_buf.lb_len = sizeof(*slave_lmm);
3169
3170         rec->rec_type = S_IFDIR;
3171         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
3172                 struct dt_object *dto;
3173                 char             *stripe_name = info->lti_key;
3174                 struct lu_name          *sname;
3175                 struct linkea_data       ldata          = { NULL };
3176                 struct lu_buf            linkea_buf;
3177
3178                 dto = lo->ldo_stripe[i];
3179
3180                 dt_write_lock(env, dto, MOR_TGT_CHILD);
3181                 rc = lod_sub_create(env, dto, attr, NULL, dof, th);
3182                 if (rc != 0) {
3183                         dt_write_unlock(env, dto);
3184                         GOTO(out, rc);
3185                 }
3186
3187                 rc = lod_sub_ref_add(env, dto, th);
3188                 dt_write_unlock(env, dto);
3189                 if (rc != 0)
3190                         GOTO(out, rc);
3191
3192                 rec->rec_fid = lu_object_fid(&dto->do_lu);
3193                 rc = lod_sub_insert(env, dto, (const struct dt_rec *)rec,
3194                                     (const struct dt_key *)dot, th, 0);
3195                 if (rc != 0)
3196                         GOTO(out, rc);
3197
3198                 rec->rec_fid = lu_object_fid(&dt->do_lu);
3199                 rc = lod_sub_insert(env, dto, (struct dt_rec *)rec,
3200                                     (const struct dt_key *)dotdot, th, 0);
3201                 if (rc != 0)
3202                         GOTO(out, rc);
3203
3204                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
3205                     cfs_fail_val != i) {
3206                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
3207                             cfs_fail_val == i)
3208                                 slave_lmm->lmv_master_mdt_index =
3209                                                         cpu_to_le32(i + 1);
3210                         else
3211                                 slave_lmm->lmv_master_mdt_index =
3212                                                         cpu_to_le32(i);
3213
3214                         rc = lod_sub_xattr_set(env, dto, &slave_lmv_buf,
3215                                                XATTR_NAME_LMV, fl, th);
3216                         if (rc != 0)
3217                                 GOTO(out, rc);
3218                 }
3219
3220                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
3221                     cfs_fail_val == i)
3222                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
3223                                  PFID(lu_object_fid(&dto->do_lu)), i + 1);
3224                 else
3225                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
3226                                  PFID(lu_object_fid(&dto->do_lu)), i);
3227
3228                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
3229                 rc = linkea_links_new(&ldata, &info->lti_linkea_buf,
3230                                       sname, lu_object_fid(&dt->do_lu));
3231                 if (rc != 0)
3232                         GOTO(out, rc);
3233
3234                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
3235                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
3236                 rc = lod_sub_xattr_set(env, dto, &linkea_buf,
3237                                        XATTR_NAME_LINK, 0, th);
3238                 if (rc != 0)
3239                         GOTO(out, rc);
3240
3241                 rec->rec_fid = lu_object_fid(&dto->do_lu);
3242                 rc = lod_sub_insert(env, dt_object_child(dt),
3243                                     (const struct dt_rec *)rec,
3244                                     (const struct dt_key *)stripe_name, th, 0);
3245                 if (rc != 0)
3246                         GOTO(out, rc);
3247
3248                 rc = lod_sub_ref_add(env, dt_object_child(dt), th);
3249                 if (rc != 0)
3250                         GOTO(out, rc);
3251         }
3252
3253         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MASTER_LMV))
3254                 rc = lod_sub_xattr_set(env, dt_object_child(dt),
3255                                        &lmv_buf, XATTR_NAME_LMV, fl, th);
3256 out:
3257         if (slave_lmm != NULL)
3258                 OBD_FREE_PTR(slave_lmm);
3259
3260         RETURN(rc);
3261 }
3262
3263 /**
3264  * Helper function to declare/execute creation of a striped directory
3265  *
3266  * Called in declare/create object path, prepare striping for a directory
3267  * and prepare defaults data striping for the objects to be created in
3268  * that directory. Notice the function calls "declaration" or "execution"
3269  * methods depending on \a declare param. This is a consequence of the
3270  * current approach while we don't have natural distributed transactions:
3271  * we basically execute non-local updates in the declare phase. So, the
3272  * arguments for the both phases are the same and this is the reason for
3273  * this function to exist.
3274  *
3275  * \param[in] env       execution environment
3276  * \param[in] dt        object
3277  * \param[in] attr      attributes the stripes will be created with
3278  * \param[in] dof       format of stripes (see OSD API description)
3279  * \param[in] th        transaction handle
3280  * \param[in] declare   where to call "declare" or "execute" methods
3281  *
3282  * \retval              0 on success
3283  * \retval              negative if failed
3284  */
3285 static int lod_dir_striping_create_internal(const struct lu_env *env,
3286                                             struct dt_object *dt,
3287                                             struct lu_attr *attr,
3288                                             struct dt_object_format *dof,
3289                                             struct thandle *th,
3290                                             bool declare)
3291 {
3292         struct lod_thread_info *info = lod_env_info(env);
3293         struct lod_object *lo = lod_dt_obj(dt);
3294         const struct lod_default_striping *lds = lo->ldo_def_striping;
3295         int rc;
3296         ENTRY;
3297
3298         LASSERT(ergo(lds != NULL,
3299                      lds->lds_def_striping_set ||
3300                      lds->lds_dir_def_striping_set));
3301
3302         if (!LMVEA_DELETE_VALUES(lo->ldo_dir_stripe_count,
3303                                  lo->ldo_dir_stripe_offset)) {
3304                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
3305                 int stripe_count = lo->ldo_dir_stripe_count;
3306
3307                 if (info->lti_ea_store_size < sizeof(*v1)) {
3308                         rc = lod_ea_store_resize(info, sizeof(*v1));
3309                         if (rc != 0)
3310                                 RETURN(rc);
3311                         v1 = info->lti_ea_store;
3312                 }
3313
3314                 memset(v1, 0, sizeof(*v1));
3315                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
3316                 v1->lum_stripe_count = cpu_to_le32(stripe_count);
3317                 v1->lum_stripe_offset =
3318                                 cpu_to_le32(lo->ldo_dir_stripe_offset);
3319
3320                 info->lti_buf.lb_buf = v1;
3321                 info->lti_buf.lb_len = sizeof(*v1);
3322
3323                 if (declare)
3324                         rc = lod_declare_xattr_set_lmv(env, dt, attr,
3325                                                        &info->lti_buf, dof, th);
3326                 else
3327                         rc = lod_xattr_set_lmv(env, dt, &info->lti_buf,
3328                                                XATTR_NAME_LMV, 0, th);
3329                 if (rc != 0)
3330                         RETURN(rc);
3331         }
3332
3333         /* Transfer default LMV striping from the parent */
3334         if (lds != NULL && lds->lds_dir_def_striping_set &&
3335             !LMVEA_DELETE_VALUES(lds->lds_dir_def_stripe_count,
3336                                  lds->lds_dir_def_stripe_offset)) {
3337                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
3338
3339                 if (info->lti_ea_store_size < sizeof(*v1)) {
3340                         rc = lod_ea_store_resize(info, sizeof(*v1));
3341                         if (rc != 0)
3342                                 RETURN(rc);
3343                         v1 = info->lti_ea_store;
3344                 }
3345
3346                 memset(v1, 0, sizeof(*v1));
3347                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
3348                 v1->lum_stripe_count =
3349                         cpu_to_le32(lds->lds_dir_def_stripe_count);
3350                 v1->lum_stripe_offset =
3351                         cpu_to_le32(lds->lds_dir_def_stripe_offset);
3352                 v1->lum_hash_type =
3353                         cpu_to_le32(lds->lds_dir_def_hash_type);
3354
3355                 info->lti_buf.lb_buf = v1;
3356                 info->lti_buf.lb_len = sizeof(*v1);
3357                 if (declare)
3358                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
3359                                                        XATTR_NAME_DEFAULT_LMV,
3360                                                        0, th);
3361                 else
3362                         rc = lod_xattr_set_default_lmv_on_dir(env, dt,
3363                                                   &info->lti_buf,
3364                                                   XATTR_NAME_DEFAULT_LMV, 0,
3365                                                   th);
3366                 if (rc != 0)
3367                         RETURN(rc);
3368         }
3369
3370         /* Transfer default LOV striping from the parent */
3371         if (lds != NULL && lds->lds_def_striping_set &&
3372             lds->lds_def_comp_cnt != 0) {
3373                 struct lov_mds_md *lmm;
3374                 int lmm_size = lod_comp_md_size(lo, true);
3375
3376                 if (info->lti_ea_store_size < lmm_size) {
3377                         rc = lod_ea_store_resize(info, lmm_size);
3378                         if (rc != 0)
3379                                 RETURN(rc);
3380                 }
3381                 lmm = info->lti_ea_store;
3382
3383                 rc = lod_generate_lovea(env, lo, lmm, &lmm_size, true);
3384                 if (rc != 0)
3385                         RETURN(rc);
3386
3387                 info->lti_buf.lb_buf = lmm;
3388                 info->lti_buf.lb_len = lmm_size;
3389
3390                 if (declare)
3391                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
3392                                                        XATTR_NAME_LOV, 0, th);
3393                 else
3394                         rc = lod_xattr_set_lov_on_dir(env, dt, &info->lti_buf,
3395                                                       XATTR_NAME_LOV, 0, th);
3396                 if (rc != 0)
3397                         RETURN(rc);
3398         }
3399
3400         RETURN(0);
3401 }
3402
3403 static int lod_declare_dir_striping_create(const struct lu_env *env,
3404                                            struct dt_object *dt,
3405                                            struct lu_attr *attr,
3406                                            struct dt_object_format *dof,
3407                                            struct thandle *th)
3408 {
3409         return lod_dir_striping_create_internal(env, dt, attr, dof, th, true);
3410 }
3411
3412 static int lod_dir_striping_create(const struct lu_env *env,
3413                                    struct dt_object *dt,
3414                                    struct lu_attr *attr,
3415                                    struct dt_object_format *dof,
3416                                    struct thandle *th)
3417 {
3418         return lod_dir_striping_create_internal(env, dt, attr, dof, th, false);
3419 }
3420
3421 /**
3422  * Make LOV EA for striped object.
3423  *
3424  * Generate striping information and store it in the LOV EA of the given
3425  * object. The caller must ensure nobody else is calling the function
3426  * against the object concurrently. The transaction must be started.
3427  * FLDB service must be running as well; it's used to map FID to the target,
3428  * which is stored in LOV EA.
3429  *
3430  * \param[in] env               execution environment for this thread
3431  * \param[in] lo                LOD object
3432  * \param[in] th                transaction handle
3433  *
3434  * \retval                      0 if LOV EA is stored successfully
3435  * \retval                      negative error number on failure
3436  */
3437 static int lod_generate_and_set_lovea(const struct lu_env *env,
3438                                       struct lod_object *lo,
3439                                       struct thandle *th)
3440 {
3441         struct lod_thread_info  *info = lod_env_info(env);
3442         struct dt_object        *next = dt_object_child(&lo->ldo_obj);
3443         struct lov_mds_md_v1    *lmm;
3444         int                      rc, lmm_size;
3445         ENTRY;
3446
3447         LASSERT(lo);
3448
3449         if (lo->ldo_comp_cnt == 0) {
3450                 lod_object_free_striping(env, lo);
3451                 rc = lod_sub_xattr_del(env, next, XATTR_NAME_LOV, th);
3452                 RETURN(rc);
3453         }
3454
3455         lmm_size = lod_comp_md_size(lo, false);
3456         if (info->lti_ea_store_size < lmm_size) {
3457                 rc = lod_ea_store_resize(info, lmm_size);
3458                 if (rc)
3459                         RETURN(rc);
3460         }
3461         lmm = info->lti_ea_store;
3462
3463         rc = lod_generate_lovea(env, lo, lmm, &lmm_size, false);
3464         if (rc)
3465                 RETURN(rc);
3466
3467         info->lti_buf.lb_buf = lmm;
3468         info->lti_buf.lb_len = lmm_size;
3469         rc = lod_sub_xattr_set(env, next, &info->lti_buf,
3470                                XATTR_NAME_LOV, 0, th);
3471         RETURN(rc);
3472 }
3473
3474 /**
3475  * Delete layout component(s)
3476  *
3477  * \param[in] env       execution environment for this thread
3478  * \param[in] dt        object
3479  * \param[in] th        transaction handle
3480  *
3481  * \retval      0 on success
3482  * \retval      negative error number on failure
3483  */
3484 static int lod_layout_del(const struct lu_env *env, struct dt_object *dt,
3485                           struct thandle *th)
3486 {
3487         struct lod_layout_component     *lod_comp;
3488         struct lod_object       *lo = lod_dt_obj(dt);
3489         struct dt_object        *next = dt_object_child(dt);
3490         struct lu_attr  *attr = &lod_env_info(env)->lti_attr;
3491         int     rc, i, j, left;
3492
3493         LASSERT(lo->ldo_is_composite);
3494         LASSERT(lo->ldo_comp_cnt > 0 && lo->ldo_comp_entries != NULL);
3495
3496         left = lo->ldo_comp_cnt;
3497         for (i = (lo->ldo_comp_cnt - 1); i >= 0; i--) {
3498                 lod_comp = &lo->ldo_comp_entries[i];
3499
3500                 if (lod_comp->llc_id != LCME_ID_INVAL)
3501                         break;
3502                 left--;
3503
3504                 /* Not instantiated component */
3505                 if (lod_comp->llc_stripe == NULL)
3506                         continue;
3507
3508                 LASSERT(lod_comp->llc_stripe_count > 0);
3509                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
3510                         struct dt_object *obj = lod_comp->llc_stripe[j];
3511
3512                         if (obj == NULL)
3513                                 continue;
3514                         rc = lod_sub_destroy(env, obj, th);
3515                         if (rc)
3516                                 GOTO(out, rc);
3517
3518                         lu_object_put(env, &obj->do_lu);
3519                         lod_comp->llc_stripe[j] = NULL;
3520                 }
3521                 OBD_FREE(lod_comp->llc_stripe, sizeof(struct dt_object *) *
3522                                         lod_comp->llc_stripes_allocated);
3523                 lod_comp->llc_stripe = NULL;
3524                 lod_comp->llc_stripes_allocated = 0;
3525                 lod_obj_set_pool(lo, i, NULL);
3526                 if (lod_comp->llc_ostlist.op_array) {
3527                         OBD_FREE(lod_comp->llc_ostlist.op_array,
3528                                  lod_comp->llc_ostlist.op_size);
3529                         lod_comp->llc_ostlist.op_array = NULL;
3530                         lod_comp->llc_ostlist.op_size = 0;
3531                 }
3532         }
3533
3534         LASSERTF(left >= 0 && left < lo->ldo_comp_cnt, "left = %d\n", left);
3535         if (left > 0) {
3536                 struct lod_layout_component     *comp_array;
3537
3538                 OBD_ALLOC(comp_array, sizeof(*comp_array) * left);
3539                 if (comp_array == NULL)
3540                         GOTO(out, rc = -ENOMEM);
3541
3542                 memcpy(&comp_array[0], &lo->ldo_comp_entries[0],
3543                        sizeof(*comp_array) * left);
3544
3545                 OBD_FREE(lo->ldo_comp_entries,
3546                          sizeof(*comp_array) * lo->ldo_comp_cnt);
3547                 lo->ldo_comp_entries = comp_array;
3548                 lo->ldo_comp_cnt = left;
3549
3550                 LASSERT(lo->ldo_mirror_count == 1);
3551                 lo->ldo_mirrors[0].lme_end = left - 1;
3552                 lod_obj_inc_layout_gen(lo);
3553         } else {
3554                 lod_free_comp_entries(lo);
3555         }
3556
3557         LASSERT(dt_object_exists(dt));
3558         rc = dt_attr_get(env, next, attr);
3559         if (rc)
3560                 GOTO(out, rc);
3561
3562         if (attr->la_size > 0) {
3563                 attr->la_size = 0;
3564                 attr->la_valid = LA_SIZE;
3565                 rc = lod_sub_attr_set(env, next, attr, th);
3566                 if (rc)
3567                         GOTO(out, rc);
3568         }
3569
3570         rc = lod_generate_and_set_lovea(env, lo, th);
3571         EXIT;
3572 out:
3573         if (rc)
3574                 lod_object_free_striping(env, lo);
3575         return rc;
3576 }
3577
3578 /**
3579  * Implementation of dt_object_operations::do_xattr_set.
3580  *
3581  * Sets specified extended attribute on the object. Three types of EAs are
3582  * special:
3583  *   LOV EA - stores striping for a regular file or default striping (when set
3584  *            on a directory)
3585  *   LMV EA - stores a marker for the striped directories
3586  *   DMV EA - stores default directory striping
3587  *
3588  * When striping is applied to a non-striped existing object (this is called
3589  * late striping), then LOD notices the caller wants to turn the object into a
3590  * striped one. The stripe objects are created and appropriate EA is set:
3591  * LOV EA storing all the stripes directly or LMV EA storing just a small header
3592  * with striping configuration.
3593  *
3594  * \see dt_object_operations::do_xattr_set() in the API description for details.
3595  */
3596 static int lod_xattr_set(const struct lu_env *env,
3597                          struct dt_object *dt, const struct lu_buf *buf,
3598                          const char *name, int fl, struct thandle *th)
3599 {
3600         struct dt_object        *next = dt_object_child(dt);
3601         int                      rc;
3602         ENTRY;
3603
3604         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
3605             strcmp(name, XATTR_NAME_LMV) == 0) {
3606                 struct lmv_mds_md_v1 *lmm = buf->lb_buf;
3607
3608                 if (lmm != NULL && le32_to_cpu(lmm->lmv_hash_type) &
3609                                                 LMV_HASH_FLAG_MIGRATION)
3610                         rc = lod_sub_xattr_set(env, next, buf, name, fl, th);
3611                 else
3612                         rc = lod_dir_striping_create(env, dt, NULL, NULL, th);
3613
3614                 RETURN(rc);
3615         }
3616
3617         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
3618             strcmp(name, XATTR_NAME_LOV) == 0) {
3619                 /* default LOVEA */
3620                 rc = lod_xattr_set_lov_on_dir(env, dt, buf, name, fl, th);
3621                 RETURN(rc);
3622         } else if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
3623                    strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
3624                 /* default LMVEA */
3625                 rc = lod_xattr_set_default_lmv_on_dir(env, dt, buf, name, fl,
3626                                                       th);
3627                 RETURN(rc);
3628         } else if (S_ISREG(dt->do_lu.lo_header->loh_attr) &&
3629                    (!strcmp(name, XATTR_NAME_LOV) ||
3630                     !strncmp(name, XATTR_LUSTRE_LOV,
3631                              strlen(XATTR_LUSTRE_LOV)))) {
3632                 /* in case of lov EA swap, just set it
3633                  * if not, it is a replay so check striping match what we
3634                  * already have during req replay, declare_xattr_set()
3635                  * defines striping, then create() does the work */
3636                 if (fl & LU_XATTR_REPLACE) {
3637                         /* free stripes, then update disk */
3638                         lod_object_free_striping(env, lod_dt_obj(dt));
3639
3640                         rc = lod_sub_xattr_set(env, next, buf, name, fl, th);
3641                 } else if (dt_object_remote(dt)) {
3642                         /* This only happens during migration, see
3643                          * mdd_migrate_create(), in which Master MDT will
3644                          * create a remote target object, and only set
3645                          * (migrating) stripe EA on the remote object,
3646                          * and does not need creating each stripes. */
3647                         rc = lod_sub_xattr_set(env, next, buf, name,
3648                                                       fl, th);
3649                 } else if (strcmp(name, XATTR_LUSTRE_LOV".del") == 0) {
3650                         /* delete component(s) */
3651                         LASSERT(lod_dt_obj(dt)->ldo_comp_cached);
3652                         rc = lod_layout_del(env, dt, th);
3653                 } else {
3654                         /*
3655                          * When 'name' is XATTR_LUSTRE_LOV or XATTR_NAME_LOV,
3656                          * it's going to create create file with specified
3657                          * component(s), the striping must have not being
3658                          * cached in this case;
3659                          *
3660                          * Otherwise, it's going to add/change component(s) to
3661                          * an existing file, the striping must have been cached
3662                          * in this case.
3663                          */
3664                         LASSERT(equi(!strcmp(name, XATTR_LUSTRE_LOV) ||
3665                                      !strcmp(name, XATTR_NAME_LOV),
3666                                 !lod_dt_obj(dt)->ldo_comp_cached));
3667
3668                         rc = lod_striped_create(env, dt, NULL, NULL, th);
3669                 }
3670                 RETURN(rc);
3671         } else if (strcmp(name, XATTR_NAME_FID) == 0) {
3672                 rc = lod_replace_parent_fid(env, dt, th, false);
3673
3674                 RETURN(rc);
3675         }
3676
3677         /* then all other xattr */
3678         rc = lod_xattr_set_internal(env, dt, buf, name, fl, th);
3679
3680         RETURN(rc);
3681 }
3682
3683 /**
3684  * Implementation of dt_object_operations::do_declare_xattr_del.
3685  *
3686  * \see dt_object_operations::do_declare_xattr_del() in the API description
3687  * for details.
3688  */
3689 static int lod_declare_xattr_del(const struct lu_env *env,
3690                                  struct dt_object *dt, const char *name,
3691                                  struct thandle *th)
3692 {
3693         struct lod_object       *lo = lod_dt_obj(dt);
3694         int                     rc;
3695         int                     i;
3696         ENTRY;
3697
3698         rc = lod_sub_declare_xattr_del(env, dt_object_child(dt), name, th);
3699         if (rc != 0)
3700                 RETURN(rc);