ldeflate/src/split.rs
//! **Axis B — split within one stream.** The write-side mirror of
//! `linflate::deflate_scan`.
//!
//! `deflate_scan::find_all_flushes` *searches* a compressed stream for
//! `00 00 FF FF` full-flush markers so the read side can hand independent
//! segments to independent workers. This module *creates* them: one large object
//! is cut into fixed byte ranges of the **uncompressed** input, each range is
//! compressed by a different worker with a compressor that has never seen the
//! others, and the results are concatenated.
//!
//! ```text
//! input: [------- seg 0 -------][------- seg 1 -------][-- seg 2 --]
//! worker 7 worker 2 worker 19
//! output: 78 9C │ deflate·FULL │ deflate·FULL │ deflate·FIN │ adler32
//! └─ 00 00 FF FF ────┴─ 00 00 FF FF ───────┘
//! ```
//!
//! ## Why the concatenation is a legal zlib stream
//!
//! `Z_FULL_FLUSH` does two things and this design leans on both. It byte-aligns
//! the bit stream (so the next segment's first byte lands on a byte boundary and
//! can be appended verbatim), and it **empties the LZ77 window** (so no
//! back-reference in segment *k+1* can point into segment *k*, which is exactly
//! why segment *k+1* could be compressed by a worker that never saw segment
//! *k*). The final segment is `Z_FINISH`, which sets `BFINAL=1`. An RFC-1950
//! header goes in front and the whole-input Adler-32 — assembled from the
//! per-segment checksums via `adler32_combine` — goes behind. A stock decoder
//! (`zlib`, `git`, `flate2`, `linflate`) sees one ordinary stream of blocks.
//!
//! This is pigz's trick. It is not novel; what matters is that it costs a small,
//! *measurable* amount of ratio (each window reset forecloses long-range
//! matches) and that the ratio cost is reported honestly rather than hidden.
//!
//! ## Determinism
//!
//! Segment boundaries are a function of `(input length, segment size)` **only** —
//! never of the worker count, never of completion order. So a 4-thread run and a
//! 32-thread run emit byte-identical output. That is asserted, not assumed
//! ([`crate::tests`] / `tests/determinism.rs`).
use crate::container::Container;
use crate::unit::{DeflateError, Encoder};
/// One segment of a split stream: where it came from and what it produced.
#[derive(Debug, Clone)]
pub struct Segment {
/// Byte range of the **uncompressed** input this segment covers.
pub src_start: usize,
pub src_end: usize,
/// The segment's compressed bytes (bare DEFLATE, ending in a full-flush
/// marker unless this is the final segment).
pub bytes: Vec<u8>,
/// Checksum of `input[src_start..src_end]` under the container's algorithm
/// (Adler-32 seeded at 1 for zlib, CRC-32 seeded at 0 for gzip). Computed on
/// the worker over that worker's own bytes, so it parallelises with the
/// compression instead of becoming a serial tail.
pub sum: u32,
}
/// Plan the segment boundaries for an input of `len` bytes at `segment` size.
///
/// Returns `(start, end, is_last)` triples. Deliberately a pure function of its
/// two arguments — this is what makes the compressor deterministic across
/// thread counts.
///
/// # Errors
///
/// [`DeflateError::TooManyPieces`] if `len / segment` exceeds
/// [`crate::limits::MAX_PIECES`]. A `segment` of 1 against a megabyte asks for
/// a million independently compressed, independently checksummed pieces; this
/// refuses **before** the `Vec` is built rather than trying. The ceiling errors
/// instead of clamping `segment` upward because the segment plan is exactly
/// what carries this crate's byte-identical-output promise — see
/// [`crate::limits`].
pub fn plan_segments(
len: usize,
segment: usize,
) -> Result<Vec<(usize, usize, bool)>, DeflateError> {
let n = crate::limits::check_pieces("split segments", len, segment)?;
if len == 0 {
return Ok(vec![(0, 0, true)]);
}
let segment = segment.max(1);
Ok((0..n)
.map(|i| {
let start = i * segment;
let end = ((i + 1) * segment).min(len);
(start, end, i + 1 == n)
})
.collect())
}
/// Compress one segment. Called on a worker thread, with `enc` being that
/// worker's reusable encoder.
///
/// `is_last` selects `Z_FINISH` (BFINAL=1) over `Z_FULL_FLUSH`.
pub fn compress_segment(
enc: &mut Encoder,
src: &[u8],
start: usize,
end: usize,
is_last: bool,
container: Container,
) -> Result<Segment, DeflateError> {
compress_segment_skipping(enc, src, start, end, is_last, container, false)
}
/// [`compress_segment`], optionally storing the segment instead of deflating it.
///
/// `store` is decided by the caller from the **whole** object (see
/// [`crate::precompressed`]), never from this segment's slice: a segment cut out
/// of the middle of a PNG carries no magic bytes of its own, and a decision made
/// per slice would make the output depend on where the cuts landed.
///
/// A stored non-final segment ends on a block boundary, which is the same seam
/// `Z_FULL_FLUSH` leaves, so stored and deflated segments concatenate freely.
pub fn compress_segment_skipping(
enc: &mut Encoder,
src: &[u8],
start: usize,
end: usize,
is_last: bool,
container: Container,
store: bool,
) -> Result<Segment, DeflateError> {
debug_assert!(!enc.is_zlib_wrapped(), "segments must be bare DEFLATE");
let chunk = &src[start..end];
let mut bytes = Vec::new();
if store {
bytes = crate::precompressed::stored_deflate(chunk, is_last);
} else if is_last {
enc.compress_into(chunk, &mut bytes)?;
} else {
enc.compress_segment_into(chunk, &mut bytes)?;
}
let sum = container.checksum(chunk);
Ok(Segment {
src_start: start,
src_end: end,
bytes,
sum,
})
}
/// Assemble planned segments (in stream order) into one complete zlib stream.
///
/// Panics in debug if the segments are not contiguous and in order — that would
/// be a scheduler bug, and a silently mis-ordered assembly produces a stream
/// that inflates to *wrong bytes* rather than failing, which is the worst
/// possible failure mode.
pub fn assemble(level: u32, segments: &[Segment], container: Container) -> Vec<u8> {
assemble_checksummed(level, segments, container).0
}
/// [`assemble`], also returning the folded whole-stream checksum.
///
/// ZIP stores each entry's CRC-32 in its local and central headers rather than
/// in a trailer, so it needs the folded value that [`assemble`] would otherwise
/// consume and discard.
pub fn assemble_checksummed(
level: u32,
segments: &[Segment],
container: Container,
) -> (Vec<u8>, u32) {
debug_assert!(!segments.is_empty(), "assemble needs at least one segment");
debug_assert!(
segments.windows(2).all(|w| w[0].src_end == w[1].src_start),
"segments must be contiguous and in stream order"
);
let body: usize = segments.iter().map(|s| s.bytes.len()).sum();
let total: u64 = segments
.iter()
.map(|s| (s.src_end - s.src_start) as u64)
.sum();
let mut out = Vec::with_capacity(12 + body + 8);
out.extend_from_slice(&container.header(level));
// Fold the per-segment checksums into the whole-stream one. This is the
// only reason `adler32_combine` / `crc32_combine` exist in this crate.
let mut sum = container.checksum_init();
for s in segments {
out.extend_from_slice(&s.bytes);
let n = (s.src_end - s.src_start) as u64;
sum = container.checksum_combine(sum, s.sum, n);
}
out.extend_from_slice(&container.trailer(sum, total));
(out, sum)
}
/// Serial reference implementation of the split path — used by tests as the
/// oracle the parallel scheduler must match byte-for-byte, and by
/// [`crate::plan`] when a caller asks for one worker.
pub fn split_compress_serial(
src: &[u8],
level: u32,
segment: usize,
) -> Result<Vec<u8>, DeflateError> {
split_compress_serial_in(src, level, segment, Container::Zlib)
}
/// [`split_compress_serial`] in an explicit container.
pub fn split_compress_serial_in(
src: &[u8],
level: u32,
segment: usize,
container: Container,
) -> Result<Vec<u8>, DeflateError> {
let mut enc = Encoder::new(level, false)?;
let plan = plan_segments(src.len(), segment)?;
let mut segs = Vec::with_capacity(plan.len());
for (start, end, last) in plan {
segs.push(compress_segment(
&mut enc, src, start, end, last, container,
)?);
}
Ok(assemble(level, &segs, container))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::plan::Config;
use std::io::Read;
fn corpus(n: usize) -> Vec<u8> {
// Compressible, but with enough entropy that segmentation actually
// costs ratio — a corpus of all-zeros would hide any boundary bug
// because every segment compresses to nothing (LAW 2: identity value).
let words: &[&[u8]] = &[
b"fn main() { println!(\"hello\"); }\n",
b"use std::collections::HashMap;\n",
b" let mut total = 0usize;\n",
b"// TODO: revisit this once the planner lands\n",
];
let mut out = Vec::with_capacity(n + 64);
let mut s: u64 = 0x9E3779B97F4A7C15;
while out.len() < n {
s = s
.wrapping_mul(6364136223846793005)
.wrapping_add(1442695040888963407);
out.extend_from_slice(words[(s >> 33) as usize % words.len()]);
}
out.truncate(n);
out
}
/// Assert one (len, seg) pair round-trips through a stock decoder.
fn round_trips(len: usize, seg: usize) {
let src = corpus(len);
let z = split_compress_serial(&src, 6, seg).unwrap();
let mut got = Vec::new();
flate2::read::ZlibDecoder::new(&z[..])
.read_to_end(&mut got)
.unwrap_or_else(|e| panic!("len={len} seg={seg}: inflate failed: {e}"));
assert_eq!(got, src, "len={len} seg={seg}: bytes differ");
}
/// **Degenerate segment sizes, against SMALL inputs.** `seg = 1` and
/// `seg = 2` are exactly the boundary cases worth testing — an empty final
/// segment, a one-byte final segment, a segment count that is not a divisor
/// — and they are cheap here because the largest input is 8 KiB, i.e. at
/// most ~8 k pieces.
///
/// **Do not re-cross these with the large inputs below.** `seg = 1` against
/// a megabyte is a million independently compressed, framed and checksummed
/// pieces; that cross is what OOM-killed this box on 2026-08-03 (the
/// original loop ran `len` up to 1_000_000 against `seg = 1` and took over
/// 6 minutes even under an address-space cap). The two loops together cover
/// strictly more shapes than the cross did and finish in under a second.
/// The library now refuses the absurd end of that cross outright — see
/// `a_degenerate_segment_size_is_refused_not_attempted`.
#[test]
fn degenerate_segment_sizes_round_trip_on_small_inputs() {
for &len in &[0usize, 1, 2, 3, 7, 8, 4095, 4096, 4097, 8192] {
for &seg in &[1usize, 2, 3, 7, 4096] {
round_trips(len, seg);
}
}
}
/// **Large inputs, against REALISTIC segment sizes.** The size axis, kept
/// separate from the degenerate-segment axis above for the reason stated
/// there. Every pair here stays in the low hundreds of pieces.
#[test]
fn large_inputs_round_trip_at_realistic_segment_sizes() {
for &len in &[100_000usize, 1_000_000] {
for &seg in &[4096usize, 65_536, 1 << 20] {
round_trips(len, seg);
}
}
}
/// **The OOM guard, split side.** A `segment` size that would cut the input
/// into an absurd number of pieces must come back as an `Err` — from the
/// planner, from the serial path, and from the public `compress_split`
/// entry point, because all three route through `plan_segments`.
///
/// Seen red on 2026-08-03 with the ceiling check removed from
/// `plan_segments`: every case below returned `Ok`, building a
/// million-element plan, and this test failed on the first `expect_err`.
#[test]
fn a_degenerate_segment_size_is_refused_not_attempted() {
use crate::limits::MAX_PIECES;
// The planner itself — the one writer all three paths share.
let just_over = MAX_PIECES + 1;
let e = plan_segments(just_over, 1).expect_err("one segment per byte must be refused");
assert!(
matches!(e, DeflateError::TooManyPieces { pieces, max, .. }
if pieces == just_over && max == MAX_PIECES),
"wrong error: {e}"
);
assert!(
e.to_string().contains("split segments"),
"message does not name the axis: {e}"
);
// The shape the old cross-product test ran: 1_000_000 bytes at seg = 1.
assert!(
plan_segments(1_000_000, 1).is_err(),
"a million one-byte segments were planned"
);
assert!(
plan_segments(1_000_000, 0).is_err(),
"segment=0 escaped the ceiling"
);
assert!(
split_compress_serial(&vec![b'x'; 1_000_000], 6, 1).is_err(),
"the serial split path attempted a million one-byte segments"
);
assert!(
crate::compress_split(
&vec![b'x'; 1_000_000],
&Config {
segment: 1,
..Config::default()
}
)
.is_err(),
"the public compress_split entry point bypassed the ceiling"
);
assert!(
crate::compress_objects(
&[&vec![b'x'; 1_000_000][..]],
&Config {
split_above: 1,
segment: 1,
..Config::default()
},
)
.is_err(),
"compress_objects bypassed the ceiling"
);
// Not vacuous: at the ceiling and at every realistic segment size the
// planner still says yes.
assert_eq!(plan_segments(MAX_PIECES, 1).unwrap().len(), MAX_PIECES);
assert!(
plan_segments(256 << 30, 1 << 20).is_ok(),
"256 GiB at 1 MiB segments"
);
assert!(split_compress_serial(&vec![b'x'; 1_000_000], 6, 4096).is_ok());
}
/// RED-when-broken: the Adler-32 trailer must be *checked* by the decoder.
/// `flate2`'s `ZlibDecoder` validates it — so corrupting one trailer byte
/// must make the read above FAIL. Without this, the round-trip test above
/// would pass even if `adler32_combine` were returning garbage, because the
/// payload bytes would still be right. This is the guard on the guard.
#[test]
fn a_wrong_adler_trailer_is_actually_rejected() {
let src = corpus(200_000);
let mut z = split_compress_serial(&src, 6, 16_384).unwrap();
// Sanity: it reads clean first.
let mut ok = Vec::new();
flate2::read::ZlibDecoder::new(&z[..])
.read_to_end(&mut ok)
.unwrap();
assert_eq!(ok, src);
// Now break the trailer. If the decoder did not verify it, this would
// still succeed and the test above would be worthless.
let n = z.len();
z[n - 1] ^= 0xff;
let mut bad = Vec::new();
let r = flate2::read::ZlibDecoder::new(&z[..]).read_to_end(&mut bad);
assert!(
r.is_err(),
"the decoder did NOT verify the Adler-32 trailer — the round-trip \
tests in this module prove nothing about adler32_combine"
);
}
/// RED-when-broken: if the final segment were emitted with FULL_FLUSH
/// instead of FINISH, no `BFINAL=1` block would ever be written. Prove a
/// decoder actually notices — otherwise `compress_segment`'s `is_last`
/// branch is untested.
#[test]
fn a_stream_with_no_final_block_is_rejected() {
let src = corpus(50_000);
let mut enc = Encoder::new(6, false).unwrap();
let plan = plan_segments(src.len(), 8192).unwrap();
let mut segs = Vec::new();
for (start, end, _last) in plan {
// Deliberately pass is_last = false for EVERY segment.
segs.push(
compress_segment(&mut enc, &src, start, end, false, Container::Zlib).unwrap(),
);
}
let z = assemble(6, &segs, Container::Zlib);
let mut got = Vec::new();
let r = flate2::read::ZlibDecoder::new(&z[..]).read_to_end(&mut got);
assert!(
r.is_err() || got != src,
"a stream whose last block is not BFINAL was accepted as complete — \
the `is_last` branch of compress_segment is not being tested"
);
}
/// The plan is a pure function of (len, segment): contiguous, covering, in
/// order, exactly one `is_last`.
///
/// Same rule as the round-trip tests above: degenerate `seg` is paired with
/// small `len`, big `len` with realistic `seg`. Crossing them asks the
/// planner for a piece count it now (correctly) refuses, and asked the old
/// planner for a million-element `Vec`.
#[test]
fn segment_plan_is_contiguous_and_covering() {
let check = |len: usize, seg: usize| {
let p = plan_segments(len, seg)
.unwrap_or_else(|e| panic!("len={len} seg={seg}: refused: {e}"));
assert!(!p.is_empty());
assert_eq!(p[0].0, 0, "len={len} seg={seg}: must start at 0");
assert_eq!(
p.last().unwrap().1,
len,
"len={len} seg={seg}: must cover to len"
);
assert_eq!(
p.iter().filter(|(_, _, l)| *l).count(),
1,
"len={len} seg={seg}: exactly one final segment"
);
for w in p.windows(2) {
assert_eq!(w[0].1, w[1].0, "len={len} seg={seg}: gap or overlap");
assert!(
!w[0].2,
"len={len} seg={seg}: is_last set on a non-final segment"
);
}
};
for &len in &[0usize, 1, 999, 1000, 1001, 9999] {
for &seg in &[1usize, 7, 1000] {
check(len, seg);
}
}
for &len in &[1usize << 20, (1 << 20) + 1, 64 << 20] {
for &seg in &[1000usize, 1 << 16, 1 << 20] {
check(len, seg);
}
}
}
/// Splitting costs ratio. Measure it here so the cost is a documented
/// number rather than a surprise — and so a future change that silently
/// stops resetting the window (which would improve ratio but break
/// segment independence) is visible.
#[test]
fn splitting_costs_ratio_and_the_cost_is_bounded() {
let src = corpus(4 << 20);
let whole = crate::unit::zlib_compress(&src, 6).unwrap();
for &seg in &[1 << 20, 256 << 10, 64 << 10] {
let split = split_compress_serial(&src, 6, seg).unwrap();
let overhead = split.len() as f64 / whole.len() as f64;
assert!(
overhead >= 1.0,
"seg={seg}: split output ({}) SMALLER than unsplit ({}) — impossible \
unless the window is not actually being reset",
split.len(),
whole.len()
);
assert!(
overhead < 1.25,
"seg={seg}: split costs {:.1}% more than unsplit — far beyond the \
expected few percent; a boundary is being emitted per block, not per segment",
(overhead - 1.0) * 100.0
);
}
}
}