Land b1_2 onto HEAD (20040304_171022)

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

diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c
index bb0c0c1..0e7f0b0 100644 (file)
--- a/lustre/ldlm/ldlm_lock.c
+++ b/lustre/ldlm/ldlm_lock.c
@@ -430,9 +430,9 @@ void ldlm_lock_addref_internal(struct ldlm_lock *lock, __u32 mode)
 {
         l_lock(&lock->l_resource->lr_namespace->ns_lock);
         ldlm_lock_remove_from_lru(lock);
-        if (mode == LCK_NL || mode == LCK_CR || mode == LCK_PR)
+        if (mode & (LCK_NL | LCK_CR | LCK_PR))
                 lock->l_readers++;
-        else
+        if (mode & (LCK_EX | LCK_CW | LCK_PW))
                 lock->l_writers++;
         lock->l_last_used = jiffies;
         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
@@ -448,10 +448,11 @@ void ldlm_lock_decref_internal(struct ldlm_lock *lock, __u32 mode)
         LDLM_DEBUG(lock, "ldlm_lock_decref(%s)", ldlm_lockname[mode]);
         ns = lock->l_resource->lr_namespace;
         l_lock(&ns->ns_lock);
-        if (mode == LCK_NL || mode == LCK_CR || mode == LCK_PR) {
+        if (mode & (LCK_NL | LCK_CR | LCK_PR)) {
                 LASSERT(lock->l_readers > 0);
                 lock->l_readers--;
-        } else {
+        }
+        if (mode & (LCK_EX | LCK_CW | LCK_PW)) {
                 LASSERT(lock->l_writers > 0);
                 lock->l_writers--;
         }
@@ -473,13 +474,11 @@ void ldlm_lock_decref_internal(struct ldlm_lock *lock, __u32 mode)
                                "warning\n");
 
                 LDLM_DEBUG(lock, "final decref done on cbpending lock");
-                l_unlock(&ns->ns_lock);
 
-                l_check_no_ns_lock(ns);
-                /* FIXME: need a real 'desc' here */
-                if (lock->l_blocking_ast != NULL)
-                        lock->l_blocking_ast(lock, NULL, lock->l_ast_data,
-                                             LDLM_CB_BLOCKING);
+                LDLM_LOCK_GET(lock); /* dropped by bl thread */
+                ldlm_lock_remove_from_lru(lock);
+                ldlm_bl_to_thread(ns, NULL, lock);
+                l_unlock(&ns->ns_lock);
         } else if (ns->ns_client == LDLM_NAMESPACE_CLIENT &&
                    !lock->l_readers && !lock->l_writers) {
                 /* If this is a client-side namespace and this was the last
@@ -577,7 +576,7 @@ static struct ldlm_lock *search_queue(struct list_head *queue, ldlm_mode_t mode,
                     lock->l_readers == 0 && lock->l_writers == 0)
                         continue;
 
-                if (lock->l_req_mode != mode)
+                if (!(lock->l_req_mode & mode))
                         continue;
 
                 if (lock->l_resource->lr_type == LDLM_EXTENT &&
@@ -593,7 +592,10 @@ static struct ldlm_lock *search_queue(struct list_head *queue, ldlm_mode_t mode,
                     !(lock->l_flags & LDLM_FL_LOCAL))
                         continue;
 
-                ldlm_lock_addref_internal(lock, mode);
+                if (flags & LDLM_FL_TEST_LOCK)
+                        LDLM_LOCK_GET(lock);
+                else
+                        ldlm_lock_addref_internal(lock, mode);
                 return lock;
         }
 
@@ -622,6 +624,8 @@ void ldlm_lock_allow_match(struct ldlm_lock *lock)
  * If 'flags' contains LDLM_FL_CBPENDING, then locks that have been marked
  *     to be canceled can still be matched as long as they still have reader
  *     or writer refernces
+ * If 'flags' contains LDLM_FL_TEST_LOCK, then don't actually reference a lock,
+ *     just tell us if we would have matched.
  *
  * Returns 1 if it finds an already-existing lock that is compatible; in this
  * case, lockh is filled in with a addref()ed lock
@@ -672,25 +676,40 @@ int ldlm_lock_match(struct ldlm_namespace *ns, int flags,
         l_unlock(&ns->ns_lock);
 
         if (lock) {
-                struct l_wait_info lwi;
                 ldlm_lock2handle(lock, lockh);
-                if (lock->l_completion_ast)
-                        lock->l_completion_ast(lock, LDLM_FL_WAIT_NOREPROC,
-                                               NULL);
+                if (!(lock->l_flags & LDLM_FL_CAN_MATCH)) {
+                        struct l_wait_info lwi;
+                        if (lock->l_completion_ast)
+                                lock->l_completion_ast(lock,
+                                                       LDLM_FL_WAIT_NOREPROC,
+                                                       NULL);
 
-                lwi = LWI_TIMEOUT_INTR(obd_timeout * HZ, NULL, NULL, NULL);
+                        lwi = LWI_TIMEOUT_INTR(obd_timeout*HZ, NULL,NULL,NULL);
 
-                /* XXX FIXME see comment about CAN_MATCH in lustre_dlm.h */
-                l_wait_event(lock->l_waitq,
-                             (lock->l_flags & LDLM_FL_CAN_MATCH), &lwi);
+                        /* XXX FIXME see comment on CAN_MATCH in lustre_dlm.h */
+                        l_wait_event(lock->l_waitq,
+                                     (lock->l_flags & LDLM_FL_CAN_MATCH), &lwi);
+                }
         }
         if (rc)
