feat(mt): move serial failure turn policy into Rust
Move the scalar failed-job skip decision and wrapping next-job transition into Rust. C keeps the serial mutex, error assertion, condition broadcast, and LDM window cleanup around the projected result. Test Plan: - ulimit -v 41943040; CARGO_BUILD_JOBS=1; cargo test --manifest-path rust/Cargo.toml - ulimit -v 41943040; CARGO_BUILD_JOBS=1; cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings - ulimit -v 41943040; make -B -C programs -j1 zstd - ulimit -v 41943040; make -C tests -j1 test-zstream ZSTREAM_TESTTIME=-T1s
This commit is contained in:
@@ -368,6 +368,12 @@ void ZSTDMT_rust_serialStateGenSequences(
|
||||
ZSTDMT_serialGenerateLdmFn generateLdm,
|
||||
ZSTDMT_serialUpdateChecksumFn updateChecksum,
|
||||
ZSTDMT_serialAdvanceFn advance);
|
||||
typedef struct {
|
||||
unsigned skip;
|
||||
unsigned nextJobID;
|
||||
} ZSTDMT_RustSerialEnsureFinishedResult;
|
||||
ZSTDMT_RustSerialEnsureFinishedResult ZSTDMT_rust_serialStateEnsureFinished(
|
||||
unsigned nextJobID, unsigned jobID);
|
||||
typedef void (*ZSTDMT_waitForLdmLockFn)(void* opaque);
|
||||
typedef int (*ZSTDMT_waitForLdmOverlapFn)(
|
||||
void* opaque, void* bufferStart, size_t bufferCapacity);
|
||||
@@ -975,11 +981,14 @@ 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);
|
||||
if (serialState->nextJobID <= jobID) {
|
||||
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 = jobID + 1;
|
||||
serialState->nextJobID = result.nextJobID;
|
||||
ZSTD_pthread_cond_broadcast(&serialState->cond);
|
||||
|
||||
ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex);
|
||||
|
||||
@@ -90,6 +90,13 @@ pub type ZSTDMT_serialGenerateLdmFn =
|
||||
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);
|
||||
|
||||
#[repr(C)]
|
||||
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
|
||||
pub struct ZSTDMT_serialStateEnsureFinishedResult {
|
||||
pub skip: c_uint,
|
||||
pub nextJobID: c_uint,
|
||||
}
|
||||
|
||||
type ZSTDMT_waitForLdmLockFn = unsafe extern "C" fn(*mut c_void);
|
||||
type ZSTDMT_waitForLdmOverlapFn = unsafe extern "C" fn(*mut c_void, *mut c_void, usize) -> c_int;
|
||||
type ZSTDMT_waitForLdmWaitFn = unsafe extern "C" fn(*mut c_void);
|
||||
@@ -836,6 +843,27 @@ 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.
|
||||
#[no_mangle]
|
||||
pub extern "C" fn ZSTDMT_rust_serialStateEnsureFinished(
|
||||
next_job_id: c_uint,
|
||||
job_id: c_uint,
|
||||
) -> ZSTDMT_serialStateEnsureFinishedResult {
|
||||
if next_job_id <= job_id {
|
||||
ZSTDMT_serialStateEnsureFinishedResult {
|
||||
skip: 1,
|
||||
nextJobID: job_id.wrapping_add(1),
|
||||
}
|
||||
} else {
|
||||
ZSTDMT_serialStateEnsureFinishedResult {
|
||||
skip: 0,
|
||||
nextJobID: next_job_id,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// C ABI entry point for the worker-job orchestration. C supplies callbacks
|
||||
/// that keep the private descriptor, pools, mutexes, and codec operations on
|
||||
/// the C side of this narrow projection.
|
||||
@@ -3481,6 +3509,38 @@ mod tests {
|
||||
assert_eq!(&*empty_events.borrow(), &["ldm", "advance"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serial_state_ensure_finished_owns_skip_and_wrapping_policy() {
|
||||
assert_eq!(
|
||||
ZSTDMT_rust_serialStateEnsureFinished(4, 4),
|
||||
ZSTDMT_serialStateEnsureFinishedResult {
|
||||
skip: 1,
|
||||
nextJobID: 5,
|
||||
}
|
||||
);
|
||||
assert_eq!(
|
||||
ZSTDMT_rust_serialStateEnsureFinished(3, 9),
|
||||
ZSTDMT_serialStateEnsureFinishedResult {
|
||||
skip: 1,
|
||||
nextJobID: 10,
|
||||
}
|
||||
);
|
||||
assert_eq!(
|
||||
ZSTDMT_rust_serialStateEnsureFinished(11, 9),
|
||||
ZSTDMT_serialStateEnsureFinishedResult {
|
||||
skip: 0,
|
||||
nextJobID: 11,
|
||||
}
|
||||
);
|
||||
assert_eq!(
|
||||
ZSTDMT_rust_serialStateEnsureFinished(c_uint::MAX, c_uint::MAX),
|
||||
ZSTDMT_serialStateEnsureFinishedResult {
|
||||
skip: 1,
|
||||
nextJobID: 0,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn compression_job_stops_on_non_first_chunk_error_and_cleans_up() {
|
||||
let state = Rc::new(RefCell::new(MockCompressionJob::default()));
|
||||
|
||||
Reference in New Issue
Block a user