From f8bec6413b739f8a8d18ae20de9e329c09a86f16 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Mon, 8 Jun 2026 17:56:10 +0300 Subject: [PATCH 1/7] test(decode): ring-vs-flat decode diagnostic harness --- zstd/examples/ring_vs_flat.rs | 122 ++++++++++++++++++++++++++++++++++ 1 file changed, 122 insertions(+) create mode 100644 zstd/examples/ring_vs_flat.rs diff --git a/zstd/examples/ring_vs_flat.rs b/zstd/examples/ring_vs_flat.rs new file mode 100644 index 000000000..7130b93a4 --- /dev/null +++ b/zstd/examples/ring_vs_flat.rs @@ -0,0 +1,122 @@ +//! Diagnostic: isolate the RingBuffer (streaming) decode cost vs the +//! FlatBuf/UserSlice (slice) decode cost on identical compressed bytes. +//! +//! The streaming production path (`StreamingDecoder` -> `RingBuffer`) keeps +//! at most `window_size` live bytes, so on inputs larger than the window the +//! ring is perpetually wrapped (`head > tail`). `RingBuffer::inline_exec_ok` +//! vetoes the donor inline match-copy on a wrapped ring, so the whole frame +//! tail runs the slow `push`/`repeat` fallback. The flat path never wraps and +//! always takes the fast inline exec. This harness measures the gap and gives +//! a C one-shot reference. +//! +//! Run on the i9 (x86 AVX2): `cargo run --release --example ring_vs_flat`. + +use std::hint::black_box; +use std::io::Read; +use std::time::Instant; + +use structured_zstd::decoding::{FrameDecoder, StreamingDecoder}; +use structured_zstd::encoding::{CompressionLevel, compress_slice_to_vec}; + +fn repeated_pattern_bytes(len: usize) -> Vec { + let pattern = b"coordinode:segment:0001|tenant=demo|label=orders|"; + let mut bytes = Vec::with_capacity(len); + while bytes.len() < len { + let remaining = len - bytes.len(); + bytes.extend_from_slice(&pattern[..pattern.len().min(remaining)]); + } + bytes +} + +fn repeated_log_lines(len: usize) -> Vec { + const LINES: &[&str] = &[ + "ts=2026-03-26T21:39:28Z level=INFO msg=\"flush memtable\" tenant=demo table=orders region=eu-west\n", + "ts=2026-03-26T21:39:29Z level=INFO msg=\"rotate segment\" tenant=demo table=orders region=eu-west\n", + "ts=2026-03-26T21:39:30Z level=INFO msg=\"compact level\" tenant=demo table=orders region=eu-west\n", + "ts=2026-03-26T21:39:31Z level=INFO msg=\"write block\" tenant=demo table=orders region=eu-west\n", + ]; + let mut bytes = Vec::with_capacity(len); + while bytes.len() < len { + for line in LINES { + if bytes.len() == len { + break; + } + let remaining = len - bytes.len(); + bytes.extend_from_slice(&line.as_bytes()[..line.len().min(remaining)]); + } + } + bytes +} + +fn time_flat(compressed: &[u8], expected: usize, iters: u32) -> f64 { + let mut target = vec![0u8; expected + structured_zstd::WILDCOPY_OVERLENGTH]; + let mut decoder = FrameDecoder::new(); + // warm + let w = decoder.decode_all(compressed, &mut target).unwrap(); + assert_eq!(w, expected); + let start = Instant::now(); + for _ in 0..iters { + let written = decoder + .decode_all(black_box(compressed), &mut target) + .unwrap(); + black_box(&target[..written]); + } + start.elapsed().as_secs_f64() * 1e3 / f64::from(iters) +} + +fn time_ring(compressed: &[u8], expected: usize, iters: u32) -> f64 { + let mut out = Vec::with_capacity(expected); + // warm + { + let mut dec = StreamingDecoder::new(compressed).unwrap(); + out.clear(); + dec.read_to_end(&mut out).unwrap(); + assert_eq!(out.len(), expected); + } + let start = Instant::now(); + for _ in 0..iters { + let mut dec = StreamingDecoder::new(black_box(compressed)).unwrap(); + out.clear(); + dec.read_to_end(&mut out).unwrap(); + black_box(&out[..]); + } + start.elapsed().as_secs_f64() * 1e3 / f64::from(iters) +} + +fn time_c_oneshot(compressed: &[u8], expected: usize, iters: u32) -> f64 { + // warm + let w = zstd::stream::decode_all(compressed).unwrap(); + assert_eq!(w.len(), expected); + let start = Instant::now(); + for _ in 0..iters { + let v = zstd::stream::decode_all(black_box(compressed)).unwrap(); + black_box(&v[..]); + } + start.elapsed().as_secs_f64() * 1e3 / f64::from(iters) +} + +fn run(name: &str, raw: &[u8], iters: u32) { + let compressed = compress_slice_to_vec(raw, CompressionLevel::Level(3)); + let expected = raw.len(); + let flat = time_flat(&compressed, expected, iters); + let ring = time_ring(&compressed, expected, iters); + let c = time_c_oneshot(&compressed, expected, iters); + println!( + "{name:>20} raw={:>8} comp={:>8} flat={flat:7.3}ms ring={ring:7.3}ms c={c:7.3}ms ring/flat={:.2}x ring/c={:.2}x flat/c={:.2}x", + expected, + compressed.len(), + ring / flat, + ring / c, + flat / c, + ); +} + +fn main() { + let mb = 1024 * 1024; + let iters = 50; + println!("== RingBuffer (stream) vs FlatBuf (slice) vs C one-shot, dfast level 3 =="); + run("low-entropy-1m", &repeated_pattern_bytes(mb), iters); + run("low-entropy-4m", &repeated_pattern_bytes(4 * mb), iters); + run("large-log-1m", &repeated_log_lines(mb), iters); + run("large-log-4m", &repeated_log_lines(4 * mb), iters); +} From e985de88ee5f61587ffd7b3cdaf157c86f1a0a34 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Mon, 8 Jun 2026 18:07:43 +0300 Subject: [PATCH 2/7] perf(decode): allow inline match-copy on contiguous wrapped ring RingBuffer::inline_exec_ok vetoed the donor inline ZSTD_execSequence body whenever the ring had wrapped (head > tail). The streaming path drains to window_size after each read, so on any input larger than the window the ring is perpetually wrapped and the whole frame tail ran the slow push/repeat fallback while the flat slice path always took the fast inline copy. Relax the gate: on a wrapped ring still take the inline path when the sequence linear write + overshoot stays in the free gap before head and the match source is the contiguous lower live segment (offset <= tail + lit). The match offset is now passed into inline_exec_ok so the ring can verify source contiguity; linear backends ignore it. The three RingBuffer exec_sequence_inline bodies drop their head<=tail assumption (which underflowed live_len on a wrapped ring) for a wrap-aware match-source assert. --- zstd/examples/ring_vs_flat.rs | 58 ++++++-- zstd/src/decoding/buffer_backend.rs | 13 +- zstd/src/decoding/ringbuffer.rs | 140 +++++++++++++----- zstd/src/decoding/seq_decoder_avx2.rs | 8 +- zstd/src/decoding/seq_decoder_bmi2.rs | 8 +- zstd/src/decoding/seq_decoder_vbmi2.rs | 8 +- zstd/src/decoding/sequence_section_decoder.rs | 16 +- 7 files changed, 184 insertions(+), 67 deletions(-) diff --git a/zstd/examples/ring_vs_flat.rs b/zstd/examples/ring_vs_flat.rs index 7130b93a4..02a93e7da 100644 --- a/zstd/examples/ring_vs_flat.rs +++ b/zstd/examples/ring_vs_flat.rs @@ -16,7 +16,9 @@ use std::io::Read; use std::time::Instant; use structured_zstd::decoding::{FrameDecoder, StreamingDecoder}; -use structured_zstd::encoding::{CompressionLevel, compress_slice_to_vec}; +use structured_zstd::encoding::{ + CompressionLevel, CompressionParameters, compress_with_parameters, +}; fn repeated_pattern_bytes(len: usize) -> Vec { let pattern = b"coordinode:segment:0001|tenant=demo|label=orders|"; @@ -95,28 +97,60 @@ fn time_c_oneshot(compressed: &[u8], expected: usize, iters: u32) -> f64 { start.elapsed().as_secs_f64() * 1e3 / f64::from(iters) } -fn run(name: &str, raw: &[u8], iters: u32) { - let compressed = compress_slice_to_vec(raw, CompressionLevel::Level(3)); +/// C streaming decode (ZSTD_decompressStream via zstd::stream::read::Decoder): +/// the apples-to-apples peer to our `StreamingDecoder` ring path — C also keeps +/// a window buffer and flushes (double-copies) into `out`. +fn time_c_stream(compressed: &[u8], expected: usize, iters: u32) -> f64 { + let mut out = Vec::with_capacity(expected); + { + let mut dec = zstd::stream::read::Decoder::new(compressed).unwrap(); + out.clear(); + dec.read_to_end(&mut out).unwrap(); + assert_eq!(out.len(), expected); + } + let start = Instant::now(); + for _ in 0..iters { + let mut dec = zstd::stream::read::Decoder::new(black_box(compressed)).unwrap(); + out.clear(); + dec.read_to_end(&mut out).unwrap(); + black_box(&out[..]); + } + start.elapsed().as_secs_f64() * 1e3 / f64::from(iters) +} + +fn run(name: &str, raw: &[u8], window_log: u32, iters: u32) { + // Compress with an explicit small window so the streaming RingBuffer + // genuinely cycles (window << src), the realistic streaming case the + // dashboard's large-log-stream/low-entropy-1m hit. With src == window the + // ring is near-full in steady state (no gap) and the inline path can never + // fire regardless of the gate, so that degenerate case measures nothing. + let params = CompressionParameters::builder(CompressionLevel::Level(3)) + .window_log(window_log) + .build() + .unwrap(); + let compressed = compress_with_parameters(raw, ¶ms); let expected = raw.len(); let flat = time_flat(&compressed, expected, iters); let ring = time_ring(&compressed, expected, iters); let c = time_c_oneshot(&compressed, expected, iters); + let cs = time_c_stream(&compressed, expected, iters); println!( - "{name:>20} raw={:>8} comp={:>8} flat={flat:7.3}ms ring={ring:7.3}ms c={c:7.3}ms ring/flat={:.2}x ring/c={:.2}x flat/c={:.2}x", + "{name:>16} wlog={window_log} raw={:>9} flat={flat:7.3} ring={ring:7.3} c1={c:7.3} cstream={cs:7.3} ms ring/flat={:.2}x ring/cstream={:.2}x ring/c1={:.2}x", expected, - compressed.len(), ring / flat, + ring / cs, ring / c, - flat / c, ); } fn main() { let mb = 1024 * 1024; - let iters = 50; - println!("== RingBuffer (stream) vs FlatBuf (slice) vs C one-shot, dfast level 3 =="); - run("low-entropy-1m", &repeated_pattern_bytes(mb), iters); - run("low-entropy-4m", &repeated_pattern_bytes(4 * mb), iters); - run("large-log-1m", &repeated_log_lines(mb), iters); - run("large-log-4m", &repeated_log_lines(4 * mb), iters); + println!( + "== RingBuffer (stream) vs FlatBuf (slice) vs C one-shot, dfast level 3, small window ==" + ); + // window_log 18 = 256 KiB window; src 4/16 MiB => the ring cycles many times. + run("low-entropy-4m", &repeated_pattern_bytes(4 * mb), 18, 30); + run("low-entropy-16m", &repeated_pattern_bytes(16 * mb), 18, 15); + run("large-log-4m", &repeated_log_lines(4 * mb), 18, 30); + run("large-log-16m", &repeated_log_lines(16 * mb), 18, 15); } diff --git a/zstd/src/decoding/buffer_backend.rs b/zstd/src/decoding/buffer_backend.rs index 41525f01a..08ea128f0 100644 --- a/zstd/src/decoding/buffer_backend.rs +++ b/zstd/src/decoding/buffer_backend.rs @@ -284,12 +284,17 @@ pub(crate) trait BufferBackend: Sized { /// Linear backends (`FlatBuf`, `UserSliceBackend`) are always contiguous, /// so the default returns `true` and their `sequence_output_fits` / /// tight-tail / grow handling covers capacity. `RingBuffer` overrides this - /// to reject the cases where the live region has wrapped or the write - /// would cross `cap`; the caller then takes the wrap-correct cold - /// `push` / `repeat` path instead. + /// to reject only the cases where this specific sequence's linear write or + /// its match source would cross the wrap boundary; a wrapped ring whose + /// write stays in the contiguous free gap before `head` and whose match + /// source is the contiguous lower live segment still takes the fast inline + /// path. The caller falls back to the wrap-correct cold `push` / `repeat` + /// path only when this returns `false`. `offset` is the resolved match + /// offset (post-repcode), needed by the ring to verify the match source is + /// contiguous; linear backends ignore it. #[allow(unused_variables)] #[inline(always)] - fn inline_exec_ok(&self, lit_length: usize, match_length: usize) -> bool { + fn inline_exec_ok(&self, lit_length: usize, match_length: usize, offset: usize) -> bool { true } diff --git a/zstd/src/decoding/ringbuffer.rs b/zstd/src/decoding/ringbuffer.rs index 550a3227f..fc931a647 100644 --- a/zstd/src/decoding/ringbuffer.rs +++ b/zstd/src/decoding/ringbuffer.rs @@ -916,26 +916,49 @@ impl super::buffer_backend::BufferBackend for RingBuffer { const SUPPORTS_INLINE_SEQUENCE_EXEC: bool = true; #[inline(always)] - fn inline_exec_ok(&self, lit_length: usize, match_length: usize) -> bool { + fn inline_exec_ok(&self, lit_length: usize, match_length: usize, offset: usize) -> bool { // The inline donor wildcopy addresses the ring linearly from `tail` - // (match source at `tail + lit_length - offset`) with up to 31 bytes of - // AVX2 wildcopy overshoot. Sound only when the live region has not - // wrapped (`head <= tail`) AND the write plus overshoot stay strictly - // below `cap`: `< cap` keeps ring invariant 4 (`tail != cap`) and puts - // the overshoot inside the trailing WILDCOPY_OVERLENGTH slack. The - // caller's `offset <= live + lit_length` invariant then places the - // match source at `>= head >= 0`, so it is contiguous and in-bounds. + // (literals at `[tail, tail+lit)`, match at `[tail+lit, tail+lit+ml)`, + // match source at `tail + lit - offset`) with up to 31 bytes of AVX2 + // wildcopy overshoot. It is sound whenever that whole span is one + // contiguous in-bounds run — which holds in two cases: + // + // * **Unwrapped** (`head <= tail`): the free region runs `[tail, cap)`. + // The write + overshoot must stay strictly below `cap` (`< cap` keeps + // ring invariant 4, `tail != cap`, and puts the overshoot inside the + // trailing WILDCOPY_OVERLENGTH slack). The caller's + // `offset <= live + lit` invariant then puts the match source at + // `>= head >= 0`, contiguous and in-bounds. + // + // * **Wrapped** (`head > tail`): the free region is the gap + // `[tail, head)` and live data is split (`[head, cap)` + `[0, tail)`). + // The donor body can still run linearly from `tail` when (a) the write + // + overshoot ends strictly before `head` (so it neither wraps nor + // clobbers the upper live segment) and (b) the match source does not + // underflow into that upper segment: `offset <= tail + lit` keeps + // `tail + lit - offset >= 0`, placing the source in the contiguous + // lower live segment `[0, tail)`. Sequences violating either bound + // (a far-back match across the wrap, or a write that reaches `head`) + // fall back to the wrap-correct `push` / `repeat` path. This is the + // subset the donor handles with its fast `ZSTD_execSequence` body; + // only its `execSequenceEnd` near the buffer boundary is the + // equivalent of our fallback. const INLINE_EXEC_MAX_OVERSHOOT: usize = 31; - if self.head > self.tail { - return false; - } - // Physical fit: write + AVX2 overshoot stays strictly below `cap`. - let physical_fit = self + let Some(end) = self .tail .checked_add(lit_length) .and_then(|v| v.checked_add(match_length)) .and_then(|v| v.checked_add(INLINE_EXEC_MAX_OVERSHOOT)) - .is_some_and(|end| end < self.cap); + else { + return false; + }; + let physical_fit = if self.head <= self.tail { + end < self.cap + } else { + // Wrapped: write + overshoot stays in the free gap before `head`, + // and the match source stays in the contiguous lower segment. + end < self.head && offset <= self.tail + lit_length + }; if !physical_fit { return false; } @@ -944,11 +967,11 @@ impl super::buffer_backend::BufferBackend for RingBuffer { // on its growth path (the decompression-bomb guard armed by // `set_block_output_ceiling`). Without this, a reused large-window // ring with physical slack could emit past `len_at_block_start + - // MAX_BLOCK_SIZE`. `head <= tail` (just checked) and `physical_fit` - // bound `tail + lit + match < cap`, so `new_len` is wrap-free and - // cannot overflow. `max_capacity == usize::MAX` between blocks makes - // this a no-op for unbounded callers. - let new_len = (self.tail - self.head) + lit_length + match_length; + // MAX_BLOCK_SIZE`. `physical_fit` bounds the write within the current + // allocation in both branches, so `new_len` cannot overflow. + // `max_capacity == usize::MAX` between blocks makes this a no-op for + // unbounded callers. + let new_len = self.len() + lit_length + match_length; new_len <= self.max_capacity } @@ -994,11 +1017,18 @@ impl super::buffer_backend::BufferBackend for RingBuffer { sequence_output_fits(lit_length, match_length, tail, cap, MAX_WILDCOPY_OVERSHOOT)?; debug_assert!(offset >= 1); debug_assert!(match_length >= 1); - debug_assert!(self.head <= tail, "exec_sequence_inline on a wrapped ring"); - let live_len = tail - self.head; + // `inline_exec_ok` admits both the unwrapped run (`head <= tail`) and a + // wrapped ring whose write stays in the gap before `head` and whose + // match source is the contiguous lower segment (`offset <= tail+lit`). + // The match-source contiguity bound differs per case; assert the one + // that applies so a future caller bypassing the gate is caught. debug_assert!( - live_len + lit_length >= offset, - "RingBuffer::exec_sequence_inline: offset {offset} exceeds live window", + if self.head <= tail { + (tail - self.head) + lit_length >= offset + } else { + offset <= tail + lit_length + }, + "RingBuffer::exec_sequence_inline: match source outside contiguous live region", ); unsafe { @@ -1046,11 +1076,15 @@ impl super::buffer_backend::BufferBackend for RingBuffer { sequence_output_fits(lit_length, match_length, tail, cap, MAX_WILDCOPY_OVERSHOOT)?; debug_assert!(offset >= 1); debug_assert!(match_length >= 1); - debug_assert!(self.head <= tail, "exec_sequence_inline on a wrapped ring"); - let live_len = tail - self.head; + // See the x86 arm: gate admits the unwrapped run and the contiguous + // wrapped subset; assert the match-source bound that applies. debug_assert!( - live_len + lit_length >= offset, - "RingBuffer::exec_sequence_inline: offset {offset} exceeds live window", + if self.head <= tail { + (tail - self.head) + lit_length >= offset + } else { + offset <= tail + lit_length + }, + "RingBuffer::exec_sequence_inline: match source outside contiguous live region", ); unsafe { @@ -1102,14 +1136,15 @@ impl super::buffer_backend::BufferBackend for RingBuffer { sequence_output_fits(lit_length, match_length, tail, cap, MAX_WILDCOPY_OVERSHOOT)?; debug_assert!(offset >= 1); debug_assert!(match_length >= 1); + // See the non-avx2 arm: gate admits the unwrapped run and the + // contiguous wrapped subset; assert the match-source bound that applies. debug_assert!( - self.head <= tail, - "exec_sequence_inline_avx2 on a wrapped ring" - ); - let live_len = tail - self.head; - debug_assert!( - live_len + lit_length >= offset, - "RingBuffer::exec_sequence_inline_avx2: offset {offset} exceeds live window", + if self.head <= tail { + (tail - self.head) + lit_length >= offset + } else { + offset <= tail + lit_length + }, + "RingBuffer::exec_sequence_inline_avx2: match source outside contiguous live region", ); unsafe { @@ -1548,16 +1583,49 @@ mod tests { // lit+match = 500 exceeds the 100-byte budget but fits physically // (1531 < cap): the inline gate must reject it. assert!( - !rb.inline_exec_ok(500, 0), + !rb.inline_exec_ok(500, 0, 1), "inline_exec_ok must reject a write past the per-block output ceiling" ); // A write within the budget stays eligible for the inline path. assert!( - rb.inline_exec_ok(50, 0), + rb.inline_exec_ok(50, 0, 1), "inline_exec_ok must allow a write within the per-block ceiling" ); } + #[test] + fn inline_exec_ok_admits_contiguous_wrapped_sequence() { + // After the ring wraps (`head > tail`), the inline donor path stays + // eligible for a sequence whose linear write fits in the free gap + // before `head` AND whose match source is the contiguous lower live + // segment (`offset <= tail + lit`). A far-back match (source across the + // wrap) or a write that would reach `head` must still be vetoed. + use super::super::buffer_backend::BufferBackend; + let mut rb = RingBuffer::new(); + rb.reserve(4096); + let cap = rb.cap; + // Force a wrapped layout: head well ahead of a small tail. + rb.head = cap - 64; + rb.tail = 32; + // Free gap is [32, cap-64): write 16 lit + 16 match + 31 overshoot = 63 + // ends at 95, far below head, and offset 8 <= tail+lit = 48 -> eligible. + assert!( + rb.inline_exec_ok(16, 16, 8), + "wrapped ring with contiguous write + in-segment source must stay inline-eligible" + ); + // Match source crosses the wrap (offset 100 > tail+lit = 48) -> veto. + assert!( + !rb.inline_exec_ok(16, 16, 100), + "wrapped ring with match source across the wrap must veto the inline path" + ); + // Write + overshoot would reach `head` (huge match) -> veto. + let huge_match = cap; // tail + lit + huge_match overflows past head + assert!( + !rb.inline_exec_ok(16, huge_match, 8), + "wrapped ring whose write would reach the upper live segment must veto" + ); + } + #[test] fn smoke() { let mut rb = RingBuffer::new(); diff --git a/zstd/src/decoding/seq_decoder_avx2.rs b/zstd/src/decoding/seq_decoder_avx2.rs index 75d253931..fe114cb93 100644 --- a/zstd/src/decoding/seq_decoder_avx2.rs +++ b/zstd/src/decoding/seq_decoder_avx2.rs @@ -129,9 +129,11 @@ macro_rules! execute_one_body { // backend (RingBuffer) veto the inline path when the live region is // not contiguous at `tail`; linear backends fold it to `true`. let inline_path_safe = B::SUPPORTS_INLINE_SEQUENCE_EXEC - && $buffer - .buffer_mut() - .inline_exec_ok(seq_ll_v as usize, seq_ml_v as usize) + && $buffer.buffer_mut().inline_exec_ok( + seq_ll_v as usize, + seq_ml_v as usize, + resolved_offset_v as usize, + ) && lit_cur_before .checked_add(16) .is_some_and(|b| b <= literals_buffer_len_v) diff --git a/zstd/src/decoding/seq_decoder_bmi2.rs b/zstd/src/decoding/seq_decoder_bmi2.rs index a70216fde..858cca7a1 100644 --- a/zstd/src/decoding/seq_decoder_bmi2.rs +++ b/zstd/src/decoding/seq_decoder_bmi2.rs @@ -106,9 +106,11 @@ macro_rules! execute_one_body { } let inline_path_safe = B::SUPPORTS_INLINE_SEQUENCE_EXEC - && $buffer - .buffer_mut() - .inline_exec_ok(seq_ll_v as usize, seq_ml_v as usize) + && $buffer.buffer_mut().inline_exec_ok( + seq_ll_v as usize, + seq_ml_v as usize, + resolved_offset_v as usize, + ) && lit_cur_before .checked_add(16) .is_some_and(|b| b <= literals_buffer_len_v) diff --git a/zstd/src/decoding/seq_decoder_vbmi2.rs b/zstd/src/decoding/seq_decoder_vbmi2.rs index 0e762f7fe..c04546fe5 100644 --- a/zstd/src/decoding/seq_decoder_vbmi2.rs +++ b/zstd/src/decoding/seq_decoder_vbmi2.rs @@ -102,9 +102,11 @@ macro_rules! execute_one_body { } let inline_path_safe = B::SUPPORTS_INLINE_SEQUENCE_EXEC - && $buffer - .buffer_mut() - .inline_exec_ok(seq_ll_v as usize, seq_ml_v as usize) + && $buffer.buffer_mut().inline_exec_ok( + seq_ll_v as usize, + seq_ml_v as usize, + resolved_offset_v as usize, + ) && lit_cur_before .checked_add(16) .is_some_and(|b| b <= literals_buffer_len_v) diff --git a/zstd/src/decoding/sequence_section_decoder.rs b/zstd/src/decoding/sequence_section_decoder.rs index b3bac73c7..7f362ea97 100644 --- a/zstd/src/decoding/sequence_section_decoder.rs +++ b/zstd/src/decoding/sequence_section_decoder.rs @@ -1023,9 +1023,11 @@ pub(crate) fn execute_one_sequence_pipelined Date: Mon, 8 Jun 2026 18:12:18 +0300 Subject: [PATCH 3/7] test(decode): ring-only streaming decode loop for profiling --- zstd/examples/decode_ring_loop.rs | 56 +++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100644 zstd/examples/decode_ring_loop.rs diff --git a/zstd/examples/decode_ring_loop.rs b/zstd/examples/decode_ring_loop.rs new file mode 100644 index 000000000..42991e138 --- /dev/null +++ b/zstd/examples/decode_ring_loop.rs @@ -0,0 +1,56 @@ +//! Tight RingBuffer (streaming) decode loop for perf-record/flamegraph on the +//! i9. Decodes a 16 MiB low-entropy frame with a small (256 KiB) window via +//! `StreamingDecoder` (the wrapped-ring path) repeatedly, so the profile is +//! dominated by the ring decode hot path + drain, not setup. +//! +//! `cargo flamegraph --example decode_ring_loop --features dict_builder` +//! or `perf record -g -- target/release/examples/decode_ring_loop`. + +use std::io::Read; + +use structured_zstd::decoding::StreamingDecoder; +use structured_zstd::encoding::{ + CompressionLevel, CompressionParameters, compress_with_parameters, +}; + +fn repeated_log_lines(len: usize) -> Vec { + const LINES: &[&str] = &[ + "ts=2026-03-26T21:39:28Z level=INFO msg=\"flush memtable\" tenant=demo table=orders region=eu-west\n", + "ts=2026-03-26T21:39:29Z level=INFO msg=\"rotate segment\" tenant=demo table=orders region=eu-west\n", + "ts=2026-03-26T21:39:30Z level=INFO msg=\"compact level\" tenant=demo table=orders region=eu-west\n", + "ts=2026-03-26T21:39:31Z level=INFO msg=\"write block\" tenant=demo table=orders region=eu-west\n", + ]; + let mut bytes = Vec::with_capacity(len); + while bytes.len() < len { + for line in LINES { + if bytes.len() == len { + break; + } + let remaining = len - bytes.len(); + bytes.extend_from_slice(&line.as_bytes()[..line.len().min(remaining)]); + } + } + bytes +} + +fn main() { + let raw = repeated_log_lines(16 * 1024 * 1024); + let params = CompressionParameters::builder(CompressionLevel::Level(3)) + .window_log(18) + .build() + .unwrap(); + let compressed = compress_with_parameters(&raw, ¶ms); + let expected = raw.len(); + let mut out = Vec::with_capacity(expected); + let iters: u32 = std::env::var("ITERS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(200); + for _ in 0..iters { + let mut dec = StreamingDecoder::new(compressed.as_slice()).unwrap(); + out.clear(); + dec.read_to_end(&mut out).unwrap(); + std::hint::black_box(&out[..]); + } + assert_eq!(out.len(), expected); +} From b0b5da8d0755356ed50f6a83c63bece8eb7b2578 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Mon, 8 Jun 2026 19:08:47 +0300 Subject: [PATCH 4/7] feat(checksum): add content-checksum modes and encoder toggle Decoder gains a ContentChecksum mode (None / EmitOnly / Verify, default EmitOnly) via FrameDecoder::set_content_checksum: - None skips the XXH64 pass entirely on both the direct and streaming/drain paths (the pass is a large share of decode time on checksum-flagged frames) - Verify compares the stored digest against the computed one per frame and fails with the new FrameDecoderError::ChecksumMismatch { expected, calculated } from decode_all and the streaming finish Encoder gains a binary toggle (FrameCompressor and StreamingEncoder set_content_checksum, semantics of ZSTD_c_checksumFlag, default emit). When off, frames set Content_Checksum_flag = 0 and carry no trailing digest; such frames are valid and decode in any mode. The enum, field, error variant, and setters compile without the hash feature (no-std + alloc); only the XXH64 work stays hash-gated, so Verify without hash degrades to no compute rather than a spurious error. --- zstd/src/decoding/decode_buffer.rs | 30 +++- zstd/src/decoding/errors.rs | 19 +++ zstd/src/decoding/frame_decoder.rs | 226 ++++++++++++++++++++++++- zstd/src/decoding/mod.rs | 2 +- zstd/src/decoding/streaming_decoder.rs | 10 ++ zstd/src/encoding/frame_compressor.rs | 53 ++++-- zstd/src/encoding/streaming_encoder.rs | 33 +++- 7 files changed, 354 insertions(+), 19 deletions(-) diff --git a/zstd/src/decoding/decode_buffer.rs b/zstd/src/decoding/decode_buffer.rs index 8c0f6b3ab..45629eafa 100644 --- a/zstd/src/decoding/decode_buffer.rs +++ b/zstd/src/decoding/decode_buffer.rs @@ -38,6 +38,13 @@ pub struct DecodeBuffer { total_output_counter: u64, #[cfg(feature = "hash")] pub hash: twox_hash::XxHash64, + /// Whether drain hashes the bytes it emits into `hash`. `false` lets a + /// `FrameDecoder` set to [`ContentChecksum::None`](crate::decoding::ContentChecksum::None) + /// skip the XXH64 pass on the streaming path. Persists across `reset` + /// (it mirrors a decoder-level setting, not per-frame state); the frame + /// layer re-applies it before each decode. + #[cfg(feature = "hash")] + compute_hash: bool, } /// Rollback token produced by [`DecodeBuffer::checkpoint`]. @@ -86,6 +93,8 @@ impl DecodeBuffer { total_output_counter: 0, #[cfg(feature = "hash")] hash: twox_hash::XxHash64::with_seed(0), + #[cfg(feature = "hash")] + compute_hash: true, } } @@ -110,9 +119,20 @@ impl DecodeBuffer { total_output_counter: 0, #[cfg(feature = "hash")] hash: twox_hash::XxHash64::with_seed(0), + #[cfg(feature = "hash")] + compute_hash: true, } } + /// Enable or disable the drain-time XXH64 pass. Set by the frame layer + /// from the decoder's [`ContentChecksum`](crate::decoding::ContentChecksum) + /// mode before each decode (`false` for `None`). + #[cfg(feature = "hash")] + #[inline] + pub(crate) fn set_compute_hash(&mut self, compute: bool) { + self.compute_hash = compute; + } + /// Arm the per-block decompressed-output ceiling for the block about to /// be decoded: the growable backend is told it may grow only up to /// `current_len + max_block_output`, so a match that would push this @@ -963,7 +983,7 @@ impl DecodeBuffer { pub fn drain(&mut self) -> Vec { let (slice1, slice2) = self.buffer.as_slices(); #[cfg(feature = "hash")] - { + if self.compute_hash { self.hash.write(slice1); self.hash.write(slice2); } @@ -1029,7 +1049,9 @@ impl DecodeBuffer { if n1 != 0 { let (written1, res1) = write_bytes(&slice1[..n1]); #[cfg(feature = "hash")] - self.hash.write(&slice1[..written1]); + if self.compute_hash { + self.hash.write(&slice1[..written1]); + } drain_guard.amount += written1; // Apparently this is what clippy thinks is the best way of expressing this @@ -1040,7 +1062,9 @@ impl DecodeBuffer { if written1 == n1 && n2 != 0 { let (written2, res2) = write_bytes(&slice2[..n2]); #[cfg(feature = "hash")] - self.hash.write(&slice2[..written2]); + if self.compute_hash { + self.hash.write(&slice2[..written2]); + } drain_guard.amount += written2; // Apparently this is what clippy thinks is the best way of expressing this diff --git a/zstd/src/decoding/errors.rs b/zstd/src/decoding/errors.rs index 049e2aa6f..6c18e362d 100644 --- a/zstd/src/decoding/errors.rs +++ b/zstd/src/decoding/errors.rs @@ -580,6 +580,16 @@ pub enum FrameDecoderError { declared: u64, produced: u64, }, + /// The frame carried a trailing XXH64 content checksum and the decoder + /// was set to [`ContentChecksum::Verify`](crate::decoding::ContentChecksum::Verify), + /// but the digest computed over the decompressed output did not match the + /// stored value. Indicates corruption in the compressed stream or its + /// trailing checksum. `expected` is the value read from the frame tail; + /// `calculated` is the digest the decoder computed (both low 32 bits). + ChecksumMismatch { + expected: u32, + calculated: u32, + }, DictNotProvided { dict_id: u32, }, @@ -788,6 +798,15 @@ impl core::fmt::Display for FrameDecoderError { "Frame content size mismatch (corrupt frame): declared {declared} bytes, blocks summed to {produced} bytes" ) } + FrameDecoderError::ChecksumMismatch { + expected, + calculated, + } => { + write!( + f, + "Content checksum mismatch (corrupt frame): frame stored 0x{expected:08X}, decoder calculated 0x{calculated:08X}" + ) + } FrameDecoderError::DictNotProvided { dict_id } => { write!( f, diff --git a/zstd/src/decoding/frame_decoder.rs b/zstd/src/decoding/frame_decoder.rs index 6e71e7aa5..46316e480 100644 --- a/zstd/src/decoding/frame_decoder.rs +++ b/zstd/src/decoding/frame_decoder.rs @@ -161,6 +161,10 @@ pub struct FrameDecoder { /// expect frames without the 4-byte magic number prefix. /// Default false (standard zstd format). magicless: bool, + /// How the optional content checksum is handled. Default + /// [`ContentChecksum::EmitOnly`] (compute + expose, no error on + /// mismatch). Set via [`Self::set_content_checksum`]. + content_checksum: ContentChecksum, /// Pinned `Dictionary_ID` expectation set via /// [`Self::expect_dict_id`]. `None` (default) disables the /// check; `Some(0)` matches frames whose header omits the @@ -195,6 +199,33 @@ pub struct FrameDecoder { computed_block_checksums: alloc::vec::Vec, } +/// How the decoder treats a frame's optional XXH64 content checksum +/// (RFC 8878 Content_Checksum_flag). The XXH64 pass over the decompressed +/// output is a measurable share of decode time, so it is made skippable. +/// +/// ``` +/// use structured_zstd::decoding::{ContentChecksum, FrameDecoder}; +/// let mut decoder = FrameDecoder::new(); +/// decoder.set_content_checksum(ContentChecksum::Verify); +/// ``` +#[derive(Copy, Clone, PartialEq, Eq, Debug, Default)] +pub enum ContentChecksum { + /// Skip the XXH64 pass entirely: no compute, no verify. + /// `get_calculated_checksum()` returns `None`. + None, + /// Compute the checksum and expose it via the accessors, but do not + /// error on a mismatch. This is the default and matches the historical + /// behaviour (callers verify manually if they wish). + #[default] + EmitOnly, + /// Compute the checksum and compare it against the frame's stored value; + /// a disagreement fails the decode with + /// [`FrameDecoderError::ChecksumMismatch`](crate::decoding::errors::FrameDecoderError::ChecksumMismatch). + /// Without the `hash` feature there is no way to compute a digest, so + /// `Verify` cannot detect a mismatch and behaves like `None`. + Verify, +} + /// Decode-relevant identity of a frame, used to reject a [`ResumeState`] /// captured from one frame being applied to a frame of a different shape. Covers /// every header field that changes how blocks decode (buffer sizing, backend @@ -672,6 +703,17 @@ impl DecoderScratchKind { Self::Flat(s) => s.buffer.hash.finish(), } } + + /// Forward the drain-time hash toggle to the inner `DecodeBuffer` + /// (streaming path). Called by the frame layer from the decoder's + /// `ContentChecksum` mode before each decode. + #[cfg(feature = "hash")] + fn set_compute_hash(&mut self, compute: bool) { + match self { + Self::Ring(s) => s.buffer.set_compute_hash(compute), + Self::Flat(s) => s.buffer.set_compute_hash(compute), + } + } } struct FrameDecoderState { @@ -836,6 +878,7 @@ impl FrameDecoder { #[cfg(not(target_has_atomic = "ptr"))] shared_dicts: (), magicless: false, + content_checksum: ContentChecksum::EmitOnly, #[cfg(feature = "lsm")] expect_dict_id: None, #[cfg(feature = "lsm")] @@ -847,6 +890,14 @@ impl FrameDecoder { } } + /// Select how the frame's optional content checksum is handled + /// (compute, expose, verify, or skip). See [`ContentChecksum`]. + /// Default [`ContentChecksum::EmitOnly`]. Takes effect on the next + /// decode; safe to call between frames on a reused decoder. + pub fn set_content_checksum(&mut self, mode: ContentChecksum) { + self.content_checksum = mode; + } + /// Opt in to per-block XXH64 verification during decode. /// Default off; zero cost when disabled. Each block's decompressed /// bytes are XXH64-hashed (low 32 bits) and appended to @@ -1283,6 +1334,11 @@ impl FrameDecoder { #[cfg(feature = "hash")] pub fn get_calculated_checksum(&self) -> Option { let state = self.state.as_ref()?; + // `ContentChecksum::None` skips the XXH64 pass entirely, so there is + // no calculated digest to report. + if self.content_checksum == ContentChecksum::None { + return None; + } if !state.frame_header.descriptor.content_checksum_flag() { return None; } @@ -1291,6 +1347,36 @@ impl FrameDecoder { Some(cksum_64bit as u32) } + /// Compare the frame's stored content checksum against the digest the + /// decoder computed, returning [`FrameDecoderError::ChecksumMismatch`] on + /// disagreement. No-op unless the mode is [`ContentChecksum::Verify`] and + /// the frame carries a trailing checksum. Call once per frame after the + /// frame is fully decoded (and, on the drain path, fully drained) so both + /// the stored value and the running digest are final. + #[cfg(feature = "hash")] + pub(crate) fn verify_content_checksum(&self) -> Result<(), FrameDecoderError> { + if self.content_checksum != ContentChecksum::Verify { + return Ok(()); + } + let Some(state) = self.state.as_ref() else { + return Ok(()); + }; + if !state.frame_header.descriptor.content_checksum_flag() { + return Ok(()); + } + let Some(expected) = state.check_sum else { + return Ok(()); + }; + let calculated = state.decoder_scratch.hash_finish() as u32; + if expected != calculated { + return Err(FrameDecoderError::ChecksumMismatch { + expected, + calculated, + }); + } + Ok(()) + } + /// Counter for how many bytes have been consumed while decoding the frame pub fn bytes_read_from_source(&self) -> u64 { let state = match &self.state { @@ -1334,7 +1420,13 @@ impl FrameDecoder { strat: BlockDecodingStrategy, ) -> Result { use FrameDecoderError as err; + // Apply the content-checksum mode to the streaming drain hash before + // any block decodes into the ring. `None` skips the XXH64 pass. + #[cfg(feature = "hash")] + let compute_hash = self.content_checksum != ContentChecksum::None; let state = self.state.as_mut().ok_or(err::NotYetInitialized)?; + #[cfg(feature = "hash")] + state.decoder_scratch.set_compute_hash(compute_hash); // Streaming entry point: pre-reserve the backing buffer to // `window_size` so multi-block frames don't pay repeated @@ -2135,6 +2227,10 @@ impl FrameDecoder { let written = self.run_direct_decode(&mut input, output, content_size)?; output = &mut output[written..]; total_bytes_written += written; + // Per-frame content-checksum verification (no-op unless the + // mode is `Verify` and the frame carries a checksum). + #[cfg(feature = "hash")] + self.verify_content_checksum()?; continue; } // Non-direct fallback: pre-reserve the backing buffer to @@ -2175,6 +2271,12 @@ impl FrameDecoder { }); } } + // Per-frame content-checksum verification on the drain path: the + // frame is fully decoded and drained here (is_finished + nothing + // left to collect), so the running digest and stored value are + // final. No-op unless the mode is `Verify`. + #[cfg(feature = "hash")] + self.verify_content_checksum()?; } Ok(total_bytes_written) @@ -2261,6 +2363,10 @@ impl FrameDecoder { let written = self.run_direct_decode(&mut input, output, content_size)?; output = &mut output[written..]; total_bytes_written += written; + // Per-frame content-checksum verification (no-op unless the + // mode is `Verify` and the frame carries a checksum). + #[cfg(feature = "hash")] + self.verify_content_checksum()?; continue; } // Non-direct fallback: pre-reserve the backing buffer to @@ -2297,6 +2403,12 @@ impl FrameDecoder { }); } } + // Per-frame content-checksum verification on the drain path: the + // frame is fully decoded and drained here (is_finished + nothing + // left to collect), so the running digest and stored value are + // final. No-op unless the mode is `Verify`. + #[cfg(feature = "hash")] + self.verify_content_checksum()?; } Ok(total_bytes_written) @@ -2616,7 +2728,9 @@ impl FrameDecoder { } } #[cfg(feature = "hash")] - if state.frame_header.descriptor.content_checksum_flag() { + if state.frame_header.descriptor.content_checksum_flag() + && self.content_checksum != ContentChecksum::None + { // Direct path bypasses the per-write hash accounting // (DecodeBuffer hashes during drain; the direct path // never drains because the user slice IS the buffer). @@ -2865,6 +2979,116 @@ mod tests { ); } + #[cfg(feature = "hash")] + #[test] + fn verify_mode_accepts_a_valid_frame() { + use crate::decoding::ContentChecksum; + let payload: Vec = (0..8192u32).map(|i| (i & 0xFF) as u8).collect(); + let mut compressor = FrameCompressor::new(CompressionLevel::Default); + compressor.set_source(payload.as_slice()); + let mut compressed = Vec::new(); + compressor.set_drain(&mut compressed); + compressor.compress(); + + let slack = super::super::buffer_backend::WILDCOPY_OVERLENGTH; + let mut dec = FrameDecoder::new(); + dec.set_content_checksum(ContentChecksum::Verify); + let mut out = alloc::vec![0u8; payload.len() + slack]; + let n = dec + .decode_all(compressed.as_slice(), &mut out) + .expect("Verify mode must accept a frame with a correct checksum"); + assert_eq!(&out[..n], payload.as_slice()); + } + + #[cfg(feature = "hash")] + #[test] + fn verify_mode_rejects_a_corrupted_checksum() { + use crate::decoding::ContentChecksum; + use crate::decoding::errors::FrameDecoderError; + let payload: Vec = (0..8192u32).map(|i| (i & 0xFF) as u8).collect(); + let mut compressor = FrameCompressor::new(CompressionLevel::Default); + compressor.set_source(payload.as_slice()); + let mut compressed = Vec::new(); + compressor.set_drain(&mut compressed); + compressor.compress(); + + // Flip a bit in the trailing 4-byte content checksum: the frame body + // still decodes to the correct bytes, but the stored digest no longer + // matches the one the decoder computes. + let last = compressed.len() - 1; + compressed[last] ^= 0xFF; + + let slack = super::super::buffer_backend::WILDCOPY_OVERLENGTH; + let mut dec = FrameDecoder::new(); + dec.set_content_checksum(ContentChecksum::Verify); + let mut out = alloc::vec![0u8; payload.len() + slack]; + let err = dec + .decode_all(compressed.as_slice(), &mut out) + .expect_err("Verify mode must reject a corrupted checksum"); + assert!( + matches!(err, FrameDecoderError::ChecksumMismatch { .. }), + "expected ChecksumMismatch, got {err:?}" + ); + } + + #[cfg(feature = "hash")] + #[test] + fn none_mode_skips_the_checksum_pass() { + use crate::decoding::ContentChecksum; + let payload: Vec = (0..8192u32).map(|i| (i & 0xFF) as u8).collect(); + let mut compressor = FrameCompressor::new(CompressionLevel::Default); + compressor.set_source(payload.as_slice()); + let mut compressed = Vec::new(); + compressor.set_drain(&mut compressed); + compressor.compress(); + + let slack = super::super::buffer_backend::WILDCOPY_OVERLENGTH; + let mut dec = FrameDecoder::new(); + dec.set_content_checksum(ContentChecksum::None); + let mut out = alloc::vec![0u8; payload.len() + slack]; + let n = dec + .decode_all(compressed.as_slice(), &mut out) + .expect("None mode must still decode correctly"); + assert_eq!(&out[..n], payload.as_slice()); + // No digest is computed in None mode, even though the frame carries one. + assert!(dec.get_checksum_from_data().is_some()); + assert!(dec.get_calculated_checksum().is_none()); + } + + #[cfg(feature = "hash")] + #[test] + fn encoder_without_checksum_emits_no_trailing_digest() { + let payload: Vec = (0..8192u32).map(|i| (i & 0xFF) as u8).collect(); + + let mut with = Vec::new(); + let mut c_with = FrameCompressor::new(CompressionLevel::Default); + c_with.set_source(payload.as_slice()); + c_with.set_drain(&mut with); + c_with.compress(); + + let mut without = Vec::new(); + let mut c_without = FrameCompressor::new(CompressionLevel::Default); + c_without.set_content_checksum(false); + c_without.set_source(payload.as_slice()); + c_without.set_drain(&mut without); + c_without.compress(); + + // The checksum-off frame is exactly the 4-byte trailing digest shorter. + assert_eq!(with.len(), without.len() + 4); + + let slack = super::super::buffer_backend::WILDCOPY_OVERLENGTH; + let mut dec = FrameDecoder::new(); + let mut out = alloc::vec![0u8; payload.len() + slack]; + let n = dec + .decode_all(without.as_slice(), &mut out) + .expect("a frame without a content checksum must decode"); + assert_eq!(&out[..n], payload.as_slice()); + assert!( + dec.get_checksum_from_data().is_none(), + "no trailing checksum should be reported" + ); + } + #[test] fn decode_all_fcs_overflow_via_corrupt_frame_returns_structured_error() { // Hand-build a corrupt frame that declares diff --git a/zstd/src/decoding/mod.rs b/zstd/src/decoding/mod.rs index 2ba31449a..392da05d3 100644 --- a/zstd/src/decoding/mod.rs +++ b/zstd/src/decoding/mod.rs @@ -34,7 +34,7 @@ mod frame_decoder; mod streaming_decoder; pub use dictionary::{Dictionary, DictionaryHandle}; -pub use frame_decoder::{BlockDecodingStrategy, FrameDecoder}; +pub use frame_decoder::{BlockDecodingStrategy, ContentChecksum, FrameDecoder}; #[cfg(feature = "lsm")] pub use frame_decoder::{PartialDecode, ResumeInput, ResumeState}; pub use streaming_decoder::StreamingDecoder; diff --git a/zstd/src/decoding/streaming_decoder.rs b/zstd/src/decoding/streaming_decoder.rs index c491080e8..5389affd6 100644 --- a/zstd/src/decoding/streaming_decoder.rs +++ b/zstd/src/decoding/streaming_decoder.rs @@ -140,6 +140,16 @@ impl> Read for StreamingDecoder Result { let decoder = self.decoder.borrow_mut(); if decoder.is_finished() && decoder.can_collect() == 0 { + // Frame fully decoded and fully drained: the running XXH64 digest + // is final, so a `Verify`-mode decoder validates the content + // checksum at this finish point. No-op in other modes. + #[cfg(feature = "hash")] + if let Err(e) = decoder.verify_content_checksum() { + #[cfg(feature = "std")] + return Err(Error::other(e)); + #[cfg(not(feature = "std"))] + return Err(Error::new(ErrorKind::Other, alloc::boxed::Box::new(e))); + } //No more bytes can ever be decoded return Ok(0); } diff --git a/zstd/src/encoding/frame_compressor.rs b/zstd/src/encoding/frame_compressor.rs index e7b18d3a2..442278483 100644 --- a/zstd/src/encoding/frame_compressor.rs +++ b/zstd/src/encoding/frame_compressor.rs @@ -102,6 +102,12 @@ pub struct FrameCompressor< /// matching format — wire-format only round-trips with a /// magicless-aware decoder. magicless: bool, + /// Whether to emit a trailing XXH64 content checksum and set the frame + /// header's `Content_Checksum_flag` (semantics of upstream + /// `ZSTD_c_checksumFlag`). Default `true`; combined with the `hash` + /// feature at frame-build time, so without `hash` no checksum is emitted + /// regardless. Set via [`Self::set_content_checksum`]. + content_checksum: bool, #[cfg(feature = "hash")] hasher: XxHash64, /// Block-layout introspection populated at the end of every @@ -599,6 +605,7 @@ impl FrameCompressor { ), }, magicless: false, + content_checksum: true, #[cfg(feature = "hash")] hasher: XxHash64::with_seed(0), #[cfg(feature = "lsm")] @@ -846,7 +853,9 @@ impl FrameCompressor { let block = &input[start..end]; let last_block = end == input.len(); #[cfg(feature = "hash")] - self.hasher.write(block); + if self.content_checksum { + self.hasher.write(block); + } crate::encoding::levels::compress_block_encoded_borrowed( &mut self.state, self.compression_level, @@ -888,6 +897,7 @@ impl FrameCompressor { }, compression_level, magicless: false, + content_checksum: true, #[cfg(feature = "hash")] hasher: XxHash64::with_seed(0), #[cfg(feature = "lsm")] @@ -912,6 +922,18 @@ impl FrameCompressor { self.magicless = magicless; } + /// Enable or disable the trailing XXH64 content checksum + /// (semantics of upstream `ZSTD_c_checksumFlag`). Default `true`. + /// + /// When `false`, emitted frames set `Content_Checksum_flag = 0` and carry + /// no trailing digest; such frames are valid (RFC 8878) and decode + /// correctly in any [`ContentChecksum`](crate::decoding::ContentChecksum) + /// mode. Without the `hash` feature no checksum is emitted regardless of + /// this setting. + pub fn set_content_checksum(&mut self, emit: bool) { + self.content_checksum = emit; + } + /// Before calling [FrameCompressor::compress] you need to set the source. /// /// This is the data that is compressed and written into the drain. @@ -1257,9 +1279,12 @@ impl FrameCompressor { last_block = false; } } - // As we read, hash that data too + // As we read, hash that data too (skipped when the content + // checksum is disabled). #[cfg(feature = "hash")] - self.hasher.write(&uncompressed_data); + if self.content_checksum { + self.hasher.write(&uncompressed_data); + } // Per-physical-block XXH64 (low 32 bits) for the optional // per-block checksum sidecar. Hashing happens INSIDE the // block emitters (RLE / Raw fast-path / Compressed / @@ -1356,7 +1381,7 @@ impl FrameCompressor { let header = FrameHeader { frame_content_size: Some(total_uncompressed), single_segment, - content_checksum: cfg!(feature = "hash"), + content_checksum: cfg!(feature = "hash") && self.content_checksum, dictionary_id: if prep.use_dictionary_state { self.dictionary.as_ref().map(|dict| dict.inner.id as u64) } else { @@ -1385,14 +1410,19 @@ impl FrameCompressor { // `self.hasher` read and the `self.compressed_data` write don't // both need `&mut self` simultaneously. #[cfg(feature = "hash")] - let checksum_bytes = (self.hasher.finish() as u32).to_le_bytes(); + let checksum_bytes = self + .content_checksum + .then(|| (self.hasher.finish() as u32).to_le_bytes()); let drain = self.compressed_data.as_mut().unwrap(); drain.write_all(&header_buf).unwrap(); drain.write_all(&all_blocks).unwrap(); - // If the `hash` feature is enabled, `content_checksum` is set in the - // header and the 32-bit digest is written at the end of the frame. + // With the `hash` feature AND the content checksum enabled, the header + // set `Content_Checksum_flag` and the 32-bit digest is written at the + // end of the frame. Disabled => no trailing bytes, flag stays 0. #[cfg(feature = "hash")] - drain.write_all(&checksum_bytes).unwrap(); + if let Some(checksum_bytes) = checksum_bytes { + drain.write_all(&checksum_bytes).unwrap(); + } #[cfg(feature = "lsm")] self.populate_frame_emit_info(header_buf.len(), &all_blocks); } @@ -1414,13 +1444,16 @@ impl FrameCompressor { prep: &FramePrep, ) { let header_buf = self.build_frame_header(total_uncompressed, prep); - let checksum_len = if cfg!(feature = "hash") { 4 } else { 0 }; + let emit_checksum = cfg!(feature = "hash") && self.content_checksum; + let checksum_len = if emit_checksum { 4 } else { 0 }; out.clear(); out.reserve(header_buf.len() + all_blocks.len() + checksum_len); out.extend_from_slice(&header_buf); out.extend_from_slice(&all_blocks); #[cfg(feature = "hash")] - out.extend_from_slice(&(self.hasher.finish() as u32).to_le_bytes()); + if self.content_checksum { + out.extend_from_slice(&(self.hasher.finish() as u32).to_le_bytes()); + } #[cfg(feature = "lsm")] self.populate_frame_emit_info(header_buf.len(), &all_blocks); } diff --git a/zstd/src/encoding/streaming_encoder.rs b/zstd/src/encoding/streaming_encoder.rs index fed8df275..150da4c26 100644 --- a/zstd/src/encoding/streaming_encoder.rs +++ b/zstd/src/encoding/streaming_encoder.rs @@ -36,6 +36,11 @@ pub struct StreamingEncoder { /// `ZSTD_f_zstd1_magicless` — omit the 4-byte magic number prefix. /// Default false. See [`Self::set_magicless`]. magicless: bool, + /// Whether to emit a trailing XXH64 content checksum and set the frame + /// header's `Content_Checksum_flag` (upstream `ZSTD_c_checksumFlag`). + /// Default `true`; combined with the `hash` feature, so without `hash` + /// no checksum is emitted regardless. See [`Self::set_content_checksum`]. + content_checksum: bool, #[cfg(feature = "hash")] hasher: XxHash64, } @@ -82,11 +87,29 @@ impl StreamingEncoder { pledged_content_size: None, bytes_consumed: 0, magicless: false, + content_checksum: true, #[cfg(feature = "hash")] hasher: XxHash64::with_seed(0), } } + /// Enable or disable the trailing XXH64 content checksum + /// (upstream `ZSTD_c_checksumFlag`). Default `true`. Must be called + /// before the first [`write`](Write::write); once the frame header is + /// emitted the flag is fixed, so a late change returns an error rather + /// than producing a header/trailer mismatch. Without the `hash` feature + /// no checksum is emitted regardless. + pub fn set_content_checksum(&mut self, emit: bool) -> Result<(), Error> { + self.ensure_open()?; + if self.frame_started { + return Err(invalid_input_error( + "content checksum must be set before the first write", + )); + } + self.content_checksum = emit; + Ok(()) + } + /// Enable or disable magicless frame format (`ZSTD_f_zstd1_magicless`). /// /// When set to `true`, the frame header serialized by this encoder @@ -203,7 +226,7 @@ impl StreamingEncoder { .expect("streaming encoder drain must be present when finishing"); #[cfg(feature = "hash")] - { + if self.content_checksum { let checksum = self.hasher.finish() as u32; drain .write_all(&checksum.to_le_bytes()) @@ -290,7 +313,7 @@ impl StreamingEncoder { let header = FrameHeader { frame_content_size: self.pledged_content_size, single_segment, - content_checksum: cfg!(feature = "hash"), + content_checksum: cfg!(feature = "hash") && self.content_checksum, dictionary_id: None, window_size: if single_segment { None @@ -456,7 +479,7 @@ impl StreamingEncoder { if moved_into_matcher { #[cfg(feature = "hash")] - { + if self.content_checksum { self.hasher.write(self.state.matcher.get_last_space()); } } else { @@ -484,7 +507,9 @@ impl StreamingEncoder { #[cfg(feature = "hash")] fn hash_block(&mut self, uncompressed_data: &[u8]) { - self.hasher.write(uncompressed_data); + if self.content_checksum { + self.hasher.write(uncompressed_data); + } } #[cfg(not(feature = "hash"))] From 8100658b97d82f5273124fc87b629707752d8acc Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Mon, 8 Jun 2026 19:21:59 +0300 Subject: [PATCH 5/7] feat(wasm): expose content-checksum params and accessors Surface the content-checksum controls through the wasm/npm bindings: - compress / compressUsingDict / ZstdCompressStream take an optional checksum bool (emit the trailing XXH64 digest or not) - decompress / decompressUsingDict / ZstdDecompressStream take an optional ContentChecksum mode (None / EmitOnly / Verify); Verify throws on a digest mismatch so TypeScript callers see a corrupt frame as a rejected promise - ZstdDecompressStream gains storedChecksum() / calculatedChecksum() accessors Defaults are off (None / no emit) to preserve the package's prior behaviour (it shipped without the checksum code) and libzstd's checksumFlag=0 default, and to skip the XXH64 cost when no digest is needed. The core hash feature is enabled for the wasm crate (twox-hash is no_std, so no std is pulled). FrameDecoder::verify_content_checksum is made public so callers driving decode_blocks directly (the wasm streaming path) can verify per frame. --- zstd-wasm/Cargo.toml | 5 +- zstd-wasm/npm/index.ts | 80 +++++++++++++++---- zstd-wasm/src/lib.rs | 122 +++++++++++++++++++++++++---- zstd/src/decoding/frame_decoder.rs | 10 ++- 4 files changed, 181 insertions(+), 36 deletions(-) diff --git a/zstd-wasm/Cargo.toml b/zstd-wasm/Cargo.toml index bdc4d2ed3..c9e566b99 100644 --- a/zstd-wasm/Cargo.toml +++ b/zstd-wasm/Cargo.toml @@ -34,7 +34,10 @@ wasm-bindgen = "0.2" # go stale on the next core bump and break packaging again. The requirement # is purely ceremonial: `path` always wins for local/workspace builds and # this crate is never published to crates.io. -structured-zstd = { version = "0", path = "../zstd", default-features = false, features = ["kernel_simd128"] } +# `hash` enables XXH64 content-checksum support (the `ContentChecksum` modes + +# encoder toggle exposed below). twox-hash is `no_std`, so this adds the +# checksum code without pulling `std`. +structured-zstd = { version = "0", path = "../zstd", default-features = false, features = ["kernel_simd128", "hash"] } # Size-tuned wasm-pack release profile: optimise for size + a wasm-opt -Oz # post-pass, since payload bytes are the budget for a published wasm module. diff --git a/zstd-wasm/npm/index.ts b/zstd-wasm/npm/index.ts index ccb4552a8..257ef6db9 100644 --- a/zstd-wasm/npm/index.ts +++ b/zstd-wasm/npm/index.ts @@ -17,6 +17,19 @@ import { simd } from "wasm-feature-detect"; +/** + * How the decoder treats a frame's optional XXH64 content checksum. Numeric + * values match the wasm-bindgen `ContentChecksum` enum. + */ +export enum ContentChecksum { + /** Skip the XXH64 pass entirely (fastest; no verification). */ + None = 0, + /** Compute the checksum but do not error on a mismatch (default). */ + EmitOnly = 1, + /** Compute and verify; a mismatch rejects the decode. */ + Verify = 2, +} + /** Shape of a wasm-pack `web`-target payload glue module (simd or scalar). */ interface Payload { /** Async wasm initialiser. Accepts wasm bytes / a URL, or nothing (browser). */ @@ -24,12 +37,21 @@ interface Payload { // (bytes / URL / Response / Module), and Node's `fs.readFile` returns // `Buffer`, wider than the DOM `BufferSource` in the .d.ts. default: (moduleOrPath?: unknown) => Promise; - compress: (data: Uint8Array, level: number) => Uint8Array; - decompress: (data: Uint8Array) => Uint8Array; - compressUsingDict: (data: Uint8Array, dict: Uint8Array, level: number) => Uint8Array; - decompressUsingDict: (data: Uint8Array, dict: Uint8Array) => Uint8Array; - ZstdDecompressStream: new () => DecompressStream; - ZstdCompressStream: new (level: number) => CompressStream; + compress: (data: Uint8Array, level: number, checksum?: boolean) => Uint8Array; + decompress: (data: Uint8Array, checksum?: ContentChecksum) => Uint8Array; + compressUsingDict: ( + data: Uint8Array, + dict: Uint8Array, + level: number, + checksum?: boolean, + ) => Uint8Array; + decompressUsingDict: ( + data: Uint8Array, + dict: Uint8Array, + checksum?: ContentChecksum, + ) => Uint8Array; + ZstdDecompressStream: new (checksum?: ContentChecksum) => DecompressStream; + ZstdCompressStream: new (level: number, checksum?: boolean) => CompressStream; } /** @@ -42,8 +64,22 @@ interface Payload { export interface DecompressStream { /** Feed compressed bytes; returns decompressed output available so far. */ push(chunk: Uint8Array): Uint8Array; - /** Signal end of input; returns the final bytes. Throws if incomplete. */ + /** + * Signal end of input; returns the final bytes. Throws if incomplete, or + * (in {@link ContentChecksum.Verify} mode) if the content checksum is wrong. + */ finish(): Uint8Array; + /** + * Content checksum stored in the frame trailer, or `undefined` if none. + * Meaningful after {@link DecompressStream.finish}. + */ + storedChecksum(): number | undefined; + /** + * XXH64 digest computed over the output (low 32 bits), or `undefined` under + * {@link ContentChecksum.None} or when the frame carried no checksum. Lets + * callers verify manually under {@link ContentChecksum.EmitOnly}. + */ + calculatedChecksum(): number | undefined; /** Release the underlying wasm handle. */ free(): void; } @@ -124,18 +160,25 @@ export async function init(): Promise { export async function compress( data: Uint8Array, level: number = DEFAULT_LEVEL, + checksum?: boolean, ): Promise { loading ??= load(); - return (await loading).compress(data, level); + return (await loading).compress(data, level, checksum); } /** * Decompress a complete Zstandard frame. Rejects if the input is not a valid, - * complete frame. + * complete frame, or — when `checksum` is {@link ContentChecksum.Verify} — if + * the content checksum does not match. Defaults to + * {@link ContentChecksum.EmitOnly}; pass {@link ContentChecksum.None} to skip + * the XXH64 pass for speed. */ -export async function decompress(data: Uint8Array): Promise { +export async function decompress( + data: Uint8Array, + checksum?: ContentChecksum, +): Promise { loading ??= load(); - return (await loading).decompress(data); + return (await loading).decompress(data, checksum); } /** @@ -148,9 +191,10 @@ export async function compressUsingDict( data: Uint8Array, dict: Uint8Array, level: number = DEFAULT_LEVEL, + checksum?: boolean, ): Promise { loading ??= load(); - return (await loading).compressUsingDict(data, dict, level); + return (await loading).compressUsingDict(data, dict, level, checksum); } /** @@ -161,9 +205,10 @@ export async function compressUsingDict( export async function decompressUsingDict( data: Uint8Array, dict: Uint8Array, + checksum?: ContentChecksum, ): Promise { loading ??= load(); - return (await loading).decompressUsingDict(data, dict); + return (await loading).decompressUsingDict(data, dict, checksum); } /** @@ -172,9 +217,11 @@ export async function decompressUsingDict( * Unlike the common npm wasm zstd packages, the frame need not be fully * buffered — the decoder window lives on the wasm side across chunks. */ -export async function createDecompressStream(): Promise { +export async function createDecompressStream( + checksum?: ContentChecksum, +): Promise { loading ??= load(); - return new (await loading).ZstdDecompressStream(); + return new (await loading).ZstdDecompressStream(checksum); } /** @@ -188,7 +235,8 @@ export async function createDecompressStream(): Promise { */ export async function createCompressStream( level: number = DEFAULT_LEVEL, + checksum?: boolean, ): Promise { loading ??= load(); - return new (await loading).ZstdCompressStream(level); + return new (await loading).ZstdCompressStream(level, checksum); } diff --git a/zstd-wasm/src/lib.rs b/zstd-wasm/src/lib.rs index 3f8bacec5..63ae65912 100644 --- a/zstd-wasm/src/lib.rs +++ b/zstd-wasm/src/lib.rs @@ -21,26 +21,70 @@ use structured_zstd::encoding::{ use structured_zstd::io::{Read, Write}; use wasm_bindgen::prelude::*; +/// How the decoder treats a frame's optional content checksum. Mirrors the +/// core `structured_zstd::decoding::ContentChecksum`. +#[wasm_bindgen] +#[derive(Copy, Clone, PartialEq, Eq, Debug)] +pub enum ContentChecksum { + /// Skip the XXH64 pass entirely (fastest; no verification). + None = 0, + /// Compute the checksum but do not error on a mismatch (default). + EmitOnly = 1, + /// Compute and verify; a mismatch throws on decode. + Verify = 2, +} + +/// Resolve an optional JS-supplied mode to the core enum. Defaults to `None` +/// (skip the XXH64 pass) when omitted: this matches the package's prior +/// behaviour (it shipped without the checksum code at all) and libzstd's +/// `checksumFlag = 0` default, and avoids paying the XXH64 cost on a digest no +/// wasm accessor even exposes. Callers opt into `Verify` explicitly. +fn core_checksum(mode: Option) -> structured_zstd::decoding::ContentChecksum { + use structured_zstd::decoding::ContentChecksum as Core; + match mode.unwrap_or(ContentChecksum::None) { + ContentChecksum::None => Core::None, + ContentChecksum::EmitOnly => Core::EmitOnly, + ContentChecksum::Verify => Core::Verify, + } +} + /// Compress `data` into a standard Zstandard frame at compression `level`. /// /// `level` follows the zstd scale: `1..=22` (higher = smaller/slower) and /// negative levels (`-7..=-1`) for the ultra-fast tier. The returned frame /// decodes in any compliant zstd decoder, including the native C library. +/// +/// `checksum` is optional (default `true`): pass `false` to emit a frame +/// without the trailing XXH64 content checksum (semantics of +/// `ZSTD_c_checksumFlag`). #[wasm_bindgen] -pub fn compress(data: &[u8], level: i32) -> Vec { - compress_slice_to_vec(data, CompressionLevel::Level(level)) +pub fn compress(data: &[u8], level: i32, checksum: Option) -> Vec { + if checksum.unwrap_or(true) { + // Default: keep the historical fast path verbatim. + compress_slice_to_vec(data, CompressionLevel::Level(level)) + } else { + let mut enc: FrameCompressor = FrameCompressor::new(CompressionLevel::Level(level)); + enc.set_content_checksum(false); + enc.compress_independent_frame(data) + } } /// Decompress a complete Zstandard frame back into its original bytes. /// -/// Throws a JavaScript `Error` if the input is not a valid, complete frame. +/// Throws a JavaScript `Error` if the input is not a valid, complete frame, +/// or (when `checksum` is `Verify`) if the content checksum does not match. +/// `checksum` is optional (default `EmitOnly`): pass `ContentChecksum.None` +/// to skip the XXH64 pass for speed, or `ContentChecksum.Verify` to validate. #[wasm_bindgen] -pub fn decompress(data: &[u8]) -> Result, JsError> { +pub fn decompress(data: &[u8], checksum: Option) -> Result, JsError> { // Stream the frame so the output Vec grows to fit — works for frames with // or without a content-size header (the fixed-size `decode_all_to_vec` // requires the caller to know the decoded length up front). let mut decoder = StreamingDecoder::new(data) .map_err(|err| JsError::new(&format!("structured-zstd: invalid frame: {err:?}")))?; + decoder + .decoder + .set_content_checksum(core_checksum(checksum)); let mut out = Vec::new(); decoder .read_to_end(&mut out) @@ -54,8 +98,14 @@ pub fn decompress(data: &[u8]) -> Result, JsError> { /// small, similar payloads compress far better. The dictionary is the raw /// zstd dictionary blob (e.g. from `zstd --train`). Throws if it is invalid. #[wasm_bindgen(js_name = compressUsingDict)] -pub fn compress_using_dict(data: &[u8], dict: &[u8], level: i32) -> Result, JsError> { +pub fn compress_using_dict( + data: &[u8], + dict: &[u8], + level: i32, + checksum: Option, +) -> Result, JsError> { let mut enc: FrameCompressor = FrameCompressor::new(CompressionLevel::Level(level)); + enc.set_content_checksum(checksum.unwrap_or(true)); enc.set_dictionary_from_bytes(dict) .map_err(|err| JsError::new(&format!("structured-zstd: invalid dictionary: {err:?}")))?; Ok(enc.compress_independent_frame(data)) @@ -67,12 +117,19 @@ pub fn compress_using_dict(data: &[u8], dict: &[u8], level: i32) -> Result Result, JsError> { +pub fn decompress_using_dict( + data: &[u8], + dict: &[u8], + checksum: Option, +) -> Result, JsError> { let mut decoder = StreamingDecoder::new_with_dictionary_bytes(data, dict).map_err(|err| { JsError::new(&format!( "structured-zstd: dict decode init failed: {err:?}" )) })?; + decoder + .decoder + .set_content_checksum(core_checksum(checksum)); let mut out = Vec::new(); decoder.read_to_end(&mut out).map_err(|err| { JsError::new(&format!("structured-zstd: dict decompress failed: {err:?}")) @@ -145,10 +202,15 @@ pub struct ZstdDecompressStream { #[wasm_bindgen] impl ZstdDecompressStream { + /// `checksum` is optional (default `EmitOnly`) and applies to the whole + /// stream, so set it here rather than mid-stream: `None` skips the XXH64 + /// pass, `Verify` validates the content checksum at [`Self::finish`]. #[wasm_bindgen(constructor)] - pub fn new() -> ZstdDecompressStream { + pub fn new(checksum: Option) -> ZstdDecompressStream { + let mut decoder = FrameDecoder::new(); + decoder.set_content_checksum(core_checksum(checksum)); ZstdDecompressStream { - decoder: FrameDecoder::new(), + decoder, pending: Vec::new(), header_done: false, checksum: false, @@ -164,7 +226,8 @@ impl ZstdDecompressStream { } /// Signal end of input; returns the final decompressed bytes. Throws if the - /// stream ended before the frame completed. + /// stream ended before the frame completed, or (in `Verify` mode) if the + /// content checksum does not match. pub fn finish(&mut self) -> Result, JsError> { let out = self.pump()?; if !self.finished { @@ -172,13 +235,36 @@ impl ZstdDecompressStream { "structured-zstd: stream ended before the frame completed", )); } + // The frame is fully decoded and drained (pump collects every block), + // so the running digest is final: validate it in Verify mode (no-op + // otherwise). The Display of `ChecksumMismatch` names it a corrupt + // frame, so the thrown JS error reads clearly on the TS side. + self.decoder + .verify_content_checksum() + .map_err(|err| JsError::new(&format!("structured-zstd: {err}")))?; Ok(out) } + + /// The content checksum stored in the frame's 4-byte trailer, or + /// `undefined` if the frame carried none. Meaningful after [`Self::finish`]. + #[wasm_bindgen(js_name = storedChecksum)] + pub fn stored_checksum(&self) -> Option { + self.decoder.get_checksum_from_data() + } + + /// The XXH64 digest the decoder computed over the output (low 32 bits), or + /// `undefined` when the mode is `None` or the frame carried no checksum. + /// Meaningful after [`Self::finish`]; lets callers verify manually under + /// `EmitOnly` without enabling the throwing `Verify` mode. + #[wasm_bindgen(js_name = calculatedChecksum)] + pub fn calculated_checksum(&self) -> Option { + self.decoder.get_calculated_checksum() + } } impl Default for ZstdDecompressStream { fn default() -> Self { - Self::new() + Self::new(None) } } @@ -243,14 +329,18 @@ pub struct ZstdCompressStream { #[wasm_bindgen] impl ZstdCompressStream { /// Open a streaming compressor at `level` (zstd scale: `1..=22`, negatives - /// for the ultra-fast tier). + /// for the ultra-fast tier). `checksum` is optional (default `true`): pass + /// `false` to seal the frame without a trailing content checksum. #[wasm_bindgen(constructor)] - pub fn new(level: i32) -> ZstdCompressStream { + pub fn new(level: i32, checksum: Option) -> ZstdCompressStream { + let mut encoder = StreamingEncoder::new(Vec::new(), CompressionLevel::Level(level)); + // Provably Ok: the encoder is fresh, so no frame header has been + // emitted yet (the only failure mode of this setter). + encoder + .set_content_checksum(checksum.unwrap_or(true)) + .expect("fresh streaming encoder accepts the content-checksum toggle"); ZstdCompressStream { - encoder: Some(StreamingEncoder::new( - Vec::new(), - CompressionLevel::Level(level), - )), + encoder: Some(encoder), } } diff --git a/zstd/src/decoding/frame_decoder.rs b/zstd/src/decoding/frame_decoder.rs index 46316e480..ba6bdd9d3 100644 --- a/zstd/src/decoding/frame_decoder.rs +++ b/zstd/src/decoding/frame_decoder.rs @@ -1350,11 +1350,15 @@ impl FrameDecoder { /// Compare the frame's stored content checksum against the digest the /// decoder computed, returning [`FrameDecoderError::ChecksumMismatch`] on /// disagreement. No-op unless the mode is [`ContentChecksum::Verify`] and - /// the frame carries a trailing checksum. Call once per frame after the - /// frame is fully decoded (and, on the drain path, fully drained) so both + /// the frame carries a trailing checksum. + /// + /// [`decode_all`](Self::decode_all) and the streaming reader call this + /// automatically. Callers driving [`decode_blocks`](Self::decode_blocks) + /// directly invoke it themselves once per frame, after the frame is fully + /// decoded AND fully drained (e.g. via [`collect`](Self::collect)), so both /// the stored value and the running digest are final. #[cfg(feature = "hash")] - pub(crate) fn verify_content_checksum(&self) -> Result<(), FrameDecoderError> { + pub fn verify_content_checksum(&self) -> Result<(), FrameDecoderError> { if self.content_checksum != ContentChecksum::Verify { return Ok(()); } From fd4ffa41dae866caa1df3be8cef246b7a2b5f654 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Mon, 8 Jun 2026 19:51:25 +0300 Subject: [PATCH 6/7] feat(encode): off-thread content checksum for one-shot frames Add FrameCompressor::set_offload_checksum (std + hash). On the one-shot compress_independent_frame* path it hashes the whole immutable input on a std::thread::scope worker concurrent with the block-compress loop (a shared read of the input, disjoint from the &mut self loop, no unsafe, no copy), and routes the joined digest into the frame trailer in place of the inline pass; the per-block inline hash is skipped in that mode. Output is byte-identical to the inline path. No effect on the streaming reader path, and a no-op without std + hash. --- zstd/src/decoding/frame_decoder.rs | 33 ++++++++++++ zstd/src/encoding/frame_compressor.rs | 76 +++++++++++++++++++++++++-- 2 files changed, 106 insertions(+), 3 deletions(-) diff --git a/zstd/src/decoding/frame_decoder.rs b/zstd/src/decoding/frame_decoder.rs index ba6bdd9d3..b645fdd61 100644 --- a/zstd/src/decoding/frame_decoder.rs +++ b/zstd/src/decoding/frame_decoder.rs @@ -3059,6 +3059,39 @@ mod tests { assert!(dec.get_calculated_checksum().is_none()); } + #[cfg(all(feature = "std", feature = "hash"))] + #[test] + fn offload_checksum_output_is_byte_identical_to_inline() { + // The scoped-worker checksum path must produce exactly the same frame + // bytes as the inline path (it only moves the XXH64 work off-thread), + // and the frame must still decode + verify. + let payload: Vec = (0..200_000u32) + .map(|i| (i.wrapping_mul(2654435761) >> 24) as u8) + .collect(); + + let mut inline: FrameCompressor = FrameCompressor::new(CompressionLevel::Default); + let inline_bytes = inline.compress_independent_frame(&payload); + + let mut offloaded: FrameCompressor = FrameCompressor::new(CompressionLevel::Default); + offloaded.set_offload_checksum(true); + let offloaded_bytes = offloaded.compress_independent_frame(&payload); + + assert_eq!( + inline_bytes, offloaded_bytes, + "offloaded checksum must yield byte-identical frame" + ); + + use crate::decoding::ContentChecksum; + let mut dec = FrameDecoder::new(); + dec.set_content_checksum(ContentChecksum::Verify); + let mut out = + alloc::vec![0u8; payload.len() + super::super::buffer_backend::WILDCOPY_OVERLENGTH]; + let n = dec + .decode_all(&offloaded_bytes, &mut out) + .expect("offloaded-checksum frame must decode and verify"); + assert_eq!(&out[..n], payload.as_slice()); + } + #[cfg(feature = "hash")] #[test] fn encoder_without_checksum_emits_no_trailing_digest() { diff --git a/zstd/src/encoding/frame_compressor.rs b/zstd/src/encoding/frame_compressor.rs index 442278483..2e4930bcc 100644 --- a/zstd/src/encoding/frame_compressor.rs +++ b/zstd/src/encoding/frame_compressor.rs @@ -108,8 +108,20 @@ pub struct FrameCompressor< /// feature at frame-build time, so without `hash` no checksum is emitted /// regardless. Set via [`Self::set_content_checksum`]. content_checksum: bool, + /// When `true` (and `std` + `hash` are active), the one-shot + /// (`compress_independent_frame*`) path computes the XXH64 content + /// checksum on a scoped worker thread concurrent with compression instead + /// of inline. Default `false`. Set via [`Self::set_offload_checksum`]. No + /// effect on the streaming `compress()` reader path. + #[cfg(feature = "hash")] + offload_checksum: bool, #[cfg(feature = "hash")] hasher: XxHash64, + /// Set by the one-shot offload path to the digest a scoped worker computed + /// over the whole input; `write_frame_to_vec` uses it in place of the + /// inline `hasher.finish()`. `None` on every non-offload path. + #[cfg(feature = "hash")] + offloaded_digest: Option, /// Block-layout introspection populated at the end of every /// successful `compress()`. `None` until the first call. /// Behind the `lsm` feature gate. @@ -607,7 +619,11 @@ impl FrameCompressor { magicless: false, content_checksum: true, #[cfg(feature = "hash")] + offload_checksum: false, + #[cfg(feature = "hash")] hasher: XxHash64::with_seed(0), + #[cfg(feature = "hash")] + offloaded_digest: None, #[cfg(feature = "lsm")] frame_emit_info: None, #[cfg(all(feature = "lsm", feature = "hash"))] @@ -753,10 +769,40 @@ impl FrameCompressor { // resolved window/header and could flip borrowed eligibility). self.source_size_hint = Some(input.len() as u64); let prep = self.prepare_frame(); - let (all_blocks, total_uncompressed) = self.run_one_frame(input, &prep); + let (all_blocks, total_uncompressed) = self.run_one_frame_maybe_offload(input, &prep); self.write_frame_to_vec(out, all_blocks, total_uncompressed, &prep); } + /// [`Self::run_one_frame`], optionally with the content checksum computed + /// on a scoped worker thread concurrent with compression (when + /// `offload_checksum` + `content_checksum` are set and `std` + `hash` are + /// available). The worker hashes the whole immutable `input` (a shared + /// read, disjoint from the `&mut self` block loop) and the joined digest is + /// stashed in `offloaded_digest` for the frame trailer; the inline per-block + /// hash is skipped in that case. Output is byte-identical to the inline + /// path; only the XXH64 work moves off the compression thread. + fn run_one_frame_maybe_offload(&mut self, input: &[u8], prep: &FramePrep) -> (Vec, u64) { + #[cfg(all(feature = "std", feature = "hash"))] + if self.offload_checksum && self.content_checksum { + use core::hash::Hasher; + let mut blocks: Option<(Vec, u64)> = None; + let digest = std::thread::scope(|s| { + let worker = s.spawn(move || { + let mut hasher = XxHash64::with_seed(0); + hasher.write(input); + hasher.finish() + }); + blocks = Some(self.run_one_frame(input, prep)); + worker + .join() + .expect("content-checksum worker thread panicked") + }); + self.offloaded_digest = Some(digest); + return blocks.expect("block loop runs inside the checksum scope"); + } + self.run_one_frame(input, prep) + } + /// Convenience wrapper over [`Self::compress_independent_frame_into`] /// that allocates and returns a fresh `Vec` per call. Prefer the /// `_into` form in tight per-block-frame loops to reuse one output @@ -852,8 +898,11 @@ impl FrameCompressor { let end = (start + block_len).min(input.len()); let block = &input[start..end]; let last_block = end == input.len(); + // Skipped when offloading: the scoped worker hashes the whole + // input concurrently, so an inline feed here would just duplicate + // the work on the compression thread. #[cfg(feature = "hash")] - if self.content_checksum { + if self.content_checksum && !self.offload_checksum { self.hasher.write(block); } crate::encoding::levels::compress_block_encoded_borrowed( @@ -899,7 +948,11 @@ impl FrameCompressor { magicless: false, content_checksum: true, #[cfg(feature = "hash")] + offload_checksum: false, + #[cfg(feature = "hash")] hasher: XxHash64::with_seed(0), + #[cfg(feature = "hash")] + offloaded_digest: None, #[cfg(feature = "lsm")] frame_emit_info: None, #[cfg(all(feature = "lsm", feature = "hash"))] @@ -934,6 +987,17 @@ impl FrameCompressor { self.content_checksum = emit; } + /// Compute the content checksum on a scoped worker thread (concurrent with + /// compression) on the one-shot `compress_independent_frame*` path, instead + /// of hashing inline. Default `false`. Requires `std` + `hash`; a no-op + /// without them or when the content checksum is disabled. The output is + /// byte-identical either way; this only moves the XXH64 work off the + /// compression thread. + #[cfg(feature = "hash")] + pub fn set_offload_checksum(&mut self, offload: bool) { + self.offload_checksum = offload; + } + /// Before calling [FrameCompressor::compress] you need to set the source. /// /// This is the data that is compressed and written into the drain. @@ -1452,7 +1516,13 @@ impl FrameCompressor { out.extend_from_slice(&all_blocks); #[cfg(feature = "hash")] if self.content_checksum { - out.extend_from_slice(&(self.hasher.finish() as u32).to_le_bytes()); + // Use the scoped worker's digest when the one-shot offload path + // produced one; otherwise the inline hasher's. + let digest = self + .offloaded_digest + .take() + .unwrap_or_else(|| self.hasher.finish()); + out.extend_from_slice(&(digest as u32).to_le_bytes()); } #[cfg(feature = "lsm")] self.populate_frame_emit_info(header_buf.len(), &all_blocks); From 80d529380c2e6d580f1e554b1d887927f56a3609 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Mon, 8 Jun 2026 20:16:38 +0300 Subject: [PATCH 7/7] test(encode): one-shot checksum offload vs inline timing harness --- zstd/examples/encode_offload.rs | 73 +++++++++++++++++++++++++++++++++ 1 file changed, 73 insertions(+) create mode 100644 zstd/examples/encode_offload.rs diff --git a/zstd/examples/encode_offload.rs b/zstd/examples/encode_offload.rs new file mode 100644 index 000000000..6274bdc39 --- /dev/null +++ b/zstd/examples/encode_offload.rs @@ -0,0 +1,73 @@ +//! Measure the one-shot encode content-checksum offload: hashing the input on +//! a scoped worker concurrent with compression vs hashing inline. Run on the +//! i9: `cargo run --release --example encode_offload`. + +use std::hint::black_box; +use std::time::Instant; + +use structured_zstd::encoding::{CompressionLevel, FrameCompressor}; + +fn repeated_log_lines(len: usize) -> Vec { + const LINES: &[&str] = &[ + "ts=2026-03-26T21:39:28Z level=INFO msg=\"flush memtable\" tenant=demo table=orders region=eu-west\n", + "ts=2026-03-26T21:39:29Z level=INFO msg=\"rotate segment\" tenant=demo table=orders region=eu-west\n", + "ts=2026-03-26T21:39:30Z level=INFO msg=\"compact level\" tenant=demo table=orders region=eu-west\n", + "ts=2026-03-26T21:39:31Z level=INFO msg=\"write block\" tenant=demo table=orders region=eu-west\n", + ]; + let mut bytes = Vec::with_capacity(len); + while bytes.len() < len { + for line in LINES { + if bytes.len() == len { + break; + } + let remaining = len - bytes.len(); + bytes.extend_from_slice(&line.as_bytes()[..line.len().min(remaining)]); + } + } + bytes +} + +fn time(raw: &[u8], level: i32, offload: bool, iters: u32) -> f64 { + let mut enc: FrameCompressor = FrameCompressor::new(CompressionLevel::Level(level)); + enc.set_offload_checksum(offload); + let mut out = Vec::new(); + // warm + enc.compress_independent_frame_into(raw, &mut out); + let start = Instant::now(); + for _ in 0..iters { + enc.compress_independent_frame_into(black_box(raw), &mut out); + black_box(&out[..]); + } + start.elapsed().as_secs_f64() * 1e3 / f64::from(iters) +} + +fn run(name: &str, raw: &[u8], level: i32, iters: u32) { + // Verify byte-identical output before timing. + let mut a: FrameCompressor = FrameCompressor::new(CompressionLevel::Level(level)); + let inline_bytes = a.compress_independent_frame(raw); + let mut b: FrameCompressor = FrameCompressor::new(CompressionLevel::Level(level)); + b.set_offload_checksum(true); + let off_bytes = b.compress_independent_frame(raw); + assert_eq!( + inline_bytes, off_bytes, + "offload output must be byte-identical" + ); + + let inline = time(raw, level, false, iters); + let off = time(raw, level, true, iters); + println!( + "{name:>16} L{level:>2} raw={:>9} inline={inline:8.3}ms offload={off:8.3}ms offload/inline={:.3}x", + raw.len(), + off / inline, + ); +} + +fn main() { + let mb = 1024 * 1024; + let data = repeated_log_lines(8 * mb); + println!("== one-shot encode: checksum inline vs scoped-offload =="); + for level in [3, 8, 12, 19] { + let iters = if level >= 19 { 6 } else { 20 }; + run("large-log-8m", &data, level, iters); + } +}