zstream: self-tuning queues

This PR removes the `qp_batch_budget` and `qp_queue_length` parameters
from the `zq_params_t` struct that's used to create `zstream_queues.`
Queue lengths and batch budgets still exist, but they're set internally
by the queue implementation. In addition, `ZQ_MAX_BATCH` is raised to
1024 items.

### Motivation

Multithreaded queues need some specific value for queue lengths and
batch sizes, but the impact of these values is largely opaque to
developers without detailed benchmarking.

Queue lengths in particular have been difficult to optimize because they
were formerly the only way to limit the amount of in-flight stream
data. Streams vary in payload density, so queue lengths were set
according to relatively pessimistic assumptions. But as of #18982, the
memory consumed by in-flight payloads is limited independently of queue
operations. Queues now have wider latitude to use throughput-optimized
parameters.

### Benchmark results

I've been investigating what those values should be. Here are the main
findings:

- Longer queues are always better. They facilitate larger batches,
  reduce (proportional) overhead, and smooth irregularities in the
  stream profile and in I/O performance. The magnitude of the
  improvement can be a reduction of up to 15% in total execution time.

- The main cost of longer queues is increased memory consumption, not
  because the storage for queue slots is significant but because long
  queues mean a long pipeline and a correspondingly larger number of
  payloads in memory.

- Despite that, driving the pipeline up against memory limits is helpful
  rather than counterproductive. Ingress throttling makes records
  arrive in a burstier pattern, but long queues also enable the
  pipeline to handle those bursts more smoothly.

- Input throttling happens only when there's a bottleneck downstream of
  the input source. Pipelines are linear, so by definition every stage
  has identical average throughput, and that throughput is set by the
  bottleneck. It's fine for the input stage to spend much of its time
  waiting for memory; in cases where that occurs, higher input
  bandwidth generally wouldn't help.

- The time needed to process a byte of payload data varies by more than
  two orders of magnitude among queues, depending on the work each
  queue is doing. This much variation makes it hard to state rules of
  thumb for how large batches ought to be. Queues need to observe their
  own work rates and set batch sizes accordingly.

- Batch sizes can cause problems both by being too small and by being
  too large. Too small, and many more dispatch loops are needed to farm
  out all the work. Too large, and latency starts to affect overall
  performance.

### Changes in this PR

This PR replaces static batch sizes with batches sized to require 400
microseconds of processing on one thread. To determine the mapping
between time and batch size, each queue accumulates the total cost it
has processed and the total wall-clock time it spent processing, and
calculates the ratio. Both are running totals over the life of the
queue, so the estimate converges early and is largely static.

The 400 microsecond figure is empirical. However, as an absolute
reference point, it should be relatively transportable across systems.

Because batch size is now derived from cost, cost functions carry more
weight than they used to: a cost that doesn't scale with processing
time will mis-size batches. This is documented in `zstream_queue.h`.

All queues are now statically sized at 4096 slots. Longer queues would
be even more performant, but 4096 slots is right about where
diminishing returns start to come into play. Note that this is a fixed
per-queue allocation of the ring buffer and its item storage, made at
queue creation time regardless of load; for `drr_packet_t` items it
works out to about 1.4 MB per queue.

Batch sizes are now capped at 1024 work items, up from 32. In theory,
unlimited batch sizes (that is, batches limited only by the 400us time
quantum) would be optimal. But real-world batch size statistics show a
predominance of modestly-sized batches with a long right tail. Few
batches have the opportunity to become larger than 1024 items.

`ZQ_SLOTS_PER_QUEUE` and `ZQ_MAX_BATCH` remain in `zstream_queue.h`,
but only so that tests and diagnostics can reason about the real
boundaries. They are not caller-tunable.

The queue scoring mechanism used during thread dispatch has been
simplified. A queue's score is now just its number of unclaimed work
items. The more complex scoring used in the past also works fine, but
the extra complexity doesn't yield better performance. I also
investigated backpressure scoring, which incorporates knowledge of the
downstream neighbor's state; that wasn't of benefit either. Stochastic
assignment remains, but it's now calculated with integer arithmetic.

Since queue length is no longer a parameter, the
`STANDARD_INPUT_STACK_Q` and `STANDARD_OUTPUT_STACK_Q` macros are gone,
and `parallel_calc_fletcher4()` no longer takes an argument.

Batch size histograms have been separated out from the
`MONITOR_QUEUES` #define into a separate option,
`SHOW_BATCH_HISTOGRAMS`. `MONITOR_QUEUES` now shows mean and standard
deviation figures for each time interval, sampled on both enqueue and
dequeue. It uses Welford's algorithm, and needs `-lm` for `sqrt()`;
that's why libm has been added to `zstream_LDADD`.

`zstream selftest queue` gains a `queue_batch_tuning` case that runs
queues with deliberately dissimilar cost-to-time relationships,
including ones that pin the derived batch size to each end of its
range. `queue_edge_cases` now sweeps stream lengths that straddle
`ZQ_MAX_BATCH` and `ZQ_SLOTS_PER_QUEUE` instead of the parameters that
no longer exist.

Reviewed-by: Brian Behlendorf <[email protected]>
Signed-off-by: Garth Snyder <[email protected]>
Closes #18991

cmd/zstream/Makefile.am

@@ -37,11 +37,13 @@ zstream_SOURCES = \
 	%D%/zstream_validate.c \
 	%D%/zstream_validate.h
 
+# -lm supplies sqrt() for MONITOR_QUEUES
 zstream_LDADD = \
 	libzfs.la \
 	libzfs_core.la \
 	libzpool.la \
-	libnvpair.la
+	libnvpair.la \
+	-lm
 
 cmd-zstream-install-exec-hook:
 	cd $(DESTDIR)$(sbindir) && $(LN_S) -f zstream zstreamdump

cmd/zstream/zstream_chain.c

@@ -234,8 +234,6 @@ zstream_chain_exec(zstream_chain_t chain, chain_attrs_t *attrs)
 				.qp_process	 = cs->cs_parallel.process,
 				.qp_cost	 = cs->cs_parallel.cost,
 				.qp_item_size	 = stats.ct_item_size,
-				.qp_batch_budget = cs->cs_parallel.batch_budget,
-				.qp_queue_length = cs->cs_parallel.queue_length,
 				.qp_context	 = cs->cs_context
 			};
 			queue = zstream_queue_create(&queue_params);

cmd/zstream/zstream_chain.h

@@ -158,8 +158,6 @@ typedef struct chain_step
 			zc_serial_process_f	*process;	/* serial */
 		} cs_serial;
 		struct {
-			size_t			queue_length;	/* parallel */
-			size_t			batch_budget;
 			zq_estimate_cost_f	*cost;
 			zq_process_item_f	*process;
 		} cs_parallel;

cmd/zstream/zstream_fletcher4.c

