4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
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.
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
23 * Copyright 2009 Sun Microsystems, Inc. All rights reserved
24 * Use is subject to license terms.
26 * Copyright (c) 2011, 2012, Intel, Inc.
29 * lustre/lod/lod_object.c
31 * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
35 # define EXPORT_SYMTAB
37 #define DEBUG_SUBSYSTEM S_MDS
40 #include <obd_class.h>
41 #include <lustre_ver.h>
42 #include <obd_support.h>
43 #include <lprocfs_status.h>
45 #include <lustre_fid.h>
46 #include <lustre_param.h>
47 #include <lustre_fid.h>
50 #include "lod_internal.h"
52 extern cfs_mem_cache_t *lod_object_kmem;
53 static const struct dt_body_operations lod_body_lnk_ops;
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)
59 struct dt_object *next = dt_object_child(dt);
60 return next->do_index_ops->dio_lookup(env, next, rec, key, capa);
63 static int lod_declare_index_insert(const struct lu_env *env,
65 const struct dt_rec *rec,
66 const struct dt_key *key,
67 struct thandle *handle)
69 return dt_declare_insert(env, dt_object_child(dt), rec, key, handle);
72 static int lod_index_insert(const struct lu_env *env,
74 const struct dt_rec *rec,
75 const struct dt_key *key,
77 struct lustre_capa *capa,
80 return dt_insert(env, dt_object_child(dt), rec, key, th, capa, ign);
83 static int lod_declare_index_delete(const struct lu_env *env,
85 const struct dt_key *key,
88 return dt_declare_delete(env, dt_object_child(dt), key, th);
91 static int lod_index_delete(const struct lu_env *env,
93 const struct dt_key *key,
95 struct lustre_capa *capa)
97 return dt_delete(env, dt_object_child(dt), key, th, capa);
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)
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;
109 it_next = next->do_index_ops->dio_it.init(env, next, attr, capa);
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
117 LASSERT(it->lit_obj == NULL);
119 it->lit_it = it_next;
122 return (struct dt_it *)it;
125 #define LOD_CHECK_IT(env, it) \
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); \
133 void lod_it_fini(const struct lu_env *env, struct dt_it *di)
135 struct lod_it *it = (struct lod_it *)di;
137 LOD_CHECK_IT(env, it);
138 it->lit_obj->do_index_ops->dio_it.fini(env, it->lit_it);
140 /* the iterator not in use any more */
145 int lod_it_get(const struct lu_env *env, struct dt_it *di,
146 const struct dt_key *key)
148 const struct lod_it *it = (const struct lod_it *)di;
150 LOD_CHECK_IT(env, it);
151 return it->lit_obj->do_index_ops->dio_it.get(env, it->lit_it, key);
154 void lod_it_put(const struct lu_env *env, struct dt_it *di)
156 struct lod_it *it = (struct lod_it *)di;
158 LOD_CHECK_IT(env, it);
159 return it->lit_obj->do_index_ops->dio_it.put(env, it->lit_it);
162 int lod_it_next(const struct lu_env *env, struct dt_it *di)
164 struct lod_it *it = (struct lod_it *)di;
166 LOD_CHECK_IT(env, it);
167 return it->lit_obj->do_index_ops->dio_it.next(env, it->lit_it);
170 struct dt_key *lod_it_key(const struct lu_env *env, const struct dt_it *di)
172 const struct lod_it *it = (const struct lod_it *)di;
174 LOD_CHECK_IT(env, it);
175 return it->lit_obj->do_index_ops->dio_it.key(env, it->lit_it);
178 int lod_it_key_size(const struct lu_env *env, const struct dt_it *di)
180 struct lod_it *it = (struct lod_it *)di;
182 LOD_CHECK_IT(env, it);
183 return it->lit_obj->do_index_ops->dio_it.key_size(env, it->lit_it);
186 int lod_it_rec(const struct lu_env *env, const struct dt_it *di,
187 struct dt_rec *rec, __u32 attr)
189 const struct lod_it *it = (const struct lod_it *)di;
191 LOD_CHECK_IT(env, it);
192 return it->lit_obj->do_index_ops->dio_it.rec(env, it->lit_it, rec, attr);
195 __u64 lod_it_store(const struct lu_env *env, const struct dt_it *di)
197 const struct lod_it *it = (const struct lod_it *)di;
199 LOD_CHECK_IT(env, it);
200 return it->lit_obj->do_index_ops->dio_it.store(env, it->lit_it);
203 int lod_it_load(const struct lu_env *env, const struct dt_it *di, __u64 hash)
205 const struct lod_it *it = (const struct lod_it *)di;
207 LOD_CHECK_IT(env, it);
208 return it->lit_obj->do_index_ops->dio_it.load(env, it->lit_it, hash);
211 int lod_it_key_rec(const struct lu_env *env, const struct dt_it *di,
214 const struct lod_it *it = (const struct lod_it *)di;
216 LOD_CHECK_IT(env, it);
217 return it->lit_obj->do_index_ops->dio_it.key_rec(env, it->lit_it, key_rec);
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,
233 .key_size = lod_it_key_size,
235 .store = lod_it_store,
237 .key_rec = lod_it_key_rec,
241 static void lod_object_read_lock(const struct lu_env *env,
242 struct dt_object *dt, unsigned role)
244 dt_read_lock(env, dt_object_child(dt), role);
247 static void lod_object_write_lock(const struct lu_env *env,
248 struct dt_object *dt, unsigned role)
250 dt_write_lock(env, dt_object_child(dt), role);
253 static void lod_object_read_unlock(const struct lu_env *env,
254 struct dt_object *dt)
256 dt_read_unlock(env, dt_object_child(dt));
259 static void lod_object_write_unlock(const struct lu_env *env,
260 struct dt_object *dt)
262 dt_write_unlock(env, dt_object_child(dt));
265 static int lod_object_write_locked(const struct lu_env *env,
266 struct dt_object *dt)
268 return dt_write_locked(env, dt_object_child(dt));
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)
276 return dt_attr_get(env, dt_object_child(dt), attr, capa);
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)
284 struct dt_object *next = dt_object_child(dt);
285 struct lod_object *lo = lod_dt_obj(dt);
290 * declare setattr on the local object
292 rc = dt_declare_attr_set(env, next, attr, handle);
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
301 rc = lod_load_striping(env, lo);
306 * if object is striped declare changes on the stripes
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);
313 CERROR("failed declaration: %d\n", rc);
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)
327 struct dt_object *next = dt_object_child(dt);
328 struct lod_object *lo = lod_dt_obj(dt);
333 * apply changes to the local object
335 rc = dt_attr_set(env, next, attr, handle, capa);
340 * if object is striped, apply changes to all the stripes
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);
347 CERROR("failed declaration: %d\n", rc);
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)
359 struct lod_thread_info *info = lod_env_info(env);
360 struct lod_device *dev = lu2lod_dev(dt->do_lu.lo_dev);
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))
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.
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));
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;
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);
400 * LOV xattr is a storage for striping, and LOD owns this xattr.
401 * but LOD allows others to control striping to some extent
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
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,
415 struct dt_object *next = dt_object_child(dt);
416 struct lu_attr *attr = &lod_env_info(env)->lti_attr;
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)
426 mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
427 if ((S_ISREG(mode) || !mode) && !strcmp(name, XATTR_NAME_LOV)) {
429 * this is a request to manipulate object's striping
431 if (dt_object_exists(dt)) {
432 rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
436 memset(attr, 0, sizeof(attr));
437 attr->la_valid = LA_TYPE | LA_MODE;
438 attr->la_mode = S_IFREG;
440 rc = lod_declare_striped_object(env, dt, attr, buf, th);
445 rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
450 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
451 struct dt_object *dt,
452 const struct lu_buf *buf,
453 const char *name, int fl,
455 struct lustre_capa *capa)
457 struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
458 struct dt_object *next = dt_object_child(dt);
459 struct lod_object *l = lod_dt_obj(dt);
460 struct lov_user_md_v1 *lum;
461 struct lov_user_md_v3 *v3 = NULL;
465 LASSERT(l->ldo_stripe == NULL);
466 l->ldo_striping_cached = 0;
467 l->ldo_def_striping_set = 0;
468 lod_object_set_pool(l, NULL);
469 l->ldo_def_stripe_size = 0;
470 l->ldo_def_stripenr = 0;
473 LASSERT(buf->lb_buf);
476 rc = lod_verify_striping(d, buf, 0);
480 if (lum->lmm_magic == LOV_USER_MAGIC_V3)
483 /* if { size, offset, count } = { 0, -1, 0 } and no pool
484 * (i.e. all default values specified) then delete default
485 * striping from dir. */
487 "set default striping: sz %u # %u offset %d %s %s\n",
488 (unsigned)lum->lmm_stripe_size,
489 (unsigned)lum->lmm_stripe_count,
490 (int)lum->lmm_stripe_offset,
491 v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
493 if (LOVEA_DELETE_VALUES((lum->lmm_stripe_size),
494 (lum->lmm_stripe_count),
495 (lum->lmm_stripe_offset)) &&
496 lum->lmm_magic == LOV_USER_MAGIC_V1) {
497 rc = dt_xattr_del(env, next, name, th, capa);
501 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
507 static int lod_xattr_set(const struct lu_env *env,
508 struct dt_object *dt, const struct lu_buf *buf,
509 const char *name, int fl, struct thandle *th,
510 struct lustre_capa *capa)
512 struct dt_object *next = dt_object_child(dt);
517 attr = dt->do_lu.lo_header->loh_attr & S_IFMT;
519 if (strncmp(name, XATTR_NAME_LOV, strlen(XATTR_NAME_LOV)) == 0)
520 rc = lod_xattr_set_lov_on_dir(env, dt, buf, name,
523 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
525 } else if (S_ISREG(attr) && !strcmp(name, XATTR_NAME_LOV)) {
527 * XXX: check striping match what we already have
528 * during req replay, declare_xattr_set() defines striping,
529 * then create() does the work
531 rc = lod_striping_create(env, dt, NULL, NULL, th);
535 * behave transparantly for all other EAs
537 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
543 static int lod_declare_xattr_del(const struct lu_env *env,
544 struct dt_object *dt, const char *name,
547 return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
550 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
551 const char *name, struct thandle *th,
552 struct lustre_capa *capa)
554 return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
557 static int lod_xattr_list(const struct lu_env *env,
558 struct dt_object *dt, struct lu_buf *buf,
559 struct lustre_capa *capa)
561 return dt_xattr_list(env, dt_object_child(dt), buf, capa);
564 int lod_object_set_pool(struct lod_object *o, char *pool)
569 len = strlen(o->ldo_pool);
570 OBD_FREE(o->ldo_pool, len + 1);
575 OBD_ALLOC(o->ldo_pool, len + 1);
576 if (o->ldo_pool == NULL)
578 strcpy(o->ldo_pool, pool);
583 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
585 return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
588 static int lod_cache_parent_striping(const struct lu_env *env,
589 struct lod_object *lp)
591 struct lov_user_md_v1 *v1 = NULL;
592 struct lov_user_md_v3 *v3 = NULL;
596 /* dt_ah_init() is called from MDD without parent being write locked
598 dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
599 if (lp->ldo_striping_cached)
600 GOTO(unlock, rc = 0);
602 rc = lod_get_lov_ea(env, lp);
606 if (rc < sizeof(struct lov_user_md)) {
607 /* don't lookup for non-existing or invalid striping */
608 lp->ldo_def_striping_set = 0;
609 lp->ldo_striping_cached = 1;
610 lp->ldo_def_stripe_size = 0;
611 lp->ldo_def_stripenr = 0;
612 lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1);
613 GOTO(unlock, rc = 0);
616 v1 = (struct lov_user_md_v1 *)lod_env_info(env)->lti_ea_store;
617 if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1))
618 lustre_swab_lov_user_md_v1(v1);
619 else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3))
620 lustre_swab_lov_user_md_v3(v3);
622 if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1)
623 GOTO(unlock, rc = 0);
625 if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
626 GOTO(unlock, rc = 0);
628 lp->ldo_def_stripenr = v1->lmm_stripe_count;
629 lp->ldo_def_stripe_size = v1->lmm_stripe_size;
630 lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
631 lp->ldo_striping_cached = 1;
632 lp->ldo_def_striping_set = 1;
634 if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
635 /* XXX: sanity check here */
636 v3 = (struct lov_user_md_v3 *) v1;
637 if (v3->lmm_pool_name[0])
638 lod_object_set_pool(lp, v3->lmm_pool_name);
641 CDEBUG(D_OTHER, "def. striping: # %d, sz %d, off %d %s%s on "DFID"\n",
642 lp->ldo_def_stripenr, lp->ldo_def_stripe_size,
643 lp->ldo_def_stripe_offset, v3 ? "from " : "",
644 v3 ? lp->ldo_pool : "", PFID(lu_object_fid(&lp->ldo_obj.do_lu)));
648 dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
653 * used to transfer default striping data to the object being created
655 static void lod_ah_init(const struct lu_env *env,
656 struct dt_allocation_hint *ah,
657 struct dt_object *parent,
658 struct dt_object *child,
659 cfs_umode_t child_mode)
661 struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
662 struct dt_object *nextp = NULL;
663 struct dt_object *nextc;
664 struct lod_object *lp = NULL;
665 struct lod_object *lc;
666 struct lov_desc *desc;
671 if (likely(parent)) {
672 nextp = dt_object_child(parent);
673 lp = lod_dt_obj(parent);
676 nextc = dt_object_child(child);
677 lc = lod_dt_obj(child);
679 LASSERT(lc->ldo_stripenr == 0);
680 LASSERT(lc->ldo_stripe == NULL);
683 * local object may want some hints
684 * in case of late striping creation, ->ah_init()
685 * can be called with local object existing
687 if (!dt_object_exists(nextc))
688 nextc->do_ops->do_ah_init(env, ah, nextp, nextc, child_mode);
690 if (S_ISDIR(child_mode)) {
691 if (lp->ldo_striping_cached == 0) {
692 /* we haven't tried to get default striping for
693 * the directory yet, let's cache it in the object */
694 lod_cache_parent_striping(env, lp);
696 /* transfer defaults to new directory */
697 if (lp->ldo_striping_cached) {
699 lod_object_set_pool(lc, lp->ldo_pool);
700 lc->ldo_def_stripenr = lp->ldo_def_stripenr;
701 lc->ldo_def_stripe_size = lp->ldo_def_stripe_size;
702 lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
703 lc->ldo_striping_cached = 1;
704 lc->ldo_def_striping_set = 1;
705 CDEBUG(D_OTHER, "inherite striping defaults\n");
711 * if object is going to be striped over OSTs, transfer default
712 * striping information to the child, so that we can use it
713 * during declaration and creation
715 if (!lod_object_will_be_striped(S_ISREG(child_mode),
716 lu_object_fid(&child->do_lu)))
720 * try from the parent
722 if (likely(parent)) {
723 if (lp->ldo_striping_cached == 0) {
724 /* we haven't tried to get default striping for
725 * the directory yet, let's cache it in the object */
726 lod_cache_parent_striping(env, lp);
729 lc->ldo_def_stripe_offset = (__u16) -1;
731 if (lp->ldo_def_striping_set) {
733 lod_object_set_pool(lc, lp->ldo_pool);
734 lc->ldo_stripenr = lp->ldo_def_stripenr;
735 lc->ldo_stripe_size = lp->ldo_def_stripe_size;
736 lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
737 CDEBUG(D_OTHER, "striping from parent: #%d, sz %d %s\n",
738 lc->ldo_stripenr, lc->ldo_stripe_size,
739 lp->ldo_pool ? lp->ldo_pool : "");
744 * if the parent doesn't provide with specific pattern, grab fs-wide one
747 if (lc->ldo_stripenr == 0)
748 lc->ldo_stripenr = desc->ld_default_stripe_count;
749 if (lc->ldo_stripe_size == 0)
750 lc->ldo_stripe_size = desc->ld_default_stripe_size;
751 CDEBUG(D_OTHER, "final striping: # %d stripes, sz %d from %s\n",
752 lc->ldo_stripenr, lc->ldo_stripe_size,
753 lc->ldo_pool ? lc->ldo_pool : "");
758 #define ll_do_div64(aaa,bbb) do_div((aaa), (bbb))
760 * this function handles a special case when truncate was done
761 * on a stripeless object and now striping is being created
762 * we can't lose that size, so we have to propagate it to newly
765 static int lod_declare_init_size(const struct lu_env *env,
766 struct dt_object *dt, struct thandle *th)
768 struct dt_object *next = dt_object_child(dt);
769 struct lod_object *lo = lod_dt_obj(dt);
770 struct lu_attr *attr = &lod_env_info(env)->lti_attr;
775 /* XXX: we support the simplest (RAID0) striping so far */
776 LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
777 LASSERT(lo->ldo_stripe_size > 0);
779 rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
780 LASSERT(attr->la_valid & LA_SIZE);
784 size = attr->la_size;
788 /* ll_do_div64(a, b) returns a % b, and a = a / b */
789 ll_do_div64(size, (__u64) lo->ldo_stripe_size);
790 stripe = ll_do_div64(size, (__u64) lo->ldo_stripenr);
792 size = size * lo->ldo_stripe_size;
793 offs = attr->la_size;
794 size += ll_do_div64(offs, lo->ldo_stripe_size);
796 attr->la_valid = LA_SIZE;
797 attr->la_size = size;
799 rc = dt_declare_attr_set(env, lo->ldo_stripe[stripe], attr, th);
806 * Create declaration of striped object
808 int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt,
809 struct lu_attr *attr,
810 const struct lu_buf *lovea, struct thandle *th)
812 struct lod_thread_info *info = lod_env_info(env);
813 struct dt_object *next = dt_object_child(dt);
814 struct lod_object *lo = lod_dt_obj(dt);
818 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) {
819 /* failed to create striping, let's reset
820 * config so that others don't get confused */
821 lod_object_free_striping(env, lo);
822 GOTO(out, rc = -ENOMEM);
825 /* choose OST and generate appropriate objects */
826 rc = lod_qos_prep_create(env, lo, attr, lovea, th);
828 /* failed to create striping, let's reset
829 * config so that others don't get confused */
830 lod_object_free_striping(env, lo);
835 * declare storage for striping data
837 info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
838 lo->ldo_pool ? LOV_MAGIC_V3 : LOV_MAGIC_V1);
839 rc = dt_declare_xattr_set(env, next, &info->lti_buf, XATTR_NAME_LOV,
845 * if striping is created with local object's size > 0,
846 * we have to propagate this size to specific object
847 * the case is possible only when local object was created previously
849 if (dt_object_exists(next))
850 rc = lod_declare_init_size(env, dt, th);
856 static int lod_declare_object_create(const struct lu_env *env,
857 struct dt_object *dt,
858 struct lu_attr *attr,
859 struct dt_allocation_hint *hint,
860 struct dt_object_format *dof,
863 struct dt_object *next = dt_object_child(dt);
864 struct lod_object *lo = lod_dt_obj(dt);
871 LASSERT(!dt_object_exists(next));
874 * first of all, we declare creation of local object
876 rc = dt_declare_create(env, next, attr, hint, dof, th);
880 if (dof->dof_type == DFT_SYM)
881 dt->do_body_ops = &lod_body_lnk_ops;
884 * it's lod_ah_init() who has decided the object will striped
886 if (dof->dof_type == DFT_REGULAR) {
887 /* callers don't want stripes */
888 /* XXX: all tricky interactions with ->ah_make_hint() decided
889 * to use striping, then ->declare_create() behaving differently
890 * should be cleaned */
891 if (dof->u.dof_reg.striped == 0)
892 lo->ldo_stripenr = 0;
893 if (lo->ldo_stripenr > 0)
894 rc = lod_declare_striped_object(env, dt, attr,
896 } else if (dof->dof_type == DFT_DIR && lo->ldo_striping_cached) {
897 struct lod_thread_info *info = lod_env_info(env);
899 info->lti_buf.lb_buf = NULL;
900 info->lti_buf.lb_len = sizeof(struct lov_user_md_v3);
901 /* to transfer default striping from the parent */
902 rc = dt_declare_xattr_set(env, next, &info->lti_buf,
903 XATTR_NAME_LOV, 0, th);
910 int lod_striping_create(const struct lu_env *env, struct dt_object *dt,
911 struct lu_attr *attr, struct dt_object_format *dof,
914 struct lod_object *lo = lod_dt_obj(dt);
918 LASSERT(lo->ldo_stripe);
919 LASSERT(lo->ldo_stripe > 0);
920 LASSERT(lo->ldo_striping_cached == 0);
922 /* create all underlying objects */
923 for (i = 0; i < lo->ldo_stripenr; i++) {
924 LASSERT(lo->ldo_stripe[i]);
925 rc = dt_create(env, lo->ldo_stripe[i], attr, NULL, dof, th);
931 rc = lod_generate_and_set_lovea(env, lo, th);
936 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
937 struct lu_attr *attr,
938 struct dt_allocation_hint *hint,
939 struct dt_object_format *dof, struct thandle *th)
941 struct dt_object *next = dt_object_child(dt);
942 struct lod_object *lo = lod_dt_obj(dt);
946 /* create local object */
947 rc = dt_create(env, next, attr, hint, dof, th);
950 if (S_ISDIR(dt->do_lu.lo_header->loh_attr))
951 rc = lod_store_def_striping(env, dt, th);
952 else if (lo->ldo_stripe)
953 rc = lod_striping_create(env, dt, attr, dof, th);
959 static int lod_declare_object_destroy(const struct lu_env *env,
960 struct dt_object *dt,
963 struct dt_object *next = dt_object_child(dt);
964 struct lod_object *lo = lod_dt_obj(dt);
969 * we declare destroy for the local object
971 rc = dt_declare_destroy(env, next, th);
976 * load striping information, notice we don't do this when object
977 * is being initialized as we don't need this information till
978 * few specific cases like destroy, chown
980 rc = lod_load_striping(env, lo);
984 /* declare destroy for all underlying objects */
985 for (i = 0; i < lo->ldo_stripenr; i++) {
986 LASSERT(lo->ldo_stripe[i]);
987 rc = dt_declare_destroy(env, lo->ldo_stripe[i], th);
996 static int lod_object_destroy(const struct lu_env *env,
997 struct dt_object *dt, struct thandle *th)
999 struct dt_object *next = dt_object_child(dt);
1000 struct lod_object *lo = lod_dt_obj(dt);
1004 /* destroy local object */
1005 rc = dt_destroy(env, next, th);
1009 /* destroy all underlying objects */
1010 for (i = 0; i < lo->ldo_stripenr; i++) {
1011 LASSERT(lo->ldo_stripe[i]);
1012 rc = dt_destroy(env, lo->ldo_stripe[i], th);
1020 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
1021 const struct dt_index_features *feat)
1023 struct dt_object *next = dt_object_child(dt);
1027 LASSERT(next->do_ops);
1028 LASSERT(next->do_ops->do_index_try);
1030 rc = next->do_ops->do_index_try(env, next, feat);
1031 if (next->do_index_ops && dt->do_index_ops == NULL)
1032 dt->do_index_ops = &lod_index_ops;
1037 static int lod_declare_ref_add(const struct lu_env *env,
1038 struct dt_object *dt, struct thandle *th)
1040 return dt_declare_ref_add(env, dt_object_child(dt), th);
1043 static int lod_ref_add(const struct lu_env *env,
1044 struct dt_object *dt, struct thandle *th)
1046 return dt_ref_add(env, dt_object_child(dt), th);
1049 static int lod_declare_ref_del(const struct lu_env *env,
1050 struct dt_object *dt, struct thandle *th)
1052 return dt_declare_ref_del(env, dt_object_child(dt), th);
1055 static int lod_ref_del(const struct lu_env *env,
1056 struct dt_object *dt, struct thandle *th)
1058 return dt_ref_del(env, dt_object_child(dt), th);
1061 static struct obd_capa *lod_capa_get(const struct lu_env *env,
1062 struct dt_object *dt,
1063 struct lustre_capa *old, __u64 opc)
1065 return dt_capa_get(env, dt_object_child(dt), old, opc);
1068 static int lod_object_sync(const struct lu_env *env, struct dt_object *dt)
1070 return dt_object_sync(env, dt_object_child(dt));
1073 struct dt_object_operations lod_obj_ops = {
1074 .do_read_lock = lod_object_read_lock,
1075 .do_write_lock = lod_object_write_lock,
1076 .do_read_unlock = lod_object_read_unlock,
1077 .do_write_unlock = lod_object_write_unlock,
1078 .do_write_locked = lod_object_write_locked,
1079 .do_attr_get = lod_attr_get,
1080 .do_declare_attr_set = lod_declare_attr_set,
1081 .do_attr_set = lod_attr_set,
1082 .do_xattr_get = lod_xattr_get,
1083 .do_declare_xattr_set = lod_declare_xattr_set,
1084 .do_xattr_set = lod_xattr_set,
1085 .do_declare_xattr_del = lod_declare_xattr_del,
1086 .do_xattr_del = lod_xattr_del,
1087 .do_xattr_list = lod_xattr_list,
1088 .do_ah_init = lod_ah_init,
1089 .do_declare_create = lod_declare_object_create,
1090 .do_create = lod_object_create,
1091 .do_declare_destroy = lod_declare_object_destroy,
1092 .do_destroy = lod_object_destroy,
1093 .do_index_try = lod_index_try,
1094 .do_declare_ref_add = lod_declare_ref_add,
1095 .do_ref_add = lod_ref_add,
1096 .do_declare_ref_del = lod_declare_ref_del,
1097 .do_ref_del = lod_ref_del,
1098 .do_capa_get = lod_capa_get,
1099 .do_object_sync = lod_object_sync,
1102 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
1103 struct lu_buf *buf, loff_t *pos,
1104 struct lustre_capa *capa)
1106 struct dt_object *next = dt_object_child(dt);
1107 return next->do_body_ops->dbo_read(env, next, buf, pos, capa);
1110 static ssize_t lod_declare_write(const struct lu_env *env,
1111 struct dt_object *dt,
1112 const loff_t size, loff_t pos,
1115 return dt_declare_record_write(env, dt_object_child(dt),
1119 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,
1120 const struct lu_buf *buf, loff_t *pos,
1121 struct thandle *th, struct lustre_capa *capa, int iq)
1123 struct dt_object *next = dt_object_child(dt);
1125 return next->do_body_ops->dbo_write(env, next, buf, pos, th, capa, iq);
1128 static const struct dt_body_operations lod_body_lnk_ops = {
1129 .dbo_read = lod_read,
1130 .dbo_declare_write = lod_declare_write,
1131 .dbo_write = lod_write
1134 static int lod_object_init(const struct lu_env *env, struct lu_object *o,
1135 const struct lu_object_conf *conf)
1137 struct lod_device *d = lu2lod_dev(o->lo_dev);
1138 struct lu_object *below;
1139 struct lu_device *under;
1143 * create local object
1145 under = &d->lod_child->dd_lu_dev;
1146 below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
1150 lu_object_add(o, below);
1155 void lod_object_free_striping(const struct lu_env *env, struct lod_object *lo)
1159 if (lo->ldo_stripe) {
1160 LASSERT(lo->ldo_stripes_allocated > 0);
1162 for (i = 0; i < lo->ldo_stripenr; i++) {
1163 if (lo->ldo_stripe[i])
1164 lu_object_put(env, &lo->ldo_stripe[i]->do_lu);
1167 i = sizeof(struct dt_object *) * lo->ldo_stripes_allocated;
1168 OBD_FREE(lo->ldo_stripe, i);
1169 lo->ldo_stripe = NULL;
1170 lo->ldo_stripes_allocated = 0;
1172 lo->ldo_stripenr = 0;
1176 * ->start is called once all slices are initialized, including header's
1177 * cache for mode (object type). using the type we can initialize ops
1179 static int lod_object_start(const struct lu_env *env, struct lu_object *o)
1181 if (S_ISLNK(o->lo_header->loh_attr & S_IFMT))
1182 lu2lod_obj(o)->ldo_obj.do_body_ops = &lod_body_lnk_ops;
1186 static void lod_object_free(const struct lu_env *env, struct lu_object *o)
1188 struct lod_object *mo = lu2lod_obj(o);
1191 * release all underlying object pinned
1194 lod_object_free_striping(env, mo);
1196 lod_object_set_pool(mo, NULL);
1199 OBD_SLAB_FREE_PTR(mo, lod_object_kmem);
1202 static void lod_object_release(const struct lu_env *env, struct lu_object *o)
1204 /* XXX: shouldn't we release everything here in case if object
1205 * creation failed before? */
1208 static int lod_object_print(const struct lu_env *env, void *cookie,
1209 lu_printer_t p, const struct lu_object *l)
1211 struct lod_object *o = lu2lod_obj((struct lu_object *) l);
1213 return (*p)(env, cookie, LUSTRE_LOD_NAME"-object@%p", o);
1216 struct lu_object_operations lod_lu_obj_ops = {
1217 .loo_object_init = lod_object_init,
1218 .loo_object_start = lod_object_start,
1219 .loo_object_free = lod_object_free,
1220 .loo_object_release = lod_object_release,
1221 .loo_object_print = lod_object_print,