static inline int mdt_object_striped(struct mdt_thread_info *mti,
struct mdt_object *obj)
{
+ struct lu_device *bottom_dev;
+ struct lu_object *bottom_obj;
int rc;
if (!S_ISDIR(obj->mot_header.loh_attr))
return 0;
- rc = mo_xattr_get(mti->mti_env, mdt_object_child(obj), &LU_BUF_NULL,
+ /* getxattr from bottom obj to avoid reading in shard FIDs */
+ bottom_dev = dt2lu_dev(mti->mti_mdt->mdt_bottom);
+ bottom_obj = lu_object_find_slice(mti->mti_env, bottom_dev,
+ mdt_object_fid(obj), NULL);
+ if (IS_ERR(bottom_obj))
+ return PTR_ERR(bottom_obj);
+
+ rc = dt_xattr_get(mti->mti_env, lu2dt(bottom_obj), &LU_BUF_NULL,
XATTR_NAME_LMV);
- if (rc <= 0)
- return rc == -ENODATA ? 0 : rc;
+ lu_object_put(mti->mti_env, bottom_obj);
- return 1;
+ return (rc > 0) ? 1 : (rc == -ENODATA) ? 0 : rc;
}
/**
/* LU-10286: compatibility check for FLR.
* Please check the comment in mdt_finish_open() for details */
- if (!exp_connect_flr(info->mti_exp)) {
+ if (!exp_connect_flr(info->mti_exp) ||
+ !exp_connect_overstriping(info->mti_exp)) {
rc = mdt_big_xattr_get(info, mo, XATTR_NAME_LOV);
if (rc < 0 && rc != -ENODATA)
GOTO(out_put, rc);
- if (rc > 0 && mdt_lmm_is_flr(info->mti_big_lmm))
- GOTO(out_put, rc = -EOPNOTSUPP);
+ if (!exp_connect_flr(info->mti_exp)) {
+ if (rc > 0 &&
+ mdt_lmm_is_flr(info->mti_big_lmm))
+ GOTO(out_put, rc = -EOPNOTSUPP);
+ }
+
+ if (!exp_connect_overstriping(info->mti_exp)) {
+ if (rc > 0 &&
+ mdt_lmm_is_overstriping(info->mti_big_lmm))
+ GOTO(out_put, rc = -EOPNOTSUPP);
+ }
}
/* For truncate, the file size sent from client
struct lu_buf *buf = &info->mti_buf;
struct lu_ucred *uc = mdt_ucred(info);
struct mdt_lock_handle *lh;
+ const char *name;
+ __u64 lockpart = MDS_INODELOCK_XATTR;
/* reject if either remote or striped dir is disabled */
if (ma->ma_valid & MA_LMV) {
GOTO(out_put, rc = -EPERM);
}
+ if (!S_ISDIR(lu_object_attr(&mo->mot_obj)))
+ GOTO(out_put, rc = -ENOTDIR);
+
if (ma->ma_attr.la_valid != 0)
GOTO(out_put, rc = -EPROTO);
- lh = &info->mti_lh[MDT_LH_PARENT];
- mdt_lock_reg_init(lh, LCK_PW);
-
- rc = mdt_object_lock(info, mo, lh, MDS_INODELOCK_XATTR);
- if (rc != 0)
- GOTO(out_put, rc);
-
if (ma->ma_valid & MA_LOV) {
buf->lb_buf = ma->ma_lmm;
buf->lb_len = ma->ma_lmm_size;
+ name = XATTR_NAME_LOV;
} else {
- buf->lb_buf = ma->ma_lmv;
+ struct lmv_user_md *lmu = &ma->ma_lmv->lmv_user_md;
+
+ buf->lb_buf = lmu;
buf->lb_len = ma->ma_lmv_size;
+
+ if (le32_to_cpu(lmu->lum_hash_type) &
+ LMV_HASH_FLAG_SPACE) {
+ /*
+ * only allow setting "space" hash flag on
+ * plain directory.
+ */
+ rc = mdt_object_striped(info, mo);
+ if (rc)
+ GOTO(out_put,
+ rc = (rc == 1) ? -EPERM : rc);
+ }
+
+ name = XATTR_NAME_DEFAULT_LMV;
+ /* force client to update dir default layout */
+ lockpart |= MDS_INODELOCK_LOOKUP;
}
+
+ lh = &info->mti_lh[MDT_LH_PARENT];
+ mdt_lock_reg_init(lh, LCK_PW);
+
+ rc = mdt_object_lock(info, mo, lh, lockpart);
+ if (rc != 0)
+ GOTO(out_put, rc);
+
rc = mo_xattr_set(info->mti_env, mdt_object_child(mo), buf,
- (ma->ma_valid & MA_LOV) ?
- XATTR_NAME_LOV : XATTR_NAME_DEFAULT_LMV,
- 0);
+ name, 0);
mdt_object_unlock(info, mo, lh, rc);
if (rc)
struct mdt_lock_handle *child_lh;
struct ldlm_enqueue_info *einfo = &info->mti_einfo[0];
__u64 lock_ibits;
- bool cos_incompat = false, discard = false;
+ bool cos_incompat = false;
int no_name = 0;
int rc;
if (!cos_incompat) {
rc = mdt_object_striped(info, mc);
if (rc < 0)
- GOTO(unlock_parent, rc);
+ GOTO(put_child, rc);
cos_incompat = rc;
if (cos_incompat) {
rc = mdt_attr_get_complex(info, mc, ma);
if (rc)
GOTO(out_stat, rc);
- } else {
- discard = mdt_dom_check_for_discard(info, mc);
+ } else if (mdt_dom_check_for_discard(info, mc)) {
+ mdt_dom_discard_data(info, mc);
}
mdt_handle_last_unlink(info, mc, ma);
mdt_object_unlock(info, mp, parent_lh, rc);
put_parent:
mdt_object_put(info->mti_env, mp);
-
- /* discard is just a PW DOM lock to drop the data on a client
- * no need to keep objects being get and locked, do that after all.
- */
- if (discard)
- mdt_dom_discard_data(info, child_fid);
-
+ CFS_RACE_WAKEUP(OBD_FAIL_OBD_ZERO_NLINK_RACE);
return rc;
}
for (i = 0; i < le32_to_cpu(lmv->lmv_stripe_count); i++) {
fid_le_to_cpu(fid, &lmv->lmv_stripe_fids[i]);
+ if (!fid_is_sane(fid))
+ continue;
+
slave = mdt_object_find(info->mti_env, mdt, fid);
if (IS_ERR(slave))
GOTO(out, rc = PTR_ERR(slave));
struct lu_fid *old_fid = &info->mti_tmp_fid1;
struct lu_fid *new_fid = &info->mti_tmp_fid2;
__u64 lock_ibits;
- bool reverse = false;
- bool cos_incompat, discard = false;
+ bool reverse = false, discard = false;
+ bool cos_incompat;
int rc;
ENTRY;
mdt_handle_last_unlink(info, mnew, ma);
discard = mdt_dom_check_for_discard(info, mnew);
}
-
mdt_rename_counter_tally(info, info->mti_mdt, req,
msrcdir, mtgtdir);
}
out_unlock_old:
mdt_object_unlock(info, mold, lh_oldp, rc);
out_put_new:
- if (mnew != NULL)
+ if (mnew && !discard)
mdt_object_put(info->mti_env, mnew);
out_put_old:
mdt_object_put(info->mti_env, mold);
out_put_srcdir:
mdt_object_put(info->mti_env, msrcdir);
- /* If 'discard' is set then new_fid must exits.
- * DOM data discard need neither object nor lock,
- * so do this at the end.
+ /* The DoM discard can be done right in the place above where it is
+ * assigned, meanwhile it is done here after rename unlock due to
+ * compatibility with old clients, for them the discard blocks
+ * the main thread until completion. Check LU-11359 for details.
*/
- if (discard)
- mdt_dom_discard_data(info, new_fid);
-
+ if (discard) {
+ mdt_dom_discard_data(info, mnew);
+ mdt_object_put(info->mti_env, mnew);
+ }
return rc;
}
git://git.whamcloud.com / fs / lustre-release.git / blobdiff