zstream: add "drop records" chain module

This patch adds a module to remove records from a stream according to a
user-specified record-type mask. For example:

```
zstream_chain_t drop_writes {
    ...
    serial_drop_record_types(DROP_WRITE_EMBEDDED | DROP_WRITE),
    ...
}
```

This is a trivial module, but there are a couple of reasons to use it
instead of just ignoring records you don't care about:

- It makes filtering a declarable API contract. Past the filter point,
  you do not need to double-check record types beyond those you're
  actually interested in.

- Payloads are stored in memory until they reach the end of the chain.
  If your don't-care list includes WRITEs or other potentially large
  records, you can make more efficient use of memory by releasing those
  records early.

- Records that you ignore still occupy space in parallel queues as they
  work their way down the pipeline. Each enqueue or dequeue is a
  (record-header) copy, and since queues are FIFO, unwanted records
  shorten queues' effective lengths just by occupying slots.

The common theme is that it's better to drop unwanted records early.
Having an off-the-shelf module makes that simple. By the time unwanted
records get to your "ok, I'm doing real work now" function, they've
already extracted most of their cost.

Unfortunately, we can't avoid reading or checksumming unwanted records
entirely because they're needed to validate stream checksums.

Reviewed-by: Brian Behlendorf <[email protected]>
Signed-off-by: Garth Snyder <[email protected]>
Closes #18755

cmd/zstream/zstream_io.c

@@ -57,6 +57,9 @@ static int next_io_context = 0;
 static checkpoint_context_t checkpoint_contexts[MAX_IO_STREAMS];
 static int next_checkpoint_context = 0;
 
+static uint32_t drop_contexts[MAX_DROP_FILTERS];
+static int next_drop_context = 0;
+
 /*
  * Run from within chain execution to initialize I/O. A NULL filename
  * indicates stdin or stdout.
@@ -484,3 +487,50 @@ serial_checkpoint(const char *name)
 	};
 	return (step);
 }
+
+static disposition_t
+chain_drop_record_types(void *item_in, void *context_in)
+{
+	drr_packet_t *item = (drr_packet_t *)item_in;
+	uint32_t *context = (uint32_t *)context_in;
+
+	if (item == NULL)
+		return (D_OK);
+
+	uint32_t type = (uint32_t)item->dp_drr.drr_type;
+	if (type >= DRR_NUMTYPES) {
+		errx(1, "invalid record type %u found at offset %llu "
+		    "(place drop filter downstream of byteswapping?)",
+		    type, (u_longlong_t)item->dp_stream_offset);
+	}
+
+	if (((UINT32_C(1) << type) & *context) != 0) {
+		if (item->dp_payload != NULL) {
+			free(item->dp_payload);
+			item->dp_payload = NULL;
+			item->dp_payload_size = 0;
+		}
+		return (D_DROP);
+	}
+	return (D_OK);
+}
+
+chain_step_t
+serial_drop_record_types(uint32_t drop_mask)
+{
+	int context_no = next_drop_context++ % MAX_DROP_FILTERS;
+	uint32_t *context = &drop_contexts[context_no];
+
+	*context = drop_mask;
+
+	chain_step_t step = {
+		.cs_type = CS_SERIAL,
+		.cs_in_size = sizeof (drr_packet_t),
+		.cs_out_size = sizeof (drr_packet_t),
+		.cs_context = context,
+		.cs_serial = {
+			.process = chain_drop_record_types
+		},
+	};
+	return (step);
+}

cmd/zstream/zstream_io.h

@@ -31,7 +31,24 @@ extern "C" {
 
 #include "zstream_chain.h"
 
-#define	MAX_IO_STREAMS 4
+#define	MAX_IO_STREAMS		4
+#define	MAX_DROP_FILTERS	4
+
+/*
+ * Masks for serial_drop_record_types()
+ */
+
+#define	DROP_BEGIN		(UINT32_C(1) << DRR_BEGIN)
+#define	DROP_OBJECT		(UINT32_C(1) << DRR_OBJECT)
+#define	DROP_FREEOBJECTS	(UINT32_C(1) << DRR_FREEOBJECTS)
+#define	DROP_WRITE		(UINT32_C(1) << DRR_WRITE)
+#define	DROP_FREE		(UINT32_C(1) << DRR_FREE)
+#define	DROP_END		(UINT32_C(1) << DRR_END)
+#define	DROP_WRITE_BYREF	(UINT32_C(1) << DRR_WRITE_BYREF)
+#define	DROP_SPILL		(UINT32_C(1) << DRR_SPILL)
+#define	DROP_WRITE_EMBEDDED	(UINT32_C(1) << DRR_WRITE_EMBEDDED)
+#define	DROP_OBJECT_RANGE	(UINT32_C(1) << DRR_OBJECT_RANGE)
+#define	DROP_REDACT		(UINT32_C(1) << DRR_REDACT)
 
 /*
  * The stream offset is the offset within the original source stream.
@@ -57,10 +74,30 @@ serial_read_stream(const char *filename);
 chain_step_t
 serial_write_stream(const char *filename);
 
-/* Report throughput periodically */
+/*
+ * Report throughput periodically
+ */
 chain_step_t
 serial_checkpoint(const char *name);
 
+/*
+ * Winnow the stream by dropping records of the given types. This frees up
+ * payload memory used by records you won't be inspecting. If there are
+ * parallel operations downstream of the filter, removing records allows the
+ * parallel queues to be used more efficiently.
+ *
+ * Use the DROP_* defines to construct a mask of the records you want to
+ * remove. If you want to remove most records, it's fine to pass an inverted
+ * mask formed by enumerating only the records you want to keep, e.g.:
+ *
+ * serial_drop_record_types((uint32_t)~(DROP_WRITE | DROP_WRITE_EMBEDDED))
+ *
+ * This step should be placed downstream of byteswapping, since it relies on
+ * being able to read drr->drr_type.
+ */
+chain_step_t
+serial_drop_record_types(uint32_t drop_mask);
+
 /*
  * Usually the output step is responsible for freeing payloads. Subcommands
  * that don't have stream outputs still need to free this memory. A