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

feat(cli): move zstd frame result policy to Rust

Browse Source 
The default zstd frame loop was already implemented in Rust, but its
completion and error-result policy still lived in the C callback wrapper.
That left the C translation unit deciding when a decoded size was valid and
when a decoding or premature-end status had to become the historical frame
sentinel. Keep the exact operator-facing diagnostics and the private
asynchronous-resource pointers in C, while giving Rust ownership of the
status-to-result policy and callback ordering.

The new C/Rust policy record has compile-time layout assertions on both sides,
and the Rust fixture covers success, decode error, premature end, sentinel
results, and callback sequencing. This keeps the bridge explicit without
making Rust depend on the private dRess_t layout.

Test Plan:
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo check --manifest-path rust/Cargo.toml --tests
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings
- ulimit -v 41943040; make -j1
- ulimit -v 41943040; make -j1 -C tests test
- git diff --check
This commit is contained in:
ddidderr
2026-07-21 16:05:56 +02:00
parent cf8953eed2
commit bf2d73fcc0
2 changed files with 254 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
programs/fileio.c
+69 -37
View File
@@ -573,6 +573,42 @@ int FIO_rust_decompressZstdFrames(void* fCtx,
U64* decodedSize,
size_t* zstdError,
FIO_rust_frame_progress_fn progress);
typedef void (*FIO_rust_zstd_frame_decoding_error_fn)(void* opaque,
const char* srcFileName,
size_t zstdError);
typedef void (*FIO_rust_zstd_frame_premature_end_fn)(void* opaque,
const char* srcFileName);
typedef struct {
void* callbackContext;
void* fCtx;
void* dctx;
ReadPoolCtx_t* readCtx;
WritePoolCtx_t* writeCtx;
const char* srcFileName;
U64 alreadyDecoded;
FIO_rust_frame_progress_fn progress;
FIO_rust_zstd_frame_decoding_error_fn displayDecodingError;
FIO_rust_zstd_frame_premature_end_fn displayPrematureEnd;
} FIO_rust_zstd_frame_policy_state;
typedef char FIO_rust_zstd_frame_policy_state_layout[
(offsetof(FIO_rust_zstd_frame_policy_state, callbackContext) == 0
&& offsetof(FIO_rust_zstd_frame_policy_state, fCtx) == sizeof(void*)
&& offsetof(FIO_rust_zstd_frame_policy_state, dctx) == 2 * sizeof(void*)
&& offsetof(FIO_rust_zstd_frame_policy_state, readCtx) == 3 * sizeof(void*)
&& offsetof(FIO_rust_zstd_frame_policy_state, writeCtx) == 4 * sizeof(void*)
&& offsetof(FIO_rust_zstd_frame_policy_state, srcFileName) == 5 * sizeof(void*)
&& offsetof(FIO_rust_zstd_frame_policy_state, alreadyDecoded) == 6 * sizeof(void*)
&& offsetof(FIO_rust_zstd_frame_policy_state, progress) == 7 * sizeof(void*)
&& offsetof(FIO_rust_zstd_frame_policy_state, displayDecodingError)
== 8 * sizeof(void*)
&& offsetof(FIO_rust_zstd_frame_policy_state, displayPrematureEnd)
== 9 * sizeof(void*)
&& sizeof(FIO_rust_zstd_frame_policy_state) == 10 * sizeof(void*)
&& sizeof(FIO_rust_zstd_frame_decoding_error_fn) == sizeof(void*)
&& sizeof(FIO_rust_zstd_frame_premature_end_fn) == sizeof(void*))
? 1 : -1];
U64 FIO_rust_decompressZstdFramePolicy(
const FIO_rust_zstd_frame_policy_state* state);
enum {
FIO_RUST_DECOMPRESS_OK = 0,
FIO_RUST_DECOMPRESS_PASS_THROUGH = 1,
@@ -4177,38 +4213,6 @@ FIO_decompressZstdFrameProgress(void* const opaque,
}
}
static unsigned long long
FIO_decompressZstdFrames(FIO_ctx_t* const fCtx, dRess_t* ress,
const FIO_prefs_t* const prefs,
const char* srcFileName,
U64 alreadyDecoded) /* for multi-frames streams */
{
U64 decodedSize = 0;
size_t zstdError = 0;
int const status = FIO_rust_decompressZstdFrames(
fCtx, ress->dctx, ress->readCtx, ress->writeCtx, srcFileName,
alreadyDecoded, &decodedSize, &zstdError,
FIO_decompressZstdFrameProgress);
int const action = FIO_rust_decompressZstdFrameAction(status);
switch (action) {
case FIO_RUST_ZSTD_FRAME_ACTION_OK:
return decodedSize;
case FIO_RUST_ZSTD_FRAME_ACTION_DECODING_ERROR:
DISPLAYLEVEL(1, "%s : Decoding error (36) : %s \n",
srcFileName, ZSTD_getErrorName(zstdError));
FIO_zstdErrorHelp(prefs, ress, zstdError, srcFileName);
return FIO_ERROR_FRAME_DECODING;
case FIO_RUST_ZSTD_FRAME_ACTION_PREMATURE_END:
DISPLAYLEVEL(1, "%s : Read error (39) : premature end \n",
srcFileName);
return FIO_ERROR_FRAME_DECODING;
default:
assert(0);
return FIO_ERROR_FRAME_DECODING;
}
}
/* Rust owns the mixed-format loop, while these adapters keep each codec and
* the private dRess_t layout in this C translation unit. */
@@ -4539,6 +4543,25 @@ static int FIO_rust_decompressLz4FrameCallback(void* opaque,
}
#endif
static void FIO_rust_decompressZstdFrame_displayDecodingError(
void* opaque, const char* srcFileName, size_t zstdError)
{
FIO_rust_decompression_projection_t* const projection =
(FIO_rust_decompression_projection_t*)opaque;
DISPLAYLEVEL(1, "%s : Decoding error (36) : %s \n",
srcFileName, ZSTD_getErrorName(zstdError));
FIO_zstdErrorHelp(projection->prefs, projection->ress,
zstdError, srcFileName);
}
static void FIO_rust_decompressZstdFrame_displayPrematureEnd(
void* opaque, const char* srcFileName)
{
(void)opaque;
DISPLAYLEVEL(1, "%s : Read error (39) : premature end \n",
srcFileName);
}
static int FIO_rust_decompressZstdFrameCallback(void* opaque,
const char* srcFileName,
U64 alreadyDecoded,
@@ -4548,13 +4571,22 @@ static int FIO_rust_decompressZstdFrameCallback(void* opaque,
{
FIO_rust_decompression_projection_t* const projection =
(FIO_rust_decompression_projection_t*)opaque;
FIO_rust_zstd_frame_policy_state state;
(void)mode;
state.callbackContext = projection;
state.fCtx = projection->fCtx;
state.dctx = projection->ress->dctx;
state.readCtx = projection->ress->readCtx;
state.writeCtx = projection->ress->writeCtx;
state.srcFileName = srcFileName;
state.alreadyDecoded = alreadyDecoded;
state.progress = FIO_decompressZstdFrameProgress;
state.displayDecodingError =
FIO_rust_decompressZstdFrame_displayDecodingError;
state.displayPrematureEnd =
FIO_rust_decompressZstdFrame_displayPrematureEnd;
*errorCode = 0;
*frameSize = FIO_decompressZstdFrames(projection->fCtx,
projection->ress,
projection->prefs,
srcFileName,
alreadyDecoded);
*frameSize = FIO_rust_decompressZstdFramePolicy(&state);
return *frameSize == FIO_ERROR_FRAME_DECODING;
}
rust/src/fileio_asyncio.rs
+185 -1
View File
@@ -18,7 +18,7 @@
use std::collections::VecDeque;
use std::ffi::{c_char, c_void};
use std::mem::{align_of, size_of};
use std::mem::{align_of, offset_of, size_of};
#[cfg(any(windows, not(any(unix, windows))))]
use std::os::raw::c_long;
use std::os::raw::{c_int, c_uint};
@@ -119,6 +119,9 @@ pub extern "C" fn FIO_rust_decompressStatusAction(status: c_int) -> c_int {
type FIO_rust_frame_progress_fn = Option<unsafe extern "C" fn(*mut c_void, *const c_char, u64)>;
pub type FIO_rust_decompress_frame_fn =
unsafe extern "C" fn(*mut c_void, *const c_char, u64, *mut u64, *mut usize, c_int) -> c_int;
type FIO_rust_zstd_frame_decoding_error_fn =
unsafe extern "C" fn(*mut c_void, *const c_char, usize);
type FIO_rust_zstd_frame_premature_end_fn = unsafe extern "C" fn(*mut c_void, *const c_char);
pub type FIO_rust_pass_through_fn = unsafe extern "C" fn(*mut c_void) -> c_int;
pub type FIO_rust_decompress_status_fn = unsafe extern "C" fn(*mut c_void, c_int, *const c_char);
pub type FIO_rust_decompress_finish_fn = unsafe extern "C" fn(*mut c_void, *const c_char, u64);
@@ -131,6 +134,39 @@ type FIO_zstd_decompress_fn = unsafe extern "C" fn(
type FIO_zstd_in_size_fn = extern "C" fn() -> usize;
type FIO_zstd_is_frame_fn = unsafe extern "C" fn(*const c_void, usize) -> c_uint;
/// Rust owns the default zstd-frame result policy. C supplies only the exact
/// display callbacks and the private asynchronous-pool pointers needed by the
/// frame loop.
#[repr(C)]
pub struct FIO_rust_zstd_frame_policy_state {
callback_context: *mut c_void,
f_ctx: *mut c_void,
dctx: *mut c_void,
read_ctx: *mut ReadPoolCtx_t,
write_ctx: *mut WritePoolCtx_t,
src_file_name: *const c_char,
already_decoded: u64,
progress: FIO_rust_frame_progress_fn,
display_decoding_error: Option<FIO_rust_zstd_frame_decoding_error_fn>,
display_premature_end: Option<FIO_rust_zstd_frame_premature_end_fn>,
}
const _: () = {
assert!(offset_of!(FIO_rust_zstd_frame_policy_state, callback_context) == 0);
assert!(offset_of!(FIO_rust_zstd_frame_policy_state, f_ctx) == size_of::<usize>());
assert!(offset_of!(FIO_rust_zstd_frame_policy_state, dctx) == 2 * size_of::<usize>());
assert!(offset_of!(FIO_rust_zstd_frame_policy_state, read_ctx) == 3 * size_of::<usize>());
assert!(offset_of!(FIO_rust_zstd_frame_policy_state, write_ctx) == 4 * size_of::<usize>());
assert!(offset_of!(FIO_rust_zstd_frame_policy_state, src_file_name) == 5 * size_of::<usize>());
assert!(offset_of!(FIO_rust_zstd_frame_policy_state, already_decoded) == 6 * size_of::<usize>());
assert!(offset_of!(FIO_rust_zstd_frame_policy_state, progress) == 7 * size_of::<usize>());
assert!(offset_of!(FIO_rust_zstd_frame_policy_state, display_decoding_error) == 8 * size_of::<usize>());
assert!(offset_of!(FIO_rust_zstd_frame_policy_state, display_premature_end) == 9 * size_of::<usize>());
assert!(size_of::<FIO_rust_zstd_frame_policy_state>() == 10 * size_of::<usize>());
assert!(size_of::<FIO_rust_zstd_frame_decoding_error_fn>() == size_of::<usize>());
assert!(size_of::<FIO_rust_zstd_frame_premature_end_fn>() == size_of::<usize>());
};
/// C supplies format-specific decoders through this projection. The opaque
/// value is normally a pointer to C's private `dRess_t`; Rust only drives the
/// callbacks and never depends on that platform-sensitive layout.
@@ -4879,6 +4915,76 @@ pub unsafe extern "C" fn FIO_rust_decompressZstdFrames(
}
}
const FIO_ERROR_FRAME_DECODING: u64 = u64::MAX - 1;
#[inline]
unsafe fn zstd_frame_policy_result(
callback_context: *mut c_void,
src_file_name: *const c_char,
status: c_int,
decoded_size: u64,
zstd_error: usize,
display_decoding_error: FIO_rust_zstd_frame_decoding_error_fn,
display_premature_end: FIO_rust_zstd_frame_premature_end_fn,
) -> u64 {
match status {
FIO_RUST_ZSTD_FRAME_OK => decoded_size,
FIO_RUST_ZSTD_FRAME_DECODING_ERROR => {
unsafe { display_decoding_error(callback_context, src_file_name, zstd_error) };
FIO_ERROR_FRAME_DECODING
}
FIO_RUST_ZSTD_FRAME_PREMATURE_END => {
unsafe { display_premature_end(callback_context, src_file_name) };
FIO_ERROR_FRAME_DECODING
}
_ => FIO_ERROR_FRAME_DECODING,
}
}
/// Run the default zstd-frame loop and apply its historical result/error
/// policy. C retains only exact diagnostics and private resource callbacks.
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_decompressZstdFramePolicy(
state: *const FIO_rust_zstd_frame_policy_state,
) -> u64 {
let Some(state) = (unsafe { state.as_ref() }) else {
return FIO_ERROR_FRAME_DECODING;
};
let Some(display_decoding_error) = state.display_decoding_error else {
return FIO_ERROR_FRAME_DECODING;
};
let Some(display_premature_end) = state.display_premature_end else {
return FIO_ERROR_FRAME_DECODING;
};
let mut decoded_size = 0;
let mut zstd_error = 0;
let status = unsafe {
FIO_rust_decompressZstdFrames(
state.f_ctx,
state.dctx,
state.read_ctx,
state.write_ctx,
state.src_file_name,
state.already_decoded,
&mut decoded_size,
&mut zstd_error,
state.progress,
)
};
unsafe {
zstd_frame_policy_result(
state.callback_context,
state.src_file_name,
status,
decoded_size,
zstd_error,
display_decoding_error,
display_premature_end,
)
}
}
/// Decompresses one gzip member through a C-owned zlib stream and the common
/// asynchronous I/O projection. Input is consumed immediately after each
/// inflate call; this is equivalent to the original C leaf's final
@@ -10279,6 +10385,84 @@ mod tests {
mod zstd_frame_tests {
use super::*;
#[derive(Default)]
struct PolicyCallbackState {
decoding_errors: Vec<usize>,
premature_ends: usize,
}
unsafe extern "C" fn record_policy_decoding_error(
opaque: *mut c_void,
_source_name: *const c_char,
error: usize,
) {
unsafe { (*opaque.cast::<PolicyCallbackState>()).decoding_errors.push(error) };
}
unsafe extern "C" fn record_policy_premature_end(
opaque: *mut c_void,
_source_name: *const c_char,
) {
unsafe { (*opaque.cast::<PolicyCallbackState>()).premature_ends += 1 };
}
#[test]
fn zstd_frame_policy_preserves_result_classes_and_callback_order() {
let mut context = PolicyCallbackState::default();
let callback_context = (&mut context as *mut PolicyCallbackState).cast();
assert_eq!(
unsafe {
zstd_frame_policy_result(
callback_context,
ptr::null(),
FIO_RUST_ZSTD_FRAME_OK,
17,
0,
record_policy_decoding_error,
record_policy_premature_end,
)
},
17
);
assert!(context.decoding_errors.is_empty());
assert_eq!(context.premature_ends, 0);
assert_eq!(
unsafe {
zstd_frame_policy_result(
callback_context,
ptr::null(),
FIO_RUST_ZSTD_FRAME_DECODING_ERROR,
0,
23,
record_policy_decoding_error,
record_policy_premature_end,
)
},
FIO_ERROR_FRAME_DECODING
);
assert_eq!(context.decoding_errors, [23]);
assert_eq!(context.premature_ends, 0);
assert_eq!(
unsafe {
zstd_frame_policy_result(
callback_context,
ptr::null(),
FIO_RUST_ZSTD_FRAME_PREMATURE_END,
0,
0,
record_policy_decoding_error,
record_policy_premature_end,
)
},
FIO_ERROR_FRAME_DECODING
);
assert_eq!(context.decoding_errors, [23]);
assert_eq!(context.premature_ends, 1);
}
#[repr(C)]
#[derive(Default)]
struct ProgressState {