scoutfs: lock invalidate only syncs dirty

Lock invalidation has to make sure that changes are visible to future
readers.  It was syncing if the current transaction is dirty.  This was
never optimal, but it wasn't catastrophic when concurrent invalidation
work could all block on one sync in progress.

With the move to a single invalidation worker serially invalidating
locks it became unacceptable.  Invalidation happening in the presence of
writers would constantly sync the current transaction while very old
unused write locks were invalidated.  Their changes had long since been
committed in previous transactions.

We add a lock field to remember the transaction sequence which could
have been dirtied under the lock.  If that transaction has already been
comitted by the time we invalidate the lock it doesn't have to sync.

Signed-off-by: Zach Brown <zab@versity.com>

kmod/src/counters.h

@@ -84,11 +84,11 @@
 	EXPAND_COUNTER(lock_grant_request)			\
 	EXPAND_COUNTER(lock_grant_response)			\
 	EXPAND_COUNTER(lock_grant_work)				\
-	EXPAND_COUNTER(lock_invalidate_commit)			\
 	EXPAND_COUNTER(lock_invalidate_coverage)		\
 	EXPAND_COUNTER(lock_invalidate_inode)			\
 	EXPAND_COUNTER(lock_invalidate_request)			\
 	EXPAND_COUNTER(lock_invalidate_response)		\
+	EXPAND_COUNTER(lock_invalidate_sync)			\
 	EXPAND_COUNTER(lock_invalidate_work)			\
 	EXPAND_COUNTER(lock_lock)				\
 	EXPAND_COUNTER(lock_lock_error)				\

kmod/src/lock.c

@@ -160,15 +160,13 @@ static int lock_invalidate(struct super_block *sb, struct scoutfs_lock *lock,
 	BUG_ON(!(prev == SCOUTFS_LOCK_WRITE && mode == SCOUTFS_LOCK_READ) &&
 	         mode != SCOUTFS_LOCK_NULL);
 
-	/* any transition from a mode allowed to dirty items has to write */
-	if (lock_mode_can_write(prev) && scoutfs_trans_has_dirty(sb)) {
+	/* sync when a write lock could have dirtied the current transaction */
+	if (lock_mode_can_write(prev) &&
+	    (lock->dirty_trans_seq == scoutfs_trans_sample_seq(sb))) {
+		scoutfs_inc_counter(sb, lock_invalidate_sync);
 		ret = scoutfs_trans_sync(sb, 1);
 		if (ret < 0)
 			return ret;
-		if (ret > 0) {
-			scoutfs_add_counter(sb, lock_invalidate_commit, ret);
-			ret = 0;
-		}
 	}
 
 	/* have to invalidate if we're not in the only usable case */
@@ -1273,6 +1271,8 @@ void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, int mode)
 
 	lock_dec_count(lock->users, mode);
 	extend_grace(sb, lock);
+	if (lock_mode_can_write(mode))
+		lock->dirty_trans_seq = scoutfs_trans_sample_seq(sb);
 
 	trace_scoutfs_lock_unlock(sb, lock);
 	wake_up(&lock->waitq);

kmod/src/lock.h

@@ -22,6 +22,7 @@ struct scoutfs_lock {
 	struct rb_node range_node;
 	u64 refresh_gen;
 	u64 write_version;
+	u64 dirty_trans_seq;
 	struct scoutfs_net_roots roots;
 	struct list_head lru_head;
 	wait_queue_head_t waitq;

kmod/src/trans.c

@@ -158,6 +158,7 @@ void scoutfs_trans_write_func(struct work_struct *work)
 						   trans_write_work.work);
 	struct super_block *sb = sbi->sb;
 	DECLARE_TRANS_INFO(sb, tri);
+	u64 trans_seq = sbi->trans_seq;
 	char *s = NULL;
 	int ret = 0;
 
@@ -177,7 +178,7 @@ void scoutfs_trans_write_func(struct work_struct *work)
 			 * seq indices but doesn't send a message for every sync
 			 * syscall.
 			 */
-			ret = scoutfs_client_advance_seq(sb, &sbi->trans_seq);
+			ret = scoutfs_client_advance_seq(sb, &trans_seq);
 			if (ret < 0)
 			      s = "clean advance seq";
 		}
@@ -194,8 +195,7 @@ void scoutfs_trans_write_func(struct work_struct *work)
 	      (s = "meta write", scoutfs_block_writer_write(sb, &tri->wri))  ?:
 	      (s = "data wait", scoutfs_inode_walk_writeback(sb, false)) ?:
 	      (s = "commit log trees", commit_btrees(sb)) ?:
-	      (s = "advance seq", scoutfs_client_advance_seq(sb,
-							     &sbi->trans_seq))?:
+	      (s = "advance seq", scoutfs_client_advance_seq(sb, &trans_seq)) ?:
 	      (s = "get log trees", scoutfs_trans_get_log_trees(sb));
 out:
 	if (ret < 0)
@@ -205,6 +205,7 @@ out:
 	spin_lock(&sbi->trans_write_lock);
 	sbi->trans_write_count++;
 	sbi->trans_write_ret = ret;
+	sbi->trans_seq = trans_seq;
 	spin_unlock(&sbi->trans_write_lock);
 	wake_up(&sbi->trans_write_wq);
 
@@ -522,6 +523,23 @@ void scoutfs_release_trans(struct super_block *sb)
 		wake_up(&sbi->trans_hold_wq);
 }
 
+/*
+ * Return the current transaction sequence.  Whether this is racing with
+ * the transaction write thread is entirely dependent on the caller's
+ * context.
+ */
+u64 scoutfs_trans_sample_seq(struct super_block *sb)
+{
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	u64 ret;
+
+	spin_lock(&sbi->trans_write_lock);
+	ret = sbi->trans_seq;
+	spin_unlock(&sbi->trans_write_lock);
+
+	return ret;
+}
+
 int scoutfs_setup_trans(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);

kmod/src/trans.h

@@ -18,6 +18,7 @@ int scoutfs_hold_trans(struct super_block *sb,
 		       const struct scoutfs_item_count cnt);
 bool scoutfs_trans_held(void);
 void scoutfs_release_trans(struct super_block *sb);
+u64 scoutfs_trans_sample_seq(struct super_block *sb);
 void scoutfs_trans_track_item(struct super_block *sb, signed items,
 			      signed vals);