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,34 +1,6 @@
 Versity ScoutFS Release Notes
 =============================
 
----
-v1.18
-\
-*Nov 7, 2023*
-
-Fixed a bug where background srch file compaction could stop making
-forward progress if a partial compaction operation was committed at a
-specific byte offset in a block.  This would cause srch file searches to
-be progressively more expensive over time.  Once this fix is running
-background compaction will resume, bringing the cost of searches back
-down.
-
----
-v1.17
-\
-*Oct 23, 2023*
-
-Add support for EL8 generation kernels.
-
----
-v1.16
-\
-*Oct 4, 2023*
-
-Fix an issue where the server could hang on startup if its persistent
-allocator structures were left in a specific degraded state by the
-previously active server.
-
 ---
 v1.15
 \

kmod/src/block.c

@@ -1078,7 +1078,7 @@ static unsigned long block_count_objects(struct shrinker *shrink, struct shrink_
 
 	scoutfs_inc_counter(sb, block_cache_count_objects);
 
-	return shrinker_min_long(atomic_read(&binf->total_inserted));
+	return shrinker_min_t_long((u64)atomic_read(&binf->total_inserted));
 }
 
 /*

kmod/src/dir.c

@@ -1058,15 +1058,16 @@ static int symlink_item_ops(struct super_block *sb, enum symlink_ops op, u64 ino
 	return ret;
 }
 
+#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
 /*
- * Fill a buffer with the null terminated symlink, and return it
- * so callers can free it once the vfs is done.
+ * 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.
  *
  * We chose to pay the runtime cost of per-call allocation and copy
  * overhead instead of wiring up symlinks to the page cache, storing
  * each small link in a full page, and later having to reclaim them.
  */
-static void *scoutfs_get_link_target(struct dentry *dentry)
+static void *scoutfs_follow_link(struct dentry *dentry, struct nameidata *nd)
 {
 	struct inode *inode = dentry->d_inode;
 	struct super_block *sb = inode->i_sb;
@@ -1125,20 +1126,10 @@ out:
 	if (ret < 0) {
 		kfree(path);
 		path = ERR_PTR(ret);
-	}
-
-	scoutfs_unlock(sb, inode_lock, SCOUTFS_LOCK_READ);
-	return path;
-}
-
-#ifdef KC_LINUX_HAVE_RHEL_IOPS_WRAPPER
-static void *scoutfs_follow_link(struct dentry *dentry, struct nameidata *nd)
-{
-	char *path;
-
-	path = scoutfs_get_link_target(dentry);
-	if (!IS_ERR_OR_NULL(path))
+	} else {
 		nd_set_link(nd, path);
+	}
+	scoutfs_unlock(sb, inode_lock, SCOUTFS_LOCK_READ);
 	return path;
 }
 
