diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 2ae1b5101..083bd56d9 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -460,6 +460,11 @@ ZSTD_frameProgression ZSTDMT_rust_frameProgression( ZSTD_frameProgression ZSTDMT_rust_frameProgressionAddJob( ZSTD_frameProgression progression, size_t srcSize, size_t consumed, size_t produced, size_t flushed); +ZSTD_frameProgression ZSTDMT_rust_frameProgressionWithJobs( + unsigned long long consumed, size_t inBuffFilled, + unsigned long long produced, unsigned currentJobID, + unsigned doneJobID, unsigned nextJobID, unsigned jobReady, + unsigned jobIDMask, void* opaque, ZSTDMT_jobProjectionFn projectJob); typedef int (*ZSTDMT_jobTableInitFn)(void* jobTable, unsigned nbJobs, size_t jobSize); typedef void (*ZSTDMT_jobTableDestroyFn)(void* jobTable, unsigned nbJobs, @@ -1523,28 +1528,12 @@ void ZSTDMT_updateCParams_whileCompressing(ZSTDMT_CCtx* mtctx, const ZSTD_CCtx_p * Note : mutex will be acquired during statistics collection inside workers. */ ZSTD_frameProgression ZSTDMT_getFrameProgression(ZSTDMT_CCtx* mtctx) { - ZSTD_frameProgression fps = ZSTDMT_rust_frameProgression( - mtctx->consumed, mtctx->inBuff.filled, mtctx->produced, - mtctx->nextJobID); + ZSTD_frameProgression const fps = ZSTDMT_rust_frameProgressionWithJobs( + mtctx->consumed, mtctx->inBuff.filled, mtctx->produced, + mtctx->nextJobID, mtctx->doneJobID, mtctx->nextJobID, + mtctx->jobReady, mtctx->jobIDMask, mtctx, + ZSTDMT_projectJob); DEBUGLOG(5, "ZSTDMT_getFrameProgression"); - { unsigned jobNb; - unsigned lastJobNb = mtctx->nextJobID + mtctx->jobReady; assert(mtctx->jobReady <= 1); - DEBUGLOG(6, "ZSTDMT_getFrameProgression: jobs: from %u to <%u (jobReady:%u)", - mtctx->doneJobID, lastJobNb, mtctx->jobReady); - for (jobNb = mtctx->doneJobID ; jobNb < lastJobNb ; jobNb++) { - unsigned const wJobID = jobNb & mtctx->jobIDMask; - ZSTDMT_jobDescription* 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); - fps = ZSTDMT_rust_frameProgressionAddJob( - fps, jobPtr->src.size, jobPtr->consumed, produced, flushed); - } - ZSTD_pthread_mutex_unlock(&mtctx->jobs[wJobID].job_mutex); - } - } return fps; } diff --git a/rust/README.md b/rust/README.md index 09fccbf44..5b6e2e655 100644 --- a/rust/README.md +++ b/rust/README.md @@ -95,7 +95,9 @@ zstd ABI: private to C while Rust owns input-retention scans, reusable input-range overlap decisions, outer scheduling and end-directive adjustments, job-creation decisions, compression-job stage sequencing and error flow, - and pending-output decisions through scalar job projections. + pending-output decisions through scalar job projections, and frame + progression's job-ring scan, error normalization, and active-worker + accounting. - Dictionary support - `zstd_ddict` owns, loads, copies, and references decode dictionaries. - Legacy decoding @@ -145,7 +147,8 @@ shared- and separate-destination multi-file compression schedulers, shared- and separate-destination decompression scheduling, single-threaded stream initialization and the buffered/stable stream state machine, MT stream initialization, MT outer scheduling and flush policy, MT -compression-job stage sequencing and error flow, public sequence-API +compression-job stage sequencing and error flow, MT frame-progression job +aggregation, public sequence-API orchestration, sequence-store and block policy, external-producer invocation and success-path validation, external-sequence-store reset, external-sequence/literals block loop, optional-format decompression loops, diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 9335a098e..edab176a6 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -1957,6 +1957,43 @@ fn frame_progression_add_job( progression } +#[inline] +fn frame_progression_with_jobs( + consumed: u64, + in_buff_filled: usize, + produced: u64, + current_job_id: c_uint, + done_job_id: c_uint, + next_job_id: c_uint, + job_ready: c_uint, + job_id_mask: c_uint, + mut project_job: F, +) -> ZSTD_frameProgression +where + F: FnMut(c_uint) -> ZSTDMT_jobProjection, +{ + let mut progression = frame_progression(consumed, in_buff_filled, produced, current_job_id); + debug_assert!(job_ready <= 1); + let last_job_id = next_job_id.wrapping_add(job_ready); + for job_id in done_job_id..last_job_id { + let projection = project_job(job_id & job_id_mask); + let (produced, flushed) = if ERR_isError(projection.cSize) { + (0, 0) + } else { + (projection.cSize, projection.dstFlushed) + }; + debug_assert!(flushed <= produced); + progression = frame_progression_add_job( + progression, + projection.srcSize, + projection.consumed, + produced, + flushed, + ); + } + progression +} + /// Construct the base MT frame progression from C-owned scalar state. #[no_mangle] pub extern "C" fn ZSTDMT_rust_frameProgression( @@ -1980,6 +2017,42 @@ pub extern "C" fn ZSTDMT_rust_frameProgressionAddJob( frame_progression_add_job(progression, src_size, consumed, produced, flushed) } +/// Aggregate the base MT progression with mutex-protected C job snapshots. +/// Rust owns only the ring scan and normalization policy; C retains the job +/// descriptor layout and takes each descriptor mutex in `projectJob`. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_frameProgressionWithJobs( + consumed: u64, + in_buff_filled: usize, + produced: u64, + current_job_id: c_uint, + done_job_id: c_uint, + next_job_id: c_uint, + job_ready: c_uint, + job_id_mask: c_uint, + opaque: *mut c_void, + project_job: Option, +) -> ZSTD_frameProgression { + let Some(project_job) = project_job else { + return frame_progression(consumed, in_buff_filled, produced, current_job_id); + }; + frame_progression_with_jobs( + consumed, + in_buff_filled, + produced, + current_job_id, + done_job_id, + next_job_id, + job_ready, + job_id_mask, + |job_id| { + let mut projection = ZSTDMT_jobProjection::default(); + unsafe { project_job(opaque, job_id, &mut projection) }; + projection + }, + ) +} + #[inline] unsafe fn rolling_hash_append(mut hash: u64, input: *const u8, size: usize) -> u64 { for pos in 0..size { @@ -4469,6 +4542,59 @@ mod tests { ); } + #[test] + fn frame_progression_with_jobs_masks_slots_and_normalizes_errors() { + let projections = [ + ZSTDMT_jobProjection { + consumed: 50, + cSize: 8, + srcSize: 50, + dstFlushed: 8, + ..ZSTDMT_jobProjection::default() + }, + ZSTDMT_jobProjection { + consumed: 17, + cSize: 42, + srcSize: 80, + dstFlushed: 10, + ..ZSTDMT_jobProjection::default() + }, + ZSTDMT_jobProjection { + consumed: 30, + cSize: ERROR(ZstdErrorCode::SequenceProducerFailed), + srcSize: 30, + dstFlushed: 29, + ..ZSTDMT_jobProjection::default() + }, + ZSTDMT_jobProjection { + consumed: 5, + cSize: 7, + srcSize: 40, + dstFlushed: 6, + ..ZSTDMT_jobProjection::default() + }, + ]; + let mut calls = Vec::new(); + + let progression = frame_progression_with_jobs(100, 20, 300, 9, 5, 8, 1, 3, |job_id| { + calls.push(job_id); + projections[job_id as usize] + }); + + assert_eq!(calls, [1, 2, 3, 0]); + assert_eq!( + progression, + ZSTD_frameProgression { + ingested: 320, + consumed: 202, + produced: 357, + flushed: 324, + currentJobID: 9, + nbActiveWorkers: 2, + } + ); + } + #[test] fn input_data_in_use_skips_first_round_and_scans_oldest_ring_slot() { let source_a = [0u8; 8];