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

feat(compress): move continue frame chunks into Rust

Browse Source 
Project the prepared frame-chunk state through the compressContinue ABI so
Rust invokes the existing frame-chunk loop directly. Remove the redundant C
frame-chunk forwarding wrapper while retaining C callbacks for the private
block-compression leaves and context-sensitive preparation operations.

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_continue -- --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
This commit is contained in:
ddidderr
2026-07-20 00:48:02 +02:00
parent 3ec8f03850
commit 58e6d2eef7
2 changed files with 115 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+35 -60
View File
@@ -874,7 +874,8 @@ typedef char ZSTD_rust_frame_chunk_state_layout[
? 1 : -1];
/* The high-level continue/block entry points are Rust-owned. This projection
* carries frame-header scalars, mutable state, and C callbacks. Window
* carries frame-header scalars, the already-projected frame-chunk state, and
* the remaining C callback for context-sensitive block compression. Window
* advancement uses direct pointers to the C-owned fields; the private CCtx
* and match-state layout never crosses the ABI. */
typedef struct {
@@ -899,7 +900,7 @@ typedef struct {
ZSTD_rust_compressContinueWindowState* windowState;
ZSTD_rust_compressContinueWindowState* ldmWindowState;
const ZSTD_rust_overflowCorrectState* overflowState;
ZSTD_rust_compressContinueBlock_f compressFrameChunk;
const ZSTD_rust_frameChunkState* frameChunkState;
ZSTD_rust_compressContinueBlock_f compressBlock;
ZSTD_compressionStage_e* stage;
unsigned long long* consumedSrcSize;
@@ -925,7 +926,7 @@ typedef char ZSTD_rust_compress_continue_state_layout[
&& offsetof(ZSTD_rust_compressContinueState, windowState) == sizeof(void*)
&& offsetof(ZSTD_rust_compressContinueState, ldmWindowState) == 2 * sizeof(void*)
&& offsetof(ZSTD_rust_compressContinueState, overflowState) == 3 * sizeof(void*)
&& offsetof(ZSTD_rust_compressContinueState, compressFrameChunk) == 4 * sizeof(void*)
&& offsetof(ZSTD_rust_compressContinueState, frameChunkState) == 4 * sizeof(void*)
&& offsetof(ZSTD_rust_compressContinueState, compressBlock) == 5 * sizeof(void*)
&& offsetof(ZSTD_rust_compressContinueState, stage) == 6 * sizeof(void*)
&& offsetof(ZSTD_rust_compressContinueState, consumedSrcSize) == 7 * sizeof(void*)
@@ -5568,62 +5569,6 @@ static size_t ZSTD_rust_frameChunk_compressInternal(
1 /* frame */);
}
/*! ZSTD_compress_frameChunk() :
* Compress a chunk of data into one or multiple blocks.
* All blocks will be terminated, all input will be consumed.
* Function will issue an error if there is not enough `dstCapacity` to hold the compressed content.
* Frame is supposed already started (header already produced)
* @return : compressed size, or an error code
*/
static size_t ZSTD_compress_frameChunk(ZSTD_CCtx* cctx,
void* dst, size_t dstCapacity,
const void* src, size_t srcSize,
U32 lastFrameChunk)
{
ZSTD_rust_frameChunkState state;
ZSTD_rust_frameChunkPrepareState prepareState;
ZSTD_rust_frameChunkClampState clampState;
ZSTD_MatchState_t* const ms = &cctx->blockState.matchState;
prepareState.callbackContext = cctx;
prepareState.maxDist = (U32)1 << cctx->appliedParams.cParams.windowLog;
prepareState.correctOverflow = ZSTD_rust_frameChunk_correctOverflow;
prepareState.checkDictValidity = ZSTD_rust_frameChunk_checkDictValidity;
prepareState.enforceMaxDist = ZSTD_rust_frameChunk_enforceMaxDist;
clampState.nextToUpdate = &ms->nextToUpdate;
clampState.lowLimit = &ms->window.lowLimit;
prepareState.clampState = &clampState;
state.callbackContext = cctx;
state.tmpWorkspace = cctx->tmpWorkspace;
state.checksumState = &cctx->xxhState;
state.isFirstBlock = &cctx->isFirstBlock;
state.stage = &cctx->stage;
state.tmpWkspSize = cctx->tmpWkspSize;
state.blockSizeMax = cctx->blockSizeMax;
state.savings = (S64)cctx->consumedSrcSize - (S64)cctx->producedCSize;
state.preBlockSplitterLevel = cctx->appliedParams.preBlockSplitter_level;
state.strategy = (int)cctx->appliedParams.cParams.strategy;
state.useTargetCBlockSize = ZSTD_useTargetCBlockSize(&cctx->appliedParams);
state.blockSplitterEnabled = ZSTD_blockSplitterEnabled(&cctx->appliedParams);
state.checksumFlag = cctx->appliedParams.fParams.checksumFlag;
state.endingStage = (int)ZSTDcs_ending;
state.prepareState = &prepareState;
state.compressTarget = ZSTD_rust_frameChunk_compressTarget;
state.compressSplit = ZSTD_rust_frameChunk_compressSplit;
state.compressInternal = ZSTD_rust_frameChunk_compressInternal;
return ZSTD_rust_compressFrameChunk(
&state, dst, dstCapacity, src, srcSize, lastFrameChunk);
}
static size_t ZSTD_rust_compressContinue_frameChunk(
void* context, void* dst, size_t dstCapacity,
const void* src, size_t srcSize, U32 lastFrameChunk)
{
return ZSTD_compress_frameChunk(
(ZSTD_CCtx*)context, dst, dstCapacity, src, srcSize,
lastFrameChunk);
}
static size_t ZSTD_rust_compressContinue_block(
void* context, void* dst, size_t dstCapacity,
const void* src, size_t srcSize, U32 lastFrameChunk)
@@ -5643,6 +5588,9 @@ static size_t ZSTD_compressContinue_dispatch(
ZSTD_rust_compressContinueState state;
ZSTD_rust_compressContinueWindowState windowState;
ZSTD_rust_compressContinueWindowState ldmWindowState;
ZSTD_rust_frameChunkState frameChunkState;
ZSTD_rust_frameChunkPrepareState frameChunkPrepareState;
ZSTD_rust_frameChunkClampState frameChunkClampState;
ZSTD_rust_overflowCorrectContext overflowContext;
ZSTD_rust_overflowCorrectState overflowState;
ZSTD_MatchState_t* const ms = &cctx->blockState.matchState;
@@ -5664,6 +5612,33 @@ static size_t ZSTD_compressContinue_dispatch(
ldmWindowState.forceNonContiguous = NULL;
ldmWindowState.nextToUpdate = NULL;
frameChunkPrepareState.callbackContext = cctx;
frameChunkPrepareState.maxDist = (U32)1 << cctx->appliedParams.cParams.windowLog;
frameChunkPrepareState.correctOverflow = ZSTD_rust_frameChunk_correctOverflow;
frameChunkPrepareState.checkDictValidity = ZSTD_rust_frameChunk_checkDictValidity;
frameChunkPrepareState.enforceMaxDist = ZSTD_rust_frameChunk_enforceMaxDist;
frameChunkClampState.nextToUpdate = &ms->nextToUpdate;
frameChunkClampState.lowLimit = &ms->window.lowLimit;
frameChunkPrepareState.clampState = &frameChunkClampState;
frameChunkState.callbackContext = cctx;
frameChunkState.tmpWorkspace = cctx->tmpWorkspace;
frameChunkState.checksumState = &cctx->xxhState;
frameChunkState.isFirstBlock = &cctx->isFirstBlock;
frameChunkState.stage = &cctx->stage;
frameChunkState.tmpWkspSize = cctx->tmpWkspSize;
frameChunkState.blockSizeMax = cctx->blockSizeMax;
frameChunkState.savings = (S64)cctx->consumedSrcSize - (S64)cctx->producedCSize;
frameChunkState.preBlockSplitterLevel = cctx->appliedParams.preBlockSplitter_level;
frameChunkState.strategy = (int)cctx->appliedParams.cParams.strategy;
frameChunkState.useTargetCBlockSize = ZSTD_useTargetCBlockSize(&cctx->appliedParams);
frameChunkState.blockSplitterEnabled = ZSTD_blockSplitterEnabled(&cctx->appliedParams);
frameChunkState.checksumFlag = cctx->appliedParams.fParams.checksumFlag;
frameChunkState.endingStage = (int)ZSTDcs_ending;
frameChunkState.prepareState = &frameChunkPrepareState;
frameChunkState.compressTarget = ZSTD_rust_frameChunk_compressTarget;
frameChunkState.compressSplit = ZSTD_rust_frameChunk_compressSplit;
frameChunkState.compressInternal = ZSTD_rust_frameChunk_compressInternal;
overflowContext.matchState = ms;
overflowContext.workspace = &cctx->workspace;
overflowContext.params = &cctx->appliedParams;
@@ -5681,7 +5656,7 @@ static size_t ZSTD_compressContinue_dispatch(
state.windowState = &windowState;
state.ldmWindowState = &ldmWindowState;
state.overflowState = &overflowState;
state.compressFrameChunk = ZSTD_rust_compressContinue_frameChunk;
state.frameChunkState = &frameChunkState;
state.compressBlock = ZSTD_rust_compressContinue_block;
state.stage = &cctx->stage;
state.consumedSrcSize = &cctx->consumedSrcSize;
rust/src/zstd_compress.rs
+80 -9
View File
@@ -841,16 +841,17 @@ const _: () = {
///
/// Rust owns stage transitions, frame-header sequencing, input progression,
/// and dispatch between the already-migrated frame-chunk/block bodies. The
/// frame-header parameters are projected as scalars. The opaque callback
/// context remains in C for overflow-correction callbacks and context-sensitive
/// block operations; window advancement is projected directly into Rust.
/// frame-header parameters and frame-chunk state are projected explicitly.
/// The opaque callback context remains in C for overflow-correction callbacks
/// and context-sensitive block operations; window advancement is projected
/// directly into Rust.
#[repr(C)]
pub struct ZSTD_rust_compressContinueState {
callback_context: *mut c_void,
window_state: *const ZSTD_rust_compressContinueWindowProjection,
ldm_window_state: *const ZSTD_rust_compressContinueWindowProjection,
overflow_state: *const ZSTD_rust_overflowCorrectState,
compress_frame_chunk: CompressContinueBlockFn,
frame_chunk_state: *const ZSTD_rust_frameChunkState,
compress_block: CompressContinueBlockFn,
stage: *mut c_int,
consumed_src_size: *mut u64,
@@ -878,7 +879,7 @@ const _: () = {
);
assert!(offset_of!(ZSTD_rust_compressContinueState, overflow_state) == 3 * size_of::<usize>());
assert!(
offset_of!(ZSTD_rust_compressContinueState, compress_frame_chunk) == 4 * size_of::<usize>()
offset_of!(ZSTD_rust_compressContinueState, frame_chunk_state) == 4 * size_of::<usize>()
);
assert!(offset_of!(ZSTD_rust_compressContinueState, compress_block) == 5 * size_of::<usize>());
assert!(offset_of!(ZSTD_rust_compressContinueState, stage) == 6 * size_of::<usize>());
@@ -938,6 +939,8 @@ const _: () = {
const ZSTD_COMPRESSION_STAGE_CREATED: c_int = 0;
const ZSTD_COMPRESSION_STAGE_INIT: c_int = 1;
const ZSTD_COMPRESSION_STAGE_ONGOING: c_int = 2;
#[cfg(test)]
const ZSTD_COMPRESSION_STAGE_ENDING: c_int = 3;
unsafe fn compress_continue_update_window(
projection: *const ZSTD_rust_compressContinueWindowProjection,
@@ -1081,9 +1084,12 @@ unsafe fn compress_continue_body_with(
}
let compressed_size = if frame != 0 {
if state.frame_chunk_state.is_null() {
return ERROR(ZstdErrorCode::Generic);
}
unsafe {
(state.compress_frame_chunk)(
state.callback_context,
ZSTD_rust_compressFrameChunk(
state.frame_chunk_state,
dst,
dst_capacity,
src,
@@ -11174,6 +11180,10 @@ mod tests {
window_projection: ZSTD_rust_compressContinueWindowProjection,
ldm_window_projection: ZSTD_rust_compressContinueWindowProjection,
overflow_state: Option<ZSTD_rust_overflowCorrectState>,
is_first_block: c_int,
frame_chunk_clamp_state: Option<ZSTD_rust_frameChunkClampState>,
frame_chunk_prepare_state: Option<ZSTD_rust_frameChunkPrepareState>,
frame_chunk_state: Option<ZSTD_rust_frameChunkState>,
}
unsafe fn compress_continue_test_context(
@@ -11207,6 +11217,21 @@ mod tests {
) {
}
unsafe extern "C" fn compress_continue_test_frame_prepare_overflow(
_context: *mut c_void,
_src: *const c_void,
_block_size: usize,
) {
}
unsafe extern "C" fn compress_continue_test_frame_prepare_window(
_context: *mut c_void,
_src: *const c_void,
_block_size: usize,
_max_dist: c_uint,
) {
}
unsafe extern "C" fn compress_continue_test_frame(
context: *mut c_void,
_dst: *mut c_void,
@@ -11257,6 +11282,7 @@ mod tests {
context.next_to_update = ZSTD_WINDOW_START_INDEX;
context.loaded_dict_end = 0;
context.dict_match_state = ptr::null();
context.is_first_block = 1;
context.window_projection = ZSTD_rust_compressContinueWindowProjection {
next_src: ptr::addr_of_mut!(context.window.nextSrc),
base: ptr::addr_of_mut!(context.window.base),
@@ -11291,12 +11317,56 @@ mod tests {
.overflow_state
.as_ref()
.map_or(ptr::null(), |state| state as *const _);
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_continue_test_frame_prepare_overflow,
check_dict_validity: compress_continue_test_frame_prepare_window,
enforce_max_dist: compress_continue_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.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,
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_continue_test_frame,
compress_split: compress_continue_test_frame,
compress_internal: compress_continue_test_frame,
});
let frame_chunk_state = context
.frame_chunk_state
.as_ref()
.map_or(ptr::null(), |state| state as *const _);
ZSTD_rust_compressContinueState {
callback_context,
window_state: &context.window_projection,
ldm_window_state: &context.ldm_window_projection,
overflow_state,
compress_frame_chunk: compress_continue_test_frame,
frame_chunk_state,
compress_block: compress_continue_test_block,
stage,
consumed_src_size,
@@ -11348,7 +11418,7 @@ mod tests {
};
assert_eq!(result, 12);
assert_eq!(stage, ZSTD_COMPRESSION_STAGE_ONGOING);
assert_eq!(stage, ZSTD_COMPRESSION_STAGE_ENDING);
assert_eq!(consumed, 12);
assert_eq!(produced, 23);
assert_eq!(context.window.nextSrc, unsafe {
@@ -11361,6 +11431,7 @@ mod tests {
assert_eq!(context.frame_calls, 1);
assert_eq!(context.block_calls, 0);
assert_eq!(context.last_frame_chunk, 1);
assert_eq!(context.is_first_block, 0);
}
#[test]