ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(compress): move frame epilogue into Rust

Browse Source 
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.
This commit is contained in:
ddidderr
2026-07-19 23:58:06 +02:00
parent d8e8683353
commit 048979aff6
3 changed files with 119 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+30 -43
View File
@@ -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);
}
rust/src/zstd_compress.rs
+85 -51
View File
@@ -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::<usize>());
assert!(offset_of!(ZSTD_rust_compressEndState, write_epilogue) == 2 * size_of::<usize>());
assert!(offset_of!(ZSTD_rust_compressEndState, trace) == 3 * size_of::<usize>());
assert!(offset_of!(ZSTD_rust_compressEndState, consumed_src_size) == 4 * size_of::<usize>());
assert!(offset_of!(ZSTD_rust_compressEndState, trace) == 2 * size_of::<usize>());
assert!(offset_of!(ZSTD_rust_compressEndState, consumed_src_size) == 3 * size_of::<usize>());
assert!(
offset_of!(ZSTD_rust_compressEndState, pledged_src_size_plus_one) == 5 * size_of::<usize>()
offset_of!(ZSTD_rust_compressEndState, pledged_src_size_plus_one) == 4 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_compressEndState, content_size_flag)
== 5 * size_of::<usize>() + size_of::<u64>()
== 4 * size_of::<usize>() + size_of::<u64>()
);
assert!(
size_of::<ZSTD_rust_compressEndState>() == if size_of::<usize>() == 8 { 56 } else { 32 }
offset_of!(ZSTD_rust_compressEndState, stage) == 5 * size_of::<usize>() + size_of::<u64>()
);
assert!(
offset_of!(ZSTD_rust_compressEndState, no_dict_id_flag)
== 6 * size_of::<usize>() + size_of::<u64>()
);
assert!(
offset_of!(ZSTD_rust_compressEndState, checksum_flag)
== 6 * size_of::<usize>() + size_of::<u64>() + size_of::<c_int>()
);
assert!(
offset_of!(ZSTD_rust_compressEndState, format)
== 6 * size_of::<usize>() + size_of::<u64>() + 2 * size_of::<c_int>()
);
assert!(
offset_of!(ZSTD_rust_compressEndState, window_log)
== 6 * size_of::<usize>() + size_of::<u64>() + 3 * size_of::<c_int>()
);
assert!(
offset_of!(ZSTD_rust_compressEndState, checksum_state)
== 6 * size_of::<usize>()
+ size_of::<u64>()
+ 3 * size_of::<c_int>()
+ size_of::<c_uint>()
);
assert!(
size_of::<ZSTD_rust_compressEndState>() == if size_of::<usize>() == 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::<u8>().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)]
rust/src/zstd_compress_frame.rs
+4 -3
View File
@@ -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,