Coverage Report

Created: 2025-07-18 06:42

/rust/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.46.1/src/macros/join.rs
Line
Count
Source (jump to first uncovered line)
1
macro_rules! doc {
2
    ($join:item) => {
3
        /// Waits on multiple concurrent branches, returning when **all** branches
4
        /// complete.
5
        ///
6
        /// The `join!` macro must be used inside of async functions, closures, and
7
        /// blocks.
8
        ///
9
        /// The `join!` macro takes a list of async expressions and evaluates them
10
        /// concurrently on the same task. Each async expression evaluates to a future
11
        /// and the futures from each expression are multiplexed on the current task.
12
        ///
13
        /// When working with async expressions returning `Result`, `join!` will wait
14
        /// for **all** branches complete regardless if any complete with `Err`. Use
15
        /// [`try_join!`] to return early when `Err` is encountered.
16
        ///
17
        /// [`try_join!`]: crate::try_join
18
        ///
19
        /// # Notes
20
        ///
21
        /// The supplied futures are stored inline and do not require allocating a
22
        /// `Vec`.
23
        ///
24
        /// ## Runtime characteristics
25
        ///
26
        /// By running all async expressions on the current task, the expressions are
27
        /// able to run **concurrently** but not in **parallel**. This means all
28
        /// expressions are run on the same thread and if one branch blocks the thread,
29
        /// all other expressions will be unable to continue. If parallelism is
30
        /// required, spawn each async expression using [`tokio::spawn`] and pass the
31
        /// join handle to `join!`.
32
        ///
33
        /// [`tokio::spawn`]: crate::spawn
34
        ///
35
        /// ## Fairness
36
        ///
37
        /// By default, `join!`'s generated future rotates which contained
38
        /// future is polled first whenever it is woken.
39
        ///
40
        /// This behavior can be overridden by adding `biased;` to the beginning of the
41
        /// macro usage. See the examples for details. This will cause `join` to poll
42
        /// the futures in the order they appear from top to bottom.
43
        ///
44
        /// You may want this if your futures may interact in a way where known polling order is significant.
45
        ///
46
        /// But there is an important caveat to this mode. It becomes your responsibility
47
        /// to ensure that the polling order of your futures is fair. If for example you
48
        /// are joining a stream and a shutdown future, and the stream has a
49
        /// huge volume of messages that takes a long time to finish processing per poll, you should
50
        /// place the shutdown future earlier in the `join!` list to ensure that it is
51
        /// always polled, and will not be delayed due to the stream future taking a long time to return
52
        /// `Poll::Pending`.
53
        ///
54
        /// # Examples
55
        ///
56
        /// Basic join with two branches
57
        ///
58
        /// ```
59
        /// async fn do_stuff_async() {
60
        ///     // async work
61
        /// }
62
        ///
63
        /// async fn more_async_work() {
64
        ///     // more here
65
        /// }
66
        ///
67
        /// #[tokio::main]
68
        /// async fn main() {
69
        ///     let (first, second) = tokio::join!(
70
        ///         do_stuff_async(),
71
        ///         more_async_work());
72
        ///
73
        ///     // do something with the values
74
        /// }
75
        /// ```
76
        ///
77
        /// Using the `biased;` mode to control polling order.
78
        ///
79
        /// ```
80
        /// async fn do_stuff_async() {
81
        ///     // async work
82
        /// }
83
        ///
84
        /// async fn more_async_work() {
85
        ///     // more here
86
        /// }
87
        ///
88
        /// #[tokio::main]
89
        /// async fn main() {
90
        ///     let (first, second) = tokio::join!(
91
        ///         biased;
92
        ///         do_stuff_async(),
93
        ///         more_async_work()
94
        ///     );
95
        ///
96
        ///     // do something with the values
97
        /// }
98
        /// ```
99
100
        #[macro_export]
101
        #[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
102
        $join
103
    };
104
}
105
106
#[cfg(doc)]
107
doc! {macro_rules! join {
108
    ($(biased;)? $($future:expr),*) => { unimplemented!() }
109
}}
110
111
#[cfg(not(doc))]
112
doc! {macro_rules! join {
113
    (@ {
114
        // Type of rotator that controls which inner future to start with
115
        // when polling our output future.
116
        rotator=$rotator:ty;
117
118
        // One `_` for each branch in the `join!` macro. This is not used once
119
        // normalization is complete.
120
        ( $($count:tt)* )
121
122
        // The expression `0+1+1+ ... +1` equal to the number of branches.
123
        ( $($total:tt)* )
124
125
        // Normalized join! branches
126
        $( ( $($skip:tt)* ) $e:expr, )*
127
128
    }) => {{
129
        use $crate::macros::support::{maybe_done, poll_fn, Future, Pin};
130
        use $crate::macros::support::Poll::{Ready, Pending};
131
132
        // Safety: nothing must be moved out of `futures`. This is to satisfy
133
        // the requirement of `Pin::new_unchecked` called below.
134
        //
135
        // We can't use the `pin!` macro for this because `futures` is a tuple
136
        // and the standard library provides no way to pin-project to the fields
137
        // of a tuple.
138
        let mut futures = ( $( maybe_done($e), )* );
139
140
        // This assignment makes sure that the `poll_fn` closure only has a
141
        // reference to the futures, instead of taking ownership of them. This
142
        // mitigates the issue described in
143
        // <https://internals.rust-lang.org/t/surprising-soundness-trouble-around-pollfn/17484>
144
        let mut futures = &mut futures;
145
146
        const COUNT: u32 = $($total)*;
147
148
        // Each time the future created by poll_fn is polled, if not using biased mode,
149
        // a different future is polled first to ensure every future passed to join!
150
        // can make progress even if one of the futures consumes the whole budget.
151
        let mut rotator = <$rotator>::default();
152
153
        poll_fn(move |cx| {
154
            let mut is_pending = false;
155
            let mut to_run = COUNT;
156
157
            // The number of futures that will be skipped in the first loop iteration.
158
            let mut skip = rotator.num_skip();
159
160
            // This loop runs twice and the first `skip` futures
161
            // are not polled in the first iteration.
162
            loop {
163
            $(
164
                if skip == 0 {
165
                    if to_run == 0 {
166
                        // Every future has been polled
167
                        break;
168
                    }
169
                    to_run -= 1;
170
171
                    // Extract the future for this branch from the tuple.
172
                    let ( $($skip,)* fut, .. ) = &mut *futures;
173
174
                    // Safety: future is stored on the stack above
175
                    // and never moved.
176
                    let mut fut = unsafe { Pin::new_unchecked(fut) };
177
178
                    // Try polling
179
                    if fut.poll(cx).is_pending() {
180
                        is_pending = true;
181
                    }
182
                } else {
183
                    // Future skipped, one less future to skip in the next iteration
184
                    skip -= 1;
185
                }
186
            )*
187
            }
188
189
            if is_pending {
190
                Pending
191
            } else {
192
                Ready(($({
193
                    // Extract the future for this branch from the tuple.
194
                    let ( $($skip,)* fut, .. ) = &mut futures;
195
196
                    // Safety: future is stored on the stack above
197
                    // and never moved.
198
                    let mut fut = unsafe { Pin::new_unchecked(fut) };
199
200
                    fut.take_output().expect("expected completed future")
201
                },)*))
202
            }
203
        }).await
204
    }};
205
206
    // ===== Normalize =====
207
208
    (@ { rotator=$rotator:ty; ( $($s:tt)* ) ( $($n:tt)* ) $($t:tt)* } $e:expr, $($r:tt)* ) => {
209
        $crate::join!(@{ rotator=$rotator; ($($s)* _) ($($n)* + 1) $($t)* ($($s)*) $e, } $($r)*)
210
    };
211
212
    // ===== Entry point =====
213
    ( biased; $($e:expr),+ $(,)?) => {
214
        $crate::join!(@{ rotator=$crate::macros::support::BiasedRotator; () (0) } $($e,)*)
215
    };
216
217
    ( $($e:expr),+ $(,)?) => {
218
        $crate::join!(@{ rotator=$crate::macros::support::Rotator<COUNT>; () (0) } $($e,)*)
219
    };
220
221
    (biased;) => { async {}.await };
222
223
    () => { async {}.await }
224
}}
225
226
/// Rotates by one each [`Self::num_skip`] call up to COUNT - 1.
227
#[derive(Default, Debug)]
228
pub struct Rotator<const COUNT: u32> {
229
    next: u32,
230
}
231
232
impl<const COUNT: u32> Rotator<COUNT> {
233
    /// Rotates by one each [`Self::num_skip`] call up to COUNT - 1
234
    #[inline]
235
0
    pub fn num_skip(&mut self) -> u32 {
236
0
        let num_skip = self.next;
237
0
        self.next += 1;
238
0
        if self.next == COUNT {
239
0
            self.next = 0;
240
0
        }
241
0
        num_skip
242
0
    }
243
}
244
245
/// [`Self::num_skip`] always returns 0.
246
#[derive(Default, Debug)]
247
pub struct BiasedRotator {}
248
249
impl BiasedRotator {
250
    /// Always returns 0.
251
    #[inline]
252
0
    pub fn num_skip(&mut self) -> u32 {
253
0
        0
254
0
    }
255
}