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 #include <assert.h>
18 #include <err.h>
19 #include <libspl.h>
20 #include <pthread.h>
21 #include <stddef.h>
22 #include <stdio.h>
23 #include <sys/abd.h>
24 #include <sys/param.h>
25 #include <sys/stdtypes.h>
26 #include <sys/zio.h>
27 #include <sys/zstd/zstd.h>
28 #include <sys/zfs_refcount.h>
29 #include <zfs_fletcher.h>
30
31 #include "zstream_chain.h"
32 #include "zstream_queue.h"
33 #include "zstream_util.h"
34
35 #define MAX_CHAIN_LENGTH 32
36
37 /*
38 * Calculated information about a chain. Not to be confused with
39 * chain_attrs_t, which is a global set of options and stream attributes
40 * available to all chain steps.
41 */
42 typedef struct {
43 int ct_num_steps;
44 int ct_num_queues;
45 size_t ct_item_size;
46 } chain_stats_t;
47
48 /*
49 * Data passed to worker threads
50 */
51 typedef struct {
52 chain_step_t *wc_steps;
53 int wc_num_steps;
54 size_t wc_buffer_size;
55 zstream_queue_t *wc_in_queue;
56 zstream_queue_t *wc_out_queue;
57 } worker_context_t;
58
59 chain_attrs_t *chain_attrs;
60
61 chain_step_t
chain_terminator(void)62 chain_terminator(void)
63 {
64 chain_step_t step = { .cs_type = CS_TERMINATE };
65 return (step);
66 }
67
68 static void
libraries_init(void)69 libraries_init(void)
70 {
71 zfs_refcount_init();
72 abd_init();
73 zio_init();
74 zstd_init();
75 libspl_init();
76 fletcher_4_init();
77 }
78
79 static void
libraries_fini(void)80 libraries_fini(void)
81 {
82 fletcher_4_fini();
83 libspl_fini();
84 zio_fini();
85 zstd_fini();
86 abd_fini();
87 zfs_refcount_fini();
88 }
89
90 /*
91 * Body function for worker threads
92 */
93 static void *
zstream_chain_worker(void * ctxt_in)94 zstream_chain_worker(void *ctxt_in)
95 {
96 worker_context_t *ctxt = (worker_context_t *)ctxt_in;
97 uint8_t buffer[ctxt->wc_buffer_size];
98 boolean_t done = B_FALSE;
99
100 while (!done) {
101 for (int i = 0; i < ctxt->wc_num_steps; i++) {
102 chain_step_t *step = &ctxt->wc_steps[i];
103 if (step->cs_type == CS_SERIAL) {
104 if (done) {
105 (void) step->cs_serial.process(NULL,
106 step->cs_context);
107 } else {
108 disposition_t dispo =
109 step->cs_serial.process(buffer,
110 step->cs_context);
111 if (dispo == D_EOF) {
112 done = B_TRUE;
113 } else if (dispo == D_DROP) {
114 break;
115 }
116 }
117 } else if (i == 0) {
118 done = done ||
119 !zstream_dequeue(ctxt->wc_in_queue, buffer);
120 } else if (done) {
121 zstream_queue_fini(ctxt->wc_out_queue);
122 } else {
123 zstream_enqueue(ctxt->wc_out_queue, buffer);
124 }
125 }
126 }
127 return (NULL);
128 }
129
130 /*
131 * Validate chain and calculate number of steps, max packet size, and number
132 * of zstream_queues that must be created.
133 */
134 static chain_stats_t
validate_chain(zstream_chain_t chain)135 validate_chain(zstream_chain_t chain)
136 {
137 int num_steps = 0;
138 int num_queues = 0;
139 size_t item_size = 0;
140
141 while (chain[num_steps].cs_type != CS_TERMINATE) {
142 if (num_steps >= MAX_CHAIN_LENGTH) {
143 errx(1, "unterminated zstream_chain");
144 }
145 chain_step_t *step = &chain[num_steps];
146 item_size = MAX(item_size, step->cs_out_size);
147 if (step->cs_type == CS_PARALLEL) {
148 num_queues++;
149 }
150 num_steps++;
151 }
152 VERIFY3U(num_steps, >, 0);
153
154 boolean_t first_parallel = chain[0].cs_type == CS_PARALLEL;
155 boolean_t last_parallel = chain[num_steps-1].cs_type == CS_PARALLEL;
156 if (first_parallel || last_parallel) {
157 errx(1, "a chain cannot start or end with a parallel step");
158 }
159
160 /*
161 * Check for consistency of input and output packet sizes in
162 * adjacent steps.
163 */
164 for (int i = 0; i < num_steps; i++) {
165 if (i > 0 && chain[i].cs_in_size != chain[i-1].cs_out_size) {
166 warnx("adjacent chain steps %d and %d declare "
167 "incompatible packet sizes", i - 1, i);
168 }
169 }
170
171 chain_stats_t stats = {
172 .ct_num_steps = num_steps,
173 .ct_num_queues = num_queues,
174 .ct_item_size = item_size
175 };
176 return (stats);
177 }
178
179 /*
180 * Execute a chain of processing steps, some parallel and some serial.
181 *
182 * For simplicity, we normalize the chain item size to that of the largest
183 * output of any step. Payloads are allocated on the heap, so the maximal
184 * item size is typically on the order of 512 bytes.
185 *
186 * Packets with data beyond the base drr_record_t should add their
187 * additional data to the end of the packet, and this area may be reused for
188 * different purposes as items travel down the chain.
189 *
190 * One worker thread is assigned to every contiguous sequence of serial
191 * steps plus the parallel steps on either side of that block (if any).
192 * Adjacent parallel steps also receive a worker. This isn't a special case,
193 * it's just the same rule with the serial block consisting of zero steps.
194 *
195 * Parallel steps are double-covered, which is the intended behavior. If a
196 * worker's domain begins with a parallel step, it dequeues items from the
197 * associated queue. If the domain ends with a parallel step, it submits
198 * items to that queue.
199 */
200 void
zstream_chain_exec(zstream_chain_t chain,chain_attrs_t * attrs)201 zstream_chain_exec(zstream_chain_t chain, chain_attrs_t *attrs)
202 {
203 chain_stats_t stats = validate_chain(chain);
204
205 int num_workers = stats.ct_num_queues + 1;
206 worker_context_t contexts[num_workers];
207 pthread_t worker_threads[num_workers];
208 zstream_queue_t *queue;
209
210 /*
211 * Create parallel queues and worker thread contexts
212 *
213 * We do not need to track zstream_queues independently of worker
214 * contexts because queues clean themselves up once the last item
215 * has been dequeued. The stream eventually ends, so some worker
216 * thread will eventually call zstream_queue_fini() on every queue.
217 */
218
219 int worker = 0;
220
221 worker_context_t context = {
222 .wc_steps = chain,
223 .wc_num_steps = 0,
224 .wc_buffer_size = stats.ct_item_size,
225 .wc_in_queue = NULL
226 };
227 contexts[worker] = context;
228
229 for (int i = 0; i < stats.ct_num_steps; i++) {
230 contexts[worker].wc_num_steps++;
231 if (chain[i].cs_type == CS_PARALLEL) {
232 chain_step_t *cs = &chain[i];
233 zq_params_t queue_params = {
234 .qp_process = cs->cs_parallel.process,
235 .qp_cost = cs->cs_parallel.cost,
236 .qp_item_size = stats.ct_item_size,
237 .qp_context = cs->cs_context
238 };
239 queue = zstream_queue_create(&queue_params);
240 contexts[worker].wc_out_queue = queue;
241 worker++;
242 worker_context_t next_context = {
243 .wc_steps = cs,
244 .wc_num_steps = 1,
245 .wc_buffer_size = stats.ct_item_size,
246 .wc_in_queue = queue
247 };
248 contexts[worker] = next_context;
249 }
250 }
251
252 contexts[worker].wc_out_queue = NULL;
253
254 chain_attrs_t backup_attrs = {0};
255 chain_attrs = attrs ? attrs : &backup_attrs;
256
257 libraries_init();
258
259 /* Spawn threads */
260 for (int i = 0; i < num_workers; i++) {
261 char name[32];
262 snprintf(name, sizeof (name), "chain-%d", i);
263 worker_threads[i] = safe_create_thread(zstream_chain_worker,
264 &contexts[i], name, B_FALSE);
265 }
266
267 /* Reap threads */
268 for (int i = 0; i < num_workers; i++) {
269 int ret = pthread_join(worker_threads[i], NULL);
270 VERIFY3S(ret, ==, 0);
271 }
272
273 libraries_fini();
274 }
275
276 /*
277 * Execute a chain linearly, without queues and without multithreading. This
278 * form of execution is intended as a debugging aid, both for clients and
279 * for the chain mechanism itself. If this variant doesn't produce results
280 * identical to zstream_chain_exec(), there's a multithreading-related bug
281 * somewhere.
282 *
283 * It is not necessary to remove parallel steps from the input chain. They
284 * are accepted as-is, but their execution won't be parallelized.
285 */
286 void
zstream_chain_exec_serialized(zstream_chain_t chain,chain_attrs_t * attrs)287 zstream_chain_exec_serialized(zstream_chain_t chain, chain_attrs_t *attrs)
288 {
289 chain_stats_t stats = validate_chain(chain);
290
291 uint8_t buffer[stats.ct_item_size];
292 boolean_t done = B_FALSE;
293
294 chain_attrs_t backup_attrs = {0};
295 chain_attrs = attrs ? attrs : &backup_attrs;
296
297 libraries_init();
298
299 while (!done) {
300 for (int i = 0; i < stats.ct_num_steps; i++) {
301 chain_step_t *step = &chain[i];
302 if (step->cs_type == CS_SERIAL) {
303 if (done) {
304 (void) step->cs_serial.process(NULL,
305 step->cs_context);
306 } else {
307 disposition_t dispo =
308 step->cs_serial.process(buffer,
309 step->cs_context);
310 if (dispo == D_EOF) {
311 done = B_TRUE;
312 } else if (dispo == D_DROP) {
313 break;
314 }
315 }
316 } else if (!done) {
317 size_t cost = step->cs_parallel.cost(buffer,
318 step->cs_context);
319 if (cost > 0) {
320 step->cs_parallel.process(buffer,
321 step->cs_context);
322 }
323 }
324 }
325 }
326
327 libraries_fini();
328 }
329