From 160fa291486d59b18433971f6344b14a0178faf5 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sun, 19 Jul 2026 20:47:58 +0200 Subject: [PATCH] feat(mt): move serial wait loop into Rust Move the serial turn wait loop out of zstdmt_compress.c and into a Rust entry point. Rust now owns the lock, condition-wait, and ready/skip ordering policy while C retains pthread synchronization behind lock and wait callbacks. The projection documents that the mutex remains locked on return for the serial advance callback, preserving the existing worker synchronization contract. Add ABI layout checks and focused tests for both waiting and skip paths. Test Plan: - cargo fmt --manifest-path rust/Cargo.toml -- --check - ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml zstdmt_compress::tests::serial_wait_for_turn -- --nocapture - ulimit -v 41943040 && make -j1 --- lib/compress/zstdmt_compress.c | 44 +++++++++++-- rust/src/zstdmt_compress.rs | 117 +++++++++++++++++++++++++++++++++ 2 files changed, 155 insertions(+), 6 deletions(-) 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); }