- Remove cancelled locks from the waiting list. Revoking locks from a dead

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

diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c
index cbf3aeb..ee4594a 100644 (file)
--- a/lustre/ldlm/ldlm_request.c
+++ b/lustre/ldlm/ldlm_request.c
@@ -12,6 +12,7 @@
 #define DEBUG_SUBSYSTEM S_LDLM
 
 #include <linux/lustre_dlm.h>
+#include <linux/obd.h>
 
 int ldlm_completion_ast(struct ldlm_lock *lock, int flags)
 {
@@ -59,7 +60,7 @@ static int ldlm_cli_enqueue_local(struct ldlm_namespace *ns,
                 LBUG();
         }
 
-        lock = ldlm_lock_create(ns, parent_lockh, res_id, type, mode, NULL, 0);
+        lock = ldlm_lock_create(ns, parent_lockh, res_id, type, mode, data, data_len);
         if (!lock)
                 GOTO(out_nolock, err = -ENOMEM);
         LDLM_DEBUG(lock, "client-side local enqueue handler, new lock created");
@@ -130,7 +131,8 @@ int ldlm_cli_enqueue(struct lustre_handle *connh,
         ldlm_lock2handle(lock, lockh);
 
         if (req == NULL) {
-                req = ptlrpc_prep_req2(connh, LDLM_ENQUEUE, 1, &size, NULL);
+                req = ptlrpc_prep_req(class_conn2cliimp(connh), LDLM_ENQUEUE, 1,
+                                      &size, NULL);
                 if (!req)
                         GOTO(out, rc = -ENOMEM);
                 req_passed_in = 0;
@@ -158,7 +160,6 @@ int ldlm_cli_enqueue(struct lustre_handle *connh,
         }
         lock->l_connh = connh;
         lock->l_export = NULL;
-        lock->l_client = client_conn2cli(connh)->cl_client;
 
         rc = ptlrpc_queue_wait(req);
         /* FIXME: status check here? */
@@ -303,7 +304,8 @@ int ldlm_cli_convert(struct lustre_handle *lockh, int new_mode, int *flags)
 
         LDLM_DEBUG(lock, "client-side convert");
 
-        req = ptlrpc_prep_req2(connh, LDLM_CONVERT, 1, &size, NULL);
+        req = ptlrpc_prep_req(class_conn2cliimp(connh), LDLM_CONVERT, 1, &size,
+                              NULL);
         if (!req)
                 GOTO(out, rc = -ENOMEM);
 
@@ -361,10 +363,11 @@ int ldlm_cli_cancel(struct lustre_handle *lockh)
                 /* Set this flag to prevent others from getting new references*/
                 l_lock(&lock->l_resource->lr_namespace->ns_lock);
                 lock->l_flags |= LDLM_FL_CBPENDING;
+                ldlm_cancel_callback(lock);
                 l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 
-                req = ptlrpc_prep_req2(lock->l_connh, LDLM_CANCEL, 1, &size,
-                                       NULL);
+                req = ptlrpc_prep_req(class_conn2cliimp(lock->l_connh), 
+                                      LDLM_CANCEL, 1, &size, NULL);
                 if (!req)
                         GOTO(out, rc = -ENOMEM);
 
@@ -398,8 +401,12 @@ int ldlm_cli_cancel(struct lustre_handle *lockh)
         return rc;
 }
 
-/* Cancel all locks on a given resource that have 0 readers/writers */
-int ldlm_cli_cancel_unused(struct ldlm_namespace *ns, __u64 *res_id)
+/* Cancel all locks on a given resource that have 0 readers/writers.
+ *
+ * If 'local_only' is true, throw the locks away without trying to notify the
+ * server. */
+int ldlm_cli_cancel_unused(struct ldlm_namespace *ns, __u64 *res_id,
+                           int local_only)
 {
         struct ldlm_resource *res;
         struct list_head *tmp, *next, list = LIST_HEAD_INIT(list);
@@ -408,7 +415,7 @@ int ldlm_cli_cancel_unused(struct ldlm_namespace *ns, __u64 *res_id)
 
         res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
         if (res == NULL)
-                RETURN(-ENOMEM);
+                RETURN(-EINVAL);
 
         l_lock(&ns->ns_lock);
         list_for_each(tmp, &res->lr_granted) {
@@ -438,15 +445,20 @@ int ldlm_cli_cancel_unused(struct ldlm_namespace *ns, __u64 *res_id)
                 int rc;
                 w = list_entry(tmp, struct ldlm_ast_work, w_list);
 
-                ldlm_lock2handle(w->w_lock, &lockh);
-                rc = ldlm_cli_cancel(&lockh);
-                if (rc != ELDLM_OK)
-                        CERROR("ldlm_cli_cancel: %d\n", rc);
-
+                if (local_only)
+                        ldlm_lock_cancel(w->w_lock);
+                else {
+                        ldlm_lock2handle(w->w_lock, &lockh);
+                        rc = ldlm_cli_cancel(&lockh);
+                        if (rc != ELDLM_OK)
+                                CERROR("ldlm_cli_cancel: %d\n", rc);
+                }
                 LDLM_LOCK_PUT(w->w_lock);
                 list_del(&w->w_list);
                 OBD_FREE(w, sizeof(*w));
         }
 
+        ldlm_resource_put(res);
+
         RETURN(0);
 }