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) 2012, 2013, Intel Corporation.
29 * lustre/lod/lod_object.c
31 * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
34 #define DEBUG_SUBSYSTEM S_MDS
37 #include <obd_class.h>
38 #include <lustre_ver.h>
39 #include <obd_support.h>
40 #include <lprocfs_status.h>
42 #include <lustre_fid.h>
43 #include <lustre_param.h>
44 #include <lustre_fid.h>
45 #include <lustre_lmv.h>
47 #include <md_object.h>
49 #include "lod_internal.h"
51 static const char dot[] = ".";
52 static const char dotdot[] = "..";
54 extern struct kmem_cache *lod_object_kmem;
55 static const struct dt_body_operations lod_body_lnk_ops;
57 static int lod_index_lookup(const struct lu_env *env, struct dt_object *dt,
58 struct dt_rec *rec, const struct dt_key *key,
59 struct lustre_capa *capa)
61 struct dt_object *next = dt_object_child(dt);
62 return next->do_index_ops->dio_lookup(env, next, rec, key, capa);
65 static int lod_declare_index_insert(const struct lu_env *env,
67 const struct dt_rec *rec,
68 const struct dt_key *key,
69 struct thandle *handle)
71 return dt_declare_insert(env, dt_object_child(dt), rec, key, handle);
74 static int lod_index_insert(const struct lu_env *env,
76 const struct dt_rec *rec,
77 const struct dt_key *key,
79 struct lustre_capa *capa,
82 return dt_insert(env, dt_object_child(dt), rec, key, th, capa, ign);
85 static int lod_declare_index_delete(const struct lu_env *env,
87 const struct dt_key *key,
90 return dt_declare_delete(env, dt_object_child(dt), key, th);
93 static int lod_index_delete(const struct lu_env *env,
95 const struct dt_key *key,
97 struct lustre_capa *capa)
99 return dt_delete(env, dt_object_child(dt), key, th, capa);
102 static struct dt_it *lod_it_init(const struct lu_env *env,
103 struct dt_object *dt, __u32 attr,
104 struct lustre_capa *capa)
106 struct dt_object *next = dt_object_child(dt);
107 struct lod_it *it = &lod_env_info(env)->lti_it;
108 struct dt_it *it_next;
111 it_next = next->do_index_ops->dio_it.init(env, next, attr, capa);
115 /* currently we do not use more than one iterator per thread
116 * so we store it in thread info. if at some point we need
117 * more active iterators in a single thread, we can allocate
119 LASSERT(it->lit_obj == NULL);
121 it->lit_it = it_next;
124 return (struct dt_it *)it;
127 #define LOD_CHECK_IT(env, 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);
296 /* osp_declare_attr_set() ignores all attributes other than
297 * UID, GID, and size, and osp_attr_set() ignores all but UID
298 * and GID. Declaration of size attr setting happens through
299 * lod_declare_init_size(), and not through this function.
300 * Therefore we need not load striping unless ownership is
301 * changing. This should save memory and (we hope) speed up
303 if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
304 if (!(attr->la_valid & (LA_UID | LA_GID)))
307 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
310 if (!(attr->la_valid & (LA_UID | LA_GID | LA_MODE |
311 LA_ATIME | LA_MTIME | LA_CTIME)))
315 * load striping information, notice we don't do this when object
316 * is being initialized as we don't need this information till
317 * few specific cases like destroy, chown
319 rc = lod_load_striping(env, lo);
323 if (lo->ldo_stripenr == 0)
326 if (!(attr->la_valid & ~(LA_ATIME | LA_MTIME | LA_CTIME))) {
327 struct lu_attr *la = &lod_env_info(env)->lti_attr;
328 bool setattr_time = false;
330 rc = dt_attr_get(env, dt_object_child(dt), la,
335 /* If it will only setattr time, it will only set
336 * time < current_time */
337 if ((attr->la_valid & LA_ATIME &&
338 attr->la_atime < la->la_atime) ||
339 (attr->la_valid & LA_CTIME &&
340 attr->la_ctime < la->la_ctime) ||
341 (attr->la_valid & LA_MTIME &&
342 attr->la_mtime < la->la_mtime))
349 * if object is striped declare changes on the stripes
351 LASSERT(lo->ldo_stripe);
352 for (i = 0; i < lo->ldo_stripenr; i++) {
353 LASSERT(lo->ldo_stripe[i]);
355 rc = dt_declare_attr_set(env, lo->ldo_stripe[i], attr, handle);
357 CERROR("failed declaration: %d\n", rc);
362 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE) &&
363 dt_object_exists(next) != 0 &&
364 dt_object_remote(next) == 0)
365 dt_declare_xattr_del(env, next, XATTR_NAME_LOV, handle);
367 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE) &&
368 dt_object_exists(next) &&
369 dt_object_remote(next) == 0 && S_ISREG(attr->la_mode)) {
370 struct lod_thread_info *info = lod_env_info(env);
371 struct lu_buf *buf = &info->lti_buf;
373 buf->lb_buf = info->lti_ea_store;
374 buf->lb_len = info->lti_ea_store_size;
375 dt_declare_xattr_set(env, next, buf, XATTR_NAME_LOV,
376 LU_XATTR_REPLACE, handle);
382 static int lod_attr_set(const struct lu_env *env,
383 struct dt_object *dt,
384 const struct lu_attr *attr,
385 struct thandle *handle,
386 struct lustre_capa *capa)
388 struct dt_object *next = dt_object_child(dt);
389 struct lod_object *lo = lod_dt_obj(dt);
394 * apply changes to the local object
396 rc = dt_attr_set(env, next, attr, handle, capa);
400 if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
401 if (!(attr->la_valid & (LA_UID | LA_GID)))
404 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
407 if (!(attr->la_valid & (LA_UID | LA_GID | LA_MODE |
408 LA_ATIME | LA_MTIME | LA_CTIME)))
412 if (lo->ldo_stripenr == 0)
415 if (!(attr->la_valid & ~(LA_ATIME | LA_MTIME | LA_CTIME))) {
416 struct lu_attr *la = &lod_env_info(env)->lti_attr;
417 bool setattr_time = false;
419 rc = dt_attr_get(env, dt_object_child(dt), la,
424 /* If it will only setattr time, it will only set
425 * time < current_time */
426 if ((attr->la_valid & LA_ATIME &&
427 attr->la_atime < la->la_atime) ||
428 (attr->la_valid & LA_CTIME &&
429 attr->la_ctime < la->la_ctime) ||
430 (attr->la_valid & LA_MTIME &&
431 attr->la_mtime < la->la_mtime))
439 * if object is striped, apply changes to all the stripes
441 LASSERT(lo->ldo_stripe);
442 for (i = 0; i < lo->ldo_stripenr; i++) {
443 LASSERT(lo->ldo_stripe[i]);
444 rc = dt_attr_set(env, lo->ldo_stripe[i], attr, handle, capa);
446 CERROR("failed declaration: %d\n", rc);
451 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE) &&
452 dt_object_exists(next) != 0 &&
453 dt_object_remote(next) == 0)
454 dt_xattr_del(env, next, XATTR_NAME_LOV, handle, BYPASS_CAPA);
456 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE) &&
457 dt_object_exists(next) &&
458 dt_object_remote(next) == 0 && S_ISREG(attr->la_mode)) {
459 struct lod_thread_info *info = lod_env_info(env);
460 struct lu_buf *buf = &info->lti_buf;
461 struct ost_id *oi = &info->lti_ostid;
462 struct lu_fid *fid = &info->lti_fid;
463 struct lov_mds_md_v1 *lmm;
464 struct lov_ost_data_v1 *objs;
468 rc1 = lod_get_lov_ea(env, lo);
472 buf->lb_buf = info->lti_ea_store;
473 buf->lb_len = info->lti_ea_store_size;
474 lmm = info->lti_ea_store;
475 magic = le32_to_cpu(lmm->lmm_magic);
476 if (magic == LOV_MAGIC_V1)
477 objs = &(lmm->lmm_objects[0]);
479 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
480 ostid_le_to_cpu(&objs->l_ost_oi, oi);
481 ostid_to_fid(fid, oi, le32_to_cpu(objs->l_ost_idx));
483 fid_to_ostid(fid, oi);
484 ostid_cpu_to_le(oi, &objs->l_ost_oi);
485 dt_xattr_set(env, next, buf, XATTR_NAME_LOV,
486 LU_XATTR_REPLACE, handle, BYPASS_CAPA);
492 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
493 struct lu_buf *buf, const char *name,
494 struct lustre_capa *capa)
496 struct lod_thread_info *info = lod_env_info(env);
497 struct lod_device *dev = lu2lod_dev(dt->do_lu.lo_dev);
501 rc = dt_xattr_get(env, dt_object_child(dt), buf, name, capa);
502 if (rc != -ENODATA || !S_ISDIR(dt->do_lu.lo_header->loh_attr & S_IFMT))
506 * lod returns default striping on the real root of the device
507 * this is like the root stores default striping for the whole
508 * filesystem. historically we've been using a different approach
509 * and store it in the config.
511 dt_root_get(env, dev->lod_child, &info->lti_fid);
512 is_root = lu_fid_eq(&info->lti_fid, lu_object_fid(&dt->do_lu));
514 if (is_root && strcmp(XATTR_NAME_LOV, name) == 0) {
515 struct lov_user_md *lum = buf->lb_buf;
516 struct lov_desc *desc = &dev->lod_desc;
518 if (buf->lb_buf == NULL) {
520 } else if (buf->lb_len >= sizeof(*lum)) {
521 lum->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V1);
522 lmm_oi_set_seq(&lum->lmm_oi, FID_SEQ_LOV_DEFAULT);
523 lmm_oi_set_id(&lum->lmm_oi, 0);
524 lmm_oi_cpu_to_le(&lum->lmm_oi, &lum->lmm_oi);
525 lum->lmm_pattern = cpu_to_le32(desc->ld_pattern);
526 lum->lmm_stripe_size = cpu_to_le32(
527 desc->ld_default_stripe_size);
528 lum->lmm_stripe_count = cpu_to_le16(
529 desc->ld_default_stripe_count);
530 lum->lmm_stripe_offset = cpu_to_le16(
531 desc->ld_default_stripe_offset);
541 static int lod_verify_md_striping(struct lod_device *lod,
542 const struct lmv_user_md_v1 *lum)
547 if (unlikely(le32_to_cpu(lum->lum_magic) != LMV_USER_MAGIC))
548 GOTO(out, rc = -EINVAL);
550 if (unlikely(le32_to_cpu(lum->lum_stripe_count) == 0))
551 GOTO(out, rc = -EINVAL);
553 if (unlikely(le32_to_cpu(lum->lum_stripe_count) >
554 lod->lod_remote_mdt_count + 1))
555 GOTO(out, rc = -EINVAL);
558 CERROR("%s: invalid lmv_user_md: magic = %x, "
559 "stripe_offset = %d, stripe_count = %u: rc = %d\n",
560 lod2obd(lod)->obd_name, le32_to_cpu(lum->lum_magic),
561 (int)le32_to_cpu(lum->lum_stripe_offset),
562 le32_to_cpu(lum->lum_stripe_count), rc);
566 int lod_prep_lmv_md(const struct lu_env *env, struct dt_object *dt,
567 struct lu_buf *lmv_buf)
569 struct lod_thread_info *info = lod_env_info(env);
570 struct lod_device *lod = lu2lod_dev(dt->do_lu.lo_dev);
571 struct lod_object *lo = lod_dt_obj(dt);
572 struct lmv_mds_md_v1 *lmm1;
575 int type = LU_SEQ_RANGE_ANY;
581 LASSERT(lo->ldo_dir_striped != 0);
582 LASSERT(lo->ldo_stripenr > 0);
583 stripe_count = lo->ldo_stripenr + 1;
584 lmm_size = lmv_mds_md_size(stripe_count, LMV_MAGIC);
585 if (info->lti_ea_store_size < lmm_size) {
586 rc = lod_ea_store_resize(info, lmm_size);
591 lmm1 = (struct lmv_mds_md_v1 *)info->lti_ea_store;
592 lmm1->lmv_magic = cpu_to_le32(LMV_MAGIC);
593 lmm1->lmv_stripe_count = cpu_to_le32(stripe_count);
594 lmm1->lmv_hash_type = cpu_to_le32(lo->ldo_dir_hash_type);
595 rc = lod_fld_lookup(env, lod, lu_object_fid(&dt->do_lu),
600 lmm1->lmv_master_mdt_index = cpu_to_le32(mdtidx);
601 fid_cpu_to_le(&lmm1->lmv_stripe_fids[0], lu_object_fid(&dt->do_lu));
602 for (i = 0; i < lo->ldo_stripenr; i++) {
603 struct dt_object *dto;
605 dto = lo->ldo_stripe[i];
606 LASSERT(dto != NULL);
607 fid_cpu_to_le(&lmm1->lmv_stripe_fids[i + 1],
608 lu_object_fid(&dto->do_lu));
611 lmv_buf->lb_buf = info->lti_ea_store;
612 lmv_buf->lb_len = lmm_size;
613 lo->ldo_dir_striping_cached = 1;
618 int lod_parse_dir_striping(const struct lu_env *env, struct lod_object *lo,
619 const struct lu_buf *buf)
621 struct lod_thread_info *info = lod_env_info(env);
622 struct lod_device *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
623 struct lod_tgt_descs *ltd = &lod->lod_mdt_descs;
624 struct dt_object **stripe;
625 union lmv_mds_md *lmm = buf->lb_buf;
626 struct lmv_mds_md_v1 *lmv1 = &lmm->lmv_md_v1;
627 struct lu_fid *fid = &info->lti_fid;
632 if (le32_to_cpu(lmv1->lmv_magic) != LMV_MAGIC_V1)
635 if (le32_to_cpu(lmv1->lmv_stripe_count) <= 1)
638 fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[0]);
639 /* Do not load striping information for slave inode */
640 if (!lu_fid_eq(fid, lu_object_fid(&lo->ldo_obj.do_lu))) {
641 lo->ldo_dir_slave_stripe = 1;
645 LASSERT(lo->ldo_stripe == NULL);
646 OBD_ALLOC(stripe, sizeof(stripe[0]) *
647 (le32_to_cpu(lmv1->lmv_stripe_count) - 1));
651 /* skip master stripe */
652 for (i = 1; i < le32_to_cpu(lmv1->lmv_stripe_count); i++) {
653 struct lod_tgt_desc *tgt;
655 int type = LU_SEQ_RANGE_ANY;
656 struct dt_object *dto;
658 fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[i]);
659 rc = lod_fld_lookup(env, lod, fid, &idx, &type);
663 tgt = LTD_TGT(ltd, idx);
665 GOTO(out, rc = -ESTALE);
667 dto = dt_locate_at(env, tgt->ltd_tgt, fid,
668 lo->ldo_obj.do_lu.lo_dev->ld_site->ls_top_dev,
671 GOTO(out, rc = PTR_ERR(dto));
676 lo->ldo_stripe = stripe;
677 lo->ldo_stripenr = le32_to_cpu(lmv1->lmv_stripe_count) - 1;
678 lo->ldo_stripes_allocated = le32_to_cpu(lmv1->lmv_stripe_count) - 1;
680 lod_object_free_striping(env, lo);
685 static int lod_prep_md_striped_create(const struct lu_env *env,
686 struct dt_object *dt,
687 struct lu_attr *attr,
688 const struct lmv_user_md_v1 *lum,
691 struct lod_device *lod = lu2lod_dev(dt->do_lu.lo_dev);
692 struct lod_tgt_descs *ltd = &lod->lod_mdt_descs;
693 struct lod_object *lo = lod_dt_obj(dt);
694 struct dt_object **stripe;
695 struct lu_buf lmv_buf;
703 /* The lum has been verifed in lod_verify_md_striping */
704 LASSERT(le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC);
705 LASSERT(le32_to_cpu(lum->lum_stripe_count) > 0);
707 /* Do not need allocated master stripe */
708 stripe_count = le32_to_cpu(lum->lum_stripe_count);
709 OBD_ALLOC(stripe, sizeof(stripe[0]) * (stripe_count - 1));
713 OBD_ALLOC(idx_array, sizeof(idx_array[0]) * stripe_count);
714 if (idx_array == NULL)
715 GOTO(out_free, rc = -ENOMEM);
717 idx_array[0] = le32_to_cpu(lum->lum_stripe_offset);
718 for (i = 1; i < stripe_count; i++) {
719 struct lod_tgt_desc *tgt;
720 struct dt_object *dto;
723 struct lu_object_conf conf = { 0 };
725 idx = (idx_array[i - 1] + 1) % (lod->lod_remote_mdt_count + 1);
727 for (j = 0; j < lod->lod_remote_mdt_count;
728 j++, idx = (idx + 1) % (lod->lod_remote_mdt_count + 1)) {
729 bool already_allocated = false;
732 CDEBUG(D_INFO, "try idx %d, mdt cnt %d,"
733 " allocated %d, last allocated %d\n", idx,
734 lod->lod_remote_mdt_count, i, idx_array[i - 1]);
736 /* Find next avaible target */
737 if (!cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx))
740 /* check whether the idx already exists
741 * in current allocated array */
742 for (k = 0; k < i; k++) {
743 if (idx_array[k] == idx) {
744 already_allocated = true;
749 if (already_allocated)
755 /* Can not allocate more stripes */
756 if (j == lod->lod_remote_mdt_count) {
757 CDEBUG(D_INFO, "%s: require stripes %d only get %d\n",
758 lod2obd(lod)->obd_name, stripe_count, i - 1);
762 CDEBUG(D_INFO, "idx %d, mdt cnt %d,"
763 " allocated %d, last allocated %d\n", idx,
764 lod->lod_remote_mdt_count, i, idx_array[i - 1]);
766 tgt = LTD_TGT(ltd, idx);
767 LASSERT(tgt != NULL);
769 rc = obd_fid_alloc(tgt->ltd_exp, &fid, NULL);
774 conf.loc_flags = LOC_F_NEW;
775 dto = dt_locate_at(env, tgt->ltd_tgt, &fid,
776 dt->do_lu.lo_dev->ld_site->ls_top_dev, &conf);
778 GOTO(out_put, rc = PTR_ERR(dto));
783 lo->ldo_dir_striped = 1;
784 lo->ldo_stripe = stripe;
785 lo->ldo_stripenr = i - 1;
786 lo->ldo_stripes_allocated = stripe_count - 1;
788 if (lo->ldo_stripenr == 0)
789 GOTO(out_put, rc = -ENOSPC);
791 rc = lod_prep_lmv_md(env, dt, &lmv_buf);
795 for (i = 0; i < lo->ldo_stripenr; i++) {
796 struct dt_object *dto;
799 /* only create slave striped object */
800 rc = dt_declare_create(env, dto, attr, NULL, NULL, th);
804 if (!dt_try_as_dir(env, dto))
805 GOTO(out_put, rc = -EINVAL);
807 rc = dt_declare_insert(env, dto,
808 (const struct dt_rec *)lu_object_fid(&dto->do_lu),
809 (const struct dt_key *)dot, th);
813 /* master stripe FID will be put to .. */
814 rc = dt_declare_insert(env, dto,
815 (const struct dt_rec *)lu_object_fid(&dt->do_lu),
816 (const struct dt_key *)dotdot, th);
820 /* probably nothing to inherite */
821 if (lo->ldo_striping_cached &&
822 !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
823 lo->ldo_def_stripenr,
824 lo->ldo_def_stripe_offset)) {
825 struct lod_thread_info *info;
826 struct lov_user_md_v3 *v3;
828 /* sigh, lti_ea_store has been used for lmv_buf,
829 * so we have to allocate buffer for default
833 GOTO(out_put, rc = -ENOMEM);
835 memset(v3, 0, sizeof(*v3));
836 v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
837 v3->lmm_stripe_count =
838 cpu_to_le32(lo->ldo_def_stripenr);
839 v3->lmm_stripe_offset =
840 cpu_to_le32(lo->ldo_def_stripe_offset);
841 v3->lmm_stripe_size =
842 cpu_to_le32(lo->ldo_def_stripe_size);
844 strncpy(v3->lmm_pool_name, lo->ldo_pool,
847 info = lod_env_info(env);
848 info->lti_buf.lb_buf = v3;
849 info->lti_buf.lb_len = sizeof(*v3);
850 rc = dt_declare_xattr_set(env, dto,
858 rc = dt_declare_xattr_set(env, dto, &lmv_buf, XATTR_NAME_LMV, 0,
864 rc = dt_declare_xattr_set(env, dt, &lmv_buf, XATTR_NAME_LMV, 0, th);
870 for (i = 0; i < stripe_count - 1; i++)
871 if (stripe[i] != NULL)
872 lu_object_put(env, &stripe[i]->do_lu);
873 OBD_FREE(stripe, sizeof(stripe[0]) * (stripe_count - 1));
877 if (idx_array != NULL)
878 OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count);
884 * Declare create striped md object.
886 static int lod_declare_xattr_set_lmv(const struct lu_env *env,
887 struct dt_object *dt,
888 struct lu_attr *attr,
889 const struct lu_buf *lum_buf,
892 struct lod_object *lo = lod_dt_obj(dt);
893 struct lod_device *lod = lu2lod_dev(dt->do_lu.lo_dev);
894 struct lmv_user_md_v1 *lum;
898 lum = lum_buf->lb_buf;
899 LASSERT(lum != NULL);
901 CDEBUG(D_INFO, "lum magic = %x count = %u offset = %d\n",
902 le32_to_cpu(lum->lum_magic), le32_to_cpu(lum->lum_stripe_count),
903 (int)le32_to_cpu(lum->lum_stripe_offset));
905 if (le32_to_cpu(lum->lum_stripe_count) <= 1)
908 rc = lod_verify_md_striping(lod, lum);
912 /* prepare dir striped objects */
913 rc = lod_prep_md_striped_create(env, dt, attr, lum, th);
915 /* failed to create striping, let's reset
916 * config so that others don't get confused */
917 lod_object_free_striping(env, lo);
925 * LOV xattr is a storage for striping, and LOD owns this xattr.
926 * but LOD allows others to control striping to some extent
928 * - to set new defined striping
929 * - to set new semi-defined striping
930 * - number of stripes is defined
931 * - number of stripes + osts are defined
934 static int lod_declare_xattr_set(const struct lu_env *env,
935 struct dt_object *dt,
936 const struct lu_buf *buf,
937 const char *name, int fl,
940 struct dt_object *next = dt_object_child(dt);
941 struct lu_attr *attr = &lod_env_info(env)->lti_attr;
947 * allow to declare predefined striping on a new (!mode) object
948 * which is supposed to be replay of regular file creation
949 * (when LOV setting is declared)
950 * LU_XATTR_REPLACE is set to indicate a layout swap
952 mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
953 if ((S_ISREG(mode) || mode == 0) && strcmp(name, XATTR_NAME_LOV) == 0 &&
954 !(fl & LU_XATTR_REPLACE)) {
956 * this is a request to manipulate object's striping
958 if (dt_object_exists(dt)) {
959 rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
963 memset(attr, 0, sizeof(*attr));
964 attr->la_valid = LA_TYPE | LA_MODE;
965 attr->la_mode = S_IFREG;
967 rc = lod_declare_striped_object(env, dt, attr, buf, th);
968 } else if (S_ISDIR(mode)) {
969 struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
970 struct lod_object *lo = lod_dt_obj(dt);
973 if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
974 struct lmv_user_md_v1 *lum;
976 LASSERT(buf != NULL && buf->lb_buf != NULL);
978 rc = lod_verify_md_striping(d, lum);
983 rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
987 /* set xattr to each stripes, if needed */
988 rc = lod_load_striping(env, lo);
992 if (lo->ldo_stripenr == 0)
995 for (i = 0; i < lo->ldo_stripenr; i++) {
996 LASSERT(lo->ldo_stripe[i]);
997 rc = dt_declare_xattr_set(env, lo->ldo_stripe[i], buf,
1003 rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
1009 static void lod_lov_stripe_cache_clear(struct lod_object *lo)
1011 lo->ldo_striping_cached = 0;
1012 lo->ldo_def_striping_set = 0;
1013 lod_object_set_pool(lo, NULL);
1014 lo->ldo_def_stripe_size = 0;
1015 lo->ldo_def_stripenr = 0;
1018 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
1019 struct dt_object *dt,
1020 const struct lu_buf *buf,
1021 const char *name, int fl,
1023 struct lustre_capa *capa)
1025 struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
1026 struct dt_object *next = dt_object_child(dt);
1027 struct lod_object *l = lod_dt_obj(dt);
1028 struct lov_user_md_v1 *lum;
1029 struct lov_user_md_v3 *v3 = NULL;
1033 /* If it is striped dir, we should clear the stripe cache for
1034 * slave stripe as well, but there are no effective way to
1035 * notify the LOD on the slave MDT, so we do not cache stripe
1036 * information for slave stripe for now. XXX*/
1037 lod_lov_stripe_cache_clear(l);
1038 LASSERT(buf != NULL && buf->lb_buf != NULL);
1041 rc = lod_verify_striping(d, buf, 0);
1045 if (lum->lmm_magic == LOV_USER_MAGIC_V3)
1048 /* if { size, offset, count } = { 0, -1, 0 } and no pool
1049 * (i.e. all default values specified) then delete default
1050 * striping from dir. */
1052 "set default striping: sz %u # %u offset %d %s %s\n",
1053 (unsigned)lum->lmm_stripe_size,
1054 (unsigned)lum->lmm_stripe_count,
1055 (int)lum->lmm_stripe_offset,
1056 v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
1058 if (LOVEA_DELETE_VALUES((lum->lmm_stripe_size),
1059 (lum->lmm_stripe_count),
1060 (lum->lmm_stripe_offset)) &&
1061 lum->lmm_magic == LOV_USER_MAGIC_V1) {
1062 rc = dt_xattr_del(env, next, name, th, capa);
1066 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
1072 static int lod_xattr_set_default_lmv_on_dir(const struct lu_env *env,
1073 struct dt_object *dt,
1074 const struct lu_buf *buf,
1075 const char *name, int fl,
1077 struct lustre_capa *capa)
1079 struct dt_object *next = dt_object_child(dt);
1080 struct lod_object *l = lod_dt_obj(dt);
1081 struct lmv_user_md_v1 *lum;
1085 LASSERT(buf != NULL && buf->lb_buf != NULL);
1088 CDEBUG(D_OTHER, "set default stripe_count # %u stripe_offset %d\n",
1089 le32_to_cpu(lum->lum_stripe_count),
1090 (int)le32_to_cpu(lum->lum_stripe_offset));
1092 if (LMVEA_DELETE_VALUES((le32_to_cpu(lum->lum_stripe_count)),
1093 le32_to_cpu(lum->lum_stripe_offset)) &&
1094 le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC) {
1095 rc = dt_xattr_del(env, next, name, th, capa);
1099 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
1103 /* Update default stripe cache */
1104 if (l->ldo_dir_stripe == NULL) {
1105 OBD_ALLOC_PTR(l->ldo_dir_stripe);
1106 if (l->ldo_dir_stripe == NULL)
1110 l->ldo_dir_striping_cached = 0;
1111 l->ldo_dir_def_striping_set = 1;
1112 l->ldo_dir_def_stripenr =
1113 le32_to_cpu(lum->lum_stripe_count) - 1;
1119 static int lod_xattr_set_lmv(const struct lu_env *env, struct dt_object *dt,
1120 const struct lu_buf *buf, const char *name,
1121 int fl, struct thandle *th,
1122 struct lustre_capa *capa)
1124 struct lod_object *lo = lod_dt_obj(dt);
1125 struct lu_buf lmv_buf;
1130 if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
1133 /* The stripes are supposed to be allocated in declare phase,
1134 * if there are no stripes being allocated, it will skip */
1135 if (lo->ldo_stripenr == 0)
1138 rc = lod_prep_lmv_md(env, dt, &lmv_buf);
1142 for (i = 0; i < lo->ldo_stripenr; i++) {
1143 struct dt_object *dto;
1144 struct lu_attr *attr = &lod_env_info(env)->lti_attr;
1146 dto = lo->ldo_stripe[i];
1147 memset(attr, 0, sizeof(*attr));
1148 attr->la_valid = LA_TYPE | LA_MODE;
1149 attr->la_mode = S_IFDIR;
1150 rc = dt_create(env, dto, attr, NULL, NULL, th);
1154 rc = dt_insert(env, dto,
1155 (const struct dt_rec *)lu_object_fid(&dto->do_lu),
1156 (const struct dt_key *)dot, th, capa, 0);
1160 rc = dt_insert(env, dto,
1161 (struct dt_rec *)lu_object_fid(&dt->do_lu),
1162 (const struct dt_key *)dotdot, th, capa, 0);
1166 if (lo->ldo_striping_cached &&
1167 !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
1168 lo->ldo_def_stripenr,
1169 lo->ldo_def_stripe_offset)) {
1170 struct lod_thread_info *info;
1171 struct lov_user_md_v3 *v3;
1173 /* sigh, lti_ea_store has been used for lmv_buf,
1174 * so we have to allocate buffer for default
1180 memset(v3, 0, sizeof(*v3));
1181 v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
1182 v3->lmm_stripe_count =
1183 cpu_to_le32(lo->ldo_def_stripenr);
1184 v3->lmm_stripe_offset =
1185 cpu_to_le32(lo->ldo_def_stripe_offset);
1186 v3->lmm_stripe_size =
1187 cpu_to_le32(lo->ldo_def_stripe_size);
1189 strncpy(v3->lmm_pool_name, lo->ldo_pool,
1192 info = lod_env_info(env);
1193 info->lti_buf.lb_buf = v3;
1194 info->lti_buf.lb_len = sizeof(*v3);
1195 rc = dt_xattr_set(env, dto, &info->lti_buf,
1196 XATTR_NAME_LOV, 0, th, capa);
1202 rc = dt_xattr_set(env, dto, &lmv_buf, XATTR_NAME_LMV, fl, th,
1206 rc = dt_xattr_set(env, dt, &lmv_buf, XATTR_NAME_LMV, fl, th, capa);
1211 static int lod_xattr_set(const struct lu_env *env,
1212 struct dt_object *dt, const struct lu_buf *buf,
1213 const char *name, int fl, struct thandle *th,
1214 struct lustre_capa *capa)
1216 struct lod_object *lo = lod_dt_obj(dt);
1217 struct dt_object *next = dt_object_child(dt);
1223 attr = dt->do_lu.lo_header->loh_attr & S_IFMT;
1224 if (S_ISDIR(attr) && strcmp(name, XATTR_NAME_LOV) == 0) {
1225 rc = lod_xattr_set_lov_on_dir(env, dt, buf, name, fl, th, capa);
1226 } else if (S_ISREG(attr) && !strcmp(name, XATTR_NAME_LOV)) {
1227 /* in case of lov EA swap, just set it
1228 * if not, it is a replay so check striping match what we
1229 * already have during req replay, declare_xattr_set()
1230 * defines striping, then create() does the work
1232 if (fl & LU_XATTR_REPLACE) {
1233 /* free stripes, then update disk */
1234 lod_object_free_striping(env, lod_dt_obj(dt));
1235 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
1237 rc = lod_striping_create(env, dt, NULL, NULL, th);
1239 } else if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
1242 rc = lod_xattr_set_default_lmv_on_dir(env, dt, buf, name, fl,
1246 * behave transparantly for all other EAs
1248 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
1251 if (rc != 0 || !S_ISDIR(attr))
1254 if (lo->ldo_stripenr == 0)
1257 for (i = 0; i < lo->ldo_stripenr; i++) {
1258 LASSERT(lo->ldo_stripe[i]);
1259 rc = dt_xattr_set(env, lo->ldo_stripe[i], buf, name, fl, th,
1268 static int lod_declare_xattr_del(const struct lu_env *env,
1269 struct dt_object *dt, const char *name,
1272 return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
1275 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
1276 const char *name, struct thandle *th,
1277 struct lustre_capa *capa)
1279 if (!strcmp(name, XATTR_NAME_LOV))
1280 lod_object_free_striping(env, lod_dt_obj(dt));
1281 return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
1284 static int lod_xattr_list(const struct lu_env *env,
1285 struct dt_object *dt, struct lu_buf *buf,
1286 struct lustre_capa *capa)
1288 return dt_xattr_list(env, dt_object_child(dt), buf, capa);
1291 int lod_object_set_pool(struct lod_object *o, char *pool)
1296 len = strlen(o->ldo_pool);
1297 OBD_FREE(o->ldo_pool, len + 1);
1302 OBD_ALLOC(o->ldo_pool, len + 1);
1303 if (o->ldo_pool == NULL)
1305 strcpy(o->ldo_pool, pool);
1310 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
1312 return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
1316 static int lod_cache_parent_lov_striping(const struct lu_env *env,
1317 struct lod_object *lp)
1319 struct lod_thread_info *info = lod_env_info(env);
1320 struct lov_user_md_v1 *v1 = NULL;
1321 struct lov_user_md_v3 *v3 = NULL;
1325 /* called from MDD without parent being write locked,
1327 dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
1328 rc = lod_get_lov_ea(env, lp);
1332 if (rc < sizeof(struct lov_user_md)) {
1333 /* don't lookup for non-existing or invalid striping */
1334 lp->ldo_def_striping_set = 0;
1335 lp->ldo_striping_cached = 1;
1336 lp->ldo_def_stripe_size = 0;
1337 lp->ldo_def_stripenr = 0;
1338 lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1);
1339 GOTO(unlock, rc = 0);
1343 v1 = info->lti_ea_store;
1344 if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1))
1345 lustre_swab_lov_user_md_v1(v1);
1346 else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3))
1347 lustre_swab_lov_user_md_v3(v3);
1349 if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1)
1350 GOTO(unlock, rc = 0);
1352 if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
1353 GOTO(unlock, rc = 0);
1355 lp->ldo_def_stripenr = v1->lmm_stripe_count;
1356 lp->ldo_def_stripe_size = v1->lmm_stripe_size;
1357 lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
1358 lp->ldo_striping_cached = 1;
1359 lp->ldo_def_striping_set = 1;
1360 if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
1361 /* XXX: sanity check here */
1362 v3 = (struct lov_user_md_v3 *) v1;
1363 if (v3->lmm_pool_name[0])
1364 lod_object_set_pool(lp, v3->lmm_pool_name);
1368 dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
1373 static int lod_cache_parent_lmv_striping(const struct lu_env *env,
1374 struct lod_object *lp)
1376 struct lod_thread_info *info = lod_env_info(env);
1377 struct lmv_user_md_v1 *v1 = NULL;
1381 /* called from MDD without parent being write locked,
1383 dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
1384 rc = lod_get_default_lmv_ea(env, lp);
1388 if (rc < sizeof(struct lmv_user_md)) {
1389 /* don't lookup for non-existing or invalid striping */
1390 lp->ldo_dir_def_striping_set = 0;
1391 lp->ldo_dir_striping_cached = 1;
1392 lp->ldo_dir_def_stripenr = 0;
1393 lp->ldo_dir_def_stripe_offset =
1394 (typeof(v1->lum_stripe_offset))(-1);
1395 lp->ldo_dir_def_hash_type = LMV_HASH_TYPE_FNV_1A_64;
1396 GOTO(unlock, rc = 0);
1400 v1 = info->lti_ea_store;
1402 lp->ldo_dir_def_stripenr = le32_to_cpu(v1->lum_stripe_count) - 1;
1403 lp->ldo_dir_def_stripe_offset = le32_to_cpu(v1->lum_stripe_offset);
1404 lp->ldo_dir_def_hash_type = le32_to_cpu(v1->lum_hash_type);
1405 lp->ldo_dir_def_striping_set = 1;
1406 lp->ldo_dir_striping_cached = 1;
1410 dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
1414 static int lod_cache_parent_striping(const struct lu_env *env,
1415 struct lod_object *lp,
1421 rc = lod_load_striping(env, lp);
1425 if (!lp->ldo_striping_cached) {
1426 /* we haven't tried to get default striping for
1427 * the directory yet, let's cache it in the object */
1428 rc = lod_cache_parent_lov_striping(env, lp);
1433 if (S_ISDIR(child_mode) && !lp->ldo_dir_striping_cached)
1434 rc = lod_cache_parent_lmv_striping(env, lp);
1440 * used to transfer default striping data to the object being created
1442 static void lod_ah_init(const struct lu_env *env,
1443 struct dt_allocation_hint *ah,
1444 struct dt_object *parent,
1445 struct dt_object *child,
1448 struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
1449 struct dt_object *nextp = NULL;
1450 struct dt_object *nextc;
1451 struct lod_object *lp = NULL;
1452 struct lod_object *lc;
1453 struct lov_desc *desc;
1458 if (likely(parent)) {
1459 nextp = dt_object_child(parent);
1460 lp = lod_dt_obj(parent);
1463 nextc = dt_object_child(child);
1464 lc = lod_dt_obj(child);
1466 LASSERT(lc->ldo_stripenr == 0);
1467 LASSERT(lc->ldo_stripe == NULL);
1470 * local object may want some hints
1471 * in case of late striping creation, ->ah_init()
1472 * can be called with local object existing
1474 if (!dt_object_exists(nextc) || dt_object_remote(nextc))
1475 nextc->do_ops->do_ah_init(env, ah, dt_object_remote(nextp) ?
1476 NULL : nextp, nextc, child_mode);
1478 if (S_ISDIR(child_mode)) {
1481 if (lc->ldo_dir_stripe == NULL) {
1482 OBD_ALLOC_PTR(lc->ldo_dir_stripe);
1483 if (lc->ldo_dir_stripe == NULL)
1487 if (lp->ldo_dir_stripe == NULL) {
1488 OBD_ALLOC_PTR(lp->ldo_dir_stripe);
1489 if (lp->ldo_dir_stripe == NULL)
1493 rc = lod_cache_parent_striping(env, lp, child_mode);
1497 /* transfer defaults to new directory */
1498 if (lp->ldo_striping_cached) {
1500 lod_object_set_pool(lc, lp->ldo_pool);
1501 lc->ldo_def_stripenr = lp->ldo_def_stripenr;
1502 lc->ldo_def_stripe_size = lp->ldo_def_stripe_size;
1503 lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
1504 lc->ldo_striping_cached = 1;
1505 lc->ldo_def_striping_set = 1;
1506 CDEBUG(D_OTHER, "inherite EA sz:%d off:%d nr:%d\n",
1507 (int)lc->ldo_def_stripe_size,
1508 (int)lc->ldo_def_stripe_offset,
1509 (int)lc->ldo_def_stripenr);
1512 /* transfer dir defaults to new directory */
1513 if (lp->ldo_dir_striping_cached) {
1514 lc->ldo_dir_def_stripenr = lp->ldo_dir_def_stripenr;
1515 lc->ldo_dir_def_stripe_offset =
1516 lp->ldo_dir_def_stripe_offset;
1517 lc->ldo_dir_def_hash_type =
1518 lp->ldo_dir_def_hash_type;
1519 lc->ldo_dir_striping_cached = 1;
1520 lc->ldo_dir_def_striping_set = 1;
1521 CDEBUG(D_INFO, "inherit default EA nr:%d off:%d t%u\n",
1522 (int)lc->ldo_dir_def_stripenr,
1523 (int)lc->ldo_dir_def_stripe_offset,
1524 lc->ldo_dir_def_hash_type);
1527 /* If the directory is specified with certain stripes */
1528 if (ah->dah_eadata != NULL && ah->dah_eadata_len != 0) {
1529 const struct lmv_user_md_v1 *lum1 = ah->dah_eadata;
1532 rc = lod_verify_md_striping(d, lum1);
1534 le32_to_cpu(lum1->lum_stripe_count) > 1) {
1535 /* Directory will be striped only if
1536 * stripe_count > 1 */
1538 le32_to_cpu(lum1->lum_stripe_count) - 1;
1539 lc->ldo_dir_stripe_offset =
1540 le32_to_cpu(lum1->lum_stripe_offset);
1541 lc->ldo_dir_hash_type =
1542 le32_to_cpu(lum1->lum_hash_type);
1543 CDEBUG(D_INFO, "set stripe EA nr:%hu off:%d\n",
1545 (int)lc->ldo_dir_stripe_offset);
1547 } else if (lp->ldo_dir_def_striping_set) {
1548 /* If there are default dir stripe from parent */
1549 lc->ldo_stripenr = lp->ldo_dir_def_stripenr;
1550 lc->ldo_dir_stripe_offset =
1551 lp->ldo_dir_def_stripe_offset;
1552 lc->ldo_dir_hash_type =
1553 lp->ldo_dir_def_hash_type;
1554 CDEBUG(D_INFO, "inherit EA nr:%hu off:%d\n",
1556 (int)lc->ldo_dir_stripe_offset);
1558 /* set default stripe for this directory */
1559 lc->ldo_stripenr = 0;
1560 lc->ldo_dir_stripe_offset = -1;
1563 CDEBUG(D_INFO, "final striping count:%hu, offset:%d\n",
1564 lc->ldo_stripenr, (int)lc->ldo_dir_stripe_offset);
1570 * if object is going to be striped over OSTs, transfer default
1571 * striping information to the child, so that we can use it
1572 * during declaration and creation
1574 if (!lod_object_will_be_striped(S_ISREG(child_mode),
1575 lu_object_fid(&child->do_lu)))
1578 * try from the parent
1580 if (likely(parent)) {
1581 lod_cache_parent_striping(env, lp, child_mode);
1583 lc->ldo_def_stripe_offset = (__u16) -1;
1585 if (lp->ldo_def_striping_set) {
1587 lod_object_set_pool(lc, lp->ldo_pool);
1588 lc->ldo_stripenr = lp->ldo_def_stripenr;
1589 lc->ldo_stripe_size = lp->ldo_def_stripe_size;
1590 lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
1591 CDEBUG(D_OTHER, "striping from parent: #%d, sz %d %s\n",
1592 lc->ldo_stripenr, lc->ldo_stripe_size,
1593 lp->ldo_pool ? lp->ldo_pool : "");
1598 * if the parent doesn't provide with specific pattern, grab fs-wide one
1600 desc = &d->lod_desc;
1601 if (lc->ldo_stripenr == 0)
1602 lc->ldo_stripenr = desc->ld_default_stripe_count;
1603 if (lc->ldo_stripe_size == 0)
1604 lc->ldo_stripe_size = desc->ld_default_stripe_size;
1605 CDEBUG(D_OTHER, "final striping: # %d stripes, sz %d from %s\n",
1606 lc->ldo_stripenr, lc->ldo_stripe_size,
1607 lc->ldo_pool ? lc->ldo_pool : "");
1610 /* we do not cache stripe information for slave stripe, see
1611 * lod_xattr_set_lov_on_dir */
1612 if (lp != NULL && lp->ldo_dir_slave_stripe)
1613 lod_lov_stripe_cache_clear(lp);
1618 #define ll_do_div64(aaa,bbb) do_div((aaa), (bbb))
1620 * this function handles a special case when truncate was done
1621 * on a stripeless object and now striping is being created
1622 * we can't lose that size, so we have to propagate it to newly
1625 static int lod_declare_init_size(const struct lu_env *env,
1626 struct dt_object *dt, struct thandle *th)
1628 struct dt_object *next = dt_object_child(dt);
1629 struct lod_object *lo = lod_dt_obj(dt);
1630 struct lu_attr *attr = &lod_env_info(env)->lti_attr;
1631 uint64_t size, offs;
1635 /* XXX: we support the simplest (RAID0) striping so far */
1636 LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
1637 LASSERT(lo->ldo_stripe_size > 0);
1639 rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
1640 LASSERT(attr->la_valid & LA_SIZE);
1644 size = attr->la_size;
1648 /* ll_do_div64(a, b) returns a % b, and a = a / b */
1649 ll_do_div64(size, (__u64) lo->ldo_stripe_size);
1650 stripe = ll_do_div64(size, (__u64) lo->ldo_stripenr);
1652 size = size * lo->ldo_stripe_size;
1653 offs = attr->la_size;
1654 size += ll_do_div64(offs, lo->ldo_stripe_size);
1656 attr->la_valid = LA_SIZE;
1657 attr->la_size = size;
1659 rc = dt_declare_attr_set(env, lo->ldo_stripe[stripe], attr, th);
1665 * Create declaration of striped object
1667 int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt,
1668 struct lu_attr *attr,
1669 const struct lu_buf *lovea, struct thandle *th)
1671 struct lod_thread_info *info = lod_env_info(env);
1672 struct dt_object *next = dt_object_child(dt);
1673 struct lod_object *lo = lod_dt_obj(dt);
1677 if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) {
1678 /* failed to create striping, let's reset
1679 * config so that others don't get confused */
1680 lod_object_free_striping(env, lo);
1681 GOTO(out, rc = -ENOMEM);
1684 /* choose OST and generate appropriate objects */
1685 rc = lod_qos_prep_create(env, lo, attr, lovea, th);
1687 /* failed to create striping, let's reset
1688 * config so that others don't get confused */
1689 lod_object_free_striping(env, lo);
1694 * declare storage for striping data
1696 info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
1697 lo->ldo_pool ? LOV_MAGIC_V3 : LOV_MAGIC_V1);
1698 rc = dt_declare_xattr_set(env, next, &info->lti_buf, XATTR_NAME_LOV,
1704 * if striping is created with local object's size > 0,
1705 * we have to propagate this size to specific object
1706 * the case is possible only when local object was created previously
1708 if (dt_object_exists(next))
1709 rc = lod_declare_init_size(env, dt, th);
1715 int lod_dir_striping_create_internal(const struct lu_env *env,
1716 struct dt_object *dt,
1717 struct lu_attr *attr,
1718 const struct dt_object_format *dof,
1722 struct lod_thread_info *info = lod_env_info(env);
1723 struct dt_object *next = dt_object_child(dt);
1724 struct lod_object *lo = lod_dt_obj(dt);
1728 if (lo->ldo_dir_def_striping_set &&
1729 !LMVEA_DELETE_VALUES(lo->ldo_stripenr,
1730 lo->ldo_dir_stripe_offset)) {
1731 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
1732 int stripe_count = lo->ldo_stripenr + 1;
1734 if (info->lti_ea_store_size < sizeof(*v1)) {
1735 rc = lod_ea_store_resize(info, sizeof(*v1));
1738 v1 = info->lti_ea_store;
1741 memset(v1, 0, sizeof(*v1));
1742 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
1743 v1->lum_stripe_count = cpu_to_le32(stripe_count);
1744 v1->lum_stripe_offset =
1745 cpu_to_le32(lo->ldo_dir_stripe_offset);
1747 info->lti_buf.lb_buf = v1;
1748 info->lti_buf.lb_len = sizeof(*v1);
1751 rc = lod_declare_xattr_set_lmv(env, dt, attr,
1752 &info->lti_buf, th);
1754 rc = lod_xattr_set_lmv(env, dt, &info->lti_buf,
1755 XATTR_NAME_LMV, 0, th,
1761 /* Transfer default LMV striping from the parent */
1762 if (lo->ldo_dir_striping_cached &&
1763 !LMVEA_DELETE_VALUES(lo->ldo_dir_def_stripenr,
1764 lo->ldo_dir_def_stripe_offset)) {
1765 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
1766 int def_stripe_count = lo->ldo_dir_def_stripenr + 1;
1768 if (info->lti_ea_store_size < sizeof(*v1)) {
1769 rc = lod_ea_store_resize(info, sizeof(*v1));
1772 v1 = info->lti_ea_store;
1775 memset(v1, 0, sizeof(*v1));
1776 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
1777 v1->lum_stripe_count = cpu_to_le32(def_stripe_count);
1778 v1->lum_stripe_offset =
1779 cpu_to_le32(lo->ldo_dir_def_stripe_offset);
1781 cpu_to_le32(lo->ldo_dir_def_hash_type);
1783 info->lti_buf.lb_buf = v1;
1784 info->lti_buf.lb_len = sizeof(*v1);
1786 rc = dt_declare_xattr_set(env, next, &info->lti_buf,
1787 XATTR_NAME_DEFAULT_LMV, 0,
1790 rc = dt_xattr_set(env, next, &info->lti_buf,
1791 XATTR_NAME_DEFAULT_LMV, 0, th,
1797 /* Transfer default LOV striping from the parent */
1798 if (lo->ldo_striping_cached &&
1799 !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
1800 lo->ldo_def_stripenr,
1801 lo->ldo_def_stripe_offset)) {
1802 struct lov_user_md_v3 *v3 = info->lti_ea_store;
1804 if (info->lti_ea_store_size < sizeof(*v3)) {
1805 rc = lod_ea_store_resize(info, sizeof(*v3));
1808 v3 = info->lti_ea_store;
1811 memset(v3, 0, sizeof(*v3));
1812 v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
1813 v3->lmm_stripe_count = cpu_to_le16(lo->ldo_def_stripenr);
1814 v3->lmm_stripe_offset = cpu_to_le16(lo->ldo_def_stripe_offset);
1815 v3->lmm_stripe_size = cpu_to_le32(lo->ldo_def_stripe_size);
1817 strncpy(v3->lmm_pool_name, lo->ldo_pool,
1820 info->lti_buf.lb_buf = v3;
1821 info->lti_buf.lb_len = sizeof(*v3);
1824 rc = dt_declare_xattr_set(env, next, &info->lti_buf,
1825 XATTR_NAME_LOV, 0, th);
1827 rc = dt_xattr_set(env, next, &info->lti_buf,
1828 XATTR_NAME_LOV, 0, th,
1837 static int lod_declare_dir_striping_create(const struct lu_env *env,
1838 struct dt_object *dt,
1839 struct lu_attr *attr,
1840 struct dt_object_format *dof,
1843 return lod_dir_striping_create_internal(env, dt, attr, dof, th, true);
1846 static int lod_dir_striping_create(const struct lu_env *env,
1847 struct dt_object *dt,
1848 struct lu_attr *attr,
1849 struct dt_object_format *dof,
1852 return lod_dir_striping_create_internal(env, dt, attr, dof, th, false);
1855 static int lod_declare_object_create(const struct lu_env *env,
1856 struct dt_object *dt,
1857 struct lu_attr *attr,
1858 struct dt_allocation_hint *hint,
1859 struct dt_object_format *dof,
1862 struct dt_object *next = dt_object_child(dt);
1863 struct lod_object *lo = lod_dt_obj(dt);
1872 * first of all, we declare creation of local object
1874 rc = dt_declare_create(env, next, attr, hint, dof, th);
1878 if (dof->dof_type == DFT_SYM)
1879 dt->do_body_ops = &lod_body_lnk_ops;
1882 * it's lod_ah_init() who has decided the object will striped
1884 if (dof->dof_type == DFT_REGULAR) {
1885 /* callers don't want stripes */
1886 /* XXX: all tricky interactions with ->ah_make_hint() decided
1887 * to use striping, then ->declare_create() behaving differently
1888 * should be cleaned */
1889 if (dof->u.dof_reg.striped == 0)
1890 lo->ldo_stripenr = 0;
1891 if (lo->ldo_stripenr > 0)
1892 rc = lod_declare_striped_object(env, dt, attr,
1894 } else if (dof->dof_type == DFT_DIR) {
1895 rc = lod_declare_dir_striping_create(env, dt, attr, dof, th);
1901 int lod_striping_create(const struct lu_env *env, struct dt_object *dt,
1902 struct lu_attr *attr, struct dt_object_format *dof,
1905 struct lod_object *lo = lod_dt_obj(dt);
1909 LASSERT(lo->ldo_striping_cached == 0);
1911 /* create all underlying objects */
1912 for (i = 0; i < lo->ldo_stripenr; i++) {
1913 LASSERT(lo->ldo_stripe[i]);
1914 rc = dt_create(env, lo->ldo_stripe[i], attr, NULL, dof, th);
1920 rc = lod_generate_and_set_lovea(env, lo, th);
1925 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
1926 struct lu_attr *attr,
1927 struct dt_allocation_hint *hint,
1928 struct dt_object_format *dof, struct thandle *th)
1930 struct dt_object *next = dt_object_child(dt);
1931 struct lod_object *lo = lod_dt_obj(dt);
1935 /* create local object */
1936 rc = dt_create(env, next, attr, hint, dof, th);
1939 if (S_ISDIR(dt->do_lu.lo_header->loh_attr))
1940 rc = lod_dir_striping_create(env, dt, attr, dof, th);
1941 else if (lo->ldo_stripe && dof->u.dof_reg.striped != 0)
1942 rc = lod_striping_create(env, dt, attr, dof, th);
1948 static int lod_declare_object_destroy(const struct lu_env *env,
1949 struct dt_object *dt,
1952 struct dt_object *next = dt_object_child(dt);
1953 struct lod_object *lo = lod_dt_obj(dt);
1958 * we declare destroy for the local object
1960 rc = dt_declare_destroy(env, next, th);
1964 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ))
1968 * load striping information, notice we don't do this when object
1969 * is being initialized as we don't need this information till
1970 * few specific cases like destroy, chown
1972 rc = lod_load_striping(env, lo);
1976 /* declare destroy for all underlying objects */
1977 for (i = 0; i < lo->ldo_stripenr; i++) {
1978 LASSERT(lo->ldo_stripe[i]);
1979 rc = dt_declare_destroy(env, lo->ldo_stripe[i], th);
1988 static int lod_object_destroy(const struct lu_env *env,
1989 struct dt_object *dt, struct thandle *th)
1991 struct dt_object *next = dt_object_child(dt);
1992 struct lod_object *lo = lod_dt_obj(dt);
1996 /* destroy local object */
1997 rc = dt_destroy(env, next, th);
2001 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ))
2004 /* destroy all underlying objects */
2005 for (i = 0; i < lo->ldo_stripenr; i++) {
2006 LASSERT(lo->ldo_stripe[i]);
2007 /* for striped directory, next == ldo_stripe[0] */
2008 if (next != lo->ldo_stripe[i]) {
2009 rc = dt_destroy(env, lo->ldo_stripe[i], th);
2018 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
2019 const struct dt_index_features *feat)
2021 struct dt_object *next = dt_object_child(dt);
2025 LASSERT(next->do_ops);
2026 LASSERT(next->do_ops->do_index_try);
2028 rc = next->do_ops->do_index_try(env, next, feat);
2029 if (next->do_index_ops && dt->do_index_ops == NULL)
2030 dt->do_index_ops = &lod_index_ops;
2035 static int lod_declare_ref_add(const struct lu_env *env,
2036 struct dt_object *dt, struct thandle *th)
2038 return dt_declare_ref_add(env, dt_object_child(dt), th);
2041 static int lod_ref_add(const struct lu_env *env,
2042 struct dt_object *dt, struct thandle *th)
2044 return dt_ref_add(env, dt_object_child(dt), th);
2047 static int lod_declare_ref_del(const struct lu_env *env,
2048 struct dt_object *dt, struct thandle *th)
2050 return dt_declare_ref_del(env, dt_object_child(dt), th);
2053 static int lod_ref_del(const struct lu_env *env,
2054 struct dt_object *dt, struct thandle *th)
2056 return dt_ref_del(env, dt_object_child(dt), th);
2059 static struct obd_capa *lod_capa_get(const struct lu_env *env,
2060 struct dt_object *dt,
2061 struct lustre_capa *old, __u64 opc)
2063 return dt_capa_get(env, dt_object_child(dt), old, opc);
2066 static int lod_object_sync(const struct lu_env *env, struct dt_object *dt)
2068 return dt_object_sync(env, dt_object_child(dt));
2071 struct lod_slave_locks {
2073 struct lustre_handle lsl_handle[0];
2076 static int lod_object_unlock_internal(const struct lu_env *env,
2077 struct dt_object *dt,
2078 struct ldlm_enqueue_info *einfo,
2079 ldlm_policy_data_t *policy)
2081 struct lod_object *lo = lod_dt_obj(dt);
2082 struct lod_slave_locks *slave_locks = einfo->ei_cbdata;
2087 if (slave_locks == NULL)
2090 for (i = 0; i < slave_locks->lsl_lock_count; i++) {
2091 if (lustre_handle_is_used(&slave_locks->lsl_handle[i])) {
2094 einfo->ei_cbdata = &slave_locks->lsl_handle[i];
2095 rc1 = dt_object_unlock(env, lo->ldo_stripe[i], einfo,
2098 rc = rc == 0 ? rc1 : rc;
2105 static int lod_object_unlock(const struct lu_env *env, struct dt_object *dt,
2106 struct ldlm_enqueue_info *einfo,
2107 union ldlm_policy_data *policy)
2109 struct lod_object *lo = lod_dt_obj(dt);
2110 struct lod_slave_locks *slave_locks = einfo->ei_cbdata;
2111 int slave_locks_size;
2115 if (slave_locks == NULL)
2118 rc = lod_load_striping(env, lo);
2122 /* Note: for remote lock for single stripe dir, MDT will cancel
2123 * the lock by lockh directly */
2124 if (lo->ldo_stripenr == 0 && dt_object_remote(dt_object_child(dt)))
2127 if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
2130 /* Only cancel slave lock for striped dir */
2131 rc = lod_object_unlock_internal(env, dt, einfo, policy);
2133 slave_locks_size = sizeof(*slave_locks) + slave_locks->lsl_lock_count *
2134 sizeof(slave_locks->lsl_handle[0]);
2135 OBD_FREE(slave_locks, slave_locks_size);
2136 einfo->ei_cbdata = NULL;
2141 static int lod_object_lock(const struct lu_env *env,
2142 struct dt_object *dt,
2143 struct lustre_handle *lh,
2144 struct ldlm_enqueue_info *einfo,
2145 union ldlm_policy_data *policy)
2147 struct lod_object *lo = lod_dt_obj(dt);
2150 int slave_locks_size;
2151 struct lod_slave_locks *slave_locks = NULL;
2154 /* remote object lock */
2155 if (!einfo->ei_enq_slave) {
2156 LASSERT(dt_object_remote(dt));
2157 return dt_object_lock(env, dt_object_child(dt), lh, einfo,
2161 if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
2164 rc = lod_load_striping(env, lo);
2169 if (lo->ldo_stripenr == 0)
2172 slave_locks_size = sizeof(*slave_locks) + lo->ldo_stripenr *
2173 sizeof(slave_locks->lsl_handle[0]);
2174 /* Freed in lod_object_unlock */
2175 OBD_ALLOC(slave_locks, slave_locks_size);
2176 if (slave_locks == NULL)
2178 slave_locks->lsl_lock_count = lo->ldo_stripenr;
2180 /* striped directory lock */
2181 for (i = 0; i < lo->ldo_stripenr; i++) {
2182 struct lustre_handle lockh;
2184 LASSERT(lo->ldo_stripe[i]);
2185 rc = dt_object_lock(env, lo->ldo_stripe[i], &lockh, einfo,
2190 slave_locks->lsl_handle[i] = lockh;
2193 einfo->ei_cbdata = slave_locks;
2196 if (rc != 0 && slave_locks != NULL) {
2197 einfo->ei_cbdata = slave_locks;
2198 lod_object_unlock_internal(env, dt, einfo, policy);
2199 OBD_FREE(slave_locks, slave_locks_size);
2200 einfo->ei_cbdata = NULL;
2206 struct dt_object_operations lod_obj_ops = {
2207 .do_read_lock = lod_object_read_lock,
2208 .do_write_lock = lod_object_write_lock,
2209 .do_read_unlock = lod_object_read_unlock,
2210 .do_write_unlock = lod_object_write_unlock,
2211 .do_write_locked = lod_object_write_locked,
2212 .do_attr_get = lod_attr_get,
2213 .do_declare_attr_set = lod_declare_attr_set,
2214 .do_attr_set = lod_attr_set,
2215 .do_xattr_get = lod_xattr_get,
2216 .do_declare_xattr_set = lod_declare_xattr_set,
2217 .do_xattr_set = lod_xattr_set,
2218 .do_declare_xattr_del = lod_declare_xattr_del,
2219 .do_xattr_del = lod_xattr_del,
2220 .do_xattr_list = lod_xattr_list,
2221 .do_ah_init = lod_ah_init,
2222 .do_declare_create = lod_declare_object_create,
2223 .do_create = lod_object_create,
2224 .do_declare_destroy = lod_declare_object_destroy,
2225 .do_destroy = lod_object_destroy,
2226 .do_index_try = lod_index_try,
2227 .do_declare_ref_add = lod_declare_ref_add,
2228 .do_ref_add = lod_ref_add,
2229 .do_declare_ref_del = lod_declare_ref_del,
2230 .do_ref_del = lod_ref_del,
2231 .do_capa_get = lod_capa_get,
2232 .do_object_sync = lod_object_sync,
2233 .do_object_lock = lod_object_lock,
2234 .do_object_unlock = lod_object_unlock,
2237 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
2238 struct lu_buf *buf, loff_t *pos,
2239 struct lustre_capa *capa)
2241 struct dt_object *next = dt_object_child(dt);
2242 return next->do_body_ops->dbo_read(env, next, buf, pos, capa);
2245 static ssize_t lod_declare_write(const struct lu_env *env,
2246 struct dt_object *dt,
2247 const loff_t size, loff_t pos,
2250 return dt_declare_record_write(env, dt_object_child(dt),
2254 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,
2255 const struct lu_buf *buf, loff_t *pos,
2256 struct thandle *th, struct lustre_capa *capa, int iq)
2258 struct dt_object *next = dt_object_child(dt);
2260 return next->do_body_ops->dbo_write(env, next, buf, pos, th, capa, iq);
2263 static const struct dt_body_operations lod_body_lnk_ops = {
2264 .dbo_read = lod_read,
2265 .dbo_declare_write = lod_declare_write,
2266 .dbo_write = lod_write
2269 static int lod_object_init(const struct lu_env *env, struct lu_object *lo,
2270 const struct lu_object_conf *conf)
2272 struct lod_device *lod = lu2lod_dev(lo->lo_dev);
2273 struct lu_device *cdev = NULL;
2274 struct lu_object *cobj;
2275 struct lod_tgt_descs *ltd = NULL;
2276 struct lod_tgt_desc *tgt;
2278 int type = LU_SEQ_RANGE_ANY;
2282 rc = lod_fld_lookup(env, lod, lu_object_fid(lo), &idx, &type);
2286 if (type == LU_SEQ_RANGE_MDT &&
2287 idx == lu_site2seq(lo->lo_dev->ld_site)->ss_node_id) {
2288 cdev = &lod->lod_child->dd_lu_dev;
2289 } else if (type == LU_SEQ_RANGE_MDT) {
2290 ltd = &lod->lod_mdt_descs;
2292 } else if (type == LU_SEQ_RANGE_OST) {
2293 ltd = &lod->lod_ost_descs;
2300 if (ltd->ltd_tgts_size > idx &&
2301 cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx)) {
2302 tgt = LTD_TGT(ltd, idx);
2304 LASSERT(tgt != NULL);
2305 LASSERT(tgt->ltd_tgt != NULL);
2307 cdev = &(tgt->ltd_tgt->dd_lu_dev);
2309 lod_putref(lod, ltd);
2312 if (unlikely(cdev == NULL))
2315 cobj = cdev->ld_ops->ldo_object_alloc(env, lo->lo_header, cdev);
2316 if (unlikely(cobj == NULL))
2319 lu_object_add(lo, cobj);
2324 void lod_object_free_striping(const struct lu_env *env, struct lod_object *lo)
2328 if (lo->ldo_dir_stripe != NULL) {
2329 OBD_FREE_PTR(lo->ldo_dir_stripe);
2330 lo->ldo_dir_stripe = NULL;
2333 if (lo->ldo_stripe) {
2334 LASSERT(lo->ldo_stripes_allocated > 0);
2336 for (i = 0; i < lo->ldo_stripenr; i++) {
2337 if (lo->ldo_stripe[i])
2338 lu_object_put(env, &lo->ldo_stripe[i]->do_lu);
2341 i = sizeof(struct dt_object *) * lo->ldo_stripes_allocated;
2342 OBD_FREE(lo->ldo_stripe, i);
2343 lo->ldo_stripe = NULL;
2344 lo->ldo_stripes_allocated = 0;
2346 lo->ldo_stripenr = 0;
2347 lo->ldo_pattern = 0;
2351 * ->start is called once all slices are initialized, including header's
2352 * cache for mode (object type). using the type we can initialize ops
2354 static int lod_object_start(const struct lu_env *env, struct lu_object *o)
2356 if (S_ISLNK(o->lo_header->loh_attr & S_IFMT))
2357 lu2lod_obj(o)->ldo_obj.do_body_ops = &lod_body_lnk_ops;
2361 static void lod_object_free(const struct lu_env *env, struct lu_object *o)
2363 struct lod_object *mo = lu2lod_obj(o);
2366 * release all underlying object pinned
2369 lod_object_free_striping(env, mo);
2371 lod_object_set_pool(mo, NULL);
2374 OBD_SLAB_FREE_PTR(mo, lod_object_kmem);
2377 static void lod_object_release(const struct lu_env *env, struct lu_object *o)
2379 /* XXX: shouldn't we release everything here in case if object
2380 * creation failed before? */
2383 static int lod_object_print(const struct lu_env *env, void *cookie,
2384 lu_printer_t p, const struct lu_object *l)
2386 struct lod_object *o = lu2lod_obj((struct lu_object *) l);
2388 return (*p)(env, cookie, LUSTRE_LOD_NAME"-object@%p", o);
2391 struct lu_object_operations lod_lu_obj_ops = {
2392 .loo_object_init = lod_object_init,
2393 .loo_object_start = lod_object_start,
2394 .loo_object_free = lod_object_free,
2395 .loo_object_release = lod_object_release,
2396 .loo_object_print = lod_object_print,