Fix parallel scheduling across concatenated bzip2 streams

Cargo.toml

@@ -31,7 +31,7 @@ rayon = "1.12.0"
 tempfile = "3.27.0"
 
 [dev-dependencies]
-crabz2 = "0.4.0"
+crabz2 = { version = "0.4.0", features = ["parallel"] }
 libbz2-rs-sys = { version = "0.2.5", default-features = false, features = ["std"] }
 
 [features]

DEV.md

@@ -26,6 +26,8 @@ The current scanner deliberately does not treat 48-bit marker matches or later `
 
 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.
 
+Parallel decoding uses a rolling candidate queue rather than stopping at stream boundaries or waiting for fixed batches. Workers reserve the maximum possible decoded block size before starting; once a block finishes, that conservative reservation shrinks to its actual output size and is released when ordered validation consumes or rejects it. Thus the `memory_limit` bounds speculative decoded output while short multistream inputs can keep the worker pool busy. The 1 GiB default admits one worst-case block per worker on the primary 18-core machine.
+
 The production decoder is safe scalar Rust designed for LLVM auto-vectorisation. Huffman decoding uses a 4096-entry direct table for codes up to 12 bits and canonical fallback for longer codes. Add narrowly scoped unsafe or architecture-specific SIMD only after profiling; `libbz2-rs-sys` remains the dev-only differential oracle.
 
 ## Commands
@@ -51,7 +53,7 @@ The same test binary contains a warmed end-to-end performance regression gate ca
 
 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 SimpleWiki benchmark
+### Local Wikipedia benchmarks
 
 `tests/wiki_perf.rs` contains a release-only local benchmark that is skipped by default. Its git-ignored fixture is a single bzip2 stream containing the first 84,423,012 decoded bytes (about 5%) of SimpleWiki, stored at `meta/simplewiki-first-5pct.xml.bz2`. Run it with:
 
@@ -69,6 +71,57 @@ cargo test --release --test wiki_perf simplewiki_full -- --ignored --nocapture
 
 The full test streams decoded bytes into a counting sink and validates every block and stream CRC. Stream headers found during the structural scan remain speculative: only the exact byte-aligned header following a validated end-of-stream marker starts another stream.
 
+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
+cargo test --release --test wiki_perf enwiki_first_1000_fastbz2_parallel -- --ignored --exact --nocapture
+cargo test --release --test wiki_perf enwiki_first_1000_crabz2_parallel -- --ignored --exact --nocapture
+cargo test --release --test wiki_perf enwiki_first_1000_fastbz2_serial -- --ignored --exact --nocapture
+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.
+
+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.
+
+#### Regenerating the fixtures
+
+Set `wiki` to the checkout containing the Wikimedia dumps:
+
+```bash
+wiki=/path/to/wiki2md
+```
+
+Recreate the SimpleWiki fixtures from its validated compressed and decoded files:
+
+```bash
+head -c 84423012 "$wiki/data/simplewiki-latest-pages-articles.xml" | bzip2 -9c > meta/simplewiki-first-5pct.xml.bz2
+ln -s "$wiki/data/simplewiki-latest-pages-articles.xml.bz2" meta/simplewiki-full.xml.bz2
+```
+
+Run the 5% test to obtain and verify its decoded length and BLAKE3. Obtain the full decoded length with `stat`; update `FIVE_PERCENT_LEN`, `FIVE_PERCENT_BLAKE3`, or `FULL_LEN` only when intentionally changing a fixture.
+
+The enwiki dump is multistream. Extract its unique compressed stream offsets from the official index, whose first page-bearing stream begins after an unindexed initial stream at byte zero:
+
+```bash
+bzcat "$wiki/data/enwiki-latest-pages-articles-multistream-index.txt.bz2" \
+  | awk -F: '!seen[$1]++ {print $1}' > "$wiki/meta/enwiki-multistream-offsets.txt"
+boundary=$(sed -n '1000p' "$wiki/meta/enwiki-multistream-offsets.txt")
+head -c "$boundary" "$wiki/data/enwiki-latest-pages-articles-multistream.xml.bz2" \
+  > "$wiki/data/enwiki-first-1000-streams.xml.bz2"
+ln -s "$wiki/data/enwiki-first-1000-streams.xml.bz2" meta/enwiki-first-1000-streams.xml.bz2
+```
+
+Line 1000 is the start of stream 1001 because byte zero is stream 1 and is absent from the page index. Thus `[0, boundary)` contains exactly 1,000 complete bzip2 streams. For the 2026-08-01 dump, `boundary` is `654362682` and the decoded bzip2 payload length is `2715335085`.
+
+To create the separately useful well-formed parser fixture, append only the XML root close after decoding; those 13 bytes are deliberately excluded from `ENWIKI_1000_LEN`:
+
+```bash
+fastbz2 decode "$wiki/data/enwiki-first-1000-streams.xml.bz2" -o "$wiki/data/enwiki-first-1000-streams.xml"
+printf '</mediawiki>\n' >> "$wiki/data/enwiki-first-1000-streams.xml"
+xmllint --stream --noout "$wiki/data/enwiki-first-1000-streams.xml"
+```
+
 ## Platforms
 
 CI tests and builds Linux on x86-64 and ARM64, and macOS on ARM64. macOS Intel remains best-effort and should not add implementation complexity. Keep the core portable: no required mmap, custom allocator, `io_uring`, assembly, or native-endian parsing. Platform-specific positional I/O belongs behind a small source abstraction.

README.md

@@ -6,7 +6,37 @@ Fast parallel and indexed bzip2 decompression for Rust and Python.
 
 The performance floor is end-to-end decompression within 20% of the maintained pure-Rust `libbz2-rs-sys` decoder on a representative corpus. Portable, SIMD-friendly Rust comes first; architecture-specific SIMD is added only when profiles justify it. The much larger Simple English Wikipedia dump is used for local throughput measurements.
 
-The implementation includes a safe structural scanner, an in-repo decoder with a tuned 12-bit Huffman lookup table, CRC-validated block decoding, bounded parallel scheduling, persistent indexes, a native CLI, and a seekable Python file API. Marker scans remain speculative until decoding establishes an exact stream chain and validates block and combined-stream CRCs.
+The implementation includes a safe structural scanner, an in-repo decoder with a tuned 12-bit Huffman lookup table, CRC-validated block decoding, memory-bounded rolling parallel scheduling, persistent indexes, a native CLI, and a seekable Python file API. Marker scans remain speculative until decoding establishes an exact stream chain and validates block and combined-stream CRCs.
+
+## Performance
+
+These are single local release-mode runs on the primary Apple Silicon development machine, using 18 workers for parallel rows. They are observations rather than statistically aggregated benchmarks. Modes are stated per row because a streaming sink, an in-memory `Vec<u8>`, and a CLI pipeline have different allocation and I/O costs; compare rows using the same method most directly.
+
+Full Simple English Wikipedia (`338 MB` compressed, `1,688,460,257` bytes decoded):
+
+| Decoder | Mode | Seconds |
+|---|---:|---:|
+| fastbz2 | parallel, 18 threads, streaming sink | 2.244 |
+| 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 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 “Local Wikipedia benchmarks” documents exact fixture generation and commands.
+
+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.
 
 ```python
 import bz2

