lgz/tests/bitsplit.rs
//! Ordinary gzip (no flush points, dynamic-Huffman blocks) decoded in parallel
//! by splitting at real block starts found at the BIT level.
use std::io::Write;
use lgz::bitsplit;
/// Compressible, not trivially repetitive text, like the files in a layer.
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(match x >> 28 {
0..=9 => b'/',
10..=13 => b'\n',
_ => 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()
}
/// Every true block start of a raw DEFLATE stream, by decoding it block by block.
fn true_block_starts(deflate: &[u8], out_len: usize) -> Vec<(usize, u32)> {
let mut starts = Vec::new();
let mut out = vec![0u8; out_len + linflate::OVERWRITE_HEADROOM];
let mut bit = 0usize;
let mut done = 0usize;
loop {
let byte = deflate[bit / 8] as u32 | (deflate[bit / 8 + 1] as u32) << 8;
let hdr = (byte >> (bit % 8)) & 7; // BFINAL | BTYPE<<1
starts.push((bit, hdr));
// decode exactly one block: stop at the first block end after `bit`
let win = done.min(32 * 1024);
let mut buf = vec![0u8; win + out_len - done + linflate::OVERWRITE_HEADROOM];
buf[..win].copy_from_slice(&out[done - win..done]);
let r = linflate::inflate_bits(deflate, bit, bit + 1, &mut buf, win).unwrap();
out[done..done + r.written].copy_from_slice(&buf[win..win + r.written]);
done += r.written;
if r.final_block {
break;
}
bit = r.end_bit;
}
starts
}
#[test]
fn finds_the_first_real_block_start_at_any_bit() {
let data = corpus(6 << 20, 7);
let deflate = miniz_oxide::deflate::compress_to_vec(&data, 6);
let starts = true_block_starts(&deflate, data.len());
let dynamic: Vec<usize> = starts
.iter()
.filter(|(_, h)| *h == 0b100) // BFINAL=0, BTYPE=2
.map(|(b, _)| *b)
.collect();
assert!(
dynamic.len() > 20,
"corpus too small: {} dynamic blocks",
dynamic.len()
);
let total = deflate.len() * 8;
let mut unaligned = 0;
for k in 1..16 {
let from = total * k / 16 + 3; // never on a byte boundary
let want = *dynamic.iter().find(|&&b| b >= from).unwrap();
let got = bitsplit::find_block_start(&deflate, from, total).expect("a block start");
assert_eq!(got, want, "split {k}: from bit {from}");
if !got.is_multiple_of(8) {
unaligned += 1;
}
}
assert!(
unaligned > 0,
"every found start was byte aligned; the test proves nothing"
);
}
#[test]
fn nothing_is_found_in_noise() {
let mut x: u64 = 0x2545_f491_4f6c_dd1d;
let noise: Vec<u8> = (0..(1 << 20))
.map(|_| {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
x as u8
})
.collect();
assert_eq!(bitsplit::find_block_start(&noise, 0, noise.len() * 8), None);
}
#[test]
fn ordinary_gzip_decodes_in_parallel_and_matches() {
for (len, level, seed) in [(120 << 20, 6, 1u32), (9 << 20, 9, 2), (60 << 20, 1, 3)] {
let data = corpus(len, seed);
let gz = gzip(&data, level);
for threads in [1usize, 4, 12] {
for chunk in [1 << 20, 4 << 20, 200 << 20] {
let (out, st) = bitsplit::decompress(&gz, threads, chunk).unwrap_or_else(|e| {
panic!("len {len} level {level} t {threads} chunk {chunk}: {e}")
});
assert!(
out == data,
"output differs: len {len} level {level} t {threads} chunk {chunk}"
);
// below the serial threshold the input goes to flate2 by design
if threads > 1 && gz.len() >= lgz::stream::SERIAL_BELOW {
assert!(
st.segments > 1,
"t {threads} chunk {chunk}: no split was used"
);
assert!(
st.marker_segments > 0,
"no segment needed the cross-border resolution"
);
}
}
}
}
}
#[test]
fn a_corrupted_stream_is_refused_not_emitted() {
let data = corpus(12 << 20, 5);
let mut gz = gzip(&data, 6);
let mid = gz.len() / 2;
gz[mid] ^= 0x10;
assert!(bitsplit::decompress(&gz, 8, 2 << 20).is_err());
// the library entry refuses it too (parallel or fallback, never bad bytes)
if let Ok(out) = lgz::decompress_gz(&gz) {
assert!(out == data, "corrupt input returned wrong bytes");
}
}
#[test]
fn decompress_gz_takes_the_parallel_path_for_ordinary_gzip() {
let data = corpus(120 << 20, 9);
let gz = gzip(&data, 6);
assert!(
gz.len() >= lgz::stream::SERIAL_BELOW,
"corpus too small: {}",
gz.len()
);
assert_eq!(lgz::decompress_gz(&gz).unwrap(), data);
let (_, st) = bitsplit::decompress(&gz, 8, 200 << 20).unwrap();
assert!(st.segments >= 4);
}
/// A layer that holds compressed files: the outer gzip stores them raw
/// (stored blocks), and their own DEFLATE blocks look exactly like real
/// block starts. A split may land on one; it must be caught and redone in
/// order, not fail the run and never give wrong bytes.
#[test]
fn a_false_start_inside_a_stored_payload_is_redone_in_order() {
let mut data = Vec::new();
for k in 0..10u32 {
data.extend_from_slice(&corpus(1 << 20, 100 + k));
data.extend_from_slice(&gzip(&corpus(6 << 20, 200 + k), 6)); // ~2 MB, incompressible
}
let gz = gzip(&data, 6);
let mut redone = 0;
for threads in [12usize, 16, 24, 31] {
let (out, st) = bitsplit::decompress(&gz, threads, 200 << 20)
.unwrap_or_else(|e| panic!("t {threads}: {e}"));
assert!(out == data, "t {threads}: output differs");
redone += st.redone + st.skipped;
}
assert!(
redone > 0,
"no split landed on a false start; the test proves nothing"
);
}