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