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 <stdint.h>
21 #include <stdio.h>
22 #include <stdlib.h>
23 #include <sys/byteorder.h>
24 #include <sys/stdtypes.h>
25 #include <sys/sysmacros.h>
26 #include <sys/types.h>
27 #include <sys/zfs_ioctl.h>
28 #include <time.h>
29 #include <unistd.h>
30
31 #include "zstream_modules.h"
32 #include "zstream_util.h"
33
34 /*
35 * Init only the filename; chain functions will prepare the FILE *
36 */
37 typedef struct {
38 const char *ic_filename;
39 FILE *ic_fp;
40 boolean_t ic_for_reading;
41 off_t ic_offset;
42 } io_context_t;
43
44 typedef struct {
45 const char *cc_name;
46 double cc_last_sec;
47 double cc_period_sec;
48 uint64_t cc_last_bytes;
49 } checkpoint_context_t;
50
51 static io_context_t io_contexts[MAX_IO_STREAMS];
52 static int next_io_context = 0;
53
54 static checkpoint_context_t checkpoint_contexts[MAX_IO_STREAMS];
55 static int next_checkpoint_context = 0;
56
57 static uint32_t drop_contexts[MAX_DROP_FILTERS];
58 static int next_drop_context = 0;
59
60 /*
61 * Run from within chain execution to initialize I/O. A NULL filename
62 * indicates stdin or stdout.
63 */
64 static void
open_file(io_context_t * context)65 open_file(io_context_t *context)
66 {
67 if (context->ic_filename) {
68 context->ic_fp = fopen(context->ic_filename,
69 context->ic_for_reading ? "rb" : "wb+");
70 if (!context->ic_fp) {
71 perror(context->ic_filename);
72 exit(1);
73 }
74 } else if (context->ic_for_reading && isatty(STDIN_FILENO)) {
75 errx(1, "stream cannot be read from a terminal. "
76 "Name a file or take input from a pipe.");
77 } else if (context->ic_for_reading) {
78 context->ic_fp = stdin;
79 } else if (isatty(STDOUT_FILENO)) {
80 errx(1, "stream cannot be written to a terminal. "
81 "Capture output to a file or pipe to another command.");
82 } else {
83 context->ic_fp = stdout;
84 }
85 }
86
87 /*
88 * Extract the payload size from a replay record that is potentially
89 * byteswapped. We want to leave the bulk of byteswapping to another module,
90 * so just take a quick, nondestructive peek.
91 *
92 * Record-specific macros such as DRR_WRITE_PAYLOAD_SIZE do not seem to be
93 * byteswap-aware. However, with the exception of DRR_OBJECT_PAYLOAD_SIZE,
94 * they happen to work with post-swapping since they are switching on either
95 * a uint8_t value or 0.
96 *
97 * DRR_WRITE and DRR_SPILL use 64-bit sizes. The other two record types have
98 * 32-bit sizes. The drr_payloadlen field shared by all record types (but
99 * used only by BEGIN records is also 32 bits.
100 */
101 static size_t
calc_payload_size(dmu_replay_record_t * drr)102 calc_payload_size(dmu_replay_record_t *drr)
103 {
104 struct drr_object *drro = &drr->drr_u.drr_object;
105 struct drr_write *drrw = &drr->drr_u.drr_write;
106 struct drr_spill *drrs = &drr->drr_u.drr_spill;
107 struct drr_write_embedded *drrwe = &drr->drr_u.drr_write_embedded;
108
109 boolean_t swap = ATTR_IS_SET(CA_BYTESWAPPED);
110 uint32_t drr_type = swap ? BSWAP_32(drr->drr_type) : drr->drr_type;
111 uint64_t size, size64 = 0;
112 uint32_t size32 = 0;
113 boolean_t round = B_FALSE;
114
115 if (drr_type == DRR_OBJECT) {
116 round = drro->drr_raw_bonuslen == 0;
117 size32 = round ? drro->drr_bonuslen : drro->drr_raw_bonuslen;
118 } else if (drr_type == DRR_WRITE) {
119 size64 = DRR_WRITE_PAYLOAD_SIZE(drrw);
120 } else if (drr_type == DRR_SPILL) {
121 size64 = DRR_SPILL_PAYLOAD_SIZE(drrs);
122 } else if (drr_type == DRR_WRITE_EMBEDDED) {
123 size32 = drrwe->drr_psize;
124 round = B_TRUE;
125 } else if (drr_type == DRR_BEGIN) {
126 size32 = drr->drr_payloadlen;
127 } else {
128 return (0);
129 }
130 if (size32 != 0) {
131 size = swap ? BSWAP_32(size32) : size32;
132 } else {
133 size = swap ? BSWAP_64(size64) : size64;
134 }
135 return (round ? P2ROUNDUP(size, 8) : size);
136 }
137
138 /*
139 * Must be called only with the first record in a stream. Must be a
140 * DRR_BEGIN record or we'll terminate with "invalid stream".
141 */
142 static void
set_stream_attributes(drr_packet_t * item)143 set_stream_attributes(drr_packet_t *item)
144 {
145 dmu_replay_record_t *drr = &item->dp_drr;
146 struct drr_begin *drrb = &drr->drr_u.drr_begin;
147 uint64_t magic = drrb->drr_magic;
148 uint64_t versioninfo = drrb->drr_versioninfo;
149 boolean_t i_am_big_endian = htonl(0xFF00) == 0xFF00;
150
151 boolean_t swap_on_output, is_deduped;
152
153 if (magic == BSWAP_64(DMU_BACKUP_MAGIC)) {
154 SET_ATTR(CA_BYTESWAPPED);
155 versioninfo = BSWAP_64(versioninfo);
156 } else if (magic != DMU_BACKUP_MAGIC) {
157 errx(1, "invalid ZFS stream, bad magic number %llx",
158 (u_longlong_t)magic);
159 }
160 if (i_am_big_endian == ATTR_IS_SET(CA_BYTESWAPPED)) {
161 SET_ATTR(CA_LITTLE_ENDIAN_INPUT);
162 } else {
163 SET_ATTR(CA_BIG_ENDIAN_INPUT);
164 }
165 chain_attrs->ca_feature_flags = DMU_GET_FEATUREFLAGS(versioninfo);
166
167 is_deduped =
168 STREAM_HAS_FEATURE(DMU_BACKUP_FEATURE_DEDUP) ||
169 STREAM_HAS_FEATURE(DMU_BACKUP_FEATURE_DEDUPPROPS);
170
171 if (OPTION_ENABLED(CA_FORBID_DEDUP) && is_deduped) {
172 errx(1, "input stream is deduplicated, but this subcommand "
173 "does not support deduplicated streams. Use 'zstream "
174 "redup' to reduplicate.");
175 }
176 boolean_t req_dedup = OPTION_ENABLED(CA_REQUIRE_DEDUP);
177 boolean_t is_dedup = STREAM_HAS_FEATURE(DMU_BACKUP_FEATURE_DEDUP);
178 if (req_dedup && !is_dedup) {
179 errx(1, "this subcommand requires a deduplicated input "
180 "stream, but the stream is not deduplicated");
181 }
182 boolean_t req_native = OPTION_ENABLED(CA_REQUIRE_NATIVE_ENDIAN);
183 boolean_t is_byteswapped = ATTR_IS_SET(CA_BYTESWAPPED);
184 if (req_native && is_byteswapped) {
185 errx(1, "this subcommand requires a native-endian "
186 "input stream");
187 }
188
189 /*
190 * Figure out output endianness. In the absence of explicit byte
191 * order instructions, we default to preserving the input byte
192 * order. Record headers are always converted to native byte order
193 * for processing, but they can be swapped back on output.
194 *
195 * zfs receive inspects the endianness of each DRR record
196 * and assumes, at least in some cases, that payload data has the
197 * same order as the DMU wrappers.
198 */
199 if (OPTION_ENABLED(CA_BIG_ENDIAN_OUT))
200 swap_on_output = !i_am_big_endian;
201 else if (OPTION_ENABLED(CA_LITTLE_ENDIAN_OUT))
202 swap_on_output = i_am_big_endian;
203 else if (OPTION_ENABLED(CA_OPPOSITE_ENDIAN_OUT))
204 swap_on_output = !ATTR_IS_SET(CA_BYTESWAPPED);
205 else
206 swap_on_output = ATTR_IS_SET(CA_BYTESWAPPED);
207
208 if (swap_on_output) {
209 ENABLE_OPTION(chain_attrs, CA_BYTESWAP_ON_OUTPUT);
210 }
211 }
212
213 static disposition_t
chain_read(void * item_in,void * context_in)214 chain_read(void *item_in, void *context_in)
215 {
216 drr_packet_t *item = (drr_packet_t *)item_in;
217 io_context_t *context = (io_context_t *)context_in;
218
219 if (item == NULL)
220 return (D_OK);
221
222 dmu_replay_record_t *drr = &item->dp_drr;
223
224 if (!context->ic_fp)
225 open_file(context);
226
227 if (fread(drr, sizeof (dmu_replay_record_t), 1, context->ic_fp) != 1) {
228 if (ferror(context->ic_fp)) {
229 err(1, "error reading record header at offset %llu",
230 (u_longlong_t)context->ic_offset);
231 }
232 fclose(context->ic_fp);
233 return (D_EOF);
234 }
235
236 if (context->ic_offset == 0)
237 set_stream_attributes(item);
238
239 size_t payload_size = calc_payload_size(&item->dp_drr);
240 if (payload_size > UINT32_MAX) {
241 errx(1, "stated packet size is greater than uint32_t"
242 "at offset %llu", (u_longlong_t)context->ic_offset);
243 }
244 item->dp_payload_size = payload_size;
245 if (item->dp_payload_size > 0) {
246 item->dp_payload = safe_malloc(item->dp_payload_size);
247 size_t n_read = fread(item->dp_payload, item->dp_payload_size,
248 1, context->ic_fp);
249 if (n_read != 1) {
250 if (ferror(context->ic_fp)) {
251 err(1, "error reading record payload at "
252 " offset %llu",
253 (u_longlong_t)context->ic_offset);
254 } else {
255 /*
256 * We can't exit here because the ZFS test
257 * suite depends on being able to process
258 * streams truncated at random places.
259 */
260 warnx("input ends mid-record at offset %llu "
261 "- stream is likely corrupt",
262 (u_longlong_t)context->ic_offset);
263 fclose(context->ic_fp);
264 free(item->dp_payload);
265 return (D_EOF);
266 }
267 }
268 } else {
269 item->dp_payload = NULL;
270 }
271 item->dp_stream_offset = context->ic_offset;
272
273 uint32_t drr_type = ATTR_IS_SET(CA_BYTESWAPPED) ?
274 BSWAP_32(drr->drr_type) : drr->drr_type;
275
276 if (drr_type >= DRR_NUMTYPES) {
277 err(1, "invalid record type %llu found at offset %llu",
278 (u_longlong_t)drr_type, (u_longlong_t)context->ic_offset);
279 }
280
281 context->ic_offset += sizeof (*drr) + item->dp_payload_size;
282
283 record_stats_t *stats = &chain_attrs->ca_stats_in[drr_type];
284 stats->rs_num_records++;
285 stats->rs_total_header_bytes += sizeof (dmu_replay_record_t);
286 stats->rs_total_payload_bytes += item->dp_payload_size;
287
288 stats = &chain_attrs->ca_totals_in;
289 stats->rs_num_records++;
290 stats->rs_total_header_bytes += sizeof (dmu_replay_record_t);
291 stats->rs_total_payload_bytes += item->dp_payload_size;
292
293 return (D_OK);
294 }
295
296 static disposition_t
chain_write(void * item_in,void * context_in)297 chain_write(void *item_in, void *context_in)
298 {
299 drr_packet_t *item = (drr_packet_t *)item_in;
300 io_context_t *context = (io_context_t *)context_in;
301
302 if (item == NULL) {
303 if (context->ic_fp) {
304 if (fclose(context->ic_fp) != 0)
305 err(1, "error closing output stream");
306 context->ic_fp = NULL;
307 }
308 return (D_OK);
309 }
310
311 if (!context->ic_fp) {
312 open_file(context);
313 }
314
315 dmu_replay_record_t *drr = &item->dp_drr;
316
317 if (fwrite(drr, sizeof (dmu_replay_record_t), 1, context->ic_fp) != 1) {
318 err(1, "error writing record header");
319 } else if (item->dp_payload_size > 0) {
320 size_t n_written = fwrite(item->dp_payload,
321 item->dp_payload_size, 1, context->ic_fp);
322 if (n_written != 1) {
323 err(1, "error writing payload");
324 } else {
325 free(item->dp_payload);
326 item->dp_payload = NULL;
327 }
328 }
329
330 uint32_t drr_type = OPTION_ENABLED(CA_BYTESWAP_ON_OUTPUT) ?
331 BSWAP_32(drr->drr_type) : drr->drr_type;
332
333 record_stats_t *stats = &chain_attrs->ca_stats_out[drr_type];
334 stats->rs_num_records++;
335 stats->rs_total_header_bytes += sizeof (dmu_replay_record_t);
336 stats->rs_total_payload_bytes += item->dp_payload_size;
337
338 stats = &chain_attrs->ca_totals_out;
339 stats->rs_num_records++;
340 stats->rs_total_header_bytes += sizeof (dmu_replay_record_t);
341 stats->rs_total_payload_bytes += item->dp_payload_size;
342
343 return (D_OK);
344 }
345
346 /*
347 * Even if the chain doesn't write out a stream, payloads still need freed.
348 */
349 static disposition_t
chain_null_output(void * item_in,void * context)350 chain_null_output(void *item_in, void *context)
351 {
352 (void) context;
353 drr_packet_t *item = (drr_packet_t *)item_in;
354
355 if (item && item->dp_payload != NULL && item->dp_payload_size > 0) {
356 free(item->dp_payload);
357 item->dp_payload = NULL;
358 item->dp_payload_size = 0;
359 }
360 return (D_OK);
361 }
362
363 /*
364 * Storage for the filename must remain valid during chain execution
365 */
366 static chain_step_t
setup_io(const char * filename,boolean_t for_reading)367 setup_io(const char *filename, boolean_t for_reading)
368 {
369 int context_num = next_io_context++ % MAX_IO_STREAMS;
370
371 io_context_t context = {
372 .ic_filename = filename,
373 .ic_for_reading = for_reading
374 };
375 io_contexts[context_num] = context;
376
377 chain_step_t step = {
378 .cs_type = CS_SERIAL,
379 .cs_in_size = for_reading ? 0 : sizeof (drr_packet_t),
380 .cs_out_size = for_reading ? sizeof (drr_packet_t) : 0,
381 .cs_context = &io_contexts[context_num],
382 .cs_serial = {
383 .process = for_reading ? chain_read : chain_write
384 }
385 };
386 return (step);
387 }
388
389 chain_step_t
serial_read_stream(const char * filename)390 serial_read_stream(const char *filename)
391 {
392 return (setup_io(filename, B_TRUE));
393 }
394
395 chain_step_t
serial_write_stream(const char * filename)396 serial_write_stream(const char *filename)
397 {
398 return (setup_io(filename, B_FALSE));
399 }
400
401 chain_step_t
serial_null_output(void)402 serial_null_output(void)
403 {
404 chain_step_t step = {
405 .cs_type = CS_SERIAL,
406 .cs_in_size = sizeof (drr_packet_t),
407 .cs_out_size = 0,
408 .cs_context = NULL,
409 .cs_serial = {
410 .process = chain_null_output
411 }
412 };
413 return (step);
414 }
415
416 size_t
constant_cost_of_one(queue_item_t * packet,void * context)417 constant_cost_of_one(queue_item_t *packet, void *context)
418 {
419 (void) context;
420 (void) packet;
421 return (1);
422 }
423
424 size_t
payload_size_as_cost(queue_item_t * packet_in,void * context)425 payload_size_as_cost(queue_item_t *packet_in, void *context)
426 {
427 (void) context;
428 drr_packet_t *packet = (drr_packet_t *)packet_in;
429 return (packet->dp_payload_size);
430 }
431
432 static disposition_t
chain_checkpoint(void * item_in,void * ctxt_in)433 chain_checkpoint(void *item_in, void *ctxt_in)
434 {
435 drr_packet_t *item = (drr_packet_t *)item_in;
436 checkpoint_context_t *ctxt = (checkpoint_context_t *)ctxt_in;
437
438 struct timespec now;
439 char buff[32];
440 uint64_t delta_b, dbdt;
441 double now_sec, delta_t;
442
443 if (item == NULL)
444 return (D_OK);
445
446 clock_gettime(CLOCK_MONOTONIC, &now);
447 now_sec = now.tv_sec + (double)now.tv_nsec / 1E9;
448 if (ctxt->cc_last_sec > 1E-9) {
449 delta_t = now_sec - ctxt->cc_last_sec;
450 if (delta_t < ctxt->cc_period_sec)
451 return (D_OK);
452 delta_b = item->dp_stream_offset - ctxt->cc_last_bytes;
453 dbdt = delta_b / delta_t;
454 zfs_nicenum(dbdt, buff, sizeof (buff));
455 fprintf(stderr, "Checkpoint %s: %s/s\n", ctxt->cc_name, buff);
456 }
457 ctxt->cc_last_sec = now_sec;
458 ctxt->cc_last_bytes = item->dp_stream_offset;
459 return (D_OK);
460 }
461
462 /*
463 * Storage for name must remain valid throughout chain execution
464 */
465 chain_step_t
serial_checkpoint(const char * name)466 serial_checkpoint(const char *name)
467 {
468 int context_no = next_checkpoint_context++ % MAX_IO_STREAMS;
469
470 checkpoint_context_t context = {
471 .cc_name = name,
472 .cc_period_sec = 1.0
473 };
474 checkpoint_contexts[context_no] = context;
475
476 chain_step_t step = {
477 .cs_type = CS_SERIAL,
478 .cs_in_size = sizeof (drr_packet_t),
479 .cs_out_size = sizeof (drr_packet_t),
480 .cs_context = &checkpoint_contexts[context_no],
481 .cs_serial = {
482 .process = chain_checkpoint
483 },
484 };
485 return (step);
486 }
487
488 static disposition_t
chain_drop_record_types(void * item_in,void * context_in)489 chain_drop_record_types(void *item_in, void *context_in)
490 {
491 drr_packet_t *item = (drr_packet_t *)item_in;
492 uint32_t *context = (uint32_t *)context_in;
493
494 if (item == NULL)
495 return (D_OK);
496
497 uint32_t type = (uint32_t)item->dp_drr.drr_type;
498 if (type >= DRR_NUMTYPES) {
499 errx(1, "invalid record type %u found at offset %llu "
500 "(place drop filter downstream of byteswapping?)",
501 type, (u_longlong_t)item->dp_stream_offset);
502 }
503
504 if (((UINT32_C(1) << type) & *context) != 0) {
505 if (item->dp_payload != NULL) {
506 free(item->dp_payload);
507 item->dp_payload = NULL;
508 item->dp_payload_size = 0;
509 }
510 return (D_DROP);
511 }
512 return (D_OK);
513 }
514
515 chain_step_t
serial_drop_record_types(uint32_t drop_mask)516 serial_drop_record_types(uint32_t drop_mask)
517 {
518 int context_no = next_drop_context++ % MAX_DROP_FILTERS;
519 uint32_t *context = &drop_contexts[context_no];
520
521 *context = drop_mask;
522
523 chain_step_t step = {
524 .cs_type = CS_SERIAL,
525 .cs_in_size = sizeof (drr_packet_t),
526 .cs_out_size = sizeof (drr_packet_t),
527 .cs_context = context,
528 .cs_serial = {
529 .process = chain_drop_record_types
530 },
531 };
532 return (step);
533 }
534