From 97469a2e558365d63eaa7a28242bb80d90485bab Mon Sep 17 00:00:00 2001 From: ddidderr Date: Mon, 20 Jul 2026 17:20:12 +0200 Subject: [PATCH] feat(mt): move job resource acquisition policy to Rust MT worker execution used to combine CCtx acquisition, raw-sequence-store acquisition, destination-buffer allocation, frame-header publication, and the LDM resource check in one C callback. That made the resource order and short-circuit policy another C-owned implementation boundary even though the operations themselves must remain private to C. Project the scalar LDM and destination-presence inputs into Rust and let Rust own the ordering and error normalization. The C callbacks now only acquire private resources, report readiness, allocate the destination buffer, and publish the frame-header destination. The ordering remains compatible with the worker cleanup path: both pool gets happen first, CCtx failure stops before destination work, destination failure stops before publication, and the LDM sequence-store check runs after publication. Test Plan: - `ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings` -- passed - `ulimit -v 41943040; cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check` -- passed - Capped GCC and Clang syntax-only checks for `lib/compress/zstdmt_compress.c` -- passed by the worker - `git diff --cached --check` -- passed - Native build and full upstream tests are deferred to the post-batch serial verification. --- lib/compress/zstdmt_compress.c | 97 ++++++++++-- rust/src/zstdmt_compress.rs | 261 +++++++++++++++++++++++++++++++++ 2 files changed, 344 insertions(+), 14 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 886b96e9c..a57571289 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -294,9 +294,33 @@ typedef char ZSTDMT_compression_job_parameters_layout[ == 2 * sizeof(int) + sizeof(unsigned)) ? 1 : -1]; +typedef struct { + int ldmEnabled; + int hasDestination; +} ZSTDMT_RustCompressionJobResourceProjection; +typedef char ZSTDMT_compression_job_resource_projection_layout[ + (offsetof(ZSTDMT_RustCompressionJobResourceProjection, ldmEnabled) == 0 + && offsetof(ZSTDMT_RustCompressionJobResourceProjection, hasDestination) + == sizeof(int) + && sizeof(ZSTDMT_RustCompressionJobResourceProjection) + == 2 * sizeof(int)) + ? 1 : -1]; + +typedef int (*ZSTDMT_compressionJobResourceReadyFn)(void* opaque); + ZSTDMT_RustCompressionJobParameters ZSTDMT_rust_prepareCompressionJobParameters( unsigned jobID, int checksumFlag, int ldmEnable, unsigned nbWorkers); +size_t ZSTDMT_rust_compressionJobAcquireResources( + const ZSTDMT_RustCompressionJobResourceProjection* projection, + void* opaque, + ZSTDMT_compressionJobVoidFn acquireCCtx, + ZSTDMT_compressionJobVoidFn acquireRawSeqStore, + ZSTDMT_compressionJobResourceReadyFn hasCCtx, + ZSTDMT_compressionJobStepFn acquireDestinationBuffer, + ZSTDMT_compressionJobResourceReadyFn hasRawSeqStore, + ZSTDMT_compressionJobVoidFn publishDestination); + size_t ZSTDMT_rust_compressionJobBegin( const ZSTDMT_RustCompressionJobBeginProjection* projection, void* opaque, @@ -1664,28 +1688,73 @@ typedef struct { ZSTDMT_RustCompressionJobFrameHeaderState frameHeaderState; } ZSTDMT_compressionJobState; +static void ZSTDMT_compressionJobAcquireCCtx(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + state->cctx = ZSTDMT_getCCtx(state->job->cctxPool); +} + +static void ZSTDMT_compressionJobAcquireRawSeqStore(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + state->rawSeqStore = ZSTDMT_getSeq(state->job->seqPool); +} + +static int ZSTDMT_compressionJobHasCCtx(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + return state->cctx != NULL; +} + +static size_t ZSTDMT_compressionJobAcquireDestinationBuffer(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + ZSTDMT_jobDescription* const job = state->job; + state->dstBuff = ZSTDMT_getBuffer(job->bufPool); + if (state->dstBuff.start == NULL) return ERROR(memory_allocation); + job->dstBuff = state->dstBuff; + return 0; +} + +static int ZSTDMT_compressionJobHasRawSeqStore(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + return state->rawSeqStore.seq != NULL; +} + +static void ZSTDMT_compressionJobPublishDestination(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + state->frameHeaderState.dst = state->dstBuff.start; + state->frameHeaderState.dstCapacity = state->dstBuff.capacity; +} + static size_t ZSTDMT_compressionJobAcquireResources(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; - state->cctx = ZSTDMT_getCCtx(job->cctxPool); - state->rawSeqStore = ZSTDMT_getSeq(job->seqPool); state->dstBuff = job->dstBuff; DEBUGLOG(5, "ZSTDMT_compressionJob: job %u", job->jobID); - if (state->cctx == NULL) return ERROR(memory_allocation); - if (state->dstBuff.start == NULL) { - state->dstBuff = ZSTDMT_getBuffer(job->bufPool); - if (state->dstBuff.start == NULL) return ERROR(memory_allocation); - job->dstBuff = state->dstBuff; - } - state->frameHeaderState.dst = state->dstBuff.start; - state->frameHeaderState.dstCapacity = state->dstBuff.capacity; - if (state->jobParams.ldmParams.enableLdm == ZSTD_ps_enable && - state->rawSeqStore.seq == NULL) - return ERROR(memory_allocation); - return 0; + ZSTDMT_RustCompressionJobResourceProjection const projection = { + state->jobParams.ldmParams.enableLdm, + state->dstBuff.start != NULL + }; + return ZSTDMT_rust_compressionJobAcquireResources( + &projection, state, + ZSTDMT_compressionJobAcquireCCtx, + ZSTDMT_compressionJobAcquireRawSeqStore, + ZSTDMT_compressionJobHasCCtx, + ZSTDMT_compressionJobAcquireDestinationBuffer, + ZSTDMT_compressionJobHasRawSeqStore, + ZSTDMT_compressionJobPublishDestination); } static void ZSTDMT_compressionJobPrepareParameters(void* opaque) diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 24d317dbf..296ff19a1 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -87,6 +87,28 @@ const _: () = { ); }; +/// Scalar inputs for worker-resource acquisition. The worker descriptor, +/// synchronization, pools, and destination storage remain private to C; +/// Rust owns the acquisition/publication order and error short-circuiting. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_compressionJobResourceProjection { + pub ldmEnabled: c_int, + pub hasDestination: c_int, +} + +const _: () = { + assert!(offset_of!(ZSTDMT_compressionJobResourceProjection, ldmEnabled) == 0); + assert!( + offset_of!(ZSTDMT_compressionJobResourceProjection, hasDestination) == size_of::() + ); + assert!(size_of::() == 2 * size_of::()); +}; + +pub type ZSTDMT_compressionJobResourceVoidFn = unsafe extern "C" fn(*mut c_void); +pub type ZSTDMT_compressionJobResourceStepFn = unsafe extern "C" fn(*mut c_void) -> usize; +pub type ZSTDMT_compressionJobResourceReadyFn = unsafe extern "C" fn(*mut c_void) -> c_int; + #[inline] fn prepare_compression_job_parameters( job_id: c_uint, @@ -117,6 +139,100 @@ pub extern "C" fn ZSTDMT_rust_prepareCompressionJobParameters( prepare_compression_job_parameters(jobID, checksumFlag, ldmEnable, nbWorkers) } +#[inline] +fn compression_job_acquire_resources_with( + projection: ZSTDMT_compressionJobResourceProjection, + mut acquire_cctx: A, + mut acquire_raw_seq_store: S, + mut has_cctx: C, + mut acquire_destination_buffer: B, + mut has_raw_seq_store: R, + mut publish_destination: P, +) -> usize +where + A: FnMut(), + S: FnMut(), + C: FnMut() -> bool, + B: FnMut() -> usize, + R: FnMut() -> bool, + P: FnMut(), +{ + /* Keep both pool gets before checking the CCtx, matching the C worker + * cleanup path that releases every resource acquired for the job. */ + acquire_cctx(); + acquire_raw_seq_store(); + if !has_cctx() { + return ERROR(ZstdErrorCode::MemoryAllocation); + } + + if projection.hasDestination == 0 { + let error = acquire_destination_buffer(); + if ERR_isError(error) { + return error; + } + } + publish_destination(); + + if projection.ldmEnabled != 0 && !has_raw_seq_store() { + return ERROR(ZstdErrorCode::MemoryAllocation); + } + 0 +} + +/// Own worker-resource acquisition policy while C retains pool, descriptor, +/// synchronization, and allocator operations behind callbacks. The order is +/// intentionally: acquire both worker resources, validate the CCtx, acquire a +/// destination only when absent, publish its scalar destination projection, +/// then validate the LDM sequence store. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_compressionJobAcquireResources( + projection: *const ZSTDMT_compressionJobResourceProjection, + opaque: *mut c_void, + acquire_cctx: Option, + acquire_raw_seq_store: Option, + has_cctx: Option, + acquire_destination_buffer: Option, + has_raw_seq_store: Option, + publish_destination: Option, +) -> usize { + let Some(projection) = (unsafe { projection.as_ref() }).copied() else { + return ERROR(ZstdErrorCode::Generic); + }; + let ( + Some(acquire_cctx), + Some(acquire_raw_seq_store), + Some(has_cctx), + Some(acquire_destination_buffer), + Some(has_raw_seq_store), + Some(publish_destination), + ) = ( + acquire_cctx, + acquire_raw_seq_store, + has_cctx, + acquire_destination_buffer, + has_raw_seq_store, + publish_destination, + ) + else { + return ERROR(ZstdErrorCode::Generic); + }; + if opaque.is_null() { + return ERROR(ZstdErrorCode::Generic); + } + + unsafe { + compression_job_acquire_resources_with( + projection, + || acquire_cctx(opaque), + || acquire_raw_seq_store(opaque), + || has_cctx(opaque) != 0, + || acquire_destination_buffer(opaque), + || has_raw_seq_store(opaque) != 0, + || publish_destination(opaque), + ) + } +} + /// Scalar job state used by the Rust compression-job scheduler. /// /// The job descriptor, pools, synchronization, and codec state remain @@ -5510,6 +5626,151 @@ mod tests { assert_eq!(&output[result..], &[0xa5; 12]); } + #[test] + fn compression_job_resources_order_pool_gets_destination_and_ldm_check() { + let events = Rc::new(RefCell::new(Vec::<&'static str>::new())); + let acquire_cctx_events = Rc::clone(&events); + let acquire_seq_events = Rc::clone(&events); + let cctx_ready_events = Rc::clone(&events); + let destination_events = Rc::clone(&events); + let seq_ready_events = Rc::clone(&events); + let publish_events = Rc::clone(&events); + + let result = compression_job_acquire_resources_with( + ZSTDMT_compressionJobResourceProjection { + ldmEnabled: 1, + hasDestination: 0, + }, + move || acquire_cctx_events.borrow_mut().push("cctx"), + move || acquire_seq_events.borrow_mut().push("seq"), + move || { + cctx_ready_events.borrow_mut().push("cctx-ready"); + true + }, + move || { + destination_events.borrow_mut().push("destination"); + 0 + }, + move || { + seq_ready_events.borrow_mut().push("seq-ready"); + true + }, + move || publish_events.borrow_mut().push("publish"), + ); + + assert_eq!(result, 0); + assert_eq!( + &*events.borrow(), + &[ + "cctx", + "seq", + "cctx-ready", + "destination", + "publish", + "seq-ready", + ] + ); + } + + #[test] + fn compression_job_resources_reject_missing_cctx_after_both_pool_gets() { + let events = Rc::new(RefCell::new(Vec::<&'static str>::new())); + let acquire_cctx_events = Rc::clone(&events); + let acquire_seq_events = Rc::clone(&events); + let cctx_ready_events = Rc::clone(&events); + let expected_error = ERROR(ZstdErrorCode::MemoryAllocation); + + let result = compression_job_acquire_resources_with( + ZSTDMT_compressionJobResourceProjection { + ldmEnabled: 1, + hasDestination: 0, + }, + move || acquire_cctx_events.borrow_mut().push("cctx"), + move || acquire_seq_events.borrow_mut().push("seq"), + move || { + cctx_ready_events.borrow_mut().push("cctx-ready"); + false + }, + || panic!("CCtx failure must stop before destination acquisition"), + || panic!("CCtx failure must stop before LDM validation"), + || panic!("CCtx failure must stop before destination publication"), + ); + + assert_eq!(result, expected_error); + assert_eq!(&*events.borrow(), &["cctx", "seq", "cctx-ready"]); + } + + #[test] + fn compression_job_resources_stops_before_publication_on_destination_error() { + let events = Rc::new(RefCell::new(Vec::<&'static str>::new())); + let acquire_cctx_events = Rc::clone(&events); + let acquire_seq_events = Rc::clone(&events); + let cctx_ready_events = Rc::clone(&events); + let destination_events = Rc::clone(&events); + let expected_error = ERROR(ZstdErrorCode::MemoryAllocation); + + let result = compression_job_acquire_resources_with( + ZSTDMT_compressionJobResourceProjection { + ldmEnabled: 1, + hasDestination: 0, + }, + move || acquire_cctx_events.borrow_mut().push("cctx"), + move || acquire_seq_events.borrow_mut().push("seq"), + move || { + cctx_ready_events.borrow_mut().push("cctx-ready"); + true + }, + move || { + destination_events.borrow_mut().push("destination"); + expected_error + }, + || panic!("destination failure must stop before LDM validation"), + || panic!("destination failure must stop before publication"), + ); + + assert_eq!(result, expected_error); + assert_eq!( + &*events.borrow(), + &["cctx", "seq", "cctx-ready", "destination"] + ); + } + + #[test] + fn compression_job_resources_publishes_before_missing_ldm_store_error() { + let events = Rc::new(RefCell::new(Vec::<&'static str>::new())); + let acquire_cctx_events = Rc::clone(&events); + let acquire_seq_events = Rc::clone(&events); + let cctx_ready_events = Rc::clone(&events); + let seq_ready_events = Rc::clone(&events); + let publish_events = Rc::clone(&events); + let expected_error = ERROR(ZstdErrorCode::MemoryAllocation); + + let result = compression_job_acquire_resources_with( + ZSTDMT_compressionJobResourceProjection { + ldmEnabled: 1, + hasDestination: 1, + }, + move || acquire_cctx_events.borrow_mut().push("cctx"), + move || acquire_seq_events.borrow_mut().push("seq"), + move || { + cctx_ready_events.borrow_mut().push("cctx-ready"); + true + }, + || panic!("an existing destination must skip buffer acquisition"), + move || { + seq_ready_events.borrow_mut().push("seq-ready"); + false + }, + move || publish_events.borrow_mut().push("publish"), + ); + + assert_eq!(result, expected_error); + assert_eq!( + &*events.borrow(), + &["cctx", "seq", "cctx-ready", "publish", "seq-ready"] + ); + } + #[test] fn compression_job_runs_first_job_stages_and_reports_final_block() { let state = Rc::new(RefCell::new(MockCompressionJob::default()));