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

refactor(compress): move frame progression dispatch to Rust

Browse Source 
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.
This commit is contained in:
ddidderr
2026-07-21 08:38:26 +02:00
parent e01f79fb1a
commit 11dc99899b
2 changed files with 194 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+56 -12
View File
@@ -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 {
rust/src/zstd_compress.rs
+138
View File
@@ -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<FrameProgressionMTorSTFn>,
nb_workers: c_int,
}
const _: () = {
assert!(size_of::<FrameProgressionMTorSTFn>() == size_of::<usize>());
assert!(size_of::<Option<FrameProgressionMTorSTFn>>() == size_of::<usize>());
assert!(offset_of!(ZSTD_rust_frameProgressionMTorSTState, consumed_src_size) == 0);
assert!(offset_of!(ZSTD_rust_frameProgressionMTorSTState, produced_c_size) == size_of::<u64>());
assert!(offset_of!(ZSTD_rust_frameProgressionMTorSTState, buffered) == 2 * size_of::<u64>());
assert!(
offset_of!(ZSTD_rust_frameProgressionMTorSTState, mt_context)
== 2 * size_of::<u64>() + size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_frameProgressionMTorSTState, mt_frame_progression)
== 2 * size_of::<u64>() + 2 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_frameProgressionMTorSTState, nb_workers)
== 2 * size_of::<u64>() + 3 * size_of::<usize>()
);
assert!(
size_of::<ZSTD_rust_frameProgressionMTorSTState>()
== if size_of::<usize>() == 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::<usize>() };
*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::<usize>() };
*calls += 1;