scoutfs: introduce CodingStyle.txt

Add a coding style document that tries to record the conventions that
the project uses.  It seemed more appropriate to put it up in the -kmod
git repo context rather than in src/ which would end up in fs/scoutfs/
upstream.

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

CodingStyle.txt

@@ -0,0 +1,82 @@
+
+We try to maintain a consistent coding style across the project.  It's
+admitedly arbitrary and starts with and is based on upstream's
+Documentation/CodingStyle.  Conventions are added here as they come up
+during review.  We'll demonstrate each sylistic preference with a diff
+snippet.
+
+== Try to make one exit point for reasonably long functions
+
+{
+-	void *a;
+-	void *b;
++	void *a = NULL;
++	void *b = NULL;
++	int ret;
+
+	a = kalloc();
+-	if (!a)
+-		return 1;
++	if (!a) {
++		ret = 1;
++		goto out;
++	}
+
+	b = kalloc();
+	if (!b) {
+-		kfree(a);
+-		return 2;
++		ret = 2;
++		goto out;
+	}
+
+-	return 3
++	ret = 3;
++out:
++	kfree(a);
++	kfree(b);
++	return ret;
+}
+
+The idea is to initialize all state at the top of the function,
+modifying it throughout, and clean it all up at the end.  Having one
+exit point also gives us a place to add tracing of function exit.
+
+== Multiple declarations on a line
+
+-	int i, j;
++	int i;
++	int j;
+
+Declare function variables one per line.  The verbose declarations
+create pressure to think about excessive stack use or over-long
+functions, makes initializers clear, and leaves room for comments.
+
+== Balance braces
+
+-	if (IS_ERR(super_block))
++	if (IS_ERR(super_block)) {
+		return PTR_ERR(super_block);
+-	else {
++	} else {
+		*super_res = *super_block;
+		kfree(super_block);
+		return 0;
+	}
+
+*nervous twitch*
+
+== Cute variable defintion waterfalls
+
++	struct block_device *meta_bdev;
+	struct scoutfs_sb_info *sbi;
+	struct mount_options opts;
+-	struct block_device *meta_bdev;
+	struct inode *inode;
+
+This isn't strictly necessary, but it's nice to try and make a pretty
+descending length of variable distributions.  It often has the
+accidental  effect of sorting definitions by decreasing complexity.  I
+tend to group types when the name lengths are pretty close, even if
+they're not strictly sorted, so that all the ints, u64s, keys, etc, are
+all together.

kmod/src/alloc.c

@@ -252,7 +252,7 @@ void scoutfs_alloc_init(struct scoutfs_alloc *alloc,
 {
 	memset(alloc, 0, sizeof(struct scoutfs_alloc));
 
-	spin_lock_init(&alloc->lock);
+	seqlock_init(&alloc->seqlock);
 	mutex_init(&alloc->mutex);
 	alloc->avail = *avail;
 	alloc->freed = *freed;
@@ -358,31 +358,24 @@ static void list_block_sort(struct scoutfs_alloc_list_block *lblk)
 
 /*
  * We're always reading blocks that we own, so we shouldn't see stale
- * references.  But the cached block can be stale and we can need to
- * invalidate it.
+ * references but we could retry reads after dropping stale cached
+ * blocks.  If we do see a stale error then we've hit persistent
+ * corruption.
  */
-static int read_list_block(struct super_block *sb,
-			   struct scoutfs_alloc_list_ref *ref,
+static int read_list_block(struct super_block *sb, struct scoutfs_block_ref *ref,
 			   struct scoutfs_block **bl_ret)
 {
-	struct scoutfs_block *bl = NULL;
+	int ret;
 
-	bl = scoutfs_block_read(sb, le64_to_cpu(ref->blkno));
-	if (!IS_ERR_OR_NULL(bl) &&
-	    !scoutfs_block_consistent_ref(sb, bl, ref->seq, ref->blkno,
-					  SCOUTFS_BLOCK_MAGIC_ALLOC_LIST)) {
-		scoutfs_inc_counter(sb, alloc_stale_cached_list_block);
-		scoutfs_block_invalidate(sb, bl);
-		scoutfs_block_put(sb, bl);
-		bl = scoutfs_block_read(sb, le64_to_cpu(ref->blkno));
-	}
-	if (IS_ERR(bl)) {
-		*bl_ret = NULL;
-		return PTR_ERR(bl);
-	}
+	ret = scoutfs_block_read_ref(sb, ref, SCOUTFS_BLOCK_MAGIC_ALLOC_LIST, bl_ret);
+	if (ret < 0) {
+		if (ret == -ESTALE) {
+			scoutfs_inc_counter(sb, alloc_stale_list_block);
+			ret = -EIO;
+		}
+	};
 
-	*bl_ret = bl;
-	return 0;
+	return ret;
 }
 
 /*
@@ -396,86 +389,12 @@ static int read_list_block(struct super_block *sb,
 static int dirty_list_block(struct super_block *sb,
 			    struct scoutfs_alloc *alloc,
 			    struct scoutfs_block_writer *wri,
-			    struct scoutfs_alloc_list_ref *ref,
+			    struct scoutfs_block_ref *ref,
 			    u64 dirty, u64 *old,
 			    struct scoutfs_block **bl_ret)
 {
-	struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super;
-	struct scoutfs_block *cow_bl = NULL;
-	struct scoutfs_block *bl = NULL;
-	struct scoutfs_alloc_list_block *lblk;
-	bool undo_alloc = false;
-	u64 blkno;
-	int ret;
-	int err;
-
-	blkno = le64_to_cpu(ref->blkno);
-	if (blkno) {
-		ret = read_list_block(sb, ref, &bl);
-		if (ret < 0)
-			goto out;
-
-		if (scoutfs_block_writer_is_dirty(sb, bl)) {
-			ret = 0;
-			goto out;
-		}
-	}
-
-	if (dirty == 0) {
-		ret = scoutfs_alloc_meta(sb, alloc, wri, &dirty);
-		if (ret < 0)
-			goto out;
-		undo_alloc = true;
-	}
-
-	cow_bl = scoutfs_block_create(sb, dirty);
-	if (IS_ERR(cow_bl)) {
-		ret = PTR_ERR(cow_bl);
-		goto out;
-	}
-
-	if (old) {
-		*old = blkno;
-	} else if (blkno) {
-		ret = scoutfs_free_meta(sb, alloc, wri, blkno);
-		if (ret < 0)
-			goto out;
-	}
-
-	if (bl)
-		memcpy(cow_bl->data, bl->data, SCOUTFS_BLOCK_LG_SIZE);
-	else
-		memset(cow_bl->data, 0, SCOUTFS_BLOCK_LG_SIZE);
-	scoutfs_block_put(sb, bl);
-	bl = cow_bl;
-	cow_bl = NULL;
-
-	lblk = bl->data;
-	lblk->hdr.magic = cpu_to_le32(SCOUTFS_BLOCK_MAGIC_ALLOC_LIST);
-	lblk->hdr.fsid = super->hdr.fsid;
-	lblk->hdr.blkno = cpu_to_le64(bl->blkno);
-	prandom_bytes(&lblk->hdr.seq, sizeof(lblk->hdr.seq));
-
-	ref->blkno = lblk->hdr.blkno;
-	ref->seq = lblk->hdr.seq;
-
-	scoutfs_block_writer_mark_dirty(sb, wri, bl);
-	ret = 0;
-
-out:
-	scoutfs_block_put(sb, cow_bl);
-	if (ret < 0 && undo_alloc) {
-		err = scoutfs_free_meta(sb, alloc, wri, dirty);
-		BUG_ON(err); /* inconsistent */
-	}
-
-	if (ret < 0) {
-		scoutfs_block_put(sb, bl);
-		bl = NULL;
-	}
-	*bl_ret = bl;
-
-	return ret;
+	return scoutfs_block_dirty_ref(sb, alloc, wri, ref, SCOUTFS_BLOCK_MAGIC_ALLOC_LIST,
+				       bl_ret, dirty, old);
 }
 
 /* Allocate a new dirty list block if we fill up more than 3/4 of the block. */
@@ -497,7 +416,7 @@ static int dirty_alloc_blocks(struct super_block *sb,
 			      struct scoutfs_alloc *alloc,
 			      struct scoutfs_block_writer *wri)
 {
-	struct scoutfs_alloc_list_ref orig_freed;
+	struct scoutfs_block_ref orig_freed;
 	struct scoutfs_alloc_list_block *lblk;
 	struct scoutfs_block *av_bl = NULL;
 	struct scoutfs_block *fr_bl = NULL;
@@ -607,7 +526,8 @@ int scoutfs_alloc_meta(struct super_block *sb, struct scoutfs_alloc *alloc,
 	if (ret < 0)
 		goto out;
 
-	spin_lock(&alloc->lock);
+	write_seqlock(&alloc->seqlock);
+
 	lblk = alloc->dirty_avail_bl->data;
 	if (WARN_ON_ONCE(lblk->nr == 0)) {
 		/* shouldn't happen, transaction should commit first */
@@ -617,7 +537,8 @@ int scoutfs_alloc_meta(struct super_block *sb, struct scoutfs_alloc *alloc,
 		list_block_remove(&alloc->avail, lblk, 1);
 		ret = 0;
 	}
-	spin_unlock(&alloc->lock);
+
+	write_sequnlock(&alloc->seqlock);
 
 out:
 	if (ret < 0)
@@ -640,7 +561,8 @@ int scoutfs_free_meta(struct super_block *sb, struct scoutfs_alloc *alloc,
 	if (ret < 0)
 		goto out;
 
-	spin_lock(&alloc->lock);
+	write_seqlock(&alloc->seqlock);
+
 	lblk = alloc->dirty_freed_bl->data;
 	if (WARN_ON_ONCE(list_block_space(lblk->nr) == 0)) {
 		/* shouldn't happen, transaction should commit first */
@@ -649,7 +571,8 @@ int scoutfs_free_meta(struct super_block *sb, struct scoutfs_alloc *alloc,
 		list_block_add(&alloc->freed, lblk, blkno);
 		ret = 0;
 	}
-	spin_unlock(&alloc->lock);
+
+	write_sequnlock(&alloc->seqlock);
 
 out:
 	scoutfs_inc_counter(sb, alloc_free_meta);
@@ -1106,7 +1029,7 @@ int scoutfs_alloc_splice_list(struct super_block *sb,
 			      struct scoutfs_alloc_list_head *src)
 {
 	struct scoutfs_alloc_list_block *lblk;
-	struct scoutfs_alloc_list_ref *ref;
+	struct scoutfs_block_ref *ref;
 	struct scoutfs_block *prev = NULL;
 	struct scoutfs_block *bl = NULL;
 	int ret = 0;
@@ -1147,17 +1070,23 @@ out:
 
 /*
  * Returns true if meta avail and free don't have room for the given
- * number of alloctions or frees.
+ * number of allocations or frees.  This is called at a significantly
+ * higher frequency than allocations as writers try to enter
+ * transactions.  This is the only reader of the seqlock which gives
+ * read-mostly sampling instead of bouncing a spinlock around all the
+ * cores.
  */
 bool scoutfs_alloc_meta_low(struct super_block *sb,
 			    struct scoutfs_alloc *alloc, u32 nr)
 {
+	unsigned int seq;
 	bool lo;
 
-	spin_lock(&alloc->lock);
-	lo = le32_to_cpu(alloc->avail.first_nr) < nr ||
-	     list_block_space(alloc->freed.first_nr) < nr;
-	spin_unlock(&alloc->lock);
+	do {
+		seq = read_seqbegin(&alloc->seqlock);
+		lo = le32_to_cpu(alloc->avail.first_nr) < nr ||
+		     list_block_space(alloc->freed.first_nr) < nr;
+	} while (read_seqretry(&alloc->seqlock, seq));
 
 	return lo;
 }
@@ -1169,8 +1098,8 @@ bool scoutfs_alloc_meta_low(struct super_block *sb,
 int scoutfs_alloc_foreach(struct super_block *sb,
 			  scoutfs_alloc_foreach_cb_t cb, void *arg)
 {
-	struct scoutfs_btree_ref stale_refs[2] = {{0,}};
-	struct scoutfs_btree_ref refs[2] = {{0,}};
+	struct scoutfs_block_ref stale_refs[2] = {{0,}};
+	struct scoutfs_block_ref refs[2] = {{0,}};
 	struct scoutfs_super_block *super = NULL;
 	struct scoutfs_srch_compact *sc;
 	struct scoutfs_log_trees lt;

kmod/src/alloc.h

@@ -72,7 +72,8 @@
  * transaction.
  */
 struct scoutfs_alloc {
-	spinlock_t lock;
+	/* writers rarely modify list_head avail/freed.  readers often check for _meta_alloc_low */
+	seqlock_t seqlock;
 	struct mutex mutex;
 	struct scoutfs_block *dirty_avail_bl;
 	struct scoutfs_block *dirty_freed_bl;

kmod/src/block.h

@@ -13,27 +13,16 @@ struct scoutfs_block {
 	void *priv;
 };
 
-__le32 scoutfs_block_calc_crc(struct scoutfs_block_header *hdr, u32 size);
-bool scoutfs_block_valid_crc(struct scoutfs_block_header *hdr, u32 size);
-bool scoutfs_block_valid_ref(struct super_block *sb,
-			     struct scoutfs_block_header *hdr,
-			     __le64 seq, __le64 blkno);
-
-struct scoutfs_block *scoutfs_block_create(struct super_block *sb, u64 blkno);
-struct scoutfs_block *scoutfs_block_read(struct super_block *sb, u64 blkno);
-void scoutfs_block_invalidate(struct super_block *sb, struct scoutfs_block *bl);
-bool scoutfs_block_consistent_ref(struct super_block *sb,
-				  struct scoutfs_block *bl,
-				  __le64 seq, __le64 blkno, u32 magic);
+int scoutfs_block_read_ref(struct super_block *sb, struct scoutfs_block_ref *ref, u32 magic,
+			   struct scoutfs_block **bl_ret);
 void scoutfs_block_put(struct super_block *sb, struct scoutfs_block *bl);
 
 void scoutfs_block_writer_init(struct super_block *sb,
 			       struct scoutfs_block_writer *wri);
-void scoutfs_block_writer_mark_dirty(struct super_block *sb,
-				     struct scoutfs_block_writer *wri,
-				     struct scoutfs_block *bl);
-bool scoutfs_block_writer_is_dirty(struct super_block *sb,
-				   struct scoutfs_block *bl);
+int scoutfs_block_dirty_ref(struct super_block *sb, struct scoutfs_alloc *alloc,
+			    struct scoutfs_block_writer *wri, struct scoutfs_block_ref *ref,
+			    u32 magic, struct scoutfs_block **bl_ret,
+			    u64 dirty_blkno, u64 *ref_blkno);
 int scoutfs_block_writer_write(struct super_block *sb,
 			       struct scoutfs_block_writer *wri);
 void scoutfs_block_writer_forget_all(struct super_block *sb,

kmod/src/btree.c

@@ -80,7 +80,7 @@ enum btree_walk_flags {
 	 BTW_NEXT	= (1 <<  0), /* return >= key */
 	 BTW_PREV	= (1 <<  1), /* return <= key */
 	 BTW_DIRTY	= (1 <<  2), /* cow stable blocks */
-	 BTW_ALLOC	= (1 <<  3), /* allocate a new block for 0 ref */
+	 BTW_ALLOC	= (1 <<  3), /* allocate a new block for 0 ref, requires dirty */
 	 BTW_INSERT	= (1 <<  4), /* walking to insert, try splitting */
 	 BTW_DELETE	= (1 <<  5), /* walking to delete, try joining */
 };
@@ -619,140 +619,36 @@ static void move_items(struct scoutfs_btree_block *dst,
  * This is used to lookup cached blocks, read blocks, cow blocks for
  * dirtying, and allocate new blocks.
  *
- * Btree blocks don't have rigid cache consistency.  We can be following
- * block references into cached blocks that are now stale or can be
- * following a stale root into blocks that have been overwritten.  If we
- * hit a block that looks stale we first invalidate the cache and retry,
- * returning -ESTALE if it still looks wrong.  The caller can retry the
- * read from a more current root or decide that this is a persistent
- * error.
+ * If we read a stale block we return stale so the caller can retry with
+ * a newer root or return an error.
  */
 static int get_ref_block(struct super_block *sb,
 			 struct scoutfs_alloc *alloc,
 			 struct scoutfs_block_writer *wri, int flags,
-			 struct scoutfs_btree_ref *ref,
+			 struct scoutfs_block_ref *ref,
 			 struct scoutfs_block **bl_ret)
 {
-	struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super;
-	struct scoutfs_btree_block *bt = NULL;
-	struct scoutfs_btree_block *new;
-	struct scoutfs_block *new_bl = NULL;
-	struct scoutfs_block *bl = NULL;
-	bool retried = false;
-	u64 blkno;
-	u64 seq;
 	int ret;
 
-	/* always get the current block, either to return or cow from */
-	if (ref && ref->blkno) {
-retry:
+	if (WARN_ON_ONCE((flags & BTW_ALLOC) && !(flags & BTW_DIRTY)))
+		return -EINVAL;
 
-		bl = scoutfs_block_read(sb, le64_to_cpu(ref->blkno));
-		if (IS_ERR(bl)) {
-			trace_scoutfs_btree_read_error(sb, ref);
-			scoutfs_inc_counter(sb, btree_read_error);
-			ret = PTR_ERR(bl);
-			goto out;
-		}
-		bt = (void *)bl->data;
-
-		if (!scoutfs_block_consistent_ref(sb, bl, ref->seq, ref->blkno,
-						  SCOUTFS_BLOCK_MAGIC_BTREE) ||
-		    scoutfs_trigger(sb, BTREE_STALE_READ)) {
-
-			scoutfs_inc_counter(sb, btree_stale_read);
-
-			scoutfs_block_invalidate(sb, bl);
-			scoutfs_block_put(sb, bl);
-			bl = NULL;
-
-			if (!retried) {
-				retried = true;
-				goto retry;
-			}
-
-			ret = -ESTALE;
-			goto out;
-		}
-
-		/*
-		 * We need to create a new dirty copy of the block if
-		 * the caller asked for it.  If the block is already
-		 * dirty then we can return it.
-		 */
-		if (!(flags & BTW_DIRTY) ||
-		    scoutfs_block_writer_is_dirty(sb, bl)) {
-			ret = 0;
-			goto out;
-		}
-
-	} else if (!(flags & BTW_ALLOC)) {
+	if (ref->blkno == 0 && !(flags & BTW_ALLOC)) {
 		ret = -ENOENT;
 		goto out;
 	}
 
-	ret = scoutfs_alloc_meta(sb, alloc, wri, &blkno);
-	if (ret < 0)
-		goto out;
-
-	prandom_bytes(&seq, sizeof(seq));
-
-	new_bl = scoutfs_block_create(sb, blkno);
-	if (IS_ERR(new_bl)) {
-		ret = scoutfs_free_meta(sb, alloc, wri, blkno);
-		BUG_ON(ret);
-		ret = PTR_ERR(new_bl);
-		goto out;
-	}
-	new = (void *)new_bl->data;
-
-	/* free old stable blkno we're about to overwrite */
-	if (ref && ref->blkno) {
-		ret = scoutfs_free_meta(sb, alloc, wri,
-					le64_to_cpu(ref->blkno));
-		if (ret) {
-			ret = scoutfs_free_meta(sb, alloc, wri, blkno);
-			BUG_ON(ret);
-			scoutfs_block_put(sb, new_bl);
-			new_bl = NULL;
-			goto out;
-		}
-	}
-
-	scoutfs_block_writer_mark_dirty(sb, wri, new_bl);
-
-	trace_scoutfs_btree_dirty_block(sb, blkno, seq,
-					bt ? le64_to_cpu(bt->hdr.blkno) : 0,
-					bt ? le64_to_cpu(bt->hdr.seq) : 0);
-
-	if (bt) {
-		/* returning a cow of an existing block */
-		memcpy(new, bt, SCOUTFS_BLOCK_LG_SIZE);
-		scoutfs_block_put(sb, bl);
-	} else {
-		/* returning a newly allocated block */
-		memset(new, 0, SCOUTFS_BLOCK_LG_SIZE);
-		new->hdr.fsid = super->hdr.fsid;
-	}
-	bl = new_bl;
-	bt = new;
-
-	bt->hdr.magic = cpu_to_le32(SCOUTFS_BLOCK_MAGIC_BTREE);
-	bt->hdr.blkno = cpu_to_le64(blkno);
-	bt->hdr.seq = cpu_to_le64(seq);
-	if (ref) {
-		ref->blkno = bt->hdr.blkno;
-		ref->seq = bt->hdr.seq;
-	}
-	ret = 0;
-
+	if (flags & BTW_DIRTY)
+		ret = scoutfs_block_dirty_ref(sb, alloc, wri, ref, SCOUTFS_BLOCK_MAGIC_BTREE,
+					      bl_ret, 0, NULL);
+	else
+		ret = scoutfs_block_read_ref(sb, ref, SCOUTFS_BLOCK_MAGIC_BTREE, bl_ret);
 out:
-	if (ret) {
-		scoutfs_block_put(sb, bl);
-		bl = NULL;
+	if (ret < 0) {
+		if (ret == -ESTALE)
+			scoutfs_inc_counter(sb, btree_stale_read);
 	}
 
-	*bl_ret = bl;
 	return ret;
 }
 
@@ -766,7 +662,7 @@ static void create_parent_item(struct scoutfs_btree_block *parent,
 {
 	struct scoutfs_avl_node *par;
 	int cmp;
-	struct scoutfs_btree_ref ref = {
+	struct scoutfs_block_ref ref = {
 		.blkno = child->hdr.blkno,
 		.seq = child->hdr.seq,
 	};
@@ -784,7 +680,7 @@ static void update_parent_item(struct scoutfs_btree_block *parent,
 			       struct scoutfs_btree_item *par_item,
 			       struct scoutfs_btree_block *child)
 {
-	struct scoutfs_btree_ref *ref = item_val(parent, par_item);
+	struct scoutfs_block_ref *ref = item_val(parent, par_item);
 
 	par_item->key = *item_key(last_item(child));
 	ref->blkno = child->hdr.blkno;
@@ -832,12 +728,13 @@ static int try_split(struct super_block *sb,
 	struct scoutfs_block *par_bl = NULL;
 	struct scoutfs_btree_block *left;
 	struct scoutfs_key max_key;
+	struct scoutfs_block_ref zeros;
 	int ret;
 	int err;
 
 	/* parents need to leave room for child references */
 	if (right->level)
-		val_len = sizeof(struct scoutfs_btree_ref);
+		val_len = sizeof(struct scoutfs_block_ref);
 
 	/* don't need to split if there's enough space for the item */
 	if (mid_free_item_room(right, val_len))
@@ -849,7 +746,8 @@ static int try_split(struct super_block *sb,
 	scoutfs_inc_counter(sb, btree_split);
 
 	/* alloc split neighbour first to avoid unwinding tree growth */
-	ret = get_ref_block(sb, alloc, wri, BTW_ALLOC, NULL, &left_bl);
+	memset(&zeros, 0, sizeof(zeros));
+	ret = get_ref_block(sb, alloc, wri, BTW_ALLOC | BTW_DIRTY, &zeros, &left_bl);
 	if (ret)
 		return ret;
 	left = left_bl->data;
@@ -857,7 +755,8 @@ static int try_split(struct super_block *sb,
 	init_btree_block(left, right->level);
 
 	if (!parent) {
-		ret = get_ref_block(sb, alloc, wri, BTW_ALLOC, NULL, &par_bl);
+		memset(&zeros, 0, sizeof(zeros));
+		ret = get_ref_block(sb, alloc, wri, BTW_ALLOC | BTW_DIRTY, &zeros, &par_bl);
 		if (ret) {
 			err = scoutfs_free_meta(sb, alloc, wri,
 						le64_to_cpu(left->hdr.blkno));
@@ -905,7 +804,7 @@ static int try_join(struct super_block *sb,
 	struct scoutfs_btree_item *sib_par_item;
 	struct scoutfs_btree_block *sib;
 	struct scoutfs_block *sib_bl;
-	struct scoutfs_btree_ref *ref;
+	struct scoutfs_block_ref *ref;
 	unsigned int sib_tot;
 	bool move_right;
 	int to_move;
@@ -1194,7 +1093,7 @@ static int btree_walk(struct super_block *sb,
 	struct scoutfs_btree_item *prev;
 	struct scoutfs_avl_node *next_node;
 	struct scoutfs_avl_node *node;
-	struct scoutfs_btree_ref *ref;
+	struct scoutfs_block_ref *ref;
 	unsigned int level;
 	unsigned int nr;
 	int ret;
@@ -1225,8 +1124,7 @@ restart:
 		if (!(flags & BTW_INSERT)) {
 			ret = -ENOENT;
 		} else {
-			ret = get_ref_block(sb, alloc, wri, BTW_ALLOC,
-					    &root->ref, &bl);
+			ret = get_ref_block(sb, alloc, wri, BTW_ALLOC | BTW_DIRTY, &root->ref, &bl);
 			if (ret == 0) {
 				bt = bl->data;
 				init_btree_block(bt, 0);

kmod/src/client.c

@@ -49,7 +49,6 @@ struct client_info {
 	struct delayed_work connect_dwork;
 
 	u64 server_term;
-	u64 greeting_umb;
 
 	bool sending_farewell;
 	int farewell_error;

kmod/src/counters.h

@@ -20,17 +20,21 @@
 	EXPAND_COUNTER(alloc_list_freed_hi)			\
 	EXPAND_COUNTER(alloc_move)				\
 	EXPAND_COUNTER(alloc_moved_extent)			\
-	EXPAND_COUNTER(alloc_stale_cached_list_block)		\
-	EXPAND_COUNTER(block_cache_access)			\
+	EXPAND_COUNTER(alloc_stale_list_block)			\
+	EXPAND_COUNTER(block_cache_access_update)		\
 	EXPAND_COUNTER(block_cache_alloc_failure)		\
 	EXPAND_COUNTER(block_cache_alloc_page_order)		\
 	EXPAND_COUNTER(block_cache_alloc_virt)			\
 	EXPAND_COUNTER(block_cache_end_io_error)		\
 	EXPAND_COUNTER(block_cache_forget)			\
 	EXPAND_COUNTER(block_cache_free)			\
-	EXPAND_COUNTER(block_cache_invalidate)			\
-	EXPAND_COUNTER(block_cache_lru_move)			\
+	EXPAND_COUNTER(block_cache_free_work)			\
+	EXPAND_COUNTER(block_cache_remove_stale)		\
 	EXPAND_COUNTER(block_cache_shrink)			\
+	EXPAND_COUNTER(block_cache_shrink_next)			\
+	EXPAND_COUNTER(block_cache_shrink_recent)		\
+	EXPAND_COUNTER(block_cache_shrink_remove)		\
+	EXPAND_COUNTER(block_cache_shrink_restart)		\
 	EXPAND_COUNTER(btree_compact_values)			\
 	EXPAND_COUNTER(btree_compact_values_enomem)		\
 	EXPAND_COUNTER(btree_delete)				\
@@ -42,7 +46,6 @@
 	EXPAND_COUNTER(btree_lookup)				\
 	EXPAND_COUNTER(btree_next)				\
 	EXPAND_COUNTER(btree_prev)				\
-	EXPAND_COUNTER(btree_read_error)			\
 	EXPAND_COUNTER(btree_split)				\
 	EXPAND_COUNTER(btree_stale_read)			\
 	EXPAND_COUNTER(btree_update)				\
@@ -73,6 +76,7 @@
 	EXPAND_COUNTER(ext_op_remove)				\
 	EXPAND_COUNTER(forest_bloom_fail)			\
 	EXPAND_COUNTER(forest_bloom_pass)			\
+	EXPAND_COUNTER(forest_bloom_stale)			\
 	EXPAND_COUNTER(forest_read_items)			\
 	EXPAND_COUNTER(forest_roots_next_hint)			\
 	EXPAND_COUNTER(forest_set_bloom_bits)			\
@@ -163,7 +167,6 @@
 	EXPAND_COUNTER(srch_compact_flush)			\
 	EXPAND_COUNTER(srch_compact_log_page)			\
 	EXPAND_COUNTER(srch_compact_removed_entry)		\
-	EXPAND_COUNTER(srch_inconsistent_ref)			\
 	EXPAND_COUNTER(srch_rotate_log)				\
 	EXPAND_COUNTER(srch_search_log)				\
 	EXPAND_COUNTER(srch_search_log_block)			\

kmod/src/forest.c

@@ -66,8 +66,8 @@ struct forest_info {
 	struct forest_info *name = SCOUTFS_SB(sb)->forest_info
 
 struct forest_refs {
-	struct scoutfs_btree_ref fs_ref;
-	struct scoutfs_btree_ref logs_ref;
+	struct scoutfs_block_ref fs_ref;
+	struct scoutfs_block_ref logs_ref;
 };
 
 /* initialize some refs that initially aren't equal */
@@ -96,20 +96,16 @@ static void calc_bloom_nrs(struct forest_bloom_nrs *bloom,
 	}
 }
 
-static struct scoutfs_block *read_bloom_ref(struct super_block *sb,
-					    struct scoutfs_btree_ref *ref)
+static struct scoutfs_block *read_bloom_ref(struct super_block *sb, struct scoutfs_block_ref *ref)
 {
 	struct scoutfs_block *bl;
+	int ret;
 
-	bl = scoutfs_block_read(sb, le64_to_cpu(ref->blkno));
-	if (IS_ERR(bl))
-		return bl;
-
-	if (!scoutfs_block_consistent_ref(sb, bl, ref->seq, ref->blkno,
-					  SCOUTFS_BLOCK_MAGIC_BLOOM)) {
-		scoutfs_block_invalidate(sb, bl);
-		scoutfs_block_put(sb, bl);
-		return ERR_PTR(-ESTALE);
+	ret = scoutfs_block_read_ref(sb, ref, SCOUTFS_BLOCK_MAGIC_BLOOM, &bl);
+	if (ret < 0) {
+		if (ret == -ESTALE)
+			scoutfs_inc_counter(sb, forest_bloom_stale);
+		bl = ERR_PTR(ret);
 	}
 
 	return bl;
@@ -381,18 +377,14 @@ out:
 int scoutfs_forest_set_bloom_bits(struct super_block *sb,
 				  struct scoutfs_lock *lock)
 {
-	struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super;
 	DECLARE_FOREST_INFO(sb, finf);
-	struct scoutfs_block *new_bl = NULL;
 	struct scoutfs_block *bl = NULL;
 	struct scoutfs_bloom_block *bb;
-	struct scoutfs_btree_ref *ref;
+	struct scoutfs_block_ref *ref;
 	struct forest_bloom_nrs bloom;
 	int nr_set = 0;
-	u64 blkno;
 	u64 nr;
 	int ret;
-	int err;
 	int i;
 
 	nr = le64_to_cpu(finf->our_log.nr);
@@ -410,53 +402,11 @@ int scoutfs_forest_set_bloom_bits(struct super_block *sb,
 
 	ref = &finf->our_log.bloom_ref;
 
-	if (ref->blkno) {
-		bl = read_bloom_ref(sb, ref);
-		if (IS_ERR(bl)) {
-			ret = PTR_ERR(bl);
-			goto unlock;
-		}
-		bb = bl->data;
-	}
-
-	if (!ref->blkno || !scoutfs_block_writer_is_dirty(sb, bl)) {
-
-		ret = scoutfs_alloc_meta(sb, finf->alloc, finf->wri, &blkno);
-		if (ret < 0)
-			goto unlock;
-
-		new_bl = scoutfs_block_create(sb, blkno);
-		if (IS_ERR(new_bl)) {
-			err = scoutfs_free_meta(sb, finf->alloc, finf->wri,
-						blkno);
-			BUG_ON(err); /* could have dirtied */
-			ret = PTR_ERR(new_bl);
-			goto unlock;
-		}
-
-		if (bl) {
-			err = scoutfs_free_meta(sb, finf->alloc, finf->wri,
-						le64_to_cpu(ref->blkno));
-			BUG_ON(err); /* could have dirtied */
-			memcpy(new_bl->data, bl->data, SCOUTFS_BLOCK_LG_SIZE);
-		} else {
-			memset(new_bl->data, 0, SCOUTFS_BLOCK_LG_SIZE);
-		}
-
-		scoutfs_block_writer_mark_dirty(sb, finf->wri, new_bl);
-
-		scoutfs_block_put(sb, bl);
-		bl = new_bl;
-		bb = bl->data;
-		new_bl = NULL;
-
-		bb->hdr.magic = cpu_to_le32(SCOUTFS_BLOCK_MAGIC_BLOOM);
-		bb->hdr.fsid = super->hdr.fsid;
-		bb->hdr.blkno = cpu_to_le64(blkno);
-		prandom_bytes(&bb->hdr.seq, sizeof(bb->hdr.seq));
-		ref->blkno = bb->hdr.blkno;
-		ref->seq = bb->hdr.seq;
-	}
+	ret = scoutfs_block_dirty_ref(sb, finf->alloc, finf->wri, ref, SCOUTFS_BLOCK_MAGIC_BLOOM,
+				      &bl, 0, NULL);
+	if (ret < 0)
+		goto unlock;
+	bb = bl->data;
 
 	for (i = 0; i < ARRAY_SIZE(bloom.nrs); i++) {
 		if (!test_and_set_bit_le(bloom.nrs[i], bb->bits)) {

kmod/src/format.h

@@ -86,11 +86,33 @@ struct scoutfs_timespec {
 	__u8 __pad[4];
 };
 
-/* XXX ipv6 */
-struct scoutfs_inet_addr {
-	__le32 addr;
+enum scoutfs_inet_family {
+	SCOUTFS_AF_NONE = 0,
+	SCOUTFS_AF_IPV4 = 1,
+	SCOUTFS_AF_IPV6 = 2,
+};
+
+struct scoutfs_inet_addr4 {
+	__le16 family;
 	__le16 port;
-	__u8 __pad[2];
+	__le32 addr;
+};
+
+/*
+ * Not yet supported by code.
+ */
+struct scoutfs_inet_addr6 {
+	__le16 family;
+	__le16 port;
+	__u8 addr[16];
+	__le32 flow_info;
+	__le32 scope_id;
+	__u8 __pad[4];
+};
+
+union scoutfs_inet_addr {
+	struct scoutfs_inet_addr4 v4;
+	struct scoutfs_inet_addr6 v6;
 };
 
 /*
@@ -106,6 +128,15 @@ struct scoutfs_block_header {
 	__le64 blkno;
 };
 
+/*
+ * A reference to a block.  The corresponding fields in the block_header
+ * must match after having read the block contents.
+ */
+struct scoutfs_block_ref {
+	__le64 blkno;
+	__le64 seq;
+};
+
 /*
  * scoutfs identifies all file system metadata items by a small key
  * struct.
@@ -202,17 +233,12 @@ struct scoutfs_avl_node {
  */
 #define SCOUTFS_BTREE_MAX_HEIGHT 20
 
-struct scoutfs_btree_ref {
-	__le64 blkno;
-	__le64 seq;
-};
-
 /*
  * A height of X means that the first block read will have level X-1 and
  * the leaves will have level 0.
  */
 struct scoutfs_btree_root {
-	struct scoutfs_btree_ref ref;
+	struct scoutfs_block_ref ref;
 	__u8 height;
 	__u8 __pad[7];
 };
@@ -253,18 +279,13 @@ struct scoutfs_btree_block {
 #define SCOUTFS_BTREE_LEAF_ITEM_HASH_BYTES \
 	(SCOUTFS_BTREE_LEAF_ITEM_HASH_NR * sizeof(__le16))
 
-struct scoutfs_alloc_list_ref {
-	__le64 blkno;
-	__le64 seq;
-};
-
 /*
  * first_nr tracks the nr of the first block in the list and is used for
  * allocation sizing. total_nr is the sum of the nr of all the blocks in
  * the list and is used for calculating total free block counts.
  */
 struct scoutfs_alloc_list_head {
-	struct scoutfs_alloc_list_ref ref;
+	struct scoutfs_block_ref ref;
 	__le64 total_nr;
 	__le32 first_nr;
 	__u8 __pad[4];
@@ -283,7 +304,7 @@ struct scoutfs_alloc_list_head {
  */
 struct scoutfs_alloc_list_block {
 	struct scoutfs_block_header hdr;
-	struct scoutfs_alloc_list_ref next;
+	struct scoutfs_block_ref next;
 	__le32 start;
 	__le32 nr;
 	__le64 blknos[0]; /* naturally aligned for sorting */
@@ -329,15 +350,10 @@ struct scoutfs_srch_entry {
 
 #define SCOUTFS_SRCH_ENTRY_MAX_BYTES	(2 + (sizeof(__u64) * 3))
 
-struct scoutfs_srch_ref {
-	__le64 blkno;
-	__le64 seq;
-};
-
 struct scoutfs_srch_file {
 	struct scoutfs_srch_entry first;
 	struct scoutfs_srch_entry last;
-	struct scoutfs_srch_ref ref;
+	struct scoutfs_block_ref ref;
 	__le64 blocks;
 	__le64 entries;
 	__u8 height;
@@ -346,13 +362,13 @@ struct scoutfs_srch_file {
 
 struct scoutfs_srch_parent {
 	struct scoutfs_block_header hdr;
-	struct scoutfs_srch_ref refs[0];
+	struct scoutfs_block_ref refs[0];
 };
 
 #define SCOUTFS_SRCH_PARENT_REFS				\
 	((SCOUTFS_BLOCK_LG_SIZE -				\
 	  offsetof(struct scoutfs_srch_parent, refs)) /		\
-	 sizeof(struct scoutfs_srch_ref))
+	 sizeof(struct scoutfs_block_ref))
 
 struct scoutfs_srch_block {
 	struct scoutfs_block_header hdr;
@@ -423,7 +439,7 @@ struct scoutfs_log_trees {
 	struct scoutfs_alloc_list_head meta_avail;
 	struct scoutfs_alloc_list_head meta_freed;
 	struct scoutfs_btree_root item_root;
-	struct scoutfs_btree_ref bloom_ref;
+	struct scoutfs_block_ref bloom_ref;
 	struct scoutfs_alloc_root data_avail;
 	struct scoutfs_alloc_root data_freed;
 	struct scoutfs_srch_file srch_file;
@@ -597,7 +613,7 @@ struct scoutfs_quorum_message {
 struct scoutfs_quorum_config {
 	__le64 version;
 	struct scoutfs_quorum_slot {
-		struct scoutfs_inet_addr addr;
+		union scoutfs_inet_addr addr;
 	} slots[SCOUTFS_QUORUM_MAX_SLOTS];
 };
 

kmod/src/inode.c

@@ -1621,19 +1621,28 @@ int scoutfs_orphan_inode(struct inode *inode)
 }
 
 /*
- * Track an inode that could have dirty pages.  Used to kick off writeback
- * on all dirty pages during transaction commit without tying ourselves in
- * knots trying to call through the high level vfs sync methods.
+ * Track an inode that could have dirty pages.  Used to kick off
+ * writeback on all dirty pages during transaction commit without tying
+ * ourselves in knots trying to call through the high level vfs sync
+ * methods.
+ *
+ * This is called by writers who hold the inode and transaction.  The
+ * inode's presence in the rbtree is removed by destroy_inode, prevented
+ * by the inode hold, and by committing the transaction, which is
+ * prevented by holding the transaction.  The inode can only go from
+ * empty to on the rbtree while we're here.
  */
 void scoutfs_inode_queue_writeback(struct inode *inode)
 {
 	DECLARE_INODE_SB_INFO(inode->i_sb, inf);
 	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
 
-	spin_lock(&inf->writeback_lock);
-	if (RB_EMPTY_NODE(&si->writeback_node))
-		insert_writeback_inode(inf, si);
-	spin_unlock(&inf->writeback_lock);
+	if (RB_EMPTY_NODE(&si->writeback_node)) {
+		spin_lock(&inf->writeback_lock);
+		if (RB_EMPTY_NODE(&si->writeback_node))
+			insert_writeback_inode(inf, si);
+		spin_unlock(&inf->writeback_lock);
+	}
 }
 
 /*

kmod/src/lock_server.c

@@ -586,7 +586,9 @@ static void init_lock_clients_key(struct scoutfs_key *key, u64 rid)
  * the client had already talked to the server then we must find an
  * existing record for it and should begin recovery.  If it doesn't have
  * a record then its timed out and we can't allow it to reconnect.  If
- * its connecting for the first time then we insert a new record.  If
+ * we're creating a new record for a client we can see EEXIST if the
+ * greeting is resent to a new server after the record was committed but
+ * before the response was received by the client.
  *
  * This is running in concurrent client greeting processing contexts.
  */
@@ -611,6 +613,8 @@ int scoutfs_lock_server_greeting(struct super_block *sb, u64 rid,
 		ret = scoutfs_btree_insert(sb, inf->alloc, inf->wri,
 					   &super->lock_clients,
 					   &key, NULL, 0);
+		if (ret == -EEXIST)
+			ret = 0;
 	}
 	mutex_unlock(&inf->mutex);
 

kmod/src/net.c

@@ -1546,9 +1546,8 @@ void scoutfs_net_client_greeting(struct super_block *sb,
  * response and they can disconnect cleanly.
  *
  * At this point our connection is idle except for send submissions and
- * shutdown being queued.  Once we shut down a We completely own a We
- * have exclusive access to a previous conn once its shutdown and we set
- * _freeing.
+ * shutdown being queued.  We have exclusive access to the previous conn
+ * once it's shutdown and we set _freeing.
  */
 void scoutfs_net_server_greeting(struct super_block *sb,
 				 struct scoutfs_net_connection *conn,

kmod/src/net.h

@@ -90,19 +90,13 @@ enum conn_flags {
 #define SIN_ARG(sin)	sin, be16_to_cpu((sin)->sin_port)
 
 static inline void scoutfs_addr_to_sin(struct sockaddr_in *sin,
-				       struct scoutfs_inet_addr *addr)
+				       union scoutfs_inet_addr *addr)
 {
-	sin->sin_family = AF_INET;
-	sin->sin_addr.s_addr = cpu_to_be32(le32_to_cpu(addr->addr));
-	sin->sin_port = cpu_to_be16(le16_to_cpu(addr->port));
-}
+	BUG_ON(addr->v4.family != cpu_to_le16(SCOUTFS_AF_IPV4));
 
-static inline void scoutfs_addr_from_sin(struct scoutfs_inet_addr *addr,
-					 struct sockaddr_in *sin)
-{
-	addr->addr = be32_to_le32(sin->sin_addr.s_addr);
-	addr->port = be16_to_le16(sin->sin_port);
-	memset(addr->__pad, 0, sizeof(addr->__pad));
+	sin->sin_family = AF_INET;
+	sin->sin_addr.s_addr = cpu_to_be32(le32_to_cpu(addr->v4.addr));
+	sin->sin_port = cpu_to_be16(le16_to_cpu(addr->v4.port));
 }
 
 struct scoutfs_net_connection *

kmod/src/quorum.c

@@ -138,7 +138,7 @@ static bool quorum_slot_present(struct scoutfs_super_block *super, int i)
 {
 	BUG_ON(i < 0 || i > SCOUTFS_QUORUM_MAX_SLOTS);
 
-	return super->qconf.slots[i].addr.addr != 0;
+	return super->qconf.slots[i].addr.v4.family == cpu_to_le16(SCOUTFS_AF_IPV4);
 }
 
 static ktime_t election_timeout(void)
@@ -976,6 +976,9 @@ static int verify_quorum_slots(struct super_block *sb)
 		}
 
 		for (j = i + 1; j < SCOUTFS_QUORUM_MAX_SLOTS; j++) {
+			if (!quorum_slot_present(super, j))
+				continue;
+
 			scoutfs_quorum_slot_sin(super, j, &other);
 
 			if (sin.sin_addr.s_addr == other.sin_addr.s_addr &&

kmod/src/scoutfs_trace.h

@@ -423,61 +423,34 @@ TRACE_EVENT(scoutfs_trans_write_func,
 	TP_printk(SCSBF" dirty %lu", SCSB_TRACE_ARGS, __entry->dirty)
 );
 
-TRACE_EVENT(scoutfs_release_trans,
-	TP_PROTO(struct super_block *sb, void *rsv, unsigned int rsv_holders,
-		 unsigned int tri_holders,
-		 unsigned int tri_writing),
+DECLARE_EVENT_CLASS(scoutfs_trans_hold_release_class,
+	TP_PROTO(struct super_block *sb, void *journal_info, int holders),
 
-	TP_ARGS(sb, rsv, rsv_holders, tri_holders, tri_writing),
+	TP_ARGS(sb, journal_info, holders),
 
 	TP_STRUCT__entry(
 		SCSB_TRACE_FIELDS
-		__field(void *, rsv)
-		__field(unsigned int, rsv_holders)
-		__field(unsigned int, tri_holders)
-		__field(unsigned int, tri_writing)
+		__field(unsigned long, journal_info)
+		__field(int, holders)
 	),
 
 	TP_fast_assign(
 		SCSB_TRACE_ASSIGN(sb);
-		__entry->rsv = rsv;
-		__entry->rsv_holders = rsv_holders;
-		__entry->tri_holders = tri_holders;
-		__entry->tri_writing = tri_writing;
+		__entry->journal_info = (unsigned long)journal_info;
+		__entry->holders = holders;
 	),
 
-	TP_printk(SCSBF" rsv %p holders %u trans holders %u writing %u",
-		  SCSB_TRACE_ARGS, __entry->rsv, __entry->rsv_holders,
-		  __entry->tri_holders, __entry->tri_writing)
+	TP_printk(SCSBF" journal_info 0x%0lx holders %d",
+		  SCSB_TRACE_ARGS, __entry->journal_info, __entry->holders)
 );
 
-TRACE_EVENT(scoutfs_trans_acquired_hold,
-	TP_PROTO(struct super_block *sb,
-		 void *rsv, unsigned int rsv_holders,
-		 unsigned int tri_holders,
-		 unsigned int tri_writing),
-
-	TP_ARGS(sb, rsv, rsv_holders, tri_holders, tri_writing),
-
-	TP_STRUCT__entry(
-		SCSB_TRACE_FIELDS
-		__field(void *, rsv)
-		__field(unsigned int, rsv_holders)
-		__field(unsigned int, tri_holders)
-		__field(unsigned int, tri_writing)
-	),
-
-	TP_fast_assign(
-		SCSB_TRACE_ASSIGN(sb);
-		__entry->rsv = rsv;
-		__entry->rsv_holders = rsv_holders;
-		__entry->tri_holders = tri_holders;
-		__entry->tri_writing = tri_writing;
-	),
-
-	TP_printk(SCSBF" rsv %p holders %u trans holders %u writing %u",
-		  SCSB_TRACE_ARGS, __entry->rsv, __entry->rsv_holders,
-		  __entry->tri_holders, __entry->tri_writing)
+DEFINE_EVENT(scoutfs_trans_hold_release_class, scoutfs_trans_acquired_hold,
+	TP_PROTO(struct super_block *sb, void *journal_info, int holders),
+	TP_ARGS(sb, journal_info, holders)
+);
+DEFINE_EVENT(scoutfs_trans_hold_release_class, scoutfs_release_trans,
+	TP_PROTO(struct super_block *sb, void *journal_info, int holders),
+	TP_ARGS(sb, journal_info, holders)
 );
 
 TRACE_EVENT(scoutfs_ioc_release,
@@ -1611,7 +1584,7 @@ TRACE_EVENT(scoutfs_get_name,
 );
 
 TRACE_EVENT(scoutfs_btree_read_error,
-	TP_PROTO(struct super_block *sb, struct scoutfs_btree_ref *ref),
+	TP_PROTO(struct super_block *sb, struct scoutfs_block_ref *ref),
 
 	TP_ARGS(sb, ref),
 
@@ -1631,37 +1604,10 @@ TRACE_EVENT(scoutfs_btree_read_error,
 		  SCSB_TRACE_ARGS, __entry->blkno, __entry->seq)
 );
 
-TRACE_EVENT(scoutfs_btree_dirty_block,
-	TP_PROTO(struct super_block *sb, u64 blkno, u64 seq,
-		 u64 bt_blkno, u64 bt_seq),
-
-	TP_ARGS(sb, blkno, seq, bt_blkno, bt_seq),
-
-	TP_STRUCT__entry(
-		SCSB_TRACE_FIELDS
-		__field(__u64, blkno)
-		__field(__u64, seq)
-		__field(__u64, bt_blkno)
-		__field(__u64, bt_seq)
-	),
-
-	TP_fast_assign(
-		SCSB_TRACE_ASSIGN(sb);
-		__entry->blkno = blkno;
-		__entry->seq = seq;
-		__entry->bt_blkno = bt_blkno;
-		__entry->bt_seq = bt_seq;
-	),
-
-	TP_printk(SCSBF" blkno %llu seq %llu bt_blkno %llu bt_seq %llu",
-		  SCSB_TRACE_ARGS, __entry->blkno, __entry->seq,
-		  __entry->bt_blkno, __entry->bt_seq)
-);
-
 TRACE_EVENT(scoutfs_btree_walk,
 	TP_PROTO(struct super_block *sb, struct scoutfs_btree_root *root,
 		 struct scoutfs_key *key, int flags, int level,
-		 struct scoutfs_btree_ref *ref),
+		 struct scoutfs_block_ref *ref),
 
 	TP_ARGS(sb, root, key, flags, level, ref),
 
@@ -1989,8 +1935,8 @@ DEFINE_EVENT(scoutfs_forest_bloom_class, scoutfs_forest_bloom_search,
 );
 
 TRACE_EVENT(scoutfs_forest_prepare_commit,
-	TP_PROTO(struct super_block *sb, struct scoutfs_btree_ref *item_ref,
-		 struct scoutfs_btree_ref *bloom_ref),
+	TP_PROTO(struct super_block *sb, struct scoutfs_block_ref *item_ref,
+		 struct scoutfs_block_ref *bloom_ref),
 	TP_ARGS(sb, item_ref, bloom_ref),
 	TP_STRUCT__entry(
 		SCSB_TRACE_FIELDS
@@ -2056,18 +2002,45 @@ TRACE_EVENT(scoutfs_forest_init_our_log,
 		  __entry->blkno, __entry->seq)
 );
 
+TRACE_EVENT(scoutfs_block_dirty_ref,
+	TP_PROTO(struct super_block *sb, u64 ref_blkno, u64 ref_seq,
+		 u64 block_blkno, u64 block_seq),
+
+	TP_ARGS(sb, ref_blkno, ref_seq, block_blkno, block_seq),
+
+	TP_STRUCT__entry(
+		SCSB_TRACE_FIELDS
+		__field(__u64, ref_blkno)
+		__field(__u64, ref_seq)
+		__field(__u64, block_blkno)
+		__field(__u64, block_seq)
+	),
+
+	TP_fast_assign(
+		SCSB_TRACE_ASSIGN(sb);
+		__entry->ref_blkno = ref_blkno;
+		__entry->ref_seq = ref_seq;
+		__entry->block_blkno = block_blkno;
+		__entry->block_seq = block_seq;
+	),
+
+	TP_printk(SCSBF" ref_blkno %llu ref_seq %llu block_blkno %llu block_seq %llu",
+		  SCSB_TRACE_ARGS, __entry->ref_blkno, __entry->ref_seq,
+		  __entry->block_blkno, __entry->block_seq)
+);
+
 DECLARE_EVENT_CLASS(scoutfs_block_class,
-	TP_PROTO(struct super_block *sb, void *bp, u64 blkno,
-		 int refcount, int io_count, unsigned long bits, u64 lru_moved),
-	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, lru_moved),
+	TP_PROTO(struct super_block *sb, void *bp, u64 blkno, int refcount, int io_count,
+		 unsigned long bits, __u64 accessed),
+	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, accessed),
 	TP_STRUCT__entry(
 		SCSB_TRACE_FIELDS
 		__field(void *, bp)
 		__field(__u64, blkno)
 		__field(int, refcount)
 		__field(int, io_count)
-		__field(unsigned long, bits)
-		__field(__u64, lru_moved)
+		__field(long, bits)
+		__field(__u64, accessed)
 	),
 	TP_fast_assign(
 		SCSB_TRACE_ASSIGN(sb);
@@ -2076,57 +2049,71 @@ DECLARE_EVENT_CLASS(scoutfs_block_class,
 		__entry->refcount = refcount;
 		__entry->io_count = io_count;
 		__entry->bits = bits;
-		__entry->lru_moved = lru_moved;
+		__entry->accessed = accessed;
 	),
-	TP_printk(SCSBF" bp %p blkno %llu refcount %d io_count %d bits 0x%lx lru_moved %llu",
-		  SCSB_TRACE_ARGS, __entry->bp, __entry->blkno,
-		  __entry->refcount, __entry->io_count, __entry->bits,
-		  __entry->lru_moved)
+	TP_printk(SCSBF" bp %p blkno %llu refcount %d io_count %d bits 0x%lx accessed %llu",
+		  SCSB_TRACE_ARGS, __entry->bp, __entry->blkno, __entry->refcount,
+		  __entry->io_count, __entry->bits, __entry->accessed)
 );
 DEFINE_EVENT(scoutfs_block_class, scoutfs_block_allocate,
 	TP_PROTO(struct super_block *sb, void *bp, u64 blkno,
-		 int refcount, int io_count, unsigned long bits, u64 lru_moved),
-	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, lru_moved)
+		 int refcount, int io_count, unsigned long bits,
+		 __u64 accessed),
+	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, accessed)
 );
 DEFINE_EVENT(scoutfs_block_class, scoutfs_block_free,
 	TP_PROTO(struct super_block *sb, void *bp, u64 blkno,
-		 int refcount, int io_count, unsigned long bits, u64 lru_moved),
-	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, lru_moved)
+		 int refcount, int io_count, unsigned long bits,
+		 __u64 accessed),
+	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, accessed)
 );
 DEFINE_EVENT(scoutfs_block_class, scoutfs_block_insert,
 	TP_PROTO(struct super_block *sb, void *bp, u64 blkno,
-		 int refcount, int io_count, unsigned long bits, u64 lru_moved),
-	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, lru_moved)
+		 int refcount, int io_count, unsigned long bits,
+		 __u64 accessed),
+	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, accessed)
+);
+DEFINE_EVENT(scoutfs_block_class, scoutfs_block_remove,
+	TP_PROTO(struct super_block *sb, void *bp, u64 blkno,
+		 int refcount, int io_count, unsigned long bits,
+		 __u64 accessed),
+	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, accessed)
 );
 DEFINE_EVENT(scoutfs_block_class, scoutfs_block_end_io,
 	TP_PROTO(struct super_block *sb, void *bp, u64 blkno,
-		 int refcount, int io_count, unsigned long bits, u64 lru_moved),
-	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, lru_moved)
+		 int refcount, int io_count, unsigned long bits,
+		 __u64 accessed),
+	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, accessed)
 );
 DEFINE_EVENT(scoutfs_block_class, scoutfs_block_submit,
 	TP_PROTO(struct super_block *sb, void *bp, u64 blkno,
-		 int refcount, int io_count, unsigned long bits, u64 lru_moved),
-	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, lru_moved)
+		 int refcount, int io_count, unsigned long bits,
+		 __u64 accessed),
+	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, accessed)
 );
 DEFINE_EVENT(scoutfs_block_class, scoutfs_block_invalidate,
 	TP_PROTO(struct super_block *sb, void *bp, u64 blkno,
-		 int refcount, int io_count, unsigned long bits, u64 lru_moved),
-	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, lru_moved)
+		 int refcount, int io_count, unsigned long bits,
+		 __u64 accessed),
+	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, accessed)
 );
 DEFINE_EVENT(scoutfs_block_class, scoutfs_block_mark_dirty,
 	TP_PROTO(struct super_block *sb, void *bp, u64 blkno,
-		 int refcount, int io_count, unsigned long bits, u64 lru_moved),
-	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, lru_moved)
+		 int refcount, int io_count, unsigned long bits,
+		 __u64 accessed),
+	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, accessed)
 );
 DEFINE_EVENT(scoutfs_block_class, scoutfs_block_forget,
 	TP_PROTO(struct super_block *sb, void *bp, u64 blkno,
-		 int refcount, int io_count, unsigned long bits, u64 lru_moved),
-	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, lru_moved)
+		 int refcount, int io_count, unsigned long bits,
+		 __u64 accessed),
+	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, accessed)
 );
 DEFINE_EVENT(scoutfs_block_class, scoutfs_block_shrink,
 	TP_PROTO(struct super_block *sb, void *bp, u64 blkno,
-		 int refcount, int io_count, unsigned long bits, u64 lru_moved),
-	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, lru_moved)
+		 int refcount, int io_count, unsigned long bits,
+		 __u64 accessed),
+	TP_ARGS(sb, bp, blkno, refcount, io_count, bits, accessed)
 );
 
 DECLARE_EVENT_CLASS(scoutfs_ext_next_class,

kmod/src/server.c

@@ -1024,6 +1024,12 @@ static void init_mounted_client_key(struct scoutfs_key *key, u64 rid)
 	};
 }
 
+/*
+ * Insert a new mounted client item for a client that is sending us a
+ * greeting that hasn't yet seen a response.  The greeting can be
+ * retransmitted to a new server after the previous inserted the item so
+ * it's acceptable to see -EEXIST.
+ */
 static int insert_mounted_client(struct super_block *sb, u64 rid,
 				 u64 gr_flags)
 {
@@ -1042,6 +1048,8 @@ static int insert_mounted_client(struct super_block *sb, u64 rid,
 	ret = scoutfs_btree_insert(sb, &server->alloc, &server->wri,
 				   &super->mounted_clients, &key, &mcv,
 				   sizeof(mcv));
+	if (ret == -EEXIST)
+		ret = 0;
 	mutex_unlock(&server->mounted_clients_mutex);
 
 	return ret;

kmod/src/srch.c

@@ -255,24 +255,9 @@ static u8 height_for_blk(u64 blk)
 	return hei;
 }
 
-static void init_file_block(struct super_block *sb, struct scoutfs_block *bl,
-			    int level)
+static inline u32 srch_level_magic(int level)
 {
-	struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super;
-	struct scoutfs_block_header *hdr;
-
-	/* don't leak uninit kernel mem.. block should do this for us? */
-	memset(bl->data, 0, SCOUTFS_BLOCK_LG_SIZE);
-
-	hdr = bl->data;
-	hdr->fsid = super->hdr.fsid;
-	hdr->blkno = cpu_to_le64(bl->blkno);
-	prandom_bytes(&hdr->seq, sizeof(hdr->seq));
-
-	if (level)
-		hdr->magic = cpu_to_le32(SCOUTFS_BLOCK_MAGIC_SRCH_PARENT);
-	else
-		hdr->magic = cpu_to_le32(SCOUTFS_BLOCK_MAGIC_SRCH_BLOCK);
+	return level ? SCOUTFS_BLOCK_MAGIC_SRCH_PARENT : SCOUTFS_BLOCK_MAGIC_SRCH_BLOCK;
 }
 
 /*
@@ -284,39 +269,15 @@ static void init_file_block(struct super_block *sb, struct scoutfs_block *bl,
  */
 static int read_srch_block(struct super_block *sb,
 			   struct scoutfs_block_writer *wri, int level,
-			   struct scoutfs_srch_ref *ref,
+			   struct scoutfs_block_ref *ref,
 			   struct scoutfs_block **bl_ret)
 {
-	struct scoutfs_block *bl;
-	int retries = 0;
-	int ret = 0;
-	int mag;
+	u32 magic = srch_level_magic(level);
+	int ret;
 
-	mag = level ? SCOUTFS_BLOCK_MAGIC_SRCH_PARENT :
-		      SCOUTFS_BLOCK_MAGIC_SRCH_BLOCK;
-retry:
-	bl = scoutfs_block_read(sb, le64_to_cpu(ref->blkno));
-	if (!IS_ERR_OR_NULL(bl) &&
-	    !scoutfs_block_consistent_ref(sb, bl, ref->seq, ref->blkno, mag)) {
-
-		scoutfs_inc_counter(sb, srch_inconsistent_ref);
-		scoutfs_block_writer_forget(sb, wri, bl);
-		scoutfs_block_invalidate(sb, bl);
-		scoutfs_block_put(sb, bl);
-		bl = NULL;
-
-		if (retries++ == 0)
-			goto retry;
-
-		bl = ERR_PTR(-ESTALE);
+	ret = scoutfs_block_read_ref(sb, ref, magic, bl_ret);
+	if (ret == -ESTALE)
 		scoutfs_inc_counter(sb, srch_read_stale);
-	}
-	if (IS_ERR(bl)) {
-		ret = PTR_ERR(bl);
-		bl = NULL;
-	}
-
-	*bl_ret = bl;
 	return ret;
 }
 
@@ -333,7 +294,7 @@ static int read_path_block(struct super_block *sb,
 {
 	struct scoutfs_block *bl = NULL;
 	struct scoutfs_srch_parent *srp;
-	struct scoutfs_srch_ref ref;
+	struct scoutfs_block_ref ref;
 	int level;
 	int ind;
 	int ret;
@@ -392,12 +353,10 @@ static int get_file_block(struct super_block *sb,
 	struct scoutfs_block_header *hdr;
 	struct scoutfs_block *bl = NULL;
 	struct scoutfs_srch_parent *srp;
-	struct scoutfs_block *new_bl;
-	struct scoutfs_srch_ref *ref;
-	u64 blkno = 0;
+	struct scoutfs_block_ref new_root_ref;
+	struct scoutfs_block_ref *ref;
 	int level;
 	int ind;
-	int err;
 	int ret;
 	u8 hei;
 
@@ -409,29 +368,21 @@ static int get_file_block(struct super_block *sb,
 			goto out;
 		}
 
-		ret = scoutfs_alloc_meta(sb, alloc, wri, &blkno);
+		memset(&new_root_ref, 0, sizeof(new_root_ref));
+		level = sfl->height;
+
+		ret = scoutfs_block_dirty_ref(sb, alloc, wri, &new_root_ref,
+					      srch_level_magic(level), &bl, 0, NULL);
 		if (ret < 0)
 			goto out;
 
-		bl = scoutfs_block_create(sb, blkno);
-		if (IS_ERR(bl)) {
-			ret = PTR_ERR(bl);
-			goto out;
-		}
-		blkno = 0;
-
-		scoutfs_block_writer_mark_dirty(sb, wri, bl);
-
-		init_file_block(sb, bl, sfl->height);
-		if (sfl->height) {
+		if (level) {
 			srp = bl->data;
-			srp->refs[0].blkno = sfl->ref.blkno;
-			srp->refs[0].seq = sfl->ref.seq;
+			srp->refs[0] = sfl->ref;
 		}
 
 		hdr = bl->data;
-		sfl->ref.blkno = hdr->blkno;
-		sfl->ref.seq = hdr->seq;
+		sfl->ref = new_root_ref;
 		sfl->height++;
 		scoutfs_block_put(sb, bl);
 		bl = NULL;
@@ -447,54 +398,13 @@ static int get_file_block(struct super_block *sb,
 			goto out;
 		}
 
-		/* read an existing block */
-		if (ref->blkno) {
-			ret = read_srch_block(sb, wri, level, ref, &bl);
-			if (ret < 0)
-				goto out;
-		}
-
-		/* allocate a new block if we need it */
-		if (!ref->blkno || ((flags & GFB_DIRTY) &&
-				    !scoutfs_block_writer_is_dirty(sb, bl))) {
-			ret = scoutfs_alloc_meta(sb, alloc, wri, &blkno);
-			if (ret < 0)
-				goto out;
-
-			new_bl = scoutfs_block_create(sb, blkno);
-			if (IS_ERR(new_bl)) {
-				ret = PTR_ERR(new_bl);
-				goto out;
-			}
-
-			if (bl) {
-				/* cow old block if we have one */
-				ret = scoutfs_free_meta(sb, alloc, wri,
-							bl->blkno);
-				if (ret)
-					goto out;
-
-				memcpy(new_bl->data, bl->data,
-				       SCOUTFS_BLOCK_LG_SIZE);
-				scoutfs_block_put(sb, bl);
-				bl = new_bl;
-				hdr = bl->data;
-				hdr->blkno = cpu_to_le64(bl->blkno);
-				prandom_bytes(&hdr->seq, sizeof(hdr->seq));
-			} else {
-				/* init new allocated block */
-				bl = new_bl;
-				init_file_block(sb, bl, level);
-			}
-
-			blkno = 0;
-			scoutfs_block_writer_mark_dirty(sb, wri, bl);
-
-			/* update file or parent block ref */
-			hdr = bl->data;
-			ref->blkno = hdr->blkno;
-			ref->seq = hdr->seq;
-		}
+		if (flags & GFB_DIRTY)
+			ret = scoutfs_block_dirty_ref(sb, alloc, wri, ref, srch_level_magic(level),
+						      &bl, 0, NULL);
+		else
+			ret = scoutfs_block_read_ref(sb, ref, srch_level_magic(level), &bl);
+		if (ret < 0)
+			goto out;
 
 		if (level == 0) {
 			ret = 0;
@@ -514,12 +424,6 @@ static int get_file_block(struct super_block *sb,
 out:
 	scoutfs_block_put(sb, parent);
 
-	/* return allocated blkno on error */
-	if (blkno > 0) {
-		err = scoutfs_free_meta(sb, alloc, wri, blkno);
-		BUG_ON(err); /* radix should have been dirty */
-	}
-
 	if (ret < 0) {
 		scoutfs_block_put(sb, bl);
 		bl = NULL;

kmod/src/trans.c

@@ -39,17 +39,15 @@
  * track the relationships between dirty blocks so there's only ever one
  * transaction being built.
  *
- * The copy of the on-disk super block in the fs sb info has its header
- * sequence advanced so that new dirty blocks inherit this dirty
- * sequence number.  It's only advanced once all those dirty blocks are
- * reachable after having first written them all out and then the new
- * super with that seq.  It's first incremented at mount.
+ * Committing the current dirty transaction can be triggered by sync, a
+ * regular background commit interval, reaching a dirty block threshold,
+ * or the transaction running out of its private allocator resources.
+ * Once all the current holders release the writing func writes out the
+ * dirty blocks while excluding holders until it finishes.
  *
- * Unfortunately writers can nest.  We don't bother trying to special
- * case holding a transaction that you're already holding because that
- * requires per-task storage.  We just let anyone hold transactions
- * regardless of waiters waiting to write, which risks waiters waiting a
- * very long time.
+ * Unfortunately writing holders can nest.  We track nested hold callers
+ * with the per-task journal_info pointer to avoid deadlocks between
+ * holders that might otherwise wait for a pending commit.
  */
 
 /* sync dirty data at least this often */
@@ -59,9 +57,7 @@
  * XXX move the rest of the super trans_ fields here.
  */
 struct trans_info {
-	spinlock_t lock;
-	unsigned holders;
-	bool writing;
+	atomic_t holders;
 
 	struct scoutfs_log_trees lt;
 	struct scoutfs_alloc alloc;
@@ -71,17 +67,9 @@ struct trans_info {
 #define DECLARE_TRANS_INFO(sb, name) \
 	struct trans_info *name = SCOUTFS_SB(sb)->trans_info
 
-static bool drained_holders(struct trans_info *tri)
-{
-	bool drained;
-
-	spin_lock(&tri->lock);
-	tri->writing = true;
-	drained = tri->holders == 0;
-	spin_unlock(&tri->lock);
-
-	return drained;
-}
+/* avoid the high sign bit out of an abundance of caution*/
+#define TRANS_HOLDERS_WRITE_FUNC_BIT	(1 << 30)
+#define TRANS_HOLDERS_COUNT_MASK	(TRANS_HOLDERS_WRITE_FUNC_BIT - 1)
 
 static int commit_btrees(struct super_block *sb)
 {
@@ -126,6 +114,36 @@ bool scoutfs_trans_has_dirty(struct super_block *sb)
 	return scoutfs_block_writer_has_dirty(sb, &tri->wri);
 }
 
+/*
+ * This is racing with wait_event conditions, make sure our atomic
+ * stores and waitqueue loads are ordered.
+ */
+static void sub_holders_and_wake(struct super_block *sb, int val)
+{
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	DECLARE_TRANS_INFO(sb, tri);
+
+	atomic_sub(val, &tri->holders);
+	smp_mb(); /* make sure sub is visible before we wake */
+	if (waitqueue_active(&sbi->trans_hold_wq))
+		wake_up(&sbi->trans_hold_wq);
+}
+
+/*
+ * called as a wait_event condition, needs to be careful to not change
+ * task state and is racing with waking paths that sub_return, test, and
+ * wake.
+ */
+static bool drained_holders(struct trans_info *tri)
+{
+	int holders;
+
+	smp_mb(); /* make sure task in wait_event queue before atomic read */
+	holders = atomic_read(&tri->holders) & TRANS_HOLDERS_COUNT_MASK;
+
+	return holders == 0;
+}
+
 /*
  * This work func is responsible for writing out all the dirty blocks
  * that make up the current dirty transaction.  It prevents writers from
@@ -162,6 +180,9 @@ void scoutfs_trans_write_func(struct work_struct *work)
 
 	sbi->trans_task = current;
 
+	/* mark that we're writing so holders wait for us to finish and clear our bit */
+	atomic_add(TRANS_HOLDERS_WRITE_FUNC_BIT, &tri->holders);
+
 	wait_event(sbi->trans_hold_wq, drained_holders(tri));
 
 	trace_scoutfs_trans_write_func(sb,
@@ -213,11 +234,8 @@ out:
 	spin_unlock(&sbi->trans_write_lock);
 	wake_up(&sbi->trans_write_wq);
 
-	spin_lock(&tri->lock);
-	tri->writing = false;
-	spin_unlock(&tri->lock);
-
-	wake_up(&sbi->trans_hold_wq);
+	/* we're done, wake waiting holders */
+	sub_holders_and_wake(sb, TRANS_HOLDERS_WRITE_FUNC_BIT);
 
 	sbi->trans_task = NULL;
 
@@ -309,53 +327,83 @@ void scoutfs_trans_restart_sync_deadline(struct super_block *sb)
 }
 
 /*
- * Each thread reserves space in the segment for their dirty items while
- * they hold the transaction.  This is calculated before the first
- * transaction hold is acquired.  It includes all the potential nested
- * item manipulation that could happen with the transaction held.
- * Including nested holds avoids having to deal with writing out partial
- * transactions while a caller still holds the transaction.
+ * We store nested holders in the lower bits of journal_info.  We use
+ * some higher bits as a magic value to detect if something goes
+ * horribly wrong and it gets clobbered.
  */
+#define TRANS_JI_MAGIC		0xd5700000
+#define TRANS_JI_MAGIC_MASK	0xfff00000
+#define TRANS_JI_COUNT_MASK	0x000fffff
 
-#define SCOUTFS_RESERVATION_MAGIC 0xd57cd13b
-struct scoutfs_reservation {
-	unsigned magic;
-	unsigned holders;
-};
+/* returns true if a caller already had a holder counted in journal_info */
+static bool inc_journal_info_holders(void)
+{
+	unsigned long holders = (unsigned long)current->journal_info;
+
+	WARN_ON_ONCE(holders != 0 && ((holders & TRANS_JI_MAGIC_MASK) != TRANS_JI_MAGIC));
+
+	if (holders == 0)
+		holders = TRANS_JI_MAGIC;
+	holders++;
+
+	current->journal_info = (void *)holders;
+	return (holders > (TRANS_JI_MAGIC | 1));
+}
+
+static void dec_journal_info_holders(void)
+{
+	unsigned long holders = (unsigned long)current->journal_info;
+
+	WARN_ON_ONCE(holders != 0 && ((holders & TRANS_JI_MAGIC_MASK) != TRANS_JI_MAGIC));
+	WARN_ON_ONCE((holders & TRANS_JI_COUNT_MASK) == 0);
+
+	holders--;
+	if (holders == TRANS_JI_MAGIC)
+		holders = 0;
+
+	current->journal_info = (void *)holders;
+}
 
 /*
- * Try to hold the transaction.  If a caller already holds the trans then
- * we piggy back on their hold.  We wait if the writer is trying to
- * write out the transation.  And if our items won't fit then we kick off
- * a write.
+ * This is called as the wait_event condition for holding a transaction.
+ * Increment the holder count unless the writer is present.  We return
+ * false to wait until the writer finishes and wakes us.
  *
- * This is called as a condition for wait_event.  It is very limited in
- * the locking (blocking) it can do because the caller has set the task
- * state before testing the condition safely race with waking after
- * setting the condition.  Our checking the amount of dirty metadata
- * blocks and free data blocks is racy, but we don't mind the risk of
- * delaying or prematurely forcing commits.
+ * This can be racing with itself while there's no waiters.  We retry
+ * the cmpxchg instead of returning and waiting.
  */
-static bool acquired_hold(struct super_block *sb,
-			  struct scoutfs_reservation *rsv)
+static bool inc_holders_unless_writer(struct trans_info *tri)
 {
-	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
-	DECLARE_TRANS_INFO(sb, tri);
-	bool acquired = false;
+	int holders;
 
-	spin_lock(&tri->lock);
+	do {
+		smp_mb(); /* make sure we read after wait puts task in queue */
+		holders = atomic_read(&tri->holders);
+		if (holders & TRANS_HOLDERS_WRITE_FUNC_BIT)
+			return false;
 
-	trace_scoutfs_trans_acquired_hold(sb, rsv, rsv->holders,
-					  tri->holders, tri->writing);
+	} while (atomic_cmpxchg(&tri->holders, holders, holders + 1) != holders);
 
-	/* use a caller's existing reservation */
-	if (rsv->holders)
-		goto hold;
+	return true;
+}
 
-	/* wait until the writing thread is finished */
-	if (tri->writing)
-		goto out;
+/*
+ * As we drop the last trans holder we try to wake a writing thread that
+ * was waiting for us to finish.
+ */
+static void release_holders(struct super_block *sb)
+{
+	dec_journal_info_holders();
+	sub_holders_and_wake(sb, 1);
+}
 
+/*
+ * The caller has incremented holders so it is blocking commits.  We
+ * make some quick checks to see if we need to trigger and wait for
+ * another commit before proceeding.
+ */
+static bool commit_before_hold(struct super_block *sb, struct trans_info *tri)
+{
 	/*
 	 * In theory each dirty item page could be straddling two full
 	 * blocks, requiring 4 allocations for each item cache page.
@@ -365,11 +413,9 @@ static bool acquired_hold(struct super_block *sb,
 	 * that it accounts for having to dirty parent blocks and
 	 * whatever dirtying is done during the transaction hold.
 	 */
-	if (scoutfs_alloc_meta_low(sb, &tri->alloc,
-				   scoutfs_item_dirty_pages(sb) * 2)) {
+	if (scoutfs_alloc_meta_low(sb, &tri->alloc, scoutfs_item_dirty_pages(sb) * 2)) {
 		scoutfs_inc_counter(sb, trans_commit_dirty_meta_full);
-		queue_trans_work(sbi);
-		goto out;
+		return true;
 	}
 
 	/*
@@ -381,57 +427,74 @@ static bool acquired_hold(struct super_block *sb,
 	 */
 	if (scoutfs_alloc_meta_low(sb, &tri->alloc, 16)) {
 		scoutfs_inc_counter(sb, trans_commit_meta_alloc_low);
-		queue_trans_work(sbi);
-		goto out;
+		return true;
 	}
 
 	/* Try to refill data allocator before premature enospc */
 	if (scoutfs_data_alloc_free_bytes(sb) <= SCOUTFS_TRANS_DATA_ALLOC_LWM) {
 		scoutfs_inc_counter(sb, trans_commit_data_alloc_low);
-		queue_trans_work(sbi);
+		return true;
+	}
+
+	return false;
+}
+
+static bool acquired_hold(struct super_block *sb)
+{
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	DECLARE_TRANS_INFO(sb, tri);
+	bool acquired;
+
+	/* if a caller already has a hold we acquire unconditionally */
+	if (inc_journal_info_holders()) {
+		atomic_inc(&tri->holders);
+		acquired = true;
 		goto out;
 	}
 
-hold:
-	rsv->holders++;
-	tri->holders++;
+	/* wait if the writer is blocking holds */
+	if (!inc_holders_unless_writer(tri)) {
+		dec_journal_info_holders();
+		acquired = false;
+		goto out;
+	}
+
+	/* wait if we're triggering another commit */
+	if (commit_before_hold(sb, tri)) {
+		release_holders(sb);
+		queue_trans_work(sbi);
+		acquired = false;
+		goto out;
+	}
+
+	trace_scoutfs_trans_acquired_hold(sb, current->journal_info, atomic_read(&tri->holders));
 	acquired = true;
-
 out:
-
-	spin_unlock(&tri->lock);
-
 	return acquired;
 }
 
+/*
+ * Try to hold the transaction.  Holding the transaction prevents it
+ * from being committed.  If a transaction is currently being written
+ * then we'll block until it's done and our hold can be granted.
+ *
+ * If a caller already holds the trans then we unconditionally acquire
+ * our hold and return to avoid deadlocks with our caller, the writing
+ * thread, and us.  We record nested holds in a call stack with the
+ * journal_info pointer in the task_struct.
+ *
+ * The writing thread marks itself as a global trans_task which
+ * short-circuits all the hold machinery so it can call code that would
+ * otherwise try to hold transactions while it is writing.
+ */
 int scoutfs_hold_trans(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
-	struct scoutfs_reservation *rsv;
-	int ret;
 
 	if (current == sbi->trans_task)
 		return 0;
 
-	rsv = current->journal_info;
-	if (rsv == NULL) {
-		rsv = kzalloc(sizeof(struct scoutfs_reservation), GFP_NOFS);
-		if (!rsv)
-			return -ENOMEM;
-
-		rsv->magic = SCOUTFS_RESERVATION_MAGIC;
-		current->journal_info = rsv;
-	}
-
-	BUG_ON(rsv->magic != SCOUTFS_RESERVATION_MAGIC);
-
-	ret = wait_event_interruptible(sbi->trans_hold_wq,
-				       acquired_hold(sb, rsv));
-	if (ret && rsv->holders == 0) {
-		current->journal_info = NULL;
-		kfree(rsv);
-	}
-	return ret;
+	return wait_event_interruptible(sbi->trans_hold_wq, acquired_hold(sb));
 }
 
 /*
@@ -441,50 +504,22 @@ int scoutfs_hold_trans(struct super_block *sb)
  */
 bool scoutfs_trans_held(void)
 {
-	struct scoutfs_reservation *rsv = current->journal_info;
+	unsigned long holders = (unsigned long)current->journal_info;
 
-	return rsv && rsv->magic == SCOUTFS_RESERVATION_MAGIC;
+	return (holders != 0 && ((holders & TRANS_JI_MAGIC_MASK) == TRANS_JI_MAGIC));
 }
 
-/*
- * As we drop the last hold in the reservation we try and wake other
- * hold attempts that were waiting for space.  As we drop the last trans
- * holder we try to wake a writing thread that was waiting for us to
- * finish.
- */
 void scoutfs_release_trans(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
-	struct scoutfs_reservation *rsv;
 	DECLARE_TRANS_INFO(sb, tri);
-	bool wake = false;
 
 	if (current == sbi->trans_task)
 		return;
 
-	rsv = current->journal_info;
-	BUG_ON(!rsv || rsv->magic != SCOUTFS_RESERVATION_MAGIC);
+	release_holders(sb);
 
-	spin_lock(&tri->lock);
-
-	trace_scoutfs_release_trans(sb, rsv, rsv->holders, tri->holders, tri->writing);
-
-	BUG_ON(rsv->holders <= 0);
-	BUG_ON(tri->holders <= 0);
-
-	if (--rsv->holders == 0) {
-		current->journal_info = NULL;
-		kfree(rsv);
-		wake = true;
-	}
-
-	if (--tri->holders == 0)
-		wake = true;
-
-	spin_unlock(&tri->lock);
-
-	if (wake)
-		wake_up(&sbi->trans_hold_wq);
+	trace_scoutfs_release_trans(sb, current->journal_info, atomic_read(&tri->holders));
 }
 
 /*
@@ -513,7 +548,7 @@ int scoutfs_setup_trans(struct super_block *sb)
 	if (!tri)
 		return -ENOMEM;
 
-	spin_lock_init(&tri->lock);
+	atomic_set(&tri->holders, 0);
 	scoutfs_block_writer_init(sb, &tri->wri);
 
 	sbi->trans_write_workq = alloc_workqueue("scoutfs_trans",

kmod/src/triggers.c

@@ -38,10 +38,7 @@ struct scoutfs_triggers {
 	struct scoutfs_triggers *name = SCOUTFS_SB(sb)->triggers
 
 static char *names[] = {
-	[SCOUTFS_TRIGGER_BTREE_STALE_READ] = "btree_stale_read",
-	[SCOUTFS_TRIGGER_BTREE_ADVANCE_RING_HALF] = "btree_advance_ring_half",
-	[SCOUTFS_TRIGGER_HARD_STALE_ERROR] = "hard_stale_error",
-	[SCOUTFS_TRIGGER_SEG_STALE_READ] = "seg_stale_read",
+	[SCOUTFS_TRIGGER_BLOCK_REMOVE_STALE] = "block_remove_stale",
 	[SCOUTFS_TRIGGER_STATFS_LOCK_PURGE] = "statfs_lock_purge",
 };
 

kmod/src/triggers.h

@@ -2,10 +2,7 @@
 #define _SCOUTFS_TRIGGERS_H_
 
 enum scoutfs_trigger {
-	SCOUTFS_TRIGGER_BTREE_STALE_READ,
-	SCOUTFS_TRIGGER_BTREE_ADVANCE_RING_HALF,
-	SCOUTFS_TRIGGER_HARD_STALE_ERROR,
-	SCOUTFS_TRIGGER_SEG_STALE_READ,
+	SCOUTFS_TRIGGER_BLOCK_REMOVE_STALE,
 	SCOUTFS_TRIGGER_STATFS_LOCK_PURGE,
 	SCOUTFS_TRIGGER_NR,
 };

tests/funcs/fs.sh

@@ -209,12 +209,19 @@ t_trigger_show() {
 	echo "trigger $which $string: $(t_trigger_get $which $nr)"
 }
 
-t_trigger_arm() {
+t_trigger_arm_silent() {
 	local which="$1"
 	local nr="$2"
 	local path=$(t_trigger_path "$nr")
 
 	echo 1 > "$path/$which"
+}
+
+t_trigger_arm() {
+	local which="$1"
+	local nr="$2"
+
+	t_trigger_arm_silent $which $nr
 	t_trigger_show $which armed $nr
 }
 
@@ -229,16 +236,44 @@ t_counter() {
 	cat "$(t_sysfs_path $nr)/counters/$which"
 }
 
+#
+# output the difference between the current value of a counter and the
+# caller's provided previous value.
+#
+t_counter_diff_value() {
+	local which="$1"
+	local old="$2"
+	local nr="$3"
+	local new="$(t_counter $which $nr)"
+
+	echo "$((new - old))"
+}
+
 #
 # output the value of the given counter for the given mount, defaulting
-# to mount 0 if a mount isn't specified.
+# to mount 0 if a mount isn't specified.  For tests which expect a
+# specific difference in counters.
 #
 t_counter_diff() {
 	local which="$1"
 	local old="$2"
 	local nr="$3"
-	local new
 
-	new="$(t_counter $which $nr)"
-	echo "counter $which diff $((new - old))"
+	echo "counter $which diff $(t_counter_diff_value $which $old $nr)"
+}
+
+#
+# output a message indicating whether or not the counter value changed.
+# For tests that expect a difference, or not, but the amount of
+# difference isn't significant.
+#
+t_counter_diff_changed() {
+	local which="$1"
+	local old="$2"
+	local nr="$3"
+	local diff="$(t_counter_diff_value $which $old $nr)"
+
+	test "$diff" -eq 0 && \
+		echo "counter $which didn't change" ||
+		echo "counter $which changed"
 }

tests/golden/block-stale-reads

@@ -0,0 +1,52 @@
+== create shared test file
+== set and get xattrs between mount pairs while retrying
+# file: /mnt/test/test/block-stale-reads/file
+user.xat="1"
+
+counter block_cache_remove_stale changed
+counter block_cache_remove_stale changed
+# file: /mnt/test/test/block-stale-reads/file
+user.xat="2"
+
+counter block_cache_remove_stale changed
+counter block_cache_remove_stale changed
+# file: /mnt/test/test/block-stale-reads/file
+user.xat="3"
+
+counter block_cache_remove_stale changed
+counter block_cache_remove_stale changed
+# file: /mnt/test/test/block-stale-reads/file
+user.xat="4"
+
+counter block_cache_remove_stale changed
+counter block_cache_remove_stale changed
+# file: /mnt/test/test/block-stale-reads/file
+user.xat="5"
+
+counter block_cache_remove_stale changed
+counter block_cache_remove_stale changed
+# file: /mnt/test/test/block-stale-reads/file
+user.xat="6"
+
+counter block_cache_remove_stale changed
+counter block_cache_remove_stale changed
+# file: /mnt/test/test/block-stale-reads/file
+user.xat="7"
+
+counter block_cache_remove_stale changed
+counter block_cache_remove_stale changed
+# file: /mnt/test/test/block-stale-reads/file
+user.xat="8"
+
+counter block_cache_remove_stale changed
+counter block_cache_remove_stale changed
+# file: /mnt/test/test/block-stale-reads/file
+user.xat="9"
+
+counter block_cache_remove_stale changed
+counter block_cache_remove_stale changed
+# file: /mnt/test/test/block-stale-reads/file
+user.xat="10"
+
+counter block_cache_remove_stale changed
+counter block_cache_remove_stale changed

tests/golden/stale-btree-read

@@ -1,11 +0,0 @@
-== create file for xattr ping pong
-# file: /mnt/test/test/stale-btree-read/file
-user.xat="initial"
-
-== retry btree block read
-trigger btree_stale_read armed: 1
-# file: /mnt/test/test/stale-btree-read/file
-user.xat="btree"
-
-trigger btree_stale_read after: 0
-counter btree_stale_read diff 1

tests/sequence

@@ -28,5 +28,5 @@ setup-error-teardown.sh
 mount-unmount-race.sh
 createmany-parallel-mounts.sh
 archive-light-cycle.sh
-stale-btree-read.sh
+block-stale-reads.sh
 xfstests.sh

tests/tests/block-stale-reads.sh

@@ -0,0 +1,61 @@
+#
+# Exercise stale block reading.
+#
+# It would be very difficult to manipulate the allocators, cache, and
+# persistent blocks to create stable block reading scenarios.    Instead
+# we use triggers to exercise how readers encounter stale blocks.
+#
+
+t_require_commands touch setfattr getfattr
+
+inc_wrap_fs_nr()
+{
+	local nr="$(($1 + 1))"
+
+	if [ "$nr" == "$T_NR_MOUNTS" ]; then
+		nr=0
+	fi
+
+	echo $nr
+}
+
+GETFATTR="getfattr --absolute-names"
+SETFATTR="setfattr"
+
+echo "== create shared test file"
+touch "$T_D0/file"
+$SETFATTR -n user.xat -v 0 "$T_D0/file"
+
+#
+# Trigger retries in the block cache as we bounce xattr values around
+# between sequential pairs of mounts.  This is a little silly because if
+# either of the mounts are the server then they'll almost certaily have
+# their trigger fired prematurely by message handling btree calls while
+# working with the t_ helpers long before we work with the xattrs.  But
+# the block cache stale retry path is still being exercised.
+#
+echo "== set and get xattrs between mount pairs while retrying"
+set_nr=0
+get_nr=$(inc_wrap_fs_nr $set_nr)
+
+for i in $(seq 1 10); do
+	eval set_file="\$T_D${set_nr}/file"
+	eval get_file="\$T_D${get_nr}/file"
+
+	old_set=$(t_counter block_cache_remove_stale $set_nr)
+	old_get=$(t_counter block_cache_remove_stale $get_nr)
+
+	t_trigger_arm_silent block_remove_stale $set_nr
+	t_trigger_arm_silent block_remove_stale $get_nr
+
+	$SETFATTR -n user.xat -v $i "$set_file"
+	$GETFATTR -n user.xat "$get_file" 2>&1 | t_filter_fs
+
+	t_counter_diff_changed block_cache_remove_stale $old_set $set_nr
+	t_counter_diff_changed block_cache_remove_stale $old_get $get_nr
+
+	set_nr="$get_nr"
+	get_nr=$(inc_wrap_fs_nr $set_nr)
+done
+
+t_pass

tests/tests/stale-btree-read.sh

@@ -1,40 +0,0 @@
-#
-# verify stale btree block reading
-#
-
-t_require_commands touch stat setfattr getfattr createmany
-t_require_mounts 2
-
-GETFATTR="getfattr --absolute-names"
-SETFATTR="setfattr"
-
-#
-# This exercises the soft retry of btree blocks when
-# inconsistent cached versions are found.  It ensures that basic hard
-# error returning turns into EIO in the case where the persistent reread
-# blocks and segments really are inconsistent.
-#
-# The triggers apply across all execution in the file system.  So to
-# trigger btree block retries in the client we make sure that the server
-# is running on the other node.
-# 
-
-cl=$(t_first_client_nr)
-sv=$(t_server_nr)
-eval cl_dir="\$T_D${cl}"
-eval sv_dir="\$T_D${sv}"
-
-echo "== create file for xattr ping pong"
-touch "$sv_dir/file"
-$SETFATTR -n user.xat -v initial "$sv_dir/file"
-$GETFATTR -n user.xat "$sv_dir/file" 2>&1 | t_filter_fs
-
-echo "== retry btree block read"
-$SETFATTR -n user.xat -v btree "$sv_dir/file"
-t_trigger_arm btree_stale_read $cl
-old=$(t_counter btree_stale_read $cl)
-$GETFATTR -n user.xat "$cl_dir/file" 2>&1 | t_filter_fs
-t_trigger_show btree_stale_read "after" $cl
-t_counter_diff btree_stale_read $old $cl
-
-t_pass

utils/src/mkfs.c

@@ -361,12 +361,12 @@ static int do_mkfs(struct mkfs_args *args)
 		struct scoutfs_quorum_slot *sl = &super->qconf.slots[i];
 		struct in_addr in;
 
-		if (sl->addr.addr == 0)
+		if (sl->addr.v4.family != cpu_to_le16(SCOUTFS_AF_IPV4))
 			continue;
 
-		in.s_addr = htonl(le32_to_cpu(sl->addr.addr));
+		in.s_addr = htonl(le32_to_cpu(sl->addr.v4.addr));
 		printf("%s%u: %s:%u", indent,
-		       i, inet_ntoa(in), le16_to_cpu(sl->addr.port));
+		       i, inet_ntoa(in), le16_to_cpu(sl->addr.v4.port));
 		indent = "\n                        ";
 	}
 	printf("\n");
@@ -395,22 +395,28 @@ static bool valid_quorum_slots(struct scoutfs_quorum_slot *slots)
 	int j;
 
 	for (i = 0; i < SCOUTFS_QUORUM_MAX_SLOTS; i++) {
-		if (slots[i].addr.addr == 0)
+		if (slots[i].addr.v4.family == cpu_to_le16(SCOUTFS_AF_NONE))
 			continue;
 
+		if (slots[i].addr.v4.family != cpu_to_le16(SCOUTFS_AF_IPV4)) {
+			fprintf(stderr, "quorum slot nr %u has invalid family %u\n",
+				i, le16_to_cpu(slots[i].addr.v4.family));
+			valid = false;
+		}
+
 		for (j = i + 1; j < SCOUTFS_QUORUM_MAX_SLOTS; j++) {
-			if (slots[j].addr.addr == 0)
+			if (slots[i].addr.v4.family != cpu_to_le16(SCOUTFS_AF_IPV4))
 				continue;
 
-			if (slots[i].addr.addr == slots[j].addr.addr &&
-			    slots[i].addr.port == slots[j].addr.port) {
+			if (slots[i].addr.v4.addr == slots[j].addr.v4.addr &&
+			    slots[i].addr.v4.port == slots[j].addr.v4.port) {
 
 				in.s_addr =
-					htonl(le32_to_cpu(slots[i].addr.addr));
+					htonl(le32_to_cpu(slots[i].addr.v4.addr));
 				addr = inet_ntoa(in);
 				fprintf(stderr, "quorum slot nr %u and %u have the same address %s:%u\n",
 					i, j, addr,
-					le16_to_cpu(slots[i].addr.port));
+					le16_to_cpu(slots[i].addr.v4.port));
 				valid = false;
 			}
 		}
@@ -430,7 +436,7 @@ static int parse_opt(int key, char *arg, struct argp_state *state)
 		ret = parse_quorum_slot(&slot, arg);
 		if (ret < 0)
 			return ret;
-		if (args->slots[ret].addr.addr != 0)
+		if (args->slots[ret].addr.v4.family != cpu_to_le16(SCOUTFS_AF_NONE))
 			argp_error(state, "Quorum slot %u already specified before slot '%s'\n",
 				   ret, arg);
 		args->slots[ret] = slot;

utils/src/parse.c

@@ -213,7 +213,8 @@ int parse_quorum_slot(struct scoutfs_quorum_slot *slot, char *arg)
 		return -EINVAL;
 	}
 
-	slot->addr.addr = cpu_to_le32(htonl(in.s_addr));
-	slot->addr.port = cpu_to_le16(port);
+	slot->addr.v4.family = cpu_to_le16(SCOUTFS_AF_IPV4);
+	slot->addr.v4.addr = cpu_to_le32(htonl(in.s_addr));
+	slot->addr.v4.port = cpu_to_le16(port);
 	return nr;
 }

utils/src/print.c

@@ -388,10 +388,10 @@ static int print_alloc_item(struct scoutfs_key *key, void *val,
 typedef int (*print_item_func)(struct scoutfs_key *key, void *val,
 			       unsigned val_len, void *arg);
 
-static int print_btree_ref(struct scoutfs_key *key, void *val,
+static int print_block_ref(struct scoutfs_key *key, void *val,
 			   unsigned val_len, print_item_func func, void *arg)
 {
-	struct scoutfs_btree_ref *ref = val;
+	struct scoutfs_block_ref *ref = val;
 
 	func(key, NULL, 0, arg);
 	printf("    ref blkno %llu seq %llu\n",
@@ -433,7 +433,7 @@ static void print_leaf_item_hash(struct scoutfs_btree_block *bt)
 }
 
 static int print_btree_block(int fd, struct scoutfs_super_block *super,
-			     char *which, struct scoutfs_btree_ref *ref,
+			     char *which, struct scoutfs_block_ref *ref,
 			     print_item_func func, void *arg, u8 level)
 {
 	struct scoutfs_btree_item *item;
@@ -500,7 +500,7 @@ static int print_btree_block(int fd, struct scoutfs_super_block *super,
 			val_len);
 
 		if (level)
-			print_btree_ref(key, val, val_len, func, arg);
+			print_block_ref(key, val, val_len, func, arg);
 		else
 			func(key, val, val_len, arg);
 	}
@@ -531,11 +531,10 @@ static int print_btree(int fd, struct scoutfs_super_block *super, char *which,
 	return ret;
 }
 
-static int print_alloc_list_block(int fd, char *str,
-				  struct scoutfs_alloc_list_ref *ref)
+static int print_alloc_list_block(int fd, char *str, struct scoutfs_block_ref *ref)
 {
 	struct scoutfs_alloc_list_block *lblk;
-	struct scoutfs_alloc_list_ref next;
+	struct scoutfs_block_ref next;
 	u64 blkno;
 	u64 start;
 	u64 len;
@@ -583,7 +582,7 @@ static int print_alloc_list_block(int fd, char *str,
 	return print_alloc_list_block(fd, str, &next);
 }
 
-static int print_srch_block(int fd, struct scoutfs_srch_ref *ref, int level)
+static int print_srch_block(int fd, struct scoutfs_block_ref *ref, int level)
 {
 	struct scoutfs_srch_parent *srp;
 	struct scoutfs_srch_block *srb;
@@ -729,7 +728,7 @@ static int print_srch_root_files(struct scoutfs_key *key, void *val,
 }
 
 static int print_btree_leaf_items(int fd, struct scoutfs_super_block *super,
-				  struct scoutfs_btree_ref *ref,
+				  struct scoutfs_block_ref *ref,
 				  print_item_func func, void *arg)
 {
 	struct scoutfs_btree_item *item;
@@ -770,7 +769,7 @@ static int print_btree_leaf_items(int fd, struct scoutfs_super_block *super,
 	return 0;
 }
 
-static char *alloc_addr_str(struct scoutfs_inet_addr *ia)
+static char *alloc_addr_str(union scoutfs_inet_addr *ia)
 {
 	struct in_addr addr;
 	char *quad;
@@ -778,12 +777,12 @@ static char *alloc_addr_str(struct scoutfs_inet_addr *ia)
 	int len;
 
 	memset(&addr, 0, sizeof(addr));
-	addr.s_addr = htonl(le32_to_cpu(ia->addr));
+	addr.s_addr = htonl(le32_to_cpu(ia->v4.addr));
 	quad = inet_ntoa(addr);
 	if (quad == NULL)
 		return NULL;
 
-	len = snprintf(NULL, 0, "%s:%u", quad, le16_to_cpu(ia->port));
+	len = snprintf(NULL, 0, "%s:%u", quad, le16_to_cpu(ia->v4.port));
 	if (len < 1 || len > 22)
 		return NULL;
 
@@ -792,7 +791,7 @@ static char *alloc_addr_str(struct scoutfs_inet_addr *ia)
 	if (!str)
 		return NULL;
 
-	snprintf(str, len, "%s:%u", quad, le16_to_cpu(ia->port));
+	snprintf(str, len, "%s:%u", quad, le16_to_cpu(ia->v4.port));
 	return str;
 }
 
@@ -916,8 +915,7 @@ static void print_super_block(struct scoutfs_super_block *super, u64 blkno)
 	printf("  quorum config version %llu\n",
 		le64_to_cpu(super->qconf.version));
 	for (i = 0; i < array_size(super->qconf.slots); i++) {
-		if (!super->qconf.slots[i].addr.addr &&
-		    !super->qconf.slots[i].addr.port)
+		if (super->qconf.slots[i].addr.v4.family != cpu_to_le16(SCOUTFS_AF_IPV4))
 			continue;
 
 		addr = alloc_addr_str(&super->qconf.slots[i].addr);