b=141,205,323

- When the MDS returns an intent lock to us, check to see if we already have a
  matching one.  If so, cancel the new one.
- Add a new mode to ldlm_lock_match, which allows you to pass in an existing
  lock handle.  It will look for any matches among other locks.

diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c
index 40dc2ff..5798d6e 100644 (file)
--- a/lustre/ldlm/ldlm_lock.c
+++ b/lustre/ldlm/ldlm_lock.c
@@ -561,7 +561,8 @@ void ldlm_grant_lock(struct ldlm_lock *lock)
 
 /* returns a referenced lock or NULL */
 static struct ldlm_lock *search_queue(struct list_head *queue, ldlm_mode_t mode,
-                                      struct ldlm_extent *extent)
+                                      struct ldlm_extent *extent,
+                                      struct ldlm_lock *old_lock)
 {
         struct ldlm_lock *lock;
         struct list_head *tmp;
@@ -569,6 +570,9 @@ static struct ldlm_lock *search_queue(struct list_head *queue, ldlm_mode_t mode,
         list_for_each(tmp, queue) {
                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
 
+                if (lock == old_lock)
+                        continue;
+
                 if (lock->l_flags & (LDLM_FL_CBPENDING | LDLM_FL_DESTROYED))
                         continue;
 
@@ -589,36 +593,53 @@ static struct ldlm_lock *search_queue(struct list_head *queue, ldlm_mode_t mode,
         return NULL;
 }
 
-/* Must be called with no resource or lock locks held.
+/* Can be called in two ways:
+ *
+ * If 'ns' is NULL, then lockh describes an existing lock that we want to look
+ * for a duplicate of.
+ *
+ * Otherwise, all of the fields must be filled in, to match against.
  *
  * Returns 1 if it finds an already-existing lock that is compatible; in this
  * case, lockh is filled in with a addref()ed lock
-*/
-int ldlm_lock_match(struct ldlm_namespace *ns, __u64 * res_id, __u32 type,
+ */
+int ldlm_lock_match(struct ldlm_namespace *ns, __u64 *res_id, __u32 type,
                     void *cookie, int cookielen, ldlm_mode_t mode,
                     struct lustre_handle *lockh)
 {
         struct ldlm_resource *res;
-        struct ldlm_lock *lock;
+        struct ldlm_lock *lock, *old_lock = NULL;
         int rc = 0;
         ENTRY;
 
+        if (ns == NULL) {
+                old_lock = ldlm_handle2lock(lockh);
+                LASSERT(old_lock);
+
+                ns = old_lock->l_resource->lr_namespace;
+                res_id = old_lock->l_resource->lr_name;
+                type = old_lock->l_resource->lr_type;
+                mode = old_lock->l_req_mode;
+        }
+
         res = ldlm_resource_get(ns, NULL, res_id, type, 0);
-        if (res == NULL)
+        if (res == NULL) {
+                LASSERT(old_lock == NULL);
                 RETURN(0);
+        }
 
         ns = res->lr_namespace;
         l_lock(&ns->ns_lock);
 
-        if ((lock = search_queue(&res->lr_granted, mode, cookie)))
+        if ((lock = search_queue(&res->lr_granted, mode, cookie, old_lock)))
                 GOTO(out, rc = 1);
-        if ((lock = search_queue(&res->lr_converting, mode, cookie)))
+        if ((lock = search_queue(&res->lr_converting, mode, cookie, old_lock)))
                 GOTO(out, rc = 1);
-        if ((lock = search_queue(&res->lr_waiting, mode, cookie)))
+        if ((lock = search_queue(&res->lr_waiting, mode, cookie, old_lock)))
                 GOTO(out, rc = 1);
 
         EXIT;
-      out:
+       out:
         ldlm_resource_put(res);
         l_unlock(&ns->ns_lock);
 
@@ -631,6 +652,10 @@ int ldlm_lock_match(struct ldlm_namespace *ns, __u64 * res_id, __u32 type,
                 LDLM_DEBUG(lock, "matched");
         else
                 LDLM_DEBUG_NOLOCK("not matched");
+
+        if (old_lock)
+                LDLM_LOCK_PUT(old_lock);
+
         return rc;
 }
 

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index 1270275..fb2f54a 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -749,7 +749,7 @@ EXPORT_SYMBOL(ldlm_register_intent);
 EXPORT_SYMBOL(ldlm_unregister_intent);
 EXPORT_SYMBOL(ldlm_lockname);
 EXPORT_SYMBOL(ldlm_typename);
-EXPORT_SYMBOL(ldlm_handle2lock);
+EXPORT_SYMBOL(__ldlm_handle2lock);
 EXPORT_SYMBOL(ldlm_lock2handle);
 EXPORT_SYMBOL(ldlm_lock_put);
 EXPORT_SYMBOL(ldlm_lock_match);

diff --git a/lustre/llite/namei.c b/lustre/llite/namei.c
index ec8dae8..5a8d882 100644 (file)
--- a/lustre/llite/namei.c
+++ b/lustre/llite/namei.c
@@ -308,6 +308,8 @@ int ll_intent_lock(struct inode *parent, struct dentry **de,
                 struct ll_inode_info *lli = ll_i2info(parent);
                 int mode;
 
+                LBUG(); /* For the moment, no non-intent locks */
+
                 /* it_disposition == 0 indicates that it just did a simple lock
                  * request, for which we are very thankful.  move along with
                  * the local lookup then. */

diff --git a/lustre/mdc/mdc_request.c b/lustre/mdc/mdc_request.c
index dd99059..f9a870b 100644 (file)
--- a/lustre/mdc/mdc_request.c
+++ b/lustre/mdc/mdc_request.c
@@ -373,7 +373,7 @@ int mdc_enqueue(struct lustre_handle *conn, int lock_type,
                 LBUG();
                 RETURN(-EINVAL);
         }
-#warning FIXME: the data here needs to be different if a lock was granted for a different inode
+
         rc = ldlm_cli_enqueue(conn, req, obddev->obd_namespace, NULL, res_id,
                               lock_type, NULL, 0, lock_mode, &flags,
                               ldlm_completion_ast, mdc_blocking_ast, data,
@@ -396,6 +396,25 @@ int mdc_enqueue(struct lustre_handle *conn, int lock_type,
         } else if (rc != 0) {
                 CERROR("ldlm_cli_enqueue: %d\n", rc);
                 RETURN(rc);
+        } else {
+                /* The server almost certainly gave us a lock other than the one
+                 * that we asked for.  If we already have a matching lock, then
+                 * cancel this one--we don't need two. */
+                struct ldlm_lock *lock = ldlm_handle2lock(lockh);
+                struct lustre_handle lockh2;
+                LASSERT(lock);
+
+                LDLM_DEBUG(lock, "matching against this");
+
+                memcpy(&lockh2, lockh, sizeof(lockh2));
+                if (ldlm_lock_match(NULL, NULL, LDLM_PLAIN, NULL, 0, LCK_NL,
+                                    &lockh2)) {
+                        /* We already have a lock; cancel the old one */
+                        ldlm_lock_decref(lockh, lock_mode);
+                        ldlm_cli_cancel(lockh);
+                        memcpy(lockh, &lockh2, sizeof(lockh2));
+                }
+                LDLM_LOCK_PUT(lock);
         }
 
         /* On replay, we don't want the lock granted. */