Branch HEAD

b=17046
i=johann
i=oleg.drokin (green)

Reconstruct flock completion ast to make it in line with ldlm_completion_ast().

diff --git a/lustre/ldlm/ldlm_flock.c b/lustre/ldlm/ldlm_flock.c
index 1089022..439a4f4 100644 (file)
--- a/lustre/ldlm/ldlm_flock.c
+++ b/lustre/ldlm/ldlm_flock.c
@@ -108,7 +108,8 @@ ldlm_flock_destroy(struct ldlm_lock *lock, ldlm_mode_t mode, int flags)
         LASSERT(list_empty(&lock->l_flock_waitq));
 
         list_del_init(&lock->l_res_link);
-        if (flags == LDLM_FL_WAIT_NOREPROC) {
+        if (flags == LDLM_FL_WAIT_NOREPROC &&
+            !(lock->l_flags & LDLM_FL_FAILED)) {
                 /* client side - set a flag to prevent sending a CANCEL */
                 lock->l_flags |= LDLM_FL_LOCAL_ONLY | LDLM_FL_CBPENDING;
 
@@ -378,7 +379,8 @@ reprocess:
                                         NULL, 0);
                         lock_res_and_lock(req);
                         if (!new2) {
-                                ldlm_flock_destroy(req, lock->l_granted_mode, *flags);
+                                ldlm_flock_destroy(req, lock->l_granted_mode,
+                                                   *flags);
                                 *err = -ENOLCK;
                                 RETURN(LDLM_ITER_STOP);
                         }
@@ -399,14 +401,15 @@ reprocess:
                 new2->l_conn_export = lock->l_conn_export;
                 if (lock->l_export != NULL) {
                         new2->l_export = class_export_get(lock->l_export);
-                        if (new2->l_export->exp_lock_hash && 
+                        if (new2->l_export->exp_lock_hash &&
                             hlist_unhashed(&new2->l_exp_hash))
                                 lustre_hash_add(new2->l_export->exp_lock_hash,
                                                 &new2->l_remote_handle,
                                                 &new2->l_exp_hash);
                 }
                 if (*flags == LDLM_FL_WAIT_NOREPROC)
-                        ldlm_lock_addref_internal_nolock(new2, lock->l_granted_mode);
+                        ldlm_lock_addref_internal_nolock(new2,
+                                                         lock->l_granted_mode);
 
                 /* insert new2 at lock */
                 ldlm_resource_add_lock(res, ownlocks, new2);
@@ -502,17 +505,26 @@ ldlm_flock_interrupted_wait(void *data)
         EXIT;
 }
 
+/**
+ * Flock completion calback function.
+ *
+ * \param lock [in,out]: A lock to be handled
+ * \param flags    [in]: flags
+ * \param *data    [in]: ldlm_work_cp_ast_lock() will use ldlm_cb_set_arg
+ *
+ * \retval 0    : success
+ * \retval <0   : failure
+ */
 int
 ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data)
 {
-        struct ldlm_namespace *ns;
-        cfs_flock_t *getlk = lock->l_ast_data;
-        struct ldlm_flock_wait_data fwd;
-        struct obd_device *obd;
-        struct obd_import *imp = NULL;
-        ldlm_error_t err;
-        int rc = 0;
-        struct l_wait_info lwi;
+        cfs_flock_t                    *getlk = lock->l_ast_data;
+        struct obd_device              *obd;
+        struct obd_import              *imp = NULL;
+        struct ldlm_flock_wait_data     fwd;
+        struct l_wait_info              lwi;
+        ldlm_error_t                    err;
+        int                             rc = 0;
         ENTRY;
 
         CDEBUG(D_DLMTRACE, "flags: 0x%x data: %p getlk: %p\n",
@@ -527,7 +539,8 @@ ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data)
             (LDLM_FL_FAILED|LDLM_FL_LOCAL_ONLY)) {
                 unlock_res_and_lock(lock);
                 if (lock->l_req_mode == lock->l_granted_mode &&
-                    lock->l_granted_mode != LCK_NL)
+                    lock->l_granted_mode != LCK_NL &&
+                    NULL == data)
                         ldlm_lock_decref_internal(lock, lock->l_req_mode);
                 RETURN(0);
         }
@@ -536,20 +549,25 @@ ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data)
         LASSERT(flags != LDLM_FL_WAIT_NOREPROC);
 
         if (!(flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
-                       LDLM_FL_BLOCK_CONV)))
-                goto  granted;
+                       LDLM_FL_BLOCK_CONV))) {
+                if (NULL == data)
+                        /* mds granted the lock in the reply */
+                        goto granted;
+                /* CP AST RPC: lock get granted, wake it up */
+                cfs_waitq_signal(&lock->l_waitq);
+                RETURN(0);
+        }
 
         LDLM_DEBUG(lock, "client-side enqueue returned a blocked lock, "
                    "sleeping");
