Crossfire

https://docs.rs/crossfire/latest/
Admin
High-performance lockless spsc/mpsc/mpmc channels with async and blocking context support.
Tokens:978,406
Snippets:9,247
Trust Score:9.1
Update:5 months ago
Show doc for...
Tokens:
Raw
### Example Searches

Source: https://docs.rs/crossfire/latest/crossfire/flavor/trait.Flavor.html?search=

Illustrative examples of search queries for the Flavor functionality, demonstrating different types and transformations.

```text
* std::vec
* u32 -> bool
* Option<T>, (T -> U) -> Option<U>
```

--------------------------------

### Example: Downgrading and Upgrading MAsyncTx

Source: https://docs.rs/crossfire/latest/crossfire/compat/type.MAsyncTx.html?search=

Demonstrates how to get a weak reference to an MAsyncTx sender using `downgrade` and then upgrade it back to a strong reference. This is useful for managing sender lifetimes without preventing receivers from being dropped.

```rust
use crossfire::*;
let (tx, rx) = mpsc::bounded_async::<usize>(100);
assert_eq!(tx.get_tx_count(), 1);
let weak_tx = tx.downgrade();
let tx_clone = weak_tx.upgrade::<MAsyncTx<_>>().unwrap();
assert_eq!(tx.get_tx_count(), 2);
drop(tx);
drop(tx_clone);
assert!(weak_tx.upgrade::<MAsyncTx<_>>().is_none());
assert_eq!(weak_tx.get_tx_count(), 0);
drop(rx);
```

--------------------------------

### mpmc::new Channel Initialization Examples

Source: https://docs.rs/crossfire/latest/crossfire/mpmc/fn.new.html?search=

Examples demonstrating how to initialize different types of MPMC channels using `mpmc::new`. The first example initializes a `List` channel for `i32`, while the second initializes an `async` channel for `mpmc::One<usize>`.

```rust
use crossfire::*;
let (tx, rx): (MTx<_>, MRx<_>) = mpmc::new::<mpmc::List<i32>, _, _>();
```

```rust
let (tx, rx): (MAsyncTx<mpmc::One<usize>>, MRx<mpmc::One<usize>>) = mpmc::new();
```

--------------------------------

### Example Searches

Source: https://docs.rs/crossfire/latest/crossfire/flavor/trait.FlavorBounded.html?search=

These examples illustrate common search patterns within the documentation, demonstrating how to search for types, type transformations, and generic function signatures.

```text
  * std::vec
  * u32 -> bool
  * Option<T>, (T -> U) -> Option<U>
```

--------------------------------

### build Function Examples

Source: https://docs.rs/crossfire/latest/crossfire/mpsc/fn.build.html?search=u32+-%3E+bool

Examples demonstrating how to use the build function to create different types of mpsc channels.

```APIDOC
## Examples

```rust
use crossfire::{*, mpsc::*};

// Example 1: Inferring types for List channel
let (tx, rx): (MTx<_>, Rx<_>) = build::<List<i32>, _, _>(List::new());

// Example 2: Inferring types for One channel with explicit async sender
let (tx, rx): (MAsyncTx<One<usize>>, Rx<One<usize>>)  = build(One::new());
```
```

--------------------------------

### Example Searches

Source: https://docs.rs/crossfire/latest/crossfire/flavor/trait.FlavorMC.html?search=

Illustrative examples of search queries that can be performed. These demonstrate searching for standard library types, type transformations, and generic type operations.

```text
std::vec
```

```text
u32 -> bool
```

```text
Option<T>, (T -> U) -> Option<U>
```

--------------------------------

### MPMC Example Usage

Source: https://docs.rs/crossfire/latest/crossfire/mpmc/index.html?search=

An example demonstrating how to set up and use MPMC channels with multiple consumers.

```APIDOC
```rust
use crossfire::*;
use std::thread;

struct Worker {
    tx: MAsyncTx<mpmc::Array<usize>>,
}

impl Worker {
    pub fn new() -> Self {
        // use type hint
        let (tx, rx): (MAsyncTx<_>, MRx<_>) = mpmc::build(mpmc::Array::<usize>::new(100));
        // equals to
        // let (tx, rx): (MAsyncTx<_>, MRx<_>) = mpmc::bounded_blocking::<usize>(100);
        for _ in 0..4 {
            let _rx = rx.clone();
            thread::spawn(move || {
                match _rx.recv() {
                    Ok(item)=>{
                        println!("recv job {}", item);
                    }
                    Err(_)=>return,
                }
            });
        }
        Self{
            tx,
        }
    }
    pub async fn submit(&self, msg: usize) {
        self.tx.send(msg).await.expect("send");
    }
}
```
```

--------------------------------

### MAsyncTx Downgrade Example

Source: https://docs.rs/crossfire/latest/src/crossfire/async_tx.rs.html?search=u32+-%3E+bool

Demonstrates obtaining a weak reference to an MAsyncTx and upgrading it. This example shows how to check sender counts and verify the lifecycle of the sender.

```rust
/// Get a weak reference of sender.
///
/// # Example
/// ```
/// use crossfire::*;
/// let (tx, rx) = mpsc::bounded_async::<usize>(100);
/// assert_eq!(tx.get_tx_count(), 1);
/// let weak_tx = tx.downgrade();
/// let tx_clone = weak_tx.upgrade::<MAsyncTx<_>>().unwrap();
/// assert_eq!(tx.get_tx_count(), 2);
/// drop(tx);
/// drop(tx_clone);
/// assert!(weak_tx.upgrade::<MAsyncTx<_>>().is_none());
/// assert_eq!(weak_tx.get_tx_count(), 0);
/// drop(rx);
/// ```
///
#[inline]
pub fn downgrade(&self) -> WeakTx<F>
where
    F: FlavorMP,
{
    WeakTx(self.shared.clone())
}
```

--------------------------------

### Example Usage of build Function

Source: https://docs.rs/crossfire/latest/crossfire/spsc/fn.build.html

Examples demonstrating how to use the `build` function to initialize different types of spsc channels. The compiler can infer the sender and receiver types.

```rust
use crossfire::{*, spsc::*};
let (tx, rx): (Tx<_>, Rx<_>) = build::<List<i32>, _, _>(List::new());
let (tx, rx): (AsyncTx<One<usize>>, Rx<One<usize>>)  = build(One::new());

