feat(cli): move adaptive decisions into Rust
Move the scalar runtime --adapt predicates and compression-level normalization into Rust. Rust now decides output blocking, output backlog, input starvation, blocked-input speedups, and bounded slower/faster level changes through an explicit projection. Keep the CLI iteration order, refresh clock, frame-progression snapshots, private FIO/ZSTD contexts, diagnostics, and ZSTD_CCtx_setParameter() mutation in C. The new C/Rust layout assertions make the scalar policy ABI explicit without exposing those private layouts. Test Plan: - focused adaptive-policy tests: 6 passed - cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1: 533 passed - cargo test --manifest-path rust/cli/Cargo.toml --all-targets -- --test-threads=1: 169 passed - legacy compression/decompression/dictionary-builder feature matrix: 588 passed - six library/CLI clippy gates with -D warnings - capped serial native library/program rebuilds, 41 CLI tests, Rust library smoke, and full test-zstd round trips including --adapt cases - capped serial stress gates: 278 fuzzer cases, 84+129+143 zstream cases, and 1,601 decode-corpus cases - every heavyweight command used CARGO_BUILD_JOBS=1 or make -j1 and ulimit -v 41943040; no worker/native process remained afterward Commit is intentionally unsigned because GPG pinentry hangs in this non-interactive environment.
This commit is contained in:
+103
-20
@@ -1968,6 +1968,68 @@ typedef enum {
|
|||||||
FIO_rust_zstd_faster
|
FIO_rust_zstd_faster
|
||||||
} FIO_rust_zstd_speed_change_e;
|
} 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 {
|
typedef struct {
|
||||||
FIO_ctx_t* fCtx;
|
FIO_ctx_t* fCtx;
|
||||||
FIO_prefs_t* prefs;
|
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_rust_zstd_projection_context_t*)opaque;
|
||||||
FIO_prefs_t* const prefs = context->prefs;
|
FIO_prefs_t* const prefs = context->prefs;
|
||||||
FIO_ctx_t* const fCtx = context->fCtx;
|
FIO_ctx_t* const fCtx = context->fCtx;
|
||||||
|
FIO_rust_zstd_adapt_projection_t adaptProjection;
|
||||||
|
|
||||||
context->inputPresented++;
|
context->inputPresented++;
|
||||||
if (oldInputPos == newInputPos) context->inputBlocked++;
|
if (oldInputPos == newInputPos) context->inputBlocked++;
|
||||||
if (!toFlushNow) context->flushWaiting = 1;
|
if (!toFlushNow) context->flushWaiting = 1;
|
||||||
|
|
||||||
/* Adaptive mode remains in C so its policy and private preference/context
|
/* C retains the clock gate, progression snapshots, diagnostics, and
|
||||||
* access stay identical while Rust owns the surrounding stream loop. */
|
* private context mutations; Rust supplies only scalar policy decisions. */
|
||||||
if (prefs->adaptiveMode &&
|
if (prefs->adaptiveMode &&
|
||||||
UTIL_clockSpanMicro(context->lastAdaptTime) > REFRESH_RATE) {
|
UTIL_clockSpanMicro(context->lastAdaptTime) > REFRESH_RATE) {
|
||||||
ZSTD_frameProgression const zfp = ZSTD_getFrameProgression(context->cctx);
|
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
|
* 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.
|
* 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 */
|
* note : exclude starting part, since currentJobID > 1 */
|
||||||
if ((zfp.consumed == context->previousZfpUpdate.consumed)
|
memset(&adaptProjection, 0, sizeof(adaptProjection));
|
||||||
&& (zfp.nbActiveWorkers == 0)) {
|
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")
|
DISPLAYLEVEL(6, "all buffers full : compression stopped => slow down \n")
|
||||||
context->speedChange = FIO_rust_zstd_slower;
|
context->speedChange = FIO_rust_zstd_slower;
|
||||||
}
|
}
|
||||||
|
|
||||||
context->previousZfpUpdate = zfp;
|
context->previousZfpUpdate = zfp;
|
||||||
|
|
||||||
if ((newlyProduced > (newlyFlushed * 9 / 8))
|
adaptProjection.newlyProduced = newlyProduced;
|
||||||
&& (context->flushWaiting == 0)) {
|
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",
|
DISPLAYLEVEL(6, "compression faster than flush (%llu > %llu), and flushed was never slowed down by lack of production => slow down \n",
|
||||||
newlyProduced, newlyFlushed);
|
newlyProduced, newlyFlushed);
|
||||||
context->speedChange = FIO_rust_zstd_slower;
|
context->speedChange = FIO_rust_zstd_slower;
|
||||||
@@ -2119,7 +2191,11 @@ static void FIO_rust_zstd_iteration(void* opaque, const char* srcFileName,
|
|||||||
|
|
||||||
/* check input speed */
|
/* check input speed */
|
||||||
if (zfp.currentJobID > (unsigned)(prefs->nbWorkers+1)) {
|
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");
|
DISPLAYLEVEL(6, "input is never blocked => input is slower than ingestion \n");
|
||||||
context->speedChange = FIO_rust_zstd_slower;
|
context->speedChange = FIO_rust_zstd_slower;
|
||||||
} else if (context->speedChange == FIO_rust_zstd_noChange) {
|
} 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,
|
(double)context->inputBlocked/context->inputPresented*100,
|
||||||
(unsigned)newlyIngested, (unsigned)newlyConsumed,
|
(unsigned)newlyIngested, (unsigned)newlyConsumed,
|
||||||
(unsigned)newlyFlushed, (unsigned)newlyProduced);
|
(unsigned)newlyFlushed, (unsigned)newlyProduced);
|
||||||
if ((context->inputBlocked > context->inputPresented / 8)
|
adaptProjection.inputBlocked = context->inputBlocked;
|
||||||
&& (newlyFlushed * 33 / 32 > newlyProduced)
|
adaptProjection.inputPresented = context->inputPresented;
|
||||||
&& (newlyIngested * 33 / 32 > newlyConsumed)) {
|
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",
|
DISPLAYLEVEL(6, "recommend faster as in(%llu) >= (%llu)comp(%llu) <= out(%llu) \n",
|
||||||
newlyIngested, newlyConsumed,
|
newlyIngested, newlyConsumed,
|
||||||
newlyProduced, newlyFlushed);
|
newlyProduced, newlyFlushed);
|
||||||
@@ -2153,22 +2235,23 @@ static void FIO_rust_zstd_iteration(void* opaque, const char* srcFileName,
|
|||||||
|
|
||||||
if (context->speedChange == FIO_rust_zstd_slower) {
|
if (context->speedChange == FIO_rust_zstd_slower) {
|
||||||
DISPLAYLEVEL(6, "slower speed , higher compression \n")
|
DISPLAYLEVEL(6, "slower speed , higher compression \n")
|
||||||
(*compressionLevel)++;
|
memset(&adaptProjection, 0, sizeof(adaptProjection));
|
||||||
if (*compressionLevel > ZSTD_maxCLevel())
|
adaptProjection.compressionLevel = *compressionLevel;
|
||||||
*compressionLevel = ZSTD_maxCLevel();
|
adaptProjection.maxAdaptLevel = prefs->maxAdaptLevel;
|
||||||
if (*compressionLevel > prefs->maxAdaptLevel)
|
adaptProjection.maxCLevel = ZSTD_maxCLevel();
|
||||||
*compressionLevel = prefs->maxAdaptLevel;
|
*compressionLevel = FIO_rust_zstd_adapt(
|
||||||
*compressionLevel += (*compressionLevel == 0);
|
FIO_RUST_ZSTD_ADAPT_LEVEL_SLOWER, &adaptProjection);
|
||||||
ZSTD_CCtx_setParameter(context->cctx,
|
ZSTD_CCtx_setParameter(context->cctx,
|
||||||
ZSTD_c_compressionLevel,
|
ZSTD_c_compressionLevel,
|
||||||
*compressionLevel);
|
*compressionLevel);
|
||||||
}
|
}
|
||||||
if (context->speedChange == FIO_rust_zstd_faster) {
|
if (context->speedChange == FIO_rust_zstd_faster) {
|
||||||
DISPLAYLEVEL(6, "faster speed , lighter compression \n")
|
DISPLAYLEVEL(6, "faster speed , lighter compression \n")
|
||||||
(*compressionLevel)--;
|
memset(&adaptProjection, 0, sizeof(adaptProjection));
|
||||||
if (*compressionLevel < prefs->minAdaptLevel)
|
adaptProjection.compressionLevel = *compressionLevel;
|
||||||
*compressionLevel = prefs->minAdaptLevel;
|
adaptProjection.minAdaptLevel = prefs->minAdaptLevel;
|
||||||
*compressionLevel -= (*compressionLevel == 0);
|
*compressionLevel = FIO_rust_zstd_adapt(
|
||||||
|
FIO_RUST_ZSTD_ADAPT_LEVEL_FASTER, &adaptProjection);
|
||||||
ZSTD_CCtx_setParameter(context->cctx,
|
ZSTD_CCtx_setParameter(context->cctx,
|
||||||
ZSTD_c_compressionLevel,
|
ZSTD_c_compressionLevel,
|
||||||
*compressionLevel);
|
*compressionLevel);
|
||||||
|
|||||||
+308
-3
@@ -294,8 +294,103 @@ pub type FIO_rust_zstd_compress_stream_fn = unsafe extern "C" fn(
|
|||||||
pub type FIO_rust_zstd_iteration_fn =
|
pub type FIO_rust_zstd_iteration_fn =
|
||||||
unsafe extern "C" fn(*mut c_void, *const c_char, *mut c_int, usize, usize, usize);
|
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,
|
pub const FIO_RUST_ZSTD_ADAPT_NO_CHANGE: c_int = 0;
|
||||||
/// adaptive policy, diagnostics, and codec calls behind opaque callbacks.
|
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::<u64>()
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, nb_active_workers)
|
||||||
|
== 2 * size_of::<u64>()
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, newly_produced)
|
||||||
|
== if size_of::<usize>() == 8 { 24 } else { 20 }
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, newly_flushed)
|
||||||
|
== if size_of::<usize>() == 8 { 32 } else { 28 }
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, flush_waiting)
|
||||||
|
== if size_of::<usize>() == 8 { 40 } else { 36 }
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, input_blocked)
|
||||||
|
== if size_of::<usize>() == 8 { 44 } else { 40 }
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, input_presented)
|
||||||
|
== if size_of::<usize>() == 8 { 48 } else { 44 }
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, newly_ingested)
|
||||||
|
== if size_of::<usize>() == 8 { 56 } else { 48 }
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, newly_consumed)
|
||||||
|
== if size_of::<usize>() == 8 { 64 } else { 56 }
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, compression_level)
|
||||||
|
== if size_of::<usize>() == 8 { 72 } else { 64 }
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, min_adapt_level)
|
||||||
|
== if size_of::<usize>() == 8 { 76 } else { 68 }
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, max_adapt_level)
|
||||||
|
== if size_of::<usize>() == 8 { 80 } else { 72 }
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
std::mem::offset_of!(FIO_rust_zstd_adapt_projection_t, max_c_level)
|
||||||
|
== if size_of::<usize>() == 8 { 84 } else { 76 }
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
size_of::<FIO_rust_zstd_adapt_projection_t>()
|
||||||
|
== if size_of::<usize>() == 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)]
|
#[repr(C)]
|
||||||
pub struct FIO_rust_zstd_compress_projection_t {
|
pub struct FIO_rust_zstd_compress_projection_t {
|
||||||
pub read_opaque: *mut c_void,
|
pub read_opaque: *mut c_void,
|
||||||
@@ -2027,9 +2122,88 @@ pub unsafe extern "C" fn FIO_rust_compressFilenameInternal(
|
|||||||
FIO_RUST_COMPRESS_OK
|
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
|
/// Compresses one zstd frame through the C-owned zstd context and adaptive
|
||||||
/// policy. Rust owns the stream loop and exact pool accounting; C callbacks
|
/// 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]
|
#[no_mangle]
|
||||||
pub unsafe extern "C" fn FIO_rust_compressZstdFrame(
|
pub unsafe extern "C" fn FIO_rust_compressZstdFrame(
|
||||||
projection: *const FIO_rust_zstd_compress_projection_t,
|
projection: *const FIO_rust_zstd_compress_projection_t,
|
||||||
@@ -3810,6 +3984,137 @@ impl PoolInner {
|
|||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
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 {
|
fn test_prefs(async_io: c_int) -> FIO_prefs_t {
|
||||||
let mut prefs: FIO_prefs_t = unsafe { std::mem::zeroed() };
|
let mut prefs: FIO_prefs_t = unsafe { std::mem::zeroed() };
|
||||||
prefs.asyncIO = async_io;
|
prefs.asyncIO = async_io;
|
||||||
|
|||||||
Reference in New Issue
Block a user