zstream: multithreading

This PR extends the `zstream_chain` mechanism introduced in #18509 to
include support for multithreading. It makes three main changes.

- It adds `zstream_queue.[ch]`, a generic FIFO queue with multiple
  worker threads. This is a freestanding construct not directly tied to
  ZFS or to the `zstream_chain` mechanism.

- It adapts `zstream_chain.[ch]` to allow both single-threaded and
  multithreaded steps.

- It converts `zstream_fletcher4.[ch]` and `zstream_recompress.[ch]`
  to use multithreading.

### Motivation

This patch significantly speeds recompression on modern CPUs. The
ultimate goal is to add a `zstream dedup` or `zstream pack` subcommand
that exploits ZFS's knowledge of streams' structure to compress them
efficiently for backup or archival purposes. See #18195.

### `zstream_queue`

Callers define a fixed item size to be used by each queue and supply two
thread-safe functions that 1) estimate individual items' processing
cost and 2) perform the actual processing.

The cost function assigns a `size_t` cost that estimates the amount of
work needed to process each item. For typical operations such as
hashing, payload length is a natural definition of cost. Items with an
assigned cost of 0 are assumed to need no work and are not presented to
the processing function. Thread dispatch granularity is specified as a
per-batch budget that is set for each queue in the same
(arbitrary) units used for item costs.

All queues share a single fixed-size thread pool that is managed to
avoid contention among queues. Threads are allocated to queues
dynamically. When multiple queues have work, threads are allocated
among them stochastically with an eye toward preventing pipeline
stalls.

### Changes to `zstream_chain`

From callers' perspective, the only change is that `chain_step_t` gains
a `CS_PARALLEL` option. By convention, functions that set up parallel
steps are named `parallel_*()` rather than `serial_*()` to indicate
this difference.

Execution changes somewhat, since it is no longer the default to
shepherd each item through the chain individually. (That behavior
remains available for debugging, however, and can be requested by
calling `zstream_chain_exec_serialized()` instead of
`zstream_chain_exec()`.)

On execution, the chain is divided into segments bounded by parallel
steps, and a worker thread is assigned to each segment. Each parallel
queue is double-covered, with one thread enqueuing there and one
thread dequeuing. Since `zstream_queue` emits items in FIFO order, the
mechanism is essentially similar to that of the original chain
implementation, but it allows for asynchronous and out-of-order work.
Load balancing and avoidance of pipeline stalls is delegated to
`zstream_queue`, which has visibility into all active queues.

### Fletcher 4 checksums

