diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 083bd56d9..25ec11390 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -354,6 +354,20 @@ typedef char ZSTDMT_rust_raw_seq_store_layout[ ZSTDMT_RustRawSeqStore ZSTDMT_rust_bufferToSeq(ZSTDMT_RustBuffer buffer); ZSTDMT_RustBuffer ZSTDMT_rust_seqToBuffer(ZSTDMT_RustRawSeqStore seq); +typedef int (*ZSTDMT_serialWaitForTurnFn)(void* opaque, unsigned jobID); +typedef void (*ZSTDMT_serialGenerateLdmFn)( + void* opaque, ZSTDMT_RustRawSeqStore* seqStore, + const void* src, size_t srcSize); +typedef void (*ZSTDMT_serialUpdateChecksumFn)( + void* opaque, const void* src, size_t srcSize); +typedef void (*ZSTDMT_serialAdvanceFn)(void* opaque); +void ZSTDMT_rust_serialStateGenSequences( + ZSTDMT_RustRawSeqStore* seqStore, const void* src, size_t srcSize, + unsigned jobID, int ldmEnabled, int checksumEnabled, void* opaque, + ZSTDMT_serialWaitForTurnFn waitForTurn, + ZSTDMT_serialGenerateLdmFn generateLdm, + ZSTDMT_serialUpdateChecksumFn updateChecksum, + ZSTDMT_serialAdvanceFn advance); unsigned ZSTDMT_rust_computeTargetJobLog(unsigned windowLog, unsigned chainLog, int strategy, int enableLdm); @@ -862,44 +876,57 @@ static void ZSTDMT_serialState_free(SerialState* serialState) ZSTD_customFree(serialState->ldmState.bucketOffsets, cMem); } -static void -ZSTDMT_serialState_genSequences(SerialState* serialState, - RawSeqStore_t* seqStore, - Range src, unsigned jobID) +/* Rust owns the serial turn/skip decision and operation ordering. The wait + * callback intentionally leaves the main serial mutex locked; the advance + * callback releases it after Rust has performed the current turn's work. */ +static int ZSTDMT_serialState_waitForTurn(void* opaque, unsigned jobID) { - /* Wait for our turn */ + SerialState* const serialState = (SerialState*)opaque; ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex); while (serialState->nextJobID < jobID) { DEBUGLOG(5, "wait for serialState->cond"); ZSTD_pthread_cond_wait(&serialState->cond, &serialState->mutex); } - /* A future job may error and skip our job */ - if (serialState->nextJobID == jobID) { - /* It is now our turn, do any processing necessary */ - if (serialState->params.ldmParams.enableLdm == ZSTD_ps_enable) { - size_t error; - DEBUGLOG(6, "ZSTDMT_serialState_genSequences: LDM update"); - assert(seqStore->seq != NULL && seqStore->pos == 0 && - seqStore->size == 0 && seqStore->capacity > 0); - assert(src.size <= serialState->params.jobSize); - ZSTD_window_update(&serialState->ldmState.window, src.start, src.size, /* forceNonContiguous */ 0); - error = ZSTD_ldm_generateSequences( - &serialState->ldmState, seqStore, - &serialState->params.ldmParams, src.start, src.size); - /* We provide a large enough buffer to never fail. */ - assert(!ZSTD_isError(error)); (void)error; - /* Update ldmWindow to match the ldmState.window and signal the main - * thread if it is waiting for a buffer. - */ - ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex); - serialState->ldmWindow = serialState->ldmState.window; - ZSTD_pthread_cond_signal(&serialState->ldmWindowCond); - ZSTD_pthread_mutex_unlock(&serialState->ldmWindowMutex); - } - if (serialState->params.fParams.checksumFlag && src.size > 0) - XXH64_update(&serialState->xxhState, src.start, src.size); - } - /* Now it is the next jobs turn */ + return serialState->nextJobID == jobID; +} + +static void ZSTDMT_serialState_generateLdm( + void* opaque, ZSTDMT_RustRawSeqStore* seqStore, + const void* src, size_t srcSize) +{ + SerialState* const serialState = (SerialState*)opaque; + RawSeqStore_t* const cSeqStore = (RawSeqStore_t*)seqStore; + size_t error; + + DEBUGLOG(6, "ZSTDMT_serialState_genSequences: LDM update"); + assert(cSeqStore->seq != NULL && cSeqStore->pos == 0 && + cSeqStore->size == 0 && cSeqStore->capacity > 0); + assert(srcSize <= serialState->params.jobSize); + ZSTD_window_update(&serialState->ldmState.window, src, srcSize, + /* forceNonContiguous */ 0); + error = ZSTD_ldm_generateSequences( + &serialState->ldmState, cSeqStore, + &serialState->params.ldmParams, src, srcSize); + /* We provide a large enough buffer to never fail. */ + assert(!ZSTD_isError(error)); (void)error; + /* Update ldmWindow to match the ldmState.window and signal the main + * thread if it is waiting for a buffer. */ + ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex); + serialState->ldmWindow = serialState->ldmState.window; + ZSTD_pthread_cond_signal(&serialState->ldmWindowCond); + ZSTD_pthread_mutex_unlock(&serialState->ldmWindowMutex); +} + +static void ZSTDMT_serialState_updateChecksum( + void* opaque, const void* src, size_t srcSize) +{ + SerialState* const serialState = (SerialState*)opaque; + XXH64_update(&serialState->xxhState, src, srcSize); +} + +static void ZSTDMT_serialState_advance(void* opaque) +{ + SerialState* const serialState = (SerialState*)opaque; serialState->nextJobID++; ZSTD_pthread_cond_broadcast(&serialState->cond); ZSTD_pthread_mutex_unlock(&serialState->mutex); @@ -1033,8 +1060,16 @@ static void ZSTDMT_compressionJobGenerateSequences(void* opaque) ZSTDMT_jobDescription* const job = state->job; /* Perform serial step as early as possible. */ - ZSTDMT_serialState_genSequences(job->serial, &state->rawSeqStore, - job->src, job->jobID); + ZSTDMT_rust_serialStateGenSequences( + (ZSTDMT_RustRawSeqStore*)&state->rawSeqStore, + job->src.start, job->src.size, job->jobID, + job->serial->params.ldmParams.enableLdm == ZSTD_ps_enable, + job->serial->params.fParams.checksumFlag, + job->serial, + ZSTDMT_serialState_waitForTurn, + ZSTDMT_serialState_generateLdm, + ZSTDMT_serialState_updateChecksum, + ZSTDMT_serialState_advance); } static size_t ZSTDMT_compressionJobBegin(void* opaque) diff --git a/rust/README.md b/rust/README.md index 13be71887..08b32918c 100644 --- a/rust/README.md +++ b/rust/README.md @@ -97,8 +97,10 @@ zstd ABI: job-creation decisions, compression-job stage sequencing and error flow, pending-output decisions through scalar job projections, and frame progression's job-ring scan, error normalization, and active-worker - accounting. Rust also owns frame-block preparation ordering; C callbacks - retain the private match-state window and workspace operations. + accounting. Rust also owns frame-block preparation ordering and the MT + serial turn/skip policy, including LDM-before-checksum sequencing; C + callbacks retain the private match-state window, workspace operations, + synchronization, LDM state, and checksum state. - Dictionary support - `zstd_ddict` owns, loads, copies, and references decode dictionaries. - Legacy decoding @@ -149,7 +151,8 @@ shared- and separate-destination decompression scheduling, single-threaded stream initialization and the buffered/stable stream state machine, MT stream initialization, MT outer scheduling and flush policy, MT compression-job stage sequencing and error flow, MT frame-progression job -aggregation, frame-block preparation ordering, public sequence-API +aggregation, frame-block preparation ordering, MT serial turn/skip and +LDM/checksum sequencing, public sequence-API orchestration, sequence-store and block policy, external-producer invocation and success-path validation, external-sequence-store reset, external-sequence/literals block loop, optional-format decompression loops, diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index edab176a6..f59f75fdf 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -84,6 +84,12 @@ pub type ZSTDMT_compressionJobCompressFn = pub type ZSTDMT_compressionJobErrorFn = unsafe extern "C" fn(*mut c_void, usize); pub type ZSTDMT_compressionJobFinishFn = unsafe extern "C" fn(*mut c_void, usize); +pub type ZSTDMT_serialWaitForTurnFn = unsafe extern "C" fn(*mut c_void, c_uint) -> c_int; +pub type ZSTDMT_serialGenerateLdmFn = + unsafe extern "C" fn(*mut c_void, *mut ZstdMtRawSeqStore, *const c_void, usize); +pub type ZSTDMT_serialUpdateChecksumFn = unsafe extern "C" fn(*mut c_void, *const c_void, usize); +pub type ZSTDMT_serialAdvanceFn = unsafe extern "C" fn(*mut c_void); + /// Scalar inputs for the MT streaming initializer. The full parameter /// object, dictionary handles, pools, buffers, and synchronization remain /// private to C. Rust owns the order in which the C callbacks are invoked and @@ -686,6 +692,82 @@ fn compression_job_with( finish_job(last_block_size); } +/// Drive one ordered MT serial-state turn. +/// +/// The wait callback returns with the C-owned serial mutex held. Rust then +/// performs the LDM-before-checksum policy only for the current job and always +/// invokes the advance callback exactly once, including when an earlier job +/// has already skipped this turn. +#[inline] +fn serial_state_gen_sequences_with( + seq_store: &mut ZstdMtRawSeqStore, + src: *const c_void, + src_size: usize, + job_id: c_uint, + ldm_enabled: bool, + checksum_enabled: bool, + mut wait_for_turn: W, + mut generate_ldm: L, + mut update_checksum: C, + mut advance: A, +) where + W: FnMut(c_uint) -> bool, + L: FnMut(&mut ZstdMtRawSeqStore, *const c_void, usize), + C: FnMut(*const c_void, usize), + A: FnMut(), +{ + if wait_for_turn(job_id) { + if ldm_enabled { + generate_ldm(seq_store, src, src_size); + } + if checksum_enabled && src_size != 0 { + update_checksum(src, src_size); + } + } + advance(); +} + +/// Own MT serial turn/skip and LDM/checksum ordering while C retains the +/// mutexes, LDM window/hash state, checksum state, and raw sequence storage. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_serialStateGenSequences( + seq_store: *mut ZstdMtRawSeqStore, + src: *const c_void, + src_size: usize, + job_id: c_uint, + ldm_enabled: c_int, + checksum_enabled: c_int, + opaque: *mut c_void, + wait_for_turn: Option, + generate_ldm: Option, + update_checksum: Option, + advance: Option, +) { + let (Some(wait_for_turn), Some(generate_ldm), Some(update_checksum), Some(advance)) = + (wait_for_turn, generate_ldm, update_checksum, advance) + else { + return; + }; + if seq_store.is_null() { + return; + } + + unsafe { + serial_state_gen_sequences_with( + &mut *seq_store, + src, + src_size, + job_id, + ldm_enabled != 0, + checksum_enabled != 0, + |job_id| wait_for_turn(opaque, job_id) != 0, + |seq_store, src, src_size| generate_ldm(opaque, seq_store, src, src_size), + |src, src_size| update_checksum(opaque, src, src_size), + || advance(opaque), + ); + } +} + /// C ABI entry point for the worker-job orchestration. C supplies callbacks /// that keep the private descriptor, pools, mutexes, and codec operations on /// the C side of this narrow projection. @@ -3156,6 +3238,86 @@ mod tests { assert_eq!(state.finished, vec![7]); } + #[test] + fn serial_turn_runs_ldm_before_checksum_and_advances_once() { + let events = Rc::new(RefCell::new(Vec::new())); + let wait_events = Rc::clone(&events); + let ldm_events = Rc::clone(&events); + let checksum_events = Rc::clone(&events); + let advance_events = Rc::clone(&events); + let mut seq_store = ZstdMtRawSeqStore::default(); + + serial_state_gen_sequences_with( + &mut seq_store, + ptr::null(), + 8, + 4, + true, + true, + move |job_id| { + assert_eq!(job_id, 4); + wait_events.borrow_mut().push("wait"); + true + }, + move |_seq_store, _src, src_size| { + assert_eq!(src_size, 8); + ldm_events.borrow_mut().push("ldm"); + }, + move |_src, src_size| { + assert_eq!(src_size, 8); + checksum_events.borrow_mut().push("checksum"); + }, + move || advance_events.borrow_mut().push("advance"), + ); + + assert_eq!(&*events.borrow(), &["wait", "ldm", "checksum", "advance"]); + } + + #[test] + fn serial_turn_skips_failed_predecessor_and_keeps_empty_ldm_turn() { + let events = Rc::new(RefCell::new(Vec::new())); + let wait_events = Rc::clone(&events); + let advance_events = Rc::clone(&events); + let mut seq_store = ZstdMtRawSeqStore::default(); + + serial_state_gen_sequences_with( + &mut seq_store, + ptr::null(), + 0, + 9, + true, + true, + move |_job_id| { + wait_events.borrow_mut().push("wait"); + false + }, + |_seq_store, _src, _src_size| panic!("skipped jobs must not generate LDM sequences"), + |_src, _src_size| panic!("skipped jobs must not update the checksum"), + move || advance_events.borrow_mut().push("advance"), + ); + assert_eq!(&*events.borrow(), &["wait", "advance"]); + + let empty_events = Rc::new(RefCell::new(Vec::new())); + let empty_ldm_events = Rc::clone(&empty_events); + let empty_advance_events = Rc::clone(&empty_events); + serial_state_gen_sequences_with( + &mut seq_store, + ptr::null(), + 0, + 10, + true, + true, + |_job_id| true, + move |_seq_store, _src, src_size| { + assert_eq!(src_size, 0); + empty_ldm_events.borrow_mut().push("ldm"); + }, + |_src, _src_size| panic!("empty input must not update the checksum"), + move || empty_advance_events.borrow_mut().push("advance"), + ); + assert_eq!(&*empty_events.borrow(), &["ldm", "advance"]); + } + #[test] fn compression_job_stops_on_non_first_chunk_error_and_cleans_up() { let state = Rc::new(RefCell::new(MockCompressionJob::default()));