```

--------------------------------

### Example Search: Option<T>, (T -> U) -> Option<U>

Source: https://docs.rs/crossfire/latest/crossfire/macro.trace_log.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

An example search query demonstrating a common Rust pattern involving Option types and function transformations.

```rust
Option<T>, (T -> U) -> Option<U>
```

--------------------------------

### Example: Downgrading and Upgrading MTx

Source: https://docs.rs/crossfire/latest/crossfire/compat/type.MTx.html?search=

Demonstrates obtaining a weak reference to an MTx sender, upgrading it back to a strong reference, and verifying sender counts. This example requires the 'compat' feature.

```rust
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);
drop(tx);
drop(tx_clone);
assert!(weak_tx.upgrade::<MTx<_>>().is_none());
assert_eq!(weak_tx.get_tx_count(), 0);
```

--------------------------------

### Multiplex Search Examples

Source: https://docs.rs/crossfire/latest/crossfire/select/struct.Multiplex.html?search=

Examples of how to perform searches within the Multiplex structure.

```APIDOC
## Multiplex Search Examples

This section provides example searches for the Multiplex structure.

### Example Searches

- `std::vec`
- `u32 -> bool`
- `Option<T>, (T -> U) -> Option<U>`
```

--------------------------------

### Start Read Operation for SegQueue

Source: https://docs.rs/crossfire/latest/src/crossfire/crossbeam/seg_queue.rs.html?search=

The `start_read` method provides a way to get a token for reading an element without immediately consuming it. It uses the internal `_pop` method with `false` for the `FINAL` flag.

```rust
    #[inline(always)]
    pub fn start_read(&self) -> Option<Token> {
        if let Some((block, offset)) = self._pop::<false>() {
            Some(Token::new(block as *const u8, offset))
        } else {
            None
        }
    }

```

--------------------------------

### Mixed Blocking/Async Sender-Receiver Example

Source: https://docs.rs/crossfire/latest/crossfire/index.html?search=

Demonstrates how to use blocking and asynchronous senders and receivers together with a bounded async channel. This example utilizes tokio for asynchronous tasks and spawn_blocking for synchronous tasks.

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

#[tokio::main] async fn main() {
    let (tx, rx) = mpmc::bounded_async::<usize>(100);
    let mut recv_counter = 0;
    let mut co_tx = Vec::new();
    let mut co_rx = Vec::new();
    const ROUND: usize = 1000;

    let _tx: MTx<mpmc::Array<usize>> = tx.clone().into_blocking();
    co_tx.push(tokio::task::spawn_blocking(move || {
        for i in 0..ROUND {
            _tx.send(i).expect("send ok");
        }
    }));
    co_tx.push(tokio::spawn(async move {
        for i in 0..ROUND {
            tx.send(i).await.expect("send ok");
        }
    }));
    let _rx: MRx<mpmc::Array<usize>> = rx.clone().into_blocking();
    co_rx.push(tokio::task::spawn_blocking(move || {
        let mut count: usize = 0;
        'A: loop {
            match _rx.recv() {
                Ok(_i) => {
                    count += 1;
                }
                Err(_) => break 'A,
            }
        }
        count
    }));
    co_rx.push(tokio::spawn(async move {
        let mut count: usize = 0;
        'A: loop {
            match rx.recv().await {
                Ok(_i) => {
                    count += 1;
                }
                Err(_) => break 'A,
            }
        }
        count
    }));
    for th in co_tx {
        let _ = th.await.unwrap();
    }
    for th in co_rx {
        recv_counter += th.await.unwrap();
    }
    assert_eq!(recv_counter, ROUND * 2);
}

```

--------------------------------

### Search Examples

Source: https://docs.rs/crossfire/latest/crossfire/oneshot/struct.RxOneshot.html?search=

Provides examples of search queries that can be used within the Crossfire Oneshot system.

```APIDOC
## Search Operations

### Description
This section provides examples of how to perform searches within the Crossfire Oneshot system. These examples illustrate the syntax and types of queries that can be executed.

### Example Searches

- `std::vec`
- `u32 -> bool`
- `Option<T>, (T -> U) -> Option<U>`
```

--------------------------------

### Example usage of bounded_blocking

Source: https://docs.rs/crossfire/latest/crossfire/mpmc/fn.bounded_blocking.html?search=

Demonstrates creating a bounded channel with a capacity of 10, sending an integer, and then receiving it. This example requires importing `crossfire::mpmc` and `crossfire::*`.

```rust
use crossfire::{mpmc, *};
let (tx, rx) = mpmc::bounded_blocking::<i32>(10);
tx.send(42).unwrap();
assert_eq!(rx.recv(), Ok(42));

```

--------------------------------

### Async Sender/Receiver Mixed Example

Source: https://docs.rs/crossfire/latest/crossfire/index.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Demonstrates mixing blocking and asynchronous senders and receivers within a Tokio runtime. This example showcases how to use `tokio::task::spawn_blocking` for blocking operations and `tokio::spawn` for async operations, all interacting with Crossfire channels.

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

