LU-11642 mdt: revoke remote LOOKUP lock in dir layout shrink

mdt_dir_layout_shrink() should revoke remote LOOKUP lock if parent
is remote, because it will alter dir layout, which is refreshed
upon lookup.

Signed-off-by: Lai Siyao <lai.siyao@whamcloud.com>
Change-Id: I26ae1af5da6142b44005e5d9ea11293af65ed7b5
Reviewed-on: https://review.whamcloud.com/33640
Tested-by: Jenkins
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Hongchao Zhang <hongchao@whamcloud.com>
Reviewed-by: Mike Pershin <mpershin@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/mdt/mdt_internal.h b/lustre/mdt/mdt_internal.h
index 35744bf..5bf811f 100644 (file)
--- a/lustre/mdt/mdt_internal.h
+++ b/lustre/mdt/mdt_internal.h
@@ -828,6 +828,9 @@ int mdt_reint_open(struct mdt_thread_info *info, struct mdt_lock_handle *lhc);
 struct mdt_file_data *mdt_open_handle2mfd(struct mdt_export_data *med,
                                        const struct lustre_handle *open_handle,
                                        bool is_replay);
 struct mdt_file_data *mdt_open_handle2mfd(struct mdt_export_data *med,
                                        const struct lustre_handle *open_handle,
                                        bool is_replay);

