diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 4bf8cc168..7021698d4 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -234,8 +234,39 @@ typedef size_t (*ZSTDMT_compressionJobStepFn)(void* opaque); typedef void (*ZSTDMT_compressionJobVoidFn)(void* opaque); typedef ZSTDMT_chunkProcessResult (*ZSTDMT_compressionJobCompressFn)( void* opaque, unsigned lastJob); -typedef void (*ZSTDMT_compressionJobErrorFn)(void* opaque, size_t error); -typedef void (*ZSTDMT_compressionJobFinishFn)(void* opaque, size_t lastBlockSize); +typedef void (*ZSTDMT_compressionJobSizeFn)(void* opaque, size_t size); + +typedef struct { + void* callbackContext; + size_t srcSize; + ZSTDMT_compressionJobVoidFn ensureFinished; + ZSTDMT_compressionJobSizeFn publishError; + ZSTDMT_compressionJobSizeFn publishLastBlockSize; + ZSTDMT_compressionJobVoidFn releaseSeq; + ZSTDMT_compressionJobVoidFn releaseCCtx; + ZSTDMT_compressionJobSizeFn publishConsumed; + ZSTDMT_compressionJobVoidFn signal; +} ZSTDMT_RustCompressionJobFinishProjection; +typedef char ZSTDMT_compression_job_finish_projection_layout[ + (offsetof(ZSTDMT_RustCompressionJobFinishProjection, callbackContext) == 0 + && offsetof(ZSTDMT_RustCompressionJobFinishProjection, srcSize) + == sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFinishProjection, ensureFinished) + == 2 * sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFinishProjection, publishError) + == 3 * sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFinishProjection, publishLastBlockSize) + == 4 * sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFinishProjection, releaseSeq) + == 5 * sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFinishProjection, releaseCCtx) + == 6 * sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFinishProjection, publishConsumed) + == 7 * sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFinishProjection, signal) + == 8 * sizeof(void*) + && sizeof(ZSTDMT_RustCompressionJobFinishProjection) == 9 * sizeof(void*)) + ? 1 : -1]; typedef struct { int checksumFlag; @@ -255,6 +286,7 @@ ZSTDMT_RustCompressionJobParameters ZSTDMT_rust_prepareCompressionJobParameters( void ZSTDMT_rust_compressionJob( const ZSTDMT_RustCompressionJobProjection* projection, + const ZSTDMT_RustCompressionJobFinishProjection* finishProjection, void* opaque, ZSTDMT_compressionJobStepFn acquireResources, ZSTDMT_compressionJobVoidFn prepareParameters, @@ -262,9 +294,7 @@ void ZSTDMT_rust_compressionJob( ZSTDMT_compressionJobStepFn beginJob, ZSTDMT_compressionJobVoidFn applySequences, ZSTDMT_compressionJobCompressFn compressJob, - ZSTDMT_compressionJobVoidFn traceJob, - ZSTDMT_compressionJobErrorFn setError, - ZSTDMT_compressionJobFinishFn finishJob); + ZSTDMT_compressionJobVoidFn traceJob); ZSTDMT_chunkProcessResult ZSTDMT_rust_compressJobChunks( ZSTD_CCtx* cctx, const void* src, size_t srcSize, @@ -1764,7 +1794,19 @@ static void ZSTDMT_compressionJobTrace(void* opaque) ZSTD_CCtx_trace(state->cctx, 0); } -static void ZSTDMT_compressionJobSetError(void* opaque, size_t error) +static void ZSTDMT_compressionJobEnsureFinished(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + ZSTDMT_jobDescription* const job = state->job; + + ZSTDMT_serialState_ensureFinished(job->serial, job->jobID, job->cSize); + if (job->prefix.size > 0) + DEBUGLOG(5, "Finished with prefix: %zx", (size_t)job->prefix.start); + DEBUGLOG(5, "Finished with source: %zx", (size_t)job->src.start); +} + +static void ZSTDMT_compressionJobPublishError(void* opaque, size_t error) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; @@ -1775,24 +1817,54 @@ static void ZSTDMT_compressionJobSetError(void* opaque, size_t error) ZSTD_pthread_mutex_unlock(&job->job_mutex); } -static void ZSTDMT_compressionJobFinish(void* opaque, size_t lastCBlockSize) +static void ZSTDMT_compressionJobPublishLastBlockSize(void* opaque, size_t lastCBlockSize) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; - ZSTDMT_serialState_ensureFinished(job->serial, job->jobID, job->cSize); - if (job->prefix.size > 0) - DEBUGLOG(5, "Finished with prefix: %zx", (size_t)job->prefix.start); - DEBUGLOG(5, "Finished with source: %zx", (size_t)job->src.start); - /* release resources */ - ZSTDMT_releaseSeq(job->seqPool, state->rawSeqStore); - ZSTDMT_releaseCCtx(job->cctxPool, state->cctx); - /* report */ ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex); - if (ZSTD_isError(job->cSize)) assert(lastCBlockSize == 0); job->cSize += lastCBlockSize; - job->consumed = job->src.size; /* when job->consumed == job->src.size , compression job is presumed completed */ + ZSTD_pthread_mutex_unlock(&job->job_mutex); +} + +static void ZSTDMT_compressionJobReleaseSeq(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + ZSTDMT_jobDescription* const job = state->job; + + ZSTDMT_releaseSeq(job->seqPool, state->rawSeqStore); +} + +static void ZSTDMT_compressionJobReleaseCCtx(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + ZSTDMT_jobDescription* const job = state->job; + + ZSTDMT_releaseCCtx(job->cctxPool, state->cctx); +} + +static void ZSTDMT_compressionJobPublishConsumed(void* opaque, size_t srcSize) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + ZSTDMT_jobDescription* const job = state->job; + + assert(srcSize == job->src.size); + ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex); + job->consumed = srcSize; /* when job->consumed == job->src.size , compression job is presumed completed */ + ZSTD_pthread_mutex_unlock(&job->job_mutex); +} + +static void ZSTDMT_compressionJobSignal(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + ZSTDMT_jobDescription* const job = state->job; + + ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex); ZSTD_pthread_cond_signal(&job->job_cond); ZSTD_pthread_mutex_unlock(&job->job_mutex); } @@ -1803,6 +1875,7 @@ static void ZSTDMT_compressionJob(void* jobDescription) { ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription; ZSTDMT_compressionJobState state; + ZSTDMT_RustCompressionJobFinishProjection finishProjection; ZSTDMT_RustCompressionJobProjection const projection = { job->firstJob, job->lastJob, @@ -1813,18 +1886,27 @@ static void ZSTDMT_compressionJob(void* jobDescription) state.job = job; state.jobParams = job->params; /* do not modify job->params ! copy it, modify the copy */ state.dstBuff = job->dstBuff; + finishProjection = (ZSTDMT_RustCompressionJobFinishProjection){ + &state, + job->src.size, + ZSTDMT_compressionJobEnsureFinished, + ZSTDMT_compressionJobPublishError, + ZSTDMT_compressionJobPublishLastBlockSize, + ZSTDMT_compressionJobReleaseSeq, + ZSTDMT_compressionJobReleaseCCtx, + ZSTDMT_compressionJobPublishConsumed, + ZSTDMT_compressionJobSignal + }; ZSTDMT_rust_compressionJob( - &projection, &state, + &projection, &finishProjection, &state, ZSTDMT_compressionJobAcquireResources, ZSTDMT_compressionJobPrepareParameters, ZSTDMT_compressionJobGenerateSequences, ZSTDMT_compressionJobBegin, ZSTDMT_compressionJobApplySequences, ZSTDMT_compressionJobCompress, - ZSTDMT_compressionJobTrace, - ZSTDMT_compressionJobSetError, - ZSTDMT_compressionJobFinish); + ZSTDMT_compressionJobTrace); } diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 940497545..89c1d85e2 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -265,8 +265,54 @@ pub type ZSTDMT_compressionJobStepFn = unsafe extern "C" fn(*mut c_void) -> usiz pub type ZSTDMT_compressionJobVoidFn = unsafe extern "C" fn(*mut c_void); pub type ZSTDMT_compressionJobCompressFn = unsafe extern "C" fn(*mut c_void, c_uint) -> ZSTDMT_chunkProcessResult; -pub type ZSTDMT_compressionJobErrorFn = unsafe extern "C" fn(*mut c_void, usize); -pub type ZSTDMT_compressionJobFinishFn = unsafe extern "C" fn(*mut c_void, usize); +pub type ZSTDMT_compressionJobSizeFn = unsafe extern "C" fn(*mut c_void, usize); + +/// C-owned leaves for the worker-job finish path. +/// +/// Rust owns the error/success branch and callback order. C keeps the job +/// descriptor, pthread objects, private pools, and serial state behind these +/// callbacks; the source size is the only scalar needed for consumed-size +/// publication. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct ZSTDMT_compressionJobFinishProjection { + pub callbackContext: *mut c_void, + pub srcSize: usize, + pub ensureFinished: Option, + pub publishError: Option, + pub publishLastBlockSize: Option, + pub releaseSeq: Option, + pub releaseCCtx: Option, + pub publishConsumed: Option, + pub signal: Option, +} + +const _: () = { + assert!(offset_of!(ZSTDMT_compressionJobFinishProjection, callbackContext) == 0); + assert!(offset_of!(ZSTDMT_compressionJobFinishProjection, srcSize) == size_of::()); + assert!( + offset_of!(ZSTDMT_compressionJobFinishProjection, ensureFinished) == 2 * size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFinishProjection, publishError) == 3 * size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFinishProjection, publishLastBlockSize) + == 4 * size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFinishProjection, releaseSeq) == 5 * size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFinishProjection, releaseCCtx) == 6 * size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFinishProjection, publishConsumed) + == 7 * size_of::() + ); + assert!(offset_of!(ZSTDMT_compressionJobFinishProjection, signal) == 8 * size_of::()); + assert!(size_of::() == 9 * size_of::()); +}; pub type ZSTDMT_serialWaitForTurnFn = unsafe extern "C" fn(*mut c_void, c_uint) -> c_int; pub type ZSTDMT_serialGenerateLdmFn = @@ -1420,16 +1466,58 @@ pub unsafe extern "C" fn ZSTDMT_rust_createCompressionJob( ) } +/// Finish a worker job with Rust-owned branch and callback ordering. +/// +/// Error publication happens before serial completion and resource release, +/// matching the original C path so failed jobs advance the serial state with +/// the published error. Success publishes the final block size only after the +/// private sequence and CCtx callbacks return. Both paths publish the full +/// consumed size and signal only after all preceding leaves have run. +#[inline] +fn finish_compression_job_with( + error: usize, + last_block_size: usize, + src_size: usize, + mut publish_error: E, + mut ensure_finished: F, + mut release_seq: R, + mut release_cctx: C, + mut publish_last_block_size: B, + mut publish_consumed: P, + mut signal: S, +) where + E: FnMut(usize), + F: FnMut(), + R: FnMut(), + C: FnMut(), + B: FnMut(usize), + P: FnMut(usize), + S: FnMut(), +{ + let failed = ERR_isError(error); + if failed { + publish_error(error); + } + ensure_finished(); + release_seq(); + release_cctx(); + if !failed { + publish_last_block_size(last_block_size); + } + publish_consumed(src_size); + signal(); +} + /// Run the high-level worker-job sequence while C owns all codec and /// synchronization operations behind callbacks. /// /// Resource acquisition and every codec-facing stage can fail with a zstd -/// error. Rust stops at the first such error, reports it before the common -/// C-owned cleanup callback, and passes the final block size only on the -/// successful compression path. +/// error. Rust stops at the first such error, normalizes the final block size, +/// then applies the C-owned finish leaves through the Rust-owned policy above. #[inline] -fn compression_job_with( +fn compression_job_with( projection: ZSTDMT_compressionJobProjection, + finish: ZSTDMT_compressionJobFinishProjection, mut acquire_resources: A, mut prepare_parameters: P, mut generate_sequences: S, @@ -1438,8 +1526,6 @@ fn compression_job_with( mut write_frame_header: H, mut compress_job: C, mut trace_job: T, - mut set_error: E, - mut finish_job: F, ) where A: FnMut() -> usize, P: FnMut(), @@ -1449,9 +1535,28 @@ fn compression_job_with( H: FnMut() -> usize, C: FnMut(c_uint) -> ZSTDMT_chunkProcessResult, T: FnMut(), - E: FnMut(usize), - F: FnMut(usize), { + let ( + Some(ensure_finished), + Some(publish_error), + Some(publish_last_block_size), + Some(release_seq), + Some(release_cctx), + Some(publish_consumed), + Some(signal), + ) = ( + finish.ensureFinished, + finish.publishError, + finish.publishLastBlockSize, + finish.releaseSeq, + finish.releaseCCtx, + finish.publishConsumed, + finish.signal, + ) + else { + return; + }; + let mut error = acquire_resources(); let mut last_block_size = 0; @@ -1479,10 +1584,22 @@ fn compression_job_with( } if ERR_isError(error) { - set_error(error); last_block_size = 0; } - finish_job(last_block_size); + finish_compression_job_with( + error, + last_block_size, + finish.srcSize, + |error| unsafe { publish_error(finish.callbackContext, error) }, + || unsafe { ensure_finished(finish.callbackContext) }, + || unsafe { release_seq(finish.callbackContext) }, + || unsafe { release_cctx(finish.callbackContext) }, + |last_block_size| unsafe { + publish_last_block_size(finish.callbackContext, last_block_size) + }, + |src_size| unsafe { publish_consumed(finish.callbackContext, src_size) }, + || unsafe { signal(finish.callbackContext) }, + ); } /// Drive one ordered MT serial-state turn. @@ -1677,6 +1794,7 @@ pub unsafe extern "C" fn ZSTDMT_rust_releaseAllJobResources( #[no_mangle] pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( projection: *const ZSTDMT_compressionJobProjection, + finishProjection: *const ZSTDMT_compressionJobFinishProjection, opaque: *mut c_void, acquireResources: Option, prepareParameters: Option, @@ -1685,12 +1803,13 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( applySequences: Option, compressJob: Option, traceJob: Option, - setError: Option, - finishJob: Option, ) { let Some(projection) = (unsafe { projection.as_ref() }).copied() else { return; }; + let Some(finish) = (unsafe { finishProjection.as_ref() }).copied() else { + return; + }; let ( Some(acquire_resources), Some(prepare_parameters), @@ -1699,8 +1818,6 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( Some(apply_sequences), Some(compress_job), Some(trace_job), - Some(set_error), - Some(finish_job), ) = ( acquireResources, prepareParameters, @@ -1709,8 +1826,6 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( applySequences, compressJob, traceJob, - setError, - finishJob, ) else { return; @@ -1718,6 +1833,7 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( compression_job_with( projection, + finish, || unsafe { acquire_resources(opaque) }, || unsafe { prepare_parameters(opaque) }, || unsafe { generate_sequences(opaque) }, @@ -1726,8 +1842,6 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( || unsafe { ZSTDMT_rust_writeFrameHeader(projection.frameHeaderState) }, |last_job| unsafe { compress_job(opaque, last_job) }, || unsafe { trace_job(opaque) }, - |error| unsafe { set_error(opaque, error) }, - |last_block_size| unsafe { finish_job(opaque, last_block_size) }, ); } @@ -4818,6 +4932,7 @@ mod tests { events: Vec<&'static str>, errors: Vec, finished: Vec, + consumed: Vec, } struct WaitForLdmTestContext { @@ -5174,6 +5289,78 @@ mod tests { state.borrow_mut().events.push(event); } + fn record_compression_job_finish_event(context: *mut c_void, event: &'static str) { + unsafe { + (*context.cast::>()) + .borrow_mut() + .events + .push(event); + } + } + + unsafe extern "C" fn compression_job_finish_test_ensure(context: *mut c_void) { + record_compression_job_finish_event(context, "ensure"); + } + + unsafe extern "C" fn compression_job_finish_test_error(context: *mut c_void, error: usize) { + unsafe { + let state = &*context.cast::>(); + state.borrow_mut().errors.push(error); + } + record_compression_job_finish_event(context, "error"); + } + + unsafe extern "C" fn compression_job_finish_test_last_block( + context: *mut c_void, + last_block_size: usize, + ) { + unsafe { + let state = &*context.cast::>(); + state.borrow_mut().finished.push(last_block_size); + } + record_compression_job_finish_event(context, "last-block"); + } + + unsafe extern "C" fn compression_job_finish_test_release_seq(context: *mut c_void) { + record_compression_job_finish_event(context, "release-seq"); + } + + unsafe extern "C" fn compression_job_finish_test_release_cctx(context: *mut c_void) { + record_compression_job_finish_event(context, "release-cctx"); + } + + unsafe extern "C" fn compression_job_finish_test_consumed( + context: *mut c_void, + src_size: usize, + ) { + unsafe { + let state = &*context.cast::>(); + state.borrow_mut().consumed.push(src_size); + } + record_compression_job_finish_event(context, "consumed"); + } + + unsafe extern "C" fn compression_job_finish_test_signal(context: *mut c_void) { + record_compression_job_finish_event(context, "signal"); + } + + fn compression_job_finish_test_projection( + state: &Rc>, + src_size: usize, + ) -> ZSTDMT_compressionJobFinishProjection { + ZSTDMT_compressionJobFinishProjection { + callbackContext: Rc::as_ptr(state).cast_mut().cast(), + srcSize: src_size, + ensureFinished: Some(compression_job_finish_test_ensure), + publishError: Some(compression_job_finish_test_error), + publishLastBlockSize: Some(compression_job_finish_test_last_block), + releaseSeq: Some(compression_job_finish_test_release_seq), + releaseCCtx: Some(compression_job_finish_test_release_cctx), + publishConsumed: Some(compression_job_finish_test_consumed), + signal: Some(compression_job_finish_test_signal), + } + } + #[test] fn compression_job_frame_header_projection_writes_header_and_clears_repcodes() { let mut stage = ZSTD_COMPRESSION_STAGE_INIT; @@ -5221,7 +5408,7 @@ mod tests { let apply_state = Rc::clone(&state); let compress_state = Rc::clone(&state); let trace_state = Rc::clone(&state); - let finish_state = Rc::clone(&state); + let finish = compression_job_finish_test_projection(&state, 19); compression_job_with( ZSTDMT_compressionJobProjection { @@ -5229,6 +5416,7 @@ mod tests { lastJob: 1, frameHeaderState: ptr::null(), }, + finish, move || { record_compression_job_event(&acquire_state, "acquire"); 0 @@ -5250,11 +5438,6 @@ mod tests { } }, move || record_compression_job_event(&trace_state, "trace"), - |_error| panic!("success must not report an error"), - move |last_block_size| { - record_compression_job_event(&finish_state, "finish"); - finish_state.borrow_mut().finished.push(last_block_size); - }, ); let state = state.borrow(); @@ -5268,11 +5451,62 @@ mod tests { "apply", "compress", "trace", - "finish" + "ensure", + "release-seq", + "release-cctx", + "last-block", + "consumed", + "signal", ] ); assert!(state.errors.is_empty()); assert_eq!(state.finished, vec![7]); + assert_eq!(state.consumed, vec![19]); + } + + #[test] + fn compression_job_normalizes_resource_error_before_finish_cleanup() { + let state = Rc::new(RefCell::new(MockCompressionJob::default())); + let acquire_state = Rc::clone(&state); + let expected_error = ERROR(ZstdErrorCode::MemoryAllocation); + let finish = compression_job_finish_test_projection(&state, 11); + + compression_job_with( + ZSTDMT_compressionJobProjection { + firstJob: 1, + lastJob: 0, + frameHeaderState: ptr::null(), + }, + finish, + move || { + record_compression_job_event(&acquire_state, "acquire"); + expected_error + }, + || panic!("resource failure must skip parameter preparation"), + || panic!("resource failure must skip sequence generation"), + || panic!("resource failure must skip context initialization"), + || panic!("resource failure must skip sequence application"), + || panic!("resource failure must skip frame-header writing"), + || panic!("resource failure must skip compression"), + || panic!("resource failure must skip tracing"), + ); + + let state = state.borrow(); + assert_eq!( + state.events, + vec![ + "acquire", + "error", + "ensure", + "release-seq", + "release-cctx", + "consumed", + "signal", + ] + ); + assert_eq!(state.errors, vec![expected_error]); + assert!(state.finished.is_empty()); + assert_eq!(state.consumed, vec![11]); } #[test] @@ -5460,9 +5694,8 @@ mod tests { let apply_state = Rc::clone(&state); let header_state = Rc::clone(&state); let compress_state = Rc::clone(&state); - let error_state = Rc::clone(&state); - let finish_state = Rc::clone(&state); let expected_error = ERROR(ZstdErrorCode::DstSizeTooSmall); + let finish = compression_job_finish_test_projection(&state, 23); compression_job_with( ZSTDMT_compressionJobProjection { @@ -5470,6 +5703,7 @@ mod tests { lastJob: 1, frameHeaderState: ptr::null(), }, + finish, move || { record_compression_job_event(&acquire_state, "acquire"); 0 @@ -5490,18 +5724,10 @@ mod tests { record_compression_job_event(&compress_state, "compress"); ZSTDMT_chunkProcessResult { error: expected_error, - lastBlockSize: 0, + lastBlockSize: 9, } }, || panic!("a failed chunk must not be traced"), - move |error| { - record_compression_job_event(&error_state, "error"); - error_state.borrow_mut().errors.push(error); - }, - move |last_block_size| { - record_compression_job_event(&finish_state, "finish"); - finish_state.borrow_mut().finished.push(last_block_size); - }, ); let state = state.borrow(); @@ -5516,11 +5742,16 @@ mod tests { "header", "compress", "error", - "finish" + "ensure", + "release-seq", + "release-cctx", + "consumed", + "signal", ] ); assert_eq!(state.errors, vec![expected_error]); - assert_eq!(state.finished, vec![0]); + assert!(state.finished.is_empty()); + assert_eq!(state.consumed, vec![23]); } fn init_projection() -> ZSTDMT_initCStreamProjection {