diff --git a/programs/fileio.c b/programs/fileio.c index 9d3880061..1927fe2b1 100644 --- a/programs/fileio.c +++ b/programs/fileio.c @@ -1552,6 +1552,34 @@ void FIO_rust_zstd_compressStreamDisplay( typedef void (*FIO_rust_zstd_iteration_fn)( void* opaque, const char* srcFileName, int* compressionLevel, size_t oldInputPos, size_t newInputPos, size_t toFlushNow); +typedef struct FIO_rust_zstd_adapt_projection_s + FIO_rust_zstd_adapt_projection_t; +typedef struct { + U64 ingested; + U64 consumed; + U64 produced; + U64 flushed; + unsigned currentJobID; + unsigned nbActiveWorkers; +} FIO_rust_zstd_progression_t; +typedef char FIO_rust_zstd_progression_layout[ + (offsetof(FIO_rust_zstd_progression_t, ingested) == 0 + && offsetof(FIO_rust_zstd_progression_t, consumed) == sizeof(U64) + && offsetof(FIO_rust_zstd_progression_t, produced) == 2 * sizeof(U64) + && offsetof(FIO_rust_zstd_progression_t, flushed) == 3 * sizeof(U64) + && offsetof(FIO_rust_zstd_progression_t, currentJobID) == 4 * sizeof(U64) + && offsetof(FIO_rust_zstd_progression_t, nbActiveWorkers) + == 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)( + void* opaque, int compressionLevel); +typedef void (*FIO_rust_zstd_adaptive_diagnostic_fn)( + void* opaque, int diagnostic, + const FIO_rust_zstd_adapt_projection_t* projection); typedef struct { void* readOpaque; @@ -1568,7 +1596,33 @@ typedef struct { FIO_rust_zstd_compress_stream_fn compressStream; FIO_rust_zstd_iteration_fn iteration; FIO_rust_zstd_compress_display_fn compressStreamDisplay; + int adaptiveMode; + int nbWorkers; + 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; } FIO_rust_zstd_compress_projection_t; +typedef char FIO_rust_zstd_compress_projection_layout[ + (offsetof(FIO_rust_zstd_compress_projection_t, adaptiveMode) + == 14 * sizeof(void*) + && offsetof(FIO_rust_zstd_compress_projection_t, nbWorkers) + == 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) + == ((14 * sizeof(void*) + 5 * sizeof(int) + sizeof(void*) - 1) + / 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(FIO_rust_zstd_compress_projection_t) + == ((14 * sizeof(void*) + 5 * sizeof(int) + sizeof(void*) - 1) + / sizeof(void*)) * sizeof(void*) + 4 * sizeof(void*)) + ? 1 : -1]; int FIO_rust_compressZstdFrame( const FIO_rust_zstd_compress_projection_t* projection, @@ -2621,15 +2675,9 @@ FIO_compressLz4Frame(cRess_t* ress, } #endif -typedef enum { - FIO_rust_zstd_noChange, - FIO_rust_zstd_slower, - FIO_rust_zstd_faster -} FIO_rust_zstd_speed_change_e; - -/* Rust owns only the scalar adaptive predicates. Keep the FIO context, +/* Rust owns scalar adaptive state and policy. Keep the FIO context, * preference, and ZSTD progression layouts private to this translation unit. */ -typedef struct { +struct FIO_rust_zstd_adapt_projection_s { U64 consumed; U64 previousConsumed; unsigned nbActiveWorkers; @@ -2644,7 +2692,7 @@ typedef struct { 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) @@ -2678,28 +2726,19 @@ typedef char FIO_rust_zstd_adapt_projection_layout[ ? 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 + FIO_RUST_ZSTD_ADAPT_DIAG_OUTPUT_BLOCKED, + FIO_RUST_ZSTD_ADAPT_DIAG_OUTPUT_BACKLOG, + FIO_RUST_ZSTD_ADAPT_DIAG_CHECK, + FIO_RUST_ZSTD_ADAPT_DIAG_INPUT_STARVATION, + FIO_RUST_ZSTD_ADAPT_DIAG_INPUT_STATS, + FIO_RUST_ZSTD_ADAPT_DIAG_RECOMMEND_FASTER, + FIO_RUST_ZSTD_ADAPT_DIAG_SLOWER_LEVEL, + FIO_RUST_ZSTD_ADAPT_DIAG_FASTER_LEVEL }; -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; ZSTD_CCtx* cctx; - ZSTD_frameProgression previousZfpUpdate; - ZSTD_frameProgression previousZfpCorrection; - FIO_rust_zstd_speed_change_e speedChange; - unsigned flushWaiting; - unsigned inputPresented; - unsigned inputBlocked; - unsigned lastJobID; UTIL_time_t lastAdaptTime; U64 srcFileSize; UTIL_HumanReadableSize_t fileHrs; @@ -2754,6 +2793,95 @@ 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) +{ + FIO_rust_zstd_projection_context_t* const context = + (FIO_rust_zstd_projection_context_t*)opaque; + ZSTD_frameProgression const zfp = ZSTD_getFrameProgression(context->cctx); + assert(projection != NULL); + projection->ingested = zfp.ingested; + projection->consumed = zfp.consumed; + projection->produced = zfp.produced; + projection->flushed = zfp.flushed; + projection->currentJobID = zfp.currentJobID; + projection->nbActiveWorkers = zfp.nbActiveWorkers; +} + +static void FIO_rust_zstd_adaptiveSetParameter(void* opaque, + int compressionLevel) +{ + FIO_rust_zstd_projection_context_t* const context = + (FIO_rust_zstd_projection_context_t*)opaque; + (void)ZSTD_CCtx_setParameter(context->cctx, + ZSTD_c_compressionLevel, + compressionLevel); +} + +static void FIO_rust_zstd_adaptiveDiagnostic( + void* opaque, int diagnostic, + const FIO_rust_zstd_adapt_projection_t* projection) +{ + (void)opaque; + switch (diagnostic) { + case FIO_RUST_ZSTD_ADAPT_DIAG_OUTPUT_BLOCKED: + DISPLAYLEVEL(6, "all buffers full : compression stopped => slow down \n") + break; + case FIO_RUST_ZSTD_ADAPT_DIAG_OUTPUT_BACKLOG: + assert(projection != NULL); + DISPLAYLEVEL(6, "compression faster than flush (%llu > %llu), and flushed was never slowed down by lack of production => slow down \n", + (unsigned long long)projection->newlyProduced, + (unsigned long long)projection->newlyFlushed); + break; + case FIO_RUST_ZSTD_ADAPT_DIAG_CHECK: + DISPLAYLEVEL(6, "compression level adaptation check \n") + break; + case FIO_RUST_ZSTD_ADAPT_DIAG_INPUT_STARVATION: + DISPLAYLEVEL(6, "input is never blocked => input is slower than ingestion \n"); + break; + case FIO_RUST_ZSTD_ADAPT_DIAG_INPUT_STATS: + assert(projection != NULL); + assert(projection->inputPresented > 0); + DISPLAYLEVEL(6, "input blocked %u/%u(%.2f) - ingested:%u vs %u:consumed - flushed:%u vs %u:produced \n", + projection->inputBlocked, projection->inputPresented, + (double)projection->inputBlocked/ + projection->inputPresented*100, + (unsigned)projection->newlyIngested, + (unsigned)projection->newlyConsumed, + (unsigned)projection->newlyFlushed, + (unsigned)projection->newlyProduced); + break; + case FIO_RUST_ZSTD_ADAPT_DIAG_RECOMMEND_FASTER: + assert(projection != NULL); + DISPLAYLEVEL(6, "recommend faster as in(%llu) >= (%llu)comp(%llu) <= out(%llu) \n", + (unsigned long long)projection->newlyIngested, + (unsigned long long)projection->newlyConsumed, + (unsigned long long)projection->newlyProduced, + (unsigned long long)projection->newlyFlushed); + break; + case FIO_RUST_ZSTD_ADAPT_DIAG_SLOWER_LEVEL: + DISPLAYLEVEL(6, "slower speed , higher compression \n") + break; + case FIO_RUST_ZSTD_ADAPT_DIAG_FASTER_LEVEL: + DISPLAYLEVEL(6, "faster speed , lighter compression \n") + break; + default: + assert(0); + break; + } +} + void FIO_rust_zstd_compressStreamDisplay( int directive, size_t inputPos, size_t inputSize, size_t outputProduced) { @@ -2769,136 +2897,10 @@ static void FIO_rust_zstd_iteration(void* opaque, const char* srcFileName, { FIO_rust_zstd_projection_context_t* const context = (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; - - /* 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); - - context->lastAdaptTime = UTIL_getTime(); - - /* check output speed */ - if (zfp.currentJobID > 1) { /* only possible if nbWorkers >= 1 */ - unsigned long long const newlyProduced = - zfp.produced - context->previousZfpUpdate.produced; - unsigned long long const newlyFlushed = - zfp.flushed - context->previousZfpUpdate.flushed; - assert(zfp.produced >= context->previousZfpUpdate.produced); - assert(prefs->nbWorkers >= 1); - - /* test if compression is blocked - * 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 */ - 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; - - 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; - } - context->flushWaiting = 0; - } - - /* course correct only if there is at least one new job completed */ - if (zfp.currentJobID > context->lastJobID) { - DISPLAYLEVEL(6, "compression level adaptation check \n") - - /* check input speed */ - if (zfp.currentJobID > (unsigned)(prefs->nbWorkers+1)) { - 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) { - unsigned long long const newlyIngested = - zfp.ingested - context->previousZfpCorrection.ingested; - unsigned long long const newlyConsumed = - zfp.consumed - context->previousZfpCorrection.consumed; - unsigned long long const newlyProduced = - zfp.produced - context->previousZfpCorrection.produced; - unsigned long long const newlyFlushed = - zfp.flushed - context->previousZfpCorrection.flushed; - context->previousZfpCorrection = zfp; - assert(context->inputPresented > 0); - DISPLAYLEVEL(6, "input blocked %u/%u(%.2f) - ingested:%u vs %u:consumed - flushed:%u vs %u:produced \n", - context->inputBlocked, context->inputPresented, - (double)context->inputBlocked/context->inputPresented*100, - (unsigned)newlyIngested, (unsigned)newlyConsumed, - (unsigned)newlyFlushed, (unsigned)newlyProduced); - 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); - context->speedChange = FIO_rust_zstd_faster; - } - } - context->inputBlocked = 0; - context->inputPresented = 0; - } - - if (context->speedChange == FIO_rust_zstd_slower) { - DISPLAYLEVEL(6, "slower speed , higher compression \n") - 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") - 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); - } - context->speedChange = FIO_rust_zstd_noChange; - context->lastJobID = zfp.currentJobID; - } - } + (void)oldInputPos; + (void)newInputPos; + (void)toFlushNow; /* Keep progress formatting and the frame-progression query in C. */ if (SHOULD_DISPLAY_PROGRESS() && READY_FOR_UPDATE()) { @@ -2964,7 +2966,6 @@ FIO_rust_compressZstdCallback(void* fCtx, void* prefs, void* ress, memset(&context, 0, sizeof(context)); context.fCtx = fCtxPtr; - context.prefs = prefsPtr; context.cctx = ressPtr->cctx; context.lastAdaptTime = UTIL_getTime(); context.srcFileSize = srcFileSize; @@ -2985,6 +2986,15 @@ FIO_rust_compressZstdCallback(void* fCtx, void* prefs, void* ress, projection.compressStream = FIO_rust_zstd_compressStream; projection.iteration = FIO_rust_zstd_iteration; projection.compressStreamDisplay = FIO_rust_zstd_compressStreamDisplay; + projection.adaptiveMode = prefsPtr->adaptiveMode; + projection.nbWorkers = prefsPtr->nbWorkers; + 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; DISPLAYLEVEL(6, "compression using zstd format \n"); diff --git a/rust/src/fileio_asyncio.rs b/rust/src/fileio_asyncio.rs index 94b8899a3..cd4a7477f 100644 --- a/rust/src/fileio_asyncio.rs +++ b/rust/src/fileio_asyncio.rs @@ -948,6 +948,15 @@ 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; +pub const FIO_RUST_ZSTD_ADAPT_DIAG_OUTPUT_BLOCKED: c_int = 0; +pub const FIO_RUST_ZSTD_ADAPT_DIAG_OUTPUT_BACKLOG: c_int = 1; +pub const FIO_RUST_ZSTD_ADAPT_DIAG_CHECK: c_int = 2; +pub const FIO_RUST_ZSTD_ADAPT_DIAG_INPUT_STARVATION: c_int = 3; +pub const FIO_RUST_ZSTD_ADAPT_DIAG_INPUT_STATS: c_int = 4; +pub const FIO_RUST_ZSTD_ADAPT_DIAG_RECOMMEND_FASTER: c_int = 5; +pub const FIO_RUST_ZSTD_ADAPT_DIAG_SLOWER_LEVEL: c_int = 6; +pub const FIO_RUST_ZSTD_ADAPT_DIAG_FASTER_LEVEL: c_int = 7; + /// 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 @@ -1030,9 +1039,44 @@ const _: () = { ); }; -/// 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. +/// Scalar frame progression returned by C's progression callback. The public +/// C structure itself never crosses this boundary. +#[repr(C)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct FIO_rust_zstd_progression_t { + pub ingested: u64, + pub consumed: u64, + pub produced: u64, + pub flushed: u64, + pub current_job_id: c_uint, + pub nb_active_workers: c_uint, +} +const _: () = { + assert!(std::mem::offset_of!(FIO_rust_zstd_progression_t, ingested) == 0); + assert!(std::mem::offset_of!(FIO_rust_zstd_progression_t, consumed) == size_of::()); + assert!(std::mem::offset_of!(FIO_rust_zstd_progression_t, produced) == 2 * size_of::()); + assert!(std::mem::offset_of!(FIO_rust_zstd_progression_t, flushed) == 3 * size_of::()); + assert!( + std::mem::offset_of!(FIO_rust_zstd_progression_t, current_job_id) == 4 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_progression_t, nb_active_workers) + == 4 * size_of::() + size_of::() + ); + assert!( + size_of::() == 4 * size_of::() + 2 * size_of::() + ); +}; + +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); +pub type FIO_rust_zstd_adaptive_diagnostic_fn = + unsafe extern "C" fn(*mut c_void, c_int, *const FIO_rust_zstd_adapt_projection_t); + +/// Rust owns the scalar adaptive state, progression ordering, and level +/// decision. C retains diagnostics and all codec/private state in callbacks. #[repr(C)] pub struct FIO_rust_zstd_compress_projection_t { pub read_opaque: *mut c_void, @@ -1049,7 +1093,46 @@ pub struct FIO_rust_zstd_compress_projection_t { pub compress_stream: Option, pub iteration: Option, pub compress_display: Option, + pub adaptive_mode: c_int, + pub nb_workers: c_int, + 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, } +const _: () = { + let callback_offset = (14 * size_of::() + 5 * size_of::() + size_of::() + - 1) + / size_of::() + * size_of::(); + assert!( + std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, adaptive_mode) + == 14 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, nb_workers) + == 14 * size_of::() + size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, max_c_level) + == 14 * size_of::() + 4 * size_of::() + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, adaptive_refresh) + == callback_offset + ); + assert!( + std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, adaptive_diagnostic) + == callback_offset + 3 * size_of::() + ); + assert!( + size_of::() + == callback_offset + 4 * size_of::() + ); +}; pub const FIO_RUST_GZIP_OK: c_int = 0; pub const FIO_RUST_GZIP_INIT_ERROR: c_int = 1; @@ -3440,6 +3523,214 @@ pub unsafe extern "C" fn FIO_rust_zstd_adapt( zstd_adapt_policy(policy, projection) } +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +struct ZstdAdaptiveState { + previous_update: FIO_rust_zstd_progression_t, + previous_correction: FIO_rust_zstd_progression_t, + speed_change: c_int, + flush_waiting: c_uint, + input_presented: c_uint, + input_blocked: c_uint, + last_job_id: c_uint, +} + +unsafe fn zstd_adaptive_report( + projection: &FIO_rust_zstd_compress_projection_t, + diagnostic: c_int, + adapt_projection: Option<&FIO_rust_zstd_adapt_projection_t>, +) { + let Some(report) = projection.adaptive_diagnostic else { + return; + }; + let projection_ptr = adapt_projection.map_or(ptr::null(), |value| value as *const _); + unsafe { + report(projection.policy_opaque, diagnostic, projection_ptr); + } +} + +unsafe fn zstd_adaptive_iteration( + state: &mut ZstdAdaptiveState, + projection: &FIO_rust_zstd_compress_projection_t, + compression_level: &mut c_int, + old_input_pos: usize, + new_input_pos: usize, + to_flush_now: usize, +) { + state.input_presented = state.input_presented.wrapping_add(1); + if old_input_pos == new_input_pos { + state.input_blocked = state.input_blocked.wrapping_add(1); + } + if to_flush_now == 0 { + state.flush_waiting = 1; + } + + if projection.adaptive_mode == 0 { + return; + } + let Some(refresh) = projection.adaptive_refresh else { + return; + }; + if unsafe { refresh(projection.policy_opaque) } == 0 { + return; + } + let Some(get_progression) = projection.adaptive_progression else { + return; + }; + let mut progression = FIO_rust_zstd_progression_t::default(); + unsafe { + get_progression(projection.policy_opaque, &mut progression); + } + + if progression.current_job_id > 1 { + let newly_produced = progression + .produced + .wrapping_sub(state.previous_update.produced); + let newly_flushed = progression + .flushed + .wrapping_sub(state.previous_update.flushed); + assert!(progression.produced >= state.previous_update.produced); + assert!(projection.nb_workers >= 1); + + let mut adapt_projection = FIO_rust_zstd_adapt_projection_t { + consumed: progression.consumed, + previous_consumed: state.previous_update.consumed, + nb_active_workers: progression.nb_active_workers, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + if zstd_adapt_policy(FIO_RUST_ZSTD_ADAPT_OUTPUT_BLOCKED, &adapt_projection) + == FIO_RUST_ZSTD_ADAPT_SLOWER + { + unsafe { + zstd_adaptive_report(projection, FIO_RUST_ZSTD_ADAPT_DIAG_OUTPUT_BLOCKED, None); + } + state.speed_change = FIO_RUST_ZSTD_ADAPT_SLOWER; + } + + state.previous_update = progression; + + adapt_projection.newly_produced = newly_produced; + adapt_projection.newly_flushed = newly_flushed; + adapt_projection.flush_waiting = state.flush_waiting; + if zstd_adapt_policy(FIO_RUST_ZSTD_ADAPT_OUTPUT_BACKLOG, &adapt_projection) + == FIO_RUST_ZSTD_ADAPT_SLOWER + { + unsafe { + zstd_adaptive_report( + projection, + FIO_RUST_ZSTD_ADAPT_DIAG_OUTPUT_BACKLOG, + Some(&adapt_projection), + ); + } + state.speed_change = FIO_RUST_ZSTD_ADAPT_SLOWER; + } + state.flush_waiting = 0; + } + + if progression.current_job_id > state.last_job_id { + unsafe { + zstd_adaptive_report(projection, FIO_RUST_ZSTD_ADAPT_DIAG_CHECK, None); + } + + if progression.current_job_id > projection.nb_workers.wrapping_add(1) as c_uint { + let mut adapt_projection = FIO_rust_zstd_adapt_projection_t { + input_blocked: state.input_blocked, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + if zstd_adapt_policy(FIO_RUST_ZSTD_ADAPT_INPUT_STARVATION, &adapt_projection) + == FIO_RUST_ZSTD_ADAPT_SLOWER + { + unsafe { + zstd_adaptive_report( + projection, + FIO_RUST_ZSTD_ADAPT_DIAG_INPUT_STARVATION, + None, + ); + } + state.speed_change = FIO_RUST_ZSTD_ADAPT_SLOWER; + } else if state.speed_change == FIO_RUST_ZSTD_ADAPT_NO_CHANGE { + let newly_ingested = progression + .ingested + .wrapping_sub(state.previous_correction.ingested); + let newly_consumed = progression + .consumed + .wrapping_sub(state.previous_correction.consumed); + let newly_produced = progression + .produced + .wrapping_sub(state.previous_correction.produced); + let newly_flushed = progression + .flushed + .wrapping_sub(state.previous_correction.flushed); + state.previous_correction = progression; + adapt_projection.input_blocked = state.input_blocked; + adapt_projection.input_presented = state.input_presented; + adapt_projection.newly_ingested = newly_ingested; + adapt_projection.newly_consumed = newly_consumed; + adapt_projection.newly_produced = newly_produced; + adapt_projection.newly_flushed = newly_flushed; + unsafe { + zstd_adaptive_report( + projection, + FIO_RUST_ZSTD_ADAPT_DIAG_INPUT_STATS, + Some(&adapt_projection), + ); + } + if zstd_adapt_policy(FIO_RUST_ZSTD_ADAPT_BLOCKED_INPUT, &adapt_projection) + == FIO_RUST_ZSTD_ADAPT_FASTER + { + unsafe { + zstd_adaptive_report( + projection, + FIO_RUST_ZSTD_ADAPT_DIAG_RECOMMEND_FASTER, + Some(&adapt_projection), + ); + } + state.speed_change = FIO_RUST_ZSTD_ADAPT_FASTER; + } + } + state.input_blocked = 0; + state.input_presented = 0; + } + + if state.speed_change == FIO_RUST_ZSTD_ADAPT_SLOWER { + unsafe { + zstd_adaptive_report(projection, FIO_RUST_ZSTD_ADAPT_DIAG_SLOWER_LEVEL, None); + } + let adapt_level_projection = FIO_rust_zstd_adapt_projection_t { + compression_level: *compression_level, + max_adapt_level: projection.max_adapt_level, + max_c_level: projection.max_c_level, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + *compression_level = + zstd_adapt_policy(FIO_RUST_ZSTD_ADAPT_LEVEL_SLOWER, &adapt_level_projection); + if let Some(set_parameter) = projection.adaptive_set_parameter { + unsafe { + set_parameter(projection.policy_opaque, *compression_level); + } + } + } + if state.speed_change == FIO_RUST_ZSTD_ADAPT_FASTER { + unsafe { + zstd_adaptive_report(projection, FIO_RUST_ZSTD_ADAPT_DIAG_FASTER_LEVEL, None); + } + let adapt_level_projection = FIO_rust_zstd_adapt_projection_t { + compression_level: *compression_level, + min_adapt_level: projection.min_adapt_level, + ..FIO_rust_zstd_adapt_projection_t::default() + }; + *compression_level = + zstd_adapt_policy(FIO_RUST_ZSTD_ADAPT_LEVEL_FASTER, &adapt_level_projection); + if let Some(set_parameter) = projection.adaptive_set_parameter { + unsafe { + set_parameter(projection.policy_opaque, *compression_level); + } + } + } + state.speed_change = FIO_RUST_ZSTD_ADAPT_NO_CHANGE; + state.last_job_id = progression.current_job_id; + } +} + type FIO_rust_zstd_to_flush_now_fn = unsafe extern "C" fn(*mut c_void) -> usize; type FIO_rust_zstd_compress_stream2_fn = unsafe extern "C" fn(*mut c_void, *mut ZSTD_outBuffer, *mut ZSTD_inBuffer, c_int) -> usize; @@ -3582,6 +3873,14 @@ pub unsafe extern "C" fn FIO_rust_compressZstdFrame( let Some(iteration) = projection.iteration else { return FIO_RUST_ZSTD_INVALID_PROJECTION; }; + if projection.adaptive_mode != 0 + && (projection.adaptive_refresh.is_none() + || projection.adaptive_progression.is_none() + || projection.adaptive_set_parameter.is_none() + || projection.adaptive_diagnostic.is_none()) + { + return FIO_RUST_ZSTD_INVALID_PROJECTION; + } let mut job = ptr::null_mut::(); let mut output = ptr::null_mut::(); @@ -3605,6 +3904,7 @@ 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(); loop { if input_pos == input_size { @@ -3692,6 +3992,14 @@ pub unsafe extern "C" fn FIO_rust_compressZstdFrame( } unsafe { + zstd_adaptive_iteration( + &mut adaptive_state, + projection, + &mut compression_level, + old_input_pos, + new_input_pos, + to_flush_now, + ); iteration( projection.policy_opaque, src_file_name, @@ -7338,6 +7646,114 @@ 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, + ) { + let state = unsafe { &*opaque.cast::() }; + unsafe { + *progression = state.progression; + } + } + + unsafe extern "C" fn adaptive_test_set_parameter(opaque: *mut c_void, level: c_int) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(100 + level); + } + + unsafe extern "C" fn adaptive_test_diagnostic( + opaque: *mut c_void, + diagnostic: c_int, + _projection: *const FIO_rust_zstd_adapt_projection_t, + ) { + let state = unsafe { &mut *opaque.cast::() }; + state.events.push(diagnostic); + } + + fn adaptive_test_projection( + state: &mut AdaptiveCallbackState, + ) -> FIO_rust_zstd_compress_projection_t { + let opaque = (state as *mut AdaptiveCallbackState).cast::(); + FIO_rust_zstd_compress_projection_t { + read_opaque: ptr::null_mut(), + write_opaque: ptr::null_mut(), + codec_opaque: ptr::null_mut(), + policy_opaque: opaque, + read_buffer_size: 1, + read_fill: None, + read_consume: None, + write_acquire: None, + write_enqueue: None, + write_release: None, + sparse_write_end: None, + compress_stream: None, + iteration: None, + compress_display: None, + adaptive_mode: 1, + nb_workers: 1, + 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), + } + } + + #[test] + fn zstd_adaptive_iteration_preserves_callback_order_and_resets_state() { + let mut callbacks = AdaptiveCallbackState { + progression: FIO_rust_zstd_progression_t { + produced: 10, + flushed: 8, + current_job_id: 3, + ..FIO_rust_zstd_progression_t::default() + }, + ..AdaptiveCallbackState::default() + }; + let projection = adaptive_test_projection(&mut callbacks); + let mut state = ZstdAdaptiveState::default(); + let mut compression_level = 3; + + unsafe { + zstd_adaptive_iteration(&mut state, &projection, &mut compression_level, 0, 1, 1); + } + + assert_eq!( + callbacks.events, + vec![ + FIO_RUST_ZSTD_ADAPT_DIAG_OUTPUT_BLOCKED, + FIO_RUST_ZSTD_ADAPT_DIAG_OUTPUT_BACKLOG, + FIO_RUST_ZSTD_ADAPT_DIAG_CHECK, + FIO_RUST_ZSTD_ADAPT_DIAG_INPUT_STARVATION, + FIO_RUST_ZSTD_ADAPT_DIAG_SLOWER_LEVEL, + 104, + ] + ); + assert_eq!(callbacks.refresh_calls, 1); + assert_eq!(compression_level, 4); + assert_eq!(state.previous_update, callbacks.progression); + assert_eq!(state.last_job_id, 3); + assert_eq!(state.speed_change, FIO_RUST_ZSTD_ADAPT_NO_CHANGE); + assert_eq!(state.input_blocked, 0); + assert_eq!(state.input_presented, 0); + assert_eq!(state.flush_waiting, 0); + } + unsafe extern "C" fn zstd_test_read_fill( opaque: *mut c_void, requested: usize, @@ -7487,6 +7903,15 @@ mod tests { compress_stream: Some(zstd_test_compress), iteration: Some(zstd_test_iteration), compress_display: None, + adaptive_mode: 0, + nb_workers: 0, + 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, } }