From 6da0296af63a218d0907104515f875e0b9dbb599 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Mon, 20 Jul 2026 01:06:14 +0200 Subject: [PATCH] feat(compress): project stream continue states into Rust Pass the shared continue and end projections directly through the buffered stream ABI so Rust owns stream dispatch into the migrated compression paths. Remove the redundant C continue/end forwarding callbacks while retaining the C reset callback for private CCtx session state. Test Plan: - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo fmt --all -- --check - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --lib zstd_compress::tests::compress_stream -- --nocapture - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --all-targets -- -D warnings - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test - ulimit -v 41943040; make -j1 - ulimit -v 41943040; make -j1 -C tests test-zstream ZSTREAM_TESTTIME=-T2s - ulimit -v 41943040; make -j1 -C tests test-fuzzer FUZZERTEST=-T3s FUZZER_FLAGS=--no-big-tests --- lib/compress/zstd_compress.c | 48 ++++---- rust/src/zstd_compress.rs | 233 ++++++++++++++++++++++++++++++++--- 2 files changed, 239 insertions(+), 42 deletions(-) diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 715295b78..2b897631d 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -1007,9 +1007,6 @@ typedef char ZSTD_rust_compress_end_state_layout[ /* Rust owns the single-threaded buffered/stable stream state machine. The * projection contains only stream bookkeeping and callback slots; operations * which still need the private CCtx layout remain C callbacks. */ -typedef size_t (*ZSTD_rust_compressStreamBlock_f)( - void* context, void* dst, size_t dstCapacity, - const void* src, size_t srcSize); typedef size_t (*ZSTD_rust_compressStreamReset_f)(void* context); typedef struct { void* callbackContext; @@ -1028,8 +1025,8 @@ typedef struct { size_t* outBuffContentSize; size_t* outBuffFlushedSize; U32* frameEnded; - ZSTD_rust_compressStreamBlock_f compressContinue; - ZSTD_rust_compressStreamBlock_f compressEnd; + const ZSTD_rust_compressContinueState* compressContinueState; + const ZSTD_rust_compressEndState* compressEndState; ZSTD_rust_compressStreamReset_f resetSession; } ZSTD_rust_compressStreamState; size_t ZSTD_rust_compressStreamGeneric( @@ -1066,9 +1063,9 @@ typedef char ZSTD_rust_compress_stream_state_layout[ == 10 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t) && offsetof(ZSTD_rust_compressStreamState, frameEnded) == 11 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t) - && offsetof(ZSTD_rust_compressStreamState, compressContinue) + && offsetof(ZSTD_rust_compressStreamState, compressContinueState) == 12 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t) - && offsetof(ZSTD_rust_compressStreamState, compressEnd) + && offsetof(ZSTD_rust_compressStreamState, compressEndState) == 13 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t) && offsetof(ZSTD_rust_compressStreamState, resetSession) == 14 * sizeof(void*) + 2 * sizeof(int) + 2 * sizeof(size_t) @@ -7233,22 +7230,6 @@ static size_t ZSTD_nextInputSizeHint(const ZSTD_CCtx* cctx) /** ZSTD_compressStream_generic(): * internal function for all *compressStream*() variants * @return : hint size for next input to complete ongoing block */ -static size_t ZSTD_rust_compressStream_continue( - void* context, void* dst, size_t dstCapacity, - const void* src, size_t srcSize) -{ - return ZSTD_compressContinue_public( - (ZSTD_CCtx*)context, dst, dstCapacity, src, srcSize); -} - -static size_t ZSTD_rust_compressStream_end( - void* context, void* dst, size_t dstCapacity, - const void* src, size_t srcSize) -{ - return ZSTD_compressEnd_public( - (ZSTD_CCtx*)context, dst, dstCapacity, src, srcSize); -} - static size_t ZSTD_rust_compressStream_reset(void* context) { return ZSTD_CCtx_reset((ZSTD_CCtx*)context, ZSTD_reset_session_only); @@ -7259,7 +7240,24 @@ static size_t ZSTD_compressStream_generic(ZSTD_CStream* zcs, ZSTD_inBuffer* input, ZSTD_EndDirective const flushMode) { + ZSTD_rust_compressContinueContext continueContext; + ZSTD_rust_compressEndState endState; ZSTD_rust_compressStreamState state; + ZSTD_CCtx* const cctx = (ZSTD_CCtx*)zcs; + ZSTD_compressContinue_prepare(cctx, cctx->blockSizeMax, + 0 /* block size already selected */, &continueContext); + endState.callbackContext = zcs; + endState.compressContinueState = &continueContext.state; + endState.trace = ZSTD_rust_compressEnd_trace; + endState.consumedSrcSize = &cctx->consumedSrcSize; + endState.pledgedSrcSizePlusOne = cctx->pledgedSrcSizePlusOne; + endState.contentSizeFlag = cctx->appliedParams.fParams.contentSizeFlag; + endState.stage = (int*)&cctx->stage; + endState.noDictIDFlag = cctx->appliedParams.fParams.noDictIDFlag; + endState.checksumFlag = cctx->appliedParams.fParams.checksumFlag; + endState.format = (int)cctx->appliedParams.format; + endState.windowLog = cctx->appliedParams.cParams.windowLog; + endState.checksumState = &cctx->xxhState; state.callbackContext = zcs; state.inBufferMode = (int)zcs->appliedParams.inBufferMode; state.outBufferMode = (int)zcs->appliedParams.outBufferMode; @@ -7276,8 +7274,8 @@ static size_t ZSTD_compressStream_generic(ZSTD_CStream* zcs, state.outBuffContentSize = &zcs->outBuffContentSize; state.outBuffFlushedSize = &zcs->outBuffFlushedSize; state.frameEnded = &zcs->frameEnded; - state.compressContinue = ZSTD_rust_compressStream_continue; - state.compressEnd = ZSTD_rust_compressStream_end; + state.compressContinueState = &continueContext.state; + state.compressEndState = &endState; state.resetSession = ZSTD_rust_compressStream_reset; return ZSTD_rust_compressStreamGeneric( &state, output, input, (int)flushMode); diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index 2a6ca5b0c..4d3db5baf 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -2914,16 +2914,15 @@ pub unsafe extern "C" fn ZSTD_rust_setParams( 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; /// Explicit projection of the single-threaded `ZSTD_compressStream_generic` /// state machine. /// /// Rust owns buffering, output-drain policy, directive handling, and stream -/// stage transitions. The opaque callback context stays in C; callbacks -/// retain access to `ZSTD_CCtx`, frame construction, and session reset. +/// stage transitions. Continue/end state projections retain the migrated +/// frame construction path; the opaque callback context remains only for +/// session reset. #[repr(C)] pub struct ZSTD_rust_compressStreamState { callback_context: *mut c_void, @@ -2942,8 +2941,8 @@ pub struct ZSTD_rust_compressStreamState { out_buff_content_size: *mut usize, out_buff_flushed_size: *mut usize, frame_ended: *mut c_uint, - compress_continue: CompressStreamBlockFn, - compress_end: CompressStreamBlockFn, + compress_continue_state: *const ZSTD_rust_compressContinueState, + compress_end_state: *const ZSTD_rust_compressEndState, reset_session: CompressStreamResetFn, } @@ -3007,11 +3006,11 @@ const _: () = { == 11 * size_of::() + 2 * size_of::() + 2 * size_of::() ); assert!( - offset_of!(ZSTD_rust_compressStreamState, compress_continue) + offset_of!(ZSTD_rust_compressStreamState, compress_continue_state) == 12 * size_of::() + 2 * size_of::() + 2 * size_of::() ); assert!( - offset_of!(ZSTD_rust_compressStreamState, compress_end) + offset_of!(ZSTD_rust_compressStreamState, compress_end_state) == 13 * size_of::() + 2 * size_of::() + 2 * size_of::() ); assert!( @@ -3441,6 +3440,8 @@ unsafe fn compress_stream_generic_body_with( || state.out_buff_content_size.is_null() || state.out_buff_flushed_size.is_null() || state.frame_ended.is_null() + || state.compress_continue_state.is_null() + || state.compress_end_state.is_null() { return ERROR(ZstdErrorCode::Generic); } @@ -3488,8 +3489,8 @@ unsafe fn compress_stream_generic_body_with( unsafe { input.src.cast::().add(input.pos).cast() } }; let c_size = unsafe { - (state.compress_end)( - state.callback_context, + ZSTD_rust_compressEnd( + state.compress_end_state, dst, output_remaining, src, @@ -3591,20 +3592,22 @@ unsafe fn compress_stream_generic_body_with( }; let c_size = unsafe { if last_block { - (state.compress_end)( - state.callback_context, + ZSTD_rust_compressEnd( + state.compress_end_state, output_dst, output_size, source, input_size, ) } else { - (state.compress_continue)( - state.callback_context, + ZSTD_rust_compressContinue( + state.compress_continue_state, output_dst, output_size, source, input_size, + 1, + 0, ) } }; @@ -11762,6 +11765,22 @@ mod tests { continue_result: usize, end_result: usize, reset_result: usize, + window: ZSTD_rust_windowUpdateState, + force_non_contiguous: c_int, + next_to_update: c_uint, + window_projection: ZSTD_rust_compressContinueWindowProjection, + frame_chunk_clamp_state: Option, + frame_chunk_prepare_state: Option, + continue_frame_chunk_state: Option, + end_frame_chunk_state: Option, + continue_state: Option, + end_continue_state: Option, + end_state: Option, + compression_stage: c_int, + is_first_block: c_int, + end_is_first_block: c_int, + consumed_src_size: u64, + produced_c_size: u64, } unsafe fn compress_stream_test_context( @@ -11776,6 +11795,7 @@ mod tests { dst_capacity: usize, _src: *const c_void, _src_size: usize, + _last_block: c_uint, ) -> usize { let context = unsafe { compress_stream_test_context(context) }; context.continue_calls += 1; @@ -11792,6 +11812,7 @@ mod tests { dst_capacity: usize, _src: *const c_void, _src_size: usize, + _last_block: c_uint, ) -> usize { let context = unsafe { compress_stream_test_context(context) }; context.end_calls += 1; @@ -11808,6 +11829,23 @@ mod tests { context.reset_result } + unsafe extern "C" fn compress_stream_test_frame_prepare_overflow( + _context: *mut c_void, + _src: *const c_void, + _block_size: usize, + ) { + } + + unsafe extern "C" fn compress_stream_test_frame_prepare_window( + _context: *mut c_void, + _src: *const c_void, + _block_size: usize, + _max_dist: c_uint, + ) { + } + + unsafe extern "C" fn compress_stream_test_trace(_context: *mut c_void, _extra_c_size: usize) {} + #[allow(clippy::too_many_arguments)] fn compress_stream_test_state( context: &mut CompressStreamTestContext, @@ -11827,8 +11865,169 @@ mod tests { out_buff_flushed_size: &mut usize, frame_ended: &mut c_uint, ) -> ZSTD_rust_compressStreamState { + let base = WINDOW_INIT_SENTINEL.as_ptr().cast::(); + context.window = ZSTD_rust_windowUpdateState { + nextSrc: base, + base, + dictBase: base, + dictLimit: ZSTD_WINDOW_START_INDEX, + lowLimit: ZSTD_WINDOW_START_INDEX, + }; + context.force_non_contiguous = 0; + context.next_to_update = ZSTD_WINDOW_START_INDEX; + context.compression_stage = ZSTD_COMPRESSION_STAGE_ONGOING; + context.is_first_block = 1; + context.end_is_first_block = 1; + context.consumed_src_size = 0; + context.produced_c_size = 0; + context.window_projection = ZSTD_rust_compressContinueWindowProjection { + next_src: ptr::addr_of_mut!(context.window.nextSrc), + base: ptr::addr_of_mut!(context.window.base), + dict_base: ptr::addr_of_mut!(context.window.dictBase), + dict_limit: ptr::addr_of_mut!(context.window.dictLimit), + low_limit: ptr::addr_of_mut!(context.window.lowLimit), + force_non_contiguous: ptr::addr_of_mut!(context.force_non_contiguous), + next_to_update: ptr::addr_of_mut!(context.next_to_update), + }; + let callback_context = (context as *mut CompressStreamTestContext).cast(); + context.frame_chunk_clamp_state = Some(ZSTD_rust_frameChunkClampState { + next_to_update: ptr::addr_of_mut!(context.next_to_update), + low_limit: ptr::addr_of!(context.window.lowLimit), + }); + let frame_chunk_clamp_state = context + .frame_chunk_clamp_state + .as_ref() + .map_or(ptr::null(), |state| state as *const _); + context.frame_chunk_prepare_state = Some(ZSTD_rust_frameChunkPrepareState { + callback_context, + max_dist: 64, + correct_overflow: compress_stream_test_frame_prepare_overflow, + check_dict_validity: compress_stream_test_frame_prepare_window, + enforce_max_dist: compress_stream_test_frame_prepare_window, + clamp_state: frame_chunk_clamp_state, + }); + let frame_chunk_prepare_state = context + .frame_chunk_prepare_state + .as_ref() + .map_or(ptr::null(), |state| state as *const _); + context.continue_frame_chunk_state = Some(ZSTD_rust_frameChunkState { + callback_context, + tmp_workspace: ptr::null_mut(), + checksum_state: ptr::null_mut(), + is_first_block: ptr::addr_of_mut!(context.is_first_block), + stage: ptr::addr_of_mut!(context.compression_stage), + tmp_wksp_size: 0, + block_size_max, + savings: 0, + pre_block_splitter_level: 1, + strategy: ZSTD_FAST, + use_target_c_block_size: 1, + block_splitter_enabled: 0, + checksum_flag: 0, + ending_stage: ZSTD_COMPRESSION_STAGE_ENDING, + prepare_state: frame_chunk_prepare_state, + compress_target: compress_stream_test_block, + compress_split: compress_stream_test_block, + compress_internal: compress_stream_test_block, + }); + let continue_frame_chunk_state = context + .continue_frame_chunk_state + .as_ref() + .map_or(ptr::null(), |state| state as *const _); + context.end_frame_chunk_state = Some(ZSTD_rust_frameChunkState { + callback_context, + tmp_workspace: ptr::null_mut(), + checksum_state: ptr::null_mut(), + is_first_block: ptr::addr_of_mut!(context.end_is_first_block), + stage: ptr::addr_of_mut!(context.compression_stage), + tmp_wksp_size: 0, + block_size_max, + savings: 0, + pre_block_splitter_level: 1, + strategy: ZSTD_FAST, + use_target_c_block_size: 1, + block_splitter_enabled: 0, + checksum_flag: 0, + ending_stage: ZSTD_COMPRESSION_STAGE_ENDING, + prepare_state: frame_chunk_prepare_state, + compress_target: compress_stream_test_end, + compress_split: compress_stream_test_end, + compress_internal: compress_stream_test_end, + }); + let end_frame_chunk_state = context + .end_frame_chunk_state + .as_ref() + .map_or(ptr::null(), |state| state as *const _); + context.continue_state = Some(ZSTD_rust_compressContinueState { + callback_context, + window_state: &context.window_projection, + ldm_window_state: ptr::null(), + overflow_state: ptr::null(), + frame_chunk_state: continue_frame_chunk_state, + compress_block: compress_stream_test_block, + stage: ptr::addr_of_mut!(context.compression_stage), + consumed_src_size: ptr::addr_of_mut!(context.consumed_src_size), + produced_c_size: ptr::addr_of_mut!(context.produced_c_size), + pledged_src_size_plus_one: 0, + block_size_max, + check_block_size: 0, + no_dict_id_flag: 0, + checksum_flag: 0, + content_size_flag: 0, + format: 0, + window_log: 20, + dict_id: 0, + ldm_enabled: 0, + }); + let continue_state = context + .continue_state + .as_ref() + .map_or(ptr::null(), |state| state as *const _); + context.end_continue_state = Some(ZSTD_rust_compressContinueState { + callback_context, + window_state: &context.window_projection, + ldm_window_state: ptr::null(), + overflow_state: ptr::null(), + frame_chunk_state: end_frame_chunk_state, + compress_block: compress_stream_test_end, + stage: ptr::addr_of_mut!(context.compression_stage), + consumed_src_size: ptr::addr_of_mut!(context.consumed_src_size), + produced_c_size: ptr::addr_of_mut!(context.produced_c_size), + pledged_src_size_plus_one: 0, + block_size_max, + check_block_size: 0, + no_dict_id_flag: 0, + checksum_flag: 0, + content_size_flag: 0, + format: 0, + window_log: 20, + dict_id: 0, + ldm_enabled: 0, + }); + let end_continue_state = context + .end_continue_state + .as_ref() + .map_or(ptr::null(), |state| state as *const _); + context.end_state = Some(ZSTD_rust_compressEndState { + callback_context, + compress_continue_state: end_continue_state, + trace: compress_stream_test_trace, + consumed_src_size: ptr::addr_of!(context.consumed_src_size), + pledged_src_size_plus_one: 0, + content_size_flag: 0, + stage: ptr::addr_of_mut!(context.compression_stage), + no_dict_id_flag: 0, + checksum_flag: 0, + format: 0, + window_log: 20, + checksum_state: &COMPRESS_END_TEST_CHECKSUM_STATE, + }); + let end_state = context + .end_state + .as_ref() + .map_or(ptr::null(), |state| state as *const _); ZSTD_rust_compressStreamState { - callback_context: (context as *mut CompressStreamTestContext).cast(), + callback_context, in_buffer_mode, out_buffer_mode, stream_stage: stage, @@ -11844,8 +12043,8 @@ mod tests { out_buff_content_size, out_buff_flushed_size, frame_ended, - compress_continue: compress_stream_test_block, - compress_end: compress_stream_test_end, + compress_continue_state: continue_state, + compress_end_state: end_state, reset_session: compress_stream_test_reset, } }