diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 25ec11390..e2700d690 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -368,6 +368,33 @@ void ZSTDMT_rust_serialStateGenSequences( ZSTDMT_serialGenerateLdmFn generateLdm, ZSTDMT_serialUpdateChecksumFn updateChecksum, ZSTDMT_serialAdvanceFn advance); +typedef void (*ZSTDMT_waitForLdmLockFn)(void* opaque); +typedef int (*ZSTDMT_waitForLdmOverlapFn)( + void* opaque, void* bufferStart, size_t bufferCapacity); +typedef void (*ZSTDMT_waitForLdmWaitFn)(void* opaque); +typedef void (*ZSTDMT_waitForLdmUnlockFn)(void* opaque); +typedef struct { + void* callbackContext; + void* bufferStart; + size_t bufferCapacity; + int ldmEnabled; + ZSTDMT_waitForLdmLockFn lock; + ZSTDMT_waitForLdmOverlapFn overlaps; + ZSTDMT_waitForLdmWaitFn wait; + ZSTDMT_waitForLdmUnlockFn unlock; +} ZSTDMT_RustWaitForLdmState; +typedef char ZSTDMT_rust_wait_for_ldm_state_layout[ + (offsetof(ZSTDMT_RustWaitForLdmState, callbackContext) == 0 + && offsetof(ZSTDMT_RustWaitForLdmState, bufferStart) == sizeof(void*) + && offsetof(ZSTDMT_RustWaitForLdmState, bufferCapacity) == 2 * sizeof(void*) + && offsetof(ZSTDMT_RustWaitForLdmState, ldmEnabled) == 3 * sizeof(void*) + && offsetof(ZSTDMT_RustWaitForLdmState, lock) == 4 * sizeof(void*) + && offsetof(ZSTDMT_RustWaitForLdmState, overlaps) == 5 * sizeof(void*) + && offsetof(ZSTDMT_RustWaitForLdmState, wait) == 6 * sizeof(void*) + && offsetof(ZSTDMT_RustWaitForLdmState, unlock) == 7 * sizeof(void*) + && sizeof(ZSTDMT_RustWaitForLdmState) == 8 * sizeof(void*)) ? 1 : -1]; +void ZSTDMT_rust_waitForLdmComplete( + const ZSTDMT_RustWaitForLdmState* state); unsigned ZSTDMT_rust_computeTargetJobLog(unsigned windowLog, unsigned chainLog, int strategy, int enableLdm); @@ -2102,22 +2129,54 @@ static int ZSTDMT_doesOverlapWindow(Buffer buffer, ZSTD_window_t window) window.lowLimit); } +static void ZSTDMT_waitForLdmLock(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + ZSTD_PTHREAD_MUTEX_LOCK(&mtctx->serial.ldmWindowMutex); +} + +static int ZSTDMT_waitForLdmOverlap( + void* opaque, void* bufferStart, size_t bufferCapacity) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + Buffer const buffer = { bufferStart, bufferCapacity }; + return ZSTDMT_doesOverlapWindow(buffer, mtctx->serial.ldmWindow); +} + +static void ZSTDMT_waitForLdmWait(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + DEBUGLOG(5, "Waiting for LDM to finish..."); + ZSTD_pthread_cond_wait(&mtctx->serial.ldmWindowCond, + &mtctx->serial.ldmWindowMutex); +} + +static void ZSTDMT_waitForLdmUnlock(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + DEBUGLOG(6, "Done waiting for LDM to finish"); + ZSTD_pthread_mutex_unlock(&mtctx->serial.ldmWindowMutex); +} + static void ZSTDMT_waitForLdmComplete(ZSTDMT_CCtx* mtctx, Buffer buffer) { - if (mtctx->params.ldmParams.enableLdm == ZSTD_ps_enable) { - ZSTD_pthread_mutex_t* mutex = &mtctx->serial.ldmWindowMutex; + int const ldmEnabled = mtctx->params.ldmParams.enableLdm == ZSTD_ps_enable; + ZSTDMT_RustWaitForLdmState state; + if (ldmEnabled) { DEBUGLOG(5, "ZSTDMT_waitForLdmComplete"); DEBUGLOG(5, "source [0x%zx, 0x%zx)", (size_t)buffer.start, (size_t)buffer.start + buffer.capacity); - ZSTD_PTHREAD_MUTEX_LOCK(mutex); - while (ZSTDMT_doesOverlapWindow(buffer, mtctx->serial.ldmWindow)) { - DEBUGLOG(5, "Waiting for LDM to finish..."); - ZSTD_pthread_cond_wait(&mtctx->serial.ldmWindowCond, mutex); - } - DEBUGLOG(6, "Done waiting for LDM to finish"); - ZSTD_pthread_mutex_unlock(mutex); } + state.callbackContext = mtctx; + state.bufferStart = buffer.start; + state.bufferCapacity = buffer.capacity; + state.ldmEnabled = ldmEnabled; + state.lock = ZSTDMT_waitForLdmLock; + state.overlaps = ZSTDMT_waitForLdmOverlap; + state.wait = ZSTDMT_waitForLdmWait; + state.unlock = ZSTDMT_waitForLdmUnlock; + ZSTDMT_rust_waitForLdmComplete(&state); } /** diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index f59f75fdf..dcdc8eee9 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -14,7 +14,7 @@ //! are Rust-owned. C retains ownership of private descriptor fields and //! platform synchronization. -use std::mem::{self, MaybeUninit}; +use std::mem::{self, offset_of, size_of, MaybeUninit}; use std::os::raw::{c_int, c_uint, c_void}; use std::ptr; use std::sync::Mutex; @@ -90,6 +90,74 @@ pub type ZSTDMT_serialGenerateLdmFn = pub type ZSTDMT_serialUpdateChecksumFn = unsafe extern "C" fn(*mut c_void, *const c_void, usize); pub type ZSTDMT_serialAdvanceFn = unsafe extern "C" fn(*mut c_void); +type ZSTDMT_waitForLdmLockFn = unsafe extern "C" fn(*mut c_void); +type ZSTDMT_waitForLdmOverlapFn = unsafe extern "C" fn(*mut c_void, *mut c_void, usize) -> c_int; +type ZSTDMT_waitForLdmWaitFn = unsafe extern "C" fn(*mut c_void); +type ZSTDMT_waitForLdmUnlockFn = unsafe extern "C" fn(*mut c_void); + +/// Projection for the MT wait that protects reusable input against LDM use. +/// +/// Rust owns the enabled branch and lock/overlap/wait ordering. C retains the +/// pthread mutex, condition variable, and private LDM window state. +#[repr(C)] +pub struct ZSTDMT_RustWaitForLdmState { + callback_context: *mut c_void, + buffer_start: *mut c_void, + buffer_capacity: usize, + ldm_enabled: c_int, + lock: Option, + overlaps: Option, + wait: Option, + unlock: Option, +} + +const _: () = { + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!(offset_of!(ZSTDMT_RustWaitForLdmState, callback_context) == 0); + assert!(offset_of!(ZSTDMT_RustWaitForLdmState, buffer_start) == size_of::()); + assert!(offset_of!(ZSTDMT_RustWaitForLdmState, buffer_capacity) == 2 * size_of::()); + assert!(offset_of!(ZSTDMT_RustWaitForLdmState, ldm_enabled) == 3 * size_of::()); + assert!(offset_of!(ZSTDMT_RustWaitForLdmState, lock) == 4 * size_of::()); + assert!(offset_of!(ZSTDMT_RustWaitForLdmState, overlaps) == 5 * size_of::()); + assert!(offset_of!(ZSTDMT_RustWaitForLdmState, wait) == 6 * size_of::()); + assert!(offset_of!(ZSTDMT_RustWaitForLdmState, unlock) == 7 * size_of::()); + assert!(size_of::() == size_of::<[usize; 8]>()); +}; + +/// Wait until the reusable input range no longer overlaps the serial LDM +/// window, preserving the C synchronization state behind callbacks. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_waitForLdmComplete(state: *const ZSTDMT_RustWaitForLdmState) { + if state.is_null() { + return; + } + let state = unsafe { &*state }; + if state.ldm_enabled == 0 || state.callback_context.is_null() { + return; + } + let (Some(lock), Some(overlaps), Some(wait), Some(unlock)) = + (state.lock, state.overlaps, state.wait, state.unlock) + else { + return; + }; + + unsafe { + lock(state.callback_context); + while overlaps( + state.callback_context, + state.buffer_start, + state.buffer_capacity, + ) != 0 + { + wait(state.callback_context); + } + unlock(state.callback_context); + } +} + /// Scalar inputs for the MT streaming initializer. The full parameter /// object, dictionary handles, pools, buffers, and synchronization remain /// private to C. Rust owns the order in which the C callbacks are invoked and @@ -3171,6 +3239,101 @@ mod tests { finished: Vec, } + struct WaitForLdmTestContext { + events: Vec<&'static str>, + overlaps: Vec, + next_overlap: usize, + } + + unsafe extern "C" fn wait_for_ldm_test_lock(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("lock") + }; + } + + unsafe extern "C" fn wait_for_ldm_test_overlaps( + context: *mut c_void, + _buffer_start: *mut c_void, + _buffer_capacity: usize, + ) -> c_int { + let context = unsafe { &mut *context.cast::() }; + context.events.push("overlap"); + let overlaps = context + .overlaps + .get(context.next_overlap) + .copied() + .unwrap_or(false); + context.next_overlap += 1; + c_int::from(overlaps) + } + + unsafe extern "C" fn wait_for_ldm_test_wait(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("wait") + }; + } + + unsafe extern "C" fn wait_for_ldm_test_unlock(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("unlock") + }; + } + + fn wait_for_ldm_test_state( + context: &mut WaitForLdmTestContext, + ldm_enabled: c_int, + ) -> ZSTDMT_RustWaitForLdmState { + ZSTDMT_RustWaitForLdmState { + callback_context: context as *mut _ as *mut c_void, + buffer_start: ptr::null_mut(), + buffer_capacity: 8, + ldm_enabled, + lock: Some(wait_for_ldm_test_lock), + overlaps: Some(wait_for_ldm_test_overlaps), + wait: Some(wait_for_ldm_test_wait), + unlock: Some(wait_for_ldm_test_unlock), + } + } + + #[test] + fn wait_for_ldm_runs_overlap_wait_loop_before_unlocking() { + let mut context = WaitForLdmTestContext { + events: Vec::new(), + overlaps: vec![true, false], + next_overlap: 0, + }; + let state = wait_for_ldm_test_state(&mut context, ZSTD_PS_ENABLE); + + unsafe { ZSTDMT_rust_waitForLdmComplete(&state) }; + + assert_eq!( + context.events, + vec!["lock", "overlap", "wait", "overlap", "unlock"] + ); + assert_eq!(context.next_overlap, 2); + } + + #[test] + fn wait_for_ldm_skips_callbacks_when_ldm_is_disabled() { + let mut context = WaitForLdmTestContext { + events: Vec::new(), + overlaps: vec![true], + next_overlap: 0, + }; + let state = wait_for_ldm_test_state(&mut context, 0); + + unsafe { ZSTDMT_rust_waitForLdmComplete(&state) }; + + assert!(context.events.is_empty()); + assert_eq!(context.next_overlap, 0); + } + fn record_compression_job_event(state: &Rc>, event: &'static str) { state.borrow_mut().events.push(event); }