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

feat(cli): move decompression result policy into Rust

Browse Source 
Move mixed-format decompression status mapping and successful-file finalization
into Rust. C keeps the diagnostic and private FIO context callbacks, while
Rust decides pass-through/error behavior and invokes final accounting only for
a successful decoded file.

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 --tests -- -D warnings
- make -B -C programs -j2 zstd
- make -B -C tests -j2 test-cli-tests
This commit is contained in:
ddidderr
2026-07-18 23:46:17 +02:00
parent 440585995b
commit cb438dc4f1
2 changed files with 190 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
programs/fileio.c
+57 -40
View File
@@ -441,6 +441,12 @@ typedef int (*FIO_rust_decompress_frame_fn)(void* opaque,
size_t* errorCode,
int mode);
typedef int (*FIO_rust_pass_through_fn)(void* opaque);
typedef void (*FIO_rust_decompress_status_fn)(void* opaque,
int status,
const char* srcFileName);
typedef void (*FIO_rust_decompress_finish_fn)(void* opaque,
const char* srcFileName,
U64 decodedSize);
typedef struct {
void* opaque;
FIO_rust_decompress_frame_fn decode_zstd;
@@ -448,12 +454,18 @@ typedef struct {
FIO_rust_decompress_frame_fn decode_lzma;
FIO_rust_decompress_frame_fn decode_lz4;
FIO_rust_pass_through_fn pass_through;
FIO_rust_decompress_status_fn report_status;
FIO_rust_decompress_finish_fn finish;
} FIO_rust_decompress_callbacks_t;
int FIO_rust_decompressFrames(ReadPoolCtx_t* readCtx,
const char* srcFileName,
int passThrough,
U64* decodedSize,
const FIO_rust_decompress_callbacks_t* callbacks);
int FIO_rust_finishDecompressFrames(int status,
const char* srcFileName,
U64 decodedSize,
const FIO_rust_decompress_callbacks_t* callbacks);
void FIO_rust_displayCompressionParameters(const FIO_prefs_t* prefs);
#ifdef ZSTD_LZ4COMPRESS
int FIO_rust_LZ4_GetBlockSize_FromBlockId(int id);
@@ -3105,6 +3117,48 @@ static int FIO_rust_decompressPassThroughCallback(void* opaque)
return FIO_passThrough(projection->ress);
}
static void FIO_rust_decompressStatusCallback(void* opaque,
int status,
const char* srcFileName)
{
(void)opaque;
switch (status) {
case FIO_RUST_DECOMPRESS_EMPTY_INPUT:
DISPLAYLEVEL(1, "zstd: %s: unexpected end of file \n", srcFileName);
break;
case FIO_RUST_DECOMPRESS_SHORT_INPUT:
DISPLAYLEVEL(1, "zstd: %s: unknown header \n", srcFileName);
break;
case FIO_RUST_DECOMPRESS_GZIP_UNSUPPORTED:
DISPLAYLEVEL(1, "zstd: %s: gzip file cannot be uncompressed (zstd compiled without HAVE_ZLIB) -- ignored \n", srcFileName);
break;
case FIO_RUST_DECOMPRESS_LZMA_UNSUPPORTED:
DISPLAYLEVEL(1, "zstd: %s: xz/lzma file cannot be uncompressed (zstd compiled without HAVE_LZMA) -- ignored \n", srcFileName);
break;
case FIO_RUST_DECOMPRESS_LZ4_UNSUPPORTED:
DISPLAYLEVEL(1, "zstd: %s: lz4 file cannot be uncompressed (zstd compiled without HAVE_LZ4) -- ignored \n", srcFileName);
break;
case FIO_RUST_DECOMPRESS_UNSUPPORTED_FORMAT:
DISPLAYLEVEL(1, "zstd: %s: unsupported format \n", srcFileName);
break;
default:
break;
}
}
static void FIO_rust_decompressFinalCallback(void* opaque,
const char* srcFileName,
U64 decodedSize)
{
FIO_rust_decompression_projection_t* const projection =
(FIO_rust_decompression_projection_t*)opaque;
projection->fCtx->totalBytesOutput += (size_t)decodedSize;
DISPLAY_PROGRESS("\r%79s\r", "");
if (FIO_shouldDisplayFileSummary(projection->fCtx))
DISPLAY_SUMMARY("%-20s: %llu bytes \n", srcFileName,
(unsigned long long)decodedSize);
}
/** FIO_decompressFrames() :
@@ -3148,49 +3202,12 @@ static int FIO_decompressFrames(FIO_ctx_t* const fCtx,
callbacks.decode_lz4 = FIO_rust_decompressLz4FrameCallback;
#endif
callbacks.pass_through = FIO_rust_decompressPassThroughCallback;
callbacks.report_status = FIO_rust_decompressStatusCallback;
callbacks.finish = FIO_rust_decompressFinalCallback;
status = FIO_rust_decompressFrames(ress.readCtx, srcFileName, passThrough,
&filesize, &callbacks);
switch (status) {
case FIO_RUST_DECOMPRESS_OK:
break;
case FIO_RUST_DECOMPRESS_PASS_THROUGH:
return 0;
case FIO_RUST_DECOMPRESS_EMPTY_INPUT:
DISPLAYLEVEL(1, "zstd: %s: unexpected end of file \n", srcFileName);
return 1;
case FIO_RUST_DECOMPRESS_SHORT_INPUT:
DISPLAYLEVEL(1, "zstd: %s: unknown header \n", srcFileName);
return 1;
case FIO_RUST_DECOMPRESS_GZIP_UNSUPPORTED:
DISPLAYLEVEL(1, "zstd: %s: gzip file cannot be uncompressed (zstd compiled without HAVE_ZLIB) -- ignored \n", srcFileName);
return 1;
case FIO_RUST_DECOMPRESS_LZMA_UNSUPPORTED:
DISPLAYLEVEL(1, "zstd: %s: xz/lzma file cannot be uncompressed (zstd compiled without HAVE_LZMA) -- ignored \n", srcFileName);
return 1;
case FIO_RUST_DECOMPRESS_LZ4_UNSUPPORTED:
DISPLAYLEVEL(1, "zstd: %s: lz4 file cannot be uncompressed (zstd compiled without HAVE_LZ4) -- ignored \n", srcFileName);
return 1;
case FIO_RUST_DECOMPRESS_UNSUPPORTED_FORMAT:
DISPLAYLEVEL(1, "zstd: %s: unsupported format \n", srcFileName);
return 1;
case FIO_RUST_DECOMPRESS_FRAME_ERROR:
case FIO_RUST_DECOMPRESS_PASS_THROUGH_ERROR:
case FIO_RUST_DECOMPRESS_ZSTD_UNSUPPORTED:
return 1;
default:
assert(0);
return 1;
}
/* Final Status */
fCtx->totalBytesOutput += (size_t)filesize;
DISPLAY_PROGRESS("\r%79s\r", "");
if (FIO_shouldDisplayFileSummary(fCtx))
DISPLAY_SUMMARY("%-20s: %llu bytes \n", srcFileName,
(unsigned long long)filesize);
return 0;
return FIO_rust_finishDecompressFrames(status, srcFileName, filesize, &callbacks);
}
/** FIO_decompressDstFile() :
rust/src/fileio_asyncio.rs
+133
View File
@@ -53,6 +53,8 @@ type FIO_rust_frame_progress_fn = Option<unsafe extern "C" fn(*mut c_void, *cons
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;
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);
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,
@@ -73,6 +75,8 @@ pub struct FIO_rust_decompress_callbacks_t {
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>,
pub report_status: Option<FIO_rust_decompress_status_fn>,
pub finish: Option<FIO_rust_decompress_finish_fn>,
}
pub const FIO_RUST_COMPRESS_OK: c_int = 0;
@@ -2854,6 +2858,49 @@ pub unsafe extern "C" fn FIO_rust_decompressFrames(
}
}
/// Maps the mixed-format loop's result to the CLI result and performs the
/// successful-file finalization. C keeps the diagnostic and accounting
/// callbacks because they touch `DISPLAY*` state and the private `FIO_ctx_t`
/// layout; Rust owns the result policy and decides which callback is legal.
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_finishDecompressFrames(
status: c_int,
src_file_name: *const c_char,
decoded_size: u64,
callbacks: *const FIO_rust_decompress_callbacks_t,
) -> c_int {
assert!(!src_file_name.is_null());
assert!(!callbacks.is_null());
let callbacks = unsafe { &*callbacks };
match status {
FIO_RUST_DECOMPRESS_OK => {
if let Some(finish) = callbacks.finish {
unsafe { finish(callbacks.opaque, src_file_name, decoded_size) };
}
0
}
FIO_RUST_DECOMPRESS_PASS_THROUGH => 0,
FIO_RUST_DECOMPRESS_EMPTY_INPUT
| FIO_RUST_DECOMPRESS_SHORT_INPUT
| FIO_RUST_DECOMPRESS_GZIP_UNSUPPORTED
| FIO_RUST_DECOMPRESS_LZMA_UNSUPPORTED
| FIO_RUST_DECOMPRESS_LZ4_UNSUPPORTED
| FIO_RUST_DECOMPRESS_FRAME_ERROR
| FIO_RUST_DECOMPRESS_UNSUPPORTED_FORMAT
| FIO_RUST_DECOMPRESS_PASS_THROUGH_ERROR
| FIO_RUST_DECOMPRESS_ZSTD_UNSUPPORTED => {
if let Some(report_status) = callbacks.report_status {
unsafe { report_status(callbacks.opaque, status, src_file_name) };
}
1
}
_ => {
unreachable!("unknown Rust decompression status: {status}");
}
}
}
#[inline]
unsafe fn read_buffer_ptr(ctx: *mut ReadPoolCtx_t) -> *const u8 {
let context = ctx.cast::<u8>();
@@ -4223,6 +4270,8 @@ mod tests {
decode_lzma: None,
decode_lz4: None,
pass_through: None,
report_status: None,
finish: None,
};
let mut decoded_size = 0;
let status = unsafe {
@@ -4241,6 +4290,90 @@ mod tests {
unsafe { AIO_ReadPool_free(read_ctx) };
}
struct DecompressResultState {
reported_statuses: Vec<c_int>,
finished_sizes: Vec<u64>,
}
unsafe extern "C" fn record_decompress_status(
opaque: *mut c_void,
status: c_int,
_src_file_name: *const c_char,
) {
let state = unsafe { &mut *opaque.cast::<DecompressResultState>() };
state.reported_statuses.push(status);
}
unsafe extern "C" fn record_decompress_finish(
opaque: *mut c_void,
_src_file_name: *const c_char,
decoded_size: u64,
) {
let state = unsafe { &mut *opaque.cast::<DecompressResultState>() };
state.finished_sizes.push(decoded_size);
}
#[test]
fn maps_decompression_status_and_finishes_only_success() {
let mut state = DecompressResultState {
reported_statuses: Vec::new(),
finished_sizes: Vec::new(),
};
let callbacks = FIO_rust_decompress_callbacks_t {
opaque: (&mut state as *mut DecompressResultState).cast(),
decode_zstd: None,
decode_gzip: None,
decode_lzma: None,
decode_lz4: None,
pass_through: None,
report_status: Some(record_decompress_status),
finish: Some(record_decompress_finish),
};
let statuses = [
(FIO_RUST_DECOMPRESS_OK, 0),
(FIO_RUST_DECOMPRESS_PASS_THROUGH, 0),
(FIO_RUST_DECOMPRESS_EMPTY_INPUT, 1),
(FIO_RUST_DECOMPRESS_SHORT_INPUT, 1),
(FIO_RUST_DECOMPRESS_GZIP_UNSUPPORTED, 1),
(FIO_RUST_DECOMPRESS_LZMA_UNSUPPORTED, 1),
(FIO_RUST_DECOMPRESS_LZ4_UNSUPPORTED, 1),
(FIO_RUST_DECOMPRESS_FRAME_ERROR, 1),
(FIO_RUST_DECOMPRESS_UNSUPPORTED_FORMAT, 1),
(FIO_RUST_DECOMPRESS_PASS_THROUGH_ERROR, 1),
(FIO_RUST_DECOMPRESS_ZSTD_UNSUPPORTED, 1),
];
for (status, expected_result) in statuses {
assert_eq!(
unsafe {
FIO_rust_finishDecompressFrames(
status,
c"decompress-result-test".as_ptr(),
123,
&callbacks,
)
},
expected_result
);
}
assert_eq!(state.finished_sizes, vec![123]);
assert_eq!(
state.reported_statuses,
vec![
FIO_RUST_DECOMPRESS_EMPTY_INPUT,
FIO_RUST_DECOMPRESS_SHORT_INPUT,
FIO_RUST_DECOMPRESS_GZIP_UNSUPPORTED,
FIO_RUST_DECOMPRESS_LZMA_UNSUPPORTED,
FIO_RUST_DECOMPRESS_LZ4_UNSUPPORTED,
FIO_RUST_DECOMPRESS_FRAME_ERROR,
FIO_RUST_DECOMPRESS_UNSUPPORTED_FORMAT,
FIO_RUST_DECOMPRESS_PASS_THROUGH_ERROR,
FIO_RUST_DECOMPRESS_ZSTD_UNSUPPORTED,
]
);
}
#[repr(C)]
struct CBase<M> {
thread_pool: *mut c_void,