Further isolate rings and compaction

Each mount was still loading the manifest and allocator rings and
starting compaction, even if they were coordinating segment reads
and writes with the server.

This moves ring and compaction setup and teardown from on mount and
unmount to as the server starts up and shuts down.  Now only the server
has the rings resident and is running compaction.

We had to null some of the super info fields so that we can repeatedly
load and destroy the ring indices over the lifetime of a mount.

We also have to be careful not to call between item transactions and
compaction.   We'll restore this functionality with the server in the
future.

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

kmod/src/alloc.c

@@ -363,6 +363,7 @@ int scoutfs_alloc_setup(struct super_block *sb)
 
 void scoutfs_alloc_destroy(struct super_block *sb)
 {
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	DECLARE_SEG_ALLOC(sb, sal);
 	struct pending_region *pend;
 	struct rb_node *node;
@@ -375,5 +376,6 @@ void scoutfs_alloc_destroy(struct super_block *sb)
 			kfree(pend);
 		}
 		kfree(sal);
+		sbi->seg_alloc = NULL;
 	}
 }

kmod/src/compact.c

@@ -22,7 +22,6 @@
 #include "cmp.h"
 #include "compact.h"
 #include "manifest.h"
-#include "trans.h"
 #include "counters.h"
 #include "alloc.h"
 #include "scoutfs_trace.h"
@@ -645,8 +644,13 @@ static void scoutfs_compact_func(struct work_struct *work)
 
 	ret = update_manifest(sb, &curs, &results);
 	if (ret == 0) {
+#if 0 /* XXX this is busted, fixing soon */
 		scoutfs_sync_fs(sb, 0);
+#endif
+
+#if 0 /* XXX where do we do this in shared? */
 		scoutfs_trans_wake_holders(sb);
+#endif
 		scoutfs_compact_kick(sb);
 	}
 out:
@@ -695,10 +699,12 @@ int scoutfs_compact_setup(struct super_block *sb)
  */
 void scoutfs_compact_destroy(struct super_block *sb)
 {
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	DECLARE_COMPACT_INFO(sb, ci);
 
 	if (ci) {
 		flush_work(&ci->work);
 		destroy_workqueue(ci->workq);
+		sbi->compact_info = NULL;
 	}
 }

kmod/src/manifest.c

@@ -1004,5 +1004,6 @@ void scoutfs_manifest_destroy(struct super_block *sb)
 		for (i = 0; i < ARRAY_SIZE(mani->compact_keys); i++)
 			scoutfs_key_free(sb, mani->compact_keys[i]);
 		kfree(mani);
+		sbi->manifest = NULL;
 	}
 }

kmod/src/net.c

@@ -27,6 +27,7 @@
 #include "bio.h"
 #include "alloc.h"
 #include "seg.h"
+#include "compact.h"
 #include "scoutfs_trace.h"
 
 /*
@@ -73,6 +74,7 @@ struct net_info {
 	/* server listens and processes requests */
 	struct delayed_work server_work;
 	struct sock_info *listening_sinf;
+	bool server_loaded;
 
 	/* server commits ring changes while processing requests */
 	struct rw_semaphore ring_commit_rwsem;
@@ -696,6 +698,13 @@ static int process_reply(struct net_info *nti, struct recv_buf *rbuf)
 	return func(sb, rbuf->nh->data, ret, arg);
 }
 
+static void destroy_server_state(struct super_block *sb)
+{
+	scoutfs_alloc_destroy(sb);
+	scoutfs_manifest_destroy(sb);
+	scoutfs_compact_destroy(sb);
+}
+
 /*
  * Process each received message in its own non-reentrant work so we get
  * concurrent request processing.
@@ -704,7 +713,35 @@ static void scoutfs_net_proc_func(struct work_struct *work)
 {
 	struct recv_buf *rbuf = container_of(work, struct recv_buf, proc_work);
 	struct net_info *nti = rbuf->nti;
-	int ret;
+	struct super_block *sb = nti->sb;
+	int ret = 0;
+
+	/*
+	 * This is the first blocking context we have once all the
+	 * server locking and networking is set up so we bring up the
+	 * rest of the server state the first time we get here.
+	 */
+	while (!nti->server_loaded) {
+		mutex_lock(&nti->mutex);
+		if (!nti->server_loaded) {
+			ret = scoutfs_read_supers(sb) ?:
+			      scoutfs_manifest_setup(sb) ?:
+			      scoutfs_alloc_setup(sb) ?:
+			      scoutfs_compact_setup(sb);
+			if (ret == 0) {
+				scoutfs_advance_dirty_super(sb);
+				nti->server_loaded = true;
+			} else {
+				destroy_server_state(sb);
+			}
+		}
+		mutex_unlock(&nti->mutex);
+		if (ret) {
+			trace_printk("server setup failed %d\n", ret);
+			queue_sock_work(rbuf->sinf, &rbuf->sinf->shutdown_work);
+			return;
+		}
+	}
 
 	if (rbuf->nh->status == SCOUTFS_NET_STATUS_REQUEST)
 		ret = process_request(nti, rbuf);
