From 28a74c5edbad1aeed23ed9a759f09b219b69f813 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Mon, 20 Jul 2026 15:59:55 +0200 Subject: [PATCH] feat(mt): move compression-job finish policy into Rust The MT worker previously reported errors through one C callback and left a combined C finish callback responsible for serial completion, private resource release, output-size publication, consumed-size publication, and signaling. That kept the branch and lifecycle policy on the C side of the existing Rust stage scheduler. Add a narrow finish projection whose C callbacks expose only those private leaves and the source-size scalar. Rust now publishes an error before serial completion on every failed resource or codec stage, releases the sequence and CCtx resources in the original order, publishes the final block size only on success, then publishes the consumed size and signals the job condition. The failed path therefore normalizes any codec-reported last-block size to zero without changing the C-owned mutex, descriptor, pool, or context layouts. Focused tests cover successful final-block publication, resource failure, codec failure, cleanup ordering, consumed-size publication, and error-path normalization. Test Plan: - `rustfmt --check --edition 2021 rust/src/zstdmt_compress.rs` -- passed - `cc -fsyntax-only -Werror=incompatible-pointer-types -Ilib -Ilib/common -Ilib/compress -Ilib/decompress -Ilib/dict -Ilib/legacy lib/compress/zstdmt_compress.c` -- passed - `git diff --check -- lib/compress/zstdmt_compress.c rust/src/zstdmt_compress.rs` and `git diff --cached --check` -- passed - Rust unit tests were added but not run because the request prohibited Cargo and heavy commands --- lib/compress/zstdmt_compress.c | 124 ++++++++++--- rust/src/zstdmt_compress.rs | 311 ++++++++++++++++++++++++++++----- 2 files changed, 374 insertions(+), 61 deletions(-) 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 {