From 28a6b82690f37c0166d0c778c3a62dd208edf87f Mon Sep 17 00:00:00 2001
From: Mark Fasheh <mfasheh@versity.com>
Date: Tue, 26 Sep 2017 17:23:23 -0500
Subject: [PATCH] scoutfs: allow some recursive locking in dlmglue

Scoutfs can get into a situation where it wants to acquire a lock twice.
This happens when we have a parent and child in the same inode group. A
create operation will lock that group twice, once for each inode.
Instead of forcing callers to remember which inode groups they've
locked, we handle this internally within dlmglue.

Add a dlmglue lock type flag that indicates we might recursively lock a
resource. During locking, when dlmglue sees that flag and we already
have the lock at an appropriate level, it will allow the lock operation
to continue even when the lock is marked blocking.

Signed-off-by: Mark Fasheh <mfasheh@versity.com>
---
 kmod/src/dlmglue.c | 23 +++++++++++++++++++++++
 kmod/src/dlmglue.h | 13 +++++++++++++
 kmod/src/lock.c    |  4 ++--
 3 files changed, 38 insertions(+), 2 deletions(-)

diff --git a/kmod/src/dlmglue.c b/kmod/src/dlmglue.c
index ffcf264d..dc1fa4d3 100644
--- a/kmod/src/dlmglue.c
+++ b/kmod/src/dlmglue.c
@@ -918,6 +918,13 @@ static inline int ocfs2_may_continue_on_blocked_lock(struct ocfs2_lock_res *lock
 	return wanted <= ocfs2_highest_compat_lock_level(lockres->l_blocking);
 }
 
+static inline int lockres_allow_recursion(struct ocfs2_lock_res *lockres,
+					  int wanted)
+{
+	return (lockres->l_ops->flags & LOCK_TYPE_RECURSIVE)
+		&& wanted <= lockres->l_level;
+}
+
 static void ocfs2_init_mask_waiter(struct ocfs2_mask_waiter *mw)
 {
 	INIT_LIST_HEAD(&mw->mw_item);
@@ -1070,6 +1077,7 @@ again:
 	}
 
 	if (lockres->l_flags & OCFS2_LOCK_BLOCKED &&
+	    !lockres_allow_recursion(lockres, level) &&
 	    !ocfs2_may_continue_on_blocked_lock(lockres, level)) {
 		/* is the lock is currently blocked on behalf of
 		 * another node */
@@ -2385,6 +2393,21 @@ recheck:
 		goto recheck;
 	}
 
+	if ((lockres->l_ops->flags & LOCK_TYPE_RECURSIVE) &&
+	    ((lockres->l_blocking == DLM_LOCK_PR && lockres->l_ex_holders) ||
+	     (lockres->l_blocking == DLM_LOCK_EX &&
+	      (lockres->l_ex_holders || lockres->l_ro_holders)))) {
+		/*
+		 * Recursive locks may have had their holder count
+		 * incremented while we were sleeping in
+		 * ->downconvert_worker. Recheck here.
+		 */
+		mlog(ML_BASTS, "lockres %s, block=%d:%d, level=%d:%d, ro=%d "
+		     "ex=%d, Recheck\n", lockres->l_name, blocking,
+		       lockres->l_blocking, level, lockres->l_level,
+		       lockres->l_ro_holders, lockres->l_ex_holders);
+		goto recheck;
+	}
 downconvert:
 	ctl->requeue = 0;
 
diff --git a/kmod/src/dlmglue.h b/kmod/src/dlmglue.h
index 0d852743..4a88b934 100644
--- a/kmod/src/dlmglue.h
+++ b/kmod/src/dlmglue.h
@@ -279,6 +279,19 @@ struct ocfs2_lock_res_ops {
  */
 #define LOCK_TYPE_USES_LVB		0x2
 
+/*
+ * Tells dlmglue to override fairness considerations when locking this
+ * lock type - the blocking flag will be ignored when a lock is
+ * requested and we already have it at the appropriate level. This
+ * allows a process to acquire a dlmglue lock on the same resource
+ * multiple times in a row without deadlocking, even if another node has
+ * asked for a competing lock on the resource.
+ *
+ * Note that lock/unlock calls must always be balanced (1 unlock for
+ * every lock), even when this flag is set.
+ */
+#define LOCK_TYPE_RECURSIVE 0x4
+
 struct ocfs2_lock_holder {
 	struct list_head oh_list;
 	struct pid *oh_owner_pid;
diff --git a/kmod/src/lock.c b/kmod/src/lock.c
index d89ea6ac..7fc15526 100644
--- a/kmod/src/lock.c
+++ b/kmod/src/lock.c
@@ -174,14 +174,14 @@ static struct ocfs2_lock_res_ops scoufs_ino_lops = {
 	.get_osb 		= get_ino_lock_osb,
 	.downconvert_worker 	= ino_lock_downconvert,
 	/* XXX: .check_downconvert that queries the item cache for dirty items */
-	.flags			= LOCK_TYPE_REQUIRES_REFRESH,
+	.flags			= LOCK_TYPE_REQUIRES_REFRESH|LOCK_TYPE_RECURSIVE,
 };
 
 static struct ocfs2_lock_res_ops scoufs_ino_index_lops = {
 	.get_osb 		= get_ino_lock_osb,
 	.downconvert_worker 	= ino_lock_downconvert,
 	/* XXX: .check_downconvert that queries the item cache for dirty items */
-	.flags			= 0,
+	.flags			= LOCK_TYPE_RECURSIVE,
 };
 
 static struct ocfs2_lock_res_ops scoutfs_global_lops = {