From 71a18ce3159fd4ca63efc1cf7d8bccd81f3f35ec Mon Sep 17 00:00:00 2001 From: ddidderr Date: Mon, 20 Jul 2026 05:42:31 +0200 Subject: [PATCH] refactor(mt): move per-job parameter policy to Rust Move the pure multithreaded worker-parameter policy into Rust: only the first job keeps the frame checksum flag, worker jobs disable LDM, and every worker context runs with zero nested workers. Keep the complete C-owned parameter structure and codec initialization in the adapter, crossing the ABI with a small layout-checked scalar projection. Test Plan: - rustfmt --edition 2021 --check rust/src/zstdmt_compress.rs - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings - ulimit -v 41943040; make -j1 - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml --all-targets - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/cli/Cargo.toml --all-targets - ulimit -v 41943040; make -j1 -C tests test --- lib/compress/zstdmt_compress.c | 32 ++++++++++--- rust/src/zstdmt_compress.rs | 84 +++++++++++++++++++++++++++++++++- 2 files changed, 108 insertions(+), 8 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index f1f936325..09c352b58 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -190,6 +190,22 @@ typedef ZSTDMT_chunkProcessResult (*ZSTDMT_compressionJobCompressFn)( typedef void (*ZSTDMT_compressionJobErrorFn)(void* opaque, size_t error); typedef void (*ZSTDMT_compressionJobFinishFn)(void* opaque, size_t lastBlockSize); +typedef struct { + int checksumFlag; + int ldmEnable; + unsigned nbWorkers; +} ZSTDMT_RustCompressionJobParameters; +typedef char ZSTDMT_compression_job_parameters_layout[ + (offsetof(ZSTDMT_RustCompressionJobParameters, checksumFlag) == 0 + && offsetof(ZSTDMT_RustCompressionJobParameters, ldmEnable) == sizeof(int) + && offsetof(ZSTDMT_RustCompressionJobParameters, nbWorkers) == 2 * sizeof(int) + && sizeof(ZSTDMT_RustCompressionJobParameters) + == 2 * sizeof(int) + sizeof(unsigned)) + ? 1 : -1]; + +ZSTDMT_RustCompressionJobParameters ZSTDMT_rust_prepareCompressionJobParameters( + unsigned jobID, int checksumFlag, int ldmEnable, unsigned nbWorkers); + void ZSTDMT_rust_compressionJob( const ZSTDMT_RustCompressionJobProjection* projection, void* opaque, @@ -1471,14 +1487,16 @@ static void ZSTDMT_compressionJobPrepareParameters(void* opaque) ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; + ZSTDMT_RustCompressionJobParameters const parameters = + ZSTDMT_rust_prepareCompressionJobParameters( + job->jobID, + state->jobParams.fParams.checksumFlag, + state->jobParams.ldmParams.enableLdm, + (unsigned)state->jobParams.nbWorkers); - /* Don't compute the checksum for chunks, since we compute it externally, - * but write it in the header. */ - if (job->jobID != 0) state->jobParams.fParams.checksumFlag = 0; - /* Don't run LDM for the chunks, since we handle it externally. */ - state->jobParams.ldmParams.enableLdm = ZSTD_ps_disable; - /* Correct nbWorkers to 0. */ - state->jobParams.nbWorkers = 0; + state->jobParams.fParams.checksumFlag = parameters.checksumFlag; + state->jobParams.ldmParams.enableLdm = parameters.ldmEnable; + state->jobParams.nbWorkers = (int)parameters.nbWorkers; } static void ZSTDMT_compressionJobGenerateSequences(void* opaque) diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index d8d726de5..2ca133704 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -45,7 +45,6 @@ const ZSTD_BTOPT: c_int = 7; const ZSTD_BTULTRA: c_int = 8; const ZSTD_BTULTRA2: c_int = 9; const ZSTD_PS_ENABLE: c_int = 1; -#[cfg(test)] const ZSTD_PS_DISABLE: c_int = 2; const RSYNC_LENGTH: usize = 32; @@ -66,6 +65,57 @@ pub struct ZSTDMT_chunkProcessResult { pub lastBlockSize: usize, } +/// Per-job parameter policy applied before a worker initializes its private +/// compression context. The full `ZSTD_CCtx_params` remains C-owned; only the +/// three scalar fields changed by the MT worker policy cross the ABI. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_compressionJobParameters { + pub checksumFlag: c_int, + pub ldmEnable: c_int, + pub nbWorkers: c_uint, +} + +const _: () = { + assert!(offset_of!(ZSTDMT_compressionJobParameters, checksumFlag) == 0); + assert!(offset_of!(ZSTDMT_compressionJobParameters, ldmEnable) == size_of::()); + assert!(offset_of!(ZSTDMT_compressionJobParameters, nbWorkers) == 2 * size_of::()); + assert!( + size_of::() + == 2 * size_of::() + size_of::() + ); +}; + +#[inline] +fn prepare_compression_job_parameters( + job_id: c_uint, + checksum_flag: c_int, + _ldm_enable: c_int, + _nb_workers: c_uint, +) -> ZSTDMT_compressionJobParameters { + ZSTDMT_compressionJobParameters { + /* The first job writes the frame checksum; later jobs have their + * checksum accounted for by the serial MT state. */ + checksumFlag: if job_id == 0 { checksum_flag } else { 0 }, + /* LDM sequences are generated by the serial state, not per worker. */ + ldmEnable: ZSTD_PS_DISABLE, + /* Worker contexts are initialized as single-threaded contexts. */ + nbWorkers: 0, + } +} + +/// Apply the pure per-worker MT parameter policy while C retains the complete +/// compression-parameter structure and all codec initialization. +#[no_mangle] +pub extern "C" fn ZSTDMT_rust_prepareCompressionJobParameters( + jobID: c_uint, + checksumFlag: c_int, + ldmEnable: c_int, + nbWorkers: c_uint, +) -> ZSTDMT_compressionJobParameters { + prepare_compression_job_parameters(jobID, checksumFlag, ldmEnable, nbWorkers) +} + /// Scalar job state used by the Rust compression-job scheduler. /// /// The job descriptor, pools, synchronization, and codec state remain @@ -7157,6 +7207,38 @@ mod tests { ); } + #[test] + fn compression_job_parameters_disable_worker_owned_features() { + assert_eq!( + prepare_compression_job_parameters(0, 1, ZSTD_PS_ENABLE, 4), + ZSTDMT_compressionJobParameters { + checksumFlag: 1, + ldmEnable: ZSTD_PS_DISABLE, + nbWorkers: 0, + } + ); + } + + #[test] + fn compression_job_parameters_clear_checksum_after_first_job() { + assert_eq!( + prepare_compression_job_parameters(3, 1, ZSTD_PS_ENABLE, 4), + ZSTDMT_compressionJobParameters { + checksumFlag: 0, + ldmEnable: ZSTD_PS_DISABLE, + nbWorkers: 0, + } + ); + assert_eq!( + prepare_compression_job_parameters(9, 0, ZSTD_PS_DISABLE, 0), + ZSTDMT_compressionJobParameters { + checksumFlag: 0, + ldmEnable: ZSTD_PS_DISABLE, + nbWorkers: 0, + } + ); + } + #[test] fn overlap_log_defaults_follow_strategy_groups() { assert_eq!(overlap_log_default(ZSTD_FAST), 6);