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

refactor(cli): move adaptive feedback policy to Rust

Browse Source 
The zstd file-I/O loop still kept adaptive compression's state machine in C: progression deltas, refresh gating, job-completion checks, input counters, speed decisions, and level updates were interleaved with private FIO and ZSTD state. That left orchestration policy behind the existing scalar Rust predicates.

Move the adaptive state and callback order into Rust. C now supplies scalar progression snapshots, the clock gate, exact diagnostics, and the ignored CCtx parameter setter through callbacks; private FIO_prefs_t, ZSTD_CCtx, ZSTD_frameProgression, clocks, and progress formatting remain C-owned. Preserve the previous-progression publication before backlog evaluation, input counter reset points, and serial/MT level-clamp behavior.

Test Plan:

- cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check

- git diff --check

- ulimit -v 41943040; CARGO_BUILD_JOBS=1 make -j1
This commit is contained in:
ddidderr
2026-07-21 11:32:06 +02:00
parent 1a3971ff10
commit 1554c5aacb
2 changed files with 594 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
programs/fileio.c
+166 -156
View File
@@ -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");
rust/src/fileio_asyncio.rs
+428 -3
View File
@@ -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::<u64>());
assert!(std::mem::offset_of!(FIO_rust_zstd_progression_t, produced) == 2 * size_of::<u64>());
assert!(std::mem::offset_of!(FIO_rust_zstd_progression_t, flushed) == 3 * size_of::<u64>());
assert!(
std::mem::offset_of!(FIO_rust_zstd_progression_t, current_job_id) == 4 * size_of::<u64>()
);
assert!(
std::mem::offset_of!(FIO_rust_zstd_progression_t, nb_active_workers)
== 4 * size_of::<u64>() + size_of::<c_uint>()
);
assert!(
size_of::<FIO_rust_zstd_progression_t>() == 4 * size_of::<u64>() + 2 * size_of::<c_uint>()
);
};
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<FIO_rust_zstd_compress_stream_fn>,
pub iteration: Option<FIO_rust_zstd_iteration_fn>,
pub compress_display: Option<FIO_rust_zstd_compress_display_fn>,
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<FIO_rust_zstd_adaptive_refresh_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_diagnostic: Option<FIO_rust_zstd_adaptive_diagnostic_fn>,
}
const _: () = {
let callback_offset = (14 * size_of::<usize>() + 5 * size_of::<c_int>() + size_of::<usize>()
- 1)
/ size_of::<usize>()
* size_of::<usize>();
assert!(
std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, adaptive_mode)
== 14 * size_of::<usize>()
);
assert!(
std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, nb_workers)
== 14 * size_of::<usize>() + size_of::<c_int>()
);
assert!(
std::mem::offset_of!(FIO_rust_zstd_compress_projection_t, max_c_level)
== 14 * size_of::<usize>() + 4 * size_of::<c_int>()
);
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::<usize>()
);
assert!(
size_of::<FIO_rust_zstd_compress_projection_t>()
== callback_offset + 4 * size_of::<usize>()
);
};
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::<c_void>();
let mut output = ptr::null_mut::<u8>();
@@ -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<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(
opaque: *mut c_void,
progression: *mut FIO_rust_zstd_progression_t,
) {
let state = unsafe { &*opaque.cast::<AdaptiveCallbackState>() };
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::<AdaptiveCallbackState>() };
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::<AdaptiveCallbackState>() };
state.events.push(diagnostic);
}
fn adaptive_test_projection(
state: &mut AdaptiveCallbackState,
) -> FIO_rust_zstd_compress_projection_t {
let opaque = (state as *mut AdaptiveCallbackState).cast::<c_void>();
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,
}
}