Server updates ring for level0 segment writes

Transaction commits currently directly modify the ring and super block
as segments are written.  As we introduce shared mounts only the server
can modify the ring and super blocks.

This adds network messages to let mounts write items in a level 0
segment while the server modifies the allocator and manifest.

The item transaction commit now sends a message to the server to get an
allocated segno for its new level0 segment and sends a manifest entry to
the server once the segment is written.  The request and reply handlers
for the functions are straight forward.  The processing paths are simple
wrappers around the allocation and update functions that transaction
writing used to call directly.

Now that the item transactions aren't updating the super sync can't
work with the super sequence numbers.

The server needs to make both allocations and manifest updates
persistent before it sends replies to the client.  We add the ability
for the server processing paths to queue and wait for commits of the
rings and super block.  We can hopefull get reasonable batching by using
a work struct for the commit.  We update the other processing path
callers that modify the rings to use the new commit mechanism.

We add a few segment and manifest functions to work with manifest
entries that describe segments.  This creats a bit of similar looking
code thorughout the segment and manifest code but we'll come back and
clean this up once we see what the final shared support looks like.

scoutfs_seg_alloc() now takes the segno from the caller for the segment
it's allocating and inserting into the cache.  Transaction commit uses
the segno it got from the server while compaction still allocates
locally.

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

kmod/src/compact.c

@@ -24,6 +24,7 @@
 #include "manifest.h"
 #include "trans.h"
 #include "counters.h"
+#include "alloc.h"
 #include "scoutfs_trace.h"
 
 /*
@@ -347,6 +348,7 @@ static int compact_segments(struct super_block *sb,
 	struct compact_seg *lower;
 	u32 key_bytes;
 	u32 nr_items;
+	u64 segno;
 	int ret;
 
 	scoutfs_inc_counter(sb, compact_operations);
@@ -444,12 +446,19 @@ static int compact_segments(struct super_block *sb,
 			break;
 		}
 
-		ret = scoutfs_seg_alloc(sb, &seg);
+		ret = scoutfs_alloc_segno(sb, &segno);
 		if (ret) {
 			kfree(cseg);
 			break;
 		}
 
+		ret = scoutfs_seg_alloc(sb, segno, &seg);
+		if (ret) {
+			scoutfs_alloc_free(sb, segno);
+			kfree(cseg);
+			break;
+		}
+
 		/* csegs will be claned up once they're on the list */
 		cseg->level = curs->lower_level;
 		cseg->seg = seg;

kmod/src/format.h

@@ -389,6 +389,8 @@ enum {
 	SCOUTFS_NET_TRADE_TIME = 0,
 	SCOUTFS_NET_ALLOC_INODES,
 	SCOUTFS_NET_MANIFEST_RANGE_ENTRIES,
+	SCOUTFS_NET_ALLOC_SEGNO,
+	SCOUTFS_NET_RECORD_SEGMENT,
 	SCOUTFS_NET_UNKNOWN,
 };
 

kmod/src/manifest.c

@@ -210,6 +210,45 @@ int scoutfs_manifest_add(struct super_block *sb,
 	return 0;
 }
 
+/*
+ * Add a manifest entry as provided by the caller instead of exploded
+ * out into arguments.
+ *
+ * This must be called with the manifest lock held.
+ */
+int scoutfs_manifest_add_ment(struct super_block *sb,
+			      struct scoutfs_manifest_entry *add)
+{
+	DECLARE_MANIFEST(sb, mani);
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	struct scoutfs_super_block *super = &sbi->super;
+	struct scoutfs_manifest_entry *ment;
+	struct manifest_search_key skey;
+	struct scoutfs_key_buf first;
+	unsigned bytes;
+
+	lockdep_assert_held(&mani->rwsem);
+
+	init_ment_keys(add, &first, NULL);
+
+	skey.key = &first;
+	skey.level = add->level;
+	skey.seq = le64_to_cpu(add->seq);
+
+	bytes = scoutfs_manifest_bytes(add);
+
+	ment = scoutfs_ring_insert(&mani->ring, &skey, bytes);
+	if (!ment)
+		return -ENOMEM;
+
+	memcpy(ment, add, bytes);
+
+	mani->nr_levels = max_t(u8, mani->nr_levels, add->level + 1);
+	add_level_count(sb, mani, super, add->level, 1);
+
+	return 0;
+}
+
 /*
  * This must be called with the manifest lock held.
  */
