- more LDLM refcount locking infrastructure

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

diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c
index 2c463c5..1b055ca 100644 (file)
--- a/lustre/ldlm/ldlm_request.c
+++ b/lustre/ldlm/ldlm_request.c
@@ -14,6 +14,7 @@
 #include <linux/lustre_dlm.h>
 
 int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
+                     struct lustre_handle *connh, 
                      struct ptlrpc_request *req,
                      struct ldlm_namespace *ns,
                      struct lustre_handle *parent_lock_handle,
@@ -38,14 +39,14 @@ int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
                                 data, data_len);
         if (lock == NULL)
                 GOTO(out, rc = -ENOMEM);
+        LDLM_DEBUG(lock, "client-side enqueue START");
         /* for the local lock, add the reference */
         ldlm_lock_addref_internal(lock, mode);
         ldlm_lock2handle(lock, lockh);
 
-        LDLM_DEBUG(lock, "client-side enqueue START");
-
         if (req == NULL) {
-                req = ptlrpc_prep_req(cl, conn, LDLM_ENQUEUE, 1, &size, NULL);
+                req = ptlrpc_prep_req2(cl, conn, connh, 
+                                       LDLM_ENQUEUE, 1, &size, NULL);
                 if (!req)
                         GOTO(out, rc = -ENOMEM);
                 req_passed_in = 0;
@@ -61,7 +62,7 @@ int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
                        sizeof(body->lock_desc.l_extent));
         body->lock_flags = *flags;
 
-        ldlm_lock2handle(lock, &body->lock_handle1);
+        memcpy(&body->lock_handle1, lockh, sizeof(*lockh));
         if (parent_lock_handle)
                 memcpy(&body->lock_handle2, parent_lock_handle,
                        sizeof(body->lock_handle2));
@@ -72,7 +73,7 @@ int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
                 req->rq_replen = lustre_msg_size(1, &size);
         }
 
-        lock->l_connection = conn;
+        lock->l_connection = ptlrpc_connection_addref(conn);
         lock->l_client = cl;
 
         rc = ptlrpc_queue_wait(req);
@@ -83,6 +84,10 @@ int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
                 LDLM_DEBUG(lock, "client-side enqueue END (%s)",
                            rc == ELDLM_LOCK_ABORTED ? "ABORTED" : "FAILED");
                 ldlm_lock_put(lock);
+                ldlm_lock_decref(lockh, mode);
+                /* FIXME: if we've already received a completion AST, this will
+                 * LBUG! */
+                ldlm_lock_destroy(lock);
                 GOTO(out, rc);
         }
 
@@ -122,18 +127,18 @@ int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
         rc = ldlm_lock_enqueue(lock, cookie, cookielen, flags, callback,
                                      callback);
 
-        LDLM_DEBUG(lock, "client-side enqueue END");
         if (*flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
                       LDLM_FL_BLOCK_CONV)) {
                 /* Go to sleep until the lock is granted. */
                 /* FIXME: or cancelled. */
-                CDEBUG(D_NET, "enqueue returned a blocked lock (%p), "
-                       "going to sleep.\n", lock);
+                LDLM_DEBUG(lock, "client-side enqueue returned a blocked lock,"
+                           " sleeping");
                 ldlm_lock_dump(lock);
                 wait_event_interruptible(lock->l_waitq, lock->l_req_mode ==
                                          lock->l_granted_mode);
-                CDEBUG(D_NET, "waking up, the lock must be granted.\n");
+                LDLM_DEBUG(lock, "client-side enqueue waking up: granted");
         }
+        LDLM_DEBUG(lock, "client-side enqueue END");
         ldlm_lock_put(lock);
         EXIT;
  out:
@@ -187,6 +192,7 @@ int ldlm_server_ast(struct lustre_handle *lockh, struct ldlm_lock_desc *desc,
 }
 
 int ldlm_cli_convert(struct ptlrpc_client *cl, struct lustre_handle *lockh,
+                     struct lustre_handle *connh, 
                      int new_mode, int *flags)
 {
         struct ldlm_request *body;
@@ -204,8 +210,8 @@ int ldlm_cli_convert(struct ptlrpc_client *cl, struct lustre_handle *lockh,
 
         LDLM_DEBUG(lock, "client-side convert");
 
-        req = ptlrpc_prep_req(cl, lock->l_connection, LDLM_CONVERT, 1, &size,
-                              NULL);
+        req = ptlrpc_prep_req(cl, lock->l_connection,
+                               LDLM_CONVERT, 1, &size, NULL);
         if (!req)
                 GOTO(out, rc = -ENOMEM);
 
@@ -244,7 +250,8 @@ int ldlm_cli_convert(struct ptlrpc_client *cl, struct lustre_handle *lockh,
         return rc;
 }
 
-int ldlm_cli_cancel(struct lustre_handle *lockh)
+int ldlm_cli_cancel(struct lustre_handle *lockh, 
+                    struct lustre_handle *connh)
 {
         struct ptlrpc_request *req;
         struct ldlm_lock *lock;
@@ -257,8 +264,8 @@ int ldlm_cli_cancel(struct lustre_handle *lockh)
                 LBUG();
 
         LDLM_DEBUG(lock, "client-side cancel");
-        req = ptlrpc_prep_req(lock->l_client, lock->l_connection, LDLM_CANCEL,
-                              1, &size, NULL);
+        req = ptlrpc_prep_req(lock->l_client, lock->l_connection,
+                              LDLM_CANCEL, 1, &size, NULL);
         if (!req)
                 GOTO(out, rc = -ENOMEM);