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, } }