Add parallel compression and archive creation

Cargo.toml

@@ -4,11 +4,11 @@ version = "0.1.9"
 edition = "2024"
 rust-version = "1.91"
 license = "Apache-2.0"
-description = "Compression-format research workbench with fast bzip2, gzip, LZ4, and ZIP decompression"
+description = "Fast parallel compression and decompression for bzip2, gzip, LZ4, tar, and ZIP"
 repository = "https://github.com/AnswerDotAI/fbz"
 homepage = "https://github.com/AnswerDotAI/fbz"
 documentation = "https://docs.rs/fbz"
-include = ["/src/**", "/Cargo.toml", "/README.md", "/LICENSE"]
+include = ["/src/**", "/Cargo.toml", "/README.md", "/DEV.md", "/LICENSE"]
 
 [lib]
 name = "fbz"

pyproject.toml

@@ -6,7 +6,7 @@ backend-path = ["."]
 [project]
 name = "fbz"
 dynamic = ["version"]
-description = "Compression-format research workbench with fast bzip2, gzip, LZ4, and ZIP decompression"
+description = "Fast parallel compression and decompression for bzip2, gzip, LZ4, tar, and ZIP"
 license = {text = "Apache-2.0"}
 requires-python = ">=3.10"
 readme = "README.md"

python/fbz/__init__.py

@@ -2,7 +2,7 @@
 import io, os
 from pathlib import Path
 
-from ._core import BadBzip2File, _IndexedReader, __version__, _build_index, _decompress, _scan, _test, bz2_crc32
+from ._core import BadBzip2File, _IndexedReader, __version__, _build_index, _compress, _decompress, _scan, _test, bz2_crc32
 
 DEFAULT_MEMORY_LIMIT = 1024 * 1024 * 1024
 DEFAULT_CACHE_LIMIT = 64 * 1024 * 1024
@@ -28,6 +28,10 @@ def decompress(data: bytes, *, threads=0, memory_limit=DEFAULT_MEMORY_LIMIT) ->
     "Decompress and fully CRC-validate one or more concatenated bzip2 streams."
     return _decompress(data, threads, memory_limit)
 
+def compress(data: bytes, format: str, *, threads=0, memory_limit=DEFAULT_MEMORY_LIMIT, level=None) -> bytes:
+    "Compress *data* as bzip2, gzip, or LZ4."
+    return _compress(data, format, threads, memory_limit, level)
+
 class IndexedBzip2File(io.RawIOBase):
     "Seekable binary reader backed by a validated bzip2 block index."
 
@@ -96,5 +100,5 @@ def test(source, *, threads=0, memory_limit=DEFAULT_MEMORY_LIMIT):
 
 __all__ = [
     "__version__", "BadBzip2File", "BlockCandidate", "DEFAULT_CACHE_LIMIT", "DEFAULT_MEMORY_LIMIT", "EndCandidate",
-    "IndexedBzip2File", "ScanResult", "StreamHeaderCandidate", "build_index", "bz2_crc32", "decompress", "open", "scan", "test"
+    "IndexedBzip2File", "ScanResult", "StreamHeaderCandidate", "build_index", "bz2_crc32", "compress", "decompress", "open", "scan", "test"
 ]

src/bin/fbz/archive_create.rs

@@ -0,0 +1,34 @@
+use std::{
+    env,
+    path::{Component, Path, PathBuf},
+};
+
+use fbz::{Error, Result};
+
+fn invalid(message: impl Into<String>) -> Error {
+    Error::InvalidConfiguration(message.into())
+}
+
+pub(super) fn archive_name(path: &Path) -> Result<PathBuf> {
+    let relative = if path.is_absolute() {
+        let current = env::current_dir()?;
+        path.strip_prefix(&current).ok().map(Path::to_path_buf).or_else(|| path.file_name().map(PathBuf::from))
+    } else {
+        Some(path.to_path_buf())
+    }
+    .ok_or_else(|| invalid(format!("cannot derive an archive name for {}", path.display())))?;
+    let mut clean = PathBuf::new();
+    for component in relative.components() {
+        match component {
+            Component::Normal(name) => clean.push(name),
+            Component::CurDir => {}
+            Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
+                return Err(invalid(format!("refusing unsafe archive input name {}", relative.display())));
+            }
+        }
+    }
+    if clean.as_os_str().is_empty() {
+        return Err(invalid(format!("cannot derive an archive name for {}", path.display())));
+    }
+    Ok(clean)
+}

