feat(cli): move zstd compression loop into Rust

The CLI's zstd-format compression path previously kept the asynchronous read, compressStream2, output-job, flush, accounting, adaptive-policy, and progress loop together in fileio.c. That made the high-level stream orchestration another large C-owned surface even though the Rust fileio module already owned the neighboring format loops. Move the format-independent zstd stream loop and read/output accounting into FIO_rust_compressZstdFrame. The C adapter now projects read-pool, write-pool, codec, and policy operations through narrow callbacks. C retains the private ZSTD_CCtx interaction, adaptive-level policy, memory diagnostics, progress formatting, and CLI error mapping, so no private C layout crosses into Rust. Focused seam tests cover input/output ordering, final flush behavior, codec error propagation, and incomplete known-size input handling. Test Plan: - `cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1` -- 513 passed under the 40 GiB virtual-memory cap. - `cargo test --manifest-path rust/cli/Cargo.toml --all-targets -- --test-threads=1` -- 169 passed under the cap. - `cargo clippy` lib/benches/tests for `rust` and `rust/cli`, with `-D warnings`, and nightly formatting -- passed. - `make -B -C lib -j1 lib` and `make -B -C programs -j1 zstd` -- passed. - Native CLI, full zstd, fuzzer, zstream, and decode-corpus targets -- passed serially under the cap. GPG signing was attempted but unavailable because no pinentry process was available; this repository's preceding commits are unsigned, so this commit uses the explicit unsigned fallback.
2026-07-19 08:48:50 +02:00
parent 747559e619
commit 423854541b
2 changed files with 913 additions and 221 deletions
@@ -259,6 +259,60 @@ pub struct FIO_rust_compress_callbacks_t {
    pub display_status: Option<FIO_rust_compress_status_display_fn>,
 }

+pub const FIO_RUST_ZSTD_OK: c_int = 0;
+pub const FIO_RUST_ZSTD_COMPRESS_ERROR: c_int = 1;
+pub const FIO_RUST_ZSTD_INCOMPLETE_INPUT: c_int = 2;
+pub const FIO_RUST_ZSTD_INVALID_PROJECTION: c_int = 3;
+
+const FIO_RUST_ZSTD_E_CONTINUE: c_int = 0;
+const FIO_RUST_ZSTD_E_END: c_int = 2;
+const UTIL_FILESIZE_UNKNOWN: u64 = u64::MAX;
+
+pub type FIO_rust_zstd_read_fill_fn =
+    unsafe extern "C" fn(*mut c_void, usize, *mut *const u8, *mut usize) -> usize;
+pub type FIO_rust_zstd_read_consume_fn = unsafe extern "C" fn(*mut c_void, usize);
+pub type FIO_rust_zstd_write_acquire_fn =
+    unsafe extern "C" fn(*mut c_void, *mut *mut c_void, *mut *mut u8, *mut usize);
+pub type FIO_rust_zstd_write_enqueue_fn =
+    unsafe extern "C" fn(*mut c_void, *mut *mut c_void, usize, *mut *mut u8, *mut usize);
+pub type FIO_rust_zstd_write_release_fn = unsafe extern "C" fn(*mut c_void, *mut c_void);
+pub type FIO_rust_zstd_sparse_write_end_fn = unsafe extern "C" fn(*mut c_void);
+pub type FIO_rust_zstd_compress_stream_fn = unsafe extern "C" fn(
+    *mut c_void,
+    *const c_char,
+    c_int,
+    *const u8,
+    usize,
+    usize,
+    *mut u8,
+    usize,
+    *mut usize,
+    *mut usize,
+    *mut usize,
+    *mut usize,
+) -> c_int;
+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.
+#[repr(C)]
+pub struct FIO_rust_zstd_compress_projection_t {
+    pub read_opaque: *mut c_void,
+    pub write_opaque: *mut c_void,
+    pub codec_opaque: *mut c_void,
+    pub policy_opaque: *mut c_void,
+    pub read_buffer_size: usize,
+    pub read_fill: Option<FIO_rust_zstd_read_fill_fn>,
+    pub read_consume: Option<FIO_rust_zstd_read_consume_fn>,
+    pub write_acquire: Option<FIO_rust_zstd_write_acquire_fn>,
+    pub write_enqueue: Option<FIO_rust_zstd_write_enqueue_fn>,
+    pub write_release: Option<FIO_rust_zstd_write_release_fn>,
+    pub sparse_write_end: Option<FIO_rust_zstd_sparse_write_end_fn>,
+    pub compress_stream: Option<FIO_rust_zstd_compress_stream_fn>,
+    pub iteration: Option<FIO_rust_zstd_iteration_fn>,
+}
+
 pub const FIO_RUST_GZIP_OK: c_int = 0;
 pub const FIO_RUST_GZIP_INIT_ERROR: c_int = 1;
 pub const FIO_RUST_GZIP_DEFLATE_ERROR: c_int = 2;