Fletcher 4 calculation uses only addition and is quite fast. However,
because of [Amdahl's Law](https://en.wikipedia.org/wiki/Amdahl%27s_law),
it's necessary to parallelize Fletcher 4 calculations to unlock the
potential of multithreading for other operations. (Keep in mind that
checksums are typically calculated twice: once on input and once on
output.)

Fletcher 4 operations are now split into a parallel portion that
pre-calculates checksums for payloads and a serial portion that
calculates record checksums and reassembles a consistent stream
checksum using data from the parallel step.

There's a limit of 8MB on the length of any given Fletcher 4 segment
because of the risk of addition overflow. See the comments in
`zfs_fletcher.c` for details. The parallel code is aware of this limit
and subdivides long payloads appropriately.

### Recompression

Decompression and recompression are handled as separate parallel steps,
with the `zstream_queues`' cost functions deciding whether any given
WRITE record requires attention. But that's essentially how the current
serial version already works.

I added a `-t num_threads` option to `zstream recompress`. This is the
only user-facing change. Without that option, `zstream_queue` spawns
one thread per CPU core, which is an aggressive default.

I looked into more architecture-aware ways of selecting a default thread
count, but given SMT/hyperthreading, differences between performance
and power-efficient cores, and processors that use various strategies
for sharing L2 cache among cores, it's difficult to do anything more
sensible without adding significant complexity. If one thread per CPU
is too aggressive, I'd suggest we just default to #cores - N for some
value of N.

### Testing

These changes are covered by the existing `zstream` test suite, which
validates both checksum calculation and recompression. In particular,
there is already a "large payload checksum" test from #18509 that was
added in anticipation of this PR.

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

cmd/zstream/Makefile.am

@@ -23,6 +23,8 @@ zstream_SOURCES = \
 	%D%/zstream_recompress.h \
 	%D%/zstream_redup.c \
 	%D%/zstream_token.c \
+	%D%/zstream_queue.c \
+	%D%/zstream_queue.h \
 	%D%/zstream_util.c \
 	%D%/zstream_util.h \
 	%D%/zstream_validate.c \

cmd/zstream/zstream.c

@@ -39,7 +39,7 @@ zstream_usage(void)
 	    "\n"
 	    "\tzstream drop_record [-v] [OBJECT,OFFSET] ...\n"
 	    "\n"
-	    "\tzstream recompress [ -l level] TYPE\n"
+	    "\tzstream recompress [-t num_threads] [-l level] TYPE\n"
 	    "\n"
 	    "\tzstream token resume_token\n"
 	    "\n"

cmd/zstream/zstream_byteswap.c

@@ -40,8 +40,11 @@ static byteswap_context_t byteswap_contexts[MAX_BYTESWAP];
 static int next_context = 0;
 
 static disposition_t
-chain_byteswap(drr_packet_t *item, byteswap_context_t *context)
+chain_byteswap(void *item_in, void *context_in)
 {
+	drr_packet_t *item = (drr_packet_t *)item_in;
+	byteswap_context_t *context = (byteswap_context_t *)context_in;
+
 	if (item == NULL) {
 		return (D_OK);
 	}
@@ -191,7 +194,7 @@ serial_byteswap(byteswap_stage_t stage)
 		.cs_out_size = sizeof (drr_packet_t),
 		.cs_context = bsc,
 		.cs_serial = {
-			.process = (zc_serial_process_f *)chain_byteswap,
+			.process = chain_byteswap,
 		}
 	};
 	return (step);

cmd/zstream/zstream_chain.c

@@ -17,9 +17,12 @@
  * Copyright (c) 2026 by Garth Snyder. All rights reserved.
  */
 
+#include <time.h>
 #include <assert.h>
 #include <err.h>
 #include <libspl.h>
+#include <pthread.h>
+#include <stddef.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <sys/abd.h>
@@ -27,34 +30,40 @@
 #include <sys/stdtypes.h>
 #include <sys/zio.h>
 #include <sys/zstd/zstd.h>
+#include <sys/zfs_refcount.h>
 #include <zfs_fletcher.h>
 
 #include "zstream_chain.h"
+#include "zstream_queue.h"
 
 #define	MAX_CHAIN_LENGTH 32
 
+/*
+ * Calculated information about a chain. Not to be confused with
+ * chain_attrs_t, which is a global set of options and stream attributes
+ * available to all chain steps.
+ */
+typedef struct {
+	int	ct_num_steps;
+	int	ct_num_queues;
+	size_t	ct_item_size;
+} chain_stats_t;
+
+/*
+ * Data passed to worker threads
+ */
+typedef struct {
+	chain_step_t	*wc_steps;
+	int		wc_num_steps;
+	size_t		wc_buffer_size;
+	zstream_queue_t	*wc_in_queue;
+	zstream_queue_t	*wc_out_queue;
+} worker_context_t;
+
+typedef void *chain_worker_f(void *);
+
 chain_attrs_t *chain_attrs;
 
-static disposition_t
-chain_null_step(void *item, void *context)
-{
-	(void) item;
-	(void) context;
-	return (D_OK);
-}
-
-chain_step_t
-serial_null_step(void)
-{
-	chain_step_t step = {
-		.cs_type = CS_SERIAL,
-		.cs_serial = {
-		    .process = (zc_serial_process_f *)chain_null_step
-		}
-	};
-	return (step);
-}
-
 chain_step_t
 chain_terminator(void)
 {
@@ -85,67 +94,241 @@ libraries_fini(void)
 }
 
 /*
- * Execute a chain of serial processing steps.
+ * Body function for worker threads
+ */
+static void *
+zstream_chain_worker(worker_context_t *ctxt)
+{
+	uint8_t buffer[ctxt->wc_buffer_size];
+	boolean_t done = B_FALSE;
+
+	while (!done) {
+		for (int i = 0; i < ctxt->wc_num_steps; i++) {
+			chain_step_t *step = &ctxt->wc_steps[i];
+			if (step->cs_type == CS_SERIAL) {
+				if (done) {
+					(void) step->cs_serial.process(NULL,
+					    step->cs_context);
+				} else {
+					disposition_t dispo =
+					    step->cs_serial.process(buffer,
+					    step->cs_context);
+					if (dispo == D_EOF) {
+						done = B_TRUE;
+					} else if (dispo == D_DROP) {
+						break;
+					}
+				}
+			} else if (i == 0) {
+				done = done ||
+				    !zstream_dequeue(ctxt->wc_in_queue, buffer);
+			} else if (done) {
+				zstream_queue_fini(ctxt->wc_out_queue);
+			} else {
+				zstream_enqueue(ctxt->wc_out_queue, buffer);
+			}
+		}
+	}
+	return (NULL);
+}
+
+/*
+ * Validate chain and calculate number of steps, max packet size, and number
+ * of zstream_queues that must be created.
+ */
+static chain_stats_t
+validate_chain(zstream_chain_t chain)
+{
+	int num_steps = 0;
+	int num_queues = 0;
+	size_t item_size = 0;
+
+	while (chain[num_steps].cs_type != CS_TERMINATE) {
+		if (num_steps >= MAX_CHAIN_LENGTH) {
+			errx(1, "unterminated zstream_chain");
+		}
+		chain_step_t *step = &chain[num_steps];
+		item_size = MAX(item_size, step->cs_out_size);
+		if (step->cs_type == CS_PARALLEL) {
+			num_queues++;
+		}
+		num_steps++;
+	}
+	VERIFY3U(num_steps, >, 0);
+
+	boolean_t first_parallel = chain[0].cs_type == CS_PARALLEL;
+	boolean_t last_parallel = chain[num_steps-1].cs_type == CS_PARALLEL;
+	if (first_parallel || last_parallel) {
+		errx(1, "a chain cannot start or end with a parallel step");
+	}
+
+	/*
+	 * Check for consistency of input and output packet sizes in
+	 * adjacent steps.
+	 */
+	for (int i = 0; i < num_steps; i++) {
+		if (i > 0 && chain[i].cs_in_size != chain[i-1].cs_out_size) {
+			warnx("adjacent chain steps %d and %d declare "
+			    "incompatible packet sizes", i - 1, i);
+		}
+	}
+
+	chain_stats_t stats = {
+	    .ct_num_steps = num_steps,
+	    .ct_num_queues = num_queues,
+	    .ct_item_size = item_size
+	};
+	return (stats);
+}
+
+/*
+ * Execute a chain of processing steps, some parallel and some serial.
  *
  * For simplicity, we normalize the chain item size to that of the largest
- * output of any step. Packets with data beyond the base drr_record_t should
- * add their additional data to the end of the packet, and this area may be
- * reused for different purposes as items travel down the chain.
+ * output of any step. Payloads are allocated on the heap, so the maximal
+ * item size is typically on the order of 512 bytes.
  *
- * Each item traverses the entire chain before the next item is read.
+ * Packets with data beyond the base drr_record_t should add their
+ * additional data to the end of the packet, and this area may be reused for
+ * different purposes as items travel down the chain.
+ *
+ * One worker thread is assigned to every contiguous sequence of serial
+ * steps plus the parallel steps on either side of that block (if any).
+ * Adjacent parallel steps also receive a worker. This isn't a special case,
+ * it's just the same rule with the serial block consisting of zero steps.
+ *
+ * Parallel steps are double-covered, which is the intended behavior. If a
+ * worker's domain begins with a parallel step, it dequeues items from the
+ * associated queue. If the domain ends with a parallel step, it submits
+ * items to that queue.
  */
 void
 zstream_chain_exec(zstream_chain_t chain, chain_attrs_t *attrs)
 {
-	int num_steps = 0;
-	size_t packet_size = 0;
-	chain_attrs_t backup_attrs = {0};
+	chain_stats_t stats = validate_chain(chain);
 
-	chain_attrs = attrs ? attrs : &backup_attrs;
-
-	while (chain[num_steps].cs_type != CS_TERMINATE) {
-		packet_size = MAX(packet_size, chain[num_steps].cs_out_size);
-		num_steps++;
-		if (num_steps >= MAX_CHAIN_LENGTH) {
-			errx(1, "unterminated zstream_chain");
-		}
-	}
-	VERIFY3U(num_steps, >, 0);
+	int num_workers = stats.ct_num_queues + 1;
+	worker_context_t contexts[num_workers];
+	pthread_t worker_threads[num_workers];
+	zstream_queue_t *queue;
 
 	/*
-	 * Check for consistency of input and output packet sizes in
-	 * adjacent steps. A declared packet size of zero waives this check.
+	 * Create parallel queues and worker thread contexts
+	 *
+	 * We do not need to track zstream_queues independently of worker
+	 * contexts because queues clean themselves up once the last item
+	 * has been dequeued. The stream eventually ends, so some worker
+	 * thread will eventually call zstream_queue_fini() on every queue.
 	 */
-	for (int i = 0; i < num_steps; i++) {
-		boolean_t mismatch = i > 0 &&
-		    chain[i].cs_in_size != 0 &&
-		    chain[i-1].cs_out_size != 0 &&
-		    chain[i].cs_in_size != chain[i-1].cs_out_size;
-		if (mismatch) {
-			warnx("note - chain steps %d and %d have "
-			    "mismatched packet sizes", i - 1, i);
+
+	int worker = 0;
+
+	worker_context_t context = {
+	    .wc_steps = chain,
+	    .wc_num_steps = 0,
+	    .wc_buffer_size = stats.ct_item_size,
+	    .wc_in_queue = NULL
+	};
+	contexts[worker] = context;
+
+	for (int i = 0; i < stats.ct_num_steps; i++) {
+		contexts[worker].wc_num_steps++;
+		if (chain[i].cs_type == CS_PARALLEL) {
+			chain_step_t *cs = &chain[i];
+			zq_params_t queue_params = {
+				.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);
+			contexts[worker].wc_out_queue = queue;
+			worker++;
+			worker_context_t next_context = {
+			    .wc_steps = cs,
+			    .wc_num_steps = 1,
+			    .wc_buffer_size = stats.ct_item_size,
+			    .wc_in_queue = queue
+			};
+			contexts[worker] = next_context;
 		}
 	}
 
+	contexts[worker].wc_out_queue = NULL;
+
+	chain_attrs_t backup_attrs = {0};
+	chain_attrs = attrs ? attrs : &backup_attrs;
+
 	libraries_init();
 
-	uint8_t buffer[packet_size];
+	/* Spawn threads */
+	for (int i = 0; i < num_workers; i++) {
+		char buff[32];
+		int ret = pthread_create(&worker_threads[i], NULL,
+		    (chain_worker_f *)zstream_chain_worker,
+		    &contexts[i]);
+		VERIFY3S(ret, ==, 0);
+		snprintf(buff, sizeof (buff), "chain-%d", i);
+		pthread_setname_np(worker_threads[i], buff);
+	}
+
+	/* Reap threads */
+	for (int i = 0; i < num_workers; i++) {
+		int ret = pthread_join(worker_threads[i], NULL);
+		VERIFY3S(ret, ==, 0);
+	}
+
+	libraries_fini();
+}
+
+/*
+ * Execute a chain linearly, without queues and without multithreading. This
+ * form of execution is intended as a debugging aid, both for clients and
+ * for the chain mechanism itself. If this variant doesn't produce results
+ * identical to zstream_chain_exec(), there's a multithreading-related bug
+ * somewhere.
+ *
+ * It is not necessary to remove parallel steps from the input chain. They
+ * are accepted as-is, but their execution won't be parallelized.
+ */
+void
+zstream_chain_exec_serialized(zstream_chain_t chain, chain_attrs_t *attrs)
+{
+	chain_stats_t stats = validate_chain(chain);
+
+	uint8_t buffer[stats.ct_item_size];
 	boolean_t done = B_FALSE;
 
+	chain_attrs_t backup_attrs = {0};
+	chain_attrs = attrs ? attrs : &backup_attrs;
+
+	libraries_init();
+
 	while (!done) {
-		for (int i = 0; i < num_steps; i++) {
-			if (done) {
-				(void) chain[i].cs_serial.process(NULL,
-				    chain[i].cs_context);
-			} else {
-				disposition_t dispo =
-				    chain[i].cs_serial.process(buffer,
-				    chain[i].cs_context);
-				if (dispo == D_EOF) {
-					VERIFY0(i);
-					done = B_TRUE;
-				} else if (dispo == D_DROP) {
-					break;
+		for (int i = 0; i < stats.ct_num_steps; i++) {
+			chain_step_t *step = &chain[i];
+			if (step->cs_type == CS_SERIAL) {
+				if (done) {
+					(void) step->cs_serial.process(NULL,
+					    step->cs_context);
+				} else {
+					disposition_t dispo =
+					    step->cs_serial.process(buffer,
+					    step->cs_context);
+					if (dispo == D_EOF) {
+						done = B_TRUE;
+					} else if (dispo == D_DROP) {
+						break;
+					}
+				}
+			} else if (!done) {
+				size_t cost = step->cs_parallel.cost(buffer,
+				    step->cs_context);
+				if (cost > 0) {
+					step->cs_parallel.process(buffer,
+					    step->cs_context);
 				}
 			}
 		}

cmd/zstream/zstream_chain.h

@@ -28,6 +28,8 @@ extern "C" {
 #include <stdint.h>
 #include <sys/zfs_ioctl.h>
 
+#include "zstream_queue.h"
+
 /*
  * A chain is a linear series of steps that process packets of data. It's
  * designed to modularize common functionality, reduce code duplication, and
@@ -36,16 +38,17 @@ extern "C" {
  * Some terms:
  *
  * **STEP** - A chain_step_t struct that represents a packet-processing
- * module and any arguments or context that it needs. Chain modules
- * generally define a function named serial_* that produces a chain_step_t
- * that can be incorporated directly into a chain.
+ * module and any arguments or context that it needs. Chain participants
+ * generally define a function named serial_xxx() or parallel_xxx() that
+ * produces a chain_step_t that can be incorporated directly into a chain.
  *
  * **CHAIN** - An array of chain_step_t's. It's just data, so you can create
  * the array however you like. But normally you'd just declare the whole
- * thing:
+ * chain at once, e.g.:
  *
  *	zstream_chain_t dump_chain = {
  *		serial_read_stream(infile),
+ *  		parallel_calc_fletcher4(1024),
  *		serial_validate_fletcher4(),
  *		serial_byteswap(BS_INCOMING),
  *		serial_validate_records(),
@@ -73,20 +76,26 @@ extern "C" {
  * **PROCESSING FUNCTION** - Each step names a processing function that does
  * the actual work of transforming an input buffer into an output buffer.
  * The transformation happens in place, in a single buffer provided by the
- * chain.
+ * chain. (Steps that run in parallel must also identify a cost function; see
+ * zstream_queue.h.)
  *
- * The processing function should return a disposition_t, normally D_OK. A
- * function can return D_DROP to remove an item from the stream entirely. It
- * can also return D_EOF to indicate that no more data will be forthcoming.
- * However, only the first step in the chain should ever return D_EOF.
+ * The processing function for a serial step should return a disposition_t,
+ * normally D_OK. A processing function can return D_DROP to remove an item
+ * from the stream entirely. It can also return D_EOF to indicate that no
+ * more data will be forthcoming. However, only the first step in the chain
+ * should ever return D_EOF.
  *
- * Functions are called with a NULL packet pointer when the end of the
- * stream passes by them.
+ * Serial functions are called with a NULL packet when the end of the
+ * stream passes by them. Since parallel functions may see packets in any
+ * order, they have no concept of "end of stream" and do not receive this
+ * notification.
  *
  * **CONTEXT** - An arbitrary void * that the chain passes along to the
  * processing function as an argument.
  *
- * **CHAIN ATTRIBUTES** - A global set of flags available to all steps.
+ * **CHAIN ATTRIBUTES** - A global set of flags available to all steps. The
+ * chain is also responsible for tracking general statistics such as the
+ * number of records of each type that have been processed.
  */
 
 #define	CA_BYTESWAPPED			(1ULL << 0)	/* ca_attrs */
@@ -133,7 +142,7 @@ typedef struct {
 	record_stats_t	ca_stats_out[DRR_NUMTYPES];
 } chain_attrs_t;
 
-typedef enum { CS_SERIAL, CS_TERMINATE } step_type_t;
+typedef enum { CS_SERIAL, CS_PARALLEL, CS_TERMINATE } step_type_t;
 typedef enum { D_OK, D_EOF, D_DROP } disposition_t;
 
 typedef disposition_t
@@ -145,31 +154,47 @@ typedef struct chain_step
 	size_t		cs_in_size;
 	size_t		cs_out_size;
 	void		*cs_context;
-	struct {
-		zc_serial_process_f	*process;
-	} cs_serial;
+	union {
+		struct {
+			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;
+	};
 } chain_step_t;
 
 typedef chain_step_t zstream_chain_t[];
 
 /*
- * Chain attributes accessible to any step on the chain. In theory this
- * could cause a race condition between reading and setting, but all
- * attributes are typically set by the time the first record has been read.
- * Ergo, nobody else will be executing while that first chain_read() runs.
+ * Chain attributes accessible to any step on the chain. These are normally
+ * accessed through the macros defined above.
  */
 extern chain_attrs_t *chain_attrs;
 
 /*
  * Execute a chain. Returns once execution is complete. You can pass NULL
  * for the attrs if you're not interested in preserving them after the chain
- * has run.
+ * has run. (The chain will allocate and dispose of a buffer for them.)
  */
 void
 zstream_chain_exec(zstream_chain_t chain, chain_attrs_t *attrs);
 
-chain_step_t
-serial_null_step(void);
+/*
+ * Execute a chain linearly, without queues and without multithreading. This
+ * form of execution is intended as a debugging aid, both for clients and
+ * for the chain mechanism itself. If this variant doesn't produce results
+ * identical to zstream_chain_exec(), there's a multithreading-related bug
+ * somewhere.
+ *
+ * It is not necessary to remove parallel steps from the input chain. They
+ * are accepted as-is, but their execution won't be parallelized.
+ */
+void
+zstream_chain_exec_serialized(zstream_chain_t chain, chain_attrs_t *attrs);
 
 chain_step_t
 chain_terminator(void);

cmd/zstream/zstream_decompress.c

@@ -47,9 +47,11 @@
 #define	KEYSIZE 64
 
 static disposition_t
-chain_decompress_named_writes(drr_packet_t *item, void *context)
+chain_decompress_named_writes(void *item_in, void *context)
 {
 	(void) context;
+	drr_packet_t *item = (drr_packet_t *)item_in;
+
 	if (item == NULL) {
 		return (D_OK);
 	}
@@ -134,8 +136,7 @@ serial_decompress_named_writes(void)
 		.cs_out_size = sizeof (drr_packet_t),
 		.cs_context = NULL,
 		.cs_serial = {
-		    .process =
-			(zc_serial_process_f *)chain_decompress_named_writes
+		    .process = chain_decompress_named_writes
 		}
 	};
 	return (step);

cmd/zstream/zstream_drop_record.c

@@ -42,9 +42,11 @@
 #define	KEYSIZE 64
 
 static disposition_t
-chain_drop_records(drr_packet_t *item, void *context)
+chain_drop_records(void *item_in, void *context)
 {
 	(void) context;
+	drr_packet_t *item = (drr_packet_t *)item_in;
+
 	if (item == NULL)
 		return (D_OK);
 
@@ -105,7 +107,7 @@ serial_drop_records(void)
 		.cs_out_size = sizeof (drr_packet_t),
 		.cs_context = NULL,
 		.cs_serial = {
-			.process = (zc_serial_process_f *)chain_drop_records
+			.process = chain_drop_records
 		}
 	};
 	return (step);

cmd/zstream/zstream_dump.c

@@ -431,8 +431,11 @@ dump_redact_record(drr_packet_t *item)
 }
 
 static disposition_t
-chain_dump_record(drr_packet_t *item, record_type_t *context)
+chain_dump_record(void *item_in, void *context_in)
 {
+	drr_packet_t *item = (drr_packet_t *)item_in;
+	record_type_t *context = (record_type_t *)context_in;
+
 	if (item == NULL) {
 		return (D_OK);
 	}
@@ -463,7 +466,7 @@ serial_dump_records(record_type_t *context)
 		.cs_out_size = sizeof (drr_packet_t),
 		.cs_context = context,
 		.cs_serial = {
-			.process = (zc_serial_process_f *)chain_dump_record
+			.process = chain_dump_record
 		}
 	};
 	return (step);

cmd/zstream/zstream_fletcher4.c

@@ -20,22 +20,36 @@
 #include <assert.h>
 #include <stddef.h>
 #include <stdint.h>
+#include <stdlib.h>
 #include <sys/byteorder.h>
+#include <sys/param.h>
 #include <sys/spa_checksum.h>
 #include <sys/stdtypes.h>
+#include <sys/sysmacros.h>
 #include <sys/types.h>
 #include <sys/zfs_ioctl.h>
 #include <zfs_fletcher.h>
 
+#include "zstream_fletcher4.h"
 #include "zstream_modules.h"
+#include "zstream_queue.h"
 #include "zstream_util.h"
 
 #define	CK_OFFSET offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum)
 #define	END_CK_OFFSET offsetof(dmu_replay_record_t, drr_u.drr_end.drr_checksum)
 
+/*
+ * Copied from zfs_fletcher.c. See comments below regarding the
+ * fletcher_4_incremental_combine() function.
+ */
+#define	MAX_FLETCHER_BLOCK	(8ULL << 20)
+
 typedef enum { F4_SET, F4_VALIDATE } fletcher4_op_t;
 
-typedef zio_cksum_t fletcher4_context_t;
+typedef struct {
+	zio_cksum_t	fc_stream_cksum;
+	fletcher4_op_t	fc_operation;
+} fletcher4_context_t;
 
 static fletcher4_context_t	fletcher4_contexts[MAX_FLETCHER_4];
 static int			next_context = 0;
@@ -50,9 +64,130 @@ fletcher_4_incremental(boolean_t swap, void *buff, size_t size, void *cksum)
 	}
 }
 
+static inline void
+fletcher_4(boolean_t swap, void *buff, size_t size, void *cksum)
+{
+	if (swap) {
+		fletcher_4_byteswap(buff, size, NULL, cksum);
+	} else {
+		fletcher_4_native(buff, size, NULL, cksum);
+	}
+}
+
 /*
- * Emit or validate (below) a replay record with proper checksums and with
- * proper maintenance of the stream checksum. That is:
+ * The function below and the MAX_FLETCHER_BLOCK define are copied from
+ * zfs_fletcher.c, where they're internal.
+ *
+ * Fletcher checksums CAN be computed in parallel, with the segments later
+ * being reassembled. However, the combine function needs to know the
+ * original length of each segment, and there's a hard limit as to how long
+ * any given segment can be because 64-bit coefficients used in the combine
+ * operation may overflow if the size is larger than 8MB.
+ *
+ * My understanding of this is that the checksum fields themselves can and
+ * will overflow for long hash texts. However, they still function properly
+ * as checksums when this happens. But overflow has to be handled correctly
+ * in a structured fashion, not by allowing intermediate calculations to
+ * overflow.
+ */
+static inline void
+fletcher4_incremental_combine(zio_cksum_t *zcp, const uint64_t size,
+    const zio_cksum_t *nzcp)
+{
+	const uint64_t c1 = size / sizeof (uint32_t);
+	const uint64_t c2 = c1 * (c1 + 1) / 2;
+	const uint64_t c3 = c2 * (c1 + 2) / 3;
+
+	/*
+	 * Value of 'c3' overflows on buffer sizes close to 16MiB. For that
+	 * reason we split incremental fletcher4 computation of large buffers
+	 * to steps of (MAX_FLETCHER_BLOCK) size.
+	 */
+	ASSERT3U(size, <=, MAX_FLETCHER_BLOCK);
+
+	zcp->zc_word[3] += nzcp->zc_word[3] + c1 * zcp->zc_word[2] +
+	    c2 * zcp->zc_word[1] + c3 * zcp->zc_word[0];
+	zcp->zc_word[2] += nzcp->zc_word[2] + c1 * zcp->zc_word[1] +
+	    c2 * zcp->zc_word[0];
+	zcp->zc_word[1] += nzcp->zc_word[1] + c1 * zcp->zc_word[0];
+	zcp->zc_word[0] += nzcp->zc_word[0];
+}
+
+/*
+ * This is the parallel portion of checksum calculation. We calculate only
+ * the checksum blocks for payloads. The records themselves are summed in
+ * the serial step.
+ *
+ * Because MAX_FLETCHER_BLOCK is 8MB, the great majority of payloads need
+ * only a single checksum calculation. The drr_fletcher4_t struct has both a
+ * first-block checksum field and a pointer to an overflow block on the
+ * heap. The overflow block is not allocated unless the payload size is
+ * greater than MAX_FLETCHER_BLOCK.
+ */
+static void
+chain_calc_fletcher4(queue_item_t *item_in, void *context)
+{
+	(void) context;
+	drr_fletcher4_t *item = (drr_fletcher4_t *)item_in;
+
+	VERIFY3U(item->dp_base.dp_payload_size, >, 0);
+
+	ssize_t remaining = item->dp_base.dp_payload_size;
+	uint8_t *data = item->dp_base.dp_payload;
+	size_t write_size = MIN(remaining, MAX_FLETCHER_BLOCK);
+	int num_overflow = DIV_ROUND_UP(remaining, MAX_FLETCHER_BLOCK) - 1;
+	zio_cksum_t *fragment = &item->dp_fletcher4_payload;
+	boolean_t swap = ATTR_IS_SET(CA_BYTESWAPPED);
+
+	fletcher_4(swap, data, write_size, fragment);
+	if (num_overflow) {
+		fragment = safe_calloc(num_overflow * sizeof (zio_cksum_t));
+		item->dp_fletcher4_overflow = fragment;
+	} else {
+		item->dp_fletcher4_overflow = NULL;
+	}
+	while (remaining -= write_size) {
+		data += write_size;
+		write_size = MIN(remaining, MAX_FLETCHER_BLOCK);
+		fletcher_4(swap, data, write_size, fragment);
+		fragment++;
+	}
+}
+
+/*
+ * Combine the payload checksum(s) produced by the parallel phase into the
+ * stream checksum. Used by the serial validation or inscription step.
+ */
+static void
+assemble_payload_cksum(drr_fletcher4_t *item, zio_cksum_t *stream_ck)
+{
+	ssize_t remaining = item->dp_base.dp_payload_size;
+	size_t read_size = MIN(remaining, MAX_FLETCHER_BLOCK);
+	zio_cksum_t *fragment = item->dp_fletcher4_overflow;
+
+	if (remaining == 0)
+		return;
+	fletcher4_incremental_combine(stream_ck, read_size,
+	    &item->dp_fletcher4_payload);
+	while (remaining -= read_size) {
+		read_size = MIN(remaining, MAX_FLETCHER_BLOCK);
+		fletcher4_incremental_combine(stream_ck, read_size, fragment);
+		fragment++;
+	}
+	if (item->dp_fletcher4_overflow != NULL) {
+		free(item->dp_fletcher4_overflow);
+		item->dp_fletcher4_overflow = NULL;
+	}
+}
+
+/*
+ * This function implements the serial portions of both validation and
+ * inscription, based on the fc_operation field of the context struct.
+ * Checksumming is either very early in a chain or very late, so records
+ * are potentially in non-native endianness in either mode.
+ *
+ * This function emits or validates a replay record with proper checksums
+ * and with proper maintenance of the stream checksum. That is:
  *
  *   1) Update stream checksum with the record header up to drr_checksum.
  *   2) Update drr_checksum field in the record header from stream checksum.
@@ -62,33 +197,74 @@ fletcher_4_incremental(boolean_t swap, void *buff, size_t size, void *cksum)
  * DRR_BEGIN records do not have record checksums. They can't, because the
  * drr_begin struct overlaps with space that would otherwise be used for the
  * end-record checksum.
+ *
+ * DRR_END records normally do have end-record checksums. However, records
+ * emitted by send_conclusion_record() in libzfs_sendrecv.c have the
+ * checksum set to zero. zfs receive ignores those checksums. DRR_END records
+ * also have an internal checksum that applies to the stream-to-date since the
+ * most recent DRR_BEGIN.
+ *
+ * Null zstream transformations should be idempotent. E.g., a zstream redup
+ * that does not redup anything should yield a stream that is bit-for-bit
+ * identical to the original stream. So, it's helpful to emulate zfs send's
+ * checksumming pattern just to minimize spurious differences between
+ * input and output streams.
  */
 static disposition_t
-chain_add_fletcher4(drr_packet_t *item, zio_cksum_t *stream_cksum)
+chain_fletcher4(queue_item_t *item_in, void *context_in)
 {
-	if (item == NULL)
+	drr_fletcher4_t *item = (drr_fletcher4_t *)item_in;
+	fletcher4_context_t *context = (fletcher4_context_t *)context_in;
+
+	if (item == NULL || (context->fc_operation == F4_VALIDATE &&
+	    OPTION_ENABLED(CA_IGNORE_CKSUMS))) {
 		return (D_OK);
+	}
 
-	dmu_replay_record_t *drr   = &item->dp_drr;
-	struct drr_end *drre	   = &item->dp_drr.drr_u.drr_end;
-	zio_cksum_t *record_cksum  = &drr->drr_u.drr_checksum.drr_checksum;
-	zio_cksum_t *end_cksum	   = &drre->drr_checksum;
+	zio_cksum_t *stream_cksum	= &context->fc_stream_cksum;
+	dmu_replay_record_t *drr	= &item->dp_base.dp_drr;
+	struct drr_end *drre		= &drr->drr_u.drr_end;
+	zio_cksum_t *record_cksum	= &drr->drr_u.drr_checksum.drr_checksum;
+	zio_cksum_t *end_cksum		= &drre->drr_checksum;
 
-	boolean_t swap = OPTION_ENABLED(CA_BYTESWAP_ON_OUTPUT);
+	boolean_t swap = (context->fc_operation == F4_VALIDATE &&
+	    ATTR_IS_SET(CA_BYTESWAPPED)) || (context->fc_operation == F4_SET &&
+	    OPTION_ENABLED(CA_BYTESWAP_ON_OUTPUT));
 	uint32_t drr_type = swap ? BSWAP_32(drr->drr_type) : drr->drr_type;
+	off_t ck_offset = offsetof(dmu_replay_record_t,
+	    drr_u.drr_checksum.drr_checksum);
 
+	if (item->dp_base.dp_stream_offset == 0) {
+		VERIFY3U(ck_offset, ==, sizeof (dmu_replay_record_t) -
+		    sizeof (zio_cksum_t));
+	}
 	if (drr_type == DRR_BEGIN) {
 		ZIO_SET_CHECKSUM(stream_cksum, 0, 0, 0, 0);
 	} else if (drr_type == DRR_END) {
-		*end_cksum = *stream_cksum;
-		if (swap)
-			ZIO_CHECKSUM_BSWAP(end_cksum);
+		if (context->fc_operation == F4_VALIDATE) {
+			off_t stream_offset = item->dp_base.dp_stream_offset +
+			    offsetof(dmu_replay_record_t,
+			    drr_u.drr_end.drr_checksum);
+			validate_or_exit(stream_cksum, end_cksum, swap,
+			    "in DRR_END record", stream_offset);
+		} else {
+			*end_cksum = *stream_cksum;
+			if (swap)
+				ZIO_CHECKSUM_BSWAP(end_cksum);
+		}
 	}
-	fletcher_4_incremental(swap, drr, CK_OFFSET, stream_cksum);
+	fletcher_4_incremental(swap, drr, ck_offset, stream_cksum);
 	if (drr_type != DRR_BEGIN && !IS_CONCLUSION(drr, drr_type)) {
-		*record_cksum = *stream_cksum;
-		if (swap)
-			ZIO_CHECKSUM_BSWAP(record_cksum);
+		if (context->fc_operation == F4_VALIDATE) {
+			off_t stream_offset =
+			    item->dp_base.dp_stream_offset + ck_offset;
+			validate_or_exit(stream_cksum, record_cksum,
+			    swap, "at DRR record end", stream_offset);
+		} else {
+			*record_cksum = *stream_cksum;
+			if (swap)
+				ZIO_CHECKSUM_BSWAP(record_cksum);
+		}
 	}
 	if (drr_type == DRR_END) {
 		ZIO_SET_CHECKSUM(stream_cksum, 0, 0, 0, 0);
@@ -96,54 +272,32 @@ chain_add_fletcher4(drr_packet_t *item, zio_cksum_t *stream_cksum)
 		fletcher_4_incremental(swap, record_cksum,
 		    sizeof (drr->drr_u.drr_checksum.drr_checksum),
 		    stream_cksum);
-		if (item->dp_payload_size > 0) {
-			fletcher_4_incremental(swap, item->dp_payload,
-			    item->dp_payload_size, stream_cksum);
-		}
+		assemble_payload_cksum(item, stream_cksum);
 	}
 	return (D_OK);
 }
 
-static disposition_t
-chain_validate_fletcher4(drr_packet_t *item, zio_cksum_t *stream_cksum)
+/*
+ * 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.
+ */
+chain_step_t
+parallel_calc_fletcher4(int queue_length)
 {
-	if (item == NULL || OPTION_ENABLED(CA_IGNORE_CKSUMS)) {
-		return (D_OK);
-	}
-
-	dmu_replay_record_t *drr   = &item->dp_drr;
-	struct drr_end *drre	   = &item->dp_drr.drr_u.drr_end;
-	zio_cksum_t *record_cksum  = &drr->drr_u.drr_checksum.drr_checksum;
-	zio_cksum_t *end_cksum	   = &drre->drr_checksum;
-
-	boolean_t swap = ATTR_IS_SET(CA_BYTESWAPPED);
-	uint32_t drr_type = swap ? BSWAP_32(drr->drr_type) : drr->drr_type;
-
-	if (drr_type == DRR_BEGIN) {
-		ZIO_SET_CHECKSUM(stream_cksum, 0, 0, 0, 0);
-	} else if (drr_type == DRR_END) {
-		off_t stream_offset = item->dp_stream_offset + END_CK_OFFSET;
-		validate_or_exit(stream_cksum, end_cksum, swap,
-		    "in DRR_END record", stream_offset);
-	}
-	fletcher_4_incremental(swap, drr, CK_OFFSET, stream_cksum);
-	if (drr_type != DRR_BEGIN && !IS_CONCLUSION(drr, drr_type)) {
-		off_t stream_offset = item->dp_stream_offset + CK_OFFSET;
-		validate_or_exit(stream_cksum, record_cksum,
-		    swap, "at DRR record end", stream_offset);
-	}
-	if (drr_type == DRR_END) {
-		ZIO_SET_CHECKSUM(stream_cksum, 0, 0, 0, 0);
-	} else {
-		fletcher_4_incremental(swap, record_cksum,
-		    sizeof (drr->drr_u.drr_checksum.drr_checksum),
-		    stream_cksum);
-		if (item->dp_payload_size > 0) {
-			fletcher_4_incremental(swap, item->dp_payload,
-			    item->dp_payload_size, stream_cksum);
-		}
-	}
-	return (D_OK);
+	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
+	    }
+	};
+	return (step);
 }
 
 static chain_step_t
@@ -152,17 +306,17 @@ fletcher4_serial_step(fletcher4_op_t operation)
 	int context_ix = next_context++ % MAX_FLETCHER_4;
 	fletcher4_context_t *context = &fletcher4_contexts[context_ix];
 
-	ZIO_SET_CHECKSUM(context, 0, 0, 0, 0);
+	context->fc_operation = operation;
+	ZIO_SET_CHECKSUM(&context->fc_stream_cksum, 0, 0, 0, 0);
+
 	chain_step_t step = {
-		.cs_type = CS_SERIAL,
-		.cs_in_size = sizeof (drr_packet_t),
-		.cs_out_size = sizeof (drr_packet_t),
-		.cs_context = context,
-		.cs_serial = {
-			.process = (zc_serial_process_f *)
-			    ((operation == F4_VALIDATE) ?
-			    chain_validate_fletcher4 : chain_add_fletcher4)
-		}
+	    .cs_type = CS_SERIAL,
+	    .cs_in_size = sizeof (drr_fletcher4_t),
+	    .cs_out_size = sizeof (drr_packet_t),
+	    .cs_context = context,
+	    .cs_serial = {
+		.process = chain_fletcher4,
+	    }
 	};
 	return (step);
 }

cmd/zstream/zstream_fletcher4.h

@@ -24,19 +24,30 @@
 extern "C" {
 #endif
 
+#include <sys/spa_checksum.h>
+#include <sys/zfs_ioctl.h>
+
 #include "zstream_io.h"
 
 /*
  * zstream_chain module for calculating, validating, and inscribing
  * Fletcher4 checksums.
  *
- * serial_validate_fletcher4() validates record checksums against the
- * running stream checksum and fails loudly on any mismatch.
+ * Checksums are calculated for the entire stream between a DRR_BEGIN record
+ * and its corresponding DRR_END, so the final checksum assembly must be
+ * performed as a serial step. However, we can pre-calculate the checksums
+ * for individual payloads in parallel.
  *
- * serial_add_fletcher4() inscribes record checksums from the running
- * stream checksum, in theory replacing whatever was there before. But
- * see note in zstream_fletcher4.c regarding zero checksums generated
- * by send_conclusion_record(), which are preserved.
+ * The normal sequence is a parallel_calc_fletcher4() step followed by a
+ * serial_validate_fletcher4() or serial_add_fletcher4() step.
+ *
+ * Fletcher4 is just addition, so it's quite fast on its own. However, not
+ * parallelizing these calculations reduces aggregate throughput because of
+ * [Amdahl's Law](https://en.wikipedia.org/wiki/Amdahl%27s_law). Briefly
+ * stated: nonparallelizable work limits the total speedup obtainable
+ * through parallelization, often to a nonintuitive degree. Here, the effect
+ * is magnified because parallelizable work such as recompression lies
+ * beyond a Fletcher4 bottleneck.
  */
 
 /*
@@ -65,6 +76,15 @@ extern "C" {
  */
 #define	MAX_FLETCHER_4 8
 
+typedef struct {
+	drr_packet_t	dp_base;
+	zio_cksum_t	dp_fletcher4_payload;
+	zio_cksum_t	*dp_fletcher4_overflow;
+} drr_fletcher4_t;
+
+chain_step_t
+parallel_calc_fletcher4(int queue_length);
+
 chain_step_t
 serial_validate_fletcher4(void);
 

cmd/zstream/zstream_io.c

@@ -19,11 +19,10 @@
 
 #include <arpa/inet.h>
 #include <err.h>
-#include <errno.h>
 #include <libzutil.h>
+#include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
-#include <string.h>
 #include <sys/byteorder.h>
 #include <sys/stdtypes.h>
 #include <sys/sysmacros.h>
@@ -32,7 +31,6 @@
 #include <time.h>
 #include <unistd.h>
 
-#include "zstream_chain.h"
 #include "zstream_modules.h"
 #include "zstream_util.h"
 
@@ -213,8 +211,11 @@ set_stream_attributes(drr_packet_t *item)
 }
 
 static disposition_t
-chain_read(drr_packet_t *item, io_context_t *context)
+chain_read(void *item_in, void *context_in)
 {
+	drr_packet_t *item = (drr_packet_t *)item_in;
+	io_context_t *context = (io_context_t *)context_in;
+
 	if (item == NULL)
 		return (D_OK);
 
@@ -293,8 +294,11 @@ chain_read(drr_packet_t *item, io_context_t *context)
 }
 
 static disposition_t
-chain_write(drr_packet_t *item, io_context_t *context)
+chain_write(void *item_in, void *context_in)
 {
+	drr_packet_t *item = (drr_packet_t *)item_in;
+	io_context_t *context = (io_context_t *)context_in;
+
 	if (item == NULL) {
 		if (context->ic_fp) {
 			if (fclose(context->ic_fp) != 0)
@@ -343,9 +347,11 @@ chain_write(drr_packet_t *item, io_context_t *context)
  * Even if the chain doesn't write out a stream, payloads still need freed.
  */
 static disposition_t
-chain_null_output(drr_packet_t *item, void *context)
+chain_null_output(void *item_in, void *context)
 {
 	(void) context;
+	drr_packet_t *item = (drr_packet_t *)item_in;
+
 	if (item && item->dp_payload != NULL && item->dp_payload_size > 0) {
 		free(item->dp_payload);
 		item->dp_payload = NULL;
@@ -370,12 +376,11 @@ setup_io(const char *filename, boolean_t for_reading)
 
 	chain_step_t step = {
 		.cs_type = CS_SERIAL,
-		.cs_in_size = 0,
-		.cs_out_size = sizeof (drr_packet_t),
+		.cs_in_size = for_reading ? 0 : sizeof (drr_packet_t),
+		.cs_out_size = for_reading ? sizeof (drr_packet_t) : 0,
 		.cs_context = &io_contexts[context_num],
 		.cs_serial = {
-			.process = (zc_serial_process_f *)
-			    (for_reading ? chain_read : chain_write),
+			.process = for_reading ? chain_read : chain_write
 		}
 	};
 	return (step);
@@ -402,15 +407,34 @@ serial_null_output(void)
 		.cs_out_size = 0,
 		.cs_context = NULL,
 		.cs_serial = {
-			.process = (zc_serial_process_f *)chain_null_output
+			.process = chain_null_output
 		}
 	};
 	return (step);
 }
 
-static disposition_t
-chain_checkpoint(drr_packet_t *item, checkpoint_context_t *ctxt)
+size_t
+constant_cost_of_one(queue_item_t *packet, void *context)
 {
+	(void) context;
+	(void) packet;
+	return (1);
+}
+
+size_t
+payload_size_as_cost(queue_item_t *packet_in, void *context)
+{
+	(void) context;
+	drr_packet_t *packet = (drr_packet_t *)packet_in;
+	return (packet->dp_payload_size);
+}
+
+static disposition_t
+chain_checkpoint(void *item_in, void *ctxt_in)
+{
+	drr_packet_t *item = (drr_packet_t *)item_in;
+	checkpoint_context_t *ctxt = (checkpoint_context_t *)ctxt_in;
+
 	struct timespec now;
 	char buff[32];
 	uint64_t delta_b, dbdt;
@@ -455,7 +479,7 @@ serial_checkpoint(const char *name)
 		.cs_out_size = sizeof (drr_packet_t),
 		.cs_context = &checkpoint_contexts[context_no],
 		.cs_serial = {
-			.process = (zc_serial_process_f *)chain_checkpoint
+			.process = chain_checkpoint
 		},
 	};
 	return (step);

cmd/zstream/zstream_io.h

@@ -24,6 +24,8 @@
 extern "C" {
 #endif
 
+#include <stddef.h>
+#include <stdint.h>
 #include <sys/types.h>
 #include <sys/zfs_ioctl.h>
 
@@ -67,6 +69,14 @@ serial_checkpoint(const char *name);
 chain_step_t
 serial_null_output(void);
 
+/* Off-the-shelf zstream_queue cost functions */
+
+size_t
+constant_cost_of_one(queue_item_t *packet, void *context);
+
+size_t
+payload_size_as_cost(queue_item_t *packet, void *context);
+
 #ifdef __cplusplus
 }
 #endif

cmd/zstream/zstream_modules.h

@@ -37,20 +37,23 @@ extern "C" {
 #include "zstream_util.h"
 #include "zstream_validate.h"
 
-#define	READ_STEP 0
-
-#define	STANDARD_INPUT_STACK(infile)					\
+#define	STANDARD_INPUT_STACK_Q(infile, queue_size) 			\
 	serial_read_stream(infile),					\
+	parallel_calc_fletcher4(queue_size),				\
 	serial_validate_fletcher4(),					\
 	serial_byteswap(BS_INCOMING),					\
 	serial_validate_records()
 
-#define	STANDARD_OUTPUT_STACK(outfile)					\
+#define	STANDARD_OUTPUT_STACK_Q(outfile, queue_size) 			\
 	serial_byteswap(BS_OUTGOING),					\
+	parallel_calc_fletcher4(queue_size),				\
 	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

@@ -0,0 +1,817 @@
+// SPDX-License-Identifier: CDDL-1.0
+/*
+ * CDDL HEADER START
+ *
+ * This file and its contents are supplied under the terms of the Common
+ * Development and Distribution License ("CDDL"), version 1.0. You may only use
+ * this file in accordance with the terms of version 1.0 of the CDDL.
+ *
+ * A full copy of the text of the CDDL should have accompanied this source. A
+ * copy of the CDDL is also available via the Internet at
+ * http://www.illumos.org/license/CDDL.
+ *
+ * CDDL HEADER END
+ */
+
+/*
+ * Copyright (c) 2026 by Garth Snyder. All rights reserved.
+ */
+
+#include <assert.h>
+#include <atomic.h>
+#include <err.h>
+#include <pthread.h>
+#include <sched.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/param.h>
+#include <sys/random.h>
+#include <sys/stdtypes.h>
+#include <unistd.h>
+
+#include "zstream_queue.h"
+#include "zstream_util.h"
+
+#define	MIN_THREADS 	6
+#define	MAX_QUEUES 	16	/* Largest # of queues simultaneously active */
+
+#define	PLENTY_OF_WORK		6	/* "Many" items to claim */
+#define	NO_WORK			0.0001	/* 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_FULL(queue)	((queue)->zq_ix.enqueue - (queue)->zq_ix.dequeue >= \
+	    (queue)->zq_params.qp_queue_length)
+
+/*
+ * A zstream_queue is a ring buffer with four indices: enqueue, claim,
+ * complete, and dequeue, in that order. No index can move beyond its
+ * preceding index. Every interval between indices contains work items in a
+ * particular state: enqueued, claimed for work, or completed. Items never
+ * leave the ring buffer, so FIFO order is guaranteed on dequeueing.
+ *
+ * In concept, every index has a corresponding condition that threads can
+ * wait on if they are interested in knowing when that index moves:
+ * enqueued, claimed, completed, dequeued. However, the exact reality
+ * deviates from this model in two ways:
+ *
+ * - There is no "claimed" condition, because no thread would wait on it.
+ * Claiming and processing are one unified operation. Dequeuers await the
+ * "completed" condition.
+ *
+ * - All queues share one thread pool, so idle threads are not bound to any
+ * particular queue. Instead of having queue-specific "enqueued" conditions,
+ * one condition is shared by all queues. On awakening, each worker thread
+ * is dynamically assigned to a queue using a scoring mechanism described
+ * in the comments at score_queue().
+ *
+ * THREAD SAFETY STRATEGY
+ *
+ * There are four types of lock:
+ *
+ * - One global lock that gates changes to the thread pool and queue cohort
+ * - A second global lock that controls the creation of new queues
+ * - A third global lock associated with the shared "enqueued" condition
+ * - One lock for each queue
+ *
+ * Although the "enqueued" condition and its associated lock are stored as
+ * part of the global thread pool, they are logically separate from the pool
+ * itself.
+ *
+ * For the most part, locking is straightforward. Any operation that adds or
+ * removes queues or threads should hold the pool lock. Any operation that
+ * moves a queue's indices should hold the queue lock. Any thread waiting
+ * for work should wait on the "enqueued" condition.
+ *
+ * Worker threads hold no locks while they are actually processing items.
+ *
+ * Several operations require multiple locks. In these cases, a standardized
+ * locking order is used to avoid deadlocks:
+ *
+ *   enqueue -> pool -> queue -> create
+ *
+ * Several operations merit additional comments about locking. These are
+ * marked with a "locking note" in the comments preceding the relevant
+ * function.
+ */
+
+typedef struct {
+	queue_item_t	*qs_item;
+	size_t		qs_cost;
+	boolean_t	qs_completed;
+	boolean_t	qs_end_of_stream;
+} queue_slot_t;
+
+typedef struct {
+	uint64_t	enqueue;
+	uint64_t	claim;
+	uint64_t	complete;
+	uint64_t	dequeue;
+} zq_indices_t;
+
+typedef struct {
+	pthread_cond_t	completed;
+	pthread_cond_t	dequeued;
+} zq_conditions_t;
+
+typedef struct {
+	int		min_depth;
+	int		max_depth;
+} zq_stats_t;
+
+struct zstream_queue {
+	queue_slot_t	*zq_slots;
+	pthread_mutex_t	zq_mutex;
+	zq_indices_t	zq_ix;
+	zq_conditions_t	zq_cond;
+	zq_params_t	zq_params;
+	zq_stats_t	zq_stats;
+	boolean_t	zq_disallow_enqueue;
+};
+
+typedef struct {
+	pthread_mutex_t	tp_pool_mutex;
+	pthread_mutex_t tp_create_mutex;
+	pthread_mutex_t tp_enqueue_mutex;
+	pthread_cond_t	tp_enqueued;
+	zstream_queue_t	*tp_queues[MAX_QUEUES];
+	int		tp_num_queues;
+	pthread_t	*tp_threads;
+	int		tp_num_threads;
+} thread_pool_t;
+
+typedef void cleanup_f(void *);
+
+static void *
+queue_worker(void *);
+
+#ifdef MONITOR_QUEUES
+static void
+start_monitor_thread(void);
+#endif
+
+static thread_pool_t	pool = {0};
+static int		num_threads = 0;
+static boolean_t	pool_initialized = B_FALSE;
+static pthread_once_t	once_control = PTHREAD_ONCE_INIT;
+
+void
+zstream_queue_set_num_threads(uint_t n)
+{
+	if (pool_initialized) {
+		errx(1, "thread pool size must be set before creating queues");
+	} else if (n < MIN_THREADS) {
+		errx(1, "number of threads must be at least %d", MIN_THREADS);
+	} else if (n > 256) {
+		warnx("num_threads = %u seems suspiciously high, setting "
+		    "anyway...", n);
+	}
+	num_threads = n;
+}
+
+static void
+thread_pool_init(void)
+{
+	pthread_mutex_init(&pool.tp_pool_mutex, NULL);
+	pthread_mutex_init(&pool.tp_create_mutex, NULL);
+	pthread_mutex_init(&pool.tp_enqueue_mutex, NULL);
+	pthread_cond_init(&pool.tp_enqueued, NULL);
+	pool_initialized = B_TRUE;
+}
+
+/*
+ * Locking note: must be called by a function holding the pool mutex
+ *
+ * If num_threads is nonzero, it sets the number of threads to spawn.
+ * Otherwise, one thread is spawned per core.
+ *
+ * sched_affinity() is a better estimate of available threads than sysconf
+ * because sysconf doesn't account for limits that might be set on, e.g., a
+ * container.
+ */
+static void
+thread_pool_spinup(void)
+{
+	pool.tp_num_queues = 0;
+	if (num_threads > 0) {
+		pool.tp_num_threads = num_threads;
+	} else {
+#ifdef	CPU_COUNT
+		cpu_set_t cpu_set;
+		sched_getaffinity(0, sizeof (cpu_set_t), &cpu_set);
+		pool.tp_num_threads = CPU_COUNT(&cpu_set);
+#else
+		pool.tp_num_threads = sysconf(_SC_NPROCESSORS_ONLN);
+#endif
+	}
+	pool.tp_num_threads = MAX(pool.tp_num_threads, MIN_THREADS);
+	pool.tp_threads = safe_malloc(sizeof (pthread_t) * pool.tp_num_threads);
+	for (int i = 0; i < pool.tp_num_threads; i++) {
+		char buff[32];
+		pthread_t *thread = &pool.tp_threads[i];
+		int ret = pthread_create(thread, NULL, queue_worker, NULL);
+		VERIFY3S(ret, ==, 0);
+		snprintf(buff, sizeof (buff), "queue-%d", i);
+		pthread_setname_np(*thread, buff);
+	}
+#ifdef MONITOR_QUEUES
+	start_monitor_thread();
+#endif
+}
+
+/*
+ * Locking note: thread_pool_spindown() is a pool-level operation and by
+ * rights should hold the pool mutex. (And in fact, the caller must already
+ * hold that mutex.)
+ *
+ * However, we can't leave the pool mutex locked while canceling threads
+ * because most worker threads will be waiting on the "enqueued" condition.
+ * That condition is protected by the enqueue mutex, which threads need to
+ * lock just to wake up and be canceled.
+ *
+ * Even though the two mutexes are locked by different threads, it is still
+ * one composite operation for which the locking order is pool -> enqueue.
+ * That's incompatible with the standard locking order of enqueue -> pool ->
+ * queue -> create, so continuing to hold the pool mutex risks deadlock.
+ *
+ * If we are here, that means there are no existing queues, so we needn't
+ * worry about operations being attempted on queues. The one potential
+ * conflict is with zstream_queue_create(). That's the reason for the
+ * seemingly redundant "create" mutex. It lets us prevent the creation of
+ * new queues while simultaneously dropping the pool lock.
+ */
+static void
+thread_pool_spindown(void)
+{
+	pthread_mutex_lock(&pool.tp_create_mutex);
+	pthread_mutex_unlock(&pool.tp_pool_mutex);
+
+	for (int i = 0; i < pool.tp_num_threads; i++) {
+		VERIFY3S(pthread_cancel(pool.tp_threads[i]), ==, 0);
+		VERIFY3S(pthread_join(pool.tp_threads[i], NULL), ==, 0);
+	}
+	free(pool.tp_threads);
+	pool.tp_threads = NULL;
+	pool.tp_num_threads = 0;
+
+	pthread_mutex_lock(&pool.tp_pool_mutex);
+	pthread_mutex_unlock(&pool.tp_create_mutex);
+}
+
+/*
+ * Locking note: see comments on thread_pool_spindown() for an explanation
+ * of why this operation acquires two locks.
+ */
+zstream_queue_t *
+zstream_queue_create(zq_params_t *params)
+{
+	pthread_once(&once_control, thread_pool_init);
+	pthread_mutex_lock(&pool.tp_pool_mutex);
+	pthread_mutex_lock(&pool.tp_create_mutex);
+	VERIFY3S(pool.tp_num_queues, <, MAX_QUEUES);
+
+	if (!pool.tp_num_threads) {
+		thread_pool_spinup();
+	}
+	zstream_queue_t *queue = safe_malloc(sizeof (zstream_queue_t));
+	pool.tp_queues[pool.tp_num_queues] = queue;
+
+	zstream_queue_t new_queue = {
+	    .zq_params = *params,
+	    .zq_slots = safe_malloc(params->qp_queue_length *
+		((sizeof (queue_slot_t)) + params->qp_item_size))
+	};
+	*queue = new_queue;
+	/*
+	 * Queue slots and item storage are allocated in one block, so we
+	 * need to manually wire each slot to its item buffer.
+	 */
+	uint8_t *item = (uint8_t *)&queue->zq_slots[params->qp_queue_length];
+	queue_slot_t *slot = &queue->zq_slots[0];
+	for (int i = 0; i < params->qp_queue_length; i++) {
+		slot->qs_item = item;
+		item += queue->zq_params.qp_item_size;
+		slot++;
+	}
+
+	pthread_mutex_init(&queue->zq_mutex, NULL);
+	pthread_cond_init(&queue->zq_cond.completed, NULL);
+	pthread_cond_init(&queue->zq_cond.dequeued, NULL);
+
+	pool.tp_num_queues++;
+
+	pthread_mutex_unlock(&pool.tp_create_mutex);
+	pthread_mutex_unlock(&pool.tp_pool_mutex);
+
+	return (queue);
+}
+
+/*
+ * Try to advance the "complete" index as far as possible by examining the
+ * qs_completed flag on each item. This can't be done directly by the
+ * threads that complete work, for a couple of reasons:
+ *
+ * - Items can be completed in any order. Just because you (a thread) have
+ * finished your batch doesn't mean that all prior batches have completed.
+ * If there are uncompleted items ahead of you in the ring buffer, you can't
+ * advance the completion index past them.
+ *
+ * - Items for which the cost function returns 0 are marked as qs_completed
+ * on enqueue and are never seen by a worker thread. So, there needs to be
+ * an independent mechanism to sweep the completion index past these items
+ * whenever that becomes possible.
+ *
+ * This function is called:
+ *
+ * - Whenever a thread completes a batch
+ * - Whenever a thread claims a batch
+ * - Whenever an item of cost 0 is enqueued
+ *
+ * Strictly speaking, advancing on claiming a batch is not logically
+ * necessary. However, the claimer already holds the queue mutex, and
+ * it's in our interest to make completed items available for dequeueing as
+ * expeditiously as possible.
+ *
+ * Locking note: the calling thread must hold the queue mutex.
+ */
+static inline void
+advance_completion_index(zstream_queue_t *queue)
+{
+	boolean_t any_completed = B_FALSE;
+	while (queue->zq_ix.complete < queue->zq_ix.claim &&
+	    Q_SLOT(queue, queue->zq_ix.complete).qs_completed) {
+		queue->zq_ix.complete++;
+		any_completed = B_TRUE;
+	}
+	if (any_completed) {
+		pthread_cond_signal(&queue->zq_cond.completed);
+	}
+}
+
+/*
+ * Locking note: the calling thread must hold the enqueue mutex and the
+ * thread pool mutex.
+ *
+ * This function scores 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.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * To score queues, a thread must hold both the thread pool mutex and the
+ * global enqueue mutex. However, it does not need to hold the mutex for the
+ * queue being scored. Several corollaries:
+ *
+ * 1) Only one thread may score queues at a time.
+ *
+ * 2) Worker threads can still complete work during scoring, so queue scores
+ * may become stale before they are used.
+ *
+ * 3) If a queue score is stale, it will always err on the side of
+ * overstating the amount of work that a queue has available. This is fine
+ * because at worst, a thread is assigned to a no-work queue and loops
+ * immediately.
+ *
+ * 4) Understatement is not possible because the enqueue mutex is locked
+ * during scoring. We don't want a queue to be scored and then receive new
+ * work while the scorer is looking at other queues. That would create a
+ * potential race condition in which a scorer concludes that there is no
+ * work available on any queue and goes back to sleep. If no further items
+ * are submitted to any queue, no worker thread will ever be awakened to
+ * process the newly-enqueued item.
+ */
+static inline double
+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, PLENTY_OF_WORK) /
+	    (double)PLENTY_OF_WORK;
+	double need = open_score + dq_score * DEQUEUE_SCORE_WEIGHT;
+	return (need * claim_factor);
+}
+
+/*
+ * Return a random index from an array of doubles, with the likelihood of
+ * index i being selected equal to weights[i] / sum(weights).
+ */
+static inline int
+select_stochastic(double weights[], int num_values)
+{
+	uint32_t numerator;
+	uint32_t denominator = UINT32_MAX;
+	double total = 0.0;
+
+	for (int i = 0; i < num_values; i++) {
+		total += weights[i];
+	}
+	random_get_pseudo_bytes((uint8_t *)&numerator, sizeof (uint32_t));
+	double select_val = total * numerator / denominator;
+	for (int i = 0; i < num_values; i++) {
+		if (select_val <= weights[i])
+			return (i);
+		select_val -= weights[i];
+	}
+	return (num_values - 1);
+}
+
+static void
+auto_unlock_mutex(pthread_mutex_t *mutex)
+{
+	pthread_mutex_unlock(mutex);
+}
+
+static void
+await_condition(pthread_cond_t *cond, pthread_mutex_t *mutex)
+{
+	pthread_cleanup_push((cleanup_f *)auto_unlock_mutex, mutex);
+	pthread_cond_wait(cond, mutex);
+	pthread_cleanup_pop(0);
+}
+
+/*
+ * Claim up to MAX_BATCH work items from the given queue, trying to
+ * accumulate at least queue->qp_batch_budget worth of work data (==
+ * "cost"). All items in a batch will be drawn from the same queue.
+ *
+ * Does not block waiting to fill the budget; returns whatever is available
+ * now.
+ *
+ * Locking note: 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
+ * mutex while destroying it (because destruction entails destroying the
+ * queue mutex, which must be unlocked), so holding the queue mutex while
+ * attempting to claim work is not a sufficient guarantee of correctness.
+ *
+ * In other contexts, we have more certainty about whether a queue still has
+ * work to do. If it does, it can't be destroyed while we hold the queue
+ * mutex alone. But here, we merely suspect that there's work available
+ * based on possibly outdated queue scoring information. By the time we get
+ * here, the queue might already have been finalized. Holding the thread
+ * pool mutex guarantees that the queue won't have been destroyed out from
+ * under us.
+ */
+static int
+claim_batch(zstream_queue_t *queue, queue_slot_t **batch)
+{
+	size_t cost_claimed = 0;
+	int count = 0;
+	boolean_t more_to_claim, more_slots, more_budget;
+	boolean_t first_and_only, ok_to_claim;
+
+	while (B_TRUE) {
+		more_to_claim = queue->zq_ix.claim < queue->zq_ix.enqueue;
+		more_slots = count < 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;
+
+		if (!more_to_claim || !more_slots || !ok_to_claim) {
+			break;
+		}
+		queue_slot_t *slot = &Q_SLOT(queue, queue->zq_ix.claim);
+		if (!slot->qs_completed) {
+			cost_claimed += slot->qs_cost;
+			batch[count++] = slot;
+		}
+		queue->zq_ix.claim++;
+	}
+
+	advance_completion_index(queue);
+	return (count);
+}
+
+/*
+ * Threads are assigned to a queue on each loop so they can be shifted
+ * dynamically to follow available work. Idle threads will typically
+ * be awaiting the "enqueued" condition within this function.
+ *
+ * Locking note: this function has complex locking behavior. At first we
+ * must hold both the enqueue mutex (to be sure new work doesn't get sneaked
+ * in after a queue is scored, which might cause it to be overlooked
+ * entirely) and the thread pool mutex (to guarantee that no queue can be
+ * destroyed out from under us).
+ *
+ * After scoring, we can release the enqueue mutex. However, we need to then
+ * obtain the mutex of the selected queue without releasing the pool mutex
+ * because there is still the potential for a claim-vs-destroy race.
+ *
+ * This sequence dictates the lock acquisition ordering for all of
+ * zstream_queue:
+ *
+ *   enqueue -> pool -> queue -> create
+ *
+ * If everyone follows that order, deadlocks can't occur. Unfortunately,
+ * thread_pool_spindown() would like to acquire two of these locks in the
+ * wrong order, so it uses a separate work-around. See the comments for that
+ * function.
+ */
+static int
+assign_queue_and_get_work(zstream_queue_t **queue, queue_slot_t **batch)
+{
+	pthread_mutex_lock(&pool.tp_enqueue_mutex);
+	pthread_mutex_lock(&pool.tp_pool_mutex);
+
+	while (B_TRUE) {
+		int num_queues = pool.tp_num_queues;
+		double weights[MAX_QUEUES];
+		int queues_with_work = 0;
+
+		for (int i = 0; i < num_queues; i++) {
+			weights[i] = score_queue(pool.tp_queues[i]);
+			if (weights[i] > NO_WORK)
+				queues_with_work++;
+		}
+		if (!queues_with_work) {
+			pthread_mutex_unlock(&pool.tp_pool_mutex);
+			await_condition(&pool.tp_enqueued,
+			    &pool.tp_enqueue_mutex);
+			pthread_mutex_lock(&pool.tp_pool_mutex);
+		} else {
+			pthread_mutex_unlock(&pool.tp_enqueue_mutex);
+			int q = select_stochastic(weights, num_queues);
+			*queue = pool.tp_queues[q];
+			pthread_mutex_lock(&(*queue)->zq_mutex);
+			int count = claim_batch(*queue, batch);
+			/*
+			 * If we didn't claim all available work, wake up
+			 * another worker thread.
+			 */
+			boolean_t more_here = (*queue)->zq_ix.claim <
+			    (*queue)->zq_ix.enqueue;
+			if (more_here || queues_with_work > 1) {
+				pthread_cond_signal(&pool.tp_enqueued);
+			}
+			pthread_mutex_unlock(&(*queue)->zq_mutex);
+			pthread_mutex_unlock(&pool.tp_pool_mutex);
+			return (count);
+		}
+	}
+}
+
+static uint32_t items_claimed = 0;  /* Used for tuning/debugging */
+
+static void *
+queue_worker(void *dummy)
+{
+	(void) dummy;
+	zstream_queue_t *queue;
+	queue_slot_t *batch[MAX_BATCH];
+	int count;
+
+	while (B_TRUE) {
+		count = assign_queue_and_get_work(&queue, batch);
+		if (count) {
+			zq_process_item_f *process =
+			    queue->zq_params.qp_process;
+			void *context = queue->zq_params.qp_context;
+			atomic_add_32(&items_claimed, count);
+			/*
+			 * Locking note: we complete the whole batch without
+			 * holding any locks. However, we can't mark items
+			 * as completed without holding the queue lock
+			 * because that creates a race condition with
+			 * advance_completion_index().
+			 */
+			for (int i = 0; i < count; i++) {
+				process(batch[i]->qs_item, context);
+			}
+			pthread_mutex_lock(&queue->zq_mutex);
+			for (int i = 0; i < count; i++) {
+				batch[i]->qs_completed = B_TRUE;
+			}
+			advance_completion_index(queue);
+			atomic_sub_32(&items_claimed, count);
+			pthread_mutex_unlock(&queue->zq_mutex);
+		}
+	}
+	return (NULL);
+}
+
+/*
+ * Implements both _enqueue and _fini. item == NULL for fini.
+ */
+void
+zstream_enqueue(zstream_queue_t *queue, queue_item_t *item)
+{
+	pthread_mutex_lock(&queue->zq_mutex);
+	VERIFY3B(queue->zq_disallow_enqueue, ==, B_FALSE);
+
+	while (Q_FULL(queue)) {
+		await_condition(&queue->zq_cond.dequeued, &queue->zq_mutex);
+	}
+
+	queue_slot_t *slot = &Q_SLOT(queue, queue->zq_ix.enqueue);
+	if (item) {
+		slot->qs_cost =
+		    queue->zq_params.qp_cost(item, queue->zq_params.qp_context);
+		slot->qs_completed = slot->qs_cost == 0;
+		slot->qs_end_of_stream = B_FALSE;
+		memcpy(slot->qs_item, item, queue->zq_params.qp_item_size);
+		if (slot->qs_completed) {
+			advance_completion_index(queue);
+		}
+	} else {
+		slot->qs_cost = 0;
+		slot->qs_completed = B_TRUE;
+		slot->qs_end_of_stream = B_TRUE;
+		queue->zq_disallow_enqueue = B_TRUE;
+	}
+	queue->zq_ix.enqueue++;
+
+#ifdef MONITOR_QUEUES
+	/* Maintain queue usage data per monitor interval */
+	int 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);
+#endif
+
+	pthread_mutex_unlock(&queue->zq_mutex);
+	pthread_mutex_lock(&pool.tp_enqueue_mutex);
+	pthread_cond_signal(&pool.tp_enqueued);
+	pthread_mutex_unlock(&pool.tp_enqueue_mutex);
+}
+
+void
+zstream_queue_fini(zstream_queue_t *queue) {
+	zstream_enqueue(queue, NULL);
+}
+
+/*
+ * Note that this function is not public. The only way to destroy a queue
+ * through the public API is to call zstream_queue_fini(), wait for all
+ * items to be processed, and then dequeue all items.
+ */
+static void
+zstream_queue_destroy(zstream_queue_t *queue)
+{
+	pthread_mutex_lock(&pool.tp_pool_mutex);
+
+	pthread_mutex_destroy(&queue->zq_mutex);
+	pthread_cond_destroy(&queue->zq_cond.dequeued);
+
+	if (pthread_cond_destroy(&queue->zq_cond.completed) != 0) {
+		errx(1, "cannot destroy zstream_queue completed condition - "
+		    "are you attempting to dequeue from multiple threads "
+		    "simultaneously?");
+	}
+
+	free(queue->zq_slots);
+	queue->zq_slots = NULL;
+	free(queue);
+	pool.tp_num_queues--;
+
+	if (pool.tp_num_queues == 0) {
+		thread_pool_spindown();  /* Unlocks pool mutex while running */
+	} else {
+		/* Gaps are not allowed in the tp_queues array */
+		zstream_queue_t **qscan = &pool.tp_queues[0];
+		int i = pool.tp_num_queues;
+		while (*qscan != queue) { qscan++; i--; }
+		memmove(qscan, qscan + 1, i * sizeof (*qscan));
+	}
+	pthread_mutex_unlock(&pool.tp_pool_mutex);
+}
+
+/*
+ * Locking note: if more than one thread attempts to dequeue items
+ * simultaneously, disaster is nearly certain. It will work fine until the
+ * end of the stream, at which point it's a tossup between a race condition
+ * with multiple attempts to destroy the whole queue vs. an attempt to
+ * delete a condition that another thread is waiting on. The latter will be
+ * trapped in zstream_queue_destroy(), but the former will likely just
+ * crash. Hence the warning not to do multithreaded dequeues in
+ * zstream_queue.h.
+ */
+boolean_t
+zstream_dequeue(zstream_queue_t *queue, queue_item_t *item)
+{
+	pthread_mutex_lock(&queue->zq_mutex);
+	while (queue->zq_ix.dequeue >= queue->zq_ix.complete) {
+		await_condition(&queue->zq_cond.completed, &queue->zq_mutex);
+	}
+	queue_slot_t *slot = &Q_SLOT(queue, queue->zq_ix.dequeue);
+	queue->zq_ix.dequeue++;
+	if (slot->qs_end_of_stream) {
+		pthread_mutex_unlock(&queue->zq_mutex);
+		/* Potential race point */
+		zstream_queue_destroy(queue);
+		return (B_FALSE);
+	} else {
+		memcpy(item, slot->qs_item, queue->zq_params.qp_item_size);
+		pthread_cond_signal(&queue->zq_cond.dequeued);
+		pthread_mutex_unlock(&queue->zq_mutex);
+		return (B_TRUE);
+	}
+}
+
+#ifdef	MONITOR_QUEUES
+
+#define	JIFFIES_PER_SEC 100
+#define	SAMPLE_DURATION_US 1000000
+
+/*
+ * 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.
+ */
+static void *
+cpu_and_queue_monitor(void *dummy)
+{
+	(void) dummy;
+	uint64_t period = SAMPLE_DURATION_US;
+	struct timespec clock = {};
+	uint64_t start_us, end_us, delta_jif;
+	uint64_t cpu_jif_prior = 0;
+	uint64_t delta_cpu_jif;
+	long unsigned int utime, stime;
+	char buff[1024];
+	boolean_t interrupt = B_FALSE;
+	FILE *fp;
+
+	/* Wait a few seconds for things to settle into steady state */
+	usleep(3 * 1000 * 1000);
+
+	while (B_TRUE) {
+		usleep(period);
+		fp = fopen("/proc/self/stat", "r");
+		VERIFY3P(fp, !=, NULL);
+		VERIFY3P(fgets(buff, sizeof (buff), fp), !=, NULL);
+		fclose(fp);
+		char *p = strrchr(buff, ')');
+		VERIFY3P(p, !=, NULL);
+		p += 2;  /* skip ") " and fields 3-13 */
+		for (int i = 0; i < 11; i++) {
+			p = strchr(p, ' ');
+			VERIFY3P(p, !=, NULL);
+			p++;
+		}
+		VERIFY3U(sscanf(p, "%lu %lu", &utime, &stime), ==, 2);
+		pthread_mutex_lock(&pool.tp_pool_mutex);
+		clock_gettime(CLOCK_MONOTONIC, &clock);
+		end_us = clock.tv_sec * 1000000 + clock.tv_nsec / 1000;
+		if (cpu_jif_prior) {
+			delta_cpu_jif = utime + stime - cpu_jif_prior;
+			delta_jif = (end_us - start_us) /
+			    (JIFFIES_PER_SEC * 100);
+			double cpu_pct = (double)delta_cpu_jif /
+			    (pool.tp_num_threads * delta_jif);
+			fprintf(stderr, "CPU: %.2f%%  ", 100 * cpu_pct);
+			/* Stop to investigate low CPU usage */
+			if (interrupt && cpu_pct < 0.85 && cpu_pct > 0.1) {
+				kill(getpid(), SIGSTOP);
+			}
+		}
+
+		/* Report queue depths */
+		for (int i = 0; i < pool.tp_num_queues; i++) {
+			zstream_queue_t *q = pool.tp_queues[i];
+			fprintf(stderr, "Queue %d: %d-%d  ", i,
+			    q->zq_stats.min_depth, q->zq_stats.max_depth);
+			q->zq_stats.min_depth = 999999999;
+			q->zq_stats.max_depth = 0;
+		}
+
+		pthread_mutex_unlock(&pool.tp_pool_mutex);
+		fprintf(stderr, "\n");
+		cpu_jif_prior = utime + stime;
+		start_us = end_us;
+	}
+	return (NULL);
+}
+
+static void
+start_monitor_thread(void)
+{
+	static boolean_t started = B_FALSE;
+	pthread_t monitor;
+
+	if (!started) {
+		pthread_create(&monitor, NULL, cpu_and_queue_monitor, NULL);
+		pthread_setname_np(monitor, "monitor-0");
+		pthread_detach(monitor);
+		started = B_TRUE;
+	}
+}
+
+#endif	/* MONITOR_QUEUES */

cmd/zstream/zstream_queue.h

@@ -0,0 +1,138 @@
+// SPDX-License-Identifier: CDDL-1.0
+/*
+ * CDDL HEADER START
+ *
+ * This file and its contents are supplied under the terms of the Common
+ * Development and Distribution License ("CDDL"), version 1.0. You may only use
+ * this file in accordance with the terms of version 1.0 of the CDDL.
+ *
+ * A full copy of the text of the CDDL should have accompanied this source. A
+ * copy of the CDDL is also available via the Internet at
+ * http://www.illumos.org/license/CDDL.
+ *
+ * CDDL HEADER END
+ */
+
+/*
+ * Copyright (c) 2026 by Garth Snyder. All rights reserved.
+ */
+
+#ifndef	_ZSTREAM_QUEUE_H
+#define	_ZSTREAM_QUEUE_H
+
+#ifdef	__cplusplus
+extern "C" {
+#endif
+
+#include <stddef.h>
+#include <sys/stdtypes.h>
+
+/*
+ * This is a generalized implementation of multithreaded FIFO work queues.
+ *
+ * Callers define a fixed item size to be used by each queue and supply two
+ * thread-safe functions that 1) estimate individual items' processing costs
+ * and 2) perform the actual processing. The queue never inspects or
+ * interprets items in the queue, so processing functions can modify them as
+ * 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.
+ *
+ * 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
+ * processing function.
+ *
+ * The cost function is run as items enter the queue, so it's
+ * single-threaded and should return a value promptly. If cost estimation is
+ * expensive and important, use a separate queue to implement it.
+ *
+ * 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 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.
+ *
+ * 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.
+ */
+
+#define	MAX_BATCH 16	/* The most items that can be claimed at once */
+
+typedef void queue_item_t;
+
+typedef struct zstream_queue zstream_queue_t;
+
+/*
+ * Signatures that cost and processing functions must conform to.
+ */
+typedef size_t
+zq_estimate_cost_f(queue_item_t *item, void *context);
+
+typedef void
+zq_process_item_f(queue_item_t *item, void *context);
+
+/*
+ * Set the number of threads to be spawned for queue work. Since all queues
+ * share a thread pool, this value affects all queues. The value must be set
+ * before any queues are created. By default, one thread is spawned for
+ * every CPU core.
+ */
+void
+zstream_queue_set_num_threads(uint_t num_threads);
+
+/*
+ * Create a queue. The qp_context field is passed to the cost and processing
+ * functions and is not examined by the queue itself.
+ */
+
+typedef struct {
+	zq_process_item_f	*qp_process;
+	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 *
+zstream_queue_create(zq_params_t *params);
+
+/*
+ * Submit a work item. Blocks if the queue is full. The work item is
+ * shallow-copied into the queue. Multiple threads may enqueue at once.
+ */
+void
+zstream_enqueue(zstream_queue_t *queue, queue_item_t *item);
+
+/*
+ * Retrieve a completed work item. The caller must provide a buffer into
+ * which the dequeued item is shallow-copied. If the next item is not yet
+ * ready, this call will block.
+ *
+ * If zstream_dequeue returns B_FALSE, the stream is complete. The returned
+ * item is not valid and no further calls may be made on the queue.
+ *
+ * Only one thread may dequeue at once.
+ */
+boolean_t
+zstream_dequeue(zstream_queue_t *queue, queue_item_t *item);
+
+/*
+ * Declare that all items have been submitted. The queue will continue to
+ * function normally for dequeuers and worker threads until zstream_dequeue()
+ * returns B_FALSE, at which point the queue will be destroyed.
+ */
+void
+zstream_queue_fini(zstream_queue_t *queue);
+
+#ifdef	__cplusplus
+}
+#endif
+
+#endif	/* _ZSTREAM_QUEUE_H */

cmd/zstream/zstream_recompress.c

@@ -35,6 +35,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/dmu.h>
 #include <sys/zfs_ioctl.h>
 #include <sys/zio_compress.h>
 #include <sys/zstd/zstd.h>
@@ -44,7 +45,8 @@
 #include "zstream.h"
 #include "zstream_chain.h"
 #include "zstream_modules.h"
-#include "zstream_util.h"
+#include "zstream_queue.h"
+#include "zstream_recompress.h"
 
 #define	MAX_COMPRESSION_STEPS 4
 
@@ -125,20 +127,22 @@ needs_decompression(drr_packet_t *item, compression_spec_t *context)
 	return (needs_modification(item, context));
 }
 
-static disposition_t
-chain_decompress_writes(drr_packet_t *item, compression_spec_t *context)
+/*
+ * We can ignore the context here because it's already been evaluated by the
+ * cost function. If the cost function returned something other than zero,
+ * we have to decompress.
+ */
+static void
+chain_decompress_writes(queue_item_t *item_in, void *context)
 {
-	if (item == NULL)
-		return (D_OK);
+	(void) context;
+	drr_packet_t *item = (drr_packet_t *)item_in;
 
 	dmu_replay_record_t *drr = &item->dp_drr;
-	struct drr_write *drrw	 = &drr->drr_u.drr_write;
+	struct drr_write *drrw	= &drr->drr_u.drr_write;
 	uint8_t *debuff;
 
-	if (drr->drr_type != DRR_WRITE || !needs_decompression(item, context)) {
-		return (D_OK);
-	}
-
+	VERIFY3U(drr->drr_type, ==, DRR_WRITE);
 	debuff = decompress_buffer(item->dp_payload, item->dp_payload_size,
 	    drrw->drr_logical_size, drrw->drr_compressiontype);
 	if (debuff == NULL) {
@@ -152,29 +156,30 @@ chain_decompress_writes(drr_packet_t *item, compression_spec_t *context)
 	item->dp_payload_size = drrw->drr_logical_size;
 	drrw->drr_compressed_size = 0;
 	drrw->drr_compressiontype = 0;
-	return (D_OK);
 }
 
-static disposition_t
-chain_compress_writes(drr_packet_t *item, compression_spec_t *context)
+/*
+ * As with chain_decompress_writes(), all the important decisions were made
+ * by the cost function. If we're here, we need to compress.
+ */
+static void
+chain_compress_writes(queue_item_t *item_in, void *context_in)
 {
-	if (item == NULL)
-		return (D_OK);
+	drr_packet_t *item = (drr_packet_t *)item_in;
+	compression_spec_t *context = (compression_spec_t *)context_in;
 
 	dmu_replay_record_t *drr = &item->dp_drr;
 
-	if (drr->drr_type != DRR_WRITE || !needs_compression(item, context)) {
-		return (D_OK);
-	}
-
-	struct drr_write *drrw  = &drr->drr_u.drr_write;
+	struct drr_write *drrw = &drr->drr_u.drr_write;
 	enum zio_compress ctype = drrw->drr_compressiontype;
 	uint8_t *cbuff;
 	size_t	csize;
 
+	VERIFY3U(drr->drr_type, ==, DRR_WRITE);
 	VERIFY3B(ctype_is_uncompressed(ctype), ==, B_TRUE);
 	cbuff = compress_buffer(item->dp_payload, item->dp_payload_size,
 	    *context, &csize);
+
 	if (cbuff == NULL) {
 		drrw->drr_compressiontype = 0;
 		drrw->drr_compressed_size = 0;
@@ -185,7 +190,51 @@ chain_compress_writes(drr_packet_t *item, compression_spec_t *context)
 		drrw->drr_compressed_size = csize;
 		drrw->drr_compressiontype = context->cs_type;
 	}
-	return (D_OK);
+}
+
+/*
+ * 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.
+ */
+static size_t
+chain_compress_cost(queue_item_t *item_in, void *context_in)
+{
+	compression_spec_t *context = (compression_spec_t *)context_in;
+	drr_packet_t *item = (drr_packet_t *)item_in;
+	dmu_replay_record_t *drr = &item->dp_drr;
+
+	if (drr->drr_type != DRR_WRITE) {
+		return (0);
+	}
+	struct drr_write *drrw = &drr->drr_u.drr_write;
+	return (needs_compression(item, context) ? drrw->drr_logical_size : 0);
+}
+
+/*
+ * Don't decompress packets that aren't compressed. And don't decompress
+ * them if their ultimate fate is to be recompressed using the compression
+ * profile that's already in use.
+ */
+static size_t
+chain_decompress_cost(queue_item_t *item_in, void *context_in)
+{
+	compression_spec_t *context = (compression_spec_t *)context_in;
+	drr_packet_t *item = (drr_packet_t *)item_in;
+	dmu_replay_record_t *drr = &item->dp_drr;
+
+	if (drr->drr_type != DRR_WRITE)
+		return (0);
+
+	struct drr_write *drrw   = &drr->drr_u.drr_write;
+	enum zio_compress ctype  = drrw->drr_compressiontype;
+
+	if (ctype_is_uncompressed(ctype))
+		return (0);
+
+	return (needs_decompression(item, context) ? item->dp_payload_size : 0);
 }
 
 /*
@@ -194,7 +243,7 @@ chain_compress_writes(drr_packet_t *item, compression_spec_t *context)
  * uncompressed).
  */
 chain_step_t
-serial_decompress_writes(compression_spec_t *target)
+parallel_decompress_writes(compression_spec_t *target)
 {
 	int this_spec = next_spec++ % MAX_COMPRESSION_STEPS;
 	compression_spec_t *context = &specs[this_spec];
@@ -205,34 +254,40 @@ serial_decompress_writes(compression_spec_t *target)
 		*context = *target;
 	}
 	chain_step_t step = {
-		.cs_type = CS_SERIAL,
-		.cs_in_size = sizeof (drr_packet_t),
-		.cs_out_size = sizeof (drr_packet_t),
-		.cs_context = context,
-		.cs_serial = {
-		    .process = (zc_serial_process_f *)chain_decompress_writes
-		}
+	    .cs_type = CS_PARALLEL,
+	    .cs_in_size = sizeof (drr_packet_t),
+	    .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
+	    }
 	};
 	return (step);
 }
 
 chain_step_t
-serial_compress_writes(compression_spec_t *target)
+parallel_compress_writes(compression_spec_t *target)
 {
 	int this_spec = next_spec++ % MAX_COMPRESSION_STEPS;
 	compression_spec_t *context = &specs[this_spec];
 
 	VERIFY3P(target, !=, NULL);
 	*context = *target;
+
 	chain_step_t step = {
-		.cs_type = CS_SERIAL,
-		.cs_in_size = sizeof (drr_packet_t),
-		.cs_out_size = sizeof (drr_packet_t),
-		.cs_context = context,
-		.cs_serial = {
-			.process =
-			    (zc_serial_process_f *)chain_compress_writes
-		}
+	    .cs_type = CS_PARALLEL,
+	    .cs_in_size = sizeof (drr_packet_t),
+	    .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
+	    }
 	};
 	return (step);
 }
@@ -242,10 +297,11 @@ zstream_do_recompress(int argc, char *argv[])
 {
 	int c;
 	int level = ZIO_COMPLEVEL_DEFAULT;
+	uint_t num_threads = 0;
 
 	chain_attrs_t attrs = { .ca_command_opts = CA_FORBID_DEDUP };
 
-	while ((c = getopt(argc, argv, "l:")) != -1) {
+	while ((c = getopt(argc, argv, "t:l:")) != -1) {
 		switch (c) {
 		case 'l':
 			if (sscanf(optarg, "%d", &level) != 1) {
@@ -253,6 +309,14 @@ zstream_do_recompress(int argc, char *argv[])
 				zstream_usage();
 			}
 			break;
+		case 't':
+			if (sscanf(optarg, "%u", &num_threads) != 1) {
+				warnx("failed to parse num_threads '%s'",
+				    optarg);
+				zstream_usage();
+			}
+			zstream_queue_set_num_threads(num_threads);
+			break;
 		case '?':
 			warnx("invalid option '%c'", optopt);
 			zstream_usage();
@@ -284,8 +348,8 @@ zstream_do_recompress(int argc, char *argv[])
 
 	zstream_chain_t recompress_chain = {
 		STANDARD_INPUT_STACK(NULL),
-		serial_decompress_writes(&spec),
-		serial_compress_writes(&spec),
+		parallel_decompress_writes(&spec),
+		parallel_compress_writes(&spec),
 		STANDARD_OUTPUT_STACK(NULL)
 	};
 

cmd/zstream/zstream_recompress.h

@@ -28,10 +28,10 @@ extern "C" {
 #include "zstream_io.h"
 
 chain_step_t
-serial_decompress_writes(compression_spec_t *target);
+parallel_decompress_writes(compression_spec_t *target);
 
 chain_step_t
-serial_compress_writes(compression_spec_t *target);
+parallel_compress_writes(compression_spec_t *target);
 
 #ifdef __cplusplus
 }

cmd/zstream/zstream_redup.c

@@ -102,8 +102,11 @@ rdt_lookup(redup_table_t *rdt,
 }
 
 static disposition_t
-chain_redup_writes(drr_packet_t *item, redup_context_t *context)
+chain_redup_writes(void *item_in, void *context_in)
 {
+	drr_packet_t *item = (drr_packet_t *)item_in;
+	redup_context_t *context = (redup_context_t *)context_in;
+
 	if (item == NULL) {
 		return (D_OK);
 	}
@@ -188,7 +191,7 @@ serial_redup_writes(redup_context_t *context)
 		.cs_out_size = sizeof (drr_packet_t),
 		.cs_context = context,
 		.cs_serial = {
-			.process = (zc_serial_process_f *)chain_redup_writes
+			.process = chain_redup_writes
 		}
 	};
 	return (step);

cmd/zstream/zstream_validate.c

@@ -43,8 +43,11 @@ static validate_context_t 	contexts[MAX_VALIDATIONS];
 static int			next_context = 0;
 
 static disposition_t
-chain_validate_records(drr_packet_t *item, validate_context_t *context)
+chain_validate_records(void *item_in, void *context_in)
 {
+	drr_packet_t *item = (drr_packet_t *)item_in;
+	validate_context_t *context = (validate_context_t *)context_in;
+
 	if (item == NULL)
 		return (D_OK);
 
@@ -122,7 +125,7 @@ serial_validate_records(void)
 		.cs_out_size = sizeof (drr_packet_t),
 		.cs_context = context,
 		.cs_serial = {
-		    .process = (zc_serial_process_f *)chain_validate_records,
+		    .process = chain_validate_records,
 		}
 	};
 	return (step);

man/man8/zstream.8

@@ -50,6 +50,7 @@
 .Ar resume_token
 .Nm
 .Cm recompress
+.Op Fl t Ar num_threads
 .Op Fl l Ar level
 .Ar algorithm
 .
@@ -176,17 +177,37 @@ Print summary of converted records.
 .It Xo
 .Nm
 .Cm recompress
+.Op Fl t Ar num_threads
 .Op Fl l Ar level
 .Ar algorithm
 .Xc
-Recompresses a send stream, provided on standard input, using the provided
-algorithm and optional level, and writes the modified stream to standard output.
-All WRITE records in the send stream will be recompressed, unless they fail
-to result in size reduction compared to being left uncompressed.
-The provided algorithm can be any valid value to the
+Recompresses a send stream from standard input using the specified compression
+.Ar algorithm
+and optional
+.Ar level ,
+and writes the modified stream to standard output.
+All WRITE records in the send stream will be compressed unless compression
+fails to reduce their size compared to being left uncompressed.
+The
+.Ar algorithm
+can be any valid value to the
 .Nm compress
 property.
 Note that encrypted send streams cannot be recompressed.
+.Bl -tag -width "-t"
+.It Fl t Ar num_threads
+Specifies the number of compression worker threads.
+By default,
+.Nm zstream Cm recompress
+creates a thread for every CPU core.
+It's generally unproductive to raise
+.Ar num_threads
+above this default.
+On some architectures, contention for resources such as memory caches can
+reduce efficiency when many threads are active.
+On those systems, it may be helpful to run with fewer threads,
+particularly when automating operations on shared servers.
+.El
 .Bl -tag -width "-l"
 .It Fl l Ar level
 Specifies compression level.