@@ -276,20 +276,22 @@ chain_fletcher4(queue_item_t *item_in, void *context_in)
 
 /*
  * Since checksumming is either very early or very late in the chain, these
- * queues effectively double as I/O buffers. Ergo, the default queue length
- * is long. The batch budget is also large because Fletcher 4 calculations
- * are fast.
+ * queues effectively double as I/O buffers, absorbing irregularity in the
+ * rate at which the stream arrives or drains.
+ *
+ * Payload length is the cost basis. Fletcher 4 runs at a nearly constant
+ * rate per byte, so cost tracks processing time closely and the queue's
+ * automatic batch sizing has an easy job of it. Batches end up large,
+ * because the per-byte cost is small.
  */
 chain_step_t
-parallel_calc_fletcher4(int queue_length)
+parallel_calc_fletcher4(void)
 {
 	chain_step_t step = {
 	    .cs_type = CS_PARALLEL,
 	    .cs_in_size = sizeof (drr_packet_t),
 	    .cs_out_size = sizeof (drr_fletcher4_t),
 	    .cs_parallel = {
-		.queue_length = queue_length,
-		.batch_budget = 256 * 1024,
 		.process = chain_calc_fletcher4,
 		.cost = payload_size_as_cost
 	    }

cmd/zstream/zstream_fletcher4.h

@@ -80,7 +80,7 @@ typedef struct {
 } drr_fletcher4_t;
 
 chain_step_t
-parallel_calc_fletcher4(int queue_length);
+parallel_calc_fletcher4(void);
 
 chain_step_t
 serial_validate_fletcher4(void);

cmd/zstream/zstream_io.c

@@ -100,14 +100,14 @@ static pthread_once_t	dif_init_control = PTHREAD_ONCE_INIT;
 static void
 initialize_memory_tracking(void)
 {
-	ssize_t pagesize = (ssize_t)sysconf(_SC_PAGESIZE);
-	ssize_t pages = (ssize_t)sysconf(_SC_PHYS_PAGES);
+	int64_t pagesize = (int64_t)sysconf(_SC_PAGESIZE);
+	int64_t pages = (int64_t)sysconf(_SC_PHYS_PAGES);
 	if (pagesize < 0 || pages < 0) {
 		warnx("unable to read system memory info");
 		payloads.dif_allowed = UINT64_MAX; /* no limit */
 	} else {
-		uint64_t total_mem = (uint64_t)pagesize * (uint64_t)pages;
-		int64_t flex = total_mem - MEMORY_BASE_CUTOFF;
+		int64_t total_mem = pagesize * pages;
+		int64_t flex = total_mem - (int64_t)MEMORY_BASE_CUTOFF;
 		int64_t addl = (double)flex * MEMORY_PCT / 100;
 		payloads.dif_allowed = MEMORY_BASE + MAX(addl, 0);
 	}

cmd/zstream/zstream_modules.h

@@ -35,23 +35,20 @@ extern "C" {
 #include "zstream_util.h"
 #include "zstream_validate.h"
 
-#define	STANDARD_INPUT_STACK_Q(infile, queue_size) 			\
+#define	STANDARD_INPUT_STACK(infile) 					\
 	serial_read_stream(infile),					\
-	parallel_calc_fletcher4(queue_size),				\
+	parallel_calc_fletcher4(),					\
 	serial_validate_fletcher4(),					\
 	serial_byteswap(BS_INCOMING),					\
 	serial_validate_records()
 
-#define	STANDARD_OUTPUT_STACK_Q(outfile, queue_size) 			\
+#define	STANDARD_OUTPUT_STACK(outfile) 					\
 	serial_byteswap(BS_OUTGOING),					\
-	parallel_calc_fletcher4(queue_size),				\
+	parallel_calc_fletcher4(),					\
 	serial_add_fletcher4(),						\
 	serial_write_stream(outfile),					\
 	chain_terminator()
 
-#define	STANDARD_INPUT_STACK(infile)	STANDARD_INPUT_STACK_Q(infile, 1024)
-#define	STANDARD_OUTPUT_STACK(outfile)	STANDARD_OUTPUT_STACK_Q(outfile, 512)
-
 #define	NULL_OUTPUT_STACK()						\
 	serial_null_output(),						\
 	chain_terminator()

cmd/zstream/zstream_queue.c

@@ -18,6 +18,7 @@
 #include <atomic.h>
 #include <err.h>
 #include <errno.h>
+#include <math.h>
 #include <pthread.h>
 #include <sched.h>
 #include <stddef.h>
@@ -38,16 +39,16 @@
 
 #define	ENQUEUE_DELAY_NSEC	(100 * 1000)		/* 100us */
 #define	DISPATCH_BACKUP_NSEC	(1000 * 1000)		/* 1ms */
+#define	BATCH_TIME_NSEC		(400 * 1000)		/* 400us */
 
-#define	PLENTY_OF_WORK		6	/* "Many" items to claim */
-#define	NO_WORK			1.0E-6	/* No-work score threshold */
-#define	DEQUEUE_SCORE_WEIGHT	0.3	/* Dequeue score relative weight */
-
-#define	Q_MOD(queue, index)	((index) % (queue)->zq_params.qp_queue_length)
-#define	Q_SLOT(queue, index)	((queue)->zq_slots[Q_MOD((queue), (index))])
+#define	Q_MOD(index)		((index) % ZQ_SLOTS_PER_QUEUE)
+#define	Q_SLOT(queue, index)	((queue)->zq_slots[Q_MOD(index)])
 
 #define	Q_FULL(queue)	((queue)->zq_ix.enqueue - (queue)->zq_ix.dequeue >= \
-	    (queue)->zq_params.qp_queue_length)
+	    ZQ_SLOTS_PER_QUEUE)
+
+// #define	MONITOR_QUEUES		/* Queue tenancy data per interval */
+// #define	SHOW_BATCH_HISTOGRAMS	/* Batch size histograms per queue */
 
 /*
  * A zstream_queue is a ring buffer with four indexes: enqueue, claim,
@@ -133,9 +134,27 @@ typedef struct {
 	pthread_cond_t	dequeued;
 } zq_conditions_t;
 
+#ifdef MONITOR_QUEUES
+/*
+ * Running mean and standard deviation of queue depth, accumulated by
+ * Welford's online algorithm. Welford is numerically stable and doesn't
+ * need wider-than-platform integers.
+ */
 typedef struct {
-	int		min_depth;
-	int		max_depth;
+	int		min;
+	int		max;
+	uint64_t	ds_samples;
+	double		ds_mean;
+	double		ds_m2;		/* Sum of squared deviations */
+} depth_stats_t;
+#endif
+
+typedef struct {
+#ifdef MONITOR_QUEUES
+	depth_stats_t	depth;
+#endif
+	uint64_t	nsec_used;
+	size_t		cost_processed;
 } zq_stats_t;
 
 struct zstream_queue {
@@ -146,8 +165,8 @@ struct zstream_queue {
 	zq_conditions_t	zq_cond;
 	zq_params_t	zq_params;
 	boolean_t	zq_disallow_enqueue;
-#ifdef MONITOR_QUEUES
 	zq_stats_t	zq_stats;
+#ifdef SHOW_BATCH_HISTOGRAMS
 	uint64_t	zq_histogram[ZQ_MAX_BATCH+1];	/* Batch sizes */
 #endif
 };
@@ -181,6 +200,11 @@ static void *dispatch_worker(void *);
 
 #ifdef MONITOR_QUEUES
 static void *cpu_and_queue_monitor(void *);
+static inline void initialize_monitor_data(zstream_queue_t *);
+static inline void update_monitor_data(zstream_queue_t *);
+#endif
+
+#ifdef SHOW_BATCH_HISTOGRAMS
 static void print_batch_size_histogram(zstream_queue_t *);
 #endif
 
@@ -285,8 +309,6 @@ zstream_queue_create(zq_params_t *params)
 	VERIFY3P(params->qp_process, !=, NULL);
 	VERIFY3P(params->qp_cost, !=, NULL);
 	VERIFY3U(params->qp_item_size, >, 0);
-	VERIFY3U(params->qp_queue_length, >, 0);
-	VERIFY3U(params->qp_queue_length, <, 1 << 18);
 
 	pthread_once(&once_control, thread_pool_init);
 	pthread_mutex_lock(&pool.tp_pool_mutex);
@@ -301,18 +323,19 @@ zstream_queue_create(zq_params_t *params)
 	*queue = (zstream_queue_t) {
 		.zq_id = next_queue_id++,
 		.zq_params = *params,
-		.zq_slots = safe_malloc(params->qp_queue_length *
-		    (sizeof (queue_slot_t))),
-#ifdef MONITOR_QUEUES
-		.zq_stats.min_depth = INT_MAX
-#endif
+		.zq_slots = safe_malloc(ZQ_SLOTS_PER_QUEUE *
+		    (sizeof (queue_slot_t)))
 	};
 	pool.tp_queues[pool.tp_num_queues] = queue;
 
+#ifdef MONITOR_QUEUES
+	initialize_monitor_data(queue);
+#endif
+
 	size_t qpis_rounded = P2ROUNDUP(params->qp_item_size,
 	    _Alignof(worst_case_alignment_t));
-	uint8_t *items = safe_malloc(params->qp_queue_length * qpis_rounded);
-	for (size_t i = 0; i < params->qp_queue_length; i++) {
+	uint8_t *items = safe_malloc(ZQ_SLOTS_PER_QUEUE * qpis_rounded);
+	for (int i = 0; i < ZQ_SLOTS_PER_QUEUE; i++) {
 		queue->zq_slots[i].qs_item =
 		    (queue_item_t *)(items + i * qpis_rounded);
 	}
@@ -390,75 +413,72 @@ advance_indexes(zstream_queue_t *queue)
 }
 
 /*
- * Score a queue according to its need for workers. Higher is better. The
- * scoring tries to assign threads to queues that are running out of space
- * for new enqueuements or that have little completed work available to
- * dequeue. The broader goal is to try to avoid pipeline stalls.
+ * Score a queue according to its need for workers. Higher is better.
  *
- * Two measures are used for scoring. The "open score" is 1/M where M is the
- * number of slots available to receive new items. The "dequeue score" is
- * 1/N where N is the number of completed items available to dequeue. These
- * two measures are added together with the dequeue score scaled by
- * DEQUEUE_SCORE_WEIGHT.
+ * Threads are distributed among queues in proportion to their scores (see
+ * select_stochastic()), so a queue with twice as much outstanding work
+ * attracts twice as many workers. Queues with no claimable work score 0 and
+ * are never selected.
  *
- * The composite score is scaled by a factor that reflects how much work is
- * actually available to be claimed on the queue; there's no point assigning
- * threads to queues that have no work.
+ * The score is just the number of unclaimed items. More elaborate scoring
+ * is possible. Earlier versions weighted queues by how close they were to
+ * being full and by how little completed work they had available to
+ * dequeue. But, benchmarking showed no benefit over the simple count below.
+ * Also not effective: backpressure-style differencing with the next
+ * downstream queue.
  *
  * Locking: the caller must hold the thread pool mutex and the queue mutex.
  */
-static inline double
+static inline uint32_t
 score_queue(zstream_queue_t *queue)
 {
-	uint64_t claimable = queue->zq_ix.enqueue - queue->zq_ix.claim;
-	uint64_t dequeueable = queue->zq_ix.complete - queue->zq_ix.dequeue;
-	uint64_t in_queue = queue->zq_ix.enqueue - queue->zq_ix.dequeue;
-	uint64_t open_slots = queue->zq_params.qp_queue_length - in_queue;
-
-	double open_score = (open_slots > 0) ? (1.0 / open_slots) : 2.0;
-	double dq_score = (dequeueable > 0) ? (1.0 / dequeueable) : 2.0;
-	double claim_factor = MIN(claimable, (uint64_t)PLENTY_OF_WORK) /
-	    (double)PLENTY_OF_WORK;
-	double need = open_score + dq_score * DEQUEUE_SCORE_WEIGHT;
-	return (need * claim_factor);
+	return (queue->zq_ix.enqueue - queue->zq_ix.claim);
 }
 
 /*
- * Return a random index from an array of doubles, with the likelihood of
- * index i being selected equal to weights[i] / sum(weights). Returns index
- * 0 if no weight is greater than 0.
+ * Return a random index into weights[], with the likelihood of index i
+ * being selected equal to weights[i] / sum(weights).
+ *
+ * The caller must guarantee that at least one weight is nonzero;
+ * a uniformly zero array has no meaningful answer and would divide by
+ * zero. assign_queue_and_get_work() checks this before calling.
  */
 static inline int
-select_stochastic(double weights[], int num_values)
+select_stochastic(uint32_t weights[], int num_values)
 {
-	const double denominator = (double)UINT64_MAX;
-	uint64_t numerator;
-	double total = 0.0;
+	uint64_t randval;
+	uint32_t total = 0;
 
 	for (int i = 0; i < num_values; i++) {
 		total += weights[i];
 	}
-	random_get_pseudo_bytes((uint8_t *)&numerator, sizeof (numerator));
-	double select_val = total * numerator / denominator;
+	if (total == 0)
+		return (0);
+	random_get_pseudo_bytes((uint8_t *)&randval, sizeof (randval));
+	uint64_t select_val = randval % total;
 	for (int i = 0; i < num_values; i++) {
 		if (select_val < weights[i])
 			return (i);
 		select_val -= weights[i];
 	}
-	/* Fallback in case of FP rounding not producing a winner */
-	for (int i = num_values - 1; i >= 0; i--) {
-		if (weights[i] != 0.0)
-			return (i);
-	}
-	return (0);
+	return (num_values - 1);
 }
 
 /*
- * Claim up to ZQ_MAX_BATCH work items from the given queue, trying to
- * accumulate at least qp_batch_budget worth of work data (== "cost"). All
- * items in a batch will be drawn from the same queue.
+ * Claim up to ZQ_MAX_BATCH work items from the given queue. All items in a
+ * batch will be drawn from the same queue.
  *
- * Does not block waiting to fill the budget; returns whatever is available.
+ * Queues track the historical relationship between cost values and
+ * processing times, which is assumed to be roughly linear. The goal is for
+ * each batch to have a processing time of BATCH_TIME_NSEC. However, most
+ * batches fall short of this goal because of the availability of work.
+ * claim_batch() does not block waiting to fill the budget; it returns
+ * whatever is available.
+ *
+ * The cost/time calculation is a cumulative average over the life of the
+ * queue, so it converges early and changes only slowly. Stationarity is
+ * assumed. If nonstationary work is being performed, callers should adjust
+ * their cost functions to reflect that.
  *
  * Locking: this function must be called with both the queue mutex and the
  * thread pool mutex held. zstream_queue_destroy() can't hold a queue's
@@ -477,21 +497,31 @@ select_stochastic(double weights[], int num_values)
 static int
 claim_batch(zstream_queue_t *queue, queue_slot_t **batch)
 {
-	size_t cost_claimed = 0;
+	size_t cost_to_claim, cost_claimed = 0;
 	int count = 0;
 	uint64_t passed = 0;
 	boolean_t more_to_claim, more_slots, more_budget;
-	boolean_t first_and_only, ok_to_claim;
+	boolean_t have_cost_data = queue->zq_stats.nsec_used != 0 &&
+	    queue->zq_stats.cost_processed != 0;
+
+	if (have_cost_data) {
+		double ns_per_cost = (double)queue->zq_stats.nsec_used /
+		    (double)queue->zq_stats.cost_processed;
+		double budget = BATCH_TIME_NSEC / ns_per_cost;
+		if (budget >= (double)SIZE_MAX)
+			cost_to_claim = SIZE_MAX;
+		else
+			cost_to_claim = MAX(1, (size_t)budget);
+	} else {
+		cost_to_claim = 1;
+	}
 
 	while (B_TRUE) {
 		more_to_claim = queue->zq_ix.claim < queue->zq_ix.enqueue;
 		more_slots = count < ZQ_MAX_BATCH;
-		more_budget = cost_claimed < queue->zq_params.qp_batch_budget;
-		first_and_only = queue->zq_params.qp_batch_budget == 0 &&
-		    count == 0;
-		ok_to_claim = first_and_only || more_budget;
+		more_budget = cost_claimed < cost_to_claim;
 
-		if (!more_to_claim || !more_slots || !ok_to_claim) {
+		if (!more_to_claim || !more_slots || !more_budget) {
 			break;
 		}
 		queue_slot_t *slot = &Q_SLOT(queue, queue->zq_ix.claim);
@@ -511,7 +541,7 @@ claim_batch(zstream_queue_t *queue, queue_slot_t **batch)
 		atomic_sub_64(&pool.tp_unclaimed, passed);
 	}
 	advance_indexes(queue);
-#ifdef MONITOR_QUEUES
+#ifdef SHOW_BATCH_HISTOGRAMS
 	queue->zq_histogram[count]++;
 #endif
 	return (count);
@@ -534,7 +564,7 @@ assign_queue_and_get_work(zstream_queue_t **queue, queue_slot_t **batch)
 
 	while (B_TRUE) {
 		int num_queues = pool.tp_num_queues;
-		double weights[ZQ_MAX_QUEUES];
+		uint32_t weights[ZQ_MAX_QUEUES];
 		int queues_with_work = 0;
 
 		for (int i = 0; i < num_queues; i++) {
@@ -542,7 +572,7 @@ assign_queue_and_get_work(zstream_queue_t **queue, queue_slot_t **batch)
 			pthread_mutex_lock(&to_score->zq_mutex);
 			weights[i] = score_queue(to_score);
 			pthread_mutex_unlock(&to_score->zq_mutex);
-			if (weights[i] > NO_WORK)
+			if (weights[i] > 0)
 				queues_with_work++;
 		}
 		if (!queues_with_work) {
@@ -567,6 +597,14 @@ assign_queue_and_get_work(zstream_queue_t **queue, queue_slot_t **batch)
 	}
 }
 
+static inline uint64_t
+time_now_ns(void)
+{
+	struct timespec ts;
+	clock_gettime(CLOCK_MONOTONIC, &ts);
+	return ((uint64_t)ts.tv_sec * 1000000000ULL + ts.tv_nsec);
+}
+
 /*
  * Batches are processed without holding any locks. The existence of the
  * items we're working on guarantees that the queue can't be destroyed out
@@ -587,16 +625,23 @@ queue_worker(void *dummy)
 	while (B_TRUE) {
 		count = assign_queue_and_get_work(&queue, batch);
 		if (count) {
+			size_t batch_cost = 0;
+			uint64_t start = time_now_ns();
 			zq_process_item_f *process =
 			    queue->zq_params.qp_process;
 			void *context = queue->zq_params.qp_context;
 			for (int i = 0; i < count; i++) {
 				process(batch[i]->qs_item, context);
+				batch_cost += batch[i]->qs_cost;
 			}
+			uint64_t nsec = time_now_ns() - start;
 			pthread_mutex_lock(&queue->zq_mutex);
 			for (int i = 0; i < count; i++) {
 				batch[i]->qs_completed = B_TRUE;
 			}
+			/* Collect processing rate data */
+			queue->zq_stats.cost_processed += batch_cost;
+			queue->zq_stats.nsec_used += nsec;
 			advance_indexes(queue);
 			pthread_mutex_unlock(&queue->zq_mutex);
 		}
@@ -621,7 +666,7 @@ maybe_wake_worker(void)
 }
 
 static inline struct timespec
-timeout_timespec(void)
+timeout_deadline(void)
 {
 	struct timespec expire;
 	struct timeval tv;
@@ -654,7 +699,7 @@ dispatch_worker(void *nope)
 	while (B_TRUE) {
 		while (!pool.tp_dispatch_requested) {
 			int rc;
-			struct timespec expire = timeout_timespec();
+			struct timespec expire = timeout_deadline();
 			rc = pthread_cond_timedwait(&pool.tp_request_dispatch,
 			    &pool.tp_dispatch_mutex, &expire);
 			if (rc == ETIMEDOUT) {
@@ -705,10 +750,7 @@ zstream_enqueue(zstream_queue_t *queue, queue_item_t *item)
 		advance_indexes(queue);
 
 #ifdef MONITOR_QUEUES
-	/* Maintain queue usage data per monitor interval */
-	uint64_t depth = queue->zq_ix.enqueue - queue->zq_ix.dequeue;
-	queue->zq_stats.max_depth = MAX(queue->zq_stats.max_depth, depth);
-	queue->zq_stats.min_depth = MIN(queue->zq_stats.min_depth, depth);
+	update_monitor_data(queue);
 #endif
 
 	pthread_mutex_unlock(&queue->zq_mutex);
@@ -740,7 +782,7 @@ zstream_queue_destroy(zstream_queue_t *queue)
 {
 	pthread_mutex_lock(&pool.tp_pool_mutex);
 
-#ifdef MONITOR_QUEUES
+#ifdef SHOW_BATCH_HISTOGRAMS
 	print_batch_size_histogram(queue);
 #endif
 
@@ -792,6 +834,11 @@ zstream_dequeue(zstream_queue_t *queue, queue_item_t *item)
 	}
 	queue_slot_t *slot = &Q_SLOT(queue, queue->zq_ix.dequeue);
 	queue->zq_ix.dequeue++;
+
+#ifdef MONITOR_QUEUES
+	update_monitor_data(queue);
+#endif
+
 	if (slot->qs_end_of_stream) {
 		pthread_mutex_unlock(&queue->zq_mutex);
 		/* Potential multi-dequeuer race point */
@@ -805,14 +852,16 @@ zstream_dequeue(zstream_queue_t *queue, queue_item_t *item)
 	}
 }
 
-#ifdef	MONITOR_QUEUES
-
-#define	USEC_PER_JIFFY		10000
-#define	SAMPLE_DURATION_USEC	1000000
-#define	CPU_FIELD_WIDTH		14
+#ifdef SHOW_BATCH_HISTOGRAMS
 
 /*
- * Called only during zstream_queue_destroy(), under the pool mutex
+ * Called only by zstream_queue_destroy(), under the pool mutex
+ *
+ * Prints one dense line per queue: the bucket counts for batch sizes 0
+ * through the largest batch observed. Now that ZQ_MAX_BATCH is 1024 and
+ * batches routinely reach the cap, that line can run to a thousand
+ * comma-separated values. Not very readable in a terminal, but still useful
+ * when plotted.
  */
 static void
 print_batch_size_histogram(zstream_queue_t *queue)
@@ -837,15 +886,66 @@ print_batch_size_histogram(zstream_queue_t *queue)
 	fflush(stderr);
 }
 
+#endif
+
+#ifdef MONITOR_QUEUES
+
+#define	USEC_PER_JIFFY		10000
+#define	SAMPLE_DURATION_USEC	1000000
+#define	CPU_FIELD_WIDTH		14
+
+static inline void
+update_depth_stats(depth_stats_t *ds, uint64_t depth)
+{
+	double delta = (double)depth - ds->ds_mean;
+	ds->ds_samples++;
+	ds->ds_mean += delta / (double)ds->ds_samples;
+	ds->ds_m2 += delta * ((double)depth - ds->ds_mean);
+}
+
+static inline double
+depth_stdev(const depth_stats_t *ds)
+{
+	if (ds->ds_samples < 2)
+		return (0.0);
+	return (sqrt(ds->ds_m2 / (double)(ds->ds_samples - 1)));
+}
+
+static inline void
+initialize_monitor_data(zstream_queue_t *queue)
+{
+	zq_stats_t *stats = &queue->zq_stats;
+	stats->depth = (depth_stats_t) {0};
+	stats->depth.min = INT_MAX;
+}
+
+/*
+ * Sample the current depth. Called on every enqueue and every dequeue, so
+ * the statistics are per-event rather than per-unit-time; a queue that sits
+ * full while nothing moves contributes no samples.
+ */
+static inline void
+update_monitor_data(zstream_queue_t *queue)
+{
+	uint64_t depth = queue->zq_ix.enqueue - queue->zq_ix.dequeue;
+
+	queue->zq_stats.depth.max = MAX(queue->zq_stats.depth.max, depth);
+	queue->zq_stats.depth.min = MIN(queue->zq_stats.depth.min, depth);
+	update_depth_stats(&queue->zq_stats.depth, depth);
+}
+
 /*
  * Monitor queue and CPU usage from a separate thread. This is all
- * Linux-specific, but it's needed only while tuning queue lengths and
- * batch sizes. Prints the minimum and maximum queue depth observed
- * during each period.
+ * Linux-specific. It's largely a development remnant. Now that queue depths
+ * are standardized and batch sizes are sized to BATCH_TIME_NSEC, there are
+ * relatively few levers that need tuning.
+ *
+ * For each period, prints the minimum and maximum queue depth observed,
+ * followed by the mean depth and its standard deviation.
  *
  * Example output:
  *
- *     CPU: 99.85%   Queue 0:  745-1024   Queue 1:  183-256
+ * CPU: 99.85% Queue 0: 745-4096 (2553 +/- 994) Queue 1: ...
  */
 static void *
 cpu_and_queue_monitor(void *dummy)
@@ -900,14 +1000,17 @@ cpu_and_queue_monitor(void *dummy)
 		for (int i = 0; i < pool.tp_num_queues; i++) {
 			zstream_queue_t *q = pool.tp_queues[i];
 			pthread_mutex_lock(&q->zq_mutex);
-			int min = q->zq_stats.min_depth;
-			int max = q->zq_stats.max_depth;
-			if (min > max)
-				min = max = 0;
-			fprintf(stderr, "Queue %d: %4d-%-4d   ",
-			    q->zq_id, min, max);
-			q->zq_stats.min_depth = INT_MAX;
-			q->zq_stats.max_depth = 0;
+			zq_stats_t *stats = &q->zq_stats;
+			double avg = stats->depth.ds_mean;
+			double stdev = depth_stdev(&stats->depth);
+			if (stats->depth.min > stats->depth.max)
+				stats->depth.min = stats->depth.max = 0;
+			const char *plusminus = "\xc2\xb1";  /* UTF-8 */
+			fprintf(stderr, "Queue %d: %4d-%-4d (%4d %s %-4d)   ",
+			    q->zq_id, stats->depth.min, stats->depth.max,
+			    (int)avg, plusminus, (int)stdev);
+			fflush(stderr);
+			initialize_monitor_data(q);
 			pthread_mutex_unlock(&q->zq_mutex);
 		}
 

cmd/zstream/zstream_queue.h

@@ -34,8 +34,8 @@ extern "C" {
  * they wish.
  *
  * The cost function assigns a size_t cost that estimates the amount of work
- * needed to process an item. For operations like hashing and data
- * compression, the natural cost is typically payload length.
+ * needed to process an item. Payload length is a natural cost basis for
+ * operations such as hashing and data compression.
  *
  * It's expected that only a subset of input items will require processing.
  * If an item's cost is 0, it is fast-tracked and never presented to the
@@ -43,28 +43,35 @@ extern "C" {
  *
  * The cost function is run as items enter the queue, with the queue mutex
  * held, so it should return a value promptly. If cost estimation is
- * expensive and important, use a separate queue to implement it.
+ * expensive and important, use a separate queue to implement it. Cost
+ * values should have a roughly linear relationship to processing time.
+ * Only the ratio between costs matters, so choose units that keep the sum
+ * of a batch's costs comfortably inside a size_t.
  *
- * Dispatch granularity is specified as a per-batch budget that is set for
- * each queue in the same (arbitrary) units used for item costs. Threads
- * claim items until the budget is met, there are no more items available,
- * or ZQ_MAX_BATCH items have been claimed. When claiming items to work on,
- * threads never block waiting for additional work to arrive. They start
- * work as quickly as possible even if the budget has not been reached.
- *
- * A batch budget of 0 means that all batches will have a size of 1.
+ * Work is processed in batches whose size is tuned automatically. Each
+ * queue observes how long its own work actually takes per unit of cost, and
+ * claims enough items to fill a fixed slice of processing time. No batch
+ * exceeds ZQ_MAX_BATCH items.
  *
  * All queues share a single thread pool that is managed to avoid
  * contention. Threads are assigned to queues dynamically according to where
  * work is available. When multiple queues have work, threads are allocated
- * among them stochastically with an eye toward preventing pipeline stalls.
+ * among them stochastically, in proportion to the amount of unclaimed work
+ * each queue is holding.
  *
  * The shared thread pool persists until the process exits.
  */
 
-#define	ZQ_MAX_BATCH	32	/* The most items that can be claimed at once */
-#define	ZQ_MAX_QUEUES	16	/* The maximum number of simultaneous queues */
-#define	ZQ_MIN_THREADS	6
+/*
+ * Implementation constants, exposed only so that tests and diagnostics can
+ * reason about the real boundaries. Except for ZQ_MIN_THREADS, these values
+ * are not caller-tunable. ZQ_MIN_THREADS can be overridden by calling
+ * zstream_queue_set_num_threads().
+ */
+#define	ZQ_SLOTS_PER_QUEUE	4096	/* Ring buffer slots, every queue */
+#define	ZQ_MAX_BATCH		1024	/* Most items claimable at once */
+#define	ZQ_MAX_QUEUES		16	/* Max number of simultaneous queues */
+#define	ZQ_MIN_THREADS		6
 
 typedef void queue_item_t;
 
@@ -98,8 +105,6 @@ typedef struct {
 	zq_estimate_cost_f	*qp_cost;
 	void			*qp_context;
 	size_t			qp_item_size;
-	size_t			qp_batch_budget;
-	size_t			qp_queue_length;
 } zq_params_t;
 
 zstream_queue_t *

cmd/zstream/zstream_recompress.c

@@ -179,11 +179,15 @@ chain_compress_writes(queue_item_t *item_in, void *context_in)
 }
 
 /*
- * A cost of zero waives processing for the current item. If we want to
- * process it, the cost will always be item->dp_payload_size. So in these
- * two cost functions, we're mostly determining which packets need
- * attention. A packet that's already compressed with the target compression
- * profile can be ignored.
+ * A cost of zero waives processing for the current item, so the following
+ * two functions are mostly determining which packets need attention. A
+ * packet that's already compressed with the target compression profile can
+ * be ignored.
+ *
+ * When a packet does need work, the cost is the number of bytes the
+ * compressor or decompressor will have to process: the logical size for
+ * compression, since that's what goes in, and the stored payload size for
+ * decompression, since that's what comes out.
  */
 static size_t
 chain_compress_cost(queue_item_t *item_in, void *context_in)
@@ -245,8 +249,6 @@ parallel_decompress_writes(compression_spec_t *target)
 	    .cs_out_size = sizeof (drr_packet_t),
 	    .cs_context = context,
 	    .cs_parallel = {
-		.queue_length = 256,
-		.batch_budget = 256 * 1024,
 		.process = chain_decompress_writes,
 		.cost = chain_decompress_cost
 	    }
@@ -269,8 +271,6 @@ parallel_compress_writes(compression_spec_t *target)
 	    .cs_out_size = sizeof (drr_packet_t),
 	    .cs_context = context,
 	    .cs_parallel = {
-		.queue_length = 1024,
-		.batch_budget = 32 * 1024,
 		.process = chain_compress_writes,
 		.cost = chain_compress_cost
 	    }

cmd/zstream/zstream_selftest_queue.c

@@ -27,23 +27,31 @@
  * Every item carries enough information to be verified independently:
  *
  * - The tuple (qi_producer, qi_seq) identifies each item; the consumer
- *   checks that each producer's items arrive in the same order they
- *   were enqueued.
+ * checks that each producer's items arrive in the same order they were
+ * enqueued.
  *
  * - qi_check is a hash of (qi_seed, qi_producer, qi_seq). The processing
- *   function verifies it and then XORs in TRANSFORM_MAGIC. The consumer
- *   checks that the transform happened iff cost > 0.
+ * function verifies it and then XORs in TRANSFORM_MAGIC. The consumer
+ * checks that the transform happened iff cost > 0.
  *
  * - qi_pattern[] is filled from qi_check and verified both by the process
- *   function and the consumer, to catch any corruption of the shallow
- *   copies in and out of the ring buffer.
+ * function and the consumer, to catch any corruption of the shallow copies
+ * in and out of the ring buffer.
  *
- * - qi_process_count counts invocations of the process function, which
- *   must be exactly one for cost > 0 items and zero for cost == 0 items.
+ * - qi_process_count counts invocations of the process function, which must
+ * be exactly one for cost > 0 items and zero for cost == 0 items.
  *
  * Global conservation checks: the number of items dequeued must equal the
  * number enqueued, and the total number of process-function invocations
  * must equal the number of nonzero-cost items enqueued.
+ *
+ * Costs are not passive values. Each queue measures how long its own work
+ * takes per unit of cost and sizes its batches accordingly. So, a
+ * workload's cost distribution decides how the implementation will behave.
+ * Configs can set qc_ns_per_cost to make processing time genuinely
+ * proportional to cost (the relationship the queue assumes), or leave it at
+ * zero to get delays unrelated to cost. Both are worth testing; see
+ * queue_batch_tuning().
  */
 
 #include <assert.h>
@@ -54,6 +62,7 @@
 #include <stdint.h>
 #include <stdio.h>
 #include <string.h>
+#include <sys/param.h>
 #include <unistd.h>
 
 #include "zstream_queue.h"
@@ -82,13 +91,12 @@ typedef struct {
 typedef struct {
 	uint32_t	qc_producers;		/* Number of producers */
 	uint64_t	qc_items;		/* Items per producer */
-	size_t		qc_queue_length;
-	size_t		qc_batch_budget;
 	size_t		qc_pattern_len;		/* Extra payload bytes */
 	uint32_t	qc_zero_cost_pct;	/* % of items fast-tracked */
 	size_t		qc_max_cost;		/* Nonzero costs are 1..max */
 	uint32_t	qc_delay_pct;		/* % of items slept on */
 	uint32_t	qc_max_delay_us;
+	uint32_t	qc_ns_per_cost;		/* Delay of cost * this, ns */
 	uint32_t	qc_producer_stall_pct;	/* % chance producer naps */
 	uint32_t	qc_consumer_stall_pct;	/* % chance consumer naps */
 	uint32_t	qc_stall_max_us;
@@ -203,7 +211,19 @@ qtest_producer(void *arg)
 			local_expect++;
 		}
 
-		if (item->qi_cost > 0 && cfg->qc_max_delay_us > 0) {
+		if (item->qi_cost > 0 && cfg->qc_ns_per_cost > 0) {
+			/*
+			 * Make processing time proportional to cost, which
+			 * is the relationship the queue's batch sizing
+			 * assumes. Costs here are small enough that the
+			 * product can't overflow, but clamp anyway so that
+			 * a future config can't turn a tuning test into a
+			 * multi-second sleep.
+			 */
+			uint64_t ns = (uint64_t)item->qi_cost *
+			    cfg->qc_ns_per_cost;
+			item->qi_delay_us = MIN(ns / 1000, 100000);
+		} else if (item->qi_cost > 0 && cfg->qc_max_delay_us > 0) {
 			if (selftest_rng_below(&rng, 1000) <
 			    LONG_DELAYS_PER_THOUSAND) {
 				item->qi_delay_us = cfg->qc_max_delay_us *
@@ -312,9 +332,7 @@ run_queue_workloads(const qtest_config_t *cfgs, int ncfg)
 			.qp_cost = qtest_cost,
 			.qp_context = &runs[i],
 			.qp_item_size =
-			    sizeof (qtest_item_t) + cfgs[i].qc_pattern_len,
-			.qp_batch_budget = cfgs[i].qc_batch_budget,
-			.qp_queue_length = cfgs[i].qc_queue_length,
+			    sizeof (qtest_item_t) + cfgs[i].qc_pattern_len
 		};
 		runs[i].qr_queue = zstream_queue_create(&params);
 	}
@@ -355,8 +373,6 @@ queue_basic(void)
 	qtest_config_t cfg = {
 		.qc_producers = 1,
 		.qc_items = 5000,
-		.qc_queue_length = 64,
-		.qc_batch_budget = 256,
 		.qc_pattern_len = 32,
 		.qc_zero_cost_pct = 20,
 		.qc_max_cost = 64,
@@ -366,9 +382,11 @@ queue_basic(void)
 
 /*
  * A long, randomized stream with heavy-tailed processing delays, a large
- * fraction of fast-tracked items, costs that exceed the batch budget, and a
- * consumer that periodically stalls so the queue backs up and enqueue
- * blocks on Q_FULL. The ring indices wrap hundreds of times.
+ * fraction of fast-tracked items, and a consumer that periodically stalls.
+ * The producer outruns the consumer badly enough that the queue sits at or
+ * near ZQ_SLOTS_PER_QUEUE for most of the run, so this is the main exercise
+ * of Q_FULL and of enqueuers blocking on the dequeued condition. 100000
+ * items wrap the ring indices a couple of dozen times.
  */
 static void
 queue_torture(void)
@@ -376,8 +394,6 @@ queue_torture(void)
 	qtest_config_t cfg = {
 		.qc_producers = 1,
 		.qc_items = 100000,
-		.qc_queue_length = 512,
-		.qc_batch_budget = 2048,
 		.qc_pattern_len = 64,
 		.qc_zero_cost_pct = 30,
 		.qc_max_cost = 4096,
@@ -391,36 +407,50 @@ queue_torture(void)
 }
 
 /*
- * Off-by-one hunting: sweep the degenerate corners of queue length,
- * batch budget, and stream length, including a zero-item stream and
+ * Off-by-one hunting: sweep stream lengths that land on and adjacent to
+ * every boundary the implementation has, including a zero-item stream and
  * zero-length payloads.
+ *
+ * Queue depth and batch size are no longer caller-supplied, so the
+ * boundaries worth probing are the implementation's own: ZQ_MAX_BATCH, the
+ * point at which claim_batch() stops filling a batch, and
+ * ZQ_SLOTS_PER_QUEUE, the point at which the ring index wraps and Q_FULL
+ * can trip. Both are exported by zstream_queue.h for exactly this purpose,
+ * so this sweep tracks them automatically if either one is retuned.
+ *
+ * A stalling consumer is used for the counts at or above
+ * ZQ_SLOTS_PER_QUEUE. Without it the consumer keeps up, the queue never
+ * approaches full, and the wrap and Q_FULL paths go untested no matter how
+ * many items are sent.
  */
 static void
 queue_edge_cases(void)
 {
-	static const size_t lengths[] =
-	    { 1, 2, ZQ_MAX_BATCH - 1, ZQ_MAX_BATCH, ZQ_MAX_BATCH + 1, 64 };
-	static const size_t budgets[] = { 0, 1, 16, SIZE_MAX / 2 };
+	static const uint64_t counts[] = {
+		0, 1, 2, 3,
+		ZQ_MAX_BATCH - 1, ZQ_MAX_BATCH, ZQ_MAX_BATCH + 1,
+		2 * ZQ_MAX_BATCH,
+		ZQ_SLOTS_PER_QUEUE - 1, ZQ_SLOTS_PER_QUEUE,
+		ZQ_SLOTS_PER_QUEUE + 1, 2 * ZQ_SLOTS_PER_QUEUE + 3
+	};
+	const int ncounts = sizeof (counts) / sizeof (counts[0]);
 	uint64_t stream = 200;
 
-	for (int l = 0; l < 6; l++) {
-		for (int b = 0; b < 4; b++) {
-			uint64_t counts[] = { 0, lengths[l], lengths[l] + 1,
-			    4 * lengths[l] + 3 };
-			for (int n = 0; n < 4; n++) {
-				qtest_config_t cfg = {
-					.qc_producers = 1,
-					.qc_items = counts[n],
-					.qc_queue_length = lengths[l],
-					.qc_batch_budget = budgets[b],
-					.qc_pattern_len =
-					    (lengths[l] & 1) ? 0 : 24,
-					.qc_zero_cost_pct = 25,
-					.qc_max_cost = 8,
-					.qc_rng_stream = stream++,
-				};
-				run_queue_workload(&cfg);
-			}
+	for (int n = 0; n < ncounts; n++) {
+		boolean_t big = counts[n] >= ZQ_SLOTS_PER_QUEUE;
+
+		for (int p = 0; p < 2; p++) {
+			qtest_config_t cfg = {
+				.qc_producers = 1,
+				.qc_items = counts[n],
+				.qc_pattern_len = p ? 24 : 0,
+				.qc_zero_cost_pct = 25,
+				.qc_max_cost = 8,
+				.qc_consumer_stall_pct = big ? 1 : 0,
+				.qc_stall_max_us = big ? 200 : 0,
+				.qc_rng_stream = stream++,
+			};
+			run_queue_workload(&cfg);
 		}
 	}
 }
@@ -438,8 +468,6 @@ queue_zero_cost(void)
 	qtest_config_t cfg = {
 		.qc_producers = 1,
 		.qc_items = 20000,
-		.qc_queue_length = 128,
-		.qc_batch_budget = 1024,
 		.qc_pattern_len = 16,
 		.qc_zero_cost_pct = 100,
 		.qc_max_cost = 8,
@@ -448,6 +476,73 @@ queue_zero_cost(void)
 	run_queue_workload(&cfg);
 }
 
+/*
+ * Batch sizing is derived from each queue's own measured ns-per-unit-cost,
+ * so the cost values reported by a caller now steer the implementation rather
+ * than just gating the fast track. This test runs four queues at once whose
+ * cost-to-time relationships are deliberately dissimilar and checks that
+ * all of them still deliver every item exactly once and in order.
+ *
+ * The four cases, in the order configured below:
+ *
+ * - Faithful. Processing time really is proportional to cost, which is
+ *   what the model assumes. Costs average ~2048 at 250ns each, putting a
+ *   typical item just past the 400us target on its own, so batches come
+ *   out at one or two items.
+ *
+ * - Too slow to batch. Every item costs 1 but takes 600us, so the implied
+ *   budget is a fraction of a cost unit and has to be clamped up to 1.
+ *   Batches should be single items; a rounding error that let the budget
+ *   reach 0 would spin claim_batch() without claiming anything.
+ *
+ * - Too fast to measure. Costs are enormous and the work is nothing, so
+ *   the implied budget overflows anything a size_t can hold and has to
+ *   saturate. Sums of these costs wrap inside both claim_batch() and the
+ *   queue's running total, which is allowed to produce silly batch sizes
+ *   but must not produce wrong answers.
+ *
+ * - Uniform cost. Every item costs the same, the degenerate case for a
+ *   ratio-based estimate.
+ */
+static void
+queue_batch_tuning(void)
+{
+	qtest_config_t cfgs[] = {
+		{
+			.qc_producers = 2,
+			.qc_items = 1500,
+			.qc_pattern_len = 16,
+			.qc_zero_cost_pct = 10,
+			.qc_max_cost = 4096,
+			.qc_ns_per_cost = 250,
+			.qc_rng_stream = 600,
+		}, {
+			.qc_producers = 1,
+			.qc_items = 400,
+			.qc_pattern_len = 8,
+			.qc_zero_cost_pct = 5,
+			.qc_max_cost = 1,
+			.qc_ns_per_cost = 600 * 1000,
+			.qc_rng_stream = 610,
+		}, {
+			.qc_producers = 2,
+			.qc_items = 5000,
+			.qc_pattern_len = 32,
+			.qc_zero_cost_pct = 20,
+			.qc_max_cost = SIZE_MAX / 2,
+			.qc_rng_stream = 620,
+		}, {
+			.qc_producers = 1,
+			.qc_items = 20000,
+			.qc_pattern_len = 0,
+			.qc_zero_cost_pct = 0,
+			.qc_max_cost = 1,
+			.qc_rng_stream = 630,
+		}
+	};
+	run_queue_workloads(cfgs, sizeof (cfgs) / sizeof (cfgs[0]));
+}
+
 /*
  * Eight producer threads hammering one queue with random pacing. The
  * consumer verifies per-producer FIFO order and exact counts.
@@ -458,8 +553,6 @@ queue_multi_producer(void)
 	qtest_config_t cfg = {
 		.qc_producers = 8,
 		.qc_items = 15000,
-		.qc_queue_length = 256,
-		.qc_batch_budget = 512,
 		.qc_pattern_len = 24,
 		.qc_zero_cost_pct = 25,
 		.qc_max_cost = 512,
@@ -486,9 +579,6 @@ queue_multi_queue(void)
 		qtest_config_t cfg = {
 			.qc_producers = producers,
 			.qc_items = 4000 / producers,
-			.qc_queue_length = (size_t)4 << (i % 6),
-			.qc_batch_budget =
-			    (i % 4 == 0) ? 0 : (size_t)64 << (i % 5),
 			.qc_pattern_len = 8 * (i % 5),
 			.qc_zero_cost_pct = 10 * (i % 6),
 			.qc_max_cost = (size_t)16 << (i % 8),
@@ -518,16 +608,19 @@ queue_stress(void)
 
 		for (int i = 0; i < nqueues; i++) {
 			uint32_t producers = 1 + selftest_rng_below(&rng, 4);
+			/*
+			 * A quarter of the queues get processing time tied
+			 * to cost, so the batch-size estimator sees a mix of
+			 * well-behaved and meaningless cost data.
+			 */
+			uint32_t ns_per_cost =
+			    (selftest_rng_below(&rng, 4) == 0) ?
+			    1 + selftest_rng_below(&rng, 400) : 0;
 			qtest_config_t cfg = {
 				.qc_producers = producers,
 				.qc_items = (2000 +
 				    selftest_rng_below(&rng, 8000)) /
 				    producers,
-				.qc_queue_length = (size_t)1 <<
-				    selftest_rng_below(&rng, 10),
-				.qc_batch_budget =
-				    (selftest_rng_below(&rng, 3) == 0) ? 0 :
-				    selftest_rng_below(&rng, 4096),
 				.qc_pattern_len =
 				    selftest_rng_below(&rng, 64),
 				.qc_zero_cost_pct =
@@ -537,6 +630,7 @@ queue_stress(void)
 				.qc_delay_pct = selftest_rng_below(&rng, 4),
 				.qc_max_delay_us =
 				    selftest_rng_below(&rng, 120),
+				.qc_ns_per_cost = ns_per_cost,
 				.qc_producer_stall_pct =
 				    selftest_rng_below(&rng, 2),
 				.qc_consumer_stall_pct =
@@ -556,6 +650,7 @@ const test_case_t selftest_queue_cases[] = {
 	{ "queue_basic",		queue_basic },
 	{ "queue_edge_cases",		queue_edge_cases },
 	{ "queue_zero_cost",		queue_zero_cost },
+	{ "queue_batch_tuning",		queue_batch_tuning },
 	{ "queue_torture",		queue_torture },
 	{ "queue_multi_producer",	queue_multi_producer },
 	{ "queue_multi_queue",		queue_multi_queue },