diff --git a/programs/fileio.c b/programs/fileio.c index 813155cd7..7c6e3fc03 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -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; } diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index a9e7d2ff2..eb407f808 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -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; 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, + display_premature_end: Option, +} + +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::()); + assert!(offset_of!(FIO_rust_zstd_frame_policy_state, dctx) == 2 * size_of::()); + assert!(offset_of!(FIO_rust_zstd_frame_policy_state, read_ctx) == 3 * size_of::()); + assert!(offset_of!(FIO_rust_zstd_frame_policy_state, write_ctx) == 4 * size_of::()); + assert!(offset_of!(FIO_rust_zstd_frame_policy_state, src_file_name) == 5 * size_of::()); + assert!(offset_of!(FIO_rust_zstd_frame_policy_state, already_decoded) == 6 * size_of::()); + assert!(offset_of!(FIO_rust_zstd_frame_policy_state, progress) == 7 * size_of::()); + assert!(offset_of!(FIO_rust_zstd_frame_policy_state, display_decoding_error) == 8 * size_of::()); + assert!(offset_of!(FIO_rust_zstd_frame_policy_state, display_premature_end) == 9 * size_of::()); + assert!(size_of::() == 10 * size_of::()); + assert!(size_of::() == size_of::()); + assert!(size_of::() == size_of::()); +}; + /// 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, + 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::()).decoding_errors.push(error) }; + } + + unsafe extern "C" fn record_policy_premature_end( + opaque: *mut c_void, + _source_name: *const c_char, + ) { + unsafe { (*opaque.cast::()).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 {