#[tokio::main]
async fn main() {
    let (tx, rx) = mpmc::bounded_async::<usize>(100);
    let mut recv_counter = 0;
    let mut co_tx = Vec::new();
    let mut co_rx = Vec::new();
    const ROUND: usize = 1000;

    let _tx: MTx<mpmc::Array<usize>> = tx.clone().into_blocking();
    co_tx.push(tokio::task::spawn_blocking(move || {
        for i in 0..ROUND {
            _tx.send(i).expect("send ok");
        }
    }));
    co_tx.push(tokio::spawn(async move {
        for i in 0..ROUND {
            tx.send(i).await.expect("send ok");
        }
    }));
    let _rx: MRx<mpmc::Array<usize>> = rx.clone().into_blocking();
    co_rx.push(tokio::task::spawn_blocking(move || {
        let mut count: usize = 0;
        'A: loop {
            match _rx.recv() {
                Ok(_i) => {
                    count += 1;
                }
                Err(_) => break 'A,
            }
        }
        count
    }));
    co_rx.push(tokio::spawn(async move {
        let mut count: usize = 0;
        'A: loop {
            match rx.recv().await {
                Ok(_i) => {
                    count += 1;
                }
                Err(_) => break 'A,
            }
        }
        count
    }));
    for th in co_tx {
        let _ = th.await.unwrap();
    }
    for th in co_rx {
        recv_counter += th.await.unwrap();
    }
    assert_eq!(recv_counter, ROUND * 2);
}
```

--------------------------------

### Get Mutable Reference to Payload

Source: https://docs.rs/crossfire/latest/src/crossfire/waker.rs.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Provides mutable access to the payload stored within WakerInner. This allows modification of the payload, for example, when resetting the waker.

```rust
#[inline(always)]
pub fn get_payload_mut(&self) -> &mut P {
    unsafe { &mut *self.payload.get() }
}
```

--------------------------------

### Multiplex Channel Creation and Message Reception

Source: https://docs.rs/crossfire/latest/crossfire/select/struct.Multiplex.html?search=u32+-%3E+bool

Creates two bounded channels within a multiplexer, sends messages to them, and then receives them. This example demonstrates the basic setup and message flow for `Multiplex`.

```rust
use crossfire::{mpsc::Array, *, select::{Multiplex, Mux}};

let mut mp = Multiplex::<Array<i32>>::new();
// Creates a bounded channel with capacity 10
let tx1: MTx<Mux<Array<i32>>> = mp.bounded_tx(10);
// Creates another bounded channel with capacity 20
let tx2: MTx<Mux<Array<i32>>> = mp.bounded_tx(20);
tx1.send(42).expect("send");
std::thread::spawn(move || {
    tx2.send(42).expect("send");
});
let value = mp.recv().unwrap();
assert_eq!(value, 42);
let value = mp.recv().unwrap();
assert_eq!(value, 42);
```

--------------------------------

### Initializing mpsc Channels with new

Source: https://docs.rs/crossfire/latest/crossfire/mpsc/fn.new.html?search=

Examples demonstrating how to initialize mpsc channels using the `mpsc::new` function. The compiler can infer the types based on the return type signature.

```rust
use crossfire::*;
let (tx, rx): (MTx<_>, Rx<_>) = mpsc::new::<mpsc::List<i32>, _, _>();
```

```rust
let (tx, rx): (MAsyncTx<mpsc::One<usize>>, Rx<mpsc::One<usize>>) = mpsc::new();
```

--------------------------------

### Example Usage of Select Interface

Source: https://docs.rs/crossfire/latest/src/crossfire/select/select.rs.html?search=std%3A%3Avec

Demonstrates how to use the `Select` interface to concurrently receive messages from multiple channels. It shows adding receivers, performing a select operation, handling results, and removing disconnected channels.

```rust
use crossfire::{mpmc, mpsc, RecvError};
use crossfire::select::Select;

let (tx1, rx1) = mpmc::bounded_blocking::<i32>(10);
let (tx2, rx2) = mpsc::bounded_blocking::<i32>(10);

// Send some messages
tx1.send(100).unwrap();
tx2.send(200).unwrap();

// Drop senders to simulate disconnection after messages are sent
drop(tx1);
drop(tx2);

let mut select = Select::new();
select.add(&rx1);
select.add(&rx2);

// Loop until all channels are disconnected and removed from select
loop {
    // When `select()` returns `Err(RecvError)`, it means all channels
    // previously added to `select` have been disconnected or removed.
    // In such a case, there's nothing left to select from, so we break.
    let res = match select.select() {
        Ok(res) => res,
        Err(RecvError) => {
            println!("All channels disconnected or removed from select. Breaking loop.");
            break;
        },
    };

    // Handle the result from the ready receiver
    if res == rx1 {
        match rx1.read_select(res) {
            Ok(val) => println!("Received from rx1: {}", val),
            Err(RecvError) => { // Now RecvError
                println!("rx1 disconnected, removing from select.");
                select.remove(&rx1); // Remove disconnected receiver
            },
        }
    } else if res == rx2 {
        match rx2.read_select(res) {
            Ok(val) => println!("Received from rx2: {}", val),
            Err(RecvError) => { // Now RecvError
                println!("rx2 disconnected, removing from select.");
                select.remove(&rx2); // Remove disconnected receiver
            },
        }
    }
}

```

--------------------------------

### Get Weak Reference of Sender

Source: https://docs.rs/crossfire/latest/src/crossfire/async_tx.rs.html?search=

Obtain a weak reference to the sender, allowing for safe upgrades and checks for sender existence. This example demonstrates creating a weak reference, upgrading it, and verifying its behavior after the original senders are dropped.

```rust
/// Get a weak reference of sender.
///
/// # Example
/// ```
/// use crossfire::*; 
/// let (tx, rx) = mpsc::bounded_async::<usize>(100);
/// assert_eq!(tx.get_tx_count(), 1);
/// let weak_tx = tx.downgrade();
/// let tx_clone = weak_tx.upgrade::<MAsyncTx<_>>().unwrap();
/// assert_eq!(tx.get_tx_count(), 2);
/// drop(tx);
/// drop(tx_clone);
/// assert!(weak_tx.upgrade::<MAsyncTx<_>>().is_none());
/// assert_eq!(weak_tx.get_tx_count(), 0);
/// drop(rx);
/// ```
#[inline]
    pub fn downgrade(&self) -> WeakTx<F>
    where
        F: FlavorMP,
    {
        WeakTx(self.shared.clone())
    }
