ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(fileio): move source compression policy into Rust

Browse Source 
Move the ordered policy from FIO_compressFilename_srcFile into a narrow Rust
projection: named-source stat and collision checks, compressed-file exclusion,
source open and close, size-based async selection, source attachment, and
successful-only --rm handling now run in Rust. The destination callback still
hands compression to the existing Rust compression loop.

Keep FIO_ctx_t, cRess_t, stat_t, FILE/pool handles, signal handling, and
CLI diagnostics behind opaque C callbacks, so the ABI carries only filenames,
scalar policy flags, and callback pointers. Focused Rust policy tests cover
named-file short circuits, stdin behavior, async threshold selection, and the
successful-compression removal gate.

Test Plan:
- `rustfmt --edition 2021 --check rust/src/fileio_asyncio.rs` — passed
- `git diff --check` and path-limited `git diff --cached --check` — passed
- Static ABI/order/diff inspection only; policy tests were not executed
- cargo/make/native/fuzzer/heavy tests were not run per request
This commit is contained in:
ddidderr
2026-07-20 02:34:59 +02:00
parent edc5a11757
commit 7af1b27140
2 changed files with 605 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/src/fileio_asyncio.rs
+412
View File
@@ -259,6 +259,112 @@ pub struct FIO_rust_compress_callbacks_t {
pub display_status: Option<FIO_rust_compress_status_display_fn>,
}
pub const FIO_RUST_COMPRESS_SRC_STAT_FAILED: c_int = 0;
pub const FIO_RUST_COMPRESS_SRC_STAT_OK: c_int = 1;
pub const FIO_RUST_COMPRESS_SRC_DIRECTORY: c_int = 2;
pub const FIO_RUST_COMPRESS_SRC_DICT_COLLISION: c_int = 3;
const FIO_RUST_COMPRESS_SRC_OPEN_OK: c_int = 0;
const FIO_RUST_COMPRESS_SRC_ASYNC_THRESHOLD: u64 = (1 << 17) * 3;
const FIO_RUST_COMPRESS_SRC_UNKNOWN_SIZE: u64 = u64::MAX;
pub type FIO_rust_compress_src_stat_fn = unsafe extern "C" fn(*mut c_void, *const c_char) -> c_int;
pub type FIO_rust_compress_src_excluded_fn =
unsafe extern "C" fn(*mut c_void, *const c_char) -> c_int;
pub type FIO_rust_compress_src_open_fn =
unsafe extern "C" fn(*mut c_void, *const c_char, *mut u64) -> c_int;
pub type FIO_rust_compress_src_async_fn = unsafe extern "C" fn(*mut c_void, c_int);
pub type FIO_rust_compress_src_attach_fn = unsafe extern "C" fn(*mut c_void);
pub type FIO_rust_compress_src_compress_fn =
unsafe extern "C" fn(*mut c_void, *const c_char, *const c_char, c_int) -> c_int;
pub type FIO_rust_compress_src_close_fn = unsafe extern "C" fn(*mut c_void);
pub type FIO_rust_compress_src_remove_fn = unsafe extern "C" fn(*mut c_void, *const c_char);
/// Rust owns the source-file policy and ordering. C retains the private
/// context/resource/stat objects and implements each operation behind opaque
/// callbacks, including diagnostics, signal handling, and actual compression.
#[repr(C)]
pub struct FIO_rust_compress_src_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 source_is_stdin: c_int,
pub exclude_compressed_files: c_int,
pub remove_src_file: c_int,
pub stat_source: Option<FIO_rust_compress_src_stat_fn>,
pub source_is_excluded: Option<FIO_rust_compress_src_excluded_fn>,
pub open_source: Option<FIO_rust_compress_src_open_fn>,
pub set_async: Option<FIO_rust_compress_src_async_fn>,
pub attach_source: Option<FIO_rust_compress_src_attach_fn>,
pub compress: Option<FIO_rust_compress_src_compress_fn>,
pub close_source: Option<FIO_rust_compress_src_close_fn>,
pub remove_source: Option<FIO_rust_compress_src_remove_fn>,
}
const _: () = {
assert!(std::mem::offset_of!(FIO_rust_compress_src_projection_t, opaque) == 0);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, dst_file_name)
== size_of::<usize>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, src_file_name)
== 2 * size_of::<usize>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, compression_level)
== 3 * size_of::<usize>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, source_is_stdin)
== 3 * size_of::<usize>() + size_of::<c_int>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, exclude_compressed_files)
== 3 * size_of::<usize>() + 2 * size_of::<c_int>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, remove_src_file)
== 3 * size_of::<usize>() + 3 * size_of::<c_int>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, stat_source)
== 3 * size_of::<usize>() + 4 * size_of::<c_int>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, source_is_excluded)
== 3 * size_of::<usize>() + 4 * size_of::<c_int>() + size_of::<usize>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, open_source)
== 3 * size_of::<usize>() + 4 * size_of::<c_int>() + 2 * size_of::<usize>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, set_async)
== 3 * size_of::<usize>() + 4 * size_of::<c_int>() + 3 * size_of::<usize>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, attach_source)
== 3 * size_of::<usize>() + 4 * size_of::<c_int>() + 4 * size_of::<usize>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, compress)
== 3 * size_of::<usize>() + 4 * size_of::<c_int>() + 5 * size_of::<usize>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, close_source)
== 3 * size_of::<usize>() + 4 * size_of::<c_int>() + 6 * size_of::<usize>()
);
assert!(
std::mem::offset_of!(FIO_rust_compress_src_projection_t, remove_source)
== 3 * size_of::<usize>() + 4 * size_of::<c_int>() + 7 * size_of::<usize>()
);
assert!(
size_of::<FIO_rust_compress_src_projection_t>()
== 3 * size_of::<usize>() + 4 * size_of::<c_int>() + 8 * size_of::<usize>()
);
};
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 =
@@ -2267,6 +2373,96 @@ pub unsafe extern "C" fn FIO_rust_compressFilenameInternal(
FIO_RUST_COMPRESS_OK
}
/// Runs the source-file policy around the existing C-owned destination and
/// compression callback. Rust owns the ordering: named-file checks,
/// exclusion, source open, size-based async selection, source attachment,
/// compression, close, and finally conditional source removal.
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_compressFilenameSrcFile(
projection: *const FIO_rust_compress_src_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());
let src_file_name = projection.src_file_name;
let source_is_stdin = projection.source_is_stdin != 0;
if !source_is_stdin {
let stat_source = projection
.stat_source
.expect("source stat callback is required for named files");
match unsafe { stat_source(projection.opaque, src_file_name) } {
FIO_RUST_COMPRESS_SRC_STAT_FAILED | FIO_RUST_COMPRESS_SRC_STAT_OK => {}
FIO_RUST_COMPRESS_SRC_DIRECTORY | FIO_RUST_COMPRESS_SRC_DICT_COLLISION => return 1,
_ => unreachable!("invalid source stat status"),
}
}
if projection.exclude_compressed_files != 0 {
let source_is_excluded = projection
.source_is_excluded
.expect("compressed-file predicate is required when exclusion is enabled");
if unsafe { source_is_excluded(projection.opaque, src_file_name) } != 0 {
return 0;
}
}
let open_source = projection
.open_source
.expect("source open callback is required");
let mut file_size = FIO_RUST_COMPRESS_SRC_UNKNOWN_SIZE;
let open_status = unsafe { open_source(projection.opaque, src_file_name, &mut file_size) };
if open_status != FIO_RUST_COMPRESS_SRC_OPEN_OK {
return 1;
}
let async_mode = if file_size != FIO_RUST_COMPRESS_SRC_UNKNOWN_SIZE
&& file_size < FIO_RUST_COMPRESS_SRC_ASYNC_THRESHOLD
{
0
} else {
1
};
let set_async = projection
.set_async
.expect("async-selection callback is required");
unsafe { set_async(projection.opaque, async_mode) };
let attach_source = projection
.attach_source
.expect("source attachment callback is required");
unsafe { attach_source(projection.opaque) };
let compress = projection
.compress
.expect("compression callback is required");
let result = unsafe {
compress(
projection.opaque,
projection.dst_file_name,
src_file_name,
projection.compression_level,
)
};
let close_source = projection
.close_source
.expect("source close callback is required");
unsafe { close_source(projection.opaque) };
if projection.remove_src_file != 0 && result == 0 && !source_is_stdin {
let remove_source = projection
.remove_source
.expect("source removal callback is required when removal is enabled");
unsafe { remove_source(projection.opaque, src_file_name) };
}
result
}
/// 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
@@ -4277,6 +4473,222 @@ impl PoolInner {
mod tests {
use super::*;
const SOURCE_POLICY_STAT: u8 = 1;
const SOURCE_POLICY_EXCLUDED: u8 = 2;
const SOURCE_POLICY_OPEN: u8 = 3;
const SOURCE_POLICY_ASYNC: u8 = 4;
const SOURCE_POLICY_ATTACH: u8 = 5;
const SOURCE_POLICY_COMPRESS: u8 = 6;
const SOURCE_POLICY_CLOSE: u8 = 7;
const SOURCE_POLICY_REMOVE: u8 = 8;
#[derive(Default)]
struct SourcePolicyState {
events: Vec<u8>,
async_modes: Vec<c_int>,
stat_status: c_int,
excluded: c_int,
open_status: c_int,
file_size: u64,
compression_status: c_int,
compression_level: c_int,
}
unsafe extern "C" fn source_policy_stat(
opaque: *mut c_void,
_src_file_name: *const c_char,
) -> c_int {
let state = unsafe { &mut *opaque.cast::<SourcePolicyState>() };
state.events.push(SOURCE_POLICY_STAT);
state.stat_status
}
unsafe extern "C" fn source_policy_excluded(
opaque: *mut c_void,
_src_file_name: *const c_char,
) -> c_int {
let state = unsafe { &mut *opaque.cast::<SourcePolicyState>() };
state.events.push(SOURCE_POLICY_EXCLUDED);
state.excluded
}
unsafe extern "C" fn source_policy_open(
opaque: *mut c_void,
_src_file_name: *const c_char,
file_size: *mut u64,
) -> c_int {
let state = unsafe { &mut *opaque.cast::<SourcePolicyState>() };
state.events.push(SOURCE_POLICY_OPEN);
unsafe { *file_size = state.file_size };
state.open_status
}
unsafe extern "C" fn source_policy_async(opaque: *mut c_void, async_mode: c_int) {
let state = unsafe { &mut *opaque.cast::<SourcePolicyState>() };
state.events.push(SOURCE_POLICY_ASYNC);
state.async_modes.push(async_mode);
}
unsafe extern "C" fn source_policy_attach(opaque: *mut c_void) {
let state = unsafe { &mut *opaque.cast::<SourcePolicyState>() };
state.events.push(SOURCE_POLICY_ATTACH);
}
unsafe extern "C" fn source_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::<SourcePolicyState>() };
state.events.push(SOURCE_POLICY_COMPRESS);
state.compression_level = compression_level;
state.compression_status
}
unsafe extern "C" fn source_policy_close(opaque: *mut c_void) {
let state = unsafe { &mut *opaque.cast::<SourcePolicyState>() };
state.events.push(SOURCE_POLICY_CLOSE);
}
unsafe extern "C" fn source_policy_remove(opaque: *mut c_void, _src_file_name: *const c_char) {
let state = unsafe { &mut *opaque.cast::<SourcePolicyState>() };
state.events.push(SOURCE_POLICY_REMOVE);
}
fn source_policy_projection(
state: &mut SourcePolicyState,
source_is_stdin: c_int,
exclude_compressed_files: c_int,
remove_src_file: c_int,
) -> FIO_rust_compress_src_projection_t {
FIO_rust_compress_src_projection_t {
opaque: (state as *mut SourcePolicyState).cast(),
dst_file_name: c"destination".as_ptr(),
src_file_name: c"source".as_ptr(),
compression_level: 7,
source_is_stdin,
exclude_compressed_files,
remove_src_file,
stat_source: Some(source_policy_stat),
source_is_excluded: Some(source_policy_excluded),
open_source: Some(source_policy_open),
set_async: Some(source_policy_async),
attach_source: Some(source_policy_attach),
compress: Some(source_policy_compress),
close_source: Some(source_policy_close),
remove_source: Some(source_policy_remove),
}
}
#[test]
fn source_policy_keeps_named_checks_before_exclusion_and_open() {
for status in [
FIO_RUST_COMPRESS_SRC_DIRECTORY,
FIO_RUST_COMPRESS_SRC_DICT_COLLISION,
] {
let mut state = SourcePolicyState {
stat_status: status,
excluded: 1,
..SourcePolicyState::default()
};
let projection = source_policy_projection(&mut state, 0, 1, 1);
assert_eq!(unsafe { FIO_rust_compressFilenameSrcFile(&projection) }, 1);
assert_eq!(state.events, vec![SOURCE_POLICY_STAT]);
}
}
#[test]
fn source_policy_short_circuits_excluded_files_before_opening() {
let mut state = SourcePolicyState {
stat_status: FIO_RUST_COMPRESS_SRC_STAT_OK,
excluded: 1,
..SourcePolicyState::default()
};
let projection = source_policy_projection(&mut state, 0, 1, 1);
assert_eq!(unsafe { FIO_rust_compressFilenameSrcFile(&projection) }, 0);
assert_eq!(
state.events,
vec![SOURCE_POLICY_STAT, SOURCE_POLICY_EXCLUDED]
);
}
#[test]
fn source_policy_selects_async_after_open_and_removes_only_after_success() {
let mut state = SourcePolicyState {
stat_status: FIO_RUST_COMPRESS_SRC_STAT_OK,
file_size: FIO_RUST_COMPRESS_SRC_ASYNC_THRESHOLD - 1,
compression_status: 0,
..SourcePolicyState::default()
};
let projection = source_policy_projection(&mut state, 0, 0, 1);
assert_eq!(unsafe { FIO_rust_compressFilenameSrcFile(&projection) }, 0);
assert_eq!(state.async_modes, vec![0]);
assert_eq!(state.compression_level, 7);
assert_eq!(
state.events,
vec![
SOURCE_POLICY_STAT,
SOURCE_POLICY_OPEN,
SOURCE_POLICY_ASYNC,
SOURCE_POLICY_ATTACH,
SOURCE_POLICY_COMPRESS,
SOURCE_POLICY_CLOSE,
SOURCE_POLICY_REMOVE,
]
);
let mut failed = SourcePolicyState {
stat_status: FIO_RUST_COMPRESS_SRC_STAT_OK,
file_size: FIO_RUST_COMPRESS_SRC_ASYNC_THRESHOLD,
compression_status: 1,
..SourcePolicyState::default()
};
let failed_projection = source_policy_projection(&mut failed, 0, 0, 1);
assert_eq!(
unsafe { FIO_rust_compressFilenameSrcFile(&failed_projection) },
1
);
assert_eq!(failed.async_modes, vec![1]);
assert_eq!(
failed.events,
vec![
SOURCE_POLICY_STAT,
SOURCE_POLICY_OPEN,
SOURCE_POLICY_ASYNC,
SOURCE_POLICY_ATTACH,
SOURCE_POLICY_COMPRESS,
SOURCE_POLICY_CLOSE,
]
);
}
#[test]
fn source_policy_skips_named_checks_and_removal_for_stdin() {
let mut state = SourcePolicyState {
file_size: FIO_RUST_COMPRESS_SRC_UNKNOWN_SIZE,
..SourcePolicyState::default()
};
let projection = source_policy_projection(&mut state, 1, 0, 1);
assert_eq!(unsafe { FIO_rust_compressFilenameSrcFile(&projection) }, 0);
assert_eq!(state.async_modes, vec![1]);
assert_eq!(
state.events,
vec![
SOURCE_POLICY_OPEN,
SOURCE_POLICY_ASYNC,
SOURCE_POLICY_ATTACH,
SOURCE_POLICY_COMPRESS,
SOURCE_POLICY_CLOSE,
]
);
}
fn run_zstd_adapt(policy: c_int, projection: &FIO_rust_zstd_adapt_projection_t) -> c_int {
unsafe { FIO_rust_zstd_adapt(policy, projection) }
}