feat(mt): move serial sequence turn policy into Rust

Move MT serial turn/skip control and the ordering of LDM generation,
checksum updates, and turn advancement into the Rust rewrite. Keep the C
serial mutexes, LDM window/hash state, checksum state, and codec callbacks
behind a narrow callback bridge so the synchronization and private layouts
remain unchanged. Add focused tests for skipped predecessors, empty LDM
turns, and LDM-before-checksum ordering.

Test Plan:
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo fmt --manifest-path rust/Cargo.toml -- --check
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml serial_turn_ -- --nocapture
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings
- ulimit -v 41943040 make -B -C programs -j1 zstd
- ulimit -v 41943040 make -C tests -j1 test-zstream ZSTREAM_TESTTIME=-T1s

lib/compress/zstdmt_compress.c

@@ -354,6 +354,20 @@ typedef char ZSTDMT_rust_raw_seq_store_layout[
 
 ZSTDMT_RustRawSeqStore ZSTDMT_rust_bufferToSeq(ZSTDMT_RustBuffer buffer);
 ZSTDMT_RustBuffer ZSTDMT_rust_seqToBuffer(ZSTDMT_RustRawSeqStore seq);
+typedef int (*ZSTDMT_serialWaitForTurnFn)(void* opaque, unsigned jobID);
+typedef void (*ZSTDMT_serialGenerateLdmFn)(
+        void* opaque, ZSTDMT_RustRawSeqStore* seqStore,
+        const void* src, size_t srcSize);
+typedef void (*ZSTDMT_serialUpdateChecksumFn)(
+        void* opaque, const void* src, size_t srcSize);
+typedef void (*ZSTDMT_serialAdvanceFn)(void* opaque);
+void ZSTDMT_rust_serialStateGenSequences(
+        ZSTDMT_RustRawSeqStore* seqStore, const void* src, size_t srcSize,
+        unsigned jobID, int ldmEnabled, int checksumEnabled, void* opaque,
+        ZSTDMT_serialWaitForTurnFn waitForTurn,
+        ZSTDMT_serialGenerateLdmFn generateLdm,
+        ZSTDMT_serialUpdateChecksumFn updateChecksum,
+        ZSTDMT_serialAdvanceFn advance);
 
 unsigned ZSTDMT_rust_computeTargetJobLog(unsigned windowLog, unsigned chainLog,
                                          int strategy, int enableLdm);
@@ -862,44 +876,57 @@ static void ZSTDMT_serialState_free(SerialState* serialState)
     ZSTD_customFree(serialState->ldmState.bucketOffsets, cMem);
 }
 
-static void
-ZSTDMT_serialState_genSequences(SerialState* serialState,
-                                RawSeqStore_t* seqStore,
-                                Range src, unsigned jobID)
+/* Rust owns the serial turn/skip decision and operation ordering.  The wait
+ * callback intentionally leaves the main serial mutex locked; the advance
+ * callback releases it after Rust has performed the current turn's work. */
+static int ZSTDMT_serialState_waitForTurn(void* opaque, unsigned jobID)
 {
-    /* Wait for our turn */
+    SerialState* const serialState = (SerialState*)opaque;
     ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex);
     while (serialState->nextJobID < jobID) {
         DEBUGLOG(5, "wait for serialState->cond");
         ZSTD_pthread_cond_wait(&serialState->cond, &serialState->mutex);
     }