```

--------------------------------

### Rx Usage Example

Source: https://docs.rs/crossfire/latest/src/crossfire/blocking_rx.rs.html?search=std%3A%3Avec

Demonstrates moving an Rx to another thread. This is allowed because Rx implements Send.

```rust
use crossfire::*
let (tx, rx) = mpsc::bounded_blocking::<usize>(100);
std::thread::spawn(move || {
    let _ = rx.recv();
});
drop(tx);

```

--------------------------------

### Example Search Query

Source: https://docs.rs/crossfire/latest/crossfire/flavor/trait.FlavorMC.html?search=u32+-%3E+bool

Illustrates a search query format for finding type conversions, specifically from u32 to bool.

```text
u32 -> bool
```

--------------------------------

### AsyncTx into_sink Example

Source: https://docs.rs/crossfire/latest/src/crossfire/async_tx.rs.html

Converts an AsyncTx into an AsyncSink. This is useful for integrating with systems that expect an AsyncSink.

```rust
pub fn into_sink(self) -> AsyncSink<F> {
        AsyncSink::new(self)
    }
```

--------------------------------

### Receive with Smol Timer Example

Source: https://docs.rs/crossfire/latest/crossfire/struct.AsyncRx.html?search=std%3A%3Avec

Example demonstrating the use of `recv_with_timer` with a `smol::Timer` for custom timeout behavior.

```rust
extern crate smol
use std::time::Duration;
use crossfire::*
async fn foo() {
    let (tx, rx) = mpmc::bounded_async::<usize>(10);
    match rx.recv_with_timer(smol::Timer::after(Duration::from_secs(1))).await {
        Ok(_item)=>{
            println!("message recv");
        }
        Err(RecvTimeoutError::Timeout)=>{
            println!("timeout");
        }
        Err(RecvTimeoutError::Disconnected)=>{
            println!("sender-side closed");
        }
    }
}
```

--------------------------------

### Initializing spsc Channels with new

Source: https://docs.rs/crossfire/latest/crossfire/spsc/fn.new.html?search=

Examples demonstrating how to initialize sender and receiver types using `spsc::new`. The compiler can infer types based on the return signature.

```rust
use crossfire::*;
let (tx, rx): (Tx<_>, Rx<_>) = spsc::new::<spsc::List<i32>, _, _>();
```

```rust
let (tx, rx): (AsyncTx<spsc::One<usize>>, Rx<spsc::One<usize>>) = spsc::new();
```

--------------------------------

### Examples of using new

Source: https://docs.rs/crossfire/latest/crossfire/mpmc/fn.new.html?search=std%3A%3Avec

Illustrates how to use the `new` function to create different types of MPMC channels, including type inference.

```APIDOC
## §Examples
```rust
use crossfire::*
let (tx, rx): (MTx<_>, MRx<_>) = mpmc::new::<mpmc::List<i32>, _, _>();
let (tx, rx): (MAsyncTx<mpmc::One<usize>>, MRx<mpmc::One<usize>>) = mpmc::new();
```
```

--------------------------------

### Send Message Example

Source: https://docs.rs/crossfire/latest/crossfire/struct.MAsyncTx.html

Shows how to send a message asynchronously using the `send` method. The operation will await if the channel is full.

```rust
send( &self, item: F::Item, ) -> impl Future<Output = Result<(), SendError<F::Item>>> + Send
where F::Item: Send + 'static + Unpin,
Send message. Will await when channel is full. Read more
```

--------------------------------

### Receive Message with Smol Timer Example

Source: https://docs.rs/crossfire/latest/crossfire/compat/type.AsyncRx.html

An example demonstrating how to use `recv_with_timer` with a `smol::Timer` for a custom timeout duration.

```rust
extern crate smol;
use std::time::Duration;
use crossfire::*;
async fn foo() {
    let (tx, rx) = mpmc::bounded_async::<usize>(10);
    match rx.recv_with_timer(smol::Timer::after(Duration::from_secs(1))).await {
        Ok(_item)=>{
            println!("message recv");
        }
        Err(RecvTimeoutError::Timeout)=>{
            println!("timeout");
        }
        Err(RecvTimeoutError::Disconnected)=>{
            println!("sender-side closed");
        }
    }
}
```

--------------------------------

### take

Source: https://docs.rs/crossfire/latest/crossfire/stream/struct.AsyncStream.html

Creates a stream that yields its first `n` elements.

```APIDOC
## take(self, n: usize) -> Take<Self>

### Description
Creates a stream that yields its first `n` elements.

### Parameters
#### Path Parameters
- **n** (usize) - Required - The number of elements to yield.
```

--------------------------------

### take

Source: https://docs.rs/crossfire/latest/crossfire/stream/struct.AsyncStream.html?search=u32+-%3E+bool

Creates a stream that yields its first `n` elements.

```APIDOC
fn take(self, n: usize) -> Take<Self>
where Self: Sized
```

--------------------------------

### take

Source: https://docs.rs/crossfire/latest/crossfire/stream/struct.AsyncStream.html?search=

Creates a stream that yields its first `n` elements.

```APIDOC
## fn take(self, n: usize) -> Take<Self>
where Self: Sized,

### Description
Creates a stream that yields its first `n` elements.

### Parameters
#### Path Parameters
- None

#### Query Parameters
- None

#### Request Body
- None

### Request Example
```rust
// Example usage not provided in source
```

### Response
#### Success Response (200)
- Returns a `Take` stream that yields at most `n` items from the original stream.

#### Response Example
```json
// Example response not provided in source
```
```

--------------------------------

### Rx Send example

Source: https://docs.rs/crossfire/latest/crossfire/struct.Rx.html

Demonstrates that an Rx instance can be moved to another thread using std::thread::spawn, as Rx implements Send.

```rust
use crossfire::*
use std::thread;

let (tx, rx) = mpsc::bounded_blocking::<usize>(100);
thread::spawn(move || {
    let _ = rx.recv();
});
drop(tx);
```

--------------------------------

### start_read

Source: https://docs.rs/crossfire/latest/src/crossfire/crossbeam/array_queue_spsc.rs.html?search=u32+-%3E+bool

Initiates a read operation from the queue. It returns an `Option<Token>` which can be used to retrieve the data.

```APIDOC
## start_read

