diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index c2bf72058..b0cb2a67f 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -25,8 +25,9 @@ #include "zstd_ldm.h" #include "zstdmt_compress.h" -/* The Rust parameter leaf receives only this scalar snapshot. Keep the - * private ZSTDMT_CCtx and ZSTD_CCtx_params layouts in this translation unit. */ +/* The Rust MT parameter-update policy receives only this scalar snapshot. + * Keep the private ZSTDMT_CCtx and ZSTD_CCtx_params layouts in this + * translation unit. */ typedef struct { int compressionLevel; int cctxSrcSizeHint; @@ -65,7 +66,18 @@ typedef char ZSTDMT_rust_cparams_update_projection_layout[ == (sizeof(void*) == 8 ? 80 : 72)) ? 1 : -1]; -ZSTD_compressionParameters ZSTD_rust_params_updateCParamsWhileCompressing( +typedef struct { + int compressionLevel; + ZSTD_compressionParameters cParams; +} ZSTDMT_RustCParamsUpdateResult; +typedef char ZSTDMT_rust_cparams_update_result_layout[ + (offsetof(ZSTDMT_RustCParamsUpdateResult, compressionLevel) == 0 + && offsetof(ZSTDMT_RustCParamsUpdateResult, cParams) == sizeof(int) + && sizeof(ZSTDMT_RustCParamsUpdateResult) + == sizeof(int) + sizeof(ZSTD_compressionParameters)) + ? 1 : -1]; + +ZSTDMT_RustCParamsUpdateResult ZSTDMT_rust_updateCParamsWhileCompressing( ZSTDMT_RustCParamsUpdateProjection projection); /* Defined in zstd_compress.c so all C translation units use the same @@ -2571,10 +2583,8 @@ static size_t ZSTDMT_resize(ZSTDMT_CCtx* mtctx, unsigned nbWorkers) void ZSTDMT_updateCParams_whileCompressing(ZSTDMT_CCtx* mtctx, const ZSTD_CCtx_params* cctxParams) { U32 const saved_wlog = mtctx->params.cParams.windowLog; /* Do not modify windowLog while compressing */ - int const compressionLevel = cctxParams->compressionLevel; DEBUGLOG(5, "ZSTDMT_updateCParams_whileCompressing (level:%i)", - compressionLevel); - mtctx->params.compressionLevel = compressionLevel; + cctxParams->compressionLevel); { ZSTDMT_RustCParamsUpdateProjection const projection = { cctxParams->compressionLevel, cctxParams->srcSizeHint, @@ -2588,8 +2598,10 @@ void ZSTDMT_updateCParams_whileCompressing(ZSTDMT_CCtx* mtctx, const ZSTD_CCtx_p ZSTD_getCParamsExclusionMask(), saved_wlog }; - mtctx->params.cParams = - ZSTD_rust_params_updateCParamsWhileCompressing(projection); + ZSTDMT_RustCParamsUpdateResult const result = + ZSTDMT_rust_updateCParamsWhileCompressing(projection); + mtctx->params.compressionLevel = result.compressionLevel; + mtctx->params.cParams = result.cParams; } } diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 1845839f2..dfb354b9e 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -25,6 +25,10 @@ use crate::errors::{ERR_isError, ZstdErrorCode, ERROR}; use crate::mem::MEM_writeLE32; use crate::zstd_compress::{ZSTD_frameProgression, ZSTD_rust_invalidateRepCodes}; use crate::zstd_compress_frame::ZSTD_rust_writeFrameHeader; +use crate::zstd_compress_params::{ + ZSTD_compressionParameters, ZSTD_rust_params_updateCParamsWhileCompressing, + ZSTD_rust_params_updateCParamsWhileCompressingState, +}; const ZSTDMT_JOBLOG_MAX: c_uint = if mem::size_of::() == 4 { 29 } else { 30 }; const ZSTD_WINDOWLOG_MAX: c_uint = if mem::size_of::() == 4 { 30 } else { 31 }; @@ -88,6 +92,121 @@ const _: () = { ); }; +/// Scalar MT state used when compression parameters are updated between +/// streaming jobs. The private `ZSTDMT_CCtx` and `ZSTD_CCtx_params` layouts +/// remain in C; Rust owns the transition policy and returns the two values C +/// must publish together. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_CParamsUpdateProjection { + pub compressionLevel: c_int, + pub cctxSrcSizeHint: c_int, + pub srcSizeHint: u64, + pub dictSize: usize, + pub mode: c_int, + pub enableLdm: c_int, + pub ldmDefaultWindowLog: c_uint, + pub overrides: ZSTD_compressionParameters, + pub useRowMatchFinder: c_int, + pub exclusionMask: c_uint, + pub savedWindowLog: c_uint, +} + +const _: () = { + assert!(offset_of!(ZSTDMT_CParamsUpdateProjection, compressionLevel) == 0); + assert!(offset_of!(ZSTDMT_CParamsUpdateProjection, cctxSrcSizeHint) == size_of::()); + assert!(offset_of!(ZSTDMT_CParamsUpdateProjection, srcSizeHint) == 2 * size_of::()); + assert!(offset_of!(ZSTDMT_CParamsUpdateProjection, dictSize) == 16); + assert!( + offset_of!(ZSTDMT_CParamsUpdateProjection, mode) + == if size_of::() == 8 { 24 } else { 20 } + ); + assert!( + offset_of!(ZSTDMT_CParamsUpdateProjection, enableLdm) + == if size_of::() == 8 { 28 } else { 24 } + ); + assert!( + offset_of!(ZSTDMT_CParamsUpdateProjection, ldmDefaultWindowLog) + == if size_of::() == 8 { 32 } else { 28 } + ); + assert!( + offset_of!(ZSTDMT_CParamsUpdateProjection, overrides) + == if size_of::() == 8 { 36 } else { 32 } + ); + assert!( + offset_of!(ZSTDMT_CParamsUpdateProjection, useRowMatchFinder) + == if size_of::() == 8 { 64 } else { 60 } + ); + assert!( + offset_of!(ZSTDMT_CParamsUpdateProjection, exclusionMask) + == if size_of::() == 8 { 68 } else { 64 } + ); + assert!( + offset_of!(ZSTDMT_CParamsUpdateProjection, savedWindowLog) + == if size_of::() == 8 { 72 } else { 68 } + ); + assert!( + size_of::() + == if size_of::() == 8 { 80 } else { 72 } + ); +}; + +/// Result of the MT parameter transition. Keeping the requested level next +/// to the derived compression parameters makes the C publication explicit: +/// C retains the private context, while Rust decides the complete scalar +/// state for the next job. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_CParamsUpdateResult { + pub compressionLevel: c_int, + pub cParams: ZSTD_compressionParameters, +} + +const _: () = { + assert!(offset_of!(ZSTDMT_CParamsUpdateResult, compressionLevel) == 0); + assert!(offset_of!(ZSTDMT_CParamsUpdateResult, cParams) == size_of::()); + assert!( + size_of::() + == size_of::() + size_of::() + ); +}; + +#[inline] +fn update_cparams_while_compressing( + projection: ZSTDMT_CParamsUpdateProjection, +) -> ZSTDMT_CParamsUpdateResult { + let c_params = ZSTD_rust_params_updateCParamsWhileCompressing( + ZSTD_rust_params_updateCParamsWhileCompressingState { + compressionLevel: projection.compressionLevel, + cctxSrcSizeHint: projection.cctxSrcSizeHint, + srcSizeHint: projection.srcSizeHint, + dictSize: projection.dictSize, + mode: projection.mode, + enableLdm: projection.enableLdm, + ldmDefaultWindowLog: projection.ldmDefaultWindowLog, + overrides: projection.overrides, + useRowMatchFinder: projection.useRowMatchFinder, + exclusionMask: projection.exclusionMask, + savedWindowLog: projection.savedWindowLog, + }, + ); + ZSTDMT_CParamsUpdateResult { + compressionLevel: projection.compressionLevel, + cParams: c_params, + } +} + +/// Derive and publish the scalar MT parameter transition for the next job. +/// C keeps the private parameter object and performs the two final writes; +/// Rust owns the rule that the requested level travels with the derived +/// parameters and that the active frame window log is preserved. +#[no_mangle] +pub extern "C" fn ZSTDMT_rust_updateCParamsWhileCompressing( + projection: ZSTDMT_CParamsUpdateProjection, +) -> ZSTDMT_CParamsUpdateResult { + update_cparams_while_compressing(projection) +} + /// 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. @@ -6837,6 +6956,95 @@ mod tests { } } + fn mt_cparams_update_projection() -> ZSTDMT_CParamsUpdateProjection { + ZSTDMT_CParamsUpdateProjection { + compressionLevel: 3, + cctxSrcSizeHint: 0, + srcSizeHint: u64::MAX, + dictSize: 0, + mode: 0, + enableLdm: ZSTD_PS_DISABLE, + ldmDefaultWindowLog: 27, + overrides: ZSTD_compressionParameters::default(), + useRowMatchFinder: 0, + exclusionMask: 0, + savedWindowLog: 18, + } + } + + fn expected_mt_cparams_update( + projection: ZSTDMT_CParamsUpdateProjection, + ) -> ZSTD_compressionParameters { + ZSTD_rust_params_updateCParamsWhileCompressing( + ZSTD_rust_params_updateCParamsWhileCompressingState { + compressionLevel: projection.compressionLevel, + cctxSrcSizeHint: projection.cctxSrcSizeHint, + srcSizeHint: projection.srcSizeHint, + dictSize: projection.dictSize, + mode: projection.mode, + enableLdm: projection.enableLdm, + ldmDefaultWindowLog: projection.ldmDefaultWindowLog, + overrides: projection.overrides, + useRowMatchFinder: projection.useRowMatchFinder, + exclusionMask: projection.exclusionMask, + savedWindowLog: projection.savedWindowLog, + }, + ) + } + + #[test] + fn mt_update_parameters_carries_level_with_next_job_params() { + let projection = mt_cparams_update_projection(); + let expected = expected_mt_cparams_update(projection); + let result = update_cparams_while_compressing(projection); + + assert_eq!(result.compressionLevel, projection.compressionLevel); + assert_eq!(result.cParams, expected); + assert_eq!(result.cParams.windowLog, projection.savedWindowLog); + assert_eq!( + ZSTDMT_rust_updateCParamsWhileCompressing(projection), + result + ); + } + + #[test] + fn mt_update_parameters_uses_context_size_hint_only_for_unknown_size() { + let mut from_context_hint = mt_cparams_update_projection(); + from_context_hint.cctxSrcSizeHint = 1 << 20; + + let mut explicit_size = mt_cparams_update_projection(); + explicit_size.srcSizeHint = 1 << 20; + + assert_eq!( + update_cparams_while_compressing(from_context_hint).cParams, + update_cparams_while_compressing(explicit_size).cParams + ); + + let mut explicit_with_stale_hint = explicit_size; + explicit_with_stale_hint.cctxSrcSizeHint = 1; + assert_eq!( + update_cparams_while_compressing(explicit_size).cParams, + update_cparams_while_compressing(explicit_with_stale_hint).cParams + ); + } + + #[test] + fn mt_update_parameters_restores_active_window_after_ldm_and_overrides() { + let mut projection = mt_cparams_update_projection(); + projection.compressionLevel = -1; + projection.enableLdm = ZSTD_PS_ENABLE; + projection.ldmDefaultWindowLog = 27; + projection.overrides.strategy = ZSTD_BTULTRA2; + projection.savedWindowLog = 19; + + let result = update_cparams_while_compressing(projection); + + assert_eq!(result.compressionLevel, -1); + assert_eq!(result.cParams.strategy, ZSTD_BTULTRA2); + assert_eq!(result.cParams.windowLog, projection.savedWindowLog); + assert_ne!(result.cParams.windowLog, projection.ldmDefaultWindowLog); + } + #[test] fn dictionary_policy_classifies_copy_borrow_and_raw_prefix_branches() { assert_eq!(