@@ -268,6 +307,41 @@ int scoutfs_manifest_bytes(struct scoutfs_manifest_entry *ment)
 	       le16_to_cpu(ment->last_key_len);
 }
 
+/*
+ * Return an allocated and filled in manifest entry.
+ */
+struct scoutfs_manifest_entry *
+scoutfs_manifest_alloc_entry(struct super_block *sb,
+			     struct scoutfs_key_buf *first,
+			     struct scoutfs_key_buf *last, u64 segno, u64 seq,
+			     u8 level)
+{
+	struct scoutfs_manifest_entry *ment;
+	struct scoutfs_key_buf ment_first;
+	struct scoutfs_key_buf ment_last;
+	unsigned key_bytes;
+	unsigned bytes;
+
+	key_bytes = first->key_len + last->key_len;
+	bytes = offsetof(struct scoutfs_manifest_entry, keys[key_bytes]);
+
+	ment = kmalloc(bytes, GFP_NOFS);
+	if (!ment)
+		return NULL;
+
+	ment->segno = cpu_to_le64(segno);
+	ment->seq = cpu_to_le64(seq);
+	ment->first_key_len = cpu_to_le16(first->key_len);
+	ment->last_key_len = cpu_to_le16(last->key_len);
+	ment->level = level;
+
+	init_ment_keys(ment, &ment_first, &ment_last);
+	scoutfs_key_copy(&ment_first, first);
+	scoutfs_key_copy(&ment_last, last);
+
+	return ment;
+}
+
 /*
  * XXX This feels pretty gross, but it's a simple way to give compaction
  * atomic updates.  It'll go away once compactions go to the trouble of

kmod/src/manifest.h

@@ -8,6 +8,8 @@ int scoutfs_manifest_add(struct super_block *sb,
 			 struct scoutfs_key_buf *first,
 			 struct scoutfs_key_buf *last, u64 segno, u64 seq,
 			 u8 level);
+int scoutfs_manifest_add_ment(struct super_block *sb,
+			      struct scoutfs_manifest_entry *add);
 int scoutfs_manifest_dirty(struct super_block *sb,
 			   struct scoutfs_key_buf *first, u64 seq, u8 level);
 int scoutfs_manifest_del(struct super_block *sb, struct scoutfs_key_buf *first,
@@ -19,6 +21,12 @@ void scoutfs_manifest_write_complete(struct super_block *sb);
 
 int scoutfs_manifest_bytes(struct scoutfs_manifest_entry *ment);
 
+struct scoutfs_manifest_entry *
+scoutfs_manifest_alloc_entry(struct super_block *sb,
+			     struct scoutfs_key_buf *first,
+			     struct scoutfs_key_buf *last, u64 segno, u64 seq,
+			     u8 level);
+
 int scoutfs_manifest_lock(struct super_block *sb);
 int scoutfs_manifest_unlock(struct super_block *sb);
 

kmod/src/net.c

@@ -24,6 +24,9 @@
 #include "counters.h"
 #include "inode.h"
 #include "manifest.h"
+#include "bio.h"
+#include "alloc.h"
+#include "seg.h"
 #include "scoutfs_trace.h"
 
 /*
@@ -71,6 +74,11 @@ struct net_info {
 	struct delayed_work server_work;
 	struct sock_info *listening_sinf;
 
+	/* server commits ring changes while processing requests */
+	struct rw_semaphore ring_commit_rwsem;
+	struct llist_head ring_commit_waiters;
+	struct work_struct ring_commit_work;
+
 	/* both track active sockets for destruction */
 	struct list_head active_socks;
 
@@ -243,6 +251,102 @@ static void scoutfs_net_send_func(struct work_struct *work)
 	mutex_unlock(&nti->mutex);
 }
 
