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

feat(compress): move frame-chunk loop into Rust

Browse Source 
Port ZSTD_compress_frameChunk's per-block orchestration into Rust behind an
explicit C-layout projection and callback table. Rust now owns block sizing,
target/split/internal dispatch, block framing and accounting, checksum
sequencing, and terminal frame-state updates. C retains CCtx and match-state
preparation plus the codec-specific callbacks.

Test Plan:
- cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1
- cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings
- make -B -C lib -j2 lib
- make -B -C programs -j2 zstd
- make -B -C tests -j2 test-zstd
This commit is contained in:
ddidderr
2026-07-18 22:08:37 +02:00
parent 896397f729
commit f3bc5e98f1
2 changed files with 630 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+142 -97
View File
@@ -79,6 +79,74 @@ void ZSTD_rust_copyCDictTableIntoCCtx(U32* dst, U32 const* src,
U64 ZSTD_rust_advanceHashSalt(U64 hashSalt, U64 hashSaltEntropy);
int ZSTD_rust_indexTooCloseToMax(size_t nextSrcBaseOffset);
int ZSTD_rust_dictTooBig(size_t loadedDictSize);
/* The frame-chunk loop is Rust-owned. Its callbacks keep the private
* ZSTD_CCtx and match-state layout in C: Rust only drives the block loop and
* passes this context back to these C-owned state-preparation/dispatch seams. */
typedef void (*ZSTD_rust_frameChunkPrepare_f)(void* context,
const void* src,
size_t blockSize);
typedef size_t (*ZSTD_rust_frameChunkCompress_f)(void* context,
void* dst,
size_t dstCapacity,
const void* src,
size_t srcSize,
U32 lastBlock);
typedef void (*ZSTD_rust_frameChunkChecksum_f)(void* state,
const void* src,
size_t srcSize);
typedef struct {
void* callbackContext;
void* tmpWorkspace;
void* checksumState;
int* isFirstBlock;
ZSTD_compressionStage_e* stage;
size_t tmpWkspSize;
size_t blockSizeMax;
S64 savings;
int preBlockSplitterLevel;
int strategy;
int useTargetCBlockSize;
int blockSplitterEnabled;
int checksumFlag;
int endingStage;
ZSTD_rust_frameChunkPrepare_f prepareBlock;
ZSTD_rust_frameChunkCompress_f compressTarget;
ZSTD_rust_frameChunkCompress_f compressSplit;
ZSTD_rust_frameChunkCompress_f compressInternal;
ZSTD_rust_frameChunkChecksum_f updateChecksum;
} ZSTD_rust_frameChunkState;
size_t ZSTD_rust_compressFrameChunk(
const ZSTD_rust_frameChunkState* state,
void* dst, size_t dstCapacity,
const void* src, size_t srcSize,
U32 lastFrameChunk);
typedef char ZSTD_rust_frame_chunk_state_layout[
(offsetof(ZSTD_rust_frameChunkState, callbackContext) == 0
&& offsetof(ZSTD_rust_frameChunkState, tmpWorkspace) == sizeof(void*)
&& offsetof(ZSTD_rust_frameChunkState, checksumState) == 2 * sizeof(void*)
&& offsetof(ZSTD_rust_frameChunkState, isFirstBlock) == 3 * sizeof(void*)
&& offsetof(ZSTD_rust_frameChunkState, stage) == 4 * sizeof(void*)
&& offsetof(ZSTD_rust_frameChunkState, tmpWkspSize) == 5 * sizeof(void*)
&& offsetof(ZSTD_rust_frameChunkState, blockSizeMax) == 6 * sizeof(void*)
&& offsetof(ZSTD_rust_frameChunkState, savings) == 7 * sizeof(void*)
&& offsetof(ZSTD_rust_frameChunkState, preBlockSplitterLevel)
== 7 * sizeof(void*) + sizeof(S64)
&& offsetof(ZSTD_rust_frameChunkState, strategy)
== 7 * sizeof(void*) + sizeof(S64) + sizeof(int)
&& offsetof(ZSTD_rust_frameChunkState, useTargetCBlockSize)
== 7 * sizeof(void*) + sizeof(S64) + 2 * sizeof(int)
&& offsetof(ZSTD_rust_frameChunkState, blockSplitterEnabled)
== 7 * sizeof(void*) + sizeof(S64) + 3 * sizeof(int)
&& offsetof(ZSTD_rust_frameChunkState, checksumFlag)
== 7 * sizeof(void*) + sizeof(S64) + 4 * sizeof(int)
&& offsetof(ZSTD_rust_frameChunkState, endingStage)
== 7 * sizeof(void*) + sizeof(S64) + 5 * sizeof(int)
&& offsetof(ZSTD_rust_frameChunkState, prepareBlock)
== 7 * sizeof(void*) + sizeof(S64) + 6 * sizeof(int)
&& sizeof(ZSTD_rust_frameChunkState)
== 12 * sizeof(void*) + sizeof(S64) + 6 * sizeof(int))
? 1 : -1];
/* The target-sized block body only needs this narrow projection of ZSTD_CCtx.
* Matchfinder/window state, sequence-store construction, and outer repeat-mode
* cleanup remain in C. */
@@ -2896,11 +2964,59 @@ static void ZSTD_overflowCorrectIfNeeded(ZSTD_MatchState_t* ms,
#include "zstd_preSplit.h"
static size_t ZSTD_optimalBlockSize(ZSTD_CCtx* cctx, const void* src, size_t srcSize, size_t blockSizeMax, int splitLevel, ZSTD_strategy strat, S64 savings)
static void ZSTD_rust_frameChunk_prepareBlock(void* context,
const void* src,
size_t blockSize)
{
return ZSTD_rust_optimalBlockSize(src, srcSize, blockSizeMax, splitLevel,
(int)strat, savings,
cctx->tmpWorkspace, cctx->tmpWkspSize);
ZSTD_CCtx* const cctx = (ZSTD_CCtx*)context;
ZSTD_MatchState_t* const ms = &cctx->blockState.matchState;
U32 const maxDist = (U32)1 << cctx->appliedParams.cParams.windowLog;
ZSTD_overflowCorrectIfNeeded(
ms, &cctx->workspace, &cctx->appliedParams,
src, (const BYTE*)src + blockSize);
ZSTD_checkDictValidity(
&ms->window, (const BYTE*)src + blockSize, maxDist,
&ms->loadedDictEnd, &ms->dictMatchState);
ZSTD_window_enforceMaxDist(
&ms->window, src, maxDist,
&ms->loadedDictEnd, &ms->dictMatchState);
/* Ensure hash/chain table insertion resumes no sooner than lowlimit. */
if (ms->nextToUpdate < ms->window.lowLimit)
ms->nextToUpdate = ms->window.lowLimit;
}
static size_t ZSTD_rust_frameChunk_compressTarget(
void* context, void* dst, size_t dstCapacity,
const void* src, size_t srcSize, U32 lastBlock)
{
return ZSTD_compressBlock_targetCBlockSize(
(ZSTD_CCtx*)context, dst, dstCapacity, src, srcSize, lastBlock);
}
static size_t ZSTD_rust_frameChunk_compressSplit(
void* context, void* dst, size_t dstCapacity,
const void* src, size_t srcSize, U32 lastBlock)
{
return ZSTD_compressBlock_splitBlock(
(ZSTD_CCtx*)context, dst, dstCapacity, src, srcSize, lastBlock);
}
static size_t ZSTD_rust_frameChunk_compressInternal(
void* context, void* dst, size_t dstCapacity,
const void* src, size_t srcSize, U32 lastBlock)
{
(void)lastBlock;
return ZSTD_compressBlock_internal(
(ZSTD_CCtx*)context, dst, dstCapacity, src, srcSize,
1 /* frame */);
}
static void ZSTD_rust_frameChunk_updateChecksum(
void* state, const void* src, size_t srcSize)
{
(void)XXH64_update((XXH64_state_t*)state, src, srcSize);
}
/*! ZSTD_compress_frameChunk() :
@@ -2915,99 +3031,28 @@ static size_t ZSTD_compress_frameChunk(ZSTD_CCtx* cctx,
const void* src, size_t srcSize,
U32 lastFrameChunk)
{
size_t blockSizeMax = cctx->blockSizeMax;
size_t remaining = srcSize;
const BYTE* ip = (const BYTE*)src;
BYTE* const ostart = (BYTE*)dst;
BYTE* op = ostart;
U32 const maxDist = (U32)1 << cctx->appliedParams.cParams.windowLog;
S64 savings = (S64)cctx->consumedSrcSize - (S64)cctx->producedCSize;
assert(cctx->appliedParams.cParams.windowLog <= ZSTD_WINDOWLOG_MAX);
DEBUGLOG(5, "ZSTD_compress_frameChunk (srcSize=%u, blockSizeMax=%u)", (unsigned)srcSize, (unsigned)blockSizeMax);
if (cctx->appliedParams.fParams.checksumFlag && srcSize)
XXH64_update(&cctx->xxhState, src, srcSize);
while (remaining) {
ZSTD_MatchState_t* const ms = &cctx->blockState.matchState;
size_t const blockSize = ZSTD_optimalBlockSize(cctx,
ip, remaining,
blockSizeMax,
cctx->appliedParams.preBlockSplitter_level,
cctx->appliedParams.cParams.strategy,
savings);
U32 const lastBlock = lastFrameChunk & (blockSize == remaining);
assert(blockSize <= remaining);
/* TODO: See 3090. We reduced MIN_CBLOCK_SIZE from 3 to 2 so to compensate we are adding
* additional 1. We need to revisit and change this logic to be more consistent */
RETURN_ERROR_IF(dstCapacity < ZSTD_blockHeaderSize + MIN_CBLOCK_SIZE + 1,
dstSize_tooSmall,
"not enough space to store compressed block");
ZSTD_overflowCorrectIfNeeded(
ms, &cctx->workspace, &cctx->appliedParams, ip, ip + blockSize);
ZSTD_checkDictValidity(&ms->window, ip + blockSize, maxDist, &ms->loadedDictEnd, &ms->dictMatchState);
ZSTD_window_enforceMaxDist(&ms->window, ip, maxDist, &ms->loadedDictEnd, &ms->dictMatchState);
/* Ensure hash/chain table insertion resumes no sooner than lowlimit */
if (ms->nextToUpdate < ms->window.lowLimit) ms->nextToUpdate = ms->window.lowLimit;
{ size_t cSize;
if (ZSTD_useTargetCBlockSize(&cctx->appliedParams)) {
cSize = ZSTD_compressBlock_targetCBlockSize(cctx, op, dstCapacity, ip, blockSize, lastBlock);
FORWARD_IF_ERROR(cSize, "ZSTD_compressBlock_targetCBlockSize failed");
assert(cSize > 0);
assert(cSize <= blockSize + ZSTD_blockHeaderSize);
} else if (ZSTD_blockSplitterEnabled(&cctx->appliedParams)) {
cSize = ZSTD_compressBlock_splitBlock(cctx, op, dstCapacity, ip, blockSize, lastBlock);
FORWARD_IF_ERROR(cSize, "ZSTD_compressBlock_splitBlock failed");
assert(cSize > 0 || cctx->seqCollector.collectSequences == 1);
} else {
cSize = ZSTD_compressBlock_internal(cctx,
op+ZSTD_blockHeaderSize, dstCapacity-ZSTD_blockHeaderSize,
ip, blockSize, 1 /* frame */);
FORWARD_IF_ERROR(cSize, "ZSTD_compressBlock_internal failed");
if (cSize == 0) { /* block is not compressible */
cSize = ZSTD_rust_noCompressBlock(op, dstCapacity, ip, blockSize, lastBlock);
FORWARD_IF_ERROR(cSize, "ZSTD_noCompressBlock failed");
} else {
ZSTD_rust_writeBlockHeader(op, cSize, blockSize, lastBlock);
cSize += ZSTD_blockHeaderSize;
}
} /* if (ZSTD_useTargetCBlockSize(&cctx->appliedParams))*/
/* @savings is employed to ensure that splitting doesn't worsen expansion of incompressible data.
* Without splitting, the maximum expansion is 3 bytes per full block.
* An adversarial input could attempt to fudge the split detector,
* and make it split incompressible data, resulting in more block headers.
* Note that, since ZSTD_COMPRESSBOUND() assumes a worst case scenario of 1KB per block,
* and the splitter never creates blocks that small (current lower limit is 8 KB),
* there is already no risk to expand beyond ZSTD_COMPRESSBOUND() limit.
* But if the goal is to not expand by more than 3-bytes per 128 KB full block,
* then yes, it becomes possible to make the block splitter oversplit incompressible data.
* Using @savings, we enforce an even more conservative condition,
* requiring the presence of enough savings (at least 3 bytes) to authorize splitting,
* otherwise only full blocks are used.
* But being conservative is fine,
* since splitting barely compressible blocks is not fruitful anyway */
savings += (S64)blockSize - (S64)cSize;
ip += blockSize;
assert(remaining >= blockSize);
remaining -= blockSize;
op += cSize;
assert(dstCapacity >= cSize);
dstCapacity -= cSize;
cctx->isFirstBlock = 0;
DEBUGLOG(5, "ZSTD_compress_frameChunk: adding a block of size %u",
(unsigned)cSize);
} }
if (lastFrameChunk && (op>ostart)) cctx->stage = ZSTDcs_ending;
return (size_t)(op-ostart);
ZSTD_rust_frameChunkState state;
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.prepareBlock = ZSTD_rust_frameChunk_prepareBlock;
state.compressTarget = ZSTD_rust_frameChunk_compressTarget;
state.compressSplit = ZSTD_rust_frameChunk_compressSplit;
state.compressInternal = ZSTD_rust_frameChunk_compressInternal;
state.updateChecksum = ZSTD_rust_frameChunk_updateChecksum;
return ZSTD_rust_compressFrameChunk(
&state, dst, dstCapacity, src, srcSize, lastFrameChunk);
}