From fe6cf3bccc3dc7bafe528436bd844555411b044a Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sun, 19 Jul 2026 11:46:18 +0200 Subject: [PATCH] feat(compress): move resetCStream policy into Rust Move the public CStream reset wrapper's pledge normalization, callback order, and error short-circuiting into Rust. Keep the C-owned context reset and pledged-size mutation behind opaque callbacks with an explicit ABI projection. Test Plan: - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml reset_cstream -- --test-threads=1 - ulimit -v 41943040; make -B -C programs -j1 zstd - git diff --cached --check --- lib/compress/zstd_compress.c | 42 ++++++++--- rust/src/zstd_compress.rs | 142 +++++++++++++++++++++++++++++++++++ 2 files changed, 175 insertions(+), 9 deletions(-) diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 259a2dbbe..e25bb2c18 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -51,6 +51,23 @@ int ZSTD_rust_updateFrameProgression(unsigned long long* consumedSrcSize, size_t ZSTD_rust_setPledgedSrcSize(int streamStage, unsigned long long pledgedSrcSize, unsigned long long* pledgedSrcSizePlusOne); +typedef size_t (*ZSTD_rust_resetCStreamReset_f)(void* context); +typedef size_t (*ZSTD_rust_resetCStreamSetPledgedSrcSize_f)( + void* context, unsigned long long pledgedSrcSize); +typedef struct { + void* callbackContext; + ZSTD_rust_resetCStreamReset_f resetSession; + ZSTD_rust_resetCStreamSetPledgedSrcSize_f setPledgedSrcSize; +} ZSTD_rust_resetCStreamState; +size_t ZSTD_rust_resetCStream(const ZSTD_rust_resetCStreamState* state, + unsigned long long pss); +typedef char ZSTD_rust_reset_cstream_state_layout[ + (offsetof(ZSTD_rust_resetCStreamState, callbackContext) == 0 + && offsetof(ZSTD_rust_resetCStreamState, resetSession) == sizeof(void*) + && offsetof(ZSTD_rust_resetCStreamState, setPledgedSrcSize) + == 2 * sizeof(void*) + && sizeof(ZSTD_rust_resetCStreamState) == 3 * sizeof(void*)) + ? 1 : -1]; ZSTD_frameProgression ZSTD_rust_frameProgression(U64 consumedSrcSize, size_t buffered, U64 producedCSize); @@ -4770,17 +4787,24 @@ static ZSTD_CParamMode_e ZSTD_getCParamMode(ZSTD_CDict const* cdict, ZSTD_CCtx_p /* ZSTD_resetCStream(): * pledgedSrcSize == 0 means "unknown" */ +static size_t ZSTD_rust_resetCStream_resetSession(void* context) +{ + return ZSTD_CCtx_reset((ZSTD_CCtx*)context, ZSTD_reset_session_only); +} + +static size_t ZSTD_rust_resetCStream_setPledgedSrcSize( + void* context, unsigned long long pledgedSrcSize) +{ + return ZSTD_CCtx_setPledgedSrcSize((ZSTD_CCtx*)context, pledgedSrcSize); +} + size_t ZSTD_resetCStream(ZSTD_CStream* zcs, unsigned long long pss) { - /* temporary : 0 interpreted as "unknown" during transition period. - * Users willing to specify "unknown" **must** use ZSTD_CONTENTSIZE_UNKNOWN. - * 0 will be interpreted as "empty" in the future. - */ - U64 const pledgedSrcSize = (pss==0) ? ZSTD_CONTENTSIZE_UNKNOWN : pss; - DEBUGLOG(4, "ZSTD_resetCStream: pledgedSrcSize = %u", (unsigned)pledgedSrcSize); - FORWARD_IF_ERROR( ZSTD_CCtx_reset(zcs, ZSTD_reset_session_only) , ""); - FORWARD_IF_ERROR( ZSTD_CCtx_setPledgedSrcSize(zcs, pledgedSrcSize) , ""); - return 0; + ZSTD_rust_resetCStreamState state; + state.callbackContext = zcs; + state.resetSession = ZSTD_rust_resetCStream_resetSession; + state.setPledgedSrcSize = ZSTD_rust_resetCStream_setPledgedSrcSize; + return ZSTD_rust_resetCStream(&state, pss); } /*! ZSTD_initCStream_internal() : diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index f36deff69..a8334d59f 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -803,6 +803,60 @@ pub unsafe extern "C" fn ZSTD_rust_compress2( unsafe { compress2_body_with(&*state, dst, dst_capacity, src, src_size) } } +type ResetCStreamResetFn = unsafe extern "C" fn(*mut c_void) -> usize; +type ResetCStreamSetPledgedSrcSizeFn = unsafe extern "C" fn(*mut c_void, u64) -> usize; + +/// Explicit projection for the public `ZSTD_resetCStream` wrapper. +/// +/// Rust owns the zero-to-unknown conversion, callback ordering, and error +/// propagation. The opaque callback context stays in C, where the callbacks +/// retain access to the private `ZSTD_CCtx` layout. +#[repr(C)] +pub struct ZSTD_rust_resetCStreamState { + callback_context: *mut c_void, + reset_session: ResetCStreamResetFn, + set_pledged_src_size: ResetCStreamSetPledgedSrcSizeFn, +} + +const _: () = { + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); + assert!(offset_of!(ZSTD_rust_resetCStreamState, callback_context) == 0); + assert!(offset_of!(ZSTD_rust_resetCStreamState, reset_session) == size_of::()); + assert!( + offset_of!(ZSTD_rust_resetCStreamState, set_pledged_src_size) == 2 * size_of::() + ); + assert!(size_of::() == 3 * size_of::()); +}; + +/// Drive the public `ZSTD_resetCStream` policy through C-owned callbacks. +#[no_mangle] +pub unsafe extern "C" fn ZSTD_rust_resetCStream( + state: *const ZSTD_rust_resetCStreamState, + pss: u64, +) -> usize { + if state.is_null() { + return ERROR(ZstdErrorCode::Generic); + } + let state = unsafe { &*state }; + let pledged_src_size = if pss == 0 { + ZSTD_CONTENTSIZE_UNKNOWN + } else { + pss + }; + + let reset_result = unsafe { (state.reset_session)(state.callback_context) }; + if ERR_isError(reset_result) { + return reset_result; + } + let pledged_size_result = + unsafe { (state.set_pledged_src_size)(state.callback_context, pledged_src_size) }; + if ERR_isError(pledged_size_result) { + return pledged_size_result; + } + 0 +} + type CompressStreamBlockFn = unsafe extern "C" fn(*mut c_void, *mut c_void, usize, *const c_void, usize) -> usize; type CompressStreamResetFn = unsafe extern "C" fn(*mut c_void) -> usize; @@ -8559,6 +8613,94 @@ mod tests { assert_eq!(dict_limit, 0x89ab_cdef); } + #[derive(Default)] + struct ResetCStreamTestContext { + events: Vec<&'static str>, + reset_result: usize, + pledged_result: usize, + pledged_src_size: u64, + } + + unsafe extern "C" fn reset_cstream_test_reset(context: *mut c_void) -> usize { + let context = unsafe { &mut *context.cast::() }; + context.events.push("reset"); + context.reset_result + } + + unsafe extern "C" fn reset_cstream_test_set_pledged_src_size( + context: *mut c_void, + pledged_src_size: u64, + ) -> usize { + let context = unsafe { &mut *context.cast::() }; + context.events.push("pledged"); + context.pledged_src_size = pledged_src_size; + context.pledged_result + } + + fn reset_cstream_test_state( + context: &mut ResetCStreamTestContext, + ) -> ZSTD_rust_resetCStreamState { + ZSTD_rust_resetCStreamState { + callback_context: (context as *mut ResetCStreamTestContext).cast(), + reset_session: reset_cstream_test_reset, + set_pledged_src_size: reset_cstream_test_set_pledged_src_size, + } + } + + #[test] + fn reset_cstream_converts_zero_pledge_to_unknown_after_reset() { + let mut context = ResetCStreamTestContext::default(); + let state = reset_cstream_test_state(&mut context); + + let result = unsafe { ZSTD_rust_resetCStream(&state, 0) }; + + assert_eq!(result, 0); + assert_eq!(context.events, ["reset", "pledged"]); + assert_eq!(context.pledged_src_size, ZSTD_CONTENTSIZE_UNKNOWN); + } + + #[test] + fn reset_cstream_forwards_nonzero_pledge() { + let mut context = ResetCStreamTestContext::default(); + let state = reset_cstream_test_state(&mut context); + + let result = unsafe { ZSTD_rust_resetCStream(&state, 123) }; + + assert_eq!(result, 0); + assert_eq!(context.events, ["reset", "pledged"]); + assert_eq!(context.pledged_src_size, 123); + } + + #[test] + fn reset_cstream_stops_before_pledge_after_reset_error() { + let mut context = ResetCStreamTestContext { + reset_result: ERROR(ZstdErrorCode::MemoryAllocation), + ..ResetCStreamTestContext::default() + }; + let state = reset_cstream_test_state(&mut context); + + let result = unsafe { ZSTD_rust_resetCStream(&state, 123) }; + + assert_eq!(result, ERROR(ZstdErrorCode::MemoryAllocation)); + assert_eq!(context.events, ["reset"]); + assert_eq!(context.pledged_src_size, 0); + } + + #[test] + fn reset_cstream_propagates_pledge_error_after_reset() { + let mut context = ResetCStreamTestContext { + pledged_result: ERROR(ZstdErrorCode::StageWrong), + ..ResetCStreamTestContext::default() + }; + let state = reset_cstream_test_state(&mut context); + + let result = unsafe { ZSTD_rust_resetCStream(&state, 123) }; + + assert_eq!(result, ERROR(ZstdErrorCode::StageWrong)); + assert_eq!(context.events, ["reset", "pledged"]); + assert_eq!(context.pledged_src_size, 123); + } + #[test] fn pledged_src_size_writes_the_init_stage_value_plus_one() { let mut pledged_src_size_plus_one = 0;