-                LDLM_DEBUG(lock, "matched");
-        else
-                LDLM_DEBUG_NOLOCK("not matched");
+                LDLM_DEBUG(lock, "matched ("LPU64" "LPU64")",
+                           type == LDLM_PLAIN ? res_id->name[2] :
+                                policy->l_extent.start,
+                           type == LDLM_PLAIN ? res_id->name[3] :
+                                policy->l_extent.end);
+        else if (!(flags & LDLM_FL_TEST_LOCK)) /* less verbose for test-only */
+                LDLM_DEBUG_NOLOCK("not matched ns %p type %u mode %u res "
+                                  LPU64"/"LPU64" ("LPU64" "LPU64")", ns,
+                                  type, mode, res_id->name[0], res_id->name[1],
+                                  type == LDLM_PLAIN ? res_id->name[2] :
+                                        policy->l_extent.start,
+                                  type == LDLM_PLAIN ? res_id->name[3] :
+                                        policy->l_extent.end);
 
         if (old_lock)
                 LDLM_LOCK_PUT(old_lock);
+        if (flags & LDLM_FL_TEST_LOCK && rc)
+                LDLM_LOCK_PUT(lock);
 
         return rc;
 }
@@ -1041,7 +1060,7 @@ void ldlm_cancel_locks_for_export(struct obd_export *exp)
         struct ldlm_resource *res;
 
         l_lock(&ns->ns_lock);
-        while(!list_empty(&exp->exp_ldlm_data.led_held_locks)) {
+        while(!list_empty(&exp->exp_ldlm_data.led_held_locks)) { 
                 lock = list_entry(exp->exp_ldlm_data.led_held_locks.next,
                                   struct ldlm_lock, l_export_chain);
                 res = ldlm_resource_getref(lock->l_resource);
@@ -1148,9 +1167,11 @@ void ldlm_lock_dump(int level, struct ldlm_lock *lock, int pos)
                "write: %d\n", (int)lock->l_req_mode, (int)lock->l_granted_mode,
                atomic_read(&lock->l_refc), lock->l_readers, lock->l_writers);
         if (lock->l_resource->lr_type == LDLM_EXTENT)
-                CDEBUG(level, "  Extent: "LPU64" -> "LPU64"\n",
+                CDEBUG(level, "  Extent: "LPU64" -> "LPU64
+                       " (req "LPU64"-"LPU64")\n",
                        lock->l_policy_data.l_extent.start,
-                       lock->l_policy_data.l_extent.end);
+                       lock->l_policy_data.l_extent.end,
+                       lock->l_req_extent.start, lock->l_req_extent.end);
         else if (lock->l_resource->lr_type == LDLM_FLOCK)
                 CDEBUG(level, "  Pid: %d Extent: "LPU64" -> "LPU64"\n",
                        lock->l_policy_data.l_flock.pid,