Parallel compression

mbrotli::compressor::parallel divides a known input into fixed independent segments and assembles them into one Brotli stream. It supports qualities 0–11 with standard windows. Dictionaries, Large Window, and framing are unsupported by this API.

The caller runs the tasks using scoped threads, Rayon, or another executor. The library does not create a thread pool.

Compressing a slice

use mbrotli::compressor::parallel::{
    BatchConfig, ParallelCompressor, ParallelConfig, TaskCount,
};
use mbrotli::{Decompressor, EncoderConfig, Quality};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let input = vec![b'a'; 8 << 20];
    let mut encoder = ParallelCompressor::new(
        EncoderConfig::default().with_quality(Quality::Q5),
        ParallelConfig::default(),
    )?;
    let mut batch = encoder.prepare_slice(
        &input,
        BatchConfig::auto(TaskCount::try_from(4)?),
    )?;
    let tasks = batch.take_tasks()?;
    std::thread::scope(|scope| {
        for task in tasks {
            scope.spawn(move || task.run());
        }
    });
    let mut output = Vec::new();
    batch.finish_into(&mut output)?;
    let mut decoder = Decompressor::new(Default::default())?;
    assert_eq!(decoder.decompress(&output)?, input);
    Ok(())
}

take_tasks transfers each task once. Every taken task must run or be dropped so its completion can be reported. Finalization collects task results, validates artifacts, and assembles them in input order. A blocking wait requires the caller’s executor to remain runnable.

Running tasks with Rayon

Add rayon = "1.12" alongside mbrotli in your dependencies. Create a pool once and reuse it across requests. Task completion order does not affect assembly order or the resulting bytes for fixed segment settings.

use mbrotli::compressor::parallel::{
    BatchConfig, ParallelCompressor, ParallelConfig, TaskCount,
};
use mbrotli::{Decompressor, EncoderConfig, Quality};
use rayon::prelude::*;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = rayon::ThreadPoolBuilder::new().num_threads(4).build()?;
    let mut encoder = ParallelCompressor::new(
        EncoderConfig::default().with_quality(Quality::Q5),
        ParallelConfig::default(),
    )?;
    let input = b"parallel compression with a reusable pool\n".repeat(220_000);
    let mut batch = encoder.prepare_slice(
        &input,
        BatchConfig::auto(TaskCount::try_from(4)?),
    )?;
    let tasks = batch.take_tasks()?;
    pool.install(|| tasks.into_par_iter().for_each(|task| task.run()));

    let mut compressed = Vec::new();
    batch.finish_into(&mut compressed)?;
    let mut decoder = Decompressor::new(Default::default())?;
    assert_eq!(decoder.decompress(&compressed)?, input);
    Ok(())
}

For many unrelated small inputs, another option is one ordinary Compressor per worker. That preserves serial compression decisions for each input. The batch API above splits one input into a single assembled stream.

Streaming input

When input arrives in chunks, start(window) returns a session and a task queue. Each full segment the session accepts becomes a task on that queue; clone the queue for each executor and run for task in queue { task.run() }. The session delivers finished parts in input order while later segments are still encoding, and closes the queue when the stream finishes, fails, or is dropped.

use mbrotli::compressor::parallel::{
    ParallelCompressor, ParallelConfig, ParallelEncoderStatus, TaskCount,
};
use mbrotli::{EncoderConfig, Operation, Quality};
use std::io::{Read, Write};

fn compress(
    mut reader: impl Read,
    mut writer: impl Write,
) -> Result<(), Box<dyn std::error::Error>> {
    let mut encoder = ParallelCompressor::new(
        EncoderConfig::default().with_quality(Quality::Q5),
        ParallelConfig::default(),
    )?;
    std::thread::scope(|scope| {
        let (mut session, queue) = encoder.start(TaskCount::try_from(4)?)?;
        for _ in 0..4 {
            let queue = queue.clone();
            scope.spawn(move || queue.into_iter().for_each(|task| task.run()));
        }
        drop(queue);
        let mut chunk = vec![0; 1 << 16];
        let mut output = vec![0; 1 << 16];
        loop {
            let read = reader.read(&mut chunk)?;
            let operation = if read == 0 { Operation::Finish } else { Operation::Process };
            let mut input = &chunk[..read];
            loop {
                let progress = session.process(input, &mut output, operation)?;
                input = &input[progress.consumed..];
                writer.write_all(&output[..progress.produced])?;
                match progress.status {
                    ParallelEncoderStatus::Finished => return Ok(()),
                    ParallelEncoderStatus::Pending => session.wait()?,
                    ParallelEncoderStatus::NeedsInput if input.is_empty() => break,
                    _ => {}
                }
            }
        }
    })
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let input = b"incremental parallel input\n".repeat(350_000);
    let mut compressed = Vec::new();
    compress(input.as_slice(), &mut compressed)?;
    let mut decoder = mbrotli::Decompressor::new(Default::default())?;
    assert_eq!(decoder.decompress(&compressed)?, input);
    Ok(())
}