+struct commit_waiter {
+	struct completion comp;
+	struct llist_node node;
+	int ret;
+};
+
+/*
+ * This is called while still holding the rwsem that prevents commits so
+ * that the caller can be sure to be woken by the next commit after they
+ * queue and release the lock.
+ *
+ * This could queue delayed work but we're first trying to have batching
+ * work by having concurrent modification line up behind a commit in
+ * flight.  Once the commit finishes it'll unlock and hopefully everyone
+ * will race to make their changes and they'll all be applied by the
+ * next commit after that.
+ */
+static void queue_commit_work(struct net_info *nti, struct commit_waiter *cw)
+{
+	lockdep_assert_held(&nti->ring_commit_rwsem);
+
+	cw->ret = 0;
+	init_completion(&cw->comp);
+	llist_add(&cw->node, &nti->ring_commit_waiters);
+	queue_work(nti->proc_wq, &nti->ring_commit_work);
+}
+
+static int wait_for_commit(struct commit_waiter *cw)
+{
+	wait_for_completion(&cw->comp);
+	return cw->ret;
+}
+
+/*
+ * A core function of request processing is to modify the manifest and
+ * allocator.  Often the processing needs to make the modifications
+ * persistent before replying.  We'd like to batch these commits as much
+ * as is reasonable so that we don't degrade to a few IO round trips per
+ * request.
+ *
+ * Getting that batching right is bound up in the concurrency of request
+ * processing so a clear way to implement the batched commits is to
+ * implement commits with work funcs like the processing.  This ring
+ * commit work is queued on the non-reentrant proc_wq so there will only
+ * ever be one commit executing at a time.
+ *
+ * Processing paths acquire the rwsem for reading while they're making
+ * multiple dependent changes.  When they're done and want it persistent
+ * they add themselves to the list of waiters and queue the commit work.
+ * This work runs, acquires the lock to exclude other writers, and
+ * performs the commit.  Readers can run concurrently with these
+ * commits.
+ */
+static void scoutfs_net_ring_commit_func(struct work_struct *work)
+{
+	struct net_info *nti = container_of(work, struct net_info,
+					    ring_commit_work);
+	struct super_block *sb = nti->sb;
+	struct scoutfs_bio_completion comp;
+	struct commit_waiter *cw;
+	struct commit_waiter *pos;
+	struct llist_node *node;
+	int ret;
+
+	scoutfs_bio_init_comp(&comp);
+
+	down_write(&nti->ring_commit_rwsem);
+
+	if (scoutfs_manifest_has_dirty(sb) || scoutfs_alloc_has_dirty(sb)) {
+		ret = scoutfs_manifest_submit_write(sb, &comp) ?:
+		      scoutfs_alloc_submit_write(sb, &comp) ?:
+		      scoutfs_bio_wait_comp(sb, &comp) ?:
+		      scoutfs_write_dirty_super(sb);
+
+		/* we'd need to loop or something */
+		BUG_ON(ret);
+
+		scoutfs_manifest_write_complete(sb);
+		scoutfs_alloc_write_complete(sb);
+
+		scoutfs_advance_dirty_super(sb);
+	} else {
+		ret = 0;
+	}
+
+	node = llist_del_all(&nti->ring_commit_waiters);
+
+	/* waiters always wait on completion, cw could be free after complete */
+	llist_for_each_entry_safe(cw, pos, node, node) {
+		cw->ret = ret;
+		complete(&cw->comp);
+	}
+
+	up_write(&nti->ring_commit_rwsem);
+}
+
 static struct send_buf *alloc_sbuf(unsigned data_len)
 {
 	unsigned len = offsetof(struct send_buf, nh[0].data[data_len]);
@@ -257,6 +361,98 @@ static struct send_buf *alloc_sbuf(unsigned data_len)
 	return sbuf;
 }
 
