zstream: zstream_queue bug fixes and simplifications

[GarthSnyder]

This PR makes several changes to zstream_queue.c aimed at bulletproofing and simplification. There are no API changes.

Remove thread pool spindown

This PR removes code that decomissioned worker threads once the last remaining queue had completed. The interlock between this operation and the creation of new queues complicated the locking system significantly and is known to have introduced at least two subtle locking bugs. With this change, the thread pool will be created once and retained until the process exits.

Remove lock-free queue scoring

The code is designed to work correctly even without scoring threads holding the queue mutexes of the queues they're examining. However, I've confirmed through testing that lock-free operation buys essentially nothing in terms of performance. The code doesn't really change, but it now locks around the scoring, ensuring that there is no possibility of skew among the observed queue indexes. Score skew is still possible (and expected) among queues.

No condition signals without locks

assign_queue_and_get_work() now retains the enqueue mutex until after signaling the "enqueued" condition if a scoring run determines that there is likely work available for more than one worker.

Optimize advancement of the "claim" index for no-work items

advance_completion_index() has been renamed advance_indexes() and sweeps both the "claim" and "completed" indexes. This does not affect correctness, but it reduces the number of empty worker loops. Formerly, advancement of the "claim" index only occured while a thread was actively collecting jobs.

Call advance_indexes() even on zstream_queue_fini()

Enqueues of zero-cost items now consistently trigger a call to advance_indexes().

Round up requested item sizes to a pessimistic boundary

Queues formerly allocated their item workspaces as a single block indexed by slot number. That caused alignment problems when API clients requested peculiarly-shaped buffers. The requested size is now rounded up to a pessimistic boundary.

Two-block memory allocation for new queues

Previously, zstream_queue allocated memory for queue slots and the item workspaces to which they point as a single block. The code is more readable when these are separate allocations.

Motivation and context

I was finally able to catch a deadlock on a GitHub test VM and get a multi-threaded backtrace. It was about what I expected, related to thread pool spindown. This feature is largely hygienic and is just as well done away with.

How has this been tested?

I've been running this in conjunction with the zstream selftest patch, which covers queues extensively. Here, I have inverted the commit order so that this one comes first. The selftests will come right after, but in the interim queues are still used by some of the existing ZTS tests.

Types of Changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Performance enhancement (non-breaking change which improves efficiency)
• Code cleanup (non-breaking change which makes code smaller or more readable)
• Quality assurance (non-breaking change which makes the code more robust against bugs)
• Breaking change (fix or feature that would cause existing functionality to change)
• Library ABI change (libzfs, libzfs_core, libnvpair and libzfsbootenv)
• Documentation (a change to man pages or other documentation)

Checklist

• My code follows the OpenZFS code style requirements.
• I have updated the documentation accordingly.
• I have read the contributing document.
• I have added tests to cover my changes.
• I have run the ZFS Test Suite with this change applied.
• All commit messages are properly formatted and contain Signed-off-by.

[Copilot]

Pull request overview

This PR refactors cmd/zstream/zstream_queue.c to simplify the zstream queue thread-pool lifecycle and locking, while improving robustness around queue index advancement and item workspace allocation/alignment.

Changes:

• Removes thread-pool “spindown” logic and associated locking complexity; pool is initialized once and retained for process lifetime.
• Locks queue scoring and adjusts signaling/locking order around the shared “enqueued” condition to reduce race potential.
• Refactors index sweeping (advance_indexes()), changes queue workspace allocation to be 8-byte aligned, and separates slot vs. item workspace allocations.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 1 out of 1 changed files in this pull request and generated 1 comment.

Suppressed comments (3)

cmd/zstream/zstream_queue.c:626

• Queue item storage is freed via queue->zq_slots[0].qs_item, relying on the invariant that all qs_item pointers come from a single contiguous allocation and that index 0 always points to the base. This ownership model is implicit and fragile (e.g., future changes to per-slot allocation or different wiring would silently break free()). Consider storing the base items pointer explicitly in struct zstream_queue (e.g., zq_items_base) and freeing that, which makes ownership and lifetime unambiguous.

	free(queue->zq_slots[0].qs_item);
	free(queue->zq_slots);
	queue->zq_slots = NULL;

cmd/zstream/zstream_queue.c:180

• This now allows zstream_queue_set_num_threads() to set tp_num_threads below MIN_THREADS (only warning), but the auto-detected path still enforces MAX(..., MIN_THREADS). That creates inconsistent behavior depending on whether the value is user-set vs auto-detected. To keep semantics consistent, either (a) clamp user-provided n up to MIN_THREADS, or (b) remove the MIN_THREADS enforcement from the auto-detected path if the minimum is no longer required.

	} else if (n < MIN_THREADS) {
		warnx("using only %u threads may limit performance, setting "
		    "anyway...", n);
	} else if (n > 256) {
		warnx("num_threads = %u seems suspiciously high, setting "
		    "anyway...", n);
	}
	pool.tp_num_threads = n;

cmd/zstream/zstream_queue.c:633

