From 69425cb8a44fb0c65f633dc1a16f3ee9671b3d55 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Mon, 20 Jul 2026 08:21:06 +0200 Subject: [PATCH] refactor(compress): move MT input hint dispatch to Rust Project the scalar single-thread hint inputs and keep the multithreaded context behind a C callback so Rust owns the worker-count dispatch. Preserve the existing buffer-mode assertion and both hint arithmetic leaves, with focused tests covering MT selection and single-thread fallback. All heavy verification was run serially with a 40 GiB virtual-memory cap and one build job. Test Plan: - git diff --cached --check - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml --all-targets (790 passed) - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/cli/Cargo.toml --all-targets (184 passed) - ulimit -v 41943040; cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check - ulimit -v 41943040; make -j1 - ulimit -v 41943040; make -j1 -C tests test --- lib/compress/zstd_compress.c | 86 +++++++++++++++++------- rust/src/zstd_compress.rs | 124 +++++++++++++++++++++++++++++++++++ 2 files changed, 187 insertions(+), 23 deletions(-) diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 26b78b174..02d73dc1d 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -1521,6 +1521,39 @@ size_t ZSTD_rust_nextInputSizeHint(int inBufferMode, size_t stableInNotConsumed, size_t inBuffTarget, size_t inBuffPos); +typedef size_t (*ZSTD_rust_nextInputSizeHintMTorST_f)(void* context); +typedef struct { + size_t blockSizeMax; + size_t stableInNotConsumed; + size_t inBuffTarget; + size_t inBuffPos; + int nbWorkers; + int inBufferMode; + void* mtContext; + ZSTD_rust_nextInputSizeHintMTorST_f mtNextInputSizeHint; +} ZSTD_rust_nextInputSizeHintMTorSTState; +typedef char ZSTD_rust_next_input_size_hint_mt_or_st_state_layout[ + (sizeof(ZSTD_rust_nextInputSizeHintMTorST_f) == sizeof(void*) + && offsetof(ZSTD_rust_nextInputSizeHintMTorSTState, blockSizeMax) == 0 + && offsetof(ZSTD_rust_nextInputSizeHintMTorSTState, stableInNotConsumed) + == sizeof(size_t) + && offsetof(ZSTD_rust_nextInputSizeHintMTorSTState, inBuffTarget) + == 2 * sizeof(size_t) + && offsetof(ZSTD_rust_nextInputSizeHintMTorSTState, inBuffPos) + == 3 * sizeof(size_t) + && offsetof(ZSTD_rust_nextInputSizeHintMTorSTState, nbWorkers) + == 4 * sizeof(size_t) + && offsetof(ZSTD_rust_nextInputSizeHintMTorSTState, inBufferMode) + == 4 * sizeof(size_t) + sizeof(int) + && offsetof(ZSTD_rust_nextInputSizeHintMTorSTState, mtContext) + == 4 * sizeof(size_t) + 2 * sizeof(int) + && offsetof(ZSTD_rust_nextInputSizeHintMTorSTState, mtNextInputSizeHint) + == 4 * sizeof(size_t) + 2 * sizeof(int) + sizeof(void*) + && sizeof(ZSTD_rust_nextInputSizeHintMTorSTState) + == 4 * sizeof(size_t) + 2 * sizeof(int) + 2 * sizeof(void*)) + ? 1 : -1]; +size_t ZSTD_rust_nextInputSizeHintMTorST( + const ZSTD_rust_nextInputSizeHintMTorSTState* state); size_t ZSTD_rust_CStreamInSize(void); size_t ZSTD_rust_CStreamOutSize(void); size_t ZSTD_rust_sizeofCDict(size_t objectSize, size_t workspaceSize); @@ -7788,24 +7821,6 @@ size_t ZSTD_initCStream(ZSTD_CStream* zcs, int compressionLevel) /*====== Compression ======*/ -static size_t ZSTD_nextInputSizeHint(const ZSTD_CCtx* cctx) -{ - int const inBufferMode = (int)cctx->appliedParams.inBufferMode; - if (inBufferMode == ZSTD_bm_stable) { - return ZSTD_rust_nextInputSizeHint(inBufferMode, - cctx->blockSizeMax, - cctx->stableIn_notConsumed, - cctx->inBuffTarget, - cctx->inBuffPos); - } - assert(inBufferMode == ZSTD_bm_buffered); - return ZSTD_rust_nextInputSizeHint(inBufferMode, - cctx->blockSizeMax, - cctx->stableIn_notConsumed, - cctx->inBuffTarget, - cctx->inBuffPos); -} - /** ZSTD_compressStream_generic(): * internal function for all *compressStream*() variants * @return : hint size for next input to complete ongoing block */ @@ -7860,16 +7875,41 @@ static size_t ZSTD_compressStream_generic(ZSTD_CStream* zcs, &state, output, input, (int)flushMode); } +/* Keep the MT context opaque to Rust; only its existing C hint leaf crosses + * this callback boundary. */ +#ifdef ZSTD_MULTITHREAD +static size_t ZSTD_nextInputSizeHint_MT(void* context) +{ + return ZSTDMT_nextInputSizeHint((const ZSTDMT_CCtx*)context); +} +#endif + static size_t ZSTD_nextInputSizeHint_MTorST(const ZSTD_CCtx* cctx) { + int const inBufferMode = (int)cctx->appliedParams.inBufferMode; + ZSTD_rust_nextInputSizeHintMTorSTState state = { + cctx->blockSizeMax, + cctx->stableIn_notConsumed, + cctx->inBuffTarget, + cctx->inBuffPos, + 0, + inBufferMode, + NULL, + NULL + }; + + if (inBufferMode != ZSTD_bm_stable) { + assert(inBufferMode == ZSTD_bm_buffered); + } #ifdef ZSTD_MULTITHREAD - if (cctx->appliedParams.nbWorkers >= 1) { - assert(cctx->mtctx != NULL); - return ZSTDMT_nextInputSizeHint(cctx->mtctx); + state.nbWorkers = cctx->appliedParams.nbWorkers; + state.mtContext = cctx->mtctx; + state.mtNextInputSizeHint = ZSTD_nextInputSizeHint_MT; + if (state.nbWorkers >= 1) { + assert(state.mtContext != NULL); } #endif - return ZSTD_nextInputSizeHint(cctx); - + return ZSTD_rust_nextInputSizeHintMTorST(&state); } size_t ZSTD_compressStream(ZSTD_CStream* zcs, ZSTD_outBuffer* output, ZSTD_inBuffer* input) diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index a488e0ab6..81decb941 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -6831,6 +6831,88 @@ pub extern "C" fn ZSTD_rust_nextInputSizeHint( ) } +type NextInputSizeHintMTorSTFn = unsafe extern "C" fn(*mut c_void) -> usize; + +/// Projection for the MT-versus-single-thread input-hint dispatch. +/// +/// Rust owns only the worker-count branch. The single-thread hint inputs are +/// scalar, while the MT context remains opaque behind its C callback. +#[repr(C)] +pub struct ZSTD_rust_nextInputSizeHintMTorSTState { + block_size_max: usize, + stable_in_not_consumed: usize, + in_buff_target: usize, + in_buff_pos: usize, + nb_workers: c_int, + in_buffer_mode: c_int, + mt_context: *mut c_void, + mt_next_input_size_hint: Option, +} + +const _: () = { + assert!(size_of::() == size_of::()); + assert!(size_of::>() == size_of::()); + assert!(offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, block_size_max) == 0); + assert!( + offset_of!( + ZSTD_rust_nextInputSizeHintMTorSTState, + stable_in_not_consumed + ) == size_of::() + ); + assert!( + offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, in_buff_target) + == 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, in_buff_pos) == 3 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, nb_workers) == 4 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, in_buffer_mode) + == 4 * size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, mt_context) + == 4 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!( + ZSTD_rust_nextInputSizeHintMTorSTState, + mt_next_input_size_hint + ) == 4 * size_of::() + 2 * size_of::() + size_of::() + ); + assert!( + size_of::() + == 4 * size_of::() + 2 * size_of::() + 2 * size_of::() + ); +}; + +/// Select the MT or single-thread input-size hint without exposing the MT +/// context representation to Rust. +#[no_mangle] +pub unsafe extern "C" fn ZSTD_rust_nextInputSizeHintMTorST( + state: *const ZSTD_rust_nextInputSizeHintMTorSTState, +) -> usize { + let Some(state) = (unsafe { state.as_ref() }) else { + return 0; + }; + if state.nb_workers >= 1 { + let Some(mt_next_input_size_hint) = state.mt_next_input_size_hint else { + return 0; + }; + return unsafe { mt_next_input_size_hint(state.mt_context) }; + } + next_input_size_hint( + state.in_buffer_mode, + state.block_size_max, + state.stable_in_not_consumed, + state.in_buff_target, + state.in_buff_pos, + ) +} + #[inline] fn mt_next_input_size_hint(target_section_size: usize, in_buff_filled: usize) -> usize { let hint_in_size = target_section_size.wrapping_sub(in_buff_filled); @@ -14731,6 +14813,48 @@ mod tests { ); } + unsafe extern "C" fn next_input_size_hint_mt_or_st_test_callback( + context: *mut c_void, + ) -> usize { + let calls = unsafe { &mut *context.cast::() }; + *calls += 1; + 777 + } + + fn next_input_size_hint_mt_or_st_test_state( + nb_workers: c_int, + calls: &mut usize, + ) -> ZSTD_rust_nextInputSizeHintMTorSTState { + ZSTD_rust_nextInputSizeHintMTorSTState { + block_size_max: 256, + stable_in_not_consumed: 37, + in_buff_target: 128, + in_buff_pos: 32, + nb_workers, + in_buffer_mode: ZSTD_BM_BUFFERED, + mt_context: calls as *mut usize as *mut c_void, + mt_next_input_size_hint: Some(next_input_size_hint_mt_or_st_test_callback), + } + } + + #[test] + fn next_input_size_hint_mt_or_st_dispatches_to_mt_callback_for_workers() { + let mut calls = 0; + let state = next_input_size_hint_mt_or_st_test_state(1, &mut calls); + + assert_eq!(unsafe { ZSTD_rust_nextInputSizeHintMTorST(&state) }, 777); + assert_eq!(calls, 1); + } + + #[test] + fn next_input_size_hint_mt_or_st_uses_single_thread_leaf_without_workers() { + let mut calls = 0; + let state = next_input_size_hint_mt_or_st_test_state(0, &mut calls); + + assert_eq!(unsafe { ZSTD_rust_nextInputSizeHintMTorST(&state) }, 96); + assert_eq!(calls, 0); + } + #[test] fn mt_next_input_size_hint_handles_empty_input_buffer() { assert_eq!(mt_next_input_size_hint(128, 0), 128);