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_CHAIN_H 18 #define _ZSTREAM_CHAIN_H 19 20 #ifdef __cplusplus 21 extern "C" { 22 #endif 23 24 #include <stddef.h> 25 #include <stdint.h> 26 #include <sys/zfs_ioctl.h> 27 28 #include "zstream_queue.h" 29 30 /* 31 * A chain is a linear series of steps that process packets of data. It's 32 * designed to modularize common functionality, reduce code duplication, and 33 * separate processing structure from implementation. 34 * 35 * Some terms: 36 * 37 * **STEP** - A chain_step_t struct that represents a packet-processing 38 * module and any arguments or context that it needs. Chain participants 39 * generally define a function named serial_xxx() or parallel_xxx() that 40 * produces a chain_step_t that can be incorporated directly into a chain. 41 * 42 * **CHAIN** - An array of chain_step_t's. It's just data, so you can create 43 * the array however you like. But normally you'd just declare the whole 44 * chain at once, e.g.: 45 * 46 * zstream_chain_t dump_chain = { 47 * serial_read_stream(infile), 48 * parallel_calc_fletcher4(1024), 49 * serial_validate_fletcher4(), 50 * serial_byteswap(BS_INCOMING), 51 * serial_validate_records(), 52 * serial_dump_records(), 53 * serial_null_output(), 54 * chain_terminator() 55 * } 56 * 57 * Or more succinctly: 58 * 59 * zstream_chain_t dump_chain = { 60 * STANDARD_INPUT_STACK(infile), 61 * serial_dump_records(), 62 * NULL_OUTPUT_STACK() 63 * }; 64 * 65 * Chains must be terminated by a step of type CS_TERMINATE. 66 * 67 * **ITEMS** - The data packets that flow through a chain. Each step accepts 68 * items of one size and emits items of another size, which may be smaller, 69 * larger, or the same size. Items will generally be structs that start with 70 * a drr_packet_t (defined in zstream_io.h) and may include additional 71 * module-specific fields. 72 * 73 * **PROCESSING FUNCTION** - Each step names a processing function that does 74 * the actual work of transforming an input buffer into an output buffer. 75 * The transformation happens in place, in a single buffer provided by the 76 * chain. (Steps that run in parallel must also identify a cost function; see 77 * zstream_queue.h.) 78 * 79 * The processing function for a serial step should return a disposition_t, 80 * normally D_OK. A processing function can return D_DROP to remove an item 81 * from the stream entirely. It can also return D_EOF to indicate that no 82 * more data will be forthcoming. However, only the first step in the chain 83 * should ever return D_EOF. 84 * 85 * Serial functions are called with a NULL packet when the end of the 86 * stream passes by them. Since parallel functions may see packets in any 87 * order, they have no concept of "end of stream" and do not receive this 88 * notification. 89 * 90 * **CONTEXT** - An arbitrary void * that the chain passes along to the 91 * processing function as an argument. 92 * 93 * **CHAIN ATTRIBUTES** - A global set of flags available to all steps. The 94 * chain is also responsible for tracking general statistics such as the 95 * number of records of each type that have been processed. 96 */ 97 98 #define CA_BYTESWAPPED (1ULL << 0) /* ca_attrs */ 99 #define CA_BIG_ENDIAN_INPUT (1ULL << 1) 100 #define CA_LITTLE_ENDIAN_INPUT (1ULL << 2) 101 102 #define CA_VERBOSE (1ULL << 0) /* ca_command_opts */ 103 #define CA_DUMP_BEGIN_AND_END (1ULL << 1) 104 #define CA_DUMP_ALL_RECORDS (1ULL << 2) 105 #define CA_DUMP_CHECKSUMS (1ULL << 3) 106 #define CA_DUMP_DATA (1ULL << 4) 107 #define CA_IGNORE_CKSUMS (1ULL << 5) 108 #define CA_DO_NOT_VALIDATE (1ULL << 6) 109 #define CA_FORBID_DEDUP (1ULL << 7) 110 #define CA_REQUIRE_DEDUP (1ULL << 8) 111 #define CA_REQUIRE_NATIVE_ENDIAN (1ULL << 9) 112 #define CA_BYTESWAP_ON_OUTPUT (1ULL << 10) 113 #define CA_BIG_ENDIAN_OUT (1ULL << 11) 114 #define CA_LITTLE_ENDIAN_OUT (1ULL << 12) 115 #define CA_OPPOSITE_ENDIAN_OUT (1ULL << 13) 116 117 #define OPTION_ENABLED(option) (!!(chain_attrs->ca_command_opts & (option))) 118 #define STREAM_HAS_FEATURE(feat) (!!(chain_attrs->ca_feature_flags & (feat))) 119 #define ATTR_IS_SET(attr) (!!(chain_attrs->ca_attrs & (attr))) 120 121 #define ENABLE_OPTION(attrs, opt) ((attrs)->ca_command_opts |= (opt)) 122 #define SET_ATTR(attr) (chain_attrs->ca_attrs |= (attr)) 123 124 typedef struct { 125 uint64_t rs_num_records; 126 uint64_t rs_total_header_bytes; 127 uint64_t rs_total_payload_bytes; 128 } record_stats_t; 129 130 /* 131 * Chain attribute flags that describe the stream. Statistics are maintained 132 * by the zstream_io modules. 133 */ 134 typedef struct { 135 uint64_t ca_feature_flags; /* From drr_versioninfo */ 136 uint64_t ca_attrs; /* Discovered attributes */ 137 uint64_t ca_command_opts; /* Command line options */ 138 record_stats_t ca_totals_in; 139 record_stats_t ca_totals_out; 140 record_stats_t ca_stats_in[DRR_NUMTYPES]; 141 record_stats_t ca_stats_out[DRR_NUMTYPES]; 142 } chain_attrs_t; 143 144 typedef enum { CS_SERIAL, CS_PARALLEL, CS_TERMINATE } step_type_t; 145 typedef enum { D_OK, D_EOF, D_DROP } disposition_t; 146 147 typedef disposition_t 148 zc_serial_process_f(void *item, void *context); 149 150 typedef struct chain_step 151 { 152 step_type_t cs_type; 153 size_t cs_in_size; 154 size_t cs_out_size; 155 void *cs_context; 156 union { 157 struct { 158 zc_serial_process_f *process; /* serial */ 159 } cs_serial; 160 struct { 161 zq_estimate_cost_f *cost; 162 zq_process_item_f *process; 163 } cs_parallel; 164 }; 165 } chain_step_t; 166 167 typedef chain_step_t zstream_chain_t[]; 168 169 /* 170 * Chain attributes accessible to any step on the chain. These are normally 171 * accessed through the macros defined above. 172 */ 173 extern chain_attrs_t *chain_attrs; 174 175 /* 176 * Execute a chain. Returns once execution is complete. You can pass NULL 177 * for the attrs if you're not interested in preserving them after the chain 178 * has run. (The chain will allocate and dispose of a buffer for them.) 179 */ 180 void 181 zstream_chain_exec(zstream_chain_t chain, chain_attrs_t *attrs); 182 183 /* 184 * Execute a chain linearly, without queues and without multithreading. This 185 * form of execution is intended as a debugging aid, both for clients and 186 * for the chain mechanism itself. If this variant doesn't produce results 187 * identical to zstream_chain_exec(), there's a multithreading-related bug 188 * somewhere. 189 * 190 * It is not necessary to remove parallel steps from the input chain. They 191 * are accepted as-is, but their execution won't be parallelized. 192 */ 193 void 194 zstream_chain_exec_serialized(zstream_chain_t chain, chain_attrs_t *attrs); 195 196 chain_step_t 197 chain_terminator(void); 198 199 #ifdef __cplusplus 200 } 201 #endif 202 203 #endif /* _ZSTREAM_CHAIN_H */ 204