b=23518 add the missing logic of enqueue wait for conflicting lock.

r=di.wang
r=vitaly.fertman

diff --git a/lustre/include/cl_object.h b/lustre/include/cl_object.h
index 91b32c9..897faf1 100644 (file)
--- a/lustre/include/cl_object.h
+++ b/lustre/include/cl_object.h
@@ -2795,12 +2795,13 @@ int   cl_lock_user_del  (const struct lu_env *env, struct cl_lock *lock);
 
 enum cl_lock_state cl_lock_intransit(const struct lu_env *env,
                                      struct cl_lock *lock);
 
 enum cl_lock_state cl_lock_intransit(const struct lu_env *env,
                                      struct cl_lock *lock);

-

 void cl_lock_extransit(const struct lu_env *env, struct cl_lock *lock,
                        enum cl_lock_state state);
 void cl_lock_extransit(const struct lu_env *env, struct cl_lock *lock,
                        enum cl_lock_state state);

-

 int cl_lock_is_intransit(struct cl_lock *lock);
 
 int cl_lock_is_intransit(struct cl_lock *lock);
 

+int cl_lock_enqueue_wait(const struct lu_env *env, struct cl_lock *lock,
+                         int keep_mutex);
+

 /** \name statemachine statemachine
  * Interface to lock state machine consists of 3 parts:
  *
 /** \name statemachine statemachine
  * Interface to lock state machine consists of 3 parts:
  *


@@ -2842,6 +2843,7 @@ int   cl_enqueue_try(const struct lu_env *env, struct cl_lock *lock,
 int   cl_unuse_try  (const struct lu_env *env, struct cl_lock *lock);
 int   cl_wait_try   (const struct lu_env *env, struct cl_lock *lock);
 int   cl_use_try    (const struct lu_env *env, struct cl_lock *lock, int atomic);
 int   cl_unuse_try  (const struct lu_env *env, struct cl_lock *lock);
 int   cl_wait_try   (const struct lu_env *env, struct cl_lock *lock);
 int   cl_use_try    (const struct lu_env *env, struct cl_lock *lock, int atomic);

+

 /** @} statemachine */
 
 void cl_lock_signal      (const struct lu_env *env, struct cl_lock *lock);
 /** @} statemachine */
 
 void cl_lock_signal      (const struct lu_env *env, struct cl_lock *lock);

diff --git a/lustre/lov/lov_lock.c b/lustre/lov/lov_lock.c
index 6ef27ee..1c118f0 100644 (file)
--- a/lustre/lov/lov_lock.c
+++ b/lustre/lov/lov_lock.c
@@ -491,50 +491,20 @@ static void lov_lock_fini(const struct lu_env *env,
         EXIT;
 }
 
         EXIT;
 }
 

