lgz/src/bin/lgz.rs
//! lgz — parallel gzip decompressor CLI.
//!
//! ONE pass, chunk by chunk, through [`lgz::stream::decode`]:
//!
//! ```text
//! reader ──chunk──▶ split (all cores at once, first border each: flush point
//! or bit-level block start) ▶ decode (all cores) ▶ in order
//! (false starts, markers, per-member CRC) ──bytes──▶ writer
//! ```
//!
//! stdin and files are read the same way; the input is never read whole.
//! Small inputs and `LGZ_THREADS=1` take streaming zlib-rs.
//!
//! Usage: lgz < input.gz > output
//! lgz input.gz -o output
//! lgz input.gz (writes to stdout)
//!
//! Env: `LGZ_THREADS` (default: physical cores), `LGZ_CHUNK_MB` (a fixed read
//! window; default: the shared plan, sized from the file).
use std::collections::HashSet;
use std::fs::File;
use std::io::{self, BufWriter, Read, Write};
use std::thread;
use std::time::Instant;
const BUF_CAP: usize = 4 * 1024 * 1024;
// ── Physical core detection ──────────────────────────────────────────────────
fn physical_cores() -> usize {
let n = (|| -> Option<usize> {
let mut seen = HashSet::new();
for e in std::fs::read_dir("/sys/devices/system/cpu").ok()? {
let e = e.ok()?;
let s = e.file_name().to_str()?.to_string();
if !s.starts_with("cpu")
|| s[3..].is_empty()
|| !s[3..].bytes().all(|b| b.is_ascii_digit())
{
continue;
}
let pkg =
std::fs::read_to_string(e.path().join("topology/physical_package_id")).ok()?;
let core = std::fs::read_to_string(e.path().join("topology/core_id")).ok()?;
seen.insert((pkg.trim().to_string(), core.trim().to_string()));
}
Some(seen.len()).filter(|&n| n > 0)
})();
n.unwrap_or_else(|| {
thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4)
})
}
/// Worker count for the decode engine. Honors `LGZ_THREADS` (mirrors lbunzip2's
/// `LBZIP2_THREADS`) so a bench harness can pin single-core (`LGZ_THREADS=1`) vs
/// all-core runs; falls back to the physical core count. Clamped to ≥1.
fn n_workers() -> usize {
std::env::var("LGZ_THREADS")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.filter(|&n| n >= 1)
.unwrap_or_else(physical_cores)
}
/// `LGZ_CHUNK_MB` pins a fixed read window; by default the windows come
/// from the shared plan (gatling::window_plan), sized from the input.
fn fixed_chunk() -> Option<usize> {
std::env::var("LGZ_CHUNK_MB")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.filter(|&n| n >= 1)
.map(|n| n << 20)
}
fn main() {
let t0 = Instant::now();
let args: Vec<String> = std::env::args().collect();
let o_idx = args.iter().position(|a| a == "-o");
let in_path = args.get(1).filter(|a| *a != "-o").cloned();
let out_path = o_idx.and_then(|i| args.get(i + 1)).cloned();
if args.len() > 4 || (o_idx.is_some() && out_path.is_none()) {
eprintln!("Usage: lgz [input.gz] [-o output]");
std::process::exit(1);
}
let input: Box<dyn Read + Send> = match &in_path {
Some(p) => Box::new(File::open(p).unwrap_or_else(|e| {
eprintln!("lgz: cannot open '{p}': {e}");
std::process::exit(1);
})),
None => Box::new(io::stdin()),
};
let open_out = || -> Box<dyn Write + Send> {
match &out_path {
Some(p) => Box::new(File::create(p).unwrap_or_else(|e| {
eprintln!("lgz: cannot create '{p}': {e}");
std::process::exit(1);
})),
None => Box::new(io::stdout()),
}
};
let n_workers = n_workers();
// a regular-file output (-o, or stdout redirected to a file at offset 0)
// takes positional writes from a writer pool: no serial copy
let pos_file: Option<File> = match &out_path {
Some(p) => File::create(p).ok(),
None => {
use std::io::Seek;
use std::os::fd::AsFd;
io::stdout()
.as_fd()
.try_clone_to_owned()
.ok()
.map(File::from)
.filter(|f| f.metadata().is_ok_and(|m| m.is_file()))
.filter(|mut f| f.stream_position().is_ok_and(|p| p == 0))
}
};
let output = BufWriter::with_capacity(BUF_CAP, open_out());
let size = in_path
.as_ref()
.and_then(|p| std::fs::metadata(p).ok())
.filter(|m| m.is_file())
.map(|m| m.len());
// a regular file is mapped and decoded in place (zero copy); the pages
// behind the in-order stage are dropped as it moves on, so the mapping
// does not stay resident
let map = match (&in_path, size, fixed_chunk()) {
(Some(p), Some(_), None) => File::open(p)
.ok()
.and_then(|f| unsafe { memmap2::Mmap::map(&f) }.ok()),
_ => None,
};
let res = match (&map, fixed_chunk()) {
(Some(m), _) => {
let release = |n: usize| {
// SAFETY: a read-only file mapping: dropped pages refault from
// the page cache if they are ever touched again
let _ =
unsafe { m.unchecked_advise_range(memmap2::UncheckedAdvice::DontNeed, 0, n) };
};
match &pos_file {
Some(f) => {
drop(output);
lgz::stream::decode_slice_to_file(m, f, n_workers, &release)
}
None => lgz::stream::decode_slice_release(m, output, n_workers, &release),
}
}
(None, Some(c)) => lgz::stream::decode(input, output, n_workers, c),
(None, None) => lgz::stream::decode_sized(input, output, n_workers, size),
};
match res {
Ok(st) => {
let s = t0.elapsed().as_secs_f64();
eprintln!(
"lgz: {} bytes in {s:.3}s ({:.0} MB/s) workers={n_workers} segments={} members={} serial={}",
st.written,
st.written as f64 / s / 1e6,
st.segments,
st.members,
st.serial,
);
lgz::functional_status(
"lgz",
"decode",
true,
&format!("gzip decode completed in {s:.3}s"),
);
}
Err(e) => {
// A file in and a file out can be redone from the start with the
// authoritative streaming decoder; a pipe cannot be taken back.
if let (Some(ip), Some(_)) = (&in_path, &out_path) {
eprintln!("{e}; redoing the whole file with streaming zlib-rs");
let f = File::open(ip).expect("reopen input");
let mut dec =
flate2::read::MultiGzDecoder::new(io::BufReader::with_capacity(BUF_CAP, f));
let mut w = BufWriter::with_capacity(BUF_CAP, open_out());
if let Err(e2) = io::copy(&mut dec, &mut w).and_then(|_| w.flush()) {
eprintln!("lgz: decompression failed: {e2}");
lgz::functional_status("lgz", "decode", false, &e2.to_string());
std::process::exit(1);
}
} else {
eprintln!("{e}");
lgz::functional_status("lgz", "decode", false, &e.to_string());
std::process::exit(1);
}
}
}
}