diff --git a/programs/fileio.c b/programs/fileio.c index 497c7cf85..3b7011699 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -1179,6 +1179,35 @@ typedef char FIO_rust_compress_multiple_separate_projection_size[ int FIO_rust_compressMultipleSeparateFilenames( const FIO_rust_compress_multiple_separate_projection_t* projection); +typedef int (*FIO_rust_decompress_multiple_file_fn)( + void* opaque, const char* outFileName, const char* srcFileName); +typedef struct { + void* fCtx; + const char** srcNamesTable; + const char* outFileName; + void* opaque; + FIO_rust_decompress_multiple_file_fn decompressFile; +} FIO_rust_decompress_multiple_projection_t; +typedef char FIO_rust_decompress_multiple_fctx_offset[ + (offsetof(FIO_rust_decompress_multiple_projection_t, fCtx) == 0) ? 1 : -1]; +typedef char FIO_rust_decompress_multiple_input_names_offset[ + (offsetof(FIO_rust_decompress_multiple_projection_t, srcNamesTable) + == sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_decompress_multiple_output_name_offset[ + (offsetof(FIO_rust_decompress_multiple_projection_t, outFileName) + == 2 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_decompress_multiple_opaque_offset[ + (offsetof(FIO_rust_decompress_multiple_projection_t, opaque) + == 3 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_decompress_multiple_callback_offset[ + (offsetof(FIO_rust_decompress_multiple_projection_t, decompressFile) + == 4 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_decompress_multiple_projection_size[ + (sizeof(FIO_rust_decompress_multiple_projection_t) + == 4 * sizeof(void*) + sizeof(FIO_rust_decompress_multiple_file_fn)) ? 1 : -1]; +int FIO_rust_decompressMultipleFilenames( + const FIO_rust_decompress_multiple_projection_t* projection); + enum { FIO_RUST_ZSTD_OK = 0, FIO_RUST_ZSTD_COMPRESS_ERROR = 1, @@ -3838,6 +3867,23 @@ static int FIO_decompressSrcFile(FIO_ctx_t* const fCtx, FIO_prefs_t* const prefs return result; } +typedef struct { + FIO_ctx_t* fCtx; + FIO_prefs_t* prefs; + dRess_t* ress; +} FIO_rust_decompress_multiple_context_t; + +/* Keep source opening, format dispatch, diagnostics, and --rm in C. */ +static int FIO_rust_decompressMultipleFileCallback(void* opaque, + const char* outFileName, + const char* srcFileName) +{ + FIO_rust_decompress_multiple_context_t* const context = + (FIO_rust_decompress_multiple_context_t*)opaque; + return FIO_decompressSrcFile( + context->fCtx, context->prefs, *context->ress, outFileName, srcFileName); +} + int FIO_decompressFilename(FIO_ctx_t* const fCtx, FIO_prefs_t* const prefs, @@ -3921,10 +3967,15 @@ FIO_decompressMultipleFilenames(FIO_ctx_t* const fCtx, if (dstFile == 0) EXM_THROW(19, "cannot open %s", outFileName); AIO_WritePool_setFile(ress.writeCtx, dstFile); } - for (; fCtx->currFileIdx < fCtx->nbFilesTotal; fCtx->currFileIdx++) { - status = FIO_decompressSrcFile(fCtx, prefs, ress, outFileName, srcNamesTable[fCtx->currFileIdx]); - if (!status) fCtx->nbFilesProcessed++; - error |= status; + { + FIO_rust_decompress_multiple_context_t callbackContext = { + fCtx, prefs, &ress + }; + FIO_rust_decompress_multiple_projection_t projection = { + fCtx, srcNamesTable, outFileName, + &callbackContext, FIO_rust_decompressMultipleFileCallback + }; + error = FIO_rust_decompressMultipleFilenames(&projection); } if ((!prefs->testMode) && (AIO_WritePool_closeFile(ress.writeCtx))) EXM_THROW(72, "Write error : %s : cannot properly close output file", diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index f4071bfe2..18f98d559 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -332,6 +332,42 @@ const _: () = { ); }; +pub type FIO_rust_decompress_multiple_file_fn = + unsafe extern "C" fn(*mut c_void, *const c_char, *const c_char) -> c_int; + +/// Rust owns only the shared-destination file iteration and aggregate error +/// handling. C retains source opening, format dispatch, diagnostics, and +/// source removal through the per-source callback. +#[repr(C)] +pub struct FIO_rust_decompress_multiple_projection_t { + pub f_ctx: *mut c_void, + pub src_names_table: *const *const c_char, + pub out_file_name: *const c_char, + pub opaque: *mut c_void, + pub decompress_file: Option, +} +const _: () = { + assert!(std::mem::offset_of!(FIO_rust_decompress_multiple_projection_t, f_ctx) == 0); + assert!( + std::mem::offset_of!(FIO_rust_decompress_multiple_projection_t, src_names_table) + == size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_decompress_multiple_projection_t, out_file_name) + == 2 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_decompress_multiple_projection_t, opaque) + == 3 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_decompress_multiple_projection_t, decompress_file) + == 4 * size_of::() + ); + assert!(size_of::() == size_of::()); + assert!(size_of::() == 5 * size_of::()); +}; + pub const FIO_RUST_ZSTD_OK: c_int = 0; pub const FIO_RUST_ZSTD_COMPRESS_ERROR: c_int = 1; pub const FIO_RUST_ZSTD_INCOMPLETE_INPUT: c_int = 2; @@ -2272,6 +2308,42 @@ pub unsafe extern "C" fn FIO_rust_compressMultipleSeparateFilenames( error } +/// Iterate files that share one already-open decompression destination. The +/// C callback retains source opening, format dispatch, diagnostics, and +/// `--rm`; Rust preserves the non-short-circuiting loop and counter updates. +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_decompressMultipleFilenames( + projection: *const FIO_rust_decompress_multiple_projection_t, +) -> c_int { + assert!(!projection.is_null()); + let projection = unsafe { &*projection }; + assert!(!projection.f_ctx.is_null()); + assert!(!projection.src_names_table.is_null()); + assert!(!projection.out_file_name.is_null()); + let decompress_file = projection + .decompress_file + .expect("shared-destination decompression callback is required"); + + let f_ctx = projection.f_ctx.cast::(); + let mut error = 0; + while unsafe { (*f_ctx).currFileIdx < (*f_ctx).nbFilesTotal } { + let file_index = unsafe { (*f_ctx).currFileIdx as usize }; + let src_file_name = unsafe { *projection.src_names_table.add(file_index) }; + let status = + unsafe { decompress_file(projection.opaque, projection.out_file_name, src_file_name) }; + if status == 0 { + unsafe { + (*f_ctx).nbFilesProcessed = (*f_ctx).nbFilesProcessed.wrapping_add(1); + } + } + error |= status; + unsafe { + (*f_ctx).currFileIdx = (*f_ctx).currFileIdx.wrapping_add(1); + } + } + error +} + #[inline] fn zstd_adapt_level(projection: &FIO_rust_zstd_adapt_projection_t, slower: bool) -> c_int { if slower { @@ -4573,6 +4645,29 @@ mod tests { status } + #[derive(Default)] + struct MultipleDecompressionState { + sources: Vec<*const c_char>, + destinations: Vec<*const c_char>, + statuses: Vec, + } + + unsafe extern "C" fn record_multiple_decompression_file( + opaque: *mut c_void, + destination: *const c_char, + source: *const c_char, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + let status = state + .statuses + .get(state.sources.len()) + .copied() + .unwrap_or(0); + state.sources.push(source); + state.destinations.push(destination); + status + } + #[test] fn compression_multiple_separate_destinations_preserves_order_and_counters() { let source_names = [c"zero".as_ptr(), c"one".as_ptr(), c"two".as_ptr()]; @@ -4707,6 +4802,76 @@ mod tests { assert_eq!(context.nbFilesProcessed, 4); } + #[test] + fn decompression_multiple_shared_destination_preserves_order_and_counters() { + let source_names = [c"zero".as_ptr(), c"one".as_ptr(), c"two".as_ptr()]; + let destination = c"archive"; + let mut context = FIO_rust_compression_context_t { + nbFilesTotal: 3, + hasStdinInput: 0, + hasStdoutOutput: 0, + currFileIdx: 1, + nbFilesProcessed: 5, + totalBytesInput: 0, + totalBytesOutput: 0, + }; + let mut state = MultipleDecompressionState { + statuses: vec![0, 0], + ..MultipleDecompressionState::default() + }; + let projection = FIO_rust_decompress_multiple_projection_t { + f_ctx: (&mut context as *mut FIO_rust_compression_context_t).cast(), + src_names_table: source_names.as_ptr(), + out_file_name: destination.as_ptr(), + opaque: (&mut state as *mut MultipleDecompressionState).cast(), + decompress_file: Some(record_multiple_decompression_file), + }; + + assert_eq!( + unsafe { FIO_rust_decompressMultipleFilenames(&projection) }, + 0 + ); + assert_eq!(state.sources, vec![source_names[1], source_names[2]]); + assert_eq!(state.destinations, vec![destination.as_ptr(); 2]); + assert_eq!(context.currFileIdx, 3); + assert_eq!(context.nbFilesProcessed, 7); + } + + #[test] + fn decompression_multiple_shared_destination_accumulates_errors_without_short_circuiting() { + let source_names = [c"one".as_ptr(), c"two".as_ptr(), c"three".as_ptr()]; + let destination = c"archive"; + let mut context = FIO_rust_compression_context_t { + nbFilesTotal: 3, + hasStdinInput: 0, + hasStdoutOutput: 0, + currFileIdx: 0, + nbFilesProcessed: 4, + totalBytesInput: 0, + totalBytesOutput: 0, + }; + let mut state = MultipleDecompressionState { + statuses: vec![1, 4, 2], + ..MultipleDecompressionState::default() + }; + let projection = FIO_rust_decompress_multiple_projection_t { + f_ctx: (&mut context as *mut FIO_rust_compression_context_t).cast(), + src_names_table: source_names.as_ptr(), + out_file_name: destination.as_ptr(), + opaque: (&mut state as *mut MultipleDecompressionState).cast(), + decompress_file: Some(record_multiple_decompression_file), + }; + + assert_eq!( + unsafe { FIO_rust_decompressMultipleFilenames(&projection) }, + 1 | 4 | 2 + ); + assert_eq!(state.sources, source_names); + assert_eq!(state.destinations, vec![destination.as_ptr(); 3]); + assert_eq!(context.currFileIdx, 3); + assert_eq!(context.nbFilesProcessed, 4); + } + struct ZstdProjectionState { input: [u8; 5], input_pos: usize,