-    /* A future job may error and skip our job */
-    if (serialState->nextJobID == jobID) {
-        /* It is now our turn, do any processing necessary */
-        if (serialState->params.ldmParams.enableLdm == ZSTD_ps_enable) {
-            size_t error;
-            DEBUGLOG(6, "ZSTDMT_serialState_genSequences: LDM update");
-            assert(seqStore->seq != NULL && seqStore->pos == 0 &&
-                   seqStore->size == 0 && seqStore->capacity > 0);
-            assert(src.size <= serialState->params.jobSize);
-            ZSTD_window_update(&serialState->ldmState.window, src.start, src.size, /* forceNonContiguous */ 0);
-            error = ZSTD_ldm_generateSequences(
-                &serialState->ldmState, seqStore,
-                &serialState->params.ldmParams, src.start, src.size);
-            /* We provide a large enough buffer to never fail. */
-            assert(!ZSTD_isError(error)); (void)error;
-            /* Update ldmWindow to match the ldmState.window and signal the main
-             * thread if it is waiting for a buffer.
-             */
-            ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex);
-            serialState->ldmWindow = serialState->ldmState.window;
-            ZSTD_pthread_cond_signal(&serialState->ldmWindowCond);
-            ZSTD_pthread_mutex_unlock(&serialState->ldmWindowMutex);
-        }
-        if (serialState->params.fParams.checksumFlag && src.size > 0)
-            XXH64_update(&serialState->xxhState, src.start, src.size);
-    }
-    /* Now it is the next jobs turn */
+    return serialState->nextJobID == jobID;
+}
+
+static void ZSTDMT_serialState_generateLdm(
+        void* opaque, ZSTDMT_RustRawSeqStore* seqStore,
+        const void* src, size_t srcSize)
+{
+    SerialState* const serialState = (SerialState*)opaque;
+    RawSeqStore_t* const cSeqStore = (RawSeqStore_t*)seqStore;
+    size_t error;
+
+    DEBUGLOG(6, "ZSTDMT_serialState_genSequences: LDM update");
+    assert(cSeqStore->seq != NULL && cSeqStore->pos == 0 &&
+           cSeqStore->size == 0 && cSeqStore->capacity > 0);
+    assert(srcSize <= serialState->params.jobSize);
+    ZSTD_window_update(&serialState->ldmState.window, src, srcSize,
+                       /* forceNonContiguous */ 0);
+    error = ZSTD_ldm_generateSequences(
+        &serialState->ldmState, cSeqStore,
+        &serialState->params.ldmParams, src, srcSize);
+    /* We provide a large enough buffer to never fail. */
+    assert(!ZSTD_isError(error)); (void)error;
+    /* Update ldmWindow to match the ldmState.window and signal the main
+     * thread if it is waiting for a buffer. */
+    ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex);
+    serialState->ldmWindow = serialState->ldmState.window;
+    ZSTD_pthread_cond_signal(&serialState->ldmWindowCond);
+    ZSTD_pthread_mutex_unlock(&serialState->ldmWindowMutex);
+}
+
+static void ZSTDMT_serialState_updateChecksum(
+        void* opaque, const void* src, size_t srcSize)
+{
+    SerialState* const serialState = (SerialState*)opaque;
+    XXH64_update(&serialState->xxhState, src, srcSize);
+}
+
+static void ZSTDMT_serialState_advance(void* opaque)
+{
+    SerialState* const serialState = (SerialState*)opaque;
     serialState->nextJobID++;
     ZSTD_pthread_cond_broadcast(&serialState->cond);
     ZSTD_pthread_mutex_unlock(&serialState->mutex);
@@ -1033,8 +1060,16 @@ static void ZSTDMT_compressionJobGenerateSequences(void* opaque)
     ZSTDMT_jobDescription* const job = state->job;
 
     /* Perform serial step as early as possible. */
-    ZSTDMT_serialState_genSequences(job->serial, &state->rawSeqStore,
-                                    job->src, job->jobID);
+    ZSTDMT_rust_serialStateGenSequences(
+            (ZSTDMT_RustRawSeqStore*)&state->rawSeqStore,
+            job->src.start, job->src.size, job->jobID,
+            job->serial->params.ldmParams.enableLdm == ZSTD_ps_enable,
+            job->serial->params.fParams.checksumFlag,
+            job->serial,
+            ZSTDMT_serialState_waitForTurn,
+            ZSTDMT_serialState_generateLdm,
+            ZSTDMT_serialState_updateChecksum,
+            ZSTDMT_serialState_advance);
 }
 
 static size_t ZSTDMT_compressionJobBegin(void* opaque)

rust/README.md

@@ -97,8 +97,10 @@ zstd ABI:
     job-creation decisions, compression-job stage sequencing and error flow,
     pending-output decisions through scalar job projections, and frame
     progression's job-ring scan, error normalization, and active-worker
