This commit contains probably 92% of the striping 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 1b055ca..eddf484 100644 (file)
--- a/lustre/ldlm/ldlm_request.c
+++ b/lustre/ldlm/ldlm_request.c
@@ -83,7 +83,7 @@ int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
         if (rc != ELDLM_OK) {
                 LDLM_DEBUG(lock, "client-side enqueue END (%s)",
                            rc == ELDLM_LOCK_ABORTED ? "ABORTED" : "FAILED");
-                ldlm_lock_put(lock);
+                LDLM_LOCK_PUT(lock);
                 ldlm_lock_decref(lockh, mode);
                 /* FIXME: if we've already received a completion AST, this will
                  * LBUG! */
@@ -107,7 +107,7 @@ int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
         /* If enqueue returned a blocked lock but the completion handler has
          * already run, then it fixed up the resource and we don't need to do it
          * again. */
-        if (*flags & LDLM_FL_LOCK_CHANGED &&
+        if ((*flags) & LDLM_FL_LOCK_CHANGED &&
             lock->l_req_mode != lock->l_granted_mode) {
                 CDEBUG(D_INFO, "remote intent success, locking %ld instead of"
                        "%ld\n", (long)reply->lock_resource_name[0],
@@ -125,7 +125,7 @@ int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
                 ptlrpc_free_req(req);
 
         rc = ldlm_lock_enqueue(lock, cookie, cookielen, flags, callback,
-                                     callback);
+                               callback);
 
         if (*flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
                       LDLM_FL_BLOCK_CONV)) {
@@ -134,17 +134,49 @@ int ldlm_cli_enqueue(struct ptlrpc_client *cl, struct ptlrpc_connection *conn,
                 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);
+#warning ldlm needs to time out
+                wait_event(lock->l_waitq,
+                           lock->l_req_mode == lock->l_granted_mode);
                 LDLM_DEBUG(lock, "client-side enqueue waking up: granted");
         }
         LDLM_DEBUG(lock, "client-side enqueue END");
-        ldlm_lock_put(lock);
+        LDLM_LOCK_PUT(lock);
         EXIT;
  out:
         return rc;
 }
 
+int ldlm_match_or_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,
+                          __u64 *res_id,
+                          __u32 type,
+                          void *cookie, int cookielen,
+                          ldlm_mode_t mode,
+                          int *flags,
+                          ldlm_lock_callback callback,
+                          void *data,
+                          __u32 data_len,
+                          struct lustre_handle *lockh)
+{
+        int rc;
+        ENTRY;
+        rc = ldlm_lock_match(ns, res_id, type, cookie, cookielen, mode, lockh);
+        if (rc == 0) {
+                rc = ldlm_cli_enqueue(cl, conn, connh, req, ns,
+                                      parent_lock_handle, res_id, type, cookie,
+                                      cookielen, mode, flags, callback, data,
+                                      data_len, lockh);
+                if (rc != ELDLM_OK)
+                        CERROR("ldlm_cli_enqueue: err: %d\n", rc);
+                RETURN(rc);
+        } else
+                RETURN(0);
+}
+
 int ldlm_server_ast(struct lustre_handle *lockh, struct ldlm_lock_desc *desc,
                     void *data, __u32 data_len)
 {
@@ -156,8 +188,10 @@ int ldlm_server_ast(struct lustre_handle *lockh, struct ldlm_lock_desc *desc,
         ENTRY;
 
         lock = ldlm_handle2lock(lockh);
-        if (lock == NULL)
+        if (lock == NULL) {
                 LBUG();
+                RETURN(-EINVAL);
+        }
         cl = &lock->l_resource->lr_namespace->ns_rpc_client;
         req = ptlrpc_prep_req(cl, lock->l_connection, LDLM_CALLBACK, 1,
                               &size, NULL);
@@ -187,7 +221,7 @@ int ldlm_server_ast(struct lustre_handle *lockh, struct ldlm_lock_desc *desc,
 
         EXIT;
  out:
-        ldlm_lock_put(lock);
+        LDLM_LOCK_PUT(lock);
         return rc;
 }
 
@@ -204,8 +238,10 @@ int ldlm_cli_convert(struct ptlrpc_client *cl, struct lustre_handle *lockh,
         ENTRY;
 
         lock = ldlm_handle2lock(lockh);
-        if (!lock)
-                LBUG(); 
+        if (!lock) {
+                LBUG();
+                RETURN(-EINVAL);
+        }
         *flags = 0;
 
         LDLM_DEBUG(lock, "client-side convert");
@@ -239,19 +275,18 @@ int ldlm_cli_convert(struct ptlrpc_client *cl, struct lustre_handle *lockh,
                 /* FIXME: or cancelled. */
                 CDEBUG(D_NET, "convert returned a blocked lock, "
                        "going to sleep.\n");
-                wait_event_interruptible(lock->l_waitq, lock->l_req_mode ==
-                                         lock->l_granted_mode);
+                wait_event(lock->l_waitq,
+                           lock->l_req_mode == lock->l_granted_mode);
                 CDEBUG(D_NET, "waking up, the lock must be granted.\n");
         }
-        ldlm_lock_put(lock);
+        LDLM_LOCK_PUT(lock);
         EXIT;
  out:
         ptlrpc_free_req(req);
         return rc;
 }
 
-int ldlm_cli_cancel(struct lustre_handle *lockh, 
-                    struct lustre_handle *connh)
+int ldlm_cli_cancel(struct lustre_handle *lockh)
 {
         struct ptlrpc_request *req;
         struct ldlm_lock *lock;
@@ -260,8 +295,15 @@ int ldlm_cli_cancel(struct lustre_handle *lockh,
         ENTRY;
 
         lock = ldlm_handle2lock(lockh); 
-        if (!lock)
-                LBUG();
+        if (!lock) {
+                /* It's possible that the decref that we did just before this
+                 * cancel was the last reader/writer, and caused a cancel before
+                 * we could call this function.  If we want to make this
+                 * impossible (by adding a dec_and_cancel() or similar), then
+                 * we can put the LBUG back. */
+                //LBUG();
+                RETURN(-EINVAL);
+        }
 
         LDLM_DEBUG(lock, "client-side cancel");
         req = ptlrpc_prep_req(lock->l_client, lock->l_connection,
@@ -282,7 +324,7 @@ int ldlm_cli_cancel(struct lustre_handle *lockh,
                 GOTO(out, rc);
 
         ldlm_lock_cancel(lock);
-        ldlm_lock_put(lock); 
+        LDLM_LOCK_PUT(lock); 
         EXIT;
  out:
         return 0;