crossfire-rs

Channels can be backed by different queue implementations (flavors) available in the flavor module. Note that while names like Array or List are used across modules, they are distinct type aliases local to their parent module (e.g., spsc::Array vs mpmc::Array).

• List: Uses crossbeam::SegQueue.
• Array: An enum wrapping crossbeam::ArrayQueue (or One if size <= 1). Note: Bounded channels with size 0 are not supported; use size 1.
• One: Optimized for size=1 scenarios using two slots to allow concurrent reader/writer operations.
• Null: A cancellation channel that only wakes up upon closing.

Crossfire provides specialized modules for different concurrency models. Each is optimized for its specific use case:

• spsc: Single-Producer Single-Consumer.
• mpsc: Multi-Producer Single-Consumer.
• mpmc: Multi-Producer Multi-Consumer.
• oneshot: Specialized sender/receiver types for single-message communication (optimized to avoid the overhead of standard Tx/Rx types).
• select: Provides Select<'a> (crossbeam-style type-erased API) and Multiplex (a stream that owns multiple receivers of the same flavor/type).

Add crossfire to your Cargo.toml dependencies to use high-performance lockless SPSC, MPSC, and MPMC channels. The current major version is v3.

[dependencies]
crossfire = "3.1"

To run the project's internal tests, use the make command. Note that due to the high performance and pressure placed on async runtimes, some hidden bugs (particularly involving atomic operations on weaker ordering platforms) may occur. The tests are located in the test-suite directory.

make test

MAsyncRx is a multi-consumer (clonable) receiver designed for asynchronous contexts. It wraps AsyncRx and implements Clone and Sync, allowing it to be shared across multiple tasks safely.

Key characteristics:

• Clonable: You can call .clone() to create additional receivers for the same channel.
• Deref: It implements Deref<Target = AsyncRx<F>>, so all methods of AsyncRx are available directly on MAsyncRx instances.
• Conversion: Can be converted to a blocking MRx via .into_blocking().

The Tx<F> struct is a single-producer (sender) designed for use in a blocking context.

Key Constraints:

• Not Clone or Sync: You cannot clone a Tx or share it across threads via Arc<Tx>. Doing so will cause compilation errors because it lacks the Sync marker.
• Send: It can be moved to other threads. This is the correct way to use it in a multi-threaded producer-consumer pattern.
• Concurrency: If you require concurrent access (multiple producers), use MTx<F> instead.

Methods available via Deref to ChannelShared<F> allow you to inspect channel state.

use crossfire::*;
let (tx, rx) = spsc::bounded_blocking::<usize>(100);
std::thread::spawn(move || {
    let _ = tx.send(1);
});
drop(rx);

AsyncRx is a single-consumer receiver designed for asynchronous contexts. It is Send but not Clone or Sync.

If you need to share the receiver across multiple tasks or threads, use MAsyncRx instead. Because AsyncRx lacks a Sync marker, wrapping it in an Arc will cause it to lose its Send property, preventing it from being moved into other coroutines (like tokio::spawn).

Key characteristics:

• Cancellation-safe: The recv() method is safe to use with select! or timeout().
• Conversion: Can be converted to a blocking Rx via .into_blocking() or into an AsyncStream via .into_stream().

use crossfire::*; 
async fn foo() {
    let (tx, rx) = mpsc::bounded_async::<usize>(100);
    tokio::spawn(async move {
        let _ = rx.recv().await;
    });
    drop(tx);
}

Crossfire allows seamless bridging between blocking and async contexts. You can convert a sender or receiver from one context to another using the .into_blocking() method (provided by the From trait implementation). This is useful for mixing tokio::task::spawn_blocking with async tasks using the same channel.

// Example: Converting an async MPMC sender to a blocking MTx
let (tx, rx) = mpmc::bounded_async::<usize>(100);
let _tx: MTx<mpmc::Array<usize>> = tx.clone().into_blocking();

// Now _tx can be used in a blocking thread
_tx.send(1).expect("send ok");

The MTx<F> struct is a multi-producer (sender) designed for blocking contexts. Unlike Tx, MTx implements Clone and Sync, allowing it to be shared across multiple threads or cloned for multiple producers.

Capabilities:

• Clone: Create multiple handles to the same channel.
• Sync: Safe to share between threads.
• downgrade(): Returns a WeakTx<F> for non-owning references to the sender.
• into_async(): Converts the blocking multi-producer sender into an MAsyncTx<F>.

use crossfire::*;
let (tx, rx) = mpsc::bounded_blocking::<usize>(100);
let weak_tx = tx.downgrade();
assert_eq!(tx.get_tx_count(), 1);
let tx_clone = weak_tx.upgrade::<MTx<_>>().unwrap();
assert_eq!(tx.get_tx_count(), 2);

AsyncTx is a single-producer (sender) interface designed for asynchronous contexts. It is Send but not Clone or Sync.

If you need to share a sender across multiple coroutines or threads, use MAsyncTx instead. Because AsyncTx lacks a Sync marker, wrapping it in an Arc will cause it to lose its Send capability, preventing it from being moved into tasks like tokio::spawn.

Key characteristics:

• Single Producer: Cannot be cloned.
• Async-ready: Provides send, try_send, and timeout-based sending.
• Conversion: Can be converted to a blocking Tx via .into_blocking() or From<Tx<F>>.

use crossfire::*;
async fn foo() {
    let (tx, rx) = spsc::bounded_async::<usize>(100);
    tokio::spawn(async move {
         let _ = tx.send(2).await;
    });
    drop(rx);
}