+static struct send_buf *process_record_segment(struct super_block *sb,
+					       void *req, int req_len)
+{
+	DECLARE_NET_INFO(sb, nti);
+	struct scoutfs_manifest_entry *ment;
+	struct commit_waiter cw;
+	struct send_buf *sbuf;
+	int ret;
+
+	if (req_len < sizeof(struct scoutfs_manifest_entry)) {
+		sbuf = ERR_PTR(-EINVAL);
+		goto out;
+	}
+
+	ment = req;
+
+	if (req_len != scoutfs_manifest_bytes(ment)) {
+		sbuf = ERR_PTR(-EINVAL);
+		goto out;
+	}
+
+	down_read(&nti->ring_commit_rwsem);
+
+	scoutfs_manifest_lock(sb);
+	ret = scoutfs_manifest_add_ment(sb, ment);
+	scoutfs_manifest_unlock(sb);
+
+	if (ret == 0)
+		queue_commit_work(nti, &cw);
+	up_read(&nti->ring_commit_rwsem);
+
+	if (ret == 0)
+		ret = wait_for_commit(&cw);
+
+	sbuf = alloc_sbuf(0);
+	if (!sbuf) {
+		sbuf = ERR_PTR(-ENOMEM);
+		goto out;
+	}
+
+	if (ret)
+		sbuf->nh->status = SCOUTFS_NET_STATUS_ERROR;
+	else
+		sbuf->nh->status = SCOUTFS_NET_STATUS_SUCCESS;
+out:
+	return sbuf;
+}
+
+static struct send_buf *process_alloc_segno(struct super_block *sb,
+					    void *req, int req_len)
+{
+	DECLARE_NET_INFO(sb, nti);
+	__le64 * __packed lesegno;
+	struct commit_waiter cw;
+	struct send_buf *sbuf;
+	u64 segno;
+	int ret;
+
+	if (req_len != 0) {
+		sbuf = ERR_PTR(-EINVAL);
+		goto out;
+	}
+
+	down_read(&nti->ring_commit_rwsem);
+
+	ret = scoutfs_alloc_segno(sb, &segno);
+	if (ret == 0)
+		queue_commit_work(nti, &cw);
+
+	up_read(&nti->ring_commit_rwsem);
+
+	if (ret == 0)
+		ret = wait_for_commit(&cw);
+
+	sbuf = alloc_sbuf(sizeof(__le64));
+	if (!sbuf) {
+		sbuf = ERR_PTR(-ENOMEM);
+		goto out;
+	}
+
+	if (ret) {
+		sbuf->nh->status = SCOUTFS_NET_STATUS_ERROR;
+	} else {
+		lesegno = (void *)sbuf->nh->data;
+		*lesegno = cpu_to_le64(segno);
+		sbuf->nh->status = SCOUTFS_NET_STATUS_SUCCESS;
+	}
+
+out:
+	return sbuf;
+}
+
 /*
  * Find the manifest entries that intersect with the request's key
  * range.  We lock the manifest and get pointers to the manifest entries
@@ -330,9 +526,11 @@ out:
 static struct send_buf *process_alloc_inodes(struct super_block *sb,
 					     void *req, int req_len)
 {
+	DECLARE_NET_INFO(sb, nti);
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_super_block *super = &sbi->super;
 	struct scoutfs_net_inode_alloc *ial;
+	struct commit_waiter cw;
 	struct send_buf *sbuf;
 	int ret;
 	u64 ino;
@@ -345,22 +543,27 @@ static struct send_buf *process_alloc_inodes(struct super_block *sb,
 	if (!sbuf)
 		return ERR_PTR(-ENOMEM);
 
-	spin_lock(&sbi->next_ino_lock);
+	down_read(&nti->ring_commit_rwsem);
 
+	spin_lock(&sbi->next_ino_lock);
 	ino = le64_to_cpu(super->next_ino);
 	nr = min(100000ULL, ~0ULL - ino);
 	le64_add_cpu(&super->next_ino, nr);
-
 	spin_unlock(&sbi->next_ino_lock);
 
-	/* XXX think about server ring commits */
-	ret = 0; //sync_or_something();
+	queue_commit_work(nti, &cw);
+	up_read(&nti->ring_commit_rwsem);
+
+	ret = wait_for_commit(&cw);
 
 	ial = (void *)sbuf->nh->data;
 	ial->ino = cpu_to_le64(ino);
 	ial->nr = cpu_to_le64(nr);
 
-	sbuf->nh->status = SCOUTFS_NET_STATUS_SUCCESS;
+	if (ret < 0)
+		sbuf->nh->status = SCOUTFS_NET_STATUS_ERROR;
+	else
+		sbuf->nh->status = SCOUTFS_NET_STATUS_SUCCESS;
 
 	return sbuf;
 }
