LU-4403 mds: extra lock during resend lock lookup.

1. If the request does not require open lock, MDT does not
need to search the lock in exp_hash_lock list, because the MDT
will release the lock anyway, and it can always re-enqueue the
request during resend.

2. Lock the resource when the MDS is trying to lookup the lock for
resend request(in mdt_intent_fixup_resent). Otherwise, if the
lock is being released, fixup_resent will return a released lock,
which will cause LBUG when this lock is being released later.

Signed-off-by: wang di <di.wang@intel.com>
Change-Id: Ice5a75f9a4cc3bcb45f8d54cf809b4a272805804
Reviewed-on: http://review.whamcloud.com/8680
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Mike Pershin <mike.pershin@intel.com>
Reviewed-by: Fan Yong <fan.yong@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 1b09982..f882b4d 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -2939,9 +2939,10 @@ int mdt_intent_lock_replace(struct mdt_thread_info *info,
 }
 
 static void mdt_intent_fixup_resent(struct mdt_thread_info *info,
-                                    struct ldlm_lock *new_lock,
-                                    struct ldlm_lock **old_lock,
-                                    struct mdt_lock_handle *lh)
+                                   struct ldlm_lock *new_lock,
+                                   struct ldlm_lock **old_lock,
+                                   struct mdt_lock_handle *lh,
+                                   enum mdt_it_code opcode)
 {
         struct ptlrpc_request  *req = mdt_info_req(info);
         struct obd_export      *exp = req->rq_export;
@@ -2954,29 +2955,37 @@ static void mdt_intent_fixup_resent(struct mdt_thread_info *info,
 
         dlmreq = req_capsule_client_get(info->mti_pill, &RMF_DLM_REQ);
         remote_hdl = dlmreq->lock_handle[0];
-
-       /* In the function below, .hs_keycmp resolves to
-        * ldlm_export_lock_keycmp() */
-       /* coverity[overrun-buffer-val] */
-        lock = cfs_hash_lookup(exp->exp_lock_hash, &remote_hdl);
-        if (lock) {
-                if (lock != new_lock) {
-                        lh->mlh_reg_lh.cookie = lock->l_handle.h_cookie;
-                        lh->mlh_reg_mode = lock->l_granted_mode;
-
-                        LDLM_DEBUG(lock, "Restoring lock cookie");
-                        DEBUG_REQ(D_DLMTRACE, req,
-                                  "restoring lock cookie "LPX64,
-                                  lh->mlh_reg_lh.cookie);
-                        if (old_lock)
-                                *old_lock = LDLM_LOCK_GET(lock);
-                        cfs_hash_put(exp->exp_lock_hash, &lock->l_exp_hash);
-                        return;
-                }
-
-                cfs_hash_put(exp->exp_lock_hash, &lock->l_exp_hash);
-        }
-
+       /* If the client does not require open lock, it does not need to
+        * search lock in exp_lock_hash, since the server thread will
+        * make sure the lock will be released, and the resend request
+        * can always re-enqueue the lock */
+       if ((opcode != MDT_IT_OPEN) || (opcode == MDT_IT_OPEN &&
+           info->mti_spec.sp_cr_flags & MDS_OPEN_LOCK)) {
+               /* In the function below, .hs_keycmp resolves to
+                * ldlm_export_lock_keycmp() */
+               /* coverity[overrun-buffer-val] */
+               lock = cfs_hash_lookup(exp->exp_lock_hash, &remote_hdl);
+               if (lock) {
+                       lock_res_and_lock(lock);
+                       if (lock != new_lock) {
+                               lh->mlh_reg_lh.cookie = lock->l_handle.h_cookie;
+                               lh->mlh_reg_mode = lock->l_granted_mode;
+
+                               LDLM_DEBUG(lock, "Restoring lock cookie");
+                               DEBUG_REQ(D_DLMTRACE, req,
+                                         "restoring lock cookie "LPX64,
+                                         lh->mlh_reg_lh.cookie);
+                               if (old_lock)
+                                       *old_lock = LDLM_LOCK_GET(lock);
+                               cfs_hash_put(exp->exp_lock_hash,
+                                            &lock->l_exp_hash);
+                               unlock_res_and_lock(lock);
+                               return;
+                       }
+                       cfs_hash_put(exp->exp_lock_hash, &lock->l_exp_hash);
+                       unlock_res_and_lock(lock);
+               }
+       }
         /*
          * If the xid matches, then we know this is a resent request, and allow
          * it. (It's probably an OPEN, for which we don't send a lock.
@@ -3009,7 +3018,7 @@ static int mdt_intent_getxattr(enum mdt_it_code opcode,
         * (for the resend case) or a new lock. Below we will use it to
         * replace the original lock.
         */
-       mdt_intent_fixup_resent(info, *lockp, NULL, lhc);
+       mdt_intent_fixup_resent(info, *lockp, NULL, lhc, opcode);
        if (!lustre_handle_is_used(&lhc->mlh_reg_lh)) {
                mdt_lock_reg_init(lhc, (*lockp)->l_req_mode);
                rc = mdt_object_lock(info, info->mti_object, lhc,
@@ -3079,8 +3088,8 @@ static int mdt_intent_getattr(enum mdt_it_code opcode,
         ldlm_rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP);
         mdt_set_disposition(info, ldlm_rep, DISP_IT_EXECD);
 
-        /* Get lock from request for possible resent case. */
-        mdt_intent_fixup_resent(info, *lockp, &new_lock, lhc);
+       /* Get lock from request for possible resent case. */
+       mdt_intent_fixup_resent(info, *lockp, &new_lock, lhc, opcode);
 
        rc = mdt_getattr_name_lock(info, lhc, child_bits, ldlm_rep);
        ldlm_rep->lock_policy_res2 = clear_serious(rc);
@@ -3188,8 +3197,8 @@ static int mdt_intent_reint(enum mdt_it_code opcode,
                 RETURN(err_serious(-EPROTO));
         }
 
-        /* Get lock from request for possible resent case. */
-        mdt_intent_fixup_resent(info, *lockp, NULL, lhc);
+       /* Get lock from request for possible resent case. */
+       mdt_intent_fixup_resent(info, *lockp, NULL, lhc, opcode);
 
         rc = mdt_reint_internal(info, lhc, opc);