From 701bc899ad4dded8ace236382f00402f84f2f72a Mon Sep 17 00:00:00 2001 From: ddidderr Date: Mon, 20 Jul 2026 18:00:40 +0200 Subject: [PATCH] refactor(mt): move job preparation order to Rust Make Rust the owner of the compression-job preparation sequence: publish the descriptor, reset the reusable input state, and publish terminal frame state last. Keep all MT job and stream layouts in C behind three focused callbacks, and preserve the nonterminal and terminal branches exactly. Test Plan: ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings; cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check; GCC and Clang syntax-only checks; make -j1 -C tests test (all 41 shell tests, fuzzer, zstd tester, and zstream tester passed). --- lib/compress/zstdmt_compress.c | 34 +++++++++- rust/src/zstdmt_compress.rs | 111 +++++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 1 deletion(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 69ffc7a2f..369217aaf 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -547,6 +547,12 @@ typedef void (*ZSTDMT_prepareJobFn)(void* opaque, unsigned jobID, const ZSTDMT_RustJobInitialization* init); typedef void (*ZSTDMT_writeEmptyJobFn)(void* opaque, unsigned jobID); typedef int (*ZSTDMT_tryAddJobFn)(void* opaque, unsigned jobID); +void ZSTDMT_rust_prepareCompressionJob( + void* opaque, unsigned jobID, + const ZSTDMT_RustJobInitialization* initialization, + ZSTDMT_prepareJobFn initializeJob, + ZSTDMT_prepareJobFn resetInputState, + ZSTDMT_prepareJobFn publishFrameState); ZSTDMT_RustCreateJobResult ZSTDMT_rust_createCompressionJob( const ZSTDMT_RustCreateJobProjection* projection, void* opaque, ZSTDMT_prepareJobFn prepareJob, @@ -2877,7 +2883,7 @@ static void ZSTDMT_writeLastEmptyBlock(ZSTDMT_jobDescription* job) assert(!ZSTD_isError(job->cSize)); } -static void ZSTDMT_prepareCompressionJob( +static void ZSTDMT_prepareCompressionJobDescriptor( void* opaque, unsigned jobID, const ZSTDMT_RustJobInitialization* initialization) { @@ -2908,7 +2914,14 @@ static void ZSTDMT_prepareCompressionJob( job->lastJob = initialization->lastJob; job->frameChecksumNeeded = initialization->frameChecksumNeeded; job->dstFlushed = 0; +} +static void ZSTDMT_prepareCompressionJobInputState( + void* opaque, unsigned jobID, + const ZSTDMT_RustJobInitialization* initialization) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + (void)jobID; /* Update the round buffer position and clear the input buffer to be reset. */ mtctx->roundBuff.pos += initialization->roundBuffPosDelta; mtctx->inBuff.buffer = g_nullBuffer; @@ -2917,6 +2930,14 @@ static void ZSTDMT_prepareCompressionJob( (const BYTE*)initialization->nextPrefixStart, initialization->nextPrefixSize }; +} + +static void ZSTDMT_prepareCompressionJobFrameState( + void* opaque, unsigned jobID, + const ZSTDMT_RustJobInitialization* initialization) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + (void)jobID; if (initialization->lastJob) { mtctx->frameEnded = 1; if (initialization->clearChecksumFlag) @@ -2924,6 +2945,17 @@ static void ZSTDMT_prepareCompressionJob( } } +static void ZSTDMT_prepareCompressionJob( + void* opaque, unsigned jobID, + const ZSTDMT_RustJobInitialization* initialization) +{ + ZSTDMT_rust_prepareCompressionJob( + opaque, jobID, initialization, + ZSTDMT_prepareCompressionJobDescriptor, + ZSTDMT_prepareCompressionJobInputState, + ZSTDMT_prepareCompressionJobFrameState); +} + static void ZSTDMT_writeEmptyCompressionJob(void* opaque, unsigned jobID) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 296ff19a1..b72f04e93 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -1547,6 +1547,68 @@ pub type ZSTDMT_prepareJobFn = unsafe extern "C" fn( pub type ZSTDMT_writeEmptyJobFn = unsafe extern "C" fn(opaque: *mut c_void, job_id: c_uint); pub type ZSTDMT_tryAddJobFn = unsafe extern "C" fn(opaque: *mut c_void, job_id: c_uint) -> c_int; +/// Order the C-owned leaves that publish a newly prepared MT job. +/// +/// The descriptor must become complete before the reusable input range is +/// cleared, and terminal frame state is published last. `lastJob` is a C +/// boolean scalar, so any non-zero value selects the terminal publication. +#[inline] +fn prepare_compression_job_with( + job_id: c_uint, + initialization: &ZSTDMT_jobInitialization, + mut initialize_job: P, + mut reset_input_state: I, + mut publish_frame_state: F, +) where + P: FnMut(c_uint, &ZSTDMT_jobInitialization), + I: FnMut(c_uint, &ZSTDMT_jobInitialization), + F: FnMut(c_uint, &ZSTDMT_jobInitialization), +{ + initialize_job(job_id, initialization); + reset_input_state(job_id, initialization); + if initialization.lastJob != 0 { + publish_frame_state(job_id, initialization); + } +} + +/// Apply the MT job-preparation ordering while C retains the private job +/// descriptor and stream-state fields behind callbacks. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_prepareCompressionJob( + opaque: *mut c_void, + job_id: c_uint, + initialization: *const ZSTDMT_jobInitialization, + initialize_job: Option, + reset_input_state: Option, + publish_frame_state: Option, +) { + let Some(initialization) = (unsafe { initialization.as_ref() }) else { + return; + }; + let (Some(initialize_job), Some(reset_input_state), Some(publish_frame_state)) = + (initialize_job, reset_input_state, publish_frame_state) + else { + return; + }; + if opaque.is_null() { + return; + } + + prepare_compression_job_with( + job_id, + initialization, + |job_id, initialization| unsafe { + initialize_job(opaque, job_id, initialization); + }, + |job_id, initialization| unsafe { + reset_input_state(opaque, job_id, initialization); + }, + |job_id, initialization| unsafe { + publish_frame_state(opaque, job_id, initialization); + }, + ); +} + #[inline] fn create_job_table_full(projection: ZSTDMT_createJobProjection) -> ZSTDMT_createJobResult { ZSTDMT_createJobResult { @@ -7034,6 +7096,55 @@ mod tests { assert_eq!(result.jobReady, 0); } + #[test] + fn job_preparation_orders_descriptor_input_and_terminal_state() { + let initialization = ZSTDMT_jobInitialization { + jobNumber: 7, + lastJob: 2, + clearChecksumFlag: 1, + ..ZSTDMT_jobInitialization::default() + }; + let events = RefCell::new(Vec::new()); + + prepare_compression_job_with( + 3, + &initialization, + |job_id, value| { + assert_eq!(job_id, 3); + assert_eq!(value.jobNumber, 7); + events.borrow_mut().push("descriptor"); + }, + |job_id, value| { + assert_eq!(job_id, 3); + assert_eq!(value.roundBuffPosDelta, 0); + events.borrow_mut().push("input"); + }, + |job_id, value| { + assert_eq!(job_id, 3); + assert_eq!(value.clearChecksumFlag, 1); + events.borrow_mut().push("frame"); + }, + ); + + assert_eq!(events.into_inner(), vec!["descriptor", "input", "frame"]); + } + + #[test] + fn job_preparation_skips_frame_state_for_nonterminal_job() { + let initialization = ZSTDMT_jobInitialization::default(); + let events = RefCell::new(Vec::new()); + + prepare_compression_job_with( + 4, + &initialization, + |_job_id, _value| events.borrow_mut().push("descriptor"), + |_job_id, _value| events.borrow_mut().push("input"), + |_job_id, _value| panic!("nonterminal job must not publish frame state"), + ); + + assert_eq!(events.into_inner(), vec!["descriptor", "input"]); + } + #[test] fn create_job_preserves_pool_success_and_failure_state() { let projection = ZSTDMT_createJobProjection {