scoutfs: add bidirectional network messages

The client and server networking code was a bit too rudimentary.

The existing code only had support for the client synchronously and
actively sending requests that the server could only passively respond
to.  We're going to need the server to be able to send requests to
connected clients and it can't block waiting for responses from each
one.

This refactors sending and receiving in both the client and server code
into shared networking code.  It's built around a connection struct that
then holds the message state.  Both peers on the connection can send
requests and send responses.

The existing code only retransmitted requests down newly established
connections.  Requests could be processed twice.

This adds robust reliability guarantees.  Requests are resend until
their response is received.  Requests are only processed once by a given
peer, regardless of the connection's transport socket.  Responses are
reiably resent until acknowledged.

This only adds the new refactored code and disables the old unused code
to keep the diff foot print minmal.  A following commit will remove all
the unused code.

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

kmod/src/Makefile

@@ -7,7 +7,7 @@ CFLAGS_scoutfs_trace.o = -I$(src) # define_trace.h double include
  
 scoutfs-y += bio.o block.o btree.o client.o compact.o counters.o data.o dir.o \
 	     export.o extents.o file.o inode.o ioctl.o item.o lock.o \
-	     manifest.o msg.o options.o per_task.o seg.o server.o \
+	     manifest.o msg.o net.o options.o per_task.o seg.o server.o \
 	     scoutfs_trace.o sock.o sort_priv.o super.o sysfs.o trans.o \
 	     triggers.o xattr.o
 

kmod/src/client.c

@@ -32,24 +32,18 @@
 #include "msg.h"
 #include "server.h"
 #include "client.h"
-#include "sock.h"
+#include "net.h"
 #include "endian_swap.h"
 
 /*
- * Client callers block sending requests to the server.  Senders connect
- * and send down the socket in their blocked context under a mutex.
- * Once a socket is connected recv work is fired up.  Destroying a
- * socket shuts down the socket and cancels the work.
- *
- * Clients are responsible for resending their requests after
- * reconnecting to a new socket.  These new socket connections might be
- * connecting to the same server.  The message sending and processing
- * paths are responsible for dealing with duplicate requests.
+ * The client always maintains a connection to the server.  It reads the
+ * super to get the address it should try and connect to.
  */
 
 #define SIN_FMT		"%pIS:%u"
 #define SIN_ARG(sin)	sin, be16_to_cpu((sin)->sin_port)
 
+#if 0
 /*
  * Have a pretty aggressive keepalive timeout of around 10 seconds.  The
  * TCP keepalives are being processed out of task context so they should
@@ -59,6 +53,7 @@
 #define KEEPCNT			3
 #define KEEPIDLE		7
 #define KEEPINTVL		1
+#endif
 
 
 /*
@@ -72,30 +67,17 @@
 struct client_info {
 	struct super_block *sb;
 
-	/* spinlock protects quick critical sections between send,recv,umount */
-	spinlock_t recv_lock;
-	struct rb_root sender_root;
+	struct scoutfs_net_connection *conn;
+	atomic_t shutting_down;
 
-	/* the sock mutex serializes connecting and sending */
-	struct mutex send_mutex;
-	bool recv_shutdown;
-	u64 next_id;
-	u64 sock_gen;
-	struct socket *sock;
-	struct sockaddr_in peername;
-	struct sockaddr_in sockname;
-
-	/* blocked senders sit on a waitq that's woken for resends */
-	wait_queue_head_t waitq;
+	struct workqueue_struct *workq;
+	struct delayed_work connect_dwork;
 
 	/* connection timeouts are tracked across attempts */
 	unsigned long conn_retry_ms;
-	unsigned long conn_retry_limit_j;
-
-	struct workqueue_struct *recv_wq;
-	struct work_struct recv_work;
 };
 
+#if 0
 struct waiting_sender {
 	struct rb_node node;
 	struct task_struct *task;
@@ -213,14 +195,20 @@ static void scoutfs_client_recv_func(struct work_struct *work)
 
 	kfree(rx);
 }
+#endif
 
-static void reset_connect_timeouts(struct client_info *client)
+static void reset_connect_timeout(struct client_info *client)
 {
 	client->conn_retry_ms = CONN_RETRY_MIN_MS;
-	client->conn_retry_limit_j = jiffies + CONN_RETRY_LIMIT_J;
 }
 
+static void grow_connect_timeout(struct client_info *client)
+{
+	client->conn_retry_ms = min(client->conn_retry_ms * 2,
+				    CONN_RETRY_MAX_MS);
+}
 
