LU-7791 ldlm: signal vs CP callback race

[fs/lustre-release.git] / lustre / ldlm / ldlm_lockd.c

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index e208689..faf7545 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -1884,15 +1884,29 @@ void ldlm_handle_bl_callback(struct ldlm_namespace *ns,
        EXIT;
 }
 
+static int ldlm_callback_reply(struct ptlrpc_request *req, int rc)
+{
+       if (req->rq_no_reply)
+               return 0;
+
+       req->rq_status = rc;
+       if (!req->rq_packed_final) {
+               rc = lustre_pack_reply(req, 1, NULL, NULL);
+               if (rc)
+                       return rc;
+       }
+       return ptlrpc_reply(req);
+}
+
 /**
  * Callback handler for receiving incoming completion ASTs.
  *
  * This only can happen on client side.
  */
-static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
-                                   struct ldlm_namespace *ns,
-                                   struct ldlm_request *dlm_req,
-                                   struct ldlm_lock *lock)
+static int ldlm_handle_cp_callback(struct ptlrpc_request *req,
+                                    struct ldlm_namespace *ns,
+                                    struct ldlm_request *dlm_req,
+                                    struct ldlm_lock *lock)
 {
        struct list_head ast_list;
        int lvb_len;
@@ -1906,6 +1920,8 @@ static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
        if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE)) {
                long to = cfs_time_seconds(1);
 
+               ldlm_callback_reply(req, 0);
+
                while (to > 0) {
                        set_current_state(TASK_INTERRUPTIBLE);
                        schedule_timeout(to);
@@ -1949,6 +1965,12 @@ static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
                lock_res_and_lock(lock);
        }
 
+       if (ldlm_is_failed(lock)) {
+               unlock_res_and_lock(lock);
+               LDLM_LOCK_RELEASE(lock);
+               RETURN(-EINVAL);
+       }
+
        if (ldlm_is_destroyed(lock) ||
            ldlm_is_granted(lock)) {
                /* b=11300: the lock has already been granted */
@@ -2021,6 +2043,8 @@ out:
                wake_up(&lock->l_waitq);
        }
        LDLM_LOCK_RELEASE(lock);
+
+       return 0;
 }
 
 /**
@@ -2077,20 +2101,6 @@ static void ldlm_handle_gl_callback(struct ptlrpc_request *req,
        EXIT;
 }
 
-static int ldlm_callback_reply(struct ptlrpc_request *req, int rc)
-{
-       if (req->rq_no_reply)
-               return 0;
-
-       req->rq_status = rc;
-       if (!req->rq_packed_final) {
-               rc = lustre_pack_reply(req, 1, NULL, NULL);
-               if (rc)
-                       return rc;
-       }
-       return ptlrpc_reply(req);
-}
-
 static int __ldlm_bl_to_thread(struct ldlm_bl_work_item *blwi,
                               enum ldlm_cancel_flags cancel_flags)
 {
@@ -2426,8 +2436,9 @@ static int ldlm_callback_handler(struct ptlrpc_request *req)
        case LDLM_CP_CALLBACK:
                CDEBUG(D_INODE, "completion ast\n");
                req_capsule_extend(&req->rq_pill, &RQF_LDLM_CP_CALLBACK);
-               ldlm_callback_reply(req, 0);
-               ldlm_handle_cp_callback(req, ns, dlm_req, lock);
+               rc = ldlm_handle_cp_callback(req, ns, dlm_req, lock);
+               if (!OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE))
+                       ldlm_callback_reply(req, rc);
                break;
        case LDLM_GL_CALLBACK:
                CDEBUG(D_INODE, "glimpse ast\n");