Parallel compression

This subsystem is omitted when no_std is enabled. See feature boundaries for the alloc-backed API that remains available.

The caller schedules tasks that encode fixed segments, each independently of the others’ encoding: at qualities 2-9 a segment may copy from the input before it (its history), which the decoder holds by then. Input that arrives in chunks uses the streaming session over the same tasks and workers. Private fragment encoders produce aligned non-final blocks, and the batch assembles them in source order into one stream. The API supports qualities 0–11 with standard windows; dictionary, Large Window, and framing extensions are unsupported.

Public and ownership boundaries

compressor::parallel exposes ParallelCompressor, ParallelConfig, BatchConfig, validated SegmentSize and TaskCount, memory/directory staging, source wrappers (including SeekSource<R>), task/batch types, polling, statistics, retention, and errors. The compressor module is public; its core remains private. Serial APIs are also re-exported at the crate root.

Validated integers use TryFrom<usize> and get(); standard values use SegmentSize::DEFAULT, TaskCount::ONE, and Default. Configuration/staging conversions use From. Source wrappers use From<Arc<[u8]>> and TryFrom<File>. These trait-based spellings follow repository API conventions. There is no mandatory executor trait, runtime dependency, or hidden thread pool. Rayon is a development dependency for examples of scheduling in tests/benchmarks. tempfile is a development dependency for test and benchmark fixtures only. The private parallel::core::spool module uses the standard library for exclusive spool creation and cleanup.

Public source rustdoc includes a custom immutable snapshot with a version token, absolute reads through SeekSource, and a compile-checked FileSource::open example. Source identity is compared as caller-supplied bytes; it is not a library-computed content hash. An open file handle does not freeze file contents, and callers must keep input immutable throughout a batch. FileSource uses positional reads on Unix and Windows and returns Unsupported elsewhere.

graph TD
    Public[compressor::parallel public wrappers] --> Planner[parallel::core::Compressor and Plan]
    Public --> Batch[parallel::core::batch]
    Public --> Task[parallel::core::task]
    Public --> Source[RandomAccessSource / SeekSource / FileSource / ArcBytesSource]
    Planner --> Reservoir[exclusive idle Workers]
    Task --> Worker[owned Worker: fragment codec, serial codec, input buffer]
    Worker --> Fragment[compressor::core::fragment]
    Fragment --> Driver[serial core::driver::Encoder]
    Driver --> Families[Fast / Greedy / HQ]
    Families --> Dispatch[Encoder true: dispatch! retained Backend per block]
    Task --> Artifact[private memory artifact or Spool]
    Artifact --> Spool[parallel::core::spool: File + cleanup guard]
    Task --> Slot[one completion slot per task]
    Slot --> Batch
    Batch --> Destination[Vec / Write]

A task’s Job is either Planned (a batch task’s contiguous segments, read from its input) or Streamed (one segment a stream session already moved into the worker); both run through the same Task::run and per-segment encode helper.

Each ScopedParallelTask<'input> owns its worker, artifact, immutable input handle, cancellation observer and completion capability. It never borrows the batch or parent compressor. OwnedParallelTask is the 'static specialization; OwnedParallelBatch similarly specializes the shared batch implementation. This supports both scoped borrowing and detached execution without duplicated engines. ParallelCompressor is Send, non-Clone, and owns no global state.

Planning, source reads, and memory

Fixed segment sizes range from 64 KiB to 16 MiB, with a 4 MiB default. Boundaries are i * segment_size .. min((i + 1) * segment_size, source_len) using u64 file coordinates. Effective task count is min(requested, segments). Contiguous balanced grouping gives the first remainder tasks one additional segment. Neither grouping nor completion order changes bytes.

Empty input creates no tasks and uses canonical serial encoding. A nonempty input below minimum_parallel_size (default 8 MiB) uses serial encoding only if it also fits one segment. That fallback still creates one caller-run task. Restricting fallback to one segment prevents user-selected thresholds from causing a whole-file allocation. Setting the threshold to zero forces independent parts.

Borrowed slices go directly into codec calls. Detached sources fill one retained segment buffer per task, reading each segment’s history together with it (Plan::history_start), so a source read covers history + segment bytes. FileSource retains an open regular file and uses Unix read_at or Windows seek_read, retrying interruptions and short reads. Other platforms return Unsupported for positional reads. File handles are never used with a shared seek cursor. Metadata snapshots include length and modification time, plus device, inode and change time on Unix. Metadata checks cannot prove immutability against every in-place write; callers can provide stronger immutable snapshot sources. Arbitrary blocking source calls cannot be forcibly interrupted.

