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

refactor(cli): move MT resource parameter policy to Rust

Browse Source 
File-resource creation already delegates the general compression-parameter
policy to Rust, but the multithreaded parameters were still applied in C in a
separate branch. That left parameter ordering, optional overlap handling, and
error short-circuiting outside the Rust policy boundary. Add a narrow ABI
projection with C callbacks for CCtx mutation and diagnostics, and let Rust
apply worker count, job size, optional overlap, and rsyncable in the original
order. The C90 declaration layout and compile-time ABI assertions keep the
existing native program configurations intact.

Test Plan:
- `cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings` -- passed
- `cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings` -- passed
- `cargo test --manifest-path rust/Cargo.toml --all-targets` -- 771 passed
- `cargo test --manifest-path rust/cli/Cargo.toml --all-targets` -- 179 passed
- `cargo +nightly fmt --manifest-path rust/Cargo.toml -- --check` -- passed
- `make -j1` under `ulimit -v 41943040` -- passed without the new C90 warning
- `make -j1 -C tests test` under `ulimit -v 41943040` -- passed
This commit is contained in:
ddidderr
2026-07-20 06:03:41 +02:00
parent 998c88c97b
commit a67e6a59df
2 changed files with 151 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
programs/fileio.c
+22 -16
View File
@@ -1733,6 +1733,7 @@ typedef size_t (*FIO_rust_createCResources_set_parameter_f)(void* context,
int parameter,
int value);
typedef int (*FIO_rust_createCResources_is_error_f)(size_t result);
typedef void (*FIO_rust_createCResources_display_overlap_f)(int overlapLog);
typedef struct {
void* callbackContext;
const FIO_prefs_t* prefs;
@@ -1740,6 +1741,7 @@ typedef struct {
int cLevel;
FIO_rust_createCResources_set_parameter_f setParameter;
FIO_rust_createCResources_is_error_f isError;
FIO_rust_createCResources_display_overlap_f displayOverlap;
} FIO_rust_createCResourcesState;
typedef char FIO_rust_create_c_resources_state_context_offset[
(offsetof(FIO_rust_createCResourcesState, callbackContext) == 0) ? 1 : -1];
@@ -1757,9 +1759,12 @@ typedef char FIO_rust_create_c_resources_state_set_parameter_offset[
typedef char FIO_rust_create_c_resources_state_is_error_offset[
(offsetof(FIO_rust_createCResourcesState, isError)
== offsetof(FIO_rust_createCResourcesState, setParameter) + sizeof(void*)) ? 1 : -1];
typedef char FIO_rust_create_c_resources_state_display_overlap_offset[
(offsetof(FIO_rust_createCResourcesState, displayOverlap)
== offsetof(FIO_rust_createCResourcesState, isError) + sizeof(void*)) ? 1 : -1];
typedef char FIO_rust_create_c_resources_state_size[
(sizeof(FIO_rust_createCResourcesState)
== offsetof(FIO_rust_createCResourcesState, setParameter) + 2 * sizeof(void*)) ? 1 : -1];
== offsetof(FIO_rust_createCResourcesState, setParameter) + 3 * sizeof(void*)) ? 1 : -1];
size_t FIO_rust_createCResources(const FIO_rust_createCResourcesState* state);
static size_t FIO_rust_createCResources_setParameter(void* context,
@@ -1774,6 +1779,13 @@ static int FIO_rust_createCResources_isError(size_t result)
return ZSTD_isError(result);
}
static void FIO_rust_createCResources_displayOverlap(int overlapLog)
{
DISPLAYLEVEL(3,"set overlapLog = %u \n", overlapLog);
}
size_t FIO_rust_setCResourcesMtParameters(const FIO_rust_createCResourcesState* state);
static cRess_t FIO_createCResources(FIO_prefs_t* const prefs,
const char* dictFileName, unsigned long long const maxSrcFileSize,
int cLevel, ZSTD_compressionParameters comprParams) {
@@ -1781,6 +1793,7 @@ static cRess_t FIO_createCResources(FIO_prefs_t* const prefs,
int forceNoUseMMap = prefs->mmapDict == ZSTD_ps_disable;
FIO_dictBufferType_t dictBufferType;
cRess_t ress;
FIO_rust_createCResourcesState policy;
memset(&ress, 0, sizeof(ress));
DISPLAYLEVEL(6, "FIO_createCResources \n");
@@ -1811,26 +1824,19 @@ static cRess_t FIO_createCResources(FIO_prefs_t* const prefs,
EXM_THROW(32, "allocation error : can't create dictBuffer");
ress.dictFileName = dictFileName;
FIO_rust_createCResourcesState const policy = {
ress.cctx,
prefs,
comprParams,
cLevel,
FIO_rust_createCResources_setParameter,
FIO_rust_createCResources_isError
};
policy.callbackContext = ress.cctx;
policy.prefs = prefs;
policy.comprParams = comprParams;
policy.cLevel = cLevel;
policy.setParameter = FIO_rust_createCResources_setParameter;
policy.isError = FIO_rust_createCResources_isError;
policy.displayOverlap = FIO_rust_createCResources_displayOverlap;
CHECK( FIO_rust_createCResources(&policy) );
/* multi-threading */
#ifdef ZSTD_MULTITHREAD
DISPLAYLEVEL(5,"set nb workers = %u \n", prefs->nbWorkers);
CHECK( ZSTD_CCtx_setParameter(ress.cctx, ZSTD_c_nbWorkers, prefs->nbWorkers) );
CHECK( ZSTD_CCtx_setParameter(ress.cctx, ZSTD_c_jobSize, prefs->blockSize) );
if (prefs->overlapLog != FIO_OVERLAP_LOG_NOTSET) {
DISPLAYLEVEL(3,"set overlapLog = %u \n", prefs->overlapLog);
CHECK( ZSTD_CCtx_setParameter(ress.cctx, ZSTD_c_overlapLog, prefs->overlapLog) );
}
CHECK( ZSTD_CCtx_setParameter(ress.cctx, ZSTD_c_rsyncable, prefs->rsyncable) );
CHECK( FIO_rust_setCResourcesMtParameters(&policy) );
#endif
/* dictionary */
if (prefs->patchFromMode) {
rust/src/fileio_prefs.rs
+129 -3
View File
@@ -57,6 +57,10 @@ const FIO_ZSTD_C_LITERAL_COMPRESSION_MODE: c_int = 1002;
const FIO_ZSTD_C_SRC_SIZE_HINT: c_int = 1004;
const FIO_ZSTD_C_ENABLE_DEDICATED_DICT_SEARCH: c_int = 1005;
const FIO_ZSTD_C_USE_ROW_MATCH_FINDER: c_int = 1011;
const FIO_ZSTD_C_NB_WORKERS: c_int = 400;
const FIO_ZSTD_C_JOB_SIZE: c_int = 401;
const FIO_ZSTD_C_OVERLAP_LOG: c_int = 402;
const FIO_ZSTD_C_RSYNCABLE: c_int = 500;
static STDOUT_MARK: &[u8] = b"/*stdout*\\\0";
static mut COMPRESSED_NAME_CAPACITY: usize = 0;
@@ -172,12 +176,14 @@ pub unsafe extern "C" fn FIO_rust_freeCResources(
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);
/// C-owned compression-resource parameter policy projection.
///
/// Rust owns the ordered, short-circuiting parameter policy. C keeps the
/// `ZSTD_CCtx` layout private and exposes only the parameter callback; resource
/// allocation, dictionary/pool setup, and dictionary loading remain in C.
/// `ZSTD_CCtx` layout private and exposes parameter/diagnostic callbacks;
/// resource allocation, dictionary/pool setup, and dictionary loading remain
/// in C.
#[repr(C)]
pub struct FIO_rust_createCResourcesState {
callback_context: *mut c_void,
@@ -186,6 +192,7 @@ pub struct FIO_rust_createCResourcesState {
compression_level: c_int,
set_parameter: FIO_createCResourcesSetParameterFn,
is_error: FIO_createCResourcesIsErrorFn,
display_overlap: FIO_createCResourcesDisplayOverlapFn,
}
const _: () = {
@@ -207,7 +214,11 @@ const _: () = {
assert!(offset_of!(FIO_rust_createCResourcesState, set_parameter) == callback_offset);
assert!(offset_of!(FIO_rust_createCResourcesState, is_error) == callback_offset + size_of::<usize>());
assert!(
size_of::<FIO_rust_createCResourcesState>() == callback_offset + 2 * size_of::<usize>()
offset_of!(FIO_rust_createCResourcesState, display_overlap)
== callback_offset + 2 * size_of::<usize>()
);
assert!(
size_of::<FIO_rust_createCResourcesState>() == callback_offset + 3 * size_of::<usize>()
);
};
@@ -300,6 +311,54 @@ pub unsafe extern "C" fn FIO_rust_createCResources(
0
}
/// Apply the multithreaded CCtx parameter policy after the general creation
/// parameters. C retains the compile-time multithreading guard and its
/// diagnostics; Rust owns the ordered callback sequence and short-circuiting
/// error policy.
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_setCResourcesMtParameters(
state: *const FIO_rust_createCResourcesState,
) -> usize {
if state.is_null() {
return usize::MAX;
}
let state = unsafe { &*state };
if state.prefs.is_null() {
return usize::MAX;
}
let prefs = unsafe { &*state.prefs };
let parameters = [
(FIO_ZSTD_C_NB_WORKERS, prefs.nbWorkers),
(FIO_ZSTD_C_JOB_SIZE, prefs.blockSize),
];
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 prefs.overlapLog != FIO_OVERLAP_LOG_NOTSET {
unsafe { (state.display_overlap)(prefs.overlapLog) };
let result = unsafe {
(state.set_parameter)(
state.callback_context,
FIO_ZSTD_C_OVERLAP_LOG,
prefs.overlapLog,
)
};
if unsafe { (state.is_error)(result) } != 0 {
return result;
}
}
let result = unsafe {
(state.set_parameter)(state.callback_context, FIO_ZSTD_C_RSYNCABLE, prefs.rsyncable)
};
if unsafe { (state.is_error)(result) } != 0 {
return result;
}
0
}
pub type FIO_listMultipleFilesIsStdinFn = unsafe extern "C" fn(*mut c_void, *const c_char) -> c_int;
pub type FIO_listMultipleFilesDisplayFn = unsafe extern "C" fn(*mut c_void);
pub type FIO_listMultipleFilesListFileFn = unsafe extern "C" fn(
@@ -2134,6 +2193,8 @@ mod tests {
c_int::from(result == usize::MAX)
}
unsafe extern "C" fn create_c_resources_display_overlap_test(_overlap_log: c_int) {}
fn create_c_resources_test_prefs() -> FIO_prefs_t {
let mut prefs: FIO_prefs_t = unsafe { std::mem::zeroed() };
prefs.contentSize = 1;
@@ -2166,6 +2227,7 @@ mod tests {
compression_level: 7,
set_parameter: create_c_resources_set_parameter_test,
is_error: create_c_resources_is_error_test,
display_overlap: create_c_resources_display_overlap_test,
}
}
@@ -2228,6 +2290,70 @@ mod tests {
);
}
#[test]
fn create_c_resources_mt_parameters_preserve_order_and_optional_overlap() {
let mut prefs = create_c_resources_test_prefs();
prefs.nbWorkers = 3;
prefs.blockSize = 4096;
prefs.overlapLog = 5;
prefs.rsyncable = 1;
let mut context = CreateCResourcesTestState::default();
let state = create_c_resources_test_state(&prefs, &mut context);
let result = unsafe { FIO_rust_setCResourcesMtParameters(&state) };
assert_eq!(result, 0);
assert_eq!(
context.events,
vec![
(FIO_ZSTD_C_NB_WORKERS, 3),
(FIO_ZSTD_C_JOB_SIZE, 4096),
(FIO_ZSTD_C_OVERLAP_LOG, 5),
(FIO_ZSTD_C_RSYNCABLE, 1),
]
);
context.events.clear();
prefs.overlapLog = FIO_OVERLAP_LOG_NOTSET;
let state = create_c_resources_test_state(&prefs, &mut context);
let result = unsafe { FIO_rust_setCResourcesMtParameters(&state) };
assert_eq!(result, 0);
assert_eq!(
context.events,
vec![
(FIO_ZSTD_C_NB_WORKERS, 3),
(FIO_ZSTD_C_JOB_SIZE, 4096),
(FIO_ZSTD_C_RSYNCABLE, 1),
]
);
}
#[test]
fn create_c_resources_mt_parameters_short_circuit_on_error() {
let mut prefs = create_c_resources_test_prefs();
prefs.nbWorkers = 3;
prefs.blockSize = 4096;
prefs.overlapLog = 5;
prefs.rsyncable = 1;
let mut context = CreateCResourcesTestState {
fail_parameter: Some(FIO_ZSTD_C_JOB_SIZE),
..CreateCResourcesTestState::default()
};
let state = create_c_resources_test_state(&prefs, &mut context);
let result = unsafe { FIO_rust_setCResourcesMtParameters(&state) };
assert_eq!(result, usize::MAX);
assert_eq!(
context.events,
vec![
(FIO_ZSTD_C_NB_WORKERS, 3),
(FIO_ZSTD_C_JOB_SIZE, 4096),
]
);
}
fn temporary_file_path(name: &str) -> std::path::PathBuf {
std::env::temp_dir().join(format!("zstd-fileio-prefs-{}-{name}", std::process::id()))
}