Branch b1_6

b=14425
i=oleg
i=bzzz

ldlm_completion_ast() assumes that a lock is granted when the req
mode is equal to the granted mode. However, it should also check
that LDLM_FL_CP_REQD is not set.

diff --git a/lustre/ChangeLog b/lustre/ChangeLog
index 9e3f552..9b0325d 100644 (file)
--- a/lustre/ChangeLog
+++ b/lustre/ChangeLog
@@ -237,6 +237,13 @@ Details    : qunit size will be changed when quota limitation is too low/high;
              this patch is landed. This bug also contains 14526, 14299, 14601
              and 13794, which are found and landed during v1_4_12.
 
+Severity   : normal
+Bugzilla   : 14225
+Description: LDLM_ENQUEUE races with LDLM_CP_CALLBACK
+Details    : ldlm_completion_ast() assumes that a lock is granted when the req
+            mode is equal to the granted mode. However, it should also check
+            that LDLM_FL_CP_REQD is not set.
+
 --------------------------------------------------------------------------------
 
 2007-12-07         Cluster File Systems, Inc. <info@clusterfs.com>

diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c
index 977ff9a..f5749ba 100644 (file)
--- a/lustre/ldlm/ldlm_lock.c
+++ b/lustre/ldlm/ldlm_lock.c
@@ -1367,16 +1367,20 @@ int ldlm_run_cp_ast_work(struct list_head *rpc_list)
         list_for_each_safe(tmp, pos, rpc_list) {
                 struct ldlm_lock *lock =
                         list_entry(tmp, struct ldlm_lock, l_cp_ast);
+                ldlm_completion_callback completion_callback;
 
                 /* nobody should touch l_cp_ast */
                 lock_res_and_lock(lock);
                 list_del_init(&lock->l_cp_ast);
                 LASSERT(lock->l_flags & LDLM_FL_CP_REQD);
+                /* save l_completion_ast since it can be changed by
+                 * mds_intent_policy(), see bug 14225 */
+                completion_callback = lock->l_completion_ast;
                 lock->l_flags &= ~LDLM_FL_CP_REQD;
                 unlock_res_and_lock(lock);
 
-                if (lock->l_completion_ast != NULL) {
-                        rc = lock->l_completion_ast(lock, 0, (void *)&arg);
+                if (completion_callback != NULL) {
+                        rc = completion_callback(lock, 0, (void *)&arg);
                         ast_count++;
                 }
                 LDLM_LOCK_PUT(lock);

diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c
index b2c48d8..699af94 100644 (file)
--- a/lustre/ldlm/ldlm_request.c
+++ b/lustre/ldlm/ldlm_request.c
@@ -103,6 +103,20 @@ int ldlm_get_enq_timeout(struct ldlm_lock *lock)
         return max(timeout, ldlm_enqueue_min);
 }
 
+static int is_granted_or_cancelled(struct ldlm_lock *lock)
+{
+        int ret = 0;
+
+        lock_res_and_lock(lock);
+        if (((lock->l_req_mode == lock->l_granted_mode) &&
+             !(lock->l_flags & LDLM_FL_CP_REQD)) ||
+            (lock->l_flags & LDLM_FL_FAILED))
+                ret = 1;
+        unlock_res_and_lock(lock);
+
+        return ret;
+}
+
 int ldlm_completion_ast(struct ldlm_lock *lock, int flags, void *data)
 {
         /* XXX ALLOCATE - 160 bytes */
@@ -162,9 +176,7 @@ noreproc:
         }
 
         /* Go to sleep until the lock is granted or cancelled. */
-        rc = l_wait_event(lock->l_waitq,
-                          ((lock->l_req_mode == lock->l_granted_mode) ||
-                           (lock->l_flags & LDLM_FL_FAILED)), &lwi);
+        rc = l_wait_event(lock->l_waitq, is_granted_or_cancelled(lock), &lwi);
 
         if (lock->l_destroyed || lock->l_flags & LDLM_FL_FAILED) {
                 LDLM_DEBUG(lock, "client-side enqueue waking up: destroyed");