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 <arpa/inet.h>
18 #include <err.h>
19 #include <libzutil.h>
20 #include <pthread.h>
21 #include <stdint.h>
22 #include <stdio.h>
23 #include <stdlib.h>
24 #include <sys/byteorder.h>
25 #include <sys/stdtypes.h>
26 #include <sys/sysmacros.h>
27 #include <sys/types.h>
28 #include <sys/zfs_ioctl.h>
29 #include <time.h>
30 #include <unistd.h>
31
32 #include "zstream_chain.h"
33 #include "zstream_modules.h"
34 #include "zstream_util.h"
35
36 /*
37 * Memory devoted to storing payloads is limited to
38 *
39 * MEMORY_BASE + (system_memory - MEMORY_BASE_CUTOFF) * MEMORY_PCT / 100
40 *
41 * When memory is exhausted, chain_read() waits until memory in use is
42 * MEMORY_HYSTERESIS bytes lower than the nominal limit.
43 */
44 #define MEMORY_BASE (512 << 20) /* 512MB */
45 #define MEMORY_BASE_CUTOFF (4ULL << 30) /* 4GB */
46 #define MEMORY_PCT 10 /* % beyond the base region */
47 #define MEMORY_HYSTERESIS (128 << 20) /* 128MB */
48
49 /*
50 * Init only the filename; chain functions will prepare the FILE *
51 */
52 typedef struct {
53 const char *ic_filename;
54 FILE *ic_fp;
55 boolean_t ic_for_reading;
56 off_t ic_offset;
57 } io_context_t;
58
59 typedef struct {
60 const char *cc_name;
61 double cc_last_sec;
62 double cc_period_sec;
63 uint64_t cc_last_bytes;
64 } checkpoint_context_t;
65
66 /*
67 * See comments at set_payload() for more information about locking and
68 * performance considerations for data-in-flight tracking. Briefly, access
69 * patterns make the mutex expensive, so it's reserved for awakening threads
70 * that are waiting for memory.
71 */
72 typedef struct {
73 pthread_mutex_t dif_mutex;
74 pthread_cond_t dif_cond;
75 uint64_t dif_current; /* atomic access only */
76 boolean_t dif_waiting; /* atomic access only */
77 uint64_t dif_allowed;
78 uint64_t dif_resume; /* dif_allowed - MEMORY_HYSTERESIS */
79 } data_in_flight_t;
80
81 static io_context_t io_contexts[MAX_IO_STREAMS];
82 static int next_io_context = 0;
83
84 static checkpoint_context_t checkpoint_contexts[MAX_IO_STREAMS];
85 static int next_checkpoint_context = 0;
86
87 static uint32_t drop_contexts[MAX_DROP_FILTERS];
88 static int next_drop_context = 0;
89
90 static data_in_flight_t payloads = {
91 .dif_mutex = PTHREAD_MUTEX_INITIALIZER,
92 .dif_cond = PTHREAD_COND_INITIALIZER
93 };
94
95 static pthread_once_t dif_init_control = PTHREAD_ONCE_INIT;
96
97 /*
98 * Called through setup_io() -> pthread_once()
99 */
100 static void
initialize_memory_tracking(void)101 initialize_memory_tracking(void)
102 {
103 int64_t pagesize = (int64_t)sysconf(_SC_PAGESIZE);
104 int64_t pages = (int64_t)sysconf(_SC_PHYS_PAGES);
105 if (pagesize < 0 || pages < 0) {
106 warnx("unable to read system memory info");
107 payloads.dif_allowed = UINT64_MAX; /* no limit */
108 } else {
109 int64_t total_mem = pagesize * pages;
110 int64_t flex = total_mem - (int64_t)MEMORY_BASE_CUTOFF;
111 int64_t addl = (double)flex * MEMORY_PCT / 100;
112 payloads.dif_allowed = MEMORY_BASE + MAX(addl, 0);
113 }
114 payloads.dif_resume = payloads.dif_allowed - MEMORY_HYSTERESIS;
115 }
116
117 /*
118 * Run from within chain execution to initialize I/O. A NULL filename
119 * indicates stdin or stdout.
120 */
121 static void
open_file(io_context_t * context)122 open_file(io_context_t *context)
123 {
124 if (context->ic_filename) {
125 context->ic_fp = fopen(context->ic_filename,
126 context->ic_for_reading ? "rb" : "wb+");
127 if (!context->ic_fp) {
128 perror(context->ic_filename);
129 exit(1);
130 }
131 } else if (context->ic_for_reading && isatty(STDIN_FILENO)) {
132 errx(1, "stream cannot be read from a terminal. "
133 "Name a file or take input from a pipe.");
134 } else if (context->ic_for_reading) {
135 context->ic_fp = stdin;
136 } else if (isatty(STDOUT_FILENO)) {
137 errx(1, "stream cannot be written to a terminal. "
138 "Capture output to a file or pipe to another command.");
139 } else {
140 context->ic_fp = stdout;
141 }
142 }
143
144 /*
145 * Extract the payload size from a replay record that is potentially
146 * byteswapped. We want to leave the bulk of byteswapping to another module,
147 * so just take a quick, nondestructive peek.
148 *
149 * Record-specific macros such as DRR_WRITE_PAYLOAD_SIZE do not seem to be
150 * byteswap-aware. However, with the exception of DRR_OBJECT_PAYLOAD_SIZE,
151 * they happen to work with post-swapping since they are switching on either
152 * a uint8_t value or 0.
153 *
154 * DRR_WRITE and DRR_SPILL use 64-bit sizes. The other two record types have
155 * 32-bit sizes. The drr_payloadlen field shared by all record types (but
156 * used only by BEGIN records is also 32 bits.
157 */
158 static size_t
calc_payload_size(dmu_replay_record_t * drr)159 calc_payload_size(dmu_replay_record_t *drr)
160 {
161 struct drr_object *drro = &drr->drr_u.drr_object;
162 struct drr_write *drrw = &drr->drr_u.drr_write;
163 struct drr_spill *drrs = &drr->drr_u.drr_spill;
164 struct drr_write_embedded *drrwe = &drr->drr_u.drr_write_embedded;
165
166 boolean_t swap = ATTR_IS_SET(CA_BYTESWAPPED);
167 uint32_t drr_type = swap ? BSWAP_32(drr->drr_type) : drr->drr_type;
168 uint64_t size, size64 = 0;
169 uint32_t size32 = 0;
170 boolean_t round = B_FALSE;
171
172 if (drr_type == DRR_OBJECT) {
173 round = drro->drr_raw_bonuslen == 0;
174 size32 = round ? drro->drr_bonuslen : drro->drr_raw_bonuslen;
175 } else if (drr_type == DRR_WRITE) {
176 size64 = DRR_WRITE_PAYLOAD_SIZE(drrw);
177 } else if (drr_type == DRR_SPILL) {
178 size64 = DRR_SPILL_PAYLOAD_SIZE(drrs);
179 } else if (drr_type == DRR_WRITE_EMBEDDED) {
180 size32 = drrwe->drr_psize;
181 round = B_TRUE;
182 } else if (drr_type == DRR_BEGIN) {
183 size32 = drr->drr_payloadlen;
184 } else {
185 return (0);
186 }
187 if (size32 != 0) {
188 size = swap ? BSWAP_32(size32) : size32;
189 } else {
190 size = swap ? BSWAP_64(size64) : size64;
191 }
192 return (round ? P2ROUNDUP(size, 8) : size);
193 }
194
195 /*
196 * Must be called only with the first record in a stream. Must be a
197 * DRR_BEGIN record or we'll terminate with "invalid stream".
198 */
199 static void
set_stream_attributes(drr_packet_t * item)200 set_stream_attributes(drr_packet_t *item)
201 {
202 dmu_replay_record_t *drr = &item->dp_drr;
203 struct drr_begin *drrb = &drr->drr_u.drr_begin;
204 uint64_t magic = drrb->drr_magic;
205 uint64_t versioninfo = drrb->drr_versioninfo;
206 boolean_t i_am_big_endian = htonl(0xFF00) == 0xFF00;
207
208 boolean_t swap_on_output, is_deduped;
209
210 if (magic == BSWAP_64(DMU_BACKUP_MAGIC)) {
211 SET_ATTR(CA_BYTESWAPPED);
212 versioninfo = BSWAP_64(versioninfo);
213 } else if (magic != DMU_BACKUP_MAGIC) {
214 errx(1, "invalid ZFS stream, bad magic number %llx",
215 (u_longlong_t)magic);
216 }
217 if (i_am_big_endian == ATTR_IS_SET(CA_BYTESWAPPED)) {
218 SET_ATTR(CA_LITTLE_ENDIAN_INPUT);
219 } else {
220 SET_ATTR(CA_BIG_ENDIAN_INPUT);
221 }
222 chain_attrs->ca_feature_flags = DMU_GET_FEATUREFLAGS(versioninfo);
223
224 is_deduped =
225 STREAM_HAS_FEATURE(DMU_BACKUP_FEATURE_DEDUP) ||
226 STREAM_HAS_FEATURE(DMU_BACKUP_FEATURE_DEDUPPROPS);
227
228 if (OPTION_ENABLED(CA_FORBID_DEDUP) && is_deduped) {
229 errx(1, "input stream is deduplicated, but this subcommand "
230 "does not support deduplicated streams. Use 'zstream "
231 "redup' to reduplicate.");
232 }
233 boolean_t req_dedup = OPTION_ENABLED(CA_REQUIRE_DEDUP);
234 boolean_t is_dedup = STREAM_HAS_FEATURE(DMU_BACKUP_FEATURE_DEDUP);
235 if (req_dedup && !is_dedup) {
236 errx(1, "this subcommand requires a deduplicated input "
237 "stream, but the stream is not deduplicated");
238 }
239 boolean_t req_native = OPTION_ENABLED(CA_REQUIRE_NATIVE_ENDIAN);
240 boolean_t is_byteswapped = ATTR_IS_SET(CA_BYTESWAPPED);
241 if (req_native && is_byteswapped) {
242 errx(1, "this subcommand requires a native-endian "
243 "input stream");
244 }
245
246 /*
247 * Figure out output endianness. In the absence of explicit byte
248 * order instructions, we default to preserving the input byte
249 * order. Record headers are always converted to native byte order
250 * for processing, but they can be swapped back on output.
251 *
252 * zfs receive inspects the endianness of each DRR record
253 * and assumes, at least in some cases, that payload data has the
254 * same order as the DMU wrappers.
255 */
256 if (OPTION_ENABLED(CA_BIG_ENDIAN_OUT))
257 swap_on_output = !i_am_big_endian;
258 else if (OPTION_ENABLED(CA_LITTLE_ENDIAN_OUT))
259 swap_on_output = i_am_big_endian;
260 else if (OPTION_ENABLED(CA_OPPOSITE_ENDIAN_OUT))
261 swap_on_output = !ATTR_IS_SET(CA_BYTESWAPPED);
262 else
263 swap_on_output = ATTR_IS_SET(CA_BYTESWAPPED);
264
265 if (swap_on_output) {
266 ENABLE_OPTION(chain_attrs, CA_BYTESWAP_ON_OUTPUT);
267 }
268 }
269
270 /*
271 * Given a desired payload size, determine whether we can read it in
272 * immediately. If not, we have to wait for memory to become available.
273 *
274 * chain_read() is a serial chain step and will always be called by the same
275 * thread. However, multiple other steps in the chain may want to modify or
276 * free payloads, so memory tracking has to be managed with multithreading
277 * in mind.
278 *
279 * The common case is that we are nowhere near the limit, which costs only a
280 * single unsynchronized read of dif_current. We lock the mutex only when we
281 * are actually going to await the condition.
282 *
283 * The store to dif_waiting and the subsequent load of dif_current are the
284 * mirror image of the sequence in set_payload_impl(), which stores
285 * dif_current and then loads dif_waiting. Both halves must be sequentially
286 * consistent: if either were weaker, the two threads could miss each
287 * other's store.
288 */
289 static inline void
maybe_wait_for_memory(size_t bytes_wanted)290 maybe_wait_for_memory(size_t bytes_wanted)
291 {
292 uint64_t current = __atomic_load_n(&payloads.dif_current,
293 __ATOMIC_RELAXED);
294
295 if (current + bytes_wanted <= payloads.dif_allowed)
296 return;
297
298 pthread_mutex_lock(&payloads.dif_mutex);
299 __atomic_store_n(&payloads.dif_waiting, B_TRUE, __ATOMIC_SEQ_CST);
300 while (__atomic_load_n(&payloads.dif_current, __ATOMIC_SEQ_CST) >
301 payloads.dif_resume) {
302 pthread_cond_wait(&payloads.dif_cond, &payloads.dif_mutex);
303 }
304 /*
305 * A freeing thread that sees a stale B_TRUE here just takes the
306 * mutex for a broadcast that no one is waiting for, so this store
307 * needs no ordering of its own.
308 */
309 __atomic_store_n(&payloads.dif_waiting, B_FALSE, __ATOMIC_RELAXED);
310 pthread_mutex_unlock(&payloads.dif_mutex);
311 }
312
313 /*
314 * Read in an item's payload. We don't do memory accounting here because
315 * that's now handled by set_payload(). This function reads the payload into
316 * a newly allocated buffer and returns the buffer. set_payload() attaches
317 * an existing buffer to a dp_drr_t item.
318 */
319 static inline uint8_t *
read_payload(io_context_t * context,size_t size)320 read_payload(io_context_t *context, size_t size)
321 {
322 maybe_wait_for_memory(size);
323 uint8_t *buff = safe_malloc(size);
324 size_t n_read = fread(buff, size, 1, context->ic_fp);
325 if (n_read != 1) {
326 if (ferror(context->ic_fp)) {
327 err(1, "error reading record payload at offset %llu",
328 (u_longlong_t)context->ic_offset);
329 } else {
330 /*
331 * We can't exit here because ZTS depends on being
332 * able to process randomly truncated streams.
333 */
334 warnx("input ends mid-record at offset %llu - "
335 "stream is likely corrupt",
336 (u_longlong_t)context->ic_offset);
337 fclose(context->ic_fp);
338 free(buff);
339 return (NULL);
340 }
341 }
342 return (buff);
343 }
344
345 static disposition_t
chain_read(void * item_in,void * context_in)346 chain_read(void *item_in, void *context_in)
347 {
348 drr_packet_t *item = (drr_packet_t *)item_in;
349 io_context_t *context = (io_context_t *)context_in;
350
351 if (item == NULL)
352 return (D_OK);
353
354 dmu_replay_record_t *drr = &item->dp_drr;
355
356 if (!context->ic_fp)
357 open_file(context);
358
359 item->dp_payload = NULL;
360 item->dp_payload_size = 0;
361 item->dp_stream_offset = context->ic_offset;
362
363 if (fread(drr, sizeof (dmu_replay_record_t), 1, context->ic_fp) != 1) {
364 if (ferror(context->ic_fp)) {
365 err(1, "error reading record header at offset %llu",
366 (u_longlong_t)context->ic_offset);
367 }
368 fclose(context->ic_fp);
369 return (D_EOF);
370 }
371
372 if (context->ic_offset == 0)
373 set_stream_attributes(item);
374
375 size_t payload_size = calc_payload_size(drr);
376 if (payload_size > UINT32_MAX) {
377 errx(1, "stated packet size is greater than uint32_t "
378 "at offset %llu", (u_longlong_t)context->ic_offset);
379 } else if (payload_size > 0) {
380 uint8_t *buff = read_payload(context, payload_size);
381 if (buff == NULL)
382 return (D_EOF);
383 set_payload(item, buff, payload_size);
384 }
385
386 uint32_t drr_type = ATTR_IS_SET(CA_BYTESWAPPED) ?
387 BSWAP_32(drr->drr_type) : drr->drr_type;
388 if (drr_type >= DRR_NUMTYPES) {
389 err(1, "invalid record type %llu found at offset %llu",
390 (u_longlong_t)drr_type, (u_longlong_t)context->ic_offset);
391 }
392
393 context->ic_offset += sizeof (*drr) + item->dp_payload_size;
394
395 record_stats_t *stats = &chain_attrs->ca_stats_in[drr_type];
396 stats->rs_num_records++;
397 stats->rs_total_header_bytes += sizeof (dmu_replay_record_t);
398 stats->rs_total_payload_bytes += item->dp_payload_size;
399
400 stats = &chain_attrs->ca_totals_in;
401 stats->rs_num_records++;
402 stats->rs_total_header_bytes += sizeof (dmu_replay_record_t);
403 stats->rs_total_payload_bytes += item->dp_payload_size;
404
405 return (D_OK);
406 }
407
408 static disposition_t
chain_write(void * item_in,void * context_in)409 chain_write(void *item_in, void *context_in)
410 {
411 drr_packet_t *item = (drr_packet_t *)item_in;
412 io_context_t *context = (io_context_t *)context_in;
413
414 if (item == NULL) {
415 if (context->ic_fp) {
416 if (fclose(context->ic_fp) != 0)
417 err(1, "error closing output stream");
418 context->ic_fp = NULL;
419 }
420 VERIFY0(__atomic_load_n(&payloads.dif_current,
421 __ATOMIC_SEQ_CST));
422 return (D_OK);
423 }
424
425 if (!context->ic_fp) {
426 open_file(context);
427 }
428
429 dmu_replay_record_t *drr = &item->dp_drr;
430
431 if (fwrite(drr, sizeof (dmu_replay_record_t), 1, context->ic_fp) != 1) {
432 err(1, "error writing record header");
433 } else if (item->dp_payload_size > 0) {
434 size_t n_written = fwrite(item->dp_payload,
435 item->dp_payload_size, 1, context->ic_fp);
436 if (n_written != 1) {
437 err(1, "error writing payload");
438 }
439 }
440
441 uint32_t drr_type = OPTION_ENABLED(CA_BYTESWAP_ON_OUTPUT) ?
442 BSWAP_32(drr->drr_type) : drr->drr_type;
443
444 record_stats_t *stats = &chain_attrs->ca_stats_out[drr_type];
445 stats->rs_num_records++;
446 stats->rs_total_header_bytes += sizeof (dmu_replay_record_t);
447 stats->rs_total_payload_bytes += item->dp_payload_size;
448
449 stats = &chain_attrs->ca_totals_out;
450 stats->rs_num_records++;
451 stats->rs_total_header_bytes += sizeof (dmu_replay_record_t);
452 stats->rs_total_payload_bytes += item->dp_payload_size;
453
454 set_payload(item, NULL, 0);
455 return (D_OK);
456 }
457
458 /*
459 * Even if the chain doesn't write out a stream, payloads still need freed.
460 */
461 static disposition_t
chain_null_output(void * item_in,void * context)462 chain_null_output(void *item_in, void *context)
463 {
464 (void) context;
465 drr_packet_t *item = (drr_packet_t *)item_in;
466
467 if (item == NULL)
468 return (D_OK);
469
470 set_payload(item, NULL, 0);
471 return (D_OK);
472 }
473
474 /*
475 * Storage for the filename must remain valid during chain execution
476 */
477 static chain_step_t
setup_io(const char * filename,boolean_t for_reading)478 setup_io(const char *filename, boolean_t for_reading)
479 {
480 pthread_once(&dif_init_control, initialize_memory_tracking);
481 int context_num = next_io_context++ % MAX_IO_STREAMS;
482
483 io_context_t context = {
484 .ic_filename = filename,
485 .ic_for_reading = for_reading
486 };
487 io_contexts[context_num] = context;
488
489 chain_step_t step = {
490 .cs_type = CS_SERIAL,
491 .cs_in_size = for_reading ? 0 : sizeof (drr_packet_t),
492 .cs_out_size = for_reading ? sizeof (drr_packet_t) : 0,
493 .cs_context = &io_contexts[context_num],
494 .cs_serial = {
495 .process = for_reading ? chain_read : chain_write
496 }
497 };
498 return (step);
499 }
500
501 chain_step_t
serial_read_stream(const char * filename)502 serial_read_stream(const char *filename)
503 {
504 return (setup_io(filename, B_TRUE));
505 }
506
507 chain_step_t
serial_write_stream(const char * filename)508 serial_write_stream(const char *filename)
509 {
510 return (setup_io(filename, B_FALSE));
511 }
512
513 chain_step_t
serial_null_output(void)514 serial_null_output(void)
515 {
516 chain_step_t step = {
517 .cs_type = CS_SERIAL,
518 .cs_in_size = sizeof (drr_packet_t),
519 .cs_out_size = 0,
520 .cs_context = NULL,
521 .cs_serial = {
522 .process = chain_null_output
523 }
524 };
525 return (step);
526 }
527
528 size_t
constant_cost_of_one(queue_item_t * packet,void * context)529 constant_cost_of_one(queue_item_t *packet, void *context)
530 {
531 (void) context;
532 (void) packet;
533 return (1);
534 }
535
536 size_t
payload_size_as_cost(queue_item_t * packet_in,void * context)537 payload_size_as_cost(queue_item_t *packet_in, void *context)
538 {
539 (void) context;
540 drr_packet_t *packet = (drr_packet_t *)packet_in;
541 return (packet->dp_payload_size);
542 }
543
544 static disposition_t
chain_checkpoint(void * item_in,void * ctxt_in)545 chain_checkpoint(void *item_in, void *ctxt_in)
546 {
547 drr_packet_t *item = (drr_packet_t *)item_in;
548 checkpoint_context_t *ctxt = (checkpoint_context_t *)ctxt_in;
549
550 struct timespec now;
551 char buff[32];
552 uint64_t delta_b, dbdt;
553 double now_sec, delta_t;
554
555 if (item == NULL)
556 return (D_OK);
557
558 clock_gettime(CLOCK_MONOTONIC, &now);
559 now_sec = now.tv_sec + (double)now.tv_nsec / 1E9;
560 if (ctxt->cc_last_sec > 1E-9) {
561 delta_t = now_sec - ctxt->cc_last_sec;
562 if (delta_t < ctxt->cc_period_sec)
563 return (D_OK);
564 delta_b = item->dp_stream_offset - ctxt->cc_last_bytes;
565 dbdt = delta_b / delta_t;
566 zfs_nicenum(dbdt, buff, sizeof (buff));
567 fprintf(stderr, "Checkpoint %s: %s/s\n", ctxt->cc_name, buff);
568 }
569 ctxt->cc_last_sec = now_sec;
570 ctxt->cc_last_bytes = item->dp_stream_offset;
571 return (D_OK);
572 }
573
574 /*
575 * Storage for name must remain valid throughout chain execution
576 */
577 chain_step_t
serial_checkpoint(const char * name)578 serial_checkpoint(const char *name)
579 {
580 int context_no = next_checkpoint_context++ % MAX_IO_STREAMS;
581
582 checkpoint_context_t context = {
583 .cc_name = name,
584 .cc_period_sec = 1.0
585 };
586 checkpoint_contexts[context_no] = context;
587
588 chain_step_t step = {
589 .cs_type = CS_SERIAL,
590 .cs_in_size = sizeof (drr_packet_t),
591 .cs_out_size = sizeof (drr_packet_t),
592 .cs_context = &checkpoint_contexts[context_no],
593 .cs_serial = {
594 .process = chain_checkpoint
595 },
596 };
597 return (step);
598 }
599
600 static disposition_t
chain_drop_record_types(void * item_in,void * context_in)601 chain_drop_record_types(void *item_in, void *context_in)
602 {
603 drr_packet_t *item = (drr_packet_t *)item_in;
604 uint32_t *context = (uint32_t *)context_in;
605
606 if (item == NULL)
607 return (D_OK);
608
609 uint32_t type = (uint32_t)item->dp_drr.drr_type;
610 if (type >= DRR_NUMTYPES) {
611 errx(1, "invalid record type %u found at offset %llu "
612 "(place drop filter downstream of byteswapping?)",
613 type, (u_longlong_t)item->dp_stream_offset);
614 }
615
616 if (((UINT32_C(1) << type) & *context) != 0) {
617 set_payload(item, NULL, 0);
618 return (D_DROP);
619 }
620 return (D_OK);
621 }
622
623 chain_step_t
serial_drop_record_types(uint32_t drop_mask)624 serial_drop_record_types(uint32_t drop_mask)
625 {
626 int context_no = next_drop_context++ % MAX_DROP_FILTERS;
627 uint32_t *context = &drop_contexts[context_no];
628
629 *context = drop_mask;
630
631 chain_step_t step = {
632 .cs_type = CS_SERIAL,
633 .cs_in_size = sizeof (drr_packet_t),
634 .cs_out_size = sizeof (drr_packet_t),
635 .cs_context = context,
636 .cs_serial = {
637 .process = chain_drop_record_types
638 },
639 };
640 return (step);
641 }
642
643 /*
644 * Every record that carries a payload passes through here at least twice,
645 * once on the way in and once on the way out. Those calls are also made by
646 * different threads, which unfortunately is something of an adversarial
647 * pattern for a pthreads mutex. The cache line containing the lock
648 * structure ping-pongs among cores, and each move is performed with
649 * restrictive memory barriers. We use atomic operations instead and reserve
650 * the mutex for waking up a sleeping memory consumer.
651 *
652 * The atomic update is done in sequentially consistent mode because the
653 * (potential) load of dif_waiting beneath must not be hoisted above the
654 * update to dif_current. dif_waiting shares a cache line with dif_current,
655 * which we have just acquired exclusively, so reading it is essentially
656 * free.
657 */
658 static void
set_payload_impl(void * item_in,void * payload_in,uint64_t size,boolean_t free_old)659 set_payload_impl(void *item_in, void *payload_in, uint64_t size,
660 boolean_t free_old)
661 {
662 drr_packet_t *item = (drr_packet_t *)item_in;
663 uint8_t *payload = (uint8_t *)payload_in;
664 VERIFY(payload != NULL || size == 0);
665 VERIFY(payload == NULL || payload != item->dp_payload);
666
667 if (free_old && item->dp_payload != NULL) {
668 free(item->dp_payload);
669 }
670 int64_t delta = (int64_t)size - (int64_t)item->dp_payload_size;
671 VERIFY3U(size, <=, UINT32_MAX);
672 item->dp_payload = payload;
673 item->dp_payload_size = (uint32_t)size;
674
675 /*
676 * Atomics are nominally unsigned operations, but because of twos
677 * complement arithmetic it's fine to add a "negative" value.
678 */
679 uint64_t current = __atomic_add_fetch(&payloads.dif_current, delta,
680 __ATOMIC_SEQ_CST);
681
682 /*
683 * Only a net reduction can release a parked reader, and only if it
684 * brings us back under the hysteresis threshold.
685 */
686 if (delta < 0 && current <= payloads.dif_resume &&
687 __atomic_load_n(&payloads.dif_waiting, __ATOMIC_SEQ_CST)) {
688 pthread_mutex_lock(&payloads.dif_mutex);
689 pthread_cond_broadcast(&payloads.dif_cond);
690 pthread_mutex_unlock(&payloads.dif_mutex);
691 }
692 }
693
694 void
set_payload(void * item_in,void * payload_in,uint64_t size)695 set_payload(void *item_in, void *payload_in, uint64_t size)
696 {
697 set_payload_impl(item_in, payload_in, size, B_TRUE);
698 }
699
700 /*
701 * Remove a payload from the chain's management without freeing. It's up to
702 * the recipient to free the buffer.
703 */
704 void
export_payload(void * item_in)705 export_payload(void *item_in)
706 {
707 set_payload_impl(item_in, NULL, 0, B_FALSE);
708 }
709