Merge pull request #49 from versity/zab/btree_merge_fixes

Zab/btree merge fixes

kmod/src/alloc.c

@@ -252,6 +252,7 @@ static struct scoutfs_ext_ops alloc_ext_ops = {
 	.next = alloc_ext_next,
 	.insert = alloc_ext_insert,
 	.remove = alloc_ext_remove,
+	.insert_overlap_warn = true,
 };
 
 static bool invalid_extent(u64 start, u64 end, u64 first, u64 last)

kmod/src/btree.c

@@ -2013,94 +2013,14 @@ int scoutfs_btree_rebalance(struct super_block *sb,
 struct merge_pos {
 	struct rb_node node;
 	struct scoutfs_btree_root *root;
-	struct scoutfs_key key;
+	struct scoutfs_block *bl;
+	struct scoutfs_btree_block *bt;
+	struct scoutfs_avl_node *avl;
+	struct scoutfs_key *key;
 	unsigned int val_len;
-	u8 val[SCOUTFS_BTREE_MAX_VAL_LEN];
+	u8 *val;
 };
 
-/*
- * Find the next item in the mpos's root after its key and make sure
- * that it's in its sorted position in the rbtree.  We're responsible
- * for freeing the mpos if we don't put it back in the pos_root.  This
- * happens naturally naturally when its item_root has no more items to
- * merge.
- */
-static int reset_mpos(struct super_block *sb, struct rb_root *pos_root,
-		      struct merge_pos *mpos, struct scoutfs_key *end,
-		      scoutfs_btree_merge_cmp_t merge_cmp)
-{
-	SCOUTFS_BTREE_ITEM_REF(iref);
-	struct merge_pos *walk;
-	struct rb_node *parent;
-	struct rb_node **node;
-	int key_cmp;
-	int val_cmp;
-	int ret;
-
-restart:
-	if (!RB_EMPTY_NODE(&mpos->node)) {
-		rb_erase(&mpos->node, pos_root);
-		RB_CLEAR_NODE(&mpos->node);
-	}
-
-	/* find the next item in the root within end */
-	ret = scoutfs_btree_next(sb, mpos->root, &mpos->key, &iref);
-	if (ret == 0) {
-		if (scoutfs_key_compare(iref.key, end) > 0) {
-			ret = -ENOENT;
-		} else {
-			mpos->key = *iref.key;
-			mpos->val_len = iref.val_len;
-			memcpy(mpos->val, iref.val, iref.val_len);
-		}
-		scoutfs_btree_put_iref(&iref);
-	}
-	if (ret < 0) {
-		kfree(mpos);
-		if (ret == -ENOENT)
-			ret = 0;
-		goto out;
-	}
-
-rewalk:
-	/* sort merge items by key then oldest to newest */
-	node = &pos_root->rb_node;
-	parent = NULL;
-	while (*node) {
-		parent = *node;
-		walk = container_of(*node, struct merge_pos, node);
-
-		key_cmp = scoutfs_key_compare(&mpos->key, &walk->key);
-		val_cmp = merge_cmp(mpos->val, mpos->val_len,
-				    walk->val, walk->val_len);
-
-		/* drop old versions of logged keys as we discover them */
-		if (key_cmp == 0) {
-			scoutfs_inc_counter(sb, btree_merge_drop_old);
-			if (val_cmp < 0)  {
-				scoutfs_key_inc(&mpos->key);
-				goto restart;
-			} else {
-				BUG_ON(val_cmp == 0);
-				rb_erase(&walk->node, pos_root);
-				kfree(walk);
-				goto rewalk;
-			}
-		}
-
-		if ((key_cmp ?: val_cmp) < 0)
-			node = &(*node)->rb_left;
-		else
-			node = &(*node)->rb_right;
-	}
-
-	rb_link_node(&mpos->node, parent, node);
-	rb_insert_color(&mpos->node, pos_root);
-	ret = 0;
-out:
-	return ret;
-}
-
 static struct merge_pos *first_mpos(struct rb_root *root)
 {
 	struct rb_node *node = rb_first(root);
@@ -2109,6 +2029,109 @@ static struct merge_pos *first_mpos(struct rb_root *root)
 	return NULL;
 }
 
+static void free_mpos(struct super_block *sb, struct merge_pos *mpos)
+{
+	scoutfs_block_put(sb, mpos->bl);
+	kfree(mpos);
+}
+
+static void insert_mpos(struct rb_root *pos_root, struct merge_pos *ins,
+			scoutfs_btree_merge_cmp_t merge_cmp)
+{
+	struct rb_node **node = &pos_root->rb_node;
+	struct rb_node *parent = NULL;
+	struct merge_pos *mpos;
+	int cmp;
+
+	parent = NULL;
+	while (*node) {
+		parent = *node;
+		mpos = container_of(*node, struct merge_pos, node);
+
+		/* sort merge items by key then newest to oldest */
+		cmp = scoutfs_key_compare(ins->key, mpos->key) ?:
+		      -merge_cmp(ins->val, ins->val_len, mpos->val, mpos->val_len);
+
+		if (cmp < 0)
+			node = &(*node)->rb_left;
+		else
+			node = &(*node)->rb_right;
+	}
+
+	rb_link_node(&ins->node, parent, node);
+	rb_insert_color(&ins->node, pos_root);
+}
+
+/*
+ * Find the next item in the merge_pos root in the caller's range and
+ * insert it into the rbtree sorted by key and version so that merging
+ * can find the next newest item at the front of the rbtree.  We free
+ * the mpos on error or if there are no more items in the range.
+ */
+static int reset_mpos(struct super_block *sb, struct rb_root *pos_root, struct merge_pos *mpos,
+		      struct scoutfs_key *start, struct scoutfs_key *end,
+		      scoutfs_btree_merge_cmp_t merge_cmp)
+{
+	struct scoutfs_btree_item *item;
+	struct scoutfs_avl_node *next;
+	struct btree_walk_key_range kr;
+	struct scoutfs_key walk_key;
+	int ret = 0;
+
+	/* always erase before freeing or inserting */
+	if (!RB_EMPTY_NODE(&mpos->node)) {
+		rb_erase(&mpos->node, pos_root);
+		RB_CLEAR_NODE(&mpos->node);
+	}
+
+	/*
+	 * advance to next item via the avl tree.   The caller's pos is
+	 * only ever incremented past the last key so we can use next to
+	 * iterate rather than using search to skip past multiple items.
+	 */
+	if (mpos->avl)
+		mpos->avl = scoutfs_avl_next(&mpos->bt->item_root, mpos->avl);
+
+	/* find the next leaf with the key if we run out of items */
+	walk_key = *start;
+	while (!mpos->avl && !scoutfs_key_is_zeros(&walk_key)) {
+		scoutfs_block_put(sb, mpos->bl);
+		mpos->bl = NULL;
+		ret = btree_walk(sb, NULL, NULL, mpos->root, BTW_NEXT, &walk_key,
+				 0, &mpos->bl, &kr, NULL);
+		if (ret < 0) {
+			if (ret == -ENOENT)
+				ret = 0;
+			free_mpos(sb, mpos);
+			goto out;
+		}
+		mpos->bt = mpos->bl->data;
+
+		mpos->avl = scoutfs_avl_search(&mpos->bt->item_root, cmp_key_item,
+					       start, NULL, NULL, &next, NULL) ?: next;
+		if (mpos->avl == NULL)
+			walk_key = kr.iter_next;
+	}
+
+	/* see if we're out of items within the range */
+	item = node_item(mpos->avl);
+	if (!item || scoutfs_key_compare(item_key(item), end) > 0) {
+		free_mpos(sb, mpos);
+		ret = 0;
+		goto out;
+	}
+
+	/* insert the next item within range at its version */
+	mpos->key = item_key(item);
+	mpos->val_len = item_val_len(item);
+	mpos->val = item_val(mpos->bt, item);
+
+	insert_mpos(pos_root, mpos, merge_cmp);
+	ret = 0;
+out:
+	return ret;
+}
+
 /*
  * Merge items from a number of read-only input roots into a writable
  * destination root.  The order of the input roots doesn't matter, the
@@ -2149,6 +2172,7 @@ int scoutfs_btree_merge(struct super_block *sb,
 	struct scoutfs_block *bl = NULL;
 	struct btree_walk_key_range kr;
 	struct scoutfs_avl_node *par;
+	struct scoutfs_key next;
 	struct merge_pos *mpos;
 	struct merge_pos *tmp;
 	int walk_val_len;
@@ -2161,17 +2185,16 @@ int scoutfs_btree_merge(struct super_block *sb,
 	scoutfs_inc_counter(sb, btree_merge);
 
 	list_for_each_entry(rhead, inputs, head) {
-		mpos = kmalloc(sizeof(*mpos), GFP_NOFS);
+		mpos = kzalloc(sizeof(*mpos), GFP_NOFS);
 		if (!mpos) {
 			ret = -ENOMEM;
 			goto out;
 		}
 
 		RB_CLEAR_NODE(&mpos->node);
-		mpos->key = *start;
 		mpos->root = &rhead->root;
 
-		ret = reset_mpos(sb, &pos_root, mpos, end, merge_cmp);
+		ret = reset_mpos(sb, &pos_root, mpos, start, end, merge_cmp);
 		if (ret < 0)
 			goto out;
 	}
@@ -2186,24 +2209,24 @@ int scoutfs_btree_merge(struct super_block *sb,
 		if (scoutfs_block_writer_dirty_bytes(sb, wri) >= dirty_limit) {
 			scoutfs_inc_counter(sb, btree_merge_dirty_limit);
 			ret = -ERANGE;
-			*next_ret = mpos->key;
+			*next_ret = *mpos->key;
 			goto out;
 		}
 
 		if (scoutfs_alloc_meta_low(sb, alloc, alloc_low)) {
 			scoutfs_inc_counter(sb, btree_merge_alloc_low);
 			ret = -ERANGE;
-			*next_ret = mpos->key;
+			*next_ret = *mpos->key;
 			goto out;
 		}
 
 		scoutfs_block_put(sb, bl);
 		bl = NULL;
 		ret = btree_walk(sb, alloc, wri, root, walk_flags,
-				 &mpos->key, walk_val_len, &bl, &kr, NULL);
+			         mpos->key, walk_val_len, &bl, &kr, NULL);
 		if (ret < 0) {
 			if (ret == -ERANGE)
-				*next_ret = mpos->key;
+				*next_ret = *mpos->key;
 			goto out;
 		}
 		bt = bl->data;
@@ -2218,15 +2241,15 @@ int scoutfs_btree_merge(struct super_block *sb,
 			}
 
 			/* walk to new leaf if we exceed parent ref key */
-			if (scoutfs_key_compare(&mpos->key, &kr.end) > 0)
+			if (scoutfs_key_compare(mpos->key, &kr.end) > 0)
 				break;
 
 			/* see if there's an existing item */
-			item = leaf_item_hash_search(sb, bt, &mpos->key);
+			item = leaf_item_hash_search(sb, bt, mpos->key);
 			is_del = merge_is_del(mpos->val, mpos->val_len);
 
 			trace_scoutfs_btree_merge_items(sb, mpos->root,
-					&mpos->key, mpos->val_len,
+					mpos->key, mpos->val_len,
 					item ? root : NULL,
 					item ? item_key(item) : NULL,
 					item ? item_val_len(item) : 0, is_del);
@@ -2241,9 +2264,9 @@ int scoutfs_btree_merge(struct super_block *sb,
 			/* insert missing non-deletion merge items */
 			if (!item && !is_del) {
 				scoutfs_avl_search(&bt->item_root,
-						   cmp_key_item, &mpos->key,
+						   cmp_key_item, mpos->key,
 						   &cmp, &par, NULL, NULL);
-				create_item(bt, &mpos->key,
+				create_item(bt, mpos->key,
 					    mpos->val + drop_val,
 					    mpos->val_len - drop_val, par, cmp);
 				scoutfs_inc_counter(sb, btree_merge_insert);
@@ -2273,12 +2296,15 @@ int scoutfs_btree_merge(struct super_block *sb,
 			walk_flags &= ~(BTW_INSERT | BTW_DELETE);
 			walk_val_len = 0;
 
-			/* finished with this merge item */
-			scoutfs_key_inc(&mpos->key);
-			ret = reset_mpos(sb, &pos_root, mpos, end, merge_cmp);
-			if (ret < 0)
-				goto out;
-			mpos = NULL;
+			/* finished with this key, skip any older items */
+			next = *mpos->key;
+			scoutfs_key_inc(&next);
+			while (mpos && scoutfs_key_compare(mpos->key, &next) < 0) {
+				ret = reset_mpos(sb, &pos_root, mpos, &next, end, merge_cmp);
+				if (ret < 0)
+					goto out;
+				mpos = first_mpos(&pos_root);
+			}
 		}
 	}
 
@@ -2286,7 +2312,7 @@ int scoutfs_btree_merge(struct super_block *sb,
 out:
 	scoutfs_block_put(sb, bl);
 	rbtree_postorder_for_each_entry_safe(mpos, tmp, &pos_root, node) {
-		kfree(mpos);
+		free_mpos(sb, mpos);
 	}
 
 	return ret;

kmod/src/client.c

@@ -32,6 +32,7 @@
 #include "endian_swap.h"
 #include "quorum.h"
 #include "omap.h"
+#include "trans.h"
 
 /*
  * The client is responsible for maintaining a connection to the server.
@@ -305,6 +306,24 @@ int scoutfs_client_resize_devices(struct super_block *sb, struct scoutfs_net_res
 					nrd, sizeof(*nrd), NULL, 0);
 }
 
+/*
+ * The server is asking that we trigger a commit of the current log
+ * trees so that they can ensure an item seq discontinuity between
+ * finalized log btrees and the next set of open log btrees.  If we're
+ * shutting down then we're already going to perform a final commit.
+ */
+static int sync_log_trees(struct super_block *sb, struct scoutfs_net_connection *conn,
+			  u8 cmd, u64 id, void *arg, u16 arg_len)
+{
+	if (arg_len != 0)
+		return -EINVAL;
+
+	if (!scoutfs_unmounting(sb))
+		scoutfs_trans_sync(sb, 0);
+
+	return scoutfs_net_response(sb, conn, cmd, id, 0, NULL, 0);
+}
+
 /* The client is receiving a invalidation request from the server */
 static int client_lock(struct super_block *sb,
 		       struct scoutfs_net_connection *conn, u8 cmd, u64 id,
@@ -516,6 +535,7 @@ out:
 }
 
 static scoutfs_net_request_t client_req_funcs[] = {
+	[SCOUTFS_NET_CMD_SYNC_LOG_TREES]	= sync_log_trees,
 	[SCOUTFS_NET_CMD_LOCK]			= client_lock,
 	[SCOUTFS_NET_CMD_LOCK_RECOVER]		= client_lock_recover,
 	[SCOUTFS_NET_CMD_OPEN_INO_MAP]		= client_open_ino_map,

kmod/src/counters.h

@@ -120,12 +120,8 @@
 	EXPAND_COUNTER(item_write_dirty)			\
 	EXPAND_COUNTER(lock_alloc)				\
 	EXPAND_COUNTER(lock_free)				\
-	EXPAND_COUNTER(lock_grace_extended)			\
-	EXPAND_COUNTER(lock_grace_set)				\
-	EXPAND_COUNTER(lock_grace_wait)				\
 	EXPAND_COUNTER(lock_grant_request)			\
 	EXPAND_COUNTER(lock_grant_response)			\
-	EXPAND_COUNTER(lock_grant_work)				\
 	EXPAND_COUNTER(lock_invalidate_coverage)		\
 	EXPAND_COUNTER(lock_invalidate_inode)			\
 	EXPAND_COUNTER(lock_invalidate_request)			\

kmod/src/ext.c

@@ -13,6 +13,7 @@
 #include <linux/kernel.h>
 #include <linux/fs.h>
 
+#include "msg.h"
 #include "ext.h"
 #include "counters.h"
 #include "scoutfs_trace.h"
@@ -191,6 +192,9 @@ int scoutfs_ext_insert(struct super_block *sb, struct scoutfs_ext_ops *ops,
 
 	/* inserting extent must not overlap */
 	if (found.len && ext_overlap(&ins, found.start, found.len)) {
+		if (ops->insert_overlap_warn)
+			scoutfs_err(sb, "inserting extent %llu.%llu overlaps existing %llu.%llu",
+					start, len, found.start, found.len);
 		ret = -EINVAL;
 		goto out;
 	}
@@ -242,6 +246,8 @@ int scoutfs_ext_remove(struct super_block *sb, struct scoutfs_ext_ops *ops,
 
 	/* removed extent must be entirely within found */
 	if (!scoutfs_ext_inside(start, len, &found)) {
+		scoutfs_err(sb, "error removing extent %llu.%llu, isn't inside existing %llu.%llu",
+				start, len, found.start, found.len);
 		ret = -EINVAL;
 		goto out;
 	}

kmod/src/ext.h

@@ -15,6 +15,8 @@ struct scoutfs_ext_ops {
 		      u64 start, u64 len, u64 map, u8 flags);
 	int (*remove)(struct super_block *sb, void *arg, u64 start, u64 len,
 		      u64 map, u8 flags);
+
+	bool insert_overlap_warn;
 };
 
 bool scoutfs_ext_can_merge(struct scoutfs_extent *left,

kmod/src/forest.c

@@ -251,10 +251,8 @@ static int forest_read_items(struct super_block *sb, struct scoutfs_key *key,
  * that covers all the blocks.  Any keys outside of this range can't be
  * trusted because we didn't visit all the trees to check their items.
  *
- * If we hit stale blocks and retry we can call the callback for
- * duplicate items.  This is harmless because the items are stable while
- * the caller holds their cluster lock and the caller has to filter out
- * item seqs anyway.
+ * We return -ESTALE if we hit stale blocks to give the caller a chance
+ * to reset their state and retry with a newer version of the btrees.
  */
 int scoutfs_forest_read_items(struct super_block *sb,
 			      struct scoutfs_lock *lock,
@@ -263,7 +261,6 @@ int scoutfs_forest_read_items(struct super_block *sb,
 			      struct scoutfs_key *end,
 			      scoutfs_forest_item_cb cb, void *arg)
 {
-	DECLARE_STALE_TRACKING_SUPER_REFS(prev_refs, refs);
 	struct forest_read_items_data rid = {
 		.cb = cb,
 		.cb_arg = arg,
@@ -281,14 +278,11 @@ int scoutfs_forest_read_items(struct super_block *sb,
 	scoutfs_inc_counter(sb, forest_read_items);
 	calc_bloom_nrs(&bloom, &lock->start);
 
-retry:
 	ret = scoutfs_client_get_roots(sb, &roots);
 	if (ret)
 		goto out;
 
 	trace_scoutfs_forest_using_roots(sb, &roots.fs_root, &roots.logs_root);
-	refs.fs_ref = roots.fs_root.ref;
-	refs.logs_ref = roots.logs_root.ref;
 
 	*start = lock->start;
 	*end = lock->end;
@@ -352,13 +346,6 @@ retry:
 
 	ret = 0;
 out:
-	if (ret == -ESTALE) {
-		if (memcmp(&prev_refs, &refs, sizeof(refs)) == 0)
-			return -EIO;
-		prev_refs = refs;
-		goto retry;
-	}
-
 	return ret;
 }
 
@@ -642,7 +629,7 @@ static void scoutfs_forest_log_merge_worker(struct work_struct *work)
 	scoutfs_alloc_init(&alloc, &req.meta_avail, &req.meta_freed);
 	scoutfs_block_writer_init(sb, &wri);
 
-	/* find finalized input log trees up to last_seq */
+	/* find finalized input log trees within the input seq */
 	for (scoutfs_key_init_log_trees(&key, 0, 0); ; scoutfs_key_inc(&key)) {
 
 		if (!rhead) {
@@ -658,10 +645,9 @@ static void scoutfs_forest_log_merge_worker(struct work_struct *work)
 			if (iref.val_len == sizeof(*lt)) {
 				key = *iref.key;
 				lt = iref.val;
-				if ((le64_to_cpu(lt->flags) &
-				     SCOUTFS_LOG_TREES_FINALIZED) &&
-				    (le64_to_cpu(lt->max_item_seq) <=
-				     le64_to_cpu(req.last_seq))) {
+				if (lt->item_root.ref.blkno != 0 &&
+				    (le64_to_cpu(lt->flags) & SCOUTFS_LOG_TREES_FINALIZED) &&
+				    (le64_to_cpu(lt->finalize_seq) < le64_to_cpu(req.input_seq))) {
 					rhead->root = lt->item_root;
 					list_add_tail(&rhead->head, &inputs);
 					rhead = NULL;

kmod/src/format.h

@@ -457,6 +457,7 @@ struct scoutfs_log_trees {
 	__le64 data_alloc_zone_blocks;
 	__le64 data_alloc_zones[SCOUTFS_DATA_ALLOC_ZONE_LE64S];
 	__le64 max_item_seq;
+	__le64 finalize_seq;
 	__le64 rid;
 	__le64 nr;
 	__le64 flags;
@@ -508,7 +509,6 @@ struct scoutfs_log_merge_status {
 	struct scoutfs_key next_range_key;
 	__le64 nr_requests;
 	__le64 nr_complete;
-	__le64 last_seq;
 	__le64 seq;
 };
 
@@ -525,7 +525,7 @@ struct scoutfs_log_merge_request {
 	struct scoutfs_btree_root root;
 	struct scoutfs_key start;
 	struct scoutfs_key end;
-	__le64 last_seq;
+	__le64 input_seq;
 	__le64 rid;
 	__le64 seq;
 	__le64 flags;
@@ -973,6 +973,7 @@ enum scoutfs_net_cmd {
 	SCOUTFS_NET_CMD_ALLOC_INODES,
 	SCOUTFS_NET_CMD_GET_LOG_TREES,
 	SCOUTFS_NET_CMD_COMMIT_LOG_TREES,
+	SCOUTFS_NET_CMD_SYNC_LOG_TREES,
 	SCOUTFS_NET_CMD_GET_ROOTS,
 	SCOUTFS_NET_CMD_ADVANCE_SEQ,
 	SCOUTFS_NET_CMD_GET_LAST_SEQ,
@@ -1060,6 +1061,7 @@ enum scoutfs_lock_trace {
 	SLT_INVALIDATE,
 	SLT_REQUEST,
 	SLT_RESPONSE,
+	SLT_NR,
 };
 
 /*

kmod/src/inode.c

@@ -34,7 +34,7 @@
 #include "client.h"
 #include "cmp.h"
 #include "omap.h"
-#include "forest.h"
+#include "btree.h"
 
 /*
  * XXX
@@ -1735,15 +1735,20 @@ static void schedule_orphan_dwork(struct inode_sb_info *inf)
  * the cached inodes pinning the inode fail to delete as they are
  * evicted from the cache -- either through crashing or errors.
  *
- * This work runs in all mounts in the background looking for orphaned
- * inodes that should be deleted.
+ * This work runs in all mounts in the background looking for those
+ * orphaned inodes that weren't fully deleted.
  *
- * We use the forest hint call to read the persistent forest trees
- * looking for orphan items without creating lock contention.  Orphan
- * items exist for O_TMPFILE users and we don't want to force them to
- * commit by trying to acquire a conflicting read lock the orphan zone.
- * There's no rush to reclaim deleted items, eventually they will be
- * found in the persistent item btrees.
+ * First, we search for items in the current persistent fs root.  We'll
+ * only find orphan items that made it to the fs root after being merged
+ * from a mount's log btree.  This naturally avoids orphan items that
+ * exist while inodes have been unlinked but are still cached, including
+ * O_TMPFILE inodes that are actively used during normal operations.
+ * Scanning the read-only persistent fs root uses cached blocks and
+ * avoids the lock contention we'd cause if we tried to use the
+ * consistent item cache.  The downside is that it adds a bit of
+ * latency.  If an orphan was created in error it'll take until the
+ * mount's log btree is finalized and merged.  A crash will have the log
+ * btree merged after it is fenced.
  *
  * Once we find candidate orphan items we can first check our local
  * inode cache for inodes that are already on their way to eviction and
@@ -1751,10 +1756,6 @@ static void schedule_orphan_dwork(struct inode_sb_info *inf)
  * the inode.  Only if we don't have it cached, and no one else does, do
  * we try and read it into our cache and evict it to trigger the final
  * inode deletion process.
- *
- * Orphaned items that make it that far should be very rare.  They can
- * only exist if all the mounts that were using an inode after it had
- * been unlinked (or created with o_tmpfile) didn't unmount cleanly.
  */
 static void inode_orphan_scan_worker(struct work_struct *work)
 {
@@ -1762,8 +1763,9 @@ static void inode_orphan_scan_worker(struct work_struct *work)
 						 orphan_scan_dwork.work);
 	struct super_block *sb = inf->sb;
 	struct scoutfs_open_ino_map omap;
+	struct scoutfs_net_roots roots;
+	SCOUTFS_BTREE_ITEM_REF(iref);
 	struct scoutfs_key last;
-	struct scoutfs_key next;
 	struct scoutfs_key key;
 	struct inode *inode;
 	u64 group_nr;
@@ -1776,6 +1778,10 @@ static void inode_orphan_scan_worker(struct work_struct *work)
 	init_orphan_key(&last, U64_MAX);
 	omap.args.group_nr = cpu_to_le64(U64_MAX);
 
+	ret = scoutfs_client_get_roots(sb, &roots);
+	if (ret)
+		goto out;
+
 	for (ino = SCOUTFS_ROOT_INO + 1; ino != 0; ino++) {
 		if (inf->stopped) {
 			ret = 0;
@@ -1784,18 +1790,21 @@ static void inode_orphan_scan_worker(struct work_struct *work)
 
 		/* find the next orphan item */
 		init_orphan_key(&key, ino);
-		ret = scoutfs_forest_next_hint(sb, &key, &next);
+		ret = scoutfs_btree_next(sb, &roots.fs_root, &key, &iref);
 		if (ret < 0) {
 			if (ret == -ENOENT)
 				break;
 			goto out;
 		}
 
-		if (scoutfs_key_compare(&next, &last) > 0)
+		key = *iref.key;
+		scoutfs_btree_put_iref(&iref);
+
+		if (scoutfs_key_compare(&key, &last) > 0)
 			break;
 
 		scoutfs_inc_counter(sb, orphan_scan_item);
-		ino = le64_to_cpu(next.sko_ino);
+		ino = le64_to_cpu(key.sko_ino);
 
 		/* locally cached inodes will already be deleted */
 		inode = scoutfs_ilookup(sb, ino);
@@ -1972,7 +1981,11 @@ void scoutfs_inode_start(struct super_block *sb)
 	schedule_orphan_dwork(inf);
 }
 
-void scoutfs_inode_stop(struct super_block *sb)
+/*
+ * Orphan scanning can instantiate inodes.  We shut it down before
+ * calling into the vfs to tear down dentries and inodes during unmount.
+ */
+void scoutfs_inode_orphan_stop(struct super_block *sb)
 {
 	DECLARE_INODE_SB_INFO(sb, inf);
 
@@ -1982,6 +1995,14 @@ void scoutfs_inode_stop(struct super_block *sb)
 	}
 }
 
+void scoutfs_inode_flush_iput(struct super_block *sb)
+{
+	DECLARE_INODE_SB_INFO(sb, inf);
+
+	if (inf)
+		flush_work(&inf->iput_work);
+}
+
 void scoutfs_inode_destroy(struct super_block *sb)
 {
 	struct inode_sb_info *inf = SCOUTFS_SB(sb)->inode_sb_info;

kmod/src/inode.h

@@ -133,7 +133,8 @@ int scoutfs_inode_init(void);
 
 int scoutfs_inode_setup(struct super_block *sb);
 void scoutfs_inode_start(struct super_block *sb);
-void scoutfs_inode_stop(struct super_block *sb);
+void scoutfs_inode_orphan_stop(struct super_block *sb);
+void scoutfs_inode_flush_iput(struct super_block *sb);
 void scoutfs_inode_destroy(struct super_block *sb);
 
 #endif

kmod/src/item.c

@@ -1445,6 +1445,11 @@ static int read_page_item(struct super_block *sb, struct scoutfs_key *key,
  * locks protect the stable items we read.  Invalidation is careful not
  * to drop pages that have items that we couldn't see because they were
  * dirty when we started reading.
+ *
+ * The forest item reader is reading stable trees that could be
+ * overwritten.  It can return -ESTALE which we return to the caller who
+ * will retry the operation and work with a new set of more recent
+ * btrees.
  */
 static int read_pages(struct super_block *sb, struct item_cache_info *cinf,
 		      struct scoutfs_key *key, struct scoutfs_lock *lock)
@@ -1615,7 +1620,7 @@ retry:
 					       &lock->end);
 		else
 			ret = read_pages(sb, cinf, key, lock);
-		if (ret < 0)
+		if (ret < 0 && ret != -ESTALE)
 			goto out;
 		goto retry;
 	}

kmod/src/lock.c

@@ -66,8 +66,6 @@
  * relative to that lock state we resend.
  */
 
-#define GRACE_PERIOD_KT	ms_to_ktime(10)
-
 /*
  * allocated per-super, freed on unmount.
  */
@@ -82,9 +80,7 @@ struct lock_info {
 	struct list_head lru_list;
 	unsigned long long lru_nr;
 	struct workqueue_struct *workq;
-	struct work_struct grant_work;
-	struct list_head grant_list;
-	struct delayed_work inv_dwork;
+	struct work_struct inv_work;
 	struct list_head inv_list;
 	struct work_struct shrink_work;
 	struct list_head shrink_list;
@@ -255,7 +251,6 @@ static void lock_free(struct lock_info *linfo, struct scoutfs_lock *lock)
 	BUG_ON(!RB_EMPTY_NODE(&lock->node));
 	BUG_ON(!RB_EMPTY_NODE(&lock->range_node));
 	BUG_ON(!list_empty(&lock->lru_head));
-	BUG_ON(!list_empty(&lock->grant_head));
 	BUG_ON(!list_empty(&lock->inv_head));
 	BUG_ON(!list_empty(&lock->shrink_head));
 	BUG_ON(!list_empty(&lock->cov_list));
@@ -283,8 +278,8 @@ static struct scoutfs_lock *lock_alloc(struct super_block *sb,
 	RB_CLEAR_NODE(&lock->node);
 	RB_CLEAR_NODE(&lock->range_node);
 	INIT_LIST_HEAD(&lock->lru_head);
-	INIT_LIST_HEAD(&lock->grant_head);
 	INIT_LIST_HEAD(&lock->inv_head);
+	INIT_LIST_HEAD(&lock->inv_list);
 	INIT_LIST_HEAD(&lock->shrink_head);
 	spin_lock_init(&lock->cov_list_lock);
 	INIT_LIST_HEAD(&lock->cov_list);
@@ -331,23 +326,6 @@ static bool lock_counts_match(int granted, unsigned int *counts)
 	return true;
 }
 
-/*
- * Returns true if there are any mode counts that match with the desired
- * mode.  There can be other non-matching counts as well but we're only
- * testing for the existence of any matching counts.
- */
-static bool lock_count_match_exists(int desired, unsigned int *counts)
-{
-	enum scoutfs_lock_mode mode;
-
-	for (mode = 0; mode < SCOUTFS_LOCK_NR_MODES; mode++) {
-		if (counts[mode] && lock_modes_match(desired, mode))
-			return true;
-	}
-
-	return false;
-}
-
 /*
  * An idle lock has nothing going on.  It can be present in the lru and
  * can be freed by the final put when it has a null mode.
@@ -565,45 +543,15 @@ static void put_lock(struct lock_info *linfo,struct scoutfs_lock *lock)
 }
 
 /*
- * Locks have a grace period that extends after activity and prevents
- * invalidation.  It's intended to let nodes do reasonable batches of
- * work as locks ping pong between nodes that are doing conflicting
- * work.
- */
-static void extend_grace(struct super_block *sb, struct scoutfs_lock *lock)
-{
-	ktime_t now = ktime_get();
-
-	if (ktime_after(now, lock->grace_deadline))
-		scoutfs_inc_counter(sb, lock_grace_set);
-	else
-		scoutfs_inc_counter(sb, lock_grace_extended);
-
-	lock->grace_deadline = ktime_add(now, GRACE_PERIOD_KT);
-}
-
-static void queue_grant_work(struct lock_info *linfo)
-{
-	assert_spin_locked(&linfo->lock);
-
-	if (!list_empty(&linfo->grant_list))
-		queue_work(linfo->workq, &linfo->grant_work);
-}
-
-/*
- * We immediately queue work on the assumption that the caller might
- * have made a change (set a lock mode) which can let one of the
- * invalidating locks make forward progress, even if other locks are
- * waiting for their grace period to elapse.  It's a trade-off between
- * invalidation latency and burning cpu repeatedly finding that locks
- * are still in their grace period.
+ * The caller has made a change (set a lock mode) which can let one of the
+ * invalidating locks make forward progress.
  */
 static void queue_inv_work(struct lock_info *linfo)
 {
 	assert_spin_locked(&linfo->lock);
 
 	if (!list_empty(&linfo->inv_list))
-		mod_delayed_work(linfo->workq, &linfo->inv_dwork, 0);
+		queue_work(linfo->workq, &linfo->inv_work);
 }
 
 /*
@@ -651,72 +599,13 @@ static void bug_on_inconsistent_grant_cache(struct super_block *sb,
 }
 
 /*
- * Each lock has received a grant response message from the server.
+ * The client is receiving a grant response message from the server.
+ * This is being called synchronously in the networking receive path so
+ * our work should be quick and reasonably non-blocking.
  *
- * Grant responses can be reordered with incoming invalidation requests
- * from the server so we have to be careful to only set the new mode
- * once the old mode matches.
- *
- * We extend the grace period as we grant the lock if there is a waiting
- * locker who can use the lock.  This stops invalidation from pulling
- * the granted lock out from under the requester, resulting in a lot of
- * churn with no forward progress.  Using the grace period avoids having
- * to identify a specific waiter and give it an acquired lock.  It's
- * also very similar to waking up the locker and having it win the race
- * against the invalidation.  In that case they'd extend the grace
- * period anyway as they unlock.
- */
-static void lock_grant_worker(struct work_struct *work)
-{
-	struct lock_info *linfo = container_of(work, struct lock_info,
-					       grant_work);
-	struct super_block *sb = linfo->sb;
-	struct scoutfs_net_lock *nl;
-	struct scoutfs_lock *lock;
-	struct scoutfs_lock *tmp;
-
-	scoutfs_inc_counter(sb, lock_grant_work);
-
-	spin_lock(&linfo->lock);
-
-	list_for_each_entry_safe(lock, tmp, &linfo->grant_list, grant_head) {
-		nl = &lock->grant_nl;
-
-		/* wait for reordered invalidation to finish */
-		if (lock->mode != nl->old_mode)
-			continue;
-
-		bug_on_inconsistent_grant_cache(sb, lock, nl->old_mode,
-						nl->new_mode);
-
-		if (!lock_mode_can_read(nl->old_mode) &&
-		    lock_mode_can_read(nl->new_mode)) {
-			lock->refresh_gen =
-				atomic64_inc_return(&linfo->next_refresh_gen);
-		}
-
-		lock->request_pending = 0;
-		lock->mode = nl->new_mode;
-		lock->write_seq = le64_to_cpu(nl->write_seq);
-
-		if (lock_count_match_exists(nl->new_mode, lock->waiters))
-			extend_grace(sb, lock);
-
-		trace_scoutfs_lock_granted(sb, lock);
-		list_del_init(&lock->grant_head);
-		wake_up(&lock->waitq);
-		put_lock(linfo, lock);
-	}
-
-	/* invalidations might be waiting for our reordered grant */
-	queue_inv_work(linfo);
-	spin_unlock(&linfo->lock);
-}
-
-/*
- * The client is receiving a grant response message from the server.  We
- * find the lock, record the response, and add it to the list for grant
- * work to process.
+ * The server's state machine can immediately send an invalidate request
+ * after sending this grant response.   We won't process the incoming
+ * invalidate request until after processing this grant response.
  */
 int scoutfs_lock_grant_response(struct super_block *sb,
 				struct scoutfs_net_lock *nl)
@@ -734,64 +623,61 @@ int scoutfs_lock_grant_response(struct super_block *sb,
 	trace_scoutfs_lock_grant_response(sb, lock);
 	BUG_ON(!lock->request_pending);
 
-	lock->grant_nl = *nl;
-	list_add_tail(&lock->grant_head, &linfo->grant_list);
-	queue_grant_work(linfo);
+	bug_on_inconsistent_grant_cache(sb, lock, nl->old_mode, nl->new_mode);
+
+	if (!lock_mode_can_read(nl->old_mode) && lock_mode_can_read(nl->new_mode))
+		lock->refresh_gen = atomic64_inc_return(&linfo->next_refresh_gen);
+
+	lock->request_pending = 0;
+	lock->mode = nl->new_mode;
+	lock->write_seq = le64_to_cpu(nl->write_seq);
+
+	trace_scoutfs_lock_granted(sb, lock);
+	wake_up(&lock->waitq);
+	put_lock(linfo, lock);
 
 	spin_unlock(&linfo->lock);
 
 	return 0;
 }
 
+struct inv_req {
+	struct list_head head;
+	struct scoutfs_lock *lock;
+	u64 net_id;
+	struct scoutfs_net_lock nl;
+};
+
 /*
  * Each lock has received a lock invalidation request from the server
- * which specifies a new mode for the lock.  The server will only send
- * one invalidation request at a time for each lock.  The server can
- * send another invalidate request after we send the response but before
- * we reacquire the lock and finish invalidation.
+ * which specifies a new mode for the lock.  Our processing state
+ * machine and server failover and lock recovery can both conspire to
+ * give us triplicate invalidation requests.  The incoming requests for
+ * a given lock need to be processed in order, but we can process locks
+ * in any order.
  *
  * This is an unsolicited request from the server so it can arrive at
- * any time after we make the server aware of the lock by initially
- * requesting it.  We wait for users of the current mode to unlock
- * before invalidating.
+ * any time after we make the server aware of the lock.  We wait for
+ * users of the current mode to unlock before invalidating.
  *
  * This can arrive on behalf of our request for a mode that conflicts
  * with our current mode.  We have to proceed while we have a request
  * pending.  We can also be racing with shrink requests being sent while
  * we're invalidating.
  *
- * This can be processed concurrently and experience reordering with a
- * grant response sent back-to-back from the server.  We carefully only
- * invalidate once the lock mode matches what the server told us to
- * invalidate.
- *
- * We delay invalidation processing until a grace period has elapsed
- * since the last unlock.  The intent is to let users do a reasonable
- * batch of work before dropping the lock.  Continuous unlocking can
- * continuously extend the deadline.
- *
  * Before we start invalidating the lock we set the lock to the new
  * mode, preventing further incompatible users of the old mode from
  * using the lock while we're invalidating.
- *
- * This does a lot of serialized inode invalidation in one context and
- * performs a lot of repeated calls to sync.  It would be nice to get
- * some concurrent inode invalidation and to more carefully only call
- * sync when needed.
  */
 static void lock_invalidate_worker(struct work_struct *work)
 {
-	struct lock_info *linfo = container_of(work, struct lock_info,
-					       inv_dwork.work);
+	struct lock_info *linfo = container_of(work, struct lock_info, inv_work);
 	struct super_block *sb = linfo->sb;
 	struct scoutfs_net_lock *nl;
 	struct scoutfs_lock *lock;
 	struct scoutfs_lock *tmp;
-	unsigned long delay = MAX_JIFFY_OFFSET;
-	ktime_t now = ktime_get();
-	ktime_t deadline;
+	struct inv_req *ireq;
 	LIST_HEAD(ready);
-	u64 net_id;
 	int ret;
 
 	scoutfs_inc_counter(sb, lock_invalidate_work);
@@ -799,25 +685,13 @@ static void lock_invalidate_worker(struct work_struct *work)
 	spin_lock(&linfo->lock);
 
 	list_for_each_entry_safe(lock, tmp, &linfo->inv_list, inv_head) {
-		nl = &lock->inv_nl;
-
-		/* wait for reordered grant to finish */
-		if (lock->mode != nl->old_mode)
-			continue;
+		ireq = list_first_entry(&lock->inv_list, struct inv_req, head);
+		nl = &ireq->nl;
 
 		/* wait until incompatible holders unlock */
 		if (!lock_counts_match(nl->new_mode, lock->users))
 			continue;
 
-		/* skip if grace hasn't elapsed, record earliest */
-		deadline = lock->grace_deadline;
-		if (!linfo->shutdown && ktime_before(now, deadline)) {
-			delay = min(delay,
-				    nsecs_to_jiffies(ktime_to_ns(
-						ktime_sub(deadline, now))));
-			scoutfs_inc_counter(linfo->sb, lock_grace_wait);
-			continue;
-		}
 		/* set the new mode, no incompatible users during inval */
 		lock->mode = nl->new_mode;
 
@@ -828,12 +702,12 @@ static void lock_invalidate_worker(struct work_struct *work)
 	spin_unlock(&linfo->lock);
 
 	if (list_empty(&ready))
-		goto out;
+		return;
 
 	/* invalidate once the lock is read */
 	list_for_each_entry(lock, &ready, inv_head) {
-		nl = &lock->inv_nl;
-		net_id = lock->inv_net_id;
+		ireq = list_first_entry(&lock->inv_list, struct inv_req, head);
+		nl = &ireq->nl;
 
 		/* only lock protocol, inv can't call subsystems after shutdown */
 		if (!linfo->shutdown) {
@@ -841,11 +715,10 @@ static void lock_invalidate_worker(struct work_struct *work)
 			BUG_ON(ret);
 		}
 
-		/* allow another request after we respond but before we finish */
-		lock->inv_net_id = 0;
-
-		/* respond with the key and modes from the request */
-		ret = scoutfs_client_lock_response(sb, net_id, nl);
+		/* respond with the key and modes from the request, server might have died */
+		ret = scoutfs_client_lock_response(sb, ireq->net_id, nl);
+		if (ret == -ENOTCONN)
+			ret = 0;
 		BUG_ON(ret);
 
 		scoutfs_inc_counter(sb, lock_invalidate_response);
@@ -855,71 +728,87 @@ static void lock_invalidate_worker(struct work_struct *work)
 	spin_lock(&linfo->lock);
 
 	list_for_each_entry_safe(lock, tmp, &ready, inv_head) {
+		ireq = list_first_entry(&lock->inv_list, struct inv_req, head);
+
 		trace_scoutfs_lock_invalidated(sb, lock);
-		if (lock->inv_net_id == 0) {
+
+		list_del(&ireq->head);
+		kfree(ireq);
+
+		if (list_empty(&lock->inv_list)) {
 			/* finish if another request didn't arrive */
 			list_del_init(&lock->inv_head);
 			lock->invalidate_pending = 0;
 			wake_up(&lock->waitq);
 		} else {
-			/* another request filled nl/net_id, back on the list and requeue */
+			/* another request arrived, back on the list and requeue */
 			list_move_tail(&lock->inv_head, &linfo->inv_list);
 			queue_inv_work(linfo);
 		}
+
 		put_lock(linfo, lock);
 	}
 
-	/* grant might have been waiting for invalidate request */
-	queue_grant_work(linfo);
 	spin_unlock(&linfo->lock);
-
-out:
-	/* queue delayed work if invalidations waiting on grace deadline */
-	if (delay != MAX_JIFFY_OFFSET)
-		queue_delayed_work(linfo->workq, &linfo->inv_dwork, delay);
 }
 
 /*
- * Record an incoming invalidate request from the server and add its
- * lock to the list for processing.  This request can be from a new
- * server and racing with invalidation that frees from an old server.
- * It's fine to not find the requested lock and send an immediate
- * response.
+ * Add an incoming invalidation request to the end of the list on the
+ * lock and queue it for blocking invalidation work.   This is being
+ * called synchronously in the net recv path to avoid reordering with
+ * grants that were sent immediately before the server sent this
+ * invalidation.
  *
- * The invalidation process drops the linfo lock to send responses.  The
- * moment it does so we can receive another invalidation request (the
- * server can ask us to go from write->read then read->null).  We allow
- * for one chain like this but it's a bug if we receive more concurrent
- * invalidation requests than that.  The server should be only sending
- * one at a time.
+ * Incoming invalidation requests are a function of the remote lock
+ * server's state machine and are slightly decoupled from our lock
+ * state.  We can receive duplicate requests if the server is quick
+ * enough to send the next request after we send a previous reply, or if
+ * pending invalidation spans server failover and lock recovery.
+ *
+ * Similarly, we can get a request to invalidate a lock we don't have if
+ * invalidation finished just after lock recovery to a new server.
+ * Happily we can just reply because we satisfy the invalidation
+ * response promise to not be using the old lock's mode if the lock
+ * doesn't exist.
  */
 int scoutfs_lock_invalidate_request(struct super_block *sb, u64 net_id,
 				    struct scoutfs_net_lock *nl)
 {
 	DECLARE_LOCK_INFO(sb, linfo);
-	struct scoutfs_lock *lock;
+	struct scoutfs_lock *lock = NULL;
+	struct inv_req *ireq;
 	int ret = 0;
 
 	scoutfs_inc_counter(sb, lock_invalidate_request);
 
+	ireq = kmalloc(sizeof(struct inv_req), GFP_NOFS);
+	BUG_ON(!ireq); /* lock server doesn't handle response errors */
+	if (ireq == NULL) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
 	spin_lock(&linfo->lock);
 	lock = get_lock(sb, &nl->key);
 	if (lock) {
-		BUG_ON(lock->inv_net_id != 0);
-		lock->inv_net_id = net_id;
-		lock->inv_nl = *nl;
-		if (list_empty(&lock->inv_head)) {
+		trace_scoutfs_lock_invalidate_request(sb, lock);
+		ireq->lock = lock;
+		ireq->net_id = net_id;
+		ireq->nl = *nl;
+		if (list_empty(&lock->inv_list)) {
 			list_add_tail(&lock->inv_head, &linfo->inv_list);
 			lock->invalidate_pending = 1;
 			queue_inv_work(linfo);
-			/* otherwise inv work queues itself when it sees inv_net_id */
 		}
-		trace_scoutfs_lock_invalidate_request(sb, lock);
+		list_add_tail(&ireq->head, &lock->inv_list);
 	}
 	spin_unlock(&linfo->lock);
 
-	if (!lock)
+out:
+	if (!lock) {
 		ret = scoutfs_client_lock_response(sb, net_id, nl);
+		BUG_ON(ret); /* lock server doesn't fence timed out client requests */
+	}
 
 	return ret;
 }
@@ -1348,10 +1237,6 @@ int scoutfs_lock_orphan(struct super_block *sb, enum scoutfs_lock_mode mode, int
 	return lock_key_range(sb, mode, flags, &start, &end, lock);
 }
 
-/*
- * As we unlock we always extend the grace period to give the caller
- * another pass at the lock before its invalidated.
- */
 void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, enum scoutfs_lock_mode mode)
 {
 	DECLARE_LOCK_INFO(sb, linfo);
@@ -1364,7 +1249,6 @@ void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, enum scou
 	spin_lock(&linfo->lock);
 
 	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);
 
@@ -1604,10 +1488,18 @@ void scoutfs_lock_unmount_begin(struct super_block *sb)
 
 	if (linfo) {
 		linfo->unmounting = true;
-		flush_delayed_work(&linfo->inv_dwork);
+		flush_work(&linfo->inv_work);
 	}
 }
 
+void scoutfs_lock_flush_invalidate(struct super_block *sb)
+{
+	DECLARE_LOCK_INFO(sb, linfo);
+
+	if (linfo)
+		flush_work(&linfo->inv_work);
+}
+
 /*
  * The caller is going to be shutting down transactions and the client.
  * We need to make sure that locking won't call either after we return.
@@ -1671,6 +1563,8 @@ void scoutfs_lock_destroy(struct super_block *sb)
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	DECLARE_LOCK_INFO(sb, linfo);
 	struct scoutfs_lock *lock;
+	struct inv_req *ireq_tmp;
+	struct inv_req *ireq;
 	struct rb_node *node;
 	enum scoutfs_lock_mode mode;
 
@@ -1697,8 +1591,6 @@ void scoutfs_lock_destroy(struct super_block *sb)
 	spin_unlock(&linfo->lock);
 
 	if (linfo->workq) {
-		/* pending grace work queues normal work */
-		flush_workqueue(linfo->workq);
 		/* now all work won't queue itself */
 		destroy_workqueue(linfo->workq);
 	}
@@ -1715,15 +1607,21 @@ void scoutfs_lock_destroy(struct super_block *sb)
 	 * of free).
 	 */
 	spin_lock(&linfo->lock);
+
 	node = rb_first(&linfo->lock_tree);
 	while (node) {
 		lock = rb_entry(node, struct scoutfs_lock, node);
 		node = rb_next(node);
+
+		list_for_each_entry_safe(ireq, ireq_tmp, &lock->inv_list, head) {
+			list_del_init(&ireq->head);
+			put_lock(linfo, ireq->lock);
+			kfree(ireq);
+		}
+
 		lock->request_pending = 0;
 		if (!list_empty(&lock->lru_head))
 			__lock_del_lru(linfo, lock);
-		if (!list_empty(&lock->grant_head))
-			list_del_init(&lock->grant_head);
 		if (!list_empty(&lock->inv_head)) {
 			list_del_init(&lock->inv_head);
 			lock->invalidate_pending = 0;
@@ -1733,6 +1631,7 @@ void scoutfs_lock_destroy(struct super_block *sb)
 		lock_remove(linfo, lock);
 		lock_free(linfo, lock);
 	}
+
 	spin_unlock(&linfo->lock);
 
 	kfree(linfo);
@@ -1757,9 +1656,7 @@ int scoutfs_lock_setup(struct super_block *sb)
 	linfo->shrinker.seeks = DEFAULT_SEEKS;
 	register_shrinker(&linfo->shrinker);
 	INIT_LIST_HEAD(&linfo->lru_list);
-	INIT_WORK(&linfo->grant_work, lock_grant_worker);
-	INIT_LIST_HEAD(&linfo->grant_list);
-	INIT_DELAYED_WORK(&linfo->inv_dwork, lock_invalidate_worker);
+	INIT_WORK(&linfo->inv_work, lock_invalidate_worker);
 	INIT_LIST_HEAD(&linfo->inv_list);
 	INIT_WORK(&linfo->shrink_work, lock_shrink_worker);
 	INIT_LIST_HEAD(&linfo->shrink_list);

kmod/src/lock.h

@@ -28,15 +28,11 @@ struct scoutfs_lock {
 	u64 dirty_trans_seq;
 	struct list_head lru_head;
 	wait_queue_head_t waitq;
-	ktime_t grace_deadline;
 	unsigned long request_pending:1,
 		      invalidate_pending:1;
 
-	struct list_head grant_head;
-	struct scoutfs_net_lock grant_nl;
-	struct list_head inv_head;
-	struct scoutfs_net_lock inv_nl;
-	u64 inv_net_id;
+	struct list_head inv_head;  /* entry in linfo's list of locks with invalidations */
+	struct list_head inv_list;  /* list of lock's invalidation requests */
 	struct list_head shrink_head;
 
 	spinlock_t cov_list_lock;
@@ -106,6 +102,7 @@ void scoutfs_free_unused_locks(struct super_block *sb);
 
 int scoutfs_lock_setup(struct super_block *sb);
 void scoutfs_lock_unmount_begin(struct super_block *sb);
+void scoutfs_lock_flush_invalidate(struct super_block *sb);
 void scoutfs_lock_shutdown(struct super_block *sb);
 void scoutfs_lock_destroy(struct super_block *sb);
 

kmod/src/lock_server.c

@@ -78,6 +78,8 @@ struct lock_server_info {
 
 	struct scoutfs_tseq_tree tseq_tree;
 	struct dentry *tseq_dentry;
+	struct scoutfs_tseq_tree stats_tseq_tree;
+	struct dentry *stats_tseq_dentry;
 
 	struct scoutfs_alloc *alloc;
 	struct scoutfs_block_writer *wri;
@@ -107,6 +109,9 @@ struct server_lock_node {
 	struct list_head granted;
 	struct list_head requested;
 	struct list_head invalidated;
+
+	struct scoutfs_tseq_entry stats_tseq_entry;
+	u64 stats[SLT_NR];
 };
 
 /*
@@ -296,6 +301,8 @@ static struct server_lock_node *alloc_server_lock(struct lock_server_info *inf,
 			snode = get_server_lock(inf, key, ins, false);
 			if (snode != ins)
 				kfree(ins);
+			else
+				scoutfs_tseq_add(&inf->stats_tseq_tree, &snode->stats_tseq_entry);
 		}
 	}
 
@@ -325,8 +332,10 @@ static void put_server_lock(struct lock_server_info *inf,
 
 	mutex_unlock(&snode->mutex);
 
-	if (should_free)
+	if (should_free) {
+		scoutfs_tseq_del(&inf->stats_tseq_tree, &snode->stats_tseq_entry);
 		kfree(snode);
+	}
 }
 
 static struct client_lock_entry *find_entry(struct server_lock_node *snode,
@@ -388,6 +397,8 @@ int scoutfs_lock_server_request(struct super_block *sb, u64 rid,
 		goto out;
 	}
 
+	snode->stats[SLT_REQUEST]++;
+
 	clent->snode = snode;
 	add_client_entry(snode, &snode->requested, clent);
 	scoutfs_tseq_add(&inf->tseq_tree, &clent->tseq_entry);
@@ -428,6 +439,8 @@ int scoutfs_lock_server_response(struct super_block *sb, u64 rid,
 		goto out;
 	}
 
+	snode->stats[SLT_RESPONSE]++;
+
 	clent = find_entry(snode, &snode->invalidated, rid);
 	if (!clent) {
 		put_server_lock(inf, snode);
@@ -508,6 +521,7 @@ static int process_waiting_requests(struct super_block *sb,
 			trace_scoutfs_lock_message(sb, SLT_SERVER,
 						   SLT_INVALIDATE, SLT_REQUEST,
 						   gr->rid, 0, &nl);
+			snode->stats[SLT_INVALIDATE]++;
 
 			add_client_entry(snode, &snode->invalidated, gr);
 		}
@@ -544,6 +558,7 @@ static int process_waiting_requests(struct super_block *sb,
 		trace_scoutfs_lock_message(sb, SLT_SERVER, SLT_GRANT,
 					   SLT_RESPONSE, req->rid,
 					   req->net_id, &nl);
+		snode->stats[SLT_GRANT]++;
 
 		/* don't track null client locks, track all else */ 
 		if (req->mode == SCOUTFS_LOCK_NULL)
@@ -786,6 +801,16 @@ static void lock_server_tseq_show(struct seq_file *m,
 		   clent->net_id);
 }
 
+static void stats_tseq_show(struct seq_file *m, struct scoutfs_tseq_entry *ent)
+{
+	struct server_lock_node *snode = container_of(ent, struct server_lock_node,
+						      stats_tseq_entry);
+
+	seq_printf(m, SK_FMT" req %llu inv %llu rsp %llu gr %llu\n",
+		   SK_ARG(&snode->key), snode->stats[SLT_REQUEST], snode->stats[SLT_INVALIDATE],
+		   snode->stats[SLT_RESPONSE], snode->stats[SLT_GRANT]);
+}
+
 /*
  * Setup the lock server.  This is called before networking can deliver
  * requests.
@@ -805,6 +830,7 @@ int scoutfs_lock_server_setup(struct super_block *sb,
 	spin_lock_init(&inf->lock);
 	inf->locks_root = RB_ROOT;
 	scoutfs_tseq_tree_init(&inf->tseq_tree, lock_server_tseq_show);
+	scoutfs_tseq_tree_init(&inf->stats_tseq_tree, stats_tseq_show);
 	inf->alloc = alloc;
 	inf->wri = wri;
 
@@ -815,6 +841,14 @@ int scoutfs_lock_server_setup(struct super_block *sb,
 		return -ENOMEM;
 	}
 
+	inf->stats_tseq_dentry = scoutfs_tseq_create("server_lock_stats", sbi->debug_root,
+						     &inf->stats_tseq_tree);
+	if (!inf->stats_tseq_dentry) {
+		debugfs_remove(inf->tseq_dentry);
+		kfree(inf);
+		return -ENOMEM;
+	}
+
 	sbi->lock_server_info = inf;
 
 	return 0;
@@ -836,6 +870,7 @@ void scoutfs_lock_server_destroy(struct super_block *sb)
 
 	if (inf) {
 		debugfs_remove(inf->tseq_dentry);
+		debugfs_remove(inf->stats_tseq_dentry);
 
 		rbtree_postorder_for_each_entry_safe(snode, stmp,
 						     &inf->locks_root, node) {

kmod/src/msg.h

@@ -4,6 +4,7 @@
 #include <linux/bitops.h>
 #include "key.h"
 #include "counters.h"
+#include "super.h"
 
 void __printf(4, 5) scoutfs_msg(struct super_block *sb, const char *prefix,
 				const char *str, const char *fmt, ...);
@@ -23,6 +24,9 @@ do {							\
 #define scoutfs_info(sb, fmt, args...) \
 	scoutfs_msg_check(sb, KERN_INFO, "", fmt, ##args)
 
+#define scoutfs_tprintk(sb, fmt, args...) \
+	trace_printk(SCSBF " " fmt "\n", SCSB_ARGS(sb), ##args);
+
 #define scoutfs_bug_on(sb, cond, fmt, args...)				\
 do {									\
 	if (cond) {							\

kmod/src/net.c

@@ -677,8 +677,15 @@ static void scoutfs_net_recv_worker(struct work_struct *work)
 
 		scoutfs_tseq_add(&ninf->msg_tseq_tree, &mrecv->tseq_entry);
 
-		/* synchronously process greeting before next recvmsg */
-		if (nh.cmd == SCOUTFS_NET_CMD_GREETING)
+		/*
+		 * Initial received greetings are processed
+		 * synchronously before any other incoming messages.
+		 *
+		 * Incoming requests or responses to the lock client are
+		 * called synchronously to avoid reordering.
+		 */
+		if (nh.cmd == SCOUTFS_NET_CMD_GREETING ||
+		    (nh.cmd == SCOUTFS_NET_CMD_LOCK && !conn->listening_conn))
 			scoutfs_net_proc_worker(&mrecv->proc_work);
 		else
 			queue_work(conn->workq, &mrecv->proc_work);

kmod/src/quorum.c

@@ -392,6 +392,51 @@ out:
 	return ret;
 }
 
+/*
+ * It's really important in raft elections that the term not go
+ * backwards in time.  We achieve this by having each participant record
+ * the greatest term they've seen in their quorum block.  It's also
+ * important that participants agree on the greatest term.  It can
+ * happen that one gets ahead of the rest, perhaps by being forcefully
+ * shutdown after having just been elected.  As everyone starts up it's
+ * possible to have N-1 have term T-1 while just one participant thinks
+ * the term is T.   That single participant will ignore all messages
+ * from older terms.  If its timeout is greater then the others it can
+ * immediately override the election of the majority and request votes
+ * and become elected.
+ *
+ * A best-effort work around is to have everyone try and start from the
+ * greatest term that they can find in everyone's blocks.  If it works
+ * then you avoid having those with greater terms ignore others.  If it
+ * doesn't work the elections will eventually stabilize after rocky
+ * periods of fencing from what looks like concurrent elections.
+ */
+static void read_greatest_term(struct super_block *sb, u64 *term)
+{
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	struct scoutfs_super_block *super = &sbi->super;
+	struct scoutfs_quorum_block blk;
+	int ret;
+	int e;
+	int s;
+
+	*term = 0;
+
+	for (s = 0; s < SCOUTFS_QUORUM_MAX_SLOTS; s++) {
+		if (!quorum_slot_present(super, s))
+			continue;
+
+		ret = read_quorum_block(sb, SCOUTFS_QUORUM_BLKNO + s, &blk, false);
+		if (ret < 0)
+			continue;
+
+		for (e = 0; e < ARRAY_SIZE(blk.events); e++) {
+			if (blk.events[e].rid)
+				*term = max(*term, le64_to_cpu(blk.events[e].term));
+		}
+	}
+}
+
 static void set_quorum_block_event(struct super_block *sb, struct scoutfs_quorum_block *blk,
 				   int event, u64 term)
 {
@@ -576,29 +621,24 @@ static void scoutfs_quorum_worker(struct work_struct *work)
 	struct quorum_info *qinf = container_of(work, struct quorum_info, work);
 	struct super_block *sb = qinf->sb;
 	struct mount_options *opts = &SCOUTFS_SB(sb)->opts;
-	struct scoutfs_quorum_block blk;
 	struct sockaddr_in unused;
 	struct quorum_host_msg msg;
 	struct quorum_status qst;
-	u64 blkno;
 	int ret;
 	int err;
 
 	/* recording votes from slots as native single word bitmap */
 	BUILD_BUG_ON(SCOUTFS_QUORUM_MAX_SLOTS > BITS_PER_LONG);
 
-	/* get our starting term from our persistent block */
-	blkno = SCOUTFS_QUORUM_BLKNO + opts->quorum_slot_nr;
-	ret = read_quorum_block(sb, blkno, &blk, false);
-	if (ret < 0)
-		goto out;
-
 	/* start out as a follower */
 	qst.role = FOLLOWER;
-	qst.term = le64_to_cpu(blk.events[SCOUTFS_QUORUM_EVENT_TERM].term);
+	qst.term = 0;
 	qst.vote_for = -1;
 	qst.vote_bits = 0;
 
+	/* read our starting term from greatest in all events in all slots */
+	read_greatest_term(sb, &qst.term);
+
 	/* see if there's a server to chose heartbeat or election timeout */
 	if (scoutfs_quorum_server_sin(sb, &unused) == 0)
 		qst.timeout = heartbeat_timeout();
@@ -697,11 +737,10 @@ static void scoutfs_quorum_worker(struct work_struct *work)
 		/* candidates count votes in their term */
 		if (qst.role == CANDIDATE &&
 		    msg.type == SCOUTFS_QUORUM_MSG_VOTE) {
-			if (test_bit(msg.from, &qst.vote_bits)) {
+			if (test_and_set_bit(msg.from, &qst.vote_bits)) {
 				scoutfs_warn(sb, "already received vote from %u in term %llu, are there multiple mounts with quorum_slot_nr=%u?",
 					     msg.from, qst.term, msg.from);
 			}
-			set_bit(msg.from, &qst.vote_bits);
 			scoutfs_inc_counter(sb, quorum_recv_vote);
 		}
 
@@ -1007,13 +1046,17 @@ static inline bool valid_ipv4_port(__be16 port)
 static int verify_quorum_slots(struct super_block *sb)
 {
 	struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super;
+	struct mount_options *opts = &SCOUTFS_SB(sb)->opts;
+	char slots[(SCOUTFS_QUORUM_MAX_SLOTS * 3) + 1];
 	DECLARE_QUORUM_INFO(sb, qinf);
 	struct sockaddr_in other;
 	struct sockaddr_in sin;
 	int found = 0;
+	int ret;
 	int i;
 	int j;
 
+
 	for (i = 0; i < SCOUTFS_QUORUM_MAX_SLOTS; i++) {
 		if (!quorum_slot_present(super, i))
 			continue;
@@ -1054,6 +1097,25 @@ static int verify_quorum_slots(struct super_block *sb)
 		return -EINVAL;
 	}
 
+	if (!quorum_slot_present(super, opts->quorum_slot_nr)) {
+		char *str = slots;
+		*str = '\0';
+		for (i = 0; i < SCOUTFS_QUORUM_MAX_SLOTS; i++) {
+			if (quorum_slot_present(super, i)) {
+				ret = snprintf(str, &slots[ARRAY_SIZE(slots)] - str, "%c%u",
+					       str == slots ? ' ' : ',', i);
+				if (ret < 2 || ret > 3) {
+					scoutfs_err(sb, "error gathering populated slots");
+					return -EINVAL;
+				}
+				str += ret;
+			}
+		}
+		scoutfs_err(sb, "quorum_slot_nr=%u option references unused slot, must be one of the following configured slots:%s",
+			    opts->quorum_slot_nr, slots);
+		return -EINVAL;
+	}
+
 	/*
 	 * Always require a majority except in the pathological cases of
 	 * 1 or 2 members.

kmod/src/scoutfs_trace.h

@@ -406,21 +406,24 @@ TRACE_EVENT(scoutfs_sync_fs,
 );
 
 TRACE_EVENT(scoutfs_trans_write_func,
-	TP_PROTO(struct super_block *sb, unsigned long dirty),
+	TP_PROTO(struct super_block *sb, u64 dirty_block_bytes, u64 dirty_item_pages),
 
-	TP_ARGS(sb, dirty),
+	TP_ARGS(sb, dirty_block_bytes, dirty_item_pages),
 
 	TP_STRUCT__entry(
 		SCSB_TRACE_FIELDS
-		__field(unsigned long, dirty)
+		__field(__u64, dirty_block_bytes)
+		__field(__u64, dirty_item_pages)
 	),
 
 	TP_fast_assign(
 		SCSB_TRACE_ASSIGN(sb);
-		__entry->dirty = dirty;
+		__entry->dirty_block_bytes = dirty_block_bytes;
+		__entry->dirty_item_pages = dirty_item_pages;
 	),
 
-	TP_printk(SCSBF" dirty %lu", SCSB_TRACE_ARGS, __entry->dirty)
+	TP_printk(SCSBF" dirty_block_bytes %llu dirty_item_pages %llu",
+		  SCSB_TRACE_ARGS, __entry->dirty_block_bytes, __entry->dirty_item_pages)
 );
 
 DECLARE_EVENT_CLASS(scoutfs_trans_hold_release_class,
@@ -2045,9 +2048,9 @@ TRACE_EVENT(scoutfs_trans_seq_last,
 
 TRACE_EVENT(scoutfs_get_log_merge_status,
 	TP_PROTO(struct super_block *sb, u64 rid, struct scoutfs_key *next_range_key,
-		 u64 nr_requests, u64 nr_complete, u64 last_seq, u64 seq),
+		 u64 nr_requests, u64 nr_complete, u64 seq),
 
-	TP_ARGS(sb, rid, next_range_key, nr_requests, nr_complete, last_seq, seq),
+	TP_ARGS(sb, rid, next_range_key, nr_requests, nr_complete, seq),
 
 	TP_STRUCT__entry(
 		SCSB_TRACE_FIELDS
@@ -2055,7 +2058,6 @@ TRACE_EVENT(scoutfs_get_log_merge_status,
 		sk_trace_define(next_range_key)
 		__field(__u64, nr_requests)
 		__field(__u64, nr_complete)
-		__field(__u64, last_seq)
 		__field(__u64, seq)
 	),
 
@@ -2065,21 +2067,20 @@ TRACE_EVENT(scoutfs_get_log_merge_status,
 		sk_trace_assign(next_range_key, next_range_key);
 		__entry->nr_requests = nr_requests;
 		__entry->nr_complete = nr_complete;
-		__entry->last_seq = last_seq;
 		__entry->seq = seq;
 	),
 
-	TP_printk(SCSBF" rid %016llx next_range_key "SK_FMT" nr_requests %llu nr_complete %llu last_seq %llu seq %llu",
+	TP_printk(SCSBF" rid %016llx next_range_key "SK_FMT" nr_requests %llu nr_complete %llu seq %llu",
 		  SCSB_TRACE_ARGS, __entry->s_rid, sk_trace_args(next_range_key),
-		  __entry->nr_requests, __entry->nr_complete, __entry->last_seq, __entry->seq)
+		  __entry->nr_requests, __entry->nr_complete, __entry->seq)
 );
 
 TRACE_EVENT(scoutfs_get_log_merge_request,
 	TP_PROTO(struct super_block *sb, u64 rid,
 		 struct scoutfs_btree_root *root, struct scoutfs_key *start,
-		 struct scoutfs_key *end, u64 last_seq, u64 seq),
+		 struct scoutfs_key *end, u64 input_seq, u64 seq),
 
-	TP_ARGS(sb, rid, root, start, end, last_seq, seq),
+	TP_ARGS(sb, rid, root, start, end, input_seq, seq),
 
 	TP_STRUCT__entry(
 		SCSB_TRACE_FIELDS
@@ -2089,7 +2090,7 @@ TRACE_EVENT(scoutfs_get_log_merge_request,
 		__field(__u8, root_height)
 		sk_trace_define(start)
 		sk_trace_define(end)
-		__field(__u64, last_seq)
+		__field(__u64, input_seq)
 		__field(__u64, seq)
 	),
 
@@ -2101,14 +2102,14 @@ TRACE_EVENT(scoutfs_get_log_merge_request,
 		__entry->root_height = root->height;
 		sk_trace_assign(start, start);
 		sk_trace_assign(end, end);
-		__entry->last_seq = last_seq;
+		__entry->input_seq = input_seq;
 		__entry->seq = seq;
 	),
 
-	TP_printk(SCSBF" rid %016llx root blkno %llu seq %llu height %u start "SK_FMT" end "SK_FMT" last_seq %llu seq %llu",
+	TP_printk(SCSBF" rid %016llx root blkno %llu seq %llu height %u start "SK_FMT" end "SK_FMT" input_seq %llu seq %llu",
 		  SCSB_TRACE_ARGS, __entry->s_rid, __entry->root_blkno,
 		  __entry->root_seq, __entry->root_height,
-		  sk_trace_args(start), sk_trace_args(end), __entry->last_seq,
+		  sk_trace_args(start), sk_trace_args(end), __entry->input_seq,
 		  __entry->seq)
 );
 

kmod/src/super.c

@@ -255,7 +255,16 @@ static void scoutfs_put_super(struct super_block *sb)
 
 	trace_scoutfs_put_super(sb);
 
-	scoutfs_inode_stop(sb);
+	/*
+	 * Wait for invalidation and iput to finish with any lingering
+	 * inode references that escaped the evict_inodes in
+	 * generic_shutdown_super.  MS_ACTIVE is clear so final iput
+	 * will always evict.
+	 */
+	scoutfs_lock_flush_invalidate(sb);
+	scoutfs_inode_flush_iput(sb);
+	WARN_ON_ONCE(!list_empty(&sb->s_inodes));
+
 	scoutfs_forest_stop(sb);
 	scoutfs_srch_destroy(sb);
 
@@ -661,10 +670,17 @@ static struct dentry *scoutfs_mount(struct file_system_type *fs_type, int flags,
  */
 static void scoutfs_kill_sb(struct super_block *sb)
 {
-	trace_scoutfs_kill_sb(sb);
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 
-	if (SCOUTFS_HAS_SBI(sb))
+	if (sbi) {
+		sbi->unmounting = true;
+		smp_wmb();
+	}
+
+	if (SCOUTFS_HAS_SBI(sb)) {
+		scoutfs_inode_orphan_stop(sb);
 		scoutfs_lock_unmount_begin(sb);
+	}
 
 	kill_block_super(sb);
 }

kmod/src/super.h

@@ -89,6 +89,7 @@ struct scoutfs_sb_info {
 	struct dentry *debug_root;
 
 	bool forced_unmount;
+	bool unmounting;
 
 	unsigned long corruption_messages_once[SC_NR_LONGS];
 };
@@ -117,6 +118,19 @@ static inline bool scoutfs_forcing_unmount(struct super_block *sb)
 	return sbi->forced_unmount;
 }
 
+/*
+ * True if we're shutting down the system and can be used as a coarse
+ * indicator that we can avoid doing some work that no longer makes
+ * sense.
+ */
+static inline bool scoutfs_unmounting(struct super_block *sb)
+{
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+
+	smp_rmb();
+	return !sbi || sbi->unmounting;
+}
+
 /*
  * A small string embedded in messages that's used to identify a
  * specific mount.  It's the three most significant bytes of the fsid

kmod/src/trans.c

@@ -190,25 +190,8 @@ void scoutfs_trans_write_func(struct work_struct *work)
 		goto out;
 	}
 
-	trace_scoutfs_trans_write_func(sb,
-			scoutfs_block_writer_dirty_bytes(sb, &tri->wri));
-
-	if (!scoutfs_block_writer_has_dirty(sb, &tri->wri) &&
-	    !scoutfs_item_dirty_pages(sb)) {
-		if (sbi->trans_deadline_expired) {
-			/*
-			 * If we're not writing data then we only advance the
-			 * seq at the sync deadline interval.  This keeps idle
-			 * mounts from pinning a seq and stopping readers of the
-			 * seq indices but doesn't send a message for every sync
-			 * syscall.
-			 */
-			ret = scoutfs_client_advance_seq(sb, &trans_seq);
-			if (ret < 0)
-			      s = "clean advance seq";
-		}
-		goto err;
-	}
+	trace_scoutfs_trans_write_func(sb, scoutfs_block_writer_dirty_bytes(sb, &tri->wri),
+				       scoutfs_item_dirty_pages(sb));
 
 	if (sbi->trans_deadline_expired)
 		scoutfs_inc_counter(sb, trans_commit_timer);
@@ -219,15 +202,12 @@ void scoutfs_trans_write_func(struct work_struct *work)
 	ret = (s = "data submit", scoutfs_inode_walk_writeback(sb, true)) ?:
 	      (s = "item dirty", scoutfs_item_write_dirty(sb))  ?:
 	      (s = "data prepare", scoutfs_data_prepare_commit(sb))  ?:
-	      (s = "alloc prepare", scoutfs_alloc_prepare_commit(sb,
-						&tri->alloc, &tri->wri))  ?:
+	      (s = "alloc prepare", scoutfs_alloc_prepare_commit(sb, &tri->alloc, &tri->wri))  ?:
 	      (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)) ?:
-	      scoutfs_item_write_done(sb) ?:
-	      (s = "advance seq", scoutfs_client_advance_seq(sb, &trans_seq)) ?:
-	      (s = "get log trees", scoutfs_trans_get_log_trees(sb));
-err:
+	      (s = "commit log trees", commit_btrees(sb)) ?: scoutfs_item_write_done(sb) ?:
+	      (s = "get log trees", scoutfs_trans_get_log_trees(sb)) ?:
+	      (s = "advance seq", scoutfs_client_advance_seq(sb, &trans_seq));
 	if (ret < 0)
 		scoutfs_err(sb, "critical transaction commit failure: %s, %d",
 			    s, ret);

tests/golden/lock-conflicting-batch-commit

@@ -1,4 +0,0 @@
-== create per mount files
-== time independent modification
-== time concurrent independent modification
-== time concurrent conflicting modification

tests/sequence

@@ -25,7 +25,6 @@ basic-posix-consistency.sh
 dirent-consistency.sh
 mkdir-rename-rmdir.sh
 lock-ex-race-processes.sh
-lock-conflicting-batch-commit.sh
 cross-mount-data-free.sh
 persistent-item-vers.sh
 setup-error-teardown.sh

tests/tests/lock-conflicting-batch-commit.sh

@@ -1,59 +0,0 @@
-#
-# If bulk work accidentally conflicts in the worst way we'd like to have
-# it not result in catastrophic performance.  Make sure that each
-# instance of bulk work is given the opportunity to get as much as it
-# can into the transaction under a lock before the lock is revoked
-# and the transaction is committed.
-#
-
-t_require_commands setfattr
-t_require_mounts 2
-
-NR=3000
-
-echo "== create per mount files" 
-for m in 0 1; do 
-	eval dir="\$T_D${m}/dir/$m"
-	t_quiet mkdir -p "$dir"
-	for a in $(seq 1 $NR); do touch "$dir/$a"; done
-done
-
-echo "== time independent modification"
-for m in 0 1; do 
-	eval dir="\$T_D${m}/dir/$m"
-	START=$SECONDS
-	for a in $(seq 1 $NR); do
-		setfattr -n user.test_grace -v $a "$dir/$a"
-	done
-	echo "mount $m: $((SECONDS - START))" >> $T_TMP.log
-done
-
-echo "== time concurrent independent modification"
-START=$SECONDS
-for m in 0 1; do 
-	eval dir="\$T_D${m}/dir/$m"
-	(for a in $(seq 1 $NR); do
-		setfattr -n user.test_grace -v $a "$dir/$a"; 
-	done) &
-done
-wait
-IND="$((SECONDS - START))"
-echo "ind: $IND" >> $T_TMP.log
-
-echo "== time concurrent conflicting modification"
-START=$SECONDS
-for m in 0 1; do 
-	eval dir="\$T_D${m}/dir/0"
-	(for a in $(seq 1 $NR); do
-		setfattr -n user.test_grace -v $a "$dir/$a"; 
-	done) &
-done
-wait
-CONF="$((SECONDS - START))"
-echo "conf: $CONF" >> $T_TMP.log
-
-if [ "$CONF" -gt "$((IND * 5))" ]; then
-	t_fail "conflicting $CONF secs is more than 5x independent $IND secs"
-fi
-
-t_pass

tests/tests/srch-basic-functionality.sh

@@ -17,8 +17,10 @@ diff_srch_find()
 	local n="$1"
 
 	sync
-	scoutfs search-xattrs "$n" -p "$T_M0" > "$T_TMP.srch"
-	find_xattrs -d "$T_D0" -m "$T_M0" -n "$n" > "$T_TMP.find"
+	scoutfs search-xattrs "$n" -p "$T_M0" > "$T_TMP.srch" || \
+		t_fail "search-xattrs failed"
+	find_xattrs -d "$T_D0" -m "$T_M0" -n "$n" > "$T_TMP.find" || \
+		t_fail "find_xattrs failed"
 
 	diff -u "$T_TMP.srch" "$T_TMP.find"
 }

utils/src/mkfs.c

@@ -238,7 +238,7 @@ static int do_mkfs(struct mkfs_args *args)
 	memset(super, 0, SCOUTFS_BLOCK_SM_SIZE);
 	super->version = cpu_to_le64(SCOUTFS_INTEROP_VERSION);
 	uuid_generate(super->uuid);
-	super->next_ino = cpu_to_le64(SCOUTFS_ROOT_INO + 1);
+	super->next_ino = cpu_to_le64(round_up(SCOUTFS_ROOT_INO + 1, SCOUTFS_LOCK_INODE_GROUP_NR));
 	super->seq = cpu_to_le64(1);
 	super->total_meta_blocks = cpu_to_le64(last_meta + 1);
 	super->total_data_blocks = cpu_to_le64(last_data + 1);

utils/src/print.c

@@ -290,6 +290,7 @@ static int print_log_trees_item(struct scoutfs_key *key, void *val,
 		       "      data_freed: "ALCROOT_F"\n"
 		       "      srch_file: "SRF_FMT"\n"
 		       "      max_item_seq: %llu\n"
+		       "      finalize_seq: %llu\n"
 		       "      rid: %016llx\n"
 		       "      nr: %llu\n"
 		       "      flags: %llx\n"
@@ -306,6 +307,7 @@ static int print_log_trees_item(struct scoutfs_key *key, void *val,
 		       ALCROOT_A(&lt->data_freed),
 		       SRF_A(&lt->srch_file),
 		       le64_to_cpu(lt->max_item_seq),
+		       le64_to_cpu(lt->finalize_seq),
 		       le64_to_cpu(lt->rid),
 		       le64_to_cpu(lt->nr),
 		       le64_to_cpu(lt->flags),
@@ -397,12 +399,10 @@ static int print_log_merge_item(struct scoutfs_key *key, void *val,
 	switch (key->sk_zone) {
 	case SCOUTFS_LOG_MERGE_STATUS_ZONE:
 		stat = val;
-		printf("    status: next_range_key "SK_FMT" nr_req %llu nr_comp %llu"
-		       " last_seq %llu seq %llu\n",
+		printf("    status: next_range_key "SK_FMT" nr_req %llu nr_comp %llu seq %llu\n",
 		       SK_ARG(&stat->next_range_key),
 		       le64_to_cpu(stat->nr_requests),
 		       le64_to_cpu(stat->nr_complete),
-		       le64_to_cpu(stat->last_seq),
 		       le64_to_cpu(stat->seq));
 		break;
 	case SCOUTFS_LOG_MERGE_RANGE_ZONE:
@@ -414,12 +414,12 @@ static int print_log_merge_item(struct scoutfs_key *key, void *val,
 	case SCOUTFS_LOG_MERGE_REQUEST_ZONE:
 		req = val;
 		printf("    request: logs_root "BTROOT_F" logs_root "BTROOT_F" start "SK_FMT
-		       " end "SK_FMT" last_seq %llu rid %016llx seq %llu flags 0x%llx\n",
+		       " end "SK_FMT" input_seq %llu rid %016llx seq %llu flags 0x%llx\n",
 		       BTROOT_A(&req->logs_root),
 		       BTROOT_A(&req->root),
 		       SK_ARG(&req->start),
 		       SK_ARG(&req->end),
-		       le64_to_cpu(req->last_seq),
+		       le64_to_cpu(req->input_seq),
 		       le64_to_cpu(req->rid),
 		       le64_to_cpu(req->seq),
 		       le64_to_cpu(req->flags));