diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 69ffc7a2f..369217aaf 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -547,6 +547,12 @@ typedef void (*ZSTDMT_prepareJobFn)(void* opaque, unsigned jobID, const ZSTDMT_RustJobInitialization* init); typedef void (*ZSTDMT_writeEmptyJobFn)(void* opaque, unsigned jobID); typedef int (*ZSTDMT_tryAddJobFn)(void* opaque, unsigned jobID); +void ZSTDMT_rust_prepareCompressionJob( + void* opaque, unsigned jobID, + const ZSTDMT_RustJobInitialization* initialization, + ZSTDMT_prepareJobFn initializeJob, + ZSTDMT_prepareJobFn resetInputState, + ZSTDMT_prepareJobFn publishFrameState); ZSTDMT_RustCreateJobResult ZSTDMT_rust_createCompressionJob( const ZSTDMT_RustCreateJobProjection* projection, void* opaque, ZSTDMT_prepareJobFn prepareJob, @@ -2877,7 +2883,7 @@ static void ZSTDMT_writeLastEmptyBlock(ZSTDMT_jobDescription* job) assert(!ZSTD_isError(job->cSize)); } -static void ZSTDMT_prepareCompressionJob( +static void ZSTDMT_prepareCompressionJobDescriptor( void* opaque, unsigned jobID, const ZSTDMT_RustJobInitialization* initialization) { @@ -2908,7 +2914,14 @@ static void ZSTDMT_prepareCompressionJob( job->lastJob = initialization->lastJob; job->frameChecksumNeeded = initialization->frameChecksumNeeded; job->dstFlushed = 0; +} +static void ZSTDMT_prepareCompressionJobInputState( + void* opaque, unsigned jobID, + const ZSTDMT_RustJobInitialization* initialization) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + (void)jobID; /* Update the round buffer position and clear the input buffer to be reset. */ mtctx->roundBuff.pos += initialization->roundBuffPosDelta; mtctx->inBuff.buffer = g_nullBuffer; @@ -2917,6 +2930,14 @@ static void ZSTDMT_prepareCompressionJob( (const BYTE*)initialization->nextPrefixStart, initialization->nextPrefixSize }; +} + +static void ZSTDMT_prepareCompressionJobFrameState( + void* opaque, unsigned jobID, + const ZSTDMT_RustJobInitialization* initialization) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + (void)jobID; if (initialization->lastJob) { mtctx->frameEnded = 1; if (initialization->clearChecksumFlag) @@ -2924,6 +2945,17 @@ static void ZSTDMT_prepareCompressionJob( } } +static void ZSTDMT_prepareCompressionJob( + void* opaque, unsigned jobID, + const ZSTDMT_RustJobInitialization* initialization) +{ + ZSTDMT_rust_prepareCompressionJob( + opaque, jobID, initialization, + ZSTDMT_prepareCompressionJobDescriptor, + ZSTDMT_prepareCompressionJobInputState, + ZSTDMT_prepareCompressionJobFrameState); +} + static void ZSTDMT_writeEmptyCompressionJob(void* opaque, unsigned jobID) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 296ff19a1..b72f04e93 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -1547,6 +1547,68 @@ pub type ZSTDMT_prepareJobFn = unsafe extern "C" fn( 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; +/// Order the C-owned leaves that publish a newly prepared MT job. +/// +/// The descriptor must become complete before the reusable input range is +/// cleared, and terminal frame state is published last. `lastJob` is a C +/// boolean scalar, so any non-zero value selects the terminal publication. +#[inline] +fn prepare_compression_job_with
(
+ job_id: c_uint,
+ initialization: &ZSTDMT_jobInitialization,
+ mut initialize_job: P,
+ mut reset_input_state: I,
+ mut publish_frame_state: F,
+) where
+ P: FnMut(c_uint, &ZSTDMT_jobInitialization),
+ I: FnMut(c_uint, &ZSTDMT_jobInitialization),
+ F: FnMut(c_uint, &ZSTDMT_jobInitialization),
+{
+ initialize_job(job_id, initialization);
+ reset_input_state(job_id, initialization);
+ if initialization.lastJob != 0 {
+ publish_frame_state(job_id, initialization);
+ }
+}
+
+/// Apply the MT job-preparation ordering while C retains the private job
+/// descriptor and stream-state fields behind callbacks.
+#[no_mangle]
+pub unsafe extern "C" fn ZSTDMT_rust_prepareCompressionJob(
+ opaque: *mut c_void,
+ job_id: c_uint,
+ initialization: *const ZSTDMT_jobInitialization,
+ initialize_job: Option