diff --git a/kmod/src/Makefile b/kmod/src/Makefile
index ba8dc505..35f9fd07 100644
--- a/kmod/src/Makefile
+++ b/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		\
diff --git a/kmod/src/client.c b/kmod/src/client.c
index 642a0c5d..99c7ba06 100644
--- a/kmod/src/client.c
+++ b/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,
 };
 
 /*
diff --git a/kmod/src/client.h b/kmod/src/client.h
index ae830ef8..c569f038 100644
--- a/kmod/src/client.h
+++ b/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);
diff --git a/kmod/src/dir.c b/kmod/src/dir.c
index 42c70a4a..2223a4ab 100644
--- a/kmod/src/dir.c
+++ b/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"
 
diff --git a/kmod/src/file.c b/kmod/src/file.c
index f35ef039..b1555b08 100644
--- a/kmod/src/file.c
+++ b/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,
diff --git a/kmod/src/format.h b/kmod/src/format.h
index 3a1a4cca..86d10d3d 100644
--- a/kmod/src/format.h
+++ b/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
diff --git a/kmod/src/inode.c b/kmod/src/inode.c
index e0e11e59..97a68aeb 100644
--- a/kmod/src/inode.c
+++ b/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;
 
diff --git a/kmod/src/lock.c b/kmod/src/lock.c
index 7bddb762..f761ba49 100644
--- a/kmod/src/lock.c
+++ b/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);
 
diff --git a/kmod/src/lock.h b/kmod/src/lock.h
index 46dcdf96..5485bb35 100644
--- a/kmod/src/lock.h
+++ b/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 {
diff --git a/kmod/src/omap.c b/kmod/src/omap.c
new file mode 100644
index 00000000..bb3ac8c4
--- /dev/null
+++ b/kmod/src/omap.c
@@ -0,0 +1,1046 @@
+/*
+ * Copyright (C) 2021 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
+ * License v2 as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ */
+#include <linux/kernel.h>
+#include <linux/fs.h>
+#include <linux/slab.h>
+#include <linux/sched.h>
+#include <linux/rhashtable.h>
+#include <linux/rcupdate.h>
+
+#include "format.h"
+#include "counters.h"
+#include "cmp.h"
+#include "inode.h"
+#include "client.h"
+#include "server.h"
+#include "omap.h"
+#include "recov.h"
+#include "scoutfs_trace.h"
+
+/*
+ * As a client removes an inode from its cache with an nlink of 0 it
+ * needs to decide if it is the last client using the inode and should
+ * fully delete all its items.  It needs to know if other mounts still
+ * have the inode in use.
+ *
+ * We need a way to communicate between mounts that an inode is open.
+ * We don't want to pay the synchronous per-file locking round trip
+ * costs associated with per-inode open locks that you'd typically see
+ * in systems to solve this problem.
+ *
+ * Instead clients maintain open bitmaps that cover groups of inodes.
+ * As inodes enter the cache their bit is set, and as the inode is
+ * evicted the bit is cleared.  As an inode is evicted messages are sent
+ * around the cluster to get the current bitmaps for that inode's group
+ * from all active mounts.  If the inode's bit is clear then it can be
+ * deleted.
+ *
+ * We associate the open bitmaps with our cluster locking of inode
+ * groups to cache these open bitmaps.  As long as we have the lock then
+ * nlink can't be changed on any remote mounts.  Specifically, it can't
+ * increase from 0 so any clear bits can gain references on remote
+ * mounts.  As long as we have the lock, all clear bits in the group for
+ * inodes with 0 nlink can be deleted.
+ *
+ * This layer maintains a list of client rids to send messages to.  The
+ * server calls us as clients enter and leave the cluster.    We can't
+ * process requests until all clients are present as a server starts up
+ * so we hook into recovery and delay processing until all previously
+ * existing clients are recovered or fenced.
+ */
+
+struct omap_rid_list {
+	int nr_rids;
+	struct list_head head;
+};
+
+struct omap_rid_entry {
+	struct list_head head;
+	u64 rid;
+};
+
+struct omap_info {
+	/* client */
+	struct rhashtable group_ht;
+
+	/* server */
+	struct rhashtable req_ht;
+	struct llist_head requests;
+	spinlock_t lock;
+	struct omap_rid_list rids;
+	atomic64_t next_req_id;
+};
+
+#define DECLARE_OMAP_INFO(sb, name) \
+	struct omap_info *name = SCOUTFS_SB(sb)->omap_info
+
+/*
+ * The presence of an inode in the inode cache increases the count of
+ * its inode number's position within its lock group.  These structs
+ * track the counts for all the inodes in a lock group and maintain a
+ * bitmap whose bits are set for each non-zero count.
+ *
+ * We don't want to add additional global synchronization of inode cache
+ * maintenance so these are tracked in an rcu hash table.  Once their
+ * total count reaches zero they're removed from the hash and queued for
+ * freeing and readers should ignore them.
+ */
+struct omap_group {
+	struct super_block *sb;
+	struct rhash_head ht_head;
+	struct rcu_head rcu;
+	u64 nr;
+	spinlock_t lock;
+	unsigned int total;
+	unsigned int *counts;
+	__le64 bits[SCOUTFS_OPEN_INO_MAP_LE64S];
+};
+
+#define trace_group(sb, which, group, bit_nr)						\
+do {											\
+	__typeof__(group) _grp = (group);						\
+	__typeof__(bit_nr) _nr = (bit_nr);						\
+											\
+	trace_scoutfs_omap_group_##which(sb, _grp, _grp->nr, _grp->total, _nr,		\
+				        _nr < 0 ? -1 : _grp->counts[_nr]);		\
+} while (0)
+
+/*
+ * Each request is initialized with the rids of currently mounted
+ * clients.  As each responds we remove their rid and send the response
+ * once everyone has contributed.
+ *
+ * The request frequency will typically be low, but in a mass rm -rf
+ * load we will see O(groups * clients) messages flying around.
+ */
+struct omap_request {
+	struct llist_node llnode;
+	struct rhash_head ht_head;
+	struct rcu_head rcu;
+	spinlock_t lock;
+	u64 client_rid;
+	u64 client_id;
+	struct omap_rid_list rids;
+	struct scoutfs_open_ino_map map;
+};
+
+/*
+ * In each inode group cluster lock we store data to track the open ino
+ * map which tracks all the inodes that the cluster lock covers.  When
+ * the version shows that the map is stale we send a request to update
+ * it.
+ */
+struct scoutfs_omap_lock_data {
+	u64 version;
+	bool req_in_flight;
+	wait_queue_head_t waitq;
+	struct scoutfs_open_ino_map map;
+};
+
+static inline void init_rid_list(struct omap_rid_list *list)
+{
+	INIT_LIST_HEAD(&list->head);
+	list->nr_rids = 0;
+}
+
+/*
+ * Negative searches almost never happen.
+ */
+static struct omap_rid_entry *find_rid(struct omap_rid_list *list, u64 rid)
+{
+	struct omap_rid_entry *entry;
+
+	list_for_each_entry(entry, &list->head, head) {
+		if (rid == entry->rid)
+			return entry;
+	}
+
+	return NULL;
+}
+
+static int free_rid(struct omap_rid_list *list, struct omap_rid_entry *entry)
+{
+	int nr;
+
+	list_del(&entry->head);
+	nr = --list->nr_rids;
+
+	kfree(entry);
+	return nr;
+}
+
+static int copy_rids(struct omap_rid_list *to, struct omap_rid_list *from, spinlock_t *from_lock)
+{
+	struct omap_rid_entry *entry;
+	struct omap_rid_entry *src;
+	struct omap_rid_entry *dst;
+	int nr;
+
+	spin_lock(from_lock);
+
+	while (to->nr_rids != from->nr_rids) {
+		nr = from->nr_rids;
+		spin_unlock(from_lock);
+
+		while (to->nr_rids < nr) {
+			entry = kmalloc(sizeof(struct omap_rid_entry), GFP_NOFS);
+			if (!entry)
+				return -ENOMEM;
+
+			list_add_tail(&entry->head, &to->head);
+			to->nr_rids++;
+		}
+
+		while (to->nr_rids > nr) {
+			entry = list_first_entry(&to->head, struct omap_rid_entry, head);
+			list_del(&entry->head);
+			kfree(entry);
+			to->nr_rids--;
+		}
+
+		spin_lock(from_lock);
+	}
+
+	dst = list_first_entry(&to->head, struct omap_rid_entry, head);
+	list_for_each_entry(src, &from->head, head) {
+		dst->rid = src->rid;
+		dst = list_next_entry(dst, head);
+	}
+
+	spin_unlock(from_lock);
+
+	return 0;
+}
+
+static void free_rids(struct omap_rid_list *list)
+{
+	struct omap_rid_entry *entry;
+	struct omap_rid_entry *tmp;
+
+	list_for_each_entry_safe(entry, tmp, &list->head, head) {
+		list_del(&entry->head);
+		kfree(entry);
+	}
+}
+
+static void calc_group_nrs(u64 ino, u64 *group_nr, int *bit_nr)
+{
+	*group_nr = ino >> SCOUTFS_OPEN_INO_MAP_SHIFT;
+	*bit_nr = ino & SCOUTFS_OPEN_INO_MAP_MASK;
+}
+
+static struct omap_group *alloc_group(struct super_block *sb, u64 group_nr)
+{
+	struct omap_group *group;
+
+	BUILD_BUG_ON((sizeof(group->counts[0]) * SCOUTFS_OPEN_INO_MAP_BITS) > PAGE_SIZE);
+
+	group = kzalloc(sizeof(struct omap_group), GFP_NOFS);
+	if (group) {
+		group->sb = sb;
+		group->nr = group_nr;
+		spin_lock_init(&group->lock);
+
+		group->counts = (void *)get_zeroed_page(GFP_NOFS);
+		if (!group->counts) {
+			kfree(group);
+			group = NULL;
+		} else {
+			trace_group(sb, alloc, group, -1);
+		}
+	}
+
+	return group;
+}
+
+static void free_group(struct super_block *sb, struct omap_group *group)
+{
+	trace_group(sb, free, group, -1);
+	free_page((unsigned long)group->counts);
+	kfree(group);
+}
+
+static void free_group_rcu(struct rcu_head *rcu)
+{
+	struct omap_group *group = container_of(rcu, struct omap_group, rcu);
+
+	free_group(group->sb, group);
+}
+
+static const struct rhashtable_params group_ht_params = {
+        .key_len = member_sizeof(struct omap_group, nr),
+        .key_offset = offsetof(struct omap_group, nr),
+        .head_offset = offsetof(struct omap_group, ht_head),
+};
+
+/*
+ * Track an cached inode in its group.  Our increment can be racing with
+ * a final decrement that removes the group from the hash, sets total to
+ * UINT_MAX, and calls rcu free.  We can retry until the dead group is
+ * no longer visible in the hash table and we can insert a new allocated
+ * group.
+ */
+int scoutfs_omap_inc(struct super_block *sb, u64 ino)
+{
+	DECLARE_OMAP_INFO(sb, ominf);
+	struct omap_group *group;
+	u64 group_nr;
+	int bit_nr;
+	bool found;
+	int ret = 0;
+
+	calc_group_nrs(ino, &group_nr, &bit_nr);
+
+retry:
+	found = false;
+	rcu_read_lock();
+	group = rhashtable_lookup(&ominf->group_ht, &group_nr, group_ht_params);
+	if (group) {
+		spin_lock(&group->lock);
+		if (group->total < UINT_MAX) {
+			found = true;
+			if (group->counts[bit_nr]++ == 0) {
+				set_bit_le(bit_nr, group->bits);
+				group->total++;
+			}
+		}
+		trace_group(sb, inc, group, bit_nr);
+		spin_unlock(&group->lock);
+	}
+	rcu_read_unlock();
+
+	if (!found) {
+		group = alloc_group(sb, group_nr);
+		if (group) {
+			ret = rhashtable_lookup_insert_fast(&ominf->group_ht, &group->ht_head,
+							    group_ht_params);
+			if (ret < 0)
+				free_group(sb, group);
+			if (ret == -EEXIST)
+				ret = 0;
+			if (ret == -EBUSY) {
+				/* wait for rehash to finish */
+				synchronize_rcu();
+				ret = 0;
+			}
+			if (ret == 0)
+				goto retry;
+		} else {
+			ret = -ENOMEM;
+		}
+	}
+
+	return ret;
+}
+
+/*
+ * Decrement a previously incremented ino count.  Not finding a count
+ * implies imbalanced inc/dec or bugs freeing groups.  We only free
+ * groups here as the last dec drops the group's total count to 0.
+ */
+void scoutfs_omap_dec(struct super_block *sb, u64 ino)
+{
+	DECLARE_OMAP_INFO(sb, ominf);
+	struct omap_group *group;
+	u64 group_nr;
+	int bit_nr;
+
+	calc_group_nrs(ino, &group_nr, &bit_nr);
+
+	rcu_read_lock();
+	group = rhashtable_lookup(&ominf->group_ht, &group_nr, group_ht_params);
+	if (group) {
+		spin_lock(&group->lock);
+		WARN_ON_ONCE(group->counts[bit_nr] == 0);
+		WARN_ON_ONCE(group->total == 0);
+		WARN_ON_ONCE(group->total == UINT_MAX);
+		if (--group->counts[bit_nr] == 0) {
+			clear_bit_le(bit_nr, group->bits);
+			if (--group->total == 0) {
+				group->total = UINT_MAX;
+				rhashtable_remove_fast(&ominf->group_ht, &group->ht_head,
+						       group_ht_params);
+				call_rcu(&group->rcu, free_group_rcu);
+			}
+		}
+		trace_group(sb, dec, group, bit_nr);
+		spin_unlock(&group->lock);
+	}
+	rcu_read_unlock();
+
+	WARN_ON_ONCE(!group);
+}
+
+/*
+ * The server adds rids as it discovers clients.  We add them to the
+ * list of rids to send map requests to.
+ */
+int scoutfs_omap_add_rid(struct super_block *sb, u64 rid)
+{
+	DECLARE_OMAP_INFO(sb, ominf);
+	struct omap_rid_entry *entry;
+	struct omap_rid_entry *found;
+
+	entry = kmalloc(sizeof(struct omap_rid_entry), GFP_NOFS);
+	if (!entry)
+		return -ENOMEM;
+
+	spin_lock(&ominf->lock);
+	found = find_rid(&ominf->rids, rid);
+	if (!found) {
+		entry->rid = rid;
+		list_add_tail(&entry->head, &ominf->rids.head);
+		ominf->rids.nr_rids++;
+	}
+	spin_unlock(&ominf->lock);
+
+	if (found)
+		kfree(entry);
+
+	return 0;
+}
+
+static void free_req(struct omap_request *req)
+{
+	free_rids(&req->rids);
+	kfree(req);
+}
+
+static void free_req_rcu(struct rcu_head *rcu)
+{
+	struct omap_request *req = container_of(rcu, struct omap_request, rcu);
+
+	free_req(req);
+}
+
+static const struct rhashtable_params req_ht_params = {
+        .key_len = member_sizeof(struct omap_request, map.args.req_id),
+        .key_offset = offsetof(struct omap_request, map.args.req_id),
+        .head_offset = offsetof(struct omap_request, ht_head),
+};
+
+/*
+ * Remove a rid from all the pending requests.  If it's the last rid we
+ * give the caller the details to send a response, they'll call back to
+ * keep removing.  If their send fails they're going to shutdown the
+ * server so we can queue freeing the request as we give it to them.
+ */
+static int remove_rid_from_reqs(struct omap_info *ominf, u64 rid, u64 *resp_rid, u64 *resp_id,
+				struct scoutfs_open_ino_map *map)
+{
+	struct omap_rid_entry *entry;
+	struct rhashtable_iter iter;
+	struct omap_request *req;
+	int ret = 0;
+
+	rhashtable_walk_enter(&ominf->req_ht, &iter);
+	rhashtable_walk_start(&iter);
+
+	for (;;) {
+		req = rhashtable_walk_next(&iter);
+		if (req == NULL)
+			break;
+		if (req == ERR_PTR(-EAGAIN))
+			continue;
+
+		spin_lock(&req->lock);
+		entry = find_rid(&req->rids, rid);
+		if (entry && free_rid(&req->rids, entry) == 0) {
+			*resp_rid = req->client_rid;
+			*resp_id = req->client_id;
+			memcpy(map, &req->map, sizeof(struct scoutfs_open_ino_map));
+			rhashtable_remove_fast(&ominf->req_ht, &req->ht_head, req_ht_params);
+			call_rcu(&req->rcu, free_req_rcu);
+			ret = 1;
+		}
+		spin_unlock(&req->lock);
+		if (ret > 0)
+			break;
+	}
+
+	rhashtable_walk_stop(&iter);
+	rhashtable_walk_exit(&iter);
+
+	if (ret <= 0) {
+		*resp_rid = 0;
+		*resp_id = 0;
+	}
+
+	return ret;
+}
+
+/*
+ * A client has been evicted.  Remove its rid from the list and walk
+ * through all the pending requests and remove its rids, sending the
+ * response if it was the last rid waiting for a response.
+ *
+ * If this returns an error then the server will shut down.
+ */
+int scoutfs_omap_remove_rid(struct super_block *sb, u64 rid)
+{
+	DECLARE_OMAP_INFO(sb, ominf);
+	struct scoutfs_open_ino_map *map = NULL;
+	struct omap_rid_entry *entry;
+	u64 resp_rid = 0;
+	u64 resp_id = 0;
+	int ret;
+
+	map = kmalloc(sizeof(struct scoutfs_open_ino_map), GFP_NOFS);
+	if (!map) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
+	spin_lock(&ominf->lock);
+	entry = find_rid(&ominf->rids, rid);
+	if (entry)
+		free_rid(&ominf->rids, entry);
+	spin_unlock(&ominf->lock);
+
+	/* the server really shouldn't be removing a rid it never added */
+	if (WARN_ON_ONCE(!entry)) {
+		ret = -ENOENT;
+		goto out;
+	}
+
+	/* remove the rid from all pending requests, sending responses if it was final */
+	for (;;) {
+		ret = remove_rid_from_reqs(ominf, rid, &resp_rid, &resp_id, map);
+		if (ret <= 0)
+			break;
+		ret = scoutfs_server_send_omap_response(sb, resp_rid, resp_id, map, 0);
+		if (ret < 0)
+			break;
+	}
+
+out:
+	kfree(map);
+	return ret;
+}
+
+/*
+ * Handle a single incoming request in the server.  This could have been
+ * delayed by recovery.  This only returns an error if we couldn't send
+ * a processing error response to the client.
+ */
+static int handle_request(struct super_block *sb, struct omap_request *req)
+{
+	DECLARE_OMAP_INFO(sb, ominf);
+	struct omap_rid_list priv_rids;
+	struct omap_rid_entry *entry;
+	int ret;
+
+	init_rid_list(&priv_rids);
+
+	ret = copy_rids(&priv_rids, &ominf->rids, &ominf->lock);
+	if (ret < 0)
+		goto out;
+
+	/* don't send a request to the client who originated this request */
+	entry = find_rid(&priv_rids, req->client_rid);
+	if (entry && free_rid(&priv_rids, entry) == 0) {
+		ret = scoutfs_server_send_omap_response(sb, req->client_rid, req->client_id,
+							&req->map, 0);
+		kfree(req);
+		req = NULL;
+		goto out;
+	}
+
+	/* this lock isn't needed but sparse gave warnings with conditional locking */
+	ret = copy_rids(&req->rids, &priv_rids, &ominf->lock);
+	if (ret < 0)
+		goto out;
+
+	do {
+		ret = rhashtable_insert_fast(&ominf->req_ht, &req->ht_head, req_ht_params);
+		if (ret == -EBUSY)
+			synchronize_rcu(); /* wait for rehash to finish */
+	} while (ret == -EBUSY);
+
+	if (ret < 0)
+		goto out;
+
+	/*
+	 * We can start getting responses the moment we send the first response.  After
+	 * we send the last request the req can be freed.
+	 */
+	while ((entry = list_first_entry_or_null(&priv_rids.head, struct omap_rid_entry, head))) {
+		ret = scoutfs_server_send_omap_request(sb, entry->rid, &req->map.args);
+		if (ret < 0) {
+			rhashtable_remove_fast(&ominf->req_ht, &req->ht_head, req_ht_params);
+			goto out;
+		}
+
+		free_rid(&priv_rids, entry);
+	}
+
+	ret = 0;
+out:
+	free_rids(&priv_rids);
+	if (ret < 0) {
+		ret = scoutfs_server_send_omap_response(sb, req->client_rid, req->client_id,
+							NULL, ret);
+		free_req(req);
+	}
+
+	/* it's fine if we couldn't send to a client that left */
+	if (ret == -ENOTCONN)
+		ret = 0;
+
+	return ret;
+}
+
+/*
+ * Handle all previously received omap requests from clients.  Once
+ * we've finished recovery and can send requests to all clients we can
+ * handle all pending requests.  The handling function frees the request
+ * and only returns an error if it couldn't send a response to the
+ * client.
+ */
+static int handle_requests(struct super_block *sb)
+{
+	DECLARE_OMAP_INFO(sb, ominf);
+	struct llist_node *requests;
+	struct omap_request *req;
+	struct omap_request *tmp;
+	int ret;
+	int err;
+
+	if (scoutfs_recov_next_pending(sb, SCOUTFS_RECOV_GREETING))
+		return 0;
+
+	ret = 0;
+	requests = llist_del_all(&ominf->requests);
+
+	llist_for_each_entry_safe(req, tmp, requests, llnode) {
+		err = handle_request(sb, req);
+		if (err < 0 && ret == 0)
+			ret = err;
+	}
+
+	return ret;
+}
+
+int scoutfs_omap_finished_recovery(struct super_block *sb)
+{
+	return handle_requests(sb);
+}
+
+/*
+ * The server is receiving a request from a client for the bitmap of all
+ * open inodes around their ino.  Queue it for processing which is
+ * typically immediate and inline but which can be deferred by recovery
+ * as the server first starts up.
+ */
+int scoutfs_omap_server_handle_request(struct super_block *sb, u64 rid, u64 id,
+				       struct scoutfs_open_ino_map_args *args)
+{
+	DECLARE_OMAP_INFO(sb, ominf);
+	struct omap_request *req;
+
+	req = kzalloc(sizeof(struct omap_request), GFP_NOFS);
+	if (req == NULL)
+		return -ENOMEM;
+
+	spin_lock_init(&req->lock);
+	req->client_rid = rid;
+	req->client_id = id;
+	init_rid_list(&req->rids);
+	req->map.args.group_nr = args->group_nr;
+	req->map.args.req_id = cpu_to_le64(atomic64_inc_return(&ominf->next_req_id));
+
+	llist_add(&req->llnode, &ominf->requests);
+
+	return handle_requests(sb);
+}
+
+/*
+ * The client is receiving a request from the server for its map for the
+ * given group.  Look up the group and copy the bits to the map for
+ * non-zero open counts.
+ *
+ * The mount originating the request for this bitmap has the inode group
+ * write locked.  We can't be adding links to any inodes in the group
+ * because that requires the lock.  Inodes bits can be set and cleared
+ * while we're sampling the bitmap.  These races are fine, they can't be
+ * adding cached inodes if nlink is 0 and we don't have the lock.  If
+ * the caller is removing a set bit then they're about to try and delete
+ * the inode themselves and will first have to acquire the cluster lock
+ * themselves.
+ */
+int scoutfs_omap_client_handle_request(struct super_block *sb, u64 id,
+				       struct scoutfs_open_ino_map_args *args)
+{
+	DECLARE_OMAP_INFO(sb, ominf);
+	u64 group_nr = le64_to_cpu(args->group_nr);
+	struct scoutfs_open_ino_map *map;
+	struct omap_group *group;
+	bool copied = false;
+	int ret;
+
+	map = kmalloc(sizeof(struct scoutfs_open_ino_map), GFP_NOFS);
+	if (!map)
+		return -ENOMEM;
+
+	map->args = *args;
+
+	rcu_read_lock();
+	group = rhashtable_lookup(&ominf->group_ht, &group_nr, group_ht_params);
+	if (group) {
+		spin_lock(&group->lock);
+		trace_group(sb, request, group, -1);
+		if (group->total > 0 && group->total < UINT_MAX) {
+			memcpy(map->bits, group->bits, sizeof(map->bits));
+			copied = true;
+		}
+		spin_unlock(&group->lock);
+	}
+	rcu_read_unlock();
+
+	if (!copied)
+		memset(map->bits, 0, sizeof(map->bits));
+
+	ret = scoutfs_client_send_omap_response(sb, id, map);
+	kfree(map);
+	return ret;
+}
+
+/*
+ * The server has received an open ino map response from a client.  Find
+ * the original request that it's serving, or in the response's map, and
+ * send a reply if this was the last response from a client we were
+ * waiting for.
+ *
+ * We can get responses for requests we're no longer tracking if, for
+ * example, sending to a client gets an error.  We'll have already sent
+ * the response to the requesting client so we drop these responses on
+ * the floor.
+ */
+int scoutfs_omap_server_handle_response(struct super_block *sb, u64 rid,
+					struct scoutfs_open_ino_map *resp_map)
+{
+	DECLARE_OMAP_INFO(sb, ominf);
+	struct scoutfs_open_ino_map *map;
+	struct omap_rid_entry *entry;
+	bool send_response = false;
+	struct omap_request *req;
+	u64 resp_rid;
+	u64 resp_id;
+	int ret;
+
+	map = kmalloc(sizeof(struct scoutfs_open_ino_map), GFP_NOFS);
+	if (!map) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
+	rcu_read_lock();
+	req = rhashtable_lookup(&ominf->req_ht, &resp_map->args.req_id, req_ht_params);
+	if (req) {
+		spin_lock(&req->lock);
+		entry = find_rid(&req->rids, rid);
+		if (entry) {
+			bitmap_or((unsigned long *)req->map.bits, (unsigned long *)req->map.bits,
+				  (unsigned long *)resp_map->bits, SCOUTFS_OPEN_INO_MAP_BITS);
+			if (free_rid(&req->rids, entry) == 0)
+				send_response = true;
+		}
+		spin_unlock(&req->lock);
+
+		if (send_response) {
+			resp_rid = req->client_rid;
+			resp_id = req->client_id;
+			memcpy(map, &req->map, sizeof(struct scoutfs_open_ino_map));
+			rhashtable_remove_fast(&ominf->req_ht, &req->ht_head, req_ht_params);
+			call_rcu(&req->rcu, free_req_rcu);
+		}
+	}
+	rcu_read_unlock();
+
+	if (send_response)
+		ret = scoutfs_server_send_omap_response(sb, resp_rid, resp_id, map, 0);
+	else
+		ret = 0;
+	kfree(map);
+out:
+	return ret;
+}
+
+/*
+ * The server is shutting down.  Free all the server state associated
+ * with ongoing request processing.  Clients who still have requests
+ * pending will resend them to the next server.
+ */
+void scoutfs_omap_server_shutdown(struct super_block *sb)
+{
+	DECLARE_OMAP_INFO(sb, ominf);
+	struct rhashtable_iter iter;
+	struct llist_node *requests;
+	struct omap_request *req;
+	struct omap_request *tmp;
+
+	rhashtable_walk_enter(&ominf->req_ht, &iter);
+	rhashtable_walk_start(&iter);
+
+	for (;;) {
+		req = rhashtable_walk_next(&iter);
+		if (req == NULL)
+			break;
+		if (req == ERR_PTR(-EAGAIN))
+			continue;
+
+		if (req->rids.nr_rids != 0) {
+			free_rids(&req->rids);
+			rhashtable_remove_fast(&ominf->req_ht, &req->ht_head, req_ht_params);
+			call_rcu(&req->rcu, free_req_rcu);
+		}
+	}
+
+	rhashtable_walk_stop(&iter);
+	rhashtable_walk_exit(&iter);
+
+	requests = llist_del_all(&ominf->requests);
+	llist_for_each_entry_safe(req, tmp, requests, llnode)
+		kfree(req);
+
+	synchronize_rcu();
+}
+
+static bool omap_req_in_flight(struct scoutfs_lock *lock, struct scoutfs_omap_lock_data *ldata)
+{
+	bool in_flight;
+
+	spin_lock(&lock->omap_spinlock);
+	in_flight = ldata->req_in_flight;
+	spin_unlock(&lock->omap_spinlock);
+
+	return in_flight;
+}
+
+/*
+ * Make sure the map covered by the cluster lock is current.  The caller
+ * holds the cluster lock so once we store lock_data on the cluster lock
+ * it won't be freed and the write_version in the cluster lock won't
+ * change.
+ *
+ * The omap_spinlock protects the omap_data in the cluster lock.  We
+ * have to drop it if we have to block to allocate lock_data, send a
+ * request for a new map, or wait for a request in flight to finish.
+ */
+static int get_current_lock_data(struct super_block *sb, struct scoutfs_lock *lock,
+				 struct scoutfs_omap_lock_data **ldata_ret, u64 group_nr)
+{
+	struct scoutfs_omap_lock_data *ldata;
+	bool send_req;
+	int ret = 0;
+
+	spin_lock(&lock->omap_spinlock);
+
+	ldata = lock->omap_data;
+	if (ldata == NULL) {
+		spin_unlock(&lock->omap_spinlock);
+		ldata = kzalloc(sizeof(struct scoutfs_omap_lock_data), GFP_NOFS);
+		spin_lock(&lock->omap_spinlock);
+
+		if (!ldata) {
+			ret = -ENOMEM;
+			goto out;
+		}
+
+		if (lock->omap_data == NULL) {
+			ldata->version = lock->write_version - 1; /* ensure refresh */
+			init_waitqueue_head(&ldata->waitq);
+
+			lock->omap_data = ldata;
+		} else {
+			kfree(ldata);
+			ldata = lock->omap_data;
+		}
+	}
+
+	while (ldata->version != lock->write_version) {
+		/* only one waiter sends a request at a time */
+		if (!ldata->req_in_flight) {
+			ldata->req_in_flight = true;
+			send_req = true;
+		} else {
+			send_req = false;
+		}
+
+		spin_unlock(&lock->omap_spinlock);
+		if (send_req)
+			ret = scoutfs_client_open_ino_map(sb, group_nr, &ldata->map);
+		else
+			wait_event(ldata->waitq, !omap_req_in_flight(lock, ldata));
+		spin_lock(&lock->omap_spinlock);
+
+		/* only sender can return error, other waiters retry */
+		if (send_req) {
+			ldata->req_in_flight = false;
+			if (ret == 0)
+				ldata->version = lock->write_version;
+			wake_up(&ldata->waitq);
+			if (ret < 0)
+				goto out;
+		}
+	}
+
+out:
+	spin_unlock(&lock->omap_spinlock);
+
+	if (ret == 0)
+		*ldata_ret = ldata;
+	else
+		*ldata_ret = NULL;
+
+	return ret;
+}
+
+/*
+ * Return 1 and give the caller a write inode lock if it is safe to be
+ * deleted.  It's safe to be deleted when it is no longer reachable and
+ * nothing is referencing it.
+ *
+ * The inode is unreachable when nlink hits zero.  Cluster locks protect
+ * modification and testing of nlink.  We use the ino_lock_cov covrage
+ * to short circuit the common case of having a locked inode that hasn't
+ * been deleted.  If it isn't locked, we have to acquire the lock to
+ * refresh the inode to see its current nlink. 
+ *
+ * Then we use an open inode bitmap that covers all the inodes in the
+ * lock group to determine if the inode is present in any other mount's
+ * caches.  We refresh it by asking the server for all clients' maps and
+ * then store it in the lock.  As long as we hold the lock nothing can
+ * increase nlink from zero and let people get a reference to the inode.
+ */
+int scoutfs_omap_should_delete(struct super_block *sb, struct inode *inode,
+			       struct scoutfs_lock **lock_ret)
+{
+	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
+	struct scoutfs_lock *lock = NULL;
+	const u64 ino = scoutfs_ino(inode);
+	struct scoutfs_omap_lock_data *ldata;
+	u64 group_nr;
+	int bit_nr;
+	int ret;
+
+	/* lock group and omap constants are defined independently */
+	BUILD_BUG_ON(SCOUTFS_OPEN_INO_MAP_BITS != SCOUTFS_LOCK_INODE_GROUP_NR);
+
+	if (scoutfs_lock_is_covered(sb, &si->ino_lock_cov) && inode->i_nlink > 0) {
+		ret = 0;
+		goto out;
+	}
+
+	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_WRITE, SCOUTFS_LKF_REFRESH_INODE, inode, &lock);
+	if (ret < 0)
+		goto out;
+
+	if (inode->i_nlink > 0) {
+		ret = 0;
+		goto out;
+	}
+
+	calc_group_nrs(ino, &group_nr, &bit_nr);
+
+	/* only one request to refresh the map at a time */
+	ret = get_current_lock_data(sb, lock, &ldata, group_nr);
+	if (ret < 0)
+		goto out;
+
+	/* can delete caller's zero nlink inode if it's not cached in other mounts */
+	ret = !test_bit_le(bit_nr, ldata->map.bits);
+out:
+	trace_scoutfs_omap_should_delete(sb, ino, inode->i_nlink, ret);
+
+	if (ret <= 0) {
+		scoutfs_unlock(sb, lock, SCOUTFS_LOCK_WRITE);
+		lock = NULL;
+	}
+
+	*lock_ret = lock;
+	return ret;
+}
+
+void scoutfs_omap_free_lock_data(struct scoutfs_omap_lock_data *ldata)
+{
+	if (ldata) {
+		WARN_ON_ONCE(ldata->req_in_flight);
+		WARN_ON_ONCE(waitqueue_active(&ldata->waitq));
+		kfree(ldata);
+	}
+}
+
+int scoutfs_omap_setup(struct super_block *sb)
+{
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	struct omap_info *ominf;
+	int ret;
+
+	ominf = kzalloc(sizeof(struct omap_info), GFP_KERNEL);
+	if (!ominf) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
+	ret = rhashtable_init(&ominf->group_ht, &group_ht_params);
+	if (ret < 0) {
+		kfree(ominf);
+		goto out;
+	}
+
+	ret = rhashtable_init(&ominf->req_ht, &req_ht_params);
+	if (ret < 0) {
+		rhashtable_destroy(&ominf->group_ht);
+		kfree(ominf);
+		goto out;
+	}
+
+	init_llist_head(&ominf->requests);
+	spin_lock_init(&ominf->lock);
+	init_rid_list(&ominf->rids);
+	atomic64_set(&ominf->next_req_id, 0);
+
+	sbi->omap_info = ominf;
+	ret = 0;
+out:
+	return ret;
+}
+
+/*
+ * To get here the server must have shut down, freeing requests, and
+ * evict must have been called on all cached inodes so we can just
+ * synchronize all the pending group frees.
+ */
+void scoutfs_omap_destroy(struct super_block *sb)
+{
+	DECLARE_OMAP_INFO(sb, ominf);
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	struct rhashtable_iter iter;
+
+	if (ominf) {
+		synchronize_rcu();
+
+		/* double check that all the groups deced to 0 and were freed */
+		rhashtable_walk_enter(&ominf->group_ht, &iter);
+		rhashtable_walk_start(&iter);
+		WARN_ON_ONCE(rhashtable_walk_peek(&iter) != NULL);
+		rhashtable_walk_stop(&iter);
+		rhashtable_walk_exit(&iter);
+
+		rhashtable_destroy(&ominf->group_ht);
+		rhashtable_destroy(&ominf->req_ht);
+		kfree(ominf);
+		sbi->omap_info = NULL;
+	}
+}
diff --git a/kmod/src/omap.h b/kmod/src/omap.h
new file mode 100644
index 00000000..0e5fa29d
--- /dev/null
+++ b/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
diff --git a/kmod/src/scoutfs_trace.h b/kmod/src/scoutfs_trace.h
index 91504e9c..8d1209e7 100644
--- a/kmod/src/scoutfs_trace.h
+++ b/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 */
diff --git a/kmod/src/server.c b/kmod/src/server.c
index 0210dbe9..369b2797 100644
--- a/kmod/src/server.c
+++ b/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);
diff --git a/kmod/src/server.h b/kmod/src/server.h
index e06c0818..8d31a271 100644
--- a/kmod/src/server.h
+++ b/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);
diff --git a/kmod/src/super.c b/kmod/src/super.c
index e66160ed..27b2f832 100644
--- a/kmod/src/super.c
+++ b/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) ?:
diff --git a/kmod/src/super.h b/kmod/src/super.h
index 8d004db3..a74166bf 100644
--- a/kmod/src/super.h
+++ b/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;