CLIO posts ENQUEUE requests asynchronously through ptlrpcd---a case that stock

ldlm_completion_ast() cannot handle as it waits until lock is
granted. Introduce new ldlm_completion_ast_async() for this. Also comment
ldlm_completion_ast().
b=16450

diff --git a/lustre/ChangeLog b/lustre/ChangeLog
index d1c47dc..7ba9637 100644 (file)
--- a/lustre/ChangeLog
+++ b/lustre/ChangeLog
@@ -1520,6 +1520,14 @@ Details    : CLIO uses lock weighting policy to keep locks over mmapped regions
             in memory---a requirement implemented through ->o_join_lru() obd
             method in HEAD. Remove this method and its users.
 
+Severity   : normal
+Bugzilla   : 16450
+Description: Add asynchronous ldlm ENQUEUE completion handler.
+Details    : CLIO posts ENQUEUE requests asynchronously through ptlrpcd---a
+            case that stock ldlm_completion_ast() cannot handle as it waits
+            until lock is granted. Introduce new ldlm_completion_ast_async()
+            for this. Also comment ldlm_completion_ast().
+       
 --------------------------------------------------------------------------------
 
 2007-08-10         Cluster File Systems, Inc. <info@clusterfs.com>

diff --git a/lustre/include/lustre_dlm.h b/lustre/include/lustre_dlm.h
index 487f690..b7a4117 100644 (file)
--- a/lustre/include/lustre_dlm.h
+++ b/lustre/include/lustre_dlm.h
@@ -980,6 +980,7 @@ int ldlm_expired_completion_wait(void *data);
 int ldlm_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
                       void *data, int flag);
 int ldlm_glimpse_ast(struct ldlm_lock *lock, void *reqp);
+int ldlm_completion_ast_async(struct ldlm_lock *lock, int flags, void *data);
 int ldlm_completion_ast(struct ldlm_lock *lock, int flags, void *data);
 int ldlm_cli_enqueue(struct obd_export *exp, struct ptlrpc_request **reqp,
                      struct ldlm_enqueue_info *einfo,

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index 41553e3..8a7acd5 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -2380,6 +2380,7 @@ EXPORT_SYMBOL(ldlm_reprocess_all_ns);
 EXPORT_SYMBOL(ldlm_lock_allow_match);
 
 /* ldlm_request.c */
+EXPORT_SYMBOL(ldlm_completion_ast_async);
 EXPORT_SYMBOL(ldlm_completion_ast);
 EXPORT_SYMBOL(ldlm_blocking_ast);
 EXPORT_SYMBOL(ldlm_glimpse_ast);

diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c
index 6dfeac7..6e08c02 100644 (file)
--- a/lustre/ldlm/ldlm_request.c
+++ b/lustre/ldlm/ldlm_request.c
@@ -135,6 +135,78 @@ static int is_granted_or_cancelled(struct ldlm_lock *lock)
         return ret;
 }
 
+/**
+ * Helper function for ldlm_completion_ast(), updating timings when lock is
+ * actually granted.
+ */
+static int ldlm_completion_tail(struct ldlm_lock *lock)
+{
+        long delay;
+        int  result;
+
+        if (lock->l_destroyed || lock->l_flags & LDLM_FL_FAILED) {
+                LDLM_DEBUG(lock, "client-side enqueue: destroyed");
+                result = -EIO;
+        } else {
+                delay = cfs_time_sub(cfs_time_current_sec(),
+                                     lock->l_enqueued_time.tv_sec);
+                LDLM_DEBUG(lock, "client-side enqueue: granted after "
+                           CFS_DURATION_T"s", delay);
+
+                /* Update our time estimate */
+                at_add(&lock->l_resource->lr_namespace->ns_at_estimate, delay);
+                result = 0;
+        }
+        return result;
+}
+
+/**
+ * Implementation of ->l_completion_ast() for a client that doesn't wait
+ * until lock is granted. Suitable for locks enqueued through ptlrpcd or
+ * other threads that cannot block for long.
+ */
+int ldlm_completion_ast_async(struct ldlm_lock *lock, int flags, void *data)
+{
+        ENTRY;
+
+        if (flags == LDLM_FL_WAIT_NOREPROC) {
+                LDLM_DEBUG(lock, "client-side enqueue waiting on pending lock");
+                RETURN(0);
+        }
+
+        if (!(flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
+                       LDLM_FL_BLOCK_CONV))) {
+                cfs_waitq_signal(&lock->l_waitq);
+                RETURN(ldlm_completion_tail(lock));
+        }
+
+        LDLM_DEBUG(lock, "client-side enqueue returned a blocked lock, "
+                   "going forward");
+        ldlm_lock_dump(D_OTHER, lock, 0);
+        RETURN(0);
+}
+
+/**
+ * Client side LDLM "completion" AST. This is called in several cases:
+ *
+ *     - when a reply to an ENQUEUE rpc is received from the server
+ *       (ldlm_cli_enqueue_fini()). Lock might be granted or not granted at
+ *       this point (determined by flags);
+ *
+ *     - when LDLM_CP_CALLBACK rpc comes to client to notify it that lock has
+ *       been granted;
+ *
+ *     - when ldlm_lock_match(LDLM_FL_LVB_READY) is about to wait until lock
+ *       gets correct lvb;
+ *
+ *     - to force all locks when resource is destroyed (cleanup_resource());
+ *
+ *     - during lock conversion (not used currently).
+ *
+ * If lock is not granted in the first case, this function waits until second
+ * or penultimate cases happen in some other thread.
+ *
+ */
 int ldlm_completion_ast(struct ldlm_lock *lock, int flags, void *data)
 {
         /* XXX ALLOCATE - 160 bytes */
@@ -160,7 +232,6 @@ int ldlm_completion_ast(struct ldlm_lock *lock, int flags, void *data)
         LDLM_DEBUG(lock, "client-side enqueue returned a blocked lock, "
                    "sleeping");
         ldlm_lock_dump(D_OTHER, lock, 0);
-        ldlm_reprocess_all(lock->l_resource);
 
 noreproc:
 
@@ -196,27 +267,13 @@ noreproc:
         /* Go to sleep until the lock is granted or cancelled. */
         rc = l_wait_event(lock->l_waitq, is_granted_or_cancelled(lock), &lwi);
 
-        if (lock->l_destroyed || lock->l_flags & LDLM_FL_FAILED) {
-                LDLM_DEBUG(lock, "client-side enqueue waking up: destroyed");
-                RETURN(-EIO);
-        }
-
         if (rc) {
                 LDLM_DEBUG(lock, "client-side enqueue waking up: failed (%d)",
                            rc);
                 RETURN(rc);
         }
 
-        LDLM_DEBUG(lock, "client-side enqueue waking up: granted after "
-                   CFS_DURATION_T"s",
-                   cfs_time_sub(cfs_time_current_sec(),
-                   lock->l_enqueued_time.tv_sec));
-
-        /* Update our time estimate */
-        at_add(&lock->l_resource->lr_namespace->ns_at_estimate,
-               cfs_time_current_sec() - lock->l_enqueued_time.tv_sec);
-
-        RETURN(0);
+        RETURN(ldlm_completion_tail(lock));
 }
 
 /*