+#if 0
 /*
  * Clients who try to send and don't see a connected socket call here to
  * connect to the server.  They get the server address and try to
@@ -527,6 +515,7 @@ static int client_request(struct client_info *client, int type, void *data,
 
 	return ret;
 }
+#endif
 
 /*
  * Ask for a new run of allocated inode numbers.  The server can return
@@ -540,8 +529,10 @@ int scoutfs_client_alloc_inodes(struct super_block *sb, u64 count,
 	__le64 lecount = cpu_to_le64(count);
 	int ret;
 
-	ret = client_request(client, SCOUTFS_NET_ALLOC_INODES,
-			     &lecount, sizeof(lecount), &ial, sizeof(ial));
+	ret = scoutfs_net_sync_request(sb, client->conn,
+				       SCOUTFS_NET_CMD_ALLOC_INODES,
+				       &lecount, sizeof(lecount),
+				       &ial, sizeof(ial));
 	if (ret == 0) {
 		*ino = le64_to_cpu(ial.ino);
 		*nr = le64_to_cpu(ial.nr);
@@ -568,8 +559,10 @@ int scoutfs_client_alloc_extent(struct super_block *sb, u64 blocks, u64 *start,
 	struct scoutfs_net_extent nex;
 	int ret;
 
-	ret = client_request(client, SCOUTFS_NET_ALLOC_EXTENT,
-			     &leblocks, sizeof(leblocks), &nex, sizeof(nex));
+	ret = scoutfs_net_sync_request(sb, client->conn,
+				       SCOUTFS_NET_CMD_ALLOC_EXTENT,
+				       &leblocks, sizeof(leblocks),
+				       &nex, sizeof(nex));
 	if (ret == 0) {
 		if (nex.len == 0) {
 			ret = -ENOSPC;
@@ -590,8 +583,9 @@ int scoutfs_client_free_extents(struct super_block *sb,
 
 	bytes = SCOUTFS_NET_EXTENT_LIST_BYTES(le64_to_cpu(nexl->nr));
 
-	return client_request(client, SCOUTFS_NET_FREE_EXTENTS,
-			      nexl, bytes, NULL, 0);
+	return scoutfs_net_sync_request(sb, client->conn,
+					SCOUTFS_NET_CMD_FREE_EXTENTS,
+					nexl, bytes, NULL, 0);
 }
 
 int scoutfs_client_alloc_segno(struct super_block *sb, u64 *segno)
@@ -600,8 +594,9 @@ int scoutfs_client_alloc_segno(struct super_block *sb, u64 *segno)
 	__le64 lesegno;
 	int ret;
 
-	ret = client_request(client, SCOUTFS_NET_ALLOC_SEGNO, NULL, 0,
-			     &lesegno, sizeof(lesegno));
+	ret = scoutfs_net_sync_request(sb, client->conn,
+				       SCOUTFS_NET_CMD_ALLOC_SEGNO,
+				       NULL, 0, &lesegno, sizeof(lesegno));
 	if (ret == 0) {
 		if (lesegno == 0)
 			ret = -ENOSPC;
@@ -622,8 +617,9 @@ int scoutfs_client_record_segment(struct super_block *sb,
 	scoutfs_seg_init_ment(&ment, level, seg);
 	scoutfs_init_ment_to_net(&net_ment, &ment);
 
-	return client_request(client, SCOUTFS_NET_RECORD_SEGMENT, &net_ment,
-			      sizeof(net_ment), NULL, 0);
+	return scoutfs_net_sync_request(sb, client->conn,
+					SCOUTFS_NET_CMD_RECORD_SEGMENT,
+					&net_ment, sizeof(net_ment), NULL, 0);
 }
 
 int scoutfs_client_advance_seq(struct super_block *sb, u64 *seq)
@@ -633,8 +629,10 @@ int scoutfs_client_advance_seq(struct super_block *sb, u64 *seq)
 	__le64 after;
 	int ret;
 
-	ret = client_request(client, SCOUTFS_NET_ADVANCE_SEQ,
-			     &before, sizeof(before), &after, sizeof(after));
+	ret = scoutfs_net_sync_request(sb, client->conn,
+				       SCOUTFS_NET_CMD_ADVANCE_SEQ,
+				       &before, sizeof(before),
+				       &after, sizeof(after));
 	if (ret == 0)
 		*seq = le64_to_cpu(after);
 
@@ -647,8 +645,9 @@ int scoutfs_client_get_last_seq(struct super_block *sb, u64 *seq)
 	__le64 last_seq;
 	int ret;
 
-	ret = client_request(client, SCOUTFS_NET_GET_LAST_SEQ,
-			     NULL, 0, &last_seq, sizeof(last_seq));
+	ret = scoutfs_net_sync_request(sb, client->conn,
+				       SCOUTFS_NET_CMD_GET_LAST_SEQ,
+				       NULL, 0, &last_seq, sizeof(last_seq));
 	if (ret == 0)
 		*seq = le64_to_cpu(last_seq);
 
@@ -660,8 +659,10 @@ int scoutfs_client_get_manifest_root(struct super_block *sb,
 {
 	struct client_info *client = SCOUTFS_SB(sb)->client_info;
 
-	return client_request(client, SCOUTFS_NET_GET_MANIFEST_ROOT,
-			      NULL, 0, root, sizeof(struct scoutfs_btree_root));
+	return scoutfs_net_sync_request(sb, client->conn,
+					SCOUTFS_NET_CMD_GET_MANIFEST_ROOT,
+					NULL, 0, root,
+					sizeof(struct scoutfs_btree_root));
 }
 
 int scoutfs_client_statfs(struct super_block *sb,
@@ -669,53 +670,147 @@ int scoutfs_client_statfs(struct super_block *sb,
 {
 	struct client_info *client = SCOUTFS_SB(sb)->client_info;
 
-	return client_request(client, SCOUTFS_NET_STATFS, NULL, 0, nstatfs,
-			      sizeof(struct scoutfs_net_statfs));
+	return scoutfs_net_sync_request(sb, client->conn,
+					SCOUTFS_NET_CMD_STATFS, NULL, 0,
+					nstatfs,
+					sizeof(struct scoutfs_net_statfs));
+}
+
+/*
+ * Attempt to connect to the listening address that the server wrote in
+ * the super block.  We keep trying indefinitely with an increasing
+ * delay if we fail to either read the address or connect to it.
+ *
+ * We're careful to only ever have one connection attempt in flight.  We
+ * only queue this work on mount, on error, or from the connection
+ * callback.
+ */
+static void scoutfs_client_connect_worker(struct work_struct *work)
+{
+	struct client_info *client = container_of(work, struct client_info,
+						  connect_dwork.work);
+	struct super_block *sb = client->sb;
+	struct scoutfs_super_block super;
+	struct sockaddr_in sin;
+	int ret;
+
+	ret = scoutfs_read_super(sb, &super);
+	if (ret)
+		goto out;
+
+	if (super.server_addr.addr == cpu_to_le32(INADDR_ANY)) {
+		ret = -EADDRNOTAVAIL;
+		goto out;
+	}
+
+	memset(&sin, 0, sizeof(sin));
+	sin.sin_family = AF_INET;
+	sin.sin_addr.s_addr = le32_to_be32(super.server_addr.addr);
+	sin.sin_port = le16_to_be16(super.server_addr.port);
+
+	scoutfs_net_connect(sb, client->conn, &sin, client->conn_retry_ms);
+	ret = 0;
+out:
+	if (ret && !atomic_read(&client->shutting_down)) {
+		queue_delayed_work(client->workq, &client->connect_dwork,
+				   msecs_to_jiffies(client->conn_retry_ms));
+		grow_connect_timeout(client);
+	}
+}
+
+static void client_notify_up(struct super_block *sb,
+			     struct scoutfs_net_connection *conn)
+{
+	struct client_info *client = SCOUTFS_SB(sb)->client_info;
+
+	reset_connect_timeout(client);
+}
+
+/*
+ * Called when either a connect attempt or established connection times
+ * out and fails.
+ */
+static void client_notify_down(struct super_block *sb,
+			       struct scoutfs_net_connection *conn)
+{
+	struct client_info *client = SCOUTFS_SB(sb)->client_info;
+
+	if (!atomic_read(&client->shutting_down)) {
+		queue_delayed_work(client->workq, &client->connect_dwork,
+				   msecs_to_jiffies(client->conn_retry_ms));
+		grow_connect_timeout(client);
+	}
 }
 
 int scoutfs_client_setup(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct client_info *client;
+	int ret;
 
 	client = kzalloc(sizeof(struct client_info), GFP_KERNEL);
-	if (!client)
-		return -ENOMEM;
+	if (!client) {
+		ret = -ENOMEM;
+		goto out;
+	}
+	sbi->client_info = client;
 
 	client->sb = sb;
-	spin_lock_init(&client->recv_lock);
-	client->sender_root = RB_ROOT;
-	mutex_init(&client->send_mutex);
-	init_waitqueue_head(&client->waitq);
-	INIT_WORK(&client->recv_work, scoutfs_client_recv_func);
-	reset_connect_timeouts(client);
+	atomic_set(&client->shutting_down, 0);
+	INIT_DELAYED_WORK(&client->connect_dwork,
+			  scoutfs_client_connect_worker);
 
-	client->recv_wq = alloc_workqueue("scoutfs_client_recv", WQ_UNBOUND, 1);
-	if (!client->recv_wq) {
-		kfree(client);
-		return -ENOMEM;
+	/* client doesn't process any incoming requests yet */
+	client->conn = scoutfs_net_alloc_conn(sb, client_notify_up,
+					      client_notify_down, NULL,
+					      "client");
+	if (!client->conn) {
+		ret = -ENOMEM;
+		goto out;
 	}
 
-	sbi->client_info = client;
-	return 0;
+	client->workq = alloc_workqueue("scoutfs_client_workq", WQ_UNBOUND, 1);
+	if (!client->workq) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
+	reset_connect_timeout(client);
+	/* delay initial connect to give a local server some time to setup */
+	queue_delayed_work(client->workq, &client->connect_dwork,
+			   msecs_to_jiffies(client->conn_retry_ms));
+	ret = 0;
+
+out:
+	if (ret)
+		scoutfs_client_destroy(sb);
+	return ret;
 }
 
 /*
- * There must be no more callers to the client send functions by the
- * time we get here.  We just need to free the socket if it's
- * still sitting around.
+ * There must be no more callers to the client request functions by the
+ * time we get here.
  */
 void scoutfs_client_destroy(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct client_info *client = SCOUTFS_SB(sb)->client_info;
+	struct scoutfs_net_connection *conn;
 
 	if (client) {
-		shutdown_sock_sync(client);
+		/* stop notify_down from queueing connect work */
+		atomic_set(&client->shutting_down, 1);
 
-		cancel_work_sync(&client->recv_work);
-		destroy_workqueue(client->recv_wq);
+		/* make sure worker isn't using the conn */
+		cancel_delayed_work_sync(&client->connect_dwork);
 
+		/* make racing conn use explode */
+		conn = client->conn;
+		client->conn = NULL;
+		scoutfs_net_free_conn(sb, conn);
+
+		if (client->workq)
+			destroy_workqueue(client->workq);
 		kfree(client);
 		sbi->client_info = NULL;
 	}

kmod/src/counters.h

@@ -100,6 +100,18 @@
 	EXPAND_COUNTER(manifest_compact_migrate)		\
 	EXPAND_COUNTER(manifest_hard_stale_error)		\
 	EXPAND_COUNTER(manifest_read_excluded_key)		\
+	EXPAND_COUNTER(net_dropped_ack)				\
+	EXPAND_COUNTER(net_dropped_response)			\
+	EXPAND_COUNTER(net_dropped_request)			\
+	EXPAND_COUNTER(net_send_bytes)				\
+	EXPAND_COUNTER(net_send_error)				\
+	EXPAND_COUNTER(net_send_messages)			\
+	EXPAND_COUNTER(net_recv_bytes)				\
+	EXPAND_COUNTER(net_recv_error)				\
+	EXPAND_COUNTER(net_recv_invalid_message)		\
+	EXPAND_COUNTER(net_recv_messages)			\
+	EXPAND_COUNTER(net_unknown_message)			\
+	EXPAND_COUNTER(net_unknown_request)			\
 	EXPAND_COUNTER(seg_alloc)				\
 	EXPAND_COUNTER(seg_free)				\
 	EXPAND_COUNTER(seg_shrink)				\

kmod/src/format.h

@@ -353,8 +353,6 @@ struct scoutfs_inet_addr {
 	__le16 port;
 } __packed;
 
-#define SCOUTFS_DEFAULT_PORT 12345
-
 struct scoutfs_super_block {
 	struct scoutfs_block_header hdr;
 	__le64 id;
@@ -506,6 +504,10 @@ struct scoutfs_lock_name {
  * messages over the wire.
  */
 
+/*
+ * Greetings verify identity of communicating nodes.  The sender
+ * sends their credentials and the receiver verifies them.
+ */
 struct scoutfs_net_greeting {
 	__le64 fsid;
 	__le64 format_hash;
@@ -517,15 +519,70 @@ struct scoutfs_net_greeting {
  * type is strictly redundant in the reply because the id will find the
  * send but we include it in both packets to make it easier to observe
  * replies without having the id from their previous request.
+ *
+ * Error is only set to a translated errno on response messages and
+ * data_len will be 0.
  */
 struct scoutfs_net_header {
 	__le64 id;
 	__le16 data_len;
-	__u8 type;
-	__u8 status;
+	__u8 msg;
+	__u8 cmd;
+	__u8 error;
 	__u8 data[0];
 } __packed;
 
+/*
+ * Greetings are the first messages sent down every newly established
+ * socket on the connection.  Every other message gets a unique
+ * increasing id over the life time of the connection.
+ */
+#define SCOUTFS_NET_ID_GREETING 1
+
+enum {
+	SCOUTFS_NET_MSG_REQUEST = 0,
+	SCOUTFS_NET_MSG_RESPONSE,
+	SCOUTFS_NET_MSG_ACK,
+	SCOUTFS_NET_MSG_UNKNOWN,
+};
+
+enum {
+	SCOUTFS_NET_CMD_GREETING = 0,
+	SCOUTFS_NET_CMD_ALLOC_INODES,
+	SCOUTFS_NET_CMD_ALLOC_EXTENT,
+	SCOUTFS_NET_CMD_FREE_EXTENTS,
+	SCOUTFS_NET_CMD_ALLOC_SEGNO,
+	SCOUTFS_NET_CMD_RECORD_SEGMENT,
+	SCOUTFS_NET_CMD_ADVANCE_SEQ,
+	SCOUTFS_NET_CMD_GET_LAST_SEQ,
+	SCOUTFS_NET_CMD_GET_MANIFEST_ROOT,
+	SCOUTFS_NET_CMD_STATFS,
+	SCOUTFS_NET_CMD_UNKNOWN,
+};
+
+/*
+ * Define a macro to evaluate another macro for each of the errnos we
+ * translate over the wire.  This lets us keep our enum in sync with the
+ * mapping arrays to and from host errnos.
+ */
+#define EXPAND_EACH_NET_ERRNO		\
+	EXPAND_NET_ERRNO(ENOENT)	\
+	EXPAND_NET_ERRNO(ENOMEM)	\
+	EXPAND_NET_ERRNO(EIO)		\
+	EXPAND_NET_ERRNO(ENOSPC)	\
+	EXPAND_NET_ERRNO(EINVAL)
+
+#undef EXPAND_NET_ERRNO
+#define EXPAND_NET_ERRNO(which) SCOUTFS_NET_ERR_##which,
+enum {
+	SCOUTFS_NET_ERR_NONE = 0,
+	EXPAND_EACH_NET_ERRNO
+	SCOUTFS_NET_ERR_UNKNOWN,
+};
+
+/* arbitrarily chosen to be safely less than mss and allow 1k with header */
+#define SCOUTFS_NET_MAX_DATA_LEN 1100
+
 /*
  * When there's no more free inodes this will be sent with ino = ~0 and
  * nr = 0.
@@ -586,26 +643,6 @@ struct scoutfs_net_extent_list {
 #define SCOUTFS_COMPACTION_MAX_UPDATE \
 	(2 * (SCOUTFS_COMPACTION_MAX_INPUT + SCOUTFS_COMPACTION_SLOP))
 
-enum {
-	SCOUTFS_NET_ALLOC_INODES = 0,
-	SCOUTFS_NET_ALLOC_EXTENT,
-	SCOUTFS_NET_FREE_EXTENTS,
-	SCOUTFS_NET_ALLOC_SEGNO,
-	SCOUTFS_NET_RECORD_SEGMENT,
-	SCOUTFS_NET_ADVANCE_SEQ,
-	SCOUTFS_NET_GET_LAST_SEQ,
-	SCOUTFS_NET_GET_MANIFEST_ROOT,
-	SCOUTFS_NET_STATFS,
-	SCOUTFS_NET_UNKNOWN,
-};
-
-enum {
-	SCOUTFS_NET_STATUS_REQUEST = 0,
-	SCOUTFS_NET_STATUS_SUCCESS,
-	SCOUTFS_NET_STATUS_ERROR,
-	SCOUTFS_NET_STATUS_UNKNOWN,
-};
-
 /*
  * Scoutfs file handle structure - this can be copied out to userspace
  * via open by handle or put on the wire from NFS.

kmod/src/net.h

@@ -0,0 +1,61 @@
+#ifndef _SCOUTFS_NET_H_
+#define _SCOUTFS_NET_H_
+
+#include <linux/in.h>
+
+#define SIN_FMT		"%pIS:%u"
+#define SIN_ARG(sin)	sin, be16_to_cpu((sin)->sin_port)
+
+struct scoutfs_net_connection;
+
+/* These are called in their own blocking context */
+typedef int (*scoutfs_net_request_t)(struct super_block *sb,
+				     struct scoutfs_net_connection *conn,
+				     u8 cmd, u64 id, void *arg, u16 arg_len);
+
+/* These are called with a spinlock held, funcs must be fast and nonblocking */
+typedef int (*scoutfs_net_response_t)(struct super_block *sb,
+				      struct scoutfs_net_connection *conn,
+				      void *resp, unsigned int resp_len,
+				      int error, void *data);
+
+typedef void (*scoutfs_net_notify_t)(struct super_block *sb,
+				     struct scoutfs_net_connection *conn);
+
+struct scoutfs_net_connection *
+scoutfs_net_alloc_conn(struct super_block *sb,
+		       scoutfs_net_notify_t notify_up,
+		       scoutfs_net_notify_t notify_down,
+		       scoutfs_net_request_t *req_funcs, char *name_suffix);
+void scoutfs_net_connect(struct super_block *sb,
+			 struct scoutfs_net_connection *conn,
+			 struct sockaddr_in *sin, unsigned long timeout_ms);
+int scoutfs_net_bind(struct super_block *sb,
+		     struct scoutfs_net_connection *conn,
+		     struct sockaddr_in *sin);
+void scoutfs_net_listen(struct super_block *sb,
+			struct scoutfs_net_connection *conn);
+int scoutfs_net_submit_request(struct super_block *sb,
+			       struct scoutfs_net_connection *conn,
+			       u8 cmd, void *arg, u16 arg_len,
+			       scoutfs_net_response_t resp_func,
+			       void *resp_data, u64 *id_ret);
+void scoutfs_net_cancel_request(struct super_block *sb,
+				struct scoutfs_net_connection *conn,
+				u8 cmd, u64 id);
+int scoutfs_net_sync_request(struct super_block *sb,
+			     struct scoutfs_net_connection *conn,
+			     u8 cmd, void *arg, unsigned arg_len,
+			     void *resp, size_t resp_len);
+int scoutfs_net_response(struct super_block *sb,
+			 struct scoutfs_net_connection *conn,
+			 u8 cmd, u64 id, int error, void *resp, u16 resp_len);
+void scoutfs_net_shutdown(struct super_block *sb,
+			  struct scoutfs_net_connection *conn);
+void scoutfs_net_free_conn(struct super_block *sb,
+			   struct scoutfs_net_connection *conn);
+
+int scoutfs_net_setup(struct super_block *sb);
+void scoutfs_net_destroy(struct super_block *sb);
+
+#endif

kmod/src/scoutfs_trace.h

@@ -1658,25 +1658,13 @@ DECLARE_EVENT_CLASS(scoutfs_net_class,
 		  snh_trace_args(nh))
 );
 
-DEFINE_EVENT(scoutfs_net_class, scoutfs_client_send_request,
+DEFINE_EVENT(scoutfs_net_class, scoutfs_net_send_message,
         TP_PROTO(struct super_block *sb, struct sockaddr_in *name,
 		 struct sockaddr_in *peer, struct scoutfs_net_header *nh),
         TP_ARGS(sb, name, peer, nh)
 );
 
-DEFINE_EVENT(scoutfs_net_class, scoutfs_server_recv_request,
-        TP_PROTO(struct super_block *sb, struct sockaddr_in *name,
-		 struct sockaddr_in *peer, struct scoutfs_net_header *nh),
-        TP_ARGS(sb, name, peer, nh)
-);
-
-DEFINE_EVENT(scoutfs_net_class, scoutfs_server_send_reply,
-        TP_PROTO(struct super_block *sb, struct sockaddr_in *name,
-		 struct sockaddr_in *peer, struct scoutfs_net_header *nh),
-        TP_ARGS(sb, name, peer, nh)
-);
-
-DEFINE_EVENT(scoutfs_net_class, scoutfs_client_recv_reply,
+DEFINE_EVENT(scoutfs_net_class, scoutfs_net_recv_message,
         TP_PROTO(struct super_block *sb, struct sockaddr_in *name,
 		 struct sockaddr_in *peer, struct scoutfs_net_header *nh),
         TP_ARGS(sb, name, peer, nh)
@@ -1706,11 +1694,51 @@ DEFINE_EVENT(scoutfs_work_class, scoutfs_server_commit_work_exit,
         TP_PROTO(struct super_block *sb, u64 data, int ret),
         TP_ARGS(sb, data, ret)
 );
-DEFINE_EVENT(scoutfs_work_class, scoutfs_server_recv_work_enter,
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_proc_work_enter,
         TP_PROTO(struct super_block *sb, u64 data, int ret),
         TP_ARGS(sb, data, ret)
 );
-DEFINE_EVENT(scoutfs_work_class, scoutfs_server_recv_work_exit,
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_proc_work_exit,
+        TP_PROTO(struct super_block *sb, u64 data, int ret),
+        TP_ARGS(sb, data, ret)
+);
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_listen_work_enter,
+        TP_PROTO(struct super_block *sb, u64 data, int ret),
+        TP_ARGS(sb, data, ret)
+);
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_listen_work_exit,
+        TP_PROTO(struct super_block *sb, u64 data, int ret),
+        TP_ARGS(sb, data, ret)
+);
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_connect_work_enter,
+        TP_PROTO(struct super_block *sb, u64 data, int ret),
+        TP_ARGS(sb, data, ret)
+);
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_connect_work_exit,
+        TP_PROTO(struct super_block *sb, u64 data, int ret),
+        TP_ARGS(sb, data, ret)
+);
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_shutdown_work_enter,
+        TP_PROTO(struct super_block *sb, u64 data, int ret),
+        TP_ARGS(sb, data, ret)
+);
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_shutdown_work_exit,
+        TP_PROTO(struct super_block *sb, u64 data, int ret),
+        TP_ARGS(sb, data, ret)
+);
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_send_work_enter,
+        TP_PROTO(struct super_block *sb, u64 data, int ret),
+        TP_ARGS(sb, data, ret)
+);
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_send_work_exit,
+        TP_PROTO(struct super_block *sb, u64 data, int ret),
+        TP_ARGS(sb, data, ret)
+);
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_recv_work_enter,
+        TP_PROTO(struct super_block *sb, u64 data, int ret),
+        TP_ARGS(sb, data, ret)
+);
+DEFINE_EVENT(scoutfs_work_class, scoutfs_net_recv_work_exit,
         TP_PROTO(struct super_block *sb, u64 data, int ret),
         TP_ARGS(sb, data, ret)
 );

