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

feat(mt): move worker finish publication to Rust

Browse Source 
Replace the C worker finish callbacks that publish errors, final block sizes,
consumed bytes, and completion signals with a Rust-owned publication path. The
nested projection keeps the job counters and pthread operations private to C,
while Rust preserves the error/success ordering and size_t wrapping behavior.
The consumed-size assertion remains a C validation callback at the original
point before publication.

Test Plan:
- git diff --cached --check
- capped cargo check --tests
- capped cargo clippy --tests -- -A clippy::manual-bits -D warnings
- capped make -j1
- capped make -j1 -C tests test
- standalone hybrid cargo test remains unavailable because its link lacks the
  pre-existing ZSTD_rust_dctx_trace_view, ZSTD_rust_dctx_view, and
  ZSTD_rust_block_context_init symbols
This commit is contained in:
ddidderr
2026-07-21 22:27:57 +02:00
parent 0b18e73d4b
commit 6b8051b98a
2 changed files with 262 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstdmt_compress.c
+59 -59
View File
@@ -300,16 +300,48 @@ typedef char ZSTDMT_compression_job_progress_projection_layout[
void ZSTDMT_rust_compressionJobProgress(
void* opaque, size_t cSize, size_t consumed);
typedef struct {
void* callbackContext;
size_t* cSize;
size_t* consumed;
size_t srcSize;
ZSTDMT_compressionJobVoidFn lock;
ZSTDMT_compressionJobVoidFn signal;
ZSTDMT_compressionJobVoidFn unlock;
ZSTDMT_compressionJobSizeFn validateConsumed;
} ZSTDMT_RustCompressionJobFinishPublicationProjection;
typedef char ZSTDMT_compression_job_finish_publication_projection_layout[
(offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection,
callbackContext) == 0
&& offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, cSize)
== sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection,
consumed)
== sizeof(void*) + sizeof(size_t)
&& offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection,
srcSize)
== sizeof(void*) + 2 * sizeof(size_t)
&& offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, lock)
== sizeof(void*) + 3 * sizeof(size_t)
&& offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, signal)
== sizeof(void*) + 3 * sizeof(size_t) + sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection,
unlock)
== sizeof(void*) + 3 * sizeof(size_t) + 2 * sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection,
validateConsumed)
== sizeof(void*) + 3 * sizeof(size_t) + 3 * sizeof(void*)
&& sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection)
== sizeof(void*) + 3 * sizeof(size_t) + 4 * sizeof(void*))
? 1 : -1];
typedef struct {
void* callbackContext;
size_t srcSize;
ZSTDMT_compressionJobVoidFn ensureFinished;
ZSTDMT_compressionJobSizeFn publishError;
ZSTDMT_compressionJobSizeFn publishLastBlockSize;
ZSTDMT_RustCompressionJobFinishPublicationProjection publication;
ZSTDMT_compressionJobVoidFn releaseSeq;
ZSTDMT_compressionJobVoidFn releaseCCtx;
ZSTDMT_compressionJobSizeFn publishConsumed;
ZSTDMT_compressionJobVoidFn signal;
} ZSTDMT_RustCompressionJobFinishProjection;
typedef char ZSTDMT_compression_job_finish_projection_layout[
(offsetof(ZSTDMT_RustCompressionJobFinishProjection, callbackContext) == 0
@@ -317,19 +349,17 @@ typedef char ZSTDMT_compression_job_finish_projection_layout[
== sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, ensureFinished)
== 2 * sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, publishError)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, publication)
== 3 * sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, publishLastBlockSize)
== 4 * sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, releaseSeq)
== 5 * sizeof(void*)
== 3 * sizeof(void*)
+ sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, releaseCCtx)
== 6 * sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, publishConsumed)
== 7 * sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, signal)
== 8 * sizeof(void*)
&& sizeof(ZSTDMT_RustCompressionJobFinishProjection) == 9 * sizeof(void*))
== 4 * sizeof(void*)
+ sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection)
&& sizeof(ZSTDMT_RustCompressionJobFinishProjection)
== 5 * sizeof(void*)
+ sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection))
? 1 : -1];
typedef struct {
@@ -2243,28 +2273,6 @@ static void ZSTDMT_compressionJobEnsureFinished(void* opaque)
DEBUGLOG(5, "Finished with source: %zx", (size_t)job->src.start);
}
static void ZSTDMT_compressionJobPublishError(void* opaque, size_t error)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex);
job->cSize = error;
ZSTD_pthread_mutex_unlock(&job->job_mutex);
}
static void ZSTDMT_compressionJobPublishLastBlockSize(void* opaque, size_t lastCBlockSize)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex);
job->cSize += lastCBlockSize;
ZSTD_pthread_mutex_unlock(&job->job_mutex);
}
static void ZSTDMT_compressionJobReleaseSeq(void* opaque)
{
ZSTDMT_compressionJobState* const state =
@@ -2283,27 +2291,11 @@ static void ZSTDMT_compressionJobReleaseCCtx(void* opaque)
ZSTDMT_releaseCCtx(job->cctxPool, state->cctx);
}
static void ZSTDMT_compressionJobPublishConsumed(void* opaque, size_t srcSize)
static void ZSTDMT_compressionJobValidateConsumed(void* opaque, size_t srcSize)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)opaque;
assert(srcSize == job->src.size);
ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex);
job->consumed = srcSize; /* when job->consumed == job->src.size , compression job is presumed completed */
ZSTD_pthread_mutex_unlock(&job->job_mutex);
}
static void ZSTDMT_compressionJobSignal(void* opaque)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex);
ZSTD_pthread_cond_signal(&job->job_cond);
ZSTD_pthread_mutex_unlock(&job->job_mutex);
}
/* ZSTDMT_compressionJob() is a POOL_function type. Rust owns the stage
@@ -2313,6 +2305,7 @@ static void ZSTDMT_compressionJob(void* jobDescription)
ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription;
ZSTDMT_compressionJobState state;
ZSTDMT_RustCompressionJobSequenceState sequenceState;
ZSTDMT_RustCompressionJobFinishPublicationProjection publication;
ZSTDMT_RustCompressionJobFinishProjection finishProjection;
ZSTDMT_RustCompressionJobProjection projection;
@@ -2331,16 +2324,23 @@ static void ZSTDMT_compressionJob(void* jobDescription)
&state.frameHeaderState,
&sequenceState
};
publication = (ZSTDMT_RustCompressionJobFinishPublicationProjection){
job,
&job->cSize,
&job->consumed,
job->src.size,
ZSTDMT_compressionJobProgressLock,
ZSTDMT_compressionJobProgressSignal,
ZSTDMT_compressionJobProgressUnlock,
ZSTDMT_compressionJobValidateConsumed
};
finishProjection = (ZSTDMT_RustCompressionJobFinishProjection){
&state,
job->src.size,
ZSTDMT_compressionJobEnsureFinished,
ZSTDMT_compressionJobPublishError,
ZSTDMT_compressionJobPublishLastBlockSize,
publication,
ZSTDMT_compressionJobReleaseSeq,
ZSTDMT_compressionJobReleaseCCtx,
ZSTDMT_compressionJobPublishConsumed,
ZSTDMT_compressionJobSignal
ZSTDMT_compressionJobReleaseCCtx
};
ZSTDMT_rust_compressionJob(
rust/src/zstdmt_compress.rs
+203 -91
View File
@@ -755,24 +755,66 @@ pub unsafe extern "C" fn ZSTDMT_rust_compressionJobBegin(
)
}
/// Live fields and synchronization callbacks for worker-job finish
/// publication. The private job descriptor and pthread objects remain in C;
/// Rust owns the counter updates and completion signal order.
#[repr(C)]
#[derive(Clone, Copy)]
pub struct ZSTDMT_compressionJobFinishPublicationProjection {
pub callbackContext: *mut c_void,
pub cSize: *mut usize,
pub consumed: *mut usize,
pub srcSize: usize,
pub lock: Option<ZSTDMT_compressionJobVoidFn>,
pub signal: Option<ZSTDMT_compressionJobVoidFn>,
pub unlock: Option<ZSTDMT_compressionJobVoidFn>,
pub validateConsumed: Option<ZSTDMT_compressionJobSizeFn>,
}
const _: () = {
assert!(
offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, callbackContext) == 0
);
assert!(
offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, cSize)
== size_of::<*mut c_void>()
);
assert!(
offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, consumed)
== size_of::<*mut c_void>() + size_of::<usize>()
);
assert!(
offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, srcSize)
== size_of::<*mut c_void>() + 2 * size_of::<usize>()
);
assert!(offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, lock)
== size_of::<*mut c_void>() + 3 * size_of::<usize>());
assert!(offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, signal)
== size_of::<*mut c_void>() + 3 * size_of::<usize>() + size_of::<usize>());
assert!(offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, unlock)
== size_of::<*mut c_void>() + 3 * size_of::<usize>() + 2 * size_of::<usize>());
assert!(
offset_of!(ZSTDMT_compressionJobFinishPublicationProjection, validateConsumed)
== size_of::<*mut c_void>() + 3 * size_of::<usize>() + 3 * size_of::<usize>()
);
assert!(size_of::<ZSTDMT_compressionJobFinishPublicationProjection>()
== size_of::<*mut c_void>() + 3 * size_of::<usize>() + 4 * size_of::<usize>());
};
/// C-owned leaves for the worker-job finish path.
///
/// Rust owns the error/success branch and callback order. C keeps the job
/// descriptor, pthread objects, private pools, and serial state behind these
/// callbacks; the source size is the only scalar needed for consumed-size
/// publication.
/// descriptor, pthread objects, private pools, and serial state behind the
/// remaining callbacks; publication state crosses through one live projection.
#[repr(C)]
#[derive(Clone, Copy)]
pub struct ZSTDMT_compressionJobFinishProjection {
pub callbackContext: *mut c_void,
pub srcSize: usize,
pub ensureFinished: Option<ZSTDMT_compressionJobVoidFn>,
pub publishError: Option<ZSTDMT_compressionJobSizeFn>,
pub publishLastBlockSize: Option<ZSTDMT_compressionJobSizeFn>,
pub publication: ZSTDMT_compressionJobFinishPublicationProjection,
pub releaseSeq: Option<ZSTDMT_compressionJobVoidFn>,
pub releaseCCtx: Option<ZSTDMT_compressionJobVoidFn>,
pub publishConsumed: Option<ZSTDMT_compressionJobSizeFn>,
pub signal: Option<ZSTDMT_compressionJobVoidFn>,
}
const _: () = {
@@ -781,25 +823,21 @@ const _: () = {
assert!(
offset_of!(ZSTDMT_compressionJobFinishProjection, ensureFinished) == 2 * size_of::<usize>()
);
assert!(offset_of!(ZSTDMT_compressionJobFinishProjection, publication)
== 3 * size_of::<usize>());
assert!(
offset_of!(ZSTDMT_compressionJobFinishProjection, publishError) == 3 * size_of::<usize>()
offset_of!(ZSTDMT_compressionJobFinishProjection, releaseSeq)
== 3 * size_of::<usize>()
+ size_of::<ZSTDMT_compressionJobFinishPublicationProjection>()
);
assert!(
offset_of!(ZSTDMT_compressionJobFinishProjection, publishLastBlockSize)
offset_of!(ZSTDMT_compressionJobFinishProjection, releaseCCtx)
== 4 * size_of::<usize>()
+ size_of::<ZSTDMT_compressionJobFinishPublicationProjection>()
);
assert!(
offset_of!(ZSTDMT_compressionJobFinishProjection, releaseSeq) == 5 * size_of::<usize>()
);
assert!(
offset_of!(ZSTDMT_compressionJobFinishProjection, releaseCCtx) == 6 * size_of::<usize>()
);
assert!(
offset_of!(ZSTDMT_compressionJobFinishProjection, publishConsumed)
== 7 * size_of::<usize>()
);
assert!(offset_of!(ZSTDMT_compressionJobFinishProjection, signal) == usize::BITS as usize);
assert!(size_of::<ZSTDMT_compressionJobFinishProjection>() == 9 * size_of::<usize>());
assert!(size_of::<ZSTDMT_compressionJobFinishProjection>()
== 5 * size_of::<usize>()
+ size_of::<ZSTDMT_compressionJobFinishPublicationProjection>());
};
pub type ZSTDMT_serialWaitForTurnFn = unsafe extern "C" fn(*mut c_void, c_uint) -> c_int;
@@ -2570,6 +2608,78 @@ pub unsafe extern "C" fn ZSTDMT_rust_createCompressionJob(
)
}
#[inline]
unsafe fn publish_finish_error(
publication: ZSTDMT_compressionJobFinishPublicationProjection,
error: usize,
) {
let (Some(lock), Some(unlock)) = (publication.lock, publication.unlock) else {
return;
};
if publication.cSize.is_null() {
return;
}
unsafe {
lock(publication.callbackContext);
*publication.cSize = error;
unlock(publication.callbackContext);
}
}
#[inline]
unsafe fn publish_finish_last_block_size(
publication: ZSTDMT_compressionJobFinishPublicationProjection,
last_block_size: usize,
) {
let (Some(lock), Some(unlock)) = (publication.lock, publication.unlock) else {
return;
};
if publication.cSize.is_null() {
return;
}
unsafe {
lock(publication.callbackContext);
*publication.cSize = (*publication.cSize).wrapping_add(last_block_size);
unlock(publication.callbackContext);
}
}
#[inline]
unsafe fn publish_finish_consumed(
publication: ZSTDMT_compressionJobFinishPublicationProjection,
src_size: usize,
) {
debug_assert_eq!(publication.srcSize, src_size);
if let Some(validate_consumed) = publication.validateConsumed {
unsafe { validate_consumed(publication.callbackContext, src_size) };
}
let (Some(lock), Some(unlock)) = (publication.lock, publication.unlock) else {
return;
};
if publication.consumed.is_null() {
return;
}
unsafe {
lock(publication.callbackContext);
*publication.consumed = src_size;
unlock(publication.callbackContext);
}
}
#[inline]
unsafe fn signal_finish(publication: ZSTDMT_compressionJobFinishPublicationProjection) {
let (Some(lock), Some(signal), Some(unlock)) =
(publication.lock, publication.signal, publication.unlock)
else {
return;
};
unsafe {
lock(publication.callbackContext);
signal(publication.callbackContext);
unlock(publication.callbackContext);
}
}
/// Finish a worker job with Rust-owned branch and callback ordering.
///
/// Error publication happens before serial completion and resource release,
@@ -2640,22 +2750,10 @@ fn compression_job_with<A, P, S, B, Q, H, C, T>(
C: FnMut(c_uint) -> ZSTDMT_chunkProcessResult,
T: FnMut(),
{
let (
Some(ensure_finished),
Some(publish_error),
Some(publish_last_block_size),
Some(release_seq),
Some(release_cctx),
Some(publish_consumed),
Some(signal),
) = (
let (Some(ensure_finished), Some(release_seq), Some(release_cctx)) = (
finish.ensureFinished,
finish.publishError,
finish.publishLastBlockSize,
finish.releaseSeq,
finish.releaseCCtx,
finish.publishConsumed,
finish.signal,
)
else {
return;
@@ -2690,19 +2788,18 @@ fn compression_job_with<A, P, S, B, Q, H, C, T>(
if ERR_isError(error) {
last_block_size = 0;
}
let publication = finish.publication;
finish_compression_job_with(
error,
last_block_size,
finish.srcSize,
|error| unsafe { publish_error(finish.callbackContext, error) },
|error| unsafe { publish_finish_error(publication, error) },
|| unsafe { ensure_finished(finish.callbackContext) },
|| unsafe { release_seq(finish.callbackContext) },
|| unsafe { release_cctx(finish.callbackContext) },
|last_block_size| unsafe {
publish_last_block_size(finish.callbackContext, last_block_size)
},
|src_size| unsafe { publish_consumed(finish.callbackContext, src_size) },
|| unsafe { signal(finish.callbackContext) },
|last_block_size| unsafe { publish_finish_last_block_size(publication, last_block_size) },
|src_size| unsafe { publish_finish_consumed(publication, src_size) },
|| unsafe { signal_finish(publication) },
);
}
@@ -6287,9 +6384,9 @@ mod tests {
#[derive(Default)]
struct MockCompressionJob {
events: Vec<&'static str>,
errors: Vec<usize>,
finished: Vec<usize>,
consumed: Vec<usize>,
c_size: usize,
consumed_size: usize,
expected_src_size: usize,
}
struct WaitForLdmTestContext {
@@ -6659,25 +6756,6 @@ mod tests {
record_compression_job_finish_event(context, "ensure");
}
unsafe extern "C" fn compression_job_finish_test_error(context: *mut c_void, error: usize) {
unsafe {
let state = &*context.cast::<RefCell<MockCompressionJob>>();
state.borrow_mut().errors.push(error);
}
record_compression_job_finish_event(context, "error");
}
unsafe extern "C" fn compression_job_finish_test_last_block(
context: *mut c_void,
last_block_size: usize,
) {
unsafe {
let state = &*context.cast::<RefCell<MockCompressionJob>>();
state.borrow_mut().finished.push(last_block_size);
}
record_compression_job_finish_event(context, "last-block");
}
unsafe extern "C" fn compression_job_finish_test_release_seq(context: *mut c_void) {
record_compression_job_finish_event(context, "release-seq");
}
@@ -6686,35 +6764,57 @@ mod tests {
record_compression_job_finish_event(context, "release-cctx");
}
unsafe extern "C" fn compression_job_finish_test_consumed(
context: *mut c_void,
src_size: usize,
) {
unsafe {
let state = &*context.cast::<RefCell<MockCompressionJob>>();
state.borrow_mut().consumed.push(src_size);
}
record_compression_job_finish_event(context, "consumed");
unsafe extern "C" fn compression_job_finish_test_lock(context: *mut c_void) {
record_compression_job_finish_event(context, "lock");
}
unsafe extern "C" fn compression_job_finish_test_signal(context: *mut c_void) {
record_compression_job_finish_event(context, "signal");
}
unsafe extern "C" fn compression_job_finish_test_unlock(context: *mut c_void) {
record_compression_job_finish_event(context, "unlock");
}
unsafe extern "C" fn compression_job_finish_test_validate(
context: *mut c_void,
src_size: usize,
) {
let state = unsafe { &*context.cast::<RefCell<MockCompressionJob>>() };
assert_eq!(state.borrow().expected_src_size, src_size);
record_compression_job_finish_event(context, "validate");
}
fn compression_job_finish_test_projection(
state: &Rc<RefCell<MockCompressionJob>>,
src_size: usize,
) -> ZSTDMT_compressionJobFinishProjection {
let state_cell = Rc::as_ptr(state);
let state_value = unsafe { (*state_cell).as_ptr() };
let (c_size, consumed) = unsafe {
(*state_value).expected_src_size = src_size;
(
&mut (*state_value).c_size as *mut usize,
&mut (*state_value).consumed_size as *mut usize,
)
};
let callback_context = state_cell.cast_mut().cast();
ZSTDMT_compressionJobFinishProjection {
callbackContext: Rc::as_ptr(state).cast_mut().cast(),
callbackContext: callback_context,
srcSize: src_size,
ensureFinished: Some(compression_job_finish_test_ensure),
publishError: Some(compression_job_finish_test_error),
publishLastBlockSize: Some(compression_job_finish_test_last_block),
publication: ZSTDMT_compressionJobFinishPublicationProjection {
callbackContext: callback_context,
cSize: c_size,
consumed,
srcSize: src_size,
lock: Some(compression_job_finish_test_lock),
signal: Some(compression_job_finish_test_signal),
unlock: Some(compression_job_finish_test_unlock),
validateConsumed: Some(compression_job_finish_test_validate),
},
releaseSeq: Some(compression_job_finish_test_release_seq),
releaseCCtx: Some(compression_job_finish_test_release_cctx),
publishConsumed: Some(compression_job_finish_test_consumed),
signal: Some(compression_job_finish_test_signal),
}
}
@@ -6998,14 +7098,18 @@ mod tests {
"ensure",
"release-seq",
"release-cctx",
"last-block",
"consumed",
"lock",
"unlock",
"validate",
"lock",
"unlock",
"lock",
"signal",
"unlock",
]
);
assert!(state.errors.is_empty());
assert_eq!(state.finished, vec![7]);
assert_eq!(state.consumed, vec![19]);
assert_eq!(state.c_size, 7);
assert_eq!(state.consumed_size, 19);
}
#[test]
@@ -7041,17 +7145,21 @@ mod tests {
state.events,
vec![
"acquire",
"error",
"lock",
"unlock",
"ensure",
"release-seq",
"release-cctx",
"consumed",
"validate",
"lock",
"unlock",
"lock",
"signal",
"unlock",
]
);
assert_eq!(state.errors, vec![expected_error]);
assert!(state.finished.is_empty());
assert_eq!(state.consumed, vec![11]);
assert_eq!(state.c_size, expected_error);
assert_eq!(state.consumed_size, 11);
}
#[test]
@@ -7324,17 +7432,21 @@ mod tests {
"apply",
"header",
"compress",
"error",
"lock",
"unlock",
"ensure",
"release-seq",
"release-cctx",
"consumed",
"validate",
"lock",
"unlock",
"lock",
"signal",
"unlock",
]
);
assert_eq!(state.errors, vec![expected_error]);
assert!(state.finished.is_empty());
assert_eq!(state.consumed, vec![23]);
assert_eq!(state.c_size, expected_error);
assert_eq!(state.consumed_size, 23);
}
fn init_projection() -> ZSTDMT_initCStreamProjection {