diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 170cd3bb9..3808861e7 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -121,10 +121,59 @@ typedef struct { size_t lastBlockSize; } ZSTDMT_chunkProcessResult; +typedef struct { + void* dst; + size_t dstCapacity; + int* stage; + int noDictIDFlag; + int checksumFlag; + int contentSizeFlag; + int format; + U32 windowLog; + U64 pledgedSrcSizePlusOne; + U32 dictID; + U32* repCodes; +} ZSTDMT_RustCompressionJobFrameHeaderState; +typedef char ZSTDMT_compression_job_frame_header_state_layout[ + (offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, dst) == 0 + && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, dstCapacity) + == sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, stage) + == 2 * sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, noDictIDFlag) + == 3 * sizeof(void*) + && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, checksumFlag) + == 3 * sizeof(void*) + sizeof(int) + && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, contentSizeFlag) + == 3 * sizeof(void*) + 2 * sizeof(int) + && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, format) + == 3 * sizeof(void*) + 3 * sizeof(int) + && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, windowLog) + == 3 * sizeof(void*) + 4 * sizeof(int) + && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, pledgedSrcSizePlusOne) + == (sizeof(void*) == 8 ? 6 * sizeof(void*) : 8 * sizeof(void*)) + && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, dictID) + == (sizeof(void*) == 8 ? 6 * sizeof(void*) : 8 * sizeof(void*)) + + sizeof(U64) + && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, repCodes) + == (sizeof(void*) == 8 ? 8 * sizeof(void*) : 11 * sizeof(void*)) + && sizeof(ZSTDMT_RustCompressionJobFrameHeaderState) + == (sizeof(void*) == 8 ? 72 : 48)) + ? 1 : -1]; + typedef struct { unsigned firstJob; unsigned lastJob; + const ZSTDMT_RustCompressionJobFrameHeaderState* frameHeaderState; } ZSTDMT_RustCompressionJobProjection; +typedef char ZSTDMT_compression_job_projection_layout[ + (offsetof(ZSTDMT_RustCompressionJobProjection, firstJob) == 0 + && offsetof(ZSTDMT_RustCompressionJobProjection, lastJob) == sizeof(unsigned) + && offsetof(ZSTDMT_RustCompressionJobProjection, frameHeaderState) + == 2 * sizeof(unsigned) + && sizeof(ZSTDMT_RustCompressionJobProjection) + == 2 * sizeof(unsigned) + sizeof(void*)) + ? 1 : -1]; typedef struct { int status; @@ -149,7 +198,6 @@ void ZSTDMT_rust_compressionJob( ZSTDMT_compressionJobVoidFn generateSequences, ZSTDMT_compressionJobStepFn beginJob, ZSTDMT_compressionJobVoidFn applySequences, - ZSTDMT_compressionJobStepFn writeFrameHeader, ZSTDMT_compressionJobCompressFn compressJob, ZSTDMT_compressionJobVoidFn traceJob, ZSTDMT_compressionJobErrorFn setError, @@ -1355,6 +1403,7 @@ typedef struct { ZSTD_CCtx* cctx; RawSeqStore_t rawSeqStore; Buffer dstBuff; + ZSTDMT_RustCompressionJobFrameHeaderState frameHeaderState; } ZSTDMT_compressionJobState; static size_t ZSTDMT_compressionJobAcquireResources(void* opaque) @@ -1373,6 +1422,8 @@ static size_t ZSTDMT_compressionJobAcquireResources(void* opaque) if (state->dstBuff.start == NULL) return ERROR(memory_allocation); job->dstBuff = state->dstBuff; } + state->frameHeaderState.dst = state->dstBuff.start; + state->frameHeaderState.dstCapacity = state->dstBuff.capacity; if (state->jobParams.ldmParams.enableLdm == ZSTD_ps_enable && state->rawSeqStore.seq == NULL) return ERROR(memory_allocation); @@ -1418,18 +1469,17 @@ static size_t ZSTDMT_compressionJobBegin(void* opaque) ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; + size_t initError; if (job->cdict) { - size_t const initError = ZSTD_compressBegin_advanced_internal( + initError = ZSTD_compressBegin_advanced_internal( state->cctx, NULL, 0, ZSTD_dct_auto, ZSTD_dtlm_fast, job->cdict, &state->jobParams, job->fullFrameSize); assert(job->firstJob); /* only allowed for first job */ - return initError; - } - - { U64 const pledgedSrcSize = job->firstJob ? job->fullFrameSize : job->src.size; + } else { + U64 const pledgedSrcSize = job->firstJob ? job->fullFrameSize : job->src.size; size_t const forceWindowError = ZSTD_CCtxParams_setParameter( - &state->jobParams, ZSTD_c_forceMaxWindow, !job->firstJob); + &state->jobParams, ZSTD_c_forceMaxWindow, !job->firstJob); if (ZSTD_isError(forceWindowError)) return forceWindowError; if (!job->firstJob) { size_t const err = ZSTD_CCtxParams_setParameter( @@ -1437,11 +1487,29 @@ static size_t ZSTDMT_compressionJobBegin(void* opaque) if (ZSTD_isError(err)) return err; } DEBUGLOG(6, "ZSTDMT_compressionJob: job %u: loading prefix of size %zu", job->jobID, job->prefix.size); - return ZSTD_compressBegin_advanced_internal( + initError = ZSTD_compressBegin_advanced_internal( state->cctx, job->prefix.start, job->prefix.size, ZSTD_dct_rawContent, ZSTD_dtlm_fast, NULL, /*cdict*/ &state->jobParams, pledgedSrcSize); } + + if (ZSTD_isError(initError)) return initError; + state->frameHeaderState.stage = (int*)&state->cctx->stage; + state->frameHeaderState.noDictIDFlag = + state->cctx->appliedParams.fParams.noDictIDFlag; + state->frameHeaderState.checksumFlag = + state->cctx->appliedParams.fParams.checksumFlag; + state->frameHeaderState.contentSizeFlag = + state->cctx->appliedParams.fParams.contentSizeFlag; + state->frameHeaderState.format = (int)state->cctx->appliedParams.format; + state->frameHeaderState.windowLog = + state->cctx->appliedParams.cParams.windowLog; + state->frameHeaderState.pledgedSrcSizePlusOne = + state->cctx->pledgedSrcSizePlusOne; + state->frameHeaderState.dictID = state->cctx->dictID; + state->frameHeaderState.repCodes = state->cctx->blockState.prevCBlock == NULL + ? NULL : state->cctx->blockState.prevCBlock->rep; + return initError; } static void ZSTDMT_compressionJobApplySequences(void* opaque) @@ -1455,20 +1523,6 @@ static void ZSTDMT_compressionJobApplySequences(void* opaque) &state->rawSeqStore); } -static size_t ZSTDMT_compressionJobWriteFrameHeader(void* opaque) -{ - ZSTDMT_compressionJobState* const state = - (ZSTDMT_compressionJobState*)opaque; - ZSTDMT_jobDescription* const job = state->job; - size_t const hSize = ZSTD_compressContinue_public( - state->cctx, state->dstBuff.start, state->dstBuff.capacity, - job->src.start, 0); - if (ZSTD_isError(hSize)) return hSize; - DEBUGLOG(5, "ZSTDMT_compressionJob: flush and overwrite %u bytes of frame header (not first job)", (U32)hSize); - ZSTD_invalidateRepCodes(state->cctx); - return hSize; -} - static ZSTDMT_chunkProcessResult ZSTDMT_compressionJobCompress( void* opaque, unsigned lastJob) { @@ -1545,7 +1599,8 @@ static void ZSTDMT_compressionJob(void* jobDescription) ZSTDMT_compressionJobState state; ZSTDMT_RustCompressionJobProjection const projection = { job->firstJob, - job->lastJob + job->lastJob, + &state.frameHeaderState }; ZSTD_memset(&state, 0, sizeof(state)); @@ -1560,7 +1615,6 @@ static void ZSTDMT_compressionJob(void* jobDescription) ZSTDMT_compressionJobGenerateSequences, ZSTDMT_compressionJobBegin, ZSTDMT_compressionJobApplySequences, - ZSTDMT_compressionJobWriteFrameHeader, ZSTDMT_compressionJobCompress, ZSTDMT_compressionJobTrace, ZSTDMT_compressionJobSetError, diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 11d61dcfe..1924a0e56 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -21,7 +21,8 @@ use std::sync::Mutex; use crate::bits::ZSTD_highbit32; use crate::errors::{ERR_isError, ZstdErrorCode, ERROR}; -use crate::zstd_compress::ZSTD_frameProgression; +use crate::zstd_compress::{ZSTD_frameProgression, ZSTD_rust_invalidateRepCodes}; +use crate::zstd_compress_frame::ZSTD_rust_writeFrameHeader; const ZSTDMT_JOBLOG_MAX: c_uint = if mem::size_of::() == 4 { 29 } else { 30 }; const ZSTD_WINDOWLOG_MAX: c_uint = if mem::size_of::() == 4 { 30 } else { 31 }; @@ -68,13 +69,145 @@ pub struct ZSTDMT_chunkProcessResult { /// Scalar job state used by the Rust compression-job scheduler. /// /// The job descriptor, pools, synchronization, and codec state remain -/// private to C. Rust uses only the frame-position flags to choose the -/// sequencing and non-first-job header stages. +/// private to C. Rust uses the frame-position flags and a scalar header +/// projection to choose the non-first-job header stage. #[repr(C)] #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct ZSTDMT_compressionJobProjection { pub firstJob: c_uint, pub lastJob: c_uint, + pub frameHeaderState: *const ZSTDMT_compressionJobFrameHeaderState, +} + +/// Scalar projection for the non-first MT job frame header. The destination, +/// stage, and repcodes point at the worker-owned C storage; all header +/// serialization and repcode invalidation happen in Rust. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_compressionJobFrameHeaderState { + dst: *mut c_void, + dst_capacity: usize, + stage: *mut c_int, + no_dict_id_flag: c_int, + checksum_flag: c_int, + content_size_flag: c_int, + format: c_int, + window_log: c_uint, + pledged_src_size_plus_one: u64, + dict_id: c_uint, + rep_codes: *mut c_uint, +} + +const _: () = { + assert!(offset_of!(ZSTDMT_compressionJobProjection, firstJob) == 0); + assert!(offset_of!(ZSTDMT_compressionJobProjection, lastJob) == size_of::()); + assert!( + offset_of!(ZSTDMT_compressionJobProjection, frameHeaderState) == size_of::<[c_uint; 2]>() + ); + assert!( + size_of::() + == size_of::<[c_uint; 2]>() + size_of::() + ); + assert!(offset_of!(ZSTDMT_compressionJobFrameHeaderState, dst) == 0); + assert!(offset_of!(ZSTDMT_compressionJobFrameHeaderState, dst_capacity) == size_of::()); + assert!(offset_of!(ZSTDMT_compressionJobFrameHeaderState, stage) == 2 * size_of::()); + assert!( + offset_of!(ZSTDMT_compressionJobFrameHeaderState, no_dict_id_flag) + == 3 * size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFrameHeaderState, checksum_flag) + == 3 * size_of::() + size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFrameHeaderState, content_size_flag) + == 3 * size_of::() + 2 * size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFrameHeaderState, format) + == 3 * size_of::() + 3 * size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFrameHeaderState, window_log) + == 3 * size_of::() + 4 * size_of::() + ); + assert!( + offset_of!( + ZSTDMT_compressionJobFrameHeaderState, + pledged_src_size_plus_one + ) == ZSTDMT_HEADER_PLEDGED_OFFSET + ); + assert!( + offset_of!(ZSTDMT_compressionJobFrameHeaderState, dict_id) + == ZSTDMT_HEADER_PLEDGED_OFFSET + size_of::() + ); + assert!( + offset_of!(ZSTDMT_compressionJobFrameHeaderState, rep_codes) + == ZSTDMT_HEADER_REP_CODES_OFFSET + ); + assert!( + size_of::() + == if size_of::() == 8 { 72 } else { 48 } + ); +}; + +const ZSTD_COMPRESSION_STAGE_CREATED: c_int = 0; +const ZSTD_COMPRESSION_STAGE_INIT: c_int = 1; +const ZSTD_COMPRESSION_STAGE_ONGOING: c_int = 2; +const ZSTDMT_HEADER_PLEDGED_OFFSET: usize = if size_of::() == 8 { + size_of::<[usize; 6]>() +} else { + size_of::<[usize; 8]>() +}; +const ZSTDMT_HEADER_REP_CODES_OFFSET: usize = if size_of::() == 8 { + size_of::<[usize; 8]>() +} else { + size_of::<[usize; 11]>() +}; + +#[no_mangle] +pub unsafe extern "C" fn ZSTDMT_rust_writeFrameHeader( + state: *const ZSTDMT_compressionJobFrameHeaderState, +) -> usize { + if state.is_null() { + return ERROR(ZstdErrorCode::Generic); + } + let state = unsafe { &*state }; + if state.stage.is_null() || state.rep_codes.is_null() { + return ERROR(ZstdErrorCode::Generic); + } + if unsafe { *state.stage } == ZSTD_COMPRESSION_STAGE_CREATED { + return ERROR(ZstdErrorCode::StageWrong); + } + + let header_size = if unsafe { *state.stage } == ZSTD_COMPRESSION_STAGE_INIT { + let header_size = unsafe { + ZSTD_rust_writeFrameHeader( + state.dst, + state.dst_capacity, + state.no_dict_id_flag, + state.checksum_flag, + state.content_size_flag, + state.format, + state.window_log, + state.pledged_src_size_plus_one.wrapping_sub(1), + state.dict_id, + ) + }; + if ERR_isError(header_size) { + return header_size; + } + if header_size > state.dst_capacity { + return ERROR(ZstdErrorCode::DstSizeTooSmall); + } + unsafe { *state.stage = ZSTD_COMPRESSION_STAGE_ONGOING }; + header_size + } else { + 0 + }; + + unsafe { ZSTD_rust_invalidateRepCodes(state.rep_codes) }; + header_size } pub type ZSTDMT_compressionJobStepFn = unsafe extern "C" fn(*mut c_void) -> usize; @@ -1116,7 +1249,6 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( generateSequences: Option, beginJob: Option, applySequences: Option, - writeFrameHeader: Option, compressJob: Option, traceJob: Option, setError: Option, @@ -1131,7 +1263,6 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( Some(generate_sequences), Some(begin_job), Some(apply_sequences), - Some(write_frame_header), Some(compress_job), Some(trace_job), Some(set_error), @@ -1142,7 +1273,6 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( generateSequences, beginJob, applySequences, - writeFrameHeader, compressJob, traceJob, setError, @@ -1159,7 +1289,7 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJob( || unsafe { generate_sequences(opaque) }, || unsafe { begin_job(opaque) }, || unsafe { apply_sequences(opaque) }, - || unsafe { write_frame_header(opaque) }, + || unsafe { ZSTDMT_rust_writeFrameHeader(projection.frameHeaderState) }, |last_job| unsafe { compress_job(opaque, last_job) }, || unsafe { trace_job(opaque) }, |error| unsafe { set_error(opaque, error) }, @@ -3994,6 +4124,43 @@ mod tests { state.borrow_mut().events.push(event); } + #[test] + fn compression_job_frame_header_projection_writes_header_and_clears_repcodes() { + let mut stage = ZSTD_COMPRESSION_STAGE_INIT; + let mut rep_codes = [11u32, 22, 33]; + let mut output = [0xa5u8; 18]; + let state = ZSTDMT_compressionJobFrameHeaderState { + dst: output.as_mut_ptr().cast(), + dst_capacity: output.len(), + stage: &mut stage, + no_dict_id_flag: 0, + checksum_flag: 0, + content_size_flag: 1, + format: 0, + window_log: 20, + pledged_src_size_plus_one: 1, + dict_id: 0, + rep_codes: rep_codes.as_mut_ptr(), + }; + + let mut created_stage = ZSTD_COMPRESSION_STAGE_CREATED; + let created_state = ZSTDMT_compressionJobFrameHeaderState { + stage: &mut created_stage, + ..state + }; + let created = unsafe { ZSTDMT_rust_writeFrameHeader(&created_state) }; + assert_eq!(created, ERROR(ZstdErrorCode::StageWrong)); + assert_eq!(created_stage, ZSTD_COMPRESSION_STAGE_CREATED); + assert_eq!(rep_codes, [11, 22, 33]); + let result = unsafe { ZSTDMT_rust_writeFrameHeader(&state) }; + + assert_eq!(result, 6); + assert_eq!(stage, ZSTD_COMPRESSION_STAGE_ONGOING); + assert_eq!(rep_codes, [0; 3]); + assert_eq!(&output[..result], &[0x28, 0xb5, 0x2f, 0xfd, 0x20, 0x00]); + assert_eq!(&output[result..], &[0xa5; 12]); + } + #[test] fn compression_job_runs_first_job_stages_and_reports_final_block() { let state = Rc::new(RefCell::new(MockCompressionJob::default())); @@ -4010,6 +4177,7 @@ mod tests { ZSTDMT_compressionJobProjection { firstJob: 1, lastJob: 1, + frameHeaderState: ptr::null(), }, move || { record_compression_job_event(&acquire_state, "acquire"); @@ -4250,6 +4418,7 @@ mod tests { ZSTDMT_compressionJobProjection { firstJob: 0, lastJob: 1, + frameHeaderState: ptr::null(), }, move || { record_compression_job_event(&acquire_state, "acquire");