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.
This commit is contained in:
+55
-5
@@ -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);
|
||||
|
||||
@@ -259,6 +259,42 @@ pub struct FIO_rust_compress_callbacks_t {
|
||||
pub display_status: Option<FIO_rust_compress_status_display_fn>,
|
||||
}
|
||||
|
||||
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<FIO_rust_compress_multiple_file_fn>,
|
||||
}
|
||||
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::<usize>()
|
||||
);
|
||||
assert!(
|
||||
std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, output_file_name)
|
||||
== 2 * size_of::<usize>()
|
||||
);
|
||||
assert!(
|
||||
std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, opaque)
|
||||
== 3 * size_of::<usize>()
|
||||
);
|
||||
assert!(
|
||||
std::mem::offset_of!(FIO_rust_compress_multiple_projection_t, compress_file)
|
||||
== 4 * size_of::<usize>()
|
||||
);
|
||||
assert!(size_of::<FIO_rust_compress_multiple_file_fn>() == size_of::<usize>());
|
||||
assert!(size_of::<FIO_rust_compress_multiple_projection_t>() == 5 * size_of::<usize>());
|
||||
};
|
||||
|
||||
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::<FIO_rust_compression_context_t>();
|
||||
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<c_int>,
|
||||
}
|
||||
|
||||
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::<MultipleCompressionState>() };
|
||||
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,
|
||||
|
||||
Reference in New Issue
Block a user