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

refactor(compress): move MT completion ordering to Rust

Browse Source 
The multithreaded stream loop already delegated scalar result classification to
Rust, but its final error/end conditional still mixed completion ordering with
the private C context. That left trace and reset sequencing embedded beside the
MT counters and made the next Rust seam depend on exposing CCtx layout.

Rust now consumes the projected loop action and sequences the final completion
side effects through opaque C callbacks. Continue and break do nothing, worker
errors reset the session, and completed end operations trace before resetting.
The reset callback result is intentionally discarded. C retains the private
context mutations in callback adapters, and the callback invocation remains
inside the existing ZSTD_MULTITHREAD branch before FORWARD_IF_ERROR(flushMin).
C/Rust repr(C) projections and ABI assertions protect the callback/result
boundary, while focused tests cover all four loop actions and end ordering.

Test Plan:
- `git diff --cached --check` -- passed
- `rustfmt --edition 2021 --check rust/src/zstd_compress.rs` -- passed
- single-thread and `-DZSTD_MULTITHREAD` C syntax checks -- passed; existing warnings only
- `CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --lib -- -D warnings` -- passed
- `CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --benches -- -D warnings` -- passed
- focused Rust test filter was attempted serially but is blocked by unrelated in-progress `rust/src/zstdmt_compress.rs` test ABI edits; no full test suites were run
This commit is contained in:
ddidderr
2026-07-21 10:29:33 +02:00
parent 5948934bd9
commit 6cc080a35a
2 changed files with 236 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+53 -5
View File
@@ -981,6 +981,37 @@ typedef char ZSTD_rust_compress_stream2_mt_loop_policy_state_layout[
&& sizeof(ZSTD_rust_compressStream2MTLoopPolicyState)
== 8 * sizeof(size_t))
? 1 : -1];
typedef void (*ZSTD_rust_compressStream2MTTrace_f)(void* context);
typedef size_t (*ZSTD_rust_compressStream2MTReset_f)(void* context);
typedef struct {
void* callbackContext;
ZSTD_rust_compressStream2MTTrace_f trace;
ZSTD_rust_compressStream2MTReset_f reset;
} ZSTD_rust_compressStream2MTCompletionCallbacks;
typedef struct {
int loopPolicy;
const ZSTD_rust_compressStream2MTCompletionCallbacks* callbacks;
} ZSTD_rust_compressStream2MTCompletionState;
void ZSTD_rust_compressStream2MTCompletion(
const ZSTD_rust_compressStream2MTCompletionState* state);
typedef char ZSTD_rust_compress_stream2_mt_completion_callbacks_layout[
(offsetof(ZSTD_rust_compressStream2MTCompletionCallbacks, callbackContext) == 0
&& offsetof(ZSTD_rust_compressStream2MTCompletionCallbacks, trace)
== sizeof(void*)
&& offsetof(ZSTD_rust_compressStream2MTCompletionCallbacks, reset)
== 2 * sizeof(void*)
&& sizeof(ZSTD_rust_compressStream2MTTrace_f) == sizeof(void*)
&& sizeof(ZSTD_rust_compressStream2MTReset_f) == sizeof(void*)
&& sizeof(ZSTD_rust_compressStream2MTCompletionCallbacks)
== 3 * sizeof(void*))
? 1 : -1];
typedef char ZSTD_rust_compress_stream2_mt_completion_state_layout[
(offsetof(ZSTD_rust_compressStream2MTCompletionState, loopPolicy) == 0
&& offsetof(ZSTD_rust_compressStream2MTCompletionState, callbacks)
== sizeof(void*)
&& sizeof(ZSTD_rust_compressStream2MTCompletionState)
== 2 * sizeof(void*))
? 1 : -1];
int ZSTD_rust_simpleCompress2Level(const void* cctx, size_t srcSize);
typedef int (*ZSTD_rust_simpleCompress2Level_f)(const void* cctx, size_t srcSize);
typedef struct {
@@ -8582,6 +8613,18 @@ static size_t ZSTD_CCtx_init_compressStream2(ZSTD_CCtx* cctx,
/* @return provides a minimum amount of data remaining to be flushed from internal buffers
*/
#ifdef ZSTD_MULTITHREAD
static void ZSTD_rust_compressStream2MT_trace(void* context)
{
ZSTD_CCtx_trace((ZSTD_CCtx*)context, 0);
}
static size_t ZSTD_rust_compressStream2MT_reset(void* context)
{
return ZSTD_CCtx_reset((ZSTD_CCtx*)context, ZSTD_reset_session_only);
}
#endif /* ZSTD_MULTITHREAD */
size_t ZSTD_compressStream2_c( ZSTD_CCtx* cctx,
ZSTD_outBuffer* output,
ZSTD_inBuffer* input,
@@ -8705,11 +8748,16 @@ size_t ZSTD_compressStream2_c( ZSTD_CCtx* cctx,
};
int const loopPolicy =
ZSTD_rust_compressStream2MTLoopPolicy(&state);
if (loopPolicy == ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_ERROR
|| loopPolicy == ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_END_COMPLETE) {
if (loopPolicy == ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_END_COMPLETE)
ZSTD_CCtx_trace(cctx, 0);
ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only);
{ ZSTD_rust_compressStream2MTCompletionCallbacks const callbacks = {
cctx,
ZSTD_rust_compressStream2MT_trace,
ZSTD_rust_compressStream2MT_reset
};
ZSTD_rust_compressStream2MTCompletionState const completionState = {
loopPolicy,
&callbacks
};
ZSTD_rust_compressStream2MTCompletion(&completionState);
}
FORWARD_IF_ERROR(flushMin, "ZSTDMT_compressStream_generic failed");
if (loopPolicy == ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_BREAK
rust/src/zstd_compress.rs
+183 -3
View File
@@ -448,9 +448,9 @@ pub unsafe extern "C" fn ZSTD_rust_compressStream2InitPolicy(
/// Scalar projection for the MT loop after a C-owned compression call.
///
/// Rust owns the result classification and progress/termination decision. C
/// retains the MT call, private progress counters, trace callback, reset, and
/// error encoding. The ordering matches `ZSTD_compressStream2_c`: errors win,
/// then a completed end directive, then the directive-specific progress rule.
/// retains the MT call, private progress counters, and error encoding. The
/// ordering matches `ZSTD_compressStream2_c`: errors win, then a completed end
/// directive, then the directive-specific progress rule.
#[repr(C)]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ZSTD_rust_compressStream2MTLoopPolicyState {
@@ -545,6 +545,84 @@ pub unsafe extern "C" fn ZSTD_rust_compressStream2MTLoopPolicy(
compress_stream2_mt_loop_policy(state)
}
type CompressStream2MTTraceFn = unsafe extern "C" fn(*mut c_void);
type CompressStream2MTResetFn = unsafe extern "C" fn(*mut c_void) -> usize;
/// Opaque C callbacks for the final MT stream completion actions.
///
/// The callback context is the private `ZSTD_CCtx`. Rust only sequences these
/// callbacks; C retains all context layout and mutation behind them.
#[repr(C)]
pub struct ZSTD_rust_compressStream2MTCompletionCallbacks {
callback_context: *mut c_void,
trace: CompressStream2MTTraceFn,
reset: CompressStream2MTResetFn,
}
/// Projected MT loop result and its opaque completion callbacks.
#[repr(C)]
pub struct ZSTD_rust_compressStream2MTCompletionState {
loop_policy: c_int,
callbacks: *const ZSTD_rust_compressStream2MTCompletionCallbacks,
}
const _: () = {
assert!(size_of::<CompressStream2MTTraceFn>() == size_of::<usize>());
assert!(size_of::<CompressStream2MTResetFn>() == size_of::<usize>());
assert!(
offset_of!(
ZSTD_rust_compressStream2MTCompletionCallbacks,
callback_context
) == 0
);
assert!(
offset_of!(ZSTD_rust_compressStream2MTCompletionCallbacks, trace) == size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_compressStream2MTCompletionCallbacks, reset) == 2 * size_of::<usize>()
);
assert!(size_of::<ZSTD_rust_compressStream2MTCompletionCallbacks>() == 3 * size_of::<usize>());
assert!(offset_of!(ZSTD_rust_compressStream2MTCompletionState, loop_policy) == 0);
assert!(
offset_of!(ZSTD_rust_compressStream2MTCompletionState, callbacks) == size_of::<usize>()
);
assert!(size_of::<ZSTD_rust_compressStream2MTCompletionState>() == 2 * size_of::<usize>());
};
/// Apply the final MT completion ordering without exposing the private C
/// context. Continue and break have no completion side effects. Worker errors
/// reset the session; a completed end traces first and then resets it.
#[inline]
unsafe fn compress_stream2_mt_completion(state: &ZSTD_rust_compressStream2MTCompletionState) {
let Some(callbacks) = (unsafe { state.callbacks.as_ref() }) else {
return;
};
match state.loop_policy {
ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_ERROR => {
let _ = unsafe { (callbacks.reset)(callbacks.callback_context) };
}
ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_END_COMPLETE => {
unsafe { (callbacks.trace)(callbacks.callback_context) };
let _ = unsafe { (callbacks.reset)(callbacks.callback_context) };
}
ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_CONTINUE
| ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_BREAK => {}
_ => {}
}
}
/// Apply MT completion side effects through opaque C callbacks.
#[no_mangle]
pub unsafe extern "C" fn ZSTD_rust_compressStream2MTCompletion(
state: *const ZSTD_rust_compressStream2MTCompletionState,
) {
let Some(state) = (unsafe { state.as_ref() }) else {
return;
};
unsafe { compress_stream2_mt_completion(state) };
}
const ZSTD_C_WINDOW_LOG: c_int = 101;
const ZSTD_C_HASH_LOG: c_int = 102;
const ZSTD_C_CHAIN_LOG: c_int = 103;
@@ -14782,6 +14860,108 @@ mod tests {
}
}
#[derive(Default)]
struct CompressStream2MTCompletionTestContext {
events: Vec<&'static str>,
}
unsafe extern "C" fn compress_stream2_mt_completion_test_trace(context: *mut c_void) {
let context = unsafe { &mut *context.cast::<CompressStream2MTCompletionTestContext>() };
context.events.push("trace");
}
unsafe extern "C" fn compress_stream2_mt_completion_test_reset(context: *mut c_void) -> usize {
let context = unsafe { &mut *context.cast::<CompressStream2MTCompletionTestContext>() };
context.events.push("reset");
/* The completion policy intentionally ignores this callback result. */
ERROR(ZstdErrorCode::Generic)
}
fn compress_stream2_mt_completion_test_state(
context: &mut CompressStream2MTCompletionTestContext,
loop_policy: c_int,
callbacks: &mut ZSTD_rust_compressStream2MTCompletionCallbacks,
) -> ZSTD_rust_compressStream2MTCompletionState {
callbacks.callback_context =
(context as *mut CompressStream2MTCompletionTestContext).cast();
ZSTD_rust_compressStream2MTCompletionState {
loop_policy,
callbacks: callbacks as *const _,
}
}
#[test]
fn compress_stream2_mt_completion_does_nothing_for_continue() {
let mut context = CompressStream2MTCompletionTestContext::default();
let mut callbacks = ZSTD_rust_compressStream2MTCompletionCallbacks {
callback_context: ptr::null_mut(),
trace: compress_stream2_mt_completion_test_trace,
reset: compress_stream2_mt_completion_test_reset,
};
let state = compress_stream2_mt_completion_test_state(
&mut context,
ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_CONTINUE,
&mut callbacks,
);
unsafe { ZSTD_rust_compressStream2MTCompletion(&state) };
assert!(context.events.is_empty());
}
#[test]
fn compress_stream2_mt_completion_does_nothing_for_break() {
let mut context = CompressStream2MTCompletionTestContext::default();
let mut callbacks = ZSTD_rust_compressStream2MTCompletionCallbacks {
callback_context: ptr::null_mut(),
trace: compress_stream2_mt_completion_test_trace,
reset: compress_stream2_mt_completion_test_reset,
};
let state = compress_stream2_mt_completion_test_state(
&mut context,
ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_BREAK,
&mut callbacks,
);
unsafe { ZSTD_rust_compressStream2MTCompletion(&state) };
assert!(context.events.is_empty());
}
#[test]
fn compress_stream2_mt_completion_resets_worker_errors() {
let mut context = CompressStream2MTCompletionTestContext::default();
let mut callbacks = ZSTD_rust_compressStream2MTCompletionCallbacks {
callback_context: ptr::null_mut(),
trace: compress_stream2_mt_completion_test_trace,
reset: compress_stream2_mt_completion_test_reset,
};
let state = compress_stream2_mt_completion_test_state(
&mut context,
ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_ERROR,
&mut callbacks,
);
unsafe { ZSTD_rust_compressStream2MTCompletion(&state) };
assert_eq!(context.events, ["reset"]);
}
#[test]
fn compress_stream2_mt_completion_traces_before_resetting_completed_end() {
let mut context = CompressStream2MTCompletionTestContext::default();
let mut callbacks = ZSTD_rust_compressStream2MTCompletionCallbacks {
callback_context: ptr::null_mut(),
trace: compress_stream2_mt_completion_test_trace,
reset: compress_stream2_mt_completion_test_reset,
};
let state = compress_stream2_mt_completion_test_state(
&mut context,
ZSTD_RUST_COMPRESS_STREAM2_MT_LOOP_POLICY_END_COMPLETE,
&mut callbacks,
);
unsafe { ZSTD_rust_compressStream2MTCompletion(&state) };
assert_eq!(context.events, ["trace", "reset"]);
}
fn compress_stream2_init_policy_state(
in_buffer_mode: c_int,
end_op: c_int,