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. For operations like hashing and data 38 * compression, the natural cost is typically payload length. 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. 47 * 48 * Dispatch granularity is specified as a per-batch budget that is set for 49 * each queue in the same (arbitrary) units used for item costs. Threads 50 * claim items until the budget is met, there are no more items available, 51 * or ZQ_MAX_BATCH items have been claimed. When claiming items to work on, 52 * threads never block waiting for additional work to arrive. They start 53 * work as quickly as possible even if the budget has not been reached. 54 * 55 * A batch budget of 0 means that all batches will have a size of 1. 56 * 57 * All queues share a single thread pool that is managed to avoid 58 * contention. Threads are assigned to queues dynamically according to where 59 * work is available. When multiple queues have work, threads are allocated 60 * among them stochastically with an eye toward preventing pipeline stalls. 61 * 62 * The shared thread pool persists until the process exits. 63 */ 64 65 #define ZQ_MAX_BATCH 32 /* The most items that can be claimed at once */ 66 #define ZQ_MAX_QUEUES 16 /* The maximum number of simultaneous queues */ 67 #define ZQ_MIN_THREADS 6 68 69 typedef void queue_item_t; 70 71 typedef struct zstream_queue zstream_queue_t; 72 73 /* 74 * Signatures that cost and processing functions must conform to. 75 */ 76 typedef size_t 77 zq_estimate_cost_f(queue_item_t *item, void *context); 78 79 typedef void 80 zq_process_item_f(queue_item_t *item, void *context); 81 82 /* 83 * Set the number of threads to be spawned for queue work. Since all queues 84 * share a thread pool, this value affects all queues. The value must be set 85 * before any queues are created. By default, one thread is spawned for 86 * every CPU core, but always at least ZQ_MIN_THREADS threads. 87 */ 88 void 89 zstream_queue_set_num_threads(int num_threads); 90 91 /* 92 * Create a queue. The qp_context field is passed to the cost and processing 93 * functions and is not examined by the queue itself. 94 */ 95 96 typedef struct { 97 zq_process_item_f *qp_process; 98 zq_estimate_cost_f *qp_cost; 99 void *qp_context; 100 size_t qp_item_size; 101 size_t qp_batch_budget; 102 size_t qp_queue_length; 103 } zq_params_t; 104 105 zstream_queue_t * 106 zstream_queue_create(zq_params_t *params); 107 108 /* 109 * Submit a work item. Blocks if the queue is full. The work item is 110 * shallow-copied into the queue. Multiple threads may enqueue at once. 111 */ 112 void 113 zstream_enqueue(zstream_queue_t *queue, queue_item_t *item); 114 115 /* 116 * Retrieve a completed work item. The caller must provide a buffer into 117 * which the dequeued item is shallow-copied. If the next item is not yet 118 * ready, this call will block. 119 * 120 * If zstream_dequeue returns B_FALSE, the stream is complete. The returned 121 * item is not valid and no further calls may be made on the queue. 122 * 123 * Only one thread may dequeue at once. 124 */ 125 boolean_t 126 zstream_dequeue(zstream_queue_t *queue, queue_item_t *item); 127 128 /* 129 * Declare that all items have been submitted. The queue will continue to 130 * function normally for dequeuers and worker threads until zstream_dequeue() 131 * returns B_FALSE, at which point the queue will be destroyed. 132 */ 133 void 134 zstream_queue_fini(zstream_queue_t *queue); 135 136 #ifdef __cplusplus 137 } 138 #endif 139 140 #endif /* _ZSTREAM_QUEUE_H */ 141