diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index cc18a0fc1..2324e6999 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -361,6 +361,24 @@ typedef void (*ZSTDMT_serialGenerateLdmFn)( typedef void (*ZSTDMT_serialUpdateChecksumFn)( void* opaque, const void* src, size_t srcSize); typedef void (*ZSTDMT_serialAdvanceFn)(void* opaque); +typedef void (*ZSTDMT_serialStateLockFn)(void* opaque); +typedef void (*ZSTDMT_serialStateWaitFn)(void* opaque); +typedef struct { + void* callbackContext; + unsigned* nextJobID; + ZSTDMT_serialStateLockFn lock; + ZSTDMT_serialStateWaitFn wait; + unsigned jobID; +} ZSTDMT_RustSerialWaitForTurnState; +typedef char ZSTDMT_rust_serial_wait_for_turn_state_layout[ + (offsetof(ZSTDMT_RustSerialWaitForTurnState, callbackContext) == 0 + && offsetof(ZSTDMT_RustSerialWaitForTurnState, nextJobID) == sizeof(void*) + && offsetof(ZSTDMT_RustSerialWaitForTurnState, lock) == 2 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialWaitForTurnState, wait) == 3 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialWaitForTurnState, jobID) == 4 * sizeof(void*) + && sizeof(ZSTDMT_RustSerialWaitForTurnState) == 5 * sizeof(void*)) ? 1 : -1]; +int ZSTDMT_rust_serialStateWaitForTurn( + const ZSTDMT_RustSerialWaitForTurnState* state); void ZSTDMT_rust_serialStateGenSequences( ZSTDMT_RustRawSeqStore* seqStore, const void* src, size_t srcSize, unsigned jobID, int ldmEnabled, int checksumEnabled, void* opaque, @@ -1089,18 +1107,32 @@ static void ZSTDMT_serialState_free(SerialState* serialState) ZSTDMT_serialState_freeTables); } +static void ZSTDMT_serialState_lock(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex); +} + +static void ZSTDMT_serialState_wait(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + DEBUGLOG(5, "wait for serialState->cond"); + ZSTD_pthread_cond_wait(&serialState->cond, &serialState->mutex); +} + /* Rust owns the serial turn/skip decision and operation ordering. The wait * callback intentionally leaves the main serial mutex locked; the advance * callback releases it after Rust has performed the current turn's work. */ static int ZSTDMT_serialState_waitForTurn(void* opaque, unsigned jobID) { SerialState* const serialState = (SerialState*)opaque; - ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex); - while (serialState->nextJobID < jobID) { - DEBUGLOG(5, "wait for serialState->cond"); - ZSTD_pthread_cond_wait(&serialState->cond, &serialState->mutex); - } - return serialState->nextJobID == jobID; + ZSTDMT_RustSerialWaitForTurnState state; + state.callbackContext = serialState; + state.nextJobID = &serialState->nextJobID; + state.lock = ZSTDMT_serialState_lock; + state.wait = ZSTDMT_serialState_wait; + state.jobID = jobID; + return ZSTDMT_rust_serialStateWaitForTurn(&state); } static void ZSTDMT_serialState_generateLdm( diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index fa586cb77..7fc871940 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -89,6 +89,56 @@ pub type ZSTDMT_serialGenerateLdmFn = unsafe extern "C" fn(*mut c_void, *mut ZstdMtRawSeqStore, *const c_void, usize); 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); +pub type ZSTDMT_serialStateLockFn = unsafe extern "C" fn(*mut c_void); +pub type ZSTDMT_serialStateWaitFn = unsafe extern "C" fn(*mut c_void); + +/// Projection for the MT serial turn wait. Rust owns the lock/wait loop and +/// comparison; C retains the pthread mutex and condition variable behind the +/// callbacks. The lock intentionally remains held when this returns so the +/// caller can perform the serial turn and release it in the advance callback. +#[repr(C)] +pub struct ZSTDMT_RustSerialWaitForTurnState { + callback_context: *mut c_void, + next_job_id: *mut c_uint, + lock: Option, + wait: Option, + job_id: c_uint, +} + +const _: () = { + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialWaitForTurnState, callback_context) == 0); + assert!(offset_of!(ZSTDMT_RustSerialWaitForTurnState, next_job_id) == size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialWaitForTurnState, lock) == 2 * size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialWaitForTurnState, wait) == 3 * size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialWaitForTurnState, job_id) == 4 * size_of::()); + assert!(size_of::() == size_of::<[usize; 5]>()); +}; + +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_serialStateWaitForTurn( + state: *const ZSTDMT_RustSerialWaitForTurnState, +) -> c_int { + if state.is_null() { + return 0; + } + let state = unsafe { &*state }; + if state.callback_context.is_null() || state.next_job_id.is_null() { + return 0; + } + let (Some(lock), Some(wait)) = (state.lock, state.wait) else { + return 0; + }; + + unsafe { + lock(state.callback_context); + while *state.next_job_id < state.job_id { + wait(state.callback_context); + } + c_int::from(*state.next_job_id == state.job_id) + } +} #[repr(C)] #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] @@ -3452,6 +3502,41 @@ mod tests { next_overlap: usize, } + struct SerialWaitForTurnTestContext { + events: Vec<&'static str>, + next_job_id: c_uint, + wake_job_id: c_uint, + } + + unsafe extern "C" fn serial_wait_for_turn_test_lock(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("lock") + }; + } + + unsafe extern "C" fn serial_wait_for_turn_test_wait(context: *mut c_void) { + unsafe { + let context = &mut *context.cast::(); + context.events.push("wait"); + context.next_job_id = context.wake_job_id; + } + } + + fn serial_wait_for_turn_test_state( + context: &mut SerialWaitForTurnTestContext, + job_id: c_uint, + ) -> ZSTDMT_RustSerialWaitForTurnState { + ZSTDMT_RustSerialWaitForTurnState { + callback_context: context as *mut _ as *mut c_void, + next_job_id: &mut context.next_job_id, + lock: Some(serial_wait_for_turn_test_lock), + wait: Some(serial_wait_for_turn_test_wait), + job_id, + } + } + unsafe extern "C" fn wait_for_ldm_test_lock(context: *mut c_void) { unsafe { (*context.cast::()) @@ -3541,6 +3626,38 @@ mod tests { assert_eq!(context.next_overlap, 0); } + #[test] + fn serial_wait_for_turn_waits_until_job_is_ready() { + let mut context = SerialWaitForTurnTestContext { + events: Vec::new(), + next_job_id: 2, + wake_job_id: 4, + }; + let state = serial_wait_for_turn_test_state(&mut context, 4); + + let result = unsafe { ZSTDMT_rust_serialStateWaitForTurn(&state) }; + + assert_eq!(result, 1); + assert_eq!(context.events, vec!["lock", "wait"]); + assert_eq!(context.next_job_id, 4); + } + + #[test] + fn serial_wait_for_turn_reports_skip_without_waiting_for_later_job() { + let mut context = SerialWaitForTurnTestContext { + events: Vec::new(), + next_job_id: 5, + wake_job_id: 5, + }; + let state = serial_wait_for_turn_test_state(&mut context, 4); + + let result = unsafe { ZSTDMT_rust_serialStateWaitForTurn(&state) }; + + assert_eq!(result, 0); + assert_eq!(context.events, vec!["lock"]); + assert_eq!(context.next_job_id, 5); + } + fn record_compression_job_event(state: &Rc>, event: &'static str) { state.borrow_mut().events.push(event); }