feat(compress): route split blocks through Rust
Project frame-chunk split blocks into Rust alongside the existing internal and target paths. Rust now builds the sequence store, asks the C-owned block-split context to derive partitions, refreshes the per-block first-block state, and emits the split blocks through the Rust post-build path. Keep the C partition discovery callback and the existing callback fallback as compatibility seams. Preserve the no-compress path without requiring sequence-store cursors that it does not use. 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
This commit is contained in:
+101
-10
@@ -180,6 +180,7 @@ 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);
|
||||
@@ -587,6 +588,8 @@ pub struct ZSTD_rust_frameChunkState {
|
||||
compress_internal: Option<FrameChunkCompressFn>,
|
||||
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>,
|
||||
}
|
||||
|
||||
const _: () = {
|
||||
@@ -635,9 +638,17 @@ const _: () = {
|
||||
== 12 * size_of::<usize>() + size_of::<c_longlong>() + 6 * size_of::<c_int>()
|
||||
);
|
||||
assert!(
|
||||
size_of::<ZSTD_rust_frameChunkState>()
|
||||
offset_of!(ZSTD_rust_frameChunkState, compress_split_state)
|
||||
== 13 * size_of::<usize>() + size_of::<c_longlong>() + 6 * size_of::<c_int>()
|
||||
);
|
||||
assert!(
|
||||
offset_of!(ZSTD_rust_frameChunkState, derive_block_splits)
|
||||
== 14 * size_of::<usize>() + size_of::<c_longlong>() + 6 * size_of::<c_int>()
|
||||
);
|
||||
assert!(
|
||||
size_of::<ZSTD_rust_frameChunkState>()
|
||||
== 15 * size_of::<usize>() + size_of::<c_longlong>() + 6 * size_of::<c_int>()
|
||||
);
|
||||
};
|
||||
|
||||
/// Rust implementation of `ZSTD_compress_frameChunk`.
|
||||
@@ -772,15 +783,85 @@ unsafe fn compress_frame_chunk_body_with(
|
||||
}
|
||||
}
|
||||
} else if state.block_splitter_enabled != 0 {
|
||||
unsafe {
|
||||
(state.compress_split)(
|
||||
state.callback_context,
|
||||
op.cast(),
|
||||
remaining_capacity,
|
||||
ip.cast(),
|
||||
block_size,
|
||||
last_block,
|
||||
)
|
||||
if !state.compress_split_state.is_null() {
|
||||
let block_state = if state.compress_internal_state.is_null() {
|
||||
return ERROR(ZstdErrorCode::Generic);
|
||||
} else {
|
||||
unsafe { &*state.compress_internal_state }
|
||||
};
|
||||
let split_state = unsafe { &*state.compress_split_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 num_splits = if bss == ZSTD_BSS_COMPRESS as usize {
|
||||
if split_state.seq_store.is_null() || split_state.partitions.is_null() {
|
||||
return ERROR(ZstdErrorCode::Generic);
|
||||
}
|
||||
let sequence_store = unsafe { &*split_state.seq_store };
|
||||
if sequence_store.sequencesStart.is_null() || sequence_store.sequences.is_null()
|
||||
{
|
||||
return ERROR(ZstdErrorCode::Generic);
|
||||
}
|
||||
let sequence_count = unsafe {
|
||||
sequence_store
|
||||
.sequences
|
||||
.offset_from(sequence_store.sequencesStart)
|
||||
};
|
||||
if sequence_count < 0 {
|
||||
return ERROR(ZstdErrorCode::Generic);
|
||||
}
|
||||
let Some(derive_block_splits) = state.derive_block_splits else {
|
||||
return ERROR(ZstdErrorCode::Generic);
|
||||
};
|
||||
let result = unsafe {
|
||||
derive_block_splits(
|
||||
state.callback_context,
|
||||
split_state.partitions.cast_mut(),
|
||||
sequence_count as c_uint,
|
||||
)
|
||||
};
|
||||
if ERR_isError(result) {
|
||||
return result;
|
||||
}
|
||||
result
|
||||
} else {
|
||||
0
|
||||
};
|
||||
let mut split_state = unsafe { ptr::read(state.compress_split_state) };
|
||||
split_state.is_first_block = unsafe { *state.is_first_block };
|
||||
unsafe {
|
||||
ZSTD_rust_compressBlockSplitAfterBuild(
|
||||
&split_state,
|
||||
op.cast(),
|
||||
remaining_capacity,
|
||||
ip.cast(),
|
||||
block_size,
|
||||
last_block,
|
||||
num_splits,
|
||||
bss as c_int,
|
||||
)
|
||||
}
|
||||
} else {
|
||||
unsafe {
|
||||
(state.compress_split)(
|
||||
state.callback_context,
|
||||
op.cast(),
|
||||
remaining_capacity,
|
||||
ip.cast(),
|
||||
block_size,
|
||||
last_block,
|
||||
)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let compressed_size = if !state.compress_internal_state.is_null() {
|
||||
@@ -10453,6 +10534,8 @@ mod tests {
|
||||
compress_internal: Some(compress_end_test_frame_target),
|
||||
compress_internal_state: ptr::null(),
|
||||
compress_target_state: ptr::null(),
|
||||
compress_split_state: ptr::null(),
|
||||
derive_block_splits: None,
|
||||
});
|
||||
let frame_chunk_state = context
|
||||
.frame_chunk_state
|
||||
@@ -11330,6 +11413,8 @@ mod tests {
|
||||
compress_internal: Some(frame_chunk_test_internal),
|
||||
compress_internal_state: ptr::null(),
|
||||
compress_target_state: ptr::null(),
|
||||
compress_split_state: ptr::null(),
|
||||
derive_block_splits: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11671,6 +11756,8 @@ mod tests {
|
||||
compress_internal: Some(compress_continue_test_frame),
|
||||
compress_internal_state: ptr::null(),
|
||||
compress_target_state: ptr::null(),
|
||||
compress_split_state: ptr::null(),
|
||||
derive_block_splits: None,
|
||||
});
|
||||
let frame_chunk_state = context
|
||||
.frame_chunk_state
|
||||
@@ -12158,6 +12245,8 @@ mod tests {
|
||||
compress_internal: Some(compress_stream_test_block),
|
||||
compress_internal_state: ptr::null(),
|
||||
compress_target_state: ptr::null(),
|
||||
compress_split_state: ptr::null(),
|
||||
derive_block_splits: None,
|
||||
});
|
||||
let continue_frame_chunk_state = context
|
||||
.continue_frame_chunk_state
|
||||
@@ -12184,6 +12273,8 @@ mod tests {
|
||||
compress_internal: Some(compress_stream_test_end),
|
||||
compress_internal_state: ptr::null(),
|
||||
compress_target_state: ptr::null(),
|
||||
compress_split_state: ptr::null(),
|
||||
derive_block_splits: None,
|
||||
});
|
||||
let end_frame_chunk_state = context
|
||||
.end_frame_chunk_state
|
||||
|
||||
Reference in New Issue
Block a user