The active-worker estimate scales segment + history rather than the segment alone, since a worker’s window and match finder hold both. Planning checks staged-byte bounds, descriptor sizes, active-worker estimates, completion metadata, assembly scratch and currently retained workspaces before returning tasks. Staged payload allowance is 2 * input + 1024 * max(segments,1). Memory staging also charges descriptor storage to its explicit limit. A memory artifact reserves its first part exactly, so a single-segment task hands over a buffer without spare capacity, and then doubles its capacity, capped at the task’s staged-payload bound, so the accounted bound also bounds its spare capacity. Reserving each part exactly reallocated on every append; with four tasks on 64 KiB segments of a 256 MiB input that took 143-191 ms against 114-119 ms doubling (4 MiB segments: about 4% apart). BatchConfig::auto(tasks) selects the same memory staging with usize::MAX as the caller ceiling. Preparation still computes checked per-task payload bounds, checks payload plus descriptors fits usize, and enforces the aggregate limit. It neither queries free RAM nor switches storage backends. Source length comes from the slice or the source’s len(); task count is capped by segment count.

ParallelSizeEstimate::maximum_staging_memory_bytes() exposes payload plus segment_count * size_of::<Descriptor>(), using the same private checked helper as memory preflight. It returns a u64 bound for either staging kind; callers convert it with usize::try_from for an explicit memory budget. The existing maximum_staged_bytes field remains payload-only. Overflow propagates as ParallelEncodeError::SizeOverflow, including for caller-modified estimates. Descriptor storage remains included once in the aggregate workspace estimate.

flowchart TD
    Input[Slice length or source len] --> Plan[Checked segments and effective tasks]
    Config[auto tasks or explicit memory ceiling] --> Plan
    Plan --> Bound[Payload bound plus descriptor storage]
    Bound --> Fit{Fits usize and caller ceiling?}
    Fit -->|no| Error[SizeOverflow or MemoryStagingLimit]
    Fit -->|yes| Aggregate{Aggregate memory limit satisfied?}
    Aggregate -->|no| WorkerError[WorkerMemoryLimit]
    Aggregate -->|yes| Prepare[Create bounded artifacts and caller-run tasks]
    Bound --> Estimate[maximum_staging_memory_bytes]

Workspace estimates are deliberately conservative: per task, 4 MiB + 16 times segment size for q0/q1, 64 MiB + 128 times segment size for q2–q4, and 256 MiB + 256 times segment size above q4. These ceilings are not measured RSS. File staging keeps payload RAM proportional to active workers and segment size; descriptor memory is proportional to segment count and included in the aggregate ceiling.

Generic source entry point and seekable readers

prepare_source<S, T> accepts S: RandomAccessSource + ?Sized and T: Into<Arc<S>>. Owned sources, shared concrete sources, custom conversions, and Arc<dyn RandomAccessSource> use this single entry point. FileSource is a source adapter. Arc and custom conversion arguments may need explicit source types due to Into ambiguity: prepare_source::<FileSource, _>(shared_file, config) or prepare_source::<dyn RandomAccessSource, _>(erased_source, config).

The conversion occurs once before planning. A private sized SharedSource<S> bridge lets both sized and unsized sources enter the existing erased core path. It forwards length, identity and reads. Tasks clone the outer shared handle; source bytes are never copied by this bridge. The bridge adds one small shared allocation per prepared batch, not per task or segment.

SeekSource<R> takes ownership through From<R>. For R: Read + Seek + Send + 'static, it implements RandomAccessSource without requiring R: Sync. The public wrapper delegates all locking and I/O to parallel::core::source. Each length query holds the mutex while seeking to the end. Each range read holds that same mutex for its absolute seek and the entire read_exact call. The reader’s original cursor is ignored; all future reads seek explicitly. A short read or interruption is handled by Read::read_exact; seek/read errors propagate as I/O errors into existing batch metadata/read errors. A reader panic poisons the mutex and subsequent access returns an I/O error. Worker panic handling continues to discard the affected codec workspace.

sequenceDiagram
    participant Caller
    participant Public as prepare_source / SeekSource
    participant Core as core::source
    participant Reader as R: Read + Seek + Send
    participant Task
    Caller->>Public: source.into() to Arc<S>
    Public->>Core: wrap shared source for type erasure
    Core->>Reader: lock, seek End(0), unlock
    Note over Core,Reader: planning snapshots current length
    Task->>Core: read_exact_at(offset, segment buffer)
    Core->>Reader: lock, seek Start(offset)
    Core->>Reader: read_exact (retry interrupted/short reads)
    Core-->>Task: unlock, bytes or I/O error
    Task->>Task: compress without the reader lock

