X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Flod%2Flod_object.c;h=cb15551fdcec1c15cbaadf829372dcbbcf15524a;hb=b78fb445555916e380b1661546c821df14098596;hp=95b59829b04cf182f5135e8b8ec4aa0b63523781;hpb=29d18f84e80b6f11d1e435c370ea962162851432;p=fs%2Flustre-release.git diff --git a/lustre/lod/lod_object.c b/lustre/lod/lod_object.c index 95b5982..cb15551 100644 --- a/lustre/lod/lod_object.c +++ b/lustre/lod/lod_object.c @@ -23,7 +23,7 @@ * Copyright 2009 Sun Microsystems, Inc. All rights reserved * Use is subject to license terms. * - * Copyright (c) 2012, 2014, Intel Corporation. + * Copyright (c) 2012, 2016, Intel Corporation. */ /* * lustre/lod/lod_object.c @@ -33,7 +33,7 @@ * local OSD object interface to the MDD layer, and abstracts the * addressing of local (OSD) and remote (OSP) objects. The API is * described in the file lustre/include/dt_object.h and in - * lustre/doc/osd-api.txt. + * Documentation/osd-api.txt. * * Author: Alex Zhuravlev */ @@ -42,16 +42,16 @@ #include #include -#include #include -#include #include -#include -#include +#include #include +#include +#include +#include +#include #include -#include #include "lod_internal.h" @@ -571,12 +571,12 @@ again: next = lo->ldo_stripe[it->lit_stripe_index]; LASSERT(next != NULL); - LASSERT(next->do_index_ops != NULL); - rc = next->do_ops->do_index_try(env, next, &dt_directory_features); if (rc != 0) RETURN(rc); + LASSERT(next->do_index_ops != NULL); + it_next = next->do_index_ops->dio_it.init(env, next, it->lit_attr); if (!IS_ERR(it_next)) { it->lit_it = it_next; @@ -1065,78 +1065,6 @@ static int lod_attr_get(const struct lu_env *env, } /** - * Mark all of the striped directory sub-stripes dead. - * - * When a striped object is a subject to removal, we have - * to mark all the stripes to prevent further access to - * them (e.g. create a new file in those). So we mark - * all the stripes with LMV_HASH_FLAG_DEAD. The function - * can be used to declare the changes and to apply them. - * If the object isn't striped, then just return success. - * - * \param[in] env execution environment - * \param[in] dt the striped object - * \param[in] handle transaction handle - * \param[in] declare whether to declare the change or apply - * - * \retval 0 on success - * \retval negative if failed - **/ -static int lod_mark_dead_object(const struct lu_env *env, - struct dt_object *dt, - struct thandle *th, - bool declare) -{ - struct lod_object *lo = lod_dt_obj(dt); - struct lmv_mds_md_v1 *lmv; - __u32 dead_hash_type; - int rc; - int i; - - ENTRY; - - if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) - RETURN(0); - - rc = lod_load_striping_locked(env, lo); - if (rc != 0) - RETURN(rc); - - if (lo->ldo_stripenr == 0) - RETURN(0); - - rc = lod_get_lmv_ea(env, lo); - if (rc <= 0) - RETURN(rc); - - lmv = lod_env_info(env)->lti_ea_store; - lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_STRIPE); - dead_hash_type = le32_to_cpu(lmv->lmv_hash_type) | LMV_HASH_FLAG_DEAD; - lmv->lmv_hash_type = cpu_to_le32(dead_hash_type); - for (i = 0; i < lo->ldo_stripenr; i++) { - struct lu_buf buf; - - lmv->lmv_master_mdt_index = i; - buf.lb_buf = lmv; - buf.lb_len = sizeof(*lmv); - if (declare) { - rc = lod_sub_object_declare_xattr_set(env, - lo->ldo_stripe[i], &buf, - XATTR_NAME_LMV, - LU_XATTR_REPLACE, th); - } else { - rc = lod_sub_object_xattr_set(env, lo->ldo_stripe[i], - &buf, XATTR_NAME_LMV, - LU_XATTR_REPLACE, th); - } - if (rc != 0) - break; - } - - RETURN(rc); -} - -/** * Implementation of dt_object_operations::do_declare_attr_set. * * If the object is striped, then apply the changes to all the stripes. @@ -1154,13 +1082,6 @@ static int lod_declare_attr_set(const struct lu_env *env, int rc, i; ENTRY; - /* Set dead object on all other stripes */ - if (attr->la_valid & LA_FLAGS && !(attr->la_valid & ~LA_FLAGS) && - attr->la_flags & LUSTRE_SLAVE_DEAD_FL) { - rc = lod_mark_dead_object(env, dt, th, true); - RETURN(rc); - } - /* * declare setattr on the local object */ @@ -1253,13 +1174,6 @@ static int lod_attr_set(const struct lu_env *env, int rc, i; ENTRY; - /* Set dead object on all other stripes */ - if (attr->la_valid & LA_FLAGS && !(attr->la_valid & ~LA_FLAGS) && - attr->la_flags & LUSTRE_SLAVE_DEAD_FL) { - rc = lod_mark_dead_object(env, dt, th, false); - RETURN(rc); - } - /* * apply changes to the local object */ @@ -1734,11 +1648,8 @@ static int lod_dir_declare_create_stripes(const struct lu_env *env, PFID(lu_object_fid(&dto->do_lu)), i); sname = lod_name_get(env, stripe_name, strlen(stripe_name)); - rc = linkea_data_new(&ldata, &info->lti_linkea_buf); - if (rc != 0) - GOTO(out, rc); - - rc = linkea_add_buf(&ldata, sname, lu_object_fid(&dt->do_lu)); + rc = linkea_links_new(&ldata, &info->lti_linkea_buf, + sname, lu_object_fid(&dt->do_lu)); if (rc != 0) GOTO(out, rc); @@ -1799,17 +1710,12 @@ static int lod_prep_md_striped_create(const struct lu_env *env, stripe_count = le32_to_cpu(lum->lum_stripe_count); - /* shrink the stripe_count to the avaible MDT count */ - if (stripe_count > lod->lod_remote_mdt_count + 1 && - !OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE)) - stripe_count = lod->lod_remote_mdt_count + 1; + OBD_ALLOC(idx_array, sizeof(idx_array[0]) * stripe_count); + if (idx_array == NULL) + RETURN(-ENOMEM); OBD_ALLOC(stripe, sizeof(stripe[0]) * stripe_count); if (stripe == NULL) - RETURN(-ENOMEM); - - OBD_ALLOC(idx_array, sizeof(idx_array[0]) * stripe_count); - if (idx_array == NULL) GOTO(out_free, rc = -ENOMEM); /* Start index will be the master MDT */ @@ -1933,8 +1839,7 @@ out_put: } out_free: - if (idx_array != NULL) - OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count); + OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count); RETURN(rc); } @@ -1997,7 +1902,6 @@ out: RETURN(rc); } - /** * Implementation of dt_object_operations::do_declare_xattr_set. * @@ -2035,16 +1939,19 @@ static int lod_dir_declare_xattr_set(const struct lu_env *env, if (rc != 0) RETURN(rc); + /* Note: Do not set LinkEA on sub-stripes, otherwise + * it will confuse the fid2path process(see mdt_path_current()). + * The linkEA between master and sub-stripes is set in + * lod_xattr_set_lmv(). */ + if (strcmp(name, XATTR_NAME_LINK) == 0) + RETURN(0); + /* set xattr to each stripes, if needed */ rc = lod_load_striping(env, lo); if (rc != 0) RETURN(rc); - /* Note: Do not set LinkEA on sub-stripes, otherwise - * it will confuse the fid2path process(see mdt_path_current()). - * The linkEA between master and sub-stripes is set in - * lod_xattr_set_lmv(). */ - if (lo->ldo_stripenr == 0 || strcmp(name, XATTR_NAME_LINK) == 0) + if (lo->ldo_stripenr == 0) RETURN(0); for (i = 0; i < lo->ldo_stripenr; i++) { @@ -2060,6 +1967,85 @@ static int lod_dir_declare_xattr_set(const struct lu_env *env, } /** + * Reset parent FID on OST object + * + * Replace parent FID with @dt object FID, which is only called during migration + * to reset the parent FID after the MDT object is migrated to the new MDT, i.e. + * the FID is changed. + * + * \param[in] env execution environment + * \param[in] dt dt_object whose stripes's parent FID will be reset + * \parem[in] th thandle + * \param[in] declare if it is declare + * + * \retval 0 if reset succeeds + * \retval negative errno if reset fais + */ +static int lod_object_replace_parent_fid(const struct lu_env *env, + struct dt_object *dt, + struct thandle *th, bool declare) +{ + struct lod_object *lo = lod_dt_obj(dt); + struct lod_thread_info *info = lod_env_info(env); + struct lu_buf *buf = &info->lti_buf; + struct filter_fid *ff; + int i, rc; + ENTRY; + + LASSERT(S_ISREG(dt->do_lu.lo_header->loh_attr)); + + /* set xattr to each stripes, if needed */ + rc = lod_load_striping(env, lo); + if (rc != 0) + RETURN(rc); + + if (lo->ldo_stripenr == 0) + RETURN(0); + + if (info->lti_ea_store_size < sizeof(*ff)) { + rc = lod_ea_store_resize(info, sizeof(*ff)); + if (rc != 0) + RETURN(rc); + } + + buf->lb_buf = info->lti_ea_store; + buf->lb_len = info->lti_ea_store_size; + + for (i = 0; i < lo->ldo_stripenr; i++) { + if (lo->ldo_stripe[i] == NULL) + continue; + + rc = dt_xattr_get(env, lo->ldo_stripe[i], buf, + XATTR_NAME_FID); + if (rc < 0) { + rc = 0; + continue; + } + + ff = buf->lb_buf; + fid_le_to_cpu(&ff->ff_parent, &ff->ff_parent); + ff->ff_parent.f_seq = lu_object_fid(&dt->do_lu)->f_seq; + ff->ff_parent.f_oid = lu_object_fid(&dt->do_lu)->f_oid; + fid_cpu_to_le(&ff->ff_parent, &ff->ff_parent); + + if (declare) { + rc = lod_sub_object_declare_xattr_set(env, + lo->ldo_stripe[i], buf, + XATTR_NAME_FID, + LU_XATTR_REPLACE, th); + } else { + rc = lod_sub_object_xattr_set(env, lo->ldo_stripe[i], + buf, XATTR_NAME_FID, + LU_XATTR_REPLACE, th); + } + if (rc < 0) + break; + } + + RETURN(rc); +} + +/** * Implementation of dt_object_operations::do_declare_xattr_set. * * \see dt_object_operations::do_declare_xattr_set() in the API description @@ -2105,6 +2091,8 @@ static int lod_declare_xattr_set(const struct lu_env *env, rc = lod_declare_striped_object(env, dt, attr, buf, th); } else if (S_ISDIR(mode)) { rc = lod_dir_declare_xattr_set(env, dt, buf, name, fl, th); + } else if (strcmp(name, XATTR_NAME_FID) == 0) { + rc = lod_object_replace_parent_fid(env, dt, th, true); } else { rc = lod_sub_object_declare_xattr_set(env, next, buf, name, fl, th); @@ -2114,22 +2102,6 @@ static int lod_declare_xattr_set(const struct lu_env *env, } /** - * Resets cached default striping in the object. - * - * \param[in] lo object - */ -static void lod_lov_stripe_cache_clear(struct lod_object *lo) -{ - lo->ldo_def_striping_set = 0; - lo->ldo_def_striping_cached = 0; - lod_object_set_pool(lo, NULL); - lo->ldo_def_stripe_size = 0; - lo->ldo_def_stripenr = 0; - if (lo->ldo_dir_stripe != NULL) - lo->ldo_dir_def_striping_cached = 0; -} - -/** * Apply xattr changes to the object. * * Applies xattr changes to the object and the stripes if the latter exist. @@ -2246,18 +2218,12 @@ static int lod_xattr_set_lov_on_dir(const struct lu_env *env, struct thandle *th) { struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev); - struct lod_object *l = lod_dt_obj(dt); struct lov_user_md_v1 *lum; struct lov_user_md_v3 *v3 = NULL; const char *pool_name = NULL; int rc; ENTRY; - /* If it is striped dir, we should clear the stripe cache for - * slave stripe as well, but there are no effective way to - * notify the LOD on the slave MDT, so we do not cache stripe - * information for slave stripe for now. XXX*/ - lod_lov_stripe_cache_clear(l); LASSERT(buf != NULL && buf->lb_buf != NULL); lum = buf->lb_buf; @@ -2317,7 +2283,6 @@ static int lod_xattr_set_default_lmv_on_dir(const struct lu_env *env, const char *name, int fl, struct thandle *th) { - struct lod_object *l = lod_dt_obj(dt); struct lmv_user_md_v1 *lum; int rc; ENTRY; @@ -2341,14 +2306,6 @@ static int lod_xattr_set_default_lmv_on_dir(const struct lu_env *env, RETURN(rc); } - /* Update default stripe cache */ - if (l->ldo_dir_stripe == NULL) { - OBD_ALLOC_PTR(l->ldo_dir_stripe); - if (l->ldo_dir_stripe == NULL) - RETURN(-ENOMEM); - } - - l->ldo_dir_def_striping_cached = 0; RETURN(rc); } @@ -2478,11 +2435,8 @@ static int lod_xattr_set_lmv(const struct lu_env *env, struct dt_object *dt, PFID(lu_object_fid(&dto->do_lu)), i); sname = lod_name_get(env, stripe_name, strlen(stripe_name)); - rc = linkea_data_new(&ldata, &info->lti_linkea_buf); - if (rc != 0) - GOTO(out, rc); - - rc = linkea_add_buf(&ldata, sname, lu_object_fid(&dt->do_lu)); + rc = linkea_links_new(&ldata, &info->lti_linkea_buf, + sname, lu_object_fid(&dt->do_lu)); if (rc != 0) GOTO(out, rc); @@ -2544,11 +2498,17 @@ static int lod_dir_striping_create_internal(const struct lu_env *env, struct thandle *th, bool declare) { - struct lod_thread_info *info = lod_env_info(env); - struct lod_object *lo = lod_dt_obj(dt); - int rc; + struct lod_thread_info *info = lod_env_info(env); + struct lod_object *lo = lod_dt_obj(dt); + const struct lod_default_striping *lds = lo->ldo_def_striping; + const char *poolname = NULL; + int rc; ENTRY; + LASSERT(ergo(lds != NULL, + lds->lds_def_striping_set || + lds->lds_dir_def_striping_set)); + if (!LMVEA_DELETE_VALUES(lo->ldo_stripenr, lo->ldo_dir_stripe_offset)) { struct lmv_user_md_v1 *v1 = info->lti_ea_store; @@ -2581,11 +2541,10 @@ static int lod_dir_striping_create_internal(const struct lu_env *env, } /* Transfer default LMV striping from the parent */ - if (lo->ldo_dir_def_striping_set && - !LMVEA_DELETE_VALUES(lo->ldo_dir_def_stripenr, - lo->ldo_dir_def_stripe_offset)) { + if (lds != NULL && lds->lds_dir_def_striping_set && + !LMVEA_DELETE_VALUES(lds->lds_dir_def_stripenr, + lds->lds_dir_def_stripe_offset)) { struct lmv_user_md_v1 *v1 = info->lti_ea_store; - int def_stripe_count = lo->ldo_dir_def_stripenr; if (info->lti_ea_store_size < sizeof(*v1)) { rc = lod_ea_store_resize(info, sizeof(*v1)); @@ -2596,11 +2555,11 @@ static int lod_dir_striping_create_internal(const struct lu_env *env, memset(v1, 0, sizeof(*v1)); v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC); - v1->lum_stripe_count = cpu_to_le32(def_stripe_count); + v1->lum_stripe_count = cpu_to_le32(lds->lds_dir_def_stripenr); v1->lum_stripe_offset = - cpu_to_le32(lo->ldo_dir_def_stripe_offset); + cpu_to_le32(lds->lds_dir_def_stripe_offset); v1->lum_hash_type = - cpu_to_le32(lo->ldo_dir_def_hash_type); + cpu_to_le32(lds->lds_dir_def_hash_type); info->lti_buf.lb_buf = v1; info->lti_buf.lb_len = sizeof(*v1); @@ -2617,12 +2576,15 @@ static int lod_dir_striping_create_internal(const struct lu_env *env, RETURN(rc); } + if (lds != NULL && lds->lds_def_pool[0] != '\0') + poolname = lds->lds_def_pool; + /* Transfer default LOV striping from the parent */ - if (lo->ldo_def_striping_set && - !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size, - lo->ldo_def_stripenr, - lo->ldo_def_stripe_offset, - lo->ldo_pool)) { + if (lds != NULL && lds->lds_def_striping_set && + !LOVEA_DELETE_VALUES(lds->lds_def_stripe_size, + lds->lds_def_stripenr, + lds->lds_def_stripe_offset, + poolname)) { struct lov_user_md_v3 *v3 = info->lti_ea_store; if (info->lti_ea_store_size < sizeof(*v3)) { @@ -2634,11 +2596,11 @@ static int lod_dir_striping_create_internal(const struct lu_env *env, memset(v3, 0, sizeof(*v3)); v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3); - v3->lmm_stripe_count = cpu_to_le16(lo->ldo_def_stripenr); - v3->lmm_stripe_offset = cpu_to_le16(lo->ldo_def_stripe_offset); - v3->lmm_stripe_size = cpu_to_le32(lo->ldo_def_stripe_size); - if (lo->ldo_pool != NULL) - strlcpy(v3->lmm_pool_name, lo->ldo_pool, + v3->lmm_stripe_count = cpu_to_le16(lds->lds_def_stripenr); + v3->lmm_stripe_offset = cpu_to_le16(lds->lds_def_stripe_offset); + v3->lmm_stripe_size = cpu_to_le32(lds->lds_def_stripe_size); + if (poolname != NULL) + strlcpy(v3->lmm_pool_name, poolname, sizeof(v3->lmm_pool_name)); info->lti_buf.lb_buf = v3; @@ -2672,14 +2634,7 @@ static int lod_dir_striping_create(const struct lu_env *env, struct dt_object_format *dof, struct thandle *th) { - struct lod_object *lo = lod_dt_obj(dt); - int rc; - - rc = lod_dir_striping_create_internal(env, dt, attr, dof, th, false); - if (rc == 0) - lo->ldo_striping_cached = 1; - - return rc; + return lod_dir_striping_create_internal(env, dt, attr, dof, th, false); } /** @@ -2757,6 +2712,10 @@ static int lod_xattr_set(const struct lu_env *env, rc = lod_striping_create(env, dt, NULL, NULL, th); } RETURN(rc); + } else if (strcmp(name, XATTR_NAME_FID) == 0) { + rc = lod_object_replace_parent_fid(env, dt, th, false); + + RETURN(rc); } /* then all other xattr */ @@ -2857,38 +2816,6 @@ static int lod_xattr_list(const struct lu_env *env, return dt_xattr_list(env, dt_object_child(dt), buf); } -/** - * Initialize a pool the object belongs to. - * - * When a striped object is being created, striping configuration - * may demand the stripes are allocated on a limited set of the - * targets. These limited sets are known as "pools". So we copy - * a pool name into the object and later actual creation methods - * (like lod_object_create()) will use this information to allocate - * the stripes properly. - * - * \param[in] o object - * \param[in] pool pool name - */ -int lod_object_set_pool(struct lod_object *o, char *pool) -{ - int len; - - if (o->ldo_pool) { - len = strlen(o->ldo_pool); - OBD_FREE(o->ldo_pool, len + 1); - o->ldo_pool = NULL; - } - if (pool) { - len = strlen(pool); - OBD_ALLOC(o->ldo_pool, len + 1); - if (o->ldo_pool == NULL) - return -ENOMEM; - strcpy(o->ldo_pool, pool); - } - return 0; -} - static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid) { return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE); @@ -2896,44 +2823,31 @@ static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fi /** - * Cache default regular striping in the object. - * - * To improve performance of striped regular object creation we cache - * default LOV striping (if it exists) in the parent directory object. + * Get default striping. * * \param[in] env execution environment - * \param[in] lp object + * \param[in] lo object + * \param[out] lds default striping * * \retval 0 on success * \retval negative if failed */ -static int lod_cache_parent_lov_striping(const struct lu_env *env, - struct lod_object *lp) +static int lod_get_default_lov_striping(const struct lu_env *env, + struct lod_object *lo, + struct lod_default_striping *lds) { - struct lod_thread_info *info = lod_env_info(env); - struct lov_user_md_v1 *v1 = NULL; - struct lov_user_md_v3 *v3 = NULL; - int rc; - ENTRY; + struct lod_thread_info *info = lod_env_info(env); + struct lov_user_md_v1 *v1 = NULL; + struct lov_user_md_v3 *v3 = NULL; + int rc; - /* called from MDD without parent being write locked, - * lock it here */ - dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0); - rc = lod_get_lov_ea(env, lp); + rc = lod_get_lov_ea(env, lo); if (rc < 0) - GOTO(unlock, rc); - - if (rc < (typeof(rc))sizeof(struct lov_user_md)) { - /* don't lookup for non-existing or invalid striping */ - lp->ldo_def_striping_set = 0; - lp->ldo_def_striping_cached = 1; - lp->ldo_def_stripe_size = 0; - lp->ldo_def_stripenr = 0; - lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1); - GOTO(unlock, rc = 0); - } + return rc; + + if (rc < (typeof(rc))sizeof(struct lov_user_md)) + return 0; - rc = 0; v1 = info->lti_ea_store; if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1)) { lustre_swab_lov_user_md_v1(v1); @@ -2943,127 +2857,126 @@ static int lod_cache_parent_lov_striping(const struct lu_env *env, } if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1) - GOTO(unlock, rc = 0); + return 0; if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0) - GOTO(unlock, rc = 0); - - CDEBUG(D_INFO, DFID" stripe_count=%d stripe_size=%d stripe_offset=%d\n", - PFID(lu_object_fid(&lp->ldo_obj.do_lu)), - (int)v1->lmm_stripe_count, - (int)v1->lmm_stripe_size, (int)v1->lmm_stripe_offset); - - lp->ldo_def_stripenr = v1->lmm_stripe_count; - lp->ldo_def_stripe_size = v1->lmm_stripe_size; - lp->ldo_def_stripe_offset = v1->lmm_stripe_offset; - lp->ldo_def_striping_cached = 1; - lp->ldo_def_striping_set = 1; + return 0; + + lds->lds_def_stripenr = v1->lmm_stripe_count; + lds->lds_def_stripe_size = v1->lmm_stripe_size; + lds->lds_def_stripe_offset = v1->lmm_stripe_offset; + lds->lds_def_striping_set = 1; if (v1->lmm_magic == LOV_USER_MAGIC_V3) { - /* XXX: sanity check here */ - v3 = (struct lov_user_md_v3 *) v1; - if (v3->lmm_pool_name[0]) - lod_object_set_pool(lp, v3->lmm_pool_name); + v3 = (struct lov_user_md_v3 *)v1; + if (v3->lmm_pool_name[0] != '\0') + strlcpy(lds->lds_def_pool, v3->lmm_pool_name, + sizeof(lds->lds_def_pool)); } - EXIT; -unlock: - dt_write_unlock(env, dt_object_child(&lp->ldo_obj)); - return rc; -} + return 0; +} /** - * Cache default directory striping in the object. - * - * To improve performance of striped directory creation we cache default - * directory striping (if it exists) in the parent directory object. + * Get default directory striping. * * \param[in] env execution environment - * \param[in] lp object + * \param[in] lo object + * \param[out] lds default striping * * \retval 0 on success * \retval negative if failed */ -static int lod_cache_parent_lmv_striping(const struct lu_env *env, - struct lod_object *lp) +static int lod_get_default_lmv_striping(const struct lu_env *env, + struct lod_object *lo, + struct lod_default_striping *lds) { struct lod_thread_info *info = lod_env_info(env); struct lmv_user_md_v1 *v1 = NULL; int rc; - ENTRY; - /* called from MDD without parent being write locked, - * lock it here */ - dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0); - rc = lod_get_default_lmv_ea(env, lp); + rc = lod_get_default_lmv_ea(env, lo); if (rc < 0) - GOTO(unlock, rc); - - if (rc < (typeof(rc))sizeof(struct lmv_user_md)) { - /* don't lookup for non-existing or invalid striping */ - lp->ldo_dir_def_striping_set = 0; - lp->ldo_dir_def_striping_cached = 1; - lp->ldo_dir_def_stripenr = 0; - lp->ldo_dir_def_stripe_offset = - (typeof(v1->lum_stripe_offset))(-1); - lp->ldo_dir_def_hash_type = LMV_HASH_TYPE_FNV_1A_64; - GOTO(unlock, rc = 0); - } + return rc; + + if (rc < (typeof(rc))sizeof(struct lmv_user_md)) + return 0; - rc = 0; v1 = info->lti_ea_store; - lp->ldo_dir_def_stripenr = le32_to_cpu(v1->lum_stripe_count); - lp->ldo_dir_def_stripe_offset = le32_to_cpu(v1->lum_stripe_offset); - lp->ldo_dir_def_hash_type = le32_to_cpu(v1->lum_hash_type); - lp->ldo_dir_def_striping_set = 1; - lp->ldo_dir_def_striping_cached = 1; + lds->lds_dir_def_stripenr = le32_to_cpu(v1->lum_stripe_count); + lds->lds_dir_def_stripe_offset = le32_to_cpu(v1->lum_stripe_offset); + lds->lds_dir_def_hash_type = le32_to_cpu(v1->lum_hash_type); + lds->lds_dir_def_striping_set = 1; - EXIT; -unlock: - dt_write_unlock(env, dt_object_child(&lp->ldo_obj)); - return rc; + return 0; } /** - * Cache default striping in the object. + * Get default striping in the object. * - * To improve performance of striped object creation we cache default striping - * (if it exists) in the parent directory object. We always cache default - * striping for the regular files (stored in LOV EA) and we cache default - * striping for the directories if requested by \a child_mode (when a new - * directory is being created). + * Get object default striping and default directory striping. * * \param[in] env execution environment - * \param[in] lp object - * \param[in] child_mode new object's mode + * \param[in] lo object + * \param[out] lds default striping * * \retval 0 on success * \retval negative if failed */ -static int lod_cache_parent_striping(const struct lu_env *env, - struct lod_object *lp, - umode_t child_mode) +static int lod_get_default_striping(const struct lu_env *env, + struct lod_object *lo, + struct lod_default_striping *lds) { - int rc = 0; - ENTRY; - - if (!lp->ldo_def_striping_cached) { - /* we haven't tried to get default striping for - * the directory yet, let's cache it in the object */ - rc = lod_cache_parent_lov_striping(env, lp); - if (rc != 0) - RETURN(rc); - } + int rc; - /* If the parent is on the remote MDT, we should always - * try to refresh the default stripeEA cache, because we - * do not cache default striping information for remote - * object. */ - if (S_ISDIR(child_mode) && (!lp->ldo_dir_def_striping_cached || - dt_object_remote(&lp->ldo_obj))) - rc = lod_cache_parent_lmv_striping(env, lp); + rc = lod_get_default_lov_striping(env, lo, lds); + if (rc == 0) + rc = lod_get_default_lmv_striping(env, lo, lds); + return rc; +} - RETURN(rc); +/** + * Apply default striping on object. + * + * If object striping pattern is not set, set to the one in default striping. + * The default striping is from parent or fs. + * + * \param[in] lo new object + * \param[in] lds default striping + * \param[in] mode new object's mode + */ +static void lod_striping_from_default(struct lod_object *lo, + const struct lod_default_striping *lds, + umode_t mode) +{ + if (lds->lds_def_striping_set && S_ISREG(mode)) { + if (lo->ldo_stripenr == 0) + lo->ldo_stripenr = lds->lds_def_stripenr; + if (lo->ldo_stripe_size == 0) + lo->ldo_stripe_size = lds->lds_def_stripe_size; + if (lo->ldo_stripe_offset == LOV_OFFSET_DEFAULT) + lo->ldo_stripe_offset = lds->lds_def_stripe_offset; + if (lo->ldo_pool == NULL && lds->lds_def_pool[0] != '\0') + lod_object_set_pool(lo, lds->lds_def_pool); + + CDEBUG(D_INFO, "striping from default: count %hu, size %u, " + "offset %d, pool %s\n", + lo->ldo_stripenr, lo->ldo_stripe_size, + (int)lo->ldo_stripe_offset, lo->ldo_pool ?: ""); + } else if (lds->lds_dir_def_striping_set && S_ISDIR(mode)) { + if (lo->ldo_stripenr == 0) + lo->ldo_stripenr = lds->lds_dir_def_stripenr; + if (lo->ldo_dir_stripe_offset == -1) + lo->ldo_dir_stripe_offset = + lds->lds_dir_def_stripe_offset; + if (lo->ldo_dir_hash_type == 0) + lo->ldo_dir_hash_type = lds->lds_dir_def_hash_type; + + CDEBUG(D_INFO, "striping from default: count %hu, offset %d, " + "hash_type %u\n", + lo->ldo_stripenr, (int)lo->ldo_dir_stripe_offset, + lo->ldo_dir_hash_type); + } } /** @@ -3084,12 +2997,13 @@ static void lod_ah_init(const struct lu_env *env, umode_t child_mode) { struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev); - struct dt_object *nextp = NULL; - struct dt_object *nextc; + struct lod_thread_info *info = lod_env_info(env); + struct lod_default_striping *lds = &info->lti_def_striping; + struct dt_object *nextp = NULL; + struct dt_object *nextc; struct lod_object *lp = NULL; struct lod_object *lc; - struct lov_desc *desc; - int rc; + struct lov_desc *desc; ENTRY; LASSERT(child); @@ -3097,9 +3011,6 @@ static void lod_ah_init(const struct lu_env *env, if (likely(parent)) { nextp = dt_object_child(parent); lp = lod_dt_obj(parent); - rc = lod_load_striping(env, lp); - if (rc != 0) - return; } nextc = dt_object_child(child); @@ -3112,93 +3023,52 @@ static void lod_ah_init(const struct lu_env *env, nextc->do_ops->do_ah_init(env, ah, nextp, nextc, child_mode); if (S_ISDIR(child_mode)) { - if (lc->ldo_dir_stripe == NULL) { - OBD_ALLOC_PTR(lc->ldo_dir_stripe); - if (lc->ldo_dir_stripe == NULL) - return; - } + /* other default values are 0 */ + lc->ldo_dir_stripe_offset = -1; - LASSERT(lp != NULL); - if (lp->ldo_dir_stripe == NULL) { - OBD_ALLOC_PTR(lp->ldo_dir_stripe); - if (lp->ldo_dir_stripe == NULL) - return; - } + memset(lds, 0, sizeof(*lds)); + lod_get_default_striping(env, lp, lds); - rc = lod_cache_parent_striping(env, lp, child_mode); - if (rc != 0) - return; - - /* transfer defaults to new directory */ - if (lp->ldo_def_striping_set) { - if (lp->ldo_pool) - lod_object_set_pool(lc, lp->ldo_pool); - lc->ldo_def_stripenr = lp->ldo_def_stripenr; - lc->ldo_def_stripe_size = lp->ldo_def_stripe_size; - lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset; - lc->ldo_def_striping_set = 1; - lc->ldo_def_striping_cached = 1; - CDEBUG(D_OTHER, "inherite EA sz:%d off:%d nr:%d\n", - (int)lc->ldo_def_stripe_size, - (int)lc->ldo_def_stripe_offset, - (int)lc->ldo_def_stripenr); - } - - /* transfer dir defaults to new directory */ - if (lp->ldo_dir_def_striping_set) { - lc->ldo_dir_def_stripenr = lp->ldo_dir_def_stripenr; - lc->ldo_dir_def_stripe_offset = - lp->ldo_dir_def_stripe_offset; - lc->ldo_dir_def_hash_type = - lp->ldo_dir_def_hash_type; - lc->ldo_dir_def_striping_set = 1; - lc->ldo_dir_def_striping_cached = 1; - CDEBUG(D_INFO, "inherit default EA nr:%d off:%d t%u\n", - (int)lc->ldo_dir_def_stripenr, - (int)lc->ldo_dir_def_stripe_offset, - lc->ldo_dir_def_hash_type); - } + /* inherit parent default striping */ + if (lds->lds_def_striping_set || lds->lds_dir_def_striping_set) + lc->ldo_def_striping = lds; /* It should always honour the specified stripes */ - if (ah->dah_eadata != NULL && ah->dah_eadata_len != 0) { + if (ah->dah_eadata != NULL && ah->dah_eadata_len != 0 && + lod_verify_md_striping(d, ah->dah_eadata) == 0) { const struct lmv_user_md_v1 *lum1 = ah->dah_eadata; - rc = lod_verify_md_striping(d, lum1); - if (rc == 0 && - le32_to_cpu(lum1->lum_stripe_count) > 1) { - /* Directory will be striped only if - * stripe_count > 1 */ - lc->ldo_stripenr = - le32_to_cpu(lum1->lum_stripe_count); - lc->ldo_dir_stripe_offset = - le32_to_cpu(lum1->lum_stripe_offset); - lc->ldo_dir_hash_type = - le32_to_cpu(lum1->lum_hash_type); - CDEBUG(D_INFO, "set stripe EA nr:%hu off:%d\n", - lc->ldo_stripenr, - (int)lc->ldo_dir_stripe_offset); - } - /* then check whether there is default stripes from parent */ - } else if (lp->ldo_dir_def_striping_set) { - /* If there are default dir stripe from parent */ - lc->ldo_stripenr = lp->ldo_dir_def_stripenr; + lc->ldo_stripenr = le32_to_cpu(lum1->lum_stripe_count); lc->ldo_dir_stripe_offset = - lp->ldo_dir_def_stripe_offset; + le32_to_cpu(lum1->lum_stripe_offset); lc->ldo_dir_hash_type = - lp->ldo_dir_def_hash_type; - CDEBUG(D_INFO, "inherit EA nr:%hu off:%d\n", - lc->ldo_stripenr, - (int)lc->ldo_dir_stripe_offset); + le32_to_cpu(lum1->lum_hash_type); + CDEBUG(D_INFO, "set dir stripe: count %hu, offset %d, " + "hash_type %u\n", + lc->ldo_stripenr, + (int)lc->ldo_dir_stripe_offset, + lc->ldo_dir_hash_type); } else { - /* set default stripe for this directory */ - lc->ldo_stripenr = 0; - lc->ldo_dir_stripe_offset = -1; + lod_striping_from_default(lc, lds, child_mode); } - CDEBUG(D_INFO, "final striping count:%hu, offset:%d\n", - lc->ldo_stripenr, (int)lc->ldo_dir_stripe_offset); + /* shrink the stripe_count to the avaible MDT count */ + if (lc->ldo_stripenr > d->lod_remote_mdt_count + 1 && + !OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE)) + lc->ldo_stripenr = d->lod_remote_mdt_count + 1; - goto out; + /* Directory will be striped only if stripe_count > 1, if + * stripe_count == 1, let's reset stripenr = 0 to avoid + * create single master stripe and also help to unify the + * stripe handling of directories and files */ + if (lc->ldo_stripenr == 1) + lc->ldo_stripenr = 0; + + CDEBUG(D_INFO, "final dir stripe [%hu %d %u]\n", + lc->ldo_stripenr, (int)lc->ldo_dir_stripe_offset, + lc->ldo_dir_hash_type); + + RETURN_EXIT; } /* @@ -3208,45 +3078,60 @@ static void lod_ah_init(const struct lu_env *env, */ if (!lod_object_will_be_striped(S_ISREG(child_mode), lu_object_fid(&child->do_lu))) - goto out; - /* - * try from the parent - */ + RETURN_EXIT; + + /* other default values are 0 */ + lc->ldo_stripe_offset = LOV_OFFSET_DEFAULT; + + /* striping from parent default */ if (likely(parent)) { - lod_cache_parent_striping(env, lp, child_mode); - - lc->ldo_def_stripe_offset = LOV_OFFSET_DEFAULT; - - if (lp->ldo_def_striping_set) { - if (lp->ldo_pool) - lod_object_set_pool(lc, lp->ldo_pool); - lc->ldo_stripenr = lp->ldo_def_stripenr; - lc->ldo_stripe_size = lp->ldo_def_stripe_size; - lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset; - CDEBUG(D_OTHER, "striping from parent: #%d, sz %d %s\n", - lc->ldo_stripenr, lc->ldo_stripe_size, - lp->ldo_pool ? lp->ldo_pool : ""); + memset(lds, 0, sizeof(*lds)); + lod_get_default_lov_striping(env, lp, lds); + lod_striping_from_default(lc, lds, child_mode); + } + + if (d->lod_md_root == NULL) { + struct dt_object *root; + struct lod_object *lroot; + + lu_root_fid(&info->lti_fid); + root = dt_locate(env, &d->lod_dt_dev, &info->lti_fid); + if (!IS_ERR(root)) { + lroot = lod_dt_obj(root); + + spin_lock(&d->lod_lock); + if (d->lod_md_root != NULL) + lu_object_put(env, + &d->lod_md_root->ldo_obj.do_lu); + d->lod_md_root = lroot; + spin_unlock(&d->lod_lock); } } + /* if parent doesn't provide all defaults, striping from fs default */ + if (d->lod_md_root != NULL && + (lc->ldo_stripenr == 0 || + lc->ldo_stripe_size == 0 || + lc->ldo_stripe_offset == LOV_OFFSET_DEFAULT || + lc->ldo_pool == NULL)) { + memset(lds, 0, sizeof(*lds)); + lod_get_default_lov_striping(env, d->lod_md_root, lds); + lod_striping_from_default(lc, lds, child_mode); + } + /* - * if the parent doesn't provide with specific pattern, grab fs-wide one + * fs default striping may not be explicitly set, or historically set + * in config log, check striping sanity here and fix to sane values. */ desc = &d->lod_desc; if (lc->ldo_stripenr == 0) lc->ldo_stripenr = desc->ld_default_stripe_count; if (lc->ldo_stripe_size == 0) lc->ldo_stripe_size = desc->ld_default_stripe_size; - CDEBUG(D_OTHER, "final striping: # %d stripes, sz %d from %s\n", - lc->ldo_stripenr, lc->ldo_stripe_size, - lc->ldo_pool ? lc->ldo_pool : ""); - -out: - /* we do not cache stripe information for slave stripe, see - * lod_xattr_set_lov_on_dir */ - if (lp != NULL && lp->ldo_dir_slave_stripe) - lod_lov_stripe_cache_clear(lp); + CDEBUG(D_INFO, "final striping [%hu %u %d %s]\n", + lc->ldo_stripenr, lc->ldo_stripe_size, + (int)lc->ldo_stripe_offset, lc->ldo_pool ?: ""); EXIT; } @@ -3281,6 +3166,9 @@ static int lod_declare_init_size(const struct lu_env *env, LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0); LASSERT(lo->ldo_stripe_size > 0); + if (lo->ldo_stripenr == 0) + RETURN(0); + rc = dt_attr_get(env, next, attr); LASSERT(attr->la_valid & LA_SIZE); if (rc) @@ -3336,22 +3224,14 @@ int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt, int rc; ENTRY; - if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) { - /* failed to create striping, let's reset - * config so that others don't get confused */ - lod_object_free_striping(env, lo); + if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) GOTO(out, rc = -ENOMEM); - } if (!dt_object_remote(next)) { /* choose OST and generate appropriate objects */ rc = lod_qos_prep_create(env, lo, attr, lovea, th); - if (rc) { - /* failed to create striping, let's reset - * config so that others don't get confused */ - lod_object_free_striping(env, lo); + if (rc) GOTO(out, rc); - } /* * declare storage for striping data @@ -3382,6 +3262,11 @@ int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt, rc = lod_declare_init_size(env, dt, th); out: + /* failed to create striping or to set initial size, let's reset + * config so that others don't get confused */ + if (rc) + lod_object_free_striping(env, lo); + RETURN(rc); } @@ -3462,15 +3347,35 @@ static int lod_declare_object_create(const struct lu_env *env, GOTO(out, rc = -EREMOTE); } else if (lo->ldo_dir_stripe_offset != ss->ss_node_id) { - GOTO(out, rc = -EREMOTE); + struct lod_device *lod; + struct lod_tgt_descs *ltd; + struct lod_tgt_desc *tgt = NULL; + bool found_mdt = false; + int i; + + lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev); + ltd = &lod->lod_mdt_descs; + cfs_foreach_bit(ltd->ltd_tgt_bitmap, i) { + tgt = LTD_TGT(ltd, i); + if (tgt->ltd_index == + lo->ldo_dir_stripe_offset) { + found_mdt = true; + break; + } + } + + /* If the MDT indicated by stripe_offset can be + * found, then tell client to resend the create + * request to the correct MDT, otherwise return + * error to client */ + if (found_mdt) + GOTO(out, rc = -EREMOTE); + else + GOTO(out, rc = -EINVAL); } } - /* Orphan object (like migrating object) does not have - * lod_dir_stripe, see lod_ah_init */ - if (lo->ldo_dir_stripe != NULL) - rc = lod_declare_dir_striping_create(env, dt, attr, - dof, th); + rc = lod_declare_dir_striping_create(env, dt, attr, dof, th); } out: RETURN(rc); @@ -3503,8 +3408,6 @@ int lod_striping_create(const struct lu_env *env, struct dt_object *dt, int rc = 0, i; ENTRY; - LASSERT(lo->ldo_striping_cached == 0); - /* create all underlying objects */ for (i = 0; i < lo->ldo_stripenr; i++) { LASSERT(lo->ldo_stripe[i]); @@ -3514,11 +3417,8 @@ int lod_striping_create(const struct lu_env *env, struct dt_object *dt, break; } - if (rc == 0) { + if (rc == 0) rc = lod_generate_and_set_lovea(env, lo, th); - if (rc == 0) - lo->ldo_striping_cached = 1; - } RETURN(rc); } @@ -3614,7 +3514,8 @@ static int lod_declare_object_destroy(const struct lu_env *env, if (rc) RETURN(rc); - if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ)) + if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ) || + OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ2)) RETURN(0); /* declare destroy all striped objects */ @@ -3686,7 +3587,8 @@ static int lod_object_destroy(const struct lu_env *env, if (rc != 0) RETURN(rc); - if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ)) + if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ) || + OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ2)) RETURN(0); /* destroy all striped objects */ @@ -3771,16 +3673,11 @@ static int lod_object_sync(const struct lu_env *env, struct dt_object *dt, return dt_object_sync(env, dt_object_child(dt), start, end); } -struct lod_slave_locks { - int lsl_lock_count; - struct lustre_handle lsl_handle[0]; -}; - /** * Release LDLM locks on the stripes of a striped directory. * * Iterates over all the locks taken on the stripe objects and - * release them using ->do_object_unlock() method. + * cancel them. * * \param[in] env execution environment * \param[in] dt striped object @@ -3793,9 +3690,9 @@ struct lod_slave_locks { static int lod_object_unlock_internal(const struct lu_env *env, struct dt_object *dt, struct ldlm_enqueue_info *einfo, - ldlm_policy_data_t *policy) + union ldlm_policy_data *policy) { - struct lod_slave_locks *slave_locks = einfo->ei_cbdata; + struct lustre_handle_array *slave_locks = einfo->ei_cbdata; int rc = 0; int i; ENTRY; @@ -3803,10 +3700,10 @@ static int lod_object_unlock_internal(const struct lu_env *env, if (slave_locks == NULL) RETURN(0); - for (i = 1; i < slave_locks->lsl_lock_count; i++) { - if (lustre_handle_is_used(&slave_locks->lsl_handle[i])) - ldlm_lock_decref(&slave_locks->lsl_handle[i], - einfo->ei_mode); + for (i = 1; i < slave_locks->count; i++) { + if (lustre_handle_is_used(&slave_locks->handles[i])) + ldlm_lock_decref_and_cancel(&slave_locks->handles[i], + einfo->ei_mode); } RETURN(rc); @@ -3824,32 +3721,33 @@ static int lod_object_unlock(const struct lu_env *env, struct dt_object *dt, struct ldlm_enqueue_info *einfo, union ldlm_policy_data *policy) { - struct lod_object *lo = lod_dt_obj(dt); - struct lod_slave_locks *slave_locks = einfo->ei_cbdata; - int slave_locks_size; - int rc; + struct lod_object *lo = lod_dt_obj(dt); + struct lustre_handle_array *slave_locks = einfo->ei_cbdata; + int slave_locks_size; + int i; ENTRY; if (slave_locks == NULL) RETURN(0); - if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) - RETURN(-ENOTDIR); - + LASSERT(S_ISDIR(dt->do_lu.lo_header->loh_attr)); + LASSERT(lo->ldo_stripenr > 1); /* Note: for remote lock for single stripe dir, MDT will cancel * the lock by lockh directly */ - if (lo->ldo_stripenr <= 1 && dt_object_remote(dt_object_child(dt))) - RETURN(0); + LASSERT(!dt_object_remote(dt_object_child(dt))); - /* Only cancel slave lock for striped dir */ - rc = lod_object_unlock_internal(env, dt, einfo, policy); + /* locks were unlocked in MDT layer */ + for (i = 1; i < slave_locks->count; i++) { + LASSERT(!lustre_handle_is_used(&slave_locks->handles[i])); + dt_invalidate(env, lo->ldo_stripe[i]); + } - slave_locks_size = sizeof(*slave_locks) + slave_locks->lsl_lock_count * - sizeof(slave_locks->lsl_handle[0]); + slave_locks_size = sizeof(*slave_locks) + slave_locks->count * + sizeof(slave_locks->handles[0]); OBD_FREE(slave_locks, slave_locks_size); einfo->ei_cbdata = NULL; - RETURN(rc); + RETURN(0); } /** @@ -3870,7 +3768,7 @@ static int lod_object_lock(const struct lu_env *env, int rc = 0; int i; int slave_locks_size; - struct lod_slave_locks *slave_locks = NULL; + struct lustre_handle_array *slave_locks = NULL; ENTRY; /* remote object lock */ @@ -3881,23 +3779,30 @@ static int lod_object_lock(const struct lu_env *env, } if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) - RETURN(-ENOTDIR); + GOTO(out, rc = -ENOTDIR); rc = lod_load_striping(env, lo); if (rc != 0) - RETURN(rc); + GOTO(out, rc); /* No stripes */ - if (lo->ldo_stripenr <= 1) - RETURN(0); + if (lo->ldo_stripenr <= 1) { + /* + * NB, ei_cbdata stores pointer to slave locks, if no locks + * taken, make sure it's set to NULL, otherwise MDT will try to + * unlock them. + */ + einfo->ei_cbdata = NULL; + GOTO(out, rc = 0); + } slave_locks_size = sizeof(*slave_locks) + lo->ldo_stripenr * - sizeof(slave_locks->lsl_handle[0]); + sizeof(slave_locks->handles[0]); /* Freed in lod_object_unlock */ OBD_ALLOC(slave_locks, slave_locks_size); if (slave_locks == NULL) - RETURN(-ENOMEM); - slave_locks->lsl_lock_count = lo->ldo_stripenr; + GOTO(out, rc = -ENOMEM); + slave_locks->count = lo->ldo_stripenr; /* striped directory lock */ for (i = 1; i < lo->ldo_stripenr; i++) { @@ -3919,6 +3824,10 @@ static int lod_object_lock(const struct lu_env *env, ldlm_completion_callback completion = einfo->ei_cb_cp; __u64 dlmflags = LDLM_FL_ATOMIC_CB; + if (einfo->ei_mode == LCK_PW || + einfo->ei_mode == LCK_EX) + dlmflags |= LDLM_FL_COS_INCOMPAT; + /* This only happens if there are mulitple stripes * on the master MDT, i.e. except stripe0, there are * other stripes on the Master MDT as well, Only @@ -3932,23 +3841,32 @@ static int lod_object_lock(const struct lu_env *env, NULL, &lockh); } if (rc != 0) - GOTO(out, rc); - slave_locks->lsl_handle[i] = lockh; + break; + slave_locks->handles[i] = lockh; } - einfo->ei_cbdata = slave_locks; -out: if (rc != 0 && slave_locks != NULL) { - einfo->ei_cbdata = slave_locks; lod_object_unlock_internal(env, dt, einfo, policy); OBD_FREE(slave_locks, slave_locks_size); - einfo->ei_cbdata = NULL; } - + EXIT; +out: + if (rc != 0) + einfo->ei_cbdata = NULL; RETURN(rc); } +/** + * Implementation of dt_object_operations::do_invalidate. + * + * \see dt_object_operations::do_invalidate() in the API description for details + */ +static int lod_invalidate(const struct lu_env *env, struct dt_object *dt) +{ + return dt_invalidate(env, dt_object_child(dt)); +} + struct dt_object_operations lod_obj_ops = { .do_read_lock = lod_object_read_lock, .do_write_lock = lod_object_write_lock, @@ -3977,6 +3895,7 @@ struct dt_object_operations lod_obj_ops = { .do_object_sync = lod_object_sync, .do_object_lock = lod_object_lock, .do_object_unlock = lod_object_unlock, + .do_invalidate = lod_invalidate, }; /** @@ -4137,14 +4056,11 @@ static int lod_object_init(const struct lu_env *env, struct lu_object *lo, */ void lod_object_free_striping(const struct lu_env *env, struct lod_object *lo) { - int i; + int len; - if (lo->ldo_dir_stripe != NULL) { - OBD_FREE_PTR(lo->ldo_dir_stripe); - lo->ldo_dir_stripe = NULL; - } + if (lo->ldo_stripe != NULL) { + int i; - if (lo->ldo_stripe) { LASSERT(lo->ldo_stripes_allocated > 0); for (i = 0; i < lo->ldo_stripenr; i++) { @@ -4152,14 +4068,12 @@ void lod_object_free_striping(const struct lu_env *env, struct lod_object *lo) lu_object_put(env, &lo->ldo_stripe[i]->do_lu); } - i = sizeof(struct dt_object *) * lo->ldo_stripes_allocated; - OBD_FREE(lo->ldo_stripe, i); + len = sizeof(struct dt_object *) * lo->ldo_stripes_allocated; + OBD_FREE(lo->ldo_stripe, len); lo->ldo_stripe = NULL; lo->ldo_stripes_allocated = 0; } - lo->ldo_striping_cached = 0; lo->ldo_stripenr = 0; - lo->ldo_pattern = 0; } /** @@ -4191,18 +4105,13 @@ static int lod_object_start(const struct lu_env *env, struct lu_object *o) */ static void lod_object_free(const struct lu_env *env, struct lu_object *o) { - struct lod_object *mo = lu2lod_obj(o); - - /* - * release all underlying object pinned - */ - - lod_object_free_striping(env, mo); - - lod_object_set_pool(mo, NULL); + struct lod_object *lo = lu2lod_obj(o); + lod_object_set_pool(lo, NULL); + /* release all underlying object pinned */ + lod_object_free_striping(env, lo); lu_object_fini(o); - OBD_SLAB_FREE_PTR(mo, lod_object_kmem); + OBD_SLAB_FREE_PTR(lo, lod_object_kmem); } /**