+int mdt_revoke_remote_lookup_lock(struct mdt_thread_info *info,
+                                 struct mdt_object *pobj,
+                                 struct mdt_object *obj);

 
 int mdt_get_info(struct tgt_session_info *tsi);
 int mdt_attr_get_complex(struct mdt_thread_info *info,
 
 int mdt_get_info(struct tgt_session_info *tsi);
 int mdt_attr_get_complex(struct mdt_thread_info *info,

diff --git a/lustre/mdt/mdt_reint.c b/lustre/mdt/mdt_reint.c
index 87f26ce..a4ea444 100644 (file)
--- a/lustre/mdt/mdt_reint.c
+++ b/lustre/mdt/mdt_reint.c
@@ -1303,9 +1303,9 @@ out_put:
  * in case obj is remote obj on its parent, revoke LOOKUP lock,
  * herein we don't really check it, just do revoke.
  */
  * in case obj is remote obj on its parent, revoke LOOKUP lock,
  * herein we don't really check it, just do revoke.
  */

-static int mdt_revoke_remote_lookup_lock(struct mdt_thread_info *info,
-                                        struct mdt_object *pobj,
-                                        struct mdt_object *obj)
+int mdt_revoke_remote_lookup_lock(struct mdt_thread_info *info,
+                                 struct mdt_object *pobj,
+                                 struct mdt_object *obj)

 {
        struct mdt_lock_handle *lh = &info->mti_lh[MDT_LH_LOCAL];
        int rc;
 {
        struct mdt_lock_handle *lh = &info->mti_lh[MDT_LH_LOCAL];
        int rc;


@@ -1314,10 +1314,11 @@ static int mdt_revoke_remote_lookup_lock(struct mdt_thread_info *info,
        mdt_lock_reg_init(lh, LCK_EX);
 
        if (mdt_object_remote(pobj)) {
        mdt_lock_reg_init(lh, LCK_EX);
 
        if (mdt_object_remote(pobj)) {

+               /* don't bother to check if pobj and obj are on the same MDT. */

                rc = mdt_remote_object_lock(info, pobj, mdt_object_fid(obj),
                                            &lh->mlh_rreg_lh, LCK_EX,
                                            MDS_INODELOCK_LOOKUP, false);
                rc = mdt_remote_object_lock(info, pobj, mdt_object_fid(obj),
                                            &lh->mlh_rreg_lh, LCK_EX,
                                            MDS_INODELOCK_LOOKUP, false);

-       } else {
+       } else if (mdt_object_remote(obj)) {

                struct ldlm_res_id *res = &info->mti_res_id;
                union ldlm_policy_data *policy = &info->mti_policy;
                __u64 dlmflags = LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB |
                struct ldlm_res_id *res = &info->mti_res_id;
                union ldlm_policy_data *policy = &info->mti_policy;
                __u64 dlmflags = LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB |


@@ -1329,6 +1330,9 @@ static int mdt_revoke_remote_lookup_lock(struct mdt_thread_info *info,
                rc = mdt_fid_lock(info->mti_env, info->mti_mdt->mdt_namespace,
                                  &lh->mlh_reg_lh, LCK_EX, policy, res,
                                  dlmflags, NULL);
                rc = mdt_fid_lock(info->mti_env, info->mti_mdt->mdt_namespace,
                                  &lh->mlh_reg_lh, LCK_EX, policy, res,
                                  dlmflags, NULL);

+       } else {
+               /* do nothing if both are local */
+               return 0;

        }
 
        if (rc != ELDLM_OK)
        }
 
        if (rc != ELDLM_OK)


@@ -1702,7 +1706,6 @@ static int mdt_migrate_object_lock(struct mdt_thread_info *info,
        int rc;
 
        if (mdt_object_remote(obj)) {
        int rc;
 
        if (mdt_object_remote(obj)) {

-               /* don't bother to check if pobj and obj are on the same MDT. */

                rc = mdt_revoke_remote_lookup_lock(info, pobj, obj);
                if (rc)
                        return rc;
                rc = mdt_revoke_remote_lookup_lock(info, pobj, obj);
                if (rc)
                        return rc;

diff --git a/lustre/mdt/mdt_xattr.c b/lustre/mdt/mdt_xattr.c
index 5101eab..e5f70e3 100644 (file)
--- a/lustre/mdt/mdt_xattr.c
+++ b/lustre/mdt/mdt_xattr.c
@@ -346,14 +346,44 @@ static int mdt_dir_layout_shrink(struct mdt_thread_info *info)
        if (IS_ERR(obj))
                RETURN(PTR_ERR(obj));
 
        if (IS_ERR(obj))
                RETURN(PTR_ERR(obj));
 

-relock:
+       /* get parent from PFID */
+       rc = mdt_attr_get_pfid(info, obj, &ma->ma_pfid);
+       if (rc)
+               GOTO(put_obj, rc);
+
+       pobj = mdt_object_find(env, mdt, &ma->ma_pfid);
+       if (IS_ERR(pobj))
+               GOTO(put_obj, rc = PTR_ERR(pobj));
+
+       /* revoke object remote LOOKUP lock */
+       if (mdt_object_remote(pobj)) {
+               rc = mdt_revoke_remote_lookup_lock(info, pobj, obj);
+               if (rc)
+                       GOTO(put_pobj, rc);
+       }
+
+       /*
+        * lock parent if dir will be shrunk to 1 stripe, because dir will be
+        * converted to normal directory, as will change dir fid and update
+        * namespace of parent.
+        */
+       lhp = &info->mti_lh[MDT_LH_PARENT];
+       mdt_lock_reg_init(lhp, LCK_PW);
+
+       if (le32_to_cpu(lmu->lum_stripe_count) < 2) {
+               rc = mdt_reint_object_lock(info, pobj, lhp,
+                                          MDS_INODELOCK_UPDATE, true);
+               if (rc)
+                       GOTO(put_pobj, rc);
+       }
+

        /* lock object */
        lhc = &info->mti_lh[MDT_LH_CHILD];
        mdt_lock_reg_init(lhc, LCK_EX);
        rc = mdt_reint_striped_lock(info, obj, lhc, MDS_INODELOCK_FULL, einfo,
                                    true);
        if (rc)
        /* lock object */
        lhc = &info->mti_lh[MDT_LH_CHILD];
        mdt_lock_reg_init(lhc, LCK_EX);
        rc = mdt_reint_striped_lock(info, obj, lhc, MDS_INODELOCK_FULL, einfo,
                                    true);
        if (rc)

-               GOTO(put_obj, rc);
+               GOTO(unlock_pobj, rc);

 
        ma->ma_lmv = info->mti_big_lmm;
        ma->ma_lmv_size = info->mti_big_lmmsize;
 
        ma->ma_lmv = info->mti_big_lmm;
        ma->ma_lmv_size = info->mti_big_lmmsize;


@@ -399,56 +429,6 @@ relock:
                GOTO(unlock_obj, rc = -EINVAL);
        }
 
                GOTO(unlock_obj, rc = -EINVAL);
        }
 

-       if (le32_to_cpu(lmu->lum_stripe_count) < 2 && !pobj) {
-               /*
-                * lock parent because dir will be shrunk to be 1 stripe, which
-                * should be converted to normal directory, but that will
-                * change dir fid and update namespace of parent.
-                */
-               lhp = &info->mti_lh[MDT_LH_PARENT];
-               mdt_lock_reg_init(lhp, LCK_PW);
-
-               /* get parent from PFID */
-               ma->ma_need |= MA_PFID;
-               ma->ma_valid = 0;
-               rc = mdt_attr_get_complex(info, obj, ma);
-               if (rc)
-                       GOTO(unlock_obj, rc);
-
-               if (!(ma->ma_valid & MA_PFID))
-                       GOTO(unlock_obj, rc = -ENOTSUPP);
-
-               pobj = mdt_object_find(env, mdt, &ma->ma_pfid);
-               if (IS_ERR(pobj)) {
-                       rc = PTR_ERR(pobj);
-                       pobj = NULL;
-                       GOTO(unlock_obj, rc);
-               }
-
-               mdt_reint_striped_unlock(info, obj, lhc, einfo, 1);
-
-               if (mdt_object_remote(pobj)) {
-                       rc = mdt_remote_object_lock(info, pobj, rr->rr_fid1,
-                                                   &lhp->mlh_rreg_lh, LCK_EX,
-                                                   MDS_INODELOCK_LOOKUP,
-                                                   false);
-                       if (rc != ELDLM_OK) {
-                               mdt_object_put(env, pobj);
-                               GOTO(put_obj, rc);
-                       }
-                       mdt_object_unlock(info, NULL, lhp, 1);
-               }
-
-               rc = mdt_reint_object_lock(info, pobj, lhp,
-                                          MDS_INODELOCK_UPDATE, true);
-               if (rc) {
-                       mdt_object_put(env, pobj);
-                       GOTO(put_obj, rc);
-               }
-
-               goto relock;
-       }
-

        buf->lb_buf = rr->rr_eadata;
        buf->lb_len = rr->rr_eadatalen;
        rc = mo_xattr_set(env, mdt_object_child(obj), buf, XATTR_NAME_LMV, 0);
        buf->lb_buf = rr->rr_eadata;
        buf->lb_len = rr->rr_eadatalen;
        rc = mo_xattr_set(env, mdt_object_child(obj), buf, XATTR_NAME_LMV, 0);


@@ -456,8 +436,10 @@ relock:
 
 unlock_obj:
        mdt_reint_striped_unlock(info, obj, lhc, einfo, rc);
 
 unlock_obj:
        mdt_reint_striped_unlock(info, obj, lhc, einfo, rc);

-       if (pobj)
-               mdt_object_unlock_put(info, pobj, lhp, rc);
+unlock_pobj:
+       mdt_object_unlock(info, pobj, lhp, rc);
+put_pobj:
+       mdt_object_put(env, pobj);

 put_obj:
        mdt_object_put(env, obj);
 
 put_obj:
        mdt_object_put(env, obj);