From de881d75a7ec09e93083cc254fa92c7eb71ec78e Mon Sep 17 00:00:00 2001
From: Vitaly Fertman <vitaly_fertman@xyratex.com>
Date: Thu, 30 Jan 2014 01:31:30 +0400
Subject: [PATCH] LELUS-203 clio: no recursive closures

ensure we do not try to enclosure recursively, in which case embedded
disclosure does disclose and some operation may happen without proper
locking.

Signed-off-by: Vitaly Fertman <Vitaly_Fertman@xyratex.com>
Change-Id: Ia998745d9e324c4d70ca99b197b5a85ff1d4dd3c
---
 lustre/lov/lov_lock.c | 74 +++++++++++++++++++++++++++++----------------------
 1 file changed, 42 insertions(+), 32 deletions(-)

diff --git a/lustre/lov/lov_lock.c b/lustre/lov/lov_lock.c
index 16b2ca2..93878b2 100644
--- a/lustre/lov/lov_lock.c
+++ b/lustre/lov/lov_lock.c
@@ -400,21 +400,31 @@ static int lov_lock_sub_init(const struct lu_env *env,
 }
 
 static int lov_sublock_release(const struct lu_env *env, struct lov_lock *lck,
-                               int i, int deluser, int rc)
+                               int i, struct lovsub_lock *sub,
+			       int deluser, int rc,
+			       struct lov_sublock_env *subenv)
 {
+        struct cl_lock_closure *closure = &lov_env_info(env)->lti_closure;
         struct cl_lock *parent = lck->lls_cl.cls_lock;
+	struct cl_lock *sublock;
 
+	LASSERT(sub != NULL);
         LASSERT(cl_lock_is_mutexed(parent));
         ENTRY;
 
+	sublock = sub->lss_cl.cls_lock;
+	cl_lock_get(sublock);
+	/* take recursive mutex of sublock */
+	cl_lock_mutex_get(env, sublock);
+	/* need to release all locks in closure
+	 * otherwise it may deadlock. LU-2683.*/
+	lov_sublock_unlock(env, sub, closure, subenv);
+
+	/* sublock and parent are held. */
+
         if (lck->lls_sub[i].sub_flags & LSF_HELD) {
-                struct cl_lock    *sublock;
                 int dying;
 
-                LASSERT(lck->lls_sub[i].sub_lock != NULL);
-                sublock = lck->lls_sub[i].sub_lock->lss_cl.cls_lock;
-                LASSERT(cl_lock_is_mutexed(sublock));
-
                 lck->lls_sub[i].sub_flags &= ~LSF_HELD;
                 if (deluser)
                         cl_lock_user_del(env, sublock);
@@ -442,6 +452,9 @@ static int lov_sublock_release(const struct lu_env *env, struct lov_lock *lck,
                  * sub-lock is destroyed.
                  */
         }
+	cl_lock_mutex_put(env, sublock);
+	cl_lock_put(env, sublock);
+
         RETURN(rc);
 }
 
@@ -646,25 +659,18 @@ static int lov_lock_enqueue(const struct lu_env *env,
                                         /* take recursive mutex, the lock is
                                          * released in lov_lock_enqueue_wait.
                                          */
+					cl_lock_get(sublock);
                                         cl_lock_mutex_get(env, sublock);
                                         lov_sublock_unlock(env, sub, closure,
                                                            subenv);
                                         rc = lov_lock_enqueue_wait(env, lck,
                                                                    sublock);
+					cl_lock_put(env, sublock);
                                         break;
                                 case CLS_CACHED:
-					cl_lock_get(sublock);
-                                        /* take recursive mutex of sublock */
-                                        cl_lock_mutex_get(env, sublock);
-					/* need to release all locks in closure
-					 * otherwise it may deadlock. LU-2683.*/
-                                        lov_sublock_unlock(env, sub, closure,
-                                                           subenv);
-					/* sublock and parent are held. */
                                         rc = lov_sublock_release(env, lck, i,
-                                                                 1, rc);
-					cl_lock_mutex_put(env, sublock);
-					cl_lock_put(env, sublock);
+                                                                 sub, 1, rc,
+								 subenv);
 					break;
                                 default:
                                         lov_sublock_unlock(env, sub, closure,
@@ -725,13 +731,14 @@ static int lov_lock_unuse(const struct lu_env *env,
 			switch(sublock->cll_state) {
 			case CLS_HELD:
 				rc = cl_unuse_try(subenv->lse_env, sublock);
-				lov_sublock_release(env, lck, i, 0, 0);
+				lov_sublock_release(env, lck, i, sub,
+						    0, 0, subenv);
 				break;
 			default:
-				lov_sublock_release(env, lck, i, 1, 0);
+				lov_sublock_release(env, lck, i, sub,
+						    1, 0, subenv);
 				break;
 			}
-			lov_sublock_unlock(env, sub, closure, subenv);
 		}
 		result = lov_subresult(result, rc);
         }
@@ -781,13 +788,14 @@ static void lov_lock_cancel(const struct lu_env *env,
                         switch(sublock->cll_state) {
                         case CLS_HELD:
 				rc = cl_unuse_try(subenv->lse_env, sublock);
-                                lov_sublock_release(env, lck, i, 0, 0);
+                                lov_sublock_release(env, lck, i, sub,
+						    0, 0, subenv);
                                 break;
                         default:
-                                lov_sublock_release(env, lck, i, 1, 0);
+                                lov_sublock_release(env, lck, i, sub,
+						    1, 0, subenv);
                                 break;
                         }
-                        lov_sublock_unlock(env, sub, closure, subenv);
                 }
 
                 if (rc == CLO_REPEAT) {
@@ -894,16 +902,21 @@ static int lov_lock_use(const struct lu_env *env,
                         lov_sublock_hold(env, lck, i);
                         if (sublock->cll_state == CLS_CACHED) {
                                 rc = cl_use_try(subenv->lse_env, sublock, 0);
+
                                 if (rc != 0)
-                                        rc = lov_sublock_release(env, lck,
-                                                                 i, 1, rc);
+					rc = lov_sublock_release(env, lck, i,
+								 sub, 1, rc,
+								 subenv);
+				else
+					lov_sublock_unlock(env, sub, closure, subenv);
                         } else if (sublock->cll_state == CLS_NEW) {
                                 /* Sub-lock might have been canceled, while
                                  * top-lock was cached. */
                                 result = -ESTALE;
-                                lov_sublock_release(env, lck, i, 1, result);
-                        }
-                        lov_sublock_unlock(env, sub, closure, subenv);
+                                lov_sublock_release(env, lck, i, sub,
+						    1, result, subenv);
+                        } else
+				lov_sublock_unlock(env, sub, closure, subenv);
                 }
                 result = lov_subresult(result, rc);
                 if (result != 0)
@@ -1155,15 +1168,12 @@ static void lov_lock_delete(const struct lu_env *env,
                 LASSERT(rc == 0);
                 LASSERT(lsl->lss_cl.cls_lock->cll_state < CLS_FREEING);
 
-                if (lls->sub_flags & LSF_HELD)
-                        lov_sublock_release(env, lck, i, 1, 0);
-
                 link = lov_lock_link_find(env, lck, lsl);
                 LASSERT(link != NULL);
                 lov_lock_unlink(env, link, lsl);
                 LASSERT(lck->lls_sub[i].sub_lock == NULL);
 
-                lov_sublock_unlock(env, lsl, closure, NULL);
+		lov_sublock_release(env, lck, i, lsl, 1, 0, NULL);
         }
 
         cl_lock_closure_fini(closure);
-- 
1.8.4.3

