disruptor-rs

Disruptor

This library is a low latency, inter-thread communication library written in Rust.

It's heavily inspired by the brilliant Disruptor library from LMAX.

Use it when you want to trade CPU resources for lower latency and higher throughput compared to e.g. Crossbeam or std::sync::mpsc channels.

Contents

• Getting Started
• Features
• Patterns
• Design Choices
• Correctness
• Performance
• Related Work
• Used By
• Contributions
• Roadmap

Getting Started

Add the following to your Cargo.toml file:

disruptor = "4.4.0"

To read details of how to use the library, check out the documentation on docs.rs/disruptor.

Processing Events

There are two ways to process events:

1. Supply a closure to the Disruptor and let it manage the processing thread(s).
2. Use the EventPoller API where you can poll for events (and manage your own threads).

Both have comparable performance so use what fits your use case best. (See also benchmarks.)

Single and Batch Publication With Managed Threads

Here's a minimal example demonstrating both single and batch publication. Note, batch publication should be used whenever possible for best latency and throughput (see benchmarks below).

…

Pinning Threads and Dependencies Between Processors

The library also supports pinning threads on cores to avoid latency induced by context switching. A more advanced usage demonstrating this and with multiple producers and multiple interdependent consumers could look like this:

…

Processors with State

If you need to store some state in the processor thread which is neither Send nor Sync, e.g. a Rc, then you can create a closure for initializing that state and pass it along with the processing closure when you build the Disruptor. The Disruptor will then pass a mutable reference to your state on each event. As an example:

…

Event Polling

An alternative to storing state in the processor is to use the Event Poller API:

…

Note, the Event Poller API can be interleaved with all the other builder functionality so you can mix and match closures submitted to managed threads, interdependencies, etc.

Out-of-Band Branches

It's possible to create an out-of-band branch with an Event Poller. This can be used to process events in parallel with the main stages.

As an example, a journaler could persist events while multiple other stages are processing in parallel:

           /-- A -> B -> C --\
Producer -> ------- J --------> D

Here, A, B and C process events in parallel with J. D only processes events after both J and C are done.

In code it looks like this:

…

Features

• Single Producer Single Consumer (SPSC).
• Single Producer Multi Consumer (SPMC) with consumer interdependencies.
• Multi Producer Single Consumer (MPSC).
• Multi Producer Multi Consumer (MPMC) with consumer interdependencies.
• Single Consumer optimized barrier.
• Busy-spin wait strategies.
• Sleep wait strategy for development/non-latency sensitive applications.
• Batch publication of events.
• Batch consumption of events.
• Event Poller API.
• Out-of-band branches with Event Pollers.
• Thread affinity can be set for the event processor thread(s).
• Set thread name of each event processor thread.

Patterns

A Disruptor with Different Event Types

Let's assume you have multiple different types of producers that each publish distinct events. It could be an exchange where you receive e.g. client logins, logouts, orders, etc.

You can model this by using an enum as event type:

enum Event {
    Login{id: i32},
    Logout{id: i32},
    Order{user_id: i32, price: f64, quantity: i32},
    Heartbeat,
}

Then you can differentiate on different event types in your processor:

…

Splitting Workload Across Processors

Let's assume you have a high ingress rate of events and you need to split the work across multiple processors to cope with the load. You can do that by assigning an id to each processor and then only process events with a sequence number that modulo the number of processors equal the id. Here, for simplicity, we split it across two processors:

let processor0 = |e: &Event, sequence: Sequence, _end_of_batch: bool| {
    if sequence % 2 == 0 {
        // Process event.
    }
}
let processor1 = |e: &Event, sequence: Sequence, _end_of_batch: bool| {
    if sequence % 2 == 1 {
        // Process event.
    }
}

This scheme ensures each event is processed once.

Dynamic Number of Consumers and Producers

Sometimes, you don't know at compile time how many producers or consumers you need at runtime. It could be your data processing pipeline that has different number of sources and sinks - based on configuration - or your trading engine that has a variable number of input and output feeds.

To accommodate that, you can use this pattern:

…

Design Choices

Everything in the library is about low-latency and this heavily influences all choices made in this library. As an example, you cannot allocate an event and move that into the ringbuffer. Instead, events are allocated on startup to ensure they are co-located in memory to increase cache coherency. However, you can still allocate e.g. a struct and move ownership to a field in the event on the Ringbuffer.

There's also no use of dynamic dispatch - everything is monomorphed.

Correctness

This library needs to use Unsafe to achieve low latency. Although the absence of bugs or data races cannot be guaranteed without formal proofs, these approaches have been used to eliminate logic errors:

• Minimal usage of Unsafe blocks.
• High test coverage.
• All tests are run on Miri in CI/CD.
• Verification in TLA+ (see the verification/ folder).

Furthermore, the code is carefully handcrafted by me with only a little bit of AI intellisense (which was carefully reviewed).

Performance

The SPSC and MPSC Disruptor variants have been benchmarked and compared to Crossbeam. See the code in the benches/spsc.rs and benches/mpsc.rs files.

The results below of the SPSC benchmark are gathered from running the benchmarks on a 2016 Macbook Pro running a 2,6 GHz Quad-Core Intel Core i7. So on a modern Intel Xeon the numbers should be even better. Furthermore, it's not possible to isolate cores on Mac and pin threads which would produce even more stable results. This is future work.

If you have any suggestions to improving the benchmarks, please feel free to open an issue.

To provide a somewhat realistic benchmark not only burst of different sizes are considered but also variable pauses between bursts: 0 ms, 1 ms and 10 ms.

The latencies below are the mean latency per element with 95% confidence interval (standard criterion settings). Capturing all latencies and calculating misc. percentiles (and in particular the max latency) is future work. However, I expect the below measurements to be representative for the actual performance you can achieve in a real application.

No Pause Between Bursts

Latency:

Throughput:

1 ms Pause Between Bursts

Latency:

Throughput:

10 ms Pause Between Bursts

Latency:

Throughput:

Conclusion

There's clearly a difference between the Disruptor and the Crossbeam libs. However, this is not because the Crossbeam library is not a great piece of software. It is. The Disruptor trades CPU and memory resources for lower latency and higher throughput and that is why it's able to achieve these results. The Disruptor also excels if you can publish batches of events as demonstrated in the benchmarks with bursts of 10 and 100 events.

Both libraries greatly improves as the burst size goes up but the Disruptor's performance is more resilient to the pauses between bursts which is one of the design goals.

Related Work

There are multiple other Rust projects that mimic the LMAX Disruptor library:

1. Turbine
2. Disrustor

A key feature that this library supports is multiple producers from different threads that neither of the above libraries support (at the time of writing).

Used By

Would you like to have your company listed here? Open an issue with the details and I'll add you to the list.

Contributions

You are welcome to open an issue with suggestions for improvements or any questions you might have. Please do this before starting the implementation of new features.

AI generated issues or PRs without your (human) involvement will be closed.

Changes are accepted solely at my discretion and I will focus on whether the changes are a good fit for the purpose and design of this crate.

Roadmap

Empty! All the items have been implemented.