diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 2330d4303..6b2eeda7f 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -362,6 +362,24 @@ typedef char ZSTDMT_compression_job_finish_projection_layout[ + sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection)) ? 1 : -1]; +typedef struct { + void* callbackContext; + unsigned firstJob; + ZSTDMT_compressionJobVoidFn assertNoExtDict; + ZSTDMT_compressionJobVoidFn trace; +} ZSTDMT_RustCompressionJobTraceProjection; +typedef char ZSTDMT_compression_job_trace_projection_layout[ + (offsetof(ZSTDMT_RustCompressionJobTraceProjection, callbackContext) == 0 + && offsetof(ZSTDMT_RustCompressionJobTraceProjection, firstJob) + == sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobTraceProjection, assertNoExtDict) + == 2 * sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobTraceProjection, trace) + == 3 * sizeof(void*) + && sizeof(ZSTDMT_RustCompressionJobTraceProjection) + == 4 * sizeof(void*)) + ? 1 : -1]; + typedef struct { int checksumFlag; int ldmEnable; @@ -422,6 +440,8 @@ void ZSTDMT_rust_compressionJob( ZSTDMT_compressionJobApplySequencesFn applySequences, ZSTDMT_compressionJobCompressFn compressJob, ZSTDMT_compressionJobVoidFn traceJob); +void ZSTDMT_rust_compressionJobTrace( + const ZSTDMT_RustCompressionJobTraceProjection* projection); ZSTDMT_chunkProcessResult ZSTDMT_rust_compressJobChunks( ZSTD_CCtx* cctx, const void* src, size_t srcSize, @@ -2247,18 +2267,34 @@ static ZSTDMT_chunkProcessResult ZSTDMT_compressionJobCompress( &progress, ZSTDMT_rust_compressionJobProgress); } +static void ZSTDMT_compressionJobTraceAssertNoExtDict(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + + /* Double check that we don't have an ext-dict, because then our repcode + * invalidation doesn't work. */ + assert(!ZSTD_window_hasExtDict(state->cctx->blockState.matchState.window)); +} + +static void ZSTDMT_compressionJobTraceRecord(void* opaque) +{ + ZSTDMT_compressionJobState* const state = + (ZSTDMT_compressionJobState*)opaque; + ZSTD_CCtx_trace(state->cctx, 0); +} + static void ZSTDMT_compressionJobTrace(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; - ZSTDMT_jobDescription* const job = state->job; - - if (!job->firstJob) { - /* Double check that we don't have an ext-dict, because then our - * repcode invalidation doesn't work. */ - assert(!ZSTD_window_hasExtDict(state->cctx->blockState.matchState.window)); - } - ZSTD_CCtx_trace(state->cctx, 0); + ZSTDMT_RustCompressionJobTraceProjection const projection = { + state, + state->job->firstJob, + ZSTDMT_compressionJobTraceAssertNoExtDict, + ZSTDMT_compressionJobTraceRecord + }; + ZSTDMT_rust_compressionJobTrace(&projection); } static void ZSTDMT_compressionJobEnsureFinished(void* opaque) diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 0a20be203..16abb1ee6 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -840,6 +840,67 @@ const _: () = { + size_of::()); }; +/// C retains the private window assertion and trace-recording leaves for an +/// MT worker. Rust owns the first-job policy that determines whether the +/// external-dictionary assertion is required before tracing. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct ZSTDMT_compressionJobTraceProjection { + pub callbackContext: *mut c_void, + pub firstJob: c_uint, + pub assertNoExtDict: Option, + pub trace: Option, +} + +const _: () = { + assert!(offset_of!(ZSTDMT_compressionJobTraceProjection, callbackContext) == 0); + assert!(offset_of!(ZSTDMT_compressionJobTraceProjection, firstJob) == size_of::()); + assert!( + offset_of!(ZSTDMT_compressionJobTraceProjection, assertNoExtDict) + == 2 * size_of::() + ); + assert!(offset_of!(ZSTDMT_compressionJobTraceProjection, trace) == 3 * size_of::()); + assert!(size_of::() == 4 * size_of::()); +}; + +#[inline] +fn compression_job_trace_with(first_job: c_uint, mut assert_no_ext_dict: A, mut trace: T) +where + A: FnMut(), + T: FnMut(), +{ + if first_job == 0 { + assert_no_ext_dict(); + } + trace(); +} + +/// Apply the worker trace policy while C retains the opaque trace leaves. +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_compressionJobTrace( + projection: *const ZSTDMT_compressionJobTraceProjection, +) { + let Some(projection) = (unsafe { projection.as_ref() }) else { + return; + }; + let Some(trace) = projection.trace else { + return; + }; + if projection.firstJob == 0 && projection.assertNoExtDict.is_none() { + return; + } + let assert_no_ext_dict = projection.assertNoExtDict; + compression_job_trace_with( + projection.firstJob, + || { + if let Some(assert_no_ext_dict) = assert_no_ext_dict { + unsafe { assert_no_ext_dict(projection.callbackContext) }; + } + }, + || unsafe { trace(projection.callbackContext) }, + ); +} + pub type ZSTDMT_serialWaitForTurnFn = unsafe extern "C" fn(*mut c_void, c_uint) -> c_int; pub type ZSTDMT_serialLdmWindowUpdateFn = unsafe extern "C" fn(*mut c_void, *mut ZstdMtRawSeqStore, *const c_void, usize); @@ -7041,6 +7102,29 @@ mod tests { ); } + #[test] + fn compression_job_trace_guards_only_continuation_jobs() { + let first_job_events = Rc::new(RefCell::new(Vec::<&'static str>::new())); + let first_job_assert_events = Rc::clone(&first_job_events); + let first_job_trace_events = Rc::clone(&first_job_events); + compression_job_trace_with( + 1, + move || first_job_assert_events.borrow_mut().push("assert"), + move || first_job_trace_events.borrow_mut().push("trace"), + ); + assert_eq!(&*first_job_events.borrow(), &["trace"]); + + let continuation_events = Rc::new(RefCell::new(Vec::<&'static str>::new())); + let continuation_assert_events = Rc::clone(&continuation_events); + let continuation_trace_events = Rc::clone(&continuation_events); + compression_job_trace_with( + 0, + move || continuation_assert_events.borrow_mut().push("assert"), + move || continuation_trace_events.borrow_mut().push("trace"), + ); + assert_eq!(&*continuation_events.borrow(), &["assert", "trace"]); + } + #[test] fn compression_job_runs_first_job_stages_and_reports_final_block() { let state = Rc::new(RefCell::new(MockCompressionJob::default()));