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

refactor(compress): move stable stream init policy to Rust

Browse Source 
ZSTD_compressStream2_c still owns private context mutation, diagnostics, buffer updates, and codec initialization, but its stable-input transparent-initialization decision was a remaining scalar policy island in C. Move the continuation validation, block-size threshold, format-specific progress hint, and wrapping size arithmetic behind a Rust projection. Preserve the original C error messages and return values while keeping all private context state on the C side.\n\nThe bridge uses explicit ABI layout assertions for the mixed pointer/size projection. Focused Rust tests cover buffered and non-continue calls, block-boundary initialization, empty and short stable input, both frame formats, matching continuation state, invalid source/position, and null bridge input.\n\nTest Plan:\n- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check\n- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings\n- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings\n- Full capped native and original-test verification follows after the batch is committed.
This commit is contained in:
ddidderr
2026-07-21 06:28:29 +02:00
parent b0e90c39e2
commit 85e9457fcc
2 changed files with 401 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+73 -11
View File
@@ -807,6 +807,51 @@ typedef char ZSTD_rust_compress_stream2_policy_state_layout[
(offsetof(ZSTD_rust_compressStream2PolicyState, endOp) == 0
&& sizeof(ZSTD_rust_compressStream2PolicyState) == sizeof(int))
? 1 : -1];
typedef struct {
int inBufferMode;
int endOp;
size_t inputSize;
size_t stableInNotConsumed;
const void* inputSrc;
size_t inputPos;
const void* expectedInputSrc;
size_t expectedInputSize;
int format;
} ZSTD_rust_compressStream2InitPolicyState;
enum {
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE = 0,
/* These values are the exact progress hints returned by the C API. */
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS = 2,
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD = 6,
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID = -1,
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID = -2
};
int ZSTD_rust_compressStream2InitPolicy(
const ZSTD_rust_compressStream2InitPolicyState* state);
typedef char ZSTD_rust_compress_stream2_init_policy_state_layout[
(offsetof(ZSTD_rust_compressStream2InitPolicyState, inBufferMode) == 0
&& offsetof(ZSTD_rust_compressStream2InitPolicyState, endOp)
== sizeof(int)
&& offsetof(ZSTD_rust_compressStream2InitPolicyState, inputSize)
== 2 * sizeof(int)
&& offsetof(ZSTD_rust_compressStream2InitPolicyState,
stableInNotConsumed)
== 2 * sizeof(int) + sizeof(size_t)
&& offsetof(ZSTD_rust_compressStream2InitPolicyState, inputSrc)
== 2 * sizeof(int) + 2 * sizeof(size_t)
&& offsetof(ZSTD_rust_compressStream2InitPolicyState, inputPos)
== 2 * sizeof(int) + 3 * sizeof(size_t)
&& offsetof(ZSTD_rust_compressStream2InitPolicyState,
expectedInputSrc)
== 2 * sizeof(int) + 4 * sizeof(size_t)
&& offsetof(ZSTD_rust_compressStream2InitPolicyState,
expectedInputSize)
== 2 * sizeof(int) + 5 * sizeof(size_t)
&& offsetof(ZSTD_rust_compressStream2InitPolicyState, format)
== 2 * sizeof(int) + 6 * sizeof(size_t)
&& sizeof(ZSTD_rust_compressStream2InitPolicyState)
== 2 * sizeof(int) + 7 * sizeof(size_t))
? 1 : -1];
int ZSTD_rust_simpleCompress2Level(const void* cctx, size_t srcSize);
typedef int (*ZSTD_rust_simpleCompress2Level_f)(const void* cctx, size_t srcSize);
typedef struct {
@@ -8351,15 +8396,28 @@ size_t ZSTD_compressStream2_c( ZSTD_CCtx* cctx,
if (cctx->streamStage == zcss_init) {
size_t const inputSize = input->size - input->pos; /* no obligation to start from pos==0 */
size_t const totalInputSize = inputSize + cctx->stableIn_notConsumed;
if ( (cctx->requestedParams.inBufferMode == ZSTD_bm_stable) /* input is presumed stable, across invocations */
&& (endOp == ZSTD_e_continue) /* no flush requested, more input to come */
&& (totalInputSize < ZSTD_BLOCKSIZE_MAX) ) { /* not even reached one block yet */
if (cctx->stableIn_notConsumed) { /* not the first time */
/* check stable source guarantees */
RETURN_ERROR_IF(input->src != cctx->expectedInBuffer.src, stabilityCondition_notRespected, "stableInBuffer condition not respected: wrong src pointer");
RETURN_ERROR_IF(input->pos != cctx->expectedInBuffer.size, stabilityCondition_notRespected, "stableInBuffer condition not respected: externally modified pos");
}
ZSTD_rust_compressStream2InitPolicyState const state = {
(int)cctx->requestedParams.inBufferMode,
(int)endOp,
inputSize,
cctx->stableIn_notConsumed,
input->src,
input->pos,
cctx->expectedInBuffer.src,
cctx->expectedInBuffer.size,
(int)cctx->requestedParams.format
};
int const initPolicy = ZSTD_rust_compressStream2InitPolicy(&state);
RETURN_ERROR_IF(
initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID,
stabilityCondition_notRespected,
"stableInBuffer condition not respected: wrong src pointer");
RETURN_ERROR_IF(
initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID,
stabilityCondition_notRespected,
"stableInBuffer condition not respected: externally modified pos");
if (initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS
|| initPolicy == ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD) {
/* pretend input was consumed, to give a sense forward progress */
input->pos = input->size;
/* save stable inBuffer, for later control, and flush/end */
@@ -8367,9 +8425,13 @@ size_t ZSTD_compressStream2_c( ZSTD_CCtx* cctx,
/* but actually input wasn't consumed, so keep track of position from where compression shall resume */
cctx->stableIn_notConsumed += inputSize;
/* don't initialize yet, wait for the first block of flush() order, for better parameters adaptation */
return ZSTD_FRAMEHEADERSIZE_MIN(cctx->requestedParams.format); /* at least some header to produce */
return (size_t)initPolicy; /* at least some header to produce */
}
FORWARD_IF_ERROR(ZSTD_CCtx_init_compressStream2(cctx, endOp, totalInputSize), "compressStream2 initialization failed");
FORWARD_IF_ERROR(
ZSTD_CCtx_init_compressStream2(
cctx, endOp,
inputSize + cctx->stableIn_notConsumed),
"compressStream2 initialization failed");
ZSTD_setBufferExpectations(cctx, output, input); /* Set initial buffer expectations now that we've initialized */
}
/* end of transparent initialization stage */
rust/src/zstd_compress.rs
+328
View File
@@ -163,6 +163,7 @@ const ZSTD_CHUNKSIZE_MAX: usize = u32::MAX as usize - ZSTD_CURRENT_MAX;
const ZSTD_E_END: c_int = 2;
const ZSTD_E_CONTINUE: c_int = 0;
const ZSTD_E_FLUSH: c_int = 1;
const ZSTD_F_ZSTD1: c_int = 0;
const ZSTD_RUST_COMPRESS_STREAM2_BUFFER_VALID: c_int = 0;
const ZSTD_RUST_COMPRESS_STREAM2_BUFFER_OUTPUT_INVALID: c_int = 1;
@@ -256,6 +257,110 @@ pub unsafe extern "C" fn ZSTD_rust_compressStream2Policy(
compress_stream2_policy(state.end_op)
}
/// Projection for the stable-input part of `ZSTD_compressStream2_c`'s
/// transparent initialization stage.
///
/// Rust owns the scalar decision, including the stable-buffer continuation
/// checks and the exact progress hint. C retains the input-buffer mutation,
/// error reporting, context initialization, and all private context state.
#[repr(C)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ZSTD_rust_compressStream2InitPolicyState {
in_buffer_mode: c_int,
end_op: c_int,
input_size: usize,
stable_in_not_consumed: usize,
input_src: *const c_void,
input_pos: usize,
expected_input_src: *const c_void,
expected_input_size: usize,
format: c_int,
}
const ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE: c_int = 0;
/* These values are the exact progress hints returned by the C API. */
const ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS: c_int = 2;
const ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD: c_int = 6;
const ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID: c_int = -1;
const ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID: c_int = -2;
const _: () = {
assert!(offset_of!(ZSTD_rust_compressStream2InitPolicyState, in_buffer_mode) == 0);
assert!(offset_of!(ZSTD_rust_compressStream2InitPolicyState, end_op) == size_of::<c_int>());
assert!(
offset_of!(ZSTD_rust_compressStream2InitPolicyState, input_size) == 2 * size_of::<c_int>()
);
assert!(
offset_of!(
ZSTD_rust_compressStream2InitPolicyState,
stable_in_not_consumed
) == 2 * size_of::<c_int>() + size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_compressStream2InitPolicyState, input_src)
== 2 * size_of::<c_int>() + 2 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_compressStream2InitPolicyState, input_pos)
== 2 * size_of::<c_int>() + 3 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_compressStream2InitPolicyState, expected_input_src)
== 2 * size_of::<c_int>() + 4 * size_of::<usize>()
);
assert!(
offset_of!(
ZSTD_rust_compressStream2InitPolicyState,
expected_input_size
) == 2 * size_of::<c_int>() + 5 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_compressStream2InitPolicyState, format)
== 2 * size_of::<c_int>() + 6 * size_of::<usize>()
);
assert!(
size_of::<ZSTD_rust_compressStream2InitPolicyState>()
== 2 * size_of::<c_int>() + 7 * size_of::<usize>()
);
};
#[inline]
fn compress_stream2_init_policy(state: &ZSTD_rust_compressStream2InitPolicyState) -> c_int {
let total_input_size = state.input_size.wrapping_add(state.stable_in_not_consumed);
if state.in_buffer_mode != ZSTD_BM_STABLE
|| state.end_op != ZSTD_E_CONTINUE
|| total_input_size >= ZSTD_BLOCKSIZE_MAX
{
return ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE;
}
if state.stable_in_not_consumed != 0 {
if state.input_src != state.expected_input_src {
return ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID;
}
if state.input_pos != state.expected_input_size {
return ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID;
}
}
if state.format == ZSTD_F_ZSTD1 {
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD
} else {
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS
}
}
/// Classify whether transparent stream initialization must be deferred.
#[no_mangle]
pub unsafe extern "C" fn ZSTD_rust_compressStream2InitPolicy(
state: *const ZSTD_rust_compressStream2InitPolicyState,
) -> c_int {
let Some(state) = (unsafe { state.as_ref() }) else {
return ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE;
};
compress_stream2_init_policy(state)
}
const ZSTD_C_WINDOW_LOG: c_int = 101;
const ZSTD_C_HASH_LOG: c_int = 102;
const ZSTD_C_CHAIN_LOG: c_int = 103;
@@ -10413,6 +10518,7 @@ mod tests {
const ZSTD_BTULTRA: c_int = 8;
const ZSTD_BTOPT: c_int = 7;
const ZSTD_BTULTRA2: c_int = 9;
const ZSTD_F_ZSTD1_MAGICLESS: c_int = 1;
struct EstimateCCtxSizeTestContext {
compression_levels: Vec<c_int>,
@@ -14220,6 +14326,228 @@ mod tests {
assert_eq!(unsafe { ZSTD_rust_compressStream2Policy(ptr::null()) }, 0);
}
fn compress_stream2_init_policy_state(
in_buffer_mode: c_int,
end_op: c_int,
input_size: usize,
stable_in_not_consumed: usize,
input_src: *const c_void,
input_pos: usize,
expected_input_src: *const c_void,
expected_input_size: usize,
format: c_int,
) -> ZSTD_rust_compressStream2InitPolicyState {
ZSTD_rust_compressStream2InitPolicyState {
in_buffer_mode,
end_op,
input_size,
stable_in_not_consumed,
input_src,
input_pos,
expected_input_src,
expected_input_size,
format,
}
}
#[test]
fn compress_stream2_init_policy_initializes_for_non_stable_or_flush_calls() {
let source = [0u8; 1];
for (in_buffer_mode, end_op) in [
(ZSTD_BM_BUFFERED, ZSTD_E_CONTINUE),
(ZSTD_BM_STABLE, ZSTD_E_FLUSH),
(ZSTD_BM_STABLE, ZSTD_E_END),
] {
let state = compress_stream2_init_policy_state(
in_buffer_mode,
end_op,
1,
0,
source.as_ptr().cast(),
0,
ptr::null(),
0,
ZSTD_F_ZSTD1,
);
assert_eq!(
compress_stream2_init_policy(&state),
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE
);
}
}
#[test]
fn compress_stream2_init_policy_initializes_at_block_boundary() {
let source = [0u8; 1];
for (input_size, stable_in_not_consumed) in
[(ZSTD_BLOCKSIZE_MAX, 0), (ZSTD_BLOCKSIZE_MAX - 1, 1)]
{
let state = compress_stream2_init_policy_state(
ZSTD_BM_STABLE,
ZSTD_E_CONTINUE,
input_size,
stable_in_not_consumed,
source.as_ptr().cast(),
0,
ptr::null(),
0,
ZSTD_F_ZSTD1,
);
assert_eq!(
compress_stream2_init_policy(&state),
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE
);
}
}
#[test]
fn compress_stream2_init_policy_returns_zstd_progress_hint_for_first_short_stable_input() {
let source = [0u8; 1];
let state = compress_stream2_init_policy_state(
ZSTD_BM_STABLE,
ZSTD_E_CONTINUE,
1,
0,
source.as_ptr().cast(),
0,
ptr::null(),
0,
ZSTD_F_ZSTD1,
);
assert_eq!(
compress_stream2_init_policy(&state),
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD
);
}
#[test]
fn compress_stream2_init_policy_defers_an_empty_stable_input() {
let state = compress_stream2_init_policy_state(
ZSTD_BM_STABLE,
ZSTD_E_CONTINUE,
0,
0,
ptr::null(),
0,
ptr::null(),
0,
ZSTD_F_ZSTD1,
);
assert_eq!(
compress_stream2_init_policy(&state),
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD
);
}
#[test]
fn compress_stream2_init_policy_accepts_matching_stable_continuation() {
let source = [0u8; 1];
let state = compress_stream2_init_policy_state(
ZSTD_BM_STABLE,
ZSTD_E_CONTINUE,
3,
4,
source.as_ptr().cast(),
8,
source.as_ptr().cast(),
8,
ZSTD_F_ZSTD1,
);
assert_eq!(
compress_stream2_init_policy(&state),
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD
);
}
#[test]
fn compress_stream2_init_policy_returns_magicless_progress_hint() {
let source = [0u8; 1];
let state = compress_stream2_init_policy_state(
ZSTD_BM_STABLE,
ZSTD_E_CONTINUE,
1,
0,
source.as_ptr().cast(),
0,
ptr::null(),
0,
ZSTD_F_ZSTD1_MAGICLESS,
);
assert_eq!(
compress_stream2_init_policy(&state),
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_MAGICLESS
);
}
#[test]
fn compress_stream2_init_policy_checks_source_before_position() {
let source = [0u8; 1];
let other_source = [0u8; 1];
let state = compress_stream2_init_policy_state(
ZSTD_BM_STABLE,
ZSTD_E_CONTINUE,
1,
1,
source.as_ptr().cast(),
7,
other_source.as_ptr().cast(),
8,
ZSTD_F_ZSTD1,
);
assert_eq!(
compress_stream2_init_policy(&state),
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_SRC_INVALID
);
}
#[test]
fn compress_stream2_init_policy_rejects_modified_stable_position() {
let source = [0u8; 1];
let state = compress_stream2_init_policy_state(
ZSTD_BM_STABLE,
ZSTD_E_CONTINUE,
1,
1,
source.as_ptr().cast(),
7,
source.as_ptr().cast(),
8,
ZSTD_F_ZSTD1,
);
assert_eq!(
compress_stream2_init_policy(&state),
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_STABLE_POS_INVALID
);
}
#[test]
fn compress_stream2_init_policy_wraps_total_input_like_c() {
let source = [0u8; 1];
let state = compress_stream2_init_policy_state(
ZSTD_BM_STABLE,
ZSTD_E_CONTINUE,
1,
usize::MAX,
source.as_ptr().cast(),
0,
source.as_ptr().cast(),
0,
ZSTD_F_ZSTD1,
);
assert_eq!(
compress_stream2_init_policy(&state),
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_DEFER_ZSTD
);
}
#[test]
fn compress_stream2_init_policy_null_state_is_conservative() {
assert_eq!(
unsafe { ZSTD_rust_compressStream2InitPolicy(ptr::null()) },
ZSTD_RUST_COMPRESS_STREAM2_INIT_POLICY_INITIALIZE
);
}
#[test]
fn reduce_table_applies_threshold_and_wrapping_subtraction() {
let mut table = [0, 1, 2, 3, 4, 5, 6, u32::MAX, 0, 0, 0, 0, 0, 0, 0, 0];