lgz/src/bitsplit.rs
//! **Parallel decode of ORDINARY gzip** — no flush points needed.
//!
//! A standard `gzip` / Go `compress/gzip` stream (every Docker Hub layer) is
//! one DEFLATE stream of dynamic-Huffman blocks that start at any BIT, and
//! whose back-references reach up to 32 KB into the previous output. This
//! module splits such a stream at real block starts and decodes the pieces on
//! all cores:
//!
//! 1. **Split** — the pipeline's split stage takes a chunk and asks, for each
//! of `n_workers − 1` nominal points (chunk / cores), for the first real
//! block start after it: [`find_block_start`].
//! 2. **Find, at every bit** — a 64-offsets-at-once bitmask filter
//! ([`candidates64`], plain integer ops on a 128-bit window, so it is the
//! same fast path on x86_64 and aarch64) keeps only offsets whose first
//! 13 bits can start a non-final dynamic block; then the 3-bit code-length
//! code must be a complete prefix code (Kraft); then the whole header must
//! build complete literal/length and distance codes with an end-of-block
//! code; then the block must decode to its end, and the next block header
//! must be valid too.
//! 3. **Decode with markers** — a segment that does not start the stream is
//! decoded into 16-bit symbols: a byte, or a MARKER naming a byte of the
//! unknown 32 KB before the segment ([`decode_marked`]). As soon as the
//! last 32 KB of a segment's own output hold no marker, the rest of the
//! segment is decoded by linflate's fast path with that output as window.
//! 4. **Resolve in order** — the collector ([`Collector`]) holds the real
//! 32 KB before each segment, replaces the markers, writes the bytes and
//! keeps the CRC32 and length; [`Collector::finish`] checks them against
//! the gzip trailer. A segment that did not end exactly where the next one
//! starts fails the run.
//!
//! The library entry [`decompress`] runs the same split → decode → collect
//! stages over a byte slice and returns nothing unless the trailer matched.
use crate::parallel::GzError;
pub mod simd;
/// **Buffer pools** for the hot path: segment outputs, 16-bit marker heads
/// and resolved heads are taken here by the workers and handed back by the
/// in-order stage after writing, so a window allocates nothing once the
/// pools are warm.
pub mod pool {
use std::sync::Mutex;
static U8: Mutex<Vec<Vec<u8>>> = Mutex::new(Vec::new());
static U16: Mutex<Vec<Vec<u16>>> = Mutex::new(Vec::new());
/// At most this many idle buffers of each kind are kept.
const KEEP: usize = 96;
fn take<T>(p: &Mutex<Vec<Vec<T>>>, min_cap: usize) -> Vec<T> {
let mut p = p.lock().unwrap();
let v = match p.iter().position(|v| v.capacity() >= min_cap) {
Some(i) => p.swap_remove(i),
None => match p.iter().enumerate().max_by_key(|(_, v)| v.capacity()) {
Some((i, _)) => p.swap_remove(i),
None => Vec::new(),
},
};
drop(p);
let mut v = v;
v.clear();
v.reserve(min_cap);
v
}
fn give<T>(p: &Mutex<Vec<Vec<T>>>, v: Vec<T>) {
if v.capacity() == 0 {
return;
}
let mut p = p.lock().unwrap();
if p.len() < KEEP {
p.push(v);
}
}
/// A byte buffer with at least `min_cap` capacity, empty.
pub fn take_u8(min_cap: usize) -> Vec<u8> {
take(&U8, min_cap)
}
/// Hand a byte buffer back.
pub fn give_u8(v: Vec<u8>) {
give(&U8, v)
}
/// A symbol buffer with at least `min_cap` capacity, empty.
pub fn take_u16(min_cap: usize) -> Vec<u16> {
take(&U16, min_cap)
}
/// Hand a symbol buffer back.
pub fn give_u16(v: Vec<u16>) {
give(&U16, v)
}
/// Hand a written segment's buffers back.
pub fn recycle(seg: super::MarkedSeg) {
give_u8(seg.tail);
give_u16(seg.head);
}
}
/// The DEFLATE window: back-references reach at most this far back.
pub const WINDOW: usize = 32 * 1024;
/// A marker for byte `w` (0 = oldest) of the unknown window before a segment.
const MARKER: u16 = 0x8000;
// ── Bit reader (LSB first, zero-padded past the end) ─────────────────────────
struct Bits<'a> {
d: &'a [u8],
pos: usize,
end: usize,
}
impl<'a> Bits<'a> {
fn new(d: &'a [u8], pos: usize) -> Self {
Self {
d,
pos,
end: d.len() * 8,
}
}
#[inline(always)]
fn word(&self) -> u64 {
let b = self.pos >> 3;
let w = if b + 8 <= self.d.len() {
u64::from_le_bytes(self.d[b..b + 8].try_into().unwrap())
} else {
let mut t = [0u8; 8];
if b < self.d.len() {
t[..self.d.len() - b].copy_from_slice(&self.d[b..]);
}
u64::from_le_bytes(t)
};
w >> (self.pos & 7)
}
/// Peek `n` ≤ 32 bits.
#[inline(always)]
fn peek(&self, n: u32) -> u32 {
(self.word() & ((1u64 << n) - 1)) as u32
}
#[inline(always)]
fn take(&mut self, n: u32) -> Result<u32, GzError> {
let v = self.peek(n);
self.pos += n as usize;
if self.pos > self.end {
return Err(GzError("bitsplit: unexpected end of input"));
}
Ok(v)
}
}
// ── Huffman decode tables (one level, `max` bits) ────────────────────────────
const LEN_BASE: [u16; 29] = [
3, 4, 5, 6, 7, 8, 9, 10, 11, 13, 15, 17, 19, 23, 27, 31, 35, 43, 51, 59, 67, 83, 99, 115, 131,
163, 195, 227, 258,
];
const LEN_EXTRA: [u8; 29] = [
0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 4, 5, 5, 5, 5, 0,
];
const DIST_BASE: [u16; 30] = [
1, 2, 3, 4, 5, 7, 9, 13, 17, 25, 33, 49, 65, 97, 129, 193, 257, 385, 513, 769, 1025, 1537,
2049, 3073, 4097, 6145, 8193, 12289, 16385, 24577,
];
const DIST_EXTRA: [u8; 30] = [
0, 0, 0, 0, 1, 1, 2, 2, 3, 3, 4, 4, 5, 5, 6, 6, 7, 7, 8, 8, 9, 9, 10, 10, 11, 11, 12, 12, 13,
13,
];
const CL_ORDER: [usize; 19] = [
16, 17, 18, 0, 8, 7, 9, 6, 10, 5, 11, 4, 12, 3, 13, 2, 14, 1, 15,
];
/// How complete a set of code lengths is.
#[derive(PartialEq, Eq, Debug)]
enum Kraft {
Complete,
/// Exactly one code, of length 1 (DEFLATE allows this one incomplete code).
Single,
Empty,
Bad,
}
fn kraft(lens: &[u8], max: u32) -> Kraft {
let mut count = [0u32; 16];
for &l in lens {
count[l as usize] += 1;
}
let used: u32 = count[1..].iter().sum();
if used == 0 {
return Kraft::Empty;
}
let mut left: i64 = 1;
for c in &count[1..=max as usize] {
left = left * 2 - *c as i64;
if left < 0 {
return Kraft::Bad;
}
}
if left == 0 {
Kraft::Complete
} else if used == 1 && count[1] == 1 {
Kraft::Single
} else {
Kraft::Bad
}
}
/// One-level table: index = next `bits` input bits; entry = sym | len << 12.
struct Table {
bits: u32,
t: Vec<u16>,
}
impl Table {
fn empty() -> Self {
Self {
bits: 1,
t: vec![0; 2],
}
}
fn build(&mut self, lens: &[u8]) {
let max = lens.iter().copied().max().unwrap_or(0).max(1) as u32;
self.bits = max;
let size = 1usize << max;
self.t.clear();
self.t.resize(size, 0); // len 0 = invalid code
let mut count = [0u16; 16];
for &l in lens {
count[l as usize] += 1;
}
count[0] = 0;
let mut next = [0u16; 16];
let mut code = 0u16;
for l in 1..16 {
code = (code + count[l - 1]) << 1;
next[l] = code;
}
for (sym, &l) in lens.iter().enumerate() {
if l == 0 {
continue;
}
let c = next[l as usize];
next[l as usize] += 1;
let rev = (c.reverse_bits() >> (16 - l)) as usize;
let entry = sym as u16 | (l as u16) << 12;
let mut i = rev;
while i < size {
self.t[i] = entry;
i += 1 << l;
}
}
}
#[inline(always)]
fn decode(&self, b: &mut Bits) -> Result<u16, GzError> {
let e = self.t[b.peek(self.bits) as usize];
let l = (e >> 12) as u32;
if l == 0 {
return Err(GzError("bitsplit: invalid Huffman code"));
}
b.take(l)?;
Ok(e & 0x0fff)
}
}
struct Codes {
lit: Table,
dist: Table,
/// the code-length code of a dynamic header (rebuilt in place)
cl: Table,
}
impl Codes {
fn new() -> Self {
Self {
lit: Table::empty(),
dist: Table::empty(),
cl: Table::empty(),
}
}
fn fixed(&mut self) {
let mut l = [0u8; 288];
l[..144].fill(8);
l[144..256].fill(9);
l[256..280].fill(7);
l[280..].fill(8);
self.lit.build(&l);
self.dist.build(&[5u8; 30]);
}
}
thread_local! {
static SEG_CODES: std::cell::RefCell<Option<Codes>> = const { std::cell::RefCell::new(None) };
}
/// Puts a worker's tables back for its next segment, however the decode ends.
struct CodesGuard(Option<Codes>);
impl Drop for CodesGuard {
fn drop(&mut self) {
let c = self.0.take();
SEG_CODES.with(|s| *s.borrow_mut() = c);
}
}
/// Read and validate a dynamic-block header (after the 3 header bits) into
/// `codes`. Strict: every check zlib makes, so a random bit position that is
/// not a real block start is refused here with overwhelming probability.
fn read_dynamic(b: &mut Bits, codes: &mut Codes) -> Result<(), GzError> {
let bad = || GzError("bitsplit: invalid dynamic header");
let hlit = b.take(5)? as usize + 257;
let hdist = b.take(5)? as usize + 1;
let hclen = b.take(4)? as usize + 4;
if hlit > 286 || hdist > 30 {
return Err(bad());
}
let mut cl = [0u8; 19];
for &i in CL_ORDER.iter().take(hclen) {
cl[i] = b.take(3)? as u8;
}
if kraft(&cl, 7) != Kraft::Complete {
return Err(bad());
}
codes.cl.build(&cl);
let mut lens = [0u8; 316];
let n = hlit + hdist;
let mut i = 0;
while i < n {
let sym = codes.cl.decode(b)?;
match sym {
0..=15 => {
lens[i] = sym as u8;
i += 1;
}
16 => {
if i == 0 {
return Err(bad());
}
let r = 3 + b.take(2)? as usize;
if i + r > n {
return Err(bad());
}
let p = lens[i - 1];
lens[i..i + r].fill(p);
i += r;
}
17 | 18 => {
let r = if sym == 17 {
3 + b.take(3)?
} else {
11 + b.take(7)?
} as usize;
if i + r > n {
return Err(bad());
}
lens[i..i + r].fill(0);
i += r;
}
_ => return Err(bad()),
}
}
let (lit, dist) = lens[..n].split_at(hlit);
if lit[256] == 0 {
return Err(bad()); // no end-of-block code
}
match kraft(lit, 15) {
Kraft::Complete | Kraft::Single => {}
_ => return Err(bad()),
}
match kraft(dist, 15) {
Kraft::Complete | Kraft::Single | Kraft::Empty => {}
Kraft::Bad => return Err(bad()),
}
codes.lit.build(lit);
codes.dist.build(dist);
Ok(())
}
// ── Marker decoder ───────────────────────────────────────────────────────────
/// A gzip member that ended inside a segment: its trailer, and where in the
/// segment's output (head then tail) it ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Mark {
/// Output offset within the segment, just past the member's last byte.
pub at: usize,
/// The member's trailer.
pub crc: u32,
pub isize: u32,
}
/// How a segment's decode ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum SegEnd {
/// At one of the split starts it was given (`end_bit` is that start).
#[default]
Stop,
/// A member's final block ended; its 8-byte trailer starts at `end_bit`
/// (byte aligned) but is not all in the input yet.
Trailer,
/// A member and its trailer ended (recorded in `marks`); the next member's
/// gzip header, or the end of the stream, starts at `end_bit`.
Header,
}
/// Decoded output of one segment.
#[derive(Debug, Default)]
pub struct MarkedSeg {
/// The first part, as 16-bit symbols: a byte, or `0x8000 | w` for byte `w`
/// of the 32 KB window before the segment.
pub head: Vec<u16>,
/// The rest, plain bytes (decoded by linflate once the window was clean):
/// `tail[tail_off..]` (the first `tail_off` bytes are the window copy).
pub tail: Vec<u8>,
pub tail_off: usize,
/// Bit just past the segment's last block (see `end`).
pub end_bit: usize,
pub end: SegEnd,
/// Whether any marker was emitted (the segment needed the window).
pub had_markers: bool,
/// Members that ended inside this segment; the first `head_marks` of them
/// fall in the head, the rest in the tail.
pub marks: Vec<Mark>,
pub head_marks: usize,
/// CRC32 of the tail, in pieces split at the marks that fall in the tail
/// (computed by the worker; the in-order stage combines them).
pub tail_crcs: Vec<flate2::Crc>,
}
impl MarkedSeg {
/// The plain-byte part of the output.
pub fn tail_bytes(&self) -> &[u8] {
&self.tail[self.tail_off..]
}
}
struct Marked<'a> {
out: &'a mut Vec<u16>,
last_marker: Option<usize>,
any: bool,
}
impl Marked<'_> {
#[inline(always)]
fn copy(&mut self, len: usize, dist: usize) -> Result<(), GzError> {
let n = self.out.len();
if dist > n + WINDOW {
return Err(GzError("bitsplit: distance beyond the window"));
}
if dist <= n {
// inside this segment's own output: a block copy (or a repeat for
// an overlapping one), then one scan of what came across for markers
let src = n - dist;
if dist >= len {
self.out.extend_from_within(src..src + len);
} else {
self.out.reserve(len);
for k in 0..len {
let v = self.out[src + k];
self.out.push(v);
}
}
if let Some(r) = self.out[n..].iter().rposition(|&v| v & MARKER != 0) {
self.last_marker = Some(n + r);
self.any = true;
}
return Ok(());
}
for k in 0..len {
let p = n + k;
let v = if dist > p {
MARKER | (WINDOW - (dist - p)) as u16
} else {
self.out[p - dist]
};
if v & MARKER != 0 {
self.last_marker = Some(p);
self.any = true;
}
self.out.push(v);
}
Ok(())
}
fn clean_window(&self) -> bool {
let n = self.out.len();
n >= WINDOW && self.last_marker.is_none_or(|m| n - m > WINDOW)
}
}
/// Decode one block (header at `b.pos`) into `m`. Returns BFINAL.
fn marked_block(b: &mut Bits, codes: &mut Codes, m: &mut Marked) -> Result<bool, GzError> {
let bfinal = b.take(1)? != 0;
match b.take(2)? {
0 => {
b.pos = b.pos.div_ceil(8) * 8;
let len = b.take(16)? as usize;
let nlen = b.take(16)? as usize;
if len != !nlen & 0xffff {
return Err(GzError("bitsplit: bad stored length"));
}
let s = b.pos / 8;
if s + len > b.d.len() {
return Err(GzError("bitsplit: unexpected end of input"));
}
m.out.extend(b.d[s..s + len].iter().map(|&x| x as u16));
b.pos += len * 8;
return Ok(bfinal);
}
1 => codes.fixed(),
2 => read_dynamic(b, codes)?,
_ => return Err(GzError("bitsplit: invalid block type")),
}
// One 64-bit read per symbol: a literal/length code (≤ 15 bits), its
// extra bits (≤ 5), a distance code (≤ 15) and its extra bits (≤ 13) are
// at most 48 bits, and `word()` always holds at least 57.
let lmask = (1u64 << codes.lit.bits) - 1;
let dmask = (1u64 << codes.dist.bits) - 1;
loop {
if b.pos > b.end {
return Err(GzError("bitsplit: unexpected end of input"));
}
let w = b.word();
let e = codes.lit.t[(w & lmask) as usize];
let l = (e >> 12) as u32;
if l == 0 {
return Err(GzError("bitsplit: invalid Huffman code"));
}
let sym = (e & 0x0fff) as usize;
if sym < 256 {
m.out.push(sym as u16);
b.pos += l as usize;
continue;
}
if sym == 256 {
b.pos += l as usize;
if b.pos > b.end {
return Err(GzError("bitsplit: unexpected end of input"));
}
return Ok(bfinal);
}
let i = sym - 257;
if i >= 29 {
return Err(GzError("bitsplit: invalid length symbol"));
}
let mut used = l;
let le = LEN_EXTRA[i] as u32;
let len = LEN_BASE[i] as usize + ((w >> used) & ((1u64 << le) - 1)) as usize;
used += le;
let e2 = codes.dist.t[((w >> used) & dmask) as usize];
let l2 = (e2 >> 12) as u32;
if l2 == 0 {
return Err(GzError("bitsplit: invalid Huffman code"));
}
used += l2;
let dsym = (e2 & 0x0fff) as usize;
if dsym >= 30 {
return Err(GzError("bitsplit: invalid distance symbol"));
}
let de = DIST_EXTRA[dsym] as u32;
let dist = DIST_BASE[dsym] as usize + ((w >> used) & ((1u64 << de) - 1)) as usize;
used += de;
b.pos += used as usize;
m.copy(len, dist)?;
}
}
// ── Finding block starts ─────────────────────────────────────────────────────
/// 64 candidate offsets at once. `w` holds the stream bits from some position
/// `p` (bit 0 = bit `p`); bit `j` of the result is set when offset `p + j` can
/// start a non-final dynamic block: BFINAL=0, BTYPE=2, HLIT ≤ 29, HDIST ≤ 29.
/// Plain integer shifts and ANDs over a 128-bit value: the same bitmask an
/// AVX2 compare would give, on every target.
#[inline(always)]
pub fn candidates64(w: u128) -> u64 {
let s = |k: u32| w >> k;
let hlit_bad = s(4) & s(5) & s(6) & s(7);
let hdist_bad = s(9) & s(10) & s(11) & s(12);
(!w & !s(1) & s(2) & !hlit_bad & !hdist_bad) as u64
}
/// Cheap second filter: the 3-bit code-length code must be complete.
#[inline]
fn precode_ok(d: &[u8], pos: usize) -> bool {
let b = Bits::new(d, pos + 13);
let hclen = b.peek(4) as usize + 4;
let mut w = Bits::new(d, pos + 17).word();
let mut cl = [0u8; 19];
for &i in CL_ORDER.iter().take(hclen) {
cl[i] = (w & 7) as u8;
w >>= 3;
}
kraft(&cl, 7) == Kraft::Complete
}
/// Full check of a candidate: the header builds valid codes, the block decodes
/// to its end-of-block, and the next block header is valid too.
fn confirm(d: &[u8], pos: usize) -> bool {
// per-thread scratch: the candidate check runs thousands of times per
// window and must not allocate
thread_local! {
static SCRATCH: std::cell::RefCell<(Codes, Vec<u16>)> =
std::cell::RefCell::new((Codes::new(), Vec::with_capacity(64 * 1024)));
}
SCRATCH.with(|s| {
let mut s = s.borrow_mut();
let (codes, out) = &mut *s;
out.clear();
confirm_with(d, pos, codes, out)
})
}
fn confirm_with(d: &[u8], pos: usize, codes: &mut Codes, out: &mut Vec<u16>) -> bool {
let mut b = Bits::new(d, pos);
let mut m = Marked {
out,
last_marker: None,
any: false,
};
match marked_block(&mut b, codes, &mut m) {
Ok(false) => {}
_ => return false,
}
if m.out.len() < 1024 {
return false; // a real dynamic block carries far more than this
}
// The next header: a stored block's LEN/NLEN, or a full dynamic header.
let mut n = Bits::new(d, b.pos);
if n.pos + 3 > n.end {
return false;
}
let _bfinal = n.take(1);
match n.take(2) {
Ok(0) => {
n.pos = n.pos.div_ceil(8) * 8;
matches!((n.take(16), n.take(16)), (Ok(l), Ok(nl)) if l == !nl & 0xffff)
}
Ok(1) => true,
Ok(2) => read_dynamic(&mut n, codes).is_ok(),
_ => false,
}
}
/// The first real block start at or after bit `from`, before bit `to`.
pub fn find_block_start(d: &[u8], from: usize, to: usize) -> Option<usize> {
find_block_start_with(simd::chosen(), d, from, to)
}
/// [`find_block_start`] with a given filter version (tests and benches).
pub fn find_block_start_with(imp: simd::Impl, d: &[u8], from: usize, to: usize) -> Option<usize> {
let to = to.min(d.len() * 8);
let mut p = from / 8 * 8; // byte aligned: the words load straight from `d`
while p < to {
let b = p / 8;
let mut w = [0u64; simd::WORDS];
if b + 8 * simd::WORDS <= d.len() {
for (k, v) in w.iter_mut().enumerate() {
*v = u64::from_le_bytes(d[b + 8 * k..b + 8 * k + 8].try_into().unwrap());
}
} else {
let mut t = [0u8; 8 * simd::WORDS];
let n = d.len().saturating_sub(b).min(t.len());
t[..n].copy_from_slice(&d[b..b + n]);
for (k, v) in w.iter_mut().enumerate() {
*v = u64::from_le_bytes(t[8 * k..8 * k + 8].try_into().unwrap());
}
}
let m = simd::masks_with(imp, &w);
for (j, &mj) in m.iter().enumerate() {
let base = p + 64 * j;
let mut mask = mj;
if base < from {
mask &= !0u64 << (from - base).min(63);
if from - base >= 64 {
mask = 0;
}
}
if base + 64 > to {
mask &= if to <= base {
0
} else {
(1u64 << (to - base)) - 1
};
}
while mask != 0 {
let q = base + mask.trailing_zeros() as usize;
mask &= mask - 1;
if precode_ok(d, q) && confirm(d, q) {
return Some(q);
}
}
}
p += 64 * simd::LANES;
}
None
}
// ── Segment decode ───────────────────────────────────────────────────────────
fn plain_into(
d: &[u8],
start_bit: usize,
stop_bit: usize,
window: &[u8],
hint: usize,
) -> Result<(Vec<u8>, linflate::BitsDecoded), GzError> {
// returns the buffer WITH the window prefix: the new bytes are
// `buf[window.len()..]`
let cap = linflate::max_inflated_size(d.len() - start_bit / 8);
let mut size = hint.clamp(256 * 1024, cap.max(256 * 1024));
loop {
let mut buf = pool::take_u8(window.len() + size + linflate::OVERWRITE_HEADROOM);
buf.extend_from_slice(window);
// SAFETY: linflate writes before it reads every byte past the window
// (back-references are bounds-checked against the bytes written), and
// we truncate to what it wrote.
#[allow(clippy::uninit_vec)]
unsafe {
buf.set_len(window.len() + size + linflate::OVERWRITE_HEADROOM)
};
match linflate::inflate_bits(d, start_bit, stop_bit, &mut buf, window.len()) {
Ok(r) => {
buf.truncate(window.len() + r.written);
return Ok((buf, r));
}
Err(linflate::InflateError::OutputOverflow) if size < cap => {
pool::give_u8(buf);
size = size.saturating_mul(2).min(cap)
}
Err(_) => return Err(GzError("bitsplit: segment decode failed")),
}
}
}
/// A gzip header at `d[h..]`: `Ok(Some(len))`, `Ok(None)` if it is not all
/// there yet, `Err` if it is not a gzip header.
pub fn gzip_header_len(d: &[u8]) -> Result<Option<usize>, GzError> {
if d.len() < 10 {
return if d.len() >= 2 && (d[0] != 0x1f || d[1] != 0x8b) {
Err(GzError("trailing data after the gzip stream"))
} else {
Ok(None)
};
}
if d[0] != 0x1f || d[1] != 0x8b || d[2] != 8 {
return Err(GzError("trailing data after the gzip stream"));
}
let f = d[3];
let mut p = 10usize;
if f & 0x04 != 0 {
if p + 2 > d.len() {
return Ok(None);
}
p += 2 + u16::from_le_bytes([d[p], d[p + 1]]) as usize;
}
for flag in [0x08u8, 0x10] {
if f & flag != 0 {
match d.get(p..).and_then(|r| r.iter().position(|&c| c == 0)) {
Some(z) => p += z + 1,
None => return Ok(None),
}
}
}
if f & 0x02 != 0 {
p += 2;
}
if p > d.len() {
return Ok(None);
}
Ok(Some(p))
}
enum After {
/// the trailer is not all in `d`: resume at this bit
Trailer(usize),
/// member ended; the next header (or the stream end) is not all in `d`
End { crc: u32, isize: u32, at: usize },
/// member ended and the next member's DEFLATE starts at this bit
Next { crc: u32, isize: u32, start: usize },
}
/// After a member's final block ends at bit `pos`: read its trailer and the
/// next member's header, if they are in `d`.
fn after_member(d: &[u8], pos: usize) -> Result<After, GzError> {
let a = pos.div_ceil(8);
if a + 8 > d.len() {
return Ok(After::Trailer(a * 8));
}
let crc = u32::from_le_bytes(d[a..a + 4].try_into().unwrap());
let isize = u32::from_le_bytes(d[a + 4..a + 8].try_into().unwrap());
let h = a + 8;
match gzip_header_len(&d[h..])? {
Some(n) => Ok(After::Next {
crc,
isize,
start: (h + n) * 8,
}),
None => Ok(After::End {
crc,
isize,
at: h * 8,
}),
}
}
/// Decode the segment `[start_bit, stop_bit)` of raw DEFLATE `d`.
///
/// `fresh` = the segment starts a gzip member (no window before it, so no
/// markers). Decodes until the first block end at or after `stop_bit`
/// (`usize::MAX` = to the end of the stream).
pub fn decode_marked(
d: &[u8],
start_bit: usize,
stop_bit: usize,
fresh: bool,
) -> Result<MarkedSeg, GzError> {
let stops: &[usize] = if stop_bit == usize::MAX {
&[]
} else {
&[stop_bit]
};
decode_until(d, start_bit, stops, fresh)
}
/// Like [`decode_marked`], with every later split start in `stops`
/// (ascending). If a block ends PAST a stop, that stop was a false start, so
/// the decode simply carries on to the next one: the worker that finds a false
/// start covers it itself, in parallel. A gzip member that ends inside the
/// segment is recorded in `marks` and decoding goes on into the next member.
pub fn decode_until(
d: &[u8],
start_bit: usize,
stops: &[usize],
fresh: bool,
) -> Result<MarkedSeg, GzError> {
decode_from(
d,
start_bit,
stops,
if fresh { Some(&[]) } else { None },
usize::MAX,
)
}
/// [`decode_until`] with the 32 KB window before the segment, when it is
/// known (`Some`: the in-order stage's re-decodes, the one-thread path; an
/// empty window = a member start): straight to linflate's fast path, no
/// markers. `None`: unknown, start in marker mode.
///
/// `soft`: also end at the first block end at or after this bit (not a split
/// start: the one-thread path's read limit); `usize::MAX` = none.
pub fn decode_from(
d: &[u8],
start_bit: usize,
stops: &[usize],
known: Option<&[u8]>,
soft: usize,
) -> Result<MarkedSeg, GzError> {
let fresh = known.is_some();
let mut t = 0usize;
let stop = |t: usize| stops.get(t).copied().unwrap_or(usize::MAX);
let hint = stop(0).saturating_sub(start_bit).min(d.len() * 8) / 8 * 4;
let mut seg = MarkedSeg::default();
let mut pos = start_bit;
let mut head_window: Vec<u8> = known
.map(|w| w[w.len().saturating_sub(WINDOW)..].to_vec())
.unwrap_or_default();
if !fresh {
seg.head = pool::take_u16(4 * WINDOW);
let mut b = Bits::new(d, start_bit);
// the worker's own tables, made once and rebuilt in place per block
let mut guard = CodesGuard(Some(
SEG_CODES
.with(|s| s.borrow_mut().take())
.unwrap_or_else(Codes::new),
));
let codes = guard.0.as_mut().unwrap();
let mut m = Marked {
out: &mut seg.head,
last_marker: None,
any: false,
};
let switched;
loop {
let fin = marked_block(&mut b, codes, &mut m)?;
pos = b.pos;
if fin {
let at = m.out.len();
seg.had_markers = m.any;
match after_member(d, pos)? {
After::Trailer(p) => {
seg.end_bit = p;
seg.end = SegEnd::Trailer;
seg.head_marks = seg.marks.len();
seg.tail_crcs.push(flate2::Crc::new());
return Ok(seg);
}
After::End { crc, isize, at: p } => {
seg.marks.push(Mark { at, crc, isize });
seg.end_bit = p;
seg.end = SegEnd::Header;
seg.head_marks = seg.marks.len();
seg.tail_crcs.push(flate2::Crc::new());
return Ok(seg);
}
After::Next { crc, isize, start } => {
seg.marks.push(Mark { at, crc, isize });
pos = start;
switched = true; // the next member needs no window
break;
}
}
}
while pos > stop(t) {
t += 1;
}
if pos == stop(t) {
seg.had_markers = m.any;
seg.end_bit = pos;
seg.head_marks = seg.marks.len();
seg.tail_crcs.push(flate2::Crc::new());
return Ok(seg);
}
if m.clean_window() {
switched = false;
break;
}
}
seg.had_markers |= m.any;
seg.head_marks = seg.marks.len();
if !switched {
// the last 32 KB are plain bytes: the rest goes to the fast path
head_window = seg.head[seg.head.len() - WINDOW..]
.iter()
.map(|&s| s as u8)
.collect();
}
}
let head_len = seg.head.len();
let mut tail: Vec<u8> = Vec::new();
let mut off = 0usize;
let mut piece = flate2::Crc::new();
loop {
let have = tail.len() - off;
let win: &[u8] = if have >= WINDOW {
&tail[tail.len() - WINDOW..]
} else if have == 0 {
&head_window
} else {
// window = end of the head part + all of the tail so far
let need = WINDOW - have;
let keep = head_window.len().saturating_sub(need);
head_window.drain(..keep);
head_window.extend_from_slice(&tail[off..]);
&head_window
};
let wl = win.len();
let (more, r) = plain_into(d, pos, stop(t).min(soft), win, hint)?;
let new_from = tail.len().max(off);
if tail.is_empty() {
tail = more;
off = wl;
} else {
tail.extend_from_slice(&more[wl..]);
pool::give_u8(more);
}
piece.update(&tail[new_from.max(off)..]);
pos = r.end_bit;
if r.final_block {
let at = head_len + tail.len() - off;
match after_member(d, pos)? {
After::Trailer(p) => {
seg.end_bit = p;
seg.end = SegEnd::Trailer;
break;
}
After::End { crc, isize, at: p } => {
seg.marks.push(Mark { at, crc, isize });
seg.end_bit = p;
seg.end = SegEnd::Header;
break;
}
After::Next { crc, isize, start } => {
seg.marks.push(Mark { at, crc, isize });
seg.tail_crcs.push(std::mem::take(&mut piece));
pos = start;
// the new member references nothing before it
head_window.clear();
if tail.len() > off {
let k = tail.len() - WINDOW.min(tail.len() - off);
head_window.extend_from_slice(&tail[k..]);
}
continue;
}
}
}
if pos >= soft {
seg.end_bit = pos;
break;
}
while pos > stop(t) {
t += 1;
}
if pos == stop(t) {
seg.end_bit = pos;
break;
}
}
seg.tail_crcs.push(piece);
seg.tail = tail;
seg.tail_off = off;
Ok(seg)
}
// ── Split stage ──────────────────────────────────────────────────────────────
/// The first border at or after bit `from`, before bit `to`: a validated
/// flush point or a bit-level block start, whichever comes first.
pub fn find_border(d: &[u8], from: usize, to: usize) -> Option<usize> {
let to = to.min(d.len() * 8);
let fb = from.div_ceil(8);
let lim = (to / 8 + 16).min(d.len());
let flush = if fb < lim {
crate::deflate_scan::find_next_valid_flush(&d[..lim], fb)
.map(|f| f.start * 8)
.filter(|&b| b < to && confirm(d, b))
} else {
None
};
let bit_to = flush.unwrap_or(to);
find_block_start(d, from, bit_to).or(flush)
}
/// Segment bit starts for a chunk of raw DEFLATE.
pub struct BitSplit {
/// Bit starts of the segments; `starts[0]` is the chunk's own start bit.
pub starts: Vec<usize>,
}
/// Split `d[start_bit..]` into up to `n_workers` segments: every worker starts
/// at once from its own nominal point (i × chunk / cores) and takes the first
/// border it finds there.
pub fn split_chunk_bits(d: &[u8], start_bit: usize, n_workers: usize) -> BitSplit {
let total = d.len() * 8;
let span = total.saturating_sub(start_bit);
// never split pieces smaller than this: the split costs more than it gains
const MIN_SEG_BITS: usize = 512 * 1024 * 8;
let n = n_workers.max(1).min((span / MIN_SEG_BITS).max(1));
let found: Vec<Option<usize>> = if n > 1 {
gatling::gatling_forkjoin::gatling_for_each(n - 1, n - 1, |k| {
let from = start_bit + span * (k + 1) / n;
let to = start_bit + span * (k + 2) / n;
find_border(d, from, to)
})
} else {
Vec::new()
};
let mut starts = vec![start_bit];
for b in found.into_iter().flatten() {
if b > *starts.last().unwrap() {
starts.push(b);
}
}
BitSplit { starts }
}
/// Resolve `head` (bytes and window markers) against `win` into `dst`.
pub fn resolve_into(head: &[u16], win: &[u8], dst: &mut [u8]) -> Result<(), GzError> {
if head.iter().all(|&v| v & MARKER == 0) {
for (o, &v) in dst.iter_mut().zip(head) {
*o = v as u8;
}
return Ok(());
}
for (o, &v) in dst.iter_mut().zip(head) {
*o = if v & MARKER == 0 {
v as u8
} else {
let back = WINDOW - (v & !MARKER) as usize;
if back > win.len() {
return Err(GzError("bitsplit: reference before the stream start"));
}
win[win.len() - back]
};
}
Ok(())
}
// ── Library entry (in memory) ────────────────────────────────────────────────
pub use crate::stream::Stats;
/// Decode a gzip slice on `threads` workers, `chunk` bytes at a time, through
/// the streaming pipeline ([`crate::stream::decode`]). Returns the bytes only
/// if every member's trailer matched.
pub fn decompress(gz: &[u8], threads: usize, chunk: usize) -> Result<(Vec<u8>, Stats), GzError> {
let cap = linflate::max_inflated_size(gz.len()).min(if gz.len() >= 4 {
u32::from_le_bytes(gz[gz.len() - 4..].try_into().unwrap()) as usize
} else {
0
});
let mut out = Vec::with_capacity(cap);
let st = crate::stream::decode(gz, &mut out, threads, chunk).map_err(|e| GzError(e.msg))?;
Ok((out, st))
}