### Description
Initiates a read operation from the queue. It returns an `Option<Token>` which can be used to retrieve the data.

### Method
`pub fn start_read(&self, final_check: bool) -> Option<Token>`

### Parameters
- `final_check` (bool) - If true, performs a final check to ensure data consistency.

### Returns
- `Option<Token>` - A token representing the read operation if successful, otherwise `None`.
```

--------------------------------

### SPSC ArrayQueue Start Read Token

Source: https://docs.rs/crossfire/latest/src/crossfire/crossbeam/array_queue_spsc.rs.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Starts a read operation and returns a `Token` which can be used later to complete the read. This is useful for deferring the finalization of the read operation.

```rust
#[inline]
    pub fn start_read(&self, final_check: bool) -> Option<Token> {
        if let Some((head, tail_cached)) = self._start_read::<true>(final_check) {
            let (slot, packed_recv) = self._read(head, tail_cached);
            Some(Token::new(slot as *const Slot<T> as *const u8, packed_recv as usize))
        } else {
            None
        }
    }
```

--------------------------------

### Example Usage of Select

Source: https://docs.rs/crossfire/latest/crossfire/select/struct.Select.html

Demonstrates how to add receivers to a Select instance, loop to receive messages, and handle disconnected channels. It's crucial to process `SelectResult` to avoid message leaks and to remove disconnected receivers from the select instance.

```rust
use crossfire::{mpmc, mpsc, RecvError};
use crossfire::select::Select;

let (tx1, rx1) = mpmc::bounded_blocking::<i32>(10);
let (tx2, rx2) = mpsc::bounded_blocking::<i32>(10);

// Send some messages
tx1.send(100).unwrap();
tx2.send(200).unwrap();

// Drop senders to simulate disconnection after messages are sent
drop(tx1);
drop(tx2);

let mut select = Select::new();
select.add(&rx1);
select.add(&rx2);

// Loop until all channels are disconnected and removed from select
loop {
    // When `select()` returns `Err(RecvError)`, it means all channels
    // previously added to `select` have been disconnected or removed.
    // In such a case, there's nothing left to select from, so we break.
    let res = match select.select() {
        Ok(res) => res,
        Err(RecvError) => {
            println!("All channels disconnected or removed from select. Breaking loop.");
            break;
        },
    };

    // Handle the result from the ready receiver
    if res == rx1 {
        match rx1.read_select(res) {
            Ok(val) => println!("Received from rx1: {}", val),
            Err(RecvError) => { // Now RecvError
                println!("rx1 disconnected, removing from select.");
                select.remove(&rx1); // Remove disconnected receiver
            },
        }
    } else if res == rx2 {
        match rx2.read_select(res) {
            Ok(val) => println!("Received from rx2: {}", val),
            Err(RecvError) => { // Now RecvError
                println!("rx2 disconnected, removing from select.");
                select.remove(&rx2); // Remove disconnected receiver
            },
        }
    }
}
```

--------------------------------

### Any for T

Source: https://docs.rs/crossfire/latest/crossfire/struct.AsyncRx.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Gets the TypeId of self.

```APIDOC
## fn type_id(&self) -> TypeId

Gets the `TypeId` of `self`. Read more
Source§
```

--------------------------------

### WeakTx Upgrade Example

Source: https://docs.rs/crossfire/latest/src/crossfire/weak.rs.html?search=u32+-%3E+bool

Demonstrates how to downgrade a sender to a weak reference, then upgrade it back, and check sender counts. It also shows that upgrading fails after all strong references are dropped.

```rust
use crossfire::*
let (tx, rx) = mpsc::bounded_blocking::<usize>(100);
let weak_tx = tx.downgrade();
let tx_clone = weak_tx.upgrade::<MTx<_>>().unwrap();
drop(tx);
drop(tx_clone);
assert!(weak_tx.upgrade::<MTx<_>>().is_none());
assert_eq!(weak_tx.get_tx_count(), 0);
drop(rx);
```

--------------------------------

### Async and Blocking Sender/Receiver Mixed Example

Source: https://docs.rs/crossfire/latest/src/crossfire/lib.rs.html

Demonstrates how to mix asynchronous and blocking senders and receivers with an MPMC bounded channel. This example utilizes tokio for asynchronous tasks and spawn_blocking for synchronous tasks.

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

#[tokio::main]
async fn main() {
    let (tx, rx) = mpmc::bounded_async::<usize>(100);
    let mut recv_counter = 0;
    let mut co_tx = Vec::new();
    let mut co_rx = Vec::new();
    const ROUND: usize = 1000;

    let _tx: MTx<mpmc::Array<usize>> = tx.clone().into_blocking();
    co_tx.push(tokio::task::spawn_blocking(move || {
        for i in 0..ROUND {
            _tx.send(i).expect("send ok");
        }
    }));
    co_tx.push(tokio::spawn(async move {
        for i in 0..ROUND {
            tx.send(i).await.expect("send ok");
        }
    }));
    let _rx: MRx<mpmc::Array<usize>> = rx.clone().into_blocking();
    co_rx.push(tokio::task::spawn_blocking(move || {
        let mut count: usize = 0;
        'A: loop {
            match _rx.recv() {
                Ok(_i) => {
                    count += 1;
                }
                Err(_) => break 'A,
            }
        }
        count
    }));
    co_rx.push(tokio::spawn(async move {
        let mut count: usize = 0;
        'A: loop {
            match rx.recv().await {
                Ok(_i) => {
                    count += 1;
                }
                Err(_) => break 'A,
            }
        }
        count
    }));
    for th in co_tx {
        let _ = th.await.unwrap();
    }
    for th in co_rx {
        recv_counter += th.await.unwrap();
    }
    assert_eq!(recv_counter, ROUND * 2);
}
```

--------------------------------

### Async Example with Dynamic Guard Cloning

