diff --git a/DEV.md b/DEV.md index 0055358..027f06d 100644 --- a/DEV.md +++ b/DEV.md @@ -2,14 +2,19 @@ `fastbz2` is a mixed Rust/PyO3 project. The Rust crate is the implementation and public Rust API; `python/fastbz2/` is the public Python package over the private `fastbz2._core` extension. -## Initial architecture +## Architecture ```text -src/ portable decoder, scanner, index, scheduler, CLI support +src/bitreader.rs bounded MSB-first in-memory bit reads +src/crc.rs bzip2 block and combined-stream CRC primitives +src/format.rs cheap structural scan for header and marker candidates +src/lib.rs public Rust API and private PyO3 binding python/fastbz2/ thin Python I/O wrapper over fastbz2._core tests/ Python API and integration tests ``` +The current scanner deliberately does not treat 48-bit marker matches or later `BZh` headers as validated structure. Full decoding must establish the exact block chain and validate every block CRC plus the combined stream CRC before marker candidates can become trusted index entries. Python integration tests use standard-library `bz2`/libbz2 as an independent fixture generator. + The 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. Start with safe scalar Rust designed for LLVM auto-vectorisation. Add narrowly scoped unsafe or architecture-specific SIMD only after profiling, with the safe implementation retained as a differential oracle. diff --git a/README.md b/README.md index a0b79c4..3d073e9 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,16 @@ Fast parallel and indexed bzip2 decompression for Rust and Python. The first performance target is end-to-end throughput within 20% of `librapidarchive`'s `indexed_bzip2` on the same host and input. Portable, SIMD-friendly Rust comes first; architecture-specific SIMD is added only when profiles justify it. -The decoder is not implemented yet. +The first implemented layer is a safe, portable MSB-first bit reader, bzip2 CRC primitives, and a structural scanner for stream headers and non-byte-aligned block/end markers. The scanner reports candidates rather than claiming validation: the decoder, exact stream-chain validation, indexing, parallel scheduler, CLI, and seekable Python file API are not implemented yet. + +```python +import bz2 +from fastbz2 import scan + +result = scan(bz2.compress(b"hello")) +result.blocks[0].bit_offset +# 32 +``` ## Inspiration and credit diff --git a/python/fastbz2/__init__.py b/python/fastbz2/__init__.py index 552aa38..b1c5abe 100644 --- a/python/fastbz2/__init__.py +++ b/python/fastbz2/__init__.py @@ -1,3 +1,24 @@ -from ._core import __version__, hello +from collections import namedtuple -__all__ = ["__version__", "hello"] +from ._core import __version__, _scan, bz2_crc32 + +StreamHeaderCandidate = namedtuple("StreamHeaderCandidate", "byte_offset block_size_100k") +BlockCandidate = namedtuple("BlockCandidate", "bit_offset expected_crc randomized orig_ptr") +EndCandidate = namedtuple("EndCandidate", "bit_offset expected_stream_crc") +ScanResult = namedtuple("ScanResult", "streams blocks stream_ends") + +def scan(data: bytes) -> ScanResult: + """Find candidate stream headers and bit-level block markers in *data*. + + This is a structural scan rather than full validation. Marker patterns after + the first header remain candidates until a decoder validates the stream. + """ + streams, blocks, stream_ends = _scan(data) + streams = [StreamHeaderCandidate(*item) for item in streams] + blocks = [BlockCandidate(*item) for item in blocks] + stream_ends = [EndCandidate(*item) for item in stream_ends] + return ScanResult(streams, blocks, stream_ends) + +__all__ = [ + "__version__", "BlockCandidate", "EndCandidate", "ScanResult", "StreamHeaderCandidate", "bz2_crc32", "scan" +] diff --git a/src/bitreader.rs b/src/bitreader.rs new file mode 100644 index 0000000..5945b5d --- /dev/null +++ b/src/bitreader.rs @@ -0,0 +1,117 @@ +use crate::{Error, Result}; + +/// An MSB-first reader over an in-memory bzip2 bitstream. +#[derive(Clone, Debug)] +pub struct BitReader<'a> { + data: &'a [u8], + bit_offset: u64, +} + +impl<'a> BitReader<'a> { + pub fn new(data: &'a [u8]) -> Self { + Self { data, bit_offset: 0 } + } + + pub fn at(data: &'a [u8], bit_offset: u64) -> Result { + let reader = Self { data, bit_offset }; + if bit_offset > reader.len_bits() { + return Err(Error::InvalidBitOffset { bit_offset, len_bits: reader.len_bits() }); + } + Ok(reader) + } + + #[inline] + pub fn position(&self) -> u64 { + self.bit_offset + } + + #[inline] + pub fn len_bits(&self) -> u64 { + u64::try_from(self.data.len()).unwrap_or(u64::MAX / 8).saturating_mul(8) + } + + #[inline] + pub fn remaining(&self) -> u64 { + self.len_bits().saturating_sub(self.bit_offset) + } + + #[inline] + pub fn read_bit(&mut self) -> Result { + if self.bit_offset >= self.len_bits() { + return Err(self.eof(1)); + } + let byte = self.data[(self.bit_offset / 8) as usize]; + let shift = 7 - (self.bit_offset & 7); + self.bit_offset += 1; + Ok((byte >> shift) & 1 != 0) + } + + /// Read up to 64 bits, returning them in the low bits of a `u64`. + #[inline] + pub fn read_bits(&mut self, count: u8) -> Result { + if count > 64 { + return Err(Error::InvalidBitCount(count)); + } + if u64::from(count) > self.remaining() { + return Err(self.eof(u64::from(count))); + } + if count == 0 { + return Ok(0); + } + + let first_byte = (self.bit_offset / 8) as usize; + let skipped = (self.bit_offset & 7) as u32; + let byte_count = (skipped + u32::from(count)).div_ceil(8) as usize; + let mut word = 0_u128; + for &byte in &self.data[first_byte..first_byte + byte_count] { + word = (word << 8) | u128::from(byte); + } + let available = (byte_count * 8) as u32; + let trailing = available - skipped - u32::from(count); + let mask = if count == 64 { u128::from(u64::MAX) } else { (1_u128 << count) - 1 }; + self.bit_offset += u64::from(count); + Ok(((word >> trailing) & mask) as u64) + } + + #[inline] + pub fn skip(&mut self, count: u64) -> Result<()> { + if count > self.remaining() { + return Err(self.eof(count)); + } + self.bit_offset += count; + Ok(()) + } + + fn eof(&self, requested: u64) -> Error { + Error::UnexpectedEof { bit_offset: self.bit_offset, requested, remaining: self.remaining() } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn reference(data: &[u8], offset: usize, count: usize) -> u64 { + (offset..offset + count).fold(0, |result, bit| (result << 1) | u64::from((data[bit / 8] >> (7 - bit % 8)) & 1)) + } + + #[test] + fn reads_every_alignment_and_width() { + let data = [0x12, 0x34, 0x56, 0x78, 0x9a, 0xbc, 0xde, 0xf0, 0x55]; + for offset in 0..8 { + for count in 0..=64.min(data.len() * 8 - offset) { + let mut reader = BitReader::at(&data, offset as u64).unwrap(); + assert_eq!(reader.read_bits(count as u8).unwrap(), reference(&data, offset, count)); + assert_eq!(reader.position(), (offset + count) as u64); + } + } + } + + #[test] + fn reads_single_bits_msb_first() { + let mut reader = BitReader::new(&[0b1010_0001]); + let bits: Vec<_> = (0..8).map(|_| reader.read_bit().unwrap()).collect(); + assert_eq!(bits, [true, false, true, false, false, false, false, true]); + assert!(matches!(reader.read_bit(), Err(Error::UnexpectedEof { .. }))); + } +} diff --git a/src/crc.rs b/src/crc.rs new file mode 100644 index 0000000..67a462f --- /dev/null +++ b/src/crc.rs @@ -0,0 +1,51 @@ +const POLYNOMIAL: u32 = 0x04c1_1db7; + +const fn make_table() -> [u32; 256] { + let mut table = [0; 256]; + let mut byte = 0; + while byte < 256 { + let mut crc = (byte as u32) << 24; + let mut bit = 0; + while bit < 8 { + crc = if crc & 0x8000_0000 != 0 { (crc << 1) ^ POLYNOMIAL } else { crc << 1 }; + bit += 1; + } + table[byte] = crc; + byte += 1; + } + table +} + +const TABLE: [u32; 256] = make_table(); + +/// Compute the CRC used for an uncompressed bzip2 block. +pub fn bz2_crc32(data: &[u8]) -> u32 { + let mut crc = u32::MAX; + for &byte in data { + let index = ((crc >> 24) as u8 ^ byte) as usize; + crc = (crc << 8) ^ TABLE[index]; + } + !crc +} + +/// Add a block CRC to bzip2's combined stream CRC. +#[inline] +pub fn combine_stream_crc(stream_crc: u32, block_crc: u32) -> u32 { + stream_crc.rotate_left(1) ^ block_crc +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn known_values() { + assert_eq!(bz2_crc32(b""), 0); + assert_eq!(bz2_crc32(b"123456789"), 0xfc89_1918); + } + + #[test] + fn combines_by_rotating_then_xoring() { + assert_eq!(combine_stream_crc(0x8000_0001, 0x1234_5678), 0x1234_567b); + } +} diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..35049ec --- /dev/null +++ b/src/error.rs @@ -0,0 +1,28 @@ +use std::{error, fmt}; + +pub type Result = std::result::Result; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Error { + InvalidBitCount(u8), + InvalidBitOffset { bit_offset: u64, len_bits: u64 }, + UnexpectedEof { bit_offset: u64, requested: u64, remaining: u64 }, + InvalidStreamHeader, +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidBitCount(count) => write!(f, "cannot read {count} bits at once; maximum is 64"), + Self::InvalidBitOffset { bit_offset, len_bits } => { + write!(f, "bit offset {bit_offset} is beyond the {len_bits}-bit input") + } + Self::UnexpectedEof { bit_offset, requested, remaining } => { + write!(f, "unexpected end of input at bit {bit_offset}: requested {requested} bits, {remaining} remain") + } + Self::InvalidStreamHeader => write!(f, "input does not start with a bzip2 BZh1-BZh9 header"), + } + } +} + +impl error::Error for Error {} diff --git a/src/format.rs b/src/format.rs new file mode 100644 index 0000000..13498ba --- /dev/null +++ b/src/format.rs @@ -0,0 +1,142 @@ +use crate::{BitReader, Error, Result}; + +pub const BLOCK_MAGIC: u64 = 0x3141_5926_5359; +pub const END_MAGIC: u64 = 0x1772_4538_5090; +const MAGIC_BITS: u8 = 48; +const MAGIC_MASK: u64 = (1_u64 << MAGIC_BITS) - 1; +const WINDOW_MASK: u64 = (1_u64 << 56) - 1; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct StreamHeaderCandidate { + pub byte_offset: u64, + pub block_size_100k: u8, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct BlockCandidate { + pub bit_offset: u64, + pub expected_crc: u32, + pub randomized: bool, + pub orig_ptr: u32, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct EndCandidate { + pub bit_offset: u64, + pub expected_stream_crc: u32, +} + +/// Candidate stream headers and bit-level block markers found in a bzip2 input. +/// +/// This is an intentionally cheap structural scan, not full validation: marker +/// bit patterns can occur in compressed payloads, so entries after the first +/// stream header are candidates until a decoder validates the chain. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ScanResult { + pub streams: Vec, + pub blocks: Vec, + pub stream_ends: Vec, +} + +pub fn scan(data: &[u8]) -> Result { + if !is_stream_header(data, 0) { + return Err(Error::InvalidStreamHeader); + } + + let streams = (0..=data.len().saturating_sub(4)) + .filter(|&offset| is_stream_header(data, offset)) + .map(|offset| StreamHeaderCandidate { byte_offset: offset as u64, block_size_100k: data[offset + 3] - b'0' }) + .collect(); + let mut blocks = Vec::new(); + let mut stream_ends = Vec::new(); + + if data.len() < 7 { + return Ok(ScanResult { streams, blocks, stream_ends }); + } + + // A 56-bit rolling window contains all eight 48-bit candidates beginning + // in one byte. The simple fixed-width inner loop is intentionally friendly + // to unrolling and auto-vectorisation on both x86-64 and ARM64. + let mut window = data[..7].iter().fold(0_u64, |word, &byte| (word << 8) | u64::from(byte)); + for byte_offset in 0..=data.len() - 7 { + for shift in 0..8_u32 { + let marker = (window >> (8 - shift)) & MAGIC_MASK; + let bit_offset = byte_offset as u64 * 8 + u64::from(shift); + if marker == BLOCK_MAGIC { + if let Some(block) = parse_block(data, bit_offset) { + blocks.push(block); + } + } else if marker == END_MAGIC + && let Some(end) = parse_end(data, bit_offset) + { + stream_ends.push(end); + } + } + if let Some(&next) = data.get(byte_offset + 7) { + window = ((window << 8) & WINDOW_MASK) | u64::from(next); + } + } + + Ok(ScanResult { streams, blocks, stream_ends }) +} + +fn is_stream_header(data: &[u8], offset: usize) -> bool { + data.get(offset..offset + 3) == Some(b"BZh") && matches!(data.get(offset + 3), Some(b'1'..=b'9')) +} + +fn parse_block(data: &[u8], bit_offset: u64) -> Option { + let mut reader = BitReader::at(data, bit_offset + u64::from(MAGIC_BITS)).ok()?; + Some(BlockCandidate { + bit_offset, + expected_crc: reader.read_bits(32).ok()? as u32, + randomized: reader.read_bit().ok()?, + orig_ptr: reader.read_bits(24).ok()? as u32, + }) +} + +fn parse_end(data: &[u8], bit_offset: u64) -> Option { + let mut reader = BitReader::at(data, bit_offset + u64::from(MAGIC_BITS)).ok()?; + Some(EndCandidate { bit_offset, expected_stream_crc: reader.read_bits(32).ok()? as u32 }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn append_bits(target: &mut Vec, value: u64, count: usize) { + target.extend((0..count).rev().map(|shift| value >> shift & 1 != 0)); + } + + fn pack(bits: &[bool]) -> Vec { + bits.chunks(8).map(|chunk| chunk.iter().fold(0, |byte, &bit| (byte << 1) | u8::from(bit)) << (8 - chunk.len())).collect() + } + + #[test] + fn finds_markers_at_every_bit_alignment() { + for prefix in 0..8 { + let mut bits = Vec::new(); + for byte in b"BZh9" { + append_bits(&mut bits, u64::from(*byte), 8); + } + bits.extend(std::iter::repeat_n(false, prefix)); + append_bits(&mut bits, BLOCK_MAGIC, 48); + append_bits(&mut bits, 0x1234_5678, 32); + append_bits(&mut bits, 1, 1); + append_bits(&mut bits, 0x00ab_cdef, 24); + let data = pack(&bits); + let result = scan(&data).unwrap(); + let [block] = result.blocks.as_slice() else { + panic!("expected one block, found {:?}", result.blocks); + }; + assert_eq!(block.bit_offset, 32 + prefix as u64); + assert_eq!(block.expected_crc, 0x1234_5678); + assert!(block.randomized); + assert_eq!(block.orig_ptr, 0x00ab_cdef); + } + } + + #[test] + fn rejects_non_bzip_input() { + assert_eq!(scan(b"not bzip2"), Err(Error::InvalidStreamHeader)); + } +} diff --git a/src/lib.rs b/src/lib.rs index 724fc3d..4f25c79 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,20 +1,38 @@ -pub fn hello(name: &str) -> String { - format!("Hello, {name}!") -} +//! Portable bzip2 primitives and format inspection. + +mod bitreader; +mod crc; +mod error; +mod format; + +pub use bitreader::BitReader; +pub use crc::{bz2_crc32, combine_stream_crc}; +pub use error::{Error, Result}; +pub use format::{BLOCK_MAGIC, BlockCandidate, END_MAGIC, EndCandidate, ScanResult, StreamHeaderCandidate, scan}; #[cfg(feature = "python")] mod python { - use super::hello; - use pyo3::prelude::*; + use pyo3::{exceptions::PyValueError, prelude::*}; + + #[pyfunction(name = "_scan")] + #[allow(clippy::type_complexity)] + fn py_scan(data: &[u8]) -> PyResult<(Vec<(u64, u8)>, Vec<(u64, u32, bool, u32)>, Vec<(u64, u32)>)> { + let result = crate::scan(data).map_err(|err| PyValueError::new_err(err.to_string()))?; + let streams = result.streams.into_iter().map(|stream| (stream.byte_offset, stream.block_size_100k)).collect(); + let blocks = result.blocks.into_iter().map(|block| (block.bit_offset, block.expected_crc, block.randomized, block.orig_ptr)).collect(); + let stream_ends = result.stream_ends.into_iter().map(|end| (end.bit_offset, end.expected_stream_crc)).collect(); + Ok((streams, blocks, stream_ends)) + } - #[pyfunction(name = "hello")] - fn py_hello(name: &str) -> String { - hello(name) + #[pyfunction(name = "bz2_crc32")] + fn py_bz2_crc32(data: &[u8]) -> u32 { + crate::bz2_crc32(data) } #[pymodule] fn _core(m: &Bound<'_, PyModule>) -> PyResult<()> { - m.add_function(wrap_pyfunction!(py_hello, m)?)?; + m.add_function(wrap_pyfunction!(py_scan, m)?)?; + m.add_function(wrap_pyfunction!(py_bz2_crc32, m)?)?; m.add("__version__", env!("CARGO_PKG_VERSION"))?; Ok(()) } diff --git a/tests/test_basic.py b/tests/test_basic.py deleted file mode 100644 index ec7419a..0000000 --- a/tests/test_basic.py +++ /dev/null @@ -1,4 +0,0 @@ -from fastbz2 import hello - -def test_hello(): - assert hello("fastship") == "Hello, fastship!" diff --git a/tests/test_scan.py b/tests/test_scan.py new file mode 100644 index 0000000..9631c58 --- /dev/null +++ b/tests/test_scan.py @@ -0,0 +1,40 @@ +import bz2 + +import pytest + +from fastbz2 import bz2_crc32, scan + +def combined_crc(blocks): + crc = 0 + for block in blocks: crc = ((crc << 1) | (crc >> 31)) & 0xffffffff ^ block.expected_crc + return crc + +def test_single_block_matches_libbz2(): + raw = b"The quick brown fox jumps over the lazy dog\n" * 100 + result = scan(bz2.compress(raw)) + + assert result.streams == [(0, 9)] + assert len(result.blocks) == len(result.stream_ends) == 1 + assert result.blocks[0].bit_offset == 32 + assert result.blocks[0].expected_crc == bz2_crc32(raw) + assert result.blocks[0].randomized is False + assert result.stream_ends[0].expected_stream_crc == result.blocks[0].expected_crc + +def test_multiblock_combined_crc_matches_libbz2(): + raw = bytes(range(256)) * 1000 + result = scan(bz2.compress(raw, compresslevel=1)) + + assert len(result.blocks) >= 2 + assert len(result.stream_ends) == 1 + assert combined_crc(result.blocks) == result.stream_ends[0].expected_stream_crc + +def test_concatenated_stream_headers(): + first = bz2.compress(b"first stream") + second = bz2.compress(b"second stream", compresslevel=3) + result = scan(first + second) + + assert result.streams == [(0, 9), (len(first), 3)] + assert len(result.blocks) == len(result.stream_ends) == 2 + +def test_rejects_non_bzip_input(): + with pytest.raises(ValueError, match="does not start with a bzip2"): scan(b"not bzip2")