@@ -369,9 +572,9 @@ static struct send_buf *process_alloc_inodes(struct super_block *sb,
  * Log the time in the request and reply with our current time.
  */
 static struct send_buf *process_trade_time(struct super_block *sb,
-					   struct scoutfs_timespec *req,
-					   int req_len)
+					   void *r, int req_len)
 {
+	struct scoutfs_timespec *req = r;
 	struct scoutfs_timespec *reply;
 	struct send_buf *sbuf;
 	struct timespec64 ts;
@@ -397,6 +600,23 @@ static struct send_buf *process_trade_time(struct super_block *sb,
 	return sbuf;
 }
 
+typedef struct send_buf *(*proc_func_t)(struct super_block *sb, void *req,
+				        int req_len);
+
+static proc_func_t type_proc_func(u8 type)
+{
+	static proc_func_t funcs[] = {
+		[SCOUTFS_NET_TRADE_TIME] = process_trade_time,
+		[SCOUTFS_NET_ALLOC_INODES] = process_alloc_inodes,
+		[SCOUTFS_NET_MANIFEST_RANGE_ENTRIES] =
+			process_manifest_range_entries,
+		[SCOUTFS_NET_ALLOC_SEGNO] = process_alloc_segno,
+		[SCOUTFS_NET_RECORD_SEGMENT] = process_record_segment,
+	};
+
+	return type < SCOUTFS_NET_UNKNOWN ? funcs[type] : NULL;
+}
+
 /*
  * Process an incoming request and queue its reply to send if the socket
  * is still open by the time we have the reply.
@@ -405,23 +625,15 @@ static int process_request(struct net_info *nti, struct recv_buf *rbuf)
 {
 	struct super_block *sb = nti->sb;
 	struct send_buf *sbuf;
+	proc_func_t proc;
 	unsigned data_len;
 
 	data_len = le16_to_cpu(rbuf->nh->data_len);
-
-	if (rbuf->nh->type == SCOUTFS_NET_TRADE_TIME)
-		sbuf = process_trade_time(sb, (void *)rbuf->nh->data,
-					  data_len);
-	else if (rbuf->nh->type == SCOUTFS_NET_ALLOC_INODES)
-		sbuf = process_alloc_inodes(sb, (void *)rbuf->nh->data,
-					    data_len);
-	else if (rbuf->nh->type == SCOUTFS_NET_MANIFEST_RANGE_ENTRIES)
-		sbuf = process_manifest_range_entries(sb,
-						      (void *)rbuf->nh->data,
-						      data_len);
+	proc = type_proc_func(rbuf->nh->type);
+	if (proc)
+		sbuf = proc(sb, (void *)rbuf->nh->data, data_len);
 	else
 		sbuf = ERR_PTR(-EINVAL);
-
 	if (IS_ERR(sbuf))
 		return PTR_ERR(sbuf);
 
@@ -844,6 +1056,93 @@ static int add_send_buf(struct super_block *sb, int type, void *data,
 	return 0;
 }
 
+struct record_segment_args {
+	struct completion comp;
+	int ret;
+};
+
+static int record_segment_reply(struct super_block *sb, void *reply, int ret,
+				void *arg)
+{
+	struct record_segment_args *args = arg;
+
+	if (ret > 0)
+		ret = -EINVAL;
+
+	args->ret = ret;
+	complete(&args->comp);
+	return args->ret;
+}
+
+int scoutfs_net_record_segment(struct super_block *sb,
+			       struct scoutfs_segment *seg, u8 level)
+{
+	struct scoutfs_manifest_entry *ment;
+	struct record_segment_args args;
+	int ret;
+
+	ment = scoutfs_seg_manifest_entry(sb, seg, level);
+	if (!ment) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
+	init_completion(&args.comp);
+
+	ret = add_send_buf(sb, SCOUTFS_NET_RECORD_SEGMENT, ment,
+			   scoutfs_manifest_bytes(ment),
+			   record_segment_reply, &args);
+	kfree(ment);
+	if (ret == 0) {
+		wait_for_completion(&args.comp);
+		ret = args.ret;
+	}
+out:
+	return ret;
+}
+
+struct alloc_segno_args {
+	u64 segno;
+	struct completion comp;
+	int ret;
+};
+
+static int alloc_segno_reply(struct super_block *sb, void *reply, int ret,
+			     void *arg)
+{
+	struct alloc_segno_args *args = arg;
+	__le64 * __packed segno = reply;
+
+	if (ret == sizeof(__le64)) {
+		args->segno = le64_to_cpup(segno);
+		args->ret = 0;
+	} else {
+		args->ret = -EINVAL;
+	}
+
+	complete(&args->comp); /* args can be freed from this point */
+	return args->ret;
+}
+
+int scoutfs_net_alloc_segno(struct super_block *sb, u64 *segno)
+{
+	struct alloc_segno_args args;
+	int ret;
+
+	init_completion(&args.comp);
+
+	ret = add_send_buf(sb, SCOUTFS_NET_ALLOC_SEGNO, NULL, 0,
+			   alloc_segno_reply, &args);
+	if (ret == 0) {
+		wait_for_completion(&args.comp);
+		*segno = args.segno;
+		ret = args.ret;
+		if (ret == 0 && *segno == 0)
+			ret = -ENOSPC;
+	}
+	return ret;
+}
+
 struct manifest_range_entries_args {
 	struct list_head *list;
 	struct completion comp;
@@ -1337,6 +1636,9 @@ int scoutfs_net_setup(struct super_block *sb)
 	INIT_LIST_HEAD(&nti->to_send);
 	nti->next_id = 1;
 	INIT_DELAYED_WORK(&nti->server_work, scoutfs_net_server_func);
+	init_rwsem(&nti->ring_commit_rwsem);
+	init_llist_head(&nti->ring_commit_waiters);
+	INIT_WORK(&nti->ring_commit_work, scoutfs_net_ring_commit_func);
 	INIT_LIST_HEAD(&nti->active_socks);
 
 	sbi->net_info = nti;
@@ -1383,8 +1685,8 @@ void scoutfs_net_destroy(struct super_block *sb)
 		mutex_unlock(&nti->mutex);
 		drain_workqueue(nti->sock_wq);
 
-		/* wait for processing to finish and free rbufs */
-		flush_workqueue(nti->proc_wq);
+		/* wait for processing (and commits) to finish and free rbufs */
+		drain_workqueue(nti->proc_wq);
 
 		/* make sure client/server work isn't queued */
 		cancel_delayed_work_sync(&nti->server_work);

kmod/src/net.h

@@ -2,6 +2,7 @@
 #define _SCOUTFS_NET_H_
 
 struct scoutfs_key_buf;
+struct scoutfs_segment;
 
 int scoutfs_net_trade_time(struct super_block *sb);
 int scoutfs_net_alloc_inodes(struct super_block *sb);
@@ -9,6 +10,9 @@ int scoutfs_net_manifest_range_entries(struct super_block *sb,
 				       struct scoutfs_key_buf *start,
 				       struct scoutfs_key_buf *end,
 				       struct list_head *list);
+int scoutfs_net_alloc_segno(struct super_block *sb, u64 *segno);
+int scoutfs_net_record_segment(struct super_block *sb,
+			       struct scoutfs_segment *seg, u8 level);
 
 int scoutfs_net_setup(struct super_block *sb);
 void scoutfs_net_destroy(struct super_block *sb);

kmod/src/seg.c

@@ -245,26 +245,18 @@ static u64 segno_to_blkno(u64 blkno)
 	return blkno << (SCOUTFS_SEGMENT_SHIFT - SCOUTFS_BLOCK_SHIFT);
 }
 
