Fully destroy inodes after all mounts evict

Today an inode's items are deleted once its nlink reaches zero and the
final iput is called in a local mount.  This can delete inodes from
under other mounts which have opened the inode before it was unlinked on
another mount.

We fix this by adding cached inode tracking.  Each mount maintains
groups of cached inode bitmaps at the same granularity as inode locking.
As a mount performs its final iput it gets a bitmap from the server
which indicates if any other mount has inodes in the group open.

This makes the two fast paths of opening and closing linked files and of
deleting a file that was unlinked locally only pay a moderate cost of
either maintaining the bitmap locally and only getting the open map once
per lock group.  Removing many files in a group will only lock and get
the open map once per group.

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

kmod/src/Makefile

@@ -27,6 +27,7 @@ scoutfs-y +=			\
 	lock_server.o		\
 	msg.o			\
 	net.o			\
+	omap.o			\
 	options.o		\
 	per_task.o		\
 	quorum.o		\

kmod/src/client.c

@@ -31,6 +31,7 @@
 #include "net.h"
 #include "endian_swap.h"
 #include "quorum.h"
+#include "omap.h"
 
 /*
  * The client is responsible for maintaining a connection to the server.
@@ -215,6 +216,39 @@ int scoutfs_client_srch_commit_compact(struct super_block *sb,
 					res, sizeof(*res), NULL, 0);
 }
 
+int scoutfs_client_send_omap_response(struct super_block *sb, u64 id,
+				      struct scoutfs_open_ino_map *map)
+{
+	struct client_info *client = SCOUTFS_SB(sb)->client_info;
+
+	return scoutfs_net_response(sb, client->conn, SCOUTFS_NET_CMD_OPEN_INO_MAP,
+				    id, 0, map, sizeof(*map));
+}
+
+/* The client is receiving an omap request from the server */
+static int client_open_ino_map(struct super_block *sb, struct scoutfs_net_connection *conn,
+			       u8 cmd, u64 id, void *arg, u16 arg_len)
+{
+	if (arg_len != sizeof(struct scoutfs_open_ino_map_args))
+		return -EINVAL;
+
+	return scoutfs_omap_client_handle_request(sb, id, arg);
+}
+
+/* The client is sending an omap request to the server */
+int scoutfs_client_open_ino_map(struct super_block *sb, u64 group_nr,
+				struct scoutfs_open_ino_map *map)
+{
+	struct client_info *client = SCOUTFS_SB(sb)->client_info;
+	struct scoutfs_open_ino_map_args args = {
+		.group_nr = cpu_to_le64(group_nr),
+		.req_id = 0,
+	};
+
+	return scoutfs_net_sync_request(sb, client->conn, SCOUTFS_NET_CMD_OPEN_INO_MAP,
+					&args, sizeof(args), map, sizeof(*map));
+}
+
 /* The client is receiving a invalidation request from the server */
 static int client_lock(struct super_block *sb,
 		       struct scoutfs_net_connection *conn, u8 cmd, u64 id,
@@ -413,6 +447,7 @@ out:
 static scoutfs_net_request_t client_req_funcs[] = {
 	[SCOUTFS_NET_CMD_LOCK]			= client_lock,
 	[SCOUTFS_NET_CMD_LOCK_RECOVER]		= client_lock_recover,
+	[SCOUTFS_NET_CMD_OPEN_INO_MAP]		= client_open_ino_map,
 };
 
 /*

kmod/src/client.h

@@ -22,6 +22,10 @@ int scoutfs_client_srch_get_compact(struct super_block *sb,
 				    struct scoutfs_srch_compact *sc);
 int scoutfs_client_srch_commit_compact(struct super_block *sb,
 				       struct scoutfs_srch_compact *res);
+int scoutfs_client_send_omap_response(struct super_block *sb, u64 id,
+				      struct scoutfs_open_ino_map *map);
+int scoutfs_client_open_ino_map(struct super_block *sb, u64 group_nr,
+				struct scoutfs_open_ino_map *map);
 
 int scoutfs_client_setup(struct super_block *sb);
 void scoutfs_client_destroy(struct super_block *sb);

kmod/src/dir.c

@@ -30,6 +30,7 @@
 #include "item.h"
 #include "lock.h"
 #include "hash.h"
+#include "omap.h"
 #include "counters.h"
 #include "scoutfs_trace.h"
 

kmod/src/file.c

@@ -27,6 +27,7 @@
 #include "file.h"
 #include "inode.h"
 #include "per_task.h"
+#include "omap.h"
 
 /* TODO: Direct I/O, AIO */
 ssize_t scoutfs_file_aio_read(struct kiocb *iocb, const struct iovec *iov,

kmod/src/format.h

@@ -840,6 +840,7 @@ enum scoutfs_net_cmd {
 	SCOUTFS_NET_CMD_LOCK_RECOVER,
 	SCOUTFS_NET_CMD_SRCH_GET_COMPACT,
 	SCOUTFS_NET_CMD_SRCH_COMMIT_COMPACT,
+	SCOUTFS_NET_CMD_OPEN_INO_MAP,
 	SCOUTFS_NET_CMD_FAREWELL,
 	SCOUTFS_NET_CMD_UNKNOWN,
 };
@@ -960,4 +961,42 @@ enum scoutfs_corruption_sources {
 
 #define SC_NR_LONGS DIV_ROUND_UP(SC_NR_SOURCES, BITS_PER_LONG)
 
+#define SCOUTFS_OPEN_INO_MAP_SHIFT	10
+#define SCOUTFS_OPEN_INO_MAP_BITS	(1 << SCOUTFS_OPEN_INO_MAP_SHIFT)
+#define SCOUTFS_OPEN_INO_MAP_MASK	(SCOUTFS_OPEN_INO_MAP_BITS - 1)
+#define SCOUTFS_OPEN_INO_MAP_LE64S	(SCOUTFS_OPEN_INO_MAP_BITS / 64)
+
+/*
+ * The request and response conversation is as follows:
+ *
+ * client[init] -> server:
+ *	group_nr = G
+ *	req_id = 0	(I)
+ * server -> client[*]
+ *	group_nr = G
+ *	req_id = R
+ * client[*] -> server
+ *	group_nr = G	(I)
+ *	req_id = R
+ *	bits
+ * server -> client[init]
+ *	group_nr = G	(I)
+ *	req_id = R	(I)
+ *	bits
+ *
+ * Many of the fields in individual messages are ignored ("I") because
+ * the net id or the omap req_id can be used to identify the
+ * conversation.  We always include them on the wire to make inspected
+ * messages easier to follow.
+ */
+struct scoutfs_open_ino_map_args {
+	__le64 group_nr;
+	__le64 req_id;
+};
+
+struct scoutfs_open_ino_map {
+	struct scoutfs_open_ino_map_args args;
+	__le64 bits[SCOUTFS_OPEN_INO_MAP_LE64S];
+};
+
 #endif

kmod/src/inode.c

@@ -33,6 +33,7 @@
 #include "item.h"
 #include "client.h"
 #include "cmp.h"
+#include "omap.h"
 
 /*
  * XXX
@@ -693,6 +694,8 @@ struct inode *scoutfs_iget(struct super_block *sb, u64 ino)
 		atomic64_set(&si->last_refreshed, 0);
 
 		ret = scoutfs_inode_refresh(inode, lock, 0);
+		if (ret == 0)
+			ret = scoutfs_omap_inc(sb, ino);
 		if (ret) {
 			iget_failed(inode);
 			inode = ERR_PTR(ret);
@@ -1405,10 +1408,17 @@ struct inode *scoutfs_new_inode(struct super_block *sb, struct inode *dir,
 	store_inode(&sinode, inode);
 	init_inode_key(&key, scoutfs_ino(inode));
 
+	ret = scoutfs_omap_inc(sb, ino);
+	if (ret < 0)
+		goto out;
+
 	ret = scoutfs_item_create(sb, &key, &sinode, sizeof(sinode), lock);
+	if (ret < 0)
+		scoutfs_omap_dec(sb, ino);
+out:
 	if (ret) {
 		iput(inode);
-		return ERR_PTR(ret);
+		inode = ERR_PTR(ret);
 	}
 
 	return inode;
@@ -1453,15 +1463,15 @@ int scoutfs_orphan_delete(struct super_block *sb, u64 ino)
 
 /*
  * Remove all the items associated with a given inode.  This is only
- * called once nlink has dropped to zero so we don't have to worry about
- * dirents referencing the inode or link backrefs.  Dropping nlink to 0
- * also created an orphan item.  That orphan item will continue
- * triggering attempts to finish previous partial deletion until all
- * deletion is complete and the orphan item is removed.
+ * called once nlink has dropped to zero and nothing has the inode open
+ * so we don't have to worry about dirents referencing the inode or link
+ * backrefs.  Dropping nlink to 0 also created an orphan item.  That
+ * orphan item will continue triggering attempts to finish previous
+ * partial deletion until all deletion is complete and the orphan item
+ * is removed.
  */
-static int delete_inode_items(struct super_block *sb, u64 ino)
+static int delete_inode_items(struct super_block *sb, u64 ino, struct scoutfs_lock *lock)
 {
-	struct scoutfs_lock *lock = NULL;
 	struct scoutfs_inode sinode;
 	struct scoutfs_key key;
 	LIST_HEAD(ind_locks);
@@ -1471,10 +1481,6 @@ static int delete_inode_items(struct super_block *sb, u64 ino)
 	u64 size;
 	int ret;
 
-	ret = scoutfs_lock_ino(sb, SCOUTFS_LOCK_WRITE, 0, ino, &lock);
-	if (ret)
-		return ret;
-
 	init_inode_key(&key, ino);
 
 	ret = scoutfs_item_lookup_exact(sb, &key, &sinode, sizeof(sinode),
@@ -1539,18 +1545,24 @@ out:
 	if (release)
 		scoutfs_release_trans(sb);
 	scoutfs_inode_index_unlock(sb, &ind_locks);
-	scoutfs_unlock(sb, lock, SCOUTFS_LOCK_WRITE);
+
 	return ret;
 }
 
 /*
  * iput_final has already written out the dirty pages to the inode
  * before we get here.  We're left with a clean inode that we have to
- * tear down.  If there are no more links to the inode then we also
- * remove all its persistent structures.
+ * tear down.  We use locking and open inode number bitmaps to decide if
+ * we should finally destroy an inode that is no longer open nor
+ * reachable through directory entries.
  */
 void scoutfs_evict_inode(struct inode *inode)
 {
+	struct super_block *sb = inode->i_sb;
+	const u64 ino = scoutfs_ino(inode);
+	struct scoutfs_lock *lock;
+	int ret;
+
 	trace_scoutfs_evict_inode(inode->i_sb, scoutfs_ino(inode),
 				  inode->i_nlink, is_bad_inode(inode));
 
@@ -1559,8 +1571,17 @@ void scoutfs_evict_inode(struct inode *inode)
 
 	truncate_inode_pages_final(&inode->i_data);
 
-	if (inode->i_nlink == 0)
-		delete_inode_items(inode->i_sb, scoutfs_ino(inode));
+	ret = scoutfs_omap_should_delete(sb, inode, &lock);
+	if (ret > 0) {
+		ret = delete_inode_items(inode->i_sb, scoutfs_ino(inode), lock);
+		scoutfs_unlock(sb, lock, SCOUTFS_LOCK_WRITE);
+	}
+	if (ret < 0)
+		scoutfs_err(sb, "error %d while checking to delete inode nr %llu, it might linger.",
+			    ret, ino);
+
+	scoutfs_omap_dec(sb, ino);
+
 clear:
 	clear_inode(inode);
 }
@@ -1588,8 +1609,10 @@ int scoutfs_scan_orphans(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_lock *lock = sbi->rid_lock;
+	struct scoutfs_lock *inode_lock = NULL;
 	struct scoutfs_key key;
 	struct scoutfs_key last;
+	u64 ino;
 	int err = 0;
 	int ret;
 
@@ -1605,7 +1628,13 @@ int scoutfs_scan_orphans(struct super_block *sb)
 		if (ret < 0)
 			goto out;
 
-		ret = delete_inode_items(sb, le64_to_cpu(key.sko_ino));
+		ino = le64_to_cpu(key.sko_ino);
+
+		ret = scoutfs_lock_ino(sb, SCOUTFS_LOCK_WRITE, 0, ino, &inode_lock);
+		if (ret == 0) {
+			ret = delete_inode_items(sb, le64_to_cpu(key.sko_ino), inode_lock);
+			scoutfs_unlock(sb, inode_lock, SCOUTFS_LOCK_WRITE);
+		}
 		if (ret && ret != -ENOENT && !err)
 			err = ret;
 

kmod/src/lock.c

@@ -34,6 +34,7 @@
 #include "data.h"
 #include "xattr.h"
 #include "item.h"
+#include "omap.h"
 
 /*
  * scoutfs uses a lock service to manage item cache consistency between
@@ -234,6 +235,7 @@ static void lock_free(struct lock_info *linfo, struct scoutfs_lock *lock)
 	BUG_ON(!list_empty(&lock->shrink_head));
 	BUG_ON(!list_empty(&lock->cov_list));
 
+	scoutfs_omap_free_lock_data(lock->omap_data);
 	kfree(lock);
 }
 
@@ -269,6 +271,7 @@ static struct scoutfs_lock *lock_alloc(struct super_block *sb,
 	lock->mode = SCOUTFS_LOCK_NULL;
 
 	atomic64_set(&lock->forest_bloom_nr, 0);
+	spin_lock_init(&lock->omap_spinlock);
 
 	trace_scoutfs_lock_alloc(sb, lock);
 

kmod/src/lock.h

@@ -10,6 +10,8 @@
 
 #define SCOUTFS_LOCK_NR_MODES		SCOUTFS_LOCK_INVALID
 
+struct scoutfs_omap_lock;
+
 /*
  * A few fields (start, end, refresh_gen, write_version, granted_mode)
  * are referenced by code outside lock.c.
@@ -47,6 +49,10 @@ struct scoutfs_lock {
 
 	/* the forest tracks which log tree last saw bloom bit updates */
 	atomic64_t forest_bloom_nr;
+
+	/* open ino mapping has a valid map for a held write lock */
+	spinlock_t omap_spinlock;
+	struct scoutfs_omap_lock_data *omap_data;
 };
 
 struct scoutfs_lock_coverage {

kmod/src/omap.h

@@ -0,0 +1,24 @@
+#ifndef _SCOUTFS_OMAP_H_
+#define _SCOUTFS_OMAP_H_
+
+int scoutfs_omap_inc(struct super_block *sb, u64 ino);
+void scoutfs_omap_dec(struct super_block *sb, u64 ino);
+int scoutfs_omap_should_delete(struct super_block *sb, struct inode *inode,
+			       struct scoutfs_lock **lock_ret);
+void scoutfs_omap_free_lock_data(struct scoutfs_omap_lock_data *ldata);
+int scoutfs_omap_client_handle_request(struct super_block *sb, u64 id,
+				       struct scoutfs_open_ino_map_args *args);
+
+int scoutfs_omap_add_rid(struct super_block *sb, u64 rid);
+int scoutfs_omap_remove_rid(struct super_block *sb, u64 rid);
+int scoutfs_omap_finished_recovery(struct super_block *sb);
+int scoutfs_omap_server_handle_request(struct super_block *sb, u64 rid, u64 id,
+				       struct scoutfs_open_ino_map_args *args);
+int scoutfs_omap_server_handle_response(struct super_block *sb, u64 rid,
+					struct scoutfs_open_ino_map *resp_map);
+void scoutfs_omap_server_shutdown(struct super_block *sb);
+
+int scoutfs_omap_setup(struct super_block *sb);
+void scoutfs_omap_destroy(struct super_block *sb);
+
+#endif

kmod/src/scoutfs_trace.h

@@ -2402,6 +2402,89 @@ TRACE_EVENT(scoutfs_item_invalidate_page,
 		  sk_trace_args(pg_start), sk_trace_args(pg_end), __entry->pgi)
 );
 
+DECLARE_EVENT_CLASS(scoutfs_omap_group_class,
+	TP_PROTO(struct super_block *sb, void *grp, u64 group_nr, unsigned int group_total,
+		 int bit_nr, int bit_count),
+
+	TP_ARGS(sb, grp, group_nr, group_total, bit_nr, bit_count),
+
+	TP_STRUCT__entry(
+		SCSB_TRACE_FIELDS
+		__field(void *, grp)
+		__field(__u64, group_nr)
+		__field(unsigned int, group_total)
+		__field(int, bit_nr)
+		__field(int, bit_count)
+	),
+
+	TP_fast_assign(
+		SCSB_TRACE_ASSIGN(sb);
+		__entry->grp = grp;
+		__entry->group_nr = group_nr;
+		__entry->group_total = group_total;
+		__entry->bit_nr = bit_nr;
+		__entry->bit_count = bit_count;
+	),
+
+	TP_printk(SCSBF" grp %p group_nr %llu group_total %u bit_nr %d bit_count %d",
+		  SCSB_TRACE_ARGS, __entry->grp, __entry->group_nr, __entry->group_total,
+		  __entry->bit_nr, __entry->bit_count)
+);
+
+DEFINE_EVENT(scoutfs_omap_group_class, scoutfs_omap_group_alloc,
+	TP_PROTO(struct super_block *sb, void *grp, u64 group_nr, unsigned int group_total,
+		 int bit_nr, int bit_count),
+	TP_ARGS(sb, grp, group_nr, group_total, bit_nr, bit_count)
+);
+DEFINE_EVENT(scoutfs_omap_group_class, scoutfs_omap_group_free,
+	TP_PROTO(struct super_block *sb, void *grp, u64 group_nr, unsigned int group_total,
+		 int bit_nr, int bit_count),
+	TP_ARGS(sb, grp, group_nr, group_total, bit_nr, bit_count)
+);
+DEFINE_EVENT(scoutfs_omap_group_class, scoutfs_omap_group_inc,
+	TP_PROTO(struct super_block *sb, void *grp, u64 group_nr, unsigned int group_total,
+		 int bit_nr, int bit_count),
+	TP_ARGS(sb, grp, group_nr, group_total, bit_nr, bit_count)
+);
+DEFINE_EVENT(scoutfs_omap_group_class, scoutfs_omap_group_dec,
+	TP_PROTO(struct super_block *sb, void *grp, u64 group_nr, unsigned int group_total,
+		 int bit_nr, int bit_count),
+	TP_ARGS(sb, grp, group_nr, group_total, bit_nr, bit_count)
+);
+DEFINE_EVENT(scoutfs_omap_group_class, scoutfs_omap_group_request,
+	TP_PROTO(struct super_block *sb, void *grp, u64 group_nr, unsigned int group_total,
+		 int bit_nr, int bit_count),
+	TP_ARGS(sb, grp, group_nr, group_total, bit_nr, bit_count)
+);
+DEFINE_EVENT(scoutfs_omap_group_class, scoutfs_omap_group_destroy,
+	TP_PROTO(struct super_block *sb, void *grp, u64 group_nr, unsigned int group_total,
+		 int bit_nr, int bit_count),
+	TP_ARGS(sb, grp, group_nr, group_total, bit_nr, bit_count)
+);
+
+TRACE_EVENT(scoutfs_omap_should_delete,
+	TP_PROTO(struct super_block *sb, u64 ino, unsigned int nlink, int ret),
+
+	TP_ARGS(sb, ino, nlink, ret),
+
+	TP_STRUCT__entry(
+		SCSB_TRACE_FIELDS
+		__field(__u64, ino)
+		__field(unsigned int, nlink)
+		__field(int, ret)
+	),
+
+	TP_fast_assign(
+		SCSB_TRACE_ASSIGN(sb);
+		__entry->ino = ino;
+		__entry->nlink = nlink;
+		__entry->ret = ret;
+	),
+
+	TP_printk(SCSBF" ino %llu nlink %u ret %d",
+		  SCSB_TRACE_ARGS, __entry->ino, __entry->nlink, __entry->ret)
+);
+
 #endif /* _TRACE_SCOUTFS_H */
 
 /* This part must be outside protection */

kmod/src/server.c

@@ -39,6 +39,7 @@
 #include "alloc.h"
 #include "forest.h"
 #include "recov.h"
+#include "omap.h"
 
 /*
  * Every active mount can act as the server that listens on a net
@@ -1020,6 +1021,60 @@ out:
 	return scoutfs_net_response(sb, conn, cmd, id, ret, NULL, 0);
 }
 
+/* The server is receiving an omap response from the client */
+static int open_ino_map_response(struct super_block *sb, struct scoutfs_net_connection *conn,
+				 void *resp, unsigned int resp_len, int error, void *data)
+{
+	u64 rid = scoutfs_net_client_rid(conn);
+
+	if (resp_len != sizeof(struct scoutfs_open_ino_map))
+		return -EINVAL;
+
+	return scoutfs_omap_server_handle_response(sb, rid, resp);
+}
+
+/* The server is sending an omap request to the client */
+int scoutfs_server_send_omap_request(struct super_block *sb, u64 rid,
+				     struct scoutfs_open_ino_map_args *args)
+{
+	struct server_info *server = SCOUTFS_SB(sb)->server_info;
+
+	return scoutfs_net_submit_request_node(sb, server->conn, rid, SCOUTFS_NET_CMD_OPEN_INO_MAP,
+					      args, sizeof(*args),
+					      open_ino_map_response, NULL, NULL);
+}
+
+/* The server is sending an omap response to the client */
+int scoutfs_server_send_omap_response(struct super_block *sb, u64 rid, u64 id,
+				      struct scoutfs_open_ino_map *map, int err)
+{
+	struct server_info *server = SCOUTFS_SB(sb)->server_info;
+
+	return scoutfs_net_response_node(sb, server->conn, rid,
+					 SCOUTFS_NET_CMD_OPEN_INO_MAP, id, err,
+					 map, sizeof(*map));
+}
+
+/* The server is receiving an omap request from the client */
+static int server_open_ino_map(struct super_block *sb, struct scoutfs_net_connection *conn,
+			       u8 cmd, u64 id, void *arg, u16 arg_len)
+{
+	u64 rid = scoutfs_net_client_rid(conn);
+	int ret;
+
+	if (arg_len != sizeof(struct scoutfs_open_ino_map_args)) {
+		ret = -EINVAL;
+		goto out;
+	}
+
+	ret = scoutfs_omap_server_handle_request(sb, rid, id, arg);
+out:
+	if (ret < 0)
+		return scoutfs_net_response(sb, conn, cmd, id, ret, NULL, 0);
+
+	return 0;
+}
+
 static void init_mounted_client_key(struct scoutfs_key *key, u64 rid)
 {
 	*key = (struct scoutfs_key) {
@@ -1237,9 +1292,10 @@ send_err:
 	scoutfs_net_server_greeting(sb, conn, le64_to_cpu(gr->rid), id,
 				    reconnecting, first_contact, farewell);
 
-	/* lock server might send recovery request */
+	/* let layers know we have a client connecting for the first time */
 	if (le64_to_cpu(gr->server_term) != server->term) {
-		ret = scoutfs_lock_server_greeting(sb, le64_to_cpu(gr->rid));
+		ret = scoutfs_lock_server_greeting(sb, le64_to_cpu(gr->rid)) ?:
+		      scoutfs_omap_add_rid(sb, le64_to_cpu(gr->rid));
 		if (ret)
 			goto out;
 	}
@@ -1273,7 +1329,8 @@ static int reclaim_rid(struct super_block *sb, u64 rid)
 	      remove_trans_seq(sb, rid) ?:
 	      reclaim_log_trees(sb, rid) ?:
 	      cancel_srch_compact(sb, rid) ?:
-	      delete_mounted_client(sb, rid);
+	      delete_mounted_client(sb, rid) ?:
+	      scoutfs_omap_remove_rid(sb, rid);
 
 	return scoutfs_server_apply_commit(sb, ret);
 }
@@ -1507,6 +1564,7 @@ static scoutfs_net_request_t server_req_funcs[] = {
 	[SCOUTFS_NET_CMD_LOCK]			= server_lock,
 	[SCOUTFS_NET_CMD_SRCH_GET_COMPACT]	= server_srch_get_compact,
 	[SCOUTFS_NET_CMD_SRCH_COMMIT_COMPACT]	= server_srch_commit_compact,
+	[SCOUTFS_NET_CMD_OPEN_INO_MAP]		= server_open_ino_map,
 	[SCOUTFS_NET_CMD_FAREWELL]		= server_farewell,
 };
 
@@ -1564,7 +1622,8 @@ static void finished_recovery(struct super_block *sb)
 
 	scoutfs_info(sb, "all clients recovered");
 
-	ret = scoutfs_lock_server_finished_recovery(sb);
+	ret = scoutfs_omap_finished_recovery(sb) ?:
+	      scoutfs_lock_server_finished_recovery(sb);
 	if (ret < 0) {
 		scoutfs_err(sb, "error %d resuming after recovery finished, shutting down", ret);
 		stop_server(server);
@@ -1787,6 +1846,7 @@ shutdown:
 	flush_work(&server->commit_work);
 
 	scoutfs_lock_server_destroy(sb);
+	scoutfs_omap_server_shutdown(sb);
 
 out:
 	scoutfs_net_free_conn(sb, conn);

kmod/src/server.h

@@ -66,6 +66,11 @@ int scoutfs_server_hold_commit(struct super_block *sb);
 int scoutfs_server_apply_commit(struct super_block *sb, int err);
 void scoutfs_server_recov_finish(struct super_block *sb, u64 rid, int which);
 
+int scoutfs_server_send_omap_request(struct super_block *sb, u64 rid,
+				     struct scoutfs_open_ino_map_args *args);
+int scoutfs_server_send_omap_response(struct super_block *sb, u64 rid, u64 id,
+				      struct scoutfs_open_ino_map *map, int err);
+
 struct sockaddr_in;
 struct scoutfs_quorum_elected_info;
 int scoutfs_server_start(struct super_block *sb, u64 term);

kmod/src/super.c

@@ -45,6 +45,7 @@
 #include "item.h"
 #include "alloc.h"
 #include "recov.h"
+#include "omap.h"
 #include "scoutfs_trace.h"
 
 static struct dentry *scoutfs_debugfs_root;
@@ -264,6 +265,7 @@ static void scoutfs_put_super(struct super_block *sb)
 	scoutfs_recov_destroy(sb);
 	scoutfs_net_destroy(sb);
 	scoutfs_lock_destroy(sb);
+	scoutfs_omap_destroy(sb);
 
 	scoutfs_block_destroy(sb);
 	scoutfs_destroy_triggers(sb);
@@ -593,6 +595,7 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 	      scoutfs_inode_setup(sb) ?:
 	      scoutfs_data_setup(sb) ?:
 	      scoutfs_setup_trans(sb) ?:
+	      scoutfs_omap_setup(sb) ?:
 	      scoutfs_lock_setup(sb) ?:
 	      scoutfs_net_setup(sb) ?:
 	      scoutfs_recov_setup(sb) ?:

kmod/src/super.h

@@ -27,6 +27,7 @@ struct block_info;
 struct forest_info;
 struct srch_info;
 struct recov_info;
+struct omap_info;
 
 struct scoutfs_sb_info {
 	struct super_block *sb;
@@ -49,6 +50,7 @@ struct scoutfs_sb_info {
 	struct block_info *block_info;
 	struct forest_info *forest_info;
 	struct srch_info *srch_info;
+	struct omap_info *omap_info;
 	struct item_cache_info *item_cache_info;
 
 	wait_queue_head_t trans_hold_wq;