diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 960cbd5f8..b0e0e8137 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -467,6 +467,16 @@ typedef struct { size_t result; size_t outputPos; } ZSTDMT_RustStreamFlushResult; +typedef struct { + size_t result; + unsigned inBuffActive; +} ZSTDMT_RustStreamCreateJobResult; +typedef char ZSTDMT_stream_create_job_result_layout[ + (offsetof(ZSTDMT_RustStreamCreateJobResult, result) == 0 + && offsetof(ZSTDMT_RustStreamCreateJobResult, inBuffActive) + == sizeof(size_t) + && sizeof(ZSTDMT_RustStreamCreateJobResult) == 2 * sizeof(size_t)) + ? 1 : -1]; typedef struct { size_t result; size_t inputPos; @@ -483,7 +493,8 @@ typedef char ZSTDMT_compress_stream_result_layout[ ? 1 : -1]; typedef int (*ZSTDMT_streamTryGetInputRangeFn)( void* opaque, ZSTDMT_RustStreamInputRangeProjection* projection); -typedef size_t (*ZSTDMT_streamCreateJobFn)(void* opaque, size_t srcSize, unsigned end); +typedef ZSTDMT_RustStreamCreateJobResult (*ZSTDMT_streamCreateJobFn)( + void* opaque, size_t srcSize, unsigned end); typedef ZSTDMT_RustStreamFlushResult (*ZSTDMT_streamFlushProducedFn)( void* opaque, void* outputDst, size_t outputSize, size_t outputPos, unsigned blockToFlush, unsigned end); @@ -3064,12 +3075,14 @@ static int ZSTDMT_tryAddCompressionJob(void* opaque, unsigned jobID) return POOL_tryAdd(mtctx->factory, ZSTDMT_compressionJob, &mtctx->jobs[jobID]); } -static size_t ZSTDMT_createCompressionJob(void* opaque, size_t srcSize, unsigned end) +static ZSTDMT_RustStreamCreateJobResult ZSTDMT_createCompressionJob( + void* opaque, size_t srcSize, unsigned end) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTD_EndDirective const endOp = (ZSTD_EndDirective)end; ZSTDMT_RustCreateJobProjection projection; ZSTDMT_RustCreateJobResult result; + ZSTDMT_RustStreamCreateJobResult streamResult; /* Rust fills the reusable input range in its local projection before * calling this C-owned job-construction callback. Publish that count * before the callback prepares a job and checks the C context. */ @@ -3090,13 +3103,19 @@ static size_t ZSTDMT_createCompressionJob(void* opaque, size_t srcSize, unsigned (unsigned)mtctx->params.fParams.checksumFlag }; result = ZSTDMT_rust_createCompressionJob( - &projection, mtctx, ZSTDMT_prepareCompressionJob, - ZSTDMT_writeEmptyCompressionJob, ZSTDMT_tryAddCompressionJob); + &projection, mtctx, ZSTDMT_prepareCompressionJob, + ZSTDMT_writeEmptyCompressionJob, ZSTDMT_tryAddCompressionJob); + /* Keep the private C mutation in this callback, but let Rust decide + * whether its local inBuffFilled value may be published afterwards. */ + streamResult = (ZSTDMT_RustStreamCreateJobResult){ + result.returnCode, + (unsigned)(mtctx->inBuff.buffer.start != NULL) + }; if (result.action == ZSTDMT_CREATE_JOB_TABLE_FULL) { DEBUGLOG(5, "ZSTDMT_createCompressionJob: will not create new job : table is full"); assert((mtctx->nextJobID & mtctx->jobIDMask) == (mtctx->doneJobID & mtctx->jobIDMask)); - return result.returnCode; + return streamResult; } mtctx->nextJobID = result.nextJobID; @@ -3104,7 +3123,7 @@ static size_t ZSTDMT_createCompressionJob(void* opaque, size_t srcSize, unsigned if (result.action == ZSTDMT_CREATE_JOB_EMPTY) { DEBUGLOG(5, "ZSTDMT_createCompressionJob: creating a last empty block to end frame"); assert(endOp == ZSTD_e_end); /* only possible case : need to end the frame with an empty last block */ - return result.returnCode; + return streamResult; } DEBUGLOG(5, "ZSTDMT_createCompressionJob: posting job %u : %u bytes (end:%u, jobNb == %u (mod:%u))", @@ -3115,7 +3134,7 @@ static size_t ZSTDMT_createCompressionJob(void* opaque, size_t srcSize, unsigned result.jobID); if (result.jobReady) DEBUGLOG(5, "ZSTDMT_createCompressionJob: no worker available for job %u", result.jobNumber); - return result.returnCode; + return streamResult; } /* The Rust flush state machine receives only this synchronized scalar view. @@ -3464,11 +3483,10 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, (U32)endOp, (U32)(input->size - input->pos)); assert(output->pos <= output->size); assert(input->pos <= input->size); - /* A prepared job clears the C range while Rust still has the pre-job - * count in its local projection. Only publish the result when the range - * remains active, which is the no-job-created path. */ - if (mtctx->inBuff.buffer.start != NULL) - mtctx->inBuff.filled = result.inBuffFilled; + /* Rust owns the decision to publish its local count. C retains the + * private field mutation and receives either the count or zero when the + * job callback cleared the range. */ + mtctx->inBuff.filled = result.inBuffFilled; input->pos = result.inputPos; output->pos = result.outputPos; DEBUGLOG(5, "end of ZSTDMT_compressStream_generic: remainingToFlush = %u", (U32)result.result); diff --git a/rust/src/zstdmt_compress.rs b/rust/src/zstdmt_compress.rs index 0797904a3..be1ea709c 100644 --- a/rust/src/zstdmt_compress.rs +++ b/rust/src/zstdmt_compress.rs @@ -1588,6 +1588,24 @@ pub struct ZSTDMT_streamFlushResult { pub outputPos: usize, } +/// Result of the C-owned job callback. C reports only whether its private +/// input range remains active; Rust owns the policy that publishes its local +/// `inBuffFilled` value after the callback. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ZSTDMT_streamCreateJobResult { + pub result: usize, + pub inBuffActive: c_uint, +} + +const _: () = { + assert!(offset_of!(ZSTDMT_streamCreateJobResult, result) == 0); + assert!( + offset_of!(ZSTDMT_streamCreateJobResult, inBuffActive) == size_of::() + ); + assert!(size_of::() == 2 * size_of::()); +}; + /// Result of one outer MT stream scheduling pass. #[repr(C)] #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] @@ -1611,7 +1629,11 @@ pub type ZSTDMT_streamTryGetInputRangeFn = unsafe extern "C" fn( projection: *mut ZSTDMT_streamInputRangeProjection, ) -> c_int; pub type ZSTDMT_streamCreateJobFn = - unsafe extern "C" fn(opaque: *mut c_void, src_size: usize, end: c_uint) -> usize; + unsafe extern "C" fn( + opaque: *mut c_void, + src_size: usize, + end: c_uint, + ) -> ZSTDMT_streamCreateJobResult; pub type ZSTDMT_streamFlushProducedFn = unsafe extern "C" fn( opaque: *mut c_void, output_dst: *mut c_void, @@ -3433,6 +3455,15 @@ fn compress_stream_result( } } +#[inline] +fn publishable_in_buff_filled(buffer_start: *mut c_void, buffer_filled: usize) -> usize { + if buffer_start.is_null() { + 0 + } else { + buffer_filled + } +} + /// Run one outer MT streaming pass while C retains the context and worker /// implementation details. /// @@ -3454,7 +3485,7 @@ unsafe fn compress_stream_generic_with( ) -> ZSTDMT_compressStreamResult where T: FnMut(&mut ZSTDMT_streamInputRangeProjection) -> c_int, - J: FnMut(usize, c_uint) -> usize, + J: FnMut(usize, c_uint) -> ZSTDMT_streamCreateJobResult, F: FnMut(*mut c_void, usize, usize, c_uint, c_uint) -> ZSTDMT_streamFlushResult, S: FnMut( &ZSTDMT_compressStreamContextProjection, @@ -3464,6 +3495,8 @@ where { let mut input_pos = input.pos; let output_pos = output.pos; + let initial_in_buff_filled = + publishable_in_buff_filled(context.inBuffStart, context.inBuffFilled); if input_pos > input.size || output_pos > output.size || context.targetSectionSize == 0 @@ -3474,7 +3507,7 @@ where ERROR(ZstdErrorCode::Generic), input_pos, output_pos, - context.inBuffFilled, + initial_in_buff_filled, ); } @@ -3483,7 +3516,7 @@ where ERROR(ZstdErrorCode::StageWrong), input_pos, output_pos, - context.inBuffFilled, + initial_in_buff_filled, ); } @@ -3520,7 +3553,10 @@ where ERROR(ZstdErrorCode::Generic), input_pos, output_pos, - input_range.bufferFilled, + publishable_in_buff_filled( + input_range.bufferStart, + input_range.bufferFilled, + ), ); }; if sync_point.toLoad > available @@ -3531,7 +3567,10 @@ where ERROR(ZstdErrorCode::Generic), input_pos, output_pos, - input_range.bufferFilled, + publishable_in_buff_filled( + input_range.bufferStart, + input_range.bufferFilled, + ), ); } @@ -3564,6 +3603,8 @@ where end_op = ZSTD_E_FLUSH; } + let mut returned_in_buff_filled = + publishable_in_buff_filled(input_range.bufferStart, input_range.bufferFilled); if context.jobReady != 0 || input_range.bufferFilled >= context.targetSectionSize || (end_op != ZSTD_E_CONTINUE && input_range.bufferFilled > 0) @@ -3571,12 +3612,17 @@ where { debug_assert!(input_range.bufferFilled <= context.targetSectionSize); let create_result = create_job(input_range.bufferFilled, end_op); - if ERR_isError(create_result) { + returned_in_buff_filled = if create_result.inBuffActive != 0 { + input_range.bufferFilled + } else { + 0 + }; + if ERR_isError(create_result.result) { return compress_stream_result( - create_result, + create_result.result, input_pos, output_pos, - input_range.bufferFilled, + returned_in_buff_filled, ); } } @@ -3598,7 +3644,7 @@ where result, input_pos, flush_result.outputPos, - input_range.bufferFilled, + returned_in_buff_filled, ) } @@ -8273,7 +8319,10 @@ mod tests { }, |src_size, end| { create_calls.push((src_size, end)); - 0 + ZSTDMT_streamCreateJobResult { + result: 0, + inBuffActive: 1, + } }, |_dst, _size, output_pos, block_to_flush, end| { flush_calls.push((block_to_flush, end)); @@ -8300,6 +8349,47 @@ mod tests { assert_eq!(output_buffer[0], 0xb6); } + #[test] + fn outer_scheduler_suppresses_filled_after_job_clears_range() { + let mut round_buffer = [0xa5u8; 4]; + let context = ZSTDMT_compressStreamContextProjection { + inBuffStart: round_buffer.as_mut_ptr().cast(), + inBuffCapacity: round_buffer.len(), + inBuffFilled: 2, + targetSectionSize: round_buffer.len(), + ..ZSTDMT_compressStreamContextProjection::default() + }; + let mut create_call = None; + + let result = unsafe { + compress_stream_generic_with( + context, + ZSTDMT_streamInputProjection::default(), + ZSTDMT_streamOutputProjection::default(), + ZSTD_E_FLUSH, + |_projection| panic!("the active range must not be reacquired"), + |src_size, end| { + create_call = Some((src_size, end)); + ZSTDMT_streamCreateJobResult { + result: 0, + inBuffActive: 0, + } + }, + |_dst, _size, output_pos, _block_to_flush, _end| { + ZSTDMT_streamFlushResult { + result: 0, + outputPos: output_pos, + } + }, + |_context, _input, _projection| panic!("no synchronization scan expected"), + ) + }; + + assert_eq!(create_call, Some((2, ZSTD_E_FLUSH))); + assert_eq!(result.result, 0); + assert_eq!(result.inBuffFilled, 0); + } + #[test] fn outer_scheduler_forces_flush_before_end_when_input_remains() { let input_bytes = [7u8, 8, 9, 10, 11]; @@ -8330,7 +8420,10 @@ mod tests { |_projection| panic!("input range should already be available"), |src_size, end| { create_end = Some((src_size, end)); - 0 + ZSTDMT_streamCreateJobResult { + result: 0, + inBuffActive: 0, + } }, |_dst, _size, output_pos, block_to_flush, end| { flush_block = Some(block_to_flush); @@ -8381,7 +8474,10 @@ mod tests { |_projection| panic!("a ready job must skip input acquisition"), |src_size, end| { create_call = Some((src_size, end)); - 0 + ZSTDMT_streamCreateJobResult { + result: 0, + inBuffActive: 0, + } }, |_dst, _size, output_pos, block_to_flush, end| { flush_call = Some((block_to_flush, end)); @@ -8425,7 +8521,10 @@ mod tests { output, ZSTD_E_CONTINUE, |_projection| panic!("input range should already be available"), - |_src_size, _end| ERROR(ZstdErrorCode::DstSizeTooSmall), + |_src_size, _end| ZSTDMT_streamCreateJobResult { + result: ERROR(ZstdErrorCode::DstSizeTooSmall), + inBuffActive: 1, + }, |_dst, _size, output_pos, _block_to_flush, _end| { flush_called = true; ZSTDMT_streamFlushResult { @@ -8444,6 +8543,7 @@ mod tests { assert_eq!(result.result, ERROR(ZstdErrorCode::DstSizeTooSmall)); assert_eq!(result.inputPos, round_buffer.len()); assert_eq!(result.outputPos, 0); + assert_eq!(result.inBuffFilled, round_buffer.len()); assert!(!flush_called); }