perf(ldm): close the level-1 long-distance gap to libzstd

zstd/examples/decode_loop_z000033.rs

@@ -22,10 +22,14 @@ fn main() {
     let iters: u32 = args.get(2).and_then(|s| s.parse().ok()).unwrap_or(50_000);
     let mode: &str = args.get(3).map(|s| s.as_str()).unwrap_or("rust");
     let corpus_path: Option<&str> = args.get(4).map(|s| s.as_str());
-    // 6th arg: "checksum" → encode with content_checksum_flag = 1 so the
-    // decode loop exercises the post-decode XXH64 verify pass. Lets us
-    // isolate the checksum cost (flag-on vs flag-off) in one harness.
-    let checksum: bool = args.get(5).map(|s| s == "checksum").unwrap_or(false);
+    // 6th arg, comma-separated encode options: "checksum" sets
+    // content_checksum_flag = 1 so the decode loop exercises the post-decode
+    // XXH64 verify pass; "ldm" turns on long-distance matching. The
+    // benchmark's `level_<N>_*_ldm` decode cells read frames with both, so
+    // reproduce one with `checksum,ldm`.
+    let options: Vec<&str> = args.get(5).map_or(Vec::new(), |s| s.split(',').collect());
+    let checksum = options.contains(&"checksum");
+    let ldm = options.contains(&"ldm");
 
     // Source resolution. This example is named after the `z000033` corpus,
     // so an explicit path or the in-tree corpus is the contract. The LCG
@@ -94,8 +98,8 @@ fn main() {
     // FFI encode once at requested level.
     let dst_cap = unsafe { zstd_sys::ZSTD_compressBound(src.len()) };
     let mut compressed: Vec<u8> = vec![0u8; dst_cap];
-    let written = if checksum {
-        // ZSTD_compress2 honours CCtx params, including the checksum flag.
+    let written = if checksum || ldm {
+        // ZSTD_compress2 honours CCtx params, the checksum and LDM flags.
         let cctx = unsafe { zstd_sys::ZSTD_createCCtx() };
         assert!(!cctx.is_null(), "ZSTD_createCCtx failed");
         unsafe {
@@ -107,7 +111,12 @@ fn main() {
             zstd_sys::ZSTD_CCtx_setParameter(
                 cctx,
                 zstd_sys::ZSTD_cParameter::ZSTD_c_checksumFlag,
-                1,
+                i32::from(checksum),
+            );
+            zstd_sys::ZSTD_CCtx_setParameter(
+                cctx,
+                zstd_sys::ZSTD_cParameter::ZSTD_c_enableLongDistanceMatching,
+                i32::from(ldm),
             );
         }
         let w = unsafe {

zstd/examples/encode_loop_oneshot_z000033.rs

@@ -11,14 +11,21 @@
 //! Output buffer allocated once and reused, as in the sibling binaries.
 //!
 //! Build: `cargo build --release -p ffi-bench --example encode_loop_oneshot_z000033 --features dict-builder`
-//! Run:   `perf stat -e cycles,instructions ./encode_loop_oneshot_z000033 <level> <iters> <corpus>`
+//! Run:   `perf stat -e cycles,instructions ./encode_loop_oneshot_z000033 <level> <iters> <corpus> [ldm]`
+//!
+//! With `ldm` it times what the benchmark's `level_<N>_*_ldm` cells time: a
+//! fresh compressor per frame, long-distance matching on, content checksum
+//! on; `ffi_encode_loop_z000033 ... ldm` is the C side of the same cell.
 
 use std::env;
 use std::fs;
 
-use structured_zstd::encoding::{CompressionLevel, FrameCompressor};
+use structured_zstd::encoding::{CompressionLevel, CompressionParameters, FrameCompressor};
 
 fn main() {
+    if env::args().nth(4).as_deref() == Some("ldm") {
+        return ldm_loop();
+    }
     let level: i32 = env::args().nth(1).and_then(|s| s.parse().ok()).unwrap_or(3);
     let iters: usize = env::args()
         .nth(2)
@@ -46,3 +53,34 @@ fn main() {
         sum
     );
 }
+
+fn ldm_loop() {
+    let level: i32 = env::args().nth(1).and_then(|s| s.parse().ok()).unwrap_or(1);
+    let iters: usize = env::args()
+        .nth(2)
+        .and_then(|s| s.parse().ok())
+        .unwrap_or(40);
+    let corpus_path = env::args().nth(3).expect("corpus path");
+    let bytes = fs::read(&corpus_path).expect("read corpus");
+    // `CompressionLevel::Level`, as the benchmark's level inventory uses.
+    let params = CompressionParameters::builder(CompressionLevel::Level(level))
+        .enable_long_distance_matching(true)
+        .build()
+        .expect("LDM-only parameters");
+    let mut out: Vec<u8> = Vec::new();
+    let mut sum = 0usize;
+    for _ in 0..iters {
+        let mut enc: FrameCompressor = FrameCompressor::new(CompressionLevel::Level(level));
+        enc.set_parameters(&params);
+        enc.set_content_checksum(true);
+        enc.compress_independent_frame_into(&bytes[..], &mut out);
+        sum += out.len();
+    }
+    println!(
+        "encoded {} bytes x {} iters at level {} with ldm; last-out-sum={}",
+        bytes.len(),
+        iters,
+        level,
+        sum
+    );
+}

zstd/examples/ffi_encode_loop_z000033.rs

@@ -9,14 +9,21 @@
 //! overhead.
 //!
 //! Build: `cargo build --release -p ffi-bench --example ffi_encode_loop_z000033`
-//! Run:   `perf stat -e cycles,instructions ./ffi_encode_loop_z000033 <level> <iters> <corpus>`
+//! Run:   `perf stat -e cycles,instructions ./ffi_encode_loop_z000033 <level> <iters> <corpus> [ldm]`
+//!
+//! With `ldm` it times the benchmark's `level_<N>_*_ldm` C arm: a fresh
+//! context per frame through `ZSTD_compress2`, long-distance matching on,
+//! content checksum and content size on.
 
 use std::env;
 use std::fs;
 
 use zstd::zstd_safe::zstd_sys;
 
 fn main() {
+    if env::args().nth(4).as_deref() == Some("ldm") {
+        return ldm_loop();
+    }
     let level: i32 = env::args().nth(1).and_then(|s| s.parse().ok()).unwrap_or(3);
     let iters: usize = env::args()
         .nth(2)
@@ -57,3 +64,54 @@ fn main() {
         sum
     );
 }
+
+fn ldm_loop() {
+    use zstd_sys::ZSTD_cParameter as P;
+    let level: i32 = env::args().nth(1).and_then(|s| s.parse().ok()).unwrap_or(1);
+    let iters: usize = env::args()
+        .nth(2)
+        .and_then(|s| s.parse().ok())
+        .unwrap_or(40);
+    let corpus_path = env::args().nth(3).expect("corpus path");
+    let bytes = fs::read(&corpus_path).expect("read corpus");
+    let bound = unsafe { zstd_sys::ZSTD_compressBound(bytes.len()) };
+    let mut out = vec![0u8; bound];
+    let mut sum = 0usize;
+    for _ in 0..iters {
+        // SAFETY: every call operates on the context created here, freed at
+        // the end of the iteration; errors are asserted.
+        unsafe {
+            let cctx = zstd_sys::ZSTD_createCCtx();
+            assert!(!cctx.is_null(), "ZSTD_createCCtx returned null");
+            for (param, value) in [
+                (P::ZSTD_c_compressionLevel, level),
+                (P::ZSTD_c_checksumFlag, 1),
+                (P::ZSTD_c_contentSizeFlag, 1),
+                (P::ZSTD_c_enableLongDistanceMatching, 1),
+            ] {
+                let rc = zstd_sys::ZSTD_CCtx_setParameter(cctx, param, value);
+                assert!(zstd_sys::ZSTD_isError(rc) == 0, "setParameter failed");
+            }
+            let written = zstd_sys::ZSTD_compress2(
+                cctx,
+                out.as_mut_ptr().cast(),
+                out.len(),
+                bytes.as_ptr().cast(),
+                bytes.len(),
+            );
+            assert!(
+                zstd_sys::ZSTD_isError(written) == 0,
+                "ZSTD_compress2 failed"
+            );
+            sum += written;
+            zstd_sys::ZSTD_freeCCtx(cctx);
+        }
+    }
+    println!(
+        "encoded {} bytes x {} iters at level {} with ldm; last-out-sum={}",
+        bytes.len(),
+        iters,
+        level,
+        sum
+    );
+}

zstd/src/encoding/frame_compressor.rs

@@ -2281,6 +2281,17 @@ impl<R: Read, W: Write> FrameCompressor<R, W, MatchGeneratorDriver> {
                 block_capacity,
                 savings,
             );
+            // A cut the splitter proposes inside a long match is moved out of
+            // it, as the owned loop does for the full block it has read.
+            let available = (input.len() - start).min(block_capacity);
+            let block_len = if block_len < available && available == block_capacity {
+                guard
+                    .0
+                    .matcher
+                    .settle_block_cut_with::<K>(available, block_len)
+            } else {
+                block_len
+            };
             let end = (start + block_len).min(input.len());
             let block = &input[start..end];
             let last_block = end == input.len();

zstd/src/encoding/ldm/gear_hash.rs

@@ -128,6 +128,35 @@ impl GearHashState {
     }
 }
 
+/// One byte through the rolling hash: `(hash << 1) + GEAR_TAB[byte]`.
+#[inline(always)]
+fn gear_step(hash: u64, byte: u8) -> u64 {
+    let entry = GEAR_TAB[byte as usize];
+    #[cfg(target_arch = "x86_64")]
+    {
+        // One `lea`, a single cycle on the hash's dependency chain, as
+        // upstream's compiled `ZSTD_ldm_gear_feed` has it. Written plainly,
+        // the table load folds into an `add` behind `add hash, hash`: two
+        // cycles a byte on the chain the whole scan waits on.
+        let mut hash = hash;
+        // SAFETY: register arithmetic only; no memory, stack or flags.
+        unsafe {
+            core::arch::asm!(
+                "lea {h}, [{e} + {h}*2]",
+                h = inout(reg) hash,
+                e = in(reg) entry,
+                options(pure, nomem, nostack, preserves_flags),
+            );
+        }
+        hash
+    }
+    // Elsewhere the shift folds into the add (aarch64 `add x, t, h, lsl #1`).
+    #[cfg(not(target_arch = "x86_64"))]
+    {
+        (hash << 1).wrapping_add(entry)
+    }
+}
+
 /// Feed `min_match_length` bytes through the rolling hash WITHOUT
 /// emitting any splits.
 ///
@@ -148,14 +177,14 @@ pub(crate) fn reset(state: &mut GearHashState, data: &[u8]) {
     // arithmetic monomorphises to the same instruction sequence on
     // every backend.
     while n + 3 < data.len() {
-        hash = (hash << 1).wrapping_add(GEAR_TAB[data[n] as usize]);
-        hash = (hash << 1).wrapping_add(GEAR_TAB[data[n + 1] as usize]);
-        hash = (hash << 1).wrapping_add(GEAR_TAB[data[n + 2] as usize]);
-        hash = (hash << 1).wrapping_add(GEAR_TAB[data[n + 3] as usize]);
+        hash = gear_step(hash, data[n]);
+        hash = gear_step(hash, data[n + 1]);
+        hash = gear_step(hash, data[n + 2]);
+        hash = gear_step(hash, data[n + 3]);
         n += 4;
     }
     while n < data.len() {
-        hash = (hash << 1).wrapping_add(GEAR_TAB[data[n] as usize]);
+        hash = gear_step(hash, data[n]);
         n += 1;
     }
     state.rolling = hash;
@@ -189,71 +218,58 @@ pub(crate) fn feed(state: &mut GearHashState, data: &[u8], splits: &mut [usize])
     );
     let mut hash = state.rolling;
     let mask = state.stop_mask;
-    let mut n: usize = 0;
     let mut num_splits: usize = 0;
 
-    // 4-wide unrolled head. Upstream zstd `zstd_ldm.c:118-123`. Each iteration
-    // increments `n` first, then tests the post-update hash so the
-    // split index points to the byte AFTER the matched window — this
-    // matches `splits[*numSplits] = n` in upstream zstd (n already
-    // incremented).
-    while n + 3 < data.len() {
-        hash = (hash << 1).wrapping_add(GEAR_TAB[data[n] as usize]);
-        n += 1;
-        if hash & mask == 0 {
-            splits[num_splits] = n;
-            num_splits += 1;
-            if num_splits == LDM_BATCH_SIZE {
-                state.rolling = hash;
-                return (n, num_splits);
-            }
-        }
-        hash = (hash << 1).wrapping_add(GEAR_TAB[data[n] as usize]);
-        n += 1;
-        if hash & mask == 0 {
-            splits[num_splits] = n;
-            num_splits += 1;
-            if num_splits == LDM_BATCH_SIZE {
-                state.rolling = hash;
-                return (n, num_splits);
-            }
-        }
-        hash = (hash << 1).wrapping_add(GEAR_TAB[data[n] as usize]);
-        n += 1;
-        if hash & mask == 0 {
-            splits[num_splits] = n;
-            num_splits += 1;
-            if num_splits == LDM_BATCH_SIZE {
-                state.rolling = hash;
-                return (n, num_splits);
+    // One byte at `$at`: the split index is the count of bytes consumed,
+    // `$at + 1`, the byte AFTER the matched window, as upstream's
+    // `splits[*numSplits] = n` with `n` already incremented. Positions are
+    // derived only here, on the rare split, so the scan carries a pointer
+    // and nothing else.
+    macro_rules! step {
+        ($byte:expr, $at:expr) => {{
+            hash = gear_step(hash, $byte);
+            if hash & mask == 0 {
+                let consumed = $at + 1;
+                if push_split(splits, &mut num_splits, consumed) {
+                    state.rolling = hash;
+                    return (consumed, num_splits);
+                }
             }
-        }
-        hash = (hash << 1).wrapping_add(GEAR_TAB[data[n] as usize]);
-        n += 1;
-        if hash & mask == 0 {
-            splits[num_splits] = n;
-            num_splits += 1;
-            if num_splits == LDM_BATCH_SIZE {
-                state.rolling = hash;
-                return (n, num_splits);
-            }
-        }
+        }};
     }
-    // Tail loop for the remaining 0..3 bytes. Upstream zstd `zstd_ldm.c:124-126`.
-    while n < data.len() {
-        hash = (hash << 1).wrapping_add(GEAR_TAB[data[n] as usize]);
-        n += 1;
-        if hash & mask == 0 {
-            splits[num_splits] = n;
-            num_splits += 1;
-            if num_splits == LDM_BATCH_SIZE {
-                state.rolling = hash;
-                return (n, num_splits);
-            }
-        }
+    // Four bytes a round, upstream zstd `zstd_ldm.c:118-123`; a chunk's length
+    // is known, so its loads carry no bounds check, and the loop's only
+    // counter is the chunk pointer, as upstream's is its index.
+    let base = data.as_ptr() as usize;
+    let (quads, rest) = data.as_chunks::<4>();
+    for quad in quads {
+        let at = quad.as_ptr() as usize - base;
+        step!(quad[0], at);
+        step!(quad[1], at + 1);
+        step!(quad[2], at + 2);
+        step!(quad[3], at + 3);
+    }
+    // The remaining 0..3 bytes, upstream zstd `zstd_ldm.c:124-126`.
+    let tail_at = data.len() - rest.len();
+    for (i, &byte) in rest.iter().enumerate() {
+        step!(byte, tail_at + i);
     }
     state.rolling = hash;
-    (n, num_splits)
+    (data.len(), num_splits)
+}
+
+/// Record a split at `consumed`; whether the batch is now full.
+///
+/// Out of line and cold: a split comes about once per `2 ^ hash_rate_log`
+/// bytes, and without the hint the record was laid out on the fall-through
+/// path, so every other byte took a branch around it. Upstream's compiled
+/// loop branches out to its record, as this one now does.
+#[cold]
+#[inline(never)]
+fn push_split(splits: &mut [usize], num_splits: &mut usize, consumed: usize) -> bool {
+    splits[*num_splits] = consumed;
+    *num_splits += 1;
+    *num_splits == LDM_BATCH_SIZE
 }
 
 /// 256-entry random-permutation table from upstream zstd

zstd/src/encoding/levels/fastest.rs

@@ -358,9 +358,10 @@ pub(crate) fn compress_block_encoded_borrowed<K: EncodeTier>(
     // incompressible.
     let is_rle = !block.is_empty() && block.iter().all(|x| block[0].eq(x));
     let window_size = state.matcher.window_size();
-    // As on the owned path: an attached dictionary keeps the block on the
-    // search rather than trusting a sample to say the dictionary is irrelevant.
-    let dict_rejects_raw = dict_active;
+    // As on the owned path: an attached dictionary, or long-distance matching,
+    // keeps the block on the search rather than trusting a sample to say
+    // matches from outside the block are irrelevant.
+    let dict_rejects_raw = dict_active || state.matcher.matches_beyond_window_tables();
     // As on the owned path: where no block may go out raw, the grid has nothing
     // to answer and is left alone.
     let raw_skip_reachable = compression_level_allows_raw_fast_path(compression_level, window_size);