ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(compress): move MT input hint dispatch to Rust

Browse Source 
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
This commit is contained in:
ddidderr
2026-07-20 08:21:06 +02:00
parent 22cb347f6e
commit 69425cb8a4
2 changed files with 187 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+63 -23
View File
@@ -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)
rust/src/zstd_compress.rs
+124
View File
@@ -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<NextInputSizeHintMTorSTFn>,
}
const _: () = {
assert!(size_of::<NextInputSizeHintMTorSTFn>() == size_of::<usize>());
assert!(size_of::<Option<NextInputSizeHintMTorSTFn>>() == size_of::<usize>());
assert!(offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, block_size_max) == 0);
assert!(
offset_of!(
ZSTD_rust_nextInputSizeHintMTorSTState,
stable_in_not_consumed
) == size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, in_buff_target)
== 2 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, in_buff_pos) == 3 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, nb_workers) == 4 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, in_buffer_mode)
== 4 * size_of::<usize>() + size_of::<c_int>()
);
assert!(
offset_of!(ZSTD_rust_nextInputSizeHintMTorSTState, mt_context)
== 4 * size_of::<usize>() + 2 * size_of::<c_int>()
);
assert!(
offset_of!(
ZSTD_rust_nextInputSizeHintMTorSTState,
mt_next_input_size_hint
) == 4 * size_of::<usize>() + 2 * size_of::<c_int>() + size_of::<usize>()
);
assert!(
size_of::<ZSTD_rust_nextInputSizeHintMTorSTState>()
== 4 * size_of::<usize>() + 2 * size_of::<c_int>() + 2 * size_of::<usize>()
);
};
/// 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::<usize>() };
*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);