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

feat(mt): move frame progression job aggregation into Rust

Browse Source 
Move the multithreaded frame-progression ring scan into Rust so the Rust
scheduler owns job-ID masking, ready-job range handling, error-output
normalization, and active-worker accounting. Keep C-owned job descriptors and
mutex-protected snapshots behind the narrow ZSTDMT_projectJob callback, which
preserves the private synchronization boundary while making the aggregation
policy directly testable in Rust.

Test Plan:
- cargo fmt --manifest-path rust/Cargo.toml -- --check
- ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml --lib (691 passed)
- ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings
- ulimit -v 41943040 && MAKEFLAGS=-j1 make -B -C programs -j1 zstd
- ulimit -v 41943040 && MAKEFLAGS=-j1 make -C tests -j1 test-zstream ZSTREAM_TESTTIME=-T1s (84 tests and both short fuzz rounds passed)
This commit is contained in:
ddidderr
2026-07-19 18:53:08 +02:00
parent fec11ae7d0
commit adaae03552
3 changed files with 141 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstdmt_compress.c
+10 -21
View File
@@ -460,6 +460,11 @@ ZSTD_frameProgression ZSTDMT_rust_frameProgression(
ZSTD_frameProgression ZSTDMT_rust_frameProgressionAddJob(
ZSTD_frameProgression progression, size_t srcSize, size_t consumed,
size_t produced, size_t flushed);
ZSTD_frameProgression ZSTDMT_rust_frameProgressionWithJobs(
unsigned long long consumed, size_t inBuffFilled,
unsigned long long produced, unsigned currentJobID,
unsigned doneJobID, unsigned nextJobID, unsigned jobReady,
unsigned jobIDMask, void* opaque, ZSTDMT_jobProjectionFn projectJob);
typedef int (*ZSTDMT_jobTableInitFn)(void* jobTable, unsigned nbJobs,
size_t jobSize);
typedef void (*ZSTDMT_jobTableDestroyFn)(void* jobTable, unsigned nbJobs,
@@ -1523,28 +1528,12 @@ void ZSTDMT_updateCParams_whileCompressing(ZSTDMT_CCtx* mtctx, const ZSTD_CCtx_p
* Note : mutex will be acquired during statistics collection inside workers. */
ZSTD_frameProgression ZSTDMT_getFrameProgression(ZSTDMT_CCtx* mtctx)
{
ZSTD_frameProgression fps = ZSTDMT_rust_frameProgression(
mtctx->consumed, mtctx->inBuff.filled, mtctx->produced,
mtctx->nextJobID);
ZSTD_frameProgression const fps = ZSTDMT_rust_frameProgressionWithJobs(
mtctx->consumed, mtctx->inBuff.filled, mtctx->produced,
mtctx->nextJobID, mtctx->doneJobID, mtctx->nextJobID,
mtctx->jobReady, mtctx->jobIDMask, mtctx,
ZSTDMT_projectJob);
DEBUGLOG(5, "ZSTDMT_getFrameProgression");
{ unsigned jobNb;
unsigned lastJobNb = mtctx->nextJobID + mtctx->jobReady; assert(mtctx->jobReady <= 1);
DEBUGLOG(6, "ZSTDMT_getFrameProgression: jobs: from %u to <%u (jobReady:%u)",
mtctx->doneJobID, lastJobNb, mtctx->jobReady);
for (jobNb = mtctx->doneJobID ; jobNb < lastJobNb ; jobNb++) {
unsigned const wJobID = jobNb & mtctx->jobIDMask;
ZSTDMT_jobDescription* jobPtr = &mtctx->jobs[wJobID];
ZSTD_pthread_mutex_lock(&jobPtr->job_mutex);
{ size_t const cResult = jobPtr->cSize;
size_t const produced = ZSTD_isError(cResult) ? 0 : cResult;
size_t const flushed = ZSTD_isError(cResult) ? 0 : jobPtr->dstFlushed;
assert(flushed <= produced);
fps = ZSTDMT_rust_frameProgressionAddJob(
fps, jobPtr->src.size, jobPtr->consumed, produced, flushed);
}
ZSTD_pthread_mutex_unlock(&mtctx->jobs[wJobID].job_mutex);
}
}
return fps;
}
rust/README.md
+5 -2
View File
@@ -95,7 +95,9 @@ zstd ABI:
private to C while Rust owns input-retention scans, reusable input-range
overlap decisions, outer scheduling and end-directive adjustments,
job-creation decisions, compression-job stage sequencing and error flow,
and pending-output decisions through scalar job projections.
pending-output decisions through scalar job projections, and frame
progression's job-ring scan, error normalization, and active-worker
accounting.
- Dictionary support
- `zstd_ddict` owns, loads, copies, and references decode dictionaries.
- Legacy decoding
@@ -145,7 +147,8 @@ shared- and separate-destination multi-file compression schedulers,
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, public sequence-API
compression-job stage sequencing and error flow, MT frame-progression job
aggregation, 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,
rust/src/zstdmt_compress.rs
+126
View File
@@ -1957,6 +1957,43 @@ fn frame_progression_add_job(
progression
}
#[inline]
fn frame_progression_with_jobs<F>(
consumed: u64,
in_buff_filled: usize,
produced: u64,
current_job_id: c_uint,
done_job_id: c_uint,
next_job_id: c_uint,
job_ready: c_uint,
job_id_mask: c_uint,
mut project_job: F,
) -> ZSTD_frameProgression
where
F: FnMut(c_uint) -> ZSTDMT_jobProjection,
{
let mut progression = frame_progression(consumed, in_buff_filled, produced, current_job_id);
debug_assert!(job_ready <= 1);
let last_job_id = next_job_id.wrapping_add(job_ready);
for job_id in done_job_id..last_job_id {
let projection = project_job(job_id & job_id_mask);
let (produced, flushed) = if ERR_isError(projection.cSize) {
(0, 0)
} else {
(projection.cSize, projection.dstFlushed)
};
debug_assert!(flushed <= produced);
progression = frame_progression_add_job(
progression,
projection.srcSize,
projection.consumed,
produced,
flushed,
);
}
progression
}
/// Construct the base MT frame progression from C-owned scalar state.
#[no_mangle]
pub extern "C" fn ZSTDMT_rust_frameProgression(
@@ -1980,6 +2017,42 @@ pub extern "C" fn ZSTDMT_rust_frameProgressionAddJob(
frame_progression_add_job(progression, src_size, consumed, produced, flushed)
}
/// Aggregate the base MT progression with mutex-protected C job snapshots.
/// Rust owns only the ring scan and normalization policy; C retains the job
/// descriptor layout and takes each descriptor mutex in `projectJob`.
#[no_mangle]
pub unsafe extern "C" fn ZSTDMT_rust_frameProgressionWithJobs(
consumed: u64,
in_buff_filled: usize,
produced: u64,
current_job_id: c_uint,
done_job_id: c_uint,
next_job_id: c_uint,
job_ready: c_uint,
job_id_mask: c_uint,
opaque: *mut c_void,
project_job: Option<ZSTDMT_jobProjectionFn>,
) -> ZSTD_frameProgression {
let Some(project_job) = project_job else {
return frame_progression(consumed, in_buff_filled, produced, current_job_id);
};
frame_progression_with_jobs(
consumed,
in_buff_filled,
produced,
current_job_id,
done_job_id,
next_job_id,
job_ready,
job_id_mask,
|job_id| {
let mut projection = ZSTDMT_jobProjection::default();
unsafe { project_job(opaque, job_id, &mut projection) };
projection
},
)
}
#[inline]
unsafe fn rolling_hash_append(mut hash: u64, input: *const u8, size: usize) -> u64 {
for pos in 0..size {
@@ -4469,6 +4542,59 @@ mod tests {
);
}
#[test]
fn frame_progression_with_jobs_masks_slots_and_normalizes_errors() {
let projections = [
ZSTDMT_jobProjection {
consumed: 50,
cSize: 8,
srcSize: 50,
dstFlushed: 8,
..ZSTDMT_jobProjection::default()
},
ZSTDMT_jobProjection {
consumed: 17,
cSize: 42,
srcSize: 80,
dstFlushed: 10,
..ZSTDMT_jobProjection::default()
},
ZSTDMT_jobProjection {
consumed: 30,
cSize: ERROR(ZstdErrorCode::SequenceProducerFailed),
srcSize: 30,
dstFlushed: 29,
..ZSTDMT_jobProjection::default()
},
ZSTDMT_jobProjection {
consumed: 5,
cSize: 7,
srcSize: 40,
dstFlushed: 6,
..ZSTDMT_jobProjection::default()
},
];
let mut calls = Vec::new();
let progression = frame_progression_with_jobs(100, 20, 300, 9, 5, 8, 1, 3, |job_id| {
calls.push(job_id);
projections[job_id as usize]
});
assert_eq!(calls, [1, 2, 3, 0]);
assert_eq!(
progression,
ZSTD_frameProgression {
ingested: 320,
consumed: 202,
produced: 357,
flushed: 324,
currentJobID: 9,
nbActiveWorkers: 2,
}
);
}
#[test]
fn input_data_in_use_skips_first_round_and_scans_oldest_ring_slot() {
let source_a = [0u8; 8];