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