Seekable input reads are serialized; task compression still overlaps. No metadata identity can be inferred from the generic traits, so this adapter provides live length checks only and callers must keep bytes immutable. FileSource retains positional reads and file identity verification; it remains the preferred file adapter when concurrent input reads are desired.

Fragment format and encoder state

Format plan version 1 uses uniform two-byte raw prefixes. Qualities 0-1 and 10-11 write explicit distances throughout every segment; qualities 2-9 write a short distance code only for a cache slot the segment’s own copies filled (see Recent distances), and copies from the segment’s history as from earlier input of the same stream. It follows RFC 7932 sections 11.2 and 11.3. Every part starts at a byte boundary (except the first part also owns the window header), contains only non-final data blocks, and ends aligned. Alignment metadata uses the existing flush writer’s when-needed empty metadata blocks. The assembler only copies validated bytes and appends the final-empty byte 0x03 at the proven byte boundary. It never parses or edits complete streams.

sequenceDiagram
    participant Task
    participant Fragment as FragmentEncoder
    participant Codec as retained Encoder
    participant Artifact
    Task->>Fragment: encode(history, segment bytes, first role)
    Fragment->>Codec: reset semantic state, preserve allocations
    Fragment->>Codec: prepend_history(history) (greedy only)
    Note over Codec: history fills the window and match finder, emits nothing
    Fragment->>Codec: begin_fragment(prefix)
    Note over Codec: greedy fills its four remembered distances with -16
    Note over Codec: greedy/HQ seed local ring and prior bytes via two-byte flush
    Fragment->>Fragment: optional header + raw min(length,2) prefix
    loop complete body input blocks
        Fragment->>Codec: non-final encode, flush final body block
        Codec-->>Fragment: completed bytes
    end
    Fragment->>Codec: verify no pending partial bits
    Fragment-->>Task: sealed AlignedFragment
    Task->>Artifact: append bytes + source/segment descriptor

Recent distances

The decoder’s distance cache carries over between meta-blocks, so at a fragment’s start it holds whatever the fragments before it left there, which the encoder cannot know. Greedy fragments make those slots unusable instead of avoiding the cache altogether: GreedyEncoder::begin_fragment writes -16 to the four remembered distances (and to saved_dist_cache, which a re-encoded meta-block restores). A copy distance is at least 1, so neither an exact match nor a ±1..3 neighbour of -16 ever equals it, compute_distance_code never returns a short code for such a slot, and the searcher’s cached-distance probe rejects the position it computes. Every copy pushes its distance as in a serial stream; after four distinct pushes the cache is entirely the fragment’s own and the rest of the fragment codes distances exactly as serial encoding does.

flowchart TD
    Start[begin_fragment] --> Seed["dist_cache[0..4] = -16<br/>saved_dist_cache = -16"]
    Seed --> Copy{copy at distance d}
    Copy --> Code["compute_distance_code(d, cache)"]
    Code --> Slot{"d equals, or is ±1..3 of,<br/>a filled slot?"}
    Slot -- yes --> Short[short code 0-15]
    Slot -- "no, or the slot still holds -16" --> Full["explicit code d + 15"]
    Short --> Push{code > 0}
    Full --> Push
    Push -- yes --> Shift[shift d into dist_cache]
    Push -- no --> Copy
    Shift --> Copy

History

A part is decoded after every part before it, so the decoder already holds the input before a segment. Greedy fragments let copies reach into the last Compressor::history_len bytes of it: a quarter of a segment, capped at the window’s backward limit (1 << lgwin) - 16, and nothing at the other qualities. GreedyEncoder::prepend_history mirrors brotli 1.0’s BrotliEncoderSetCustomDictionary: it trims the history to the backward limit, copies it into the ring buffer, sets input_pos, last_processed_pos and last_flush_pos past it and the prior bytes to its last two, prepares the match finder for a stream that is not one-shot, and stores the history positions whose lookahead lies inside it (MatchFinder::prepend, HasherPrependCustomDictionary); the boundary positions are stitched when the segment arrives, as between any two blocks. No meta-block is written for the history.

Unlike the reference, prepend skips positions inside short-period runs (history_ranges): where the history repeats a period of at most 1 KiB for more than four periods, only the run’s last four periods are stored. That is the rule commit_match applies to a stream’s own copies (a copy shorter than four times its distance stores everything, a longer one only its last 4 * distance positions), so the buckets hold what a serial stream’s would. Storing every position put all of a zero run or a repeated 256-byte record under a few keys, and every search there measured each candidate to the full run length: 1 MiB of repeated bytes 0-255 at q9 took 7.8 ms instead of 0.4 ms. Runs are found through a 4096-entry table of each 4-byte key’s last position and confirmed eight bytes at a time; each byte is measured at most once.