Source: https://docs.rs/crossfire/latest/src/crossfire/waitgroup.rs.html?search=u32+-%3E+bool

Demonstrates task and sub-task management in an async context, where `WaitGroupGuard` can be cloned to dynamically increase the reference count.

```rust
use crossfire::waitgroup::WaitGroup
use std::time::Duration

#[tokio::test]
async fn wait_group_async_example() {
    let wg = WaitGroup::new((), 0)
    for _j in 0..4 {
        // Create a guard for the manager task.
        let parent_guard = wg.add_guard()
        tokio::spawn(async move {
            // This manager task will spawn 2 workers.
            for i in 0..2 {
                let child_guard = parent_guard.clone()

```

--------------------------------

### Select::new

Source: https://docs.rs/crossfire/latest/crossfire/select/struct.Select.html

Initializes a new Select instance with a fair, round-robin selection strategy.

```APIDOC
## Select::new

### Description
Initializes Select with fair, round-robin strategy.

### Method
```rust
pub fn new() -> Self
```
```

--------------------------------

### type_id

Source: https://docs.rs/crossfire/latest/crossfire/select/struct.Select.html?search=std%3A%3Avec

Gets the `TypeId` of `self`.

```APIDOC
## fn type_id(&self) -> TypeId

### Description
Gets the `TypeId` of `self`.

### Method
`type_id`

### Parameters
* `self`: &Self - An immutable reference to self.
```

--------------------------------

### Async Task and Sub-task Example

Source: https://docs.rs/crossfire/latest/crossfire/waitgroup/index.html

Demonstrates managing tasks and sub-tasks in an asynchronous context using `WaitGroup`. It shows how to dynamically increase the reference count by cloning `WaitGroupGuard` for worker tasks spawned by a manager task.

```rust
use crossfire::waitgroup::WaitGroup;
use std::time::Duration;

#[tokio::test] async fn wait_group_async_example() {
    let wg = WaitGroup::new((), 0);
    for _j in 0..4 {
        // Create a guard for the manager task.
        let parent_guard = wg.add_guard();
        tokio::spawn(async move {
            // This manager task will spawn 2 workers.
            for i in 0..2 {
                let child_guard = parent_guard.clone();
                tokio::spawn(async move {
                    // Do some work...
                    tokio::time::sleep(Duration::from_millis(50 * (i + 1))).await;
                    // worker_guard is dropped here.
                });
            }
            // The manager's work is to spawn workers,
            // so it drops its own guard after the loop.
            drop(manager_guard);
        });
    }
    // Wait until the manager guard and all its clones are dropped.
    wg.wait_async().await;
    assert_eq!(wg.get_left_seqcst(), 0);
}
```

--------------------------------

### Any::type_id

Source: https://docs.rs/crossfire/latest/crossfire/flavor/struct.One.html

Gets the `TypeId` of `self`.

```APIDOC
## fn type_id(&self) -> TypeId

### Description
Gets the `TypeId` of `self`.

### Method
`type_id`

### Returns
The `TypeId` of `self`.
```

--------------------------------

### type_id

Source: https://docs.rs/crossfire/latest/crossfire/struct.Tx.html?search=u32+-%3E+bool

Gets the TypeId of the Tx value.

```APIDOC
## type_id

### Description
Gets the `TypeId` of `self`.

### Method
`type_id(&self) -> TypeId`
```

--------------------------------

### start_read

Source: https://docs.rs/crossfire/latest/src/crossfire/crossbeam/array_queue_mpsc.rs.html?search=std%3A%3Avec

Initiates a read operation from the queue. It prepares the necessary internal state for reading and returns a `Token` if a read is possible, otherwise `None`.

```APIDOC
## start_read

### Description
Initiates a read operation from the queue. It prepares the necessary internal state for reading and returns a `Token` if a read is possible, otherwise `None`.

### Method
`pub fn start_read(&self, final_check: bool) -> Option<Token>`

### Parameters
- `final_check` (bool) - A flag to indicate if a final check should be performed.

### Return Value
- `Option<Token>` - A `Token` representing the read operation if successful, `None` otherwise.
```

--------------------------------

### Initialize mpsc Channel with List Flavor

Source: https://docs.rs/crossfire/latest/crossfire/mpsc/fn.new.html

Example of initializing a multi-producer, single-consumer channel using the `List` flavor for `i32` elements. The sender and receiver types are explicitly specified.

```rust
use crossfire::*;
let (tx, rx): (MTx<_>, Rx<_>) = mpsc::new::<mpsc::List<i32>, _, _>();

```

--------------------------------

### Deref::capacity

Source: https://docs.rs/crossfire/latest/crossfire/struct.Tx.html?search=std%3A%3Avec

Gets the capacity of the channel.

```APIDOC
## Deref::capacity

### Description
The capacity of the channel. Returns `None` for unbounded channels.

### Method
`capacity`

### Parameters
None

### Returns
`Option<usize>` - The capacity of the channel, or `None` if it's unbounded.
```

--------------------------------

### new (SenderType)

Source: https://docs.rs/crossfire/latest/crossfire/compat/type.AsyncTx.html?search=

Creates a new AsyncTx instance.

```APIDOC
## new (SenderType)

### Description
Creates a new AsyncTx instance.

### Method
`new(shared: Arc<ChannelShared<F>>)`

### Parameters
* `shared` (Arc<ChannelShared<F>>) - The shared channel state.

### Return Value
`Self`

### Associated Type
`Flavor = F`
```

--------------------------------

### fn type_id()

Source: https://docs.rs/crossfire/latest/crossfire/struct.MRx.html

Gets the `TypeId` of `self`.

```APIDOC
## fn type_id()

### Description
Gets the `TypeId` of `self`.

### Method
`type_id`

### Parameters
None

### Response
- `TypeId`: The `TypeId` of the object.
```

--------------------------------

### Async Sender/Receiver Mixed Example

Source: https://docs.rs/crossfire/latest/crossfire/index.html?search=u32+-%3E+bool