-int scoutfs_seg_alloc(struct super_block *sb, struct scoutfs_segment **seg_ret)
+int scoutfs_seg_alloc(struct super_block *sb, u64 segno,
+		      struct scoutfs_segment **seg_ret)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct segment_cache *cac = sbi->segment_cache;
 	struct scoutfs_segment *existing;
 	struct scoutfs_segment *seg;
 	unsigned long flags;
-	u64 segno;
 	int ret;
 
-	*seg_ret = NULL;
-
-	ret = scoutfs_alloc_segno(sb, &segno);
-	if (ret)
-		goto out;
-
 	seg = alloc_seg(segno);
 	if (!seg) {
-		ret = scoutfs_alloc_free(sb, segno);
-		BUG_ON(ret); /* XXX could make pending when allocating */
 		ret = -ENOMEM;
 		goto out;
 	}
@@ -281,9 +273,9 @@ int scoutfs_seg_alloc(struct super_block *sb, struct scoutfs_segment **seg_ret)
 	if (existing)
 		scoutfs_seg_put(existing);
 
-	*seg_ret = seg;
 	ret = 0;
 out:
+	*seg_ret = seg;
 	return ret;
 
 }
@@ -632,6 +624,32 @@ int scoutfs_seg_manifest_del(struct super_block *sb,
 	return scoutfs_manifest_del(sb, &first, le64_to_cpu(sblk->seq), level);
 }
 
