ldeflate/src/stream.rs
//! **The 31+1 split** — N−1 workers compress, one dedicated thread streams the
//! result down to the device.
//!
//! [`crate::plan::compress_objects`] materialises every output stream in RAM and
//! hands back a `Vec<Vec<u8>>`. For a repack of a large repository that is the
//! wrong shape: peak RSS becomes the size of the whole compressed pack, and the
//! device sits idle for the entire compress phase and then has to absorb
//! everything at once.
//!
//! This module is the streaming entry point. It uses
//! `gatling::gatling::ordered::run_ordered_sink`, whose **collector runs on the
//! calling thread** — so the caller's thread *is* the one dedicated writer, and
//! `n_workers` compressors run beside it. Ask for `cores - 1` workers and the
//! topology is literally 31 compute + 1 I/O on this box.
//!
//! ```text
//! producer (lazy, back-pressured)
//! │ ≤ cap objects alive at any instant
//! ▼
//! [31 workers] compress ──► reorder buffer ──► writer (calling thread)
//! self-dispatch, no barrier bounded by cap one uncontended pwrite
//! ```
//!
//! ## Why one writer and not 32
//!
//! Because on local NVMe a write is a memcpy-class syscall with no round-trip
//! latency to hide, one uncontended writer already saturates the device, and
//! adding writers buys lock, seek and allocator contention. That is not a guess
//! — it is skade's documented rationale for keeping its local path sequential
//! (`skade/.nornir/s3-ingest-concurrency.md`, "Why local stays sequential (and
//! must)"). The inverse holds for object storage, where the cost is per-op
//! latency and only concurrency hides it; that case wants `gatling::io`, not
//! this.
use std::io::Write;
use crate::plan::Config;
use crate::unit::{DeflateError, EncoderPool};
/// Where one object landed in the output stream.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Placed {
/// Byte offset of this object's zlib stream in the output.
pub offset: u64,
/// Length of the zlib stream in bytes.
pub len: u64,
}
/// Default worker count for the streaming path: **one fewer than the box has**,
/// because the calling thread is the writer and wants a core to itself.
pub fn compute_workers() -> usize {
std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(2)
.saturating_sub(1)
.max(1)
}
/// Compress `objects` and stream every result to `out` in input order, with
/// N−1 compressors running beside the single writer.
///
/// Returns one [`Placed`] per object, so the caller can build a pack index
/// without a second pass.
///
/// `cfg.workers == 0` resolves to [`compute_workers`] (i.e. cores − 1), not to
/// "one per core" — the difference is the whole point of this entry point.
///
/// Note this path compresses each object as **one whole stream**; it does not
/// split large objects across workers, because the ordered sink emits per item.
/// For a batch job dominated by a few huge blobs use
/// [`crate::plan::compress_objects`], which does split. For a pack of thousands
/// of small-to-medium objects — the git case — this is the right shape.
pub fn compress_objects_to_writer<W: Write>(
objects: &[&[u8]],
cfg: &Config,
out: &mut W,
) -> Result<Vec<Placed>, DeflateError> {
let workers = if cfg.workers == 0 {
compute_workers()
} else {
cfg.workers
};
let cap = (workers * 4).max(8);
let level = cfg.level;
let wb = cfg.window_bits;
let ct = cfg.container;
let mut next = 0usize;
let producer = move || -> Option<(usize, &[u8])> {
if next < objects.len() {
let i = next;
next += 1;
// Borrowed, never cloned: the worker compresses the caller's bytes.
Some((i, objects[i]))
} else {
None
}
};
// Lazy: the first checkout on each worker constructs, every later one is a
// `reset`. Shared with `plan.rs` rather than twinned (LAW 5).
let pool = EncoderPool::new(level, ct.encoder_is_zlib_wrapped(), wb);
let mut placed: Vec<Placed> = Vec::with_capacity(objects.len());
let mut offset: u64 = 0;
let mut sink_err: Option<DeflateError> = None;
// The collector — and therefore this closure — runs on THIS thread. It is
// the one dedicated I/O core.
let mut sink = |_seq: u64, r: Result<Vec<u8>, DeflateError>| -> anyhow::Result<()> {
match r {
Ok(bytes) => {
out.write_all(&bytes)
.map_err(|e| anyhow::anyhow!("ldeflate: write failed: {e}"))?;
placed.push(Placed {
offset,
len: bytes.len() as u64,
});
offset += bytes.len() as u64;
Ok(())
}
Err(e) => {
sink_err.get_or_insert(e.clone());
Err(anyhow::anyhow!("{e}"))
}
}
};
let run = gatling::gatling::ordered::run_ordered_sink(
producer,
workers,
cap,
|_label: usize, input: &[u8]| -> Result<Vec<u8>, DeflateError> {
// ONE encoder per WORKER for the whole call, not one per object.
// This path called `Encoder::new_with_window` inside the map closure,
// i.e. `Compress::new` (**4082 ns**) for every object — against a
// total per-object cost of about 5 us on a git-median 1130 B object.
// See [`EncoderPool`] for why a pool and not a `thread_local!`.
let mut enc = pool.checkout()?;
let mut o = Vec::new();
enc.compress_into(input, &mut o)?;
if ct.encoder_is_zlib_wrapped() {
Ok(o)
} else {
Ok(ct.wrap(level, &o, input))
}
},
&mut sink,
);
if let Some(e) = sink_err {
return Err(e);
}
run.map_err(|_| DeflateError::Encode("ordered sink aborted"))?;
gatling::functional_status(
"ldeflate",
"compress_objects_to_writer",
true,
&"N-1 compressors + 1 dedicated writer on the calling thread",
);
Ok(placed)
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Read;
fn objs(n: usize) -> Vec<Vec<u8>> {
(0..n)
.map(|i| {
let mut v = Vec::new();
let tag = format!("obj{i}:");
// Wildly uneven sizes, so a barrier or a static carve-up would
// show up as a slow tail.
let target = if i % 97 == 0 {
400_000
} else {
40 + (i * 7) % 3_000
};
while v.len() < target {
v.extend_from_slice(tag.as_bytes());
v.extend_from_slice(b"the quick brown fox ");
}
v.truncate(target);
v
})
.collect()
}
/// The streamed bytes must be the exact concatenation of the per-object
/// streams, **in input order**, with correct offsets — at every worker
/// count. An out-of-order emit corrupts a pack silently, so this is checked
/// by decoding each recorded range back to its object rather than by
/// comparing a hash of the whole file.
#[test]
fn streamed_output_is_ordered_and_offsets_are_right() {
let data = objs(600);
let r: Vec<&[u8]> = data.iter().map(|v| v.as_slice()).collect();
for w in [1usize, 2, 7, 31] {
let mut buf: Vec<u8> = Vec::new();
let placed =
compress_objects_to_writer(&r, &Config::default().workers(w), &mut buf).unwrap();
assert_eq!(placed.len(), data.len(), "workers={w}");
assert_eq!(
placed.last().unwrap().offset + placed.last().unwrap().len,
buf.len() as u64,
"workers={w}: recorded extents must cover the whole output"
);
for (i, p) in placed.iter().enumerate() {
let slice = &buf[p.offset as usize..(p.offset + p.len) as usize];
let mut got = Vec::new();
flate2::read::ZlibDecoder::new(slice)
.read_to_end(&mut got)
.unwrap_or_else(|e| panic!("workers={w} obj {i}: {e}"));
assert_eq!(
got, data[i],
"workers={w}: object {i} landed at the wrong offset"
);
}
}
}
/// RED-when-broken: the ordering assertion above must be able to FAIL.
/// Shuffle the recorded extents and prove the check trips — otherwise
/// "decoded to the right object" would be true for any permutation.
#[test]
fn the_ordering_check_can_actually_fail() {
let data = objs(50);
let r: Vec<&[u8]> = data.iter().map(|v| v.as_slice()).collect();
let mut buf: Vec<u8> = Vec::new();
let mut placed = compress_objects_to_writer(&r, &Config::default(), &mut buf).unwrap();
// Swap two extents of DIFFERENT objects; the decode-and-compare must now
// disagree for at least one index.
placed.swap(3, 17);
let mut mismatches = 0;
for (i, p) in placed.iter().enumerate() {
let slice = &buf[p.offset as usize..(p.offset + p.len) as usize];
let mut got = Vec::new();
if flate2::read::ZlibDecoder::new(slice)
.read_to_end(&mut got)
.is_err()
|| got != data[i]
{
mismatches += 1;
}
}
assert!(
mismatches > 0,
"a deliberately shuffled extent table still passed the per-object decode \
check — that check cannot detect mis-ordering and proves nothing"
);
}
/// The 31+1 default must actually leave a core for the writer.
#[test]
fn default_worker_count_reserves_one_core() {
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(2);
let w = compute_workers();
assert_eq!(
w,
(cores - 1).max(1),
"expected cores-1 workers, got {w} on {cores} cores"
);
assert!(
w < cores || cores == 1,
"the writer must not be sharing a core with a compressor"
);
}
}