lgz/tests/stream.rs
//! Round 2: ONE pass, chunk by chunk. The library's Read → Write API and the
//! CLI both stream: no whole-file read, stdin works, memory stays near
//! chunk size × chunks in flight, small inputs are never slower than flate2.
//!
//! The real-layer tests read `LGZ_LAYERS` (a directory of fetched OCI layer
//! blobs, `benches/fetch_oci_layers.py`) and are skipped when it is unset.
use std::io::{Read, Write};
use std::process::{Command, Stdio};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;
fn corpus(len: usize, seed: u32) -> Vec<u8> {
let words: [&[u8]; 16] = [
b"usr",
b"lib",
b"python3.12",
b"site-packages",
b"share",
b"locale",
b"etc",
b"ssl",
b"certs",
b"bin",
b"include",
b"pkgconfig",
b"__pycache__",
b"dist-info",
b"LICENSE",
b"README",
];
let mut out = Vec::with_capacity(len + 64);
let mut x: u32 = seed | 1;
while out.len() < len {
x ^= x << 13;
x ^= x >> 17;
x ^= x << 5;
out.extend_from_slice(words[(x % 16) as usize]);
out.push(if x >> 29 == 0 { b'\n' } else { b'/' });
if x & 0x3f00 == 0 {
out.extend_from_slice(&x.to_le_bytes());
}
}
out.truncate(len);
out
}
fn gzip(data: &[u8], level: u32) -> Vec<u8> {
let mut e = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::new(level));
e.write_all(data).unwrap();
e.finish().unwrap()
}
fn flate2_decode(gz: &[u8]) -> Vec<u8> {
let mut out = Vec::new();
flate2::read::MultiGzDecoder::new(gz)
.read_to_end(&mut out)
.unwrap();
out
}
/// Counts what the decoder has pulled from the input.
struct CountingReader<R> {
inner: R,
pulled: Arc<AtomicU64>,
}
impl<R: Read> Read for CountingReader<R> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let n = self.inner.read(buf)?;
self.pulled.fetch_add(n as u64, Ordering::Relaxed);
Ok(n)
}
}
/// Records how much input had been pulled when the first output byte arrived.
struct FirstWrite {
out: Vec<u8>,
pulled: Arc<AtomicU64>,
at_first: Option<u64>,
}
impl Write for FirstWrite {
fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
if self.at_first.is_none() && !b.is_empty() {
self.at_first = Some(self.pulled.load(Ordering::Relaxed));
}
self.out.extend_from_slice(b);
Ok(b.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
#[test]
fn the_library_streams_read_to_write_without_reading_the_whole_input_first() {
let data = corpus(256 << 20, 11);
let gz = gzip(&data, 6);
let pulled = Arc::new(AtomicU64::new(0));
let r = CountingReader {
inner: &gz[..],
pulled: pulled.clone(),
};
let mut w = FirstWrite {
out: Vec::new(),
pulled: pulled.clone(),
at_first: None,
};
let st = lgz::stream::decode(r, &mut w, 8, 4 << 20).expect("stream decode");
assert!(w.out == data, "output differs");
assert!(st.segments > 8, "not parallel: {st:?}");
// at most a few chunks in flight (reader, coordinator, writer) before the
// first byte comes out, never the whole input
let first = w.at_first.unwrap();
assert!(
first <= 6 * (4 << 20) && (first as usize) < gz.len() * 2 / 3,
"first output only after {first} of {} input bytes",
gz.len()
);
}
#[test]
fn multi_member_and_pigz_style_flush_streams_stream_too() {
// concatenated members (cat a.gz b.gz), the second one with sync flushes
let a = corpus(9 << 20, 21);
let b = corpus(11 << 20, 22);
let mut e = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::new(6));
for piece in b.chunks(256 * 1024) {
e.write_all(piece).unwrap();
e.flush().unwrap(); // Z_SYNC_FLUSH: 00 00 FF FF points
}
let mut gz = gzip(&a, 6);
gz.extend_from_slice(&e.finish().unwrap());
let mut want = a.clone();
want.extend_from_slice(&b);
for threads in [1usize, 4, 12] {
for chunk in [1 << 20, 4 << 20, 64 << 20] {
let mut out = Vec::new();
lgz::stream::decode(&gz[..], &mut out, threads, chunk)
.unwrap_or_else(|e| panic!("t {threads} chunk {chunk}: {e}"));
assert!(out == want, "t {threads} chunk {chunk}: output differs");
}
}
}
#[test]
fn a_bad_trailer_is_an_error_that_says_what_was_written() {
let data = corpus(16 << 20, 31);
let mut gz = gzip(&data, 6);
let n = gz.len();
gz[n - 6] ^= 1; // CRC32 byte
let mut out = Vec::new();
let e = lgz::stream::decode(&gz[..], &mut out, 8, 2 << 20).unwrap_err();
assert_eq!(e.written, out.len() as u64);
assert!(e.to_string().contains("written"), "{e}");
}
#[test]
fn stdin_decodes_in_parallel() {
let data = corpus(200 << 20, 41);
let gz = gzip(&data, 6);
let mut child = Command::new(env!("CARGO_BIN_EXE_lgz"))
.env("LGZ_THREADS", "8")
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.unwrap();
let mut stdin = child.stdin.take().unwrap();
let feeder = std::thread::spawn(move || stdin.write_all(&gz).unwrap());
let out = child.wait_with_output().unwrap();
feeder.join().unwrap();
let err = String::from_utf8_lossy(&out.stderr);
assert!(out.status.success(), "lgz failed: {err}");
assert!(out.stdout == data, "stdin output differs; stderr: {err}");
let segs: usize = err
.split("segments=")
.nth(1)
.and_then(|s| s.split(|c: char| !c.is_ascii_digit()).next())
.and_then(|s| s.parse().ok())
.unwrap_or(0);
assert!(segs > 1, "stdin was not decoded in parallel: {err}");
}
#[test]
fn small_inputs_are_never_slower_than_flate2() {
for len in [256 << 10, 2 << 20, 8 << 20, 16 << 20] {
let data = corpus(len, 51);
let gz = gzip(&data, 6);
// interleaved, best of 15 each: the two are the same work on a
// shared host, so only the minimum is a fair comparison
let (mut ref_s, mut lgz_s) = (f64::MAX, f64::MAX);
for _ in 0..15 {
let t = Instant::now();
assert_eq!(flate2_decode(&gz).len(), data.len());
ref_s = ref_s.min(t.elapsed().as_secs_f64());
let t = Instant::now();
let mut o = Vec::new();
lgz::stream::decode(&gz[..], &mut o, 12, lgz::stream::DEFAULT_CHUNK).unwrap();
assert_eq!(o.len(), data.len());
lgz_s = lgz_s.min(t.elapsed().as_secs_f64());
}
assert!(
lgz_s <= ref_s * 1.10 + 0.002,
"{} KiB: lgz {lgz_s:.4}s vs flate2 {ref_s:.4}s",
len >> 10
);
}
}
fn layers() -> Vec<std::path::PathBuf> {
let Some(dir) = std::env::var_os("LGZ_LAYERS") else {
eprintln!("LGZ_LAYERS unset: real-layer test skipped");
return Vec::new();
};
let mut v: Vec<_> = std::fs::read_dir(dir)
.unwrap()
.filter_map(|e| e.ok().map(|e| e.path()))
.filter(|p| p.extension().is_some_and(|x| x == "gz"))
.collect();
v.sort();
v
}
/// Peak RSS of a child process (KiB), via wait4.
fn run_rss(cmd: &mut Command) -> (std::process::ExitStatus, Vec<u8>, i64) {
use std::os::unix::process::ExitStatusExt;
let mut child = cmd.stdout(Stdio::piped()).spawn().unwrap();
let mut out = Vec::new();
child.stdout.take().unwrap().read_to_end(&mut out).unwrap();
let mut status = 0;
let mut ru: libc::rusage = unsafe { std::mem::zeroed() };
unsafe { libc::wait4(child.id() as i32, &mut status, 0, &mut ru) };
(
std::process::ExitStatus::from_raw(status),
out,
ru.ru_maxrss,
)
}
#[test]
fn every_real_layer_is_byte_identical_through_the_cli() {
for p in layers() {
let gz = std::fs::read(&p).unwrap();
let want = flate2_decode(&gz);
let (st, out, _) = run_rss(Command::new(env!("CARGO_BIN_EXE_lgz")).arg(&p));
assert!(st.success(), "{}: exit {st}", p.display());
assert!(out == want, "{}: output differs", p.display());
}
}
#[test]
fn peak_rss_on_the_python_layer_stays_near_chunks_in_flight() {
let Some(p) = layers()
.into_iter()
.find(|p| p.to_string_lossy().contains("python-425148f8"))
else {
return;
};
// stdout is drained by this test (counted, not kept) so the bound is lgz's own
let mut child = Command::new(env!("CARGO_BIN_EXE_lgz"))
.arg(&p)
.stdout(Stdio::piped())
.spawn()
.unwrap();
let mut so = child.stdout.take().unwrap();
let mut buf = vec![0u8; 1 << 20];
let mut n = 0u64;
loop {
let k = so.read(&mut buf).unwrap();
if k == 0 {
break;
}
n += k as u64;
}
let mut status = 0;
let mut ru: libc::rusage = unsafe { std::mem::zeroed() };
unsafe { libc::wait4(child.id() as i32, &mut status, 0, &mut ru) };
assert_eq!(status, 0);
assert!(n > 600_000_000, "output {n}");
let mb = ru.ru_maxrss / 1024;
// windows in flight (reader, producer, jobs) plus segment outputs; the
// zero-copy / preallocated-slot round takes this toward rapidgzip's ~330 MB
assert!(mb < 700, "peak RSS {mb} MB on the 226 MB python layer");
}
#[test]
fn small_chunks_on_a_layer_with_false_starts() {
// stored payloads holding compressed files give false starts; with small
// chunks the carry start itself is often one
let mut data = Vec::new();
for k in 0..12u32 {
data.extend_from_slice(&corpus(1 << 20, 300 + k));
data.extend_from_slice(&gzip(&corpus(5 << 20, 400 + k), 6));
}
let gz = gzip(&data, 6);
for chunk in [1 << 20, 2 << 20, 3 << 20] {
for threads in [6usize, 12, 31] {
let mut out = Vec::new();
lgz::stream::decode(&gz[..], &mut out, threads, chunk)
.unwrap_or_else(|e| panic!("t {threads} chunk {chunk}: {e}"));
assert!(out == data, "t {threads} chunk {chunk}: output differs");
}
}
}
#[test]
fn high_ratio_segments_never_write_past_their_buffer() {
// ratio far above the 4× size hint: every segment's first guess is too
// small, so the overflow-and-retry path runs on every one. A write past the
// buffer there corrupts the heap (seen as "double free or corruption").
let data = corpus(240 << 20, 61); // ~8× ratio, ~30 MB compressed
let gz = gzip(&data, 6);
for _ in 0..3 {
for threads in [12usize, 16, 24] {
let mut out = Vec::new();
lgz::stream::decode(&gz[..], &mut out, threads, 4 << 20).unwrap();
assert!(out == data, "t {threads}: output differs");
}
}
}
#[test]
fn one_thread_on_a_large_input_is_never_slower_than_flate2() {
// many reads on the one-thread path: each must end near its read limit
// (a soft stop), not run on to the end and start over
let data = corpus(320 << 20, 71);
let gz = gzip(&data, 6);
assert!(gz.len() > 30 << 20, "{}", gz.len());
let (mut ref_s, mut lgz_s) = (f64::MAX, f64::MAX);
for _ in 0..4 {
let t = Instant::now();
assert_eq!(flate2_decode(&gz).len(), data.len());
ref_s = ref_s.min(t.elapsed().as_secs_f64());
let t = Instant::now();
let mut o = Vec::new();
let st = lgz::stream::decode(&gz[..], &mut o, 1, lgz::stream::DEFAULT_CHUNK).unwrap();
assert!(st.serial && o == data);
lgz_s = lgz_s.min(t.elapsed().as_secs_f64());
}
assert!(
lgz_s <= ref_s * 1.05,
"lgz {lgz_s:.3}s vs flate2 {ref_s:.3}s"
);
}
#[test]
fn the_slice_path_decodes_members_and_refuses_a_bad_trailer() {
let a = corpus(9 << 20, 81);
let b = corpus(30 << 20, 82);
let mut gz = gzip(&a, 6);
gz.extend_from_slice(&gzip(&b, 9));
let mut want = a.clone();
want.extend_from_slice(&b);
for threads in [1usize, 12] {
let mut out = Vec::new();
let st = lgz::stream::decode_slice(&gz, &mut out, threads).unwrap();
assert!(out == want, "t {threads}");
if threads == 1 {
assert_eq!(st.members, 2);
}
}
let n = gz.len();
gz[n - 6] ^= 1;
let mut out = Vec::new();
assert!(lgz::stream::decode_slice(&gz, &mut out, 1).is_err());
}
#[test]
fn the_parallel_slice_path_handles_false_starts_and_high_ratios() {
let mut data = Vec::new();
for k in 0..12u32 {
data.extend_from_slice(&corpus(2 << 20, 500 + k));
data.extend_from_slice(&gzip(&corpus(8 << 20, 600 + k), 6));
}
let gz = gzip(&data, 6);
let hi = corpus(240 << 20, 71);
let gz_hi = gzip(&hi, 6);
for threads in [6usize, 12, 31] {
let mut out = Vec::new();
lgz::stream::decode_slice(&gz, &mut out, threads)
.unwrap_or_else(|e| panic!("t {threads}: {e}"));
assert!(out == data, "false starts, t {threads}");
let mut out = Vec::new();
lgz::stream::decode_slice(&gz_hi, &mut out, threads).unwrap();
assert!(out == hi, "high ratio, t {threads}");
}
}
#[test]
fn the_cli_writes_a_regular_file_output_positionally_and_correctly() {
let data = corpus(200 << 20, 91);
let gz = gzip(&data, 6);
let dir = std::env::temp_dir().join(format!("lgz-pos-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let inp = dir.join("in.gz");
std::fs::write(&inp, &gz).unwrap();
// -o, and stdout redirected to a file
let o1 = dir.join("o1");
let st = Command::new(env!("CARGO_BIN_EXE_lgz"))
.env("LGZ_THREADS", "8")
.arg(&inp)
.arg("-o")
.arg(&o1)
.status()
.unwrap();
assert!(st.success());
assert!(std::fs::read(&o1).unwrap() == data, "-o output differs");
let o2 = dir.join("o2");
let st = Command::new(env!("CARGO_BIN_EXE_lgz"))
.env("LGZ_THREADS", "8")
.arg(&inp)
.stdout(std::fs::File::create(&o2).unwrap())
.status()
.unwrap();
assert!(st.success());
assert!(
std::fs::read(&o2).unwrap() == data,
"stdout-to-file output differs"
);
let _ = std::fs::remove_dir_all(&dir);
}