refactor(mt): move context teardown order to Rust
Move ZSTDMT_freeCCtx ownership branches and teardown ordering into Rust while keeping the MT context, worker pool, synchronization, dictionary, and custom allocator layouts private to C callbacks. Preserve the original factory and round-buffer conditions, release jobs before destroying the pools, and keep free-on-null compatible with the C API. Test Plan: - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml free_cctx --lib - ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings - ulimit -v 41943040; make -j1
This commit is contained in:
+108
-14
@@ -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)
|
||||
|
||||
@@ -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<ZSTDMT_freeCCtxFn>,
|
||||
release_all_job_resources: Option<ZSTDMT_freeCCtxFn>,
|
||||
free_jobs: Option<ZSTDMT_freeCCtxFn>,
|
||||
free_buffer_pool: Option<ZSTDMT_freeCCtxFn>,
|
||||
free_cctx_pool: Option<ZSTDMT_freeCCtxFn>,
|
||||
free_seq_pool: Option<ZSTDMT_freeCCtxFn>,
|
||||
free_serial_state: Option<ZSTDMT_freeCCtxFn>,
|
||||
free_cdict: Option<ZSTDMT_freeCCtxFn>,
|
||||
free_round_buffer: Option<ZSTDMT_freeCCtxFn>,
|
||||
free_mtctx: Option<ZSTDMT_freeCCtxFn>,
|
||||
}
|
||||
|
||||
const _: () = {
|
||||
assert!(size_of::<ZSTDMT_freeCCtxFn>() == size_of::<usize>());
|
||||
assert!(offset_of!(ZSTDMT_RustFreeCCtxState, callback_context) == 0);
|
||||
assert!(offset_of!(ZSTDMT_RustFreeCCtxState, provided_factory) == size_of::<usize>());
|
||||
assert!(offset_of!(ZSTDMT_RustFreeCCtxState, round_buffer_present) == 2 * size_of::<usize>());
|
||||
assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_factory) == 3 * size_of::<usize>());
|
||||
assert!(
|
||||
offset_of!(ZSTDMT_RustFreeCCtxState, release_all_job_resources) == 4 * size_of::<usize>()
|
||||
);
|
||||
assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_jobs) == 5 * size_of::<usize>());
|
||||
assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_buffer_pool) == 6 * size_of::<usize>());
|
||||
assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_cctx_pool) == 7 * size_of::<usize>());
|
||||
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::<usize>());
|
||||
assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_cdict) == 10 * size_of::<usize>());
|
||||
assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_round_buffer) == 11 * size_of::<usize>());
|
||||
assert!(offset_of!(ZSTDMT_RustFreeCCtxState, free_mtctx) == 12 * size_of::<usize>());
|
||||
assert!(size_of::<ZSTDMT_RustFreeCCtxState>() == 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::<FreeCCtxTestContext>()).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<usize>,
|
||||
end_inputs: Vec<usize>,
|
||||
|
||||
Reference in New Issue
Block a user