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

feat(compress): move CCtx reset tail orchestration into Rust

Browse Source 
Move the post-workspace CCtx reset order into Rust: initialize the applied
context state, reset the previous compressed-block state, reset match state,
and publish the remaining storage in sequence with error propagation between
steps. Keep private CCtx and workspace mutations in C-owned callbacks and
preserve the C/Rust state layout assertions.

Add focused tests for successful ordering and match-state/storage failure
boundaries.

Test Plan:
- `ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml`
- `ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings`
- `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`
- `ulimit -v 41943040 && cargo fmt --manifest-path rust/Cargo.toml -- --check`
- `git diff --check`
This commit is contained in:
ddidderr
2026-07-19 21:41:02 +02:00
parent f882006e7a
commit 0be6d23512
2 changed files with 279 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+101 -29
View File
@@ -1447,6 +1447,7 @@ typedef size_t (*ZSTD_rust_resetCCtxWorkspaceCreate_f)(
void* context, size_t neededSpace);
typedef void* (*ZSTD_rust_resetCCtxWorkspaceReserveObject_f)(
void* context, size_t size);
typedef size_t (*ZSTD_rust_resetCCtxTailCallback_f)(void* context);
typedef struct {
void* callbackContext;
int ldmEnable;
@@ -1559,6 +1560,24 @@ typedef char ZSTD_rust_reset_cctx_workspace_state_layout[
== offsetof(ZSTD_rust_resetCCtxWorkspaceState, clearWorkspace)
+ sizeof(void*))
? 1 : -1];
typedef struct {
void* callbackContext;
ZSTD_rust_resetCCtxStorageCallback_f initialize;
ZSTD_rust_resetCCtxStorageCallback_f resetCompressedBlockState;
ZSTD_rust_resetCCtxTailCallback_f resetMatchState;
ZSTD_rust_resetCCtxTailCallback_f resetStorage;
} ZSTD_rust_resetCCtxTailState;
typedef char ZSTD_rust_reset_cctx_tail_state_layout[
(offsetof(ZSTD_rust_resetCCtxTailState, callbackContext) == 0
&& offsetof(ZSTD_rust_resetCCtxTailState, initialize) == sizeof(void*)
&& offsetof(ZSTD_rust_resetCCtxTailState, resetCompressedBlockState)
== 2 * sizeof(void*)
&& offsetof(ZSTD_rust_resetCCtxTailState, resetMatchState)
== 3 * sizeof(void*)
&& sizeof(ZSTD_rust_resetCCtxTailState)
== offsetof(ZSTD_rust_resetCCtxTailState, resetStorage)
+ sizeof(void*))
? 1 : -1];
enum {
ZSTD_RUST_RESET_CCTX_RESERVE_ALIGNED64 = 0,
ZSTD_RUST_RESET_CCTX_RESERVE_BUFFER = 1
@@ -1595,6 +1614,7 @@ enum {
size_t ZSTD_rust_resetCCtxStorage(const ZSTD_rust_resetCCtxStorageState* state);
size_t ZSTD_rust_resetCCtxWorkspace(
const ZSTD_rust_resetCCtxWorkspaceState* state);
size_t ZSTD_rust_resetCCtxTail(const ZSTD_rust_resetCCtxTailState* state);
size_t ZSTD_rust_estimateCCtxWorkspaceSize(
ZSTD_compressionParameters cParams,
int ldmEnable, U32 ldmHashLog, U32 ldmBucketSizeLog,
@@ -4228,6 +4248,69 @@ static void ZSTD_rust_resetCCtxWorkspace_clear(void* opaque)
ZSTD_cwksp_clear(context->ws);
}
typedef struct {
ZSTD_CCtx* cctx;
const ZSTD_CCtx_params* params;
size_t blockSize;
U64 pledgedSrcSize;
ZSTD_cwksp* ws;
ZSTD_compResetPolicy_e compResetPolicy;
ZSTD_indexResetPolicy_e indexResetPolicy;
ZSTD_rust_resetCCtxStorageState* storageState;
} ZSTD_rust_resetCCtxTailContext;
static void ZSTD_rust_resetCCtxTail_initialize(void* opaque)
{
ZSTD_rust_resetCCtxTailContext* const context =
(ZSTD_rust_resetCCtxTailContext*)opaque;
ZSTD_CCtx* const cctx = context->cctx;
cctx->blockState.matchState.cParams = context->params->cParams;
cctx->blockState.matchState.prefetchCDictTables =
context->params->prefetchCDictTables == ZSTD_ps_enable;
cctx->pledgedSrcSizePlusOne = context->pledgedSrcSize + 1;
cctx->consumedSrcSize = 0;
cctx->producedCSize = 0;
if (context->pledgedSrcSize == ZSTD_CONTENTSIZE_UNKNOWN)
cctx->appliedParams.fParams.contentSizeFlag = 0;
DEBUGLOG(4, "pledged content size : %u ; flag : %u",
(unsigned)context->pledgedSrcSize,
cctx->appliedParams.fParams.contentSizeFlag);
cctx->blockSizeMax = context->blockSize;
XXH64_reset(&cctx->xxhState, 0);
cctx->stage = ZSTDcs_init;
cctx->dictID = 0;
cctx->dictContentSize = 0;
}
static void ZSTD_rust_resetCCtxTail_resetCompressedBlockState(void* opaque)
{
ZSTD_rust_resetCCtxTailContext* const context =
(ZSTD_rust_resetCCtxTailContext*)opaque;
ZSTD_reset_compressedBlockState(context->cctx->blockState.prevCBlock);
}
static size_t ZSTD_rust_resetCCtxTail_resetMatchState(void* opaque)
{
ZSTD_rust_resetCCtxTailContext* const context =
(ZSTD_rust_resetCCtxTailContext*)opaque;
return ZSTD_reset_matchState(
&context->cctx->blockState.matchState,
context->ws,
&context->params->cParams,
context->params->useRowMatchFinder,
context->compResetPolicy,
context->indexResetPolicy,
ZSTD_resetTarget_CCtx);
}
static size_t ZSTD_rust_resetCCtxTail_resetStorage(void* opaque)
{
ZSTD_rust_resetCCtxTailContext* const context =
(ZSTD_rust_resetCCtxTailContext*)opaque;
return ZSTD_rust_resetCCtxStorage(context->storageState);
}
/*! ZSTD_resetCCtx_internal() :
* @param loadedDictSize The size of the dictionary to be loaded
* into the context, if any. If no dictionary is used, or the
@@ -4367,35 +4450,24 @@ static size_t ZSTD_resetCCtx_internal(ZSTD_CCtx* zc,
FORWARD_IF_ERROR(ZSTD_rust_resetCCtxWorkspace(&workspaceState), "");
}
/* init params */
zc->blockState.matchState.cParams = params->cParams;
zc->blockState.matchState.prefetchCDictTables = params->prefetchCDictTables == ZSTD_ps_enable;
zc->pledgedSrcSizePlusOne = pledgedSrcSize+1;
zc->consumedSrcSize = 0;
zc->producedCSize = 0;
if (pledgedSrcSize == ZSTD_CONTENTSIZE_UNKNOWN)
zc->appliedParams.fParams.contentSizeFlag = 0;
DEBUGLOG(4, "pledged content size : %u ; flag : %u",
(unsigned)pledgedSrcSize, zc->appliedParams.fParams.contentSizeFlag);
zc->blockSizeMax = blockSize;
XXH64_reset(&zc->xxhState, 0);
zc->stage = ZSTDcs_init;
zc->dictID = 0;
zc->dictContentSize = 0;
ZSTD_reset_compressedBlockState(zc->blockState.prevCBlock);
FORWARD_IF_ERROR(ZSTD_reset_matchState(
&zc->blockState.matchState,
ws,
&params->cParams,
params->useRowMatchFinder,
crp,
needsIndexReset,
ZSTD_resetTarget_CCtx), "");
FORWARD_IF_ERROR(ZSTD_rust_resetCCtxStorage(&storageState), "");
{ ZSTD_rust_resetCCtxTailContext tailContext;
ZSTD_rust_resetCCtxTailState tailState;
tailContext.cctx = zc;
tailContext.params = params;
tailContext.blockSize = blockSize;
tailContext.pledgedSrcSize = pledgedSrcSize;
tailContext.ws = ws;
tailContext.compResetPolicy = crp;
tailContext.indexResetPolicy = needsIndexReset;
tailContext.storageState = &storageState;
tailState.callbackContext = &tailContext;
tailState.initialize = ZSTD_rust_resetCCtxTail_initialize;
tailState.resetCompressedBlockState =
ZSTD_rust_resetCCtxTail_resetCompressedBlockState;
tailState.resetMatchState = ZSTD_rust_resetCCtxTail_resetMatchState;
tailState.resetStorage = ZSTD_rust_resetCCtxTail_resetStorage;
FORWARD_IF_ERROR(ZSTD_rust_resetCCtxTail(&tailState), "");
}
DEBUGLOG(3, "wksp: finished allocating, %zd bytes remain available", ZSTD_cwksp_available_space(ws));
assert(ZSTD_cwksp_estimated_space_within_bounds(ws, neededSpace));
rust/src/zstd_compress.rs
+178
View File
@@ -6720,6 +6720,7 @@ const _: () = {
type ResetCCtxWorkspaceCreate = unsafe extern "C" fn(*mut c_void, usize) -> usize;
type ResetCCtxWorkspaceReserveObject = unsafe extern "C" fn(*mut c_void, usize) -> *mut c_void;
type ResetCCtxTailCallback = unsafe extern "C" fn(*mut c_void) -> usize;
/// C-owned workspace operations used by the CCtx reset policy.
///
@@ -6800,6 +6801,34 @@ const _: () = {
);
};
/// C-owned operations used by the post-workspace CCtx reset tail.
///
/// Rust owns the operation order and error propagation. C retains the
/// private context initialization, compressed-block reset, match-state reset,
/// and storage publication operations.
#[repr(C)]
pub struct ZSTD_rust_resetCCtxTailState {
callbackContext: *mut c_void,
initialize: Option<ResetCCtxStorageCallback>,
resetCompressedBlockState: Option<ResetCCtxStorageCallback>,
resetMatchState: Option<ResetCCtxTailCallback>,
resetStorage: Option<ResetCCtxTailCallback>,
}
const _: () = {
assert!(offset_of!(ZSTD_rust_resetCCtxTailState, callbackContext) == 0);
assert!(offset_of!(ZSTD_rust_resetCCtxTailState, initialize) == size_of::<usize>());
assert!(
offset_of!(ZSTD_rust_resetCCtxTailState, resetCompressedBlockState)
== 2 * size_of::<usize>()
);
assert!(offset_of!(ZSTD_rust_resetCCtxTailState, resetMatchState) == 3 * size_of::<usize>());
assert!(
size_of::<ZSTD_rust_resetCCtxTailState>()
== offset_of!(ZSTD_rust_resetCCtxTailState, resetStorage) + size_of::<usize>()
);
};
/// Reserve and publish the private CCtx storage that follows match-state reset.
#[no_mangle]
pub unsafe extern "C" fn ZSTD_rust_resetCCtxStorage(
@@ -7082,6 +7111,46 @@ pub unsafe extern "C" fn ZSTD_rust_resetCCtxWorkspace(
0
}
/// Own the callback order for the post-workspace CCtx reset tail.
#[no_mangle]
pub unsafe extern "C" fn ZSTD_rust_resetCCtxTail(
state: *const ZSTD_rust_resetCCtxTailState,
) -> usize {
if state.is_null() {
return ERROR(ZstdErrorCode::Generic);
}
let state = unsafe { &*state };
if state.callbackContext.is_null() {
return ERROR(ZstdErrorCode::Generic);
}
let Some(initialize) = state.initialize else {
return ERROR(ZstdErrorCode::Generic);
};
let Some(reset_compressed_block_state) = state.resetCompressedBlockState else {
return ERROR(ZstdErrorCode::Generic);
};
let Some(reset_match_state) = state.resetMatchState else {
return ERROR(ZstdErrorCode::Generic);
};
let Some(reset_storage) = state.resetStorage else {
return ERROR(ZstdErrorCode::Generic);
};
unsafe {
initialize(state.callbackContext);
reset_compressed_block_state(state.callbackContext);
let result = reset_match_state(state.callbackContext);
if ERR_isError(result) {
return result;
}
let result = reset_storage(state.callbackContext);
if ERR_isError(result) {
return result;
}
}
0
}
#[inline]
fn max_estimate_cctx_size(
estimate0: usize,
@@ -16653,6 +16722,115 @@ mod tests {
assert!(context.events.is_empty());
}
#[derive(Default)]
struct ResetCCtxTailTestContext {
events: Vec<&'static str>,
match_result: usize,
storage_result: usize,
}
unsafe fn reset_cctx_tail_test_context(
context: *mut c_void,
) -> &'static mut ResetCCtxTailTestContext {
unsafe { &mut *context.cast::<ResetCCtxTailTestContext>() }
}
unsafe extern "C" fn reset_cctx_tail_test_initialize(context: *mut c_void) {
let context = unsafe { reset_cctx_tail_test_context(context) };
context.events.push("initialize");
}
unsafe extern "C" fn reset_cctx_tail_test_reset_compressed_block_state(context: *mut c_void) {
let context = unsafe { reset_cctx_tail_test_context(context) };
context.events.push("reset-compressed-block-state");
}
unsafe extern "C" fn reset_cctx_tail_test_reset_match_state(context: *mut c_void) -> usize {
let context = unsafe { reset_cctx_tail_test_context(context) };
context.events.push("reset-match-state");
context.match_result
}
unsafe extern "C" fn reset_cctx_tail_test_reset_storage(context: *mut c_void) -> usize {
let context = unsafe { reset_cctx_tail_test_context(context) };
context.events.push("reset-storage");
context.storage_result
}
fn reset_cctx_tail_test_state(
context: &mut ResetCCtxTailTestContext,
) -> ZSTD_rust_resetCCtxTailState {
ZSTD_rust_resetCCtxTailState {
callbackContext: (context as *mut ResetCCtxTailTestContext).cast(),
initialize: Some(reset_cctx_tail_test_initialize),
resetCompressedBlockState: Some(reset_cctx_tail_test_reset_compressed_block_state),
resetMatchState: Some(reset_cctx_tail_test_reset_match_state),
resetStorage: Some(reset_cctx_tail_test_reset_storage),
}
}
#[test]
fn cctx_reset_tail_preserves_post_workspace_callback_order() {
let mut context = ResetCCtxTailTestContext::default();
let state = reset_cctx_tail_test_state(&mut context);
assert_eq!(unsafe { ZSTD_rust_resetCCtxTail(&state) }, 0);
assert_eq!(
context.events,
[
"initialize",
"reset-compressed-block-state",
"reset-match-state",
"reset-storage",
]
);
}
#[test]
fn cctx_reset_tail_stops_before_storage_after_match_state_error() {
let mut context = ResetCCtxTailTestContext {
match_result: ERROR(ZstdErrorCode::StageWrong),
..ResetCCtxTailTestContext::default()
};
let state = reset_cctx_tail_test_state(&mut context);
assert_eq!(
unsafe { ZSTD_rust_resetCCtxTail(&state) },
ERROR(ZstdErrorCode::StageWrong)
);
assert_eq!(
context.events,
[
"initialize",
"reset-compressed-block-state",
"reset-match-state"
]
);
}
#[test]
fn cctx_reset_tail_propagates_storage_error_after_match_state() {
let mut context = ResetCCtxTailTestContext {
storage_result: ERROR(ZstdErrorCode::MemoryAllocation),
..ResetCCtxTailTestContext::default()
};
let state = reset_cctx_tail_test_state(&mut context);
assert_eq!(
unsafe { ZSTD_rust_resetCCtxTail(&state) },
ERROR(ZstdErrorCode::MemoryAllocation)
);
assert_eq!(
context.events,
[
"initialize",
"reset-compressed-block-state",
"reset-match-state",
"reset-storage",
]
);
}
#[test]
fn estimate_cctx_workspace_size_keeps_static_and_buffer_components_separate() {
let sizing = cctx_workspace_test_sizing();