diff --git a/programs/fileio.c b/programs/fileio.c index df2d6777d..c569bcc0b 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -331,8 +331,6 @@ int FIO_rust_adjustParamsForPatchFromMode(FIO_prefs_t* prefs, unsigned* fileWindowLog, int* autoLdm, int* optimalParser); -void FIO_setInBuffer(ZSTD_inBuffer* output, const void* buf, size_t s, size_t pos); -void FIO_setOutBuffer(ZSTD_outBuffer* output, void* buf, size_t s, size_t pos); const char* FIO_determineCompressedName(const char* srcFileName, const char* outDirName, const char* suffix); const char* FIO_rust_determineDstName(const char* srcFileName, const char* outDirName, const char* const* suffixList, const char* suffixListStr); @@ -1333,6 +1331,14 @@ typedef int (*FIO_rust_zstd_compress_stream_fn)( unsigned char* output, size_t outputSize, size_t* inputPosAfter, size_t* outputProduced, size_t* toFlushNow, size_t* zstdResult); +int FIO_rust_zstd_compressStream( + void* opaque, const char* srcFileName, int directive, + const unsigned char* input, size_t inputSize, size_t inputPos, + unsigned char* output, size_t outputSize, + size_t* inputPosAfter, size_t* outputProduced, + size_t* toFlushNow, size_t* zstdResult); +void FIO_rust_zstd_compressStreamDisplay( + int directive, size_t inputPos, size_t inputSize, size_t outputProduced); typedef void (*FIO_rust_zstd_iteration_fn)( void* opaque, const char* srcFileName, int* compressionLevel, size_t oldInputPos, size_t newInputPos, size_t toFlushNow); @@ -2524,36 +2530,12 @@ static void FIO_rust_zstd_sparseWriteEnd(void* opaque) AIO_WritePool_sparseWriteEnd((WritePoolCtx_t*)opaque); } -static int FIO_rust_zstd_compressStream( - void* opaque, const char* srcFileName, int directive, - const unsigned char* input, size_t inputSize, size_t inputPos, - unsigned char* output, size_t outputSize, - size_t* inputPosAfter, size_t* outputProduced, - size_t* toFlushNow, size_t* zstdResult) +void FIO_rust_zstd_compressStreamDisplay( + int directive, size_t inputPos, size_t inputSize, size_t outputProduced) { - FIO_rust_zstd_projection_context_t* const context = - (FIO_rust_zstd_projection_context_t*)opaque; - ZSTD_inBuffer inBuff; - ZSTD_outBuffer outBuff; - size_t toFlush; - size_t result; - FIO_setInBuffer(&inBuff, input, inputSize, inputPos); - FIO_setOutBuffer(&outBuff, output, outputSize, 0); - toFlush = ZSTD_toFlushNow(context->cctx); - result = ZSTD_compressStream2( - context->cctx, &outBuff, &inBuff, (ZSTD_EndDirective)directive); - - *inputPosAfter = inBuff.pos; - *outputProduced = outBuff.pos; - *toFlushNow = toFlush; - *zstdResult = result; - if (!ZSTD_isError(result)) { - DISPLAYLEVEL(6, "ZSTD_compress_generic(end:%u) => input pos(%u)<=(%u)size ; output generated %u bytes \n", - (unsigned)directive, (unsigned)inBuff.pos, - (unsigned)inBuff.size, (unsigned)outBuff.pos); - } - (void)srcFileName; - return ZSTD_isError(result); + DISPLAYLEVEL(6, "ZSTD_compress_generic(end:%u) => input pos(%u)<=(%u)size ; output generated %u bytes \n", + (unsigned)directive, (unsigned)inputPos, + (unsigned)inputSize, (unsigned)outputProduced); } static void FIO_rust_zstd_iteration(void* opaque, const char* srcFileName, @@ -2766,7 +2748,7 @@ FIO_rust_compressZstdCallback(void* fCtx, void* prefs, void* ress, memset(&projection, 0, sizeof(projection)); projection.readOpaque = (void*)ressPtr->readCtx; projection.writeOpaque = (void*)ressPtr->writeCtx; - projection.codecOpaque = &context; + projection.codecOpaque = (void*)ressPtr->cctx; projection.policyOpaque = &context; projection.readBufferSize = ZSTD_CStreamInSize(); projection.readFill = FIO_rust_zstd_readFill; diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index 0ff6d9d9f..cd4f4d67b 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -26,6 +26,19 @@ use std::ptr; use std::sync::{Arc, Condvar, Mutex}; use std::thread::{self, JoinHandle}; +use crate::zstd_compress::{ZSTD_inBuffer, ZSTD_outBuffer}; + +#[cfg(not(test))] +unsafe extern "C" { + fn ZSTD_toFlushNow(cctx: *mut c_void) -> usize; + fn FIO_rust_zstd_compressStreamDisplay( + directive: c_int, + input_pos: usize, + input_size: usize, + output_produced: usize, + ); +} + const MAX_IO_JOBS: usize = 10; const IO_QUEUE_SIZE: usize = MAX_IO_JOBS - 2; const SPARSE_SEGMENT_SIZE: usize = 32 * 1024; @@ -3318,10 +3331,106 @@ pub unsafe extern "C" fn FIO_rust_zstd_adapt( zstd_adapt_policy(policy, projection) } +type FIO_rust_zstd_to_flush_now_fn = unsafe extern "C" fn(*mut c_void) -> usize; +type FIO_rust_zstd_compress_stream2_fn = + unsafe extern "C" fn(*mut c_void, *mut ZSTD_outBuffer, *mut ZSTD_inBuffer, c_int) -> usize; +type FIO_rust_zstd_compress_display_fn = unsafe extern "C" fn(c_int, usize, usize, usize); + +/// Build the public stream-buffer views, call the Rust compressor, and publish +/// the same scalar results as the former C-only callback. The codec context +/// stays opaque; only the public buffer ABI crosses into the compressor. +unsafe fn fio_zstd_compress_stream_with( + cctx: *mut c_void, + directive: c_int, + input: *const u8, + input_size: usize, + input_pos: usize, + output: *mut u8, + output_size: usize, + input_pos_after: *mut usize, + output_produced: *mut usize, + to_flush_now: *mut usize, + zstd_result: *mut usize, + to_flush_now_fn: FIO_rust_zstd_to_flush_now_fn, + compress_stream2_fn: FIO_rust_zstd_compress_stream2_fn, + display_fn: Option, +) -> c_int { + let mut input_view = ZSTD_inBuffer { + src: input.cast::(), + size: input_size, + pos: input_pos, + }; + let mut output_view = ZSTD_outBuffer { + dst: output.cast::(), + size: output_size, + pos: 0, + }; + + let pending_flush = unsafe { to_flush_now_fn(cctx) }; + let result = unsafe { compress_stream2_fn(cctx, &mut output_view, &mut input_view, directive) }; + + unsafe { + *input_pos_after = input_view.pos; + *output_produced = output_view.pos; + *to_flush_now = pending_flush; + *zstd_result = result; + } + + if !crate::errors::ERR_isError(result) { + if let Some(display) = display_fn { + unsafe { + display(directive, input_view.pos, input_view.size, output_view.pos); + } + } + 0 + } else { + 1 + } +} + +/// The codec callback used by the zstd file-I/O projection. C retains only +/// the iteration callback's private adaptive/diagnostic context; this callback +/// receives the opaque `ZSTD_CCtx` directly and delegates to Rust's public +/// `ZSTD_compressStream2` implementation. +#[cfg(not(test))] +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_zstd_compressStream( + cctx: *mut c_void, + _src_file_name: *const c_char, + directive: c_int, + input: *const u8, + input_size: usize, + input_pos: usize, + output: *mut u8, + output_size: usize, + input_pos_after: *mut usize, + output_produced: *mut usize, + to_flush_now: *mut usize, + zstd_result: *mut usize, +) -> c_int { + unsafe { + fio_zstd_compress_stream_with( + cctx, + directive, + input, + input_size, + input_pos, + output, + output_size, + input_pos_after, + output_produced, + to_flush_now, + zstd_result, + ZSTD_toFlushNow, + crate::zstd_compress::ZSTD_compressStream2, + Some(FIO_rust_zstd_compressStreamDisplay), + ) + } +} + /// Compresses one zstd frame through the C-owned zstd context and adaptive -/// policy. Rust owns the stream loop and exact pool accounting; C callbacks -/// retain `ZSTD_compressStream2()`, adaptive diagnostics, and all private CLI -/// state. +/// policy. Rust owns the stream loop and codec call; C callbacks retain the +/// adaptive diagnostics and all private CLI state. #[no_mangle] pub unsafe extern "C" fn FIO_rust_compressZstdFrame( projection: *const FIO_rust_zstd_compress_projection_t, @@ -6796,6 +6905,113 @@ mod tests { assert_eq!(context.nbFilesProcessed, 4); } + struct ZstdCodecCallbackTestState { + pending_flush: usize, + codec_result: usize, + observed_input: Option<(usize, usize, usize)>, + observed_output: Option<(usize, usize, usize)>, + observed_directive: c_int, + } + + unsafe extern "C" fn zstd_test_to_flush_now(opaque: *mut c_void) -> usize { + let state = unsafe { &mut *opaque.cast::() }; + state.pending_flush + } + + unsafe extern "C" fn zstd_test_compress_stream2( + opaque: *mut c_void, + output: *mut ZSTD_outBuffer, + input: *mut ZSTD_inBuffer, + directive: c_int, + ) -> usize { + let state = unsafe { &mut *opaque.cast::() }; + let input = unsafe { &mut *input }; + let output = unsafe { &mut *output }; + state.observed_input = Some((input.src as usize, input.size, input.pos)); + state.observed_output = Some((output.dst as usize, output.size, output.pos)); + state.observed_directive = directive; + input.pos = input.size; + output.pos = 3; + state.codec_result + } + + #[test] + fn zstd_codec_callback_builds_public_views_and_preserves_results() { + let input = b"abcdef"; + let mut output = [0_u8; 8]; + let mut state = ZstdCodecCallbackTestState { + pending_flush: 19, + codec_result: 0, + observed_input: None, + observed_output: None, + observed_directive: -1, + }; + let mut input_pos_after = 0; + let mut output_produced = 0; + let mut to_flush_now = 0; + let mut zstd_result = 0; + let cctx = (&mut state as *mut ZstdCodecCallbackTestState).cast::(); + + assert_eq!( + unsafe { + fio_zstd_compress_stream_with( + cctx, + 2, + input.as_ptr(), + input.len(), + 1, + output.as_mut_ptr(), + output.len(), + &mut input_pos_after, + &mut output_produced, + &mut to_flush_now, + &mut zstd_result, + zstd_test_to_flush_now, + zstd_test_compress_stream2, + None, + ) + }, + 0 + ); + assert_eq!( + state.observed_input, + Some((input.as_ptr() as usize, input.len(), 1)) + ); + assert_eq!( + state.observed_output, + Some((output.as_mut_ptr() as usize, output.len(), 0)) + ); + assert_eq!(state.observed_directive, 2); + assert_eq!(input_pos_after, input.len()); + assert_eq!(output_produced, 3); + assert_eq!(to_flush_now, 19); + assert_eq!(zstd_result, 0); + + state.codec_result = usize::MAX; + assert_eq!( + unsafe { + fio_zstd_compress_stream_with( + cctx, + 0, + input.as_ptr(), + input.len(), + 0, + output.as_mut_ptr(), + output.len(), + &mut input_pos_after, + &mut output_produced, + &mut to_flush_now, + &mut zstd_result, + zstd_test_to_flush_now, + zstd_test_compress_stream2, + None, + ) + }, + 1 + ); + assert_eq!(zstd_result, usize::MAX); + } + struct ZstdProjectionState { input: [u8; 5], input_pos: usize, diff --git a/rust/src/zstd_compress.rs b/rust/src/zstd_compress.rs index 13acc4e6b..0c83967a5 100644 --- a/rust/src/zstd_compress.rs +++ b/rust/src/zstd_compress.rs @@ -6282,16 +6282,16 @@ pub extern "C" fn ZSTD_rust_targetCBlockSizeAction( #[repr(C)] pub struct ZSTD_inBuffer { - src: *const c_void, - size: usize, - pos: usize, + pub src: *const c_void, + pub size: usize, + pub pos: usize, } #[repr(C)] pub struct ZSTD_outBuffer { - dst: *mut c_void, - size: usize, - pos: usize, + pub dst: *mut c_void, + pub size: usize, + pub pos: usize, } #[no_mangle]