@@ -1973,6 +2027,196 @@ pub unsafe extern "C" fn FIO_rust_compressFilenameInternal(
    FIO_RUST_COMPRESS_OK
 }

+/// 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.
+#[no_mangle]
+pub unsafe extern "C" fn FIO_rust_compressZstdFrame(
+    projection: *const FIO_rust_zstd_compress_projection_t,
+    src_file_name: *const c_char,
+    src_file_size: u64,
+    compression_level: c_int,
+    read_size: *mut u64,
+    compressed_size: *mut u64,
+    zstd_result: *mut usize,
+) -> c_int {
+    assert!(!projection.is_null());
+    assert!(!src_file_name.is_null());
+    assert!(!read_size.is_null());
+    assert!(!compressed_size.is_null());
+    assert!(!zstd_result.is_null());
+
+    unsafe {
+        *read_size = 0;
+        *compressed_size = 0;
+        *zstd_result = 0;
+    }
+
+    let projection = unsafe { &*projection };
+    if projection.read_buffer_size == 0 {
+        return FIO_RUST_ZSTD_INVALID_PROJECTION;
+    }
+    let Some(read_fill) = projection.read_fill else {
+        return FIO_RUST_ZSTD_INVALID_PROJECTION;
+    };
+    let Some(read_consume) = projection.read_consume else {
+        return FIO_RUST_ZSTD_INVALID_PROJECTION;
+    };
+    let Some(write_acquire) = projection.write_acquire else {
+        return FIO_RUST_ZSTD_INVALID_PROJECTION;
+    };
+    let Some(write_enqueue) = projection.write_enqueue else {
+        return FIO_RUST_ZSTD_INVALID_PROJECTION;
+    };
+    let Some(write_release) = projection.write_release else {
+        return FIO_RUST_ZSTD_INVALID_PROJECTION;
+    };
+    let Some(sparse_write_end) = projection.sparse_write_end else {
+        return FIO_RUST_ZSTD_INVALID_PROJECTION;
+    };
+    let Some(compress_stream) = projection.compress_stream else {
+        return FIO_RUST_ZSTD_INVALID_PROJECTION;
+    };
+    let Some(iteration) = projection.iteration else {
+        return FIO_RUST_ZSTD_INVALID_PROJECTION;
+    };
+
+    let mut job = ptr::null_mut::<c_void>();
+    let mut output = ptr::null_mut::<u8>();
+    let mut output_size = 0_usize;
+    unsafe {
+        write_acquire(
+            projection.write_opaque,
+            &mut job,
+            &mut output,
+            &mut output_size,
+        );
+    }
+    if job.is_null() || (output.is_null() && output_size != 0) {
+        return FIO_RUST_ZSTD_INVALID_PROJECTION;
+    }
+
+    let mut input = ptr::null::<u8>();
+    let mut input_size = 0_usize;
+    let mut input_pos = 0_usize;
+    let mut in_file_size = 0_u64;
+    let mut out_file_size = 0_u64;
+    let mut directive = FIO_RUST_ZSTD_E_CONTINUE;
+    let mut compression_level = compression_level;
+
+    loop {
+        if input_pos == input_size {
+            let mut loaded = 0_usize;
+            let added = unsafe {
+                read_fill(
+                    projection.read_opaque,
+                    projection.read_buffer_size,
+                    &mut input,
+                    &mut loaded,
+                )
+            };
+            if loaded != 0 && input.is_null() {
+                return FIO_RUST_ZSTD_INVALID_PROJECTION;
+            }
+            input_size = loaded;
+            input_pos = 0;
+            in_file_size = in_file_size.wrapping_add(added as u64);
+            unsafe { *read_size = in_file_size };
+
+            if loaded == 0
+                || (src_file_size != UTIL_FILESIZE_UNKNOWN && in_file_size == src_file_size)
+            {
+                directive = FIO_RUST_ZSTD_E_END;
+            }
+        }
+
+        let mut still_to_flush = 1_usize;
+        while input_pos != input_size || (directive == FIO_RUST_ZSTD_E_END && still_to_flush != 0) {
+            let old_input_pos = input_pos;
+            let mut new_input_pos = input_pos;
+            let mut output_produced = 0_usize;
+            let mut to_flush_now = 0_usize;
+            let mut codec_result = 0_usize;
+            let status = unsafe {
+                compress_stream(
+                    projection.codec_opaque,
+                    src_file_name,
+                    directive,
+                    input,
+                    input_size,
+                    input_pos,
+                    output,
+                    output_size,
+                    &mut new_input_pos,
+                    &mut output_produced,
+                    &mut to_flush_now,
+                    &mut codec_result,
+                )
+            };
+            if status != 0 {
+                unsafe {
+                    *zstd_result = codec_result;
+                    *read_size = in_file_size;
+                    *compressed_size = out_file_size;
+                }
+                return FIO_RUST_ZSTD_COMPRESS_ERROR;
+            }
+            still_to_flush = codec_result;
+            assert!(new_input_pos >= old_input_pos);
+            assert!(new_input_pos <= input_size);
+            assert!(output_produced <= output_size);
+
+            unsafe { read_consume(projection.read_opaque, new_input_pos - old_input_pos) };
+            input_pos = new_input_pos;
+
+            if output_produced != 0 {
+                unsafe {
+                    write_enqueue(
+                        projection.write_opaque,
+                        &mut job,
+                        output_produced,
+                        &mut output,
+                        &mut output_size,
+                    );
+                }
+                out_file_size = out_file_size.wrapping_add(output_produced as u64);
+                unsafe { *compressed_size = out_file_size };
+            }
+
+            unsafe {
+                iteration(
+                    projection.policy_opaque,
+                    src_file_name,
+                    &mut compression_level,
+                    old_input_pos,
+                    new_input_pos,
+                    to_flush_now,
+                )
+            };
+        }
+
+        if directive == FIO_RUST_ZSTD_E_END {
+            break;
+        }
+    }
+
+    if src_file_size != UTIL_FILESIZE_UNKNOWN && in_file_size != src_file_size {
+        unsafe {
+            *read_size = in_file_size;
+            *compressed_size = out_file_size;
+        }
+        return FIO_RUST_ZSTD_INCOMPLETE_INPUT;
+    }
+
+    unsafe {
+        *read_size = in_file_size;
+        *compressed_size = out_file_size;
+        write_release(projection.write_opaque, job);
+        sparse_write_end(projection.write_opaque);
+    }
+    FIO_RUST_ZSTD_OK
+}
+
 /// Compresses one gzip member through the C-owned zlib and asynchronous
 /// resource callbacks.  The loop mirrors the original `FIO_compressGzFrame`
 /// sequencing: input is counted when a read-pool buffer is loaded, consumed