Demonstrates mixing blocking and asynchronous senders and receivers within a Tokio runtime. This example showcases how to use `mpmc::bounded_async` with both blocking and async tasks, ensuring proper handling of message counts.

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

#[tokio::main]
async fn main() {
    let (tx, rx) = mpmc::bounded_async::<usize>(100);
    let mut recv_counter = 0;
    let mut co_tx = Vec::new();
    let mut co_rx = Vec::new();
    const ROUND: usize = 1000;

    let _tx: MTx<mpmc::Array<usize>> = tx.clone().into_blocking();
    co_tx.push(tokio::task::spawn_blocking(move || {
        for i in 0..ROUND {
            _tx.send(i).expect("send ok");
        }
    }));
    co_tx.push(tokio::spawn(async move {
        for i in 0..ROUND {
            tx.send(i).await.expect("send ok");
        }
    }));
    let _rx: MRx<mpmc::Array<usize>> = rx.clone().into_blocking();
    co_rx.push(tokio::task::spawn_blocking(move || {
        let mut count: usize = 0;
        'A: loop {
            match _rx.recv() {
                Ok(_i) => {
                    count += 1;
                }
                Err(_) => break 'A,
            }
        }
        count
    }));
    co_rx.push(tokio::spawn(async move {
        let mut count: usize = 0;
        'A: loop {
            match rx.recv().await {
                Ok(_i) => {
                    count += 1;
                }
                Err(_) => break 'A,
            }
        }
        count
    }));
    for th in co_tx {
        let _ = th.await.unwrap();
    }
    for th in co_rx {
        recv_counter += th.await.unwrap();
    }
    assert_eq!(recv_counter, ROUND * 2);
}
```

--------------------------------

### Create new RegistrySingle for sending

Source: https://docs.rs/crossfire/latest/src/crossfire/waker_registry.rs.html

Initializes a new RegistrySingle instance for use as a send-side registry with a WeakCell storage.

```rust
#[inline(always)]
fn new() -> Self {
    //Self { cell: _OneSpmc::new(), _tag: "tx" }
    Self { cell: WeakCell::new(), _tag: "tx" }
}
```

--------------------------------

### Building mpmc Channels with build Function

Source: https://docs.rs/crossfire/latest/crossfire/mpmc/fn.build.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Examples demonstrating how to use the build function to initialize different types of mpmc channels. The compiler can infer types based on the return signature.

```rust
use crossfire::{*, mpmc::*};
let (tx, rx): (MTx<_>, MRx<_>) = build::<List<i32>, _, _>(List::new());
let (tx, rx): (MAsyncTx<One<usize>>, MRx<One<usize>>)  = build(One::new());
```

--------------------------------

### Internal Start Read Logic

Source: https://docs.rs/crossfire/latest/src/crossfire/crossbeam/array_queue.rs.html

The core logic for starting a read operation. It manages head pointer loading, slot inspection, and atomic head updates to ensure safe concurrent access. Uses backoff strategies for contention.

```rust
#[inline]
fn _start_read(&self, final_check: bool) -> Option<(&Slot<T>, usize)> {
    let mut head;
    if final_check {
        // because we need to check is_empty before park,
        // use SeqCst to make Miri happy
        head = self.head.load(Ordering::SeqCst);
    } else {
        let order = if MC { Ordering::Relaxed } else { Ordering::Acquire };
        head = self.head.load(order);
    }
    let backoff = Backoff::new();
    loop {
        // Deconstruct the head.
        let index = head & (self.one_lap - 1);
        // Inspect the corresponding slot.
        debug_assert!(index < self.buffer.len());
        let slot = unsafe { self.buffer.get_unchecked(index) };
        let stamp = slot.stamp.load(Ordering::Acquire);

        // If the stamp is ahead of the head by 1, we may attempt to pop.
        if head + 1 == stamp {
            let new = if index + 1 < self.capacity() {
                // Same lap, incremented index.
                // Set to `{ lap: lap, index: index + 1 }`.
                head + 1
            } else {
                let lap = head & !(self.one_lap - 1);
                // One lap forward, index wraps around to zero.
                // Set to `{ lap: lap.wrapping_add(1), index: 0 }`.
                lap.wrapping_add(self.one_lap)
            };
            if MC {
                // Try moving the head.
                if let Err(new_head) = self.head.compare_exchange_weak(
                    head,
                    new,
                    Ordering::SeqCst,
                    Ordering::Relaxed,
                ) {
                    head = new_head;
                    backoff.spin();
                    continue;
                }
            } else {
                self.head.store(new, Ordering::SeqCst);
            }
            let new_head = head.wrapping_add(self.one_lap);
            return Some((slot, new_head));
        } else {
            if stamp == head {
                // Check full
                let tail = if MP || MC {
                    // NOTE: The fence is preventing live lock
                    atomic::fence(Ordering::SeqCst);
                    self.tail.load(Ordering::Relaxed)
                } else {
                    self.tail.load(Ordering::SeqCst)
                };
                // If the tail equals the head, that means the channel is empty.
                if tail == head {
                    return None;
                }
                backoff.spin();
            } else {
                // Snooze because we need to wait for the stamp to get updated.
                backoff.snooze();
            }
            if MC {
                head = self.head.load(Ordering::Relaxed);
            }
            continue;
        }
    }
}
```

--------------------------------

### MTx::downgrade

Source: https://docs.rs/crossfire/latest/crossfire/struct.MTx.html

Gets a weak reference of the sender.

```APIDOC
## MTx::downgrade

### Description
Get a weak reference of sender.

### Method
`downgrade`

### Parameters
None

### Returns
WeakTx<F>

### Example
```rust
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);
drop(tx);
drop(tx_clone);
assert!(weak_tx.upgrade::<MTx<_>>().is_none());
assert_eq!(weak_tx.get_tx_count(), 0);
```
```

--------------------------------

### MAsyncTx::downgrade

Source: https://docs.rs/crossfire/latest/crossfire/struct.MAsyncTx.html?search=

Gets a weak reference to the sender.

```APIDOC
## MAsyncTx::downgrade

