Whamcloud - gitweb
LU-12624 lod: alloc dir stripes by QoS
[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 /**
1949  * Allocate a striping on a predefined set of MDTs.
1950  *
1951  * Allocates new striping using the MDT index range provided by the data from
1952  * the lum_obejcts contained in the lmv_user_md passed to this method if
1953  * \a is_specific is true; or allocates new layout starting from MDT index in
1954  * lo->ldo_dir_stripe_offset. The exact order of MDTs is not important and
1955  * varies depending on MDT status. The number of stripes needed and stripe
1956  * offset are taken from the object. If that number cannot be met, then the
1957  * function returns an error and then it's the caller's responsibility to
1958  * release the stripes allocated. All the internal structures are protected,
1959  * but no concurrent allocation is allowed on the same objects.
1960  *
1961  * \param[in] env               execution environment for this thread
1962  * \param[in] lo                LOD object
1963  * \param[out] stripes          striping created
1964  * \param[out] mdt_indices      MDT indices of striping created
1965  * \param[in] is_specific       true if the MDTs are provided by lum; false if
1966  *                              only the starting MDT index is provided
1967  *
1968  * \retval positive     stripes allocated, including the first stripe allocated
1969  *                      outside
1970  * \retval negative     errno on failure
1971  */
1972 static int lod_mdt_alloc_specific(const struct lu_env *env,
1973                                   struct lod_object *lo,
1974                                   struct dt_object **stripes,
1975                                   __u32 *mdt_indices, bool is_specific)
1976 {
1977         struct lod_thread_info  *info = lod_env_info(env);
1978         struct lod_device *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1979         struct lu_tgt_descs *ltd = &lod->lod_mdt_descs;
1980         struct lu_tgt_desc *tgt = NULL;
1981         struct lu_object_conf conf = { .loc_flags = LOC_F_NEW };
1982         struct dt_device *tgt_dt = NULL;
1983         struct lu_fid fid = { 0 };
1984         struct dt_object *dto;
1985         u32 master_index;
1986         u32 stripe_count = lo->ldo_dir_stripe_count;
1987         int stripe_idx = 1;
1988         int j;
1989         int idx;
1990         int rc;
1991
1992         master_index = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id;
1993         if (stripe_count > 1)
1994                 /* Set the start index for the 2nd stripe allocation */
1995                 mdt_indices[1] = (mdt_indices[0] + 1) %
1996                                         (lod->lod_remote_mdt_count + 1);
1997
1998         for (; stripe_idx < stripe_count; stripe_idx++) {
1999                 /* Try to find next avaible target */
2000                 idx = mdt_indices[stripe_idx];
2001                 for (j = 0; j < lod->lod_remote_mdt_count;
2002                      j++, idx = (idx + 1) % (lod->lod_remote_mdt_count + 1)) {
2003                         bool already_allocated = false;
2004                         __u32 k;
2005
2006                         CDEBUG(D_INFO, "try idx %d, mdt cnt %u, allocated %u\n",
2007                                idx, lod->lod_remote_mdt_count + 1, stripe_idx);
2008
2009                         if (likely(!is_specific &&
2010                                    !OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE))) {
2011                                 /* check whether the idx already exists
2012                                  * in current allocated array */
2013                                 for (k = 0; k < stripe_idx; k++) {
2014                                         if (mdt_indices[k] == idx) {
2015                                                 already_allocated = true;
2016                                                 break;
2017                                         }
2018                                 }
2019
2020                                 if (already_allocated)
2021                                         continue;
2022                         }
2023
2024                         /* Sigh, this index is not in the bitmap, let's check
2025                          * next available target */
2026                         if (!cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx) &&
2027                             idx != master_index)
2028                                 continue;
2029
2030                         if (idx == master_index) {
2031                                 /* Allocate the FID locally */
2032                                 rc = obd_fid_alloc(env, lod->lod_child_exp,
2033                                                    &fid, NULL);
2034                                 if (rc < 0)
2035                                         continue;
2036                                 tgt_dt = lod->lod_child;
2037                                 break;
2038                         }
2039
2040                         /* check the status of the OSP */
2041                         tgt = LTD_TGT(ltd, idx);
2042                         if (!tgt)
2043                                 continue;
2044
2045                         tgt_dt = tgt->ltd_tgt;
2046                         rc = dt_statfs(env, tgt_dt, &info->lti_osfs);
2047                         if (rc)
2048                                 /* this OSP doesn't feel well */
2049                                 continue;
2050
2051                         rc = obd_fid_alloc(env, tgt->ltd_exp, &fid, NULL);
2052                         if (rc < 0)
2053                                 continue;
2054
2055                         break;
2056                 }
2057
2058                 /* Can not allocate more stripes */
2059                 if (j == lod->lod_remote_mdt_count) {
2060                         CDEBUG(D_INFO, "%s: require stripes %u only get %d\n",
2061                                lod2obd(lod)->obd_name, stripe_count,
2062                                stripe_idx);
2063                         break;
2064                 }
2065
2066                 CDEBUG(D_INFO, "Get idx %d, for stripe %d "DFID"\n",
2067                        idx, stripe_idx, PFID(&fid));
2068                 mdt_indices[stripe_idx] = idx;
2069                 /* Set the start index for next stripe allocation */
2070                 if (!is_specific && stripe_idx < stripe_count - 1) {
2071                         /*
2072                          * for large dir test, put all other slaves on one
2073                          * remote MDT, otherwise we may save too many local
2074                          * slave locks which will exceed RS_MAX_LOCKS.
2075                          */
2076                         if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE)))
2077                                 idx = master_index;
2078                         mdt_indices[stripe_idx + 1] = (idx + 1) %
2079                                            (lod->lod_remote_mdt_count + 1);
2080                 }
2081                 /* tgt_dt and fid must be ready after search avaible OSP
2082                  * in the above loop */
2083                 LASSERT(tgt_dt != NULL);
2084                 LASSERT(fid_is_sane(&fid));
2085
2086                 /* fail a remote stripe FID allocation */
2087                 if (stripe_idx && OBD_FAIL_CHECK(OBD_FAIL_MDS_STRIPE_FID))
2088                         continue;
2089
2090                 dto = dt_locate_at(env, tgt_dt, &fid,
2091                                   lo->ldo_obj.do_lu.lo_dev->ld_site->ls_top_dev,
2092                                   &conf);
2093                 if (IS_ERR(dto)) {
2094                         rc = PTR_ERR(dto);
2095                         goto error;
2096                 }
2097
2098                 stripes[stripe_idx] = dto;
2099         }
2100
2101         return stripe_idx;
2102
2103 error:
2104         for (j = 1; j < stripe_idx; j++) {
2105                 LASSERT(stripes[j] != NULL);
2106                 dt_object_put(env, stripes[j]);
2107                 stripes[j] = NULL;
2108         }
2109         return rc;
2110 }
2111
2112 static int lod_prep_md_striped_create(const struct lu_env *env,
2113                                       struct dt_object *dt,
2114                                       struct lu_attr *attr,
2115                                       const struct lmv_user_md_v1 *lum,
2116                                       struct dt_object_format *dof,
2117                                       struct thandle *th)
2118 {
2119         struct lod_device *lod = lu2lod_dev(dt->do_lu.lo_dev);
2120         struct lod_object *lo = lod_dt_obj(dt);
2121         struct dt_object **stripes;
2122         struct lu_object_conf conf = { .loc_flags = LOC_F_NEW };
2123         struct lu_fid fid = { 0 };
2124         __u32 stripe_count;
2125         int i;
2126         int rc = 0;
2127
2128         ENTRY;
2129
2130         /* The lum has been verifed in lod_verify_md_striping */
2131         LASSERT(le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC ||
2132                 le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC_SPECIFIC);
2133
2134         stripe_count = lo->ldo_dir_stripe_count;
2135
2136         OBD_ALLOC(stripes, sizeof(stripes[0]) * stripe_count);
2137         if (!stripes)
2138                 RETURN(-ENOMEM);
2139
2140         /* Allocate the first stripe locally */
2141         rc = obd_fid_alloc(env, lod->lod_child_exp, &fid, NULL);
2142         if (rc < 0)
2143                 GOTO(out, rc);
2144
2145         stripes[0] = dt_locate_at(env, lod->lod_child, &fid,
2146                                   dt->do_lu.lo_dev->ld_site->ls_top_dev, &conf);
2147         if (IS_ERR(stripes[0]))
2148                 GOTO(out, rc = PTR_ERR(stripes[0]));
2149
2150         if (lo->ldo_dir_stripe_offset == LMV_OFFSET_DEFAULT) {
2151                 lod_qos_statfs_update(env, lod, &lod->lod_mdt_descs);
2152                 rc = lod_mdt_alloc_qos(env, lo, stripes);
2153                 if (rc == -EAGAIN)
2154                         rc = lod_mdt_alloc_rr(env, lo, stripes);
2155         } else {
2156                 int *idx_array;
2157                 bool is_specific = false;
2158
2159                 OBD_ALLOC(idx_array, sizeof(idx_array[0]) * stripe_count);
2160                 if (!idx_array)
2161                         GOTO(out, rc = -ENOMEM);
2162
2163                 if (le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC_SPECIFIC) {
2164                         is_specific = true;
2165                         for (i = 0; i < stripe_count; i++)
2166                                 idx_array[i] =
2167                                        le32_to_cpu(lum->lum_objects[i].lum_mds);
2168                 }
2169
2170                 /* stripe 0 is local */
2171                 idx_array[0] =
2172                         lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id;
2173                 rc = lod_mdt_alloc_specific(env, lo, stripes, idx_array,
2174                                             is_specific);
2175                 OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count);
2176         }
2177
2178         if (rc < 0)
2179                 GOTO(out, rc);
2180
2181         LASSERT(rc > 0);
2182
2183         lo->ldo_dir_striped = 1;
2184         lo->ldo_stripe = stripes;
2185         lo->ldo_dir_stripe_count = rc;
2186         lo->ldo_dir_stripes_allocated = stripe_count;
2187         smp_mb();
2188         lo->ldo_dir_stripe_loaded = 1;
2189
2190         rc = lod_dir_declare_create_stripes(env, dt, attr, dof, th);
2191         if (rc < 0)
2192                 lod_striping_free(env, lo);
2193
2194         RETURN(rc);
2195
2196 out:
2197         LASSERT(rc < 0);
2198         if (!IS_ERR_OR_NULL(stripes[0]))
2199                 dt_object_put(env, stripes[0]);
2200         for (i = 1; i < stripe_count; i++)
2201                 LASSERT(!stripes[i]);
2202         OBD_FREE(stripes, sizeof(stripes[0]) * stripe_count);
2203
2204         return rc;
2205 }
2206
2207 /**
2208  *
2209  * Alloc cached foreign LMV
2210  *
2211  * \param[in] lo        object
2212  * \param[in] size      size of foreign LMV
2213  *
2214  * \retval              0 on success
2215  * \retval              negative if failed
2216  */
2217 int lod_alloc_foreign_lmv(struct lod_object *lo, size_t size)
2218 {
2219         OBD_ALLOC_LARGE(lo->ldo_foreign_lmv, size);
2220         if (lo->ldo_foreign_lmv == NULL)
2221                 return -ENOMEM;
2222         lo->ldo_foreign_lmv_size = size;
2223         lo->ldo_dir_is_foreign = 1;
2224
2225         return 0;
2226 }
2227
2228 /**
2229  * Declare create striped md object.
2230  *
2231  * The function declares intention to create a striped directory. This is a
2232  * wrapper for lod_prep_md_striped_create(). The only additional functionality
2233  * is to verify pattern \a lum_buf is good. Check that function for the details.
2234  *
2235  * \param[in] env       execution environment
2236  * \param[in] dt        object
2237  * \param[in] attr      attributes to initialize the objects with
2238  * \param[in] lum_buf   a pattern specifying the number of stripes and
2239  *                      MDT to start from
2240  * \param[in] dof       type of objects to be created
2241  * \param[in] th        transaction handle
2242  *
2243  * \retval              0 on success
2244  * \retval              negative if failed
2245  *
2246  */
2247 static int lod_declare_xattr_set_lmv(const struct lu_env *env,
2248                                      struct dt_object *dt,
2249                                      struct lu_attr *attr,
2250                                      const struct lu_buf *lum_buf,
2251                                      struct dt_object_format *dof,
2252                                      struct thandle *th)
2253 {
2254         struct lod_object       *lo = lod_dt_obj(dt);
2255         struct lmv_user_md_v1   *lum = lum_buf->lb_buf;
2256         int                     rc;
2257         ENTRY;
2258
2259         LASSERT(lum != NULL);
2260
2261         CDEBUG(D_INFO, "lum magic = %x count = %u offset = %d\n",
2262                le32_to_cpu(lum->lum_magic), le32_to_cpu(lum->lum_stripe_count),
2263                (int)le32_to_cpu(lum->lum_stripe_offset));
2264
2265         if (lo->ldo_dir_stripe_count == 0) {
2266                 if (lo->ldo_dir_is_foreign) {
2267                         rc = lod_alloc_foreign_lmv(lo, lum_buf->lb_len);
2268                         if (rc != 0)
2269                                 GOTO(out, rc);
2270                         memcpy(lo->ldo_foreign_lmv, lum, lum_buf->lb_len);
2271                         lo->ldo_dir_stripe_loaded = 1;
2272                 }
2273                 GOTO(out, rc = 0);
2274         }
2275
2276         /* prepare dir striped objects */
2277         rc = lod_prep_md_striped_create(env, dt, attr, lum, dof, th);
2278         if (rc != 0) {
2279                 /* failed to create striping, let's reset
2280                  * config so that others don't get confused */
2281                 lod_striping_free(env, lo);
2282                 GOTO(out, rc);
2283         }
2284 out:
2285         RETURN(rc);
2286 }
2287
2288 /**
2289  * Append source stripes after target stripes for migrating directory. NB, we
2290  * only need to declare this, the append is done inside lod_xattr_set_lmv().
2291  *
2292  * \param[in] env       execution environment
2293  * \param[in] dt        target object
2294  * \param[in] buf       LMV buf which contains source stripe fids
2295  * \param[in] th        transaction handle
2296  *
2297  * \retval              0 on success
2298  * \retval              negative if failed
2299  */
2300 static int lod_dir_declare_layout_add(const struct lu_env *env,
2301                                       struct dt_object *dt,
2302                                       const struct lu_buf *buf,
2303                                       struct thandle *th)
2304 {
2305         struct lod_thread_info *info = lod_env_info(env);
2306         struct lod_device *lod = lu2lod_dev(dt->do_lu.lo_dev);
2307         struct lod_tgt_descs *ltd = &lod->lod_mdt_descs;
2308         struct lod_object *lo = lod_dt_obj(dt);
2309         struct dt_object *next = dt_object_child(dt);
2310         struct dt_object_format *dof = &info->lti_format;
2311         struct lmv_mds_md_v1 *lmv = buf->lb_buf;
2312         struct dt_object **stripe;
2313         __u32 stripe_count = le32_to_cpu(lmv->lmv_stripe_count);
2314         struct lu_fid *fid = &info->lti_fid;
2315         struct lod_tgt_desc *tgt;
2316         struct dt_object *dto;
2317         struct dt_device *tgt_dt;
2318         int type = LU_SEQ_RANGE_ANY;
2319         struct dt_insert_rec *rec = &info->lti_dt_rec;
2320         char *stripe_name = info->lti_key;
2321         struct lu_name *sname;
2322         struct linkea_data ldata = { NULL };
2323         struct lu_buf linkea_buf;
2324         __u32 idx;
2325         int i;
2326         int rc;
2327
2328         ENTRY;
2329
2330         if (le32_to_cpu(lmv->lmv_magic) != LMV_MAGIC_V1)
2331                 RETURN(-EINVAL);
2332
2333         if (stripe_count == 0)
2334                 RETURN(-EINVAL);
2335
2336         dof->dof_type = DFT_DIR;
2337
2338         OBD_ALLOC(stripe,
2339                   sizeof(*stripe) * (lo->ldo_dir_stripe_count + stripe_count));
2340         if (stripe == NULL)
2341                 RETURN(-ENOMEM);
2342
2343         for (i = 0; i < lo->ldo_dir_stripe_count; i++)
2344                 stripe[i] = lo->ldo_stripe[i];
2345
2346         for (i = 0; i < stripe_count; i++) {
2347                 fid_le_to_cpu(fid,
2348                         &lmv->lmv_stripe_fids[i]);
2349                 if (!fid_is_sane(fid))
2350                         continue;
2351
2352                 rc = lod_fld_lookup(env, lod, fid, &idx, &type);
2353                 if (rc)
2354                         GOTO(out, rc);
2355
2356                 if (idx == lod2lu_dev(lod)->ld_site->ld_seq_site->ss_node_id) {
2357                         tgt_dt = lod->lod_child;
2358                 } else {
2359                         tgt = LTD_TGT(ltd, idx);
2360                         if (tgt == NULL)
2361                                 GOTO(out, rc = -ESTALE);
2362                         tgt_dt = tgt->ltd_tgt;
2363                 }
2364
2365                 dto = dt_locate_at(env, tgt_dt, fid,
2366                                   lo->ldo_obj.do_lu.lo_dev->ld_site->ls_top_dev,
2367                                   NULL);
2368                 if (IS_ERR(dto))
2369                         GOTO(out, rc = PTR_ERR(dto));
2370
2371                 stripe[i + lo->ldo_dir_stripe_count] = dto;
2372
2373                 if (!dt_try_as_dir(env, dto))
2374                         GOTO(out, rc = -ENOTDIR);
2375
2376                 rc = lod_sub_declare_ref_add(env, dto, th);
2377                 if (rc)
2378                         GOTO(out, rc);
2379
2380                 rc = lod_sub_declare_insert(env, dto,
2381                                             (const struct dt_rec *)rec,
2382                                             (const struct dt_key *)dot, th);
2383                 if (rc)
2384                         GOTO(out, rc);
2385
2386                 rc = lod_sub_declare_insert(env, dto,
2387                                             (const struct dt_rec *)rec,
2388                                             (const struct dt_key *)dotdot, th);
2389                 if (rc)
2390                         GOTO(out, rc);
2391
2392                 rc = lod_sub_declare_xattr_set(env, dto, buf,
2393                                                 XATTR_NAME_LMV, 0, th);
2394                 if (rc)
2395                         GOTO(out, rc);
2396
2397                 snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
2398                          PFID(lu_object_fid(&dto->do_lu)),
2399                          i + lo->ldo_dir_stripe_count);
2400
2401                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
2402                 rc = linkea_links_new(&ldata, &info->lti_linkea_buf,
2403                                       sname, lu_object_fid(&dt->do_lu));
2404                 if (rc)
2405                         GOTO(out, rc);
2406
2407                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
2408                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
2409                 rc = lod_sub_declare_xattr_set(env, dto, &linkea_buf,
2410                                                XATTR_NAME_LINK, 0, th);
2411                 if (rc)
2412                         GOTO(out, rc);
2413
2414                 rc = lod_sub_declare_insert(env, next,
2415                                             (const struct dt_rec *)rec,
2416                                             (const struct dt_key *)stripe_name,
2417                                             th);
2418                 if (rc)
2419                         GOTO(out, rc);
2420
2421                 rc = lod_sub_declare_ref_add(env, next, th);
2422                 if (rc)
2423                         GOTO(out, rc);
2424         }
2425
2426         if (lo->ldo_stripe)
2427                 OBD_FREE(lo->ldo_stripe,
2428                          sizeof(*stripe) * lo->ldo_dir_stripes_allocated);
2429         lo->ldo_stripe = stripe;
2430         lo->ldo_dir_migrate_offset = lo->ldo_dir_stripe_count;
2431         lo->ldo_dir_migrate_hash = le32_to_cpu(lmv->lmv_hash_type);
2432         lo->ldo_dir_stripe_count += stripe_count;
2433         lo->ldo_dir_stripes_allocated += stripe_count;
2434         lo->ldo_dir_hash_type |= LMV_HASH_FLAG_MIGRATION;
2435
2436         RETURN(0);
2437 out:
2438         i = lo->ldo_dir_stripe_count;
2439         while (i < lo->ldo_dir_stripe_count + stripe_count && stripe[i])
2440                 dt_object_put(env, stripe[i++]);
2441
2442         OBD_FREE(stripe,
2443                  sizeof(*stripe) * (stripe_count + lo->ldo_dir_stripe_count));
2444         RETURN(rc);
2445 }
2446
2447 static int lod_dir_declare_layout_delete(const struct lu_env *env,
2448                                          struct dt_object *dt,
2449                                          const struct lu_buf *buf,
2450                                          struct thandle *th)
2451 {
2452         struct lod_thread_info *info = lod_env_info(env);
2453         struct lod_object *lo = lod_dt_obj(dt);
2454         struct dt_object *next = dt_object_child(dt);
2455         struct lmv_user_md *lmu = buf->lb_buf;
2456         __u32 final_stripe_count;
2457         char *stripe_name = info->lti_key;
2458         struct dt_object *dto;
2459         int i;
2460         int rc = 0;
2461
2462         if (!lmu)
2463                 return -EINVAL;
2464
2465         final_stripe_count = le32_to_cpu(lmu->lum_stripe_count);
2466         if (final_stripe_count >= lo->ldo_dir_stripe_count)
2467                 return -EINVAL;
2468
2469         for (i = final_stripe_count; i < lo->ldo_dir_stripe_count; i++) {
2470                 dto = lo->ldo_stripe[i];
2471                 if (!dto)
2472                         continue;
2473
2474                 if (!dt_try_as_dir(env, dto))
2475                         return -ENOTDIR;
2476
2477                 rc = lod_sub_declare_delete(env, dto,
2478                                             (const struct dt_key *)dot, th);
2479                 if (rc)
2480                         return rc;
2481
2482                 rc = lod_sub_declare_ref_del(env, dto, th);
2483                 if (rc)
2484                         return rc;
2485
2486                 rc = lod_sub_declare_delete(env, dto,
2487                                         (const struct dt_key *)dotdot, th);
2488                 if (rc)
2489                         return rc;
2490
2491                 snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
2492                          PFID(lu_object_fid(&dto->do_lu)), i);
2493
2494                 rc = lod_sub_declare_delete(env, next,
2495                                         (const struct dt_key *)stripe_name, th);
2496                 if (rc)
2497                         return rc;
2498
2499                 rc = lod_sub_declare_ref_del(env, next, th);
2500                 if (rc)
2501                         return rc;
2502         }
2503
2504         return 0;
2505 }
2506
2507 /*
2508  * delete stripes from dir master object, the lum_stripe_count in argument is
2509  * the final stripe count, the stripes after that will be deleted, NB, they
2510  * are not destroyed, but deleted from it's parent namespace, this function
2511  * will be called in two places:
2512  * 1. mdd_migrate_create() delete stripes from source, and append them to
2513  *    target.
2514  * 2. mdd_dir_layout_shrink() delete stripes from source, and destroy them.
2515  */
2516 static int lod_dir_layout_delete(const struct lu_env *env,
2517                                  struct dt_object *dt,
2518                                  const struct lu_buf *buf,
2519                                  struct thandle *th)
2520 {
2521         struct lod_thread_info *info = lod_env_info(env);
2522         struct lod_object *lo = lod_dt_obj(dt);
2523         struct dt_object *next = dt_object_child(dt);
2524         struct lmv_user_md *lmu = buf->lb_buf;
2525         __u32 final_stripe_count;
2526         char *stripe_name = info->lti_key;
2527         struct dt_object *dto;
2528         int i;
2529         int rc = 0;
2530
2531         ENTRY;
2532
2533         if (!lmu)
2534                 RETURN(-EINVAL);
2535
2536         final_stripe_count = le32_to_cpu(lmu->lum_stripe_count);
2537         if (final_stripe_count >= lo->ldo_dir_stripe_count)
2538                 RETURN(-EINVAL);
2539
2540         for (i = final_stripe_count; i < lo->ldo_dir_stripe_count; i++) {
2541                 dto = lo->ldo_stripe[i];
2542                 if (!dto)
2543                         continue;
2544
2545                 rc = lod_sub_delete(env, dto,
2546                                     (const struct dt_key *)dotdot, th);
2547                 if (rc)
2548                         break;
2549
2550                 snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
2551                          PFID(lu_object_fid(&dto->do_lu)), i);
2552
2553                 rc = lod_sub_delete(env, next,
2554                                     (const struct dt_key *)stripe_name, th);
2555                 if (rc)
2556                         break;
2557
2558                 rc = lod_sub_ref_del(env, next, th);
2559                 if (rc)
2560                         break;
2561         }
2562
2563         lod_striping_free(env, lod_dt_obj(dt));
2564
2565         RETURN(rc);
2566 }
2567
2568 /**
2569  * Implementation of dt_object_operations::do_declare_xattr_set.
2570  *
2571  * Used with regular (non-striped) objects. Basically it
2572  * initializes the striping information and applies the
2573  * change to all the stripes.
2574  *
2575  * \see dt_object_operations::do_declare_xattr_set() in the API description
2576  * for details.
2577  */
2578 static int lod_dir_declare_xattr_set(const struct lu_env *env,
2579                                      struct dt_object *dt,
2580                                      const struct lu_buf *buf,
2581                                      const char *name, int fl,
2582                                      struct thandle *th)
2583 {
2584         struct dt_object        *next = dt_object_child(dt);
2585         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2586         struct lod_object       *lo = lod_dt_obj(dt);
2587         int                     i;
2588         int                     rc;
2589         ENTRY;
2590
2591         if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
2592                 struct lmv_user_md_v1 *lum;
2593
2594                 LASSERT(buf != NULL && buf->lb_buf != NULL);
2595                 lum = buf->lb_buf;
2596                 rc = lod_verify_md_striping(d, lum);
2597                 if (rc != 0)
2598                         RETURN(rc);
2599         } else if (strcmp(name, XATTR_NAME_LOV) == 0) {
2600                 rc = lod_verify_striping(d, lo, buf, false);
2601                 if (rc != 0)
2602                         RETURN(rc);
2603         }
2604
2605         rc = lod_sub_declare_xattr_set(env, next, buf, name, fl, th);
2606         if (rc != 0)
2607                 RETURN(rc);
2608
2609         /* Note: Do not set LinkEA on sub-stripes, otherwise
2610          * it will confuse the fid2path process(see mdt_path_current()).
2611          * The linkEA between master and sub-stripes is set in
2612          * lod_xattr_set_lmv(). */
2613         if (strcmp(name, XATTR_NAME_LINK) == 0)
2614                 RETURN(0);
2615
2616         /* set xattr to each stripes, if needed */
2617         rc = lod_striping_load(env, lo);
2618         if (rc != 0)
2619                 RETURN(rc);
2620
2621         if (lo->ldo_dir_stripe_count == 0)
2622                 RETURN(0);
2623
2624         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
2625                 if (!lo->ldo_stripe[i])
2626                         continue;
2627
2628                 if (!dt_object_exists(lo->ldo_stripe[i]))
2629                         continue;
2630
2631                 rc = lod_sub_declare_xattr_set(env, lo->ldo_stripe[i],
2632                                                buf, name, fl, th);
2633                 if (rc != 0)
2634                         break;
2635         }
2636
2637         RETURN(rc);
2638 }
2639
2640 static int
2641 lod_obj_stripe_replace_parent_fid_cb(const struct lu_env *env,
2642                                      struct lod_object *lo,
2643                                      struct dt_object *dt, struct thandle *th,
2644                                      int comp_idx, int stripe_idx,
2645                                      struct lod_obj_stripe_cb_data *data)
2646 {
2647         struct lod_thread_info *info = lod_env_info(env);
2648         struct lod_layout_component *comp = &lo->ldo_comp_entries[comp_idx];
2649         struct filter_fid *ff = &info->lti_ff;
2650         struct lu_buf *buf = &info->lti_buf;
2651         int rc;
2652
2653         buf->lb_buf = ff;
2654         buf->lb_len = sizeof(*ff);
2655         rc = dt_xattr_get(env, dt, buf, XATTR_NAME_FID);
2656         if (rc < 0) {
2657                 if (rc == -ENODATA)
2658                         return 0;
2659                 return rc;
2660         }
2661
2662         /*
2663          * locd_buf is set if it's called by dir migration, which doesn't check
2664          * pfid and comp id.
2665          */
2666         if (data->locd_buf) {
2667                 memset(ff, 0, sizeof(*ff));
2668                 ff->ff_parent = *(struct lu_fid *)data->locd_buf->lb_buf;
2669         } else {
2670                 filter_fid_le_to_cpu(ff, ff, sizeof(*ff));
2671
2672                 if (lu_fid_eq(lod_object_fid(lo), &ff->ff_parent) &&
2673                     ff->ff_layout.ol_comp_id == comp->llc_id)
2674                         return 0;
2675
2676                 memset(ff, 0, sizeof(*ff));
2677                 ff->ff_parent = *lu_object_fid(&lo->ldo_obj.do_lu);
2678         }
2679
2680         /* rewrite filter_fid */
2681         ff->ff_parent.f_ver = stripe_idx;
2682         ff->ff_layout.ol_stripe_size = comp->llc_stripe_size;
2683         ff->ff_layout.ol_stripe_count = comp->llc_stripe_count;
2684         ff->ff_layout.ol_comp_id = comp->llc_id;
2685         ff->ff_layout.ol_comp_start = comp->llc_extent.e_start;
2686         ff->ff_layout.ol_comp_end = comp->llc_extent.e_end;
2687         filter_fid_cpu_to_le(ff, ff, sizeof(*ff));
2688
2689         if (data->locd_declare)
2690                 rc = lod_sub_declare_xattr_set(env, dt, buf, XATTR_NAME_FID,
2691                                                LU_XATTR_REPLACE, th);
2692         else
2693                 rc = lod_sub_xattr_set(env, dt, buf, XATTR_NAME_FID,
2694                                        LU_XATTR_REPLACE, th);
2695
2696         return rc;
2697 }
2698
2699 /**
2700  * Reset parent FID on OST object
2701  *
2702  * Replace parent FID with @dt object FID, which is only called during migration
2703  * to reset the parent FID after the MDT object is migrated to the new MDT, i.e.
2704  * the FID is changed.
2705  *
2706  * \param[in] env execution environment
2707  * \param[in] dt dt_object whose stripes's parent FID will be reset
2708  * \parem[in] th thandle
2709  * \param[in] declare if it is declare
2710  *
2711  * \retval      0 if reset succeeds
2712  * \retval      negative errno if reset fails
2713  */
2714 static int lod_replace_parent_fid(const struct lu_env *env,
2715                                   struct dt_object *dt,
2716                                   const struct lu_buf *buf,
2717                                   struct thandle *th, bool declare)
2718 {
2719         struct lod_object *lo = lod_dt_obj(dt);
2720         struct lod_thread_info  *info = lod_env_info(env);
2721         struct filter_fid *ff;
2722         struct lod_obj_stripe_cb_data data = { { 0 } };
2723         int rc;
2724         ENTRY;
2725
2726         LASSERT(S_ISREG(dt->do_lu.lo_header->loh_attr));
2727
2728         /* set xattr to each stripes, if needed */
2729         rc = lod_striping_load(env, lo);
2730         if (rc != 0)
2731                 RETURN(rc);
2732
2733         if (!lod_obj_is_striped(dt))
2734                 RETURN(0);
2735
2736         if (info->lti_ea_store_size < sizeof(*ff)) {
2737                 rc = lod_ea_store_resize(info, sizeof(*ff));
2738                 if (rc != 0)
2739                         RETURN(rc);
2740         }
2741
2742         data.locd_declare = declare;
2743         data.locd_stripe_cb = lod_obj_stripe_replace_parent_fid_cb;
2744         data.locd_buf = buf;
2745         rc = lod_obj_for_each_stripe(env, lo, th, &data);
2746
2747         RETURN(rc);
2748 }
2749
2750 inline __u16 lod_comp_entry_stripe_count(struct lod_object *lo,
2751                                          struct lod_layout_component *entry,
2752                                          bool is_dir)
2753 {
2754         struct lod_device *lod = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
2755
2756         if (is_dir)
2757                 return  0;
2758         else if (lod_comp_inited(entry))
2759                 return entry->llc_stripe_count;
2760         else if ((__u16)-1 == entry->llc_stripe_count)
2761                 return lod->lod_ost_count;
2762         else
2763                 return lod_get_stripe_count(lod, lo,
2764                                             entry->llc_stripe_count, false);
2765 }
2766
2767 static int lod_comp_md_size(struct lod_object *lo, bool is_dir)
2768 {
2769         int magic, size = 0, i;
2770         struct lod_layout_component *comp_entries;
2771         __u16 comp_cnt;
2772         bool is_composite, is_foreign = false;
2773
2774         if (is_dir) {
2775                 comp_cnt = lo->ldo_def_striping->lds_def_comp_cnt;
2776                 comp_entries = lo->ldo_def_striping->lds_def_comp_entries;
2777                 is_composite =
2778                         lo->ldo_def_striping->lds_def_striping_is_composite;
2779         } else {
2780                 comp_cnt = lo->ldo_comp_cnt;
2781                 comp_entries = lo->ldo_comp_entries;
2782                 is_composite = lo->ldo_is_composite;
2783                 is_foreign = lo->ldo_is_foreign;
2784         }
2785
2786         if (is_foreign)
2787                 return lo->ldo_foreign_lov_size;
2788
2789         LASSERT(comp_cnt != 0 && comp_entries != NULL);
2790         if (is_composite) {
2791                 size = sizeof(struct lov_comp_md_v1) +
2792                        sizeof(struct lov_comp_md_entry_v1) * comp_cnt;
2793                 LASSERT(size % sizeof(__u64) == 0);
2794         }
2795
2796         for (i = 0; i < comp_cnt; i++) {
2797                 __u16 stripe_count;
2798
2799                 magic = comp_entries[i].llc_pool ? LOV_MAGIC_V3 : LOV_MAGIC_V1;
2800                 stripe_count = lod_comp_entry_stripe_count(lo, &comp_entries[i],
2801                                                            is_dir);
2802                 if (!is_dir && is_composite)
2803                         lod_comp_shrink_stripe_count(&comp_entries[i],
2804                                                      &stripe_count);
2805
2806                 size += lov_user_md_size(stripe_count, magic);
2807                 LASSERT(size % sizeof(__u64) == 0);
2808         }
2809         return size;
2810 }
2811
2812 /**
2813  * Declare component add. The xattr name is XATTR_LUSTRE_LOV.add, and
2814  * the xattr value is binary lov_comp_md_v1 which contains component(s)
2815  * to be added.
2816   *
2817  * \param[in] env       execution environment
2818  * \param[in] dt        dt_object to add components on
2819  * \param[in] buf       buffer contains components to be added
2820  * \parem[in] th        thandle
2821  *
2822  * \retval      0 on success
2823  * \retval      negative errno on failure
2824  */
2825 static int lod_declare_layout_add(const struct lu_env *env,
2826                                   struct dt_object *dt,
2827                                   const struct lu_buf *buf,
2828                                   struct thandle *th)
2829 {
2830         struct lod_thread_info  *info = lod_env_info(env);
2831         struct lod_layout_component *comp_array, *lod_comp, *old_array;
2832         struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
2833         struct dt_object *next = dt_object_child(dt);
2834         struct lov_desc *desc = &d->lod_ost_descs.ltd_lov_desc;
2835         struct lod_object *lo = lod_dt_obj(dt);
2836         struct lov_user_md_v3 *v3;
2837         struct lov_comp_md_v1 *comp_v1 = buf->lb_buf;
2838         __u32 magic;
2839         int i, rc, array_cnt, old_array_cnt;
2840         ENTRY;
2841
2842         LASSERT(lo->ldo_is_composite);
2843
2844         if (lo->ldo_flr_state != LCM_FL_NONE)
2845                 RETURN(-EBUSY);
2846
2847         rc = lod_verify_striping(d, lo, buf, false);
2848         if (rc != 0)
2849                 RETURN(rc);
2850
2851         magic = comp_v1->lcm_magic;
2852         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
2853                 lustre_swab_lov_comp_md_v1(comp_v1);
2854                 magic = comp_v1->lcm_magic;
2855         }
2856
2857         if (magic != LOV_USER_MAGIC_COMP_V1)
2858                 RETURN(-EINVAL);
2859
2860         array_cnt = lo->ldo_comp_cnt + comp_v1->lcm_entry_count;
2861         OBD_ALLOC(comp_array, sizeof(*comp_array) * array_cnt);
2862         if (comp_array == NULL)
2863                 RETURN(-ENOMEM);
2864
2865         memcpy(comp_array, lo->ldo_comp_entries,
2866                sizeof(*comp_array) * lo->ldo_comp_cnt);
2867
2868         for (i = 0; i < comp_v1->lcm_entry_count; i++) {
2869                 struct lov_user_md_v1 *v1;
2870                 struct lu_extent *ext;
2871
2872                 v1 = (struct lov_user_md *)((char *)comp_v1 +
2873                                 comp_v1->lcm_entries[i].lcme_offset);
2874                 ext = &comp_v1->lcm_entries[i].lcme_extent;
2875
2876                 lod_comp = &comp_array[lo->ldo_comp_cnt + i];
2877                 lod_comp->llc_extent.e_start = ext->e_start;
2878                 lod_comp->llc_extent.e_end = ext->e_end;
2879                 lod_comp->llc_stripe_offset = v1->lmm_stripe_offset;
2880                 lod_comp->llc_flags = comp_v1->lcm_entries[i].lcme_flags;
2881
2882                 lod_comp->llc_stripe_count = v1->lmm_stripe_count;
2883                 lod_comp->llc_stripe_size = v1->lmm_stripe_size;
2884                 lod_adjust_stripe_info(lod_comp, desc, 0);
2885
2886                 if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
2887                         v3 = (struct lov_user_md_v3 *) v1;
2888                         if (v3->lmm_pool_name[0] != '\0') {
2889                                 rc = lod_set_pool(&lod_comp->llc_pool,
2890                                                   v3->lmm_pool_name);
2891                                 if (rc)
2892                                         GOTO(error, rc);
2893                         }
2894                 }
2895         }
2896
2897         old_array = lo->ldo_comp_entries;
2898         old_array_cnt = lo->ldo_comp_cnt;
2899
2900         lo->ldo_comp_entries = comp_array;
2901         lo->ldo_comp_cnt = array_cnt;
2902
2903         /* No need to increase layout generation here, it will be increased
2904          * later when generating component ID for the new components */
2905
2906         info->lti_buf.lb_len = lod_comp_md_size(lo, false);
2907         rc = lod_sub_declare_xattr_set(env, next, &info->lti_buf,
2908                                               XATTR_NAME_LOV, 0, th);
2909         if (rc) {
2910                 lo->ldo_comp_entries = old_array;
2911                 lo->ldo_comp_cnt = old_array_cnt;
2912                 GOTO(error, rc);
2913         }
2914
2915         OBD_FREE(old_array, sizeof(*lod_comp) * old_array_cnt);
2916
2917         LASSERT(lo->ldo_mirror_count == 1);
2918         lo->ldo_mirrors[0].lme_end = array_cnt - 1;
2919
2920         RETURN(0);
2921
2922 error:
2923         for (i = lo->ldo_comp_cnt; i < array_cnt; i++) {
2924                 lod_comp = &comp_array[i];
2925                 if (lod_comp->llc_pool != NULL) {
2926                         OBD_FREE(lod_comp->llc_pool,
2927                                  strlen(lod_comp->llc_pool) + 1);
2928                         lod_comp->llc_pool = NULL;
2929                 }
2930         }
2931         OBD_FREE(comp_array, sizeof(*comp_array) * array_cnt);
2932         RETURN(rc);
2933 }
2934
2935 /**
2936  * lod_last_non_stale_mirror() - Check if a mirror is the last non-stale mirror.
2937  * @mirror_id: Mirror id to be checked.
2938  * @lo:        LOD object.
2939  *
2940  * This function checks if a mirror with specified @mirror_id is the last
2941  * non-stale mirror of a LOD object @lo.
2942  *
2943  * Return: true or false.
2944  */
2945 static inline
2946 bool lod_last_non_stale_mirror(__u16 mirror_id, struct lod_object *lo)
2947 {
2948         struct lod_layout_component *lod_comp;
2949         bool has_stale_flag;
2950         int i;
2951
2952         for (i = 0; i < lo->ldo_mirror_count; i++) {
2953                 if (lo->ldo_mirrors[i].lme_id == mirror_id ||
2954                     lo->ldo_mirrors[i].lme_stale)
2955                         continue;
2956
2957                 has_stale_flag = false;
2958                 lod_foreach_mirror_comp(lod_comp, lo, i) {
2959                         if (lod_comp->llc_flags & LCME_FL_STALE) {
2960                                 has_stale_flag = true;
2961                                 break;
2962                         }
2963                 }
2964                 if (!has_stale_flag)
2965                         return false;
2966         }
2967
2968         return true;
2969 }
2970
2971 /**
2972  * Declare component set. The xattr is name XATTR_LUSTRE_LOV.set.$field,
2973  * the '$field' can only be 'flags' now. The xattr value is binary
2974  * lov_comp_md_v1 which contains the component ID(s) and the value of
2975  * the field to be modified.
2976  *
2977  * \param[in] env       execution environment
2978  * \param[in] dt        dt_object to be modified
2979  * \param[in] op        operation string, like "set.flags"
2980  * \param[in] buf       buffer contains components to be set
2981  * \parem[in] th        thandle
2982  *
2983  * \retval      0 on success
2984  * \retval      negative errno on failure
2985  */
2986 static int lod_declare_layout_set(const struct lu_env *env,
2987                                   struct dt_object *dt,
2988                                   char *op, const struct lu_buf *buf,
2989                                   struct thandle *th)
2990 {
2991         struct lod_layout_component     *lod_comp;
2992         struct lod_thread_info  *info = lod_env_info(env);
2993         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2994         struct lod_object       *lo = lod_dt_obj(dt);
2995         struct lov_comp_md_v1   *comp_v1 = buf->lb_buf;
2996         __u32   magic;
2997         int     i, j, rc;
2998         bool    changed = false;
2999         ENTRY;
3000
3001         if (strcmp(op, "set.flags") != 0) {
3002                 CDEBUG(D_LAYOUT, "%s: operation (%s) not supported.\n",
3003                        lod2obd(d)->obd_name, op);
3004                 RETURN(-ENOTSUPP);
3005         }
3006
3007         magic = comp_v1->lcm_magic;
3008         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
3009                 lustre_swab_lov_comp_md_v1(comp_v1);
3010                 magic = comp_v1->lcm_magic;
3011         }
3012
3013         if (magic != LOV_USER_MAGIC_COMP_V1)
3014                 RETURN(-EINVAL);
3015
3016         if (comp_v1->lcm_entry_count == 0) {
3017                 CDEBUG(D_LAYOUT, "%s: entry count is zero.\n",
3018                        lod2obd(d)->obd_name);
3019                 RETURN(-EINVAL);
3020         }
3021
3022         for (i = 0; i < comp_v1->lcm_entry_count; i++) {
3023                 __u32 id = comp_v1->lcm_entries[i].lcme_id;
3024                 __u32 flags = comp_v1->lcm_entries[i].lcme_flags;
3025                 __u32 mirror_flag = flags & LCME_MIRROR_FLAGS;
3026                 __u16 mirror_id = mirror_id_of(id);
3027                 bool neg = flags & LCME_FL_NEG;
3028
3029                 if (flags & LCME_FL_INIT) {
3030                         if (changed)
3031                                 lod_striping_free(env, lo);
3032                         RETURN(-EINVAL);
3033                 }
3034
3035                 flags &= ~(LCME_MIRROR_FLAGS | LCME_FL_NEG);
3036                 for (j = 0; j < lo->ldo_comp_cnt; j++) {
3037                         lod_comp = &lo->ldo_comp_entries[j];
3038
3039                         /* lfs only put one flag in each entry */
3040                         if ((flags && id != lod_comp->llc_id) ||
3041                             (mirror_flag && mirror_id !=
3042                                             mirror_id_of(lod_comp->llc_id)))
3043                                 continue;
3044
3045                         if (neg) {
3046                                 if (flags)
3047                                         lod_comp->llc_flags &= ~flags;
3048                                 if (mirror_flag)
3049                                         lod_comp->llc_flags &= ~mirror_flag;
3050                         } else {
3051                                 if (flags) {
3052                                         if ((flags & LCME_FL_STALE) &&
3053                                             lod_last_non_stale_mirror(mirror_id,
3054                                                                       lo))
3055                                                 RETURN(-EUCLEAN);
3056                                         lod_comp->llc_flags |= flags;
3057                                 }
3058                                 if (mirror_flag) {
3059                                         lod_comp->llc_flags |= mirror_flag;
3060                                         if (mirror_flag & LCME_FL_NOSYNC)
3061                                                 lod_comp->llc_timestamp =
3062                                                        ktime_get_real_seconds();
3063                                 }
3064                         }
3065                         changed = true;
3066                 }
3067         }
3068
3069         if (!changed) {
3070                 CDEBUG(D_LAYOUT, "%s: requested component(s) not found.\n",
3071                        lod2obd(d)->obd_name);
3072                 RETURN(-EINVAL);
3073         }
3074
3075         lod_obj_inc_layout_gen(lo);
3076
3077         info->lti_buf.lb_len = lod_comp_md_size(lo, false);
3078         rc = lod_sub_declare_xattr_set(env, dt_object_child(dt), &info->lti_buf,
3079                                        XATTR_NAME_LOV, LU_XATTR_REPLACE, th);
3080         RETURN(rc);
3081 }
3082
3083 /**
3084  * Declare component deletion. The xattr name is XATTR_LUSTRE_LOV.del,
3085  * and the xattr value is a unique component ID or a special lcme_id.
3086  *
3087  * \param[in] env       execution environment
3088  * \param[in] dt        dt_object to be operated on
3089  * \param[in] buf       buffer contains component ID or lcme_id
3090  * \parem[in] th        thandle
3091  *
3092  * \retval      0 on success
3093  * \retval      negative errno on failure
3094  */
3095 static int lod_declare_layout_del(const struct lu_env *env,
3096                                   struct dt_object *dt,
3097                                   const struct lu_buf *buf,
3098                                   struct thandle *th)
3099 {
3100         struct lod_thread_info  *info = lod_env_info(env);
3101         struct dt_object *next = dt_object_child(dt);
3102         struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
3103         struct lod_object *lo = lod_dt_obj(dt);
3104         struct lu_attr *attr = &lod_env_info(env)->lti_attr;
3105         struct lov_comp_md_v1 *comp_v1 = buf->lb_buf;
3106         __u32 magic, id, flags, neg_flags = 0;
3107         int rc, i, j, left;
3108         ENTRY;
3109
3110         LASSERT(lo->ldo_is_composite);
3111
3112         if (lo->ldo_flr_state != LCM_FL_NONE)
3113                 RETURN(-EBUSY);
3114
3115         magic = comp_v1->lcm_magic;
3116         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
3117                 lustre_swab_lov_comp_md_v1(comp_v1);
3118                 magic = comp_v1->lcm_magic;
3119         }
3120
3121         if (magic != LOV_USER_MAGIC_COMP_V1)
3122                 RETURN(-EINVAL);
3123
3124         id = comp_v1->lcm_entries[0].lcme_id;
3125         flags = comp_v1->lcm_entries[0].lcme_flags;
3126
3127         if (id > LCME_ID_MAX || (flags & ~LCME_KNOWN_FLAGS)) {
3128                 CDEBUG(D_LAYOUT, "%s: invalid component id %#x, flags %#x\n",
3129                        lod2obd(d)->obd_name, id, flags);
3130                 RETURN(-EINVAL);
3131         }
3132
3133         if (id != LCME_ID_INVAL && flags != 0) {
3134                 CDEBUG(D_LAYOUT, "%s: specified both id and flags.\n",
3135                        lod2obd(d)->obd_name);
3136                 RETURN(-EINVAL);
3137         }
3138
3139         if (id == LCME_ID_INVAL && !flags) {
3140                 CDEBUG(D_LAYOUT, "%s: no id or flags specified.\n",
3141                        lod2obd(d)->obd_name);
3142                 RETURN(-EINVAL);
3143         }
3144
3145         if (flags & LCME_FL_NEG) {
3146                 neg_flags = flags & ~LCME_FL_NEG;
3147                 flags = 0;
3148         }
3149
3150         left = lo->ldo_comp_cnt;
3151         if (left <= 0)
3152                 RETURN(-EINVAL);
3153
3154         for (i = (lo->ldo_comp_cnt - 1); i >= 0; i--) {
3155                 struct lod_layout_component *lod_comp;
3156
3157                 lod_comp = &lo->ldo_comp_entries[i];
3158
3159                 if (id != LCME_ID_INVAL && id != lod_comp->llc_id)
3160                         continue;
3161                 else if (flags && !(flags & lod_comp->llc_flags))
3162                         continue;
3163                 else if (neg_flags && (neg_flags & lod_comp->llc_flags))
3164                         continue;
3165
3166                 if (left != (i + 1)) {
3167                         CDEBUG(D_LAYOUT, "%s: this deletion will create "
3168                                "a hole.\n", lod2obd(d)->obd_name);
3169                         RETURN(-EINVAL);
3170                 }
3171                 left--;
3172
3173                 /* Mark the component as deleted */
3174                 lod_comp->llc_id = LCME_ID_INVAL;
3175
3176                 /* Not instantiated component */
3177                 if (lod_comp->llc_stripe == NULL)
3178                         continue;
3179
3180                 LASSERT(lod_comp->llc_stripe_count > 0);
3181                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
3182                         struct dt_object *obj = lod_comp->llc_stripe[j];
3183
3184                         if (obj == NULL)
3185                                 continue;
3186                         rc = lod_sub_declare_destroy(env, obj, th);
3187                         if (rc)
3188                                 RETURN(rc);
3189                 }
3190         }
3191
3192         LASSERTF(left >= 0, "left = %d\n", left);
3193         if (left == lo->ldo_comp_cnt) {
3194                 CDEBUG(D_LAYOUT, "%s: requested component id:%#x not found\n",
3195                        lod2obd(d)->obd_name, id);
3196                 RETURN(-EINVAL);
3197         }
3198
3199         memset(attr, 0, sizeof(*attr));
3200         attr->la_valid = LA_SIZE;
3201         rc = lod_sub_declare_attr_set(env, next, attr, th);
3202         if (rc)
3203                 RETURN(rc);
3204
3205         if (left > 0) {
3206                 info->lti_buf.lb_len = lod_comp_md_size(lo, false);
3207                 rc = lod_sub_declare_xattr_set(env, next, &info->lti_buf,
3208                                                XATTR_NAME_LOV, 0, th);
3209         } else {
3210