feat(compress): aggregate MT frame progression in Rust

Move the arithmetic aggregation in ZSTDMT_getFrameProgression across the existing C/Rust boundary. C still owns MT state traversal, mutex locking, jobIDMask ordering, nextJobID plus jobReady handling, error normalization, and the flushed <= produced assertion. Rust constructs the base result and folds each compact, C-normalized job snapshot into the complete repr(C) ZSTD_frameProgression, preserving size_t-to-U64 wrapping and active-worker counting without exposing private job or mutex layouts. Focused Rust tests cover zero jobs, a ready job, completed jobs, normalized error output, and mixed active/completed aggregation. Test Plan: - All three compression clippy commands passed before formatting. - cargo +nightly fmt --manifest-path rust/Cargo.toml -- passed. - All three compression clippy commands passed after formatting. - cargo test --manifest-path rust/Cargo.toml --no-default-features --features compression -- 332 passed. - make -B -C lib -j2 lib-mt -- passed. - make -B -C lib -j2 lib-nomt -- passed. - make -B -C tests -j2 test-zstream -- passed: 84 named, 6,004, and 8,392 randomized cases. - git diff --check and git diff --cached --check -- passed. The zstream build retained the pre-existing warning at tests/zstreamtest.c:1899 about an unterminated initializer string.
2026-07-18 14:05:46 +02:00
parent 4c114b8e7b
commit 3bd4dba2ee
2 changed files with 143 additions and 11 deletions
@@ -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);
        }
@@ -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::<usize>() == 4 { 29 } else { 30 };
 const ZSTD_WINDOWLOG_MAX: c_uint = if mem::size_of::<usize>() == 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,