python/fastbz2/__init__.py

@@ -1,11 +1,10 @@
 from collections import namedtuple
-import io
-import os
+import io, os
 from pathlib import Path
 
 from ._core import BadBzip2File, _IndexedReader, __version__, _build_index, _decompress, _scan, _test, bz2_crc32
 
-DEFAULT_MEMORY_LIMIT = 512 * 1024 * 1024
+DEFAULT_MEMORY_LIMIT = 1024 * 1024 * 1024
 DEFAULT_CACHE_LIMIT = 64 * 1024 * 1024
 
 StreamHeaderCandidate = namedtuple("StreamHeaderCandidate", "byte_offset block_size_100k")
@@ -26,17 +25,16 @@ def scan(data: bytes) -> ScanResult:
     return ScanResult(streams, blocks, stream_ends)
 
 def decompress(data: bytes, *, threads=0, memory_limit=DEFAULT_MEMORY_LIMIT) -> bytes:
-    """Decompress and fully CRC-validate one or more concatenated bzip2 streams."""
+    "Decompress and fully CRC-validate one or more concatenated bzip2 streams."
     return _decompress(data, threads, memory_limit)
 
 class IndexedBzip2File(io.RawIOBase):
-    """Seekable binary reader backed by a validated bzip2 block index."""
+    "Seekable binary reader backed by a validated bzip2 block index."
 
     def __init__(self, source, *, threads=0, index=None, memory_limit=DEFAULT_MEMORY_LIMIT, cache_limit=DEFAULT_CACHE_LIMIT):
         super().__init__()
         if isinstance(source, (bytes, bytearray, memoryview)):
