From 11dc99899b3e5ad6be468327e88b21bca38ee210 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Tue, 21 Jul 2026 08:38:26 +0200 Subject: [PATCH] refactor(compress): move frame progression dispatch to Rust Move the public ZSTD_getFrameProgression MT-versus-single-thread worker-count branch into a Rust-owned scalar projection. C continues to compute the single-thread scalar inputs and keeps the private ZSTDMT context behind a callback; the compile-time ZSTD_MULTITHREAD adapter remains local to C. Add C/Rust layout assertions and focused Rust tests for both dispatch paths and null-state fallback. Test Plan: - `git diff --cached --check` - Not run: Cargo, make, builds, native tests, and heavy verification per request. --- lib/compress/zstd_compress.c | 68 ++++++++++++++--- rust/src/zstd_compress.rs | 138 +++++++++++++++++++++++++++++++++++ 2 files changed, 194 insertions(+), 12 deletions(-) diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 7a96f3ab4..222d2b40a 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -729,6 +729,34 @@ typedef char ZSTD_rust_compress_begin_advanced_state_layout[ ZSTD_frameProgression ZSTD_rust_frameProgression(U64 consumedSrcSize, size_t buffered, U64 producedCSize); +typedef ZSTD_frameProgression (*ZSTD_rust_frameProgressionMTorST_f)( + void* context); +typedef struct { + U64 consumedSrcSize; + U64 producedCSize; + size_t buffered; + void* mtContext; + ZSTD_rust_frameProgressionMTorST_f mtFrameProgression; + int nbWorkers; +} ZSTD_rust_frameProgressionMTorSTState; +typedef char ZSTD_rust_frame_progression_mt_or_st_state_layout[ + (sizeof(ZSTD_rust_frameProgressionMTorST_f) == sizeof(void*) + && offsetof(ZSTD_rust_frameProgressionMTorSTState, consumedSrcSize) == 0 + && offsetof(ZSTD_rust_frameProgressionMTorSTState, producedCSize) + == sizeof(U64) + && offsetof(ZSTD_rust_frameProgressionMTorSTState, buffered) + == 2 * sizeof(U64) + && offsetof(ZSTD_rust_frameProgressionMTorSTState, mtContext) + == 2 * sizeof(U64) + sizeof(size_t) + && offsetof(ZSTD_rust_frameProgressionMTorSTState, mtFrameProgression) + == 2 * sizeof(U64) + sizeof(size_t) + sizeof(void*) + && offsetof(ZSTD_rust_frameProgressionMTorSTState, nbWorkers) + == 2 * sizeof(U64) + sizeof(size_t) + 2 * sizeof(void*) + && sizeof(ZSTD_rust_frameProgressionMTorSTState) + == (sizeof(void*) == 8 ? 48 : 32)) + ? 1 : -1]; +ZSTD_frameProgression ZSTD_rust_frameProgressionMTorST( + const ZSTD_rust_frameProgressionMTorSTState* state); size_t ZSTD_rust_resetCCtxForSimpleCompression(void* cctx); size_t ZSTD_rust_prepareCCtxForSimpleCompression(void* cctx, size_t srcSize, @@ -4411,21 +4439,37 @@ size_t ZSTD_estimateCStreamSize(int compressionLevel) * tells how much data has been consumed (input) and produced (output) for current frame. * able to count progression inside worker threads (non-blocking mode). */ -ZSTD_frameProgression ZSTD_getFrameProgression(const ZSTD_CCtx* cctx) +static ZSTD_frameProgression ZSTD_rust_frameProgression_MT(void* context) { #ifdef ZSTD_MULTITHREAD - if (cctx->appliedParams.nbWorkers > 0) { - return ZSTDMT_getFrameProgression(cctx->mtctx); - } + return ZSTDMT_getFrameProgression((const ZSTDMT_CCtx*)context); +#else + (void)context; + return (ZSTD_frameProgression){ 0 }; #endif - { - size_t const buffered = (cctx->inBuff == NULL) ? 0 : - cctx->inBuffPos - cctx->inToCompress; - if (buffered) assert(cctx->inBuffPos >= cctx->inToCompress); - assert(buffered <= ZSTD_BLOCKSIZE_MAX); - return ZSTD_rust_frameProgression(cctx->consumedSrcSize, buffered, - cctx->producedCSize); - } } +} + +ZSTD_frameProgression ZSTD_getFrameProgression(const ZSTD_CCtx* cctx) +{ + size_t const buffered = (cctx->inBuff == NULL) ? 0 : + cctx->inBuffPos - cctx->inToCompress; + ZSTD_rust_frameProgressionMTorSTState state = { + cctx->consumedSrcSize, + cctx->producedCSize, + buffered, + NULL, + NULL, + 0 + }; + if (buffered) assert(cctx->inBuffPos >= cctx->inToCompress); + assert(buffered <= ZSTD_BLOCKSIZE_MAX); +#ifdef ZSTD_MULTITHREAD + state.mtContext = cctx->mtctx; + state.mtFrameProgression = ZSTD_rust_frameProgression_MT; + state.nbWorkers = (int)cctx->appliedParams.nbWorkers; +#endif + return ZSTD_rust_frameProgressionMTorST(&state); +} typedef size_t (*ZSTD_rust_toFlushNow_f)(void* context); typedef struct { diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index 125411b8c..1b566a77d 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -7025,6 +7025,69 @@ pub extern "C" fn ZSTD_rust_frameProgression( frame_progression(consumed_src_size, buffered, produced_c_size) } +type FrameProgressionMTorSTFn = unsafe extern "C" fn(*mut c_void) -> ZSTD_frameProgression; + +/// Projection for the MT-versus-single-thread frame-progression dispatch. +/// +/// Rust owns only the worker-count branch. The single-thread progression +/// inputs are scalar, while the private MT context remains opaque behind its +/// C callback. +#[repr(C)] +pub struct ZSTD_rust_frameProgressionMTorSTState { + consumed_src_size: u64, + produced_c_size: u64, + buffered: usize, + mt_context: *mut c_void, + mt_frame_progression: Option, + nb_workers: c_int, +} + +const _: () = { + assert!(size_of::() == size_of::()); + assert!(size_of::>() == size_of::()); + assert!(offset_of!(ZSTD_rust_frameProgressionMTorSTState, consumed_src_size) == 0); + assert!(offset_of!(ZSTD_rust_frameProgressionMTorSTState, produced_c_size) == size_of::()); + assert!(offset_of!(ZSTD_rust_frameProgressionMTorSTState, buffered) == 2 * size_of::()); + assert!( + offset_of!(ZSTD_rust_frameProgressionMTorSTState, mt_context) + == 2 * size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_frameProgressionMTorSTState, mt_frame_progression) + == 2 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_frameProgressionMTorSTState, nb_workers) + == 2 * size_of::() + 3 * size_of::() + ); + assert!( + size_of::() + == if size_of::() == 8 { 48 } else { 32 } + ); +}; + +/// Select the MT or single-thread frame progression without exposing the MT +/// context representation to Rust. +#[no_mangle] +pub unsafe extern "C" fn ZSTD_rust_frameProgressionMTorST( + state: *const ZSTD_rust_frameProgressionMTorSTState, +) -> ZSTD_frameProgression { + let Some(state) = (unsafe { state.as_ref() }) else { + return ZSTD_frameProgression::default(); + }; + if state.nb_workers > 0 { + let Some(mt_frame_progression) = state.mt_frame_progression else { + return ZSTD_frameProgression::default(); + }; + return unsafe { mt_frame_progression(state.mt_context) }; + } + frame_progression( + state.consumed_src_size, + state.buffered, + state.produced_c_size, + ) +} + type ToFlushNowFn = unsafe extern "C" fn(*mut c_void) -> usize; /// Explicit projection for the `ZSTD_toFlushNow` multithread policy. @@ -15749,6 +15812,81 @@ mod tests { assert_eq!(progression.nbActiveWorkers, 0); } + unsafe extern "C" fn frame_progression_mt_or_st_test_callback( + context: *mut c_void, + ) -> ZSTD_frameProgression { + let calls = unsafe { &mut *context.cast::() }; + *calls += 1; + ZSTD_frameProgression { + ingested: 101, + consumed: 103, + produced: 107, + flushed: 109, + currentJobID: 11, + nbActiveWorkers: 13, + } + } + + fn frame_progression_mt_or_st_test_state( + nb_workers: c_int, + calls: &mut usize, + ) -> ZSTD_rust_frameProgressionMTorSTState { + ZSTD_rust_frameProgressionMTorSTState { + consumed_src_size: 11, + produced_c_size: 53, + buffered: 37, + mt_context: calls as *mut usize as *mut c_void, + mt_frame_progression: Some(frame_progression_mt_or_st_test_callback), + nb_workers, + } + } + + #[test] + fn frame_progression_mt_or_st_dispatches_to_mt_callback_for_workers() { + let mut calls = 0; + let state = frame_progression_mt_or_st_test_state(1, &mut calls); + + assert_eq!( + unsafe { ZSTD_rust_frameProgressionMTorST(&state) }, + ZSTD_frameProgression { + ingested: 101, + consumed: 103, + produced: 107, + flushed: 109, + currentJobID: 11, + nbActiveWorkers: 13, + } + ); + assert_eq!(calls, 1); + } + + #[test] + fn frame_progression_mt_or_st_uses_single_thread_leaf_without_workers() { + let mut calls = 0; + let state = frame_progression_mt_or_st_test_state(0, &mut calls); + + assert_eq!( + unsafe { ZSTD_rust_frameProgressionMTorST(&state) }, + ZSTD_frameProgression { + ingested: 48, + consumed: 11, + produced: 53, + flushed: 53, + currentJobID: 0, + nbActiveWorkers: 0, + } + ); + assert_eq!(calls, 0); + } + + #[test] + fn frame_progression_mt_or_st_rejects_a_null_state() { + assert_eq!( + unsafe { ZSTD_rust_frameProgressionMTorST(ptr::null()) }, + ZSTD_frameProgression::default() + ); + } + unsafe extern "C" fn to_flush_now_test_callback(context: *mut c_void) -> usize { let calls = unsafe { &mut *context.cast::() }; *calls += 1;