feat(cli): move adaptive refresh timing to Rust

The adaptive compression loop still delegated its refresh clock gate to a
small C callback even though Rust already owned the iteration state and all
adaptive policy decisions. That left timing policy, the last-refresh scalar,
and one projection callback in the C-side orchestration boundary.

Move the one-sixth-second monotonic refresh gate into ZstdAdaptiveState. The
Rust loop now initializes the first refresh timestamp at frame start and
preserves the C callback's strict-greater-than interval check. C retains only
the private progression, parameter-setting, and diagnostic callbacks needed
by the existing file I/O context.

Shrink both sides of the projection together and keep compile-time offset and
size assertions aligned. The adaptive unit fixture now verifies Rust records
the refresh event while continuing to exercise the existing progression and
policy callbacks.

Test Plan:
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo check --manifest-path rust/Cargo.toml --tests
- 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 make -j1
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 make -j1 -C tests invalidDictionaries
- ulimit -v 41943040; make -j1 -C tests test
- git diff --check
This commit is contained in:
2026-07-21 16:39:33 +02:00
parent bf2d73fcc0
commit 02d3579da3
2 changed files with 30 additions and 43 deletions
+6 -19
View File
@@ -1643,7 +1643,6 @@ typedef char FIO_rust_zstd_progression_layout[
== 4 * sizeof(U64) + sizeof(unsigned) == 4 * sizeof(U64) + sizeof(unsigned)
&& sizeof(FIO_rust_zstd_progression_t) && sizeof(FIO_rust_zstd_progression_t)
== 4 * sizeof(U64) + 2 * sizeof(unsigned)) ? 1 : -1]; == 4 * sizeof(U64) + 2 * sizeof(unsigned)) ? 1 : -1];
typedef int (*FIO_rust_zstd_adaptive_refresh_fn)(void* opaque);
typedef void (*FIO_rust_zstd_adaptive_progression_fn)( typedef void (*FIO_rust_zstd_adaptive_progression_fn)(
void* opaque, FIO_rust_zstd_progression_t* progression); void* opaque, FIO_rust_zstd_progression_t* progression);
typedef void (*FIO_rust_zstd_adaptive_set_parameter_fn)( typedef void (*FIO_rust_zstd_adaptive_set_parameter_fn)(
@@ -1672,7 +1671,6 @@ typedef struct {
int minAdaptLevel; int minAdaptLevel;
int maxAdaptLevel; int maxAdaptLevel;
int maxCLevel; int maxCLevel;
FIO_rust_zstd_adaptive_refresh_fn adaptiveRefresh;
FIO_rust_zstd_adaptive_progression_fn adaptiveProgression; FIO_rust_zstd_adaptive_progression_fn adaptiveProgression;
FIO_rust_zstd_adaptive_set_parameter_fn adaptiveSetParameter; FIO_rust_zstd_adaptive_set_parameter_fn adaptiveSetParameter;
FIO_rust_zstd_adaptive_diagnostic_fn adaptiveDiagnostic; FIO_rust_zstd_adaptive_diagnostic_fn adaptiveDiagnostic;
@@ -1684,15 +1682,18 @@ typedef char FIO_rust_zstd_compress_projection_layout[
== 14 * sizeof(void*) + sizeof(int) == 14 * sizeof(void*) + sizeof(int)
&& offsetof(FIO_rust_zstd_compress_projection_t, maxCLevel) && offsetof(FIO_rust_zstd_compress_projection_t, maxCLevel)
== 14 * sizeof(void*) + 4 * sizeof(int) == 14 * sizeof(void*) + 4 * sizeof(int)
&& offsetof(FIO_rust_zstd_compress_projection_t, adaptiveRefresh) && offsetof(FIO_rust_zstd_compress_projection_t, adaptiveProgression)
== ((14 * sizeof(void*) + 5 * sizeof(int) + sizeof(void*) - 1) == ((14 * sizeof(void*) + 5 * sizeof(int) + sizeof(void*) - 1)
/ sizeof(void*)) * sizeof(void*) / sizeof(void*)) * sizeof(void*)
&& offsetof(FIO_rust_zstd_compress_projection_t, adaptiveSetParameter)
== ((14 * sizeof(void*) + 5 * sizeof(int) + sizeof(void*) - 1)
/ sizeof(void*)) * sizeof(void*) + sizeof(void*)
&& offsetof(FIO_rust_zstd_compress_projection_t, adaptiveDiagnostic) && offsetof(FIO_rust_zstd_compress_projection_t, adaptiveDiagnostic)
== ((14 * sizeof(void*) + 5 * sizeof(int) + sizeof(void*) - 1) == ((14 * sizeof(void*) + 5 * sizeof(int) + sizeof(void*) - 1)
/ sizeof(void*)) * sizeof(void*) + 3 * sizeof(void*) / sizeof(void*)) * sizeof(void*) + 2 * sizeof(void*)
&& sizeof(FIO_rust_zstd_compress_projection_t) && sizeof(FIO_rust_zstd_compress_projection_t)
== ((14 * sizeof(void*) + 5 * sizeof(int) + sizeof(void*) - 1) == ((14 * sizeof(void*) + 5 * sizeof(int) + sizeof(void*) - 1)
/ sizeof(void*)) * sizeof(void*) + 4 * sizeof(void*)) / sizeof(void*)) * sizeof(void*) + 3 * sizeof(void*))
? 1 : -1]; ? 1 : -1];
int FIO_rust_compressZstdFrame( int FIO_rust_compressZstdFrame(
@@ -2882,7 +2883,6 @@ enum {
typedef struct { typedef struct {
FIO_ctx_t* fCtx; FIO_ctx_t* fCtx;
ZSTD_CCtx* cctx; ZSTD_CCtx* cctx;
UTIL_time_t lastAdaptTime;
U64 srcFileSize; U64 srcFileSize;
UTIL_HumanReadableSize_t fileHrs; UTIL_HumanReadableSize_t fileHrs;
} FIO_rust_zstd_projection_context_t; } FIO_rust_zstd_projection_context_t;
@@ -2936,17 +2936,6 @@ static void FIO_rust_zstd_sparseWriteEnd(void* opaque)
AIO_WritePool_sparseWriteEnd((WritePoolCtx_t*)opaque); AIO_WritePool_sparseWriteEnd((WritePoolCtx_t*)opaque);
} }
static int FIO_rust_zstd_adaptiveRefresh(void* opaque)
{
FIO_rust_zstd_projection_context_t* const context =
(FIO_rust_zstd_projection_context_t*)opaque;
if (UTIL_clockSpanMicro(context->lastAdaptTime) > REFRESH_RATE) {
context->lastAdaptTime = UTIL_getTime();
return 1;
}
return 0;
}
static void FIO_rust_zstd_adaptiveProgression( static void FIO_rust_zstd_adaptiveProgression(
void* opaque, FIO_rust_zstd_progression_t* projection) void* opaque, FIO_rust_zstd_progression_t* projection)
{ {
@@ -3110,7 +3099,6 @@ FIO_rust_compressZstdCallback(void* fCtx, void* prefs, void* ress,
memset(&context, 0, sizeof(context)); memset(&context, 0, sizeof(context));
context.fCtx = fCtxPtr; context.fCtx = fCtxPtr;
context.cctx = ressPtr->cctx; context.cctx = ressPtr->cctx;
context.lastAdaptTime = UTIL_getTime();
context.srcFileSize = srcFileSize; context.srcFileSize = srcFileSize;
context.fileHrs = UTIL_makeHumanReadableSize(srcFileSize); context.fileHrs = UTIL_makeHumanReadableSize(srcFileSize);
@@ -3134,7 +3122,6 @@ FIO_rust_compressZstdCallback(void* fCtx, void* prefs, void* ress,
projection.minAdaptLevel = prefsPtr->minAdaptLevel; projection.minAdaptLevel = prefsPtr->minAdaptLevel;
projection.maxAdaptLevel = prefsPtr->maxAdaptLevel; projection.maxAdaptLevel = prefsPtr->maxAdaptLevel;
projection.maxCLevel = ZSTD_maxCLevel(); projection.maxCLevel = ZSTD_maxCLevel();
projection.adaptiveRefresh = FIO_rust_zstd_adaptiveRefresh;
projection.adaptiveProgression = FIO_rust_zstd_adaptiveProgression; projection.adaptiveProgression = FIO_rust_zstd_adaptiveProgression;
projection.adaptiveSetParameter = FIO_rust_zstd_adaptiveSetParameter; projection.adaptiveSetParameter = FIO_rust_zstd_adaptiveSetParameter;
projection.adaptiveDiagnostic = FIO_rust_zstd_adaptiveDiagnostic; projection.adaptiveDiagnostic = FIO_rust_zstd_adaptiveDiagnostic;
+24 -24
View File
@@ -25,6 +25,7 @@ use std::os::raw::{c_int, c_uint};
use std::ptr; use std::ptr;
use std::sync::{Arc, Condvar, Mutex}; use std::sync::{Arc, Condvar, Mutex};
use std::thread::{self, JoinHandle}; use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
use crate::zstd_compress::{ZSTD_inBuffer, ZSTD_outBuffer}; use crate::zstd_compress::{ZSTD_inBuffer, ZSTD_outBuffer};
@@ -1104,7 +1105,6 @@ const _: () = {
); );
}; };
pub type FIO_rust_zstd_adaptive_refresh_fn = unsafe extern "C" fn(*mut c_void) -> c_int;
pub type FIO_rust_zstd_adaptive_progression_fn = pub type FIO_rust_zstd_adaptive_progression_fn =
unsafe extern "C" fn(*mut c_void, *mut FIO_rust_zstd_progression_t); unsafe extern "C" fn(*mut c_void, *mut FIO_rust_zstd_progression_t);
pub type FIO_rust_zstd_adaptive_set_parameter_fn = unsafe extern "C" fn(*mut c_void, c_int); pub type FIO_rust_zstd_adaptive_set_parameter_fn = unsafe extern "C" fn(*mut c_void, c_int);
@@ -1134,7 +1134,6 @@ pub struct FIO_rust_zstd_compress_projection_t {
pub min_adapt_level: c_int, pub min_adapt_level: c_int,
pub max_adapt_level: c_int, pub max_adapt_level: c_int,
pub max_c_level: c_int, pub max_c_level: c_int,
pub adaptive_refresh: Option<FIO_rust_zstd_adaptive_refresh_fn>,
pub adaptive_progression: Option<FIO_rust_zstd_adaptive_progression_fn>, pub adaptive_progression: Option<FIO_rust_zstd_adaptive_progression_fn>,
pub adaptive_set_parameter: Option<FIO_rust_zstd_adaptive_set_parameter_fn>, pub adaptive_set_parameter: Option<FIO_rust_zstd_adaptive_set_parameter_fn>,
pub adaptive_diagnostic: Option<FIO_rust_zstd_adaptive_diagnostic_fn>, pub adaptive_diagnostic: Option<FIO_rust_zstd_adaptive_diagnostic_fn>,
@@ -1156,16 +1155,20 @@ const _: () = {
== 14 * size_of::<usize>() + 4 * size_of::<c_int>() == 14 * size_of::<usize>() + 4 * size_of::<c_int>()
); );
assert!( assert!(
std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, adaptive_refresh) std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, adaptive_progression)
== callback_offset == callback_offset
); );
assert!(
std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, adaptive_set_parameter)
== callback_offset + size_of::<usize>()
);
assert!( assert!(
std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, adaptive_diagnostic) std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, adaptive_diagnostic)
== callback_offset + 3 * size_of::<usize>() == callback_offset + 2 * size_of::<usize>()
); );
assert!( assert!(
size_of::<FIO_rust_zstd_compress_projection_t>() size_of::<FIO_rust_zstd_compress_projection_t>()
== callback_offset + 4 * size_of::<usize>() == callback_offset + 3 * size_of::<usize>()
); );
}; };
@@ -3558,7 +3561,9 @@ pub unsafe extern "C" fn FIO_rust_zstd_adapt(
zstd_adapt_policy(policy, projection) zstd_adapt_policy(policy, projection)
} }
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] const ZSTD_ADAPT_REFRESH_RATE: Duration = Duration::from_micros(1_000_000 / 6);
#[derive(Debug, Default)]
struct ZstdAdaptiveState { struct ZstdAdaptiveState {
previous_update: FIO_rust_zstd_progression_t, previous_update: FIO_rust_zstd_progression_t,
previous_correction: FIO_rust_zstd_progression_t, previous_correction: FIO_rust_zstd_progression_t,
@@ -3567,6 +3572,7 @@ struct ZstdAdaptiveState {
input_presented: c_uint, input_presented: c_uint,
input_blocked: c_uint, input_blocked: c_uint,
last_job_id: c_uint, last_job_id: c_uint,
last_refresh: Option<Instant>,
} }
unsafe fn zstd_adaptive_report( unsafe fn zstd_adaptive_report(
@@ -3602,12 +3608,12 @@ unsafe fn zstd_adaptive_iteration(
if projection.adaptive_mode == 0 { if projection.adaptive_mode == 0 {
return; return;
} }
let Some(refresh) = projection.adaptive_refresh else { if let Some(last_refresh) = state.last_refresh.as_ref() {
return; if last_refresh.elapsed() <= ZSTD_ADAPT_REFRESH_RATE {
}; return;
if unsafe { refresh(projection.policy_opaque) } == 0 { }
return;
} }
state.last_refresh = Some(Instant::now());
let Some(get_progression) = projection.adaptive_progression else { let Some(get_progression) = projection.adaptive_progression else {
return; return;
}; };
@@ -3909,8 +3915,7 @@ pub unsafe extern "C" fn FIO_rust_compressZstdFrame(
return FIO_RUST_ZSTD_INVALID_PROJECTION; return FIO_RUST_ZSTD_INVALID_PROJECTION;
}; };
if projection.adaptive_mode != 0 if projection.adaptive_mode != 0
&& (projection.adaptive_refresh.is_none() && (projection.adaptive_progression.is_none()
|| projection.adaptive_progression.is_none()
|| projection.adaptive_set_parameter.is_none() || projection.adaptive_set_parameter.is_none()
|| projection.adaptive_diagnostic.is_none()) || projection.adaptive_diagnostic.is_none())
{ {
@@ -3939,7 +3944,10 @@ pub unsafe extern "C" fn FIO_rust_compressZstdFrame(
let mut out_file_size = 0_u64; let mut out_file_size = 0_u64;
let mut directive = FIO_RUST_ZSTD_E_CONTINUE; let mut directive = FIO_RUST_ZSTD_E_CONTINUE;
let mut compression_level = compression_level; let mut compression_level = compression_level;
let mut adaptive_state = ZstdAdaptiveState::default(); let mut adaptive_state = ZstdAdaptiveState {
last_refresh: Some(Instant::now()),
..ZstdAdaptiveState::default()
};
loop { loop {
if input_pos == input_size { if input_pos == input_size {
@@ -7754,16 +7762,9 @@ mod tests {
#[derive(Default)] #[derive(Default)]
struct AdaptiveCallbackState { struct AdaptiveCallbackState {
progression: FIO_rust_zstd_progression_t, progression: FIO_rust_zstd_progression_t,
refresh_calls: usize,
events: Vec<c_int>, events: Vec<c_int>,
} }
unsafe extern "C" fn adaptive_test_refresh(opaque: *mut c_void) -> c_int {
let state = unsafe { &mut *opaque.cast::<AdaptiveCallbackState>() };
state.refresh_calls += 1;
1
}
unsafe extern "C" fn adaptive_test_progression( unsafe extern "C" fn adaptive_test_progression(
opaque: *mut c_void, opaque: *mut c_void,
progression: *mut FIO_rust_zstd_progression_t, progression: *mut FIO_rust_zstd_progression_t,
@@ -7812,7 +7813,6 @@ mod tests {
min_adapt_level: 1, min_adapt_level: 1,
max_adapt_level: 5, max_adapt_level: 5,
max_c_level: 22, max_c_level: 22,
adaptive_refresh: Some(adaptive_test_refresh),
adaptive_progression: Some(adaptive_test_progression), adaptive_progression: Some(adaptive_test_progression),
adaptive_set_parameter: Some(adaptive_test_set_parameter), adaptive_set_parameter: Some(adaptive_test_set_parameter),
adaptive_diagnostic: Some(adaptive_test_diagnostic), adaptive_diagnostic: Some(adaptive_test_diagnostic),
@@ -7832,6 +7832,7 @@ mod tests {
}; };
let projection = adaptive_test_projection(&mut callbacks); let projection = adaptive_test_projection(&mut callbacks);
let mut state = ZstdAdaptiveState::default(); let mut state = ZstdAdaptiveState::default();
assert!(state.last_refresh.is_none());
let mut compression_level = 3; let mut compression_level = 3;
unsafe { unsafe {
@@ -7849,7 +7850,7 @@ mod tests {
104, 104,
] ]
); );
assert_eq!(callbacks.refresh_calls, 1); assert!(state.last_refresh.is_some());
assert_eq!(compression_level, 4); assert_eq!(compression_level, 4);
assert_eq!(state.previous_update, callbacks.progression); assert_eq!(state.previous_update, callbacks.progression);
assert_eq!(state.last_job_id, 3); assert_eq!(state.last_job_id, 3);
@@ -8013,7 +8014,6 @@ mod tests {
min_adapt_level: 0, min_adapt_level: 0,
max_adapt_level: 0, max_adapt_level: 0,
max_c_level: 0, max_c_level: 0,
adaptive_refresh: None,
adaptive_progression: None, adaptive_progression: None,
adaptive_set_parameter: None, adaptive_set_parameter: None,
adaptive_diagnostic: None, adaptive_diagnostic: None,