LU-1070 agl: update lock state when AGL upcall

The AGL RPC sponsor may exits the cl_lock processing without
wait() called before related OSC lock upcall(). So when AGL
upcall(), it needs to update the cl_lock state according to
the enqueue result through the general cl_lock API wait().

Originally, it is done in "lov_lock_unuse()" against toplock.
But osc_lock upcall() is against sublock. So the sublock state
may be un-updated and inconsistent with the low layer osc_lock
state, then causes the sublock cannot be cached.

On the other hand, cl_lock::cll_descr::cld_enq_flags should not be
changed during unuse() for passing parameter, as the replacement,
it can be done through cl_lock::cll_flags.

Signed-off-by: Fan Yong <yong.fan@whamcloud.com>
Change-Id: I4bc4c0ef0b93d5e0c2e59304158a35c57635ad70
Reviewed-on: http://review.whamcloud.com/2099
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Jinshan Xiong <jinshan.xiong@whamcloud.com>
Reviewed-by: Niu Yawei <niu@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/cl_object.h b/lustre/include/cl_object.h
index c0d6223..d0efdc7 100644 (file)
--- a/lustre/include/cl_object.h
+++ b/lustre/include/cl_object.h
@@ -1442,7 +1442,9 @@ enum cl_lock_flags {
         /** cancellation is pending for this lock. */
         CLF_CANCELPEND = 1 << 1,
         /** destruction is pending for this lock. */
-        CLF_DOOMED     = 1 << 2
+        CLF_DOOMED     = 1 << 2,
+        /** from enqueue RPC reply upcall. */
+        CLF_FROM_UPCALL= 1 << 3,
 };
 
 /**
@@ -2160,13 +2162,9 @@ enum cl_enq_flags {
          */
         CEF_AGL          = 0x00000020,
         /**
-         * do not trigger re-enqueue.
-         */
-        CEF_NO_REENQUEUE = 0x00000040,
-        /**
          * mask of enq_flags.
          */
-        CEF_MASK         = 0x0000007f,
+        CEF_MASK         = 0x0000003f,
 };
 
 /**

diff --git a/lustre/lov/lov_lock.c b/lustre/lov/lov_lock.c
index c12f246..35548e5 100644 (file)
--- a/lustre/lov/lov_lock.c
+++ b/lustre/lov/lov_lock.c
@@ -701,19 +701,6 @@ static int lov_lock_unuse(const struct lu_env *env,
                         if (lls->sub_flags & LSF_HELD) {
                                 LASSERT(sublock->cll_state == CLS_HELD ||
                                         sublock->cll_state == CLS_ENQUEUED);
-                                /* For AGL case, the sublock state maybe not
-                                 * match the lower layer state, so sync them
-                                 * before unuse. */
-                                if (sublock->cll_users == 1 &&
-                                    sublock->cll_state == CLS_ENQUEUED) {
-                                        __u32 save;
-
-                                        save = sublock->cll_descr.cld_enq_flags;
-                                        sublock->cll_descr.cld_enq_flags |=
-                                                        CEF_NO_REENQUEUE;
-                                        cl_wait_try(env, sublock);
-                                        sublock->cll_descr.cld_enq_flags = save;
-                                }
                                 rc = cl_unuse_try(subenv->lse_env, sublock);
                                 rc = lov_sublock_release(env, lck, i, 0, rc);
                         }

diff --git a/lustre/osc/osc_lock.c b/lustre/osc/osc_lock.c
index 7d59098..4e137bc 100644 (file)
--- a/lustre/osc/osc_lock.c
+++ b/lustre/osc/osc_lock.c
@@ -539,10 +539,8 @@ static int osc_lock_upcall(void *cookie, int errcode)
                                 LDLM_LOCK_PUT(dlmlock);
                         }
                 } else {
-                        if (olck->ols_glimpse) {
+                        if (olck->ols_glimpse)
                                 olck->ols_glimpse = 0;
-                                olck->ols_agl = 0 ;
-                        }
                         osc_lock_upcall0(env, olck);
                 }
 
@@ -566,6 +564,17 @@ static int osc_lock_upcall(void *cookie, int errcode)
                 }
 
                 if (rc == 0) {
+                        /* For AGL case, the RPC sponsor may exits the cl_lock
+                        *  processing without wait() called before related OSC
+                        *  lock upcall(). So update the lock status according
+                        *  to the enqueue result inside AGL upcall(). */
+                        if (olck->ols_agl) {
+                                lock->cll_flags |= CLF_FROM_UPCALL;
+                                cl_wait_try(env, lock);
+                                lock->cll_flags &= ~CLF_FROM_UPCALL;
+                                if (!olck->ols_glimpse)
+                                        olck->ols_agl = 0;
+                        }
                         cl_lock_signal(env, lock);
                         /* del user for lock upcall cookie */
                         cl_unuse_try(env, lock);
@@ -1247,7 +1256,12 @@ static int osc_lock_wait(const struct lu_env *env,
                 if (olck->ols_flags & LDLM_FL_LVB_READY) {
                         return 0;
                 } else if (olck->ols_agl) {
-                        olck->ols_state = OLS_NEW;
+                        if (lock->cll_flags & CLF_FROM_UPCALL)
+                                /* It is from enqueue RPC reply upcall for
+                                 * updating state. Do not re-enqueue. */
+                                return -ENAVAIL;
+                        else
+                                olck->ols_state = OLS_NEW;
                 } else {
                         LASSERT(lock->cll_error);
                         return lock->cll_error;
@@ -1255,20 +1269,15 @@ static int osc_lock_wait(const struct lu_env *env,
         }
 
         if (olck->ols_state == OLS_NEW) {
-                if (lock->cll_descr.cld_enq_flags & CEF_NO_REENQUEUE) {
-                        return -ENAVAIL;
-                } else {
-                        int rc;
+                int rc;
 
-                        LASSERT(olck->ols_agl);
+                LASSERT(olck->ols_agl);
 
-                        rc = osc_lock_enqueue(env, slice, NULL, CEF_ASYNC |
-                                                                CEF_MUST);
-                        if (rc != 0)
-                                return rc;
-                        else
-                                return CLO_REENQUEUED;
-                }
+                rc = osc_lock_enqueue(env, slice, NULL, CEF_ASYNC | CEF_MUST);
+                if (rc != 0)
+                        return rc;
+                else
+                        return CLO_REENQUEUED;
         }
 
         LASSERT(equi(olck->ols_state >= OLS_UPCALL_RECEIVED &&