diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index b0cb2a67f..960cbd5f8 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -957,6 +957,19 @@ typedef struct { size_t bufferCapacity; size_t roundBufferPos; } ZSTDMT_RustTryGetInputRangeResult; +typedef char ZSTDMT_rust_try_get_input_range_result_layout[ + (offsetof(ZSTDMT_RustTryGetInputRangeResult, ready) == 0 + && offsetof(ZSTDMT_RustTryGetInputRangeResult, movePrefix) + == sizeof(unsigned) + && offsetof(ZSTDMT_RustTryGetInputRangeResult, bufferStart) + == 2 * sizeof(unsigned) + && offsetof(ZSTDMT_RustTryGetInputRangeResult, bufferCapacity) + == 2 * sizeof(unsigned) + sizeof(void*) + && offsetof(ZSTDMT_RustTryGetInputRangeResult, roundBufferPos) + == 2 * sizeof(unsigned) + sizeof(void*) + sizeof(size_t) + && sizeof(ZSTDMT_RustTryGetInputRangeResult) + == 2 * sizeof(unsigned) + sizeof(void*) + 2 * sizeof(size_t)) + ? 1 : -1]; typedef void (*ZSTDMT_jobProjectionFn)(void* opaque, unsigned jobID, ZSTDMT_RustJobProjection* projection); ZSTDMT_RustInputRange ZSTDMT_rust_getInputDataInUse( @@ -965,6 +978,19 @@ ZSTDMT_RustInputRange ZSTDMT_rust_getInputDataInUse( void* opaque, ZSTDMT_jobProjectionFn projectJob); ZSTDMT_RustTryGetInputRangeResult ZSTDMT_rust_tryGetInputRange( const ZSTDMT_RustTryGetInputRangeProjection* projection); +typedef void (*ZSTDMT_inputRangeWaitFn)( + void* opaque, void* bufferStart, size_t bufferCapacity); +typedef void (*ZSTDMT_inputRangeMovePrefixFn)( + void* opaque, void* destination, const void* source, size_t size, + size_t roundBufferPos); +typedef void (*ZSTDMT_inputRangePublishBufferFn)( + void* opaque, void* bufferStart, size_t bufferCapacity); +int ZSTDMT_rust_commitInputRange( + const ZSTDMT_RustTryGetInputRangeResult* result, + void* roundBufferStart, const void* prefixStart, size_t prefixSize, + void* opaque, ZSTDMT_inputRangeWaitFn waitForLdm, + ZSTDMT_inputRangeMovePrefixFn movePrefix, + ZSTDMT_inputRangePublishBufferFn publishBuffer); size_t ZSTDMT_rust_toFlushNow(unsigned doneJobID, unsigned nextJobID, unsigned jobIDMask, void* opaque, ZSTDMT_jobProjectionFn projectJob); @@ -3296,6 +3322,43 @@ static void ZSTDMT_waitForLdmComplete(ZSTDMT_CCtx* mtctx, Buffer buffer) ZSTDMT_rust_waitForLdmComplete(&state); } +static void ZSTDMT_inputRangeWaitForLdm( + void* opaque, void* bufferStart, size_t bufferCapacity) +{ + ZSTDMT_waitForLdmComplete((ZSTDMT_CCtx*)opaque, + (Buffer){ bufferStart, bufferCapacity }); +} + +static void ZSTDMT_inputRangeMovePrefix( + void* opaque, void* destination, const void* source, size_t size, + size_t roundBufferPos) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + /* ZSTD_invalidateRepCodes() doesn't work for extDict variants. + * Simply copy the prefix to the beginning in that case. */ + ZSTD_memmove(destination, source, size); + mtctx->inBuff.prefix.start = (const BYTE*)destination; + mtctx->roundBuff.pos = roundBufferPos; +} + +static void ZSTDMT_inputRangePublishBuffer( + void* opaque, void* bufferStart, size_t bufferCapacity) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + Buffer const buffer = { bufferStart, bufferCapacity }; + + DEBUGLOG(5, "Using prefix range [%zx, %zx)", + (size_t)mtctx->inBuff.prefix.start, + (size_t)mtctx->inBuff.prefix.start + mtctx->inBuff.prefix.size); + DEBUGLOG(5, "Using source range [%zx, %zx)", + (size_t)buffer.start, + (size_t)buffer.start + buffer.capacity); + + mtctx->inBuff.buffer = buffer; + mtctx->inBuff.filled = 0; + assert(mtctx->roundBuff.pos + buffer.capacity <= mtctx->roundBuff.capacity); +} + /** * Attempts to set the inBuff to the next section to fill. * If any part of the new section is still in use we give up. @@ -3316,47 +3379,22 @@ static int ZSTDMT_tryGetInputRange(ZSTDMT_CCtx* mtctx) }; ZSTDMT_RustTryGetInputRangeResult const result = ZSTDMT_rust_tryGetInputRange(&projection); - Buffer buffer; DEBUGLOG(5, "ZSTDMT_tryGetInputRange"); assert(mtctx->inBuff.buffer.start == NULL); - if (!result.ready) { + if (!ZSTDMT_rust_commitInputRange( + &result, + mtctx->roundBuff.buffer, + mtctx->inBuff.prefix.start, + mtctx->inBuff.prefix.size, + mtctx, + ZSTDMT_inputRangeWaitForLdm, + ZSTDMT_inputRangeMovePrefix, + ZSTDMT_inputRangePublishBuffer)) { DEBUGLOG(5, "Waiting for buffer..."); return 0; } - - if (result.movePrefix) { - /* ZSTD_invalidateRepCodes() doesn't work for extDict variants. - * Simply copy the prefix to the beginning in that case. - */ - BYTE* const start = (BYTE*)mtctx->roundBuff.buffer; - size_t const prefixSize = mtctx->inBuff.prefix.size; - - buffer.start = start; - buffer.capacity = prefixSize; - ZSTDMT_waitForLdmComplete(mtctx, buffer); - ZSTD_memmove(start, mtctx->inBuff.prefix.start, prefixSize); - mtctx->inBuff.prefix.start = start; - mtctx->roundBuff.pos = result.roundBufferPos; - } - - buffer.start = (BYTE*)result.bufferStart; - buffer.capacity = result.bufferCapacity; - - ZSTDMT_waitForLdmComplete(mtctx, buffer); - - DEBUGLOG(5, "Using prefix range [%zx, %zx)", - (size_t)mtctx->inBuff.prefix.start, - (size_t)mtctx->inBuff.prefix.start + mtctx->inBuff.prefix.size); - DEBUGLOG(5, "Using source range [%zx, %zx)", - (size_t)buffer.start, - (size_t)buffer.start + buffer.capacity); - - - mtctx->inBuff.buffer = buffer; - mtctx->inBuff.filled = 0; - assert(mtctx->roundBuff.pos + buffer.capacity <= mtctx->roundBuff.capacity); return 1; } diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index dfb354b9e..ddd5e365f 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -1689,9 +1689,9 @@ pub struct ZSTDMT_tryGetInputRangeProjection { pub inUseSize: usize, } -/// Result of the MT input-range selection. C applies the returned decision, -/// including synchronization waits, prefix copying, and private field -/// updates. +/// Result of the MT input-range selection. Rust also owns the ordering of the +/// synchronization waits and publication callbacks that commit this result; +/// C retains the private buffer mutations and synchronization state. #[repr(C)] #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct ZSTDMT_tryGetInputRangeResult { @@ -1702,6 +1702,126 @@ pub struct ZSTDMT_tryGetInputRangeResult { pub roundBufferPos: usize, } +const _: () = { + assert!(offset_of!(ZSTDMT_tryGetInputRangeResult, ready) == 0); + assert!(offset_of!(ZSTDMT_tryGetInputRangeResult, movePrefix) == size_of::()); + assert!( + offset_of!(ZSTDMT_tryGetInputRangeResult, bufferStart) + == 2 * size_of::() + ); + assert!( + offset_of!(ZSTDMT_tryGetInputRangeResult, bufferCapacity) + == 2 * size_of::() + size_of::<*mut c_void>() + ); + assert!( + offset_of!(ZSTDMT_tryGetInputRangeResult, roundBufferPos) + == 2 * size_of::() + size_of::<*mut c_void>() + size_of::() + ); + assert!( + size_of::() + == 2 * size_of::() + size_of::<*mut c_void>() + 2 * size_of::() + ); +}; + +pub type ZSTDMT_inputRangeWaitFn = unsafe extern "C" fn( + opaque: *mut c_void, + buffer_start: *mut c_void, + buffer_capacity: usize, +); +pub type ZSTDMT_inputRangeMovePrefixFn = unsafe extern "C" fn( + opaque: *mut c_void, + destination: *mut c_void, + source: *const c_void, + size: usize, + round_buffer_pos: usize, +); +pub type ZSTDMT_inputRangePublishBufferFn = unsafe extern "C" fn( + opaque: *mut c_void, + buffer_start: *mut c_void, + buffer_capacity: usize, +); + +#[inline] +fn commit_input_range_with( + result: ZSTDMT_tryGetInputRangeResult, + round_buffer_start: *mut c_void, + prefix_start: *const c_void, + prefix_size: usize, + opaque: *mut c_void, + mut wait_for_ldm: W, + mut move_prefix: M, + mut publish_buffer: P, +) -> c_int +where + W: FnMut(*mut c_void, *mut c_void, usize), + M: FnMut(*mut c_void, *mut c_void, *const c_void, usize, usize), + P: FnMut(*mut c_void, *mut c_void, usize), +{ + if result.ready == 0 { + return 0; + } + + if result.movePrefix != 0 { + wait_for_ldm(opaque, round_buffer_start, prefix_size); + move_prefix( + opaque, + round_buffer_start, + prefix_start, + prefix_size, + result.roundBufferPos, + ); + } + + wait_for_ldm(opaque, result.bufferStart, result.bufferCapacity); + publish_buffer(opaque, result.bufferStart, result.bufferCapacity); + 1 +} + +/// Commit the selected MT input range in the original order. Rust owns the +/// prefix-wrap branch and the wait/copy/wait/publish sequencing; C callbacks +/// retain the private round-buffer, input-buffer, and LDM synchronization +/// operations. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_commitInputRange( + result: *const ZSTDMT_tryGetInputRangeResult, + roundBufferStart: *mut c_void, + prefixStart: *const c_void, + prefixSize: usize, + opaque: *mut c_void, + waitForLdm: Option, + movePrefix: Option, + publishBuffer: Option, +) -> c_int { + let Some(result) = (unsafe { result.as_ref() }).copied() else { + return 0; + }; + let (Some(wait_for_ldm), Some(move_prefix), Some(publish_buffer)) = + (waitForLdm, movePrefix, publishBuffer) + else { + return 0; + }; + if opaque.is_null() { + return 0; + } + + commit_input_range_with( + result, + roundBufferStart, + prefixStart, + prefixSize, + opaque, + |opaque, buffer_start, buffer_capacity| unsafe { + wait_for_ldm(opaque, buffer_start, buffer_capacity); + }, + |opaque, destination, source, size, round_buffer_pos| unsafe { + move_prefix(opaque, destination, source, size, round_buffer_pos); + }, + |opaque, buffer_start, buffer_capacity| unsafe { + publish_buffer(opaque, buffer_start, buffer_capacity); + }, + ) +} + pub type ZSTDMT_jobProjectionFn = unsafe extern "C" fn( opaque: *mut c_void, job_id: c_uint, @@ -8944,6 +9064,115 @@ mod tests { assert_eq!(try_get_input_range_with(blocked_source).ready, 0); } + #[test] + fn commit_input_range_wraps_prefix_before_waiting_for_source() { + let mut round_buffer = [0u8; 32]; + let prefix = [1u8; 4]; + let round_buffer_start = round_buffer.as_mut_ptr(); + let selected_buffer = round_buffer_start.wrapping_add(prefix.len()); + let result = ZSTDMT_tryGetInputRangeResult { + ready: 1, + movePrefix: 1, + bufferStart: selected_buffer.cast(), + bufferCapacity: 8, + roundBufferPos: prefix.len(), + }; + let events = Rc::new(RefCell::new(Vec::new())); + let wait_events = Rc::clone(&events); + let move_events = Rc::clone(&events); + let publish_events = Rc::clone(&events); + let wait_calls = Rc::new(RefCell::new(Vec::new())); + let wait_call_records = Rc::clone(&wait_calls); + + let committed = commit_input_range_with( + result, + round_buffer_start.cast(), + prefix.as_ptr().cast(), + prefix.len(), + ptr::null_mut(), + move |_, buffer_start, buffer_capacity| { + wait_call_records + .borrow_mut() + .push((buffer_start, buffer_capacity)); + wait_events.borrow_mut().push("wait"); + }, + move |_, destination, source, size, round_buffer_pos| { + assert_eq!(destination, round_buffer_start.cast()); + assert_eq!(source, prefix.as_ptr().cast()); + assert_eq!(size, prefix.len()); + assert_eq!(round_buffer_pos, prefix.len()); + unsafe { + ptr::copy_nonoverlapping( + source.cast::(), + destination.cast::(), + size, + ); + } + move_events.borrow_mut().push("move"); + }, + move |_, buffer_start, buffer_capacity| { + assert_eq!(buffer_start, selected_buffer.cast()); + assert_eq!(buffer_capacity, 8); + publish_events.borrow_mut().push("publish"); + }, + ); + + assert_eq!(committed, 1); + assert_eq!(&*events.borrow(), &["wait", "move", "wait", "publish"]); + assert_eq!( + &*wait_calls.borrow(), + &[ + (round_buffer_start.cast(), prefix.len()), + (selected_buffer.cast(), 8), + ] + ); + assert_eq!(&round_buffer[..prefix.len()], &prefix); + } + + #[test] + fn commit_input_range_without_wrap_waits_once_before_publish() { + let mut round_buffer = [0u8; 32]; + let round_buffer_start = round_buffer.as_mut_ptr(); + let selected_buffer = round_buffer_start.wrapping_add(8); + let result = ZSTDMT_tryGetInputRangeResult { + ready: 1, + bufferStart: selected_buffer.cast(), + bufferCapacity: 8, + roundBufferPos: 8, + ..ZSTDMT_tryGetInputRangeResult::default() + }; + let events = Rc::new(RefCell::new(Vec::new())); + let wait_events = Rc::clone(&events); + let move_called = Rc::new(RefCell::new(false)); + let move_called_by_callback = Rc::clone(&move_called); + let publish_events = Rc::clone(&events); + + let committed = commit_input_range_with( + result, + round_buffer_start.cast(), + ptr::null(), + 0, + ptr::null_mut(), + move |_, buffer_start, buffer_capacity| { + assert_eq!(buffer_start, selected_buffer.cast()); + assert_eq!(buffer_capacity, 8); + wait_events.borrow_mut().push("wait"); + }, + move |_, _, _, _, _| { + *move_called_by_callback.borrow_mut() = true; + }, + move |_, buffer_start, buffer_capacity| { + assert_eq!(buffer_start, selected_buffer.cast()); + assert_eq!(buffer_capacity, 8); + publish_events.borrow_mut().push("publish"); + }, + ); + + assert_eq!(committed, 1); + assert_eq!(&*events.borrow(), &["wait", "publish"]); + assert!(!*move_called.borrow()); + } + #[test] fn to_flush_now_projects_oldest_slot_and_normalizes_errors() { let projection = ZSTDMT_jobProjection {