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

feat(mt): move serial reset orchestration into Rust

Browse Source 
Move the MT serial-state reset ordering into Rust. C retains parameter adjustment, private LDM table allocation and clearing, dictionary loading, window/checksum state, and allocator behavior behind callbacks; Rust now owns the reset sequencing and allocation-failure stop point.

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 -B -C programs -j1 zstd
- ulimit -v 41943040; make -C tests -j1 test-zstream ZSTREAM_TESTTIME=-T1s
This commit is contained in:
ddidderr
2026-07-19 20:11:35 +02:00
parent 2557319583
commit e3fe5b5a24
2 changed files with 371 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstdmt_compress.c
+141 -44
View File
@@ -451,6 +451,19 @@ typedef void (*ZSTDMT_initSetBufferSizeFn)(void* opaque, size_t size);
typedef size_t (*ZSTDMT_initResizeRoundBufferFn)(void* opaque, size_t capacity);
typedef void (*ZSTDMT_initResetStreamFn)(void* opaque);
typedef size_t (*ZSTDMT_initSerialResetFn)(void* opaque, size_t targetSectionSize);
typedef void (*ZSTDMT_serialResetVoidFn)(void* opaque);
typedef void (*ZSTDMT_serialResetSetNbSeqFn)(void* opaque, size_t nbSeq);
typedef int (*ZSTDMT_serialResetResizeFn)(void* opaque);
int ZSTDMT_rust_serialStateReset(
int enableLdm, int checksumEnabled, size_t maxNbSeq, void* opaque,
ZSTDMT_serialResetVoidFn resetNextJob,
ZSTDMT_serialResetVoidFn resetChecksum,
ZSTDMT_serialResetSetNbSeqFn setNbSeq,
ZSTDMT_serialResetVoidFn resetWindow,
ZSTDMT_serialResetResizeFn resizeTables,
ZSTDMT_serialResetVoidFn zeroTables,
ZSTDMT_serialResetVoidFn loadDictionary,
ZSTDMT_serialResetVoidFn copyWindow);
size_t ZSTDMT_rust_initCStream(
const ZSTDMT_RustInitCStreamProjection* projection, void* opaque,
ZSTDMT_initResizeFn resize, ZSTDMT_initDrainFn drain,
@@ -824,6 +837,99 @@ typedef struct {
ZSTD_window_t ldmWindow; /* A thread-safe copy of ldmState.window */
} SerialState;
typedef struct {
SerialState* serialState;
ZSTDMT_seqPool* seqPool;
ZSTD_CCtx_params* params;
const void* dict;
size_t dictSize;
ZSTD_dictContentType_e dictContentType;
size_t hashSize;
size_t numBuckets;
unsigned bucketLog;
unsigned prevBucketLog;
ZSTD_customMem cMem;
} ZSTDMT_serialResetContext;
static void ZSTDMT_serialResetNextJob(void* opaque)
{
ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque;
context->serialState->nextJobID = 0;
}
static void ZSTDMT_serialResetChecksum(void* opaque)
{
ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque;
XXH64_reset(&context->serialState->xxhState, 0);
}
static void ZSTDMT_serialResetSetNbSeq(void* opaque, size_t nbSeq)
{
ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque;
ZSTDMT_setNbSeq(context->seqPool, nbSeq);
}
static void ZSTDMT_serialResetWindow(void* opaque)
{
ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque;
ZSTD_window_init(&context->serialState->ldmState.window);
}
static int ZSTDMT_serialResetResizeTables(void* opaque)
{
ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque;
SerialState* const serialState = context->serialState;
ldmParams_t const* const ldmParams = &context->params->ldmParams;
if (serialState->ldmState.hashTable == NULL ||
serialState->params.ldmParams.hashLog < ldmParams->hashLog) {
ZSTD_customFree(serialState->ldmState.hashTable, context->cMem);
serialState->ldmState.hashTable = (ldmEntry_t*)ZSTD_customMalloc(
context->hashSize, context->cMem);
}
if (serialState->ldmState.bucketOffsets == NULL ||
context->prevBucketLog < context->bucketLog) {
ZSTD_customFree(serialState->ldmState.bucketOffsets, context->cMem);
serialState->ldmState.bucketOffsets = (BYTE*)ZSTD_customMalloc(
context->numBuckets, context->cMem);
}
return !serialState->ldmState.hashTable || !serialState->ldmState.bucketOffsets;
}
static void ZSTDMT_serialResetZeroTables(void* opaque)
{
ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque;
SerialState* const serialState = context->serialState;
ZSTD_memset(serialState->ldmState.hashTable, 0, context->hashSize);
ZSTD_memset(serialState->ldmState.bucketOffsets, 0, context->numBuckets);
}
static void ZSTDMT_serialResetLoadDictionary(void* opaque)
{
ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque;
SerialState* const serialState = context->serialState;
serialState->ldmState.loadedDictEnd = 0;
if (context->dictSize > 0 &&
context->dictContentType == ZSTD_dct_rawContent) {
BYTE const* const dictEnd = (const BYTE*)context->dict + context->dictSize;
ZSTD_window_update(&serialState->ldmState.window,
context->dict, context->dictSize,
/* forceNonContiguous */ 0);
ZSTD_ldm_fillHashTable(&serialState->ldmState,
(const BYTE*)context->dict, dictEnd,
&context->params->ldmParams);
serialState->ldmState.loadedDictEnd = context->params->forceWindow
? 0 : (U32)(dictEnd - serialState->ldmState.window.base);
}
}
static void ZSTDMT_serialResetCopyWindow(void* opaque)
{
ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque;
context->serialState->ldmWindow = context->serialState->ldmState.window;
}
static int
ZSTDMT_serialState_reset(SerialState* serialState,
ZSTDMT_seqPool* seqPool,
@@ -832,6 +938,8 @@ ZSTDMT_serialState_reset(SerialState* serialState,
const void* dict, size_t const dictSize,
ZSTD_dictContentType_e dictContentType)
{
ZSTDMT_serialResetContext context;
/* Adjust parameters */
if (params.ldmParams.enableLdm == ZSTD_ps_enable) {
DEBUGLOG(4, "LDM window size = %u KB", (1U << params.cParams.windowLog) >> 10);
@@ -841,53 +949,42 @@ ZSTDMT_serialState_reset(SerialState* serialState,
} else {
ZSTD_memset(&params.ldmParams, 0, sizeof(params.ldmParams));
}
serialState->nextJobID = 0;
if (params.fParams.checksumFlag)
XXH64_reset(&serialState->xxhState, 0);
context.serialState = serialState;
context.seqPool = seqPool;
context.params = &params;
context.dict = dict;
context.dictSize = dictSize;
context.dictContentType = dictContentType;
context.cMem = params.customMem;
context.hashSize = 0;
context.numBuckets = 0;
context.bucketLog = 0;
context.prevBucketLog = 0;
if (params.ldmParams.enableLdm == ZSTD_ps_enable) {
ZSTD_customMem cMem = params.customMem;
unsigned const hashLog = params.ldmParams.hashLog;
size_t const hashSize = ((size_t)1 << hashLog) * sizeof(ldmEntry_t);
unsigned const bucketLog =
params.ldmParams.hashLog - params.ldmParams.bucketSizeLog;
unsigned const prevBucketLog =
serialState->params.ldmParams.hashLog -
serialState->params.ldmParams.bucketSizeLog;
size_t const numBuckets = (size_t)1 << bucketLog;
/* Size the seq pool tables */
ZSTDMT_setNbSeq(seqPool, ZSTD_ldm_getMaxNbSeq(params.ldmParams, jobSize));
/* Reset the window */
ZSTD_window_init(&serialState->ldmState.window);
/* Resize tables and output space if necessary. */
if (serialState->ldmState.hashTable == NULL || serialState->params.ldmParams.hashLog < hashLog) {
ZSTD_customFree(serialState->ldmState.hashTable, cMem);
serialState->ldmState.hashTable = (ldmEntry_t*)ZSTD_customMalloc(hashSize, cMem);
}
if (serialState->ldmState.bucketOffsets == NULL || prevBucketLog < bucketLog) {
ZSTD_customFree(serialState->ldmState.bucketOffsets, cMem);
serialState->ldmState.bucketOffsets = (BYTE*)ZSTD_customMalloc(numBuckets, cMem);
}
if (!serialState->ldmState.hashTable || !serialState->ldmState.bucketOffsets)
return 1;
/* Zero the tables */
ZSTD_memset(serialState->ldmState.hashTable, 0, hashSize);
ZSTD_memset(serialState->ldmState.bucketOffsets, 0, numBuckets);
context.hashSize = ((size_t)1 << hashLog) * sizeof(ldmEntry_t);
context.bucketLog = params.ldmParams.hashLog - params.ldmParams.bucketSizeLog;
context.prevBucketLog =
serialState->params.ldmParams.hashLog -
serialState->params.ldmParams.bucketSizeLog;
context.numBuckets = (size_t)1 << context.bucketLog;
}
/* Update window state and fill hash table with dict */
serialState->ldmState.loadedDictEnd = 0;
if (dictSize > 0) {
if (dictContentType == ZSTD_dct_rawContent) {
BYTE const* const dictEnd = (const BYTE*)dict + dictSize;
ZSTD_window_update(&serialState->ldmState.window, dict, dictSize, /* forceNonContiguous */ 0);
ZSTD_ldm_fillHashTable(&serialState->ldmState, (const BYTE*)dict, dictEnd, &params.ldmParams);
serialState->ldmState.loadedDictEnd = params.forceWindow ? 0 : (U32)(dictEnd - serialState->ldmState.window.base);
} else {
/* don't even load anything */
}
}
/* Initialize serialState's copy of ldmWindow. */
serialState->ldmWindow = serialState->ldmState.window;
if (ZSTDMT_rust_serialStateReset(
params.ldmParams.enableLdm == ZSTD_ps_enable,
params.fParams.checksumFlag,
ZSTD_ldm_getMaxNbSeq(params.ldmParams, jobSize),
&context,
ZSTDMT_serialResetNextJob,
ZSTDMT_serialResetChecksum,
ZSTDMT_serialResetSetNbSeq,
ZSTDMT_serialResetWindow,
ZSTDMT_serialResetResizeTables,
ZSTDMT_serialResetZeroTables,
ZSTDMT_serialResetLoadDictionary,
ZSTDMT_serialResetCopyWindow)) {
return 1;
}
serialState->params = params;
rust/src/zstdmt_compress.rs
+230
View File
@@ -199,6 +199,9 @@ pub type ZSTDMT_initSetBufferSizeFn = unsafe extern "C" fn(*mut c_void, usize);
pub type ZSTDMT_initResizeRoundBufferFn = unsafe extern "C" fn(*mut c_void, usize) -> usize;
pub type ZSTDMT_initResetStreamFn = unsafe extern "C" fn(*mut c_void);
pub type ZSTDMT_initSerialResetFn = unsafe extern "C" fn(*mut c_void, usize) -> usize;
pub type ZSTDMT_serialResetVoidFn = unsafe extern "C" fn(*mut c_void);
pub type ZSTDMT_serialResetSetNbSeqFn = unsafe extern "C" fn(*mut c_void, usize);
pub type ZSTDMT_serialResetResizeFn = unsafe extern "C" fn(*mut c_void) -> c_int;
#[repr(C)]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
@@ -1134,6 +1137,67 @@ where
serial_reset(target_section_size)
}
/// Run the MT serial-state reset policy while C retains its private LDM
/// tables, dictionary/window state, checksum state, and allocator callbacks.
/// The callback order mirrors the original reset path and stops immediately
/// when table allocation reports failure.
#[no_mangle]
pub unsafe extern "C" fn ZSTDMT_rust_serialStateReset(
enable_ldm: c_int,
checksum_enabled: c_int,
max_nb_seq: usize,
opaque: *mut c_void,
reset_next_job: Option<ZSTDMT_serialResetVoidFn>,
reset_checksum: Option<ZSTDMT_serialResetVoidFn>,
set_nb_seq: Option<ZSTDMT_serialResetSetNbSeqFn>,
reset_window: Option<ZSTDMT_serialResetVoidFn>,
resize_tables: Option<ZSTDMT_serialResetResizeFn>,
zero_tables: Option<ZSTDMT_serialResetVoidFn>,
load_dictionary: Option<ZSTDMT_serialResetVoidFn>,
copy_window: Option<ZSTDMT_serialResetVoidFn>,
) -> c_int {
let (
Some(reset_next_job),
Some(reset_checksum),
Some(set_nb_seq),
Some(reset_window),
Some(resize_tables),
Some(zero_tables),
Some(load_dictionary),
Some(copy_window),
) = (
reset_next_job,
reset_checksum,
set_nb_seq,
reset_window,
resize_tables,
zero_tables,
load_dictionary,
copy_window,
)
else {
return 1;
};
unsafe {
reset_next_job(opaque);
if checksum_enabled != 0 {
reset_checksum(opaque);
}
if enable_ldm == ZSTD_PS_ENABLE {
set_nb_seq(opaque, max_nb_seq);
reset_window(opaque);
if resize_tables(opaque) != 0 {
return 1;
}
zero_tables(opaque);
load_dictionary(opaque);
copy_window(opaque);
}
}
0
}
/// C ABI entry point for the MT streaming initializer. C owns every
/// allocation, dictionary handle, synchronization object, and private context
/// mutation; this wrapper only connects those operations to the Rust policy.
@@ -3734,6 +3798,172 @@ mod tests {
}
}
#[derive(Default)]
struct SerialResetTestContext {
events: Vec<&'static str>,
max_nb_seq: usize,
resize_result: c_int,
}
unsafe extern "C" fn serial_reset_test_next_job(context: *mut c_void) {
unsafe {
(*context.cast::<SerialResetTestContext>())
.events
.push("next-job");
}
}
unsafe extern "C" fn serial_reset_test_checksum(context: *mut c_void) {
unsafe {
(*context.cast::<SerialResetTestContext>())
.events
.push("checksum");
}
}
unsafe extern "C" fn serial_reset_test_set_nb_seq(context: *mut c_void, max_nb_seq: usize) {
unsafe {
let context = &mut *context.cast::<SerialResetTestContext>();
context.events.push("seq-size");
context.max_nb_seq = max_nb_seq;
}
}
unsafe extern "C" fn serial_reset_test_window(context: *mut c_void) {
unsafe {
(*context.cast::<SerialResetTestContext>())
.events
.push("window");
}
}
unsafe extern "C" fn serial_reset_test_resize(context: *mut c_void) -> c_int {
unsafe {
let context = &mut *context.cast::<SerialResetTestContext>();
context.events.push("resize");
context.resize_result
}
}
unsafe extern "C" fn serial_reset_test_zero(context: *mut c_void) {
unsafe {
(*context.cast::<SerialResetTestContext>())
.events
.push("zero");
}
}
unsafe extern "C" fn serial_reset_test_dictionary(context: *mut c_void) {
unsafe {
(*context.cast::<SerialResetTestContext>())
.events
.push("dictionary");
}
}
unsafe extern "C" fn serial_reset_test_copy_window(context: *mut c_void) {
unsafe {
(*context.cast::<SerialResetTestContext>())
.events
.push("copy-window");
}
}
type SerialResetTestCallbacks = (
Option<ZSTDMT_serialResetVoidFn>,
Option<ZSTDMT_serialResetVoidFn>,
Option<ZSTDMT_serialResetSetNbSeqFn>,
Option<ZSTDMT_serialResetVoidFn>,
Option<ZSTDMT_serialResetResizeFn>,
Option<ZSTDMT_serialResetVoidFn>,
Option<ZSTDMT_serialResetVoidFn>,
Option<ZSTDMT_serialResetVoidFn>,
);
fn serial_reset_test_callbacks() -> SerialResetTestCallbacks {
(
Some(serial_reset_test_next_job),
Some(serial_reset_test_checksum),
Some(serial_reset_test_set_nb_seq),
Some(serial_reset_test_window),
Some(serial_reset_test_resize),
Some(serial_reset_test_zero),
Some(serial_reset_test_dictionary),
Some(serial_reset_test_copy_window),
)
}
#[test]
fn serial_state_reset_runs_ldm_operations_in_c_order() {
let mut context = SerialResetTestContext::default();
let callbacks = serial_reset_test_callbacks();
let result = unsafe {
ZSTDMT_rust_serialStateReset(
ZSTD_PS_ENABLE,
1,
123,
(&mut context as *mut SerialResetTestContext).cast(),
callbacks.0,
callbacks.1,
callbacks.2,
callbacks.3,
callbacks.4,
callbacks.5,
callbacks.6,
callbacks.7,
)
};
assert_eq!(result, 0);
assert_eq!(context.max_nb_seq, 123);
assert_eq!(
context.events,
vec![
"next-job",
"checksum",
"seq-size",
"window",
"resize",
"zero",
"dictionary",
"copy-window",
]
);
}
#[test]
fn serial_state_reset_stops_before_zeroing_after_resize_failure() {
let mut context = SerialResetTestContext {
resize_result: 1,
..Default::default()
};
let callbacks = serial_reset_test_callbacks();
let result = unsafe {
ZSTDMT_rust_serialStateReset(
ZSTD_PS_ENABLE,
0,
123,
(&mut context as *mut SerialResetTestContext).cast(),
callbacks.0,
callbacks.1,
callbacks.2,
callbacks.3,
callbacks.4,
callbacks.5,
callbacks.6,
callbacks.7,
)
};
assert_eq!(result, 1);
assert_eq!(
context.events,
vec!["next-job", "seq-size", "window", "resize"]
);
}
#[test]
fn init_c_stream_preserves_success_order_and_normalization() {
let projection = init_projection();