LU-1715 ptlrpc: flock deadlock detection does not work

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

diff --git a/lustre/ldlm/ldlm_flock.c b/lustre/ldlm/ldlm_flock.c
index f59c354..c141438 100644 (file)
--- a/lustre/ldlm/ldlm_flock.c
+++ b/lustre/ldlm/ldlm_flock.c
@@ -100,20 +100,12 @@ ldlm_flocks_overlap(struct ldlm_lock *lock, struct ldlm_lock *new)
                 lock->l_policy_data.l_flock.start));
 }
 
-static inline int ldlm_flock_blocking_link(struct ldlm_lock *req,
-                                          struct ldlm_lock *lock)
+static inline void ldlm_flock_blocking_link(struct ldlm_lock *req,
+                                           struct ldlm_lock *lock)
 {
-       int rc = 0;
-
         /* For server only */
         if (req->l_export == NULL)
-               return 0;
-
-       if (unlikely(req->l_export->exp_flock_hash == NULL)) {
-               rc = ldlm_init_flock_export(req->l_export);
-               if (rc)
-                       goto error;
-       }
+               return;
 
        LASSERT(cfs_hlist_unhashed(&req->l_exp_flock_hash));
 
@@ -126,8 +118,6 @@ static inline int ldlm_flock_blocking_link(struct ldlm_lock *req,
        cfs_hash_add(req->l_export->exp_flock_hash,
                     &req->l_policy_data.l_flock.owner,
                     &req->l_exp_flock_hash);
-error:
-       return rc;
 }
 
 static inline void ldlm_flock_blocking_unlink(struct ldlm_lock *req)
@@ -203,6 +193,7 @@ ldlm_flock_deadlock(struct ldlm_lock *req, struct ldlm_lock *bl_lock)
                if (lock == NULL)
                        break;
 
+               LASSERT(req != lock);
                flock = &lock->l_policy_data.l_flock;
                LASSERT(flock->owner == bl_owner);
                 bl_owner = flock->blocking_owner;
@@ -222,6 +213,26 @@ ldlm_flock_deadlock(struct ldlm_lock *req, struct ldlm_lock *bl_lock)
         return 0;
 }
 
+static void ldlm_flock_cancel_on_deadlock(struct ldlm_lock *lock,
+                                                cfs_list_t *work_list)
+{
+       CDEBUG(D_INFO, "reprocess deadlock req=%p\n", lock);
+
+       if ((exp_connect_flags(lock->l_export) &
+                               OBD_CONNECT_FLOCK_DEAD) == 0) {
+               CERROR("deadlock found, but client doesn't "
+                               "support flock canceliation\n");
+       } else {
+               LASSERT(lock->l_completion_ast);
+               LASSERT((lock->l_flags & LDLM_FL_AST_SENT) == 0);
+               lock->l_flags |= LDLM_FL_AST_SENT | LDLM_FL_CANCEL_ON_BLOCK |
+                       LDLM_FL_FLOCK_DEADLOCK;
+               ldlm_flock_blocking_unlink(lock);
+               ldlm_resource_unlink_lock(lock);
+               ldlm_add_ast_work_item(lock, NULL, work_list);
+       }
+}
+
 /**
  * Process a granting attempt for flock lock.
  * Must be called under ns lock held.
@@ -257,7 +268,6 @@ ldlm_process_flock_lock(struct ldlm_lock *req, __u64 *flags, int first_enq,
         int overlaps = 0;
         int splitted = 0;
         const struct ldlm_callback_suite null_cbs = { NULL };
-       int rc;
         ENTRY;
 
        CDEBUG(D_DLMTRACE, "flags %#llx owner "LPU64" pid %u mode %u start "
@@ -291,6 +301,7 @@ reprocess:
                         }
                 }
         } else {
+               int reprocess_failed = 0;
                 lockmode_verify(mode);
 
                 /* This loop determines if there are existing locks
@@ -312,8 +323,15 @@ reprocess:
                         if (!ldlm_flocks_overlap(lock, req))
                                 continue;
 
-                        if (!first_enq)
-                                RETURN(LDLM_ITER_CONTINUE);
+                       if (!first_enq) {
+                               reprocess_failed = 1;
+                               if (ldlm_flock_deadlock(req, lock)) {
+                                       ldlm_flock_cancel_on_deadlock(req,
+                                                       work_list);
+                                       RETURN(LDLM_ITER_CONTINUE);
+                               }
+                               continue;
+                       }
 
                         if (*flags & LDLM_FL_BLOCK_NOWAIT) {
                                 ldlm_flock_destroy(req, mode, *flags);
@@ -334,22 +352,23 @@ reprocess:
                                 RETURN(LDLM_ITER_STOP);
                         }
 
-                        if (ldlm_flock_deadlock(req, lock)) {
-                                ldlm_flock_destroy(req, mode, *flags);
-                                *err = -EDEADLK;
-                                RETURN(LDLM_ITER_STOP);
-                        }
+                       /* add lock to blocking list before deadlock
+                        * check to prevent race */
+                       ldlm_flock_blocking_link(req, lock);
 
