1 // SPDX-License-Identifier: CDDL-1.0
2 /*
3 * This file and its contents are supplied under the terms of the
4 * Common Development and Distribution License ("CDDL"), version 1.0.
5 * You may only use this file in accordance with the terms of version
6 * 1.0 of the CDDL.
7 *
8 * A full copy of the text of the CDDL should have accompanied this
9 * source. A copy of the CDDL is also available via the Internet at
10 * https://opensource.org/license/CDDL-1.0.
11 */
12
13 /*
14 * Copyright (c) 2026 by Garth Snyder. All rights reserved.
15 */
16
17 /*
18 * Selftests for the zstream_queue multithreaded FIFO queue API.
19 *
20 * All tests are built on one generic workload runner. A workload is
21 * described by a qtest_config_t: some number of producer threads each
22 * enqueue a stream of self-describing items with randomized costs,
23 * payloads, and processing delays, while one consumer thread per queue
24 * dequeues and verifies. Several workloads can run concurrently on separate
25 * queues to exercise the shared thread pool.
26 *
27 * Every item carries enough information to be verified independently:
28 *
29 * - The tuple (qi_producer, qi_seq) identifies each item; the consumer
30 * checks that each producer's items arrive in the same order they
31 * were enqueued.
32 *
33 * - qi_check is a hash of (qi_seed, qi_producer, qi_seq). The processing
34 * function verifies it and then XORs in TRANSFORM_MAGIC. The consumer
35 * checks that the transform happened iff cost > 0.
36 *
37 * - qi_pattern[] is filled from qi_check and verified both by the process
38 * function and the consumer, to catch any corruption of the shallow
39 * copies in and out of the ring buffer.
40 *
41 * - qi_process_count counts invocations of the process function, which
42 * must be exactly one for cost > 0 items and zero for cost == 0 items.
43 *
44 * Global conservation checks: the number of items dequeued must equal the
45 * number enqueued, and the total number of process-function invocations
46 * must equal the number of nonzero-cost items enqueued.
47 */
48
49 #include <assert.h>
50 #include <atomic.h>
51 #include <err.h>
52 #include <pthread.h>
53 #include <stdalign.h>
54 #include <stdint.h>
55 #include <stdio.h>
56 #include <string.h>
57 #include <unistd.h>
58
59 #include "zstream_queue.h"
60 #include "zstream_selftest.h"
61
62 #define TRANSFORM_MAGIC 0xf00dfeedbeefcafeULL
63
64 /*
65 * Number of times per 1000 processing function invocations to use an
66 * extra-long "outlier" processing delay to force overtly out-of-order
67 * completion.
68 */
69 #define LONG_DELAYS_PER_THOUSAND 3
70 #define LONG_DELAY_MULTIPLIER 20
71
72 typedef struct {
73 uint32_t qi_producer;
74 uint32_t qi_delay_us;
75 uint64_t qi_seq;
76 uint64_t qi_check;
77 size_t qi_cost;
78 uint32_t qi_process_count;
79 uint8_t qi_pattern[];
80 } qtest_item_t;
81
82 typedef struct {
83 uint32_t qc_producers; /* Number of producers */
84 uint64_t qc_items; /* Items per producer */
85 size_t qc_queue_length;
86 size_t qc_batch_budget;
87 size_t qc_pattern_len; /* Extra payload bytes */
88 uint32_t qc_zero_cost_pct; /* % of items fast-tracked */
89 size_t qc_max_cost; /* Nonzero costs are 1..max */
90 uint32_t qc_delay_pct; /* % of items slept on */
91 uint32_t qc_max_delay_us;
92 uint32_t qc_producer_stall_pct; /* % chance producer naps */
93 uint32_t qc_consumer_stall_pct; /* % chance consumer naps */
94 uint32_t qc_stall_max_us;
95 uint64_t qc_rng_stream; /* base PRNG stream number */
96 } qtest_config_t;
97
98 typedef struct {
99 const qtest_config_t *qr_cfg;
100 zstream_queue_t *qr_queue;
101 uint32_t qr_producers_left;
102 uint64_t qr_expect_processed; /* Atomic */
103 uint64_t qr_processed; /* Atomic */
104 uint64_t qr_dequeued;
105 } qtest_run_t;
106
107 typedef struct {
108 qtest_run_t *qp_run;
109 uint32_t qp_id;
110 } qtest_producer_arg_t;
111
112 static uint64_t
item_check_value(uint32_t producer,uint64_t seq)113 item_check_value(uint32_t producer, uint64_t seq)
114 {
115 return (selftest_mix64(selftest_seed ^
116 (((uint64_t)producer << 40) + seq)));
117 }
118
119 static void
fill_pattern(uint8_t * pattern,size_t len,uint64_t check)120 fill_pattern(uint8_t *pattern, size_t len, uint64_t check)
121 {
122 for (size_t i = 0; i < len; i++)
123 pattern[i] = (uint8_t)(check >> ((i & 7) << 3)) ^ (uint8_t)i;
124 }
125
126 static void
verify_pattern(const uint8_t * pattern,size_t len,uint64_t check,const char * who)127 verify_pattern(const uint8_t *pattern, size_t len, uint64_t check,
128 const char *who)
129 {
130 for (size_t i = 0; i < len; i++) {
131 uint8_t expect =
132 (uint8_t)(check >> ((i & 7) << 3)) ^ (uint8_t)i;
133 if (pattern[i] != expect) {
134 errx(1, "%s: payload corrupted at byte %zu "
135 "(0x%02x != 0x%02x)", who, i, pattern[i], expect);
136 }
137 }
138 }
139
140 static size_t
qtest_cost(void * item_in,void * context)141 qtest_cost(void *item_in, void *context)
142 {
143 (void) context;
144 qtest_item_t *item = item_in;
145 return (item->qi_cost);
146 }
147
148 static void
qtest_process(void * item_in,void * context)149 qtest_process(void *item_in, void *context)
150 {
151 qtest_run_t *run = context;
152 qtest_item_t *item = item_in;
153
154 /* Cost-0 items should never reach the process function */
155 VERIFY3U(item->qi_cost, >, 0);
156 VERIFY3U(item->qi_check, ==,
157 item_check_value(item->qi_producer, item->qi_seq));
158 verify_pattern(item->qi_pattern, run->qr_cfg->qc_pattern_len,
159 item->qi_check, "process");
160 VERIFY3U(atomic_add_32_nv(&item->qi_process_count, 1), ==, 1);
161
162 if (item->qi_delay_us > 0)
163 (void) usleep(item->qi_delay_us);
164
165 item->qi_check ^= TRANSFORM_MAGIC;
166 atomic_add_64(&run->qr_processed, 1);
167 }
168
169 /*
170 * Pthreads worker function for enqueuers
171 */
172 static void *
qtest_producer(void * arg)173 qtest_producer(void *arg)
174 {
175 qtest_producer_arg_t *pa = arg;
176 qtest_run_t *run = pa->qp_run;
177 const qtest_config_t *cfg = run->qr_cfg;
178 uint64_t local_expect = 0;
179 selftest_rng_t rng;
180 alignas(uint64_t) uint8_t item_buffer[sizeof (qtest_item_t) +
181 cfg->qc_pattern_len];
182 qtest_item_t *item = (qtest_item_t *)item_buffer;
183
184 selftest_rng_init(&rng, cfg->qc_rng_stream + 1000 + pa->qp_id);
185
186 for (uint64_t seq = 0; seq < cfg->qc_items; seq++) {
187
188 qtest_item_t item_xfer = {
189 .qi_producer = pa->qp_id,
190 .qi_seq = seq,
191 .qi_process_count = 0,
192 .qi_check = item_check_value(pa->qp_id, seq)
193 };
194 *item = item_xfer;
195 fill_pattern(item->qi_pattern, cfg->qc_pattern_len,
196 item->qi_check);
197
198 if (selftest_rng_below(&rng, 100) < cfg->qc_zero_cost_pct) {
199 item->qi_cost = 0;
200 } else {
201 item->qi_cost =
202 1 + selftest_rng_below(&rng, cfg->qc_max_cost);
203 local_expect++;
204 }
205
206 if (item->qi_cost > 0 && cfg->qc_max_delay_us > 0) {
207 if (selftest_rng_below(&rng, 1000) <
208 LONG_DELAYS_PER_THOUSAND) {
209 item->qi_delay_us = cfg->qc_max_delay_us *
210 LONG_DELAY_MULTIPLIER;
211 } else if (selftest_rng_below(&rng, 100) <
212 cfg->qc_delay_pct) {
213 item->qi_delay_us = selftest_rng_below(&rng,
214 cfg->qc_max_delay_us);
215 }
216 }
217
218 if (cfg->qc_producer_stall_pct > 0 &&
219 selftest_rng_below(&rng, 100) < cfg->qc_producer_stall_pct)
220 (void) usleep(selftest_rng_below(&rng,
221 cfg->qc_stall_max_us));
222
223 zstream_enqueue(run->qr_queue, item);
224 }
225
226 atomic_add_64(&run->qr_expect_processed, local_expect);
227 if (atomic_add_32_nv(&run->qr_producers_left, -1) == 0)
228 zstream_queue_fini(run->qr_queue);
229 return (NULL);
230 }
231
232 /*
233 * Pthreads worker function for dequeuers
234 */
235 static void *
qtest_consumer(void * arg)236 qtest_consumer(void *arg)
237 {
238 qtest_run_t *run = arg;
239 const qtest_config_t *cfg = run->qr_cfg;
240 selftest_rng_t rng;
241 uint64_t expected_seq[cfg->qc_producers];
242 alignas(uint64_t) uint8_t item_buffer[sizeof (qtest_item_t) +
243 cfg->qc_pattern_len];
244 qtest_item_t *item = (qtest_item_t *)item_buffer;
245
246 memset(expected_seq, 0, sizeof (expected_seq));
247 selftest_rng_init(&rng, cfg->qc_rng_stream + 999);
248
249 while (zstream_dequeue(run->qr_queue, item)) {
250 VERIFY3U(item->qi_producer, <, cfg->qc_producers);
251 if (item->qi_seq != expected_seq[item->qi_producer]) {
252 errx(1, "consumer: FIFO order violated: got "
253 "producer %u seq %ju, expected seq %ju",
254 item->qi_producer, (uintmax_t)item->qi_seq,
255 (uintmax_t)expected_seq[item->qi_producer]);
256 }
257 expected_seq[item->qi_producer]++;
258
259 uint64_t check =
260 item_check_value(item->qi_producer, item->qi_seq);
261 if (item->qi_cost > 0) {
262 VERIFY3U(item->qi_process_count, ==, 1);
263 VERIFY3U(item->qi_check, ==, check ^ TRANSFORM_MAGIC);
264 } else {
265 VERIFY3U(item->qi_process_count, ==, 0);
266 VERIFY3U(item->qi_check, ==, check);
267 }
268 verify_pattern(item->qi_pattern, cfg->qc_pattern_len, check,
269 "consumer");
270 run->qr_dequeued++;
271
272 if (cfg->qc_consumer_stall_pct > 0 &&
273 selftest_rng_below(&rng, 100) < cfg->qc_consumer_stall_pct)
274 (void) usleep(selftest_rng_below(&rng,
275 cfg->qc_stall_max_us));
276 }
277
278 for (uint32_t p = 0; p < cfg->qc_producers; p++)
279 VERIFY3U(expected_seq[p], ==, cfg->qc_items);
280 VERIFY3U(run->qr_dequeued, ==,
281 (uint64_t)cfg->qc_producers * cfg->qc_items);
282
283 return (NULL);
284 }
285
286 /*
287 * Run several workloads at once, one queue per config, with a dedicated
288 * consumer thread and qc_producers producer threads per queue. Returns
289 * after every queue has been drained to end-of-stream (and therefore
290 * destroyed) and all verification checks have passed.
291 */
292 static void
run_queue_workloads(const qtest_config_t * cfgs,int ncfg)293 run_queue_workloads(const qtest_config_t *cfgs, int ncfg)
294 {
295 qtest_run_t runs[ncfg];
296 pthread_t consumers[ncfg];
297 uint32_t total_producers = 0;
298
299 for (int i = 0; i < ncfg; i++)
300 total_producers += cfgs[i].qc_producers;
301
302 pthread_t producers[total_producers];
303 qtest_producer_arg_t pargs[total_producers];
304 memset(runs, 0, sizeof (runs));
305 memset(pargs, 0, sizeof (pargs));
306
307 for (int i = 0; i < ncfg; i++) {
308 runs[i].qr_cfg = &cfgs[i];
309 runs[i].qr_producers_left = cfgs[i].qc_producers;
310 zq_params_t params = {
311 .qp_process = qtest_process,
312 .qp_cost = qtest_cost,
313 .qp_context = &runs[i],
314 .qp_item_size =
315 sizeof (qtest_item_t) + cfgs[i].qc_pattern_len,
316 .qp_batch_budget = cfgs[i].qc_batch_budget,
317 .qp_queue_length = cfgs[i].qc_queue_length,
318 };
319 runs[i].qr_queue = zstream_queue_create(¶ms);
320 }
321
322 int p = 0;
323 for (int i = 0; i < ncfg; i++) {
324 VERIFY3S(pthread_create(&consumers[i], NULL, qtest_consumer,
325 &runs[i]), ==, 0);
326 for (uint32_t j = 0; j < cfgs[i].qc_producers; j++, p++) {
327 pargs[p].qp_run = &runs[i];
328 pargs[p].qp_id = j;
329 VERIFY3S(pthread_create(&producers[p], NULL,
330 qtest_producer, &pargs[p]), ==, 0);
331 }
332 }
333
334 for (uint32_t i = 0; i < total_producers; i++)
335 VERIFY3S(pthread_join(producers[i], NULL), ==, 0);
336 for (int i = 0; i < ncfg; i++)
337 VERIFY3S(pthread_join(consumers[i], NULL), ==, 0);
338
339 for (int i = 0; i < ncfg; i++)
340 VERIFY3U(runs[i].qr_processed, ==, runs[i].qr_expect_processed);
341 }
342
343 static void
run_queue_workload(const qtest_config_t * cfg)344 run_queue_workload(const qtest_config_t *cfg)
345 {
346 run_queue_workloads(cfg, 1);
347 }
348
349 /*
350 * Basic single-producer smoke test: deterministic-ish costs, no delays.
351 */
352 static void
queue_basic(void)353 queue_basic(void)
354 {
355 qtest_config_t cfg = {
356 .qc_producers = 1,
357 .qc_items = 5000,
358 .qc_queue_length = 64,
359 .qc_batch_budget = 256,
360 .qc_pattern_len = 32,
361 .qc_zero_cost_pct = 20,
362 .qc_max_cost = 64,
363 };
364 run_queue_workload(&cfg);
365 }
366
367 /*
368 * A long, randomized stream with heavy-tailed processing delays, a large
369 * fraction of fast-tracked items, costs that exceed the batch budget, and a
370 * consumer that periodically stalls so the queue backs up and enqueue
371 * blocks on Q_FULL. The ring indices wrap hundreds of times.
372 */
373 static void
queue_torture(void)374 queue_torture(void)
375 {
376 qtest_config_t cfg = {
377 .qc_producers = 1,
378 .qc_items = 100000,
379 .qc_queue_length = 512,
380 .qc_batch_budget = 2048,
381 .qc_pattern_len = 64,
382 .qc_zero_cost_pct = 30,
383 .qc_max_cost = 4096,
384 .qc_delay_pct = 5,
385 .qc_max_delay_us = 100,
386 .qc_consumer_stall_pct = 1,
387 .qc_stall_max_us = 500,
388 .qc_rng_stream = 100,
389 };
390 run_queue_workload(&cfg);
391 }
392
393 /*
394 * Off-by-one hunting: sweep the degenerate corners of queue length,
395 * batch budget, and stream length, including a zero-item stream and
396 * zero-length payloads.
397 */
398 static void
queue_edge_cases(void)399 queue_edge_cases(void)
400 {
401 static const size_t lengths[] =
402 { 1, 2, ZQ_MAX_BATCH - 1, ZQ_MAX_BATCH, ZQ_MAX_BATCH + 1, 64 };
403 static const size_t budgets[] = { 0, 1, 16, SIZE_MAX / 2 };
404 uint64_t stream = 200;
405
406 for (int l = 0; l < 6; l++) {
407 for (int b = 0; b < 4; b++) {
408 uint64_t counts[] = { 0, lengths[l], lengths[l] + 1,
409 4 * lengths[l] + 3 };
410 for (int n = 0; n < 4; n++) {
411 qtest_config_t cfg = {
412 .qc_producers = 1,
413 .qc_items = counts[n],
414 .qc_queue_length = lengths[l],
415 .qc_batch_budget = budgets[b],
416 .qc_pattern_len =
417 (lengths[l] & 1) ? 0 : 24,
418 .qc_zero_cost_pct = 25,
419 .qc_max_cost = 8,
420 .qc_rng_stream = stream++,
421 };
422 run_queue_workload(&cfg);
423 }
424 }
425 }
426 }
427
428 /*
429 * All items cost 0, so every item takes the fast track and the process
430 * function must never run (qtest_process VERIFYs cost > 0, and the
431 * conservation check at the end of the run confirms zero invocations).
432 * This exercises the completion-index sweep for items no worker ever
433 * touches.
434 */
435 static void
queue_zero_cost(void)436 queue_zero_cost(void)
437 {
438 qtest_config_t cfg = {
439 .qc_producers = 1,
440 .qc_items = 20000,
441 .qc_queue_length = 128,
442 .qc_batch_budget = 1024,
443 .qc_pattern_len = 16,
444 .qc_zero_cost_pct = 100,
445 .qc_max_cost = 8,
446 .qc_rng_stream = 300,
447 };
448 run_queue_workload(&cfg);
449 }
450
451 /*
452 * Eight producer threads hammering one queue with random pacing. The
453 * consumer verifies per-producer FIFO order and exact counts.
454 */
455 static void
queue_multi_producer(void)456 queue_multi_producer(void)
457 {
458 qtest_config_t cfg = {
459 .qc_producers = 8,
460 .qc_items = 15000,
461 .qc_queue_length = 256,
462 .qc_batch_budget = 512,
463 .qc_pattern_len = 24,
464 .qc_zero_cost_pct = 25,
465 .qc_max_cost = 512,
466 .qc_delay_pct = 2,
467 .qc_max_delay_us = 50,
468 .qc_producer_stall_pct = 1,
469 .qc_stall_max_us = 200,
470 .qc_rng_stream = 400,
471 };
472 run_queue_workload(&cfg);
473 }
474
475 /*
476 * Many dissimilar queues live at once, stressing worker scoring and
477 * assignment, per-queue index isolation, and destruction of queues while
478 * others remain active (which compacts the pool's queue array).
479 */
480 static void
queue_multi_queue(void)481 queue_multi_queue(void)
482 {
483 qtest_config_t cfgs[12];
484 for (int i = 0; i < 12; i++) {
485 uint32_t producers = 1 + i % 3;
486 qtest_config_t cfg = {
487 .qc_producers = producers,
488 .qc_items = 4000 / producers,
489 .qc_queue_length = (size_t)4 << (i % 6),
490 .qc_batch_budget =
491 (i % 4 == 0) ? 0 : (size_t)64 << (i % 5),
492 .qc_pattern_len = 8 * (i % 5),
493 .qc_zero_cost_pct = 10 * (i % 6),
494 .qc_max_cost = (size_t)16 << (i % 8),
495 .qc_delay_pct = i % 3,
496 .qc_max_delay_us = 60,
497 .qc_rng_stream = 500 + i * 10000,
498 };
499 cfgs[i] = cfg;
500 }
501 run_queue_workloads(cfgs, 12);
502 }
503
504 /*
505 * Seeded chaos: randomize every workload parameter within sane bounds
506 * and run a few rounds of concurrent queues. Whatever the targeted tests
507 * miss, this net catches over many CI runs; failures replay with -s.
508 */
509 static void
queue_stress(void)510 queue_stress(void)
511 {
512 selftest_rng_t rng;
513 selftest_rng_init(&rng, 900);
514
515 for (int iter = 0; iter < 8; iter++) {
516 int nqueues = 1 + selftest_rng_below(&rng, 4);
517 qtest_config_t cfgs[4];
518
519 for (int i = 0; i < nqueues; i++) {
520 uint32_t producers = 1 + selftest_rng_below(&rng, 4);
521 qtest_config_t cfg = {
522 .qc_producers = producers,
523 .qc_items = (2000 +
524 selftest_rng_below(&rng, 8000)) /
525 producers,
526 .qc_queue_length = (size_t)1 <<
527 selftest_rng_below(&rng, 10),
528 .qc_batch_budget =
529 (selftest_rng_below(&rng, 3) == 0) ? 0 :
530 selftest_rng_below(&rng, 4096),
531 .qc_pattern_len =
532 selftest_rng_below(&rng, 64),
533 .qc_zero_cost_pct =
534 selftest_rng_below(&rng, 101),
535 .qc_max_cost =
536 1 + selftest_rng_below(&rng, 2048),
537 .qc_delay_pct = selftest_rng_below(&rng, 4),
538 .qc_max_delay_us =
539 selftest_rng_below(&rng, 120),
540 .qc_producer_stall_pct =
541 selftest_rng_below(&rng, 2),
542 .qc_consumer_stall_pct =
543 selftest_rng_below(&rng, 2),
544 .qc_stall_max_us =
545 selftest_rng_below(&rng, 400),
546 .qc_rng_stream = 1000000 + iter * 1000 +
547 i * 100,
548 };
549 cfgs[i] = cfg;
550 }
551 run_queue_workloads(cfgs, nqueues);
552 }
553 }
554
555 const test_case_t selftest_queue_cases[] = {
556 { "queue_basic", queue_basic },
557 { "queue_edge_cases", queue_edge_cases },
558 { "queue_zero_cost", queue_zero_cost },
559 { "queue_torture", queue_torture },
560 { "queue_multi_producer", queue_multi_producer },
561 { "queue_multi_queue", queue_multi_queue },
562 { "queue_stress", queue_stress },
563 { NULL, NULL },
564 };
565