diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 6aa5db922..cc18a0fc1 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -464,6 +464,20 @@ int ZSTDMT_rust_serialStateReset( ZSTDMT_serialResetVoidFn zeroTables, ZSTDMT_serialResetVoidFn loadDictionary, ZSTDMT_serialResetVoidFn copyWindow); +typedef void (*ZSTDMT_serialStateVoidFn)(void* opaque); +typedef int (*ZSTDMT_serialStateInitFn)(void* opaque); +int ZSTDMT_rust_serialStateInit( + void* opaque, ZSTDMT_serialStateVoidFn zeroState, + ZSTDMT_serialStateInitFn initMutex, + ZSTDMT_serialStateInitFn initCond, + ZSTDMT_serialStateInitFn initLdmMutex, + ZSTDMT_serialStateInitFn initLdmCond); +void ZSTDMT_rust_serialStateFree( + void* opaque, ZSTDMT_serialStateVoidFn destroyMutex, + ZSTDMT_serialStateVoidFn destroyCond, + ZSTDMT_serialStateVoidFn destroyLdmMutex, + ZSTDMT_serialStateVoidFn destroyLdmCond, + ZSTDMT_serialStateVoidFn freeTables); size_t ZSTDMT_rust_initCStream( const ZSTDMT_RustInitCStreamProjection* projection, void* opaque, ZSTDMT_initResizeFn resize, ZSTDMT_initDrainFn drain, @@ -992,26 +1006,87 @@ ZSTDMT_serialState_reset(SerialState* serialState, return 0; } +static void ZSTDMT_serialState_zero(void* opaque) +{ + ZSTD_memset(opaque, 0, sizeof(SerialState)); +} + +static int ZSTDMT_serialState_initMutex(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + return ZSTD_pthread_mutex_init(&serialState->mutex, NULL); +} + +static int ZSTDMT_serialState_initCond(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + return ZSTD_pthread_cond_init(&serialState->cond, NULL); +} + +static int ZSTDMT_serialState_initLdmMutex(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + return ZSTD_pthread_mutex_init(&serialState->ldmWindowMutex, NULL); +} + +static int ZSTDMT_serialState_initLdmCond(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + return ZSTD_pthread_cond_init(&serialState->ldmWindowCond, NULL); +} + +static void ZSTDMT_serialState_destroyMutex(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_pthread_mutex_destroy(&serialState->mutex); +} + +static void ZSTDMT_serialState_destroyCond(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_pthread_cond_destroy(&serialState->cond); +} + +static void ZSTDMT_serialState_destroyLdmMutex(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_pthread_mutex_destroy(&serialState->ldmWindowMutex); +} + +static void ZSTDMT_serialState_destroyLdmCond(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_pthread_cond_destroy(&serialState->ldmWindowCond); +} + +static void ZSTDMT_serialState_freeTables(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; + ZSTD_customMem const cMem = serialState->params.customMem; + ZSTD_customFree(serialState->ldmState.hashTable, cMem); + ZSTD_customFree(serialState->ldmState.bucketOffsets, cMem); +} + static int ZSTDMT_serialState_init(SerialState* serialState) { - int initError = 0; - ZSTD_memset(serialState, 0, sizeof(*serialState)); - initError |= ZSTD_pthread_mutex_init(&serialState->mutex, NULL); - initError |= ZSTD_pthread_cond_init(&serialState->cond, NULL); - initError |= ZSTD_pthread_mutex_init(&serialState->ldmWindowMutex, NULL); - initError |= ZSTD_pthread_cond_init(&serialState->ldmWindowCond, NULL); - return initError; + return ZSTDMT_rust_serialStateInit( + serialState, + ZSTDMT_serialState_zero, + ZSTDMT_serialState_initMutex, + ZSTDMT_serialState_initCond, + ZSTDMT_serialState_initLdmMutex, + ZSTDMT_serialState_initLdmCond); } static void ZSTDMT_serialState_free(SerialState* serialState) { - ZSTD_customMem cMem = serialState->params.customMem; - ZSTD_pthread_mutex_destroy(&serialState->mutex); - ZSTD_pthread_cond_destroy(&serialState->cond); - ZSTD_pthread_mutex_destroy(&serialState->ldmWindowMutex); - ZSTD_pthread_cond_destroy(&serialState->ldmWindowCond); - ZSTD_customFree(serialState->ldmState.hashTable, cMem); - ZSTD_customFree(serialState->ldmState.bucketOffsets, cMem); + ZSTDMT_rust_serialStateFree( + serialState, + ZSTDMT_serialState_destroyMutex, + ZSTDMT_serialState_destroyCond, + ZSTDMT_serialState_destroyLdmMutex, + ZSTDMT_serialState_destroyLdmCond, + ZSTDMT_serialState_freeTables); } /* Rust owns the serial turn/skip decision and operation ordering. The wait diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 5ea6ce550..fa586cb77 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -202,6 +202,8 @@ pub type ZSTDMT_initSerialResetFn = unsafe extern "C" fn(*mut c_void, 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; +pub type ZSTDMT_serialStateVoidFn = unsafe extern "C" fn(*mut c_void); +pub type ZSTDMT_serialStateInitFn = unsafe extern "C" fn(*mut c_void) -> c_int; #[repr(C)] #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] @@ -1198,6 +1200,79 @@ pub unsafe extern "C" fn ZSTDMT_rust_serialStateReset( 0 } +/// Run the MT serial-state synchronization initialization order and preserve +/// the original bitwise error aggregation. C owns each pthread primitive and +/// the zeroing callback. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_serialStateInit( + opaque: *mut c_void, + zero_state: Option, + init_mutex: Option, + init_cond: Option, + init_ldm_mutex: Option, + init_ldm_cond: Option, +) -> c_int { + let ( + Some(zero_state), + Some(init_mutex), + Some(init_cond), + Some(init_ldm_mutex), + Some(init_ldm_cond), + ) = ( + zero_state, + init_mutex, + init_cond, + init_ldm_mutex, + init_ldm_cond, + ) + else { + return 1; + }; + + unsafe { + zero_state(opaque); + init_mutex(opaque) | init_cond(opaque) | init_ldm_mutex(opaque) | init_ldm_cond(opaque) + } +} + +/// Run the MT serial-state synchronization destruction and table-free order. +/// The callbacks retain ownership of platform synchronization and allocator +/// details on the C side. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_serialStateFree( + opaque: *mut c_void, + destroy_mutex: Option, + destroy_cond: Option, + destroy_ldm_mutex: Option, + destroy_ldm_cond: Option, + free_tables: Option, +) { + let ( + Some(destroy_mutex), + Some(destroy_cond), + Some(destroy_ldm_mutex), + Some(destroy_ldm_cond), + Some(free_tables), + ) = ( + destroy_mutex, + destroy_cond, + destroy_ldm_mutex, + destroy_ldm_cond, + free_tables, + ) + else { + return; + }; + + unsafe { + destroy_mutex(opaque); + destroy_cond(opaque); + destroy_ldm_mutex(opaque); + destroy_ldm_cond(opaque); + free_tables(opaque); + } +} + /// 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. @@ -3964,6 +4039,134 @@ mod tests { ); } + #[derive(Default)] + struct SerialLifecycleTestContext { + events: Vec<&'static str>, + init_errors: [c_int; 4], + } + + unsafe fn serial_lifecycle_init_step( + context: *mut c_void, + index: usize, + event: &'static str, + ) -> c_int { + let context = unsafe { &mut *context.cast::() }; + context.events.push(event); + context.init_errors[index] + } + + unsafe extern "C" fn serial_lifecycle_zero(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("zero"); + } + } + + unsafe extern "C" fn serial_lifecycle_init_mutex(context: *mut c_void) -> c_int { + unsafe { serial_lifecycle_init_step(context, 0, "mutex") } + } + + unsafe extern "C" fn serial_lifecycle_init_cond(context: *mut c_void) -> c_int { + unsafe { serial_lifecycle_init_step(context, 1, "cond") } + } + + unsafe extern "C" fn serial_lifecycle_init_ldm_mutex(context: *mut c_void) -> c_int { + unsafe { serial_lifecycle_init_step(context, 2, "ldm-mutex") } + } + + unsafe extern "C" fn serial_lifecycle_init_ldm_cond(context: *mut c_void) -> c_int { + unsafe { serial_lifecycle_init_step(context, 3, "ldm-cond") } + } + + unsafe extern "C" fn serial_lifecycle_destroy_mutex(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("destroy-mutex"); + } + } + + unsafe extern "C" fn serial_lifecycle_destroy_cond(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("destroy-cond"); + } + } + + unsafe extern "C" fn serial_lifecycle_destroy_ldm_mutex(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("destroy-ldm-mutex"); + } + } + + unsafe extern "C" fn serial_lifecycle_destroy_ldm_cond(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("destroy-ldm-cond"); + } + } + + unsafe extern "C" fn serial_lifecycle_free_tables(context: *mut c_void) { + unsafe { + (*context.cast::()) + .events + .push("free-tables"); + } + } + + #[test] + fn serial_state_lifecycle_preserves_order_and_aggregates_init_errors() { + let mut context = SerialLifecycleTestContext { + init_errors: [0, 2, 0, 4], + ..Default::default() + }; + let opaque = (&mut context as *mut SerialLifecycleTestContext).cast(); + + let result = unsafe { + ZSTDMT_rust_serialStateInit( + opaque, + Some(serial_lifecycle_zero), + Some(serial_lifecycle_init_mutex), + Some(serial_lifecycle_init_cond), + Some(serial_lifecycle_init_ldm_mutex), + Some(serial_lifecycle_init_ldm_cond), + ) + }; + assert_eq!(result, 6); + + unsafe { + ZSTDMT_rust_serialStateFree( + opaque, + Some(serial_lifecycle_destroy_mutex), + Some(serial_lifecycle_destroy_cond), + Some(serial_lifecycle_destroy_ldm_mutex), + Some(serial_lifecycle_destroy_ldm_cond), + Some(serial_lifecycle_free_tables), + ) + }; + + assert_eq!( + context.events, + vec![ + "zero", + "mutex", + "cond", + "ldm-mutex", + "ldm-cond", + "destroy-mutex", + "destroy-cond", + "destroy-ldm-mutex", + "destroy-ldm-cond", + "free-tables", + ] + ); + } + #[test] fn init_c_stream_preserves_success_order_and_normalization() { let projection = init_projection();