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