Source code
Revision control
Copy as Markdown
Other Tools
/// Waits on multiple concurrent branches, returning when **all** branches
/// complete.
///
/// The `join!` macro must be used inside of async functions, closures, and
/// blocks.
///
/// The `join!` macro takes a list of async expressions and evaluates them
/// concurrently on the same task. Each async expression evaluates to a future
/// and the futures from each expression are multiplexed on the current task.
///
/// When working with async expressions returning `Result`, `join!` will wait
/// for **all** branches complete regardless if any complete with `Err`. Use
/// [`try_join!`] to return early when `Err` is encountered.
///
/// [`try_join!`]: crate::try_join
///
/// # Notes
///
/// The supplied futures are stored inline and does not require allocating a
/// `Vec`.
///
/// ### Runtime characteristics
///
/// By running all async expressions on the current task, the expressions are
/// able to run **concurrently** but not in **parallel**. This means all
/// expressions are run on the same thread and if one branch blocks the thread,
/// all other expressions will be unable to continue. If parallelism is
/// required, spawn each async expression using [`tokio::spawn`] and pass the
/// join handle to `join!`.
///
/// [`tokio::spawn`]: crate::spawn
///
/// # Examples
///
/// Basic join with two branches
///
/// ```
/// async fn do_stuff_async() {
/// // async work
/// }
///
/// async fn more_async_work() {
/// // more here
/// }
///
/// #[tokio::main]
/// async fn main() {
/// let (first, second) = tokio::join!(
/// do_stuff_async(),
/// more_async_work());
///
/// // do something with the values
/// }
/// ```
#[macro_export]
#[cfg_attr(docsrs, doc(cfg(feature = "macros")))]
macro_rules! join {
(@ {
// One `_` for each branch in the `join!` macro. This is not used once
// normalization is complete.
( $($count:tt)* )
// The expression `0+1+1+ ... +1` equal to the number of branches.
( $($total:tt)* )
// Normalized join! branches
$( ( $($skip:tt)* ) $e:expr, )*
}) => {{
use $crate::macros::support::{maybe_done, poll_fn, Future, Pin};
use $crate::macros::support::Poll::{Ready, Pending};
// Safety: nothing must be moved out of `futures`. This is to satisfy
// the requirement of `Pin::new_unchecked` called below.
//
// We can't use the `pin!` macro for this because `futures` is a tuple
// and the standard library provides no way to pin-project to the fields
// of a tuple.
let mut futures = ( $( maybe_done($e), )* );
// This assignment makes sure that the `poll_fn` closure only has a
// reference to the futures, instead of taking ownership of them. This
// mitigates the issue described in
let mut futures = &mut futures;
// Each time the future created by poll_fn is polled, a different future will be polled first
// to ensure every future passed to join! gets a chance to make progress even if
// one of the futures consumes the whole budget.
//
// This is number of futures that will be skipped in the first loop
// iteration the next time.
let mut skip_next_time: u32 = 0;
poll_fn(move |cx| {
const COUNT: u32 = $($total)*;
let mut is_pending = false;
let mut to_run = COUNT;
// The number of futures that will be skipped in the first loop iteration.
let mut skip = skip_next_time;
skip_next_time = if skip + 1 == COUNT { 0 } else { skip + 1 };
// This loop runs twice and the first `skip` futures
// are not polled in the first iteration.
loop {
$(
if skip == 0 {
if to_run == 0 {
// Every future has been polled
break;
}
to_run -= 1;
// Extract the future for this branch from the tuple.
let ( $($skip,)* fut, .. ) = &mut *futures;
// Safety: future is stored on the stack above
// and never moved.
let mut fut = unsafe { Pin::new_unchecked(fut) };
// Try polling
if fut.poll(cx).is_pending() {
is_pending = true;
}
} else {
// Future skipped, one less future to skip in the next iteration
skip -= 1;
}
)*
}
if is_pending {
Pending
} else {
Ready(($({
// Extract the future for this branch from the tuple.
let ( $($skip,)* fut, .. ) = &mut futures;
// Safety: future is stored on the stack above
// and never moved.
let mut fut = unsafe { Pin::new_unchecked(fut) };
fut.take_output().expect("expected completed future")
},)*))
}
}).await
}};
// ===== Normalize =====
(@ { ( $($s:tt)* ) ( $($n:tt)* ) $($t:tt)* } $e:expr, $($r:tt)* ) => {
$crate::join!(@{ ($($s)* _) ($($n)* + 1) $($t)* ($($s)*) $e, } $($r)*)
};
// ===== Entry point =====
( $($e:expr),+ $(,)?) => {
$crate::join!(@{ () (0) } $($e,)*)
};
() => { async {}.await }
}