• process never blocks. Pending means the window is full or a flush or finish waits for queued tasks; call wait, which blocks until one completes.
• window bounds the segments queued, running or awaiting delivery, and so the session’s memory. A full window stops input from being consumed.
• Without a flush, the output is byte-identical to prepare_slice over the same bytes, however the input was chunked. Operation::Flush ends the current segment early, which moves later segment boundaries.
• On a single thread, run queue.try_recv() tasks before wait; nothing else will. Dropping every queue handle fails the stream instead of hanging.
• into_session(window) returns an owned session without a lifetime; into_compressor returns the compressor.

Reader and writer adapters

writer(sink, window) and reader(source, window) wrap the same session in std::io::Write and std::io::Read and return the task queue beside the adapter, so the loop above becomes std::io::copy:

use mbrotli::compressor::parallel::{ParallelCompressor, ParallelConfig, TaskCount};
use mbrotli::io::FinishError;
use mbrotli::{EncoderConfig, Quality};
use std::io::{Read, Write};

fn compress<W: Write>(
    mut reader: impl Read,
    writer: W,
) -> Result<W, Box<dyn std::error::Error>> {
    let mut encoder = ParallelCompressor::new(
        EncoderConfig::default().with_quality(Quality::Q5),
        ParallelConfig::default(),
    )?;
    std::thread::scope(|scope| {
        let (mut sink, queue) = encoder.writer(writer, TaskCount::try_from(4)?)?;
        for _ in 0..4 {
            let queue = queue.clone();
            scope.spawn(move || queue.into_iter().for_each(|task| task.run()));
        }
        drop(queue);
        std::io::copy(&mut reader, &mut sink)?;
        Ok(sink.finish().map_err(FinishError::into_error)?)
    })
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let input = b"stream through caller-owned workers\n".repeat(250_000);
    let compressed = compress(input.as_slice(), Vec::new())?;
    let mut decoder = mbrotli::Decompressor::new(Default::default())?;
    assert_eq!(decoder.decompress(&compressed)?, input);
    Ok(())
}

• Unlike process, the adapters block: a write waits while the window is full, a read waits for the next part, and finish waits for every part. Run the queue on threads other than the one using the adapter.
• Output is byte-identical to prepare_slice over the same bytes. The writer’s flush is Operation::Flush: it makes everything written decodable and moves later segment boundaries.
• The writer keeps every compressed byte until the sink takes it, so short writes, WouldBlock and sink errors lose nothing, and finish hands the adapter back inside FinishError for a retry, as EncoderWriter does. reader.into_parts() returns source bytes read ahead of the session.
• Session failures arrive as std::io::Error with the ParallelEncodeError as their source. Dropping an adapter abandons the stream and closes the queue.

Segments and compressed size

Segment size defaults to 4 MiB and accepts 64 KiB–16 MiB. Effective task count is at most the number of segments. For fixed input and segment settings, task grouping and completion order do not change output bytes.

Segments are encoded independently. At qualities 2-9 a segment may copy from the input before it, up to a quarter of a segment within the window, and codes distances as a serial stream does; at qualities 0-1 and 10-11 it starts from nothing and writes every distance explicitly, which costs most on input that repeats one distance, such as tables and logs. The compression ratio can be worse than serial compression, especially for repetition further back than that history. Compare both output size and elapsed time.

Empty input uses serial encoding without tasks. A nonempty input below the default 8 MiB parallel threshold uses a single serial task only when it also fits in one segment.

Sources and staging

FileSource uses positional reads on Unix and Windows. SeekSource accepts Read + Seek + Send, serializes seek/read operations, and lets compression run concurrently. Sources must remain immutable for the batch lifetime; metadata checks cannot detect every in-place mutation.

Owned sources and shared handles use the same generic entry point. For an existing Arc<FileSource>, type inference can require prepare_source::<FileSource, _>(shared_file, config).

BatchConfig::auto(tasks) stages in memory using bounds derived from source length and configured segment size. It caps effective tasks at the segment count and still enforces any configured aggregate memory limit. It does not query available RAM or switch to files.

BatchConfig::memory(tasks, max_bytes) sets an explicit payload-and-descriptor budget. To obtain its exact required preflight allowance, call estimate_source with BatchConfig::auto(tasks), then convert estimate.maximum_staging_memory_bytes()? using usize::try_from. maximum_staged_bytes remains the payload-only bound. Planning also accounts for worker and assembly storage. These estimates are not a process RSS measurement. BatchConfig::directory(tasks, directory) stages output in exclusive temporary files in an existing directory, keeping payload RAM proportional to active workers and segment size. File cleanup is best effort on drop.

Output and cancellation

finish_into appends to a vector and restores its length on failure. finish_to_writer consumes the batch and returns the destination with statistics on success. On failure it preserves destination ownership and the number of accepted bytes; the consumed batch cannot retry assembly. The caller controls flushing, file durability, and publication.

Cancellation is cooperative around source reads and segment encoding. A running codec call or blocking source read can delay it. Dropping a batch cancels work without waiting for detached tasks. Worker retention is configurable and defaults to retaining no workers.

The file example uses directory staging and creates a new output file:

cargo run --release --example parallel -- INPUT OUTPUT

See parallel mechanics for lifecycle, resource accounting, source checks, and error propagation.