diff --git a/programs/fileio.c b/programs/fileio.c index 3d2a33c25..ad9ac5260 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -421,6 +421,39 @@ int FIO_rust_decompressZstdFrames(void* fCtx, U64* decodedSize, size_t* zstdError, FIO_rust_frame_progress_fn progress); +enum { + FIO_RUST_DECOMPRESS_OK = 0, + FIO_RUST_DECOMPRESS_PASS_THROUGH = 1, + FIO_RUST_DECOMPRESS_EMPTY_INPUT = 2, + FIO_RUST_DECOMPRESS_SHORT_INPUT = 3, + FIO_RUST_DECOMPRESS_GZIP_UNSUPPORTED = 4, + FIO_RUST_DECOMPRESS_LZMA_UNSUPPORTED = 5, + FIO_RUST_DECOMPRESS_LZ4_UNSUPPORTED = 6, + FIO_RUST_DECOMPRESS_FRAME_ERROR = 7, + FIO_RUST_DECOMPRESS_UNSUPPORTED_FORMAT = 8, + FIO_RUST_DECOMPRESS_PASS_THROUGH_ERROR = 9, + FIO_RUST_DECOMPRESS_ZSTD_UNSUPPORTED = 10 +}; +typedef int (*FIO_rust_decompress_frame_fn)(void* opaque, + const char* srcFileName, + U64 alreadyDecoded, + U64* frameSize, + size_t* errorCode, + int mode); +typedef int (*FIO_rust_pass_through_fn)(void* opaque); +typedef struct { + void* opaque; + FIO_rust_decompress_frame_fn decode_zstd; + FIO_rust_decompress_frame_fn decode_gzip; + FIO_rust_decompress_frame_fn decode_lzma; + FIO_rust_decompress_frame_fn decode_lz4; + FIO_rust_pass_through_fn pass_through; +} 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); void FIO_rust_displayCompressionParameters(const FIO_prefs_t* prefs); #ifdef ZSTD_LZ4COMPRESS int FIO_rust_LZ4_GetBlockSize_FromBlockId(int id); @@ -2474,6 +2507,93 @@ FIO_decompressLz4Frame(dRess_t* ress, const char* srcFileName) } #endif +/* Rust owns the mixed-format loop, while these adapters keep each codec and + * the private dRess_t layout in this C translation unit. */ +typedef struct { + FIO_ctx_t* fCtx; + dRess_t* ress; + const FIO_prefs_t* prefs; +} FIO_rust_decompression_projection_t; + +static int FIO_rust_decompressZstdFrameCallback(void* opaque, + const char* srcFileName, + U64 alreadyDecoded, + U64* frameSize, + size_t* errorCode, + int mode) +{ + FIO_rust_decompression_projection_t* const projection = + (FIO_rust_decompression_projection_t*)opaque; + (void)mode; + *errorCode = 0; + *frameSize = FIO_decompressZstdFrames(projection->fCtx, + projection->ress, + projection->prefs, + srcFileName, + alreadyDecoded); + return *frameSize == FIO_ERROR_FRAME_DECODING; +} + +#ifdef ZSTD_GZDECOMPRESS +static int FIO_rust_decompressGzFrameCallback(void* opaque, + const char* srcFileName, + U64 alreadyDecoded, + U64* frameSize, + size_t* errorCode, + int mode) +{ + FIO_rust_decompression_projection_t* const projection = + (FIO_rust_decompression_projection_t*)opaque; + (void)alreadyDecoded; + (void)mode; + *errorCode = 0; + *frameSize = FIO_decompressGzFrame(projection->ress, srcFileName); + return *frameSize == FIO_ERROR_FRAME_DECODING; +} +#endif + +#ifdef ZSTD_LZMADECOMPRESS +static int FIO_rust_decompressLzmaFrameCallback(void* opaque, + const char* srcFileName, + U64 alreadyDecoded, + U64* frameSize, + size_t* errorCode, + int mode) +{ + FIO_rust_decompression_projection_t* const projection = + (FIO_rust_decompression_projection_t*)opaque; + (void)alreadyDecoded; + *errorCode = 0; + *frameSize = FIO_decompressLzmaFrame(projection->ress, srcFileName, mode); + return *frameSize == FIO_ERROR_FRAME_DECODING; +} +#endif + +#ifdef ZSTD_LZ4DECOMPRESS +static int FIO_rust_decompressLz4FrameCallback(void* opaque, + const char* srcFileName, + U64 alreadyDecoded, + U64* frameSize, + size_t* errorCode, + int mode) +{ + FIO_rust_decompression_projection_t* const projection = + (FIO_rust_decompression_projection_t*)opaque; + (void)alreadyDecoded; + (void)mode; + *errorCode = 0; + *frameSize = FIO_decompressLz4Frame(projection->ress, srcFileName); + return *frameSize == FIO_ERROR_FRAME_DECODING; +} +#endif + +static int FIO_rust_decompressPassThroughCallback(void* opaque) +{ + FIO_rust_decompression_projection_t* const projection = + (FIO_rust_decompression_projection_t*)opaque; + return FIO_passThrough(projection->ress); +} + /** FIO_decompressFrames() : @@ -2486,9 +2606,11 @@ static int FIO_decompressFrames(FIO_ctx_t* const fCtx, dRess_t ress, const FIO_prefs_t* const prefs, const char* dstFileName, const char* srcFileName) { - unsigned readSomething = 0; - unsigned long long filesize = 0; + U64 filesize = 0; int passThrough = prefs->passThrough; + FIO_rust_decompression_projection_t projection; + FIO_rust_decompress_callbacks_t callbacks; + int status; if (passThrough == -1) { /* If pass-through mode is not explicitly enabled or disabled, @@ -2499,72 +2621,63 @@ static int FIO_decompressFrames(FIO_ctx_t* const fCtx, } assert(passThrough == 0 || passThrough == 1); - /* for each frame */ - for ( ; ; ) { - /* check magic number -> version */ - size_t const toRead = 4; - const BYTE* buf; - AIO_ReadPool_fillBuffer(ress.readCtx, toRead); - buf = (const BYTE*)ress.readCtx->srcBuffer; - if (ress.readCtx->srcBufferLoaded==0) { - if (readSomething==0) { /* srcFile is empty (which is invalid) */ - DISPLAYLEVEL(1, "zstd: %s: unexpected end of file \n", srcFileName); - return 1; - } /* else, just reached frame boundary */ - break; /* no more input */ - } - readSomething = 1; /* there is at least 1 byte in srcFile */ - if (ress.readCtx->srcBufferLoaded < toRead) { /* not enough input to check magic number */ - if (passThrough) { - return FIO_passThrough(&ress); - } + projection.fCtx = fCtx; + projection.ress = &ress; + projection.prefs = prefs; + memset(&callbacks, 0, sizeof(callbacks)); + callbacks.opaque = &projection; + callbacks.decode_zstd = FIO_rust_decompressZstdFrameCallback; +#ifdef ZSTD_GZDECOMPRESS + callbacks.decode_gzip = FIO_rust_decompressGzFrameCallback; +#endif +#ifdef ZSTD_LZMADECOMPRESS + callbacks.decode_lzma = FIO_rust_decompressLzmaFrameCallback; +#endif +#ifdef ZSTD_LZ4DECOMPRESS + callbacks.decode_lz4 = FIO_rust_decompressLz4FrameCallback; +#endif + callbacks.pass_through = FIO_rust_decompressPassThroughCallback; + + 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; - } - if (ZSTD_isFrame(buf, ress.readCtx->srcBufferLoaded)) { - unsigned long long const frameSize = FIO_decompressZstdFrames(fCtx, &ress, prefs, srcFileName, filesize); - if (frameSize == FIO_ERROR_FRAME_DECODING) return 1; - filesize += frameSize; - } else if (buf[0] == 31 && buf[1] == 139) { /* gz magic number */ -#ifdef ZSTD_GZDECOMPRESS - unsigned long long const frameSize = FIO_decompressGzFrame(&ress, srcFileName); - if (frameSize == FIO_ERROR_FRAME_DECODING) return 1; - filesize += frameSize; -#else + 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; -#endif - } else if ((buf[0] == 0xFD && buf[1] == 0x37) /* xz magic number */ - || (buf[0] == 0x5D && buf[1] == 0x00)) { /* lzma header (no magic number) */ -#ifdef ZSTD_LZMADECOMPRESS - unsigned long long const frameSize = FIO_decompressLzmaFrame(&ress, srcFileName, buf[0] != 0xFD); - if (frameSize == FIO_ERROR_FRAME_DECODING) return 1; - filesize += frameSize; -#else + 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; -#endif - } else if (MEM_readLE32(buf) == LZ4_MAGICNUMBER) { -#ifdef ZSTD_LZ4DECOMPRESS - unsigned long long const frameSize = FIO_decompressLz4Frame(&ress, srcFileName); - if (frameSize == FIO_ERROR_FRAME_DECODING) return 1; - filesize += frameSize; -#else + 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; -#endif - } else if (passThrough) { - return FIO_passThrough(&ress); - } else { + case FIO_RUST_DECOMPRESS_UNSUPPORTED_FORMAT: DISPLAYLEVEL(1, "zstd: %s: unsupported format \n", srcFileName); return 1; - } } /* for each frame */ + 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, filesize); + DISPLAY_SUMMARY("%-20s: %llu bytes \n", srcFileName, + (unsigned long long)filesize); return 0; } diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index d6b19ff1e..bab6a43d4 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -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; +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, + pub decode_gzip: Option, + pub decode_lzma: Option, + pub decode_lz4: Option, + pub pass_through: Option, +} + /// 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(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::(); @@ -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::() }; + 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 { thread_pool: *mut c_void,