Crossfire

High-performance spsc/mpsc/mpmc channels.

It supports async context, or communicates between async-blocking context.

Implemented with lockless in mind, low level is based on crossbeam-channel. For the concept, please refer to wiki.

Stability and versions

Crossfire v1.0 has been released and used in production since 2022.12. Heavily tested on X86_64 and ARM.

V2.0 has refactored the codebase and API at 2025.6. By removing generic types of ChannelShared object in sender and receiver, it's easier to remember and code.

V2.0.x branch will remain in maintenance mode. Further optimization might be in v2.x_beta version until long run tests prove to be stable.

Performance

We focus on optimization of async logic, outperforming other async capability channels in most cases.

Due to context switching between sleep and wake, there is a certain overhead on async context over crossbeam-channel which in blocking context.

Benchmark is written in criterion framework. You can run benchmark by:

cargo bench --bench crossfire

More benchmark data is on wiki. Here are some of the results:

APIs

modules and functions

There are 3 modules: spsc, mpsc, mpmc, providing functions to allocate different types of channels.

The SP or SC interface, only for non-concurrent operation, it's more memory efficient than MP or MC implementations, and sometimes slightly faster.

The return types in these 3 modules are different:

• mpmc::bounded_blocking() : (tx blocking, rx blocking)

• mpmc::bounded_async() : (tx async, rx async)

• mpmc::bounded_tx_async_rx_blocking() : (tx async, rx blocking)

• mpmc::bounded_tx_blocking_rx_async() : (tx blocking, rx async)

• mpmc::unbounded_blocking() : (tx non-blocking, rx blocking)

• mpmc::unbounded_async() : (tx non-blocking, rx async)

NOTE : For bounded channel, 0 size case is not supported yet. (Temporary rewrite as 1 size).

Types

NOTE: For SP / SC version [AsyncTx], [AsyncRx], [Tx], [Rx], is not Clone, and without Sync, Although can be moved to other thread, but not allowed to use send/recv while in Arc. (Refer to the compile_fail examples in type document).

Error types

Error types are re-exported from crossbeam-channel: TrySendError, SendError, TryRecvError, RecvError

Feature flags

• tokio: Enable send_timeout, recv_timeout API for async context, based on tokio.

• async_std: Enable send_timeout, recv_timeout API for async context, base on async-std.

Async compatibility

Tested on tokio-1.x and async-std-1.x, by default we do not depend on any async runtime.

In async context, tokio-select! or future-select! can be used. Cancelling is supported. You can combine recv() future with tokio::time::timeout.

When feature "tokio" or "async_std" enable, we also provide two additional functions:

send_timeout which will return the message failed to sent in SendTimeoutError.

recv_timeout

While using MAsyncTx or MAsyncRx, there's memory overhead to pass along small size wakers for pending async producer or consumer. Because we aim to be lockless, when the sending/receiving futures are cancelled (like tokio::time::timeout()), might trigger immediate cleanup if non-conflict conditions are met. Otherwise will rely on lazy cleanup. (waker will be consumed by actual message send and recv).

Never the less, for close notification without sending anything, I suggest that use tokio::sync::oneshot instead.

Usage

Cargo.toml:

[dependencies]
crossfire = "2.0"

Example with tokio::select!

extern crate crossfire;
use crossfire::*;
#[macro_use]
extern crate tokio;
use tokio::time::{sleep, interval, Duration};

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::bounded_async::<i32>(100);
    for _ in 0..10 {
        let _tx = tx.clone();
        tokio::spawn(async move {
            for i in 0i32..10 {
                let _ = _tx.send(i).await;
                sleep(Duration::from_millis(100)).await;
                println!("sent {}", i);
            }
        });
    }
    drop(tx);
    let mut inv = tokio::time::interval(Duration::from_millis(500));
    loop {
        tokio::select! {
            _ = inv.tick() =>{
                println!("tick");
            }
            r = rx.recv() => {
                if let Ok(_i) = r {
                    println!("recv {}", _i);
                } else {
                    println!("rx closed");
                    break;
                }
            }
        }
    }
}