diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 6018a1c5e..1a280b001 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -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) diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 68db326ca..cdf9b76a8 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -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::()); + assert!(offset_of!(ZSTDMT_frameChecksumProjection, cSize) == 2 * size_of::()); + assert!(offset_of!(ZSTDMT_frameChecksumProjection, checksum) == 3 * size_of::()); + assert!(size_of::() == 4); + assert!(size_of::() == 4 * size_of::()); +}; + +#[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::().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 {