From f8cabd39a481f7e384a1edf3c83c92ce8f849fb1 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sun, 19 Jul 2026 20:01:41 +0200 Subject: [PATCH] feat(mt): move job-resource slot iteration into Rust Move the deterministic inclusive job-table cleanup loop into the Rust MT seam. C retains each descriptor's mutex and condition variables, buffer-pool release, descriptor clearing, and the containing context's final state reset behind a per-slot callback. 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 | 35 +++++++++++++++---------- rust/src/zstdmt_compress.rs | 47 ++++++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 14 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 50d3c28aa..d34121287 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -379,6 +379,9 @@ typedef void (*ZSTDMT_waitForJobCompleteFn)( unsigned ZSTDMT_rust_waitForAllJobsCompleted( unsigned doneJobID, unsigned nextJobID, unsigned jobIDMask, void* opaque, ZSTDMT_waitForJobCompleteFn waitForJob); +typedef void (*ZSTDMT_releaseJobResourceFn)(void* opaque, unsigned jobID); +void ZSTDMT_rust_releaseAllJobResources( + unsigned jobIDMask, void* opaque, ZSTDMT_releaseJobResourceFn releaseJob); typedef void (*ZSTDMT_waitForLdmLockFn)(void* opaque); typedef int (*ZSTDMT_waitForLdmOverlapFn)( void* opaque, void* bufferStart, size_t bufferCapacity); @@ -1483,25 +1486,29 @@ ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbWorkers, ZSTD_customMem cMem, } +static void ZSTDMT_releaseJobResource(void* opaque, unsigned jobID) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + /* Copy the mutex/cond out */ + ZSTD_pthread_mutex_t const mutex = mtctx->jobs[jobID].job_mutex; + ZSTD_pthread_cond_t const cond = mtctx->jobs[jobID].job_cond; + + DEBUGLOG(4, "job%02u: release dst address %08X", jobID, (U32)(size_t)mtctx->jobs[jobID].dstBuff.start); + ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].dstBuff); + + /* Clear the job description, but keep the mutex/cond */ + ZSTD_memset(&mtctx->jobs[jobID], 0, sizeof(mtctx->jobs[jobID])); + mtctx->jobs[jobID].job_mutex = mutex; + mtctx->jobs[jobID].job_cond = cond; +} + /* ZSTDMT_releaseAllJobResources() : * note : ensure all workers are killed first ! */ static void ZSTDMT_releaseAllJobResources(ZSTDMT_CCtx* mtctx) { - unsigned jobID; DEBUGLOG(3, "ZSTDMT_releaseAllJobResources"); - for (jobID=0; jobID <= mtctx->jobIDMask; jobID++) { - /* Copy the mutex/cond out */ - ZSTD_pthread_mutex_t const mutex = mtctx->jobs[jobID].job_mutex; - ZSTD_pthread_cond_t const cond = mtctx->jobs[jobID].job_cond; - - DEBUGLOG(4, "job%02u: release dst address %08X", jobID, (U32)(size_t)mtctx->jobs[jobID].dstBuff.start); - ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].dstBuff); - - /* Clear the job description, but keep the mutex/cond */ - ZSTD_memset(&mtctx->jobs[jobID], 0, sizeof(mtctx->jobs[jobID])); - mtctx->jobs[jobID].job_mutex = mutex; - mtctx->jobs[jobID].job_cond = cond; - } + ZSTDMT_rust_releaseAllJobResources( + mtctx->jobIDMask, mtctx, ZSTDMT_releaseJobResource); mtctx->inBuff.buffer = g_nullBuffer; mtctx->inBuff.filled = 0; mtctx->allJobsCompleted = 1; diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 39a3a4315..9583c091f 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -97,6 +97,7 @@ pub struct ZSTDMT_serialStateEnsureFinishedResult { pub nextJobID: c_uint, } 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); 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; @@ -886,6 +887,23 @@ pub unsafe extern "C" fn ZSTDMT_rust_waitForAllJobsCompleted( done_job_id } +/// Visit every MT job slot so C can release its private descriptor resources. +/// Rust owns the deterministic inclusive ring-table iteration; C retains the +/// buffer-pool, mutex/condition, and descriptor cleanup side effects. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_releaseAllJobResources( + job_id_mask: c_uint, + opaque: *mut c_void, + release_job: Option, +) { + let Some(release_job) = release_job else { + return; + }; + for job_id in 0..=job_id_mask { + unsafe { release_job(opaque, job_id) }; + } +} + /// C ABI entry point for the worker-job orchestration. C supplies callbacks /// that keep the private descriptor, pools, mutexes, and codec operations on /// the C side of this narrow projection. @@ -3597,6 +3615,35 @@ mod tests { assert_eq!(context.events, vec![(3, 3), (0, 4), (1, 5)]); } + struct ReleaseAllJobsTestContext { + job_ids: Vec, + } + + unsafe extern "C" fn release_all_jobs_test_callback(context: *mut c_void, job_id: c_uint) { + unsafe { + (*context.cast::()) + .job_ids + .push(job_id); + } + } + + #[test] + fn release_all_jobs_visits_each_slot_in_order() { + let mut context = ReleaseAllJobsTestContext { + job_ids: Vec::new(), + }; + + unsafe { + ZSTDMT_rust_releaseAllJobResources( + 3, + (&mut context as *mut ReleaseAllJobsTestContext).cast(), + Some(release_all_jobs_test_callback), + ) + }; + + assert_eq!(context.job_ids, vec![0, 1, 2, 3]); + } + #[test] fn compression_job_stops_on_non_first_chunk_error_and_cleans_up() { let state = Rc::new(RefCell::new(MockCompressionJob::default()));