Whamcloud - gitweb
LU-4665 utils: lfs setstripe to specify OSTs
[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, 2013, 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                         struct lov_user_md_v3   *v3;
1842
1843                         /* sigh, lti_ea_store has been used for lmv_buf,
1844                          * so we have to allocate buffer for default
1845                          * stripe EA */
1846                         OBD_ALLOC_PTR(v3);
1847                         if (v3 == NULL)
1848                                 GOTO(out_put, rc = -ENOMEM);
1849
1850                         memset(v3, 0, sizeof(*v3));
1851                         v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
1852                         v3->lmm_stripe_count =
1853                                 cpu_to_le16(lo->ldo_def_stripenr);
1854                         v3->lmm_stripe_offset =
1855                                 cpu_to_le16(lo->ldo_def_stripe_offset);
1856                         v3->lmm_stripe_size =
1857                                 cpu_to_le32(lo->ldo_def_stripe_size);
1858                         if (lo->ldo_pool != NULL)
1859                                 strlcpy(v3->lmm_pool_name, lo->ldo_pool,
1860                                         sizeof(v3->lmm_pool_name));
1861
1862                         info->lti_buf.lb_buf = v3;
1863                         info->lti_buf.lb_len = sizeof(*v3);
1864                         rc = dt_declare_xattr_set(env, dto,
1865                                                   &info->lti_buf,
1866                                                   XATTR_NAME_LOV,
1867                                                   0, th);
1868                         OBD_FREE_PTR(v3);
1869                         if (rc != 0)
1870                                 GOTO(out_put, rc);
1871                 }
1872
1873                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
1874                     cfs_fail_val != i) {
1875                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
1876                             cfs_fail_val == i)
1877                                 slave_lmm->lmv_master_mdt_index =
1878                                                         cpu_to_le32(i + 1);
1879                         else
1880                                 slave_lmm->lmv_master_mdt_index =
1881                                                         cpu_to_le32(i);
1882                         rc = dt_declare_xattr_set(env, dto, &slave_lmv_buf,
1883                                                   XATTR_NAME_LMV, 0, th);
1884                         if (rc != 0)
1885                                 GOTO(out_put, rc);
1886                 }
1887
1888                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
1889                     cfs_fail_val == i)
1890                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1891                                 PFID(lu_object_fid(&dto->do_lu)), i + 1);
1892                 else
1893                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1894                                 PFID(lu_object_fid(&dto->do_lu)), i);
1895
1896                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
1897                 rc = linkea_data_new(&ldata, &info->lti_linkea_buf);
1898                 if (rc != 0)
1899                         GOTO(out_put, rc);
1900
1901                 rc = linkea_add_buf(&ldata, sname, lu_object_fid(&dt->do_lu));
1902                 if (rc != 0)
1903                         GOTO(out_put, rc);
1904
1905                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
1906                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
1907                 rc = dt_declare_xattr_set(env, dto, &linkea_buf,
1908                                           XATTR_NAME_LINK, 0, th);
1909                 if (rc != 0)
1910                         GOTO(out_put, rc);
1911
1912                 rec->rec_fid = lu_object_fid(&dto->do_lu);
1913                 rc = dt_declare_insert(env, dt_object_child(dt),
1914                                        (const struct dt_rec *)rec,
1915                                        (const struct dt_key *)stripe_name, th);
1916                 if (rc != 0)
1917                         GOTO(out_put, rc);
1918
1919                 rc = dt_declare_ref_add(env, dt_object_child(dt), th);
1920                 if (rc != 0)
1921                         GOTO(out_put, rc);
1922         }
1923
1924         rc = dt_declare_xattr_set(env, dt_object_child(dt), &lmv_buf,
1925                                   XATTR_NAME_LMV, 0, th);
1926         if (rc != 0)
1927                 GOTO(out_put, rc);
1928
1929 out_put:
1930         if (rc < 0) {
1931                 for (i = 0; i < stripe_count; i++)
1932                         if (stripe[i] != NULL)
1933                                 lu_object_put(env, &stripe[i]->do_lu);
1934                 OBD_FREE(stripe, sizeof(stripe[0]) * stripe_count);
1935                 lo->ldo_stripenr = 0;
1936                 lo->ldo_stripes_allocated = 0;
1937                 lo->ldo_stripe = NULL;
1938         }
1939
1940 out_free:
1941         if (idx_array != NULL)
1942                 OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count);
1943         if (slave_lmm != NULL)
1944                 OBD_FREE_PTR(slave_lmm);
1945
1946         RETURN(rc);
1947 }
1948
1949 /**
1950  * Declare create striped md object.
1951  *
1952  * The function declares intention to create a striped directory. This is a
1953  * wrapper for lod_prep_md_striped_create(). The only additional functionality
1954  * is to verify pattern \a lum_buf is good. Check that function for the details.
1955  *
1956  * \param[in] env       execution environment
1957  * \param[in] dt        object
1958  * \param[in] attr      attributes to initialize the objects with
1959  * \param[in] lum_buf   a pattern specifying the number of stripes and
1960  *                      MDT to start from
1961  * \param[in] dof       type of objects to be created
1962  * \param[in] th        transaction handle
1963  *
1964  * \retval              0 on success
1965  * \retval              negative if failed
1966  *
1967  */
1968 static int lod_declare_xattr_set_lmv(const struct lu_env *env,
1969                                      struct dt_object *dt,
1970                                      struct lu_attr *attr,
1971                                      const struct lu_buf *lum_buf,
1972                                      struct dt_object_format *dof,
1973                                      struct thandle *th)
1974 {
1975         struct lod_object       *lo = lod_dt_obj(dt);
1976         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1977         struct lmv_user_md_v1   *lum;
1978         int                     rc;
1979         ENTRY;
1980
1981         lum = lum_buf->lb_buf;
1982         LASSERT(lum != NULL);
1983
1984         CDEBUG(D_INFO, "lum magic = %x count = %u offset = %d\n",
1985                le32_to_cpu(lum->lum_magic), le32_to_cpu(lum->lum_stripe_count),
1986                (int)le32_to_cpu(lum->lum_stripe_offset));
1987
1988         if (le32_to_cpu(lum->lum_stripe_count) == 0)
1989                 GOTO(out, rc = 0);
1990
1991         rc = lod_verify_md_striping(lod, lum);
1992         if (rc != 0)
1993                 GOTO(out, rc);
1994
1995         /* prepare dir striped objects */
1996         rc = lod_prep_md_striped_create(env, dt, attr, lum, dof, th);
1997         if (rc != 0) {
1998                 /* failed to create striping, let's reset
1999                  * config so that others don't get confused */
2000                 lod_object_free_striping(env, lo);
2001                 GOTO(out, rc);
2002         }
2003 out:
2004         RETURN(rc);
2005 }
2006
2007
2008 /**
2009  * Implementation of dt_object_operations::do_declare_xattr_set.
2010  *
2011  * Used with regular (non-striped) objects. Basically it
2012  * initializes the striping information and applies the
2013  * change to all the stripes.
2014  *
2015  * \see dt_object_operations::do_declare_xattr_set() in the API description
2016  * for details.
2017  */
2018 static int lod_dir_declare_xattr_set(const struct lu_env *env,
2019                                      struct dt_object *dt,
2020                                      const struct lu_buf *buf,
2021                                      const char *name, int fl,
2022                                      struct thandle *th)
2023 {
2024         struct dt_object        *next = dt_object_child(dt);
2025         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2026         struct lod_object       *lo = lod_dt_obj(dt);
2027         int                     i;
2028         int                     rc;
2029         ENTRY;
2030
2031         if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
2032                 struct lmv_user_md_v1 *lum;
2033
2034                 LASSERT(buf != NULL && buf->lb_buf != NULL);
2035                 lum = buf->lb_buf;
2036                 rc = lod_verify_md_striping(d, lum);
2037                 if (rc != 0)
2038                         RETURN(rc);
2039         }
2040
2041         rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
2042         if (rc != 0)
2043                 RETURN(rc);
2044
2045         /* set xattr to each stripes, if needed */
2046         rc = lod_load_striping(env, lo);
2047         if (rc != 0)
2048                 RETURN(rc);
2049
2050         /* Note: Do not set LinkEA on sub-stripes, otherwise
2051          * it will confuse the fid2path process(see mdt_path_current()).
2052          * The linkEA between master and sub-stripes is set in
2053          * lod_xattr_set_lmv(). */
2054         if (lo->ldo_stripenr == 0 || strcmp(name, XATTR_NAME_LINK) == 0)
2055                 RETURN(0);
2056
2057         for (i = 0; i < lo->ldo_stripenr; i++) {
2058                 LASSERT(lo->ldo_stripe[i]);
2059                 rc = dt_declare_xattr_set(env, lo->ldo_stripe[i], buf,
2060                                           name, fl, th);
2061                 if (rc != 0)
2062                         break;
2063         }
2064
2065         RETURN(rc);
2066 }
2067
2068 /**
2069  * Implementation of dt_object_operations::do_declare_xattr_set.
2070  *
2071  * \see dt_object_operations::do_declare_xattr_set() in the API description
2072  * for details.
2073  *
2074  * the extension to the API:
2075  *   - declaring LOVEA requests striping creation
2076  *   - LU_XATTR_REPLACE means layout swap
2077  */
2078 static int lod_declare_xattr_set(const struct lu_env *env,
2079                                  struct dt_object *dt,
2080                                  const struct lu_buf *buf,
2081                                  const char *name, int fl,
2082                                  struct thandle *th)
2083 {
2084         struct dt_object *next = dt_object_child(dt);
2085         struct lu_attr   *attr = &lod_env_info(env)->lti_attr;
2086         __u32             mode;
2087         int               rc;
2088         ENTRY;
2089
2090         /*
2091          * allow to declare predefined striping on a new (!mode) object
2092          * which is supposed to be replay of regular file creation
2093          * (when LOV setting is declared)
2094          * LU_XATTR_REPLACE is set to indicate a layout swap
2095          */
2096         mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
2097         if ((S_ISREG(mode) || mode == 0) && strcmp(name, XATTR_NAME_LOV) == 0 &&
2098              !(fl & LU_XATTR_REPLACE)) {
2099                 /*
2100                  * this is a request to manipulate object's striping
2101                  */
2102                 if (dt_object_exists(dt)) {
2103                         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
2104                         if (rc)
2105                                 RETURN(rc);
2106                 } else {
2107                         memset(attr, 0, sizeof(*attr));
2108                         attr->la_valid = LA_TYPE | LA_MODE;
2109                         attr->la_mode = S_IFREG;
2110                 }
2111                 rc = lod_declare_striped_object(env, dt, attr, buf, th);
2112         } else if (S_ISDIR(mode)) {
2113                 rc = lod_dir_declare_xattr_set(env, dt, buf, name, fl, th);
2114         } else {
2115                 rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
2116         }
2117
2118         RETURN(rc);
2119 }
2120
2121 /**
2122  * Resets cached default striping in the object.
2123  *
2124  * \param[in] lo        object
2125  */
2126 static void lod_lov_stripe_cache_clear(struct lod_object *lo)
2127 {
2128         lo->ldo_striping_cached = 0;
2129         lo->ldo_def_striping_set = 0;
2130         lod_object_set_pool(lo, NULL);
2131         lo->ldo_def_stripe_size = 0;
2132         lo->ldo_def_stripenr = 0;
2133         if (lo->ldo_dir_stripe != NULL)
2134                 lo->ldo_dir_striping_cached = 0;
2135 }
2136
2137 /**
2138  * Apply xattr changes to the object.
2139  *
2140  * Applies xattr changes to the object and the stripes if the latter exist.
2141  *
2142  * \param[in] env       execution environment
2143  * \param[in] dt        object
2144  * \param[in] buf       buffer pointing to the new value of xattr
2145  * \param[in] name      name of xattr
2146  * \param[in] fl        flags
2147  * \param[in] th        transaction handle
2148  * \param[in] capa      not used currently
2149  *
2150  * \retval              0 on success
2151  * \retval              negative if failed
2152  */
2153 static int lod_xattr_set_internal(const struct lu_env *env,
2154                                   struct dt_object *dt,
2155                                   const struct lu_buf *buf,
2156                                   const char *name, int fl, struct thandle *th,
2157                                   struct lustre_capa *capa)
2158 {
2159         struct dt_object        *next = dt_object_child(dt);
2160         struct lod_object       *lo = lod_dt_obj(dt);
2161         int                     rc;
2162         int                     i;
2163         ENTRY;
2164
2165         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
2166         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
2167                 RETURN(rc);
2168
2169         /* Note: Do not set LinkEA on sub-stripes, otherwise
2170          * it will confuse the fid2path process(see mdt_path_current()).
2171          * The linkEA between master and sub-stripes is set in
2172          * lod_xattr_set_lmv(). */
2173         if (lo->ldo_stripenr == 0 || strcmp(name, XATTR_NAME_LINK) == 0)
2174                 RETURN(0);
2175
2176         for (i = 0; i < lo->ldo_stripenr; i++) {
2177                 LASSERT(lo->ldo_stripe[i]);
2178                 rc = dt_xattr_set(env, lo->ldo_stripe[i], buf, name, fl, th,
2179                                   capa);
2180                 if (rc != 0)
2181                         break;
2182         }
2183
2184         RETURN(rc);
2185 }
2186
2187 /**
2188  * Delete an extended attribute.
2189  *
2190  * Deletes specified xattr from the object and the stripes if the latter exist.
2191  *
2192  * \param[in] env       execution environment
2193  * \param[in] dt        object
2194  * \param[in] name      name of xattr
2195  * \param[in] th        transaction handle
2196  * \param[in] capa      not used currently
2197  *
2198  * \retval              0 on success
2199  * \retval              negative if failed
2200  */
2201 static int lod_xattr_del_internal(const struct lu_env *env,
2202                                   struct dt_object *dt,
2203                                   const char *name, struct thandle *th,
2204                                   struct lustre_capa *capa)
2205 {
2206         struct dt_object        *next = dt_object_child(dt);
2207         struct lod_object       *lo = lod_dt_obj(dt);
2208         int                     rc;
2209         int                     i;
2210         ENTRY;
2211
2212         rc = dt_xattr_del(env, next, name, th, capa);
2213         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
2214                 RETURN(rc);
2215
2216         if (lo->ldo_stripenr == 0)
2217                 RETURN(rc);
2218
2219         for (i = 0; i < lo->ldo_stripenr; i++) {
2220                 LASSERT(lo->ldo_stripe[i]);
2221                 rc = dt_xattr_del(env, lo->ldo_stripe[i], name, th,
2222                                   capa);
2223                 if (rc != 0)
2224                         break;
2225         }
2226
2227         RETURN(rc);
2228 }
2229
2230 /**
2231  * Set default striping on a directory.
2232  *
2233  * Sets specified striping on a directory object unless it matches the default
2234  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
2235  * EA. This striping will be used when regular file is being created in this
2236  * directory.
2237  *
2238  * \param[in] env       execution environment
2239  * \param[in] dt        the striped object
2240  * \param[in] buf       buffer with the striping
2241  * \param[in] name      name of EA
2242  * \param[in] fl        xattr flag (see OSD API description)
2243  * \param[in] th        transaction handle
2244  * \param[in] capa      not used
2245  *
2246  * \retval              0 on success
2247  * \retval              negative if failed
2248  */
2249 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
2250                                     struct dt_object *dt,
2251                                     const struct lu_buf *buf,
2252                                     const char *name, int fl,
2253                                     struct thandle *th,
2254                                     struct lustre_capa *capa)
2255 {
2256         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2257         struct lod_object       *l = lod_dt_obj(dt);
2258         struct lov_user_md_v1   *lum;
2259         struct lov_user_md_v3   *v3 = NULL;
2260         int                      rc;
2261         ENTRY;
2262
2263         /* If it is striped dir, we should clear the stripe cache for
2264          * slave stripe as well, but there are no effective way to
2265          * notify the LOD on the slave MDT, so we do not cache stripe
2266          * information for slave stripe for now. XXX*/
2267         lod_lov_stripe_cache_clear(l);
2268         LASSERT(buf != NULL && buf->lb_buf != NULL);
2269         lum = buf->lb_buf;
2270
2271         rc = lod_verify_striping(d, buf, false);
2272         if (rc)
2273                 RETURN(rc);
2274
2275         if (lum->lmm_magic == LOV_USER_MAGIC_V3)
2276                 v3 = buf->lb_buf;
2277
2278         /* if { size, offset, count } = { 0, -1, 0 } and no pool
2279          * (i.e. all default values specified) then delete default
2280          * striping from dir. */
2281         CDEBUG(D_OTHER,
2282                 "set default striping: sz %u # %u offset %d %s %s\n",
2283                 (unsigned)lum->lmm_stripe_size,
2284                 (unsigned)lum->lmm_stripe_count,
2285                 (int)lum->lmm_stripe_offset,
2286                 v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
2287
2288         if (LOVEA_DELETE_VALUES((lum->lmm_stripe_size),
2289                                 (lum->lmm_stripe_count),
2290                                 (lum->lmm_stripe_offset)) &&
2291                         lum->lmm_magic == LOV_USER_MAGIC_V1) {
2292                 rc = lod_xattr_del_internal(env, dt, name, th, capa);
2293                 if (rc == -ENODATA)
2294                         rc = 0;
2295         } else {
2296                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th, capa);
2297         }
2298
2299         RETURN(rc);
2300 }
2301
2302 /**
2303  * Set default striping on a directory object.
2304  *
2305  * Sets specified striping on a directory object unless it matches the default
2306  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
2307  * EA. This striping will be used when a new directory is being created in the
2308  * directory.
2309  *
2310  * \param[in] env       execution environment
2311  * \param[in] dt        the striped object
2312  * \param[in] buf       buffer with the striping
2313  * \param[in] name      name of EA
2314  * \param[in] fl        xattr flag (see OSD API description)
2315  * \param[in] th        transaction handle
2316  * \param[in] capa      not used
2317  *
2318  * \retval              0 on success
2319  * \retval              negative if failed
2320  */
2321 static int lod_xattr_set_default_lmv_on_dir(const struct lu_env *env,
2322                                             struct dt_object *dt,
2323                                             const struct lu_buf *buf,
2324                                             const char *name, int fl,
2325                                             struct thandle *th,
2326                                             struct lustre_capa *capa)
2327 {
2328         struct lod_object       *l = lod_dt_obj(dt);
2329         struct lmv_user_md_v1   *lum;
2330         int                      rc;
2331         ENTRY;
2332
2333         LASSERT(buf != NULL && buf->lb_buf != NULL);
2334         lum = buf->lb_buf;
2335
2336         CDEBUG(D_OTHER, "set default stripe_count # %u stripe_offset %d\n",
2337               le32_to_cpu(lum->lum_stripe_count),
2338               (int)le32_to_cpu(lum->lum_stripe_offset));
2339
2340         if (LMVEA_DELETE_VALUES((le32_to_cpu(lum->lum_stripe_count)),
2341                                  le32_to_cpu(lum->lum_stripe_offset)) &&
2342                                 le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC) {
2343                 rc = lod_xattr_del_internal(env, dt, name, th, capa);
2344                 if (rc == -ENODATA)
2345                         rc = 0;
2346         } else {
2347                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th, capa);
2348                 if (rc != 0)
2349                         RETURN(rc);
2350         }
2351
2352         /* Update default stripe cache */
2353         if (l->ldo_dir_stripe == NULL) {
2354                 OBD_ALLOC_PTR(l->ldo_dir_stripe);
2355                 if (l->ldo_dir_stripe == NULL)
2356                         RETURN(-ENOMEM);
2357         }
2358
2359         l->ldo_dir_striping_cached = 0;
2360         l->ldo_dir_def_striping_set = 1;
2361         l->ldo_dir_def_stripenr = le32_to_cpu(lum->lum_stripe_count);
2362
2363         RETURN(rc);
2364 }
2365
2366 /**
2367  * Turn directory into a striped directory.
2368  *
2369  * During replay the client sends the striping created before MDT
2370  * failure, then the layer above LOD sends this defined striping
2371  * using ->do_xattr_set(), so LOD uses this method to replay creation
2372  * of the stripes. Notice the original information for the striping
2373  * (#stripes, FIDs, etc) was transfered in declare path.
2374  *
2375  * \param[in] env       execution environment
2376  * \param[in] dt        the striped object
2377  * \param[in] buf       not used currently
2378  * \param[in] name      not used currently
2379  * \param[in] fl        xattr flag (see OSD API description)
2380  * \param[in] th        transaction handle
2381  * \param[in] capa      not used
2382  *
2383  * \retval              0 on success
2384  * \retval              negative if failed
2385  */
2386 static int lod_xattr_set_lmv(const struct lu_env *env, struct dt_object *dt,
2387                              const struct lu_buf *buf, const char *name,
2388                              int fl, struct thandle *th,
2389                              struct lustre_capa *capa)
2390 {
2391         struct lod_object       *lo = lod_dt_obj(dt);
2392         struct lod_thread_info  *info = lod_env_info(env);
2393         struct lu_attr          *attr = &info->lti_attr;
2394         struct dt_object_format *dof = &info->lti_format;
2395         struct lu_buf           lmv_buf;
2396         struct lu_buf           slave_lmv_buf;
2397         struct lmv_mds_md_v1    *lmm;
2398         struct lmv_mds_md_v1    *slave_lmm = NULL;
2399         struct dt_insert_rec    *rec = &info->lti_dt_rec;
2400         int                     i;
2401         int                     rc;
2402         ENTRY;
2403
2404         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
2405                 RETURN(-ENOTDIR);
2406
2407         /* The stripes are supposed to be allocated in declare phase,
2408          * if there are no stripes being allocated, it will skip */
2409         if (lo->ldo_stripenr == 0)
2410                 RETURN(0);
2411
2412         rc = dt_attr_get(env, dt_object_child(dt), attr, BYPASS_CAPA);
2413         if (rc != 0)
2414                 RETURN(rc);
2415
2416         attr->la_valid = LA_TYPE | LA_MODE;
2417         dof->dof_type = DFT_DIR;
2418
2419         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
2420         if (rc != 0)
2421                 RETURN(rc);
2422         lmm = lmv_buf.lb_buf;
2423
2424         OBD_ALLOC_PTR(slave_lmm);
2425         if (slave_lmm == NULL)
2426                 RETURN(-ENOMEM);
2427
2428         lod_prep_slave_lmv_md(slave_lmm, lmm);
2429         slave_lmv_buf.lb_buf = slave_lmm;
2430         slave_lmv_buf.lb_len = sizeof(*slave_lmm);
2431
2432         rec->rec_type = S_IFDIR;
2433         for (i = 0; i < lo->ldo_stripenr; i++) {
2434                 struct dt_object        *dto;
2435                 char                    *stripe_name    = info->lti_key;
2436                 struct lu_name          *sname;
2437                 struct linkea_data       ldata          = { 0 };
2438                 struct lu_buf            linkea_buf;
2439
2440                 dto = lo->ldo_stripe[i];
2441                 dt_write_lock(env, dto, MOR_TGT_CHILD);
2442                 rc = dt_create(env, dto, attr, NULL, dof, th);
2443                 dt_write_unlock(env, dto);
2444                 if (rc != 0)
2445                         RETURN(rc);
2446
2447                 rec->rec_fid = lu_object_fid(&dto->do_lu);
2448                 rc = dt_insert(env, dto, (const struct dt_rec *)rec,
2449                                (const struct dt_key *)dot, th, capa, 0);
2450                 if (rc != 0)
2451                         RETURN(rc);
2452
2453                 rec->rec_fid = lu_object_fid(&dt->do_lu);
2454                 rc = dt_insert(env, dto, (struct dt_rec *)rec,
2455                                (const struct dt_key *)dotdot, th, capa, 0);
2456                 if (rc != 0)
2457                         RETURN(rc);
2458
2459                 if (lo->ldo_striping_cached &&
2460                     !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
2461                                          lo->ldo_def_stripenr,
2462                                          lo->ldo_def_stripe_offset)) {
2463                         struct lov_user_md_v3   *v3;
2464
2465                         /* sigh, lti_ea_store has been used for lmv_buf,
2466                          * so we have to allocate buffer for default
2467                          * stripe EA */
2468                         OBD_ALLOC_PTR(v3);
2469                         if (v3 == NULL)
2470                                 GOTO(out, rc);
2471
2472                         memset(v3, 0, sizeof(*v3));
2473                         v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
2474                         v3->lmm_stripe_count =
2475                                 cpu_to_le16(lo->ldo_def_stripenr);
2476                         v3->lmm_stripe_offset =
2477                                 cpu_to_le16(lo->ldo_def_stripe_offset);
2478                         v3->lmm_stripe_size =
2479                                 cpu_to_le32(lo->ldo_def_stripe_size);
2480                         if (lo->ldo_pool != NULL)
2481                                 strlcpy(v3->lmm_pool_name, lo->ldo_pool,
2482                                         sizeof(v3->lmm_pool_name));
2483
2484                         info->lti_buf.lb_buf = v3;
2485                         info->lti_buf.lb_len = sizeof(*v3);
2486                         rc = dt_xattr_set(env, dto, &info->lti_buf,
2487                                           XATTR_NAME_LOV, 0, th, capa);
2488                         OBD_FREE_PTR(v3);
2489                         if (rc != 0)
2490                                 GOTO(out, rc);
2491                 }
2492
2493                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
2494                     cfs_fail_val != i) {
2495                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
2496                             cfs_fail_val == i)
2497                                 slave_lmm->lmv_master_mdt_index =
2498                                                         cpu_to_le32(i + 1);
2499                         else
2500                                 slave_lmm->lmv_master_mdt_index =
2501                                                         cpu_to_le32(i);
2502                         rc = dt_xattr_set(env, dto, &slave_lmv_buf,
2503                                           XATTR_NAME_LMV, fl, th, capa);
2504                         if (rc != 0)
2505                                 GOTO(out, rc);
2506                 }
2507
2508                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
2509                     cfs_fail_val == i)
2510                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
2511                                  PFID(lu_object_fid(&dto->do_lu)), i + 1);
2512                 else
2513                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
2514                                  PFID(lu_object_fid(&dto->do_lu)), i);
2515
2516                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
2517                 rc = linkea_data_new(&ldata, &info->lti_linkea_buf);
2518                 if (rc != 0)
2519                         GOTO(out, rc);
2520
2521                 rc = linkea_add_buf(&ldata, sname, lu_object_fid(&dt->do_lu));
2522                 if (rc != 0)
2523                         GOTO(out, rc);
2524
2525                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
2526                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
2527                 rc = dt_xattr_set(env, dto, &linkea_buf, XATTR_NAME_LINK,
2528                                   0, th, BYPASS_CAPA);
2529                 if (rc != 0)
2530                         GOTO(out, rc);
2531
2532                 rec->rec_fid = lu_object_fid(&dto->do_lu);
2533                 rc = dt_insert(env, dt_object_child(dt),
2534                                (const struct dt_rec *)rec,
2535                                (const struct dt_key *)stripe_name, th, capa, 0);
2536                 if (rc != 0)
2537                         GOTO(out, rc);
2538
2539                 rc = dt_ref_add(env, dt_object_child(dt), th);
2540                 if (rc != 0)
2541                         GOTO(out, rc);
2542         }
2543
2544         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MASTER_LMV))
2545                 rc = dt_xattr_set(env, dt_object_child(dt), &lmv_buf,
2546                                   XATTR_NAME_LMV, fl, th, capa);
2547
2548 out:
2549         if (slave_lmm != NULL)
2550                 OBD_FREE_PTR(slave_lmm);
2551
2552         RETURN(rc);
2553 }
2554
2555 /**
2556  * Helper function to declare/execute creation of a striped directory
2557  *
2558  * Called in declare/create object path, prepare striping for a directory
2559  * and prepare defaults data striping for the objects to be created in
2560  * that directory. Notice the function calls "declaration" or "execution"
2561  * methods depending on \a declare param. This is a consequence of the
2562  * current approach while we don't have natural distributed transactions:
2563  * we basically execute non-local updates in the declare phase. So, the
2564  * arguments for the both phases are the same and this is the reason for
2565  * this function to exist.
2566  *
2567  * \param[in] env       execution environment
2568  * \param[in] dt        object
2569  * \param[in] attr      attributes the stripes will be created with
2570  * \param[in] dof       format of stripes (see OSD API description)
2571  * \param[in] th        transaction handle
2572  * \param[in] declare   where to call "declare" or "execute" methods
2573  *
2574  * \retval              0 on success
2575  * \retval              negative if failed
2576  */
2577 int lod_dir_striping_create_internal(const struct lu_env *env,
2578                                      struct dt_object *dt,
2579                                      struct lu_attr *attr,
2580                                      struct dt_object_format *dof,
2581                                      struct thandle *th,
2582                                      bool declare)
2583 {
2584         struct lod_thread_info  *info = lod_env_info(env);
2585         struct lod_object       *lo = lod_dt_obj(dt);
2586         int                     rc;
2587         ENTRY;
2588
2589         if (!LMVEA_DELETE_VALUES(lo->ldo_stripenr,
2590                                  lo->ldo_dir_stripe_offset)) {
2591                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
2592                 int stripe_count = lo->ldo_stripenr;
2593
2594                 if (info->lti_ea_store_size < sizeof(*v1)) {
2595                         rc = lod_ea_store_resize(info, sizeof(*v1));
2596                         if (rc != 0)
2597                                 RETURN(rc);
2598                         v1 = info->lti_ea_store;
2599                 }
2600
2601                 memset(v1, 0, sizeof(*v1));
2602                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
2603                 v1->lum_stripe_count = cpu_to_le32(stripe_count);
2604                 v1->lum_stripe_offset =
2605                                 cpu_to_le32(lo->ldo_dir_stripe_offset);
2606
2607                 info->lti_buf.lb_buf = v1;
2608                 info->lti_buf.lb_len = sizeof(*v1);
2609
2610                 if (declare)
2611                         rc = lod_declare_xattr_set_lmv(env, dt, attr,
2612                                                        &info->lti_buf, dof, th);
2613                 else
2614                         rc = lod_xattr_set_lmv(env, dt, &info->lti_buf,
2615                                                XATTR_NAME_LMV, 0, th,
2616                                                BYPASS_CAPA);
2617                 if (rc != 0)
2618                         RETURN(rc);
2619         }
2620
2621         /* Transfer default LMV striping from the parent */
2622         if (lo->ldo_dir_striping_cached &&
2623             !LMVEA_DELETE_VALUES(lo->ldo_dir_def_stripenr,
2624                                  lo->ldo_dir_def_stripe_offset)) {
2625                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
2626                 int def_stripe_count = lo->ldo_dir_def_stripenr;
2627
2628                 if (info->lti_ea_store_size < sizeof(*v1)) {
2629                         rc = lod_ea_store_resize(info, sizeof(*v1));
2630                         if (rc != 0)
2631                                 RETURN(rc);
2632                         v1 = info->lti_ea_store;
2633                 }
2634
2635                 memset(v1, 0, sizeof(*v1));
2636                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
2637                 v1->lum_stripe_count = cpu_to_le32(def_stripe_count);
2638                 v1->lum_stripe_offset =
2639                                 cpu_to_le32(lo->ldo_dir_def_stripe_offset);
2640                 v1->lum_hash_type =
2641                                 cpu_to_le32(lo->ldo_dir_def_hash_type);
2642
2643                 info->lti_buf.lb_buf = v1;
2644                 info->lti_buf.lb_len = sizeof(*v1);
2645                 if (declare)
2646                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
2647                                                        XATTR_NAME_DEFAULT_LMV,
2648                                                        0, th);
2649                 else
2650                         rc = lod_xattr_set_default_lmv_on_dir(env, dt,
2651                                                   &info->lti_buf,
2652                                                   XATTR_NAME_DEFAULT_LMV, 0,
2653                                                   th, BYPASS_CAPA);
2654                 if (rc != 0)
2655                         RETURN(rc);
2656         }
2657
2658         /* Transfer default LOV striping from the parent */
2659         if (lo->ldo_striping_cached &&
2660             !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
2661                                  lo->ldo_def_stripenr,
2662                                  lo->ldo_def_stripe_offset)) {
2663                 struct lov_user_md_v3 *v3 = info->lti_ea_store;
2664
2665                 if (info->lti_ea_store_size < sizeof(*v3)) {
2666                         rc = lod_ea_store_resize(info, sizeof(*v3));
2667                         if (rc != 0)
2668                                 RETURN(rc);
2669                         v3 = info->lti_ea_store;
2670                 }
2671
2672                 memset(v3, 0, sizeof(*v3));
2673                 v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
2674                 v3->lmm_stripe_count = cpu_to_le16(lo->ldo_def_stripenr);
2675                 v3->lmm_stripe_offset = cpu_to_le16(lo->ldo_def_stripe_offset);
2676                 v3->lmm_stripe_size = cpu_to_le32(lo->ldo_def_stripe_size);
2677                 if (lo->ldo_pool != NULL)
2678                         strlcpy(v3->lmm_pool_name, lo->ldo_pool,
2679                                 sizeof(v3->lmm_pool_name));
2680
2681                 info->lti_buf.lb_buf = v3;
2682                 info->lti_buf.lb_len = sizeof(*v3);
2683
2684                 if (declare)
2685                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
2686                                                        XATTR_NAME_LOV, 0, th);
2687                 else
2688                         rc = lod_xattr_set_lov_on_dir(env, dt, &info->lti_buf,
2689                                                       XATTR_NAME_LOV, 0, th,
2690                                                       BYPASS_CAPA);
2691                 if (rc != 0)
2692                         RETURN(rc);
2693         }
2694
2695         RETURN(0);
2696 }
2697
2698 static int lod_declare_dir_striping_create(const struct lu_env *env,
2699                                            struct dt_object *dt,
2700                                            struct lu_attr *attr,
2701                                            struct dt_object_format *dof,
2702                                            struct thandle *th)
2703 {
2704         return lod_dir_striping_create_internal(env, dt, attr, dof, th, true);
2705 }
2706
2707 static int lod_dir_striping_create(const struct lu_env *env,
2708                                    struct dt_object *dt,
2709                                    struct lu_attr *attr,
2710                                    struct dt_object_format *dof,
2711                                    struct thandle *th)
2712 {
2713         return lod_dir_striping_create_internal(env, dt, attr, dof, th, false);
2714 }
2715
2716 /**
2717  * Implementation of dt_object_operations::do_xattr_set.
2718  *
2719  * Sets specified extended attribute on the object. Three types of EAs are
2720  * special:
2721  *   LOV EA - stores striping for a regular file or default striping (when set
2722  *            on a directory)
2723  *   LMV EA - stores a marker for the striped directories
2724  *   DMV EA - stores default directory striping
2725  *
2726  * When striping is applied to a non-striped existing object (this is called
2727  * late striping), then LOD notices the caller wants to turn the object into a
2728  * striped one. The stripe objects are created and appropriate EA is set:
2729  * LOV EA storing all the stripes directly or LMV EA storing just a small header
2730  * with striping configuration.
2731  *
2732  * \see dt_object_operations::do_xattr_set() in the API description for details.
2733  */
2734 static int lod_xattr_set(const struct lu_env *env,
2735                          struct dt_object *dt, const struct lu_buf *buf,
2736                          const char *name, int fl, struct thandle *th,
2737                          struct lustre_capa *capa)
2738 {
2739         struct dt_object        *next = dt_object_child(dt);
2740         int                      rc;
2741         ENTRY;
2742
2743         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
2744             strcmp(name, XATTR_NAME_LMV) == 0) {
2745                 struct lmv_mds_md_v1 *lmm = buf->lb_buf;
2746
2747                 if (lmm != NULL && le32_to_cpu(lmm->lmv_hash_type) &
2748                                                 LMV_HASH_FLAG_MIGRATION)
2749                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
2750                 else
2751                         rc = lod_dir_striping_create(env, dt, NULL, NULL, th);
2752
2753                 RETURN(rc);
2754         }
2755
2756         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
2757             strcmp(name, XATTR_NAME_LOV) == 0) {
2758                 /* default LOVEA */
2759                 rc = lod_xattr_set_lov_on_dir(env, dt, buf, name, fl, th, capa);
2760                 RETURN(rc);
2761         } else if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
2762                    strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
2763                 /* default LMVEA */
2764                 rc = lod_xattr_set_default_lmv_on_dir(env, dt, buf, name, fl,
2765                                                       th, capa);
2766                 RETURN(rc);
2767         } else if (S_ISREG(dt->do_lu.lo_header->loh_attr) &&
2768                    !strcmp(name, XATTR_NAME_LOV)) {
2769                 /* in case of lov EA swap, just set it
2770                  * if not, it is a replay so check striping match what we
2771                  * already have during req replay, declare_xattr_set()
2772                  * defines striping, then create() does the work
2773                 */
2774                 if (fl & LU_XATTR_REPLACE) {
2775                         /* free stripes, then update disk */
2776                         lod_object_free_striping(env, lod_dt_obj(dt));
2777                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
2778                 } else {
2779                         rc = lod_striping_create(env, dt, NULL, NULL, th);
2780                 }
2781                 RETURN(rc);
2782         }
2783
2784         /* then all other xattr */
2785         rc = lod_xattr_set_internal(env, dt, buf, name, fl, th, capa);
2786
2787         RETURN(rc);
2788 }
2789
2790 /**
2791  * Implementation of dt_object_operations::do_declare_xattr_del.
2792  *
2793  * \see dt_object_operations::do_declare_xattr_del() in the API description
2794  * for details.
2795  */
2796 static int lod_declare_xattr_del(const struct lu_env *env,
2797                                  struct dt_object *dt, const char *name,
2798                                  struct thandle *th)
2799 {
2800         return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
2801 }
2802
2803 /**
2804  * Implementation of dt_object_operations::do_xattr_del.
2805  *
2806  * If EA storing a regular striping is being deleted, then release
2807  * all the references to the stripe objects in core.
2808  *
2809  * \see dt_object_operations::do_xattr_del() in the API description for details.
2810  */
2811 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
2812                          const char *name, struct thandle *th,
2813                          struct lustre_capa *capa)
2814 {
2815         if (!strcmp(name, XATTR_NAME_LOV))
2816                 lod_object_free_striping(env, lod_dt_obj(dt));
2817         return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
2818 }
2819
2820 /**
2821  * Implementation of dt_object_operations::do_xattr_list.
2822  *
2823  * \see dt_object_operations::do_xattr_list() in the API description
2824  * for details.
2825  */
2826 static int lod_xattr_list(const struct lu_env *env,
2827                           struct dt_object *dt, struct lu_buf *buf,
2828                           struct lustre_capa *capa)
2829 {
2830         return dt_xattr_list(env, dt_object_child(dt), buf, capa);
2831 }
2832
2833 /**
2834  * Initialize a pool the object belongs to.
2835  *
2836  * When a striped object is being created, striping configuration
2837  * may demand the stripes are allocated on a limited set of the
2838  * targets. These limited sets are known as "pools". So we copy
2839  * a pool name into the object and later actual creation methods
2840  * (like lod_object_create()) will use this information to allocate
2841  * the stripes properly.
2842  *
2843  * \param[in] o         object
2844  * \param[in] pool      pool name
2845  */
2846 int lod_object_set_pool(struct lod_object *o, char *pool)
2847 {
2848         int len;
2849
2850         if (o->ldo_pool) {
2851                 len = strlen(o->ldo_pool);
2852                 OBD_FREE(o->ldo_pool, len + 1);
2853                 o->ldo_pool = NULL;
2854         }
2855         if (pool) {
2856                 len = strlen(pool);
2857                 OBD_ALLOC(o->ldo_pool, len + 1);
2858                 if (o->ldo_pool == NULL)
2859                         return -ENOMEM;
2860                 strcpy(o->ldo_pool, pool);
2861         }
2862         return 0;
2863 }
2864
2865 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
2866 {
2867         return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
2868 }
2869
2870
2871 /**
2872  * Cache default regular striping in the object.
2873  *
2874  * To improve performance of striped regular object creation we cache
2875  * default LOV striping (if it exists) in the parent directory object.
2876  *
2877  * \param[in] env               execution environment
2878  * \param[in] lp                object
2879  *
2880  * \retval              0 on success
2881  * \retval              negative if failed
2882  */
2883 static int lod_cache_parent_lov_striping(const struct lu_env *env,
2884                                          struct lod_object *lp)
2885 {
2886         struct lod_thread_info  *info = lod_env_info(env);
2887         struct lov_user_md_v1   *v1 = NULL;
2888         struct lov_user_md_v3   *v3 = NULL;
2889         int                      rc;
2890         ENTRY;
2891
2892         /* called from MDD without parent being write locked,
2893          * lock it here */
2894         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
2895         rc = lod_get_lov_ea(env, lp);
2896         if (rc < 0)
2897                 GOTO(unlock, rc);
2898
2899         if (rc < (typeof(rc))sizeof(struct lov_user_md)) {
2900                 /* don't lookup for non-existing or invalid striping */
2901                 lp->ldo_def_striping_set = 0;
2902                 lp->ldo_striping_cached = 1;
2903                 lp->ldo_def_stripe_size = 0;
2904                 lp->ldo_def_stripenr = 0;
2905                 lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1);
2906                 GOTO(unlock, rc = 0);
2907         }
2908
2909         rc = 0;
2910         v1 = info->lti_ea_store;
2911         if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1)) {
2912                 lustre_swab_lov_user_md_v1(v1);
2913         } else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3)) {
2914                 v3 = (struct lov_user_md_v3 *)v1;
2915                 lustre_swab_lov_user_md_v3(v3);
2916         }
2917
2918         if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1)
2919                 GOTO(unlock, rc = 0);
2920
2921         if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
2922                 GOTO(unlock, rc = 0);
2923
2924         CDEBUG(D_INFO, DFID" stripe_count=%d stripe_size=%d stripe_offset=%d\n",
2925                PFID(lu_object_fid(&lp->ldo_obj.do_lu)),
2926                (int)v1->lmm_stripe_count,
2927                (int)v1->lmm_stripe_size, (int)v1->lmm_stripe_offset);
2928
2929         lp->ldo_def_stripenr = v1->lmm_stripe_count;
2930         lp->ldo_def_stripe_size = v1->lmm_stripe_size;
2931         lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
2932         lp->ldo_striping_cached = 1;
2933         lp->ldo_def_striping_set = 1;
2934         if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
2935                 /* XXX: sanity check here */
2936                 v3 = (struct lov_user_md_v3 *) v1;
2937                 if (v3->lmm_pool_name[0])
2938                         lod_object_set_pool(lp, v3->lmm_pool_name);
2939         }
2940         EXIT;
2941 unlock:
2942         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
2943         return rc;
2944 }
2945
2946
2947 /**
2948  * Cache default directory striping in the object.
2949  *
2950  * To improve performance of striped directory creation we cache default
2951  * directory striping (if it exists) in the parent directory object.
2952  *
2953  * \param[in] env               execution environment
2954  * \param[in] lp                object
2955  *
2956  * \retval              0 on success
2957  * \retval              negative if failed
2958  */
2959 static int lod_cache_parent_lmv_striping(const struct lu_env *env,
2960                                          struct lod_object *lp)
2961 {
2962         struct lod_thread_info  *info = lod_env_info(env);
2963         struct lmv_user_md_v1   *v1 = NULL;
2964         int                      rc;
2965         ENTRY;
2966
2967         /* called from MDD without parent being write locked,
2968          * lock it here */
2969         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
2970         rc = lod_get_default_lmv_ea(env, lp);
2971         if (rc < 0)
2972                 GOTO(unlock, rc);
2973
2974         if (rc < (typeof(rc))sizeof(struct lmv_user_md)) {
2975                 /* don't lookup for non-existing or invalid striping */
2976                 lp->ldo_dir_def_striping_set = 0;
2977                 lp->ldo_dir_striping_cached = 1;
2978                 lp->ldo_dir_def_stripenr = 0;
2979                 lp->ldo_dir_def_stripe_offset =
2980                                         (typeof(v1->lum_stripe_offset))(-1);
2981                 lp->ldo_dir_def_hash_type = LMV_HASH_TYPE_FNV_1A_64;
2982                 GOTO(unlock, rc = 0);
2983         }
2984
2985         rc = 0;
2986         v1 = info->lti_ea_store;
2987
2988         lp->ldo_dir_def_stripenr = le32_to_cpu(v1->lum_stripe_count);
2989         lp->ldo_dir_def_stripe_offset = le32_to_cpu(v1->lum_stripe_offset);
2990         lp->ldo_dir_def_hash_type = le32_to_cpu(v1->lum_hash_type);
2991         lp->ldo_dir_def_striping_set = 1;
2992         lp->ldo_dir_striping_cached = 1;
2993
2994         EXIT;
2995 unlock:
2996         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
2997         return rc;
2998 }
2999
3000 /**
3001  * Cache default striping in the object.
3002  *
3003  * To improve performance of striped object creation we cache default striping
3004  * (if it exists) in the parent directory object. We always cache default
3005  * striping for the regular files (stored in LOV EA) and we cache default
3006  * striping for the directories if requested by \a child_mode (when a new
3007  * directory is being created).
3008  *
3009  * \param[in] env               execution environment
3010  * \param[in] lp                object
3011  * \param[in] child_mode        new object's mode
3012  *
3013  * \retval              0 on success
3014  * \retval              negative if failed
3015  */
3016 static int lod_cache_parent_striping(const struct lu_env *env,
3017                                      struct lod_object *lp,
3018                                      umode_t child_mode)
3019 {
3020         int rc = 0;
3021         ENTRY;
3022
3023         rc = lod_load_striping(env, lp);
3024         if (rc != 0)
3025                 RETURN(rc);
3026
3027         if (!lp->ldo_striping_cached) {
3028                 /* we haven't tried to get default striping for
3029                  * the directory yet, let's cache it in the object */
3030                 rc = lod_cache_parent_lov_striping(env, lp);
3031                 if (rc != 0)
3032                         RETURN(rc);
3033         }
3034
3035         if (S_ISDIR(child_mode) && !lp->ldo_dir_striping_cached)
3036                 rc = lod_cache_parent_lmv_striping(env, lp);
3037
3038         RETURN(rc);
3039 }
3040
3041 /**
3042  * Implementation of dt_object_operations::do_ah_init.
3043  *
3044  * This method is used to make a decision on the striping configuration for the
3045  * object being created. It can be taken from the \a parent object if it exists,
3046  * or filesystem's default. The resulting configuration (number of stripes,
3047  * stripe size/offset, pool name, etc) is stored in the object itself and will
3048  * be used by the methods like ->doo_declare_create().
3049  *
3050  * \see dt_object_operations::do_ah_init() in the API description for details.
3051  */
3052 static void lod_ah_init(const struct lu_env *env,
3053                         struct dt_allocation_hint *ah,
3054                         struct dt_object *parent,
3055                         struct dt_object *child,
3056                         umode_t child_mode)
3057 {
3058         struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
3059         struct dt_object  *nextp = NULL;
3060         struct dt_object  *nextc;
3061         struct lod_object *lp = NULL;
3062         struct lod_object *lc;
3063         struct lov_desc   *desc;
3064         int               rc;
3065         ENTRY;
3066
3067         LASSERT(child);
3068
3069         if (likely(parent)) {
3070                 nextp = dt_object_child(parent);
3071                 lp = lod_dt_obj(parent);
3072                 rc = lod_load_striping(env, lp);
3073                 if (rc != 0)
3074                         return;
3075         }
3076
3077         nextc = dt_object_child(child);
3078         lc = lod_dt_obj(child);
3079
3080         LASSERT(lc->ldo_stripenr == 0);
3081         LASSERT(lc->ldo_stripe == NULL);
3082
3083         /*
3084          * local object may want some hints
3085          * in case of late striping creation, ->ah_init()
3086          * can be called with local object existing
3087          */
3088         if (!dt_object_exists(nextc) || dt_object_remote(nextc))
3089                 nextc->do_ops->do_ah_init(env, ah, dt_object_remote(nextp) ?
3090                                           NULL : nextp, nextc, child_mode);
3091
3092         if (S_ISDIR(child_mode)) {
3093                 if (lc->ldo_dir_stripe == NULL) {
3094                         OBD_ALLOC_PTR(lc->ldo_dir_stripe);
3095                         if (lc->ldo_dir_stripe == NULL)
3096                                 return;
3097                 }
3098
3099                 if (lp->ldo_dir_stripe == NULL) {
3100                         OBD_ALLOC_PTR(lp->ldo_dir_stripe);
3101                         if (lp->ldo_dir_stripe == NULL)
3102                                 return;
3103                 }
3104
3105                 rc = lod_cache_parent_striping(env, lp, child_mode);
3106                 if (rc != 0)
3107                         return;
3108
3109                 /* transfer defaults to new directory */
3110                 if (lp->ldo_striping_cached) {
3111                         if (lp->ldo_pool)
3112                                 lod_object_set_pool(lc, lp->ldo_pool);
3113                         lc->ldo_def_stripenr = lp->ldo_def_stripenr;
3114                         lc->ldo_def_stripe_size = lp->ldo_def_stripe_size;
3115                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
3116                         lc->ldo_striping_cached = 1;
3117                         lc->ldo_def_striping_set = 1;
3118                         CDEBUG(D_OTHER, "inherite EA sz:%d off:%d nr:%d\n",
3119                                (int)lc->ldo_def_stripe_size,
3120                                (int)lc->ldo_def_stripe_offset,
3121                                (int)lc->ldo_def_stripenr);
3122                 }
3123
3124                 /* transfer dir defaults to new directory */
3125                 if (lp->ldo_dir_striping_cached) {
3126                         lc->ldo_dir_def_stripenr = lp->ldo_dir_def_stripenr;
3127                         lc->ldo_dir_def_stripe_offset =
3128                                                   lp->ldo_dir_def_stripe_offset;
3129                         lc->ldo_dir_def_hash_type =
3130                                                   lp->ldo_dir_def_hash_type;
3131                         lc->ldo_dir_striping_cached = 1;
3132                         lc->ldo_dir_def_striping_set = 1;
3133                         CDEBUG(D_INFO, "inherit default EA nr:%d off:%d t%u\n",
3134                                (int)lc->ldo_dir_def_stripenr,
3135                                (int)lc->ldo_dir_def_stripe_offset,
3136                                lc->ldo_dir_def_hash_type);
3137                 }
3138
3139                 /* It should always honour the specified stripes */
3140                 if (ah->dah_eadata != NULL && ah->dah_eadata_len != 0) {
3141                         const struct lmv_user_md_v1 *lum1 = ah->dah_eadata;
3142
3143                         rc = lod_verify_md_striping(d, lum1);
3144                         if (rc == 0 &&
3145                                 le32_to_cpu(lum1->lum_stripe_count) > 1) {
3146                                 /* Directory will be striped only if
3147                                  * stripe_count > 1 */
3148                                 lc->ldo_stripenr =
3149                                         le32_to_cpu(lum1->lum_stripe_count);
3150                                 lc->ldo_dir_stripe_offset =
3151                                         le32_to_cpu(lum1->lum_stripe_offset);
3152                                 lc->ldo_dir_hash_type =
3153                                         le32_to_cpu(lum1->lum_hash_type);
3154                                 CDEBUG(D_INFO, "set stripe EA nr:%hu off:%d\n",
3155                                        lc->ldo_stripenr,
3156                                        (int)lc->ldo_dir_stripe_offset);
3157                         }
3158                 /* then check whether there is default stripes from parent */
3159                 } else if (lp->ldo_dir_def_striping_set) {
3160                         /* If there are default dir stripe from parent */
3161                         lc->ldo_stripenr = lp->ldo_dir_def_stripenr;
3162                         lc->ldo_dir_stripe_offset =
3163                                         lp->ldo_dir_def_stripe_offset;
3164                         lc->ldo_dir_hash_type =
3165                                         lp->ldo_dir_def_hash_type;
3166                         CDEBUG(D_INFO, "inherit EA nr:%hu off:%d\n",
3167                                lc->ldo_stripenr,
3168                                (int)lc->ldo_dir_stripe_offset);
3169                 } else {
3170                         /* set default stripe for this directory */
3171                         lc->ldo_stripenr = 0;
3172                         lc->ldo_dir_stripe_offset = -1;
3173                 }
3174
3175                 CDEBUG(D_INFO, "final striping count:%hu, offset:%d\n",
3176                        lc->ldo_stripenr, (int)lc->ldo_dir_stripe_offset);
3177
3178                 goto out;
3179         }
3180
3181         /*
3182          * if object is going to be striped over OSTs, transfer default
3183          * striping information to the child, so that we can use it
3184          * during declaration and creation