From 69a2d3969d2a4c332003f084d5a18a8586b8a27c Mon Sep 17 00:00:00 2001 From: ddidderr Date: Mon, 20 Jul 2026 16:48:10 +0200 Subject: [PATCH] feat(mt): move compression-job begin policy to Rust The worker context previously selected the cdict or raw-prefix initialization path, applied non-first-job parameter updates, and published the frame-header projection directly in C. That left the branch and failure order intertwined with private CCtx and parameter layouts. Project only the first-job and cdict flags into Rust. Rust now validates cdict placement, selects the initialization path, stops on force-window or prefix-policy errors, and publishes the header projection only after successful initialization. C callbacks retain the private cdict, CCtx, parameter mutation, and frame-header field operations, including the original pledged-size and force-window behavior. Focused tests cover cdict ordering, non-first parameter ordering, parameter and initialization failures, header-publication suppression, and invalid cdict placement. Test Plan: - `rustfmt +nightly --edition 2021 --check rust/src/zstdmt_compress.rs` -- passed. - Capped GCC syntax-only check for `zstdmt_compress.c` -- passed. - Capped Clang syntax-only check for `zstdmt_compress.c` -- passed. - `git diff --cached --check` -- passed. - Cargo, native builds, fuzzers, and large tests were not run per assignment. --- lib/compress/zstdmt_compress.c | 112 ++++++++++++---- rust/src/zstdmt_compress.rs | 238 +++++++++++++++++++++++++++++++++ 2 files changed, 326 insertions(+), 24 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index efbff5159..886b96e9c 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -222,6 +222,18 @@ typedef char ZSTDMT_compression_job_projection_layout[ == 2 * sizeof(unsigned) + sizeof(void*)) ? 1 : -1]; +typedef struct { + unsigned firstJob; + unsigned hasCDict; +} ZSTDMT_RustCompressionJobBeginProjection; +typedef char ZSTDMT_compression_job_begin_projection_layout[ + (offsetof(ZSTDMT_RustCompressionJobBeginProjection, firstJob) == 0 + && offsetof(ZSTDMT_RustCompressionJobBeginProjection, hasCDict) + == sizeof(unsigned) + && sizeof(ZSTDMT_RustCompressionJobBeginProjection) + == 2 * sizeof(unsigned)) + ? 1 : -1]; + typedef struct { int status; size_t toFlush; @@ -235,6 +247,7 @@ typedef void (*ZSTDMT_compressionJobVoidFn)(void* opaque); 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 struct { void* callbackContext; @@ -284,6 +297,15 @@ typedef char ZSTDMT_compression_job_parameters_layout[ ZSTDMT_RustCompressionJobParameters ZSTDMT_rust_prepareCompressionJobParameters( unsigned jobID, int checksumFlag, int ldmEnable, unsigned nbWorkers); +size_t ZSTDMT_rust_compressionJobBegin( + const ZSTDMT_RustCompressionJobBeginProjection* projection, + void* opaque, + ZSTDMT_compressionJobStepFn beginWithCDict, + ZSTDMT_compressionJobSetParameterFn setForceMaxWindow, + ZSTDMT_compressionJobStepFn setDeterministicRefPrefix, + ZSTDMT_compressionJobStepFn beginWithPrefix, + ZSTDMT_compressionJobVoidFn publishFrameHeader); + void ZSTDMT_rust_compressionJob( const ZSTDMT_RustCompressionJobProjection* projection, const ZSTDMT_RustCompressionJobFinishProjection* finishProjection, @@ -1702,36 +1724,79 @@ static void ZSTDMT_compressionJobGenerateSequences(void* opaque) ZSTDMT_serialState_advance); } +static size_t ZSTDMT_compressionJobBeginWithCDict(void* opaque); +static size_t ZSTDMT_compressionJobSetForceMaxWindow(void* opaque, int value); +static size_t ZSTDMT_compressionJobSetDeterministicRefPrefix(void* opaque); +static size_t ZSTDMT_compressionJobBeginWithPrefix(void* opaque); +static void ZSTDMT_compressionJobPublishFrameHeader(void* opaque); + static size_t ZSTDMT_compressionJobBegin(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; - size_t initError; - if (job->cdict) { - initError = ZSTD_compressBegin_advanced_internal( - state->cctx, NULL, 0, ZSTD_dct_auto, ZSTD_dtlm_fast, job->cdict, - &state->jobParams, job->fullFrameSize); - assert(job->firstJob); /* only allowed for first job */ - } else { - U64 const pledgedSrcSize = job->firstJob ? job->fullFrameSize : job->src.size; - size_t const forceWindowError = ZSTD_CCtxParams_setParameter( - &state->jobParams, ZSTD_c_forceMaxWindow, !job->firstJob); - if (ZSTD_isError(forceWindowError)) return forceWindowError; - if (!job->firstJob) { - size_t const err = ZSTD_CCtxParams_setParameter( - &state->jobParams, ZSTD_c_deterministicRefPrefix, 0); - if (ZSTD_isError(err)) return err; - } - DEBUGLOG(6, "ZSTDMT_compressionJob: job %u: loading prefix of size %zu", job->jobID, job->prefix.size); - initError = ZSTD_compressBegin_advanced_internal( - state->cctx, job->prefix.start, job->prefix.size, - ZSTD_dct_rawContent, ZSTD_dtlm_fast, NULL, /*cdict*/ - &state->jobParams, pledgedSrcSize); - } + ZSTDMT_RustCompressionJobBeginProjection const projection = { + job->firstJob, + (unsigned)(job->cdict != NULL) + }; + return ZSTDMT_rust_compressionJobBegin( + &projection, state, + ZSTDMT_compressionJobBeginWithCDict, + ZSTDMT_compressionJobSetForceMaxWindow, + ZSTDMT_compressionJobSetDeterministicRefPrefix, + ZSTDMT_compressionJobBeginWithPrefix, + ZSTDMT_compressionJobPublishFrameHeader); +} + +static size_t ZSTDMT_compressionJobBeginWithCDict(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + ZSTDMT_jobDescription* const job = state->job; + + return ZSTD_compressBegin_advanced_internal( + state->cctx, NULL, 0, ZSTD_dct_auto, ZSTD_dtlm_fast, job->cdict, + &state->jobParams, job->fullFrameSize); +} + +static size_t ZSTDMT_compressionJobSetForceMaxWindow(void* opaque, int value) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + + return ZSTD_CCtxParams_setParameter( + &state->jobParams, ZSTD_c_forceMaxWindow, value); +} + +static size_t ZSTDMT_compressionJobSetDeterministicRefPrefix(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + + return ZSTD_CCtxParams_setParameter( + &state->jobParams, ZSTD_c_deterministicRefPrefix, 0); +} + +static size_t ZSTDMT_compressionJobBeginWithPrefix(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + ZSTDMT_jobDescription* const job = state->job; + U64 const pledgedSrcSize = job->firstJob ? job->fullFrameSize : job->src.size; + + DEBUGLOG(6, "ZSTDMT_compressionJob: job %u: loading prefix of size %zu", job->jobID, job->prefix.size); + return ZSTD_compressBegin_advanced_internal( + state->cctx, job->prefix.start, job->prefix.size, + ZSTD_dct_rawContent, ZSTD_dtlm_fast, NULL, /*cdict*/ + &state->jobParams, pledgedSrcSize); +} + +static void ZSTDMT_compressionJobPublishFrameHeader(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; - if (ZSTD_isError(initError)) return initError; state->frameHeaderState.stage = (int*)&state->cctx->stage; state->frameHeaderState.noDictIDFlag = state->cctx->appliedParams.fParams.noDictIDFlag; @@ -1747,7 +1812,6 @@ static size_t ZSTDMT_compressionJobBegin(void* opaque) state->frameHeaderState.dictID = state->cctx->dictID; state->frameHeaderState.repCodes = state->cctx->blockState.prevCBlock == NULL ? NULL : state->cctx->blockState.prevCBlock->rep; - return initError; } static void ZSTDMT_compressionJobApplySequences(void* opaque) diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 64de5cd6d..ae298f019 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -266,6 +266,107 @@ 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_compressionJobSizeFn = unsafe extern "C" fn(*mut c_void, usize); +pub type ZSTDMT_compressionJobSetParameterFn = unsafe extern "C" fn(*mut c_void, c_int) -> usize; + +/// Scalar view used to choose the private worker-context initialization path. +/// C retains the cdict, parameter, and CCtx objects; Rust owns the branch and +/// the rule that the frame-header projection is published only after init. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_compressionJobBeginProjection { + pub firstJob: c_uint, + pub hasCDict: c_uint, +} + +const _: () = { + assert!(offset_of!(ZSTDMT_compressionJobBeginProjection, firstJob) == 0); + assert!(offset_of!(ZSTDMT_compressionJobBeginProjection, hasCDict) == size_of::()); + assert!(size_of::() == 2 * size_of::()); +}; + +#[inline] +fn begin_compression_job_with( + projection: ZSTDMT_compressionJobBeginProjection, + mut begin_with_cdict: B, + mut set_force_max_window: F, + mut set_deterministic_ref_prefix: D, + mut begin_with_prefix: P, + mut publish_frame_header: H, +) -> usize +where + B: FnMut() -> usize, + F: FnMut(c_int) -> usize, + D: FnMut() -> usize, + P: FnMut() -> usize, + H: FnMut(), +{ + let init_error = if projection.hasCDict != 0 { + if projection.firstJob == 0 { + return ERROR(ZstdErrorCode::StageWrong); + } + begin_with_cdict() + } else { + let force_window_error = set_force_max_window((projection.firstJob == 0) as c_int); + if ERR_isError(force_window_error) { + return force_window_error; + } + if projection.firstJob == 0 { + let error = set_deterministic_ref_prefix(); + if ERR_isError(error) { + return error; + } + } + begin_with_prefix() + }; + + if ERR_isError(init_error) { + return init_error; + } + publish_frame_header(); + init_error +} + +/// Own the worker-context initialization branch and failure/publication order +/// while C keeps all private codec operations behind callbacks. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_compressionJobBegin( + projection: *const ZSTDMT_compressionJobBeginProjection, + opaque: *mut c_void, + beginWithCDict: Option, + setForceMaxWindow: Option, + setDeterministicRefPrefix: Option, + beginWithPrefix: Option, + publishFrameHeader: Option, +) -> usize { + let Some(projection) = (unsafe { projection.as_ref() }).copied() else { + return ERROR(ZstdErrorCode::Generic); + }; + let ( + Some(begin_with_cdict), + Some(set_force_max_window), + Some(set_deterministic_ref_prefix), + Some(begin_with_prefix), + Some(publish_frame_header), + ) = ( + beginWithCDict, + setForceMaxWindow, + setDeterministicRefPrefix, + beginWithPrefix, + publishFrameHeader, + ) + else { + return ERROR(ZstdErrorCode::Generic); + }; + + begin_compression_job_with( + *projection, + || unsafe { begin_with_cdict(opaque) }, + |value| unsafe { set_force_max_window(opaque, value) }, + || unsafe { set_deterministic_ref_prefix(opaque) }, + || unsafe { begin_with_prefix(opaque) }, + || unsafe { publish_frame_header(opaque) }, + ) +} /// C-owned leaves for the worker-job finish path. /// @@ -7312,6 +7413,143 @@ mod tests { assert_eq!(result.clearSource, 0); } + #[test] + fn compression_job_begin_uses_cdict_and_publishes_header_after_init() { + let events = Rc::new(RefCell::new(Vec::<&'static str>::new())); + let projection = ZSTDMT_compressionJobBeginProjection { + firstJob: 1, + hasCDict: 1, + }; + + let cdict_events = Rc::clone(&events); + let header_events = Rc::clone(&events); + let result = begin_compression_job_with( + projection, + move || { + cdict_events.borrow_mut().push("cdict"); + 0 + }, + |_| panic!("cdict jobs must not update force-window parameters"), + || panic!("cdict jobs must not update deterministic-prefix parameters"), + || panic!("cdict jobs must not initialize from a raw prefix"), + move || header_events.borrow_mut().push("header"), + ); + + assert_eq!(result, 0); + assert_eq!(&*events.borrow(), &["cdict", "header"]); + } + + #[test] + fn compression_job_begin_orders_nonfirst_parameter_policy_before_prefix_init() { + let events = Rc::new(RefCell::new(Vec::<&'static str>::new())); + let projection = ZSTDMT_compressionJobBeginProjection { + firstJob: 0, + hasCDict: 0, + }; + + let force_events = Rc::clone(&events); + let deterministic_events = Rc::clone(&events); + let prefix_events = Rc::clone(&events); + let header_events = Rc::clone(&events); + let result = begin_compression_job_with( + projection, + || panic!("raw-prefix jobs must not use a cdict"), + move |value| { + assert_eq!(value, 1); + force_events.borrow_mut().push("force-window"); + 0 + }, + move || { + deterministic_events + .borrow_mut() + .push("deterministic-prefix"); + 0 + }, + move || { + prefix_events.borrow_mut().push("prefix"); + 0 + }, + move || header_events.borrow_mut().push("header"), + ); + + assert_eq!(result, 0); + assert_eq!( + &*events.borrow(), + &["force-window", "deterministic-prefix", "prefix", "header"] + ); + } + + #[test] + fn compression_job_begin_stops_after_force_window_error() { + let events = Rc::new(RefCell::new(Vec::<&'static str>::new())); + let expected_error = ERROR(ZstdErrorCode::MemoryAllocation); + let projection = ZSTDMT_compressionJobBeginProjection { + firstJob: 0, + hasCDict: 0, + }; + let force_events = Rc::clone(&events); + + let result = begin_compression_job_with( + projection, + || panic!("parameter failure must stop before cdict initialization"), + move |_| { + force_events.borrow_mut().push("force-window"); + expected_error + }, + || panic!("parameter failure must stop before deterministic-prefix update"), + || panic!("parameter failure must stop before prefix initialization"), + || panic!("parameter failure must stop before header publication"), + ); + + assert_eq!(result, expected_error); + assert_eq!(&*events.borrow(), &["force-window"]); + } + + #[test] + fn compression_job_begin_does_not_publish_header_after_init_error() { + let events = Rc::new(RefCell::new(Vec::<&'static str>::new())); + let expected_error = ERROR(ZstdErrorCode::Generic); + let projection = ZSTDMT_compressionJobBeginProjection { + firstJob: 1, + hasCDict: 1, + }; + let cdict_events = Rc::clone(&events); + + let result = begin_compression_job_with( + projection, + move || { + cdict_events.borrow_mut().push("cdict"); + expected_error + }, + |_| panic!("cdict initialization failure must not update parameters"), + || panic!("cdict initialization failure must not update prefix policy"), + || panic!("cdict initialization failure must not initialize a raw prefix"), + || panic!("cdict initialization failure must not publish a header"), + ); + + assert_eq!(result, expected_error); + assert_eq!(&*events.borrow(), &["cdict"]); + } + + #[test] + fn compression_job_begin_rejects_cdict_on_nonfirst_job() { + let projection = ZSTDMT_compressionJobBeginProjection { + firstJob: 0, + hasCDict: 1, + }; + + let result = begin_compression_job_with( + projection, + || panic!("invalid cdict placement must not initialize"), + |_| panic!("invalid cdict placement must not update parameters"), + || panic!("invalid cdict placement must not update prefix policy"), + || panic!("invalid cdict placement must not initialize a raw prefix"), + || panic!("invalid cdict placement must not publish a header"), + ); + + assert_eq!(result, ERROR(ZstdErrorCode::StageWrong)); + } + #[test] fn chunk_loop_handles_empty_jobs_without_compression() { let mut compressor = MockChunkCompressor {