diff --git a/Cargo.toml b/Cargo.toml index 8c749cb..edf81ce 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.6" edition = "2024" rust-version = "1.91" license = "Apache-2.0" -description = "Compression-format research workbench with fast bzip2 and gzip decompression" +description = "Compression-format research workbench with fast bzip2, gzip, and ZIP decompression" repository = "https://github.com/AnswerDotAI/fastbz2" homepage = "https://github.com/AnswerDotAI/fastbz2" documentation = "https://github.com/AnswerDotAI/fastbz2" @@ -21,6 +21,7 @@ test = false [profile.release] lto = true codegen-units = 1 +strip = true [dependencies] blake3 = "1.8.7" @@ -32,6 +33,7 @@ rayon = "1.12.0" serde_json = "1.0.151" tempfile = "3.27.0" tar = { version = "0.4.46", default-features = false } +zip = { version = "8.6.0", default-features = false } [dev-dependencies] crabz2 = { version = "0.4.0", features = ["parallel"] } diff --git a/DEV.md b/DEV.md index edca9f5..7f082a6 100644 --- a/DEV.md +++ b/DEV.md @@ -11,14 +11,17 @@ 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/gzip.rs gzip framing, LSB-first DEFLATE implementation, CRC32, and reports +src/deflate.rs format-neutral raw-DEFLATE API shared by gzip and ZIP 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 +src/bin/fastbz2/archive_extract.rs shared same-filesystem staging and atomic commit +src/bin/fastbz2/tar_extract.rs bounded decode-to-tar bridge +src/bin/fastbz2/zip_extract.rs ZIP parsing policy and adaptive entry extraction 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 @@ -32,10 +35,14 @@ The gzip decoder is an in-repo RFC 1952/RFC 1951 implementation rather than a wr 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. An `OutputSink` wrapper enforces output-size limits, so each decoder has one code path for files, stdout, validation, listing, and tar extraction. +Core decode APIs report completed compressed and decoded byte counts without knowing anything about terminals. The CLI selects bzip2, gzip, or ZIP 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 archive 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. +ZIP structure semantics use `zip` 8.6.0 with all codec features disabled. The crate parses the central directory, Zip64 fields, data descriptors, names, modes, symlink kinds, and timestamp extra fields; fastbz2 reads each raw stored/DEFLATE range and sends it through its own validated codec path. Because the crate intentionally collapses duplicate raw names into its index map, a small bounds-checked central-record count detects and rejects that ambiguity before using its metadata. Parsing also rejects encryption, unsupported methods, escaping or equivalent paths, non-directory ancestors, overlapping data ranges, and ranges crossing the central directory. An aggregate declared-size check runs before extraction, and a per-entry sink prevents output exceeding its declaration before size and CRC32 are checked. ZIP and tar share the same staging/preflight/rename implementation. + +ZIP scheduling deliberately uses one parallelism level at a time. A sole DEFLATE entry at least 16 MiB compressed, or an entry at least 64 MiB in a multi-entry archive, uses all requested workers inside the raw DEFLATE decoder. Remaining entries run serial inner decoders concurrently on one Rayon pool. This keeps thread ownership and memory behaviour obvious, avoids nested oversubscription, and lets many ordinary entries naturally absorb stragglers through work stealing. + 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 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. @@ -57,7 +64,7 @@ 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. 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 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 and raw-DEFLATE tests cover stored, fixed-Huffman, and dynamic-Huffman blocks; optional headers and FHCRC; concatenated members; exact end-of-stream boundaries; 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 and ZIP archives and cover wrappers and extension dispatch, stored/DEFLATE entries, Zip64 and streaming descriptors, metadata and safe symlinks, stdin, raw-tar output, output limits, overwrite preflight, late checksum failure, atomicity, 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. @@ -71,7 +78,7 @@ cargo test --release --test archive_perf tgz_system_reference -- --ignored --exa 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 +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: @@ -87,12 +94,67 @@ The cadence benchmark identified ordered gzip output as the main overlap limit. 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. +### Local ZIP benchmarks + +`tests/zip_perf.rs` creates two deterministic ZIPs from the same 84,423,012-byte SimpleWiki prefix used by the tar benchmarks: one DEFLATE entry, and 18 equal-sized DEFLATE entries. `tests/support::simplewiki_prefix` decodes `meta/simplewiki-first-5pct.xml.bz2` before timing, and the ZIP builder then runs before timing. Thus regenerating the dataset is exactly the SimpleWiki 5% procedure below; no ZIP fixture is stored or needs hand-maintained lengths. + +Each speed test warms its selected executable once, measures it once, and verifies all extracted bytes. Run only the row whose implementation changed: + +```bash +cargo test --release --test zip_perf zip_single_fastbz2 -- --ignored --exact --nocapture +cargo test --release --test zip_perf zip_single_unzip -- --ignored --exact --nocapture +cargo test --release --test zip_perf zip_many_fastbz2 -- --ignored --exact --nocapture +cargo test --release --test zip_perf zip_many_unzip -- --ignored --exact --nocapture +``` + +The established external baseline is `UnZip 6.00 of 20 April 2009, by Info-ZIP, with modifications by Apple Inc.` The measured archive was 25.8 MiB and decoded to 80.5 MiB: + +| Shape | fastbz2 | Info-ZIP `unzip` | fastbz2/reference | +|---|---:|---:|---:| +| One entry | 35.571 ms | 355.235 ms | 0.100x | +| 18 entries | 29.700 ms | 360.008 ms | 0.083x | + +Both fastbz2 measurements used the default automatic worker selection, resolving to 18 cores on the primary machine. `FASTBZ2_THREADS` is only an optional diagnostic override; normal tests and benchmarks should leave it unset so `-P 0` follows the machine automatically. + +The ungated child-process memory diagnostics use the same many-entry fixture: + +```bash +cargo test --release --test zip_perf zip_many_fastbz2_process_metrics -- --ignored --exact --nocapture +cargo test --release --test zip_perf zip_many_unzip_process_metrics -- --ignored --exact --nocapture +``` + +The single sampled runs measured fastbz2 at 66.4 MiB peak RSS and 36.7 MiB physical footprint, versus `unzip` at 2.8 MiB RSS and 2.4 MiB physical footprint. That is the explicit cost of decoding 18 entries concurrently; it remains small in absolute terms and avoids the much larger speculative-history footprint of parallelising each small entry internally. + 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 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`. +This section retains in-process and library-oriented research comparisons that are useful to implementation work but deliberately excluded from the user-facing README. Full Simple English Wikipedia (`338 MB` compressed, `1,688,460,257` bytes decoded): + +| Decoder | Mode | Seconds | +|---|---|---:| +| 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 | + +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 | +|---|---|---:| +| crabz2 0.4.0 | parallel, in process | 3.815 | +| fastbz2 | parallel, 18 threads, in process | 3.881 | +| fastbz2 | serial, in process | 37.198 | +| crabz2 0.4.0 | serial, in process | 40.602 | +| pbzip2 1.1.13 | 18-thread CLI + byte comparison | 88.080 | +| bzip2 | serial CLI + byte comparison | 92.960 | + +The last two rows stream 2.5 GB through `cmp` against the validated XML, so their absolute times are not directly comparable with the in-process rows. The tables record distinct implementation experiments, not a user-facing CLI ranking. + The quick bzip2 iteration test reads `meta/simplewiki-first-5pct.xml.bz2` before timing, then verifies decoded length, all CRCs, and BLAKE3: ```bash @@ -112,7 +174,7 @@ cargo test --release --test wiki_perf gzip_fastbz2_validation -- --ignored --exa cargo test --release --test wiki_perf gzip_reference_ratio -- --ignored --exact --nocapture ``` -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. +The default selects available parallelism automatically. `FASTBZ2_THREADS` is an optional diagnostic override, and `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: @@ -122,7 +184,7 @@ cargo test --release --test wiki_perf rapidgzip_rust_process_metrics -- --ignore 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 metrics helper uses `wait4` and, on macOS, `proc_pid_rusage`'s `ri_phys_footprint` field 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 @@ -132,7 +194,7 @@ cargo test --release --test wiki_perf enwiki_first_1000_fastbz2_serial -- --igno cargo test --release --test wiki_perf enwiki_first_1000_crabz2_serial -- --ignored --exact --nocapture ``` -It compares fastbz2 and crabz2 in parallel and serial modes. Set `FASTBZ2_THREADS` to give both parallel decoders an explicit thread count. Each implementation validates the bzip2 CRCs; the benchmark also checks the exact decoded length. +It compares fastbz2 and crabz2 in parallel and serial modes. Automatic parallelism is the default; `FASTBZ2_THREADS` can give both parallel decoders an explicit count for a diagnostic comparison. Each implementation validates the bzip2 CRCs; the benchmark also checks the exact decoded length. The decoded lengths and the 5% BLAKE3 in `tests/wiki_perf.rs` are acceptance values, not parameters used by the decoder. They prevent a truncated decode from appearing artificially fast without reading a separate multi-gigabyte reference during each timed run. Regenerating a fixture requires independently validating it and updating the corresponding acceptance value. diff --git a/README.md b/README.md index 1daf019..0b6c224 100644 --- a/README.md +++ b/README.md @@ -1,10 +1,10 @@ # fastbz2 -An active compression-format research workbench with fast bzip2/gzip decompression and streaming tar extraction. +An active compression-format research workbench with fast bzip2, gzip, and ZIP decompression plus safe 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, 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. +`fastbz2` provides a native CLI, a Rust library, and a Python module. The CLI auto-selects bzip2, gzip, or ZIP handling from the filename extension, falling back to stream magic when needed. It streams compressed tar variants through a bounded extractor and adaptively parallelises ZIP work within or across entries. Every path validates decoded sizes and checksums. The bzip2 implementation also provides persistent random-access indexes. -Compression is planned, but the current implementation decompresses bzip2 and gzip and extracts their tar-wrapped variants. +Compression is planned, but the current implementation decompresses bzip2 and gzip, extracts their tar-wrapped variants, and extracts stored or DEFLATE-compressed ZIP archives. ## Install @@ -25,14 +25,15 @@ 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. 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`. +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`—and `.zip` automatically extract into the current directory or `-C/--output-dir`. `-x/--extract` forces archive extraction for stdin or an unusual filename; an explicit `-o/--output` instead writes a decoded tar stream, but is invalid for ZIP because ZIP has no single decoded byte stream. For stdin and unrecognised extensions, bzip2, gzip, or ZIP magic selects the format. Other non-archive input names gain `.out`. ```bash 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 dataset.zip -C unpacked # extract ZIP entries adaptively in parallel +fastbz2 --extract -C unpacked - # extract tar or ZIP 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 @@ -44,6 +45,7 @@ Multiple inputs are processed in order, with parallelism applied inside each com fastbz2 data/*.bz2 logs/*.gz -C decoded fastbz2 data/*.bz2 logs/*.gz -C decoded --skip-existing fastbz2 backups/*.tgz -C restored +fastbz2 datasets/*.zip -C restored ``` Validation and inspection remain flags rather than subcommands: @@ -52,6 +54,7 @@ Validation and inspection remain flags rather than subcommands: fastbz2 --test dump.xml.bz2 # fully decode and validate, writing nothing 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 dataset.zip # print the validated entry layout fastbz2 --list --json dump.xml.bz2 # emit the complete layout as JSON ``` @@ -61,14 +64,14 @@ fastbz2 --list --json dump.xml.bz2 # emit the complete layout as JSON - 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. +- Tar entries stream into a same-filesystem staging directory through a bounded pipe. ZIP entries decode directly into the same staging scheme. Entries are preflighted and moved into the destination only after every relevant compression stream and archive structure validates, so a late CRC failure leaves no extracted files. +- Tar and ZIP paths and link targets are confined to the destination. ZIP rejects unsafe or duplicate paths; tar safely skips unsafe entries. New entries use the archive's permissions and modification times where provided. 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. -`-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. +`-P/--threads 0`, the default, uses the machine's available parallelism; an explicit positive value is honoured by every decoder. `--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. ZIP uses one level of parallelism at a time: large entries use the parallel DEFLATE engine, while archives of ordinary entries decode files concurrently without nested worker pools. ## Python @@ -141,52 +144,38 @@ 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 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. +The raw shared codec is available as `fastbz2::deflate::decompress_to_sink_with_options_and_progress`; gzip framing and ZIP extraction both use this exact decoder. -Full Simple English Wikipedia (`338 MB` compressed, `1,688,460,257` bytes decoded): - -| Decoder | Mode | Seconds | -|---|---|---:| -| 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 | +## Performance +These are single local release-mode CLI runs on the primary 18-core Apple Silicon development machine. 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. ZIP likewise warms each CLI once and then measures one extraction. These are observations rather than statistical aggregates. In-process codec and library-oracle comparisons live in [DEV.md](DEV.md#local-wikipedia-benchmarks), not this user-facing table. Full SimpleWiki recompressed with system `gzip -6` (`438,904,466` bytes compressed, `1,688,460,257` bytes decoded): -| Decoder | Mode | Seconds | Peak physical footprint | +| CLI | Mode | Seconds | Peak physical footprint | |---|---|---:|---:| -| rapidgzip-rust, local checkout | auto parallel, validation sink | 0.363 | 585 MiB | -| fastbz2 | auto parallel, validation sink | 0.326 | 325 MiB | +| rapidgzip-rust, local checkout | auto parallel, `--test` | 0.363 | 460.0 MiB | +| fastbz2 | auto parallel, `--test` | 0.326 | 335.9 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: +ZIPs containing the same 80.5 MiB SimpleWiki prefix (`25.8 MiB` compressed), after one untimed warm-up per executable: -| Decoder | Mode | Seconds | -|---|---|---:| -| crabz2 0.4.0 | parallel, in process | 3.815 | -| fastbz2 | parallel, 18 threads, in process | 3.881 | -| fastbz2 | serial, in process | 37.198 | -| crabz2 0.4.0 | serial, in process | 40.602 | -| pbzip2 1.1.13 | 18-thread CLI + byte comparison | 88.080 | -| bzip2 | serial CLI + byte comparison | 92.960 | +| Shape | fastbz2, auto parallel | Info-ZIP unzip 6.00 (Apple) | Speedup | +|---|---:|---:|---:| +| One DEFLATE entry | 35.571 ms | 355.235 ms | 10.0x | +| 18 DEFLATE entries | 29.700 ms | 360.008 ms | 12.1x | -The CLI rows in the second table stream 2.5 GB through `cmp` against the validated XML, so their absolute times are not directly comparable with the in-process rows. [DEV.md](DEV.md#local-wikipedia-benchmarks) documents exact fixture generation and commands. +The many-entry sampled run used 36.7 MiB physical footprint for fastbz2 versus 2.4 MiB for `unzip`; its 18-way file parallelism deliberately spends modest memory to obtain the throughput above. Both fastbz2 rows used automatic thread selection, which resolved to 18 available cores on this machine. -Homebrew `pbzip2` 1.1.13 could not safely decompress the complete 26,668,484,995-byte English Wikipedia multistream dump on this machine. It segfaulted, and repeated attempts produced divergent and truncated plaintext. Its successful 1,000-stream result above does not establish full-file reliability. +Homebrew `pbzip2` 1.1.13 could not safely decompress the complete 26,668,484,995-byte English Wikipedia multistream dump on this machine. It segfaulted, and repeated attempts produced divergent and truncated plaintext. Successful smaller-file results therefore do not establish full-file reliability. ## Implementation and compatibility 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. +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. ZIP reuses that raw DEFLATE core and uses the mature `zip` crate only for container structure and metadata. It supports stored and DEFLATE entries, Zip64, streaming data descriptors, Unix symlinks/modes, and Unix/NTFS modification-time fields; encryption and uncommon legacy compression methods are intentionally unsupported. `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. @@ -200,6 +189,7 @@ The open-source implementations and codebases consulted were: - [`rapidgzip-rust`](https://github.com/COMBINE-lab/rapidgzip-rust), a pure-Rust reimplementation and fastbz2's local gzip performance and memory reference. - [`librapidarchive`](https://github.com/mxmlnkn/librapidarchive), an experimental shared architecture for parallel bzip2 and gzip access. - [`indexed_bzip2`](https://github.com/mxmlnkn/indexed_bzip2), for non-byte-aligned marker scanning, independent bzip2 block decoding, ordered prefetch, and indexed seeking. +- [`zip`](https://github.com/zip-rs/zip2), used without codec features for maintained ZIP structure and metadata handling. - Rob Landley's 0BSD [`bzcat` implementation in Toybox](https://github.com/landley/toybox), from which fastbz2's specialised bzip2 decoder is derived. ## Development diff --git a/src/bin/fastbz2.rs b/src/bin/fastbz2.rs index fa5d25b..e8eb9c5 100644 --- a/src/bin/fastbz2.rs +++ b/src/bin/fastbz2.rs @@ -1,5 +1,9 @@ +#[path = "fastbz2/archive_extract.rs"] +mod archive_extract; #[path = "fastbz2/tar_extract.rs"] mod tar_extract; +#[path = "fastbz2/zip_extract.rs"] +mod zip_extract; use std::{ fs, @@ -20,18 +24,19 @@ use tempfile::NamedTempFile; #[derive(Parser)] #[command( version, - 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.", + about = "Parallel bzip2/gzip/ZIP decompression and safe archive extraction", + long_about = "Parallel bzip2 and gzip decompression, streaming tar extraction, and adaptive parallel ZIP extraction. Decoding is the default operation. Recognised codec suffixes are removed from normal output names. ZIP and compressed tar archives extract automatically unless -o is given; -o is not valid for ZIP.", 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 dataset.zip -C restored Extract ZIP entries in parallel 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 except with --index. + /// Input files, or - for stdin except with --index or --list. #[arg(required = true, num_args = 1..)] inputs: Vec, /// Fully decode and validate checksums without writing output. @@ -40,10 +45,10 @@ struct Cli { /// Build validated, source-bound .fbz2i indexes for bzip2 inputs. #[arg(long)] index: bool, - /// Validate and show bzip2 streams/blocks or gzip members/blocks. + /// Validate and show bzip2 streams/blocks, gzip members/blocks, or ZIP entries. #[arg(long)] list: bool, - /// Extract a decoded tar stream; automatic for recognised compressed-tar suffixes. + /// Extract a tar or ZIP archive; automatic for recognised archive suffixes. #[arg(short = 'x', long)] extract: bool, /// Write decoded bytes to PATH, or - for stdout; requires one input and disables automatic extraction. @@ -82,6 +87,7 @@ struct Cli { enum Format { Bzip2, Gzip, + Zip, } fn main() -> ExitCode { @@ -99,7 +105,7 @@ fn run(cli: Cli) -> fastbz2::Result<()> { let options = DecodeOptions { threads: cli.threads, memory_limit: cli.memory_limit }; if cli.test { for input in &cli.inputs { - decode_input(input, &mut io::sink(), options, cli.max_output, cli.quiet)?; + test_input(input, options, cli.max_output, cli.quiet)?; } return Ok(()); } @@ -116,6 +122,9 @@ fn validate_cli(cli: &Cli) -> fastbz2::Result<()> { if cli.output.is_some() && cli.inputs.len() != 1 { return Err(invalid("--output requires exactly one input")); } + if cli.output.is_some() && cli.inputs.iter().any(|input| is_zip_archive(input)) { + return Err(invalid("--output is not supported for ZIP archives")); + } if cli.inputs.iter().any(|input| input == "-") && cli.inputs.len() != 1 { return Err(invalid("stdin must be the only input")); } @@ -129,7 +138,7 @@ fn validate_cli(cli: &Cli) -> fastbz2::Result<()> { } fn should_extract(cli: &Cli, input: &str) -> bool { - cli.extract || (cli.output.is_none() && is_tar_archive(input)) + cli.extract || (cli.output.is_none() && is_archive(input)) } fn run_decode(cli: &Cli, options: DecodeOptions) -> fastbz2::Result<()> { @@ -214,6 +223,15 @@ fn run_list(cli: &Cli, options: DecodeOptions) -> fastbz2::Result<()> { print_gzip_report((cli.inputs.len() > 1).then_some(input), &report); } } + Format::Zip => { + let mut display = ProgressDisplay::new(input, source.as_slice().len() as u64, cli.quiet); + let report = zip_extract::validate(source.as_slice(), options, cli.max_output, |progress| display.update(progress))?; + if cli.json { + values.push(zip_json(input, &report)); + } else { + print_zip_report((cli.inputs.len() > 1).then_some(input), &report); + } + } } } if cli.json { @@ -244,7 +262,31 @@ fn extract_data( max_output: Option, quiet: bool, ) -> fastbz2::Result<()> { - tar_extract::unpack(destination, overwrite, |writer| decode_data_to_sink(data, label, writer, options, max_output, quiet)) + if select_format(label, data)? == Format::Zip { + let mut display = ProgressDisplay::new(label, data.len() as u64, quiet); + zip_extract::unpack(data, destination, overwrite, options, max_output, |progress| display.update(progress)).map(|_| ()) + } else { + tar_extract::unpack(destination, overwrite, |writer| decode_data_to_sink(data, label, writer, options, max_output, quiet)) + } +} + +fn test_input(input: &str, options: DecodeOptions, max_output: Option, quiet: bool) -> fastbz2::Result<()> { + if input == "-" { + let mut data = Vec::new(); + io::stdin().lock().read_to_end(&mut data)?; + return test_data(&data, "stdin", options, max_output, quiet); + } + let source = Source::open(input)?; + test_data(source.as_slice(), input, options, max_output, quiet) +} + +fn test_data(data: &[u8], label: &str, options: DecodeOptions, max_output: Option, quiet: bool) -> fastbz2::Result<()> { + if select_format(label, data)? == Format::Zip { + let mut display = ProgressDisplay::new(label, data.len() as u64, quiet); + zip_extract::validate(data, options, max_output, |progress| display.update(progress)).map(|_| ()) + } else { + decode_data(data, label, &mut io::sink(), options, max_output, quiet) + } } fn decode_input(input: &str, output: &mut impl Write, options: DecodeOptions, max_output: Option, quiet: bool) -> fastbz2::Result<()> { @@ -276,6 +318,7 @@ fn decode_data_to_sink( match select_format(label, data)? { 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(|_| ()), + Format::Zip => Err(invalid("ZIP archives extract to a directory and cannot be decoded to one output stream")), } } @@ -455,6 +498,7 @@ fn format_extension(input: &Path) -> Option<(Format, &'static str)> { "tbz" | "tbz2" => Some((Format::Bzip2, "tar")), "gz" | "gzip" => Some((Format::Gzip, "")), "tgz" => Some((Format::Gzip, "tar")), + "zip" => Some((Format::Zip, "")), _ => None, } } @@ -464,6 +508,14 @@ fn is_tar_archive(input: &str) -> bool { [".tar.bz2", ".tar.bzip2", ".tbz", ".tbz2", ".tar.gz", ".tar.gzip", ".tgz"].iter().any(|extension| input.ends_with(extension)) } +fn is_zip_archive(input: &str) -> bool { + input.to_ascii_lowercase().ends_with(".zip") +} + +fn is_archive(input: &str) -> bool { + is_tar_archive(input) || is_zip_archive(input) +} + fn default_output(input: &Path) -> PathBuf { format_extension(input).map_or_else(|| PathBuf::from(format!("{}.out", input.display())), |(_, extension)| input.with_extension(extension)) } @@ -476,8 +528,10 @@ fn select_format(input: &str, data: &[u8]) -> fastbz2::Result { Ok(Format::Bzip2) } else if data.starts_with(&[0x1f, 0x8b]) { Ok(Format::Gzip) + } else if data.starts_with(b"PK\x03\x04") || data.starts_with(b"PK\x05\x06") || data.starts_with(b"PK\x07\x08") { + Ok(Format::Zip) } else { - Err(invalid(format!("cannot determine compression format for {input}; expected a bzip2/gzip extension or magic"))) + Err(invalid(format!("cannot determine compression format for {input}; expected a bzip2, gzip, or ZIP extension or magic"))) } } @@ -520,6 +574,26 @@ fn print_gzip_report(input: Option<&String>, report: &gzip::Report) { } } +fn print_zip_report(input: Option<&String>, report: &zip_extract::Report) { + if let Some(input) = input { + println!("input\t{input}"); + } + println!("format\tzip"); + println!("compressed_bytes\t{}", report.source_len); + println!("decoded_bytes\t{}", report.decoded_len); + println!("entries\t{}", report.entries.len()); + for (number, entry) in report.entries.iter().enumerate() { + println!( + "entry\t{number}\tmethod={}\tcompressed={}\tdecoded={}\tcrc={:08x}\tpath={}", + entry.compression_method, + entry.compressed_size, + entry.decoded_size, + entry.crc, + entry.path.display(), + ); + } +} + fn member_block_count(report: &gzip::Report, member: usize) -> usize { report.blocks.iter().filter(|block| block.member as usize == member).count() } @@ -589,6 +663,23 @@ fn gzip_json(input: &str, report: &gzip::Report) -> Value { }) } +fn zip_json(input: &str, report: &zip_extract::Report) -> Value { + json!({ + "input": input, + "format": "zip", + "source_bytes": report.source_len, + "decoded_bytes": report.decoded_len, + "entries": report.entries.iter().enumerate().map(|(number, entry)| json!({ + "number": number, + "path": entry.path, + "compression_method": entry.compression_method, + "compressed_bytes": entry.compressed_size, + "decoded_bytes": entry.decoded_size, + "expected_crc": entry.crc, + })).collect::>(), + }) +} + fn hex(bytes: &[u8]) -> String { bytes.iter().map(|byte| format!("{byte:02x}")).collect() } @@ -638,7 +729,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::InvalidGzip(_) | Error::Decode { .. } | Error::InvalidIndex(_) => 3, + Error::InvalidStreamHeader | Error::InvalidGzip(_) | Error::InvalidZip(_) | Error::Decode { .. } | Error::InvalidIndex(_) => 3, _ => 4, } } diff --git a/src/bin/fastbz2/archive_extract.rs b/src/bin/fastbz2/archive_extract.rs new file mode 100644 index 0000000..39d278c --- /dev/null +++ b/src/bin/fastbz2/archive_extract.rs @@ -0,0 +1,110 @@ +use std::{ + fs, io, + path::{Path, PathBuf}, +}; + +use fastbz2::{Error, Result}; +use tempfile::TempDir; + +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(()) +} + +pub(super) 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) +} + +pub(super) 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(()) +} diff --git a/src/bin/fastbz2/tar_extract.rs b/src/bin/fastbz2/tar_extract.rs index 040ce6b..d922179 100644 --- a/src/bin/fastbz2/tar_extract.rs +++ b/src/bin/fastbz2/tar_extract.rs @@ -1,13 +1,14 @@ use std::{ - cmp, fs, + cmp, io::{self, Read}, - path::{Path, PathBuf}, + path::Path, sync::mpsc::{Receiver, SyncSender, sync_channel}, thread, }; use fastbz2::{Error, OutputSink, Result}; -use tempfile::TempDir; + +use super::archive_extract; struct Chunk { bytes: Vec, @@ -74,113 +75,11 @@ 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)?; + let staging = archive_extract::staging(destination)?; thread::scope(|scope| { let (mut writer, mut reader) = pipe(); let decoder = scope.spawn(move || decode(&mut writer)); @@ -195,7 +94,7 @@ where 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), + (Ok(()), Ok(()), Ok(())) => archive_extract::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/bin/fastbz2/zip_extract.rs b/src/bin/fastbz2/zip_extract.rs new file mode 100644 index 0000000..31d1183 --- /dev/null +++ b/src/bin/fastbz2/zip_extract.rs @@ -0,0 +1,437 @@ +use std::{ + collections::HashMap, + fs, + io::{self, Cursor}, + path::{Component, Path, PathBuf}, + sync::{ + Mutex, + atomic::{AtomicU64, AtomicUsize, Ordering}, + }, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use fastbz2::{DecodeOptions, DecodeProgress, Error, OutputSink, Result, WriterSink, deflate, gzip}; +use rayon::prelude::*; +use zip::{CompressionMethod, ZipArchive, extra_fields::ExtraField}; + +use super::archive_extract; + +const MULTI_ENTRY_INTRA_THRESHOLD: u64 = 64 * 1024 * 1024; +const SINGLE_ENTRY_INTRA_THRESHOLD: u64 = 16 * 1024 * 1024; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum EntryKind { + File, + Directory, + Symlink, +} + +#[derive(Clone, Debug)] +pub(super) struct Entry { + pub path: PathBuf, + pub compression_method: u16, + pub compressed_size: u64, + pub decoded_size: u64, + pub crc: u32, + method: CompressionMethod, + data_start: usize, + data_end: usize, + mode: Option, + modified: Option, + kind: EntryKind, +} + +#[derive(Clone, Debug)] +pub(super) struct Report { + pub source_len: u64, + pub decoded_len: u64, + pub entries: Vec, +} + +fn invalid(message: impl Into) -> Error { + Error::InvalidZip(message.into()) +} + +fn zip_error(error: zip::result::ZipError) -> Error { + invalid(error.to_string()) +} + +fn ntfs_time(ticks: u64) -> Option { + const UNIX_EPOCH_TICKS: u64 = 116_444_736_000_000_000; + let duration = Duration::from_nanos(ticks.abs_diff(UNIX_EPOCH_TICKS).checked_mul(100)?); + if ticks >= UNIX_EPOCH_TICKS { UNIX_EPOCH.checked_add(duration) } else { UNIX_EPOCH.checked_sub(duration) } +} + +fn modified_time(file: &zip::read::ZipFile<'_, impl io::Read>) -> Option { + file.extra_data_fields().find_map(|field| match field { + ExtraField::ExtendedTimestamp(timestamp) => timestamp.mod_time().and_then(|seconds| UNIX_EPOCH.checked_add(Duration::from_secs(seconds.into()))), + ExtraField::Ntfs(timestamp) => ntfs_time(timestamp.mtime()), + }) +} + +fn central_entry_count(data: &[u8], start: u64) -> Result { + let mut position = usize::try_from(start).map_err(|_| invalid("central directory offset exceeds this platform"))?; + let mut count = 0; + while data.get(position..).is_some_and(|remaining| remaining.starts_with(b"PK\x01\x02")) { + let header = data.get(position..).and_then(|remaining| remaining.get(..46)).ok_or_else(|| invalid("truncated central directory entry"))?; + let name_len = u16::from_le_bytes(header[28..30].try_into().unwrap()) as usize; + let extra_len = u16::from_le_bytes(header[30..32].try_into().unwrap()) as usize; + let comment_len = u16::from_le_bytes(header[32..34].try_into().unwrap()) as usize; + position = position + .checked_add(46) + .and_then(|value| value.checked_add(name_len)) + .and_then(|value| value.checked_add(extra_len)) + .and_then(|value| value.checked_add(comment_len)) + .filter(|&value| value <= data.len()) + .ok_or_else(|| invalid("central directory entry exceeds the archive"))?; + count += 1; + } + Ok(count) +} + +fn parse(data: &[u8], max_output: Option) -> Result { + let mut archive = ZipArchive::new(Cursor::new(data)).map_err(zip_error)?; + let central_start = archive.central_directory_start(); + if central_entry_count(data, central_start)? != archive.len() { + return Err(invalid("central directory contains duplicate entry names")); + } + let mut entries = Vec::with_capacity(archive.len()); + let mut decoded_len = 0_u64; + for index in 0..archive.len() { + let file = archive.by_index_raw(index).map_err(zip_error)?; + if file.encrypted() { + return Err(invalid(format!("encrypted entry {:?} is not supported", file.name()))); + } + let path = file.enclosed_name().ok_or_else(|| invalid(format!("unsafe entry path {:?}", file.name())))?; + if path.as_os_str().is_empty() { + return Err(invalid("empty entry path")); + } + let data_start = usize::try_from(file.data_start().ok_or_else(|| invalid(format!("cannot locate entry {:?}", file.name())))?) + .map_err(|_| invalid(format!("entry offset for {:?} exceeds this platform", file.name())))?; + let data_end_u64 = + (data_start as u64).checked_add(file.compressed_size()).ok_or_else(|| invalid(format!("compressed range for {:?} overflows", file.name())))?; + if data_end_u64 > central_start || data_end_u64 > data.len() as u64 { + return Err(invalid(format!("compressed range for {:?} overlaps the central directory or exceeds the archive", file.name()))); + } + let data_end = data_end_u64 as usize; + let kind = if file.is_dir() { + EntryKind::Directory + } else if file.is_symlink() { + EntryKind::Symlink + } else { + EntryKind::File + }; + decoded_len = decoded_len.checked_add(file.size()).ok_or_else(|| invalid("total decoded size overflows u64"))?; + if let Some(limit) = max_output + && decoded_len > limit as u64 + { + return Err(invalid(format!("decoded output exceeds {limit} bytes"))); + } + let method = file.compression(); + #[allow(deprecated)] + let compression_method = method.to_u16(); + entries.push(Entry { + path, + compression_method, + compressed_size: file.compressed_size(), + decoded_size: file.size(), + crc: file.crc32(), + method, + data_start, + data_end, + mode: file.unix_mode(), + modified: modified_time(&file), + kind, + }); + } + validate_layout(&entries)?; + Ok(Report { source_len: data.len() as u64, decoded_len, entries }) +} + +fn validate_layout(entries: &[Entry]) -> Result<()> { + let mut paths = HashMap::with_capacity(entries.len()); + for entry in entries { + if paths.insert(entry.path.clone(), entry.kind).is_some() { + return Err(invalid(format!("duplicate entry path {}", entry.path.display()))); + } + } + for entry in entries { + for ancestor in entry.path.ancestors().skip(1).filter(|path| !path.as_os_str().is_empty()) { + if paths.get(ancestor).is_some_and(|kind| *kind != EntryKind::Directory) { + return Err(invalid(format!("non-directory entry {} is an ancestor of {}", ancestor.display(), entry.path.display()))); + } + } + } + let mut ranges: Vec<_> = entries.iter().filter(|entry| entry.compressed_size != 0).map(|entry| (entry.data_start, entry.data_end, &entry.path)).collect(); + ranges.sort_unstable_by_key(|range| range.0); + for pair in ranges.windows(2) { + if pair[0].1 > pair[1].0 { + return Err(invalid(format!("compressed data for {} overlaps {}", pair[0].2.display(), pair[1].2.display()))); + } + } + Ok(()) +} + +struct ExpectedOutput { + inner: W, + expected: u64, + written: u64, +} + +impl ExpectedOutput { + fn new(inner: W, expected: u64) -> Self { + Self { inner, expected, written: 0 } + } + + fn check(&self, count: usize) -> io::Result<()> { + if self.written.saturating_add(count as u64) > self.expected { + return Err(io::Error::new(io::ErrorKind::InvalidData, format!("ZIP entry exceeds its declared {}-byte size", self.expected))); + } + Ok(()) + } +} + +impl OutputSink for ExpectedOutput { + fn write_borrowed(&mut self, bytes: &[u8]) -> io::Result<()> { + self.check(bytes.len())?; + self.inner.write_borrowed(bytes)?; + self.written += bytes.len() as u64; + Ok(()) + } + + fn write_owned_from(&mut self, bytes: Vec, start: usize) -> io::Result<()> { + let count = bytes.get(start..).ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "owned chunk start exceeds its length"))?.len(); + self.check(count)?; + self.inner.write_owned_from(bytes, start)?; + self.written += count as u64; + Ok(()) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} + +fn entry_error(entry: &Entry, error: Error) -> Error { + let detail = match error { + Error::InvalidGzip(message) | Error::InvalidZip(message) => message, + other => other.to_string(), + }; + invalid(format!("entry {}: {detail}", entry.path.display())) +} + +fn decode_entry(data: &[u8], entry: &Entry, output: &mut impl OutputSink, options: DecodeOptions, progress: &mut dyn FnMut(DecodeProgress)) -> Result<()> { + let compressed = &data[entry.data_start..entry.data_end]; + let mut output = ExpectedOutput::new(output, entry.decoded_size); + let (decoded_size, crc) = if entry.method == CompressionMethod::STORE { + if entry.compressed_size != entry.decoded_size { + return Err(invalid(format!("stored entry {} has different compressed and decoded sizes", entry.path.display()))); + } + output.write_borrowed(compressed)?; + progress(DecodeProgress { compressed_bytes: entry.compressed_size, decoded_bytes: entry.decoded_size }); + (compressed.len() as u64, gzip::crc32(compressed)) + } else if entry.method == CompressionMethod::DEFLATE { + let report = + deflate::decompress_to_sink_with_options_and_progress(compressed, &mut output, options, progress).map_err(|error| entry_error(entry, error))?; + (report.decoded_len, report.crc) + } else { + return Err(invalid(format!("entry {} uses unsupported compression method {}", entry.path.display(), entry.compression_method))); + }; + output.flush()?; + if decoded_size != entry.decoded_size { + return Err(invalid(format!("entry {} size mismatch: expected {}, decoded {decoded_size}", entry.path.display(), entry.decoded_size))); + } + if crc != entry.crc { + return Err(invalid(format!("entry {} CRC32 mismatch: expected {:08x}, decoded {crc:08x}", entry.path.display(), entry.crc))); + } + Ok(()) +} + +fn uses_intra_entry(entry: &Entry, entry_count: usize) -> bool { + if entry.method != CompressionMethod::DEFLATE { + return false; + } + let threshold = if entry_count == 1 { SINGLE_ENTRY_INTRA_THRESHOLD } else { MULTI_ENTRY_INTRA_THRESHOLD }; + entry.compressed_size >= threshold +} + +fn run_entries

