feat(mt): move input-range reuse policy into Rust

Move multithreaded input-range selection and overlap decisions into Rust using
an explicit scalar projection. C retains the round-buffer mutation, LDM
synchronization, prefix copying, and private job state, while Rust decides
whether a reusable section is available and where it begins.

Test Plan:
- cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1
- cargo clippy --manifest-path rust/Cargo.toml --tests -- -D warnings
- make -B -C lib -j2 lib
- make -B -C tests -j2 test-cli-tests
- make -B -C tests -j2 test-rust-lib-smoke
This commit is contained in:
2026-07-18 23:46:02 +02:00
parent 693b142837
commit dc95de5212
2 changed files with 220 additions and 27 deletions
+180
View File
@@ -170,6 +170,35 @@ pub struct ZSTDMT_inputRange {
pub size: usize,
}
/// Scalar input-buffer state used by the MT range-selection policy. C keeps
/// the input and round-buffer structs private; Rust decides only whether the
/// next target section can be reused and where that section starts.
#[repr(C)]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ZSTDMT_tryGetInputRangeProjection {
pub roundBufferStart: *const c_void,
pub roundBufferCapacity: usize,
pub roundBufferPos: usize,
pub prefixStart: *const c_void,
pub prefixSize: usize,
pub targetSectionSize: usize,
pub inUseStart: *const c_void,
pub inUseSize: usize,
}
/// Result of the MT input-range selection. C applies the returned decision,
/// including synchronization waits, prefix copying, and private field
/// updates.
#[repr(C)]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ZSTDMT_tryGetInputRangeResult {
pub ready: c_uint,
pub movePrefix: c_uint,
pub bufferStart: *mut c_void,
pub bufferCapacity: usize,
pub roundBufferPos: usize,
}
pub type ZSTDMT_jobProjectionFn = unsafe extern "C" fn(
opaque: *mut c_void,
job_id: c_uint,
@@ -962,6 +991,93 @@ pub unsafe extern "C" fn ZSTDMT_rust_toFlushNow(
})
}
#[inline]
fn try_get_input_range_with(
projection: ZSTDMT_tryGetInputRangeProjection,
) -> ZSTDMT_tryGetInputRangeResult {
let invalid = || ZSTDMT_tryGetInputRangeResult::default();
debug_assert!(projection.roundBufferPos <= projection.roundBufferCapacity);
debug_assert!(projection.targetSectionSize > 0);
if projection.roundBufferPos > projection.roundBufferCapacity
|| projection.targetSectionSize == 0
|| projection.targetSectionSize > projection.roundBufferCapacity
{
return invalid();
}
let space_left = projection.roundBufferCapacity - projection.roundBufferPos;
let (round_buffer_pos, move_prefix) = if space_left < projection.targetSectionSize {
/* The C implementation first checks the prefix range at the start of
* the round buffer before resetting the position there. */
if is_overlapped(
projection.roundBufferStart,
projection.prefixSize,
projection.inUseStart,
projection.inUseSize,
) != 0
{
return invalid();
}
if projection.prefixSize > projection.roundBufferCapacity - projection.targetSectionSize {
return invalid();
}
(projection.prefixSize, 1)
} else {
(projection.roundBufferPos, 0)
};
let buffer_start = projection
.roundBufferStart
.cast::<u8>()
.wrapping_add(round_buffer_pos)
.cast_mut()
.cast::<c_void>();
if move_prefix == 0 {
/* This is an internal invariant in the original C helper: a live
* prefix must never be overwritten by the next source section. */
debug_assert_eq!(
is_overlapped(
buffer_start,
projection.targetSectionSize,
projection.prefixStart,
projection.prefixSize,
),
0
);
}
if is_overlapped(
buffer_start,
projection.targetSectionSize,
projection.inUseStart,
projection.inUseSize,
) != 0
{
return invalid();
}
ZSTDMT_tryGetInputRangeResult {
ready: 1,
movePrefix: move_prefix,
bufferStart: buffer_start,
bufferCapacity: projection.targetSectionSize,
roundBufferPos: round_buffer_pos,
}
}
/// Select the next reusable round-buffer section for MT input. The C adapter
/// retains all synchronization, prefix copying, and private context updates.
#[cfg(not(test))]
#[no_mangle]
pub unsafe extern "C" fn ZSTDMT_rust_tryGetInputRange(
projection: *const ZSTDMT_tryGetInputRangeProjection,
) -> ZSTDMT_tryGetInputRangeResult {
let Some(projection) = (unsafe { projection.as_ref() }).copied() else {
return ZSTDMT_tryGetInputRangeResult::default();
};
try_get_input_range_with(projection)
}
#[inline]
fn cycle_log(chain_log: c_uint, strategy: c_int) -> c_uint {
chain_log.wrapping_sub((strategy >= ZSTD_BTLAZY2) as c_uint)
@@ -3087,6 +3203,70 @@ mod tests {
assert_eq!(result.size, source.len());
}
#[test]
fn try_get_input_range_selects_next_section_without_wrapping() {
let round_buffer = [0u8; 32];
let prefix = [1u8; 4];
let projection = ZSTDMT_tryGetInputRangeProjection {
roundBufferStart: round_buffer.as_ptr().cast(),
roundBufferCapacity: round_buffer.len(),
roundBufferPos: 8,
prefixStart: prefix.as_ptr().cast(),
prefixSize: prefix.len(),
targetSectionSize: 8,
..ZSTDMT_tryGetInputRangeProjection::default()
};
let result = try_get_input_range_with(projection);
assert_eq!(result.ready, 1);
assert_eq!(result.movePrefix, 0);
assert_eq!(result.roundBufferPos, 8);
assert_eq!(
result.bufferStart,
round_buffer.as_ptr().wrapping_add(8).cast_mut().cast()
);
assert_eq!(result.bufferCapacity, 8);
}
#[test]
fn try_get_input_range_wraps_after_prefix_and_rejects_live_ranges() {
let round_buffer = [0u8; 32];
let prefix = [1u8; 4];
let wrapped = ZSTDMT_tryGetInputRangeProjection {
roundBufferStart: round_buffer.as_ptr().cast(),
roundBufferCapacity: round_buffer.len(),
roundBufferPos: 28,
prefixStart: prefix.as_ptr().cast(),
prefixSize: prefix.len(),
targetSectionSize: 8,
..ZSTDMT_tryGetInputRangeProjection::default()
};
let result = try_get_input_range_with(wrapped);
assert_eq!(result.ready, 1);
assert_eq!(result.movePrefix, 1);
assert_eq!(result.roundBufferPos, prefix.len());
assert_eq!(
result.bufferStart,
round_buffer.as_ptr().wrapping_add(4).cast_mut().cast()
);
let blocked_prefix = ZSTDMT_tryGetInputRangeProjection {
inUseStart: round_buffer.as_ptr().cast(),
inUseSize: prefix.len(),
..wrapped
};
assert_eq!(try_get_input_range_with(blocked_prefix).ready, 0);
let blocked_source = ZSTDMT_tryGetInputRangeProjection {
roundBufferPos: 8,
inUseStart: round_buffer.as_ptr().wrapping_add(8).cast(),
inUseSize: 8,
..wrapped
};
assert_eq!(try_get_input_range_with(blocked_source).ready, 0);
}
#[test]
fn to_flush_now_projects_oldest_slot_and_normalizes_errors() {
let projection = ZSTDMT_jobProjection {