From 1cf5a92e023ccc03a6b58097ad5f03df1d447d7c Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sun, 19 Jul 2026 13:22:25 +0200 Subject: [PATCH] feat(compress): move CCtx reset policy into Rust Move ZSTD_CCtx_reset directive handling into a Rust-owned state machine while keeping dictionary cleanup and parameter reset behind C callbacks. Preserve the original session ordering, stage error boundary, completion/max-block flag updates, and reset error propagation across the ABI bridge. Test Plan: - cargo test --manifest-path rust/Cargo.toml --lib - cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings - make -B -C programs -j1 zstd - focused reset_cctx unit tests --- lib/compress/zstd_compress.c | 64 ++++++--- rust/src/zstd_compress.rs | 242 +++++++++++++++++++++++++++++++++++ 2 files changed, 291 insertions(+), 15 deletions(-) diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 22cf70830..31748ca66 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -247,6 +247,34 @@ typedef char ZSTD_rust_set_params_state_layout[ && offsetof(ZSTD_rust_setParamsState, setCParams) == 3 * sizeof(void*) && sizeof(ZSTD_rust_setParamsState) == 4 * sizeof(void*)) ? 1 : -1]; +typedef void (*ZSTD_rust_resetCCtxClearAllDicts_f)(void* context); +typedef size_t (*ZSTD_rust_resetCCtxResetParams_f)(void* context); +typedef struct { + void* callbackContext; + unsigned* rustSimpleCompress2Completed; + ZSTD_cStreamStage* streamStage; + unsigned long long* pledgedSrcSizePlusOne; + unsigned* rustSimpleCompress2MaxBlockSizeSet; + ZSTD_rust_resetCCtxClearAllDicts_f clearAllDicts; + ZSTD_rust_resetCCtxResetParams_f resetParams; +} ZSTD_rust_resetCCtxState; +size_t ZSTD_rust_resetCCtx(const ZSTD_rust_resetCCtxState* state, int reset); +typedef char ZSTD_rust_reset_cctx_state_layout[ + (offsetof(ZSTD_rust_resetCCtxState, callbackContext) == 0 + && offsetof(ZSTD_rust_resetCCtxState, rustSimpleCompress2Completed) + == sizeof(void*) + && offsetof(ZSTD_rust_resetCCtxState, streamStage) + == 2 * sizeof(void*) + && offsetof(ZSTD_rust_resetCCtxState, pledgedSrcSizePlusOne) + == 3 * sizeof(void*) + && offsetof(ZSTD_rust_resetCCtxState, rustSimpleCompress2MaxBlockSizeSet) + == 4 * sizeof(void*) + && offsetof(ZSTD_rust_resetCCtxState, clearAllDicts) + == 5 * sizeof(void*) + && offsetof(ZSTD_rust_resetCCtxState, resetParams) + == 6 * sizeof(void*) + && sizeof(ZSTD_rust_resetCCtxState) == 7 * sizeof(void*)) + ? 1 : -1]; ZSTD_frameProgression ZSTD_rust_frameProgression(U64 consumedSrcSize, size_t buffered, U64 producedCSize); @@ -2099,25 +2127,31 @@ size_t ZSTD_CCtx_refPrefix_advanced( ZSTD_clearAllDicts_callback, ZSTD_assignPrefixDict_callback); } +static void ZSTD_rust_resetCCtx_clearAllDicts(void* context) +{ + ZSTD_clearAllDicts((ZSTD_CCtx*)context); +} + +static size_t ZSTD_rust_resetCCtx_resetParams(void* context) +{ + ZSTD_CCtx* const cctx = (ZSTD_CCtx*)context; + return ZSTD_CCtxParams_reset(&cctx->requestedParams); +} + /*! ZSTD_CCtx_reset() : * Also dumps dictionary */ size_t ZSTD_CCtx_reset(ZSTD_CCtx* cctx, ZSTD_ResetDirective reset) { - cctx->rustSimpleCompress2Completed = 0; - if ( (reset == ZSTD_reset_session_only) - || (reset == ZSTD_reset_session_and_parameters) ) { - cctx->streamStage = zcss_init; - cctx->pledgedSrcSizePlusOne = 0; - } - if ( (reset == ZSTD_reset_parameters) - || (reset == ZSTD_reset_session_and_parameters) ) { - RETURN_ERROR_IF(cctx->streamStage != zcss_init, stage_wrong, - "Reset parameters is only possible during init stage."); - ZSTD_clearAllDicts(cctx); - cctx->rustSimpleCompress2MaxBlockSizeSet = 0; - return ZSTD_CCtxParams_reset(&cctx->requestedParams); - } - return 0; + ZSTD_rust_resetCCtxState state; + state.callbackContext = cctx; + state.rustSimpleCompress2Completed = &cctx->rustSimpleCompress2Completed; + state.streamStage = &cctx->streamStage; + state.pledgedSrcSizePlusOne = &cctx->pledgedSrcSizePlusOne; + state.rustSimpleCompress2MaxBlockSizeSet = + &cctx->rustSimpleCompress2MaxBlockSizeSet; + state.clearAllDicts = ZSTD_rust_resetCCtx_clearAllDicts; + state.resetParams = ZSTD_rust_resetCCtx_resetParams; + return ZSTD_rust_resetCCtx(&state, (int)reset); } diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index 8207db998..f982dc70b 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -123,6 +123,9 @@ const ZSTD_C_STRATEGY: c_int = 107; const ZSTD_C_CONTENT_SIZE_FLAG: c_int = 200; const ZSTD_C_CHECKSUM_FLAG: c_int = 201; const ZSTD_C_DICT_ID_FLAG: c_int = 202; +const ZSTD_RESET_SESSION_ONLY: c_int = 1; +const ZSTD_RESET_PARAMETERS: c_int = 2; +const ZSTD_RESET_SESSION_AND_PARAMETERS: c_int = 3; const ZSTD_CSTREAM_STAGE_LOAD: c_int = 1; const ZSTD_CSTREAM_STAGE_FLUSH: c_int = 2; const ZSTD_BSS_COMPRESS: c_int = 0; @@ -814,6 +817,80 @@ pub unsafe extern "C" fn ZSTD_rust_compress2( unsafe { compress2_body_with(&*state, dst, dst_capacity, src, src_size) } } +type ResetCCtxClearAllDictsFn = unsafe extern "C" fn(*mut c_void); +type ResetCCtxResetParamsFn = unsafe extern "C" fn(*mut c_void) -> usize; + +/// Explicit projection for the public `ZSTD_CCtx_reset` wrapper. +/// +/// Rust owns the reset-directive policy and ordering. The callback context +/// remains in C, where the dictionary and parameter reset operations retain +/// access to the private `ZSTD_CCtx` layout. +#[repr(C)] +pub struct ZSTD_rust_resetCCtxState { + callback_context: *mut c_void, + rust_simple_compress2_completed: *mut c_uint, + stream_stage: *mut c_int, + pledged_src_size_plus_one: *mut u64, + rust_simple_compress2_max_block_size_set: *mut c_uint, + clear_all_dicts: ResetCCtxClearAllDictsFn, + reset_params: ResetCCtxResetParamsFn, +} + +const _: () = { + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!(offset_of!(ZSTD_rust_resetCCtxState, callback_context) == 0); + assert!( + offset_of!(ZSTD_rust_resetCCtxState, rust_simple_compress2_completed) == size_of::() + ); + assert!(offset_of!(ZSTD_rust_resetCCtxState, stream_stage) == 2 * size_of::()); + assert!( + offset_of!(ZSTD_rust_resetCCtxState, pledged_src_size_plus_one) == 3 * size_of::() + ); + assert!( + offset_of!( + ZSTD_rust_resetCCtxState, + rust_simple_compress2_max_block_size_set + ) == 4 * size_of::() + ); + assert!(offset_of!(ZSTD_rust_resetCCtxState, clear_all_dicts) == 5 * size_of::()); + assert!(offset_of!(ZSTD_rust_resetCCtxState, reset_params) == 6 * size_of::()); + assert!(size_of::() == 7 * size_of::()); +}; + +/// Drive the public `ZSTD_CCtx_reset` policy through C-owned callbacks. +#[no_mangle] +pub unsafe extern "C" fn ZSTD_rust_resetCCtx( + state: *const ZSTD_rust_resetCCtxState, + reset: c_int, +) -> usize { + if state.is_null() { + return ERROR(ZstdErrorCode::Generic); + } + let state = unsafe { &*state }; + + unsafe { + *state.rust_simple_compress2_completed = 0; + } + if reset == ZSTD_RESET_SESSION_ONLY || reset == ZSTD_RESET_SESSION_AND_PARAMETERS { + unsafe { + *state.stream_stage = ZSTD_CSTREAM_STAGE_INIT; + *state.pledged_src_size_plus_one = 0; + } + } + if reset == ZSTD_RESET_PARAMETERS || reset == ZSTD_RESET_SESSION_AND_PARAMETERS { + if unsafe { *state.stream_stage } != ZSTD_CSTREAM_STAGE_INIT { + return ERROR(ZstdErrorCode::StageWrong); + } + unsafe { (state.clear_all_dicts)(state.callback_context) }; + unsafe { + *state.rust_simple_compress2_max_block_size_set = 0; + } + return unsafe { (state.reset_params)(state.callback_context) }; + } + 0 +} + type ResetCStreamResetFn = unsafe extern "C" fn(*mut c_void) -> usize; type ResetCStreamSetPledgedSrcSizeFn = unsafe extern "C" fn(*mut c_void, u64) -> usize; @@ -9116,6 +9193,171 @@ mod tests { assert_eq!(dict_limit, 0x89ab_cdef); } + #[derive(Default)] + struct ResetCCtxTestContext { + events: Vec<&'static str>, + rust_simple_compress2_completed: c_uint, + stream_stage: c_int, + pledged_src_size_plus_one: u64, + rust_simple_compress2_max_block_size_set: c_uint, + reset_params_result: usize, + } + + unsafe fn reset_cctx_test_context(context: *mut c_void) -> &'static mut ResetCCtxTestContext { + unsafe { &mut *context.cast::() } + } + + unsafe extern "C" fn reset_cctx_test_clear_all_dicts(context: *mut c_void) { + unsafe { reset_cctx_test_context(context) } + .events + .push("clear-dicts"); + } + + unsafe extern "C" fn reset_cctx_test_reset_params(context: *mut c_void) -> usize { + let context = unsafe { reset_cctx_test_context(context) }; + assert_eq!(context.rust_simple_compress2_max_block_size_set, 0); + context.events.push("reset-params"); + context.reset_params_result + } + + fn reset_cctx_test_state(context: &mut ResetCCtxTestContext) -> ZSTD_rust_resetCCtxState { + ZSTD_rust_resetCCtxState { + callback_context: (context as *mut ResetCCtxTestContext).cast(), + rust_simple_compress2_completed: &mut context.rust_simple_compress2_completed, + stream_stage: &mut context.stream_stage, + pledged_src_size_plus_one: &mut context.pledged_src_size_plus_one, + rust_simple_compress2_max_block_size_set: &mut context + .rust_simple_compress2_max_block_size_set, + clear_all_dicts: reset_cctx_test_clear_all_dicts, + reset_params: reset_cctx_test_reset_params, + } + } + + #[test] + fn reset_cctx_session_only_clears_session_state() { + let mut context = ResetCCtxTestContext { + rust_simple_compress2_completed: 1, + stream_stage: ZSTD_CSTREAM_STAGE_LOAD, + pledged_src_size_plus_one: 123, + rust_simple_compress2_max_block_size_set: 1, + ..ResetCCtxTestContext::default() + }; + let state = reset_cctx_test_state(&mut context); + + let result = unsafe { ZSTD_rust_resetCCtx(&state, ZSTD_RESET_SESSION_ONLY) }; + + assert_eq!(result, 0); + assert_eq!(context.rust_simple_compress2_completed, 0); + assert_eq!(context.stream_stage, ZSTD_CSTREAM_STAGE_INIT); + assert_eq!(context.pledged_src_size_plus_one, 0); + assert_eq!(context.rust_simple_compress2_max_block_size_set, 1); + assert!(context.events.is_empty()); + } + + #[test] + fn reset_cctx_parameters_runs_callbacks_in_order_and_clears_max_block_flag() { + let mut context = ResetCCtxTestContext { + rust_simple_compress2_completed: 1, + stream_stage: ZSTD_CSTREAM_STAGE_INIT, + pledged_src_size_plus_one: 123, + rust_simple_compress2_max_block_size_set: 1, + ..ResetCCtxTestContext::default() + }; + let state = reset_cctx_test_state(&mut context); + + let result = unsafe { ZSTD_rust_resetCCtx(&state, ZSTD_RESET_PARAMETERS) }; + + assert_eq!(result, 0); + assert_eq!(context.rust_simple_compress2_completed, 0); + assert_eq!(context.stream_stage, ZSTD_CSTREAM_STAGE_INIT); + assert_eq!(context.pledged_src_size_plus_one, 123); + assert_eq!(context.rust_simple_compress2_max_block_size_set, 0); + assert_eq!(context.events, ["clear-dicts", "reset-params"]); + } + + #[test] + fn reset_cctx_session_and_parameters_resets_session_before_callbacks() { + let mut context = ResetCCtxTestContext { + rust_simple_compress2_completed: 1, + stream_stage: ZSTD_CSTREAM_STAGE_LOAD, + pledged_src_size_plus_one: 123, + rust_simple_compress2_max_block_size_set: 1, + ..ResetCCtxTestContext::default() + }; + let state = reset_cctx_test_state(&mut context); + + let result = unsafe { ZSTD_rust_resetCCtx(&state, ZSTD_RESET_SESSION_AND_PARAMETERS) }; + + assert_eq!(result, 0); + assert_eq!(context.rust_simple_compress2_completed, 0); + assert_eq!(context.stream_stage, ZSTD_CSTREAM_STAGE_INIT); + assert_eq!(context.pledged_src_size_plus_one, 0); + assert_eq!(context.rust_simple_compress2_max_block_size_set, 0); + assert_eq!(context.events, ["clear-dicts", "reset-params"]); + } + + #[test] + fn reset_cctx_parameters_rejects_non_init_stage_before_mutating_parameters() { + let mut context = ResetCCtxTestContext { + rust_simple_compress2_completed: 1, + stream_stage: ZSTD_CSTREAM_STAGE_LOAD, + pledged_src_size_plus_one: 123, + rust_simple_compress2_max_block_size_set: 1, + ..ResetCCtxTestContext::default() + }; + let state = reset_cctx_test_state(&mut context); + + let result = unsafe { ZSTD_rust_resetCCtx(&state, ZSTD_RESET_PARAMETERS) }; + + assert_eq!(result, ERROR(ZstdErrorCode::StageWrong)); + assert_eq!(context.rust_simple_compress2_completed, 0); + assert_eq!(context.stream_stage, ZSTD_CSTREAM_STAGE_LOAD); + assert_eq!(context.pledged_src_size_plus_one, 123); + assert_eq!(context.rust_simple_compress2_max_block_size_set, 1); + assert!(context.events.is_empty()); + } + + #[test] + fn reset_cctx_propagates_parameter_reset_error_after_clearing_state() { + let mut context = ResetCCtxTestContext { + rust_simple_compress2_completed: 1, + stream_stage: ZSTD_CSTREAM_STAGE_INIT, + pledged_src_size_plus_one: 123, + rust_simple_compress2_max_block_size_set: 1, + reset_params_result: ERROR(ZstdErrorCode::MemoryAllocation), + ..ResetCCtxTestContext::default() + }; + let state = reset_cctx_test_state(&mut context); + + let result = unsafe { ZSTD_rust_resetCCtx(&state, ZSTD_RESET_PARAMETERS) }; + + assert_eq!(result, ERROR(ZstdErrorCode::MemoryAllocation)); + assert_eq!(context.rust_simple_compress2_completed, 0); + assert_eq!(context.rust_simple_compress2_max_block_size_set, 0); + assert_eq!(context.events, ["clear-dicts", "reset-params"]); + } + + #[test] + fn reset_cctx_ignores_unknown_directives_after_clearing_completion() { + let mut context = ResetCCtxTestContext { + rust_simple_compress2_completed: 1, + stream_stage: ZSTD_CSTREAM_STAGE_LOAD, + pledged_src_size_plus_one: 123, + rust_simple_compress2_max_block_size_set: 1, + ..ResetCCtxTestContext::default() + }; + let state = reset_cctx_test_state(&mut context); + + let result = unsafe { ZSTD_rust_resetCCtx(&state, 99) }; + + assert_eq!(result, 0); + assert_eq!(context.rust_simple_compress2_completed, 0); + assert_eq!(context.stream_stage, ZSTD_CSTREAM_STAGE_LOAD); + assert_eq!(context.pledged_src_size_plus_one, 123); + assert_eq!(context.rust_simple_compress2_max_block_size_set, 1); + assert!(context.events.is_empty()); + } + #[derive(Default)] struct ResetCStreamTestContext { events: Vec<&'static str>,