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

@@ -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;
+    assert(mtctx->doneJobID <= mtctx->nextJobID);
-    unsigned const jobID = mtctx->doneJobID;
+    return ZSTDMT_rust_toFlushNow(mtctx->doneJobID, mtctx->nextJobID,
-    assert(jobID <= mtctx->nextJobID);
+                                   mtctx->jobIDMask, mtctx,
-    if (jobID == mtctx->nextJobID) return 0;   /* no active job => nothing to flush */
+                                   ZSTDMT_projectJob);
-
-    /* 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;
 }
 
 
@@ -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;
+    ZSTDMT_RustInputRange const range = ZSTDMT_rust_getInputDataInUse(
-    unsigned const lastJobID = mtctx->nextJobID;
+            mtctx->doneJobID, mtctx->nextJobID, mtctx->jobIDMask,
-    unsigned jobID;
+            mtctx->roundBuff.capacity, mtctx->targetSectionSize,
-
+            mtctx, ZSTDMT_projectJob);
-    /* no need to check during first round */
+    return (Range){ range.start, range.size };
-    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;
 }
 
 /**

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<F>(
+    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_jobProjectionFn>,
+) -> 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<F>(
+    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<ZSTDMT_jobProjectionFn>,
+) -> 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,