-                       rc = ldlm_flock_blocking_link(req, lock);
-                       if (rc) {
+                       if (ldlm_flock_deadlock(req, lock)) {
+                               ldlm_flock_blocking_unlink(req);
                                ldlm_flock_destroy(req, mode, *flags);
-                               *err = rc;
+                               *err = -EDEADLK;
                                RETURN(LDLM_ITER_STOP);
                        }
+
                         ldlm_resource_add_lock(res, &res->lr_waiting, req);
                         *flags |= LDLM_FL_BLOCK_GRANTED;
                         RETURN(LDLM_ITER_STOP);
                 }
+               if (reprocess_failed)
+                       RETURN(LDLM_ITER_CONTINUE);
         }
 
         if (*flags & LDLM_FL_TEST_LOCK) {
@@ -675,10 +694,10 @@ ldlm_flock_completion_ast(struct ldlm_lock *lock, __u64 flags, void *data)
 granted:
         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_CP_CB_WAIT, 10);
 
-        if (lock->l_destroyed) {
-                LDLM_DEBUG(lock, "client-side enqueue waking up: destroyed");
-                RETURN(0);
-        }
+       if (lock->l_flags & LDLM_FL_DESTROYED) {
+               LDLM_DEBUG(lock, "client-side enqueue waking up: destroyed");
+               RETURN(0);
+       }
 
         if (lock->l_flags & LDLM_FL_FAILED) {
                 LDLM_DEBUG(lock, "client-side enqueue waking up: failed");
@@ -701,7 +720,10 @@ granted:
         /* ldlm_lock_enqueue() has already placed lock on the granted list. */
         cfs_list_del_init(&lock->l_res_link);
 
-        if (flags & LDLM_FL_TEST_LOCK) {
+       if (lock->l_flags & LDLM_FL_FLOCK_DEADLOCK) {
+               LDLM_DEBUG(lock, "client-side enqueue deadlock received");
+               rc = -EDEADLK;
+       } else if (flags & LDLM_FL_TEST_LOCK) {
                 /* fcntl(F_GETLK) request */
                 /* The old mode was saved in getlk->fl_type so that if the mode
                  * in the lock changes we can decref the appropriate refcount.*/
@@ -730,7 +752,7 @@ granted:
                ldlm_process_flock_lock(lock, &noreproc, 1, &err, NULL);
        }
        unlock_res_and_lock(lock);
-       RETURN(0);
+       RETURN(rc);
 }
 EXPORT_SYMBOL(ldlm_flock_completion_ast);
 
@@ -858,6 +880,9 @@ static cfs_hash_ops_t ldlm_export_flock_ops = {
 
 int ldlm_init_flock_export(struct obd_export *exp)
 {
+       if( strcmp(exp->exp_obd->obd_type->typ_name, LUSTRE_MDT_NAME) != 0)
+               RETURN(0);
+
        exp->exp_flock_hash =
                cfs_hash_create(obd_uuid2str(&exp->exp_client_uuid),
                                HASH_EXP_LOCK_CUR_BITS,