( + entries: &[Entry], + source_len: u64, + options: DecodeOptions, + run: impl Fn(&Entry, DecodeOptions, &mut dyn FnMut(DecodeProgress)) -> Result<()> + Sync, + progress: P, +) -> Result<()> +where + P: FnMut(DecodeProgress) + Send, +{ + let threads = options.resolved_threads(); + let compressed = AtomicU64::new(0); + let decoded = AtomicU64::new(0); + let completed = AtomicUsize::new(0); + let progress = Mutex::new(progress); + let run = |entry: &Entry, entry_options| { + let mut entry_compressed = 0; + let mut entry_decoded = 0; + let mut update = |entry_progress: DecodeProgress| { + let compressed_delta = entry_progress.compressed_bytes.saturating_sub(entry_compressed); + let decoded_delta = entry_progress.decoded_bytes.saturating_sub(entry_decoded); + entry_compressed = entry_progress.compressed_bytes; + entry_decoded = entry_progress.decoded_bytes; + let compressed_bytes = compressed.fetch_add(compressed_delta, Ordering::Relaxed) + compressed_delta; + let decoded_bytes = decoded.fetch_add(decoded_delta, Ordering::Relaxed) + decoded_delta; + progress.lock().unwrap_or_else(std::sync::PoisonError::into_inner)(DecodeProgress { compressed_bytes, decoded_bytes }); + }; + run(entry, entry_options, &mut update)?; + let compressed_delta = entry.compressed_size.saturating_sub(entry_compressed); + let decoded_delta = entry.decoded_size.saturating_sub(entry_decoded); + let compressed = compressed.fetch_add(compressed_delta, Ordering::Relaxed) + compressed_delta; + let decoded = decoded.fetch_add(decoded_delta, Ordering::Relaxed) + decoded_delta; + let completed = completed.fetch_add(1, Ordering::Relaxed) + 1; + let compressed_bytes = if completed == entries.len() { source_len } else { compressed }; + progress.lock().unwrap_or_else(std::sync::PoisonError::into_inner)(DecodeProgress { compressed_bytes, decoded_bytes: decoded }); + Ok(()) + }; + let (within, across): (Vec<_>, Vec<_>) = entries.iter().partition(|entry| uses_intra_entry(entry, entries.len())); + for entry in within { + run(entry, DecodeOptions { threads, ..options })?; + } + if across.is_empty() { + return Ok(()); + } + let pool = rayon::ThreadPoolBuilder::new() + .num_threads(threads) + .thread_name(|index| format!("fastbz2-zip-{index}")) + .build() + .map_err(|error| invalid(error.to_string()))?; + pool.install(|| across.par_iter().try_for_each(|entry| run(entry, DecodeOptions { threads: 1, ..options }))) +} + +pub(super) fn validate(data: &[u8], options: DecodeOptions, max_output: Option, progress: impl FnMut(DecodeProgress) + Send) -> Result { + let report = parse(data, max_output)?; + run_entries( + &report.entries, + report.source_len, + options, + |entry, entry_options, progress| { + let mut output = WriterSink::new(io::sink()); + decode_entry(data, entry, &mut output, entry_options, progress) + }, + progress, + )?; + Ok(report) +} + +fn prepare_directories(root: &Path, entries: &[Entry]) -> Result<()> { + for entry in entries { + let path = root.join(&entry.path); + if entry.kind == EntryKind::Directory { + fs::create_dir_all(&path)?; + } else if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + } + Ok(()) +} + +#[cfg(unix)] +fn create_symlink(root: &Path, entry: &Entry, target: &[u8]) -> Result<()> { + use std::{ + ffi::OsStr, + os::unix::{ffi::OsStrExt, fs::symlink}, + }; + + let target = Path::new(OsStr::from_bytes(target)); + if target.is_absolute() { + return Err(invalid(format!("entry {} has absolute symlink target {}", entry.path.display(), target.display()))); + } + let mut depth = entry.path.parent().map_or(0, |parent| parent.components().count()); + for component in target.components() { + match component { + Component::Normal(_) => depth += 1, + Component::CurDir => {} + Component::ParentDir if depth != 0 => depth -= 1, + Component::ParentDir | Component::RootDir | Component::Prefix(_) => { + return Err(invalid(format!("entry {} has escaping symlink target {}", entry.path.display(), target.display()))); + } + } + } + symlink(target, root.join(&entry.path))?; + Ok(()) +} + +#[cfg(not(unix))] +fn create_symlink(_root: &Path, entry: &Entry, _target: &[u8]) -> Result<()> { + Err(invalid(format!("symlink entry {} is not supported on this platform", entry.path.display()))) +} + +fn extract_entry(data: &[u8], root: &Path, entry: &Entry, options: DecodeOptions, progress: &mut dyn FnMut(DecodeProgress)) -> Result<()> { + if entry.kind == EntryKind::Directory { + return Ok(()); + } + let path = root.join(&entry.path); + if entry.kind == EntryKind::Symlink { + if entry.decoded_size > 64 * 1024 { + return Err(invalid(format!("symlink target in {} is too large", entry.path.display()))); + } + let mut target = Vec::with_capacity(entry.decoded_size as usize); + let mut output = WriterSink::new(&mut target); + decode_entry(data, entry, &mut output, options, progress)?; + return create_symlink(root, entry, &target); + } + let mut file = fs::OpenOptions::new().write(true).create_new(true).open(&path)?; + let mut output = WriterSink::new(&mut file); + decode_entry(data, entry, &mut output, options, progress)?; + set_mode(&path, entry.mode)?; + set_modified(&path, entry.modified)?; + Ok(()) +} + +fn set_modified(path: &Path, modified: Option) -> Result<()> { + if let Some(modified) = modified { + fs::File::open(path)?.set_times(fs::FileTimes::new().set_modified(modified))?; + } + Ok(()) +} + +#[cfg(unix)] +fn set_mode(path: &Path, mode: Option) -> Result<()> { + use std::os::unix::fs::PermissionsExt; + if let Some(mode) = mode { + fs::set_permissions(path, fs::Permissions::from_mode(mode & 0o7777))?; + } + Ok(()) +} + +#[cfg(not(unix))] +fn set_mode(_path: &Path, _mode: Option) -> Result<()> { + Ok(()) +} + +pub(super) fn unpack( + data: &[u8], + destination: &Path, + overwrite: bool, + options: DecodeOptions, + max_output: Option, + progress: impl FnMut(DecodeProgress) + Send, +) -> Result { + let report = parse(data, max_output)?; + let staging = archive_extract::staging(destination)?; + prepare_directories(staging.path(), &report.entries)?; + run_entries( + &report.entries, + report.source_len, + options, + |entry, entry_options, progress| extract_entry(data, staging.path(), entry, entry_options, progress), + progress, + )?; + let mut directories: Vec<_> = report.entries.iter().filter(|entry| entry.kind == EntryKind::Directory).collect(); + directories.sort_unstable_by_key(|entry| std::cmp::Reverse(entry.path.components().count())); + for entry in directories { + set_mode(&staging.path().join(&entry.path), entry.mode)?; + set_modified(&staging.path().join(&entry.path), entry.modified)?; + } + archive_extract::commit(staging.path(), destination, overwrite)?; + Ok(report) +} diff --git a/src/deflate.rs b/src/deflate.rs new file mode 100644 index 0000000..4b29d79 --- /dev/null +++ b/src/deflate.rs @@ -0,0 +1,16 @@ +//! Raw DEFLATE decompression shared by gzip and ZIP framing. + +use crate::{DecodeOptions, DecodeProgress, OutputSink, Result, gzip}; + +pub use gzip::DeflateReport as Report; + +/// Decode one raw DEFLATE stream into an output sink. +#[doc(hidden)] +pub fn decompress_to_sink_with_options_and_progress( + data: &[u8], + output: &mut impl OutputSink, + options: DecodeOptions, + progress: impl FnMut(DecodeProgress), +) -> Result { + gzip::decompress_deflate_to_sink_with_options_and_progress(data, output, options, progress) +} diff --git a/src/error.rs b/src/error.rs index d79ffaf..60d179f 100644 --- a/src/error.rs +++ b/src/error.rs @@ -38,6 +38,7 @@ pub enum Error { UnexpectedEof { bit_offset: u64, requested: u64, remaining: u64 }, InvalidStreamHeader, InvalidGzip(String), + InvalidZip(String), Decode { bit_offset: u64, source: DecodeError }, InvalidIndex(String), InvalidConfiguration(String), @@ -56,6 +57,7 @@ impl fmt::Display for Error { } Self::InvalidStreamHeader => write!(f, "input does not start with a bzip2 BZh1-BZh9 header"), Self::InvalidGzip(message) => write!(f, "invalid gzip stream: {message}"), + Self::InvalidZip(message) => write!(f, "invalid ZIP archive: {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 index dabe263..07fac3c 100644 --- a/src/gzip.rs +++ b/src/gzip.rs @@ -75,6 +75,17 @@ pub struct Report { pub fallback_chunks: u64, } +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct DeflateReport { + pub source_len: u64, + pub compressed_end_bit: u64, + pub decoded_len: u64, + pub crc: u32, + pub blocks: Vec, + pub speculative_chunks: u64, + pub fallback_chunks: u64, +} + #[derive(Clone, Debug)] struct Header { deflate_start: usize, @@ -125,26 +136,13 @@ pub fn decompress_to_sink_with_options_and_progress( 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_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_sink_with_progress(data, output, progress), - }; - decompress_parallel_to_sink(data, output, options, threads, &mut progress, initial_segment) -} - -fn decompress_serial_to_sink_with_progress(data: &[u8], output: &mut impl OutputSink, mut progress: impl FnMut(DecodeProgress)) -> Result { + let options = options.validate()?; let mut members = Vec::new(); let mut blocks = Vec::new(); let mut position = 0_usize; let mut decoded_total = 0_u64; + let mut speculative_total = 0_u64; + let mut fallback_total = 0_u64; while position < data.len() { if !members.is_empty() && data[position..].iter().all(|&byte| byte == 0) { @@ -153,30 +151,30 @@ fn decompress_serial_to_sink_with_progress(data: &[u8], output: &mut impl Output 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 stream = DeflateStream { start_byte: header.deflate_start, end_byte: data.len(), member: member_number, decoded_base: decoded_total }; + let decoded = decompress_deflate_stream(data, stream, output, options, &mut progress)?; + let trailer = (decoded.compressed_end_bit as usize).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 (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.crc != expected_crc { + return Err(invalid_at(trailer, format!("CRC32 mismatch: expected {expected_crc:08x}, decoded {:08x}", decoded.crc))); } - if decoded_len as u32 != expected_size { - return Err(invalid_at(trailer + 4, format!("ISIZE mismatch: expected {expected_size}, decoded {}", decoded_len as u32))); + if decoded.decoded_len as u32 != expected_size { + return Err(invalid_at(trailer + 4, format!("ISIZE mismatch: expected {expected_size}, decoded {}", decoded.decoded_len as u32))); } - decoded_total = decoded_total.checked_add(decoded_len).ok_or_else(|| invalid("decoded offset overflow"))?; + decoded_total = decoded_total.checked_add(decoded.decoded_len).ok_or_else(|| invalid("decoded offset overflow"))?; + speculative_total += decoded.speculative_chunks; + fallback_total += decoded.fallback_chunks; + blocks.extend(decoded.blocks); 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, + decoded_start: decoded_total - decoded.decoded_len, + decoded_len: decoded.decoded_len, expected_crc, mtime: header.mtime, extra_flags: header.extra_flags, @@ -191,7 +189,33 @@ fn decompress_serial_to_sink_with_progress(data: &[u8], output: &mut impl Output } 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 }) + Ok(Report { + source_len: data.len() as u64, + decoded_len: decoded_total, + members, + blocks, + speculative_chunks: speculative_total, + fallback_chunks: fallback_total, + }) +} + +/// Decode one raw DEFLATE stream into an output that can take ownership of completed chunks. +#[doc(hidden)] +pub(crate) fn decompress_deflate_to_sink_with_options_and_progress( + data: &[u8], + output: &mut impl OutputSink, + options: DecodeOptions, + mut progress: impl FnMut(DecodeProgress), +) -> Result { + let options = options.validate()?; + let stream = DeflateStream { start_byte: 0, end_byte: data.len(), member: 0, decoded_base: 0 }; + let report = decompress_deflate_stream(data, stream, output, options, &mut progress)?; + if (report.compressed_end_bit as usize).div_ceil(8) != data.len() { + return Err(invalid_bit(report.compressed_end_bit as usize, "trailing data after final DEFLATE block")); + } + output.flush()?; + progress(DecodeProgress { compressed_bytes: data.len() as u64, decoded_bytes: report.decoded_len }); + Ok(report) } #[derive(Clone, Copy)] @@ -551,218 +575,173 @@ impl<'a, W: OutputSink + ?Sized, P: FnMut(DecodeProgress) + ?Sized> SegmentCommi } } -fn decompress_parallel_to_sink( +#[derive(Clone, Copy)] +struct DeflateStream { + start_byte: usize, + end_byte: usize, + member: u32, + decoded_base: u64, +} + +fn decompress_deflate_stream( + data: &[u8], + stream: DeflateStream, + output: &mut impl OutputSink, + mut options: DecodeOptions, + progress: &mut impl FnMut(DecodeProgress), +) -> Result { + let DeflateStream { start_byte, end_byte, member, decoded_base } = stream; + let data = data.get(..end_byte).ok_or_else(|| invalid("DEFLATE end exceeds input"))?; + if start_byte > end_byte { + return Err(invalid("DEFLATE start exceeds end")); + } + let threads = options.resolved_threads(); + options.threads = threads; + if threads == 1 || end_byte - start_byte < MIN_PARALLEL_INPUT || options.memory_limit < PARALLEL_JOB_MEMORY { + return decompress_deflate_serial_stream(data, start_byte, output, member, decoded_base, progress); + } + let first_grid = start_byte.saturating_add(PARALLEL_GRID); + let initial_segment = match decode_segment(data, start_byte * 8, first_grid.min(end_byte) * 8, InitialHistory::Empty, PARALLEL_OUTPUT_LIMIT) { + Ok(segment) => segment, + Err(_) => return decompress_deflate_serial_stream(data, start_byte, output, member, decoded_base, progress), + }; + decompress_deflate_parallel_stream(data, start_byte, end_byte, output, options, threads, member, decoded_base, progress, initial_segment) +} + +fn decompress_deflate_serial_stream( + data: &[u8], + start_byte: usize, + output: &mut impl OutputSink, + member: u32, + decoded_base: u64, + progress: &mut impl FnMut(DecodeProgress), +) -> Result { + let mut blocks = Vec::new(); + let mut emitter = Emitter::new(output, decoded_base); + let mut bits = Bits::new(data, start_byte); + decode_deflate(&mut bits, &mut emitter, member, &mut blocks, progress)?; + let compressed_end_bit = bits.position_bits() as u64; + let (crc, decoded_len) = emitter.finish()?; + Ok(DeflateReport { source_len: (data.len() - start_byte) as u64, compressed_end_bit, decoded_len, crc, blocks, speculative_chunks: 0, fallback_chunks: 0 }) +} + +#[allow(clippy::too_many_arguments)] +fn decompress_deflate_parallel_stream( data: &[u8], + start_byte: usize, + end_byte: usize, output: &mut impl OutputSink, options: DecodeOptions, threads: usize, + member: u32, + decoded_base: u64, 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_sink_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, - }); - } +) -> Result { + let per_job = PARALLEL_JOB_MEMORY; + let output_limit = PARALLEL_OUTPUT_LIMIT; + let horizon = threads.saturating_add(2); + let parallel_budget = options.memory_limit.min(per_job.saturating_mul(horizon)); + let first_grid = start_byte.saturating_add(PARALLEL_GRID); + let mut jobs = Vec::new(); + let mut key = 1; + let mut grid = first_grid; + while grid < end_byte { + jobs.push(Job { + key, + reservation: per_job, + payload: GzipJob { + search_start: grid * 8, + search_end: grid.saturating_add(2 * PARALLEL_GRID).min(end_byte) * 8, + stop_bit: grid.saturating_add(PARALLEL_GRID).min(end_byte) * 8, + output_limit, }, - }; - 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; + }); + key += 1; + grid = grid.saturating_add(PARALLEL_GRID); + } - 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)?; + let mut blocks = Vec::new(); + let (compressed_end_bit, decoded_len, 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, decoded_base, 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 = initial_segment; + 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 = start_byte.saturating_add((key + 1) * PARALLEL_GRID).min(end_byte) * 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)?; - predecessor = next_window; - key += 1; - continue; + next_resolve += 1; + outstanding -= 1; } - - speculative_chunks += 1; - let segment = speculative.unwrap(); + 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)?; - results.submit(resolve_sequence, lease, ResolveTask { segment, predecessor })?; + let resolved = resolve_segment(segment, &predecessor)?; + committer.commit(resolved)?; 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; - } + continue; } - while outstanding != 0 { + 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; } + } - 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 }); - } + while outstanding != 0 { + let resolved = results.take_stage(next_resolve)??; + committer.commit(resolved)?; + next_resolve += 1; + outstanding -= 1; + } - 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, - }) + let SegmentCommitter { decoded, crc, .. } = committer; + Ok((next_start as u64, decoded, crc.finalize(), speculative_chunks, fallback_chunks)) + }, + )?; + Ok(DeflateReport { source_len: (end_byte - start_byte) as u64, compressed_end_bit, decoded_len, crc, blocks, speculative_chunks, fallback_chunks }) } fn parse_header(data: &[u8], start: usize) -> Result

