Add rpm spec file support for el8 builds

The rpmbuild support files no longer define the previously used kernel
module macros. This carves out the differences between el7 and el8 with
conditionals based on the distro we are building for.

Signed-off-by: Ben McClelland <ben.mcclelland@versity.com>

ReleaseNotes.md

@@ -1,6 +1,67 @@
 Versity ScoutFS Release Notes
 =============================
 
+---
+v1.15
+\
+*Jul 17, 2023*
+
+Process log btree merge splicing in multiple commits.  This prevents a
+rare case where pending log merge completions contain more work than can
+be done in a single server commit, causing the server to trigger an
+assert shortly after starting.
+
+Fix spurious EINVAL from data writes when data\_prealloc\_contig\_only was
+set to 0.
+
+---
+v1.14
+\
+*Jun 29, 2023*
+
+Add get\_referring\_entries ioctl for getting directory entries that
+refer to an inode.
+
+Fix excessive CPU use in the move\_blocks interface when moving a large
+number of extents.
+
+Reduce fragmented data allocation when contig\_only prealloc is not in
+use by more consistently allocating multi-block extents within each
+aligned prealloc region.
+
+Avoid rare deadlock in metadata block cache recalim under both heavy
+load and memory pressure.
+
+Fix crash when using quorum\_heartbeat\_timeout\_ms mount option.
+
+---
+v1.13
+\
+*May 19, 2023*
+
+Add the quorum\_heartbeat\_timeout\_ms mount option to set the quorum
+heartbeat timeout.
+
+Change some task prioritization and allocation behavior of the quorum
+agent to help reduce delays in sending and receiving heartbeat messages.
+
+---
+v1.12
+\
+*Apr 17, 2023*
+
+Add the prepare-empty-data-device scoutfs command.  A data device can be
+unused when no files have data blocks, perhaps because they're archived
+and offline.  In this case the data device can be swapped out for
+another device without changes to the metadata device.
+
+Fix an oversight which limited inode timestamps to second granularity
+for some operations.  All operations now record timestamps with full
+nanosecond precision.
+
+Fix spurious ENOENT failures when renaming from other directories into
+the root directory.
+
 ---
 v1.11
 \

kmod/Makefile

@@ -31,12 +31,12 @@ TARFILE = scoutfs-kmod-$(RPM_VERSION).tar
 all: module
 
 module:
-	make $(SCOUTFS_ARGS)
-	$(SP) make C=2 CF="-D__CHECK_ENDIAN__" $(SCOUTFS_ARGS)
+	$(MAKE) $(SCOUTFS_ARGS)
+	$(SP) $(MAKE) C=2 CF="-D__CHECK_ENDIAN__" $(SCOUTFS_ARGS)
 
 
 modules_install:
-	make $(SCOUTFS_ARGS) modules_install
+	$(MAKE) $(SCOUTFS_ARGS) modules_install
 
 
 %.spec: %.spec.in .FORCE
@@ -50,4 +50,4 @@ dist: scoutfs-kmod.spec
 	@ tar rf $(TARFILE) --transform="s@\(.*\)@scoutfs-kmod-$(RPM_VERSION)/\1@" scoutfs-kmod.spec
 
 clean:
-	make $(SCOUTFS_ARGS) clean
+	$(MAKE) $(SCOUTFS_ARGS) clean

kmod/scoutfs-kmod.spec.in

@@ -3,16 +3,28 @@
 %define kmod_git_hash @@GITHASH@@
 %define pkg_date %(date +%%Y%%m%%d)
 
+# Disable the building of the debug package(s).
+%define debug_package %{nil}
+
 # take kernel version or default to uname -r
 %{!?kversion: %global kversion %(uname -r)}
 %global kernel_version %{kversion}
 
+%if 0%{?el7}
 %global kernel_source() /usr/src/kernels/%{kernel_version}.$(arch)
-%global kernel_release() %{kversion}
+%endif
+%if 0%{?el8}
+%global kernel_source() /usr/src/kernels/%{kernel_version}
+%endif
 
 %{!?_release: %global _release 0.%{pkg_date}git%{kmod_git_hash}}
 
+%if 0%{?el7}
 Name:           %{kmod_name}
+%endif
+%if 0%{?el8}
+Name:           kmod-%{kmod_name}
+%endif
 Summary:        %{kmod_name} kernel module
 Version:        %{kmod_version}
 Release:        %{_release}%{?dist}
@@ -20,24 +32,30 @@ License:        GPLv2
 Group:          System/Kernel
 URL:            http://scoutfs.org/
 
+%if 0%{?el7}
 BuildRequires:  %{kernel_module_package_buildreqs}
-BuildRequires:  git
+%endif
+%if 0%{?el8}
+BuildRequires:  elfutils-libelf-devel
+%endif
 BuildRequires:  kernel-devel-uname-r = %{kernel_version}
+BuildRequires:  git
 BuildRequires:  module-init-tools
 
 ExclusiveArch:  x86_64
 
 Source:		%{kmod_name}-kmod-%{kmod_version}.tar
 
+%if 0%{?el7}
 # Build only for standard kernel variant(s); for debug packages, append "debug"
 # after "default" (separated by space)
 %kernel_module_package default
+%endif
 
-# Disable the building of the debug package(s).
-%define debug_package %{nil}
-
-%global install_mod_dir extra/%{name}
-
+%global install_mod_dir extra/%{kmod_name}
+%if 0%{?el8}
+%global flavors_to_build x86_64
+%endif
 
 %description
 %{kmod_name} - kernel module
@@ -66,7 +84,7 @@ export INSTALL_MOD_DIR=%{install_mod_dir}
 mkdir -p %{install_mod_dir}
 for flavor in %{flavors_to_build}; do
 	export KSRC=%{kernel_source $flavor}
-	export KVERSION=%{kernel_release $KSRC}
+	export KVERSION=%{kversion}
 	install -d $INSTALL_MOD_PATH/lib/modules/$KVERSION/%{install_mod_dir}
 	cp $PWD/obj/$flavor/src/scoutfs.ko $INSTALL_MOD_PATH/lib/modules/$KVERSION/%{install_mod_dir}/
 done
@@ -74,6 +92,14 @@ done
 # mark modules executable so that strip-to-file can strip them
 find %{buildroot} -type f -name \*.ko -exec %{__chmod} u+x \{\} \;
 
+%if 0%{?el8}
+%files
+/lib/modules
+
+%post
+weak-modules --add-kernel --no-initramfs
+depmod -a
+%endif
 
 %clean
 rm -rf %{buildroot}

kmod/src/Makefile

@@ -25,6 +25,7 @@ scoutfs-y +=			\
 	inode.o			\
 	ioctl.o			\
 	item.o			\
+	kernelcompat.o		\
 	lock.o			\
 	lock_server.o		\
 	msg.o			\

kmod/src/Makefile.kernelcompat

@@ -26,6 +26,16 @@ ifneq (,$(shell grep 'dir_emit_dots' include/linux/fs.h))
 ccflags-y += -DKC_DIR_EMIT_DOTS
 endif
 
+#
+# v3.18-rc2-19-gb5ae6b15bd73
+# 
+# Folds d_materialise_unique into d_splice_alias. Note reversal
+# of arguments (Also note Documentation/filesystems/porting.rst)
+#
+ifneq (,$(shell grep 'd_materialise_unique' include/linux/dcache.h))
+ccflags-y += -DKC_D_MATERIALISE_UNIQUE=1
+endif
+
 #
 # RHEL extended the fop struct so to use it we have to set
 # a flag to indicate that the struct is large enough and
@@ -40,6 +50,211 @@ endif
 #
 # Added user_ns argument to posix_acl_valid
 #
-ifneq (,$(shell grep 'posix_acl_valid.*user_ns,' include/linux/posix_acl.h))
+ifneq (,$(shell grep 'posix_acl_valid.*user_namespace' include/linux/posix_acl.h))
 ccflags-y += -DKC_POSIX_ACL_VALID_USER_NS
 endif
+
+#
+# v5.3-12296-g6d2052d188d9
+#
+# The RBCOMPUTE function is now passed an extra flag, and should return a bool
+# to indicate whether the propagated callback should stop or not.
+#
+ifneq (,$(shell grep 'static inline bool RBNAME.*_compute_max' include/linux/rbtree_augmented.h))
+ccflags-y += -DKC_RB_TREE_AUGMENTED_COMPUTE_MAX
+endif
+
+#
+# v3.13-25-g37bc15392a23
+#
+# Renames posix_acl_create to __posix_acl_create and provide some
+# new interfaces for creating ACLs
+#
+ifneq (,$(shell grep '__posix_acl_create' include/linux/posix_acl.h))
+ccflags-y += -DKC___POSIX_ACL_CREATE
+endif
+
+#
+# v4.8-rc1-29-g31051c85b5e2
+#
+# inode_change_ok() removed - replace with setattr_prepare()
+#
+ifneq (,$(shell grep 'extern int setattr_prepare' include/linux/fs.h))
+ccflags-y += -DKC_SETATTR_PREPARE
+endif
+
+#
+# v4.15-rc3-4-gae5e165d855d
+#
+# linux/iversion.h needs to manually be included for code that
+# manipulates this field.
+#
+ifneq (,$(shell grep -s 'define _LINUX_IVERSION_H' include/linux/iversion.h))
+ccflags-y += -DKC_NEED_LINUX_IVERSION_H=1
+endif
+
+# v4.11-12447-g104b4e5139fe
+#
+# Renamed __percpu_counter_add to percpu_counter_add_batch to clarify
+# that the __ wasn't less safe, just took an extra parameter.
+#
+ifneq (,$(shell grep 'percpu_counter_add_batch' include/linux/percpu_counter.h))
+ccflags-y += -DKC_PERCPU_COUNTER_ADD_BATCH
+endif
+
+#
+# v4.11-4550-g7dea19f9ee63
+#
+# Introduced memalloc_nofs_{save,restore} preferred instead of _noio_.
+#
+ifneq (,$(shell grep 'memalloc_nofs_save' include/linux/sched/mm.h))
+ccflags-y += -DKC_MEMALLOC_NOFS_SAVE
+endif
+
+#
+# v4.7-12414-g1eff9d322a44
+#
+# Renamed bi_rw to bi_opf to force old code to catch up.  We use it as a
+# single switch between old and new bio structures.
+#
+ifneq (,$(shell grep 'bi_opf' include/linux/blk_types.h))
+ccflags-y += -DKC_BIO_BI_OPF
+endif
+
+#
+# v4.12-rc2-201-g4e4cbee93d56
+#
+# Moves to bi_status BLK_STS_ API instead of having a mix of error
+# end_io args or bi_error.
+#
+ifneq (,$(shell grep 'bi_status' include/linux/blk_types.h))
+ccflags-y += -DKC_BIO_BI_STATUS
+endif
+
+#
+# v3.11-8765-ga0b02131c5fc
+#
+# Remove the old ->shrink() API, ->{scan,count}_objects is preferred.
+#
+ifneq (,$(shell grep '(*shrink)' include/linux/shrinker.h))
+ccflags-y += -DKC_SHRINKER_SHRINK
+endif
+
+#
+# v3.19-4777-g6bec00352861
+#
+# backing_dev_info is removed from address_space. Instead we need to use
+# inode_to_bdi() inline from <backing-dev.h>.
+#
+ifneq (,$(shell grep 'struct backing_dev_info.*backing_dev_info' include/linux/fs.h))
+ccflags-y += -DKC_LINUX_BACKING_DEV_INFO=1
+endif
+
+#
+# v4.3-9290-ge409de992e3e
+#
+# xattr handlers are now passed a struct that contains `flags`
+#
+ifneq (,$(shell grep 'int...get..const struct xattr_handler.*struct dentry.*dentry,' include/linux/xattr.h))
+ccflags-y += -DKC_XATTR_STRUCT_XATTR_HANDLER=1
+endif
+
+#
+# v4.16-rc1-1-g9b2c45d479d0
+#
+# kernel_getsockname() and kernel_getpeername dropped addrlen arg
+#
+ifneq (,$(shell grep 'kernel_getsockname.*,$$' include/linux/net.h))
+ccflags-y += -DKC_KERNEL_GETSOCKNAME_ADDRLEN=1
+endif
+
+#
+# v4.1-rc1-410-geeb1bd5c40ed
+#
+# Adds a struct net parameter to sock_create_kern
+#
+ifneq (,$(shell grep 'sock_create_kern.*struct net' include/linux/net.h))
+ccflags-y += -DKC_SOCK_CREATE_KERN_NET=1
+endif
+
+#
+# v3.18-rc6-1619-gc0371da6047a
+#
+# iov_iter is now part of struct msghdr
+#
+ifneq (,$(shell grep 'struct iov_iter.*msg_iter' include/linux/socket.h))
+ccflags-y += -DKC_MSGHDR_STRUCT_IOV_ITER=1
+endif
+
+#
+# v4.17-rc6-7-g95582b008388
+#
+# Kernel has current_time(inode) to uniformly retreive timespec in the right unit
+#
+ifneq (,$(shell grep 'extern struct timespec64 current_time' include/linux/fs.h))
+ccflags-y += -DKC_CURRENT_TIME_INODE=1
+endif
+
+#
+# v4.9-12228-g530e9b76ae8f
+#
+# register_cpu_notifier and family were all removed and to be
+# replaced with cpuhp_* API calls.
+#
+ifneq (,$(shell grep 'define register_hotcpu_notifier' include/linux/cpu.h))
+ccflags-y += -DKC_CPU_NOTIFIER
+endif
+
+#
+# v3.14-rc8-130-gccad2365668f
+#
+# generic_file_buffered_write is removed, backport it
+#
+ifneq (,$(shell grep 'extern ssize_t generic_file_buffered_write' include/linux/fs.h))
+ccflags-y += -DKC_GENERIC_FILE_BUFFERED_WRITE=1
+endif
+
+#
+# v5.7-438-g8151b4c8bee4
+#
+# struct address_space_operations switches away from .readpages to .readahead
+#
+# RHEL has backported this feature all the way to RHEL8, as part of RHEL_KABI,
+# which means we need to detect this very precisely
+#
+ifneq (,$(shell grep 'readahead.*struct readahead_control' include/linux/fs.h))
+ccflags-y += -DKC_FILE_AOPS_READAHEAD
+endif
+
+#
+# v4.0-rc7-1743-g8436318205b9
+#
+# .aio_read and .aio_write no longer exist. All reads and writes now use the
+# .read_iter and .write_iter methods, or must implement .read and .write (which
+# we don't).
+#
+ifneq (,$(shell grep 'ssize_t.*aio_read' include/linux/fs.h))
+ccflags-y += -DKC_LINUX_HAVE_FOP_AIO_READ=1
+endif
+
+#
+# rhel7 has a custom inode_operations_wrapper struct that is discarded
+# entirely in favor of upstream structure since rhel8.
+#
+ifneq (,$(shell grep 'void.*follow_link.*struct dentry' include/linux/fs.h))
+ccflags-y += -DKC_LINUX_HAVE_RHEL_IOPS_WRAPPER=1
+endif
+
+ifneq (,$(shell grep 'size_t.*ki_left;' include/linux/aio.h))
+ccflags-y += -DKC_LINUX_AIO_KI_LEFT=1
+endif
+
+#
+# v4.4-rc4-4-g98e9cb5711c6
+#
+# Introduces a new xattr_handler .name member that can be used to match the
+# entire field, instead of just a prefix. For these kernels, we must use
+# the new .name field instead.
+ifneq (,$(shell grep 'static inline const char .xattr_prefix' include/linux/xattr.h))
+ccflags-y += -DKC_XATTR_HANDLER_NAME=1
+endif

kmod/src/acl.c

@@ -69,12 +69,14 @@ struct posix_acl *scoutfs_get_acl_locked(struct inode *inode, int type, struct s
 	char *name;
 	int ret;
 
+#ifndef KC___POSIX_ACL_CREATE
 	if (!IS_POSIXACL(inode))
 		return NULL;
 
 	acl = get_cached_acl(inode, type);
 	if (acl != ACL_NOT_CACHED)
 		return acl;
+#endif
 
 	ret = acl_xattr_name_len(type, &name, NULL);
 	if (ret < 0)
@@ -96,9 +98,11 @@ struct posix_acl *scoutfs_get_acl_locked(struct inode *inode, int type, struct s
 		acl = ERR_PTR(ret);
 	}
 
+#ifndef KC___POSIX_ACL_CREATE
 	/* can set null negative cache */
 	if (!IS_ERR(acl))
 		set_cached_acl(inode, type, acl);
+#endif
 
 	kfree(value);
 
@@ -112,8 +116,10 @@ struct posix_acl *scoutfs_get_acl(struct inode *inode, int type)
 	struct posix_acl *acl;
 	int ret;
 
+#ifndef KC___POSIX_ACL_CREATE
 	if (!IS_POSIXACL(inode))
 		return NULL;
+#endif
 
 	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_READ, 0, inode, &lock);
 	if (ret < 0) {
@@ -183,13 +189,15 @@ int scoutfs_set_acl_locked(struct inode *inode, struct posix_acl *acl, int type,
 		if (!value) {
 			/* can be setting an acl that only affects mode, didn't need xattr */
 			inode_inc_iversion(inode);
-			inode->i_ctime = CURRENT_TIME;
+			inode->i_ctime = current_time(inode);
 		}
 	}
 
 out:
+#ifndef KC___POSIX_ACL_CREATE
 	if (!ret)
 		set_cached_acl(inode, type, acl);
+#endif
 
 	kfree(value);
 
@@ -218,10 +226,17 @@ int scoutfs_set_acl(struct inode *inode, struct posix_acl *acl, int type)
 	scoutfs_unlock(sb, lock, SCOUTFS_LOCK_WRITE);
 	return ret;
 }
