feat(cli): move mixed-format dispatch into Rust

Move file I/O's format probing, frame classification, callback dispatch, and
size accounting into Rust. C retains the codec-specific decompression,
pass-through, metadata, and private resource adapters, while Rust coordinates
mixed-format streams and maps callback outcomes back to the CLI error paths.

Test Plan:
- cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1
- cargo test --manifest-path rust/cli/Cargo.toml --all-targets -- --test-threads=1
- cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings
- cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings
- make -B -C programs -j2 zstd
- make -B -C tests -j2 test-cli-tests
- make -B -C tests -j2 test-zstd
This commit is contained in:
2026-07-18 22:09:05 +02:00
parent 2a588f89f2
commit 139b45832e
2 changed files with 471 additions and 53 deletions
+305
View File
@@ -37,7 +37,22 @@ pub const FIO_RUST_ZSTD_FRAME_OK: c_int = 0;
pub const FIO_RUST_ZSTD_FRAME_DECODING_ERROR: c_int = 1;
pub const FIO_RUST_ZSTD_FRAME_PREMATURE_END: c_int = 2;
pub const FIO_RUST_DECOMPRESS_OK: c_int = 0;
pub const FIO_RUST_DECOMPRESS_PASS_THROUGH: c_int = 1;
pub const FIO_RUST_DECOMPRESS_EMPTY_INPUT: c_int = 2;
pub const FIO_RUST_DECOMPRESS_SHORT_INPUT: c_int = 3;
pub const FIO_RUST_DECOMPRESS_GZIP_UNSUPPORTED: c_int = 4;
pub const FIO_RUST_DECOMPRESS_LZMA_UNSUPPORTED: c_int = 5;
pub const FIO_RUST_DECOMPRESS_LZ4_UNSUPPORTED: c_int = 6;
pub const FIO_RUST_DECOMPRESS_FRAME_ERROR: c_int = 7;
pub const FIO_RUST_DECOMPRESS_UNSUPPORTED_FORMAT: c_int = 8;
pub const FIO_RUST_DECOMPRESS_PASS_THROUGH_ERROR: c_int = 9;
pub const FIO_RUST_DECOMPRESS_ZSTD_UNSUPPORTED: c_int = 10;
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;
pub type FIO_rust_pass_through_fn = unsafe extern "C" fn(*mut c_void) -> c_int;
type FIO_zstd_reset_fn = unsafe extern "C" fn(*mut c_void, c_int) -> usize;
type FIO_zstd_decompress_fn = unsafe extern "C" fn(
*mut c_void,
@@ -47,6 +62,19 @@ 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;
/// 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.
#[repr(C)]
pub struct FIO_rust_decompress_callbacks_t {
pub opaque: *mut c_void,
pub decode_zstd: Option<FIO_rust_decompress_frame_fn>,
pub decode_gzip: Option<FIO_rust_decompress_frame_fn>,
pub decode_lzma: Option<FIO_rust_decompress_frame_fn>,
pub decode_lz4: Option<FIO_rust_decompress_frame_fn>,
pub pass_through: Option<FIO_rust_pass_through_fn>,
}
/// C's `FIO_prefs_t` from `programs/fileio_types.h`.
///
/// `fileio_prefs.rs` contains the same C layout for the preferences API. It
@@ -1663,6 +1691,178 @@ pub unsafe extern "C" fn FIO_rust_decompressZstdFrames(
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum DecompressionFormat {
Zstd,
Gzip,
Xz,
Lzma,
Lz4,
ShortHeader,
Unsupported,
}
fn classify_decompression_format<F>(buffer: &[u8], is_zstd_frame: F) -> DecompressionFormat
where
F: Fn(&[u8]) -> bool,
{
if buffer.len() < 4 {
return DecompressionFormat::ShortHeader;
}
if is_zstd_frame(buffer) {
return DecompressionFormat::Zstd;
}
if buffer[0] == 31 && buffer[1] == 139 {
return DecompressionFormat::Gzip;
}
if (buffer[0] == 0xFD && buffer[1] == 0x37) || (buffer[0] == 0x5D && buffer[1] == 0x00) {
return if buffer[0] == 0xFD {
DecompressionFormat::Xz
} else {
DecompressionFormat::Lzma
};
}
if u32::from_le_bytes([buffer[0], buffer[1], buffer[2], buffer[3]]) == 0x184D2204 {
return DecompressionFormat::Lz4;
}
DecompressionFormat::Unsupported
}
#[inline]
unsafe fn is_zstd_frame_for_dispatch(buffer: &[u8]) -> bool {
#[cfg(test)]
{
/* Standalone Rust tests do not link the C legacy-decoder shim. The
* dispatch tests only need the modern frame magic; production keeps
* the complete public predicate below. */
buffer.starts_with(&[0x28, 0xB5, 0x2F, 0xFD])
}
#[cfg(not(test))]
{
unsafe { crate::zstd_decompress::ZSTD_isFrame(buffer.as_ptr().cast(), buffer.len()) != 0 }
}
}
unsafe fn run_pass_through_callback(callbacks: &FIO_rust_decompress_callbacks_t) -> c_int {
let Some(callback) = callbacks.pass_through else {
return FIO_RUST_DECOMPRESS_PASS_THROUGH_ERROR;
};
if unsafe { callback(callbacks.opaque) } == 0 {
FIO_RUST_DECOMPRESS_PASS_THROUGH
} else {
FIO_RUST_DECOMPRESS_PASS_THROUGH_ERROR
}
}
/// Drives the CLI's mixed-format decompression loop.
///
/// Rust owns input probing, format selection, repeated callback dispatch, and
/// decoded-size accumulation. Codec implementations and their private C
/// resource layout stay behind the callback projection above. A successful
/// pass-through is reported separately because the C caller historically
/// returns before final decompression accounting in that case.
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_decompressFrames(
read_ctx: *mut ReadPoolCtx_t,
src_file_name: *const c_char,
pass_through: c_int,
decoded_size: *mut u64,
callbacks: *const FIO_rust_decompress_callbacks_t,
) -> c_int {
assert!(!read_ctx.is_null());
assert!(!src_file_name.is_null());
assert!(!decoded_size.is_null());
assert!(!callbacks.is_null());
assert!(pass_through == 0 || pass_through == 1);
let callbacks = unsafe { &*callbacks };
unsafe { *decoded_size = 0 };
let mut read_something = false;
loop {
unsafe { AIO_ReadPool_fillBuffer(read_ctx, 4) };
let loaded = unsafe { read_buffer_loaded(read_ctx) };
if loaded == 0 {
return if read_something {
FIO_RUST_DECOMPRESS_OK
} else {
FIO_RUST_DECOMPRESS_EMPTY_INPUT
};
}
read_something = true;
if loaded < 4 {
return if pass_through != 0 {
unsafe { run_pass_through_callback(callbacks) }
} else {
FIO_RUST_DECOMPRESS_SHORT_INPUT
};
}
let source = unsafe { read_buffer_ptr(read_ctx) };
let buffer = unsafe { std::slice::from_raw_parts(source, loaded) };
let format = classify_decompression_format(buffer, |bytes| unsafe {
is_zstd_frame_for_dispatch(bytes)
});
let (callback, missing_status, mode) = match format {
DecompressionFormat::Zstd => (
callbacks.decode_zstd,
FIO_RUST_DECOMPRESS_ZSTD_UNSUPPORTED,
0,
),
DecompressionFormat::Gzip => (
callbacks.decode_gzip,
FIO_RUST_DECOMPRESS_GZIP_UNSUPPORTED,
0,
),
DecompressionFormat::Xz => (
callbacks.decode_lzma,
FIO_RUST_DECOMPRESS_LZMA_UNSUPPORTED,
0,
),
DecompressionFormat::Lzma => (
callbacks.decode_lzma,
FIO_RUST_DECOMPRESS_LZMA_UNSUPPORTED,
1,
),
DecompressionFormat::Lz4 => {
(callbacks.decode_lz4, FIO_RUST_DECOMPRESS_LZ4_UNSUPPORTED, 0)
}
DecompressionFormat::ShortHeader => unreachable!(),
DecompressionFormat::Unsupported => {
return if pass_through != 0 {
unsafe { run_pass_through_callback(callbacks) }
} else {
FIO_RUST_DECOMPRESS_UNSUPPORTED_FORMAT
};
}
};
let Some(callback) = callback else {
return missing_status;
};
let mut frame_size = 0_u64;
let mut error_code = 0_usize;
let status = unsafe {
callback(
callbacks.opaque,
src_file_name,
*decoded_size,
&mut frame_size,
&mut error_code,
mode,
)
};
if status != 0 {
return FIO_RUST_DECOMPRESS_FRAME_ERROR;
}
unsafe {
*decoded_size = (*decoded_size).wrapping_add(frame_size);
}
}
}
#[inline]
unsafe fn read_buffer_ptr(ctx: *mut ReadPoolCtx_t) -> *const u8 {
let context = ctx.cast::<u8>();
@@ -1696,6 +1896,111 @@ mod tests {
prefs
}
#[test]
fn classifies_all_cli_decompression_headers() {
let is_mock_zstd = |buffer: &[u8]| buffer.starts_with(&[0x28, 0xB5, 0x2F, 0xFD]);
assert_eq!(
classify_decompression_format(&[0x28, 0xB5, 0x2F, 0xFD], is_mock_zstd),
DecompressionFormat::Zstd
);
assert_eq!(
classify_decompression_format(&[31, 139, 8, 0], is_mock_zstd),
DecompressionFormat::Gzip
);
assert_eq!(
classify_decompression_format(&[0xFD, 0x37, 0x7A, 0x58], is_mock_zstd),
DecompressionFormat::Xz
);
assert_eq!(
classify_decompression_format(&[0x5D, 0x00, 0x00, 0x80], is_mock_zstd),
DecompressionFormat::Lzma
);
assert_eq!(
classify_decompression_format(&[0x04, 0x22, 0x4D, 0x18], is_mock_zstd),
DecompressionFormat::Lz4
);
assert_eq!(
classify_decompression_format(&[0x00, 0x01, 0x02, 0x03], is_mock_zstd),
DecompressionFormat::Unsupported
);
assert_eq!(
classify_decompression_format(&[0x00, 0x01, 0x02], is_mock_zstd),
DecompressionFormat::ShortHeader
);
}
#[cfg(unix)]
struct DispatchTestState {
read_ctx: *mut ReadPoolCtx_t,
calls: Vec<(u64, c_int)>,
}
#[cfg(unix)]
unsafe extern "C" fn record_dispatch_callback(
opaque: *mut c_void,
_src_file_name: *const c_char,
already_decoded: u64,
frame_size: *mut u64,
error_code: *mut usize,
mode: c_int,
) -> c_int {
let state = unsafe { &mut *opaque.cast::<DispatchTestState>() };
state.calls.push((already_decoded, mode));
let loaded = unsafe { read_buffer_loaded(state.read_ctx) };
unsafe {
AIO_ReadPool_consumeBytes(state.read_ctx, loaded);
*frame_size = 17;
*error_code = 0;
}
0
}
#[cfg(unix)]
#[test]
fn dispatches_a_format_and_accumulates_callback_output() {
let input_file = unsafe { libc::tmpfile() };
assert!(!input_file.is_null());
let header = [31_u8, 139, 8, 0];
assert_eq!(
unsafe { libc::fwrite(header.as_ptr().cast(), 1, header.len(), input_file) },
header.len()
);
assert_eq!(unsafe { libc::fflush(input_file) }, 0);
assert_eq!(unsafe { libc::fseek(input_file, 0, libc::SEEK_SET) }, 0);
let prefs = test_prefs(0);
let read_ctx = unsafe { AIO_ReadPool_create(&prefs, 4) };
unsafe { AIO_ReadPool_setFile(read_ctx, input_file) };
let mut state = DispatchTestState {
read_ctx,
calls: Vec::new(),
};
let callbacks = FIO_rust_decompress_callbacks_t {
opaque: (&mut state as *mut DispatchTestState).cast(),
decode_zstd: None,
decode_gzip: Some(record_dispatch_callback),
decode_lzma: None,
decode_lz4: None,
pass_through: None,
};
let mut decoded_size = 0;
let status = unsafe {
FIO_rust_decompressFrames(
read_ctx,
c"dispatch-test".as_ptr(),
0,
&mut decoded_size,
&callbacks,
)
};
assert_eq!(status, FIO_RUST_DECOMPRESS_OK);
assert_eq!(decoded_size, 17);
assert_eq!(state.calls, vec![(0, 0)]);
unsafe { AIO_ReadPool_free(read_ctx) };
}
#[repr(C)]
struct CBase<M> {
thread_pool: *mut c_void,