scoutfs: better tracking of recursive lock holders

This replaces the fragile recursive locking logic in dlmglue. In particular
that code fails when we have a pending downconvert and a process comes in
for a level that's compatible with the existing level. The downconvert will
still happen which causes us to now believe we are holding a lock that we
are not! We could go back to checking for holders that raced our downconvert
worker but that had problems of its own (see commit e8f7ef0).

Instead of trying to infer from lock state what we are allowed to do, let's
be explicit. Each lock now has a tree of task refs. If you come in to
acquire a lock, we look for our task in that tree. If it's not there, we
know this is the first time this task wanted that lock, so we can continue.
Otherwise we incremement a count on the task ref and return the already
locked lock. Unlock does the opposite - it finds the task ref and decreases
the count. On zero it will proceed with the actual unlock.

The owning task is the only process allowed to manipulate a task ref, so we
only have to lock manipulation of the tree. We make an exception for
global locks which might be unlocked from another process context (in this
case that means the node id lock).

Signed-off-by: Mark Fasheh <mfasheh@versity.com>

kmod/src/dlmglue.c

@@ -419,7 +419,7 @@ static inline int __levels_compat(int lockres_level, int wanted)
 	return level_compat_matrix[wanted + 1][lockres_level + 1];
 }
 
-static inline int levels_compat(struct ocfs2_lock_res *lockres, int wanted)
+int ocfs2_levels_compat(struct ocfs2_lock_res *lockres, int wanted)
 {
 	return __levels_compat(lockres->l_level, wanted);
 }
