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.
2026-07-19 10:12:50 +02:00
parent 3b989ab3ee
commit b048937992
2 changed files with 411 additions and 23 deletions
@@ -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::<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)]
 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;