diff --git a/cmd/zstream/Makefile.am b/cmd/zstream/Makefile.am index 72c421ad3bfa..1efd5251433d 100644 --- a/cmd/zstream/Makefile.am +++ b/cmd/zstream/Makefile.am @@ -7,6 +7,8 @@ CPPCHECKTARGETS += zstream zstream_SOURCES = \ %D%/zstream.c \ %D%/zstream.h \ + %D%/zstream_backtrace.c \ + %D%/zstream_backtrace.h \ %D%/zstream_byteswap.c \ %D%/zstream_byteswap.h \ %D%/zstream_chain.c \ @@ -24,6 +26,9 @@ zstream_SOURCES = \ %D%/zstream_recompress.c \ %D%/zstream_recompress.h \ %D%/zstream_redup.c \ + %D%/zstream_selftest.c \ + %D%/zstream_selftest.h \ + %D%/zstream_selftest_queue.c \ %D%/zstream_token.c \ %D%/zstream_queue.c \ %D%/zstream_queue.h \ diff --git a/cmd/zstream/zstream.c b/cmd/zstream/zstream.c index db43a4bbff3b..450f9a4c35b8 100644 --- a/cmd/zstream/zstream.c +++ b/cmd/zstream/zstream.c @@ -19,11 +19,14 @@ * Copyright (c) 2020 by Datto Inc. All rights reserved. */ +#include +#include #include #include #include #include "zstream.h" +#include "zstream_util.h" void zstream_usage(void) @@ -50,9 +53,27 @@ zstream_usage(void) exit(1); } +/* + * Set the signal mask to allow THREAD_BACKTRACE_SIGNAL. WATCHDOG_SIGNAL + * must be blocked in all threads so that its intended recipient can listen + * for it with sigwait(), which detects only pending signals. + */ +static void +set_signal_mask(void) +{ + sigset_t mask; + + safe_pthread_sigmask(SIG_SETMASK, NULL, &mask); + sigaddset(&mask, WATCHDOG_SIGNAL); + sigdelset(&mask, THREAD_BACKTRACE_SIGNAL); + safe_pthread_sigmask(SIG_SETMASK, &mask, NULL); +} + int main(int argc, char *argv[]) { + set_signal_mask(); + char *basename = strrchr(argv[0], '/'); basename = basename ? (basename + 1) : argv[0]; if (argc >= 1 && strcmp(basename, "zstreamdump") == 0) @@ -77,6 +98,9 @@ main(int argc, char *argv[]) return (zstream_do_token(argc - 1, argv + 1)); } else if (strcmp(subcommand, "redup") == 0) { return (zstream_do_redup(argc - 1, argv + 1)); + } else if (strcmp(subcommand, "selftest") == 0) { + /* Undocumented; used by the ZFS test suite */ + return (zstream_do_selftest(argc - 1, argv + 1)); } else { zstream_usage(); } diff --git a/cmd/zstream/zstream.h b/cmd/zstream/zstream.h index b38d30a0f647..528394553452 100644 --- a/cmd/zstream/zstream.h +++ b/cmd/zstream/zstream.h @@ -21,10 +21,26 @@ #ifndef _ZSTREAM_H #define _ZSTREAM_H +#include + #ifdef __cplusplus extern "C" { #endif +/* + * Signals used by the watchdog timer and zstream_queue. The signal mask + * must be set properly before any threads are spawned. + */ +#define WATCHDOG_SIGNAL SIGALRM +#define THREAD_BACKTRACE_SIGNAL (SIGRTMIN) + +/* + * Establishes the process-wide signal mask. Must be called before any + * thread is created so that every thread inherits the mask without having + * to make any calls of its own. + */ +extern void zstream_signal_init(void); + extern int zstream_do_redup(int, char *[]); extern int zstream_do_dump(int, char *[]); extern int zstream_do_decompress(int argc, char *argv[]); @@ -32,6 +48,7 @@ extern int zstream_do_drop_record(int argc, char *argv[]); extern int zstream_do_recompress(int argc, char *argv[]); extern int zstream_do_token(int, char *[]); extern int zstream_do_raw(int, char *[]); +extern int zstream_do_selftest(int, char *[]); extern void zstream_usage(void) __attribute__((noreturn)); #ifdef __cplusplus diff --git a/cmd/zstream/zstream_backtrace.c b/cmd/zstream/zstream_backtrace.c new file mode 100644 index 000000000000..c9fcc4165f2c --- /dev/null +++ b/cmd/zstream/zstream_backtrace.c @@ -0,0 +1,187 @@ +// 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. + */ + +/* + * This is a watchdog timer and multithread backtrace dumper that's used by + * zstream selftest. libspl_backtrace() does all the real work, but it can + * only dump the current thread's stack. We need to get every thread to call + * libspl_backtrace() in an organized sequence. However, there's no "give me + * a list of all pthreads" function in the POSIX API. + * + * Rather than constructing an ad-hoc thread registry, we can approach the + * problem by unblocking SIGRTMIN (an arbitrary choice) and designating a + * handler for it when zstream first starts. All created threads then + * inherit this signal mask and handler. + * + * The SIGRTMIN handler calls libspl_backtrace(), which is signal-handler + * safe. It then signals a semaphore to indicate that it's finished and + * enters pause(). + * + * The watchdog itself is a separate thread that sigwait()s for SIGALRM. It + * blocks SIGRTMIN at the thread level and runs libspl_backtrace(). It then + * enters a loop in which it sends SIGRTMIN to the process as a whole and + * runs sem_timedwait() to see if any thread woke up and dumped its + * backtrace. If that call times out, either the receiving thread wedged + * while trying to backtrace or there were no more threads to backtrace. + * + * These two scenarios can be distinguished by calling sigpending(). If a + * SIGRTMIN still shows as being pending on the process, then there was no + * thread to receive it and we are done. If there's no pending signal, then + * some thread did receive the signal but failed to post to the semaphore; + * we print a "thread wedged while backtracing" message and continue the + * loop. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "zstream.h" +#include "zstream_backtrace.h" + +/* + * Watchdog timeout in seconds. A test that hits this limit is almost + * certainly deadlocked, and the watchdog converts the hang into a test + * failure instead of a stuck test run. + */ +#define WATCHDOG_TIMEOUT_SECS 120 +#define MAX_SECS_FOR_BACKTRACE 2 + +static sem_t sem_thread_bt_complete; /* From thread to watchdog */ + +/* + * Signal handler for THREAD_BACKTRACE_SIGNAL, run by all threads except the + * watchdog thread + */ +static void +backtrace_self(int signal) +{ + (void) signal; + ssize_t dummy __maybe_unused = write(STDERR_FILENO, "\n", 1); + libspl_backtrace(STDERR_FILENO); + sem_post(&sem_thread_bt_complete); + + sigset_t mask; + sigfillset(&mask); + sigsuspend(&mask); +} + +static void +backtrace_all_threads(void) +{ + while (B_TRUE) { + if (kill(getpid(), THREAD_BACKTRACE_SIGNAL) != 0) + err(1, "failed to send thread backtrace signal"); + struct timespec deadline; + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_sec += MAX_SECS_FOR_BACKTRACE; + if (sem_timedwait(&sem_thread_bt_complete, &deadline) != 0) { + sigset_t pending; + if (sigpending(&pending) != 0) + err(1, "sigpending failed"); + if (sigismember(&pending, THREAD_BACKTRACE_SIGNAL)) { + return; + } else { + warnx("a thread failed to generate a backtrace," + " continuing..."); + } + } + } +} + +/* + * Body of the watchdog thread + */ +static void * +watchdog(void *nope) +{ + (void) nope; + int signal; + sigset_t bt_mask, dog_mask; + + /* + * This thread does its own backtrace, so we block the + * backtrace signal. + */ + sigemptyset(&bt_mask); + sigaddset(&bt_mask, THREAD_BACKTRACE_SIGNAL); + if (pthread_sigmask(SIG_BLOCK, &bt_mask, NULL) != 0) + err(1, "pthread_sigmask failed"); + + sigemptyset(&dog_mask); + sigaddset(&dog_mask, WATCHDOG_SIGNAL); + + int rc = sigwait(&dog_mask, &signal); + if (rc != 0) { + errno = rc; + err(1, "watchdog sigwait failed"); + } else if (signal != WATCHDOG_SIGNAL) { + errx(1, "unexpected signal %d received by watchdog", signal); + } + + fprintf(stderr, "\n\nWATCHDOG TIMER EXPIRED\n" + "Dumping backtrace for all threads...\n\n"); + fflush(stderr); + libspl_backtrace(STDERR_FILENO); + backtrace_all_threads(); + fprintf(stderr, "\nAll threads backtraced, exiting.\n"); + exit(1); +} + +/* + * This function must be called before any (extra) threads are created. + */ +void +watchdog_init(void) +{ + if (sem_init(&sem_thread_bt_complete, 0, 0) != 0) + err(1, "watchdog sem_init failed"); + + safe_create_thread(watchdog, NULL, "watchdog", B_TRUE); + + struct sigaction sa = { + .sa_handler = backtrace_self, + .sa_flags = SA_RESTART + }; + if (sigaction(THREAD_BACKTRACE_SIGNAL, &sa, NULL) != 0) + err(1, "backtrace sigaction failed"); +} + +void +watchdog_arm(void) +{ + (void) alarm(WATCHDOG_TIMEOUT_SECS); +} + +void +watchdog_disarm(void) +{ + (void) alarm(0); +} diff --git a/cmd/zstream/zstream_backtrace.h b/cmd/zstream/zstream_backtrace.h new file mode 100644 index 000000000000..12fbdee47724 --- /dev/null +++ b/cmd/zstream/zstream_backtrace.h @@ -0,0 +1,40 @@ +// 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_BACKTRACE_H +#define _ZSTREAM_BACKTRACE_H + +#ifdef __cplusplus +extern "C" { +#endif + +extern void +watchdog_init(void); + +void +watchdog_arm(void); + +void +watchdog_disarm(void); + +#ifdef __cplusplus +} +#endif + +#endif /* _ZSTREAM_BACKTRACE_H */ diff --git a/cmd/zstream/zstream_chain.c b/cmd/zstream/zstream_chain.c index 35e46dd429b6..cddaf1d5e194 100644 --- a/cmd/zstream/zstream_chain.c +++ b/cmd/zstream/zstream_chain.c @@ -33,8 +33,11 @@ #include #include +#include "zstream.h" #include "zstream_chain.h" #include "zstream_queue.h" +#include "zstream_selftest.h" +#include "zstream_util.h" #define MAX_CHAIN_LENGTH 32 @@ -60,8 +63,6 @@ typedef struct { zstream_queue_t *wc_out_queue; } worker_context_t; -typedef void *chain_worker_f(void *); - chain_attrs_t *chain_attrs; chain_step_t @@ -265,13 +266,11 @@ zstream_chain_exec(zstream_chain_t chain, chain_attrs_t *attrs) /* 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); + char name[32]; + snprintf(name, sizeof (name), "chain-%d", i); + worker_threads[i] = safe_pthread_create( + (thread_f *)zstream_chain_worker, &contexts[i], + name, B_FALSE); } /* Reap threads */ diff --git a/cmd/zstream/zstream_queue.c b/cmd/zstream/zstream_queue.c index 37cfc35d4180..4cb44782c4b2 100644 --- a/cmd/zstream/zstream_queue.c +++ b/cmd/zstream/zstream_queue.c @@ -31,6 +31,7 @@ #include #include +#include "zstream.h" #include "zstream_queue.h" #include "zstream_util.h" @@ -71,11 +72,10 @@ * * THREAD SAFETY STRATEGY * - * There are four types of lock: + * There are three 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 + * - A second global lock associated with the shared "enqueued" condition * - One lock for each queue * * Although the "enqueued" condition and its associated lock are stored as @@ -92,11 +92,7 @@ * 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. + * enqueue -> pool -> queue */ typedef struct { @@ -135,16 +131,20 @@ struct zstream_queue { 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; + boolean_t tp_threads_created; int tp_num_threads; } thread_pool_t; -typedef void cleanup_f(void *); +typedef union { + long long ll; + long double ld; + void *p; + void (*fp)(void); +} worst_case_alignment_t; static void * queue_worker(void *); @@ -155,38 +155,44 @@ 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; +static void +thread_pool_init(void) +{ + pthread_mutex_init(&pool.tp_pool_mutex, NULL); + pthread_mutex_init(&pool.tp_enqueue_mutex, NULL); + pthread_cond_init(&pool.tp_enqueued, NULL); +} + +/* + * If this function is to be called at all, it must be called before any + * queues have been created. + */ void zstream_queue_set_num_threads(uint_t n) { - if (pool_initialized) { + pthread_once(&once_control, thread_pool_init); + pthread_mutex_lock(&pool.tp_pool_mutex); + if (pool.tp_threads_created) { 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) { - errx(1, "number of threads must be at least %d", 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); } - 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; + pool.tp_num_threads = n; + pthread_mutex_unlock(&pool.tp_pool_mutex); } /* - * Locking note: must be called by a function holding the pool mutex + * Locking: the caller must hold the pool mutex. * - * If num_threads is nonzero, it sets the number of threads to spawn. + * If tp_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 @@ -196,10 +202,7 @@ thread_pool_init(void) static void thread_pool_spinup(void) { - pool.tp_num_queues = 0; - if (num_threads > 0) { - pool.tp_num_threads = num_threads; - } else { + if (pool.tp_num_threads == 0) { #ifdef CPU_COUNT cpu_set_t cpu_set; sched_getaffinity(0, sizeof (cpu_set_t), &cpu_set); @@ -207,95 +210,49 @@ thread_pool_spinup(void) #else pool.tp_num_threads = sysconf(_SC_NPROCESSORS_ONLN); #endif + pool.tp_num_threads = MAX(pool.tp_num_threads, MIN_THREADS); } - 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); + char name[32]; + snprintf(name, sizeof (name), "queue-%d", i); + safe_pthread_create(queue_worker, NULL, name, B_TRUE); } #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) { + VERIFY3P(params->qp_process, !=, NULL); + VERIFY3P(params->qp_cost, !=, NULL); + VERIFY3U(params->qp_item_size, >, 0); + VERIFY3U(params->qp_queue_length, >, 0); + 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) { + if (!pool.tp_threads_created) { thread_pool_spinup(); + pool.tp_threads_created = B_TRUE; } 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 = (zstream_queue_t) { + .zq_params = *params, + .zq_slots = safe_malloc(params->qp_queue_length * + (sizeof (queue_slot_t))) }; - *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]; + + size_t qpis_rounded = P2ROUNDUP(params->qp_item_size, + _Alignof(worst_case_alignment_t)); + uint8_t *items = safe_malloc(params->qp_queue_length * qpis_rounded); for (int i = 0; i < params->qp_queue_length; i++) { - slot->qs_item = item; - item += queue->zq_params.qp_item_size; - slot++; + queue->zq_slots[i].qs_item = + (queue_item_t *)(items + i * qpis_rounded); } pthread_mutex_init(&queue->zq_mutex, NULL); @@ -303,22 +260,19 @@ zstream_queue_create(zq_params_t *params) 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: + * Try to advance the "claim" and "complete" indexes 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. + * advance the completion index past them on your way out. * * - 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 @@ -333,15 +287,25 @@ zstream_queue_create(zq_params_t *params) * * 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. + * it's in our interest to make completed items available for dequeueing + * as expeditiously as possible. + * + * It's also expedient to sweep the "claim" index if we can. This is not + * necessary for correctness. However, if we don't do it here, it can only + * be done by threads as they claim jobs to work on. In some cases, not + * advancing the "claim" index here can result in an empty batch and a + * wasted claim cycle. * - * Locking note: the calling thread must hold the queue mutex. + * Locking: the caller must hold the queue mutex. */ static inline void -advance_completion_index(zstream_queue_t *queue) +advance_indexes(zstream_queue_t *queue) { boolean_t any_completed = B_FALSE; + while (queue->zq_ix.claim < queue->zq_ix.enqueue && + Q_SLOT(queue, queue->zq_ix.claim).qs_completed) { + queue->zq_ix.claim++; + } while (queue->zq_ix.complete < queue->zq_ix.claim && Q_SLOT(queue, queue->zq_ix.complete).qs_completed) { queue->zq_ix.complete++; @@ -353,9 +317,6 @@ advance_completion_index(zstream_queue_t *queue) } /* - * 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 @@ -372,14 +333,13 @@ advance_completion_index(zstream_queue_t *queue) * 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: + * Locking: the caller must hold the enqueue mutex, the thread pool mutex, + * and the queue mutex. 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. + * 2) Worker threads can still complete work during the process of scoring + * all queues, so there will likely be time skew among scores. * * 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 @@ -434,29 +394,14 @@ select_stochastic(double weights[], int num_values) 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. + * Does not block waiting to fill the budget; returns whatever is available. * - * Locking note: this function must be called with both the queue mutex and + * Locking: this function must be called with both the queue mutex and * the thread pool mutex held. zstream_queue_destroy() can't hold a queue's * mutex while destroying it (because destruction entails destroying the * queue mutex, which must be unlocked), so holding the queue mutex while @@ -497,34 +442,33 @@ claim_batch(zstream_queue_t *queue, queue_slot_t **batch) queue->zq_ix.claim++; } - advance_completion_index(queue); + advance_indexes(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. + * 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). + * Locking: 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). We also lock individual queues while scoring them. * - * 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. + * After queue selection, we retain the enqueue mutex while claiming a batch + * so that we can safely signal the "enqueued" condition if there appears to + * be enough work for more than one thread dispatch. We need to 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 + * enqueue -> pool -> queue * - * 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. + * If everyone follows that order, deadlocks should not occur. */ static int assign_queue_and_get_work(zstream_queue_t **queue, queue_slot_t **batch) @@ -538,17 +482,19 @@ assign_queue_and_get_work(zstream_queue_t **queue, queue_slot_t **batch) int queues_with_work = 0; for (int i = 0; i < num_queues; i++) { - weights[i] = score_queue(pool.tp_queues[i]); + zstream_queue_t *to_score = pool.tp_queues[i]; + pthread_mutex_lock(&to_score->zq_mutex); + weights[i] = score_queue(to_score); + pthread_mutex_unlock(&to_score->zq_mutex); if (weights[i] > NO_WORK) queues_with_work++; } if (!queues_with_work) { pthread_mutex_unlock(&pool.tp_pool_mutex); - await_condition(&pool.tp_enqueued, + pthread_cond_wait(&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); @@ -559,17 +505,18 @@ assign_queue_and_get_work(zstream_queue_t **queue, queue_slot_t **batch) */ boolean_t more_here = (*queue)->zq_ix.claim < (*queue)->zq_ix.enqueue; + pthread_mutex_unlock(&(*queue)->zq_mutex); 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); + pthread_mutex_unlock(&pool.tp_enqueue_mutex); return (count); } } } -static uint32_t items_claimed = 0; /* Used for tuning/debugging */ +static uint64_t items_in_claimed_state = 0; /* Used for tuning/debugging */ static void * queue_worker(void *dummy) @@ -585,13 +532,13 @@ queue_worker(void *dummy) zq_process_item_f *process = queue->zq_params.qp_process; void *context = queue->zq_params.qp_context; - atomic_add_32(&items_claimed, count); + atomic_add_64(&items_in_claimed_state, 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(). + * advance_indexes(). */ for (int i = 0; i < count; i++) { process(batch[i]->qs_item, context); @@ -600,8 +547,8 @@ queue_worker(void *dummy) for (int i = 0; i < count; i++) { batch[i]->qs_completed = B_TRUE; } - advance_completion_index(queue); - atomic_sub_32(&items_claimed, count); + advance_indexes(queue); + atomic_sub_64(&items_in_claimed_state, count); pthread_mutex_unlock(&queue->zq_mutex); } } @@ -614,11 +561,12 @@ queue_worker(void *dummy) void zstream_enqueue(zstream_queue_t *queue, queue_item_t *item) { + VERIFY(queue != NULL); 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); + pthread_cond_wait(&queue->zq_cond.dequeued, &queue->zq_mutex); } queue_slot_t *slot = &Q_SLOT(queue, queue->zq_ix.enqueue); @@ -628,9 +576,6 @@ zstream_enqueue(zstream_queue_t *queue, queue_item_t *item) 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; @@ -638,6 +583,9 @@ zstream_enqueue(zstream_queue_t *queue, queue_item_t *item) queue->zq_disallow_enqueue = B_TRUE; } queue->zq_ix.enqueue++; + if (slot->qs_completed) { + advance_indexes(queue); + } #ifdef MONITOR_QUEUES /* Maintain queue usage data per monitor interval */ @@ -658,9 +606,12 @@ zstream_queue_fini(zstream_queue_t *queue) { } /* - * 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. + * 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. + * + * Locking: the caller must NOT hold the queue lock. The pool mutex is + * held while destroying the queue. */ static void zstream_queue_destroy(zstream_queue_t *queue) @@ -676,14 +627,13 @@ zstream_queue_destroy(zstream_queue_t *queue) "simultaneously?"); } + free(queue->zq_slots[0].qs_item); 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 { + 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; @@ -694,27 +644,29 @@ zstream_queue_destroy(zstream_queue_t *queue) } /* - * 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. + * Locking: if more than one thread attempts to dequeue items + * simultaneously, disaster is likely. 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. + * + * Returns B_TRUE if real data is returned, B_FALSE if the end of the queue + * has been reached. */ 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); + pthread_cond_wait(&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 */ + /* Potential multi-dequeuer race point */ zstream_queue_destroy(queue); return (B_FALSE); } else { @@ -807,9 +759,8 @@ start_monitor_thread(void) pthread_t monitor; if (!started) { - pthread_create(&monitor, NULL, cpu_and_queue_monitor, NULL); - pthread_setname_np(monitor, "monitor-0"); - pthread_detach(monitor); + safe_pthread_create(cpu_and_queue_monitor, NULL, + "monitor", B_TRUE); started = B_TRUE; } } diff --git a/cmd/zstream/zstream_queue.h b/cmd/zstream/zstream_queue.h index 73425ed01f25..bcd78709ea78 100644 --- a/cmd/zstream/zstream_queue.h +++ b/cmd/zstream/zstream_queue.h @@ -55,14 +55,18 @@ extern "C" { * threads never block waiting for additional work to arrive. They start * work as quickly as possible even if the budget has not been reached. * + * A batch budget of 0 means that all batches will have a size of 1. + * * 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. + * + * The shared thread pool persists until the process exits. */ -#define MAX_BATCH 16 /* The most items that can be claimed at once */ +#define MAX_BATCH 32 /* The most items that can be claimed at once */ typedef void queue_item_t; diff --git a/cmd/zstream/zstream_selftest.c b/cmd/zstream/zstream_selftest.c new file mode 100644 index 000000000000..ff69433eadab --- /dev/null +++ b/cmd/zstream/zstream_selftest.c @@ -0,0 +1,212 @@ +// 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. + */ + +/* + * zstream selftest: in-process unit tests for zstream's internal machinery. + * + * zstream selftest [-l] [-s seed] [-t nthreads] module [test ...] + * + * Tests are grouped into modules (see zstream_selftest.h). With no test + * names, all of a module's tests run in order. -l lists the available + * tests. -s replays a previous run's PRNG seed. -t sets the size of the + * shared worker thread pool before any test runs. + * + * This subcommand is intentionally undocumented in zstream_usage() and the + * man page; it exists to be driven by the ZFS test suite. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "zstream.h" +#include "zstream_backtrace.h" +#include "zstream_queue.h" +#include "zstream_selftest.h" + +typedef struct { + const char *sm_name; + const test_case_t *sm_cases; +} selftest_module_t; + +static const selftest_module_t modules[] = { + { "queue", selftest_queue_cases }, +}; + +#define NUM_MODULES (sizeof (modules) / sizeof (modules[0])) + +uint64_t selftest_seed; + +static const char *current_test = "(startup)"; + +static void +selftest_usage(void) +{ + (void) fprintf(stderr, + "usage: zstream selftest [-l] [-s seed] [-t nthreads] " + "module [test ...]\n" + "\n" + "\t-l list available tests\n" + "\t-s seed seed for pseudo-random workloads (for replays)\n" + "\t-t num size of the shared worker thread pool\n" + "\n" + "Available modules:"); + for (int i = 0; i < NUM_MODULES; i++) + (void) fprintf(stderr, " %s", modules[i].sm_name); + (void) fprintf(stderr, "\n"); + exit(1); +} + +static const selftest_module_t * +find_module(const char *name) +{ + for (int i = 0; i < NUM_MODULES; i++) { + if (strcmp(name, modules[i].sm_name) == 0) + return (&modules[i]); + } + warnx("unknown module '%s'", name); + selftest_usage(); + return (NULL); /* NOTREACHED */ +} + +static void +list_tests(const selftest_module_t *module) +{ + for (int i = 0; i < NUM_MODULES; i++) { + if (module != NULL && module != &modules[i]) + continue; + (void) printf("%s:\n", modules[i].sm_name); + for (const test_case_t *tc = modules[i].sm_cases; + tc->tc_name != NULL; tc++) { + (void) printf("\t%s\n", tc->tc_name); + } + } +} + +static void +run_case(const test_case_t *tc) +{ + (void) printf("Running %-20s ... ", tc->tc_name); + (void) fflush(stdout); + current_test = tc->tc_name; + watchdog_arm(); + tc->tc_func(); + watchdog_disarm(); + (void) printf("OK\n"); +} + +static const test_case_t * +find_case(const selftest_module_t *module, const char *name) +{ + for (const test_case_t *tc = module->sm_cases; + tc->tc_name != NULL; tc++) { + if (strcmp(name, tc->tc_name) == 0) + return (tc); + } + errx(2, "module '%s' has no test named '%s' (try -l)", + module->sm_name, name); + return (NULL); /* NOTREACHED */ +} + +int +zstream_do_selftest(int argc, char *argv[]) +{ + boolean_t list_only = B_FALSE; + boolean_t have_seed = B_FALSE; + uint_t nthreads = 0; + char *end; + int c; + + while ((c = getopt(argc, argv, "ls:t:")) != -1) { + switch (c) { + case 'l': + list_only = B_TRUE; + break; + case 's': + selftest_seed = strtoull(optarg, &end, 0); + if (*optarg == '\0' || *end != '\0') { + warnx("failed to parse seed '%s'", optarg); + selftest_usage(); + } + have_seed = B_TRUE; + break; + case 't': + if (sscanf(optarg, "%u", &nthreads) != 1 || + nthreads == 0) { + warnx("failed to parse num_threads '%s'", + optarg); + selftest_usage(); + } + break; + case '?': + warnx("invalid option '%c'", optopt); + selftest_usage(); + break; + } + } + argc -= optind; + argv += optind; + + if (list_only) { + list_tests(argc > 0 ? find_module(argv[0]) : NULL); + return (0); + } + if (argc < 1) + selftest_usage(); + + const selftest_module_t *module = find_module(argv[0]); + + /* Needed for random_get_pseudo_bytes() */ + libspl_init(); + watchdog_init(); + + if (!have_seed) + random_get_pseudo_bytes((uint8_t *)&selftest_seed, + sizeof (selftest_seed)); + (void) printf("Using seed 0x%016jx (replay with -s 0x%jx)\n", + (uintmax_t)selftest_seed, (uintmax_t)selftest_seed); + + if (nthreads > 0) + zstream_queue_set_num_threads(nthreads); + + + int count = 0; + if (argc == 1) { + for (const test_case_t *tc = module->sm_cases; + tc->tc_name != NULL; tc++) { + run_case(tc); + count++; + } + } else { + for (int i = 1; i < argc; i++) { + run_case(find_case(module, argv[i])); + count++; + } + } + (void) printf("All %d %s selftest%s passed\n", count, module->sm_name, + count == 1 ? "" : "s"); + libspl_fini(); + return (0); +} diff --git a/cmd/zstream/zstream_selftest.h b/cmd/zstream/zstream_selftest.h new file mode 100644 index 000000000000..287693a718e0 --- /dev/null +++ b/cmd/zstream/zstream_selftest.h @@ -0,0 +1,99 @@ +// 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_SELFTEST_H +#define _ZSTREAM_SELFTEST_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include + +/* + * Shared harness for "zstream selftest". Each module under test supplies a + * NULL-terminated array of named test cases. The harness in + * zstream_selftest.c handles argument parsing, test selection, seeding of + * pseudo-random number generators, watchdog timeouts, and status output. + * + * Test cases report failure by exiting with a nonzero exit code. Any test + * case that returns has passed. + */ + +typedef void test_function_f(void); + +typedef struct { + const char *tc_name; + test_function_f *tc_func; +} test_case_t; + +/* + * Modules with test cases to offer. Each array ends with a NULL tc_name. + */ +extern const test_case_t selftest_queue_cases[]; + +/* + * The seed for this run, set by the harness before any test runs. Printed + * at startup and settable with -s so failures can be replayed. + */ +extern uint64_t selftest_seed; + +/* + * A small deterministic PRNG (splitmix64). Tests derive per-thread + * generators from selftest_seed plus a caller-chosen stream number, so + * workloads are reproducible for a given seed regardless of scheduling. + */ +typedef struct { + uint64_t sr_state; +} selftest_rng_t; + +static inline uint64_t +selftest_mix64(uint64_t z) +{ + z += 0x9e3779b97f4a7c15ULL; + z = (z ^ (z >> 30)) * 0xbf58476d1ce4e5b9ULL; + z = (z ^ (z >> 27)) * 0x94d049bb133111ebULL; + return (z ^ (z >> 31)); +} + +static inline void +selftest_rng_init(selftest_rng_t *rng, uint64_t stream) +{ + rng->sr_state = selftest_seed ^ selftest_mix64(stream); +} + +static inline uint64_t +selftest_rng_next(selftest_rng_t *rng) +{ + rng->sr_state += 0x9e3779b97f4a7c15ULL; + return (selftest_mix64(rng->sr_state)); +} + +/* Uniform value in [0, bound); returns 0 if bound is 0 */ +static inline uint64_t +selftest_rng_below(selftest_rng_t *rng, uint64_t bound) +{ + return (bound ? selftest_rng_next(rng) % bound : 0); +} + +#ifdef __cplusplus +} +#endif + +#endif /* _ZSTREAM_SELFTEST_H */ diff --git a/cmd/zstream/zstream_selftest_queue.c b/cmd/zstream/zstream_selftest_queue.c new file mode 100644 index 000000000000..c03995316ef8 --- /dev/null +++ b/cmd/zstream/zstream_selftest_queue.c @@ -0,0 +1,617 @@ +// 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. + */ + +/* + * Selftests for the zstream_queue multithreaded FIFO queue API. + * + * All tests are built on one generic workload runner. A workload is + * described by a qtest_config_t: some number of producer threads each + * enqueue a stream of self-describing items with randomized costs, + * payloads, and processing delays, while one consumer thread per queue + * dequeues and verifies. Several workloads can run concurrently on separate + * queues to exercise the shared thread pool. + * + * Every item carries enough information to be verified independently: + * + * - The tuple (qi_producer, qi_seq) identifies each item; the consumer + * checks that each producer's items arrive in the same order they + * were enqueued. + * + * - qi_check is a hash of (qi_seed, qi_producer, qi_seq). The processing + * function verifies it and then XORs in TRANSFORM_MAGIC. The consumer + * checks that the transform happened iff cost > 0. + * + * - qi_pattern[] is filled from qi_check and verified both by the process + * function and the consumer, to catch any corruption of the shallow + * copies in and out of the ring buffer. + * + * - qi_process_count counts invocations of the process function, which + * must be exactly one for cost > 0 items and zero for cost == 0 items. + * + * Global conservation checks: the number of items dequeued must equal the + * number enqueued, and the total number of process-function invocations + * must equal the number of nonzero-cost items enqueued. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "zstream_queue.h" +#include "zstream_selftest.h" + +#define TRANSFORM_MAGIC 0xf00dfeedbeefcafeULL + +/* + * Number of times per 1000 processing function invocations to use an + * extra-long "outlier" processing delay to force overtly out-of-order + * completion. + */ +#define LONG_DELAYS_PER_THOUSAND 3 +#define LONG_DELAY_MULTIPLIER 20 + +typedef struct { + uint32_t qi_producer; + uint32_t qi_delay_us; + uint64_t qi_seq; + uint64_t qi_check; + size_t qi_cost; + uint32_t qi_process_count; + uint8_t qi_pattern[]; +} qtest_item_t; + +typedef struct { + uint32_t qc_producers; /* Number of producers */ + uint64_t qc_items; /* Items per producer */ + size_t qc_queue_length; + size_t qc_batch_budget; + size_t qc_pattern_len; /* Extra payload bytes */ + uint32_t qc_zero_cost_pct; /* % of items fast-tracked */ + size_t qc_max_cost; /* Nonzero costs are 1..max */ + uint32_t qc_delay_pct; /* % of items slept on */ + uint32_t qc_max_delay_us; + uint32_t qc_producer_stall_pct; /* % chance producer naps */ + uint32_t qc_consumer_stall_pct; /* % chance consumer naps */ + uint32_t qc_stall_max_us; + uint64_t qc_rng_stream; /* base PRNG stream number */ +} qtest_config_t; + +typedef struct { + const qtest_config_t *qr_cfg; + zstream_queue_t *qr_queue; + uint32_t qr_producers_left; + uint64_t qr_expect_processed; /* Atomic */ + uint64_t qr_processed; /* Atomic */ + uint64_t qr_dequeued; +} qtest_run_t; + +typedef struct { + qtest_run_t *qp_run; + uint32_t qp_id; +} qtest_producer_arg_t; + +static uint64_t +item_check_value(uint32_t producer, uint64_t seq) +{ + return (selftest_mix64(selftest_seed ^ + (((uint64_t)producer << 40) + seq))); +} + +static void +fill_pattern(uint8_t *pattern, size_t len, uint64_t check) +{ + for (size_t i = 0; i < len; i++) + pattern[i] = (uint8_t)(check >> ((i & 7) << 3)) ^ (uint8_t)i; +} + +static void +verify_pattern(const uint8_t *pattern, size_t len, uint64_t check, + const char *who) +{ + for (size_t i = 0; i < len; i++) { + uint8_t expect = + (uint8_t)(check >> ((i & 7) << 3)) ^ (uint8_t)i; + if (pattern[i] != expect) { + errx(1, "%s: payload corrupted at byte %zu " + "(0x%02x != 0x%02x)", who, i, pattern[i], expect); + } + } +} + +static size_t +qtest_cost(void *item_in, void *context) +{ + (void) context; + qtest_item_t *item = item_in; + return (item->qi_cost); +} + +static void +qtest_process(void *item_in, void *context) +{ + qtest_run_t *run = context; + qtest_item_t *item = item_in; + + /* Cost-0 items should never reach the process function */ + VERIFY3U(item->qi_cost, >, 0); + VERIFY3U(item->qi_check, ==, + item_check_value(item->qi_producer, item->qi_seq)); + verify_pattern(item->qi_pattern, run->qr_cfg->qc_pattern_len, + item->qi_check, "process"); + VERIFY3U(atomic_add_32_nv(&item->qi_process_count, 1), ==, 1); + + if (item->qi_delay_us > 0) + (void) usleep(item->qi_delay_us); + + item->qi_check ^= TRANSFORM_MAGIC; + atomic_add_64(&run->qr_processed, 1); +} + +/* + * Pthreads worker function for enqueuers + */ +static void * +qtest_producer(void *arg) +{ + qtest_producer_arg_t *pa = arg; + qtest_run_t *run = pa->qp_run; + const qtest_config_t *cfg = run->qr_cfg; + uint64_t local_expect = 0; + selftest_rng_t rng; + alignas(uint64_t) uint8_t item_buffer[sizeof (qtest_item_t) + + cfg->qc_pattern_len]; + qtest_item_t *item = (qtest_item_t *)item_buffer; + + selftest_rng_init(&rng, cfg->qc_rng_stream + 1000 + pa->qp_id); + + for (uint64_t seq = 0; seq < cfg->qc_items; seq++) { + + qtest_item_t item_xfer = { + .qi_producer = pa->qp_id, + .qi_seq = seq, + .qi_process_count = 0, + .qi_check = item_check_value(pa->qp_id, seq) + }; + *item = item_xfer; + fill_pattern(item->qi_pattern, cfg->qc_pattern_len, + item->qi_check); + + if (selftest_rng_below(&rng, 100) < cfg->qc_zero_cost_pct) { + item->qi_cost = 0; + } else { + item->qi_cost = + 1 + selftest_rng_below(&rng, cfg->qc_max_cost); + local_expect++; + } + + if (item->qi_cost > 0 && cfg->qc_max_delay_us > 0) { + if (selftest_rng_below(&rng, 1000) < + LONG_DELAYS_PER_THOUSAND) { + item->qi_delay_us = cfg->qc_max_delay_us * + LONG_DELAY_MULTIPLIER; + } else if (selftest_rng_below(&rng, 100) < + cfg->qc_delay_pct) { + item->qi_delay_us = selftest_rng_below(&rng, + cfg->qc_max_delay_us); + } + } + + if (cfg->qc_producer_stall_pct > 0 && + selftest_rng_below(&rng, 100) < cfg->qc_producer_stall_pct) + (void) usleep(selftest_rng_below(&rng, + cfg->qc_stall_max_us)); + + zstream_enqueue(run->qr_queue, item); + } + + atomic_add_64(&run->qr_expect_processed, local_expect); + if (atomic_add_32_nv(&run->qr_producers_left, -1) == 0) + zstream_queue_fini(run->qr_queue); + return (NULL); +} + +/* + * Pthreads worker function for dequeuers + */ +static void * +qtest_consumer(void *arg) +{ + qtest_run_t *run = arg; + const qtest_config_t *cfg = run->qr_cfg; + selftest_rng_t rng; + uint64_t expected_seq[cfg->qc_producers]; + alignas(uint64_t) uint8_t item_buffer[sizeof (qtest_item_t) + + cfg->qc_pattern_len]; + qtest_item_t *item = (qtest_item_t *)item_buffer; + + memset(expected_seq, 0, sizeof (expected_seq)); + selftest_rng_init(&rng, cfg->qc_rng_stream + 999); + + while (zstream_dequeue(run->qr_queue, item)) { + VERIFY3U(item->qi_producer, <, cfg->qc_producers); + if (item->qi_seq != expected_seq[item->qi_producer]) { + errx(1, "consumer: FIFO order violated: got " + "producer %u seq %ju, expected seq %ju", + item->qi_producer, (uintmax_t)item->qi_seq, + (uintmax_t)expected_seq[item->qi_producer]); + } + expected_seq[item->qi_producer]++; + + uint64_t check = + item_check_value(item->qi_producer, item->qi_seq); + if (item->qi_cost > 0) { + VERIFY3U(item->qi_process_count, ==, 1); + VERIFY3U(item->qi_check, ==, check ^ TRANSFORM_MAGIC); + } else { + VERIFY3U(item->qi_process_count, ==, 0); + VERIFY3U(item->qi_check, ==, check); + } + verify_pattern(item->qi_pattern, cfg->qc_pattern_len, check, + "consumer"); + run->qr_dequeued++; + + if (cfg->qc_consumer_stall_pct > 0 && + selftest_rng_below(&rng, 100) < cfg->qc_consumer_stall_pct) + (void) usleep(selftest_rng_below(&rng, + cfg->qc_stall_max_us)); + } + + for (uint32_t p = 0; p < cfg->qc_producers; p++) + VERIFY3U(expected_seq[p], ==, cfg->qc_items); + VERIFY3U(run->qr_dequeued, ==, + (uint64_t)cfg->qc_producers * cfg->qc_items); + + return (NULL); +} + +/* + * Run several workloads at once, one queue per config, with a dedicated + * consumer thread and qc_producers producer threads per queue. Returns + * after every queue has been drained to end-of-stream (and therefore + * destroyed) and all verification checks have passed. + */ +static void +run_queue_workloads(const qtest_config_t *cfgs, int ncfg) +{ + qtest_run_t runs[ncfg]; + pthread_t consumers[ncfg]; + uint32_t total_producers = 0; + + for (int i = 0; i < ncfg; i++) + total_producers += cfgs[i].qc_producers; + + pthread_t producers[total_producers]; + qtest_producer_arg_t pargs[total_producers]; + memset(runs, 0, sizeof (runs)); + memset(pargs, 0, sizeof (pargs)); + + for (int i = 0; i < ncfg; i++) { + runs[i].qr_cfg = &cfgs[i]; + runs[i].qr_producers_left = cfgs[i].qc_producers; + zq_params_t params = { + .qp_process = qtest_process, + .qp_cost = qtest_cost, + .qp_context = &runs[i], + .qp_item_size = + sizeof (qtest_item_t) + cfgs[i].qc_pattern_len, + .qp_batch_budget = cfgs[i].qc_batch_budget, + .qp_queue_length = cfgs[i].qc_queue_length, + }; + runs[i].qr_queue = zstream_queue_create(¶ms); + } + + int p = 0; + for (int i = 0; i < ncfg; i++) { + VERIFY3S(pthread_create(&consumers[i], NULL, qtest_consumer, + &runs[i]), ==, 0); + for (uint32_t j = 0; j < cfgs[i].qc_producers; j++, p++) { + pargs[p].qp_run = &runs[i]; + pargs[p].qp_id = j; + VERIFY3S(pthread_create(&producers[p], NULL, + qtest_producer, &pargs[p]), ==, 0); + } + } + + for (uint32_t i = 0; i < total_producers; i++) + VERIFY3S(pthread_join(producers[i], NULL), ==, 0); + for (int i = 0; i < ncfg; i++) + VERIFY3S(pthread_join(consumers[i], NULL), ==, 0); + + for (int i = 0; i < ncfg; i++) + VERIFY3U(runs[i].qr_processed, ==, runs[i].qr_expect_processed); +} + +static void +run_queue_workload(const qtest_config_t *cfg) +{ + run_queue_workloads(cfg, 1); +} + +/* + * Basic single-producer smoke test: deterministic-ish costs, no delays. + */ +static void +queue_basic(void) +{ + qtest_config_t cfg = { + .qc_producers = 1, + .qc_items = 5000, + .qc_queue_length = 64, + .qc_batch_budget = 256, + .qc_pattern_len = 32, + .qc_zero_cost_pct = 20, + .qc_max_cost = 64, + }; + run_queue_workload(&cfg); +} + +/* + * A long, randomized stream with heavy-tailed processing delays, a large + * fraction of fast-tracked items, costs that exceed the batch budget, and a + * consumer that periodically stalls so the queue backs up and enqueue + * blocks on Q_FULL. The ring indices wrap hundreds of times. + */ +static void +queue_torture(void) +{ + qtest_config_t cfg = { + .qc_producers = 1, + .qc_items = 100000, + .qc_queue_length = 512, + .qc_batch_budget = 2048, + .qc_pattern_len = 64, + .qc_zero_cost_pct = 30, + .qc_max_cost = 4096, + .qc_delay_pct = 5, + .qc_max_delay_us = 100, + .qc_consumer_stall_pct = 1, + .qc_stall_max_us = 500, + .qc_rng_stream = 100, + }; + run_queue_workload(&cfg); +} + +/* + * Off-by-one hunting: sweep the degenerate corners of queue length, + * batch budget, and stream length, including a zero-item stream and + * zero-length payloads. + */ +static void +queue_edge_cases(void) +{ + static const size_t lengths[] = + { 1, 2, MAX_BATCH - 1, MAX_BATCH, MAX_BATCH + 1, 64 }; + static const size_t budgets[] = { 0, 1, 16, SIZE_MAX / 2 }; + uint64_t stream = 200; + + for (int l = 0; l < 6; l++) { + for (int b = 0; b < 4; b++) { + uint64_t counts[] = { 0, lengths[l], lengths[l] + 1, + 4 * lengths[l] + 3 }; + for (int n = 0; n < 4; n++) { + qtest_config_t cfg = { + .qc_producers = 1, + .qc_items = counts[n], + .qc_queue_length = lengths[l], + .qc_batch_budget = budgets[b], + .qc_pattern_len = + (lengths[l] & 1) ? 0 : 24, + .qc_zero_cost_pct = 25, + .qc_max_cost = 8, + .qc_rng_stream = stream++, + }; + run_queue_workload(&cfg); + } + } + } +} + +/* + * All items cost 0, so every item takes the fast track and the process + * function must never run (qtest_process VERIFYs cost > 0, and the + * conservation check at the end of the run confirms zero invocations). + * This exercises the completion-index sweep for items no worker ever + * touches. + */ +static void +queue_zero_cost(void) +{ + qtest_config_t cfg = { + .qc_producers = 1, + .qc_items = 20000, + .qc_queue_length = 128, + .qc_batch_budget = 1024, + .qc_pattern_len = 16, + .qc_zero_cost_pct = 100, + .qc_max_cost = 8, + .qc_rng_stream = 300, + }; + run_queue_workload(&cfg); +} + +/* + * Eight producer threads hammering one queue with random pacing. The + * consumer verifies per-producer FIFO order and exact counts. + */ +static void +queue_multi_producer(void) +{ + qtest_config_t cfg = { + .qc_producers = 8, + .qc_items = 15000, + .qc_queue_length = 256, + .qc_batch_budget = 512, + .qc_pattern_len = 24, + .qc_zero_cost_pct = 25, + .qc_max_cost = 512, + .qc_delay_pct = 2, + .qc_max_delay_us = 50, + .qc_producer_stall_pct = 1, + .qc_stall_max_us = 200, + .qc_rng_stream = 400, + }; + run_queue_workload(&cfg); +} + +/* + * Many dissimilar queues live at once, stressing worker scoring and + * assignment, per-queue index isolation, and destruction of queues while + * others remain active (which compacts the pool's queue array). + */ +static void +queue_multi_queue(void) +{ + qtest_config_t cfgs[12]; + for (int i = 0; i < 12; i++) { + uint32_t producers = 1 + i % 3; + qtest_config_t cfg = { + .qc_producers = producers, + .qc_items = 4000 / producers, + .qc_queue_length = (size_t)4 << (i % 6), + .qc_batch_budget = + (i % 4 == 0) ? 0 : (size_t)64 << (i % 5), + .qc_pattern_len = 8 * (i % 5), + .qc_zero_cost_pct = 10 * (i % 6), + .qc_max_cost = (size_t)16 << (i % 8), + .qc_delay_pct = i % 3, + .qc_max_delay_us = 60, + .qc_rng_stream = 500 + i * 10000, + }; + cfgs[i] = cfg; + } + run_queue_workloads(cfgs, 12); +} + +static void * +run_queue_workload_thread(void *arg) +{ + run_queue_workload(arg); + return (NULL); +} + +/* + * Spin the thread pool up and down repeatedly. After every drain the + * process must be back to its baseline thread count. The second phase + * races a fresh zstream_queue_create() against the previous queue's + * spin-down to exercise the create-vs-spindown locking. + */ +static void +queue_cycles(void) +{ + qtest_config_t cfg = { + .qc_producers = 1, + .qc_items = 400, + .qc_queue_length = 32, + .qc_batch_budget = 64, + .qc_pattern_len = 16, + .qc_zero_cost_pct = 20, + .qc_max_cost = 32, + .qc_rng_stream = 600, + }; + + for (int iter = 0; iter < 30; iter++) { + cfg.qc_producers = 1 + iter % 2; + cfg.qc_rng_stream = 600 + iter; + run_queue_workload(&cfg); + } + + selftest_rng_t rng; + selftest_rng_init(&rng, 650); + for (int iter = 0; iter < 10; iter++) { + qtest_config_t racer = cfg; + racer.qc_producers = 1; + racer.qc_rng_stream = 700 + iter; + pthread_t bg; + VERIFY3S(pthread_create(&bg, NULL, run_queue_workload_thread, + &racer), ==, 0); + (void) usleep(selftest_rng_below(&rng, 2000)); + cfg.qc_rng_stream = 800 + iter; + run_queue_workload(&cfg); + VERIFY3S(pthread_join(bg, NULL), ==, 0); + } +} + +/* + * Seeded chaos: randomize every workload parameter within sane bounds + * and run a few rounds of concurrent queues. Whatever the targeted tests + * miss, this net catches over many CI runs; failures replay with -s. + */ +static void +queue_stress(void) +{ + selftest_rng_t rng; + selftest_rng_init(&rng, 900); + + for (int iter = 0; iter < 8; iter++) { + int nqueues = 1 + selftest_rng_below(&rng, 4); + qtest_config_t cfgs[4]; + + for (int i = 0; i < nqueues; i++) { + uint32_t producers = 1 + selftest_rng_below(&rng, 4); + qtest_config_t cfg = { + .qc_producers = producers, + .qc_items = (2000 + + selftest_rng_below(&rng, 8000)) / + producers, + .qc_queue_length = (size_t)1 << + selftest_rng_below(&rng, 10), + .qc_batch_budget = + (selftest_rng_below(&rng, 3) == 0) ? 0 : + selftest_rng_below(&rng, 4096), + .qc_pattern_len = + selftest_rng_below(&rng, 64), + .qc_zero_cost_pct = + selftest_rng_below(&rng, 101), + .qc_max_cost = + 1 + selftest_rng_below(&rng, 2048), + .qc_delay_pct = selftest_rng_below(&rng, 4), + .qc_max_delay_us = + selftest_rng_below(&rng, 120), + .qc_producer_stall_pct = + selftest_rng_below(&rng, 2), + .qc_consumer_stall_pct = + selftest_rng_below(&rng, 2), + .qc_stall_max_us = + selftest_rng_below(&rng, 400), + .qc_rng_stream = 1000000 + iter * 1000 + + i * 100, + }; + cfgs[i] = cfg; + } + run_queue_workloads(cfgs, nqueues); + } +} + +const test_case_t selftest_queue_cases[] = { + { "queue_basic", queue_basic }, + { "queue_edge_cases", queue_edge_cases }, + { "queue_zero_cost", queue_zero_cost }, + { "queue_torture", queue_torture }, + { "queue_multi_producer", queue_multi_producer }, + { "queue_multi_queue", queue_multi_queue }, + { "queue_cycles", queue_cycles }, + { "queue_stress", queue_stress }, + { NULL, NULL }, +}; diff --git a/cmd/zstream/zstream_util.c b/cmd/zstream/zstream_util.c index 5660c67cc015..6acdc557917a 100644 --- a/cmd/zstream/zstream_util.c +++ b/cmd/zstream/zstream_util.c @@ -29,8 +29,10 @@ */ #include +#include #include #include +#include #include #include #include @@ -67,6 +69,30 @@ safe_calloc(size_t size) return (rv); } +void +safe_pthread_sigmask(int how, const sigset_t *set, sigset_t *oldset) +{ + int error = pthread_sigmask(how, set, oldset); + if (error != 0) { + errno = error; + err(1, "pthread_sigmask failed"); + } +} + +pthread_t +safe_create_thread(thread_f *body, void *body_arg, const char *name, + boolean_t detach) +{ + pthread_t tid; + if (pthread_create(&tid, NULL, body, body_arg) != 0) + err(1, "pthread_create for %s failed", name); + if (pthread_setname_np(tid, name) != 0) + err(1, "could not set name of %s thread", name); + if (detach && pthread_detach(tid) != 0) + err(1, "failed to detach %s thread", name); + return (tid); +} + char * checksum_str(zio_cksum_t *cksum, char *buff, size_t buff_size) { diff --git a/cmd/zstream/zstream_util.h b/cmd/zstream/zstream_util.h index 8ae470a08e27..e3cde1de80b6 100644 --- a/cmd/zstream/zstream_util.h +++ b/cmd/zstream/zstream_util.h @@ -25,8 +25,11 @@ extern "C" { #endif +#include +#include #include #include +#include #include #include #include @@ -36,17 +39,27 @@ typedef struct { int cs_level; } compression_spec_t; +typedef void * +thread_f(void *); + /* * The safe_ versions of the functions below terminate the process if the * operation doesn't succeed instead of returning an error. */ -extern void * +void * safe_malloc(size_t size); -extern void * +void * safe_calloc(size_t n); -extern char * +void +safe_pthread_sigmask(int how, const sigset_t *set, sigset_t *oldset); + +pthread_t +safe_create_thread(thread_f *body, void *body_arg, const char *name, + boolean_t detach); + +char * checksum_str(zio_cksum_t *cksum, char *buff, size_t buff_size); /* diff --git a/tests/runfiles/common.run b/tests/runfiles/common.run index 51559f185606..39100599e231 100644 --- a/tests/runfiles/common.run +++ b/tests/runfiles/common.run @@ -1170,6 +1170,7 @@ tests = ['zstream_checksum_001_pos', 'zstream_recompress_003_pos', 'zstream_recompress_004_pos', 'zstream_recompress_005_pos', 'zstream_redup_001_pos', + 'zstream_selftest_queue_001_pos', 'zstream_validate_001_neg'] tags = ['functional', 'zstream'] diff --git a/tests/zfs-tests/tests/Makefile.am b/tests/zfs-tests/tests/Makefile.am index e283021de98c..11012aa5696f 100644 --- a/tests/zfs-tests/tests/Makefile.am +++ b/tests/zfs-tests/tests/Makefile.am @@ -2462,6 +2462,7 @@ nobase_dist_datadir_zfs_tests_tests_SCRIPTS += \ functional/zstream/zstream_recompress_005_pos.ksh \ functional/zstream/zstream_redup_001_pos.ksh \ functional/zstream/zstream_validate_001_neg.ksh \ + functional/zstream/zstream_selftest_queue_001_pos.ksh \ functional/zvol/zvol_cli/cleanup.ksh \ functional/zvol/zvol_cli/setup.ksh \ functional/zvol/zvol_cli/zvol_cli_001_pos.ksh \ diff --git a/tests/zfs-tests/tests/functional/zstream/zstream_selftest_queue_001_pos.ksh b/tests/zfs-tests/tests/functional/zstream/zstream_selftest_queue_001_pos.ksh new file mode 100755 index 000000000000..fe7366ff93e8 --- /dev/null +++ b/tests/zfs-tests/tests/functional/zstream/zstream_selftest_queue_001_pos.ksh @@ -0,0 +1,37 @@ +#!/bin/ksh -p +# SPDX-License-Identifier: CDDL-1.0 + +# +# 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. +# + +# +# Copyright (c) 2026 by Garth Snyder. All rights reserved. +# + +. $STF_SUITE/include/libtest.shlib + +# +# Description: +# Run zstream's built-in unit tests ("zstream selftest"). +# +# Strategy: +# 1. Run all selftests for each module with the default worker thread pool. +# 2. Run them again with a single worker thread. +# + +verify_runnable "both" + +log_assert "zstream self-tests for zstream_queue all pass" + +log_must zstream selftest queue +log_must zstream selftest -t 1 queue + +log_pass "zstream self-tests for zstream_queue all pass"