async_selector/lib.rs
1//! # async-selector
2//!
3//! [](https://crates.io/crates/async-selector)
4//! [](https://docs.rs/async-selector)
5//! [](./LICENSE)
6//! [](https://github.com/Razz4780/async-selector/actions/workflows/ci.yaml)
7//! [](https://crates.io/crates/async-selector)
8//!
9//! Fast and flexible selector for asynchronous tasks (generalized [`Future`]s and [`Stream`](futures::Stream)s).
10//!
11//! Inspired by [`FuturesUnordered`](futures::stream::FuturesUnordered), but more flexible and optimized.
12//! Provides optimal performance when polling a large number of tasks
13//! (see [example](https://github.com/Razz4780/async-selector/blob/main/examples/speed.rs)).
14//!
15//! Allows for:
16//! * Polling multiple tasks concurrently on the same thread
17//! * Safe injection of mutable shared state into the polling logic,
18//! meaning that caller code can provide a custom polling function with a strongly typed context
19//! * O(1) task access and removal with unique IDs
20//!
21//! The main struct is [`Selector`], which works with any [`Task`](crate::task::Task) implementor.
22//! For convenience, this crate exposes multiple specializations of the selector,
23//! including [`FutureSelector`] and [`StreamSelector`] (*almost* API-compatible with
24//! [`FuturesUnordered`](futures::stream::FuturesUnordered)/[`SelectAll`](futures::stream::SelectAll)).
25//!
26//! ## Examples
27//!
28//! Simply flatten a set of streams:
29//!
30//! ```rust
31//! # use async_selector::StreamSelector;
32//! # use futures::{StreamExt, channel::mpsc};
33//! #
34//! # #[tokio::main(flavor = "current_thread")]
35//! # async fn main() {
36//! let mut selector = StreamSelector::default();
37//! for i in 0..3 {
38//! let stream = futures::stream::repeat(i);
39//! selector.push(stream);
40//! }
41//! let collected = (&mut selector).take(6).collect::<Vec<_>>().await;
42//! assert_eq!(
43//! collected,
44//! vec![0, 1, 2, 0, 1, 2],
45//! );
46//! # }
47//! ```
48//!
49//! Use as a map of streams:
50//!
51//! ```rust
52//! # use async_selector::StreamWithIdSelector;
53//! # use futures::{SinkExt, StreamExt, channel::mpsc};
54//! #
55//! # #[tokio::main(flavor = "current_thread")]
56//! # async fn main() {
57//! let mut selector = StreamWithIdSelector::default();
58//! let ids = (0..3)
59//! .map(|i| {
60//! let stream = futures::stream::repeat(i);
61//! selector.push(stream).id().clone()
62//! })
63//! .collect::<Vec<_>>();
64//! let item = selector.next().await.unwrap();
65//! assert_eq!(item.0, ids[0]);
66//! assert_eq!(item.1, 0);
67//! selector.remove(&ids[1]);
68//! let item = selector.next().await.unwrap();
69//! assert_eq!(item.0, ids[2]);
70//! assert_eq!(item.1, 2);
71//! # }
72//! ```
73//!
74//! More examples live [here](https://github.com/Razz4780/async-selector/tree/main/examples).
75//!
76//! ## Performance
77//!
78//! The implementation of [`Selector`] is very similar to that of [`FuturesUnordered`](futures::stream::FuturesUnordered).
79//! However, some optimizations were made:
80//! 1. Reduced the number of CAS instructions
81//! 2. Removed repeated memory allocations when polling [`Stream`](futures::Stream)s
82//!
83//! [Here](https://github.com/Razz4780/async-selector/tree/main/examples/speed.rs) lives the source code
84//! of an example used to compare performance. It is not a proper benchmark,
85//! but strongly suggests that this implementation is at least as fast as
86//! [`FuturesUnordered`](futures::stream::FuturesUnordered).
87//!
88//! The results below were obtained with:
89//! * rustc 1.96.0
90//! * 13th Gen Intel(R) Core(TM) i9-13900HX
91//! * Tokio runtime with 32 worker threads
92//! * `cargo run --example speed --profile release`
93//!
94//! **Scenario 1**: concurrently drain 32 instances, each draining 1k streams, each stream producing 16k values (yielding once before producing each value)
95//!
96//! * [`StreamSelector`] - 2.522s
97//! * [`SelectAll`](futures::stream::SelectAll) - 5.979s
98//!
99//! **Scenario 2**: concurrently drain 32 instances, each resolving 1k futures, each future yielding 16k times
100//!
101//! * [`FutureSelector`] - 1.479s
102//! * [`FuturesUnordered`](futures::stream::FuturesUnordered) - 4.416s
103
104#![deny(unused_crate_dependencies)]
105
106use crate::{selector::Selector, task::strategy};
107
108mod list;
109mod queue;
110pub mod selector;
111pub mod task;
112
113/// [`Selector`] using [`strategy::FutureBasic`].
114pub type FutureSelector<F> = Selector<F, strategy::FutureBasic>;
115
116/// [`Selector`] using [`strategy::FutureReclaim`].
117pub type FutureReclaimSelector<F> = Selector<F, strategy::FutureReclaim>;
118
119/// [`Selector`] using [`strategy::StreamBasic`].
120pub type StreamSelector<S> = Selector<S, strategy::StreamBasic>;
121
122/// [`Selector`] using [`strategy::StreamWithId`].
123pub type StreamWithIdSelector<S> = Selector<S, strategy::StreamWithId>;
124
125/// [`Selector`] using [`strategy::StreamReclaim`].
126pub type StreamReclaimSelector<S> = Selector<S, strategy::StreamReclaim>;
127
128/// [`Selector`] using [`strategy::TryStreamBasic`].
129pub type TryStreamSelector<S> = Selector<S, strategy::TryStreamBasic>;
130
131/// [`Selector`] using [`strategy::TryStreamWithId`].
132pub type TryStreamWithIdSelector<S> = Selector<S, strategy::TryStreamWithId>;
133
134/// [`Selector`] using [`strategy::TryStreamReclaim`].
135pub type TryStreamReclaimSelector<S> = Selector<S, strategy::TryStreamReclaim>;