kmod/src/server.c

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2017 Versity Software, Inc.  All rights reserved.
+ * Copyright (C) 2018 Versity Software, Inc.  All rights reserved.
  *
  * This program is free software; you can redistribute it and/or
  * modify it under the terms of the GNU General Public
@@ -32,9 +32,20 @@
 #include "msg.h"
 #include "client.h"
 #include "server.h"
-#include "sock.h"
+#include "net.h"
 #include "endian_swap.h"
 
+/*
+ * Every active mount can act as the server that listens on a net
+ * connection and accepts connections from all the other mounts acting
+ * as clients.
+ *
+ * It queues long-lived work that blocks trying to acquire a lock.  If
+ * it acquires the lock it listens on a socket and serves requests.  If
+ * it sees errors it shuts down the server in the hopes that another
+ * mount will have less trouble.
+ */
+
 #define SIN_FMT		"%pIS:%u"
 #define SIN_ARG(sin)	sin, be16_to_cpu((sin)->sin_port)
 
@@ -43,19 +54,14 @@ struct server_info {
 
 	struct workqueue_struct *wq;
 	struct delayed_work dwork;
-
-	struct mutex mutex;
-	bool shutting_down;
+	struct completion shutdown_comp;
 	bool bind_warned;
-	struct task_struct *listen_task;
-	struct socket *listen_sock;
 
 	/* request processing coordinates committing manifest and alloc */
 	struct rw_semaphore commit_rwsem;
 	struct llist_head commit_waiters;
 	struct work_struct commit_work;
 
-
 	/* adding new segments can have to wait for compaction */
 	wait_queue_head_t compaction_waitq;
 
@@ -72,6 +78,10 @@ struct server_info {
 	struct list_head pending_frees;
 };
 
+#define DECLARE_SERVER_INFO(sb, name) \
+	struct server_info *name = SCOUTFS_SB(sb)->server_info
+
+#if 0
 struct server_request {
 	struct server_connection *conn;
 	struct work_struct work;
@@ -89,6 +99,7 @@ struct server_connection {
 	struct work_struct recv_work;
 	struct mutex send_mutex;
 };
+#endif
 
 struct commit_waiter {
 	struct completion comp;
@@ -454,17 +465,9 @@ out:
 	return ret;
 }
 
-/*
- * Trigger a server shutdown by shutting down the listening socket.  The
- * server thread will break out of accept and exit.
- */
-static void shut_down_server(struct server_info *server)
+static void shutdown_server(struct server_info *server)
 {
-	mutex_lock(&server->mutex);
-	server->shutting_down = true;
-	if (server->listen_sock)
-		kernel_sock_shutdown(server->listen_sock, SHUT_RDWR);
-	mutex_unlock(&server->mutex);
+	complete(&server->shutdown_comp);
 }
 
 /*
@@ -495,21 +498,11 @@ static void queue_commit_work(struct server_info *server,
 }
 
 /*
- * Commit errors are fatal and shut down the server.  This is called
- * from request processing which shutdown will wait for.
+ * Wait for a commit during request processing and return its status.
  */
-static int wait_for_commit(struct server_info *server,
-			   struct commit_waiter *cw, u64 id, u8 type)
+static inline int wait_for_commit(struct commit_waiter *cw)
 {
-	struct super_block *sb = server->sb;
-
 	wait_for_completion(&cw->comp);
-	if (cw->ret < 0) {
-		scoutfs_err(sb, "commit error %d processing req id %llu type %u",
-				cw->ret, id, type);
-
-		shut_down_server(server);
-	}
 	return cw->ret;
 }
 
@@ -546,17 +539,18 @@ static void scoutfs_server_commit_func(struct work_struct *work)
 
 	down_write(&server->commit_rwsem);
 
-	if (!scoutfs_btree_has_dirty(sb)) {
-		ret = 0;
-		goto out;
-	}
-
+	/* try to free first which can dirty the btrees */
 	ret = apply_pending_frees(sb);
 	if (ret) {
 		scoutfs_err(sb, "server error freeing extents: %d", ret);
 		goto out;
 	}
 
+	if (!scoutfs_btree_has_dirty(sb)) {
+		ret = 0;
+		goto out;
+	}
+
 	ret = scoutfs_btree_write_dirty(sb);
 	if (ret) {
 		scoutfs_err(sb, "server error writing btree blocks: %d", ret);
@@ -591,6 +585,7 @@ out:
 	trace_scoutfs_server_commit_work_exit(sb, 0, ret);
 }
 
+#if 0
 /*
  * Request processing synchronously sends their reply from within their
  * processing work.  If this fails the socket is shutdown.
@@ -639,6 +634,7 @@ static int send_reply(struct server_connection *conn, u64 id,
 
 	return ret;
 }
+#endif
 
 void scoutfs_init_ment_to_net(struct scoutfs_net_manifest_entry *net_ment,
 			      struct scoutfs_manifest_entry *ment)
@@ -660,26 +656,26 @@ void scoutfs_init_ment_from_net(struct scoutfs_manifest_entry *ment,
 	ment->last = net_ment->last;
 }
 
-static int process_alloc_inodes(struct server_connection *conn,
-				u64 id, u8 type, void *data, unsigned data_len)
+static int server_alloc_inodes(struct super_block *sb,
+			       struct scoutfs_net_connection *conn,
+			       u8 cmd, u64 id, void *arg, u16 arg_len)
 {
-	struct server_info *server = conn->server;
-	struct super_block *sb = server->sb;
+	DECLARE_SERVER_INFO(sb, server);
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_super_block *super = &sbi->super;
-	struct scoutfs_net_inode_alloc ial;
+	struct scoutfs_net_inode_alloc ial = { 0, };
 	struct commit_waiter cw;
 	__le64 lecount;
 	u64 ino;
 	u64 nr;
 	int ret;
 
-	if (data_len != sizeof(lecount)) {
+	if (arg_len != sizeof(lecount)) {
 		ret = -EINVAL;
 		goto out;
 	}
 
-	memcpy(&lecount, data, data_len);
+	memcpy(&lecount, arg, arg_len);
 
 	down_read(&server->commit_rwsem);
 
@@ -695,52 +691,48 @@ static int process_alloc_inodes(struct server_connection *conn,
 	ial.ino = cpu_to_le64(ino);
 	ial.nr = cpu_to_le64(nr);
 
-	ret = wait_for_commit(server, &cw, id, type);
+	ret = wait_for_commit(&cw);
 out:
-	return send_reply(conn, id, type, ret, &ial, sizeof(ial));
+	return scoutfs_net_response(sb, conn, cmd, id, ret, &ial, sizeof(ial));
 }
 
 /*
  * Give the client an extent allocation of len blocks.  We leave the
  * details to the extent allocator.
  */
-static int process_alloc_extent(struct server_connection *conn,
-			       u64 id, u8 type, void *data, unsigned data_len)
+static int server_alloc_extent(struct super_block *sb,
+			       struct scoutfs_net_connection *conn,
+			       u8 cmd, u64 id, void *arg, u16 arg_len)
 {
-	struct server_info *server = conn->server;
-	struct super_block *sb = server->sb;
+	DECLARE_SERVER_INFO(sb, server);
 	struct commit_waiter cw;
-	struct scoutfs_net_extent nex;
+	struct scoutfs_net_extent nex = {0,};
 	__le64 leblocks;
 	u64 start;
 	u64 len;
 	int ret;
 
-	if (data_len != sizeof(leblocks)) {
+	if (arg_len != sizeof(leblocks)) {
 		ret = -EINVAL;
 		goto out;
 	}
 
-	memcpy(&leblocks, data, data_len);
+	memcpy(&leblocks, arg, arg_len);
 
 	down_read(&server->commit_rwsem);
 	ret = alloc_extent(sb, le64_to_cpu(leblocks), &start, &len);
-	if (ret == -ENOSPC) {
-		start = 0;
-		len = 0;
-		ret = 0;
-	}
-	if (ret == 0) {
-		nex.start = cpu_to_le64(start);
-		nex.len = cpu_to_le64(len);
-		queue_commit_work(server, &cw);
-	}
-	up_read(&server->commit_rwsem);
-
 	if (ret == 0)
-		ret = wait_for_commit(server, &cw, id, type);
+		queue_commit_work(server, &cw);
+	up_read(&server->commit_rwsem);
+	if (ret == 0)
+		ret = wait_for_commit(&cw);
+	if (ret)
+		goto out;
+
+	nex.start = cpu_to_le64(start);
+	nex.len = cpu_to_le64(len);
 out:
-	return send_reply(conn, id, type, ret, &nex, sizeof(nex));
+	return scoutfs_net_response(sb, conn, cmd, id, ret, &nex, sizeof(nex));
 }
 
 static bool invalid_net_extent_list(struct scoutfs_net_extent_list *nexl,
@@ -752,19 +744,19 @@ static bool invalid_net_extent_list(struct scoutfs_net_extent_list *nexl,
 				    extents[le64_to_cpu(nexl->nr)]));
 }
 
-static int process_free_extents(struct server_connection *conn,
-			       u64 id, u8 type, void *data, unsigned data_len)
+static int server_free_extents(struct super_block *sb,
+			       struct scoutfs_net_connection *conn,
+			       u8 cmd, u64 id, void *arg, u16 arg_len)
 {
-	struct server_info *server = conn->server;
-	struct super_block *sb = server->sb;
+	DECLARE_SERVER_INFO(sb, server);
 	struct scoutfs_net_extent_list *nexl;
 	struct commit_waiter cw;
-	int ret = 0;
+	int ret;
 	int err;
 	u64 i;
 
-	nexl = data;
-	if (invalid_net_extent_list(nexl, data_len)) {
+	nexl = arg;
+	if (invalid_net_extent_list(nexl, arg_len)) {
 		ret = -EINVAL;
 		goto out;
 	}
@@ -783,70 +775,66 @@ static int process_free_extents(struct server_connection *conn,
 	up_read(&server->commit_rwsem);
 
 	if (i > 0) {
-		err = wait_for_commit(server, &cw, id, type);
+		err = wait_for_commit(&cw);
 		if (ret == 0)
 			ret = err;
 	}
+
 out:
-	return send_reply(conn, id, type, ret, NULL, 0);
+	return scoutfs_net_response(sb, conn, cmd, id, ret, NULL, 0);
 }
 
 /*
  * We still special case segno allocation because it's aligned and we'd
  * like to keep that detail in the server.
  */
-static int process_alloc_segno(struct server_connection *conn,
-			       u64 id, u8 type, void *data, unsigned data_len)
+static int server_alloc_segno(struct super_block *sb,
+			      struct scoutfs_net_connection *conn,
+			      u8 cmd, u64 id, void *arg, u16 arg_len)
 {
-	struct server_info *server = conn->server;
-	struct super_block *sb = server->sb;
+	DECLARE_SERVER_INFO(sb, server);
 	struct commit_waiter cw;
 	__le64 lesegno = 0;
 	u64 segno;
 	int ret;
 
-	if (data_len != 0) {
+	if (arg_len != 0) {
 		ret = -EINVAL;
 		goto out;
 	}
 
 	down_read(&server->commit_rwsem);
 	ret = alloc_segno(sb, &segno);
-	if (ret == 0) {
-		lesegno = cpu_to_le64(segno);
+	if (ret == 0)
 		queue_commit_work(server, &cw);
-	} else if (ret == -ENOSPC) {
-		ret = 0;
-	}
 	up_read(&server->commit_rwsem);
+	if (ret == 0)
+		ret = wait_for_commit(&cw);
+	if (ret)
+		goto out;
 
-	if (ret == 0 && lesegno != 0)
-		ret = wait_for_commit(server, &cw, id, type);
+	lesegno = cpu_to_le64(segno);
 out:
-	return send_reply(conn, id, type, ret, &lesegno, sizeof(lesegno));
+	return scoutfs_net_response(sb, conn, cmd, id, ret,
+				    &lesegno, sizeof(lesegno));
 }
 
-static int process_record_segment(struct server_connection *conn, u64 id,
-				  u8 type, void *data, unsigned data_len)
+static int server_record_segment(struct super_block *sb,
+				 struct scoutfs_net_connection *conn,
+				 u8 cmd, u64 id, void *arg, u16 arg_len)
 {
-	struct server_info *server = conn->server;
-	struct super_block *sb = server->sb;
+	DECLARE_SERVER_INFO(sb, server);
 	struct scoutfs_net_manifest_entry *net_ment;
 	struct scoutfs_manifest_entry ment;
 	struct commit_waiter cw;
 	int ret;
 
-	if (data_len < sizeof(struct scoutfs_net_manifest_entry)) {
+	if (arg_len != sizeof(struct scoutfs_net_manifest_entry)) {
 		ret = -EINVAL;
 		goto out;
 	}
 
-	net_ment = data;
-
-	if (data_len != sizeof(*net_ment))  {
-		ret = -EINVAL;
-		goto out;
-	}
+	net_ment = arg;
 
 retry:
 	down_read(&server->commit_rwsem);
@@ -871,12 +859,13 @@ retry:
 	up_read(&server->commit_rwsem);
 
 	if (ret == 0) {
-		ret = wait_for_commit(server, &cw, id, type);
+		ret = wait_for_commit(&cw);
 		if (ret == 0)
 			scoutfs_compact_kick(sb);
 	}
+
 out:
-	return send_reply(conn, id, type, ret, NULL, 0);
+	return scoutfs_net_response(sb, conn, cmd, id, ret, NULL, 0);
 }
 
 struct pending_seq {
@@ -896,21 +885,21 @@ struct pending_seq {
  * XXX The pending seq tracking should be persistent so that it survives
  * server failover.
  */
-static int process_advance_seq(struct server_connection *conn, u64 id, u8 type,
-			       void *data, unsigned data_len)
+static int server_advance_seq(struct super_block *sb,
+			      struct scoutfs_net_connection *conn,
+			      u8 cmd, u64 id, void *arg, u16 arg_len)
 {
-	struct server_info *server = conn->server;
-	struct super_block *sb = server->sb;
+	DECLARE_SERVER_INFO(sb, server);
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_super_block *super = &sbi->super;
 	struct pending_seq *next_ps;
 	struct pending_seq *ps;
 	struct commit_waiter cw;
-	__le64 * __packed their_seq = data;
+	__le64 * __packed their_seq = arg;
 	__le64 next_seq;
 	int ret;
 
-	if (data_len != sizeof(__le64)) {
+	if (arg_len != sizeof(__le64)) {
 		ret = -EINVAL;
 		goto out;
 	}
@@ -941,24 +930,24 @@ static int process_advance_seq(struct server_connection *conn, u64 id, u8 type,
 	spin_unlock(&server->seq_lock);
 	queue_commit_work(server, &cw);
 	up_read(&server->commit_rwsem);
-	ret = wait_for_commit(server, &cw, id, type);
-
+	ret = wait_for_commit(&cw);
 out:
-	return send_reply(conn, id, type, ret, &next_seq, sizeof(next_seq));
+	return scoutfs_net_response(sb, conn, cmd, id, ret,
+				    &next_seq, sizeof(next_seq));
 }
 
-static int process_get_last_seq(struct server_connection *conn, u64 id,
-				u8 type, void *data, unsigned data_len)
+static int server_get_last_seq(struct super_block *sb,
+			       struct scoutfs_net_connection *conn,
+			       u8 cmd, u64 id, void *arg, u16 arg_len)
 {
-	struct server_info *server = conn->server;
-	struct super_block *sb = server->sb;
+	DECLARE_SERVER_INFO(sb, server);
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_super_block *super = &sbi->super;
 	struct pending_seq *ps;
 	__le64 last_seq;
 	int ret;
 
-	if (data_len != 0) {
+	if (arg_len != 0) {
 		ret = -EINVAL;
 		goto out;
 	}
@@ -975,18 +964,20 @@ static int process_get_last_seq(struct server_connection *conn, u64 id,
 	spin_unlock(&server->seq_lock);
 	ret = 0;
 out:
-	return send_reply(conn, id, type, ret, &last_seq, sizeof(last_seq));
+	return scoutfs_net_response(sb, conn, cmd, id, ret,
+				    &last_seq, sizeof(last_seq));
 }
 
-static int process_get_manifest_root(struct server_connection *conn, u64 id,
-				     u8 type, void *data, unsigned data_len)
+static int server_get_manifest_root(struct super_block *sb,
+				    struct scoutfs_net_connection *conn,
+				    u8 cmd, u64 id, void *arg, u16 arg_len)
 {
-	struct server_info *server = conn->server;
+	DECLARE_SERVER_INFO(sb, server);
 	struct scoutfs_btree_root root;
 	unsigned int start;
 	int ret;
 
-	if (data_len == 0) {
+	if (arg_len == 0) {
 		do {
 			start = read_seqcount_begin(&server->stable_seqcount);
 			root = server->stable_manifest_root;
@@ -996,24 +987,25 @@ static int process_get_manifest_root(struct server_connection *conn, u64 id,
 		ret = -EINVAL;
 	}
 
-	return send_reply(conn, id, type, ret, &root, sizeof(root));
+	return scoutfs_net_response(sb, conn, cmd, id, ret,
+				    &root, sizeof(root));
 }
 
 /*
  * Sample the super stats that the client wants for statfs by serializing
  * with each component.
  */
-static int process_statfs(struct server_connection *conn, u64 id, u8 type,
-			  void *data, unsigned data_len)
+static int server_statfs(struct super_block *sb,
+			 struct scoutfs_net_connection *conn,
+			 u8 cmd, u64 id, void *arg, u16 arg_len)
 {
-	struct server_info *server = conn->server;
-	struct super_block *sb = server->sb;
+	DECLARE_SERVER_INFO(sb, server);
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_super_block *super = &sbi->super;
 	struct scoutfs_net_statfs nstatfs;
 	int ret;
 
-	if (data_len == 0) {
+	if (arg_len == 0) {
 		/* uuid and total_segs are constant, so far */
 		memcpy(nstatfs.uuid, super->uuid, sizeof(nstatfs.uuid));
 
@@ -1030,7 +1022,8 @@ static int process_statfs(struct server_connection *conn, u64 id, u8 type,
 		ret = -EINVAL;
 	}
 
-	return send_reply(conn, id, type, ret, &nstatfs, sizeof(nstatfs));
+	return scoutfs_net_response(sb, conn, cmd, id, ret,
+				    &nstatfs, sizeof(nstatfs));
 }
 
 /*
@@ -1078,7 +1071,7 @@ int scoutfs_client_get_compaction(struct super_block *sb, void *curs)
 	up_read(&server->commit_rwsem);
 
 	if (ret == 0)
-		ret = wait_for_commit(server, &cw, U64_MAX, 1);
+		ret = wait_for_commit(&cw);
 
 	return ret;
 }
@@ -1117,13 +1110,14 @@ int scoutfs_client_finish_compaction(struct super_block *sb, void *curs,
 	up_read(&server->commit_rwsem);
 
 	if (ret == 0)
-		ret = wait_for_commit(server, &cw, U64_MAX, 2);
+		ret = wait_for_commit(&cw);
 
 	scoutfs_compact_kick(sb);
 
 	return ret;
 }
 
+#if 0
 typedef int (*process_func_t)(struct server_connection *conn, u64 id,
 			      u8 type, void *data, unsigned data_len);
 
@@ -1168,7 +1162,9 @@ static void scoutfs_server_process_func(struct work_struct *work)
 	/* process_one_work explicitly allows freeing work in its func */
 	kfree(req);
 }
+#endif
 
+#if 0
 /*
  * Always block receiving from the socket.  This owns the socket.  If
  * receive fails this shuts down and frees the socket.
@@ -1292,6 +1288,7 @@ out:
 
 	trace_scoutfs_server_recv_work_exit(sb, 0, ret);
 }
+#endif
 
 /*
  * This relies on the caller having read the current super and advanced
@@ -1308,6 +1305,7 @@ static int write_server_addr(struct super_block *sb, struct sockaddr_in *sin)
 	return scoutfs_write_dirty_super(sb);
 }
 
+#if 0
 static bool barrier_list_empty_careful(struct list_head *list)
 {
 	/* store caller's task state before loading wake condition */
@@ -1315,7 +1313,27 @@ static bool barrier_list_empty_careful(struct list_head *list)
 
 	return list_empty_careful(list);
 }
+#endif
 
+static scoutfs_net_request_t server_req_funcs[] = {
+	[SCOUTFS_NET_CMD_ALLOC_INODES]		= server_alloc_inodes,
+	[SCOUTFS_NET_CMD_ALLOC_EXTENT]		= server_alloc_extent,
+	[SCOUTFS_NET_CMD_FREE_EXTENTS]		= server_free_extents,
+	[SCOUTFS_NET_CMD_ALLOC_SEGNO]		= server_alloc_segno,
+	[SCOUTFS_NET_CMD_RECORD_SEGMENT]	= server_record_segment,
+	[SCOUTFS_NET_CMD_ADVANCE_SEQ]		= server_advance_seq,
+	[SCOUTFS_NET_CMD_GET_LAST_SEQ]		= server_get_last_seq,
+	[SCOUTFS_NET_CMD_GET_MANIFEST_ROOT]	= server_get_manifest_root,
+	[SCOUTFS_NET_CMD_STATFS]		= server_statfs,
+};
+
+static void server_notify_down(struct super_block *sb,
+			       struct scoutfs_net_connection *conn)
+{
+	DECLARE_SERVER_INFO(sb, server);
+
+	shutdown_server(server);
+}
 /*
  * This work is always running or has a delayed timer set while a super
  * is mounted.  It tries to grab the lock to become the server.  If it
@@ -1323,51 +1341,45 @@ static bool barrier_list_empty_careful(struct list_head *list)
  * anything goes wrong it releases the lock and sets a timer to try to
  * become the server all over again.
  */
-static void scoutfs_server_func(struct work_struct *work)
+static void scoutfs_server_worker(struct work_struct *work)
 {
 	struct server_info *server = container_of(work, struct server_info,
 						  dwork.work);
 	struct super_block *sb = server->sb;
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_super_block *super = &sbi->super;
+	struct scoutfs_net_connection *conn = NULL;
 	static struct sockaddr_in zeros = {0,};
-	struct socket *new_sock;
-	struct socket *sock = NULL;
 	struct scoutfs_lock *lock = NULL;
-	struct server_connection *conn;
 	struct pending_seq *ps;
 	struct pending_seq *ps_tmp;
 	DECLARE_WAIT_QUEUE_HEAD(waitq);
 	struct sockaddr_in sin;
 	LIST_HEAD(conn_list);
-	int addrlen;
-	int optval;
 	int ret;
 
 	trace_scoutfs_server_work_enter(sb, 0, 0);
 
-	init_waitqueue_head(&waitq);
+	init_completion(&server->shutdown_comp);
 
 	ret = scoutfs_lock_global(sb, DLM_LOCK_EX, 0,
 				  SCOUTFS_LOCK_TYPE_GLOBAL_SERVER, &lock);
 	if (ret)
 		goto out;
 
+	conn = scoutfs_net_alloc_conn(sb, NULL, server_notify_down,
+				      server_req_funcs, "server");
+	if (!conn) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
 	sin.sin_family = AF_INET;
 	sin.sin_addr.s_addr = le32_to_be32(sbi->opts.listen_addr.addr);
 	sin.sin_port = le16_to_be16(sbi->opts.listen_addr.port);
 
-	optval = 1;
-	ret = sock_create_kern(AF_INET, SOCK_STREAM, IPPROTO_TCP, &sock) ?:
-	      kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY,
-			        (char *)&optval, sizeof(optval)) ?:
-	      kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
-			        (char *)&optval, sizeof(optval));
-	if (ret)
-		goto out;
-
-	addrlen = sizeof(sin);
-	ret = kernel_bind(sock, (struct sockaddr *)&sin, addrlen);
+	/* get the address of our listening socket */
+	ret = scoutfs_net_bind(sb, conn, &sin);
 	if (ret) {
 		if (!server->bind_warned) {
 			scoutfs_err(sb, "server failed to bind to "SIN_FMT", errno %d%s.  Retrying indefinitely..",
@@ -1378,15 +1390,6 @@ static void scoutfs_server_func(struct work_struct *work)
 		}
 		goto out;
 	}
-	server->bind_warned = false;
-
-	kernel_getsockname(sock, (struct sockaddr *)&sin, &addrlen);
-	if (ret)
-		goto out;
-
-	ret = kernel_listen(sock, 255);
-	if (ret)
-		goto out;
 
 	/* publish the address for clients to connect to */
 	ret = scoutfs_read_super(sb, super);
@@ -1398,17 +1401,7 @@ static void scoutfs_server_func(struct work_struct *work)
 	if (ret)
 		goto out;
 
-	/* either see shutting down or they'll shutdown our sock */
-	mutex_lock(&server->mutex);
-	server->listen_task = current;
-	server->listen_sock = sock;
-	if (server->shutting_down)
-		ret = -ESHUTDOWN;
-	mutex_unlock(&server->mutex);
-	if (ret)
-		goto out;
-
-	/* finally start up the server subsystems before accepting */
+	/* start up the server subsystems before accepting */
 	ret = scoutfs_btree_setup(sb) ?:
 	      scoutfs_manifest_setup(sb) ?:
 	      scoutfs_compact_setup(sb);
@@ -1420,69 +1413,24 @@ static void scoutfs_server_func(struct work_struct *work)
 
 	scoutfs_info(sb, "server started on "SIN_FMT, SIN_ARG(&sin));
 
-	for (;;) {
-		ret = kernel_accept(sock, &new_sock, 0);
-		if (ret < 0)
-			break;
+	/* start accepting connections and processing work */
+	scoutfs_net_listen(sb, conn);
 
-		conn = kmalloc(sizeof(struct server_connection), GFP_NOFS);
-		if (!conn) {
-			sock_release(new_sock);
-			ret = -ENOMEM;
-			continue;
-		}
-
-		addrlen = sizeof(struct sockaddr_in);
-		ret = kernel_getsockname(new_sock,
-					 (struct sockaddr *)&conn->sockname,
-					 &addrlen) ?:
-		      kernel_getpeername(new_sock,
-					 (struct sockaddr *)&conn->peername,
-					 &addrlen);
-		if (ret) {
-			sock_release(new_sock);
-			continue;
-		}
-
-		/*
-		 * XXX yeah, ok, killing the sock and accepting a new
-		 * one is racey.  think about that in all the code.  Are
-		 * we destroying a resource to shutdown that the thing
-		 * we're canceling creates?
-		 */
-
-		conn->server = server;
-		conn->sock = new_sock;
-		mutex_init(&conn->send_mutex);
-
-		scoutfs_info(sb, "server accepted "SIN_FMT" -> "SIN_FMT,
-			     SIN_ARG(&conn->peername),
-			     SIN_ARG(&conn->sockname));
-
-		/* recv work owns the conn once its in the list */
-		mutex_lock(&server->mutex);
-		list_add(&conn->head, &conn_list);
-		mutex_unlock(&server->mutex);
-
-		INIT_WORK(&conn->recv_work, scoutfs_server_recv_func);
-		queue_work(server->wq, &conn->recv_work);
-	}
-
-	/* shutdown send and recv on all accepted sockets */
-	mutex_lock(&server->mutex);
-	list_for_each_entry(conn, &conn_list, head)
-		kernel_sock_shutdown(conn->sock, SHUT_RDWR);
-	mutex_unlock(&server->mutex);
-
-	/* wait for all recv work to finish and free connections */
-	wait_event(waitq, barrier_list_empty_careful(&conn_list));
+	/* wait for listening down or umount, conn can still be live */
+	wait_for_completion_interruptible(&server->shutdown_comp);
 
 	scoutfs_info(sb, "server shutting down on "SIN_FMT, SIN_ARG(&sin));
 
 shutdown:
+	/* wait for request processing */
+	scoutfs_net_shutdown(sb, conn);
+	/* wait for commit queued by request processing */
+	flush_work(&server->commit_work);
 
 	/* shut down all the server subsystems */
 	scoutfs_compact_destroy(sb);
+	/* (wait for possible double commit work queued by compaction) */
+	flush_work(&server->commit_work);
 	destroy_pending_frees(sb);
 	scoutfs_manifest_destroy(sb);
 	scoutfs_btree_destroy(sb);
@@ -1496,9 +1444,7 @@ shutdown:
 	write_server_addr(sb, &zeros);
 
 out:
-	if (sock)
-		sock_release(sock);
-
+	scoutfs_net_free_conn(sb, conn);
 	scoutfs_unlock(sb, lock, DLM_LOCK_EX);
 
 	/* always requeues, cancel_delayed_work_sync cancels on shutdown */
@@ -1517,8 +1463,9 @@ int scoutfs_server_setup(struct super_block *sb)
 		return -ENOMEM;
 
 	server->sb = sb;
-	INIT_DELAYED_WORK(&server->dwork, scoutfs_server_func);
-	mutex_init(&server->mutex);
+	init_completion(&server->shutdown_comp);
+	server->bind_warned = false;
+	INIT_DELAYED_WORK(&server->dwork, scoutfs_server_worker);
 	init_rwsem(&server->commit_rwsem);
 	init_llist_head(&server->commit_waiters);
 	INIT_WORK(&server->commit_work, scoutfs_server_commit_func);
@@ -1547,7 +1494,7 @@ void scoutfs_server_destroy(struct super_block *sb)
 	struct server_info *server = sbi->server_info;
 
 	if (server) {
-		shut_down_server(server);
+		shutdown_server(server);
 
 		/* wait for server work to wait for everything to shut down */
 		cancel_delayed_work_sync(&server->dwork);

kmod/src/server.h

@@ -22,13 +22,14 @@ do {									\
 	__entry->name##_addr & 255,		\
 	__entry->name##_port
 
-#define SNH_FMT	"id %llu data_len %u type %u status %u"
+#define SNH_FMT	"id %llu data_len %u msg %u cmd %u error %u"
 
 #define snh_trace_define(name)		\
 	__field(__u64, name##_id)	\
 	__field(__u16, name##_data_len)	\
-	__field(__u8, name##_type)	\
-	__field(__u8, name##_status)
+	__field(__u8, name##_msg)	\
+	__field(__u8, name##_cmd)	\
+	__field(__u8, name##_error)
 
 #define snh_trace_assign(name, nh)				\
 do {								\
@@ -36,13 +37,14 @@ do {								\
 								\
 	__entry->name##_id = le64_to_cpu(_nh->id);		\
 	__entry->name##_data_len = le16_to_cpu(_nh->data_len);	\
-	__entry->name##_type = _nh->type;			\
-	__entry->name##_status = _nh->status;			\
+	__entry->name##_msg = _nh->msg;				\
+	__entry->name##_cmd = _nh->cmd;				\
+	__entry->name##_error = _nh->error;			\
 } while (0)
 
 #define snh_trace_args(name) \
-	__entry->name##_id, __entry->name##_data_len, __entry->name##_type, \
-	__entry->name##_status
+	__entry->name##_id, __entry->name##_data_len, __entry->name##_msg, \
+	__entry->name##_cmd, __entry->name##_error
 
 void scoutfs_init_ment_to_net(struct scoutfs_net_manifest_entry *net_ment,
 			      struct scoutfs_manifest_entry *ment);

kmod/src/super.c

@@ -38,6 +38,7 @@
 #include "compact.h"
 #include "data.h"
 #include "lock.h"
+#include "net.h"
 #include "client.h"
 #include "server.h"
 #include "options.h"
@@ -133,6 +134,7 @@ static void scoutfs_put_super(struct super_block *sb)
 	/* the server locks the listen address and compacts */
 	scoutfs_lock_shutdown(sb);
 	scoutfs_server_destroy(sb);
+	scoutfs_net_destroy(sb);
 	scoutfs_seg_destroy(sb);
 	scoutfs_lock_destroy(sb);
 
@@ -333,6 +335,7 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 	      scoutfs_data_setup(sb) ?:
 	      scoutfs_setup_trans(sb) ?:
 	      scoutfs_lock_setup(sb) ?:
+	      scoutfs_net_setup(sb) ?:
 	      scoutfs_server_setup(sb) ?:
 	      scoutfs_client_setup(sb) ?:
 	      scoutfs_lock_node_id(sb, DLM_LOCK_EX, 0, sbi->node_id,

kmod/src/super.h

@@ -22,6 +22,7 @@ struct inode_sb_info;
 struct btree_info;
 struct sysfs_info;
 struct options_sb_info;
+struct net_info;
 
 struct scoutfs_sb_info {
 	struct super_block *sb;
@@ -42,6 +43,7 @@ struct scoutfs_sb_info {
 	struct data_info *data_info;
 	struct inode_sb_info *inode_sb_info;
 	struct btree_info *btree_info;
+	struct net_info *net_info;
 
 	wait_queue_head_t trans_hold_wq;
 	struct task_struct *trans_task;