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);
106 return next->do_index_ops->dio_it.init(env, next, attr, capa);
109 static struct dt_index_operations lod_index_ops = {
110 .dio_lookup = lod_index_lookup,
111 .dio_declare_insert = lod_declare_index_insert,
112 .dio_insert = lod_index_insert,
113 .dio_declare_delete = lod_declare_index_delete,
114 .dio_delete = lod_index_delete,
120 static void lod_object_read_lock(const struct lu_env *env,
121 struct dt_object *dt, unsigned role)
123 dt_read_lock(env, dt_object_child(dt), role);
126 static void lod_object_write_lock(const struct lu_env *env,
127 struct dt_object *dt, unsigned role)
129 dt_write_lock(env, dt_object_child(dt), role);
132 static void lod_object_read_unlock(const struct lu_env *env,
133 struct dt_object *dt)
135 dt_read_unlock(env, dt_object_child(dt));
138 static void lod_object_write_unlock(const struct lu_env *env,
139 struct dt_object *dt)
141 dt_write_unlock(env, dt_object_child(dt));
144 static int lod_object_write_locked(const struct lu_env *env,
145 struct dt_object *dt)
147 return dt_write_locked(env, dt_object_child(dt));
150 static int lod_attr_get(const struct lu_env *env,
151 struct dt_object *dt,
152 struct lu_attr *attr,
153 struct lustre_capa *capa)
155 return dt_attr_get(env, dt_object_child(dt), attr, capa);
158 static int lod_declare_attr_set(const struct lu_env *env,
159 struct dt_object *dt,
160 const struct lu_attr *attr,
161 struct thandle *handle)
163 struct dt_object *next = dt_object_child(dt);
164 struct lod_object *lo = lod_dt_obj(dt);
169 * declare setattr on the local object
171 rc = dt_declare_attr_set(env, next, attr, handle);
176 * load striping information, notice we don't do this when object
177 * is being initialized as we don't need this information till
178 * few specific cases like destroy, chown
180 rc = lod_load_striping(env, lo);
185 * if object is striped declare changes on the stripes
187 LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
188 for (i = 0; i < lo->ldo_stripenr; i++) {
189 LASSERT(lo->ldo_stripe[i]);
190 rc = dt_declare_attr_set(env, lo->ldo_stripe[i], attr, handle);
192 CERROR("failed declaration: %d\n", rc);
200 static int lod_attr_set(const struct lu_env *env,
201 struct dt_object *dt,
202 const struct lu_attr *attr,
203 struct thandle *handle,
204 struct lustre_capa *capa)
206 struct dt_object *next = dt_object_child(dt);
207 struct lod_object *lo = lod_dt_obj(dt);
212 * apply changes to the local object
214 rc = dt_attr_set(env, next, attr, handle, capa);
219 * if object is striped, apply changes to all the stripes
221 LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
222 for (i = 0; i < lo->ldo_stripenr; i++) {
223 LASSERT(lo->ldo_stripe[i]);
224 rc = dt_attr_set(env, lo->ldo_stripe[i], attr, handle, capa);
226 CERROR("failed declaration: %d\n", rc);
234 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
235 struct lu_buf *buf, const char *name,
236 struct lustre_capa *capa)
238 struct lod_thread_info *info = lod_env_info(env);
239 struct lod_device *dev = lu2lod_dev(dt->do_lu.lo_dev);
243 rc = dt_xattr_get(env, dt_object_child(dt), buf, name, capa);
244 if (rc != -ENODATA || !S_ISDIR(dt->do_lu.lo_header->loh_attr & S_IFMT))
248 * lod returns default striping on the real root of the device
249 * this is like the root stores default striping for the whole
250 * filesystem. historically we've been using a different approach
251 * and store it in the config.
253 dt_root_get(env, dev->lod_child, &info->lti_fid);
254 is_root = lu_fid_eq(&info->lti_fid, lu_object_fid(&dt->do_lu));
256 if (is_root && strcmp(XATTR_NAME_LOV, name) == 0) {
257 struct lov_user_md *lum = buf->lb_buf;
258 struct lov_desc *desc = &dev->lod_desc;
260 if (buf->lb_buf == NULL) {
261 rc = sizeof(struct lov_user_md_v1);
262 } else if (buf->lb_len >= sizeof(struct lov_user_md_v1)) {
263 lum->lmm_magic = LOV_USER_MAGIC_V1;
264 lum->lmm_object_seq = FID_SEQ_LOV_DEFAULT;
265 lum->lmm_pattern = desc->ld_pattern;
266 lum->lmm_stripe_size = desc->ld_default_stripe_size;
267 lum->lmm_stripe_count = desc->ld_default_stripe_count;
268 lum->lmm_stripe_offset = desc->ld_default_stripe_offset;
269 rc = sizeof(struct lov_user_md_v1);
279 * LOV xattr is a storage for striping, and LOD owns this xattr.
280 * but LOD allows others to control striping to some extent
282 * - to set new defined striping
283 * - to set new semi-defined striping
284 * - number of stripes is defined
285 * - number of stripes + osts are defined
288 static int lod_declare_xattr_set(const struct lu_env *env,
289 struct dt_object *dt,
290 const struct lu_buf *buf,
291 const char *name, int fl,
294 struct dt_object *next = dt_object_child(dt);
295 struct lu_attr *attr = &lod_env_info(env)->lti_attr;
301 * allow to declare predefined striping on a new (!mode) object
302 * which is supposed to be replay of regular file creation
303 * (when LOV setting is declared)
305 mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
306 if ((S_ISREG(mode) || !mode) && !strcmp(name, XATTR_NAME_LOV)) {
308 * this is a request to manipulate object's striping
310 if (dt_object_exists(dt)) {
311 rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
315 memset(attr, 0, sizeof(attr));
316 attr->la_valid = LA_TYPE | LA_MODE;
317 attr->la_mode = S_IFREG;
319 rc = lod_declare_striped_object(env, dt, attr, buf, th);
324 rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
329 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
330 struct dt_object *dt,
331 const struct lu_buf *buf,
332 const char *name, int fl,
334 struct lustre_capa *capa)
336 struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
337 struct dt_object *next = dt_object_child(dt);
338 struct lod_object *l = lod_dt_obj(dt);
339 struct lov_user_md_v1 *lum;
340 struct lov_user_md_v3 *v3 = NULL;
344 LASSERT(l->ldo_stripe == NULL);
345 l->ldo_striping_cached = 0;
346 l->ldo_def_striping_set = 0;
347 lod_object_set_pool(l, NULL);
348 l->ldo_def_stripe_size = 0;
349 l->ldo_def_stripenr = 0;
352 LASSERT(buf->lb_buf);
355 rc = lod_verify_striping(d, buf, 0);
359 if (lum->lmm_magic == LOV_USER_MAGIC_V3)
362 /* if { size, offset, count } = { 0, -1, 0 } and no pool
363 * (i.e. all default values specified) then delete default
364 * striping from dir. */
366 "set default striping: sz %u # %u offset %d %s %s\n",
367 (unsigned)lum->lmm_stripe_size,
368 (unsigned)lum->lmm_stripe_count,
369 (int)lum->lmm_stripe_offset,
370 v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
372 if (LOVEA_DELETE_VALUES((lum->lmm_stripe_size),
373 (lum->lmm_stripe_count),
374 (lum->lmm_stripe_offset)) &&
375 lum->lmm_magic == LOV_USER_MAGIC_V1) {
376 rc = dt_xattr_del(env, next, name, th, capa);
380 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
386 static int lod_xattr_set(const struct lu_env *env,
387 struct dt_object *dt, const struct lu_buf *buf,
388 const char *name, int fl, struct thandle *th,
389 struct lustre_capa *capa)
391 struct dt_object *next = dt_object_child(dt);
396 attr = dt->do_lu.lo_header->loh_attr & S_IFMT;
398 if (strncmp(name, XATTR_NAME_LOV, strlen(XATTR_NAME_LOV)) == 0)
399 rc = lod_xattr_set_lov_on_dir(env, dt, buf, name,
402 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
404 } else if (S_ISREG(attr) && !strcmp(name, XATTR_NAME_LOV)) {
406 * XXX: check striping match what we already have
407 * during req replay, declare_xattr_set() defines striping,
408 * then create() does the work
410 rc = lod_striping_create(env, dt, NULL, NULL, th);
414 * behave transparantly for all other EAs
416 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
422 static int lod_declare_xattr_del(const struct lu_env *env,
423 struct dt_object *dt, const char *name,
426 return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
429 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
430 const char *name, struct thandle *th,
431 struct lustre_capa *capa)
433 return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
436 static int lod_xattr_list(const struct lu_env *env,
437 struct dt_object *dt, struct lu_buf *buf,
438 struct lustre_capa *capa)
440 return dt_xattr_list(env, dt_object_child(dt), buf, capa);
443 int lod_object_set_pool(struct lod_object *o, char *pool)
448 len = strlen(o->ldo_pool);
449 OBD_FREE(o->ldo_pool, len + 1);
454 OBD_ALLOC(o->ldo_pool, len + 1);
455 if (o->ldo_pool == NULL)
457 strcpy(o->ldo_pool, pool);
462 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
464 return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
467 static int lod_cache_parent_striping(const struct lu_env *env,
468 struct lod_object *lp)
470 struct lov_user_md_v1 *v1 = NULL;
471 struct lov_user_md_v3 *v3 = NULL;
475 /* dt_ah_init() is called from MDD without parent being write locked
477 dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
478 if (lp->ldo_striping_cached)
479 GOTO(unlock, rc = 0);
481 rc = lod_get_lov_ea(env, lp);
485 if (rc < sizeof(struct lov_user_md)) {
486 /* don't lookup for non-existing or invalid striping */
487 lp->ldo_def_striping_set = 0;
488 lp->ldo_striping_cached = 1;
489 lp->ldo_def_stripe_size = 0;
490 lp->ldo_def_stripenr = 0;
491 lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1);
492 GOTO(unlock, rc = 0);
495 v1 = (struct lov_user_md_v1 *)lod_env_info(env)->lti_ea_store;
496 if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1))
497 lustre_swab_lov_user_md_v1(v1);
498 else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3))
499 lustre_swab_lov_user_md_v3(v3);
501 if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1)
502 GOTO(unlock, rc = 0);
504 if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
505 GOTO(unlock, rc = 0);
507 lp->ldo_def_stripenr = v1->lmm_stripe_count;
508 lp->ldo_def_stripe_size = v1->lmm_stripe_size;
509 lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
510 lp->ldo_striping_cached = 1;
511 lp->ldo_def_striping_set = 1;
513 if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
514 /* XXX: sanity check here */
515 v3 = (struct lov_user_md_v3 *) v1;
516 if (v3->lmm_pool_name[0])
517 lod_object_set_pool(lp, v3->lmm_pool_name);
520 CDEBUG(D_OTHER, "def. striping: # %d, sz %d, off %d %s%s on "DFID"\n",
521 lp->ldo_def_stripenr, lp->ldo_def_stripe_size,
522 lp->ldo_def_stripe_offset, v3 ? "from " : "",
523 v3 ? lp->ldo_pool : "", PFID(lu_object_fid(&lp->ldo_obj.do_lu)));
527 dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
532 * used to transfer default striping data to the object being created
534 static void lod_ah_init(const struct lu_env *env,
535 struct dt_allocation_hint *ah,
536 struct dt_object *parent,
537 struct dt_object *child,
538 cfs_umode_t child_mode)
540 struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
541 struct dt_object *nextp = NULL;
542 struct dt_object *nextc;
543 struct lod_object *lp = NULL;
544 struct lod_object *lc;
545 struct lov_desc *desc;
550 if (likely(parent)) {
551 nextp = dt_object_child(parent);
552 lp = lod_dt_obj(parent);
555 nextc = dt_object_child(child);
556 lc = lod_dt_obj(child);
558 LASSERT(lc->ldo_stripenr == 0);
559 LASSERT(lc->ldo_stripe == NULL);
562 * local object may want some hints
563 * in case of late striping creation, ->ah_init()
564 * can be called with local object existing
566 if (!dt_object_exists(nextc))
567 nextc->do_ops->do_ah_init(env, ah, nextp, nextc, child_mode);
569 if (S_ISDIR(child_mode)) {
570 if (lp->ldo_striping_cached == 0) {
571 /* we haven't tried to get default striping for
572 * the directory yet, let's cache it in the object */
573 lod_cache_parent_striping(env, lp);
575 /* transfer defaults to new directory */
576 if (lp->ldo_striping_cached) {
578 lod_object_set_pool(lc, lp->ldo_pool);
579 lc->ldo_def_stripenr = lp->ldo_def_stripenr;
580 lc->ldo_def_stripe_size = lp->ldo_def_stripe_size;
581 lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
582 lc->ldo_striping_cached = 1;
583 lc->ldo_def_striping_set = 1;
584 CDEBUG(D_OTHER, "inherite striping defaults\n");
590 * if object is going to be striped over OSTs, transfer default
591 * striping information to the child, so that we can use it
592 * during declaration and creation
594 if (!lod_object_will_be_striped(S_ISREG(child_mode),
595 lu_object_fid(&child->do_lu)))
599 * try from the parent
601 if (likely(parent)) {
602 if (lp->ldo_striping_cached == 0) {
603 /* we haven't tried to get default striping for
604 * the directory yet, let's cache it in the object */
605 lod_cache_parent_striping(env, lp);
608 lc->ldo_def_stripe_offset = (__u16) -1;
610 if (lp->ldo_def_striping_set) {
612 lod_object_set_pool(lc, lp->ldo_pool);
613 lc->ldo_stripenr = lp->ldo_def_stripenr;
614 lc->ldo_stripe_size = lp->ldo_def_stripe_size;
615 lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
616 CDEBUG(D_OTHER, "striping from parent: #%d, sz %d %s\n",
617 lc->ldo_stripenr, lc->ldo_stripe_size,
618 lp->ldo_pool ? lp->ldo_pool : "");
623 * if the parent doesn't provide with specific pattern, grab fs-wide one
626 if (lc->ldo_stripenr == 0)
627 lc->ldo_stripenr = desc->ld_default_stripe_count;
628 if (lc->ldo_stripe_size == 0)
629 lc->ldo_stripe_size = desc->ld_default_stripe_size;
630 CDEBUG(D_OTHER, "final striping: # %d stripes, sz %d from %s\n",
631 lc->ldo_stripenr, lc->ldo_stripe_size,
632 lc->ldo_pool ? lc->ldo_pool : "");
638 * Create declaration of striped object
640 int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt,
641 struct lu_attr *attr,
642 const struct lu_buf *lovea, struct thandle *th)
644 struct lod_thread_info *info = lod_env_info(env);
645 struct dt_object *next = dt_object_child(dt);
646 struct lod_object *lo = lod_dt_obj(dt);
650 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) {
651 /* failed to create striping, let's reset
652 * config so that others don't get confused */
653 lod_object_free_striping(env, lo);
654 GOTO(out, rc = -ENOMEM);
657 /* XXX: there will be a call to QoS here */
661 * declare storage for striping data
663 info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
664 lo->ldo_pool ? LOV_MAGIC_V3 : LOV_MAGIC_V1);
665 rc = dt_declare_xattr_set(env, next, &info->lti_buf, XATTR_NAME_LOV,
674 static int lod_declare_object_create(const struct lu_env *env,
675 struct dt_object *dt,
676 struct lu_attr *attr,
677 struct dt_allocation_hint *hint,
678 struct dt_object_format *dof,
681 struct dt_object *next = dt_object_child(dt);
682 struct lod_object *lo = lod_dt_obj(dt);
689 LASSERT(!dt_object_exists(next));
692 * first of all, we declare creation of local object
694 rc = dt_declare_create(env, next, attr, hint, dof, th);
698 if (dof->dof_type == DFT_SYM)
699 dt->do_body_ops = &lod_body_lnk_ops;
702 * it's lod_ah_init() who has decided the object will striped
704 if (dof->dof_type == DFT_REGULAR) {
705 /* callers don't want stripes */
706 /* XXX: all tricky interactions with ->ah_make_hint() decided
707 * to use striping, then ->declare_create() behaving differently
708 * should be cleaned */
709 if (dof->u.dof_reg.striped == 0)
710 lo->ldo_stripenr = 0;
711 if (lo->ldo_stripenr > 0)
712 rc = lod_declare_striped_object(env, dt, attr,
714 } else if (dof->dof_type == DFT_DIR && lo->ldo_striping_cached) {
715 struct lod_thread_info *info = lod_env_info(env);
717 info->lti_buf.lb_buf = NULL;
718 info->lti_buf.lb_len = sizeof(struct lov_user_md_v3);
719 /* to transfer default striping from the parent */
720 rc = dt_declare_xattr_set(env, next, &info->lti_buf,
721 XATTR_NAME_LOV, 0, th);
728 int lod_striping_create(const struct lu_env *env, struct dt_object *dt,
729 struct lu_attr *attr, struct dt_object_format *dof,
732 struct lod_object *lo = lod_dt_obj(dt);
736 LASSERT(lo->ldo_stripe);
737 LASSERT(lo->ldo_stripe > 0);
738 LASSERT(lo->ldo_striping_cached == 0);
740 /* create all underlying objects */
741 for (i = 0; i < lo->ldo_stripenr; i++) {
742 LASSERT(lo->ldo_stripe[i]);
743 rc = dt_create(env, lo->ldo_stripe[i], attr, NULL, dof, th);
749 rc = lod_generate_and_set_lovea(env, lo, th);
754 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
755 struct lu_attr *attr,
756 struct dt_allocation_hint *hint,
757 struct dt_object_format *dof, struct thandle *th)
759 struct dt_object *next = dt_object_child(dt);
760 struct lod_object *lo = lod_dt_obj(dt);
764 /* create local object */
765 rc = dt_create(env, next, attr, hint, dof, th);
768 if (S_ISDIR(dt->do_lu.lo_header->loh_attr))
769 rc = lod_store_def_striping(env, dt, th);
770 else if (lo->ldo_stripe)
771 rc = lod_striping_create(env, dt, attr, dof, th);
777 static int lod_declare_object_destroy(const struct lu_env *env,
778 struct dt_object *dt,
781 struct dt_object *next = dt_object_child(dt);
782 struct lod_object *lo = lod_dt_obj(dt);
787 * we declare destroy for the local object
789 rc = dt_declare_destroy(env, next, th);
794 * load striping information, notice we don't do this when object
795 * is being initialized as we don't need this information till
796 * few specific cases like destroy, chown
798 rc = lod_load_striping(env, lo);
802 /* declare destroy for all underlying objects */
803 for (i = 0; i < lo->ldo_stripenr; i++) {
804 LASSERT(lo->ldo_stripe[i]);
805 rc = dt_declare_destroy(env, lo->ldo_stripe[i], th);
814 static int lod_object_destroy(const struct lu_env *env,
815 struct dt_object *dt, struct thandle *th)
817 struct dt_object *next = dt_object_child(dt);
818 struct lod_object *lo = lod_dt_obj(dt);
822 /* destroy local object */
823 rc = dt_destroy(env, next, th);
827 /* destroy all underlying objects */
828 for (i = 0; i < lo->ldo_stripenr; i++) {
829 LASSERT(lo->ldo_stripe[i]);
830 rc = dt_destroy(env, lo->ldo_stripe[i], th);
838 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
839 const struct dt_index_features *feat)
841 struct dt_object *next = dt_object_child(dt);
845 LASSERT(next->do_ops);
846 LASSERT(next->do_ops->do_index_try);
848 rc = next->do_ops->do_index_try(env, next, feat);
849 if (next->do_index_ops && dt->do_index_ops == NULL) {
850 dt->do_index_ops = &lod_index_ops;
851 /* XXX: iterators don't accept device, so bypass LOD */
852 /* will be fixed with DNE */
853 if (lod_index_ops.dio_it.fini == NULL) {
854 lod_index_ops.dio_it = next->do_index_ops->dio_it;
855 lod_index_ops.dio_it.init = lod_it_init;
862 static int lod_declare_ref_add(const struct lu_env *env,
863 struct dt_object *dt, struct thandle *th)
865 return dt_declare_ref_add(env, dt_object_child(dt), th);
868 static int lod_ref_add(const struct lu_env *env,
869 struct dt_object *dt, struct thandle *th)
871 return dt_ref_add(env, dt_object_child(dt), th);
874 static int lod_declare_ref_del(const struct lu_env *env,
875 struct dt_object *dt, struct thandle *th)
877 return dt_declare_ref_del(env, dt_object_child(dt), th);
880 static int lod_ref_del(const struct lu_env *env,
881 struct dt_object *dt, struct thandle *th)
883 return dt_ref_del(env, dt_object_child(dt), th);
886 static struct obd_capa *lod_capa_get(const struct lu_env *env,
887 struct dt_object *dt,
888 struct lustre_capa *old, __u64 opc)
890 return dt_capa_get(env, dt_object_child(dt), old, opc);
893 static int lod_object_sync(const struct lu_env *env, struct dt_object *dt)
895 return dt_object_sync(env, dt_object_child(dt));
898 struct dt_object_operations lod_obj_ops = {
899 .do_read_lock = lod_object_read_lock,
900 .do_write_lock = lod_object_write_lock,
901 .do_read_unlock = lod_object_read_unlock,
902 .do_write_unlock = lod_object_write_unlock,
903 .do_write_locked = lod_object_write_locked,
904 .do_attr_get = lod_attr_get,
905 .do_declare_attr_set = lod_declare_attr_set,
906 .do_attr_set = lod_attr_set,
907 .do_xattr_get = lod_xattr_get,
908 .do_declare_xattr_set = lod_declare_xattr_set,
909 .do_xattr_set = lod_xattr_set,
910 .do_declare_xattr_del = lod_declare_xattr_del,
911 .do_xattr_del = lod_xattr_del,
912 .do_xattr_list = lod_xattr_list,
913 .do_ah_init = lod_ah_init,
914 .do_declare_create = lod_declare_object_create,
915 .do_create = lod_object_create,
916 .do_declare_destroy = lod_declare_object_destroy,
917 .do_destroy = lod_object_destroy,
918 .do_index_try = lod_index_try,
919 .do_declare_ref_add = lod_declare_ref_add,
920 .do_ref_add = lod_ref_add,
921 .do_declare_ref_del = lod_declare_ref_del,
922 .do_ref_del = lod_ref_del,
923 .do_capa_get = lod_capa_get,
924 .do_object_sync = lod_object_sync,
927 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
928 struct lu_buf *buf, loff_t *pos,
929 struct lustre_capa *capa)
931 struct dt_object *next = dt_object_child(dt);
932 return next->do_body_ops->dbo_read(env, next, buf, pos, capa);
935 static ssize_t lod_declare_write(const struct lu_env *env,
936 struct dt_object *dt,
937 const loff_t size, loff_t pos,
940 return dt_declare_record_write(env, dt_object_child(dt),
944 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,
945 const struct lu_buf *buf, loff_t *pos,
946 struct thandle *th, struct lustre_capa *capa, int iq)
948 struct dt_object *next = dt_object_child(dt);
950 return next->do_body_ops->dbo_write(env, next, buf, pos, th, capa, iq);
953 static const struct dt_body_operations lod_body_lnk_ops = {
954 .dbo_read = lod_read,
955 .dbo_declare_write = lod_declare_write,
956 .dbo_write = lod_write
959 static int lod_object_init(const struct lu_env *env, struct lu_object *o,
960 const struct lu_object_conf *conf)
962 struct lod_device *d = lu2lod_dev(o->lo_dev);
963 struct lu_object *below;
964 struct lu_device *under;
968 * create local object
970 under = &d->lod_child->dd_lu_dev;
971 below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
975 lu_object_add(o, below);
980 void lod_object_free_striping(const struct lu_env *env, struct lod_object *lo)
984 if (lo->ldo_stripe) {
985 LASSERT(lo->ldo_stripes_allocated > 0);
987 for (i = 0; i < lo->ldo_stripenr; i++) {
988 if (lo->ldo_stripe[i])
989 lu_object_put(env, &lo->ldo_stripe[i]->do_lu);
992 i = sizeof(struct dt_object *) * lo->ldo_stripes_allocated;
993 OBD_FREE(lo->ldo_stripe, i);
994 lo->ldo_stripe = NULL;
995 lo->ldo_stripes_allocated = 0;
997 lo->ldo_stripenr = 0;
1001 * ->start is called once all slices are initialized, including header's
1002 * cache for mode (object type). using the type we can initialize ops
1004 static int lod_object_start(const struct lu_env *env, struct lu_object *o)
1006 if (S_ISLNK(o->lo_header->loh_attr & S_IFMT))
1007 lu2lod_obj(o)->ldo_obj.do_body_ops = &lod_body_lnk_ops;
1011 static void lod_object_free(const struct lu_env *env, struct lu_object *o)
1013 struct lod_object *mo = lu2lod_obj(o);
1016 * release all underlying object pinned
1019 lod_object_free_striping(env, mo);
1021 lod_object_set_pool(mo, NULL);
1024 OBD_SLAB_FREE_PTR(mo, lod_object_kmem);
1027 static void lod_object_release(const struct lu_env *env, struct lu_object *o)
1029 /* XXX: shouldn't we release everything here in case if object
1030 * creation failed before? */
1033 static int lod_object_print(const struct lu_env *env, void *cookie,
1034 lu_printer_t p, const struct lu_object *l)
1036 struct lod_object *o = lu2lod_obj((struct lu_object *) l);
1038 return (*p)(env, cookie, LUSTRE_LOD_NAME"-object@%p", o);
1041 struct lu_object_operations lod_lu_obj_ops = {
1042 .loo_object_init = lod_object_init,
1043 .loo_object_start = lod_object_start,
1044 .loo_object_free = lod_object_free,
1045 .loo_object_release = lod_object_release,
1046 .loo_object_print = lod_object_print,