ldeflate/src/plan.rs
//! **The two-queue planner** — work that does not earn splitting goes in one
//! queue, work that does goes in another.
//!
//! This is the scheduling core, and it is the part that is actually new. The
//! shape is borrowed from `ljar`, but the borrow needs stating precisely,
//! because ljar does *not* implement the thing this crate needs (see
//! `.nornir/gatling-tricks.md` for the full note):
//!
//! - ljar's **10 MB threshold** (`ljar/src/lib.rs:48`) picks between two *entry
//! points* for a whole archive — an in-memory pass or a streaming reader. It
//! is not a per-item size class.
//! - ljar's **batch of 64** (`ljar/src/reader.rs:20`) is a **RAM bound** — "only
//! one batch of decompressed output is live at a time" — not dispatch
//! amortisation.
//! - ljar dispatches **one gatling unit per entry, with claim size 1**
//! (`ljar/src/parallel.rs:32`, `ljar/src/batch.rs:109`). It never coalesces.
//!
//! So ljar has the *silhouette* of the two-queue design and none of the
//! coalescing. For git objects that gap is the whole problem: dispatching a task
//! to deflate 200 bytes costs more than the deflate. What this module adds is
//! the small queue.
//!
//! ## The three unit kinds
//!
//! | kind | when | why |
//! |---|---|---|
//! | [`Unit::Batch`] | object < `coalesce_below` | one task drains many objects **reusing one `Encoder`**. Measured: this is the ENTIRE win. Reusing a compressor (`reset`, ~1.6 us) instead of building one (`Compress::new`, ~4.1 us) saves ~2.5 us per object; the gatling claim it also amortises is only ~62 ns and never mattered. |
//! | [`Unit::Whole`] | `coalesce_below` ≤ object < `split_above` | one task, one object — the classic axis |
//! | [`Unit::Segment`] | object ≥ `split_above` | one object across many tasks, so a single 400 MB blob cannot serialise a worker |
//!
//! ## Determinism
//!
//! The plan is a pure function of `(object sizes, Config)`. It does not read the
//! worker count, the core count, or anything about completion order. That is
//! what lets the compressor promise byte-identical output at 1 thread and at 32.
use crate::container::Container;
use crate::precompressed::SkipPolicy;
use crate::split::{self, Segment};
use crate::unit::{DeflateError, Encoder, EncoderPool};
/// Tuning for [`compress_objects`]. Every default here is **measured** on this
/// machine (AMD Threadripper PRO 3975WX, 32 cores, 2026-08-03) — see
/// `.nornir/ldeflate-design.md` for the sweeps that produced them, and for the
/// two hypotheses those sweeps FALSIFIED.
///
/// | field | default | what the measurement said |
/// |---|---|---|
/// | `coalesce_below` | 64 KiB | monotone win up to here, no knee: 1.15x fewer CPU-seconds on the sub-64 KiB majority, 1.02x over a whole real repository. Covers 93.8% of real git objects. |
/// | `batch_bytes` | 512 KiB | keeps ~15 units per worker on a 46 k-object corpus — big enough to amortise, small enough that LPT can still balance. |
/// | `claim` | 4 | measured **irrelevant** here (0.3% across 1/4/16). Kept at 4 because it costs nothing; the house note that claim 1 is a 2x regression is about far cheaper units than a deflate. |
/// | `split_above` / `segment` | 4 MiB / 1 MiB | 10.3x wall throughput on large compressible objects for a 2.2% ratio cost. `segment` must stay <= `split_above` or splitting never fires. |
#[derive(Debug, Clone, Copy)]
pub struct Config {
/// DEFLATE level, 0-9.
pub level: u32,
/// Objects **strictly smaller** than this are coalesced into batches.
pub coalesce_below: usize,
/// A coalesced batch closes once it holds this many source bytes...
pub batch_bytes: usize,
/// ...or this many objects, whichever comes first.
pub batch_count: usize,
/// Objects **at or above** this are split within-stream.
pub split_above: usize,
/// Uncompressed bytes per segment on the split path.
pub segment: usize,
/// Worker threads. `0` ⇒ one per core.
pub workers: usize,
/// gatling claim size — indices taken per atomic `fetch_add`. `1` is a
/// measured regression on cheap units; the planner already coalesces, so
/// this is a second-order knob.
pub claim: usize,
/// LZ77 window, `2^window_bits` bytes (9..=15). See
/// [`Encoder::new_with_window`] — for git-sized objects the default 15 is
/// mostly a per-object state-reset tax, not compression.
pub window_bits: u8,
/// Which framing wraps each output stream: zlib (default), gzip, or bare
/// DEFLATE. The scheduler is identical for all three — see
/// [`crate::container`].
pub container: Container,
/// Whether to skip the entropy coder for content that is already
/// compressed. **On by default** — see [`crate::precompressed`].
///
/// Applies to all three containers at once because it is applied where the
/// bytes are compressed ([`run_unit`]), not in any one container's writer.
pub skip: SkipPolicy,
}
impl Default for Config {
fn default() -> Self {
Config {
level: 6,
coalesce_below: 64 * 1024,
batch_bytes: 512 * 1024,
batch_count: 256,
split_above: 4 << 20,
segment: 1 << 20,
workers: 0,
claim: 4,
window_bits: crate::unit::WINDOW_BITS_DEFAULT,
container: Container::Zlib,
skip: SkipPolicy::resolve(),
}
}
}
impl Config {
/// Level 6 — git's own default, and the level `git repack` uses unless told
/// otherwise.
pub fn level(mut self, level: u32) -> Self {
self.level = level;
self
}
/// Force a worker count (`0` ⇒ one per core). Used by the scaling bench.
pub fn workers(mut self, w: usize) -> Self {
self.workers = w;
self
}
/// Disable coalescing — the "one task per object" arm the bench compares
/// against. Sets the small threshold to zero so no object is ever batched.
pub fn without_coalescing(mut self) -> Self {
self.coalesce_below = 0;
self
}
/// Disable within-stream splitting — the "objects only" arm.
pub fn without_splitting(mut self) -> Self {
self.split_above = usize::MAX;
self
}
/// Emit each stream in `container` instead of zlib.
pub fn container(mut self, container: Container) -> Self {
self.container = container;
self
}
/// Set the already-compressed skip policy (default: probe the bytes).
pub fn skip(mut self, skip: SkipPolicy) -> Self {
self.skip = skip;
self
}
/// Run the entropy coder over everything, probing nothing — the pre-probe
/// behaviour, by name. The arm the bench compares against.
pub fn compress_everything(mut self) -> Self {
self.skip = SkipPolicy::compress_everything();
self
}
}
/// One dispatchable piece of work. Holds only indices and ranges — never bytes.
/// The objects stay where the caller put them and every worker reads them
/// through a shared `&[&[u8]]`; nothing is cloned into the pool.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Unit {
/// Objects `[lo, hi)`, each compressed to its own stream, all by one worker.
Batch { lo: usize, hi: usize },
/// One whole object.
Whole { idx: usize },
/// Segment `seg` of object `idx`, covering source bytes `[start, end)`.
Segment {
idx: usize,
seg: usize,
start: usize,
end: usize,
is_last: bool,
},
}
/// Build the work plan for `sizes` under `cfg`.
///
/// Pure: same inputs ⇒ same plan, always, on any machine with any core count.
///
/// # Errors
///
/// [`DeflateError::TooManyPieces`] if any one object would be cut into more
/// than [`crate::limits::MAX_PIECES`] segments at `cfg.segment`. The object
/// *count* is deliberately not capped — those buffers are the caller's and
/// already resident; only counts this crate amplifies out of a size parameter
/// are checked. See [`crate::limits`].
pub fn plan(sizes: &[usize], cfg: &Config) -> Result<Vec<Unit>, DeflateError> {
let mut units = Vec::new();
let mut batch_lo: Option<usize> = None;
let mut batch_bytes = 0usize;
let mut batch_count = 0usize;
// Close the currently-open coalesced batch, if any.
let flush = |units: &mut Vec<Unit>,
batch_lo: &mut Option<usize>,
batch_bytes: &mut usize,
batch_count: &mut usize,
hi: usize| {
if let Some(lo) = batch_lo.take() {
if hi > lo {
units.push(Unit::Batch { lo, hi });
}
}
*batch_bytes = 0;
*batch_count = 0;
};
for (idx, &size) in sizes.iter().enumerate() {
if size < cfg.coalesce_below {
// ── small queue: accumulate ─────────────────────────────────────
if batch_lo.is_none() {
batch_lo = Some(idx);
}
batch_bytes += size;
batch_count += 1;
if batch_bytes >= cfg.batch_bytes || batch_count >= cfg.batch_count {
flush(
&mut units,
&mut batch_lo,
&mut batch_bytes,
&mut batch_count,
idx + 1,
);
}
continue;
}
// A non-small object ends any open batch (batches stay contiguous, so
// reassembly is a straight walk with no index bookkeeping).
flush(
&mut units,
&mut batch_lo,
&mut batch_bytes,
&mut batch_count,
idx,
);
if size >= cfg.split_above {
// ── large queue: split within the stream ────────────────────────
for (seg, (start, end, is_last)) in split::plan_segments(size, cfg.segment)?
.into_iter()
.enumerate()
{
units.push(Unit::Segment {
idx,
seg,
start,
end,
is_last,
});
}
} else {
units.push(Unit::Whole { idx });
}
}
flush(
&mut units,
&mut batch_lo,
&mut batch_bytes,
&mut batch_count,
sizes.len(),
);
Ok(units)
}
/// LPT weight for one unit — the source bytes it covers. `gatling_map_balanced`
/// schedules heaviest-first, so the 400 MB blob's segments are claimed before
/// the 200-byte trees and the cheap tail fills the gaps.
#[inline]
pub fn weight(u: &Unit, sizes: &[usize]) -> u64 {
match *u {
Unit::Batch { lo, hi } => sizes[lo..hi].iter().map(|&s| s as u64).sum(),
Unit::Whole { idx } => sizes[idx] as u64,
Unit::Segment { start, end, .. } => (end - start) as u64,
}
}
/// What one unit produced. The `u32` beside each finished stream is the
/// whole-object checksum under the container's algorithm, computed **on the
/// worker** — and only when the caller asked for it (`want_sum`), so the git
/// path never pays for a checksum it does not use.
enum UnitOut {
Batch(Vec<(Vec<u8>, u32)>),
Whole((Vec<u8>, u32)),
Segment(Segment),
}
/// Run one unit on a worker thread. `pool_wrapped` / `pool_raw` hand out that
/// worker's reusable encoders (zlib-wrapped for whole objects, bare for
/// segments).
///
/// **Both checkouts are lazy and each happens at most once per unit.** A
/// coalesced batch of ~95 git objects therefore pays one pool round-trip, and a
/// unit that never reaches the bare encoder never constructs one. Before the
/// pool this function's caller built BOTH encoders unconditionally, so the git
/// path — for which the bare one is dead code — paid `Compress::new` (4082 ns)
/// per unit for an encoder that was never touched.
fn run_unit(
u: &Unit,
objects: &[&[u8]],
pool_wrapped: &EncoderPool,
pool_raw: &EncoderPool,
level: u32,
container: Container,
want_sum: bool,
skip: SkipPolicy,
) -> Result<UnitOut, DeflateError> {
// For zlib the encoder writes the wrapper itself (fast path, and it computes
// Adler-32 internally). For gzip and raw the body is compressed bare and
// framed here, so there is exactly one framing code path.
// The checksum is an extra pass over the input, so it is computed only when
// a caller needs it out of band (ZIP headers). For zlib and gzip the trailer
// already carries it and `want_sum` is false, so the git path is unchanged.
//
// There is only ONE encoder in play here, not two: `pool_wrapped` is built
// with `zlib_wrapped = container.encoder_is_zlib_wrapped()`, so on the
// gzip/raw containers it *already is* the bare encoder the `else` arm wanted.
// The old two-encoder signature made that look like a choice; it never was,
// and the second encoder was pure construction cost. `pool_raw` is reached
// only by [`Unit::Segment`].
let one = |src: &[u8], ew: &mut Encoder| -> Result<(Vec<u8>, u32), DeflateError> {
let sum = if want_sum { container.checksum(src) } else { 0 };
// ── the skip, for every container at once ───────────────────────────
// Deciding here rather than in gzip.rs/zip.rs is the whole point: this
// closure is the only place in the crate where an object's bytes meet
// the entropy coder, so one branch covers zlib, gzip, ZIP and JAR and
// there is no per-format copy to keep in agreement. `stored_deflate` is
// a length-prefixed memcpy, and `Container::wrap` already knows how to
// frame a bare body for each container.
if skip.skip(src) {
let body = crate::precompressed::stored_deflate(src, true);
return Ok((container.wrap(level, &body, src), sum));
}
let mut o = Vec::new();
ew.compress_into(src, &mut o)?;
if container.encoder_is_zlib_wrapped() {
Ok((o, sum))
} else {
Ok((container.wrap(level, &o, src), sum))
}
};
match *u {
Unit::Batch { lo, hi } => {
// ONE checkout for the whole batch — ~95 objects on the measured
// corpus — so the pool lock is touched twice per ~95 deflates.
let mut ew = pool_wrapped.checkout()?;
let mut outs = Vec::with_capacity(hi - lo);
for obj in &objects[lo..hi] {
outs.push(one(obj, &mut ew)?);
}
Ok(UnitOut::Batch(outs))
}
Unit::Whole { idx } => {
let mut ew = pool_wrapped.checkout()?;
Ok(UnitOut::Whole(one(objects[idx], &mut ew)?))
}
Unit::Segment {
idx,
start,
end,
is_last,
..
} => {
// The probe is on the WHOLE object, not on this segment's slice, so
// every segment of one object agrees — a segment from the middle of
// a PNG has no magic bytes of its own, and a plan that stored some
// segments and deflated others would still be correct but would
// depend on where the cuts landed.
let mut er = pool_raw.checkout()?;
Ok(UnitOut::Segment(split::compress_segment_skipping(
&mut er,
objects[idx],
start,
end,
is_last,
container,
skip.skip(objects[idx]),
)?))
}
}
}
/// Compress every object in `objects` into its own complete zlib (RFC 1950)
/// stream, in input order.
///
/// This is the entry point Gunnar's repack/pack-serve path wants: thousands of
/// independent small streams, plus a thin tail of large ones, all fanned out
/// across one no-barrier gatling pass.
///
/// **Zero copy:** `objects` is borrowed for the whole call. Workers read the
/// caller's bytes in place; the only allocation on the hot path is each
/// object's output buffer, which has to exist anyway.
///
/// **Zero wait:** every unit is claimed off one shared atomic cursor. A worker
/// that finishes a 200-byte tree grabs the next unit immediately; it never waits
/// on a barrier, a scheduler, or a sibling that drew a 400 MB blob.
pub fn compress_objects(objects: &[&[u8]], cfg: &Config) -> Result<Vec<Vec<u8>>, DeflateError> {
Ok(compress_objects_inner(objects, cfg, false)?
.into_iter()
.map(|(b, _)| b)
.collect())
}
/// [`compress_objects`], also returning each object's whole-object checksum
/// under `cfg.container`.
///
/// For a container that stores its checksum **out of band** — ZIP keeps each
/// entry's CRC-32 in the local and central headers, not in a trailer — this
/// avoids a second serial pass over every entry. The checksum is computed on the
/// worker that compressed the object, or folded from the per-segment checksums
/// when the object took the split path.
pub fn compress_objects_checksummed(
objects: &[&[u8]],
cfg: &Config,
) -> Result<Vec<(Vec<u8>, u32)>, DeflateError> {
compress_objects_inner(objects, cfg, true)
}
fn compress_objects_inner(
objects: &[&[u8]],
cfg: &Config,
want_sum: bool,
) -> Result<Vec<(Vec<u8>, u32)>, DeflateError> {
let sizes: Vec<usize> = objects.iter().map(|o| o.len()).collect();
let units = plan(&sizes, cfg)?;
if units.is_empty() {
return Ok(Vec::new());
}
let level = cfg.level;
let wb = cfg.window_bits;
let ct = cfg.container;
// Two pools, both LAZY: nothing is constructed until a unit asks for it, so
// the git path — which never reaches `pool_r` — constructs from it ZERO
// times. Owned by this call, not a `thread_local!`, so the crate stays
// reentrant.
let pool_z = EncoderPool::new(level, ct.encoder_is_zlib_wrapped(), wb);
let pool_r = EncoderPool::new(level, false, wb);
// The ONE fan-out. LPT (heaviest first) + batched atomic claims; results
// come back in unit order, so reassembly below is a single forward walk.
let results: Vec<Result<UnitOut, DeflateError>> =
gatling::gatling_forkjoin::gatling_map_balanced(
&units,
cfg.workers,
cfg.claim,
|u| weight(u, &sizes),
|_i, u| {
// Encoders come from the pool, so `Compress::new` is paid once
// per WORKER for the whole call instead of twice per unit.
run_unit(u, objects, &pool_z, &pool_r, level, ct, want_sum, cfg.skip)
},
);
// ── Reassemble, in order, on one thread ─────────────────────────────────
let mut out: Vec<(Vec<u8>, u32)> = Vec::with_capacity(objects.len());
let mut pending: Vec<Segment> = Vec::new();
let mut pending_idx: Option<usize> = None;
for (u, r) in units.iter().zip(results) {
match (u, r?) {
(Unit::Batch { .. }, UnitOut::Batch(v)) => out.extend(v),
(Unit::Whole { .. }, UnitOut::Whole(v)) => out.push(v),
(Unit::Segment { idx, is_last, .. }, UnitOut::Segment(s)) => {
debug_assert!(pending_idx.is_none_or(|p| p == *idx));
pending_idx = Some(*idx);
pending.push(s);
if *is_last {
out.push(split::assemble_checksummed(level, &pending, ct));
pending.clear();
pending_idx = None;
}
}
_ => unreachable!("unit kind and output kind cannot disagree"),
}
}
debug_assert!(pending.is_empty(), "a split object was left unassembled");
debug_assert_eq!(
out.len(),
objects.len(),
"one output stream per input object"
);
gatling::functional_status(
"ldeflate",
"compress_objects",
true,
&"objects fanned out over one no-barrier gatling pass",
);
Ok(out)
}
/// The serial reference. Same output, one thread, no gatling — the baseline
/// every scaling number in the bench is divided by, and the oracle the parallel
/// path is asserted equal to.
pub fn compress_objects_serial(
objects: &[&[u8]],
cfg: &Config,
) -> Result<Vec<Vec<u8>>, DeflateError> {
let sizes: Vec<usize> = objects.iter().map(|o| o.len()).collect();
let units = plan(&sizes, cfg)?;
// The same two pools the parallel path uses, so the serial oracle and the
// scheduler run byte-identical code (LAW 5). With one thread a pool is a
// one-element free list: `Compress::new` once, `reset` thereafter.
let pool_z = EncoderPool::new(
cfg.level,
cfg.container.encoder_is_zlib_wrapped(),
cfg.window_bits,
);
let pool_r = EncoderPool::new(cfg.level, false, cfg.window_bits);
let mut out: Vec<Vec<u8>> = Vec::with_capacity(objects.len());
let mut pending: Vec<Segment> = Vec::new();
for u in &units {
match run_unit(
u,
objects,
&pool_z,
&pool_r,
cfg.level,
cfg.container,
false,
cfg.skip,
)? {
UnitOut::Batch(v) => out.extend(v.into_iter().map(|(b, _)| b)),
UnitOut::Whole((b, _)) => out.push(b),
UnitOut::Segment(s) => {
let last = matches!(u, Unit::Segment { is_last: true, .. });
pending.push(s);
if last {
out.push(split::assemble(cfg.level, &pending, cfg.container));
pending.clear();
}
}
}
}
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
fn objs(sizes: &[usize]) -> Vec<Vec<u8>> {
sizes
.iter()
.enumerate()
.map(|(i, &n)| {
// Distinct, compressible-but-not-trivial content per object, so a
// mis-ordered reassembly cannot pass by luck.
let tag = format!("object-{i}-");
let mut v = Vec::with_capacity(n);
while v.len() < n {
v.extend_from_slice(tag.as_bytes());
v.extend_from_slice(b"payload payload payload ");
}
v.truncate(n);
v
})
.collect()
}
fn refs(v: &[Vec<u8>]) -> Vec<&[u8]> {
v.iter().map(|x| x.as_slice()).collect()
}
/// The plan must cover every object exactly once, in order — whatever the
/// mix of sizes and thresholds.
#[test]
fn plan_covers_every_object_exactly_once() {
let cfg = Config::default();
let sizes: Vec<usize> = (0..500)
.map(|i| match i % 7 {
0 => 120,
1 => 900,
2 => 4_000,
3 => 70_000,
4 => 300_000,
5 => 3_000_000,
_ => 40,
})
.collect();
let units = plan(&sizes, &cfg).unwrap();
let mut seen = vec![0usize; sizes.len()];
let mut order = Vec::new();
for u in &units {
match *u {
Unit::Batch { lo, hi } => {
for i in lo..hi {
seen[i] += 1;
order.push(i);
}
}
Unit::Whole { idx } => {
seen[idx] += 1;
order.push(idx);
}
Unit::Segment { idx, is_last, .. } => {
if is_last {
seen[idx] += 1;
order.push(idx);
}
}
}
}
assert!(
seen.iter().all(|&c| c == 1),
"every object exactly once: {seen:?}"
);
assert!(
order.windows(2).all(|w| w[0] < w[1]),
"objects must stay in order"
);
}
/// **The scaling-bench guard, seen red.** The parallel path and the serial
/// path must produce byte-identical output at every worker count. If they
/// did not, the bench would be timing two different computations and its
/// speedup number would be meaningless.
#[test]
fn parallel_equals_serial_at_every_worker_count() {
let data = objs(&[80, 300, 900, 64_000, 200_000, 12 << 20, 40, 5, 1_500_000]);
let r = refs(&data);
let base = compress_objects_serial(&r, &Config::default()).unwrap();
for w in [1usize, 2, 3, 8, 32] {
let got = compress_objects(&r, &Config::default().workers(w)).unwrap();
assert_eq!(
got, base,
"workers={w}: output differs from the serial oracle"
);
}
}
/// Determinism across worker counts is only meaningful if the output would
/// actually *change* under a thread-count-dependent plan. Prove the plan is
/// the thing carrying determinism: a plan built for a different segment size
/// must produce different bytes, so "identical" above is not vacuous.
#[test]
fn segment_size_actually_changes_the_bytes() {
// Must exceed `split_above` (4 MiB) or nothing is split and the two
// configs trivially agree — the failure mode this test exists to catch.
let data = objs(&[24 << 20]);
let r = refs(&data);
assert!(
data[0].len() >= Config::default().split_above,
"fixture must reach the split queue"
);
let a = compress_objects(
&r,
&Config {
segment: 256 << 10,
..Config::default()
},
)
.unwrap();
let b = compress_objects(
&r,
&Config {
segment: 1 << 20,
..Config::default()
},
)
.unwrap();
assert_ne!(
a, b,
"two different segment sizes produced identical bytes — the split path \
is not actually segmenting, so the determinism test above proves nothing"
);
}
/// Every stream this crate emits, whichever queue produced it, must inflate
/// back to its object.
#[test]
fn every_queue_round_trips() {
use std::io::Read;
// Sizes chosen to hit all three queues under the default config.
let data = objs(&[10, 200, 65_000, 300_000, 9 << 20, 7, 1_100_000]);
let r = refs(&data);
let out = compress_objects(&r, &Config::default()).unwrap();
assert_eq!(out.len(), data.len());
for (i, (z, src)) in out.iter().zip(&data).enumerate() {
let mut got = Vec::new();
flate2::read::ZlibDecoder::new(&z[..])
.read_to_end(&mut got)
.unwrap_or_else(|e| panic!("object {i} ({} B) failed to inflate: {e}", src.len()));
assert_eq!(&got, src, "object {i} ({} B) round-trip differs", src.len());
}
}
/// The coalescing threshold must actually change the plan shape — otherwise
/// `without_coalescing()` (the bench's A/B arm) is a no-op and the batching
/// measurement is fake.
#[test]
fn coalescing_actually_changes_the_unit_count() {
let sizes = vec![200usize; 10_000];
let batched = plan(&sizes, &Config::default()).unwrap();
let unbatched = plan(&sizes, &Config::default().without_coalescing()).unwrap();
assert_eq!(
unbatched.len(),
10_000,
"no-coalesce must be one unit per object"
);
assert!(
batched.len() * 8 < unbatched.len(),
"coalescing produced {} units vs {} — barely fewer, so the small queue \
is not doing anything and the A/B arm is measuring noise",
batched.len(),
unbatched.len()
);
// And the two must still agree byte-for-byte.
let data = objs(&vec![200usize; 400]);
let r = refs(&data);
assert_eq!(
compress_objects(&r, &Config::default()).unwrap(),
compress_objects(&r, &Config::default().without_coalescing()).unwrap(),
"coalescing must not change a single output byte"
);
}
/// Empty input, empty objects, one object — the shapes that break planners.
#[test]
fn degenerate_inputs() {
let cfg = Config::default();
assert!(compress_objects(&[], &cfg).unwrap().is_empty());
let empties: Vec<Vec<u8>> = vec![Vec::new(); 5];
let r = refs(&empties);
let out = compress_objects(&r, &cfg).unwrap();
assert_eq!(out.len(), 5);
use std::io::Read;
for z in &out {
let mut got = Vec::new();
flate2::read::ZlibDecoder::new(&z[..])
.read_to_end(&mut got)
.unwrap();
assert!(got.is_empty());
}
}
}