-/**
- *
- * \retval 0 if state-transition can proceed
- * \retval -ve otherwise.
- */

 static int lov_lock_enqueue_wait(const struct lu_env *env,
                                  struct lov_lock *lck,
                                  struct cl_lock *sublock)
 {
 static int lov_lock_enqueue_wait(const struct lu_env *env,
                                  struct lov_lock *lck,
                                  struct cl_lock *sublock)
 {

-        struct cl_lock *lock     = lck->lls_cl.cls_lock;
-        struct cl_lock *conflict = sublock->cll_conflict;
-        int result = CLO_REPEAT;
+        struct cl_lock *lock = lck->lls_cl.cls_lock;
+        int             result;

         ENTRY;
 
         LASSERT(cl_lock_is_mutexed(lock));
         ENTRY;
 
         LASSERT(cl_lock_is_mutexed(lock));

-        LASSERT(cl_lock_is_mutexed(sublock));
-        LASSERT(sublock->cll_state == CLS_QUEUING);
-        LASSERT(conflict != NULL);

 
 

-        sublock->cll_conflict = NULL;

         cl_lock_mutex_put(env, lock);
         cl_lock_mutex_put(env, lock);

-        cl_lock_mutex_put(env, sublock);
-
-        LASSERT(cl_lock_nr_mutexed(env) == 0);
-
-        cl_lock_mutex_get(env, conflict);
-        cl_lock_cancel(env, conflict);
-        cl_lock_delete(env, conflict);
-        while (conflict->cll_state != CLS_FREEING) {
-                int rc = 0;
-
-                rc = cl_lock_state_wait(env, conflict);
-                if (rc == 0)
-                        continue;
-
-                result = lov_subresult(result, rc);
-                break;
-        }
-        cl_lock_mutex_put(env, conflict);
-        lu_ref_del(&conflict->cll_reference, "cancel-wait", sublock);
-        cl_lock_put(env, conflict);
-
+        result = cl_lock_enqueue_wait(env, sublock, 0);

         cl_lock_mutex_get(env, lock);
         cl_lock_mutex_get(env, lock);

-        RETURN(result);
+        RETURN(result ?: CLO_REPEAT);

 }
 
 /**
 }
 
 /**

diff --git a/lustre/obdclass/cl_lock.c b/lustre/obdclass/cl_lock.c
index 5b3767b..77935bf 100644 (file)
--- a/lustre/obdclass/cl_lock.c
+++ b/lustre/obdclass/cl_lock.c
@@ -1225,6 +1225,51 @@ int cl_enqueue_try(const struct lu_env *env, struct cl_lock *lock,
 }
 EXPORT_SYMBOL(cl_enqueue_try);
 
 }
 EXPORT_SYMBOL(cl_enqueue_try);
 

+/**
+ * Cancel the conflicting lock found during previous enqueue.
+ *
+ * \retval 0 conflicting lock has been canceled.
+ * \retval -ve error code.
+ */
+int cl_lock_enqueue_wait(const struct lu_env *env,
+                         struct cl_lock *lock,
+                         int keep_mutex)
+{
+        struct cl_lock  *conflict;
+        int              rc = 0;
+        ENTRY;
+
+        LASSERT(cl_lock_is_mutexed(lock));
+        LASSERT(lock->cll_state == CLS_QUEUING);
+        LASSERT(lock->cll_conflict != NULL);
+
+        conflict = lock->cll_conflict;
+        lock->cll_conflict = NULL;
+
+        cl_lock_mutex_put(env, lock);
+        LASSERT(cl_lock_nr_mutexed(env) == 0);
+
+        cl_lock_mutex_get(env, conflict);
+        cl_lock_cancel(env, conflict);
+        cl_lock_delete(env, conflict);
+
+        while (conflict->cll_state != CLS_FREEING) {
+                rc = cl_lock_state_wait(env, conflict);
+                if (rc != 0)
+                        break;
+        }
+        cl_lock_mutex_put(env, conflict);
+        lu_ref_del(&conflict->cll_reference, "cancel-wait", lock);
+        cl_lock_put(env, conflict);
+
+        if (keep_mutex)
+                cl_lock_mutex_get(env, lock);
+
+        LASSERT(rc <= 0);
+        RETURN(rc);
+}
+EXPORT_SYMBOL(cl_lock_enqueue_wait);
+

 static int cl_enqueue_locked(const struct lu_env *env, struct cl_lock *lock,
                              struct cl_io *io, __u32 enqflags)
 {
 static int cl_enqueue_locked(const struct lu_env *env, struct cl_lock *lock,
                              struct cl_io *io, __u32 enqflags)
 {


@@ -1240,7 +1285,10 @@ static int cl_enqueue_locked(const struct lu_env *env, struct cl_lock *lock,
         do {
                 result = cl_enqueue_try(env, lock, io, enqflags);
                 if (result == CLO_WAIT) {
         do {
                 result = cl_enqueue_try(env, lock, io, enqflags);
                 if (result == CLO_WAIT) {

-                        result = cl_lock_state_wait(env, lock);
+                        if (lock->cll_conflict != NULL)
+                                result = cl_lock_enqueue_wait(env, lock, 1);
+                        else
+                                result = cl_lock_state_wait(env, lock);

                         if (result == 0)
                                 continue;
                 }
                         if (result == 0)
                                 continue;
                 }