From 1d8bbb421d85c28a38074bb8a373caf7d2dcc016 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sun, 19 Jul 2026 20:52:10 +0200 Subject: [PATCH] feat(mt): move serial advance orchestration into Rust Move the serial turn increment, broadcast, and unlock ordering into a Rust entry point. C retains the pthread condition variable and mutex operations behind callbacks, while Rust preserves wrapping unsigned job-counter semantics and the required broadcast-before-unlock contract. Add ABI layout checks and focused tests for ordinary and wrapping advancement. 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_ -- --nocapture - ulimit -v 41943040 && make -j1 --- lib/compress/zstdmt_compress.c | 36 +++++++++++- rust/src/zstdmt_compress.rs | 101 +++++++++++++++++++++++++++++++++ 2 files changed, 134 insertions(+), 3 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 2324e6999..d33308707 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -379,6 +379,21 @@ typedef char ZSTDMT_rust_serial_wait_for_turn_state_layout[ && sizeof(ZSTDMT_RustSerialWaitForTurnState) == 5 * sizeof(void*)) ? 1 : -1]; int ZSTDMT_rust_serialStateWaitForTurn( const ZSTDMT_RustSerialWaitForTurnState* state); +typedef void (*ZSTDMT_serialStateBroadcastFn)(void* opaque); +typedef void (*ZSTDMT_serialStateUnlockFn)(void* opaque); +typedef struct { + void* callbackContext; + unsigned* nextJobID; + ZSTDMT_serialStateBroadcastFn broadcast; + ZSTDMT_serialStateUnlockFn unlock; +} ZSTDMT_RustSerialAdvanceState; +typedef char ZSTDMT_rust_serial_advance_state_layout[ + (offsetof(ZSTDMT_RustSerialAdvanceState, callbackContext) == 0 + && offsetof(ZSTDMT_RustSerialAdvanceState, nextJobID) == sizeof(void*) + && offsetof(ZSTDMT_RustSerialAdvanceState, broadcast) == 2 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialAdvanceState, unlock) == 3 * sizeof(void*) + && sizeof(ZSTDMT_RustSerialAdvanceState) == 4 * sizeof(void*)) ? 1 : -1]; +void ZSTDMT_rust_serialStateAdvance(const ZSTDMT_RustSerialAdvanceState* state); void ZSTDMT_rust_serialStateGenSequences( ZSTDMT_RustRawSeqStore* seqStore, const void* src, size_t srcSize, unsigned jobID, int ldmEnabled, int checksumEnabled, void* opaque, @@ -1120,6 +1135,18 @@ static void ZSTDMT_serialState_wait(void* opaque) ZSTD_pthread_cond_wait(&serialState->cond, &serialState->mutex); } +static void ZSTDMT_serialState_broadcast(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_pthread_cond_broadcast(&serialState->cond); +} + +static void ZSTDMT_serialState_unlock(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_pthread_mutex_unlock(&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. */ @@ -1172,9 +1199,12 @@ static void ZSTDMT_serialState_updateChecksum( static void ZSTDMT_serialState_advance(void* opaque) { SerialState* const serialState = (SerialState*)opaque; - serialState->nextJobID++; - ZSTD_pthread_cond_broadcast(&serialState->cond); - ZSTD_pthread_mutex_unlock(&serialState->mutex); + ZSTDMT_RustSerialAdvanceState state; + state.callbackContext = serialState; + state.nextJobID = &serialState->nextJobID; + state.broadcast = ZSTDMT_serialState_broadcast; + state.unlock = ZSTDMT_serialState_unlock; + ZSTDMT_rust_serialStateAdvance(&state); } static void diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 188741fe8..8aea7aa75 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -91,6 +91,8 @@ pub type ZSTDMT_serialUpdateChecksumFn = unsafe extern "C" fn(*mut c_void, *cons 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); +pub type ZSTDMT_serialStateBroadcastFn = unsafe extern "C" fn(*mut c_void); +pub type ZSTDMT_serialStateUnlockFn = 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 @@ -143,6 +145,49 @@ pub unsafe extern "C" fn ZSTDMT_rust_serialStateWaitForTurn( } } +/// Projection for advancing the MT serial turn. Rust owns the increment and +/// callback order; C retains the condition variable broadcast and mutex +/// unlock operations behind the callbacks. +#[repr(C)] +pub struct ZSTDMT_RustSerialAdvanceState { + callback_context: *mut c_void, + next_job_id: *mut c_uint, + broadcast: Option, + unlock: Option, +} + +const _: () = { + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialAdvanceState, callback_context) == 0); + assert!(offset_of!(ZSTDMT_RustSerialAdvanceState, next_job_id) == size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialAdvanceState, broadcast) == 2 * size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialAdvanceState, unlock) == 3 * size_of::()); + assert!(size_of::() == size_of::<[usize; 4]>()); +}; + +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_serialStateAdvance( + state: *const ZSTDMT_RustSerialAdvanceState, +) { + if state.is_null() { + return; + } + let state = unsafe { &*state }; + if state.callback_context.is_null() || state.next_job_id.is_null() { + return; + } + let (Some(broadcast), Some(unlock)) = (state.broadcast, state.unlock) else { + return; + }; + + unsafe { + *state.next_job_id = (*state.next_job_id).wrapping_add(1); + broadcast(state.callback_context); + unlock(state.callback_context); + } +} + #[repr(C)] #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct ZSTDMT_serialStateEnsureFinishedResult { @@ -3540,6 +3585,38 @@ mod tests { } } + struct SerialAdvanceTestContext { + events: Vec<&'static str>, + } + + unsafe extern "C" fn serial_advance_test_broadcast(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("broadcast") + }; + } + + unsafe extern "C" fn serial_advance_test_unlock(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("unlock") + }; + } + + fn serial_advance_test_state( + context: &mut SerialAdvanceTestContext, + next_job_id: &mut c_uint, + ) -> ZSTDMT_RustSerialAdvanceState { + ZSTDMT_RustSerialAdvanceState { + callback_context: context as *mut _ as *mut c_void, + next_job_id, + broadcast: Some(serial_advance_test_broadcast), + unlock: Some(serial_advance_test_unlock), + } + } + unsafe extern "C" fn wait_for_ldm_test_lock(context: *mut c_void) { unsafe { (*context.cast::()) @@ -3661,6 +3738,30 @@ mod tests { assert_eq!(context.next_job_id, 5); } + #[test] + fn serial_advance_increments_before_broadcast_and_unlock() { + let mut context = SerialAdvanceTestContext { events: Vec::new() }; + let mut next_job_id = 7; + let state = serial_advance_test_state(&mut context, &mut next_job_id); + + unsafe { ZSTDMT_rust_serialStateAdvance(&state) }; + + assert_eq!(next_job_id, 8); + assert_eq!(context.events, vec!["broadcast", "unlock"]); + } + + #[test] + fn serial_advance_wraps_the_job_counter_like_c_unsigned_arithmetic() { + let mut context = SerialAdvanceTestContext { events: Vec::new() }; + let mut next_job_id = c_uint::MAX; + let state = serial_advance_test_state(&mut context, &mut next_job_id); + + unsafe { ZSTDMT_rust_serialStateAdvance(&state) }; + + assert_eq!(next_job_id, 0); + assert_eq!(context.events, vec!["broadcast", "unlock"]); + } + fn record_compression_job_event(state: &Rc>, event: &'static str) { state.borrow_mut().events.push(event); }