diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 541f508d3..0420151ee 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -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 diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index 3770c2551..f276cfb65 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -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::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!( + offset_of!( + ZSTD_rust_compressStream2MTCompletionCallbacks, + callback_context + ) == 0 + ); + assert!( + offset_of!(ZSTD_rust_compressStream2MTCompletionCallbacks, trace) == size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressStream2MTCompletionCallbacks, reset) == 2 * size_of::() + ); + assert!(size_of::() == 3 * size_of::()); + assert!(offset_of!(ZSTD_rust_compressStream2MTCompletionState, loop_policy) == 0); + assert!( + offset_of!(ZSTD_rust_compressStream2MTCompletionState, callbacks) == size_of::() + ); + assert!(size_of::() == 2 * size_of::()); +}; + +/// 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::() }; + 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::() }; + 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,