diff --git a/programs/fileio.c b/programs/fileio.c index 3b7011699..f86769747 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -1208,6 +1208,32 @@ typedef char FIO_rust_decompress_multiple_projection_size[ int FIO_rust_decompressMultipleFilenames( const FIO_rust_decompress_multiple_projection_t* projection); +typedef int (*FIO_rust_decompress_multiple_separate_file_fn)( + void* opaque, const char* srcFileName); +typedef struct { + void* fCtx; + const char** srcNamesTable; + void* opaque; + FIO_rust_decompress_multiple_separate_file_fn decompressFile; +} FIO_rust_decompress_multiple_separate_projection_t; +typedef char FIO_rust_decompress_multiple_separate_fctx_offset[ + (offsetof(FIO_rust_decompress_multiple_separate_projection_t, fCtx) == 0) ? 1 : -1]; +typedef char FIO_rust_decompress_multiple_separate_input_names_offset[ + (offsetof(FIO_rust_decompress_multiple_separate_projection_t, srcNamesTable) + == sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_decompress_multiple_separate_opaque_offset[ + (offsetof(FIO_rust_decompress_multiple_separate_projection_t, opaque) + == 2 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_decompress_multiple_separate_callback_offset[ + (offsetof(FIO_rust_decompress_multiple_separate_projection_t, decompressFile) + == 3 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_decompress_multiple_separate_projection_size[ + (sizeof(FIO_rust_decompress_multiple_separate_projection_t) + == 3 * sizeof(void*) + + sizeof(FIO_rust_decompress_multiple_separate_file_fn)) ? 1 : -1]; +int FIO_rust_decompressMultipleSeparateFilenames( + const FIO_rust_decompress_multiple_separate_projection_t* projection); + enum { FIO_RUST_ZSTD_OK = 0, FIO_RUST_ZSTD_COMPRESS_ERROR = 1, @@ -3884,6 +3910,43 @@ static int FIO_rust_decompressMultipleFileCallback(void* opaque, context->fCtx, context->prefs, *context->ress, outFileName, srcFileName); } +typedef struct { + FIO_ctx_t* fCtx; + FIO_prefs_t* prefs; + dRess_t* ress; + const char* outMirroredRootDirName; + const char* outDirName; +} FIO_rust_decompress_multiple_separate_context_t; + +static const char* FIO_determineDstName(const char* srcFileName, const char* outDirName); + +/* Keep destination-name construction, diagnostics, and decompression in C. */ +static int FIO_rust_decompressMultipleSeparateFileCallback(void* opaque, + const char* srcFileName) +{ + FIO_rust_decompress_multiple_separate_context_t* const context = + (FIO_rust_decompress_multiple_separate_context_t*)opaque; + const char* dstFileName = NULL; + + if (context->outMirroredRootDirName) { + char* validMirroredDirName = UTIL_createMirroredDestDirName( + srcFileName, context->outMirroredRootDirName); + if (validMirroredDirName) { + dstFileName = FIO_determineDstName(srcFileName, validMirroredDirName); + free(validMirroredDirName); + } else { + DISPLAYLEVEL(2, "zstd: --output-dir-mirror cannot decompress '%s' into '%s'\n", + srcFileName, context->outMirroredRootDirName); + } + } else { + dstFileName = FIO_determineDstName(srcFileName, context->outDirName); + } + + if (dstFileName == NULL) return 1; + return FIO_decompressSrcFile( + context->fCtx, context->prefs, *context->ress, dstFileName, srcFileName); +} + int FIO_decompressFilename(FIO_ctx_t* const fCtx, FIO_prefs_t* const prefs, @@ -3953,7 +4016,6 @@ FIO_decompressMultipleFilenames(FIO_ctx_t* const fCtx, const char* outDirName, const char* outFileName, const char* dictFileName) { - int status; int error = 0; dRess_t ress = FIO_createDResources(prefs, dictFileName); @@ -3981,28 +4043,18 @@ FIO_decompressMultipleFilenames(FIO_ctx_t* const fCtx, EXM_THROW(72, "Write error : %s : cannot properly close output file", strerror(errno)); } else { + FIO_rust_decompress_multiple_separate_context_t callbackContext = { + fCtx, prefs, &ress, outMirroredRootDirName, outDirName + }; + FIO_rust_decompress_multiple_separate_projection_t projection = { + fCtx, srcNamesTable, &callbackContext, + FIO_rust_decompressMultipleSeparateFileCallback + }; if (outMirroredRootDirName) UTIL_mirrorSourceFilesDirectories(srcNamesTable, (unsigned)fCtx->nbFilesTotal, outMirroredRootDirName); - for (; fCtx->currFileIdx < fCtx->nbFilesTotal; fCtx->currFileIdx++) { /* create dstFileName */ - const char* const srcFileName = srcNamesTable[fCtx->currFileIdx]; - const char* dstFileName = NULL; - if (outMirroredRootDirName) { - char* validMirroredDirName = UTIL_createMirroredDestDirName(srcFileName, outMirroredRootDirName); - if (validMirroredDirName) { - dstFileName = FIO_determineDstName(srcFileName, validMirroredDirName); - free(validMirroredDirName); - } else { - DISPLAYLEVEL(2, "zstd: --output-dir-mirror cannot decompress '%s' into '%s'\n", srcFileName, outMirroredRootDirName); - } - } else { - dstFileName = FIO_determineDstName(srcFileName, outDirName); - } - if (dstFileName == NULL) { error=1; continue; } - status = FIO_decompressSrcFile(fCtx, prefs, ress, dstFileName, srcFileName); - if (!status) fCtx->nbFilesProcessed++; - error |= status; - } + error = FIO_rust_decompressMultipleSeparateFilenames(&projection); + if (outDirName) FIO_checkFilenameCollisions(srcNamesTable , (unsigned)fCtx->nbFilesTotal); } diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index 18f98d559..66d272076 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -334,6 +334,8 @@ const _: () = { pub type FIO_rust_decompress_multiple_file_fn = unsafe extern "C" fn(*mut c_void, *const c_char, *const c_char) -> c_int; +pub type FIO_rust_decompress_multiple_separate_file_fn = + unsafe extern "C" fn(*mut c_void, *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 @@ -368,6 +370,40 @@ const _: () = { assert!(size_of::() == 5 * size_of::()); }; +/// Rust owns only the separate-destination file iteration and aggregate error +/// handling. C retains destination-name construction, source opening, format +/// dispatch, diagnostics, and source removal through the per-source callback. +#[repr(C)] +pub struct FIO_rust_decompress_multiple_separate_projection_t { + pub f_ctx: *mut c_void, + pub src_names_table: *const *const c_char, + pub opaque: *mut c_void, + pub decompress_file: Option, +} +const _: () = { + assert!(std::mem::offset_of!(FIO_rust_decompress_multiple_separate_projection_t, f_ctx) == 0); + assert!( + std::mem::offset_of!( + FIO_rust_decompress_multiple_separate_projection_t, + src_names_table + ) == size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_decompress_multiple_separate_projection_t, opaque) + == 2 * size_of::() + ); + assert!( + std::mem::offset_of!( + FIO_rust_decompress_multiple_separate_projection_t, + decompress_file + ) == 3 * size_of::() + ); + assert!(size_of::() == size_of::()); + assert!( + size_of::() == 4 * 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; @@ -2344,6 +2380,41 @@ pub unsafe extern "C" fn FIO_rust_decompressMultipleFilenames( error } +/// Iterate files that each receive a separate destination. The C callback +/// retains destination-name construction, 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_decompressMultipleSeparateFilenames( + projection: *const FIO_rust_decompress_multiple_separate_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()); + let decompress_file = projection + .decompress_file + .expect("separate-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, 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 { @@ -4668,6 +4739,20 @@ mod tests { status } + unsafe extern "C" fn record_multiple_separate_decompression_file( + opaque: *mut c_void, + 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); + 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()]; @@ -4872,6 +4957,70 @@ mod tests { assert_eq!(context.nbFilesProcessed, 4); } + #[test] + fn decompression_multiple_separate_destinations_preserves_order_and_counters() { + let source_names = [c"zero".as_ptr(), c"one".as_ptr(), c"two".as_ptr()]; + 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_separate_projection_t { + f_ctx: (&mut context as *mut FIO_rust_compression_context_t).cast(), + src_names_table: source_names.as_ptr(), + opaque: (&mut state as *mut MultipleDecompressionState).cast(), + decompress_file: Some(record_multiple_separate_decompression_file), + }; + + assert_eq!( + unsafe { FIO_rust_decompressMultipleSeparateFilenames(&projection) }, + 0 + ); + assert_eq!(state.sources, vec![source_names[1], source_names[2]]); + assert_eq!(context.currFileIdx, 3); + assert_eq!(context.nbFilesProcessed, 7); + } + + #[test] + fn decompression_multiple_separate_destinations_accumulates_errors_without_short_circuiting() { + let source_names = [c"one".as_ptr(), c"two".as_ptr(), c"three".as_ptr()]; + 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_separate_projection_t { + f_ctx: (&mut context as *mut FIO_rust_compression_context_t).cast(), + src_names_table: source_names.as_ptr(), + opaque: (&mut state as *mut MultipleDecompressionState).cast(), + decompress_file: Some(record_multiple_separate_decompression_file), + }; + + assert_eq!( + unsafe { FIO_rust_decompressMultipleSeparateFilenames(&projection) }, + 1 | 4 | 2 + ); + assert_eq!(state.sources, source_names); + assert_eq!(context.currFileIdx, 3); + assert_eq!(context.nbFilesProcessed, 4); + } + struct ZstdProjectionState { input: [u8; 5], input_pos: usize,