diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index b0e0e8137..947d32d91 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -1024,6 +1024,30 @@ typedef int (*ZSTDMT_jobTableInitFn)(void* jobTable, unsigned nbJobs, size_t jobSize); typedef void (*ZSTDMT_jobTableDestroyFn)(void* jobTable, unsigned nbJobs, size_t jobSize); +typedef struct { + void* jobTable; + unsigned nbJobs; + size_t jobSize; + ZSTD_customMem cMem; + ZSTDMT_jobTableDestroyFn destroySync; +} ZSTDMT_RustFreeJobsTableProjection; +typedef char ZSTDMT_rust_free_jobs_table_projection_layout[ + (sizeof(ZSTDMT_jobTableDestroyFn) == sizeof(void*) + && offsetof(ZSTDMT_RustFreeJobsTableProjection, jobTable) == 0 + && offsetof(ZSTDMT_RustFreeJobsTableProjection, nbJobs) + == sizeof(void*) + && offsetof(ZSTDMT_RustFreeJobsTableProjection, jobSize) + == (sizeof(void*) == 8 ? 16 : 8) + && offsetof(ZSTDMT_RustFreeJobsTableProjection, cMem) + == offsetof(ZSTDMT_RustFreeJobsTableProjection, jobSize) + + sizeof(size_t) + && offsetof(ZSTDMT_RustFreeJobsTableProjection, destroySync) + == offsetof(ZSTDMT_RustFreeJobsTableProjection, cMem) + + sizeof(ZSTD_customMem) + && sizeof(ZSTDMT_RustFreeJobsTableProjection) + == offsetof(ZSTDMT_RustFreeJobsTableProjection, destroySync) + + sizeof(ZSTDMT_jobTableDestroyFn)) + ? 1 : -1]; void* ZSTDMT_rust_createJobsTable( unsigned* nbJobsPtr, size_t jobSize, ZSTD_customMem cMem, ZSTDMT_jobTableInitFn initSync, ZSTDMT_jobTableDestroyFn destroySync); @@ -1032,6 +1056,8 @@ size_t ZSTDMT_rust_expandJobsTable( size_t jobSize, ZSTD_customMem cMem, ZSTDMT_jobTableInitFn initSync, ZSTDMT_jobTableDestroyFn destroySync); +void ZSTDMT_rust_freeJobsTable( + const ZSTDMT_RustFreeJobsTableProjection* projection); typedef struct ZSTDMT_bufferPool_s { ZSTDMT_RustBufferPool* rustPool; @@ -2214,8 +2240,14 @@ static void ZSTDMT_projectJob(void* opaque, unsigned jobID, static void ZSTDMT_freeJobsTable(ZSTDMT_jobDescription* jobTable, U32 nbJobs, ZSTD_customMem cMem) { if (jobTable == NULL) return; - ZSTDMT_job_table_destroy_sync(jobTable, nbJobs, sizeof(*jobTable)); - ZSTDMT_rust_job_table_free(jobTable, nbJobs, sizeof(*jobTable), cMem); + ZSTDMT_RustFreeJobsTableProjection const projection = { + jobTable, + nbJobs, + sizeof(*jobTable), + cMem, + ZSTDMT_job_table_destroy_sync + }; + ZSTDMT_rust_freeJobsTable(&projection); } int ZSTDMT_job_table_init_sync(void* jobTable, unsigned nbJobs, size_t jobSize) diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index b3ca2618f..97d2b63cf 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -4654,6 +4654,45 @@ unsafe fn free_job_table_storage(job_table: *mut c_void, custom_mem: ZstdCustomM type JobTableInitSync = unsafe extern "C" fn(*mut c_void, c_uint, usize) -> c_int; type JobTableDestroySync = unsafe extern "C" fn(*mut c_void, c_uint, usize); +/// Projection for the final MT job-table release. Rust owns the +/// destroy-before-storage-free order; C retains the descriptor synchronization +/// callback and custom allocator behind this opaque projection. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct ZSTDMT_RustFreeJobsTableProjection { + jobTable: *mut c_void, + nbJobs: c_uint, + jobSize: usize, + customMem: ZstdCustomMem, + destroySync: Option, +} + +const _: () = { + assert!(size_of::() == size_of::()); + assert!(offset_of!(ZSTDMT_RustFreeJobsTableProjection, jobTable) == 0); + assert!( + offset_of!(ZSTDMT_RustFreeJobsTableProjection, nbJobs) == size_of::() + ); + assert!( + offset_of!(ZSTDMT_RustFreeJobsTableProjection, jobSize) + == if size_of::() == 8 { 16 } else { 8 } + ); + assert!( + offset_of!(ZSTDMT_RustFreeJobsTableProjection, customMem) + == offset_of!(ZSTDMT_RustFreeJobsTableProjection, jobSize) + size_of::() + ); + assert!( + offset_of!(ZSTDMT_RustFreeJobsTableProjection, destroySync) + == offset_of!(ZSTDMT_RustFreeJobsTableProjection, customMem) + + size_of::() + ); + assert!( + size_of::() + == offset_of!(ZSTDMT_RustFreeJobsTableProjection, destroySync) + + size_of::() + ); +}; + unsafe fn free_job_table_with_sync( job_table: *mut c_void, nb_jobs: c_uint, @@ -4661,11 +4700,29 @@ unsafe fn free_job_table_with_sync( custom_mem: ZstdCustomMem, destroy_sync: JobTableDestroySync, ) { - if job_table.is_null() { + let projection = ZSTDMT_RustFreeJobsTableProjection { + jobTable: job_table, + nbJobs: nb_jobs, + jobSize: job_size, + customMem: custom_mem, + destroySync: Some(destroy_sync), + }; + unsafe { free_job_table_with_sync_projection(projection) }; +} + +unsafe fn free_job_table_with_sync_projection( + projection: ZSTDMT_RustFreeJobsTableProjection, +) { + if projection.jobTable.is_null() { return; } - unsafe { destroy_sync(job_table, nb_jobs, job_size) }; - unsafe { ZSTDMT_rust_job_table_free(job_table, nb_jobs, job_size, custom_mem) }; + let Some(destroy_sync) = projection.destroySync else { + return; + }; + unsafe { + destroy_sync(projection.jobTable, projection.nbJobs, projection.jobSize); + free_job_table_storage(projection.jobTable, projection.customMem); + } } unsafe fn create_job_table_with_sync( @@ -4777,6 +4834,19 @@ pub unsafe extern "C" fn ZSTDMT_rust_job_table_free( unsafe { free_job_table_storage(job_table, custom_mem) } } +/// Release the C-owned synchronization objects before Rust releases the +/// opaque job-table storage. This is the normal teardown counterpart to the +/// same order used by failed create/expand transactions. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_freeJobsTable( + projection: *const ZSTDMT_RustFreeJobsTableProjection, +) { + let Some(projection) = (unsafe { projection.as_ref() }).copied() else { + return; + }; + unsafe { free_job_table_with_sync_projection(projection) }; +} + /// Expand the C-owned MT job table when the requested worker count outgrows /// its masked capacity. The table storage and custom allocator calls stay in /// Rust; C supplies callbacks for the private descriptor synchronization @@ -8862,6 +8932,63 @@ mod tests { JOB_TABLE_DESTROY_CALLS.fetch_add(1, Ordering::Relaxed); } + #[derive(Default)] + struct FreeJobsTableTestContext { + events: Vec<&'static str>, + } + + unsafe extern "C" fn free_jobs_table_test_alloc( + _opaque: *mut c_void, + _size: usize, + ) -> *mut c_void { + ptr::null_mut() + } + + unsafe extern "C" fn free_jobs_table_test_destroy( + job_table: *mut c_void, + _nb_jobs: c_uint, + _job_size: usize, + ) { + unsafe { + (*job_table.cast::()) + .events + .push("destroy-sync"); + } + } + + unsafe extern "C" fn free_jobs_table_test_free( + opaque: *mut c_void, + allocation: *mut c_void, + ) { + assert_eq!(allocation, opaque); + unsafe { + (*opaque.cast::()) + .events + .push("free-storage"); + } + } + + #[test] + fn free_jobs_table_orders_sync_destruction_before_storage_release() { + let mut context = FreeJobsTableTestContext::default(); + let context_ptr = (&mut context as *mut FreeJobsTableTestContext).cast::(); + let projection = ZSTDMT_RustFreeJobsTableProjection { + jobTable: context_ptr, + nbJobs: 8, + jobSize: 128, + customMem: ZstdCustomMem { + customAlloc: Some(free_jobs_table_test_alloc), + customFree: Some(free_jobs_table_test_free), + opaque: context_ptr, + }, + destroySync: Some(free_jobs_table_test_destroy), + }; + + unsafe { ZSTDMT_rust_freeJobsTable(&projection) }; + + assert_eq!(context.events, ["destroy-sync", "free-storage"]); + } + fn rsync_prime_power() -> u64 { (0..RSYNC_LENGTH - 1).fold(1, |power, _| power.wrapping_mul(PRIME8_BYTES)) }