repository·main·Indexed 21 days ago
https://github.com/nicholassm/disruptor-rsA low-latency, inter-thread communication library for Rust inspired by the LMAX Disruptor. It utilizes a ringbuffer to achieve high throughput and low latency, offering features such as thread pinning, consumer dependency management, and support for both single and multi-producer configurations. Version 4.3.0.
There are two primary ways to process events:
EventPoller API to manually poll for events. This allows you to manage your own threads and execution loop.Both methods offer comparable performance.
You can create parallel processing branches using .new_branch() and .join(branch). This allows an 'out-of-band' processor (like a journaler) to run in parallel with the main pipeline. You can then rejoin the branch into the main flow, ensuring subsequent stages only run after both the main flow and the branch are complete.
use disruptor::*;
let mut builder = disruptor::build_single_producer(64, factory, BusySpin);
// 1. Create a branch
let branch = builder.new_branch();
// 2. Define main pipeline
let builder = builder
.handle_events_with(a).and_then()
.handle_events_with(b).and_then()
.handle_events_with(c);
// 3. Join branch back into main flow
// This returns a poller for the branch and the builder for the main flow
let (mut event_poller_j, builder) = builder.join(branch);
// 4. Ensure 'd' only runs after 'c' AND 'j' (the branch)
let builder = builder
.and_then()
.handle_events_with(d);The disruptor-rs library is optimized for low latency through several key design choices:
Add disruptor to your Cargo.toml to use this low-latency, inter-thread communication library. It is designed to trade CPU resources for lower latency and higher throughput compared to std::sync::mpsc or Crossbeam.
disruptor = "4.3.0"If the number of producers or consumers is only known at runtime (e.g., based on configuration), use the build_multi_producer builder pattern. You can iteratively call new_event_poller() to add consumers and then clone() the final producer to create multiple producers.
Note: You must have at least one consumer and one producer.
fn build_disruptor<E, F, W>(
size: usize,
event_factory: F,
wait_strategy: W,
producer_count: usize,
consumer_count: usize,
) -> Result<(
Vec<MultiProducer<E, MultiConsumerBarrier>>,
Vec<EventPoller<E, MultiProducerBarrier>>,
)>
where
F: FnMut() -> E,
E: 'static + Send + Sync,
W: 'static + WaitStrategy,
{
if producer_count == 0 || consumer_count == 0 {
bail!("Must have at least one consumer and producer.");
}
let mut builder = disruptor::build_multi_producer(size, event_factory, wait_strategy)
.with_multi_consumer();
// Create consumers (EventPollers):
let mut consumers = Vec::new();
for _ in 0..consumer_count {
let (poller, next_builder) = builder.new_event_poller();
consumers.push(poller);
builder = next_builder;
}
// Create producers:
let producer = builder.build();
let mut producers = Vec::new();
for _ in 1..producer_count {
producers.push(producer.clone());
}
producers.push(producer);
Ok((producers, consumers))
}The repository includes TLA+ models to verify Single-Producer Multi-Consumer (SPMC) and Multi-Producer Multi-Consumer (MPMC) scenarios. To ensure verifications complete in under a minute, use the following model configurations:
MaxPublished <- 10Size <- 8Writers <- { "w" }Readers <- { "r1", "r2" }NULL <- [ model value ]MaxPublished <- 10Size <- 8Writers <- { "w1", "w2" }Readers <- { "r1", "r2" }NULL <- [ model value ]To avoid latency induced by context switching, you can pin processor threads to specific CPU cores using .pin_at_core(core_id). You can also define dependencies between processors using .and_then().
.handle_events_with(h1): Registers a processor..and_then(): Ensures the subsequent processor only runs after the preceding one(s) have completed.use disruptor::*;
// ... setup factory and handlers ...
let mut producer = disruptor::build_multi_producer(64, factory, BusySpin)
.pin_at_core(1).handle_events_with(h1)
.pin_at_core(2).handle_events_with(h2)
.and_then()
.pin_at_core(3).handle_events_with(h3)
.build();The MultiProducer uses a SharedProducer protected by a Mutex to track the number of active producer handles.
clone() a MultiProducer, the internal counter is incremented. If the number of clones exceeds a safety threshold (i64::MAX/2), the process will abort to prevent overflow.MultiProducer implements Drop. When the last producer handle is dropped (the counter reaches 0), the producer sets a shutdown_at_sequence to signal consumers to stop. This ensures a clean lifecycle when all publishers are finished.You can create Directed Acyclic Graph (DAG) topologies by branching the event flow.
new_branch() on the builder. This returns a Branch<E, B>.join(branch) to merge the branch back into the main flow. This returns an EventPoller for the branch and a new builder state to continue the main flow. The main flow will now depend on the branch's progress.You can create parallel execution paths (branches) that diverge from the main pipeline and later merge back (join).
builder.new_branch() to create a new branch. This returns a branch handle and a new builder instance.builder.join(branch_handle) to merge a branch back into the main flow. This returns an EventPoller for that branch and a new builder instance.builder.and_then() to create a dependency, ensuring downstream stages only proceed after upstream stages have advanced.// 1. Create a branch
let b1 = builder.new_branch();
// 2. Join it back to the main flow
let (mut ep_b1, builder) = builder.join(b1);
// 3. Create a dependency (A depends on the current state of the pipeline)
let (mut ep_a, builder) = builder.and_then().new_event_poller();The Disruptor supports Directed Acyclic Graph (DAG) topologies using branches.
Call .new_branch() on the MPBuilder. This returns a Branch<E, B>. This branch runs in parallel with the main flow.
To merge a branch back into the main flow, call .join(branch) on the builder. This returns the EventPoller for the branch and a new builder state that incorporates the branch's cursor, allowing subsequent stages to depend on the branch's progress.
You can define complex consumer topologies using the builder pattern:
Multiple calls to .handle_events_with(...) add consumers that run in parallel, all receiving every event.
Use .and_then() to create a dependency. A consumer added after .and_then() will only receive events after the preceding consumer(s) have processed them.
Use .pin_at_core(core_id) to pin a specific consumer thread to a CPU core for lower latency.
Use .thread_name("name") to set the thread name for debugging.
You can create a Directed Acyclic Graph (DAG) where a branch runs in parallel with a pipeline and is later joined back:
.new_branch() to create a Branch..join(branch) to merge that branch back into the main dependency chain.EventPoller for the branch, ensuring safe access to that specific path.// Example: Pinned, dependent consumers
let mut producer = build_multi_producer(64, factory, BusySpin)
.pin_at_core(1).handle_events_with(h1)
.pin_at_core(2).handle_events_with(h2)
.and_then() // h3 depends on h1 and h2
.pin_at_core(3).handle_events_with(h3)
.build();