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

feat(compress): move continue orchestration into Rust

Browse Source 
The public and deprecated block APIs previously entered a C-owned
ZSTD_compressContinue_internal routine, which mixed frame-header sequencing,
window maintenance, block dispatch, and frame-size accounting with the private
CCtx match state.  That made the high-level compression boundary both harder
to test and easy to diverge from the already-ported frame-chunk body.

Add an explicit C/Rust projection containing only scalar state and callback
slots.  Rust now owns stage validation and transitions, header sequencing,
frame-versus-block dispatch, and consumed/produced progression, while C
callbacks retain access to the private match state and context-sensitive block
operations.  The same path now covers compressContinue, deprecated block
compression, and compressEnd without exposing the unstable CCtx layout.

Test Plan:
- `cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1` -- 473 passed.
- `cargo test --manifest-path rust/Cargo.toml --no-default-features --features compression,decompression,dict-builder,legacy-v01,legacy-v02,legacy-v03,legacy-v04,legacy-v05,legacy-v06,legacy-v07 --all-targets -- --test-threads=1` -- 528 passed.
- Native `test-fuzzer`, `test-zstream`, `test-decodecorpus`, `test-cli-tests`, and `test-zstd` -- passed.
- `cargo clippy` default, benches, and tests for library and CLI -- passed.
This commit is contained in:
ddidderr
2026-07-18 22:40:35 +02:00
parent 5aeef8096e
commit 217cc76e74
2 changed files with 644 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/src/zstd_compress.rs
+500
View File
@@ -340,6 +340,185 @@ pub unsafe extern "C" fn ZSTD_rust_compressFrameChunk(
}
}
type CompressContinueHeaderFn = unsafe extern "C" fn(*mut c_void, *mut c_void, usize) -> usize;
type CompressContinueWindowFn = unsafe extern "C" fn(*mut c_void, *const c_void, usize);
type CompressContinueBlockFn =
unsafe extern "C" fn(*mut c_void, *mut c_void, usize, *const c_void, usize, c_uint) -> usize;
/// Explicit projection for the high-level continue and deprecated block APIs.
///
/// Rust owns stage transitions, frame-header sequencing, input progression,
/// and dispatch between the already-migrated frame-chunk/block bodies. The
/// opaque callback context remains in C, where callbacks update the private
/// match-state windows and invoke the C-owned context-sensitive operations.
#[repr(C)]
pub struct ZSTD_rust_compressContinueState {
callback_context: *mut c_void,
write_frame_header: CompressContinueHeaderFn,
update_window: CompressContinueWindowFn,
correct_overflow: CompressContinueWindowFn,
compress_frame_chunk: CompressContinueBlockFn,
compress_block: CompressContinueBlockFn,
stage: *mut c_int,
consumed_src_size: *mut u64,
produced_c_size: *mut u64,
pledged_src_size_plus_one: u64,
block_size_max: usize,
check_block_size: c_int,
}
const _: () = {
assert!(offset_of!(ZSTD_rust_compressContinueState, callback_context) == 0);
assert!(offset_of!(ZSTD_rust_compressContinueState, write_frame_header) == size_of::<usize>());
assert!(offset_of!(ZSTD_rust_compressContinueState, update_window) == 2 * size_of::<usize>());
assert!(
offset_of!(ZSTD_rust_compressContinueState, correct_overflow) == 3 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_compressContinueState, compress_frame_chunk) == 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>());
assert!(
offset_of!(ZSTD_rust_compressContinueState, consumed_src_size) == 7 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_compressContinueState, produced_c_size)
== 8 * (usize::BITS as usize / 8)
);
assert!(
offset_of!(ZSTD_rust_compressContinueState, pledged_src_size_plus_one)
== 9 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_compressContinueState, block_size_max)
== 9 * size_of::<usize>() + size_of::<u64>()
);
assert!(
offset_of!(ZSTD_rust_compressContinueState, check_block_size)
== 9 * size_of::<usize>() + size_of::<u64>() + size_of::<usize>()
);
assert!(
size_of::<ZSTD_rust_compressContinueState>()
== if size_of::<usize>() == 8 { 96 } else { 52 }
);
};
const ZSTD_COMPRESSION_STAGE_CREATED: c_int = 0;
const ZSTD_COMPRESSION_STAGE_INIT: c_int = 1;
const ZSTD_COMPRESSION_STAGE_ONGOING: c_int = 2;
#[allow(clippy::too_many_arguments)]
unsafe fn compress_continue_body_with(
state: &ZSTD_rust_compressContinueState,
mut dst: *mut c_void,
mut dst_capacity: usize,
src: *const c_void,
src_size: usize,
frame: c_uint,
last_frame_chunk: c_uint,
) -> usize {
if state.stage.is_null() || state.consumed_src_size.is_null() || state.produced_c_size.is_null()
{
return ERROR(ZstdErrorCode::Generic);
}
/* The deprecated block entry point performed this check before entering
* the old internal routine. Keep that ordering in the Rust orchestration. */
if state.check_block_size != 0 && src_size > state.block_size_max {
return ERROR(ZstdErrorCode::SrcSizeWrong);
}
if unsafe { *state.stage } == ZSTD_COMPRESSION_STAGE_CREATED {
return ERROR(ZstdErrorCode::StageWrong);
}
let mut frame_header_size = 0usize;
if frame != 0 && unsafe { *state.stage } == ZSTD_COMPRESSION_STAGE_INIT {
frame_header_size =
unsafe { (state.write_frame_header)(state.callback_context, dst, dst_capacity) };
if ERR_isError(frame_header_size) {
return frame_header_size;
}
if frame_header_size > dst_capacity {
return ERROR(ZstdErrorCode::DstSizeTooSmall);
}
unsafe {
*state.stage = ZSTD_COMPRESSION_STAGE_ONGOING;
}
dst = unsafe { dst.cast::<u8>().add(frame_header_size).cast() };
dst_capacity -= frame_header_size;
}
if src_size == 0 {
return frame_header_size;
}
unsafe { (state.update_window)(state.callback_context, src, src_size) };
if frame == 0 {
unsafe { (state.correct_overflow)(state.callback_context, src, src_size) };
}
let compressed_size = if frame != 0 {
unsafe {
(state.compress_frame_chunk)(
state.callback_context,
dst,
dst_capacity,
src,
src_size,
last_frame_chunk,
)
}
} else {
unsafe {
(state.compress_block)(state.callback_context, dst, dst_capacity, src, src_size, 0)
}
};
if ERR_isError(compressed_size) {
return compressed_size;
}
let source_size_wrong = update_frame_progression(
unsafe { &mut *state.consumed_src_size },
unsafe { &mut *state.produced_c_size },
state.pledged_src_size_plus_one,
src_size,
compressed_size,
frame_header_size,
);
if state.pledged_src_size_plus_one != 0 && source_size_wrong {
return ERROR(ZstdErrorCode::SrcSizeWrong);
}
compressed_size.wrapping_add(frame_header_size)
}
#[no_mangle]
pub unsafe extern "C" fn ZSTD_rust_compressContinue(
state: *const ZSTD_rust_compressContinueState,
dst: *mut c_void,
dst_capacity: usize,
src: *const c_void,
src_size: usize,
frame: c_uint,
last_frame_chunk: c_uint,
) -> usize {
if state.is_null() {
return ERROR(ZstdErrorCode::Generic);
}
unsafe {
compress_continue_body_with(
&*state,
dst,
dst_capacity,
src,
src_size,
frame,
last_frame_chunk,
)
}
}
/// Explicit projection of the state used by `ZSTD_compressSequences_internal`.
///
/// The C context and its function-pointer-bearing parameter structure remain
@@ -3161,6 +3340,327 @@ mod tests {
assert_eq!(output, [0xa5; 5]);
}
#[derive(Default)]
struct CompressContinueTestContext {
header_calls: usize,
update_window_calls: usize,
overflow_calls: usize,
frame_calls: usize,
block_calls: usize,
last_frame_chunk: c_uint,
header_result: usize,
frame_result: usize,
block_result: usize,
}
unsafe fn compress_continue_test_context(
context: *mut c_void,
) -> &'static mut CompressContinueTestContext {
unsafe { &mut *context.cast::<CompressContinueTestContext>() }
}
unsafe extern "C" fn compress_continue_test_header(
context: *mut c_void,
_dst: *mut c_void,
_dst_capacity: usize,
) -> usize {
let context = unsafe { compress_continue_test_context(context) };
context.header_calls += 1;
context.header_result
}
unsafe extern "C" fn compress_continue_test_window(
context: *mut c_void,
_src: *const c_void,
_src_size: usize,
) {
let context = unsafe { compress_continue_test_context(context) };
context.update_window_calls += 1;
}
unsafe extern "C" fn compress_continue_test_overflow(
context: *mut c_void,
_src: *const c_void,
_src_size: usize,
) {
let context = unsafe { compress_continue_test_context(context) };
context.overflow_calls += 1;
}
unsafe extern "C" fn compress_continue_test_frame(
context: *mut c_void,
_dst: *mut c_void,
_dst_capacity: usize,
_src: *const c_void,
_src_size: usize,
last_frame_chunk: c_uint,
) -> usize {
let context = unsafe { compress_continue_test_context(context) };
context.frame_calls += 1;
context.last_frame_chunk = last_frame_chunk;
context.frame_result
}
unsafe extern "C" fn compress_continue_test_block(
context: *mut c_void,
_dst: *mut c_void,
_dst_capacity: usize,
_src: *const c_void,
_src_size: usize,
_last_frame_chunk: c_uint,
) -> usize {
let context = unsafe { compress_continue_test_context(context) };
context.block_calls += 1;
context.block_result
}
fn compress_continue_test_state(
context: &mut CompressContinueTestContext,
stage: &mut c_int,
consumed_src_size: &mut u64,
produced_c_size: &mut u64,
pledged_src_size_plus_one: u64,
block_size_max: usize,
check_block_size: c_int,
) -> ZSTD_rust_compressContinueState {
ZSTD_rust_compressContinueState {
callback_context: (context as *mut CompressContinueTestContext).cast(),
write_frame_header: compress_continue_test_header,
update_window: compress_continue_test_window,
correct_overflow: compress_continue_test_overflow,
compress_frame_chunk: compress_continue_test_frame,
compress_block: compress_continue_test_block,
stage,
consumed_src_size,
produced_c_size,
pledged_src_size_plus_one,
block_size_max,
check_block_size,
}
}
#[test]
fn compress_continue_starts_frame_and_updates_progression() {
let mut context = CompressContinueTestContext {
header_result: 4,
frame_result: 6,
..CompressContinueTestContext::default()
};
let mut stage = ZSTD_COMPRESSION_STAGE_INIT;
let mut consumed = 7;
let mut produced = 11;
let state = compress_continue_test_state(
&mut context,
&mut stage,
&mut consumed,
&mut produced,
0,
16,
0,
);
let source = [0x11u8; 5];
let mut output = [0xa5u8; 32];
let result = unsafe {
ZSTD_rust_compressContinue(
&state,
output.as_mut_ptr().cast(),
output.len(),
source.as_ptr().cast(),
source.len(),
1,
1,
)
};
assert_eq!(result, 10);
assert_eq!(stage, ZSTD_COMPRESSION_STAGE_ONGOING);
assert_eq!(consumed, 12);
assert_eq!(produced, 21);
assert_eq!(context.header_calls, 1);
assert_eq!(context.update_window_calls, 1);
assert_eq!(context.overflow_calls, 0);
assert_eq!(context.frame_calls, 1);
assert_eq!(context.block_calls, 0);
assert_eq!(context.last_frame_chunk, 1);
}
#[test]
fn compress_continue_empty_frame_only_writes_header() {
let mut context = CompressContinueTestContext {
header_result: 3,
..CompressContinueTestContext::default()
};
let mut stage = ZSTD_COMPRESSION_STAGE_INIT;
let mut consumed = 7;
let mut produced = 11;
let state = compress_continue_test_state(
&mut context,
&mut stage,
&mut consumed,
&mut produced,
0,
16,
0,
);
let mut output = [0xa5u8; 8];
let result = unsafe {
ZSTD_rust_compressContinue(
&state,
output.as_mut_ptr().cast(),
output.len(),
ptr::null(),
0,
1,
0,
)
};
assert_eq!(result, 3);
assert_eq!(stage, ZSTD_COMPRESSION_STAGE_ONGOING);
assert_eq!((consumed, produced), (7, 11));
assert_eq!(context.header_calls, 1);
assert_eq!(context.update_window_calls, 0);
assert_eq!(context.frame_calls, 0);
assert_eq!(output, [0xa5; 8]);
}
#[test]
fn compress_continue_rejects_invalid_stage_and_oversized_block_before_callbacks() {
let mut context = CompressContinueTestContext::default();
let mut stage = ZSTD_COMPRESSION_STAGE_INIT;
let mut consumed = 4;
let mut produced = 9;
let mut state = compress_continue_test_state(
&mut context,
&mut stage,
&mut consumed,
&mut produced,
0,
4,
1,
);
let source = [0x22u8; 5];
let mut output = [0xa5u8; 16];
let oversized = unsafe {
ZSTD_rust_compressContinue(
&state,
output.as_mut_ptr().cast(),
output.len(),
source.as_ptr().cast(),
source.len(),
0,
0,
)
};
assert_eq!(oversized, ERROR(ZstdErrorCode::SrcSizeWrong));
assert_eq!(stage, ZSTD_COMPRESSION_STAGE_INIT);
assert_eq!((consumed, produced), (4, 9));
assert_eq!(context.update_window_calls, 0);
assert_eq!(context.block_calls, 0);
state.check_block_size = 0;
stage = ZSTD_COMPRESSION_STAGE_CREATED;
let created = unsafe {
ZSTD_rust_compressContinue(
&state,
output.as_mut_ptr().cast(),
output.len(),
source.as_ptr().cast(),
source.len(),
0,
0,
)
};
assert_eq!(created, ERROR(ZstdErrorCode::StageWrong));
assert_eq!(stage, ZSTD_COMPRESSION_STAGE_CREATED);
assert_eq!(context.update_window_calls, 0);
assert_eq!(context.block_calls, 0);
}
#[test]
fn compress_continue_propagates_block_error_without_progression() {
let mut context = CompressContinueTestContext {
block_result: ERROR(ZstdErrorCode::DstSizeTooSmall),
..CompressContinueTestContext::default()
};
let mut stage = ZSTD_COMPRESSION_STAGE_ONGOING;
let mut consumed = 4;
let mut produced = 9;
let state = compress_continue_test_state(
&mut context,
&mut stage,
&mut consumed,
&mut produced,
0,
16,
0,
);
let source = [0x33u8; 5];
let mut output = [0xa5u8; 16];
let result = unsafe {
ZSTD_rust_compressContinue(
&state,
output.as_mut_ptr().cast(),
output.len(),
source.as_ptr().cast(),
source.len(),
0,
0,
)
};
assert_eq!(result, ERROR(ZstdErrorCode::DstSizeTooSmall));
assert_eq!(
(stage, consumed, produced),
(ZSTD_COMPRESSION_STAGE_ONGOING, 4, 9)
);
assert_eq!(context.update_window_calls, 1);
assert_eq!(context.overflow_calls, 1);
assert_eq!(context.block_calls, 1);
}
#[test]
fn compress_continue_reports_pledge_overrun_after_updating_counters() {
let mut context = CompressContinueTestContext {
block_result: 3,
..CompressContinueTestContext::default()
};
let mut stage = ZSTD_COMPRESSION_STAGE_ONGOING;
let mut consumed = 0;
let mut produced = 0;
let state = compress_continue_test_state(
&mut context,
&mut stage,
&mut consumed,
&mut produced,
6,
16,
0,
);
let source = [0x44u8; 6];
let mut output = [0xa5u8; 16];
let result = unsafe {
ZSTD_rust_compressContinue(
&state,
output.as_mut_ptr().cast(),
output.len(),
source.as_ptr().cast(),
source.len(),
0,
0,
)
};
assert_eq!(result, ERROR(ZstdErrorCode::SrcSizeWrong));
assert_eq!((consumed, produced), (6, 3));
assert_eq!(context.block_calls, 1);
}
#[test]
fn simple_strategy_follows_source_size_tiers() {
assert_eq!(ZSTD_rust_compressCCtxStrategy(0, 1), ZSTD_FAST);