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

refactor(fileio): move zstd stream callback into Rust

Browse Source 
The fileio zstd projection used to route its stream callback through a C
implementation that constructed the public input and output buffer views,
queried pending output, called the streaming codec, and copied four scalar
results back to the Rust-owned loop.  That left the central codec operation in
the CLI C translation unit even though the surrounding stream loop was already
in Rust.

Move that callback into Rust while keeping the C CCtx opaque across the
boundary.  Rust now builds the public ZSTD_inBuffer and ZSTD_outBuffer views,
invokes ZSTD_toFlushNow and ZSTD_compressStream2, publishes the original
positions and result, and preserves the existing success diagnostic through a
small C display callback.  The C projection passes the real CCtx as codecOpaque
and retains the adaptive iteration context and all private CLI state.  The
buffer structs are made public within the Rust crate so this seam can reuse the
existing C-compatible definitions without duplicating them.

Test Plan:
- `cargo +nightly fmt --manifest-path rust/Cargo.toml --all` -- passed
- `cc -fsyntax-only -Werror=incompatible-pointer-types -Ilib -Ilib/common -Ilib/compress -Ilib/decompress -Ilib/dict -Ilib/legacy programs/fileio.c` -- passed
- `git diff --check` and `git diff --cached --check` -- passed
- Full capped native and Rust verification remains to be run after this seam
This commit is contained in:
ddidderr
2026-07-20 15:09:23 +02:00
parent 0e93557879
commit 725877ad7a
3 changed files with 239 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
programs/fileio.c
+14 -32
View File
@@ -331,8 +331,6 @@ int FIO_rust_adjustParamsForPatchFromMode(FIO_prefs_t* prefs,
unsigned* fileWindowLog,
int* autoLdm,
int* optimalParser);
void FIO_setInBuffer(ZSTD_inBuffer* output, const void* buf, size_t s, size_t pos);
void FIO_setOutBuffer(ZSTD_outBuffer* output, void* buf, size_t s, size_t pos);
const char* FIO_determineCompressedName(const char* srcFileName, const char* outDirName, const char* suffix);
const char* FIO_rust_determineDstName(const char* srcFileName, const char* outDirName,
const char* const* suffixList, const char* suffixListStr);
@@ -1333,6 +1331,14 @@ typedef int (*FIO_rust_zstd_compress_stream_fn)(
unsigned char* output, size_t outputSize,
size_t* inputPosAfter, size_t* outputProduced,
size_t* toFlushNow, size_t* zstdResult);
int FIO_rust_zstd_compressStream(
void* opaque, const char* srcFileName, int directive,
const unsigned char* input, size_t inputSize, size_t inputPos,
unsigned char* output, size_t outputSize,
size_t* inputPosAfter, size_t* outputProduced,
size_t* toFlushNow, size_t* zstdResult);
void FIO_rust_zstd_compressStreamDisplay(
int directive, size_t inputPos, size_t inputSize, size_t outputProduced);
typedef void (*FIO_rust_zstd_iteration_fn)(
void* opaque, const char* srcFileName, int* compressionLevel,
size_t oldInputPos, size_t newInputPos, size_t toFlushNow);
@@ -2524,36 +2530,12 @@ static void FIO_rust_zstd_sparseWriteEnd(void* opaque)
AIO_WritePool_sparseWriteEnd((WritePoolCtx_t*)opaque);
}
static int FIO_rust_zstd_compressStream(
void* opaque, const char* srcFileName, int directive,
const unsigned char* input, size_t inputSize, size_t inputPos,
unsigned char* output, size_t outputSize,
size_t* inputPosAfter, size_t* outputProduced,
size_t* toFlushNow, size_t* zstdResult)
void FIO_rust_zstd_compressStreamDisplay(
int directive, size_t inputPos, size_t inputSize, size_t outputProduced)
{
FIO_rust_zstd_projection_context_t* const context =
(FIO_rust_zstd_projection_context_t*)opaque;
ZSTD_inBuffer inBuff;
ZSTD_outBuffer outBuff;
size_t toFlush;
size_t result;
FIO_setInBuffer(&inBuff, input, inputSize, inputPos);
FIO_setOutBuffer(&outBuff, output, outputSize, 0);
toFlush = ZSTD_toFlushNow(context->cctx);
result = ZSTD_compressStream2(
context->cctx, &outBuff, &inBuff, (ZSTD_EndDirective)directive);
*inputPosAfter = inBuff.pos;
*outputProduced = outBuff.pos;
*toFlushNow = toFlush;
*zstdResult = result;
if (!ZSTD_isError(result)) {
DISPLAYLEVEL(6, "ZSTD_compress_generic(end:%u) => input pos(%u)<=(%u)size ; output generated %u bytes \n",
(unsigned)directive, (unsigned)inBuff.pos,
(unsigned)inBuff.size, (unsigned)outBuff.pos);
}
(void)srcFileName;
return ZSTD_isError(result);
DISPLAYLEVEL(6, "ZSTD_compress_generic(end:%u) => input pos(%u)<=(%u)size ; output generated %u bytes \n",
(unsigned)directive, (unsigned)inputPos,
(unsigned)inputSize, (unsigned)outputProduced);
}
static void FIO_rust_zstd_iteration(void* opaque, const char* srcFileName,
@@ -2766,7 +2748,7 @@ FIO_rust_compressZstdCallback(void* fCtx, void* prefs, void* ress,
memset(&projection, 0, sizeof(projection));
projection.readOpaque = (void*)ressPtr->readCtx;
projection.writeOpaque = (void*)ressPtr->writeCtx;
projection.codecOpaque = &context;
projection.codecOpaque = (void*)ressPtr->cctx;
projection.policyOpaque = &context;
projection.readBufferSize = ZSTD_CStreamInSize();
projection.readFill = FIO_rust_zstd_readFill;
rust/src/fileio_asyncio.rs
+219 -3
View File
@@ -26,6 +26,19 @@ use std::ptr;
use std::sync::{Arc, Condvar, Mutex};
use std::thread::{self, JoinHandle};
use crate::zstd_compress::{ZSTD_inBuffer, ZSTD_outBuffer};
#[cfg(not(test))]
unsafe extern "C" {
fn ZSTD_toFlushNow(cctx: *mut c_void) -> usize;
fn FIO_rust_zstd_compressStreamDisplay(
directive: c_int,
input_pos: usize,
input_size: usize,
output_produced: usize,
);
}
const MAX_IO_JOBS: usize = 10;
const IO_QUEUE_SIZE: usize = MAX_IO_JOBS - 2;
const SPARSE_SEGMENT_SIZE: usize = 32 * 1024;
@@ -3318,10 +3331,106 @@ pub unsafe extern "C" fn FIO_rust_zstd_adapt(
zstd_adapt_policy(policy, projection)
}
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;
type FIO_rust_zstd_compress_display_fn = unsafe extern "C" fn(c_int, usize, usize, usize);
/// Build the public stream-buffer views, call the Rust compressor, and publish
/// the same scalar results as the former C-only callback. The codec context
/// stays opaque; only the public buffer ABI crosses into the compressor.
unsafe fn fio_zstd_compress_stream_with(
cctx: *mut c_void,
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,
to_flush_now_fn: FIO_rust_zstd_to_flush_now_fn,
compress_stream2_fn: FIO_rust_zstd_compress_stream2_fn,
display_fn: Option<FIO_rust_zstd_compress_display_fn>,
) -> c_int {
let mut input_view = ZSTD_inBuffer {
src: input.cast::<c_void>(),
size: input_size,
pos: input_pos,
};
let mut output_view = ZSTD_outBuffer {
dst: output.cast::<c_void>(),
size: output_size,
pos: 0,
};
let pending_flush = unsafe { to_flush_now_fn(cctx) };
let result = unsafe { compress_stream2_fn(cctx, &mut output_view, &mut input_view, directive) };
unsafe {
*input_pos_after = input_view.pos;
*output_produced = output_view.pos;
*to_flush_now = pending_flush;
*zstd_result = result;
}
if !crate::errors::ERR_isError(result) {
if let Some(display) = display_fn {
unsafe {
display(directive, input_view.pos, input_view.size, output_view.pos);
}
}
0
} else {
1
}
}
/// The codec callback used by the zstd file-I/O projection. C retains only
/// the iteration callback's private adaptive/diagnostic context; this callback
/// receives the opaque `ZSTD_CCtx` directly and delegates to Rust's public
/// `ZSTD_compressStream2` implementation.
#[cfg(not(test))]
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_zstd_compressStream(
cctx: *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 {
unsafe {
fio_zstd_compress_stream_with(
cctx,
directive,
input,
input_size,
input_pos,
output,
output_size,
input_pos_after,
output_produced,
to_flush_now,
zstd_result,
ZSTD_toFlushNow,
crate::zstd_compress::ZSTD_compressStream2,
Some(FIO_rust_zstd_compressStreamDisplay),
)
}
}
/// 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()`, adaptive diagnostics, and all private CLI
/// state.
/// policy. Rust owns the stream loop and codec call; C callbacks retain the
/// 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,
@@ -6796,6 +6905,113 @@ mod tests {
assert_eq!(context.nbFilesProcessed, 4);
}
struct ZstdCodecCallbackTestState {
pending_flush: usize,
codec_result: usize,
observed_input: Option<(usize, usize, usize)>,
observed_output: Option<(usize, usize, usize)>,
observed_directive: c_int,
}
unsafe extern "C" fn zstd_test_to_flush_now(opaque: *mut c_void) -> usize {
let state = unsafe { &mut *opaque.cast::<ZstdCodecCallbackTestState>() };
state.pending_flush
}
unsafe extern "C" fn zstd_test_compress_stream2(
opaque: *mut c_void,
output: *mut ZSTD_outBuffer,
input: *mut ZSTD_inBuffer,
directive: c_int,
) -> usize {
let state = unsafe { &mut *opaque.cast::<ZstdCodecCallbackTestState>() };
let input = unsafe { &mut *input };
let output = unsafe { &mut *output };
state.observed_input = Some((input.src as usize, input.size, input.pos));
state.observed_output = Some((output.dst as usize, output.size, output.pos));
state.observed_directive = directive;
input.pos = input.size;
output.pos = 3;
state.codec_result
}
#[test]
fn zstd_codec_callback_builds_public_views_and_preserves_results() {
let input = b"abcdef";
let mut output = [0_u8; 8];
let mut state = ZstdCodecCallbackTestState {
pending_flush: 19,
codec_result: 0,
observed_input: None,
observed_output: None,
observed_directive: -1,
};
let mut input_pos_after = 0;
let mut output_produced = 0;
let mut to_flush_now = 0;
let mut zstd_result = 0;
let cctx = (&mut state as *mut ZstdCodecCallbackTestState).cast::<c_void>();
assert_eq!(
unsafe {
fio_zstd_compress_stream_with(
cctx,
2,
input.as_ptr(),
input.len(),
1,
output.as_mut_ptr(),
output.len(),
&mut input_pos_after,
&mut output_produced,
&mut to_flush_now,
&mut zstd_result,
zstd_test_to_flush_now,
zstd_test_compress_stream2,
None,
)
},
0
);
assert_eq!(
state.observed_input,
Some((input.as_ptr() as usize, input.len(), 1))
);
assert_eq!(
state.observed_output,
Some((output.as_mut_ptr() as usize, output.len(), 0))
);
assert_eq!(state.observed_directive, 2);
assert_eq!(input_pos_after, input.len());
assert_eq!(output_produced, 3);
assert_eq!(to_flush_now, 19);
assert_eq!(zstd_result, 0);
state.codec_result = usize::MAX;
assert_eq!(
unsafe {
fio_zstd_compress_stream_with(
cctx,
0,
input.as_ptr(),
input.len(),
0,
output.as_mut_ptr(),
output.len(),
&mut input_pos_after,
&mut output_produced,
&mut to_flush_now,
&mut zstd_result,
zstd_test_to_flush_now,
zstd_test_compress_stream2,
None,
)
},
1
);
assert_eq!(zstd_result, usize::MAX);
}
struct ZstdProjectionState {
input: [u8; 5],
input_pos: usize,
rust/src/zstd_compress.rs
+6 -6
View File
@@ -6282,16 +6282,16 @@ pub extern "C" fn ZSTD_rust_targetCBlockSizeAction(
#[repr(C)]
pub struct ZSTD_inBuffer {
src: *const c_void,
size: usize,
pos: usize,
pub src: *const c_void,
pub size: usize,
pub pos: usize,
}
#[repr(C)]
pub struct ZSTD_outBuffer {
dst: *mut c_void,
size: usize,
pos: usize,
pub dst: *mut c_void,
pub size: usize,
pub pos: usize,
}
#[no_mangle]