-    accounting.  Rust also owns frame-block preparation ordering; C callbacks
-    retain the private match-state window and workspace operations.
+    accounting.  Rust also owns frame-block preparation ordering and the MT
+    serial turn/skip policy, including LDM-before-checksum sequencing; C
+    callbacks retain the private match-state window, workspace operations,
+    synchronization, LDM state, and checksum state.
 - Dictionary support
   - `zstd_ddict` owns, loads, copies, and references decode dictionaries.
 - Legacy decoding
@@ -149,7 +151,8 @@ 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, MT frame-progression job
-aggregation, frame-block preparation ordering, public sequence-API
+aggregation, frame-block preparation ordering, MT serial turn/skip and
+LDM/checksum sequencing, 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,

rust/src/zstdmt_compress.rs

@@ -84,6 +84,12 @@ pub type ZSTDMT_compressionJobCompressFn =
 pub type ZSTDMT_compressionJobErrorFn = unsafe extern "C" fn(*mut c_void, usize);
 pub type ZSTDMT_compressionJobFinishFn = unsafe extern "C" fn(*mut c_void, usize);
 
+pub type ZSTDMT_serialWaitForTurnFn = unsafe extern "C" fn(*mut c_void, c_uint) -> c_int;
+pub type ZSTDMT_serialGenerateLdmFn =
+    unsafe extern "C" fn(*mut c_void, *mut ZstdMtRawSeqStore, *const c_void, usize);
+pub type ZSTDMT_serialUpdateChecksumFn = unsafe extern "C" fn(*mut c_void, *const c_void, usize);
+pub type ZSTDMT_serialAdvanceFn = unsafe extern "C" fn(*mut c_void);
+
 /// Scalar inputs for the MT streaming initializer.  The full parameter
 /// object, dictionary handles, pools, buffers, and synchronization remain
 /// private to C. Rust owns the order in which the C callbacks are invoked and
@@ -686,6 +692,82 @@ fn compression_job_with<A, P, S, B, Q, H, C, T, E, F>(
     finish_job(last_block_size);
 }
 
+/// Drive one ordered MT serial-state turn.
+///
+/// The wait callback returns with the C-owned serial mutex held.  Rust then
+/// performs the LDM-before-checksum policy only for the current job and always
+/// invokes the advance callback exactly once, including when an earlier job
+/// has already skipped this turn.
+#[inline]
+fn serial_state_gen_sequences_with<W, L, C, A>(
+    seq_store: &mut ZstdMtRawSeqStore,
+    src: *const c_void,
+    src_size: usize,
+    job_id: c_uint,
+    ldm_enabled: bool,
+    checksum_enabled: bool,
+    mut wait_for_turn: W,
+    mut generate_ldm: L,
+    mut update_checksum: C,
+    mut advance: A,
+) where
+    W: FnMut(c_uint) -> bool,
+    L: FnMut(&mut ZstdMtRawSeqStore, *const c_void, usize),
+    C: FnMut(*const c_void, usize),
+    A: FnMut(),
+{
+    if wait_for_turn(job_id) {
+        if ldm_enabled {
+            generate_ldm(seq_store, src, src_size);
+        }
+        if checksum_enabled && src_size != 0 {
+            update_checksum(src, src_size);
+        }
+    }
+    advance();
+}
+
+/// Own MT serial turn/skip and LDM/checksum ordering while C retains the
+/// mutexes, LDM window/hash state, checksum state, and raw sequence storage.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_serialStateGenSequences(
+    seq_store: *mut ZstdMtRawSeqStore,
+    src: *const c_void,
+    src_size: usize,
+    job_id: c_uint,
+    ldm_enabled: c_int,
+    checksum_enabled: c_int,
+    opaque: *mut c_void,
+    wait_for_turn: Option<ZSTDMT_serialWaitForTurnFn>,
+    generate_ldm: Option<ZSTDMT_serialGenerateLdmFn>,
+    update_checksum: Option<ZSTDMT_serialUpdateChecksumFn>,
+    advance: Option<ZSTDMT_serialAdvanceFn>,
+) {
+    let (Some(wait_for_turn), Some(generate_ldm), Some(update_checksum), Some(advance)) =
+        (wait_for_turn, generate_ldm, update_checksum, advance)
+    else {
+        return;
+    };
+    if seq_store.is_null() {
+        return;
+    }
+
+    unsafe {
+        serial_state_gen_sequences_with(
+            &mut *seq_store,
+            src,
+            src_size,
+            job_id,
+            ldm_enabled != 0,
+            checksum_enabled != 0,
+            |job_id| wait_for_turn(opaque, job_id) != 0,
+            |seq_store, src, src_size| generate_ldm(opaque, seq_store, src, src_size),
+            |src, src_size| update_checksum(opaque, src, src_size),
+            || advance(opaque),
+        );
+    }
+}
+
 /// C ABI entry point for the worker-job orchestration. C supplies callbacks
 /// that keep the private descriptor, pools, mutexes, and codec operations on
 /// the C side of this narrow projection.
