From 484ffa4d93fb9652bc2b879fbd0bd75a929f7202 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Tue, 21 Jul 2026 20:42:30 +0200 Subject: [PATCH] feat(mt): move external sequence gate to Rust The MT worker previously kept the external-sequence non-empty gate and LDM invariant assertion in C, bundled with the private CCtx operation. Project the CCtx pointer and raw-sequence store through a layout-checked seam so Rust owns the gate and the apply point while C retains only the opaque ZSTD_referenceExternalSequences() leaf. Preserve the callback ordering after worker CCtx initialization and keep empty stores as a no-op. Test Plan: - git diff --cached --check -- passed before commit. - Static rg inspection confirmed all projection initializers include the new sequence-state pointer and no old serialState_applySequences symbol remains. - Cargo, make, compiler checks, and full tests were not run per worker scope. --- lib/compress/zstdmt_compress.c | 78 ++++++++++++------- rust/src/zstdmt_compress.rs | 136 ++++++++++++++++++++++++++++++++- 2 files changed, 182 insertions(+), 32 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 8d1de4e07..b8be2e421 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -220,18 +220,24 @@ typedef char ZSTDMT_compression_job_frame_header_state_layout[ == (sizeof(void*) == 8 ? 72 : 48)) ? 1 : -1]; +typedef struct ZSTDMT_RustCompressionJobSequenceState_s + ZSTDMT_RustCompressionJobSequenceState; + typedef struct { unsigned firstJob; unsigned lastJob; const ZSTDMT_RustCompressionJobFrameHeaderState* frameHeaderState; + const ZSTDMT_RustCompressionJobSequenceState* sequenceState; } ZSTDMT_RustCompressionJobProjection; typedef char ZSTDMT_compression_job_projection_layout[ (offsetof(ZSTDMT_RustCompressionJobProjection, firstJob) == 0 && offsetof(ZSTDMT_RustCompressionJobProjection, lastJob) == sizeof(unsigned) && offsetof(ZSTDMT_RustCompressionJobProjection, frameHeaderState) == 2 * sizeof(unsigned) + && offsetof(ZSTDMT_RustCompressionJobProjection, sequenceState) + == 2 * sizeof(unsigned) + sizeof(void*) && sizeof(ZSTDMT_RustCompressionJobProjection) - == 2 * sizeof(unsigned) + sizeof(void*)) + == 2 * sizeof(unsigned) + 2 * sizeof(void*)) ? 1 : -1]; typedef struct { @@ -260,6 +266,8 @@ typedef ZSTDMT_chunkProcessResult (*ZSTDMT_compressionJobCompressFn)( void* opaque, unsigned lastJob); typedef void (*ZSTDMT_compressionJobSizeFn)(void* opaque, size_t size); typedef size_t (*ZSTDMT_compressionJobSetParameterFn)(void* opaque, int value); +typedef void (*ZSTDMT_compressionJobApplySequencesFn)( + void* opaque, void* cctx, void* sequences, size_t nbSequences); typedef struct { void* callbackContext; @@ -350,7 +358,7 @@ void ZSTDMT_rust_compressionJob( ZSTDMT_compressionJobVoidFn prepareParameters, ZSTDMT_compressionJobVoidFn generateSequences, ZSTDMT_compressionJobStepFn beginJob, - ZSTDMT_compressionJobVoidFn applySequences, + ZSTDMT_compressionJobApplySequencesFn applySequences, ZSTDMT_compressionJobCompressFn compressJob, ZSTDMT_compressionJobVoidFn traceJob); @@ -599,6 +607,21 @@ typedef char ZSTDMT_rust_raw_seq_store_layout[ && offsetof(ZSTDMT_RustRawSeqStore, capacity) == offsetof(RawSeqStore_t, capacity)) ? 1 : -1]; +struct ZSTDMT_RustCompressionJobSequenceState_s { + ZSTD_CCtx** cctx; + ZSTDMT_RustRawSeqStore* rawSeqStore; + int ldmEnabled; +}; +typedef char ZSTDMT_compression_job_sequence_state_layout[ + (offsetof(ZSTDMT_RustCompressionJobSequenceState, cctx) == 0 + && offsetof(ZSTDMT_RustCompressionJobSequenceState, rawSeqStore) + == sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobSequenceState, ldmEnabled) + == 2 * sizeof(void*) + && sizeof(ZSTDMT_RustCompressionJobSequenceState) + == (sizeof(void*) == 8 ? 24 : 12)) + ? 1 : -1]; + ZSTDMT_RustRawSeqStore ZSTDMT_rust_bufferToSeq(ZSTDMT_RustBuffer buffer); ZSTDMT_RustBuffer ZSTDMT_rust_seqToBuffer(ZSTDMT_RustRawSeqStore seq); typedef int (*ZSTDMT_serialWaitForTurnFn)(void* opaque, unsigned jobID); @@ -1840,17 +1863,17 @@ static void ZSTDMT_serialState_onSkip(void* opaque, unsigned jobID, size_t cSize DEBUGLOG(5, "Skipping past job %u because of error", jobID); } -static void -ZSTDMT_serialState_applySequences(const SerialState* serialState, /* just for an assert() check */ - ZSTD_CCtx* jobCCtx, - const RawSeqStore_t* seqStore) +/* Rust owns the non-empty/LDM gate and the point at which this callback is + * invoked. Keep only the private codec operation on the C side. */ +static void ZSTDMT_compressionJobReferenceSequences( + void* opaque, void* cctx, void* sequences, size_t nbSequences) { - if (seqStore->size > 0) { - DEBUGLOG(5, "ZSTDMT_serialState_applySequences: uploading %u external sequences", (unsigned)seqStore->size); - assert(serialState->params.ldmParams.enableLdm == ZSTD_ps_enable); (void)serialState; - assert(jobCCtx); - ZSTD_referenceExternalSequences(jobCCtx, seqStore->seq, seqStore->size); - } + (void)opaque; + assert(cctx != NULL); + DEBUGLOG(5, "ZSTDMT_compressionJob: uploading %u external sequences", + (unsigned)nbSequences); + ZSTD_referenceExternalSequences((ZSTD_CCtx*)cctx, + (rawSeq*)sequences, nbSequences); } static void ZSTDMT_serialState_ensureFinished(SerialState* serialState, @@ -2120,17 +2143,6 @@ static void ZSTDMT_compressionJobPublishFrameHeader(void* opaque) ? NULL : state->cctx->blockState.prevCBlock->rep; } -static void ZSTDMT_compressionJobApplySequences(void* opaque) -{ - ZSTDMT_compressionJobState* const state = - (ZSTDMT_compressionJobState*)opaque; - ZSTDMT_jobDescription* const job = state->job; - - /* External Sequences can only be applied after CCtx initialization. */ - ZSTDMT_serialState_applySequences(job->serial, state->cctx, - &state->rawSeqStore); -} - static ZSTDMT_chunkProcessResult ZSTDMT_compressionJobCompress( void* opaque, unsigned lastJob) { @@ -2247,17 +2259,25 @@ static void ZSTDMT_compressionJob(void* jobDescription) { ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription; ZSTDMT_compressionJobState state; + ZSTDMT_RustCompressionJobSequenceState sequenceState; ZSTDMT_RustCompressionJobFinishProjection finishProjection; - ZSTDMT_RustCompressionJobProjection const projection = { - job->firstJob, - job->lastJob, - &state.frameHeaderState - }; + ZSTDMT_RustCompressionJobProjection projection; ZSTD_memset(&state, 0, sizeof(state)); state.job = job; state.jobParams = job->params; /* do not modify job->params ! copy it, modify the copy */ state.dstBuff = job->dstBuff; + sequenceState = (ZSTDMT_RustCompressionJobSequenceState){ + &state.cctx, + (ZSTDMT_RustRawSeqStore*)&state.rawSeqStore, + job->serial->params.ldmParams.enableLdm + }; + projection = (ZSTDMT_RustCompressionJobProjection){ + job->firstJob, + job->lastJob, + &state.frameHeaderState, + &sequenceState + }; finishProjection = (ZSTDMT_RustCompressionJobFinishProjection){ &state, job->src.size, @@ -2276,7 +2296,7 @@ static void ZSTDMT_compressionJob(void* jobDescription) ZSTDMT_compressionJobPrepareParameters, ZSTDMT_compressionJobGenerateSequences, ZSTDMT_compressionJobBegin, - ZSTDMT_compressionJobApplySequences, + ZSTDMT_compressionJobReferenceSequences, ZSTDMT_compressionJobCompress, ZSTDMT_compressionJobTrace); } diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 9e76263f3..5da5eccf8 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -365,6 +365,18 @@ pub struct ZSTDMT_compressionJobProjection { pub firstJob: c_uint, pub lastJob: c_uint, pub frameHeaderState: *const ZSTDMT_compressionJobFrameHeaderState, + pub sequenceState: *const ZSTDMT_compressionJobSequenceState, +} + +/// Indirection into the worker-owned state used by the external-sequence +/// gate. Rust sees only pointers to the opaque CCtx and the already-projected +/// raw-sequence store; the private worker-state layout remains in C. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_compressionJobSequenceState { + pub cctx: *mut *mut c_void, + pub rawSeqStore: *const ZstdMtRawSeqStore, + pub ldmEnabled: c_int, } /// Scalar projection for the non-first MT job frame header. The destination, @@ -393,9 +405,24 @@ const _: () = { offset_of!(ZSTDMT_compressionJobProjection, frameHeaderState) == size_of::<[c_uint; 2]>() ); assert!( - size_of::() + offset_of!(ZSTDMT_compressionJobProjection, sequenceState) == size_of::<[c_uint; 2]>() + size_of::() ); + assert!( + size_of::() + == size_of::<[c_uint; 2]>() + 2 * size_of::() + ); + assert!(offset_of!(ZSTDMT_compressionJobSequenceState, cctx) == 0); + assert!( + offset_of!(ZSTDMT_compressionJobSequenceState, rawSeqStore) == size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobSequenceState, ldmEnabled) == 2 * size_of::() + ); + assert!( + size_of::() + == if size_of::() == 8 { 24 } else { 12 } + ); assert!(offset_of!(ZSTDMT_compressionJobFrameHeaderState, dst) == 0); assert!(offset_of!(ZSTDMT_compressionJobFrameHeaderState, dst_capacity) == size_of::()); assert!(offset_of!(ZSTDMT_compressionJobFrameHeaderState, stage) == 2 * size_of::()); @@ -504,6 +531,57 @@ pub type ZSTDMT_compressionJobCompressFn = unsafe extern "C" fn(*mut c_void, c_uint) -> ZSTDMT_chunkProcessResult; pub type ZSTDMT_compressionJobSizeFn = unsafe extern "C" fn(*mut c_void, usize); pub type ZSTDMT_compressionJobSetParameterFn = unsafe extern "C" fn(*mut c_void, c_int) -> usize; +pub type ZSTDMT_compressionJobApplySequencesFn = + unsafe extern "C" fn(*mut c_void, *mut c_void, *mut c_void, usize); + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +struct ZSTDMT_compressionJobSequenceProjection { + cctx: *mut c_void, + sequences: *mut c_void, + nbSequences: usize, + ldmEnabled: c_int, +} + +#[inline] +unsafe fn compression_job_sequence_projection( + state: *const ZSTDMT_compressionJobSequenceState, +) -> Option { + let state = unsafe { state.as_ref() }?; + if state.cctx.is_null() || state.rawSeqStore.is_null() { + return None; + } + + let cctx = unsafe { *state.cctx }; + let raw_seq_store = unsafe { &*state.rawSeqStore }; + Some(ZSTDMT_compressionJobSequenceProjection { + cctx, + sequences: raw_seq_store.seq.cast(), + nbSequences: raw_seq_store.size, + ldmEnabled: state.ldmEnabled, + }) +} + +/// Apply the worker's external-sequence policy after CCtx initialization. +/// Rust owns the non-empty gate and LDM invariant; C retains only the opaque +/// `ZSTD_referenceExternalSequences()` leaf behind the callback. +#[inline] +fn apply_compression_job_sequences_with( + projection: Option, + mut reference_sequences: R, +) where + R: FnMut(*mut c_void, *mut c_void, usize), +{ + let Some(projection) = projection else { + return; + }; + if projection.nbSequences == 0 { + return; + } + + debug_assert_eq!(projection.ldmEnabled, ZSTD_PS_ENABLE); + debug_assert!(!projection.cctx.is_null()); + reference_sequences(projection.cctx, projection.sequences, projection.nbSequences); +} /// Scalar view used to choose the private worker-context initialization path. /// C retains the cdict, parameter, and CCtx objects; Rust owns the branch and @@ -2779,7 +2857,7 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( prepareParameters: Option, generateSequences: Option, beginJob: Option, - applySequences: Option, + applySequences: Option, compressJob: Option, traceJob: Option, ) { @@ -2810,6 +2888,8 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( return; }; + let sequence_state = projection.sequenceState; + compression_job_with( projection, finish, @@ -2817,7 +2897,13 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( || unsafe { prepare_parameters(opaque) }, || unsafe { generate_sequences(opaque) }, || unsafe { begin_job(opaque) }, - || unsafe { apply_sequences(opaque) }, + || { + let sequence_projection = + unsafe { compression_job_sequence_projection(sequence_state) }; + apply_compression_job_sequences_with(sequence_projection, |cctx, sequences, count| { + unsafe { apply_sequences(opaque, cctx, sequences, count) } + }); + }, || unsafe { ZSTDMT_rust_writeFrameHeader(projection.frameHeaderState) }, |last_job| unsafe { compress_job(opaque, last_job) }, || unsafe { trace_job(opaque) }, @@ -6591,6 +6677,47 @@ mod tests { assert_eq!(&output[result..], &[0xa5; 12]); } + #[test] + fn compression_job_sequence_gate_skips_empty_store_without_ldm_assertion() { + let mut calls = Vec::new(); + let projection = ZSTDMT_compressionJobSequenceProjection { + cctx: 0x10usize as *mut c_void, + sequences: ptr::null_mut(), + nbSequences: 0, + ldmEnabled: 0, + }; + + apply_compression_job_sequences_with(Some(projection), |cctx, sequences, count| { + calls.push((cctx, sequences, count)); + }); + + assert!(calls.is_empty()); + } + + #[test] + fn compression_job_sequence_gate_forwards_nonempty_store_after_ldm_check() { + let mut calls = Vec::new(); + let projection = ZSTDMT_compressionJobSequenceProjection { + cctx: 0x10usize as *mut c_void, + sequences: 0x20usize as *mut c_void, + nbSequences: 3, + ldmEnabled: ZSTD_PS_ENABLE, + }; + + apply_compression_job_sequences_with(Some(projection), |cctx, sequences, count| { + calls.push((cctx, sequences, count)); + }); + + assert_eq!( + calls, + vec![( + 0x10usize as *mut c_void, + 0x20usize as *mut c_void, + 3 + )] + ); + } + #[test] fn compression_job_resources_order_pool_gets_destination_and_ldm_check() { let events = Rc::new(RefCell::new(Vec::<&'static str>::new())); @@ -6753,6 +6880,7 @@ mod tests { firstJob: 1, lastJob: 1, frameHeaderState: ptr::null(), + sequenceState: ptr::null(), }, finish, move || { @@ -6814,6 +6942,7 @@ mod tests { firstJob: 1, lastJob: 0, frameHeaderState: ptr::null(), + sequenceState: ptr::null(), }, finish, move || { @@ -7077,6 +7206,7 @@ mod tests { firstJob: 0, lastJob: 1, frameHeaderState: ptr::null(), + sequenceState: ptr::null(), }, finish, move || {