feat(mt): move LDM window wait policy into Rust

Move the MT reusable-input wait decision and lock/overlap/condition-wait
ordering into Rust through a small callback projection. C retains ownership
of the pthread synchronization objects, private LDM window, and diagnostics.
Add layout assertions and callback-order tests for enabled and disabled LDM.

Test Plan:
- 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

@@ -368,6 +368,33 @@ void ZSTDMT_rust_serialStateGenSequences(
         ZSTDMT_serialGenerateLdmFn generateLdm,
         ZSTDMT_serialUpdateChecksumFn updateChecksum,
         ZSTDMT_serialAdvanceFn advance);
+typedef void (*ZSTDMT_waitForLdmLockFn)(void* opaque);
+typedef int (*ZSTDMT_waitForLdmOverlapFn)(
+        void* opaque, void* bufferStart, size_t bufferCapacity);
+typedef void (*ZSTDMT_waitForLdmWaitFn)(void* opaque);
+typedef void (*ZSTDMT_waitForLdmUnlockFn)(void* opaque);
+typedef struct {
+    void* callbackContext;
+    void* bufferStart;
+    size_t bufferCapacity;
+    int ldmEnabled;
+    ZSTDMT_waitForLdmLockFn lock;
+    ZSTDMT_waitForLdmOverlapFn overlaps;
+    ZSTDMT_waitForLdmWaitFn wait;
+    ZSTDMT_waitForLdmUnlockFn unlock;
+} ZSTDMT_RustWaitForLdmState;
+typedef char ZSTDMT_rust_wait_for_ldm_state_layout[
+        (offsetof(ZSTDMT_RustWaitForLdmState, callbackContext) == 0
+         && offsetof(ZSTDMT_RustWaitForLdmState, bufferStart) == sizeof(void*)
+         && offsetof(ZSTDMT_RustWaitForLdmState, bufferCapacity) == 2 * sizeof(void*)
+         && offsetof(ZSTDMT_RustWaitForLdmState, ldmEnabled) == 3 * sizeof(void*)
+         && offsetof(ZSTDMT_RustWaitForLdmState, lock) == 4 * sizeof(void*)
+         && offsetof(ZSTDMT_RustWaitForLdmState, overlaps) == 5 * sizeof(void*)
+         && offsetof(ZSTDMT_RustWaitForLdmState, wait) == 6 * sizeof(void*)
+         && offsetof(ZSTDMT_RustWaitForLdmState, unlock) == 7 * sizeof(void*)
+         && sizeof(ZSTDMT_RustWaitForLdmState) == 8 * sizeof(void*)) ? 1 : -1];
+void ZSTDMT_rust_waitForLdmComplete(
+        const ZSTDMT_RustWaitForLdmState* state);
 
 unsigned ZSTDMT_rust_computeTargetJobLog(unsigned windowLog, unsigned chainLog,
                                          int strategy, int enableLdm);
@@ -2102,22 +2129,54 @@ static int ZSTDMT_doesOverlapWindow(Buffer buffer, ZSTD_window_t window)
                                          window.lowLimit);
 }
 