Every part shares the configured lgwin, which the first part’s header declares, and the encoder never emits a distance beyond min(position, backward limit), its position counting the history. The decoder’s position is at least as far, so such a copy is an ordinary backward reference there, never a static-dictionary one. The size hint covers history and segment, so the match finder is sized for every position it stores.

flowchart LR
    Input[input before the segment] --> Tail["last history_len bytes<br/>(batch: Plan::history_start,<br/>stream: history buffer)"]
    Tail --> Prepend["prepend_history:<br/>ring, positions, prior bytes,<br/>prepare match finder"]
    Prepend --> Runs["history_ranges:<br/>skip all but the last four periods<br/>of short-period runs"]
    Runs --> Store["store_range for kept positions"]
    Store --> Begin
    Begin["begin_fragment:<br/>header off, -16 distances,<br/>two-byte prefix"]
    Begin --> Body["segment body:<br/>copies reach into history"]

A quarter of a segment keeps most of the ratio a whole segment of history gives at about a quarter of the hashing time. With 1 MiB segments (one task, best of 7, Apple M5 Pro): alice29.txt repeated to 4 MiB is 51,261 bytes at q6 with either a 19- or a 22-bit window, against 203,708 for its segments compressed separately; for seq, three copies of /usr/share/dict/words and a mixed binary input the gain was 0-6%, and q9 with a 22-bit window took 11-12% longer than the separate segments (a whole segment of history: up to 45%).

Dictionary overloads are absent. Parallel Large Window configuration returns a typed configuration error.

Completion, cancellation, and reuse

stateDiagram-v2
    [*] --> Prepared
    Prepared --> Running: caller runs consumed task
    Prepared --> Abandoned: dropped task guard
    Running --> Success: commit artifact
    Running --> Error: source / codec / staging failure
    Running --> Panicked: catch_unwind, discard worker
    Running --> Cancelled: cancellation checkpoint
    Success --> Published
    Error --> Published
    Panicked --> Published
    Cancelled --> Published
    Abandoned --> Published
    Published --> [*]: one slot and one completion ID

An unbounded channel carries exactly one small ID per task. Payloads and workers move through single-assignment slots, protected only during ownership transfer. Codec execution holds no mutex. Sending never blocks, so tasks do not need a coordinator to drain their payload or another task to progress.

Each task catches unwinding panics and discards the affected worker. Ordinary errors return resettable workers. The drop guard reports abandonment once. panic=abort remains unrecoverable. Cancellation uses a batch-local relaxed atomic flag, carrying no other published state; checks run around reads and segment encoding and before artifact commit. Successful bytes do not depend on cancellation timing.

The coordinator collects by task ID and chooses the lowest failing ID after all completions, preferring originating errors over cancellation fallout. Repeated poll/wait errors use BatchFailed(Arc<ParallelEncodeError>) to preserve the typed cause and its source chain. Blocking wait requires the caller’s executor to remain runnable. A timeout does not cancel. Dropping a batch cancels, drops untaken tasks, collects only already-published results, and never waits for detached work. Late tasks release their own resources. The default retention count is zero; callers can retain compatible workspaces and trim them by aggregate size.

Encoder reconfiguration

ParallelCompressor::reconfigure(EncoderConfig) delegates to the private core::Compressor, using the same validation path as construction. Invalid encoder settings propagate as ParallelConfigError::Encoder with their source; otherwise valid Large Window settings return UnsupportedParallelWindow. Validation completes before any mutation, so rejection preserves both settings and idle workers. Identical settings preserve workers as well.

Changing encoder settings drops the idle worker reservoir and its allocations. The next batch plans with the new configuration and lazily constructs matching fragment and serial fallback encoders. Parallel configuration and the backend selected at construction remain unchanged; no SIMD detection is repeated. A live batch holds an exclusive borrow of the parent, preventing reconfiguration. After a batch is dropped, detached tasks keep their old exclusive workers and release them on completion; they cannot return them to a reconfigured parent. The parallel Criterion suite validates identical output after reconfiguration and measures reconfiguration plus encoding with one and four tasks, alongside retained-worker and C/Rust serial cases on the same corpora and codec settings.

flowchart TD
    Request[reconfigure encoder settings] --> Validate[Validate encoder and standard window]
    Validate -->|invalid| Error[Return typed error without mutation]
    Validate -->|valid| Compare{Settings changed?}
    Compare -->|no| Keep[Keep idle workers]
    Compare -->|yes| Drop[Release idle workers and store settings]
    Drop --> Next[Next batch plans and creates workers with new settings]
    Keep --> Reuse[Next batch can reuse workers]

