Commented proc code

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

diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c
index c50bb47..3731f7d 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)
 {
@@ -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);
 
@@ -358,8 +360,13 @@ int ldlm_cli_cancel(struct lustre_handle *lockh)
 
         if (lock->l_connh) {
                 LDLM_DEBUG(lock, "client-side cancel");
-                req = ptlrpc_prep_req2(lock->l_connh, LDLM_CANCEL, 1, &size,
-                                       NULL);
+                /* 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;
+                l_unlock(&lock->l_resource->lr_namespace->ns_lock);
+
+                req = ptlrpc_prep_req(class_conn2cliimp(lock->l_connh), 
+                                      LDLM_CANCEL, 1, &size, NULL);
                 if (!req)
                         GOTO(out, rc = -ENOMEM);
 
@@ -392,3 +399,56 @@ int ldlm_cli_cancel(struct lustre_handle *lockh)
         LDLM_LOCK_PUT(lock);
         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)
+{
+        struct ldlm_resource *res;
+        struct list_head *tmp, *next, list = LIST_HEAD_INIT(list);
+        struct ldlm_ast_work *w;
+        ENTRY;
+
+        res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
+        if (res == NULL)
+                RETURN(-EINVAL);
+
+        l_lock(&ns->ns_lock);
+        list_for_each(tmp, &res->lr_granted) {
+                struct ldlm_lock *lock;
+                lock = list_entry(tmp, struct ldlm_lock, l_res_link);
+
+                if (lock->l_readers || lock->l_writers)
+                        continue;
+
+                /* Setting the CBPENDING flag is a little misleading, but
+                 * prevents an important race; namely, once CBPENDING is set,
+                 * the lock can accumulate no more readers/writers.  Since
+                 * readers and writers are already zero here, ldlm_lock_decref
+                 * won't see this flag and call l_blocking_ast */
+                lock->l_flags |= LDLM_FL_CBPENDING;
+
+                OBD_ALLOC(w, sizeof(*w));
+                LASSERT(w);
+
+                w->w_lock = LDLM_LOCK_GET(lock);
+                list_add(&w->w_list, &list);
+        }
+        l_unlock(&ns->ns_lock);
+
+        list_for_each_safe(tmp, next, &list) {
+                struct lustre_handle lockh;
+                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);
+
+                LDLM_LOCK_PUT(w->w_lock);
+                list_del(&w->w_list);
+                OBD_FREE(w, sizeof(*w));
+        }
+
+        RETURN(0);
+}