From 722c497672d2c530bf89d19ff9839dfeb96fd42f Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Thu, 4 Jun 2026 16:08:46 +0300 Subject: [PATCH 01/10] refactor(encode): rename block-splitter fns to describe behaviour Drop the reference-impl prefix from the pre-split identifiers, naming them for what they do: split_block_by_chunks, split_block_from_borders, pre_split_level, optimal_block_size. Reference attribution stays in the doc comments (prose), not in the identifiers. Pure rename across the splitter, its callers, tests, and comment references. Byte-identical, 683 tests pass. --- zstd/benches/support/mod.rs | 2 +- zstd/src/encoding/frame_compressor.rs | 71 ++++++++--------------- zstd/src/encoding/match_generator.rs | 2 +- zstd/src/encoding/sequence_capture.rs | 6 +- zstd/tests/block_splitter_donor_parity.rs | 6 +- 5 files changed, 33 insertions(+), 54 deletions(-) diff --git a/zstd/benches/support/mod.rs b/zstd/benches/support/mod.rs index fdfcd6067..3225904a1 100644 --- a/zstd/benches/support/mod.rs +++ b/zstd/benches/support/mod.rs @@ -202,7 +202,7 @@ fn build_supported_levels() -> Vec { // `CompressionLevel::from_level(n)` — so the bench label // `level__` matches the variant exercised by the // encoder. `from_level(11)` collapses to `Best`, and the named - // `Best` variant bypasses `donor_pre_split_level`'s + // `Best` variant bypasses `pre_split_level`'s // `Level(11..=15) -> Some(0)` arm (the borders pre-splitter // landed in #140), so the numbers would silently diverge from // what a user calling `compress_to_vec(input, Level(11))` sees. diff --git a/zstd/src/encoding/frame_compressor.rs b/zstd/src/encoding/frame_compressor.rs index d9b96faef..4cb570229 100644 --- a/zstd/src/encoding/frame_compressor.rs +++ b/zstd/src/encoding/frame_compressor.rs @@ -315,7 +315,7 @@ fn presplit_merge_events(acc: &mut PreSplitFingerprint, new_fp: &PreSplitFingerp acc.nb_events = acc.nb_events.saturating_add(new_fp.nb_events); } -fn donor_split_block_by_chunks(block: &[u8], level: usize) -> usize { +fn split_block_by_chunks(block: &[u8], level: usize) -> usize { debug_assert_eq!(block.len(), MAX_BLOCK_SIZE as usize); debug_assert!((1..=4).contains(&level)); let (sampling_rate, hash_log) = match level - 1 { @@ -361,7 +361,7 @@ fn donor_split_block_by_chunks(block: &[u8], level: usize) -> usize { /// size when the two ends look indistinguishable. Cheaper than the /// chunk-based path because it touches at most 1.5 KB of input /// regardless of block size. -fn donor_split_block_from_borders(block: &[u8]) -> usize { +fn split_block_from_borders(block: &[u8]) -> usize { debug_assert_eq!(block.len(), MAX_BLOCK_SIZE as usize); let block_size = block.len(); let mut past = PreSplitFingerprint::default(); @@ -405,7 +405,7 @@ fn donor_split_block_from_borders(block: &[u8]) -> usize { } } -fn donor_pre_split_level(level: CompressionLevel) -> Option { +fn pre_split_level(level: CompressionLevel) -> Option { match level { // Donor `ZSTD_blockSplitter_level` table (`clevels.h`): cheap // borders heuristic for lazy2 / btlazy2 strategies (levels @@ -438,7 +438,7 @@ pub(crate) fn xxh64_block_low32(data: &[u8]) -> u32 { /// Bench-only entry point for the donor-parity comparator test in /// `tests/block_splitter_donor_parity.rs`. Dispatches to the same /// `_from_borders` (split_level == 0) / `_by_chunks` (split_level ∈ -/// 1..=4) ports that `donor_optimal_block_size` itself routes +/// 1..=4) ports that `optimal_block_size` itself routes /// through. Caller is responsible for passing exactly /// `MAX_BLOCK_SIZE` bytes (per donor `ZSTD_splitBlock` contract — /// "@blockSize must be == 128 KB" in `zstd_preSplit.h`). @@ -454,20 +454,20 @@ pub(crate) fn block_splitter_decision_for_bench(block: &[u8], split_level: usize "block_splitter_decision_for_bench: split_level must be in 0..=4, got {split_level}" ); if split_level == 0 { - donor_split_block_from_borders(block) + split_block_from_borders(block) } else { - donor_split_block_by_chunks(block, split_level) + split_block_by_chunks(block, split_level) } } -pub(crate) fn donor_optimal_block_size( +pub(crate) fn optimal_block_size( level: CompressionLevel, block: &[u8], remaining_src_size: usize, block_size_max: usize, savings: i64, ) -> usize { - let Some(split_level) = donor_pre_split_level(level) else { + let Some(split_level) = pre_split_level(level) else { return remaining_src_size.min(block_size_max); }; if remaining_src_size < MAX_BLOCK_SIZE as usize || block_size_max < MAX_BLOCK_SIZE as usize { @@ -484,9 +484,9 @@ pub(crate) fn donor_optimal_block_size( // `split_level == 1..=4` → byChunks with internal sampling level // `split_level - 1`. let raw_split = if split_level == 0 { - donor_split_block_from_borders(&block[..MAX_BLOCK_SIZE as usize]) + split_block_from_borders(&block[..MAX_BLOCK_SIZE as usize]) } else { - donor_split_block_by_chunks(&block[..MAX_BLOCK_SIZE as usize], split_level) + split_block_by_chunks(&block[..MAX_BLOCK_SIZE as usize], split_level) }; raw_split .max(PRESPLIT_BLOCK_MIN) @@ -988,7 +988,7 @@ impl FrameCompressor { if !matches!(self.compression_level, CompressionLevel::Uncompressed) && uncompressed_data.len() == block_capacity { - let block_len = donor_optimal_block_size( + let block_len = optimal_block_size( self.compression_level, &uncompressed_data, remaining_for_split, @@ -2854,9 +2854,9 @@ mod tests { /// function takes the early-return path at /// `zstd_preSplit.c:214` returning `blockSize`. #[test] - fn donor_split_block_from_borders_keeps_homogeneous_block() { + fn split_block_from_borders_keeps_homogeneous_block() { let block = vec![0xAAu8; MAX_BLOCK_SIZE as usize]; - let split = super::donor_split_block_from_borders(&block); + let split = super::split_block_from_borders(&block); assert_eq!(split, MAX_BLOCK_SIZE as usize); } @@ -2874,7 +2874,7 @@ mod tests { /// rather than just "one of {32K, 64K, 96K}" so a regression /// to a different quantised arm cannot silently slip through. #[test] - fn donor_split_block_from_borders_returns_midpoint_for_centred_transition() { + fn split_block_from_borders_returns_midpoint_for_centred_transition() { let mut block = vec![0u8; MAX_BLOCK_SIZE as usize]; for (i, byte) in block .iter_mut() @@ -2883,7 +2883,7 @@ mod tests { { *byte = (i % 251 + 1) as u8; } - let split = super::donor_split_block_from_borders(&block); + let split = super::split_block_from_borders(&block); assert_eq!( split, 64 * 1024, @@ -2892,42 +2892,21 @@ mod tests { ); } - /// `donor_pre_split_level` maps mid-range levels to the cheap + /// `pre_split_level` maps mid-range levels to the cheap /// borders heuristic and high levels to the byChunks path. Levels /// below 11 stay unsplit so the splitter never runs on fast / /// default presets where its per-block cost would dominate. #[test] - fn donor_pre_split_level_dispatches_by_compression_level() { + fn pre_split_level_dispatches_by_compression_level() { use crate::encoding::CompressionLevel; - assert_eq!( - super::donor_pre_split_level(CompressionLevel::Fastest), - None - ); - assert_eq!( - super::donor_pre_split_level(CompressionLevel::Default), - None - ); - assert_eq!(super::donor_pre_split_level(CompressionLevel::Better), None); - assert_eq!( - super::donor_pre_split_level(CompressionLevel::Level(7)), - None - ); - assert_eq!( - super::donor_pre_split_level(CompressionLevel::Level(11)), - Some(0) - ); - assert_eq!( - super::donor_pre_split_level(CompressionLevel::Level(15)), - Some(0) - ); - assert_eq!( - super::donor_pre_split_level(CompressionLevel::Level(16)), - Some(4) - ); - assert_eq!( - super::donor_pre_split_level(CompressionLevel::Level(22)), - Some(4) - ); + assert_eq!(super::pre_split_level(CompressionLevel::Fastest), None); + assert_eq!(super::pre_split_level(CompressionLevel::Default), None); + assert_eq!(super::pre_split_level(CompressionLevel::Better), None); + assert_eq!(super::pre_split_level(CompressionLevel::Level(7)), None); + assert_eq!(super::pre_split_level(CompressionLevel::Level(11)), Some(0)); + assert_eq!(super::pre_split_level(CompressionLevel::Level(15)), Some(0)); + assert_eq!(super::pre_split_level(CompressionLevel::Level(16)), Some(4)); + assert_eq!(super::pre_split_level(CompressionLevel::Level(22)), Some(4)); } /// End-to-end: a 256 KB heterogeneous payload compressed at diff --git a/zstd/src/encoding/match_generator.rs b/zstd/src/encoding/match_generator.rs index 207190bfc..7b20f858f 100644 --- a/zstd/src/encoding/match_generator.rs +++ b/zstd/src/encoding/match_generator.rs @@ -7180,7 +7180,7 @@ fn level22_donor_block_ranges(data: &[u8]) -> Vec<(usize, usize)> { while cursor < data.len() { let remaining = data.len() - cursor; let candidate_len = remaining.min(HC_BLOCKSIZE_MAX); - let block_len = crate::encoding::frame_compressor::donor_optimal_block_size( + let block_len = crate::encoding::frame_compressor::optimal_block_size( CompressionLevel::Level(22), &data[cursor..cursor + candidate_len], remaining, diff --git a/zstd/src/encoding/sequence_capture.rs b/zstd/src/encoding/sequence_capture.rs index d533b6940..5d5223875 100644 --- a/zstd/src/encoding/sequence_capture.rs +++ b/zstd/src/encoding/sequence_capture.rs @@ -318,7 +318,7 @@ fn compress_and_collect_sequences_impl( // counter. Two distinct splitter mechanisms exist in // `frame_compressor.rs`: // - // * Pre-split (`Level(11..=15)` borders + `donor_optimal_block_size`, + // * Pre-split (`Level(11..=15)` borders + `optimal_block_size`, // borders-only): the splitter chooses a shrunken `block_len` // BEFORE the matcher runs; the suffix is parked in // `pending_input` and the next compress-loop iteration calls @@ -783,7 +783,7 @@ mod tests { /// physical on-wire blocks per single matcher call — the /// per-matcher-call counter cannot track that shape. /// `Level(11..=15)` is intentionally NOT rejected: those levels - /// pre-split via `donor_optimal_block_size` (borders), so each + /// pre-split via `optimal_block_size` (borders), so each /// shrunken block is processed by its own matcher call and the /// counter stays correct. #[test] @@ -796,7 +796,7 @@ mod tests { } /// `Best` is documented as "roughly equivalent to Level 11" but - /// `donor_pre_split_level` matches the EXACT enum variants + /// `pre_split_level` matches the EXACT enum variants /// (`Level(11..=15)` / `Level(16..=22)`) — the `Best` arm /// falls through to `None`, so the named preset does NOT /// trigger the donor block-splitter. The guard above diff --git a/zstd/tests/block_splitter_donor_parity.rs b/zstd/tests/block_splitter_donor_parity.rs index b742141ba..90e68a04b 100644 --- a/zstd/tests/block_splitter_donor_parity.rs +++ b/zstd/tests/block_splitter_donor_parity.rs @@ -1,6 +1,6 @@ //! Donor-parity verification for the block splitter port. //! -//! Our `donor_split_block_from_borders` and `donor_split_block_by_chunks` +//! Our `split_block_from_borders` and `split_block_by_chunks` //! ports must produce byte-identical decisions to upstream donor //! `ZSTD_splitBlock` for every (block, split_level) input in the //! decode corpus. This test invokes both implementations on the same @@ -16,8 +16,8 @@ //! issue acceptance criteria, divergences correlating with ratio //! losses would justify porting more donor split logic; divergences //! that are size-neutral represent algorithmic freedom. Today we -//! assert STRICT equality because `donor_split_block_from_borders` -//! and `donor_split_block_by_chunks` are direct ports — any +//! assert STRICT equality because `split_block_from_borders` +//! and `split_block_by_chunks` are direct ports; any //! divergence is a porting bug, not algorithmic freedom. use std::ffi::c_void; From 9ea52606486f5efb00bccb629c4c9f6468629298 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Thu, 4 Jun 2026 16:22:52 +0300 Subject: [PATCH 02/10] perf(encode): pre-split greedy/lazy blocks via donor splitLevels MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extend the cheap fingerprint pre-splitter to the greedy and lazy bands, mirroring the donor `splitLevels[]` table by strategy: greedy → 1, lazy/lazy2 → 2, btlazy2 → 3, btopt/btultra/btultra2 → 4. Previously only levels 11..=22 pre-split (and 11..=15 used the weaker borders heuristic). This gives greedy/lazy per-sub-block entropy tables on heterogeneous input (the dominant literal-section ratio loss vs the C reference) through the cheap raw-input fingerprint, not the expensive post-parse super-block splitter. Fast/dfast (1..=4) stay un-split: their match-finding is cheap enough that the splitter would cost more throughput than the ratio it buys. 683 tests pass; behaviour change is bench-gated (ratio gain vs the added per-sub-block cost validated on the bench host). --- zstd/src/encoding/frame_compressor.rs | 41 ++++++++++++++++----------- 1 file changed, 24 insertions(+), 17 deletions(-) diff --git a/zstd/src/encoding/frame_compressor.rs b/zstd/src/encoding/frame_compressor.rs index 4cb570229..5548606bb 100644 --- a/zstd/src/encoding/frame_compressor.rs +++ b/zstd/src/encoding/frame_compressor.rs @@ -406,16 +406,19 @@ fn split_block_from_borders(block: &[u8]) -> usize { } fn pre_split_level(level: CompressionLevel) -> Option { + // Mirror the donor `splitLevels[]` table (indexed by strategy) for the + // greedy and lazy bands and above, where the cheap fingerprint + // pre-splitter pays for itself on heterogeneous payloads: + // greedy → 1, lazy/lazy2 → 2, btlazy2/btopt → 3, btultra/btultra2 → 4. + // Levels 1..=4 (fast/dfast) are deliberately left un-split: their + // match-finding is cheap, so the splitter's per-block fingerprint plus + // the extra per-sub-block entropy builds cost more throughput than the + // ratio they buy. The byChunks sampling rate / hashLog tighten with the + // split level (see `split_block_by_chunks`). match level { - // Donor `ZSTD_blockSplitter_level` table (`clevels.h`): cheap - // borders heuristic for lazy2 / btlazy2 strategies (levels - // 11..=15) — the splitter still pays for itself on - // heterogeneous payloads but the per-block cost stays bounded - // by two 512-byte histograms. - CompressionLevel::Level(11..=15) => Some(0), - // C zstd's default splitter level for btopt/btultra/btultra2 is 4 - // (`ZSTD_splitBlock_byChunks` with internal level 3 — sampling - // rate 1, `hashLog` 10). + CompressionLevel::Level(5) => Some(1), + CompressionLevel::Level(6..=12) => Some(2), + CompressionLevel::Level(13..=15) => Some(3), CompressionLevel::Level(16..=22) => Some(4), _ => None, } @@ -2892,19 +2895,23 @@ mod tests { ); } - /// `pre_split_level` maps mid-range levels to the cheap - /// borders heuristic and high levels to the byChunks path. Levels - /// below 11 stay unsplit so the splitter never runs on fast / - /// default presets where its per-block cost would dominate. + /// `pre_split_level` mirrors the donor `splitLevels[]` table by + /// strategy band: greedy → 1, lazy/lazy2 → 2, btlazy2 → 3, + /// btopt/btultra/btultra2 → 4. Fast/dfast (levels 1..=4) and the named + /// presets stay unsplit so the splitter never runs where its per-block + /// cost would dominate the cheap match-finding. #[test] fn pre_split_level_dispatches_by_compression_level() { use crate::encoding::CompressionLevel; assert_eq!(super::pre_split_level(CompressionLevel::Fastest), None); assert_eq!(super::pre_split_level(CompressionLevel::Default), None); - assert_eq!(super::pre_split_level(CompressionLevel::Better), None); - assert_eq!(super::pre_split_level(CompressionLevel::Level(7)), None); - assert_eq!(super::pre_split_level(CompressionLevel::Level(11)), Some(0)); - assert_eq!(super::pre_split_level(CompressionLevel::Level(15)), Some(0)); + assert_eq!(super::pre_split_level(CompressionLevel::Level(1)), None); + assert_eq!(super::pre_split_level(CompressionLevel::Level(4)), None); + assert_eq!(super::pre_split_level(CompressionLevel::Level(5)), Some(1)); + assert_eq!(super::pre_split_level(CompressionLevel::Level(7)), Some(2)); + assert_eq!(super::pre_split_level(CompressionLevel::Level(12)), Some(2)); + assert_eq!(super::pre_split_level(CompressionLevel::Level(13)), Some(3)); + assert_eq!(super::pre_split_level(CompressionLevel::Level(15)), Some(3)); assert_eq!(super::pre_split_level(CompressionLevel::Level(16)), Some(4)); assert_eq!(super::pre_split_level(CompressionLevel::Level(22)), Some(4)); } From 6108d5cf0c0c2cfeb732dc5d988e44bdc5a5cdf1 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Thu, 4 Jun 2026 16:41:04 +0300 Subject: [PATCH 03/10] refactor(encode): move block-split level into LevelParams Surface the cheap pre-splitter level as a `LevelParams::pre_split()` knob (the C-like `blockSplitterLevel`), derived per strategy like `backend()` and `parse()`, and expose it via `level_pre_split()`. The frame loop reads it there instead of hardcoding the level-to-split mapping at the call site, so the split config sits next to search/parse/mls in the per-level table. Named presets (Fastest/Default/Better/Best) keep whole blocks: splitting them regressed ratio on tiny highly-compressible frames where the per-sub-block header + entropy-table overhead outweighs the fit gain. Only explicit numeric levels carry the split. 683 tests pass. --- zstd/src/encoding/frame_compressor.rs | 51 ++++++++------------------- zstd/src/encoding/match_generator.rs | 38 ++++++++++++++++++++ 2 files changed, 53 insertions(+), 36 deletions(-) diff --git a/zstd/src/encoding/frame_compressor.rs b/zstd/src/encoding/frame_compressor.rs index 5548606bb..ffa820bd0 100644 --- a/zstd/src/encoding/frame_compressor.rs +++ b/zstd/src/encoding/frame_compressor.rs @@ -405,25 +405,6 @@ fn split_block_from_borders(block: &[u8]) -> usize { } } -fn pre_split_level(level: CompressionLevel) -> Option { - // Mirror the donor `splitLevels[]` table (indexed by strategy) for the - // greedy and lazy bands and above, where the cheap fingerprint - // pre-splitter pays for itself on heterogeneous payloads: - // greedy → 1, lazy/lazy2 → 2, btlazy2/btopt → 3, btultra/btultra2 → 4. - // Levels 1..=4 (fast/dfast) are deliberately left un-split: their - // match-finding is cheap, so the splitter's per-block fingerprint plus - // the extra per-sub-block entropy builds cost more throughput than the - // ratio they buy. The byChunks sampling rate / hashLog tighten with the - // split level (see `split_block_by_chunks`). - match level { - CompressionLevel::Level(5) => Some(1), - CompressionLevel::Level(6..=12) => Some(2), - CompressionLevel::Level(13..=15) => Some(3), - CompressionLevel::Level(16..=22) => Some(4), - _ => None, - } -} - /// XXH64 (low 32 bits, seed 0) over `data`. Shared helper for the /// per-physical-block checksum sidecar so encoder and decoder hash /// the exact same byte ranges with the exact same parameters. Gated @@ -470,7 +451,7 @@ pub(crate) fn optimal_block_size( block_size_max: usize, savings: i64, ) -> usize { - let Some(split_level) = pre_split_level(level) else { + let Some(split_level) = crate::encoding::match_generator::level_pre_split(level) else { return remaining_src_size.min(block_size_max); }; if remaining_src_size < MAX_BLOCK_SIZE as usize || block_size_max < MAX_BLOCK_SIZE as usize { @@ -2895,25 +2876,23 @@ mod tests { ); } - /// `pre_split_level` mirrors the donor `splitLevels[]` table by - /// strategy band: greedy → 1, lazy/lazy2 → 2, btlazy2 → 3, - /// btopt/btultra/btultra2 → 4. Fast/dfast (levels 1..=4) and the named - /// presets stay unsplit so the splitter never runs where its per-block - /// cost would dominate the cheap match-finding. + /// `level_pre_split` resolves the per-level split knob through the + /// `LevelParams` table (donor `splitLevels[]` by strategy): greedy → 1, + /// the lazy band → 2, the btopt/btultra/btultra2 band → 4. Fast/dfast + /// (levels 1..=4) and the speed-first named presets stay unsplit. #[test] fn pre_split_level_dispatches_by_compression_level() { use crate::encoding::CompressionLevel; - assert_eq!(super::pre_split_level(CompressionLevel::Fastest), None); - assert_eq!(super::pre_split_level(CompressionLevel::Default), None); - assert_eq!(super::pre_split_level(CompressionLevel::Level(1)), None); - assert_eq!(super::pre_split_level(CompressionLevel::Level(4)), None); - assert_eq!(super::pre_split_level(CompressionLevel::Level(5)), Some(1)); - assert_eq!(super::pre_split_level(CompressionLevel::Level(7)), Some(2)); - assert_eq!(super::pre_split_level(CompressionLevel::Level(12)), Some(2)); - assert_eq!(super::pre_split_level(CompressionLevel::Level(13)), Some(3)); - assert_eq!(super::pre_split_level(CompressionLevel::Level(15)), Some(3)); - assert_eq!(super::pre_split_level(CompressionLevel::Level(16)), Some(4)); - assert_eq!(super::pre_split_level(CompressionLevel::Level(22)), Some(4)); + use crate::encoding::match_generator::level_pre_split; + assert_eq!(level_pre_split(CompressionLevel::Fastest), None); + assert_eq!(level_pre_split(CompressionLevel::Default), None); + assert_eq!(level_pre_split(CompressionLevel::Level(1)), None); + assert_eq!(level_pre_split(CompressionLevel::Level(4)), None); + assert_eq!(level_pre_split(CompressionLevel::Level(5)), Some(1)); + assert_eq!(level_pre_split(CompressionLevel::Level(7)), Some(2)); + assert_eq!(level_pre_split(CompressionLevel::Level(15)), Some(2)); + assert_eq!(level_pre_split(CompressionLevel::Level(16)), Some(4)); + assert_eq!(level_pre_split(CompressionLevel::Level(22)), Some(4)); } /// End-to-end: a 256 KB heterogeneous payload compressed at diff --git a/zstd/src/encoding/match_generator.rs b/zstd/src/encoding/match_generator.rs index 7b20f858f..ab62ca9c1 100644 --- a/zstd/src/encoding/match_generator.rs +++ b/zstd/src/encoding/match_generator.rs @@ -298,6 +298,25 @@ impl LevelParams { _ => super::strategy::ParseMode::from_lazy_depth(self.lazy_depth), } } + + /// Cheap fingerprint pre-splitter level (donor `splitLevels[]` by + /// strategy), or `None` to keep the whole 128 KiB block. `Fast`/`Dfast` + /// stay un-split: their match-finding is cheap, so the splitter's + /// per-block fingerprint plus extra per-sub-block entropy builds cost + /// more throughput than the ratio they buy. The btopt/btultra/btultra2 + /// band keeps level 4 (matching the pre-existing high-level splitter). + /// This is the C-like `blockSplitterLevel` knob, regulated per level + /// here rather than scattered across the frame loop. + fn pre_split(&self) -> Option { + match self.strategy_tag { + super::strategy::StrategyTag::Fast | super::strategy::StrategyTag::Dfast => None, + super::strategy::StrategyTag::Greedy => Some(1), + super::strategy::StrategyTag::Lazy => Some(2), + super::strategy::StrategyTag::BtOpt + | super::strategy::StrategyTag::BtUltra + | super::strategy::StrategyTag::BtUltra2 => Some(4), + } + } } fn dfast_hash_bits_for_window(max_window_size: usize) -> usize { @@ -540,6 +559,25 @@ fn resolve_level_params(level: CompressionLevel, source_size: Option) -> Le } } +/// The cheap fingerprint pre-splitter level for a compression level (the +/// C-like `blockSplitterLevel`), resolved through the same per-level +/// `LevelParams` table as every other tuning knob. `None` keeps the whole +/// 128 KiB block. The frame loop reads this instead of hardcoding the +/// level→split mapping at the call site. +pub(crate) fn level_pre_split(level: CompressionLevel) -> Option { + // Only the explicit numeric levels carry the split knob; the named + // presets (Fastest/Default/Better/Best/Uncompressed) keep whole blocks. + // Splitting the named presets regressed ratio on highly-compressible + // inputs (the per-sub-block header + entropy-table overhead outweighs + // the fit gain when the whole frame is already a few hundred bytes). + if !matches!(level, CompressionLevel::Level(_)) { + return None; + } + resolve_level_params(level, None) + .pre_split() + .map(usize::from) +} + /// Backend storage for [`MatchGeneratorDriver`]. Exactly one match-finder /// state lives in the driver at a time — the active variant. Backend /// transitions in [`Matcher::reset`] drain the current variant's allocations From 9a6f440888bef7609998055689a9de988834554d Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Thu, 4 Jun 2026 17:00:00 +0300 Subject: [PATCH 04/10] refactor(encode): named levels are pure aliases; scope pre-split to greedy Resolve named presets to their numeric level via `numeric_level()` so the split knob (and the rest of the per-level config) is read uniformly from the `LevelParams` table; named presets are aliases, never a separate path. Removes the earlier named-preset special-case in `level_pre_split`. Scope the cheap pre-splitter to the greedy band (the one whose single-block literal section actually loses to the reference). Lazy stays unsplit: its ratio already tracks the reference, and splitting it regresses on highly-compressible frames until the per-block entropy path reuses tables as aggressively as the reference (tracked separately). Fast/dfast stay unsplit (cheap match-finding). btopt+ keep their existing level-4 split. 683 tests pass, clippy + fmt clean. --- zstd/src/encoding/frame_compressor.rs | 18 ++++++++++++------ zstd/src/encoding/match_generator.rs | 27 ++++++++++++++++----------- zstd/src/encoding/mod.rs | 16 ++++++++++++++++ 3 files changed, 44 insertions(+), 17 deletions(-) diff --git a/zstd/src/encoding/frame_compressor.rs b/zstd/src/encoding/frame_compressor.rs index ffa820bd0..4a6c0f352 100644 --- a/zstd/src/encoding/frame_compressor.rs +++ b/zstd/src/encoding/frame_compressor.rs @@ -2877,20 +2877,26 @@ mod tests { } /// `level_pre_split` resolves the per-level split knob through the - /// `LevelParams` table (donor `splitLevels[]` by strategy): greedy → 1, - /// the lazy band → 2, the btopt/btultra/btultra2 band → 4. Fast/dfast - /// (levels 1..=4) and the speed-first named presets stay unsplit. + /// `LevelParams` table, with named presets as pure numeric aliases: + /// greedy (level 5) → 1, btopt/btultra/btultra2 (16..=22) → 4. Fast, + /// dfast and the lazy band stay unsplit (lazy split is deferred until + /// the per-block entropy path reuses tables like the reference). #[test] fn pre_split_level_dispatches_by_compression_level() { use crate::encoding::CompressionLevel; use crate::encoding::match_generator::level_pre_split; + assert_eq!(level_pre_split(CompressionLevel::Uncompressed), None); assert_eq!(level_pre_split(CompressionLevel::Fastest), None); assert_eq!(level_pre_split(CompressionLevel::Default), None); - assert_eq!(level_pre_split(CompressionLevel::Level(1)), None); + // Better is a pure alias for level 7 (lazy): unsplit, same as Level(7). + assert_eq!( + level_pre_split(CompressionLevel::Better), + level_pre_split(CompressionLevel::Level(7)), + ); assert_eq!(level_pre_split(CompressionLevel::Level(4)), None); assert_eq!(level_pre_split(CompressionLevel::Level(5)), Some(1)); - assert_eq!(level_pre_split(CompressionLevel::Level(7)), Some(2)); - assert_eq!(level_pre_split(CompressionLevel::Level(15)), Some(2)); + assert_eq!(level_pre_split(CompressionLevel::Level(7)), None); + assert_eq!(level_pre_split(CompressionLevel::Level(15)), None); assert_eq!(level_pre_split(CompressionLevel::Level(16)), Some(4)); assert_eq!(level_pre_split(CompressionLevel::Level(22)), Some(4)); } diff --git a/zstd/src/encoding/match_generator.rs b/zstd/src/encoding/match_generator.rs index ab62ca9c1..a8d8b57ea 100644 --- a/zstd/src/encoding/match_generator.rs +++ b/zstd/src/encoding/match_generator.rs @@ -309,9 +309,18 @@ impl LevelParams { /// here rather than scattered across the frame loop. fn pre_split(&self) -> Option { match self.strategy_tag { - super::strategy::StrategyTag::Fast | super::strategy::StrategyTag::Dfast => None, + // Fast/Dfast: cheap match-finding, the splitter costs more than + // it buys. Lazy: its ratio already tracks the reference, and + // splitting it regresses on highly-compressible frames until the + // per-block entropy path reuses tables as aggressively as the + // reference (tracked separately); keep whole blocks for now. + super::strategy::StrategyTag::Fast + | super::strategy::StrategyTag::Dfast + | super::strategy::StrategyTag::Lazy => None, + // Greedy is the band whose single-block literal section loses to + // the reference; the cheap fingerprint pre-split fits per-sub- + // block entropy and recovers it. super::strategy::StrategyTag::Greedy => Some(1), - super::strategy::StrategyTag::Lazy => Some(2), super::strategy::StrategyTag::BtOpt | super::strategy::StrategyTag::BtUltra | super::strategy::StrategyTag::BtUltra2 => Some(4), @@ -565,15 +574,11 @@ fn resolve_level_params(level: CompressionLevel, source_size: Option) -> Le /// 128 KiB block. The frame loop reads this instead of hardcoding the /// level→split mapping at the call site. pub(crate) fn level_pre_split(level: CompressionLevel) -> Option { - // Only the explicit numeric levels carry the split knob; the named - // presets (Fastest/Default/Better/Best/Uncompressed) keep whole blocks. - // Splitting the named presets regressed ratio on highly-compressible - // inputs (the per-sub-block header + entropy-table overhead outweighs - // the fit gain when the whole frame is already a few hundred bytes). - if !matches!(level, CompressionLevel::Level(_)) { - return None; - } - resolve_level_params(level, None) + // Named presets are pure aliases: resolve to their numeric level and + // read the split knob from the same `LevelParams` table as everything + // else. `Uncompressed` (raw blocks) has no numeric equivalent. + let numeric = level.numeric_level()?; + resolve_level_params(CompressionLevel::Level(numeric), None) .pre_split() .map(usize::from) } diff --git a/zstd/src/encoding/mod.rs b/zstd/src/encoding/mod.rs index 1ab620888..94b31357f 100644 --- a/zstd/src/encoding/mod.rs +++ b/zstd/src/encoding/mod.rs @@ -377,6 +377,22 @@ impl CompressionLevel { _ => Self::Level(level), } } + + /// Numeric level a named preset is a shortcut for, so per-level config + /// (the `LevelParams` table) is resolved uniformly: named presets are + /// pure aliases, never a separate config path. `Uncompressed` is the + /// one genuine mode (raw blocks) with no numeric equivalent and returns + /// `None`. + pub(crate) const fn numeric_level(self) -> Option { + match self { + Self::Uncompressed => None, + Self::Fastest => Some(1), + Self::Default => Some(Self::DEFAULT_LEVEL), + Self::Better => Some(7), + Self::Best => Some(11), + Self::Level(n) => Some(n), + } + } } /// Trait used by the encoder that users can use to extend the matching facilities with their own algorithm From 6fca720814afcac24d9a159c409c49e42bf2f7ae Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Thu, 4 Jun 2026 17:07:33 +0300 Subject: [PATCH 05/10] perf(encode): fast-band FSE repeat-mode to make splitting cheap Add the donor `preferRepeat` shortcut to the FSE table selector: for fast/dfast/greedy with a valid (symbol-covering) previous table and fewer than 1000 sequences, reuse it without building a new table or emitting a descriptor. This mirrors `ZSTD_selectEncodingType` (zstd_compress_sequences.c:179-204) and is the per-block cost that made greedy block-splitting expensive: each sub-block was rebuilding and re-emitting all three FSE tables instead of reusing the previous block's. Thread the strategy through both the real encode path and the splitter's size estimator so split decisions match the emitted bytes. 684 tests pass (roundtrip + cross-validation unchanged), clippy + fmt clean. --- zstd/src/encoding/blocks/compressed.rs | 73 +++++++++++++++++++++++++- 1 file changed, 72 insertions(+), 1 deletion(-) diff --git a/zstd/src/encoding/blocks/compressed.rs b/zstd/src/encoding/blocks/compressed.rs index 39cd974e3..ab6bfb77f 100644 --- a/zstd/src/encoding/blocks/compressed.rs +++ b/zstd/src/encoding/blocks/compressed.rs @@ -546,6 +546,7 @@ fn encode_block_parts_with_sequence_scratch( &ll_counts, total, 9, + state.strategy_tag, ); let ml_mode = choose_table_from_counts( state.fse_tables.ml_previous.as_ref(), @@ -553,6 +554,7 @@ fn encode_block_parts_with_sequence_scratch( &ml_counts, total, 9, + state.strategy_tag, ); let of_mode = choose_table_from_counts( state.fse_tables.of_previous.as_ref(), @@ -560,6 +562,7 @@ fn encode_block_parts_with_sequence_scratch( &of_counts, total, 8, + state.strategy_tag, ); writer.write_bits(encode_fse_table_modes(&ll_mode, &ml_mode, &of_mode), 8); @@ -643,6 +646,7 @@ fn estimate_block_parts_size( &mut workspace.ll_counts, &mut workspace.ml_counts, &mut workspace.of_counts, + state.strategy_tag, ) }; @@ -787,6 +791,7 @@ fn estimate_sequences_section_bytes( ll_counts: &mut [usize; 256], ml_counts: &mut [usize; 256], of_counts: &mut [usize; 256], + strategy: crate::encoding::strategy::StrategyTag, ) -> usize { ll_counts.fill(0); ml_counts.fill(0); @@ -810,18 +815,21 @@ fn estimate_sequences_section_bytes( fse_tables.ll_default_ref(), sequences.iter().map(|seq| encode_literal_length(seq.ll).0), 9, + strategy, ); let ml_mode = choose_table( fse_tables.ml_previous.as_ref(), fse_tables.ml_default_ref(), sequences.iter().map(|seq| encode_match_len(seq.ml).0), 9, + strategy, ); let of_mode = choose_table( fse_tables.of_previous.as_ref(), fse_tables.of_default_ref(), sequences.iter().map(|seq| encode_offset(seq.of).0), 8, + strategy, ); let ll_bits_chosen = @@ -1460,6 +1468,7 @@ fn choose_table<'a>( default_table: &'a FSETable, data: impl Iterator, max_log: u8, + strategy: crate::encoding::strategy::StrategyTag, ) -> FseTableMode<'a> { // Collect symbol distribution let mut counts = [0usize; 256]; @@ -1468,7 +1477,7 @@ fn choose_table<'a>( counts[symbol as usize] += 1; total += 1; } - choose_table_from_counts(previous, default_table, &counts, total, max_log) + choose_table_from_counts(previous, default_table, &counts, total, max_log, strategy) } /// Same decision logic as [`choose_table`] but takes pre-computed @@ -1484,6 +1493,7 @@ fn choose_table_from_counts<'a>( counts: &[usize; 256], total: usize, max_log: u8, + strategy: crate::encoding::strategy::StrategyTag, ) -> FseTableMode<'a> { if total == 0 { return FseTableMode::Predefined(default_table); @@ -1508,6 +1518,27 @@ fn choose_table_from_counts<'a>( return FseTableMode::Rle(symbol); } + // Fast-band preferRepeat (donor `ZSTD_selectEncodingType`, + // `zstd_compress_sequences.c:179-204`): for fast/dfast/greedy with a + // valid previous table and `< 1000` sequences, reuse it without building + // a new one. Trades a negligible ratio loss for skipping the per-block + // FSE table build + header descriptor — the dominant per-sub-block cost + // when these cheap-match strategies split a block. The validity probe + // (`fse_bit_cost` is finite) guarantees the previous table covers every + // symbol in this block, so the reuse can never produce an invalid stream. + if matches!( + strategy, + crate::encoding::strategy::StrategyTag::Fast + | crate::encoding::strategy::StrategyTag::Dfast + | crate::encoding::strategy::StrategyTag::Greedy + ) && total < 1000 + && let Some(prev) = previous + && let Some(table) = prev.as_table(default_table) + && fse_bit_cost(counts, max_symbol, table).is_some() + { + return FseTableMode::RepeatLast(prev); + } + let use_low_prob_count = total >= 2048; let new_table = (distinct_symbols > 1).then(|| { build_table_from_symbol_counts(&counts[..=max_symbol], max_log, use_low_prob_count) @@ -2616,23 +2647,29 @@ mod tests { )); let sample_codes = [0u8, 1u8]; + // Lazy is a non-fast-band strategy, so this exercises the cost-based + // repeat decision (not the fast-band shortcut). + let strat = crate::encoding::strategy::StrategyTag::Lazy; let ll_repeat = choose_table( fse_tables.ll_previous.as_ref(), fse_tables.ll_default_ref(), sample_codes.iter().copied(), 9, + strat, ); let ml_repeat = choose_table( fse_tables.ml_previous.as_ref(), fse_tables.ml_default_ref(), sample_codes.iter().copied(), 9, + strat, ); let of_repeat = choose_table( fse_tables.of_previous.as_ref(), fse_tables.of_default_ref(), sample_codes.iter().copied(), 8, + strat, ); assert!(matches!(ll_repeat, FseTableMode::RepeatLast(_))); @@ -2640,6 +2677,38 @@ mod tests { assert!(matches!(of_repeat, FseTableMode::RepeatLast(_))); } + /// Fast-band strategies (fast/dfast/greedy) reuse a covering previous FSE + /// table without building a new one (donor `preferRepeat`), even on a + /// fresh distribution where the cost-based path could pick a new table. + /// A non-fast-band strategy on an identical distribution takes the + /// cost-based path instead. + #[test] + fn fast_band_strategies_prefer_repeat_fse_table() { + use crate::encoding::strategy::StrategyTag; + let prev = build_table_from_symbol_counts(&[8, 1], 9, false); + let previous = PreviousFseTable::Custom(Box::new(prev)); + let fse_tables = FseTables::new(); + // Distribution over symbols {0,1}, both covered by `previous`. + let mut counts = [0usize; 256]; + counts[0] = 4; + counts[1] = 6; + let total = 10; + + // Greedy is fast-band → unconditional reuse of the covering table. + let mode = super::choose_table_from_counts( + Some(&previous), + fse_tables.ll_default_ref(), + &counts, + total, + 9, + StrategyTag::Greedy, + ); + assert!( + matches!(mode, FseTableMode::RepeatLast(_)), + "fast-band greedy must reuse the covering previous table", + ); + } + #[test] fn remember_last_used_tables_reuses_existing_custom_slot_for_repeat() { let mut fse_tables = FseTables::new(); @@ -2676,6 +2745,7 @@ mod tests { fse_tables.ll_default_ref(), core::iter::repeat_n(0u8, 32), 9, + crate::encoding::strategy::StrategyTag::Lazy, ); assert!(matches!(mode, FseTableMode::Rle(0))); } @@ -2688,6 +2758,7 @@ mod tests { &only_zero_one_table, [1u8, 2].into_iter().cycle().take(32), 5, + crate::encoding::strategy::StrategyTag::Lazy, ); assert!(matches!(mode, FseTableMode::Encoded(_))); } From c8ed6016166d8ecd8714f6d8d8789591848d8099 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Thu, 4 Jun 2026 18:11:17 +0300 Subject: [PATCH 06/10] feat(decoding): block-precise error positions (lsm) Add feature-gated `FailedToReadBlockHeaderAt` / `FailedToReadBlockBodyAt` variants to `FrameDecoderError` carrying the failing block's 0-based index and frame-absolute offset (and, for the body, the block's `FrameBlock` metadata reconstructed from its header). Lets consumers doing per-block ECC repair locate and repair exactly the bad block instead of re-fetching the whole frame. New variants are `#[cfg(feature = "lsm")]`; without the feature the legacy positionless variants are unchanged, so the default build's error surface (and the cdylib C-FFI drop-in) stays byte-identical. Coordinates are captured before each block read across all three decode loops (decode_blocks, the streaming loop, and the direct decode_all path); the offset matches the encoder's `FrameEmitInfo.blocks[index].offset_in_frame`. Closes #174. 684 tests (default) + 713 (lsm) pass, clippy + fmt clean both. --- zstd/src/decoding/errors.rs | 54 ++++++++ zstd/src/decoding/frame_decoder.rs | 206 +++++++++++++++++++++++++++-- zstd/src/tests/mod.rs | 15 ++- 3 files changed, 259 insertions(+), 16 deletions(-) diff --git a/zstd/src/decoding/errors.rs b/zstd/src/decoding/errors.rs index d9fcc1819..7831721c6 100644 --- a/zstd/src/decoding/errors.rs +++ b/zstd/src/decoding/errors.rs @@ -618,6 +618,33 @@ pub enum FrameDecoderError { expected: u8, found: Option, }, + /// Block-precise variant of [`Self::FailedToReadBlockHeader`]: a block + /// header read failed and the decoder captured WHERE. `block_index` is + /// the 0-based index of the failing block in the frame; `frame_offset` + /// is the frame-absolute byte offset of that block's `Block_Header` + /// (matches `FrameEmitInfo.blocks[block_index].offset_in_frame` from the + /// encode side). Lets per-block recovery (ECC repair) target the one bad + /// block instead of re-fetching the whole frame. + #[cfg(feature = "lsm")] + FailedToReadBlockHeaderAt { + source: BlockHeaderReadError, + block_index: u32, + frame_offset: u32, + }, + /// Block-precise variant of [`Self::FailedToReadBlockBody`]: a block + /// body decode failed. Carries the same `block_index` / `frame_offset` + /// coordinates plus the failing block's structural metadata + /// ([`FrameBlock`]) reconstructed from its header, so a consumer can + /// locate and repair exactly this block. + /// + /// [`FrameBlock`]: crate::encoding::frame_emit_info::FrameBlock + #[cfg(feature = "lsm")] + FailedToReadBlockBodyAt { + source: DecodeBlockContentError, + block_index: u32, + frame_offset: u32, + block: crate::encoding::frame_emit_info::FrameBlock, + }, } #[cfg(feature = "std")] @@ -629,6 +656,10 @@ impl StdError for FrameDecoderError { FrameDecoderError::DictionaryDecodeError(source) => Some(source), FrameDecoderError::FailedToReadBlockHeader(source) => Some(source), FrameDecoderError::FailedToReadBlockBody(source) => Some(source), + #[cfg(feature = "lsm")] + FrameDecoderError::FailedToReadBlockHeaderAt { source, .. } => Some(source), + #[cfg(feature = "lsm")] + FrameDecoderError::FailedToReadBlockBodyAt { source, .. } => Some(source), FrameDecoderError::FailedToReadChecksum(source) => Some(source), FrameDecoderError::FailedToInitialize(source) => Some(source), FrameDecoderError::FailedToDrainDecodebuffer(source) => Some(source), @@ -663,6 +694,29 @@ impl core::fmt::Display for FrameDecoderError { FrameDecoderError::FailedToReadBlockBody(e) => { write!(f, "Failed to parse block header: {e}") } + #[cfg(feature = "lsm")] + FrameDecoderError::FailedToReadBlockHeaderAt { + source, + block_index, + frame_offset, + } => { + write!( + f, + "Failed to read block header at block {block_index} (frame offset {frame_offset}): {source}" + ) + } + #[cfg(feature = "lsm")] + FrameDecoderError::FailedToReadBlockBodyAt { + source, + block_index, + frame_offset, + .. + } => { + write!( + f, + "Failed to decode block body at block {block_index} (frame offset {frame_offset}): {source}" + ) + } FrameDecoderError::FailedToReadChecksum(e) => { write!(f, "Failed to read checksum: {e}") } diff --git a/zstd/src/decoding/frame_decoder.rs b/zstd/src/decoding/frame_decoder.rs index 93dc557f8..e899d044e 100644 --- a/zstd/src/decoding/frame_decoder.rs +++ b/zstd/src/decoding/frame_decoder.rs @@ -21,6 +21,75 @@ use core::convert::TryInto; use crate::common::MAXIMUM_ALLOWED_WINDOW_SIZE; +/// Build the block-header decode error. With the `lsm` feature it captures +/// the failing block's index and frame offset (block-precise recovery); +/// without it, the legacy positionless variant — so the default build's +/// error surface stays byte-identical to the donor. +#[cfg(feature = "lsm")] +fn block_header_decode_error( + source: crate::decoding::errors::BlockHeaderReadError, + block_index: u32, + frame_offset: u32, +) -> FrameDecoderError { + FrameDecoderError::FailedToReadBlockHeaderAt { + source, + block_index, + frame_offset, + } +} +#[cfg(not(feature = "lsm"))] +fn block_header_decode_error( + source: crate::decoding::errors::BlockHeaderReadError, + _block_index: u32, + _frame_offset: u32, +) -> FrameDecoderError { + FrameDecoderError::FailedToReadBlockHeader(source) +} + +/// Build the block-body decode error. With `lsm` it captures the block +/// index, frame offset, and the failing block's structural metadata +/// (reconstructed from its header); without it, the legacy variant. +#[cfg(feature = "lsm")] +fn block_body_decode_error( + source: DecodeBlockContentError, + block_index: u32, + frame_offset: u32, + header: &crate::blocks::block::BlockHeader, + header_size: u8, +) -> FrameDecoderError { + use crate::blocks::block::BlockType; + // Physical wire body vs the raw `Block_Size` field: RLE writes a single + // body byte while `Block_Size` carries the repeat count; Raw/Compressed + // bodies match the field. + let (body_size, block_size_field) = match header.block_type { + BlockType::RLE => (1u32, header.decompressed_size), + _ => (header.content_size, header.content_size), + }; + FrameDecoderError::FailedToReadBlockBodyAt { + source, + block_index, + frame_offset, + block: crate::encoding::frame_emit_info::FrameBlock { + offset_in_frame: frame_offset, + header_size, + body_size, + block_size_field, + block_type: header.block_type, + last_block: header.last_block, + }, + } +} +#[cfg(not(feature = "lsm"))] +fn block_body_decode_error( + source: DecodeBlockContentError, + _block_index: u32, + _frame_offset: u32, + _header: &crate::blocks::block::BlockHeader, + _header_size: u8, +) -> FrameDecoderError { + FrameDecoderError::FailedToReadBlockBody(source) +} + /// Low level Zstandard decoder that can be used to decompress frames with fine control over when and how many bytes are decoded. /// /// This decoder is able to decode frames only partially and gives control @@ -951,9 +1020,17 @@ impl FrameDecoder { vprintln!("################"); vprintln!("Next Block: {}", state.block_counter); vprintln!("################"); - let (block_header, block_header_size) = block_dec - .read_block_header(&mut source) - .map_err(err::FailedToReadBlockHeader)?; + // Capture the failing-block coordinates BEFORE the header read so + // the error carries where it happened: `bytes_read_counter` is the + // frame-absolute offset of this block's header (not yet advanced), + // `block_counter` its 0-based index. Used by both the header- and + // body-error builders below (block-precise recovery under `lsm`). + let block_index = state.block_counter as u32; + let block_frame_offset = state.bytes_read_counter as u32; + let (block_header, block_header_size) = + block_dec.read_block_header(&mut source).map_err(|source| { + block_header_decode_error(source, block_index, block_frame_offset) + })?; state.bytes_read_counter += u64::from(block_header_size); vprintln!(); @@ -973,7 +1050,15 @@ impl FrameDecoder { let bytes_read_in_block_body = state .decoder_scratch .decode_block_content(&mut block_dec, &block_header, &mut source) - .map_err(err::FailedToReadBlockBody)?; + .map_err(|source| { + block_body_decode_error( + source, + block_index, + block_frame_offset, + &block_header, + block_header_size, + ) + })?; state.bytes_read_counter += bytes_read_in_block_body; // Per-block XXH64 (low 32 bits) of the just-decompressed @@ -1130,9 +1215,13 @@ impl FrameDecoder { if mt_source.len() < 3 { break; } + let block_index = state.block_counter as u32; + let block_frame_offset = state.bytes_read_counter as u32; let (block_header, block_header_size) = block_dec .read_block_header(&mut mt_source) - .map_err(err::FailedToReadBlockHeader)?; + .map_err(|source| { + block_header_decode_error(source, block_index, block_frame_offset) + })?; // check the needed size for the block before updating counters. // If not enough bytes are in the source, the header will have to be read again, so act like we never read it in the first place @@ -1144,7 +1233,15 @@ impl FrameDecoder { let bytes_read_in_block_body = state .decoder_scratch .decode_block_content(&mut block_dec, &block_header, &mut mt_source) - .map_err(err::FailedToReadBlockBody)?; + .map_err(|source| { + block_body_decode_error( + source, + block_index, + block_frame_offset, + &block_header, + block_header_size, + ) + })?; state.bytes_read_counter += bytes_read_in_block_body; state.block_counter += 1; @@ -1691,9 +1788,14 @@ impl FrameDecoder { } else { None }; - let (block_header, hsize) = block_dec - .read_block_header(&mut *input) - .map_err(err::FailedToReadBlockHeader)?; + // Failing-block coordinates captured before the header read (see + // the `decode_blocks` loop for the rationale). + let block_index = state.block_counter as u32; + let block_frame_offset = state.bytes_read_counter as u32; + let (block_header, hsize) = + block_dec.read_block_header(&mut *input).map_err(|source| { + block_header_decode_error(source, block_index, block_frame_offset) + })?; state.bytes_read_counter += u64::from(hsize); // Pre-flight FCS check ONLY for Raw / RLE blocks where // `decompressed_size` is the actual block output size. @@ -1737,7 +1839,15 @@ impl FrameDecoder { .saturating_add(u64::from(block_header.decompressed_size)), }); } - Err(e) => return Err(err::FailedToReadBlockBody(e)), + Err(e) => { + return Err(block_body_decode_error( + e, + block_index, + block_frame_offset, + &block_header, + hsize, + )); + } }; produced = direct.buffer.buffer_ref().tail() as u64; // Post-decode FCS overflow check. @@ -2048,6 +2158,82 @@ mod tests { ); } + /// Block-precise error positions (#174): a failing block header / body + /// reports its 0-based index and frame-absolute offset, consistent with + /// the encoder's `FrameEmitInfo.blocks[index].offset_in_frame`. + #[cfg(feature = "lsm")] + #[test] + fn block_precise_errors_carry_index_and_offset() { + use crate::encoding::{CompressionLevel, FrameCompressor}; + // ~1.3 MiB of incompressible (xorshift) bytes → many 128 KiB raw + // blocks, so blocks 3 and 7 both exist and are not the last block. + let mut data = alloc::vec::Vec::with_capacity(1_300_000); + let mut s: u64 = 0x2545_F491_4F6C_DD1D; + while data.len() < 1_300_000 { + s ^= s << 13; + s ^= s >> 7; + s ^= s << 17; + data.push((s >> 33) as u8); + } + + let mut frame = alloc::vec::Vec::new(); + let blocks = { + let mut fc = FrameCompressor::new(CompressionLevel::Level(1)); + fc.set_source(data.as_slice()); + fc.set_drain(&mut frame); + fc.compress(); + fc.last_frame_emit_info() + .expect("emit info present under lsm") + .blocks + .clone() + }; + assert!(blocks.len() > 7, "need >7 blocks, got {}", blocks.len()); + + let mut out = alloc::vec![0u8; data.len() + 4096]; + + // (1) Corrupt block 7's header: force its Block_Type to Reserved (3) + // by setting both type bits — fails the header read at block 7. + let off7 = blocks[7].offset_in_frame as usize; + let mut corrupt = frame.clone(); + corrupt[off7] |= 0b0000_0110; + let mut dec = FrameDecoder::new(); + let err = dec + .decode_all(&corrupt, &mut out) + .expect_err("reserved block-7 header must fail"); + match err { + super::FrameDecoderError::FailedToReadBlockHeaderAt { + block_index, + frame_offset, + .. + } => { + assert_eq!(block_index, 7); + assert_eq!(frame_offset, blocks[7].offset_in_frame); + } + other => panic!("expected FailedToReadBlockHeaderAt, got {other:?}"), + } + + // (2) Truncate at block 3's body start: header intact, body missing + // → the body decode fails at block 3 with its FrameBlock metadata. + let body3 = blocks[3].offset_in_frame as usize + blocks[3].header_size as usize; + let mut dec = FrameDecoder::new(); + let err = dec + .decode_all(&frame[..body3], &mut out) + .expect_err("truncated block-3 body must fail"); + match err { + super::FrameDecoderError::FailedToReadBlockBodyAt { + block_index, + frame_offset, + block, + .. + } => { + assert_eq!(block_index, 3); + assert_eq!(frame_offset, blocks[3].offset_in_frame); + assert_eq!(block.offset_in_frame, blocks[3].offset_in_frame); + } + other => panic!("expected FailedToReadBlockBodyAt, got {other:?}"), + } + } + #[test] fn decode_all_falls_back_when_output_too_small_for_wildcopy_slack() { // Output sized exactly to frame_content_size (no diff --git a/zstd/src/tests/mod.rs b/zstd/src/tests/mod.rs index e1b75c7fb..8c8759bea 100644 --- a/zstd/src/tests/mod.rs +++ b/zstd/src/tests/mod.rs @@ -532,14 +532,17 @@ fn test_decode_all() { assert_eq!(result, original.len()); assert_eq!(&output[..result], original); - // decode_all with truncated regular frame. + // decode_all with truncated regular frame. Under `lsm` the decoder + // raises the block-precise variant; without it, the legacy one. let mut output = vec![0; original.len()]; let result = decoder.decode_all(&input[..input.len() - 600], &mut output); - assert!( - matches!(result, Err(FrameDecoderError::FailedToReadBlockBody(_))), - "{:?}", - result - ); + let is_block_body_err = match result { + Err(FrameDecoderError::FailedToReadBlockBody(_)) => true, + #[cfg(feature = "lsm")] + Err(FrameDecoderError::FailedToReadBlockBodyAt { .. }) => true, + _ => false, + }; + assert!(is_block_body_err, "{result:?}"); // decode_all with truncated skip frame. let mut output = vec![0; original.len()]; From a831ff79824338b65b2b32dfc0d4366f2f336222 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Thu, 4 Jun 2026 18:49:37 +0300 Subject: [PATCH 07/10] feat(encode): add reusable compression context (CCtx-equivalent) Add `FrameCompressor::compress_independent_frame{,_into}` for emitting N independent, self-describing frames from one reused compressor, mirroring C `ZSTD_CCtx` + `ZSTD_compress2`: - one independent frame per call (own header/FCS/checksum, no cross-frame match history), reusing the matcher tables, scratch, FSE/Huffman seeds, and any sticky dictionary so per-frame setup is paid once, not N times - the `_into` form also reuses the caller's output buffer, matching C's caller-owned `dst` (no per-frame output allocation) - input is read in place; its lifetime is not baked into the compressor type, so successive calls may pass slices with unrelated lifetimes Refactor to share one frame pipeline: - `run_owned_block_loop` reads from a caller-supplied source so the streaming and slice paths share it without a lifetime-bound reader field - extract `run_one_frame` (borrowed/owned dispatch), `build_frame_header`, `write_frame_to_vec`, and `populate_frame_emit_info` from `finish_frame` - `compress_slice_to_vec` now wraps `compress_independent_frame`, dropping the internal `compress_oneshot_borrowed` and `compress_bound` helpers - `FrameCompressor` gains default type params so the bare type names the reusable shape Tests: reuse emits byte-identical frames to a fresh compressor across Fast (borrowed) and owned backends, the `_into` buffer is replaced not appended, and a sticky dictionary is re-primed into every frame. Part of #316 --- zstd/src/encoding/frame_compressor.rs | 613 ++++++++++++++++++-------- zstd/src/encoding/mod.rs | 183 ++------ 2 files changed, 466 insertions(+), 330 deletions(-) diff --git a/zstd/src/encoding/frame_compressor.rs b/zstd/src/encoding/frame_compressor.rs index 4a6c0f352..c4f174182 100644 --- a/zstd/src/encoding/frame_compressor.rs +++ b/zstd/src/encoding/frame_compressor.rs @@ -84,7 +84,11 @@ impl EncoderDictionary { /// // `compress` writes the compressed output into the provided buffer. /// compressor.compress(); /// ``` -pub struct FrameCompressor { +pub struct FrameCompressor< + R: Read = &'static [u8], + W: Write = Vec, + M: Matcher = MatchGeneratorDriver, +> { uncompressed_data: Option, compressed_data: Option, compression_level: CompressionLevel, @@ -575,64 +579,125 @@ impl FrameCompressor { } } - /// One-shot compress of a contiguous `&[u8]` input. When the Fast - /// (Simple) backend is selected and no dictionary is active, the - /// matcher references the input in place as a borrowed window — - /// skipping the per-block copy into the owned `history` that the - /// streaming path performs (the dominant peak-allocation cost on Fast - /// one-shot compress). Over-window inputs are included: the borrowed - /// scan bounds matches with the same `window_low = block_end - - /// advertised_window` the owned (evicting) path uses, so it produces - /// byte-identical output without ever copying the input. Non-Fast / - /// dictionary / `Uncompressed` cases fall back to the owned loop. + /// Whether the borrowed (no per-block history copy) one-shot loop is + /// valid for an `input_len`-byte slice under the resolved `prep`. /// - /// Crate-internal: the only caller is [`crate::encoding::compress_slice_to_vec`], - /// which passes the SAME slice to both [`Self::set_source`] (for the - /// owned fallback) and this method. Not `pub` because the contract — - /// `input` must equal the configured source, and a source must be set - /// for the fallback — is a footgun for external callers who could pass - /// a mismatched slice and silently compress the wrong data. - pub(crate) fn compress_oneshot_borrowed(&mut self, input: &[u8]) { + /// `Uncompressed` resolves to `StrategyTag::Fast` but must emit stored + /// Raw blocks, which the borrowed loop's + /// `compress_block_encoded_borrowed` (RLE/raw-fast/compressed) does NOT + /// do, so exclude it; it then takes the owned path's dedicated + /// Uncompressed arm. + /// + /// No window-size gate: over-window inputs are handled too. The owned + /// path bounds matches to the last `advertised_window` bytes via + /// `window_low` and evicts/rehashes its history; the borrowed path + /// computes the identical `window_low = block_end - advertised_window` + /// and the kernel rejects any hash candidate below it, while the + /// per-position `put` during the scan keeps in-window slots current, + /// so it produces byte-identical output to the owned (evicting) path + /// without ever copying the input into `history`, even when the input + /// far exceeds the window. + /// + /// BUT gate on `input_len <= u32::MAX`: the Fast kernel stores ABSOLUTE + /// positions in a `u32` hash table, and the borrowed scan walks + /// absolute input offsets up to `block_end == input.len()`. Past 4 GiB + /// those offsets truncate / overflow the `u32` position math + /// (`base_off + ip0 as u32`, `window_low`), panicking or corrupting. + /// The owned/evicting path keeps the scanned window bounded (positions + /// stay small), so >4 GiB inputs fall back to it. + fn borrowed_eligible(&self, input_len: usize, prep: &FramePrep) -> bool { use crate::encoding::strategy::StrategyTag; - // Derive frame sizing from the actual payload, not whatever hint a - // previous call on a reused compressor left behind — a stale hint - // would change the resolved window/header and could even flip - // `borrowed_eligible` for this slice. - self.source_size_hint = Some(input.len() as u64); - let prep = self.prepare_frame(); - // `Uncompressed` resolves to `StrategyTag::Fast` but must emit - // stored Raw blocks, which the borrowed loop's - // `compress_block_encoded_borrowed` (RLE/raw-fast/compressed) - // does NOT do — exclude it so it takes the owned path's dedicated - // Uncompressed arm. - // - // No window-size gate: over-window inputs are handled too. The - // owned path bounds matches to the last `advertised_window` bytes - // via `window_low` and evicts/rehashes its history; the borrowed - // path computes the identical `window_low = block_end - - // advertised_window` and the kernel rejects any hash candidate - // below it, while the per-position `put` during the scan keeps - // in-window slots current — so it produces byte-identical output - // to the owned (evicting) path without ever copying the input - // into `history`, even when the input far exceeds the window. - // - // BUT gate on `input.len() <= u32::MAX`: the Fast kernel stores - // ABSOLUTE positions in a `u32` hash table, and the borrowed scan - // walks absolute input offsets up to `block_end == input.len()`. - // Past 4 GiB those offsets truncate / overflow the `u32` position - // math (`base_off + ip0 as u32`, `window_low`), panicking or - // corrupting. The owned/evicting path keeps the scanned window - // bounded (positions stay small), so >4 GiB inputs fall back to it. - let borrowed_eligible = !prep.use_dictionary_state + !prep.use_dictionary_state && !matches!(self.compression_level, CompressionLevel::Uncompressed) && self.state.strategy_tag == StrategyTag::Fast - && input.len() <= u32::MAX as usize; - let (all_blocks, total_uncompressed) = if borrowed_eligible { + && input_len <= u32::MAX as usize + } + + /// Compress `input` as one frame's worth of blocks: the borrowed + /// in-place loop when [`Self::borrowed_eligible`], else the owned + /// (history-copying) loop fed an in-place `&[u8]` cursor. Returns + /// `(all_blocks, total_uncompressed)`; the caller emits the frame tail + /// (`finish_frame` for a configured drain, `write_frame_to_vec` for a + /// returned buffer). + fn run_one_frame(&mut self, input: &[u8], prep: &FramePrep) -> (Vec, u64) { + if self.borrowed_eligible(input.len(), prep) { self.run_borrowed_block_loop(input, prep.initial_size_hint) } else { - self.run_owned_block_loop(prep.initial_size_hint) - }; - self.finish_frame(all_blocks, total_uncompressed, &prep); + let mut cursor: &[u8] = input; + self.run_owned_block_loop(&mut cursor, prep.initial_size_hint) + } + } + + /// Compress one contiguous `&[u8]` as a single independent Zstd frame, + /// writing the frame bytes into `out` (its previous contents are + /// replaced and its allocation reused), reusing this compressor's heavy + /// state across calls. + /// + /// This is the reusable-compression-context (CCtx-equivalent) entry + /// point, mirroring C `ZSTD_compress2` over a reused `ZSTD_CCtx`: + /// construct ONE `FrameCompressor` and call this in a loop to emit N + /// independent, self-describing frames (each carrying its own header, + /// blocks, and checksum, decodable in isolation, with no cross-frame + /// match history). Every call resets the per-frame state via + /// [`Self::prepare_frame`]: only the allocations are kept, so the + /// dominant per-frame setup cost (table allocation + dictionary prime) + /// is paid once instead of N times. Passing the same `out` buffer each + /// call additionally reuses the output allocation, matching C's + /// caller-owned `dst` buffer (no per-frame output allocation). + /// + /// Reusing the context + `out` across many small frames (the typical + /// per-block-frame workload) is far cheaper than a fresh + /// [`compress_slice_to_vec`](crate::encoding::compress_slice_to_vec) + /// per block, which allocates and primes from scratch each time. + /// + /// The input is read in place: no [`Self::set_source`] / + /// [`Self::set_drain`] setup is required, and the input lifetime is not + /// baked into the compressor type, so successive calls may pass slices + /// with unrelated lifetimes. When the Fast (Simple) backend is active + /// and no dictionary is set, the matcher references the input directly + /// (no per-block history copy); other backends / dictionary use copy + /// each block into history exactly as the streaming + /// [`compress`](Self::compress) path does. The source-size hint is + /// derived from the input length on every call, so per-frame table + /// sizing tracks each frame's actual size regardless of any earlier + /// hint. + /// + /// A sticky dictionary set via + /// [`set_dictionary`](Self::set_dictionary) (or its variants) is primed + /// into every frame, mirroring `ZSTD_CCtx_loadDictionary` / + /// `ZSTD_CCtx_refCDict`. + /// + /// # Panics + /// + /// Panics on encoder error, matching [`Self::compress`] and + /// [`compress_slice_to_vec`](crate::encoding::compress_slice_to_vec). + pub fn compress_independent_frame_into(&mut self, input: &[u8], out: &mut Vec) { + // Size the next frame from the actual payload, not a stale hint a + // previous call may have left behind (a wrong hint would change the + // 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); + self.write_frame_to_vec(out, all_blocks, total_uncompressed, &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 + /// buffer across frames (the CCtx-equivalent zero-per-call-alloc + /// output, matching C's caller-owned `dst`). + /// + /// ```rust + /// use structured_zstd::encoding::{FrameCompressor, CompressionLevel}; + /// let mut cctx: FrameCompressor = FrameCompressor::new(CompressionLevel::Default); + /// let frame_a = cctx.compress_independent_frame(b"first block payload"); + /// let frame_b = cctx.compress_independent_frame(b"second block payload"); + /// assert!(!frame_a.is_empty() && !frame_b.is_empty()); + /// ``` + pub fn compress_independent_frame(&mut self, input: &[u8]) -> Vec { + let mut out = Vec::new(); + self.compress_independent_frame_into(input, &mut out); + out } /// Borrowed one-shot block loop: walks `input` in `MAX_BLOCK_SIZE` @@ -799,7 +864,17 @@ impl FrameCompressor { /// same reset / dict-prime / entropy-seed setup and frame tail. pub fn compress(&mut self) { let prep = self.prepare_frame(); - let (all_blocks, total_uncompressed) = self.run_owned_block_loop(prep.initial_size_hint); + // Take the reader out so `run_owned_block_loop` can borrow it + // mutably alongside `&mut self` (the rest of the loop touches + // `self.state` / `self.hasher`, disjoint from the reader). Restored + // before the frame tail so a reused compressor keeps its source. + let mut source = self + .uncompressed_data + .take() + .expect("source must be set via set_source before compress()"); + let (all_blocks, total_uncompressed) = + self.run_owned_block_loop(&mut source, prep.initial_size_hint); + self.uncompressed_data = Some(source); self.finish_frame(all_blocks, total_uncompressed, &prep); } @@ -913,13 +988,21 @@ impl FrameCompressor { } } - /// Owned streaming block loop: reads blocks from the source `Read`, - /// optionally pre-splits, hashes for the content checksum, and emits - /// each block via `compress_block_encoded`, accumulating the block - /// bytes. Returns `(all_blocks, total_uncompressed)`. Shared by - /// `compress` and the borrowed one-shot path's fallback. - fn run_owned_block_loop(&mut self, initial_size_hint: Option) -> (Vec, u64) { - let source = self.uncompressed_data.as_mut().unwrap(); + /// Owned streaming block loop: reads blocks from the caller-provided + /// `source` reader, optionally pre-splits, hashes for the content + /// checksum, and emits each block via `compress_block_encoded`, + /// accumulating the block bytes. Returns `(all_blocks, + /// total_uncompressed)`. The source is passed in (rather than read + /// from `self.uncompressed_data`) so the streaming `compress` path can + /// feed the configured reader while the slice paths + /// (`compress_oneshot_borrowed`, `compress_independent_frame`) feed an + /// in-place `&[u8]` cursor without baking its lifetime into the + /// compressor type. + fn run_owned_block_loop( + &mut self, + source: &mut Rd, + initial_size_hint: Option, + ) -> (Vec, u64) { // Accumulate all compressed blocks; the frame header is written // after all input has been read so Frame_Content_Size is known. // Seed capacity by source-size hint — see `initial_all_blocks_cap`. @@ -1071,27 +1154,22 @@ impl FrameCompressor { (all_blocks, total_uncompressed) } - /// Write the frame header (with now-known FCS / single_segment), the - /// accumulated block bytes, and the optional trailing content - /// checksum; populate `frame_emit_info` (lsm). Shared by `compress` - /// and the borrowed one-shot path. - fn finish_frame(&mut self, all_blocks: Vec, total_uncompressed: u64, prep: &FramePrep) { - let window_size = prep.window_size; - let use_dictionary_state = prep.use_dictionary_state; - let source_size_hint_known = prep.source_size_hint_known; - let drain = self.compressed_data.as_mut().unwrap(); - // Now that total_uncompressed is known, write the frame header with FCS. + /// Build the frame header bytes once the total payload size is known + /// (so `Frame_Content_Size` / `single_segment` can be set). Shared by + /// the drain (`finish_frame`) and returned-buffer + /// (`finish_frame_to_vec`) tails. + fn build_frame_header(&self, total_uncompressed: u64, prep: &FramePrep) -> Vec { // Match the donor framing policy for pledged one-shot inputs: use a // single-segment frame whenever the source fits the active window. - let single_segment = !use_dictionary_state - && source_size_hint_known + let single_segment = !prep.use_dictionary_state + && prep.source_size_hint_known && total_uncompressed >= 512 - && total_uncompressed <= window_size; + && total_uncompressed <= prep.window_size; let header = FrameHeader { frame_content_size: Some(total_uncompressed), single_segment, content_checksum: cfg!(feature = "hash"), - dictionary_id: if use_dictionary_state { + dictionary_id: if prep.use_dictionary_state { self.dictionary.as_ref().map(|dict| dict.inner.id as u64) } else { None @@ -1099,137 +1177,169 @@ impl FrameCompressor { window_size: if single_segment { None } else { - Some(window_size) + Some(prep.window_size) }, magicless: self.magicless, }; - // Write the frame header and compressed blocks separately to avoid - // shifting the entire `all_blocks` buffer to prepend the header. let mut header_buf: Vec = Vec::with_capacity(14); header.serialize(&mut header_buf); + header_buf + } + + /// Write the frame header, accumulated block bytes, and optional + /// trailing content checksum to the configured drain; populate + /// `frame_emit_info` (lsm). Header and blocks are written separately to + /// avoid shifting `all_blocks` to prepend the header. Used by + /// `compress` and `compress_oneshot_borrowed`. + fn finish_frame(&mut self, all_blocks: Vec, total_uncompressed: u64, prep: &FramePrep) { + let header_buf = self.build_frame_header(total_uncompressed, prep); + // Snapshot the checksum before borrowing the drain field so the + // `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 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, then `content_checksum` is set to true in the header - // and a 32 bit hash is written at the end of the data. + // 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. #[cfg(feature = "hash")] - { - // Because we only have the data as a reader, we need to read all of it to calculate the checksum - // Possible TODO: create a wrapper around self.uncompressed data that hashes the data as it's read? - let content_checksum = self.hasher.finish(); - drain - .write_all(&(content_checksum as u32).to_le_bytes()) - .unwrap(); - } - - // FrameEmitInfo population (lsm feature): walk all_blocks to - // recover per-block layout. Each Block_Header is 3 bytes LE - // packing `(block_size << 3) | (block_type << 1) | last_block`. - // Physical body size differs by type: RLE bodies are always 1 - // byte (the repeated byte), Raw/Compressed bodies span - // `block_size` bytes. + drain.write_all(&checksum_bytes).unwrap(); #[cfg(feature = "lsm")] - { - use crate::blocks::block::BlockType as BT; - use crate::encoding::frame_emit_info::{FrameBlock, FrameEmitInfo}; - // All frame-offset arithmetic below is bounded by u32 on - // the wire (Block_Size is a 21-bit field, frames bounded - // by MAX_BLOCK_SIZE * #blocks). A pathologically large - // frame whose total emitted size exceeds u32::MAX would - // overflow the cast — bail out by leaving - // `frame_emit_info` at `None` rather than handing the - // caller a silently-truncated layout. Checked once for - // header / all_blocks / cursor up front + once per push; - // the overflow path is statically unreachable on every - // realistic frame so the predictor amortises the branch - // to zero cost on the hot path. - let frame_header_len: u32 = match u32::try_from(header_buf.len()) { - Ok(v) => v, - Err(_) => return, + self.populate_frame_emit_info(header_buf.len(), &all_blocks); + } + + /// Assemble the frame (header + blocks + optional checksum) into the + /// caller-provided `out` buffer, replacing its contents, and populate + /// `frame_emit_info` (lsm). `out` is cleared first (its allocation is + /// reused, the CCtx-equivalent zero-per-call-alloc output path) then + /// grown once to the exact frame size. Used by + /// `compress_independent_frame_into`. The single `all_blocks` copy into + /// `out` is the same one copy `finish_frame` performs writing + /// `all_blocks` into a `Vec` drain, no extra buffering vs the drain + /// path. + fn write_frame_to_vec( + &mut self, + out: &mut Vec, + all_blocks: Vec, + total_uncompressed: u64, + prep: &FramePrep, + ) { + let header_buf = self.build_frame_header(total_uncompressed, prep); + let checksum_len = if cfg!(feature = "hash") { 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()); + #[cfg(feature = "lsm")] + self.populate_frame_emit_info(header_buf.len(), &all_blocks); + } + + /// Walk `all_blocks` to recover per-block layout and store it in + /// `frame_emit_info`. Each Block_Header is 3 bytes LE packing + /// `(block_size << 3) | (block_type << 1) | last_block`. Physical body + /// size differs by type: RLE bodies are always 1 byte (the repeated + /// byte), Raw/Compressed bodies span `block_size`. `header_len` is the + /// serialized frame-header length (frame offset of the first block). + #[cfg(feature = "lsm")] + fn populate_frame_emit_info(&mut self, header_len: usize, all_blocks: &[u8]) { + use crate::blocks::block::BlockType as BT; + use crate::encoding::frame_emit_info::{FrameBlock, FrameEmitInfo}; + // All frame-offset arithmetic below is bounded by u32 on the wire + // (Block_Size is a 21-bit field, frames bounded by MAX_BLOCK_SIZE * + // #blocks). A pathologically large frame whose total emitted size + // exceeds u32::MAX would overflow the cast; bail out by leaving + // `frame_emit_info` at `None` rather than handing the caller a + // silently-truncated layout. The overflow path is statically + // unreachable on every realistic frame so the predictor amortises + // the branch to zero cost. + let frame_header_len: u32 = match u32::try_from(header_len) { + Ok(v) => v, + Err(_) => return, + }; + let all_blocks_len_u32: u32 = match u32::try_from(all_blocks.len()) { + Ok(v) => v, + Err(_) => return, + }; + let mut blocks: Vec = Vec::new(); + let mut cursor: usize = 0; + while cursor + 3 <= all_blocks.len() { + let mut header_u32 = [0u8; 4]; + header_u32[..3].copy_from_slice(&all_blocks[cursor..cursor + 3]); + let raw = u32::from_le_bytes(header_u32); + let last_block = (raw & 1) != 0; + let block_type = match (raw >> 1) & 0b11 { + 0 => BT::Raw, + 1 => BT::RLE, + 2 => BT::Compressed, + _ => BT::Reserved, + }; + let block_size_field = raw >> 3; + // RLE bodies are always 1 byte physical on the wire (the single + // repeated byte); the spec's Block_Size field carries the + // logical repeat count. Raw and Compressed bodies physically + // span block_size_field bytes. Store the physical length in + // body_size so the 'offset + header + body_size' arithmetic + // always lands on the next block boundary, and surface the raw + // spec field separately as block_size_field. + let physical_body: u32 = match block_type { + BT::RLE => 1, + _ => block_size_field, }; - let all_blocks_len_u32: u32 = match u32::try_from(all_blocks.len()) { + let cursor_u32: u32 = match u32::try_from(cursor) { Ok(v) => v, Err(_) => return, }; - let mut blocks: Vec = Vec::new(); - let mut cursor: usize = 0; - while cursor + 3 <= all_blocks.len() { - let mut header_u32 = [0u8; 4]; - header_u32[..3].copy_from_slice(&all_blocks[cursor..cursor + 3]); - let raw = u32::from_le_bytes(header_u32); - let last_block = (raw & 1) != 0; - let block_type = match (raw >> 1) & 0b11 { - 0 => BT::Raw, - 1 => BT::RLE, - 2 => BT::Compressed, - _ => BT::Reserved, - }; - let block_size_field = raw >> 3; - // RLE bodies are always 1 byte physical on the wire - // (the single repeated byte); the spec's Block_Size - // field carries the logical repeat count. Raw and - // Compressed bodies physically span block_size_field - // bytes. Store the physical length in body_size so the - // 'offset + header + body_size' arithmetic always - // lands on the next block boundary, and surface the - // raw spec field separately as block_size_field. - let physical_body: u32 = match block_type { - BT::RLE => 1, - _ => block_size_field, - }; - let cursor_u32: u32 = match u32::try_from(cursor) { - Ok(v) => v, - Err(_) => return, - }; - let offset_in_frame = match frame_header_len.checked_add(cursor_u32) { - Some(v) => v, - None => return, - }; - blocks.push(FrameBlock { - offset_in_frame, - header_size: 3, - body_size: physical_body, - block_size_field, - block_type, - last_block, - }); - cursor += 3 + physical_body as usize; - if last_block { - break; - } - } - let checksum_range = if cfg!(feature = "hash") { - let cs_start = match frame_header_len.checked_add(all_blocks_len_u32) { - Some(v) => v, - None => return, - }; - let cs_end = match cs_start.checked_add(4) { - Some(v) => v, - None => return, - }; - Some(cs_start..cs_end) - } else { - None - }; - let body_total = match frame_header_len.checked_add(all_blocks_len_u32) { + let offset_in_frame = match frame_header_len.checked_add(cursor_u32) { Some(v) => v, None => return, }; - let total_size = if checksum_range.is_some() { - match body_total.checked_add(4) { - Some(v) => v, - None => return, - } - } else { - body_total - }; - self.frame_emit_info = Some(FrameEmitInfo { - frame_header_range: 0..frame_header_len, - blocks, - checksum_range, - total_size, + blocks.push(FrameBlock { + offset_in_frame, + header_size: 3, + body_size: physical_body, + block_size_field, + block_type, + last_block, }); + cursor += 3 + physical_body as usize; + if last_block { + break; + } } + let checksum_range = if cfg!(feature = "hash") { + let cs_start = match frame_header_len.checked_add(all_blocks_len_u32) { + Some(v) => v, + None => return, + }; + let cs_end = match cs_start.checked_add(4) { + Some(v) => v, + None => return, + }; + Some(cs_start..cs_end) + } else { + None + }; + let body_total = match frame_header_len.checked_add(all_blocks_len_u32) { + Some(v) => v, + None => return, + }; + let total_size = if checksum_range.is_some() { + match body_total.checked_add(4) { + Some(v) => v, + None => return, + } + } else { + body_total + }; + self.frame_emit_info = Some(FrameEmitInfo { + frame_header_range: 0..frame_header_len, + blocks, + checksum_range, + total_size, + }); } /// Layout of the most recently emitted frame. @@ -3057,4 +3167,131 @@ mod tests { ), } } + + /// A reused `FrameCompressor` must emit byte-identical frames to a + /// fresh compressor per input across both the borrowed (Fast) and + /// owned (Dfast/Lazy/Greedy/Uncompressed) backends. This proves + /// `prepare_frame` fully resets the per-frame state (matcher window, + /// content hasher, FSE/Huffman seeds) between independent frames; a + /// missed reset would corrupt frame N>=2's header checksum or matches. + /// Each emitted frame must also round-trip. + #[test] + fn compress_independent_frame_reuse_matches_fresh_and_roundtrips() { + use crate::encoding::{CompressionLevel, compress_slice_to_vec}; + let levels = [ + CompressionLevel::Uncompressed, + CompressionLevel::Fastest, + CompressionLevel::Default, + CompressionLevel::Better, + CompressionLevel::Best, + CompressionLevel::Level(5), + ]; + let inputs: Vec> = vec![ + Vec::new(), + vec![0x00], + b"the quick brown fox jumps over the lazy dog\n".to_vec(), + vec![0x7Eu8; 50_000], // highly compressible + generate_data(0xABCD, 70_000), // pseudo-random + generate_data(0x1234, 200_000), + ]; + for level in levels { + let mut cctx: FrameCompressor = FrameCompressor::new(level); + for data in &inputs { + let reused = cctx.compress_independent_frame(data); + let fresh = compress_slice_to_vec(data, level); + assert_eq!( + reused, + fresh, + "reused frame != fresh frame for len={} level={:?}", + data.len(), + level, + ); + let mut decoder = FrameDecoder::new(); + let mut decoded = Vec::with_capacity(data.len()); + decoder.decode_all_to_vec(&reused, &mut decoded).unwrap(); + assert_eq!( + decoded, + *data, + "roundtrip failed for len={} level={:?}", + data.len(), + level, + ); + } + } + } + + /// `compress_independent_frame_into` must replace (not append to) the + /// caller's buffer each call, so a smaller frame after a larger one + /// yields exactly the smaller frame, and the reused buffer's content + /// matches a fresh compression of the same input. + #[test] + fn compress_independent_frame_into_replaces_buffer_contents() { + use crate::encoding::{CompressionLevel, compress_slice_to_vec}; + let large = vec![0x11u8; 40_000]; + let small = b"short payload".to_vec(); + let mut cctx: FrameCompressor = FrameCompressor::new(CompressionLevel::Default); + let mut out = Vec::new(); + cctx.compress_independent_frame_into(&large, &mut out); + let frame_large = out.clone(); + // Reusing the same buffer for a smaller frame must clear it first. + cctx.compress_independent_frame_into(&small, &mut out); + assert_eq!( + out, + compress_slice_to_vec(&small, CompressionLevel::Default), + "reused buffer must hold exactly the second frame", + ); + // The first frame, captured before reuse, still round-trips. + let mut decoder = FrameDecoder::new(); + let mut decoded = Vec::with_capacity(large.len()); + decoder + .decode_all_to_vec(&frame_large, &mut decoded) + .unwrap(); + assert_eq!(decoded, large); + } + + /// A sticky dictionary set once on a reused compressor must be primed + /// into every independent frame (mirroring `ZSTD_CCtx_loadDictionary`): + /// each frame decodes with the dictionary and is byte-identical to a + /// fresh compressor carrying the same dictionary. This proves + /// `prepare_frame` re-primes the dictionary (matcher content + offset + /// history + entropy seed) every call rather than only on the first. + #[test] + fn compress_independent_frame_reuses_sticky_dictionary() { + use crate::encoding::CompressionLevel; + let dict_raw = include_bytes!("../../dict_tests/dictionary"); + let dict_content = crate::decoding::Dictionary::decode_dict(dict_raw).unwrap(); + let mut payload_a = Vec::new(); + for _ in 0..8 { + payload_a.extend_from_slice(&dict_content.dict_content[..2048]); + } + let payload_b = b"a different second frame payload, still dict-attached".to_vec(); + let inputs = [payload_a, payload_b]; + + let mut cctx: FrameCompressor = FrameCompressor::new(CompressionLevel::Fastest); + cctx.set_dictionary_from_bytes(dict_raw) + .expect("dictionary bytes should parse"); + + for data in &inputs { + let reused = cctx.compress_independent_frame(data); + // Fresh compressor carrying the same sticky dictionary. + let mut fresh_enc: FrameCompressor = FrameCompressor::new(CompressionLevel::Fastest); + fresh_enc + .set_dictionary_from_bytes(dict_raw) + .expect("dictionary bytes should parse"); + let fresh = fresh_enc.compress_independent_frame(data); + assert_eq!( + reused, + fresh, + "reused dict frame != fresh dict frame, len={}", + data.len(), + ); + // Round-trip with the dictionary on the decode side. + let dict_for_decoder = crate::decoding::Dictionary::decode_dict(dict_raw).unwrap(); + let mut decoder = FrameDecoder::new(); + decoder.add_dict(dict_for_decoder).unwrap(); + let mut decoded = Vec::with_capacity(data.len()); + decoder.decode_all_to_vec(&reused, &mut decoded).unwrap(); + assert_eq!(&decoded, data, "dict roundtrip failed, len={}", data.len()); + } + } } diff --git a/zstd/src/encoding/mod.rs b/zstd/src/encoding/mod.rs index 94b31357f..58bada471 100644 --- a/zstd/src/encoding/mod.rs +++ b/zstd/src/encoding/mod.rs @@ -98,55 +98,6 @@ use alloc::vec::Vec; pub(crate) const BETTER_WINDOW_LOG: u8 = 23; -/// Worst-case compressed-size bound for `src_size` uncompressed -/// bytes. Mirrors the upstream `ZSTD_COMPRESSBOUND(srcSize)` macro -/// from `lib/zstd.h` (zstd 1.5.7) — the C *macro*, NOT the public -/// `ZSTD_compressBound()` function. The function adds frame/block- -/// header headroom in some code paths; the macro is a tighter -/// expression suitable for sizing an internal output Vec, which is -/// the only call site here. Callers that need a hard upper bound on -/// the wire-format compressed size should NOT reuse this — use the -/// `zstd-sys` function call directly. If the macro changes in a -/// future upstream revision, treat the upstream header as -/// authoritative and re-derive this function — do not trust the -/// inline copy without cross-checking. -/// -/// Inline approximation (this revision): -/// -/// ```text -/// srcSize -/// + (srcSize >> 8) -/// + (srcSize < 128 KiB ? ((128 KiB - srcSize) >> 11) : 0) -/// ``` -/// -/// Consulted by [`compress_slice_to_vec`] for the small-input branch -/// of its output-`Vec` seed: the seed is -/// `min(compress_bound(src), OUTPUT_BLOCK_CAP = 128 KiB)`, so for -/// inputs that fit within one donor block (`≤ ~128 KiB`) the -/// destination never reallocates inside the measured window. Without -/// the bound, `Vec::push` growth doubles in powers of two and pins -/// `~2 × final_compressed_size` resident at the last realloc, which -/// shows up as ~1 MiB of peak-RSS noise on 1 MiB inputs and prevents -/// FFI-parity on the memory bench. For inputs larger than the cap the -/// `Vec` still grows by amortized doubling — see -/// [`compress_slice_to_vec`] for the trade-off rationale. -#[inline] -pub(crate) const fn compress_bound(src_size: usize) -> usize { - const SMALL_INPUT_THRESHOLD: usize = 128 * 1024; - let tail = if src_size < SMALL_INPUT_THRESHOLD { - (SMALL_INPUT_THRESHOLD - src_size) >> 11 - } else { - 0 - }; - // `saturating_add` guards against the corner case where `src_size` - // is close to `usize::MAX`. The current sole caller - // (`compress_slice_to_vec`) clamps the result with `.min(OUTPUT_BLOCK_CAP)` - // so an overflow there would still produce a sensible cap, but - // future `pub(crate)` callers may use the raw bound — keep the - // arithmetic itself overflow-safe. - src_size.saturating_add(src_size >> 8).saturating_add(tail) -} - /// Convenience function to compress some source into a target without reusing any resources of the compressor /// ```rust /// use structured_zstd::encoding::{compress, CompressionLevel}; @@ -168,46 +119,23 @@ pub fn compress(source: R, target: W, level: CompressionLevel /// therefore be roughly `input_size + output_size`. For very large payloads or /// tighter memory budgets, prefer streaming APIs such as [`StreamingEncoder`]. /// -/// **Peak-memory shape change in this revision.** The implementation -/// delegates to [`compress_slice_to_vec`], which seeds the output -/// `Vec` with `min(compress_bound(input.len()), OUTPUT_BLOCK_CAP = -/// 128 KiB)` instead of the previous `Vec::new()` (zero-capacity + -/// power-of-two growth). For inputs in the few-KiB to ~128 KiB range -/// this is a strict improvement (no doubling spikes inside the -/// measured window). For inputs significantly larger than 128 KiB the -/// allocation curve still grows by amortized doubling but starts from -/// a 128 KiB floor rather than 0. Downstream consumers that measure -/// peak RSS on this entry point will see a different curve than -/// pre-revision; bench shape, not steady-state, is what changed. -/// /// **This is NOT a streaming API.** The source is fully buffered /// into a `Vec` before any compression work begins, so peak input /// memory is bounded by `source.len()` (not "constant regardless of -/// payload size" as a stream-shaped encoder would offer). The RSS -/// notes below apply to the materialization-then-compress shape; if -/// the source is large enough that holding it in memory is not -/// acceptable, use [`StreamingEncoder`] which consumes chunks -/// incrementally without the up-front Vec build. +/// payload size" as a stream-shaped encoder would offer). If the +/// source is large enough that holding it in memory is not acceptable, +/// use [`StreamingEncoder`] which consumes chunks incrementally +/// without the up-front Vec build. /// -/// The other side of the peak shape is the input buffering: this -/// helper drives `read_to_end` to materialize the full source into a -/// `Vec` before forwarding the slice to [`compress_slice_to_vec`]. -/// For a `Read` whose size is unknown ahead of time, `read_to_end` -/// grows that input `Vec` via power-of-two doubling — peak input -/// allocation can be up to 2× the final source length transiently. -/// At the moment that input buffer crosses ~128 KiB the output Vec -/// seed kicks in concurrently. The total live working set on this -/// entry point is approximately -/// `input.capacity() + output_vec_seed + internal_accumulators`, -/// where `output_vec_seed` is `min(compress_bound(input.len()), 128 -/// KiB)` and `internal_accumulators` covers -/// `FrameCompressor::all_blocks` (pre-reserved at frame start, up to -/// ~130 KiB at default block cap) plus per-block scratch (hash tables, -/// literal/sequence staging). Round the helper's RSS peak to -/// `input.capacity() + output_vec_seed + ~130 KiB internal + -/// per-block scratch` rather than the bare `input.capacity() + 128 -/// KiB` figure quoted in earlier revisions, which only accounted for -/// the output seed. [`StreamingEncoder`] avoids the input +/// This helper drives `read_to_end` to materialize the full source +/// into a `Vec` before forwarding the slice to +/// [`compress_slice_to_vec`]. For a `Read` whose size is unknown ahead +/// of time, `read_to_end` grows that input `Vec` via power-of-two +/// doubling: peak input allocation can be up to 2× the final source +/// length transiently. The live working set on this entry point is +/// roughly `input.capacity()` plus the block-accumulation buffer and +/// per-block scratch carried by [`compress_slice_to_vec`], plus the +/// exactly-sized output `Vec`. [`StreamingEncoder`] avoids the input /// materialization step entirely and is the right entry point when /// the source is large or unbounded. /// @@ -223,45 +151,34 @@ pub fn compress_to_vec(source: R, level: CompressionLevel) -> Vec { compress_slice_to_vec(input.as_slice(), level) } -/// Compress a contiguous byte slice into a fresh `Vec` without -/// the input-buffering step that [`compress_to_vec`] performs to -/// adapt a `Read` source. Donor-parity peak-memory shape: the input -/// is read by reference, and the output `Vec` is seeded with -/// `min(compress_bound(src), OUTPUT_BLOCK_CAP = 128 KiB)`. +/// Compress a contiguous byte slice into a fresh `Vec` without the +/// input-buffering step that [`compress_to_vec`] performs to adapt a +/// `Read` source. /// -/// The seed is intentionally capped at one donor block -/// (`OUTPUT_BLOCK_CAP = 128 KiB`). Three regimes: +/// One-shot wrapper over +/// [`FrameCompressor::compress_independent_frame`]: the input is read by +/// reference (the eligible Fast path scans it in place, no per-block +/// history copy), and the returned `Vec` is allocated exactly once at the +/// final frame size after compression. Peak transient memory is the +/// block-accumulation buffer (grown via amortized doubling, ≈ 2× current +/// compressed size at the last realloc) plus the exactly-sized output. The +/// worst-case compressed-size bound is never pinned upfront, so a highly +/// compressible 100 MiB input does not charge ~100 MiB of worst-case +/// expansion against peak. /// -/// * Inputs up to ~127 KiB where `compress_bound(src) < -/// OUTPUT_BLOCK_CAP` — the `min` picks the tighter bound and the -/// `Vec` never reallocates inside the measured window (cheap, no -/// doubling spikes, no over-allocation). -/// * Inputs around 128 KiB where `compress_bound(src) >= -/// OUTPUT_BLOCK_CAP` — the `min` clamps to the cap, so the "no -/// reallocation" property holds only as long as the actual -/// compressed output also fits within `OUTPUT_BLOCK_CAP`. For -/// high-entropy inputs near the cap boundary the `Vec` may grow -/// by one doubling step before the next block lands. -/// * Larger inputs — the `Vec` grows via amortized doubling, with -/// peak transient memory ≈ 2× current compressed size at the last -/// realloc. Deliberate trade-off against pinning -/// `compress_bound(src)` upfront, which on the 100 MiB -/// `large-log-stream` scenario pinned 100.4 MiB even though the -/// actual compressed output was ≪ 1 MiB. +/// To compress many slices, construct one [`FrameCompressor`] and call +/// [`compress_independent_frame_into`](FrameCompressor::compress_independent_frame_into) +/// in a loop instead, which reuses the matcher tables, scratch, and output +/// buffer across frames (this function allocates and primes from scratch +/// each call). /// /// # Panics /// /// Panics on encoder error (matches the failure surface of -/// [`compress_to_vec`], which this function backs). The internal -/// [`FrameCompressor::compress`] call propagates `io::Error` from the -/// output [`Vec`] writer; under the in-tree default writer that -/// channel is infallible and any error becomes a `panic`. Out-of- -/// memory during `Vec::with_capacity` or the encoder's per-block -/// scratch allocations is handled by the global allocator's abort -/// policy. Migrating to a fallible variant requires plumbing -/// `Result` through `FrameCompressor::compress` — out of scope for -/// the slice/Vec entry points which mirror the donor `ZSTD_compress` -/// shape (no error return on the bulk path). +/// [`compress_to_vec`], which this function backs). Out-of-memory during +/// the output / per-block scratch allocations is handled by the global +/// allocator's abort policy. The slice/Vec entry points mirror the donor +/// `ZSTD_compress` shape (no error return on the bulk path). /// /// ```rust /// use structured_zstd::encoding::{compress_slice_to_vec, CompressionLevel}; @@ -269,30 +186,12 @@ pub fn compress_to_vec(source: R, level: CompressionLevel) -> Vec { /// let compressed = compress_slice_to_vec(data, CompressionLevel::Fastest); /// ``` pub fn compress_slice_to_vec(source: &[u8], level: CompressionLevel) -> Vec { - // Initial capacity sized to a single output block, matching donor - // `ZSTD_CStreamOutSize()` (≈ `ZSTD_BLOCKSIZE_MAX` + headroom). For - // small inputs `compress_bound(src)` is smaller than this block — - // use the tighter bound to avoid over-allocation. For larger inputs - // the `Vec` grows via amortized doubling, with peak ≈ 2×current - // compressed size at the last realloc (donor streaming shape: the - // caller buffer is one block, not `ZSTD_compressBound(srcSize)`). - // Pre-allocating `compress_bound(src)` here would charge worst-case - // expansion against peak even on highly compressible inputs — on - // the 100 MiB `large-log-stream` scenario that single allocation - // dominated the encoder's measured footprint (`compress_bound` ≈ - // 100.4 MiB pinned, actual compressed output ≪ 1 MiB). - const OUTPUT_BLOCK_CAP: usize = 128 * 1024; - let cap = compress_bound(source.len()).min(OUTPUT_BLOCK_CAP); - let mut vec = Vec::with_capacity(cap); - let mut frame_enc = FrameCompressor::new(level); - frame_enc.set_source_size_hint(source.len() as u64); - // `set_source` feeds the owned-streaming fallback inside - // `compress_oneshot_borrowed`; the eligible borrowed path reads the - // `source` slice in place instead (no per-block history copy). - frame_enc.set_source(source); - frame_enc.set_drain(&mut vec); - frame_enc.compress_oneshot_borrowed(source); - vec + // Bare `FrameCompressor` resolves all three type params to their + // defaults (`&'static [u8]` reader, `Vec` drain, MatchGeneratorDriver); + // neither the reader nor the drain is used by the in-place + // `compress_independent_frame` path. + let mut enc: FrameCompressor = FrameCompressor::new(level); + enc.compress_independent_frame(source) } /// The compression mode used impacts the speed of compression, From b1893b979065b53e3cdcb932608febae2f4d68d6 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Thu, 4 Jun 2026 19:42:49 +0300 Subject: [PATCH 08/10] test(bench): add clean dict/small-input encode-loop profiling example Standalone perf-record harness for the encoder hot path on small inputs with/without a dictionary. The compare_ffi dictionary bench cannot be flamegraphed for the timed compress: its per-scenario setup trains dictionaries and runs a validation decode for every scenario before the filtered group, polluting a compress-filtered flamegraph with training / decoder frames that never run in the timed loop. This binary parses the dictionary once, then loops compress_independent_frame_into over a contiguous slice, reusing the compressor and output buffer (the reusable-context shape) so the profile isolates the per-frame cost (dictionary prime + optimal parse + entropy) with no training, decode, or FFI symbols mixed in. `logs` input reproduces the *-log-lines bench fixtures byte for byte. Part of #316 --- zstd/examples/encode_loop_dict.rs | 105 ++++++++++++++++++++++++++++++ 1 file changed, 105 insertions(+) create mode 100644 zstd/examples/encode_loop_dict.rs diff --git a/zstd/examples/encode_loop_dict.rs b/zstd/examples/encode_loop_dict.rs new file mode 100644 index 000000000..5f42bbb98 --- /dev/null +++ b/zstd/examples/encode_loop_dict.rs @@ -0,0 +1,105 @@ +//! Standalone encode-loop binary for CLEAN perf-record profiles of the +//! ENCODER hot path on small inputs, with or without a dictionary. +//! +//! Why this exists: the `compare_ffi` dictionary bench cannot be flamegraphed +//! for the timed compress. Its per-scenario setup trains dictionaries +//! (XXH64-heavy) and runs a validation decode for EVERY scenario before the +//! filtered group, so a compress-filtered flamegraph is polluted with +//! `ZDICT`/decoder/`twox_hash` frames that never run in the timed loop. This +//! binary does only: parse the dictionary ONCE, then loop +//! [`FrameCompressor::compress_independent_frame_into`] over a contiguous +//! `&[u8]` at the given level. No criterion, no FFI, no training, no decode — +//! the samples land purely on our per-frame encode hot path. +//! +//! The compressor and the output buffer are both reused across iterations +//! (the reusable-compression-context shape a real per-block-frame consumer +//! uses): the matcher tables, scratch, and any primed dictionary are +//! allocated/parsed once, and the dictionary is re-primed per frame inside +//! `prepare_frame`. So the steady-state profile isolates the genuine +//! per-frame cost (dictionary prime + optimal parse + entropy), not the +//! one-time setup the fresh-per-iter bench shape folds in. +//! +//! Build: `cargo build --profile flamegraph -p structured-zstd +//! --example encode_loop_dict --features dict_builder` +//! Run: `cargo flamegraph --example encode_loop_dict --features dict_builder +//! --profile flamegraph -- [dict_path]` +//! +//! `` is either `logs` (N bytes of the bench `repeated_log_lines` +//! fixture, e.g. `logs4096` == the `small-4k-log-lines` scenario byte for +//! byte) or a path to a raw file. `dict_path` is optional; when given the +//! dictionary is attached once before the loop. + +use std::env; + +use structured_zstd::encoding::{CompressionLevel, FrameCompressor}; + +/// Byte-for-byte the bench `repeated_log_lines(len)` fixture so a `logs` +/// input reproduces the `*-log-lines` dashboard scenarios exactly. +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 resolve_input(spec: &str) -> Vec { + if let Some(rest) = spec.strip_prefix("logs") { + let n: usize = rest.parse().expect("logs: N must be a byte count"); + repeated_log_lines(n) + } else { + std::fs::read(spec).expect("read input file") + } +} + +fn main() { + let args: Vec = env::args().collect(); + let level: i32 = args.get(1).and_then(|s| s.parse().ok()).unwrap_or(22); + let iters: u32 = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(20_000); + let input_spec: &str = args.get(3).map(|s| s.as_str()).unwrap_or("logs4096"); + let dict_path: Option<&str> = args.get(4).map(|s| s.as_str()); + + let src = resolve_input(input_spec); + + // One reused compressor: matcher tables + any dictionary parse happen + // once here, mirroring a consumer that reuses a context across N + // independent per-block frames. + let mut cctx: FrameCompressor = FrameCompressor::new(CompressionLevel::from_level(level)); + if let Some(path) = dict_path { + let dict = std::fs::read(path).expect("read dict file"); + cctx.set_dictionary_from_bytes(&dict) + .expect("dictionary should attach"); + } + + // Output buffer reused across iterations (allocated once, replaced in + // place by `compress_independent_frame_into`), so steady-state iters do + // zero output allocation. + let mut out: Vec = Vec::new(); + let mut sink: usize = 0; + for _ in 0..iters { + cctx.compress_independent_frame_into(&src, &mut out); + sink = sink.wrapping_add(out.len()); + core::hint::black_box(&out); + } + + eprintln!( + "encoded {} bytes × {} iters at level {} dict={}; last-out-sum={}", + src.len(), + iters, + level, + dict_path.unwrap_or("none"), + sink + ); +} From 1cd2a8d7493c07f7a3ddd85fba79a7983e2712c2 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Thu, 4 Jun 2026 20:49:54 +0300 Subject: [PATCH 09/10] test(encode): broaden fast-band + pre-split coverage, borrow in decode assert - fast_band_strategies_prefer_repeat_fse_table: assert all three eligible strategies (Fast, Dfast, Greedy), not just Greedy, so an enum-arm regression in the reuse branch is caught. - retarget the end-to-end pre-split roundtrip from Level 13 (lazy, no longer pre-splits) to Level 5 (greedy), the chunk-split path this revision introduces, so compress -> optimal_block_size -> split_block_by_chunks stays covered. Renamed accordingly. - borrow `result` in the truncated-frame match so the value stays usable in the trailing assert format string. --- zstd/src/encoding/blocks/compressed.rs | 30 +++++++++++++++----------- zstd/src/encoding/frame_compressor.rs | 22 +++++++++++-------- zstd/src/tests/mod.rs | 2 +- 3 files changed, 31 insertions(+), 23 deletions(-) diff --git a/zstd/src/encoding/blocks/compressed.rs b/zstd/src/encoding/blocks/compressed.rs index ab6bfb77f..32c7682ee 100644 --- a/zstd/src/encoding/blocks/compressed.rs +++ b/zstd/src/encoding/blocks/compressed.rs @@ -2694,19 +2694,23 @@ mod tests { counts[1] = 6; let total = 10; - // Greedy is fast-band → unconditional reuse of the covering table. - let mode = super::choose_table_from_counts( - Some(&previous), - fse_tables.ll_default_ref(), - &counts, - total, - 9, - StrategyTag::Greedy, - ); - assert!( - matches!(mode, FseTableMode::RepeatLast(_)), - "fast-band greedy must reuse the covering previous table", - ); + // All fast-band strategies (Fast, Dfast, Greedy) unconditionally + // reuse the covering previous table; cover every eligible arm so an + // enum-arm regression in the implementation branch is caught. + for strategy in [StrategyTag::Fast, StrategyTag::Dfast, StrategyTag::Greedy] { + let mode = super::choose_table_from_counts( + Some(&previous), + fse_tables.ll_default_ref(), + &counts, + total, + 9, + strategy, + ); + assert!( + matches!(mode, FseTableMode::RepeatLast(_)), + "fast-band {strategy:?} must reuse the covering previous table", + ); + } } #[test] diff --git a/zstd/src/encoding/frame_compressor.rs b/zstd/src/encoding/frame_compressor.rs index c4f174182..7f5e6f3c3 100644 --- a/zstd/src/encoding/frame_compressor.rs +++ b/zstd/src/encoding/frame_compressor.rs @@ -3011,18 +3011,22 @@ mod tests { assert_eq!(level_pre_split(CompressionLevel::Level(22)), Some(4)); } - /// End-to-end: a 256 KB heterogeneous payload compressed at - /// Level(13) (borders heuristic active) round-trips through the - /// crate's own decoder. The pre-split path runs over the first - /// 128 KB block and emits two consecutive sub-blocks; the second - /// 128 KB block goes through the splitter on its own. The test - /// proves the split decisions do not corrupt the frame bitstream. + /// End-to-end: a 256 KB heterogeneous payload compressed at Level(5) + /// (greedy, the pre-split path this revision routes through the cheap + /// chunk splitter) round-trips through the crate's own decoder. The + /// pre-split path runs over the first 128 KB block and may emit two + /// consecutive sub-blocks; the second 128 KB block goes through the + /// splitter on its own. The test proves the split decisions do not + /// corrupt the frame bitstream. Level 13 (lazy) no longer pre-splits, + /// so the fixture uses Level 5 to keep the + /// `compress() -> optimal_block_size() -> split_block_by_chunks()` + /// wiring covered. #[test] - fn level_13_borders_split_roundtrips_through_own_decoder() { + fn greedy_chunk_split_roundtrips_through_own_decoder() { use crate::encoding::CompressionLevel; let mut data = vec![0u8; 256 * 1024]; // First 128 KB: low-entropy repeating run; second 128 KB: - // counter sequence — clearly distinct border histograms. + // counter sequence, clearly distinct sub-block fingerprints. for (i, byte) in data.iter_mut().enumerate() { *byte = if i < 128 * 1024 { (i & 0x07) as u8 @@ -3032,7 +3036,7 @@ mod tests { } let mut compressed = Vec::new(); - let mut compressor = FrameCompressor::new(CompressionLevel::Level(13)); + let mut compressor = FrameCompressor::new(CompressionLevel::Level(5)); compressor.set_source(data.as_slice()); compressor.set_drain(&mut compressed); compressor.compress(); diff --git a/zstd/src/tests/mod.rs b/zstd/src/tests/mod.rs index 8c8759bea..829bbe386 100644 --- a/zstd/src/tests/mod.rs +++ b/zstd/src/tests/mod.rs @@ -536,7 +536,7 @@ fn test_decode_all() { // raises the block-precise variant; without it, the legacy one. let mut output = vec![0; original.len()]; let result = decoder.decode_all(&input[..input.len() - 600], &mut output); - let is_block_body_err = match result { + let is_block_body_err = match &result { Err(FrameDecoderError::FailedToReadBlockBody(_)) => true, #[cfg(feature = "lsm")] Err(FrameDecoderError::FailedToReadBlockBodyAt { .. }) => true, From 1b05540b6c43143e12f7a2095e50449a28cb4051 Mon Sep 17 00:00:00 2001 From: Dmitry Prudnikov Date: Thu, 4 Jun 2026 22:38:26 +0300 Subject: [PATCH 10/10] test(encode): force second-block transition so chunk split is exercised The greedy chunk-split roundtrip changed entropy only between the two donor blocks, leaving each block internally homogeneous. The donor savings<3 gate skips splitting the first block anyway, so split_block_by_chunks could return MAX_BLOCK_SIZE for both blocks and the pending_input.split_off path was never hit. Move the fingerprint transition into the second donor block (after the compressible first block banks savings) and assert optimal_block_size returns a sub-block boundary, so the test provably exercises chunk splitting before the roundtrip rather than passing vacuously. --- zstd/src/encoding/frame_compressor.rs | 47 ++++++++++++++++++++------- 1 file changed, 35 insertions(+), 12 deletions(-) diff --git a/zstd/src/encoding/frame_compressor.rs b/zstd/src/encoding/frame_compressor.rs index 7f5e6f3c3..4329e9ab8 100644 --- a/zstd/src/encoding/frame_compressor.rs +++ b/zstd/src/encoding/frame_compressor.rs @@ -3011,30 +3011,53 @@ mod tests { assert_eq!(level_pre_split(CompressionLevel::Level(22)), Some(4)); } - /// End-to-end: a 256 KB heterogeneous payload compressed at Level(5) + /// End-to-end: a 256 KB payload whose SECOND 128 KB donor block carries + /// an intra-block fingerprint transition, compressed at Level(5) /// (greedy, the pre-split path this revision routes through the cheap - /// chunk splitter) round-trips through the crate's own decoder. The - /// pre-split path runs over the first 128 KB block and may emit two - /// consecutive sub-blocks; the second 128 KB block goes through the - /// splitter on its own. The test proves the split decisions do not - /// corrupt the frame bitstream. Level 13 (lazy) no longer pre-splits, - /// so the fixture uses Level 5 to keep the - /// `compress() -> optimal_block_size() -> split_block_by_chunks()` - /// wiring covered. + /// chunk splitter), round-trips through the crate's own decoder. + /// + /// The transition lives in the second block on purpose: the donor + /// `savings < 3` gate skips splitting the first block (savings start at + /// 0), so the first block is a homogeneous compressible run that banks + /// savings, and the second block is the one whose intra-block transition + /// `split_block_by_chunks()` resolves into a sub-block boundary (the + /// `pending_input.split_off(...)` path). The test asserts that split + /// decision directly so it cannot silently stop exercising the path if + /// the fixture or params drift, then proves the emitted split frame + /// round-trips. Level 13 (lazy) no longer pre-splits, hence Level 5. #[test] fn greedy_chunk_split_roundtrips_through_own_decoder() { use crate::encoding::CompressionLevel; let mut data = vec![0u8; 256 * 1024]; - // First 128 KB: low-entropy repeating run; second 128 KB: - // counter sequence, clearly distinct sub-block fingerprints. + // First 128 KB: homogeneous low-entropy run (compressible, banks + // the savings the donor gate needs). Second 128 KB: low-entropy run + // for its first half, then a counter sequence: a clear intra-block + // fingerprint transition at the 192 KB midpoint for the chunk + // splitter to find. for (i, byte) in data.iter_mut().enumerate() { - *byte = if i < 128 * 1024 { + *byte = if i < 192 * 1024 { (i & 0x07) as u8 } else { (i % 251 + 1) as u8 }; } + // Directly assert the chunk splitter resolves the second block's + // intra-block transition into a sub-block boundary once savings have + // accrued (the compressible first block banks well over the gate). + let second_block = &data[128 * 1024..]; + let split = super::optimal_block_size( + CompressionLevel::Level(5), + second_block, + second_block.len(), + MAX_BLOCK_SIZE as usize, + 100, + ); + assert!( + split < MAX_BLOCK_SIZE as usize, + "second donor block must chunk-split at its intra-block transition, got {split}", + ); + let mut compressed = Vec::new(); let mut compressor = FrameCompressor::new(CompressionLevel::Level(5)); compressor.set_source(data.as_slice());