diff options
author | Carl Lerche <me@carllerche.com> | 2020-01-23 13:24:30 -0800 |
---|---|---|
committer | GitHub <noreply@github.com> | 2020-01-23 13:24:30 -0800 |
commit | 7079bcd60975f592e08fcd575991f6ae2a409a1f (patch) | |
tree | b4dc701039f9a8d984fa9d5b3835b8094d6359f1 /tokio/src/macros/join.rs | |
parent | f8714e9901981d5d1cc66c9a03d5a8a0b2f2eb4b (diff) |
future: provide join! macro (#2158)
Provides a `join!` macro that supports concurrently driving multiple
futures on the same task and await the completion of all futures.
Diffstat (limited to 'tokio/src/macros/join.rs')
-rw-r--r-- | tokio/src/macros/join.rs | 112 |
1 files changed, 112 insertions, 0 deletions
diff --git a/tokio/src/macros/join.rs b/tokio/src/macros/join.rs new file mode 100644 index 00000000..c164bb04 --- /dev/null +++ b/tokio/src/macros/join.rs @@ -0,0 +1,112 @@ +/// Wait 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. +/// +/// # 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] +macro_rules! join { + (@ { + // One `_` for each branch in the `join!` macro. This is not used once + // normalization is complete. + ( $($count: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. + let mut futures = ( $( maybe_done($e), )* ); + + poll_fn(move |cx| { + let mut is_pending = false; + + $( + // 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; + } + )* + + 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)* ) $($t:tt)* } $e:expr, $($r:tt)* ) => { + $crate::join!(@{ ($($s)* _) $($t)* ($($s)*) $e, } $($r)*) + }; + + // ===== Entry point ===== + + ( $($e:expr),* $(,)?) => { + $crate::join!(@{ () } $($e,)*) + }; +} |