@@ -1151,12 +1142,67 @@ static void scoutfs_put_link(struct dentry *dentry, struct nameidata *nd,
 #else
 static const char *scoutfs_get_link(struct dentry *dentry, struct inode *inode, struct delayed_call *done)
 {
-	char *path;
+	struct super_block *sb = inode->i_sb;
+	struct scoutfs_lock *inode_lock = NULL;
+	char *path = NULL;
+	loff_t size;
+	int ret;
 
-	path = scoutfs_get_link_target(dentry);
-	if (!IS_ERR_OR_NULL(path))
+	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

kmod/src/item.c

@@ -2541,7 +2541,7 @@ static unsigned long item_cache_count_objects(struct shrinker *shrink,
 
 	scoutfs_inc_counter(sb, item_cache_count_objects);
 
-	return shrinker_min_long(cinf->lru_pages);
+	return shrinker_min_t_long((u64)(cinf->lru_pages));
 }
 
 /*

kmod/src/lock.c

@@ -1409,7 +1409,7 @@ static unsigned long lock_count_objects(struct shrinker *shrink,
 
 	scoutfs_inc_counter(sb, lock_count_objects);
 
-	return shrinker_min_long(linfo->lru_nr);
+	return shrinker_min_t_long((u64)(linfo->lru_nr));
 }
 
 /*

kmod/src/scoutfs_trace.h

@@ -1896,9 +1896,8 @@ DEFINE_EVENT(scoutfs_server_client_count_class, scoutfs_server_client_down,
 
 DECLARE_EVENT_CLASS(scoutfs_server_commit_users_class,
         TP_PROTO(struct super_block *sb, int holding, int applying, int nr_holders,
-		 u32 avail_before, u32 freed_before, int committing, int exceeded),
-        TP_ARGS(sb, holding, applying, nr_holders, avail_before, freed_before, committing,
-		exceeded),
+		 u32 avail_before, u32 freed_before, int exceeded),
+        TP_ARGS(sb, holding, applying, nr_holders, avail_before, freed_before, exceeded),
         TP_STRUCT__entry(
 		SCSB_TRACE_FIELDS
 		__field(int, holding)
@@ -1906,7 +1905,6 @@ DECLARE_EVENT_CLASS(scoutfs_server_commit_users_class,
 		__field(int, nr_holders)
 		__field(__u32, avail_before)
 		__field(__u32, freed_before)
-		__field(int, committing)
 		__field(int, exceeded)
         ),
         TP_fast_assign(
@@ -1916,33 +1914,31 @@ DECLARE_EVENT_CLASS(scoutfs_server_commit_users_class,
 		__entry->nr_holders = nr_holders;
 		__entry->avail_before = avail_before;
 		__entry->freed_before = freed_before;
-		__entry->committing = !!committing;
 		__entry->exceeded = !!exceeded;
         ),
-	TP_printk(SCSBF" holding %u applying %u nr %u avail_before %u freed_before %u committing %u exceeded %u",
+	TP_printk(SCSBF" holding %u applying %u nr %u avail_before %u freed_before %u exceeded %u",
 		  SCSB_TRACE_ARGS, __entry->holding, __entry->applying, __entry->nr_holders,
-		  __entry->avail_before, __entry->freed_before, __entry->committing,
-		  __entry->exceeded)
+		  __entry->avail_before, __entry->freed_before, __entry->exceeded)
 );
 DEFINE_EVENT(scoutfs_server_commit_users_class, scoutfs_server_commit_hold,
         TP_PROTO(struct super_block *sb, int holding, int applying, int nr_holders,
-		 u32 avail_before, u32 freed_before, int committing, int exceeded),
-        TP_ARGS(sb, holding, applying, nr_holders, avail_before, freed_before, committing, exceeded)
+		 u32 avail_before, u32 freed_before, int exceeded),
+        TP_ARGS(sb, holding, applying, nr_holders, avail_before, freed_before, exceeded)
 );
 DEFINE_EVENT(scoutfs_server_commit_users_class, scoutfs_server_commit_apply,
         TP_PROTO(struct super_block *sb, int holding, int applying, int nr_holders,
-		 u32 avail_before, u32 freed_before, int committing, int exceeded),
-        TP_ARGS(sb, holding, applying, nr_holders, avail_before, freed_before, committing, exceeded)
+		 u32 avail_before, u32 freed_before, int exceeded),
+        TP_ARGS(sb, holding, applying, nr_holders, avail_before, freed_before, exceeded)
 );
 DEFINE_EVENT(scoutfs_server_commit_users_class, scoutfs_server_commit_start,
         TP_PROTO(struct super_block *sb, int holding, int applying, int nr_holders,
-		 u32 avail_before, u32 freed_before, int committing, int exceeded),
-        TP_ARGS(sb, holding, applying, nr_holders, avail_before, freed_before, committing, exceeded)
+		 u32 avail_before, u32 freed_before, int exceeded),
+        TP_ARGS(sb, holding, applying, nr_holders, avail_before, freed_before, exceeded)
 );
 DEFINE_EVENT(scoutfs_server_commit_users_class, scoutfs_server_commit_end,
         TP_PROTO(struct super_block *sb, int holding, int applying, int nr_holders,
-		 u32 avail_before, u32 freed_before, int committing, int exceeded),
-        TP_ARGS(sb, holding, applying, nr_holders, avail_before, freed_before, committing, exceeded)
+		 u32 avail_before, u32 freed_before, int exceeded),
+        TP_ARGS(sb, holding, applying, nr_holders, avail_before, freed_before, exceeded)
 );
 
 #define slt_symbolic(mode)						\

kmod/src/server.c

@@ -67,7 +67,6 @@ struct commit_users {
 	unsigned int nr_holders;
 	u32 avail_before;
 	u32 freed_before;
-	bool committing;
 	bool exceeded;
 };
 
@@ -85,13 +84,12 @@ do {												\
 	__typeof__(cusers) _cusers = (cusers);							\
 	trace_scoutfs_server_commit_##which(sb, !list_empty(&_cusers->holding),			\
 		!list_empty(&_cusers->applying), _cusers->nr_holders, _cusers->avail_before,	\
-		_cusers->freed_before, _cusers->committing, _cusers->exceeded);			\
+		_cusers->freed_before, _cusers->exceeded);					\
 } while (0)
 
 struct server_info {
 	struct super_block *sb;
 	spinlock_t lock;
-	seqlock_t seqlock;
 	wait_queue_head_t waitq;
 
 	struct workqueue_struct *wq;
@@ -133,9 +131,11 @@ struct server_info {
 	struct mutex mounted_clients_mutex;
 
 	/* stable super stored from commits, given in locks and rpcs */
+	seqcount_t stable_seqcount;
 	struct scoutfs_super_block stable_super;
 
 	/* serializing and get and set volume options */
+	seqcount_t volopt_seqcount;
 	struct mutex volopt_mutex;
 	struct scoutfs_volume_options volopt;
 
@@ -181,7 +181,7 @@ static bool get_volopt_val(struct server_info *server, int nr, u64 *val)
 	unsigned seq;
 
 	do {
-		seq = read_seqbegin(&server->seqlock);
+		seq = read_seqcount_begin(&server->volopt_seqcount);
 		if ((le64_to_cpu(server->volopt.set_bits) & bit)) {
 			is_set = true;
 			*val = le64_to_cpup(opt);
@@ -189,7 +189,7 @@ static bool get_volopt_val(struct server_info *server, int nr, u64 *val)
 			is_set = false;
 			*val = 0;
 		};
-	} while (read_seqretry(&server->seqlock, seq));
+	} while (read_seqcount_retry(&server->volopt_seqcount, seq));
 
 	return is_set;
 }