Assembly and output failure

Directory staging resolves the existing directory to an absolute path. Each task uses a fresh RandomState to hash candidate names, then atomically opens a read/write file with OpenOptions::create_new(true). Existing files and symlinks are never opened or overwritten. Name collisions retry up to 128 times; other filesystem errors propagate immediately through the existing staging I/O error path. Unix files are created with mode 0600 (subject to the process umask); other platforms use their default file permissions.

The spool owns its File before a separate path cleanup guard in declaration order, so dropping an artifact closes the handle before attempting removal, including on Windows. Success, errors, cancellation and unwinding all use this same ownership cleanup. Removal is best effort because destructors cannot return I/O errors. Callers must keep the staging directory stable and trusted; path replacement, process termination or filesystem errors can prevent cleanup. Spooling adds no SIMD dispatch or changes to encoded bytes.

stateDiagram-v2
    [*] --> Creating: canonicalize directory
    Creating --> Creating: candidate already exists, retry within limit
    Creating --> Failed: filesystem error or collision limit
    Creating --> Open: exclusive read/write create
    Open --> Open: append / validate length / seek and copy
    Open --> Closed: artifact dropped, File drops first
    Closed --> Removed: path guard removes file
    Closed --> LeftBehind: removal fails, error ignored
    Failed --> [*]
    Removed --> [*]
    LeftBehind --> [*]

flowchart TD
    Wait[all task completions] --> Verify[source snapshot + ordered descriptors + spool lengths]
    Verify --> Valid{all valid?}
    Valid -->|no| Error[return typed error, destination untouched]
    Valid -->|yes| Kind{destination}
    Kind --> Vec[reserve exact append size, rollback length on error]
    Kind --> Write[copy in order, count every accepted short write]

Before the first destination write, assembly validates segment order and exact source ranges, artifact offsets/nonzero lengths, total segment count, and actual spool length. Directory artifacts use sequential reads and a 128 KiB copy buffer. Interrupted, short reads/writes and WriteZero are handled. A generic writer failure returns its ownership and exact accepted byte count. It does not flush or sync implicitly and is not retryable through the consumed batch.

Files are ordinary Write destinations. Callers own file creation, buffering, flushing, durability, and any temporary-file publication policy. No path-specific finish API or file commit policy is exposed.

Verification and known gaps

The crate-level documentation in src/lib.rs demonstrates automatic memory staging and two scoped-thread tasks over an 8 MiB input at quality 5. Its doctest checks that both tasks are used and the assembled output is smaller than the repetitive input.

ParallelCompressor rustdoc examples execute the borrowed-slice workflow with scoped standard-library threads and Rayon parallel iterators at quality 0 to keep doctest execution fast: prepare from input.as_bytes(), take and run tasks, then finish into a vector. Both examples propagate configuration, preparation, and assembly errors through a Result. They run as doctests with the default feature set; Rayon remains a development dependency and must be added explicitly by callers who choose that executor.

tests/parallel.rs covers all qualities, standard windows, host backends, modes, segment edges, fallback, source/staging identity, std/Rayon scoped and detached execution, reverse task order, worker reuse, source panics/errors/mutation, abandonment, timeout, cancellation, file output, and every output failure byte for a representative fast stream. Private fragment tests independently decode joined parts, and regressions cover q0 short explicit copies omitted by the serial starter tree and q1 copy-two depth/ordering aliases found by AFL. The full existing serial differential/streaming suite is also required. fuzz/afl adds the parallel target and committed seeds. benches/parallel.rs validates before timing and prints sizes alongside scaling and serial C/Rust measurements; those policies are labeled separately.

Known gaps

• Cancellation occurs around segment encoding, not inside q11 optimization. A running codec call or blocking source read may delay cancellation.
• Segments lose dictionary coding and any history beyond a quarter of a segment (repetition further back, such as a file repeated after 2.4 MB, is not found), and at qualities 0-1 and 10-11 they take no history at all, so their ratio and speed are distinct from serial encoding.
• q0/q1 and q10-11 fragments still write every distance explicitly. Input that repeats one distance compresses several times worse there than serially: 9.3 MB of seq 1 1300000 at q1 with 1 MiB segments gave 4.2 MB instead of 1.0 MB. Greedy qualities 2-9 resume the cache as described in Recent distances.
• Source metadata checks cannot detect every in-place mutation.
• Spool cleanup is best effort; process termination and filesystem errors can leave files behind.
• Positional file reads are implemented for Unix and Windows. Other platforms return Unsupported.