Whamcloud - gitweb
LU-9771 flr: Send write intent RPC to mdt
[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, 2016, 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 <obd.h>
44 #include <obd_class.h>
45 #include <obd_support.h>
46
47 #include <lustre_fid.h>
48 #include <lustre_linkea.h>
49 #include <lustre_lmv.h>
50 #include <uapi/linux/lustre/lustre_param.h>
51 #include <lustre_swab.h>
52 #include <uapi/linux/lustre/lustre_ver.h>
53 #include <lprocfs_status.h>
54 #include <md_object.h>
55
56 #include "lod_internal.h"
57
58 static const char dot[] = ".";
59 static const char dotdot[] = "..";
60
61 /**
62  * Implementation of dt_index_operations::dio_lookup
63  *
64  * Used with regular (non-striped) objects.
65  *
66  * \see dt_index_operations::dio_lookup() in the API description for details.
67  */
68 static int lod_lookup(const struct lu_env *env, struct dt_object *dt,
69                       struct dt_rec *rec, const struct dt_key *key)
70 {
71         struct dt_object *next = dt_object_child(dt);
72         return next->do_index_ops->dio_lookup(env, next, rec, key);
73 }
74
75 /**
76  * Implementation of dt_index_operations::dio_declare_insert.
77  *
78  * Used with regular (non-striped) objects.
79  *
80  * \see dt_index_operations::dio_declare_insert() in the API description
81  * for details.
82  */
83 static int lod_declare_insert(const struct lu_env *env, struct dt_object *dt,
84                               const struct dt_rec *rec,
85                               const struct dt_key *key, struct thandle *th)
86 {
87         return lod_sub_declare_insert(env, dt_object_child(dt), rec, key, th);
88 }
89
90 /**
91  * Implementation of dt_index_operations::dio_insert.
92  *
93  * Used with regular (non-striped) objects
94  *
95  * \see dt_index_operations::dio_insert() in the API description for details.
96  */
97 static int lod_insert(const struct lu_env *env, struct dt_object *dt,
98                       const struct dt_rec *rec, const struct dt_key *key,
99                       struct thandle *th, int ign)
100 {
101         return lod_sub_insert(env, dt_object_child(dt), rec, key, th, ign);
102 }
103
104 /**
105  * Implementation of dt_index_operations::dio_declare_delete.
106  *
107  * Used with regular (non-striped) objects.
108  *
109  * \see dt_index_operations::dio_declare_delete() in the API description
110  * for details.
111  */
112 static int lod_declare_delete(const struct lu_env *env, struct dt_object *dt,
113                               const struct dt_key *key, struct thandle *th)
114 {
115         return lod_sub_declare_delete(env, dt_object_child(dt), key, th);
116 }
117
118 /**
119  * Implementation of dt_index_operations::dio_delete.
120  *
121  * Used with regular (non-striped) objects.
122  *
123  * \see dt_index_operations::dio_delete() in the API description for details.
124  */
125 static int lod_delete(const struct lu_env *env, struct dt_object *dt,
126                       const struct dt_key *key, struct thandle *th)
127 {
128         return lod_sub_delete(env, dt_object_child(dt), key, th);
129 }
130
131 /**
132  * Implementation of dt_it_ops::init.
133  *
134  * Used with regular (non-striped) objects.
135  *
136  * \see dt_it_ops::init() in the API description for details.
137  */
138 static struct dt_it *lod_it_init(const struct lu_env *env,
139                                  struct dt_object *dt, __u32 attr)
140 {
141         struct dt_object        *next = dt_object_child(dt);
142         struct lod_it           *it = &lod_env_info(env)->lti_it;
143         struct dt_it            *it_next;
144
145         it_next = next->do_index_ops->dio_it.init(env, next, attr);
146         if (IS_ERR(it_next))
147                 return it_next;
148
149         /* currently we do not use more than one iterator per thread
150          * so we store it in thread info. if at some point we need
151          * more active iterators in a single thread, we can allocate
152          * additional ones */
153         LASSERT(it->lit_obj == NULL);
154
155         it->lit_it = it_next;
156         it->lit_obj = next;
157
158         return (struct dt_it *)it;
159 }
160
161 #define LOD_CHECK_IT(env, it)                                   \
162 do {                                                            \
163         LASSERT((it)->lit_obj != NULL);                         \
164         LASSERT((it)->lit_it != NULL);                          \
165 } while (0)
166
167 /**
168  * Implementation of dt_index_operations::dio_it.fini.
169  *
170  * Used with regular (non-striped) objects.
171  *
172  * \see dt_index_operations::dio_it.fini() in the API description for details.
173  */
174 static void lod_it_fini(const struct lu_env *env, struct dt_it *di)
175 {
176         struct lod_it *it = (struct lod_it *)di;
177
178         LOD_CHECK_IT(env, it);
179         it->lit_obj->do_index_ops->dio_it.fini(env, it->lit_it);
180
181         /* the iterator not in use any more */
182         it->lit_obj = NULL;
183         it->lit_it = NULL;
184 }
185
186 /**
187  * Implementation of dt_it_ops::get.
188  *
189  * Used with regular (non-striped) objects.
190  *
191  * \see dt_it_ops::get() in the API description for details.
192  */
193 static int lod_it_get(const struct lu_env *env, struct dt_it *di,
194                       const struct dt_key *key)
195 {
196         const struct lod_it *it = (const struct lod_it *)di;
197
198         LOD_CHECK_IT(env, it);
199         return it->lit_obj->do_index_ops->dio_it.get(env, it->lit_it, key);
200 }
201
202 /**
203  * Implementation of dt_it_ops::put.
204  *
205  * Used with regular (non-striped) objects.
206  *
207  * \see dt_it_ops::put() in the API description for details.
208  */
209 static void lod_it_put(const struct lu_env *env, struct dt_it *di)
210 {
211         struct lod_it *it = (struct lod_it *)di;
212
213         LOD_CHECK_IT(env, it);
214         return it->lit_obj->do_index_ops->dio_it.put(env, it->lit_it);
215 }
216
217 /**
218  * Implementation of dt_it_ops::next.
219  *
220  * Used with regular (non-striped) objects
221  *
222  * \see dt_it_ops::next() in the API description for details.
223  */
224 static int lod_it_next(const struct lu_env *env, struct dt_it *di)
225 {
226         struct lod_it *it = (struct lod_it *)di;
227
228         LOD_CHECK_IT(env, it);
229         return it->lit_obj->do_index_ops->dio_it.next(env, it->lit_it);
230 }
231
232 /**
233  * Implementation of dt_it_ops::key.
234  *
235  * Used with regular (non-striped) objects.
236  *
237  * \see dt_it_ops::key() in the API description for details.
238  */
239 static struct dt_key *lod_it_key(const struct lu_env *env,
240                                  const struct dt_it *di)
241 {
242         const struct lod_it *it = (const struct lod_it *)di;
243
244         LOD_CHECK_IT(env, it);
245         return it->lit_obj->do_index_ops->dio_it.key(env, it->lit_it);
246 }
247
248 /**
249  * Implementation of dt_it_ops::key_size.
250  *
251  * Used with regular (non-striped) objects.
252  *
253  * \see dt_it_ops::key_size() in the API description for details.
254  */
255 static int lod_it_key_size(const struct lu_env *env, const struct dt_it *di)
256 {
257         struct lod_it *it = (struct lod_it *)di;
258
259         LOD_CHECK_IT(env, it);
260         return it->lit_obj->do_index_ops->dio_it.key_size(env, it->lit_it);
261 }
262
263 /**
264  * Implementation of dt_it_ops::rec.
265  *
266  * Used with regular (non-striped) objects.
267  *
268  * \see dt_it_ops::rec() in the API description for details.
269  */
270 static int lod_it_rec(const struct lu_env *env, const struct dt_it *di,
271                       struct dt_rec *rec, __u32 attr)
272 {
273         const struct lod_it *it = (const struct lod_it *)di;
274
275         LOD_CHECK_IT(env, it);
276         return it->lit_obj->do_index_ops->dio_it.rec(env, it->lit_it, rec,
277                                                      attr);
278 }
279
280 /**
281  * Implementation of dt_it_ops::rec_size.
282  *
283  * Used with regular (non-striped) objects.
284  *
285  * \see dt_it_ops::rec_size() in the API description for details.
286  */
287 static int lod_it_rec_size(const struct lu_env *env, const struct dt_it *di,
288                            __u32 attr)
289 {
290         const struct lod_it *it = (const struct lod_it *)di;
291
292         LOD_CHECK_IT(env, it);
293         return it->lit_obj->do_index_ops->dio_it.rec_size(env, it->lit_it,
294                                                           attr);
295 }
296
297 /**
298  * Implementation of dt_it_ops::store.
299  *
300  * Used with regular (non-striped) objects.
301  *
302  * \see dt_it_ops::store() in the API description for details.
303  */
304 static __u64 lod_it_store(const struct lu_env *env, const struct dt_it *di)
305 {
306         const struct lod_it *it = (const struct lod_it *)di;
307
308         LOD_CHECK_IT(env, it);
309         return it->lit_obj->do_index_ops->dio_it.store(env, it->lit_it);
310 }
311
312 /**
313  * Implementation of dt_it_ops::load.
314  *
315  * Used with regular (non-striped) objects.
316  *
317  * \see dt_it_ops::load() in the API description for details.
318  */
319 static int lod_it_load(const struct lu_env *env, const struct dt_it *di,
320                        __u64 hash)
321 {
322         const struct lod_it *it = (const struct lod_it *)di;
323
324         LOD_CHECK_IT(env, it);
325         return it->lit_obj->do_index_ops->dio_it.load(env, it->lit_it, hash);
326 }
327
328 /**
329  * Implementation of dt_it_ops::key_rec.
330  *
331  * Used with regular (non-striped) objects.
332  *
333  * \see dt_it_ops::rec() in the API description for details.
334  */
335 static int lod_it_key_rec(const struct lu_env *env, const struct dt_it *di,
336                           void *key_rec)
337 {
338         const struct lod_it *it = (const struct lod_it *)di;
339
340         LOD_CHECK_IT(env, it);
341         return it->lit_obj->do_index_ops->dio_it.key_rec(env, it->lit_it,
342                                                          key_rec);
343 }
344
345 static struct dt_index_operations lod_index_ops = {
346         .dio_lookup             = lod_lookup,
347         .dio_declare_insert     = lod_declare_insert,
348         .dio_insert             = lod_insert,
349         .dio_declare_delete     = lod_declare_delete,
350         .dio_delete             = lod_delete,
351         .dio_it = {
352                 .init           = lod_it_init,
353                 .fini           = lod_it_fini,
354                 .get            = lod_it_get,
355                 .put            = lod_it_put,
356                 .next           = lod_it_next,
357                 .key            = lod_it_key,
358                 .key_size       = lod_it_key_size,
359                 .rec            = lod_it_rec,
360                 .rec_size       = lod_it_rec_size,
361                 .store          = lod_it_store,
362                 .load           = lod_it_load,
363                 .key_rec        = lod_it_key_rec,
364         }
365 };
366
367 /**
368  * Implementation of dt_it_ops::init.
369  *
370  * Used with striped objects. Internally just initializes the iterator
371  * on the first stripe.
372  *
373  * \see dt_it_ops::init() in the API description for details.
374  */
375 static struct dt_it *lod_striped_it_init(const struct lu_env *env,
376                                          struct dt_object *dt, __u32 attr)
377 {
378         struct lod_object       *lo = lod_dt_obj(dt);
379         struct dt_object        *next;
380         struct lod_it           *it = &lod_env_info(env)->lti_it;
381         struct dt_it            *it_next;
382         ENTRY;
383
384         LASSERT(lo->ldo_dir_stripe_count > 0);
385         next = lo->ldo_stripe[0];
386         LASSERT(next != NULL);
387         LASSERT(next->do_index_ops != NULL);
388
389         it_next = next->do_index_ops->dio_it.init(env, next, attr);
390         if (IS_ERR(it_next))
391                 return it_next;
392
393         /* currently we do not use more than one iterator per thread
394          * so we store it in thread info. if at some point we need
395          * more active iterators in a single thread, we can allocate
396          * additional ones */
397         LASSERT(it->lit_obj == NULL);
398
399         it->lit_stripe_index = 0;
400         it->lit_attr = attr;
401         it->lit_it = it_next;
402         it->lit_obj = dt;
403
404         return (struct dt_it *)it;
405 }
406
407 #define LOD_CHECK_STRIPED_IT(env, it, lo)                               \
408 do {                                                                    \
409         LASSERT((it)->lit_obj != NULL);                                 \
410         LASSERT((it)->lit_it != NULL);                                  \
411         LASSERT((lo)->ldo_dir_stripe_count > 0);                        \
412         LASSERT((it)->lit_stripe_index < (lo)->ldo_dir_stripe_count);   \
413 } while (0)
414
415 /**
416  * Implementation of dt_it_ops::fini.
417  *
418  * Used with striped objects.
419  *
420  * \see dt_it_ops::fini() in the API description for details.
421  */
422 static void lod_striped_it_fini(const struct lu_env *env, struct dt_it *di)
423 {
424         struct lod_it           *it = (struct lod_it *)di;
425         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
426         struct dt_object        *next;
427
428         /* If lit_it == NULL, then it means the sub_it has been finished,
429          * which only happens in failure cases, see lod_striped_it_next() */
430         if (it->lit_it != NULL) {
431                 LOD_CHECK_STRIPED_IT(env, it, lo);
432
433                 next = lo->ldo_stripe[it->lit_stripe_index];
434                 LASSERT(next != NULL);
435                 LASSERT(next->do_index_ops != NULL);
436
437                 next->do_index_ops->dio_it.fini(env, it->lit_it);
438         }
439
440         /* the iterator not in use any more */
441         it->lit_obj = NULL;
442         it->lit_it = NULL;
443         it->lit_stripe_index = 0;
444 }
445
446 /**
447  * Implementation of dt_it_ops::get.
448  *
449  * Right now it's not used widely, only to reset the iterator to the
450  * initial position. It should be possible to implement a full version
451  * which chooses a correct stripe to be able to position with any key.
452  *
453  * \see dt_it_ops::get() in the API description for details.
454  */
455 static int lod_striped_it_get(const struct lu_env *env, struct dt_it *di,
456                               const struct dt_key *key)
457 {
458         const struct lod_it     *it = (const struct lod_it *)di;
459         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
460         struct dt_object        *next;
461         ENTRY;
462
463         LOD_CHECK_STRIPED_IT(env, it, lo);
464
465         next = lo->ldo_stripe[it->lit_stripe_index];
466         LASSERT(next != NULL);
467         LASSERT(next->do_index_ops != NULL);
468
469         return next->do_index_ops->dio_it.get(env, it->lit_it, key);
470 }
471
472 /**
473  * Implementation of dt_it_ops::put.
474  *
475  * Used with striped objects.
476  *
477  * \see dt_it_ops::put() in the API description for details.
478  */
479 static void lod_striped_it_put(const struct lu_env *env, struct dt_it *di)
480 {
481         struct lod_it           *it = (struct lod_it *)di;
482         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
483         struct dt_object        *next;
484
485         LOD_CHECK_STRIPED_IT(env, it, lo);
486
487         next = lo->ldo_stripe[it->lit_stripe_index];
488         LASSERT(next != NULL);
489         LASSERT(next->do_index_ops != NULL);
490
491         return next->do_index_ops->dio_it.put(env, it->lit_it);
492 }
493
494 /**
495  * Implementation of dt_it_ops::next.
496  *
497  * Used with striped objects. When the end of the current stripe is
498  * reached, the method takes the next stripe's iterator.
499  *
500  * \see dt_it_ops::next() in the API description for details.
501  */
502 static int lod_striped_it_next(const struct lu_env *env, struct dt_it *di)
503 {
504         struct lod_it           *it = (struct lod_it *)di;
505         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
506         struct dt_object        *next;
507         struct dt_it            *it_next;
508         int                     rc;
509         ENTRY;
510
511         LOD_CHECK_STRIPED_IT(env, it, lo);
512
513         next = lo->ldo_stripe[it->lit_stripe_index];
514         LASSERT(next != NULL);
515         LASSERT(next->do_index_ops != NULL);
516 again:
517         rc = next->do_index_ops->dio_it.next(env, it->lit_it);
518         if (rc < 0)
519                 RETURN(rc);
520
521         if (rc == 0 && it->lit_stripe_index == 0)
522                 RETURN(rc);
523
524         if (rc == 0 && it->lit_stripe_index > 0) {
525                 struct lu_dirent *ent;
526
527                 ent = (struct lu_dirent *)lod_env_info(env)->lti_key;
528
529                 rc = next->do_index_ops->dio_it.rec(env, it->lit_it,
530                                                     (struct dt_rec *)ent,
531                                                     it->lit_attr);
532                 if (rc != 0)
533                         RETURN(rc);
534
535                 /* skip . and .. for slave stripe */
536                 if ((strncmp(ent->lde_name, ".",
537                              le16_to_cpu(ent->lde_namelen)) == 0 &&
538                      le16_to_cpu(ent->lde_namelen) == 1) ||
539                     (strncmp(ent->lde_name, "..",
540                              le16_to_cpu(ent->lde_namelen)) == 0 &&
541                      le16_to_cpu(ent->lde_namelen) == 2))
542                         goto again;
543
544                 RETURN(rc);
545         }
546
547         /* go to next stripe */
548         if (it->lit_stripe_index + 1 >= lo->ldo_dir_stripe_count)
549                 RETURN(1);
550
551         it->lit_stripe_index++;
552
553         next->do_index_ops->dio_it.put(env, it->lit_it);
554         next->do_index_ops->dio_it.fini(env, it->lit_it);
555         it->lit_it = NULL;
556
557         next = lo->ldo_stripe[it->lit_stripe_index];
558         LASSERT(next != NULL);
559         rc = next->do_ops->do_index_try(env, next, &dt_directory_features);
560         if (rc != 0)
561                 RETURN(rc);
562
563         LASSERT(next->do_index_ops != NULL);
564
565         it_next = next->do_index_ops->dio_it.init(env, next, it->lit_attr);
566         if (!IS_ERR(it_next)) {
567                 it->lit_it = it_next;
568                 goto again;
569         } else {
570                 rc = PTR_ERR(it_next);
571         }
572
573         RETURN(rc);
574 }
575
576 /**
577  * Implementation of dt_it_ops::key.
578  *
579  * Used with striped objects.
580  *
581  * \see dt_it_ops::key() in the API description for details.
582  */
583 static struct dt_key *lod_striped_it_key(const struct lu_env *env,
584                                          const struct dt_it *di)
585 {
586         const struct lod_it     *it = (const struct lod_it *)di;
587         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
588         struct dt_object        *next;
589
590         LOD_CHECK_STRIPED_IT(env, it, lo);
591
592         next = lo->ldo_stripe[it->lit_stripe_index];
593         LASSERT(next != NULL);
594         LASSERT(next->do_index_ops != NULL);
595
596         return next->do_index_ops->dio_it.key(env, it->lit_it);
597 }
598
599 /**
600  * Implementation of dt_it_ops::key_size.
601  *
602  * Used with striped objects.
603  *
604  * \see dt_it_ops::size() in the API description for details.
605  */
606 static int lod_striped_it_key_size(const struct lu_env *env,
607                                    const struct dt_it *di)
608 {
609         struct lod_it           *it = (struct lod_it *)di;
610         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
611         struct dt_object        *next;
612
613         LOD_CHECK_STRIPED_IT(env, it, lo);
614
615         next = lo->ldo_stripe[it->lit_stripe_index];
616         LASSERT(next != NULL);
617         LASSERT(next->do_index_ops != NULL);
618
619         return next->do_index_ops->dio_it.key_size(env, it->lit_it);
620 }
621
622 /**
623  * Implementation of dt_it_ops::rec.
624  *
625  * Used with striped objects.
626  *
627  * \see dt_it_ops::rec() in the API description for details.
628  */
629 static int lod_striped_it_rec(const struct lu_env *env, const struct dt_it *di,
630                               struct dt_rec *rec, __u32 attr)
631 {
632         const struct lod_it     *it = (const struct lod_it *)di;
633         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
634         struct dt_object        *next;
635
636         LOD_CHECK_STRIPED_IT(env, it, lo);
637
638         next = lo->ldo_stripe[it->lit_stripe_index];
639         LASSERT(next != NULL);
640         LASSERT(next->do_index_ops != NULL);
641
642         return next->do_index_ops->dio_it.rec(env, it->lit_it, rec, attr);
643 }
644
645 /**
646  * Implementation of dt_it_ops::rec_size.
647  *
648  * Used with striped objects.
649  *
650  * \see dt_it_ops::rec_size() in the API description for details.
651  */
652 static int lod_striped_it_rec_size(const struct lu_env *env,
653                                    const struct dt_it *di, __u32 attr)
654 {
655         struct lod_it           *it = (struct lod_it *)di;
656         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
657         struct dt_object        *next;
658
659         LOD_CHECK_STRIPED_IT(env, it, lo);
660
661         next = lo->ldo_stripe[it->lit_stripe_index];
662         LASSERT(next != NULL);
663         LASSERT(next->do_index_ops != NULL);
664
665         return next->do_index_ops->dio_it.rec_size(env, it->lit_it, attr);
666 }
667
668 /**
669  * Implementation of dt_it_ops::store.
670  *
671  * Used with striped objects.
672  *
673  * \see dt_it_ops::store() in the API description for details.
674  */
675 static __u64 lod_striped_it_store(const struct lu_env *env,
676                                   const struct dt_it *di)
677 {
678         const struct lod_it     *it = (const struct lod_it *)di;
679         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
680         struct dt_object        *next;
681
682         LOD_CHECK_STRIPED_IT(env, it, lo);
683
684         next = lo->ldo_stripe[it->lit_stripe_index];
685         LASSERT(next != NULL);
686         LASSERT(next->do_index_ops != NULL);
687
688         return next->do_index_ops->dio_it.store(env, it->lit_it);
689 }
690
691 /**
692  * Implementation of dt_it_ops::load.
693  *
694  * Used with striped objects.
695  *
696  * \see dt_it_ops::load() in the API description for details.
697  */
698 static int lod_striped_it_load(const struct lu_env *env,
699                                const struct dt_it *di, __u64 hash)
700 {
701         const struct lod_it     *it = (const struct lod_it *)di;
702         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
703         struct dt_object        *next;
704
705         LOD_CHECK_STRIPED_IT(env, it, lo);
706
707         next = lo->ldo_stripe[it->lit_stripe_index];
708         LASSERT(next != NULL);
709         LASSERT(next->do_index_ops != NULL);
710
711         return next->do_index_ops->dio_it.load(env, it->lit_it, hash);
712 }
713
714 static struct dt_index_operations lod_striped_index_ops = {
715         .dio_lookup             = lod_lookup,
716         .dio_declare_insert     = lod_declare_insert,
717         .dio_insert             = lod_insert,
718         .dio_declare_delete     = lod_declare_delete,
719         .dio_delete             = lod_delete,
720         .dio_it = {
721                 .init           = lod_striped_it_init,
722                 .fini           = lod_striped_it_fini,
723                 .get            = lod_striped_it_get,
724                 .put            = lod_striped_it_put,
725                 .next           = lod_striped_it_next,
726                 .key            = lod_striped_it_key,
727                 .key_size       = lod_striped_it_key_size,
728                 .rec            = lod_striped_it_rec,
729                 .rec_size       = lod_striped_it_rec_size,
730                 .store          = lod_striped_it_store,
731                 .load           = lod_striped_it_load,
732         }
733 };
734
735 /**
736  * Append the FID for each shard of the striped directory after the
737  * given LMV EA header.
738  *
739  * To simplify striped directory and the consistency verification,
740  * we only store the LMV EA header on disk, for both master object
741  * and slave objects. When someone wants to know the whole LMV EA,
742  * such as client readdir(), we can build the entrie LMV EA on the
743  * MDT side (in RAM) via iterating the sub-directory entries that
744  * are contained in the master object of the stripe directory.
745  *
746  * For the master object of the striped directroy, the valid name
747  * for each shard is composed of the ${shard_FID}:${shard_idx}.
748  *
749  * There may be holes in the LMV EA if some shards' name entries
750  * are corrupted or lost.
751  *
752  * \param[in] env       pointer to the thread context
753  * \param[in] lo        pointer to the master object of the striped directory
754  * \param[in] buf       pointer to the lu_buf which will hold the LMV EA
755  * \param[in] resize    whether re-allocate the buffer if it is not big enough
756  *
757  * \retval              positive size of the LMV EA
758  * \retval              0 for nothing to be loaded
759  * \retval              negative error number on failure
760  */
761 int lod_load_lmv_shards(const struct lu_env *env, struct lod_object *lo,
762                         struct lu_buf *buf, bool resize)
763 {
764         struct lu_dirent        *ent    =
765                         (struct lu_dirent *)lod_env_info(env)->lti_key;
766         struct lod_device       *lod    = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
767         struct dt_object        *obj    = dt_object_child(&lo->ldo_obj);
768         struct lmv_mds_md_v1    *lmv1   = buf->lb_buf;
769         struct dt_it            *it;
770         const struct dt_it_ops  *iops;
771         __u32                    stripes;
772         __u32                    magic  = le32_to_cpu(lmv1->lmv_magic);
773         size_t                   lmv1_size;
774         int                      rc;
775         ENTRY;
776
777         /* If it is not a striped directory, then load nothing. */
778         if (magic != LMV_MAGIC_V1)
779                 RETURN(0);
780
781         /* If it is in migration (or failure), then load nothing. */
782         if (le32_to_cpu(lmv1->lmv_hash_type) & LMV_HASH_FLAG_MIGRATION)
783                 RETURN(0);
784
785         stripes = le32_to_cpu(lmv1->lmv_stripe_count);
786         if (stripes < 1)
787                 RETURN(0);
788
789         rc = lmv_mds_md_size(stripes, magic);
790         if (rc < 0)
791                 RETURN(rc);
792         lmv1_size = rc;
793         if (buf->lb_len < lmv1_size) {
794                 struct lu_buf tbuf;
795
796                 if (!resize)
797                         RETURN(-ERANGE);
798
799                 tbuf = *buf;
800                 buf->lb_buf = NULL;
801                 buf->lb_len = 0;
802                 lu_buf_alloc(buf, lmv1_size);
803                 lmv1 = buf->lb_buf;
804                 if (lmv1 == NULL)
805                         RETURN(-ENOMEM);
806
807                 memcpy(buf->lb_buf, tbuf.lb_buf, tbuf.lb_len);
808         }
809
810         if (unlikely(!dt_try_as_dir(env, obj)))
811                 RETURN(-ENOTDIR);
812
813         memset(&lmv1->lmv_stripe_fids[0], 0, stripes * sizeof(struct lu_fid));
814         iops = &obj->do_index_ops->dio_it;
815         it = iops->init(env, obj, LUDA_64BITHASH);
816         if (IS_ERR(it))
817                 RETURN(PTR_ERR(it));
818
819         rc = iops->load(env, it, 0);
820         if (rc == 0)
821                 rc = iops->next(env, it);
822         else if (rc > 0)
823                 rc = 0;
824
825         while (rc == 0) {
826                 char             name[FID_LEN + 2] = "";
827                 struct lu_fid    fid;
828                 __u32            index;
829                 int              len;
830
831                 rc = iops->rec(env, it, (struct dt_rec *)ent, LUDA_64BITHASH);
832                 if (rc != 0)
833                         break;
834
835                 rc = -EIO;
836
837                 fid_le_to_cpu(&fid, &ent->lde_fid);
838                 ent->lde_namelen = le16_to_cpu(ent->lde_namelen);
839                 if (ent->lde_name[0] == '.') {
840                         if (ent->lde_namelen == 1)
841                                 goto next;
842
843                         if (ent->lde_namelen == 2 && ent->lde_name[1] == '.')
844                                 goto next;
845                 }
846
847                 len = snprintf(name, sizeof(name),
848                                DFID":", PFID(&ent->lde_fid));
849                 /* The ent->lde_name is composed of ${FID}:${index} */
850                 if (ent->lde_namelen < len + 1 ||
851                     memcmp(ent->lde_name, name, len) != 0) {
852                         CDEBUG(lod->lod_lmv_failout ? D_ERROR : D_INFO,
853                                "%s: invalid shard name %.*s with the FID "DFID
854                                " for the striped directory "DFID", %s\n",
855                                lod2obd(lod)->obd_name, ent->lde_namelen,
856                                ent->lde_name, PFID(&fid),
857                                PFID(lu_object_fid(&obj->do_lu)),
858                                lod->lod_lmv_failout ? "failout" : "skip");
859
860                         if (lod->lod_lmv_failout)
861                                 break;
862
863                         goto next;
864                 }
865
866                 index = 0;
867                 do {
868                         if (ent->lde_name[len] < '0' ||
869                             ent->lde_name[len] > '9') {
870                                 CDEBUG(lod->lod_lmv_failout ? D_ERROR : D_INFO,
871                                        "%s: invalid shard name %.*s with the "
872                                        "FID "DFID" for the striped directory "
873                                        DFID", %s\n",
874                                        lod2obd(lod)->obd_name, ent->lde_namelen,
875                                        ent->lde_name, PFID(&fid),
876                                        PFID(lu_object_fid(&obj->do_lu)),
877                                        lod->lod_lmv_failout ?
878                                        "failout" : "skip");
879
880                                 if (lod->lod_lmv_failout)
881                                         break;
882
883                                 goto next;
884                         }
885
886                         index = index * 10 + ent->lde_name[len++] - '0';
887                 } while (len < ent->lde_namelen);
888
889                 if (len == ent->lde_namelen) {
890                         /* Out of LMV EA range. */
891                         if (index >= stripes) {
892                                 CERROR("%s: the shard %.*s for the striped "
893                                        "directory "DFID" is out of the known "
894                                        "LMV EA range [0 - %u], failout\n",
895                                        lod2obd(lod)->obd_name, ent->lde_namelen,
896                                        ent->lde_name,
897                                        PFID(lu_object_fid(&obj->do_lu)),
898                                        stripes - 1);
899
900                                 break;
901                         }
902
903                         /* The slot has been occupied. */
904                         if (!fid_is_zero(&lmv1->lmv_stripe_fids[index])) {
905                                 struct lu_fid fid0;
906
907                                 fid_le_to_cpu(&fid0,
908                                         &lmv1->lmv_stripe_fids[index]);
909                                 CERROR("%s: both the shard "DFID" and "DFID
910                                        " for the striped directory "DFID
911                                        " claim the same LMV EA slot at the "
912                                        "index %d, failout\n",
913                                        lod2obd(lod)->obd_name,
914                                        PFID(&fid0), PFID(&fid),
915                                        PFID(lu_object_fid(&obj->do_lu)), index);
916
917                                 break;
918                         }
919
920                         /* stored as LE mode */
921                         lmv1->lmv_stripe_fids[index] = ent->lde_fid;
922
923 next:
924                         rc = iops->next(env, it);
925                 }
926         }
927
928         iops->put(env, it);
929         iops->fini(env, it);
930
931         RETURN(rc > 0 ? lmv_mds_md_size(stripes, magic) : rc);
932 }
933
934 /**
935  * Implementation of dt_object_operations::do_index_try.
936  *
937  * \see dt_object_operations::do_index_try() in the API description for details.
938  */
939 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
940                          const struct dt_index_features *feat)
941 {
942         struct lod_object       *lo = lod_dt_obj(dt);
943         struct dt_object        *next = dt_object_child(dt);
944         int                     rc;
945         ENTRY;
946
947         LASSERT(next->do_ops);
948         LASSERT(next->do_ops->do_index_try);
949
950         rc = lod_load_striping_locked(env, lo);
951         if (rc != 0)
952                 RETURN(rc);
953
954         rc = next->do_ops->do_index_try(env, next, feat);
955         if (rc != 0)
956                 RETURN(rc);
957
958         if (lo->ldo_dir_stripe_count > 0) {
959                 int i;
960
961                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
962                         if (dt_object_exists(lo->ldo_stripe[i]) == 0)
963                                 continue;
964                         rc = lo->ldo_stripe[i]->do_ops->do_index_try(env,
965                                                 lo->ldo_stripe[i], feat);
966                         if (rc != 0)
967                                 RETURN(rc);
968                 }
969                 dt->do_index_ops = &lod_striped_index_ops;
970         } else {
971                 dt->do_index_ops = &lod_index_ops;
972         }
973
974         RETURN(rc);
975 }
976
977 /**
978  * Implementation of dt_object_operations::do_read_lock.
979  *
980  * \see dt_object_operations::do_read_lock() in the API description for details.
981  */
982 static void lod_read_lock(const struct lu_env *env, struct dt_object *dt,
983                           unsigned role)
984 {
985         dt_read_lock(env, dt_object_child(dt), role);
986 }
987
988 /**
989  * Implementation of dt_object_operations::do_write_lock.
990  *
991  * \see dt_object_operations::do_write_lock() in the API description for
992  * details.
993  */
994 static void lod_write_lock(const struct lu_env *env, struct dt_object *dt,
995                            unsigned role)
996 {
997         dt_write_lock(env, dt_object_child(dt), role);
998 }
999
1000 /**
1001  * Implementation of dt_object_operations::do_read_unlock.
1002  *
1003  * \see dt_object_operations::do_read_unlock() in the API description for
1004  * details.
1005  */
1006 static void lod_read_unlock(const struct lu_env *env, struct dt_object *dt)
1007 {
1008         dt_read_unlock(env, dt_object_child(dt));
1009 }
1010
1011 /**
1012  * Implementation of dt_object_operations::do_write_unlock.
1013  *
1014  * \see dt_object_operations::do_write_unlock() in the API description for
1015  * details.
1016  */
1017 static void lod_write_unlock(const struct lu_env *env, struct dt_object *dt)
1018 {
1019         dt_write_unlock(env, dt_object_child(dt));
1020 }
1021
1022 /**
1023  * Implementation of dt_object_operations::do_write_locked.
1024  *
1025  * \see dt_object_operations::do_write_locked() in the API description for
1026  * details.
1027  */
1028 static int lod_write_locked(const struct lu_env *env, struct dt_object *dt)
1029 {
1030         return dt_write_locked(env, dt_object_child(dt));
1031 }
1032
1033 /**
1034  * Implementation of dt_object_operations::do_attr_get.
1035  *
1036  * \see dt_object_operations::do_attr_get() in the API description for details.
1037  */
1038 static int lod_attr_get(const struct lu_env *env,
1039                         struct dt_object *dt,
1040                         struct lu_attr *attr)
1041 {
1042         /* Note: for striped directory, client will merge attributes
1043          * from all of the sub-stripes see lmv_merge_attr(), and there
1044          * no MDD logic depend on directory nlink/size/time, so we can
1045          * always use master inode nlink and size for now. */
1046         return dt_attr_get(env, dt_object_child(dt), attr);
1047 }
1048
1049 int lod_obj_for_each_stripe(const struct lu_env *env, struct lod_object *lo,
1050                             struct thandle *th,
1051                             struct lod_obj_stripe_cb_data *data)
1052 {
1053         struct lod_layout_component *lod_comp;
1054         int i, j, rc;
1055         ENTRY;
1056
1057         LASSERT(lo->ldo_comp_cnt != 0 && lo->ldo_comp_entries != NULL);
1058         for (i = 0; i < lo->ldo_comp_cnt; i++) {
1059                 lod_comp = &lo->ldo_comp_entries[i];
1060
1061                 if (lod_comp->llc_stripe == NULL)
1062                         continue;
1063
1064                 /* has stripe but not inited yet, this component has been
1065                  * declared to be created, but hasn't created yet.
1066                  */
1067                 if (!lod_comp_inited(lod_comp))
1068                         continue;
1069
1070                 if (data->locd_comp_skip_cb &&
1071                     data->locd_comp_skip_cb(env, lo, i, data))
1072                         continue;
1073
1074                 LASSERT(lod_comp->llc_stripe_count > 0);
1075                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
1076                         struct dt_object *dt = lod_comp->llc_stripe[j];
1077
1078                         if (dt == NULL)
1079                                 continue;
1080                         rc = data->locd_stripe_cb(env, lo, dt, th, i, j, data);
1081                         if (rc != 0)
1082                                 RETURN(rc);
1083                 }
1084         }
1085         RETURN(0);
1086 }
1087
1088 static bool lod_obj_attr_set_comp_skip_cb(const struct lu_env *env,
1089                 struct lod_object *lo, int comp_idx,
1090                 struct lod_obj_stripe_cb_data *data)
1091 {
1092         struct lod_layout_component *lod_comp = &lo->ldo_comp_entries[comp_idx];
1093         bool skipped = false;
1094
1095         if (!(data->locd_attr->la_valid & LA_LAYOUT_VERSION))
1096                 return skipped;
1097
1098         switch (lo->ldo_flr_state) {
1099         case LCM_FL_WRITE_PENDING: {
1100                 int i;
1101
1102                 /* skip stale components */
1103                 if (lod_comp->llc_flags & LCME_FL_STALE) {
1104                         skipped = true;
1105                         break;
1106                 }
1107
1108                 /* skip valid and overlapping components, therefore any
1109                  * attempts to write overlapped components will never succeed
1110                  * because client will get EINPROGRESS. */
1111                 for (i = 0; i < lo->ldo_comp_cnt; i++) {
1112                         if (i == comp_idx)
1113                                 continue;
1114
1115                         if (lo->ldo_comp_entries[i].llc_flags & LCME_FL_STALE)
1116                                 continue;
1117
1118                         if (lu_extent_is_overlapped(&lod_comp->llc_extent,
1119                                         &lo->ldo_comp_entries[i].llc_extent)) {
1120                                 skipped = true;
1121                                 break;
1122                         }
1123                 }
1124                 break;
1125         }
1126         default:
1127                 LASSERTF(0, "impossible: %d\n", lo->ldo_flr_state);
1128         case LCM_FL_SYNC_PENDING:
1129                 break;
1130         }
1131
1132         CDEBUG(D_LAYOUT, DFID": %s to set component %x to version: %u\n",
1133                PFID(lu_object_fid(&lo->ldo_obj.do_lu)),
1134                skipped ? "skipped" : "chose", lod_comp->llc_id,
1135                data->locd_attr->la_layout_version);
1136
1137         return skipped;
1138 }
1139
1140 static inline int
1141 lod_obj_stripe_attr_set_cb(const struct lu_env *env, struct lod_object *lo,
1142                            struct dt_object *dt, struct thandle *th,
1143                            int comp_idx, int stripe_idx,
1144                            struct lod_obj_stripe_cb_data *data)
1145 {
1146         if (data->locd_declare)
1147                 return lod_sub_declare_attr_set(env, dt, data->locd_attr, th);
1148
1149         return lod_sub_attr_set(env, dt, data->locd_attr, th);
1150 }
1151
1152 /**
1153  * Implementation of dt_object_operations::do_declare_attr_set.
1154  *
1155  * If the object is striped, then apply the changes to all the stripes.
1156  *
1157  * \see dt_object_operations::do_declare_attr_set() in the API description
1158  * for details.
1159  */
1160 static int lod_declare_attr_set(const struct lu_env *env,
1161                                 struct dt_object *dt,
1162                                 const struct lu_attr *attr,
1163                                 struct thandle *th)
1164 {
1165         struct dt_object  *next = dt_object_child(dt);
1166         struct lod_object *lo = lod_dt_obj(dt);
1167         int                rc, i;
1168         ENTRY;
1169
1170         /*
1171          * declare setattr on the local object
1172          */
1173         rc = lod_sub_declare_attr_set(env, next, attr, th);
1174         if (rc)
1175                 RETURN(rc);
1176
1177         /* osp_declare_attr_set() ignores all attributes other than
1178          * UID, GID, PROJID, and size, and osp_attr_set() ignores all
1179          * but UID, GID and PROJID. Declaration of size attr setting
1180          * happens through lod_declare_init_size(), and not through
1181          * this function. Therefore we need not load striping unless
1182          * ownership is changing.  This should save memory and (we hope)
1183          * speed up rename().
1184          */
1185         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1186                 if (!(attr->la_valid & LA_REMOTE_ATTR_SET))
1187                         RETURN(rc);
1188
1189                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
1190                         RETURN(0);
1191         } else {
1192                 if (!(attr->la_valid & (LA_UID | LA_GID | LA_PROJID | LA_MODE |
1193                                         LA_ATIME | LA_MTIME | LA_CTIME |
1194                                         LA_FLAGS)))
1195                         RETURN(rc);
1196         }
1197         /*
1198          * load striping information, notice we don't do this when object
1199          * is being initialized as we don't need this information till
1200          * few specific cases like destroy, chown
1201          */
1202         rc = lod_load_striping(env, lo);
1203         if (rc)
1204                 RETURN(rc);
1205
1206         if (!lod_obj_is_striped(dt))
1207                 RETURN(0);
1208
1209         /*
1210          * if object is striped declare changes on the stripes
1211          */
1212         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1213                 LASSERT(lo->ldo_stripe);
1214                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
1215                         if (lo->ldo_stripe[i] == NULL)
1216                                 continue;
1217                         rc = lod_sub_declare_attr_set(env, lo->ldo_stripe[i],
1218                                                       attr, th);
1219                         if (rc != 0)
1220                                 RETURN(rc);
1221                 }
1222         } else {
1223                 struct lod_obj_stripe_cb_data data = { { 0 } };
1224
1225                 data.locd_attr = attr;
1226                 data.locd_declare = true;
1227                 data.locd_stripe_cb = lod_obj_stripe_attr_set_cb;
1228                 rc = lod_obj_for_each_stripe(env, lo, th, &data);
1229         }
1230
1231         if (rc)
1232                 RETURN(rc);
1233
1234         if (!dt_object_exists(next) || dt_object_remote(next) ||
1235             !S_ISREG(attr->la_mode))
1236                 RETURN(0);
1237
1238         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE)) {
1239                 rc = lod_sub_declare_xattr_del(env, next, XATTR_NAME_LOV, th);
1240                 RETURN(rc);
1241         }
1242
1243         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE) ||
1244             OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PFL_RANGE)) {
1245                 struct lod_thread_info *info = lod_env_info(env);
1246                 struct lu_buf *buf = &info->lti_buf;
1247
1248                 buf->lb_buf = info->lti_ea_store;
1249                 buf->lb_len = info->lti_ea_store_size;
1250                 rc = lod_sub_declare_xattr_set(env, next, buf, XATTR_NAME_LOV,
1251                                                LU_XATTR_REPLACE, th);
1252         }
1253
1254         RETURN(rc);
1255 }
1256
1257 /**
1258  * Implementation of dt_object_operations::do_attr_set.
1259  *
1260  * If the object is striped, then apply the changes to all or subset of
1261  * the stripes depending on the object type and specific attributes.
1262  *
1263  * \see dt_object_operations::do_attr_set() in the API description for details.
1264  */
1265 static int lod_attr_set(const struct lu_env *env,
1266                         struct dt_object *dt,
1267                         const struct lu_attr *attr,
1268                         struct thandle *th)
1269 {
1270         struct dt_object        *next = dt_object_child(dt);
1271         struct lod_object       *lo = lod_dt_obj(dt);
1272         int                     rc, i;
1273         ENTRY;
1274
1275         /*
1276          * apply changes to the local object
1277          */
1278         rc = lod_sub_attr_set(env, next, attr, th);
1279         if (rc)
1280                 RETURN(rc);
1281
1282         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1283                 if (!(attr->la_valid & LA_REMOTE_ATTR_SET))
1284                         RETURN(rc);
1285
1286                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
1287                         RETURN(0);
1288         } else {
1289                 if (!(attr->la_valid & (LA_UID | LA_GID | LA_MODE | LA_PROJID |
1290                                         LA_ATIME | LA_MTIME | LA_CTIME |
1291                                         LA_FLAGS)))
1292                         RETURN(rc);
1293         }
1294
1295         /* FIXME: a tricky case in the code path of mdd_layout_change():
1296          * the in-memory striping information has been freed in lod_xattr_set()
1297          * due to layout change. It has to load stripe here again. It only
1298          * changes flags of layout so declare_attr_set() is still accurate */
1299         rc = lod_load_striping_locked(env, lo);
1300         if (rc)
1301                 RETURN(rc);
1302
1303         if (!lod_obj_is_striped(dt))
1304                 RETURN(0);
1305
1306         /*
1307          * if object is striped, apply changes to all the stripes
1308          */
1309         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1310                 LASSERT(lo->ldo_stripe);
1311                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
1312                         if (unlikely(lo->ldo_stripe[i] == NULL))
1313                                 continue;
1314
1315                         if ((dt_object_exists(lo->ldo_stripe[i]) == 0))
1316                                 continue;
1317
1318                         rc = lod_sub_attr_set(env, lo->ldo_stripe[i], attr, th);
1319                         if (rc != 0)
1320                                 break;
1321                 }
1322         } else {
1323                 struct lod_obj_stripe_cb_data data = { { 0 } };
1324
1325                 data.locd_attr = attr;
1326                 data.locd_declare = false;
1327                 data.locd_comp_skip_cb = lod_obj_attr_set_comp_skip_cb;
1328                 data.locd_stripe_cb = lod_obj_stripe_attr_set_cb;
1329                 rc = lod_obj_for_each_stripe(env, lo, th, &data);
1330         }
1331
1332         if (rc)
1333                 RETURN(rc);
1334
1335         if (!dt_object_exists(next) || dt_object_remote(next) ||
1336             !S_ISREG(attr->la_mode))
1337                 RETURN(0);
1338
1339         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE)) {
1340                 rc = lod_sub_xattr_del(env, next, XATTR_NAME_LOV, th);
1341                 RETURN(rc);
1342         }
1343
1344         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE)) {
1345                 struct lod_thread_info *info = lod_env_info(env);
1346                 struct lu_buf *buf = &info->lti_buf;
1347                 struct ost_id *oi = &info->lti_ostid;
1348                 struct lu_fid *fid = &info->lti_fid;
1349                 struct lov_mds_md_v1 *lmm;
1350                 struct lov_ost_data_v1 *objs;
1351                 __u32 magic;
1352
1353                 rc = lod_get_lov_ea(env, lo);
1354                 if (rc <= 0)
1355                         RETURN(rc);
1356
1357                 buf->lb_buf = info->lti_ea_store;
1358                 buf->lb_len = info->lti_ea_store_size;
1359                 lmm = info->lti_ea_store;
1360                 magic = le32_to_cpu(lmm->lmm_magic);
1361                 if (magic == LOV_MAGIC_COMP_V1) {
1362                         struct lov_comp_md_v1 *lcm = buf->lb_buf;
1363                         struct lov_comp_md_entry_v1 *lcme =
1364                                                 &lcm->lcm_entries[0];
1365
1366                         lmm = buf->lb_buf + le32_to_cpu(lcme->lcme_offset);
1367                         magic = le32_to_cpu(lmm->lmm_magic);
1368                 }
1369
1370                 if (magic == LOV_MAGIC_V1)
1371                         objs = &(lmm->lmm_objects[0]);
1372                 else
1373                         objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
1374                 ostid_le_to_cpu(&objs->l_ost_oi, oi);
1375                 ostid_to_fid(fid, oi, le32_to_cpu(objs->l_ost_idx));
1376                 fid->f_oid--;
1377                 fid_to_ostid(fid, oi);
1378                 ostid_cpu_to_le(oi, &objs->l_ost_oi);
1379
1380                 rc = lod_sub_xattr_set(env, next, buf, XATTR_NAME_LOV,
1381                                        LU_XATTR_REPLACE, th);
1382         } else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PFL_RANGE)) {
1383                 struct lod_thread_info *info = lod_env_info(env);
1384                 struct lu_buf *buf = &info->lti_buf;
1385                 struct lov_comp_md_v1 *lcm;
1386                 struct lov_comp_md_entry_v1 *lcme;
1387
1388                 rc = lod_get_lov_ea(env, lo);
1389                 if (rc <= 0)
1390                         RETURN(rc);
1391
1392                 buf->lb_buf = info->lti_ea_store;
1393                 buf->lb_len = info->lti_ea_store_size;
1394                 lcm = buf->lb_buf;
1395                 if (le32_to_cpu(lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
1396                         RETURN(-EINVAL);
1397
1398                 le32_add_cpu(&lcm->lcm_layout_gen, 1);
1399                 lcme = &lcm->lcm_entries[0];
1400                 le64_add_cpu(&lcme->lcme_extent.e_start, 1);
1401                 le64_add_cpu(&lcme->lcme_extent.e_end, -1);
1402
1403                 rc = lod_sub_xattr_set(env, next, buf, XATTR_NAME_LOV,
1404                                        LU_XATTR_REPLACE, th);
1405         }
1406
1407         RETURN(rc);
1408 }
1409
1410 /**
1411  * Implementation of dt_object_operations::do_xattr_get.
1412  *
1413  * If LOV EA is requested from the root object and it's not
1414  * found, then return default striping for the filesystem.
1415  *
1416  * \see dt_object_operations::do_xattr_get() in the API description for details.
1417  */
1418 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
1419                          struct lu_buf *buf, const char *name)
1420 {
1421         struct lod_thread_info *info = lod_env_info(env);
1422         struct lod_device *dev = lu2lod_dev(dt->do_lu.lo_dev);
1423         int is_root;
1424         int rc;
1425         ENTRY;
1426
1427         rc = dt_xattr_get(env, dt_object_child(dt), buf, name);
1428         if (strcmp(name, XATTR_NAME_LMV) == 0) {
1429                 struct lmv_mds_md_v1    *lmv1;
1430                 int                      rc1 = 0;
1431
1432                 if (rc > (typeof(rc))sizeof(*lmv1))
1433                         RETURN(rc);
1434
1435                 if (rc < (typeof(rc))sizeof(*lmv1))
1436                         RETURN(rc = rc > 0 ? -EINVAL : rc);
1437
1438                 if (buf->lb_buf == NULL || buf->lb_len == 0) {
1439                         CLASSERT(sizeof(*lmv1) <= sizeof(info->lti_key));
1440
1441                         info->lti_buf.lb_buf = info->lti_key;
1442                         info->lti_buf.lb_len = sizeof(*lmv1);
1443                         rc = dt_xattr_get(env, dt_object_child(dt),
1444                                           &info->lti_buf, name);
1445                         if (unlikely(rc != sizeof(*lmv1)))
1446                                 RETURN(rc = rc > 0 ? -EINVAL : rc);
1447
1448                         lmv1 = info->lti_buf.lb_buf;
1449                         /* The on-disk LMV EA only contains header, but the
1450                          * returned LMV EA size should contain the space for
1451                          * the FIDs of all shards of the striped directory. */
1452                         if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_V1)
1453                                 rc = lmv_mds_md_size(
1454                                         le32_to_cpu(lmv1->lmv_stripe_count),
1455                                         LMV_MAGIC_V1);
1456                 } else {
1457                         rc1 = lod_load_lmv_shards(env, lod_dt_obj(dt),
1458                                                   buf, false);
1459                 }
1460
1461                 RETURN(rc = rc1 != 0 ? rc1 : rc);
1462         }
1463
1464         if (rc != -ENODATA || !S_ISDIR(dt->do_lu.lo_header->loh_attr & S_IFMT))
1465                 RETURN(rc);
1466
1467         /*
1468          * XXX: Only used by lfsck
1469          *
1470          * lod returns default striping on the real root of the device
1471          * this is like the root stores default striping for the whole
1472          * filesystem. historically we've been using a different approach
1473          * and store it in the config.
1474          */
1475         dt_root_get(env, dev->lod_child, &info->lti_fid);
1476         is_root = lu_fid_eq(&info->lti_fid, lu_object_fid(&dt->do_lu));
1477
1478         if (is_root && strcmp(XATTR_NAME_LOV, name) == 0) {
1479                 struct lov_user_md *lum = buf->lb_buf;
1480                 struct lov_desc    *desc = &dev->lod_desc;
1481
1482                 if (buf->lb_buf == NULL) {
1483                         rc = sizeof(*lum);
1484                 } else if (buf->lb_len >= sizeof(*lum)) {
1485                         lum->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V1);
1486                         lmm_oi_set_seq(&lum->lmm_oi, FID_SEQ_LOV_DEFAULT);
1487                         lmm_oi_set_id(&lum->lmm_oi, 0);
1488                         lmm_oi_cpu_to_le(&lum->lmm_oi, &lum->lmm_oi);
1489                         lum->lmm_pattern = cpu_to_le32(desc->ld_pattern);
1490                         lum->lmm_stripe_size = cpu_to_le32(
1491                                                 desc->ld_default_stripe_size);
1492                         lum->lmm_stripe_count = cpu_to_le16(
1493                                                 desc->ld_default_stripe_count);
1494                         lum->lmm_stripe_offset = cpu_to_le16(
1495                                                 desc->ld_default_stripe_offset);
1496                         rc = sizeof(*lum);
1497                 } else {
1498                         rc = -ERANGE;
1499                 }
1500         }
1501
1502         RETURN(rc);
1503 }
1504
1505 /**
1506  * Verify LVM EA.
1507  *
1508  * Checks that the magic of the stripe is sane.
1509  *
1510  * \param[in] lod       lod device
1511  * \param[in] lum       a buffer storing LMV EA to verify
1512  *
1513  * \retval              0 if the EA is sane
1514  * \retval              negative otherwise
1515  */
1516 static int lod_verify_md_striping(struct lod_device *lod,
1517                                   const struct lmv_user_md_v1 *lum)
1518 {
1519         if (unlikely(le32_to_cpu(lum->lum_magic) != LMV_USER_MAGIC)) {
1520                 CERROR("%s: invalid lmv_user_md: magic = %x, "
1521                        "stripe_offset = %d, stripe_count = %u: rc = %d\n",
1522                        lod2obd(lod)->obd_name, le32_to_cpu(lum->lum_magic),
1523                        (int)le32_to_cpu(lum->lum_stripe_offset),
1524                        le32_to_cpu(lum->lum_stripe_count), -EINVAL);
1525                 return -EINVAL;
1526         }
1527
1528         return 0;
1529 }
1530
1531 /**
1532  * Initialize LMV EA for a slave.
1533  *
1534  * Initialize slave's LMV EA from the master's LMV EA.
1535  *
1536  * \param[in] master_lmv        a buffer containing master's EA
1537  * \param[out] slave_lmv        a buffer where slave's EA will be stored
1538  *
1539  */
1540 static void lod_prep_slave_lmv_md(struct lmv_mds_md_v1 *slave_lmv,
1541                                   const struct lmv_mds_md_v1 *master_lmv)
1542 {
1543         *slave_lmv = *master_lmv;
1544         slave_lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_STRIPE);
1545 }
1546
1547 /**
1548  * Generate LMV EA.
1549  *
1550  * Generate LMV EA from the object passed as \a dt. The object must have
1551  * the stripes created and initialized.
1552  *
1553  * \param[in] env       execution environment
1554  * \param[in] dt        object
1555  * \param[out] lmv_buf  buffer storing generated LMV EA
1556  *
1557  * \retval              0 on success
1558  * \retval              negative if failed
1559  */
1560 static int lod_prep_lmv_md(const struct lu_env *env, struct dt_object *dt,
1561                            struct lu_buf *lmv_buf)
1562 {
1563         struct lod_thread_info  *info = lod_env_info(env);
1564         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1565         struct lod_object       *lo = lod_dt_obj(dt);
1566         struct lmv_mds_md_v1    *lmm1;
1567         int                     stripe_count;
1568         int                     type = LU_SEQ_RANGE_ANY;
1569         int                     rc;
1570         __u32                   mdtidx;
1571         ENTRY;
1572
1573         LASSERT(lo->ldo_dir_striped != 0);
1574         LASSERT(lo->ldo_dir_stripe_count > 0);
1575         stripe_count = lo->ldo_dir_stripe_count;
1576         /* Only store the LMV EA heahder on the disk. */
1577         if (info->lti_ea_store_size < sizeof(*lmm1)) {
1578                 rc = lod_ea_store_resize(info, sizeof(*lmm1));
1579                 if (rc != 0)
1580                         RETURN(rc);
1581         } else {
1582                 memset(info->lti_ea_store, 0, sizeof(*lmm1));
1583         }
1584
1585         lmm1 = (struct lmv_mds_md_v1 *)info->lti_ea_store;
1586         lmm1->lmv_magic = cpu_to_le32(LMV_MAGIC);
1587         lmm1->lmv_stripe_count = cpu_to_le32(stripe_count);
1588         lmm1->lmv_hash_type = cpu_to_le32(lo->ldo_dir_hash_type);
1589         rc = lod_fld_lookup(env, lod, lu_object_fid(&dt->do_lu),
1590                             &mdtidx, &type);
1591         if (rc != 0)
1592                 RETURN(rc);
1593
1594         lmm1->lmv_master_mdt_index = cpu_to_le32(mdtidx);
1595         lmv_buf->lb_buf = info->lti_ea_store;
1596         lmv_buf->lb_len = sizeof(*lmm1);
1597
1598         RETURN(rc);
1599 }
1600
1601 /**
1602  * Create in-core represenation for a striped directory.
1603  *
1604  * Parse the buffer containing LMV EA and instantiate LU objects
1605  * representing the stripe objects. The pointers to the objects are
1606  * stored in ldo_stripe field of \a lo. This function is used when
1607  * we need to access an already created object (i.e. load from a disk).
1608  *
1609  * \param[in] env       execution environment
1610  * \param[in] lo        lod object
1611  * \param[in] buf       buffer containing LMV EA
1612  *
1613  * \retval              0 on success
1614  * \retval              negative if failed
1615  */
1616 int lod_parse_dir_striping(const struct lu_env *env, struct lod_object *lo,
1617                            const struct lu_buf *buf)
1618 {
1619         struct lod_thread_info  *info = lod_env_info(env);
1620         struct lod_device       *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1621         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
1622         struct dt_object        **stripe;
1623         union lmv_mds_md        *lmm = buf->lb_buf;
1624         struct lmv_mds_md_v1    *lmv1 = &lmm->lmv_md_v1;
1625         struct lu_fid           *fid = &info->lti_fid;
1626         unsigned int            i;
1627         int                     rc = 0;
1628         ENTRY;
1629
1630         if (le32_to_cpu(lmv1->lmv_hash_type) & LMV_HASH_FLAG_MIGRATION)
1631                 RETURN(0);
1632
1633         if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_STRIPE) {
1634                 lo->ldo_dir_slave_stripe = 1;
1635                 RETURN(0);
1636         }
1637
1638         if (le32_to_cpu(lmv1->lmv_magic) != LMV_MAGIC_V1)
1639                 RETURN(-EINVAL);
1640
1641         if (le32_to_cpu(lmv1->lmv_stripe_count) < 1)
1642                 RETURN(0);
1643
1644         LASSERT(lo->ldo_stripe == NULL);
1645         OBD_ALLOC(stripe, sizeof(stripe[0]) *
1646                   (le32_to_cpu(lmv1->lmv_stripe_count)));
1647         if (stripe == NULL)
1648                 RETURN(-ENOMEM);
1649
1650         for (i = 0; i < le32_to_cpu(lmv1->lmv_stripe_count); i++) {
1651                 struct dt_device        *tgt_dt;
1652                 struct dt_object        *dto;
1653                 int                     type = LU_SEQ_RANGE_ANY;
1654                 __u32                   idx;
1655
1656                 fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[i]);
1657                 if (!fid_is_sane(fid))
1658                         GOTO(out, rc = -ESTALE);
1659
1660                 rc = lod_fld_lookup(env, lod, fid, &idx, &type);
1661                 if (rc != 0)
1662                         GOTO(out, rc);
1663
1664                 if (idx == lod2lu_dev(lod)->ld_site->ld_seq_site->ss_node_id) {
1665                         tgt_dt = lod->lod_child;
1666                 } else {
1667                         struct lod_tgt_desc     *tgt;
1668
1669                         tgt = LTD_TGT(ltd, idx);
1670                         if (tgt == NULL)
1671                                 GOTO(out, rc = -ESTALE);
1672                         tgt_dt = tgt->ltd_tgt;
1673                 }
1674
1675                 dto = dt_locate_at(env, tgt_dt, fid,
1676                                   lo->ldo_obj.do_lu.lo_dev->ld_site->ls_top_dev,
1677                                   NULL);
1678                 if (IS_ERR(dto))
1679                         GOTO(out, rc = PTR_ERR(dto));
1680
1681                 stripe[i] = dto;
1682         }
1683 out:
1684         lo->ldo_stripe = stripe;
1685         lo->ldo_dir_stripe_count = le32_to_cpu(lmv1->lmv_stripe_count);
1686         lo->ldo_dir_stripes_allocated = le32_to_cpu(lmv1->lmv_stripe_count);
1687         if (rc != 0)
1688                 lod_object_free_striping(env, lo);
1689
1690         RETURN(rc);
1691 }
1692
1693 /**
1694  * Declare create a striped directory.
1695  *
1696  * Declare creating a striped directory with a given stripe pattern on the
1697  * specified MDTs. A striped directory is represented as a regular directory
1698  * - an index listing all the stripes. The stripes point back to the master
1699  * object with ".." and LinkEA. The master object gets LMV EA which
1700  * identifies it as a striped directory. The function allocates FIDs
1701  * for all stripes.
1702  *
1703  * \param[in] env       execution environment
1704  * \param[in] dt        object
1705  * \param[in] attr      attributes to initialize the objects with
1706  * \param[in] dof       type of objects to be created
1707  * \param[in] th        transaction handle
1708  *
1709  * \retval              0 on success
1710  * \retval              negative if failed
1711  */
1712 static int lod_dir_declare_create_stripes(const struct lu_env *env,
1713                                           struct dt_object *dt,
1714                                           struct lu_attr *attr,
1715                                           struct dt_object_format *dof,
1716                                           struct thandle *th)
1717 {
1718         struct lod_thread_info  *info = lod_env_info(env);
1719         struct lu_buf           lmv_buf;
1720         struct lu_buf           slave_lmv_buf;
1721         struct lmv_mds_md_v1    *lmm;
1722         struct lmv_mds_md_v1    *slave_lmm = NULL;
1723         struct dt_insert_rec    *rec = &info->lti_dt_rec;
1724         struct lod_object       *lo = lod_dt_obj(dt);
1725         int                     rc;
1726         __u32                   i;
1727         ENTRY;
1728
1729         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
1730         if (rc != 0)
1731                 GOTO(out, rc);
1732         lmm = lmv_buf.lb_buf;
1733
1734         OBD_ALLOC_PTR(slave_lmm);
1735         if (slave_lmm == NULL)
1736                 GOTO(out, rc = -ENOMEM);
1737
1738         lod_prep_slave_lmv_md(slave_lmm, lmm);
1739         slave_lmv_buf.lb_buf = slave_lmm;
1740         slave_lmv_buf.lb_len = sizeof(*slave_lmm);
1741
1742         if (!dt_try_as_dir(env, dt_object_child(dt)))
1743                 GOTO(out, rc = -EINVAL);
1744
1745         rec->rec_type = S_IFDIR;
1746         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
1747                 struct dt_object        *dto = lo->ldo_stripe[i];
1748                 char                    *stripe_name = info->lti_key;
1749                 struct lu_name          *sname;
1750                 struct linkea_data       ldata          = { NULL };
1751                 struct lu_buf           linkea_buf;
1752
1753                 rc = lod_sub_declare_create(env, dto, attr, NULL, dof, th);
1754                 if (rc != 0)
1755                         GOTO(out, rc);
1756
1757                 if (!dt_try_as_dir(env, dto))
1758                         GOTO(out, rc = -EINVAL);
1759
1760                 rc = lod_sub_declare_ref_add(env, dto, th);
1761                 if (rc != 0)
1762                         GOTO(out, rc);
1763
1764                 rec->rec_fid = lu_object_fid(&dto->do_lu);
1765                 rc = lod_sub_declare_insert(env, dto,
1766                                             (const struct dt_rec *)rec,
1767                                             (const struct dt_key *)dot, th);
1768                 if (rc != 0)
1769                         GOTO(out, rc);
1770
1771                 /* master stripe FID will be put to .. */
1772                 rec->rec_fid = lu_object_fid(&dt->do_lu);
1773                 rc = lod_sub_declare_insert(env, dto,
1774                                             (const struct dt_rec *)rec,
1775                                             (const struct dt_key *)dotdot, th);
1776                 if (rc != 0)
1777                         GOTO(out, rc);
1778
1779                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
1780                     cfs_fail_val != i) {
1781                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
1782                             cfs_fail_val == i)
1783                                 slave_lmm->lmv_master_mdt_index =
1784                                                         cpu_to_le32(i + 1);
1785                         else
1786                                 slave_lmm->lmv_master_mdt_index =
1787                                                         cpu_to_le32(i);
1788                         rc = lod_sub_declare_xattr_set(env, dto, &slave_lmv_buf,
1789                                                        XATTR_NAME_LMV, 0, th);
1790                         if (rc != 0)
1791                                 GOTO(out, rc);
1792                 }
1793
1794                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
1795                     cfs_fail_val == i)
1796                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1797                                 PFID(lu_object_fid(&dto->do_lu)), i + 1);
1798                 else
1799                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1800                                 PFID(lu_object_fid(&dto->do_lu)), i);
1801
1802                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
1803                 rc = linkea_links_new(&ldata, &info->lti_linkea_buf,
1804                                       sname, lu_object_fid(&dt->do_lu));
1805                 if (rc != 0)
1806                         GOTO(out, rc);
1807
1808                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
1809                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
1810                 rc = lod_sub_declare_xattr_set(env, dto, &linkea_buf,
1811                                                XATTR_NAME_LINK, 0, th);
1812                 if (rc != 0)
1813                         GOTO(out, rc);
1814
1815                 rec->rec_fid = lu_object_fid(&dto->do_lu);
1816                 rc = lod_sub_declare_insert(env, dt_object_child(dt),
1817                                             (const struct dt_rec *)rec,
1818                                             (const struct dt_key *)stripe_name,
1819                                             th);
1820                 if (rc != 0)
1821                         GOTO(out, rc);
1822
1823                 rc = lod_sub_declare_ref_add(env, dt_object_child(dt), th);
1824                 if (rc != 0)
1825                         GOTO(out, rc);
1826         }
1827
1828         rc = lod_sub_declare_xattr_set(env, dt_object_child(dt),
1829                                        &lmv_buf, XATTR_NAME_LMV, 0, th);
1830         if (rc != 0)
1831                 GOTO(out, rc);
1832 out:
1833         if (slave_lmm != NULL)
1834                 OBD_FREE_PTR(slave_lmm);
1835
1836         RETURN(rc);
1837 }
1838
1839 static int lod_prep_md_striped_create(const struct lu_env *env,
1840                                       struct dt_object *dt,
1841                                       struct lu_attr *attr,
1842                                       const struct lmv_user_md_v1 *lum,
1843                                       struct dt_object_format *dof,
1844                                       struct thandle *th)
1845 {
1846         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1847         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
1848         struct lod_object       *lo = lod_dt_obj(dt);
1849         struct dt_object        **stripe;
1850         __u32                   stripe_count;
1851         int                     *idx_array;
1852         __u32                   master_index;
1853         int                     rc = 0;
1854         __u32                   i;
1855         __u32                   j;
1856         ENTRY;
1857
1858         /* The lum has been verifed in lod_verify_md_striping */
1859         LASSERT(le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC);
1860         LASSERT(le32_to_cpu(lum->lum_stripe_count) > 0);
1861
1862         stripe_count = le32_to_cpu(lum->lum_stripe_count);
1863
1864         OBD_ALLOC(idx_array, sizeof(idx_array[0]) * stripe_count);
1865         if (idx_array == NULL)
1866                 RETURN(-ENOMEM);
1867
1868         OBD_ALLOC(stripe, sizeof(stripe[0]) * stripe_count);
1869         if (stripe == NULL)
1870                 GOTO(out_free, rc = -ENOMEM);
1871
1872         /* Start index must be the master MDT */
1873         master_index = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id;
1874         idx_array[0] = master_index;
1875         for (i = 0; i < stripe_count; i++) {
1876                 struct lod_tgt_desc     *tgt = NULL;
1877                 struct dt_object        *dto;
1878                 struct lu_fid           fid = { 0 };
1879                 int                     idx;
1880                 struct lu_object_conf   conf = { 0 };
1881                 struct dt_device        *tgt_dt = NULL;
1882
1883                 /* Try to find next avaible target */
1884                 idx = idx_array[i];
1885                 for (j = 0; j < lod->lod_remote_mdt_count;
1886                      j++, idx = (idx + 1) % (lod->lod_remote_mdt_count + 1)) {
1887                         bool already_allocated = false;
1888                         __u32 k;
1889
1890                         CDEBUG(D_INFO, "try idx %d, mdt cnt %u, allocated %u\n",
1891                                idx, lod->lod_remote_mdt_count + 1, i);
1892
1893                         if (likely(!OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE))) {
1894                                 /* check whether the idx already exists
1895                                  * in current allocated array */
1896                                 for (k = 0; k < i; k++) {
1897                                         if (idx_array[k] == idx) {
1898                                                 already_allocated = true;
1899                                                 break;
1900                                         }
1901                                 }
1902
1903                                 if (already_allocated)
1904                                         continue;
1905                         }
1906
1907                         /* Sigh, this index is not in the bitmap, let's check
1908                          * next available target */
1909                         if (!cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx) &&
1910                             idx != master_index)
1911                                 continue;
1912
1913                         if (idx == master_index) {
1914                                 /* Allocate the FID locally */
1915                                 rc = obd_fid_alloc(env, lod->lod_child_exp,
1916                                                    &fid, NULL);
1917                                 if (rc < 0)
1918                                         GOTO(out_put, rc);
1919                                 tgt_dt = lod->lod_child;
1920                                 break;
1921                         }
1922
1923                         /* check the status of the OSP */
1924                         tgt = LTD_TGT(ltd, idx);
1925                         if (tgt == NULL)
1926                                 continue;
1927
1928                         tgt_dt = tgt->ltd_tgt;
1929                         rc = dt_statfs(env, tgt_dt, NULL);
1930                         if (rc) {
1931                                 /* this OSP doesn't feel well */
1932                                 rc = 0;
1933                                 continue;
1934                         }
1935
1936                         rc = obd_fid_alloc(env, tgt->ltd_exp, &fid, NULL);
1937                         if (rc < 0) {
1938                                 rc = 0;
1939                                 continue;
1940                         }
1941
1942                         break;
1943                 }
1944
1945                 /* Can not allocate more stripes */
1946                 if (j == lod->lod_remote_mdt_count) {
1947                         CDEBUG(D_INFO, "%s: require stripes %u only get %d\n",
1948                                lod2obd(lod)->obd_name, stripe_count, i);
1949                         break;
1950                 }
1951
1952                 CDEBUG(D_INFO, "Get idx %d, for stripe %d "DFID"\n",
1953                        idx, i, PFID(&fid));
1954                 idx_array[i] = idx;
1955                 /* Set the start index for next stripe allocation */
1956                 if (i < stripe_count - 1)
1957                         idx_array[i + 1] = (idx + 1) %
1958                                            (lod->lod_remote_mdt_count + 1);
1959                 /* tgt_dt and fid must be ready after search avaible OSP
1960                  * in the above loop */
1961                 LASSERT(tgt_dt != NULL);
1962                 LASSERT(fid_is_sane(&fid));
1963                 conf.loc_flags = LOC_F_NEW;
1964                 dto = dt_locate_at(env, tgt_dt, &fid,
1965                                    dt->do_lu.lo_dev->ld_site->ls_top_dev,
1966                                    &conf);
1967                 if (IS_ERR(dto))
1968                         GOTO(out_put, rc = PTR_ERR(dto));
1969                 stripe[i] = dto;
1970         }
1971
1972         lo->ldo_dir_stripe_loaded = 1;
1973         lo->ldo_dir_striped = 1;
1974         lo->ldo_stripe = stripe;
1975         lo->ldo_dir_stripe_count = i;
1976         lo->ldo_dir_stripes_allocated = stripe_count;
1977
1978         if (lo->ldo_dir_stripe_count == 0)
1979                 GOTO(out_put, rc = -ENOSPC);
1980
1981         rc = lod_dir_declare_create_stripes(env, dt, attr, dof, th);
1982         if (rc != 0)
1983                 GOTO(out_put, rc);
1984
1985 out_put:
1986         if (rc < 0) {
1987                 for (i = 0; i < stripe_count; i++)
1988                         if (stripe[i] != NULL)
1989                                 dt_object_put(env, stripe[i]);
1990                 OBD_FREE(stripe, sizeof(stripe[0]) * stripe_count);
1991                 lo->ldo_dir_stripe_count = 0;
1992                 lo->ldo_dir_stripes_allocated = 0;
1993                 lo->ldo_stripe = NULL;
1994         }
1995
1996 out_free:
1997         OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count);
1998
1999         RETURN(rc);
2000 }
2001
2002 /**
2003  * Declare create striped md object.
2004  *
2005  * The function declares intention to create a striped directory. This is a
2006  * wrapper for lod_prep_md_striped_create(). The only additional functionality
2007  * is to verify pattern \a lum_buf is good. Check that function for the details.
2008  *
2009  * \param[in] env       execution environment
2010  * \param[in] dt        object
2011  * \param[in] attr      attributes to initialize the objects with
2012  * \param[in] lum_buf   a pattern specifying the number of stripes and
2013  *                      MDT to start from
2014  * \param[in] dof       type of objects to be created
2015  * \param[in] th        transaction handle
2016  *
2017  * \retval              0 on success
2018  * \retval              negative if failed
2019  *
2020  */
2021 static int lod_declare_xattr_set_lmv(const struct lu_env *env,
2022                                      struct dt_object *dt,
2023                                      struct lu_attr *attr,
2024                                      const struct lu_buf *lum_buf,
2025                                      struct dt_object_format *dof,
2026                                      struct thandle *th)
2027 {
2028         struct lod_object       *lo = lod_dt_obj(dt);
2029         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
2030         struct lmv_user_md_v1   *lum;
2031         int                     rc;
2032         ENTRY;
2033
2034         lum = lum_buf->lb_buf;
2035         LASSERT(lum != NULL);
2036
2037         CDEBUG(D_INFO, "lum magic = %x count = %u offset = %d\n",
2038                le32_to_cpu(lum->lum_magic), le32_to_cpu(lum->lum_stripe_count),
2039                (int)le32_to_cpu(lum->lum_stripe_offset));
2040
2041         if (le32_to_cpu(lum->lum_stripe_count) == 0)
2042                 GOTO(out, rc = 0);
2043
2044         rc = lod_verify_md_striping(lod, lum);
2045         if (rc != 0)
2046                 GOTO(out, rc);
2047
2048         /* prepare dir striped objects */
2049         rc = lod_prep_md_striped_create(env, dt, attr, lum, dof, th);
2050         if (rc != 0) {
2051                 /* failed to create striping, let's reset
2052                  * config so that others don't get confused */
2053                 lod_object_free_striping(env, lo);
2054                 GOTO(out, rc);
2055         }
2056 out:
2057         RETURN(rc);
2058 }
2059
2060 /**
2061  * Implementation of dt_object_operations::do_declare_xattr_set.
2062  *
2063  * Used with regular (non-striped) objects. Basically it
2064  * initializes the striping information and applies the
2065  * change to all the stripes.
2066  *
2067  * \see dt_object_operations::do_declare_xattr_set() in the API description
2068  * for details.
2069  */
2070 static int lod_dir_declare_xattr_set(const struct lu_env *env,
2071                                      struct dt_object *dt,
2072                                      const struct lu_buf *buf,
2073                                      const char *name, int fl,
2074                                      struct thandle *th)
2075 {
2076         struct dt_object        *next = dt_object_child(dt);
2077         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2078         struct lod_object       *lo = lod_dt_obj(dt);
2079         int                     i;
2080         int                     rc;
2081         ENTRY;
2082
2083         if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
2084                 struct lmv_user_md_v1 *lum;
2085
2086                 LASSERT(buf != NULL && buf->lb_buf != NULL);
2087                 lum = buf->lb_buf;
2088                 rc = lod_verify_md_striping(d, lum);
2089                 if (rc != 0)
2090                         RETURN(rc);
2091         } else if (strcmp(name, XATTR_NAME_LOV) == 0) {
2092                 rc = lod_verify_striping(d, lo, buf, false);
2093                 if (rc != 0)
2094                         RETURN(rc);
2095         }
2096
2097         rc = lod_sub_declare_xattr_set(env, next, buf, name, fl, th);
2098         if (rc != 0)
2099                 RETURN(rc);
2100
2101         /* Note: Do not set LinkEA on sub-stripes, otherwise
2102          * it will confuse the fid2path process(see mdt_path_current()).
2103          * The linkEA between master and sub-stripes is set in
2104          * lod_xattr_set_lmv(). */
2105         if (strcmp(name, XATTR_NAME_LINK) == 0)
2106                 RETURN(0);
2107
2108         /* set xattr to each stripes, if needed */
2109         rc = lod_load_striping(env, lo);
2110         if (rc != 0)
2111                 RETURN(rc);
2112
2113         if (lo->ldo_dir_stripe_count == 0)
2114                 RETURN(0);
2115
2116         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
2117                 LASSERT(lo->ldo_stripe[i]);
2118
2119                 rc = lod_sub_declare_xattr_set(env, lo->ldo_stripe[i],
2120                                                buf, name, fl, th);
2121                 if (rc != 0)
2122                         break;
2123         }
2124
2125         RETURN(rc);
2126 }
2127
2128 static int
2129 lod_obj_stripe_replace_parent_fid_cb(const struct lu_env *env,
2130                                      struct lod_object *lo,
2131                                      struct dt_object *dt, struct thandle *th,
2132                                      int comp_idx, int stripe_idx,
2133                                      struct lod_obj_stripe_cb_data *data)
2134 {
2135         struct lod_thread_info *info = lod_env_info(env);
2136         struct filter_fid *ff = &info->lti_ff;
2137         struct lu_buf *buf = &info->lti_buf;
2138         int rc;
2139
2140         buf->lb_buf = ff;
2141         buf->lb_len = sizeof(*ff);
2142         rc = dt_xattr_get(env, dt, buf, XATTR_NAME_FID);
2143         if (rc == -ENODATA)
2144                 return 0;
2145
2146         if (rc < 0)
2147                 return rc;
2148
2149         ff->ff_parent = *lu_object_fid(&lo->ldo_obj.do_lu);
2150         ff->ff_parent.f_ver = stripe_idx;
2151         fid_cpu_to_le(&ff->ff_parent, &ff->ff_parent);
2152         if (data->locd_declare)
2153                 rc = lod_sub_declare_xattr_set(env, dt, buf, XATTR_NAME_FID,
2154                                                LU_XATTR_REPLACE, th);
2155         else
2156                 rc = lod_sub_xattr_set(env, dt, buf, XATTR_NAME_FID,
2157                                        LU_XATTR_REPLACE, th);
2158
2159         return rc;
2160 }
2161
2162 /**
2163  * Reset parent FID on OST object
2164  *
2165  * Replace parent FID with @dt object FID, which is only called during migration
2166  * to reset the parent FID after the MDT object is migrated to the new MDT, i.e.
2167  * the FID is changed.
2168  *
2169  * \param[in] env execution environment
2170  * \param[in] dt dt_object whose stripes's parent FID will be reset
2171  * \parem[in] th thandle
2172  * \param[in] declare if it is declare
2173  *
2174  * \retval      0 if reset succeeds
2175  * \retval      negative errno if reset fails
2176  */
2177 static int lod_replace_parent_fid(const struct lu_env *env,
2178                                   struct dt_object *dt,
2179                                   struct thandle *th, bool declare)
2180 {
2181         struct lod_object *lo = lod_dt_obj(dt);
2182         struct lod_thread_info  *info = lod_env_info(env);
2183         struct lu_buf *buf = &info->lti_buf;
2184         struct filter_fid *ff;
2185         struct lod_obj_stripe_cb_data data = { { 0 } };
2186         int rc;
2187         ENTRY;
2188
2189         LASSERT(S_ISREG(dt->do_lu.lo_header->loh_attr));
2190
2191         /* set xattr to each stripes, if needed */
2192         rc = lod_load_striping(env, lo);
2193         if (rc != 0)
2194                 RETURN(rc);
2195
2196         if (!lod_obj_is_striped(dt))
2197                 RETURN(0);
2198
2199         if (info->lti_ea_store_size < sizeof(*ff)) {
2200                 rc = lod_ea_store_resize(info, sizeof(*ff));
2201                 if (rc != 0)
2202                         RETURN(rc);
2203         }
2204
2205         buf->lb_buf = info->lti_ea_store;
2206         buf->lb_len = info->lti_ea_store_size;
2207
2208         data.locd_declare = declare;
2209         data.locd_stripe_cb = lod_obj_stripe_replace_parent_fid_cb;
2210         rc = lod_obj_for_each_stripe(env, lo, th, &data);
2211
2212         RETURN(rc);
2213 }
2214
2215 inline __u16 lod_comp_entry_stripe_count(struct lod_object *lo,
2216                                          struct lod_layout_component *entry,
2217                                          bool is_dir)
2218 {
2219         struct lod_device *lod = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
2220
2221         if (is_dir)
2222                 return  0;
2223         else if (lod_comp_inited(entry))
2224                 return entry->llc_stripe_count;
2225         else if ((__u16)-1 == entry->llc_stripe_count)
2226                 return lod->lod_desc.ld_tgt_count;
2227         else
2228                 return lod_get_stripe_count(lod, lo, entry->llc_stripe_count);
2229 }
2230
2231 static int lod_comp_md_size(struct lod_object *lo, bool is_dir)
2232 {
2233         int magic, size = 0, i;
2234         struct lod_layout_component *comp_entries;
2235         __u16 comp_cnt;
2236         bool is_composite;
2237
2238         if (is_dir) {
2239                 comp_cnt = lo->ldo_def_striping->lds_def_comp_cnt;
2240                 comp_entries = lo->ldo_def_striping->lds_def_comp_entries;
2241                 is_composite =
2242                         lo->ldo_def_striping->lds_def_striping_is_composite;
2243         } else {
2244                 comp_cnt = lo->ldo_comp_cnt;
2245                 comp_entries = lo->ldo_comp_entries;
2246                 is_composite = lo->ldo_is_composite;
2247         }
2248
2249
2250         LASSERT(comp_cnt != 0 && comp_entries != NULL);
2251         if (is_composite) {
2252                 size = sizeof(struct lov_comp_md_v1) +
2253                        sizeof(struct lov_comp_md_entry_v1) * comp_cnt;
2254                 LASSERT(size % sizeof(__u64) == 0);
2255         }
2256
2257         for (i = 0; i < comp_cnt; i++) {
2258                 __u16 stripe_count;
2259
2260                 magic = comp_entries[i].llc_pool ? LOV_MAGIC_V3 : LOV_MAGIC_V1;
2261                 stripe_count = lod_comp_entry_stripe_count(lo, &comp_entries[i],
2262                                                            is_dir);
2263                 if (!is_dir && is_composite)
2264                         lod_comp_shrink_stripe_count(&comp_entries[i],
2265                                                      &stripe_count);
2266
2267                 size += lov_user_md_size(stripe_count, magic);
2268                 LASSERT(size % sizeof(__u64) == 0);
2269         }
2270         return size;
2271 }
2272
2273 /**
2274  * Declare component add. The xattr name is XATTR_LUSTRE_LOV.add, and
2275  * the xattr value is binary lov_comp_md_v1 which contains component(s)
2276  * to be added.
2277   *
2278  * \param[in] env       execution environment
2279  * \param[in] dt        dt_object to add components on
2280  * \param[in] buf       buffer contains components to be added
2281  * \parem[in] th        thandle
2282  *
2283  * \retval      0 on success
2284  * \retval      negative errno on failure
2285  */
2286 static int lod_declare_layout_add(const struct lu_env *env,
2287                                   struct dt_object *dt,
2288                                   const struct lu_buf *buf,
2289                                   struct thandle *th)
2290 {
2291         struct lod_thread_info  *info = lod_env_info(env);
2292         struct lod_layout_component *comp_array, *lod_comp, *old_array;
2293         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2294         struct dt_object *next = dt_object_child(dt);
2295         struct lov_desc         *desc = &d->lod_desc;
2296         struct lod_object       *lo = lod_dt_obj(dt);
2297         struct lov_user_md_v3   *v3;
2298         struct lov_comp_md_v1   *comp_v1 = buf->lb_buf;
2299         __u32   magic;
2300         int     i, rc, array_cnt, old_array_cnt;
2301         ENTRY;
2302
2303         LASSERT(lo->ldo_is_composite);
2304
2305         if (lo->ldo_flr_state != LCM_FL_NOT_FLR)
2306                 RETURN(-EBUSY);
2307
2308         rc = lod_verify_striping(d, lo, buf, false);
2309         if (rc != 0)
2310                 RETURN(rc);
2311
2312         magic = comp_v1->lcm_magic;
2313         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
2314                 lustre_swab_lov_comp_md_v1(comp_v1);
2315                 magic = comp_v1->lcm_magic;
2316         }
2317
2318         if (magic != LOV_USER_MAGIC_COMP_V1)
2319                 RETURN(-EINVAL);
2320
2321         array_cnt = lo->ldo_comp_cnt + comp_v1->lcm_entry_count;
2322         OBD_ALLOC(comp_array, sizeof(*comp_array) * array_cnt);
2323         if (comp_array == NULL)
2324                 RETURN(-ENOMEM);
2325
2326         memcpy(comp_array, lo->ldo_comp_entries,
2327                sizeof(*comp_array) * lo->ldo_comp_cnt);
2328
2329         for (i = 0; i < comp_v1->lcm_entry_count; i++) {
2330                 struct lov_user_md_v1 *v1;
2331                 struct lu_extent *ext;
2332
2333                 v1 = (struct lov_user_md *)((char *)comp_v1 +
2334                                 comp_v1->lcm_entries[i].lcme_offset);
2335                 ext = &comp_v1->lcm_entries[i].lcme_extent;
2336
2337                 lod_comp = &comp_array[lo->ldo_comp_cnt + i];
2338                 lod_comp->llc_extent.e_start = ext->e_start;
2339                 lod_comp->llc_extent.e_end = ext->e_end;
2340                 lod_comp->llc_stripe_offset = v1->lmm_stripe_offset;
2341                 lod_comp->llc_flags = comp_v1->lcm_entries[i].lcme_flags;
2342
2343                 lod_comp->llc_stripe_count = v1->lmm_stripe_count;
2344                 if (!lod_comp->llc_stripe_count ||
2345                     lod_comp->llc_stripe_count == (__u16)-1)
2346                         lod_comp->llc_stripe_count =
2347                                 desc->ld_default_stripe_count;
2348                 lod_comp->llc_stripe_size = v1->lmm_stripe_size;
2349                 if (!lod_comp->llc_stripe_size)
2350                         lod_comp->llc_stripe_size =
2351                                 desc->ld_default_stripe_size;
2352
2353                 if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
2354                         v3 = (struct lov_user_md_v3 *) v1;
2355                         if (v3->lmm_pool_name[0] != '\0') {
2356                                 rc = lod_set_pool(&lod_comp->llc_pool,
2357                                                   v3->lmm_pool_name);
2358                                 if (rc)
2359                                         GOTO(error, rc);
2360                         }
2361                 }
2362         }
2363
2364         old_array = lo->ldo_comp_entries;
2365         old_array_cnt = lo->ldo_comp_cnt;
2366
2367         lo->ldo_comp_entries = comp_array;
2368         lo->ldo_comp_cnt = array_cnt;
2369
2370         /* No need to increase layout generation here, it will be increased
2371          * later when generating component ID for the new components */
2372
2373         info->lti_buf.lb_len = lod_comp_md_size(lo, false);
2374         rc = lod_sub_declare_xattr_set(env, next, &info->lti_buf,
2375                                               XATTR_NAME_LOV, 0, th);
2376         if (rc) {
2377                 lo->ldo_comp_entries = old_array;
2378                 lo->ldo_comp_cnt = old_array_cnt;
2379                 GOTO(error, rc);
2380         }
2381
2382         OBD_FREE(old_array, sizeof(*lod_comp) * old_array_cnt);
2383
2384         LASSERT(lo->ldo_mirror_count == 1);
2385         lo->ldo_mirrors[0].lme_end = array_cnt - 1;
2386
2387         RETURN(0);
2388
2389 error:
2390         for (i = lo->ldo_comp_cnt; i < array_cnt; i++) {
2391                 lod_comp = &comp_array[i];
2392                 if (lod_comp->llc_pool != NULL) {
2393                         OBD_FREE(lod_comp->llc_pool,
2394                                  strlen(lod_comp->llc_pool) + 1);
2395                         lod_comp->llc_pool = NULL;
2396                 }
2397         }
2398         OBD_FREE(comp_array, sizeof(*comp_array) * array_cnt);
2399         RETURN(rc);
2400 }
2401
2402 /**
2403  * Declare component set. The xattr is name XATTR_LUSTRE_LOV.set.$field,
2404  * the '$field' can only be 'flags' now. The xattr value is binary
2405  * lov_comp_md_v1 which contains the component ID(s) and the value of
2406  * the field to be modified.
2407  *
2408  * \param[in] env       execution environment
2409  * \param[in] dt        dt_object to be modified
2410  * \param[in] op        operation string, like "set.flags"
2411  * \param[in] buf       buffer contains components to be set
2412  * \parem[in] th        thandle
2413  *
2414  * \retval      0 on success
2415  * \retval      negative errno on failure
2416  */
2417 static int lod_declare_layout_set(const struct lu_env *env,
2418                                   struct dt_object *dt,
2419                                   char *op, const struct lu_buf *buf,
2420                                   struct thandle *th)
2421 {
2422         struct lod_layout_component     *lod_comp;
2423         struct lod_thread_info  *info = lod_env_info(env);
2424         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2425         struct lod_object       *lo = lod_dt_obj(dt);
2426         struct lov_comp_md_v1   *comp_v1 = buf->lb_buf;
2427         __u32   magic, id;
2428         int     i, j, rc;
2429         bool    changed = false;
2430         ENTRY;
2431
2432         if (strcmp(op, "set.flags") != 0) {
2433                 CDEBUG(D_LAYOUT, "%s: operation (%s) not supported.\n",
2434                        lod2obd(d)->obd_name, op);
2435                 RETURN(-ENOTSUPP);
2436         }
2437
2438         magic = comp_v1->lcm_magic;
2439         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
2440                 lustre_swab_lov_comp_md_v1(comp_v1);
2441                 magic = comp_v1->lcm_magic;
2442         }
2443
2444         if (magic != LOV_USER_MAGIC_COMP_V1)
2445                 RETURN(-EINVAL);
2446
2447         if (comp_v1->lcm_entry_count == 0) {
2448                 CDEBUG(D_LAYOUT, "%s: entry count is zero.\n",
2449                        lod2obd(d)->obd_name);
2450                 RETURN(-EINVAL);
2451         }
2452
2453         for (i = 0; i < comp_v1->lcm_entry_count; i++) {
2454                 id = comp_v1->lcm_entries[i].lcme_id;
2455
2456                 for (j = 0; j < lo->ldo_comp_cnt; j++) {
2457                         lod_comp = &lo->ldo_comp_entries[j];
2458                         if (id == lod_comp->llc_id || id == LCME_ID_ALL) {
2459                                 lod_comp->llc_flags =
2460                                         comp_v1->lcm_entries[i].lcme_flags;
2461                                 changed = true;
2462                         }
2463                 }
2464         }
2465
2466         if (!changed) {
2467                 CDEBUG(D_LAYOUT, "%s: requested component(s) not found.\n",
2468                        lod2obd(d)->obd_name);
2469                 RETURN(-EINVAL);
2470         }
2471
2472         lod_obj_inc_layout_gen(lo);
2473
2474         info->lti_buf.lb_len = lod_comp_md_size(lo, false);
2475         rc = lod_sub_declare_xattr_set(env, dt, &info->lti_buf,
2476                                        XATTR_NAME_LOV, 0, th);
2477         RETURN(rc);
2478 }
2479
2480 /**
2481  * Declare component deletion. The xattr name is XATTR_LUSTRE_LOV.del,
2482  * and the xattr value is a unique component ID or a special lcme_id.
2483  *
2484  * \param[in] env       execution environment
2485  * \param[in] dt        dt_object to be operated on
2486  * \param[in] buf       buffer contains component ID or lcme_id
2487  * \parem[in] th        thandle
2488  *
2489  * \retval      0 on success
2490  * \retval      negative errno on failure
2491  */
2492 static int lod_declare_layout_del(const struct lu_env *env,
2493                                   struct dt_object *dt,
2494                                   const struct lu_buf *buf,
2495                                   struct thandle *th)
2496 {
2497         struct lod_thread_info  *info = lod_env_info(env);
2498         struct dt_object *next = dt_object_child(dt);
2499         struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
2500         struct lod_object *lo = lod_dt_obj(dt);
2501         struct lu_attr *attr = &lod_env_info(env)->lti_attr;
2502         struct lov_comp_md_v1 *comp_v1 = buf->lb_buf;
2503         __u32 magic, id, flags, neg_flags = 0;
2504         int rc, i, j, left;
2505         ENTRY;
2506
2507         LASSERT(lo->ldo_is_composite);
2508
2509         if (lo->ldo_flr_state != LCM_FL_NOT_FLR)
2510                 RETURN(-EBUSY);
2511
2512         magic = comp_v1->lcm_magic;
2513         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
2514                 lustre_swab_lov_comp_md_v1(comp_v1);
2515                 magic = comp_v1->lcm_magic;
2516         }
2517
2518         if (magic != LOV_USER_MAGIC_COMP_V1)
2519                 RETURN(-EINVAL);
2520
2521         id = comp_v1->lcm_entries[0].lcme_id;
2522         flags = comp_v1->lcm_entries[0].lcme_flags;
2523
2524         if (id > LCME_ID_MAX || (flags & ~LCME_KNOWN_FLAGS)) {
2525                 CDEBUG(D_LAYOUT, "%s: invalid component id %#x, flags %#x\n",
2526                        lod2obd(d)->obd_name, id, flags);
2527                 RETURN(-EINVAL);
2528         }
2529
2530         if (id != LCME_ID_INVAL && flags != 0) {
2531                 CDEBUG(D_LAYOUT, "%s: specified both id and flags.\n",
2532                        lod2obd(d)->obd_name);
2533                 RETURN(-EINVAL);
2534         }
2535
2536         if (flags & LCME_FL_NEG) {
2537                 neg_flags = flags & ~LCME_FL_NEG;
2538                 flags = 0;
2539         }
2540
2541         left = lo->ldo_comp_cnt;
2542         if (left <= 0)
2543                 RETURN(-EINVAL);
2544
2545         for (i = (lo->ldo_comp_cnt - 1); i >= 0; i--) {
2546                 struct lod_layout_component *lod_comp;
2547
2548                 lod_comp = &lo->ldo_comp_entries[i];
2549
2550                 if (id != LCME_ID_INVAL && id != lod_comp->llc_id)
2551                         continue;
2552                 else if (flags && !(flags & lod_comp->llc_flags))
2553                         continue;
2554                 else if (neg_flags && (neg_flags & lod_comp->llc_flags))
2555                         continue;
2556
2557                 if (left != (i + 1)) {
2558                         CDEBUG(D_LAYOUT, "%s: this deletion will create "
2559                                "a hole.\n", lod2obd(d)->obd_name);
2560                         RETURN(-EINVAL);
2561                 }
2562                 left--;
2563
2564                 /* Mark the component as deleted */
2565                 lod_comp->llc_id = LCME_ID_INVAL;
2566
2567                 /* Not instantiated component */
2568                 if (lod_comp->llc_stripe == NULL)
2569                         continue;
2570
2571                 LASSERT(lod_comp->llc_stripe_count > 0);
2572                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
2573                         struct dt_object *obj = lod_comp->llc_stripe[j];
2574
2575                         if (obj == NULL)
2576                                 continue;
2577                         rc = lod_sub_declare_destroy(env, obj, th);
2578                         if (rc)
2579                                 RETURN(rc);
2580                 }
2581         }
2582
2583         LASSERTF(left >= 0, "left = %d\n", left);
2584         if (left == lo->ldo_comp_cnt) {
2585                 CDEBUG(D_LAYOUT, "%s: requested component id:%#x not found\n",
2586                        lod2obd(d)->obd_name, id);
2587                 RETURN(-EINVAL);
2588         }
2589
2590         memset(attr, 0, sizeof(*attr));
2591         attr->la_valid = LA_SIZE;
2592         rc = lod_sub_declare_attr_set(env, next, attr, th);
2593         if (rc)
2594                 RETURN(rc);
2595
2596         if (left > 0) {
2597                 info->lti_buf.lb_len = lod_comp_md_size(lo, false);
2598                 rc = lod_sub_declare_xattr_set(env, next, &info->lti_buf,
2599                                                XATTR_NAME_LOV, 0, th);
2600         } else {
2601                 rc = lod_sub_declare_xattr_del(env, next, XATTR_NAME_LOV, th);
2602         }
2603
2604         RETURN(rc);
2605 }
2606
2607 /**
2608  * Declare layout add/set/del operations issued by special xattr names:
2609  *
2610  * XATTR_LUSTRE_LOV.add         add component(s) to existing file
2611  * XATTR_LUSTRE_LOV.del         delete component(s) from existing file
2612  * XATTR_LUSTRE_LOV.set.$field  set specified field of certain component(s)
2613  *
2614  * \param[in] env       execution environment
2615  * \param[in] dt        object
2616  * \param[in] name      name of xattr
2617  * \param[in] buf       lu_buf contains xattr value
2618  * \param[in] th        transaction handle
2619  *
2620  * \retval              0 on success
2621  * \retval              negative if failed
2622  */
2623 static int lod_declare_modify_layout(const struct lu_env *env,
2624                                      struct dt_object *dt,
2625                                      const char *name,
2626                                      const struct lu_buf *buf,
2627                                      struct thandle *th)
2628 {
2629         struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
2630         struct lod_object *lo = lod_dt_obj(dt);
2631         struct dt_object *next = dt_object_child(&lo->ldo_obj);
2632         char *op;
2633         int rc, len = strlen(XATTR_LUSTRE_LOV);
2634         ENTRY;
2635
2636         LASSERT(dt_object_exists(dt));
2637
2638         if (strlen(name) <= len || name[len] != '.') {
2639                 CDEBUG(D_LAYOUT, "%s: invalid xattr name: %s\n",
2640                        lod2obd(d)->obd_name, name);
2641                 RETURN(-EINVAL);
2642         }
2643         len++;
2644
2645         dt_write_lock(env, next, 0);
2646         rc = lod_load_striping_locked(env, lo);
2647         if (rc)
2648                 GOTO(unlock, rc);
2649
2650         /* the layout to be modified must be a composite layout */
2651         if (!lo->ldo_is_composite) {
2652                 CDEBUG(D_LAYOUT, "%s: object "DFID" isn't a composite file.\n",
2653                        lod2obd(d)->obd_name, PFID(lu_object_fid(&dt->do_lu)));
2654                 GOTO(unlock, rc = -EINVAL);
2655         }
2656
2657         op = (char *)name + len;
2658         if (strcmp(op, "add") == 0) {
2659                 rc = lod_declare_layout_add(env, dt, buf, th);
2660         } else if (strcmp(op, "del") == 0) {
2661                 rc = lod_declare_layout_del(env, dt, buf, th);
2662         } else if (strncmp(op, "set", strlen("set")) == 0) {
2663                 rc = lod_declare_layout_set(env, dt, op, buf, th);
2664         } else  {
2665                 CDEBUG(D_LAYOUT, "%s: unsupported xattr name:%s\n",
2666                        lod2obd(d)->obd_name, name);
2667                 GOTO(unlock, rc = -ENOTSUPP);
2668         }
2669 unlock:
2670         if (rc)
2671                 lod_object_free_striping(env, lo);
2672         dt_write_unlock(env, next);
2673
2674         RETURN(rc);
2675 }
2676
2677 /**
2678  * Merge layouts to form a mirrored file.
2679  */
2680 static int lod_declare_layout_merge(const struct lu_env *env,
2681                 struct dt_object *dt, const struct lu_buf *mbuf,
2682                 struct thandle *th)
2683 {
2684         struct lod_thread_info  *info = lod_env_info(env);
2685         struct lu_buf           *buf = &info->lti_buf;
2686         struct lod_object       *lo = lod_dt_obj(dt);
2687         struct lov_comp_md_v1   *lcm;
2688         struct lov_comp_md_v1   *cur_lcm;
2689         struct lov_comp_md_v1   *merge_lcm;
2690         struct lov_comp_md_entry_v1     *lcme;
2691         size_t size = 0;
2692         size_t offset;
2693         __u16 cur_entry_count;
2694         __u16 merge_entry_count;
2695         __u32 id = 0;
2696         __u16 mirror_id = 0;
2697         __u32 mirror_count;
2698         int     rc, i;
2699         ENTRY;
2700
2701         merge_lcm = mbuf->lb_buf;
2702         if (mbuf->lb_len < sizeof(*merge_lcm))
2703                 RETURN(-EINVAL);
2704
2705         /* must be an existing layout from disk */
2706         if (le32_to_cpu(merge_lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
2707                 RETURN(-EINVAL);
2708
2709         merge_entry_count = le16_to_cpu(merge_lcm->lcm_entry_count);
2710
2711         /* do not allow to merge two mirrored files */
2712         if (le16_to_cpu(merge_lcm->lcm_mirror_count))
2713                 RETURN(-EBUSY);
2714
2715         /* verify the target buffer */
2716         rc = lod_get_lov_ea(env, lo);
2717         if (rc <= 0)
2718                 RETURN(rc ? : -ENODATA);
2719
2720         cur_lcm = info->lti_ea_store;
2721         if (le32_to_cpu(cur_lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
2722                 RETURN(-EINVAL);
2723
2724         cur_entry_count = le16_to_cpu(cur_lcm->lcm_entry_count);
2725
2726         /* 'lcm_mirror_count + 1' is the current # of mirrors the file has */
2727         mirror_count = le16_to_cpu(cur_lcm->lcm_mirror_count) + 1;
2728         if (mirror_count + 1 > LUSTRE_MIRROR_COUNT_MAX)
2729                 RETURN(-ERANGE);
2730
2731         /* size of new layout */
2732         size = le32_to_cpu(cur_lcm->lcm_size) +
2733                le32_to_cpu(merge_lcm->lcm_size) - sizeof(*cur_lcm);
2734
2735         memset(buf, 0, sizeof(*buf));
2736         lu_buf_alloc(buf, size);
2737         if (buf->lb_buf == NULL)
2738                 RETURN(-ENOMEM);
2739
2740         lcm = buf->lb_buf;
2741         memcpy(lcm, cur_lcm, sizeof(*lcm) + cur_entry_count * sizeof(*lcme));
2742
2743         offset = sizeof(*lcm) +
2744                  sizeof(*lcme) * (cur_entry_count + merge_entry_count);
2745         for (i = 0; i < cur_entry_count; i++) {
2746                 struct lov_comp_md_entry_v1 *cur_lcme;
2747
2748                 lcme = &lcm->lcm_entries[i];
2749                 cur_lcme = &cur_lcm->lcm_entries[i];
2750
2751                 lcme->lcme_offset = cpu_to_le32(offset);
2752                 memcpy((char *)lcm + offset,
2753                        (char *)cur_lcm + le32_to_cpu(cur_lcme->lcme_offset),
2754                        le32_to_cpu(lcme->lcme_size));
2755
2756                 offset += le32_to_cpu(lcme->lcme_size);
2757
2758                 if (mirror_count == 1) {
2759                         /* new mirrored file, create new mirror ID */
2760                         id = pflr_id(1, i + 1);
2761                         lcme->lcme_id = cpu_to_le32(id);
2762                 }
2763
2764                 id = MAX(le32_to_cpu(lcme->lcme_id), id);
2765         }
2766
2767         mirror_id = mirror_id_of(id) + 1;
2768         for (i = 0; i < merge_entry_count; i++) {
2769                 struct lov_comp_md_entry_v1 *merge_lcme;
2770
2771                 merge_lcme = &merge_lcm->lcm_entries[i];
2772                 lcme = &lcm->lcm_entries[cur_entry_count + i];
2773
2774                 *lcme = *merge_lcme;
2775                 lcme->lcme_offset = cpu_to_le32(offset);
2776
2777                 id = pflr_id(mirror_id, i + 1);
2778                 lcme->lcme_id = cpu_to_le32(id);
2779
2780                 memcpy((char *)lcm + offset,
2781                        (char *)merge_lcm + le32_to_cpu(merge_lcme->lcme_offset),
2782                        le32_to_cpu(lcme->lcme_size));
2783
2784                 offset += le32_to_cpu(lcme->lcme_size);
2785         }
2786
2787         /* fixup layout information */
2788         lod_obj_inc_layout_gen(lo);
2789         lcm->lcm_layout_gen = cpu_to_le32(lo->ldo_layout_gen);
2790         lcm->lcm_size = cpu_to_le32(size);
2791         lcm->lcm_entry_count = cpu_to_le16(cur_entry_count + merge_entry_count);
2792         lcm->lcm_mirror_count = cpu_to_le16(mirror_count);
2793         if ((le16_to_cpu(lcm->lcm_flags) & LCM_FL_FLR_MASK) == LCM_FL_NOT_FLR)
2794                 lcm->lcm_flags = cpu_to_le32(LCM_FL_RDONLY);
2795
2796         LASSERT(dt_write_locked(env, dt_object_child(dt)));
2797         lod_object_free_striping(env, lo);
2798         rc = lod_parse_striping(env, lo, buf);
2799         if (rc)
2800                 GOTO(out, rc);
2801
2802         rc = lod_sub_declare_xattr_set(env, dt_object_child(dt), buf,
2803                                         XATTR_NAME_LOV, LU_XATTR_REPLACE, th);
2804
2805 out:
2806         lu_buf_free(buf);
2807         RETURN(rc);
2808 }
2809
2810 /**
2811  * Implementation of dt_object_operations::do_declare_xattr_set.
2812  *
2813  * \see dt_object_operations::do_declare_xattr_set() in the API description
2814  * for details.
2815  *
2816  * the extension to the API:
2817  *   - declaring LOVEA requests striping creation
2818  *   - LU_XATTR_REPLACE means layout swap
2819  */
2820 static int lod_declare_xattr_set(const struct lu_env *env,
2821                                  struct dt_object *dt,
2822                                  const struct lu_buf *buf,
2823                                  const char *name, int fl,
2824                                  struct thandle *th)
2825 {
2826         struct dt_object *next = dt_object_child(dt);
2827         struct lu_attr   *attr = &lod_env_info(env)->lti_attr;
2828         __u32             mode;
2829         int               rc;
2830         ENTRY;
2831
2832         mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
2833         if ((S_ISREG(mode) || mode == 0) &&
2834             !(fl & (LU_XATTR_REPLACE | LU_XATTR_MERGE)) &&
2835             (strcmp(name, XATTR_NAME_LOV) == 0 ||
2836              strcmp(name, XATTR_LUSTRE_LOV) == 0)) {
2837                 /*
2838                  * this is a request to create object's striping.
2839                  *
2840                  * allow to declare predefined striping on a new (!mode) object
2841                  * which is supposed to be replay of regular file creation
2842                  * (when LOV setting is declared)
2843                  *
2844                  * LU_XATTR_REPLACE is set to indicate a layout swap
2845                  */
2846                 if (dt_object_exists(dt)) {
2847                         rc = dt_attr_get(env, next, attr);
2848                         if (rc)
2849                                 RETURN(rc);
2850                 } else {
2851                         memset(attr, 0, sizeof(*attr));
2852                         attr->la_valid = LA_TYPE | LA_MODE;
2853                         attr->la_mode = S_IFREG;
2854                 }
2855                 rc = lod_declare_striped_create(env, dt, attr, buf, th);
2856         } else if (fl & LU_XATTR_MERGE) {
2857                 LASSERT(strcmp(name, XATTR_NAME_LOV) == 0 ||
2858                         strcmp(name, XATTR_LUSTRE_LOV) == 0);
2859                 rc = lod_declare_layout_merge(env, dt, buf, th);
2860         } else if (S_ISREG(mode) &&
2861                    strlen(name) > strlen(XATTR_LUSTRE_LOV) + 1 &&
2862                    strncmp(name, XATTR_LUSTRE_LOV,
2863                            strlen(XATTR_LUSTRE_LOV)) == 0) {
2864                 /*
2865                  * this is a request to modify object's striping.
2866                  * add/set/del component(s).
2867                  */
2868                 if (!dt_object_exists(dt))
2869                         RETURN(-ENOENT);
2870
2871                 rc = lod_declare_modify_layout(env, dt, name, buf, th);
2872         } else if (S_ISDIR(mode)) {
2873                 rc = lod_dir_declare_xattr_set(env, dt, buf, name, fl, th);
2874         } else if (strcmp(name, XATTR_NAME_FID) == 0) {
2875                 rc = lod_replace_parent_fid(env, dt, th, true);
2876         } else {
2877                 rc = lod_sub_declare_xattr_set(env, next, buf, name, fl, th);
2878         }
2879
2880         RETURN(rc);
2881 }
2882
2883 /**
2884  * Apply xattr changes to the object.
2885  *
2886  * Applies xattr changes to the object and the stripes if the latter exist.
2887  *
2888  * \param[in] env       execution environment
2889  * \param[in] dt        object
2890  * \param[in] buf       buffer pointing to the new value of xattr
2891  * \param[in] name      name of xattr
2892  * \param[in] fl        flags
2893  * \param[in] th        transaction handle
2894  *
2895  * \retval              0 on success
2896  * \retval              negative if failed
2897  */
2898 static int lod_xattr_set_internal(const struct lu_env *env,
2899                                   struct dt_object *dt,
2900                                   const struct lu_buf *buf,
2901                                   const char *name, int fl,
2902                                   struct thandle *th)
2903 {
2904         struct dt_object        *next = dt_object_child(dt);
2905         struct lod_object       *lo = lod_dt_obj(dt);
2906         int                     rc;
2907         int                     i;
2908         ENTRY;
2909
2910         rc = lod_sub_xattr_set(env, next, buf, name, fl, th);
2911         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
2912                 RETURN(rc);
2913
2914         /* Note: Do not set LinkEA on sub-stripes, otherwise
2915          * it will confuse the fid2path process(see mdt_path_current()).
2916          * The linkEA between master and sub-stripes is set in
2917          * lod_xattr_set_lmv(). */
2918         if (lo->ldo_dir_stripe_count == 0 || strcmp(name, XATTR_NAME_LINK) == 0)
2919                 RETURN(0);
2920
2921         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
2922                 LASSERT(lo->ldo_stripe[i]);
2923
2924                 rc = lod_sub_xattr_set(env, lo->ldo_stripe[i], buf, name,
2925                                        fl, th);
2926                 if (rc != 0)
2927                         break;
2928         }
2929
2930         RETURN(rc);
2931 }
2932
2933 /**
2934  * Delete an extended attribute.
2935  *
2936  * Deletes specified xattr from the object and the stripes if the latter exist.
2937  *
2938  * \param[in] env       execution environment
2939  * \param[in] dt        object
2940  * \param[in] name      name of xattr
2941  * \param[in] th        transaction handle
2942  *
2943  * \retval              0 on success
2944  * \retval              negative if failed
2945  */
2946 static int lod_xattr_del_internal(const struct lu_env *env,
2947                                   struct dt_object *dt,
2948                                   const char *name, struct thandle *th)
2949 {
2950         struct dt_object        *next = dt_object_child(dt);
2951         struct lod_object       *lo = lod_dt_obj(dt);
2952         int                     rc;
2953         int                     i;
2954         ENTRY;
2955
2956         rc = lod_sub_xattr_del(env, next, name, th);
2957         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
2958                 RETURN(rc);
2959
2960         if (lo->ldo_dir_stripe_count == 0)
2961                 RETURN(rc);
2962
2963         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
2964                 LASSERT(lo->ldo_stripe[i]);
2965
2966                 rc = lod_sub_xattr_del(env, lo->ldo_stripe[i], name, th);
2967                 if (rc != 0)
2968                         break;
2969         }
2970
2971         RETURN(rc);
2972 }
2973
2974 /**
2975  * Set default striping on a directory.
2976  *
2977  * Sets specified striping on a directory object unless it matches the default
2978  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
2979  * EA. This striping will be used when regular file is being created in this
2980  * directory.
2981  *
2982  * \param[in] env       execution environment
2983  * \param[in] dt        the striped object
2984  * \param[in] buf       buffer with the striping
2985  * \param[in] name      name of EA
2986  * \param[in] fl        xattr flag (see OSD API description)
2987  * \param[in] th        transaction handle
2988  *
2989  * \retval              0 on success
2990  * \retval              negative if failed
2991  */
2992 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
2993                                     struct dt_object *dt,
2994                                     const struct lu_buf *buf,
2995                                     const char *name, int fl,
2996                                     struct thandle *th)
2997 {
2998         struct lov_user_md_v1   *lum;
2999         struct lov_user_md_v3   *v3 = NULL;
3000         const char              *pool_name = NULL;
3001         int                      rc;
3002         bool                     is_del;
3003         ENTRY;
3004
3005         LASSERT(buf != NULL && buf->lb_buf != NULL);
3006         lum = buf->lb_buf;
3007
3008         switch (lum->lmm_magic) {
3009         case LOV_USER_MAGIC_V3:
3010                 v3 = buf->lb_buf;
3011                 if (v3->lmm_pool_name[0] != '\0')
3012                         pool_name = v3->lmm_pool_name;
3013                 /* fall through */
3014         case LOV_USER_MAGIC_V1:
3015                 /* if { size, offset, count } = { 0, -1, 0 } and no pool
3016                  * (i.e. all default values specified) then delete default
3017                  * striping from dir. */
3018                 CDEBUG(D_LAYOUT,
3019                        "set default striping: sz %u # %u offset %d %s %s\n",
3020                        (unsigned)lum->lmm_stripe_size,
3021                        (unsigned)lum->lmm_stripe_count,
3022                        (int)lum->lmm_stripe_offset,
3023                        v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
3024
3025                 is_del = LOVEA_DELETE_VALUES(lum->lmm_stripe_size,
3026                                              lum->lmm_stripe_count,
3027                                              lum->lmm_stripe_offset,
3028                                              pool_name);
3029                 break;
3030         case LOV_USER_MAGIC_COMP_V1:
3031                 is_del = false;
3032                 break;
3033         default:
3034                 CERROR("Invalid magic %x\n", lum->lmm_magic);
3035                 RETURN(-EINVAL);
3036         }
3037
3038         if (is_del) {
3039                 rc = lod_xattr_del_internal(env, dt, name, th);
3040                 if (rc == -ENODATA)
3041                         rc = 0;
3042         } else {
3043                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th);
3044         }
3045
3046         RETURN(rc);
3047 }
3048
3049 /**
3050  * Set default striping on a directory object.
3051  *
3052  * Sets specified striping on a directory object unless it matches the default
3053  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
3054  * EA. This striping will be used when a new directory is being created in the
3055  * directory.
3056  *
3057  * \param[in] env       execution environment
3058  * \param[in] dt        the striped object
3059  * \param[in] buf       buffer with the striping
3060  * \param[in] name      name of EA
3061  * \param[in] fl        xattr flag (see OSD API description)
3062  * \param[in] th        transaction handle
3063  *
3064  * \retval              0 on success
3065  * \retval              negative if failed
3066  */
3067 static int lod_xattr_set_default_lmv_on_dir(const struct lu_env *env,
3068                                             struct dt_object *dt,
3069                                             const struct lu_buf *buf,
3070                                             const char *name, int fl,
3071                                             struct thandle *th)
3072 {
3073         struct lmv_user_md_v1   *lum;
3074         int                      rc;
3075         ENTRY;
3076
3077         LASSERT(buf != NULL && buf->lb_buf != NULL);
3078         lum = buf->lb_buf;
3079
3080         CDEBUG(D_OTHER, "set default stripe_count # %u stripe_offset %d\n",
3081               le32_to_cpu(lum->lum_stripe_count),
3082               (int)le32_to_cpu(lum->lum_stripe_offset));
3083
3084         if (LMVEA_DELETE_VALUES((le32_to_cpu(lum->lum_stripe_count)),
3085                                  le32_to_cpu(lum->lum_stripe_offset)) &&
3086                                 le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC) {
3087                 rc = lod_xattr_del_internal(env, dt, name, th);
3088                 if (rc == -ENODATA)
3089                         rc = 0;
3090         } else {
3091                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th);
3092                 if (rc != 0)
3093                         RETURN(rc);
3094         }
3095
3096         RETURN(rc);
3097 }
3098
3099 /**
3100  * Turn directory into a striped directory.
3101  *
3102  * During replay the client sends the striping created before MDT
3103  * failure, then the layer above LOD sends this defined striping
3104  * using ->do_xattr_set(), so LOD uses this method to replay creation
3105  * of the stripes. Notice the original information for the striping
3106  * (#stripes, FIDs, etc) was transferred in declare path.
3107  *
3108  * \param[in] env       execution environment
3109  * \param[in] dt        the striped object
3110  * \param[in] buf       not used currently
3111  * \param[in] name      not used currently
3112  * \param[in] fl        xattr flag (see OSD API description)
3113  * \param[in] th        transaction handle
3114  *
3115  * \retval              0 on success
3116  * \retval              negative if failed
3117  */
3118 static int lod_xattr_set_lmv(const struct lu_env *env, struct dt_object *dt,
3119                              const struct lu_buf *buf, const char *name,
3120                              int fl, struct thandle *th)
3121 {
3122         struct lod_object       *lo = lod_dt_obj(dt);
3123         struct lod_thread_info  *info = lod_env_info(env);
3124         struct lu_attr          *attr = &info->lti_attr;
3125         struct dt_object_format *dof = &info->lti_format;
3126         struct lu_buf           lmv_buf;
3127         struct lu_buf           slave_lmv_buf;
3128         struct lmv_mds_md_v1    *lmm;
3129         struct lmv_mds_md_v1    *slave_lmm = NULL;
3130         struct dt_insert_rec    *rec = &info->lti_dt_rec;
3131         int                     i;
3132         int                     rc;
3133         ENTRY;
3134
3135         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
3136                 RETURN(-ENOTDIR);
3137
3138         /* The stripes are supposed to be allocated in declare phase,
3139          * if there are no stripes being allocated, it will skip */
3140         if (lo->ldo_dir_stripe_count == 0)
3141                 RETURN(0);
3142
3143         rc = dt_attr_get(env, dt_object_child(dt), attr);
3144         if (rc != 0)
3145                 RETURN(rc);
3146
3147         attr->la_valid = LA_ATIME | LA_MTIME | LA_CTIME |
3148                          LA_MODE | LA_UID | LA_GID | LA_TYPE | LA_PROJID;
3149         dof->dof_type = DFT_DIR;
3150
3151         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
3152         if (rc != 0)
3153                 RETURN(rc);
3154         lmm = lmv_buf.lb_buf;
3155
3156         OBD_ALLOC_PTR(slave_lmm);
3157         if (slave_lmm == NULL)
3158                 RETURN(-ENOMEM);
3159
3160         lod_prep_slave_lmv_md(slave_lmm, lmm);
3161         slave_lmv_buf.lb_buf = slave_lmm;
3162         slave_lmv_buf.lb_len = sizeof(*slave_lmm);
3163
3164         rec->rec_type = S_IFDIR;
3165         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
3166                 struct dt_object *dto;
3167                 char             *stripe_name = info->lti_key;
3168                 struct lu_name          *sname;
3169                 struct linkea_data       ldata          = { NULL };
3170                 struct lu_buf            linkea_buf;
3171
3172                 dto = lo->ldo_stripe[i];
3173
3174                 dt_write_lock(env, dto, MOR_TGT_CHILD);
3175                 rc = lod_sub_create(env, dto, attr, NULL, dof, th);
3176                 if (rc != 0) {
3177                         dt_write_unlock(env, dto);
3178                         GOTO(out, rc);
3179                 }
3180
3181                 rc = lod_sub_ref_add(env, dto, th);
3182                 dt_write_unlock(env, dto);
3183                 if (rc != 0)
3184                         GOTO(out, rc);
3185
3186                 rec->rec_fid = lu_object_fid(&dto->do_lu);
3187                 rc = lod_sub_insert(env, dto, (const struct dt_rec *)rec,
3188                                     (const struct dt_key *)dot, th, 0);
3189                 if (rc != 0)
3190                         GOTO(out, rc);
3191
3192                 rec->rec_fid = lu_object_fid(&dt->do_lu);
3193                 rc = lod_sub_insert(env, dto, (struct dt_rec *)rec,
3194                                     (const struct dt_key *)dotdot, th, 0);
3195                 if (rc != 0)
3196                         GOTO(out, rc);
3197
3198                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
3199                     cfs_fail_val != i) {
3200                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
3201                             cfs_fail_val == i)
3202                                 slave_lmm->lmv_master_mdt_index =
3203                                                         cpu_to_le32(i + 1);
3204                         else
3205                                 slave_lmm->lmv_master_mdt_index =
3206                                                         cpu_to_le32(i);
3207
3208                         rc = lod_sub_xattr_set(env, dto, &slave_lmv_buf,
3209                                                XATTR_NAME_LMV, fl, th);
3210                         if (rc != 0)
3211                                 GOTO(out, rc);
3212                 }
3213
3214                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
3215                     cfs_fail_val == i)
3216                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
3217                                  PFID(lu_object_fid(&dto->do_lu)), i + 1);
3218                 else
3219                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
3220                                  PFID(lu_object_fid(&dto->do_lu)), i);
3221
3222                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
3223                 rc = linkea_links_new(&ldata, &info->lti_linkea_buf,
3224                                       sname, lu_object_fid(&dt->do_lu));
3225                 if (rc != 0)
3226                         GOTO(out, rc);
3227
3228                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
3229                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
3230                 rc = lod_sub_xattr_set(env, dto, &linkea_buf,
3231                                        XATTR_NAME_LINK, 0, th);
3232                 if (rc != 0)
3233                         GOTO(out, rc);
3234
3235                 rec->rec_fid = lu_object_fid(&dto->do_lu);
3236                 rc = lod_sub_insert(env, dt_object_child(dt),
3237                                     (const struct dt_rec *)rec,
3238                                     (const struct dt_key *)stripe_name, th, 0);
3239                 if (rc != 0)
3240                         GOTO(out, rc);
3241
3242                 rc = lod_sub_ref_add(env, dt_object_child(dt), th);
3243                 if (rc != 0)
3244                         GOTO(out, rc);
3245         }
3246
3247         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MASTER_LMV))
3248                 rc = lod_sub_xattr_set(env, dt_object_child(dt),
3249                                        &lmv_buf, XATTR_NAME_LMV, fl, th);
3250 out:
3251         if (slave_lmm != NULL)
3252                 OBD_FREE_PTR(slave_lmm);
3253
3254         RETURN(rc);
3255 }
3256
3257 /**
3258  * Helper function to declare/execute creation of a striped directory
3259  *
3260  * Called in declare/create object path, prepare striping for a directory
3261  * and prepare defaults data striping for the objects to be created in
3262  * that directory. Notice the function calls "declaration" or "execution"
3263  * methods depending on \a declare param. This is a consequence of the
3264  * current approach while we don't have natural distributed transactions:
3265  * we basically execute non-local updates in the declare phase. So, the
3266  * arguments for the both phases are the same and this is the reason for
3267  * this function to exist.
3268  *
3269  * \param[in] env       execution environment
3270  * \param[in] dt        object
3271  * \param[in] attr      attributes the stripes will be created with
3272  * \param[in] dof       format of stripes (see OSD API description)
3273  * \param[in] th        transaction handle
3274  * \param[in] declare   where to call "declare" or "execute" methods
3275  *
3276  * \retval              0 on success
3277  * \retval              negative if failed
3278  */
3279 static int lod_dir_striping_create_internal(const struct lu_env *env,
3280                                             struct dt_object *dt,
3281                                             struct lu_attr *attr,
3282                                             struct dt_object_format *dof,
3283                                             struct thandle *th,
3284                                             bool declare)
3285 {
3286         struct lod_thread_info *info = lod_env_info(env);
3287         struct lod_object *lo = lod_dt_obj(dt);
3288         const struct lod_default_striping *lds = lo->ldo_def_striping;
3289         int rc;
3290         ENTRY;
3291
3292         LASSERT(ergo(lds != NULL,
3293                      lds->lds_def_striping_set ||
3294                      lds->lds_dir_def_striping_set));
3295
3296         if (!LMVEA_DELETE_VALUES(lo->ldo_dir_stripe_count,
3297                                  lo->ldo_dir_stripe_offset)) {
3298                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
3299                 int stripe_count = lo->ldo_dir_stripe_count;
3300
3301                 if (info->lti_ea_store_size < sizeof(*v1)) {
3302                         rc = lod_ea_store_resize(info, sizeof(*v1));
3303                         if (rc != 0)
3304                                 RETURN(rc);
3305                         v1 = info->lti_ea_store;
3306                 }
3307
3308                 memset(v1, 0, sizeof(*v1));
3309                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
3310                 v1->lum_stripe_count = cpu_to_le32(stripe_count);
3311                 v1->lum_stripe_offset =
3312                                 cpu_to_le32(lo->ldo_dir_stripe_offset);
3313
3314                 info->lti_buf.lb_buf = v1;
3315                 info->lti_buf.lb_len = sizeof(*v1);
3316
3317                 if (declare)
3318                         rc = lod_declare_xattr_set_lmv(env, dt, attr,
3319                                                        &info->lti_buf, dof, th);
3320                 else
3321                         rc = lod_xattr_set_lmv(env, dt, &info->lti_buf,
3322                                                XATTR_NAME_LMV, 0, th);
3323                 if (rc != 0)
3324                         RETURN(rc);
3325         }
3326
3327         /* Transfer default LMV striping from the parent */
3328         if (lds != NULL && lds->lds_dir_def_striping_set &&
3329             !LMVEA_DELETE_VALUES(lds->lds_dir_def_stripe_count,
3330                                  lds->lds_dir_def_stripe_offset)) {
3331                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
3332
3333                 if (info->lti_ea_store_size < sizeof(*v1)) {
3334                         rc = lod_ea_store_resize(info, sizeof(*v1));
3335                         if (rc != 0)
3336                                 RETURN(rc);
3337                         v1 = info->lti_ea_store;
3338                 }
3339
3340                 memset(v1, 0, sizeof(*v1));
3341                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
3342                 v1->lum_stripe_count =
3343                         cpu_to_le32(lds->lds_dir_def_stripe_count);
3344                 v1->lum_stripe_offset =
3345                         cpu_to_le32(lds->lds_dir_def_stripe_offset);
3346                 v1->lum_hash_type =
3347                         cpu_to_le32(lds->lds_dir_def_hash_type);
3348
3349                 info->lti_buf.lb_buf = v1;
3350                 info->lti_buf.lb_len = sizeof(*v1);
3351                 if (declare)
3352                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
3353                                                        XATTR_NAME_DEFAULT_LMV,
3354                                                        0, th);
3355                 else
3356                         rc = lod_xattr_set_default_lmv_on_dir(env, dt,
3357                                                   &info->lti_buf,
3358                                                   XATTR_NAME_DEFAULT_LMV, 0,
3359                                                   th);
3360                 if (rc != 0)
3361                         RETURN(rc);
3362         }
3363
3364         /* Transfer default LOV striping from the parent */
3365         if (lds != NULL && lds->lds_def_striping_set &&
3366             lds->lds_def_comp_cnt != 0) {
3367                 struct lov_mds_md *lmm;
3368                 int lmm_size = lod_comp_md_size(lo, true);
3369
3370                 if (info->lti_ea_store_size < lmm_size) {
3371                         rc = lod_ea_store_resize(info, lmm_size);
3372                         if (rc != 0)
3373                                 RETURN(rc);
3374                 }
3375                 lmm = info->lti_ea_store;
3376
3377                 rc = lod_generate_lovea(env, lo, lmm, &lmm_size, true);
3378                 if (rc != 0)
3379                         RETURN(rc);
3380
3381                 info->lti_buf.lb_buf = lmm;
3382                 info->lti_buf.lb_len = lmm_size;
3383
3384                 if (declare)
3385                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
3386                                                        XATTR_NAME_LOV, 0, th);
3387                 else
3388                         rc = lod_xattr_set_lov_on_dir(env, dt, &info->lti_buf,
3389                                                       XATTR_NAME_LOV, 0, th);
3390                 if (rc != 0)
3391                         RETURN(rc);
3392         }
3393
3394         RETURN(0);
3395 }
3396
3397 static int lod_declare_dir_striping_create(const struct lu_env *env,
3398                                            struct dt_object *dt,
3399                                            struct lu_attr *attr,
3400                                            struct dt_object_format *dof,
3401                                            struct thandle *th)
3402 {
3403         return lod_dir_striping_create_internal(env, dt, attr, dof, th, true);
3404 }
3405
3406 static int lod_dir_striping_create(const struct lu_env *env,
3407                                    struct dt_object *dt,
3408                                    struct lu_attr *attr,
3409                                    struct dt_object_format *dof,
3410                                    struct thandle *th)
3411 {
3412         return lod_dir_striping_create_internal(env, dt, attr, dof, th, false);
3413 }
3414
3415 /**
3416  * Make LOV EA for striped object.
3417  *
3418  * Generate striping information and store it in the LOV EA of the given
3419  * object. The caller must ensure nobody else is calling the function
3420  * against the object concurrently. The transaction must be started.
3421  * FLDB service must be running as well; it's used to map FID to the target,
3422  * which is stored in LOV EA.
3423  *
3424  * \param[in] env               execution environment for this thread
3425  * \param[in] lo                LOD object
3426  * \param[in] th                transaction handle
3427  *
3428  * \retval                      0 if LOV EA is stored successfully
3429  * \retval                      negative error number on failure
3430  */
3431 static int lod_generate_and_set_lovea(const struct lu_env *env,
3432                                       struct lod_object *lo,
3433                                       struct thandle *th)
3434 {
3435         struct lod_thread_info  *info = lod_env_info(env);
3436         struct dt_object        *next = dt_object_child(&lo->ldo_obj);
3437         struct lov_mds_md_v1    *lmm;
3438         int                      rc, lmm_size;
3439         ENTRY;
3440
3441         LASSERT(lo);
3442
3443         if (lo->ldo_comp_cnt == 0) {
3444                 lod_object_free_striping(env, lo);
3445                 rc = lod_sub_xattr_del(env, next, XATTR_NAME_LOV, th);
3446                 RETURN(rc);
3447         }
3448
3449         lmm_size = lod_comp_md_size(lo, false);
3450         if (info->lti_ea_store_size < lmm_size) {
3451                 rc = lod_ea_store_resize(info, lmm_size);
3452                 if (rc)
3453                         RETURN(rc);
3454         }
3455         lmm = info->lti_ea_store;
3456
3457         rc = lod_generate_lovea(env, lo, lmm, &lmm_size, false);
3458         if (rc)
3459                 RETURN(rc);
3460
3461         info->lti_buf.lb_buf = lmm;
3462         info->lti_buf.lb_len = lmm_size;
3463         rc = lod_sub_xattr_set(env, next, &info->lti_buf,
3464                                XATTR_NAME_LOV, 0, th);
3465         RETURN(rc);
3466 }
3467
3468 /**
3469  * Delete layout component(s)
3470  *
3471  * \param[in] env       execution environment for this thread
3472  * \param[in] dt        object
3473  * \param[in] th        transaction handle
3474  *
3475  * \retval      0 on success
3476  * \retval      negative error number on failure
3477  */
3478 static int lod_layout_del(const struct lu_env *env, struct dt_object *dt,
3479                           struct thandle *th)
3480 {
3481         struct lod_layout_component     *lod_comp;
3482         struct lod_object       *lo = lod_dt_obj(dt);
3483         struct dt_object        *next = dt_object_child(dt);
3484         struct lu_attr  *attr = &lod_env_info(env)->lti_attr;
3485         int     rc, i, j, left;
3486
3487         LASSERT(lo->ldo_is_composite);
3488         LASSERT(lo->ldo_comp_cnt > 0 && lo->ldo_comp_entries != NULL);
3489
3490         left = lo->ldo_comp_cnt;
3491         for (i = (lo->ldo_comp_cnt - 1); i >= 0; i--) {
3492                 lod_comp = &lo->ldo_comp_entries[i];
3493
3494                 if (lod_comp->llc_id != LCME_ID_INVAL)
3495                         break;
3496                 left--;
3497
3498                 /* Not instantiated component */
3499                 if (lod_comp->llc_stripe == NULL)
3500                         continue;
3501
3502                 LASSERT(lod_comp->llc_stripe_count > 0);
3503                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
3504                         struct dt_object *obj = lod_comp->llc_stripe[j];
3505
3506                         if (obj == NULL)
3507                                 continue;
3508                         rc = lod_sub_destroy(env, obj, th);
3509                         if (rc)
3510                                 GOTO(out, rc);
3511
3512                         lu_object_put(env, &obj->do_lu);
3513                         lod_comp->llc_stripe[j] = NULL;
3514                 }
3515                 OBD_FREE(lod_comp->llc_stripe, sizeof(struct dt_object *) *
3516                                         lod_comp->llc_stripes_allocated);
3517                 lod_comp->llc_stripe = NULL;
3518                 lod_comp->llc_stripes_allocated = 0;
3519                 lod_obj_set_pool(lo, i, NULL);
3520                 if (lod_comp->llc_ostlist.op_array) {
3521                         OBD_FREE(lod_comp->llc_ostlist.op_array,
3522                                  lod_comp->llc_ostlist.op_size);
3523                         lod_comp->llc_ostlist.op_array = NULL;
3524                         lod_comp->llc_ostlist.op_size = 0;
3525                 }
3526         }
3527
3528         LASSERTF(left >= 0 && left < lo->ldo_comp_cnt, "left = %d\n", left);
3529         if (left > 0) {
3530                 struct lod_layout_component     *comp_array;
3531
3532                 OBD_ALLOC(comp_array, sizeof(*comp_array) * left);
3533                 if (comp_array == NULL)
3534                         GOTO(out, rc = -ENOMEM);
3535
3536                 memcpy(&comp_array[0], &lo->ldo_comp_entries[0],
3537                        sizeof(*comp_array) * left);
3538
3539                 OBD_FREE(lo->ldo_comp_entries,
3540                          sizeof(*comp_array) * lo->ldo_comp_cnt);
3541                 lo->ldo_comp_entries = comp_array;
3542                 lo->ldo_comp_cnt = left;
3543
3544                 LASSERT(lo->ldo_mirror_count == 1);
3545                 lo->ldo_mirrors[0].lme_end = left - 1;
3546                 lod_obj_inc_layout_gen(lo);
3547         } else {
3548                 lod_free_comp_entries(lo);
3549         }
3550
3551         LASSERT(dt_object_exists(dt));
3552         rc = dt_attr_get(env, next, attr);
3553         if (rc)
3554                 GOTO(out, rc);
3555
3556         if (attr->la_size > 0) {
3557                 attr->la_size = 0;
3558                 attr->la_valid = LA_SIZE;
3559                 rc = lod_sub_attr_set(env, next, attr, th);
3560                 if (rc)
3561                         GOTO(out, rc);
3562         }
3563
3564         rc = lod_generate_and_set_lovea(env, lo, th);
3565         EXIT;
3566 out:
3567         if (rc)
3568                 lod_object_free_striping(env, lo);
3569         return rc;
3570 }
3571
3572 /**
3573  * Implementation of dt_object_operations::do_xattr_set.
3574  *
3575  * Sets specified extended attribute on the object. Three types of EAs are
3576  * special:
3577  *   LOV EA - stores striping for a regular file or default striping (when set
3578  *            on a directory)
3579  *   LMV EA - stores a marker for the striped directories
3580  *   DMV EA - stores default directory striping
3581  *
3582  * When striping is applied to a non-striped existing object (this is called
3583  * late striping), then LOD notices the caller wants to turn the object into a
3584  * striped one. The stripe objects are created and appropriate EA is set:
3585  * LOV EA storing all the stripes directly or LMV EA storing just a small header
3586  * with striping configuration.
3587  *
3588  * \see dt_object_operations::do_xattr_set() in the API description for details.
3589  */
3590 static int lod_xattr_set(const struct lu_env *env,
3591                          struct dt_object *dt, const struct lu_buf *buf,
3592                          const char *name, int fl, struct thandle *th)
3593 {
3594         struct dt_object        *next = dt_object_child(dt);
3595         int                      rc;
3596         ENTRY;
3597
3598         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
3599             strcmp(name, XATTR_NAME_LMV) == 0) {
3600                 struct lmv_mds_md_v1 *lmm = buf->lb_buf;
3601
3602                 if (lmm != NULL && le32_to_cpu(lmm->lmv_hash_type) &
3603                                                 LMV_HASH_FLAG_MIGRATION)
3604                         rc = lod_sub_xattr_set(env, next, buf, name, fl, th);
3605                 else
3606                         rc = lod_dir_striping_create(env, dt, NULL, NULL, th);
3607
3608                 RETURN(rc);
3609         }
3610
3611         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
3612             strcmp(name, XATTR_NAME_LOV) == 0) {
3613                 /* default LOVEA */
3614                 rc = lod_xattr_set_lov_on_dir(env, dt, buf, name, fl, th);
3615                 RETURN(rc);
3616         } else if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
3617                    strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
3618                 /* default LMVEA */
3619                 rc = lod_xattr_set_default_lmv_on_dir(env, dt, buf, name, fl,
3620                                                       th);
3621                 RETURN(rc);
3622         } else if (S_ISREG(dt->do_lu.lo_header->loh_attr) &&
3623                    (!strcmp(name, XATTR_NAME_LOV) ||
3624                     !strncmp(name, XATTR_LUSTRE_LOV,
3625                              strlen(XATTR_LUSTRE_LOV)))) {
3626                 /* in case of lov EA swap, just set it
3627                  * if not, it is a replay so check striping match what we
3628                  * already have during req replay, declare_xattr_set()
3629                  * defines striping, then create() does the work */
3630                 if (fl & LU_XATTR_REPLACE) {
3631                         /* free stripes, then update disk */
3632                         lod_object_free_striping(env, lod_dt_obj(dt));
3633
3634                         rc = lod_sub_xattr_set(env, next, buf, name, fl, th);
3635                 } else if (dt_object_remote(dt)) {
3636                         /* This only happens during migration, see
3637                          * mdd_migrate_create(), in which Master MDT will
3638                          * create a remote target object, and only set
3639                          * (migrating) stripe EA on the remote object,
3640                          * and does not need creating each stripes. */
3641                         rc = lod_sub_xattr_set(env, next, buf, name,
3642                                                       fl, th);
3643                 } else if (strcmp(name, XATTR_LUSTRE_LOV".del") == 0) {
3644                         /* delete component(s) */
3645                         LASSERT(lod_dt_obj(dt)->ldo_comp_cached);
3646                         rc = lod_layout_del(env, dt, th);
3647                 } else {
3648                         /*
3649                          * When 'name' is XATTR_LUSTRE_LOV or XATTR_NAME_LOV,
3650                          * it's going to create create file with specified
3651                          * component(s), the striping must have not being
3652                          * cached in this case;
3653                          *
3654                          * Otherwise, it's going to add/change component(s) to
3655                          * an existing file, the striping must have been cached
3656                          * in this case.
3657                          */
3658                         LASSERT(equi(!strcmp(name, XATTR_LUSTRE_LOV) ||
3659                                      !strcmp(name, XATTR_NAME_LOV),
3660                                 !lod_dt_obj(dt)->ldo_comp_cached));
3661
3662                         rc = lod_striped_create(env, dt, NULL, NULL, th);
3663                 }
3664                 RETURN(rc);
3665         } else if (strcmp(name, XATTR_NAME_FID) == 0) {
3666                 rc = lod_replace_parent_fid(env, dt, th, false);
3667
3668                 RETURN(rc);
3669         }
3670
3671         /* then all other xattr */
3672         rc = lod_xattr_set_internal(env, dt, buf, name, fl, th);
3673
3674         RETURN(rc);
3675 }
3676
3677 /**
3678  * Implementation of dt_object_operations::do_declare_xattr_del.
3679  *
3680  * \see dt_object_operations::do_declare_xattr_del() in the API description
3681  * for details.
3682  */
3683 static int lod_declare_xattr_del(const struct lu_env *env,
3684                                  struct dt_object *dt, const char *name,
3685                                  struct thandle *th)
3686 {
3687         struct lod_object       *lo = lod_dt_obj(dt);
3688         int                     rc;
3689         int                     i;
3690         ENTRY;
3691
3692         rc = lod_sub_declare_xattr_del(env, dt_object_child(dt), name, th);
3693         if (rc != 0)
3694                 RETURN(rc);
3695
3696         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
3697                 RETURN(0);
3698
3699         /* set xattr to each stripes, if needed */
3700         rc = lod_load_striping(env, lo);
3701         if (rc != 0)
3702                 RETURN(rc);
3703
3704         if (lo->ldo_dir_stripe_count == 0)
3705                 RETURN(0);
3706
3707         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
3708                 LASSERT(lo->ldo_stripe[i]);
3709                 rc = lod_sub_declare_xattr_del(env, lo->ldo_stripe[i],
3710                                                name, th);
3711                 if (rc != 0)
3712                         break;
3713         }
3714
3715         RETURN(rc);
3716 }
3717
3718 /**
3719  * Implementation of dt_object_operations::do_xattr_del.
3720  *
3721  * If EA storing a regular striping is being deleted, then release
3722  * all the references to the stripe objects in core.
3723  *
3724  * \see dt_object_operations::do_xattr_del() in the API description for details.
3725  */
3726 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
3727                          const char *name, struct thandle *th)
3728 {
3729         struct dt_object        *next = dt_object_child(dt);
3730         struct lod_object       *lo = lod_dt_obj(dt);
3731         int                     rc;
3732         int                     i;
3733         ENTRY;
3734
3735         if (!strcmp(name, XATTR_NAME_LOV))
3736                 lod_object_free_striping(env, lod_dt_obj(dt));
3737
3738         rc = lod_sub_xattr_del(env, next, name, th);
3739         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
3740                 RETURN(rc);
3741
3742         if (lo->ldo_dir_stripe_count == 0)
3743                 RETURN(0);
3744
3745         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
3746                 LASSERT(lo->ldo_stripe[i]);
3747
3748                 rc = lod_sub_xattr_del(env, lo->ldo_stripe[i], name, th);
3749                 if (rc != 0)
3750                         break;
3751         }
3752
3753         RETURN(rc);
3754 }
3755
3756 /**
3757  * Implementation of dt_object_operations::do_xattr_list.
3758  *
3759  * \see dt_object_operations::do_xattr_list() in the API description
3760  * for details.
3761  */
3762 static int lod_xattr_list(const struct lu_env *env,
3763                           struct dt_object *dt, const struct lu_buf *buf)
3764 {
3765         return dt_xattr_list(env, dt_object_child(dt), buf);
3766 }
3767
3768 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
3769 {
3770         return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
3771 }
3772
3773
3774 /**
3775  * Get default striping.
3776  *
3777  * \param[in] env               execution environment
3778  * \param[in] lo                object
3779  * \param[out] lds              default striping
3780  *
3781  * \retval              0 on success
3782  * \retval              negative if failed
3783  */
3784 static int lod_get_default_lov_striping(const struct lu_env *env,
3785                                         struct lod_object *lo,
3786                                         struct lod_default_striping *lds)
3787 {
3788         struct lod_thread_info *info = lod_env_info(env);
3789         struct lov_user_md_v1 *v1 = NULL;
3790         struct lov_user_md_v3 *v3 = NULL;
3791         struct lov_comp_md_v1 *comp_v1 = NULL;
3792         __u16   comp_cnt;
3793         __u16   mirror_cnt;
3794         bool    composite;
3795         int     rc, i;
3796         ENTRY;
3797
3798         lds->lds_def_striping_set = 0;
3799
3800         rc = lod_get_lov_ea(env, lo);
3801         if (rc < 0)
3802                 RETURN(rc);
3803
3804         if (rc < (typeof(rc))sizeof(struct lov_user_md))
3805                 RETURN(0);
3806
3807         v1 = info->lti_ea_store;
3808         if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1)) {
3809                 lustre_swab_lov_user_md_v1(v1);
3810         } else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3)) {
3811                 v3 = (struct lov_user_md_v3 *)v1;
3812                 lustre_swab_lov_user_md_v3(v3);
3813         } else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
3814                 comp_v1 = (struct lov_comp_md_v1 *)v1;
3815                 lustre_swab_lov_comp_md_v1(comp_v1);
3816         }
3817
3818         if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1 &&
3819             v1->lmm_magic != LOV_MAGIC_COMP_V1)
3820                 RETURN(-ENOTSUPP);
3821
3822         if (v1->lmm_magic == LOV_MAGIC_COMP_V1) {
3823                 comp_v1 = (struct lov_comp_md_v1 *)v1;
3824                 comp_cnt = comp_v1->lcm_entry_count;
3825                 if (comp_cnt == 0)
3826                         RETURN(-EINVAL);
3827                 mirror_cnt = comp_v1->lcm_mirror_count + 1;
3828                 composite = true;
3829         } else {
3830                 comp_cnt = 1;
3831                 mirror_cnt = 0;
3832                 composite = false;
3833         }
3834
3835         /* realloc default comp entries if necessary */
3836         rc = lod_def_striping_comp_resize(lds, comp_cnt);
3837         if (rc < 0)
3838                 RETURN(rc);
3839
3840         lds->lds_def_comp_cnt = comp_cnt;
3841         lds->lds_def_striping_is_composite = composite;
3842         lds->lds_def_mirror_cnt = mirror_cnt;
3843
3844         for (i = 0; i < comp_cnt; i++) {
3845                 struct lod_layout_component *lod_comp;
3846                 struct lu_extent *ext;
3847                 char *pool;
3848
3849                 lod_comp = &lds->lds_def_comp_entries[i];
3850                 /*
3851                  * reset lod_comp values, llc_stripes is always NULL in
3852                  * the default striping template, llc_pool will be reset
3853                  * later below.
3854                  */
3855                 memset(lod_comp, 0, offsetof(typeof(*lod_comp), llc_pool));
3856
3857                 if (composite) {
3858                         v1 = (struct lov_user_md *)((char *)comp_v1 +
3859                                         comp_v1->lcm_entries[i].lcme_offset);
3860                         ext = &comp_v1->lcm_entries[i].lcme_extent;
3861                         lod_comp->llc_extent = *ext;
3862                 }
3863
3864                 if (v1->lmm_pattern != LOV_PATTERN_RAID0 &&
3865                     v1->lmm_pattern != LOV_PATTERN_MDT &&
3866                     v1->lmm_pattern != 0) {
3867                         lod_free_def_comp_entries(lds);
3868                         RETURN(-EINVAL);
3869                 }
3870
3871                 CDEBUG(D_LAYOUT, DFID" stripe_count=%d stripe_size=%d "
3872                        "stripe_offset=%d\n",
3873                        PFID(lu_object_fid(&lo->ldo_obj.do_lu)),
3874                        (int)v1->lmm_stripe_count, (int)v1->lmm_stripe_size,
3875                        (int)v1->lmm_stripe_offset);
3876
3877                 lod_comp->llc_stripe_count = v1->lmm_stripe_count;
3878                 lod_comp->llc_stripe_size = v1->lmm_stripe_size;
3879                 lod_comp->llc_stripe_offset = v1->lmm_stripe_offset;
3880                 lod_comp->llc_pattern = v1->lmm_pattern;
3881
3882                 pool = NULL;
3883                 if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
3884                         /* XXX: sanity check here */
3885                         v3 = (struct lov_user_md_v3 *) v1;
3886                         if (v3->lmm_pool_name[0] != '\0')
3887                                 pool = v3->lmm_pool_name;
3888                 }
3889                 lod_set_def_pool(lds, i, pool);
3890         }
3891
3892         lds->lds_def_striping_set = 1;
3893         RETURN(rc);
3894 }
3895
3896 /**
3897  * Get default directory striping.
3898  *
3899  * \param[in] env               execution environment
3900  * \param[in] lo                object
3901  * \param[out] lds              default striping
3902  *
3903  * \retval              0 on success
3904  * \retval              negative if failed
3905  */
3906 static int lod_get_default_lmv_striping(const struct lu_env *env,
3907                                         struct lod_object *lo,
3908                                         struct lod_default_striping *lds)
3909 {
3910         struct lod_thread_info  *info = lod_env_info(env);
3911         struct lmv_user_md_v1   *v1 = NULL;
3912         int                      rc;
3913         ENTRY;
3914
3915         lds->lds_dir_def_striping_set = 0;
3916         rc = lod_get_default_lmv_ea(env, lo);
3917         if (rc < 0)
3918                 RETURN(rc);
3919
3920         if (rc < (typeof(rc))sizeof(struct lmv_user_md))
3921                 RETURN(0);
3922
3923         v1 = info->lti_ea_store;
3924
3925         lds->lds_dir_def_stripe_count = le32_to_cpu(v1->lum_stripe_count);
3926         lds->lds_dir_def_stripe_offset = le32_to_cpu(v1->lum_stripe_offset);
3927         lds->lds_dir_def_hash_type = le32_to_cpu(v1->lum_hash_type);
3928         lds->lds_dir_def_striping_set = 1;
3929
3930         RETURN(0);
3931 }
3932
3933 /**
3934  * Get default striping in the object.
3935  *
3936  * Get object default striping and default directory striping.
3937  *
3938  * \param[in] env               execution environment
3939  * \param[in] lo                object
3940  * \param[out] lds              default striping
3941  *
3942  * \retval              0 on success
3943  * \retval              negative if failed
3944  */
3945 static int lod_get_default_striping(const struct lu_env *env,
3946                                     struct lod_object *lo,
3947                                     struct lod_default_striping *lds)
3948 {
3949         int rc, rc1;
3950
3951         rc = lod_get_default_lov_striping(env, lo, lds);
3952         rc1 = lod_get_default_lmv_striping(env, lo, lds);
3953         if (rc == 0 && rc1 < 0)
3954                 rc = rc1;
3955
3956         return rc;
3957 }
3958
3959 /**
3960  * Apply default striping on object.
3961  *
3962  * If object striping pattern is not set, set to the one in default striping.
3963  * The default striping is from parent or fs.
3964  *
3965  * \param[in] lo                new object
3966  * \param[in] lds               default striping
3967  * \param[in] mode              new object's mode
3968  */
3969 static void lod_striping_from_default(struct lod_object *lo,
3970                                       const struct lod_default_striping *lds,
3971                                       umode_t mode)
3972 {
3973         struct lod_device *d = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
3974         struct lov_desc *desc = &d->lod_desc;
3975         int i, rc;
3976
3977         if (lds->lds_def_striping_set && S_ISREG(mode)) {
3978                 rc = lod_alloc_comp_entries(lo, lds->lds_def_mirror_cnt,
3979                                             lds->lds_def_comp_cnt);
3980                 if (rc != 0)
3981                         return;
3982
3983                 lo->ldo_is_composite = lds->lds_def_striping_is_composite;
3984
3985                 for (i = 0; i < lo->ldo_comp_cnt; i++) {
3986                         struct lod_layout_component *obj_comp =
3987                                                 &lo->ldo_comp_entries[i];
3988                         struct lod_layout_component *def_comp =
3989                                                 &lds->lds_def_comp_entries[i];
3990
3991                         CDEBUG(D_LAYOUT, "Inherite from default: size:%hu "
3992                                "nr:%u offset:%u pattern %#x %s\n",
3993                                def_comp->llc_stripe_size,
3994                                def_comp->llc_stripe_count,
3995                                def_comp->llc_stripe_offset,
3996                                def_comp->llc_pattern,
3997                                def_comp->llc_pool ?: "");
3998
3999                         *obj_comp = *def_comp;
4000                         if (def_comp->llc_pool != NULL) {
4001                                 /* pointer was copied from def_comp */
4002                                 obj_comp->llc_pool = NULL;
4003                                 lod_obj_set_pool(lo, i, def_comp->llc_pool);
4004                         }
4005
4006                         /*
4007                          * Don't initialize these fields for plain layout
4008                          * (v1/v3) here, they are inherited in the order of
4009                          * 'parent' -> 'fs default (root)' -> 'global default
4010                          * values for stripe_count & stripe_size'.
4011                          *
4012                          * see lod_ah_init().
4013                          */
4014                         if (!lo->ldo_is_composite)
4015                                 continue;
4016
4017                         if (obj_comp->llc_stripe_count <= 0 &&
4018                             obj_comp->llc_pattern != LOV_PATTERN_MDT)
4019                                 obj_comp->llc_stripe_count =
4020                                         desc->ld_default_stripe_count;
4021                         if (obj_comp->llc_stripe_size <= 0)
4022                                 obj_comp->llc_stripe_size =
4023                                         desc->ld_default_stripe_size;
4024                 }
4025         } else if (lds->lds_dir_def_striping_set && S_ISDIR(mode)) {
4026                 if (lo->ldo_dir_stripe_count == 0)
4027                         lo->ldo_dir_stripe_count =
4028                                 lds->lds_dir_def_stripe_count;
4029                 if (lo->ldo_dir_stripe_offset == -1)
4030                         lo->ldo_dir_stripe_offset =
4031                                 lds->lds_dir_def_stripe_offset;
4032                 if (lo->ldo_dir_hash_type == 0)
4033                         lo->ldo_dir_hash_type = lds->lds_dir_def_hash_type;
4034
4035                 CDEBUG(D_LAYOUT, "striping from default dir: count:%hu, "
4036                        "offset:%u, hash_type:%u\n",
4037                        lo->ldo_dir_stripe_count, lo->ldo_dir_stripe_offset,
4038                        lo->ldo_dir_hash_type);
4039         }
4040 }
4041
4042 static inline bool lod_need_inherit_more(struct lod_object *lo, bool from_root)
4043 {
4044         struct lod_layout_component *lod_comp;
4045
4046         if (lo->ldo_comp_cnt == 0)
4047                 return true;
4048
4049         if (lo->ldo_is_composite)
4050                 return false;
4051
4052         lod_comp = &lo->ldo_comp_entries[0];
4053
4054         if (lod_comp->llc_stripe_count <= 0 ||
4055             lod_comp->llc_stripe_size <= 0)
4056                 return true;
4057
4058         if (from_root && (lod_comp->llc_pool == NULL ||
4059                           lod_comp->llc_stripe_offset == LOV_OFFSET_DEFAULT))
4060                 return true;
4061
4062         return false;
4063 }
4064
4065 /**
4066  * Implementation of dt_object_operations::do_ah_init.
4067  *
4068  * This method is used to make a decision on the striping configuration for the
4069  * object being created. It can be taken from the \a parent object if it exists,
4070  * or filesystem's default. The resulting configuration (number of stripes,
4071  * stripe size/offset, pool name, etc) is stored in the object itself and will
4072  * be used by the methods like ->doo_declare_create().
4073  *
4074  * \see dt_object_operations::do_ah_init() in the API description for details.
4075  */
4076 static void lod_ah_init(const struct lu_env *env,
4077                         struct dt_allocation_hint *ah,
4078                         struct dt_object *parent,
4079                         struct dt_object *child,
4080                         umode_t child_mode)
4081 {
4082         struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
4083         struct lod_thread_info *info = lod_env_info(env);
4084         struct lod_default_striping *lds = &info->lti_def_striping;
4085         struct dt_object *nextp = NULL;
4086         struct dt_object *nextc;
4087         struct lod_object *lp = NULL;
4088         struct lod_object *lc;
4089         struct lov_desc *desc;
4090         struct lod_layout_component *lod_comp;
4091         int rc;
4092         ENTRY;
4093
4094         LASSERT(child);
4095
4096         if (likely(parent)) {
4097                 nextp = dt_object_child(parent);
4098                 lp = lod_dt_obj(parent);
4099         }
4100
4101         nextc = dt_object_child(child);
4102         lc = lod_dt_obj(child);
4103
4104         LASSERT(!lod_obj_is_striped(child));
4105         /* default layout template may have been set on the regular file
4106          * when this is called from mdd_create_data() */
4107         if (S_ISREG(child_mode))
4108                 lod_free_comp_entries(lc);
4109
4110         if (!dt_object_exists(nextc))
4111                 nextc->do_ops->do_ah_init(env, ah, nextp, nextc, child_mode);
4112
4113         if (S_ISDIR(child_mode)) {
4114                 /* other default values are 0 */
4115                 lc->ldo_dir_stripe_offset = -1;
4116
4117                 /* get default striping from parent object */
4118                 if (likely(lp != NULL))
4119                         lod_get_default_striping(env, lp, lds);
4120
4121                 /* set child default striping info, default value is NULL */
4122                 if (lds->lds_def_striping_set || lds->lds_dir_def_striping_set)
4123                         lc->ldo_def_striping = lds;
4124
4125                 /* It should always honour the specified stripes */
4126                 if (ah->dah_eadata != NULL && ah->dah_eadata_len != 0 &&
4127                     lod_verify_md_striping(d, ah->dah_eadata) == 0) {
4128                         const struct lmv_user_md_v1 *lum1 = ah->dah_eadata;
4129
4130                         lc->ldo_dir_stripe_count =
4131                                 le32_to_cpu(lum1->lum_stripe_count);
4132                         lc->ldo_dir_stripe_offset =
4133                                 le32_to_cpu(lum1->lum_stripe_offset);
4134                         lc->ldo_dir_hash_type =
4135                                 le32_to_cpu(lum1->lum_hash_type);
4136                         CDEBUG(D_INFO,
4137                                "set dirstripe: count %hu, offset %d, hash %u\n",
4138                                 lc->ldo_dir_stripe_count,
4139                                 (int)lc->ldo_dir_stripe_offset,
4140                                 lc->ldo_dir_hash_type);
4141                 } else {
4142                         /* transfer defaults LMV to new directory */
4143                         lod_striping_from_default(lc, lds, child_mode);
4144                 }
4145
4146                 /* shrink the stripe_count to the avaible MDT count */
4147                 if (lc->ldo_dir_stripe_count > d->lod_remote_mdt_count + 1 &&
4148                     !OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE))
4149                         lc->ldo_dir_stripe_count = d->lod_remote_mdt_count + 1;
4150
4151                 /* Directory will be striped only if stripe_count > 1, if
4152                  * stripe_count == 1, let's reset stripe_count = 0 to avoid
4153                  * create single master stripe and also help to unify the
4154                  * stripe handling of directories and files */
4155                 if (lc->ldo_dir_stripe_count == 1)
4156                         lc->ldo_dir_stripe_count = 0;
4157
4158                 CDEBUG(D_INFO, "final dir stripe [%hu %d %u]\n",
4159                        lc->ldo_dir_stripe_count,
4160                        (int)lc->ldo_dir_stripe_offset, lc->ldo_dir_hash_type);
4161
4162                 RETURN_EXIT;
4163         }
4164
4165         /* child object regular file*/
4166
4167         if (!lod_object_will_be_striped(S_ISREG(child_mode),
4168                                         lu_object_fid(&child->do_lu)))
4169                 RETURN_EXIT;
4170
4171         /* If object is going to be striped over OSTs, transfer default
4172          * striping information to the child, so that we can use it
4173          * during declaration and creation.
4174          *
4175          * Try from the parent first.
4176          */
4177         if (likely(lp != NULL)) {
4178                 rc = lod_get_default_lov_striping(env, lp, lds);
4179                 if (rc == 0)
4180                         lod_striping_from_default(lc, lds, child_mode);
4181         }
4182
4183         /* Initialize lod_device::lod_md_root object reference */
4184         if (d->lod_md_root == NULL) {
4185                 struct dt_object *root;
4186                 struct lod_object *lroot;
4187
4188                 lu_root_fid(&info->lti_fid);
4189                 root = dt_locate(env, &d->lod_dt_dev, &info->lti_fid);
4190                 if (!IS_ERR(root)) {
4191                         lroot = lod_dt_obj(root);
4192
4193                         spin_lock(&d->lod_lock);
4194                         if (d->lod_md_root != NULL)
4195                                 dt_object_put(env, &d->lod_md_root->ldo_obj);
4196                         d->lod_md_root = lroot;
4197                         spin_unlock(&d->lod_lock);
4198                 }
4199         }
4200
4201         /* try inherit layout from the root object (fs default) when:
4202          *  - parent does not have default layout; or
4203          *  - parent has plain(v1/v3) default layout, and some attributes
4204          *    are not specified in the default layout;
4205          */
4206         if (d->lod_md_root != NULL && lod_need_inherit_more(lc, true)) {
4207                 rc = lod_get_default_lov_striping(env, d->lod_md_root, lds);
4208                 if (rc)
4209                         goto out;
4210                 if (lc->ldo_comp_cnt == 0) {
4211                         lod_striping_from_default(lc, lds, child_mode);
4212                 } else if (!lds->lds_def_striping_is_composite) {
4213                         struct lod_layout_component *def_comp;
4214
4215                         LASSERT(!lc->ldo_is_composite);
4216                         lod_comp = &lc->ldo_comp_entries[0];
4217                         def_comp = &lds->lds_def_comp_entries[0];
4218
4219                         if (lod_comp->llc_stripe_count <= 0)
4220                                 lod_comp->llc_stripe_count =
4221                                         def_comp->llc_stripe_count;
4222                         if (lod_comp->llc_stripe_size <= 0)
4223                                 lod_comp->llc_stripe_size =
4224                                         def_comp->llc_stripe_size;
4225                         if (lod_comp->llc_stripe_offset == LOV_OFFSET_DEFAULT)
4226                                 lod_comp->llc_stripe_offset =
4227                                         def_comp->llc_stripe_offset;
4228                         if (lod_comp->llc_pool == NULL)
4229                                 lod_obj_set_pool(lc, 0, def_comp->llc_pool);
4230                 }
4231         }
4232 out:
4233         /*
4234          * fs default striping may not be explicitly set, or historically set
4235          * in config log, use them.
4236          */
4237         if (lod_need_inherit_more(lc, false)) {
4238
4239                 if (lc->ldo_comp_cnt == 0) {
4240                         rc = lod_alloc_comp_entries(lc, 0, 1);
4241                         if (rc)
4242                                 /* fail to allocate memory, will create a
4243                                  * non-striped file. */
4244                                 RETURN_EXIT;
4245                         lc->ldo_is_composite = 0;
4246                         lod_comp = &lc->ldo_comp_entries[0];
4247                         lod_comp->llc_stripe_offset = LOV_OFFSET_DEFAULT;
4248                 }
4249                 LASSERT(!lc->ldo_is_composite);
4250                 lod_comp = &lc->ldo_comp_entries[0];
4251                 desc = &d->lod_desc;
4252                 if (lod_comp->llc_stripe_count <= 0)
4253                         lod_comp->llc_stripe_count =
4254                                 desc->ld_default_stripe_count;
4255                 if (lod_comp->llc_stripe_size <= 0)
4256                         lod_comp->llc_stripe_size =
4257                                 desc->ld_default_stripe_size;
4258         }
4259
4260         EXIT;
4261 }
4262
4263 #define ll_do_div64(aaa,bbb)    do_div((aaa), (bbb))
4264 /**
4265  * Size initialization on late striping.
4266  *
4267  * Propagate the size of a truncated object to a deferred striping.
4268  * This function handles a special case when truncate was done on a
4269  * non-striped object and now while the striping is being created
4270  * we can't lose that size, so we have to propagate it to the stripes
4271  * being created.
4272  *
4273  * \param[in] env       execution environment
4274  * \param[in] dt        object
4275  * \param[in] th        transaction handle
4276  *
4277  * \retval              0 on success
4278  * \retval              negative if failed
4279  */
4280 static int lod_declare_init_size(const struct lu_env *env,
4281                                  struct dt_object *dt, struct thandle *th)
4282 {
4283         struct dt_object        *next = dt_object_child(dt);
4284         struct lod_object       *lo = lod_dt_obj(dt);
4285         struct dt_object        **objects = NULL;
4286         struct lu_attr  *attr = &lod_env_info(env)->lti_attr;
4287         uint64_t        size, offs;
4288         int     i, rc, stripe, stripe_count = 0, stripe_size = 0;
4289         ENTRY;
4290
4291         if (!lod_obj_is_striped(dt))
4292                 RETURN(0);
4293
4294         rc = dt_attr_get(env, next, attr);
4295         LASSERT(attr->la_valid & LA_SIZE);
4296         if (rc)
4297                 RETURN(rc);
4298
4299         size = attr->la_size;
4300         if (size == 0)
4301                 RETURN(0);
4302
4303         for (i = 0; i < lo->ldo_comp_cnt; i++) {
4304                 struct lod_layout_component *lod_comp;
4305                 struct lu_extent *extent;
4306
4307                 lod_comp = &lo->ldo_comp_entries[i];
4308
4309                 if (lod_comp->llc_stripe == NULL)
4310                         continue;
4311
4312                 extent = &lod_comp->llc_extent;
4313                 CDEBUG(D_INFO, "%lld [%lld, %lld)\n",
4314                        size, extent->e_start, extent->e_end);
4315                 if (!lo->ldo_is_composite ||
4316                     (size >= extent->e_start && size < extent->e_end)) {
4317                         objects = lod_comp->llc_stripe;
4318                         stripe_count = lod_comp->llc_stripe_count;
4319                         stripe_size = lod_comp->llc_stripe_size;
4320                         break;
4321                 }
4322         }
4323
4324         if (stripe_count == 0)
4325                 RETURN(0);
4326
4327         LASSERT(objects != NULL && stripe_size != 0);
4328
4329         /* ll_do_div64(a, b) returns a % b, and a = a / b */
4330         ll_do_div64(size, (__u64)stripe_size);
4331         stripe = ll_do_div64(size, (__u64)stripe_count);
4332         LASSERT(objects[stripe] != NULL);
4333
4334         size = size * stripe_size;
4335         offs = attr->la_size;
4336         size += ll_do_div64(offs, stripe_size);
4337
4338         attr->la_valid = LA_SIZE;
4339         attr->la_size = size;
4340
4341         rc = lod_sub_declare_attr_set(env, objects[stripe], attr, th);
4342
4343         RETURN(rc);
4344 }
4345
4346 /**
4347  * Declare creation of striped object.
4348  *
4349  * The function declares creation stripes for a regular object. The function
4350  * also declares whether the stripes will be created with non-zero size if
4351  * previously size was set non-zero on the master object. If object \a dt is
4352  * not local, then only fully defined striping can be applied in \a lovea.
4353  * Otherwise \a lovea can be in the form of pattern, see lod_qos_parse_config()
4354  * for the details.
4355  *
4356  * \param[in] env       execution environment
4357  * \param[in] dt        object
4358  * \param[in] attr      attributes the stripes will be created with
4359  * \param[in] lovea     a buffer containing striping description
4360  * \param[in] th        transaction handle
4361  *
4362  * \retval              0 on success
4363  * \retval              negative if failed
4364  */
4365 int lod_declare_striped_create(const struct lu_env *env, struct dt_object *dt,
4366                                struct lu_attr *attr,
4367                                const struct lu_buf *lovea, struct thandle *th)
4368 {
4369         struct lod_thread_info  *info = lod_env_info(env);
4370         struct dt_object        *next = dt_object_child(dt);
4371         struct lod_object       *lo = lod_dt_obj(dt);
4372         int                      rc;
4373         ENTRY;
4374
4375         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO))
4376                 GOTO(out, rc = -ENOMEM);
4377
4378         if (!dt_object_remote(next)) {
4379                 /* choose OST and generate appropriate objects */
4380                 rc = lod_prepare_create(env, lo, attr, lovea, th);
4381                 if (rc)
4382                         GOTO(out, rc);
4383
4384                 /*
4385                  * declare storage for striping data
4386                  */
4387                 info->lti_buf.lb_len = lod_comp_md_size(lo, false);
4388         } else {
4389                 /* LOD can not choose OST objects for remote objects, i.e.
4390                  * stripes must be ready before that. Right now, it can only
4391                  * happen during migrate, i.e. migrate process needs to create
4392                  * remote regular file (mdd_migrate_create), then the migrate
4393                  * process will provide stripeEA. */
4394                 LASSERT(lovea != NULL);
4395                 info->lti_buf = *lovea;
4396         }
4397
4398         rc = lod_sub_declare_xattr_set(env, next, &info->lti_buf,
4399                                        XATTR_NAME_LOV, 0, th);
4400         if (rc)
4401                 GOTO(out, rc);
4402
4403         /*
4404          * if striping is created with local object's size > 0,
4405          * we have to propagate this size to specific object
4406          * the case is possible only when local object was created previously
4407          */
4408         if (dt_object_exists(next))
4409                 rc = lod_declare_init_size(env, dt, th);
4410
4411 out:
4412         /* failed to create striping or to set initial size, let's reset
4413          * config so that others don't get confused */
4414         if (rc)
4415                 lod_object_free_striping(env, lo);
4416
4417         RETURN(rc);
4418 }
4419
4420 /**
4421  * Implementation of dt_object_operations::do_declare_create.
4422  *
4423  * The method declares creation of a new object. If the object will be striped,
4424  * then helper functions are called to find FIDs for the stripes, declare
4425  * creation of the stripes and declare initialization of the striping
4426  * information to be stored in the master object.
4427  *
4428  * \see dt_object_operations::do_declare_create() in the API description
4429  * for details.
4430  */
4431 static int lod_declare_create(const struct lu_env *env, struct dt_object *dt,
4432                               struct lu_attr *attr,
4433                               struct dt_allocation_hint *hint,
4434                               struct dt_object_format *dof, struct thandle *th)
4435 {
4436         struct dt_object   *next = dt_object_child(dt);
4437         struct lod_object  *lo = lod_dt_obj(dt);
4438         int                 rc;
4439         ENTRY;
4440
4441         LASSERT(dof);
4442         LASSERT(attr);
4443         LASSERT(th);
4444
4445         /*
4446          * first of all, we declare creation of local object
4447          */
4448         rc = lod_sub_declare_create(env, next, attr, hint, dof, th);
4449         if (rc != 0)
4450                 GOTO(out, rc);
4451
4452         /*
4453          * it's lod_ah_init() that has decided the object will be striped
4454          */
4455         if (dof->dof_type == DFT_REGULAR) {
4456                 /* callers don't want stripes */
4457                 /* XXX: all tricky interactions with ->ah_make_hint() decided
4458                  * to use striping, then ->declare_create() behaving differently
4459                  * should be cleaned */
4460                 if (dof->u.dof_reg.striped != 0)
4461                         rc = lod_declare_striped_create(env, dt, attr,
4462                                                         NULL, th);
4463         } else if (dof->dof_type == DFT_DIR) {
4464                 struct seq_server_site *ss;
4465
4466                 ss = lu_site2seq(dt->do_lu.lo_dev->ld_site);
4467
4468                 /* If the parent has default stripeEA, and client
4469                  * did not find it before sending create request,
4470                  * then MDT will return -EREMOTE, and client will
4471                  * retrieve the default stripeEA and re-create the
4472                  * sub directory.
4473                  *
4474                  * Note: if dah_eadata != NULL, it means creating the
4475                  * striped directory with specified stripeEA, then it
4476                  * should ignore the default stripeEA */
4477                 if (hint != NULL && hint->dah_eadata == NULL) {
4478                         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_STALE_DIR_LAYOUT))
4479                                 GOTO(out, rc = -EREMOTE);
4480
4481                         if (lo->ldo_dir_stripe_offset == -1) {
4482                                 /* child and parent should be in the same MDT */
4483                                 if (hint->dah_parent != NULL &&
4484                                     dt_object_remote(hint->dah_parent))
4485                                         GOTO(out, rc = -EREMOTE);
4486                         } else if (lo->ldo_dir_stripe_offset !=
4487                                    ss->ss_node_id) {
4488                                 struct lod_device *lod;
4489                                 struct lod_tgt_descs *ltd;
4490                                 struct lod_tgt_desc *tgt = NULL;
4491                                 bool found_mdt = false;
4492                                 int i;
4493
4494                                 lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
4495                                 ltd = &lod->lod_mdt_descs;
4496                                 cfs_foreach_bit(ltd->ltd_tgt_bitmap, i) {
4497                                         tgt = LTD_TGT(ltd, i);
4498                                         if (tgt->ltd_index ==
4499                                                 lo->ldo_dir_stripe_offset) {
4500                                                 found_mdt = true;
4501                                                 break;
4502                                         }
4503                                 }
4504
4505                                 /* If the MDT indicated by stripe_offset can be
4506                                  * found, then tell client to resend the create
4507                                  * request to the correct MDT, otherwise return
4508                                  * error to client */
4509                                 if (found_mdt)
4510                                         GOTO(out, rc = -EREMOTE);
4511                                 else
4512                                         GOTO(out, rc = -EINVAL);
4513                         }
4514                 }
4515
4516                 rc = lod_declare_dir_striping_create(env, dt, attr, dof, th);
4517         }
4518 out:
4519         /* failed to create striping or to set initial size, let's reset
4520          * config so that others don't get confused */
4521         if (rc)
4522                 lod_object_free_striping(env, lo);
4523         RETURN(rc);
4524 }
4525
4526 /**
4527  * Generate component ID for new created component.
4528  *
4529  * \param[in] lo                LOD object
4530  * \param[in] comp_idx          index of ldo_comp_entries
4531  *
4532  * \retval                      component ID on success
4533  * \retval                      LCME_ID_INVAL on failure
4534  */
4535 static __u32 lod_gen_component_id(struct lod_object *lo,
4536                                   int mirror_id, int comp_idx)
4537 {
4538         struct lod_layout_component *lod_comp;
4539         __u32   id, start, end;
4540         int     i;
4541
4542         LASSERT(lo->ldo_comp_entries[comp_idx].llc_id == LCME_ID_INVAL);
4543
4544         lod_obj_inc_layout_gen(lo);
4545         id = lo->ldo_layout_gen;
4546         if (likely(id <= SEQ_ID_MAX))
4547                 RETURN(pflr_id(mirror_id, id & SEQ_ID_MASK));
4548
4549         /* Layout generation wraps, need to check collisions. */
4550         start = id & SEQ_ID_MASK;
4551         end = SEQ_ID_MAX;
4552 again:
4553         for (id = start; id <= end; id++) {
4554                 for (i = 0; i < lo->ldo_comp_cnt; i++) {
4555                         lod_comp = &lo->ldo_comp_entries[i];
4556                         if (pflr_id(mirror_id, id) == lod_comp->llc_id)
4557                                 break;
4558                 }
4559                 /* Found the ununsed ID */
4560                 if (i == lo->ldo_comp_cnt)
4561                         RETURN(pflr_id(mirror_id, id));
4562         }
4563         if (end == LCME_ID_MAX) {
4564                 start = 1;
4565                 end = min(lo->ldo_layout_gen & LCME_ID_MASK,
4566                           (__u32)(LCME_ID_MAX - 1));
4567                 goto again;
4568         }
4569
4570         RETURN(LCME_ID_INVAL);
4571 }
4572
4573 /**
4574  * Creation of a striped regular object.
4575  *
4576  * The function is called to create the stripe objects for a regular
4577  * striped file. This can happen at the initial object creation or
4578  * when the caller asks LOD to do so using ->do_xattr_set() method
4579  * (so called late striping). Notice all the information are already
4580  * prepared in the form of the list of objects (ldo_stripe field).
4581  * This is done during declare phase.
4582  *
4583  * \param[in] env       execution environment
4584  * \param[in] dt        object
4585  * \param[in] attr      attributes the stripes will be created with
4586  * \param[in] dof       format of stripes (see OSD API description)
4587  * \param[in] th        transaction handle
4588  *
4589  * \retval              0 on success
4590  * \retval              negative if failed
4591  */
4592 int lod_striped_create(const struct lu_env *env, struct dt_object *dt,
4593                        struct lu_attr *attr, struct dt_object_format *dof,
4594                        struct thandle *th)
4595 {
4596         struct lod_layout_component     *lod_comp;
4597         struct lod_object       *lo = lod_dt_obj(dt);
4598         int     rc = 0, i, j;
4599         ENTRY;
4600
4601         LASSERT(lo->ldo_comp_cnt != 0 && lo->ldo_comp_entries != NULL);
4602
4603         /* create all underlying objects */
4604         for (i = 0; i < lo->ldo_comp_cnt; i++) {
4605                 lod_comp = &lo->ldo_comp_entries[i];
4606
4607                 if (lod_comp->llc_id == LCME_ID_INVAL) {
4608                         lod_comp->llc_id = lod_gen_component_id(lo, 0, i);
4609                         if (lod_comp->llc_id == LCME_ID_INVAL)
4610                                 GOTO(out, rc = -ERANGE);
4611                 }
4612
4613                 if (lod_comp_inited(lod_comp))
4614                         continue;
4615
4616                 if (lod_comp->llc_pattern & LOV_PATTERN_F_RELEASED)
4617                         lod_comp_set_init(lod_comp);
4618
4619                 if (lov_pattern(lod_comp->llc_pattern) == LOV_PATTERN_MDT)
4620                         lod_comp_set_init(lod_comp);
4621
4622                 if (lod_comp->llc_stripe == NULL)
4623                         continue;
4624
4625                 LASSERT(lod_comp->llc_stripe_count);
4626                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
4627                         struct dt_object *object = lod_comp->llc_stripe[j];
4628                         LASSERT(object != NULL);
4629                         rc = lod_sub_create(env, object, attr, NULL, dof, th);
4630                         if (rc)
4631                                 GOTO(out, rc);
4632                 }
4633                 lod_comp_set_init(lod_comp);
4634         }
4635
4636         rc = lod_fill_mirrors(lo);
4637         if (rc)
4638                 GOTO(out, rc);
4639
4640         rc = lod_generate_and_set_lovea(env, lo, th);
4641         if (rc)
4642                 GOTO(out, rc);
4643
4644         lo->ldo_comp_cached = 1;
4645         RETURN(0);
4646
4647 out:
4648         lod_object_free_striping(env, lo);
4649         RETURN(rc);
4650 }
4651
4652 /**
4653  * Implementation of dt_object_operations::do_create.
4654  *
4655  * If any of preceeding methods (like ->do_declare_create(),
4656  * ->do_ah_init(), etc) chose to create a striped object,
4657  * then this method will create the master and the stripes.
4658  *
4659  * \see dt_object_operations::do_create() in the API description for details.
4660  */
4661 static int lod_create(const struct lu_env *env, struct dt_object *dt,
4662                       struct lu_attr *attr, struct dt_allocation_hint *hint,
4663                       struct dt_object_format *dof, struct thandle *th)
4664 {
4665         int                 rc;
4666         ENTRY;
4667
4668         /* create local object */
4669         rc = lod_sub_create(env, dt_object_child(dt), attr, hint, dof, th);
4670         if (rc != 0)
4671                 RETURN(rc);
4672
4673         if (S_ISREG(dt->do_lu.lo_header->loh_attr) &&
4674             lod_obj_is_striped(dt) && dof->u.dof_reg.striped != 0) {
4675                 LASSERT(lod_dt_obj(dt)->ldo_comp_cached == 0);
4676                 rc = lod_striped_create(env, dt, attr, dof, th);
4677         }
4678
4679         RETURN(rc);
4680 }
4681
4682 static inline int
4683 lod_obj_stripe_destroy_cb(const struct lu_env *env, struct lod_object *lo,
4684                           struct dt_object *dt, struct thandle *th,
4685                           int comp_idx, int stripe_idx,
4686                           struct lod_obj_stripe_cb_data *data)
4687 {
4688         if (data->locd_declare)
4689                 return lod_sub_declare_destroy(env, dt, th);
4690         else if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SPEOBJ) ||
4691                  stripe_idx == cfs_fail_val)
4692                 return lod_sub_destroy(env, dt, th);
4693         else
4694                 return 0;
4695 }
4696
4697 /**
4698  * Implementation of dt_object_operations::do_declare_destroy.
4699  *
4700  * If the object is a striped directory, then the function declares reference
4701  * removal from the master object (this is an index) to the stripes and declares
4702  * destroy of all the stripes. In all the cases, it declares an intention to
4703  * destroy the object itself.
4704  *
4705  * \see dt_object_operations::do_declare_destroy() in the API description
4706  * for details.
4707  */
4708 static int lod_declare_destroy(const struct lu_env *env, struct dt_object *dt,
4709                                struct thandle *th)
4710 {
4711         struct dt_object   *next = dt_object_child(dt);
4712         struct lod_object  *lo = lod_dt_obj(dt);
4713         struct lod_thread_info *info = lod_env_info(env);
4714         char               *stripe_name = info->lti_key;
4715         int                 rc, i;
4716         ENTRY;
4717
4718         /*
4719          * load striping information, notice we don't do this when object
4720          * is being initialized as we don't need this information till
4721          * few specific cases like destroy, chown
4722          */
4723         rc = lod_load_striping(env, lo);
4724         if (rc)
4725                 RETURN(rc);
4726
4727         /* declare destroy for all underlying objects */
4728         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
4729                 rc = next->do_ops->do_index_try(env, next,
4730                                                 &dt_directory_features);
4731                 if (rc != 0)
4732                         RETURN(rc);
4733
4734                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
4735                         rc = lod_sub_declare_ref_del(env, next, th);
4736                         if (rc != 0)
4737                                 RETURN(rc);
4738
4739                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
4740                                 PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)),
4741                                 i);
4742                         rc = lod_sub_declare_delete(env, next,
4743                                         (const struct dt_key *)stripe_name, th);
4744                         if (rc != 0)
4745                                 RETURN(rc);
4746                 }
4747         }
4748
4749         /*
4750          * we declare destroy for the local object
4751          */
4752         rc = lod_sub_declare_destroy(env, next, th);
4753         if (rc)
4754                 RETURN(rc);
4755
4756         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ) ||
4757             OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ2))
4758                 RETURN(0);
4759
4760         if (!lod_obj_is_striped(dt))
4761                 RETURN(0);
4762
4763         /* declare destroy all striped objects */
4764         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
4765                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
4766                         if (lo->ldo_stripe[i] == NULL)
4767                                 continue;
4768
4769                         rc = lod_sub_declare_ref_del(env, lo->ldo_stripe[i],
4770                                                      th);
4771
4772                         rc = lod_sub_declare_destroy(env, lo->ldo_stripe[i],
4773                                                      th);
4774                         if (rc != 0)
4775                                 break;
4776                 }
4777         } else {
4778                 struct lod_obj_stripe_cb_data data = { { 0 } };
4779
4780                 data.locd_declare = true;
4781                 data.locd_stripe_cb = lod_obj_stripe_destroy_cb;
4782                 rc = lod_obj_for_each_stripe(env, lo, th, &data);
4783         }
4784
4785         RETURN(rc);
4786 }
4787
4788 /**
4789  * Implementation of dt_object_operations::do_destroy.
4790  *
4791  * If the object is a striped directory, then the function removes references
4792  * from the master object (this is an index) to the stripes and destroys all
4793  * the stripes. In all the cases, the function destroys the object itself.
4794  *
4795  * \see dt_object_operations::do_destroy() in the API description for details.
4796  */
4797 static int lod_destroy(const struct lu_env *env, struct dt_object *dt,
4798                        struct thandle *th)
4799 {
4800         struct dt_object  *next = dt_object_child(dt);
4801         struct lod_object *lo = lod_dt_obj(dt);
4802         struct lod_thread_info *info = lod_env_info(env);
4803         char               *stripe_name = info->lti_key;
4804         unsigned int       i;
4805         int                rc;
4806         ENTRY;
4807
4808         /* destroy sub-stripe of master object */
4809         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
4810                 rc = next->do_ops->do_index_try(env, next,
4811                                                 &dt_directory_features);
4812                 if (rc != 0)
4813                         RETURN(rc);
4814
4815                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
4816                         rc = lod_sub_ref_del(env, next, th);
4817                         if (rc != 0)
4818                                 RETURN(rc);
4819
4820                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
4821                                 PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)),
4822                                 i);
4823
4824                         CDEBUG(D_INFO, DFID" delete stripe %s "DFID"\n",
4825                                PFID(lu_object_fid(&dt->do_lu)), stripe_name,
4826                                PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)));
4827
4828                         rc = lod_sub_delete(env, next,
4829                                        (const struct dt_key *)stripe_name, th);
4830                         if (rc != 0)
4831                                 RETURN(rc);
4832                 }
4833         }
4834
4835         rc = lod_sub_destroy(env, next, th);
4836         if (rc != 0)
4837                 RETURN(rc);
4838
4839         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ) ||
4840             OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ2))
4841                 RETURN(0);
4842
4843         if (!lod_obj_is_striped(dt))
4844                 RETURN(0);
4845
4846         /* destroy all striped objects */
4847         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
4848                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
4849                         if (lo->ldo_stripe[i] == NULL)
4850                                 continue;
4851                         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SPEOBJ) ||
4852                             i == cfs_fail_val) {
4853                                 dt_write_lock(env, lo->ldo_stripe[i],
4854                                               MOR_TGT_CHILD);
4855                                 rc = lod_sub_ref_del(env, lo->ldo_stripe[i],
4856                                                      th);
4857                                 dt_write_unlock(env, lo->ldo_stripe[i]);
4858                                 if (rc != 0)
4859                                         break;
4860
4861                                 rc = lod_sub_destroy(env, lo->ldo_stripe[i],
4862                                                      th);
4863                                 if (rc != 0)
4864                                         break;
4865                         }
4866                 }
4867         } else {
4868                 struct lod_obj_stripe_cb_data data = { { 0 } };
4869
4870                 data.locd_declare = false;
4871                 data.locd_stripe_cb = lod_obj_stripe_destroy_cb;
4872                 rc = lod_obj_for_each_stripe(env, lo, th, &data);
4873         }
4874
4875         RETURN(rc);
4876 }
4877
4878 /**
4879  * Implementation of dt_object_operations::do_declare_ref_add.
4880  *
4881  * \see dt_object_operations::do_declare_ref_add() in the API description
4882  * for details.
4883  */
4884 static int lod_declare_ref_add(const struct lu_env *env,
4885                                struct dt_object *dt, struct thandle *th)
4886 {
4887         return lod_sub_declare_ref_add(env, dt_object_child(dt), th);
4888 }
4889
4890 /**
4891  * Implementation of dt_object_operations::do_ref_add.
4892  *
4893  * \see dt_object_operations::do_ref_add() in the API description for details.
4894  */
4895 static int lod_ref_add(const struct lu_env *env,
4896                        struct dt_object *dt, struct thandle *th)
4897 {
4898         return lod_sub_ref_add(env, dt_object_child(dt), th);
4899 }
4900
4901 /**
4902  * Implementation of dt_object_operations::do_declare_ref_del.
4903  *
4904  * \see dt_object_operations::do_declare_ref_del() in the API description
4905  * for details.
4906  */
4907 static int lod_declare_ref_del(const struct lu_env *env,
4908                                struct dt_object *dt, struct thandle *th)
4909 {
4910         return lod_sub_declare_ref_del(env, dt_object_child(dt), th);
4911 }
4912
4913 /**
4914  * Implementation of dt_object_operations::do_ref_del
4915  *
4916  * \see dt_object_operations::do_ref_del() in the API description for details.
4917  */
4918 static int lod_ref_del(const struct lu_env *env,
4919                        struct dt_object *dt, struct thandle *th)
4920 {
4921         return lod_sub_ref_del(env, dt_object_child(dt), th);
4922 }
4923
4924 /**
4925  * Implementation of dt_object_operations::do_object_sync.
4926  *
4927  * \see dt_object_operations::do_object_sync() in the API description
4928  * for details.
4929  */
4930 static int lod_object_sync(const struct lu_env *env, struct dt_object *dt,
4931                            __u64 start, __u64 end)
4932 {
4933         return dt_object_sync(env, dt_object_child(dt), start, end);
4934 }
4935
4936 /**
4937  * Release LDLM locks on the stripes of a striped directory.
4938  *
4939  * Iterates over all the locks taken on the stripe objects and
4940  * cancel them.
4941  *
4942  * \param[in] env       execution environment
4943  * \param[in] dt        striped object
4944  * \param[in] einfo     lock description
4945  * \param[in] policy    data describing requested lock
4946  *
4947  * \retval              0 on success
4948  * \retval              negative if failed
4949  */
4950 static int lod_object_unlock_internal(const struct lu_env *env,
4951                                       struct dt_object *dt,
4952                                       struct ldlm_enqueue_info *einfo,
4953                                       union ldlm_policy_data *policy)
4954 {
4955         struct lustre_handle_array *slave_locks = einfo->ei_cbdata;
4956         int                     rc = 0;
4957         int                     i;
4958         ENTRY;
4959
4960         if (slave_locks == NULL)
4961                 RETURN(0);
4962
4963         for (i = 1; i < slave_locks->count; i++) {
4964                 if (lustre_handle_is_used(&slave_locks->handles[i]))
4965                         ldlm_lock_decref_and_cancel(&slave_locks->handles[i],
4966                                                     einfo->ei_mode);
4967         }
4968
4969         RETURN(rc);
4970 }
4971
4972 /**
4973  * Implementation of dt_object_operations::do_object_unlock.
4974  *
4975  * Used to release LDLM lock(s).
4976  *
4977  * \see dt_object_operations::do_object_unlock() in the API description
4978  * for details.
4979  */
4980 static int lod_object_unlock(const struct lu_env *env, struct dt_object *dt,
4981                              struct ldlm_enqueue_info *einfo,
4982                              union ldlm_policy_data *policy)
4983 {
4984         struct lod_object *lo = lod_dt_obj(dt);
4985         struct lustre_handle_array *slave_locks = einfo->ei_cbdata;
4986         int slave_locks_size;
4987         int i;
4988         ENTRY;
4989
4990         if (slave_locks == NULL)
4991                 RETURN(0);
4992
4993         LASSERT(S_ISDIR(dt->do_lu.lo_header->loh_attr));
4994         LASSERT(lo->ldo_dir_stripe_count > 1);
4995         /* Note: for remote lock for single stripe dir, MDT will cancel
4996          * the lock by lockh directly */
4997         LASSERT(!dt_object_remote(dt_object_child(dt)));
4998
4999         /* locks were unlocked in MDT layer */
5000         for (i = 1; i < slave_locks->count; i++) {
5001                 LASSERT(!lustre_handle_is_used(&slave_locks->handles[i]));
5002                 dt_invalidate(env, lo->ldo_stripe[i]);
5003         }
5004
5005         slave_locks_size = sizeof(*slave_locks) + slave_locks->count *
5006                            sizeof(slave_locks->handles[0]);
5007         OBD_FREE(slave_locks, slave_locks_size);
5008         einfo->ei_cbdata = NULL;
5009
5010         RETURN(0);
5011 }
5012
5013 /**
5014  * Implementation of dt_object_operations::do_object_lock.
5015  *
5016  * Used to get LDLM lock on the non-striped and striped objects.
5017  *
5018  * \see dt_object_operations::do_object_lock() in the API description
5019  * for details.
5020  */
5021 static int lod_object_lock(const struct lu_env *env,
5022                            struct dt_object *dt,
5023                            struct lustre_handle *lh,
5024                            struct ldlm_enqueue_info *einfo,
5025                            union ldlm_policy_data *policy)
5026 {
5027         struct lod_object       *lo = lod_dt_obj(dt);
5028         int                     rc = 0;
5029         int                     i;
5030         int                     slave_locks_size;
5031         struct lustre_handle_array *slave_locks = NULL;
5032         ENTRY;
5033
5034         /* remote object lock */
5035         if (!einfo->ei_enq_slave) {
5036                 LASSERT(dt_object_remote(dt));
5037                 return dt_object_lock(env, dt_object_child(dt), lh, einfo,
5038                                       policy);
5039         }
5040
5041         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
5042                 GOTO(out, rc = -ENOTDIR);
5043
5044         rc = lod_load_striping(env, lo);
5045         if (rc != 0)
5046                 GOTO(out, rc);
5047
5048         /* No stripes */
5049         if (lo->ldo_dir_stripe_count <= 1) {
5050                 /*
5051                  * NB, ei_cbdata stores pointer to slave locks, if no locks
5052                  * taken, make sure it's set to NULL, otherwise MDT will try to
5053                  * unlock them.
5054                  */
5055                 einfo->ei_cbdata = NULL;
5056                 GOTO(out, rc = 0);
5057         }
5058
5059         slave_locks_size = sizeof(*slave_locks) + lo->ldo_dir_stripe_count *
5060                            sizeof(slave_locks->handles[0]);
5061         /* Freed in lod_object_unlock */
5062         OBD_ALLOC(slave_locks, slave_locks_size);
5063         if (slave_locks == NULL)
5064                 GOTO(out, rc = -ENOMEM);
5065         slave_locks->count = lo->ldo_dir_stripe_count;
5066
5067         /* striped directory lock */
5068         for (i = 1; i < lo->ldo_dir_stripe_count; i++) {
5069                 struct lustre_handle    lockh;
5070                 struct ldlm_res_id      *res_id;
5071
5072                 res_id = &lod_env_info(env)->lti_res_id;
5073                 fid_build_reg_res_name(lu_object_fid(&lo->ldo_stripe[i]->do_lu),
5074                                        res_id);
5075                 einfo->ei_res_id = res_id;
5076
5077                 LASSERT(lo->ldo_stripe[i] != NULL);
5078                 if (likely(dt_object_remote(lo->ldo_stripe[i]))) {
5079                         rc = dt_object_lock(env, lo->ldo_stripe[i], &lockh,
5080                                             einfo, policy);
5081                 } else {
5082                         struct ldlm_namespace *ns = einfo->ei_namespace;
5083                         ldlm_blocking_callback blocking = einfo->ei_cb_local_bl;
5084                         ldlm_completion_callback completion = einfo->ei_cb_cp;
5085                         __u64   dlmflags = LDLM_FL_ATOMIC_CB;
5086
5087                         if (einfo->ei_mode == LCK_PW ||
5088                             einfo->ei_mode == LCK_EX)
5089                                 dlmflags |= LDLM_FL_COS_INCOMPAT;
5090
5091                         /* This only happens if there are mulitple stripes
5092                          * on the master MDT, i.e. except stripe0, there are
5093                          * other stripes on the Master MDT as well, Only
5094                          * happens in the test case right now. */
5095                         LASSERT(ns != NULL);
5096                         rc = ldlm_cli_enqueue_local(ns, res_id, LDLM_IBITS,
5097                                                     policy, einfo->ei_mode,
5098                                                     &dlmflags, blocking,
5099                                                     completion, NULL,
5100                                                     NULL, 0, LVB_T_NONE,
5101                                                     NULL, &lockh);
5102                 }
5103                 if (rc != 0)
5104                         break;
5105                 slave_locks->handles[i] = lockh;
5106         }
5107         einfo->ei_cbdata = slave_locks;
5108
5109         if (rc != 0 && slave_locks != NULL) {
5110                 lod_object_unlock_internal(env, dt, einfo, policy);
5111                 OBD_FREE(slave_locks, slave_locks_size);
5112         }
5113         EXIT;
5114 out:
5115         if (rc != 0)
5116                 einfo->ei_cbdata = NULL;
5117         RETURN(rc);
5118 }
5119
5120 /**
5121  * Implementation of dt_object_operations::do_invalidate.
5122  *
5123  * \see dt_object_operations::do_invalidate() in the API description for details
5124  */
5125 static int lod_invalidate(const struct lu_env *env, struct dt_object *dt)
5126 {
5127         return dt_invalidate(env, dt_object_child(dt));
5128 }
5129
5130 static int lod_layout_data_init(struct lod_thread_info *info, __u32 comp_cnt)
5131 {
5132         ENTRY;
5133
5134         /* clear memory region that will be used for layout change */
5135         memset(&info->lti_layout_attr, 0, sizeof(struct lu_attr));
5136         info->lti_count = 0;
5137
5138         if (info->lti_comp_size >= comp_cnt)
5139                 RETURN(0);
5140
5141         if (info->lti_comp_size > 0) {
5142                 OBD_FREE(info->lti_comp_idx,
5143                          info->lti_comp_size * sizeof(__u32));
5144                 info->lti_comp_size = 0;
5145         }
5146
5147         OBD_ALLOC(info->lti_comp_idx, comp_cnt * sizeof(__u32));
5148         if (!info->lti_comp_idx)
5149                 RETURN(-ENOMEM);
5150
5151         info->lti_comp_size = comp_cnt;
5152         RETURN(0);
5153 }
5154
5155 static int lod_declare_instantiate_components(const struct lu_env *env,
5156                 struct lod_object *lo, struct thandle *th)
5157 {
5158         struct lod_thread_info *info = lod_env_info(env);
5159         struct ost_pool *inuse = &info->lti_inuse_osts;
5160         int i;
5161         int rc = 0;
5162         ENTRY;
5163
5164         LASSERT(info->lti_count < lo->ldo_comp_cnt);
5165         if (info->lti_count > 0) {
5166                 /* Prepare inuse array for composite file */
5167                 rc = lod_prepare_inuse(env, lo);
5168                 if (rc)
5169                         RETURN(rc);
5170         }
5171
5172         for (i = 0; i < info->lti_count; i++) {
5173                 rc = lod_qos_prep_create(env, lo, NULL, th,
5174                                          info->lti_comp_idx[i], inuse);
5175                 if (rc)
5176                         break;
5177         }
5178
5179         if (!rc) {
5180                 info->lti_buf.lb_len = lod_comp_md_size(lo, false);
5181                 rc = lod_sub_declare_xattr_set(env, lod_object_child(lo),
5182                                 &info->lti_buf, XATTR_NAME_LOV, 0, th);
5183         }
5184
5185         RETURN(rc);
5186 }
5187
5188 static int lod_declare_update_plain(const struct lu_env *env,
5189                 struct lod_object *lo, struct layout_intent *layout,
5190                 const struct lu_buf *buf, struct thandle *th)
5191 {
5192         struct lod_thread_info *info = lod_env_info(env);
5193         struct lod_device *d = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
5194         struct lod_layout_component *lod_comp;
5195         struct lov_comp_md_v1 *comp_v1 = NULL;
5196         bool replay = false;
5197         int i, rc;
5198         ENTRY;
5199
5200         LASSERT(lo->ldo_flr_state == LCM_FL_NOT_FLR);
5201
5202         /*
5203          * In case the client is passing lovea, which only happens during
5204          * the replay of layout intent write RPC for now, we may need to
5205          * parse the lovea and apply new layout configuration.
5206          */
5207         if (buf && buf->lb_len)  {
5208                 struct lov_user_md_v1 *v1 = buf->lb_buf;
5209
5210                 if (v1->lmm_magic != (LOV_MAGIC_DEFINED | LOV_MAGIC_COMP_V1) &&
5211                     v1->lmm_magic != __swab32(LOV_MAGIC_DEFINED |
5212                                               LOV_MAGIC_COMP_V1)) {
5213                         CERROR("%s: the replay buffer of layout extend "
5214                                "(magic %#x) does not contain expected "
5215                                "composite layout.\n",
5216                                lod2obd(d)->obd_name, v1->lmm_magic);
5217                         GOTO(out, rc = -EINVAL);
5218                 }
5219
5220                 lod_object_free_striping(env, lo);
5221                 rc = lod_use_defined_striping(env, lo, buf);
5222                 if (rc)
5223                         GOTO(out, rc);
5224
5225                 rc = lod_get_lov_ea(env, lo);
5226                 if (rc <= 0)
5227                         GOTO(out, rc);
5228                 /* old on-disk EA is stored in info->lti_buf */
5229                 comp_v1 = (struct lov_comp_md_v1 *)info->lti_buf.lb_buf;
5230                 replay = true;
5231         } else {
5232                 /* non replay path */
5233                 rc = lod_load_striping_locked(env, lo);
5234                 if (rc)
5235                         GOTO(out, rc);
5236         }
5237
5238         /* Make sure defined layout covers the requested write range. */
5239         lod_comp = &lo->ldo_comp_entries[lo->ldo_comp_cnt - 1];
5240         if (lo->ldo_comp_cnt > 1 &&
5241             lod_comp->llc_extent.e_end != OBD_OBJECT_EOF &&
5242             lod_comp->llc_extent.e_end < layout->li_extent.e_end) {
5243                 CDEBUG(replay ? D_ERROR : D_LAYOUT,
5244                        "%s: the defined layout [0, %#llx) does not covers "
5245                        "the write range "DEXT"\n",
5246                        lod2obd(d)->obd_name, lod_comp->llc_extent.e_end,
5247                        PEXT(&layout->li_extent));
5248                 GOTO(out, rc = -EINVAL);
5249         }
5250
5251         CDEBUG(D_LAYOUT, "%s: "DFID": instantiate components "DEXT"\n",
5252                lod2obd(d)->obd_name, PFID(lod_object_fid(lo)),
5253                PEXT(&layout->li_extent));
5254
5255         /*
5256          * Iterate ld->ldo_comp_entries, find the component whose extent under
5257          * the write range and not instantianted.
5258          */
5259         for (i = 0; i < lo->ldo_comp_cnt; i++) {
5260                 lod_comp = &lo->ldo_comp_entries[i];
5261
5262                 if (lod_comp->llc_extent.e_start >= layout->li_extent.e_end)
5263                         break;
5264
5265                 if (!replay) {
5266                         if (lod_comp_inited(lod_comp))
5267                                 continue;
5268                 } else {
5269                         /**
5270                          * In replay path, lod_comp is the EA passed by
5271                          * client replay buffer,  comp_v1 is the pre-recovery
5272                          * on-disk EA, we'd sift out those components which
5273                          * were init-ed in the on-disk EA.
5274                          */
5275                         if (le32_to_cpu(comp_v1->lcm_entries[i].lcme_flags) &
5276                             LCME_FL_INIT)
5277                                 continue;
5278                 }
5279                 /*
5280                  * this component hasn't instantiated in normal path, or during
5281                  * replay it needs replay the instantiation.
5282                  */
5283
5284                 /* A released component is being extended */
5285                 if (lod_comp->llc_pattern & LOV_PATTERN_F_RELEASED)
5286                         GOTO(out, rc = -EINVAL);
5287
5288                 LASSERT(info->lti_comp_idx != NULL);
5289                 info->lti_comp_idx[info->lti_count++] = i;
5290         }
5291
5292         if (info->lti_count == 0)
5293                 RETURN(-EALREADY);
5294
5295         lod_obj_inc_layout_gen(lo);
5296         rc = lod_declare_instantiate_components(env, lo, th);
5297 out:
5298         if (rc)
5299                 lod_object_free_striping(env, lo);
5300         RETURN(rc);
5301 }
5302
5303 #define lod_foreach_mirror_comp(comp, lo, mirror_idx)                      \
5304 for (comp = &lo->ldo_comp_entries[lo->ldo_mirrors[mirror_idx].lme_start];  \
5305      comp <= &lo->ldo_comp_entries[lo->ldo_mirrors[mirror_idx].lme_end];   \
5306      comp++)
5307
5308 static inline int lod_comp_index(struct lod_object *lo,
5309                                  struct lod_layout_component *lod_comp)
5310 {
5311         LASSERT(lod_comp >= lo->ldo_comp_entries &&
5312                 lod_comp <= &lo->ldo_comp_entries[lo->ldo_comp_cnt - 1]);
5313
5314         return lod_comp - lo->ldo_comp_entries;
5315 }
5316
5317 /**
5318  * Stale other mirrors by writing extent.
5319  */
5320 static void lod_stale_components(struct lod_object *lo, int primary,
5321                                  struct lu_extent *extent)
5322 {
5323         struct lod_layout_component *pri_comp, *lod_comp;
5324         int i;
5325
5326         /* The writing extent decides which components in the primary
5327          * are affected... */
5328         lod_foreach_mirror_comp(pri_comp, lo, primary) {
5329                 if (!lu_extent_is_overlapped(extent, &pri_comp->llc_extent))
5330                         continue;
5331
5332                 for (i = 0; i < lo->ldo_mirror_count; i++) {
5333                         if (i == primary)
5334                                 continue;
5335
5336                         /* ... and then stale other components that are
5337                          * overlapping with primary components */
5338                         lod_foreach_mirror_comp(lod_comp, lo, i) {
5339                                 if (!lu_extent_is_overlapped(
5340                                                         &pri_comp->llc_extent,
5341                                                         &lod_comp->llc_extent))
5342                                         continue;
5343
5344                                 CDEBUG(D_LAYOUT, "stale: %u / %u\n",
5345                                       i, lod_comp_index(lo, lod_comp));
5346
5347                                 lod_comp->llc_flags |= LCME_FL_STALE;
5348                                 lo->ldo_mirrors[i].lme_stale = 1;
5349                         }
5350                 }
5351         }
5352 }
5353
5354 static int lod_declare_update_rdonly(const struct lu_env *env,
5355                 struct lod_object *lo, struct md_layout_change *mlc,
5356                 struct thandle *th)
5357 {
5358         struct lod_thread_info *info = lod_env_info(env);
5359         struct lu_attr *layout_attr = &info->lti_layout_attr;
5360         struct lod_layout_component *lod_comp;
5361         struct layout_intent *layout = mlc->mlc_intent;
5362         struct lu_extent extent = layout->li_extent;
5363         int picked;
5364         int i;
5365         int rc;
5366         ENTRY;
5367
5368         LASSERT(mlc->mlc_opc == MD_LAYOUT_WRITE);
5369         LASSERT(lo->ldo_flr_state == LCM_FL_RDONLY);
5370         LASSERT(lo->ldo_mirror_count > 0);
5371
5372         CDEBUG(D_LAYOUT, DFID": trying to write :"DEXT"\n",
5373                PFID(lod_object_fid(lo)), PEXT(&extent));
5374
5375         /**
5376          * Pick a mirror as the primary.
5377          * Now it only picks the first mirror, this algo can be
5378          * revised later after knowing the topology of cluster or
5379          * the availability of OSTs.
5380          */
5381         for (picked = -1, i = 0; i < lo->ldo_mirror_count; i++) {
5382                 if (!lo->ldo_mirrors[i].lme_stale) {
5383                         picked = i;
5384                         break;
5385                 }
5386         }
5387         if (picked < 0) /* failed to pick a primary */
5388                 RETURN(-ENODATA);
5389
5390         CDEBUG(D_LAYOUT, DFID": picked mirror %u as primary\n",
5391                PFID(lod_object_fid(lo)), lo->ldo_mirrors[picked].lme_id);
5392
5393         /* stale overlapping components from other mirrors */
5394         lod_stale_components(lo, picked, &extent);
5395
5396         /* instantiate components for the picked mirror, start from 0 */
5397         extent = (struct lu_extent) { 0, layout->li_extent.e_end };
5398         lod_foreach_mirror_comp(lod_comp, lo, picked) {
5399                 if (!lu_extent_is_overlapped(&extent,
5400                                              &lod_comp->llc_extent))
5401                         break;
5402
5403                 if (lod_comp_inited(lod_comp))
5404                         continue;
5405
5406                 CDEBUG(D_LAYOUT, "instantiate: %u / %u\n",
5407                        i, lod_comp_index(lo, lod_comp));
5408
5409                 info->lti_comp_idx[info->lti_count++] =
5410                                                 lod_comp_index(lo, lod_comp);
5411         }
5412
5413         lo->ldo_flr_state = LCM_FL_WRITE_PENDING;
5414
5415         /* Reset the layout version once it's becoming too large.
5416          * This way it can make sure that the layout version is
5417          * monotonously increased in this writing era. */
5418         lod_obj_inc_layout_gen(lo);
5419         if (lo->ldo_layout_gen > (LCME_ID_MAX >> 1)) {
5420                 __u32 layout_version;
5421
5422                 cfs_get_random_bytes(&layout_version, sizeof(layout_version));
5423                 lo->ldo_layout_gen = layout_version & 0xffff;
5424         }
5425
5426         rc = lod_declare_instantiate_components(env, lo, th);
5427         if (rc)
5428                 GOTO(out, rc);
5429
5430         layout_attr->la_valid = LA_LAYOUT_VERSION;
5431         layout_attr->la_layout_version = 0; /* set current version */
5432         rc = lod_declare_attr_set(env, &lo->ldo_obj, layout_attr, th);
5433         if (rc)
5434                 GOTO(out, rc);
5435
5436 out:
5437         if (rc)
5438                 lod_object_free_striping(env, lo);
5439         RETURN(rc);
5440 }
5441
5442 static int lod_declare_update_write_pending(const struct lu_env *env,
5443                 struct lod_object *lo, struct md_layout_change *mlc,
5444                 struct thandle *th)
5445 {
5446         struct lod_thread_info *info = lod_env_info(env);
5447         struct lu_attr *layout_attr = &info->lti_layout_attr;
5448         struct lod_layout_component *lod_comp;
5449         struct lu_extent extent = { 0 };
5450         int primary = -1;
5451         int i;
5452         int rc;
5453         ENTRY;
5454
5455         LASSERT(lo->ldo_flr_state == LCM_FL_WRITE_PENDING);
5456         LASSERT(mlc->mlc_opc == MD_LAYOUT_WRITE);
5457
5458         /* look for the primary mirror */
5459         for (i = 0; i < lo->ldo_mirror_count; i++) {
5460                 if (lo->ldo_mirrors[i].lme_stale)
5461                         continue;
5462
5463                 LASSERTF(primary < 0, DFID " has multiple primary: %u / %u",
5464                          PFID(lod_object_fid(lo)),
5465                          lo->ldo_mirrors[i].lme_id,
5466                          lo->ldo_mirrors[primary].lme_id);
5467
5468                 primary = i;
5469         }
5470         if (primary < 0) {
5471                 CERROR(DFID ": doesn't have a primary mirror\n",
5472                        PFID(lod_object_fid(lo)));
5473                 GOTO(out, rc = -ENODATA);
5474         }
5475
5476         CDEBUG(D_LAYOUT, DFID": found primary %u\n",
5477                PFID(lod_object_fid(lo)), lo->ldo_mirrors[primary].lme_id);
5478
5479         LASSERT(!lo->ldo_mirrors[primary].lme_stale);
5480
5481         /* for LAYOUT_WRITE opc, it has to do the following operations:
5482          * 1. stale overlapping componets from stale mirrors;
5483          * 2. instantiate components of the primary mirror;
5484          * 3. transfter layout version to all objects of the primary; */
5485
5486         if (mlc->mlc_opc == MD_LAYOUT_WRITE) {
5487                 LASSERT(mlc->mlc_intent != NULL);
5488
5489                 extent = mlc->mlc_intent->li_extent;
5490
5491                 CDEBUG(D_LAYOUT, DFID": intent to write: "DEXT"\n",
5492                        PFID(lod_object_fid(lo)), PEXT(&extent));
5493
5494                 /* 1. stale overlapping components */
5495                 lod_stale_components(lo, primary, &extent);
5496
5497                 /* 2. find out the components need instantiating.
5498                  * instantiate [0, mlc->mlc_intent->e_end) */
5499                 extent.e_start = 0;
5500                 lod_foreach_mirror_comp(lod_comp, lo, primary) {
5501                         if (!lu_extent_is_overlapped(&extent,
5502                                                      &lod_comp->llc_extent))
5503                                 break;
5504
5505                         if (lod_comp_inited(lod_comp))
5506                                 continue;
5507
5508                         CDEBUG(D_LAYOUT, "write instantiate %d / %d\n",
5509                                primary, lod_comp_index(lo, lod_comp));
5510                         info->lti_comp_idx[info->lti_count++] =
5511                                                 lod_comp_index(lo, lod_comp);
5512                 }
5513         }
5514
5515         rc = lod_declare_instantiate_components(env, lo, th);
5516         if (rc)
5517                 GOTO(out, rc);
5518
5519         layout_attr->la_valid = LA_LAYOUT_VERSION;
5520         layout_attr->la_layout_version = 0; /* set current version */
5521         rc = lod_declare_attr_set(env, &lo->ldo_obj, layout_attr, th);
5522         if (rc)
5523                 GOTO(out, rc);
5524
5525         lod_obj_inc_layout_gen(lo);
5526 out:
5527         if (rc)
5528                 lod_object_free_striping(env, lo);
5529         RETURN(rc);
5530 }
5531
5532 static int lod_declare_layout_change(const struct lu_env *env,
5533                 struct dt_object *dt, struct md_layout_change *mlc,
5534                 struct thandle *th)
5535 {
5536         struct lod_thread_info  *info = lod_env_info(env);
5537         struct lod_object *lo = lod_dt_obj(dt);
5538         int rc;
5539         ENTRY;
5540
5541         if (!S_ISREG(dt->do_lu.lo_header->loh_attr) || !dt_object_exists(dt) ||
5542             dt_object_remote(dt_object_child(dt)))
5543                 RETURN(-EINVAL);
5544
5545         lod_write_lock(env, dt, 0);
5546         rc = lod_load_striping_locked(env, lo);
5547         if (rc)
5548                 GOTO(out, rc);
5549
5550         LASSERT(lo->ldo_comp_cnt > 0);
5551
5552         rc = lod_layout_data_init(info, lo->ldo_comp_cnt);
5553         if (rc)
5554                 GOTO(out, rc);
5555
5556         switch (lo->ldo_flr_state) {
5557         case LCM_FL_NOT_FLR:
5558                 rc = lod_declare_update_plain(env, lo, mlc->mlc_intent,
5559                                               &mlc->mlc_buf, th);
5560                 break;
5561         case LCM_FL_RDONLY:
5562                 rc = lod_declare_update_rdonly(env, lo, mlc, th);
5563                 break;
5564         case LCM_FL_WRITE_PENDING:
5565                 rc = lod_declare_update_write_pending(env, lo, mlc, th);
5566                 break;
5567         case LCM_FL_SYNC_PENDING:
5568         default:
5569                 rc = -ENOTSUPP;
5570                 break;
5571         }
5572 out:
5573         dt_write_unlock(env, dt);
5574         RETURN(rc);
5575 }
5576
5577 /**
5578  * Instantiate layout component objects which covers the intent write offset.
5579  */
5580 static int lod_layout_change(const struct lu_env *env, struct dt_object *dt,
5581                              struct md_layout_change *mlc, struct thandle *th)
5582 {
5583         struct lu_attr *attr = &lod_env_info(env)->lti_attr;
5584         struct lu_attr *layout_attr = &lod_env_info(env)->lti_layout_attr;
5585         struct lod_object *lo = lod_dt_obj(dt);
5586         int rc;
5587
5588         rc = lod_striped_create(env, dt, attr, NULL, th);
5589         if (!rc && layout_attr->la_valid & LA_LAYOUT_VERSION) {
5590                 layout_attr->la_layout_version |= lo->ldo_layout_gen;
5591                 rc = lod_attr_set(env, dt, layout_attr, th);
5592         }
5593
5594         return rc;
5595 }
5596
5597 struct dt_object_operations lod_obj_ops = {
5598         .do_read_lock           = lod_read_lock,
5599         .do_write_lock          = lod_write_lock,
5600         .do_read_unlock         = lod_read_unlock,
5601         .do_write_unlock        = lod_write_unlock,
5602         .do_write_locked        = lod_write_locked,
5603         .do_attr_get            = lod_attr_get,
5604         .do_declare_attr_set    = lod_declare_attr_set,
5605         .do_attr_set            = lod_attr_set,
5606         .do_xattr_get           = lod_xattr_get,
5607         .do_declare_xattr_set   = lod_declare_xattr_set,
5608         .do_xattr_set           = lod_xattr_set,
5609         .do_declare_xattr_del   = lod_declare_xattr_del,
5610         .do_xattr_del           = lod_xattr_del,
5611         .do_xattr_list          = lod_xattr_list,
5612         .do_ah_init             = lod_ah_init,
5613         .do_declare_create      = lod_declare_create,
5614         .do_create              = lod_create,
5615         .do_declare_destroy     = lod_declare_destroy,
5616         .do_destroy             = lod_destroy,
5617         .do_index_try           = lod_index_try,
5618         .do_declare_ref_add     = lod_declare_ref_add,
5619         .do_ref_add             = lod_ref_add,
5620         .do_declare_ref_del     = lod_declare_ref_del,
5621         .do_ref_del             = lod_ref_del,
5622         .do_object_sync         = lod_object_sync,
5623         .do_object_lock         = lod_object_lock,
5624         .do_object_unlock       = lod_object_unlock,
5625         .do_invalidate          = lod_invalidate,
5626         .do_declare_layout_change = lod_declare_layout_change,
5627         .do_layout_change       = lod_layout_change,
5628 };
5629
5630 /**
5631  * Implementation of dt_body_operations::dbo_read.
5632  *
5633  * \see dt_body_operations::dbo_read() in the API description for details.
5634  */
5635 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
5636                         struct lu_buf *buf, loff_t *pos)
5637 {
5638         struct dt_object *next = dt_object_child(dt);
5639
5640         LASSERT(S_ISREG(dt->do_lu.lo_header->loh_attr) ||
5641                 S_ISLNK(dt->do_lu.lo_header->loh_attr));
5642         return next->do_body_ops->dbo_read(env, next, buf, pos);
5643 }
5644
5645 /**
5646  * Implementation of dt_body_operations::dbo_declare_write.
5647  *
5648  * \see dt_body_operations::dbo_declare_write() in the API description
5649  * for details.
5650  */
5651 static ssize_t lod_declare_write(const struct lu_env *env,
5652                                  struct dt_object *dt,
5653                                  const struct lu_buf *buf, loff_t pos,
5654                                  struct thandle *th)
5655 {
5656         return lod_sub_declare_write(env, dt_object_child(dt), buf, pos, th);
5657 }
5658
5659 /**
5660  * Implementation of dt_body_operations::dbo_write.
5661  *
5662  * \see dt_body_operations::dbo_write() in the API description for details.
5663  */
5664 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,
5665                          const struct lu_buf *buf, loff_t *pos,
5666                          struct thandle *th, int iq)
5667 {
5668         LASSERT(S_ISREG(dt->do_lu.lo_header->loh_attr) ||
5669                 S_ISLNK(dt->do_lu.lo_header->loh_attr));
5670         return lod_sub_write(env, dt_object_child(dt), buf, pos, th, iq);
5671 }
5672
5673 static int lod_declare_punch(const struct lu_env *env, struct dt_object *dt,
5674                              __u64 start, __u64 end, struct thandle *th)
5675 {
5676         if (dt_object_remote(dt))
5677                 return -ENOTSUPP;
5678
5679         return lod_sub_declare_punch(env, dt_object_child(dt), start, end, th);
5680 }
5681
5682 static int lod_punch(const struct lu_env *env, struct dt_object *dt,
5683                      __u64 start, __u64 end, struct thandle *th)
5684 {
5685         if (dt_object_remote(dt))
5686                 return -ENOTSUPP;
5687
5688         LASSERT(S_ISREG(dt->do_lu.lo_header->loh_attr));
5689         return lod_sub_punch(env, dt_object_child(dt), start, end, th);
5690 }
5691
5692 /*
5693  * different type of files use the same body_ops because object may be created
5694  * in OUT, where there is no chance to set correct body_ops for each type, so
5695  * body_ops themselves will check file type inside, see lod_read/write/punch for
5696  * details.
5697  */
5698 const struct dt_body_operations lod_body_ops = {
5699         .dbo_read               = lod_read,
5700         .dbo_declare_write      = lod_declare_write,
5701         .dbo_write              = lod_write,
5702         .dbo_declare_punch      = lod_declare_punch,
5703         .dbo_punch              = lod_punch,
5704 };
5705
5706 /**
5707  * Implementation of lu_object_operations::loo_object_init.
5708  *
5709  * The function determines the type and the index of the target device using
5710  * sequence of the object's FID. Then passes control down to the
5711  * corresponding device:
5712  *  OSD for the local objects, OSP for remote
5713  *
5714  * \see lu_object_operations::loo_object_init() in the API description
5715  * for details.
5716  */
5717 static int lod_object_init(const struct lu_env *env, struct lu_object *lo,
5718                            const struct lu_object_conf *conf)
5719 {
5720         struct lod_device       *lod    = lu2lod_dev(lo->lo_dev);
5721         struct lu_device        *cdev   = NULL;
5722         struct lu_object        *cobj;
5723         struct lod_tgt_descs    *ltd    = NULL;
5724         struct lod_tgt_desc     *tgt;
5725         u32                      idx    = 0;
5726         int                      type   = LU_SEQ_RANGE_ANY;
5727         int                      rc;
5728         ENTRY;
5729
5730         rc = lod_fld_lookup(env, lod, lu_object_fid(lo), &idx, &type);
5731         if (rc != 0) {
5732                 /* Note: Sometimes, it will Return EAGAIN here, see
5733                  * ptrlpc_import_delay_req(), which might confuse
5734                  * lu_object_find_at() and make it wait there incorrectly.
5735                  * so we convert it to EIO here.*/
5736                 if (rc == -EAGAIN)
5737                         rc = -EIO;
5738
5739                 RETURN(rc);
5740         }
5741
5742         if (type == LU_SEQ_RANGE_MDT &&
5743             idx == lu_site2seq(lo->lo_dev->ld_site)->ss_node_id) {
5744                 cdev = &lod->lod_child->dd_lu_dev;
5745         } else if (type == LU_SEQ_RANGE_MDT) {
5746                 ltd = &lod->lod_mdt_descs;
5747                 lod_getref(ltd);
5748         } else if (type == LU_SEQ_RANGE_OST) {
5749                 ltd = &lod->lod_ost_descs;
5750                 lod_getref(ltd);
5751         } else {
5752                 LBUG();
5753         }
5754
5755         if (ltd != NULL) {
5756                 if (ltd->ltd_tgts_size > idx &&
5757                     cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx)) {
5758                         tgt = LTD_TGT(ltd, idx);
5759
5760                         LASSERT(tgt != NULL);
5761                         LASSERT(tgt->ltd_tgt != NULL);
5762
5763                         cdev = &(tgt->ltd_tgt->dd_lu_dev);
5764                 }
5765                 lod_putref(lod, ltd);
5766         }
5767
5768         if (unlikely(cdev == NULL))
5769                 RETURN(-ENOENT);
5770
5771         cobj = cdev->ld_ops->ldo_object_alloc(env, lo->lo_header, cdev);
5772         if (unlikely(cobj == NULL))
5773                 RETURN(-ENOMEM);
5774
5775         lu2lod_obj(lo)->ldo_obj.do_body_ops = &lod_body_ops;
5776
5777         lu_object_add(lo, cobj);
5778
5779         RETURN(0);
5780 }
5781
5782 /**
5783  *
5784  * Release resources associated with striping.
5785  *
5786  * If the object is striped (regular or directory), then release
5787  * the stripe objects references and free the ldo_stripe array.
5788  *
5789  * \param[in] env       execution environment
5790  * \param[in] lo        object
5791  */
5792 void lod_object_free_striping(const struct lu_env *env, struct lod_object *lo)
5793 {
5794         struct lod_layout_component *lod_comp;
5795         int i, j;
5796
5797         if (lo->ldo_stripe != NULL) {
5798                 LASSERT(lo->ldo_comp_entries == NULL);
5799                 LASSERT(lo->ldo_dir_stripes_allocated > 0);
5800
5801                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
5802                         if (lo->ldo_stripe[i])
5803                                 dt_object_put(env, lo->ldo_stripe[i]);
5804                 }
5805
5806                 j = sizeof(struct dt_object *) * lo->ldo_dir_stripes_allocated;
5807                 OBD_FREE(lo->ldo_stripe, j);
5808                 lo->ldo_stripe = NULL;
5809                 lo->ldo_dir_stripes_allocated = 0;
5810                 lo->ldo_dir_stripe_loaded = 0;
5811                 lo->ldo_dir_stripe_count = 0;
5812         } else if (lo->ldo_comp_entries != NULL) {
5813                 for (i = 0; i < lo->ldo_comp_cnt; i++) {
5814                         /* free lod_layout_component::llc_stripe array */
5815                         lod_comp = &lo->ldo_comp_entries[i];
5816
5817                         if (lod_comp->llc_stripe == NULL)
5818                                 continue;
5819                         LASSERT(lod_comp->llc_stripes_allocated != 0);
5820                         for (j = 0; j < lod_comp->llc_stripes_allocated; j++) {
5821                                 if (lod_comp->llc_stripe[j] != NULL)
5822                                         lu_object_put(env,
5823                                                &lod_comp->llc_stripe[j]->do_lu);
5824                         }
5825                         OBD_FREE(lod_comp->llc_stripe,
5826                                  sizeof(struct dt_object *) *
5827                                  lod_comp->llc_stripes_allocated);
5828                         lod_comp->llc_stripe = NULL;
5829                         lod_comp->llc_stripes_allocated = 0;
5830                 }
5831                 lod_free_comp_entries(lo);
5832                 lo->ldo_comp_cached = 0;
5833         }
5834 }
5835
5836 /**
5837  * Implementation of lu_object_operations::loo_object_free.
5838  *
5839  * \see lu_object_operations::loo_object_free() in the API description
5840  * for details.
5841  */
5842 static void lod_object_free(const struct lu_env *env, struct lu_object *o)
5843 {
5844         struct lod_object *lo = lu2lod_obj(o);
5845
5846         /* release all underlying object pinned */
5847         lod_object_free_striping(env, lo);
5848         lu_object_fini(o);
5849         OBD_SLAB_FREE_PTR(lo, lod_object_kmem);
5850 }
5851
5852 /**
5853  * Implementation of lu_object_operations::loo_object_release.
5854  *
5855  * \see lu_object_operations::loo_object_release() in the API description
5856  * for details.
5857  */
5858 static void lod_object_release(const struct lu_env *env, struct lu_object *o)
5859 {
5860         /* XXX: shouldn't we release everything here in case if object
5861          * creation failed before? */
5862 }
5863
5864 /**
5865  * Implementation of lu_object_operations::loo_object_print.
5866  *
5867  * \see lu_object_operations::loo_object_print() in the API description
5868  * for details.
5869  */
5870 static int lod_object_print(const struct lu_env *env, void *cookie,
5871                             lu_printer_t p, const struct lu_object *l)
5872 {
5873         struct lod_object *o = lu2lod_obj((struct lu_object *) l);
5874
5875         return (*p)(env, cookie, LUSTRE_LOD_NAME"-object@%p", o);
5876 }
5877
5878 struct lu_object_operations lod_lu_obj_ops = {
5879         .loo_object_init        = lod_object_init,
5880         .loo_object_free        = lod_object_free,
5881         .loo_object_release     = lod_object_release,
5882         .loo_object_print       = lod_object_print,
5883 };