{ @@ -1008,11 +987,17 @@ impl Huffman { 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]; + let remaining = bits.data.len().saturating_mul(8).saturating_sub(bits.bit); + let peek_bits = usize::from(self.max_bits).min(remaining) as u8; + let packed = self.table[bits.peek(peek_bits)? as usize]; if packed == u16::MAX { return Err(invalid_bit(bits.position_bits(), "invalid Huffman code")); } - bits.drop((packed >> 9) as u8); + let length = (packed >> 9) as u8; + if usize::from(length) > remaining { + return Err(invalid_bit(bits.position_bits(), "truncated Huffman code")); + } + bits.drop(length); Ok(packed & 0x01ff) } } @@ -1044,10 +1029,6 @@ impl<'a> Bits<'a> { 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); @@ -1238,7 +1219,10 @@ fn invalid_bit(bit: usize, message: impl Into) -> Error { mod tests { use std::io::Write as _; - use flate2::{Compression, GzBuilder, write::GzEncoder}; + use flate2::{ + Compression, GzBuilder, + write::{DeflateEncoder, GzEncoder}, + }; use super::*; @@ -1252,6 +1236,37 @@ mod tests { encoder.finish().unwrap() } + fn compress_raw(data: &[u8]) -> Vec { + let mut encoder = DeflateEncoder::new(Vec::new(), Compression::new(6)); + encoder.write_all(data).unwrap(); + encoder.finish().unwrap() + } + + #[test] + fn raw_deflate_reuses_serial_and_parallel_decoder() { + let plain = patterned(20 * 1024 * 1024); + let compressed = compress_raw(&plain); + for threads in [1, 4] { + let mut output = Vec::new(); + let mut sink = WriterSink::new(&mut output); + let report = + decompress_deflate_to_sink_with_options_and_progress(&compressed, &mut sink, DecodeOptions { threads, ..DecodeOptions::default() }, |_| {}) + .unwrap(); + assert_eq!(output, plain); + assert_eq!(report.decoded_len, plain.len() as u64); + assert_eq!(report.crc, crc32(&plain)); + } + for size in 0..256 { + let plain = patterned(size); + let compressed = compress_raw(&plain); + let mut output = Vec::new(); + let mut sink = WriterSink::new(&mut output); + decompress_deflate_to_sink_with_options_and_progress(&compressed, &mut sink, DecodeOptions { threads: 1, ..DecodeOptions::default() }, |_| {}) + .unwrap(); + assert_eq!(output, plain); + } + } + #[test] fn decodes_stored_fixed_and_dynamic_blocks() { let cases = diff --git a/src/lib.rs b/src/lib.rs index c1a0dd0..5a47ff2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,6 +5,7 @@ mod block; mod crc; mod decode; mod decoder; +pub mod deflate; mod error; mod format; pub mod gzip; diff --git a/tests/archive_perf.rs b/tests/archive_perf.rs index d97998f..d64a227 100644 --- a/tests/archive_perf.rs +++ b/tests/archive_perf.rs @@ -6,9 +6,13 @@ use std::{ }; use crabz2::{Level, compress}; -use fastbz2::{DecodeOptions, OutputSink, decompress, gzip as gzip_decoder}; +use fastbz2::{DecodeOptions, OutputSink, gzip as gzip_decoder}; use flate2::{Compression, write::GzEncoder}; +#[allow(dead_code)] +mod support; +use support::simplewiki_prefix; + fn requested_threads() -> usize { std::env::var("FASTBZ2_THREADS").ok().map(|value| value.parse().expect("FASTBZ2_THREADS must be an integer")).unwrap_or(0) } @@ -19,14 +23,6 @@ fn binary() -> Command { 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(); { diff --git a/tests/cli.rs b/tests/cli.rs index a7a7c61..ec8a1dd 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -1,6 +1,6 @@ use std::{ fs, - io::Write, + io::{Cursor, Write}, path::Path, process::{Command, Stdio}, time::{Duration, UNIX_EPOCH}, @@ -8,6 +8,11 @@ use std::{ use crabz2::{Level, compress}; use flate2::{Compression, write::GzEncoder}; +use zip::{CompressionMethod as ZipCompression, ZipArchive, ZipWriter, write::FullFileOptions}; + +#[allow(dead_code)] +mod support; +use support::{ZipMethod, zip_bytes, zip_with_modes}; fn binary() -> Command { Command::new(env!("CARGO_BIN_EXE_fastbz2")) @@ -26,6 +31,26 @@ fn gzip_bytes(plain: &[u8]) -> Vec { fn write_gzip(path: &Path, plain: &[u8]) { fs::write(path, gzip_bytes(plain)).unwrap(); } + +fn linked_zip() -> Vec { + zip_with_modes(&[ + ("nested/", b"", ZipMethod::Stored, 0o040750), + ("nested/root.txt", b"linked contents", ZipMethod::Deflate, 0o100640), + ("nested/symbolic.txt", b"root.txt", ZipMethod::Deflate, 0o120777), + ]) +} + +fn streaming_zip64() -> Vec { + let mut archive = ZipWriter::new_stream(Vec::new()); + let modified = 1_700_000_123_u32; + let mut timestamp = vec![1]; + timestamp.extend_from_slice(&modified.to_le_bytes()); + let mut options = FullFileOptions::default().compression_method(ZipCompression::STORE).large_file(true).unix_permissions(0o600); + options.add_extra_data(0x5455, timestamp, true).unwrap(); + archive.start_file("zip64.txt", options).unwrap(); + archive.write_all(b"small payload with ZIP64 fields and a data descriptor").unwrap(); + archive.finish().unwrap().into_inner() +} fn tar_bytes(entries: &[(&str, &[u8])]) -> Vec { let mut archive = Vec::new(); { @@ -466,3 +491,122 @@ fn tar_staging_preserves_symbolic_and_hard_links() { 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); } + +#[test] +#[cfg(unix)] +fn zip_auto_extracts_validates_lists_and_preserves_links() { + use std::os::unix::fs::PermissionsExt; + + let directory = tempfile::tempdir().unwrap(); + let input = directory.path().join("mixed.zip"); + let output = directory.path().join("output"); + fs::write(&input, linked_zip()).unwrap(); + + let tested = binary().args(["--test", input.to_str().unwrap()]).output().unwrap(); + assert!(tested.status.success(), "{}", String::from_utf8_lossy(&tested.stderr)); + + let listed = binary().args(["--list", "--json", input.to_str().unwrap()]).output().unwrap(); + assert!(listed.status.success(), "{}", String::from_utf8_lossy(&listed.stderr)); + let listing: serde_json::Value = serde_json::from_slice(&listed.stdout).unwrap(); + assert_eq!(listing["format"], "zip"); + assert_eq!(listing["entries"].as_array().unwrap().len(), 3); + + let extracted = binary().args(["-C", output.to_str().unwrap(), input.to_str().unwrap()]).output().unwrap(); + assert!(extracted.status.success(), "{}", String::from_utf8_lossy(&extracted.stderr)); + assert_eq!(fs::read(output.join("nested/root.txt")).unwrap(), b"linked contents"); + assert_eq!(fs::read_link(output.join("nested/symbolic.txt")).unwrap(), Path::new("root.txt")); + assert_eq!(fs::metadata(output.join("nested/root.txt")).unwrap().permissions().mode() & 0o777, 0o640); + assert_eq!(fs::metadata(output.join("nested")).unwrap().permissions().mode() & 0o777, 0o750); + + let raw = binary().args([input.to_str().unwrap(), "-o", directory.path().join("raw").to_str().unwrap()]).output().unwrap(); + assert_eq!(raw.status.code(), Some(2)); +} + +#[test] +fn zip_stored_deflate_limits_corruption_and_paths_are_atomic() { + let directory = tempfile::tempdir().unwrap(); + let input = directory.path().join("data.zip"); + let plain = b"parallel deflate entry".repeat(10_000); + let stored = b"stored bytes".repeat(1_000); + let archive = zip_bytes(&[("deflated.txt", &plain, ZipMethod::Deflate), ("stored.bin", &stored, ZipMethod::Stored)]); + fs::write(&input, &archive).unwrap(); + + let limited = directory.path().join("limited"); + let result = binary() + .args(["--max-output", &(plain.len() + stored.len() - 1).to_string(), "-C", limited.to_str().unwrap(), input.to_str().unwrap()]) + .output() + .unwrap(); + assert_eq!(result.status.code(), Some(3)); + assert_eq!(fs::read_dir(&limited).unwrap().count(), 0); + + let mut corrupt = archive.clone(); + let data_byte = { + let mut parsed = ZipArchive::new(Cursor::new(corrupt.as_slice())).unwrap(); + let entry = parsed.by_index_raw(0).unwrap(); + entry.data_start().unwrap() as usize + entry.compressed_size() as usize / 2 + }; + corrupt[data_byte] ^= 0x40; + fs::write(&input, corrupt).unwrap(); + let output = directory.path().join("corrupt-output"); + let result = binary().args(["-C", output.to_str().unwrap(), input.to_str().unwrap()]).output().unwrap(); + assert_eq!(result.status.code(), Some(3)); + assert_eq!(fs::read_dir(&output).unwrap().count(), 0); + + let mut traversal = zip_bytes(&[("aa/x.txt", b"outside", ZipMethod::Deflate)]); + for offset in 0..=traversal.len() - b"aa/x.txt".len() { + if &traversal[offset..offset + b"aa/x.txt".len()] == b"aa/x.txt" { + traversal[offset..offset + b"../x.txt".len()].copy_from_slice(b"../x.txt"); + } + } + fs::write(&input, traversal).unwrap(); + let traversal_output = directory.path().join("traversal-output"); + let result = binary().args(["-C", traversal_output.to_str().unwrap(), input.to_str().unwrap()]).output().unwrap(); + assert_eq!(result.status.code(), Some(3)); + assert_eq!(fs::read_dir(&traversal_output).unwrap().count(), 0); + assert!(!directory.path().join("x.txt").exists()); +} + +#[test] +fn zip64_and_streaming_data_descriptors_use_the_same_extractor() { + let directory = tempfile::tempdir().unwrap(); + let input = directory.path().join("streaming.zip"); + let output = directory.path().join("output"); + fs::write(&input, streaming_zip64()).unwrap(); + let result = binary().args(["-C", output.to_str().unwrap(), input.to_str().unwrap()]).output().unwrap(); + assert!(result.status.success(), "{}", String::from_utf8_lossy(&result.stderr)); + let extracted = output.join("zip64.txt"); + assert_eq!(fs::read(&extracted).unwrap(), b"small payload with ZIP64 fields and a data descriptor"); + assert_eq!(fs::metadata(extracted).unwrap().modified().unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs(), 1_700_000_123); +} + +#[test] +fn zip_rejects_encryption_unsupported_methods_and_duplicate_paths() { + fn header(archive: &[u8], signature: &[u8; 4]) -> usize { + archive.windows(signature.len()).position(|bytes| bytes == signature).unwrap() + } + fn rejected(directory: &Path, name: &str, archive: &[u8], expected: &str) { + let input = directory.join(name); + fs::write(&input, archive).unwrap(); + let result = binary().args(["--test", input.to_str().unwrap()]).output().unwrap(); + assert_eq!(result.status.code(), Some(3), "{name}: {}", String::from_utf8_lossy(&result.stderr)); + assert!(String::from_utf8_lossy(&result.stderr).contains(expected), "{}", String::from_utf8_lossy(&result.stderr)); + } + + let directory = tempfile::tempdir().unwrap(); + let mut encrypted = zip_bytes(&[("secret.txt", b"not actually encrypted", ZipMethod::Stored)]); + let local = header(&encrypted, b"PK\x03\x04"); + let central = header(&encrypted, b"PK\x01\x02"); + encrypted[local + 6] |= 1; + encrypted[central + 8] |= 1; + rejected(directory.path(), "encrypted.zip", &encrypted, "encrypted entry"); + + let mut unsupported = zip_bytes(&[("modern.txt", b"unsupported codec", ZipMethod::Stored)]); + let local = header(&unsupported, b"PK\x03\x04"); + let central = header(&unsupported, b"PK\x01\x02"); + unsupported[local + 8..local + 10].copy_from_slice(&93_u16.to_le_bytes()); + unsupported[central + 10..central + 12].copy_from_slice(&93_u16.to_le_bytes()); + rejected(directory.path(), "unsupported.zip", &unsupported, "unsupported compression method 93"); + + let duplicate = zip_bytes(&[("same.txt", b"first", ZipMethod::Deflate), ("same.txt", b"second", ZipMethod::Deflate)]); + rejected(directory.path(), "duplicate.zip", &duplicate, "central directory contains duplicate entry names"); +} diff --git a/tests/common/process.rs b/tests/common/process.rs index 2bfc576..d20150c 100644 --- a/tests/common/process.rs +++ b/tests/common/process.rs @@ -58,7 +58,9 @@ fn physical_footprint(pid: libc::pid_t) -> Option { // 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]) + // `values` starts immediately after `ri_uuid`; in rusage_info_v4, + // ri_phys_footprint is the eighth u64 field. + (result == 0).then_some(usage.values[7]) } #[derive(Clone, Copy, Debug)] diff --git a/tests/support/mod.rs b/tests/support/mod.rs new file mode 100644 index 0000000..7f5afc2 --- /dev/null +++ b/tests/support/mod.rs @@ -0,0 +1,88 @@ +use std::{fs, io::Write, path::Path}; + +use fastbz2::{DecodeOptions, decompress}; +use flate2::{Compression, write::DeflateEncoder}; + +#[derive(Clone, Copy)] +pub(crate) enum ZipMethod { + #[allow(dead_code)] + Stored, + Deflate, +} + +pub(crate) fn simplewiki_prefix() -> Vec { + let 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 push_u16(output: &mut Vec, value: u16) { + output.extend_from_slice(&value.to_le_bytes()); +} + +fn push_u32(output: &mut Vec, value: u32) { + output.extend_from_slice(&value.to_le_bytes()); +} + +pub(crate) fn zip_with_modes(entries: &[(&str, &[u8], ZipMethod, u32)]) -> Vec { + let mut archive = Vec::new(); + let mut central = Vec::new(); + for (path, contents, method, mode) in entries { + let method_id = match method { + ZipMethod::Stored => 0, + ZipMethod::Deflate => 8, + }; + let compressed = match method { + ZipMethod::Stored => contents.to_vec(), + ZipMethod::Deflate => { + let mut encoder = DeflateEncoder::new(Vec::new(), Compression::new(6)); + encoder.write_all(contents).unwrap(); + encoder.finish().unwrap() + } + }; + let offset = archive.len() as u32; + let crc = crc32fast::hash(contents); + push_u32(&mut archive, 0x0403_4b50); + for value in [20, 0, method_id, 0, 0] { + push_u16(&mut archive, value); + } + for value in [crc, compressed.len() as u32, contents.len() as u32] { + push_u32(&mut archive, value); + } + push_u16(&mut archive, path.len() as u16); + push_u16(&mut archive, 0); + archive.extend_from_slice(path.as_bytes()); + archive.extend_from_slice(&compressed); + + push_u32(&mut central, 0x0201_4b50); + for value in [0x0314, 20, 0, method_id, 0, 0] { + push_u16(&mut central, value); + } + for value in [crc, compressed.len() as u32, contents.len() as u32] { + push_u32(&mut central, value); + } + for value in [path.len() as u16, 0, 0, 0, 0] { + push_u16(&mut central, value); + } + push_u32(&mut central, mode << 16); + push_u32(&mut central, offset); + central.extend_from_slice(path.as_bytes()); + } + let central_offset = archive.len() as u32; + archive.extend_from_slice(¢ral); + push_u32(&mut archive, 0x0605_4b50); + for value in [0, 0, entries.len() as u16, entries.len() as u16] { + push_u16(&mut archive, value); + } + push_u32(&mut archive, central.len() as u32); + push_u32(&mut archive, central_offset); + push_u16(&mut archive, 0); + archive +} + +pub(crate) fn zip_bytes(entries: &[(&str, &[u8], ZipMethod)]) -> Vec { + let entries: Vec<_> = entries.iter().map(|(path, contents, method)| (*path, *contents, *method, 0o100640)).collect(); + zip_with_modes(&entries) +} diff --git a/tests/zip_perf.rs b/tests/zip_perf.rs new file mode 100644 index 0000000..a33ed02 --- /dev/null +++ b/tests/zip_perf.rs @@ -0,0 +1,162 @@ +#[allow(dead_code, unused_imports)] +mod common; +mod support; + +use std::{ + fs, + path::PathBuf, + process::Command, + time::{Duration, Instant}, +}; + +use support::{ZipMethod, simplewiki_prefix, zip_bytes}; + +const ENTRY_COUNT: usize = 18; + +fn requested_threads() -> usize { + std::env::var("FASTBZ2_THREADS").ok().map(|value| value.parse().expect("FASTBZ2_THREADS must be an integer")).unwrap_or(0) +} + +#[derive(Clone, Copy)] +enum Shape { + Single, + Many, +} + +struct Fixture { + directory: tempfile::TempDir, + input: PathBuf, + contents: Vec, + chunk_size: usize, + shape: Shape, +} + +impl Fixture { + fn new(shape: Shape) -> Self { + let directory = tempfile::tempdir().unwrap(); + let contents = simplewiki_prefix(); + let (archive, chunk_size) = match shape { + Shape::Single => (zip_bytes(&[("payload.bin", &contents, ZipMethod::Deflate)]), contents.len()), + Shape::Many => { + let chunk_size = contents.len().div_ceil(ENTRY_COUNT); + let names: Vec<_> = (0..ENTRY_COUNT).map(|index| format!("parts/{index:02}.bin")).collect(); + let entries: Vec<_> = + contents.chunks(chunk_size).enumerate().map(|(index, chunk)| (names[index].as_str(), chunk, ZipMethod::Deflate)).collect(); + (zip_bytes(&entries), chunk_size) + } + }; + let input = directory.path().join("fixture.zip"); + fs::write(&input, &archive).unwrap(); + eprintln!( + "ZIP fixture: {} entries, {:.1} MiB compressed, {:.1} MiB decoded", + match shape { + Shape::Single => 1, + Shape::Many => ENTRY_COUNT, + }, + archive.len() as f64 / 1_048_576.0, + contents.len() as f64 / 1_048_576.0 + ); + Self { directory, input, contents, chunk_size, shape } + } + + fn verify(&self, output: &std::path::Path) { + match self.shape { + Shape::Single => assert_eq!(fs::read(output.join("payload.bin")).unwrap(), self.contents), + Shape::Many => { + for (index, expected) in self.contents.chunks(self.chunk_size).enumerate() { + assert_eq!(fs::read(output.join(format!("parts/{index:02}.bin"))).unwrap(), expected); + } + } + } + } +} + +fn fastbz2_command(input: &std::path::Path, output: &std::path::Path) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_fastbz2")); + command.args(["-q", "-P", &requested_threads().to_string(), "-C"]).arg(output).arg(input); + command +} + +fn unzip_command(input: &std::path::Path, output: &std::path::Path) -> Command { + let mut command = Command::new("unzip"); + command.args(["-qq"]).arg(input).args(["-d"]).arg(output); + command +} + +fn timed(command: &mut Command) -> Duration { + let start = Instant::now(); + assert!(command.status().unwrap().success()); + start.elapsed() +} + +fn benchmark(shape: Shape, fastbz2: bool) { + let fixture = Fixture::new(shape); + let warm = fixture.directory.path().join("warm"); + let measured = fixture.directory.path().join("measured"); + let warm_time = if fastbz2 { timed(&mut fastbz2_command(&fixture.input, &warm)) } else { timed(&mut unzip_command(&fixture.input, &warm)) }; + let elapsed = if fastbz2 { timed(&mut fastbz2_command(&fixture.input, &measured)) } else { timed(&mut unzip_command(&fixture.input, &measured)) }; + fixture.verify(&measured); + eprintln!("{}: warm {warm_time:.3?}, measured {elapsed:.3?}", if fastbz2 { "fastbz2" } else { "Info-ZIP unzip 6.00 (Apple)" }); +} + +#[test] +#[ignore = "local single-run one-entry ZIP extraction benchmark"] +fn zip_single_fastbz2() { + benchmark(Shape::Single, true); +} + +#[test] +#[ignore = "local single-run one-entry Info-ZIP baseline"] +fn zip_single_unzip() { + benchmark(Shape::Single, false); +} + +#[test] +#[ignore = "local single-run many-entry ZIP extraction benchmark"] +fn zip_many_fastbz2() { + benchmark(Shape::Many, true); +} + +#[test] +#[ignore = "local single-run many-entry Info-ZIP baseline"] +fn zip_many_unzip() { + benchmark(Shape::Many, false); +} + +#[test] +#[cfg(unix)] +#[ignore = "local ZIP extraction time and peak-memory benchmark"] +fn zip_many_fastbz2_process_metrics() { + let fixture = Fixture::new(Shape::Many); + let output = fixture.directory.path().join("measured"); + let metrics = common::measure(&mut fastbz2_command(&fixture.input, &output)).unwrap(); + assert!(metrics.status.success()); + fixture.verify(&output); + eprintln!( + "fastbz2 ZIP: 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 / 1_048_576.0, + metrics.peak_phys_footprint_bytes.map_or(f64::NAN, |bytes| bytes as f64 / 1_048_576.0), + ); +} + +#[test] +#[cfg(unix)] +#[ignore = "local Info-ZIP extraction time and peak-memory baseline"] +fn zip_many_unzip_process_metrics() { + let fixture = Fixture::new(Shape::Many); + let output = fixture.directory.path().join("measured"); + let metrics = common::measure(&mut unzip_command(&fixture.input, &output)).unwrap(); + assert!(metrics.status.success()); + fixture.verify(&output); + eprintln!( + "Info-ZIP unzip 6.00 (Apple): 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 / 1_048_576.0, + metrics.peak_phys_footprint_bytes.map_or(f64::NAN, |bytes| bytes as f64 / 1_048_576.0), + ); +}