diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 3c5fffb72..cc5b8d028 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -157,6 +157,12 @@ size_t ZSTDMT_rust_sizeofCCtx(size_t mtctxSize, size_t factorySize, size_t bufferPoolSize, size_t jobsSize, size_t cctxPoolSize, size_t seqPoolSize, size_t cdictSize, size_t roundBuffSize); +ZSTD_frameProgression ZSTDMT_rust_frameProgression( + unsigned long long consumed, size_t inBuffFilled, + unsigned long long produced, unsigned currentJobID); +ZSTD_frameProgression ZSTDMT_rust_frameProgressionAddJob( + ZSTD_frameProgression progression, size_t srcSize, size_t consumed, + size_t produced, size_t flushed); typedef struct ZSTDMT_bufferPool_s { ZSTDMT_RustBufferPool* rustPool; @@ -1101,13 +1107,10 @@ 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; + ZSTD_frameProgression fps = ZSTDMT_rust_frameProgression( + mtctx->consumed, mtctx->inBuff.filled, mtctx->produced, + mtctx->nextJobID); DEBUGLOG(5, "ZSTDMT_getFrameProgression"); - fps.ingested = mtctx->consumed + mtctx->inBuff.filled; - fps.consumed = mtctx->consumed; - fps.produced = fps.flushed = mtctx->produced; - fps.currentJobID = mtctx->nextJobID; - fps.nbActiveWorkers = 0; { unsigned jobNb; unsigned lastJobNb = mtctx->nextJobID + mtctx->jobReady; assert(mtctx->jobReady <= 1); DEBUGLOG(6, "ZSTDMT_getFrameProgression: jobs: from %u to <%u (jobReady:%u)", @@ -1120,11 +1123,8 @@ ZSTD_frameProgression ZSTDMT_getFrameProgression(ZSTDMT_CCtx* mtctx) size_t const produced = ZSTD_isError(cResult) ? 0 : cResult; size_t const flushed = ZSTD_isError(cResult) ? 0 : jobPtr->dstFlushed; assert(flushed <= produced); - fps.ingested += jobPtr->src.size; - fps.consumed += jobPtr->consumed; - fps.produced += produced; - fps.flushed += flushed; - fps.nbActiveWorkers += (jobPtr->consumed < jobPtr->src.size); + fps = ZSTDMT_rust_frameProgressionAddJob( + fps, jobPtr->src.size, jobPtr->consumed, produced, flushed); } ZSTD_pthread_mutex_unlock(&mtctx->jobs[wJobID].job_mutex); } diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 21397ade5..57d595f35 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -18,6 +18,8 @@ use std::os::raw::{c_int, c_uint, c_void}; use std::ptr; use std::sync::Mutex; +use crate::zstd_compress::ZSTD_frameProgression; + const ZSTDMT_JOBLOG_MAX: c_uint = if mem::size_of::() == 4 { 29 } else { 30 }; const ZSTD_WINDOWLOG_MAX: c_uint = if mem::size_of::() == 4 { 30 } else { 31 }; @@ -146,6 +148,65 @@ pub extern "C" fn ZSTDMT_rust_computeOverlapSize( compute_overlap_size(windowLog, chainLog, strategy, overlapLog, enableLdm) } +#[inline] +fn frame_progression( + consumed: u64, + in_buff_filled: usize, + produced: u64, + current_job_id: c_uint, +) -> ZSTD_frameProgression { + ZSTD_frameProgression { + // C's usual arithmetic conversions promote size_t to U64 before wrapping. + ingested: consumed.wrapping_add(in_buff_filled as u64), + consumed, + produced, + flushed: produced, + currentJobID: current_job_id, + nbActiveWorkers: 0, + } +} + +#[inline] +fn frame_progression_add_job( + mut progression: ZSTD_frameProgression, + src_size: usize, + consumed: usize, + produced: usize, + flushed: usize, +) -> ZSTD_frameProgression { + progression.ingested = progression.ingested.wrapping_add(src_size as u64); + progression.consumed = progression.consumed.wrapping_add(consumed as u64); + progression.produced = progression.produced.wrapping_add(produced as u64); + progression.flushed = progression.flushed.wrapping_add(flushed as u64); + progression.nbActiveWorkers = progression + .nbActiveWorkers + .wrapping_add((consumed < src_size) as c_uint); + progression +} + +/// Construct the base MT frame progression from C-owned scalar state. +#[no_mangle] +pub extern "C" fn ZSTDMT_rust_frameProgression( + consumed: u64, + in_buff_filled: usize, + produced: u64, + current_job_id: c_uint, +) -> ZSTD_frameProgression { + frame_progression(consumed, in_buff_filled, produced, current_job_id) +} + +/// Add one C-normalized, mutex-protected job snapshot to MT frame progression. +#[no_mangle] +pub extern "C" fn ZSTDMT_rust_frameProgressionAddJob( + progression: ZSTD_frameProgression, + src_size: usize, + consumed: usize, + produced: usize, + flushed: usize, +) -> ZSTD_frameProgression { + frame_progression_add_job(progression, src_size, consumed, produced, flushed) +} + #[inline] unsafe fn rolling_hash_append(mut hash: u64, input: *const u8, size: usize) -> u64 { for pos in 0..size { @@ -1093,6 +1154,77 @@ mod tests { assert_eq!(ZSTDMT_rust_rollingHashPrimePower(0), 0x12A9_3A33_31E0_3D4B); } + #[test] + fn frame_progression_zero_jobs_keeps_mt_base_values() { + assert_eq!( + ZSTDMT_rust_frameProgression(11, 7, 13, 4), + ZSTD_frameProgression { + ingested: 18, + consumed: 11, + produced: 13, + flushed: 13, + currentJobID: 4, + nbActiveWorkers: 0, + } + ); + } + + #[test] + fn frame_progression_one_ready_job_counts_as_active() { + let base = ZSTDMT_rust_frameProgression(100, 20, 300, 9); + assert_eq!( + ZSTDMT_rust_frameProgressionAddJob(base, 80, 17, 42, 10), + ZSTD_frameProgression { + ingested: 200, + consumed: 117, + produced: 342, + flushed: 310, + currentJobID: 9, + nbActiveWorkers: 1, + } + ); + } + + #[test] + fn frame_progression_completed_job_is_not_active() { + let base = ZSTDMT_rust_frameProgression(100, 20, 300, 9); + let progression = ZSTDMT_rust_frameProgressionAddJob(base, 80, 80, 42, 42); + + assert_eq!(progression.nbActiveWorkers, 0); + assert_eq!(progression.ingested, 200); + assert_eq!(progression.consumed, 180); + } + + #[test] + fn frame_progression_error_job_uses_normalized_zero_output() { + let base = ZSTDMT_rust_frameProgression(100, 20, 300, 9); + let progression = ZSTDMT_rust_frameProgressionAddJob(base, 80, 0, 0, 0); + + assert_eq!(progression.produced, 300); + assert_eq!(progression.flushed, 300); + assert_eq!(progression.nbActiveWorkers, 1); + } + + #[test] + fn frame_progression_aggregates_active_and_completed_jobs() { + let base = ZSTDMT_rust_frameProgression(100, 20, 300, 9); + let progression = ZSTDMT_rust_frameProgressionAddJob(base, 80, 17, 42, 10); + let progression = ZSTDMT_rust_frameProgressionAddJob(progression, 30, 30, 8, 8); + let progression = ZSTDMT_rust_frameProgressionAddJob(progression, 40, 5, 7, 6); + + assert_eq!( + progression, + ZSTD_frameProgression { + ingested: 270, + consumed: 152, + produced: 357, + flushed: 324, + currentJobID: 9, + nbActiveWorkers: 2, + } + ); + } + fn call_synchronization_point( input: &[u8], input_pos: usize,