@@ -3837,6 +4081,300 @@ mod tests {
        assert_eq!(context.totalBytesOutput, 47);
    }

+    struct ZstdProjectionState {
+        input: [u8; 5],
+        input_pos: usize,
+        output_buffer: [u8; 4],
+        consumed: Vec<usize>,
+        enqueue_sizes: Vec<usize>,
+        output_chunks: Vec<Vec<u8>>,
+        directives: Vec<c_int>,
+        iterations: Vec<(usize, usize, usize, c_int)>,
+        acquire_calls: usize,
+        release_calls: usize,
+        sparse_end_calls: usize,
+        compress_calls: usize,
+        codec_error: Option<usize>,
+    }
+
+    impl Default for ZstdProjectionState {
+        fn default() -> Self {
+            Self {
+                input: *b"abcde",
+                input_pos: 0,
+                output_buffer: [0; 4],
+                consumed: Vec::new(),
+                enqueue_sizes: Vec::new(),
+                output_chunks: Vec::new(),
+                directives: Vec::new(),
+                iterations: Vec::new(),
+                acquire_calls: 0,
+                release_calls: 0,
+                sparse_end_calls: 0,
+                compress_calls: 0,
+                codec_error: None,
+            }
+        }
+    }
+
+    unsafe extern "C" fn zstd_test_read_fill(
+        opaque: *mut c_void,
+        requested: usize,
+        buffer: *mut *const u8,
+        loaded: *mut usize,
+    ) -> usize {
+        let state = unsafe { &mut *opaque.cast::<ZstdProjectionState>() };
+        let available = state.input.len() - state.input_pos;
+        let amount = available.min(requested);
+        unsafe {
+            *buffer = state.input.as_ptr().add(state.input_pos);
+            *loaded = amount;
+        }
+        amount
+    }
+
+    unsafe extern "C" fn zstd_test_read_consume(opaque: *mut c_void, amount: usize) {
+        let state = unsafe { &mut *opaque.cast::<ZstdProjectionState>() };
+        assert!(amount <= state.input.len() - state.input_pos);
+        state.input_pos += amount;
+        state.consumed.push(amount);
+    }
+
+    unsafe extern "C" fn zstd_test_write_acquire(
+        opaque: *mut c_void,
+        job: *mut *mut c_void,
+        buffer: *mut *mut u8,
+        buffer_size: *mut usize,
+    ) {
+        let state = unsafe { &mut *opaque.cast::<ZstdProjectionState>() };
+        state.acquire_calls += 1;
+        unsafe {
+            *job = opaque;
+            *buffer = state.output_buffer.as_mut_ptr();
+            *buffer_size = state.output_buffer.len();
+        }
+    }
+
+    unsafe extern "C" fn zstd_test_write_enqueue(
+        opaque: *mut c_void,
+        job: *mut *mut c_void,
+        used: usize,
+        buffer: *mut *mut u8,
+        buffer_size: *mut usize,
+    ) {
+        let state = unsafe { &mut *opaque.cast::<ZstdProjectionState>() };
+        assert_eq!(unsafe { *job }, opaque);
+        assert!(used <= state.output_buffer.len());
+        state.enqueue_sizes.push(used);
+        state
+            .output_chunks
+            .push(state.output_buffer[..used].to_vec());
+        unsafe {
+            *buffer = state.output_buffer.as_mut_ptr();
+            *buffer_size = state.output_buffer.len();
+        }
+    }
+
+    unsafe extern "C" fn zstd_test_write_release(opaque: *mut c_void, job: *mut c_void) {
+        let state = unsafe { &mut *opaque.cast::<ZstdProjectionState>() };
+        assert_eq!(job, opaque);
+        state.release_calls += 1;
+    }
+
+    unsafe extern "C" fn zstd_test_sparse_end(opaque: *mut c_void) {
+        let state = unsafe { &mut *opaque.cast::<ZstdProjectionState>() };
+        state.sparse_end_calls += 1;
+    }
+
+    unsafe extern "C" fn zstd_test_compress(
+        opaque: *mut c_void,
+        _src_file_name: *const c_char,
+        directive: c_int,
+        _input: *const u8,
+        input_size: usize,
+        input_pos: usize,
+        output: *mut u8,
+        output_size: usize,
+        input_pos_after: *mut usize,
+        output_produced: *mut usize,
+        to_flush_now: *mut usize,
+        zstd_result: *mut usize,
+    ) -> c_int {
+        let state = unsafe { &mut *opaque.cast::<ZstdProjectionState>() };
+        state.compress_calls += 1;
+        state.directives.push(directive);
+        let remaining = input_size - input_pos;
+        let consumed = remaining.min(2);
+        let needs_final_flush = directive == FIO_RUST_ZSTD_E_END && consumed == remaining;
+        let result = if needs_final_flush && remaining != 0 {
+            1
+        } else {
+            0
+        };
+
+        unsafe {
+            *input_pos_after = input_pos + consumed;
+            *output_produced = 1;
+            *to_flush_now = 1;
+            *zstd_result = result;
+            assert!(output_size != 0);
+            *output = b'x';
+        }
+        if let Some(error) = state.codec_error {
+            unsafe {
+                *input_pos_after = input_pos;
+                *output_produced = 0;
+                *zstd_result = error;
+            }
+            return 1;
+        }
+        0
+    }
+
+    unsafe extern "C" fn zstd_test_iteration(
+        opaque: *mut c_void,
+        _src_file_name: *const c_char,
+        compression_level: *mut c_int,
+        old_input_pos: usize,
+        new_input_pos: usize,
+        to_flush_now: usize,
+    ) {
+        let state = unsafe { &mut *opaque.cast::<ZstdProjectionState>() };
+        state
+            .iterations
+            .push((old_input_pos, new_input_pos, to_flush_now, unsafe {
+                *compression_level
+            }));
+    }
+
+    fn zstd_test_projection(
+        state: &mut ZstdProjectionState,
+    ) -> FIO_rust_zstd_compress_projection_t {
+        let opaque = (state as *mut ZstdProjectionState).cast::<c_void>();
+        FIO_rust_zstd_compress_projection_t {
+            read_opaque: opaque,
+            write_opaque: opaque,
+            codec_opaque: opaque,
+            policy_opaque: opaque,
+            read_buffer_size: 3,
+            read_fill: Some(zstd_test_read_fill),
+            read_consume: Some(zstd_test_read_consume),
+            write_acquire: Some(zstd_test_write_acquire),
+            write_enqueue: Some(zstd_test_write_enqueue),
+            write_release: Some(zstd_test_write_release),
+            sparse_write_end: Some(zstd_test_sparse_end),
+            compress_stream: Some(zstd_test_compress),
+            iteration: Some(zstd_test_iteration),
+        }
+    }
+
+    #[test]
+    fn zstd_projection_preserves_stream_order_and_accounting() {
+        let mut state = ZstdProjectionState::default();
+        let projection = zstd_test_projection(&mut state);
+        let mut read_size = 0;
+        let mut compressed_size = 0;
+        let mut zstd_result = 0;
+
+        assert_eq!(
+            unsafe {
+                FIO_rust_compressZstdFrame(
+                    &projection,
+                    c"zstd-test".as_ptr(),
+                    5,
+                    3,
+                    &mut read_size,
+                    &mut compressed_size,
+                    &mut zstd_result,
+                )
+            },
+            FIO_RUST_ZSTD_OK
+        );
+        assert_eq!(read_size, 5);
+        assert_eq!(compressed_size, 4);
+        assert_eq!(zstd_result, 0);
+        assert_eq!(state.consumed, vec![2, 1, 2, 0]);
+        assert_eq!(state.enqueue_sizes, vec![1, 1, 1, 1]);
+        assert_eq!(state.output_chunks, vec![vec![b'x']; 4]);
+        assert_eq!(state.compress_calls, 4);
+        assert_eq!(
+            state.directives,
+            vec![
+                FIO_RUST_ZSTD_E_CONTINUE,
+                FIO_RUST_ZSTD_E_CONTINUE,
+                FIO_RUST_ZSTD_E_END,
+                FIO_RUST_ZSTD_E_END,
+            ]
+        );
+        assert_eq!(state.iterations.len(), 4);
+        assert_eq!(state.acquire_calls, 1);
+        assert_eq!(state.release_calls, 1);
+        assert_eq!(state.sparse_end_calls, 1);
+    }
+
+    #[test]
+    fn zstd_projection_does_not_consume_input_after_codec_error() {
+        let mut state = ZstdProjectionState {
+            codec_error: Some(17),
+            ..ZstdProjectionState::default()
+        };
+        let projection = zstd_test_projection(&mut state);
+        let mut read_size = 0;
+        let mut compressed_size = 0;
+        let mut zstd_result = 0;
+
+        assert_eq!(
+            unsafe {
+                FIO_rust_compressZstdFrame(
+                    &projection,
+                    c"zstd-test".as_ptr(),
+                    5,
+                    3,
+                    &mut read_size,
+                    &mut compressed_size,
+                    &mut zstd_result,
+                )
+            },
+            FIO_RUST_ZSTD_COMPRESS_ERROR
+        );
+        assert_eq!(zstd_result, 17);
+        assert_eq!(read_size, 3);
+        assert_eq!(compressed_size, 0);
+        assert_eq!(state.compress_calls, 1);
+        assert!(state.consumed.is_empty());
+        assert_eq!(state.release_calls, 0);
+        assert_eq!(state.sparse_end_calls, 0);
+    }
+
+    #[test]
+    fn zstd_projection_reports_incomplete_input_without_releasing_output() {
+        let mut state = ZstdProjectionState::default();
+        let projection = zstd_test_projection(&mut state);
+        let mut read_size = 0;
+        let mut compressed_size = 0;
+        let mut zstd_result = 0;
+
+        assert_eq!(
+            unsafe {
+                FIO_rust_compressZstdFrame(
+                    &projection,
+                    c"zstd-test".as_ptr(),
+                    7,
+                    3,
+                    &mut read_size,
+                    &mut compressed_size,
+                    &mut zstd_result,
+                )
+            },
+            FIO_RUST_ZSTD_INCOMPLETE_INPUT
+        );
+        assert_eq!(read_size, 5);
+        assert_eq!(compressed_size, 4);
+        assert_eq!(zstd_result, 0);
+        assert_eq!(state.release_calls, 0);
+        assert_eq!(state.sparse_end_calls, 0);
+    }
+
    struct GzipProjectionState {
        input: [u8; 5],
        input_pos: usize,