diff --git a/DEV.md b/DEV.md index a570ac4..79897e1 100644 --- a/DEV.md +++ b/DEV.md @@ -39,7 +39,7 @@ The decoder remains independent of files, Python, and the CLI. Parallel scanning Core decode APIs report completed compressed and decoded byte counts without knowing anything about terminals. The CLI selects bzip2, gzip, LZ4, 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. -The LZ4 decoder parses current frames, concatenation, and skippable frames itself. It validates descriptor bits and XXH32 header, block, and content checksums, and bounds every literal and match before writing. Independent compressed blocks become ordinary `pipeline::Job`s, reserving the frame's declared maximum decoded block size; stored blocks borrow their source bytes and reserve no decoded allocation. Frames containing only stored blocks without block checksums bypass the worker pool because their only remaining work is ordered output and optional content hashing. Retained results remain charged at their allocation capacity rather than logical length, so highly compressible blocks cannot understate memory use. The coordinator commits results in source order and updates the content checksum. Linked frames use the same parser, block decoder, output sink, progress, and report path, but decode serially with a rolling 64 KiB history. This is one code path with a scheduling branch, not separate serial and parallel implementations. Concatenated frames currently run in frame order; parallelising across small independent frames is a possible measured optimization, not pre-built machinery. +The LZ4 decoder parses current frames, concatenation, and skippable frames itself. It validates descriptor bits and XXH32 header, block, and content checksums, and bounds every literal and match before writing. A frame header creates an incremental block cursor rather than a complete layout. Independent blocks are gathered into at most 64-entry batches—only until there is enough work to amortize the pool—and become ordinary `pipeline::Job`s. A parse failure discovered during bounded look-ahead is held until every earlier valid block has decoded and emitted, preserving stream error order. Compressed jobs reserve the frame's declared maximum decoded block size; stored blocks borrow their source bytes and reserve no decoded allocation. Frames containing only stored blocks without block checksums bypass the worker pool because their only remaining work is ordered output and optional content hashing. Retained results remain charged at their allocation capacity rather than logical length, so highly compressible blocks cannot understate memory use. If fewer than two natural blocks fit the speculative budget, decoding proceeds incrementally on the coordinator instead of rejecting the frame. The pool is created lazily and reused across concatenated frames. Linked frames use the same parser, block decoder, output sink, progress, and report path, but decode serially with a rolling 64 KiB history. This is one code path with a scheduling branch, not separate Reader and writer implementations. 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. @@ -49,7 +49,7 @@ ZIP scheduling deliberately uses one parallelism level at a time. A sole DEFLATE 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. LZ4 independent blocks use the simpler ordered-job path and charge each retained allocation against the budget. Decode results and outstanding resolution results have bounded horizons, preventing dependency stalls from causing unbounded memory. -The shared `OutputSink` boundary accepts owned decoder chunks. Direct writer APIs adapt those chunks to `Write::write_all` without a channel or allocation copy. `Reader` and streaming tar extraction instead use the same zero-capacity owned-chunk pipe, so completed bzip2 blocks and resolved parallel-gzip segments move into the consumer rather than being copied into an intermediate pipe buffer. The pipe adds at most the consumer's current chunk and the producer's next blocked chunk beyond the scheduler budget. Its receiver owns a cancellation flag; parallel bzip2 scanning checks that flag between bounded waves. +The shared `OutputSink` boundary accepts owned decoder chunks. Direct writer APIs adapt those chunks to `Write::write_all` without a channel or allocation copy. `Reader` and streaming tar extraction instead use the same zero-capacity owned-chunk pipe, so completed bzip2 blocks, resolved parallel-gzip segments, and decoded LZ4 blocks move into the consumer rather than being copied into an intermediate pipe buffer. The pipe adds at most the consumer's current chunk and the producer's next blocked chunk beyond the scheduler budget. Its receiver owns a cancellation flag; bzip2 scanning checks that flag between bounded waves, while LZ4 checks it before parsing each serial block or parallel batch. The non-indexing parallel bzip2 path decodes and emits its first CRC-validated block before scanning the remainder of the compressed source. That candidate is then reused by normal ordered assembly rather than decoded twice. Index construction retains the full scan-first path because it produces no output. `Reader` sends decoder success or failure separately from the byte pipe and treats a missing terminal status as an error, so a worker panic or corrupt trailer cannot appear as EOF. Reader errors are sticky across subsequent calls, and drop disconnects the pipe before joining the decoder thread. @@ -81,18 +81,20 @@ The normal release path contains warmed end-to-end performance regression gates ### Local Reader benchmark -`tests/reader_perf.rs` compares the public `Read` adapter with the direct writer path on both 84.4 MiB SimpleWiki fixtures. Each row is one release-mode run, and both timings include opening the file. The Reader path necessarily copies into the caller's buffer but transfers decoder-owned chunks into its rendezvous pipe without another copy: +`tests/reader_perf.rs` compares the public `Read` adapter with the direct writer path on 84.4 MiB SimpleWiki inputs. Each row is one release-mode run, and both timings include opening the file. The Reader path necessarily copies into the caller's buffer but transfers decoder-owned chunks into its rendezvous pipe without another copy. The LZ4 test builds its frame before timing: ```bash cargo test --release --test reader_perf reader_writer_comparison -- --ignored --exact --nocapture +cargo test --release --test reader_perf lz4_reader_writer_comparison -- --ignored --exact --nocapture ``` | Format | First byte | Reader | Direct writer | Reader/writer | |---|---:|---:|---:|---:| | bzip2 | 36.627 ms | 205.081 ms | 183.465 ms | 1.118x | | gzip | 11.509 ms | 28.119 ms | 24.515 ms | 1.147x | +| LZ4 | 6.513 ms | 39.213 ms | 39.020 ms | 1.005x | -The bzip2 first-byte measurement includes opening, mapping, and decoding its first block, but not the subsequent whole-input parallel marker scan. The gzip path parses only the current member header before beginning DEFLATE work. Keep this diagnostic single-run. +The bzip2 first-byte measurement includes opening, mapping, and decoding its first block, but not the subsequent whole-input parallel marker scan. Gzip parses only the current member header before beginning DEFLATE work. LZ4 parses the current frame header and a bounded independent-block batch; it never constructs a complete frame layout. Keep this diagnostic single-run. ### Local LZ4 benchmark reproduction and diagnostics diff --git a/README.md b/README.md index 1a9e3fc..c3fbe4b 100644 --- a/README.md +++ b/README.md @@ -159,7 +159,7 @@ It accepts standard independent or linked blocks, stored blocks, all four standa ### Streaming reads -`fbz::Reader` provides a normal `std::io::Read` over bzip2 or gzip files without a preliminary indexing or validation pass: +`fbz::Reader` provides a normal `std::io::Read` over bzip2, gzip, or LZ4 files without a preliminary indexing or validation pass: ```rust use std::io::{BufReader, Read}; @@ -176,7 +176,7 @@ fn main() -> Result<(), Box> { Magic takes priority over the filename extension, with the extension used as a fallback for damaged headers. The decoder runs on an owned worker thread and transfers completed decoder allocations through a zero-capacity pipe; it neither materializes the plaintext nor writes an intermediate file. `DecodeOptions` controls decoder threads and speculative memory. Dropping early disconnects the pipe, cancels outstanding work, and joins the worker. -Checksum errors discovered after output has begun are returned by a later `read()` call. Therefore only successful EOF establishes that the complete stream was valid; dropping early deliberately does not finish validation. Compressed tar inputs yield the decoded tar byte stream rather than extracting it. LZ4 and ZIP are not currently exposed through `Reader`: LZ4's frame-layout pass must first become incremental, while ZIP has no single decoded byte stream. +Checksum errors discovered after output has begun are returned by a later `read()` call. Therefore only successful EOF establishes that the complete stream was valid; dropping early deliberately does not finish validation. Compressed tar inputs yield the decoded tar byte stream rather than extracting it. ZIP is not exposed through `Reader` because an archive has no single decoded byte stream. ## Performance @@ -212,8 +212,8 @@ The same 80.5 MiB prefix in a standard independent-block LZ4 frame (`40.2 MiB` c | CLI | Milliseconds | Peak RSS | fbz/reference | |---|---:|---:|---:| -| fbz, auto (4 workers) | 51.188 | 71.0 MiB | 0.944x | -| Homebrew `lz4` 1.10.0 | 54.228 | 32.0 MiB | — | +| fbz, auto (4 workers) | 55.313 | 58.9 MiB | 0.996x | +| Homebrew `lz4` 1.10.0 | 55.537 | 32.0 MiB | — | The larger fbz RSS includes its memory-mapped 40.2 MiB source plus bounded in-flight decoded blocks; it does not grow with decoded file size. Testing higher worker counts showed no meaningful throughput gain and raised RSS, so automatic LZ4 decoding stops at four workers; `-P` remains an explicit override. @@ -223,7 +223,7 @@ Homebrew `pbzip2` 1.1.13 could not safely decompress the complete 26,668,484,995 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. LZ4 framing and block decoding are likewise implemented in safe Rust. Independent blocks use the same ordered, byte-budgeted scheduler as bzip2; linked blocks retain only the preceding 64 KiB window. LZ4 and DEFLATE share one optimized overlapping back-reference expansion primitive. Header, block, and content XXH32 checksums are validated where present. ZIP reuses the 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` and `twox-hash` are the production checksum helpers; `flate2` and `lz4_flex` are dev-only differential oracles. +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. LZ4 framing and block decoding are likewise implemented in safe Rust. It parses one frame header at a time and schedules independent blocks in bounded batches, so output can begin without a whole-frame layout pass. Independent blocks use the same ordered, byte-budgeted scheduler as bzip2; linked blocks retain only the preceding 64 KiB window. LZ4 and DEFLATE share one optimized overlapping back-reference expansion primitive. Header, block, and content XXH32 checksums are validated where present. ZIP reuses the 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` and `twox-hash` are the production checksum helpers; `flate2` and `lz4_flex` are dev-only differential oracles. Legacy randomized blocks generated by bzip2 releases before 0.9.5 are intentionally unsupported. Normal `BZh1` through `BZh9` streams and concatenated streams are supported. diff --git a/src/lz4.rs b/src/lz4.rs index c0de632..0d6ae18 100644 --- a/src/lz4.rs +++ b/src/lz4.rs @@ -2,7 +2,7 @@ use std::{hash::Hasher, io::Write}; -use rayon::ThreadPoolBuilder; +use rayon::{ThreadPool, ThreadPoolBuilder}; use twox_hash::XxHash32; use crate::history::extend_match; @@ -17,6 +17,7 @@ const UNCOMPRESSED_BIT: u32 = 1 << 31; const WINDOW_SIZE: usize = 64 * 1024; const MIN_PARALLEL_INPUT: usize = 1024 * 1024; const AUTO_THREAD_LIMIT: usize = 4; +const MAX_BATCH_BLOCKS: usize = 64; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum BlockMode { @@ -66,22 +67,24 @@ struct BlockLayout { } #[derive(Clone, Debug)] -struct FrameLayout { +struct FrameHeader { source_start: usize, - source_end: usize, + blocks_start: usize, max_block_size: usize, mode: BlockMode, block_checksums: bool, content_checksum: bool, content_size: Option, - expected_content_checksum: Option, - blocks: Vec, } fn invalid(message: impl Into) -> Error { Error::InvalidLz4(message.into()) } +fn check_output(output: &impl OutputSink) -> Result<()> { + if output.is_cancelled() { Err(Error::Io(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "output reader stopped reading"))) } else { Ok(()) } +} + fn worker_threads(options: DecodeOptions) -> usize { let requested = options.resolved_threads(); if options.threads == 0 { requested.min(AUTO_THREAD_LIMIT) } else { requested } @@ -98,7 +101,7 @@ fn xxhash32(data: &[u8]) -> u32 { hasher.finish() as u32 } -fn parse_frame(data: &[u8], start: usize) -> Result { +fn parse_frame_header(data: &[u8], start: usize) -> Result { let flg = *data.get(start + 4).ok_or_else(|| invalid("truncated frame descriptor"))?; let bd = *data.get(start + 5).ok_or_else(|| invalid("truncated frame descriptor"))?; if flg & 0xc0 != 0x40 { @@ -138,78 +141,40 @@ fn parse_frame(data: &[u8], start: usize) -> Result { return Err(invalid(format!("header checksum mismatch: expected {expected_header_checksum:02x}, decoded {header_checksum:02x}"))); } position += 1; - let mut blocks = Vec::new(); - loop { - let value = read_u32(data, position, "block header")?; - position += 4; - if value == 0 { - break; - } - let stored = value & UNCOMPRESSED_BIT != 0; - let size = (value & !UNCOMPRESSED_BIT) as usize; - if size == 0 || size > max_block_size { - return Err(invalid(format!("block at byte {} has invalid size {size} for {max_block_size}-byte frames", position - 4))); - } - let data_start = position; - let data_end = position.checked_add(size).filter(|&end| end <= data.len()).ok_or_else(|| invalid("block data exceeds the frame"))?; - position = data_end; - let expected_checksum = if block_checksums { - let checksum = read_u32(data, position, "block checksum")?; - position += 4; - Some(checksum) - } else { - None - }; - blocks.push(BlockLayout { data_start, data_end, stored, expected_checksum }); + Ok(FrameHeader { source_start: start, blocks_start: position, max_block_size, mode, block_checksums, content_checksum, content_size }) +} + +fn next_block(data: &[u8], position: &mut usize, frame: &FrameHeader) -> Result> { + let value = read_u32(data, *position, "block header")?; + *position += 4; + if value == 0 { + return Ok(None); + } + let stored = value & UNCOMPRESSED_BIT != 0; + let size = (value & !UNCOMPRESSED_BIT) as usize; + if size == 0 || size > frame.max_block_size { + return Err(invalid(format!("block at byte {} has invalid size {size} for {}-byte frames", *position - 4, frame.max_block_size))); } - let expected_content_checksum = if content_checksum { - let checksum = read_u32(data, position, "content checksum")?; - position += 4; + let data_start = *position; + let data_end = position.checked_add(size).filter(|&end| end <= data.len()).ok_or_else(|| invalid("block data exceeds the frame"))?; + *position = data_end; + let expected_checksum = if frame.block_checksums { + let checksum = read_u32(data, *position, "block checksum")?; + *position += 4; Some(checksum) } else { None }; - Ok(FrameLayout { - source_start: start, - source_end: position, - max_block_size, - mode, - block_checksums, - content_checksum, - content_size, - expected_content_checksum, - blocks, - }) -} - -fn parse(data: &[u8]) -> Result> { - let mut frames = Vec::new(); - let mut position = 0; - while position < data.len() { - let magic = read_u32(data, position, "frame magic")?; - if (SKIPPABLE_MAGIC_START..=SKIPPABLE_MAGIC_END).contains(&magic) { - let length = read_u32(data, position + 4, "skippable-frame length")? as usize; - position = position - .checked_add(8) - .and_then(|value| value.checked_add(length)) - .filter(|&end| end <= data.len()) - .ok_or_else(|| invalid("skippable frame exceeds the input"))?; - continue; - } - if magic == LEGACY_MAGIC { - return Err(invalid("legacy LZ4 frames are not supported")); - } - if magic != FRAME_MAGIC { - return Err(invalid(format!("wrong magic {magic:08x} at byte {position}"))); - } - let frame = parse_frame(data, position)?; - position = frame.source_end; - frames.push(frame); - } - if frames.is_empty() { - return Err(invalid("input contains no LZ4 frames")); - } - Ok(frames) + Ok(Some(BlockLayout { data_start, data_end, stored, expected_checksum })) +} + +fn skip_frame(data: &[u8], position: usize) -> Result { + let length = read_u32(data, position + 4, "skippable-frame length")? as usize; + position + .checked_add(8) + .and_then(|value| value.checked_add(length)) + .filter(|&end| end <= data.len()) + .ok_or_else(|| invalid("skippable frame exceeds the input")) } fn read_length(input: &[u8], position: &mut usize, initial: usize) -> Result { @@ -346,42 +311,100 @@ impl FrameCommitter<'_, S, P> { fn decode_independent( data: &[u8], - frame: &FrameLayout, + frame: &FrameHeader, + position: &mut usize, options: DecodeOptions, + pool: &mut Option, committer: &mut FrameCommitter<'_, S, P>, ) -> Result<()> { let threads = worker_threads(options); - let source_bytes = frame.source_end - frame.source_start; - let parallel_work = frame.blocks.iter().any(|block| !block.stored || block.expected_checksum.is_some()); - if threads == 1 || frame.blocks.len() < 2 || source_bytes < MIN_PARALLEL_INPUT || !parallel_work { - for block in &frame.blocks { - let decoded = decode_layout_block(data, block, &[], frame.max_block_size)?; - committer.commit(block, decoded)?; + let parallel_slots = options.memory_limit / frame.max_block_size; + if threads == 1 || parallel_slots < 2 { + loop { + check_output(committer.output)?; + let Some(block) = next_block(data, position, frame)? else { break }; + let decoded = decode_layout_block(data, &block, &[], frame.max_block_size)?; + committer.commit(&block, decoded)?; } return Ok(()); } - let jobs: Vec<_> = frame - .blocks - .iter() - .cloned() - .enumerate() - .map(|(key, block)| Job { key, reservation: if block.stored { 0 } else { frame.max_block_size }, payload: block }) - .collect(); - let pool = - ThreadPoolBuilder::new().num_threads(threads).thread_name(|index| format!("fbz-lz4-{index}")).build().map_err(|error| invalid(error.to_string()))?; - run_ordered( - &pool, - &jobs, - PipelineLimits { memory: options.memory_limit, active: threads.saturating_add(2) }, - |block| decode_layout_block(data, block, &[], frame.max_block_size), - |result| result.as_ref().map_or(0, DecodedBlock::retained_bytes), - |results| { - for (key, block) in frame.blocks.iter().enumerate() { - committer.commit(block, results.take(key)??)?; + + let mut ended = false; + while !ended { + check_output(committer.output)?; + let mut batch = Vec::new(); + let mut estimated_work = 0_usize; + let mut parse_error = None; + while batch.len() < MAX_BATCH_BLOCKS && (batch.len() < threads || estimated_work < MIN_PARALLEL_INPUT) { + match next_block(data, position, frame) { + Ok(Some(block)) => { + estimated_work = estimated_work.saturating_add(if block.stored { + if block.expected_checksum.is_some() { block.data_end - block.data_start } else { 0 } + } else { + frame.max_block_size + }); + batch.push(block); + } + Ok(None) => { + ended = true; + break; + } + Err(error) => { + parse_error = Some(error); + ended = true; + break; + } } - Ok(()) - }, - ) + } + if batch.is_empty() { + return match parse_error { + Some(error) => Err(error), + None => Ok(()), + }; + } + let parallel_work = + batch.len() >= 2 && estimated_work >= MIN_PARALLEL_INPUT && batch.iter().any(|block| !block.stored || block.expected_checksum.is_some()); + if !parallel_work { + for block in batch { + let decoded = decode_layout_block(data, &block, &[], frame.max_block_size)?; + committer.commit(&block, decoded)?; + } + } else { + if pool.is_none() { + *pool = Some( + ThreadPoolBuilder::new() + .num_threads(threads) + .thread_name(|index| format!("fbz-lz4-{index}")) + .build() + .map_err(|error| invalid(error.to_string()))?, + ); + } + let pool = pool.as_ref().unwrap(); + let jobs: Vec<_> = batch + .iter() + .cloned() + .enumerate() + .map(|(key, block)| Job { key, reservation: if block.stored { 0 } else { frame.max_block_size }, payload: block }) + .collect(); + run_ordered( + pool, + &jobs, + PipelineLimits { memory: options.memory_limit, active: threads.saturating_add(2) }, + |block| decode_layout_block(data, block, &[], frame.max_block_size), + |result| result.as_ref().map_or(0, DecodedBlock::retained_bytes), + |results| { + for (key, block) in batch.iter().enumerate() { + committer.commit(block, results.take(key)??)?; + } + Ok(()) + }, + )?; + } + if let Some(error) = parse_error { + return Err(error); + } + } + Ok(()) } fn update_history(history: &mut Vec, bytes: &[u8]) { @@ -397,12 +420,19 @@ fn update_history(history: &mut Vec, bytes: &[u8]) { history.extend_from_slice(bytes); } -fn decode_linked(data: &[u8], frame: &FrameLayout, committer: &mut FrameCommitter<'_, S, P>) -> Result<()> { +fn decode_linked( + data: &[u8], + frame: &FrameHeader, + position: &mut usize, + committer: &mut FrameCommitter<'_, S, P>, +) -> Result<()> { let mut history = Vec::with_capacity(WINDOW_SIZE); - for block in &frame.blocks { - let decoded = decode_layout_block(data, block, &history, frame.max_block_size)?; + loop { + check_output(committer.output)?; + let Some(block) = next_block(data, position, frame)? else { break }; + let decoded = decode_layout_block(data, &block, &history, frame.max_block_size)?; update_history(&mut history, decoded.bytes(data)); - committer.commit(block, decoded)?; + committer.commit(&block, decoded)?; } Ok(()) } @@ -447,53 +477,76 @@ pub fn decompress_to_sink_with_options_and_progress Result { let options = options.validate()?; - let layouts = parse(data)?; - let mut frames = Vec::with_capacity(layouts.len()); + let mut frames = Vec::new(); let mut blocks = Vec::new(); let mut decoded_total = 0_u64; - for (frame_number, layout) in layouts.iter().enumerate() { - let first_block = blocks.len(); - let mut committer = FrameCommitter { - data, - output, - progress: &mut progress, - hasher: layout.content_checksum.then(|| XxHash32::with_seed(0)), - decoded_base: decoded_total, - decoded: 0, - frame_number: u32::try_from(frame_number).map_err(|_| invalid("too many frames"))?, - blocks: &mut blocks, - }; - match layout.mode { - BlockMode::Independent => decode_independent(data, layout, options, &mut committer)?, - BlockMode::Linked => decode_linked(data, layout, &mut committer)?, + let mut position = 0_usize; + let mut pool = None; + while position < data.len() { + check_output(output)?; + let magic = read_u32(data, position, "frame magic")?; + if (SKIPPABLE_MAGIC_START..=SKIPPABLE_MAGIC_END).contains(&magic) { + position = skip_frame(data, position)?; + continue; } - if let Some(expected) = layout.content_size - && committer.decoded != expected - { - return Err(invalid(format!("content size mismatch: expected {expected}, decoded {}", committer.decoded))); + if magic == LEGACY_MAGIC { + return Err(invalid("legacy LZ4 frames are not supported")); } - if let Some(expected) = layout.expected_content_checksum { - let actual = committer.hasher.take().unwrap().finish() as u32; - if actual != expected { - return Err(invalid(format!("content checksum mismatch: expected {expected:08x}, decoded {actual:08x}"))); - } + if magic != FRAME_MAGIC { + return Err(invalid(format!("wrong magic {magic:08x} at byte {position}"))); } - let decoded_len = committer.decoded; + let frame_number = frames.len(); + let frame = parse_frame_header(data, position)?; + position = frame.blocks_start; + let first_block = blocks.len(); + let decoded_len = { + let mut committer = FrameCommitter { + data, + output, + progress: &mut progress, + hasher: frame.content_checksum.then(|| XxHash32::with_seed(0)), + decoded_base: decoded_total, + decoded: 0, + frame_number: u32::try_from(frame_number).map_err(|_| invalid("too many frames"))?, + blocks: &mut blocks, + }; + match frame.mode { + BlockMode::Independent => decode_independent(data, &frame, &mut position, options, &mut pool, &mut committer)?, + BlockMode::Linked => decode_linked(data, &frame, &mut position, &mut committer)?, + } + if let Some(expected) = frame.content_size + && committer.decoded != expected + { + return Err(invalid(format!("content size mismatch: expected {expected}, decoded {}", committer.decoded))); + } + if frame.content_checksum { + let expected = read_u32(data, position, "content checksum")?; + position += 4; + let actual = committer.hasher.take().unwrap().finish() as u32; + if actual != expected { + return Err(invalid(format!("content checksum mismatch: expected {expected:08x}, decoded {actual:08x}"))); + } + } + committer.decoded + }; decoded_total = decoded_total.checked_add(decoded_len).ok_or_else(|| invalid("decoded length overflows u64"))?; frames.push(Frame { - compressed_start: layout.source_start as u64, - compressed_end: layout.source_end as u64, + compressed_start: frame.source_start as u64, + compressed_end: position as u64, decoded_start: decoded_total - decoded_len, decoded_len, - block_max_size: layout.max_block_size as u32, - block_mode: layout.mode, - block_checksums: layout.block_checksums, - content_checksum: layout.content_checksum, - declared_content_size: layout.content_size, + block_max_size: frame.max_block_size as u32, + block_mode: frame.mode, + block_checksums: frame.block_checksums, + content_checksum: frame.content_checksum, + declared_content_size: frame.content_size, first_block, block_count: blocks.len() - first_block, }); - progress(DecodeProgress { compressed_bytes: layout.source_end as u64, decoded_bytes: decoded_total }); + progress(DecodeProgress { compressed_bytes: position as u64, decoded_bytes: decoded_total }); + } + if frames.is_empty() { + return Err(invalid("input contains no LZ4 frames")); } output.flush()?; progress(DecodeProgress { compressed_bytes: data.len() as u64, decoded_bytes: decoded_total }); diff --git a/src/reader.rs b/src/reader.rs index ea74377..32872ec 100644 --- a/src/reader.rs +++ b/src/reader.rs @@ -29,7 +29,7 @@ enum State { Failed(StoredError), } -/// A streaming, parallel bzip2 or gzip decoder for file-backed streams. +/// A streaming, parallel bzip2, gzip, or LZ4 decoder for file-backed streams. /// /// Successful EOF means the complete stream and its checksums were validated. /// A corrupt trailer can therefore produce decoded bytes before a later call to @@ -43,7 +43,7 @@ pub struct Reader { } impl Reader { - /// Open a bzip2 or gzip file and start its decoder coordinator. + /// Open a bzip2, gzip, or LZ4 file and start its decoder coordinator. /// /// This opens and memory-maps only the compressed source. Format and option /// errors are returned here; compressed-data errors are returned later by @@ -53,8 +53,8 @@ impl Reader { let path = path.as_ref(); let source = Source::open(path)?; let format = Format::detect(path, source.as_slice())?; - if !matches!(format, Format::Bzip2 | Format::Gzip) { - return Err(Error::UnsupportedFormat("fbz::Reader currently supports bzip2 and gzip streams".into())); + if !matches!(format, Format::Bzip2 | Format::Gzip | Format::Lz4) { + return Err(Error::UnsupportedFormat("fbz::Reader supports bzip2, gzip, and LZ4 streams".into())); } let (mut output, pipe) = output_pipe(); diff --git a/tests/lz4_oracle.rs b/tests/lz4_oracle.rs index 729ecbe..133f456 100644 --- a/tests/lz4_oracle.rs +++ b/tests/lz4_oracle.rs @@ -1,6 +1,6 @@ use std::io::{Read, Write}; -use fbz::{DecodeOptions, lz4}; +use fbz::{DecodeOptions, MAX_DECODED_BLOCK, lz4}; use lz4_flex::frame::{BlockMode, BlockSize, FrameDecoder, FrameEncoder, FrameInfo}; fn encode(data: &[u8], block_size: BlockSize, block_mode: BlockMode, block_checksum: bool, content_checksum: bool, content_size: bool) -> Vec { @@ -74,3 +74,10 @@ fn incompressible_multiblock_input_exercises_parallel_scheduler() { assert!(encoded.len() > 1024 * 1024); assert_matches(&data, &encoded, DecodeOptions { threads: 4, ..DecodeOptions::default() }); } + +#[test] +fn a_small_speculative_budget_falls_back_to_incremental_serial_blocks() { + let data = b"large declared blocks do not require speculative memory ".repeat(120_000); + let encoded = encode(&data, BlockSize::Max4MB, BlockMode::Independent, true, true, true); + assert_matches(&data, &encoded, DecodeOptions { threads: 4, memory_limit: MAX_DECODED_BLOCK }); +} diff --git a/tests/reader.rs b/tests/reader.rs index b9fa3d2..b0d13c4 100644 --- a/tests/reader.rs +++ b/tests/reader.rs @@ -7,6 +7,7 @@ use std::{ use crabz2::{Level, compress}; use fbz::{DecodeOptions, Error, Reader}; use flate2::{Compression, write::GzEncoder}; +use lz4_flex::frame::{BlockMode, BlockSize, FrameEncoder, FrameInfo}; fn gzip(data: &[u8]) -> Vec { let mut encoder = GzEncoder::new(Vec::new(), Compression::new(6)); @@ -14,6 +15,14 @@ fn gzip(data: &[u8]) -> Vec { encoder.finish().unwrap() } +fn lz4(data: &[u8], mode: BlockMode) -> Vec { + let info = + FrameInfo::new().block_size(BlockSize::Max64KB).block_mode(mode).block_checksums(true).content_checksum(true).content_size(Some(data.len() as u64)); + let mut encoder = FrameEncoder::with_frame_info(info, Vec::new()); + encoder.write_all(data).unwrap(); + encoder.finish().unwrap() +} + fn read(path: &Path, options: DecodeOptions) -> std::io::Result> { let mut reader = Reader::open(path, options).unwrap(); let mut output = Vec::new(); @@ -25,7 +34,12 @@ fn read(path: &Path, options: DecodeOptions) -> std::io::Result> { fn reader_detects_stream_formats_without_prevalidation() { let directory = tempfile::tempdir().unwrap(); let plain = b"streaming reader format detection ".repeat(20_000); - let cases = [("bzip-magic.data", compress(&plain, Level::FASTEST)), ("gzip-magic.bz2", gzip(&plain))]; + let cases = [ + ("bzip-magic.data", compress(&plain, Level::FASTEST)), + ("gzip-magic.bz2", gzip(&plain)), + ("lz4-magic.gz", lz4(&plain, BlockMode::Independent)), + ("lz4-linked.data", lz4(&plain, BlockMode::Linked)), + ]; for (name, encoded) in cases { let path = directory.path().join(name); fs::write(&path, encoded).unwrap(); @@ -38,6 +52,41 @@ fn reader_detects_stream_formats_without_prevalidation() { assert_eq!(reader.read(&mut [0; 1]).unwrap_err().kind(), std::io::ErrorKind::InvalidData); } +#[test] +fn lz4_reader_emits_valid_blocks_before_a_later_parse_error() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("incremental.lz4"); + let plain = (0_u8..=255).collect::>().repeat(2_000); + let mut encoded = lz4(&plain, BlockMode::Independent); + let first_size = u32::from_le_bytes(encoded[15..19].try_into().unwrap()) as usize & 0x7fff_ffff; + let second_header = 19 + first_size + 4; + encoded[second_header..second_header + 4].copy_from_slice(&0x7fff_ffff_u32.to_le_bytes()); + fs::write(&path, encoded).unwrap(); + + let mut reader = Reader::open(path, DecodeOptions { threads: 4, ..DecodeOptions::default() }).unwrap(); + let mut first = [0]; + assert_eq!(reader.read(&mut first).unwrap(), 1); + assert_eq!(first[0], plain[0]); + assert_eq!(reader.read_to_end(&mut Vec::new()).unwrap_err().kind(), std::io::ErrorKind::InvalidData); +} + +#[test] +fn lz4_late_checksum_failure_is_sticky_after_output() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("late-error.lz4"); + let plain = b"validated only at the LZ4 trailer ".repeat(40_000); + let mut encoded = lz4(&plain, BlockMode::Independent); + *encoded.last_mut().unwrap() ^= 1; + fs::write(&path, encoded).unwrap(); + + let mut reader = Reader::open(path, DecodeOptions { threads: 4, ..DecodeOptions::default() }).unwrap(); + let mut output = Vec::new(); + let error = reader.read_to_end(&mut output).unwrap_err(); + assert_eq!(error.kind(), std::io::ErrorKind::InvalidData); + assert_eq!(output, plain); + assert_eq!(reader.read(&mut [0; 1]).unwrap_err().kind(), std::io::ErrorKind::InvalidData); +} + #[test] fn late_checksum_failure_is_an_error_after_the_decoded_prefix() { let directory = tempfile::tempdir().unwrap(); @@ -76,7 +125,11 @@ fn bzip2_failure_does_not_turn_a_valid_prefix_into_eof() { #[test] fn dropping_under_backpressure_joins_the_decoder() { let directory = tempfile::tempdir().unwrap(); - let inputs = [("early-drop.gz", gzip(&vec![7; 4 * 1024 * 1024]), 1), ("early-drop.bz2", compress(&vec![9; 2 * 1024 * 1024], Level::FASTEST), 4)]; + let inputs = [ + ("early-drop.gz", gzip(&vec![7; 4 * 1024 * 1024]), 1), + ("early-drop.bz2", compress(&vec![9; 2 * 1024 * 1024], Level::FASTEST), 4), + ("early-drop.lz4", lz4(&vec![11; 8 * 1024 * 1024], BlockMode::Independent), 4), + ]; for (name, encoded, threads) in inputs { let path = directory.path().join(name); fs::write(&path, encoded).unwrap(); @@ -96,10 +149,6 @@ fn reader_rejects_non_stream_archives_and_bad_options_at_open() { fs::write(&zip, b"PK\x05\x06\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0").unwrap(); assert!(matches!(Reader::open(&zip, DecodeOptions::default()), Err(Error::UnsupportedFormat(_)))); - let lz4 = directory.path().join("data.lz4"); - fs::write(&lz4, [0x04, 0x22, 0x4d, 0x18]).unwrap(); - assert!(matches!(Reader::open(&lz4, DecodeOptions::default()), Err(Error::UnsupportedFormat(_)))); - let gzip_path = directory.path().join("data.gz"); fs::write(&gzip_path, gzip(b"data")).unwrap(); assert!(matches!(Reader::open(gzip_path, DecodeOptions { memory_limit: 1, ..DecodeOptions::default() }), Err(Error::InvalidConfiguration(_)))); diff --git a/tests/reader_perf.rs b/tests/reader_perf.rs index 4775464..9898b7c 100644 --- a/tests/reader_perf.rs +++ b/tests/reader_perf.rs @@ -1,10 +1,16 @@ +#[allow(dead_code)] +mod support; + use std::{ - io::{self, Read}, + fs, + io::{self, Read, Write}, path::{Path, PathBuf}, time::{Duration, Instant}, }; -use fbz::{DecodeOptions, Reader, Source, decompress_to_writer, gzip}; +use fbz::{DecodeOptions, Reader, Source, decompress_to_writer, gzip, lz4}; +use lz4_flex::frame::{BlockMode, BlockSize, FrameEncoder, FrameInfo}; +use support::simplewiki_prefix; const DECODED_LEN: u64 = 84_423_012; @@ -19,13 +25,23 @@ fn reader_run(path: &Path) -> (Duration, u64) { (started.elapsed(), decoded) } -fn writer_run(path: &Path, gzip_input: bool) -> Duration { +enum Codec { + Bzip2, + Gzip, + Lz4, +} + +fn writer_run(path: &Path, codec: Codec) -> Duration { let started = Instant::now(); let source = Source::open(path).unwrap(); - if gzip_input { - gzip::decompress_to_writer_with_options(source.as_slice(), &mut io::sink(), DecodeOptions::default()).unwrap(); - } else { - decompress_to_writer(source.as_slice(), &mut io::sink(), DecodeOptions::default()).unwrap(); + match codec { + Codec::Bzip2 => decompress_to_writer(source.as_slice(), &mut io::sink(), DecodeOptions::default()).unwrap(), + Codec::Gzip => { + gzip::decompress_to_writer_with_options(source.as_slice(), &mut io::sink(), DecodeOptions::default()).unwrap(); + } + Codec::Lz4 => { + lz4::decompress_to_writer_with_options(source.as_slice(), &mut io::sink(), DecodeOptions::default()).unwrap(); + } } started.elapsed() } @@ -33,7 +49,7 @@ fn writer_run(path: &Path, gzip_input: bool) -> Duration { #[test] #[ignore = "local single-run Reader latency and writer-path comparison"] fn reader_writer_comparison() { - for (extension, gzip_input) in [("bz2", false), ("gz", true)] { + for (extension, codec) in [("bz2", Codec::Bzip2), ("gz", Codec::Gzip)] { let path = fixture(extension); let first_started = Instant::now(); let mut reader = Reader::open(&path, DecodeOptions::default()).unwrap(); @@ -43,7 +59,7 @@ fn reader_writer_comparison() { drop(reader); let (reader_time, decoded) = reader_run(&path); - let writer_time = writer_run(&path, gzip_input); + let writer_time = writer_run(&path, codec); assert_eq!(decoded, DECODED_LEN); println!( "{extension}: first byte {first_byte:?}, Reader {reader_time:?}, writer {writer_time:?}, ratio {:.3}x", @@ -51,3 +67,35 @@ fn reader_writer_comparison() { ); } } + +#[test] +#[ignore = "local single-run incremental LZ4 Reader latency and writer comparison"] +fn lz4_reader_writer_comparison() { + let directory = tempfile::tempdir().unwrap(); + let plain = simplewiki_prefix(); + let info = FrameInfo::new() + .block_size(BlockSize::Max4MB) + .block_mode(BlockMode::Independent) + .block_checksums(false) + .content_checksum(true) + .content_size(Some(plain.len() as u64)); + let mut encoder = FrameEncoder::with_frame_info(info, Vec::new()); + encoder.write_all(&plain).unwrap(); + let path = directory.path().join("simplewiki-first-5pct.xml.lz4"); + fs::write(&path, encoder.finish().unwrap()).unwrap(); + + let first_started = Instant::now(); + let mut reader = Reader::open(&path, DecodeOptions::default()).unwrap(); + let mut first = [0]; + reader.read_exact(&mut first).unwrap(); + let first_byte = first_started.elapsed(); + drop(reader); + + let (reader_time, decoded) = reader_run(&path); + let writer_time = writer_run(&path, Codec::Lz4); + assert_eq!(decoded, plain.len() as u64); + println!( + "lz4: first byte {first_byte:?}, Reader {reader_time:?}, writer {writer_time:?}, ratio {:.3}x", + reader_time.as_secs_f64() / writer_time.as_secs_f64() + ); +}