-            if index is not None and not isinstance(index, (bytes, bytearray, memoryview)):
-                index = Path(index).read_bytes()
+            if index is not None and not isinstance(index, (bytes, bytearray, memoryview)): index = Path(index).read_bytes()
             self._reader = _IndexedReader.from_bytes(bytes(source), threads, memory_limit, index, cache_limit)
         else:
             if index is not None and isinstance(index, (bytes, bytearray, memoryview)):
@@ -78,11 +76,11 @@ def close(self):
         super().close()
 
 def open(source, *, threads=0, index=None, memory_limit=DEFAULT_MEMORY_LIMIT, cache_limit=DEFAULT_CACHE_LIMIT):
-    """Open a path or bytes object as a seekable bzip2 binary file."""
+    "Open a path or bytes object as a seekable bzip2 binary file."
     return IndexedBzip2File(source, threads=threads, index=index, memory_limit=memory_limit, cache_limit=cache_limit)
 
 def build_index(source, path=None, *, threads=0, memory_limit=DEFAULT_MEMORY_LIMIT) -> bytes:
-    """Fully validate *source* and return its source-bound binary block index."""
+    "Fully validate *source* and return its source-bound binary block index."
     if isinstance(source, (bytes, bytearray, memoryview)):
         reader = _IndexedReader.from_bytes(bytes(source), threads, memory_limit, None, DEFAULT_CACHE_LIMIT)
         encoded = reader.index_bytes()
@@ -91,7 +89,7 @@ def build_index(source, path=None, *, threads=0, memory_limit=DEFAULT_MEMORY_LIM
     return encoded
 
 def test(source, *, threads=0, memory_limit=DEFAULT_MEMORY_LIMIT):
-    """Fully decode and CRC-validate *source*, returning ``None`` on success."""
+    "Fully decode and CRC-validate *source*, returning ``None`` on success."
     if isinstance(source, (bytes, bytearray, memoryview)):
         _IndexedReader.from_bytes(bytes(source), threads, memory_limit, None, DEFAULT_CACHE_LIMIT)
     else: _test(os.fspath(source), threads, memory_limit)

src/bin/fastbz2.rs

@@ -27,7 +27,7 @@ enum Command {
         stdout: bool,
         #[arg(short = 'P', long, default_value_t = 0)]
         threads: usize,
-        #[arg(long, default_value = "512M", value_parser = parse_size)]
+        #[arg(long, default_value = "1G", value_parser = parse_size)]
         memory_limit: usize,
         #[arg(short, long)]
         force: bool,
@@ -37,7 +37,7 @@ enum Command {
         input: String,
         #[arg(short = 'P', long, default_value_t = 0)]
         threads: usize,
-        #[arg(long, default_value = "512M", value_parser = parse_size)]
+        #[arg(long, default_value = "1G", value_parser = parse_size)]
         memory_limit: usize,
     },
     /// Build a validated, source-bound block index.
@@ -47,7 +47,7 @@ enum Command {
         output: Option<PathBuf>,
         #[arg(short = 'P', long, default_value_t = 0)]
         threads: usize,
-        #[arg(long, default_value = "512M", value_parser = parse_size)]
+        #[arg(long, default_value = "1G", value_parser = parse_size)]
         memory_limit: usize,
         #[arg(short, long)]
         force: bool,
@@ -57,7 +57,7 @@ enum Command {
         input: PathBuf,
         #[arg(short = 'P', long, default_value_t = 0)]
         threads: usize,
-        #[arg(long, default_value = "512M", value_parser = parse_size)]
+        #[arg(long, default_value = "1G", value_parser = parse_size)]
         memory_limit: usize,
     },
 }
@@ -80,7 +80,7 @@ fn run(cli: Cli) -> fastbz2::Result<()> {
                 if stdout || output.is_none() {
                     return decode_stdin(&mut io::stdout().lock());
                 }
-                return atomic_write(output.as_ref().unwrap(), force, |writer| decode_stdin(writer));
+                return atomic_write(output.as_ref().unwrap(), force, decode_stdin);
             }
             let input_path = Path::new(&input);
             let source = Source::open(input_path)?;