### Description
Gets a weak reference of sender.

### Method
`downgrade()`

### Example
```rust
use crossfire::*;
let (tx, rx) = mpsc::bounded_async::<usize>(100);
assert_eq!(tx.get_tx_count(), 1);
let weak_tx = tx.downgrade();
let tx_clone = weak_tx.upgrade::<MAsyncTx<_>>().unwrap();
assert_eq!(tx.get_tx_count(), 2);
drop(tx);
drop(tx_clone);
assert!(weak_tx.upgrade::<MAsyncTx<_>>().is_none());
assert_eq!(weak_tx.get_tx_count(), 0);
drop(rx);
```
```

--------------------------------

### Try Select with Token

Source: https://docs.rs/crossfire/latest/src/crossfire/flavor/one.rs.html?search=std%3A%3Avec

Attempts to select an item for reading. It uses `_start_read` with specified ordering and returns a `Token` containing the position and stamp if successful.

```rust
if let Some((index, head)) = 
    self._start_read(if final_check { Ordering::SeqCst } else { Ordering::Acquire }) 
{
    Some(Token::new(
        &self.slots[index as usize] as *const Slot<T> as *const u8,
        head as usize,
    ))
} else {
    None
}
```

--------------------------------

### Try Select Implementation

Source: https://docs.rs/crossfire/latest/src/crossfire/flavor/one.rs.html

Attempts to select an item for reading. It uses `_start_read` with specified ordering and returns a `Token` if successful.

```rust
    #[inline]
    fn try_select(&self, final_check: bool) -> Option<Token> {
        if let Some((index, head)) =
            self._start_read(if final_check { Ordering::SeqCst } else { Ordering::Acquire })
        {
            Some(Token::new(
                &self.slots[index as usize] as *const Slot<T> as *const u8,
                head as usize,
            ))
        } else {
            None
        }
    }
```

--------------------------------

### nth

Source: https://docs.rs/crossfire/latest/crossfire/stream/struct.AsyncStream.html?search=

Gets the `n`th item of the stream.

```APIDOC
## nth(&mut self, n: usize) -> NthFuture<'_, Self>

### Description
Gets the `n`th item of the stream.

### Constraints
- `Self` must implement `Unpin`.

### Parameters
- `n` (usize): The index of the item to retrieve.
```

--------------------------------

### nth

Source: https://docs.rs/crossfire/latest/crossfire/stream/struct.AsyncStream.html?search=u32+-%3E+bool

Gets the `n`th item of the stream.

```APIDOC
## nth(&mut self, n: usize) -> NthFuture<'_, Self>

### Description
Gets the `n`th item of the stream.

### Parameters
- **n**: usize - The index of the item to retrieve.
```

--------------------------------

### Run Benchmarks

Source: https://docs.rs/crossfire/latest/crossfire/index.html?search=

Commands to run the benchmark tests for the crossfire crate and its select functionality.

```bash
make bench crossfire
make bench crossfire_select
```

--------------------------------

### nth

Source: https://docs.rs/crossfire/latest/crossfire/stream/struct.AsyncStream.html

Gets the `n`th item of the stream.

```APIDOC
## nth(&mut self, n: usize) -> NthFuture<'_, Self>

### Description
Gets the `n`th item of the stream.

### Parameters
- **n** (usize) - The index of the item to retrieve (0-based).

### Constraints
- `Self` must implement `Unpin`.

### Returns
A future that resolves to `Some(item)` if the `n`th item exists, or `None` otherwise.
```

--------------------------------

### new

Source: https://docs.rs/crossfire/latest/crossfire/struct.Tx.html?search=u32+-%3E+bool

Creates a new Tx instance with the given shared channel data.

```APIDOC
## new

### Description
Creates a new Tx instance.

### Method
`new(shared: Arc<ChannelShared<F>>) -> Self`
```

--------------------------------

### nth

Source: https://docs.rs/crossfire/latest/crossfire/stream/struct.AsyncStream.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Gets the `n`th item of the stream.

```APIDOC
## nth

### Description
Gets the `n`th item of the stream.

### Signature
fn nth(&mut self, n: usize) -> NthFuture<'_, Self>

### Parameters
* `n`: The index of the item to retrieve.
```

--------------------------------

### OneSpsc start_read Method

Source: https://docs.rs/crossfire/latest/src/crossfire/flavor/one_spmc.rs.html?search=

Initiates a read operation by checking if there is data available. It returns the tail position if data is present, otherwise None.

```rust
#[inline(always)]
fn start_read(&self, order: Ordering) -> Option<u32> {
    let pos = self.pos.load(order);
    let (head, tail) = Self::unpack(pos);
    if head == tail {
        None
    } else {
        debug_assert_eq!(head.wrapping_add(1), tail);
        Some(tail)
    }
}

```

--------------------------------

### Deref::len

Source: https://docs.rs/crossfire/latest/crossfire/struct.Tx.html?search=std%3A%3Avec

Gets the number of messages currently in the channel.

```APIDOC
## Deref::len

### Description
The number of messages in the channel.

### Method
`len`

### Parameters
None

### Returns
`usize` - The number of messages in the channel.
```

--------------------------------

### Select::new_with

Source: https://docs.rs/crossfire/latest/crossfire/select/struct.Select.html

Initializes a new Select instance with a specified selection mode.

```APIDOC
## Select::new_with

### Description
Initialize Select with a specified selection mode.

### Method
```rust
pub fn new_with(mode: SelectMode) -> Self
```
```

--------------------------------

### fn type_id(&self) -> TypeId

Source: https://docs.rs/crossfire/latest/crossfire/struct.MRx.html?search=

Gets the `TypeId` of `self`.

```APIDOC
## fn type_id(&self) -> TypeId

### Description
Gets the `TypeId` of `self`.

### Method
`type_id`

### Parameters
None

### Response
- `TypeId`: The `TypeId` of the instance.
```
Tokens:
Raw