Crate thread_broker

Expand description

Dividing one thread budget between decoding and the work that consumes it.

§The problem

A tool that reads compressed input has to split -t between threads that decompress and threads that do the real work — mapping, aligning, counting. The split matters enormously: measured on a 150 M read-pair FASTQ archive at a fixed 64 threads, it was worth 5.75x, while doubling the whole budget bought 9%.

It is also unpredictable. The right answer moves by an order of magnitude with the workload — measured per-thread consumption ranged 0.064 GB/s on a 96.3 M-k-mer index to 0.43 GB/s on a 1.5 M-k-mer one, a 6.7x spread — so no constant spans it, and every attempt to predict it from a pre-run measurement landed 7-60% off the optimum depending on the workload.

§The approach

Do not search for the split. Solve for it.

Writing s_p and s_c for the two stages’ per-item costs in thread-time, a budget N finishes both sides together when d / s_p = (N - d) / s_c, so

    d* = N * s_p / (s_p + s_c)

and neither cost has to be measured on its own. Over a window in which the pipeline moved X items the producer spent X * s_p thread-nanoseconds and the consumer X * s_c, so their ratio of busy times is the ratio of costs exactly — the X cancels. The whole control law is

    d* = N * busy_producer / (busy_producer + busy_consumer)

measured over the same window. No shared unit of work is needed, and a run mixing compressed and uncompressed inputs is handled without a special case: the producer simply does less work, and the split follows.

The one thing that makes any of this valid is measuring busy time, not wall time. Wall time contains blocking; blocking is a function of the split; so a controller built on wall time is measuring its own actuation. Busy time is a property of the work rather than of the current assignment, which is what makes the equation solvable rather than circular. DS2 (Kalavri et al., OSDI’18) calls the distinction true versus observed rate. See Work, where the contract lives.

An earlier version of this crate searched instead — grow whichever side looks starved, one step at a time. It fails for a structural reason worth recording: consumer starvation reaches zero at the balance point and stays zero above it, so the target set is a half-line and the controller has no restoring force on it. Measured, it walked to 47 producer slots of 64 and ran 44% slower than the best split, reporting near-zero error the whole way. See ProducerPressure and notes/thread-broker/design.md.

A model can still be wrong — a producer reading from a saturated disk cannot use the slots it is given — so the jump is guarded: capped against observed saturation, ratified against measured throughput, and reverted if it made things worse. Both requirements — resizing either side mid-run, and both sides reporting their own busy time — are met by paraseq (consumer) and rapidgzip (producer).

Measured against swept optima on a 10x Flex dataset: 5.0% off at -t 32 (one move, settled in 1.4 s) and 3.4% off at -t 64 (three moves, 2.6 s).

§What this crate is not

It is not a thread pool. It owns no threads except the one that samples, and it creates no workers. It reads two signals and calls two setters. The pools themselves belong to the caller.

§Wiring it up

Implement Consumer for the side doing the real work and Producer for the decompressor, then let the broker run for the duration of the job:

let broker = ThreadBroker::builder(mapper, decoders)
    .budget(32)
    .sample_interval(Duration::from_millis(100))
    .build()?;

let running = broker.start()?;
// ... run the job ...
let report = running.finish()?;
tracing::info!(
    "decode settled at {} of 32 threads after {:?}",
    report.final_producer_limit,
    report.time_to_converge,
);

§Minimum viable adoption

A first integration does not need to tune every BrokerConfig field. The minimum contract is:

Define one real execution-slot budget shared by both pools. Construct the pools so every permitted split obeys that budget, choose any valid opening split, and set only .budget(...) plus real consumer/producer floors.
Implement cumulative Work counters on both adapters. Busy time must exclude queue/source blocking; counters must be monotonic; progress must publish often enough that ordinary decision windows are not empty.
Make resize acknowledgement truthful. Consumer::live_threads and Producer::active_slots must continue to include work admitted before a shrink until it actually retires. Batch geometry must let that happen inside BrokerConfig::resize_timeout.
Implement Producer::pressure as a directional classification, not a desired thread count. If the producer has native cumulative busy time, the measurement-mode and monitoring hooks can keep their defaults.
Keep broker lifecycle failure separate from application correctness. A configuration error should fail before work starts; a runtime tuning failure normally leaves the last valid split in force and should not make successfully produced output disappear.

Leave warm-up, sampling, smoothing, blackout, ratification, cap history, deadband, resurvey, regression tolerance, and OpeningPolicy at their defaults initially. Choose a SteadyStatePolicy only from the application’s workload contract: responsive for possible regime changes, or a freeze policy for stable long runs where recurring work should become zero.

Tune cadence or batch/progress granularity only after measurement shows a concrete failure: stale/empty decision windows, shrink acknowledgement near the timeout, poor oracle-gap/convergence results, or material overhead. Qualify a new adapter with canonical-output equality, a pinned-split sweep, source-bound and inelastic cases, a mid-run regime change, and both absolute administrative CPU and whole-process fractional overhead on representative long runs. That is the adopter boundary; modality-specific knobs belong in the application until repeated integrations demonstrate a generic policy.

Structs§

BatchTimer: Accumulates work time at a granularity the broker can actually resolve.
BrokerConfig: Tuning, all of it overridable.
BrokerConfigError: Invalid ThreadBroker configuration.
BrokerError: A broker failure, optionally carrying the report accumulated before it.
BrokerReport: What the broker did, for logging and for tests.
BusyMeter: A shared counter of time spent doing real work.
CompletedItem: Completion guard returned by BatchTimer::completed_item.
EngagementPolicy: Whether engaging a parallel producer can pay for the threads it takes.
Model: The solved model at one instant.
OpeningBracketConfig: Bounds for OpeningPolicy::Bracket.
OpeningBracketReport: Auditable cost and result of OpeningPolicy::Bracket.
ProducerMeasurementStats: Bounded diagnostics for producer-side busy-time measurement.
Rejection
ResizeError: A pool refused a resize request.
RunningBroker: A broker sampling in the background. Dropping it stops the loop.
ThreadBroker: A configured, not-yet-running broker.
ThreadBrokerBuilder: Builder for ThreadBroker.
Work: Cumulative work done by one stage.

Enums§

BrokerErrorKind: Why a broker could not start or complete safely.
BrokerPhase
OpeningBracketOutcome
OpeningPolicy: How the broker treats the caller’s initial split.
ProducerCapReason: Evidence behind an empirical producer-concurrency ceiling.
ProducerMeasurementMode: Requested fidelity for an adapter that reconstructs cumulative busy time from instantaneous observations.
ProducerPressure: Whether more producer concurrency would buy anything.
ResizeSide: Which pipeline side was involved in a broker operation or failure.
SteadyStatePolicy: What the broker does after it has established a stable split.

Constants§

DEFAULT_FLUSH_EVERY: Records published every this many items by default.

Traits§

Consumer: The side that consumes decoded bytes: mapping, aligning, counting.
Producer: The side that produces decoded bytes: a decompressor or pool of them.