From 73843019559a3220a426db6d2f40fe9b2ee479c4 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Mon, 20 Jul 2026 01:47:02 +0200 Subject: [PATCH] feat(compress): route target blocks through Rust Project the target-sized block state into frame-chunk compression and invoke Rust sequence-store construction followed by the existing target-block post-build leaf. Refresh the per-block isFirstBlock snapshot to preserve the former C adapter's behavior while leaving the split-block branch on its existing callback path for a later seam. 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 'zstd_compress::tests::frame_chunk' -- --nocapture - 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 --- lib/compress/zstd_compress.c | 25 ++++++++++++-- rust/src/zstd_compress.rs | 64 ++++++++++++++++++++++++++++++------ 2 files changed, 76 insertions(+), 13 deletions(-) diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 14536ee2a..6cf2ec4e4 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -817,6 +817,8 @@ typedef char ZSTD_rust_frame_chunk_prepare_state_layout[ ? 1 : -1]; typedef struct ZSTD_rust_compressContinueBlockState_s ZSTD_rust_compressContinueBlockState; +typedef struct ZSTD_rust_targetCBlockSizeState_s + ZSTD_rust_targetCBlockSizeState; typedef size_t (*ZSTD_rust_frameChunkCompress_f)(void* context, void* dst, size_t dstCapacity, @@ -843,6 +845,7 @@ typedef struct { ZSTD_rust_frameChunkCompress_f compressSplit; ZSTD_rust_frameChunkCompress_f compressInternal; const ZSTD_rust_compressContinueBlockState* compressInternalState; + const ZSTD_rust_targetCBlockSizeState* compressTargetState; } ZSTD_rust_frameChunkState; size_t ZSTD_rust_compressFrameChunk( const ZSTD_rust_frameChunkState* state, @@ -874,8 +877,10 @@ typedef char ZSTD_rust_frame_chunk_state_layout[ == 7 * sizeof(void*) + sizeof(S64) + 6 * sizeof(int) && offsetof(ZSTD_rust_frameChunkState, compressInternalState) == 11 * sizeof(void*) + sizeof(S64) + 6 * sizeof(int) + && offsetof(ZSTD_rust_frameChunkState, compressTargetState) + == 12 * sizeof(void*) + sizeof(S64) + 6 * sizeof(int) && sizeof(ZSTD_rust_frameChunkState) - == 12 * sizeof(void*) + sizeof(S64) + 6 * sizeof(int)) + == 13 * sizeof(void*) + sizeof(S64) + 6 * sizeof(int)) ? 1 : -1]; /* The high-level continue/block entry points are Rust-owned. This projection @@ -1244,7 +1249,7 @@ size_t ZSTD_rust_compressStreamInit( /* 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. */ -typedef struct { +struct ZSTD_rust_targetCBlockSizeState_s { const SeqStore_t* seqStore; ZSTD_compressedBlockState_t** prevCBlock; ZSTD_compressedBlockState_t** nextCBlock; @@ -1256,7 +1261,7 @@ typedef struct { U32 windowLog; size_t targetCBlockSize; int isFirstBlock; -} ZSTD_rust_targetCBlockSizeState; +}; size_t ZSTD_rust_compressBlockTargetCBlockSize( const ZSTD_rust_targetCBlockSizeState* state, void* dst, size_t dstCapacity, @@ -5426,6 +5431,7 @@ typedef struct { ZSTD_rust_buildSeqStoreState buildSeqStoreState; ZSTD_rust_blockInternalState blockInternalState; ZSTD_rust_compressContinueBlockState blockState; + ZSTD_rust_targetCBlockSizeState targetBlockState; } ZSTD_rust_compressContinueContext; static int ZSTD_rust_overflowCorrect_need( @@ -5604,6 +5610,7 @@ static void ZSTD_compressContinue_prepare( context->frameChunkState.compressTarget = ZSTD_rust_frameChunk_compressTarget; context->frameChunkState.compressSplit = ZSTD_rust_frameChunk_compressSplit; context->frameChunkState.compressInternal = NULL; + context->frameChunkState.compressTargetState = &context->targetBlockState; ZSTD_initBuildSeqStoreState(cctx, &context->buildSeqStoreState); context->blockInternalState.seqStore = &cctx->seqStore; @@ -5620,6 +5627,18 @@ static void ZSTD_compressContinue_prepare( context->blockState.buildSeqStoreState = &context->buildSeqStoreState; context->blockState.blockInternalState = &context->blockInternalState; context->frameChunkState.compressInternalState = &context->blockState; + context->targetBlockState.seqStore = &cctx->seqStore; + context->targetBlockState.prevCBlock = &cctx->blockState.prevCBlock; + context->targetBlockState.nextCBlock = &cctx->blockState.nextCBlock; + context->targetBlockState.tmpWorkspace = cctx->tmpWorkspace; + context->targetBlockState.tmpWkspSize = cctx->tmpWkspSize; + context->targetBlockState.strategy = (int)cctx->appliedParams.cParams.strategy; + context->targetBlockState.disableLiteralCompression = + ZSTD_literalsCompressionIsDisabled(&cctx->appliedParams); + context->targetBlockState.bmi2 = cctx->bmi2; + context->targetBlockState.windowLog = cctx->appliedParams.cParams.windowLog; + context->targetBlockState.targetCBlockSize = cctx->appliedParams.targetCBlockSize; + context->targetBlockState.isFirstBlock = cctx->isFirstBlock; context->overflowContext.matchState = ms; context->overflowContext.workspace = &cctx->workspace; diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index b323d6fbf..2e2a23616 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -586,6 +586,7 @@ pub struct ZSTD_rust_frameChunkState { compress_split: FrameChunkCompressFn, compress_internal: Option, compress_internal_state: *const ZSTD_rust_compressContinueBlockState, + compress_target_state: *const ZSTD_rust_targetCBlockSizeState, } const _: () = { @@ -630,9 +631,13 @@ const _: () = { == 11 * size_of::() + size_of::() + 6 * size_of::() ); assert!( - size_of::() + offset_of!(ZSTD_rust_frameChunkState, compress_target_state) == 12 * size_of::() + size_of::() + 6 * size_of::() ); + assert!( + size_of::() + == 13 * size_of::() + size_of::() + 6 * size_of::() + ); }; /// Rust implementation of `ZSTD_compress_frameChunk`. @@ -722,15 +727,49 @@ unsafe fn compress_frame_chunk_body_with( unsafe { prepare_frame_chunk_block(&*state.prepare_state, ip.cast(), block_size) }; let c_size = if state.use_target_c_block_size != 0 { - unsafe { - (state.compress_target)( - state.callback_context, - op.cast(), - remaining_capacity, - ip.cast(), - block_size, - last_block, - ) + if !state.compress_target_state.is_null() { + let block_state = if state.compress_internal_state.is_null() { + return ERROR(ZstdErrorCode::Generic); + } else { + unsafe { &*state.compress_internal_state } + }; + if block_state.build_seq_store_state.is_null() { + return ERROR(ZstdErrorCode::Generic); + } + let bss = unsafe { + ZSTD_rust_buildSeqStore( + block_state.build_seq_store_state, + ip.cast(), + block_size, + ) + }; + if ERR_isError(bss) { + return bss; + } + let mut target_state = unsafe { ptr::read(state.compress_target_state) }; + target_state.is_first_block = unsafe { *state.is_first_block }; + unsafe { + ZSTD_rust_compressBlockTargetCBlockSizeAfterBuild( + &target_state, + op.cast(), + remaining_capacity, + ip.cast(), + block_size, + bss as c_int, + last_block, + ) + } + } else { + unsafe { + (state.compress_target)( + state.callback_context, + op.cast(), + remaining_capacity, + ip.cast(), + block_size, + last_block, + ) + } } } else if state.block_splitter_enabled != 0 { unsafe { @@ -10413,6 +10452,7 @@ mod tests { compress_split: compress_end_test_frame_target, compress_internal: Some(compress_end_test_frame_target), compress_internal_state: ptr::null(), + compress_target_state: ptr::null(), }); let frame_chunk_state = context .frame_chunk_state @@ -11289,6 +11329,7 @@ mod tests { compress_split: frame_chunk_test_split, compress_internal: Some(frame_chunk_test_internal), compress_internal_state: ptr::null(), + compress_target_state: ptr::null(), } } @@ -11629,6 +11670,7 @@ mod tests { compress_split: compress_continue_test_frame, compress_internal: Some(compress_continue_test_frame), compress_internal_state: ptr::null(), + compress_target_state: ptr::null(), }); let frame_chunk_state = context .frame_chunk_state @@ -12115,6 +12157,7 @@ mod tests { compress_split: compress_stream_test_block, compress_internal: Some(compress_stream_test_block), compress_internal_state: ptr::null(), + compress_target_state: ptr::null(), }); let continue_frame_chunk_state = context .continue_frame_chunk_state @@ -12140,6 +12183,7 @@ mod tests { compress_split: compress_stream_test_end, compress_internal: Some(compress_stream_test_end), compress_internal_state: ptr::null(), + compress_target_state: ptr::null(), }); let end_frame_chunk_state = context .end_frame_chunk_state