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