@@ -519,7 +519,7 @@ static inline void ocfs2_generic_handle_downconvert_action(struct ocfs2_lock_res
 	BUG_ON(lockres->l_blocking <= DLM_LOCK_NL);
 
 	lockres->l_level = lockres->l_requested;
-	if (levels_compat(lockres, dc_level)) {
+	if (ocfs2_levels_compat(lockres, dc_level)) {
 		lockres->l_blocking = DLM_LOCK_NL;
 		lockres_clear_flags(lockres, OCFS2_LOCK_BLOCKED);
 	}
@@ -1005,16 +1005,6 @@ static inline int ocfs2_may_continue_on_blocked_lock(struct ocfs2_lock_res *lock
 	return wanted <= ocfs2_downconvert_level(lockres, lockres->l_blocking);
 }
 
-/* the caller doesn't have to wait on a blocked lock if their wanted level
- * is compatible with it and there are already holders of the lock */
-static inline int lockres_allow_recursion(struct ocfs2_lock_res *lockres,
-					  int wanted)
-{
-	return (lockres->l_ops->flags & LOCK_TYPE_RECURSIVE) &&
-	       levels_compat(lockres, wanted) &&
-	       lockres_has_holders(lockres, H_ANY);
-}
-
 static void ocfs2_init_mask_waiter(struct ocfs2_mask_waiter *mw)
 {
 	INIT_LIST_HEAD(&mw->mw_item);
@@ -1153,7 +1143,7 @@ again:
 	 * here. If the lock is blocked waiting on a downconvert,
 	 * we'll get caught below. */
 	if (lockres->l_flags & OCFS2_LOCK_BUSY &&
-	    !levels_compat(lockres, level)) {
+	    !ocfs2_levels_compat(lockres, level)) {
 		/* is someone sitting in dlm_lock? If so, wait on
 		 * them. */
 		lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
@@ -1176,12 +1166,11 @@ again:
 		 * OCFS2_LOCK_BLOCKED check to ensure that there is no pending
 		 * downconvert request.
 		 */
-		if (levels_compat(lockres, level))
+		if (ocfs2_levels_compat(lockres, level))
 			goto update_holders;
 	}
 
 	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 */

kmod/src/dlmglue.h

@@ -286,20 +286,6 @@ 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 and the
- * resource is currently held.  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;
@@ -352,6 +338,7 @@ void ocfs2_wake_downconvert_thread(struct ocfs2_super *osb);
 
 struct ocfs2_dlm_debug *ocfs2_new_dlm_debug(void);
 void ocfs2_put_dlm_debug(struct ocfs2_dlm_debug *dlm_debug);
+int ocfs2_levels_compat(struct ocfs2_lock_res *lockres, int wanted);
 
 #if 0
 /* To set the locking protocol on module initialization */

kmod/src/lock.c

@@ -61,6 +61,108 @@ struct lock_info {
 
 static void scoutfs_lock_reclaim(struct work_struct *work);
 
+struct task_ref {
+	struct task_struct *task;
+	struct rb_node node;
+	int count;
+	int mode;/* for debugging */
+};
+
+static struct task_ref *find_task_ref(struct scoutfs_lock *lock,
+				     struct task_struct *task)
+{
+	struct rb_node *n;
+	struct task_ref *tmp;
+
+	spin_lock(&lock->task_refs_lock);
+	n = lock->task_refs.rb_node;
+	while (n) {
+		tmp = rb_entry(n, struct task_ref, node);
+
+		if (tmp->task < task)
+			n = n->rb_left;
+		else if (tmp->task > task)
+			n = n->rb_right;
+		else {
+			spin_unlock(&lock->task_refs_lock);
+			return tmp;
+		}
+	}
+	spin_unlock(&lock->task_refs_lock);
+
+	return NULL;
+}
+
+static struct task_ref *alloc_task_ref(struct task_struct *task, int mode)
+{
+	struct task_ref *ref = kzalloc(sizeof(*ref), GFP_NOFS);
+	if (ref) {
+		ref->task = task;
+		ref->count = 1;
+		ref->mode = mode;
+		RB_CLEAR_NODE(&ref->node);
+	}
+	return ref;
+}
+
+static void insert_task_ref(struct scoutfs_lock *lock, struct task_ref *ref)
+{
+	struct task_ref *tmp;
+	struct rb_node *parent = NULL;
+	struct rb_node **p;
+
+	spin_lock(&lock->task_refs_lock);
+	p = &lock->task_refs.rb_node;
+	while (*p) {
+		parent = *p;
+
+		tmp = rb_entry(parent, struct task_ref, node);
+
+		if (tmp->task < ref->task)
+			p = &(*p)->rb_left;
+		else if (tmp->task > ref->task)
+			p = &(*p)->rb_right;
+		else
+			BUG(); /* We should never find a duplicate */
+	}
+
+	rb_link_node(&ref->node, parent, p);
+	rb_insert_color(&ref->node, &lock->task_refs);
+	spin_unlock(&lock->task_refs_lock);
+}
+
+static void get_task_ref(struct task_ref *ref)
+{
+	ref->count++;
+}
+
+static struct task_ref *new_task_ref(struct scoutfs_lock *lock,
+				     struct task_struct *task, int mode)
+{
+	struct task_ref *ref = alloc_task_ref(task, mode);
+	if (ref)
+		insert_task_ref(lock, ref);
+
+	return ref;
+}
+
+static int put_task_ref(struct scoutfs_lock *lock, struct task_ref *ref)
+{
+	if (!ref)
+		return 0;
+
+	ref->count--;
+	if (ref->count == 0) {
+		spin_lock(&lock->task_refs_lock);
+		rb_erase(&ref->node, &lock->task_refs);
+		spin_unlock(&lock->task_refs_lock);
+
+		kfree(ref);
+		return 0;
+	}
+	return 1;
+}
+
 /*
  * Invalidate caches on this because another node wants a lock
  * with the a lock with the given mode and range. We always have to
@@ -196,7 +298,7 @@ static struct ocfs2_lock_res_ops scoufs_ino_lops = {
 	.downconvert_worker 	= ino_lock_downconvert,
 	/* XXX: .check_downconvert that queries the item cache for dirty items */
 	.print			= lock_name_string,
-	.flags			= LOCK_TYPE_REQUIRES_REFRESH|LOCK_TYPE_RECURSIVE,
+	.flags			= LOCK_TYPE_REQUIRES_REFRESH,
 };
 
 static struct ocfs2_lock_res_ops scoufs_ino_index_lops = {
@@ -204,7 +306,6 @@ static struct ocfs2_lock_res_ops scoufs_ino_index_lops = {
 	.downconvert_worker 	= ino_lock_downconvert,
 	/* XXX: .check_downconvert that queries the item cache for dirty items */
 	.print			= lock_name_string,
-	.flags			= LOCK_TYPE_RECURSIVE,
 };
 
 static struct ocfs2_lock_res_ops scoutfs_global_lops = {
@@ -251,6 +352,8 @@ static struct scoutfs_lock *alloc_scoutfs_lock(struct super_block *sb,
 		}
 	}
 
+	spin_lock_init(&lock->task_refs_lock);
+	lock->task_refs = RB_ROOT;
 	RB_CLEAR_NODE(&lock->node);
 	lock->sb = sb;
 	lock->lock_name = *lock_name;
@@ -491,6 +594,7 @@ static int lock_name_keys(struct super_block *sb, int mode, int flags,
 {
 	DECLARE_LOCK_INFO(sb, linfo);
 	struct scoutfs_lock *lock;
+	struct task_ref *ref = NULL;
 	int lkm_flags;
 	int ret;
 
@@ -506,13 +610,37 @@ static int lock_name_keys(struct super_block *sb, int mode, int flags,
 
 	trace_scoutfs_lock_resource(sb, lock);
 
+	if (!(flags & SCOUTFS_LKF_NO_TASK_REF)) {
+		ref = find_task_ref(lock, current);
+		if (ref) {
+			/*
+			 * We found a ref, which means we have already locked
+			 * this resource. Check that the calling task isn't
+			 * trying to switch modes in the middle of a recursive
+			 * lock request.
+			 */
+			BUG_ON(!ocfs2_levels_compat(&lock->lockres, mode));
+			get_task_ref(ref);
+			ret = 0;
+			goto out;
+		}
+
+		ref = new_task_ref(lock, current, mode);
+		if (!ref) {
+			ret = -ENOMEM;
+			goto out;
+		}
+	}
+
 	lkm_flags = DLM_LKF_NOORDER;
 	if (flags & SCOUTFS_LKF_TRYLOCK)
 		lkm_flags |= DLM_LKF_NOQUEUE; /* maybe also NONBLOCK? */
 
 	ret = ocfs2_cluster_lock(&linfo->dlmglue, &lock->lockres, mode,
 				 lkm_flags, 0);
+out:
 	if (ret) {
+		put_task_ref(lock, ref);
 		dec_lock_users(lock);
 		put_scoutfs_lock(sb, lock);
 	} else {
@@ -837,9 +965,10 @@ int scoutfs_lock_node_id(struct super_block *sb, int mode, int flags,
 			      &scoutfs_node_id_lops, &start, &end, lock);
 }
 
-void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock,
-		    int level)
+void scoutfs_unlock_flags(struct super_block *sb, struct scoutfs_lock *lock,
+			  int level, int flags)
 {
+	struct task_ref *ref;
 	DECLARE_LOCK_INFO(sb, linfo);
 
 	if (!lock)
@@ -847,6 +976,13 @@ void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock,
 
 	trace_scoutfs_unlock(sb, lock);
 
+	if (!(flags & SCOUTFS_LKF_NO_TASK_REF)) {
+		ref = find_task_ref(lock, current);
+		BUG_ON(!ref);
+		if (put_task_ref(lock, ref))
+			return;
+	}
+
 	ocfs2_cluster_unlock(&linfo->dlmglue, &lock->lockres, level);
 
 	dec_lock_users(lock);
@@ -854,6 +990,12 @@ void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock,
 	put_scoutfs_lock(sb, lock);
 }
 
+void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock,
+		    int level)
+{
+	scoutfs_unlock_flags(sb, lock, level, 0);
+}
+
 /*
  * The moment this is done we can have other mounts start asking
  * us to write back and invalidate, so do this very very late.

kmod/src/lock.h

@@ -7,6 +7,7 @@
 
 #define SCOUTFS_LKF_REFRESH_INODE	0x01 /* update stale inode from item */
 #define SCOUTFS_LKF_TRYLOCK		0x02 /* EAGAIN if contention */
+#define SCOUTFS_LKF_NO_TASK_REF  	0x04 /* don't create a task ref */
 
 /* flags for scoutfs_lock->flags */
 enum {
@@ -30,6 +31,8 @@ struct scoutfs_lock {
 	unsigned int users; /* Tracks active users of this lock */
 	unsigned long flags;
 	wait_queue_head_t waitq;
+	struct rb_root task_refs;
+	spinlock_t task_refs_lock;
 };
 
 u64 scoutfs_lock_refresh_gen(struct scoutfs_lock *lock);
@@ -54,6 +57,8 @@ int scoutfs_lock_node_id(struct super_block *sb, int mode, int flags,
 			 u64 node_id, struct scoutfs_lock **lock);
 void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock,
 		    int level);
+void scoutfs_unlock_flags(struct super_block *sb, struct scoutfs_lock *lock,
+			  int level, int flags);
 
 int scoutfs_lock_setup(struct super_block *sb);
 void scoutfs_lock_destroy(struct super_block *sb);

kmod/src/server.c

@@ -929,7 +929,8 @@ static void scoutfs_server_func(struct work_struct *work)
 
 	init_waitqueue_head(&waitq);
 
-	ret = scoutfs_lock_global(sb, DLM_LOCK_EX, SCOUTFS_LKF_TRYLOCK,
+	ret = scoutfs_lock_global(sb, DLM_LOCK_EX,
+				  SCOUTFS_LKF_TRYLOCK|SCOUTFS_LKF_NO_TASK_REF,
 				  SCOUTFS_LOCK_TYPE_GLOBAL_SERVER,
 				  &lock);
 	if (ret)

kmod/src/super.c

@@ -332,7 +332,8 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 out:
 	if (ret) {
 		scoutfs_server_destroy(sb);
-		scoutfs_unlock(sb, sbi->node_id_lock, DLM_LOCK_EX);
+		scoutfs_unlock_flags(sb, sbi->node_id_lock, DLM_LOCK_EX,
+				     SCOUTFS_LKF_NO_TASK_REF);
 		sbi->node_id_lock = NULL;
 	}
 	return ret;
@@ -362,7 +363,8 @@ static void scoutfs_kill_sb(struct super_block *sb)
 	kill_block_super(sb);
 
 	if (sbi) {
-		scoutfs_unlock(sb, sbi->node_id_lock, DLM_LOCK_EX);
+		scoutfs_unlock_flags(sb, sbi->node_id_lock, DLM_LOCK_EX,
+			SCOUTFS_LKF_NO_TASK_REF);
 		sbi->node_id_lock = NULL;
 
 		scoutfs_lock_destroy(sb);