refactor(compress): move MT checksum serialization to Rust

Keep XXH64 state ownership and checksum calculation in the C MT scheduler, but project the final four-byte frame write through Rust. Rust now owns the little-endian serialization and destination-capacity policy, with ABI layout assertions and flush-state error propagation preserving the existing cleanup path. Focused tests cover byte order, exact capacity, invalid destinations, and callback error handling. Test Plan: - `ulimit -v 41943040; cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check` - `ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings` - `ulimit -v 41943040; make -j1` - `ulimit -v 41943040; make -j1 -C tests test` - `ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings` - `ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/cli/Cargo.toml --all-targets` The integrated checks ran under a serial 40 GiB virtual-memory limit. The full upstream suite passed, including large-file, multithreaded, API, fuzzer, and zstream coverage. Standalone root Rust unit linking remains unavailable because the crate imports C-owned bridge symbols without a Cargo build/link setup.
2026-07-20 09:29:17 +02:00
parent b550b14ceb
commit dd87b328a4
2 changed files with 190 additions and 5 deletions
@@ -285,6 +285,24 @@ typedef struct {
    size_t dstFlushed;
    unsigned frameChecksumNeeded;
 } ZSTDMT_RustFlushJobProjection;
+typedef struct {
+    void* dst;
+    size_t dstCapacity;
+    size_t cSize;
+    U32 checksum;
+} ZSTDMT_RustFrameChecksumProjection;
+typedef char ZSTDMT_frame_checksum_projection_layout[
+        (offsetof(ZSTDMT_RustFrameChecksumProjection, dst) == 0
+         && offsetof(ZSTDMT_RustFrameChecksumProjection, dstCapacity)
+                == sizeof(void*)
+         && offsetof(ZSTDMT_RustFrameChecksumProjection, cSize)
+                == 2 * sizeof(void*)
+         && offsetof(ZSTDMT_RustFrameChecksumProjection, checksum)
+                == 3 * sizeof(void*)
+         && sizeof(U32) == 4
+         && sizeof(ZSTDMT_RustFrameChecksumProjection)
+                == 4 * sizeof(void*))
+        ? 1 : -1];
 typedef struct {
    unsigned doneJobID;
    unsigned nextJobID;
@@ -319,6 +337,8 @@ ZSTDMT_RustFlushProducedResult ZSTDMT_rust_flushProduced(
        ZSTDMT_flushUpdateJobFn updateJob,
        ZSTDMT_flushCompleteJobFn completeJob,
        ZSTDMT_flushErrorFn onError);
+size_t ZSTDMT_rust_writeFrameChecksum(
+        const ZSTDMT_RustFrameChecksumProjection* projection);

 /* The Rust outer scheduler sees only this scalar snapshot.  The MT context,
 * reusable input buffer, worker pool, and all synchronization remain private
@@ -2786,15 +2806,23 @@ static void ZSTDMT_addFrameChecksum(void* opaque, unsigned jobID,
    ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
    ZSTDMT_jobDescription* const job = &mtctx->jobs[jobID];
    U32 const checksum = (U32)XXH64_digest(&mtctx->serial.xxhState);
+    ZSTDMT_RustFrameChecksumProjection const checksumProjection = {
+        job->dstBuff.start,
+        job->dstBuff.capacity,
+        job->cSize,
+        checksum
+    };
+    size_t const cSize = ZSTDMT_rust_writeFrameChecksum(&checksumProjection);

    assert(projection->frameChecksumNeeded);
    assert(projection->consumed == projection->srcSize);
    assert(projection->cSize == job->cSize);
-    MEM_writeLE32((char*)job->dstBuff.start + job->cSize, checksum);
-    job->cSize += 4;
-    job->frameChecksumNeeded = 0;
-    projection->cSize = job->cSize;
-    projection->frameChecksumNeeded = 0;
+    job->cSize = cSize;
+    projection->cSize = cSize;
+    if (!ZSTD_isError(cSize)) {
+        job->frameChecksumNeeded = 0;
+        projection->frameChecksumNeeded = 0;
+    }
 }

 static void ZSTDMT_updateFlushJob(void* opaque, unsigned jobID, size_t dstFlushed)
@@ -21,6 +21,7 @@ use std::sync::Mutex;

 use crate::bits::ZSTD_highbit32;
 use crate::errors::{ERR_isError, ZstdErrorCode, ERROR};
+use crate::mem::MEM_writeLE32;
 use crate::zstd_compress::{ZSTD_frameProgression, ZSTD_rust_invalidateRepCodes};
 use crate::zstd_compress_frame::ZSTD_rust_writeFrameHeader;

@@ -883,6 +884,55 @@ pub struct ZSTDMT_flushJobProjection {
    pub frameChecksumNeeded: c_uint,
 }

+/// Narrow projection for serializing the C-owned frame checksum.  C retains
+/// the serial XXH64 state and computes the checksum; Rust owns only the
+/// deterministic little-endian write and destination policy.
+#[repr(C)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ZSTDMT_frameChecksumProjection {
+    pub dst: *mut c_void,
+    pub dstCapacity: usize,
+    pub cSize: usize,
+    pub checksum: c_uint,
+}
+
+const _: () = {
+    assert!(offset_of!(ZSTDMT_frameChecksumProjection, dst) == 0);
+    assert!(offset_of!(ZSTDMT_frameChecksumProjection, dstCapacity) == size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_frameChecksumProjection, cSize) == 2 * size_of::<usize>());
+    assert!(offset_of!(ZSTDMT_frameChecksumProjection, checksum) == 3 * size_of::<usize>());
+    assert!(size_of::<c_uint>() == 4);
+    assert!(size_of::<ZSTDMT_frameChecksumProjection>() == 4 * size_of::<usize>());
+};
+
+#[inline]
+fn write_frame_checksum(projection: &ZSTDMT_frameChecksumProjection) -> usize {
+    if projection.dst.is_null()
+        || projection.dstCapacity < 4
+        || projection.cSize > projection.dstCapacity - 4
+    {
+        return ERROR(ZstdErrorCode::DstSizeTooSmall);
+    }
+
+    unsafe {
+        MEM_writeLE32(
+            projection.dst.cast::<u8>().add(projection.cSize).cast(),
+            projection.checksum,
+        );
+    }
+    projection.cSize + 4
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_writeFrameChecksum(
+    projection: *const ZSTDMT_frameChecksumProjection,
+) -> usize {
+    let Some(projection) = (unsafe { projection.as_ref() }) else {
+        return ERROR(ZstdErrorCode::Generic);
+    };
+    write_frame_checksum(projection)
+}
+
 /// Scalar view of the C-owned MT state needed after the current job is
 /// published.  The C adapter applies descriptor and context mutations made by
 /// Rust's decisions through callbacks.
@@ -2207,6 +2257,11 @@ where
            add_frame_checksum(job_id, &mut projection);
        }

+        if ERR_isError(projection.cSize) {
+            on_error();
+            return flush_produced_result(projection.cSize, output_pos);
+        }
+
        let c_size = projection.cSize;
        let mut dst_flushed = projection.dstFlushed;
        if c_size > 0 {
@@ -6506,6 +6561,108 @@ mod tests {
        assert_eq!(&output[..7], &job);
    }

+    #[test]
+    fn flush_state_machine_propagates_checksum_projection_error() {
+        let job = [0x80u8, 0x81, 0x82, 0x83];
+        let mut output = [0xa5u8; 8];
+        let projection = ZSTDMT_flushJobProjection {
+            consumed: job.len(),
+            cSize: 3,
+            srcSize: job.len(),
+            dstStart: job.as_ptr().cast(),
+            dstCapacity: job.len(),
+            frameChecksumNeeded: 1,
+            ..ZSTDMT_flushJobProjection::default()
+        };
+        let mut error_cleanup = false;
+        let mut completed = false;
+
+        let result = unsafe {
+            flush_produced_with(
+                ZSTDMT_flushContextProjection {
+                    doneJobID: 0,
+                    nextJobID: 1,
+                    jobIDMask: 0,
+                    ..ZSTDMT_flushContextProjection::default()
+                },
+                output.as_mut_ptr().cast(),
+                output.len(),
+                0,
+                0,
+                1,
+                |_job_id, _block_to_flush| projection,
+                |_job_id, projected| {
+                    projected.cSize = ERROR(ZstdErrorCode::DstSizeTooSmall);
+                },
+                |_job_id, _dst_flushed| panic!("unexpected update callback"),
+                |_job_id, _src_size, _c_size| completed = true,
+                || error_cleanup = true,
+            )
+        };
+
+        assert_eq!(result.result, ERROR(ZstdErrorCode::DstSizeTooSmall));
+        assert_eq!(result.outputPos, 0);
+        assert!(error_cleanup);
+        assert!(!completed);
+        assert_eq!(output, [0xa5; 8]);
+    }
+
+    #[test]
+    fn frame_checksum_is_serialized_little_endian() {
+        let mut dst = [0xa5u8; 8];
+        let projection = ZSTDMT_frameChecksumProjection {
+            dst: dst.as_mut_ptr().cast(),
+            dstCapacity: dst.len(),
+            cSize: 4,
+            checksum: 0x7856_3412,
+        };
+
+        assert_eq!(write_frame_checksum(&projection), 8);
+        assert_eq!(&dst[..4], &[0xa5; 4]);
+        assert_eq!(&dst[4..], &[0x12, 0x34, 0x56, 0x78]);
+    }
+
+    #[test]
+    fn frame_checksum_accepts_exact_destination_capacity() {
+        let mut dst = [0xa5u8; 6];
+        let projection = ZSTDMT_frameChecksumProjection {
+            dst: dst.as_mut_ptr().cast(),
+            dstCapacity: dst.len(),
+            cSize: 2,
+            checksum: 0x1122_3344,
+        };
+
+        assert_eq!(write_frame_checksum(&projection), dst.len());
+        assert_eq!(&dst, &[0xa5, 0xa5, 0x44, 0x33, 0x22, 0x11]);
+    }
+
+    #[test]
+    fn frame_checksum_rejects_too_small_or_null_destination() {
+        let mut dst = [0xa5u8; 8];
+        let too_small = ZSTDMT_frameChecksumProjection {
+            dst: dst.as_mut_ptr().cast(),
+            dstCapacity: 7,
+            cSize: 4,
+            checksum: 0x0102_0304,
+        };
+        let null_destination = ZSTDMT_frameChecksumProjection {
+            dst: ptr::null_mut(),
+            dstCapacity: 4,
+            cSize: 0,
+            checksum: 0x0102_0304,
+        };
+
+        assert_eq!(
+            write_frame_checksum(&too_small),
+            ERROR(ZstdErrorCode::DstSizeTooSmall)
+        );
+        assert_eq!(
+            write_frame_checksum(&null_destination),
+            ERROR(ZstdErrorCode::DstSizeTooSmall)
+        );
+        assert_eq!(dst, [0xa5; 8]);
+    }
+
    #[test]
    fn flush_state_machine_reports_end_of_frame_state() {
        let not_ended = unsafe {