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