From 815fc4a81611fff6b1f5487e63225cbfe4942ce9 Mon Sep 17 00:00:00 2001 From: Jeremy Howard Date: Tue, 25 Aug 2026 11:09:18 +1000 Subject: [PATCH] Add streaming tar extraction and faster parallel gzip --- Cargo.toml | 1 + DEV.md | 39 ++++++- README.md | 41 ++++--- src/bin/fastbz2.rs | 153 +++++++++++++++++++----- src/bin/fastbz2/tar_extract.rs | 204 ++++++++++++++++++++++++++++++++ src/decode.rs | 39 ++++--- src/decoder.rs | 16 +-- src/gzip.rs | 63 ++++++---- src/lib.rs | 4 +- src/output.rs | 53 +++++++++ tests/archive_perf.rs | 207 +++++++++++++++++++++++++++++++++ tests/cli.rs | 194 +++++++++++++++++++++++++++++- tests/wiki_perf.rs | 14 +++ 13 files changed, 921 insertions(+), 107 deletions(-) create mode 100644 src/bin/fastbz2/tar_extract.rs create mode 100644 src/output.rs create mode 100644 tests/archive_perf.rs diff --git a/Cargo.toml b/Cargo.toml index 7d604c3..ea86a59 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,6 +31,7 @@ pyo3 = { version = ">=0.29.2", optional = true } rayon = "1.12.0" serde_json = "1.0.151" tempfile = "3.27.0" +tar = { version = "0.4.46", default-features = false } [dev-dependencies] crabz2 = { version = "0.4.0", features = ["parallel"] } diff --git a/DEV.md b/DEV.md index 3a83adf..edca9f5 100644 --- a/DEV.md +++ b/DEV.md @@ -12,11 +12,13 @@ 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/output.rs owned/borrowed decoded-output sink abstraction 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 src/source.rs owned and memory-mapped compressed sources +src/bin/fastbz2/tar_extract.rs bounded decode-to-tar bridge, staging, and commit python/fastbz2/ thin Python I/O wrapper over fastbz2._core build_backend.py stage the native CLI for PEP 517 wheel builds tests/corpus/ selected upstream conformance and corruption fixtures @@ -26,11 +28,13 @@ tools/stage_binaries.py copy the release executable into Maturin wheel data 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 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. Primary jobs compute the CRC of each known clean suffix before ordered resolution. Resolution workers resolve the marker prefix, hash that prefix, and combine the two CRCs without rescanning the clean bytes. A marker-free history switches the same decoder to byte output. 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. -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. +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. An `OutputSink` wrapper enforces output-size limits, so each decoder has one code path for files, stdout, validation, listing, and tar extraction. + +Tar format semantics use the mature `tar` crate, pinned from 0.4.46 and built without its optional xattr feature. It handles streaming GNU/PAX/long-name/link entries and confines extracted paths to the destination. A zero-capacity rendezvous channel transfers each owned decoder chunk and its live suffix offset to `tar::Archive`. The channel queues no chunks and applies backpressure. `tar::Archive` pulls data through `Read`, which copies once from the current chunk into its request buffer. Extraction writes immediately into a same-filesystem temporary directory, drains all trailing tar padding so codec validation completes, then preflights every destination conflict and moves entries into place with renames. Multiple inputs remain sequential so their per-codec worker pools cannot oversubscribe the global thread budget. 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. @@ -53,10 +57,36 @@ 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 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 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. CLI tests generate tar archives and cover gzip/bzip2 wrappers, compound-extension dispatch, long names, stdin, raw-tar output, output limits, overwrite preflight, late checksum failure, and traversal confinement. 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. +### Local archive extraction benchmarks + +`tests/archive_perf.rs` measures the tar layer on the real `meta/simplewiki-first-5pct.xml.bz2` corpus. Fixture decoding and gzip/bzip2 recompression finish before timing. Each ignored test warms one target and measures it once. Run only the implementation changed: + +```bash +cargo test --release --test archive_perf tgz_fastbz2_overhead -- --ignored --exact --nocapture +cargo test --release --test archive_perf tgz_system_reference -- --ignored --exact --nocapture +cargo test --release --test archive_perf tbz2_fastbz2_overhead -- --ignored --exact --nocapture +cargo test --release --test archive_perf tbz2_system_reference -- --ignored --exact --nocapture +cargo test --release --test archive_perf tar_crate_reference -- --ignored --exact --nocapture +FASTBZ2_THREADS=18 cargo test --release --test archive_perf tgz_output_cadence -- --ignored --exact --nocapture +``` + +These are single runs after owned-suffix transfer and the 512 KiB gzip grid change: + +| Format | Raw decode | fastbz2 extraction | Extraction/raw | System `tar` | Extraction/system | +|---|---:|---:|---:|---:|---:| +| `.tgz` | 39.919 ms | 56.850 ms | 1.424x | 117.962 ms | 0.482x | +| `.tar.bz2` | 148.411 ms | 151.783 ms | 1.023x | 1.168 s | 0.130x | + +Direct extraction of the uncompressed in-memory tar through the `tar` crate took 32.069 ms. Raw gzip decode plus direct tar extraction totals 71.988 ms. The combined pipeline takes 56.850 ms and hides 15.138 ms, or 47%, of the direct tar work. + +The cadence benchmark identified ordered gzip output as the main overlap limit. With a 1 MiB speculative grid, output began at 8.491 ms, reached 25% at 29.595 ms, and completed at 33.724 ms. A 512 KiB grid began at 4.798 ms, reached 25% at 23.993 ms, and completed at 32.915 ms. A 256 KiB grid emitted earlier but slowed raw decode to 38.073 ms and extraction to 57.792 ms. The 512 KiB grid gave the best measured balance. Computing each clean suffix CRC in its primary job moved 25% output to 19.818 ms, 75% to 30.355 ms, and completion to 30.678 ms. The corresponding extraction run was effectively flat at 56.850 ms. Tar cannot process later bytes while an earlier ordered gzip segment remains incomplete. A custom tar parser would not remove that dependency. A one-chunk channel buffer regressed extraction to 59.698 ms, so the bridge retains its zero-capacity rendezvous. + +System `tar` remains the external reference and 1.2x remains the research target. Raw-tar output is a lower bound rather than an extractor reference. The fastbz2 tests use a broad 3x raw-decode regression guard. Keep the measurements single-run; change an implementation before rerunning it. + 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 @@ -75,9 +105,10 @@ The full bzip2 confirmation streams to a counting sink and validates every block cargo test --release --test wiki_perf simplewiki_full -- --ignored --exact --nocapture ``` -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: +The fastbz2-only gzip test warms with `meta/simplewiki-first-5pct.xml.gz` and performs one full-dump validation. The ratio test warms both executables, measures each full dump once, and fails above 1.2x the sibling rapidgzip-rust checkout: ```bash +cargo test --release --test wiki_perf gzip_fastbz2_validation -- --ignored --exact --nocapture cargo test --release --test wiki_perf gzip_reference_ratio -- --ignored --exact --nocapture ``` diff --git a/README.md b/README.md index 9beb438..1daf019 100644 --- a/README.md +++ b/README.md @@ -1,10 +1,10 @@ # fastbz2 -An active compression-format research workbench with fast bzip2 and gzip decompression. +An active compression-format research workbench with fast bzip2/gzip decompression and streaming tar extraction. -`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. +`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, and streams compressed tar variants through a bounded extractor. Both decoders handle concatenated streams and fully validate their checksums. The bzip2 implementation also provides parallel decoding and persistent random-access indexes. -Compression and additional formats are planned, but the current implementation decompresses bzip2 and gzip. +Compression is planned, but the current implementation decompresses bzip2 and gzip and extracts their tar-wrapped variants. ## Install @@ -25,24 +25,28 @@ cargo add fastbz2 --git https://github.com/AnswerDotAI/fastbz2 ## CLI -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`. +Decoding is the default operation. `.bz2`, `.bzip2`, `.gz`, and `.gzip` select their corresponding decoder and are removed from the output name. Compressed tar names—`.tar.bz2`, `.tar.bzip2`, `.tbz`, `.tbz2`, `.tar.gz`, `.tar.gzip`, and `.tgz`—automatically extract into the current directory or `-C/--output-dir`. `-x/--extract` forces tar extraction for stdin or an unusual filename; an explicit `-o/--output` instead writes the decoded tar stream. For stdin and unrecognised extensions, bzip2 or gzip magic selects the decoder. Other non-archive 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 +fastbz2 dump.xml.bz2 # write dump.xml +fastbz2 events.json.gz # write events.json +fastbz2 source.tar.gz # extract into the current directory +fastbz2 source.tbz2 -C unpacked # extract into unpacked/ +fastbz2 --extract -C unpacked - # extract gzip/bzip2 tar data from stdin +fastbz2 source.tgz -o source.tar # decode without extracting +fastbz2 dump.xml.bz2 -o result.xml # choose the decoded output path +fastbz2 dump.xml.bz2 -o - # write decoded bytes to stdout ``` -Multiple inputs are decoded in order, with parallelism applied inside each file. `-C/--output-dir` collects their outputs in one directory: +Multiple inputs are processed in order, with parallelism applied inside each compressed stream. `-C/--output-dir` collects decoded files and is the extraction root for archives: ```bash fastbz2 data/*.bz2 logs/*.gz -C decoded fastbz2 data/*.bz2 logs/*.gz -C decoded --skip-existing +fastbz2 backups/*.tgz -C restored ``` -The alternative modes are flags rather than subcommands: +Validation and inspection remain flags rather than subcommands: ```bash fastbz2 --test dump.xml.bz2 # fully decode and validate, writing nothing @@ -51,15 +55,16 @@ fastbz2 --list events.json.gz # print the validated member/block layout fastbz2 --list --json dump.xml.bz2 # emit the complete layout as JSON ``` -`--test`, `--index`, and `--list` are mutually exclusive. Human-readable `--list` output labels each input when given multiple files; JSON output is one object for one input and an array for multiple inputs. +`--test`, `--index`, `--list`, and explicit `--extract` are mutually exclusive. Human-readable `--list` output labels each input when given multiple files; JSON output is one object for one input and an array for multiple inputs. ### Output safety -- Existing outputs are rejected by default. Use `--force` to replace them or `--skip-existing` to leave them untouched. -- File outputs are written to a temporary file in the destination directory and persisted atomically only after successful CRC validation. -- Extracted files inherit the compressed input's permissions and modification time. -- `--rm` removes each compressed input only after its output has been persisted and its metadata copied successfully. -- `--max-output SIZE` limits decoded bytes per input. Sizes accept binary suffixes such as `K`, `MiB`, and `G`. +- Existing decoded files and archive entries are rejected by default. `--force` replaces them; `--skip-existing` applies to decoded files rather than archives. +- Decoded-file outputs use a same-directory temporary file and become visible atomically only after successful checksum validation. +- Tar entries stream into a same-filesystem staging directory through a bounded pipe. They are preflighted and moved into the destination only after both the compression stream and tar archive validate, so a late CRC failure leaves no extracted files. +- Tar paths and link targets are confined to the destination; unsafe entries are skipped. New entries use the archive's permissions and modification times. Standalone decoded files inherit those values from the compressed input. +- `--rm` removes each compressed input only after its decoded file or all archive entries have been committed successfully. +- `--max-output SIZE` limits decoded bytes per input, including tar framing and padding. Sizes accept binary suffixes such as `K`, `MiB`, and `G`. 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. @@ -157,7 +162,7 @@ Full SimpleWiki recompressed with system `gzip -6` (`438,904,466` bytes compress | 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 | +| fastbz2 | auto parallel, validation sink | 0.326 | 325 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. diff --git a/src/bin/fastbz2.rs b/src/bin/fastbz2.rs index a74c12d..fa5d25b 100644 --- a/src/bin/fastbz2.rs +++ b/src/bin/fastbz2.rs @@ -1,3 +1,6 @@ +#[path = "fastbz2/tar_extract.rs"] +mod tar_extract; + use std::{ fs, io::{self, IsTerminal, Read, Write}, @@ -8,7 +11,8 @@ 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, gzip, + DecodeOptions, DecodeProgress, Error, Index, OutputSink, Source, WriterSink, build_index_with_progress, decode_to_writer_with_progress, + decompress_to_sink_with_progress, gzip, }; use serde_json::{Value, json}; use tempfile::NamedTempFile; @@ -16,50 +20,60 @@ use tempfile::NamedTempFile; #[derive(Parser)] #[command( version, - about = "Fast compression-format research workbench", - group(ArgGroup::new("mode").args(["test", "index", "list"])) + about = "Parallel bzip2/gzip decompression and streaming tar extraction", + long_about = "Parallel bzip2 and gzip decompression with streaming tar extraction. Decoding is the default operation. Recognised codec suffixes are removed from normal output names. Compressed tar suffixes extract automatically unless -o is given.", + after_help = r#"Examples: + fastbz2 dump.xml.bz2 Write dump.xml + fastbz2 events.json.gz -o - Write decoded bytes to stdout + fastbz2 backup.tgz -C restored Extract into restored/ + fastbz2 --test archive.tar.bz2 Validate without writing output + fastbz2 --list --json data.gz Show the validated layout as JSON"#, + group(ArgGroup::new("mode").args(["test", "index", "list", "extract"])) )] struct Cli { - /// Input files, or - for stdin when decoding or testing. + /// Input files, or - for stdin except with --index. #[arg(required = true, num_args = 1..)] inputs: Vec, - /// Fully decode and validate without writing plaintext. + /// Fully decode and validate checksums without writing output. #[arg(long)] test: bool, - /// Build validated, source-bound bzip2 block indexes. + /// Build validated, source-bound .fbz2i indexes for bzip2 inputs. #[arg(long)] index: bool, - /// Validate and show stream/block layouts. + /// Validate and show bzip2 streams/blocks or gzip members/blocks. #[arg(long)] list: bool, - /// Output path, or - for stdout; requires one input. - #[arg(short, long, conflicts_with_all = ["test", "list", "output_dir"])] + /// Extract a decoded tar stream; automatic for recognised compressed-tar suffixes. + #[arg(short = 'x', long)] + extract: bool, + /// Write decoded bytes to PATH, or - for stdout; requires one input and disables automatic extraction. + #[arg(short, long, conflicts_with_all = ["test", "list", "extract", "output_dir"])] output: Option, - /// Put decoded files in DIRECTORY. + /// Put decoded files or extracted archive entries in DIRECTORY. #[arg(short = 'C', long = "output-dir", conflicts_with_all = ["test", "index", "list", "output"])] output_dir: Option, - /// Bzip2 worker threads; 0 uses all available CPUs. + /// Decoder worker threads; 0 uses all available CPUs. #[arg(short = 'P', long, default_value_t = 0)] threads: usize, - /// Maximum speculative bzip2 output. + /// Maximum speculative decoder output; accepts binary size suffixes. #[arg(long, default_value = "1G", value_parser = parse_size)] memory_limit: usize, - /// Refuse to decode more than SIZE bytes per input. + /// Maximum decoded bytes per input; accepts binary size suffixes. #[arg(long, value_parser = parse_size)] max_output: Option, - /// Replace existing output files. + /// Replace existing output files or archive entries. #[arg(short, long, conflicts_with_all = ["test", "list", "skip_existing"])] force: bool, - /// Skip existing output files. - #[arg(long, conflicts_with_all = ["test", "list", "force"])] + /// Skip existing decoded output files. + #[arg(long, conflicts_with_all = ["test", "list", "extract", "force"])] skip_existing: bool, - /// Remove compressed inputs after successful extraction. + /// Remove compressed inputs after successful decoding or extraction. #[arg(long = "rm", conflicts_with_all = ["test", "index", "list"])] remove_input: bool, /// Suppress progress and skip notices. #[arg(short, long)] quiet: bool, - /// Emit list output as JSON. + /// Emit `--list` output as JSON. #[arg(long, requires = "list")] json: bool, } @@ -108,14 +122,30 @@ fn validate_cli(cli: &Cli) -> fastbz2::Result<()> { if (cli.index || cli.list) && cli.inputs.iter().any(|input| input == "-") { return Err(invalid("stdin is supported only for decoding and --test")); } + if cli.skip_existing && cli.inputs.iter().any(|input| should_extract(cli, input)) { + return Err(invalid("--skip-existing is not supported when extracting archives")); + } Ok(()) } +fn should_extract(cli: &Cli, input: &str) -> bool { + cli.extract || (cli.output.is_none() && is_tar_archive(input)) +} + fn run_decode(cli: &Cli, options: DecodeOptions) -> fastbz2::Result<()> { if let Some(directory) = &cli.output_dir { fs::create_dir_all(directory)?; } for input in &cli.inputs { + let extract = should_extract(cli, input); + if extract { + let destination = cli.output_dir.as_deref().unwrap_or_else(|| Path::new(".")); + extract_input(input, destination, cli.force, options, cli.max_output, cli.quiet)?; + if cli.remove_input && input != "-" { + fs::remove_file(input)?; + } + continue; + } if input == "-" || cli.output.as_deref() == Some(Path::new("-")) { decode_input(input, &mut io::stdout().lock(), options, cli.max_output, cli.quiet)?; if cli.remove_input && input != "-" { @@ -194,6 +224,29 @@ fn run_list(cli: &Cli, options: DecodeOptions) -> fastbz2::Result<()> { Ok(()) } +fn extract_input(input: &str, destination: &Path, overwrite: bool, options: DecodeOptions, max_output: Option, quiet: bool) -> fastbz2::Result<()> { + if input == "-" { + let mut data = Vec::new(); + io::stdin().lock().read_to_end(&mut data)?; + extract_data(&data, "stdin", destination, overwrite, options, max_output, quiet) + } else { + let source = Source::open(input)?; + extract_data(source.as_slice(), input, destination, overwrite, options, max_output, quiet) + } +} + +fn extract_data( + data: &[u8], + label: &str, + destination: &Path, + overwrite: bool, + options: DecodeOptions, + max_output: Option, + quiet: bool, +) -> fastbz2::Result<()> { + tar_extract::unpack(destination, overwrite, |writer| decode_data_to_sink(data, label, writer, options, max_output, quiet)) +} + fn decode_input(input: &str, output: &mut impl Write, options: DecodeOptions, max_output: Option, quiet: bool) -> fastbz2::Result<()> { if input == "-" { let mut data = Vec::new(); @@ -206,16 +259,28 @@ 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 output = WriterSink::new(output); + decode_data_to_sink(data, label, &mut output, options, max_output, quiet) +} + +fn decode_data_to_sink( + data: &[u8], + label: &str, + output: &mut impl OutputSink, + options: DecodeOptions, + max_output: Option, + quiet: bool, +) -> fastbz2::Result<()> { + let mut output = LimitedOutput::new(output, max_output); let mut display = ProgressDisplay::new(label, data.len() as u64, quiet); 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(|_| ()), + Format::Bzip2 => decompress_to_sink_with_progress(data, &mut output, options, |progress| display.update(progress)), + Format::Gzip => gzip::decompress_to_sink_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 sink = LimitedOutput::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)) } @@ -223,30 +288,35 @@ fn build_gzip_report_data(data: &[u8], label: &str, options: DecodeOptions, max_ 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); if let Some(limit) = max_output { - let mut sink = LimitedWriter::new(io::sink(), Some(limit)); + let mut sink = LimitedOutput::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)) } } -struct LimitedWriter { +struct LimitedOutput { inner: W, written: usize, limit: Option, } -impl LimitedWriter { +impl LimitedOutput { fn new(inner: W, limit: Option) -> Self { Self { inner, written: 0, limit } } -} -impl Write for LimitedWriter { - fn write(&mut self, buffer: &[u8]) -> io::Result { - if self.limit.is_some_and(|limit| self.written.saturating_add(buffer.len()) > limit) { + fn check(&self, count: usize) -> io::Result<()> { + if self.limit.is_some_and(|limit| self.written.saturating_add(count) > limit) { return Err(io::Error::new(io::ErrorKind::InvalidData, format!("decoded output exceeds {}", format_bytes(self.limit.unwrap() as u64)))); } + Ok(()) + } +} + +impl Write for LimitedOutput { + fn write(&mut self, buffer: &[u8]) -> io::Result { + self.check(buffer.len())?; let written = self.inner.write(buffer)?; self.written += written; Ok(written) @@ -256,6 +326,26 @@ impl Write for LimitedWriter { self.inner.flush() } } +impl OutputSink for LimitedOutput { + fn write_borrowed(&mut self, buffer: &[u8]) -> io::Result<()> { + self.check(buffer.len())?; + self.inner.write_borrowed(buffer)?; + self.written += buffer.len(); + Ok(()) + } + + fn write_owned_from(&mut self, buffer: Vec, start: usize) -> io::Result<()> { + let length = buffer.get(start..).ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "owned chunk start exceeds its length"))?.len(); + self.check(length)?; + self.inner.write_owned_from(buffer, start)?; + self.written += length; + Ok(()) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} struct ProgressDisplay { stderr: io::Stderr, @@ -369,6 +459,11 @@ fn format_extension(input: &Path) -> Option<(Format, &'static str)> { } } +fn is_tar_archive(input: &str) -> bool { + let input = input.to_ascii_lowercase(); + [".tar.bz2", ".tar.bzip2", ".tbz", ".tbz2", ".tar.gz", ".tar.gzip", ".tgz"].iter().any(|extension| input.ends_with(extension)) +} + fn default_output(input: &Path) -> PathBuf { format_extension(input).map_or_else(|| PathBuf::from(format!("{}.out", input.display())), |(_, extension)| input.with_extension(extension)) } diff --git a/src/bin/fastbz2/tar_extract.rs b/src/bin/fastbz2/tar_extract.rs new file mode 100644 index 0000000..040ce6b --- /dev/null +++ b/src/bin/fastbz2/tar_extract.rs @@ -0,0 +1,204 @@ +use std::{ + cmp, fs, + io::{self, Read}, + path::{Path, PathBuf}, + sync::mpsc::{Receiver, SyncSender, sync_channel}, + thread, +}; + +use fastbz2::{Error, OutputSink, Result}; +use tempfile::TempDir; + +struct Chunk { + bytes: Vec, + offset: usize, +} + +pub(super) struct PipeWriter { + sender: SyncSender, +} + +impl PipeWriter { + fn send(&self, bytes: Vec, offset: usize) -> io::Result<()> { + let suffix = bytes.get(offset..).ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "owned chunk start exceeds its length"))?; + if suffix.is_empty() { + return Ok(()); + } + self.sender.send(Chunk { bytes, offset }).map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "tar extractor stopped reading")) + } +} + +impl OutputSink for PipeWriter { + fn write_borrowed(&mut self, buffer: &[u8]) -> io::Result<()> { + self.send(buffer.to_vec(), 0) + } + + fn write_owned_from(&mut self, buffer: Vec, start: usize) -> io::Result<()> { + self.send(buffer, start) + } +} + +struct PipeReader { + receiver: Receiver, + chunk: Vec, + offset: usize, +} + +impl Read for PipeReader { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + if buffer.is_empty() { + return Ok(0); + } + if self.offset == self.chunk.len() { + match self.receiver.recv() { + Ok(chunk) => { + self.chunk = chunk.bytes; + self.offset = chunk.offset; + } + Err(_) => return Ok(0), + } + } + let count = cmp::min(buffer.len(), self.chunk.len() - self.offset); + buffer[..count].copy_from_slice(&self.chunk[self.offset..self.offset + count]); + self.offset += count; + Ok(count) + } +} + +fn pipe() -> (PipeWriter, PipeReader) { + let (sender, receiver) = sync_channel(0); + (PipeWriter { sender }, PipeReader { receiver, chunk: Vec::new(), offset: 0 }) +} + +fn broken_pipe(error: &Error) -> bool { + matches!(error, Error::Io(source) if source.kind() == io::ErrorKind::BrokenPipe) +} + +fn path_metadata(path: &Path) -> io::Result> { + match fs::symlink_metadata(path) { + Ok(metadata) => Ok(Some(metadata)), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error), + } +} + +fn children(path: &Path) -> io::Result> { + let mut result = fs::read_dir(path)?.map(|entry| entry.map(|entry| entry.path())).collect::>>()?; + result.sort_unstable(); + Ok(result) +} + +fn existing_error(path: &Path) -> Error { + Error::Io(io::Error::new(io::ErrorKind::AlreadyExists, format!("{} already exists (use --force)", path.display()))) +} +fn directory_collision(path: &Path) -> Error { + Error::Io(io::Error::new(io::ErrorKind::AlreadyExists, format!("refusing to replace directory {} with an archive entry", path.display()))) +} + +fn preflight(source: &Path, target: &Path, overwrite: bool) -> Result<()> { + let Some(target_metadata) = path_metadata(target)? else { + return Ok(()); + }; + let source_metadata = fs::symlink_metadata(source)?; + if source_metadata.is_dir() && target_metadata.is_dir() { + for child in children(source)? { + preflight(&child, &target.join(child.file_name().unwrap()), overwrite)?; + } + return Ok(()); + } + if target_metadata.is_dir() { + return Err(directory_collision(target)); + } + if overwrite { Ok(()) } else { Err(existing_error(target)) } +} + +#[cfg(unix)] +fn make_directory_mutable(path: &Path, metadata: &fs::Metadata) -> io::Result<()> { + use std::os::unix::fs::PermissionsExt; + fs::set_permissions(path, fs::Permissions::from_mode(metadata.permissions().mode() | 0o700)) +} + +#[cfg(not(unix))] +fn make_directory_mutable(path: &Path, metadata: &fs::Metadata) -> io::Result<()> { + let mut permissions = metadata.permissions(); + permissions.set_readonly(false); + fs::set_permissions(path, permissions) +} + +fn commit_entry(source: &Path, target: &Path, overwrite: bool) -> Result<()> { + let source_metadata = fs::symlink_metadata(source)?; + let Some(target_metadata) = path_metadata(target)? else { + fs::rename(source, target)?; + return Ok(()); + }; + if source_metadata.is_dir() && target_metadata.is_dir() { + make_directory_mutable(source, &source_metadata)?; + for child in children(source)? { + commit_entry(&child, &target.join(child.file_name().unwrap()), overwrite)?; + } + fs::remove_dir(source)?; + return Ok(()); + } + if target_metadata.is_dir() { + return Err(directory_collision(target)); + } + if !overwrite { + return Err(existing_error(target)); + } + fs::remove_file(target)?; + fs::rename(source, target)?; + Ok(()) +} + +fn staging(destination: &Path) -> Result { + let parent = match path_metadata(destination)? { + Some(metadata) if metadata.is_dir() => destination, + Some(_) => { + return Err(Error::Io(io::Error::new(io::ErrorKind::NotADirectory, format!("{} is not a directory", destination.display())))); + } + None => destination.parent().filter(|path| !path.as_os_str().is_empty()).unwrap_or_else(|| Path::new(".")), + }; + fs::create_dir_all(parent)?; + TempDir::new_in(parent).map_err(Error::from) +} + +fn commit(staging: &Path, destination: &Path, overwrite: bool) -> Result<()> { + if path_metadata(destination)?.is_none() { + fs::create_dir(destination)?; + } + let entries = children(staging)?; + for source in &entries { + preflight(source, &destination.join(source.file_name().unwrap()), overwrite)?; + } + for source in entries { + commit_entry(&source, &destination.join(source.file_name().unwrap()), overwrite)?; + } + Ok(()) +} + +pub(super) fn unpack(destination: &Path, overwrite: bool, decode: F) -> Result<()> +where + F: FnOnce(&mut PipeWriter) -> Result<()> + Send, +{ + let staging = staging(destination)?; + thread::scope(|scope| { + let (mut writer, mut reader) = pipe(); + let decoder = scope.spawn(move || decode(&mut writer)); + let extracted = { + let mut archive = tar::Archive::new(&mut reader); + archive.set_overwrite(true); + archive.unpack(staging.path()).map_err(Error::from) + }; + + let drained = if extracted.is_ok() { io::copy(&mut reader, &mut io::sink()).map(|_| ()).map_err(Error::from) } else { Ok(()) }; + drop(reader); + + let decoded = decoder.join().map_err(|_| Error::InvalidConfiguration("decoder worker panicked while extracting tar".into()))?; + match (decoded, extracted, drained) { + (Ok(()), Ok(()), Ok(())) => commit(staging.path(), destination, overwrite), + (Err(error), Err(archive_error), _) if broken_pipe(&error) => Err(archive_error), + (Err(error), _, _) => Err(error), + (Ok(()), Err(error), _) | (Ok(()), Ok(()), Err(error)) => Err(error), + } + }) +} diff --git a/src/decode.rs b/src/decode.rs index 8cfee23..0c4fa01 100644 --- a/src/decode.rs +++ b/src/decode.rs @@ -4,7 +4,10 @@ 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}; +use crate::{ + BlockCandidate, BlockIndex, DecodeError, EndCandidate, Error, Index, MAX_DECODED_BLOCK, OutputSink, Result, StreamIndex, WriterSink, combine_stream_crc, + decoder, +}; pub const DEFAULT_MEMORY_LIMIT: usize = 1024 * 1024 * 1024; @@ -57,12 +60,6 @@ impl Marker { } pub fn decompress(data: &[u8], options: DecodeOptions) -> Result> { - let options = options.validate()?; - if options.resolved_threads() == 1 { - let mut output = Vec::new(); - decoder::decode_serial(data, &mut output)?; - return Ok(output); - } let mut output = Vec::new(); decompress_to_writer(data, &mut output, options)?; Ok(output) @@ -74,9 +71,16 @@ pub fn decompress_to_writer(data: &[u8], output: &mut impl Write, options: Decod decompress_to_writer_with_progress(data, output, options, |_| {}) } -pub fn decompress_to_writer_with_progress( +pub fn decompress_to_writer_with_progress(data: &[u8], output: &mut impl Write, options: DecodeOptions, progress: impl FnMut(DecodeProgress)) -> Result<()> { + let mut output = WriterSink::new(output); + decompress_to_sink_with_progress(data, &mut output, options, progress) +} + +/// Decode into an output that can take ownership of completed chunks. +#[doc(hidden)] +pub fn decompress_to_sink_with_progress( data: &[u8], - output: &mut impl Write, + output: &mut impl OutputSink, options: DecodeOptions, mut progress: impl FnMut(DecodeProgress), ) -> Result<()> { @@ -86,7 +90,7 @@ pub fn decompress_to_writer_with_progress( progress(DecodeProgress { compressed_bytes, decoded_bytes }); }); } - decode_to_writer_impl(data, output, options, &mut progress).map(|_| ()) + decode_to_sink_impl(data, output, options, &mut progress).map(|_| ()) } pub fn build_index(data: &[u8], options: DecodeOptions) -> Result { @@ -94,20 +98,23 @@ pub fn build_index(data: &[u8], options: DecodeOptions) -> Result { } pub fn build_index_with_progress(data: &[u8], options: DecodeOptions, mut progress: impl FnMut(DecodeProgress)) -> Result { - decode_to_writer_impl(data, &mut std::io::sink(), options, &mut progress) + let mut output = WriterSink::new(std::io::sink()); + decode_to_sink_impl(data, &mut output, options, &mut progress) } /// Decode a complete bzip2 input, validate every CRC, and build its block index. pub fn decode_to_writer(data: &[u8], output: &mut impl Write, options: DecodeOptions) -> Result { - decode_to_writer_impl(data, output, options, &mut |_| {}) + let mut output = WriterSink::new(output); + decode_to_sink_impl(data, &mut output, options, &mut |_| {}) } /// Decode a complete bzip2 input, return its index, and report completed work. pub fn decode_to_writer_with_progress(data: &[u8], output: &mut impl Write, options: DecodeOptions, mut progress: impl FnMut(DecodeProgress)) -> Result { - decode_to_writer_impl(data, output, options, &mut progress) + let mut output = WriterSink::new(output); + decode_to_sink_impl(data, &mut output, options, &mut progress) } -fn decode_to_writer_impl(data: &[u8], output: &mut impl Write, options: DecodeOptions, progress: &mut impl FnMut(DecodeProgress)) -> Result { +fn decode_to_sink_impl(data: &[u8], output: &mut impl OutputSink, options: DecodeOptions, progress: &mut impl FnMut(DecodeProgress)) -> Result { let options = options.validate()?; let threads = options.resolved_threads(); let pool = thread_pool(threads)?; @@ -152,7 +159,7 @@ fn decode_to_writer_impl(data: &[u8], output: &mut impl Write, options: DecodeOp fn assemble( data: &[u8], - output: &mut impl Write, + output: &mut impl OutputSink, markers: &[Marker], candidates: &mut impl Candidates, progress: &mut impl FnMut(DecodeProgress), @@ -197,8 +204,8 @@ fn assemble( } let end_index = marker_at(markers, decoded.end_bit)?; candidates.discard_before(end_index); - output.write_all(&decoded.output)?; let decoded_len = decoded.output.len() as u64; + output.write_owned_from(decoded.output, 0)?; blocks.push(BlockIndex { compressed_start_bit: block.bit_offset, compressed_end_bit: markers[end_index].bit_offset(), diff --git a/src/decoder.rs b/src/decoder.rs index 8c95a0a..2cc0e92 100644 --- a/src/decoder.rs +++ b/src/decoder.rs @@ -1,6 +1,4 @@ -use std::io::Write; - -use crate::{DecodeError, Error, Result, bz2_crc32, combine_stream_crc}; +use crate::{DecodeError, Error, OutputSink, Result, bz2_crc32, combine_stream_crc}; const BLOCK_MAGIC: u64 = 0x3141_5926_5359; const END_MAGIC: u64 = 0x1772_4538_5090; @@ -400,11 +398,7 @@ pub(crate) fn decode_block(data: &[u8], start_bit: u64, end_bit: u64, level: u8, Ok(output) } -pub(crate) fn decode_serial(data: &[u8], output: &mut impl Write) -> Result<()> { - decode_serial_with_progress(data, output, &mut |_, _| {}) -} - -pub(crate) fn decode_serial_with_progress(data: &[u8], output: &mut impl Write, progress: &mut impl FnMut(u64, u64)) -> Result<()> { +pub(crate) fn decode_serial_with_progress(data: &[u8], output: &mut impl OutputSink, progress: &mut impl FnMut(u64, u64)) -> Result<()> { let mut bits = Bits::at(data, 0)?; let mut decoder = Decoder::new(); let mut decoded_bytes = 0_u64; @@ -427,9 +421,9 @@ pub(crate) fn decode_serial_with_progress(data: &[u8], output: &mut impl Write, match bits.magic()? { BLOCK_MAGIC => { let (block, crc, _) = decoder.block(&mut bits, level, None)?; - output.write_all(&block)?; - decoded_bytes = - decoded_bytes.checked_add(block.len() as u64).ok_or_else(|| Error::InvalidConfiguration("decoded offset overflow".into()))?; + let block_len = block.len() as u64; + output.write_owned_from(block, 0)?; + decoded_bytes = decoded_bytes.checked_add(block_len).ok_or_else(|| Error::InvalidConfiguration("decoded offset overflow".into()))?; progress(bits.position().div_ceil(8), decoded_bytes); combined_crc = combine_stream_crc(combined_crc, crc); } diff --git a/src/gzip.rs b/src/gzip.rs index 0be7612..dabe263 100644 --- a/src/gzip.rs +++ b/src/gzip.rs @@ -3,7 +3,7 @@ use std::{io::Write, sync::OnceLock}; use crate::{ - DecodeOptions, DecodeProgress, Error, Result, + DecodeOptions, DecodeProgress, Error, OutputSink, Result, WriterSink, pipeline::{Job, PipelineLimits, run_staged_ordered}, }; @@ -11,8 +11,8 @@ 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_GRID: usize = 512 * 1024; +const MIN_PARALLEL_INPUT: usize = 16 * 1024 * 1024; const PARALLEL_OUTPUT_LIMIT: usize = 8 * 1024 * 1024; const PARALLEL_JOB_MEMORY: usize = 2 * PARALLEL_OUTPUT_LIMIT + 64 * 1024; @@ -111,24 +111,36 @@ pub fn decompress_to_writer_with_options_and_progress( data: &[u8], output: &mut impl Write, options: DecodeOptions, + progress: impl FnMut(DecodeProgress), +) -> Result { + let mut output = WriterSink::new(output); + decompress_to_sink_with_options_and_progress(data, &mut output, options, progress) +} + +/// Decode into an output that can take ownership of completed chunks. +#[doc(hidden)] +pub fn decompress_to_sink_with_options_and_progress( + data: &[u8], + output: &mut impl OutputSink, + 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); + return decompress_serial_to_sink_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), + Err(_) => return decompress_serial_to_sink_with_progress(data, output, progress), }; - decompress_parallel_to_writer(data, output, options, threads, &mut progress, initial_segment) + decompress_parallel_to_sink(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 { +fn decompress_serial_to_sink_with_progress(data: &[u8], output: &mut impl OutputSink, mut progress: impl FnMut(DecodeProgress)) -> Result { let mut members = Vec::new(); let mut blocks = Vec::new(); let mut position = 0_usize; @@ -306,6 +318,7 @@ struct Segment { marked: Vec, clean: Vec, clean_start: usize, + clean_crc: crc32fast::Hasher, blocks: Vec, final_block: bool, } @@ -335,12 +348,15 @@ fn decode_segment(data: &[u8], start_bit: usize, stop_bit: usize, history: Initi break false; } }; + let mut clean_crc = crc32fast::Hasher::new(); + clean_crc.update(&emitter.clean[emitter.clean_start..]); Ok(Segment { start_bit, end_bit: bits.position_bits(), marked: emitter.marked, clean: emitter.clean, clean_start: emitter.clean_start, + clean_crc, blocks, final_block, }) @@ -499,7 +515,7 @@ fn resolve_segment(segment: Segment, predecessor: &[u8]) -> Result { progress: &'a mut P, } -impl<'a, W: Write + ?Sized, P: FnMut(DecodeProgress) + ?Sized> SegmentCommitter<'a, W, P> { +impl<'a, W: OutputSink + ?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 { + fn commit(&mut self, segment: ResolvedSegment) -> Result<()> { + let ResolvedSegment { marked, clean, clean_start, mut blocks, compressed_end_bit, crc } = segment; + let decoded_len = marked.len() + clean.len() - clean_start; + self.output.write_owned_from(marked, 0)?; + self.output.write_owned_from(clean, clean_start)?; + self.crc.combine(&crc); + for block in &mut 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 }); + self.blocks.append(&mut blocks); + (self.progress)(DecodeProgress { compressed_bytes: compressed_end_bit.div_ceil(8) as u64, decoded_bytes: self.decoded_base + self.decoded }); Ok(()) } } -fn decompress_parallel_to_writer( +fn decompress_parallel_to_sink( data: &[u8], - output: &mut impl Write, + output: &mut impl OutputSink, options: DecodeOptions, threads: usize, progress: &mut impl FnMut(DecodeProgress), @@ -566,7 +583,7 @@ fn decompress_parallel_to_writer( 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| { + let mut suffix = decompress_serial_to_sink_with_progress(&data[position..], output, |item| { progress(DecodeProgress { compressed_bytes: compressed_base + item.compressed_bytes, decoded_bytes: decoded_base + item.decoded_bytes, @@ -1095,7 +1112,7 @@ struct Emitter<'a, W> { decoded_base: u64, } -impl<'a, W: Write> Emitter<'a, W> { +impl<'a, W: OutputSink> Emitter<'a, W> { fn new(output: &'a mut W, decoded_base: u64) -> Self { Self { output, @@ -1164,7 +1181,7 @@ impl<'a, W: Write> Emitter<'a, W> { if pending.is_empty() { return Ok(()); } - self.output.write_all(pending)?; + self.output.write_borrowed(pending)?; self.crc.update(pending); self.history_len = self.buffer.len(); if self.buffer.len() >= HISTORY_COMPACT { @@ -1183,7 +1200,7 @@ impl<'a, W: Write> Emitter<'a, W> { } } -impl DeflateOutput for Emitter<'_, W> { +impl DeflateOutput for Emitter<'_, W> { fn total_decoded(&self) -> u64 { self.decoded_position() } diff --git a/src/lib.rs b/src/lib.rs index bbee4b0..c1a0dd0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -10,6 +10,7 @@ mod format; pub mod gzip; mod index; mod indexed; +mod output; mod pipeline; mod source; @@ -18,12 +19,13 @@ pub use block::{MAX_DECODED_BLOCK, MAX_ENCODED_BLOCK, decode_block}; pub use crc::{bz2_crc32, combine_stream_crc}; pub use decode::{ DEFAULT_MEMORY_LIMIT, DecodeOptions, DecodeProgress, build_index, build_index_with_progress, decode_to_writer, decode_to_writer_with_progress, decompress, - decompress_to_writer, decompress_to_writer_with_progress, + decompress_to_sink_with_progress, decompress_to_writer, decompress_to_writer_with_progress, }; pub use error::{DecodeError, Error, Result}; pub use format::{BLOCK_MAGIC, BlockCandidate, END_MAGIC, EndCandidate, ScanResult, StreamHeaderCandidate, scan}; pub use index::{BlockIndex, Index, StreamIndex}; pub use indexed::{DEFAULT_CACHE_LIMIT, IndexedReader}; +pub use output::{OutputSink, WriterSink}; pub use source::Source; #[cfg(feature = "python")] diff --git a/src/output.rs b/src/output.rs new file mode 100644 index 0000000..35fbffe --- /dev/null +++ b/src/output.rs @@ -0,0 +1,53 @@ +use std::io::{self, Write}; + +/// Receives decoded bytes and can take ownership of decoder chunks. +/// +/// Implementations that cannot use ownership only need to implement +/// `write_borrowed`. The default `write_owned_from` forwards the suffix. +pub trait OutputSink { + fn write_borrowed(&mut self, bytes: &[u8]) -> io::Result<()>; + + fn write_owned_from(&mut self, bytes: Vec, start: usize) -> io::Result<()> { + let suffix = bytes.get(start..).ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "owned chunk start exceeds its length"))?; + self.write_borrowed(suffix) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +impl OutputSink for &mut S { + fn write_borrowed(&mut self, bytes: &[u8]) -> io::Result<()> { + (**self).write_borrowed(bytes) + } + + fn write_owned_from(&mut self, bytes: Vec, start: usize) -> io::Result<()> { + (**self).write_owned_from(bytes, start) + } + + fn flush(&mut self) -> io::Result<()> { + (**self).flush() + } +} + +/// Adapts any `std::io::Write` destination to an `OutputSink`. +pub struct WriterSink { + writer: W, +} + +impl WriterSink { + pub fn new(writer: W) -> Self { + Self { writer } + } +} + +impl OutputSink for WriterSink { + fn write_borrowed(&mut self, bytes: &[u8]) -> io::Result<()> { + self.writer.write_all(bytes) + } + + fn flush(&mut self) -> io::Result<()> { + self.writer.flush() + } +} diff --git a/tests/archive_perf.rs b/tests/archive_perf.rs new file mode 100644 index 0000000..d97998f --- /dev/null +++ b/tests/archive_perf.rs @@ -0,0 +1,207 @@ +use std::{ + fs, + io::{self, Write}, + process::Command, + time::{Duration, Instant}, +}; + +use crabz2::{Level, compress}; +use fastbz2::{DecodeOptions, OutputSink, decompress, gzip as gzip_decoder}; +use flate2::{Compression, write::GzEncoder}; + +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 binary() -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_fastbz2")); + command.args(["-P", &requested_threads().to_string()]); + command +} + +fn simplewiki_prefix() -> Vec { + let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("meta/simplewiki-first-5pct.xml.bz2"); + let encoded = fs::read(&path).unwrap_or_else(|error| panic!("could not read {}: {error}", path.display())); + let contents = decompress(&encoded, DecodeOptions::default()).unwrap(); + assert_eq!(contents.len(), 84_423_012); + contents +} + +fn tar_bytes(contents: &[u8]) -> Vec { + let mut archive = Vec::new(); + { + let mut builder = tar::Builder::new(&mut archive); + let mut header = tar::Header::new_gnu(); + header.set_size(contents.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + builder.append_data(&mut header, "payload.bin", contents).unwrap(); + builder.finish().unwrap(); + } + archive +} + +fn gzip(contents: &[u8]) -> Vec { + let mut encoder = GzEncoder::new(Vec::new(), Compression::fast()); + encoder.write_all(contents).unwrap(); + encoder.finish().unwrap() +} + +fn timed(action: impl FnOnce()) -> Duration { + let started = Instant::now(); + action(); + started.elapsed() +} + +fn timed_command(command: &mut Command) -> Duration { + timed(|| assert!(command.status().unwrap().success())) +} + +struct Fixture { + directory: tempfile::TempDir, + contents: Vec, + input: std::path::PathBuf, +} + +fn fixture(extension: &str, encode: impl Fn(&[u8]) -> Vec) -> Fixture { + let directory = tempfile::tempdir().unwrap(); + let contents = simplewiki_prefix(); + let archive = tar_bytes(&contents); + let input = directory.path().join(format!("archive.{extension}")); + let encoded = encode(&archive); + eprintln!("{extension}: {} MiB decoded from {:.1} MiB compressed", archive.len() / (1024 * 1024), encoded.len() as f64 / (1024.0 * 1024.0)); + fs::write(&input, encoded).unwrap(); + Fixture { directory, contents, input } +} + +fn fastbz2_overhead(extension: &str, encode: impl Fn(&[u8]) -> Vec) { + let fixture = fixture(extension, encode); + let warm = fixture.directory.path().join("warm"); + fs::create_dir(&warm).unwrap(); + assert!(binary().args(["-C", warm.to_str().unwrap(), fixture.input.to_str().unwrap()]).status().unwrap().success()); + + let raw = fixture.directory.path().join("archive.tar"); + let raw_time = timed_command(binary().args([fixture.input.to_str().unwrap(), "-o", raw.to_str().unwrap()])); + let extracted = fixture.directory.path().join("extracted"); + let extract_time = timed_command(binary().args(["-C", extracted.to_str().unwrap(), fixture.input.to_str().unwrap()])); + assert_eq!(fs::read(extracted.join("payload.bin")).unwrap(), fixture.contents); + + let raw_ratio = extract_time.as_secs_f64() / raw_time.as_secs_f64(); + eprintln!("{extension}: raw tar {raw_time:.3?}, fastbz2 extract {extract_time:.3?} ({raw_ratio:.3}x raw)"); + assert!(raw_ratio <= 3.0, "tar extraction exceeded the broad 3x raw-decode guard; measured {raw_ratio:.3}x"); +} + +fn system_reference(extension: &str, encode: impl Fn(&[u8]) -> Vec) { + let fixture = fixture(extension, encode); + let warm = fixture.directory.path().join("warm"); + fs::create_dir(&warm).unwrap(); + assert!(Command::new("tar").args(["-xf", fixture.input.to_str().unwrap(), "-C", warm.to_str().unwrap()]).status().unwrap().success()); + + let extracted = fixture.directory.path().join("extracted"); + fs::create_dir(&extracted).unwrap(); + let extract_time = timed_command(Command::new("tar").args(["-xf", fixture.input.to_str().unwrap(), "-C", extracted.to_str().unwrap()])); + assert_eq!(fs::read(extracted.join("payload.bin")).unwrap(), fixture.contents); + eprintln!("{extension}: system tar {extract_time:.3?}"); +} + +#[test] +#[ignore = "local single-run tar crate extraction reference"] +fn tar_crate_reference() { + let directory = tempfile::tempdir().unwrap(); + let contents = simplewiki_prefix(); + let archive = tar_bytes(&contents); + let unpack = |destination: &std::path::Path| { + fs::create_dir(destination).unwrap(); + tar::Archive::new(archive.as_slice()).unpack(destination).unwrap(); + }; + + unpack(&directory.path().join("warm")); + let extracted = directory.path().join("extracted"); + let extract_time = timed(|| unpack(&extracted)); + assert_eq!(fs::read(extracted.join("payload.bin")).unwrap(), contents); + eprintln!("uncompressed tar crate: {extract_time:.3?}"); +} + +struct CadenceSink { + started: Instant, + bytes: usize, + events: Vec<(usize, Duration)>, +} + +impl CadenceSink { + fn new() -> Self { + Self { started: Instant::now(), bytes: 0, events: Vec::new() } + } + + fn record(&mut self, bytes: usize) { + if bytes == 0 { + return; + } + self.bytes += bytes; + self.events.push((self.bytes, self.started.elapsed())); + } + + fn milestone(&self, bytes: usize) -> Duration { + self.events.iter().find(|(total, _)| *total >= bytes).unwrap().1 + } +} + +impl OutputSink for CadenceSink { + fn write_borrowed(&mut self, buffer: &[u8]) -> io::Result<()> { + self.record(buffer.len()); + Ok(()) + } + + fn write_owned_from(&mut self, buffer: Vec, start: usize) -> io::Result<()> { + let bytes = buffer.len().checked_sub(start).ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "owned chunk start exceeds its length"))?; + self.record(bytes); + Ok(()) + } +} + +#[test] +#[ignore = "local single-run parallel gzip output cadence"] +fn tgz_output_cadence() { + let contents = simplewiki_prefix(); + let archive = tar_bytes(&contents); + let encoded = gzip(&archive); + let options = DecodeOptions { threads: requested_threads(), ..DecodeOptions::default() }; + let threads = options.resolved_threads(); + let mut output = CadenceSink::new(); + let report = gzip_decoder::decompress_to_sink_with_options_and_progress(&encoded, &mut output, options, |_| {}).unwrap(); + let elapsed = output.started.elapsed(); + assert_eq!(output.bytes, archive.len()); + assert_eq!(report.decoded_len, archive.len() as u64); + eprintln!( + "gzip {threads} threads, {} chunks: first {:.3?}, 25% {:.3?}, 50% {:.3?}, 75% {:.3?}, complete {elapsed:.3?}", + output.events.len(), + output.events[0].1, + output.milestone(archive.len() / 4), + output.milestone(archive.len() / 2), + output.milestone(archive.len() * 3 / 4), + ); +} + +#[test] +#[ignore = "local single-run fastbz2 gzip tar extraction overhead"] +fn tgz_fastbz2_overhead() { + fastbz2_overhead("tgz", gzip); +} + +#[test] +#[ignore = "local single-run system gzip tar extraction reference"] +fn tgz_system_reference() { + system_reference("tgz", gzip); +} + +#[test] +#[ignore = "local single-run fastbz2 bzip2 tar extraction overhead"] +fn tbz2_fastbz2_overhead() { + fastbz2_overhead("tbz2", |contents| compress(contents, Level::BEST)); +} + +#[test] +#[ignore = "local single-run system bzip2 tar extraction reference"] +fn tbz2_system_reference() { + system_reference("tbz2", |contents| compress(contents, Level::BEST)); +} diff --git a/tests/cli.rs b/tests/cli.rs index 492a39d..a7a7c61 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -17,10 +17,81 @@ fn write_compressed(path: &Path, plain: &[u8]) { fs::write(path, compress(plain, Level::FASTEST)).unwrap(); } -fn write_gzip(path: &Path, plain: &[u8]) { +fn gzip_bytes(plain: &[u8]) -> Vec { let mut encoder = GzEncoder::new(Vec::new(), Compression::best()); encoder.write_all(plain).unwrap(); - fs::write(path, encoder.finish().unwrap()).unwrap(); + encoder.finish().unwrap() +} + +fn write_gzip(path: &Path, plain: &[u8]) { + fs::write(path, gzip_bytes(plain)).unwrap(); +} +fn tar_bytes(entries: &[(&str, &[u8])]) -> Vec { + let mut archive = Vec::new(); + { + let mut builder = tar::Builder::new(&mut archive); + for (path, contents) in entries { + let mut header = tar::Header::new_gnu(); + header.set_size(contents.len() as u64); + header.set_mode(0o640); + header.set_mtime(1_700_000_123); + header.set_cksum(); + builder.append_data(&mut header, path, *contents).unwrap(); + } + builder.finish().unwrap(); + } + archive +} + +fn write_tgz(path: &Path, entries: &[(&str, &[u8])]) { + write_gzip(path, &tar_bytes(entries)); +} + +fn traversal_tar() -> Vec { + let contents = b"must stay inside destination"; + let mut header = tar::Header::new_gnu(); + header.set_size(contents.len() as u64); + header.set_mode(0o644); + header.set_path("safe.txt").unwrap(); + header.as_mut_bytes()[..100].fill(0); + header.as_mut_bytes()[..14].copy_from_slice(b"../outside.txt"); + header.set_cksum(); + + let mut archive = Vec::new(); + { + let mut builder = tar::Builder::new(&mut archive); + builder.append(&header, contents.as_slice()).unwrap(); + builder.finish().unwrap(); + } + archive +} + +fn linked_tar() -> Vec { + let contents = b"linked contents"; + let mut archive = Vec::new(); + { + let mut builder = tar::Builder::new(&mut archive); + let mut file = tar::Header::new_gnu(); + file.set_size(contents.len() as u64); + file.set_mode(0o640); + file.set_mtime(1_700_000_123); + file.set_cksum(); + builder.append_data(&mut file, "root.txt", contents.as_slice()).unwrap(); + + for (path, entry_type) in [("symbolic.txt", tar::EntryType::Symlink), ("hard.txt", tar::EntryType::Link)] { + let mut link = tar::Header::new_gnu(); + link.set_path(path).unwrap(); + link.set_link_name("root.txt").unwrap(); + link.set_entry_type(entry_type); + link.set_size(0); + link.set_mode(0o777); + link.set_mtime(1_700_000_123); + link.set_cksum(); + builder.append(&link, std::io::empty()).unwrap(); + } + builder.finish().unwrap(); + } + archive } #[test] @@ -266,7 +337,7 @@ fn gzip_magic_fallback_stdin_limits_and_corruption_work() { } #[test] -fn mixed_bzip2_and_gzip_inputs_share_output_policy() { +fn mixed_bzip2_gzip_and_tar_inputs_share_output_policy() { let directory = tempfile::tempdir().unwrap(); let bzip2 = directory.path().join("first.bz2"); let gzip = directory.path().join("second.gz"); @@ -274,11 +345,124 @@ fn mixed_bzip2_and_gzip_inputs_share_output_policy() { let output_dir = directory.path().join("decoded"); write_compressed(&bzip2, b"bzip2"); write_gzip(&gzip, b"gzip"); - write_gzip(&tgz, b"tar payload"); + write_tgz(&tgz, &[("from-tar.txt", 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"); + assert_eq!(fs::read(output_dir.join("from-tar.txt")).unwrap(), b"tar payload"); +} + +#[test] +fn tar_gzip_and_bzip2_auto_extract_or_decode_raw() { + let directory = tempfile::tempdir().unwrap(); + let tar_gzip = directory.path().join("bundle.tar.gz"); + let tar_bzip2 = directory.path().join("bundle.tar.bz2"); + let gzip_output = directory.path().join("from-gzip"); + let bzip2_output = directory.path().join("from-bzip2"); + let raw_output = directory.path().join("bundle.tar"); + let long_path = format!("nested/{}/contents.txt", "long-segment-".repeat(10)); + let entries = [("root.txt", b"root contents".as_slice()), (long_path.as_str(), b"nested contents".as_slice())]; + let plain_tar = tar_bytes(&entries); + write_gzip(&tar_gzip, &plain_tar); + write_compressed(&tar_bzip2, &plain_tar); + + let gzip = binary().args(["-C", gzip_output.to_str().unwrap(), tar_gzip.to_str().unwrap()]).output().unwrap(); + assert!(gzip.status.success(), "{}", String::from_utf8_lossy(&gzip.stderr)); + let bzip2 = binary().args(["-C", bzip2_output.to_str().unwrap(), tar_bzip2.to_str().unwrap()]).output().unwrap(); + assert!(bzip2.status.success(), "{}", String::from_utf8_lossy(&bzip2.stderr)); + for output in [&gzip_output, &bzip2_output] { + assert_eq!(fs::read(output.join("root.txt")).unwrap(), b"root contents"); + assert_eq!(fs::read(output.join(&long_path)).unwrap(), b"nested contents"); + } + + let raw = binary().args([tar_gzip.to_str().unwrap(), "-o", raw_output.to_str().unwrap()]).output().unwrap(); + assert!(raw.status.success()); + assert_eq!(fs::read(raw_output).unwrap(), plain_tar); +} + +#[test] +fn tar_extraction_is_validated_before_entries_are_committed() { + let directory = tempfile::tempdir().unwrap(); + let input = directory.path().join("policy.tgz"); + let output = directory.path().join("output"); + fs::create_dir(&output).unwrap(); + fs::write(output.join("existing.txt"), b"keep me").unwrap(); + write_tgz(&input, &[("new.txt", b"new"), ("existing.txt", b"replacement")]); + + let skipped_output = directory.path().join("skipped"); + let skipped = binary().args(["--skip-existing", "-C", skipped_output.to_str().unwrap(), input.to_str().unwrap()]).output().unwrap(); + assert_eq!(skipped.status.code(), Some(2)); + assert!(!skipped_output.exists()); + + let rejected = binary().args(["-C", output.to_str().unwrap(), input.to_str().unwrap()]).output().unwrap(); + assert_eq!(rejected.status.code(), Some(1)); + assert_eq!(fs::read(output.join("existing.txt")).unwrap(), b"keep me"); + assert!(!output.join("new.txt").exists()); + + let replaced = binary().args(["--force", "-C", output.to_str().unwrap(), input.to_str().unwrap()]).output().unwrap(); + assert!(replaced.status.success(), "{}", String::from_utf8_lossy(&replaced.stderr)); + assert_eq!(fs::read(output.join("existing.txt")).unwrap(), b"replacement"); + assert_eq!(fs::read(output.join("new.txt")).unwrap(), b"new"); + + fs::remove_dir_all(&output).unwrap(); + let mut corrupt = fs::read(&input).unwrap(); + let crc = corrupt.len() - 8; + corrupt[crc] ^= 1; + fs::write(&input, corrupt).unwrap(); + let corrupt_result = binary().args(["--rm", "-C", output.to_str().unwrap(), input.to_str().unwrap()]).output().unwrap(); + assert_eq!(corrupt_result.status.code(), Some(3)); + assert!(input.exists()); + assert!(!output.join("new.txt").exists()); +} + +#[test] +fn explicit_extract_supports_stdin_limits_and_rejects_traversal() { + let directory = tempfile::tempdir().unwrap(); + let output = directory.path().join("output"); + let plain_tar = tar_bytes(&[("stdin.txt", b"streamed")]); + let compressed = gzip_bytes(&plain_tar); + + let mut child = binary().args(["--extract", "-C", output.to_str().unwrap(), "-"]).stdin(Stdio::piped()).stdout(Stdio::piped()).spawn().unwrap(); + child.stdin.take().unwrap().write_all(&compressed).unwrap(); + let extracted = child.wait_with_output().unwrap(); + assert!(extracted.status.success(), "{}", String::from_utf8_lossy(&extracted.stderr)); + assert_eq!(fs::read(output.join("stdin.txt")).unwrap(), b"streamed"); + + let limited = directory.path().join("limited"); + let limit = (plain_tar.len() - 1).to_string(); + let mut limited_child = binary().args(["--extract", "--max-output", &limit, "-C", limited.to_str().unwrap(), "-"]).stdin(Stdio::piped()).spawn().unwrap(); + limited_child.stdin.take().unwrap().write_all(&compressed).unwrap(); + let limited_result = limited_child.wait().unwrap(); + assert_eq!(limited_result.code(), Some(3)); + assert!(!limited.join("stdin.txt").exists()); + + let traversal = directory.path().join("traversal.tgz"); + write_gzip(&traversal, &traversal_tar()); + let traversal_output = directory.path().join("traversal-output"); + let result = binary().args(["-C", traversal_output.to_str().unwrap(), traversal.to_str().unwrap()]).output().unwrap(); + assert!(result.status.success()); + assert_eq!(fs::read_dir(&traversal_output).unwrap().count(), 0); + assert!(!directory.path().join("outside.txt").exists()); +} +#[test] +#[cfg(unix)] +fn tar_staging_preserves_symbolic_and_hard_links() { + use std::os::unix::fs::{MetadataExt, PermissionsExt}; + + let directory = tempfile::tempdir().unwrap(); + let input = directory.path().join("links.tgz"); + let output = directory.path().join("output"); + write_gzip(&input, &linked_tar()); + + let result = binary().args(["--rm", "-C", output.to_str().unwrap(), input.to_str().unwrap()]).output().unwrap(); + assert!(result.status.success(), "{}", String::from_utf8_lossy(&result.stderr)); + assert!(!input.exists()); + assert_eq!(fs::read_link(output.join("symbolic.txt")).unwrap(), Path::new("root.txt")); + assert_eq!(fs::read(output.join("hard.txt")).unwrap(), b"linked contents"); + let root_metadata = fs::metadata(output.join("root.txt")).unwrap(); + assert_eq!(root_metadata.ino(), fs::metadata(output.join("hard.txt")).unwrap().ino()); + assert_eq!(root_metadata.permissions().mode() & 0o777, 0o640); + assert_eq!(root_metadata.modified().unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs(), 1_700_000_123); } diff --git a/tests/wiki_perf.rs b/tests/wiki_perf.rs index 9c924f7..d36170a 100644 --- a/tests/wiki_perf.rs +++ b/tests/wiki_perf.rs @@ -165,6 +165,20 @@ fn rapidgzip_rust_process_metrics() { print_process_metrics("rapidgzip-rust", &metrics); } +#[test] +#[cfg(unix)] +#[ignore = "local single-run fastbz2 full gzip validation"] +fn gzip_fastbz2_validation() { + let path = corpus_path("simplewiki-full.xml.gz"); + let warm_path = corpus_path("simplewiki-first-5pct.xml.gz"); + let threads = requested_threads(); + let binary = env!("CARGO_BIN_EXE_fastbz2"); + warm_validation(binary, &warm_path, threads); + let result = timed_validation(binary, &path, threads); + assert!(result.status.success()); + eprintln!("fastbz2 full gzip: {:.3}s", result.wall.as_secs_f64()); +} + #[test] #[cfg(unix)] #[ignore = "local single-run gzip performance ratio against rapidgzip-rust"]