- fixes to let a client do correct lock cancellation during namespace

  free.

diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c
index 4a6e264..8087e6a 100644 (file)
--- a/lustre/ldlm/ldlm_lock.c
+++ b/lustre/ldlm/ldlm_lock.c
@@ -21,6 +21,7 @@
 
 /* lock types */
 char *ldlm_lockname[] = {
+        [0]      "--",
         [LCK_EX] "EX", 
         [LCK_PW] "PW",
         [LCK_PR] "PR",
@@ -28,7 +29,6 @@ char *ldlm_lockname[] = {
         [LCK_CR] "CR",
         [LCK_NL] "NL"
 };
-
 char *ldlm_typename[] = {
         [LDLM_PLAIN]     "PLN",
         [LDLM_EXTENT]    "EXT",
@@ -55,41 +55,6 @@ ldlm_res_policy ldlm_res_policy_table [] = {
         [LDLM_MDSINTENT] ldlm_intent_policy
 };
 
-void ldlm_lock2handle(struct ldlm_lock *lock, struct lustre_handle *lockh)
-{
-        lockh->addr = (__u64)(unsigned long)lock;
-        lockh->cookie = lock->l_random;
-}
-
-/* 
- *  HANDLES
- */ 
-
-struct ldlm_lock *ldlm_handle2lock(struct lustre_handle *handle)
-{
-        struct ldlm_lock *lock = NULL;
-        ENTRY;
-
-        if (!handle || !handle->addr)
-                RETURN(NULL);
-
-        lock = (struct ldlm_lock *)(unsigned long)(handle->addr);
-        if (!kmem_cache_validate(ldlm_lock_slab, (void *)lock))
-                RETURN(NULL);
-
-        l_lock(&lock->l_resource->lr_namespace->ns_lock);
-        if (lock->l_random != handle->cookie)
-                GOTO(out, handle = NULL);
-
-        if (lock->l_flags & LDLM_FL_DESTROYED)
-                GOTO(out, handle = NULL);
-
-        lock->l_refc++;
-        EXIT;
- out:
-        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
-        return  lock;
-}
 
 /*
  * REFCOUNTED LOCK OBJECTS
@@ -157,14 +122,12 @@ void ldlm_lock_destroy(struct ldlm_lock *lock)
 
         if (lock->l_flags & LDLM_FL_DESTROYED) {
                 EXIT;
-                ldlm_lock_put(lock);
                 return;
         }
 
         lock->l_flags = LDLM_FL_DESTROYED;
         l_unlock(&lock->l_resource->lr_namespace->ns_lock);
         ldlm_lock_put(lock);
-        ldlm_lock_put(lock);
         EXIT;
         return;
 }
@@ -241,6 +204,45 @@ int ldlm_lock_change_resource(struct ldlm_lock *lock, __u64 new_resid[3])
         RETURN(0);
 }
 
+/* 
+ *  HANDLES
+ */ 
+
+void ldlm_lock2handle(struct ldlm_lock *lock, struct lustre_handle *lockh)
+{
+        lockh->addr = (__u64)(unsigned long)lock;
+        lockh->cookie = lock->l_random;
+}
+
+
+struct ldlm_lock *ldlm_handle2lock(struct lustre_handle *handle)
+{
+        struct ldlm_lock *lock = NULL;
+        ENTRY;
+
+        if (!handle || !handle->addr)
+                RETURN(NULL);
+
+        lock = (struct ldlm_lock *)(unsigned long)(handle->addr);
+        if (!kmem_cache_validate(ldlm_lock_slab, (void *)lock))
+                RETURN(NULL);
+
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
+        if (lock->l_random != handle->cookie)
+                GOTO(out, handle = NULL);
+
+        if (lock->l_flags & LDLM_FL_DESTROYED)
+                GOTO(out, handle = NULL);
+
+        ldlm_lock_get(lock);
+        EXIT;
+ out:
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
+        return  lock;
+}
+
+
+
 static int ldlm_intent_policy(struct ldlm_lock *lock, void *req_cookie,
                               ldlm_mode_t mode, void *data)
 {
@@ -706,6 +708,7 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_lock *lock,
                         *flags |= LDLM_FL_LOCK_CHANGED;
                 } else if (rc == ELDLM_LOCK_ABORTED) {
                         ldlm_lock_destroy(lock);
+                        ldlm_lock_put(lock);
                         RETURN(rc);
                 }
         }

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index 645d94c..3b4370b 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -459,6 +459,7 @@ static void __exit ldlm_exit(void)
 
 EXPORT_SYMBOL(ldlm_lockname);
 EXPORT_SYMBOL(ldlm_typename);
+EXPORT_SYMBOL(ldlm_handle2lock);
 EXPORT_SYMBOL(ldlm_lock_match);
 EXPORT_SYMBOL(ldlm_lock_addref);
 EXPORT_SYMBOL(ldlm_lock_decref);

diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c
index aedff87..c7d44e3 100644 (file)
--- a/lustre/ldlm/ldlm_request.c
+++ b/lustre/ldlm/ldlm_request.c
@@ -255,7 +255,6 @@ int ldlm_cli_cancel(struct lustre_handle *lockh)
         struct ptlrpc_request *req;
         struct ldlm_lock *lock;
         struct ldlm_request *body;
-        struct ldlm_resource *res;
         int rc, size = sizeof(*body);
         ENTRY;
 
@@ -282,7 +281,7 @@ int ldlm_cli_cancel(struct lustre_handle *lockh)
                 GOTO(out, rc);
 
         ldlm_lock_cancel(lock);
-        ldlm_reprocess_all(res);
+        ldlm_reprocess_all(lock->l_resource);
         ldlm_lock_put(lock); 
         EXIT;
  out:

diff --git a/lustre/ldlm/ldlm_resource.c b/lustre/ldlm/ldlm_resource.c
index 7d2288c..a92454d 100644 (file)
--- a/lustre/ldlm/ldlm_resource.c
+++ b/lustre/ldlm/ldlm_resource.c
@@ -62,7 +62,9 @@ struct ldlm_namespace *ldlm_namespace_new(char *name, __u32 client)
         return NULL;
 }
 
