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 #ifndef _ZSTREAM_QUEUE_H 18 #define _ZSTREAM_QUEUE_H 19 20 #ifdef __cplusplus 21 extern "C" { 22 #endif 23 24 #include <stddef.h> 25 #include <sys/stdtypes.h> 26 27 /* 28 * This is a generalized implementation of multithreaded FIFO work queues. 29 * 30 * Callers define a fixed item size to be used by each queue and supply two 31 * thread-safe functions that 1) estimate individual items' processing costs 32 * and 2) perform the actual processing. The queue never inspects or 33 * interprets items in the queue, so processing functions can modify them as 34 * they wish. 35 * 36 * The cost function assigns a size_t cost that estimates the amount of work 37 * needed to process an item. Payload length is a natural cost basis for 38 * operations such as hashing and data compression. 39 * 40 * It's expected that only a subset of input items will require processing. 41 * If an item's cost is 0, it is fast-tracked and never presented to the 42 * processing function. 43 * 44 * The cost function is run as items enter the queue, with the queue mutex 45 * held, so it should return a value promptly. If cost estimation is 46 * expensive and important, use a separate queue to implement it. Cost 47 * values should have a roughly linear relationship to processing time. 48 * Only the ratio between costs matters, so choose units that keep the sum 49 * of a batch's costs comfortably inside a size_t. 50 * 51 * Work is processed in batches whose size is tuned automatically. Each 52 * queue observes how long its own work actually takes per unit of cost, and 53 * claims enough items to fill a fixed slice of processing time. No batch 54 * exceeds ZQ_MAX_BATCH items. 55 * 56 * All queues share a single thread pool that is managed to avoid 57 * contention. Threads are assigned to queues dynamically according to where 58 * work is available. When multiple queues have work, threads are allocated 59 * among them stochastically, in proportion to the amount of unclaimed work 60 * each queue is holding. 61 * 62 * The shared thread pool persists until the process exits. 63 */ 64 65 /* 66 * Implementation constants, exposed only so that tests and diagnostics can 67 * reason about the real boundaries. Except for ZQ_MIN_THREADS, these values 68 * are not caller-tunable. ZQ_MIN_THREADS can be overridden by calling 69 * zstream_queue_set_num_threads(). 70 */ 71 #define ZQ_SLOTS_PER_QUEUE 4096 /* Ring buffer slots, every queue */ 72 #define ZQ_MAX_BATCH 1024 /* Most items claimable at once */ 73 #define ZQ_MAX_QUEUES 16 /* Max number of simultaneous queues */ 74 #define ZQ_MIN_THREADS 6 75 76 typedef void queue_item_t; 77 78 typedef struct zstream_queue zstream_queue_t; 79 80 /* 81 * Signatures that cost and processing functions must conform to. 82 */ 83 typedef size_t 84 zq_estimate_cost_f(queue_item_t *item, void *context); 85 86 typedef void 87 zq_process_item_f(queue_item_t *item, void *context); 88 89 /* 90 * Set the number of threads to be spawned for queue work. Since all queues 91 * share a thread pool, this value affects all queues. The value must be set 92 * before any queues are created. By default, one thread is spawned for 93 * every CPU core, but always at least ZQ_MIN_THREADS threads. 94 */ 95 void 96 zstream_queue_set_num_threads(int num_threads); 97 98 /* 99 * Create a queue. The qp_context field is passed to the cost and processing 100 * functions and is not examined by the queue itself. 101 */ 102 103 typedef struct { 104 zq_process_item_f *qp_process; 105 zq_estimate_cost_f *qp_cost; 106 void *qp_context; 107 size_t qp_item_size; 108 } zq_params_t; 109 110 zstream_queue_t * 111 zstream_queue_create(zq_params_t *params); 112 113 /* 114 * Submit a work item. Blocks if the queue is full. The work item is 115 * shallow-copied into the queue. Multiple threads may enqueue at once. 116 */ 117 void 118 zstream_enqueue(zstream_queue_t *queue, queue_item_t *item); 119 120 /* 121 * Retrieve a completed work item. The caller must provide a buffer into 122 * which the dequeued item is shallow-copied. If the next item is not yet 123 * ready, this call will block. 124 * 125 * If zstream_dequeue returns B_FALSE, the stream is complete. The returned 126 * item is not valid and no further calls may be made on the queue. 127 * 128 * Only one thread may dequeue at once. 129 */ 130 boolean_t 131 zstream_dequeue(zstream_queue_t *queue, queue_item_t *item); 132 133 /* 134 * Declare that all items have been submitted. The queue will continue to 135 * function normally for dequeuers and worker threads until zstream_dequeue() 136 * returns B_FALSE, at which point the queue will be destroyed. 137 */ 138 void 139 zstream_queue_fini(zstream_queue_t *queue); 140 141 #ifdef __cplusplus 142 } 143 #endif 144 145 #endif /* _ZSTREAM_QUEUE_H */ 146