@@ -3156,6 +3238,86 @@ mod tests {
         assert_eq!(state.finished, vec![7]);
     }
 
+    #[test]
+    fn serial_turn_runs_ldm_before_checksum_and_advances_once() {
+        let events = Rc::new(RefCell::new(Vec::new()));
+        let wait_events = Rc::clone(&events);
+        let ldm_events = Rc::clone(&events);
+        let checksum_events = Rc::clone(&events);
+        let advance_events = Rc::clone(&events);
+        let mut seq_store = ZstdMtRawSeqStore::default();
+
+        serial_state_gen_sequences_with(
+            &mut seq_store,
+            ptr::null(),
+            8,
+            4,
+            true,
+            true,
+            move |job_id| {
+                assert_eq!(job_id, 4);
+                wait_events.borrow_mut().push("wait");
+                true
+            },
+            move |_seq_store, _src, src_size| {
+                assert_eq!(src_size, 8);
+                ldm_events.borrow_mut().push("ldm");
+            },
+            move |_src, src_size| {
+                assert_eq!(src_size, 8);
+                checksum_events.borrow_mut().push("checksum");
+            },
+            move || advance_events.borrow_mut().push("advance"),
+        );
+
+        assert_eq!(&*events.borrow(), &["wait", "ldm", "checksum", "advance"]);
+    }
+
+    #[test]
+    fn serial_turn_skips_failed_predecessor_and_keeps_empty_ldm_turn() {
+        let events = Rc::new(RefCell::new(Vec::new()));
+        let wait_events = Rc::clone(&events);
+        let advance_events = Rc::clone(&events);
+        let mut seq_store = ZstdMtRawSeqStore::default();
+
+        serial_state_gen_sequences_with(
+            &mut seq_store,
+            ptr::null(),
+            0,
+            9,
+            true,
+            true,
+            move |_job_id| {
+                wait_events.borrow_mut().push("wait");
+                false
+            },
+            |_seq_store, _src, _src_size| panic!("skipped jobs must not generate LDM sequences"),
+            |_src, _src_size| panic!("skipped jobs must not update the checksum"),
+            move || advance_events.borrow_mut().push("advance"),
+        );
+        assert_eq!(&*events.borrow(), &["wait", "advance"]);
+
+        let empty_events = Rc::new(RefCell::new(Vec::new()));
+        let empty_ldm_events = Rc::clone(&empty_events);
+        let empty_advance_events = Rc::clone(&empty_events);
+        serial_state_gen_sequences_with(
+            &mut seq_store,
+            ptr::null(),
+            0,
+            10,
+            true,
+            true,
+            |_job_id| true,
+            move |_seq_store, _src, src_size| {
+                assert_eq!(src_size, 0);
+                empty_ldm_events.borrow_mut().push("ldm");
+            },
+            |_src, _src_size| panic!("empty input must not update the checksum"),
+            move || empty_advance_events.borrow_mut().push("advance"),
+        );
+        assert_eq!(&*empty_events.borrow(), &["ldm", "advance"]);
+    }
+
     #[test]
     fn compression_job_stops_on_non_first_chunk_error_and_cleans_up() {
         let state = Rc::new(RefCell::new(MockCompressionJob::default()));