@@ -990,8 +1027,13 @@ static void scoutfs_net_shutdown_func(struct work_struct *work)
 	mutex_lock(&nti->mutex);
 
 	if (sinf == nti->listening_sinf) {
-		/* clear addr lvb and try to reacquire lock and listen */
 		nti->listening_sinf = NULL;
+
+		/* shutdown the server, processing won't leave rings dirty */
+		destroy_server_state(sb);
+		nti->server_loaded = false;
+
+		/* clear addr lvb and try to reacquire lock and listen */
 		memset(&sinf->addr, 0, sizeof(sinf->addr));
 		lock_addr_lvb(sb, SCOUTFS_LOCK_MODE_WRITE, &sinf->addr);
 		scoutfs_unlock_range(sb, &sinf->listen_lck);

kmod/src/super.c

@@ -133,7 +133,13 @@ int scoutfs_write_dirty_super(struct super_block *sb)
 	return ret;
 }
 
-static int read_supers(struct super_block *sb)
+/*
+ * Read the pair of super blocks and store the most recent one in the sb
+ * info.  Clients reference but don't modify the super.  The server has
+ * to re-read the super every time it comes up so that it can work from
+ * the most recent persistent state.
+ */
+int scoutfs_read_supers(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_super_block *super;
@@ -211,14 +217,11 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 		return -ENOMEM;
 
 	ret = scoutfs_setup_counters(sb) ?:
-	      read_supers(sb) ?:
+	      scoutfs_read_supers(sb) ?:
 	      scoutfs_seg_setup(sb) ?:
-	      scoutfs_manifest_setup(sb) ?:
 	      scoutfs_item_setup(sb) ?:
 	      scoutfs_inode_setup(sb) ?:
 	      scoutfs_data_setup(sb) ?:
-	      scoutfs_alloc_setup(sb) ?:
-	      scoutfs_compact_setup(sb) ?:
 	      scoutfs_setup_trans(sb) ?:
 	      scoutfs_lock_setup(sb) ?:
 	      scoutfs_net_setup(sb);
@@ -227,8 +230,6 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 
 	scoutfs_net_trade_time(sb);
 
-	scoutfs_advance_dirty_super(sb);
-
 	inode = scoutfs_iget(sb, SCOUTFS_ROOT_INO);
 	if (IS_ERR(inode))
 		return PTR_ERR(inode);

kmod/src/super.h

@@ -56,6 +56,7 @@ static inline struct scoutfs_sb_info *SCOUTFS_SB(struct super_block *sb)
 	return sb->s_fs_info;
 }
 
+int scoutfs_read_supers(struct super_block *sb);
 void scoutfs_advance_dirty_super(struct super_block *sb);
 int scoutfs_write_dirty_super(struct super_block *sb);
 

kmod/src/trans.c

@@ -24,7 +24,6 @@
 #include "item.h"
 #include "manifest.h"
 #include "seg.h"
-#include "compact.h"
 #include "counters.h"
 #include "net.h"
 #include "scoutfs_trace.h"
@@ -239,11 +238,13 @@ static bool hold_acquired(struct super_block *sb)
 		if (holds < 0)
 			return false;
 
+#if 0 /* XXX where will we do this in the shared universe? */
 		/* only hold when there's no level 0 segments, XXX for now */
 		if (scoutfs_manifest_level_count(sb, 0) > 0) {
 			scoutfs_compact_kick(sb);
 			return false;
 		}
+#endif
 
 		/* see if we all would fill the segment */
 		with_us = holds + 1;