From 035001b769891913c58f89ef5358baec9065e0b9 Mon Sep 17 00:00:00 2001 From: ddidderr Date: Sun, 19 Jul 2026 10:31:19 +0200 Subject: [PATCH] feat(cli): move shared-destination scheduling into Rust Move the shared-output arm of FIO_compressMultipleFilenames into a Rust scheduler. Rust now owns input-order iteration, processed-file accounting, index advancement, and non-short-circuiting error aggregation while C retains file validation, private resources, callbacks, diagnostics, and destination lifetime. Test Plan: - cargo +nightly fmt --manifest-path rust/Cargo.toml - cargo test --manifest-path rust/Cargo.toml compression_multiple_shared_destination -- --test-threads=1 (2 passed) - cargo clippy --manifest-path rust/Cargo.toml --tests -- -D warnings - make -B -C lib -j1 lib - make -B -C programs -j1 zstd - make -B -C tests -j1 test-cli-tests (41 passed) - all commands used CARGO_BUILD_JOBS=1 where applicable and ulimit -v 41943040; no heavyweight jobs ran concurrently Commit is intentionally unsigned because GPG pinentry hangs in this non-interactive environment. --- programs/fileio.c | 60 +++++++++++-- rust/src/fileio_asyncio.rs | 171 +++++++++++++++++++++++++++++++++++++ 2 files changed, 226 insertions(+), 5 deletions(-) diff --git a/programs/fileio.c b/programs/fileio.c index 806f9dde2..046159273 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -1125,6 +1125,35 @@ typedef struct { FIO_rust_compress_status_display_fn display_status; } FIO_rust_compress_callbacks_t; +typedef int (*FIO_rust_compress_multiple_file_fn)( + void* opaque, const char* dstFileName, const char* srcFileName); +typedef struct { + void* fCtx; + const char** inFileNamesTable; + const char* outFileName; + void* opaque; + FIO_rust_compress_multiple_file_fn compressFile; +} FIO_rust_compress_multiple_projection_t; +typedef char FIO_rust_compress_multiple_fctx_offset[ + (offsetof(FIO_rust_compress_multiple_projection_t, fCtx) == 0) ? 1 : -1]; +typedef char FIO_rust_compress_multiple_input_names_offset[ + (offsetof(FIO_rust_compress_multiple_projection_t, inFileNamesTable) + == sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_compress_multiple_output_name_offset[ + (offsetof(FIO_rust_compress_multiple_projection_t, outFileName) + == 2 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_compress_multiple_opaque_offset[ + (offsetof(FIO_rust_compress_multiple_projection_t, opaque) + == 3 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_compress_multiple_callback_offset[ + (offsetof(FIO_rust_compress_multiple_projection_t, compressFile) + == 4 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_compress_multiple_projection_size[ + (sizeof(FIO_rust_compress_multiple_projection_t) + == 4 * sizeof(void*) + sizeof(FIO_rust_compress_multiple_file_fn)) ? 1 : -1]; +int FIO_rust_compressMultipleFilenames( + const FIO_rust_compress_multiple_projection_t* projection); + enum { FIO_RUST_ZSTD_OK = 0, FIO_RUST_ZSTD_COMPRESS_ERROR = 1, @@ -2897,6 +2926,24 @@ static unsigned long long FIO_getLargestFileSize(const char** inFileNames, unsig return FIO_rust_getLargestFileSize(inFileNames, nbFiles); } +typedef struct { + FIO_ctx_t* fCtx; + FIO_prefs_t* prefs; + cRess_t* ress; + int compressionLevel; +} FIO_rust_compress_multiple_context_t; + +static int FIO_rust_compressMultipleFileCallback(void* opaque, + const char* dstFileName, + const char* srcFileName) +{ + FIO_rust_compress_multiple_context_t* const context = + (FIO_rust_compress_multiple_context_t*)opaque; + return FIO_compressFilename_srcFile( + context->fCtx, context->prefs, *context->ress, + dstFileName, srcFileName, context->compressionLevel); +} + /* FIO_compressMultipleFilenames() : * compress nbFiles files * into either one destination (outFileName), @@ -2930,12 +2977,15 @@ int FIO_compressMultipleFilenames(FIO_ctx_t* const fCtx, if (dstFile == NULL) { /* could not open outFileName */ error = 1; } else { + FIO_rust_compress_multiple_context_t callbackContext = { + fCtx, prefs, &ress, compressionLevel + }; + FIO_rust_compress_multiple_projection_t projection = { + fCtx, inFileNamesTable, outFileName, + &callbackContext, FIO_rust_compressMultipleFileCallback + }; AIO_WritePool_setFile(ress.writeCtx, dstFile); - for (; fCtx->currFileIdx < fCtx->nbFilesTotal; ++fCtx->currFileIdx) { - status = FIO_compressFilename_srcFile(fCtx, prefs, ress, outFileName, inFileNamesTable[fCtx->currFileIdx], compressionLevel); - if (!status) fCtx->nbFilesProcessed++; - error |= status; - } + error = FIO_rust_compressMultipleFilenames(&projection); if (AIO_WritePool_closeFile(ress.writeCtx)) EXM_THROW(29, "Write error (%s) : cannot properly close %s", strerror(errno), outFileName); diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index d2a478447..4e90bef23 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -259,6 +259,42 @@ pub struct FIO_rust_compress_callbacks_t { pub display_status: Option, } +pub type FIO_rust_compress_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 the private preferences/resources and supplies one +/// callback for each source file. +#[repr(C)] +pub struct FIO_rust_compress_multiple_projection_t { + pub f_ctx: *mut c_void, + pub input_file_names: *const *const c_char, + pub output_file_name: *const c_char, + pub opaque: *mut c_void, + pub compress_file: Option, +} +const _: () = { + assert!(std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, f_ctx) == 0); + assert!( + std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, input_file_names) + == size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, output_file_name) + == 2 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, opaque) + == 3 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, compress_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; @@ -2122,6 +2158,48 @@ pub unsafe extern "C" fn FIO_rust_compressFilenameInternal( FIO_RUST_COMPRESS_OK } +/// Iterate the files that share one already-open destination. The C +/// callback retains source validation, resource state, diagnostics, and +/// compression dispatch; Rust preserves the original non-short-circuiting +/// loop and its counter updates. +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_compressMultipleFilenames( + projection: *const FIO_rust_compress_multiple_projection_t, +) -> c_int { + assert!(!projection.is_null()); + let projection = unsafe { &*projection }; + assert!(!projection.f_ctx.is_null()); + assert!(!projection.input_file_names.is_null()); + assert!(!projection.output_file_name.is_null()); + let compress_file = projection + .compress_file + .expect("shared-destination compression 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.input_file_names.add(file_index) }; + let status = unsafe { + compress_file( + projection.opaque, + projection.output_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 { @@ -4386,6 +4464,99 @@ mod tests { assert_eq!(context.totalBytesOutput, 47); } + #[derive(Default)] + struct MultipleCompressionState { + sources: Vec<*const c_char>, + destinations: Vec<*const c_char>, + statuses: Vec, + } + + unsafe extern "C" fn record_multiple_compression_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_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.zst"; + 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 = MultipleCompressionState { + statuses: vec![0, 0], + ..MultipleCompressionState::default() + }; + let projection = FIO_rust_compress_multiple_projection_t { + f_ctx: (&mut context as *mut FIO_rust_compression_context_t).cast(), + input_file_names: source_names.as_ptr(), + output_file_name: destination.as_ptr(), + opaque: (&mut state as *mut MultipleCompressionState).cast(), + compress_file: Some(record_multiple_compression_file), + }; + + assert_eq!( + unsafe { FIO_rust_compressMultipleFilenames(&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 compression_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.zst"; + 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 = MultipleCompressionState { + statuses: vec![1, 4, 2], + ..MultipleCompressionState::default() + }; + let projection = FIO_rust_compress_multiple_projection_t { + f_ctx: (&mut context as *mut FIO_rust_compression_context_t).cast(), + input_file_names: source_names.as_ptr(), + output_file_name: destination.as_ptr(), + opaque: (&mut state as *mut MultipleCompressionState).cast(), + compress_file: Some(record_multiple_compression_file), + }; + + assert_eq!( + unsafe { FIO_rust_compressMultipleFilenames(&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,