diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index f3c69b24f..f1f936325 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -593,6 +593,39 @@ void ZSTDMT_rust_serialStateFree( ZSTDMT_serialStateVoidFn destroyLdmMutex, ZSTDMT_serialStateVoidFn destroyLdmCond, ZSTDMT_serialStateVoidFn freeTables); +typedef void (*ZSTDMT_freeCCtxCallbackFn)(void* opaque); +typedef struct { + void* callbackContext; + size_t providedFactory; + size_t roundBuffPresent; + ZSTDMT_freeCCtxCallbackFn freeFactory; + ZSTDMT_freeCCtxCallbackFn releaseAllJobResources; + ZSTDMT_freeCCtxCallbackFn freeJobs; + ZSTDMT_freeCCtxCallbackFn freeBufferPool; + ZSTDMT_freeCCtxCallbackFn freeCCtxPool; + ZSTDMT_freeCCtxCallbackFn freeSeqPool; + ZSTDMT_freeCCtxCallbackFn freeSerialState; + ZSTDMT_freeCCtxCallbackFn freeCDict; + ZSTDMT_freeCCtxCallbackFn freeRoundBuffer; + ZSTDMT_freeCCtxCallbackFn freeMTctx; +} ZSTDMT_RustFreeCCtxState; +typedef char ZSTDMT_rust_free_cctx_state_layout[ + (sizeof(ZSTDMT_freeCCtxCallbackFn) == sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, callbackContext) == 0 + && offsetof(ZSTDMT_RustFreeCCtxState, providedFactory) == sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, roundBuffPresent) == 2 * sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, freeFactory) == 3 * sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, releaseAllJobResources) == 4 * sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, freeJobs) == 5 * sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, freeBufferPool) == 6 * sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, freeCCtxPool) == 7 * sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, freeSeqPool) == 8 * sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, freeSerialState) == 9 * sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, freeCDict) == 10 * sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, freeRoundBuffer) == 11 * sizeof(void*) + && offsetof(ZSTDMT_RustFreeCCtxState, freeMTctx) == 12 * sizeof(void*) + && sizeof(ZSTDMT_RustFreeCCtxState) == 13 * sizeof(void*)) ? 1 : -1]; +size_t ZSTDMT_rust_freeCCtx(const ZSTDMT_RustFreeCCtxState* state); size_t ZSTDMT_rust_initCStream( const ZSTDMT_RustInitCStreamProjection* projection, void* opaque, ZSTDMT_initResizeFn resize, ZSTDMT_initDrainFn drain, @@ -1885,22 +1918,83 @@ static void ZSTDMT_waitForAllJobsCompleted(ZSTDMT_CCtx* mtctx) mtctx, ZSTDMT_waitForJobComplete); } +static void ZSTDMT_rust_freeCCtxFactory(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + POOL_free(mtctx->factory); /* stop and free worker threads */ +} + +static void ZSTDMT_rust_freeCCtxReleaseJobResources(void* opaque) +{ + ZSTDMT_releaseAllJobResources((ZSTDMT_CCtx*)opaque); +} + +static void ZSTDMT_rust_freeCCtxJobs(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + ZSTDMT_freeJobsTable(mtctx->jobs, mtctx->jobIDMask+1, mtctx->cMem); +} + +static void ZSTDMT_rust_freeCCtxBufferPool(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + ZSTDMT_freeBufferPool(mtctx->bufPool); +} + +static void ZSTDMT_rust_freeCCtxPool(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + ZSTDMT_freeCCtxPool(mtctx->cctxPool); +} + +static void ZSTDMT_rust_freeCCtxSeqPool(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + ZSTDMT_freeSeqPool(mtctx->seqPool); +} + +static void ZSTDMT_rust_freeCCtxSerialState(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + ZSTDMT_serialState_free(&mtctx->serial); +} + +static void ZSTDMT_rust_freeCCtxCDict(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + ZSTD_freeCDict(mtctx->cdictLocal); +} + +static void ZSTDMT_rust_freeCCtxRoundBuffer(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + ZSTD_customFree(mtctx->roundBuff.buffer, mtctx->cMem); +} + +static void ZSTDMT_rust_freeCCtxContext(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + ZSTD_customFree(mtctx, mtctx->cMem); +} + size_t ZSTDMT_freeCCtx(ZSTDMT_CCtx* mtctx) { - if (mtctx==NULL) return 0; /* compatible with free on NULL */ - if (!mtctx->providedFactory) - POOL_free(mtctx->factory); /* stop and free worker threads */ - ZSTDMT_releaseAllJobResources(mtctx); /* release job resources into pools first */ - ZSTDMT_freeJobsTable(mtctx->jobs, mtctx->jobIDMask+1, mtctx->cMem); - ZSTDMT_freeBufferPool(mtctx->bufPool); - ZSTDMT_freeCCtxPool(mtctx->cctxPool); - ZSTDMT_freeSeqPool(mtctx->seqPool); - ZSTDMT_serialState_free(&mtctx->serial); - ZSTD_freeCDict(mtctx->cdictLocal); - if (mtctx->roundBuff.buffer) - ZSTD_customFree(mtctx->roundBuff.buffer, mtctx->cMem); - ZSTD_customFree(mtctx, mtctx->cMem); - return 0; + ZSTDMT_RustFreeCCtxState const state = { + mtctx, + mtctx != NULL && mtctx->providedFactory, + mtctx != NULL && mtctx->roundBuff.buffer != NULL, + ZSTDMT_rust_freeCCtxFactory, + ZSTDMT_rust_freeCCtxReleaseJobResources, + ZSTDMT_rust_freeCCtxJobs, + ZSTDMT_rust_freeCCtxBufferPool, + ZSTDMT_rust_freeCCtxPool, + ZSTDMT_rust_freeCCtxSeqPool, + ZSTDMT_rust_freeCCtxSerialState, + ZSTDMT_rust_freeCCtxCDict, + ZSTDMT_rust_freeCCtxRoundBuffer, + ZSTDMT_rust_freeCCtxContext + }; + return ZSTDMT_rust_freeCCtx(mtctx == NULL ? NULL : &state); } size_t ZSTDMT_sizeof_CCtx(ZSTDMT_CCtx* mtctx) diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 26cce4a0c..d8d726de5 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -376,6 +376,48 @@ pub struct ZSTDMT_serialStateEnsureFinishedResult { pub type ZSTDMT_waitForJobCompleteFn = unsafe extern "C" fn(*mut c_void, c_uint, c_uint); pub type ZSTDMT_releaseJobResourceFn = unsafe extern "C" fn(*mut c_void, c_uint); +pub type ZSTDMT_freeCCtxFn = unsafe extern "C" fn(*mut c_void); + +/// Projection for MT context teardown. The context and every allocation or +/// synchronization object remain private to C; Rust owns only the ownership +/// branches and teardown ordering between callbacks. +#[repr(C)] +pub struct ZSTDMT_RustFreeCCtxState { + callback_context: *mut c_void, + provided_factory: usize, + round_buffer_present: usize, + free_factory: Option, + release_all_job_resources: Option, + free_jobs: Option, + free_buffer_pool: Option, + free_cctx_pool: Option, + free_seq_pool: Option, + free_serial_state: Option, + free_cdict: Option, + free_round_buffer: Option, + free_mtctx: Option, +} + +const _: () = { + assert!(size_of::() == size_of::()); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, callback_context) == 0); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, provided_factory) == size_of::()); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, round_buffer_present) == 2 * size_of::()); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_factory) == 3 * size_of::()); + assert!( + offset_of!(ZSTDMT_RustFreeCCtxState, release_all_job_resources) == 4 * size_of::() + ); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_jobs) == 5 * size_of::()); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_buffer_pool) == 6 * size_of::()); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_cctx_pool) == 7 * size_of::()); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_seq_pool) == 8 * (usize::BITS as usize / 8)); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_serial_state) == 9 * size_of::()); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_cdict) == 10 * size_of::()); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_round_buffer) == 11 * size_of::()); + assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_mtctx) == 12 * size_of::()); + assert!(size_of::() == size_of::<[usize; 13]>()); +}; + type ZSTDMT_waitForLdmLockFn = unsafe extern "C" fn(*mut c_void); type ZSTDMT_waitForLdmOverlapFn = unsafe extern "C" fn(*mut c_void, *mut c_void, usize) -> c_int; type ZSTDMT_waitForLdmWaitFn = unsafe extern "C" fn(*mut c_void); @@ -1597,6 +1639,60 @@ pub unsafe extern "C" fn ZSTDMT_rust_serialStateFree( } } +/// Run the MT context teardown order while C callbacks retain all private +/// layouts, allocators, worker pools, and synchronization details. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_freeCCtx(state: *const ZSTDMT_RustFreeCCtxState) -> usize { + if state.is_null() { + return 0; + } + let state = unsafe { &*state }; + let ( + Some(free_factory), + Some(release_all_job_resources), + Some(free_jobs), + Some(free_buffer_pool), + Some(free_cctx_pool), + Some(free_seq_pool), + Some(free_serial_state), + Some(free_cdict), + Some(free_round_buffer), + Some(free_mtctx), + ) = ( + state.free_factory, + state.release_all_job_resources, + state.free_jobs, + state.free_buffer_pool, + state.free_cctx_pool, + state.free_seq_pool, + state.free_serial_state, + state.free_cdict, + state.free_round_buffer, + state.free_mtctx, + ) + else { + return 0; + }; + + unsafe { + if state.provided_factory == 0 { + free_factory(state.callback_context); + } + release_all_job_resources(state.callback_context); + free_jobs(state.callback_context); + free_buffer_pool(state.callback_context); + free_cctx_pool(state.callback_context); + free_seq_pool(state.callback_context); + free_serial_state(state.callback_context); + free_cdict(state.callback_context); + if state.round_buffer_present != 0 { + free_round_buffer(state.callback_context); + } + free_mtctx(state.callback_context); + } + 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. @@ -3781,6 +3877,124 @@ mod tests { static JOB_TABLE_FAIL_INIT: AtomicBool = AtomicBool::new(false); static JOB_TABLE_TEST_LOCK: Mutex<()> = Mutex::new(()); + #[derive(Default)] + struct FreeCCtxTestContext { + events: Vec<&'static str>, + } + + fn record_free_cctx_event(context: *mut c_void, event: &'static str) { + unsafe { + (*context.cast::()).events.push(event); + } + } + + unsafe extern "C" fn free_cctx_test_factory(context: *mut c_void) { + record_free_cctx_event(context, "factory"); + } + + unsafe extern "C" fn free_cctx_test_release_jobs(context: *mut c_void) { + record_free_cctx_event(context, "release-jobs"); + } + + unsafe extern "C" fn free_cctx_test_jobs(context: *mut c_void) { + record_free_cctx_event(context, "jobs"); + } + + unsafe extern "C" fn free_cctx_test_buffer_pool(context: *mut c_void) { + record_free_cctx_event(context, "buffer-pool"); + } + + unsafe extern "C" fn free_cctx_test_cctx_pool(context: *mut c_void) { + record_free_cctx_event(context, "cctx-pool"); + } + + unsafe extern "C" fn free_cctx_test_seq_pool(context: *mut c_void) { + record_free_cctx_event(context, "seq-pool"); + } + + unsafe extern "C" fn free_cctx_test_serial_state(context: *mut c_void) { + record_free_cctx_event(context, "serial-state"); + } + + unsafe extern "C" fn free_cctx_test_cdict(context: *mut c_void) { + record_free_cctx_event(context, "cdict"); + } + + unsafe extern "C" fn free_cctx_test_round_buffer(context: *mut c_void) { + record_free_cctx_event(context, "round-buffer"); + } + + unsafe extern "C" fn free_cctx_test_mtctx(context: *mut c_void) { + record_free_cctx_event(context, "mtctx"); + } + + fn free_cctx_test_state( + context: &mut FreeCCtxTestContext, + provided_factory: usize, + round_buffer_present: usize, + ) -> ZSTDMT_RustFreeCCtxState { + ZSTDMT_RustFreeCCtxState { + callback_context: context as *mut FreeCCtxTestContext as *mut c_void, + provided_factory, + round_buffer_present, + free_factory: Some(free_cctx_test_factory), + release_all_job_resources: Some(free_cctx_test_release_jobs), + free_jobs: Some(free_cctx_test_jobs), + free_buffer_pool: Some(free_cctx_test_buffer_pool), + free_cctx_pool: Some(free_cctx_test_cctx_pool), + free_seq_pool: Some(free_cctx_test_seq_pool), + free_serial_state: Some(free_cctx_test_serial_state), + free_cdict: Some(free_cctx_test_cdict), + free_round_buffer: Some(free_cctx_test_round_buffer), + free_mtctx: Some(free_cctx_test_mtctx), + } + } + + #[test] + fn free_cctx_null_state_returns_zero_without_callbacks() { + assert_eq!(unsafe { ZSTDMT_rust_freeCCtx(ptr::null()) }, 0); + } + + #[test] + fn free_cctx_preserves_callback_order_and_ownership_conditions() { + let mut context = FreeCCtxTestContext::default(); + let state = free_cctx_test_state(&mut context, 0, 1); + + assert_eq!(unsafe { ZSTDMT_rust_freeCCtx(&state) }, 0); + assert_eq!( + context.events, + vec![ + "factory", + "release-jobs", + "jobs", + "buffer-pool", + "cctx-pool", + "seq-pool", + "serial-state", + "cdict", + "round-buffer", + "mtctx", + ] + ); + + context.events.clear(); + let state = free_cctx_test_state(&mut context, 1, 0); + assert_eq!(unsafe { ZSTDMT_rust_freeCCtx(&state) }, 0); + assert_eq!( + context.events, + vec![ + "release-jobs", + "jobs", + "buffer-pool", + "cctx-pool", + "seq-pool", + "serial-state", + "cdict", + "mtctx", + ] + ); + } + struct MockChunkCompressor { continue_inputs: Vec, end_inputs: Vec,