feat(cli): move optional decompression loops into Rust

Move gzip, xz/LZMA, and LZ4 frame I/O loops into Rust projections while
keeping zlib, liblzma, LZ4F, asynchronous pool operations, diagnostics, and
private C handles in C callbacks.  Preserve trailing-input accounting and
format-dispatch behavior, and stop faulty LZ4 callbacks from producing an
unbounded no-progress output loop while retaining the codec error value.

Test Plan:
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml --lib lz4_decompress_projection_reports_errors_and_unfinished_input -- --test-threads=1 (1 passed)
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --lib -- -D warnings (passed)
- ulimit -v 41943040; make -B -C tests -j1 test-cli-tests (41 passed)
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1 (507 passed before final error-value assertion)
This commit is contained in:
2026-07-19 08:20:46 +02:00
parent eb1aa3b6dc
commit 7282ec4046
2 changed files with 1643 additions and 261 deletions
+1221
View File
@@ -79,6 +79,140 @@ pub struct FIO_rust_decompress_callbacks_t {
pub finish: Option<FIO_rust_decompress_finish_fn>,
}
pub type FIO_rust_decompress_read_fill_fn =
unsafe extern "C" fn(*mut c_void, usize, *mut *const u8, *mut usize);
pub type FIO_rust_decompress_read_consume_fn = unsafe extern "C" fn(*mut c_void, usize);
pub type FIO_rust_decompress_write_acquire_fn =
unsafe extern "C" fn(*mut c_void, *mut *mut c_void, *mut *mut u8, *mut usize);
pub type FIO_rust_decompress_write_enqueue_fn =
unsafe extern "C" fn(*mut c_void, *mut *mut c_void, usize, *mut *mut u8, *mut usize);
pub type FIO_rust_decompress_write_release_fn = unsafe extern "C" fn(*mut c_void, *mut c_void);
pub type FIO_rust_decompress_sparse_write_end_fn = unsafe extern "C" fn(*mut c_void);
/// The optional-format leaves use the same opaque asynchronous I/O projection.
/// `read_fill(0, ...)` returns the bytes already staged by the mixed-format
/// dispatcher; non-zero requests refill the C-owned read pool after the Rust
/// loop has consumed the current input.
#[repr(C)]
pub struct FIO_rust_decompress_io_projection_t {
pub read_opaque: *mut c_void,
pub write_opaque: *mut c_void,
pub read_buffer_size: usize,
pub read_fill: Option<FIO_rust_decompress_read_fill_fn>,
pub read_consume: Option<FIO_rust_decompress_read_consume_fn>,
pub write_acquire: Option<FIO_rust_decompress_write_acquire_fn>,
pub write_enqueue: Option<FIO_rust_decompress_write_enqueue_fn>,
pub write_release: Option<FIO_rust_decompress_write_release_fn>,
pub sparse_write_end: Option<FIO_rust_decompress_sparse_write_end_fn>,
}
pub const FIO_RUST_GZIP_DECOMPRESS_OK: c_int = 0;
pub const FIO_RUST_GZIP_DECOMPRESS_INIT_ERROR: c_int = 1;
pub const FIO_RUST_GZIP_DECOMPRESS_BUF_ERROR: c_int = 2;
pub const FIO_RUST_GZIP_DECOMPRESS_INFLATE_ERROR: c_int = 3;
pub const FIO_RUST_GZIP_DECOMPRESS_END_ERROR: c_int = 4;
pub const FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION: c_int = 5;
const FIO_RUST_GZIP_DECOMPRESS_Z_OK: c_int = 0;
const FIO_RUST_GZIP_DECOMPRESS_Z_STREAM_END: c_int = 1;
const FIO_RUST_GZIP_DECOMPRESS_Z_BUF_ERROR: c_int = -5;
const FIO_RUST_GZIP_DECOMPRESS_Z_NO_FLUSH: c_int = 0;
const FIO_RUST_GZIP_DECOMPRESS_Z_FINISH: c_int = 4;
pub type FIO_rust_gzip_decompress_init_fn = unsafe extern "C" fn(*mut c_void) -> c_int;
pub type FIO_rust_gzip_decompress_inflate_fn = unsafe extern "C" fn(
*mut c_void,
*const u8,
usize,
*mut u8,
usize,
c_int,
*mut usize,
*mut usize,
) -> c_int;
pub type FIO_rust_gzip_decompress_end_fn = unsafe extern "C" fn(*mut c_void) -> c_int;
/// Rust owns the gzip read/inflate/output loop. The C side retains only the
/// zlib stream and its calls, plus the opaque asynchronous pool operations.
#[repr(C)]
pub struct FIO_rust_gzip_decompress_projection_t {
pub io: FIO_rust_decompress_io_projection_t,
pub zlib_opaque: *mut c_void,
pub zlib_init: Option<FIO_rust_gzip_decompress_init_fn>,
pub zlib_inflate: Option<FIO_rust_gzip_decompress_inflate_fn>,
pub zlib_end: Option<FIO_rust_gzip_decompress_end_fn>,
}
pub const FIO_RUST_LZMA_DECOMPRESS_OK: c_int = 0;
pub const FIO_RUST_LZMA_DECOMPRESS_INIT_ERROR: c_int = 1;
pub const FIO_RUST_LZMA_DECOMPRESS_BUF_ERROR: c_int = 2;
pub const FIO_RUST_LZMA_DECOMPRESS_CODE_ERROR: c_int = 3;
pub const FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION: c_int = 4;
const FIO_RUST_LZMA_DECOMPRESS_OK_CODE: c_int = 0;
const FIO_RUST_LZMA_DECOMPRESS_STREAM_END: c_int = 1;
const FIO_RUST_LZMA_DECOMPRESS_BUF_ERROR_CODE: c_int = 10;
const FIO_RUST_LZMA_DECOMPRESS_RUN: c_int = 0;
const FIO_RUST_LZMA_DECOMPRESS_FINISH: c_int = 3;
pub type FIO_rust_lzma_decompress_init_fn = unsafe extern "C" fn(*mut c_void, c_int) -> c_int;
pub type FIO_rust_lzma_decompress_code_fn = unsafe extern "C" fn(
*mut c_void,
*const u8,
usize,
*mut u8,
usize,
c_int,
*mut usize,
*mut usize,
) -> c_int;
pub type FIO_rust_lzma_decompress_end_fn = unsafe extern "C" fn(*mut c_void);
/// Rust owns the xz/LZMA read/code/output loop while liblzma's stream remains
/// an opaque C-owned handle.
#[repr(C)]
pub struct FIO_rust_lzma_decompress_projection_t {
pub io: FIO_rust_decompress_io_projection_t,
pub lzma_opaque: *mut c_void,
pub lzma_init: Option<FIO_rust_lzma_decompress_init_fn>,
pub lzma_code: Option<FIO_rust_lzma_decompress_code_fn>,
pub lzma_end: Option<FIO_rust_lzma_decompress_end_fn>,
}
pub const FIO_RUST_LZ4_DECOMPRESS_OK: c_int = 0;
pub const FIO_RUST_LZ4_DECOMPRESS_CREATE_ERROR: c_int = 1;
pub const FIO_RUST_LZ4_DECOMPRESS_CODE_ERROR: c_int = 2;
pub const FIO_RUST_LZ4_DECOMPRESS_UNFINISHED: c_int = 3;
pub const FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION: c_int = 4;
pub type FIO_rust_lz4_decompress_create_fn =
unsafe extern "C" fn(*mut c_void, c_uint, *mut usize) -> c_int;
pub type FIO_rust_lz4_decompress_code_fn = unsafe extern "C" fn(
*mut c_void,
*mut u8,
*mut usize,
*const u8,
*mut usize,
*mut usize,
) -> c_int;
pub type FIO_rust_lz4_decompress_free_fn = unsafe extern "C" fn(*mut c_void);
pub type FIO_rust_lz4_decompress_progress_fn = unsafe extern "C" fn(*mut c_void, u64);
/// Rust owns the LZ4 frame loop. LZ4F's context, error values, and calls are
/// kept behind callbacks so no optional-library or private-handle layout is
/// part of the Rust/C projection.
#[repr(C)]
pub struct FIO_rust_lz4_decompress_projection_t {
pub io: FIO_rust_decompress_io_projection_t,
pub codec_opaque: *mut c_void,
pub progress_opaque: *mut c_void,
pub version: c_uint,
pub create: Option<FIO_rust_lz4_decompress_create_fn>,
pub code: Option<FIO_rust_lz4_decompress_code_fn>,
pub free_context: Option<FIO_rust_lz4_decompress_free_fn>,
pub progress: Option<FIO_rust_lz4_decompress_progress_fn>,
}
pub const FIO_RUST_COMPRESS_OK: c_int = 0;
pub const FIO_RUST_COMPRESS_GZIP_UNSUPPORTED: c_int = 1;
pub const FIO_RUST_COMPRESS_LZMA_UNSUPPORTED: c_int = 2;
@@ -2686,6 +2820,512 @@ pub unsafe extern "C" fn FIO_rust_decompressZstdFrames(
}
}
/// Decompresses one gzip member through a C-owned zlib stream and the common
/// asynchronous I/O projection. Input is consumed immediately after each
/// inflate call; this is equivalent to the original C leaf's final
/// `avail_in` accounting while keeping any following member or format in the
/// read pool.
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_decompressGzipFrame(
projection: *const FIO_rust_gzip_decompress_projection_t,
frame_size: *mut u64,
zlib_result: *mut c_int,
) -> c_int {
assert!(!projection.is_null());
assert!(!frame_size.is_null());
assert!(!zlib_result.is_null());
unsafe {
*frame_size = 0;
*zlib_result = FIO_RUST_GZIP_DECOMPRESS_Z_OK;
}
let projection = unsafe { &*projection };
if projection.io.read_buffer_size == 0 {
return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION;
}
let Some(read_fill) = projection.io.read_fill else {
return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION;
};
let Some(read_consume) = projection.io.read_consume else {
return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION;
};
let Some(write_acquire) = projection.io.write_acquire else {
return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION;
};
let Some(write_enqueue) = projection.io.write_enqueue else {
return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION;
};
let Some(write_release) = projection.io.write_release else {
return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION;
};
let Some(sparse_write_end) = projection.io.sparse_write_end else {
return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION;
};
let Some(zlib_init) = projection.zlib_init else {
return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION;
};
let Some(zlib_inflate) = projection.zlib_inflate else {
return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION;
};
let Some(zlib_end) = projection.zlib_end else {
return FIO_RUST_GZIP_DECOMPRESS_INVALID_PROJECTION;
};
let init_result = unsafe { zlib_init(projection.zlib_opaque) };
if init_result != FIO_RUST_GZIP_DECOMPRESS_Z_OK {
unsafe { *zlib_result = init_result };
return FIO_RUST_GZIP_DECOMPRESS_INIT_ERROR;
}
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.io.write_opaque,
&mut job,
&mut output,
&mut output_size,
);
}
assert!(!job.is_null());
assert!(!output.is_null() || output_size == 0);
let mut input = ptr::null();
let mut input_size = 0_usize;
unsafe {
read_fill(projection.io.read_opaque, 0, &mut input, &mut input_size);
}
let mut flush = FIO_RUST_GZIP_DECOMPRESS_Z_NO_FLUSH;
let mut decoded = 0_u64;
let mut status = FIO_RUST_GZIP_DECOMPRESS_OK;
loop {
if input_size == 0 {
unsafe {
read_fill(
projection.io.read_opaque,
projection.io.read_buffer_size,
&mut input,
&mut input_size,
);
}
if input_size == 0 {
flush = FIO_RUST_GZIP_DECOMPRESS_Z_FINISH;
}
}
let mut consumed = 0_usize;
let mut produced = 0_usize;
let result = unsafe {
zlib_inflate(
projection.zlib_opaque,
input,
input_size,
output,
output_size,
flush,
&mut consumed,
&mut produced,
)
};
assert!(consumed <= input_size);
assert!(produced <= output_size);
unsafe { read_consume(projection.io.read_opaque, consumed) };
if consumed != 0 {
input = unsafe { input.add(consumed) };
}
input_size -= consumed;
if result == FIO_RUST_GZIP_DECOMPRESS_Z_BUF_ERROR {
unsafe { *zlib_result = result };
status = FIO_RUST_GZIP_DECOMPRESS_BUF_ERROR;
break;
}
if result != FIO_RUST_GZIP_DECOMPRESS_Z_OK
&& result != FIO_RUST_GZIP_DECOMPRESS_Z_STREAM_END
{
unsafe { *zlib_result = result };
status = FIO_RUST_GZIP_DECOMPRESS_INFLATE_ERROR;
break;
}
if produced != 0 {
unsafe {
write_enqueue(
projection.io.write_opaque,
&mut job,
produced,
&mut output,
&mut output_size,
);
}
decoded = decoded.wrapping_add(produced as u64);
}
if result == FIO_RUST_GZIP_DECOMPRESS_Z_STREAM_END {
break;
}
}
let end_result = unsafe { zlib_end(projection.zlib_opaque) };
if status == FIO_RUST_GZIP_DECOMPRESS_OK && end_result != FIO_RUST_GZIP_DECOMPRESS_Z_OK {
unsafe { *zlib_result = end_result };
status = FIO_RUST_GZIP_DECOMPRESS_END_ERROR;
}
unsafe {
write_release(projection.io.write_opaque, job);
sparse_write_end(projection.io.write_opaque);
}
if status == FIO_RUST_GZIP_DECOMPRESS_OK {
unsafe { *frame_size = decoded };
}
status
}
/// Decompresses one xz or plain-LZMA stream through a C-owned liblzma
/// stream. The action switches to `LZMA_FINISH` only after the read pool
/// reaches EOF, matching the legacy C loop and preserving unread trailing
/// bytes after `LZMA_STREAM_END`.
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_decompressLzmaFrame(
projection: *const FIO_rust_lzma_decompress_projection_t,
plain_lzma: c_int,
frame_size: *mut u64,
lzma_result: *mut c_int,
) -> c_int {
assert!(!projection.is_null());
assert!(!frame_size.is_null());
assert!(!lzma_result.is_null());
unsafe {
*frame_size = 0;
*lzma_result = FIO_RUST_LZMA_DECOMPRESS_OK_CODE;
}
let projection = unsafe { &*projection };
if projection.io.read_buffer_size == 0 {
return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION;
}
let Some(read_fill) = projection.io.read_fill else {
return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION;
};
let Some(read_consume) = projection.io.read_consume else {
return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION;
};
let Some(write_acquire) = projection.io.write_acquire else {
return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION;
};
let Some(write_enqueue) = projection.io.write_enqueue else {
return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION;
};
let Some(write_release) = projection.io.write_release else {
return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION;
};
let Some(sparse_write_end) = projection.io.sparse_write_end else {
return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION;
};
let Some(lzma_init) = projection.lzma_init else {
return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION;
};
let Some(lzma_code) = projection.lzma_code else {
return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION;
};
let Some(lzma_end) = projection.lzma_end else {
return FIO_RUST_LZMA_DECOMPRESS_INVALID_PROJECTION;
};
let init_result = unsafe { lzma_init(projection.lzma_opaque, plain_lzma) };
if init_result != FIO_RUST_LZMA_DECOMPRESS_OK_CODE {
unsafe { *lzma_result = init_result };
return FIO_RUST_LZMA_DECOMPRESS_INIT_ERROR;
}
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.io.write_opaque,
&mut job,
&mut output,
&mut output_size,
);
}
assert!(!job.is_null());
assert!(!output.is_null() || output_size == 0);
let mut input = ptr::null();
let mut input_size = 0_usize;
unsafe {
read_fill(projection.io.read_opaque, 0, &mut input, &mut input_size);
}
let mut action = FIO_RUST_LZMA_DECOMPRESS_RUN;
let mut decoded = 0_u64;
let mut status = FIO_RUST_LZMA_DECOMPRESS_OK;
loop {
if input_size == 0 {
unsafe {
read_fill(
projection.io.read_opaque,
projection.io.read_buffer_size,
&mut input,
&mut input_size,
);
}
if input_size == 0 {
action = FIO_RUST_LZMA_DECOMPRESS_FINISH;
}
}
let mut consumed = 0_usize;
let mut produced = 0_usize;
let result = unsafe {
lzma_code(
projection.lzma_opaque,
input,
input_size,
output,
output_size,
action,
&mut consumed,
&mut produced,
)
};
assert!(consumed <= input_size);
assert!(produced <= output_size);
unsafe { read_consume(projection.io.read_opaque, consumed) };
if consumed != 0 {
input = unsafe { input.add(consumed) };
}
input_size -= consumed;
if result == FIO_RUST_LZMA_DECOMPRESS_BUF_ERROR_CODE {
unsafe { *lzma_result = result };
status = FIO_RUST_LZMA_DECOMPRESS_BUF_ERROR;
break;
}
if result != FIO_RUST_LZMA_DECOMPRESS_OK_CODE
&& result != FIO_RUST_LZMA_DECOMPRESS_STREAM_END
{
unsafe { *lzma_result = result };
status = FIO_RUST_LZMA_DECOMPRESS_CODE_ERROR;
break;
}
if produced != 0 {
unsafe {
write_enqueue(
projection.io.write_opaque,
&mut job,
produced,
&mut output,
&mut output_size,
);
}
decoded = decoded.wrapping_add(produced as u64);
}
if result == FIO_RUST_LZMA_DECOMPRESS_STREAM_END {
break;
}
}
unsafe { lzma_end(projection.lzma_opaque) };
unsafe {
write_release(projection.io.write_opaque, job);
sparse_write_end(projection.io.write_opaque);
}
if status == FIO_RUST_LZMA_DECOMPRESS_OK {
unsafe { *frame_size = decoded };
}
status
}
/// Decompresses one LZ4 frame while preserving LZ4F's recommended input
/// request and full-output-buffer retry protocol. A successful frame ends at
/// the first zero `next_to_load`; any EOF while that hint remains non-zero is
/// reported as an unfinished stream.
#[no_mangle]
pub unsafe extern "C" fn FIO_rust_decompressLz4Frame(
projection: *const FIO_rust_lz4_decompress_projection_t,
frame_size: *mut u64,
lz4_result: *mut usize,
) -> c_int {
assert!(!projection.is_null());
assert!(!frame_size.is_null());
assert!(!lz4_result.is_null());
unsafe {
*frame_size = 0;
*lz4_result = 0;
}
let projection = unsafe { &*projection };
let Some(read_fill) = projection.io.read_fill else {
return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION;
};
let Some(read_consume) = projection.io.read_consume else {
return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION;
};
let Some(write_acquire) = projection.io.write_acquire else {
return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION;
};
let Some(write_enqueue) = projection.io.write_enqueue else {
return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION;
};
let Some(write_release) = projection.io.write_release else {
return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION;
};
let Some(sparse_write_end) = projection.io.sparse_write_end else {
return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION;
};
let Some(create) = projection.create else {
return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION;
};
let Some(code) = projection.code else {
return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION;
};
let Some(free_context) = projection.free_context else {
return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION;
};
let Some(progress) = projection.progress else {
return FIO_RUST_LZ4_DECOMPRESS_INVALID_PROJECTION;
};
let mut create_result = 0_usize;
let create_status = unsafe {
create(
projection.codec_opaque,
projection.version,
&mut create_result,
)
};
if create_status != 0 {
unsafe { *lz4_result = create_result };
return FIO_RUST_LZ4_DECOMPRESS_CREATE_ERROR;
}
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.io.write_opaque,
&mut job,
&mut output,
&mut output_size,
);
}
assert!(!job.is_null());
assert!(!output.is_null() || output_size == 0);
let mut next_to_load = 4_usize;
let mut decoded = 0_u64;
let mut status = FIO_RUST_LZ4_DECOMPRESS_OK;
while next_to_load != 0 {
let mut input = ptr::null();
let mut loaded = 0_usize;
unsafe {
read_fill(
projection.io.read_opaque,
next_to_load,
&mut input,
&mut loaded,
);
}
if loaded == 0 {
break;
}
let mut pos = 0_usize;
let mut full_buffer_decoded = false;
let mut unchanged_input_calls = 0_usize;
while pos < loaded || full_buffer_decoded {
let previous_next_to_load = next_to_load;
let mut produced = output_size;
let mut remaining = loaded - pos;
let mut next = 0_usize;
let result = unsafe {
code(
projection.codec_opaque,
output,
&mut produced,
input.add(pos),
&mut remaining,
&mut next,
)
};
if result != 0 {
/* The C adapter returns a boolean failure status but keeps
* LZ4F's actual error code in the next-input slot. */
unsafe { *lz4_result = next };
status = FIO_RUST_LZ4_DECOMPRESS_CODE_ERROR;
next_to_load = 0;
break;
}
assert!(produced <= output_size);
assert!(remaining <= loaded - pos);
pos += remaining;
next_to_load = next;
full_buffer_decoded = produced == output_size;
if produced != 0 {
unsafe {
write_enqueue(
projection.io.write_opaque,
&mut job,
produced,
&mut output,
&mut output_size,
);
}
decoded = decoded.wrapping_add(produced as u64);
unsafe { progress(projection.progress_opaque, decoded) };
}
if next_to_load == 0 {
break;
}
/* A codec must eventually consume input or change its next-input
* request. Without this guard, a faulty callback can keep
* producing output while receiving an empty input slice forever,
* which would turn an ordinary decode error into an unbounded
* output/memory loop. */
if remaining == 0 && next == previous_next_to_load {
unchanged_input_calls += 1;
if unchanged_input_calls >= 2 {
status = FIO_RUST_LZ4_DECOMPRESS_UNFINISHED;
break;
}
} else {
unchanged_input_calls = 0;
}
}
unsafe { read_consume(projection.io.read_opaque, pos) };
if status != FIO_RUST_LZ4_DECOMPRESS_OK {
break;
}
}
if status == FIO_RUST_LZ4_DECOMPRESS_OK && next_to_load != 0 {
status = FIO_RUST_LZ4_DECOMPRESS_UNFINISHED;
}
unsafe {
free_context(projection.codec_opaque);
write_release(projection.io.write_opaque, job);
sparse_write_end(projection.io.write_opaque);
}
if status == FIO_RUST_LZ4_DECOMPRESS_OK {
unsafe { *frame_size = decoded };
}
status
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum DecompressionFormat {
Zstd,
@@ -4183,6 +4823,587 @@ mod tests {
}
}
#[derive(Default)]
struct OptionalDecompressIoState {
input: Vec<u8>,
input_pos: usize,
output: [u8; 4],
fill_requests: Vec<usize>,
consumed: Vec<usize>,
enqueue_sizes: Vec<usize>,
output_chunks: Vec<Vec<u8>>,
acquire_calls: usize,
release_calls: usize,
sparse_end_calls: usize,
}
impl OptionalDecompressIoState {
fn new(input: &[u8]) -> Self {
Self {
input: input.to_vec(),
..Self::default()
}
}
}
unsafe extern "C" fn optional_decompress_read_fill(
opaque: *mut c_void,
requested: usize,
buffer: *mut *const u8,
loaded: *mut usize,
) {
let state = unsafe { &mut *opaque.cast::<OptionalDecompressIoState>() };
state.fill_requests.push(requested);
let available = state.input.len() - state.input_pos;
let amount = if requested == 0 {
available
} else {
available.min(requested)
};
unsafe {
*buffer = state.input.as_ptr().add(state.input_pos);
*loaded = amount;
}
}
unsafe extern "C" fn optional_decompress_read_consume(opaque: *mut c_void, amount: usize) {
let state = unsafe { &mut *opaque.cast::<OptionalDecompressIoState>() };
assert!(amount <= state.input.len() - state.input_pos);
state.input_pos += amount;
state.consumed.push(amount);
}
unsafe extern "C" fn optional_decompress_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::<OptionalDecompressIoState>() };
state.acquire_calls += 1;
unsafe {
*job = opaque;
*buffer = state.output.as_mut_ptr();
*buffer_size = state.output.len();
}
}
unsafe extern "C" fn optional_decompress_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::<OptionalDecompressIoState>() };
assert_eq!(unsafe { *job }, opaque);
assert!(used <= state.output.len());
state.enqueue_sizes.push(used);
state.output_chunks.push(state.output[..used].to_vec());
unsafe {
*buffer = state.output.as_mut_ptr();
*buffer_size = state.output.len();
}
}
unsafe extern "C" fn optional_decompress_write_release(opaque: *mut c_void, job: *mut c_void) {
let state = unsafe { &mut *opaque.cast::<OptionalDecompressIoState>() };
assert_eq!(job, opaque);
state.release_calls += 1;
}
unsafe extern "C" fn optional_decompress_sparse_end(opaque: *mut c_void) {
let state = unsafe { &mut *opaque.cast::<OptionalDecompressIoState>() };
state.sparse_end_calls += 1;
}
fn optional_decompress_io(
state: &mut OptionalDecompressIoState,
read_buffer_size: usize,
) -> FIO_rust_decompress_io_projection_t {
let opaque = (state as *mut OptionalDecompressIoState).cast::<c_void>();
FIO_rust_decompress_io_projection_t {
read_opaque: opaque,
write_opaque: opaque,
read_buffer_size,
read_fill: Some(optional_decompress_read_fill),
read_consume: Some(optional_decompress_read_consume),
write_acquire: Some(optional_decompress_write_acquire),
write_enqueue: Some(optional_decompress_write_enqueue),
write_release: Some(optional_decompress_write_release),
sparse_write_end: Some(optional_decompress_sparse_end),
}
}
struct GzipDecompressState {
io: OptionalDecompressIoState,
init_result: c_int,
inflate_calls: usize,
stop_after_calls: usize,
finish_result: c_int,
end_result: c_int,
flushes: Vec<c_int>,
end_calls: usize,
}
impl GzipDecompressState {
fn new(input: &[u8]) -> Self {
Self {
io: OptionalDecompressIoState::new(input),
init_result: FIO_RUST_GZIP_DECOMPRESS_Z_OK,
inflate_calls: 0,
stop_after_calls: 0,
finish_result: FIO_RUST_GZIP_DECOMPRESS_Z_STREAM_END,
end_result: FIO_RUST_GZIP_DECOMPRESS_Z_OK,
flushes: Vec::new(),
end_calls: 0,
}
}
}
unsafe extern "C" fn gzip_decompress_test_init(opaque: *mut c_void) -> c_int {
unsafe { (*opaque.cast::<GzipDecompressState>()).init_result }
}
unsafe extern "C" fn gzip_decompress_test_inflate(
opaque: *mut c_void,
_input: *const u8,
input_size: usize,
output: *mut u8,
output_size: usize,
flush: c_int,
consumed: *mut usize,
produced: *mut usize,
) -> c_int {
let state = unsafe { &mut *opaque.cast::<GzipDecompressState>() };
state.inflate_calls += 1;
state.flushes.push(flush);
unsafe {
*consumed = 0;
*produced = 0;
}
if flush == FIO_RUST_GZIP_DECOMPRESS_Z_NO_FLUSH {
assert!(input_size != 0);
assert!(output_size != 0);
if state.stop_after_calls != 0 && state.inflate_calls > state.stop_after_calls {
unsafe {
*produced = 1;
*output = b'y';
}
return FIO_RUST_GZIP_DECOMPRESS_Z_STREAM_END;
}
unsafe {
*consumed = input_size.min(2);
*produced = 1;
*output = b'x';
}
return FIO_RUST_GZIP_DECOMPRESS_Z_OK;
}
assert_eq!(flush, FIO_RUST_GZIP_DECOMPRESS_Z_FINISH);
assert_eq!(input_size, 0);
assert!(output_size != 0);
unsafe {
*produced = 1;
*output = b'f';
}
state.finish_result
}
unsafe extern "C" fn gzip_decompress_test_end(opaque: *mut c_void) -> c_int {
let state = unsafe { &mut *opaque.cast::<GzipDecompressState>() };
state.end_calls += 1;
state.end_result
}
fn gzip_decompress_test_projection(
state: &mut GzipDecompressState,
) -> FIO_rust_gzip_decompress_projection_t {
let codec_opaque = (state as *mut GzipDecompressState).cast::<c_void>();
let io = optional_decompress_io(&mut state.io, 3);
FIO_rust_gzip_decompress_projection_t {
io,
zlib_opaque: codec_opaque,
zlib_init: Some(gzip_decompress_test_init),
zlib_inflate: Some(gzip_decompress_test_inflate),
zlib_end: Some(gzip_decompress_test_end),
}
}
#[test]
fn gzip_decompress_projection_preserves_trailing_input_and_accounting() {
let mut state = GzipDecompressState::new(b"frame-tail");
state.stop_after_calls = 1;
let projection = gzip_decompress_test_projection(&mut state);
let mut frame_size = 0;
let mut zlib_result = 0;
assert_eq!(
unsafe { FIO_rust_decompressGzipFrame(&projection, &mut frame_size, &mut zlib_result) },
FIO_RUST_GZIP_DECOMPRESS_OK
);
assert_eq!(frame_size, 2);
assert_eq!(zlib_result, FIO_RUST_GZIP_DECOMPRESS_Z_OK);
assert_eq!(state.io.input_pos, 2);
assert_eq!(state.io.consumed, vec![2, 0]);
assert_eq!(state.io.enqueue_sizes, vec![1, 1]);
assert_eq!(state.io.output_chunks, vec![vec![b'x'], vec![b'y']]);
assert_eq!(state.flushes, vec![FIO_RUST_GZIP_DECOMPRESS_Z_NO_FLUSH; 2]);
assert_eq!(state.end_calls, 1);
assert_eq!(state.io.acquire_calls, 1);
assert_eq!(state.io.release_calls, 1);
assert_eq!(state.io.sparse_end_calls, 1);
}
#[test]
fn gzip_decompress_projection_maps_eof_to_premature_error() {
let mut state = GzipDecompressState::new(b"abc");
state.finish_result = FIO_RUST_GZIP_DECOMPRESS_Z_BUF_ERROR;
let projection = gzip_decompress_test_projection(&mut state);
let mut frame_size = 41;
let mut zlib_result = 0;
assert_eq!(
unsafe { FIO_rust_decompressGzipFrame(&projection, &mut frame_size, &mut zlib_result) },
FIO_RUST_GZIP_DECOMPRESS_BUF_ERROR
);
assert_eq!(frame_size, 0);
assert_eq!(zlib_result, FIO_RUST_GZIP_DECOMPRESS_Z_BUF_ERROR);
assert_eq!(state.io.input_pos, 3);
assert_eq!(state.io.consumed, vec![2, 1, 0]);
assert_eq!(state.end_calls, 1);
assert_eq!(state.io.release_calls, 1);
assert_eq!(state.io.sparse_end_calls, 1);
}
struct LzmaDecompressState {
io: OptionalDecompressIoState,
init_result: c_int,
init_modes: Vec<c_int>,
actions: Vec<c_int>,
finish_result: c_int,
code_error: Option<c_int>,
end_calls: usize,
}
impl LzmaDecompressState {
fn new(input: &[u8]) -> Self {
Self {
io: OptionalDecompressIoState::new(input),
init_result: FIO_RUST_LZMA_DECOMPRESS_OK_CODE,
init_modes: Vec::new(),
actions: Vec::new(),
finish_result: FIO_RUST_LZMA_DECOMPRESS_STREAM_END,
code_error: None,
end_calls: 0,
}
}
}
unsafe extern "C" fn lzma_decompress_test_init(
opaque: *mut c_void,
plain_lzma: c_int,
) -> c_int {
let state = unsafe { &mut *opaque.cast::<LzmaDecompressState>() };
state.init_modes.push(plain_lzma);
state.init_result
}
unsafe extern "C" fn lzma_decompress_test_code(
opaque: *mut c_void,
_input: *const u8,
input_size: usize,
output: *mut u8,
output_size: usize,
action: c_int,
consumed: *mut usize,
produced: *mut usize,
) -> c_int {
let state = unsafe { &mut *opaque.cast::<LzmaDecompressState>() };
state.actions.push(action);
unsafe {
*consumed = 0;
*produced = 0;
}
if action == FIO_RUST_LZMA_DECOMPRESS_RUN {
assert!(input_size != 0);
assert!(output_size != 0);
unsafe {
*consumed = input_size.min(2);
*produced = 1;
*output = b'r';
}
return state.code_error.unwrap_or(FIO_RUST_LZMA_DECOMPRESS_OK_CODE);
}
assert_eq!(action, FIO_RUST_LZMA_DECOMPRESS_FINISH);
assert_eq!(input_size, 0);
assert!(output_size != 0);
unsafe {
*produced = 1;
*output = b'f';
}
state.finish_result
}
unsafe extern "C" fn lzma_decompress_test_end(opaque: *mut c_void) {
let state = unsafe { &mut *opaque.cast::<LzmaDecompressState>() };
state.end_calls += 1;
}
fn lzma_decompress_test_projection(
state: &mut LzmaDecompressState,
) -> FIO_rust_lzma_decompress_projection_t {
let codec_opaque = (state as *mut LzmaDecompressState).cast::<c_void>();
let io = optional_decompress_io(&mut state.io, 3);
FIO_rust_lzma_decompress_projection_t {
io,
lzma_opaque: codec_opaque,
lzma_init: Some(lzma_decompress_test_init),
lzma_code: Some(lzma_decompress_test_code),
lzma_end: Some(lzma_decompress_test_end),
}
}
#[test]
fn lzma_decompress_projection_preserves_mode_and_short_read_accounting() {
let mut state = LzmaDecompressState::new(b"abcde");
let projection = lzma_decompress_test_projection(&mut state);
let mut frame_size = 0;
let mut lzma_result = 0;
assert_eq!(
unsafe {
FIO_rust_decompressLzmaFrame(&projection, 1, &mut frame_size, &mut lzma_result)
},
FIO_RUST_LZMA_DECOMPRESS_OK
);
assert_eq!(frame_size, 4);
assert_eq!(lzma_result, FIO_RUST_LZMA_DECOMPRESS_OK_CODE);
assert_eq!(state.init_modes, vec![1]);
assert_eq!(
state.actions,
vec![
FIO_RUST_LZMA_DECOMPRESS_RUN,
FIO_RUST_LZMA_DECOMPRESS_RUN,
FIO_RUST_LZMA_DECOMPRESS_RUN,
FIO_RUST_LZMA_DECOMPRESS_FINISH,
]
);
assert_eq!(state.io.input_pos, 5);
assert_eq!(state.io.consumed, vec![2, 2, 1, 0]);
assert_eq!(state.io.enqueue_sizes, vec![1, 1, 1, 1]);
assert_eq!(state.end_calls, 1);
assert_eq!(state.io.release_calls, 1);
assert_eq!(state.io.sparse_end_calls, 1);
}
#[test]
fn lzma_decompress_projection_propagates_code_error_after_consumption() {
let mut state = LzmaDecompressState::new(b"abc");
state.code_error = Some(17);
let projection = lzma_decompress_test_projection(&mut state);
let mut frame_size = 41;
let mut lzma_result = 0;
assert_eq!(
unsafe {
FIO_rust_decompressLzmaFrame(&projection, 0, &mut frame_size, &mut lzma_result)
},
FIO_RUST_LZMA_DECOMPRESS_CODE_ERROR
);
assert_eq!(frame_size, 0);
assert_eq!(lzma_result, 17);
assert_eq!(state.io.consumed, vec![2]);
assert_eq!(state.end_calls, 1);
assert_eq!(state.io.release_calls, 1);
assert_eq!(state.io.sparse_end_calls, 1);
}
struct Lz4DecompressState {
io: OptionalDecompressIoState,
create_status: c_int,
create_result: usize,
code_calls: usize,
code_error: Option<usize>,
next_after_code: usize,
unfinished: bool,
free_calls: usize,
progress: Vec<u64>,
}
impl Lz4DecompressState {
fn new(input: &[u8]) -> Self {
Self {
io: OptionalDecompressIoState::new(input),
create_status: 0,
create_result: 0,
code_calls: 0,
code_error: None,
next_after_code: 0,
unfinished: false,
free_calls: 0,
progress: Vec::new(),
}
}
}
unsafe extern "C" fn lz4_decompress_test_create(
opaque: *mut c_void,
_version: c_uint,
result: *mut usize,
) -> c_int {
let state = unsafe { &mut *opaque.cast::<Lz4DecompressState>() };
unsafe { *result = state.create_result };
state.create_status
}
unsafe extern "C" fn lz4_decompress_test_code(
opaque: *mut c_void,
output: *mut u8,
output_size: *mut usize,
_input: *const u8,
input_size: *mut usize,
next_to_load: *mut usize,
) -> c_int {
let state = unsafe { &mut *opaque.cast::<Lz4DecompressState>() };
state.code_calls += 1;
if let Some(error) = state.code_error {
unsafe {
*input_size = 0;
*next_to_load = error;
}
return error as c_int;
}
let consumed = if state.code_calls == 1 {
(*input_size).min(2)
} else {
*input_size
};
let produced = if state.unfinished && state.code_calls > 1 {
1
} else {
*output_size
};
assert!(produced != 0);
unsafe {
*input_size = consumed;
std::ptr::write_bytes(output, b'u', produced);
*next_to_load = if state.code_calls == 1 || state.unfinished {
state.next_after_code
} else {
0
};
}
0
}
unsafe extern "C" fn lz4_decompress_test_free(opaque: *mut c_void) {
let state = unsafe { &mut *opaque.cast::<Lz4DecompressState>() };
state.free_calls += 1;
}
unsafe extern "C" fn lz4_decompress_test_progress(opaque: *mut c_void, decoded: u64) {
let state = unsafe { &mut *opaque.cast::<Lz4DecompressState>() };
state.progress.push(decoded);
}
fn lz4_decompress_test_projection(
state: &mut Lz4DecompressState,
) -> FIO_rust_lz4_decompress_projection_t {
let io_opaque = (&mut state.io as *mut OptionalDecompressIoState).cast::<c_void>();
let codec_opaque = (state as *mut Lz4DecompressState).cast::<c_void>();
FIO_rust_lz4_decompress_projection_t {
io: FIO_rust_decompress_io_projection_t {
read_opaque: io_opaque,
write_opaque: io_opaque,
read_buffer_size: 0,
read_fill: Some(optional_decompress_read_fill),
read_consume: Some(optional_decompress_read_consume),
write_acquire: Some(optional_decompress_write_acquire),
write_enqueue: Some(optional_decompress_write_enqueue),
write_release: Some(optional_decompress_write_release),
sparse_write_end: Some(optional_decompress_sparse_end),
},
codec_opaque,
progress_opaque: codec_opaque,
version: 100,
create: Some(lz4_decompress_test_create),
code: Some(lz4_decompress_test_code),
free_context: Some(lz4_decompress_test_free),
progress: Some(lz4_decompress_test_progress),
}
}
#[test]
fn lz4_decompress_projection_retries_full_output_and_accounts_input() {
let mut state = Lz4DecompressState::new(b"abcdTAIL");
state.next_after_code = 2;
let projection = lz4_decompress_test_projection(&mut state);
let mut frame_size = 0;
let mut lz4_result = 0;
assert_eq!(
unsafe { FIO_rust_decompressLz4Frame(&projection, &mut frame_size, &mut lz4_result) },
FIO_RUST_LZ4_DECOMPRESS_OK
);
assert_eq!(frame_size, 8);
assert_eq!(lz4_result, 0);
assert_eq!(state.code_calls, 2);
assert_eq!(state.io.input_pos, 4);
assert_eq!(state.io.consumed, vec![4]);
assert_eq!(state.io.enqueue_sizes, vec![4, 4]);
assert_eq!(state.progress, vec![4, 8]);
assert_eq!(state.free_calls, 1);
assert_eq!(state.io.release_calls, 1);
assert_eq!(state.io.sparse_end_calls, 1);
}
#[test]
fn lz4_decompress_projection_reports_errors_and_unfinished_input() {
let mut error_state = Lz4DecompressState::new(b"abcd");
error_state.code_error = Some(23);
let error_projection = lz4_decompress_test_projection(&mut error_state);
let mut frame_size = 41;
let mut lz4_result = 0;
assert_eq!(
unsafe {
FIO_rust_decompressLz4Frame(&error_projection, &mut frame_size, &mut lz4_result)
},
FIO_RUST_LZ4_DECOMPRESS_CODE_ERROR
);
assert_eq!(frame_size, 0);
assert_eq!(lz4_result, 23);
assert_eq!(error_state.io.input_pos, 0);
assert_eq!(error_state.free_calls, 1);
assert_eq!(error_state.io.release_calls, 1);
assert_eq!(error_state.io.sparse_end_calls, 1);
let mut unfinished_state = Lz4DecompressState::new(b"abcd");
unfinished_state.next_after_code = 4;
unfinished_state.unfinished = true;
let unfinished_projection = lz4_decompress_test_projection(&mut unfinished_state);
let mut frame_size = 41;
let mut lz4_result = 0;
assert_eq!(
unsafe {
FIO_rust_decompressLz4Frame(
&unfinished_projection,
&mut frame_size,
&mut lz4_result,
)
},
FIO_RUST_LZ4_DECOMPRESS_UNFINISHED
);
assert_eq!(frame_size, 0);
assert_eq!(unfinished_state.io.input_pos, 4);
assert_eq!(unfinished_state.free_calls, 1);
assert_eq!(unfinished_state.io.release_calls, 1);
assert_eq!(unfinished_state.io.sparse_end_calls, 1);
}
#[test]
fn classifies_all_cli_decompression_headers() {
let is_mock_zstd = |buffer: &[u8]| buffer.starts_with(&[0x28, 0xB5, 0x2F, 0xFD]);