Whamcloud - gitweb
LU-1187 tests: Add DNE test cases in sanity.
[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) 2011, 2012, Intel, Inc.
27  */
28 /*
29  * lustre/lod/lod_object.c
30  *
31  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
32  */
33
34 #ifndef EXPORT_SYMTAB
35 # define EXPORT_SYMTAB
36 #endif
37 #define DEBUG_SUBSYSTEM S_MDS
38
39 #include <obd.h>
40 #include <obd_class.h>
41 #include <lustre_ver.h>
42 #include <obd_support.h>
43 #include <lprocfs_status.h>
44
45 #include <lustre_fid.h>
46 #include <lustre_param.h>
47 #include <lustre_fid.h>
48 #include <obd_lov.h>
49
50 #include "lod_internal.h"
51
52 extern cfs_mem_cache_t *lod_object_kmem;
53 static const struct dt_body_operations lod_body_lnk_ops;
54
55 static int lod_index_lookup(const struct lu_env *env, struct dt_object *dt,
56                             struct dt_rec *rec, const struct dt_key *key,
57                             struct lustre_capa *capa)
58 {
59         struct dt_object *next = dt_object_child(dt);
60         return next->do_index_ops->dio_lookup(env, next, rec, key, capa);
61 }
62
63 static int lod_declare_index_insert(const struct lu_env *env,
64                                     struct dt_object *dt,
65                                     const struct dt_rec *rec,
66                                     const struct dt_key *key,
67                                     struct thandle *handle)
68 {
69         return dt_declare_insert(env, dt_object_child(dt), rec, key, handle);
70 }
71
72 static int lod_index_insert(const struct lu_env *env,
73                             struct dt_object *dt,
74                             const struct dt_rec *rec,
75                             const struct dt_key *key,
76                             struct thandle *th,
77                             struct lustre_capa *capa,
78                             int ign)
79 {
80         return dt_insert(env, dt_object_child(dt), rec, key, th, capa, ign);
81 }
82
83 static int lod_declare_index_delete(const struct lu_env *env,
84                                     struct dt_object *dt,
85                                     const struct dt_key *key,
86                                     struct thandle *th)
87 {
88         return dt_declare_delete(env, dt_object_child(dt), key, th);
89 }
90
91 static int lod_index_delete(const struct lu_env *env,
92                             struct dt_object *dt,
93                             const struct dt_key *key,
94                             struct thandle *th,
95                             struct lustre_capa *capa)
96 {
97         return dt_delete(env, dt_object_child(dt), key, th, capa);
98 }
99
100 static struct dt_it *lod_it_init(const struct lu_env *env,
101                                  struct dt_object *dt, __u32 attr,
102                                  struct lustre_capa *capa)
103 {
104         struct dt_object        *next = dt_object_child(dt);
105         struct lod_it           *it = &lod_env_info(env)->lti_it;
106         struct dt_it            *it_next;
107
108
109         it_next = next->do_index_ops->dio_it.init(env, next, attr, capa);
110         if (IS_ERR(it_next))
111                 return it_next;
112
113         /* currently we do not use more than one iterator per thread
114          * so we store it in thread info. if at some point we need
115          * more active iterators in a single thread, we can allocate
116          * additional ones */
117         LASSERT(it->lit_obj == NULL);
118
119         it->lit_it = it_next;
120         it->lit_obj = next;
121
122         return (struct dt_it *)it;
123 }
124
125 #define LOD_CHECK_IT(env, it)                                   \
126 {                                                               \
127         /* IT is supposed to be in thread info always */        \
128         LASSERT((it) == &lod_env_info(env)->lti_it);            \
129         LASSERT((it)->lit_obj != NULL);                         \
130         LASSERT((it)->lit_it != NULL);                          \
131 } while(0)
132
133 void lod_it_fini(const struct lu_env *env, struct dt_it *di)
134 {
135         struct lod_it *it = (struct lod_it *)di;
136
137         LOD_CHECK_IT(env, it);
138         it->lit_obj->do_index_ops->dio_it.fini(env, it->lit_it);
139
140         /* the iterator not in use any more */
141         it->lit_obj = NULL;
142         it->lit_it = NULL;
143 }
144
145 int lod_it_get(const struct lu_env *env, struct dt_it *di,
146                const struct dt_key *key)
147 {
148         const struct lod_it *it = (const struct lod_it *)di;
149
150         LOD_CHECK_IT(env, it);
151         return it->lit_obj->do_index_ops->dio_it.get(env, it->lit_it, key);
152 }
153
154 void lod_it_put(const struct lu_env *env, struct dt_it *di)
155 {
156         struct lod_it *it = (struct lod_it *)di;
157
158         LOD_CHECK_IT(env, it);
159         return it->lit_obj->do_index_ops->dio_it.put(env, it->lit_it);
160 }
161
162 int lod_it_next(const struct lu_env *env, struct dt_it *di)
163 {
164         struct lod_it *it = (struct lod_it *)di;
165
166         LOD_CHECK_IT(env, it);
167         return it->lit_obj->do_index_ops->dio_it.next(env, it->lit_it);
168 }
169
170 struct dt_key *lod_it_key(const struct lu_env *env, const struct dt_it *di)
171 {
172         const struct lod_it *it = (const struct lod_it *)di;
173
174         LOD_CHECK_IT(env, it);
175         return it->lit_obj->do_index_ops->dio_it.key(env, it->lit_it);
176 }
177
178 int lod_it_key_size(const struct lu_env *env, const struct dt_it *di)
179 {
180         struct lod_it *it = (struct lod_it *)di;
181
182         LOD_CHECK_IT(env, it);
183         return it->lit_obj->do_index_ops->dio_it.key_size(env, it->lit_it);
184 }
185
186 int lod_it_rec(const struct lu_env *env, const struct dt_it *di,
187                struct dt_rec *rec, __u32 attr)
188 {
189         const struct lod_it *it = (const struct lod_it *)di;
190
191         LOD_CHECK_IT(env, it);
192         return it->lit_obj->do_index_ops->dio_it.rec(env, it->lit_it, rec, attr);
193 }
194
195 __u64 lod_it_store(const struct lu_env *env, const struct dt_it *di)
196 {
197         const struct lod_it *it = (const struct lod_it *)di;
198
199         LOD_CHECK_IT(env, it);
200         return it->lit_obj->do_index_ops->dio_it.store(env, it->lit_it);
201 }
202
203 int lod_it_load(const struct lu_env *env, const struct dt_it *di, __u64 hash)
204 {
205         const struct lod_it *it = (const struct lod_it *)di;
206
207         LOD_CHECK_IT(env, it);
208         return it->lit_obj->do_index_ops->dio_it.load(env, it->lit_it, hash);
209 }
210
211 int lod_it_key_rec(const struct lu_env *env, const struct dt_it *di,
212                    void* key_rec)
213 {
214         const struct lod_it *it = (const struct lod_it *)di;
215
216         LOD_CHECK_IT(env, it);
217         return it->lit_obj->do_index_ops->dio_it.key_rec(env, it->lit_it, key_rec);
218 }
219
220 static struct dt_index_operations lod_index_ops = {
221         .dio_lookup             = lod_index_lookup,
222         .dio_declare_insert     = lod_declare_index_insert,
223         .dio_insert             = lod_index_insert,
224         .dio_declare_delete     = lod_declare_index_delete,
225         .dio_delete             = lod_index_delete,
226         .dio_it = {
227                 .init           = lod_it_init,
228                 .fini           = lod_it_fini,
229                 .get            = lod_it_get,
230                 .put            = lod_it_put,
231                 .next           = lod_it_next,
232                 .key            = lod_it_key,
233                 .key_size       = lod_it_key_size,
234                 .rec            = lod_it_rec,
235                 .store          = lod_it_store,
236                 .load           = lod_it_load,
237                 .key_rec        = lod_it_key_rec,
238         }
239 };
240
241 static void lod_object_read_lock(const struct lu_env *env,
242                                  struct dt_object *dt, unsigned role)
243 {
244         dt_read_lock(env, dt_object_child(dt), role);
245 }
246
247 static void lod_object_write_lock(const struct lu_env *env,
248                                   struct dt_object *dt, unsigned role)
249 {
250         dt_write_lock(env, dt_object_child(dt), role);
251 }
252
253 static void lod_object_read_unlock(const struct lu_env *env,
254                                    struct dt_object *dt)
255 {
256         dt_read_unlock(env, dt_object_child(dt));
257 }
258
259 static void lod_object_write_unlock(const struct lu_env *env,
260                                     struct dt_object *dt)
261 {
262         dt_write_unlock(env, dt_object_child(dt));
263 }
264
265 static int lod_object_write_locked(const struct lu_env *env,
266                                    struct dt_object *dt)
267 {
268         return dt_write_locked(env, dt_object_child(dt));
269 }
270
271 static int lod_attr_get(const struct lu_env *env,
272                         struct dt_object *dt,
273                         struct lu_attr *attr,
274                         struct lustre_capa *capa)
275 {
276         return dt_attr_get(env, dt_object_child(dt), attr, capa);
277 }
278
279 static int lod_declare_attr_set(const struct lu_env *env,
280                                 struct dt_object *dt,
281                                 const struct lu_attr *attr,
282                                 struct thandle *handle)
283 {
284         struct dt_object  *next = dt_object_child(dt);
285         struct lod_object *lo = lod_dt_obj(dt);
286         int                rc, i;
287         ENTRY;
288
289         /*
290          * declare setattr on the local object
291          */
292         rc = dt_declare_attr_set(env, next, attr, handle);
293         if (rc)
294                 RETURN(rc);
295
296         /*
297          * load striping information, notice we don't do this when object
298          * is being initialized as we don't need this information till
299          * few specific cases like destroy, chown
300          */
301         rc = lod_load_striping(env, lo);
302         if (rc)
303                 RETURN(rc);
304
305         /*
306          * if object is striped declare changes on the stripes
307          */
308         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
309         for (i = 0; i < lo->ldo_stripenr; i++) {
310                 LASSERT(lo->ldo_stripe[i]);
311                 rc = dt_declare_attr_set(env, lo->ldo_stripe[i], attr, handle);
312                 if (rc) {
313                         CERROR("failed declaration: %d\n", rc);
314                         break;
315                 }
316         }
317
318         RETURN(rc);
319 }
320
321 static int lod_attr_set(const struct lu_env *env,
322                         struct dt_object *dt,
323                         const struct lu_attr *attr,
324                         struct thandle *handle,
325                         struct lustre_capa *capa)
326 {
327         struct dt_object  *next = dt_object_child(dt);
328         struct lod_object *lo = lod_dt_obj(dt);
329         int                rc, i;
330         ENTRY;
331
332         /*
333          * apply changes to the local object
334          */
335         rc = dt_attr_set(env, next, attr, handle, capa);
336         if (rc)
337                 RETURN(rc);
338
339         /*
340          * if object is striped, apply changes to all the stripes
341          */
342         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
343         for (i = 0; i < lo->ldo_stripenr; i++) {
344                 LASSERT(lo->ldo_stripe[i]);
345                 rc = dt_attr_set(env, lo->ldo_stripe[i], attr, handle, capa);
346                 if (rc) {
347                         CERROR("failed declaration: %d\n", rc);
348                         break;
349                 }
350         }
351
352         RETURN(rc);
353 }
354
355 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
356                          struct lu_buf *buf, const char *name,
357                          struct lustre_capa *capa)
358 {
359         struct lod_thread_info  *info = lod_env_info(env);
360         struct lod_device       *dev = lu2lod_dev(dt->do_lu.lo_dev);
361         int                      rc, is_root;
362         ENTRY;
363
364         rc = dt_xattr_get(env, dt_object_child(dt), buf, name, capa);
365         if (rc != -ENODATA || !S_ISDIR(dt->do_lu.lo_header->loh_attr & S_IFMT))
366                 RETURN(rc);
367
368         /*
369          * lod returns default striping on the real root of the device
370          * this is like the root stores default striping for the whole
371          * filesystem. historically we've been using a different approach
372          * and store it in the config.
373          */
374         dt_root_get(env, dev->lod_child, &info->lti_fid);
375         is_root = lu_fid_eq(&info->lti_fid, lu_object_fid(&dt->do_lu));
376
377         if (is_root && strcmp(XATTR_NAME_LOV, name) == 0) {
378                 struct lov_user_md *lum = buf->lb_buf;
379                 struct lov_desc    *desc = &dev->lod_desc;
380
381                 if (buf->lb_buf == NULL) {
382                         rc = sizeof(struct lov_user_md_v1);
383                 } else if (buf->lb_len >= sizeof(struct lov_user_md_v1)) {
384                         lum->lmm_magic = LOV_USER_MAGIC_V1;
385                         lum->lmm_object_seq = FID_SEQ_LOV_DEFAULT;
386                         lum->lmm_pattern = desc->ld_pattern;
387                         lum->lmm_stripe_size = desc->ld_default_stripe_size;
388                         lum->lmm_stripe_count = desc->ld_default_stripe_count;
389                         lum->lmm_stripe_offset = desc->ld_default_stripe_offset;
390                         rc = sizeof(struct lov_user_md_v1);
391                 } else {
392                         rc = -ERANGE;
393                 }
394         }
395
396         RETURN(rc);
397 }
398
399 /*
400  * LOV xattr is a storage for striping, and LOD owns this xattr.
401  * but LOD allows others to control striping to some extent
402  * - to reset strping
403  * - to set new defined striping
404  * - to set new semi-defined striping
405  *   - number of stripes is defined
406  *   - number of stripes + osts are defined
407  *   - ??
408  */
409 static int lod_declare_xattr_set(const struct lu_env *env,
410                                  struct dt_object *dt,
411                                  const struct lu_buf *buf,
412                                  const char *name, int fl,
413                                  struct thandle *th)
414 {
415         struct dt_object *next = dt_object_child(dt);
416         struct lu_attr   *attr = &lod_env_info(env)->lti_attr;
417         __u32             mode;
418         int               rc;
419         ENTRY;
420
421         /*
422          * allow to declare predefined striping on a new (!mode) object
423          * which is supposed to be replay of regular file creation
424          * (when LOV setting is declared)
425          * LU_XATTR_REPLACE is set to indicate a layout swap
426          */
427         mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
428         if ((S_ISREG(mode) || !mode) && !strcmp(name, XATTR_NAME_LOV) &&
429              !(fl & LU_XATTR_REPLACE)) {
430                 /*
431                  * this is a request to manipulate object's striping
432                  */
433                 if (dt_object_exists(dt)) {
434                         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
435                         if (rc)
436                                 RETURN(rc);
437                 } else {
438                         memset(attr, 0, sizeof(attr));
439                         attr->la_valid = LA_TYPE | LA_MODE;
440                         attr->la_mode = S_IFREG;
441                 }
442                 rc = lod_declare_striped_object(env, dt, attr, buf, th);
443                 if (rc)
444                         RETURN(rc);
445         }
446
447         rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
448
449         RETURN(rc);
450 }
451
452 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
453                                     struct dt_object *dt,
454                                     const struct lu_buf *buf,
455                                     const char *name, int fl,
456                                     struct thandle *th,
457                                     struct lustre_capa *capa)
458 {
459         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
460         struct dt_object        *next = dt_object_child(dt);
461         struct lod_object       *l = lod_dt_obj(dt);
462         struct lov_user_md_v1   *lum;
463         struct lov_user_md_v3   *v3 = NULL;
464         int                      rc;
465         ENTRY;
466
467         LASSERT(l->ldo_stripe == NULL);
468         l->ldo_striping_cached = 0;
469         l->ldo_def_striping_set = 0;
470         lod_object_set_pool(l, NULL);
471         l->ldo_def_stripe_size = 0;
472         l->ldo_def_stripenr = 0;
473
474         LASSERT(buf);
475         LASSERT(buf->lb_buf);
476         lum = buf->lb_buf;
477
478         rc = lod_verify_striping(d, buf, 0);
479         if (rc)
480                 RETURN(rc);
481
482         if (lum->lmm_magic == LOV_USER_MAGIC_V3)
483                 v3 = buf->lb_buf;
484
485         /* if { size, offset, count } = { 0, -1, 0 } and no pool
486          * (i.e. all default values specified) then delete default
487          * striping from dir. */
488         CDEBUG(D_OTHER,
489                 "set default striping: sz %u # %u offset %d %s %s\n",
490                 (unsigned)lum->lmm_stripe_size,
491                 (unsigned)lum->lmm_stripe_count,
492                 (int)lum->lmm_stripe_offset,
493                 v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
494
495         if (LOVEA_DELETE_VALUES((lum->lmm_stripe_size),
496                                 (lum->lmm_stripe_count),
497                                 (lum->lmm_stripe_offset)) &&
498                         lum->lmm_magic == LOV_USER_MAGIC_V1) {
499                 rc = dt_xattr_del(env, next, name, th, capa);
500                 if (rc == -ENODATA)
501                         rc = 0;
502         } else {
503                 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
504         }
505
506         RETURN(rc);
507 }
508
509 static int lod_xattr_set(const struct lu_env *env,
510                          struct dt_object *dt, const struct lu_buf *buf,
511                          const char *name, int fl, struct thandle *th,
512                          struct lustre_capa *capa)
513 {
514         struct dt_object        *next = dt_object_child(dt);
515         __u32                    attr;
516         int                      rc;
517         ENTRY;
518
519         attr = dt->do_lu.lo_header->loh_attr & S_IFMT;
520         if (S_ISDIR(attr)) {
521                 if (strncmp(name, XATTR_NAME_LOV, strlen(XATTR_NAME_LOV)) == 0)
522                         rc = lod_xattr_set_lov_on_dir(env, dt, buf, name,
523                                                       fl, th, capa);
524                 else
525                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
526
527         } else if (S_ISREG(attr) && !strcmp(name, XATTR_NAME_LOV)) {
528                 /* in case of lov EA swap, just set it
529                  * if not, it is a replay so check striping match what we
530                  * already have during req replay, declare_xattr_set()
531                  * defines striping, then create() does the work
532                 */
533                 if (fl & LU_XATTR_REPLACE)
534                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
535                 else
536                         rc = lod_striping_create(env, dt, NULL, NULL, th);
537                 RETURN(rc);
538         } else {
539                 /*
540                  * behave transparantly for all other EAs
541                  */
542                 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
543         }
544
545         RETURN(rc);
546 }
547
548 static int lod_declare_xattr_del(const struct lu_env *env,
549                                  struct dt_object *dt, const char *name,
550                                  struct thandle *th)
551 {
552         return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
553 }
554
555 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
556                          const char *name, struct thandle *th,
557                          struct lustre_capa *capa)
558 {
559         return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
560 }
561
562 static int lod_xattr_list(const struct lu_env *env,
563                           struct dt_object *dt, struct lu_buf *buf,
564                           struct lustre_capa *capa)
565 {
566         return dt_xattr_list(env, dt_object_child(dt), buf, capa);
567 }
568
569 int lod_object_set_pool(struct lod_object *o, char *pool)
570 {
571         int len;
572
573         if (o->ldo_pool) {
574                 len = strlen(o->ldo_pool);
575                 OBD_FREE(o->ldo_pool, len + 1);
576                 o->ldo_pool = NULL;
577         }
578         if (pool) {
579                 len = strlen(pool);
580                 OBD_ALLOC(o->ldo_pool, len + 1);
581                 if (o->ldo_pool == NULL)
582                         return -ENOMEM;
583                 strcpy(o->ldo_pool, pool);
584         }
585         return 0;
586 }
587
588 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
589 {
590         return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
591 }
592
593 static int lod_cache_parent_striping(const struct lu_env *env,
594                                      struct lod_object *lp)
595 {
596         struct lov_user_md_v1   *v1 = NULL;
597         struct lov_user_md_v3   *v3 = NULL;
598         int                      rc;
599         ENTRY;
600
601         /* dt_ah_init() is called from MDD without parent being write locked
602          * lock it here */
603         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
604         if (lp->ldo_striping_cached)
605                 GOTO(unlock, rc = 0);
606
607         rc = lod_get_lov_ea(env, lp);
608         if (rc < 0)
609                 GOTO(unlock, rc);
610
611         if (rc < sizeof(struct lov_user_md)) {
612                 /* don't lookup for non-existing or invalid striping */
613                 lp->ldo_def_striping_set = 0;
614                 lp->ldo_striping_cached = 1;
615                 lp->ldo_def_stripe_size = 0;
616                 lp->ldo_def_stripenr = 0;
617                 lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1);
618                 GOTO(unlock, rc = 0);
619         }
620
621         v1 = (struct lov_user_md_v1 *)lod_env_info(env)->lti_ea_store;
622         if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1))
623                 lustre_swab_lov_user_md_v1(v1);
624         else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3))
625                 lustre_swab_lov_user_md_v3(v3);
626
627         if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1)
628                 GOTO(unlock, rc = 0);
629
630         if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
631                 GOTO(unlock, rc = 0);
632
633         lp->ldo_def_stripenr = v1->lmm_stripe_count;
634         lp->ldo_def_stripe_size = v1->lmm_stripe_size;
635         lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
636         lp->ldo_striping_cached = 1;
637         lp->ldo_def_striping_set = 1;
638
639         if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
640                 /* XXX: sanity check here */
641                 v3 = (struct lov_user_md_v3 *) v1;
642                 if (v3->lmm_pool_name[0])
643                         lod_object_set_pool(lp, v3->lmm_pool_name);
644         }
645
646         CDEBUG(D_OTHER, "def. striping: # %d, sz %d, off %d %s%s on "DFID"\n",
647                lp->ldo_def_stripenr, lp->ldo_def_stripe_size,
648                lp->ldo_def_stripe_offset, v3 ? "from " : "",
649                v3 ? lp->ldo_pool : "", PFID(lu_object_fid(&lp->ldo_obj.do_lu)));
650
651         EXIT;
652 unlock:
653         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
654         return rc;
655 }
656
657 /**
658  * used to transfer default striping data to the object being created
659  */
660 static void lod_ah_init(const struct lu_env *env,
661                         struct dt_allocation_hint *ah,
662                         struct dt_object *parent,
663                         struct dt_object *child,
664                         cfs_umode_t child_mode)
665 {
666         struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
667         struct dt_object  *nextp = NULL;
668         struct dt_object  *nextc;
669         struct lod_object *lp = NULL;
670         struct lod_object *lc;
671         struct lov_desc   *desc;
672         ENTRY;
673
674         LASSERT(child);
675
676         if (likely(parent)) {
677                 nextp = dt_object_child(parent);
678                 lp = lod_dt_obj(parent);
679         }
680
681         nextc = dt_object_child(child);
682         lc = lod_dt_obj(child);
683
684         LASSERT(lc->ldo_stripenr == 0);
685         LASSERT(lc->ldo_stripe == NULL);
686
687         /*
688          * local object may want some hints
689          * in case of late striping creation, ->ah_init()
690          * can be called with local object existing
691          */
692         if (!dt_object_exists(nextc))
693                 nextc->do_ops->do_ah_init(env, ah, nextp, nextc, child_mode);
694
695         if (S_ISDIR(child_mode)) {
696                 if (lp->ldo_striping_cached == 0) {
697                         /* we haven't tried to get default striping for
698                          * the directory yet, let's cache it in the object */
699                         lod_cache_parent_striping(env, lp);
700                 }
701                 /* transfer defaults to new directory */
702                 if (lp->ldo_striping_cached) {
703                         if (lp->ldo_pool)
704                                 lod_object_set_pool(lc, lp->ldo_pool);
705                         lc->ldo_def_stripenr = lp->ldo_def_stripenr;
706                         lc->ldo_def_stripe_size = lp->ldo_def_stripe_size;
707                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
708                         lc->ldo_striping_cached = 1;
709                         lc->ldo_def_striping_set = 1;
710                         CDEBUG(D_OTHER, "inherite striping defaults\n");
711                 }
712                 return;
713         }
714
715         /*
716          * if object is going to be striped over OSTs, transfer default
717          * striping information to the child, so that we can use it
718          * during declaration and creation
719          */
720         if (!lod_object_will_be_striped(S_ISREG(child_mode),
721                                         lu_object_fid(&child->do_lu)))
722                 return;
723
724         /*
725          * try from the parent
726          */
727         if (likely(parent)) {
728                 if (lp->ldo_striping_cached == 0) {
729                         /* we haven't tried to get default striping for
730                          * the directory yet, let's cache it in the object */
731                         lod_cache_parent_striping(env, lp);
732                 }
733
734                 lc->ldo_def_stripe_offset = (__u16) -1;
735
736                 if (lp->ldo_def_striping_set) {
737                         if (lp->ldo_pool)
738                                 lod_object_set_pool(lc, lp->ldo_pool);
739                         lc->ldo_stripenr = lp->ldo_def_stripenr;
740                         lc->ldo_stripe_size = lp->ldo_def_stripe_size;
741                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
742                         CDEBUG(D_OTHER, "striping from parent: #%d, sz %d %s\n",
743                                lc->ldo_stripenr, lc->ldo_stripe_size,
744                                lp->ldo_pool ? lp->ldo_pool : "");
745                 }
746         }
747
748         /*
749          * if the parent doesn't provide with specific pattern, grab fs-wide one
750          */
751         desc = &d->lod_desc;
752         if (lc->ldo_stripenr == 0)
753                 lc->ldo_stripenr = desc->ld_default_stripe_count;
754         if (lc->ldo_stripe_size == 0)
755                 lc->ldo_stripe_size = desc->ld_default_stripe_size;
756         CDEBUG(D_OTHER, "final striping: # %d stripes, sz %d from %s\n",
757                lc->ldo_stripenr, lc->ldo_stripe_size,
758                lc->ldo_pool ? lc->ldo_pool : "");
759
760         EXIT;
761 }
762
763 #define ll_do_div64(aaa,bbb)    do_div((aaa), (bbb))
764 /*
765  * this function handles a special case when truncate was done
766  * on a stripeless object and now striping is being created
767  * we can't lose that size, so we have to propagate it to newly
768  * created object
769  */
770 static int lod_declare_init_size(const struct lu_env *env,
771                                  struct dt_object *dt, struct thandle *th)
772 {
773         struct dt_object   *next = dt_object_child(dt);
774         struct lod_object  *lo = lod_dt_obj(dt);
775         struct lu_attr     *attr = &lod_env_info(env)->lti_attr;
776         uint64_t            size, offs;
777         int                 rc, stripe;
778         ENTRY;
779
780         /* XXX: we support the simplest (RAID0) striping so far */
781         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
782         LASSERT(lo->ldo_stripe_size > 0);
783
784         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
785         LASSERT(attr->la_valid & LA_SIZE);
786         if (rc)
787                 RETURN(rc);
788
789         size = attr->la_size;
790         if (size == 0)
791                 RETURN(0);
792
793         /* ll_do_div64(a, b) returns a % b, and a = a / b */
794         ll_do_div64(size, (__u64) lo->ldo_stripe_size);
795         stripe = ll_do_div64(size, (__u64) lo->ldo_stripenr);
796
797         size = size * lo->ldo_stripe_size;
798         offs = attr->la_size;
799         size += ll_do_div64(offs, lo->ldo_stripe_size);
800
801         attr->la_valid = LA_SIZE;
802         attr->la_size = size;
803
804         rc = dt_declare_attr_set(env, lo->ldo_stripe[stripe], attr, th);
805
806         RETURN(rc);
807 }
808
809
810 /**
811  * Create declaration of striped object
812  */
813 int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt,
814                                struct lu_attr *attr,
815                                const struct lu_buf *lovea, struct thandle *th)
816 {
817         struct lod_thread_info  *info = lod_env_info(env);
818         struct dt_object        *next = dt_object_child(dt);
819         struct lod_object       *lo = lod_dt_obj(dt);
820         int                      rc;
821         ENTRY;
822
823         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) {
824                 /* failed to create striping, let's reset
825                  * config so that others don't get confused */
826                 lod_object_free_striping(env, lo);
827                 GOTO(out, rc = -ENOMEM);
828         }
829
830         /* choose OST and generate appropriate objects */
831         rc = lod_qos_prep_create(env, lo, attr, lovea, th);
832         if (rc) {
833                 /* failed to create striping, let's reset
834                  * config so that others don't get confused */
835                 lod_object_free_striping(env, lo);
836                 GOTO(out, rc);
837         }
838
839         /*
840          * declare storage for striping data
841          */
842         info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
843                                 lo->ldo_pool ?  LOV_MAGIC_V3 : LOV_MAGIC_V1);
844         rc = dt_declare_xattr_set(env, next, &info->lti_buf, XATTR_NAME_LOV,
845                                   0, th);
846         if (rc)
847                 GOTO(out, rc);
848
849         /*
850          * if striping is created with local object's size > 0,
851          * we have to propagate this size to specific object
852          * the case is possible only when local object was created previously
853          */
854         if (dt_object_exists(next))
855                 rc = lod_declare_init_size(env, dt, th);
856
857 out:
858         RETURN(rc);
859 }
860
861 static int lod_declare_object_create(const struct lu_env *env,
862                                      struct dt_object *dt,
863                                      struct lu_attr *attr,
864                                      struct dt_allocation_hint *hint,
865                                      struct dt_object_format *dof,
866                                      struct thandle *th)
867 {
868         struct dt_object   *next = dt_object_child(dt);
869         struct lod_object  *lo = lod_dt_obj(dt);
870         int                 rc;
871         ENTRY;
872
873         LASSERT(dof);
874         LASSERT(attr);
875         LASSERT(th);
876         LASSERT(!dt_object_exists(next));
877
878         /*
879          * first of all, we declare creation of local object
880          */
881         rc = dt_declare_create(env, next, attr, hint, dof, th);
882         if (rc)
883                 GOTO(out, rc);
884
885         if (dof->dof_type == DFT_SYM)
886                 dt->do_body_ops = &lod_body_lnk_ops;
887
888         /*
889          * it's lod_ah_init() who has decided the object will striped
890          */
891         if (dof->dof_type == DFT_REGULAR) {
892                 /* callers don't want stripes */
893                 /* XXX: all tricky interactions with ->ah_make_hint() decided
894                  * to use striping, then ->declare_create() behaving differently
895                  * should be cleaned */
896                 if (dof->u.dof_reg.striped == 0)
897                         lo->ldo_stripenr = 0;
898                 if (lo->ldo_stripenr > 0)
899                         rc = lod_declare_striped_object(env, dt, attr,
900                                                         NULL, th);
901         } else if (dof->dof_type == DFT_DIR && lo->ldo_striping_cached) {
902                 struct lod_thread_info *info = lod_env_info(env);
903
904                 info->lti_buf.lb_buf = NULL;
905                 info->lti_buf.lb_len = sizeof(struct lov_user_md_v3);
906                 /* to transfer default striping from the parent */
907                 rc = dt_declare_xattr_set(env, next, &info->lti_buf,
908                                           XATTR_NAME_LOV, 0, th);
909         }
910
911 out:
912         RETURN(rc);
913 }
914
915 int lod_striping_create(const struct lu_env *env, struct dt_object *dt,
916                         struct lu_attr *attr, struct dt_object_format *dof,
917                         struct thandle *th)
918 {
919         struct lod_object *lo = lod_dt_obj(dt);
920         int                rc = 0, i;
921         ENTRY;
922
923         LASSERT(lo->ldo_stripe);
924         LASSERT(lo->ldo_stripe > 0);
925         LASSERT(lo->ldo_striping_cached == 0);
926
927         /* create all underlying objects */
928         for (i = 0; i < lo->ldo_stripenr; i++) {
929                 LASSERT(lo->ldo_stripe[i]);
930                 rc = dt_create(env, lo->ldo_stripe[i], attr, NULL, dof, th);
931
932                 if (rc)
933                         break;
934         }
935         if (rc == 0)
936                 rc = lod_generate_and_set_lovea(env, lo, th);
937
938         RETURN(rc);
939 }
940
941 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
942                              struct lu_attr *attr,
943                              struct dt_allocation_hint *hint,
944                              struct dt_object_format *dof, struct thandle *th)
945 {
946         struct dt_object   *next = dt_object_child(dt);
947         struct lod_object  *lo = lod_dt_obj(dt);
948         int                 rc;
949         ENTRY;
950
951         /* create local object */
952         rc = dt_create(env, next, attr, hint, dof, th);
953
954         if (rc == 0) {
955                 if (S_ISDIR(dt->do_lu.lo_header->loh_attr))
956                         rc = lod_store_def_striping(env, dt, th);
957                 else if (lo->ldo_stripe)
958                         rc = lod_striping_create(env, dt, attr, dof, th);
959         }
960
961         RETURN(rc);
962 }
963
964 static int lod_declare_object_destroy(const struct lu_env *env,
965                                       struct dt_object *dt,
966                                       struct thandle *th)
967 {
968         struct dt_object   *next = dt_object_child(dt);
969         struct lod_object  *lo = lod_dt_obj(dt);
970         int                 rc, i;
971         ENTRY;
972
973         /*
974          * we declare destroy for the local object
975          */
976         rc = dt_declare_destroy(env, next, th);
977         if (rc)
978                 RETURN(rc);
979
980         /*
981          * load striping information, notice we don't do this when object
982          * is being initialized as we don't need this information till
983          * few specific cases like destroy, chown
984          */
985         rc = lod_load_striping(env, lo);
986         if (rc)
987                 RETURN(rc);
988
989         /* declare destroy for all underlying objects */
990         for (i = 0; i < lo->ldo_stripenr; i++) {
991                 LASSERT(lo->ldo_stripe[i]);
992                 rc = dt_declare_destroy(env, lo->ldo_stripe[i], th);
993
994                 if (rc)
995                         break;
996         }
997
998         RETURN(rc);
999 }
1000
1001 static int lod_object_destroy(const struct lu_env *env,
1002                 struct dt_object *dt, struct thandle *th)
1003 {
1004         struct dt_object  *next = dt_object_child(dt);
1005         struct lod_object *lo = lod_dt_obj(dt);
1006         int                rc, i;
1007         ENTRY;
1008
1009         /* destroy local object */
1010         rc = dt_destroy(env, next, th);
1011         if (rc)
1012                 RETURN(rc);
1013
1014         /* destroy all underlying objects */
1015         for (i = 0; i < lo->ldo_stripenr; i++) {
1016                 LASSERT(lo->ldo_stripe[i]);
1017                 rc = dt_destroy(env, lo->ldo_stripe[i], th);
1018                 if (rc)
1019                         break;
1020         }
1021
1022         RETURN(rc);
1023 }
1024
1025 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
1026                          const struct dt_index_features *feat)
1027 {
1028         struct dt_object *next = dt_object_child(dt);
1029         int               rc;
1030         ENTRY;
1031
1032         LASSERT(next->do_ops);
1033         LASSERT(next->do_ops->do_index_try);
1034
1035         rc = next->do_ops->do_index_try(env, next, feat);
1036         if (next->do_index_ops && dt->do_index_ops == NULL)
1037                 dt->do_index_ops = &lod_index_ops;
1038
1039         RETURN(rc);
1040 }
1041
1042 static int lod_declare_ref_add(const struct lu_env *env,
1043                                struct dt_object *dt, struct thandle *th)
1044 {
1045         return dt_declare_ref_add(env, dt_object_child(dt), th);
1046 }
1047
1048 static int lod_ref_add(const struct lu_env *env,
1049                        struct dt_object *dt, struct thandle *th)
1050 {
1051         return dt_ref_add(env, dt_object_child(dt), th);
1052 }
1053
1054 static int lod_declare_ref_del(const struct lu_env *env,
1055                                struct dt_object *dt, struct thandle *th)
1056 {
1057         return dt_declare_ref_del(env, dt_object_child(dt), th);
1058 }
1059
1060 static int lod_ref_del(const struct lu_env *env,
1061                        struct dt_object *dt, struct thandle *th)
1062 {
1063         return dt_ref_del(env, dt_object_child(dt), th);
1064 }
1065
1066 static struct obd_capa *lod_capa_get(const struct lu_env *env,
1067                                      struct dt_object *dt,
1068                                      struct lustre_capa *old, __u64 opc)
1069 {
1070         return dt_capa_get(env, dt_object_child(dt), old, opc);
1071 }
1072
1073 static int lod_object_sync(const struct lu_env *env, struct dt_object *dt)
1074 {
1075         return dt_object_sync(env, dt_object_child(dt));
1076 }
1077
1078 struct dt_object_operations lod_obj_ops = {
1079         .do_read_lock           = lod_object_read_lock,
1080         .do_write_lock          = lod_object_write_lock,
1081         .do_read_unlock         = lod_object_read_unlock,
1082         .do_write_unlock        = lod_object_write_unlock,
1083         .do_write_locked        = lod_object_write_locked,
1084         .do_attr_get            = lod_attr_get,
1085         .do_declare_attr_set    = lod_declare_attr_set,
1086         .do_attr_set            = lod_attr_set,
1087         .do_xattr_get           = lod_xattr_get,
1088         .do_declare_xattr_set   = lod_declare_xattr_set,
1089         .do_xattr_set           = lod_xattr_set,
1090         .do_declare_xattr_del   = lod_declare_xattr_del,
1091         .do_xattr_del           = lod_xattr_del,
1092         .do_xattr_list          = lod_xattr_list,
1093         .do_ah_init             = lod_ah_init,
1094         .do_declare_create      = lod_declare_object_create,
1095         .do_create              = lod_object_create,
1096         .do_declare_destroy     = lod_declare_object_destroy,
1097         .do_destroy             = lod_object_destroy,
1098         .do_index_try           = lod_index_try,
1099         .do_declare_ref_add     = lod_declare_ref_add,
1100         .do_ref_add             = lod_ref_add,
1101         .do_declare_ref_del     = lod_declare_ref_del,
1102         .do_ref_del             = lod_ref_del,
1103         .do_capa_get            = lod_capa_get,
1104         .do_object_sync         = lod_object_sync,
1105 };
1106
1107 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
1108                         struct lu_buf *buf, loff_t *pos,
1109                         struct lustre_capa *capa)
1110 {
1111         struct dt_object *next = dt_object_child(dt);
1112         return next->do_body_ops->dbo_read(env, next, buf, pos, capa);
1113 }
1114
1115 static ssize_t lod_declare_write(const struct lu_env *env,
1116                                  struct dt_object *dt,
1117                                  const loff_t size, loff_t pos,
1118                                  struct thandle *th)
1119 {
1120         return dt_declare_record_write(env, dt_object_child(dt),
1121                                        size, pos, th);
1122 }
1123
1124 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,
1125                          const struct lu_buf *buf, loff_t *pos,
1126                          struct thandle *th, struct lustre_capa *capa, int iq)
1127 {
1128         struct dt_object *next = dt_object_child(dt);
1129         LASSERT(next);
1130         return next->do_body_ops->dbo_write(env, next, buf, pos, th, capa, iq);
1131 }
1132
1133 static const struct dt_body_operations lod_body_lnk_ops = {
1134         .dbo_read               = lod_read,
1135         .dbo_declare_write      = lod_declare_write,
1136         .dbo_write              = lod_write
1137 };
1138
1139 static int lod_object_init(const struct lu_env *env, struct lu_object *o,
1140                            const struct lu_object_conf *conf)
1141 {
1142         struct lod_device *d = lu2lod_dev(o->lo_dev);
1143         struct lu_object  *below;
1144         struct lu_device  *under;
1145         ENTRY;
1146
1147         /*
1148          * create local object
1149          */
1150         under = &d->lod_child->dd_lu_dev;
1151         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
1152         if (below == NULL)
1153                 RETURN(-ENOMEM);
1154
1155         lu_object_add(o, below);
1156
1157         RETURN(0);
1158 }
1159
1160 void lod_object_free_striping(const struct lu_env *env, struct lod_object *lo)
1161 {
1162         int i;
1163
1164         if (lo->ldo_stripe) {
1165                 LASSERT(lo->ldo_stripes_allocated > 0);
1166
1167                 for (i = 0; i < lo->ldo_stripenr; i++) {
1168                         if (lo->ldo_stripe[i])
1169                                 lu_object_put(env, &lo->ldo_stripe[i]->do_lu);
1170                 }
1171
1172                 i = sizeof(struct dt_object *) * lo->ldo_stripes_allocated;
1173                 OBD_FREE(lo->ldo_stripe, i);
1174                 lo->ldo_stripe = NULL;
1175                 lo->ldo_stripes_allocated = 0;
1176         }
1177         lo->ldo_stripenr = 0;
1178 }
1179
1180 /*
1181  * ->start is called once all slices are initialized, including header's
1182  * cache for mode (object type). using the type we can initialize ops
1183  */
1184 static int lod_object_start(const struct lu_env *env, struct lu_object *o)
1185 {
1186         if (S_ISLNK(o->lo_header->loh_attr & S_IFMT))
1187                 lu2lod_obj(o)->ldo_obj.do_body_ops = &lod_body_lnk_ops;
1188         return 0;
1189 }
1190
1191 static void lod_object_free(const struct lu_env *env, struct lu_object *o)
1192 {
1193         struct lod_object *mo = lu2lod_obj(o);
1194
1195         /*
1196          * release all underlying object pinned
1197          */
1198
1199         lod_object_free_striping(env, mo);
1200
1201         lod_object_set_pool(mo, NULL);
1202
1203         lu_object_fini(o);
1204         OBD_SLAB_FREE_PTR(mo, lod_object_kmem);
1205 }
1206
1207 static void lod_object_release(const struct lu_env *env, struct lu_object *o)
1208 {
1209         /* XXX: shouldn't we release everything here in case if object
1210          * creation failed before? */
1211 }
1212
1213 static int lod_object_print(const struct lu_env *env, void *cookie,
1214                             lu_printer_t p, const struct lu_object *l)
1215 {
1216         struct lod_object *o = lu2lod_obj((struct lu_object *) l);
1217
1218         return (*p)(env, cookie, LUSTRE_LOD_NAME"-object@%p", o);
1219 }
1220
1221 struct lu_object_operations lod_lu_obj_ops = {
1222         .loo_object_init        = lod_object_init,
1223         .loo_object_start       = lod_object_start,
1224         .loo_object_free        = lod_object_free,
1225         .loo_object_release     = lod_object_release,
1226         .loo_object_print       = lod_object_print,
1227 };