From 048979aff65eb032c8d0887222823f0196f272a7 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sun, 19 Jul 2026 23:58:06 +0200 Subject: [PATCH] feat(compress): move frame epilogue into Rust Project the end-frame stage, format flags, window log, and XXH64 state into Rust so the Rust end-of-frame orchestrator computes the checksum and invokes the existing Rust epilogue serializer directly. Remove the C epilogue and checksum adapter while preserving callback order, output accounting, stage transitions, and error precedence. Test Plan: - CARGO_BUILD_JOBS=1 cargo test --lib zstd_compress::tests::compress_end -- --nocapture - CARGO_BUILD_JOBS=1 cargo clippy --all-targets -- -D warnings - CARGO_BUILD_JOBS=1 cargo test - CARGO_BUILD_JOBS=1 make -j1 - CARGO_BUILD_JOBS=1 make -j1 -C tests test-zstream ZSTREAM_TESTTIME=-T2s - CARGO_BUILD_JOBS=1 make -j1 -C tests test-fuzzer FUZZERTEST=-T3s FUZZER_FLAGS=--no-big-tests All commands were run serially with a 40 GiB virtual-memory cap. --- lib/compress/zstd_compress.c | 73 +++++++---------- rust/src/zstd_compress.rs | 136 ++++++++++++++++++++------------ rust/src/zstd_compress_frame.rs | 7 +- 3 files changed, 119 insertions(+), 97 deletions(-) diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index be965c57e..ce1750205 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -39,10 +39,6 @@ size_t ZSTD_rust_writeFrameHeader(void* dst, size_t dstCapacity, U32 dictID); void ZSTD_rust_writeBlockHeader(void* op, size_t cSize, size_t blockSize, U32 lastBlock); -size_t ZSTD_rust_writeEpilogue(void* dst, size_t dstCapacity, int* stage, - int noDictIDFlag, int checksumFlag, - int contentSizeFlag, int format, U32 windowLog, - U32 checksum); int ZSTD_rust_updateFrameProgression(unsigned long long* consumedSrcSize, unsigned long long* producedCSize, unsigned long long pledgedSrcSizePlusOne, @@ -935,7 +931,8 @@ typedef char ZSTD_rust_compress_continue_state_layout[ ? 1 : -1]; /* Rust owns the end-of-frame orchestration. The callbacks retain the - * private CCtx-dependent continue, epilogue, and trace operations in C. */ + * private CCtx-dependent continue and trace operations in C; checksum and + * frame-epilogue fields are projected explicitly. */ typedef size_t (*ZSTD_rust_compressEndContinue_f)(void* context, void* dst, size_t dstCapacity, @@ -943,19 +940,21 @@ typedef size_t (*ZSTD_rust_compressEndContinue_f)(void* context, size_t srcSize, U32 frame, U32 lastFrameChunk); -typedef size_t (*ZSTD_rust_compressEndEpilogue_f)(void* context, - void* dst, - size_t dstCapacity); typedef void (*ZSTD_rust_compressEndTrace_f)(void* context, size_t extraCSize); typedef struct { void* callbackContext; ZSTD_rust_compressEndContinue_f compressContinue; - ZSTD_rust_compressEndEpilogue_f writeEpilogue; ZSTD_rust_compressEndTrace_f trace; unsigned long long* consumedSrcSize; U64 pledgedSrcSizePlusOne; int contentSizeFlag; + int* stage; + int noDictIDFlag; + int checksumFlag; + int format; + U32 windowLog; + const XXH64_state_t* checksumState; } ZSTD_rust_compressEndState; size_t ZSTD_rust_compressEnd(const ZSTD_rust_compressEndState* state, void* dst, size_t dstCapacity, @@ -964,18 +963,28 @@ typedef char ZSTD_rust_compress_end_state_layout[ (offsetof(ZSTD_rust_compressEndState, callbackContext) == 0 && offsetof(ZSTD_rust_compressEndState, compressContinue) == sizeof(void*) - && offsetof(ZSTD_rust_compressEndState, writeEpilogue) - == 2 * sizeof(void*) && offsetof(ZSTD_rust_compressEndState, trace) - == 3 * sizeof(void*) + == 2 * sizeof(void*) && offsetof(ZSTD_rust_compressEndState, consumedSrcSize) - == 4 * sizeof(void*) + == 3 * sizeof(void*) && offsetof(ZSTD_rust_compressEndState, pledgedSrcSizePlusOne) - == 5 * sizeof(void*) + == 4 * sizeof(void*) && offsetof(ZSTD_rust_compressEndState, contentSizeFlag) + == 4 * sizeof(void*) + sizeof(U64) + && offsetof(ZSTD_rust_compressEndState, stage) == 5 * sizeof(void*) + sizeof(U64) + && offsetof(ZSTD_rust_compressEndState, noDictIDFlag) + == 6 * sizeof(void*) + sizeof(U64) + && offsetof(ZSTD_rust_compressEndState, checksumFlag) + == 6 * sizeof(void*) + sizeof(U64) + sizeof(int) + && offsetof(ZSTD_rust_compressEndState, format) + == 6 * sizeof(void*) + sizeof(U64) + 2 * sizeof(int) + && offsetof(ZSTD_rust_compressEndState, windowLog) + == 6 * sizeof(void*) + sizeof(U64) + 3 * sizeof(int) + && offsetof(ZSTD_rust_compressEndState, checksumState) + == 6 * sizeof(void*) + sizeof(U64) + 3 * sizeof(int) + sizeof(U32) && sizeof(ZSTD_rust_compressEndState) - == (sizeof(void*) == 8 ? 56 : 32)) + == (sizeof(void*) == 8 ? 80 : 48)) ? 1 : -1]; /* Rust owns the single-threaded buffered/stable stream state machine. The @@ -6161,27 +6170,6 @@ size_t ZSTD_compressBegin(ZSTD_CCtx* cctx, int compressionLevel) } -/*! ZSTD_writeEpilogue() : -* Ends a frame. -* @return : nb of bytes written into dst (or an error code) */ -static size_t ZSTD_writeEpilogue(ZSTD_CCtx* cctx, void* dst, size_t dstCapacity) -{ - U32 const checksum = cctx->appliedParams.fParams.checksumFlag - ? (U32)XXH64_digest(&cctx->xxhState) - : 0; - DEBUGLOG(4, "ZSTD_writeEpilogue"); - if (cctx->appliedParams.fParams.checksumFlag) - DEBUGLOG(4, "ZSTD_writeEpilogue: write checksum : %08X", (unsigned)checksum); - return ZSTD_rust_writeEpilogue( - dst, dstCapacity, (int*)&cctx->stage, - cctx->appliedParams.fParams.noDictIDFlag, - cctx->appliedParams.fParams.checksumFlag, - cctx->appliedParams.fParams.contentSizeFlag, - (int)cctx->appliedParams.format, - cctx->appliedParams.cParams.windowLog, - checksum); -} - void ZSTD_CCtx_trace(ZSTD_CCtx* cctx, size_t extraCSize) { #if ZSTD_TRACE @@ -6217,12 +6205,6 @@ static size_t ZSTD_rust_compressEnd_continue( 0 /* block size already selected */); } -static size_t ZSTD_rust_compressEnd_writeEpilogue( - void* context, void* dst, size_t dstCapacity) -{ - return ZSTD_writeEpilogue((ZSTD_CCtx*)context, dst, dstCapacity); -} - static void ZSTD_rust_compressEnd_trace(void* context, size_t extraCSize) { ZSTD_CCtx_trace((ZSTD_CCtx*)context, extraCSize); @@ -6235,11 +6217,16 @@ size_t ZSTD_compressEnd_public(ZSTD_CCtx* cctx, ZSTD_rust_compressEndState state; state.callbackContext = cctx; state.compressContinue = ZSTD_rust_compressEnd_continue; - state.writeEpilogue = ZSTD_rust_compressEnd_writeEpilogue; state.trace = ZSTD_rust_compressEnd_trace; state.consumedSrcSize = &cctx->consumedSrcSize; state.pledgedSrcSizePlusOne = cctx->pledgedSrcSizePlusOne; state.contentSizeFlag = cctx->appliedParams.fParams.contentSizeFlag; + state.stage = (int*)&cctx->stage; + state.noDictIDFlag = cctx->appliedParams.fParams.noDictIDFlag; + state.checksumFlag = cctx->appliedParams.fParams.checksumFlag; + state.format = (int)cctx->appliedParams.format; + state.windowLog = cctx->appliedParams.cParams.windowLog; + state.checksumState = &cctx->xxhState; return ZSTD_rust_compressEnd( &state, dst, dstCapacity, src, srcSize); } diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index ce9a7208f..13e842008 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -15,11 +15,13 @@ use crate::common::MINMATCH; use crate::errors::{ERR_isError, ZstdErrorCode, ERROR}; +use crate::xxhash::{XXH64_digest, XXH64_state_t}; #[cfg(not(test))] use crate::zstd_compress_api::ZSTD_compressBound; use crate::zstd_compress_frame::{ write_raw_block, ZSTD_rust_noCompressBlock, ZSTD_rust_rleCompressBlock, - ZSTD_rust_writeBlockHeader, ZSTD_rust_writeFrameHeader, ZSTD_writeLastEmptyBlock, + ZSTD_rust_writeBlockHeader, ZSTD_rust_writeEpilogue, ZSTD_rust_writeFrameHeader, + ZSTD_writeLastEmptyBlock, }; use crate::zstd_compress_literals::min_gain; use crate::zstd_compress_params::{ @@ -976,40 +978,70 @@ type CompressEndContinueFn = unsafe extern "C" fn( c_uint, c_uint, ) -> usize; -type CompressEndEpilogueFn = unsafe extern "C" fn(*mut c_void, *mut c_void, usize) -> usize; type CompressEndTraceFn = unsafe extern "C" fn(*mut c_void, usize); /// Explicit projection for the public end-of-frame orchestration. /// /// Rust owns callback ordering, output offset/capacity accounting, and -/// pledged-size validation. The opaque callback context retains the private -/// CCtx-dependent continue, epilogue, checksum, and trace operations in C. +/// pledged-size validation. The opaque callback context retains only the +/// private CCtx-dependent continue and trace operations in C; Rust serializes +/// the epilogue from explicit frame and checksum projections. #[repr(C)] pub struct ZSTD_rust_compressEndState { callback_context: *mut c_void, compress_continue: CompressEndContinueFn, - write_epilogue: CompressEndEpilogueFn, trace: CompressEndTraceFn, consumed_src_size: *const u64, pledged_src_size_plus_one: u64, content_size_flag: c_int, + stage: *mut c_int, + no_dict_id_flag: c_int, + checksum_flag: c_int, + format: c_int, + window_log: c_uint, + checksum_state: *const XXH64_state_t, } const _: () = { assert!(offset_of!(ZSTD_rust_compressEndState, callback_context) == 0); assert!(offset_of!(ZSTD_rust_compressEndState, compress_continue) == size_of::()); - assert!(offset_of!(ZSTD_rust_compressEndState, write_epilogue) == 2 * size_of::()); - assert!(offset_of!(ZSTD_rust_compressEndState, trace) == 3 * size_of::()); - assert!(offset_of!(ZSTD_rust_compressEndState, consumed_src_size) == 4 * size_of::()); + assert!(offset_of!(ZSTD_rust_compressEndState, trace) == 2 * size_of::()); + assert!(offset_of!(ZSTD_rust_compressEndState, consumed_src_size) == 3 * size_of::()); assert!( - offset_of!(ZSTD_rust_compressEndState, pledged_src_size_plus_one) == 5 * size_of::() + offset_of!(ZSTD_rust_compressEndState, pledged_src_size_plus_one) == 4 * size_of::() ); assert!( offset_of!(ZSTD_rust_compressEndState, content_size_flag) - == 5 * size_of::() + size_of::() + == 4 * size_of::() + size_of::() ); assert!( - size_of::() == if size_of::() == 8 { 56 } else { 32 } + offset_of!(ZSTD_rust_compressEndState, stage) == 5 * size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressEndState, no_dict_id_flag) + == 6 * size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressEndState, checksum_flag) + == 6 * size_of::() + size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressEndState, format) + == 6 * size_of::() + size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressEndState, window_log) + == 6 * size_of::() + size_of::() + 3 * size_of::() + ); + assert!( + offset_of!(ZSTD_rust_compressEndState, checksum_state) + == 6 * size_of::() + + size_of::() + + 3 * size_of::() + + size_of::() + ); + assert!( + size_of::() == if size_of::() == 8 { 80 } else { 48 } ); }; @@ -1020,7 +1052,8 @@ unsafe fn compress_end_body_with( src: *const c_void, src_size: usize, ) -> usize { - if state.consumed_src_size.is_null() { + if state.consumed_src_size.is_null() || state.stage.is_null() || state.checksum_state.is_null() + { return ERROR(ZstdErrorCode::Generic); } @@ -1039,11 +1072,22 @@ unsafe fn compress_end_body_with( return c_size; } + let checksum = if state.checksum_flag != 0 { + unsafe { XXH64_digest(state.checksum_state) as u32 } + } else { + 0 + }; let end_result = unsafe { - (state.write_epilogue)( - state.callback_context, + ZSTD_rust_writeEpilogue( dst.cast::().add(c_size).cast(), dst_capacity.wrapping_sub(c_size), + state.stage, + state.no_dict_id_flag, + state.checksum_flag, + state.content_size_flag, + state.format, + state.window_log, + checksum, ) }; if ERR_isError(end_result) { @@ -9911,15 +9955,21 @@ mod tests { struct CompressEndTestContext { events: Vec<&'static str>, continue_result: usize, - epilogue_result: usize, - epilogue_offset: usize, - epilogue_capacity: usize, trace_extra: usize, continue_frame: c_uint, continue_last_frame_chunk: c_uint, - dst_base: usize, + stage: c_int, } + static COMPRESS_END_TEST_CHECKSUM_STATE: XXH64_state_t = XXH64_state_t { + total_len: 0, + v: [0; 4], + mem64: [0; 4], + memsize: 0, + reserved32: 0, + reserved64: 0, + }; + unsafe fn compress_end_test_context( context: *mut c_void, ) -> &'static mut CompressEndTestContext { @@ -9942,18 +9992,6 @@ mod tests { context.continue_result } - unsafe extern "C" fn compress_end_test_epilogue( - context: *mut c_void, - dst: *mut c_void, - dst_capacity: usize, - ) -> usize { - let context = unsafe { compress_end_test_context(context) }; - context.events.push("epilogue"); - context.epilogue_offset = (dst as usize).wrapping_sub(context.dst_base); - context.epilogue_capacity = dst_capacity; - context.epilogue_result - } - unsafe extern "C" fn compress_end_test_trace(context: *mut c_void, extra_c_size: usize) { let context = unsafe { compress_end_test_context(context) }; context.events.push("trace"); @@ -9966,24 +10004,28 @@ mod tests { pledged_src_size_plus_one: u64, content_size_flag: c_int, ) -> ZSTD_rust_compressEndState { + context.stage = 1; ZSTD_rust_compressEndState { callback_context: (context as *mut CompressEndTestContext).cast(), compress_continue: compress_end_test_continue, - write_epilogue: compress_end_test_epilogue, trace: compress_end_test_trace, consumed_src_size: consumed_src_size as *const u64, pledged_src_size_plus_one, content_size_flag, + stage: &mut context.stage, + no_dict_id_flag: 0, + checksum_flag: 1, + format: 0, + window_log: 20, + checksum_state: &COMPRESS_END_TEST_CHECKSUM_STATE, } } #[test] fn compress_end_preserves_callback_order_and_output_accounting() { - let mut dst = [0u8; 16]; + let mut dst = [0u8; 32]; let mut context = CompressEndTestContext { continue_result: 3, - epilogue_result: 5, - dst_base: dst.as_mut_ptr() as usize, ..CompressEndTestContext::default() }; let consumed_src_size = 7; @@ -9993,17 +10035,14 @@ mod tests { ZSTD_rust_compressEnd(&state, dst.as_mut_ptr().cast(), dst.len(), ptr::null(), 0) }; - assert_eq!(result, 8); - assert_eq!(context.events, ["continue", "epilogue", "trace"]); + assert_eq!(result, 16); + assert_eq!(context.events, ["continue", "trace"]); assert_eq!( (context.continue_frame, context.continue_last_frame_chunk), (1, 1) ); - assert_eq!( - (context.epilogue_offset, context.epilogue_capacity), - (3, 13) - ); - assert_eq!(context.trace_extra, 5); + assert_eq!(context.trace_extra, 13); + assert_eq!(context.stage, 0); } #[test] @@ -10011,7 +10050,6 @@ mod tests { let mut dst = [0u8; 16]; let mut context = CompressEndTestContext { continue_result: ERROR(ZstdErrorCode::MemoryAllocation), - dst_base: dst.as_mut_ptr() as usize, ..CompressEndTestContext::default() }; let consumed_src_size = 7; @@ -10027,11 +10065,9 @@ mod tests { #[test] fn compress_end_stops_before_validation_and_trace_when_epilogue_fails() { - let mut dst = [0u8; 16]; + let mut dst = [0u8; 8]; let mut context = CompressEndTestContext { continue_result: 3, - epilogue_result: ERROR(ZstdErrorCode::MemoryAllocation), - dst_base: dst.as_mut_ptr() as usize, ..CompressEndTestContext::default() }; let consumed_src_size = 7; @@ -10041,17 +10077,15 @@ mod tests { ZSTD_rust_compressEnd(&state, dst.as_mut_ptr().cast(), dst.len(), ptr::null(), 0) }; - assert_eq!(result, ERROR(ZstdErrorCode::MemoryAllocation)); - assert_eq!(context.events, ["continue", "epilogue"]); + assert_eq!(result, ERROR(ZstdErrorCode::DstSizeTooSmall)); + assert_eq!(context.events, ["continue"]); } #[test] fn compress_end_rejects_pledged_size_before_trace() { - let mut dst = [0u8; 16]; + let mut dst = [0u8; 32]; let mut context = CompressEndTestContext { continue_result: 3, - epilogue_result: 5, - dst_base: dst.as_mut_ptr() as usize, ..CompressEndTestContext::default() }; let consumed_src_size = 7; @@ -10062,7 +10096,7 @@ mod tests { }; assert_eq!(result, ERROR(ZstdErrorCode::SrcSizeWrong)); - assert_eq!(context.events, ["continue", "epilogue"]); + assert_eq!(context.events, ["continue"]); } #[derive(Default)] diff --git a/rust/src/zstd_compress_frame.rs b/rust/src/zstd_compress_frame.rs index 2b6d3d4b9..21f3a6a92 100644 --- a/rust/src/zstd_compress_frame.rs +++ b/rust/src/zstd_compress_frame.rs @@ -346,10 +346,11 @@ pub unsafe extern "C" fn ZSTD_rust_optimalBlockSize( /// Rust implementation of the private `ZSTD_writeEpilogue()` serializer. /// -/// The C caller owns the compression context and passes its stage by scalar +/// The caller owns the compression context and passes its stage by scalar /// pointer so the transition to `ongoing` after an empty-frame header remains -/// visible even when a later write fails. The checksum is already finalized -/// by C; Rust only serializes its low 32 bits. +/// visible even when a later write fails. The checksum is supplied as its +/// low 32 bits; the Rust end-of-frame orchestrator computes that value from +/// the projected XXH64 state. #[no_mangle] pub unsafe extern "C" fn ZSTD_rust_writeEpilogue( dst: *mut c_void,