feat(mt): move serial lifecycle ordering into Rust
Move serial-state synchronization initialization and destruction order into Rust, including the original bitwise initialization-error aggregation. C retains pthread operations, state zeroing, and custom-memory table cleanup behind callbacks. 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:
@@ -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
|
||||
|
||||
@@ -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<ZSTDMT_serialStateVoidFn>,
|
||||
init_mutex: Option<ZSTDMT_serialStateInitFn>,
|
||||
init_cond: Option<ZSTDMT_serialStateInitFn>,
|
||||
init_ldm_mutex: Option<ZSTDMT_serialStateInitFn>,
|
||||
init_ldm_cond: Option<ZSTDMT_serialStateInitFn>,
|
||||
) -> 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<ZSTDMT_serialStateVoidFn>,
|
||||
destroy_cond: Option<ZSTDMT_serialStateVoidFn>,
|
||||
destroy_ldm_mutex: Option<ZSTDMT_serialStateVoidFn>,
|
||||
destroy_ldm_cond: Option<ZSTDMT_serialStateVoidFn>,
|
||||
free_tables: Option<ZSTDMT_serialStateVoidFn>,
|
||||
) {
|
||||
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::<SerialLifecycleTestContext>() };
|
||||
context.events.push(event);
|
||||
context.init_errors[index]
|
||||
}
|
||||
|
||||
unsafe extern "C" fn serial_lifecycle_zero(context: *mut c_void) {
|
||||
unsafe {
|
||||
(*context.cast::<SerialLifecycleTestContext>())
|
||||
.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::<SerialLifecycleTestContext>())
|
||||
.events
|
||||
.push("destroy-mutex");
|
||||
}
|
||||
}
|
||||
|
||||
unsafe extern "C" fn serial_lifecycle_destroy_cond(context: *mut c_void) {
|
||||
unsafe {
|
||||
(*context.cast::<SerialLifecycleTestContext>())
|
||||
.events
|
||||
.push("destroy-cond");
|
||||
}
|
||||
}
|
||||
|
||||
unsafe extern "C" fn serial_lifecycle_destroy_ldm_mutex(context: *mut c_void) {
|
||||
unsafe {
|
||||
(*context.cast::<SerialLifecycleTestContext>())
|
||||
.events
|
||||
.push("destroy-ldm-mutex");
|
||||
}
|
||||
}
|
||||
|
||||
unsafe extern "C" fn serial_lifecycle_destroy_ldm_cond(context: *mut c_void) {
|
||||
unsafe {
|
||||
(*context.cast::<SerialLifecycleTestContext>())
|
||||
.events
|
||||
.push("destroy-ldm-cond");
|
||||
}
|
||||
}
|
||||
|
||||
unsafe extern "C" fn serial_lifecycle_free_tables(context: *mut c_void) {
|
||||
unsafe {
|
||||
(*context.cast::<SerialLifecycleTestContext>())
|
||||
.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();
|
||||
|
||||
Reference in New Issue
Block a user