-
         fwd.fwd_lock = lock;
         obd = class_exp2obd(lock->l_conn_export);
 
-        /* if this is a local lock, then there is no import */
-        if (obd != NULL)
+        /* if this is a local lock, there is no import */
+        if (NULL != obd)
                 imp = obd->u.cli.cl_import;
 
-        if (imp != NULL) {
+        if (NULL != imp) {
                 spin_lock(&imp->imp_lock);
                 fwd.fwd_generation = imp->imp_generation;
                 spin_unlock(&imp->imp_lock);
@@ -558,27 +576,31 @@ ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data)
         lwi = LWI_TIMEOUT_INTR(0, NULL, ldlm_flock_interrupted_wait, &fwd);
 
         /* Go to sleep until the lock is granted. */
-        rc = l_wait_event(lock->l_waitq,
-                          ((lock->l_req_mode == lock->l_granted_mode) ||
-                           lock->l_destroyed), &lwi);
+        rc = l_wait_event(lock->l_waitq, is_granted_or_cancelled(lock), &lwi);
 
-        LDLM_DEBUG(lock, "client-side enqueue waking up: rc = %d", rc);
-        RETURN(rc);
+        if (rc) {
+                LDLM_DEBUG(lock, "client-side enqueue waking up: failed (%d)",
+                           rc);
+                RETURN(rc);
+        }
 
 granted:
         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_CP_CB_WAIT, 10);
-        LDLM_DEBUG(lock, "client-side enqueue granted");
-        ns = lock->l_resource->lr_namespace;
-        lock_res_and_lock(lock);
 
-        /* before flock's complete ast gets here, the flock
-         * can possibly be freed by another thread
-         */
-        if (lock->l_destroyed) {
-                LDLM_DEBUG(lock, "already destroyed by another thread");
+        lock_res_and_lock(lock);
+        if (lock->l_destroyed || lock->l_flags & LDLM_FL_FAILED) {
+                LDLM_DEBUG(lock, "client-side enqueue waking up: destroyed");
                 unlock_res(lock->l_resource);
-                RETURN(0);
+                RETURN(-EIO);
         }
+        if (rc) {
+                LDLM_DEBUG(lock, "client-side enqueue waking up: failed (%d)",
+                           rc);
+                unlock_res_and_lock(lock);
+                RETURN(rc);
+        }
+
+        LDLM_DEBUG(lock, "client-side enqueue granted");
 
         /* take lock off the deadlock detection waitq. */
         spin_lock(&ldlm_flock_waitq_lock);
@@ -591,8 +613,9 @@ granted:
         if (flags & LDLM_FL_TEST_LOCK) {
                 /* fcntl(F_GETLK) request */
                 /* The old mode was saved in getlk->fl_type so that if the mode
-                 * in the lock changes we can decref the approprate refcount. */
-                ldlm_flock_destroy(lock, cfs_flock_type(getlk), LDLM_FL_WAIT_NOREPROC);
+                 * in the lock changes we can decref the appropriate refcount.*/
+                ldlm_flock_destroy(lock, cfs_flock_type(getlk),
+                                   LDLM_FL_WAIT_NOREPROC);
                 switch (lock->l_granted_mode) {
                 case LCK_PR:
                         cfs_flock_set_type(getlk, F_RDLCK);
@@ -603,17 +626,18 @@ granted:
                 default:
                         cfs_flock_set_type(getlk, F_UNLCK);
                 }
-                cfs_flock_set_pid(getlk, (pid_t)lock->l_policy_data.l_flock.pid);
-                cfs_flock_set_start(getlk, (loff_t)lock->l_policy_data.l_flock.start);
-                cfs_flock_set_end(getlk, (loff_t)lock->l_policy_data.l_flock.end);
+                cfs_flock_set_pid(getlk,
+                                  (pid_t)lock->l_policy_data.l_flock.pid);
+                cfs_flock_set_start(getlk,
+                                   (loff_t)lock->l_policy_data.l_flock.start);
+                cfs_flock_set_end(getlk,
+                                  (loff_t)lock->l_policy_data.l_flock.end);
         } else {
                 int noreproc = LDLM_FL_WAIT_NOREPROC;
 
                 /* We need to reprocess the lock to do merges or splits
                  * with existing locks owned by this process. */
                 ldlm_process_flock_lock(lock, &noreproc, 1, &err, NULL);
-                if (flags == 0)
-                        cfs_waitq_signal(&lock->l_waitq);
         }
         unlock_res_and_lock(lock);
         RETURN(0);

diff --git a/lustre/ldlm/ldlm_internal.h b/lustre/ldlm/ldlm_internal.h
index 5332b2f..5ff07e7 100644 (file)
--- a/lustre/ldlm/ldlm_internal.h
+++ b/lustre/ldlm/ldlm_internal.h
@@ -239,3 +239,17 @@ typedef enum ldlm_policy_res ldlm_policy_res_t;
                 return rc;                                                  \
         }                                                                   \
         struct __##var##__dummy_write {;} /* semicolon catcher */
