feat(compress): move MT job-table storage lifecycle to Rust
The multithreaded compressor already uses Rust-owned buffer and CCtx pools, but zstdmt_compress.c still allocated and freed its job table directly. Move the raw table storage and power-of-two sizing behind Rust's custom allocator ABI. Keep descriptor field access and platform mutex/condition initialization in C because those layouts remain private and platform-specific. The C wrapper initializes and destroys synchronization primitives around the Rust storage calls, preserving failure cleanup while leaving worker job setup, scheduling, and stream entry points in C as the fallback implementation. Keep the Rust storage-only ABI free of MT-only C references so single-threaded archives can omit zstdmt_compress.c without acquiring new unresolved symbols. Test Plan: - `cargo test --manifest-path rust/Cargo.toml --no-default-features --features compression zstdmt_compress --lib` -- passed (5 tests). - `make -B -C lib lib-mt` -- passed. - `make -B -C tests -j2 fullbench zstreamtest poolTests` -- passed. - `./poolTests` -- passed. - `./fullbench -i1 -B1000 README.md` -- passed, including -T2 scenarios. - Rebuilt `programs/zstd` and ran a `-T2` compress/decompress `cmp` round-trip -- passed. - `rustfmt` and `cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check` -- passed. - Full `cargo clippy -D warnings` remains blocked by unrelated warnings in the concurrent `rust/src/fileio_asyncio.rs` worktree changes. - `./zstreamtest -T5s` reaches an unrelated single-thread maxBlockSize assertion at `tests/zstreamtest.c:2157`; MT fullbench and CLI smoke pass.
This commit is contained in:
@@ -796,13 +796,39 @@ struct ZSTDMT_CCtx_s {
|
||||
|
||||
static void ZSTDMT_freeJobsTable(ZSTDMT_jobDescription* jobTable, U32 nbJobs, ZSTD_customMem cMem)
|
||||
{
|
||||
U32 jobNb;
|
||||
if (jobTable == NULL) return;
|
||||
ZSTDMT_job_table_destroy_sync(jobTable, nbJobs, sizeof(*jobTable));
|
||||
ZSTDMT_rust_job_table_free(jobTable, nbJobs, sizeof(*jobTable), cMem);
|
||||
}
|
||||
|
||||
int ZSTDMT_job_table_init_sync(void* jobTable, unsigned nbJobs, size_t jobSize)
|
||||
{
|
||||
U32 jobNb;
|
||||
int initError = 0;
|
||||
BYTE* const table = (BYTE*)jobTable;
|
||||
assert(jobSize == sizeof(ZSTDMT_jobDescription));
|
||||
if (jobTable == NULL) return 1;
|
||||
for (jobNb=0; jobNb<nbJobs; jobNb++) {
|
||||
ZSTD_pthread_mutex_destroy(&jobTable[jobNb].job_mutex);
|
||||
ZSTD_pthread_cond_destroy(&jobTable[jobNb].job_cond);
|
||||
ZSTDMT_jobDescription* const job =
|
||||
(ZSTDMT_jobDescription*)(table + jobNb * jobSize);
|
||||
initError |= ZSTD_pthread_mutex_init(&job->job_mutex, NULL);
|
||||
initError |= ZSTD_pthread_cond_init(&job->job_cond, NULL);
|
||||
}
|
||||
return initError;
|
||||
}
|
||||
|
||||
void ZSTDMT_job_table_destroy_sync(void* jobTable, unsigned nbJobs, size_t jobSize)
|
||||
{
|
||||
U32 jobNb;
|
||||
BYTE* const table = (BYTE*)jobTable;
|
||||
if (jobTable == NULL) return;
|
||||
assert(jobSize == sizeof(ZSTDMT_jobDescription));
|
||||
for (jobNb=0; jobNb<nbJobs; jobNb++) {
|
||||
ZSTDMT_jobDescription* const job =
|
||||
(ZSTDMT_jobDescription*)(table + jobNb * jobSize);
|
||||
ZSTD_pthread_mutex_destroy(&job->job_mutex);
|
||||
ZSTD_pthread_cond_destroy(&job->job_cond);
|
||||
}
|
||||
ZSTD_customFree(jobTable, cMem);
|
||||
}
|
||||
|
||||
/* ZSTDMT_allocJobsTable()
|
||||
@@ -810,20 +836,11 @@ static void ZSTDMT_freeJobsTable(ZSTDMT_jobDescription* jobTable, U32 nbJobs, ZS
|
||||
* update *nbJobsPtr to next power of 2 value, as size of table */
|
||||
static ZSTDMT_jobDescription* ZSTDMT_createJobsTable(U32* nbJobsPtr, ZSTD_customMem cMem)
|
||||
{
|
||||
U32 const nbJobsLog2 = ZSTD_highbit32(*nbJobsPtr) + 1;
|
||||
U32 const nbJobs = 1 << nbJobsLog2;
|
||||
U32 jobNb;
|
||||
ZSTDMT_jobDescription* const jobTable = (ZSTDMT_jobDescription*)
|
||||
ZSTD_customCalloc(nbJobs * sizeof(ZSTDMT_jobDescription), cMem);
|
||||
int initError = 0;
|
||||
if (jobTable==NULL) return NULL;
|
||||
*nbJobsPtr = nbJobs;
|
||||
for (jobNb=0; jobNb<nbJobs; jobNb++) {
|
||||
initError |= ZSTD_pthread_mutex_init(&jobTable[jobNb].job_mutex, NULL);
|
||||
initError |= ZSTD_pthread_cond_init(&jobTable[jobNb].job_cond, NULL);
|
||||
}
|
||||
if (initError != 0) {
|
||||
ZSTDMT_freeJobsTable(jobTable, nbJobs, cMem);
|
||||
ZSTDMT_rust_job_table_create(nbJobsPtr, sizeof(ZSTDMT_jobDescription), cMem);
|
||||
if (jobTable == NULL) return NULL;
|
||||
if (ZSTDMT_job_table_init_sync(jobTable, *nbJobsPtr, sizeof(*jobTable)) != 0) {
|
||||
ZSTDMT_freeJobsTable(jobTable, *nbJobsPtr, cMem);
|
||||
return NULL;
|
||||
}
|
||||
return jobTable;
|
||||
|
||||
@@ -43,6 +43,17 @@
|
||||
|
||||
/* === Memory management === */
|
||||
typedef struct ZSTDMT_CCtx_s ZSTDMT_CCtx;
|
||||
|
||||
/* The Rust adapter owns job-table storage and custom allocator cleanup. The
|
||||
* descriptor layout and platform synchronization primitives remain in C, so
|
||||
* these callbacks are the explicit C fallback seam for job initialization. */
|
||||
void* ZSTDMT_rust_job_table_create(unsigned* nbJobsPtr, size_t jobSize,
|
||||
ZSTD_customMem cMem);
|
||||
void ZSTDMT_rust_job_table_free(void* jobTable, unsigned nbJobs,
|
||||
size_t jobSize, ZSTD_customMem cMem);
|
||||
int ZSTDMT_job_table_init_sync(void* jobTable, unsigned nbJobs, size_t jobSize);
|
||||
void ZSTDMT_job_table_destroy_sync(void* jobTable, unsigned nbJobs, size_t jobSize);
|
||||
|
||||
/* Requires ZSTD_MULTITHREAD to be defined during compilation, otherwise it will return NULL. */
|
||||
ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbWorkers,
|
||||
ZSTD_customMem cMem,
|
||||
|
||||
@@ -5,11 +5,12 @@
|
||||
|
||||
//! Rust-owned resource pools used by the multithreaded compressor.
|
||||
//!
|
||||
//! The job table, serial LDM state, and streaming state still use private C
|
||||
//! layouts. `zstdmt_compress.c` therefore keeps those parts and projects only
|
||||
//! the allocation pools into this module. The pool entry points below are a
|
||||
//! narrow C ABI: buffers and `ZSTD_CCtx *` values remain opaque to Rust, while
|
||||
//! allocation, reuse, expansion, and synchronization are Rust-owned.
|
||||
//! The serial LDM state, job descriptor fields, worker callback, and streaming
|
||||
//! state still use private C layouts. `zstdmt_compress.c` therefore keeps
|
||||
//! those operations and projects only allocation/lifecycle pieces into this
|
||||
//! module. The entry points below are narrow C ABIs: buffers, `ZSTD_CCtx *`
|
||||
//! values, and job descriptors remain opaque to Rust, while allocation, reuse,
|
||||
//! expansion, and synchronization of Rust-owned state are Rust-owned.
|
||||
|
||||
use std::mem::{self, MaybeUninit};
|
||||
use std::os::raw::{c_uint, c_void};
|
||||
@@ -109,6 +110,74 @@ fn checked_array_size<T>(len: usize) -> Option<usize> {
|
||||
mem::size_of::<T>().checked_mul(len)
|
||||
}
|
||||
|
||||
fn rounded_job_count(requested: c_uint) -> Option<c_uint> {
|
||||
if requested == 0 {
|
||||
return None;
|
||||
}
|
||||
// The original C expression is `1 << (highbit(requested) + 1)`, which
|
||||
// deliberately chooses a strictly larger power of two when requested is
|
||||
// already a power of two.
|
||||
let shift = usize::BITS - (requested as usize).leading_zeros();
|
||||
let count = 1usize.checked_shl(shift)?;
|
||||
if count > c_uint::MAX as usize {
|
||||
return None;
|
||||
}
|
||||
Some(count as c_uint)
|
||||
}
|
||||
|
||||
unsafe fn create_job_table(
|
||||
nb_jobs_ptr: *mut c_uint,
|
||||
job_size: usize,
|
||||
custom_mem: ZstdCustomMem,
|
||||
) -> *mut c_void {
|
||||
if nb_jobs_ptr.is_null() || job_size == 0 {
|
||||
return ptr::null_mut();
|
||||
}
|
||||
let Some(nb_jobs) = (unsafe { rounded_job_count(*nb_jobs_ptr) }) else {
|
||||
return ptr::null_mut();
|
||||
};
|
||||
let Some(table_size) = job_size.checked_mul(nb_jobs as usize) else {
|
||||
return ptr::null_mut();
|
||||
};
|
||||
let table = unsafe { custom_calloc(table_size, custom_mem) };
|
||||
if table.is_null() {
|
||||
return ptr::null_mut();
|
||||
}
|
||||
|
||||
unsafe { *nb_jobs_ptr = nb_jobs };
|
||||
table
|
||||
}
|
||||
|
||||
unsafe fn free_job_table_storage(job_table: *mut c_void, custom_mem: ZstdCustomMem) {
|
||||
if job_table.is_null() {
|
||||
return;
|
||||
}
|
||||
unsafe { custom_free(job_table, custom_mem) };
|
||||
}
|
||||
|
||||
#[no_mangle]
|
||||
pub unsafe extern "C" fn ZSTDMT_rust_job_table_create(
|
||||
nb_jobs_ptr: *mut c_uint,
|
||||
job_size: usize,
|
||||
custom_mem: ZstdCustomMem,
|
||||
) -> *mut c_void {
|
||||
unsafe { create_job_table(nb_jobs_ptr, job_size, custom_mem) }
|
||||
}
|
||||
|
||||
#[no_mangle]
|
||||
pub unsafe extern "C" fn ZSTDMT_rust_job_table_free(
|
||||
job_table: *mut c_void,
|
||||
_nb_jobs: c_uint,
|
||||
_job_size: usize,
|
||||
custom_mem: ZstdCustomMem,
|
||||
) {
|
||||
// The MT C adapter destroys its platform mutexes and condition variables
|
||||
// before calling this storage-only release function. Keeping this Rust
|
||||
// side free of MT-only C references also preserves single-threaded builds
|
||||
// where zstdmt_compress.c is intentionally omitted.
|
||||
unsafe { free_job_table_storage(job_table, custom_mem) }
|
||||
}
|
||||
|
||||
unsafe fn create_buffer_pool(
|
||||
max_nb_buffers: usize,
|
||||
custom_mem: ZstdCustomMem,
|
||||
@@ -601,4 +670,14 @@ mod tests {
|
||||
ZSTDMT_rust_buffer_pool_free(pool);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn job_table_count_matches_c_power_of_two_contract() {
|
||||
assert_eq!(rounded_job_count(0), None);
|
||||
assert_eq!(rounded_job_count(1), Some(2));
|
||||
assert_eq!(rounded_job_count(3), Some(4));
|
||||
assert_eq!(rounded_job_count(4), Some(8));
|
||||
assert_eq!(rounded_job_count(255), Some(256));
|
||||
assert_eq!(rounded_job_count(256), Some(512));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user