diff --git a/programs/fileio.c b/programs/fileio.c index 49a1ee3c7..53bcd84ee 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -1545,6 +1545,81 @@ int FIO_rust_compressFilenameInternal( U64 fileSize, int compressionLevel, const FIO_rust_compress_callbacks_t* callbacks); +enum { + FIO_RUST_COMPRESS_DST_OPEN_OK = 0, +}; +typedef int (*FIO_rust_compress_dst_open_fn)(void* opaque, + const char* srcFileName, + const char* dstFileName, + int transferStat, int* dstFd); +typedef void (*FIO_rust_compress_dst_handler_fn)(void* opaque); +typedef int (*FIO_rust_compress_dst_file_fn)(void* opaque, + const char* dstFileName, + const char* srcFileName, + int compressionLevel); +typedef void (*FIO_rust_compress_dst_stat_fn)(void* opaque, int dstFd, + const char* dstFileName); +typedef int (*FIO_rust_compress_dst_close_fn)(void* opaque); +typedef void (*FIO_rust_compress_dst_remove_fn)(void* opaque, + const char* dstFileName); + +/* Rust owns the per-file destination lifecycle; C retains private resource + * handles, diagnostics, metadata operations, and format dispatch in opaque + * callbacks. */ +typedef struct { + void* opaque; + const char* dstFileName; + const char* srcFileName; + int compressionLevel; + int destinationAlreadyOpen; + int sourceIsStdin; + int destinationIsStdout; + int sourceIsRegular; + FIO_rust_compress_dst_open_fn openDestination; + FIO_rust_compress_dst_handler_fn attachDestination; + FIO_rust_compress_dst_handler_fn addHandler; + FIO_rust_compress_dst_file_fn compress; + FIO_rust_compress_dst_handler_fn clearHandler; + FIO_rust_compress_dst_stat_fn setFDStat; + FIO_rust_compress_dst_close_fn closeDestination; + FIO_rust_compress_dst_handler_fn utimeDestination; + FIO_rust_compress_dst_remove_fn removeDestination; +} FIO_rust_compress_dst_projection_t; +typedef char FIO_rust_compress_dst_opaque_offset[ + (offsetof(FIO_rust_compress_dst_projection_t, opaque) == 0) ? 1 : -1]; +typedef char FIO_rust_compress_dst_dst_name_offset[ + (offsetof(FIO_rust_compress_dst_projection_t, dstFileName) + == sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_compress_dst_src_name_offset[ + (offsetof(FIO_rust_compress_dst_projection_t, srcFileName) + == 2 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_compress_dst_level_offset[ + (offsetof(FIO_rust_compress_dst_projection_t, compressionLevel) + == 3 * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_compress_dst_opened_offset[ + (offsetof(FIO_rust_compress_dst_projection_t, destinationAlreadyOpen) + == 3 * sizeof(void*) + sizeof(int)) ? 1 : -1]; +typedef char FIO_rust_compress_dst_stdin_offset[ + (offsetof(FIO_rust_compress_dst_projection_t, sourceIsStdin) + == 3 * sizeof(void*) + 2 * sizeof(int)) ? 1 : -1]; +typedef char FIO_rust_compress_dst_stdout_offset[ + (offsetof(FIO_rust_compress_dst_projection_t, destinationIsStdout) + == 3 * sizeof(void*) + 3 * sizeof(int)) ? 1 : -1]; +typedef char FIO_rust_compress_dst_regular_offset[ + (offsetof(FIO_rust_compress_dst_projection_t, sourceIsRegular) + == 3 * sizeof(void*) + 4 * sizeof(int)) ? 1 : -1]; +typedef char FIO_rust_compress_dst_callback_offset[ + (offsetof(FIO_rust_compress_dst_projection_t, openDestination) + == ((3 * sizeof(void*) + 5 * sizeof(int) + sizeof(void*) - 1) + / sizeof(void*)) * sizeof(void*)) ? 1 : -1]; +typedef char FIO_rust_compress_dst_projection_size[ + (sizeof(FIO_rust_compress_dst_projection_t) + == ((3 * sizeof(void*) + 5 * sizeof(int) + sizeof(void*) - 1) + / sizeof(void*)) * sizeof(void*) + + 9 * sizeof(FIO_rust_compress_dst_open_fn)) ? 1 : -1]; +int FIO_rust_compressFilenameDstFile( + const FIO_rust_compress_dst_projection_t* projection); + enum { FIO_RUST_COMPRESS_SRC_STAT_FAILED = 0, FIO_RUST_COMPRESS_SRC_STAT_OK = 1, @@ -2851,11 +2926,109 @@ FIO_compressFilename_internal(FIO_ctx_t* const fCtx, } +typedef struct { + FIO_ctx_t* fCtx; + FIO_prefs_t* prefs; + cRess_t* ress; + const char* dstFileName; + const stat_t* srcFileStat; + FILE* dstFile; +} FIO_rust_compress_dst_context_t; + +static int FIO_rust_compressDestinationOpen(void* opaque, + const char* srcFileName, + const char* dstFileName, + int transferStat, int* dstFd) +{ + FIO_rust_compress_dst_context_t* const context = + (FIO_rust_compress_dst_context_t*)opaque; + int const permissions = transferStat + ? TEMPORARY_FILE_PERMISSIONS + : DEFAULT_FILE_PERMISSIONS; + + DISPLAYLEVEL(6, "FIO_compressFilename_dstFile: opening dst: %s \n", dstFileName); + context->dstFile = FIO_openDstFile( + context->fCtx, context->prefs, srcFileName, dstFileName, permissions); + if (context->dstFile == NULL) + return 1; + *dstFd = fileno(context->dstFile); + return FIO_RUST_COMPRESS_DST_OPEN_OK; +} + +static void FIO_rust_compressDestinationAttach(void* opaque) +{ + FIO_rust_compress_dst_context_t* const context = + (FIO_rust_compress_dst_context_t*)opaque; + AIO_WritePool_setFile(context->ress->writeCtx, context->dstFile); +} + +static void FIO_rust_compressAddHandler(void* opaque) +{ + FIO_rust_compress_dst_context_t* const context = + (FIO_rust_compress_dst_context_t*)opaque; + /* Add the handler only after FIO_openDstFile() succeeds. */ + addHandler(context->dstFileName); +} + +static int FIO_rust_compressDestinationFile(void* opaque, + const char* dstFileName, + const char* srcFileName, + int compressionLevel) +{ + FIO_rust_compress_dst_context_t* const context = + (FIO_rust_compress_dst_context_t*)opaque; + return FIO_compressFilename_internal( + context->fCtx, context->prefs, *context->ress, + dstFileName, srcFileName, compressionLevel); +} + +static void FIO_rust_compressClearHandler(void* opaque) +{ + (void)opaque; + clearHandler(); +} + +static void FIO_rust_compressSetFDStat(void* opaque, int dstFd, + const char* dstFileName) +{ + FIO_rust_compress_dst_context_t* const context = + (FIO_rust_compress_dst_context_t*)opaque; + UTIL_setFDStat(dstFd, dstFileName, context->srcFileStat); +} + +static int FIO_rust_compressDestinationClose(void* opaque) +{ + FIO_rust_compress_dst_context_t* const context = + (FIO_rust_compress_dst_context_t*)opaque; + int const result = AIO_WritePool_closeFile(context->ress->writeCtx); + context->dstFile = NULL; + if (result) { + DISPLAYLEVEL(1, "zstd: %s: %s \n", context->dstFileName, strerror(errno)); + return 1; + } + return 0; +} + +static void FIO_rust_compressDestinationUtime(void* opaque) +{ + FIO_rust_compress_dst_context_t* const context = + (FIO_rust_compress_dst_context_t*)opaque; + UTIL_utime(context->dstFileName, context->srcFileStat); +} + +static void FIO_rust_compressDestinationRemove(void* opaque, + const char* dstFileName) +{ + (void)opaque; + (void)FIO_removeFile(dstFileName); +} + /*! FIO_compressFilename_dstFile() : - * open dstFileName, or pass-through if ress.file != NULL, + * open the destination, or pass through if the write pool already owns it, * then start compression with FIO_compressFilename_internal(). - * Manages source removal (--rm) and file permissions transfer. - * note : ress.srcFile must be != NULL, + * Rust owns the lifecycle ordering; C retains opaque resources, diagnostics, + * metadata operations, and format-specific compression callbacks. + * note : ress.readCtx must already have a source file attached, * so reach this function through FIO_compressFilename_srcFile(). * @return : 0 : compression completed correctly, * 1 : pb @@ -2868,62 +3041,41 @@ static int FIO_compressFilename_dstFile(FIO_ctx_t* const fCtx, const stat_t* srcFileStat, int compressionLevel) { - int closeDstFile = 0; - int result; - int transferStat = 0; - int dstFd = -1; + FIO_rust_compress_dst_context_t context; + FIO_rust_compress_dst_projection_t projection; assert(AIO_ReadPool_getFile(ress.readCtx) != NULL); - if (AIO_WritePool_getFile(ress.writeCtx) == NULL) { - int dstFileInitialPermissions = DEFAULT_FILE_PERMISSIONS; - if ( strcmp (srcFileName, stdinmark) - && strcmp (dstFileName, stdoutmark) - && UTIL_isRegularFileStat(srcFileStat) ) { - transferStat = 1; - dstFileInitialPermissions = TEMPORARY_FILE_PERMISSIONS; - } - closeDstFile = 1; - DISPLAYLEVEL(6, "FIO_compressFilename_dstFile: opening dst: %s \n", dstFileName); - { FILE *dstFile = FIO_openDstFile(fCtx, prefs, srcFileName, dstFileName, dstFileInitialPermissions); - if (dstFile==NULL) return 1; /* could not open dstFileName */ - dstFd = fileno(dstFile); - AIO_WritePool_setFile(ress.writeCtx, dstFile); - } - /* Must only be added after FIO_openDstFile() succeeds. - * Otherwise we may delete the destination file if it already exists, - * and the user presses Ctrl-C when asked if they wish to overwrite. - */ - addHandler(dstFileName); - } + memset(&context, 0, sizeof(context)); + context.fCtx = fCtx; + context.prefs = prefs; + context.ress = &ress; + context.dstFileName = dstFileName; + context.srcFileStat = srcFileStat; - result = FIO_compressFilename_internal(fCtx, prefs, ress, dstFileName, srcFileName, compressionLevel); + memset(&projection, 0, sizeof(projection)); + projection.opaque = &context; + projection.dstFileName = dstFileName; + projection.srcFileName = srcFileName; + projection.compressionLevel = compressionLevel; + projection.destinationAlreadyOpen = AIO_WritePool_getFile(ress.writeCtx) != NULL; + projection.sourceIsStdin = !strcmp(srcFileName, stdinmark); + projection.destinationIsStdout = !strcmp(dstFileName, stdoutmark); + projection.sourceIsRegular = (projection.sourceIsStdin + || projection.destinationIsStdout) + ? 0 + : UTIL_isRegularFileStat(srcFileStat); + projection.openDestination = FIO_rust_compressDestinationOpen; + projection.attachDestination = FIO_rust_compressDestinationAttach; + projection.addHandler = FIO_rust_compressAddHandler; + projection.compress = FIO_rust_compressDestinationFile; + projection.clearHandler = FIO_rust_compressClearHandler; + projection.setFDStat = FIO_rust_compressSetFDStat; + projection.closeDestination = FIO_rust_compressDestinationClose; + projection.utimeDestination = FIO_rust_compressDestinationUtime; + projection.removeDestination = FIO_rust_compressDestinationRemove; - if (closeDstFile) { - clearHandler(); - - if (transferStat) { - UTIL_setFDStat(dstFd, dstFileName, srcFileStat); - } - - DISPLAYLEVEL(6, "FIO_compressFilename_dstFile: closing dst: %s \n", dstFileName); - if (AIO_WritePool_closeFile(ress.writeCtx)) { /* error closing file */ - DISPLAYLEVEL(1, "zstd: %s: %s \n", dstFileName, strerror(errno)); - result=1; - } - - if (transferStat) { - UTIL_utime(dstFileName, srcFileStat); - } - - if ( (result != 0) /* operation failure */ - && strcmp(dstFileName, stdoutmark) /* special case : don't remove() stdout */ - ) { - FIO_removeFile(dstFileName); /* remove compression artefact; note don't do anything special if remove() fails */ - } - } - - return result; + return FIO_rust_compressFilenameDstFile(&projection); } /* List used to compare file extensions (used with --exclude-compressed flag) diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index 067182da0..e9f063f21 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -365,6 +365,115 @@ const _: () = { ); }; +pub const FIO_RUST_COMPRESS_DST_OPEN_OK: c_int = 0; + +pub type FIO_rust_compress_dst_open_fn = + unsafe extern "C" fn(*mut c_void, *const c_char, *const c_char, c_int, *mut c_int) -> c_int; +pub type FIO_rust_compress_dst_handler_fn = unsafe extern "C" fn(*mut c_void); +pub type FIO_rust_compress_dst_file_fn = + unsafe extern "C" fn(*mut c_void, *const c_char, *const c_char, c_int) -> c_int; +pub type FIO_rust_compress_dst_stat_fn = unsafe extern "C" fn(*mut c_void, c_int, *const c_char); +pub type FIO_rust_compress_dst_close_fn = unsafe extern "C" fn(*mut c_void) -> c_int; +pub type FIO_rust_compress_dst_remove_fn = unsafe extern "C" fn(*mut c_void, *const c_char); + +/// Rust owns the per-file compression destination lifecycle. C retains the +/// private resource and file handles, diagnostics, metadata operations, and +/// format dispatch behind opaque callbacks. +#[repr(C)] +pub struct FIO_rust_compress_dst_projection_t { + pub opaque: *mut c_void, + pub dst_file_name: *const c_char, + pub src_file_name: *const c_char, + pub compression_level: c_int, + pub destination_already_open: c_int, + pub source_is_stdin: c_int, + pub destination_is_stdout: c_int, + pub source_is_regular: c_int, + pub open_destination: Option, + pub attach_destination: Option, + pub add_handler: Option, + pub compress: Option, + pub clear_handler: Option, + pub set_fd_stat: Option, + pub close_destination: Option, + pub utime_destination: Option, + pub remove_destination: Option, +} +const _: () = { + let callback_offset = (3 * size_of::() + 5 * size_of::() + size_of::() + - 1) + / size_of::() + * size_of::(); + assert!(std::mem::offset_of!(FIO_rust_compress_dst_projection_t, opaque) == 0); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, dst_file_name) + == size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, src_file_name) + == 2 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, compression_level) + == 3 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, destination_already_open) + == 3 * size_of::() + size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, source_is_stdin) + == 3 * size_of::() + 2 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, destination_is_stdout) + == 3 * size_of::() + 3 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, source_is_regular) + == 3 * size_of::() + 4 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, open_destination) + == callback_offset + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, attach_destination) + == callback_offset + size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, add_handler) + == callback_offset + 2 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, compress) + == callback_offset + 3 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, clear_handler) + == callback_offset + 4 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, set_fd_stat) + == callback_offset + 5 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, close_destination) + == callback_offset + 6 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, utime_destination) + == callback_offset + 7 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_compress_dst_projection_t, remove_destination) + == callback_offset + 8 * size_of::() + ); + assert!( + size_of::() == callback_offset + 9 * size_of::() + ); +}; + pub type FIO_rust_compress_multiple_file_fn = unsafe extern "C" fn(*mut c_void, *const c_char, *const c_char) -> c_int; pub type FIO_rust_compress_multiple_separate_file_fn = @@ -2616,6 +2725,113 @@ pub unsafe extern "C" fn FIO_rust_compressFilenameSrcFile( result } +unsafe fn compress_destination(projection: &FIO_rust_compress_dst_projection_t) -> c_int { + let compress = projection + .compress + .expect("compression callback is required"); + if projection.destination_already_open != 0 { + return unsafe { + compress( + projection.opaque, + projection.dst_file_name, + projection.src_file_name, + projection.compression_level, + ) + }; + } + + let transfer_stat = projection.source_is_stdin == 0 + && projection.destination_is_stdout == 0 + && projection.source_is_regular != 0; + let mut destination_fd = -1; + let open_destination = projection + .open_destination + .expect("destination open callback is required"); + let open_status = unsafe { + open_destination( + projection.opaque, + projection.src_file_name, + projection.dst_file_name, + c_int::from(transfer_stat), + &mut destination_fd, + ) + }; + if open_status != FIO_RUST_COMPRESS_DST_OPEN_OK { + return 1; + } + + let attach_destination = projection + .attach_destination + .expect("destination attachment callback is required"); + unsafe { attach_destination(projection.opaque) }; + + let add_handler = projection + .add_handler + .expect("destination handler callback is required"); + unsafe { add_handler(projection.opaque) }; + + let mut result = unsafe { + compress( + projection.opaque, + projection.dst_file_name, + projection.src_file_name, + projection.compression_level, + ) + }; + + let clear_handler = projection + .clear_handler + .expect("handler-clear callback is required"); + unsafe { clear_handler(projection.opaque) }; + + if transfer_stat { + let set_fd_stat = projection + .set_fd_stat + .expect("destination stat callback is required"); + unsafe { set_fd_stat(projection.opaque, destination_fd, projection.dst_file_name) }; + } + + let close_destination = projection + .close_destination + .expect("destination close callback is required"); + if unsafe { close_destination(projection.opaque) } != 0 { + result = 1; + } + + if transfer_stat { + let utime_destination = projection + .utime_destination + .expect("destination timestamp callback is required"); + unsafe { utime_destination(projection.opaque) }; + } + + if result != 0 && projection.destination_is_stdout == 0 { + let remove_destination = projection + .remove_destination + .expect("destination removal callback is required"); + unsafe { remove_destination(projection.opaque, projection.dst_file_name) }; + } + + result +} + +/// Runs the per-file compression destination policy around C-owned resources +/// and the existing format-dispatch callback. The callback order mirrors the +/// original `FIO_compressFilename_dstFile()` lifecycle, including shared +/// destinations, metadata, close errors, and failed-output cleanup. +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_compressFilenameDstFile( + projection: *const FIO_rust_compress_dst_projection_t, +) -> c_int { + assert!(!projection.is_null()); + let projection = unsafe { &*projection }; + assert!(!projection.opaque.is_null()); + assert!(!projection.dst_file_name.is_null()); + assert!(!projection.src_file_name.is_null()); + + unsafe { compress_destination(projection) } +} + unsafe fn decompress_destination( projection: &FIO_rust_decompress_file_projection_t, source_is_regular: c_int, @@ -5012,6 +5228,211 @@ mod tests { ); } + const COMPRESS_DST_POLICY_OPEN: u8 = 1; + const COMPRESS_DST_POLICY_ATTACH: u8 = 2; + const COMPRESS_DST_POLICY_ADD_HANDLER: u8 = 3; + const COMPRESS_DST_POLICY_COMPRESS: u8 = 4; + const COMPRESS_DST_POLICY_CLEAR_HANDLER: u8 = 5; + const COMPRESS_DST_POLICY_SET_FD_STAT: u8 = 6; + const COMPRESS_DST_POLICY_CLOSE: u8 = 7; + const COMPRESS_DST_POLICY_UTIME: u8 = 8; + const COMPRESS_DST_POLICY_REMOVE: u8 = 9; + + #[derive(Default)] + struct CompressDestinationPolicyState { + events: Vec, + transfer_stats: Vec, + destination_fds: Vec, + compression_levels: Vec, + open_status: c_int, + close_status: c_int, + compression_status: c_int, + } + + unsafe extern "C" fn compress_destination_policy_open( + opaque: *mut c_void, + _src_file_name: *const c_char, + _dst_file_name: *const c_char, + transfer_stat: c_int, + destination_fd: *mut c_int, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(COMPRESS_DST_POLICY_OPEN); + state.transfer_stats.push(transfer_stat); + unsafe { *destination_fd = 41 }; + state.open_status + } + + unsafe extern "C" fn compress_destination_policy_handler(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(COMPRESS_DST_POLICY_ATTACH); + } + + unsafe extern "C" fn compress_destination_policy_add_handler(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(COMPRESS_DST_POLICY_ADD_HANDLER); + } + + unsafe extern "C" fn compress_destination_policy_clear_handler(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(COMPRESS_DST_POLICY_CLEAR_HANDLER); + } + + unsafe extern "C" fn compress_destination_policy_compress( + opaque: *mut c_void, + _dst_file_name: *const c_char, + _src_file_name: *const c_char, + compression_level: c_int, + ) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(COMPRESS_DST_POLICY_COMPRESS); + state.compression_levels.push(compression_level); + state.compression_status + } + + unsafe extern "C" fn compress_destination_policy_set_fd_stat( + opaque: *mut c_void, + destination_fd: c_int, + _dst_file_name: *const c_char, + ) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(COMPRESS_DST_POLICY_SET_FD_STAT); + state.destination_fds.push(destination_fd); + } + + unsafe extern "C" fn compress_destination_policy_close(opaque: *mut c_void) -> c_int { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(COMPRESS_DST_POLICY_CLOSE); + state.close_status + } + + unsafe extern "C" fn compress_destination_policy_utime(opaque: *mut c_void) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(COMPRESS_DST_POLICY_UTIME); + } + + unsafe extern "C" fn compress_destination_policy_remove( + opaque: *mut c_void, + _dst_file_name: *const c_char, + ) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(COMPRESS_DST_POLICY_REMOVE); + } + + fn compress_destination_policy_projection( + state: &mut CompressDestinationPolicyState, + destination_already_open: c_int, + source_is_stdin: c_int, + destination_is_stdout: c_int, + source_is_regular: c_int, + ) -> FIO_rust_compress_dst_projection_t { + FIO_rust_compress_dst_projection_t { + opaque: (state as *mut CompressDestinationPolicyState).cast(), + dst_file_name: c"destination".as_ptr(), + src_file_name: c"source".as_ptr(), + compression_level: 7, + destination_already_open, + source_is_stdin, + destination_is_stdout, + source_is_regular, + open_destination: Some(compress_destination_policy_open), + attach_destination: Some(compress_destination_policy_handler), + add_handler: Some(compress_destination_policy_add_handler), + compress: Some(compress_destination_policy_compress), + clear_handler: Some(compress_destination_policy_clear_handler), + set_fd_stat: Some(compress_destination_policy_set_fd_stat), + close_destination: Some(compress_destination_policy_close), + utime_destination: Some(compress_destination_policy_utime), + remove_destination: Some(compress_destination_policy_remove), + } + } + + #[test] + fn compression_destination_policy_preserves_metadata_and_cleanup_order() { + let mut state = CompressDestinationPolicyState::default(); + let projection = compress_destination_policy_projection(&mut state, 0, 0, 0, 1); + + assert_eq!(unsafe { FIO_rust_compressFilenameDstFile(&projection) }, 0); + assert_eq!(state.transfer_stats, vec![1]); + assert_eq!(state.destination_fds, vec![41]); + assert_eq!(state.compression_levels, vec![7]); + assert_eq!( + state.events, + vec![ + COMPRESS_DST_POLICY_OPEN, + COMPRESS_DST_POLICY_ATTACH, + COMPRESS_DST_POLICY_ADD_HANDLER, + COMPRESS_DST_POLICY_COMPRESS, + COMPRESS_DST_POLICY_CLEAR_HANDLER, + COMPRESS_DST_POLICY_SET_FD_STAT, + COMPRESS_DST_POLICY_CLOSE, + COMPRESS_DST_POLICY_UTIME, + ] + ); + } + + #[test] + fn compression_destination_policy_propagates_close_error_and_removes_output() { + let mut state = CompressDestinationPolicyState { + close_status: 1, + ..CompressDestinationPolicyState::default() + }; + let projection = compress_destination_policy_projection(&mut state, 0, 0, 0, 1); + + assert_eq!(unsafe { FIO_rust_compressFilenameDstFile(&projection) }, 1); + assert_eq!( + state.events, + vec![ + COMPRESS_DST_POLICY_OPEN, + COMPRESS_DST_POLICY_ATTACH, + COMPRESS_DST_POLICY_ADD_HANDLER, + COMPRESS_DST_POLICY_COMPRESS, + COMPRESS_DST_POLICY_CLEAR_HANDLER, + COMPRESS_DST_POLICY_SET_FD_STAT, + COMPRESS_DST_POLICY_CLOSE, + COMPRESS_DST_POLICY_UTIME, + COMPRESS_DST_POLICY_REMOVE, + ] + ); + } + + #[test] + fn compression_destination_policy_skips_lifecycle_for_shared_destination() { + let mut state = CompressDestinationPolicyState { + compression_status: 1, + ..CompressDestinationPolicyState::default() + }; + let projection = compress_destination_policy_projection(&mut state, 1, 0, 0, 1); + + assert_eq!(unsafe { FIO_rust_compressFilenameDstFile(&projection) }, 1); + assert_eq!(state.events, vec![COMPRESS_DST_POLICY_COMPRESS]); + assert!(state.transfer_stats.is_empty()); + assert!(state.destination_fds.is_empty()); + } + + #[test] + fn compression_destination_policy_keeps_stdout_cleanup_guard() { + let mut state = CompressDestinationPolicyState { + compression_status: 1, + ..CompressDestinationPolicyState::default() + }; + let projection = compress_destination_policy_projection(&mut state, 0, 0, 1, 1); + + assert_eq!(unsafe { FIO_rust_compressFilenameDstFile(&projection) }, 1); + assert_eq!(state.transfer_stats, vec![0]); + assert_eq!( + state.events, + vec![ + COMPRESS_DST_POLICY_OPEN, + COMPRESS_DST_POLICY_ATTACH, + COMPRESS_DST_POLICY_ADD_HANDLER, + COMPRESS_DST_POLICY_COMPRESS, + COMPRESS_DST_POLICY_CLEAR_HANDLER, + COMPRESS_DST_POLICY_CLOSE, + ] + ); + } + const DECOMPRESS_POLICY_OPEN_SOURCE: u8 = 1; const DECOMPRESS_POLICY_ASYNC: u8 = 2; const DECOMPRESS_POLICY_ATTACH_SOURCE: u8 = 3;