-
+#ifdef KC_XATTR_STRUCT_XATTR_HANDLER
+int scoutfs_acl_get_xattr(const struct xattr_handler *handler, struct dentry *dentry,
+			  struct inode *inode, const char *name, void *value,
+			  size_t size)
+{
+	int type = handler->flags;
+#else
 int scoutfs_acl_get_xattr(struct dentry *dentry, const char *name, void *value, size_t size,
 			  int type)
 {
+#endif
 	struct posix_acl *acl;
 	int ret = 0;
 
@@ -240,9 +255,17 @@ int scoutfs_acl_get_xattr(struct dentry *dentry, const char *name, void *value,
 	return ret;
 }
 
+#ifdef KC_XATTR_STRUCT_XATTR_HANDLER
+int scoutfs_acl_set_xattr(const struct xattr_handler *handler, struct dentry *dentry,
+			  struct inode *inode, const char *name, const void *value,
+			  size_t size, int flags)
+{
+	int type = handler->flags;
+#else
 int scoutfs_acl_set_xattr(struct dentry *dentry, const char *name, const void *value, size_t size,
 			  int flags, int type)
 {
+#endif
 	struct posix_acl *acl = NULL;
 	int ret;
 
@@ -301,7 +324,7 @@ int scoutfs_init_acl_locked(struct inode *inode, struct inode *dir,
 			if (ret)
 				goto out;
 		}
-		ret = posix_acl_create(&acl, GFP_NOFS, &inode->i_mode);
+		ret = __posix_acl_create(&acl, GFP_NOFS, &inode->i_mode);
 		if (ret < 0)
 			return ret;
 		if (ret > 0)
@@ -345,7 +368,7 @@ int scoutfs_acl_chmod_locked(struct inode *inode, struct iattr *attr,
 	if (IS_ERR_OR_NULL(acl))
 		return PTR_ERR(acl);
 
-	ret = posix_acl_chmod(&acl, GFP_KERNEL, attr->ia_mode);
+	ret = __posix_acl_chmod(&acl, GFP_KERNEL, attr->ia_mode);
 	if (ret)
 		return ret;
 

kmod/src/acl.h

@@ -6,10 +6,19 @@ struct posix_acl *scoutfs_get_acl_locked(struct inode *inode, int type, struct s
 int scoutfs_set_acl(struct inode *inode, struct posix_acl *acl, int type);
 int scoutfs_set_acl_locked(struct inode *inode, struct posix_acl *acl, int type,
 			   struct scoutfs_lock *lock, struct list_head *ind_locks);
+#ifdef KC_XATTR_STRUCT_XATTR_HANDLER
+int scoutfs_acl_get_xattr(const struct xattr_handler *, struct dentry *dentry,
+			  struct inode *inode, const char *name, void *value,
+			  size_t size);
+int scoutfs_acl_set_xattr(const struct xattr_handler *, struct dentry *dentry,
+			  struct inode *inode, const char *name, const void *value,
+			  size_t size, int flags);
+#else
 int scoutfs_acl_get_xattr(struct dentry *dentry, const char *name, void *value, size_t size,
 			  int type);
 int scoutfs_acl_set_xattr(struct dentry *dentry, const char *name, const void *value, size_t size,
 			  int flags, int type);
+#endif
 int scoutfs_acl_chmod_locked(struct inode *inode, struct iattr *attr,
 			     struct scoutfs_lock *lock, struct list_head *ind_locks);
 int scoutfs_init_acl_locked(struct inode *inode, struct inode *dir,

kmod/src/block.c

@@ -21,6 +21,7 @@
 #include <linux/blkdev.h>
 #include <linux/rhashtable.h>
 #include <linux/random.h>
+#include <linux/sched/mm.h>
 
 #include "format.h"
 #include "super.h"
@@ -30,6 +31,7 @@
 #include "scoutfs_trace.h"
 #include "alloc.h"
 #include "triggers.h"
+#include "util.h"
 
 /*
  * The scoutfs block cache manages metadata blocks that can be larger
@@ -57,7 +59,7 @@ struct block_info {
 	atomic64_t access_counter;
 	struct rhashtable ht;
 	wait_queue_head_t waitq;
-	struct shrinker shrinker;
+	KC_DEFINE_SHRINKER(shrinker);
 	struct work_struct free_work;
 	struct llist_head free_llist;
 };
@@ -128,7 +130,7 @@ static __le32 block_calc_crc(struct scoutfs_block_header *hdr, u32 size)
 static struct block_private *block_alloc(struct super_block *sb, u64 blkno)
 {
 	struct block_private *bp;
-	unsigned int noio_flags;
+	unsigned int nofs_flags;
 
 	/*
 	 * If we had multiple blocks per page we'd need to be a little
@@ -156,9 +158,9 @@ static struct block_private *block_alloc(struct super_block *sb, u64 blkno)
 		 * spurious reclaim-on dependencies and warnings.
 		 */
 		lockdep_off();
-		noio_flags = memalloc_noio_save();
+		nofs_flags = memalloc_nofs_save();
 		bp->virt = __vmalloc(SCOUTFS_BLOCK_LG_SIZE, GFP_NOFS | __GFP_HIGHMEM, PAGE_KERNEL);
-		memalloc_noio_restore(noio_flags);
+		memalloc_nofs_restore(nofs_flags);
 		lockdep_on();
 
 		if (!bp->virt) {
@@ -436,11 +438,10 @@ static void block_remove_all(struct super_block *sb)
  * possible.  Final freeing, verifying checksums, and unlinking errored
  * blocks are all done by future users of the blocks.
  */
-static void block_end_io(struct super_block *sb, int rw,
+static void block_end_io(struct super_block *sb, unsigned int opf,
 			 struct block_private *bp, int err)
 {
 	DECLARE_BLOCK_INFO(sb, binf);
-	bool is_read = !(rw & WRITE);
 
 	if (err) {
 		scoutfs_inc_counter(sb, block_cache_end_io_error);
@@ -450,7 +451,7 @@ static void block_end_io(struct super_block *sb, int rw,
 	if (!atomic_dec_and_test(&bp->io_count))
 		return;
 
-	if (is_read && !test_bit(BLOCK_BIT_ERROR, &bp->bits))
+	if (!op_is_write(opf) && !test_bit(BLOCK_BIT_ERROR, &bp->bits))
 		set_bit(BLOCK_BIT_UPTODATE, &bp->bits);
 
 	clear_bit(BLOCK_BIT_IO_BUSY, &bp->bits);
@@ -463,13 +464,13 @@ static void block_end_io(struct super_block *sb, int rw,
 		wake_up(&binf->waitq);
 }
 
-static void block_bio_end_io(struct bio *bio, int err)
+static void KC_DECLARE_BIO_END_IO(block_bio_end_io, struct bio *bio)
 {
 	struct block_private *bp = bio->bi_private;
 	struct super_block *sb = bp->sb;
 
 	TRACE_BLOCK(end_io, bp);
-	block_end_io(sb, bio->bi_rw, bp, err);
+	block_end_io(sb, kc_bio_get_opf(bio), bp, kc_bio_get_errno(bio));
 	bio_put(bio);
 }
 
@@ -477,7 +478,7 @@ static void block_bio_end_io(struct bio *bio, int err)
  * Kick off IO for a single block.
  */
 static int block_submit_bio(struct super_block *sb, struct block_private *bp,
-			    int rw)
+			    unsigned int opf)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct bio *bio = NULL;
@@ -510,8 +511,9 @@ static int block_submit_bio(struct super_block *sb, struct block_private *bp,
 				break;
 			}
 
-			bio->bi_sector = sector + (off >> 9);
-			bio->bi_bdev = sbi->meta_bdev;
+			kc_bio_set_opf(bio, opf);
+			kc_bio_set_sector(bio, sector + (off >> 9));
+			bio_set_dev(bio, sbi->meta_bdev);
 			bio->bi_end_io = block_bio_end_io;
 			bio->bi_private = bp;
 
@@ -528,18 +530,18 @@ static int block_submit_bio(struct super_block *sb, struct block_private *bp,
 			BUG();
 
 		if (!bio_add_page(bio, page, PAGE_SIZE, 0)) {
-			submit_bio(rw, bio);
+			kc_submit_bio(bio);
 			bio = NULL;
 		}
 	}
 
 	if (bio)
-		submit_bio(rw, bio);
+		kc_submit_bio(bio);
 
 	blk_finish_plug(&plug);
 
 	/* let racing end_io know we're done */
-	block_end_io(sb, rw, bp, ret);
+	block_end_io(sb, opf, bp, ret);
 
 	return ret;
 }
@@ -640,7 +642,7 @@ static struct block_private *block_read(struct super_block *sb, u64 blkno)
 
 	if (!test_bit(BLOCK_BIT_UPTODATE, &bp->bits) &&
 	     test_and_clear_bit(BLOCK_BIT_NEW, &bp->bits)) {
-		ret = block_submit_bio(sb, bp, READ);
+		ret = block_submit_bio(sb, bp, REQ_OP_READ);
 		if (ret < 0)
 			goto out;
 	}
@@ -969,7 +971,7 @@ int scoutfs_block_writer_write(struct super_block *sb,
 		/* retry previous write errors */
 		clear_bit(BLOCK_BIT_ERROR, &bp->bits);
 
-		ret = block_submit_bio(sb, bp, WRITE);
+		ret = block_submit_bio(sb, bp, REQ_OP_WRITE);
 		if (ret < 0)
 			break;
 	}
@@ -1069,6 +1071,16 @@ u64 scoutfs_block_writer_dirty_bytes(struct super_block *sb,
 	return wri->nr_dirty_blocks * SCOUTFS_BLOCK_LG_SIZE;
 }
 
+static unsigned long block_count_objects(struct shrinker *shrink, struct shrink_control *sc)
+{
+	struct block_info *binf = KC_SHRINKER_CONTAINER_OF(shrink, struct block_info);
+	struct super_block *sb = binf->sb;
+
+	scoutfs_inc_counter(sb, block_cache_count_objects);
+
+	return shrinker_min_t_long((u64)atomic_read(&binf->total_inserted));
+}
+
 /*
  * Remove a number of cached blocks that haven't been used recently.
  *
@@ -1089,25 +1101,19 @@ u64 scoutfs_block_writer_dirty_bytes(struct super_block *sb,
  * atomically remove blocks when the only references are ours and the
  * hash table.
  */
-static int block_shrink(struct shrinker *shrink, struct shrink_control *sc)
+static unsigned long block_scan_objects(struct shrinker *shrink, struct shrink_control *sc)
 {
-	struct block_info *binf = container_of(shrink, struct block_info,
-					       shrinker);
+	struct block_info *binf = KC_SHRINKER_CONTAINER_OF(shrink, struct block_info);
 	struct super_block *sb = binf->sb;
 	struct rhashtable_iter iter;
 	struct block_private *bp;
-	unsigned long nr;
+	bool stop = false;
+	unsigned long freed = 0;
+	unsigned long nr = sc->nr_to_scan;
 	u64 recently;
 
-	nr = sc->nr_to_scan;
-	if (nr == 0)
-		goto out;
+	scoutfs_inc_counter(sb, block_cache_scan_objects);
 
-	scoutfs_inc_counter(sb, block_cache_shrink);
-
-	nr = DIV_ROUND_UP(nr, SCOUTFS_BLOCK_LG_PAGES_PER);
-
-restart:
 	recently = accessed_recently(binf);
 	rhashtable_walk_enter(&binf->ht, &iter);
 	rhashtable_walk_start(&iter);
@@ -1129,12 +1135,15 @@ restart:
 		if (bp == NULL)
 			break;
 		if (bp == ERR_PTR(-EAGAIN)) {
-			/* hard exit to wait for rcu rebalance to finish */
-			rhashtable_walk_stop(&iter);
-			rhashtable_walk_exit(&iter);
-			scoutfs_inc_counter(sb, block_cache_shrink_restart);
-			synchronize_rcu();
-			goto restart;
+			/*
+			 * We can be called from reclaim in the allocation
+			 * to resize the hash table itself.  We have to
+			 * return so that the caller can proceed and
+			 * enable hash table iteration again.
+			 */
+			scoutfs_inc_counter(sb, block_cache_shrink_stop);
+			stop = true;
+			break;
 		}
 
 		scoutfs_inc_counter(sb, block_cache_shrink_next);
@@ -1148,6 +1157,7 @@ restart:
 			if (block_remove_solo(sb, bp)) {
 				scoutfs_inc_counter(sb, block_cache_shrink_remove);
 				TRACE_BLOCK(shrink, bp);
+				freed++;
 				nr--;
 			}
 			block_put(sb, bp);
@@ -1156,9 +1166,11 @@ restart:
 
 	rhashtable_walk_stop(&iter);
 	rhashtable_walk_exit(&iter);
-out:
-	return min_t(u64, (u64)atomic_read(&binf->total_inserted) * SCOUTFS_BLOCK_LG_PAGES_PER,
-		     INT_MAX);
+
+	if (stop)
+		return SHRINK_STOP;
+	else
+		return freed;
 }
 
 struct sm_block_completion {
@@ -1166,11 +1178,11 @@ struct sm_block_completion {
 	int err;
 };
 
-static void sm_block_bio_end_io(struct bio *bio, int err)
+static void KC_DECLARE_BIO_END_IO(sm_block_bio_end_io, struct bio *bio)
 {
 	struct sm_block_completion *sbc = bio->bi_private;
 
-	sbc->err = err;
+	sbc->err = kc_bio_get_errno(bio);
 	complete(&sbc->comp);
 	bio_put(bio);
 }
@@ -1185,9 +1197,8 @@ static void sm_block_bio_end_io(struct bio *bio, int err)
  * only layer that sees the full block buffer so we pass the calculated
  * crc to the caller for them to check in their context.
  */
-static int sm_block_io(struct super_block *sb, struct block_device *bdev, int rw, u64 blkno,
-		       struct scoutfs_block_header *hdr, size_t len,
-		       __le32 *blk_crc)
+static int sm_block_io(struct super_block *sb, struct block_device *bdev, unsigned int opf,
+		       u64 blkno, struct scoutfs_block_header *hdr, size_t len, __le32 *blk_crc)
 {
 	struct scoutfs_block_header *pg_hdr;
 	struct sm_block_completion sbc;
@@ -1201,7 +1212,7 @@ static int sm_block_io(struct super_block *sb, struct block_device *bdev, int rw
 		return -EIO;
 
 	if (WARN_ON_ONCE(len > SCOUTFS_BLOCK_SM_SIZE) ||
-	    WARN_ON_ONCE(!(rw & WRITE) && !blk_crc))
+	    WARN_ON_ONCE(!op_is_write(opf) && !blk_crc))
 		return -EINVAL;
 
 	page = alloc_page(GFP_NOFS);
@@ -1210,7 +1221,7 @@ static int sm_block_io(struct super_block *sb, struct block_device *bdev, int rw
 
 	pg_hdr = page_address(page);
 
-	if (rw & WRITE) {
+	if (op_is_write(opf)) {
 		memcpy(pg_hdr, hdr, len);
 		if (len < SCOUTFS_BLOCK_SM_SIZE)
 			memset((char *)pg_hdr + len, 0,
@@ -1224,8 +1235,9 @@ static int sm_block_io(struct super_block *sb, struct block_device *bdev, int rw
 		goto out;
 	}
 
-	bio->bi_sector = blkno << (SCOUTFS_BLOCK_SM_SHIFT - 9);
-	bio->bi_bdev = bdev;
+	kc_bio_set_opf(bio, opf | REQ_SYNC);
+	kc_bio_set_sector(bio, blkno << (SCOUTFS_BLOCK_SM_SHIFT - 9));
+	bio_set_dev(bio, bdev);
 	bio->bi_end_io = sm_block_bio_end_io;
 	bio->bi_private = &sbc;
 	bio_add_page(bio, page, SCOUTFS_BLOCK_SM_SIZE, 0);
@@ -1233,12 +1245,12 @@ static int sm_block_io(struct super_block *sb, struct block_device *bdev, int rw
 	init_completion(&sbc.comp);
 	sbc.err = 0;
 
-	submit_bio((rw & WRITE) ? WRITE_SYNC : READ_SYNC, bio);
+	kc_submit_bio(bio);
 
 	wait_for_completion(&sbc.comp);
 	ret = sbc.err;
 
-	if (ret == 0 && !(rw & WRITE)) {
+	if (ret == 0 && !op_is_write(opf)) {
 		memcpy(hdr, pg_hdr, len);
 		*blk_crc = block_calc_crc(pg_hdr, SCOUTFS_BLOCK_SM_SIZE);
 	}
@@ -1252,14 +1264,14 @@ int scoutfs_block_read_sm(struct super_block *sb,
 			  struct scoutfs_block_header *hdr, size_t len,
 			  __le32 *blk_crc)
 {
-	return sm_block_io(sb, bdev, READ, blkno, hdr, len, blk_crc);
+	return sm_block_io(sb, bdev, REQ_OP_READ, blkno, hdr, len, blk_crc);
 }
 
 int scoutfs_block_write_sm(struct super_block *sb,
 			   struct block_device *bdev, u64 blkno,
 			   struct scoutfs_block_header *hdr, size_t len)
 {
-	return sm_block_io(sb, bdev, WRITE, blkno, hdr, len, NULL);
+	return sm_block_io(sb, bdev, REQ_OP_WRITE, blkno, hdr, len, NULL);
 }
 
 int scoutfs_block_setup(struct super_block *sb)
@@ -1284,9 +1296,9 @@ int scoutfs_block_setup(struct super_block *sb)
 	atomic_set(&binf->total_inserted, 0);
 	atomic64_set(&binf->access_counter, 0);
 	init_waitqueue_head(&binf->waitq);
-	binf->shrinker.shrink = block_shrink;
-	binf->shrinker.seeks = DEFAULT_SEEKS;
-	register_shrinker(&binf->shrinker);
+	KC_INIT_SHRINKER_FUNCS(&binf->shrinker, block_count_objects,
+			       block_scan_objects);
+	KC_REGISTER_SHRINKER(&binf->shrinker);
 	INIT_WORK(&binf->free_work, block_free_work);
 	init_llist_head(&binf->free_llist);
 
@@ -1306,7 +1318,7 @@ void scoutfs_block_destroy(struct super_block *sb)
 	struct block_info *binf = SCOUTFS_SB(sb)->block_info;
 
 	if (binf) {
-		unregister_shrinker(&binf->shrinker);
+		KC_UNREGISTER_SHRINKER(&binf->shrinker);
 		block_remove_all(sb);
 		flush_work(&binf->free_work);
 		rhashtable_destroy(&binf->ht);

kmod/src/counters.h

@@ -30,11 +30,13 @@
 	EXPAND_COUNTER(block_cache_free)			\
 	EXPAND_COUNTER(block_cache_free_work)			\
 	EXPAND_COUNTER(block_cache_remove_stale)		\
+	EXPAND_COUNTER(block_cache_count_objects)		\
+	EXPAND_COUNTER(block_cache_scan_objects)		\
 	EXPAND_COUNTER(block_cache_shrink)			\
 	EXPAND_COUNTER(block_cache_shrink_next)			\
 	EXPAND_COUNTER(block_cache_shrink_recent)		\
 	EXPAND_COUNTER(block_cache_shrink_remove)		\
-	EXPAND_COUNTER(block_cache_shrink_restart)		\
+	EXPAND_COUNTER(block_cache_shrink_stop)			\
 	EXPAND_COUNTER(btree_compact_values)			\
 	EXPAND_COUNTER(btree_compact_values_enomem)		\
 	EXPAND_COUNTER(btree_delete)				\
@@ -88,6 +90,8 @@
 	EXPAND_COUNTER(forest_read_items)			\
 	EXPAND_COUNTER(forest_roots_next_hint)			\
 	EXPAND_COUNTER(forest_set_bloom_bits)			\
+	EXPAND_COUNTER(item_cache_count_objects)		\
+	EXPAND_COUNTER(item_cache_scan_objects)			\
 	EXPAND_COUNTER(item_clear_dirty)			\
 	EXPAND_COUNTER(item_create)				\
 	EXPAND_COUNTER(item_delete)				\
@@ -121,6 +125,7 @@
 	EXPAND_COUNTER(item_update)				\
 	EXPAND_COUNTER(item_write_dirty)			\
 	EXPAND_COUNTER(lock_alloc)				\
+	EXPAND_COUNTER(lock_count_objects)			\
 	EXPAND_COUNTER(lock_free)				\
 	EXPAND_COUNTER(lock_grant_request)			\
 	EXPAND_COUNTER(lock_grant_response)			\
@@ -134,6 +139,7 @@
 	EXPAND_COUNTER(lock_lock_error)				\
 	EXPAND_COUNTER(lock_nonblock_eagain)			\
 	EXPAND_COUNTER(lock_recover_request)			\
+	EXPAND_COUNTER(lock_scan_objects)			\
 	EXPAND_COUNTER(lock_shrink_attempted)			\
 	EXPAND_COUNTER(lock_shrink_aborted)			\
 	EXPAND_COUNTER(lock_shrink_work)			\
@@ -166,6 +172,7 @@
 	EXPAND_COUNTER(quorum_recv_resignation)			\
 	EXPAND_COUNTER(quorum_recv_vote)			\
 	EXPAND_COUNTER(quorum_send_heartbeat)			\
+	EXPAND_COUNTER(quorum_send_heartbeat_dropped)		\
 	EXPAND_COUNTER(quorum_send_resignation)			\
 	EXPAND_COUNTER(quorum_send_request)			\
 	EXPAND_COUNTER(quorum_send_vote)			\
@@ -231,12 +238,12 @@ struct scoutfs_counters {
 #define SCOUTFS_PCPU_COUNTER_BATCH (1 << 30)
 
 #define scoutfs_inc_counter(sb, which)					\
-	__percpu_counter_add(&SCOUTFS_SB(sb)->counters->which, 1,	\
-			     SCOUTFS_PCPU_COUNTER_BATCH)
+	percpu_counter_add_batch(&SCOUTFS_SB(sb)->counters->which, 1,	\
+				 SCOUTFS_PCPU_COUNTER_BATCH)
 
 #define scoutfs_add_counter(sb, which, cnt)				\
-	__percpu_counter_add(&SCOUTFS_SB(sb)->counters->which, cnt,	\
-			     SCOUTFS_PCPU_COUNTER_BATCH)
+	percpu_counter_add_batch(&SCOUTFS_SB(sb)->counters->which, cnt,	\
+				 SCOUTFS_PCPU_COUNTER_BATCH)
 
 void __init scoutfs_init_counters(void);
 int scoutfs_setup_counters(struct super_block *sb);

kmod/src/data.c

@@ -307,7 +307,7 @@ int scoutfs_data_truncate_items(struct super_block *sb, struct inode *inode,
 	LIST_HEAD(ind_locks);
 	s64 ret = 0;
 
-	WARN_ON_ONCE(inode && !mutex_is_locked(&inode->i_mutex));
+	WARN_ON_ONCE(inode && !inode_is_locked(inode));
 
 	/* clamp last to the last possible block? */
 	if (last > SCOUTFS_BLOCK_SM_MAX)
@@ -456,11 +456,11 @@ static int alloc_block(struct super_block *sb, struct inode *inode,
 
 	} else {
 		/*
-		 * Preallocation of aligned regions only preallocates if
-		 * the aligned region contains no extents at all.  This
-		 * could be fooled by offline sparse extents but we
-		 * don't want to iterate over all offline extents in the
-		 * aligned region.
+		 * Preallocation within aligned regions tries to
+		 * allocate an extent to fill the hole in the region
+		 * that contains iblock.  We'd have to add a bit of plumbing
+		 * to find previous extents so we only search for a next
+		 * extent from the front of the region and from iblock.
 		 */
 		div64_u64_rem(iblock, opts.data_prealloc_blocks, &rem);
 		start = iblock - rem;
@@ -468,8 +468,20 @@ static int alloc_block(struct super_block *sb, struct inode *inode,
 		ret = scoutfs_ext_next(sb, &data_ext_ops, &args, start, 1, &found);
 		if (ret < 0 && ret != -ENOENT)
 			goto out;
-		if (found.len && found.start < start + count)
-			count = 1;
+
+		/* trim count if there's an extent in the region before iblock */
+		if (found.len && found.start < iblock) {
+			count -= iblock - start;
+			start = iblock;
+			/* see if there's also an extent after iblock */
+			ret = scoutfs_ext_next(sb, &data_ext_ops, &args, iblock, 1, &found);
+			if (ret < 0 && ret != -ENOENT)
+				goto out;
+		}
+
+		/* trim count by next extent after iblock */
+		if (found.len && found.start > start && found.start < start + count)
+			count = (found.start - start);
 	}
 
 	/* overall prealloc limit */
@@ -546,7 +558,7 @@ static int scoutfs_get_block(struct inode *inode, sector_t iblock,
 	u64 offset;
 	int ret;
 
-	WARN_ON_ONCE(create && !mutex_is_locked(&inode->i_mutex));
+	WARN_ON_ONCE(create && !inode_is_locked(inode));
 
 	/* make sure caller holds a cluster lock */
 	lock = scoutfs_per_task_get(&si->pt_data_lock);
@@ -692,7 +704,7 @@ static int scoutfs_readpage(struct file *file, struct page *page)
 
 	if (scoutfs_per_task_add_excl(&si->pt_data_lock, &pt_ent, inode_lock)) {
 		ret = scoutfs_data_wait_check(inode, page_offset(page),
-					      PAGE_CACHE_SIZE, SEF_OFFLINE,
+					      PAGE_SIZE, SEF_OFFLINE,
 					      SCOUTFS_IOC_DWO_READ, &dw,
 					      inode_lock);
 		if (ret != 0) {
@@ -717,6 +729,7 @@ static int scoutfs_readpage(struct file *file, struct page *page)
 	return ret;
 }
 
+#ifndef KC_FILE_AOPS_READAHEAD
 /*
  * This is used for opportunistic read-ahead which can throw the pages
  * away if it needs to.  If the caller didn't deal with offline extents
@@ -742,14 +755,14 @@ static int scoutfs_readpages(struct file *file, struct address_space *mapping,
 
 	list_for_each_entry_safe(page, tmp, pages, lru) {
 		ret = scoutfs_data_wait_check(inode, page_offset(page),
-					      PAGE_CACHE_SIZE, SEF_OFFLINE,
+					      PAGE_SIZE, SEF_OFFLINE,
 					      SCOUTFS_IOC_DWO_READ, NULL,
 					      inode_lock);
 		if (ret < 0)
 			goto out;
 		if (ret > 0) {
 			list_del(&page->lru);
-			page_cache_release(page);
+			put_page(page);
 			if (--nr_pages == 0) {
 				ret = 0;
 				goto out;
@@ -763,6 +776,29 @@ out:
 	BUG_ON(!list_empty(pages));
 	return ret;
 }
+#else
+static void scoutfs_readahead(struct readahead_control *rac)
+{
+	struct inode *inode = rac->file->f_inode;
+	struct super_block *sb = inode->i_sb;
+	struct scoutfs_lock *inode_lock = NULL;
+	int ret;
+
+	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_READ,
+				 SCOUTFS_LKF_REFRESH_INODE, inode, &inode_lock);
+	if (ret)
+		return;
+
+	ret = scoutfs_data_wait_check(inode, readahead_pos(rac),
+				      readahead_length(rac), SEF_OFFLINE,
+				      SCOUTFS_IOC_DWO_READ, NULL,
+				      inode_lock);
+	if (ret == 0)
+		mpage_readahead(rac, scoutfs_get_block_read);
+
+	scoutfs_unlock(sb, inode_lock, SCOUTFS_LOCK_READ);
+}
+#endif
 
 static int scoutfs_writepage(struct page *page, struct writeback_control *wbc)
 {
@@ -1045,7 +1081,7 @@ long scoutfs_fallocate(struct file *file, int mode, loff_t offset, loff_t len)
 		goto out;
 	}
 
-	mutex_lock(&inode->i_mutex);
+	inode_lock(inode);
 
 	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_WRITE,
 				 SCOUTFS_LKF_REFRESH_INODE, inode, &lock);
@@ -1106,7 +1142,7 @@ out_extent:
 	up_write(&si->extent_sem);
 out_mutex:
 	scoutfs_unlock(sb, lock, SCOUTFS_LOCK_WRITE);
-	mutex_unlock(&inode->i_mutex);
+	inode_unlock(inode);
 
 out:
 	trace_scoutfs_data_fallocate(sb, ino, mode, offset, len, ret);
@@ -1209,7 +1245,7 @@ int scoutfs_data_move_blocks(struct inode *from, u64 from_off,
 	struct data_ext_args from_args;
 	struct data_ext_args to_args;
 	struct scoutfs_extent ext;
-	struct timespec cur_time;
+	struct kc_timespec cur_time;
 	LIST_HEAD(locks);
 	bool done = false;
 	loff_t from_size;
@@ -1253,6 +1289,7 @@ int scoutfs_data_move_blocks(struct inode *from, u64 from_off,
 	from_iblock = from_off >> SCOUTFS_BLOCK_SM_SHIFT;
 	count = (byte_len + SCOUTFS_BLOCK_SM_MASK) >> SCOUTFS_BLOCK_SM_SHIFT;
 	to_iblock = to_off >> SCOUTFS_BLOCK_SM_SHIFT;
+	from_start = from_iblock;
 
 	/* only move extent blocks inside i_size, careful not to wrap */
 	from_size = i_size_read(from);
@@ -1329,7 +1366,7 @@ int scoutfs_data_move_blocks(struct inode *from, u64 from_off,
 
 			/* find the next extent to move */
 			ret = scoutfs_ext_next(sb, &data_ext_ops, &from_args,
-					       from_iblock, 1, &ext);
+					       from_start, 1, &ext);
 			if (ret < 0) {
 				if (ret == -ENOENT) {
 					done = true;
@@ -1417,13 +1454,19 @@ int scoutfs_data_move_blocks(struct inode *from, u64 from_off,
 							i_size_read(from);
 				i_size_write(to, to_size);
 			}
+
+			/* find next after moved extent, avoiding wrapping */
+			if (from_start + len < from_start)
+				from_start = from_iblock + count + 1;
+			else
+				from_start += len;
 		}
 
 
 		up_write(&from_si->extent_sem);
 		up_write(&to_si->extent_sem);
 
-		cur_time = CURRENT_TIME;
+		cur_time = current_time(from);
 		if (!is_stage) {
 			to->i_ctime = to->i_mtime = cur_time;
 			inode_inc_iversion(to);
@@ -1510,7 +1553,7 @@ int scoutfs_data_fiemap(struct inode *inode, struct fiemap_extent_info *fieinfo,
 	if (ret)
 		goto out;
 
-	mutex_lock(&inode->i_mutex);
+	inode_lock(inode);
 	down_read(&si->extent_sem);
 
 	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_READ, 0, inode, &lock);
@@ -1564,7 +1607,7 @@ int scoutfs_data_fiemap(struct inode *inode, struct fiemap_extent_info *fieinfo,
 unlock:
 	scoutfs_unlock(sb, lock, SCOUTFS_LOCK_READ);
 	up_read(&si->extent_sem);
-	mutex_unlock(&inode->i_mutex);
+	inode_unlock(inode);
 
 out:
 	if (ret == 1)
@@ -1764,6 +1807,37 @@ int scoutfs_data_wait_check_iov(struct inode *inode, const struct iovec *iov,
 	return ret;
 }
 
+int scoutfs_data_wait_check_iter(struct inode *inode, loff_t pos, struct iov_iter *iter,
+				 u8 sef, u8 op, struct scoutfs_data_wait *dw,
+				 struct scoutfs_lock *lock)
+{
+	size_t count = iov_iter_count(iter);
+	size_t off = iter->iov_offset;
+	const struct iovec *iov;
+	size_t len;
+	int ret = 0;
+
+	for (iov = iter->iov; count > 0; iov++) {
+		len = iov->iov_len - off;
+		if (len == 0)
+			continue;
+
+		/* aren't we waiting on too much data here ? */
+		ret = scoutfs_data_wait_check(inode, pos, len,
+					      sef, op, dw, lock);
+
+		if (ret != 0)
+			break;
+
+
+		pos += len;
+		count -= len;
+		off = 0;
+	}
+
+	return ret;
+}
+
 int scoutfs_data_wait(struct inode *inode, struct scoutfs_data_wait *dw)
 {
 	DECLARE_DATA_WAIT_ROOT(inode->i_sb, rt);
@@ -1854,7 +1928,11 @@ int scoutfs_data_waiting(struct super_block *sb, u64 ino, u64 iblock,
 
 const struct address_space_operations scoutfs_file_aops = {
 	.readpage		= scoutfs_readpage,
+#ifndef KC_FILE_AOPS_READAHEAD
 	.readpages		= scoutfs_readpages,
+#else
+	.readahead		= scoutfs_readahead,
+#endif
 	.writepage		= scoutfs_writepage,
 	.writepages		= scoutfs_writepages,
 	.write_begin		= scoutfs_write_begin,
@@ -1862,10 +1940,15 @@ const struct address_space_operations scoutfs_file_aops = {
 };
 
 const struct file_operations scoutfs_file_fops = {
+#ifdef KC_LINUX_HAVE_FOP_AIO_READ
 	.read		= do_sync_read,
 	.write		= do_sync_write,
 	.aio_read	= scoutfs_file_aio_read,
 	.aio_write	= scoutfs_file_aio_write,
+#else
+	.read_iter	= scoutfs_file_read_iter,
+	.write_iter	= scoutfs_file_write_iter,
+#endif
 	.unlocked_ioctl	= scoutfs_ioctl,
 	.fsync		= scoutfs_file_fsync,
 	.llseek		= scoutfs_file_llseek,

kmod/src/data.h

@@ -65,6 +65,9 @@ int scoutfs_data_wait_check_iov(struct inode *inode, const struct iovec *iov,
 				unsigned long nr_segs, loff_t pos, u8 sef,
 				u8 op, struct scoutfs_data_wait *ow,
 				struct scoutfs_lock *lock);
+int scoutfs_data_wait_check_iter(struct inode *inode, loff_t pos, struct iov_iter *iter,
+				 u8 sef, u8 op, struct scoutfs_data_wait *ow,
+				 struct scoutfs_lock *lock);
 bool scoutfs_data_wait_found(struct scoutfs_data_wait *ow);
 int scoutfs_data_wait(struct inode *inode,
 			      struct scoutfs_data_wait *ow);

kmod/src/dir.c

@@ -272,7 +272,7 @@ static void set_dentry_fsdata(struct dentry *dentry, struct scoutfs_lock *lock)
 
 static bool test_dentry_fsdata(struct dentry *dentry, u64 refresh)
 {
-	u64 fsd = (unsigned long)ACCESS_ONCE(dentry->d_fsdata);
+	u64 fsd = (unsigned long)READ_ONCE(dentry->d_fsdata);
 
 	return fsd == refresh;
 }
@@ -735,7 +735,7 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode,
 	set_dentry_fsdata(dentry, dir_lock);
 
 	i_size_write(dir, i_size_read(dir) + dentry->d_name.len);
-	dir->i_mtime = dir->i_ctime = CURRENT_TIME;
+	dir->i_mtime = dir->i_ctime = current_time(inode);
 	inode->i_mtime = inode->i_atime = inode->i_ctime = dir->i_mtime;
 	si->crtime = inode->i_mtime;
 	inode_inc_iversion(dir);
@@ -859,7 +859,7 @@ retry:
 	set_dentry_fsdata(dentry, dir_lock);
 
 	i_size_write(dir, dir_size);
-	dir->i_mtime = dir->i_ctime = CURRENT_TIME;
+	dir->i_mtime = dir->i_ctime = current_time(inode);
 	inode->i_ctime = dir->i_mtime;
 	inc_nlink(inode);
 	inode_inc_iversion(dir);
@@ -900,7 +900,7 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry)
 {
 	struct super_block *sb = dir->i_sb;
 	struct inode *inode = dentry->d_inode;
-	struct timespec ts = current_kernel_time();
+	struct kc_timespec ts = current_time(inode);
 	struct scoutfs_lock *inode_lock = NULL;
 	struct scoutfs_lock *orph_lock = NULL;
 	struct scoutfs_lock *dir_lock = NULL;
@@ -1058,6 +1058,7 @@ static int symlink_item_ops(struct super_block *sb, enum symlink_ops op, u64 ino
 	return ret;
 }
 
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 /*
  * Full a buffer with the null terminated symlink, point nd at it, and
  * return it so put_link can free it once the vfs is done.
@@ -1138,19 +1139,73 @@ static void scoutfs_put_link(struct dentry *dentry, struct nameidata *nd,
 	if (!IS_ERR_OR_NULL(cookie))
 		kfree(cookie);
 }
+#else
+static const char *scoutfs_get_link(struct dentry *dentry, struct inode *inode, struct delayed_call *done)
+{
+	struct super_block *sb = inode->i_sb;
+	struct scoutfs_lock *inode_lock = NULL;
+	char *path = NULL;
+	loff_t size;
+	int ret;
 
-const struct inode_operations scoutfs_symlink_iops = {
-	.readlink       = generic_readlink,
-	.follow_link    = scoutfs_follow_link,
-	.put_link       = scoutfs_put_link,
-	.getattr	= scoutfs_getattr,
-	.setattr	= scoutfs_setattr,
-	.setxattr	= generic_setxattr,
-	.getxattr	= generic_getxattr,
-	.listxattr	= scoutfs_listxattr,
-	.removexattr	= generic_removexattr,
-	.get_acl	= scoutfs_get_acl,
-};
+	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_READ,
+				 SCOUTFS_LKF_REFRESH_INODE, inode, &inode_lock);
+	if (ret)
+		return ERR_PTR(ret);
+
+	size = i_size_read(inode);
+
+	if (size == 0 || size > SCOUTFS_SYMLINK_MAX_SIZE) {
+		scoutfs_corruption(sb, SC_SYMLINK_INODE_SIZE,
+				   corrupt_symlink_inode_size,
+				   "ino %llu size %llu",
+				   scoutfs_ino(inode), (u64)size);
+		ret = -EIO;
+		goto out;
+	}
+
+	/* unlikely, but possible I suppose */
+	if (size > PATH_MAX) {
+		ret = -ENAMETOOLONG;
+		goto out;
+	}
+
+	path = kmalloc(size, GFP_NOFS);
+	if (!path) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
+	ret = symlink_item_ops(sb, SYM_LOOKUP, scoutfs_ino(inode), inode_lock,
+			       path, size);
+
+	if (ret == -ENOENT) {
+		scoutfs_corruption(sb, SC_SYMLINK_MISSING_ITEM,
+				   corrupt_symlink_missing_item,
+				   "ino %llu size %llu", scoutfs_ino(inode),
+				   size);
+		ret = -EIO;
+
+	} else if (ret == 0 && path[size - 1]) {
+		scoutfs_corruption(sb, SC_SYMLINK_NOT_NULL_TERM,
+				   corrupt_symlink_not_null_term,
+				   "ino %llu last %u",
+				   scoutfs_ino(inode), path[size - 1]);
+		ret = -EIO;
+	}
+
+	if (ret != -EIO)
+		set_delayed_call(done, kfree_link, path);
+
+out:
+	if (ret < 0) {
+		kfree(path);
+		path = ERR_PTR(ret);
+	}
+	scoutfs_unlock(sb, inode_lock, SCOUTFS_LOCK_READ);
+	return path;
+}
+#endif
 
 /*
  * Symlink target paths can be annoyingly large.  We store relatively
@@ -1204,7 +1259,7 @@ static int scoutfs_symlink(struct inode *dir, struct dentry *dentry,
 	set_dentry_fsdata(dentry, dir_lock);
 
 	i_size_write(dir, i_size_read(dir) + dentry->d_name.len);
-	dir->i_mtime = dir->i_ctime = CURRENT_TIME;
+	dir->i_mtime = dir->i_ctime = current_time(inode);
 	inode_inc_iversion(dir);
 
 	inode->i_ctime = dir->i_mtime;
@@ -1253,75 +1308,93 @@ int scoutfs_symlink_drop(struct super_block *sb, u64 ino,
 }
 
 /*
- * Find the next link backref key for the given ino starting from the
- * given dir inode and final entry position.  If we find a backref item
- * we add an allocated copy of it to the head of the caller's list.
+ * Find the next link backref items for the given ino starting from the
+ * given dir inode and final entry position.  For each backref item we
+ * add an allocated copy of it to the head of the caller's list.
  *
- * Returns 0 if we added an entry, -ENOENT if we didn't, and -errno for
- * search errors.
+ * Callers who are building a path can add one entry for each parent.
+ * They're left with a list of entries from the root down in list order.
+ *
+ * Callers who are gathering multiple entries for one inode get the
+ * entries in the opposite order that their items are found.
+ *
+ * Returns +ve for number of entries added, -ENOENT if no entries were
+ * found, or -errno on error.  It weirdly won't return 0, but early
+ * callers preferred -ENOENT so we use that for the case of no entries.
  *
  * Callers are comfortable with the race inherent to incrementally
- * building up a path with individual locked backref item lookups.
+ * gathering backrefs across multiple lock acquisitions.
  */
-int scoutfs_dir_add_next_linkref(struct super_block *sb, u64 ino,
-				 u64 dir_ino, u64 dir_pos,
-				 struct list_head *list)
+int scoutfs_dir_add_next_linkrefs(struct super_block *sb, u64 ino, u64 dir_ino, u64 dir_pos,
+				  int count, struct list_head *list)
 {
+	struct scoutfs_link_backref_entry *prev_ent = NULL;
 	struct scoutfs_link_backref_entry *ent = NULL;
 	struct scoutfs_lock *lock = NULL;
 	struct scoutfs_key last_key;
 	struct scoutfs_key key;
+	int nr = 0;
 	int len;
 	int ret;
 
-	ent = kmalloc(offsetof(struct scoutfs_link_backref_entry,
-			       dent.name[SCOUTFS_NAME_LEN]), GFP_KERNEL);
-	if (!ent) {
-		ret = -ENOMEM;
-		goto out;
-	}
-
-	INIT_LIST_HEAD(&ent->head);
-
 	init_dirent_key(&key, SCOUTFS_LINK_BACKREF_TYPE, ino, dir_ino, dir_pos);
-	init_dirent_key(&last_key, SCOUTFS_LINK_BACKREF_TYPE, ino, U64_MAX,
-			U64_MAX);
+	init_dirent_key(&last_key, SCOUTFS_LINK_BACKREF_TYPE, ino, U64_MAX, U64_MAX);
 
 	ret = scoutfs_lock_ino(sb, SCOUTFS_LOCK_READ, 0, ino, &lock);
 	if (ret)
 		goto out;
 
-	ret = scoutfs_item_next(sb, &key, &last_key, &ent->dent,
-				dirent_bytes(SCOUTFS_NAME_LEN), lock);
-	scoutfs_unlock(sb, lock, SCOUTFS_LOCK_READ);
-	lock = NULL;
-	if (ret < 0)
-		goto out;
+	while (nr < count) {
+		ent = kmalloc(offsetof(struct scoutfs_link_backref_entry,
+				       dent.name[SCOUTFS_NAME_LEN]), GFP_NOFS);
+		if (!ent) {
+			ret = -ENOMEM;
+			goto out;
+		}
 
-	len = ret - sizeof(struct scoutfs_dirent);
-	if (len < 1 || len > SCOUTFS_NAME_LEN) {
-		scoutfs_corruption(sb, SC_DIRENT_BACKREF_NAME_LEN,
-				   corrupt_dirent_backref_name_len,
-				   "ino %llu dir_ino %llu pos %llu key "SK_FMT" len %d",
-				   ino, dir_ino, dir_pos, SK_ARG(&key), len);
-		ret = -EIO;
-		goto out;
+		INIT_LIST_HEAD(&ent->head);
+
+		ret = scoutfs_item_next(sb, &key, &last_key, &ent->dent,
+					dirent_bytes(SCOUTFS_NAME_LEN), lock);
+		if (ret < 0) {
+			if (ret == -ENOENT && prev_ent)
+				prev_ent->last = true;
+			goto out;
+		}
+
+		len = ret - sizeof(struct scoutfs_dirent);
+		if (len < 1 || len > SCOUTFS_NAME_LEN) {
+			scoutfs_corruption(sb, SC_DIRENT_BACKREF_NAME_LEN,
+					   corrupt_dirent_backref_name_len,
+					   "ino %llu dir_ino %llu pos %llu key "SK_FMT" len %d",
+					   ino, dir_ino, dir_pos, SK_ARG(&key), len);
+			ret = -EIO;
+			goto out;
+		}
+
+		ent->dir_ino = le64_to_cpu(key.skd_major);
+		ent->dir_pos = le64_to_cpu(key.skd_minor);
+		ent->name_len = len;
+		ent->d_type = dentry_type(ent->dent.type);
+		ent->last = false;
+
+		trace_scoutfs_dir_add_next_linkref_found(sb, ino, ent->dir_ino, ent->dir_pos,
+							 ent->name_len);
+
+		list_add(&ent->head, list);
+		prev_ent = ent;
+		ent = NULL;
+		nr++;
+		scoutfs_key_inc(&key);
 	}
 
-	list_add(&ent->head, list);
-	ent->dir_ino = le64_to_cpu(key.skd_major);
-	ent->dir_pos = le64_to_cpu(key.skd_minor);
-	ent->name_len = len;
 	ret = 0;
 out:
-	trace_scoutfs_dir_add_next_linkref(sb, ino, dir_ino, dir_pos, ret,
-					   ent ? ent->dir_ino : 0,
-					   ent ? ent->dir_pos : 0,
-					   ent ? ent->name_len : 0);
+	scoutfs_unlock(sb, lock, SCOUTFS_LOCK_READ);
+	trace_scoutfs_dir_add_next_linkrefs(sb, ino, dir_ino, dir_pos, count, nr, ret);
 
-	if (ent && list_empty(&ent->head))
-		kfree(ent);
-	return ret;
+	kfree(ent);
+	return nr ?: ret;
 }
 
 static u64 first_backref_dir_ino(struct list_head *list)
@@ -1396,7 +1469,7 @@ retry:
 	}
 
 	/* get the next link name to the given inode */
-	ret = scoutfs_dir_add_next_linkref(sb, ino, dir_ino, dir_pos, list);
+	ret = scoutfs_dir_add_next_linkrefs(sb, ino, dir_ino, dir_pos, 1, list);
 	if (ret < 0)
 		goto out;
 
@@ -1404,7 +1477,7 @@ retry:
 	par_ino = first_backref_dir_ino(list);
 	while (par_ino != SCOUTFS_ROOT_INO) {
 
-		ret = scoutfs_dir_add_next_linkref(sb, par_ino, 0, 0, list);
+		ret = scoutfs_dir_add_next_linkrefs(sb, par_ino, 0, 0, 1, list);
 		if (ret < 0) {
 			if (ret == -ENOENT) {
 				/* restart if there was no parent component */
@@ -1416,6 +1489,8 @@ retry:
 
 		par_ino = first_backref_dir_ino(list);
 	}
+
+	ret = 0;
 out:
 	if (ret < 0)
 		scoutfs_dir_free_backref_path(sb, list);
@@ -1538,7 +1613,7 @@ static int scoutfs_rename_common(struct inode *old_dir,
 	struct scoutfs_lock *orph_lock = NULL;
 	struct scoutfs_dirent new_dent;
 	struct scoutfs_dirent old_dent;
-	struct timespec now;
+	struct kc_timespec now;
 	bool ins_new = false;
 	bool del_new = false;
 	bool ins_old = false;
@@ -1704,7 +1779,7 @@ retry:
 		inc_nlink(new_dir);
 	}
 
-	now = CURRENT_TIME;
+	now = current_time(old_inode);
 	old_dir->i_ctime = now;
 	old_dir->i_mtime = now;
 	if (new_dir != old_dir) {
@@ -1791,12 +1866,14 @@ out_unlock:
 	return ret;
 }
 
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 static int scoutfs_rename(struct inode *old_dir,
 			  struct dentry *old_dentry, struct inode *new_dir,
 			  struct dentry *new_dentry)
 {
 	return scoutfs_rename_common(old_dir, old_dentry, new_dir, new_dentry, 0);
 }
+#endif
 
 static int scoutfs_rename2(struct inode *old_dir,
 			  struct dentry *old_dentry, struct inode *new_dir,
@@ -1841,7 +1918,7 @@ static int scoutfs_tmpfile(struct inode *dir, struct dentry *dentry, umode_t mod
 	if (ret < 0)
 		goto out; /* XXX returning error but items created */
 
-	inode->i_mtime = inode->i_atime = inode->i_ctime = CURRENT_TIME;
+	inode->i_mtime = inode->i_atime = inode->i_ctime = current_time(inode);
 	si->crtime = inode->i_mtime;
 	insert_inode_hash(inode);
 	ihold(inode); /* need to update inode modifications in d_tmpfile */
@@ -1866,6 +1943,37 @@ out:
 	return ret;
 }
 
+const struct inode_operations scoutfs_symlink_iops = {
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
+	.readlink       = generic_readlink,
+	.follow_link    = scoutfs_follow_link,
+	.put_link       = scoutfs_put_link,
+#else
+	.get_link	= scoutfs_get_link,
+#endif
+	.getattr	= scoutfs_getattr,
+	.setattr	= scoutfs_setattr,
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
+	.setxattr	= generic_setxattr,
+	.getxattr	= generic_getxattr,
+#endif
+	.listxattr	= scoutfs_listxattr,
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
+	.removexattr	= generic_removexattr,
+#endif
+	.get_acl	= scoutfs_get_acl,
+#ifndef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
+	.tmpfile	= scoutfs_tmpfile,
+	.rename		= scoutfs_rename_common,
+	.symlink	= scoutfs_symlink,
+	.unlink		= scoutfs_unlink,
+	.link		= scoutfs_link,
+	.mkdir		= scoutfs_mkdir,
+	.create		= scoutfs_create,
+	.lookup		= scoutfs_lookup,
+#endif
+};
+
 const struct file_operations scoutfs_dir_fops = {
 	.KC_FOP_READDIR	= scoutfs_readdir,
 #ifdef KC_FMODE_KABI_ITERATE
@@ -1877,9 +1985,12 @@ const struct file_operations scoutfs_dir_fops = {
 };
 
 
-
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 const struct inode_operations_wrapper scoutfs_dir_iops = {
 	.ops = {
+#else
+const struct inode_operations scoutfs_dir_iops = {
+#endif
 	.lookup		= scoutfs_lookup,
 	.mknod		= scoutfs_mknod,
 	.create		= scoutfs_create,
@@ -1887,17 +1998,25 @@ const struct inode_operations_wrapper scoutfs_dir_iops = {
 	.link		= scoutfs_link,
 	.unlink		= scoutfs_unlink,
 	.rmdir		= scoutfs_unlink,
-	.rename		= scoutfs_rename,
 	.getattr	= scoutfs_getattr,
 	.setattr	= scoutfs_setattr,
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
+	.rename		= scoutfs_rename,
 	.setxattr	= generic_setxattr,
 	.getxattr	= generic_getxattr,
-	.listxattr	= scoutfs_listxattr,
 	.removexattr	= generic_removexattr,
+#endif
+	.listxattr	= scoutfs_listxattr,
 	.get_acl	= scoutfs_get_acl,
 	.symlink	= scoutfs_symlink,
 	.permission	= scoutfs_permission,
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 	},
+#endif
 	.tmpfile	= scoutfs_tmpfile,
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 	.rename2	= scoutfs_rename2,
+#else
+	.rename		= scoutfs_rename2,
+#endif
 };

kmod/src/dir.h

@@ -5,7 +5,11 @@
 #include "lock.h"
 
 extern const struct file_operations scoutfs_dir_fops;
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 extern const struct inode_operations_wrapper scoutfs_dir_iops;
+#else
+extern const struct inode_operations scoutfs_dir_iops;
+#endif
 extern const struct inode_operations scoutfs_symlink_iops;
 
 extern const struct dentry_operations scoutfs_dentry_ops;
@@ -15,6 +19,8 @@ struct scoutfs_link_backref_entry {
 	u64 dir_ino;
 	u64 dir_pos;
 	u16 name_len;
+	u8 d_type;
+	bool last;
 	struct scoutfs_dirent dent;
 	/* the full name is allocated and stored in dent.name[] */
 };
@@ -24,9 +30,8 @@ int scoutfs_dir_get_backref_path(struct super_block *sb, u64 ino, u64 dir_ino,
 void scoutfs_dir_free_backref_path(struct super_block *sb,
 				   struct list_head *list);
 
-int scoutfs_dir_add_next_linkref(struct super_block *sb, u64 ino,
-				 u64 dir_ino, u64 dir_pos,
-				 struct list_head *list);
+int scoutfs_dir_add_next_linkrefs(struct super_block *sb, u64 ino, u64 dir_ino, u64 dir_pos,
+				  int count, struct list_head *list);
 
 int scoutfs_symlink_drop(struct super_block *sb, u64 ino,
 			 struct scoutfs_lock *lock, u64 i_size);

kmod/src/export.c

@@ -114,8 +114,8 @@ static struct dentry *scoutfs_get_parent(struct dentry *child)
 	int ret;
 	u64 ino;
 
-	ret = scoutfs_dir_add_next_linkref(sb, scoutfs_ino(inode), 0, 0, &list);
-	if (ret)
+	ret = scoutfs_dir_add_next_linkrefs(sb, scoutfs_ino(inode), 0, 0, 1, &list);
+	if (ret < 0)
 		return ERR_PTR(ret);
 
 	ent = list_first_entry(&list, struct scoutfs_link_backref_entry, head);
@@ -138,9 +138,9 @@ static int scoutfs_get_name(struct dentry *parent, char *name,
 	LIST_HEAD(list);
 	int ret;
 
-	ret = scoutfs_dir_add_next_linkref(sb, scoutfs_ino(inode), dir_ino,
-					   0, &list);
-	if (ret)
+	ret = scoutfs_dir_add_next_linkrefs(sb, scoutfs_ino(inode), dir_ino,
+					    0, 1, &list);
+	if (ret < 0)
 		return ret;
 
 	ret = -ENOENT;

kmod/src/file.c

@@ -29,6 +29,7 @@
 #include "per_task.h"
 #include "omap.h"
 
+#ifdef KC_LINUX_HAVE_FOP_AIO_READ
 /*
  * Start a high level file read.  We check for offline extents in the
  * read region here so that we only check the extents once.  We use the
@@ -42,27 +43,27 @@ ssize_t scoutfs_file_aio_read(struct kiocb *iocb, const struct iovec *iov,
 	struct inode *inode = file_inode(file);
 	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
 	struct super_block *sb = inode->i_sb;
-	struct scoutfs_lock *inode_lock = NULL;
+	struct scoutfs_lock *scoutfs_inode_lock = NULL;
 	SCOUTFS_DECLARE_PER_TASK_ENTRY(pt_ent);
 	DECLARE_DATA_WAIT(dw);
 	int ret;
 
 retry:
 	/* protect checked extents from release */
-	mutex_lock(&inode->i_mutex);
+	inode_lock(inode);
 	atomic_inc(&inode->i_dio_count);
-	mutex_unlock(&inode->i_mutex);
+	inode_unlock(inode);
 
 	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_READ,
-				 SCOUTFS_LKF_REFRESH_INODE, inode, &inode_lock);
+				 SCOUTFS_LKF_REFRESH_INODE, inode, &scoutfs_inode_lock);
 	if (ret)
 		goto out;
 
-	if (scoutfs_per_task_add_excl(&si->pt_data_lock, &pt_ent, inode_lock)) {
+	if (scoutfs_per_task_add_excl(&si->pt_data_lock, &pt_ent, scoutfs_inode_lock)) {
 		ret = scoutfs_data_wait_check_iov(inode, iov, nr_segs, pos,
 						  SEF_OFFLINE,
 						  SCOUTFS_IOC_DWO_READ,
-						  &dw, inode_lock);
+						  &dw, scoutfs_inode_lock);
 		if (ret != 0)
 			goto out;
 	} else {
@@ -74,7 +75,7 @@ retry:
 out:
 	inode_dio_done(inode);
 	scoutfs_per_task_del(&si->pt_data_lock, &pt_ent);
-	scoutfs_unlock(sb, inode_lock, SCOUTFS_LOCK_READ);
+	scoutfs_unlock(sb, scoutfs_inode_lock, SCOUTFS_LOCK_READ);
 
 	if (scoutfs_data_wait_found(&dw)) {
 		ret = scoutfs_data_wait(inode, &dw);
@@ -92,7 +93,7 @@ ssize_t scoutfs_file_aio_write(struct kiocb *iocb, const struct iovec *iov,
 	struct inode *inode = file_inode(file);
 	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
 	struct super_block *sb = inode->i_sb;
-	struct scoutfs_lock *inode_lock = NULL;
+	struct scoutfs_lock *scoutfs_inode_lock = NULL;
 	SCOUTFS_DECLARE_PER_TASK_ENTRY(pt_ent);
 	DECLARE_DATA_WAIT(dw);
 	int ret;
@@ -101,22 +102,22 @@ ssize_t scoutfs_file_aio_write(struct kiocb *iocb, const struct iovec *iov,
 		return 0;
 
 retry:
-	mutex_lock(&inode->i_mutex);
+	inode_lock(inode);
 	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_WRITE,
-				 SCOUTFS_LKF_REFRESH_INODE, inode, &inode_lock);
+				 SCOUTFS_LKF_REFRESH_INODE, inode, &scoutfs_inode_lock);
 	if (ret)
 		goto out;
 
-	ret = scoutfs_complete_truncate(inode, inode_lock);
+	ret = scoutfs_complete_truncate(inode, scoutfs_inode_lock);
 	if (ret)
 		goto out;
 
-	if (scoutfs_per_task_add_excl(&si->pt_data_lock, &pt_ent, inode_lock)) {
+	if (scoutfs_per_task_add_excl(&si->pt_data_lock, &pt_ent, scoutfs_inode_lock)) {
 		/* data_version is per inode, whole file must be online */
 		ret = scoutfs_data_wait_check(inode, 0, i_size_read(inode),
 					      SEF_OFFLINE,
 					      SCOUTFS_IOC_DWO_WRITE,
-					      &dw, inode_lock);
+					      &dw, scoutfs_inode_lock);
 		if (ret != 0)
 			goto out;
 	}
@@ -127,8 +128,8 @@ retry:
 
 out:
 	scoutfs_per_task_del(&si->pt_data_lock, &pt_ent);
-	scoutfs_unlock(sb, inode_lock, SCOUTFS_LOCK_WRITE);
-	mutex_unlock(&inode->i_mutex);
+	scoutfs_unlock(sb, scoutfs_inode_lock, SCOUTFS_LOCK_WRITE);
+	inode_unlock(inode);
 
 	if (scoutfs_data_wait_found(&dw)) {
 		ret = scoutfs_data_wait(inode, &dw);
@@ -146,6 +147,113 @@ out:
 
 	return ret;
 }
+#else
+ssize_t scoutfs_file_read_iter(struct kiocb *iocb, struct iov_iter *to)
+{
+	struct file *file = iocb->ki_filp;
+	struct inode *inode = file_inode(file);
+	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
+	struct super_block *sb = inode->i_sb;
+	struct scoutfs_lock *scoutfs_inode_lock = NULL;
+	SCOUTFS_DECLARE_PER_TASK_ENTRY(pt_ent);
+	DECLARE_DATA_WAIT(dw);
+	int ret;
+
+retry:
+	/* protect checked extents from release */
+	inode_lock(inode);
+	atomic_inc(&inode->i_dio_count);
+	inode_unlock(inode);
+
+	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_READ,
+				 SCOUTFS_LKF_REFRESH_INODE, inode, &scoutfs_inode_lock);
+	if (ret)
+		goto out;
+
+	if (scoutfs_per_task_add_excl(&si->pt_data_lock, &pt_ent, scoutfs_inode_lock)) {
+		ret = scoutfs_data_wait_check_iter(inode, iocb->ki_pos, to,
+						   SEF_OFFLINE,
+						   SCOUTFS_IOC_DWO_READ,
+						   &dw, scoutfs_inode_lock);
+		if (ret != 0)
+			goto out;
+	} else {
+		WARN_ON_ONCE(true);
+	}
+
+	ret = generic_file_read_iter(iocb, to);
+
+out:
+	inode_dio_end(inode);
+	scoutfs_per_task_del(&si->pt_data_lock, &pt_ent);
+	scoutfs_unlock(sb, scoutfs_inode_lock, SCOUTFS_LOCK_READ);
+
+	if (scoutfs_data_wait_found(&dw)) {
+		ret = scoutfs_data_wait(inode, &dw);
+		if (ret == 0)
+			goto retry;
+	}
+	return ret;
+}
+
+ssize_t scoutfs_file_write_iter(struct kiocb *iocb, struct iov_iter *from)
+{
+	struct file *file = iocb->ki_filp;
+	struct inode *inode = file_inode(file);
+	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
+	struct super_block *sb = inode->i_sb;
+	struct scoutfs_lock *scoutfs_inode_lock = NULL;
+	SCOUTFS_DECLARE_PER_TASK_ENTRY(pt_ent);
+	DECLARE_DATA_WAIT(dw);
+	int ret;
+	int written;
+
+retry:
+	inode_lock(inode);
+	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_WRITE,
+				 SCOUTFS_LKF_REFRESH_INODE, inode, &scoutfs_inode_lock);
+	if (ret)
+		goto out;
+
+	ret = generic_write_checks(iocb, from);
+	if (ret <= 0)
+		goto out;
+
+	ret = scoutfs_complete_truncate(inode, scoutfs_inode_lock);
+	if (ret)
+		goto out;
+
+	if (scoutfs_per_task_add_excl(&si->pt_data_lock, &pt_ent, scoutfs_inode_lock)) {
+		/* data_version is per inode, whole file must be online */
+		ret = scoutfs_data_wait_check_iter(inode, iocb->ki_pos, from,
+						   SEF_OFFLINE,
+						   SCOUTFS_IOC_DWO_WRITE,
+						   &dw, scoutfs_inode_lock);
+		if (ret != 0)
+			goto out;
+	}
+
+	/* XXX: remove SUID bit */
+
+	written = __generic_file_write_iter(iocb, from);
+
+out:
+	scoutfs_per_task_del(&si->pt_data_lock, &pt_ent);
+	scoutfs_unlock(sb, scoutfs_inode_lock, SCOUTFS_LOCK_WRITE);
+	inode_unlock(inode);
+
+	if (scoutfs_data_wait_found(&dw)) {
+		ret = scoutfs_data_wait(inode, &dw);
+		if (ret == 0)
+			goto retry;
+	}
+
+	if (ret > 0 || ret == -EIOCBQUEUED)
+		ret = generic_write_sync(iocb, written);
+
+	return written ? written : ret;
+}
+#endif
 
 int scoutfs_permission(struct inode *inode, int mask)
 {

kmod/src/file.h

@@ -1,10 +1,15 @@
 #ifndef _SCOUTFS_FILE_H_
 #define _SCOUTFS_FILE_H_
 
+#ifdef KC_LINUX_HAVE_FOP_AIO_READ
 ssize_t scoutfs_file_aio_read(struct kiocb *iocb, const struct iovec *iov,
 			      unsigned long nr_segs, loff_t pos);
 ssize_t scoutfs_file_aio_write(struct kiocb *iocb, const struct iovec *iov,
 			       unsigned long nr_segs, loff_t pos);
+#else
+ssize_t scoutfs_file_read_iter(struct kiocb *, struct iov_iter *);
+ssize_t scoutfs_file_write_iter(struct kiocb *, struct iov_iter *);
+#endif
 int scoutfs_permission(struct inode *inode, int mask);
 loff_t scoutfs_file_llseek(struct file *file, loff_t offset, int whence);
 

kmod/src/format.h

@@ -683,16 +683,19 @@ struct scoutfs_xattr_totl_val {
 #define SCOUTFS_QUORUM_ELECT_VAR_MS	100
 
 /*
- * Once a leader is elected they send out heartbeats at regular
- * intervals to force members to wait the much longer heartbeat timeout.
- * Once heartbeat timeout expires without receiving a heartbeat they'll
- * switch over the performing elections.
+ * Once a leader is elected they send heartbeat messages to all quorum
+ * members at regular intervals to force members to wait the much longer
+ * heartbeat timeout.  Once the heartbeat timeout expires without
+ * receiving a heartbeat message a member will start an election.
  *
  * These determine how long it could take members to notice that a
- * leader has gone silent and start to elect a new leader.
+ * leader has gone silent and start to elect a new leader.  The
+ * heartbeat timeout can be changed at run time by options.
  */
 #define SCOUTFS_QUORUM_HB_IVAL_MS	100
-#define SCOUTFS_QUORUM_HB_TIMEO_MS	(5 * MSEC_PER_SEC)
+#define SCOUTFS_QUORUM_MIN_HB_TIMEO_MS	(2 * MSEC_PER_SEC)
+#define SCOUTFS_QUORUM_DEF_HB_TIMEO_MS	(10 * MSEC_PER_SEC)
+#define SCOUTFS_QUORUM_MAX_HB_TIMEO_MS	(60 * MSEC_PER_SEC)
 
 /*
  * A newly elected leader will give fencing some time before giving up and

kmod/src/inode.c

@@ -143,10 +143,12 @@ void scoutfs_destroy_inode(struct inode *inode)
 static const struct inode_operations scoutfs_file_iops = {
 	.getattr	= scoutfs_getattr,
 	.setattr	= scoutfs_setattr,
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 	.setxattr	= generic_setxattr,
 	.getxattr	= generic_getxattr,
-	.listxattr	= scoutfs_listxattr,
 	.removexattr	= generic_removexattr,
+#endif
+	.listxattr	= scoutfs_listxattr,
 	.get_acl	= scoutfs_get_acl,
 	.fiemap		= scoutfs_data_fiemap,
 };
@@ -154,10 +156,12 @@ static const struct inode_operations scoutfs_file_iops = {
 static const struct inode_operations scoutfs_special_iops = {
 	.getattr	= scoutfs_getattr,
 	.setattr	= scoutfs_setattr,
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 	.setxattr	= generic_setxattr,
 	.getxattr	= generic_getxattr,
-	.listxattr	= scoutfs_listxattr,
 	.removexattr	= generic_removexattr,
+#endif
+	.listxattr	= scoutfs_listxattr,
 	.get_acl	= scoutfs_get_acl,
 };
 
@@ -174,8 +178,12 @@ static void set_inode_ops(struct inode *inode)
 		inode->i_fop = &scoutfs_file_fops;
 		break;
 	case S_IFDIR:
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 		inode->i_op = &scoutfs_dir_iops.ops;
 		inode->i_flags |= S_IOPS_WRAPPER;
+#else
+		inode->i_op = &scoutfs_dir_iops;
+#endif
 		inode->i_fop = &scoutfs_dir_fops;
 		break;
 	case S_IFLNK:
@@ -247,7 +255,7 @@ static void load_inode(struct inode *inode, struct scoutfs_inode *cinode)
 	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
 
 	i_size_write(inode, le64_to_cpu(cinode->size));
-	inode->i_version = le64_to_cpu(cinode->version);
+	inode_set_iversion_queried(inode, le64_to_cpu(cinode->version));
 	set_nlink(inode, le32_to_cpu(cinode->nlink));
 	i_uid_write(inode, le32_to_cpu(cinode->uid));
 	i_gid_write(inode, le32_to_cpu(cinode->gid));
@@ -340,10 +348,17 @@ int scoutfs_inode_refresh(struct inode *inode, struct scoutfs_lock *lock)
 	return ret;
 }
 
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 int scoutfs_getattr(struct vfsmount *mnt, struct dentry *dentry,
 		    struct kstat *stat)
 {
 	struct inode *inode = dentry->d_inode;
+#else
+int scoutfs_getattr(const struct path *path, struct kstat *stat,
+		    u32 request_mask, unsigned int query_flags)
+{
+	struct inode *inode = d_inode(path->dentry);
+#endif
 	struct super_block *sb = inode->i_sb;
 	struct scoutfs_lock *lock = NULL;
 	int ret;
@@ -384,7 +399,7 @@ static int set_inode_size(struct inode *inode, struct scoutfs_lock *lock,
 		scoutfs_inode_inc_data_version(inode);
 
 	truncate_setsize(inode, new_size);
-	inode->i_ctime = inode->i_mtime = CURRENT_TIME;
+	inode->i_ctime = inode->i_mtime = current_time(inode);
 	if (truncate)
 		si->flags |= SCOUTFS_INO_FLAG_TRUNCATE;
 	scoutfs_inode_set_data_seq(inode);
@@ -467,8 +482,7 @@ retry:
 				 SCOUTFS_LKF_REFRESH_INODE, inode, &lock);
 	if (ret)
 		return ret;
-
-	ret = inode_change_ok(inode, attr);
+	ret = setattr_prepare(dentry, attr);
 	if (ret)
 		goto out;
 
@@ -496,9 +510,9 @@ retry:
 				scoutfs_unlock(sb, lock, SCOUTFS_LOCK_WRITE);
 
 				/* XXX callee locks instead? */
-				mutex_unlock(&inode->i_mutex);
+				inode_unlock(inode);
 				ret = scoutfs_data_wait(inode, &dw);
-				mutex_lock(&inode->i_mutex);
+				inode_lock(inode);
 
 				if (ret == 0)
 					goto retry;
@@ -750,7 +764,7 @@ struct inode *scoutfs_iget(struct super_block *sb, u64 ino, int lkf, int igf)
 		/* XXX ensure refresh, instead clear in drop_inode? */
 		si = SCOUTFS_I(inode);
 		atomic64_set(&si->last_refreshed, 0);
-		inode->i_version = 0;
+		inode_set_iversion_queried(inode, 0);
 	}
 
 	ret = scoutfs_inode_refresh(inode, lock);
@@ -798,7 +812,7 @@ static void store_inode(struct scoutfs_inode *cinode, struct inode *inode)
 	scoutfs_inode_get_onoff(inode, &online_blocks, &offline_blocks);
 
 	cinode->size = cpu_to_le64(i_size_read(inode));
-	cinode->version = cpu_to_le64(inode->i_version);
+	cinode->version = cpu_to_le64(inode_peek_iversion(inode));
 	cinode->nlink = cpu_to_le32(inode->i_nlink);
 	cinode->uid = cpu_to_le32(i_uid_read(inode));
 	cinode->gid = cpu_to_le32(i_gid_read(inode));
@@ -1475,7 +1489,7 @@ int scoutfs_new_inode(struct super_block *sb, struct inode *dir, umode_t mode, d
 	inode->i_ino = ino; /* XXX overflow */
 	inode_init_owner(inode, dir, mode);
 	inode_set_bytes(inode, 0);
-	inode->i_mtime = inode->i_atime = inode->i_ctime = CURRENT_TIME;
+	inode->i_mtime = inode->i_atime = inode->i_ctime = current_time(inode);
 	inode->i_rdev = rdev;
 	set_inode_ops(inode);
 

kmod/src/inode.h

@@ -22,7 +22,7 @@ struct scoutfs_inode_info {
 	u64 online_blocks;
 	u64 offline_blocks;
 	u32 flags;
-	struct timespec crtime;
+	struct kc_timespec crtime;
 
 	/*
 	 * Protects per-inode extent items, most particularly readers
@@ -123,8 +123,13 @@ void scoutfs_inode_get_onoff(struct inode *inode, s64 *on, s64 *off);
 int scoutfs_complete_truncate(struct inode *inode, struct scoutfs_lock *lock);
 
 int scoutfs_inode_refresh(struct inode *inode, struct scoutfs_lock *lock);
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 int scoutfs_getattr(struct vfsmount *mnt, struct dentry *dentry,
 		    struct kstat *stat);
+#else
+int scoutfs_getattr(const struct path *path, struct kstat *stat,
+		    u32 request_mask, unsigned int query_flags);
+#endif
 int scoutfs_setattr(struct dentry *dentry, struct iattr *attr);
 
 int scoutfs_inode_orphan_create(struct super_block *sb, u64 ino, struct scoutfs_lock *lock,

kmod/src/ioctl.c

@@ -22,6 +22,7 @@
 #include <linux/sched.h>
 #include <linux/aio.h>
 #include <linux/list_sort.h>
+#include <linux/backing-dev.h>
 
 #include "format.h"
 #include "key.h"
@@ -302,7 +303,7 @@ static long scoutfs_ioc_release(struct file *file, unsigned long arg)
 	if (ret)
 		return ret;
 
-	mutex_lock(&inode->i_mutex);
+	inode_lock(inode);
 
 	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_WRITE,
 				 SCOUTFS_LKF_REFRESH_INODE, inode, &lock);
@@ -351,7 +352,7 @@ static long scoutfs_ioc_release(struct file *file, unsigned long arg)
 
 out:
 	scoutfs_unlock(sb, lock, SCOUTFS_LOCK_WRITE);
-	mutex_unlock(&inode->i_mutex);
+	inode_unlock(inode);
 	mnt_drop_write_file(file);
 
 	trace_scoutfs_ioc_release_ret(sb, scoutfs_ino(inode), ret);
@@ -393,7 +394,7 @@ static long scoutfs_ioc_data_wait_err(struct file *file, unsigned long arg)
 		goto out;
 	}
 
-	mutex_lock(&inode->i_mutex);
+	inode_lock(inode);
 
 	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_READ,
 				 SCOUTFS_LKF_REFRESH_INODE, inode, &lock);
@@ -411,7 +412,7 @@ static long scoutfs_ioc_data_wait_err(struct file *file, unsigned long arg)
 
 	scoutfs_unlock(sb, lock, SCOUTFS_LOCK_READ);
 unlock:
-	mutex_unlock(&inode->i_mutex);
+	inode_unlock(inode);
 	iput(inode);
 out:
 	return ret;
@@ -448,7 +449,6 @@ static long scoutfs_ioc_stage(struct file *file, unsigned long arg)
 {
 	struct inode *inode = file_inode(file);
 	struct super_block *sb = inode->i_sb;
-	struct address_space *mapping = inode->i_mapping;
 	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
 	SCOUTFS_DECLARE_PER_TASK_ENTRY(pt_ent);
 	struct scoutfs_ioctl_stage args;
@@ -480,8 +480,10 @@ static long scoutfs_ioc_stage(struct file *file, unsigned long arg)
 	/* the iocb is really only used for the file pointer :P */
 	init_sync_kiocb(&kiocb, file);
 	kiocb.ki_pos = args.offset;
+#ifdef KC_LINUX_AIO_KI_LEFT
 	kiocb.ki_left = args.length;
 	kiocb.ki_nbytes = args.length;
+#endif
 	iov.iov_base = (void __user *)(unsigned long)args.buf_ptr;
 	iov.iov_len = args.length;
 
@@ -489,7 +491,7 @@ static long scoutfs_ioc_stage(struct file *file, unsigned long arg)
 	if (ret)
 		return ret;
 
-	mutex_lock(&inode->i_mutex);
+	inode_lock(inode);
 
 	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_WRITE,
 				 SCOUTFS_LKF_REFRESH_INODE, inode, &lock);
@@ -516,7 +518,7 @@ static long scoutfs_ioc_stage(struct file *file, unsigned long arg)
 	}
 
 	si->staging = true;
-	current->backing_dev_info = mapping->backing_dev_info;
+	current->backing_dev_info = inode_to_bdi(inode);
 
 	pos = args.offset;
 	written = 0;
@@ -533,7 +535,7 @@ static long scoutfs_ioc_stage(struct file *file, unsigned long arg)
 out:
 	scoutfs_per_task_del(&si->pt_data_lock, &pt_ent);
 	scoutfs_unlock(sb, lock, SCOUTFS_LOCK_WRITE);
-	mutex_unlock(&inode->i_mutex);
+	inode_unlock(inode);
 	mnt_drop_write_file(file);
 
 	trace_scoutfs_ioc_stage_ret(sb, scoutfs_ino(inode), ret);
@@ -652,7 +654,7 @@ static long scoutfs_ioc_setattr_more(struct file *file, unsigned long arg)
 	if (ret)
 		goto out;
 
-	mutex_lock(&inode->i_mutex);
+	inode_lock(inode);
 
 	ret = scoutfs_lock_inode(sb, SCOUTFS_LOCK_WRITE,
 				 SCOUTFS_LKF_REFRESH_INODE, inode, &lock);
@@ -696,7 +698,7 @@ static long scoutfs_ioc_setattr_more(struct file *file, unsigned long arg)
 unlock:
 	scoutfs_inode_index_unlock(sb, &ind_locks);
 	scoutfs_unlock(sb, lock, SCOUTFS_LOCK_WRITE);
-	mutex_unlock(&inode->i_mutex);
+	inode_unlock(inode);
 	mnt_drop_write_file(file);
 out:
 
@@ -1398,6 +1400,110 @@ out:
 	return ret ?: nr;
 }
 
+/*
+ * Copy entries that point to an inode to the user's buffer.  We copy to
+ * userspace from copies of the entries that are acquired under a lock
+ * so that we don't fault while holding cluster locks.  It also gives us
+ * a chance to limit the amount of work under each lock hold.
+ */
+static long scoutfs_ioc_get_referring_entries(struct file *file, unsigned long arg)
+{
+	struct super_block *sb = file_inode(file)->i_sb;
+	struct scoutfs_ioctl_get_referring_entries gre;
+	struct scoutfs_link_backref_entry *bref = NULL;
+	struct scoutfs_link_backref_entry *bref_tmp;
+	struct scoutfs_ioctl_dirent __user *uent;
+	struct scoutfs_ioctl_dirent ent;
+	LIST_HEAD(list);
+	u64 copied;
+	int name_len;
+	int bytes;
+	long nr;
+	int ret;
+
+	if (!capable(CAP_DAC_READ_SEARCH))
+		return -EPERM;
+
+	if (copy_from_user(&gre, (void __user *)arg, sizeof(gre)))
+		return -EFAULT;
+
+	uent = (void __user *)(unsigned long)gre.entries_ptr;
+	copied = 0;
+	nr = 0;
+
+	/* use entry as cursor between calls */
+	ent.dir_ino = gre.dir_ino;
+	ent.dir_pos = gre.dir_pos;
+
+	for (;;) {
+		ret = scoutfs_dir_add_next_linkrefs(sb, gre.ino, ent.dir_ino, ent.dir_pos, 1024,
+						    &list);
+		if (ret < 0) {
+			if (ret == -ENOENT)
+				ret = 0;
+			goto out;
+		}
+
+		/* _add_next adds each entry to the head, _reverse for key order */
+		list_for_each_entry_safe_reverse(bref, bref_tmp, &list, head) {
+			list_del_init(&bref->head);
+
+			name_len = bref->name_len;
+			bytes = ALIGN(offsetof(struct scoutfs_ioctl_dirent, name[name_len + 1]),
+				      16);
+			if (copied + bytes > gre.entries_bytes) {
+				ret = -EINVAL;
+				goto out;
+			}
+
+			ent.dir_ino = bref->dir_ino;
+			ent.dir_pos = bref->dir_pos;
+			ent.ino = gre.ino;
+			ent.entry_bytes = bytes;
+			ent.flags = bref->last ? SCOUTFS_IOCTL_DIRENT_FLAG_LAST : 0;
+			ent.d_type = bref->d_type;
+			ent.name_len = name_len;
+
+			if (copy_to_user(uent, &ent, sizeof(struct scoutfs_ioctl_dirent)) ||
+			    copy_to_user(&uent->name[0], bref->dent.name, name_len) ||
+			    put_user('\0', &uent->name[name_len])) {
+				ret = -EFAULT;
+				goto out;
+			}
+
+			kfree(bref);
+			bref = NULL;
+
+			uent = (void __user *)uent + bytes;
+			copied += bytes;
+			nr++;
+
+			if (nr == LONG_MAX || (ent.flags & SCOUTFS_IOCTL_DIRENT_FLAG_LAST)) {
+				ret = 0;
+				goto out;
+			}
+		}
+
+		/* advance cursor pos from last copied entry */
+		if (++ent.dir_pos == 0) {
+			if (++ent.dir_ino == 0) {
+				ret = 0;
+				goto out;
+			}
+		}
+	}
+
+	ret = 0;
+out:
+	kfree(bref);
+	list_for_each_entry_safe(bref, bref_tmp, &list, head) {
+		list_del_init(&bref->head);
+		kfree(bref);
+	}
+
+	return nr ?: ret;
+}
+
 long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 {
 	switch (cmd) {
@@ -1433,6 +1539,8 @@ long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 		return scoutfs_ioc_read_xattr_totals(file, arg);
 	case SCOUTFS_IOC_GET_ALLOCATED_INOS:
 		return scoutfs_ioc_get_allocated_inos(file, arg);
+	case SCOUTFS_IOC_GET_REFERRING_ENTRIES:
+		return scoutfs_ioc_get_referring_entries(file, arg);
 	}
 
 	return -ENOTTY;

kmod/src/ioctl.h

@@ -559,4 +559,118 @@ struct scoutfs_ioctl_get_allocated_inos {
 #define SCOUTFS_IOC_GET_ALLOCATED_INOS \
 	_IOW(SCOUTFS_IOCTL_MAGIC, 16, struct scoutfs_ioctl_get_allocated_inos)
 
+/*
+ * Get directory entries that refer to a specific inode.
+ *
+ * @ino: The target ino that we're finding referring entries to.
+ * Constant across all the calls that make up an iteration over all the
+ * inode's entries.
+ *
+ * @dir_ino: The inode number of a directory containing the entry to our
+ * inode to search from.  If this parent directory contains no more
+ * entries to our inode then we'll search through other parent directory
+ * inodes in inode order.
+ *
+ * @dir_pos: The position in the dir_ino parent directory of the entry
+ * to our inode to search from.  If there is no entry at this position
+ * then we'll search through other entry positions in increasing order.
+ * If we exhaust the parent directory then we'll search through
+ * additional parent directories in inode order.
+ *
+ * @entries_ptr: A pointer to the buffer where found entries will be
+ * stored.  The pointer must be aligned to 16 bytes.
+ *
+ * @entries_bytes: The size of the buffer that will contain entries.
+ *
+ * To start iterating set the desired target ino, dir_ino to 0, dir_pos
+ * to 0, and set result_ptr and _bytes to a sufficiently large buffer.
+ * Each entry struct that's stored in the buffer adds some overhead so a
+ * large multiple of the largest possible name is a reasonable choice.
+ * (A few multiples of PATH_MAX perhaps.)
+ *
+ * Each call returns the total number of entries that were stored in the
+ * entries buffer.  Zero is returned when the search was successful and
+ * no referring entries were found.  The entries can be iterated over by
+ * advancing each starting struct offset by the total number of bytes in
+ * each entry.  If the _LAST flag is set on an entry then there were no
+ * more entries referring to the inode at the time of the call and
+ * iteration can be stopped.
+ *
+ * To resume iteration set the next call's starting dir_ino and dir_pos
+ * to one past the last entry seen.  Increment the last entry's dir_pos,
+ * and if it wrapped to 0, increment its dir_ino.
+ *
+ * This does not check that the caller has permission to read the
+ * entries found in each containing directory.  It requires
+ * CAP_DAC_READ_SEARCH which bypasses path traversal permissions
+ * checking.
+ *
+ * Entries returned by a single call can reflect any combination of
+ * racing creation and removal of entries.  Each entry existed at the
+ * time it was read though it may have changed in the time it took to
+ * return from the call.  The set of entries returned may no longer
+ * reflect the current set of entries and may not have existed at the
+ * same time.
+ *
+ * This has no knowledge of the life cycle of the inode.  It can return
+ * 0 when there are no referring entries because either the target inode
+ * doesn't exist, it is in the process of being deleted, or because it
+ * is still open while being unlinked.
+ *
+ * On success this returns the number of entries filled in the buffer.
+ * A return of 0 indicates that no entries referred to the inode.
+ *
+ * EINVAL is returned when there is a problem with the buffer.  Either
+ * it was not aligned or it was not large enough for the first entry.
+ *
+ * Many other errnos indicate hard failure to find the next entry.
+ */
+struct scoutfs_ioctl_get_referring_entries {
+	__u64 ino;
+	__u64 dir_ino;
+	__u64 dir_pos;
+	__u64 entries_ptr;
+	__u64 entries_bytes;
+};
+
+/*
+ * @dir_ino: The inode of the directory containing the entry.
+ *
+ * @dir_pos: The readdir f_pos position of the entry within the
+ * directory.
+ *
+ * @ino: The inode number of the target of the entry.
+ *
+ * @flags: Flags associated with this entry.
+ *
+ * @d_type: Inode type as specified with DT_ enum values in readdir(3).
+ *
+ * @entry_bytes: The total bytes taken by the entry in memory, including
+ * the name and any alignment padding.  The start of a following entry
+ * will be found after this number of bytes.
+ *
+ * @name_len: The number of bytes in the name not including the trailing
+ * null, ala strlen(3).
+ *
+ * @name: The null terminated name of the referring entry.  In the
+ * struct definition this array is sized to naturally align the struct.
+ * That number of padded bytes are not necessarily found in the buffer
+ * returned by _get_referring_entries;
+ */
+struct scoutfs_ioctl_dirent {
+	__u64 dir_ino;
+	__u64 dir_pos;
+	__u64 ino;
+	__u16 entry_bytes;
+	__u8  flags;
+	__u8  d_type;
+	__u8  name_len;
+	__u8  name[3];
+};
+
+#define SCOUTFS_IOCTL_DIRENT_FLAG_LAST (1 << 0)
+
+#define SCOUTFS_IOC_GET_REFERRING_ENTRIES \
+	_IOW(SCOUTFS_IOCTL_MAGIC, 17, struct scoutfs_ioctl_get_referring_entries)
+
 #endif

kmod/src/item.c

@@ -27,6 +27,7 @@
 #include "trans.h"
 #include "counters.h"
 #include "scoutfs_trace.h"
+#include "util.h"
 
 /*
  * The item cache maintains a consistent view of items that are read
@@ -76,8 +77,10 @@ struct item_cache_info {
 	/* almost always read, barely written */
 	struct super_block *sb;
 	struct item_percpu_pages __percpu *pcpu_pages;
-	struct shrinker shrinker;
+	KC_DEFINE_SHRINKER(shrinker);
+#ifdef KC_CPU_NOTIFIER
 	struct notifier_block notifier;
+#endif
 
 	/* often walked, but per-cpu refs are fast path */
 	rwlock_t rwlock;
@@ -2277,7 +2280,7 @@ int scoutfs_item_write_dirty(struct super_block *sb)
 		ret = -ENOMEM;
 		goto out;
 	}
-	list_add(&page->list, &pages);
+	list_add(&page->lru, &pages);
 
 	first = NULL;
 	prev = &first;
@@ -2290,7 +2293,7 @@ int scoutfs_item_write_dirty(struct super_block *sb)
 				ret = -ENOMEM;
 				goto out;
 			}
-			list_add(&second->list, &pages);
+			list_add(&second->lru, &pages);
 		}
 
 		/* read lock next sorted page, we're only dirty_list user */
@@ -2347,8 +2350,8 @@ int scoutfs_item_write_dirty(struct super_block *sb)
 	/* write all the dirty items into log btree blocks */
 	ret = scoutfs_forest_insert_list(sb, first);
 out:
-	list_for_each_entry_safe(page, second, &pages, list) {
-		list_del_init(&page->list);
+	list_for_each_entry_safe(page, second, &pages, lru) {
+		list_del_init(&page->lru);
 		__free_page(page);
 	}
 
@@ -2530,27 +2533,35 @@ retry:
 	put_pg(sb, right);
 }
 
+static unsigned long item_cache_count_objects(struct shrinker *shrink,
+					      struct shrink_control *sc)
+{
+	struct item_cache_info *cinf = KC_SHRINKER_CONTAINER_OF(shrink, struct item_cache_info);
+	struct super_block *sb = cinf->sb;
+
+	scoutfs_inc_counter(sb, item_cache_count_objects);
+
+	return shrinker_min_t_long((u64)(cinf->lru_pages));
+}
+
 /*
  * Shrink the size the item cache.  We're operating against the fast
  * path lock ordering and we skip pages if we can't acquire locks.  We
  * can run into dirty pages or pages with items that weren't visible to
  * the earliest active reader which must be skipped.
  */
-static int item_lru_shrink(struct shrinker *shrink,
-			   struct shrink_control *sc)
+static unsigned long item_cache_scan_objects(struct shrinker *shrink,
+					     struct shrink_control *sc)
 {
-	struct item_cache_info *cinf = container_of(shrink,
-						    struct item_cache_info,
-						    shrinker);
+	struct item_cache_info *cinf = KC_SHRINKER_CONTAINER_OF(shrink, struct item_cache_info);
 	struct super_block *sb = cinf->sb;
 	struct cached_page *tmp;
 	struct cached_page *pg;
+	unsigned long freed = 0;
 	u64 first_reader_seq;
-	int nr;
+	int nr = sc->nr_to_scan;
 
-	if (sc->nr_to_scan == 0)
-		goto out;
-	nr = sc->nr_to_scan;
+	scoutfs_inc_counter(sb, item_cache_scan_objects);
 
 	/* can't invalidate pages with items that weren't visible to first reader */
 	first_reader_seq = first_active_reader_seq(cinf);
@@ -2582,6 +2593,7 @@ static int item_lru_shrink(struct shrinker *shrink,
 		rbtree_erase(&pg->node, &cinf->pg_root);
 		invalidate_pcpu_page(pg);
 		write_unlock(&pg->rwlock);
+		freed++;
 
 		put_pg(sb, pg);
 
@@ -2591,10 +2603,11 @@ static int item_lru_shrink(struct shrinker *shrink,
 
 	write_unlock(&cinf->rwlock);
 	spin_unlock(&cinf->lru_lock);
-out:
-	return min_t(unsigned long, cinf->lru_pages, INT_MAX);
+
+	return freed;
 }
 
+#ifdef KC_CPU_NOTIFIER
 static int item_cpu_callback(struct notifier_block *nfb,
 			     unsigned long action, void *hcpu)
 {
@@ -2609,6 +2622,7 @@ static int item_cpu_callback(struct notifier_block *nfb,
 
 	return NOTIFY_OK;
 }
+#endif
 
 int scoutfs_item_setup(struct super_block *sb)
 {
@@ -2638,11 +2652,13 @@ int scoutfs_item_setup(struct super_block *sb)
 	for_each_possible_cpu(cpu)
 		init_pcpu_pages(cinf, cpu);
 
-	cinf->shrinker.shrink = item_lru_shrink;
-	cinf->shrinker.seeks = DEFAULT_SEEKS;
-	register_shrinker(&cinf->shrinker);
+	KC_INIT_SHRINKER_FUNCS(&cinf->shrinker, item_cache_count_objects,
+			       item_cache_scan_objects);
+	KC_REGISTER_SHRINKER(&cinf->shrinker);
+#ifdef KC_CPU_NOTIFIER
         cinf->notifier.notifier_call = item_cpu_callback;
         register_hotcpu_notifier(&cinf->notifier);
+#endif
 
 	sbi->item_cache_info = cinf;
 	return 0;
@@ -2662,8 +2678,10 @@ void scoutfs_item_destroy(struct super_block *sb)
 	if (cinf) {
 		BUG_ON(!list_empty(&cinf->active_list));
 
+#ifdef KC_CPU_NOTIFIER
 		unregister_hotcpu_notifier(&cinf->notifier);
-		unregister_shrinker(&cinf->shrinker);
+#endif
+		KC_UNREGISTER_SHRINKER(&cinf->shrinker);
 
 		for_each_possible_cpu(cpu)
 			drop_pcpu_pages(sb, cinf, cpu);

kmod/src/kernelcompat.c

@@ -0,0 +1,84 @@
+
+#include <linux/uio.h>
+
+#include "kernelcompat.h"
+
+#ifdef KC_SHRINKER_SHRINK
+#include <linux/shrinker.h>
+/*
+ * If a target doesn't have that .{count,scan}_objects() interface then
+ * we have a .shrink() helper that performs the shrink work in terms of
+ * count/scan.
+ */
+int kc_shrink_wrapper_fn(struct shrinker *shrink, struct shrink_control *sc)
+{
+	struct kc_shrinker_wrapper *wrapper = container_of(shrink, struct kc_shrinker_wrapper, shrink);
+	unsigned long nr;
+	unsigned long rc;
+
+	if (sc->nr_to_scan != 0) {
+		rc = wrapper->scan_objects(shrink, sc);
+		/* translate magic values to the equivalent for older kernels */
+		if (rc == SHRINK_STOP)
+			return -1;
+		else if (rc == SHRINK_EMPTY)
+			return 0;
+	}
+
+	nr = wrapper->count_objects(shrink, sc);
+
+	return min_t(unsigned long, nr, INT_MAX);
+}
+#endif
+
+#ifndef KC_CURRENT_TIME_INODE
+struct timespec64 kc_current_time(struct inode *inode)
+{
+	struct timespec64 now;
+	unsigned gran;
+
+	getnstimeofday64(&now);
+
+	if (unlikely(!inode->i_sb)) {
+		WARN(1, "current_time() called with uninitialized super_block in the inode");
+		return now;
+	}
+
+	gran = inode->i_sb->s_time_gran;
+
+	/* Avoid division in the common cases 1 ns and 1 s. */
+	if (gran == 1) {
+		/* nothing */
+	} else if (gran == NSEC_PER_SEC) {
+		now.tv_nsec = 0;
+	} else if (gran > 1 && gran < NSEC_PER_SEC) {
+		now.tv_nsec -= now.tv_nsec % gran;
+	} else {
+		WARN(1, "illegal file time granularity: %u", gran);
+	}
+
+	return now;
+}
+#endif
+
+#ifndef KC_GENERIC_FILE_BUFFERED_WRITE
+ssize_t
+kc_generic_file_buffered_write(struct kiocb *iocb, const struct iovec *iov,
+			       unsigned long nr_segs, loff_t pos, loff_t *ppos,
+			       size_t count, ssize_t written)
+{
+	struct file *file = iocb->ki_filp;
+	ssize_t status;
+	struct iov_iter i;
+
+	iov_iter_init(&i, WRITE, iov, nr_segs, count);
+	status = generic_perform_write(file, &i, pos);
+
+	if (likely(status >= 0)) {
+		written += status;
+		*ppos = pos + status;
+	}
+
+	return written ? written : status;
+}
+#endif

kmod/src/kernelcompat.h

@@ -1,8 +1,35 @@
 #ifndef _SCOUTFS_KERNELCOMPAT_H_
 #define _SCOUTFS_KERNELCOMPAT_H_
 
-#ifndef KC_ITERATE_DIR_CONTEXT
+#include <linux/kernel.h>
 #include <linux/fs.h>
+
+/*
+ * v4.15-rc3-4-gae5e165d855d
+ *
+ * new API for handling inode->i_version. This forces us to
+ * include this API where we need. We include it here for
+ * convenience instead of where it's needed.
+ */
+#ifdef KC_NEED_LINUX_IVERSION_H
+#include <linux/iversion.h>
+#else
+/*
+ * Kernels before above version will need to fall back to
+ * manipulating inode->i_version as previous with degraded
+ * methods.
+ */
+#define inode_set_iversion_queried(inode, val)	\
+do {						\
+	(inode)->i_version = val;		\
+} while (0)
+#define inode_peek_iversion(inode)		\
+({						\
+	(inode)->i_version;			\
+})
+#endif
+
+#ifndef KC_ITERATE_DIR_CONTEXT
 typedef filldir_t kc_readdir_ctx_t;
 #define KC_DECLARE_READDIR(name, file, dirent, ctx) name(file, dirent, ctx)
 #define KC_FOP_READDIR readdir
@@ -52,4 +79,198 @@ static inline int dir_emit_dots(struct file *file, void *dirent,
 #define kc_posix_acl_valid(user_ns, acl) posix_acl_valid(acl)
 #endif
 
+/*
+ * v3.6-rc1-24-gdbf2576e37da
+ *
+ * All workqueues are now non-reentrant, and the bit flag is removed
+ * shortly after its uses were removed.
+ */
+#ifndef WQ_NON_REENTRANT
+#define WQ_NON_REENTRANT 0
+#endif
+
+/*
+ * v3.18-rc2-19-gb5ae6b15bd73
+ *
+ * Folds d_materialise_unique into d_splice_alias. Note reversal
+ * of arguments (Also note Documentation/filesystems/porting.rst)
+ */
+#ifndef KC_D_MATERIALISE_UNIQUE
+#define d_materialise_unique(dentry, inode) d_splice_alias(inode, dentry)
+#endif
+
+/*
+ * v4.8-rc1-29-g31051c85b5e2
+ *
+ * fall back to inode_change_ok() if setattr_prepare() isn't available
+ */
+#ifndef KC_SETATTR_PREPARE
+#define setattr_prepare(dentry, attr) inode_change_ok(d_inode(dentry), attr)
+#endif
+
+#ifndef KC___POSIX_ACL_CREATE
+#define __posix_acl_create posix_acl_create
+#define __posix_acl_chmod posix_acl_chmod
+#endif
+
+#ifndef KC_PERCPU_COUNTER_ADD_BATCH
+#define percpu_counter_add_batch __percpu_counter_add
+#endif
+
+#ifndef KC_MEMALLOC_NOFS_SAVE
+#define memalloc_nofs_save memalloc_noio_save
+#define memalloc_nofs_restore memalloc_noio_restore
+#endif
+
+#ifdef KC_BIO_BI_OPF
+#define kc_bio_get_opf(bio)		\
+({					\
+	(bio)->bi_opf;			\
+})
+#define kc_bio_set_opf(bio, opf)	\
+do {					\
+	(bio)->bi_opf = opf;		\
+} while (0)
+#define kc_bio_set_sector(bio, sect)	\
+do {					\
+	(bio)->bi_iter.bi_sector = sect;\
+} while (0)
+#define kc_submit_bio(bio) submit_bio(bio)
+#else
+#define kc_bio_get_opf(bio)		\
+({					\
+	(bio)->bi_rw;			\
+})
+#define kc_bio_set_opf(bio, opf)	\
+do {					\
+	(bio)->bi_rw = opf;		\
+} while (0)
+#define kc_bio_set_sector(bio, sect)	\
+do {					\
+	(bio)->bi_sector = sect;	\
+} while (0)
+#define kc_submit_bio(bio)		\
+do {					\
+	submit_bio((bio)->bi_rw, bio);	\
+} while (0)
+#define bio_set_dev(bio, bdev)		\
+do {					\
+	(bio)->bi_bdev = (bdev);	\
+} while (0)
+#endif
+
+#ifdef KC_BIO_BI_STATUS
+#define KC_DECLARE_BIO_END_IO(name, bio)	name(bio)
+#define kc_bio_get_errno(bio)			({ blk_status_to_errno((bio)->bi_status); })
+#else
+#define KC_DECLARE_BIO_END_IO(name, bio)	name(bio, int _error_arg)
+#define kc_bio_get_errno(bio)			({ (int)((void)(bio), _error_arg); })
+#endif
+
+/*
+ * v4.13-rc1-6-ge462ec50cb5f
+ *
+ * MS_* (mount) flags from <linux/mount.h> should not be used in the kernel
+ * anymore from 4.x onwards. Instead, we need to use the SB_* (superblock) flags
+ */
+#ifndef SB_POSIXACL
+#define SB_POSIXACL MS_POSIXACL
+#define SB_I_VERSION MS_I_VERSION
+#endif
+
+#ifndef KC_CURRENT_TIME_INODE
+struct timespec64 kc_current_time(struct inode *inode);
+#define current_time kc_current_time
+#define kc_timespec timespec
+#else
+#define kc_timespec timespec64
+#endif
+
+#ifndef KC_SHRINKER_SHRINK
+
+#define KC_DEFINE_SHRINKER(name) struct shrinker name
+#define KC_INIT_SHRINKER_FUNCS(name, countfn, scanfn) do {	\
+	__typeof__(name) _shrink = (name);			\
+	_shrink->count_objects = (countfn);			\
+	_shrink->scan_objects = (scanfn);			\
+	_shrink->seeks = DEFAULT_SEEKS;			\
+} while (0)
+
+#define KC_SHRINKER_CONTAINER_OF(ptr, type) container_of(ptr, type, shrinker)
+#define KC_REGISTER_SHRINKER(ptr) (register_shrinker(ptr))
+#define KC_UNREGISTER_SHRINKER(ptr) (unregister_shrinker(ptr))
+#define KC_SHRINKER_FN(ptr) (ptr)
+#else
+
+#include <linux/shrinker.h>
+#ifndef SHRINK_STOP
+#define SHRINK_STOP (~0UL)
+#define SHRINK_EMPTY (~0UL - 1)
+#endif
+
+int kc_shrink_wrapper_fn(struct shrinker *shrink, struct shrink_control *sc);
+struct kc_shrinker_wrapper {
+	unsigned long (*count_objects)(struct shrinker *, struct shrink_control *sc);
+	unsigned long (*scan_objects)(struct shrinker *, struct shrink_control *sc);
+	struct shrinker shrink;
+};
+
+#define KC_DEFINE_SHRINKER(name) struct kc_shrinker_wrapper name;
+#define KC_INIT_SHRINKER_FUNCS(name, countfn, scanfn) do {	\
+	struct kc_shrinker_wrapper *_wrap = (name);		\
+	_wrap->count_objects = (countfn);			\
+	_wrap->scan_objects = (scanfn);				\
+	_wrap->shrink.shrink = kc_shrink_wrapper_fn;		\
+	_wrap->shrink.seeks = DEFAULT_SEEKS;			\
+} while (0)
+#define KC_SHRINKER_CONTAINER_OF(ptr, type) container_of(container_of(ptr, struct kc_shrinker_wrapper, shrink), type, shrinker)
+#define KC_REGISTER_SHRINKER(ptr) (register_shrinker(ptr.shrink))
+#define KC_UNREGISTER_SHRINKER(ptr) (unregister_shrinker(ptr.shrink))
+#define KC_SHRINKER_FN(ptr) (ptr.shrink)
+
+#endif /* KC_SHRINKER_SHRINK */
+
+#ifdef KC_KERNEL_GETSOCKNAME_ADDRLEN
+#include <linux/net.h>
+#include <linux/inet.h>
+static inline int kc_kernel_getsockname(struct socket *sock, struct sockaddr *addr)
+{
+	int addrlen = sizeof(struct sockaddr_in);
+	int ret = kernel_getsockname(sock, addr, &addrlen);
+	if (ret == 0 && addrlen != sizeof(struct sockaddr_in))
+		return -EAFNOSUPPORT;
+	else if (ret < 0)
+		return ret;
+
+	return sizeof(struct sockaddr_in);
+}
+static inline int kc_kernel_getpeername(struct socket *sock, struct sockaddr *addr)
+{
+	int addrlen = sizeof(struct sockaddr_in);
+	int ret = kernel_getpeername(sock, addr, &addrlen);
+	if (ret == 0 && addrlen != sizeof(struct sockaddr_in))
+		return -EAFNOSUPPORT;
+	else if (ret < 0)
+		return ret;
+
+	return sizeof(struct sockaddr_in);
+}
+#else
+#define kc_kernel_getsockname(sock, addr) kernel_getsockname(sock, addr)
+#define kc_kernel_getpeername(sock, addr) kernel_getpeername(sock, addr)
+#endif
+
+#ifdef KC_SOCK_CREATE_KERN_NET
+#define kc_sock_create_kern(family, type, proto, res) sock_create_kern(&init_net, family, type, proto, res)
+#else
+#define kc_sock_create_kern sock_create_kern
+#endif
+
+#ifndef KC_GENERIC_FILE_BUFFERED_WRITE
+ssize_t kc_generic_file_buffered_write(struct kiocb *iocb, const struct iovec *iov,
+               unsigned long nr_segs, loff_t pos, loff_t *ppos,
+               size_t count, ssize_t written);
+#define generic_file_buffered_write kc_generic_file_buffered_write
+#endif
+
 #endif

kmod/src/lock.c

@@ -12,7 +12,6 @@
  */
 #include <linux/kernel.h>
 #include <linux/fs.h>
-#include <linux/preempt_mask.h> /* a rhel shed.h needed preempt_offset? */
 #include <linux/sched.h>
 #include <linux/slab.h>
 #include <linux/mm.h>
@@ -36,6 +35,7 @@
 #include "xattr.h"
 #include "item.h"
 #include "omap.h"
+#include "util.h"
 
 /*
  * scoutfs uses a lock service to manage item cache consistency between
@@ -77,7 +77,7 @@ struct lock_info {
 	bool unmounting;
 	struct rb_root lock_tree;
 	struct rb_root lock_range_tree;
-	struct shrinker shrinker;
+	KC_DEFINE_SHRINKER(shrinker);
 	struct list_head lru_list;
 	unsigned long long lru_nr;
 	struct workqueue_struct *workq;
@@ -1346,7 +1346,7 @@ void scoutfs_lock_del_coverage(struct super_block *sb,
 bool scoutfs_lock_protected(struct scoutfs_lock *lock, struct scoutfs_key *key,
 			    enum scoutfs_lock_mode mode)
 {
-	signed char lock_mode = ACCESS_ONCE(lock->mode);
+	signed char lock_mode = READ_ONCE(lock->mode);
 
 	return lock_modes_match(lock_mode, mode) &&
 	       scoutfs_key_compare_ranges(key, key,
@@ -1401,6 +1401,17 @@ static void lock_shrink_worker(struct work_struct *work)
 	}
 }
 
+static unsigned long lock_count_objects(struct shrinker *shrink,
+					struct shrink_control *sc)
+{
+	struct lock_info *linfo = KC_SHRINKER_CONTAINER_OF(shrink, struct lock_info);
+	struct super_block *sb = linfo->sb;
+
+	scoutfs_inc_counter(sb, lock_count_objects);
+
+	return shrinker_min_t_long((u64)(linfo->lru_nr));
+}
+
 /*
  * Start the shrinking process for locks on the lru.  If a lock is on
  * the lru then it can't have any active users.  We don't want to block
@@ -1413,21 +1424,18 @@ static void lock_shrink_worker(struct work_struct *work)
  * mode which will prevent the lock from being freed when the null
  * response arrives.
  */
-static int scoutfs_lock_shrink(struct shrinker *shrink,
-			       struct shrink_control *sc)
+static unsigned long lock_scan_objects(struct shrinker *shrink,
+				       struct shrink_control *sc)
 {
-	struct lock_info *linfo = container_of(shrink, struct lock_info,
-					       shrinker);
+	struct lock_info *linfo = KC_SHRINKER_CONTAINER_OF(shrink, struct lock_info);
 	struct super_block *sb = linfo->sb;
 	struct scoutfs_lock *lock;
 	struct scoutfs_lock *tmp;
-	unsigned long nr;
+	unsigned long freed = 0;
+	unsigned long nr = sc->nr_to_scan;
 	bool added = false;
-	int ret;
 
-	nr = sc->nr_to_scan;
-	if (nr == 0)
-		goto out;
+	scoutfs_inc_counter(sb, lock_scan_objects);
 
 	spin_lock(&linfo->lock);
 
@@ -1445,6 +1453,7 @@ restart:
 		lock->request_pending = 1;
 		list_add_tail(&lock->shrink_head, &linfo->shrink_list);
 		added = true;
+		freed++;
 
 		scoutfs_inc_counter(sb, lock_shrink_attempted);
 		trace_scoutfs_lock_shrink(sb, lock);
@@ -1459,10 +1468,8 @@ restart:
 	if (added)
 		queue_work(linfo->workq, &linfo->shrink_work);
 
-out:
-	ret = min_t(unsigned long, linfo->lru_nr, INT_MAX);
-	trace_scoutfs_lock_shrink_exit(sb, sc->nr_to_scan, ret);
-	return ret;
+	trace_scoutfs_lock_shrink_exit(sb, sc->nr_to_scan, freed);
+	return freed;
 }
 
 void scoutfs_free_unused_locks(struct super_block *sb)
@@ -1473,7 +1480,7 @@ void scoutfs_free_unused_locks(struct super_block *sb)
 		.nr_to_scan = INT_MAX,
 	};
 
-	linfo->shrinker.shrink(&linfo->shrinker, &sc);
+	lock_scan_objects(KC_SHRINKER_FN(&linfo->shrinker), &sc);
 }
 
 static void lock_tseq_show(struct seq_file *m, struct scoutfs_tseq_entry *ent)
@@ -1580,7 +1587,7 @@ void scoutfs_lock_shutdown(struct super_block *sb)
 	trace_scoutfs_lock_shutdown(sb, linfo);
 
 	/* stop the shrinker from queueing work */
-	unregister_shrinker(&linfo->shrinker);
+	KC_UNREGISTER_SHRINKER(&linfo->shrinker);
 	flush_work(&linfo->shrink_work);
 
 	/* cause current and future lock calls to return errors */
@@ -1699,9 +1706,9 @@ int scoutfs_lock_setup(struct super_block *sb)
 	spin_lock_init(&linfo->lock);
 	linfo->lock_tree = RB_ROOT;
 	linfo->lock_range_tree = RB_ROOT;
-	linfo->shrinker.shrink = scoutfs_lock_shrink;
-	linfo->shrinker.seeks = DEFAULT_SEEKS;
-	register_shrinker(&linfo->shrinker);
+	KC_INIT_SHRINKER_FUNCS(&linfo->shrinker, lock_count_objects,
+			       lock_scan_objects);
+	KC_REGISTER_SHRINKER(&linfo->shrinker);
 	INIT_LIST_HEAD(&linfo->lru_list);
 	INIT_WORK(&linfo->inv_work, lock_invalidate_worker);
 	INIT_LIST_HEAD(&linfo->inv_list);

kmod/src/net.c

@@ -549,12 +549,16 @@ static int recvmsg_full(struct socket *sock, void *buf, unsigned len)
 
 	while (len) {
 		memset(&msg, 0, sizeof(msg));
-		msg.msg_iov = (struct iovec *)&kv;
-		msg.msg_iovlen = 1;
 		msg.msg_flags = MSG_NOSIGNAL;
 		kv.iov_base = buf;
 		kv.iov_len = len;
 
+#ifndef KC_MSGHDR_STRUCT_IOV_ITER
+		msg.msg_iov = (struct iovec *)&kv;
+		msg.msg_iovlen = 1;
+#else
+		iov_iter_init(&msg.msg_iter, READ, (struct iovec *)&kv, len, 1);
+#endif
 		ret = kernel_recvmsg(sock, &msg, &kv, 1, len, msg.msg_flags);
 		if (ret <= 0)
 			return -ECONNABORTED;
@@ -707,12 +711,16 @@ static int sendmsg_full(struct socket *sock, void *buf, unsigned len)
 
 	while (len) {
 		memset(&msg, 0, sizeof(msg));
-		msg.msg_iov = (struct iovec *)&kv;
-		msg.msg_iovlen = 1;
 		msg.msg_flags = MSG_NOSIGNAL;
 		kv.iov_base = buf;
 		kv.iov_len = len;
 
+#ifndef KC_MSGHDR_STRUCT_IOV_ITER
+		msg.msg_iov = (struct iovec *)&kv;
+		msg.msg_iovlen = 1;
+#else
+		iov_iter_init(&msg.msg_iter, WRITE, (struct iovec *)&kv, len, 1);
+#endif
 		ret = kernel_sendmsg(sock, &msg, &kv, 1, len);
 		if (ret <= 0)
 			return -ECONNABORTED;
@@ -897,7 +905,6 @@ static int sock_opts_and_names(struct scoutfs_net_connection *conn,
 			       struct socket *sock)
 {
 	struct timeval tv;
-	int addrlen;
 	int optval;
 	int ret;
 
@@ -947,23 +954,18 @@ static int sock_opts_and_names(struct scoutfs_net_connection *conn,
 	if (ret)
 		goto out;
 
-	addrlen = sizeof(struct sockaddr_in);
-	ret = kernel_getsockname(sock, (struct sockaddr *)&conn->sockname,
-				 &addrlen);
-	if (ret == 0 && addrlen != sizeof(struct sockaddr_in))
-		ret = -EAFNOSUPPORT;
-	if (ret)
+	ret = kc_kernel_getsockname(sock, (struct sockaddr *)&conn->sockname);
+	if (ret < 0)
 		goto out;
 
-	addrlen = sizeof(struct sockaddr_in);
-	ret = kernel_getpeername(sock, (struct sockaddr *)&conn->peername,
-				 &addrlen);
-	if (ret == 0 && addrlen != sizeof(struct sockaddr_in))
-		ret = -EAFNOSUPPORT;
-	if (ret)
+	ret = kc_kernel_getpeername(sock, (struct sockaddr *)&conn->peername);
+	if (ret < 0)
 		goto out;
 
+	ret = 0;
+
 	conn->last_peername = conn->peername;
+
 out:
 	return ret;
 }
@@ -1052,7 +1054,7 @@ static void scoutfs_net_connect_worker(struct work_struct *work)
 
 	trace_scoutfs_net_connect_work_enter(sb, 0, 0);
 
-	ret = sock_create_kern(AF_INET, SOCK_STREAM, IPPROTO_TCP, &sock);
+	ret = kc_sock_create_kern(AF_INET, SOCK_STREAM, IPPROTO_TCP, &sock);
 	if (ret)
 		goto out;
 
@@ -1453,7 +1455,7 @@ int scoutfs_net_bind(struct super_block *sb,
 	if (WARN_ON_ONCE(conn->sock))
 		return -EINVAL;
 
-	ret = sock_create_kern(AF_INET, SOCK_STREAM, IPPROTO_TCP, &sock);
+	ret = kc_sock_create_kern(AF_INET, SOCK_STREAM, IPPROTO_TCP, &sock);
 	if (ret)
 		goto out;
 
@@ -1471,20 +1473,18 @@ int scoutfs_net_bind(struct super_block *sb,
 		goto out;
 
 	ret = kernel_listen(sock, 255);
-	if (ret)
+	if (ret < 0)
 		goto out;
 
-	addrlen = sizeof(struct sockaddr_in);
-	ret = kernel_getsockname(sock, (struct sockaddr *)&conn->sockname,
-				 &addrlen);
-	if (ret == 0 && addrlen != sizeof(struct sockaddr_in))
-		ret = -EAFNOSUPPORT;
-	if (ret)
+	ret = kc_kernel_getsockname(sock, (struct sockaddr *)&conn->sockname);
+	if (ret < 0)
 		goto out;
 
+	ret = 0;
+
 	conn->sock = sock;
 	*sin = conn->sockname;
-	ret = 0;
+
 out:
 	if (ret < 0 && sock)
 		sock_release(sock);

kmod/src/options.c

@@ -36,6 +36,7 @@ enum {
 	Opt_metadev_path,
 	Opt_noacl,
 	Opt_orphan_scan_delay_ms,
+	Opt_quorum_heartbeat_timeout_ms,
 	Opt_quorum_slot_nr,
 	Opt_err,
 };
@@ -47,6 +48,7 @@ static const match_table_t tokens = {
 	{Opt_metadev_path, "metadev_path=%s"},
 	{Opt_noacl, "noacl"},
 	{Opt_orphan_scan_delay_ms, "orphan_scan_delay_ms=%s"},
+	{Opt_quorum_heartbeat_timeout_ms, "quorum_heartbeat_timeout_ms=%s"},
 	{Opt_quorum_slot_nr, "quorum_slot_nr=%s"},
 	{Opt_err, NULL}
 };
@@ -124,8 +126,24 @@ static void init_default_options(struct scoutfs_mount_options *opts)
 
 	opts->data_prealloc_blocks = SCOUTFS_DATA_PREALLOC_DEFAULT_BLOCKS;
 	opts->data_prealloc_contig_only = 1;
-	opts->quorum_slot_nr = -1;
 	opts->orphan_scan_delay_ms = -1;
+	opts->quorum_heartbeat_timeout_ms = SCOUTFS_QUORUM_DEF_HB_TIMEO_MS;
+	opts->quorum_slot_nr = -1;
+}
+
+static int verify_quorum_heartbeat_timeout_ms(struct super_block *sb, int ret, u64 val)
+{
+	if (ret < 0) {
+		scoutfs_err(sb, "failed to parse quorum_heartbeat_timeout_ms value");
+		return -EINVAL;
+	}
+	if (val < SCOUTFS_QUORUM_MIN_HB_TIMEO_MS || val > SCOUTFS_QUORUM_MAX_HB_TIMEO_MS) {
+		scoutfs_err(sb, "invalid quorum_heartbeat_timeout_ms value %llu, must be between %lu and %lu",
+			    val, SCOUTFS_QUORUM_MIN_HB_TIMEO_MS, SCOUTFS_QUORUM_MAX_HB_TIMEO_MS);
+		return -EINVAL;
+	}
+
+	return 0;
 }
 
 /*
@@ -151,7 +169,7 @@ static int parse_options(struct super_block *sb, char *options, struct scoutfs_m
 		switch (token) {
 
 		case Opt_acl:
-			sb->s_flags |= MS_POSIXACL;
+			sb->s_flags |= SB_POSIXACL;
 			break;
 
 		case Opt_data_prealloc_blocks:
@@ -185,7 +203,7 @@ static int parse_options(struct super_block *sb, char *options, struct scoutfs_m
 			break;
 
 		case Opt_noacl:
-			sb->s_flags &= ~MS_POSIXACL;
+			sb->s_flags &= ~SB_POSIXACL;
 			break;
 
 		case Opt_orphan_scan_delay_ms:
@@ -206,6 +224,14 @@ static int parse_options(struct super_block *sb, char *options, struct scoutfs_m
 			opts->orphan_scan_delay_ms = nr;
 			break;
 
+		case Opt_quorum_heartbeat_timeout_ms:
+			ret = match_u64(args, &nr64);
+			ret = verify_quorum_heartbeat_timeout_ms(sb, ret, nr64);
+			if (ret < 0)
+				return ret;
+			opts->quorum_heartbeat_timeout_ms = nr64;
+			break;
+
 		case Opt_quorum_slot_nr:
 			if (opts->quorum_slot_nr != -1) {
 				scoutfs_err(sb, "multiple quorum_slot_nr options provided, only provide one.");
@@ -301,7 +327,7 @@ int scoutfs_options_show(struct seq_file *seq, struct dentry *root)
 {
 	struct super_block *sb = root->d_sb;
 	struct scoutfs_mount_options opts;
-	const bool is_acl = !!(sb->s_flags & MS_POSIXACL);
+	const bool is_acl = !!(sb->s_flags & SB_POSIXACL);
 
 	scoutfs_options_read(sb, &opts);
 
@@ -448,6 +474,43 @@ static ssize_t orphan_scan_delay_ms_store(struct kobject *kobj, struct kobj_attr
 }
 SCOUTFS_ATTR_RW(orphan_scan_delay_ms);
 
+static ssize_t quorum_heartbeat_timeout_ms_show(struct kobject *kobj, struct kobj_attribute *attr,
+						char *buf)
+{
+	struct super_block *sb = SCOUTFS_SYSFS_ATTRS_SB(kobj);
+	struct scoutfs_mount_options opts;
+
+	scoutfs_options_read(sb, &opts);
+
+	return snprintf(buf, PAGE_SIZE, "%llu", opts.quorum_heartbeat_timeout_ms);
+}
+static ssize_t quorum_heartbeat_timeout_ms_store(struct kobject *kobj, struct kobj_attribute *attr,
+						 const char *buf, size_t count)
+{
+	struct super_block *sb = SCOUTFS_SYSFS_ATTRS_SB(kobj);
+	DECLARE_OPTIONS_INFO(sb, optinf);
+	char nullterm[30]; /* more than enough for octal -U64_MAX */
+	u64 val;
+	int len;
+	int ret;
+
+	len = min(count, sizeof(nullterm) - 1);
+	memcpy(nullterm, buf, len);
+	nullterm[len] = '\0';
+
+	ret = kstrtoll(nullterm, 0, &val);
+	ret = verify_quorum_heartbeat_timeout_ms(sb, ret, val);
+	if (ret == 0) {
+		write_seqlock(&optinf->seqlock);
+		optinf->opts.quorum_heartbeat_timeout_ms = val;
+		write_sequnlock(&optinf->seqlock);
+		ret = count;
+	}
+
+	return ret;
+}
+SCOUTFS_ATTR_RW(quorum_heartbeat_timeout_ms);
+
 static ssize_t quorum_slot_nr_show(struct kobject *kobj, struct kobj_attribute *attr, char *buf)
 {
 	struct super_block *sb = SCOUTFS_SYSFS_ATTRS_SB(kobj);
@@ -464,6 +527,7 @@ static struct attribute *options_attrs[] = {
 	SCOUTFS_ATTR_PTR(data_prealloc_contig_only),
 	SCOUTFS_ATTR_PTR(metadev_path),
 	SCOUTFS_ATTR_PTR(orphan_scan_delay_ms),
+	SCOUTFS_ATTR_PTR(quorum_heartbeat_timeout_ms),
 	SCOUTFS_ATTR_PTR(quorum_slot_nr),
 	NULL,
 };

kmod/src/options.h

@@ -11,7 +11,7 @@ struct scoutfs_mount_options {
 	char *metadev_path;
 	unsigned int orphan_scan_delay_ms;
 	int quorum_slot_nr;
-
+	u64 quorum_heartbeat_timeout_ms;
 };
 
 void scoutfs_options_read(struct super_block *sb, struct scoutfs_mount_options *opts);

kmod/src/quorum.c

@@ -100,6 +100,11 @@ struct last_msg {
 	ktime_t ts;
 };
 
+struct count_recent {
+	u64 count;
+	ktime_t recent;
+};
+
 enum quorum_role { FOLLOWER, CANDIDATE, LEADER };
 
 struct quorum_status {
@@ -112,9 +117,12 @@ struct quorum_status {
 	ktime_t timeout;
 };
 
+#define HB_DELAY_NR		(SCOUTFS_QUORUM_MAX_HB_TIMEO_MS / MSEC_PER_SEC)
+
 struct quorum_info {
 	struct super_block *sb;
 	struct scoutfs_quorum_config qconf;
+	struct workqueue_struct *workq;
 	struct work_struct work;
 	struct socket *sock;
 	bool shutdown;
@@ -126,6 +134,8 @@ struct quorum_info {
 	struct quorum_status show_status;
 	struct last_msg last_send[SCOUTFS_QUORUM_MAX_SLOTS];
 	struct last_msg last_recv[SCOUTFS_QUORUM_MAX_SLOTS];
+	struct count_recent *hb_delay;
+	unsigned long max_hb_delay;
 
 	struct scoutfs_sysfs_attrs ssa;
 };
@@ -160,9 +170,9 @@ static ktime_t heartbeat_interval(void)
 	return ktime_add_ms(ktime_get(), SCOUTFS_QUORUM_HB_IVAL_MS);
 }
 
-static ktime_t heartbeat_timeout(void)
+static ktime_t heartbeat_timeout(struct scoutfs_mount_options *opts)
 {
-	return ktime_add_ms(ktime_get(), SCOUTFS_QUORUM_HB_TIMEO_MS);
+	return ktime_add_ms(ktime_get(), opts->quorum_heartbeat_timeout_ms);
 }
 
 static int create_socket(struct super_block *sb)
@@ -173,13 +183,14 @@ static int create_socket(struct super_block *sb)
 	int addrlen;
 	int ret;
 
-	ret = sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock);
+	ret = kc_sock_create_kern(PF_INET, SOCK_DGRAM, IPPROTO_UDP, &sock);
 	if (ret) {
 		scoutfs_err(sb, "quorum couldn't create udp socket: %d", ret);
 		goto out;
 	}
 
-	sock->sk->sk_allocation = GFP_NOFS;
+	/* rather fail and retry than block waiting for free */
+	sock->sk->sk_allocation = GFP_ATOMIC;
 
 	quorum_slot_sin(&qinf->qconf, qinf->our_quorum_slot_nr, &sin);
 
@@ -208,12 +219,16 @@ static __le32 quorum_message_crc(struct scoutfs_quorum_message *qmes)
 	return cpu_to_le32(crc32c(~0, qmes, len));
 }
 
-static void send_msg_members(struct super_block *sb, int type, u64 term,
-			     int only)
+/*
+ * Returns the number of failures from sendmsg.
+ */
+static int send_msg_members(struct super_block *sb, int type, u64 term, int only)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	DECLARE_QUORUM_INFO(sb, qinf);
+	int failed = 0;
 	ktime_t now;
+	int ret;
 	int i;
 
 	struct scoutfs_quorum_message qmes = {
@@ -228,8 +243,10 @@ static void send_msg_members(struct super_block *sb, int type, u64 term,
 	};
 	struct sockaddr_in sin;
 	struct msghdr mh = {
+#ifndef KC_MSGHDR_STRUCT_IOV_ITER
 		.msg_iov = (struct iovec *)&kv,
 		.msg_iovlen = 1,
+#endif
 		.msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL,
 		.msg_name = &sin,
 		.msg_namelen = sizeof(sin),
@@ -239,15 +256,24 @@ static void send_msg_members(struct super_block *sb, int type, u64 term,
 
 	qmes.crc = quorum_message_crc(&qmes);
 
-
 	for (i = 0; i < SCOUTFS_QUORUM_MAX_SLOTS; i++) {
 		if (!quorum_slot_present(&qinf->qconf, i) ||
 		    (only >= 0 && i != only) || i == qinf->our_quorum_slot_nr)
 			continue;
 
+		if (scoutfs_forcing_unmount(sb)) {
+			failed = 0;
+			break;
+		}
+
 		scoutfs_quorum_slot_sin(&qinf->qconf, i, &sin);
 		now = ktime_get();
-		kernel_sendmsg(qinf->sock, &mh, &kv, 1, kv.iov_len);
+#ifdef KC_MSGHDR_STRUCT_IOV_ITER
+		iov_iter_init(&mh.msg_iter, WRITE, (struct iovec *)&kv, sizeof(qmes), 1);
+#endif
+		ret = kernel_sendmsg(qinf->sock, &mh, &kv, 1, kv.iov_len);
+		if (ret != kv.iov_len)
+			failed++;
 
 		spin_lock(&qinf->show_lock);
 		qinf->last_send[i].msg.term = term;
@@ -258,6 +284,8 @@ static void send_msg_members(struct super_block *sb, int type, u64 term,
 		if (i == only)
 			break;
 	}
+
+	return failed;
 }
 
 #define send_msg_to(sb, type, term, nr)  send_msg_members(sb, type, term, nr)
@@ -285,8 +313,10 @@ static int recv_msg(struct super_block *sb, struct quorum_host_msg *msg,
 		.iov_len = sizeof(struct scoutfs_quorum_message),
 	};
 	struct msghdr mh = {
+#ifndef KC_MSGHDR_STRUCT_IOV_ITER
 		.msg_iov = (struct iovec *)&kv,
 		.msg_iovlen = 1,
+#endif
 		.msg_flags = MSG_NOSIGNAL,
 	};
 
@@ -308,10 +338,16 @@ static int recv_msg(struct super_block *sb, struct quorum_host_msg *msg,
 			return ret;
 	}
 
+#ifdef KC_MSGHDR_STRUCT_IOV_ITER
+	iov_iter_init(&mh.msg_iter, READ, (struct iovec *)&kv, sizeof(struct scoutfs_quorum_message), 1);
+#endif
 	ret = kernel_recvmsg(qinf->sock, &mh, &kv, 1, kv.iov_len, mh.msg_flags);
 	if (ret < 0)
 		return ret;
 
+	if (scoutfs_forcing_unmount(sb))
+		return 0;
+
 	now = ktime_get();
 
 	if (ret != sizeof(qmes) ||
@@ -599,6 +635,71 @@ out:
 	return ret;
 }
 
+static void clear_hb_delay(struct quorum_info *qinf)
+{
+	int i;
+
+	spin_lock(&qinf->show_lock);
+	qinf->max_hb_delay = 0;
+	for (i = 0; i < HB_DELAY_NR; i++) {
+		qinf->hb_delay[i].recent = ns_to_ktime(0);
+		qinf->hb_delay[i].count = 0;
+	}
+	spin_unlock(&qinf->show_lock);
+}
+
+struct hb_recording {
+	ktime_t prev;
+	int count;
+};
+
+/*
+ * Record long heartbeat delays.  We only record the delay between back
+ * to back send attempts in the leader or back to back recv messages in
+ * the followers.  The worker caller sets record_hb when their iteration
+ * sent or received a heartbeat.  An iteration that does anything else
+ * resets the tracking.
+ */
+static void record_hb_delay(struct super_block *sb, struct quorum_info *qinf,
+			    struct hb_recording *hbr, bool record_hb, int role)
+{
+	bool log = false;
+	ktime_t now;
+	s64 s;
+
+	if (!record_hb) {
+		hbr->count = 0;
+		return;
+	}
+
+	now = ktime_get();
+
+	if (hbr->count < 2 && ++hbr->count < 2) {
+		hbr->prev = now;
+		return;
+	}
+
+	s = ktime_ms_delta(now, hbr->prev) / MSEC_PER_SEC;
+	hbr->prev = now;
+
+	if (s <= 0 || s >= HB_DELAY_NR)
+		return;
+
+	spin_lock(&qinf->show_lock);
+	if (qinf->max_hb_delay < s) {
+		qinf->max_hb_delay = s;
+		if (s >= 3)
+			log = true;
+	}
+	qinf->hb_delay[s].recent = now;
+	qinf->hb_delay[s].count++;
+	spin_unlock(&qinf->show_lock);
+
+	if (log)
+		scoutfs_info(sb, "longest quorum heartbeat %s delay of %lld sec",
+			     role == LEADER ? "send" : "recv", s);
+}
+
 /*
  * The main quorum task maintains its private status.  It seemed cleaner
  * to occasionally copy the status for showing in sysfs/debugfs files
@@ -623,16 +724,23 @@ static void update_show_status(struct quorum_info *qinf, struct quorum_status *q
 static void scoutfs_quorum_worker(struct work_struct *work)
 {
 	struct quorum_info *qinf = container_of(work, struct quorum_info, work);
+	struct scoutfs_mount_options opts;
 	struct super_block *sb = qinf->sb;
 	struct sockaddr_in unused;
 	struct quorum_host_msg msg;
 	struct quorum_status qst = {0,};
+	struct hb_recording hbr;
+	bool record_hb;
 	int ret;
 	int err;
 
+	memset(&hbr, 0, sizeof(struct hb_recording));
+
 	/* recording votes from slots as native single word bitmap */
 	BUILD_BUG_ON(SCOUTFS_QUORUM_MAX_SLOTS > BITS_PER_LONG);
 
+	scoutfs_options_read(sb, &opts);
+
 	/* start out as a follower */
 	qst.role = FOLLOWER;
 	qst.vote_for = -1;
@@ -642,7 +750,7 @@ static void scoutfs_quorum_worker(struct work_struct *work)
 
 	/* see if there's a server to chose heartbeat or election timeout */
 	if (scoutfs_quorum_server_sin(sb, &unused) == 0)
-		qst.timeout = heartbeat_timeout();
+		qst.timeout = heartbeat_timeout(&opts);
 	else
 		qst.timeout = election_timeout();
 
@@ -666,14 +774,16 @@ static void scoutfs_quorum_worker(struct work_struct *work)
 			ret = 0;
 		}
 
+		scoutfs_options_read(sb, &opts);
+		record_hb = false;
+
 		/* ignore messages from older terms */
 		if (msg.type != SCOUTFS_QUORUM_MSG_INVALID &&
 		    msg.term < qst.term)
 			msg.type = SCOUTFS_QUORUM_MSG_INVALID;
 
 		trace_scoutfs_quorum_loop(sb, qst.role, qst.term, qst.vote_for,
-					  qst.vote_bits,
-					  ktime_to_timespec64(qst.timeout));
+					  qst.vote_bits, ktime_to_ns(qst.timeout));
 
 		/* receiving greater terms resets term, becomes follower */
 		if (msg.type != SCOUTFS_QUORUM_MSG_INVALID &&
@@ -681,6 +791,7 @@ static void scoutfs_quorum_worker(struct work_struct *work)
 			if (qst.role == LEADER) {
 				scoutfs_warn(sb, "saw msg type %u from %u for term %llu while leader in term %llu, shutting down server.",
 					     msg.type, msg.from, msg.term, qst.term);
+				clear_hb_delay(qinf);
 			}
 			qst.role = FOLLOWER;
 			qst.term = msg.term;
@@ -689,7 +800,7 @@ static void scoutfs_quorum_worker(struct work_struct *work)
 			scoutfs_inc_counter(sb, quorum_term_follower);
 
 			if (msg.type == SCOUTFS_QUORUM_MSG_HEARTBEAT)
-				qst.timeout = heartbeat_timeout();
+				qst.timeout = heartbeat_timeout(&opts);
 			else
 				qst.timeout = election_timeout();
 
@@ -699,6 +810,21 @@ static void scoutfs_quorum_worker(struct work_struct *work)
 				goto out;
 		}
 
+		/* receiving heartbeats extends timeout, delaying elections */
+		if (msg.type == SCOUTFS_QUORUM_MSG_HEARTBEAT) {
+			qst.timeout = heartbeat_timeout(&opts);
+			scoutfs_inc_counter(sb, quorum_recv_heartbeat);
+			record_hb = true;
+		}
+
+		/* receiving a resignation from server starts election */
+		if (msg.type == SCOUTFS_QUORUM_MSG_RESIGNATION &&
+		    qst.role == FOLLOWER &&
+		    msg.term == qst.term) {
+			qst.timeout = election_timeout();
+			scoutfs_inc_counter(sb, quorum_recv_resignation);
+		}
+
 		/* followers and candidates start new election on timeout */
 		if (qst.role != LEADER &&
 		    ktime_after(ktime_get(), qst.timeout)) {
@@ -751,6 +877,7 @@ static void scoutfs_quorum_worker(struct work_struct *work)
 			qst.timeout = heartbeat_interval();
 
 			update_show_status(qinf, &qst);
+			clear_hb_delay(qinf);
 
 			/* record that we've been elected before starting up server */
 			ret = update_quorum_block(sb, SCOUTFS_QUORUM_EVENT_ELECT, qst.term, true);
@@ -805,6 +932,7 @@ static void scoutfs_quorum_worker(struct work_struct *work)
 				send_msg_others(sb, SCOUTFS_QUORUM_MSG_RESIGNATION,
 						qst.server_start_term);
 				scoutfs_inc_counter(sb, quorum_send_resignation);
+				clear_hb_delay(qinf);
 			}
 
 			ret = update_quorum_block(sb, SCOUTFS_QUORUM_EVENT_STOP,
@@ -818,24 +946,16 @@ static void scoutfs_quorum_worker(struct work_struct *work)
 		/* leaders regularly send heartbeats to delay elections */
 		if (qst.role == LEADER &&
 		    ktime_after(ktime_get(), qst.timeout)) {
-			send_msg_others(sb, SCOUTFS_QUORUM_MSG_HEARTBEAT,
-					qst.term);
+			ret = send_msg_others(sb, SCOUTFS_QUORUM_MSG_HEARTBEAT, qst.term);
+			if (ret > 0) {
+				scoutfs_add_counter(sb, quorum_send_heartbeat_dropped, ret);
+				ret = 0;
+			}
+
 			qst.timeout = heartbeat_interval();
 			scoutfs_inc_counter(sb, quorum_send_heartbeat);
-		}
+			record_hb = true;
 
-		/* receiving heartbeats extends timeout, delaying elections */
-		if (msg.type == SCOUTFS_QUORUM_MSG_HEARTBEAT) {
-			qst.timeout = heartbeat_timeout();
-			scoutfs_inc_counter(sb, quorum_recv_heartbeat);
-		}
-
-		/* receiving a resignation from server starts election */
-		if (msg.type == SCOUTFS_QUORUM_MSG_RESIGNATION &&
-		    qst.role == FOLLOWER &&
-		    msg.term == qst.term) {
-			qst.timeout = election_timeout();
-			scoutfs_inc_counter(sb, quorum_recv_resignation);
 		}
 
 		/* followers vote once per term */
@@ -847,6 +967,8 @@ static void scoutfs_quorum_worker(struct work_struct *work)
 				    msg.from);
 			scoutfs_inc_counter(sb, quorum_send_vote);
 		}
+
+		record_hb_delay(sb, qinf, &hbr, record_hb, qst.role);
 	}
 
 	update_show_status(qinf, &qst);
@@ -983,9 +1105,11 @@ static ssize_t status_show(struct kobject *kobj, struct kobj_attribute *attr,
 {
 	DECLARE_QUORUM_INFO_KOBJ(kobj, qinf);
 	struct quorum_status qst;
+	struct count_recent cr;
 	struct last_msg last;
 	struct timespec64 ts;
 	const ktime_t now = ktime_get();
+	unsigned long ul;
 	size_t size;
 	int ret;
 	int i;
@@ -1043,6 +1167,26 @@ static ssize_t status_show(struct kobject *kobj, struct kobj_attribute *attr,
 			     (s64)ts.tv_sec, (int)ts.tv_nsec);
 	}
 
+	spin_lock(&qinf->show_lock);
+	ul = qinf->max_hb_delay;
+	spin_unlock(&qinf->show_lock);
+	if (ul)
+		snprintf_ret(buf, size, &ret, "HB Delay(s)      Count  Secs Since\n");
+
+	for (i = 1; i <= ul && i < HB_DELAY_NR; i++) {
+		spin_lock(&qinf->show_lock);
+		cr = qinf->hb_delay[i];
+		spin_unlock(&qinf->show_lock);
+
+		if (cr.count == 0)
+			continue;
+
+		ts = ktime_to_timespec64(ktime_sub(now, cr.recent));
+		snprintf_ret(buf, size, &ret,
+			     "%11u  %9llu  %lld.%09u\n",
+			     i, cr.count, (s64)ts.tv_sec, (int)ts.tv_nsec);
+	}
+
 	return ret;
 }
 SCOUTFS_ATTR_RO(status);
@@ -1180,7 +1324,12 @@ int scoutfs_quorum_setup(struct super_block *sb)
 
 	qinf = kzalloc(sizeof(struct quorum_info), GFP_KERNEL);
 	super = kmalloc(sizeof(struct scoutfs_super_block), GFP_KERNEL);
-	if (!qinf || !super) {
+	if (qinf)
+		qinf->hb_delay = __vmalloc(HB_DELAY_NR * sizeof(struct count_recent),
+					   GFP_KERNEL | __GFP_ZERO, PAGE_KERNEL);
+	if (!qinf || !super || !qinf->hb_delay) {
+		if (qinf)
+			vfree(qinf->hb_delay);
 		kfree(qinf);
 		ret = -ENOMEM;
 		goto out;
@@ -1195,6 +1344,15 @@ int scoutfs_quorum_setup(struct super_block *sb)
 	sbi->quorum_info = qinf;
 	qinf->sb = sb;
 
+	/* a high priority single threaded context without mem reclaim */
+	qinf->workq = alloc_workqueue("scoutfs_quorum_work",
+				       WQ_NON_REENTRANT | WQ_UNBOUND |
+				       WQ_HIGHPRI, 1);
+	if (!qinf->workq) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
 	ret = scoutfs_read_super(sb, super);
 	if (ret < 0)
 		goto out;
@@ -1213,7 +1371,7 @@ int scoutfs_quorum_setup(struct super_block *sb)
 	if (ret < 0)
 		goto out;
 
-	schedule_work(&qinf->work);
+	queue_work(qinf->workq, &qinf->work);
 
 out:
 	if (ret)
@@ -1243,10 +1401,14 @@ void scoutfs_quorum_destroy(struct super_block *sb)
 		qinf->shutdown = true;
 		flush_work(&qinf->work);
 
+		if (qinf->workq)
+			destroy_workqueue(qinf->workq);
+
 		scoutfs_sysfs_destroy_attrs(sb, &qinf->ssa);
 		if (qinf->sock)
 			sock_release(qinf->sock);
 
+		vfree(qinf->hb_delay);
 		kfree(qinf);
 		sbi->quorum_info = NULL;
 	}

kmod/src/scoutfs_trace.h

@@ -817,22 +817,17 @@ TRACE_EVENT(scoutfs_advance_dirty_super,
 	TP_printk(SCSBF" super seq now %llu", SCSB_TRACE_ARGS, __entry->seq)
 );
 
-TRACE_EVENT(scoutfs_dir_add_next_linkref,
+TRACE_EVENT(scoutfs_dir_add_next_linkref_found,
 	TP_PROTO(struct super_block *sb, __u64 ino, __u64 dir_ino,
-		 __u64 dir_pos, int ret, __u64 found_dir_ino,
-		 __u64 found_dir_pos, unsigned int name_len),
+		 __u64 dir_pos, unsigned int name_len),
 
-	TP_ARGS(sb, ino, dir_ino, dir_pos, ret, found_dir_pos, found_dir_ino,
-		name_len),
+	TP_ARGS(sb, ino, dir_ino, dir_pos, name_len),
 
 	TP_STRUCT__entry(
 		SCSB_TRACE_FIELDS
 		__field(__u64, ino)
 		__field(__u64, dir_ino)
 		__field(__u64, dir_pos)
-		__field(int, ret)
-		__field(__u64, found_dir_ino)
-		__field(__u64, found_dir_pos)
 		__field(unsigned int, name_len)
 	),
 
@@ -841,16 +836,43 @@ TRACE_EVENT(scoutfs_dir_add_next_linkref,
 		__entry->ino = ino;
 		__entry->dir_ino = dir_ino;
 		__entry->dir_pos = dir_pos;
-		__entry->ret = ret;
-		__entry->found_dir_ino = dir_ino;
-		__entry->found_dir_pos = dir_pos;
 		__entry->name_len = name_len;
 	),
 
-	TP_printk(SCSBF" ino %llu dir_ino %llu dir_pos %llu ret %d found_dir_ino %llu found_dir_pos %llu name_len %u",
-		  SCSB_TRACE_ARGS, __entry->ino, __entry->dir_pos,
-		  __entry->dir_ino, __entry->ret, __entry->found_dir_pos,
-		  __entry->found_dir_ino, __entry->name_len)
+	TP_printk(SCSBF" ino %llu dir_ino %llu dir_pos %llu name_len %u",
+		  SCSB_TRACE_ARGS, __entry->ino, __entry->dir_ino,
+		  __entry->dir_pos, __entry->name_len)
+);
+
+TRACE_EVENT(scoutfs_dir_add_next_linkrefs,
+	TP_PROTO(struct super_block *sb, __u64 ino, __u64 dir_ino,
+		 __u64 dir_pos, int count, int nr, int ret),
+
+	TP_ARGS(sb, ino, dir_ino, dir_pos, count, nr, ret),
+
+	TP_STRUCT__entry(
+		SCSB_TRACE_FIELDS
+		__field(__u64, ino)
+		__field(__u64, dir_ino)
+		__field(__u64, dir_pos)
+		__field(int, count)
+		__field(int, nr)
+		__field(int, ret)
+	),
+
+	TP_fast_assign(
+		SCSB_TRACE_ASSIGN(sb);
+		__entry->ino = ino;
+		__entry->dir_ino = dir_ino;
+		__entry->dir_pos = dir_pos;
+		__entry->count = count;
+		__entry->nr = nr;
+		__entry->ret = ret;
+	),
+
+	TP_printk(SCSBF" ino %llu dir_ino %llu dir_pos %llu count %d nr %d ret %d",
+		  SCSB_TRACE_ARGS, __entry->ino, __entry->dir_ino,
+		  __entry->dir_pos, __entry->count, __entry->nr, __entry->ret)
 );
 
 TRACE_EVENT(scoutfs_write_begin,
@@ -1998,9 +2020,9 @@ DEFINE_EVENT(scoutfs_quorum_message_class, scoutfs_quorum_recv_message,
 
 TRACE_EVENT(scoutfs_quorum_loop,
 	TP_PROTO(struct super_block *sb, int role, u64 term, int vote_for,
-		 unsigned long vote_bits, struct timespec64 timeout),
+		 unsigned long vote_bits, unsigned long long nsecs),
 
-	TP_ARGS(sb, role, term, vote_for, vote_bits, timeout),
+	TP_ARGS(sb, role, term, vote_for, vote_bits, nsecs),
 
 	TP_STRUCT__entry(
 		SCSB_TRACE_FIELDS
@@ -2009,8 +2031,7 @@ TRACE_EVENT(scoutfs_quorum_loop,
 		__field(int, vote_for)
 		__field(unsigned long, vote_bits)
 		__field(unsigned long, vote_count)
-		__field(unsigned long long, timeout_sec)
-		__field(int, timeout_nsec)
+		__field(unsigned long long, nsecs)
 	),
 
 	TP_fast_assign(
@@ -2020,14 +2041,13 @@ TRACE_EVENT(scoutfs_quorum_loop,
 		__entry->vote_for = vote_for;
 		__entry->vote_bits = vote_bits;
 		__entry->vote_count = hweight_long(vote_bits);
-		__entry->timeout_sec = timeout.tv_sec;
-		__entry->timeout_nsec = timeout.tv_nsec;
+		__entry->nsecs = nsecs;
 	),
 
-	TP_printk(SCSBF" term %llu role %d vote_for %d vote_bits 0x%lx vote_count %lu timeout %llu.%u",
+	TP_printk(SCSBF" term %llu role %d vote_for %d vote_bits 0x%lx vote_count %lu timeout %llu",
 		  SCSB_TRACE_ARGS, __entry->term, __entry->role,
 		  __entry->vote_for, __entry->vote_bits, __entry->vote_count,
-		  __entry->timeout_sec, __entry->timeout_nsec)
+		  __entry->nsecs)
 );
 
 TRACE_EVENT(scoutfs_trans_seq_last,

kmod/src/server.c

@@ -2058,6 +2058,13 @@ out:
  * reset the next range key if there's still work to do.  If the
  * operation is complete then we tear down the input log_trees items and
  * delete the status.
+ *
+ * Processing all the completions can take more than one transaction.
+ * We return -EINPROGRESS if we have to commit a transaction and the
+ * caller will apply the commit and immediate call back in so we can
+ * perform another commit.  We need to be very careful to leave the
+ * status in a state where requests won't be issued at the wrong time
+ * (by forcing nr_completions to a batch while we delete them).
  */
 static int splice_log_merge_completions(struct super_block *sb,
 					struct scoutfs_log_merge_status *stat,
@@ -2070,15 +2077,29 @@ static int splice_log_merge_completions(struct super_block *sb,
 	struct scoutfs_log_merge_range rng;
 	struct scoutfs_log_trees lt = {{{0,}}};
 	SCOUTFS_BTREE_ITEM_REF(iref);
+	bool upd_stat = true;
+	int einprogress = 0;
 	struct scoutfs_key key;
 	char *err_str = NULL;
+	u32 alloc_low;
+	u32 tmp;
 	u64 seq;
 	int ret;
+	int err;
 
 	/* musn't rebalance fs tree parents while reqs rely on their key bounds */
 	if (WARN_ON_ONCE(le64_to_cpu(stat->nr_requests) > 0))
 		return -EIO;
 
+	/*
+	 * Be overly conservative about how low the allocator can get
+	 * before we commit.  This gives us a lot of work to do in a
+	 * commit while also allowing a pretty big smallest allocator to
+	 * work with the theoretically unbounded alloc list splicing.
+	 */
+	scoutfs_alloc_meta_remaining(&server->alloc, &alloc_low, &tmp);
+	alloc_low = min(alloc_low, tmp) / 4;
+
 	/*
 	 * Splice in all the completed subtrees at the initial parent
 	 * blocks in the main fs_tree before rebalancing any of them.
@@ -2100,6 +2121,22 @@ static int splice_log_merge_completions(struct super_block *sb,
 
 		seq = le64_to_cpu(comp.seq);
 
+		/*
+		 * Use having cleared the lists as an indication that
+		 * we've already set the parents and don't need to dirty
+		 * the btree blocks to do it all over again.  This is
+		 * safe because there is always an fs block that the
+		 * merge dirties and frees into the meta_freed list.
+		 */
+		if (comp.meta_avail.ref.blkno == 0 && comp.meta_freed.ref.blkno == 0)
+			continue;
+
+		if (scoutfs_alloc_meta_low(sb, &server->alloc, alloc_low)) {
+			einprogress = -EINPROGRESS;
+			ret = 0;
+			goto out;
+		}
+
 		ret = scoutfs_btree_set_parent(sb, &server->alloc, &server->wri,
 					       &super->fs_root, &comp.start,
 					       &comp.root);
@@ -2134,6 +2171,14 @@ static int splice_log_merge_completions(struct super_block *sb,
 		}
 	}
 
+	/*
+	 * Once we start rebalancing we force the number of completions
+	 * to a batch so that requests won't be issued.  Once we're done
+	 * we clear the completion count and requests can flow again.
+	 */
+	if (le64_to_cpu(stat->nr_complete) < LOG_MERGE_SPLICE_BATCH)
+		stat->nr_complete = cpu_to_le64(LOG_MERGE_SPLICE_BATCH);
+
 	/*
 	 * Now with all the parent blocks spliced in, rebalance items
 	 * amongst parents that needed to split/join and delete the
@@ -2155,6 +2200,12 @@ static int splice_log_merge_completions(struct super_block *sb,
 
 		seq = le64_to_cpu(comp.seq);
 
+		if (scoutfs_alloc_meta_low(sb, &server->alloc, alloc_low)) {
+			einprogress = -EINPROGRESS;
+			ret = 0;
+			goto out;
+		}
+
 		/* balance when there was a remaining key range */
 		if (le64_to_cpu(comp.flags) & SCOUTFS_LOG_MERGE_COMP_REMAIN) {
 			ret = scoutfs_btree_rebalance(sb, &server->alloc,
@@ -2194,18 +2245,11 @@ static int splice_log_merge_completions(struct super_block *sb,
 		}
 	}
 
-	/* update the status once all completes are processed */
-	scoutfs_key_set_zeros(&stat->next_range_key);
-	stat->nr_complete = 0;
-
 	/* update counts and done if there's still ranges to process */
 	if (!no_ranges) {
-		init_log_merge_key(&key, SCOUTFS_LOG_MERGE_STATUS_ZONE, 0, 0);
-		ret = scoutfs_btree_update(sb, &server->alloc, &server->wri,
-					   &super->log_merge, &key,
-					   stat, sizeof(*stat));
-		if (ret < 0)
-			err_str = "update status";
+		scoutfs_key_set_zeros(&stat->next_range_key);
+		stat->nr_complete = 0;
+		ret = 0;
 		goto out;
 	}
 
@@ -2241,6 +2285,12 @@ static int splice_log_merge_completions(struct super_block *sb,
 		      (le64_to_cpu(lt.finalize_seq) < le64_to_cpu(stat->seq))))
 			continue;
 
+		if (scoutfs_alloc_meta_low(sb, &server->alloc, alloc_low)) {
+			einprogress = -EINPROGRESS;
+			ret = 0;
+			goto out;
+		}
+
 		fr.root = lt.item_root;
 		scoutfs_key_set_zeros(&fr.key);
 		fr.seq = cpu_to_le64(scoutfs_server_next_seq(sb));
@@ -2274,9 +2324,10 @@ static int splice_log_merge_completions(struct super_block *sb,
 		}
 
 		le64_add_cpu(&super->inode_count, le64_to_cpu(lt.inode_count_delta));
-
 	}
 
+	/* everything's done, remove the merge operation */
+	upd_stat = false;
 	init_log_merge_key(&key, SCOUTFS_LOG_MERGE_STATUS_ZONE, 0, 0);
 	ret = scoutfs_btree_delete(sb, &server->alloc, &server->wri,
 				   &super->log_merge, &key);
@@ -2285,12 +2336,23 @@ static int splice_log_merge_completions(struct super_block *sb,
 	else
 		err_str = "deleting merge status item";
 out:
+	if (upd_stat) {
+		init_log_merge_key(&key, SCOUTFS_LOG_MERGE_STATUS_ZONE, 0, 0);
+		err = scoutfs_btree_update(sb, &server->alloc, &server->wri,
+					   &super->log_merge, &key,
+					   stat, sizeof(struct scoutfs_log_merge_status));
+		if (err && !ret) {
+			err_str = "updating merge status item";
+			ret = err;
+		}
+	}
+
 	if (ret < 0)
 		scoutfs_err(sb, "server error %d splicing log merge completion: %s", ret, err_str);
 
 	BUG_ON(ret); /* inconsistent */
 
-	return ret;
+	return ret ?: einprogress;
 }
 
 /*
@@ -2465,6 +2527,12 @@ static void server_log_merge_free_work(struct work_struct *work)
 }
 
 /*
+ * Clients regularly ask if there is log merge work to do.  We process
+ * completions inline before responding so that we don't create large
+ * delays between completion processing and the next request.  We don't
+ * mind if the client get_log_merge request sees high latency, the
+ * blocked caller has nothing else to do.
+ *
  * This will return ENOENT to the client if there is no work to do.
  */
 static int server_get_log_merge(struct super_block *sb,
@@ -2532,14 +2600,22 @@ restart:
 			goto out;
 		}
 
-		/* maybe splice now that we know if there's ranges */
+		/* splice if we have a batch or ran out of ranges */
 		no_next = ret == -ENOENT;
 		no_ranges = scoutfs_key_is_zeros(&stat.next_range_key) && ret == -ENOENT;
 		if (le64_to_cpu(stat.nr_requests) == 0 &&
 		    (no_next || le64_to_cpu(stat.nr_complete) >= LOG_MERGE_SPLICE_BATCH)) {
 			ret = splice_log_merge_completions(sb, &stat, no_ranges);
-			if (ret < 0)
+			if (ret == -EINPROGRESS) {
+				mutex_unlock(&server->logs_mutex);
+				ret = server_apply_commit(sb, &hold, 0);
+				if (ret < 0)
+					goto respond;
+				server_hold_commit(sb, &hold);
+				mutex_lock(&server->logs_mutex);
+			} else if (ret < 0) {
 				goto out;
+			}
 			/* splicing resets key and adds ranges, could finish status */
 			goto restart;
 		}
@@ -2741,6 +2817,7 @@ out:
 	mutex_unlock(&server->logs_mutex);
 	ret = server_apply_commit(sb, &hold, ret);
 
+respond:
 	return scoutfs_net_response(sb, conn, cmd, id, ret, &req, sizeof(req));
 }
 
@@ -4364,7 +4441,7 @@ void scoutfs_server_stop_wait(struct super_block *sb)
 	DECLARE_SERVER_INFO(sb, server);
 
 	stop_server(server);
-	flush_work_sync(&server->work);
+	flush_work(&server->work);
 }
 
 int scoutfs_server_setup(struct super_block *sb)

kmod/src/srch.c

@@ -1747,7 +1747,7 @@ static int compact_logs(struct super_block *sb,
 				goto out;
 			}
 			page->private = 0;
-			list_add_tail(&page->list, &pages);
+			list_add_tail(&page->lru, &pages);
 			nr_pages++;
 			scoutfs_inc_counter(sb, srch_compact_log_page);
 		}
@@ -1800,7 +1800,7 @@ static int compact_logs(struct super_block *sb,
 
 	/* sort page entries and reset private for _next */
 	i = 0;
-	list_for_each_entry(page, &pages, list) {
+	list_for_each_entry(page, &pages, lru) {
 		args[i++] = page;
 
 		if (atomic_read(&srinf->shutdown)) {
@@ -1821,7 +1821,7 @@ static int compact_logs(struct super_block *sb,
 		goto out;
 
 	/* make sure we finished all the pages */
-	list_for_each_entry(page, &pages, list) {
+	list_for_each_entry(page, &pages, lru) {
 		sre = page_priv_sre(page);
 		if (page->private < SRES_PER_PAGE && sre->ino != 0) {
 			ret = -ENOSPC;
@@ -1834,8 +1834,8 @@ static int compact_logs(struct super_block *sb,
 out:
 	scoutfs_block_put(sb, bl);
 	vfree(args);
-	list_for_each_entry_safe(page, tmp, &pages, list) {
-		list_del(&page->list);
+	list_for_each_entry_safe(page, tmp, &pages, lru) {
+		list_del(&page->lru);
 		__free_page(page);
 	}
 

kmod/src/super.c

@@ -13,6 +13,7 @@
 #include <linux/kernel.h>
 #include <linux/module.h>
 #include <linux/fs.h>
+#include <linux/blkdev.h>
 #include <linux/slab.h>
 #include <linux/pagemap.h>
 #include <linux/magic.h>
@@ -178,7 +179,7 @@ static void scoutfs_put_super(struct super_block *sb)
 	/*
 	 * Wait for invalidation and iput to finish with any lingering
 	 * inode references that escaped the evict_inodes in
-	 * generic_shutdown_super.  MS_ACTIVE is clear so final iput
+	 * generic_shutdown_super.  SB_ACTIVE is clear so final iput
 	 * will always evict.
 	 */
 	scoutfs_lock_flush_invalidate(sb);
@@ -485,7 +486,8 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 	sb->s_d_op = &scoutfs_dentry_ops;
 	sb->s_export_op = &scoutfs_export_ops;
 	sb->s_xattr = scoutfs_xattr_handlers;
-	sb->s_flags |= MS_I_VERSION | MS_POSIXACL;
+	sb->s_flags |= SB_I_VERSION | SB_POSIXACL;
+	sb->s_time_gran = 1;
 
 	/* btree blocks use long lived bh->b_data refs */
 	mapping_set_gfp_mask(sb->s_bdev->bd_inode->i_mapping, GFP_NOFS);
@@ -673,14 +675,14 @@ out:
 		teardown_module();
 	return ret;
 }
-module_init(scoutfs_module_init)
+module_init(scoutfs_module_init);
 
 static void __exit scoutfs_module_exit(void)
 {
 	unregister_filesystem(&scoutfs_fs_type);
 	teardown_module();
 }
-module_exit(scoutfs_module_exit)
+module_exit(scoutfs_module_exit);
 
 MODULE_AUTHOR("Zach Brown <zab@versity.com>");
 MODULE_LICENSE("GPL");

kmod/src/tseq.c

@@ -46,6 +46,23 @@ static struct scoutfs_tseq_entry *tseq_rb_next(struct scoutfs_tseq_entry *ent)
 	return rb_entry(node, struct scoutfs_tseq_entry, node);
 }
 
+#ifdef KC_RB_TREE_AUGMENTED_COMPUTE_MAX
+static bool tseq_compute_total(struct scoutfs_tseq_entry *ent, bool exit)
+{
+	loff_t total = 1 + tseq_node_total(ent->node.rb_left) +
+		       tseq_node_total(ent->node.rb_right);
+
+	if (exit && ent->total == total)
+		return true;
+
+	ent->total = total;
+	return false;
+}
+
+RB_DECLARE_CALLBACKS(static, tseq_rb_callbacks, struct scoutfs_tseq_entry,
+		     node, total, tseq_compute_total);
+#else
+
 static loff_t tseq_compute_total(struct scoutfs_tseq_entry *ent)
 {
 	return 1 + tseq_node_total(ent->node.rb_left) +
@@ -53,7 +70,8 @@ static loff_t tseq_compute_total(struct scoutfs_tseq_entry *ent)
 }
 
 RB_DECLARE_CALLBACKS(static, tseq_rb_callbacks, struct scoutfs_tseq_entry,
-		     node, loff_t, total, tseq_compute_total)
+		     node, loff_t, total, tseq_compute_total);
+#endif
 
 void scoutfs_tseq_tree_init(struct scoutfs_tseq_tree *tree,
 			    scoutfs_tseq_show_t show)

kmod/src/util.h

@@ -17,4 +17,15 @@ static inline void down_write_two(struct rw_semaphore *a,
 	down_write_nested(b, SINGLE_DEPTH_NESTING);
 }
 
+/*
+ * When returning shrinker counts from scan_objects, we should steer
+ * clear of the magic SHRINK_STOP and SHRINK_EMPTY values, which are near
+ * ~0UL values. Hence, we cap count to ~0L, which is arbitarily high
+ * enough to avoid it.
+ */
+static inline unsigned long shrinker_min_t_long(unsigned long count)
+{
+	return min_t(u64, count, LONG_MAX);
+}
+
 #endif

kmod/src/xattr.c

@@ -773,7 +773,7 @@ int scoutfs_xattr_set_locked(struct inode *inode, const char *name, size_t name_
 
 	/* XXX do these want i_mutex or anything? */
 	inode_inc_iversion(inode);
-	inode->i_ctime = CURRENT_TIME;
+	inode->i_ctime = current_time(inode);
 	ret = 0;
 
 out:
@@ -850,6 +850,7 @@ unlock:
 	return ret;
 }
 
+#ifndef KC_XATTR_STRUCT_XATTR_HANDLER
 /*
  * Future kernels have this amazing hack to rewind the name to get the
  * skipped prefix.  We're back in the stone ages without the handler
@@ -857,22 +858,41 @@ unlock:
  * compat hook to either call the kernel's xattr_full_name(handler), or
  * our hack to use the flags as the prefix length.
  */
-static const char *full_name_hack(void *handler, const char *name, int len)
+static const char *full_name_hack(const char *name, int len)
 {
 	return name - len;
 }
+#endif
 
-static int scoutfs_xattr_get_handler(struct dentry *dentry, const char *name,
-				     void *value, size_t size, int handler_flags)
+static int scoutfs_xattr_get_handler
+#ifdef KC_XATTR_STRUCT_XATTR_HANDLER
+		(const struct xattr_handler *handler, struct dentry *dentry,
+		 struct inode *inode, const char *name, void *value,
+		 size_t size)
 {
-	name = full_name_hack(NULL, name, handler_flags);
+	name = xattr_full_name(handler, name);
+#else
+		(struct dentry *dentry, const char *name,
+		 void *value, size_t size, int handler_flags)
+{
+	name = full_name_hack(name, handler_flags);
+#endif
 	return scoutfs_xattr_get(dentry, name, value, size);
 }
 
-static int scoutfs_xattr_set_handler(struct dentry *dentry, const char *name,
-				     const void *value, size_t size, int flags, int handler_flags)
+static int scoutfs_xattr_set_handler
+#ifdef KC_XATTR_STRUCT_XATTR_HANDLER
+		(const struct xattr_handler *handler, struct dentry *dentry,
+		 struct inode *inode, const char *name, const void *value,
+		 size_t size, int flags)
 {
-	name = full_name_hack(NULL, name, handler_flags);
+	name = xattr_full_name(handler, name);
+#else
+		(struct dentry *dentry, const char *name,
+		 const void *value, size_t size, int flags, int handler_flags)
+{
+	name = full_name_hack(name, handler_flags);
+#endif
 	return scoutfs_xattr_set(dentry, name, value, size, flags);
 }
 
@@ -905,14 +925,22 @@ static const struct xattr_handler scoutfs_xattr_security_handler = {
 };
 
 static const struct xattr_handler scoutfs_xattr_acl_access_handler = {
+#ifdef KC_XATTR_HANDLER_NAME
+	.name   = XATTR_NAME_POSIX_ACL_ACCESS,
+#else
 	.prefix = XATTR_NAME_POSIX_ACL_ACCESS,
+#endif
 	.flags  = ACL_TYPE_ACCESS,
 	.get    = scoutfs_acl_get_xattr,
 	.set    = scoutfs_acl_set_xattr,
 };
 
 static const struct xattr_handler scoutfs_xattr_acl_default_handler = {
+#ifdef KC_XATTR_HANDLER_NAME
+	.name   = XATTR_NAME_POSIX_ACL_DEFAULT,
+#else
 	.prefix = XATTR_NAME_POSIX_ACL_DEFAULT,
+#endif
 	.flags  = ACL_TYPE_DEFAULT,
 	.get    = scoutfs_acl_get_xattr,
 	.set    = scoutfs_acl_set_xattr,

tests/funcs/exec.sh

@@ -39,6 +39,18 @@ t_quiet()
 		t_fail "quiet command failed"
 }
 
+#
+# Quietly run a command during a test.  The output is logged but only
+# the return code is printed, presumably because the output contains
+# a lot of invocation specific text that is difficult to filter.
+#
+t_rc()
+{
+	echo "# $*" >> "$T_TMP.rc.log"
+	"$@" >> "$T_TMP.rc.log" 2>&1
+	echo "rc: $?"
+}
+
 #
 # redirect test output back to the output of the invoking script intead
 # of the compared output.

tests/funcs/filter.sh

@@ -18,6 +18,7 @@ t_filter_dmesg()
 
 	# the kernel can just be noisy
 	re=" used greatest stack depth: "
+	re="$re|sched: RT throttling activated"
 
 	# mkfs/mount checks partition tables
 	re="$re|unknown partition table"
@@ -61,6 +62,7 @@ t_filter_dmesg()
 	re="$re|scoutfs .* error: meta_super META flag not set"
 	re="$re|scoutfs .* error: could not open metadev:.*"
 	re="$re|scoutfs .* error: Unknown or malformed option,.*"
+	re="$re|scoutfs .* error: invalid quorum_heartbeat_timeout_ms value"
 
 	# in debugging kernels we can slow things down a bit
 	re="$re|hrtimer: interrupt took .*"
@@ -81,6 +83,10 @@ t_filter_dmesg()
 	re="$re|scoutfs .* error .* freeing merged btree blocks.*.final commit del.upd freeing item"
 	re="$re|scoutfs .* error .*reading quorum block.*to update event.*"
 	re="$re|scoutfs .* error.*server failed to bind to.*"
+	re="$re|scoutfs .* critical transaction commit failure.*"
+
+	# change-devices causes loop device resizing
+	re="$re|loop[0-9].* detected capacity change from.*"
 
 	egrep -v "($re)" 
 }

tests/funcs/fs.sh

@@ -75,6 +75,15 @@ t_fs_nrs()
 	seq 0 $((T_NR_MOUNTS - 1))
 }
 
+#
+# output the fs nrs of quorum nodes, we "know" that
+# the quorum nrs are the first consequtive nrs
+#
+t_quorum_nrs()
+{
+	seq 0 $((T_QUORUM - 1))
+}
+
 #
 # outputs "1" if the fs number has "1" in its quorum/is_leader file.
 # All other cases output 0, including the fs nr being a client which
@@ -144,7 +153,27 @@ t_mount()
 	test "$nr" -lt "$T_NR_MOUNTS" || \
 		t_fail "fs nr $nr invalid"
 
-	eval t_quiet mount -t scoutfs \$T_O$nr \$T_DB$nr \$T_M$nr
+	eval t_quiet mount -t scoutfs \$T_O$nr\$opt \$T_DB$nr \$T_M$nr
+}
+
+#
+# Mount with an optional mount option string.  If the string is empty
+# then the saved mount options are used.  If the string has contents
+# then it is appended to the end of the saved options with a separating
+# comma.
+#
+# Unlike t_mount this won't inherently fail in t_quiet, errors are
+# returned so bad options can be tested.
+#
+t_mount_opt()
+{
+	local nr="$1"
+	local opt="${2:+,$2}"
+
+	test "$nr" -lt "$T_NR_MOUNTS" || \
+		t_fail "fs nr $nr invalid"
+
+	eval mount -t scoutfs \$T_O$nr\$opt \$T_DB$nr \$T_M$nr
 }
 
 t_umount()
@@ -391,7 +420,7 @@ t_set_sysfs_mount_option() {
 	local val="$3"
 	local opt="$(t_sysfs_path $nr)/mount_options/$name"
 
-	echo "$val" > "$opt"
+	echo "$val" > "$opt" 2>/dev/null
 }
 
 t_set_all_sysfs_mount_options() {

tests/golden/basic-posix-consistency

@@ -47,7 +47,7 @@ four
 --- dir within dir
 --- overwrite file
 --- can't overwrite non-empty dir
-mv: cannot move ‘/mnt/test/test/basic-posix-consistency/dir/c/clobber’ to ‘/mnt/test/test/basic-posix-consistency/dir/a/dir’: Directory not empty
+mv: cannot move '/mnt/test/test/basic-posix-consistency/dir/c/clobber' to '/mnt/test/test/basic-posix-consistency/dir/a/dir': Directory not empty
 --- can overwrite empty dir
 --- can rename into root
 == path resoluion

tests/golden/change-devices

@@ -0,0 +1,27 @@
+== make tmp sparse data dev files
+== make scratch fs
+== small new data device fails
+rc: 1
+== check sees data device errors
+rc: 1
+rc: 0
+== preparing while mounted fails
+rc: 1
+== preparing without recovery fails
+rc: 1
+== check sees metadata errors
+rc: 1
+rc: 1
+== preparing with file data fails
+rc: 1
+== preparing after emptied
+rc: 0
+== checks pass
+rc: 0
+rc: 0
+== using prepared
+== preparing larger and resizing
+rc: 0
+equal_prepared
+large_prepared
+resized larger test rc: 0

tests/golden/data-prealloc

@@ -24,3 +24,307 @@
 /mnt/test/test/data-prealloc/file-2: 5 extents found
 /mnt/test/test/data-prealloc/file-1: 3 extents found
 /mnt/test/test/data-prealloc/file-2: 3 extents found
+== block writes into region allocs hole
+wrote blk 24
+wrote blk 32
+wrote blk 40
+wrote blk 55
+wrote blk 63
+wrote blk 71
+wrote blk 72
+wrote blk 79
+wrote blk 80
+wrote blk 87
+wrote blk 88
+wrote blk 95
+before:
+24.. 1: 
+32.. 1: 
+40.. 1: 
+55.. 1: 
+63.. 1: 
+71.. 2: 
+79.. 2: 
+87.. 2: 
+95.. 1: eof
+writing into existing 0 at pos 0
+wrote blk 0
+0.. 1: 
+1.. 7: unwritten
+24.. 1: 
+32.. 1: 
+40.. 1: 
+55.. 1: 
+63.. 1: 
+71.. 2: 
+79.. 2: 
+87.. 2: 
+95.. 1: eof
+writing into existing 0 at pos 1
+wrote blk 15
+0.. 1: 
+1.. 14: unwritten
+15.. 1: 
+24.. 1: 
+32.. 1: 
+40.. 1: 
+55.. 1: 
+63.. 1: 
+71.. 2: 
+79.. 2: 
+87.. 2: 
+95.. 1: eof
+writing into existing 0 at pos 2
+wrote blk 19
+0.. 1: 
+1.. 14: unwritten
+15.. 1: 
+16.. 3: unwritten
+19.. 1: 
+20.. 4: unwritten
+24.. 1: 
+32.. 1: 
+40.. 1: 
+55.. 1: 
+63.. 1: 
+71.. 2: 
+79.. 2: 
+87.. 2: 
+95.. 1: eof
+writing into existing 1 at pos 0
+wrote blk 25
+0.. 1: 
+1.. 14: unwritten
+15.. 1: 
+16.. 3: unwritten
+19.. 1: 
+20.. 4: unwritten
+24.. 1: 
+25.. 1: 
+26.. 6: unwritten
+32.. 1: 
+40.. 1: 
+55.. 1: 
+63.. 1: 
+71.. 2: 
+79.. 2: 
+87.. 2: 
+95.. 1: eof
+writing into existing 1 at pos 1
+wrote blk 39
+0.. 1: 
+1.. 14: unwritten
+15.. 1: 
+16.. 3: unwritten
+19.. 1: 
+20.. 4: unwritten
+24.. 1: 
+25.. 1: 
+26.. 6: unwritten
+32.. 1: 
+39.. 1: 
+40.. 1: 
+55.. 1: 
+63.. 1: 
+71.. 2: 
+79.. 2: 
+87.. 2: 
+95.. 1: eof
+writing into existing 1 at pos 2
+wrote blk 44
+0.. 1: 
+1.. 14: unwritten
+15.. 1: 
+16.. 3: unwritten
+19.. 1: 
+20.. 4: unwritten
+24.. 1: 
+25.. 1: 
+26.. 6: unwritten
+32.. 1: 
+39.. 1: 
+40.. 1: 
+44.. 1: 
+45.. 3: unwritten
+55.. 1: 
+63.. 1: 
+71.. 2: 
+79.. 2: 
+87.. 2: 
+95.. 1: eof
+writing into existing 2 at pos 0
+wrote blk 48
+0.. 1: 
+1.. 14: unwritten
+15.. 1: 
+16.. 3: unwritten
+19.. 1: 
+20.. 4: unwritten
+24.. 1: 
+25.. 1: 
+26.. 6: unwritten
+32.. 1: 
+39.. 1: 
+40.. 1: 
+44.. 1: 
+45.. 3: unwritten
+48.. 1: 
+49.. 6: unwritten
+55.. 1: 
+63.. 1: 
+71.. 2: 
+79.. 2: 
+87.. 2: 
+95.. 1: eof
+writing into existing 2 at pos 1
+wrote blk 62
+0.. 1: 
+1.. 14: unwritten
+15.. 1: 
+16.. 3: unwritten
+19.. 1: 
+20.. 4: unwritten
+24.. 1: 
+25.. 1: 
+26.. 6: unwritten
+32.. 1: 
+39.. 1: 
+40.. 1: 
+44.. 1: 
+45.. 3: unwritten
+48.. 1: 
+49.. 6: unwritten
+55.. 1: 
+56.. 6: unwritten
+62.. 1: 
+63.. 1: 
+71.. 2: 
+79.. 2: 
+87.. 2: 
+95.. 1: eof
+writing into existing 2 at pos 2
+wrote blk 67
+0.. 1: 
+1.. 14: unwritten
+15.. 1: 
+16.. 3: unwritten
+19.. 1: 
+20.. 4: unwritten
+24.. 1: 
+25.. 1: 
+26.. 6: unwritten
+32.. 1: 
+39.. 1: 
+40.. 1: 
+44.. 1: 
+45.. 3: unwritten
+48.. 1: 
+49.. 6: unwritten
+55.. 1: 
+56.. 6: unwritten
+62.. 1: 
+63.. 1: 
+64.. 3: unwritten
+67.. 1: 
+68.. 3: unwritten
+71.. 2: 
+79.. 2: 
+87.. 2: 
+95.. 1: eof
+writing into existing 3 at pos 0
+wrote blk 73
+0.. 1: 
+1.. 14: unwritten
+15.. 1: 
+16.. 3: unwritten
+19.. 1: 
+20.. 4: unwritten
+24.. 1: 
+25.. 1: 
+26.. 6: unwritten
+32.. 1: 
+39.. 1: 
+40.. 1: 
+44.. 1: 
+45.. 3: unwritten
+48.. 1: 
+49.. 6: unwritten
+55.. 1: 
+56.. 6: unwritten
+62.. 1: 
+63.. 1: 
+64.. 3: unwritten
+67.. 1: 
+68.. 3: unwritten
+71.. 2: 
+73.. 1: 
+74.. 5: unwritten
+79.. 2: 
+87.. 2: 
+95.. 1: eof
+writing into existing 3 at pos 1
+wrote blk 86
+0.. 1: 
+1.. 14: unwritten
+15.. 1: 
+16.. 3: unwritten
+19.. 1: 
+20.. 4: unwritten
+24.. 1: 
+25.. 1: 
+26.. 6: unwritten
+32.. 1: 
+39.. 1: 
+40.. 1: 
+44.. 1: 
+45.. 3: unwritten
+48.. 1: 
+49.. 6: unwritten
+55.. 1: 
+56.. 6: unwritten
+62.. 1: 
+63.. 1: 
+64.. 3: unwritten
+67.. 1: 
+68.. 3: unwritten
+71.. 2: 
+73.. 1: 
+74.. 5: unwritten
+79.. 2: 
+86.. 1: 
+87.. 2: 
+95.. 1: eof
+writing into existing 3 at pos 2
+wrote blk 92
+0.. 1: 
+1.. 14: unwritten
+15.. 1: 
+16.. 3: unwritten
+19.. 1: 
+20.. 4: unwritten
+24.. 1: 
+25.. 1: 
+26.. 6: unwritten
+32.. 1: 
+39.. 1: 
+40.. 1: 
+44.. 1: 
+45.. 3: unwritten
+48.. 1: 
+49.. 6: unwritten
+55.. 1: 
+56.. 6: unwritten
+62.. 1: 
+63.. 1: 
+64.. 3: unwritten
+67.. 1: 
+68.. 3: unwritten
+71.. 2: 
+73.. 1: 
+74.. 5: unwritten
+79.. 2: 
+86.. 1: 
+87.. 2: 
+92.. 1: 
+93.. 2: unwritten
+95.. 1: eof

tests/golden/get-referring-entries

@@ -0,0 +1,18 @@
+== root inode returns nothing
+== crazy large unused inode does nothing
+== basic entry
+file
+== rename
+renamed
+== hard link
+file
+link
+== removal
+== different dirs
+== file types
+type b name block
+type c name char
+type d name dir
+type f name file
+type l name symlink
+== all name lengths work

tests/golden/inode-deletion

@@ -17,7 +17,7 @@ ino not found in dseq index
 mount 0 contents after mount 1 rm: contents
 ino found in dseq index
 ino found in dseq index
-stat: cannot stat ‘/mnt/test/test/inode-deletion/file’: No such file or directory
+stat: cannot stat '/mnt/test/test/inode-deletion/file': No such file or directory
 ino not found in dseq index
 ino not found in dseq index
 == lots of deletions use one open map

tests/golden/offline-extent-waiting

@@ -20,10 +20,10 @@ offline waiting should now have two known entries:
 data_wait_err found 2 waiters.
 offline waiting should now have 0 known entries:
 0
-dd: error reading ‘/mnt/test/test/offline-extent-waiting/dir/file’: Input/output error
+dd: error reading '/mnt/test/test/offline-extent-waiting/dir/file': Input/output error
 0+0 records in
 0+0 records out
-dd: error reading ‘/mnt/test/test/offline-extent-waiting/dir/file’: Input/output error
+dd: error reading '/mnt/test/test/offline-extent-waiting/dir/file': Input/output error
 0+0 records in
 0+0 records out
 offline waiting should be empty again:

tests/golden/quorum-heartbeat-timeout

@@ -0,0 +1,5 @@
+== bad timeout values fail
+== bad mount option fails
+== mount option
+== sysfs
+== reset all options

tests/golden/xfstests

@@ -241,7 +241,6 @@ generic/312
 generic/314
 generic/316
 generic/317
-generic/318
 generic/324
 generic/326
 generic/327

tests/run-tests.sh

@@ -1,5 +1,8 @@
 #!/usr/bin/bash
 
+# Force system tools to use ASCII quotes
+export LC_ALL=C
+
 #
 # XXX
 #  - could have helper functions for waiting for pids

tests/sequence

@@ -5,6 +5,7 @@ inode-items-updated.sh
 simple-inode-index.sh
 simple-staging.sh
 simple-release-extents.sh
+get-referring-entries.sh
 fallocate.sh
 basic-truncate.sh
 data-prealloc.sh
@@ -36,7 +37,9 @@ cross-mount-data-free.sh
 persistent-item-vers.sh
 setup-error-teardown.sh
 resize-devices.sh
+change-devices.sh
 fence-and-reclaim.sh
+quorum-heartbeat-timeout.sh
 orphan-inodes.sh
 mount-unmount-race.sh
 client-unmount-recovery.sh

tests/src/handle_fsetxattr.c

@@ -48,7 +48,7 @@ struct our_handle {
 static void exit_usage(void)
 {
 	printf(" -h/-?         output this usage message and exit\n"
-	       " -e            keep trying on enoent, consider success an error\n"
+	       " -e            keep trying on enoent and estale, consider success an error\n"
 	       " -i <num>      64bit inode number for handle open, can be multiple\n"
 	       " -m <string>   scoutfs mount path string for ioctl fd\n"
 	       " -n <string>   optional xattr name string, defaults to \""DEFAULT_NAME"\"\n"
@@ -149,7 +149,7 @@ int main(int argc, char **argv)
 
 				fd = open_by_handle_at(mntfd, &handle.handle, O_RDWR);
 				if (fd == -1) {
-					if (!enoent_success_err || errno != ENOENT) {
+					if (!enoent_success_err || ( errno != ENOENT && errno != ESTALE )) {
 						perror("open_by_handle_at");
 						return 1;
 					}

tests/tests/basic-bad-mounts.sh

@@ -12,7 +12,7 @@ mount_fail()
 }
 
 echo "== prepare devices, mount point, and logs"
-SCR="/mnt/scoutfs.extra"
+SCR="$T_TMPDIR/mnt.scratch"
 mkdir -p "$SCR"
 > $T_TMP.mount.out
 scoutfs mkfs -f -Q 0,127.0.0.1,53000 "$T_EX_META_DEV" "$T_EX_DATA_DEV" > $T_TMP.mkfs.out 2>&1 \

tests/tests/change-devices.sh

@@ -0,0 +1,76 @@
+#
+# test changing devices
+#
+
+echo "== make tmp sparse data dev files"
+sz=$(blockdev --getsize64 "$T_EX_DATA_DEV")
+large_sz=$((sz * 2))
+touch "$T_TMP."{small,equal,large}
+truncate -s 1MB "$T_TMP.small"
+truncate -s $sz "$T_TMP.equal"
+truncate -s $large_sz "$T_TMP.large"
+
+echo "== make scratch fs"
+t_quiet scoutfs mkfs -f -Q 0,127.0.0.1,53000 "$T_EX_META_DEV" "$T_EX_DATA_DEV"
+SCR="$T_TMPDIR/mnt.scratch"
+mkdir -p "$SCR"
+
+echo "== small new data device fails"
+t_rc scoutfs prepare-empty-data-device "$T_EX_META_DEV" "$T_TMP.small"
+
+echo "== check sees data device errors"
+t_rc scoutfs prepare-empty-data-device --check "$T_EX_META_DEV" "$T_TMP.small"
+t_rc scoutfs prepare-empty-data-device --check "$T_EX_META_DEV"
+
+echo "== preparing while mounted fails"
+mount -t scoutfs -o metadev_path=$T_EX_META_DEV,quorum_slot_nr=0 "$T_EX_DATA_DEV" "$SCR"
+t_rc scoutfs prepare-empty-data-device "$T_EX_META_DEV" "$T_TMP.equal"
+umount "$SCR"
+
+echo "== preparing without recovery fails"
+mount -t scoutfs -o metadev_path=$T_EX_META_DEV,quorum_slot_nr=0 "$T_EX_DATA_DEV" "$SCR"
+umount -f "$SCR"
+t_rc scoutfs prepare-empty-data-device "$T_EX_META_DEV" "$T_TMP.equal"
+
+echo "== check sees metadata errors"
+t_rc scoutfs prepare-empty-data-device --check "$T_EX_META_DEV"
+t_rc scoutfs prepare-empty-data-device --check "$T_EX_META_DEV" "$T_TMP.equal"
+
+echo "== preparing with file data fails"
+mount -t scoutfs -o metadev_path=$T_EX_META_DEV,quorum_slot_nr=0 "$T_EX_DATA_DEV" "$SCR"
+echo hi > "$SCR"/file
+umount "$SCR"
+scoutfs print "$T_EX_META_DEV" > "$T_TMP.print"
+t_rc scoutfs prepare-empty-data-device "$T_EX_META_DEV" "$T_TMP.equal"
+
+echo "== preparing after emptied"
+mount -t scoutfs -o metadev_path=$T_EX_META_DEV,quorum_slot_nr=0 "$T_EX_DATA_DEV" "$SCR"
+rm -f "$SCR"/file
+umount "$SCR"
+t_rc scoutfs prepare-empty-data-device "$T_EX_META_DEV" "$T_TMP.equal"
+
+echo "== checks pass"
+t_rc scoutfs prepare-empty-data-device --check "$T_EX_META_DEV"
+t_rc scoutfs prepare-empty-data-device --check "$T_EX_META_DEV" "$T_TMP.equal"
+
+echo "== using prepared"
+scr_loop=$(losetup --find --show "$T_TMP.equal")
+mount -t scoutfs -o metadev_path=$T_EX_META_DEV,quorum_slot_nr=0 "$scr_loop" "$SCR"
+touch "$SCR"/equal_prepared
+equal_tot=$(scoutfs statfs -s total_data_blocks -p "$SCR")
+umount "$SCR"
+losetup -d "$scr_loop"
+
+echo "== preparing larger and resizing"
+t_rc scoutfs prepare-empty-data-device "$T_EX_META_DEV" "$T_TMP.large"
+scr_loop=$(losetup --find --show "$T_TMP.large")
+mount -t scoutfs -o metadev_path=$T_EX_META_DEV,quorum_slot_nr=0 "$scr_loop" "$SCR"
+touch "$SCR"/large_prepared
+ls "$SCR"
+scoutfs resize-devices -p "$SCR" -d $large_sz
+large_tot=$(scoutfs statfs -s total_data_blocks -p "$SCR")
+test "$large_tot" -gt "$equal_tot" ; echo "resized larger test rc: $?"
+umount "$SCR"
+losetup -d "$scr_loop"
+
+t_pass

tests/tests/data-prealloc.sh

@@ -6,6 +6,15 @@
 #
 t_require_commands scoutfs stat filefrag dd touch truncate
 
+write_block()
+{
+	local file="$1"
+	local blk="$2"
+
+	dd if=/dev/zero of="$file" bs=4096 seek=$blk count=1 conv=notrunc status=none
+	echo "wrote blk $blk"
+}
+
 write_forwards()
 {
 	local prefix="$1"
@@ -70,6 +79,25 @@ print_extents_found()
 	filefrag "$prefix"* 2>&1 | grep "extent.*found" | t_filter_fs
 }
 
+#
+# print the logical start, len, and flags if they're there.
+#
+print_logical_extents()
+{
+	local file="$1"
+
+	filefrag -v -b4096 "$file" 2>&1 | t_filter_fs | awk '
+		($1 ~ /[0-9]+:/) {
+			if ($NF !~  /[0-9]+:/) {
+				flags=$NF
+			} else {
+				flags=""
+			}
+			print $2, $6, flags
+		}
+	' | sed 's/last,eof/eof/'
+}
+
 t_save_all_sysfs_mount_options data_prealloc_blocks
 t_save_all_sysfs_mount_options data_prealloc_contig_only
 restore_options()
@@ -133,4 +161,71 @@ t_set_sysfs_mount_option 0 data_prealloc_contig_only 0
 write_forwards $prefix 3
 print_extents_found $prefix
 
+#
+# prepare aligned regions of 8 blocks that we'll write into.
+# We'll right into the first, last, and middle block of each
+# region which was prepared with no existing extents, one at
+# the start, and one at the end.
+#
+# Let's keep this last because it creates a ton of output to read
+# through.  The correct output is tied to preallocation strategy so it
+# has to be verified each time we change preallocation.
+#
+echo "== block writes into region allocs hole" 
+t_set_sysfs_mount_option 0 data_prealloc_blocks 8
+t_set_sysfs_mount_option 0 data_prealloc_contig_only 1
+touch "$prefix"
+truncate -s 0 "$prefix"
+
+# write initial blocks in regions
+base=0
+for sides in 0 1 2 3; do
+	for i in 0 1 2; do
+                case "$sides" in
+			# none
+			0) ;;
+			# left
+			1) write_block $prefix $((base + 0)) ;;
+			# right
+			2) write_block $prefix $((base + 7)) ;;
+			# both
+			3) write_block $prefix $((base + 0)) 
+			   write_block $prefix $((base + 7)) ;;
+		esac
+		((base+=8))
+	done
+done
+
+echo before:
+print_logical_extents "$prefix"
+
+# now write into the first, middle, and last empty block of each
+t_set_sysfs_mount_option 0 data_prealloc_contig_only 0
+base=0
+for sides in 0 1 2 3; do
+	for i in 0 1 2; do
+		echo "writing into existing $sides at pos $i"
+		case "$sides" in
+			# none
+			0) left=$base; right=$((base + 7));;
+			# left
+			1) left=$((base + 1)); right=$((base + 7));;
+			# right
+			2) left=$((base)); right=$((base + 6));;
+			# both
+			3) left=$((base + 1)); right=$((base + 6));;
+		esac
+		case "$i" in
+			# start
+			0) write_block $prefix $left ;;
+			# end
+			1) write_block $prefix $right ;;
+			# mid (both has 6 blocks internally)
+			2) write_block $prefix $((left + 3)) ;;
+		esac
+		print_logical_extents "$prefix"
+		((base+=8))
+	done
+done
+
 t_pass

tests/tests/enospc.sh

@@ -59,7 +59,7 @@ echo "== make small meta fs"
 # meta device just big enough for reserves and the metadata we'll fill
 scoutfs mkfs -A -f -Q 0,127.0.0.1,53000 -m 10G "$T_EX_META_DEV" "$T_EX_DATA_DEV" > $T_TMP.mkfs.out 2>&1 || \
 	t_fail "mkfs failed"
-SCR="/mnt/scoutfs.enospc"
+SCR="$T_TMPDIR/mnt.scratch"
 mkdir -p "$SCR"
 mount -t scoutfs -o metadev_path=$T_EX_META_DEV,quorum_slot_nr=0 \
 	"$T_EX_DATA_DEV" "$SCR"

tests/tests/get-referring-entries.sh

@@ -0,0 +1,99 @@
+
+#
+# Test _GET_REFERRING_ENTRIES ioctl via the get-referring-entries cli
+# command
+#
+
+# consistently print only entry names
+filter_names() {
+	exec cut -d ' ' -f 8- | sort
+}
+
+# print entries with type characters to match find.  not happy with hard
+# coding, but abi won't change much.
+filter_types() {
+	exec cut -d ' ' -f 5- | \
+	sed \
+		-e 's/type 1 /type p /' \
+		-e 's/type 2 /type c /' \
+		-e 's/type 4 /type d /' \
+		-e 's/type 6 /type b /' \
+		-e 's/type 8 /type f /' \
+		-e 's/type 10 /type l /' \
+		-e 's/type 12 /type s /' \
+		| \
+	sort
+}
+
+n_chars() {
+	local n="$1"
+	printf 'A%.0s' $(eval echo {1..\$n})
+}
+
+GRE="scoutfs get-referring-entries -p $T_M0"
+
+echo "== root inode returns nothing"
+$GRE 1
+
+echo "== crazy large unused inode does nothing"
+$GRE 4611686018427387904 # 1 << 62
+
+echo "== basic entry"
+touch $T_D0/file
+ino=$(stat -c '%i' $T_D0/file)
+$GRE $ino | filter_names
+
+echo "== rename"
+mv $T_D0/file $T_D0/renamed
+$GRE $ino | filter_names
+
+echo "== hard link"
+mv $T_D0/renamed $T_D0/file
+ln $T_D0/file $T_D0/link
+$GRE $ino | filter_names
+
+echo "== removal"
+rm $T_D0/file $T_D0/link
+$GRE $ino
+
+echo "== different dirs"
+touch $T_D0/file
+ino=$(stat -c '%i' $T_D0/file)
+for i in $(seq 1 10); do
+	mkdir $T_D0/dir-$i
+	ln $T_D0/file $T_D0/dir-$i/file-$i
+done
+diff -u <(find $T_D0 -type f -printf '%f\n' | sort) <($GRE $ino | filter_names)
+rm $T_D0/file
+
+echo "== file types"
+mkdir $T_D0/dir
+touch $T_D0/dir/file
+mkdir $T_D0/dir/dir
+ln -s $T_D0/dir/file $T_D0/dir/symlink
+mknod $T_D0/dir/char c 1 3 # null
+mknod $T_D0/dir/block b 7 0 # loop0
+for name in $(ls -UA $T_D0/dir | sort); do
+	ino=$(stat -c '%i' $T_D0/dir/$name)
+	$GRE $ino | filter_types
+done
+rm -rf $T_D0/dir
+
+echo "== all name lengths work"
+mkdir $T_D0/dir
+touch $T_D0/dir/file
+ino=$(stat -c '%i' $T_D0/dir/file)
+name=""
+> $T_TMP.unsorted
+for i in $(seq 1 255); do
+	name+="a"
+	echo "$name" >> $T_TMP.unsorted
+	ln $T_D0/dir/file $T_D0/dir/$name
+done
+sort $T_TMP.unsorted > $T_TMP.sorted
+rm $T_D0/dir/file
+$GRE $ino | filter_names > $T_TMP.gre
+diff -u $T_TMP.sorted $T_TMP.gre
+rm -rf $T_D0/dir
+
+t_pass

tests/tests/inode-deletion.sh

@@ -72,7 +72,7 @@ check_ino_index "$ino" "$dseq" "$T_M0"
 check_ino_index "$ino" "$dseq" "$T_M1"
 exec {FD}>&-  # close
 # we know that revalidating will unhash the remote dentry
-stat "$T_D0/file" 2>&1 | t_filter_fs
+stat "$T_D0/file" 2>&1 | sed 's/cannot statx/cannot stat/' | t_filter_fs
 check_ino_index "$ino" "$dseq" "$T_M0"
 check_ino_index "$ino" "$dseq" "$T_M1"
 

tests/tests/quorum-heartbeat-timeout.sh

@@ -0,0 +1,117 @@
+#
+# test that the quorum_heartbeat_time_ms option affects how long it
+# takes to recover from a failed mount.
+#
+
+t_require_mounts 2
+
+time_ms()
+{
+	# time_t in seconds, then trunate nanoseconds to 3 most dig digits
+	date +%s%3N
+}
+
+set_bad_timeout() {
+	local to="$1"
+	t_set_sysfs_mount_option 0 quorum_heartbeat_timeout_ms $to && \
+		t_fail "set bad q hb to $to"
+}
+
+set_timeout()
+{
+	local nr="$1"
+	local how="$2"
+	local to="$3"
+	local is
+
+	if [ $how == "sysfs" ]; then
+		t_set_sysfs_mount_option $nr quorum_heartbeat_timeout_ms $to
+	fi
+	if [ $how == "mount" ]; then
+		t_umount $nr
+		t_mount_opt $nr "quorum_heartbeat_timeout_ms=$to"
+	fi
+
+	is=$(t_get_sysfs_mount_option $nr quorum_heartbeat_timeout_ms)
+
+	if [ "$is" != "$to" ]; then
+		t_fail "tried to set qhbto on $nr via $how to $to but got $is"
+	fi
+}
+
+test_timeout()
+{
+	local how="$1"
+	local to="$2"
+	local start
+	local nr
+	local sv
+	local delay
+	local low
+	local high
+
+	# set timeout on non-server quorum mounts
+	sv=$(t_server_nr)
+	for nr in $(t_quorum_nrs); do
+		if [ $nr -ne $sv ]; then
+			set_timeout $nr $how $to
+		fi
+	done
+
+	# give followers time to recv heartbeats and reset timeouts
+	sleep 1
+
+	# tear down the current server/leader
+	t_force_umount $sv
+
+	# see how long it takes for the next leader to start
+	start=$(time_ms)
+	t_wait_for_leader
+	delay=$(($(time_ms) - start))
+
+	# kind of fun to have these logged
+	echo "to $to delay $delay" >> $T_TMP.delay
+
+	# restore the mount that we tore down
+	t_mount $sv
+
+	# make sure the new leader delay was reasonable, allowing for some slack
+	low=$((to - 1000))
+	high=$((to + 5000))
+
+	# make sure the new leader delay was reasonable
+	test "$delay" -lt "$low" && t_fail "delay $delay < low $low (to $to)"
+	test "$delay" -gt "$high" && t_fail "delay $delay > high $high (to $to)"
+}
+
+echo "== bad timeout values fail"
+set_bad_timeout 0
+set_bad_timeout -1
+set_bad_timeout 1000000
+
+echo "== bad mount option fails"
+if [ "$(t_server_nr)" == 0 ]; then
+	nr=1
+else
+	nr=0
+fi
+t_umount $nr
+t_mount_opt $nr "quorum_heartbeat_timeout_ms=1000000" 2>/dev/null && \
+	t_fail "bad mount option succeeded"
+t_mount $nr
+
+echo "== mount option"
+def=$(t_get_sysfs_mount_option 0 quorum_heartbeat_timeout_ms)
+test_timeout mount $def
+test_timeout mount 3000
+test_timeout mount $((def + 19000))
+
+echo "== sysfs"
+test_timeout sysfs $def
+test_timeout sysfs 3000
+test_timeout sysfs $((def + 19000))
+
+echo "== reset all options"
+t_remount_all
+
+t_pass

tests/tests/resize-devices.sh

@@ -73,7 +73,7 @@ echo "== make initial small fs"
 scoutfs mkfs -A -f -Q 0,127.0.0.1,53000 -m $quarter_meta -d $quarter_data \
 	"$T_EX_META_DEV" "$T_EX_DATA_DEV" > $T_TMP.mkfs.out 2>&1 || \
 		t_fail "mkfs failed"
-SCR="/mnt/scoutfs.enospc"
+SCR="$T_TMPDIR/mnt.scratch"
 mkdir -p "$SCR"
 mount -t scoutfs -o metadev_path=$T_EX_META_DEV,quorum_slot_nr=0 \
 	"$T_EX_DATA_DEV" "$SCR"

tests/tests/setattr_more.sh

@@ -55,10 +55,17 @@ scoutfs setattr -t 67305985.999999999 -V 1 -s 1 "$FILE" 2>&1 | t_filter_fs
 TZ=GMT stat -c "%z" "$FILE"
 rm "$FILE"
 
+#
+# With e2fsprogs-v1.42.10-10-g29758d2f, the output of filefrag 'flags' changes
+# significantly. First, the _LAST flag is now output. Second, the 'unknown'
+# flag is now printed out as 'unknown_loc'. To compensate for this, we check
+# and replace the "correct" output for new versions here with the expected
+# value.
+#
 echo "== large offline extents are created"
 touch "$FILE"
 scoutfs setattr -V 1 -o -s $((10007 * 4096)) "$FILE" 2>&1 | t_filter_fs
-filefrag -v -b4096 "$FILE" 2>&1 | t_filter_fs
+filefrag -v -b4096 "$FILE" 2>&1 | sed 's/last,unknown_loc,eof$/unknown,eof/' | t_filter_fs
 rm "$FILE"
 
 # had a bug where we were creating extents that were too long

tests/tests/simple-xattr-unit.sh

@@ -27,15 +27,9 @@ test_xattr_lengths() {
 	echo "key len $name_len val len $val_len" >> "$T_TMP.log"
 	setfattr -n $name -v \"$val\" "$FILE"
 
-	# grep has trouble with enormous args?  so we dump the
-	# name=value to a file and compare with a known good file
-	getfattr -d --absolute-names "$FILE" | grep "$name" > "$T_TMP.got"
+	getfattr -d --only-values --absolute-names "$FILE" -n "$name" > "$T_TMP.got"
+	echo -n "$val" > "$T_TMP.good"
 
-	if [ $val_len == 0 ]; then
-		echo "$name" > "$T_TMP.good"
-	else
-		echo "$name=\"$val\"" > "$T_TMP.good"
-	fi
 	cmp "$T_TMP.good" "$T_TMP.got" || \
 		t_fail "cmp failed name len $name_len val len $val_len"
 

tests/tests/xfstests.sh

@@ -75,6 +75,7 @@ generic/215	# mmap missing
 generic/246	# mmap missing
 generic/247	# mmap missing
 generic/248	# mmap missing
+generic/318	# can't support user namespaces until v5.11
 generic/321	# requires selinux enabled for '+' in ls?
 generic/325	# mmap missing
 generic/338	# BUG_ON update inode error handling

utils/man/scoutfs.5

@@ -85,6 +85,25 @@ the options directory in the mount's sysfs directory.  Writing a new
 value will cause the next pending orphan scan to be rescheduled
 with the newly written delay time.
 .TP
+.B quorum_heartbeat_timeout_ms=<number>
+This option sets the amount of time, in milliseconds, that a quorum
+member will wait without receiving heartbeat messages from the current
+leader before trying to take over as leader.  This setting is per-mount
+and only changes the behavior of that mount.
+.sp
+This determines how long it may take before a failed leader is replaced
+by a waiting quorum member.  Setting it too low may lead to spurious
+fencing as active leaders are prematurely replaced due to task or
+network delays that prevent the quorum members from promptly sending and
+receiving messages.  The ideal setting is the longest acceptable
+downtime during server failover.  The default is 10000 (10s) and it can
+not be less than 2000 greater than 60000.
+.sp
+This option can be changed in an active mount by writing to its file in
+the options directory in the mount's sysfs directory.  Writing a new
+value will take effect the next time the quorum agent receives a
+heartbeat message and sets the next timeout.
+.TP
 .B quorum_slot_nr=<number>
 The quorum_slot_nr option assigns a quorum member slot to the mount.
 The mount will use the slot assignment to claim exclusive ownership of

utils/man/scoutfs.8

@@ -76,6 +76,97 @@ run when the file system will not be mounted.
 .RE
 .PD
 
+.TP
+.BI "counters [-t|--table] SYSFS-DIR"
+.sp
+Display the counters and their values for a mounted ScoutFS filesystem.
+.RS 1.0i
+.PD 0
+.sp
+.TP
+.B SYSFS-DIR
+The mount's sysfs directory in which to find the
+.B counters/
+directory when then contains files for each counter.
+The sysfs directory is
+of the form
+.I /sys/fs/scoutfs/f.<fsid>.r.<rid>/
+\&.
+.TP
+.B "-t, --table"
+Format the counters into a columnar table that fills the width of the display
+instead of printing one counter per line.
+.RE
+.PD
+
+.TP
+.BI "data-waiting {-I|--inode} INODE-NUM {-B|--block} BLOCK-NUM [-p|--path PATH]"
+.sp
+Display all the files and blocks for which there is a task blocked waiting on
+offline data.
+.sp
+The results are sorted by the file's inode number and the
+logical block offset that is being waited on.
+.sp
+Each line of output describes a block in a file that has a task waiting
+and is formatted as:
+.I "ino <nr> iblock <nr> ops [str]"
+\&. The ops string indicates blocked operations seperated by commas and can
+include
+.B read
+for a read operation,
+.B write
+for a write operation, and
+.B change_size
+for a truncate or extending write.
+.RS 1.0i
+.PD 0
+.sp
+.TP
+.B "-I, --inode INODE-NUM"
+Start iterating over waiting tasks from the given inode number.
+Value of 0 will show all waiting tasks.
+.TP
+.B "-B, --block BLOCK-NUM"
+Start iterating over waiting tasks from the given logical block number
+in the starting inode.  Value of 0 will show blocks in the first inode
+and then continue to show all blocks with tasks waiting in all the
+remaining inodes.
+.TP
+.B "-p, --path PATH"
+A path within a ScoutFS filesystem.
+.RE
+.PD
+
+.TP
+.BI "data-wait-err {-I|--inode} INODE-NUM {-V|--version} VER-NUM {-F|--offset} OFF-NUM {-C|--count} COUNT {-O|--op} OP {-E|--err} ERR [-p|--path PATH]"
+.sp
+Return error from matching waiters.
+.RS 1.0i
+.PD 0
+.sp
+.TP
+.B "-C, --count COUNT"
+Count.
+.TP
+.B "-E, --err ERR"
+Error.
+.TP
+.B "-F, --offset OFF-NUM"
+Offset. May be expressed in bytes, or with KMGTP (Kibi, Mibi, etc.) size
+suffixes.
+.TP
+.B "-I, --inode INODE-NUM"
+Inode number.
+.TP
+.B "-O, --op OP"
+Operation. One of: "read", "write", "change_size".
+.TP
+.B "-p, --path PATH"
+A path within a ScoutFS filesystem.
+.RE
+.PD
+
 .TP
 .BI "df [-h|--human-readable] [-p|--path PATH]"
 .sp
@@ -93,6 +184,95 @@ A path within a ScoutFS filesystem.
 .RE
 .PD
 
+.TP
+.BI "get-allocated-inos [-i|--ino INO] [-s|--single] [-p|--path PATH]"
+.sp
+This debugging command prints allocated inode numbers.  It only prints
+inodes
+found in the group that contains the starting inode.  The printed inode
+numbers aren't necessarily reachable.  They could be anywhere in the
+process from being unlinked to finally deleted when their items
+were found.
+.RS 1.0i
+.PD 0
+.TP
+.sp
+.B "-i, --ino INO"
+The first 64bit inode number which could be printed.
+.TP
+.B "-s, --single"
+Only print the single starting inode when it is allocated, all other allocated
+inode numbers will be ignored.
+.TP
+.B "-p, --path PATH"
+A path within a ScoutFS filesystem.
+.RE
+.PD
+
+.TP
+.BI "get-referring-entries [-p|--path PATH] INO"
+.sp
+Find directory entries that reference an inode number.
+.sp
+Display all the directory entries that refer to a given inode.  Each
+entry includes the inode number of the directory that contains it, the
+d_off and d_type values for the entry as described by
+.BR readdir (3)
+, and the name of the entry.
+.RS 1.0i
+.PD 0
+.TP
+.sp
+.TP
+.B "-p, --path PATH"
+A path within a ScoutFS filesystem.
+.TP
+.B "INO"
+The inode number of the target inode.
+.RE
+.PD
+
+.TP
+.BI "ino-path INODE-NUM [-p|--path PATH]"
+.sp
+Display all paths that reference an inode number.
+.sp
+Ongoing filesystem changes, such as renaming a common parent of multiple paths,
+can cause displayed paths to be inconsistent.
+.RS 1.0i
+.PD 0
+.sp
+.TP
+.B "INODE-NUM"
+The inode number of the target inode.
+.TP
+.B "-p|--path PATH"
+A path within a ScoutFS filesystem.
+.RE
+.PD
+
+.TP
+.BI "list-hidden-xattrs FILE"
+.sp
+Display extended attributes starting with the
+.BR scoutfs.
+prefix and containing the
+.BR hide.
+tag
+which makes them invisible to
+.BR listxattr (2) .
+The names of each attribute are output, one per line.  Their order
+is not specified.
+.RS 1.0i
+.PD 0
+.TP
+.sp
+.B "FILE"
+The path to a file within a ScoutFS filesystem.  File permissions must allow
+reading.
+.RE
+.PD
+
 .TP
 .BI "mkfs META-DEVICE DATA-DEVICE {-Q|--quorum-slot} NR,ADDR,PORT [-m|--max-meta-size SIZE] [-d|--max-data-size SIZE] [-z|--data-alloc-zone-blocks BLOCKS] [-f|--force] [-A|--allow-small-size] [-V|--format-version VERS]"
 .sp
@@ -171,6 +351,79 @@ The range of supported versions is visible in the output of
 .RE
 .PD
 
+.TP
+.BI "prepare-empty-data-device {-c|--check} META-DEVICE DATA-DEVICE"
+.sp
+Prepare an unused device for use as the data device for an existing file
+system.  This will write an initialized super block to the specified
+data device, destroying any existing contents.  The specified metadata
+device will not be modified.  The file system must be fully unmounted
+and any client mount recovery must be complete.
+.sp
+The existing metadata device is read to ensure that it's safe to stop
+using the old data device.  The data block allocators must indicate that
+all data blocks are free.  If there are still data blocks referenced by
+files then the command will fail.  The contents of these files must be
+freed for the command to proceed.
+.sp
+A new super block is written to the new data device.  The device can
+then be used as the data device to mount the file system.  As this
+switch is made all client mounts must refer to the new device.  The old
+device is not modified and still contains a valid data super block that
+could be mounted, creating data device writes that wouldn't be read by
+mounts using the new device.
+.sp
+The number of data blocks available to the file system will not change
+as the new data device is used.  The new device must be large enough to
+store all the data blocks that were available on the old device.  If the
+new device is larger then its added capacity can be used by growing the
+new data device with the resize-devices command once it is mounted.
+.RS 1.0i
+.PD 0
+.TP
+.sp
+.B "-c, --check"
+Only check for errors that would prevent a new empty data device from
+being used.  No changes will be made to the data device.  If the data
+device is provided then its size will be checked to make sure that it is
+large enough.  This can be used to test the metadata for data references
+before destroying an old empty data device.
+.RE
+.PD
+
+.TP
+.BI "print {-S|--skip-likely-huge} META-DEVICE"
+.sp
+Prints out all of the metadata in the file system.  This makes no effort
+to ensure that the structures are consistent as they're traversed and
+can present structures that seem corrupt as they change as they're
+output.
+.RS 1.0i
+.PD 0
+.TP
+.sp
+.B "-S, --skip-likely-huge"
+Skip printing structures that are likely to be very large.  The
+structures that are skipped tend to be global and whose size tends to be
+related to the size of the volume.   Examples of skipped structures include
+the global fs items, srch files, and metadata and data
+allocators.  Similar structures that are not skipped are related to the
+number of mounts and are maintained at a relatively reasonable size.
+These include per-mount log trees, srch files, allocators, and the
+metadata allocators used by server commits.
+.sp
+Skipping the larger structures limits the print output to a relatively
+constant size rather than being a large multiple of the used metadata
+space of the volume making the output much more useful for inspection.
+.TP
+.B "META-DEVICE"
+The path to the metadata device for the filesystem whose metadata will be
+printed.  An attempt will be made to flush the host's buffer cache for
+this device with the BLKFLSBUF ioctl, or with posix_fadvise() if
+the path refers to a regular file.
+.RE
+.PD
+
 .TP
 .BI "resize-devices [-p|--path PATH] [-m|--meta-size SIZE] [-d|--data-size SIZE]"
 .sp
@@ -229,6 +482,92 @@ kibibytes, mebibytes, etc.
 .RE
 .PD
 
+.TP
+.BI "search-xattrs XATTR-NAME [-p|--path PATH]"
+.sp
+Display the inode numbers of inodes in the filesystem which may have
+an extended attribute with the given name.
+.sp
+The results may contain false positives.  The returned inode numbers
+should be checked to verify that the extended attribute is in fact
+present on the inode.
+.RS 1.0i
+.PD 0
+.TP
+.sp
+.B XATTR-NAME
+The full name of the extended attribute to search for as
+described in the
+.BR xattr (7)
+manual page.
+.TP
+.B "-p|--path PATH"
+A path within a ScoutFS filesystem.
+.RE
+.PD
+
+.TP
+.BI "setattr FILE [-d, --data-version=VERSION [-s, --size=SIZE [-o, --offline]]] [-t, --ctime=TIMESPEC]"
+.sp
+Set ScoutFS-specific attributes on a newly created zero-length file.
+.RS 1.0i
+.PD 0
+.sp
+.TP
+.B "-V, --data-version=VERSION"
+Set data version.
+.TP
+.B "-o, --offline"
+Set file contents as offline, not sparse. Requires
+.I --size
+option also be present.
+.TP
+.B "-s, --size=SIZE"
+Set file size. May be expressed in bytes, or with
+KMGTP (Kibi, Mibi, etc.) size suffixes. Requires
+.I --data-version
+option also be present.
+.TP
+.B "-t, --ctime=TIMESPEC"
+Set creation time using
+.I "<seconds-since-epoch>.<nanoseconds>"
+format.
+.RE
+.PD
+
+.TP
+.BI "stage ARCHIVE-FILE FILE {-V|--version} VERSION [-o, --offset OFF-NUM] [-l, --length LENGTH]"
+.sp
+.B Stage
+(i.e. return to online) the previously-offline contents of a file by copying a
+region from another file, the archive, and without updating regular inode
+metadata.  Any operations that are blocked by the existence of an offline
+region will proceed once the region has been staged.
+.RS 1.0i
+.PD 0
+.TP
+.sp
+.B "ARCHIVE-FILE"
+The source file for the file contents being staged.
+.TP
+.B "FILE"
+The regular file whose contents will be staged.
+.TP
+.B "-V, --version VERSION"
+The data_version of the contents to be staged.  It must match the
+current data_version of the file.
+.TP
+.B "-o, --offset OFF-NUM"
+The starting byte offset of the region to write.  May be expressed in bytes, or with
+KMGTP (Kibi, Mibi, etc.) size suffixes. Default is 0.
+.TP
+.B "-l, --length LENGTH"
+Length of range (bytes or KMGTP units) of file to stage. Default is the file's
+total size.
+.RE
+.PD
+
+.TP
 .BI "stat FILE [-s|--single-field FIELD-NAME]"
 .sp
 Display ScoutFS-specific metadata fields for the given file.
@@ -314,221 +653,6 @@ The total number of 4K data blocks in the filesystem.
 .RE
 .PD
 
-.TP
-.BI "counters [-t|--table] SYSFS-DIR"
-.sp
-Display the counters and their values for a mounted ScoutFS filesystem.
-.RS 1.0i
-.PD 0
-.sp
-.TP
-.B SYSFS-DIR
-The mount's sysfs directory in which to find the
-.B counters/
-directory when then contains files for each counter.
-The sysfs directory is
-of the form
-.I /sys/fs/scoutfs/f.<fsid>.r.<rid>/
-\&.
-.TP
-.B "-t, --table"
-Format the counters into a columnar table that fills the width of the display
-instead of printing one counter per line.
-.RE
-.PD
-
-.TP
-.BI "search-xattrs XATTR-NAME [-p|--path PATH]"
-.sp
-Display the inode numbers of inodes in the filesystem which may have
-an extended attribute with the given name.
-.sp
-The results may contain false positives.  The returned inode numbers
-should be checked to verify that the extended attribute is in fact
-present on the inode.
-.RS 1.0i
-.PD 0
-.TP
-.sp
-.B XATTR-NAME
-The full name of the extended attribute to search for as
-described in the
-.BR xattr (7)
-manual page.
-.TP
-.B "-p|--path PATH"
-A path within a ScoutFS filesystem.
-.RE
-.PD
-
-.TP
-.BI "list-hidden-xattrs FILE"
-.sp
-Display extended attributes starting with the
-.BR scoutfs.
-prefix and containing the
-.BR hide.
-tag
-which makes them invisible to
-.BR listxattr (2) .
-The names of each attribute are output, one per line.  Their order
-is not specified.
-.RS 1.0i
-.PD 0
-.TP
-.sp
-.B "FILE"
-The path to a file within a ScoutFS filesystem.  File permissions must allow
-reading.
-.RE
-.PD
-
-.TP
-.BI "walk-inodes {meta_seq|data_seq} FIRST-INODE LAST-INODE [-p|--path PATH]"
-.sp
-Walk an inode index in the file system and output the inode numbers
-that are found between the first and last positions in the index.
-.RS 1.0i
-.PD 0
-.sp
-.TP
-.BR meta_seq , data_seq
-Which index to walk.
-.TP
-.B "FIRST-INODE"
-An integer index value giving starting position of the index walk.
-.I 0
-is the first possible position.
-.TP
-.B "LAST-INODE"
-An integer index value giving the last position to include in the index walk.
-.I \-1
-can be given to indicate the last possible position.
-.TP
-.B "-p|--path PATH"
-A path within a ScoutFS filesystem.
-.RE
-.PD
-
-.TP
-.BI "ino-path INODE-NUM [-p|--path PATH]"
-.sp
-Display all paths that reference an inode number.
-.sp
-Ongoing filesystem changes, such as renaming a common parent of multiple paths,
-can cause displayed paths to be inconsistent.
-.RS 1.0i
-.PD 0
-.sp
-.TP
-.B "INODE-NUM"
-The inode number of the target inode.
-.TP
-.B "-p|--path PATH"
-A path within a ScoutFS filesystem.
-.RE
-.PD
-
-.TP
-.BI "data-waiting {-I|--inode} INODE-NUM {-B|--block} BLOCK-NUM [-p|--path PATH]"
-.sp
-Display all the files and blocks for which there is a task blocked waiting on
-offline data.
-.sp
-The results are sorted by the file's inode number and the
-logical block offset that is being waited on.
-.sp
-Each line of output describes a block in a file that has a task waiting
-and is formatted as:
-.I "ino <nr> iblock <nr> ops [str]"
-\&. The ops string indicates blocked operations seperated by commas and can
-include
-.B read
-for a read operation,
-.B write
-for a write operation, and
-.B change_size
-for a truncate or extending write.
-.RS 1.0i
-.PD 0
-.sp
-.TP
-.B "-I, --inode INODE-NUM"
-Start iterating over waiting tasks from the given inode number.
-Value of 0 will show all waiting tasks.
-.TP
-.B "-B, --block BLOCK-NUM"
-Start iterating over waiting tasks from the given logical block number
-in the starting inode.  Value of 0 will show blocks in the first inode
-and then continue to show all blocks with tasks waiting in all the
-remaining inodes.
-.TP
-.B "-p, --path PATH"
-A path within a ScoutFS filesystem.
-.RE
-.PD
-
-.TP
-.BI "data-wait-err {-I|--inode} INODE-NUM {-V|--version} VER-NUM {-F|--offset} OFF-NUM {-C|--count} COUNT {-O|--op} OP {-E|--err} ERR [-p|--path PATH]"
-.sp
-Return error from matching waiters.
-.RS 1.0i
-.PD 0
-.sp
-.TP
-.B "-C, --count COUNT"
-Count.
-.TP
-.B "-E, --err ERR"
-Error.
-.TP
-.B "-F, --offset OFF-NUM"
-Offset. May be expressed in bytes, or with KMGTP (Kibi, Mibi, etc.) size
-suffixes.
-.TP
-.B "-I, --inode INODE-NUM"
-Inode number.
-.TP
-.B "-O, --op OP"
-Operation. One of: "read", "write", "change_size".
-.TP
-.B "-p, --path PATH"
-A path within a ScoutFS filesystem.
-.RE
-.PD
-
-.TP
-.BI "stage ARCHIVE-FILE FILE {-V|--version} VERSION [-o, --offset OFF-NUM] [-l, --length LENGTH]"
-.sp
-.B Stage
-(i.e. return to online) the previously-offline contents of a file by copying a
-region from another file, the archive, and without updating regular inode
-metadata.  Any operations that are blocked by the existence of an offline
-region will proceed once the region has been staged.
-.RS 1.0i
-.PD 0
-.TP
-.sp
-.B "ARCHIVE-FILE"
-The source file for the file contents being staged.
-.TP
-.B "FILE"
-The regular file whose contents will be staged.
-.TP
-.B "-V, --version VERSION"
-The data_version of the contents to be staged.  It must match the
-current data_version of the file.
-.TP
-.B "-o, --offset OFF-NUM"
-The starting byte offset of the region to write.  May be expressed in bytes, or with
-KMGTP (Kibi, Mibi, etc.) size suffixes. Default is 0.
-.TP
-.B "-l, --length LENGTH"
-Length of range (bytes or KMGTP units) of file to stage. Default is the file's
-total size.
-.RE
-.PD
-
 .TP
 .BI "release FILE {-V|--version} VERSION [-o, --offset OFF-NUM] [-l, --length LENGTH]"
 .sp
@@ -568,88 +692,28 @@ total size.
 .PD
 
 .TP
-.BI "setattr FILE [-d, --data-version=VERSION [-s, --size=SIZE [-o, --offline]]] [-t, --ctime=TIMESPEC]"
+.BI "walk-inodes {meta_seq|data_seq} FIRST-INODE LAST-INODE [-p|--path PATH]"
 .sp
-Set ScoutFS-specific attributes on a newly created zero-length file.
+Walk an inode index in the file system and output the inode numbers
+that are found between the first and last positions in the index.
 .RS 1.0i
 .PD 0
 .sp
 .TP
-.B "-V, --data-version=VERSION"
-Set data version.
+.BR meta_seq , data_seq
+Which index to walk.
 .TP
-.B "-o, --offline"
-Set file contents as offline, not sparse. Requires
-.I --size
-option also be present.
+.B "FIRST-INODE"
+An integer index value giving starting position of the index walk.
+.I 0
+is the first possible position.
 .TP
-.B "-s, --size=SIZE"
-Set file size. May be expressed in bytes, or with
-KMGTP (Kibi, Mibi, etc.) size suffixes. Requires
-.I --data-version
-option also be present.
+.B "LAST-INODE"
+An integer index value giving the last position to include in the index walk.
+.I \-1
+can be given to indicate the last possible position.
 .TP
-.B "-t, --ctime=TIMESPEC"
-Set creation time using
-.I "<seconds-since-epoch>.<nanoseconds>"
-format.
-.RE
-.PD
-
-.TP
-.BI "print {-S|--skip-likely-huge} META-DEVICE"
-.sp
-Prints out all of the metadata in the file system.  This makes no effort
-to ensure that the structures are consistent as they're traversed and
-can present structures that seem corrupt as they change as they're
-output.
-.RS 1.0i
-.PD 0
-.TP
-.sp
-.B "-S, --skip-likely-huge"
-Skip printing structures that are likely to be very large.  The
-structures that are skipped tend to be global and whose size tends to be
-related to the size of the volume.   Examples of skipped structures include
-the global fs items, srch files, and metadata and data
-allocators.  Similar structures that are not skipped are related to the
-number of mounts and are maintained at a relatively reasonable size.
-These include per-mount log trees, srch files, allocators, and the
-metadata allocators used by server commits.
-.sp
-Skipping the larger structures limits the print output to a relatively
-constant size rather than being a large multiple of the used metadata
-space of the volume making the output much more useful for inspection.
-.TP
-.B "META-DEVICE"
-The path to the metadata device for the filesystem whose metadata will be
-printed.  An attempt will be made to flush the host's buffer cache for
-this device with the BLKFLSBUF ioctl, or with posix_fadvise() if
-the path refers to a regular file.
-.RE
-.PD
-
-.TP
-.BI "get-allocated-inos [-i|--ino INO] [-s|--single] [-p|--path PATH]"
-.sp
-This debugging command prints allocated inode numbers.  It only prints
-inodes
-found in the group that contains the starting inode.  The printed inode
-numbers aren't necessarily reachable.  They could be anywhere in the
-process from being unlinked to finally deleted when their items
-were found.
-.RS 1.0i
-.PD 0
-.TP
-.sp
-.B "-i, --ino INO"
-The first 64bit inode number which could be printed.
-.TP
-.B "-s, --single"
-Only print the single starting inode when it is allocated, all other allocated
-inode numbers will be ignored.
-.TP
-.B "-p, --path PATH"
+.B "-p|--path PATH"
 A path within a ScoutFS filesystem.
 .RE
 .PD

utils/scoutfs-utils.spec.in

@@ -61,7 +61,7 @@ install -m 644 -D fenced/scoutfs-fenced.conf.example $RPM_BUILD_ROOT%{_sysconfdi
 %files
 %defattr(644,root,root,755)
 %{_mandir}/man*/scoutfs*.gz
-%{_unitdir}/scoutfs-fenced.service
+/%{_unitdir}/scoutfs-fenced.service
 %{_sysconfdir}/scoutfs
 %defattr(755,root,root,755)
 %{_sbindir}/scoutfs

utils/src/dev.c

@@ -12,13 +12,10 @@
 #include "sparse.h"
 #include "dev.h"
 
-int device_size(char *path, int fd,
-		u64 min_size, u64 max_size, bool allow_small_size,
-		char *use_type, u64 *size_ret)
+int get_device_size(char *path, int fd, u64 *size_ret)
 {
 	struct stat st;
 	u64 size;
-	char *target_type;
 	int ret;
 
 	if (fstat(fd, &st)) {
@@ -30,7 +27,6 @@ int device_size(char *path, int fd,
 
 	if (S_ISREG(st.st_mode)) {
 		size = st.st_size;
-		target_type = "file";
 	} else if (S_ISBLK(st.st_mode)) {
 		if (ioctl(fd, BLKGETSIZE64, &size)) {
 			ret = -errno;
@@ -38,13 +34,26 @@ int device_size(char *path, int fd,
 				path, strerror(errno), errno);
 			return ret;
 		}
-		target_type = "device";
 	} else {
 		fprintf(stderr, "path isn't regular or device file '%s'\n",
 			path);
 		return -EINVAL;
 	}
 
+	*size_ret = size;
+	return 0;
+}
+
+int limit_device_size(char *path, int fd, u64 min_size, u64 max_size, bool allow_small_size,
+		      char *use_type, u64 *size_ret)
+{
+	u64 size;
+	int ret;
+
+	ret = get_device_size(path, fd, &size);
+	if (ret < 0)
+		return ret;
+
 	if (max_size) {
 		if (size > max_size) {
 			printf("Limiting use of "BASE_SIZE_FMT
@@ -64,9 +73,9 @@ int device_size(char *path, int fd,
 
 	if (size < min_size) {
 		fprintf(stderr,
-			BASE_SIZE_FMT" %s too small for min "
+			BASE_SIZE_FMT" too small for min "
 			BASE_SIZE_FMT" %s device%s\n",
-			BASE_SIZE_ARGS(size), target_type,
+			BASE_SIZE_ARGS(size),
 			BASE_SIZE_ARGS(min_size), use_type,
 			allow_small_size ? ", allowing with -A" : "");
 

utils/src/dev.h

@@ -9,9 +9,9 @@
 #define SIZE_FMT "%llu (%.2f %s)"
 #define SIZE_ARGS(nr, sz) (nr), size_flt(nr, sz), size_str(nr, sz)
 
-int device_size(char *path, int fd,
-		u64 min_size, u64 max_size, bool allow_small_size,
-		char *use_type, u64 *size_ret);
+int get_device_size(char *path, int fd, u64 *size_ret);
+int limit_device_size(char *path, int fd, u64 min_size, u64 max_size, bool allow_small_size,
+		      char *use_type, u64 *size_ret);
 float size_flt(u64 nr, unsigned size);
 char *size_str(u64 nr, unsigned size);
 int flush_device(int fd);

utils/src/get_referring_entries.c

@@ -0,0 +1,150 @@
+#include <unistd.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/ioctl.h>
+#include <fcntl.h>
+#include <errno.h>
+#include <string.h>
+#include <limits.h>
+#include <argp.h>
+
+#include "sparse.h"
+#include "parse.h"
+#include "util.h"
+#include "format.h"
+#include "ioctl.h"
+#include "parse.h"
+#include "cmd.h"
+
+struct gre_args {
+	char *path;
+	u64 ino;
+};
+
+static int do_get_referring_entries(struct gre_args *args)
+{
+	struct scoutfs_ioctl_get_referring_entries gre;
+	struct scoutfs_ioctl_dirent *dent;
+	unsigned int bytes;
+	void *buf;
+	int ret;
+	int fd;
+
+	fd = get_path(args->path, O_RDONLY);
+	if (fd < 0)
+		return fd;
+
+	bytes = PATH_MAX * 1024;
+	buf = malloc(bytes);
+	if (!buf) {
+		fprintf(stderr, "couldn't allocate %u byte buffer\n", bytes);
+		ret = -ENOMEM;
+		goto out;
+	}
+
+	gre.ino = args->ino;
+	gre.dir_ino = 0;
+	gre.dir_pos = 0;
+	gre.entries_ptr = (intptr_t)buf;
+	gre.entries_bytes = bytes;
+
+	for (;;) {
+		ret = ioctl(fd, SCOUTFS_IOC_GET_REFERRING_ENTRIES, &gre);
+		if (ret <= 0) {
+			if (ret < 0) {
+				ret = -errno;
+				fprintf(stderr, "ioctl failed: %s (%d)\n", strerror(errno), errno);
+			}
+			goto out;
+		}
+
+		dent = buf;
+		while (ret-- > 0) {
+			printf("dir %llu pos %llu type %u name %s\n",
+			       dent->dir_ino, dent->dir_pos, dent->d_type, dent->name);
+
+			gre.dir_ino = dent->dir_ino;
+			gre.dir_pos = dent->dir_pos;
+
+			if (dent->flags & SCOUTFS_IOCTL_DIRENT_FLAG_LAST) {
+				ret = 0;
+				goto out;
+			}
+
+			dent = (void *)dent + dent->entry_bytes;
+		}
+
+		if (++gre.dir_pos == 0) {
+			if (++gre.dir_ino == 0) {
+				ret = 0;
+				goto out;
+			}
+		}
+	}
+
+out:
+	close(fd);
+	free(buf);
+
+	return ret;
+};
+
+static int parse_opt(int key, char *arg, struct argp_state *state)
+{
+	struct gre_args *args = state->input;
+	int ret;
+
+	switch (key) {
+	case 'p':
+		args->path = strdup_or_error(state, arg);
+		break;
+	case ARGP_KEY_ARG:
+		if (args->ino)
+			argp_error(state, "more than one argument given");
+		ret = parse_u64(arg, &args->ino);
+		if (ret)
+			argp_error(state, "inode parse error");
+		break;
+	case ARGP_KEY_FINI:
+		if (!args->ino) {
+			argp_error(state, "must provide inode number");
+		}
+		break;
+	default:
+		break;
+	}
+
+	return 0;
+}
+
+static struct argp_option options[] = {
+	{ "path", 'p', "PATH", 0, "Path to ScoutFS filesystem"},
+	{ NULL }
+};
+
+static struct argp argp = {
+	options,
+	parse_opt,
+	"INODE-NUM",
+	"Print directory entries that refer to inode number"
+};
+
+static int get_referring_entries_cmd(int argc, char **argv)
+{
+	struct gre_args args = {NULL};
+	int ret;
+
+	ret = argp_parse(&argp, argc, argv, 0, NULL, &args);
+	if (ret)
+		return ret;
+
+	return do_get_referring_entries(&args);
+}
+
+
+static void __attribute__((constructor)) get_referring_entries_ctor(void)
+{
+	cmd_register_argp("get-referring-entries", &argp, GROUP_SEARCH, get_referring_entries_cmd);
+}

utils/src/mkfs.c

@@ -206,14 +206,14 @@ static int do_mkfs(struct mkfs_args *args)
 	}
 
 	/* minumum meta device size to make reserved blocks reasonably large */
-	ret = device_size(args->meta_device, meta_fd, 64ULL * (1024 * 1024 * 1024),
-			  args->max_meta_size, args->allow_small_size, "meta", &meta_size);
+	ret = limit_device_size(args->meta_device, meta_fd, 64ULL * (1024 * 1024 * 1024),
+				args->max_meta_size, args->allow_small_size, "meta", &meta_size);
 	if (ret)
 		goto out;
 
 	/* .. then arbitrarily the same minimum data device size */
-	ret = device_size(args->data_device, data_fd, 64ULL * (1024 * 1024 * 1024),
-			  args->max_data_size, args->allow_small_size, "data", &data_size);
+	ret = limit_device_size(args->data_device, data_fd, 64ULL * (1024 * 1024 * 1024),
+				args->max_data_size, args->allow_small_size, "data", &data_size);
 	if (ret)
 		goto out;
 

utils/src/prepare_empty_data_device.c

@@ -0,0 +1,247 @@
+#define _GNU_SOURCE /* O_DIRECT */
+#include <unistd.h>
+#include <stdbool.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+#include <errno.h>
+#include <sys/time.h>
+#include <uuid/uuid.h>
+#include <fcntl.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <unistd.h>
+#include <assert.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <ctype.h>
+#include <inttypes.h>
+#include <argp.h>
+
+#include "sparse.h"
+#include "cmd.h"
+#include "util.h"
+#include "format.h"
+#include "parse.h"
+#include "crc.h"
+#include "rand.h"
+#include "dev.h"
+#include "key.h"
+#include "bitops.h"
+#include "btree.h"
+#include "leaf_item_hash.h"
+#include "blkid.h"
+#include "quorum.h"
+
+struct prepare_empty_data_dev_args {
+	char *meta_device;
+	char *data_device;
+	bool check;
+};
+
+static int do_prepare_empty_data_dev(struct prepare_empty_data_dev_args *args)
+{
+	struct scoutfs_super_block *meta_super = NULL;
+	struct scoutfs_super_block *data_super = NULL;
+	char uuid_str[37];
+	int meta_fd = -1;
+	int data_fd = -1;
+	u64 data_blocks;
+	u64 data_size;
+	u64 in_use;
+	int ret;
+
+	ret = posix_memalign((void **)&data_super, SCOUTFS_BLOCK_SM_SIZE, SCOUTFS_BLOCK_SM_SIZE);
+	if (ret < 0) {
+		ret = -errno;
+		fprintf(stderr, "failed to allocate data super block: %s (%d)\n",
+			strerror(errno), errno);
+		goto out;
+	}
+
+	meta_fd = open(args->meta_device, O_DIRECT | O_SYNC | O_RDONLY | O_EXCL);
+	if (meta_fd < 0) {
+		ret = -errno;
+		fprintf(stderr, "failed to open meta device '%s': %s (%d)\n",
+			args->meta_device, strerror(errno), errno);
+		goto out;
+	}
+
+	ret = read_block_verify(meta_fd, SCOUTFS_BLOCK_MAGIC_SUPER, 0, SCOUTFS_SUPER_BLKNO,
+				SCOUTFS_BLOCK_SM_SHIFT, (void **)&meta_super);
+	if (ret) {
+		ret = -errno;
+		fprintf(stderr, "failed to read meta super block: %s (%d)\n",
+			strerror(errno), errno);
+		goto out;
+	}
+
+	ret = meta_super_in_use(meta_fd, meta_super);
+	if (ret < 0) {
+		if (ret == -EBUSY)
+			fprintf(stderr, "The filesystem must be fully recovered and cleanly unmounted to determine if the data device is empty.\n");
+		goto out;
+	}
+
+	in_use = (le64_to_cpu(meta_super->total_data_blocks) - SCOUTFS_DATA_DEV_START_BLKNO) -
+		 le64_to_cpu(meta_super->data_alloc.total_len);
+	if (in_use) {
+		fprintf(stderr, "Data block allocator metadata shows "SIZE_FMT" data blocks used by files.  They must be removed, truncated, or released before a new empty data device can be used.\n",
+		       SIZE_ARGS(in_use, SCOUTFS_BLOCK_SM_SIZE));
+		ret = -EINVAL;
+		goto out;
+	}
+
+	if (args->data_device) {
+		data_fd = open(args->data_device, O_DIRECT | O_EXCL |
+						  (args->check ? O_RDONLY : O_RDWR | O_SYNC));
+		if (data_fd < 0) {
+			ret = -errno;
+			fprintf(stderr, "failed to open data device '%s': %s (%d)\n",
+				args->data_device, strerror(errno), errno);
+			goto out;
+		}
+
+		ret = get_device_size(args->data_device, data_fd, &data_size);
+		if (ret < 0)
+			goto out;
+
+		data_blocks = data_size >> SCOUTFS_BLOCK_SM_SHIFT;
+
+		if (data_blocks < le64_to_cpu(meta_super->total_data_blocks)) {
+			fprintf(stderr, "new data device %s of size "BASE_SIZE_FMT" has %llu 4KiB blocks, it needs at least "SIZE_FMT" blocks.\n",
+			       args->data_device,
+			       BASE_SIZE_ARGS(data_size),
+			       data_blocks,
+			       SIZE_ARGS(le64_to_cpu(meta_super->total_data_blocks),
+					  SCOUTFS_BLOCK_SM_SIZE));
+			ret = -EINVAL;
+			goto out;
+		}
+	}
+
+	if (args->check) {
+		ret = 0;
+		goto out;
+	}
+
+	/* the data device superblock only needs fs identifying fields */
+	memset(data_super, 0, sizeof(struct scoutfs_super_block));
+	data_super->id = meta_super->id;
+	data_super->fmt_vers = meta_super->fmt_vers;
+	data_super->flags = meta_super->flags &~ cpu_to_le64(SCOUTFS_FLAG_IS_META_BDEV);
+	memcpy(data_super->uuid, meta_super->uuid,sizeof(data_super->uuid));
+	data_super->seq = meta_super->seq;
+	data_super->total_meta_blocks = meta_super->total_meta_blocks;
+	data_super->total_data_blocks = meta_super->total_data_blocks;
+
+	ret = write_block(data_fd, SCOUTFS_BLOCK_MAGIC_SUPER, meta_super->hdr.fsid, 1,
+			  SCOUTFS_SUPER_BLKNO, SCOUTFS_BLOCK_SM_SHIFT, &data_super->hdr);
+	if (ret < 0) {
+		ret = -errno;
+		fprintf(stderr, "Error writing super block to new data device '%s': %s (%d)\n",
+			args->data_device, strerror(errno), errno);
+		goto out;
+	}
+
+	uuid_unparse(meta_super->uuid, uuid_str);
+
+	printf("Successfully initialized empty data device for scoutfs filesystem:\n"
+	       "  meta device path:       %s\n"
+	       "  data device path:       %s\n"
+	       "  fsid:                   %llx\n"
+	       "  uuid:                   %s\n"
+	       "  format version:         %llu\n"
+	       "  64KB metadata blocks:   "SIZE_FMT"\n"
+	       "  4KB data blocks:        "SIZE_FMT"\n",
+		args->meta_device,
+	        args->data_device,
+		le64_to_cpu(meta_super->hdr.fsid),
+		uuid_str,
+		le64_to_cpu(meta_super->fmt_vers),
+		SIZE_ARGS(le64_to_cpu(meta_super->total_meta_blocks),
+			  SCOUTFS_BLOCK_LG_SIZE),
+		SIZE_ARGS(le64_to_cpu(meta_super->total_data_blocks),
+			  SCOUTFS_BLOCK_SM_SIZE));
+
+	ret = 0;
+out:
+	if (args->check) {
+		if (ret == 0)
+			printf("All checks passed.\n");
+		else
+			printf("Errors were found that must be addressed before a new empty data device could be prepared and used.\n");
+	}
+
+	if (meta_super)
+		free(meta_super);
+	if (data_super)
+		free(data_super);
+	if (meta_fd != -1)
+		close(meta_fd);
+	if (data_fd != -1)
+		close(data_fd);
+	return ret;
+}
+
+static int parse_opt(int key, char *arg, struct argp_state *state)
+{
+	struct prepare_empty_data_dev_args *args = state->input;
+
+	switch (key) {
+	case 'c':
+		args->check = true;
+		break;
+	case ARGP_KEY_ARG:
+		if (!args->meta_device)
+			args->meta_device = strdup_or_error(state, arg);
+		else if (!args->data_device)
+			args->data_device = strdup_or_error(state, arg);
+		else
+			argp_error(state, "more than two device arguments given");
+		break;
+	case ARGP_KEY_FINI:
+		if (!args->meta_device)
+			argp_error(state, "no metadata device argument given");
+		if (!args->data_device && !args->check)
+			argp_error(state, "no data device argument given");
+		break;
+	default:
+		break;
+	}
+
+	return 0;
+}
+
+static struct argp_option options[] = {
+	{ "check", 'c', NULL, 0, "Only check for errors and do not write", },
+	{ NULL }
+};
+
+static struct argp argp = {
+	options,
+	parse_opt,
+	"META-DEVICE DATA-DEVICE",
+	"Prepare empty data device for use with an existing ScoutFS filesystem"
+};
+
+static int prepare_empty_data_dev_cmd(int argc, char *argv[])
+{
+	struct prepare_empty_data_dev_args prepare_empty_data_dev_args = { 
+		.check = false,
+	};
+	int ret;
+
+	ret = argp_parse(&argp, argc, argv, 0, NULL, &prepare_empty_data_dev_args);
+	if (ret)
+		return ret;
+
+	return do_prepare_empty_data_dev(&prepare_empty_data_dev_args);
+}
+
+static void __attribute__((constructor)) prepare_empty_data_dev_ctor(void)
+{
+	cmd_register_argp("prepare-empty-data-device", &argp, GROUP_CORE,
+			  prepare_empty_data_dev_cmd);
+}