ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(mt): move input publication policy to Rust

Browse Source 
Keep the C-owned compression job callback responsible for private buffer and job-state mutation, but return the range-active decision alongside its result. Rust now owns the outer scheduler policy that publishes or suppresses the reusable input count after job creation, including the error path. The focused seam test covers the callback-clears-range case without exposing the MT context layout.

Test Plan: Not run in this atomic commit; the capped serial Rust, native, smoke, and original test-suite verification follows.
This commit is contained in:
ddidderr
2026-07-21 07:38:20 +02:00
parent c522906d07
commit 8373de7917
2 changed files with 144 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstdmt_compress.c
+30 -12
View File
@@ -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);
rust/src/zstdmt_compress.rs
+114 -14
View File
@@ -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::<usize>()
);
assert!(size_of::<ZSTDMT_streamCreateJobResult>() == 2 * size_of::<usize>());
};
/// 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<T, J, F, S>(
) -> 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);
}