From 591794e15aa1e5b845264aeb2691e79f4815f9e3 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sun, 19 Jul 2026 20:58:18 +0200 Subject: [PATCH] feat(mt): move failed-job serial cleanup into Rust Move the failed-job serial-state orchestration out of zstdmt_compress.c. Rust now owns the lock, skip decision, serial-counter publication, broadcast, LDM cleanup ordering, and final unlock. C retains the pthread objects, private LDM window, and error/debug leaves behind callbacks. Add ABI layout checks and focused tests for both skipped predecessors and later jobs that only need the lock/unlock pair. Test Plan: - cargo fmt --manifest-path rust/Cargo.toml -- --check - ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings - ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml zstdmt_compress::tests::serial_ensure_finished -- --nocapture - ulimit -v 41943040 && make -j1 - ulimit -v 41943040 && make -j1 -C tests test-fuzzer FUZZERTEST=-T3s FUZZER_FLAGS=--no-big-tests --- lib/compress/zstdmt_compress.c | 101 +++++++++++--- rust/src/zstdmt_compress.rs | 234 ++++++++++++++++++++++++++++++++- 2 files changed, 313 insertions(+), 22 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index d33308707..170cd3bb9 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -407,6 +407,40 @@ typedef struct { } ZSTDMT_RustSerialEnsureFinishedResult; ZSTDMT_RustSerialEnsureFinishedResult ZSTDMT_rust_serialStateEnsureFinished( unsigned nextJobID, unsigned jobID); +typedef void (*ZSTDMT_serialStateCallbackFn)(void* opaque); +typedef void (*ZSTDMT_serialStateSkipFn)( + void* opaque, unsigned jobID, size_t cSize); +typedef struct { + void* callbackContext; + unsigned* nextJobID; + ZSTDMT_serialStateCallbackFn lock; + ZSTDMT_serialStateCallbackFn broadcast; + ZSTDMT_serialStateCallbackFn ldmLock; + ZSTDMT_serialStateCallbackFn clearLdmWindow; + ZSTDMT_serialStateCallbackFn ldmSignal; + ZSTDMT_serialStateCallbackFn ldmUnlock; + ZSTDMT_serialStateCallbackFn unlock; + ZSTDMT_serialStateSkipFn onSkip; + size_t cSize; + unsigned jobID; +} ZSTDMT_RustSerialEnsureFinishedState; +typedef char ZSTDMT_rust_serial_ensure_finished_state_layout[ + (offsetof(ZSTDMT_RustSerialEnsureFinishedState, callbackContext) == 0 + && offsetof(ZSTDMT_RustSerialEnsureFinishedState, nextJobID) == sizeof(void*) + && offsetof(ZSTDMT_RustSerialEnsureFinishedState, lock) == 2 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialEnsureFinishedState, broadcast) == 3 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialEnsureFinishedState, ldmLock) == 4 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialEnsureFinishedState, clearLdmWindow) == 5 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialEnsureFinishedState, ldmSignal) == 6 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialEnsureFinishedState, ldmUnlock) == 7 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialEnsureFinishedState, unlock) == 8 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialEnsureFinishedState, onSkip) == 9 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialEnsureFinishedState, cSize) == 10 * sizeof(void*) + && offsetof(ZSTDMT_RustSerialEnsureFinishedState, jobID) + == 10 * sizeof(void*) + sizeof(size_t) + && sizeof(ZSTDMT_RustSerialEnsureFinishedState) == 12 * sizeof(void*)) ? 1 : -1]; +void ZSTDMT_rust_serialStateEnsureFinishedOrchestrated( + const ZSTDMT_RustSerialEnsureFinishedState* state); typedef void (*ZSTDMT_waitForJobCompleteFn)( void* opaque, unsigned jobID, unsigned doneJobID); unsigned ZSTDMT_rust_waitForAllJobsCompleted( @@ -1207,6 +1241,40 @@ static void ZSTDMT_serialState_advance(void* opaque) ZSTDMT_rust_serialStateAdvance(&state); } +static void ZSTDMT_serialState_ldmLock(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex); +} + +static void ZSTDMT_serialState_clearLdmWindow(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_rust_windowClear((size_t)(serialState->ldmWindow.nextSrc - + serialState->ldmWindow.base), + &serialState->ldmWindow.lowLimit, + &serialState->ldmWindow.dictLimit); +} + +static void ZSTDMT_serialState_ldmSignal(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_pthread_cond_signal(&serialState->ldmWindowCond); +} + +static void ZSTDMT_serialState_ldmUnlock(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_pthread_mutex_unlock(&serialState->ldmWindowMutex); +} + +static void ZSTDMT_serialState_onSkip(void* opaque, unsigned jobID, size_t cSize) +{ + (void)opaque; + assert(ZSTD_isError(cSize)); (void)cSize; + DEBUGLOG(5, "Skipping past job %u because of error", jobID); +} + static void ZSTDMT_serialState_applySequences(const SerialState* serialState, /* just for an assert() check */ ZSTD_CCtx* jobCCtx, @@ -1223,25 +1291,20 @@ ZSTDMT_serialState_applySequences(const SerialState* serialState, /* just for an static void ZSTDMT_serialState_ensureFinished(SerialState* serialState, unsigned jobID, size_t cSize) { - ZSTDMT_RustSerialEnsureFinishedResult result; - ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex); - result = ZSTDMT_rust_serialStateEnsureFinished( - serialState->nextJobID, jobID); - if (result.skip) { - assert(ZSTD_isError(cSize)); (void)cSize; - DEBUGLOG(5, "Skipping past job %u because of error", jobID); - serialState->nextJobID = result.nextJobID; - ZSTD_pthread_cond_broadcast(&serialState->cond); - - ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex); - ZSTD_rust_windowClear((size_t)(serialState->ldmWindow.nextSrc - - serialState->ldmWindow.base), - &serialState->ldmWindow.lowLimit, - &serialState->ldmWindow.dictLimit); - ZSTD_pthread_cond_signal(&serialState->ldmWindowCond); - ZSTD_pthread_mutex_unlock(&serialState->ldmWindowMutex); - } - ZSTD_pthread_mutex_unlock(&serialState->mutex); + ZSTDMT_RustSerialEnsureFinishedState state; + state.callbackContext = serialState; + state.nextJobID = &serialState->nextJobID; + state.lock = ZSTDMT_serialState_lock; + state.broadcast = ZSTDMT_serialState_broadcast; + state.ldmLock = ZSTDMT_serialState_ldmLock; + state.clearLdmWindow = ZSTDMT_serialState_clearLdmWindow; + state.ldmSignal = ZSTDMT_serialState_ldmSignal; + state.ldmUnlock = ZSTDMT_serialState_ldmUnlock; + state.unlock = ZSTDMT_serialState_unlock; + state.onSkip = ZSTDMT_serialState_onSkip; + state.cSize = cSize; + state.jobID = jobID; + ZSTDMT_rust_serialStateEnsureFinishedOrchestrated(&state); } diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 8aea7aa75..11d61dcfe 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -93,6 +93,8 @@ 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); +pub type ZSTDMT_serialStateCallbackFn = unsafe extern "C" fn(*mut c_void); +pub type ZSTDMT_serialStateSkipFn = unsafe extern "C" fn(*mut c_void, c_uint, usize); /// 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 @@ -188,6 +190,50 @@ pub unsafe extern "C" fn ZSTDMT_rust_serialStateAdvance( } } +/// Projection for the failed-job serial cleanup path. Rust owns the lock, +/// skip decision, serial-counter publication, and callback ordering; C keeps +/// the mutex, condition variables, and private LDM-window cleanup behind the +/// callbacks. +#[repr(C)] +pub struct ZSTDMT_RustSerialEnsureFinishedState { + callback_context: *mut c_void, + next_job_id: *mut c_uint, + lock: Option, + broadcast: Option, + ldm_lock: Option, + clear_ldm_window: Option, + ldm_signal: Option, + ldm_unlock: Option, + unlock: Option, + on_skip: Option, + c_size: usize, + job_id: c_uint, +} + +const _: () = { + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, callback_context) == 0); + assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, next_job_id) == size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, lock) == 2 * size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, broadcast) == 3 * size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, ldm_lock) == 4 * size_of::()); + assert!( + offset_of!(ZSTDMT_RustSerialEnsureFinishedState, clear_ldm_window) + == 5 * size_of::() + ); + assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, ldm_signal) == 6 * size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, ldm_unlock) == 7 * size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, unlock) == size_of::<[usize; 8]>()); + assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, on_skip) == 9 * size_of::()); + assert!(offset_of!(ZSTDMT_RustSerialEnsureFinishedState, c_size) == 10 * size_of::()); + assert!( + offset_of!(ZSTDMT_RustSerialEnsureFinishedState, job_id) + == 10 * size_of::() + size_of::() + ); + assert!(size_of::() == size_of::<[usize; 12]>()); +}; + #[repr(C)] #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct ZSTDMT_serialStateEnsureFinishedResult { @@ -948,9 +994,7 @@ pub unsafe extern "C" fn ZSTDMT_rust_serialStateGenSequences( } } -/// Decide whether a failed job must advance the shared serial turn. C keeps -/// the mutex, error assertion, condition broadcast, and LDM-window cleanup; -/// Rust owns the scalar comparison and wrapping job-id transition. +/// Decide whether a failed job must advance the shared serial turn. #[no_mangle] pub extern "C" fn ZSTDMT_rust_serialStateEnsureFinished( next_job_id: c_uint, @@ -969,6 +1013,58 @@ pub extern "C" fn ZSTDMT_rust_serialStateEnsureFinished( } } +/// Own the failed-job serial cleanup order while C retains synchronization and +/// private LDM-window operations behind callbacks. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_serialStateEnsureFinishedOrchestrated( + state: *const ZSTDMT_RustSerialEnsureFinishedState, +) { + if state.is_null() { + return; + } + let state = unsafe { &*state }; + if state.callback_context.is_null() || state.next_job_id.is_null() { + return; + } + let ( + Some(lock), + Some(broadcast), + Some(ldm_lock), + Some(clear_ldm_window), + Some(ldm_signal), + Some(ldm_unlock), + Some(unlock), + Some(on_skip), + ) = ( + state.lock, + state.broadcast, + state.ldm_lock, + state.clear_ldm_window, + state.ldm_signal, + state.ldm_unlock, + state.unlock, + state.on_skip, + ) + else { + return; + }; + + unsafe { + lock(state.callback_context); + let result = ZSTDMT_rust_serialStateEnsureFinished(*state.next_job_id, state.job_id); + if result.skip != 0 { + on_skip(state.callback_context, state.job_id, state.c_size); + *state.next_job_id = result.nextJobID; + broadcast(state.callback_context); + ldm_lock(state.callback_context); + clear_ldm_window(state.callback_context); + ldm_signal(state.callback_context); + ldm_unlock(state.callback_context); + } + unlock(state.callback_context); + } +} + /// Wait for each submitted MT job in ring order. C retains the per-job /// mutex, condition variable, and consumed/source counters behind one /// callback; Rust owns the ring-slot and completion progression policy. @@ -3617,6 +3713,102 @@ mod tests { } } + struct SerialEnsureFinishedTestContext { + events: Vec<&'static str>, + } + + unsafe extern "C" fn serial_ensure_finished_test_lock(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("lock") + }; + } + + unsafe extern "C" fn serial_ensure_finished_test_broadcast(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("broadcast") + }; + } + + unsafe extern "C" fn serial_ensure_finished_test_ldm_lock(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("ldm_lock") + }; + } + + unsafe extern "C" fn serial_ensure_finished_test_clear_ldm(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("clear_ldm") + }; + } + + unsafe extern "C" fn serial_ensure_finished_test_ldm_signal(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("ldm_signal") + }; + } + + unsafe extern "C" fn serial_ensure_finished_test_ldm_unlock(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("ldm_unlock") + }; + } + + unsafe extern "C" fn serial_ensure_finished_test_unlock(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("unlock") + }; + } + + unsafe extern "C" fn serial_ensure_finished_test_on_skip( + context: *mut c_void, + job_id: c_uint, + c_size: usize, + ) { + assert_eq!(job_id, 4); + assert_eq!(c_size, 123); + unsafe { + (*context.cast::()) + .events + .push("skip") + }; + } + + fn serial_ensure_finished_test_state( + context: &mut SerialEnsureFinishedTestContext, + next_job_id: &mut c_uint, + job_id: c_uint, + c_size: usize, + ) -> ZSTDMT_RustSerialEnsureFinishedState { + ZSTDMT_RustSerialEnsureFinishedState { + callback_context: context as *mut _ as *mut c_void, + next_job_id, + lock: Some(serial_ensure_finished_test_lock), + broadcast: Some(serial_ensure_finished_test_broadcast), + ldm_lock: Some(serial_ensure_finished_test_ldm_lock), + clear_ldm_window: Some(serial_ensure_finished_test_clear_ldm), + ldm_signal: Some(serial_ensure_finished_test_ldm_signal), + ldm_unlock: Some(serial_ensure_finished_test_ldm_unlock), + unlock: Some(serial_ensure_finished_test_unlock), + on_skip: Some(serial_ensure_finished_test_on_skip), + c_size, + job_id, + } + } + unsafe extern "C" fn wait_for_ldm_test_lock(context: *mut c_void) { unsafe { (*context.cast::()) @@ -3762,6 +3954,42 @@ mod tests { assert_eq!(context.events, vec!["broadcast", "unlock"]); } + #[test] + fn serial_ensure_finished_runs_failed_job_cleanup_in_order() { + let mut context = SerialEnsureFinishedTestContext { events: Vec::new() }; + let mut next_job_id = 4; + let state = serial_ensure_finished_test_state(&mut context, &mut next_job_id, 4, 123); + + unsafe { ZSTDMT_rust_serialStateEnsureFinishedOrchestrated(&state) }; + + assert_eq!(next_job_id, 5); + assert_eq!( + context.events, + vec![ + "lock", + "skip", + "broadcast", + "ldm_lock", + "clear_ldm", + "ldm_signal", + "ldm_unlock", + "unlock" + ] + ); + } + + #[test] + fn serial_ensure_finished_unlocks_without_cleanup_for_later_job() { + let mut context = SerialEnsureFinishedTestContext { events: Vec::new() }; + let mut next_job_id = 6; + let state = serial_ensure_finished_test_state(&mut context, &mut next_job_id, 4, 123); + + unsafe { ZSTDMT_rust_serialStateEnsureFinishedOrchestrated(&state) }; + + assert_eq!(next_job_id, 6); + assert_eq!(context.events, vec!["lock", "unlock"]); + } + fn record_compression_job_event(state: &Rc>, event: &'static str) { state.borrow_mut().events.push(event); }