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