v2.0.26 (legacy):

Debug locally:

Use --features trace_log to run the bench or test until it hangs, then press ctrl+c or send SIGINT, there will be latest log dump to /tmp/crossfire_ring.log (refer to tests/common.rs _setup_log())

Debug with github workflow: #37

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

The SP or SC interface is 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 a bounded channel, a 0 size case is not supported yet. (Temporary rewrite as 1 size).

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

The benefit of using the SP / SC API is completely lockless waker registration, in exchange for a performance boost.

The sender/receiver can use the From trait to convert between blocking and async context counterparts.

Error types are the same as crossbeam-channel: TrySendError, SendError, TryRecvError, RecvError

• tokio: Enable send_timeout, recv_timeout API for async context, based on tokio. And will detect the right backoff strategy for the type of runtime (multi-threaded / current-thread).

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

Tested on tokio-1.x and async-std-1.x, crossfire is runtime-agnostic.

The following scenarios are considered:

• The AsyncTx::send() and AsyncRx:recv() operations are cancellation-safe in an async context. You can safely use the select! macro and timeout() function in tokio/futures in combination with recv(). On cancellation, [SendFuture] and [RecvFuture] will trigger drop(), which will clean up the state of the waker, making sure there is no mem-leak and deadlock. But you cannot know the true result from SendFuture, since it's dropped upon cancellation. Thus, we suggest using AsyncTx::send_timeout() instead.

• When the "tokio" or "async_std" feature is enabled, we also provide two additional functions:

• AsyncTx::send_timeout(), which will return the message that failed to be sent in [SendTimeoutError]. We guarantee the result is atomic. Alternatively, you can use AsyncTx::send_with_timer().

• AsyncRx::recv_timeout(), we guarantee the result is atomic. Alternatively, you can use crate::AsyncRx::recv_with_timer().

• Between blocking context and async context, and between different async runtime instances.

• The async waker footprint.

When using a multi-producer and multi-consumer scenario, there's a small memory overhead to pass along a Weak reference of wakers. Because we aim to be lockless, when the sending/receiving futures are canceled (like tokio::time::timeout()), it might trigger an immediate cleanup if the try-lock is successful, otherwise will rely on lazy cleanup. (This won't be an issue because weak wakers will be consumed by actual message send and recv). On an idle-select scenario, like a notification for close, the waker will be reused as much as possible if poll() returns pending.

Cargo.toml:

[dependencies]
crossfire = "2.1"

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;
                }
            }
        }
    }
}