diff --git a/programs/fileio.c b/programs/fileio.c index 7c6e3fc03..ad55b137e 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -1643,7 +1643,6 @@ typedef char FIO_rust_zstd_progression_layout[ == 4 * sizeof(U64) + sizeof(unsigned) && sizeof(FIO_rust_zstd_progression_t) == 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)( void* opaque, FIO_rust_zstd_progression_t* progression); typedef void (*FIO_rust_zstd_adaptive_set_parameter_fn)( @@ -1672,7 +1671,6 @@ typedef struct { int minAdaptLevel; int maxAdaptLevel; int maxCLevel; - FIO_rust_zstd_adaptive_refresh_fn adaptiveRefresh; FIO_rust_zstd_adaptive_progression_fn adaptiveProgression; FIO_rust_zstd_adaptive_set_parameter_fn adaptiveSetParameter; FIO_rust_zstd_adaptive_diagnostic_fn adaptiveDiagnostic; @@ -1684,15 +1682,18 @@ typedef char FIO_rust_zstd_compress_projection_layout[ == 14 * sizeof(void*) + sizeof(int) && offsetof(FIO_rust_zstd_compress_projection_t, maxCLevel) == 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) / 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) == ((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) == ((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]; int FIO_rust_compressZstdFrame( @@ -2882,7 +2883,6 @@ enum { typedef struct { FIO_ctx_t* fCtx; ZSTD_CCtx* cctx; - UTIL_time_t lastAdaptTime; U64 srcFileSize; UTIL_HumanReadableSize_t fileHrs; } FIO_rust_zstd_projection_context_t; @@ -2936,17 +2936,6 @@ static void FIO_rust_zstd_sparseWriteEnd(void* 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( 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)); context.fCtx = fCtxPtr; context.cctx = ressPtr->cctx; - context.lastAdaptTime = UTIL_getTime(); context.srcFileSize = srcFileSize; context.fileHrs = UTIL_makeHumanReadableSize(srcFileSize); @@ -3134,7 +3122,6 @@ FIO_rust_compressZstdCallback(void* fCtx, void* prefs, void* ress, projection.minAdaptLevel = prefsPtr->minAdaptLevel; projection.maxAdaptLevel = prefsPtr->maxAdaptLevel; projection.maxCLevel = ZSTD_maxCLevel(); - projection.adaptiveRefresh = FIO_rust_zstd_adaptiveRefresh; projection.adaptiveProgression = FIO_rust_zstd_adaptiveProgression; projection.adaptiveSetParameter = FIO_rust_zstd_adaptiveSetParameter; projection.adaptiveDiagnostic = FIO_rust_zstd_adaptiveDiagnostic; diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index eb407f808..269904925 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -25,6 +25,7 @@ use std::os::raw::{c_int, c_uint}; use std::ptr; use std::sync::{Arc, Condvar, Mutex}; use std::thread::{self, JoinHandle}; +use std::time::{Duration, Instant}; 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 = 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); @@ -1134,7 +1134,6 @@ pub struct FIO_rust_zstd_compress_projection_t { pub min_adapt_level: c_int, pub max_adapt_level: c_int, pub max_c_level: c_int, - pub adaptive_refresh: Option, pub adaptive_progression: Option, pub adaptive_set_parameter: Option, pub adaptive_diagnostic: Option, @@ -1156,16 +1155,20 @@ const _: () = { == 14 * size_of::() + 4 * size_of::() ); 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 ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, adaptive_set_parameter) + == callback_offset + size_of::() + ); assert!( std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, adaptive_diagnostic) - == callback_offset + 3 * size_of::() + == callback_offset + 2 * size_of::() ); assert!( size_of::() - == callback_offset + 4 * size_of::() + == callback_offset + 3 * size_of::() ); }; @@ -3558,7 +3561,9 @@ pub unsafe extern "C" fn FIO_rust_zstd_adapt( 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 { previous_update: FIO_rust_zstd_progression_t, previous_correction: FIO_rust_zstd_progression_t, @@ -3567,6 +3572,7 @@ struct ZstdAdaptiveState { input_presented: c_uint, input_blocked: c_uint, last_job_id: c_uint, + last_refresh: Option, } unsafe fn zstd_adaptive_report( @@ -3602,12 +3608,12 @@ unsafe fn zstd_adaptive_iteration( if projection.adaptive_mode == 0 { return; } - let Some(refresh) = projection.adaptive_refresh else { - return; - }; - if unsafe { refresh(projection.policy_opaque) } == 0 { - return; + if let Some(last_refresh) = state.last_refresh.as_ref() { + if last_refresh.elapsed() <= ZSTD_ADAPT_REFRESH_RATE { + return; + } } + state.last_refresh = Some(Instant::now()); let Some(get_progression) = projection.adaptive_progression else { return; }; @@ -3909,8 +3915,7 @@ pub unsafe extern "C" fn FIO_rust_compressZstdFrame( return FIO_RUST_ZSTD_INVALID_PROJECTION; }; 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_diagnostic.is_none()) { @@ -3939,7 +3944,10 @@ pub unsafe extern "C" fn FIO_rust_compressZstdFrame( let mut out_file_size = 0_u64; let mut directive = FIO_RUST_ZSTD_E_CONTINUE; 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 { if input_pos == input_size { @@ -7754,16 +7762,9 @@ mod tests { #[derive(Default)] struct AdaptiveCallbackState { progression: FIO_rust_zstd_progression_t, - refresh_calls: usize, events: Vec, } - unsafe extern "C" fn adaptive_test_refresh(opaque: *mut c_void) -> c_int { - let state = unsafe { &mut *opaque.cast::() }; - state.refresh_calls += 1; - 1 - } - unsafe extern "C" fn adaptive_test_progression( opaque: *mut c_void, progression: *mut FIO_rust_zstd_progression_t, @@ -7812,7 +7813,6 @@ mod tests { min_adapt_level: 1, max_adapt_level: 5, max_c_level: 22, - adaptive_refresh: Some(adaptive_test_refresh), adaptive_progression: Some(adaptive_test_progression), adaptive_set_parameter: Some(adaptive_test_set_parameter), adaptive_diagnostic: Some(adaptive_test_diagnostic), @@ -7832,6 +7832,7 @@ mod tests { }; let projection = adaptive_test_projection(&mut callbacks); let mut state = ZstdAdaptiveState::default(); + assert!(state.last_refresh.is_none()); let mut compression_level = 3; unsafe { @@ -7849,7 +7850,7 @@ mod tests { 104, ] ); - assert_eq!(callbacks.refresh_calls, 1); + assert!(state.last_refresh.is_some()); assert_eq!(compression_level, 4); assert_eq!(state.previous_update, callbacks.progression); assert_eq!(state.last_job_id, 3); @@ -8013,7 +8014,6 @@ mod tests { min_adapt_level: 0, max_adapt_level: 0, max_c_level: 0, - adaptive_refresh: None, adaptive_progression: None, adaptive_set_parameter: None, adaptive_diagnostic: None,