diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 6532f9e98..2330d4303 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -300,16 +300,48 @@ typedef char ZSTDMT_compression_job_progress_projection_layout[ void ZSTDMT_rust_compressionJobProgress( void* opaque, size_t cSize, size_t consumed); +typedef struct { + void* callbackContext; + size_t* cSize; + size_t* consumed; + size_t srcSize; + ZSTDMT_compressionJobVoidFn lock; + ZSTDMT_compressionJobVoidFn signal; + ZSTDMT_compressionJobVoidFn unlock; + ZSTDMT_compressionJobSizeFn validateConsumed; +} ZSTDMT_RustCompressionJobFinishPublicationProjection; +typedef char ZSTDMT_compression_job_finish_publication_projection_layout[ + (offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, + callbackContext) == 0 + && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, cSize) + == sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, + consumed) + == sizeof(void*) + sizeof(size_t) + && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, + srcSize) + == sizeof(void*) + 2 * sizeof(size_t) + && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, lock) + == sizeof(void*) + 3 * sizeof(size_t) + && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, signal) + == sizeof(void*) + 3 * sizeof(size_t) + sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, + unlock) + == sizeof(void*) + 3 * sizeof(size_t) + 2 * sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, + validateConsumed) + == sizeof(void*) + 3 * sizeof(size_t) + 3 * sizeof(void*) + && sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection) + == sizeof(void*) + 3 * sizeof(size_t) + 4 * sizeof(void*)) + ? 1 : -1]; + typedef struct { void* callbackContext; size_t srcSize; ZSTDMT_compressionJobVoidFn ensureFinished; - ZSTDMT_compressionJobSizeFn publishError; - ZSTDMT_compressionJobSizeFn publishLastBlockSize; + ZSTDMT_RustCompressionJobFinishPublicationProjection publication; 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 @@ -317,19 +349,17 @@ typedef char ZSTDMT_compression_job_finish_projection_layout[ == sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobFinishProjection, ensureFinished) == 2 * sizeof(void*) - && offsetof(ZSTDMT_RustCompressionJobFinishProjection, publishError) + && offsetof(ZSTDMT_RustCompressionJobFinishProjection, publication) == 3 * sizeof(void*) - && offsetof(ZSTDMT_RustCompressionJobFinishProjection, publishLastBlockSize) - == 4 * sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobFinishProjection, releaseSeq) - == 5 * sizeof(void*) + == 3 * sizeof(void*) + + sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection) && 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*)) + == 4 * sizeof(void*) + + sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection) + && sizeof(ZSTDMT_RustCompressionJobFinishProjection) + == 5 * sizeof(void*) + + sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection)) ? 1 : -1]; typedef struct { @@ -2243,28 +2273,6 @@ static void ZSTDMT_compressionJobEnsureFinished(void* opaque) 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; - ZSTDMT_jobDescription* const job = state->job; - - ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex); - job->cSize = error; - ZSTD_pthread_mutex_unlock(&job->job_mutex); -} - -static void ZSTDMT_compressionJobPublishLastBlockSize(void* opaque, size_t lastCBlockSize) -{ - ZSTDMT_compressionJobState* const state = - (ZSTDMT_compressionJobState*)opaque; - ZSTDMT_jobDescription* const job = state->job; - - ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex); - job->cSize += lastCBlockSize; - ZSTD_pthread_mutex_unlock(&job->job_mutex); -} - static void ZSTDMT_compressionJobReleaseSeq(void* opaque) { ZSTDMT_compressionJobState* const state = @@ -2283,27 +2291,11 @@ static void ZSTDMT_compressionJobReleaseCCtx(void* opaque) ZSTDMT_releaseCCtx(job->cctxPool, state->cctx); } -static void ZSTDMT_compressionJobPublishConsumed(void* opaque, size_t srcSize) +static void ZSTDMT_compressionJobValidateConsumed(void* opaque, size_t srcSize) { - ZSTDMT_compressionJobState* const state = - (ZSTDMT_compressionJobState*)opaque; - ZSTDMT_jobDescription* const job = state->job; + ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)opaque; 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); } /* ZSTDMT_compressionJob() is a POOL_function type. Rust owns the stage @@ -2313,6 +2305,7 @@ static void ZSTDMT_compressionJob(void* jobDescription) ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription; ZSTDMT_compressionJobState state; ZSTDMT_RustCompressionJobSequenceState sequenceState; + ZSTDMT_RustCompressionJobFinishPublicationProjection publication; ZSTDMT_RustCompressionJobFinishProjection finishProjection; ZSTDMT_RustCompressionJobProjection projection; @@ -2331,16 +2324,23 @@ static void ZSTDMT_compressionJob(void* jobDescription) &state.frameHeaderState, &sequenceState }; + publication = (ZSTDMT_RustCompressionJobFinishPublicationProjection){ + job, + &job->cSize, + &job->consumed, + job->src.size, + ZSTDMT_compressionJobProgressLock, + ZSTDMT_compressionJobProgressSignal, + ZSTDMT_compressionJobProgressUnlock, + ZSTDMT_compressionJobValidateConsumed + }; finishProjection = (ZSTDMT_RustCompressionJobFinishProjection){ &state, job->src.size, ZSTDMT_compressionJobEnsureFinished, - ZSTDMT_compressionJobPublishError, - ZSTDMT_compressionJobPublishLastBlockSize, + publication, ZSTDMT_compressionJobReleaseSeq, - ZSTDMT_compressionJobReleaseCCtx, - ZSTDMT_compressionJobPublishConsumed, - ZSTDMT_compressionJobSignal + ZSTDMT_compressionJobReleaseCCtx }; ZSTDMT_rust_compressionJob( diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index eb40ef915..0a20be203 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -755,24 +755,66 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJobBegin( ) } +/// Live fields and synchronization callbacks for worker-job finish +/// publication. The private job descriptor and pthread objects remain in C; +/// Rust owns the counter updates and completion signal order. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct ZSTDMT_compressionJobFinishPublicationProjection { + pub callbackContext: *mut c_void, + pub cSize: *mut usize, + pub consumed: *mut usize, + pub srcSize: usize, + pub lock: Option, + pub signal: Option, + pub unlock: Option, + pub validateConsumed: Option, +} + +const _: () = { + assert!( + offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, callbackContext) == 0 + ); + assert!( + offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, cSize) + == size_of::<*mut c_void>() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, consumed) + == size_of::<*mut c_void>() + size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, srcSize) + == size_of::<*mut c_void>() + 2 * size_of::() + ); + assert!(offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, lock) + == size_of::<*mut c_void>() + 3 * size_of::()); + assert!(offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, signal) + == size_of::<*mut c_void>() + 3 * size_of::() + size_of::()); + assert!(offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, unlock) + == size_of::<*mut c_void>() + 3 * size_of::() + 2 * size_of::()); + assert!( + offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, validateConsumed) + == size_of::<*mut c_void>() + 3 * size_of::() + 3 * size_of::() + ); + assert!(size_of::() + == size_of::<*mut c_void>() + 3 * size_of::() + 4 * size_of::()); +}; + /// 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. +/// descriptor, pthread objects, private pools, and serial state behind the +/// remaining callbacks; publication state crosses through one live projection. #[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 publication: ZSTDMT_compressionJobFinishPublicationProjection, pub releaseSeq: Option, pub releaseCCtx: Option, - pub publishConsumed: Option, - pub signal: Option, } const _: () = { @@ -781,25 +823,21 @@ const _: () = { assert!( offset_of!(ZSTDMT_compressionJobFinishProjection, ensureFinished) == 2 * size_of::() ); + assert!(offset_of!(ZSTDMT_compressionJobFinishProjection, publication) + == 3 * size_of::()); assert!( - offset_of!(ZSTDMT_compressionJobFinishProjection, publishError) == 3 * size_of::() + offset_of!(ZSTDMT_compressionJobFinishProjection, releaseSeq) + == 3 * size_of::() + + size_of::() ); assert!( - offset_of!(ZSTDMT_compressionJobFinishProjection, publishLastBlockSize) + offset_of!(ZSTDMT_compressionJobFinishProjection, releaseCCtx) == 4 * size_of::() + + 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) == usize::BITS as usize); - assert!(size_of::() == 9 * size_of::()); + assert!(size_of::() + == 5 * size_of::() + + size_of::()); }; pub type ZSTDMT_serialWaitForTurnFn = unsafe extern "C" fn(*mut c_void, c_uint) -> c_int; @@ -2570,6 +2608,78 @@ pub unsafe extern "C" fn ZSTDMT_rust_createCompressionJob( ) } +#[inline] +unsafe fn publish_finish_error( + publication: ZSTDMT_compressionJobFinishPublicationProjection, + error: usize, +) { + let (Some(lock), Some(unlock)) = (publication.lock, publication.unlock) else { + return; + }; + if publication.cSize.is_null() { + return; + } + unsafe { + lock(publication.callbackContext); + *publication.cSize = error; + unlock(publication.callbackContext); + } +} + +#[inline] +unsafe fn publish_finish_last_block_size( + publication: ZSTDMT_compressionJobFinishPublicationProjection, + last_block_size: usize, +) { + let (Some(lock), Some(unlock)) = (publication.lock, publication.unlock) else { + return; + }; + if publication.cSize.is_null() { + return; + } + unsafe { + lock(publication.callbackContext); + *publication.cSize = (*publication.cSize).wrapping_add(last_block_size); + unlock(publication.callbackContext); + } +} + +#[inline] +unsafe fn publish_finish_consumed( + publication: ZSTDMT_compressionJobFinishPublicationProjection, + src_size: usize, +) { + debug_assert_eq!(publication.srcSize, src_size); + if let Some(validate_consumed) = publication.validateConsumed { + unsafe { validate_consumed(publication.callbackContext, src_size) }; + } + let (Some(lock), Some(unlock)) = (publication.lock, publication.unlock) else { + return; + }; + if publication.consumed.is_null() { + return; + } + unsafe { + lock(publication.callbackContext); + *publication.consumed = src_size; + unlock(publication.callbackContext); + } +} + +#[inline] +unsafe fn signal_finish(publication: ZSTDMT_compressionJobFinishPublicationProjection) { + let (Some(lock), Some(signal), Some(unlock)) = + (publication.lock, publication.signal, publication.unlock) + else { + return; + }; + unsafe { + lock(publication.callbackContext); + signal(publication.callbackContext); + unlock(publication.callbackContext); + } +} + /// Finish a worker job with Rust-owned branch and callback ordering. /// /// Error publication happens before serial completion and resource release, @@ -2640,22 +2750,10 @@ fn compression_job_with( C: FnMut(c_uint) -> ZSTDMT_chunkProcessResult, T: FnMut(), { - let ( - Some(ensure_finished), - Some(publish_error), - Some(publish_last_block_size), - Some(release_seq), - Some(release_cctx), - Some(publish_consumed), - Some(signal), - ) = ( + let (Some(ensure_finished), Some(release_seq), Some(release_cctx)) = ( finish.ensureFinished, - finish.publishError, - finish.publishLastBlockSize, finish.releaseSeq, finish.releaseCCtx, - finish.publishConsumed, - finish.signal, ) else { return; @@ -2690,19 +2788,18 @@ fn compression_job_with( if ERR_isError(error) { last_block_size = 0; } + let publication = finish.publication; finish_compression_job_with( error, last_block_size, finish.srcSize, - |error| unsafe { publish_error(finish.callbackContext, error) }, + |error| unsafe { publish_finish_error(publication, 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) }, + |last_block_size| unsafe { publish_finish_last_block_size(publication, last_block_size) }, + |src_size| unsafe { publish_finish_consumed(publication, src_size) }, + || unsafe { signal_finish(publication) }, ); } @@ -6287,9 +6384,9 @@ mod tests { #[derive(Default)] struct MockCompressionJob { events: Vec<&'static str>, - errors: Vec, - finished: Vec, - consumed: Vec, + c_size: usize, + consumed_size: usize, + expected_src_size: usize, } struct WaitForLdmTestContext { @@ -6659,25 +6756,6 @@ mod tests { 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"); } @@ -6686,35 +6764,57 @@ mod tests { 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_lock(context: *mut c_void) { + record_compression_job_finish_event(context, "lock"); } unsafe extern "C" fn compression_job_finish_test_signal(context: *mut c_void) { record_compression_job_finish_event(context, "signal"); } + unsafe extern "C" fn compression_job_finish_test_unlock(context: *mut c_void) { + record_compression_job_finish_event(context, "unlock"); + } + + unsafe extern "C" fn compression_job_finish_test_validate( + context: *mut c_void, + src_size: usize, + ) { + let state = unsafe { &*context.cast::>() }; + assert_eq!(state.borrow().expected_src_size, src_size); + record_compression_job_finish_event(context, "validate"); + } + fn compression_job_finish_test_projection( state: &Rc>, src_size: usize, ) -> ZSTDMT_compressionJobFinishProjection { + let state_cell = Rc::as_ptr(state); + let state_value = unsafe { (*state_cell).as_ptr() }; + let (c_size, consumed) = unsafe { + (*state_value).expected_src_size = src_size; + ( + &mut (*state_value).c_size as *mut usize, + &mut (*state_value).consumed_size as *mut usize, + ) + }; + let callback_context = state_cell.cast_mut().cast(); ZSTDMT_compressionJobFinishProjection { - callbackContext: Rc::as_ptr(state).cast_mut().cast(), + callbackContext: callback_context, 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), + publication: ZSTDMT_compressionJobFinishPublicationProjection { + callbackContext: callback_context, + cSize: c_size, + consumed, + srcSize: src_size, + lock: Some(compression_job_finish_test_lock), + signal: Some(compression_job_finish_test_signal), + unlock: Some(compression_job_finish_test_unlock), + validateConsumed: Some(compression_job_finish_test_validate), + }, 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), } } @@ -6998,14 +7098,18 @@ mod tests { "ensure", "release-seq", "release-cctx", - "last-block", - "consumed", + "lock", + "unlock", + "validate", + "lock", + "unlock", + "lock", "signal", + "unlock", ] ); - assert!(state.errors.is_empty()); - assert_eq!(state.finished, vec![7]); - assert_eq!(state.consumed, vec![19]); + assert_eq!(state.c_size, 7); + assert_eq!(state.consumed_size, 19); } #[test] @@ -7041,17 +7145,21 @@ mod tests { state.events, vec![ "acquire", - "error", + "lock", + "unlock", "ensure", "release-seq", "release-cctx", - "consumed", + "validate", + "lock", + "unlock", + "lock", "signal", + "unlock", ] ); - assert_eq!(state.errors, vec![expected_error]); - assert!(state.finished.is_empty()); - assert_eq!(state.consumed, vec![11]); + assert_eq!(state.c_size, expected_error); + assert_eq!(state.consumed_size, 11); } #[test] @@ -7324,17 +7432,21 @@ mod tests { "apply", "header", "compress", - "error", + "lock", + "unlock", "ensure", "release-seq", "release-cctx", - "consumed", + "validate", + "lock", + "unlock", + "lock", "signal", + "unlock", ] ); - assert_eq!(state.errors, vec![expected_error]); - assert!(state.finished.is_empty()); - assert_eq!(state.consumed, vec![23]); + assert_eq!(state.c_size, expected_error); + assert_eq!(state.consumed_size, 23); } fn init_projection() -> ZSTDMT_initCStreamProjection {