feat(cli): move zstd frame decode loop into Rust
Move the zstd frame read/decode/write loop behind a narrow Rust ABI. Rust now owns decoder reset, asynchronous input/output buffer coordination, frame progress accounting, frame-boundary handling, and status classification. C retains the mixed-format dispatcher, user-facing diagnostics, window-limit help, progress formatting callback, source/destination policy, cleanup, and metadata handling. The bridge passes opaque decoder and I/O-pool pointers plus scalar counters and out-parameters. Decoder errors return before consuming the current input buffer so the C diagnostic path can inspect it. The Rust test seam covers valid streaming, concatenated frame boundaries, pre-increment progress semantics, decoder-error input preservation, and premature EOF. Test Plan: - cargo test --manifest-path rust/Cargo.toml --lib fileio_asyncio -- --test-threads=1 - cargo clippy --manifest-path rust/Cargo.toml --lib -- -D warnings - cargo +nightly fmt --manifest-path rust/Cargo.toml -- --check - cargo test --manifest-path rust/Cargo.toml --lib -- --test-threads=1 - make -B -C tests -j2 test-cli-tests (all 41 passed) - git diff --cached --check
This commit is contained in:
+61
-59
@@ -404,6 +404,23 @@ void FIO_rust_freeDict(int dictBufferType,
|
||||
void** dictHandle);
|
||||
int FIO_rust_removeFile(const char* path);
|
||||
int FIO_rust_passThrough(ReadPoolCtx_t* readCtx, WritePoolCtx_t* writeCtx);
|
||||
enum {
|
||||
FIO_RUST_ZSTD_FRAME_OK = 0,
|
||||
FIO_RUST_ZSTD_FRAME_DECODING_ERROR = 1,
|
||||
FIO_RUST_ZSTD_FRAME_PREMATURE_END = 2,
|
||||
};
|
||||
typedef void (*FIO_rust_frame_progress_fn)(void* opaque,
|
||||
const char* srcFileName,
|
||||
U64 decodedSize);
|
||||
int FIO_rust_decompressZstdFrame(void* fCtx,
|
||||
void* dctx,
|
||||
ReadPoolCtx_t* readCtx,
|
||||
WritePoolCtx_t* writeCtx,
|
||||
const char* srcFileName,
|
||||
U64 alreadyDecoded,
|
||||
U64* frameSize,
|
||||
size_t* zstdError,
|
||||
FIO_rust_frame_progress_fn progress);
|
||||
void FIO_rust_displayCompressionParameters(const FIO_prefs_t* prefs);
|
||||
#ifdef ZSTD_LZ4COMPRESS
|
||||
int FIO_rust_LZ4_GetBlockSize_FromBlockId(int id);
|
||||
@@ -2192,6 +2209,31 @@ FIO_zstdErrorHelp(const FIO_prefs_t* const prefs,
|
||||
* @return : size of decoded zstd frame, or an error code
|
||||
*/
|
||||
#define FIO_ERROR_FRAME_DECODING ((unsigned long long)(-2))
|
||||
static void
|
||||
FIO_decompressZstdFrameProgress(void* const opaque,
|
||||
const char* const srcFileName,
|
||||
U64 const decodedSize)
|
||||
{
|
||||
FIO_ctx_t* const fCtx = (FIO_ctx_t*)opaque;
|
||||
const char* srcFName20 = srcFileName;
|
||||
size_t const srcFileLength = strlen(srcFileName);
|
||||
UTIL_HumanReadableSize_t const hrs = UTIL_makeHumanReadableSize(decodedSize);
|
||||
|
||||
/* display last 20 characters only when not --verbose */
|
||||
if ((srcFileLength>20) && (g_display_prefs.displayLevel<3))
|
||||
srcFName20 += srcFileLength-20;
|
||||
|
||||
if (fCtx->nbFilesTotal > 1) {
|
||||
DISPLAYUPDATE_PROGRESS(
|
||||
"\rDecompress: %2u/%2u files. Current: %s : %.*f%s... ",
|
||||
fCtx->currFileIdx+1, fCtx->nbFilesTotal, srcFName20,
|
||||
hrs.precision, hrs.value, hrs.suffix);
|
||||
} else {
|
||||
DISPLAYUPDATE_PROGRESS("\r%-20.20s : %.*f%s... ",
|
||||
srcFName20, hrs.precision, hrs.value, hrs.suffix);
|
||||
}
|
||||
}
|
||||
|
||||
static unsigned long long
|
||||
FIO_decompressZstdFrame(FIO_ctx_t* const fCtx, dRess_t* ress,
|
||||
const FIO_prefs_t* const prefs,
|
||||
@@ -2199,68 +2241,28 @@ FIO_decompressZstdFrame(FIO_ctx_t* const fCtx, dRess_t* ress,
|
||||
U64 alreadyDecoded) /* for multi-frames streams */
|
||||
{
|
||||
U64 frameSize = 0;
|
||||
const char* srcFName20 = srcFileName;
|
||||
IOJob_t* writeJob = AIO_WritePool_acquireJob(ress->writeCtx);
|
||||
assert(writeJob);
|
||||
size_t zstdError = 0;
|
||||
int const status = FIO_rust_decompressZstdFrame(
|
||||
fCtx, ress->dctx, ress->readCtx, ress->writeCtx, srcFileName,
|
||||
alreadyDecoded, &frameSize, &zstdError,
|
||||
FIO_decompressZstdFrameProgress);
|
||||
|
||||
/* display last 20 characters only when not --verbose */
|
||||
{ size_t const srcFileLength = strlen(srcFileName);
|
||||
if ((srcFileLength>20) && (g_display_prefs.displayLevel<3))
|
||||
srcFName20 += srcFileLength-20;
|
||||
if (status == FIO_RUST_ZSTD_FRAME_OK)
|
||||
return frameSize;
|
||||
if (status == FIO_RUST_ZSTD_FRAME_DECODING_ERROR) {
|
||||
DISPLAYLEVEL(1, "%s : Decoding error (36) : %s \n",
|
||||
srcFileName, ZSTD_getErrorName(zstdError));
|
||||
FIO_zstdErrorHelp(prefs, ress, zstdError, srcFileName);
|
||||
return FIO_ERROR_FRAME_DECODING;
|
||||
}
|
||||
if (status == FIO_RUST_ZSTD_FRAME_PREMATURE_END) {
|
||||
DISPLAYLEVEL(1, "%s : Read error (39) : premature end \n",
|
||||
srcFileName);
|
||||
return FIO_ERROR_FRAME_DECODING;
|
||||
}
|
||||
|
||||
ZSTD_DCtx_reset(ress->dctx, ZSTD_reset_session_only);
|
||||
|
||||
/* Header loading : ensures ZSTD_getFrameHeader() will succeed */
|
||||
AIO_ReadPool_fillBuffer(ress->readCtx, ZSTD_FRAMEHEADERSIZE_MAX);
|
||||
|
||||
/* Main decompression Loop */
|
||||
while (1) {
|
||||
ZSTD_inBuffer inBuff = setInBuffer( ress->readCtx->srcBuffer, ress->readCtx->srcBufferLoaded, 0 );
|
||||
ZSTD_outBuffer outBuff= setOutBuffer( writeJob->buffer, writeJob->bufferSize, 0 );
|
||||
size_t const readSizeHint = ZSTD_decompressStream(ress->dctx, &outBuff, &inBuff);
|
||||
UTIL_HumanReadableSize_t const hrs = UTIL_makeHumanReadableSize(alreadyDecoded+frameSize);
|
||||
if (ZSTD_isError(readSizeHint)) {
|
||||
DISPLAYLEVEL(1, "%s : Decoding error (36) : %s \n",
|
||||
srcFileName, ZSTD_getErrorName(readSizeHint));
|
||||
FIO_zstdErrorHelp(prefs, ress, readSizeHint, srcFileName);
|
||||
AIO_WritePool_releaseIoJob(writeJob);
|
||||
return FIO_ERROR_FRAME_DECODING;
|
||||
}
|
||||
|
||||
/* Write block */
|
||||
writeJob->usedBufferSize = outBuff.pos;
|
||||
AIO_WritePool_enqueueAndReacquireWriteJob(&writeJob);
|
||||
frameSize += outBuff.pos;
|
||||
if (fCtx->nbFilesTotal > 1) {
|
||||
DISPLAYUPDATE_PROGRESS(
|
||||
"\rDecompress: %2u/%2u files. Current: %s : %.*f%s... ",
|
||||
fCtx->currFileIdx+1, fCtx->nbFilesTotal, srcFName20, hrs.precision, hrs.value, hrs.suffix);
|
||||
} else {
|
||||
DISPLAYUPDATE_PROGRESS("\r%-20.20s : %.*f%s... ",
|
||||
srcFName20, hrs.precision, hrs.value, hrs.suffix);
|
||||
}
|
||||
|
||||
AIO_ReadPool_consumeBytes(ress->readCtx, inBuff.pos);
|
||||
|
||||
if (readSizeHint == 0) break; /* end of frame */
|
||||
|
||||
/* Fill input buffer */
|
||||
{ size_t const toDecode = MIN(readSizeHint, ZSTD_DStreamInSize()); /* support large skippable frames */
|
||||
if (ress->readCtx->srcBufferLoaded < toDecode) {
|
||||
size_t const readSize = AIO_ReadPool_fillBuffer(ress->readCtx, toDecode);
|
||||
if (readSize==0) {
|
||||
DISPLAYLEVEL(1, "%s : Read error (39) : premature end \n",
|
||||
srcFileName);
|
||||
AIO_WritePool_releaseIoJob(writeJob);
|
||||
return FIO_ERROR_FRAME_DECODING;
|
||||
}
|
||||
} } }
|
||||
|
||||
AIO_WritePool_releaseIoJob(writeJob);
|
||||
AIO_WritePool_sparseWriteEnd(ress->writeCtx);
|
||||
|
||||
return frameSize;
|
||||
assert(0);
|
||||
return FIO_ERROR_FRAME_DECODING;
|
||||
}
|
||||
|
||||
|
||||
|
||||
+517
-1
@@ -1,6 +1,7 @@
|
||||
#![allow(non_camel_case_types)]
|
||||
#![allow(non_snake_case)]
|
||||
#![allow(clippy::missing_safety_doc)]
|
||||
#![allow(clippy::too_many_arguments)]
|
||||
|
||||
//! Rust implementation of the exported asynchronous file-I/O pools.
|
||||
//!
|
||||
@@ -16,7 +17,7 @@
|
||||
//! program, library, and C-test archives.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::ffi::c_void;
|
||||
use std::ffi::{c_char, c_void};
|
||||
use std::mem::{align_of, size_of};
|
||||
#[cfg(any(windows, not(any(unix, windows))))]
|
||||
use std::os::raw::c_long;
|
||||
@@ -30,6 +31,20 @@ const IO_QUEUE_SIZE: usize = MAX_IO_JOBS - 2;
|
||||
const SPARSE_SEGMENT_SIZE: usize = 32 * 1024;
|
||||
const SPARSE_SKIP_CHUNK: u64 = 1 << 30;
|
||||
const PASS_THROUGH_MAX_BLOCK_SIZE: usize = 64 * 1024;
|
||||
const ZSTD_FRAMEHEADERSIZE_MAX: usize = 18;
|
||||
|
||||
pub const FIO_RUST_ZSTD_FRAME_OK: c_int = 0;
|
||||
pub const FIO_RUST_ZSTD_FRAME_DECODING_ERROR: c_int = 1;
|
||||
pub const FIO_RUST_ZSTD_FRAME_PREMATURE_END: c_int = 2;
|
||||
|
||||
type FIO_rust_frame_progress_fn = Option<unsafe extern "C" fn(*mut c_void, *const c_char, u64)>;
|
||||
type FIO_zstd_reset_fn = unsafe extern "C" fn(*mut c_void, c_int) -> usize;
|
||||
type FIO_zstd_decompress_fn = unsafe extern "C" fn(
|
||||
*mut c_void,
|
||||
*mut crate::zstd_decompress::ZSTD_outBuffer,
|
||||
*mut crate::zstd_decompress::ZSTD_inBuffer,
|
||||
) -> usize;
|
||||
type FIO_zstd_in_size_fn = extern "C" fn() -> usize;
|
||||
|
||||
/// C's `FIO_prefs_t` from `programs/fileio_types.h`.
|
||||
///
|
||||
@@ -1393,6 +1408,168 @@ pub unsafe extern "C" fn FIO_rust_passThrough(
|
||||
0
|
||||
}
|
||||
|
||||
/// Decompresses exactly one zstd frame using the existing asynchronous pools.
|
||||
///
|
||||
/// C retains the frame dispatcher and all user-facing diagnostics. This ABI
|
||||
/// deliberately exposes only opaque contexts, scalar counters, and output
|
||||
/// parameters so the C-owned `dRess_t` never crosses into Rust. The input
|
||||
/// pool is advanced only after a successful decoder call; in particular, an
|
||||
/// error leaves the current input buffer untouched for `FIO_zstdErrorHelp()`.
|
||||
unsafe fn decompress_zstd_frame_with(
|
||||
f_ctx: *mut c_void,
|
||||
dctx: *mut c_void,
|
||||
read_ctx: *mut ReadPoolCtx_t,
|
||||
write_ctx: *mut WritePoolCtx_t,
|
||||
src_file_name: *const c_char,
|
||||
already_decoded: u64,
|
||||
frame_size: *mut u64,
|
||||
zstd_error: *mut usize,
|
||||
progress: FIO_rust_frame_progress_fn,
|
||||
reset: FIO_zstd_reset_fn,
|
||||
decompress: FIO_zstd_decompress_fn,
|
||||
dstream_in_size: FIO_zstd_in_size_fn,
|
||||
) -> c_int {
|
||||
assert!(!dctx.is_null());
|
||||
assert!(!read_ctx.is_null());
|
||||
assert!(!write_ctx.is_null());
|
||||
assert!(!src_file_name.is_null());
|
||||
assert!(!frame_size.is_null());
|
||||
assert!(!zstd_error.is_null());
|
||||
|
||||
unsafe {
|
||||
*frame_size = 0;
|
||||
*zstd_error = 0;
|
||||
}
|
||||
|
||||
let mut write_job = unsafe { AIO_WritePool_acquireJob(write_ctx) };
|
||||
assert!(!write_job.is_null());
|
||||
|
||||
unsafe {
|
||||
// The C implementation intentionally ignores this reset result.
|
||||
reset(dctx, 1); // ZSTD_reset_session_only
|
||||
AIO_ReadPool_fillBuffer(read_ctx, ZSTD_FRAMEHEADERSIZE_MAX);
|
||||
}
|
||||
|
||||
loop {
|
||||
let mut input = crate::zstd_decompress::ZSTD_inBuffer {
|
||||
src: unsafe { read_buffer_ptr(read_ctx) }.cast::<c_void>(),
|
||||
size: unsafe { read_buffer_loaded(read_ctx) },
|
||||
pos: 0,
|
||||
};
|
||||
let mut output = crate::zstd_decompress::ZSTD_outBuffer {
|
||||
dst: unsafe { (*write_job).buffer },
|
||||
size: unsafe { (*write_job).bufferSize },
|
||||
pos: 0,
|
||||
};
|
||||
let read_size_hint = unsafe { decompress(dctx, &mut output, &mut input) };
|
||||
|
||||
if crate::errors::ERR_isError(read_size_hint) {
|
||||
unsafe {
|
||||
*zstd_error = read_size_hint;
|
||||
AIO_WritePool_releaseIoJob(write_job);
|
||||
}
|
||||
return FIO_RUST_ZSTD_FRAME_DECODING_ERROR;
|
||||
}
|
||||
|
||||
let progress_size = unsafe { already_decoded.wrapping_add(*frame_size) };
|
||||
unsafe {
|
||||
(*write_job).usedBufferSize = output.pos;
|
||||
AIO_WritePool_enqueueAndReacquireWriteJob(&mut write_job);
|
||||
*frame_size = (*frame_size).wrapping_add(output.pos as u64);
|
||||
AIO_ReadPool_consumeBytes(read_ctx, input.pos);
|
||||
}
|
||||
|
||||
if let Some(callback) = progress {
|
||||
unsafe { callback(f_ctx, src_file_name, progress_size) };
|
||||
}
|
||||
|
||||
if read_size_hint == 0 {
|
||||
unsafe {
|
||||
AIO_WritePool_releaseIoJob(write_job);
|
||||
AIO_WritePool_sparseWriteEnd(write_ctx);
|
||||
}
|
||||
return FIO_RUST_ZSTD_FRAME_OK;
|
||||
}
|
||||
|
||||
let to_decode = read_size_hint.min(dstream_in_size());
|
||||
let loaded = unsafe { read_buffer_loaded(read_ctx) };
|
||||
if loaded < to_decode {
|
||||
let read_size = unsafe { AIO_ReadPool_fillBuffer(read_ctx, to_decode) };
|
||||
if read_size == 0 {
|
||||
unsafe { AIO_WritePool_releaseIoJob(write_job) };
|
||||
return FIO_RUST_ZSTD_FRAME_PREMATURE_END;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
unsafe extern "C" fn fio_zstd_reset(dctx: *mut c_void, reset: c_int) -> usize {
|
||||
unsafe {
|
||||
crate::zstd_decompress::ZSTD_DCtx_reset(
|
||||
dctx.cast::<crate::zstd_decompress::ZSTD_DCtx>(),
|
||||
reset,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
unsafe extern "C" fn fio_zstd_decompress(
|
||||
dctx: *mut c_void,
|
||||
output: *mut crate::zstd_decompress::ZSTD_outBuffer,
|
||||
input: *mut crate::zstd_decompress::ZSTD_inBuffer,
|
||||
) -> usize {
|
||||
unsafe {
|
||||
crate::zstd_decompress::ZSTD_decompressStream(
|
||||
dctx.cast::<crate::zstd_decompress::ZSTD_DStream>(),
|
||||
output,
|
||||
input,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
#[no_mangle]
|
||||
pub unsafe extern "C" fn FIO_rust_decompressZstdFrame(
|
||||
f_ctx: *mut c_void,
|
||||
dctx: *mut c_void,
|
||||
read_ctx: *mut ReadPoolCtx_t,
|
||||
write_ctx: *mut WritePoolCtx_t,
|
||||
src_file_name: *const c_char,
|
||||
already_decoded: u64,
|
||||
frame_size: *mut u64,
|
||||
zstd_error: *mut usize,
|
||||
progress: FIO_rust_frame_progress_fn,
|
||||
) -> c_int {
|
||||
unsafe {
|
||||
decompress_zstd_frame_with(
|
||||
f_ctx,
|
||||
dctx,
|
||||
read_ctx,
|
||||
write_ctx,
|
||||
src_file_name,
|
||||
already_decoded,
|
||||
frame_size,
|
||||
zstd_error,
|
||||
progress,
|
||||
fio_zstd_reset,
|
||||
fio_zstd_decompress,
|
||||
crate::zstd_decompress::ZSTD_DStreamInSize,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
unsafe fn read_buffer_ptr(ctx: *mut ReadPoolCtx_t) -> *const u8 {
|
||||
let context = ctx.cast::<u8>();
|
||||
let inner = unsafe { base_inner(context) };
|
||||
unsafe { read_at::<*const u8>(context, inner.layout.read_src_buffer) }
|
||||
}
|
||||
|
||||
#[inline]
|
||||
unsafe fn read_buffer_loaded(ctx: *mut ReadPoolCtx_t) -> usize {
|
||||
let context = ctx.cast::<u8>();
|
||||
let inner = unsafe { base_inner(context) };
|
||||
unsafe { read_at::<usize>(context, inner.layout.read_src_buffer_loaded) }
|
||||
}
|
||||
|
||||
impl PoolInner {
|
||||
unsafe fn all_jobs_available(&self) -> bool {
|
||||
let state = self.jobs.lock().unwrap_or_else(|error| error.into_inner());
|
||||
@@ -1851,4 +2028,343 @@ mod tests {
|
||||
assert!(run_pass_through(&[], 5, 3, async_io).is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(all(unix, feature = "compression"))]
|
||||
mod zstd_frame_tests {
|
||||
use super::*;
|
||||
|
||||
#[repr(C)]
|
||||
#[derive(Default)]
|
||||
struct ProgressState {
|
||||
calls: usize,
|
||||
last_decoded: u64,
|
||||
}
|
||||
|
||||
unsafe extern "C" fn record_progress(
|
||||
opaque: *mut c_void,
|
||||
source_name: *const c_char,
|
||||
decoded_size: u64,
|
||||
) {
|
||||
assert!(!opaque.is_null());
|
||||
assert!(!source_name.is_null());
|
||||
let state = unsafe { &mut *opaque.cast::<ProgressState>() };
|
||||
state.calls += 1;
|
||||
state.last_decoded = decoded_size;
|
||||
}
|
||||
|
||||
const MOCK_FRAME_HEADER_SIZE: usize = 5;
|
||||
const MOCK_DSTREAM_IN_SIZE: usize = 8;
|
||||
const MOCK_FRAME_MAGIC: u8 = 0xA5;
|
||||
|
||||
struct MockDecoder {
|
||||
header: [u8; MOCK_FRAME_HEADER_SIZE],
|
||||
header_len: usize,
|
||||
remaining: usize,
|
||||
}
|
||||
|
||||
impl MockDecoder {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
header: [0; MOCK_FRAME_HEADER_SIZE],
|
||||
header_len: 0,
|
||||
remaining: 0,
|
||||
}
|
||||
}
|
||||
|
||||
fn reset(&mut self) {
|
||||
self.header = [0; MOCK_FRAME_HEADER_SIZE];
|
||||
self.header_len = 0;
|
||||
self.remaining = 0;
|
||||
}
|
||||
}
|
||||
|
||||
fn encode_frame(input: &[u8]) -> Vec<u8> {
|
||||
assert!(u32::try_from(input.len()).is_ok());
|
||||
let mut frame = Vec::with_capacity(MOCK_FRAME_HEADER_SIZE + input.len());
|
||||
frame.push(MOCK_FRAME_MAGIC);
|
||||
frame.extend_from_slice(&(input.len() as u32).to_le_bytes());
|
||||
frame.extend_from_slice(input);
|
||||
frame
|
||||
}
|
||||
|
||||
unsafe extern "C" fn mock_reset(dctx: *mut c_void, _reset: c_int) -> usize {
|
||||
unsafe { (*dctx.cast::<MockDecoder>()).reset() };
|
||||
0
|
||||
}
|
||||
|
||||
unsafe extern "C" fn mock_decompress(
|
||||
dctx: *mut c_void,
|
||||
output: *mut crate::zstd_decompress::ZSTD_outBuffer,
|
||||
input: *mut crate::zstd_decompress::ZSTD_inBuffer,
|
||||
) -> usize {
|
||||
let decoder = unsafe { &mut *dctx.cast::<MockDecoder>() };
|
||||
let input = unsafe { &mut *input };
|
||||
let output = unsafe { &mut *output };
|
||||
let call_start = input.pos;
|
||||
|
||||
while decoder.header_len < MOCK_FRAME_HEADER_SIZE && input.pos < input.size {
|
||||
decoder.header[decoder.header_len] =
|
||||
unsafe { input.src.cast::<u8>().add(input.pos).read() };
|
||||
decoder.header_len += 1;
|
||||
input.pos += 1;
|
||||
}
|
||||
if decoder.header_len < MOCK_FRAME_HEADER_SIZE {
|
||||
return (MOCK_FRAME_HEADER_SIZE - decoder.header_len).max(1);
|
||||
}
|
||||
if decoder.header[0] != MOCK_FRAME_MAGIC {
|
||||
input.pos = call_start;
|
||||
return crate::errors::ERROR(crate::errors::ZstdErrorCode::Generic);
|
||||
}
|
||||
if decoder.remaining == 0 {
|
||||
let length = u32::from_le_bytes([
|
||||
decoder.header[1],
|
||||
decoder.header[2],
|
||||
decoder.header[3],
|
||||
decoder.header[4],
|
||||
]);
|
||||
decoder.remaining = length as usize;
|
||||
if decoder.remaining == 0 {
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
let available = input.size - input.pos;
|
||||
let writable = output.size - output.pos;
|
||||
let copied = decoder.remaining.min(available).min(writable);
|
||||
if copied != 0 {
|
||||
unsafe {
|
||||
ptr::copy_nonoverlapping(
|
||||
input.src.cast::<u8>().add(input.pos),
|
||||
output.dst.cast::<u8>().add(output.pos),
|
||||
copied,
|
||||
);
|
||||
}
|
||||
input.pos += copied;
|
||||
output.pos += copied;
|
||||
decoder.remaining -= copied;
|
||||
}
|
||||
if decoder.remaining == 0 {
|
||||
0
|
||||
} else {
|
||||
decoder.remaining.min(MOCK_DSTREAM_IN_SIZE)
|
||||
}
|
||||
}
|
||||
|
||||
extern "C" fn mock_dstream_in_size() -> usize {
|
||||
MOCK_DSTREAM_IN_SIZE
|
||||
}
|
||||
|
||||
struct FrameHarness {
|
||||
_prefs: Box<FIO_prefs_t>,
|
||||
_decoder: Box<MockDecoder>,
|
||||
read_ctx: *mut ReadPoolCtx_t,
|
||||
write_ctx: *mut WritePoolCtx_t,
|
||||
dctx: *mut c_void,
|
||||
output_file: *mut libc::FILE,
|
||||
}
|
||||
|
||||
impl FrameHarness {
|
||||
fn new(input: &[u8], read_buffer_size: usize, async_io: c_int) -> Self {
|
||||
let mut prefs = Box::new(test_prefs(async_io));
|
||||
prefs.testMode = 0;
|
||||
prefs.sparseFileSupport = 0;
|
||||
|
||||
let input_file = unsafe { libc::tmpfile() };
|
||||
assert!(!input_file.is_null());
|
||||
assert_eq!(
|
||||
unsafe { libc::fwrite(input.as_ptr().cast(), 1, input.len(), input_file) },
|
||||
input.len()
|
||||
);
|
||||
assert_eq!(unsafe { libc::fflush(input_file) }, 0);
|
||||
assert_eq!(unsafe { libc::fseek(input_file, 0, libc::SEEK_SET) }, 0);
|
||||
|
||||
let output_file = unsafe { libc::tmpfile() };
|
||||
assert!(!output_file.is_null());
|
||||
let read_ctx = unsafe { AIO_ReadPool_create(&*prefs, read_buffer_size) };
|
||||
let write_ctx = unsafe { AIO_WritePool_create(&*prefs, 128 * 1024) };
|
||||
assert!(!read_ctx.is_null());
|
||||
assert!(!write_ctx.is_null());
|
||||
unsafe {
|
||||
AIO_ReadPool_setFile(read_ctx, input_file);
|
||||
AIO_WritePool_setFile(write_ctx, output_file);
|
||||
}
|
||||
let mut decoder = Box::new(MockDecoder::new());
|
||||
let dctx = (&mut *decoder as *mut MockDecoder).cast::<c_void>();
|
||||
|
||||
Self {
|
||||
_prefs: prefs,
|
||||
_decoder: decoder,
|
||||
read_ctx,
|
||||
write_ctx,
|
||||
dctx,
|
||||
output_file,
|
||||
}
|
||||
}
|
||||
|
||||
fn decompress(
|
||||
&mut self,
|
||||
already_decoded: u64,
|
||||
progress_state: Option<&mut ProgressState>,
|
||||
) -> (c_int, u64, usize) {
|
||||
let mut frame_size = 0;
|
||||
let mut zstd_error = 0;
|
||||
let progress_context = progress_state
|
||||
.map(|state| state as *mut ProgressState as *mut c_void)
|
||||
.unwrap_or(ptr::null_mut());
|
||||
let progress_callback = if progress_context.is_null() {
|
||||
None
|
||||
} else {
|
||||
Some(record_progress as unsafe extern "C" fn(*mut c_void, *const c_char, u64))
|
||||
};
|
||||
let status = unsafe {
|
||||
decompress_zstd_frame_with(
|
||||
progress_context,
|
||||
self.dctx,
|
||||
self.read_ctx,
|
||||
self.write_ctx,
|
||||
c"frame-test.zst".as_ptr(),
|
||||
already_decoded,
|
||||
&mut frame_size,
|
||||
&mut zstd_error,
|
||||
progress_callback,
|
||||
mock_reset,
|
||||
mock_decompress,
|
||||
mock_dstream_in_size,
|
||||
)
|
||||
};
|
||||
(status, frame_size, zstd_error)
|
||||
}
|
||||
|
||||
fn unread_bytes(&self) -> Vec<u8> {
|
||||
let loaded = unsafe { read_buffer_loaded(self.read_ctx) };
|
||||
let source = unsafe { read_buffer_ptr(self.read_ctx) };
|
||||
unsafe { std::slice::from_raw_parts(source, loaded) }.to_vec()
|
||||
}
|
||||
|
||||
fn output_bytes(&self) -> Vec<u8> {
|
||||
assert_eq!(
|
||||
unsafe { libc::fseek(self.output_file, 0, libc::SEEK_END) },
|
||||
0
|
||||
);
|
||||
let size = unsafe { libc::ftell(self.output_file) };
|
||||
assert!(size >= 0);
|
||||
let size = usize::try_from(size).unwrap();
|
||||
assert_eq!(
|
||||
unsafe { libc::fseek(self.output_file, 0, libc::SEEK_SET) },
|
||||
0
|
||||
);
|
||||
let mut output = vec![0_u8; size];
|
||||
if size != 0 {
|
||||
assert_eq!(
|
||||
unsafe {
|
||||
libc::fread(output.as_mut_ptr().cast(), 1, size, self.output_file)
|
||||
},
|
||||
size
|
||||
);
|
||||
}
|
||||
output
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for FrameHarness {
|
||||
fn drop(&mut self) {
|
||||
unsafe {
|
||||
AIO_WritePool_free(self.write_ctx);
|
||||
AIO_ReadPool_free(self.read_ctx);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decompresses_frame_streams_output_and_reports_progress() {
|
||||
let input: Vec<u8> = (0..300_000)
|
||||
.map(|index| (index as u32).wrapping_mul(37).rotate_left(5) as u8)
|
||||
.collect();
|
||||
let frame = encode_frame(&input);
|
||||
|
||||
for async_io in [0, 1] {
|
||||
let mut harness = FrameHarness::new(&frame, 97, async_io);
|
||||
let mut progress = ProgressState::default();
|
||||
let (status, frame_size, zstd_error) = harness.decompress(0, Some(&mut progress));
|
||||
assert_eq!(status, FIO_RUST_ZSTD_FRAME_OK);
|
||||
assert_eq!(frame_size, input.len() as u64);
|
||||
assert_eq!(zstd_error, 0);
|
||||
assert!(progress.calls > 1);
|
||||
assert!(progress.last_decoded < input.len() as u64);
|
||||
assert_eq!(harness.output_bytes(), input);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn progress_uses_the_pre_increment_frame_size() {
|
||||
let input = b"one streaming output block";
|
||||
let frame = encode_frame(input);
|
||||
let mut harness = FrameHarness::new(&frame, frame.len() + 1, 0);
|
||||
let mut progress = ProgressState::default();
|
||||
|
||||
let (status, frame_size, zstd_error) = harness.decompress(37, Some(&mut progress));
|
||||
assert_eq!(status, FIO_RUST_ZSTD_FRAME_OK);
|
||||
assert_eq!(frame_size, input.len() as u64);
|
||||
assert_eq!(zstd_error, 0);
|
||||
assert_eq!(progress.calls, 1);
|
||||
assert_eq!(progress.last_decoded, 37);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decompresses_concatenated_frames_without_crossing_boundary() {
|
||||
let first = b"first frame ".repeat(20_000);
|
||||
let second = b"second frame ".repeat(20_000);
|
||||
let first_frame = encode_frame(&first);
|
||||
let second_frame = encode_frame(&second);
|
||||
let mut stream = first_frame.clone();
|
||||
stream.extend_from_slice(&second_frame);
|
||||
|
||||
let mut harness = FrameHarness::new(&stream, stream.len(), 1);
|
||||
let mut progress = ProgressState::default();
|
||||
let (status, frame_size, zstd_error) = harness.decompress(0, Some(&mut progress));
|
||||
assert_eq!(status, FIO_RUST_ZSTD_FRAME_OK);
|
||||
assert_eq!(frame_size, first.len() as u64);
|
||||
assert_eq!(zstd_error, 0);
|
||||
assert_eq!(harness.unread_bytes(), second_frame);
|
||||
assert_eq!(harness.output_bytes(), first);
|
||||
|
||||
let (status, frame_size, zstd_error) =
|
||||
harness.decompress(first.len() as u64, Some(&mut progress));
|
||||
assert_eq!(status, FIO_RUST_ZSTD_FRAME_OK);
|
||||
assert_eq!(frame_size, second.len() as u64);
|
||||
assert_eq!(zstd_error, 0);
|
||||
assert!(harness.unread_bytes().is_empty());
|
||||
assert!(progress.last_decoded >= first.len() as u64);
|
||||
assert!(progress.last_decoded < (first.len() + second.len()) as u64);
|
||||
|
||||
let mut expected = first;
|
||||
expected.extend_from_slice(&second);
|
||||
assert_eq!(harness.output_bytes(), expected);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decoder_error_preserves_the_current_input_buffer() {
|
||||
let input = b"decoder error input";
|
||||
let mut frame = encode_frame(input);
|
||||
frame[0] ^= 1;
|
||||
let mut harness = FrameHarness::new(&frame, frame.len() + 1, 0);
|
||||
|
||||
let (status, frame_size, zstd_error) = harness.decompress(0, None);
|
||||
assert_eq!(status, FIO_RUST_ZSTD_FRAME_DECODING_ERROR);
|
||||
assert_eq!(frame_size, 0);
|
||||
assert_ne!(zstd_error, 0);
|
||||
assert_eq!(harness.unread_bytes(), frame);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncated_frame_reports_premature_end() {
|
||||
let frame = encode_frame(&b"truncated frame".repeat(10_000));
|
||||
let truncated = &frame[..frame.len() - 1];
|
||||
let mut harness = FrameHarness::new(truncated, truncated.len() + 1, 0);
|
||||
|
||||
let (status, _frame_size, zstd_error) = harness.decompress(0, None);
|
||||
assert_eq!(status, FIO_RUST_ZSTD_FRAME_PREMATURE_END);
|
||||
assert_eq!(zstd_error, 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user