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

fix(compress): preserve post-init MT stream dispatch

Browse Source 
Transparent stream initialization can resolve an unknown source size to the
multithreaded path and update appliedParams.nbWorkers. The Rust stream2
fallback previously captured that value before initialization, then entered the
serial generic adapter while the context stage was still created. Unknown-size
stdin compression consequently returned StageWrong instead of producing a
frame.

Keep the worker-mode projection live through initialization so Rust observes
the applied value at the existing MT-versus-serial decision point. The ABI
layout assertions now cover the pointer projection, and a focused regression
test verifies that an initialization callback which selects MT dispatches the
MT step and preserves the expected callback order.

Test Plan:
- capped cargo +nightly fmt for the Rust workspace -- passed
- capped root and CLI cargo clippy with -D warnings -- passed
- capped serial make -j1 for single-threaded library, MT library, and CLI -- passed without the prior pointer-sign warning
- capped stdin compression/decompression round trip for unknown-size input -- passed
- capped original suite, including CLI tests, native tests, fuzzer, and both zstream phases -- passed; the final signedness-only ABI spelling cleanup was followed by a clean capped rebuild
- targeted root cargo test attempted but remains blocked at link time by pre-existing decompression bridge symbols (ZSTD_rust_dctx_trace_view, ZSTD_rust_dctx_view, and ZSTD_rust_block_context_init)
This commit is contained in:
ddidderr
2026-07-21 13:37:58 +02:00
parent b394cae70f
commit 7d7b18d60a
2 changed files with 101 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+7 -9
View File
@@ -1064,7 +1064,8 @@ typedef struct {
int inBufferMode;
int outBufferMode;
int format;
int nbWorkers;
/* Transparent init may resolve a different worker mode for unknown sizes. */
const int* nbWorkers;
} ZSTD_rust_compressStream2Fields;
typedef struct {
ZSTD_rust_compressStream2Init_f initCompressStream2;
@@ -1112,10 +1113,11 @@ typedef char ZSTD_rust_compress_stream2_fields_layout[
&& offsetof(ZSTD_rust_compressStream2Fields, format)
== 10 * sizeof(void*) + sizeof(unsigned long long) + 2 * sizeof(int)
&& offsetof(ZSTD_rust_compressStream2Fields, nbWorkers)
== 10 * sizeof(void*) + sizeof(unsigned long long) + 3 * sizeof(int)
== ((10 * sizeof(void*) + sizeof(unsigned long long) + 3 * sizeof(int)
+ sizeof(void*) - 1) / sizeof(void*)) * sizeof(void*)
&& sizeof(ZSTD_rust_compressStream2Fields)
== ((10 * sizeof(void*) + sizeof(unsigned long long) + 4 * sizeof(int)
+ sizeof(void*) - 1) / sizeof(void*)) * sizeof(void*))
== offsetof(ZSTD_rust_compressStream2Fields, nbWorkers)
+ sizeof(const int*))
? 1 : -1];
typedef char ZSTD_rust_compress_stream2_callbacks_layout[
(offsetof(ZSTD_rust_compressStream2Callbacks, initCompressStream2) == 0
@@ -8922,11 +8924,7 @@ size_t ZSTD_compressStream2_c( ZSTD_CCtx* cctx,
(int)cctx->requestedParams.inBufferMode,
(int)cctx->requestedParams.outBufferMode,
(int)cctx->requestedParams.format,
#ifdef ZSTD_MULTITHREAD
(int)cctx->appliedParams.nbWorkers,
#else
0,
#endif
&cctx->appliedParams.nbWorkers,
};
ZSTD_rust_compressStream2Callbacks const callbacks = {
ZSTD_rust_compressStream2_init,
rust/src/zstd_compress.rs
+94 -18
View File
@@ -922,7 +922,8 @@ pub struct ZSTD_rust_compressStream2Fields {
in_buffer_mode: c_int,
out_buffer_mode: c_int,
format: c_int,
nb_workers: c_int,
// Transparent init may update the applied worker mode before dispatch.
nb_workers: *const c_int,
}
#[repr(C)]
@@ -988,14 +989,8 @@ const _: () = {
== 10 * size_of::<usize>() + size_of::<u64>() + 2 * size_of::<c_int>()
);
assert!(
offset_of!(ZSTD_rust_compressStream2Fields, nb_workers)
== 10 * size_of::<usize>() + size_of::<u64>() + 3 * size_of::<c_int>()
);
assert!(
size_of::<ZSTD_rust_compressStream2Fields>()
== (10 * size_of::<usize>() + size_of::<u64>() + 4 * size_of::<c_int>())
.div_ceil(size_of::<usize>())
* size_of::<usize>()
offset_of!(ZSTD_rust_compressStream2Fields, nb_workers) + size_of::<*const c_int>()
== size_of::<ZSTD_rust_compressStream2Fields>()
);
assert!(size_of::<CompressStream2InitFn>() == size_of::<usize>());
assert!(size_of::<CompressStream2CheckBufferFn>() == size_of::<usize>());
@@ -1069,7 +1064,7 @@ unsafe fn compress_stream2_fallback(state: &ZSTD_rust_compressStream2State) -> u
let Some(set_buffer_expectations) = callbacks.set_buffer_expectations else {
return ERROR(ZstdErrorCode::Generic);
};
if fields.callback_context.is_null() {
if fields.callback_context.is_null() || fields.nb_workers.is_null() {
return ERROR(ZstdErrorCode::Generic);
}
if output.pos > output.size {
@@ -1183,7 +1178,7 @@ unsafe fn compress_stream2_fallback(state: &ZSTD_rust_compressStream2State) -> u
return stability_result;
}
if fields.nb_workers > 0 {
if unsafe { *fields.nb_workers } > 0 {
if !fields.c_params_changed.is_null() {
let c_params_changed = unsafe { &mut *fields.c_params_changed };
if *c_params_changed != 0 {
@@ -22928,6 +22923,8 @@ mod tests {
struct Stream2FallbackProbe {
events: Vec<&'static str>,
compress_result: usize,
init_nb_workers: c_int,
nb_workers: c_int,
}
unsafe extern "C" fn stream2_test_init(
@@ -22935,9 +22932,9 @@ mod tests {
_end_op: c_int,
_input_size: usize,
) -> usize {
unsafe { &mut *context.cast::<Stream2FallbackProbe>() }
.events
.push("init");
let probe = unsafe { &mut *context.cast::<Stream2FallbackProbe>() };
probe.events.push("init");
probe.nb_workers = probe.init_nb_workers;
0
}
@@ -22977,6 +22974,27 @@ mod tests {
probe.compress_result
}
unsafe extern "C" fn stream2_test_mt_step(
context: *mut c_void,
output: *mut ZSTD_outBuffer,
input: *mut ZSTD_inBuffer,
_end_op: c_int,
) -> usize {
let probe = unsafe { &mut *context.cast::<Stream2FallbackProbe>() };
probe.events.push("mt");
unsafe {
(*input).pos = (*input).size;
(*output).pos += 3;
}
0
}
unsafe extern "C" fn stream2_test_mt_trace(_context: *mut c_void) {}
unsafe extern "C" fn stream2_test_mt_reset(_context: *mut c_void) -> usize {
0
}
fn stream2_test_state<'a>(
probe: &'a mut Stream2FallbackProbe,
output: &'a mut ZSTD_outBuffer,
@@ -23003,7 +23021,7 @@ mod tests {
in_buffer_mode: ZSTD_BM_BUFFERED,
out_buffer_mode: ZSTD_BM_BUFFERED,
format: ZSTD_F_ZSTD1,
nb_workers: 0,
nb_workers: &probe.nb_workers,
});
let callbacks = Box::new(ZSTD_rust_compressStream2Callbacks {
init_compress_stream2: Some(stream2_test_init),
@@ -23011,9 +23029,9 @@ mod tests {
set_buffer_expectations: Some(stream2_test_set),
compress_stream: Some(stream2_test_compress),
update_mt_cparams: None,
mt_step: None,
mt_trace: None,
mt_reset: None,
mt_step: Some(stream2_test_mt_step),
mt_trace: Some(stream2_test_mt_trace),
mt_reset: Some(stream2_test_mt_reset),
});
ZSTD_rust_compressStream2State {
fields: Box::into_raw(fields),
@@ -23027,6 +23045,8 @@ mod tests {
let mut probe = Stream2FallbackProbe {
events: Vec::new(),
compress_result: 7,
init_nb_workers: 0,
nb_workers: 0,
};
let mut output_bytes = [0u8; 32];
let input_bytes = [1u8; 8];
@@ -23079,6 +23099,8 @@ mod tests {
let mut probe = Stream2FallbackProbe {
events: Vec::new(),
compress_result: 0,
init_nb_workers: 0,
nb_workers: 0,
};
let input_bytes = [1u8; 16];
let mut output = ZSTD_outBuffer {
@@ -23134,4 +23156,58 @@ mod tests {
drop(Box::from_raw(state.callbacks.cast_mut()));
}
}
#[test]
fn stream2_fallback_reads_worker_mode_after_initialization() {
let mut probe = Stream2FallbackProbe {
events: Vec::new(),
compress_result: 0,
init_nb_workers: 1,
nb_workers: 0,
};
let mut output_bytes = [0u8; 32];
let input_bytes = [1u8; 8];
let mut output = ZSTD_outBuffer {
dst: output_bytes.as_mut_ptr().cast(),
size: output_bytes.len(),
pos: 0,
};
let mut input = ZSTD_inBuffer {
src: input_bytes.as_ptr().cast(),
size: input_bytes.len(),
pos: 0,
};
let mut stage = ZSTD_CSTREAM_STAGE_INIT;
let mut stable_in_not_consumed = 0;
let mut expected_in_buffer = ZSTD_inBuffer {
src: ptr::null(),
size: 0,
pos: 0,
};
let mut simple_compress2_completed = 0;
let out_buff_content_size = 0;
let out_buff_flushed_size = 0;
let state = stream2_test_state(
&mut probe,
&mut output,
&mut input,
&mut stage,
&mut stable_in_not_consumed,
&mut expected_in_buffer,
&mut simple_compress2_completed,
&out_buff_content_size,
&out_buff_flushed_size,
);
let result = unsafe { ZSTD_rust_compressStream2(&state) };
assert_eq!(result, 0);
assert_eq!(probe.events, ["init", "check", "mt", "set"]);
assert_eq!(input.pos, input.size);
assert_eq!(output.pos, 3);
unsafe {
drop(Box::from_raw(state.fields.cast_mut()));
drop(Box::from_raw(state.callbacks.cast_mut()));
}
}
}