src/bin/fbz/tar_create.rs

@@ -0,0 +1,43 @@
+use std::{collections::HashSet, fs, io::Write, path::Path};
+
+use fbz::{Bzip2EncodeReport, Bzip2Encoder, EncodeOptions, Error, Result, gzip, lz4};
+
+use super::archive_create::archive_name;
+
+fn invalid(message: impl Into<String>) -> Error {
+    Error::InvalidConfiguration(message.into())
+}
+
+fn append_inputs<W: Write>(inputs: &[String], encoder: W) -> Result<W> {
+    let mut archive = tar::Builder::new(encoder);
+    let mut names = HashSet::new();
+    for input in inputs {
+        if input == "-" {
+            return Err(invalid("stdin cannot be used as a tar archive entry"));
+        }
+        let path = Path::new(input);
+        let name = archive_name(path)?;
+        if !names.insert(name.clone()) {
+            return Err(invalid(format!("duplicate archive root {}", name.display())));
+        }
+        let metadata = fs::symlink_metadata(path)?;
+        if metadata.is_dir() {
+            archive.append_dir_all(&name, path)?;
+        } else {
+            archive.append_path_with_name(path, &name)?;
+        }
+    }
+    archive.into_inner().map_err(Error::from)
+}
+
+pub(super) fn pack_gzip<W: Write + ?Sized>(inputs: &[String], output: &mut W, options: EncodeOptions) -> Result<gzip::EncodeReport> {
+    append_inputs(inputs, gzip::Encoder::new(output, options)?)?.finish().map(|(_, report)| report)
+}
+
+pub(super) fn pack_lz4<W: Write + ?Sized>(inputs: &[String], output: &mut W, options: EncodeOptions) -> Result<lz4::EncodeReport> {
+    append_inputs(inputs, lz4::Encoder::new(output, options)?)?.finish().map(|(_, report)| report)
+}
+
+pub(super) fn pack_bzip2<W: Write + ?Sized>(inputs: &[String], output: &mut W, options: EncodeOptions) -> Result<Bzip2EncodeReport> {
+    append_inputs(inputs, Bzip2Encoder::new(output, options)?)?.finish().map(|(_, report)| report)
+}

src/bz2_encode/LICENSE-MIT

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 John Murray
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

src/bz2_encode/bitwriter.rs