@@ -282,14 +282,6 @@ struct commit_hold {
  * per-holder allocation consumption tracking.   The best we can do is
  * flag all the current holders so that as they release we can see
  * everyone involved in crossing the limit.
- *
- * The consumption of space to record freed blocks is tricky.  The
- * freed_before value was the space available as the holder started.
- * But that happens before we actually dirty the first block in the
- * freed list.  If that block is too full then we just allocate a new
- * empty first block.  In that case the current remaining here can be a
- * lot more than the initial freed_before.  We account for that and
- * treat freed_before as the maximum capacity.
  */
 static void check_holder_budget(struct super_block *sb, struct server_info *server,
 				struct commit_users *cusers)
@@ -309,13 +301,8 @@ static void check_holder_budget(struct super_block *sb, struct server_info *serv
 		return;
 
 	scoutfs_alloc_meta_remaining(&server->alloc, &avail_now, &freed_now);
-
 	avail_used = cusers->avail_before - avail_now;
-	if (freed_now < cusers->freed_before)
-		freed_used = cusers->freed_before - freed_now;
-	else
-		freed_used = SCOUTFS_ALLOC_LIST_MAX_BLOCKS - freed_now;
-
+	freed_used = cusers->freed_before - freed_now;
 	budget = cusers->nr_holders * COMMIT_HOLD_ALLOC_BUDGET;
 	if (avail_used <= budget && freed_used <= budget)
 		return;
@@ -338,18 +325,31 @@ static void check_holder_budget(struct super_block *sb, struct server_info *serv
 /*
  * We don't have per-holder consumption.   We allow commit holders as
  * long as the total budget of all the holders doesn't exceed the alloc
- * resources that were available.  If a hold is waiting for budget
- * availability in the allocators then we try and kick off a commit to
- * fill and use the next allocators after the current transaction.
+ * resources that were available
  */
+static bool commit_alloc_has_room(struct server_info *server, struct commit_users *cusers,
+				  unsigned int more_holders)
+{
+	u32 avail_before;
+	u32 freed_before;
+	u32 budget;
+
+	if (cusers->nr_holders > 0) {
+		avail_before = cusers->avail_before;
+		freed_before = cusers->freed_before;
+	} else {
+		scoutfs_alloc_meta_remaining(&server->alloc, &avail_before, &freed_before);
+	}
+
+	budget = (cusers->nr_holders + more_holders) * COMMIT_HOLD_ALLOC_BUDGET;
+
+	return avail_before >= budget && freed_before >= budget;
+}
+
 static bool hold_commit(struct super_block *sb, struct server_info *server,
 			struct commit_users *cusers, struct commit_hold *hold)
 {
-	bool has_room;
-	bool held;
-	u32 budget;
-	u32 av;
-	u32 fr;
+	bool held = false;
 
 	spin_lock(&cusers->lock);
 
@@ -357,39 +357,19 @@ static bool hold_commit(struct super_block *sb, struct server_info *server,
 
 	check_holder_budget(sb, server, cusers);
 
-	if (cusers->nr_holders == 0) {
-		scoutfs_alloc_meta_remaining(&server->alloc, &av, &fr);
-	} else {
-		av = cusers->avail_before;
-		fr = cusers->freed_before;
-	}
-
 	/* +2 for our additional hold and then for the final commit work the server does */
-	budget = (cusers->nr_holders + 2) * COMMIT_HOLD_ALLOC_BUDGET;
-	has_room = av >= budget && fr >= budget;
-	/* checking applying so holders drain once an apply caller starts waiting */
-	held = !cusers->committing && has_room && list_empty(&cusers->applying);
-
-	if (held) {
+	if (list_empty(&cusers->applying) && commit_alloc_has_room(server, cusers, 2)) {
+		scoutfs_alloc_meta_remaining(&server->alloc, &hold->avail, &hold->freed);
 		if (cusers->nr_holders == 0) {
-			cusers->avail_before = av;
-			cusers->freed_before = fr;
-			hold->avail = av;
-			hold->freed = fr;
+			cusers->avail_before = hold->avail;
+			cusers->freed_before = hold->freed;
 			cusers->exceeded = false;
-		} else {
-			scoutfs_alloc_meta_remaining(&server->alloc, &hold->avail, &hold->freed);
 		}
-
 		hold->exceeded = false;
 		hold->start = ktime_get();
 		list_add_tail(&hold->entry, &cusers->holding);
-
 		cusers->nr_holders++;
-
-	} else if (!has_room && cusers->nr_holders == 0 && !cusers->committing) {
-		cusers->committing = true;
-		queue_work(server->wq, &server->commit_work);
+		held = true;
 	}
 
 	spin_unlock(&cusers->lock);
@@ -423,6 +403,7 @@ static int server_apply_commit(struct super_block *sb, struct commit_hold *hold,
 	DECLARE_SERVER_INFO(sb, server);
 	struct commit_users *cusers = &server->cusers;
 	struct timespec ts;
+	bool start_commit;
 
 	spin_lock(&cusers->lock);
 
@@ -443,15 +424,13 @@ static int server_apply_commit(struct super_block *sb, struct commit_hold *hold,
 		list_del_init(&hold->entry);
 		hold->ret = err;
 	}
-
 	cusers->nr_holders--;
-	if (cusers->nr_holders == 0 && !cusers->committing && !list_empty(&cusers->applying)) {
-		cusers->committing = true;
-		queue_work(server->wq, &server->commit_work);
-	}
-
+	start_commit = cusers->nr_holders == 0 && !list_empty(&cusers->applying);
 	spin_unlock(&cusers->lock);
 
+	if (start_commit)
+		queue_work(server->wq, &server->commit_work);
+
 	wait_event(cusers->waitq, list_empty_careful(&hold->entry));
 	smp_rmb(); /* entry load before ret */
 	return hold->ret;
@@ -459,8 +438,8 @@ static int server_apply_commit(struct super_block *sb, struct commit_hold *hold,
 
 /*
  * Start a commit from the commit work.  We should only have been queued
- * while there are no active holders and someone started the commit.
- * There may or may not be blocked apply callers waiting for the result.
+ * while a holder is waiting to apply after all active holders have
+ * finished.
  */
 static int commit_start(struct super_block *sb, struct commit_users *cusers)
 {
@@ -469,7 +448,7 @@ static int commit_start(struct super_block *sb, struct commit_users *cusers)
 	/* make sure holders held off once commit started */
 	spin_lock(&cusers->lock);
 	TRACE_COMMIT_USERS(sb, cusers, start);
-	if (WARN_ON_ONCE(!cusers->committing || cusers->nr_holders != 0))
+	if (WARN_ON_ONCE(list_empty(&cusers->applying) || cusers->nr_holders != 0))
 		ret = -EINVAL;
 	spin_unlock(&cusers->lock);
 
@@ -492,7 +471,6 @@ static void commit_end(struct super_block *sb, struct commit_users *cusers, int
 	smp_wmb(); /* ret stores before list updates */
 	list_for_each_entry_safe(hold, tmp, &cusers->applying, entry)
 		list_del_init(&hold->entry);
-	cusers->committing = false;
 	spin_unlock(&cusers->lock);
 
 	wake_up(&cusers->waitq);
@@ -505,7 +483,7 @@ static void get_stable(struct super_block *sb, struct scoutfs_super_block *super
 	unsigned int seq;
 
 	do {
-		seq = read_seqbegin(&server->seqlock);
+		seq = read_seqcount_begin(&server->stable_seqcount);
 		if (super)
 			*super = server->stable_super;
 		if (roots) {
@@ -513,7 +491,7 @@ static void get_stable(struct super_block *sb, struct scoutfs_super_block *super
 			roots->logs_root = server->stable_super.logs_root;
 			roots->srch_root = server->stable_super.srch_root;
 		}
-	} while (read_seqretry(&server->seqlock, seq));
+	} while (read_seqcount_retry(&server->stable_seqcount, seq));
 }
 
 u64 scoutfs_server_seq(struct super_block *sb)
@@ -547,9 +525,11 @@ void scoutfs_server_set_seq_if_greater(struct super_block *sb, u64 seq)
 
 static void set_stable_super(struct server_info *server, struct scoutfs_super_block *super)
 {
-	write_seqlock(&server->seqlock);
+	preempt_disable();
+	write_seqcount_begin(&server->stable_seqcount);
 	server->stable_super = *super;
-	write_sequnlock(&server->seqlock);
+	write_seqcount_end(&server->stable_seqcount);
+	preempt_enable();
 }
 
 /*
@@ -563,7 +543,7 @@ static void set_stable_super(struct server_info *server, struct scoutfs_super_bl
  * implement commits with a single pending work func.
  *
  * Processing paths hold the commit while they're making multiple
- * dependent changes.  When they're done and want it persistent they
+ * dependent changes.  When they're done and want it persistent they add
  * queue the commit work.  This work runs, performs the commit, and
  * wakes all the applying waiters with the result.  Readers can run
  * concurrently with these commits.
@@ -1966,7 +1946,9 @@ static int server_srch_get_compact(struct super_block *sb,
 	ret = scoutfs_srch_get_compact(sb, &server->alloc, &server->wri,
 				       &super->srch_root, rid, sc);
 	mutex_unlock(&server->srch_mutex);
-	if (ret < 0 || (ret == 0 && sc->nr == 0))
+	if (ret == 0 && sc->nr == 0)
+		ret = -ENOENT;
+	if (ret < 0)
 		goto apply;
 
 	mutex_lock(&server->alloc_mutex);
@@ -3068,9 +3050,9 @@ static int server_get_volopt(struct super_block *sb, struct scoutfs_net_connecti
 	}
 
 	do {
-		seq = read_seqbegin(&server->seqlock);
+		seq = read_seqcount_begin(&server->volopt_seqcount);
 		volopt = server->volopt;
-	} while (read_seqretry(&server->seqlock, seq));
+	} while (read_seqcount_retry(&server->volopt_seqcount, seq));
 
 out:
 	return scoutfs_net_response(sb, conn, cmd, id, ret, &volopt, sizeof(volopt));
@@ -3139,12 +3121,12 @@ static int server_set_volopt(struct super_block *sb, struct scoutfs_net_connecti
 apply:
 	ret = server_apply_commit(sb, &hold, ret);
 
-	write_seqlock(&server->seqlock);
+	write_seqcount_begin(&server->volopt_seqcount);
 	if (ret == 0)
 		server->volopt = super->volopt;
 	else
 		super->volopt = server->volopt;
-	write_sequnlock(&server->seqlock);
+	write_seqcount_end(&server->volopt_seqcount);
 
 	mutex_unlock(&server->volopt_mutex);
 out:
@@ -3187,12 +3169,12 @@ static int server_clear_volopt(struct super_block *sb, struct scoutfs_net_connec
 
 	ret = server_apply_commit(sb, &hold, ret);
 
-	write_seqlock(&server->seqlock);
+	write_seqcount_begin(&server->volopt_seqcount);
 	if (ret == 0)
 		server->volopt = super->volopt;
 	else
 		super->volopt = server->volopt;
-	write_sequnlock(&server->seqlock);
+	write_seqcount_end(&server->volopt_seqcount);
 
 	mutex_unlock(&server->volopt_mutex);
 out:
@@ -4331,9 +4313,9 @@ static void scoutfs_server_worker(struct work_struct *work)
 	}
 
 	/* update volume options early, possibly for use during startup */
-	write_seqlock(&server->seqlock);
+	write_seqcount_begin(&server->volopt_seqcount);
 	server->volopt = super->volopt;
-	write_sequnlock(&server->seqlock);
+	write_seqcount_end(&server->volopt_seqcount);
 
 	atomic64_set(&server->seq_atomic, le64_to_cpu(super->seq));
 	set_stable_super(server, super);
@@ -4473,7 +4455,6 @@ int scoutfs_server_setup(struct super_block *sb)
 
 	server->sb = sb;
 	spin_lock_init(&server->lock);
-	seqlock_init(&server->seqlock);
 	init_waitqueue_head(&server->waitq);
 	INIT_WORK(&server->work, scoutfs_server_worker);
 	server->status = SERVER_DOWN;
@@ -4488,6 +4469,8 @@ int scoutfs_server_setup(struct super_block *sb)
 	INIT_WORK(&server->log_merge_free_work, server_log_merge_free_work);
 	mutex_init(&server->srch_mutex);
 	mutex_init(&server->mounted_clients_mutex);
+	seqcount_init(&server->stable_seqcount);
+	seqcount_init(&server->volopt_seqcount);
 	mutex_init(&server->volopt_mutex);
 	INIT_WORK(&server->fence_pending_recov_work, fence_pending_recov_worker);
 	INIT_DELAYED_WORK(&server->reclaim_dwork, reclaim_worker);

kmod/src/srch.c

@@ -30,7 +30,6 @@
 #include "client.h"
 #include "counters.h"
 #include "scoutfs_trace.h"
-#include "triggers.h"
 
 /*
  * This srch subsystem gives us a way to find inodes that have a given
@@ -521,95 +520,6 @@ out:
 	return ret;
 }
 
-/*
- * Padded entries are encoded in pairs after an existing entry.  All of
- * the pairs cancel each other out by all readers (the second encoding
- * looks like deletion) so they aren't visible to the first/last bounds of
- * the block or file.
- */
-static int append_padded_entry(struct scoutfs_srch_file *sfl, u64 blk,
-			       struct scoutfs_srch_block *srb, struct scoutfs_srch_entry *sre)
-{
-	int ret;
-
-	ret = encode_entry(srb->entries + le32_to_cpu(srb->entry_bytes),
-			   sre, &srb->tail);
-	if (ret > 0) {
-		srb->tail = *sre;
-		le32_add_cpu(&srb->entry_nr, 1);
-		le32_add_cpu(&srb->entry_bytes, ret);
-		le64_add_cpu(&sfl->entries, 1);
-		ret = 0;
-	}
-
-	return ret;
-}
-
-/*
- * This is called by a testing trigger to create a very specific case of
- * encoded entry offsets.  We want the last entry in the block to start
- * precisely at the _SAFE_BYTES offset.
- *
- * This is called when there is a single existing entry in the block.
- * We have the entire block to work with.  We encode pairs of matching
- * entries.  This hides them from readers (both searches and merging) as
- * they're interpreted as creation and deletion and are deleted.  We use
- * the existing hash value of the first entry in the block but then set
- * the inode to an impossibly large number so it doesn't interfere with
- * anything.
- *
- * To hit the specific offset we very carefully manage the amount of
- * bytes of change between fields in the entry.  We know that if we
- * change all the byte of the ino and id we end up with a 20 byte
- * (2+8+8,2) encoding of the pair of entries.  To have the last entry
- * start at the _SAFE_POS offset we know that the final 20 byte pair
- * encoding needs to end at 2 bytes (second entry encoding) after the
- * _SAFE_POS offset.
- *
- * So as we encode pairs we watch the delta of our current offset from
- * that desired final offset of 2 past _SAFE_POS.  If we're a multiple
- * of 20 away then we encode the full 20 byte pairs.  If we're not, then
- * we drop a byte to encode 19 bytes.  That'll slowly change the offset
- * to be a multiple of 20 again while encoding large entries.
- */
-static void pad_entries_at_safe(struct scoutfs_srch_file *sfl, u64 blk,
-				struct scoutfs_srch_block *srb)
-{
-	struct scoutfs_srch_entry sre;
-	u32 target;
-	s32 diff;
-	u64 hash;
-	u64 ino;
-	u64 id;
-	int ret;
-
-	hash = le64_to_cpu(srb->tail.hash);
-	ino = le64_to_cpu(srb->tail.ino) | (1ULL << 62);
-	id = le64_to_cpu(srb->tail.id);
-
-	target = SCOUTFS_SRCH_BLOCK_SAFE_BYTES + 2;
-
-	while ((diff = target - le32_to_cpu(srb->entry_bytes)) > 0) {
-		ino ^= 1ULL << (7 * 8);
-		if (diff % 20 == 0) {
-			id ^= 1ULL << (7 * 8);
-		} else {
-			id ^= 1ULL << (6 * 8);
-		}
-
-		sre.hash = cpu_to_le64(hash);
-		sre.ino = cpu_to_le64(ino);
-		sre.id = cpu_to_le64(id);
-
-		ret = append_padded_entry(sfl, blk, srb, &sre);
-		if (ret == 0)
-			ret = append_padded_entry(sfl, blk, srb, &sre);
-		BUG_ON(ret != 0);
-
-		diff = target - le32_to_cpu(srb->entry_bytes);
-	}
-}
-
 /*
  * The caller is dropping an ino/id because the tracking rbtree is full.
  * This loses information so we can't return any entries at or after the
@@ -1077,9 +987,6 @@ int scoutfs_srch_rotate_log(struct super_block *sb,
 	struct scoutfs_key key;
 	int ret;
 
-	if (sfl->ref.blkno && !force && scoutfs_trigger(sb, SRCH_FORCE_LOG_ROTATE))
-		force = true;
-
 	if (sfl->ref.blkno == 0 ||
 	    (!force && le64_to_cpu(sfl->blocks) < SCOUTFS_SRCH_LOG_BLOCK_LIMIT))
 		return 0;
@@ -1555,7 +1462,7 @@ static int kway_merge(struct super_block *sb,
 		      struct scoutfs_block_writer *wri,
 		      struct scoutfs_srch_file *sfl,
 		      kway_get_t kway_get, kway_advance_t kway_adv,
-		      void **args, int nr, bool logs_input)
+		      void **args, int nr)
 {
 	DECLARE_SRCH_INFO(sb, srinf);
 	struct scoutfs_srch_block *srb = NULL;
@@ -1660,15 +1567,6 @@ static int kway_merge(struct super_block *sb,
 				blk++;
 			}
 
-			/* end sorted block on _SAFE offset for testing */
-			if (bl && le32_to_cpu(srb->entry_nr) == 1 && logs_input &&
-			    scoutfs_trigger(sb, SRCH_COMPACT_LOGS_PAD_SAFE)) {
-				pad_entries_at_safe(sfl, blk, srb);
-				scoutfs_block_put(sb, bl);
-				bl = NULL;
-				blk++;
-			}
-
 			scoutfs_inc_counter(sb, srch_compact_entry);
 
 		} else {
@@ -1711,8 +1609,6 @@ static int kway_merge(struct super_block *sb,
 			empty++;
 			ret = 0;
 		} else if (ret < 0) {
-			if (ret == -ENOANO) /* just testing trigger */
-				ret = 0;
 			goto out;
 		}
 
@@ -1920,7 +1816,7 @@ static int compact_logs(struct super_block *sb,
 	}
 
 	ret = kway_merge(sb, alloc, wri, &sc->out, kway_get_page, kway_adv_page,
-			 args, nr_pages, true);
+			 args, nr_pages);
 	if (ret < 0)
 		goto out;
 
@@ -1978,18 +1874,12 @@ static int kway_get_reader(struct super_block *sb,
 	srb = rdr->bl->data;
 
 	if (rdr->pos > SCOUTFS_SRCH_BLOCK_SAFE_BYTES ||
-	    rdr->skip > SCOUTFS_SRCH_BLOCK_SAFE_BYTES ||
+	    rdr->skip >= SCOUTFS_SRCH_BLOCK_SAFE_BYTES ||
 	    rdr->skip >= le32_to_cpu(srb->entry_bytes)) {
 		/* XXX inconsistency */
 		return -EIO;
 	}
 
-	if (rdr->decoded_bytes == 0 && rdr->pos == SCOUTFS_SRCH_BLOCK_SAFE_BYTES &&
-	    scoutfs_trigger(sb, SRCH_MERGE_STOP_SAFE)) {
-		/* only used in testing */
-		return -ENOANO;
-	}
-
 	/* decode entry, possibly skipping start of the block */
 	while (rdr->decoded_bytes == 0 || rdr->pos < rdr->skip) {
 		ret = decode_entry(srb->entries + rdr->pos,
@@ -2079,7 +1969,7 @@ static int compact_sorted(struct super_block *sb,
 	}
 
 	ret = kway_merge(sb, alloc, wri, &sc->out, kway_get_reader,
-			 kway_adv_reader, args, nr, false);
+			 kway_adv_reader, args, nr);
 
 	sc->flags |= SCOUTFS_SRCH_COMPACT_FLAG_DONE;
 	for (i = 0; i < nr; i++) {
@@ -2289,7 +2179,7 @@ out:
 
 	scoutfs_block_writer_forget_all(sb, &wri);
 	if (!atomic_read(&srinf->shutdown)) {
-		delay = (sc->nr > 0 && ret == 0) ? 0 : msecs_to_jiffies(SRCH_COMPACT_DELAY_MS);
+		delay = ret == 0 ? 0 : msecs_to_jiffies(SRCH_COMPACT_DELAY_MS);
 		queue_delayed_work(srinf->workq, &srinf->compact_dwork, delay);
 	}
 

kmod/src/triggers.c

@@ -39,9 +39,6 @@ struct scoutfs_triggers {
 
 static char *names[] = {
 	[SCOUTFS_TRIGGER_BLOCK_REMOVE_STALE] = "block_remove_stale",
-	[SCOUTFS_TRIGGER_SRCH_COMPACT_LOGS_PAD_SAFE] = "srch_compact_logs_pad_safe",
-	[SCOUTFS_TRIGGER_SRCH_FORCE_LOG_ROTATE] = "srch_force_log_rotate",
-	[SCOUTFS_TRIGGER_SRCH_MERGE_STOP_SAFE] = "srch_merge_stop_safe",
 	[SCOUTFS_TRIGGER_STATFS_LOCK_PURGE] = "statfs_lock_purge",
 };
 

kmod/src/triggers.h

@@ -3,9 +3,6 @@
 
 enum scoutfs_trigger {
 	SCOUTFS_TRIGGER_BLOCK_REMOVE_STALE,
-	SCOUTFS_TRIGGER_SRCH_COMPACT_LOGS_PAD_SAFE,
-	SCOUTFS_TRIGGER_SRCH_FORCE_LOG_ROTATE,
-	SCOUTFS_TRIGGER_SRCH_MERGE_STOP_SAFE,
 	SCOUTFS_TRIGGER_STATFS_LOCK_PURGE,
 	SCOUTFS_TRIGGER_NR,
 };

kmod/src/util.h

@@ -23,9 +23,9 @@ static inline void down_write_two(struct rw_semaphore *a,
  * ~0UL values. Hence, we cap count to ~0L, which is arbitarily high
  * enough to avoid it.
  */
-static inline long shrinker_min_long(long count)
+static inline unsigned long shrinker_min_t_long(unsigned long count)
 {
-	return min(count, LONG_MAX);
+	return min_t(u64, count, LONG_MAX);
 }
 
 #endif

tests/funcs/exec.sh

@@ -35,7 +35,7 @@ t_fail()
 t_quiet()
 {
 	echo "# $*" >> "$T_TMPDIR/quiet.log"
-	"$@" >> "$T_TMPDIR/quiet.log" 2>&1 || \
+	"$@" > "$T_TMPDIR/quiet.log" 2>&1 || \
 		t_fail "quiet command failed"
 }
 

tests/funcs/filter.sh

@@ -88,8 +88,5 @@ t_filter_dmesg()
 	# change-devices causes loop device resizing
 	re="$re|loop[0-9].* detected capacity change from.*"
 
-	# ignore systemd-journal rotating
-	re="$re|systemd-journald.*"
-
 	egrep -v "($re)" 
 }

tests/funcs/fs.sh

@@ -265,15 +265,6 @@ t_trigger_get() {
 	cat "$(t_trigger_path "$nr")/$which"
 }
 
-t_trigger_set() {
-	local which="$1"
-	local nr="$2"
-	local val="$3"
-	local path=$(t_trigger_path "$nr")
-
-	echo "$val" > "$path/$which"
-}
-
 t_trigger_show() {
 	local which="$1"
 	local string="$2"
@@ -285,8 +276,9 @@ t_trigger_show() {
 t_trigger_arm_silent() {
 	local which="$1"
 	local nr="$2"
+	local path=$(t_trigger_path "$nr")
 
-	t_trigger_set "$which" "$nr" 1
+	echo 1 > "$path/$which"
 }
 
 t_trigger_arm() {

tests/golden/srch-safe-merge-pos

@@ -1,81 +0,0 @@
-== snapshot errors
-== arm compaction triggers
-trigger srch_compact_logs_pad_safe armed: 1
-trigger srch_merge_stop_safe armed: 1
-trigger srch_compact_logs_pad_safe armed: 1
-trigger srch_merge_stop_safe armed: 1
-trigger srch_compact_logs_pad_safe armed: 1
-trigger srch_merge_stop_safe armed: 1
-trigger srch_compact_logs_pad_safe armed: 1
-trigger srch_merge_stop_safe armed: 1
-trigger srch_compact_logs_pad_safe armed: 1
-trigger srch_merge_stop_safe armed: 1
-== force lots of small rotated log files for compaction
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-trigger srch_force_log_rotate armed: 1
-== wait for compaction
-== test and disarm compaction triggers
-== verify triggers and errors
-== cleanup

tests/sequence

@@ -14,7 +14,6 @@ offline-extent-waiting.sh
 move-blocks.sh
 large-fragmented-free.sh
 enospc.sh
-srch-safe-merge-pos.sh
 srch-basic-functionality.sh
 simple-xattr-unit.sh
 totl-xattr-tag.sh

tests/tests/basic-truncate.sh

@@ -11,13 +11,8 @@ FILE="$T_D0/file"
 # final block as we truncated past it.
 #
 echo "== truncate writes zeroed partial end of file block"
-yes | dd of="$FILE" bs=8K count=1 status=none iflag=fullblock
+yes | dd of="$FILE" bs=8K count=1 status=none
 sync
-
-# not passing iflag=fullblock causes the file occasionally to just be
-# 4K, so just to be safe we should at least check size once
-test `stat --printf="%s\n" "$FILE"` -eq 8192 || t_fail "test file incorrect start size"
-
 truncate -s 6K "$FILE"
 truncate -s 12K "$FILE"
 echo 3 > /proc/sys/vm/drop_caches

tests/tests/fence-and-reclaim.sh

@@ -7,11 +7,14 @@ t_require_mounts 2
 
 #
 # Make sure that all mounts can read the results of a write from each
-# mount.
+# mount.  And make sure that the greatest of all the written seqs is
+# visible after the writes were commited by remote reads.
 #
 check_read_write()
 {
 	local expected
+	local greatest=0
+	local seq
 	local path
 	local saw
 	local w
@@ -22,6 +25,11 @@ check_read_write()
 		eval path="\$T_D${w}/written"
 		echo "$expected" > "$path"
 
+		seq=$(scoutfs stat -s meta_seq $path)
+		if [ "$seq" -gt "$greatest" ]; then
+			greatest=$seq
+		fi
+
 		for r in $(t_fs_nrs); do
 			eval path="\$T_D${r}/written"
 			saw=$(cat "$path")
@@ -30,6 +38,11 @@ check_read_write()
 			fi
 		done
 	done
+
+	seq=$(scoutfs statfs -s committed_seq -p $T_D0)
+	if [ "$seq" -lt "$greatest" ]; then
+		echo "committed_seq $seq less than greatest $greatest"
+	fi
 }
 
 # verify that fenced ran our testing fence script

tests/tests/resize-devices.sh

@@ -2,8 +2,6 @@
 # Some basic tests of online resizing metadata and data devices.
 #
 
-t_require_commands bc
-
 statfs_total() {
 	local single="total_$1_blocks"
 	local mnt="$2"

tests/tests/srch-safe-merge-pos.sh

@@ -1,69 +0,0 @@
-#
-# There was a bug where srch file compaction could get stuck if a
-# partial compaction finished at the specific _SAFE_BYTES offset in a
-# block.  Resuming from that position would return an error and
-# compaction would stop making forward progress.
-#
-# We use triggers to make sure that we create the circumstance where a
-# sorted srch block ends at the _SAFE_BYTES offsset and that a merge
-# request stops with a partial block at that specific offset.  We then
-# watch error counters to make sure compaction doesn't get stuck.
-#
-
-# forcing rotation, so just a few
-NR=10
-SEQF="%.20g"
-COMPACT_NR=4
-
-echo "== snapshot errors"
-declare -a err
-for nr in $(t_fs_nrs); do
-	err[$nr]=$(t_counter srch_compact_error $nr)
-done
-
-echo "== arm compaction triggers"
-for nr in $(t_fs_nrs); do
-	t_trigger_arm srch_compact_logs_pad_safe $nr
-	t_trigger_arm srch_merge_stop_safe $nr
-done
-
-echo "== force lots of small rotated log files for compaction"
-sv=$(t_server_nr)
-iter=1
-while [ $iter -le $((COMPACT_NR * COMPACT_NR * COMPACT_NR)) ]; do
-	t_trigger_arm srch_force_log_rotate $sv
-
-	seq -f "f-$iter-$SEQF" 1 10 | src/bulk_create_paths -S -d "$T_D0" > /dev/null
-	sync
-
-	test "$(t_trigger_get srch_force_log_rotate $sv)" == "0" || \
-		t_fail "srch_force_log_rotate didn't trigger"
-
-	((iter++))
-done
-
-echo "== wait for compaction"
-sleep 15
-
-echo "== test and disarm compaction triggers"
-pad=0
-merge_stop=0
-for nr in $(t_fs_nrs); do
-	test "$(t_trigger_get srch_compact_logs_pad_safe $nr)" == "0" && pad=1
-	t_trigger_set srch_compact_logs_pad_safe $nr 0
-	test "$(t_trigger_get srch_merge_stop_safe $nr)" == "0" && merge_stop=1
-	t_trigger_set srch_merge_stop_safe $nr 0
-done
-
-echo "== verify triggers and errors" 
-test $pad == 1 || t_fail "srch_compact_logs_pad_safe didn't trigger"
-test $merge_stop == 1 || t_fail "srch_merge_stop_safe didn't trigger"
-for nr in $(t_fs_nrs); do
-	test "$(t_counter srch_compact_error $nr)" == "${err[$nr]}" || \
-		t_fail "srch_compact_error counter increased on mount $nr"
-done
-
-echo "== cleanup"
-find "$T_D0" -type f -name 'f-*' -delete
-
-t_pass