feat(mt): move compression job creation policy into Rust

Move the multithreaded compression job-creation decision tree into a Rust
projection while keeping C-owned job descriptors, input buffers, pools,
synchronization, worker callbacks, and terminal empty-block serialization
behind explicit callbacks. The Rust policy now preserves ring-table full
checks, prepared-job retry state, prefix advancement, frame checksum rules,
and the non-first empty terminal job path without crossing private C layouts.

Test Plan:
- cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1
- cargo test --manifest-path rust/Cargo.toml --no-default-features --features compression,decompression,dict-builder,legacy-v01,legacy-v02,legacy-v03,legacy-v04,legacy-v05,legacy-v06,legacy-v07 --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 tests -j2 test-zstd test-pool test-legacy test-invalidDictionaries

rust/src/zstdmt_compress.rs

@@ -240,6 +240,207 @@ pub unsafe extern "C" fn ZSTDMT_rust_writeLastEmptyBlock(
     unsafe { write_last_empty_block_with(projection, || get_buffer(opaque)) }
 }
 
+const CREATE_JOB_TABLE_FULL: c_uint = 0;
+const CREATE_JOB_POST: c_uint = 1;
+const CREATE_JOB_EMPTY: c_uint = 2;
+
+/// Scalar MT state supplied by C for one compression-job creation attempt.
+///
+/// The job descriptor, input-buffer ownership, and synchronization objects
+/// remain private to C. Rust owns the ring-capacity and scheduling policy over
+/// this scalar view.
+#[repr(C)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ZSTDMT_createJobProjection {
+    pub doneJobID: c_uint,
+    pub nextJobID: c_uint,
+    pub jobIDMask: c_uint,
+    pub jobReady: c_uint,
+    pub srcStart: *const c_void,
+    pub srcSize: usize,
+    pub inBuffFilled: usize,
+    pub prefixStart: *const c_void,
+    pub prefixSize: usize,
+    pub targetPrefixSize: usize,
+    pub endFrame: c_uint,
+    pub checksumFlag: c_uint,
+}
+
+/// The scalar fields needed by C to initialize its private job descriptor and
+/// the surrounding input state after Rust chooses to prepare a new job.
+#[repr(C)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ZSTDMT_jobInitialization {
+    pub srcStart: *const c_void,
+    pub srcSize: usize,
+    pub prefixStart: *const c_void,
+    pub prefixSize: usize,
+    pub nextPrefixStart: *const c_void,
+    pub nextPrefixSize: usize,
+    pub roundBuffPosDelta: usize,
+    pub jobNumber: c_uint,
+    pub firstJob: c_uint,
+    pub lastJob: c_uint,
+    pub frameChecksumNeeded: c_uint,
+    pub clearChecksumFlag: c_uint,
+}
+
+#[repr(C)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ZSTDMT_createJobResult {
+    pub returnCode: usize,
+    pub action: c_uint,
+    pub jobID: c_uint,
+    pub jobNumber: c_uint,
+    pub nextJobID: c_uint,
+    pub jobReady: c_uint,
+}
+
+pub type ZSTDMT_prepareJobFn = unsafe extern "C" fn(
+    opaque: *mut c_void,
+    job_id: c_uint,
+    initialization: *const ZSTDMT_jobInitialization,
+);
+pub type ZSTDMT_writeEmptyJobFn = unsafe extern "C" fn(opaque: *mut c_void, job_id: c_uint);
+pub type ZSTDMT_tryAddJobFn = unsafe extern "C" fn(opaque: *mut c_void, job_id: c_uint) -> c_int;
+
+#[inline]
+fn create_job_table_full(projection: ZSTDMT_createJobProjection) -> ZSTDMT_createJobResult {
+    ZSTDMT_createJobResult {
+        action: CREATE_JOB_TABLE_FULL,
+        jobID: projection.nextJobID & projection.jobIDMask,
+        jobNumber: projection.nextJobID,
+        nextJobID: projection.nextJobID,
+        jobReady: projection.jobReady,
+        ..ZSTDMT_createJobResult::default()
+    }
+}
+
+#[inline]
+fn create_job_initialization(
+    projection: ZSTDMT_createJobProjection,
+) -> (c_uint, ZSTDMT_jobInitialization) {
+    let job_number = projection.nextJobID;
+    let first_job = (job_number == 0) as c_uint;
+    let last_job = (projection.endFrame != 0) as c_uint;
+    let frame_checksum_needed =
+        (projection.checksumFlag != 0 && last_job != 0 && job_number > 0) as c_uint;
+
+    let (next_prefix_start, next_prefix_size) = if last_job != 0 {
+        (ptr::null(), 0)
+    } else {
+        let next_prefix_size = projection.srcSize.min(projection.targetPrefixSize);
+        let next_prefix_start = projection
+            .srcStart
+            .cast::<u8>()
+            .wrapping_add(projection.srcSize - next_prefix_size)
+            .cast();
+        (next_prefix_start, next_prefix_size)
+    };
+
+    (
+        projection.nextJobID & projection.jobIDMask,
+        ZSTDMT_jobInitialization {
+            srcStart: projection.srcStart,
+            srcSize: projection.srcSize,
+            prefixStart: projection.prefixStart,
+            prefixSize: projection.prefixSize,
+            nextPrefixStart: next_prefix_start,
+            nextPrefixSize: next_prefix_size,
+            roundBuffPosDelta: projection.srcSize,
+            jobNumber: job_number,
+            firstJob: first_job,
+            lastJob: last_job,
+            frameChecksumNeeded: frame_checksum_needed,
+            clearChecksumFlag: (last_job != 0 && first_job != 0) as c_uint,
+        },
+    )
+}
+
+#[inline]
+fn create_compression_job_with<P, E, T>(
+    projection: ZSTDMT_createJobProjection,
+    mut prepare_job: P,
+    mut write_empty_job: E,
+    mut try_add_job: T,
+) -> ZSTDMT_createJobResult
+where
+    P: FnMut(c_uint, &ZSTDMT_jobInitialization),
+    E: FnMut(c_uint),
+    T: FnMut(c_uint) -> bool,
+{
+    /* Match the original unsigned C comparison, including wraparound. */
+    if projection.nextJobID > projection.doneJobID.wrapping_add(projection.jobIDMask) {
+        return create_job_table_full(projection);
+    }
+
+    let job_id = projection.nextJobID & projection.jobIDMask;
+    let mut result = ZSTDMT_createJobResult {
+        action: CREATE_JOB_POST,
+        jobID: job_id,
+        jobNumber: projection.nextJobID,
+        nextJobID: projection.nextJobID,
+        jobReady: projection.jobReady,
+        ..ZSTDMT_createJobResult::default()
+    };
+
+    if projection.jobReady == 0 {
+        debug_assert!(projection.inBuffFilled >= projection.srcSize);
+        let (_, initialization) = create_job_initialization(projection);
+        prepare_job(job_id, &initialization);
+
+        /* A non-first empty job is represented by the terminal empty block,
+         * not by a worker submission. */
+        if projection.srcSize == 0 && projection.nextJobID > 0 {
+            write_empty_job(job_id);
+            result.action = CREATE_JOB_EMPTY;
+            result.nextJobID = projection.nextJobID.wrapping_add(1);
+            result.jobReady = 0;
+            return result;
+        }
+    }
+
+    if try_add_job(job_id) {
+        result.nextJobID = projection.nextJobID.wrapping_add(1);
+        result.jobReady = 0;
+    } else {
+        result.jobReady = 1;
+    }
+    result
+}
+
+/// Apply MT scalar scheduling policy while C retains all private descriptor,
+/// pool, mutex, condition-variable, and worker-callback operations.
+#[cfg(not(test))]
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_createCompressionJob(
+    projection: *const ZSTDMT_createJobProjection,
+    opaque: *mut c_void,
+    prepareJob: Option<ZSTDMT_prepareJobFn>,
+    writeEmptyJob: Option<ZSTDMT_writeEmptyJobFn>,
+    tryAddJob: Option<ZSTDMT_tryAddJobFn>,
+) -> ZSTDMT_createJobResult {
+    let Some(projection) = (unsafe { projection.as_ref() }).copied() else {
+        return ZSTDMT_createJobResult::default();
+    };
+    let (Some(prepare_job), Some(write_empty_job), Some(try_add_job)) =
+        (prepareJob, writeEmptyJob, tryAddJob)
+    else {
+        return ZSTDMT_createJobResult::default();
+    };
+
+    create_compression_job_with(
+        projection,
+        |job_id, initialization| unsafe {
+            prepare_job(opaque, job_id, initialization);
+        },
+        |job_id| unsafe {
+            write_empty_job(opaque, job_id);
+        },
+        |job_id| unsafe { try_add_job(opaque, job_id) != 0 },
+    )
+}
+
 #[inline]
 fn invalid_flush_publication(
     output_pos: usize,
@@ -2155,6 +2356,162 @@ mod tests {
         assert_eq!(output, original);
     }
 
+    #[test]
+    fn create_job_rejects_full_table_without_callbacks() {
+        let projection = ZSTDMT_createJobProjection {
+            doneJobID: 0,
+            nextJobID: 8,
+            jobIDMask: 7,
+            ..ZSTDMT_createJobProjection::default()
+        };
+        let result = create_compression_job_with(
+            projection,
+            |_job_id, _initialization| panic!("table-full job must not be prepared"),
+            |_job_id| panic!("table-full job must not be written"),
+            |_job_id| panic!("table-full job must not be posted"),
+        );
+
+        assert_eq!(result.action, CREATE_JOB_TABLE_FULL);
+        assert_eq!(result.jobID, 0);
+        assert_eq!(result.jobNumber, 8);
+        assert_eq!(result.nextJobID, 8);
+        assert_eq!(result.jobReady, 0);
+    }
+
+    #[test]
+    fn create_job_projects_ordinary_initialization_and_prefix_advance() {
+        let source = [0u8; 8];
+        let prefix = [1u8; 3];
+        let projection = ZSTDMT_createJobProjection {
+            doneJobID: 0,
+            nextJobID: 3,
+            jobIDMask: 7,
+            srcStart: source.as_ptr().cast(),
+            srcSize: source.len(),
+            inBuffFilled: source.len(),
+            prefixStart: prefix.as_ptr().cast(),
+            prefixSize: prefix.len(),
+            targetPrefixSize: 4,
+            checksumFlag: 1,
+            ..ZSTDMT_createJobProjection::default()
+        };
+        let mut initialization = None;
+        let mut posted = None;
+        let result = create_compression_job_with(
+            projection,
+            |job_id, value| {
+                assert_eq!(job_id, 3);
+                initialization = Some(*value);
+            },
+            |_job_id| panic!("ordinary job must not use the empty-block path"),
+            |job_id| {
+                posted = Some(job_id);
+                true
+            },
+        );
+
+        let initialization = initialization.expect("job should be initialized");
+        assert_eq!(initialization.srcStart, source.as_ptr().cast());
+        assert_eq!(initialization.srcSize, source.len());
+        assert_eq!(initialization.prefixStart, prefix.as_ptr().cast());
+        assert_eq!(initialization.prefixSize, prefix.len());
+        assert_eq!(
+            initialization.nextPrefixStart,
+            source.as_ptr().wrapping_add(4).cast()
+        );
+        assert_eq!(initialization.nextPrefixSize, 4);
+        assert_eq!(initialization.roundBuffPosDelta, source.len());
+        assert_eq!(initialization.jobNumber, 3);
+        assert_eq!(initialization.firstJob, 0);
+        assert_eq!(initialization.lastJob, 0);
+        assert_eq!(initialization.frameChecksumNeeded, 0);
+        assert_eq!(posted, Some(3));
+        assert_eq!(result.action, CREATE_JOB_POST);
+        assert_eq!(result.jobID, 3);
+        assert_eq!(result.nextJobID, 4);
+        assert_eq!(result.jobReady, 0);
+    }
+
+    #[test]
+    fn create_job_projects_terminal_empty_frame() {
+        let prefix = [2u8; 2];
+        let projection = ZSTDMT_createJobProjection {
+            doneJobID: 0,
+            nextJobID: 2,
+            jobIDMask: 7,
+            inBuffFilled: 0,
+            prefixStart: prefix.as_ptr().cast(),
+            prefixSize: prefix.len(),
+            endFrame: 1,
+            checksumFlag: 1,
+            ..ZSTDMT_createJobProjection::default()
+        };
+        let mut initialization = None;
+        let mut empty_job = None;
+        let result = create_compression_job_with(
+            projection,
+            |job_id, value| {
+                assert_eq!(job_id, 2);
+                initialization = Some(*value);
+            },
+            |job_id| empty_job = Some(job_id),
+            |_job_id| panic!("terminal empty job must not be posted"),
+        );
+
+        let initialization = initialization.expect("empty job should be initialized");
+        assert_eq!(initialization.srcSize, 0);
+        assert_eq!(initialization.prefixStart, prefix.as_ptr().cast());
+        assert_eq!(initialization.prefixSize, prefix.len());
+        assert!(initialization.nextPrefixStart.is_null());
+        assert_eq!(initialization.nextPrefixSize, 0);
+        assert_eq!(initialization.roundBuffPosDelta, 0);
+        assert_eq!(initialization.jobNumber, 2);
+        assert_eq!(initialization.firstJob, 0);
+        assert_eq!(initialization.lastJob, 1);
+        assert_eq!(initialization.frameChecksumNeeded, 1);
+        assert_eq!(initialization.clearChecksumFlag, 0);
+        assert_eq!(empty_job, Some(2));
+        assert_eq!(result.action, CREATE_JOB_EMPTY);
+        assert_eq!(result.nextJobID, 3);
+        assert_eq!(result.jobReady, 0);
+    }
+
+    #[test]
+    fn create_job_preserves_pool_success_and_failure_state() {
+        let projection = ZSTDMT_createJobProjection {
+            doneJobID: 0,
+            nextJobID: 4,
+            jobIDMask: 7,
+            jobReady: 1,
+            ..ZSTDMT_createJobProjection::default()
+        };
+        let success = create_compression_job_with(
+            projection,
+            |_job_id, _initialization| panic!("prepared job must not be reinitialized"),
+            |_job_id| panic!("prepared job must not use the empty-block path"),
+            |job_id| {
+                assert_eq!(job_id, 4);
+                true
+            },
+        );
+        assert_eq!(success.action, CREATE_JOB_POST);
+        assert_eq!(success.nextJobID, 5);
+        assert_eq!(success.jobReady, 0);
+
+        let failure = create_compression_job_with(
+            projection,
+            |_job_id, _initialization| panic!("prepared job must not be reinitialized"),
+            |_job_id| panic!("prepared job must not use the empty-block path"),
+            |job_id| {
+                assert_eq!(job_id, 4);
+                false
+            },
+        );
+        assert_eq!(failure.action, CREATE_JOB_POST);
+        assert_eq!(failure.nextJobID, 4);
+        assert_eq!(failure.jobReady, 1);
+    }
+
     #[test]
     fn flush_state_machine_preserves_offsets_and_completes_job() {
         let job = [0x60u8, 0x61, 0x62, 0x63, 0x64, 0x65];