+
+static inline int is_granted_or_cancelled(struct ldlm_lock *lock)
+{
+        int ret = 0;
+
+        lock_res_and_lock(lock);
+        if (((lock->l_req_mode == lock->l_granted_mode) &&
+             !(lock->l_flags & LDLM_FL_CP_REQD)) ||
+            (lock->l_flags & LDLM_FL_FAILED))
+                ret = 1;
+        unlock_res_and_lock(lock);
+
+        return ret;
+}

diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c
index 7b84a27..46e7607 100644 (file)
--- a/lustre/ldlm/ldlm_request.c
+++ b/lustre/ldlm/ldlm_request.c
@@ -122,20 +122,6 @@ int ldlm_get_enq_timeout(struct ldlm_lock *lock)
 }
 EXPORT_SYMBOL(ldlm_get_enq_timeout);
 
-static int is_granted_or_cancelled(struct ldlm_lock *lock)
-{
-        int ret = 0;
-
-        lock_res_and_lock(lock);
-        if (((lock->l_req_mode == lock->l_granted_mode) &&
-             !(lock->l_flags & LDLM_FL_CP_REQD)) ||
-            (lock->l_flags & LDLM_FL_FAILED))
-                ret = 1;
-        unlock_res_and_lock(lock);
-
-        return ret;
-}
-
 /**
  * Helper function for ldlm_completion_ast(), updating timings when lock is
  * actually granted.
@@ -266,14 +252,14 @@ noreproc:
                 spin_unlock(&imp->imp_lock);
         }
 
-        if (ns_is_client(lock->l_resource->lr_namespace) && 
+        if (ns_is_client(lock->l_resource->lr_namespace) &&
             OBD_FAIL_CHECK_RESET(OBD_FAIL_LDLM_INTR_CP_AST,
                                  OBD_FAIL_LDLM_CP_BL_RACE | OBD_FAIL_ONCE)) {
                 lock->l_flags |= LDLM_FL_FAIL_LOC;
                 rc = -EINTR;
         } else {
                 /* Go to sleep until the lock is granted or cancelled. */
-                rc = l_wait_event(lock->l_waitq, 
+                rc = l_wait_event(lock->l_waitq,
                                   is_granted_or_cancelled(lock), &lwi);
         }
 
@@ -460,20 +446,20 @@ static void failed_lock_cleanup(struct ldlm_namespace *ns,
         /* Set a flag to prevent us from sending a CANCEL (bug 407) */
         lock_res_and_lock(lock);
         /* Check that lock is not granted or failed, we might race. */
-        if ((lock->l_req_mode != lock->l_granted_mode) && 
+        if ((lock->l_req_mode != lock->l_granted_mode) &&
             !(lock->l_flags & LDLM_FL_FAILED)) {
                 /* Make sure that this lock will not be found by raced
-                 * bl_ast and -EINVAL reply is sent to server anyways. 
+                 * bl_ast and -EINVAL reply is sent to server anyways.
                  * bug 17645 */
-                lock->l_flags |= LDLM_FL_LOCAL_ONLY | LDLM_FL_FAILED | 
+                lock->l_flags |= LDLM_FL_LOCAL_ONLY | LDLM_FL_FAILED |
                                  LDLM_FL_ATOMIC_CB;
                 need_cancel = 1;
         }
         unlock_res_and_lock(lock);
-  
+
         if (need_cancel) {
-                LDLM_DEBUG(lock, 
-                           "setting FL_LOCAL_ONLY | LDLM_FL_FAILED | " 
+                LDLM_DEBUG(lock,
+                           "setting FL_LOCAL_ONLY | LDLM_FL_FAILED | "
                            "LDLM_FL_ATOMIC_CB");
                 ldlm_lock_decref_and_cancel(lockh, mode);
         } else {