From 380d8075c68935c326e271b1ae4d069ab9affc84 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sat, 18 Jul 2026 22:08:46 +0200 Subject: [PATCH] feat(mt): move job-ring decisions into Rust Move multithreaded compression's input-retention and pending-output policy behind scalar job projections. Rust scans the ring and selects source or prefix ranges while C keeps job descriptors, mutexes, pointer ownership, and worker-private layout behind the projection callback. Test Plan: - cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1 - cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings - make -B -C lib -j2 lib - make -B -C programs -j2 zstd - make -B -C tests -j2 test-fuzzer - make -B -C tests -j2 test-zstd --- lib/compress/zstdmt_compress.c | 106 +++++++------- rust/src/zstdmt_compress.rs | 258 +++++++++++++++++++++++++++++++++ 2 files changed, 307 insertions(+), 57 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index d9d17e05f..8cda31afa 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -177,6 +177,28 @@ void ZSTDMT_rust_findSynchronizationPoint(const void* inputSrc, size_t inputSize U64 ZSTDMT_rust_rollingHashPrimePower(U32 length); size_t ZSTDMT_rust_nextInputSizeHint(size_t targetSectionSize, size_t inBuffFilled); +typedef struct { + size_t consumed; + size_t cSize; + const void* srcStart; + size_t srcSize; + const void* prefixStart; + size_t prefixSize; + size_t dstFlushed; +} ZSTDMT_RustJobProjection; +typedef struct { + const void* start; + size_t size; +} ZSTDMT_RustInputRange; +typedef void (*ZSTDMT_jobProjectionFn)(void* opaque, unsigned jobID, + ZSTDMT_RustJobProjection* projection); +ZSTDMT_RustInputRange ZSTDMT_rust_getInputDataInUse( + unsigned firstJobID, unsigned lastJobID, unsigned jobIDMask, + size_t roundBufferCapacity, size_t targetSectionSize, + void* opaque, ZSTDMT_jobProjectionFn projectJob); +size_t ZSTDMT_rust_toFlushNow(unsigned doneJobID, unsigned nextJobID, + unsigned jobIDMask, void* opaque, + ZSTDMT_jobProjectionFn projectJob); size_t ZSTDMT_rust_sizeofCCtx(size_t mtctxSize, size_t factorySize, size_t bufferPoolSize, size_t jobsSize, size_t cctxPoolSize, size_t seqPoolSize, @@ -885,6 +907,24 @@ struct ZSTDMT_CCtx_s { unsigned providedFactory: 1; }; +/* Project only the scalar job state needed by Rust's read-only MT + * orchestration. The descriptor layout and its mutex remain private here. */ +static void ZSTDMT_projectJob(void* opaque, unsigned jobID, + ZSTDMT_RustJobProjection* projection) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + ZSTDMT_jobDescription* const job = &mtctx->jobs[jobID]; + ZSTD_pthread_mutex_lock(&job->job_mutex); + projection->consumed = job->consumed; + projection->cSize = job->cSize; + projection->dstFlushed = job->dstFlushed; + ZSTD_pthread_mutex_unlock(&job->job_mutex); + projection->srcStart = job->src.start; + projection->srcSize = job->src.size; + projection->prefixStart = job->prefix.start; + projection->prefixSize = job->prefix.size; +} + static void ZSTDMT_freeJobsTable(ZSTDMT_jobDescription* jobTable, U32 nbJobs, ZSTD_customMem cMem) { if (jobTable == NULL) return; @@ -1160,34 +1200,10 @@ ZSTD_frameProgression ZSTDMT_getFrameProgression(ZSTDMT_CCtx* mtctx) size_t ZSTDMT_toFlushNow(ZSTDMT_CCtx* mtctx) { - size_t toFlush; - unsigned const jobID = mtctx->doneJobID; - assert(jobID <= mtctx->nextJobID); - if (jobID == mtctx->nextJobID) return 0; /* no active job => nothing to flush */ - - /* look into oldest non-fully-flushed job */ - { unsigned const wJobID = jobID & mtctx->jobIDMask; - ZSTDMT_jobDescription* const jobPtr = &mtctx->jobs[wJobID]; - ZSTD_pthread_mutex_lock(&jobPtr->job_mutex); - { size_t const cResult = jobPtr->cSize; - size_t const produced = ZSTD_isError(cResult) ? 0 : cResult; - size_t const flushed = ZSTD_isError(cResult) ? 0 : jobPtr->dstFlushed; - assert(flushed <= produced); - assert(jobPtr->consumed <= jobPtr->src.size); - toFlush = produced - flushed; - /* if toFlush==0, nothing is available to flush. - * However, jobID is expected to still be active: - * if jobID was already completed and fully flushed, - * ZSTDMT_flushProduced() should have already moved onto next job. - * Therefore, some input has not yet been consumed. */ - if (toFlush==0) { - assert(jobPtr->consumed < jobPtr->src.size); - } - } - ZSTD_pthread_mutex_unlock(&mtctx->jobs[wJobID].job_mutex); - } - - return toFlush; + assert(mtctx->doneJobID <= mtctx->nextJobID); + return ZSTDMT_rust_toFlushNow(mtctx->doneJobID, mtctx->nextJobID, + mtctx->jobIDMask, mtctx, + ZSTDMT_projectJob); } @@ -1556,35 +1572,11 @@ static size_t ZSTDMT_flushProduced(ZSTDMT_CCtx* mtctx, ZSTD_outBuffer* output, u */ static Range ZSTDMT_getInputDataInUse(ZSTDMT_CCtx* mtctx) { - unsigned const firstJobID = mtctx->doneJobID; - unsigned const lastJobID = mtctx->nextJobID; - unsigned jobID; - - /* no need to check during first round */ - size_t roundBuffCapacity = mtctx->roundBuff.capacity; - size_t nbJobs1stRoundMin = roundBuffCapacity / mtctx->targetSectionSize; - if (lastJobID < nbJobs1stRoundMin) return kNullRange; - - for (jobID = firstJobID; jobID < lastJobID; ++jobID) { - unsigned const wJobID = jobID & mtctx->jobIDMask; - size_t consumed; - - ZSTD_PTHREAD_MUTEX_LOCK(&mtctx->jobs[wJobID].job_mutex); - consumed = mtctx->jobs[wJobID].consumed; - ZSTD_pthread_mutex_unlock(&mtctx->jobs[wJobID].job_mutex); - - if (consumed < mtctx->jobs[wJobID].src.size) { - Range range = mtctx->jobs[wJobID].prefix; - if (range.size == 0) { - /* Empty prefix */ - range = mtctx->jobs[wJobID].src; - } - /* Job source in multiple segments not supported yet */ - assert(range.start <= mtctx->jobs[wJobID].src.start); - return range; - } - } - return kNullRange; + ZSTDMT_RustInputRange const range = ZSTDMT_rust_getInputDataInUse( + mtctx->doneJobID, mtctx->nextJobID, mtctx->jobIDMask, + mtctx->roundBuff.capacity, mtctx->targetSectionSize, + mtctx, ZSTDMT_projectJob); + return (Range){ range.start, range.size }; } /** diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index ea1ea65c6..d89da97f6 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -65,6 +65,36 @@ pub struct ZSTDMT_flushPublicationResult { pub dstFlushed: usize, } +/// Scalar snapshot of one C-owned job descriptor. +/// +/// The C adapter fills this projection while holding the descriptor mutex. +/// Rust owns only the ring-scan decisions below; the descriptor layout, +/// synchronization objects, and all pointer ownership stay in C. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_jobProjection { + pub consumed: usize, + pub cSize: usize, + pub srcStart: *const c_void, + pub srcSize: usize, + pub prefixStart: *const c_void, + pub prefixSize: usize, + pub dstFlushed: usize, +} + +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_inputRange { + pub start: *const c_void, + pub size: usize, +} + +pub type ZSTDMT_jobProjectionFn = unsafe extern "C" fn( + opaque: *mut c_void, + job_id: c_uint, + projection: *mut ZSTDMT_jobProjection, +); + const FLUSH_PUBLICATION_OK: c_int = 0; const FLUSH_PUBLICATION_INVALID_BOUNDS: c_int = 1; @@ -276,6 +306,140 @@ pub unsafe extern "C" fn ZSTDMT_rust_publishJobOutput( } } +#[inline] +fn get_input_data_in_use_with( + first_job_id: c_uint, + last_job_id: c_uint, + job_id_mask: c_uint, + round_buffer_capacity: usize, + target_section_size: usize, + mut project_job: F, +) -> ZSTDMT_inputRange +where + F: FnMut(c_uint) -> ZSTDMT_jobProjection, +{ + debug_assert!(target_section_size > 0); + if target_section_size == 0 { + return ZSTDMT_inputRange::default(); + } + + /* No job can refer to the first round buffer before this point. */ + let jobs_before_round_buffer_reuse = round_buffer_capacity / target_section_size; + /* C's usual arithmetic conversions promote `lastJobID` to size_t here. */ + if (last_job_id as usize) < jobs_before_round_buffer_reuse { + return ZSTDMT_inputRange::default(); + } + + for job_id in first_job_id..last_job_id { + let projection = project_job(job_id & job_id_mask); + if projection.consumed >= projection.srcSize { + continue; + } + + if projection.prefixSize > 0 { + debug_assert!((projection.prefixStart as usize) <= (projection.srcStart as usize)); + return ZSTDMT_inputRange { + start: projection.prefixStart, + size: projection.prefixSize, + }; + } + return ZSTDMT_inputRange { + start: projection.srcStart, + size: projection.srcSize, + }; + } + + ZSTDMT_inputRange::default() +} + +/// Find the earliest still-in-use input range in the C-owned job ring. +/// +/// C supplies each snapshot under its private job mutex. Rust performs the +/// ring ordering and prefix/source selection without receiving the job table +/// or any of its private synchronization layout. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_getInputDataInUse( + firstJobID: c_uint, + lastJobID: c_uint, + jobIDMask: c_uint, + roundBufferCapacity: usize, + targetSectionSize: usize, + opaque: *mut c_void, + projectJob: Option, +) -> ZSTDMT_inputRange { + let Some(project_job) = projectJob else { + return ZSTDMT_inputRange::default(); + }; + get_input_data_in_use_with( + firstJobID, + lastJobID, + jobIDMask, + roundBufferCapacity, + targetSectionSize, + |job_id| { + let mut projection = ZSTDMT_jobProjection::default(); + unsafe { project_job(opaque, job_id, &mut projection) }; + projection + }, + ) +} + +#[inline] +fn to_flush_now_with( + done_job_id: c_uint, + next_job_id: c_uint, + job_id_mask: c_uint, + mut project_job: F, +) -> usize +where + F: FnMut(c_uint) -> ZSTDMT_jobProjection, +{ + if done_job_id == next_job_id { + return 0; + } + + let projection = project_job(done_job_id & job_id_mask); + let produced = if ERR_isError(projection.cSize) { + 0 + } else { + projection.cSize + }; + let flushed = if ERR_isError(projection.cSize) { + 0 + } else { + projection.dstFlushed + }; + debug_assert!(flushed <= produced); + debug_assert!(projection.consumed <= projection.srcSize); + + let to_flush = produced.wrapping_sub(flushed); + if to_flush == 0 { + /* A live job with no output yet must still have input remaining. */ + debug_assert!(projection.consumed < projection.srcSize); + } + to_flush +} + +/// Return the output already produced by the oldest active C job but not yet +/// published to the caller's output buffer. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_toFlushNow( + doneJobID: c_uint, + nextJobID: c_uint, + jobIDMask: c_uint, + opaque: *mut c_void, + projectJob: Option, +) -> usize { + let Some(project_job) = projectJob else { + return 0; + }; + to_flush_now_with(doneJobID, nextJobID, jobIDMask, |job_id| { + let mut projection = ZSTDMT_jobProjection::default(); + unsafe { project_job(opaque, job_id, &mut projection) }; + projection + }) +} + #[inline] fn cycle_log(chain_log: c_uint, strategy: c_int) -> c_uint { chain_log.wrapping_sub((strategy >= ZSTD_BTLAZY2) as c_uint) @@ -1917,6 +2081,100 @@ mod tests { ); } + #[test] + fn input_data_in_use_skips_first_round_and_scans_oldest_ring_slot() { + let source_a = [0u8; 8]; + let backing = [1u8; 16]; + let source_b = &backing[8..]; + let prefix_b = &backing[..3]; + let projections = [ + ZSTDMT_jobProjection { + consumed: source_a.len(), + srcStart: source_a.as_ptr().cast(), + srcSize: source_a.len(), + ..ZSTDMT_jobProjection::default() + }, + ZSTDMT_jobProjection { + consumed: 2, + srcStart: source_b.as_ptr().cast(), + srcSize: source_b.len(), + prefixStart: prefix_b.as_ptr().cast(), + prefixSize: prefix_b.len(), + ..ZSTDMT_jobProjection::default() + }, + ]; + + let mut calls = 0; + let result = get_input_data_in_use_with(4, 6, 1, 16, 8, |slot| { + calls += 1; + projections[slot as usize] + }); + + assert_eq!(calls, 2); + assert_eq!(result.start, prefix_b.as_ptr().cast()); + assert_eq!(result.size, prefix_b.len()); + + let mut first_round_calls = 0; + let result = get_input_data_in_use_with(0, 3, 1, 32, 8, |slot| { + first_round_calls += 1; + projections[slot as usize] + }); + assert_eq!(first_round_calls, 0); + assert_eq!(result, ZSTDMT_inputRange::default()); + } + + #[test] + fn input_data_in_use_falls_back_to_source_without_prefix() { + let source = [3u8; 11]; + let projection = ZSTDMT_jobProjection { + consumed: 4, + srcStart: source.as_ptr().cast(), + srcSize: source.len(), + ..ZSTDMT_jobProjection::default() + }; + + let result = get_input_data_in_use_with(7, 8, 7, 8, 8, |_| projection); + assert_eq!(result.start, source.as_ptr().cast()); + assert_eq!(result.size, source.len()); + } + + #[test] + fn to_flush_now_projects_oldest_slot_and_normalizes_errors() { + let projection = ZSTDMT_jobProjection { + consumed: 5, + cSize: 23, + srcSize: 17, + dstFlushed: 9, + ..ZSTDMT_jobProjection::default() + }; + let mut seen_slot = None; + let to_flush = to_flush_now_with(5, 9, 3, |slot| { + seen_slot = Some(slot); + projection + }); + assert_eq!(seen_slot, Some(1)); + assert_eq!(to_flush, 14); + + let error_projection = ZSTDMT_jobProjection { + consumed: 0, + cSize: ERROR(ZstdErrorCode::Generic), + srcSize: 17, + dstFlushed: 23, + ..ZSTDMT_jobProjection::default() + }; + assert_eq!(to_flush_now_with(1, 2, 1, |_| error_projection), 0); + + let mut called = false; + assert_eq!( + to_flush_now_with(2, 2, 1, |_| { + called = true; + projection + }), + 0 + ); + assert!(!called); + } + fn call_synchronization_point( input: &[u8], input_pos: usize,