+/*
+ * Return an allocated manifest entry that describes the segment, returns
+ * NULL if it couldn't allocate.
+ */
+struct scoutfs_manifest_entry *
+scoutfs_seg_manifest_entry(struct super_block *sb,
+			   struct scoutfs_segment *seg, u8 level)
+{
+	struct scoutfs_segment_block *sblk = off_ptr(seg, 0);
+	struct scoutfs_segment_item *item;
+	struct scoutfs_key_buf first;
+	struct scoutfs_key_buf last;
+
+	item = pos_ptr(seg, 0);
+	scoutfs_key_init(&first, off_ptr(seg, le32_to_cpu(item->key_off)),
+				 le16_to_cpu(item->key_len));
+
+	item = pos_ptr(seg, le32_to_cpu(sblk->nr_items) - 1);
+	scoutfs_key_init(&last, off_ptr(seg, le32_to_cpu(item->key_off)),
+				 le16_to_cpu(item->key_len));
+
+	return scoutfs_manifest_alloc_entry(sb, &first, &last,
+					    le64_to_cpu(sblk->segno),
+					    le64_to_cpu(sblk->seq), level);
+}
+
 /*
  * We maintain an LRU of segments so that the shrinker can free the
  * oldest under memory pressure.  Segments are only present in the LRU

kmod/src/seg.h

@@ -19,7 +19,8 @@ int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int pos,
 void scoutfs_seg_get(struct scoutfs_segment *seg);
 void scoutfs_seg_put(struct scoutfs_segment *seg);
 
-int scoutfs_seg_alloc(struct super_block *sb, struct scoutfs_segment **seg_ret);
+int scoutfs_seg_alloc(struct super_block *sb, u64 segno,
+		      struct scoutfs_segment **seg_ret);
 int scoutfs_seg_free_segno(struct super_block *sb,
 			   struct scoutfs_segment *seg);
 bool scoutfs_seg_fits_single(u32 nr_items, u32 key_bytes, u32 val_bytes);
@@ -41,6 +42,10 @@ int scoutfs_seg_submit_write(struct super_block *sb,
 			     struct scoutfs_segment *seg,
 			     struct scoutfs_bio_completion *comp);
 
+struct scoutfs_manifest_entry *
+scoutfs_seg_manifest_entry(struct super_block *sb,
+			   struct scoutfs_segment *seg, u8 level);
+
 int scoutfs_seg_setup(struct super_block *sb);
 void scoutfs_seg_destroy(struct super_block *sb);
 

kmod/src/trans.c

@@ -24,10 +24,9 @@
 #include "item.h"
 #include "manifest.h"
 #include "seg.h"
-#include "alloc.h"
-#include "ring.h"
 #include "compact.h"
 #include "counters.h"
+#include "net.h"
 #include "scoutfs_trace.h"
 
 /*
@@ -82,7 +81,7 @@ void scoutfs_trans_write_func(struct work_struct *work)
 	struct super_block *sb = sbi->sb;
 	struct scoutfs_bio_completion comp;
 	struct scoutfs_segment *seg;
-	bool advance = false;
+	u64 segno;
 	int ret = 0;
 
 	scoutfs_bio_init_comp(&comp);
@@ -91,42 +90,25 @@ void scoutfs_trans_write_func(struct work_struct *work)
 	wait_event(sbi->trans_hold_wq,
 		   atomic_cmpxchg(&sbi->trans_holds, 0, -1) == 0);
 
-	trace_printk("items dirty %d manifest dirty %d alloc dirty %d\n",
-		     scoutfs_item_has_dirty(sb),
-		     scoutfs_manifest_has_dirty(sb),
-		     scoutfs_alloc_has_dirty(sb));
+	trace_printk("items dirty %d\n", scoutfs_item_has_dirty(sb));
 
-	/*
-	 * XXX this needs serious work to handle errors.
-	 */
-	while (scoutfs_item_has_dirty(sb)) {
-		seg = NULL;
-		ret = scoutfs_seg_alloc(sb, &seg) ?:
+	if (scoutfs_item_has_dirty(sb)) {
+		/*
+		 * XXX only straight pass through, we're not worrying
+		 * about leaking segnos nor duplicate manifest entries
+		 * on crashes between us and the server.
+		 */
+		ret = scoutfs_net_alloc_segno(sb, &segno) ?:
+		      scoutfs_seg_alloc(sb, segno, &seg) ?:
 		      scoutfs_item_dirty_seg(sb, seg) ?:
-		      scoutfs_manifest_lock(sb) ?:
-		      scoutfs_seg_manifest_add(sb, seg, 0) ?:
-		      scoutfs_manifest_unlock(sb) ?:
-		      scoutfs_seg_submit_write(sb, seg, &comp);
-		scoutfs_seg_put(seg);
+		      scoutfs_seg_submit_write(sb, seg, &comp) ?:
+		      scoutfs_bio_wait_comp(sb, &comp) ?:
+		      scoutfs_net_record_segment(sb, seg, 0);
 		if (ret)
 			goto out;
 
 		scoutfs_inc_counter(sb, trans_level0_seg_write);
 	}
-
-	if (scoutfs_manifest_has_dirty(sb) || scoutfs_alloc_has_dirty(sb)) {
-		ret = scoutfs_manifest_submit_write(sb, &comp) ?:
-		      scoutfs_alloc_submit_write(sb, &comp) ?:
-		      scoutfs_bio_wait_comp(sb, &comp) ?:
-		      scoutfs_write_dirty_super(sb);
-		if (ret)
-			goto out;
-
-		scoutfs_manifest_write_complete(sb);
-		scoutfs_alloc_write_complete(sb);
-		advance = true;
-	}
-
 out:
 	/* XXX this all needs serious work for dealing with errors */
 	WARN_ON_ONCE(ret);
@@ -135,8 +117,6 @@ out:
 	scoutfs_data_end_writeback(sb, ret);
 
 	spin_lock(&sbi->trans_write_lock);
-	if (advance)
-		scoutfs_advance_dirty_super(sb);
 	sbi->trans_write_count++;
 	sbi->trans_write_ret = ret;
 	spin_unlock(&sbi->trans_write_lock);
@@ -149,7 +129,6 @@ out:
 }
 
 struct write_attempt {
-	u64 seq;
 	u64 count;
 	int ret;
 };
