From e3fe5b5a2489757d6c1241830b523691a3a0f49e Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sun, 19 Jul 2026 20:11:35 +0200 Subject: [PATCH] feat(mt): move serial reset orchestration into Rust 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 --- lib/compress/zstdmt_compress.c | 185 +++++++++++++++++++------- rust/src/zstdmt_compress.rs | 230 +++++++++++++++++++++++++++++++++ 2 files changed, 371 insertions(+), 44 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index d34121287..6aa5db922 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -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(¶ms.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 = ¶ms; + 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, ¶ms.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; diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 9583c091f..5ea6ce550 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -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, + reset_checksum: Option, + set_nb_seq: Option, + reset_window: Option, + resize_tables: Option, + zero_tables: Option, + load_dictionary: Option, + copy_window: Option, +) -> 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::()) + .events + .push("next-job"); + } + } + + unsafe extern "C" fn serial_reset_test_checksum(context: *mut c_void) { + unsafe { + (*context.cast::()) + .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::(); + 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::()) + .events + .push("window"); + } + } + + unsafe extern "C" fn serial_reset_test_resize(context: *mut c_void) -> c_int { + unsafe { + let context = &mut *context.cast::(); + context.events.push("resize"); + context.resize_result + } + } + + unsafe extern "C" fn serial_reset_test_zero(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("zero"); + } + } + + unsafe extern "C" fn serial_reset_test_dictionary(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("dictionary"); + } + } + + unsafe extern "C" fn serial_reset_test_copy_window(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("copy-window"); + } + } + + type SerialResetTestCallbacks = ( + Option, + Option, + Option, + Option, + Option, + Option, + Option, + Option, + ); + + 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();