From b19ca569f7b3c2706ab0b84881592ff59920a8b2 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sat, 18 Jul 2026 06:18:30 +0200 Subject: [PATCH] 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 | 102 +------- rust/src/zstdmt_compress.rs | 424 +++++++++++++++++++++++++++++++++ 2 files changed, 436 insertions(+), 90 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 0b7674778..ddcd8c0ef 100644 --- a/lib/compress/zstdmt_compress.c +++ b/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; - BYTE const* prev; - size_t pos; - - syncPoint.toLoad = MIN(input.size - input.pos, mtctx->targetSectionSize - mtctx->inBuff.filled); - syncPoint.flush = 0; - if (!mtctx->params.rsyncable) - /* 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); + ZSTDMT_rust_findSynchronizationPoint( + input.src, input.size, input.pos, + mtctx->targetSectionSize, + mtctx->inBuff.buffer.start, mtctx->inBuff.filled, + mtctx->params.rsyncable, + mtctx->rsync.primePower, mtctx->rsync.hitMask, + &syncPoint.toLoad, &syncPoint.flush); return syncPoint; } diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index cbc9eea51..3b8c2abfc 100644 --- a/rust/src/zstdmt_compress.rs +++ b/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::().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::() + .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::() + .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 { + 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) { + 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];