@@ -161,9 +140,7 @@ static int write_attempted(struct scoutfs_sb_info *sbi,
 	int done = 1;
 
 	spin_lock(&sbi->trans_write_lock);
-	if (le64_to_cpu(sbi->super.hdr.seq) > attempt->seq)
-		attempt->ret = 0;
-	else if (sbi->trans_write_count > attempt->count)
+	if (sbi->trans_write_count > attempt->count)
 		attempt->ret = sbi->trans_write_ret;
 	else
 		done = 0;
@@ -178,10 +155,12 @@ static void queue_trans_work(struct scoutfs_sb_info *sbi)
 }
 
 /*
- * sync records the current dirty seq and write count and waits for
- * either to change.  If there's nothing to write or the write returned
- * an error then only the write count advances and sets the appropriate
- * return code.
+ * Wait for a trans commit to finish and return its error code.  There
+ * can already be one in flight that we end up waiting for the
+ * completion of.  This is safe because dirtying and trans commits are
+ * serialized.  There's no way that there could have been dirty data
+ * before the caller got here that wouldn't be covered by a commit
+ * that's in flight. 
  */
 int scoutfs_sync_fs(struct super_block *sb, int wait)
 {
@@ -197,7 +176,6 @@ int scoutfs_sync_fs(struct super_block *sb, int wait)
 	}
 
 	spin_lock(&sbi->trans_write_lock);
-	attempt.seq = le64_to_cpu(sbi->super.hdr.seq);
 	attempt.count = sbi->trans_write_count;
 	spin_unlock(&sbi->trans_write_lock);