• With thread_pool_spindown() removed and no equivalent path when tp_num_queues reaches 0, worker threads and pool.tp_threads appear to persist indefinitely after the last queue is destroyed. If this code is used in a long-running process that can create/destroy queues multiple times, this changes lifecycle behavior and can retain resources (threads, stack memory) unexpectedly. Consider either restoring a safe spindown when the last queue is destroyed, or introducing an explicit global/thread-pool shutdown API/atexit cleanup to ensure resources can be reclaimed deterministically.

	if (pool.tp_num_queues > 0) {
		/* Gaps are not allowed in the tp_queues array */
		zstream_queue_t **qscan = &pool.tp_queues[0];
		int i = pool.tp_num_queues;

[Copilot]

Pull request overview

Copilot reviewed 1 out of 1 changed files in this pull request and generated 2 comments.

Suppressed comments (2)

cmd/zstream/zstream_queue.c:626

• Freeing the item storage via queue->zq_slots[0].qs_item couples destruction logic to an implicit invariant (single contiguous item allocation and slot[0] pointing to its base). To make ownership explicit and less fragile, store the allocated items base pointer in the queue struct (e.g., zq_items_base) and free that directly in zstream_queue_destroy().

	free(queue->zq_slots[0].qs_item);
	free(queue->zq_slots);

cmd/zstream/zstream_queue.c:490

• pthread_cond_wait() is a cancellation point. Previously, this file used a cleanup handler wrapper to ensure the mutex is released if a waiting thread is cancelled. If worker thread cancellation is still possible (now or in future changes), cancellation during this wait can exit while holding tp_enqueue_mutex, risking deadlock. Either ensure worker threads are never cancelled and enforce that invariant, or restore a cleanup-handler approach / temporarily disable cancellation around waits.

			pthread_cond_wait(&pool.tp_enqueued,
			    &pool.tp_enqueue_mutex);

[Copilot]

Pull request overview

Copilot reviewed 1 out of 1 changed files in this pull request and generated 3 comments.

Suppressed comments (4)

cmd/zstream/zstream_queue.c:251

• Rounding item strides to 8 bytes doesn’t guarantee correct alignment for queue_item_t on all platforms/ABIs (e.g., if queue_item_t requires >8-byte alignment). This can lead to undefined behavior when qp_process dereferences qs_item. Prefer rounding up to _Alignof(queue_item_t) (or _Alignof(max_align_t) if queue_item_t is opaque) rather than a hard-coded 8.

	size_t qpis_rounded = P2ROUNDUP(params->qp_item_size, 8);
	uint8_t *items = safe_malloc(params->qp_queue_length * qpis_rounded);
	for (int i = 0; i < params->qp_queue_length; i++) {
		queue->zq_slots[i].qs_item =
		    (queue_item_t *)(items + i * qpis_rounded);
	}

cmd/zstream/zstream_queue.c:627

• Freeing the item-storage block via queue->zq_slots[0].qs_item creates a fragile coupling between slot 0 and the allocation base address. A more robust approach is to store the items base pointer explicitly in struct zstream_queue (e.g., zq_items_base) and free that, which avoids relying on slot contents never changing and makes ownership clearer.

	free(queue->zq_slots[0].qs_item);
	free(queue->zq_slots);
	queue->zq_slots = NULL;

cmd/zstream/zstream_queue.c:176

• This changes the API behavior from rejecting values < MIN_THREADS to allowing them. If other code relies on MIN_THREADS as a correctness/sizing invariant (not just a performance recommendation), this can introduce subtle scheduling/throughput issues. Consider either (a) preserving the previous hard validation for < MIN_THREADS, or (b) enforcing pool.tp_num_threads = MAX(n, MIN_THREADS) even for user-provided values and adjusting the warning messaging accordingly.

	} else if (n == 0) {
		errx(1, "number of threads must be at least 1");
	} else if (n < MIN_THREADS) {
		warnx("using only %u threads may limit performance, setting "
		    "anyway...", n);
	} else if (n > 256) {

cmd/zstream/zstream_queue.c:634

• With thread_pool_spindown() removed, the worker threads and pool.tp_threads allocation are never torn down when the last queue is destroyed. In long-running processes or repeated create/destroy cycles, this can cause permanent thread/resource retention. Consider adding an explicit thread-pool shutdown path (e.g., when tp_num_queues reaches 0) that cleanly stops workers and frees pool.tp_threads, or provide a documented global “fini” API for callers to release resources deterministically.

	if (pool.tp_num_queues > 0) {
		/* Gaps are not allowed in the tp_queues array */
		zstream_queue_t **qscan = &pool.tp_queues[0];
		int i = pool.tp_num_queues;

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated no new comments.

Suppressed comments (2)

cmd/zstream/zstream_queue.c:180

• zstream_queue_set_num_threads(0) now fatals, but thread_pool_spinup() explicitly treats tp_num_threads == 0 as the "use one thread per core" default. Since the header docs also describe 0 as the implicit default, rejecting 0 here changes the valid parameter contract and makes it impossible to explicitly reset/request the default behavior.

	if (pool.tp_threads != NULL) {
		errx(1, "thread pool size must be set before creating queues");
	} else if (n == 0) {
		errx(1, "number of threads must be at least 1");
	} else if (n < MIN_THREADS) {

cmd/zstream/zstream_queue.c:235

• zstream_queue_create() rejects qp_batch_budget == 0, but claim_batch() still contains explicit logic for the qp_batch_budget == 0 case (first_and_only). Either 0 is meant to be a supported value (e.g., force single-item batches) or the special-case should be removed; as written, the validation makes that code path unreachable and may break callers that previously relied on 0 being valid.

	VERIFY3P(params->qp_process, !=, NULL);
	VERIFY3P(params->qp_cost, !=, NULL);
	VERIFY3U(params->qp_item_size, >, 0);
	VERIFY3U(params->qp_batch_budget, >, 0);
	VERIFY3U(params->qp_queue_length, >, 0);