diff --git a/Cargo.toml b/Cargo.toml index 0ab7f8c..7d604c3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.5" edition = "2024" rust-version = "1.91" license = "Apache-2.0" -description = "Fast parallel and indexed bzip2 decompression for Rust and Python" +description = "Compression-format research workbench with fast bzip2 and gzip decompression" repository = "https://github.com/AnswerDotAI/fastbz2" homepage = "https://github.com/AnswerDotAI/fastbz2" documentation = "https://github.com/AnswerDotAI/fastbz2" @@ -25,6 +25,7 @@ codegen-units = 1 [dependencies] blake3 = "1.8.7" clap = { version = "4.6.6", features = ["derive"] } +crc32fast = "1.5.1" memmap2 = "0.9.11" pyo3 = { version = ">=0.29.2", optional = true } rayon = "1.12.0" @@ -33,6 +34,8 @@ tempfile = "3.27.0" [dev-dependencies] crabz2 = { version = "0.4.0", features = ["parallel"] } +flate2 = { version = "1.1", default-features = false, features = ["rust_backend"] } +libc = "0.2.175" libbz2-rs-sys = { version = "0.2.5", default-features = false, features = ["std"] } [features] diff --git a/DEV.md b/DEV.md index 47f769b..3a83adf 100644 --- a/DEV.md +++ b/DEV.md @@ -11,6 +11,8 @@ src/crc.rs bzip2 block and combined-stream CRC primitives src/decode.rs serial/parallel decode scheduling and index construction src/decoder.rs bzip2 block machinery and 12-bit Huffman fast tables src/format.rs cheap structural scan for header and marker candidates +src/gzip.rs gzip framing, LSB-first DEFLATE, CRC32, and block reports +src/pipeline.rs shared ordered, byte-budgeted, staged worker scheduler src/index.rs stable persistent index format src/indexed.rs seekable decoded view and block cache src/lib.rs public Rust API and private PyO3 binding @@ -22,15 +24,17 @@ tests/ Rust CLI/corpus and Python API integration tests tools/stage_binaries.py copy the release executable into Maturin wheel data ``` -The current scanner deliberately does not treat 48-bit marker matches or later `BZh` headers as validated structure. Full decoding must establish the exact block chain and validate every block CRC plus the combined stream CRC before marker candidates can become trusted index entries. Python integration tests use standard-library `bz2`/libbz2 as an independent fixture generator. +The current bzip2 scanner deliberately does not treat 48-bit marker matches or later `BZh` headers as validated structure. Full decoding must establish the exact block chain and validate every block CRC plus the combined stream CRC before marker candidates can become trusted index entries. Python integration tests use standard-library `bz2`/libbz2 as an independent fixture generator. + +The gzip decoder is an in-repo RFC 1952/RFC 1951 implementation rather than a wrapper around a production codec. It parses optional headers and concatenated members, decodes stored/fixed/dynamic blocks, maintains the 32 KiB LZ77 history, and validates FHCRC, CRC32, and ISIZE. For large inputs, independently discovered dynamic-block boundaries seed unknown history with compact markers. A marker-free history switches the same decoder to byte output; otherwise the coordinator resolves only the suffix needed by the successor and queues full resolution plus CRC on the shared staged scheduler. Reports retain member boundaries, DEFLATE block ranges, and accepted/fallback chunk counts. `crc32fast` is the sole production helper; `flate2` is dev-only. The decoder remains independent of files, threads, Python, and the CLI. Parallel scanning/decoding and indexed seeking are layered over it. Native workers never call Python. Large offsets use explicit 64-bit bit/byte types, and speculative block-marker hits are accepted only when they form an exact stream chain with valid block and combined stream CRCs. -The core decode APIs can report completed compressed and decoded byte counts without knowing anything about terminals. The CLI layers delayed, rate-limited TTY progress rendering over those callbacks; redirected stderr and `--quiet` produce no progress output. Decoded files use same-directory temporary files and atomic persistence, then inherit the compressed input's modification time and permissions. `--rm` removes an input only after decode, persistence, and metadata copying all succeed. Output-size limits are enforced by a writer wrapper, so the decoder has one code path for files, stdout, validation, and indexing. +Both core decode APIs report completed compressed and decoded byte counts without knowing anything about terminals. The CLI selects bzip2 or gzip by a recognised extension and falls back to magic for stdin or unknown names. It layers delayed, rate-limited TTY progress rendering over the shared callbacks; redirected stderr and `--quiet` produce no progress output. Decoded files use same-directory temporary files and atomic persistence, then inherit the compressed input's modification time and permissions. `--rm` removes an input only after decode, persistence, and metadata copying all succeed. Output-size limits are enforced by a writer wrapper, so each decoder has one code path for files, stdout, validation, and listing. -Parallel decoding uses a rolling candidate queue rather than stopping at stream boundaries or waiting for fixed batches. Workers reserve the maximum possible decoded block size before starting; once a block finishes, that conservative reservation shrinks to its actual output size and is released when ordered validation consumes or rejects it. Thus the `memory_limit` bounds speculative decoded output while short multistream inputs can keep the worker pool busy. The 1 GiB default admits one worst-case block per worker on the primary 18-core machine. +The shared `pipeline.rs` scheduler provides ordered results, byte-budgeted admission, cancellation, and a staged priority queue. Bzip2 uses the rolling candidate path: workers reserve the maximum possible decoded block size, then shrink that reservation to actual retained output until ordered validation consumes or rejects it. Gzip uses the staged path: native workers alternate speculative DEFLATE decoding with higher-priority marker resolution, while the coordinator advances only the 32 KiB dependency windows and emits resolved chunks in order. Decode results and outstanding resolution results have separate bounded horizons, preventing either dependency stalls or unbounded memory. -The production decoder is safe scalar Rust designed for LLVM auto-vectorisation. Huffman decoding uses a 4096-entry direct table for codes up to 12 bits and canonical fallback for longer codes. Add narrowly scoped unsafe or architecture-specific SIMD only after profiling; `libbz2-rs-sys` remains the dev-only differential oracle. +The bzip2 decoder is safe scalar Rust designed for LLVM auto-vectorisation. Huffman decoding uses a 4096-entry direct table for codes up to 12 bits and canonical fallback for longer codes. Gzip uses full canonical lookup tables packed into `u16`, a branch-free 64 KiB marker-resolution lookup for large chunks, and `crc32fast::Hasher::combine` so CRC scanning runs with the resolution workers rather than serially in the coordinator. The only unsafe codec operation marks a just-initialized `Vec` result as initialized after writing every spare-capacity byte. Add architecture-specific SIMD or further cross-codec abstraction only after profiling; `libbz2-rs-sys` and `flate2` remain dev-only differential oracles. ## Commands @@ -49,30 +53,45 @@ Run `cargo fmt --check` after Rust edits and `chkstyle` after Python edits once ## Correctness and performance acceptance -The normal release test path decodes selected valid and corrupt cases from the maintained upstream `bzip2-testfiles` collection. Generated byte distributions add differential coverage. Valid outputs are compared byte-for-byte with `libbz2-rs-sys`, which is a dev-only oracle and never part of production decoding. +The normal release test path decodes selected valid and corrupt cases from the maintained upstream `bzip2-testfiles` collection. Generated byte distributions add differential coverage. Valid bzip2 outputs are compared byte-for-byte with `libbz2-rs-sys`. Gzip tests cover stored, fixed-Huffman, and dynamic-Huffman blocks; optional headers and FHCRC; concatenated members; truncation; and trailer corruption across varied inputs and compression levels generated by `flate2`. Both oracles are dev-only and never part of production decoding. -The same test binary contains a warmed end-to-end performance regression gate capped at 1.3 times the oracle, allowing for noise on shared runners. Representative local acceptance remains 1.2 times the oracle. Keep the whole release test suite below five seconds on the primary development laptop; individual timed workloads should normally be about 0.1 seconds or less. Use the Simple English Wikipedia dump for heavier local throughput, scaling, memory, and time-to-first-output checks. `librapidarchive` was a one-time design comparison, not a retained baseline. +The normal release path contains warmed end-to-end performance regression gates capped at 1.3 times each oracle, allowing for noise on shared runners. The gzip gates independently exercise a highly compressible LZ77-heavy shape and an incompressible literal-heavy shape against `flate2`; the bzip2 gate uses `libbz2-rs-sys`. Representative local acceptance remains 1.2 times the corresponding oracle. The ignored full-wiki gzip test applies that threshold to rapidgzip-rust. Keep the whole release test suite below five seconds on the primary development laptop; individual timed workloads should normally be about 0.1 seconds or less. Legacy randomized blocks produced by bzip2 versions before 0.9.5 are intentionally unsupported. Supporting that obsolete format would add complexity to the production decoder for data that is not realistically encountered today. ### Local Wikipedia benchmarks -`tests/wiki_perf.rs` contains a release-only local benchmark that is skipped by default. Its git-ignored fixture is a single bzip2 stream containing the first 84,423,012 decoded bytes (about 5%) of SimpleWiki, stored at `meta/simplewiki-first-5pct.xml.bz2`. Run it with: +`tests/wiki_perf.rs` contains release-mode local benchmarks that are skipped by default. The git-ignored SimpleWiki fixtures are a bzip2-compressed 5% prefix, the full bzip2 dump, the same 5% prefix recompressed as gzip, and the full XML recompressed with system `gzip -6`. + +The quick bzip2 iteration test reads `meta/simplewiki-first-5pct.xml.bz2` before timing, then verifies decoded length, all CRCs, and BLAKE3: ```bash -cargo test --release --test wiki_perf simplewiki_first_five_percent -- --ignored --nocapture +cargo test --release --test wiki_perf simplewiki_first_five_percent -- --ignored --exact --nocapture ``` -It uses all available CPUs by default. Set `FASTBZ2_THREADS` to compare an explicit thread count. The timed section includes allocation and decompression but excludes reading the compressed file and calculating its BLAKE3; decoded length, block/stream CRCs, and BLAKE3 are all checked. +The full bzip2 confirmation streams to a counting sink and validates every block and stream CRC: + +```bash +cargo test --release --test wiki_perf simplewiki_full -- --ignored --exact --nocapture +``` -For an occasional full-dump confirmation, create `meta/simplewiki-full.xml.bz2` as a symlink to the local dump and run: +The gzip acceptance test warms both executables with `meta/simplewiki-first-5pct.xml.gz`, performs exactly one measured full-dump validation with each, and fails above 1.2× the sibling rapidgzip-rust checkout: ```bash -cargo test --release --test wiki_perf simplewiki_full -- --ignored --nocapture +cargo test --release --test wiki_perf gzip_reference_ratio -- --ignored --exact --nocapture ``` -The full test streams decoded bytes into a counting sink and validates every block and stream CRC. Stream headers found during the structural scan remain speculative: only the exact byte-aligned header following a validated end-of-stream marker starts another stream. +Set `FASTBZ2_THREADS` to use an explicit worker count. `RAPIDGZIP_BIN` can point at another reference executable. The warm-up is deliberately the small fixture, not an unreported repeat of the measured full workload. + +Time/CPU/RSS and ungated macOS physical-footprint diagnostics are separate because process inspection can perturb sub-second parallel timings: + +```bash +cargo test --release --test wiki_perf gzip_cli_process_metrics -- --ignored --exact --nocapture +cargo test --release --test wiki_perf rapidgzip_rust_process_metrics -- --ignored --exact --nocapture +cargo test --release --test wiki_perf system_gzip_process_metrics -- --ignored --exact --nocapture +``` +The metrics helper uses `wait4` and, on macOS, `proc_pid_rusage` on its own child; it needs no task-inspection permission. Treat its wall time as diagnostic and use `gzip_reference_ratio` for the speed acceptance ratio. The 1,000-stream enwiki comparison has a separate ignored test for each implementation and mode so a changed decoder can be measured without rerunning unchanged baselines. Each test reads the compressed fixture before starting its single timed decode, with no warmups or repeats: ```bash @@ -91,7 +110,7 @@ The decoded lengths and the 5% BLAKE3 in `tests/wiki_perf.rs` are acceptance val Set `wiki` to the checkout containing the Wikimedia dumps: ```bash -wiki=/path/to/wiki2md +wiki=/path/to/parse-wiki ``` Recreate the SimpleWiki fixtures from its validated compressed and decoded files: @@ -99,6 +118,8 @@ Recreate the SimpleWiki fixtures from its validated compressed and decoded files ```bash head -c 84423012 "$wiki/data/simplewiki-latest-pages-articles.xml" | bzip2 -9c > meta/simplewiki-first-5pct.xml.bz2 ln -s "$wiki/data/simplewiki-latest-pages-articles.xml.bz2" meta/simplewiki-full.xml.bz2 +head -c 84423012 "$wiki/data/simplewiki-latest-pages-articles.xml" | gzip -6c > meta/simplewiki-first-5pct.xml.gz +gzip -6c "$wiki/data/simplewiki-latest-pages-articles.xml" > meta/simplewiki-full.xml.gz ``` Run the 5% test to obtain and verify its decoded length and BLAKE3. Obtain the full decoded length with `stat`; update `FIVE_PERCENT_LEN`, `FIVE_PERCENT_BLAKE3`, or `FULL_LEN` only when intentionally changing a fixture. diff --git a/README.md b/README.md index f6c92e3..7966db7 100644 --- a/README.md +++ b/README.md @@ -1,10 +1,10 @@ # fastbz2 -Fast parallel and indexed bzip2 decompression for Rust and Python. +An active compression-format research workbench with fast bzip2 and gzip decompression. -`fastbz2` provides a native CLI, a Rust library, and a Python module. It handles ordinary and concatenated bzip2 streams, validates every block and stream CRC, and keeps speculative parallel output within a configurable memory bound. Persistent indexes support efficient random access from Python without first expanding the whole file. +`fastbz2` provides a native CLI, a Rust library, and a Python module. The CLI auto-selects an in-repo bzip2 or gzip decoder from the filename extension, falling back to stream magic when needed. Both decoders handle concatenated streams and fully validate their checksums. The bzip2 implementation also provides parallel decoding and persistent random-access indexes. -This project decompresses bzip2; it does not compress it. +Compression and additional formats are planned, but the current implementation decompresses bzip2 and gzip. ## Install @@ -25,10 +25,11 @@ cargo add fastbz2 --git https://github.com/AnswerDotAI/fastbz2 ## CLI -Decoding is the default operation. A `.bz2` suffix is removed for the output name; other input names gain `.out`. +Decoding is the default operation. `.bz2`, `.bzip2`, `.gz`, and `.gzip` select their corresponding decoder and are removed from the output name. `.tbz`, `.tbz2`, and `.tgz` produce a `.tar` filename; this currently decompresses the tar stream rather than extracting its entries. For stdin and unrecognised extensions, bzip2 or gzip magic selects the decoder. Other input names gain `.out`. ```bash fastbz2 dump.xml.bz2 # write dump.xml +fastbz2 events.json.gz # write events.json fastbz2 dump.xml.bz2 -o result.xml # choose the output path fastbz2 dump.xml.bz2 -o - # write plaintext to stdout fastbz2 - # read compressed data from stdin @@ -37,16 +38,16 @@ fastbz2 - # read compressed data from stdin Multiple inputs are decoded in order, with parallelism applied inside each file. `-C/--output-dir` collects their outputs in one directory: ```bash -fastbz2 *.bz2 -C decoded -fastbz2 *.bz2 -C decoded --skip-existing +fastbz2 data/*.bz2 logs/*.gz -C decoded +fastbz2 data/*.bz2 logs/*.gz -C decoded --skip-existing ``` The alternative modes are flags rather than subcommands: ```bash fastbz2 --test dump.xml.bz2 # fully decode and validate, writing nothing -fastbz2 --index dump.xml.bz2 # write dump.xml.bz2.fbz2i -fastbz2 --list dump.xml.bz2 # print the validated stream/block layout +fastbz2 --index dump.xml.bz2 # write dump.xml.bz2.fbz2i (bzip2 only) +fastbz2 --list events.json.gz # print the validated member/block layout fastbz2 --list --json dump.xml.bz2 # emit the complete layout as JSON ``` @@ -62,10 +63,12 @@ fastbz2 --list --json dump.xml.bz2 # emit the complete layout as JSON Long interactive operations report completion, decoded throughput, compression ratio, and ETA on stderr. Progress is disabled automatically when stderr is redirected; `-q/--quiet` also suppresses progress and skip notices. -`-P/--threads 0`, the default, uses all available CPUs. `--memory-limit` bounds speculative decoded output and defaults to `1G`. +`-P/--threads 0`, the default, uses the machine's available parallelism; an explicit positive value is honoured by either codec. `--memory-limit` bounds speculative output in the shared scheduler and defaults to `1G`. Gzip uses parallel dynamic-block discovery only when the input and memory budget can amortize it, otherwise selecting its serial path automatically. ## Python +The Python API currently exposes the bzip2 backend. Unified Python dispatch will follow the CLI workbench rather than being designed ahead of it. + ### One-shot decompression and validation ```python @@ -125,21 +128,40 @@ fn main() -> fastbz2::Result<()> { `decompress` returns a `Vec`. `decode_to_writer` returns a validated `Index` while streaming output, `build_index` validates into a sink, and their `*_with_progress` variants report completed compressed and decoded byte counts. `IndexedReader` implements `Read` and `Seek`; it can build an index itself or load a persisted one with `open_with_index`. +The in-repo gzip decoder is available separately so callers can choose explicitly: + +```rust +let plain = fastbz2::gzip::decompress(&compressed_gzip)?; +``` + +`gzip::decompress_to_writer` and `gzip::decompress_to_writer_with_options` return a validated report containing gzip member metadata, each DEFLATE block's kind and ranges, and counts of accepted speculative and serial-fallback chunks. They support stored, fixed-Huffman, and dynamic-Huffman blocks, optional gzip headers, and concatenated members. + ## Performance -These are single local release-mode runs on the primary Apple Silicon development machine, using 18 workers for parallel rows. They are observations rather than statistically aggregated benchmarks. Modes are stated per row because a streaming sink, an in-memory `Vec`, and a CLI pipeline have different allocation and I/O costs; compare rows using the same method most directly. +These are single local release-mode runs on the primary 18-core Apple Silicon development machine. Bzip2 parallel rows use 18 workers. The gzip comparison warms each executable with the 5% fixture, then measures exactly one full validation run; peak physical footprint comes from a separate sampled run because process inspection can perturb such a short workload. These are observations rather than statistical aggregates. Modes are stated per row because a streaming sink, an in-memory `Vec`, and a CLI pipeline have different allocation and I/O costs; compare rows using the same method most directly. Full Simple English Wikipedia (`338 MB` compressed, `1,688,460,257` bytes decoded): | Decoder | Mode | Seconds | |---|---|---:| -| fastbz2 | parallel, 18 threads, streaming sink | 2.244 | +| fastbz2 | parallel, 18 threads, streaming sink | 2.515 | | crabz2 0.4.0 | parallel | 4.460 | | bzip2 | serial CLI | 20.310 | | pbzip2 1.1.13 | CLI | 20.240 | | libbz2-rs 0.2.5 | serial, in process | 20.700 | | fastbz2 | serial, in process | 21.279 | + +Full SimpleWiki recompressed with system `gzip -6` (`438,904,466` bytes compressed, `1,688,460,257` bytes decoded): + +| Decoder | Mode | Seconds | Peak physical footprint | +|---|---|---:|---:| +| rapidgzip-rust, local checkout | auto parallel, validation sink | 0.363 | 585 MiB | +| fastbz2 | auto parallel, validation sink | 0.357 | 552 MiB | +| Apple gzip | serial, stdout discarded | 1.371 | 1.2 MiB | + +The memory values use macOS physical footprint rather than `ru_maxrss`. The fastbz2 CLI memory-maps its 419 MiB input, so clean reclaimable file pages make RSS look roughly 419 MiB larger; `pread`-based tools leave the same cached pages outside process RSS. Physical footprint makes the comparison meaningful. + The first 1,000 streams of English Wikipedia (`654,362,682` bytes compressed, `2,715,335,085` bytes decoded, 99,853 pages) exercise scheduling across many short concatenated streams: | Decoder | Mode | Seconds | @@ -157,11 +179,13 @@ Homebrew `pbzip2` 1.1.13 could not safely decompress the complete 26,668,484,995 ## Implementation and compatibility -The decoder is safe, portable Rust with a tuned 4096-entry Huffman lookup table for codes up to 12 bits and canonical fallback for longer codes. A structural scan finds possible non-byte-aligned block markers; these remain speculative until ordered decoding establishes the exact stream chain and validates all block and combined-stream CRCs. A rolling scheduler keeps workers busy across concatenated streams while bounding decoded results awaiting validation. +The production codec logic is portable Rust. The bzip2 decoder uses a tuned 4096-entry Huffman lookup table for codes up to 12 bits and canonical fallback for longer codes. A structural scan finds possible non-byte-aligned block markers; these remain speculative until ordered decoding establishes the exact stream chain and validates all block and combined-stream CRCs. A rolling scheduler keeps workers busy across concatenated streams while bounding decoded results awaiting validation. + +The gzip backend implements RFC 1952 framing and DEFLATE directly in this repository. For sufficiently large dynamic-Huffman inputs it discovers independently decodable boundaries, decodes speculative chunks through the shared byte-budgeted scheduler, and represents unknown predecessor bytes as compact markers. The ordered coordinator resolves only the suffix needed to derive the next 32 KiB history window; full marker resolution and per-chunk CRC run as priority work on the same staged worker queue, and CRCs are combined in order. Once a chunk has a marker-free window, the same decoder switches its remaining output from `u16` markers to ordinary bytes. Small, stored-heavy, fixed-heavy, one-thread, and low-memory inputs use the serial path; concatenated members may independently choose either path. FHCRC, CRC32, and ISIZE are always validated. `crc32fast` is the only production codec helper; `flate2` is dev-only. Legacy randomized blocks generated by bzip2 releases before 0.9.5 are intentionally unsupported. Normal `BZh1` through `BZh9` streams and concatenated streams are supported. -The architecture was inspired by Maximilian Knespel's [`librapidarchive`](https://github.com/mxmlnkn/librapidarchive) and [`indexed_bzip2`](https://github.com/mxmlnkn/indexed_bzip2): in particular, scanning for non-byte-aligned bzip2 block markers, independently decoding blocks, ordered prefetch, and indexed seeking. That project's specialised decoder is itself derived from Rob Landley's 0BSD [`bzcat` implementation in Toybox](https://github.com/landley/toybox). +The architecture was inspired by Maximilian Knespel's [`librapidarchive`](https://github.com/mxmlnkn/librapidarchive), [`indexed_bzip2`](https://github.com/mxmlnkn/indexed_bzip2), and [`rapidgzip-rust`](https://github.com/mxmlnkn/rapidgzip-rust). The bzip2 design draws on non-byte-aligned marker scanning, independent block decoding, ordered prefetch, and indexed seeking; the gzip design draws on structurally validated dynamic-block discovery, predecessor markers, marker-free handoff to byte output, and suffix-only window resolution. The specialised bzip2 decoder is itself derived from Rob Landley's 0BSD [`bzcat` implementation in Toybox](https://github.com/landley/toybox). ## Development diff --git a/pyproject.toml b/pyproject.toml index 49f464b..af71fe0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,7 @@ backend-path = ["."] [project] name = "fastbz2" dynamic = ["version"] -description = "Fast parallel and indexed bzip2 decompression for Rust and Python" +description = "Compression-format research workbench with fast bzip2 and gzip decompression" license = {text = "Apache-2.0"} requires-python = ">=3.10" readme = "README.md" diff --git a/src/bin/fastbz2.rs b/src/bin/fastbz2.rs index 1e39f72..a74c12d 100644 --- a/src/bin/fastbz2.rs +++ b/src/bin/fastbz2.rs @@ -8,7 +8,7 @@ use std::{ use clap::{ArgGroup, Parser}; use fastbz2::{ - DecodeOptions, DecodeProgress, Error, Index, Source, build_index_with_progress, decode_to_writer_with_progress, decompress_to_writer_with_progress, + DecodeOptions, DecodeProgress, Error, Index, Source, build_index_with_progress, decode_to_writer_with_progress, decompress_to_writer_with_progress, gzip, }; use serde_json::{Value, json}; use tempfile::NamedTempFile; @@ -16,7 +16,7 @@ use tempfile::NamedTempFile; #[derive(Parser)] #[command( version, - about = "Fast parallel and indexed bzip2 decompression", + about = "Fast compression-format research workbench", group(ArgGroup::new("mode").args(["test", "index", "list"])) )] struct Cli { @@ -26,7 +26,7 @@ struct Cli { /// Fully decode and validate without writing plaintext. #[arg(long)] test: bool, - /// Build validated, source-bound block indexes. + /// Build validated, source-bound bzip2 block indexes. #[arg(long)] index: bool, /// Validate and show stream/block layouts. @@ -38,10 +38,10 @@ struct Cli { /// Put decoded files in DIRECTORY. #[arg(short = 'C', long = "output-dir", conflicts_with_all = ["test", "index", "list", "output"])] output_dir: Option, - /// Worker threads; 0 uses all available CPUs. + /// Bzip2 worker threads; 0 uses all available CPUs. #[arg(short = 'P', long, default_value_t = 0)] threads: usize, - /// Maximum speculative decoded output. + /// Maximum speculative bzip2 output. #[arg(long, default_value = "1G", value_parser = parse_size)] memory_limit: usize, /// Refuse to decode more than SIZE bytes per input. @@ -64,6 +64,12 @@ struct Cli { json: bool, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Format { + Bzip2, + Gzip, +} + fn main() -> ExitCode { match run(Cli::parse()) { Ok(()) => ExitCode::SUCCESS, @@ -143,6 +149,9 @@ fn run_index(cli: &Cli, options: DecodeOptions) -> fastbz2::Result<()> { continue; } let source = Source::open(input_path)?; + if select_format(input, source.as_slice())? != Format::Bzip2 { + return Err(invalid("--index is currently supported only for bzip2 inputs")); + } let index = build_index_data(source.as_slice(), input, options, cli.max_output, cli.quiet)?; let encoded = index.to_bytes(); if output == Path::new("-") { @@ -158,11 +167,23 @@ fn run_list(cli: &Cli, options: DecodeOptions) -> fastbz2::Result<()> { let mut values = Vec::new(); for input in &cli.inputs { let source = Source::open(input)?; - let index = build_index_data(source.as_slice(), input, options, cli.max_output, cli.quiet)?; - if cli.json { - values.push(index_json(input, &index)); - } else { - print_index((cli.inputs.len() > 1).then_some(input), &index); + match select_format(input, source.as_slice())? { + Format::Bzip2 => { + let index = build_index_data(source.as_slice(), input, options, cli.max_output, cli.quiet)?; + if cli.json { + values.push(index_json(input, &index)); + } else { + print_index((cli.inputs.len() > 1).then_some(input), &index); + } + } + Format::Gzip => { + let report = build_gzip_report_data(source.as_slice(), input, options, cli.max_output, cli.quiet)?; + if cli.json { + values.push(gzip_json(input, &report)); + } else { + print_gzip_report((cli.inputs.len() > 1).then_some(input), &report); + } + } } } if cli.json { @@ -187,21 +208,26 @@ fn decode_input(input: &str, output: &mut impl Write, options: DecodeOptions, ma fn decode_data(data: &[u8], label: &str, output: &mut impl Write, options: DecodeOptions, max_output: Option, quiet: bool) -> fastbz2::Result<()> { let mut output = LimitedWriter::new(output, max_output); let mut display = ProgressDisplay::new(label, data.len() as u64, quiet); - let result = decompress_to_writer_with_progress(data, &mut output, options, |progress| display.update(progress)); - display.finish(); - result + match select_format(label, data)? { + Format::Bzip2 => decompress_to_writer_with_progress(data, &mut output, options, |progress| display.update(progress)), + Format::Gzip => gzip::decompress_to_writer_with_options_and_progress(data, &mut output, options, |progress| display.update(progress)).map(|_| ()), + } +} + +fn build_gzip_report_data(data: &[u8], label: &str, options: DecodeOptions, max_output: Option, quiet: bool) -> fastbz2::Result { + let mut sink = LimitedWriter::new(io::sink(), max_output); + let mut display = ProgressDisplay::new(label, data.len() as u64, quiet); + gzip::decompress_to_writer_with_options_and_progress(data, &mut sink, options, |progress| display.update(progress)) } fn build_index_data(data: &[u8], label: &str, options: DecodeOptions, max_output: Option, quiet: bool) -> fastbz2::Result { let mut display = ProgressDisplay::new(label, data.len() as u64, quiet); - let result = if let Some(limit) = max_output { + if let Some(limit) = max_output { let mut sink = LimitedWriter::new(io::sink(), Some(limit)); decode_to_writer_with_progress(data, &mut sink, options, |progress| display.update(progress)) } else { build_index_with_progress(data, options, |progress| display.update(progress)) - }; - display.finish(); - result + } } struct LimitedWriter { @@ -332,10 +358,31 @@ fn output_in(input: &Path, directory: Option<&Path>) -> PathBuf { directory.map_or_else(|| default_output(input), |directory| directory.join(output)) } +fn format_extension(input: &Path) -> Option<(Format, &'static str)> { + let extension = input.extension()?.to_str()?.to_ascii_lowercase(); + match extension.as_str() { + "bz2" | "bzip2" => Some((Format::Bzip2, "")), + "tbz" | "tbz2" => Some((Format::Bzip2, "tar")), + "gz" | "gzip" => Some((Format::Gzip, "")), + "tgz" => Some((Format::Gzip, "tar")), + _ => None, + } +} + fn default_output(input: &Path) -> PathBuf { - match input.extension().and_then(|extension| extension.to_str()) { - Some("bz2") => input.with_extension(""), - _ => PathBuf::from(format!("{}.out", input.display())), + format_extension(input).map_or_else(|| PathBuf::from(format!("{}.out", input.display())), |(_, extension)| input.with_extension(extension)) +} + +fn select_format(input: &str, data: &[u8]) -> fastbz2::Result { + if let Some((format, _)) = format_extension(Path::new(input)) { + return Ok(format); + } + if data.starts_with(b"BZh") { + Ok(Format::Bzip2) + } else if data.starts_with(&[0x1f, 0x8b]) { + Ok(Format::Gzip) + } else { + Err(invalid(format!("cannot determine compression format for {input}; expected a bzip2/gzip extension or magic"))) } } @@ -355,9 +402,37 @@ fn print_index(input: Option<&String>, index: &Index) { } } +fn print_gzip_report(input: Option<&String>, report: &gzip::Report) { + if let Some(input) = input { + println!("input\t{input}"); + } + println!("format\tgzip"); + println!("compressed_bytes\t{}", report.source_len); + println!("decoded_bytes\t{}", report.decoded_len); + println!("members\t{}", report.members.len()); + println!("blocks\t{}", report.blocks.len()); + println!("speculative_chunks\t{}", report.speculative_chunks); + println!("fallback_chunks\t{}", report.fallback_chunks); + for (number, member) in report.members.iter().enumerate() { + let name = member.name.as_deref().map(String::from_utf8_lossy).unwrap_or_default(); + println!( + "member\t{number}\tblocks={}\tdecoded={}\tmtime={}\tos={}\tname={name}", + member_block_count(report, number), + member.decoded_len, + member.mtime, + member.operating_system + ); + } +} + +fn member_block_count(report: &gzip::Report, member: usize) -> usize { + report.blocks.iter().filter(|block| block.member as usize == member).count() +} + fn index_json(input: &str, index: &Index) -> Value { json!({ "input": input, + "format": "bzip2", "source_bytes": index.source_len, "source_hash": hex(&index.source_hash), "decoded_bytes": index.decoded_len, @@ -384,6 +459,41 @@ fn index_json(input: &str, index: &Index) -> Value { }) } +fn gzip_json(input: &str, report: &gzip::Report) -> Value { + json!({ + "input": input, + "format": "gzip", + "source_bytes": report.source_len, + "decoded_bytes": report.decoded_len, + "speculative_chunks": report.speculative_chunks, + "fallback_chunks": report.fallback_chunks, + "members": report.members.iter().enumerate().map(|(number, member)| json!({ + "number": number, + "compressed_start": member.compressed_start, + "deflate_start": member.deflate_start, + "compressed_end": member.compressed_end, + "decoded_start": member.decoded_start, + "decoded_bytes": member.decoded_len, + "expected_crc": member.expected_crc, + "mtime": member.mtime, + "extra_flags": member.extra_flags, + "operating_system": member.operating_system, + "name": member.name.as_deref().map(|name| String::from_utf8_lossy(name)), + "comment": member.comment.as_deref().map(|comment| String::from_utf8_lossy(comment)), + })).collect::>(), + "blocks": report.blocks.iter().enumerate().map(|(number, block)| json!({ + "number": number, + "member": block.member, + "kind": block.kind.as_str(), + "final": block.final_block, + "compressed_start_bit": block.compressed_start_bit, + "compressed_end_bit": block.compressed_end_bit, + "decoded_start": block.decoded_start, + "decoded_bytes": block.decoded_len, + })).collect::>(), + }) +} + fn hex(bytes: &[u8]) -> String { bytes.iter().map(|byte| format!("{byte:02x}")).collect() } @@ -433,7 +543,7 @@ fn exit_status(error: &Error) -> u8 { Error::Io(source) if source.kind() == io::ErrorKind::InvalidData => 3, Error::Io(_) => 1, Error::InvalidConfiguration(_) => 2, - Error::InvalidStreamHeader | Error::Decode { .. } | Error::InvalidIndex(_) => 3, + Error::InvalidStreamHeader | Error::InvalidGzip(_) | Error::Decode { .. } | Error::InvalidIndex(_) => 3, _ => 4, } } diff --git a/src/decode.rs b/src/decode.rs index f6df7d4..8cfee23 100644 --- a/src/decode.rs +++ b/src/decode.rs @@ -1,13 +1,9 @@ -use std::{ - collections::HashMap, - io::Write, - sync::{Arc, Condvar, Mutex, mpsc}, - thread, -}; +use std::{io::Write, sync::Arc, thread}; use rayon::{ThreadPool, ThreadPoolBuilder}; use crate::format::scan_with_pool; +use crate::pipeline::{Job, OrderedResults, PipelineLimits, run_ordered}; use crate::{BlockCandidate, BlockIndex, DecodeError, EndCandidate, Error, Index, MAX_DECODED_BLOCK, Result, StreamIndex, combine_stream_crc, decoder}; pub const DEFAULT_MEMORY_LIMIT: usize = 1024 * 1024 * 1024; @@ -37,7 +33,7 @@ impl DecodeOptions { if self.threads != 0 { self.threads } else { thread::available_parallelism().map(usize::from).unwrap_or(1) } } - fn validate(self) -> Result { + pub(crate) fn validate(self) -> Result { if self.memory_limit < MAX_DECODED_BLOCK { return Err(Error::InvalidConfiguration(format!("memory limit must be at least {MAX_DECODED_BLOCK} bytes"))); } @@ -133,25 +129,25 @@ fn decode_to_writer_impl(data: &[u8], output: &mut impl Write, options: DecodeOp .iter() .enumerate() .filter_map(|(marker_index, marker)| match marker { - Marker::Block(block) => Some(CandidateJob { marker_index, start_bit: block.bit_offset, expected_crc: block.expected_crc }), + Marker::Block(block) => Some(Job { + key: marker_index, + reservation: MAX_DECODED_BLOCK, + payload: CandidateJob { start_bit: block.bit_offset, expected_crc: block.expected_crc }, + }), Marker::End(_) => None, }) .collect(); - let work = WorkQueue::new(options.memory_limit); - let (sender, receiver) = mpsc::channel(); - - thread::scope(|scope| { - let worker = scope.spawn(|| { - pool.broadcast(|_| worker_loop(data, &jobs, &work, &sender)); - }); - let result = { - let mut candidates = ParallelCandidates { receiver, ready: HashMap::new(), work: &work }; + run_ordered( + &pool, + &jobs, + PipelineLimits { memory: options.memory_limit, active: usize::MAX }, + |job| decoder::decode_candidate(data, job.start_bit, job.expected_crc), + candidate_len, + |results| { + let mut candidates = ParallelCandidates { results }; assemble(data, output, &markers, &mut candidates, progress) - }; - work.cancel(); - worker.join().map_err(|_| Error::InvalidConfiguration("parallel decoder worker panicked".into()))?; - result - }) + }, + ) } fn assemble( @@ -253,111 +249,29 @@ impl Candidates for SerialCandidates<'_> { #[derive(Clone, Copy)] struct CandidateJob { - marker_index: usize, start_bit: u64, expected_crc: u32, } -struct WorkState { - next: usize, - reserved: usize, - cancelled: bool, -} - -struct WorkQueue { - limit: usize, - state: Mutex, - wake: Condvar, -} - -impl WorkQueue { - fn new(limit: usize) -> Self { - Self { limit, state: Mutex::new(WorkState { next: 0, reserved: 0, cancelled: false }), wake: Condvar::new() } - } - - fn next(&self, job_count: usize) -> Option { - let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); - while state.reserved > self.limit - MAX_DECODED_BLOCK && state.next < job_count && !state.cancelled { - state = self.wake.wait(state).unwrap_or_else(|error| error.into_inner()); - } - if state.cancelled || state.next >= job_count { - return None; - } - let next = state.next; - state.next += 1; - state.reserved += MAX_DECODED_BLOCK; - Some(next) - } - - fn complete(&self, decoded_len: usize) { - let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); - state.reserved -= MAX_DECODED_BLOCK - decoded_len; - self.wake.notify_all(); - } - - fn retire(&self, decoded_len: usize) { - let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); - state.reserved -= decoded_len; - self.wake.notify_all(); - } - - fn cancel(&self) { - let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); - state.cancelled = true; - self.wake.notify_all(); - } -} - -type CandidateResult = (usize, Result); - -fn worker_loop(data: &[u8], jobs: &[CandidateJob], work: &WorkQueue, sender: &mpsc::Sender) { - while let Some(index) = work.next(jobs.len()) { - let job = jobs[index]; - let result = decoder::decode_candidate(data, job.start_bit, job.expected_crc); - let decoded_len = candidate_len(&result); - work.complete(decoded_len); - if sender.send((job.marker_index, result)).is_err() { - work.retire(decoded_len); - return; - } - } -} - fn candidate_len(result: &Result) -> usize { result.as_ref().map_or(0, |decoded| decoded.output.len()) } -struct ParallelCandidates<'a> { - receiver: mpsc::Receiver, - ready: HashMap>, - work: &'a WorkQueue, +struct ParallelCandidates<'results, 'pipeline> { + results: &'results mut OrderedResults<'pipeline, Result>, } -impl Candidates for ParallelCandidates<'_> { +impl Candidates for ParallelCandidates<'_, '_> { fn take(&mut self, marker_index: usize) -> Result { - while !self.ready.contains_key(&marker_index) { - let (index, result) = self.receiver.recv().map_err(|_| Error::InvalidConfiguration("parallel decoder stopped early".into()))?; - if index < marker_index { - self.work.retire(candidate_len(&result)); - } else { - self.ready.insert(index, result); - } - } - let result = self.ready.remove(&marker_index).unwrap(); - self.work.retire(candidate_len(&result)); - result + self.results.take(marker_index)? } fn discard_before(&mut self, marker_index: usize) { - let stale: Vec<_> = self.ready.keys().copied().filter(|&index| index < marker_index).collect(); - for index in stale { - let result = self.ready.remove(&index).unwrap(); - self.work.retire(candidate_len(&result)); - } + self.results.discard_before(marker_index); } } -fn thread_pool(threads: usize) -> Result>> { +pub(crate) fn thread_pool(threads: usize) -> Result>> { if threads <= 1 { return Ok(None); } diff --git a/src/error.rs b/src/error.rs index bd62af1..d79ffaf 100644 --- a/src/error.rs +++ b/src/error.rs @@ -37,6 +37,7 @@ pub enum Error { InvalidBitOffset { bit_offset: u64, len_bits: u64 }, UnexpectedEof { bit_offset: u64, requested: u64, remaining: u64 }, InvalidStreamHeader, + InvalidGzip(String), Decode { bit_offset: u64, source: DecodeError }, InvalidIndex(String), InvalidConfiguration(String), @@ -54,6 +55,7 @@ impl fmt::Display for Error { write!(f, "unexpected end of input at bit {bit_offset}: requested {requested} bits, {remaining} remain") } Self::InvalidStreamHeader => write!(f, "input does not start with a bzip2 BZh1-BZh9 header"), + Self::InvalidGzip(message) => write!(f, "invalid gzip stream: {message}"), Self::Decode { bit_offset, source } => write!(f, "bzip2 decode error at bit {bit_offset}: {source}"), Self::InvalidIndex(message) => write!(f, "invalid fastbz2 index: {message}"), Self::InvalidConfiguration(message) => write!(f, "invalid configuration: {message}"), diff --git a/src/gzip.rs b/src/gzip.rs new file mode 100644 index 0000000..0be7612 --- /dev/null +++ b/src/gzip.rs @@ -0,0 +1,1350 @@ +//! Gzip framing and DEFLATE decompression implemented in safe Rust. + +use std::{io::Write, sync::OnceLock}; + +use crate::{ + DecodeOptions, DecodeProgress, Error, Result, + pipeline::{Job, PipelineLimits, run_staged_ordered}, +}; + +const WINDOW_SIZE: usize = 32 * 1024; +const OUTPUT_CHUNK: usize = 64 * 1024; +const HISTORY_COMPACT: usize = 1024 * 1024; +const MAX_CODE_BITS: usize = 15; +const PARALLEL_GRID: usize = 1024 * 1024; +const MIN_PARALLEL_INPUT: usize = 16 * PARALLEL_GRID; +const PARALLEL_OUTPUT_LIMIT: usize = 8 * 1024 * 1024; +const PARALLEL_JOB_MEMORY: usize = 2 * PARALLEL_OUTPUT_LIMIT + 64 * 1024; + +const LENGTH_BASE: [usize; 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 LENGTH_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 DISTANCE_BASE: [usize; 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 DISTANCE_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]; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum BlockKind { + Stored, + FixedHuffman, + DynamicHuffman, +} + +impl BlockKind { + pub fn as_str(self) -> &'static str { + match self { + Self::Stored => "stored", + Self::FixedHuffman => "fixed", + Self::DynamicHuffman => "dynamic", + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Member { + pub compressed_start: u64, + pub deflate_start: u64, + pub compressed_end: u64, + pub decoded_start: u64, + pub decoded_len: u64, + pub expected_crc: u32, + pub mtime: u32, + pub extra_flags: u8, + pub operating_system: u8, + pub name: Option>, + pub comment: Option>, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Block { + pub member: u32, + pub kind: BlockKind, + pub final_block: bool, + pub compressed_start_bit: u64, + pub compressed_end_bit: u64, + pub decoded_start: u64, + pub decoded_len: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Report { + pub source_len: u64, + pub decoded_len: u64, + pub members: Vec, + pub blocks: Vec, + pub speculative_chunks: u64, + pub fallback_chunks: u64, +} + +#[derive(Clone, Debug)] +struct Header { + deflate_start: usize, + mtime: u32, + extra_flags: u8, + operating_system: u8, + name: Option>, + comment: Option>, +} + +pub fn decompress(data: &[u8]) -> Result> { + decompress_with_options(data, DecodeOptions::default()) +} + +pub fn decompress_with_options(data: &[u8], options: DecodeOptions) -> Result> { + let mut output = Vec::new(); + decompress_to_writer_with_options(data, &mut output, options)?; + Ok(output) +} + +pub fn decompress_to_writer(data: &[u8], output: &mut impl Write) -> Result { + decompress_to_writer_with_options(data, output, DecodeOptions::default()) +} + +pub fn decompress_to_writer_with_options(data: &[u8], output: &mut impl Write, options: DecodeOptions) -> Result { + decompress_to_writer_with_options_and_progress(data, output, options, |_| {}) +} + +pub fn decompress_to_writer_with_progress(data: &[u8], output: &mut impl Write, progress: impl FnMut(DecodeProgress)) -> Result { + decompress_to_writer_with_options_and_progress(data, output, DecodeOptions::default(), progress) +} + +pub fn decompress_to_writer_with_options_and_progress( + data: &[u8], + output: &mut impl Write, + options: DecodeOptions, + mut progress: impl FnMut(DecodeProgress), +) -> Result { + let mut options = options.validate()?; + let threads = options.resolved_threads(); + options.threads = threads; + if threads == 1 || data.len() < MIN_PARALLEL_INPUT || options.memory_limit < PARALLEL_JOB_MEMORY { + return decompress_serial_to_writer_with_progress(data, output, progress); + } + let header = parse_header(data, 0)?; + let first_grid = header.deflate_start.saturating_add(PARALLEL_GRID); + let initial_segment = match decode_segment(data, header.deflate_start * 8, first_grid.min(data.len()) * 8, InitialHistory::Empty, PARALLEL_OUTPUT_LIMIT) { + Ok(segment) => segment, + Err(_) => return decompress_serial_to_writer_with_progress(data, output, progress), + }; + decompress_parallel_to_writer(data, output, options, threads, &mut progress, initial_segment) +} + +fn decompress_serial_to_writer_with_progress(data: &[u8], output: &mut impl Write, mut progress: impl FnMut(DecodeProgress)) -> Result { + let mut members = Vec::new(); + let mut blocks = Vec::new(); + let mut position = 0_usize; + let mut decoded_total = 0_u64; + + while position < data.len() { + if !members.is_empty() && data[position..].iter().all(|&byte| byte == 0) { + break; + } + let member_start = position; + let header = parse_header(data, position)?; + let member_number = u32::try_from(members.len()).map_err(|_| invalid("too many gzip members"))?; + let mut emitter = Emitter::new(output, decoded_total); + let mut bits = Bits::new(data, header.deflate_start); + decode_deflate(&mut bits, &mut emitter, member_number, &mut blocks, &mut progress)?; + bits.align_byte(); + let trailer = bits.byte_position(); + let trailer_end = trailer.checked_add(8).ok_or_else(|| invalid("trailer offset overflow"))?; + let trailer_bytes = data.get(trailer..trailer_end).ok_or_else(|| invalid_at(trailer, "truncated member trailer"))?; + let expected_crc = u32::from_le_bytes(trailer_bytes[..4].try_into().unwrap()); + let expected_size = u32::from_le_bytes(trailer_bytes[4..].try_into().unwrap()); + let (actual_crc, decoded_len) = emitter.finish()?; + if actual_crc != expected_crc { + return Err(invalid_at(trailer, format!("CRC32 mismatch: expected {expected_crc:08x}, decoded {actual_crc:08x}"))); + } + if decoded_len as u32 != expected_size { + return Err(invalid_at(trailer + 4, format!("ISIZE mismatch: expected {expected_size}, decoded {}", decoded_len as u32))); + } + decoded_total = decoded_total.checked_add(decoded_len).ok_or_else(|| invalid("decoded offset overflow"))?; + position = trailer_end; + members.push(Member { + compressed_start: member_start as u64, + deflate_start: header.deflate_start as u64, + compressed_end: position as u64, + decoded_start: decoded_total - decoded_len, + decoded_len, + expected_crc, + mtime: header.mtime, + extra_flags: header.extra_flags, + operating_system: header.operating_system, + name: header.name, + comment: header.comment, + }); + progress(DecodeProgress { compressed_bytes: position as u64, decoded_bytes: decoded_total }); + } + if members.is_empty() { + return Err(invalid("input contains no gzip members")); + } + output.flush()?; + progress(DecodeProgress { compressed_bytes: data.len() as u64, decoded_bytes: decoded_total }); + Ok(Report { source_len: data.len() as u64, decoded_len: decoded_total, members, blocks, speculative_chunks: 0, fallback_chunks: 0 }) +} + +#[derive(Clone, Copy)] +enum InitialHistory { + Empty, + Unknown, +} + +fn extend_match(output: &mut Vec, distance: usize, length: usize) { + output.reserve(length); + let append_start = output.len(); + let first = distance.min(length); + output.extend_from_within(append_start - distance..append_start - distance + first); + let mut copied = first; + while copied < length { + let count = copied.min(length - copied); + output.extend_from_within(append_start..append_start + count); + copied += count; + } +} + +struct MarkerOutput { + marked: Vec, + clean: Vec, + clean_start: usize, + history: InitialHistory, + limit: usize, + clean_mode: bool, +} + +impl MarkerOutput { + fn new(history: InitialHistory, limit: usize) -> Self { + Self { marked: Vec::new(), clean: Vec::new(), clean_start: 0, history, limit, clean_mode: matches!(history, InitialHistory::Empty) } + } + + fn len(&self) -> usize { + self.marked.len() + self.clean.len().saturating_sub(self.clean_start) + } + + fn ensure_capacity(&self, additional: usize) -> Result<()> { + if additional > self.limit.saturating_sub(self.len()) { + return Err(invalid("parallel DEFLATE chunk exceeded its memory budget")); + } + Ok(()) + } + + fn try_clean(&mut self) { + if self.clean_mode || self.marked.len() < WINDOW_SIZE { + return; + } + let suffix = &self.marked[self.marked.len() - WINDOW_SIZE..]; + if suffix.iter().any(|&symbol| symbol > u8::MAX as u16) { + return; + } + self.clean = Vec::with_capacity(WINDOW_SIZE + 4 * PARALLEL_GRID); + self.clean.extend(suffix.iter().map(|&symbol| symbol as u8)); + self.clean_start = WINDOW_SIZE; + self.clean_mode = true; + } +} + +impl DeflateOutput for MarkerOutput { + fn total_decoded(&self) -> u64 { + self.len() as u64 + } + + fn emit(&mut self, byte: u8) -> Result<()> { + self.ensure_capacity(1)?; + if self.clean_mode { + self.clean.push(byte); + } else { + self.marked.push(u16::from(byte)); + } + Ok(()) + } + + fn extend(&mut self, bytes: &[u8]) -> Result<()> { + self.ensure_capacity(bytes.len())?; + if self.clean_mode { + self.clean.extend_from_slice(bytes); + } else { + self.marked.extend(bytes.iter().map(|&byte| u16::from(byte))); + } + Ok(()) + } + + fn copy(&mut self, distance: usize, length: usize) -> Result<()> { + self.ensure_capacity(length)?; + if self.clean_mode { + let available = self.clean.len().min(WINDOW_SIZE); + if distance == 0 || distance > available { + return Err(invalid(format!("back-reference distance {distance} exceeds {available} available bytes"))); + } + extend_match(&mut self.clean, distance, length); + return Ok(()); + } + + let available = match self.history { + InitialHistory::Empty => self.marked.len().min(WINDOW_SIZE), + InitialHistory::Unknown => WINDOW_SIZE, + }; + if distance == 0 || distance > available { + return Err(invalid(format!("back-reference distance {distance} exceeds {available} available bytes"))); + } + self.marked.reserve(length); + let append_start = self.marked.len(); + let mut copied = 0; + if distance > append_start { + let from_window = (distance - append_start).min(length); + let first = WINDOW_SIZE + append_start - distance; + self.marked.extend((0..from_window).map(|offset| (WINDOW_SIZE + first + offset) as u16)); + copied = from_window; + } + if copied < length { + extend_match(&mut self.marked, distance, length - copied); + } + Ok(()) + } +} + +struct Segment { + start_bit: usize, + end_bit: usize, + marked: Vec, + clean: Vec, + clean_start: usize, + blocks: Vec, + final_block: bool, +} + +impl Segment { + fn clean_output(&self) -> &[u8] { + &self.clean[self.clean_start..] + } + + fn retained_bytes(&self) -> usize { + self.marked.capacity() * size_of::() + self.clean.capacity() + self.blocks.capacity() * size_of::() + } +} + +fn decode_segment(data: &[u8], start_bit: usize, stop_bit: usize, history: InitialHistory, output_limit: usize) -> Result { + let mut bits = Bits::at(data, start_bit)?; + let mut emitter = MarkerOutput::new(history, output_limit); + let mut blocks = Vec::new(); + let final_block = loop { + let final_block = decode_block(&mut bits, &mut emitter, 0, &mut blocks, &mut |_| {})?; + emitter.try_clean(); + if final_block { + bits.align_byte(); + break true; + } + if bits.position_bits() >= stop_bit && blocks.last().is_some_and(|block| block.kind == BlockKind::DynamicHuffman) { + break false; + } + }; + Ok(Segment { + start_bit, + end_bit: bits.position_bits(), + marked: emitter.marked, + clean: emitter.clean, + clean_start: emitter.clean_start, + blocks, + final_block, + }) +} + +#[derive(Clone, Copy)] +struct GzipJob { + search_start: usize, + search_end: usize, + stop_bit: usize, + output_limit: usize, +} + +fn run_gzip_job(data: &[u8], job: &GzipJob) -> Result { + let mut search = job.search_start; + loop { + let start = find_dynamic_boundary(data, search, job.search_end) + .ok_or_else(|| invalid_bit(job.search_start, "no decodable dynamic DEFLATE boundary found near the parallel grid point"))?; + match decode_segment(data, start, job.stop_bit, InitialHistory::Unknown, job.output_limit) { + Ok(segment) => return Ok(segment), + Err(_) => search = start.saturating_add(1), + } + } +} + +fn find_dynamic_boundary(data: &[u8], start_bit: usize, end_bit: usize) -> Option { + let end = end_bit.min(data.len().saturating_mul(8)); + let first_byte = start_bit / 8; + let last_byte = end.div_ceil(8).min(data.len()); + for byte_offset in first_byte..last_byte { + let low = u16::from(data[byte_offset]); + let high = data.get(byte_offset + 1).map_or(0, |byte| u16::from(*byte)); + let header_window = low | (high << 8); + for bit_in_byte in 0..8 { + let bit = byte_offset * 8 + bit_in_byte; + if bit < start_bit || bit.saturating_add(13) >= end || ((header_window >> bit_in_byte) & 0b111) != 0b100 { + continue; + } + let mut header = Bits::at(data, bit + 3).ok()?; + if header.read(5).ok()? > 29 || header.read(5).ok()? > 29 || !valid_precode_shape(data, bit) { + continue; + } + let mut validation = Bits::at(data, bit + 3).ok()?; + if dynamic_tables(&mut validation).is_ok() { + return Some(bit); + } + } + } + None +} + +fn valid_precode_shape(data: &[u8], block_bit: usize) -> bool { + const PRECODE_BITS: usize = 4 + 19 * 3; + let Some(precode_bit) = block_bit.checked_add(13) else { return false }; + if precode_bit.checked_add(PRECODE_BITS).is_none_or(|end| end > data.len().saturating_mul(8)) { + return false; + } + let byte = precode_bit / 8; + let shift = precode_bit & 7; + let low = word_at(data, byte); + let bits = if shift == 0 { low } else { (low >> shift) | (u64::from(data.get(byte + 8).copied().unwrap_or(0)) << (64 - shift)) }; + let count = 4 + (bits & 0b1111) as usize; + let lengths = bits >> 4; + let mut counts = [0_u8; 8]; + let mut used = 0_u8; + for index in 0..count { + let length = ((lengths >> (index * 3)) & 0b111) as usize; + if length != 0 { + counts[length] += 1; + used += 1; + } + } + if used == 0 { + return false; + } + let mut remaining = 1_i16; + for count in counts.iter().skip(1) { + remaining = remaining * 2 - i16::from(*count); + if remaining < 0 { + return false; + } + } + remaining == 0 || used == 1 +} + +#[inline(always)] +fn word_at(data: &[u8], byte: usize) -> u64 { + if data.len().saturating_sub(byte) >= 8 { + u64::from_le_bytes(data[byte..byte + 8].try_into().unwrap()) + } else { + data[byte..].iter().take(8).enumerate().fold(0_u64, |word, (index, &value)| word | u64::from(value) << (index * 8)) + } +} + +fn resolve_symbols(symbols: &[u16], predecessor: &[u8]) -> Result> { + let mut resolved = Vec::with_capacity(symbols.len()); + if predecessor.len() == WINDOW_SIZE && symbols.len() >= 128 * 1024 { + let mut lookup = [0_u8; u16::MAX as usize + 1]; + for (value, byte) in lookup[..=u8::MAX as usize].iter_mut().enumerate() { + *byte = value as u8; + } + lookup[WINDOW_SIZE..].copy_from_slice(predecessor); + for (target, &symbol) in resolved.spare_capacity_mut().iter_mut().zip(symbols) { + target.write(lookup[symbol as usize]); + } + // SAFETY: the loop initialized one distinct spare-capacity byte per input symbol. + unsafe { resolved.set_len(symbols.len()) }; + } else { + let missing = WINDOW_SIZE.saturating_sub(predecessor.len()); + for &symbol in symbols { + let byte = if symbol <= u8::MAX as u16 { + symbol as u8 + } else { + let index = symbol as usize - WINDOW_SIZE; + if index < missing { + return Err(invalid("speculative marker references unavailable predecessor history")); + } + predecessor[index - missing] + }; + resolved.push(byte); + } + } + Ok(resolved) +} + +fn successor_window(segment: &Segment, predecessor: &[u8]) -> Result> { + let clean = segment.clean_output(); + if clean.len() >= WINDOW_SIZE { + return Ok(clean[clean.len() - WINDOW_SIZE..].to_vec()); + } + let marked_count = WINDOW_SIZE.saturating_sub(clean.len()).min(segment.marked.len()); + let marked = resolve_symbols(&segment.marked[segment.marked.len() - marked_count..], predecessor)?; + let keep_predecessor = WINDOW_SIZE.saturating_sub(marked.len() + clean.len()).min(predecessor.len()); + let mut window = Vec::with_capacity(keep_predecessor + marked.len() + clean.len()); + window.extend_from_slice(&predecessor[predecessor.len() - keep_predecessor..]); + window.extend_from_slice(&marked); + window.extend_from_slice(clean); + Ok(window) +} + +struct ResolveTask { + segment: Segment, + predecessor: Vec, +} + +struct ResolvedSegment { + marked: Vec, + clean: Vec, + clean_start: usize, + blocks: Vec, + compressed_end_bit: usize, + crc: crc32fast::Hasher, +} + +fn resolve_segment(segment: Segment, predecessor: &[u8]) -> Result { + let marked = resolve_symbols(&segment.marked, predecessor)?; + let mut crc = crc32fast::Hasher::new(); + crc.update(&marked); + crc.update(&segment.clean[segment.clean_start..]); + Ok(ResolvedSegment { marked, clean: segment.clean, clean_start: segment.clean_start, blocks: segment.blocks, compressed_end_bit: segment.end_bit, crc }) +} + +struct SegmentCommitter<'a, W: ?Sized, P: ?Sized> { + member: u32, + decoded_base: u64, + decoded: u64, + output: &'a mut W, + crc: crc32fast::Hasher, + blocks: &'a mut Vec, + progress: &'a mut P, +} + +impl<'a, W: Write + ?Sized, P: FnMut(DecodeProgress) + ?Sized> SegmentCommitter<'a, W, P> { + fn new(member: u32, decoded_base: u64, output: &'a mut W, blocks: &'a mut Vec, progress: &'a mut P) -> Self { + Self { member, decoded_base, decoded: 0, output, crc: crc32fast::Hasher::new(), blocks, progress } + } + + fn commit(&mut self, mut segment: ResolvedSegment) -> Result<()> { + self.output.write_all(&segment.marked)?; + self.output.write_all(&segment.clean[segment.clean_start..])?; + self.crc.combine(&segment.crc); + let decoded_len = segment.marked.len() + segment.clean.len() - segment.clean_start; + for block in &mut segment.blocks { + block.member = self.member; + block.decoded_start += self.decoded_base + self.decoded; + } + self.decoded = self.decoded.checked_add(decoded_len as u64).ok_or_else(|| invalid("decoded offset overflow"))?; + self.blocks.append(&mut segment.blocks); + (self.progress)(DecodeProgress { compressed_bytes: segment.compressed_end_bit.div_ceil(8) as u64, decoded_bytes: self.decoded_base + self.decoded }); + Ok(()) + } +} + +fn decompress_parallel_to_writer( + data: &[u8], + output: &mut impl Write, + options: DecodeOptions, + threads: usize, + progress: &mut impl FnMut(DecodeProgress), + initial_segment: Segment, +) -> Result { + let mut members = Vec::new(); + let mut blocks = Vec::new(); + let mut position = 0; + let mut decoded_total = 0_u64; + let mut speculative_total = 0_u64; + let mut fallback_total = 0_u64; + let mut initial_segment = Some(initial_segment); + + while position < data.len() { + if !members.is_empty() && data[position..].iter().all(|&byte| byte == 0) { + break; + } + let member_start = position; + let header = parse_header(data, position)?; + let member_number = u32::try_from(members.len()).map_err(|_| invalid("too many gzip members"))?; + let first_grid = header.deflate_start.saturating_add(PARALLEL_GRID); + let member_initial = match initial_segment.take() { + Some(segment) => segment, + None => match decode_segment(data, header.deflate_start * 8, first_grid.min(data.len()) * 8, InitialHistory::Empty, PARALLEL_OUTPUT_LIMIT) { + Ok(segment) => segment, + Err(_) => { + let compressed_base = position as u64; + let decoded_base = decoded_total; + let member_base = u32::try_from(members.len()).map_err(|_| invalid("too many gzip members"))?; + let mut suffix = decompress_serial_to_writer_with_progress(&data[position..], output, |item| { + progress(DecodeProgress { + compressed_bytes: compressed_base + item.compressed_bytes, + decoded_bytes: decoded_base + item.decoded_bytes, + }); + })?; + for member in &mut suffix.members { + member.compressed_start += compressed_base; + member.deflate_start += compressed_base; + member.compressed_end += compressed_base; + member.decoded_start += decoded_base; + } + let compressed_base_bits = compressed_base.saturating_mul(8); + for block in &mut suffix.blocks { + block.member += member_base; + block.compressed_start_bit += compressed_base_bits; + block.compressed_end_bit += compressed_base_bits; + block.decoded_start += decoded_base; + } + decoded_total = decoded_total.checked_add(suffix.decoded_len).ok_or_else(|| invalid("decoded offset overflow"))?; + members.append(&mut suffix.members); + blocks.append(&mut suffix.blocks); + return Ok(Report { + source_len: data.len() as u64, + decoded_len: decoded_total, + members, + blocks, + speculative_chunks: speculative_total, + fallback_chunks: fallback_total, + }); + } + }, + }; + let per_job = PARALLEL_JOB_MEMORY; + let output_limit = PARALLEL_OUTPUT_LIMIT; + let horizon = options.resolved_threads().saturating_add(2); + let parallel_budget = options.memory_limit.min(per_job.saturating_mul(horizon)); + let mut jobs = Vec::new(); + let mut key = 1; + let mut grid = first_grid; + while grid < data.len() { + jobs.push(Job { + key, + reservation: per_job, + payload: GzipJob { + search_start: grid * 8, + search_end: grid.saturating_add(2 * PARALLEL_GRID).min(data.len()) * 8, + stop_bit: grid.saturating_add(PARALLEL_GRID).min(data.len()) * 8, + output_limit, + }, + }); + key += 1; + grid = grid.saturating_add(PARALLEL_GRID); + } + + let (trailer, decoded_len, expected_crc, speculative_chunks, fallback_chunks) = run_staged_ordered( + threads, + &jobs, + PipelineLimits { memory: parallel_budget, active: horizon.saturating_mul(2) }, + |job| run_gzip_job(data, job), + |result| result.as_ref().map_or(0, Segment::retained_bytes), + |task: ResolveTask| resolve_segment(task.segment, &task.predecessor), + |results| { + let mut predecessor = Vec::new(); + let mut committer = SegmentCommitter::new(member_number, decoded_total, output, &mut blocks, progress); + let mut key = 1; + let mut resolve_sequence = 0; + let mut next_resolve = 0; + let mut outstanding = 0; + let mut speculative_chunks = 0_u64; + let mut fallback_chunks = 0_u64; + + let mut segment = member_initial; + let mut next_start = segment.end_bit; + let mut final_block = segment.final_block; + let next_window = successor_window(&segment, &predecessor)?; + let resolved = resolve_segment(segment, &predecessor)?; + committer.commit(resolved)?; + predecessor = next_window; + + while !final_block { + let estimated_stop = header.deflate_start.saturating_add((key + 1) * PARALLEL_GRID).min(data.len()) * 8; + let (lease, speculative) = results.take_primary(key)?; + let accepted = matches!(&speculative, Ok(candidate) if candidate.start_bit == next_start); + if !accepted { + results.retire(lease); + fallback_chunks += 1; + while outstanding != 0 { + let resolved = results.take_stage(next_resolve)??; + committer.commit(resolved)?; + next_resolve += 1; + outstanding -= 1; + } + segment = decode_segment(data, next_start, estimated_stop, InitialHistory::Unknown, output_limit)?; + next_start = segment.end_bit; + final_block = segment.final_block; + let next_window = successor_window(&segment, &predecessor)?; + let resolved = resolve_segment(segment, &predecessor)?; + committer.commit(resolved)?; + predecessor = next_window; + key += 1; + continue; + } + + speculative_chunks += 1; + let segment = speculative.unwrap(); + next_start = segment.end_bit; + final_block = segment.final_block; + let next_window = successor_window(&segment, &predecessor)?; + results.submit(resolve_sequence, lease, ResolveTask { segment, predecessor })?; + predecessor = next_window; + resolve_sequence += 1; + outstanding += 1; + key += 1; + + if outstanding >= threads { + let resolved = results.take_stage(next_resolve)??; + committer.commit(resolved)?; + next_resolve += 1; + outstanding -= 1; + } + } + + while outstanding != 0 { + let resolved = results.take_stage(next_resolve)??; + committer.commit(resolved)?; + next_resolve += 1; + outstanding -= 1; + } + + let trailer = next_start.div_ceil(8); + let trailer_end = trailer.checked_add(8).ok_or_else(|| invalid("trailer offset overflow"))?; + let trailer_bytes = data.get(trailer..trailer_end).ok_or_else(|| invalid_at(trailer, "truncated member trailer"))?; + let expected_crc = u32::from_le_bytes(trailer_bytes[..4].try_into().unwrap()); + let expected_size = u32::from_le_bytes(trailer_bytes[4..].try_into().unwrap()); + let SegmentCommitter { decoded: member_decoded, crc, .. } = committer; + let actual_crc = crc.finalize(); + if actual_crc != expected_crc { + return Err(invalid_at(trailer, format!("CRC32 mismatch: expected {expected_crc:08x}, decoded {actual_crc:08x}"))); + } + if member_decoded as u32 != expected_size { + return Err(invalid_at(trailer + 4, format!("ISIZE mismatch: expected {expected_size}, decoded {}", member_decoded as u32))); + } + Ok((trailer_end, member_decoded, expected_crc, speculative_chunks, fallback_chunks)) + }, + )?; + speculative_total += speculative_chunks; + fallback_total += fallback_chunks; + + decoded_total = decoded_total.checked_add(decoded_len).ok_or_else(|| invalid("decoded offset overflow"))?; + position = trailer; + members.push(Member { + compressed_start: member_start as u64, + deflate_start: header.deflate_start as u64, + compressed_end: position as u64, + decoded_start: decoded_total - decoded_len, + decoded_len, + expected_crc, + mtime: header.mtime, + extra_flags: header.extra_flags, + operating_system: header.operating_system, + name: header.name, + comment: header.comment, + }); + progress(DecodeProgress { compressed_bytes: position as u64, decoded_bytes: decoded_total }); + } + + if members.is_empty() { + return Err(invalid("input contains no gzip members")); + } + output.flush()?; + progress(DecodeProgress { compressed_bytes: data.len() as u64, decoded_bytes: decoded_total }); + Ok(Report { + source_len: data.len() as u64, + decoded_len: decoded_total, + members, + blocks, + speculative_chunks: speculative_total, + fallback_chunks: fallback_total, + }) +} + +fn parse_header(data: &[u8], start: usize) -> Result
{ + let fixed_end = start.checked_add(10).ok_or_else(|| invalid("header offset overflow"))?; + let fixed = data.get(start..fixed_end).ok_or_else(|| invalid_at(start, "truncated member header"))?; + if fixed[0..2] != [0x1f, 0x8b] { + return Err(invalid_at(start, "missing 1f 8b magic")); + } + if fixed[2] != 8 { + return Err(invalid_at(start + 2, format!("unsupported compression method {}", fixed[2]))); + } + let flags = fixed[3]; + if flags & 0xe0 != 0 { + return Err(invalid_at(start + 3, format!("reserved header flags set: {flags:02x}"))); + } + let mtime = u32::from_le_bytes(fixed[4..8].try_into().unwrap()); + let mut cursor = fixed_end; + if flags & 0x04 != 0 { + let length_bytes = data.get(cursor..cursor + 2).ok_or_else(|| invalid_at(cursor, "truncated FEXTRA length"))?; + let length = u16::from_le_bytes(length_bytes.try_into().unwrap()) as usize; + cursor = cursor.checked_add(2 + length).ok_or_else(|| invalid("header offset overflow"))?; + if cursor > data.len() { + return Err(invalid_at(cursor.saturating_sub(length), "truncated FEXTRA data")); + } + } + let name = if flags & 0x08 != 0 { Some(read_zero_terminated(data, &mut cursor, "FNAME")?) } else { None }; + let comment = if flags & 0x10 != 0 { Some(read_zero_terminated(data, &mut cursor, "FCOMMENT")?) } else { None }; + if flags & 0x02 != 0 { + let expected_bytes = data.get(cursor..cursor + 2).ok_or_else(|| invalid_at(cursor, "truncated FHCRC"))?; + let expected = u16::from_le_bytes(expected_bytes.try_into().unwrap()); + let actual = crc32(&data[start..cursor]) as u16; + if actual != expected { + return Err(invalid_at(cursor, format!("header CRC16 mismatch: expected {expected:04x}, decoded {actual:04x}"))); + } + cursor += 2; + } + Ok(Header { deflate_start: cursor, mtime, extra_flags: fixed[8], operating_system: fixed[9], name, comment }) +} + +fn read_zero_terminated(data: &[u8], cursor: &mut usize, field: &str) -> Result> { + let rest = data.get(*cursor..).ok_or_else(|| invalid_at(*cursor, format!("truncated {field}")))?; + let length = rest.iter().position(|&byte| byte == 0).ok_or_else(|| invalid_at(*cursor, format!("unterminated {field}")))?; + let value = rest[..length].to_vec(); + *cursor = cursor.checked_add(length + 1).ok_or_else(|| invalid("header offset overflow"))?; + Ok(value) +} + +fn decode_deflate( + bits: &mut Bits<'_>, + emitter: &mut impl DeflateOutput, + member: u32, + blocks: &mut Vec, + progress: &mut impl FnMut(DecodeProgress), +) -> Result<()> { + loop { + if decode_block(bits, emitter, member, blocks, progress)? { + return Ok(()); + } + } +} + +fn decode_block( + bits: &mut Bits<'_>, + emitter: &mut impl DeflateOutput, + member: u32, + blocks: &mut Vec, + progress: &mut impl FnMut(DecodeProgress), +) -> Result { + let compressed_start_bit = bits.position_bits() as u64; + let decoded_start = emitter.total_decoded(); + let final_block = bits.read(1)? != 0; + let kind = match bits.read(2)? { + 0 => { + decode_stored(bits, emitter)?; + BlockKind::Stored + } + 1 => { + let (literal, distance) = fixed_tables()?; + decode_huffman_block(bits, emitter, literal, distance)?; + BlockKind::FixedHuffman + } + 2 => { + let (literal, distance) = dynamic_tables(bits)?; + decode_huffman_block(bits, emitter, &literal, &distance)?; + BlockKind::DynamicHuffman + } + _ => return Err(invalid_bit(bits.position_bits().saturating_sub(2), "reserved DEFLATE block type")), + }; + let decoded_end = emitter.total_decoded(); + let compressed_end_bit = bits.position_bits() as u64; + blocks.push(Block { member, kind, final_block, compressed_start_bit, compressed_end_bit, decoded_start, decoded_len: decoded_end - decoded_start }); + progress(DecodeProgress { compressed_bytes: compressed_end_bit.div_ceil(8), decoded_bytes: decoded_end }); + Ok(final_block) +} + +fn decode_stored(bits: &mut Bits<'_>, emitter: &mut impl DeflateOutput) -> Result<()> { + bits.align_byte(); + let length = bits.read(16)? as u16; + let complement = bits.read(16)? as u16; + if length != !complement { + return Err(invalid_bit(bits.position_bits().saturating_sub(16), "stored-block LEN/NLEN mismatch")); + } + emitter.extend(bits.read_aligned_bytes(length as usize)?)?; + Ok(()) +} + +fn decode_huffman_block(bits: &mut Bits<'_>, emitter: &mut impl DeflateOutput, literal: &Huffman, distance: &Huffman) -> Result<()> { + loop { + let symbol = literal.decode(bits)?; + match symbol { + 0..=255 => emitter.emit(symbol as u8)?, + 256 => return Ok(()), + 257..=285 => { + let length_index = symbol as usize - 257; + let length = LENGTH_BASE[length_index] + bits.read(LENGTH_EXTRA[length_index])? as usize; + let distance_symbol = distance.decode(bits)? as usize; + if distance_symbol >= DISTANCE_BASE.len() { + return Err(invalid_bit(bits.position_bits(), format!("invalid distance symbol {distance_symbol}"))); + } + let distance = DISTANCE_BASE[distance_symbol] + bits.read(DISTANCE_EXTRA[distance_symbol])? as usize; + emitter.copy(distance, length)?; + } + _ => return Err(invalid_bit(bits.position_bits(), format!("invalid literal/length symbol {symbol}"))), + } + } +} + +fn dynamic_tables(bits: &mut Bits<'_>) -> Result<(Huffman, Huffman)> { + const ORDER: [usize; 19] = [16, 17, 18, 0, 8, 7, 9, 6, 10, 5, 11, 4, 12, 3, 13, 2, 14, 1, 15]; + let literal_count = bits.read(5)? as usize + 257; + let distance_count = bits.read(5)? as usize + 1; + let code_count = bits.read(4)? as usize + 4; + let mut code_lengths = [0_u8; 19]; + for &symbol in &ORDER[..code_count] { + code_lengths[symbol] = bits.read(3)? as u8; + } + let code_table = Huffman::new(&code_lengths)?; + let total = literal_count + distance_count; + let mut lengths = Vec::with_capacity(total); + while lengths.len() < total { + match code_table.decode(bits)? { + length @ 0..=15 => lengths.push(length as u8), + 16 => { + let previous = *lengths.last().ok_or_else(|| invalid_bit(bits.position_bits(), "repeat code 16 has no previous length"))?; + let count = bits.read(2)? as usize + 3; + append_lengths(&mut lengths, total, previous, count, bits.position_bits())?; + } + 17 => { + let count = bits.read(3)? as usize + 3; + append_lengths(&mut lengths, total, 0, count, bits.position_bits())?; + } + 18 => { + let count = bits.read(7)? as usize + 11; + append_lengths(&mut lengths, total, 0, count, bits.position_bits())?; + } + symbol => return Err(invalid_bit(bits.position_bits(), format!("invalid code-length symbol {symbol}"))), + } + } + if lengths[256] == 0 { + return Err(invalid_bit(bits.position_bits(), "literal/length table has no end-of-block symbol")); + } + Ok((Huffman::new(&lengths[..literal_count])?, Huffman::new(&lengths[literal_count..])?)) +} + +fn append_lengths(lengths: &mut Vec, total: usize, value: u8, count: usize, bit: usize) -> Result<()> { + if lengths.len().saturating_add(count) > total { + return Err(invalid_bit(bit, "code-length repeat exceeds table")); + } + lengths.resize(lengths.len() + count, value); + Ok(()) +} + +fn fixed_tables() -> Result<&'static (Huffman, Huffman)> { + static TABLES: OnceLock<(Huffman, Huffman)> = OnceLock::new(); + if let Some(tables) = TABLES.get() { + return Ok(tables); + } + let mut literal_lengths = [0_u8; 288]; + literal_lengths[..144].fill(8); + literal_lengths[144..256].fill(9); + literal_lengths[256..280].fill(7); + literal_lengths[280..].fill(8); + let tables = (Huffman::new(&literal_lengths)?, Huffman::new(&[5; 32])?); + let _ = TABLES.set(tables); + Ok(TABLES.get().unwrap()) +} + +#[derive(Clone, Debug)] +struct Huffman { + // `(bit length << 9) | symbol`, indexed by the next `max_bits` stream bits. + table: Vec, + max_bits: u8, +} + +impl Huffman { + fn new(lengths: &[u8]) -> Result { + let mut counts = [0_u16; MAX_CODE_BITS + 1]; + let mut max_bits = 0_u8; + for &length in lengths { + if length as usize > MAX_CODE_BITS { + return Err(invalid(format!("Huffman code length {length} exceeds {MAX_CODE_BITS}"))); + } + if length != 0 { + counts[length as usize] += 1; + max_bits = max_bits.max(length); + } + } + if max_bits == 0 { + return Ok(Self { table: Vec::new(), max_bits: 0 }); + } + let mut remaining = 1_i32; + for &count in &counts[1..] { + remaining = (remaining << 1) - i32::from(count); + if remaining < 0 { + return Err(invalid("oversubscribed Huffman table")); + } + } + let mut next_code = [0_u16; MAX_CODE_BITS + 1]; + let mut code = 0_u16; + for bits in 1..=MAX_CODE_BITS { + code = (code + counts[bits - 1]) << 1; + next_code[bits] = code; + } + let mut table = vec![u16::MAX; 1_usize << max_bits]; + for (symbol, &length) in lengths.iter().enumerate() { + if length == 0 { + continue; + } + let canonical = next_code[length as usize]; + next_code[length as usize] += 1; + let reversed = reverse_low_bits(canonical, length) as usize; + let suffix_bits = max_bits - length; + let packed = (u16::from(length) << 9) | symbol as u16; + for suffix in 0..(1_usize << suffix_bits) { + table[reversed | suffix << length] = packed; + } + } + Ok(Self { table, max_bits }) + } + + #[inline(always)] + fn decode(&self, bits: &mut Bits<'_>) -> Result { + if self.max_bits == 0 { + return Err(invalid_bit(bits.position_bits(), "attempted to decode an empty Huffman table")); + } + let packed = self.table[bits.peek(self.max_bits)? as usize]; + if packed == u16::MAX { + return Err(invalid_bit(bits.position_bits(), "invalid Huffman code")); + } + bits.drop((packed >> 9) as u8); + Ok(packed & 0x01ff) + } +} + +fn reverse_low_bits(value: u16, count: u8) -> u16 { + value.reverse_bits() >> (u16::BITS as u8 - count) +} + +#[derive(Clone)] +struct Bits<'a> { + data: &'a [u8], + bit: usize, +} + +impl<'a> Bits<'a> { + fn new(data: &'a [u8], start_byte: usize) -> Self { + Self { data, bit: start_byte.saturating_mul(8) } + } + + fn at(data: &'a [u8], bit: usize) -> Result { + if bit > data.len().saturating_mul(8) { + return Err(invalid_bit(bit, "unexpected end of DEFLATE data")); + } + Ok(Self { data, bit }) + } + + #[inline(always)] + fn position_bits(&self) -> usize { + self.bit + } + + fn byte_position(&self) -> usize { + self.bit.div_ceil(8) + } + + #[inline(always)] + fn peek(&self, count: u8) -> Result { + let count = usize::from(count); + if self.bit.checked_add(count).is_none_or(|end| end > self.data.len().saturating_mul(8)) { + return Err(invalid_bit(self.bit, "unexpected end of DEFLATE data")); + } + let byte = self.bit / 8; + let shift = self.bit & 7; + let word = if self.data.len().saturating_sub(byte) >= 8 { + u64::from_le_bytes(self.data[byte..byte + 8].try_into().unwrap()) + } else { + self.data[byte..].iter().take(8).enumerate().fold(0_u64, |word, (index, &value)| word | u64::from(value) << (index * 8)) + }; + let mask = if count == 0 { 0 } else { (1_u64 << count) - 1 }; + Ok(((word >> shift) & mask) as u32) + } + + #[inline(always)] + fn drop(&mut self, count: u8) { + self.bit += usize::from(count); + } + + #[inline(always)] + fn read(&mut self, count: u8) -> Result { + if count == 0 { + return Ok(0); + } + let value = self.peek(count)?; + self.drop(count); + Ok(value) + } + + fn align_byte(&mut self) { + self.bit = self.bit.saturating_add(7) & !7; + } + + fn read_aligned_bytes(&mut self, count: usize) -> Result<&'a [u8]> { + if self.bit & 7 != 0 { + return Err(invalid_bit(self.bit, "internal unaligned byte read")); + } + let start = self.bit / 8; + let end = start.checked_add(count).ok_or_else(|| invalid("DEFLATE offset overflow"))?; + let bytes = self.data.get(start..end).ok_or_else(|| invalid_bit(self.bit, "truncated stored block"))?; + self.bit = end * 8; + Ok(bytes) + } +} + +trait DeflateOutput { + fn total_decoded(&self) -> u64; + fn emit(&mut self, byte: u8) -> Result<()>; + fn extend(&mut self, bytes: &[u8]) -> Result<()>; + fn copy(&mut self, distance: usize, length: usize) -> Result<()>; +} + +struct Emitter<'a, W> { + output: &'a mut W, + buffer: Vec, + history_len: usize, + crc: crc32fast::Hasher, + member_decoded: u64, + decoded_base: u64, +} + +impl<'a, W: Write> Emitter<'a, W> { + fn new(output: &'a mut W, decoded_base: u64) -> Self { + Self { + output, + buffer: Vec::with_capacity(WINDOW_SIZE + OUTPUT_CHUNK + 258), + history_len: 0, + crc: crc32fast::Hasher::new(), + member_decoded: 0, + decoded_base, + } + } + + fn decoded_position(&self) -> u64 { + self.decoded_base + self.member_decoded + } + + fn emit_byte(&mut self, byte: u8) -> Result<()> { + self.buffer.push(byte); + self.member_decoded = self.member_decoded.checked_add(1).ok_or_else(|| invalid("decoded offset overflow"))?; + if self.buffer.len() - self.history_len >= OUTPUT_CHUNK { + self.flush_pending()?; + } + Ok(()) + } + + fn extend_bytes(&mut self, bytes: &[u8]) -> Result<()> { + self.buffer.extend_from_slice(bytes); + self.member_decoded = self.member_decoded.checked_add(bytes.len() as u64).ok_or_else(|| invalid("decoded offset overflow"))?; + if self.buffer.len() - self.history_len >= OUTPUT_CHUNK { + self.flush_pending()?; + } + Ok(()) + } + + fn copy_match(&mut self, distance: usize, length: usize) -> Result<()> { + let available = self.member_decoded.min(WINDOW_SIZE as u64) as usize; + if distance == 0 || distance > available { + return Err(invalid(format!("back-reference distance {distance} exceeds {available} available bytes"))); + } + let original = self.buffer.len(); + if distance >= length { + self.buffer.extend_from_within(original - distance..original - distance + length); + } else if distance == 1 { + self.buffer.resize(original + length, self.buffer[original - 1]); + } else if length <= distance * 2 { + self.buffer.extend_from_within(original - distance..original); + self.buffer.extend_from_within(original..original + length - distance); + } else { + self.buffer.resize(original + length, 0); + self.buffer.copy_within(original - distance..original, original); + let mut copied = distance; + while copied < length { + let count = copied.min(length - copied); + self.buffer.copy_within(original..original + count, original + copied); + copied += count; + } + } + self.member_decoded = self.member_decoded.checked_add(length as u64).ok_or_else(|| invalid("decoded offset overflow"))?; + if self.buffer.len() - self.history_len >= OUTPUT_CHUNK { + self.flush_pending()?; + } + Ok(()) + } + + fn flush_pending(&mut self) -> Result<()> { + let pending = &self.buffer[self.history_len..]; + if pending.is_empty() { + return Ok(()); + } + self.output.write_all(pending)?; + self.crc.update(pending); + self.history_len = self.buffer.len(); + if self.buffer.len() >= HISTORY_COMPACT { + let keep = self.buffer.len().min(WINDOW_SIZE); + let start = self.buffer.len() - keep; + self.buffer.copy_within(start.., 0); + self.buffer.truncate(keep); + self.history_len = keep; + } + Ok(()) + } + + fn finish(mut self) -> Result<(u32, u64)> { + self.flush_pending()?; + Ok((self.crc.finalize(), self.member_decoded)) + } +} + +impl DeflateOutput for Emitter<'_, W> { + fn total_decoded(&self) -> u64 { + self.decoded_position() + } + + fn emit(&mut self, byte: u8) -> Result<()> { + self.emit_byte(byte) + } + + fn extend(&mut self, bytes: &[u8]) -> Result<()> { + self.extend_bytes(bytes) + } + + fn copy(&mut self, distance: usize, length: usize) -> Result<()> { + self.copy_match(distance, length) + } +} + +pub fn crc32(data: &[u8]) -> u32 { + crc32fast::hash(data) +} + +fn invalid(message: impl Into) -> Error { + Error::InvalidGzip(message.into()) +} + +fn invalid_at(byte: usize, message: impl Into) -> Error { + invalid(format!("at byte {byte}: {}", message.into())) +} + +fn invalid_bit(bit: usize, message: impl Into) -> Error { + invalid(format!("at bit {bit}: {}", message.into())) +} + +#[cfg(test)] +mod tests { + use std::io::Write as _; + + use flate2::{Compression, GzBuilder, write::GzEncoder}; + + use super::*; + + fn patterned(size: usize) -> Vec { + (0..size).map(|index| ((index * 37 + index / 251) & 255) as u8).collect() + } + + fn compress(data: &[u8], level: Compression) -> Vec { + let mut encoder = GzEncoder::new(Vec::new(), level); + encoder.write_all(data).unwrap(); + encoder.finish().unwrap() + } + + #[test] + fn decodes_stored_fixed_and_dynamic_blocks() { + let cases = + [(b"stored bytes".repeat(2_000), Compression::none()), (b"fixed huffman".to_vec(), Compression::fast()), (patterned(350_000), Compression::best())]; + let mut kinds = Vec::new(); + for (plain, level) in cases { + let compressed = compress(&plain, level); + let mut output = Vec::new(); + let report = decompress_to_writer(&compressed, &mut output).unwrap(); + assert_eq!(output, plain); + kinds.extend(report.blocks.into_iter().map(|block| block.kind)); + } + assert!(kinds.contains(&BlockKind::Stored)); + assert!(kinds.contains(&BlockKind::FixedHuffman)); + assert!(kinds.contains(&BlockKind::DynamicHuffman)); + } + + #[test] + fn validates_concatenated_members_and_reports_progress() { + let first = patterned(180_000); + let second = b"second member".repeat(4_000); + let mut compressed = compress(&first, Compression::fast()); + compressed.extend_from_slice(&compress(&second, Compression::best())); + let mut expected = first; + expected.extend_from_slice(&second); + let mut output = Vec::new(); + let mut reports = Vec::new(); + let report = decompress_to_writer_with_progress(&compressed, &mut output, |progress| reports.push(progress)).unwrap(); + assert_eq!(output, expected); + assert_eq!(report.members.len(), 2); + assert_eq!(reports.last(), Some(&DecodeProgress { compressed_bytes: compressed.len() as u64, decoded_bytes: expected.len() as u64 })); + assert!(reports.windows(2).all(|pair| pair[0].compressed_bytes <= pair[1].compressed_bytes && pair[0].decoded_bytes <= pair[1].decoded_bytes)); + } + + #[test] + fn parses_optional_header_fields_and_header_crc() { + let plain = b"header metadata"; + let base = GzBuilder::new().mtime(123456).operating_system(3).write(Vec::new(), Compression::fast()); + let mut encoder = base; + encoder.write_all(plain).unwrap(); + let base = encoder.finish().unwrap(); + let mut compressed = base[..10].to_vec(); + compressed[3] = 0x1e; + compressed.extend_from_slice(&(3_u16).to_le_bytes()); + compressed.extend_from_slice(b"xyz"); + compressed.extend_from_slice(b"name.txt\0comment\0"); + let header_crc = crc32(&compressed) as u16; + compressed.extend_from_slice(&header_crc.to_le_bytes()); + compressed.extend_from_slice(&base[10..]); + let mut output = Vec::new(); + let report = decompress_to_writer(&compressed, &mut output).unwrap(); + assert_eq!(output, plain); + assert_eq!(report.members[0].mtime, 123456); + assert_eq!(report.members[0].name.as_deref(), Some(b"name.txt".as_slice())); + assert_eq!(report.members[0].comment.as_deref(), Some(b"comment".as_slice())); + } + + #[test] + fn rejects_header_payload_and_size_corruption() { + let plain = patterned(20_000); + let compressed = compress(&plain, Compression::best()); + + let mut reserved = compressed.clone(); + reserved[3] |= 0x20; + assert!(matches!(decompress(&reserved), Err(Error::InvalidGzip(_)))); + + let mut payload = compressed.clone(); + payload[12] ^= 1; + assert!(decompress(&payload).is_err()); + + let mut crc = compressed.clone(); + let crc_byte = crc.len() - 8; + crc[crc_byte] ^= 1; + assert!(matches!(decompress(&crc), Err(Error::InvalidGzip(message)) if message.contains("CRC32 mismatch"))); + + let mut size = compressed; + let size_byte = size.len() - 4; + size[size_byte] ^= 1; + assert!(matches!(decompress(&size), Err(Error::InvalidGzip(message)) if message.contains("ISIZE mismatch"))); + } + + #[test] + fn differentially_decodes_varied_inputs_and_levels() { + let mut random = Vec::with_capacity(70_000); + let mut state = 0x1234_5678_u32; + for _ in 0..70_000 { + state ^= state << 13; + state ^= state >> 17; + state ^= state << 5; + random.push(state as u8); + } + let inputs = [Vec::new(), vec![0], vec![0; 32_769], b"abc".repeat(20_000), patterned(70_000), random]; + for plain in inputs { + for level in [Compression::none(), Compression::fast(), Compression::new(6), Compression::best()] { + let compressed = compress(&plain, level); + assert_eq!(decompress(&compressed).unwrap(), plain); + } + } + } + + #[test] + fn malformed_inputs_return_errors_without_panicking() { + let valid = compress(&patterned(4_000), Compression::best()); + for end in 0..valid.len() { + assert!(decompress(&valid[..end]).is_err()); + } + for byte in 0..valid.len().min(64) { + let mut damaged = valid.clone(); + damaged[byte] ^= 0x5a; + let _ = decompress(&damaged); + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 15ef4e6..bbee4b0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,8 +7,10 @@ mod decode; mod decoder; mod error; mod format; +pub mod gzip; mod index; mod indexed; +mod pipeline; mod source; pub use bitreader::BitReader; diff --git a/src/pipeline.rs b/src/pipeline.rs new file mode 100644 index 0000000..fd30622 --- /dev/null +++ b/src/pipeline.rs @@ -0,0 +1,310 @@ +use std::{ + collections::HashMap, + sync::{Condvar, Mutex, mpsc}, + thread, +}; + +use rayon::ThreadPool; + +use crate::{Error, Result}; + +pub(crate) struct Job { + pub key: usize, + pub reservation: usize, + pub payload: T, +} + +pub(crate) struct PipelineLimits { + pub memory: usize, + pub active: usize, +} + +struct State { + next: usize, + reserved: usize, + active: usize, + cancelled: bool, +} + +struct Budget { + limit: usize, + max_active: usize, + state: Mutex, + wake: Condvar, +} + +impl Budget { + fn new(limit: usize, max_active: usize) -> Self { + Self { limit, max_active, state: Mutex::new(State { next: 0, reserved: 0, active: 0, cancelled: false }), wake: Condvar::new() } + } + + fn next(&self, jobs: &[Job]) -> Option { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + loop { + if state.cancelled || state.next >= jobs.len() { + return None; + } + let reservation = jobs[state.next].reservation; + if state.active < self.max_active && reservation <= self.limit.saturating_sub(state.reserved) { + let next = state.next; + state.next += 1; + state.reserved += reservation; + state.active += 1; + return Some(next); + } + state = self.wake.wait(state).unwrap_or_else(|error| error.into_inner()); + } + } + + fn complete(&self, reservation: usize, retained: usize) { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + debug_assert!(retained <= reservation); + state.reserved = state.reserved.saturating_sub(reservation).saturating_add(retained.min(reservation)); + self.wake.notify_all(); + } + + fn retire(&self, retained: usize) { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + state.reserved = state.reserved.saturating_sub(retained); + state.active = state.active.saturating_sub(1); + self.wake.notify_all(); + } + + fn cancel(&self) { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + state.cancelled = true; + self.wake.notify_all(); + } +} + +type Message = (usize, usize, T); + +fn execute_job( + index: usize, + jobs: &[Job], + budget: &Budget, + sender: &mpsc::Sender>, + execute: &impl Fn(&T) -> O, + retained_size: &impl Fn(&O) -> usize, +) -> bool { + let job = &jobs[index]; + let value = execute(&job.payload); + let retained = retained_size(&value).min(job.reservation); + budget.complete(job.reservation, retained); + if sender.send((job.key, retained, value)).is_ok() { + true + } else { + budget.retire(retained); + false + } +} + +pub(crate) struct OrderedResults<'a, T> { + receiver: mpsc::Receiver>, + ready: HashMap, + budget: &'a Budget, +} + +impl OrderedResults<'_, T> { + pub fn take(&mut self, key: usize) -> Result { + while !self.ready.contains_key(&key) { + let (received_key, retained, value) = self.receiver.recv().map_err(|_| Error::InvalidConfiguration("parallel decoder stopped early".into()))?; + self.ready.insert(received_key, (retained, value)); + } + let (retained, value) = self.ready.remove(&key).unwrap(); + self.budget.retire(retained); + Ok(value) + } + + pub fn discard_before(&mut self, key: usize) { + let stale: Vec<_> = self.ready.keys().copied().filter(|&candidate| candidate < key).collect(); + for candidate in stale { + let (retained, _) = self.ready.remove(&candidate).unwrap(); + self.budget.retire(retained); + } + } +} + +pub(crate) fn run_ordered( + pool: &ThreadPool, + jobs: &[Job], + limits: PipelineLimits, + execute: impl Fn(&T) -> O + Sync, + retained_size: impl Fn(&O) -> usize + Sync, + consume: impl FnOnce(&mut OrderedResults<'_, O>) -> Result, +) -> Result +where + T: Sync, + O: Send, +{ + if jobs.iter().any(|job| job.reservation > limits.memory) { + return Err(Error::InvalidConfiguration("a parallel job reservation exceeds the memory limit".into())); + } + let budget = Budget::new(limits.memory, limits.active); + let (sender, receiver) = mpsc::channel(); + thread::scope(|scope| { + let worker = scope.spawn(|| { + pool.broadcast(|_| { + while let Some(index) = budget.next(jobs) { + if !execute_job(index, jobs, &budget, &sender, &execute, &retained_size) { + return; + } + } + }); + }); + let result = { + let mut results = OrderedResults { receiver, ready: HashMap::new(), budget: &budget }; + consume(&mut results) + }; + budget.cancel(); + worker.join().map_err(|_| Error::InvalidConfiguration("parallel decoder worker panicked".into()))?; + result + }) +} + +enum TryNext { + Ready(usize), + Pending, + Cancelled, +} + +impl Budget { + fn try_next(&self, jobs: &[Job]) -> TryNext { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + if state.cancelled { + return TryNext::Cancelled; + } + if state.next >= jobs.len() { + return TryNext::Pending; + } + let reservation = jobs[state.next].reservation; + if state.active >= self.max_active || reservation > self.limit.saturating_sub(state.reserved) { + return TryNext::Pending; + } + let next = state.next; + state.next += 1; + state.reserved += reservation; + state.active += 1; + TryNext::Ready(next) + } + + fn wait_briefly(&self) { + let state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let _ = self.wake.wait_timeout(state, std::time::Duration::from_millis(1)); + } + + fn notify(&self) { + self.wake.notify_all(); + } +} + +pub(crate) struct Lease { + retained: usize, +} + +pub(crate) struct StagedResults<'a, O, S, Q> { + primary: OrderedResults<'a, O>, + stage_sender: mpsc::Sender>, + stage_receiver: mpsc::Receiver<(usize, Q)>, + stage_ready: HashMap, +} + +impl<'a, O, S, Q> StagedResults<'a, O, S, Q> { + pub fn take_primary(&mut self, key: usize) -> Result<(Lease, O)> { + while !self.primary.ready.contains_key(&key) { + let (received_key, retained, value) = + self.primary.receiver.recv().map_err(|_| Error::InvalidConfiguration("parallel decoder stopped early".into()))?; + self.primary.ready.insert(received_key, (retained, value)); + } + let (retained, value) = self.primary.ready.remove(&key).unwrap(); + Ok((Lease { retained }, value)) + } + + pub fn retire(&self, lease: Lease) { + self.primary.budget.retire(lease.retained); + } + + pub fn submit(&self, key: usize, lease: Lease, payload: S) -> Result<()> { + if let Err(mpsc::SendError((_, retained, _))) = self.stage_sender.send((key, lease.retained, payload)) { + self.primary.budget.retire(retained); + return Err(Error::InvalidConfiguration("parallel decoder stopped before staged work was submitted".into())); + } + self.primary.budget.notify(); + Ok(()) + } + + pub fn take_stage(&mut self, key: usize) -> Result { + while !self.stage_ready.contains_key(&key) { + let (received_key, value) = self.stage_receiver.recv().map_err(|_| Error::InvalidConfiguration("parallel staged worker stopped early".into()))?; + self.stage_ready.insert(received_key, value); + } + Ok(self.stage_ready.remove(&key).unwrap()) + } +} + +pub(crate) fn run_staged_ordered( + worker_count: usize, + jobs: &[Job], + limits: PipelineLimits, + execute: impl Fn(&T) -> O + Sync, + retained_size: impl Fn(&O) -> usize + Sync, + execute_stage: impl Fn(S) -> Q + Sync, + consume: impl FnOnce(&mut StagedResults<'_, O, S, Q>) -> Result, +) -> Result +where + T: Sync, + O: Send, + S: Send, + Q: Send, +{ + if jobs.iter().any(|job| job.reservation > limits.memory) { + return Err(Error::InvalidConfiguration("a parallel job reservation exceeds the memory limit".into())); + } + let budget = Budget::new(limits.memory, limits.active); + let (primary_sender, primary_receiver) = mpsc::channel(); + let (stage_sender, stage_job_receiver) = mpsc::channel(); + let stage_job_receiver = Mutex::new(stage_job_receiver); + let (stage_result_sender, stage_receiver) = mpsc::channel(); + thread::scope(|scope| { + let workers: Vec<_> = (0..worker_count) + .map(|_| { + scope.spawn(|| { + loop { + let stage = stage_job_receiver.lock().unwrap_or_else(|error| error.into_inner()).try_recv(); + match stage { + Ok((key, retained, payload)) => { + let value = execute_stage(payload); + budget.retire(retained); + if stage_result_sender.send((key, value)).is_err() { + return; + } + continue; + } + Err(mpsc::TryRecvError::Disconnected) => return, + Err(mpsc::TryRecvError::Empty) => {} + } + match budget.try_next(jobs) { + TryNext::Ready(index) => { + if !execute_job(index, jobs, &budget, &primary_sender, &execute, &retained_size) { + return; + } + } + TryNext::Pending => budget.wait_briefly(), + TryNext::Cancelled => return, + } + } + }) + }) + .collect(); + let result = { + let primary = OrderedResults { receiver: primary_receiver, ready: HashMap::new(), budget: &budget }; + let mut results = StagedResults { primary, stage_sender, stage_receiver, stage_ready: HashMap::new() }; + consume(&mut results) + }; + budget.cancel(); + for worker in workers { + worker.join().map_err(|_| Error::InvalidConfiguration("parallel decoder worker panicked".into()))?; + } + result + }) +} diff --git a/tests/cli.rs b/tests/cli.rs index 2f38adb..492a39d 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -7,6 +7,7 @@ use std::{ }; use crabz2::{Level, compress}; +use flate2::{Compression, write::GzEncoder}; fn binary() -> Command { Command::new(env!("CARGO_BIN_EXE_fastbz2")) @@ -16,6 +17,12 @@ fn write_compressed(path: &Path, plain: &[u8]) { fs::write(path, compress(plain, Level::FASTEST)).unwrap(); } +fn write_gzip(path: &Path, plain: &[u8]) { + let mut encoder = GzEncoder::new(Vec::new(), Compression::best()); + encoder.write_all(plain).unwrap(); + fs::write(path, encoder.finish().unwrap()).unwrap(); +} + #[test] fn decode_test_index_and_list() { let directory = tempfile::tempdir().unwrap(); @@ -23,7 +30,7 @@ fn decode_test_index_and_list() { let output = directory.path().join("sample"); let index = directory.path().join("sample.fbz2i"); let plain: Vec<_> = (0..250_000).map(|i| ((i * 31 + i / 97) & 255) as u8).collect(); - fs::write(&input, compress(&plain, Level::FASTEST)).unwrap(); + write_compressed(&input, &plain); let decoded = binary().args([input.to_str().unwrap(), "-P", "2"]).status().unwrap(); assert!(decoded.success()); @@ -196,3 +203,82 @@ fn list_json_describes_one_or_many_inputs() { let missing_mode = binary().args(["--json", first.to_str().unwrap()]).output().unwrap(); assert_eq!(missing_mode.status.code(), Some(2)); } + +#[test] +fn gzip_extension_selects_decoder_across_cli_modes() { + let directory = tempfile::tempdir().unwrap(); + let input = directory.path().join("sample.gz"); + let output = directory.path().join("sample"); + let plain: Vec<_> = (0..250_000).map(|i| ((i * 31 + i / 97) & 255) as u8).collect(); + write_gzip(&input, &plain); + + let decoded = binary().arg(input.to_str().unwrap()).output().unwrap(); + assert!(decoded.status.success(), "{}", String::from_utf8_lossy(&decoded.stderr)); + assert_eq!(fs::read(output).unwrap(), plain); + + let tested = binary().args(["--test", input.to_str().unwrap()]).output().unwrap(); + assert!(tested.status.success()); + + let listed = binary().args(["--list", "--json", input.to_str().unwrap()]).output().unwrap(); + assert!(listed.status.success()); + let value: serde_json::Value = serde_json::from_slice(&listed.stdout).unwrap(); + assert_eq!(value["format"], "gzip"); + assert_eq!(value["decoded_bytes"], plain.len()); + assert_eq!(value["members"].as_array().unwrap().len(), 1); + assert!(!value["blocks"].as_array().unwrap().is_empty()); + + let indexed = binary().args(["--index", input.to_str().unwrap()]).output().unwrap(); + assert_eq!(indexed.status.code(), Some(2)); + assert!(String::from_utf8(indexed.stderr).unwrap().contains("only for bzip2")); +} + +#[test] +fn gzip_magic_fallback_stdin_limits_and_corruption_work() { + let directory = tempfile::tempdir().unwrap(); + let input = directory.path().join("mystery.data"); + let output = directory.path().join("mystery.data.out"); + let plain = b"gzip magic fallback".repeat(2_000); + write_gzip(&input, &plain); + + let decoded = binary().arg(input.to_str().unwrap()).output().unwrap(); + assert!(decoded.status.success()); + assert_eq!(fs::read(&output).unwrap(), plain); + fs::remove_file(&output).unwrap(); + + let compressed = fs::read(&input).unwrap(); + let mut child = binary().arg("-").stdin(Stdio::piped()).stdout(Stdio::piped()).spawn().unwrap(); + child.stdin.take().unwrap().write_all(&compressed).unwrap(); + let stdin = child.wait_with_output().unwrap(); + assert!(stdin.status.success()); + assert_eq!(stdin.stdout, plain); + + let limited = binary().args(["--max-output", "1K", input.to_str().unwrap()]).output().unwrap(); + assert_eq!(limited.status.code(), Some(3)); + assert!(!output.exists()); + + let mut corrupt = compressed; + let crc = corrupt.len() - 8; + corrupt[crc] ^= 1; + fs::write(&input, corrupt).unwrap(); + let rejected = binary().arg(input.to_str().unwrap()).output().unwrap(); + assert_eq!(rejected.status.code(), Some(3)); + assert!(!output.exists()); +} + +#[test] +fn mixed_bzip2_and_gzip_inputs_share_output_policy() { + let directory = tempfile::tempdir().unwrap(); + let bzip2 = directory.path().join("first.bz2"); + let gzip = directory.path().join("second.gz"); + let tgz = directory.path().join("bundle.tgz"); + let output_dir = directory.path().join("decoded"); + write_compressed(&bzip2, b"bzip2"); + write_gzip(&gzip, b"gzip"); + write_gzip(&tgz, b"tar payload"); + + let decoded = binary().args(["-C", output_dir.to_str().unwrap(), bzip2.to_str().unwrap(), gzip.to_str().unwrap(), tgz.to_str().unwrap()]).output().unwrap(); + assert!(decoded.status.success(), "{}", String::from_utf8_lossy(&decoded.stderr)); + assert_eq!(fs::read(output_dir.join("first")).unwrap(), b"bzip2"); + assert_eq!(fs::read(output_dir.join("second")).unwrap(), b"gzip"); + assert_eq!(fs::read(output_dir.join("bundle.tar")).unwrap(), b"tar payload"); +} diff --git a/tests/common/benchmark.rs b/tests/common/benchmark.rs new file mode 100644 index 0000000..7722efb --- /dev/null +++ b/tests/common/benchmark.rs @@ -0,0 +1,8 @@ +pub fn elapsed(repeats: usize, mut run: impl FnMut()) -> std::time::Duration { + run(); + let start = std::time::Instant::now(); + for _ in 0..repeats { + run() + } + start.elapsed() +} diff --git a/tests/common/mod.rs b/tests/common/mod.rs new file mode 100644 index 0000000..f85bae7 --- /dev/null +++ b/tests/common/mod.rs @@ -0,0 +1,5 @@ +#[cfg(unix)] +mod process; + +#[cfg(unix)] +pub use process::{ProcessMetrics, Timing, measure, measure_timing}; diff --git a/tests/common/process.rs b/tests/common/process.rs new file mode 100644 index 0000000..2bfc576 --- /dev/null +++ b/tests/common/process.rs @@ -0,0 +1,132 @@ +use std::{ + io, + mem::MaybeUninit, + os::unix::process::ExitStatusExt, + process::{Command, ExitStatus}, + time::{Duration, Instant}, +}; + +struct FootprintSampler { + stop: std::sync::Arc, + worker: std::thread::JoinHandle>, +} + +#[cfg(target_os = "macos")] +impl FootprintSampler { + fn start(pid: libc::pid_t) -> Self { + use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }; + let stop = Arc::new(AtomicBool::new(false)); + let worker_stop = Arc::clone(&stop); + let initial = physical_footprint(pid); + let worker = std::thread::spawn(move || { + let mut maximum = initial; + while !worker_stop.load(Ordering::Relaxed) { + if let Some(value) = physical_footprint(pid) { + maximum = Some(maximum.map_or(value, |previous: u64| previous.max(value))); + } + std::thread::sleep(Duration::from_millis(50)); + } + maximum + }); + Self { stop, worker } + } +} + +#[cfg(not(target_os = "macos"))] +impl FootprintSampler { + fn start(_pid: libc::pid_t) -> Self { + let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let worker = std::thread::spawn(|| None); + Self { stop, worker } + } +} + +#[cfg(target_os = "macos")] +fn physical_footprint(pid: libc::pid_t) -> Option { + #[repr(C)] + struct RusageInfoV4 { + uuid: [u8; 16], + values: [u64; 35], + } + unsafe extern "C" { + fn proc_pid_rusage(pid: libc::c_int, flavor: libc::c_int, buffer: *mut libc::c_void) -> libc::c_int; + } + let mut usage = RusageInfoV4 { uuid: [0; 16], values: [0; 35] }; + // SAFETY: flavor 4 requests the exact repr(C) buffer above, which remains + // writable for the call; `pid` is the benchmark process's own child. + let result = unsafe { proc_pid_rusage(pid, 4, (&mut usage as *mut RusageInfoV4).cast()) }; + (result == 0).then_some(usage.values[28]) +} + +#[derive(Clone, Copy, Debug)] +pub struct Timing { + pub status: ExitStatus, + pub wall: Duration, +} + +#[derive(Clone, Copy, Debug)] +pub struct ProcessMetrics { + pub status: ExitStatus, + pub wall: Duration, + pub user: Duration, + pub system: Duration, + pub peak_rss_bytes: u64, + pub peak_phys_footprint_bytes: Option, +} + +pub fn measure_timing(command: &mut Command) -> io::Result { + let started = Instant::now(); + let status = command.status()?; + Ok(Timing { status, wall: started.elapsed() }) +} + +pub fn measure(command: &mut Command) -> io::Result { + let started = Instant::now(); + let child = command.spawn()?; + let pid = child.id() as libc::pid_t; + let sampler = FootprintSampler::start(pid); + let mut status = 0; + let mut usage = MaybeUninit::::uninit(); + loop { + // SAFETY: `pid` names our live child, and both output pointers refer to + // valid writable storage for the duration of the call. + let result = unsafe { libc::wait4(pid, &mut status, 0, usage.as_mut_ptr()) }; + if result >= 0 { + break; + } + let error = io::Error::last_os_error(); + if error.kind() != io::ErrorKind::Interrupted { + return Err(error); + } + } + // SAFETY: a successful `wait4` initialized the complete `rusage` value. + let usage = unsafe { usage.assume_init() }; + let duration = |value: libc::timeval| Duration::new(value.tv_sec as u64, (value.tv_usec as u32) * 1_000); + #[cfg(target_os = "macos")] + let peak_rss_bytes = usage.ru_maxrss as u64; + #[cfg(not(target_os = "macos"))] + let peak_rss_bytes = (usage.ru_maxrss as u64).saturating_mul(1024); + sampler.stop.store(true, std::sync::atomic::Ordering::Relaxed); + let peak_phys_footprint_bytes = sampler.worker.join().unwrap_or(None); + Ok(ProcessMetrics { + status: ExitStatus::from_raw(status), + wall: started.elapsed(), + user: duration(usage.ru_utime), + system: duration(usage.ru_stime), + peak_rss_bytes, + peak_phys_footprint_bytes, + }) +} + +#[test] +fn collects_child_metrics_without_platform_time_tools() { + let metrics = measure(&mut Command::new("/usr/bin/true")).unwrap(); + assert!(metrics.status.success()); + assert!(metrics.wall > Duration::ZERO); + assert!(metrics.peak_rss_bytes > 0); + #[cfg(target_os = "macos")] + assert!(metrics.peak_phys_footprint_bytes.is_some()); +} diff --git a/tests/corpus.rs b/tests/corpus.rs index 763291a..cdafb7f 100644 --- a/tests/corpus.rs +++ b/tests/corpus.rs @@ -1,3 +1,7 @@ +#[cfg(not(debug_assertions))] +#[path = "common/benchmark.rs"] +mod benchmark; + use std::{ ffi::{c_char, c_uint}, fs, @@ -128,16 +132,6 @@ fn generated_shapes_match_oracle() { } } -#[cfg(not(debug_assertions))] -fn elapsed(repeats: usize, mut decode: impl FnMut()) -> std::time::Duration { - decode(); - let start = std::time::Instant::now(); - for _ in 0..repeats { - decode() - } - start.elapsed() -} - #[test] #[cfg(not(debug_assertions))] fn performance_regression_stays_bounded() { @@ -145,10 +139,10 @@ fn performance_regression_stays_bounded() { let plain = source.repeat(2); let encoded = compress(&plain, Level::FASTEST); let repeats = 3; - let fastbz2_time = elapsed(repeats, || { + let fastbz2_time = benchmark::elapsed(repeats, || { std::hint::black_box(decompress(&encoded, DecodeOptions { threads: 2, ..DecodeOptions::default() }).unwrap()); }); - let oracle_time = elapsed(repeats, || { + let oracle_time = benchmark::elapsed(repeats, || { std::hint::black_box(oracle_decompress(&encoded).unwrap()); }); assert!(fastbz2_time.as_secs_f64() <= oracle_time.as_secs_f64() * 1.3, "fastbz2 {fastbz2_time:?} exceeded 1.3x oracle {oracle_time:?}"); diff --git a/tests/gzip_oracle.rs b/tests/gzip_oracle.rs new file mode 100644 index 0000000..76effb0 --- /dev/null +++ b/tests/gzip_oracle.rs @@ -0,0 +1,111 @@ +#[cfg(not(debug_assertions))] +#[path = "common/benchmark.rs"] +mod benchmark; + +use std::io::Read; + +#[cfg(not(debug_assertions))] +use fastbz2::DecodeOptions; +use fastbz2::gzip; +use flate2::{Compression, read::MultiGzDecoder, write::GzEncoder}; + +fn oracle_decompress(input: &[u8]) -> Vec { + let mut output = Vec::new(); + MultiGzDecoder::new(input).read_to_end(&mut output).unwrap(); + output +} + +fn patterned(size: usize) -> Vec { + (0..size).map(|index| ((index * 37 + index / 251) & 255) as u8).collect() +} + +#[cfg(not(debug_assertions))] +fn random_bytes(size: usize) -> Vec { + let mut output = Vec::with_capacity(size); + let mut state = 0x1234_5678_u32; + for _ in 0..size { + state ^= state << 13; + state ^= state >> 17; + state ^= state << 5; + output.push(state as u8); + } + output +} + +#[cfg(not(debug_assertions))] +fn random_nibbles(size: usize) -> Vec { + let mut bytes = random_bytes(size); + for byte in &mut bytes { + *byte &= 0x0f; + } + bytes +} + +fn compress(input: &[u8]) -> Vec { + use std::io::Write as _; + let mut encoder = GzEncoder::new(Vec::new(), Compression::new(6)); + encoder.write_all(input).unwrap(); + encoder.finish().unwrap() +} + +#[test] +fn gzip_public_api_matches_oracle() { + let plain = patterned(250_000); + let encoded = compress(&plain); + assert_eq!(gzip::decompress(&encoded).unwrap(), oracle_decompress(&encoded)); +} + +#[test] +#[cfg(not(debug_assertions))] +fn large_stored_gzip_matches_oracle() { + let plain = random_bytes(20 * 1024 * 1024); + let encoded = compress(&plain); + assert!(encoded.len() >= 16 * 1024 * 1024); + let options = DecodeOptions { threads: 4, memory_limit: 256 * 1024 * 1024 }; + assert_eq!(gzip::decompress_with_options(&encoded, options).unwrap(), oracle_decompress(&encoded)); +} + +#[test] +#[cfg(not(debug_assertions))] +fn parallel_dynamic_gzip_matches_oracle() { + let mut plain = random_nibbles(40 * 1024 * 1024); + let mut encoded = compress(&plain); + let stored = random_bytes(20 * 1024 * 1024); + encoded.extend(compress(&stored)); + plain.extend(stored); + assert!(encoded.len() >= 32 * 1024 * 1024); + let options = DecodeOptions { threads: 4, memory_limit: 256 * 1024 * 1024 }; + let mut decoded = Vec::new(); + let report = gzip::decompress_to_writer_with_options(&encoded, &mut decoded, options).unwrap(); + assert!(report.speculative_chunks > 0); + assert_eq!(report.members.len(), 2); + assert_eq!(decoded, plain); +} + +#[cfg(not(debug_assertions))] +fn assert_performance(plain: &[u8], limit: f64) { + let encoded = compress(plain); + assert_eq!(gzip::decompress(&encoded).unwrap(), oracle_decompress(&encoded)); + let repeats = 2; + let fastbz2_time = benchmark::elapsed(repeats, || { + std::hint::black_box(gzip::decompress(&encoded).unwrap()); + }); + let oracle_time = benchmark::elapsed(repeats, || { + std::hint::black_box(oracle_decompress(&encoded)); + }); + assert!(fastbz2_time.as_secs_f64() <= oracle_time.as_secs_f64() * limit, "fastbz2 gzip {fastbz2_time:?} exceeded {limit}x oracle {oracle_time:?}"); +} + +#[test] +#[cfg(not(debug_assertions))] +fn gzip_performance_regression_stays_bounded() { + let plain = patterned(8 * 1024 * 1024); + assert_performance(&plain, 1.3); +} + +#[test] +#[cfg(not(debug_assertions))] +fn gzip_literal_performance_regression_stays_bounded() { + let plain = random_bytes(4 * 1024 * 1024); + assert_performance(&plain, 1.3); +} diff --git a/tests/test_install.py b/tests/test_install.py index 0484b63..ef04c6b 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -1,4 +1,4 @@ -import os, subprocess, sysconfig +import gzip, os, subprocess, sysconfig from pathlib import Path import fastbz2 @@ -9,3 +9,11 @@ def test_pip_installs_native_cli(): assert not executable.read_bytes().startswith(b"#!") result = subprocess.run([executable, "--version"], check=True, capture_output=True, text=True) assert result.stdout.strip() == f"fastbz2 {fastbz2.__version__}" + +def test_pip_installed_cli_decodes_gzip(tmp_path): + executable = Path(sysconfig.get_path("scripts")) / ("fastbz2.exe" if os.name == "nt" else "fastbz2") + plain = b"wheel-installed gzip decoder" * 10_000 + source = tmp_path / "sample.gz" + source.write_bytes(gzip.compress(plain, mtime=0)) + subprocess.run([executable, source], check=True) + assert source.with_suffix("").read_bytes() == plain diff --git a/tests/wiki_perf.rs b/tests/wiki_perf.rs index 7ddb44e..9c924f7 100644 --- a/tests/wiki_perf.rs +++ b/tests/wiki_perf.rs @@ -1,7 +1,11 @@ +mod common; + use std::{ + ffi::OsStr, fs, io::{self, Write}, path::Path, + process::{Command, Stdio}, time::{Duration, Instant}, }; @@ -34,6 +38,41 @@ fn requested_threads() -> usize { std::env::var("FASTBZ2_THREADS").ok().map(|value| value.parse().expect("FASTBZ2_THREADS must be an integer")).unwrap_or(0) } +fn physical_footprint_mib(metrics: &common::ProcessMetrics) -> f64 { + metrics.peak_phys_footprint_bytes.map_or(f64::NAN, |bytes| bytes as f64 / (1024.0 * 1024.0)) +} + +fn validation_command(binary: impl AsRef, path: &Path, threads: usize) -> Command { + let mut command = Command::new(binary); + command.arg("--test").arg("-P").arg(threads.to_string()).arg(path); + command +} + +fn quiet_validation_command(binary: impl AsRef, path: &Path, threads: usize) -> Command { + let mut command = validation_command(binary, path, threads); + command.stdout(Stdio::null()).stderr(Stdio::null()); + command +} + +fn warm_validation(binary: impl AsRef, path: &Path, threads: usize) { + assert!(quiet_validation_command(binary, path, threads).status().unwrap().success()); +} + +fn timed_validation(binary: impl AsRef, path: &Path, threads: usize) -> common::Timing { + common::measure_timing(&mut quiet_validation_command(binary, path, threads)).unwrap() +} + +fn print_process_metrics(label: &str, metrics: &common::ProcessMetrics) { + eprintln!( + "{label}: wall {:.3}s, CPU {:.3}s user + {:.3}s system, peak RSS {:.1} MiB, peak physical footprint {:.1} MiB", + metrics.wall.as_secs_f64(), + metrics.user.as_secs_f64(), + metrics.system.as_secs_f64(), + metrics.peak_rss_bytes as f64 / (1024.0 * 1024.0), + physical_footprint_mib(metrics), + ); +} + fn timed_fastbz2(path: &Path, threads: usize) -> Duration { let start = Instant::now(); let source = Source::open(path).unwrap(); @@ -74,6 +113,81 @@ fn simplewiki_full() { eprintln!("fastbz2 ({} threads): {elapsed:.3?}", options.resolved_threads()); } +#[test] +#[cfg(unix)] +#[ignore = "local full SimpleWiki subprocess time and peak-memory benchmark"] +fn simplewiki_cli_process_metrics() { + let path = corpus_path("simplewiki-full.xml.bz2"); + let mut command = validation_command(env!("CARGO_BIN_EXE_fastbz2"), &path, requested_threads()); + let metrics = common::measure(&mut command).unwrap(); + assert!(metrics.status.success()); + print_process_metrics("fastbz2 bzip2", &metrics); +} + +#[test] +#[cfg(unix)] +#[ignore = "local gzip subprocess time and peak-memory benchmark"] +fn gzip_cli_process_metrics() { + let path = corpus_path("simplewiki-full.xml.gz"); + let threads = requested_threads(); + let mut command = validation_command(env!("CARGO_BIN_EXE_fastbz2"), &path, threads); + let metrics = common::measure(&mut command).unwrap(); + assert!(metrics.status.success()); + let threads = if threads == 0 { "auto".to_owned() } else { threads.to_string() }; + print_process_metrics(&format!("fastbz2 gzip ({threads} threads)"), &metrics); +} + +#[test] +#[cfg(unix)] +#[ignore = "local system gzip subprocess time and peak-memory benchmark"] +fn system_gzip_process_metrics() { + let path = corpus_path("simplewiki-full.xml.gz"); + let mut command = Command::new("gzip"); + command.arg("-dc").arg(&path).stdout(Stdio::null()); + let metrics = common::measure(&mut command).unwrap(); + assert!(metrics.status.success()); + print_process_metrics("system gzip", &metrics); +} + +fn rapidgzip_binary() -> std::path::PathBuf { + let default = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../git/rapidgzip-rust/target/release/rapidgzip-rust"); + std::env::var_os("RAPIDGZIP_BIN").map(Into::into).unwrap_or(default) +} + +#[test] +#[cfg(unix)] +#[ignore = "local rapidgzip-rust subprocess time and peak-memory benchmark"] +fn rapidgzip_rust_process_metrics() { + let path = corpus_path("simplewiki-full.xml.gz"); + let mut command = validation_command(rapidgzip_binary(), &path, requested_threads()); + let metrics = common::measure(&mut command).unwrap(); + assert!(metrics.status.success()); + print_process_metrics("rapidgzip-rust", &metrics); +} + +#[test] +#[cfg(unix)] +#[ignore = "local single-run gzip performance ratio against rapidgzip-rust"] +fn gzip_reference_ratio() { + let path = corpus_path("simplewiki-full.xml.gz"); + let warm_path = corpus_path("simplewiki-first-5pct.xml.gz"); + let threads = requested_threads(); + let ours_binary = env!("CARGO_BIN_EXE_fastbz2"); + let reference_binary = rapidgzip_binary(); + + warm_validation(ours_binary, &warm_path, threads); + warm_validation(&reference_binary, &warm_path, threads); + + let ours = timed_validation(ours_binary, &path, threads); + let reference = timed_validation(&reference_binary, &path, threads); + assert!(ours.status.success()); + assert!(reference.status.success()); + + let ratio = ours.wall.as_secs_f64() / reference.wall.as_secs_f64(); + eprintln!("fastbz2 {:.3}s / rapidgzip-rust {:.3}s = {ratio:.3}x", ours.wall.as_secs_f64(), reference.wall.as_secs_f64()); + assert!(ratio <= 1.2, "fastbz2 must remain within 20% of rapidgzip-rust; measured {ratio:.3}x"); +} + fn timed_vec(name: &str, decode: impl FnOnce() -> Vec) { let start = Instant::now(); let decoded = decode();