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

feat(cli): move compression resource orchestration to Rust

Browse Source 
Make the Rust policy layer own the ordered compression-resource lifecycle:
create the CCtx, prepare patch/dictionary state, create the write and read
pools, validate the dictionary, apply general and multithreaded parameters,
and finally load the dictionary.  Keep cRess_t, FIO_Dict_t, file statistics,
AIO pools, and CLI diagnostics in C callbacks so the existing resource
ownership and error behavior remain local to the C backend.

Preserve adaptive window defaults and patch-from parameter adjustment while
adding ABI layout assertions and a lifecycle-order test for the new boundary.

Test Plan:
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo +nightly fmt --manifest-path rust/cli/Cargo.toml --all
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/cli/Cargo.toml --all-targets (205 passed)
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 make -j1
This commit is contained in:
ddidderr
2026-07-21 14:06:00 +02:00
parent c561a4dc8a
commit 877eb89e6c
2 changed files with 364 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/src/fileio_prefs.rs
+244 -136
View File
@@ -288,20 +288,14 @@ pub struct FIO_rust_freeCResourcesState {
const _: () = {
assert!(offset_of!(FIO_rust_freeCResourcesState, callback_context) == 0);
assert!(offset_of!(FIO_rust_freeCResourcesState, free_dict) == size_of::<usize>());
assert!(
offset_of!(FIO_rust_freeCResourcesState, free_write_pool) == 2 * size_of::<usize>()
);
assert!(
offset_of!(FIO_rust_freeCResourcesState, free_read_pool) == 3 * size_of::<usize>()
);
assert!(offset_of!(FIO_rust_freeCResourcesState, free_write_pool) == 2 * size_of::<usize>());
assert!(offset_of!(FIO_rust_freeCResourcesState, free_read_pool) == 3 * size_of::<usize>());
assert!(offset_of!(FIO_rust_freeCResourcesState, free_cctx) == 4 * size_of::<usize>());
assert!(size_of::<FIO_rust_freeCResourcesState>() == 5 * size_of::<usize>());
};
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_freeCResources(
state: *const FIO_rust_freeCResourcesState,
) {
pub unsafe extern "C" fn FIO_rust_freeCResources(state: *const FIO_rust_freeCResourcesState) {
if state.is_null() {
return;
}
@@ -439,15 +433,14 @@ fn ceil_log2_window_size(window_size: u64) -> c_uint {
if window_size == 0 {
return 0;
}
highbit64(window_size)
+ c_uint::from((window_size & window_size.wrapping_sub(1)) != 0)
highbit64(window_size) + c_uint::from((window_size & window_size.wrapping_sub(1)) != 0)
}
#[cfg(feature = "decompression")]
const _: () = {
let callback_offset = (offset_of!(FIO_rust_zstdErrorHelpState, error_code)
+ size_of::<c_int>())
.div_ceil(size_of::<usize>())
.div_ceil(size_of::<usize>())
* size_of::<usize>();
assert!(offset_of!(FIO_rust_zstdErrorHelpState, callback_context) == 0);
assert!(offset_of!(FIO_rust_zstdErrorHelpState, src_file_name) == size_of::<usize>());
@@ -537,68 +530,105 @@ pub type FIO_createCResourcesSetParameterFn =
unsafe extern "C" fn(*mut c_void, c_int, c_int) -> usize;
pub type FIO_createCResourcesIsErrorFn = unsafe extern "C" fn(usize) -> c_int;
pub type FIO_createCResourcesDisplayOverlapFn = unsafe extern "C" fn(c_int);
pub type FIO_createCResourcesCreateCctxFn = unsafe extern "C" fn(*mut c_void);
pub type FIO_createCResourcesPrepareDictionaryFn =
unsafe extern "C" fn(*mut c_void, *const FIO_prefs_t, *const c_char, u64, c_int, *mut c_void);
pub type FIO_createCResourcesCreatePoolFn = unsafe extern "C" fn(*mut c_void, *const FIO_prefs_t);
pub type FIO_createCResourcesValidateDictionaryFn = unsafe extern "C" fn(*mut c_void);
pub type FIO_createCResourcesLoadDictionaryFn = unsafe extern "C" fn(*mut c_void, c_int) -> usize;
/// C-owned compression-resource parameter policy projection.
/// Compression-resource initialization projection.
///
/// Rust owns the ordered, short-circuiting parameter policy. C keeps the
/// `ZSTD_CCtx` layout private and exposes parameter/diagnostic callbacks;
/// resource allocation, dictionary/pool setup, and dictionary loading remain
/// in C.
/// Rust owns the resource lifecycle and parameter order. C keeps the
/// `cRess_t`, `ZSTD_CCtx`, dictionary, and asynchronous-pool layouts private
/// behind callbacks which retain the original diagnostics and side effects.
#[repr(C)]
#[derive(Clone, Copy)]
pub struct FIO_rust_createCResourcesState {
callback_context: *mut c_void,
prefs: *const FIO_prefs_t,
compression_params: FIO_compressionParameters,
compression_level: c_int,
dict_file_name: *const c_char,
max_src_file_size: u64,
multithreaded: c_int,
create_cctx: FIO_createCResourcesCreateCctxFn,
prepare_dictionary: FIO_createCResourcesPrepareDictionaryFn,
create_write_pool: FIO_createCResourcesCreatePoolFn,
create_read_pool: FIO_createCResourcesCreatePoolFn,
validate_dictionary: FIO_createCResourcesValidateDictionaryFn,
load_dictionary: FIO_createCResourcesLoadDictionaryFn,
set_parameter: FIO_createCResourcesSetParameterFn,
is_error: FIO_createCResourcesIsErrorFn,
display_overlap: FIO_createCResourcesDisplayOverlapFn,
}
const _: () = {
let callback_offset = (2 * size_of::<usize>()
+ size_of::<FIO_compressionParameters>()
+ size_of::<c_int>())
.div_ceil(size_of::<usize>())
* size_of::<usize>();
let word = size_of::<usize>();
let dict_file_name_offset =
(2 * word + size_of::<FIO_compressionParameters>() + size_of::<c_int>()).div_ceil(word)
* word;
let callback_offset = (dict_file_name_offset + word + size_of::<u64>() + size_of::<c_int>())
.div_ceil(word)
* word;
assert!(offset_of!(FIO_rust_createCResourcesState, callback_context) == 0);
assert!(offset_of!(FIO_rust_createCResourcesState, prefs) == size_of::<usize>());
assert!(
offset_of!(FIO_rust_createCResourcesState, compression_params)
== 2 * size_of::<usize>()
);
assert!(offset_of!(FIO_rust_createCResourcesState, prefs) == word);
assert!(offset_of!(FIO_rust_createCResourcesState, compression_params) == 2 * word);
assert!(
offset_of!(FIO_rust_createCResourcesState, compression_level)
== 2 * size_of::<usize>() + size_of::<FIO_compressionParameters>()
== 2 * word + size_of::<FIO_compressionParameters>()
);
assert!(offset_of!(FIO_rust_createCResourcesState, set_parameter) == callback_offset);
assert!(offset_of!(FIO_rust_createCResourcesState, is_error) == callback_offset + size_of::<usize>());
assert!(offset_of!(FIO_rust_createCResourcesState, dict_file_name) == dict_file_name_offset);
assert!(
offset_of!(FIO_rust_createCResourcesState, display_overlap)
== callback_offset + 2 * size_of::<usize>()
offset_of!(FIO_rust_createCResourcesState, max_src_file_size)
== dict_file_name_offset + word
);
assert!(
size_of::<FIO_rust_createCResourcesState>() == callback_offset + 3 * size_of::<usize>()
offset_of!(FIO_rust_createCResourcesState, multithreaded)
== dict_file_name_offset + word + size_of::<u64>()
);
assert!(offset_of!(FIO_rust_createCResourcesState, create_cctx) == callback_offset);
assert!(
offset_of!(FIO_rust_createCResourcesState, prepare_dictionary) == callback_offset + word
);
assert!(
offset_of!(FIO_rust_createCResourcesState, create_write_pool) == callback_offset + 2 * word
);
assert!(
offset_of!(FIO_rust_createCResourcesState, create_read_pool) == callback_offset + 3 * word
);
assert!(
offset_of!(FIO_rust_createCResourcesState, validate_dictionary)
== callback_offset + 4 * word
);
assert!(
offset_of!(FIO_rust_createCResourcesState, load_dictionary) == callback_offset + 5 * word
);
assert!(
offset_of!(FIO_rust_createCResourcesState, set_parameter) == callback_offset + 6 * word
);
assert!(offset_of!(FIO_rust_createCResourcesState, is_error) == callback_offset + 7 * word);
assert!(
offset_of!(FIO_rust_createCResourcesState, display_overlap) == callback_offset + 8 * word
);
assert!(size_of::<FIO_rust_createCResourcesState>() == callback_offset + 9 * word);
};
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_createCResources(
state: *const FIO_rust_createCResourcesState,
) -> usize {
if state.is_null() {
return usize::MAX;
}
let state = unsafe { &*state };
#[inline]
unsafe fn set_c_resources_parameters(state: &FIO_rust_createCResourcesState) -> usize {
if state.prefs.is_null() {
return usize::MAX;
}
let prefs = unsafe { &*state.prefs };
let mut compression_params = state.compression_params;
if prefs.adaptiveMode != 0 && prefs.ldmFlag == 0 && compression_params.windowLog == 0 {
compression_params.windowLog = FIO_ADAPT_WINDOWLOG_DEFAULT;
}
let compression_params = state.compression_params;
let set_parameter = |parameter: c_int, value: c_int| {
let result = unsafe { (state.set_parameter)(state.callback_context, parameter, value) };
if unsafe { (state.is_error)(result) } != 0 {
Some(result)
} else {
None
}
};
let parameters = [
(FIO_ZSTD_C_CONTENT_SIZE_FLAG, prefs.contentSize),
(FIO_ZSTD_C_DICT_ID_FLAG, prefs.dictIDFlag),
@@ -614,33 +644,18 @@ pub unsafe extern "C" fn FIO_rust_createCResources(
(FIO_ZSTD_C_LDM_MIN_MATCH, prefs.ldmMinMatch),
];
for (parameter, value) in parameters {
let result = unsafe { (state.set_parameter)(state.callback_context, parameter, value) };
if unsafe { (state.is_error)(result) } != 0 {
return result;
if let Some(error) = set_parameter(parameter, value) {
return error;
}
}
if prefs.ldmBucketSizeLog != FIO_LDM_PARAM_NOTSET {
let result = unsafe {
(state.set_parameter)(
state.callback_context,
FIO_ZSTD_C_LDM_BUCKET_SIZE_LOG,
prefs.ldmBucketSizeLog,
)
};
if unsafe { (state.is_error)(result) } != 0 {
return result;
if let Some(error) = set_parameter(FIO_ZSTD_C_LDM_BUCKET_SIZE_LOG, prefs.ldmBucketSizeLog) {
return error;
}
}
if prefs.ldmHashRateLog != FIO_LDM_PARAM_NOTSET {
let result = unsafe {
(state.set_parameter)(
state.callback_context,
FIO_ZSTD_C_LDM_HASH_RATE_LOG,
prefs.ldmHashRateLog,
)
};
if unsafe { (state.is_error)(result) } != 0 {
return result;
if let Some(error) = set_parameter(FIO_ZSTD_C_LDM_HASH_RATE_LOG, prefs.ldmHashRateLog) {
return error;
}
}
@@ -663,11 +678,62 @@ pub unsafe extern "C" fn FIO_rust_createCResources(
(FIO_ZSTD_C_ENABLE_DEDICATED_DICT_SEARCH, 1),
];
for (parameter, value) in parameters {
let result = unsafe { (state.set_parameter)(state.callback_context, parameter, value) };
if let Some(error) = set_parameter(parameter, value) {
return error;
}
}
0
}
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_createCResources(
state: *const FIO_rust_createCResourcesState,
) -> usize {
let Some(state) = (unsafe { state.as_ref() }) else {
return usize::MAX;
};
if state.prefs.is_null() {
return usize::MAX;
}
let prefs = unsafe { &*state.prefs };
unsafe { (state.create_cctx)(state.callback_context) };
let mut policy = *state;
let mut compression_params = state.compression_params;
unsafe {
(state.prepare_dictionary)(
state.callback_context,
state.prefs,
state.dict_file_name,
state.max_src_file_size,
state.compression_level,
(&mut compression_params as *mut FIO_compressionParameters).cast(),
);
(state.create_write_pool)(state.callback_context, state.prefs);
(state.create_read_pool)(state.callback_context, state.prefs);
(state.validate_dictionary)(state.callback_context);
}
if prefs.adaptiveMode != 0 && prefs.ldmFlag == 0 && compression_params.windowLog == 0 {
compression_params.windowLog = FIO_ADAPT_WINDOWLOG_DEFAULT;
}
policy.compression_params = compression_params;
let result = unsafe { set_c_resources_parameters(&policy) };
if unsafe { (state.is_error)(result) } != 0 {
return result;
}
if state.multithreaded != 0 {
let result = unsafe { FIO_rust_setCResourcesMtParameters(&policy) };
if unsafe { (state.is_error)(result) } != 0 {
return result;
}
}
let patch_from_mode = prefs.patchFromMode;
let result = unsafe { (state.load_dictionary)(state.callback_context, patch_from_mode) };
if unsafe { (state.is_error)(result) } != 0 {
return result;
}
0
}
@@ -711,7 +777,11 @@ pub unsafe extern "C" fn FIO_rust_setCResourcesMtParameters(
}
}
let result = unsafe {
(state.set_parameter)(state.callback_context, FIO_ZSTD_C_RSYNCABLE, prefs.rsyncable)
(state.set_parameter)(
state.callback_context,
FIO_ZSTD_C_RSYNCABLE,
prefs.rsyncable,
)
};
if unsafe { (state.is_error)(result) } != 0 {
return result;
@@ -730,19 +800,12 @@ pub type FIO_listMultipleFilesListFileFn = unsafe extern "C" fn(
pub type FIO_listMultipleFilesDisplayTotalFn =
unsafe extern "C" fn(*mut c_void, *const FIO_listFileInfoProjection_t);
pub type FIO_listFileGetInfoFn = unsafe extern "C" fn(
*mut c_void,
*const c_char,
*mut FIO_fileInfo_t,
) -> c_int;
pub type FIO_listFileGetInfoFn =
unsafe extern "C" fn(*mut c_void, *const c_char, *mut FIO_fileInfo_t) -> c_int;
pub type FIO_listFileDisplayStatusFn =
unsafe extern "C" fn(*mut c_void, c_int, *const c_char, c_int);
pub type FIO_listFileDisplayInfoFn = unsafe extern "C" fn(
*mut c_void,
*const c_char,
*const FIO_fileInfo_t,
c_int,
);
pub type FIO_listFileDisplayInfoFn =
unsafe extern "C" fn(*mut c_void, *const c_char, *const FIO_fileInfo_t, c_int);
/// Callbacks for the C-owned `--list` file operation.
///
@@ -902,11 +965,7 @@ pub extern "C" fn FIO_rust_compressDestinationTransferStat(
destination_is_stdout: c_int,
source_is_regular: c_int,
) -> c_int {
compress_destination_transfer_stat(
source_is_stdin,
destination_is_stdout,
source_is_regular,
)
compress_destination_transfer_stat(source_is_stdin, destination_is_stdout, source_is_regular)
}
/// Callback projection for the C-owned `--list` file operations.
@@ -2895,18 +2954,12 @@ mod tests {
#[test]
fn list_file_status_action_preserves_cli_mapping_and_abi() {
for (status, action) in [
(
FIO_LIST_INFO_SUCCESS,
FIO_RUST_LIST_FILE_ACTION_SUCCESS,
),
(FIO_LIST_INFO_SUCCESS, FIO_RUST_LIST_FILE_ACTION_SUCCESS),
(
FIO_LIST_INFO_FRAME_ERROR,
FIO_RUST_LIST_FILE_ACTION_FRAME_ERROR,
),
(
FIO_LIST_INFO_NOT_ZSTD,
FIO_RUST_LIST_FILE_ACTION_NOT_ZSTD,
),
(FIO_LIST_INFO_NOT_ZSTD, FIO_RUST_LIST_FILE_ACTION_NOT_ZSTD),
(
FIO_LIST_INFO_FILE_ERROR,
FIO_RUST_LIST_FILE_ACTION_FILE_ERROR,
@@ -2935,10 +2988,7 @@ mod tests {
#[test]
fn remove_file_action_preserves_status_mapping_and_unknown_failure_policy() {
for (status, action) in [
(
FIO_REMOVE_SUCCESS,
FIO_RUST_REMOVE_FILE_ACTION_SUCCESS,
),
(FIO_REMOVE_SUCCESS, FIO_RUST_REMOVE_FILE_ACTION_SUCCESS),
(
FIO_REMOVE_STAT_FAILED,
FIO_RUST_REMOVE_FILE_ACTION_STAT_FAILED,
@@ -2947,10 +2997,7 @@ mod tests {
FIO_REMOVE_NON_REGULAR,
FIO_RUST_REMOVE_FILE_ACTION_NON_REGULAR,
),
(
FIO_REMOVE_FAILED,
FIO_RUST_REMOVE_FILE_ACTION_FAILED,
),
(FIO_REMOVE_FAILED, FIO_RUST_REMOVE_FILE_ACTION_FAILED),
(-1, FIO_RUST_REMOVE_FILE_ACTION_INVALID),
(c_int::MAX, FIO_RUST_REMOVE_FILE_ACTION_INVALID),
] {
@@ -3289,9 +3336,75 @@ mod tests {
#[derive(Default)]
struct CreateCResourcesTestState {
events: Vec<(c_int, c_int)>,
lifecycle: Vec<&'static str>,
fail_parameter: Option<c_int>,
}
unsafe extern "C" fn create_c_resources_create_cctx_test(context: *mut c_void) {
unsafe {
(*context.cast::<CreateCResourcesTestState>())
.lifecycle
.push("create-cctx");
}
}
unsafe extern "C" fn create_c_resources_prepare_dictionary_test(
context: *mut c_void,
_prefs: *const FIO_prefs_t,
_dict_file_name: *const c_char,
_max_src_file_size: u64,
_compression_level: c_int,
_compression_params: *mut c_void,
) {
unsafe {
(*context.cast::<CreateCResourcesTestState>())
.lifecycle
.push("prepare-dictionary");
}
}
unsafe extern "C" fn create_c_resources_create_write_pool_test(
context: *mut c_void,
_prefs: *const FIO_prefs_t,
) {
unsafe {
(*context.cast::<CreateCResourcesTestState>())
.lifecycle
.push("create-write-pool");
}
}
unsafe extern "C" fn create_c_resources_create_read_pool_test(
context: *mut c_void,
_prefs: *const FIO_prefs_t,
) {
unsafe {
(*context.cast::<CreateCResourcesTestState>())
.lifecycle
.push("create-read-pool");
}
}
unsafe extern "C" fn create_c_resources_validate_dictionary_test(context: *mut c_void) {
unsafe {
(*context.cast::<CreateCResourcesTestState>())
.lifecycle
.push("validate-dictionary");
}
}
unsafe extern "C" fn create_c_resources_load_dictionary_test(
context: *mut c_void,
_patch_from_mode: c_int,
) -> usize {
unsafe {
(*context.cast::<CreateCResourcesTestState>())
.lifecycle
.push("load-dictionary");
}
0
}
unsafe extern "C" fn create_c_resources_set_parameter_test(
context: *mut c_void,
parameter: c_int,
@@ -3342,6 +3455,15 @@ mod tests {
strategy: 6,
},
compression_level: 7,
dict_file_name: ptr::null(),
max_src_file_size: 0,
multithreaded: 0,
create_cctx: create_c_resources_create_cctx_test,
prepare_dictionary: create_c_resources_prepare_dictionary_test,
create_write_pool: create_c_resources_create_write_pool_test,
create_read_pool: create_c_resources_create_read_pool_test,
validate_dictionary: create_c_resources_validate_dictionary_test,
load_dictionary: create_c_resources_load_dictionary_test,
set_parameter: create_c_resources_set_parameter_test,
is_error: create_c_resources_is_error_test,
display_overlap: create_c_resources_display_overlap_test,
@@ -3357,6 +3479,17 @@ mod tests {
let result = unsafe { FIO_rust_createCResources(&state) };
assert_eq!(result, 0);
assert_eq!(
context.lifecycle,
vec![
"create-cctx",
"prepare-dictionary",
"create-write-pool",
"create-read-pool",
"validate-dictionary",
"load-dictionary",
]
);
assert_eq!(
context.events,
vec![
@@ -3464,10 +3597,7 @@ mod tests {
assert_eq!(result, usize::MAX);
assert_eq!(
context.events,
vec![
(FIO_ZSTD_C_NB_WORKERS, 3),
(FIO_ZSTD_C_JOB_SIZE, 4096),
]
vec![(FIO_ZSTD_C_NB_WORKERS, 3), (FIO_ZSTD_C_JOB_SIZE, 4096),]
);
}
@@ -4564,12 +4694,7 @@ mod tests {
let mut output = sentinel;
let result = unsafe {
FIO_rust_listFile(
file_name.as_ptr(),
display_level,
&mut output,
&callbacks,
)
FIO_rust_listFile(file_name.as_ptr(), display_level, &mut output, &callbacks)
};
assert_eq!(result, expected_return, "status {status}");
@@ -4803,20 +4928,15 @@ mod tests {
let output_dir = CString::new("out").unwrap();
unsafe {
let name =
FIO_determineCompressedName(source.as_ptr(), ptr::null(), suffix.as_ptr());
let name = FIO_determineCompressedName(source.as_ptr(), ptr::null(), suffix.as_ptr());
assert_eq!(CStr::from_ptr(name).to_bytes(), b"input/nested/file.zst");
let name = FIO_determineCompressedName(
source.as_ptr(),
output_dir.as_ptr(),
suffix.as_ptr(),
);
let name =
FIO_determineCompressedName(source.as_ptr(), output_dir.as_ptr(), suffix.as_ptr());
assert_eq!(CStr::from_ptr(name).to_bytes(), b"out/file.zst");
let stdin = CString::new("/*stdin*\\").unwrap();
let name =
FIO_determineCompressedName(stdin.as_ptr(), ptr::null(), suffix.as_ptr());
let name = FIO_determineCompressedName(stdin.as_ptr(), ptr::null(), suffix.as_ptr());
assert_eq!(CStr::from_ptr(name).to_bytes(), b"/*stdout*\\");
}
}
@@ -5105,33 +5225,21 @@ mod tests {
for display_level in [0, 2, 3] {
FIO_setNotificationLevel(display_level);
assert_eq!(
unsafe { FIO_shouldDisplayMultipleFileSummary(&ctx) },
0
);
assert_eq!(unsafe { FIO_shouldDisplayMultipleFileSummary(&ctx) }, 0);
}
ctx.nbFilesProcessed = 1;
for display_level in [0, 2, 3] {
FIO_setNotificationLevel(display_level);
assert_eq!(
unsafe { FIO_shouldDisplayMultipleFileSummary(&ctx) },
1
);
assert_eq!(unsafe { FIO_shouldDisplayMultipleFileSummary(&ctx) }, 1);
}
ctx.nbFilesProcessed = 2;
FIO_setNotificationLevel(0);
assert_eq!(
unsafe { FIO_shouldDisplayMultipleFileSummary(&ctx) },
1
);
assert_eq!(unsafe { FIO_shouldDisplayMultipleFileSummary(&ctx) }, 1);
ctx.nbFilesTotal = 1;
assert_eq!(
unsafe { FIO_shouldDisplayMultipleFileSummary(&ctx) },
0
);
assert_eq!(unsafe { FIO_shouldDisplayMultipleFileSummary(&ctx) }, 0);
FIO_setNotificationLevel(2);
}