@@ -0,0 +1,165 @@
+//! MSB-first bit writer — the mirror image of the decoder's `BitReader`.
+//!
+//! Sans-io by construction: everything lands in a `Vec<u8>` the caller owns.
+
+use std::vec::Vec;
+
+/// Accumulates bits most-significant-first into a byte buffer.
+pub struct BitWriter {
+    out: Vec<u8>,
+    acc: u64,
+    nbits: u32,
+}
+
+impl BitWriter {
+    pub fn new() -> Self {
+        BitWriter { out: Vec::new(), acc: 0, nbits: 0 }
+    }
+
+    pub fn with_capacity(cap: usize) -> Self {
+        BitWriter { out: Vec::with_capacity(cap), acc: 0, nbits: 0 }
+    }
+
+    /// Write the low `n` bits of `val`, most significant first. `n <= 32`.
+    #[inline]
+    pub fn write_bits(&mut self, n: u32, val: u32) {
+        debug_assert!(n <= 32);
+        let val = if n == 32 { val as u64 } else { (val as u64) & ((1u64 << n) - 1) };
+        self.acc = (self.acc << n) | val;
+        self.nbits += n;
+        while self.nbits >= 8 {
+            self.nbits -= 8;
+            self.out.push((self.acc >> self.nbits) as u8);
+        }
+        self.acc &= (1u64 << self.nbits) - 1;
+    }
+
+    #[inline]
+    pub fn write_bit(&mut self, bit: u32) {
+        self.write_bits(1, bit);
+    }
+
+    /// Write a 48-bit block or end-of-stream magic.
+    pub fn write_magic(&mut self, magic: u64) {
+        self.write_bits(24, (magic >> 24) as u32);
+        self.write_bits(24, (magic & 0xff_ffff) as u32);
+    }
+
+    pub fn write_u8(&mut self, byte: u8) {
+        self.write_bits(8, byte as u32);
+    }
+
+    pub fn write_u32(&mut self, val: u32) {
+        self.write_bits(32, val);
+    }
+
+    /// Take the whole bytes emitted so far, leaving any partial byte behind.
+    ///
+    /// This is what keeps a streaming writer's memory bounded: output can be
+    /// handed off as it is produced instead of accumulating until `finish`.
+    pub fn drain(&mut self) -> Vec<u8> {
+        std::mem::take(&mut self.out)
+    }
+
+    /// Append an exact MSB-first bit buffer whose final byte may be padded.
+    pub fn write_buffer(&mut self, bytes: &[u8], bit_len: usize) {
+        debug_assert!(bit_len <= bytes.len() * 8);
+        let full = bit_len / 8;
+        for &byte in &bytes[..full] {
+            self.write_u8(byte);
+        }
+        let trailing = bit_len % 8;
+        if trailing != 0 {
+            self.write_bits(trailing as u32, u32::from(bytes[full] >> (8 - trailing)));
+        }
+    }
+
+    /// Pad a copy of the final byte while retaining the exact meaningful length.
+    pub fn finish_bits(mut self) -> (Vec<u8>, usize) {
+        let bit_len = self.out.len() * 8 + self.nbits as usize;
+        if self.nbits > 0 {
+            self.out.push((self.acc << (8 - self.nbits)) as u8);
+        }
+        (self.out, bit_len)
+    }
+
+    /// Pad the final partial byte with zero bits and take the buffer.
+    pub fn finish(mut self) -> Vec<u8> {
+        if self.nbits > 0 {
+            let pad = 8 - self.nbits;
+            self.write_bits(pad, 0);
+        }
+        debug_assert_eq!(self.nbits, 0);
+        self.out
+    }
+}
+
+impl Default for BitWriter {
+    fn default() -> Self {
+        BitWriter::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::vec;
+
+    use super::*;
+
+    #[test]
+    fn packs_msb_first() {
+        let mut w = BitWriter::new();
+        w.write_bits(4, 0b1010);
+        w.write_bits(4, 0b0011);
+        assert_eq!(w.finish(), vec![0b1010_0011]);
+    }
+
+    #[test]
+    fn pads_trailing_byte_with_zeros() {
+        let mut w = BitWriter::new();
+        w.write_bits(3, 0b101);
+        assert_eq!(w.finish(), vec![0b1010_0000]);
+    }
+
+    #[test]
+    fn writes_wide_fields() {
+        let mut w = BitWriter::new();
+        w.write_u32(0xdead_beef);
+        assert_eq!(w.finish(), vec![0xde, 0xad, 0xbe, 0xef]);
+    }
+
+    #[test]
+    fn writes_stream_header_shape() {
+        // "BZh9" then the 48-bit block magic, exactly as the decoder expects.
+        let mut w = BitWriter::new();
+        w.write_u8(b'B');
+        w.write_u8(b'Z');
+        w.write_u8(b'h');
+        w.write_u8(b'9');
+        w.write_magic(0x3141_5926_5359);
+        assert_eq!(w.finish(), vec![0x42, 0x5a, 0x68, 0x39, 0x31, 0x41, 0x59, 0x26, 0x53, 0x59]);
+    }
+
+    #[test]
+    fn drain_hands_back_whole_bytes_only() {
+        let mut w = BitWriter::new();
+        w.write_bits(12, 0xabc);
+        assert_eq!(w.drain(), vec![0xab]);
+        assert_eq!(w.drain(), Vec::<u8>::new());
+        w.write_bits(4, 0xd);
+        assert_eq!(w.finish(), vec![0xcd]);
+    }
+
+    #[test]
+    fn straddles_accumulator_boundaries() {
+        let mut w = BitWriter::new();
+        for _ in 0..10 {
+            w.write_bits(24, 0xab_cdef);
+        }
+        let out = w.finish();
+        assert_eq!(out.len(), 30);
+        for chunk in out.chunks(3) {
+            assert_eq!(chunk, &[0xab, 0xcd, 0xef]);
+        }
+    }
+}