feat(compress): move MT synchronization policy to Rust

Port the rsyncable synchronization-point scan and rolling hash policy behind a scalar ABI. Keep the MT scheduler, input buffers, and context ownership in C while adding boundary-focused Rust coverage for disabled, short, buffered, spanning, and hit/no-hit paths.

Test Plan:

- cargo test --manifest-path rust/Cargo.toml --no-default-features --features compression

- cargo clippy --manifest-path rust/Cargo.toml

- cargo clippy --manifest-path rust/Cargo.toml --benches

- cargo clippy --manifest-path rust/Cargo.toml --tests

- cargo +nightly fmt --manifest-path rust/Cargo.toml --all

- make -B -C lib -j2 lib

- make -B -C tests -j2 test-zstream (pending post-commit native gate)

lib/compress/zstdmt_compress.c

@@ -144,6 +144,11 @@ int ZSTDMT_rust_isOverlapped(const void* bufferStart, size_t bufferCapacity,
 int ZSTDMT_rust_doesOverlapWindow(const void* bufferStart, size_t bufferCapacity,
                                   const void* nextSrc, const void* base,
                                   const void* dictBase, U32 dictLimit, U32 lowLimit);
+void ZSTDMT_rust_findSynchronizationPoint(const void* inputSrc, size_t inputSize,
+                                           size_t inputPos, size_t targetSectionSize,
+                                           const void* inBuffStart, size_t inBuffFilled,
+                                           int rsyncable, U64 primePower, U64 hitMask,
+                                           size_t* toLoad, int* flush);
 
 typedef struct ZSTDMT_bufferPool_s {
     ZSTDMT_RustBufferPool* rustPool;
@@ -1647,97 +1652,14 @@ typedef struct {
 static SyncPoint
 findSynchronizationPoint(ZSTDMT_CCtx const* mtctx, ZSTD_inBuffer const input)
 {
-    BYTE const* const istart = (BYTE const*)input.src + input.pos;
-    U64 const primePower = mtctx->rsync.primePower;
-    U64 const hitMask = mtctx->rsync.hitMask;
-
     SyncPoint syncPoint;
-    U64 hash;
+    ZSTDMT_rust_findSynchronizationPoint(
-    BYTE const* prev;
+        input.src, input.size, input.pos,
-    size_t pos;
+        mtctx->targetSectionSize,
-
+        mtctx->inBuff.buffer.start, mtctx->inBuff.filled,
-    syncPoint.toLoad = MIN(input.size - input.pos, mtctx->targetSectionSize - mtctx->inBuff.filled);
+        mtctx->params.rsyncable,
-    syncPoint.flush = 0;
+        mtctx->rsync.primePower, mtctx->rsync.hitMask,
-    if (!mtctx->params.rsyncable)
+        &syncPoint.toLoad, &syncPoint.flush);
-        /* Rsync is disabled. */
-        return syncPoint;
-    if (mtctx->inBuff.filled + input.size - input.pos < RSYNC_MIN_BLOCK_SIZE)
-        /* We don't emit synchronization points if it would produce too small blocks.
-         * We don't have enough input to find a synchronization point, so don't look.
-         */
-        return syncPoint;
-    if (mtctx->inBuff.filled + syncPoint.toLoad < RSYNC_LENGTH)
-        /* Not enough to compute the hash.
-         * We will miss any synchronization points in this RSYNC_LENGTH byte
-         * window. However, since it depends only in the internal buffers, if the
-         * state is already synchronized, we will remain synchronized.
-         * Additionally, the probability that we miss a synchronization point is
-         * low: RSYNC_LENGTH / targetSectionSize.
-         */
-        return syncPoint;
-    /* Initialize the loop variables. */
-    if (mtctx->inBuff.filled < RSYNC_MIN_BLOCK_SIZE) {
-        /* We don't need to scan the first RSYNC_MIN_BLOCK_SIZE positions
-         * because they can't possibly be a sync point. So we can start
-         * part way through the input buffer.
-         */
-        pos = RSYNC_MIN_BLOCK_SIZE - mtctx->inBuff.filled;
-        if (pos >= RSYNC_LENGTH) {
-            prev = istart + pos - RSYNC_LENGTH;
-            hash = ZSTD_rollingHash_compute(prev, RSYNC_LENGTH);
-        } else {
-            assert(mtctx->inBuff.filled >= RSYNC_LENGTH);
-            prev = (BYTE const*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled - RSYNC_LENGTH;
-            hash = ZSTD_rollingHash_compute(prev + pos, (RSYNC_LENGTH - pos));
-            hash = ZSTD_rollingHash_append(hash, istart, pos);
-        }
-    } else {
-        /* We have enough bytes buffered to initialize the hash,
-         * and have processed enough bytes to find a sync point.
-         * Start scanning at the beginning of the input.
-         */
-        assert(mtctx->inBuff.filled >= RSYNC_MIN_BLOCK_SIZE);
-        assert(RSYNC_MIN_BLOCK_SIZE >= RSYNC_LENGTH);
-        pos = 0;
-        prev = (BYTE const*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled - RSYNC_LENGTH;
-        hash = ZSTD_rollingHash_compute(prev, RSYNC_LENGTH);
-        if ((hash & hitMask) == hitMask) {
-            /* We're already at a sync point so don't load any more until
-             * we're able to flush this sync point.
-             * This likely happened because the job table was full so we
-             * couldn't add our job.
-             */
-            syncPoint.toLoad = 0;
-            syncPoint.flush = 1;
-            return syncPoint;
-        }
-    }
-    /* Starting with the hash of the previous RSYNC_LENGTH bytes, roll
-     * through the input. If we hit a synchronization point, then cut the
-     * job off, and tell the compressor to flush the job. Otherwise, load
-     * all the bytes and continue as normal.
-     * If we go too long without a synchronization point (targetSectionSize)
-     * then a block will be emitted anyways, but this is okay, since if we
-     * are already synchronized we will remain synchronized.
-     */
-    assert(pos < RSYNC_LENGTH || ZSTD_rollingHash_compute(istart + pos - RSYNC_LENGTH, RSYNC_LENGTH) == hash);
-    for (; pos < syncPoint.toLoad; ++pos) {
-        BYTE const toRemove = pos < RSYNC_LENGTH ? prev[pos] : istart[pos - RSYNC_LENGTH];
-        /* This assert is very expensive, and Debian compiles with asserts enabled.
-         * So disable it for now. We can get similar coverage by checking it at the
-         * beginning & end of the loop.
-         * assert(pos < RSYNC_LENGTH || ZSTD_rollingHash_compute(istart + pos - RSYNC_LENGTH, RSYNC_LENGTH) == hash);
-         */
-        hash = ZSTD_rollingHash_rotate(hash, toRemove, istart[pos], primePower);
-        assert(mtctx->inBuff.filled + pos >= RSYNC_MIN_BLOCK_SIZE);
-        if ((hash & hitMask) == hitMask) {
-            syncPoint.toLoad = pos + 1;
-            syncPoint.flush = 1;
-            ++pos; /* for assert */
-            break;
-        }
-    }
-    assert(pos < RSYNC_LENGTH || ZSTD_rollingHash_compute(istart + pos - RSYNC_LENGTH, RSYNC_LENGTH) == hash);
     return syncPoint;
 }
 

rust/src/zstdmt_compress.rs

@@ -34,6 +34,13 @@ const ZSTD_PS_ENABLE: c_int = 1;
 #[cfg(test)]
 const ZSTD_PS_DISABLE: c_int = 2;
 
+const RSYNC_LENGTH: usize = 32;
+const RSYNC_MIN_BLOCK_LOG: usize = 17;
+const RSYNC_MIN_BLOCK_SIZE: usize = 1 << RSYNC_MIN_BLOCK_LOG;
+const PRIME8_BYTES: u64 = 0xCF1B_BCDC_B7A5_6463;
+const ROLL_HASH_CHAR_OFFSET: u64 = 10;
+const _: () = assert!(RSYNC_MIN_BLOCK_SIZE >= RSYNC_LENGTH);
+
 #[inline]
 fn cycle_log(chain_log: c_uint, strategy: c_int) -> c_uint {
     chain_log.wrapping_sub((strategy >= ZSTD_BTLAZY2) as c_uint)
@@ -139,6 +146,165 @@ pub extern "C" fn ZSTDMT_rust_computeOverlapSize(
     compute_overlap_size(windowLog, chainLog, strategy, overlapLog, enableLdm)
 }
 
+#[inline]
+unsafe fn rolling_hash_append(mut hash: u64, input: *const u8, size: usize) -> u64 {
+    for pos in 0..size {
+        let byte = u64::from(unsafe { *input.wrapping_add(pos) });
+        hash = hash
+            .wrapping_mul(PRIME8_BYTES)
+            .wrapping_add(byte.wrapping_add(ROLL_HASH_CHAR_OFFSET));
+    }
+    hash
+}
+
+#[inline]
+unsafe fn rolling_hash_compute(input: *const u8, size: usize) -> u64 {
+    unsafe { rolling_hash_append(0, input, size) }
+}
+
+#[inline]
+fn rolling_hash_rotate(hash: u64, to_remove: u8, to_add: u8, prime_power: u64) -> u64 {
+    hash.wrapping_sub(
+        u64::from(to_remove)
+            .wrapping_add(ROLL_HASH_CHAR_OFFSET)
+            .wrapping_mul(prime_power),
+    )
+    .wrapping_mul(PRIME8_BYTES)
+    .wrapping_add(u64::from(to_add).wrapping_add(ROLL_HASH_CHAR_OFFSET))
+}
+
+/// Search for a synchronization point without exposing the C MT context or
+/// its private buffer types across the FFI boundary.
+#[inline]
+unsafe fn find_synchronization_point(
+    input_src: *const c_void,
+    input_size: usize,
+    input_pos: usize,
+    target_section_size: usize,
+    in_buff_start: *const c_void,
+    in_buff_filled: usize,
+    rsyncable: c_int,
+    prime_power: u64,
+    hit_mask: u64,
+) -> (usize, c_int) {
+    let istart = input_src.cast::<u8>().wrapping_add(input_pos);
+    let available = input_size.wrapping_sub(input_pos);
+    let mut to_load = available.min(target_section_size.wrapping_sub(in_buff_filled));
+    let mut flush = 0;
+
+    if rsyncable == 0 {
+        return (to_load, flush);
+    }
+    if in_buff_filled.wrapping_add(available) < RSYNC_MIN_BLOCK_SIZE {
+        return (to_load, flush);
+    }
+    if in_buff_filled.wrapping_add(to_load) < RSYNC_LENGTH {
+        return (to_load, flush);
+    }
+
+    let (mut pos, prev, mut hash) = if in_buff_filled < RSYNC_MIN_BLOCK_SIZE {
+        let pos = RSYNC_MIN_BLOCK_SIZE.wrapping_sub(in_buff_filled);
+        if pos >= RSYNC_LENGTH {
+            let prev = istart.wrapping_add(pos.wrapping_sub(RSYNC_LENGTH));
+            let hash = unsafe { rolling_hash_compute(prev, RSYNC_LENGTH) };
+            (pos, prev, hash)
+        } else {
+            debug_assert!(in_buff_filled >= RSYNC_LENGTH);
+            let prev = in_buff_start
+                .cast::<u8>()
+                .wrapping_add(in_buff_filled.wrapping_sub(RSYNC_LENGTH));
+            let hash = unsafe { rolling_hash_compute(prev.wrapping_add(pos), RSYNC_LENGTH - pos) };
+            let hash = unsafe { rolling_hash_append(hash, istart, pos) };
+            (pos, prev, hash)
+        }
+    } else {
+        let pos = 0;
+        let prev = in_buff_start
+            .cast::<u8>()
+            .wrapping_add(in_buff_filled.wrapping_sub(RSYNC_LENGTH));
+        let hash = unsafe { rolling_hash_compute(prev, RSYNC_LENGTH) };
+        if (hash & hit_mask) == hit_mask {
+            return (0, 1);
+        }
+        (pos, prev, hash)
+    };
+
+    debug_assert!(
+        pos < RSYNC_LENGTH
+            || unsafe {
+                rolling_hash_compute(
+                    istart.wrapping_add(pos.wrapping_sub(RSYNC_LENGTH)),
+                    RSYNC_LENGTH,
+                ) == hash
+            }
+    );
+    while pos < to_load {
+        let to_remove = if pos < RSYNC_LENGTH {
+            unsafe { *prev.wrapping_add(pos) }
+        } else {
+            unsafe { *istart.wrapping_add(pos.wrapping_sub(RSYNC_LENGTH)) }
+        };
+        let to_add = unsafe { *istart.wrapping_add(pos) };
+        hash = rolling_hash_rotate(hash, to_remove, to_add, prime_power);
+        debug_assert!(in_buff_filled.wrapping_add(pos) >= RSYNC_MIN_BLOCK_SIZE);
+        if (hash & hit_mask) == hit_mask {
+            to_load = pos + 1;
+            flush = 1;
+            pos += 1;
+            break;
+        }
+        pos += 1;
+    }
+    debug_assert!(
+        pos < RSYNC_LENGTH
+            || unsafe {
+                rolling_hash_compute(
+                    istart.wrapping_add(pos.wrapping_sub(RSYNC_LENGTH)),
+                    RSYNC_LENGTH,
+                ) == hash
+            }
+    );
+    (to_load, flush)
+}
+
+/// C ABI for the pure MT synchronization-point policy.
+///
+/// The caller retains ownership of the MT context, input position, and
+/// buffer updates.  Rust only scans the two byte ranges and writes the two
+/// scalar results.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_findSynchronizationPoint(
+    inputSrc: *const c_void,
+    inputSize: usize,
+    inputPos: usize,
+    targetSectionSize: usize,
+    inBuffStart: *const c_void,
+    inBuffFilled: usize,
+    rsyncable: c_int,
+    primePower: u64,
+    hitMask: u64,
+    toLoad: *mut usize,
+    flush: *mut c_int,
+) {
+    let (computed_to_load, computed_flush) = unsafe {
+        find_synchronization_point(
+            inputSrc,
+            inputSize,
+            inputPos,
+            targetSectionSize,
+            inBuffStart,
+            inBuffFilled,
+            rsyncable,
+            primePower,
+            hitMask,
+        )
+    };
+    unsafe {
+        *toLoad = computed_to_load;
+        *flush = computed_flush;
+    }
+}
+
 type ZstdAllocFunction = unsafe extern "C" fn(*mut c_void, usize) -> *mut c_void;
 type ZstdFreeFunction = unsafe extern "C" fn(*mut c_void, *mut c_void);
 
@@ -849,6 +1015,264 @@ mod tests {
         opaque: ptr::null_mut(),
     };
 
+    fn rsync_prime_power() -> u64 {
+        (0..RSYNC_LENGTH - 1).fold(1, |power, _| power.wrapping_mul(PRIME8_BYTES))
+    }
+
+    fn patterned_bytes(length: usize, seed: u8) -> Vec<u8> {
+        let mut value = seed;
+        let mut bytes = vec![0; length];
+        for byte in &mut bytes {
+            value = value.wrapping_mul(17).wrapping_add(29);
+            *byte = value;
+        }
+        bytes
+    }
+
+    fn rolling_hash_bytes(bytes: &[u8]) -> u64 {
+        bytes.iter().fold(0, |hash, &byte| {
+            hash.wrapping_mul(PRIME8_BYTES)
+                .wrapping_add(u64::from(byte).wrapping_add(ROLL_HASH_CHAR_OFFSET))
+        })
+    }
+
+    fn call_synchronization_point(
+        input: &[u8],
+        input_pos: usize,
+        in_buff: &[u8],
+        in_buff_filled: usize,
+        target_section_size: usize,
+        rsyncable: c_int,
+        hit_mask: u64,
+    ) -> (usize, c_int) {
+        let mut to_load = usize::MAX;
+        let mut flush = -1;
+        unsafe {
+            ZSTDMT_rust_findSynchronizationPoint(
+                input.as_ptr().cast(),
+                input.len(),
+                input_pos,
+                target_section_size,
+                in_buff.as_ptr().cast(),
+                in_buff_filled,
+                rsyncable,
+                rsync_prime_power(),
+                hit_mask,
+                &mut to_load,
+                &mut flush,
+            );
+        }
+        (to_load, flush)
+    }
+
+    fn input_hashes_after_each_byte(
+        in_buff: &[u8],
+        in_buff_filled: usize,
+        input: &[u8],
+        input_pos: usize,
+    ) -> (u64, Vec<u64>) {
+        let mut history = in_buff[in_buff_filled - RSYNC_LENGTH..in_buff_filled].to_vec();
+        history.extend_from_slice(&input[input_pos..]);
+        let initial = rolling_hash_bytes(&history[..RSYNC_LENGTH]);
+        let hashes = (0..input.len() - input_pos)
+            .map(|pos| rolling_hash_bytes(&history[pos + 1..pos + 1 + RSYNC_LENGTH]))
+            .collect();
+        (initial, hashes)
+    }
+
+    fn hit_mask_for_index(initial: u64, hashes: &[u64], target: usize) -> u64 {
+        let target_hash = hashes[target];
+        let prior_matches = |mask: u64| {
+            (initial & mask) == mask || hashes[..target].iter().any(|hash| (hash & mask) == mask)
+        };
+        if target_hash != 0 && !prior_matches(target_hash) {
+            return target_hash;
+        }
+        for mask in 1..=u64::from(u16::MAX) {
+            if target_hash & mask == mask && !prior_matches(mask) {
+                return mask;
+            }
+        }
+        panic!("could not isolate synchronization hash at position {target}");
+    }
+
+    #[test]
+    fn synchronization_point_keeps_rsync_disabled() {
+        let input = patterned_bytes(20, 3);
+        let in_buff = patterned_bytes(64, 5);
+
+        assert_eq!(
+            call_synchronization_point(&input, 3, &in_buff, 40, 100, 0, 0),
+            (17, 0)
+        );
+    }
+
+    #[test]
+    fn synchronization_point_does_not_scan_a_short_hash_window() {
+        let input = patterned_bytes(RSYNC_LENGTH - 1, 7);
+        let in_buff = patterned_bytes(RSYNC_MIN_BLOCK_SIZE, 11);
+
+        assert_eq!(
+            call_synchronization_point(
+                &input,
+                0,
+                &in_buff,
+                0,
+                RSYNC_MIN_BLOCK_SIZE + input.len(),
+                1,
+                0,
+            ),
+            (input.len(), 0)
+        );
+    }
+
+    #[test]
+    fn synchronization_point_respects_min_block_boundaries() {
+        let in_buff = patterned_bytes(RSYNC_MIN_BLOCK_SIZE, 13);
+        for &in_buff_filled in &[
+            0,
+            RSYNC_LENGTH - 1,
+            RSYNC_LENGTH,
+            RSYNC_MIN_BLOCK_SIZE - 1,
+            RSYNC_MIN_BLOCK_SIZE,
+        ] {
+            let input_len = if in_buff_filled < RSYNC_MIN_BLOCK_SIZE {
+                RSYNC_MIN_BLOCK_SIZE - in_buff_filled + 1
+            } else {
+                1
+            };
+            let input = patterned_bytes(input_len, in_buff_filled as u8);
+            let expected = if in_buff_filled == RSYNC_MIN_BLOCK_SIZE {
+                (0, 1)
+            } else {
+                (input_len, 1)
+            };
+
+            assert_eq!(
+                call_synchronization_point(
+                    &input,
+                    0,
+                    &in_buff,
+                    in_buff_filled,
+                    in_buff_filled + input_len,
+                    1,
+                    0,
+                ),
+                expected,
+                "inBuff.filled={in_buff_filled}"
+            );
+        }
+    }
+
+    #[test]
+    fn synchronization_point_reports_initial_first_middle_and_final_hits() {
+        let in_buff = patterned_bytes(RSYNC_MIN_BLOCK_SIZE, 17);
+        let input = patterned_bytes(96, 19);
+        let (initial, hashes) =
+            input_hashes_after_each_byte(&in_buff, RSYNC_MIN_BLOCK_SIZE, &input, 0);
+
+        assert_eq!(
+            call_synchronization_point(
+                &input,
+                0,
+                &in_buff,
+                RSYNC_MIN_BLOCK_SIZE,
+                RSYNC_MIN_BLOCK_SIZE + input.len(),
+                1,
+                0,
+            ),
+            (0, 1)
+        );
+
+        for &(name, index) in &[("first", 0), ("middle", 37), ("final", 95)] {
+            let hit_mask = hit_mask_for_index(initial, &hashes, index);
+            assert_eq!(
+                call_synchronization_point(
+                    &input,
+                    0,
+                    &in_buff,
+                    RSYNC_MIN_BLOCK_SIZE,
+                    RSYNC_MIN_BLOCK_SIZE + input.len(),
+                    1,
+                    hit_mask,
+                ),
+                (index + 1, 1),
+                "{name} hit"
+            );
+        }
+    }
+
+    #[test]
+    fn synchronization_point_loads_all_input_when_no_hash_hits() {
+        let in_buff = patterned_bytes(RSYNC_MIN_BLOCK_SIZE, 23);
+        let input = patterned_bytes(96, 29);
+        let (initial, hashes) =
+            input_hashes_after_each_byte(&in_buff, RSYNC_MIN_BLOCK_SIZE, &input, 0);
+        assert_ne!(initial, u64::MAX);
+        assert!(hashes.iter().all(|&hash| hash != u64::MAX));
+
+        assert_eq!(
+            call_synchronization_point(
+                &input,
+                0,
+                &in_buff,
+                RSYNC_MIN_BLOCK_SIZE,
+                RSYNC_MIN_BLOCK_SIZE + input.len(),
+                1,
+                u64::MAX,
+            ),
+            (input.len(), 0)
+        );
+    }
+
+    #[test]
+    fn synchronization_point_preserves_nonzero_input_position() {
+        let in_buff = patterned_bytes(RSYNC_MIN_BLOCK_SIZE, 31);
+        let prefix = patterned_bytes(7, 37);
+        let payload = patterned_bytes(64, 41);
+        let mut input = prefix;
+        input.extend_from_slice(&payload);
+        let (initial, hashes) =
+            input_hashes_after_each_byte(&in_buff, RSYNC_MIN_BLOCK_SIZE, &input, 7);
+        let hit_mask = hit_mask_for_index(initial, &hashes, 23);
+
+        assert_eq!(
+            call_synchronization_point(
+                &input,
+                7,
+                &in_buff,
+                RSYNC_MIN_BLOCK_SIZE,
+                RSYNC_MIN_BLOCK_SIZE + payload.len(),
+                1,
+                hit_mask,
+            ),
+            (24, 1)
+        );
+    }
+
+    #[test]
+    fn synchronization_point_handles_a_window_spanning_the_buffer_boundary() {
+        let in_buff = patterned_bytes(RSYNC_MIN_BLOCK_SIZE, 43);
+        let input = [0xe1, 0x7a];
+        let in_buff_filled = RSYNC_MIN_BLOCK_SIZE - 1;
+        let mut history = in_buff[in_buff_filled - RSYNC_LENGTH..in_buff_filled].to_vec();
+        history.extend_from_slice(&input);
+        let hit_mask = rolling_hash_bytes(&history[2..2 + RSYNC_LENGTH]);
+
+        assert_eq!(
+            call_synchronization_point(
+                &input,
+                0,
+                &in_buff,
+                in_buff_filled,
+                in_buff_filled + input.len(),
+                1,
+                hit_mask,
+            ),
+            (input.len(), 1)
+        );
+    }
+
     #[test]
     fn raw_seq_buffer_conversion_uses_whole_element_capacity() {
         let mut sequences = [ZstdMtRawSeq::default(); 3];