diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 26f98387c..124c9ed83 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -545,6 +545,7 @@ enum { }; typedef void (*ZSTDMT_prepareJobFn)(void* opaque, unsigned jobID, const ZSTDMT_RustJobInitialization* init); +typedef void (*ZSTDMT_prepareFrameStateFn)(void* opaque); typedef void (*ZSTDMT_writeEmptyJobFn)(void* opaque, unsigned jobID); typedef int (*ZSTDMT_tryAddJobFn)(void* opaque, unsigned jobID); void ZSTDMT_rust_prepareCompressionJob( @@ -552,7 +553,8 @@ void ZSTDMT_rust_prepareCompressionJob( const ZSTDMT_RustJobInitialization* initialization, ZSTDMT_prepareJobFn initializeJob, ZSTDMT_prepareJobFn resetInputState, - ZSTDMT_prepareJobFn publishFrameState); + ZSTDMT_prepareFrameStateFn setFrameEnded, + ZSTDMT_prepareFrameStateFn clearChecksumFlag); ZSTDMT_RustCreateJobResult ZSTDMT_rust_createCompressionJob( const ZSTDMT_RustCreateJobProjection* projection, void* opaque, ZSTDMT_prepareJobFn prepareJob, @@ -2931,17 +2933,16 @@ static void ZSTDMT_prepareCompressionJobInputState( }; } -static void ZSTDMT_prepareCompressionJobFrameState( - void* opaque, unsigned jobID, - const ZSTDMT_RustJobInitialization* initialization) +static void ZSTDMT_prepareCompressionJobSetFrameEnded(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; - (void)jobID; - if (initialization->lastJob) { - mtctx->frameEnded = 1; - if (initialization->clearChecksumFlag) - mtctx->params.fParams.checksumFlag = 0; - } + mtctx->frameEnded = 1; +} + +static void ZSTDMT_prepareCompressionJobClearChecksumFlag(void* opaque) +{ + ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; + mtctx->params.fParams.checksumFlag = 0; } static void ZSTDMT_prepareCompressionJob( @@ -2952,7 +2953,8 @@ static void ZSTDMT_prepareCompressionJob( opaque, jobID, initialization, ZSTDMT_prepareCompressionJobDescriptor, ZSTDMT_prepareCompressionJobInputState, - ZSTDMT_prepareCompressionJobFrameState); + ZSTDMT_prepareCompressionJobSetFrameEnded, + ZSTDMT_prepareCompressionJobClearChecksumFlag); } static void ZSTDMT_writeEmptyCompressionJob(void* opaque, unsigned jobID) diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 6938aacb0..34e21483e 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -1544,30 +1544,65 @@ pub type ZSTDMT_prepareJobFn = unsafe extern "C" fn( job_id: c_uint, initialization: *const ZSTDMT_jobInitialization, ); +pub type ZSTDMT_prepareFrameStateFn = unsafe extern "C" fn(opaque: *mut c_void); 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; +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum CompressionJobFrameStateAction { + None, + SetFrameEnded, + SetFrameEndedAndClearChecksum, +} + +#[inline] +fn classify_compression_job_frame_state( + last_job: c_uint, + clear_checksum_flag: c_uint, +) -> CompressionJobFrameStateAction { + if last_job == 0 { + CompressionJobFrameStateAction::None + } else if clear_checksum_flag == 0 { + CompressionJobFrameStateAction::SetFrameEnded + } else { + CompressionJobFrameStateAction::SetFrameEndedAndClearChecksum + } +} + /// 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. +/// cleared, and terminal frame state is published last. `lastJob` and +/// `clearChecksumFlag` are classified in Rust; C callbacks retain the private +/// context mutations. #[inline] -fn prepare_compression_job_with( +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, + mut set_frame_ended: S, + mut clear_checksum_flag: C, ) where P: FnMut(c_uint, &ZSTDMT_jobInitialization), I: FnMut(c_uint, &ZSTDMT_jobInitialization), - F: FnMut(c_uint, &ZSTDMT_jobInitialization), + S: FnMut(), + C: FnMut(), { + let frame_state_action = classify_compression_job_frame_state( + initialization.lastJob, + initialization.clearChecksumFlag, + ); + initialize_job(job_id, initialization); reset_input_state(job_id, initialization); - if initialization.lastJob != 0 { - publish_frame_state(job_id, initialization); + match frame_state_action { + CompressionJobFrameStateAction::None => {} + CompressionJobFrameStateAction::SetFrameEnded => set_frame_ended(), + CompressionJobFrameStateAction::SetFrameEndedAndClearChecksum => { + set_frame_ended(); + clear_checksum_flag(); + } } } @@ -1580,13 +1615,23 @@ pub unsafe extern "C" fn ZSTDMT_rust_prepareCompressionJob( initialization: *const ZSTDMT_jobInitialization, initialize_job: Option, reset_input_state: Option, - publish_frame_state: Option, + set_frame_ended: Option, + clear_checksum_flag: Option, ) { let Some(initialization) = (unsafe { initialization.as_ref() }) else { return; }; - let (Some(initialize_job), Some(reset_input_state), Some(publish_frame_state)) = - (initialize_job, reset_input_state, publish_frame_state) + let ( + Some(initialize_job), + Some(reset_input_state), + Some(set_frame_ended), + Some(clear_checksum_flag), + ) = ( + initialize_job, + reset_input_state, + set_frame_ended, + clear_checksum_flag, + ) else { return; }; @@ -1603,8 +1648,11 @@ pub unsafe extern "C" fn ZSTDMT_rust_prepareCompressionJob( |job_id, initialization| unsafe { reset_input_state(opaque, job_id, initialization); }, - |job_id, initialization| unsafe { - publish_frame_state(opaque, job_id, initialization); + || unsafe { + set_frame_ended(opaque); + }, + || unsafe { + clear_checksum_flag(opaque); }, ); } @@ -7125,6 +7173,25 @@ mod tests { assert_eq!(result.jobReady, 0); } + #[test] + fn compression_job_frame_state_classifies_all_flag_combinations() { + for (last_job, clear_checksum_flag, expected) in [ + (0, 0, CompressionJobFrameStateAction::None), + (0, 1, CompressionJobFrameStateAction::None), + (1, 0, CompressionJobFrameStateAction::SetFrameEnded), + ( + 1, + 1, + CompressionJobFrameStateAction::SetFrameEndedAndClearChecksum, + ), + ] { + assert_eq!( + classify_compression_job_frame_state(last_job, clear_checksum_flag), + expected + ); + } + } + #[test] fn job_preparation_orders_descriptor_input_and_terminal_state() { let initialization = ZSTDMT_jobInitialization { @@ -7148,14 +7215,14 @@ mod tests { assert_eq!(value.roundBuffPosDelta, 0); events.borrow_mut().push("input"); }, - |job_id, value| { - assert_eq!(job_id, 3); - assert_eq!(value.clearChecksumFlag, 1); - events.borrow_mut().push("frame"); - }, + || events.borrow_mut().push("frame-ended"), + || events.borrow_mut().push("checksum"), ); - assert_eq!(events.into_inner(), vec!["descriptor", "input", "frame"]); + assert_eq!( + events.into_inner(), + vec!["descriptor", "input", "frame-ended", "checksum"] + ); } #[test] @@ -7168,7 +7235,8 @@ mod tests { &initialization, |_job_id, _value| events.borrow_mut().push("descriptor"), |_job_id, _value| events.borrow_mut().push("input"), - |_job_id, _value| panic!("nonterminal job must not publish frame state"), + || panic!("nonterminal job must not set frame-ended state"), + || panic!("nonterminal job must not clear checksum state"), ); assert_eq!(events.into_inner(), vec!["descriptor", "input"]);