ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(mt): move serial wait loop into Rust

Browse Source 
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
This commit is contained in:
ddidderr
2026-07-19 20:47:58 +02:00
parent b8e70c1467
commit 160fa29148
2 changed files with 155 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstdmt_compress.c
+38 -6
View File
@@ -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(
rust/src/zstdmt_compress.rs
+117
View File
@@ -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<ZSTDMT_serialStateLockFn>,
wait: Option<ZSTDMT_serialStateWaitFn>,
job_id: c_uint,
}
const _: () = {
assert!(size_of::<ZSTDMT_serialStateLockFn>() == size_of::<usize>());
assert!(size_of::<ZSTDMT_serialStateWaitFn>() == size_of::<usize>());
assert!(offset_of!(ZSTDMT_RustSerialWaitForTurnState, callback_context) == 0);
assert!(offset_of!(ZSTDMT_RustSerialWaitForTurnState, next_job_id) == size_of::<usize>());
assert!(offset_of!(ZSTDMT_RustSerialWaitForTurnState, lock) == 2 * size_of::<usize>());
assert!(offset_of!(ZSTDMT_RustSerialWaitForTurnState, wait) == 3 * size_of::<usize>());
assert!(offset_of!(ZSTDMT_RustSerialWaitForTurnState, job_id) == 4 * size_of::<usize>());
assert!(size_of::<ZSTDMT_RustSerialWaitForTurnState>() == 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::<SerialWaitForTurnTestContext>())
.events
.push("lock")
};
}
unsafe extern "C" fn serial_wait_for_turn_test_wait(context: *mut c_void) {
unsafe {
let context = &mut *context.cast::<SerialWaitForTurnTestContext>();
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::<WaitForLdmTestContext>())
@@ -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<RefCell<MockCompressionJob>>, event: &'static str) {
state.borrow_mut().events.push(event);
}