diff --git a/programs/fileio.c b/programs/fileio.c index 7e76b7437..806f9dde2 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -1968,6 +1968,68 @@ typedef enum { FIO_rust_zstd_faster } FIO_rust_zstd_speed_change_e; +/* Rust owns only the scalar adaptive predicates. Keep the FIO context, + * preference, and ZSTD progression layouts private to this translation unit. */ +typedef struct { + U64 consumed; + U64 previousConsumed; + unsigned nbActiveWorkers; + U64 newlyProduced; + U64 newlyFlushed; + unsigned flushWaiting; + unsigned inputBlocked; + unsigned inputPresented; + U64 newlyIngested; + U64 newlyConsumed; + int compressionLevel; + int minAdaptLevel; + int maxAdaptLevel; + int maxCLevel; +} FIO_rust_zstd_adapt_projection_t; +typedef char FIO_rust_zstd_adapt_projection_layout[ + (offsetof(FIO_rust_zstd_adapt_projection_t, consumed) == 0 + && offsetof(FIO_rust_zstd_adapt_projection_t, previousConsumed) + == sizeof(U64) + && offsetof(FIO_rust_zstd_adapt_projection_t, nbActiveWorkers) + == 2 * sizeof(U64) + && offsetof(FIO_rust_zstd_adapt_projection_t, newlyProduced) + == (sizeof(void*) == 8 ? 24 : 20) + && offsetof(FIO_rust_zstd_adapt_projection_t, newlyFlushed) + == (sizeof(void*) == 8 ? 32 : 28) + && offsetof(FIO_rust_zstd_adapt_projection_t, flushWaiting) + == (sizeof(void*) == 8 ? 40 : 36) + && offsetof(FIO_rust_zstd_adapt_projection_t, inputBlocked) + == (sizeof(void*) == 8 ? 44 : 40) + && offsetof(FIO_rust_zstd_adapt_projection_t, inputPresented) + == (sizeof(void*) == 8 ? 48 : 44) + && offsetof(FIO_rust_zstd_adapt_projection_t, newlyIngested) + == (sizeof(void*) == 8 ? 56 : 48) + && offsetof(FIO_rust_zstd_adapt_projection_t, newlyConsumed) + == (sizeof(void*) == 8 ? 64 : 56) + && offsetof(FIO_rust_zstd_adapt_projection_t, compressionLevel) + == (sizeof(void*) == 8 ? 72 : 64) + && offsetof(FIO_rust_zstd_adapt_projection_t, minAdaptLevel) + == (sizeof(void*) == 8 ? 76 : 68) + && offsetof(FIO_rust_zstd_adapt_projection_t, maxAdaptLevel) + == (sizeof(void*) == 8 ? 80 : 72) + && offsetof(FIO_rust_zstd_adapt_projection_t, maxCLevel) + == (sizeof(void*) == 8 ? 84 : 76) + && sizeof(FIO_rust_zstd_adapt_projection_t) + == (sizeof(void*) == 8 ? 88 : 80)) + ? 1 : -1]; + +enum { + FIO_RUST_ZSTD_ADAPT_OUTPUT_BLOCKED, + FIO_RUST_ZSTD_ADAPT_OUTPUT_BACKLOG, + FIO_RUST_ZSTD_ADAPT_INPUT_STARVATION, + FIO_RUST_ZSTD_ADAPT_BLOCKED_INPUT, + FIO_RUST_ZSTD_ADAPT_LEVEL_SLOWER, + FIO_RUST_ZSTD_ADAPT_LEVEL_FASTER +}; + +int FIO_rust_zstd_adapt( + int policy, const FIO_rust_zstd_adapt_projection_t* projection); + typedef struct { FIO_ctx_t* fCtx; FIO_prefs_t* prefs; @@ -2070,13 +2132,14 @@ static void FIO_rust_zstd_iteration(void* opaque, const char* srcFileName, (FIO_rust_zstd_projection_context_t*)opaque; FIO_prefs_t* const prefs = context->prefs; FIO_ctx_t* const fCtx = context->fCtx; + FIO_rust_zstd_adapt_projection_t adaptProjection; context->inputPresented++; if (oldInputPos == newInputPos) context->inputBlocked++; if (!toFlushNow) context->flushWaiting = 1; - /* Adaptive mode remains in C so its policy and private preference/context - * access stay identical while Rust owns the surrounding stream loop. */ + /* C retains the clock gate, progression snapshots, diagnostics, and + * private context mutations; Rust supplies only scalar policy decisions. */ if (prefs->adaptiveMode && UTIL_clockSpanMicro(context->lastAdaptTime) > REFRESH_RATE) { ZSTD_frameProgression const zfp = ZSTD_getFrameProgression(context->cctx); @@ -2096,16 +2159,25 @@ static void FIO_rust_zstd_iteration(void* opaque, const char* srcFileName, * either because output is slow and all buffers are full * or because input is slow and no job can start while waiting for at least one buffer to be filled. * note : exclude starting part, since currentJobID > 1 */ - if ((zfp.consumed == context->previousZfpUpdate.consumed) - && (zfp.nbActiveWorkers == 0)) { + memset(&adaptProjection, 0, sizeof(adaptProjection)); + adaptProjection.consumed = zfp.consumed; + adaptProjection.previousConsumed = context->previousZfpUpdate.consumed; + adaptProjection.nbActiveWorkers = zfp.nbActiveWorkers; + if (FIO_rust_zstd_adapt( + FIO_RUST_ZSTD_ADAPT_OUTPUT_BLOCKED, &adaptProjection) + == FIO_rust_zstd_slower) { DISPLAYLEVEL(6, "all buffers full : compression stopped => slow down \n") context->speedChange = FIO_rust_zstd_slower; } context->previousZfpUpdate = zfp; - if ((newlyProduced > (newlyFlushed * 9 / 8)) - && (context->flushWaiting == 0)) { + adaptProjection.newlyProduced = newlyProduced; + adaptProjection.newlyFlushed = newlyFlushed; + adaptProjection.flushWaiting = context->flushWaiting; + if (FIO_rust_zstd_adapt( + FIO_RUST_ZSTD_ADAPT_OUTPUT_BACKLOG, &adaptProjection) + == FIO_rust_zstd_slower) { DISPLAYLEVEL(6, "compression faster than flush (%llu > %llu), and flushed was never slowed down by lack of production => slow down \n", newlyProduced, newlyFlushed); context->speedChange = FIO_rust_zstd_slower; @@ -2119,7 +2191,11 @@ static void FIO_rust_zstd_iteration(void* opaque, const char* srcFileName, /* check input speed */ if (zfp.currentJobID > (unsigned)(prefs->nbWorkers+1)) { - if (context->inputBlocked <= 0) { + memset(&adaptProjection, 0, sizeof(adaptProjection)); + adaptProjection.inputBlocked = context->inputBlocked; + if (FIO_rust_zstd_adapt( + FIO_RUST_ZSTD_ADAPT_INPUT_STARVATION, &adaptProjection) + == FIO_rust_zstd_slower) { DISPLAYLEVEL(6, "input is never blocked => input is slower than ingestion \n"); context->speedChange = FIO_rust_zstd_slower; } else if (context->speedChange == FIO_rust_zstd_noChange) { @@ -2138,9 +2214,15 @@ static void FIO_rust_zstd_iteration(void* opaque, const char* srcFileName, (double)context->inputBlocked/context->inputPresented*100, (unsigned)newlyIngested, (unsigned)newlyConsumed, (unsigned)newlyFlushed, (unsigned)newlyProduced); - if ((context->inputBlocked > context->inputPresented / 8) - && (newlyFlushed * 33 / 32 > newlyProduced) - && (newlyIngested * 33 / 32 > newlyConsumed)) { + adaptProjection.inputBlocked = context->inputBlocked; + adaptProjection.inputPresented = context->inputPresented; + adaptProjection.newlyIngested = newlyIngested; + adaptProjection.newlyConsumed = newlyConsumed; + adaptProjection.newlyProduced = newlyProduced; + adaptProjection.newlyFlushed = newlyFlushed; + if (FIO_rust_zstd_adapt( + FIO_RUST_ZSTD_ADAPT_BLOCKED_INPUT, &adaptProjection) + == FIO_rust_zstd_faster) { DISPLAYLEVEL(6, "recommend faster as in(%llu) >= (%llu)comp(%llu) <= out(%llu) \n", newlyIngested, newlyConsumed, newlyProduced, newlyFlushed); @@ -2153,22 +2235,23 @@ static void FIO_rust_zstd_iteration(void* opaque, const char* srcFileName, if (context->speedChange == FIO_rust_zstd_slower) { DISPLAYLEVEL(6, "slower speed , higher compression \n") - (*compressionLevel)++; - if (*compressionLevel > ZSTD_maxCLevel()) - *compressionLevel = ZSTD_maxCLevel(); - if (*compressionLevel > prefs->maxAdaptLevel) - *compressionLevel = prefs->maxAdaptLevel; - *compressionLevel += (*compressionLevel == 0); + memset(&adaptProjection, 0, sizeof(adaptProjection)); + adaptProjection.compressionLevel = *compressionLevel; + adaptProjection.maxAdaptLevel = prefs->maxAdaptLevel; + adaptProjection.maxCLevel = ZSTD_maxCLevel(); + *compressionLevel = FIO_rust_zstd_adapt( + FIO_RUST_ZSTD_ADAPT_LEVEL_SLOWER, &adaptProjection); ZSTD_CCtx_setParameter(context->cctx, ZSTD_c_compressionLevel, *compressionLevel); } if (context->speedChange == FIO_rust_zstd_faster) { DISPLAYLEVEL(6, "faster speed , lighter compression \n") - (*compressionLevel)--; - if (*compressionLevel < prefs->minAdaptLevel) - *compressionLevel = prefs->minAdaptLevel; - *compressionLevel -= (*compressionLevel == 0); + memset(&adaptProjection, 0, sizeof(adaptProjection)); + adaptProjection.compressionLevel = *compressionLevel; + adaptProjection.minAdaptLevel = prefs->minAdaptLevel; + *compressionLevel = FIO_rust_zstd_adapt( + FIO_RUST_ZSTD_ADAPT_LEVEL_FASTER, &adaptProjection); ZSTD_CCtx_setParameter(context->cctx, ZSTD_c_compressionLevel, *compressionLevel); diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index bd5e57bb1..d2a478447 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -294,8 +294,103 @@ pub type FIO_rust_zstd_compress_stream_fn = unsafe extern "C" fn( pub type FIO_rust_zstd_iteration_fn = unsafe extern "C" fn(*mut c_void, *const c_char, *mut c_int, usize, usize, usize); -/// Rust owns the zstd read/compress/write loop. C keeps the zstd context, -/// adaptive policy, diagnostics, and codec calls behind opaque callbacks. +pub const FIO_RUST_ZSTD_ADAPT_NO_CHANGE: c_int = 0; +pub const FIO_RUST_ZSTD_ADAPT_SLOWER: c_int = 1; +pub const FIO_RUST_ZSTD_ADAPT_FASTER: c_int = 2; +pub const FIO_RUST_ZSTD_ADAPT_INVALID_PROJECTION: c_int = -1; + +pub const FIO_RUST_ZSTD_ADAPT_OUTPUT_BLOCKED: c_int = 0; +pub const FIO_RUST_ZSTD_ADAPT_OUTPUT_BACKLOG: c_int = 1; +pub const FIO_RUST_ZSTD_ADAPT_INPUT_STARVATION: c_int = 2; +pub const FIO_RUST_ZSTD_ADAPT_BLOCKED_INPUT: c_int = 3; +pub const FIO_RUST_ZSTD_ADAPT_LEVEL_SLOWER: c_int = 4; +pub const FIO_RUST_ZSTD_ADAPT_LEVEL_FASTER: c_int = 5; + +/// Scalar inputs for the adaptive decision policy. C keeps the +/// `FIO_rust_zstd_projection_context_t`, `ZSTD_frameProgression`, and +/// preference layouts private; this projection carries only the values used +/// by the policy predicates and level clamps. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct FIO_rust_zstd_adapt_projection_t { + pub consumed: u64, + pub previous_consumed: u64, + pub nb_active_workers: c_uint, + pub newly_produced: u64, + pub newly_flushed: u64, + pub flush_waiting: c_uint, + pub input_blocked: c_uint, + pub input_presented: c_uint, + pub newly_ingested: u64, + pub newly_consumed: u64, + pub compression_level: c_int, + pub min_adapt_level: c_int, + pub max_adapt_level: c_int, + pub max_c_level: c_int, +} +const _: () = { + assert!(std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, consumed) == 0); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, previous_consumed) + == size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, nb_active_workers) + == 2 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, newly_produced) + == if size_of::() == 8 { 24 } else { 20 } + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, newly_flushed) + == if size_of::() == 8 { 32 } else { 28 } + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, flush_waiting) + == if size_of::() == 8 { 40 } else { 36 } + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, input_blocked) + == if size_of::() == 8 { 44 } else { 40 } + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, input_presented) + == if size_of::() == 8 { 48 } else { 44 } + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, newly_ingested) + == if size_of::() == 8 { 56 } else { 48 } + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, newly_consumed) + == if size_of::() == 8 { 64 } else { 56 } + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, compression_level) + == if size_of::() == 8 { 72 } else { 64 } + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, min_adapt_level) + == if size_of::() == 8 { 76 } else { 68 } + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, max_adapt_level) + == if size_of::() == 8 { 80 } else { 72 } + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, max_c_level) + == if size_of::() == 8 { 84 } else { 76 } + ); + assert!( + size_of::() + == if size_of::() == 8 { 88 } else { 80 } + ); +}; + +/// Rust owns only the scalar adaptive predicates and compression-level +/// normalization. C retains the surrounding iteration order, diagnostics, +/// progression snapshots, and `ZSTD_CCtx_setParameter()` mutation. #[repr(C)] pub struct FIO_rust_zstd_compress_projection_t { pub read_opaque: *mut c_void, @@ -2027,9 +2122,88 @@ pub unsafe extern "C" fn FIO_rust_compressFilenameInternal( FIO_RUST_COMPRESS_OK } +#[inline] +fn zstd_adapt_level(projection: &FIO_rust_zstd_adapt_projection_t, slower: bool) -> c_int { + if slower { + let mut level = projection.compression_level.wrapping_add(1); + if level > projection.max_c_level { + level = projection.max_c_level; + } + if level > projection.max_adapt_level { + level = projection.max_adapt_level; + } + level.wrapping_add(i32::from(level == 0)) + } else { + let mut level = projection.compression_level.wrapping_sub(1); + if level < projection.min_adapt_level { + level = projection.min_adapt_level; + } + level.wrapping_sub(i32::from(level == 0)) + } +} + +#[inline] +fn zstd_adapt_policy(policy: c_int, projection: &FIO_rust_zstd_adapt_projection_t) -> c_int { + match policy { + FIO_RUST_ZSTD_ADAPT_OUTPUT_BLOCKED => { + if projection.consumed == projection.previous_consumed + && projection.nb_active_workers == 0 + { + FIO_RUST_ZSTD_ADAPT_SLOWER + } else { + FIO_RUST_ZSTD_ADAPT_NO_CHANGE + } + } + FIO_RUST_ZSTD_ADAPT_OUTPUT_BACKLOG => { + if projection.newly_produced > projection.newly_flushed.wrapping_mul(9) / 8 + && projection.flush_waiting == 0 + { + FIO_RUST_ZSTD_ADAPT_SLOWER + } else { + FIO_RUST_ZSTD_ADAPT_NO_CHANGE + } + } + FIO_RUST_ZSTD_ADAPT_INPUT_STARVATION => { + if projection.input_blocked == 0 { + FIO_RUST_ZSTD_ADAPT_SLOWER + } else { + FIO_RUST_ZSTD_ADAPT_NO_CHANGE + } + } + FIO_RUST_ZSTD_ADAPT_BLOCKED_INPUT => { + if projection.input_blocked > projection.input_presented / 8 + && projection.newly_flushed.wrapping_mul(33) / 32 > projection.newly_produced + && projection.newly_ingested.wrapping_mul(33) / 32 > projection.newly_consumed + { + FIO_RUST_ZSTD_ADAPT_FASTER + } else { + FIO_RUST_ZSTD_ADAPT_NO_CHANGE + } + } + FIO_RUST_ZSTD_ADAPT_LEVEL_SLOWER => zstd_adapt_level(projection, true), + FIO_RUST_ZSTD_ADAPT_LEVEL_FASTER => zstd_adapt_level(projection, false), + _ => FIO_RUST_ZSTD_ADAPT_INVALID_PROJECTION, + } +} + +/// Evaluate one scalar adaptive-policy branch. C owns the caller's branch +/// order and commits the returned decision or level through its private +/// context and diagnostics. +#[no_mangle] +pub unsafe extern "C" fn FIO_rust_zstd_adapt( + policy: c_int, + projection: *const FIO_rust_zstd_adapt_projection_t, +) -> c_int { + let Some(projection) = (unsafe { projection.as_ref() }) else { + return FIO_RUST_ZSTD_ADAPT_INVALID_PROJECTION; + }; + zstd_adapt_policy(policy, projection) +} + /// Compresses one zstd frame through the C-owned zstd context and adaptive /// policy. Rust owns the stream loop and exact pool accounting; C callbacks -/// retain `ZSTD_compressStream2()`, diagnostics, and all private CLI state. +/// retain `ZSTD_compressStream2()`, adaptive diagnostics, and all private CLI +/// state. #[no_mangle] pub unsafe extern "C" fn FIO_rust_compressZstdFrame( projection: *const FIO_rust_zstd_compress_projection_t, @@ -3810,6 +3984,137 @@ impl PoolInner { mod tests { use super::*; + fn run_zstd_adapt(policy: c_int, projection: &FIO_rust_zstd_adapt_projection_t) -> c_int { + unsafe { FIO_rust_zstd_adapt(policy, projection) } + } + + #[test] + fn zstd_adapt_slows_when_output_backlog_grows() { + let projection = FIO_rust_zstd_adapt_projection_t { + newly_produced: 10, + newly_flushed: 8, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + + assert_eq!( + run_zstd_adapt(FIO_RUST_ZSTD_ADAPT_OUTPUT_BACKLOG, &projection), + FIO_RUST_ZSTD_ADAPT_SLOWER + ); + } + + #[test] + fn zstd_adapt_slows_when_output_is_blocked() { + let projection = FIO_rust_zstd_adapt_projection_t { + consumed: 100, + previous_consumed: 100, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + + assert_eq!( + run_zstd_adapt(FIO_RUST_ZSTD_ADAPT_OUTPUT_BLOCKED, &projection), + FIO_RUST_ZSTD_ADAPT_SLOWER + ); + } + + #[test] + fn zstd_adapt_slows_when_input_never_blocks() { + let projection = FIO_rust_zstd_adapt_projection_t::default(); + + assert_eq!( + run_zstd_adapt(FIO_RUST_ZSTD_ADAPT_INPUT_STARVATION, &projection), + FIO_RUST_ZSTD_ADAPT_SLOWER + ); + } + + #[test] + fn zstd_adapt_speeds_up_when_blocked_input_lags_both_sides() { + let projection = FIO_rust_zstd_adapt_projection_t { + input_blocked: 2, + input_presented: 8, + newly_ingested: 34, + newly_consumed: 32, + newly_produced: 32, + newly_flushed: 34, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + + assert_eq!( + run_zstd_adapt(FIO_RUST_ZSTD_ADAPT_BLOCKED_INPUT, &projection), + FIO_RUST_ZSTD_ADAPT_FASTER + ); + } + + #[test] + fn zstd_adapt_preserves_noop_boundaries() { + let backlog_waiting = FIO_rust_zstd_adapt_projection_t { + newly_produced: 10, + newly_flushed: 8, + flush_waiting: 1, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + let insufficiently_blocked = FIO_rust_zstd_adapt_projection_t { + input_blocked: 1, + input_presented: 8, + newly_ingested: 34, + newly_consumed: 32, + newly_produced: 32, + newly_flushed: 34, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + + assert_eq!( + run_zstd_adapt(FIO_RUST_ZSTD_ADAPT_OUTPUT_BACKLOG, &backlog_waiting), + FIO_RUST_ZSTD_ADAPT_NO_CHANGE + ); + assert_eq!( + run_zstd_adapt(FIO_RUST_ZSTD_ADAPT_BLOCKED_INPUT, &insufficiently_blocked), + FIO_RUST_ZSTD_ADAPT_NO_CHANGE + ); + } + + #[test] + fn zstd_adapt_clamps_levels_and_avoids_zero() { + let slower_max = FIO_rust_zstd_adapt_projection_t { + compression_level: 21, + max_adapt_level: 20, + max_c_level: 22, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + let faster_min = FIO_rust_zstd_adapt_projection_t { + compression_level: 2, + min_adapt_level: 3, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + let slower_zero = FIO_rust_zstd_adapt_projection_t { + compression_level: 0, + max_adapt_level: 0, + max_c_level: 22, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + let faster_zero = FIO_rust_zstd_adapt_projection_t { + compression_level: 1, + min_adapt_level: 0, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + + assert_eq!( + run_zstd_adapt(FIO_RUST_ZSTD_ADAPT_LEVEL_SLOWER, &slower_max), + 20 + ); + assert_eq!( + run_zstd_adapt(FIO_RUST_ZSTD_ADAPT_LEVEL_FASTER, &faster_min), + 3 + ); + assert_eq!( + run_zstd_adapt(FIO_RUST_ZSTD_ADAPT_LEVEL_SLOWER, &slower_zero), + 1 + ); + assert_eq!( + run_zstd_adapt(FIO_RUST_ZSTD_ADAPT_LEVEL_FASTER, &faster_zero), + -1 + ); + } + fn test_prefs(async_io: c_int) -> FIO_prefs_t { let mut prefs: FIO_prefs_t = unsafe { std::mem::zeroed() }; prefs.asyncIO = async_io;