From cb438dc4f178fdcd5ed235e454489807e07e2cc3 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sat, 18 Jul 2026 23:46:17 +0200 Subject: [PATCH] feat(cli): move decompression result policy into Rust 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 --- programs/fileio.c | 97 ++++++++++++++++----------- rust/src/fileio_asyncio.rs | 133 +++++++++++++++++++++++++++++++++++++ 2 files changed, 190 insertions(+), 40 deletions(-) diff --git a/programs/fileio.c b/programs/fileio.c index df8768252..394d8ba5f 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -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() : diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index 8c4b5298e..3eb74de33 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -53,6 +53,8 @@ type FIO_rust_frame_progress_fn = Option 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, pub decode_lz4: Option, pub pass_through: Option, + pub report_status: Option, + pub finish: Option, } 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::(); @@ -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, + finished_sizes: Vec, + } + + 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::() }; + 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::() }; + 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 { thread_pool: *mut c_void,