diff --git a/zstd/benches/compare_ffi.rs b/zstd/benches/compare_ffi.rs index 601eb3154..867a8aa82 100644 --- a/zstd/benches/compare_ffi.rs +++ b/zstd/benches/compare_ffi.rs @@ -293,6 +293,8 @@ fn bench_compress(c: &mut Criterion) { emit_report_line(scenario, level, &rust_compressed, &ffi_compressed); emit_frame_header_report(scenario, level, "rust", &rust_compressed); emit_frame_header_report(scenario, level, "ffi", &ffi_compressed); + emit_block_structure_report(scenario, level, "rust", &rust_compressed); + emit_block_structure_report(scenario, level, "ffi", &ffi_compressed); } let benchmark_name = format!("compress/{}/{}/{}", level.name, scenario.id, "matrix"); @@ -1085,6 +1087,114 @@ fn emit_frame_header_report( ); } +fn emit_block_structure_report( + scenario: &Scenario, + level: LevelConfig, + encoder: &'static str, + compressed: &[u8], +) { + if compressed.len() < 5 { + return; + } + // Raw hex of the frame head: lets us read the literals-section header and + // Huffman tree description (FSE weights vs direct nibbles) by hand when + // comparing rust vs ffi generation byte-for-byte on a fixture. + let head_len = compressed.len().min(48); + let hex: String = compressed[..head_len] + .iter() + .map(|b| format!("{b:02x}")) + .collect::>() + .join(" "); + println!( + "REPORT_HEX scenario={} level={} encoder={} head=[{}]", + scenario.id, level.name, encoder, hex + ); + let desc = compressed[4]; + let fcs_flag = desc >> 6; + let single_segment = ((desc >> 5) & 0x1) == 1; + let checksum = ((desc >> 2) & 0x1) == 1; + let dict_id_bytes: usize = match desc & 0x3 { + 0 => 0, + 1 => 1, + 2 => 2, + _ => 4, + }; + let fcs_bytes: usize = match fcs_flag { + 0 => usize::from(single_segment), + 1 => 2, + 2 => 4, + _ => 8, + }; + let mut pos = 4 + 1 + usize::from(!single_segment) + dict_id_bytes + fcs_bytes; + let payload_end = compressed.len() - if checksum { 4 } else { 0 }; + // A truncated/invalid frame must report `parse=error`, not fall through to a + // normal REPORT_BLK that would make it look like a valid empty/partial frame. + if pos > payload_end { + println!( + "REPORT_BLK scenario={} level={} encoder={} parse=error", + scenario.id, level.name, encoder + ); + return; + } + let mut blocks: Vec<(char, usize, bool)> = Vec::new(); + let mut saw_last = false; + while pos + 3 <= payload_end { + let h = compressed[pos] as usize + | (compressed[pos + 1] as usize) << 8 + | (compressed[pos + 2] as usize) << 16; + let last = (h & 1) == 1; + let btype = (h >> 1) & 0x3; + let bsize = h >> 3; + let (kind, advance) = match btype { + 0 => ('R', bsize), // Raw + 1 => ('L', 1), // RLE (1 byte payload) + 2 => ('C', bsize), // Compressed + _ => { + println!( + "REPORT_BLK scenario={} level={} encoder={} parse=error", + scenario.id, level.name, encoder + ); + return; + } + }; + if pos + 3 + advance > payload_end { + println!( + "REPORT_BLK scenario={} level={} encoder={} parse=error", + scenario.id, level.name, encoder + ); + return; + } + blocks.push((kind, bsize, last)); + pos += 3 + advance; + if last { + saw_last = true; + break; + } + } + // A clean frame ends exactly on a terminal (`last`) block with no leftover + // bytes; otherwise it was truncated (e.g. 1-2 trailing bytes that cannot + // form a block header) and must report a parse error, not a block list. + if !saw_last || pos != payload_end { + println!( + "REPORT_BLK scenario={} level={} encoder={} parse=error", + scenario.id, level.name, encoder + ); + return; + } + let list: Vec = blocks + .iter() + .map(|(k, s, l)| format!("{k}{s}{}", if *l { "!" } else { "" })) + .collect(); + println!( + "REPORT_BLK scenario={} level={} encoder={} n_blocks={} blocks=[{}]", + scenario.id, + level.name, + encoder, + blocks.len(), + list.join(","), + ); +} + fn emit_report_line( scenario: &Scenario, level: LevelConfig, diff --git a/zstd/src/decoding/frame.rs b/zstd/src/decoding/frame.rs index d563950b3..c4cec87f9 100644 --- a/zstd/src/decoding/frame.rs +++ b/zstd/src/decoding/frame.rs @@ -18,6 +18,7 @@ pub(crate) fn read_frame_header(r: impl Read) -> Result<(FrameHeader, u8), ReadF /// frame detection is bypassed — the caller MUST know out-of-band /// that the stream is magicless. Upstream zstd parity: /// `ZSTD_f_zstd1_magicless` via `ZSTD_d_format`. +#[inline] pub fn read_frame_header_with_format( mut r: impl Read, magicless: bool, @@ -60,6 +61,10 @@ pub fn read_frame_header_with_format( window_descriptor: 0, }; + // Each variable header field is read with its own field-specific error so + // a truncated frame reports which field is missing. The slice `read_exact` + // override (`io_nostd`) makes each of these a single bounds-checked copy, + // so the small per-field reads are not the cost they once were. if !desc.single_segment_flag() { r.read_exact(&mut buf[0..1]) .map_err(err::WindowDescriptorReadError)?; @@ -73,10 +78,8 @@ pub fn read_frame_header_with_format( r.read_exact(buf).map_err(err::DictionaryIdReadError)?; bytes_read += dict_id_len; let mut dict_id = 0u32; - - #[allow(clippy::needless_range_loop)] - for i in 0..dict_id_len { - dict_id += (buf[i] as u32) << (8 * i); + for (i, &b) in buf.iter().enumerate() { + dict_id += (b as u32) << (8 * i); } if dict_id != 0 { frame_header.dict_id = Some(dict_id); @@ -91,10 +94,8 @@ pub fn read_frame_header_with_format( .map_err(err::FrameContentSizeReadError)?; bytes_read += fcs_len; let mut fcs = 0u64; - - #[allow(clippy::needless_range_loop)] - for i in 0..fcs_len { - fcs += (fcs_buf[i] as u64) << (8 * i); + for (i, &b) in fcs_buf.iter().enumerate() { + fcs += (b as u64) << (8 * i); } if fcs_len == 2 { fcs += 256; diff --git a/zstd/src/decoding/frame_decoder.rs b/zstd/src/decoding/frame_decoder.rs index 11a7b1faa..aa291170e 100644 --- a/zstd/src/decoding/frame_decoder.rs +++ b/zstd/src/decoding/frame_decoder.rs @@ -854,6 +854,7 @@ impl FrameDecoderState { /// non-direct fallback reserves `window_size` once in /// `decode_all_impl` / `decode_blocks` via `reserve_buffer` before /// any block write. + #[inline] pub(crate) fn new_with_format( source: impl Read, magicless: bool, @@ -898,6 +899,7 @@ impl FrameDecoderState { /// `DecoderScratchKind::reserve_buffer(window_size)` before any block /// write. A reused scratch whose new frame fits within prior capacity /// reuses it; a larger one grows on that same `reserve_buffer` call. + #[inline] pub(crate) fn reset_with_format( &mut self, source: impl Read, @@ -1186,6 +1188,7 @@ impl FrameDecoder { /// Note that all bytes currently in the decodebuffer from any previous frame will be lost. Collect them with collect()/collect_to_writer() /// /// equivalent to reset() + #[inline] pub fn init(&mut self, source: impl Read) -> Result<(), FrameDecoderError> { self.reset(source) } @@ -1222,6 +1225,7 @@ impl FrameDecoder { /// Note that all bytes currently in the decodebuffer from any previous frame will be lost. Collect them with collect()/collect_to_writer() /// /// equivalent to init() + #[inline] pub fn reset(&mut self, source: impl Read) -> Result<(), FrameDecoderError> { use FrameDecoderError as err; // Fresh frame → start with an empty per-block checksum vec so diff --git a/zstd/src/decoding/literals_section_decoder.rs b/zstd/src/decoding/literals_section_decoder.rs index f781e7446..cb343ec36 100644 --- a/zstd/src/decoding/literals_section_decoder.rs +++ b/zstd/src/decoding/literals_section_decoder.rs @@ -446,13 +446,46 @@ fn decompress_literals_impl( // a no-op; eliding it also lets us hold `target_ptr` without an // intervening `&mut Vec` borrow that invalidates the pointer // under stacked-borrows. + // Per-stream tail decode, mirroring upstream zstd `HUF_decodeStreamX1` + // (huf_decompress.c:546): decode in groups of 4 with ONE reload per + // group, then the final < 4 symbols per-symbol. The previous form + // reloaded the bit reader on EVERY symbol, so each trailing symbol near + // a stream end paid `refill_slow` — the dominant drain cost on a + // literal-heavy frame. + let group_bits = 4 * max_num_bits; for i in 0..4 { - while brs[i].bits_remaining() > -max_bits && cursors[i] < ends[i] { - let byte = decoders[i].decode_symbol_and_advance(&mut brs[i]); - unsafe { - target_ptr.add(cursors[i]).write(byte); + // Phase 1 (upstream `HUF_decodeStreamX1` lines 551-557): decode in + // groups of four with a SINGLE reload per group. Output-based bound + // (`cursors[i] + 4 <= ends[i]`), like upstream's `p < pEnd-3`, so a + // group only runs while four whole symbols remain — it never reads + // past the last symbol into the zero padding. + while cursors[i] + 4 <= ends[i] { + brs[i].ensure_bits(group_bits); + for _ in 0..4 { + let byte = decoders[i].decode_symbol_and_advance_no_refill(&mut brs[i]); + unsafe { + target_ptr.add(cursors[i]).write(byte); + } + cursors[i] += 1; + } + } + // Phase 2 (upstream `HUF_decodeStreamX1` line 568 `while (p < pEnd)`): + // the final < 4 symbols. ONE reload covers them (<= 3 * max_num_bits + // bits), then NO per-symbol reload — so the trailing symbols at the + // stream end never pay the cold `refill_slow` each. `bits_remaining` + // is reload-timing-independent (the padding lands in either + // `extra_bits` or `bits_consumed`, and `(64 - bits_consumed) - + // extra_bits` is identical), so the end-of-stream check below still + // holds exactly. + if cursors[i] < ends[i] { + brs[i].ensure_bits(group_bits); + while cursors[i] < ends[i] { + let byte = decoders[i].decode_symbol_and_advance_no_refill(&mut brs[i]); + unsafe { + target_ptr.add(cursors[i]).write(byte); + } + cursors[i] += 1; } - cursors[i] += 1; } if brs[i].bits_remaining() != -max_bits { return Err(DecompressLiteralsError::BitstreamReadMismatch { diff --git a/zstd/src/encoding/dfast/mod.rs b/zstd/src/encoding/dfast/mod.rs index dfb137911..d3fb1c711 100644 --- a/zstd/src/encoding/dfast/mod.rs +++ b/zstd/src/encoding/dfast/mod.rs @@ -21,9 +21,9 @@ use super::fastpath::{FastpathKernel, select_kernel}; use super::incompressible::block_looks_incompressible; use super::levels::config::MIN_WINDOW_LOG; use super::match_generator::{ - DFAST_EMPTY_SLOT, DFAST_HASH_BITS, DFAST_INCOMPRESSIBLE_SKIP_STEP, DFAST_MAX_SKIP_STEP, - DFAST_MIN_MATCH_LEN, DFAST_REBASE_GUARD_BAND, DFAST_SHORT_HASH_BITS_DELTA, - DFAST_SHORT_HASH_LOOKAHEAD, DFAST_SKIP_STEP_GROWTH_INTERVAL, + DFAST_EMPTY_SLOT, DFAST_HASH_BITS, DFAST_INCOMPRESSIBLE_SKIP_STEP, DFAST_MIN_MATCH_LEN, + DFAST_REBASE_GUARD_BAND, DFAST_SHORT_HASH_BITS_DELTA, DFAST_SHORT_HASH_LOOKAHEAD, + DFAST_SKIP_STEP_GROWTH_INTERVAL, }; use super::match_table::helpers::{common_prefix_len_with_kernel, extend_backwards_shared}; use super::match_table::storage::{REBASE_RESET_FLOOR_CEILING, check_stream_abs_headroom}; @@ -51,6 +51,11 @@ const HASH_READ_SIZE: usize = 8; /// rep emissions that upstream produces. const DFAST_REP_MIN_MATCH_LEN: usize = 4; +/// Cached `DFTRACE` env flag for the dfast commit-path diagnostic (read once; +/// see the `DFTRACE` gate in the fast-loop commit handler). +#[cfg(feature = "std")] +static DFTRACE_ENABLED: std::sync::OnceLock = std::sync::OnceLock::new(); + #[derive(Clone)] pub(crate) struct DfastMatchGenerator { pub(crate) max_window_size: usize, @@ -1205,58 +1210,16 @@ impl DfastMatchGenerator { if match_len < DFAST_REP_MIN_MATCH_LEN { break; } - // Sparse complementary insertion (upstream parity, - // `zstd_double_fast.c:300-304`): upstream inserts ONLY at - // `curr+2`, `ip-2`, `ip-1` after a match — three specific - // positions, not the whole match range. The previous - // `insert_positions(abs_pos, abs_pos + match_len)` made - // sense only under the 4-slot bucket; with single-slot - // upstream zstd parity it would just overwrite every bucket along - // the match span and discard whichever positions the - // producer was about to re-probe. - // - // At the floor `match_len == DFAST_REP_MIN_MATCH_LEN (= 4)` - // the three targets collapse to two distinct positions. - // With `post_match_end = abs_pos + 4` the three offsets - // resolve to: - // `curr + 2` = abs_pos + 2 - // `ip - 2` = abs_pos + 4 - 2 = abs_pos + 2 ← same as curr+2 - // `ip - 1` = abs_pos + 4 - 1 = abs_pos + 3 ← distinct - // So `curr+2` and `ip-2` write the same slot twice; - // single-slot overwrite is idempotent, so the duplicate - // write is correctness-neutral. It's one wasted store on - // the shortest rep extension and not worth a branch to - // dedup. For `match_len >= 5` all three offsets are - // distinct. - // - // Why `abs_pos` itself is NOT in the insert set, despite - // not being written by the fast loop's pre-check insert - // (the previous range form `insert_positions(abs_pos, - // post_match_end)` did include it): upstream - // `ZSTD_compressBlock_doubleFast_*` likewise does not - // insert at the rep-extension start. After a rep match, - // upstream just advances `ip` and reruns the outer fast - // loop, which writes the new `curr` (the post-rep cursor) - // — never the rep's own start. The three offsets above - // are upstream's primary-match-emit insertion pattern, - // mirrored here to preserve hit rate on the chains that - // follow a rep extension. The hash-state delta vs the - // prior range-insert behavior is intentional and tracked - // by the sequence-stream comparator harness (the - // per-sequence diff vs FFI mentioned in the PR body's - // deferred section); the audit signed off on the current - // sparse set as the closest faithful mirror of upstream. - let post_match_end = abs_pos + match_len; - let insert_targets = [ - abs_pos + 2, // curr + 2 - post_match_end.saturating_sub(2), // ip - 2 (post-match cursor) - post_match_end.saturating_sub(1), // ip - 1 - ]; - for &target in &insert_targets { - if target > abs_pos && target < post_match_end { - self.insert_position(target); - } - } + // Upstream zstd immediate-repcode insertion (zstd_double_fast.c:314-315): + // INSIDE the rep chain, upstream writes BOTH hash tables at the rep + // position itself (`hashSmall[hash(ip)] = ip; hashLong[hash(ip)] = ip`) + // before advancing — NOT the `curr+2 / ip-2 / ip-1` primary-match + // complementary set (that pattern belongs to `_match_found`, lines + // 300-304, reached only by the non-rep store path). Inserting the + // wrong set here leaves the rep-position keys stale, so a later + // position re-resolves the long hash to an older far candidate + // instead of the stable rep offset — the dfast ratio gap vs C. + self.insert_position(abs_pos); // Emit zero-literal rep sequence. handle_sequence(Sequence::Triple { literals: &[], @@ -1386,6 +1349,7 @@ impl DfastMatchGenerator { current_abs_start: usize, literals_start: &mut usize, candidate: MatchCandidate, + scan_pos: usize, handle_sequence: &mut impl for<'a> FnMut(Sequence<'a>), ) -> usize { // Upstream zstd `zstd_double_fast.c` parity: the inner search loop already @@ -1396,18 +1360,34 @@ impl DfastMatchGenerator { // prefixStart)` + the growing `step`). Match interior: upstream zstd fills only // the sparse 3-target set (`curr+2`, `ip-2`, `ip-1`), each clamped to // the open match interval. - let match_start = candidate.start; + // Upstream zstd complementary insertion (zstd_double_fast.c:300-304) is + // ASYMMETRIC across the two tables: + // hashLong: curr+2, ip-2 + // hashSmall: curr+2, ip-1 + // The previous form inserted curr+2/ip-2/ip-1 into BOTH tables, putting + // ip-1 into long and ip-2 into short that upstream never writes — + // polluting the long table so later positions resolve to the wrong + // (far, non-rep) candidate. Mirror the exact per-table target set. + // + // `curr` is the iteration's SCAN position (`scan_pos`, upstream `curr` + // = `(U32)(ip-base)` fixed at zstd_double_fast.c:184), NOT the match + // start: upstream advances `ip` for a rep1 (`ip++`, the match begins at + // `scan+1`) and rewinds it during backward catch-up, but `curr` stays + // pinned to the scan cursor. Anchoring `curr+2` on `match_start` instead + // shifts the insert by +1 on every rep1 (and by the catch-up length on + // extended matches), seeding the short hash with the wrong positions — + // a divergence that compounds across a block. let post_match_end = candidate.start + candidate.match_len; - let insert_targets = [ - match_start + 2, // curr + 2 - post_match_end.saturating_sub(2), // ip - 2 (post-match cursor) - post_match_end.saturating_sub(1), // ip - 1 - ]; - for &target in &insert_targets { - if target > match_start && target < post_match_end { - self.insert_position(target); - } - } + // `match_len >= DFAST_REP_MIN_MATCH_LEN` (4) for every committed match, + // so `post_match_end >= 4`: the `- 2` / `- 1` cannot underflow (plain + // arithmetic, no `saturating_*` masking). + let curr_plus_2 = scan_pos + 2; + let ip_minus_2 = post_match_end - 2; + let ip_minus_1 = post_match_end - 1; + self.insert_long(curr_plus_2); + self.insert_long(ip_minus_2); + self.insert_short(curr_plus_2); + self.insert_short(ip_minus_1); // Inline the trailing-block slice rather than calling // `get_last_space()` so this matches the gate pattern used by // `skip_matching` / `start_matching` (read `window_blocks.back()` @@ -1629,6 +1609,27 @@ impl DfastMatchGenerator { #[inline] pub(crate) fn insert_position(&mut self, pos: usize) { + self.insert_masked::(pos); + } + + /// Insert `pos` into ONLY the long hash table (upstream zstd asymmetric + /// complementary insertion: `hashLong` at `curr+2` and `ip-2`). + fn insert_long(&mut self, pos: usize) { + self.insert_masked::(pos); + } + + /// Insert `pos` into ONLY the short hash table (upstream zstd: `hashSmall` + /// at `curr+2` and `ip-1`). + fn insert_short(&mut self, pos: usize) { + self.insert_masked::(pos); + } + + /// Const-generic insertion core. `SHORT` / `LONG` select which tables to + /// write; both `true` is the symmetric `insert_position`. The const flags + /// fold at compile time, so `insert_position` stays byte-identical to the + /// previous single-function form while the asymmetric variants emit only + /// their one write. + fn insert_masked(&mut self, pos: usize) { // Source the bytes + rebase coordinates through `scan_source()` so a // borrowed window's seam / tail re-seeds hash the in-place input // exactly as the owned path hashes its `history` concat. @@ -1669,7 +1670,7 @@ impl DfastMatchGenerator { // `start_matching` seam re-seed picks it up once the next block // extends the source far enough to form its full 5-byte key. let concat = unsafe { core::slice::from_raw_parts(base_ptr.add(start_offset), concat_len) }; - if idx + 5 <= concat_len { + if SHORT && idx + 5 <= concat_len { let short = self.short_hash_index(&concat[idx..]); debug_assert!(short < self.short_len()); // Short region starts at `long_len`. @@ -1677,7 +1678,7 @@ impl DfastMatchGenerator { unsafe { *self.tables.get_unchecked_mut(slot) = packed }; } - if idx + 8 <= concat_len { + if LONG && idx + 8 <= concat_len { let long = self.long_hash_index(&concat[idx..]); debug_assert!(long < self.long_len()); unsafe { *self.tables.get_unchecked_mut(long) = packed }; @@ -1746,9 +1747,8 @@ macro_rules! start_matching_fast_loop_body { // early-skip path (the `block_looks_incompressible_strict` short // circuit + `miss_run` / `DFAST_LOCAL_SKIP_TRIGGER` thresholding). // The step ramp is now driven purely by distance traveled - // (`DFAST_SKIP_STEP_GROWTH_INTERVAL = 64`, upstream zstd parity except - // upstream zstd uses 256 — see "Upstream zstd-deviation audit" in the PR body), - // so blocks the strict gate used to bail out of early now scan + // (`DFAST_SKIP_STEP_GROWTH_INTERVAL = 256`, matching upstream zstd's + // `kStepIncr = 1 << kSearchStrength`), so blocks the strict gate used to bail out of early now scan // through the standard ramp. `block_looks_incompressible_strict` // is still used by `levels/fastest.rs` for the Fastest preset // and by `incompressible.rs` unit tests, so the helper itself @@ -1937,6 +1937,7 @@ macro_rules! start_matching_fast_loop_body { $current_abs_start, &mut literals_start, committed, + $current_abs_start + ip0, $handle_sequence, ); pos = start + committed.match_len; @@ -2030,7 +2031,13 @@ macro_rules! start_matching_fast_loop_body { // previous `Option` + sibling `usize` // pairing relied on a comment-block to flag the coupling. enum InnerExit { - Committed(MatchCandidate), + // `u8` is a debug path tag (0=rep1, 1=long@ip0, 2=short/_search_next_long, + // 3=dict-long, 4=dict-snl), surfaced by the `DFTRACE` env gate in the + // commit handler to diagnose match-path / offset divergence vs C ffi. + // `usize` is the iteration's scan position `abs_ip0` (upstream `curr`, + // zstd_double_fast.c:184) — the complementary insertion anchors on it, + // NOT on the (rep1 `+1` / catch-up adjusted) match start. + Committed(MatchCandidate, u8, usize), Tail(usize), } @@ -2106,9 +2113,15 @@ macro_rules! start_matching_fast_loop_body { let rep1 = $self.offset_hist[0] as usize; if rep1 != 0 && rep1 <= abs_ip1 { let cand_pos_r = abs_ip1 - rep1; - if cand_pos_r >= wlow1 - && cand_pos_r >= $current_abs_start + literals_start - { + // Window-low bound ONLY (upstream zstd `zstd_double_fast.c:190` + // gates the rep on nothing but `offset_1 > 0` and the 4-byte + // equality at `ip+1-offset_1`, i.e. the candidate is in the + // prefix). A prior extra `cand_pos_r >= literals_start` clause + // rejected essentially EVERY rep — after a match + // `literals_start ≈ ip`, so any back-reference (cand before + // the literal cursor) failed it — collapsing the rep path and + // forcing the long-hash to mint creeping offsets. + if cand_pos_r >= wlow1 { let cand_idx_r = cand_pos_r - history_abs_start; // 4-byte gate; full forward count only if it passes. let cand4 = unsafe { @@ -2163,7 +2176,7 @@ macro_rules! start_matching_fast_loop_body { match_len, lit_len_ip1, ); - break 'inner InnerExit::Committed(rep_cand); + break 'inner InnerExit::Committed(rep_cand, 0, abs_ip0); } } } @@ -2255,7 +2268,18 @@ macro_rules! start_matching_fast_loop_body { match_len, lit_len_ip0, ); - break 'inner InnerExit::Committed(cand); + // Upstream zstd `_match_found` (zstd_double_fast.c:287): + // `if (step < 4) hashLong[hl1] = ip1`. Insert the + // lookahead position's long hash before storing the + // match — safe only while `step < 4` (then `ip1` is + // below the post-match cursor `ip + mLength`). + if step < 4 { + let packed_ip1 = ((abs_ip1 - position_base) as u32) + 1; + unsafe { + *long_hash_ptr.add(hl1_idx) = packed_ip1; + } + } + break 'inner InnerExit::Committed(cand, 1, abs_ip0); } } } @@ -2324,7 +2348,7 @@ macro_rules! start_matching_fast_loop_body { match_len, lit_len_ip0, ); - break 'inner InnerExit::Committed(cand); + break 'inner InnerExit::Committed(cand, 3, abs_ip0); } } } @@ -2533,7 +2557,15 @@ macro_rules! start_matching_fast_loop_body { } } if short_hit_valid || retry_upgraded { - break 'inner InnerExit::Committed(chosen); + // Upstream zstd `_match_found` (zstd_double_fast.c:287): + // `if (step < 4) hashLong[hl1] = ip1`. + if step < 4 { + let packed_ip1 = ((abs_ip1 - position_base) as u32) + 1; + unsafe { + *long_hash_ptr.add(hl1_idx) = packed_ip1; + } + } + break 'inner InnerExit::Committed(chosen, 2, abs_ip0); } // Below-floor short hit with no retry // upgrade — fall through to the step bump @@ -2611,7 +2643,7 @@ macro_rules! start_matching_fast_loop_body { lit_len_ip0, ); if dcand.match_len >= DFAST_MIN_MATCH_LEN { - break 'inner InnerExit::Committed(dcand); + break 'inner InnerExit::Committed(dcand, 4, abs_ip0); } } } @@ -2619,8 +2651,10 @@ macro_rules! start_matching_fast_loop_body { } // Step bump on distance (upstream zstd `zstd_double_fast.c:224-228`). + // Upstream grows the step unbounded (one per `kStepIncr` travelled); + // no cap, so the scan stride matches byte-for-byte. if ip1 >= next_step_pos { - step = (step + 1).min(DFAST_MAX_SKIP_STEP); + step += 1; next_step_pos += DFAST_SKIP_STEP_GROWTH_INTERVAL; } @@ -2639,11 +2673,28 @@ macro_rules! start_matching_fast_loop_body { }; match inner_exit { - InnerExit::Committed(candidate) => { + InnerExit::Committed(candidate, _path_tag, scan_pos) => { + // `DFTRACE` env gate: dump each committed match's path tag + + // (offset, match_len, literal_len) so the match-path / offset + // stream can be diffed against C ffi when chasing a dfast + // ratio divergence. The env is read ONCE into a cached flag + // (a per-commit `getenv` showed up at ~3% of small-frame + // encode); off by default, an atomic load in production. + #[cfg(feature = "std")] + if *DFTRACE_ENABLED.get_or_init(|| std::env::var_os("DFTRACE").is_some()) { + std::eprintln!( + "DFT path={} off={} ml={} ll={}", + _path_tag, + candidate.offset, + candidate.match_len, + candidate.start - $current_abs_start - literals_start, + ); + } let start = $self.emit_candidate( $current_abs_start, &mut literals_start, candidate, + scan_pos, $handle_sequence, ); pos = start + candidate.match_len; diff --git a/zstd/src/encoding/frame_compressor.rs b/zstd/src/encoding/frame_compressor.rs index b37e2eac1..c2c32ae19 100644 --- a/zstd/src/encoding/frame_compressor.rs +++ b/zstd/src/encoding/frame_compressor.rs @@ -701,30 +701,32 @@ pub(crate) struct CompressState { /// `CompressState` by hand must also supply a value. pub(crate) strategy_tag: crate::encoding::strategy::StrategyTag, /// Whether the HUF literal table build runs the #167 table-log search - /// (`true`) or the cheap single-build (`false`). For the Fast strategy the - /// search is a clean ratio win over upstream zstd but costs ~1.5 us per - /// literal section — negligible on large inputs, ~20% on small ones — and - /// the Fast matcher is byte-faithful to upstream zstd (so the cheap path - /// ties it). So it is gated ON only for large Fast frames; non-Fast - /// strategies always keep it (their matchers diverge, making the search - /// load-bearing for ratio — gating those is deferred). Set per frame - /// alongside `strategy_tag` via [`fast_huf_search_enabled`]. + /// (`true`) or the cheap single-build (`false`). The search is a clean + /// ratio win over upstream zstd but costs ~1.5 us per literal section — + /// negligible on large inputs, ~20% on small ones. The Fast and DoubleFast + /// matchers are byte-faithful to upstream zstd, so the cheap path ties them; + /// the search is therefore gated ON only for large (> 128 KiB) Fast and + /// DoubleFast frames. Higher strategies always keep it (their matchers + /// diverge, making the search load-bearing for ratio). Set per frame + /// alongside `strategy_tag` via [`huf_search_enabled`]. pub(crate) huf_optimal_search: bool, } -/// Whether the Fast-strategy HUF literal build should run the #167 table-log -/// search for a frame of `source_size` bytes (see -/// [`CompressState::huf_optimal_search`]). Non-Fast strategies always search. -/// For Fast, enable the search only above 128 KiB (one full block) — below -/// that the per-frame setup dominates and the search's ~1.5 us is a large -/// relative cost for a ratio gain the cheap path forgoes while still tying -/// upstream zstd. Unknown size (streaming) keeps the search for conservative -/// ratio. -pub(crate) fn fast_huf_search_enabled( +/// Whether the HUF literal build should run the #167 table-log search for a +/// frame of `source_size` bytes (see [`CompressState::huf_optimal_search`]). +/// The Fast and DoubleFast strategies do little matching work, so per-frame +/// HUF setup dominates a small frame and the search's ~1.5 us is a large +/// relative cost for a ratio gain the cheap single-build forgoes while still +/// tying upstream zstd. Gate the search to frames above 128 KiB (one full +/// block) for those two. Higher strategies do enough matching that the search +/// is a negligible fraction and always run it for the ratio edge. Unknown size +/// (streaming) keeps the search for conservative ratio. +pub(crate) fn huf_search_enabled( strategy: crate::encoding::strategy::StrategyTag, source_size: Option, ) -> bool { - if strategy != crate::encoding::strategy::StrategyTag::Fast { + use crate::encoding::strategy::StrategyTag; + if !matches!(strategy, StrategyTag::Fast | StrategyTag::Dfast) { return true; } match source_size { @@ -802,10 +804,12 @@ fn initial_all_blocks_cap(initial_size_hint: Option, block_capacity: usize) /// `fill_block` appends source bytes to `buf` (which already holds any /// carried pre-split suffix) until `buf.len() == block_capacity` or the /// source is exhausted, returning `(bytes_appended, reached_eof)`. -/// `reached_eof` is true iff the block could NOT be filled to -/// `block_capacity` — the boundary the historical `Read`-loop produced (an -/// input that is an exact multiple of the block size still yields a -/// trailing empty last block on the next iteration). +/// `reached_eof` is true when no more input follows this block: either the +/// block could not be filled to `block_capacity`, or it filled exactly and the +/// source is confirmed exhausted (the slice knows its length; the reader probes +/// one byte ahead). An input that is an exact multiple of the block size +/// therefore marks its final full block `last_block` rather than emitting a +/// spurious trailing empty block. /// /// The slice impl exists so the slice entry points /// (`compress_independent_frame_into`, `compress_oneshot_*` fallbacks) @@ -832,7 +836,14 @@ impl OwnedBlockSource for &[u8] { let take = want.min(self.len()); buf.extend_from_slice(&self[..take]); *self = &self[take..]; - (take, take < want) + // EOF when this fill could not top the block to `block_capacity` + // (`take < want`) OR it exactly consumed the last input bytes + // (`self` now empty). The slice knows its own length, so a block that + // exactly fills capacity at end-of-input is reported as the final + // block here — the loop marks it `last_block` instead of emitting a + // spurious trailing empty block on the next iteration. Mirrors the C + // encoder, which marks the last real block last on `ZSTD_e_end`. + (take, take < want || self.is_empty()) } } @@ -840,7 +851,26 @@ impl OwnedBlockSource for &[u8] { /// preserves the historical sizing behaviour — an initialized target region /// bounded by the source-size hint, grown (doubling, capped) only when the /// hint under-counted. -pub(crate) struct ReaderBlockSource(pub(crate) Rd); +/// `peeked` holds a single look-ahead byte: when a block fills exactly to +/// `block_capacity`, `fill_block` reads one more byte to learn whether the +/// stream ended on that boundary. A `None` from that probe sets EOF (so the +/// just-filled block is marked last, mirroring the C encoder on `ZSTD_e_end`); +/// a byte is stashed here and prepended to the next block instead of leaking a +/// spurious trailing empty block when the input is an exact multiple of the +/// block size. +pub(crate) struct ReaderBlockSource { + pub(crate) reader: Rd, + peeked: Option, +} + +impl ReaderBlockSource { + pub(crate) fn new(reader: Rd) -> Self { + Self { + reader, + peeked: None, + } + } +} impl OwnedBlockSource for ReaderBlockSource { fn fill_block( @@ -852,6 +882,14 @@ impl OwnedBlockSource for ReaderBlockSource { let start = buf.len(); let mut filled = start; let mut reached_eof = false; + // Prepend the look-ahead byte read past the previous full block. In + // stream order it follows any carried pre-split suffix already in + // `buf`, so it is appended after that suffix and counted as part of + // this block's appended bytes. + if let Some(b) = self.peeked.take() { + buf.push(b); + filled += 1; + } // Size the read buffer to the bytes this block actually expects // rather than always zero-filling a full MAX_BLOCK_SIZE: a small // frame otherwise pays a 128 KiB `resize(_, 0)` memset per block @@ -892,13 +930,30 @@ impl OwnedBlockSource for ReaderBlockSource { buf.resize(grow_to, 0); } let read_end = buf.len(); - let new_bytes = self.0.read(&mut buf[filled..read_end]).unwrap(); + let new_bytes = self.reader.read(&mut buf[filled..read_end]).unwrap(); if new_bytes == 0 { reached_eof = true; break; } filled += new_bytes; } + // Look ahead one byte when the block filled exactly to capacity: a + // 0-byte read means the stream ended on the block boundary, so this + // block is the last one (the loop marks it `last_block`); otherwise + // stash the byte for the next block. Without this, an input that is an + // exact multiple of the block size would emit a spurious trailing + // empty block (the next iteration reads 0 and serializes an empty + // last Raw block). A blocking reader's probe read is consistent with + // the existing pull model — the next `fill_block` would block on the + // same byte anyway. + if !reached_eof && filled == block_capacity { + let mut probe = [0u8; 1]; + if self.reader.read(&mut probe).unwrap() == 0 { + reached_eof = true; + } else { + self.peeked = Some(probe[0]); + } + } buf.truncate(filled); (filled - start, reached_eof) } @@ -978,7 +1033,7 @@ impl FrameCompressor { crate::encoding::strategy::StrategyTag::for_compression_level(self.compression_level) }); self.state.huf_optimal_search = - fast_huf_search_enabled(self.state.strategy_tag, self.source_size_hint); + huf_search_enabled(self.state.strategy_tag, self.source_size_hint); self.state.matcher.set_param_overrides(Some(overrides)); } @@ -1506,7 +1561,7 @@ impl FrameCompressor { prep.initial_size_hint, self.block_capacity(), )); - let mut block_source = ReaderBlockSource(&mut source); + let mut block_source = ReaderBlockSource::new(&mut source); let total_uncompressed = self.run_owned_block_loop( &mut block_source, prep.initial_size_hint, @@ -1575,7 +1630,7 @@ impl FrameCompressor { // `initial_size_hint` (captured before the `.take()` above) — by here // `self.source_size_hint` is None. self.state.huf_optimal_search = - fast_huf_search_enabled(self.state.strategy_tag, initial_size_hint); + huf_search_enabled(self.state.strategy_tag, initial_size_hint); let cached_entropy = if use_dictionary_state { self.dictionary_entropy_cache.as_ref() } else { diff --git a/zstd/src/encoding/frame_compressor/tests.rs b/zstd/src/encoding/frame_compressor/tests.rs index 69a0f5f69..1b7cb6754 100644 --- a/zstd/src/encoding/frame_compressor/tests.rs +++ b/zstd/src/encoding/frame_compressor/tests.rs @@ -2514,3 +2514,115 @@ fn compress_independent_frame_reuses_sticky_dictionary() { assert_eq!(&decoded, data, "dict roundtrip failed, len={}", data.len()); } } + +/// Walk a frame's block list, returning `(block_type, block_size, last)` per +/// physical block. `block_type`: 0 = Raw, 1 = RLE, 2 = Compressed. +fn frame_block_list(frame: &[u8]) -> Vec<(u8, usize, bool)> { + let desc = frame[4]; + let fcs_flag = desc >> 6; + let single_segment = (desc >> 5) & 1 == 1; + let checksum = (desc >> 2) & 1 == 1; + let dict_id_bytes = match desc & 3 { + 0 => 0, + 1 => 1, + 2 => 2, + _ => 4, + }; + let fcs_bytes = match fcs_flag { + 0 => usize::from(single_segment), + 1 => 2, + 2 => 4, + _ => 8, + }; + let mut pos = 4 + 1 + usize::from(!single_segment) + dict_id_bytes + fcs_bytes; + let end = frame.len() - if checksum { 4 } else { 0 }; + let mut blocks = Vec::new(); + while pos + 3 <= end { + let h = + frame[pos] as usize | (frame[pos + 1] as usize) << 8 | (frame[pos + 2] as usize) << 16; + let last = h & 1 == 1; + let btype = ((h >> 1) & 3) as u8; + let bsize = h >> 3; + let advance = if btype == 1 { 1 } else { bsize }; + blocks.push((btype, bsize, last)); + pos += 3 + advance; + if last { + break; + } + } + blocks +} + +/// An input that is an exact multiple of `MAX_BLOCK_SIZE` must NOT emit a +/// spurious trailing empty Raw block: the last REAL block carries the +/// `last_block` flag, matching the C encoder on `ZSTD_e_end`. Exercises both +/// the one-shot slice loop (`compress_independent_frame`) and the streaming +/// `Read` loop (`set_source` + `compress`), on incompressible input (Raw +/// blocks) and highly compressible input (Compressed/RLE blocks). Before the +/// fix, each frame ended with an extra `R0!` block (3 wasted bytes). +#[test] +fn exact_block_multiple_marks_last_real_block() { + // The fix is driven by `block_capacity`, so cover both the default 128 KiB + // cap (`None`) and a smaller configured `target_block_size` — the EOF path + // must mark the last real block in both. + const CUSTOM_CAP: u32 = 16 * 1024; + for &(cap, target) in &[ + (MAX_BLOCK_SIZE as usize, None), + (CUSTOM_CAP as usize, Some(CUSTOM_CAP)), + ] { + for &nblk in &[1usize, 2, 3] { + for &compressible in &[false, true] { + let input: Vec = if compressible { + vec![0x7Au8; cap * nblk] + } else { + generate_data(0xC0FF_EE11, cap * nblk) + }; + + // One-shot slice path. + let mut oneshot: FrameCompressor = + FrameCompressor::new(super::CompressionLevel::Default); + oneshot.set_target_block_size(target); + let frame_os = oneshot.compress_independent_frame(&input); + let blocks_os = frame_block_list(&frame_os); + let last_os = *blocks_os.last().expect("at least one block"); + assert!( + last_os.2, + "one-shot last block must set last_block (cap={cap}, nblk={nblk}, compressible={compressible}): {blocks_os:?}" + ); + assert!( + !(last_os.0 == 0 && last_os.1 == 0), + "one-shot must not emit a trailing empty Raw block (cap={cap}, nblk={nblk}, compressible={compressible}): {blocks_os:?}" + ); + + // Streaming Read loop. + let mut output = Vec::new(); + let mut streaming = FrameCompressor::new(super::CompressionLevel::Default); + streaming.set_target_block_size(target); + streaming.set_source(input.as_slice()); + streaming.set_drain(&mut output); + streaming.compress(); + let blocks_st = frame_block_list(&output); + let last_st = *blocks_st.last().expect("at least one block"); + assert!( + last_st.2, + "streaming last block must set last_block (cap={cap}, nblk={nblk}, compressible={compressible}): {blocks_st:?}" + ); + assert!( + !(last_st.0 == 0 && last_st.1 == 0), + "streaming must not emit a trailing empty Raw block (cap={cap}, nblk={nblk}, compressible={compressible}): {blocks_st:?}" + ); + + // Both frames must round-trip back to the original bytes. + for frame in [&frame_os, &output] { + let mut decoder = FrameDecoder::new(); + let mut decoded = Vec::with_capacity(input.len()); + decoder.decode_all_to_vec(frame, &mut decoded).unwrap(); + assert_eq!( + decoded, input, + "roundtrip mismatch (cap={cap}, nblk={nblk}, compressible={compressible})" + ); + } + } + } + } +} diff --git a/zstd/src/encoding/match_generator/mod.rs b/zstd/src/encoding/match_generator/mod.rs index d992ffd42..c71554f28 100644 --- a/zstd/src/encoding/match_generator/mod.rs +++ b/zstd/src/encoding/match_generator/mod.rs @@ -103,9 +103,14 @@ pub(crate) const DFAST_EMPTY_SLOT: u32 = 0; /// advance `position_base` so future inserts stay inside the `u32` /// window. Same scheme as `encoding/ldm/table.rs`. pub(crate) const DFAST_REBASE_GUARD_BAND: u32 = 1u32 << 30; -pub(crate) const DFAST_SKIP_SEARCH_STRENGTH: usize = 6; +// `kSearchStrength` (upstream `zstd_compress_internal.h:32`). The dfast step +// ramp grows one position every `1 << kSearchStrength` = 256 bytes travelled +// (upstream `kStepIncr`, zstd_double_fast.c:131). A smaller value accelerates +// the scan faster and skips source positions upstream still inserts, which +// drops the short matches upstream finds at a block start — so the +// `#167`-disabled path must use the upstream 8 to stay byte-identical. +pub(crate) const DFAST_SKIP_SEARCH_STRENGTH: usize = 8; pub(crate) const DFAST_SKIP_STEP_GROWTH_INTERVAL: usize = 1 << DFAST_SKIP_SEARCH_STRENGTH; -pub(crate) const DFAST_MAX_SKIP_STEP: usize = 8; pub(crate) const DFAST_INCOMPRESSIBLE_SKIP_STEP: usize = 16; pub(crate) const ROW_HASH_BITS: usize = 20; pub(crate) const ROW_LOG: usize = 5; diff --git a/zstd/src/encoding/streaming_encoder.rs b/zstd/src/encoding/streaming_encoder.rs index 9225c9f3e..a5f64fa13 100644 --- a/zstd/src/encoding/streaming_encoder.rs +++ b/zstd/src/encoding/streaming_encoder.rs @@ -120,7 +120,7 @@ impl StreamingEncoder { self.state.strategy_tag = self.strategy_override.unwrap_or_else(|| { crate::encoding::strategy::StrategyTag::for_compression_level(self.compression_level) }); - self.state.huf_optimal_search = crate::encoding::frame_compressor::fast_huf_search_enabled( + self.state.huf_optimal_search = crate::encoding::frame_compressor::huf_search_enabled( self.state.strategy_tag, self.pledged_content_size.or(self.source_size_hint), ); @@ -574,7 +574,7 @@ impl StreamingEncoder { self.state.strategy_tag = self.strategy_override.unwrap_or_else(|| { crate::encoding::strategy::StrategyTag::for_compression_level(self.compression_level) }); - self.state.huf_optimal_search = crate::encoding::frame_compressor::fast_huf_search_enabled( + self.state.huf_optimal_search = crate::encoding::frame_compressor::huf_search_enabled( self.state.strategy_tag, self.pledged_content_size.or(self.source_size_hint), ); diff --git a/zstd/src/huff0/huff0_decoder.rs b/zstd/src/huff0/huff0_decoder.rs index 0523cd4b6..ecc350356 100644 --- a/zstd/src/huff0/huff0_decoder.rs +++ b/zstd/src/huff0/huff0_decoder.rs @@ -259,6 +259,26 @@ impl<'t> HuffmanDecoder<'t> { packed as u8 } + /// Decode one symbol WITHOUT refilling the bit reader. The caller must have + /// guaranteed enough buffered bits (via `ensure_bits`) for this and any + /// batched siblings. Mirrors upstream zstd `HUF_decodeSymbolX1` + /// (`BIT_lookBitsFast` + `BIT_skipBits`, no per-symbol reload): the drain + /// reloads once per group of symbols instead of once per symbol, so the + /// tail symbols near a stream end don't each pay `refill_slow`. The advance + /// uses the scalar form (byte-identical to the BMI2 `_bzhi` path) since the + /// drain handles only a handful of trailing symbols per stream. + #[inline(always)] + pub fn decode_symbol_and_advance_no_refill( + &mut self, + br: &mut BitReaderReversed<'_, K>, + ) -> u8 { + let packed = self.table.packed_decode[self.state as usize]; + let num_bits = (packed >> 8) as u8; + let new_bits = br.get_bits_unchecked(num_bits); + self.state = ((self.state << num_bits) & self.table.state_mask) | new_bits; + packed as u8 + } + #[cfg(any(target_arch = "x86", target_arch = "x86_64"))] #[target_feature(enable = "bmi2")] unsafe fn decode_symbol_and_advance_x86_bmi2( diff --git a/zstd/src/huff0/huff0_encoder.rs b/zstd/src/huff0/huff0_encoder.rs index e9ca137b7..90c52e1ce 100644 --- a/zstd/src/huff0/huff0_encoder.rs +++ b/zstd/src/huff0/huff0_encoder.rs @@ -346,7 +346,16 @@ impl>> HuffmanEncoder<'_, '_, V> { } let raw_description_is_representable = weights.len() <= 128; - let raw_description_bytes = weights.len().div_ceil(2); + // Upstream zstd `HUF_writeCTable_wksp` (huf_compress.c:276) keeps the FSE + // weight description ONLY when `hSize < maxSymbolValue/2` (FLOOR). Our + // `weights.len()` equals `maxSymbolValue` (the implicit last symbol is not + // written; see `write_raw_weight_description`), so the threshold is + // `weights.len() / 2`. `div_ceil` was 1 byte too permissive on odd symbol + // counts: it kept the FSE description in a boundary band where upstream + // emits the cheap direct nibbles, producing a frame that is fractionally + // smaller but markedly slower to decode (the decoder pays an FSE + // weight-table build + FSE weight decode instead of a nibble unpack). + let raw_description_bytes = weights.len() / 2; if encoded.len() > 1 && (encoded.len() < raw_description_bytes || !raw_description_is_representable) { diff --git a/zstd/src/io_nostd.rs b/zstd/src/io_nostd.rs index 630979ab7..5dbc0a5d8 100644 --- a/zstd/src/io_nostd.rs +++ b/zstd/src/io_nostd.rs @@ -169,6 +169,22 @@ impl Read for &[u8] { *self = rest; Ok(size) } + + /// Direct slice read_exact: one bounds check + one copy + advance, instead + /// of the default trait loop (`read` in a `while` with a re-slice and a + /// trailing emptiness check per call). On the in-memory decode path the + /// frame-header parser issues several small `read_exact`s per frame, so the + /// default loop's per-call overhead dominated a tiny-frame decompress. + #[inline] + fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), Error> { + if self.len() < buf.len() { + return Err(Error::from(ErrorKind::UnexpectedEof)); + } + let (head, rest) = self.split_at(buf.len()); + buf.copy_from_slice(head); + *self = rest; + Ok(()) + } } impl Read for &mut T @@ -178,6 +194,14 @@ where fn read(&mut self, buf: &mut [u8]) -> Result { (*self).read(buf) } + + /// Delegate to the inner reader's `read_exact` so a `&mut &[u8]` (the + /// in-memory decode source) reaches the direct slice override above rather + /// than the default trait loop. + #[inline] + fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), Error> { + (*self).read_exact(buf) + } } pub struct Take {