-static int cleanup_resource(struct ldlm_resource *res, struct list_head *q)
+extern struct ldlm_lock *ldlm_lock_get(struct ldlm_lock *lock);
+
+static void cleanup_resource(struct ldlm_resource *res, struct list_head *q)
 {
         struct list_head *tmp, *pos;
         int rc = 0, client = res->lr_namespace->ns_client;
@@ -71,6 +73,7 @@ static int cleanup_resource(struct ldlm_resource *res, struct list_head *q)
         list_for_each_safe(tmp, pos, q) {
                 struct ldlm_lock *lock;
                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
+                ldlm_lock_get(lock);
 
                 if (client) {
                         struct lustre_handle lockh;
@@ -80,25 +83,22 @@ static int cleanup_resource(struct ldlm_resource *res, struct list_head *q)
                                 CERROR("ldlm_cli_cancel: %d\n", rc);
                                 LBUG();
                         }
-                        if (rc == ELDLM_RESOURCE_FREED)
-                                rc = 1;
                 } else {
                         CERROR("Freeing a lock still held by a client node.\n");
 
                         ldlm_resource_unlink_lock(lock);
                         ldlm_lock_destroy(lock);
-
-                        rc = ldlm_resource_put(res);
                 }
+                ldlm_lock_put(lock);
         }
 
-        RETURN(rc);
+        return; 
 }
 
 int ldlm_namespace_free(struct ldlm_namespace *ns)
 {
         struct list_head *tmp, *pos;
-        int i, rc;
+        int i;
 
         if (!ns)
                 RETURN(ELDLM_OK);
@@ -109,20 +109,19 @@ int ldlm_namespace_free(struct ldlm_namespace *ns)
                 list_for_each_safe(tmp, pos, &(ns->ns_hash[i])) {
                         struct ldlm_resource *res;
                         res = list_entry(tmp, struct ldlm_resource, lr_hash);
+                        ldlm_resource_getref(res);
 
-                        rc = cleanup_resource(res, &res->lr_granted);
-                        if (!rc)
-                                rc = cleanup_resource(res, &res->lr_converting);
-                        if (!rc)
-                                rc = cleanup_resource(res, &res->lr_waiting);
-
-                        if (rc == 0) {
+                        cleanup_resource(res, &res->lr_granted);
+                        cleanup_resource(res, &res->lr_converting);
+                        cleanup_resource(res, &res->lr_waiting);
+                        
+                        if (!ldlm_resource_put(res)) { 
                                 CERROR("Resource refcount nonzero (%d) after "
                                        "lock cleanup; forcing cleanup.\n",
                                        atomic_read(&res->lr_refcount));
                                 ldlm_resource_dump(res);
                                 atomic_set(&res->lr_refcount, 1);
-                                rc = ldlm_resource_put(res);
+                                ldlm_resource_put(res);
                         }
                 }
         }

diff --git a/lustre/mds/mds_reint.c b/lustre/mds/mds_reint.c
index 8753fad..0b536c6 100644 (file)
--- a/lustre/mds/mds_reint.c
+++ b/lustre/mds/mds_reint.c
@@ -229,7 +229,7 @@ static int mds_reint_create(struct mds_update_record *rec, int offset,
                         GOTO(out_create_de, rc = -EIO);
                 }
         } else {
-                lock = lustre_handle2object(&lockh);
+                lock = ldlm_handle2lock(&lockh);
                 LDLM_DEBUG(lock, "matched");
         }
         ldlm_lock_dump((void *)(unsigned long)lockh.addr);