+static void ZSTDMT_waitForLdmLock(void* opaque)
+{
+    ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
+    ZSTD_PTHREAD_MUTEX_LOCK(&mtctx->serial.ldmWindowMutex);
+}
+
+static int ZSTDMT_waitForLdmOverlap(
+        void* opaque, void* bufferStart, size_t bufferCapacity)
+{
+    ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
+    Buffer const buffer = { bufferStart, bufferCapacity };
+    return ZSTDMT_doesOverlapWindow(buffer, mtctx->serial.ldmWindow);
+}
+
+static void ZSTDMT_waitForLdmWait(void* opaque)
+{
+    ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
+    DEBUGLOG(5, "Waiting for LDM to finish...");
+    ZSTD_pthread_cond_wait(&mtctx->serial.ldmWindowCond,
+                           &mtctx->serial.ldmWindowMutex);
+}
+
+static void ZSTDMT_waitForLdmUnlock(void* opaque)
+{
+    ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
+    DEBUGLOG(6, "Done waiting for LDM to finish");
+    ZSTD_pthread_mutex_unlock(&mtctx->serial.ldmWindowMutex);
+}
+
 static void ZSTDMT_waitForLdmComplete(ZSTDMT_CCtx* mtctx, Buffer buffer)
 {
-    if (mtctx->params.ldmParams.enableLdm == ZSTD_ps_enable) {
+    int const ldmEnabled = mtctx->params.ldmParams.enableLdm == ZSTD_ps_enable;
-        ZSTD_pthread_mutex_t* mutex = &mtctx->serial.ldmWindowMutex;
+    ZSTDMT_RustWaitForLdmState state;
+    if (ldmEnabled) {
         DEBUGLOG(5, "ZSTDMT_waitForLdmComplete");
         DEBUGLOG(5, "source  [0x%zx, 0x%zx)",
                     (size_t)buffer.start,
                     (size_t)buffer.start + buffer.capacity);
-        ZSTD_PTHREAD_MUTEX_LOCK(mutex);
-        while (ZSTDMT_doesOverlapWindow(buffer, mtctx->serial.ldmWindow)) {
-            DEBUGLOG(5, "Waiting for LDM to finish...");
-            ZSTD_pthread_cond_wait(&mtctx->serial.ldmWindowCond, mutex);
-        }
-        DEBUGLOG(6, "Done waiting for LDM to finish");
-        ZSTD_pthread_mutex_unlock(mutex);
     }
+    state.callbackContext = mtctx;
+    state.bufferStart = buffer.start;
+    state.bufferCapacity = buffer.capacity;
+    state.ldmEnabled = ldmEnabled;
+    state.lock = ZSTDMT_waitForLdmLock;
+    state.overlaps = ZSTDMT_waitForLdmOverlap;
+    state.wait = ZSTDMT_waitForLdmWait;
+    state.unlock = ZSTDMT_waitForLdmUnlock;
+    ZSTDMT_rust_waitForLdmComplete(&state);
 }
 
 /**

rust/src/zstdmt_compress.rs

@@ -14,7 +14,7 @@
 //! are Rust-owned. C retains ownership of private descriptor fields and
 //! platform synchronization.
 
-use std::mem::{self, MaybeUninit};
+use std::mem::{self, offset_of, size_of, MaybeUninit};
 use std::os::raw::{c_int, c_uint, c_void};
 use std::ptr;
 use std::sync::Mutex;
@@ -90,6 +90,74 @@ pub type ZSTDMT_serialGenerateLdmFn =
 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);
 
+type ZSTDMT_waitForLdmLockFn = unsafe extern "C" fn(*mut c_void);
+type ZSTDMT_waitForLdmOverlapFn = unsafe extern "C" fn(*mut c_void, *mut c_void, usize) -> c_int;
+type ZSTDMT_waitForLdmWaitFn = unsafe extern "C" fn(*mut c_void);
+type ZSTDMT_waitForLdmUnlockFn = unsafe extern "C" fn(*mut c_void);
+
+/// Projection for the MT wait that protects reusable input against LDM use.
+///
+/// Rust owns the enabled branch and lock/overlap/wait ordering. C retains the
+/// pthread mutex, condition variable, and private LDM window state.
+#[repr(C)]
+pub struct ZSTDMT_RustWaitForLdmState {
+    callback_context: *mut c_void,
+    buffer_start: *mut c_void,
+    buffer_capacity: usize,
+    ldm_enabled: c_int,
+    lock: Option<ZSTDMT_waitForLdmLockFn>,
+    overlaps: Option<ZSTDMT_waitForLdmOverlapFn>,
+    wait: Option<ZSTDMT_waitForLdmWaitFn>,
+    unlock: Option<ZSTDMT_waitForLdmUnlockFn>,
+}
+
+const _: () = {
+    assert!(size_of::<ZSTDMT_waitForLdmLockFn>() == size_of::<usize>());
+    assert!(size_of::<ZSTDMT_waitForLdmOverlapFn>() == size_of::<usize>());
+    assert!(size_of::<ZSTDMT_waitForLdmWaitFn>() == size_of::<usize>());
+    assert!(size_of::<ZSTDMT_waitForLdmUnlockFn>() == size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustWaitForLdmState, callback_context) == 0);
+    assert!(offset_of!(ZSTDMT_RustWaitForLdmState, buffer_start) == size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustWaitForLdmState, buffer_capacity) == 2 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustWaitForLdmState, ldm_enabled) == 3 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustWaitForLdmState, lock) == 4 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustWaitForLdmState, overlaps) == 5 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustWaitForLdmState, wait) == 6 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_RustWaitForLdmState, unlock) == 7 * size_of::<usize>());
+    assert!(size_of::<ZSTDMT_RustWaitForLdmState>() == size_of::<[usize; 8]>());
+};
+
+/// Wait until the reusable input range no longer overlaps the serial LDM
+/// window, preserving the C synchronization state behind callbacks.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_waitForLdmComplete(state: *const ZSTDMT_RustWaitForLdmState) {
+    if state.is_null() {
+        return;
+    }
+    let state = unsafe { &*state };
+    if state.ldm_enabled == 0 || state.callback_context.is_null() {
+        return;
+    }
+    let (Some(lock), Some(overlaps), Some(wait), Some(unlock)) =
+        (state.lock, state.overlaps, state.wait, state.unlock)
+    else {
+        return;
+    };
+
+    unsafe {
+        lock(state.callback_context);
+        while overlaps(
+            state.callback_context,
+            state.buffer_start,
+            state.buffer_capacity,
+        ) != 0
+        {
+            wait(state.callback_context);
+        }
+        unlock(state.callback_context);
+    }
+}
+
 /// 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
@@ -3171,6 +3239,101 @@ mod tests {
         finished: Vec<usize>,
     }
 
+    struct WaitForLdmTestContext {
+        events: Vec<&'static str>,
+        overlaps: Vec<bool>,
+        next_overlap: usize,
+    }
+
+    unsafe extern "C" fn wait_for_ldm_test_lock(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<WaitForLdmTestContext>())
+                .events
+                .push("lock")
+        };
+    }
+
+    unsafe extern "C" fn wait_for_ldm_test_overlaps(
+        context: *mut c_void,
+        _buffer_start: *mut c_void,
+        _buffer_capacity: usize,
+    ) -> c_int {
+        let context = unsafe { &mut *context.cast::<WaitForLdmTestContext>() };
+        context.events.push("overlap");
+        let overlaps = context
+            .overlaps
+            .get(context.next_overlap)
+            .copied()
+            .unwrap_or(false);
+        context.next_overlap += 1;
+        c_int::from(overlaps)
+    }
+
+    unsafe extern "C" fn wait_for_ldm_test_wait(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<WaitForLdmTestContext>())
+                .events
+                .push("wait")
+        };
+    }
+
+    unsafe extern "C" fn wait_for_ldm_test_unlock(context: *mut c_void) {
+        unsafe {
+            (*context.cast::<WaitForLdmTestContext>())
+                .events
+                .push("unlock")
+        };
+    }
+
+    fn wait_for_ldm_test_state(
+        context: &mut WaitForLdmTestContext,
+        ldm_enabled: c_int,
+    ) -> ZSTDMT_RustWaitForLdmState {
+        ZSTDMT_RustWaitForLdmState {
+            callback_context: context as *mut _ as *mut c_void,
+            buffer_start: ptr::null_mut(),
+            buffer_capacity: 8,
+            ldm_enabled,
+            lock: Some(wait_for_ldm_test_lock),
+            overlaps: Some(wait_for_ldm_test_overlaps),
+            wait: Some(wait_for_ldm_test_wait),
+            unlock: Some(wait_for_ldm_test_unlock),
+        }
+    }
+
+    #[test]
+    fn wait_for_ldm_runs_overlap_wait_loop_before_unlocking() {
+        let mut context = WaitForLdmTestContext {
+            events: Vec::new(),
+            overlaps: vec![true, false],
+            next_overlap: 0,
+        };
+        let state = wait_for_ldm_test_state(&mut context, ZSTD_PS_ENABLE);
+
+        unsafe { ZSTDMT_rust_waitForLdmComplete(&state) };
+
+        assert_eq!(
+            context.events,
+            vec!["lock", "overlap", "wait", "overlap", "unlock"]
+        );
+        assert_eq!(context.next_overlap, 2);
+    }
+
+    #[test]
+    fn wait_for_ldm_skips_callbacks_when_ldm_is_disabled() {
+        let mut context = WaitForLdmTestContext {
+            events: Vec::new(),
+            overlaps: vec![true],
+            next_overlap: 0,
+        };
+        let state = wait_for_ldm_test_state(&mut context, 0);
+
+        unsafe { ZSTDMT_rust_waitForLdmComplete(&state) };
+
+        assert!(context.events.is_empty());
+        assert_eq!(context.next_overlap, 0);
+    }
+
     fn record_compression_job_event(state: &Rc<RefCell<MockCompressionJob>>, event: &'static str) {
         state.borrow_mut().events.push(event);
     }