541 lines
30 KiB
HTML
541 lines
30 KiB
HTML
<!DOCTYPE html>
|
||
<html lang="en">
|
||
<head>
|
||
<meta charset="UTF-8">
|
||
<title>Applying Concurrency with Async</title>
|
||
</head>
|
||
<body>
|
||
<!-- Old headings. Do not remove or links may break. -->
|
||
<p><a id="concurrency-with-async"></a></p>
|
||
<h2 id="applying-concurrency-with-async"><a class="header" href="#applying-concurrency-with-async">Applying Concurrency with Async</a></h2>
|
||
<p>In this section, we’ll apply async to some of the same concurrency challenges
|
||
we tackled with threads in Chapter 16. Because we already talked about a lot of
|
||
the key ideas there, in this section we’ll focus on what’s different between
|
||
threads and futures.</p>
|
||
<p>In many cases, the APIs for working with concurrency using async are very
|
||
similar to those for using threads. In other cases, they end up being quite
|
||
different. Even when the APIs <em>look</em> similar between threads and async, they
|
||
often have different behavior—and they nearly always have different performance
|
||
characteristics.</p>
|
||
<!-- Old headings. Do not remove or links may break. -->
|
||
<p><a id="counting"></a></p>
|
||
<h3 id="creating-a-new-task-with-spawn_task"><a class="header" href="#creating-a-new-task-with-spawn_task">Creating a New Task with <code>spawn_task</code></a></h3>
|
||
<p>The first operation we tackled in the <a href="../ch16/ch16-01-threads.html#creating-a-new-thread-with-spawn">“Creating a New Thread with
|
||
<code>spawn</code>”</a><!-- ignore --> section in Chapter 16 was counting up on
|
||
two separate threads. Let’s do the same using async. The <code>trpl</code> crate supplies
|
||
a <code>spawn_task</code> function that looks very similar to the <code>thread::spawn</code> API, and
|
||
a <code>sleep</code> function that is an async version of the <code>thread::sleep</code> API. We can
|
||
use these together to implement the counting example, as shown in Listing 17-6.</p>
|
||
<figure class="listing" id="listing-17-6">
|
||
<span class="file-name">Filename: src/main.rs</span>
|
||
<pre class="playground"><code class="language-rust edition2024"><span class="boring">extern crate trpl; // required for mdbook test
|
||
</span><span class="boring">
|
||
</span>use std::time::Duration;
|
||
|
||
fn main() {
|
||
trpl::block_on(async {
|
||
trpl::spawn_task(async {
|
||
for i in 1..10 {
|
||
println!("hi number {i} from the first task!");
|
||
trpl::sleep(Duration::from_millis(500)).await;
|
||
}
|
||
});
|
||
|
||
for i in 1..5 {
|
||
println!("hi number {i} from the second task!");
|
||
trpl::sleep(Duration::from_millis(500)).await;
|
||
}
|
||
});
|
||
}</code></pre>
|
||
<figcaption><a href="#listing-17-6">Listing 17-6</a>: Creating a new task to print one thing while the main task prints something else</figcaption>
|
||
</figure>
|
||
<p>As our starting point, we set up our <code>main</code> function with <code>trpl::block_on</code> so
|
||
that our top-level function can be async.</p>
|
||
<section class="note" aria-role="note">
|
||
<p>Note: From this point forward in the chapter, every example will include this
|
||
exact same wrapping code with <code>trpl::block_on</code> in <code>main</code>, so we’ll often skip it
|
||
just as we do with <code>main</code>. Remember to include it in your code!</p>
|
||
</section>
|
||
<p>Then we write two loops within that block, each containing a <code>trpl::sleep</code>
|
||
call, which waits for half a second (500 milliseconds) before sending the next
|
||
message. We put one loop in the body of a <code>trpl::spawn_task</code> and the other in a
|
||
top-level <code>for</code> loop. We also add an <code>await</code> after the <code>sleep</code> calls.</p>
|
||
<p>This code behaves similarly to the thread-based implementation—including the
|
||
fact that you may see the messages appear in a different order in your own
|
||
terminal when you run it:</p>
|
||
<!-- Not extracting output because changes to this output aren't significant;
|
||
the changes are likely to be due to the threads running differently rather than
|
||
changes in the compiler -->
|
||
<pre><code class="language-text">hi number 1 from the second task!
|
||
hi number 1 from the first task!
|
||
hi number 2 from the first task!
|
||
hi number 2 from the second task!
|
||
hi number 3 from the first task!
|
||
hi number 3 from the second task!
|
||
hi number 4 from the first task!
|
||
hi number 4 from the second task!
|
||
hi number 5 from the first task!
|
||
</code></pre>
|
||
<p>This version stops as soon as the <code>for</code> loop in the body of the main async
|
||
block finishes, because the task spawned by <code>spawn_task</code> is shut down when the
|
||
<code>main</code> function ends. If you want it to run all the way to the task’s
|
||
completion, you will need to use a join handle to wait for the first task to
|
||
complete. With threads, we used the <code>join</code> method to “block” until the thread
|
||
was done running. In Listing 17-7, we can use <code>await</code> to do the same thing,
|
||
because the task handle itself is a future. Its <code>Output</code> type is a <code>Result</code>, so
|
||
we also unwrap it after awaiting it.</p>
|
||
<figure class="listing" id="listing-17-7">
|
||
<span class="file-name">Filename: src/main.rs</span>
|
||
<pre class="playground"><code class="language-rust edition2024"><span class="boring">extern crate trpl; // required for mdbook test
|
||
</span><span class="boring">
|
||
</span><span class="boring">use std::time::Duration;
|
||
</span><span class="boring">
|
||
</span><span class="boring">fn main() {
|
||
</span><span class="boring"> trpl::block_on(async {
|
||
</span> let handle = trpl::spawn_task(async {
|
||
for i in 1..10 {
|
||
println!("hi number {i} from the first task!");
|
||
trpl::sleep(Duration::from_millis(500)).await;
|
||
}
|
||
});
|
||
|
||
for i in 1..5 {
|
||
println!("hi number {i} from the second task!");
|
||
trpl::sleep(Duration::from_millis(500)).await;
|
||
}
|
||
|
||
handle.await.unwrap();
|
||
<span class="boring"> });
|
||
</span><span class="boring">}</span></code></pre>
|
||
<figcaption><a href="#listing-17-7">Listing 17-7</a>: Using <code>await</code> with a join handle to run a task to completion</figcaption>
|
||
</figure>
|
||
<p>This updated version runs until <em>both</em> loops finish:</p>
|
||
<!-- Not extracting output because changes to this output aren't significant;
|
||
the changes are likely to be due to the threads running differently rather than
|
||
changes in the compiler -->
|
||
<pre><code class="language-text">hi number 1 from the second task!
|
||
hi number 1 from the first task!
|
||
hi number 2 from the first task!
|
||
hi number 2 from the second task!
|
||
hi number 3 from the first task!
|
||
hi number 3 from the second task!
|
||
hi number 4 from the first task!
|
||
hi number 4 from the second task!
|
||
hi number 5 from the first task!
|
||
hi number 6 from the first task!
|
||
hi number 7 from the first task!
|
||
hi number 8 from the first task!
|
||
hi number 9 from the first task!
|
||
</code></pre>
|
||
<p>So far, it looks like async and threads give us similar outcomes, just with
|
||
different syntax: using <code>await</code> instead of calling <code>join</code> on the join handle,
|
||
and awaiting the <code>sleep</code> calls.</p>
|
||
<p>The bigger difference is that we didn’t need to spawn another operating system
|
||
thread to do this. In fact, we don’t even need to spawn a task here. Because
|
||
async blocks compile to anonymous futures, we can put each loop in an async
|
||
block and have the runtime run them both to completion using the <code>trpl::join</code>
|
||
function.</p>
|
||
<p>In the <a href="../ch16/ch16-01-threads.html#waiting-for-all-threads-to-finish">“Waiting for All Threads to Finish”</a><!-- ignore -->
|
||
section in Chapter 16, we showed how to use the <code>join</code> method on the
|
||
<code>JoinHandle</code> type returned when you call <code>std::thread::spawn</code>. The <code>trpl::join</code>
|
||
function is similar, but for futures. When you give it two futures, it produces
|
||
a single new future whose output is a tuple containing the output of each
|
||
future you passed in once they <em>both</em> complete. Thus, in Listing 17-8, we use
|
||
<code>trpl::join</code> to wait for both <code>fut1</code> and <code>fut2</code> to finish. We do <em>not</em> await
|
||
<code>fut1</code> and <code>fut2</code> but instead the new future produced by <code>trpl::join</code>. We
|
||
ignore the output, because it’s just a tuple containing two unit values.</p>
|
||
<figure class="listing" id="listing-17-8">
|
||
<span class="file-name">Filename: src/main.rs</span>
|
||
<pre class="playground"><code class="language-rust edition2024"><span class="boring">extern crate trpl; // required for mdbook test
|
||
</span><span class="boring">
|
||
</span><span class="boring">use std::time::Duration;
|
||
</span><span class="boring">
|
||
</span><span class="boring">fn main() {
|
||
</span><span class="boring"> trpl::block_on(async {
|
||
</span> let fut1 = async {
|
||
for i in 1..10 {
|
||
println!("hi number {i} from the first task!");
|
||
trpl::sleep(Duration::from_millis(500)).await;
|
||
}
|
||
};
|
||
|
||
let fut2 = async {
|
||
for i in 1..5 {
|
||
println!("hi number {i} from the second task!");
|
||
trpl::sleep(Duration::from_millis(500)).await;
|
||
}
|
||
};
|
||
|
||
trpl::join(fut1, fut2).await;
|
||
<span class="boring"> });
|
||
</span><span class="boring">}</span></code></pre>
|
||
<figcaption><a href="#listing-17-8">Listing 17-8</a>: Using <code>trpl::join</code> to await two anonymous futures</figcaption>
|
||
</figure>
|
||
<p>When we run this, we see both futures run to completion:</p>
|
||
<!-- Not extracting output because changes to this output aren't significant;
|
||
the changes are likely to be due to the threads running differently rather than
|
||
changes in the compiler -->
|
||
<pre><code class="language-text">hi number 1 from the first task!
|
||
hi number 1 from the second task!
|
||
hi number 2 from the first task!
|
||
hi number 2 from the second task!
|
||
hi number 3 from the first task!
|
||
hi number 3 from the second task!
|
||
hi number 4 from the first task!
|
||
hi number 4 from the second task!
|
||
hi number 5 from the first task!
|
||
hi number 6 from the first task!
|
||
hi number 7 from the first task!
|
||
hi number 8 from the first task!
|
||
hi number 9 from the first task!
|
||
</code></pre>
|
||
<p>Now, you’ll see the exact same order every time, which is very different from
|
||
what we saw with threads and with <code>trpl::spawn_task</code> in Listing 17-7. That is
|
||
because the <code>trpl::join</code> function is <em>fair</em>, meaning it checks each future
|
||
equally often, alternating between them, and never lets one race ahead if the
|
||
other is ready. With threads, the operating system decides which thread to
|
||
check and how long to let it run. With async Rust, the runtime decides which
|
||
task to check. (In practice, the details get complicated because an async
|
||
runtime might use operating system threads under the hood as part of how it
|
||
manages concurrency, so guaranteeing fairness can be more work for a
|
||
runtime—but it’s still possible!) Runtimes don’t have to guarantee fairness for
|
||
any given operation, and they often offer different APIs to let you choose
|
||
whether or not you want fairness.</p>
|
||
<p>Try some of these variations on awaiting the futures and see what they do:</p>
|
||
<ul>
|
||
<li>Remove the async block from around either or both of the loops.</li>
|
||
<li>Await each async block immediately after defining it.</li>
|
||
<li>Wrap only the first loop in an async block, and await the resulting future
|
||
after the body of second loop.</li>
|
||
</ul>
|
||
<p>For an extra challenge, see if you can figure out what the output will be in
|
||
each case <em>before</em> running the code!</p>
|
||
<!-- Old headings. Do not remove or links may break. -->
|
||
<p><a id="message-passing"></a>
|
||
<a id="counting-up-on-two-tasks-using-message-passing"></a></p>
|
||
<h3 id="sending-data-between-two-tasks-using-message-passing"><a class="header" href="#sending-data-between-two-tasks-using-message-passing">Sending Data Between Two Tasks Using Message Passing</a></h3>
|
||
<p>Sharing data between futures will also be familiar: we’ll use message passing
|
||
again, but this time with async versions of the types and functions. We’ll take
|
||
a slightly different path than we did in the <a href="../ch16/ch16-02-message-passing.html">“Transfer Data Between Threads
|
||
with Message Passing”</a><!-- ignore --> section in
|
||
Chapter 16 to illustrate some of the key differences between thread-based and
|
||
futures-based concurrency. In Listing 17-9, we’ll begin with just a single
|
||
async block—<em>not</em> spawning a separate task as we spawned a separate thread.</p>
|
||
<figure class="listing" id="listing-17-9">
|
||
<span class="file-name">Filename: src/main.rs</span>
|
||
<pre class="playground"><code class="language-rust edition2024"><span class="boring">extern crate trpl; // required for mdbook test
|
||
</span><span class="boring">
|
||
</span><span class="boring">fn main() {
|
||
</span><span class="boring"> trpl::block_on(async {
|
||
</span> let (tx, mut rx) = trpl::channel();
|
||
|
||
let val = String::from("hi");
|
||
tx.send(val).unwrap();
|
||
|
||
let received = rx.recv().await.unwrap();
|
||
println!("received '{received}'");
|
||
<span class="boring"> });
|
||
</span><span class="boring">}</span></code></pre>
|
||
<figcaption><a href="#listing-17-9">Listing 17-9</a>: Creating an async channel and assigning the two halves to <code>tx</code> and <code>rx</code></figcaption>
|
||
</figure>
|
||
<p>Here, we use <code>trpl::channel</code>, an async version of the multiple-producer,
|
||
single-consumer channel API we used with threads back in Chapter 16. The async
|
||
version of the API is only a little different from the thread-based version: it
|
||
uses a mutable rather than an immutable receiver <code>rx</code>, and its <code>recv</code> method
|
||
produces a future we need to await rather than producing the value directly.
|
||
Now we can send messages from the sender to the receiver. Notice that we don’t
|
||
have to spawn a separate thread or even a task; we merely need to await the
|
||
<code>rx.recv</code> call.</p>
|
||
<p>The synchronous <code>Receiver::recv</code> method in <code>std::mpsc::channel</code> blocks until it
|
||
receives a message. The <code>trpl::Receiver::recv</code> method does not, because it is
|
||
async. Instead of blocking, it hands control back to the runtime until either a
|
||
message is received or the send side of the channel closes. By contrast, we
|
||
don’t await the <code>send</code> call, because it doesn’t block. It doesn’t need to,
|
||
because the channel we’re sending it into is unbounded.</p>
|
||
<section class="note" aria-role="note">
|
||
<p>Note: Because all of this async code runs in an async block in a
|
||
<code>trpl::block_on</code> call, everything within it can avoid blocking. However, the
|
||
code <em>outside</em> it will block on the <code>block_on</code> function returning. That’s the
|
||
whole point of the <code>trpl::block_on</code> function: it lets you <em>choose</em> where to
|
||
block on some set of async code, and thus where to transition between sync
|
||
and async code.</p>
|
||
</section>
|
||
<p>Notice two things about this example. First, the message will arrive right
|
||
away. Second, although we use a future here, there’s no concurrency yet.
|
||
Everything in the listing happens in sequence, just as it would if there were
|
||
no futures involved.</p>
|
||
<p>Let’s address the first part by sending a series of messages and sleeping in
|
||
between them, as shown in Listing 17-10.</p>
|
||
<!-- We cannot test this one because it never stops! -->
|
||
<figure class="listing" id="listing-17-10">
|
||
<span class="file-name">Filename: src/main.rs</span>
|
||
<pre><code class="language-rust ignore"><span class="boring">extern crate trpl; // required for mdbook test
|
||
</span><span class="boring">
|
||
</span><span class="boring">use std::time::Duration;
|
||
</span><span class="boring">
|
||
</span><span class="boring">fn main() {
|
||
</span><span class="boring"> trpl::block_on(async {
|
||
</span> let (tx, mut rx) = trpl::channel();
|
||
|
||
let vals = vec![
|
||
String::from("hi"),
|
||
String::from("from"),
|
||
String::from("the"),
|
||
String::from("future"),
|
||
];
|
||
|
||
for val in vals {
|
||
tx.send(val).unwrap();
|
||
trpl::sleep(Duration::from_millis(500)).await;
|
||
}
|
||
|
||
while let Some(value) = rx.recv().await {
|
||
println!("received '{value}'");
|
||
}
|
||
<span class="boring"> });
|
||
</span><span class="boring">}</span></code></pre>
|
||
<figcaption><a href="#listing-17-10">Listing 17-10</a>: Sending and receiving multiple messages over the async channel and sleeping with an <code>await</code> between each message</figcaption>
|
||
</figure>
|
||
<p>In addition to sending the messages, we need to receive them. In this case,
|
||
because we know how many messages are coming in, we could do that manually by
|
||
calling <code>rx.recv().await</code> four times. In the real world, though, we’ll generally
|
||
be waiting on some <em>unknown</em> number of messages, so we need to keep waiting
|
||
until we determine that there are no more messages.</p>
|
||
<p>In Listing 16-10, we used a <code>for</code> loop to process all the items received from a
|
||
synchronous channel. Rust doesn’t yet have a way to use a <code>for</code> loop with an
|
||
<em>asynchronously produced</em> series of items, however, so we need to use a loop we
|
||
haven’t seen before: the <code>while let</code> conditional loop. This is the loop version
|
||
of the <code>if let</code> construct we saw back in the <a href="../ch06/ch06-03-if-let.html">“Concise Control Flow with <code>if let</code> and <code>let...else</code>”</a><!-- ignore --> section in Chapter 6. The loop
|
||
will continue executing as long as the pattern it specifies continues to match
|
||
the value.</p>
|
||
<p>The <code>rx.recv</code> call produces a future, which we await. The runtime will pause
|
||
the future until it is ready. Once a message arrives, the future will resolve
|
||
to <code>Some(message)</code> as many times as a message arrives. When the channel closes,
|
||
regardless of whether <em>any</em> messages have arrived, the future will instead
|
||
resolve to <code>None</code> to indicate that there are no more values and thus we should
|
||
stop polling—that is, stop awaiting.</p>
|
||
<p>The <code>while let</code> loop pulls all of this together. If the result of calling
|
||
<code>rx.recv().await</code> is <code>Some(message)</code>, we get access to the message and we can
|
||
use it in the loop body, just as we could with <code>if let</code>. If the result is
|
||
<code>None</code>, the loop ends. Every time the loop completes, it hits the await point
|
||
again, so the runtime pauses it again until another message arrives.</p>
|
||
<p>The code now successfully sends and receives all of the messages.
|
||
Unfortunately, there are still a couple of problems. For one thing, the
|
||
messages do not arrive at half-second intervals. They arrive all at once, 2
|
||
seconds (2,000 milliseconds) after we start the program. For another, this
|
||
program also never exits! Instead, it waits forever for new messages. You will
|
||
need to shut it down using <kbd>ctrl</kbd>-<kbd>C</kbd>.</p>
|
||
<h4 id="code-within-one-async-block-executes-linearly"><a class="header" href="#code-within-one-async-block-executes-linearly">Code Within One Async Block Executes Linearly</a></h4>
|
||
<p>Let’s start by examining why the messages come in all at once after the full
|
||
delay, rather than coming in with delays between each one. Within a given async
|
||
block, the order in which <code>await</code> keywords appear in the code is also the order
|
||
in which they’re executed when the program runs.</p>
|
||
<p>There’s only one async block in Listing 17-10, so everything in it runs
|
||
linearly. There’s still no concurrency. All the <code>tx.send</code> calls happen,
|
||
interspersed with all of the <code>trpl::sleep</code> calls and their associated await
|
||
points. Only then does the <code>while let</code> loop get to go through any of the
|
||
<code>await</code> points on the <code>recv</code> calls.</p>
|
||
<p>To get the behavior we want, where the sleep delay happens between each
|
||
message, we need to put the <code>tx</code> and <code>rx</code> operations in their own async blocks,
|
||
as shown in Listing 17-11. Then the runtime can execute each of them separately
|
||
using <code>trpl::join</code>, just as in Listing 17-8. Once again, we await the result of
|
||
calling <code>trpl::join</code>, not the individual futures. If we awaited the individual
|
||
futures in sequence, we would just end up back in a sequential flow—exactly
|
||
what we’re trying <em>not</em> to do.</p>
|
||
<!-- We cannot test this one because it never stops! -->
|
||
<figure class="listing" id="listing-17-11">
|
||
<span class="file-name">Filename: src/main.rs</span>
|
||
<pre><code class="language-rust ignore"><span class="boring">extern crate trpl; // required for mdbook test
|
||
</span><span class="boring">
|
||
</span><span class="boring">use std::time::Duration;
|
||
</span><span class="boring">
|
||
</span><span class="boring">fn main() {
|
||
</span><span class="boring"> trpl::block_on(async {
|
||
</span><span class="boring"> let (tx, mut rx) = trpl::channel();
|
||
</span><span class="boring">
|
||
</span> let tx_fut = async {
|
||
let vals = vec![
|
||
String::from("hi"),
|
||
String::from("from"),
|
||
String::from("the"),
|
||
String::from("future"),
|
||
];
|
||
|
||
for val in vals {
|
||
tx.send(val).unwrap();
|
||
trpl::sleep(Duration::from_millis(500)).await;
|
||
}
|
||
};
|
||
|
||
let rx_fut = async {
|
||
while let Some(value) = rx.recv().await {
|
||
println!("received '{value}'");
|
||
}
|
||
};
|
||
|
||
trpl::join(tx_fut, rx_fut).await;
|
||
<span class="boring"> });
|
||
</span><span class="boring">}</span></code></pre>
|
||
<figcaption><a href="#listing-17-11">Listing 17-11</a>: Separating <code>send</code> and <code>recv</code> into their own <code>async</code> blocks and awaiting the futures for those blocks</figcaption>
|
||
</figure>
|
||
<p>With the updated code in Listing 17-11, the messages get printed at
|
||
500-millisecond intervals, rather than all in a rush after 2 seconds.</p>
|
||
<h4 id="moving-ownership-into-an-async-block"><a class="header" href="#moving-ownership-into-an-async-block">Moving Ownership Into an Async Block</a></h4>
|
||
<p>The program still never exits, though, because of the way the <code>while let</code> loop
|
||
interacts with <code>trpl::join</code>:</p>
|
||
<ul>
|
||
<li>The future returned from <code>trpl::join</code> completes only once <em>both</em> futures
|
||
passed to it have completed.</li>
|
||
<li>The <code>tx_fut</code> future completes once it finishes sleeping after sending the last
|
||
message in <code>vals</code>.</li>
|
||
<li>The <code>rx_fut</code> future won’t complete until the <code>while let</code> loop ends.</li>
|
||
<li>The <code>while let</code> loop won’t end until awaiting <code>rx.recv</code> produces <code>None</code>.</li>
|
||
<li>Awaiting <code>rx.recv</code> will return <code>None</code> only once the other end of the channel
|
||
is closed.</li>
|
||
<li>The channel will close only if we call <code>rx.close</code> or when the sender side,
|
||
<code>tx</code>, is dropped.</li>
|
||
<li>We don’t call <code>rx.close</code> anywhere, and <code>tx</code> won’t be dropped until the
|
||
outermost async block passed to <code>trpl::block_on</code> ends.</li>
|
||
<li>The block can’t end because it is blocked on <code>trpl::join</code> completing, which
|
||
takes us back to the top of this list.</li>
|
||
</ul>
|
||
<p>Right now, the async block where we send the messages only <em>borrows</em> <code>tx</code>
|
||
because sending a message doesn’t require ownership, but if we could <em>move</em>
|
||
<code>tx</code> into that async block, it would be dropped once that block ends. In the
|
||
<a href="../ch13/ch13-01-closures.html#capturing-references-or-moving-ownership">“Capturing References or Moving Ownership”</a><!-- ignore -->
|
||
section in Chapter 13, you learned how to use the <code>move</code> keyword with closures,
|
||
and, as discussed in the <a href="../ch16/ch16-01-threads.html#using-move-closures-with-threads">“Using <code>move</code> Closures with
|
||
Threads”</a><!-- ignore --> section in Chapter 16, we often need to
|
||
move data into closures when working with threads. The same basic dynamics
|
||
apply to async blocks, so the <code>move</code> keyword works with async blocks just as it
|
||
does with closures.</p>
|
||
<p>In Listing 17-12, we change the block used to send messages from <code>async</code> to
|
||
<code>async move</code>.</p>
|
||
<figure class="listing" id="listing-17-12">
|
||
<span class="file-name">Filename: src/main.rs</span>
|
||
<pre class="playground"><code class="language-rust edition2024"><span class="boring">extern crate trpl; // required for mdbook test
|
||
</span><span class="boring">
|
||
</span><span class="boring">use std::time::Duration;
|
||
</span><span class="boring">
|
||
</span><span class="boring">fn main() {
|
||
</span><span class="boring"> trpl::block_on(async {
|
||
</span> let (tx, mut rx) = trpl::channel();
|
||
|
||
let tx_fut = async move {
|
||
// --snip--
|
||
<span class="boring"> let vals = vec![
|
||
</span><span class="boring"> String::from("hi"),
|
||
</span><span class="boring"> String::from("from"),
|
||
</span><span class="boring"> String::from("the"),
|
||
</span><span class="boring"> String::from("future"),
|
||
</span><span class="boring"> ];
|
||
</span><span class="boring">
|
||
</span><span class="boring"> for val in vals {
|
||
</span><span class="boring"> tx.send(val).unwrap();
|
||
</span><span class="boring"> trpl::sleep(Duration::from_millis(500)).await;
|
||
</span><span class="boring"> }
|
||
</span><span class="boring"> };
|
||
</span><span class="boring">
|
||
</span><span class="boring"> let rx_fut = async {
|
||
</span><span class="boring"> while let Some(value) = rx.recv().await {
|
||
</span><span class="boring"> println!("received '{value}'");
|
||
</span><span class="boring"> }
|
||
</span><span class="boring"> };
|
||
</span><span class="boring">
|
||
</span><span class="boring"> trpl::join(tx_fut, rx_fut).await;
|
||
</span><span class="boring"> });
|
||
</span><span class="boring">}</span></code></pre>
|
||
<figcaption><a href="#listing-17-12">Listing 17-12</a>: A revision of the code from Listing 17-11 that correctly shuts down when complete</figcaption>
|
||
</figure>
|
||
<p>When we run <em>this</em> version of the code, it shuts down gracefully after the last
|
||
message is sent and received. Next, let’s see what would need to change to send
|
||
data from more than one future.</p>
|
||
<h4 id="joining-a-number-of-futures-with-the-join-macro"><a class="header" href="#joining-a-number-of-futures-with-the-join-macro">Joining a Number of Futures with the <code>join!</code> Macro</a></h4>
|
||
<p>This async channel is also a multiple-producer channel, so we can call <code>clone</code>
|
||
on <code>tx</code> if we want to send messages from multiple futures, as shown in Listing
|
||
17-13.</p>
|
||
<figure class="listing" id="listing-17-13">
|
||
<span class="file-name">Filename: src/main.rs</span>
|
||
<pre class="playground"><code class="language-rust edition2024"><span class="boring">extern crate trpl; // required for mdbook test
|
||
</span><span class="boring">
|
||
</span><span class="boring">use std::time::Duration;
|
||
</span><span class="boring">
|
||
</span><span class="boring">fn main() {
|
||
</span><span class="boring"> trpl::block_on(async {
|
||
</span> let (tx, mut rx) = trpl::channel();
|
||
|
||
let tx1 = tx.clone();
|
||
let tx1_fut = async move {
|
||
let vals = vec![
|
||
String::from("hi"),
|
||
String::from("from"),
|
||
String::from("the"),
|
||
String::from("future"),
|
||
];
|
||
|
||
for val in vals {
|
||
tx1.send(val).unwrap();
|
||
trpl::sleep(Duration::from_millis(500)).await;
|
||
}
|
||
};
|
||
|
||
let rx_fut = async {
|
||
while let Some(value) = rx.recv().await {
|
||
println!("received '{value}'");
|
||
}
|
||
};
|
||
|
||
let tx_fut = async move {
|
||
let vals = vec![
|
||
String::from("more"),
|
||
String::from("messages"),
|
||
String::from("for"),
|
||
String::from("you"),
|
||
];
|
||
|
||
for val in vals {
|
||
tx.send(val).unwrap();
|
||
trpl::sleep(Duration::from_millis(1500)).await;
|
||
}
|
||
};
|
||
|
||
trpl::join!(tx1_fut, tx_fut, rx_fut);
|
||
<span class="boring"> });
|
||
</span><span class="boring">}</span></code></pre>
|
||
<figcaption><a href="#listing-17-13">Listing 17-13</a>: Using multiple producers with async blocks</figcaption>
|
||
</figure>
|
||
<p>First, we clone <code>tx</code>, creating <code>tx1</code> outside the first async block. We move
|
||
<code>tx1</code> into that block just as we did before with <code>tx</code>. Then, later, we move the
|
||
original <code>tx</code> into a <em>new</em> async block, where we send more messages on a
|
||
slightly slower delay. We happen to put this new async block after the async
|
||
block for receiving messages, but it could go before it just as well. The key is
|
||
the order in which the futures are awaited, not in which they’re created.</p>
|
||
<p>Both of the async blocks for sending messages need to be <code>async move</code> blocks so
|
||
that both <code>tx</code> and <code>tx1</code> get dropped when those blocks finish. Otherwise, we’ll
|
||
end up back in the same infinite loop we started out in.</p>
|
||
<p>Finally, we switch from <code>trpl::join</code> to <code>trpl::join!</code> to handle the additional
|
||
future: the <code>join!</code> macro awaits an arbitrary number of futures where we know
|
||
the number of futures at compile time. We’ll discuss awaiting a collection of
|
||
an unknown number of futures later in this chapter.</p>
|
||
<p>Now we see all the messages from both sending futures, and because the sending
|
||
futures use slightly different delays after sending, the messages are also
|
||
received at those different intervals:</p>
|
||
<!-- Not extracting output because changes to this output aren't significant;
|
||
the changes are likely to be due to the threads running differently rather than
|
||
changes in the compiler -->
|
||
<pre><code class="language-text">received 'hi'
|
||
received 'more'
|
||
received 'from'
|
||
received 'the'
|
||
received 'messages'
|
||
received 'future'
|
||
received 'for'
|
||
received 'you'
|
||
</code></pre>
|
||
<p>We’ve explored how to use message passing to send data between futures, how
|
||
code within an async block runs sequentially, how to move ownership into an
|
||
async block, and how to join multiple futures. Next, let’s discuss how and why
|
||
to tell the runtime it can switch to another task.</p>
|
||
</body>
|
||
</html>
|