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