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

feat(compress): project split discovery into Rust

Browse Source 
Replace the frame-chunk split-discovery callback with an explicit projection of the sequence stores, block-state pointer slots, entropy metadata, parameters, and workspace required by the Rust estimator. Rust now invokes the existing split estimator directly before the Rust-owned partition loop, while C retains only private block-split storage and projection assembly.

Keep the block-state pointer slots live through estimation instead of capturing entropy-table addresses during frame preparation. Sequence-store work can change which compressed-block state is active, and following the slots preserves the original C behavior for dictionary-backed streaming compression.

Test Plan:
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo fmt --manifest-path rust/Cargo.toml --all -- --check
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml --lib
- 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
This commit is contained in:
ddidderr
2026-07-20 02:07:57 +02:00
parent 59b3c7021b
commit 93eaef18b8
2 changed files with 200 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+93 -39
View File
@@ -821,15 +821,14 @@ typedef struct ZSTD_rust_targetCBlockSizeState_s
ZSTD_rust_targetCBlockSizeState;
typedef struct ZSTD_rust_splitBlockState_s
ZSTD_rust_splitBlockState;
typedef struct ZSTD_rust_deriveBlockSplitsState_s
ZSTD_rust_deriveBlockSplitsState;
typedef size_t (*ZSTD_rust_frameChunkCompress_f)(void* context,
void* dst,
size_t dstCapacity,
const void* src,
size_t srcSize,
U32 lastBlock);
typedef size_t (*ZSTD_rust_frameChunkDeriveSplits_f)(void* context,
U32* partitions,
U32 nbSeq);
typedef struct {
void* callbackContext;
void* tmpWorkspace;
@@ -852,7 +851,7 @@ typedef struct {
const ZSTD_rust_compressContinueBlockState* compressInternalState;
const ZSTD_rust_targetCBlockSizeState* compressTargetState;
const ZSTD_rust_splitBlockState* compressSplitState;
ZSTD_rust_frameChunkDeriveSplits_f deriveBlockSplits;
const ZSTD_rust_deriveBlockSplitsState* deriveBlockSplitsState;
} ZSTD_rust_frameChunkState;
size_t ZSTD_rust_compressFrameChunk(
const ZSTD_rust_frameChunkState* state,
@@ -888,7 +887,7 @@ typedef char ZSTD_rust_frame_chunk_state_layout[
== 12 * sizeof(void*) + sizeof(S64) + 6 * sizeof(int)
&& offsetof(ZSTD_rust_frameChunkState, compressSplitState)
== 13 * sizeof(void*) + sizeof(S64) + 6 * sizeof(int)
&& offsetof(ZSTD_rust_frameChunkState, deriveBlockSplits)
&& offsetof(ZSTD_rust_frameChunkState, deriveBlockSplitsState)
== 14 * sizeof(void*) + sizeof(S64) + 6 * sizeof(int)
&& sizeof(ZSTD_rust_frameChunkState)
== 15 * sizeof(void*) + sizeof(S64) + 6 * sizeof(int))
@@ -1303,9 +1302,47 @@ typedef char ZSTD_rust_target_cblock_state_layout[
&& sizeof(ZSTD_rust_targetCBlockSizeState)
== (sizeof(void*) == 8 ? 72 : 44))
? 1 : -1];
/* The post-split partition loop only needs this projection of ZSTD_CCtx.
* Split discovery remains in the surrounding C function; the Rust body owns
* partition accounting, repcode simulation, and per-partition emission. */
/* Split discovery and the post-split partition loop use narrow projections of
* ZSTD_CCtx. The private block-split context remains in C; Rust owns the
* estimator call, partition accounting, repcode simulation, and emission. */
struct ZSTD_rust_deriveBlockSplitsState_s {
const SeqStore_t* originalSeqStore;
SeqStore_t* fullSeqStoreChunk;
SeqStore_t* firstHalfSeqStore;
SeqStore_t* secondHalfSeqStore;
ZSTD_compressedBlockState_t** prevCBlock;
ZSTD_compressedBlockState_t** nextCBlock;
ZSTD_entropyCTablesMetadata_t* entropyMetadata;
void* workspace;
size_t workspaceSize;
int strategy;
int disableLiteralCompression;
};
typedef char ZSTD_rust_derive_block_splits_state_layout[
(offsetof(ZSTD_rust_deriveBlockSplitsState, originalSeqStore) == 0
&& offsetof(ZSTD_rust_deriveBlockSplitsState, fullSeqStoreChunk)
== sizeof(void*)
&& offsetof(ZSTD_rust_deriveBlockSplitsState, firstHalfSeqStore)
== 2 * sizeof(void*)
&& offsetof(ZSTD_rust_deriveBlockSplitsState, secondHalfSeqStore)
== 3 * sizeof(void*)
&& offsetof(ZSTD_rust_deriveBlockSplitsState, prevCBlock)
== 4 * sizeof(void*)
&& offsetof(ZSTD_rust_deriveBlockSplitsState, nextCBlock)
== 5 * sizeof(void*)
&& offsetof(ZSTD_rust_deriveBlockSplitsState, entropyMetadata)
== 6 * sizeof(void*)
&& offsetof(ZSTD_rust_deriveBlockSplitsState, workspace)
== 7 * sizeof(void*)
&& offsetof(ZSTD_rust_deriveBlockSplitsState, workspaceSize)
== 8 * sizeof(void*)
&& offsetof(ZSTD_rust_deriveBlockSplitsState, strategy)
== 8 * sizeof(void*) + sizeof(size_t)
&& offsetof(ZSTD_rust_deriveBlockSplitsState, disableLiteralCompression)
== 8 * sizeof(void*) + sizeof(size_t) + sizeof(int)
&& sizeof(ZSTD_rust_deriveBlockSplitsState)
== 8 * sizeof(void*) + sizeof(size_t) + 2 * sizeof(int))
? 1 : -1];
struct ZSTD_rust_splitBlockState_s {
const SeqStore_t* seqStore;
const U32* partitions;
@@ -5301,28 +5338,6 @@ size_t ZSTD_buildBlockEntropyStats(
workspace, wkspSize);
}
/* Base recursive function.
* Populates a table with intra-block partition indices that can improve compression ratio.
*
* @return: number of splits made (which equals the size of the partition table - 1).
*/
static size_t ZSTD_deriveBlockSplits(ZSTD_CCtx* zc, U32 partitions[], U32 nbSeq)
{
return ZSTD_rust_deriveBlockSplits(
partitions, nbSeq,
&zc->seqStore,
&zc->blockSplitCtx.fullSeqStoreChunk,
&zc->blockSplitCtx.firstHalfSeqStore,
&zc->blockSplitCtx.secondHalfSeqStore,
&zc->blockState.prevCBlock->entropy,
&zc->blockState.nextCBlock->entropy,
(int)zc->appliedParams.cParams.strategy,
ZSTD_literalsCompressionIsDisabled(&zc->appliedParams),
&zc->blockSplitCtx.entropyMetadata,
zc->tmpWorkspace,
zc->tmpWkspSize);
}
/* ZSTD_compressBlock_splitBlock():
* Attempts to split a given block into multiple blocks to improve compression ratio.
*
@@ -5339,7 +5354,32 @@ ZSTD_compressBlock_splitBlock_internal(ZSTD_CCtx* zc,
size_t numSplits = 0;
if (bss == ZSTDbss_compress) {
numSplits = ZSTD_deriveBlockSplits(zc, partitions, nbSeq);
ZSTD_rust_deriveBlockSplitsState deriveState;
deriveState.originalSeqStore = &zc->seqStore;
deriveState.fullSeqStoreChunk = &zc->blockSplitCtx.fullSeqStoreChunk;
deriveState.firstHalfSeqStore = &zc->blockSplitCtx.firstHalfSeqStore;
deriveState.secondHalfSeqStore = &zc->blockSplitCtx.secondHalfSeqStore;
deriveState.prevCBlock = &zc->blockState.prevCBlock;
deriveState.nextCBlock = &zc->blockState.nextCBlock;
deriveState.entropyMetadata = &zc->blockSplitCtx.entropyMetadata;
deriveState.workspace = zc->tmpWorkspace;
deriveState.workspaceSize = zc->tmpWkspSize;
deriveState.strategy = (int)zc->appliedParams.cParams.strategy;
deriveState.disableLiteralCompression =
ZSTD_literalsCompressionIsDisabled(&zc->appliedParams);
numSplits = ZSTD_rust_deriveBlockSplits(
partitions, nbSeq,
deriveState.originalSeqStore,
deriveState.fullSeqStoreChunk,
deriveState.firstHalfSeqStore,
deriveState.secondHalfSeqStore,
&deriveState.prevCBlock[0]->entropy,
&deriveState.nextCBlock[0]->entropy,
deriveState.strategy,
deriveState.disableLiteralCompression,
deriveState.entropyMetadata,
deriveState.workspace,
deriveState.workspaceSize);
FORWARD_IF_ERROR(numSplits, "Deriving block splits failed!");
}
@@ -5443,6 +5483,7 @@ typedef struct {
ZSTD_rust_blockInternalState blockInternalState;
ZSTD_rust_compressContinueBlockState blockState;
ZSTD_rust_targetCBlockSizeState targetBlockState;
ZSTD_rust_deriveBlockSplitsState deriveBlockSplitsState;
ZSTD_rust_splitBlockState splitBlockState;
} ZSTD_rust_compressContinueContext;
@@ -5573,13 +5614,6 @@ static size_t ZSTD_rust_frameChunk_compressSplit(
(ZSTD_CCtx*)context, dst, dstCapacity, src, srcSize, lastBlock);
}
static size_t ZSTD_rust_frameChunk_deriveBlockSplits(
void* context, U32* partitions, U32 nbSeq)
{
return ZSTD_deriveBlockSplits(
(ZSTD_CCtx*)context, partitions, nbSeq);
}
static void ZSTD_compressContinue_prepare(
ZSTD_CCtx* cctx, size_t blockSizeMax, int checkBlockSize,
ZSTD_rust_compressContinueContext* context)
@@ -5631,7 +5665,8 @@ static void ZSTD_compressContinue_prepare(
context->frameChunkState.compressInternal = NULL;
context->frameChunkState.compressTargetState = &context->targetBlockState;
context->frameChunkState.compressSplitState = &context->splitBlockState;
context->frameChunkState.deriveBlockSplits = ZSTD_rust_frameChunk_deriveBlockSplits;
context->frameChunkState.deriveBlockSplitsState =
&context->deriveBlockSplitsState;
ZSTD_initBuildSeqStoreState(cctx, &context->buildSeqStoreState);
context->blockInternalState.seqStore = &cctx->seqStore;
@@ -5660,6 +5695,25 @@ static void ZSTD_compressContinue_prepare(
context->targetBlockState.windowLog = cctx->appliedParams.cParams.windowLog;
context->targetBlockState.targetCBlockSize = cctx->appliedParams.targetCBlockSize;
context->targetBlockState.isFirstBlock = cctx->isFirstBlock;
context->deriveBlockSplitsState.originalSeqStore = &cctx->seqStore;
context->deriveBlockSplitsState.fullSeqStoreChunk =
&cctx->blockSplitCtx.fullSeqStoreChunk;
context->deriveBlockSplitsState.firstHalfSeqStore =
&cctx->blockSplitCtx.firstHalfSeqStore;
context->deriveBlockSplitsState.secondHalfSeqStore =
&cctx->blockSplitCtx.secondHalfSeqStore;
context->deriveBlockSplitsState.prevCBlock =
&cctx->blockState.prevCBlock;
context->deriveBlockSplitsState.nextCBlock =
&cctx->blockState.nextCBlock;
context->deriveBlockSplitsState.entropyMetadata =
&cctx->blockSplitCtx.entropyMetadata;
context->deriveBlockSplitsState.workspace = cctx->tmpWorkspace;
context->deriveBlockSplitsState.workspaceSize = cctx->tmpWkspSize;
context->deriveBlockSplitsState.strategy =
(int)cctx->appliedParams.cParams.strategy;
context->deriveBlockSplitsState.disableLiteralCompression =
ZSTD_literalsCompressionIsDisabled(&cctx->appliedParams);
context->splitBlockState.seqStore = &cctx->seqStore;
context->splitBlockState.partitions = cctx->blockSplitCtx.partitions;
context->splitBlockState.nextSeqStore = &cctx->blockSplitCtx.nextSeqStore;
rust/src/zstd_compress.rs
+107 -24
View File
@@ -21,6 +21,7 @@ use crate::xxhash::XXH64_reset;
use crate::xxhash::{XXH64_digest, XXH64_state_t, XXH64_update};
#[cfg(not(test))]
use crate::zstd_compress_api::ZSTD_compressBound;
use crate::zstd_compress_block_split::ZSTD_rust_deriveBlockSplits;
use crate::zstd_compress_frame::{
write_raw_block, ZSTD_rust_noCompressBlock, ZSTD_rust_rleCompressBlock,
ZSTD_rust_writeBlockHeader, ZSTD_rust_writeEpilogue, ZSTD_rust_writeFrameHeader,
@@ -42,14 +43,15 @@ use crate::zstd_compress_params_api::{
use crate::zstd_compress_sequences::SeqDef;
use crate::zstd_compress_stats::{
update_rep, SeqCollector, SeqStore_t, ZSTD_Sequence, ZSTD_SequencePosition,
ZSTD_compressedBlockState_t, ZSTD_entropyCTables_t, ZSTD_rust_confirmRepcodesAndEntropyTables,
ZSTD_rust_convertSequencesNoRepcodes, ZSTD_rust_copyBlockSequences,
ZSTD_rust_countSeqStoreLiteralsBytes, ZSTD_rust_countSeqStoreMatchBytes,
ZSTD_rust_deriveSeqStoreChunk, ZSTD_rust_determineBlockSize, ZSTD_rust_entropyCompressSeqStore,
ZSTD_rust_entropyCompressSeqStore_internal, ZSTD_rust_fastSequenceLengthSum,
ZSTD_rust_finalizeOffBase, ZSTD_rust_get1BlockSummary, ZSTD_rust_isRLE, ZSTD_rust_maybeRLE,
ZSTD_rust_postProcessSequenceProducerResult, ZSTD_rust_resetCompressedBlockState,
ZSTD_rust_resetSeqStore, ZSTD_rust_seqStore_resolveOffCodes, ZSTD_rust_storeLastLiterals,
ZSTD_compressedBlockState_t, ZSTD_entropyCTablesMetadata_t, ZSTD_entropyCTables_t,
ZSTD_rust_confirmRepcodesAndEntropyTables, ZSTD_rust_convertSequencesNoRepcodes,
ZSTD_rust_copyBlockSequences, ZSTD_rust_countSeqStoreLiteralsBytes,
ZSTD_rust_countSeqStoreMatchBytes, ZSTD_rust_deriveSeqStoreChunk, ZSTD_rust_determineBlockSize,
ZSTD_rust_entropyCompressSeqStore, ZSTD_rust_entropyCompressSeqStore_internal,
ZSTD_rust_fastSequenceLengthSum, ZSTD_rust_finalizeOffBase, ZSTD_rust_get1BlockSummary,
ZSTD_rust_isRLE, ZSTD_rust_maybeRLE, ZSTD_rust_postProcessSequenceProducerResult,
ZSTD_rust_resetCompressedBlockState, ZSTD_rust_resetSeqStore,
ZSTD_rust_seqStore_resolveOffCodes, ZSTD_rust_storeLastLiterals,
ZSTD_rust_transferSequencesNoDelim, ZSTD_rust_transferSequencesWBlockDelim,
ZSTD_rust_validateSeqStore, ZSTD_LLT_LITERAL_LENGTH, ZSTD_LLT_MATCH_LENGTH,
};
@@ -180,7 +182,6 @@ type FrameChunkPrepareOverflowFn = unsafe extern "C" fn(*mut c_void, *const c_vo
type FrameChunkPrepareWindowFn = unsafe extern "C" fn(*mut c_void, *const c_void, usize, c_uint);
type FrameChunkCompressFn =
unsafe extern "C" fn(*mut c_void, *mut c_void, usize, *const c_void, usize, c_uint) -> usize;
type FrameChunkDeriveSplitsFn = unsafe extern "C" fn(*mut c_void, *mut u32, c_uint) -> usize;
type BuildSeqStoreSkipFn = unsafe extern "C" fn(*mut c_void, usize);
type BuildSeqStorePrepareFn = unsafe extern "C" fn(*mut c_void, *const c_void, usize);
@@ -560,6 +561,62 @@ const _: () = {
assert!(size_of::<ZSTD_rust_compressContinueBlockState>() == 2 * size_of::<usize>());
};
/// Explicit projection of the state used by the block-split estimator.
///
/// C retains the private block-split context and owns projection assembly;
/// Rust owns the estimator implementation and receives only its required
/// sequence stores, entropy tables, parameters, and workspace.
#[repr(C)]
pub struct ZSTD_rust_deriveBlockSplitsState {
original_seq_store: *const SeqStore_t,
full_seq_store_chunk: *mut SeqStore_t,
first_half_seq_store: *mut SeqStore_t,
second_half_seq_store: *mut SeqStore_t,
prev_c_block: *mut *mut ZSTD_compressedBlockState_t,
next_c_block: *mut *mut ZSTD_compressedBlockState_t,
entropy_metadata: *mut ZSTD_entropyCTablesMetadata_t,
workspace: *mut c_void,
workspace_size: usize,
strategy: c_int,
disable_literal_compression: c_int,
}
const _: () = {
assert!(offset_of!(ZSTD_rust_deriveBlockSplitsState, original_seq_store) == 0);
assert!(
offset_of!(ZSTD_rust_deriveBlockSplitsState, full_seq_store_chunk) == size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_deriveBlockSplitsState, first_half_seq_store)
== 2 * size_of::<usize>()
);
assert!(
offset_of!(ZSTD_rust_deriveBlockSplitsState, second_half_seq_store)
== 3 * size_of::<usize>()
);
assert!(offset_of!(ZSTD_rust_deriveBlockSplitsState, prev_c_block) == 4 * size_of::<usize>());
assert!(offset_of!(ZSTD_rust_deriveBlockSplitsState, next_c_block) == 5 * size_of::<usize>());
assert!(
offset_of!(ZSTD_rust_deriveBlockSplitsState, entropy_metadata) == 6 * size_of::<usize>()
);
assert!(offset_of!(ZSTD_rust_deriveBlockSplitsState, workspace) == 7 * size_of::<usize>());
assert!(offset_of!(ZSTD_rust_deriveBlockSplitsState, workspace_size) == usize::BITS as usize);
assert!(
offset_of!(ZSTD_rust_deriveBlockSplitsState, strategy)
== usize::BITS as usize + size_of::<usize>()
);
assert!(
offset_of!(
ZSTD_rust_deriveBlockSplitsState,
disable_literal_compression
) == usize::BITS as usize + size_of::<usize>() + size_of::<c_int>()
);
assert!(
size_of::<ZSTD_rust_deriveBlockSplitsState>()
== usize::BITS as usize + size_of::<usize>() + 2 * size_of::<c_int>()
);
};
/// Explicit projection of the state used by `ZSTD_compress_frameChunk`.
///
/// The Rust side owns the per-frame block loop and its savings/dispatch
@@ -589,7 +646,7 @@ pub struct ZSTD_rust_frameChunkState {
compress_internal_state: *const ZSTD_rust_compressContinueBlockState,
compress_target_state: *const ZSTD_rust_targetCBlockSizeState,
compress_split_state: *const ZSTD_rust_splitBlockState,
derive_block_splits: Option<FrameChunkDeriveSplitsFn>,
derive_block_splits_state: *const ZSTD_rust_deriveBlockSplitsState,
}
const _: () = {
@@ -642,7 +699,7 @@ const _: () = {
== 13 * size_of::<usize>() + size_of::<c_longlong>() + 6 * size_of::<c_int>()
);
assert!(
offset_of!(ZSTD_rust_frameChunkState, derive_block_splits)
offset_of!(ZSTD_rust_frameChunkState, derive_block_splits_state)
== 14 * size_of::<usize>() + size_of::<c_longlong>() + 6 * size_of::<c_int>()
);
assert!(
@@ -820,14 +877,40 @@ unsafe fn compress_frame_chunk_body_with(
if sequence_count < 0 {
return ERROR(ZstdErrorCode::Generic);
}
let Some(derive_block_splits) = state.derive_block_splits else {
if state.derive_block_splits_state.is_null() {
return ERROR(ZstdErrorCode::Generic);
};
}
let derive_state = unsafe { &*state.derive_block_splits_state };
if derive_state.original_seq_store.is_null()
|| derive_state.full_seq_store_chunk.is_null()
|| derive_state.first_half_seq_store.is_null()
|| derive_state.second_half_seq_store.is_null()
|| derive_state.prev_c_block.is_null()
|| derive_state.next_c_block.is_null()
|| derive_state.entropy_metadata.is_null()
{
return ERROR(ZstdErrorCode::Generic);
}
let prev_c_block = unsafe { *derive_state.prev_c_block };
let next_c_block = unsafe { *derive_state.next_c_block };
if prev_c_block.is_null() || next_c_block.is_null() {
return ERROR(ZstdErrorCode::Generic);
}
let result = unsafe {
derive_block_splits(
state.callback_context,
ZSTD_rust_deriveBlockSplits(
split_state.partitions.cast_mut(),
sequence_count as c_uint,
sequence_count as u32,
derive_state.original_seq_store,
derive_state.full_seq_store_chunk,
derive_state.first_half_seq_store,
derive_state.second_half_seq_store,
ptr::addr_of!((*prev_c_block).entropy),
ptr::addr_of_mut!((*next_c_block).entropy),
derive_state.strategy,
derive_state.disable_literal_compression,
derive_state.entropy_metadata,
derive_state.workspace,
derive_state.workspace_size,
)
};
if ERR_isError(result) {
@@ -5304,9 +5387,9 @@ fn split_single_block_state(
/// Rust implementation of the post-split partition loop.
///
/// C retains `ZSTD_deriveBlockSplits()` and the private `ZSTD_CCtx`; Rust
/// receives only the sequence-store views and block-emission state needed to
/// reproduce the original loop.
/// C retains the private `ZSTD_CCtx` and block-split storage; Rust receives
/// projected estimator inputs alongside the sequence-store views and owns the
/// partition accounting, repcode simulation, and block emission.
#[allow(clippy::too_many_arguments)]
unsafe fn compress_block_split_body_with(
state: &ZSTD_rust_splitBlockState,
@@ -10535,7 +10618,7 @@ mod tests {
compress_internal_state: ptr::null(),
compress_target_state: ptr::null(),
compress_split_state: ptr::null(),
derive_block_splits: None,
derive_block_splits_state: ptr::null(),
});
let frame_chunk_state = context
.frame_chunk_state
@@ -11414,7 +11497,7 @@ mod tests {
compress_internal_state: ptr::null(),
compress_target_state: ptr::null(),
compress_split_state: ptr::null(),
derive_block_splits: None,
derive_block_splits_state: ptr::null(),
}
}
@@ -11757,7 +11840,7 @@ mod tests {
compress_internal_state: ptr::null(),
compress_target_state: ptr::null(),
compress_split_state: ptr::null(),
derive_block_splits: None,
derive_block_splits_state: ptr::null(),
});
let frame_chunk_state = context
.frame_chunk_state
@@ -12246,7 +12329,7 @@ mod tests {
compress_internal_state: ptr::null(),
compress_target_state: ptr::null(),
compress_split_state: ptr::null(),
derive_block_splits: None,
derive_block_splits_state: ptr::null(),
});
let continue_frame_chunk_state = context
.continue_frame_chunk_state
@@ -12274,7 +12357,7 @@ mod tests {
compress_internal_state: ptr::null(),
compress_target_state: ptr::null(),
compress_split_state: ptr::null(),
derive_block_splits: None,
derive_block_splits_state: ptr::null(),
});
let end_frame_chunk_state = context
.end_frame_chunk_state