codas-flow 0.7.2

Low-latency, high-throughput bounded queues ("data flows") for (a)synchronous and event-driven systems, inspired by the LMAX Disruptor and built for codas.

What's Here

This crate provides the flow data structure: A "ring" buffer which concurrent, (a)synchronous tasks can publish data to and receive data from.

flows work kind of like the broadcast channels in Tokio, with some key differences:

1. Zero-Copy Multicast Reads: Every data published to a flow is immediately available to each subscriber, in parallel, with no copies or cloning.

2. Lock-Free by default: No locks or mutexes are used when publishing or receiving data from the flow on supported targets.

3. Broad Compatibility:

• no-std by default.
• async and synchronous APIs.
• async functionality doesn't depend on a specific runtime or framework (not even futures!).

flows work wherever channels or queues would work, but they're built specifically for systems that need the same data processed concurrently (or in parallel) by multiple tasks.

Examples

Flows are created with Flow::new, which returns a tuple of (flow, subscribers):

use codas_flow::*;

// Create a flow with a capacity of 32 strings,
// and one subscriber.
let (mut flow, [mut sub]) = Flow::<String>::new(32);

// Publish "Hello!" to the next data sequence in the flow.
let seq = flow.try_next().unwrap();
seq.publish("Hello!".to_string());

// Receive the next published data sequence from the flow.
let seq = sub.try_next().unwrap();
assert_eq!("Hello!", *seq);

Data is published into a flow via Flow::try_next (or await Flow::next), which returns an UnpublishedData reference. Once this reference is published (via UnpublishedData::publish), or dropped, it becomes receivable by every subscriber.

Data is received from a flow via FlowSubscriber::try_next (or await FlowSubscriber::next), which returns a PublishedData reference.

Subscribers

Using slice patterns, any number of subscribers can be returned by Flow::new:

use codas_flow::*;

// Create a flow with a capacity of 32 strings,
// and 2 subscribers.
let (mut flow, [mut sub_a, mut sub_b]) = Flow::<String>::new(32);

New subscribers cannot be added to an active flow. To overcome this challenge, any subscriber can be wrapped in a Stage.

Stages

A stage is a dynamic group of data processors that share a single subscriber:

# use core::sync::atomic::Ordering;
# use portable_atomic::AtomicU64;
# use portable_atomic_util::Arc;
use codas_flow::{*, stage::*};

// Create a flow.
let (mut flow, [mut sub]) = Flow::<String>::new(32);

// Wrap the subscriber in a processing stage.
let mut stage = Stage::from(sub);

// Add a data processor to the stage; an indefinite 
// number of processors can be added to a stage, even
// while the flow is active.
let calls = Arc::new(AtomicU64::new(0));
let closure_calls = calls.clone();
stage.add_proc(move |proc: &mut Proc, data: &String| {
   assert_eq!("Hello!", *data);
   closure_calls.add(1, Ordering::SeqCst);
});

// Publish "Hello!" to the next data sequence in the flow.
let seq = flow.try_next().unwrap();
seq.publish("Hello!".to_string());

// Run processors for a set of data in the flow.
stage.proc();
assert_eq!(1, calls.load(Ordering::SeqCst));

Stages only receive data from the flow when one of the Stage::proc* functions is invoked; refer to the Stage docs for more information.

Lock-Free Targets

This crate uses AtomicU64 to coordinate flow access without locks. This type is lock-free where possible, but may use locks on some platforms or compile targets.

This section contains a list of the primary targets supported by this crate, along with their support for lock-free behavior.

1 WASM targets don't technically support atomic instructions. However, because WASM code is executed in a single-thread, regular variables are simply substituted for their atomic counterparts, enabling full lock-free support.

2 Confirmation required; a safe assumption is that 32-bit targets don't support atomic operations on 64-bit values.

Relative Performance ("Benchmarks")

First, a caveat: Benchmarks are quite noisy, and shouldn't be used as absolute references--particularly benchmarks from different platforms. Instead, these benchmarks should be used to understand the relevant performance of different scenarios and frameworks on the same platform.

Each benchmark table contains a Scenario column, which describes the number of producers and consumers in the test:

• Many(1) scenarios use a multi-producer/consumer capable channel with just one producer and/or consumer.

• Many(N) scenarios use a multi-producer/consumer capable channel with two or more producers and/or consumers.

One outlier in the benchmarks is the Tokio broadcast channel. This channel has strong performance in the Many(N):* scenarios, but doesn't support back pressure: If consumers fall behind the producer(s), the channel will drop messages entirely to make room for new ones. For this reason, we list it for completeness, but don't consider it a true apples-to-apples comparison.

Benchmarks on a 13" MacBook Air M3 (2024, 16GB):

Benchmarks on a a Hetzner CCX23 AMD EPYC, 4 dedicated vCPUs, 16GB:

License

Copyright © 2024 - 2026 With Caer, LLC and Alicorn Systems, LLC.

Licensed under the MIT license. Refer to the license file for more info.