alex/g3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
68c91359130fef51a757c91f8eac990e28d6b999
g3/tmp/tokio_channels.html
Dhanji R. Prasanna 68c9135913 Fix research tool UI: remove duplicate header, add footer spacing, remove spinner, widen command display 
- Remove duplicate tool header (lib.rs already prints it)
- Add newline before timing footer for visual separation
- Remove spinner animation (incompatible with update_tool_output_line)
- Change shell command format to " > `cmd` ..." with 60 char width
2026-01-10 15:20:40 +11:00

335 lines
73 KiB
HTML
Raw Blame History

<html><head><meta charset="utf-8" data-next-head=""><meta name="viewport" content="width=device-width, initial-scale=1" data-next-head=""><title data-next-head="">Channels | Tokio - An asynchronous Rust runtime</title><meta name="description" content="Tokio is a runtime for writing reliable asynchronous applications with Rust. It provides async I/O, networking, scheduling, timers, and more." data-next-head=""><link rel="alternate icon" type="image/png" href="/favicon-32x32.png" data-next-head=""><link rel="icon" type="image/svg+xml" href="/favicon.svg" data-next-head=""><link rel="alternate" type="application/rss+xml" href="/blog/index.xml" data-next-head=""><link rel="preload" href="/_next/static/css/56c6dcfe840a75e8.css" as="style"><link rel="preload" href="/_next/static/css/8aea088cdc4338f0.css" as="style"><link rel="stylesheet" href="/_next/static/css/56c6dcfe840a75e8.css" data-n-g=""><link rel="stylesheet" href="/_next/static/css/8aea088cdc4338f0.css" data-n-p=""><noscript data-n-css=""></noscript><script defer="" nomodule="" src="/_next/static/chunks/polyfills-42372ed130431b0a.js"></script><script src="/_next/static/chunks/webpack-f8185607c88ffb0b.js" defer=""></script><script src="/_next/static/chunks/framework-516982be4eea35ea.js" defer=""></script><script src="/_next/static/chunks/main-11f62a962ab540a8.js" defer=""></script><script src="/_next/static/chunks/pages/_app-7e9c39878777ce10.js" defer=""></script><script src="/_next/static/chunks/53-e49eff5fcccf0cba.js" defer=""></script><script src="/_next/static/chunks/547-573b033aa0b73315.js" defer=""></script><script src="/_next/static/chunks/pages/%5B...slug%5D-0bd34730781e0ad8.js" defer=""></script><script src="/_next/static/h5xuILhXLuVXe-k6nNHpb/_buildManifest.js" defer=""></script><script src="/_next/static/h5xuILhXLuVXe-k6nNHpb/_ssgManifest.js" defer=""></script></head><body><link rel="preload" as="image" href="/img/tokio-horizontal.svg"><link rel="preload" as="image" href="/img/arrow-left.svg"><link rel="preload" as="image" href="/img/arrow-right.svg"><div id="__next"><main class="__className_6a21cc"><nav class="navbar is-spaced" role="navigation" aria-label="main navigation"><div class="container"><div class="navbar-brand"><a href="/" class="navbar-item"><img src="/img/tokio-horizontal.svg" alt="tokio-logo" width="133" height="56"></a><a role="button" class="navbar-burger" aria-label="menu" aria-expanded="false"><span aria-hidden="true"></span><span aria-hidden="true"></span><span aria-hidden="true"></span></a></div><div class="navbar-menu"><div class="navbar-end"><a href="/tokio/tutorial" class="navbar-item navbar-text">Learn</a><a href="https://docs.rs/tokio" class="navbar-item navbar-text">API Docs</a><a href="/blog/2025-09-26-announcing-tokio-conf-cfp" class="navbar-item navbar-text">Blog</a><hr class="is-hidden-mobile"><a target="_blank" class="navbar-item navbar-icon tk-social" href="https://twitter.com/tokio_rs"><span class="icon "><svg width="45" height="45" viewBox="0 0 1792 1792" xmlns="http://www.w3.org/2000/svg"><path class="tk-svg-path" d="m 1033.0757,785.67723 536.386,-610.09838 H 1442.3554 L 976.61152,705.31871 604.62327,175.57885 H 175.57884 L 738.0981,976.63931 175.57884,1616.4212 h 127.11352 l 491.83748,-559.4233 392.84686,559.4233 h 429.0444 L 1033.0445,785.67723 Z M 858.97635,983.69686 801.9814,903.9293 348.49316,269.21053 h 195.2389 l 365.97034,512.23831 56.9949,79.76754 475.7181,665.83142 H 1247.1765 L 858.97635,983.72736 Z" fill="#fff"></path></svg></span></a><a target="_blank" class="navbar-item navbar-icon tk-social" href="https://github.com/tokio-rs/tokio"><span class="icon "><svg width="45" height="45" viewBox="0 0 1792 1792" xmlns="http://www.w3.org/2000/svg"><path class="tk-svg-path" d="M896 128q209 0 385.5 103t279.5 279.5 103 385.5q0 251-146.5 451.5t-378.5 277.5q-27 5-40-7t-13-30q0-3 .5-76.5t.5-134.5q0-97-52-142 57-6 102.5-18t94-39 81-66.5 53-105 20.5-150.5q0-119-79-206 37-91-8-204-28-9-81 11t-92 44l-38 24q-93-26-192-26t-192 26q-16-11-42.5-27t-83.5-38.5-85-13.5q-45 113-8 204-79 87-79 206 0 85 20.5 150t52.5 105 80.5 67 94 39 102.5 18q-39 36-49 103-21 10-45 15t-57 5-65.5-21.5-55.5-62.5q-19-32-48.5-52t-49.5-24l-20-3q-21 0-29 4.5t-5 11.5 9 14 13 12l7 5q22 10 43.5 38t31.5 51l10 23q13 38 44 61.5t67 30 69.5 7 55.5-3.5l23-4q0 38 .5 88.5t.5 54.5q0 18-13 30t-40 7q-232-77-378.5-277.5t-146.5-451.5q0-209 103-385.5t279.5-279.5 385.5-103zm-477 1103q3-7-7-12-10-3-13 2-3 7 7 12 9 6 13-2zm31 34q7-5-2-16-10-9-16-3-7 5 2 16 10 10 16 3zm30 45q9-7 0-19-8-13-17-6-9 5 0 18t17 7zm42 42q8-8-4-19-12-12-20-3-9 8 4 19 12 12 20 3zm57 25q3-11-13-16-15-4-19 7t13 15q15 6 19-6zm63 5q0-13-17-11-16 0-16 11 0 13 17 11 16 0 16-11zm58-10q-2-11-18-9-16 3-14 15t18 8 14-14z" fill="#fff"></path></svg></span></a><a target="_blank" class="navbar-item navbar-icon tk-social" href="https://discord.gg/tokio"><span class="icon "><svg width="45" height="45" viewBox="0 0 245 240" xmlns="http://www.w3.org/2000/svg"><path class="tk-svg-path" fill="#FFF" d="M104.4 103.9c-5.7 0-10.2 5-10.2 11.1s4.6 11.1 10.2 11.1c5.7 0 10.2-5 10.2-11.1.1-6.1-4.5-11.1-10.2-11.1zM140.9 103.9c-5.7 0-10.2 5-10.2 11.1s4.6 11.1 10.2 11.1c5.7 0 10.2-5 10.2-11.1s-4.5-11.1-10.2-11.1z"></path><path class="tk-svg-path" fill="#FFF" d="M189.5 20h-134C44.2 20 35 29.2 35 40.6v135.2c0 11.4 9.2 20.6 20.5 20.6h113.4l-5.3-18.5 12.8 11.9 12.1 11.2 21.5 19V40.6c0-11.4-9.2-20.6-20.5-20.6zm-38.6 130.6s-3.6-4.3-6.6-8.1c13.1-3.7 18.1-11.9 18.1-11.9-4.1 2.7-8 4.6-11.5 5.9-5 2.1-9.8 3.5-14.5 4.3-9.6 1.8-18.4 1.3-25.9-.1-5.7-1.1-10.6-2.7-14.7-4.3-2.3-.9-4.8-2-7.3-3.4-.3-.2-.6-.3-.9-.5-.2-.1-.3-.2-.4-.3-1.8-1-2.8-1.7-2.8-1.7s4.8 8 17.5 11.8c-3 3.8-6.7 8.3-6.7 8.3-22.1-.7-30.5-15.2-30.5-15.2 0-32.2 14.4-58.3 14.4-58.3 14.4-10.8 28.1-10.5 28.1-10.5l1 1.2c-18 5.2-26.3 13.1-26.3 13.1s2.2-1.2 5.9-2.9c10.7-4.7 19.2-6 22.7-6.3.6-.1 1.1-.2 1.7-.2 6.1-.8 13-1 20.2-.2 9.5 1.1 19.7 3.9 30.1 9.6 0 0-7.9-7.5-24.9-12.7l1.4-1.6s13.7-.3 28.1 10.5c0 0 14.4 26.1 14.4 58.3 0 0-8.5 14.5-30.6 15.2z"></path></svg></span></a></div></div></div></nav><div class="columns is-marginless tk-docs"><div class="column is-one-quarter tk-docs-nav"><aside class="menu"><div class="tk-toc is-hidden-tablet"><a href="#">TABLE OF CONTENTS<span class="icon"><span class="tk-arrow"></span></span></a></div><div class="tk-menu-body"><p class="menu-label">Tokio</p><ul class="menu-list"><li class="is-active"><a href="/tokio/tutorial">Tutorial</a><ul><ul><li class=""><a href="/tokio/tutorial">Overview</a></li><li class=""><a href="/tokio/tutorial/setup">Setup</a></li><li class=""><a href="/tokio/tutorial/hello-tokio">Hello Tokio</a></li><li class=""><a href="/tokio/tutorial/spawning">Spawning</a></li><li class=""><a href="/tokio/tutorial/shared-state">Shared state</a></li><li class="is-active"><a href="/tokio/tutorial/channels">Channels</a></li><li class=""><a href="/tokio/tutorial/io">I/O</a></li><li class=""><a href="/tokio/tutorial/framing">Framing</a></li><li class=""><a href="/tokio/tutorial/async">Async in depth</a></li><li class=""><a href="/tokio/tutorial/select">Select</a></li><li class=""><a href="/tokio/tutorial/streams">Streams</a></li></ul></ul></li><li class=""><a href="/tokio/topics">Topics</a><ul><ul><li class=""><a href="/tokio/topics/bridging">Bridging with sync code</a></li><li class=""><a href="/tokio/topics/shutdown">Graceful Shutdown</a></li><li class=""><a href="/tokio/topics/tracing">Getting started with Tracing</a></li><li class=""><a href="/tokio/topics/tracing-next-steps">Next steps with Tracing</a></li><li class=""><a href="/tokio/topics/testing">Unit Testing</a></li></ul></ul></li><li class=""><a href="/tokio/glossary">Glossary</a></li><li class=""><a href="https://docs.rs/tokio">API documentation</a></li></ul></div><div class="is-hidden-mobile"></div></aside></div><div class="column is-three-quarters tk-content"><section class="section content"><div class="columns"><div class="column is-two-thirds tk-markdown"><h1 class="title " id="">Channels</h1><div><p>Now that we have learned a bit about concurrency with Tokio, let's apply this on
the client side. Put the server code we wrote before into an explicit binary
file:</p>
<pre><code class="hljs language-bash">$ <span class="hljs-built_in">mkdir</span> src/bin
$ <span class="hljs-built_in">mv</span> src/main.rs src/bin/server.rs
</code></pre>
<p>and create a new binary file that will contain the client code:</p>
<pre><code class="hljs language-bash">$ <span class="hljs-built_in">touch</span> src/bin/client.rs
</code></pre>
<p>In this file you will write this page's code. Whenever you want to run it,
you will have to launch the server first in a separate terminal window:</p>
<pre><code class="hljs language-bash">$ cargo run --bin server
</code></pre>
<p>And then the client, <strong>separately</strong>:</p>
<pre><code class="hljs language-bash">$ cargo run --bin client
</code></pre>
<p>That being said, let's code!</p>
<p>Say we want to run two concurrent Redis commands. We can spawn
one task per command. Then the two commands would happen concurrently.</p>
<p>At first, we might try something like:</p>
<pre><code class="hljs language-rust,compile_fail"><span class="hljs-keyword">use</span> mini_redis::client;
<span class="hljs-meta">#[tokio::main]</span>
<span class="hljs-keyword">async</span> <span class="hljs-keyword">fn</span> <span class="hljs-title function_">main</span>() {
<span class="hljs-comment">// Establish a connection to the server</span>
<span class="hljs-keyword">let</span> <span class="hljs-keyword">mut </span><span class="hljs-variable">client</span> = client::<span class="hljs-title function_ invoke__">connect</span>(<span class="hljs-string">"127.0.0.1:6379"</span>).<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
<span class="hljs-comment">// Spawn two tasks, one gets a key, the other sets a key</span>
<span class="hljs-keyword">let</span> <span class="hljs-variable">t1</span> = tokio::<span class="hljs-title function_ invoke__">spawn</span>(<span class="hljs-keyword">async</span> {
<span class="hljs-keyword">let</span> <span class="hljs-variable">res</span> = client.<span class="hljs-title function_ invoke__">get</span>(<span class="hljs-string">"foo"</span>).<span class="hljs-keyword">await</span>;
});
<span class="hljs-keyword">let</span> <span class="hljs-variable">t2</span> = tokio::<span class="hljs-title function_ invoke__">spawn</span>(<span class="hljs-keyword">async</span> {
client.<span class="hljs-title function_ invoke__">set</span>(<span class="hljs-string">"foo"</span>, <span class="hljs-string">"bar"</span>.<span class="hljs-title function_ invoke__">into</span>()).<span class="hljs-keyword">await</span>;
});
t1.<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
t2.<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
}
</code></pre>
<p>This does not compile because both tasks need to access the <code>client</code> somehow.
As <code>Client</code> does not implement <code>Copy</code>, it will not compile without some code to
facilitate this sharing. Additionally, <code>Client::set</code> takes <code>&amp;mut self</code>, which
means that exclusive access is required to call it. We could open a connection
per task, but that is not ideal. We cannot use <code>std::sync::Mutex</code> as <code>.await</code>
would need to be called with the lock held. We could use <code>tokio::sync::Mutex</code>,
but that would only allow a single in-flight request. If the client implements
<a href="https://redis.io/topics/pipelining">pipelining</a> (in short, sending many commands without waiting for each prior
command's response), an async mutex results in underutilizing
the connection.</p>
<h1 id="message-passing">Message passing</h1>
<p>The answer is to use message passing. The pattern involves spawning a dedicated
task to manage the <code>client</code> resource. Any task that wishes to issue a request
sends a message to the <code>client</code> task. The <code>client</code> task issues the request on
behalf of the sender, and the response is sent back to the sender.</p>
<p>Using this strategy, a single connection is established. The task managing the
<code>client</code> is able to get exclusive access in order to call <code>get</code> and <code>set</code>.
Additionally, the channel works as a buffer. Operations may be sent to the
<code>client</code> task while the <code>client</code> task is busy. Once the <code>client</code> task is
available to process new requests, it pulls the next request from the channel.
This can result in better throughput, and be extended to support connection
pooling.</p>
<h1 id="tokios-channel-primitives">Tokio's channel primitives</h1>
<p>Tokio provides a <a href="https://docs.rs/tokio/1/tokio/sync/index.html">number of channels</a>, each serving a different purpose.</p>
<ul>
<li><a href="https://docs.rs/tokio/1/tokio/sync/mpsc/index.html">mpsc</a>: multi-producer, single-consumer channel. Many values can be sent.</li>
<li><a href="https://docs.rs/tokio/1/tokio/sync/oneshot/index.html">oneshot</a>: single-producer, single consumer channel. A single value can be sent.</li>
<li><a href="https://docs.rs/tokio/1/tokio/sync/broadcast/index.html">broadcast</a>: multi-producer, multi-consumer. Many values can be sent. Each
receiver sees every value.</li>
<li><a href="https://docs.rs/tokio/1/tokio/sync/watch/index.html">watch</a>: multi-producer, multi-consumer. Many values can be sent, but no
history is kept. Receivers only see the most recent value.</li>
</ul>
<p>If you need a multi-producer multi-consumer channel where only one consumer sees
each message, you can use the <a href="https://docs.rs/async-channel/"><code>async-channel</code></a> crate. There are also channels
for use outside of asynchronous Rust, such as <a href="https://doc.rust-lang.org/stable/std/sync/mpsc/index.html"><code>std::sync::mpsc</code></a> and
<a href="https://docs.rs/crossbeam/latest/crossbeam/channel/index.html"><code>crossbeam::channel</code></a>. These channels wait for messages by blocking the
thread, which is not allowed in asynchronous code.</p>
<p>In this section, we will use <a href="https://docs.rs/tokio/1/tokio/sync/mpsc/index.html">mpsc</a> and <a href="https://docs.rs/tokio/1/tokio/sync/oneshot/index.html">oneshot</a>. The other types of message
passing channels are explored in later sections. The full code from this section
is found <a href="https://github.com/tokio-rs/website/blob/master/tutorial-code/channels/src/main.rs">here</a>.</p>
<h1 id="define-the-message-type">Define the message type</h1>
<p>In most cases, when using message passing, the task receiving the messages
responds to more than one command. In our case, the task will respond to <code>GET</code> and
<code>SET</code> commands. To model this, we first define a <code>Command</code> enum and include a
variant for each command type.</p>
<pre><code class="hljs language-rust"><span class="hljs-keyword">use</span> bytes::Bytes;
<span class="hljs-meta">#[derive(Debug)]</span>
<span class="hljs-keyword">enum</span> <span class="hljs-title class_">Command</span> {
Get {
key: <span class="hljs-type">String</span>,
},
Set {
key: <span class="hljs-type">String</span>,
val: Bytes,
}
}
</code></pre>
<h1 id="create-the-channel">Create the channel</h1>
<p>In the <code>main</code> function, an <code>mpsc</code> channel is created.</p>
<pre><code class="hljs language-rust"><span class="hljs-keyword">use</span> tokio::sync::mpsc;
<span class="hljs-meta">#[tokio::main]</span>
<span class="hljs-keyword">async</span> <span class="hljs-keyword">fn</span> <span class="hljs-title function_">main</span>() {
<span class="hljs-comment">// Create a new channel with a capacity of at most 32.</span>
<span class="hljs-keyword">let</span> (tx, <span class="hljs-keyword">mut</span> rx) = mpsc::<span class="hljs-title function_ invoke__">channel</span>(<span class="hljs-number">32</span>);
<span class="hljs-comment">// ... Rest comes here</span>
}
</code></pre>
<p>The <code>mpsc</code> channel is used to <strong>send</strong> commands to the task managing the redis
connection. The multi-producer capability allows messages to be sent from many
tasks. Creating the channel returns two values, a sender and a receiver. The two
handles are used separately. They may be moved to different tasks.</p>
<p>The channel is created with a capacity of 32. If messages are sent faster than
they are received, the channel will store them. Once the 32 messages are stored
in the channel, calling <code>send(...).await</code> will go to sleep until a message has
been removed by the receiver.</p>
<p>Sending from multiple tasks is done by <strong>cloning</strong> the <code>Sender</code>. For example:</p>
<pre><code class="hljs language-rust"><span class="hljs-keyword">use</span> tokio::sync::mpsc;
<span class="hljs-meta">#[tokio::main]</span>
<span class="hljs-keyword">async</span> <span class="hljs-keyword">fn</span> <span class="hljs-title function_">main</span>() {
<span class="hljs-keyword">let</span> (tx, <span class="hljs-keyword">mut</span> rx) = mpsc::<span class="hljs-title function_ invoke__">channel</span>(<span class="hljs-number">32</span>);
<span class="hljs-keyword">let</span> <span class="hljs-variable">tx2</span> = tx.<span class="hljs-title function_ invoke__">clone</span>();
tokio::<span class="hljs-title function_ invoke__">spawn</span>(<span class="hljs-keyword">async</span> <span class="hljs-keyword">move</span> {
tx.<span class="hljs-title function_ invoke__">send</span>(<span class="hljs-string">"sending from first handle"</span>).<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
});
tokio::<span class="hljs-title function_ invoke__">spawn</span>(<span class="hljs-keyword">async</span> <span class="hljs-keyword">move</span> {
tx2.<span class="hljs-title function_ invoke__">send</span>(<span class="hljs-string">"sending from second handle"</span>).<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
});
<span class="hljs-keyword">while</span> <span class="hljs-keyword">let</span> <span class="hljs-variable">Some</span>(message) = rx.<span class="hljs-title function_ invoke__">recv</span>().<span class="hljs-keyword">await</span> {
<span class="hljs-built_in">println!</span>(<span class="hljs-string">"GOT = {}"</span>, message);
}
}
</code></pre>
<p>Both messages are sent to the single <code>Receiver</code> handle. It is not possible to
clone the receiver of an <code>mpsc</code> channel.</p>
<p>When every <code>Sender</code> has gone out of scope or has otherwise been dropped, it is
no longer possible to send more messages into the channel. At this point, the
<code>recv</code> call on the <code>Receiver</code> will return <code>None</code>, which means that all senders
are gone and the channel is closed.</p>
<p>In our case of a task that manages the Redis connection, it knows that it
can close the Redis connection once the channel is closed, as the connection
will not be used again.</p>
<h1 id="spawn-manager-task">Spawn manager task</h1>
<p>Next, spawn a task that processes messages from the channel. First, a client
connection is established to Redis. Then, received commands are issued via the
Redis connection.</p>
<pre><code class="hljs language-rust"><span class="hljs-keyword">use</span> mini_redis::client;
<span class="hljs-comment">// The `move` keyword is used to **move** ownership of `rx` into the task.</span>
<span class="hljs-keyword">let</span> <span class="hljs-variable">manager</span> = tokio::<span class="hljs-title function_ invoke__">spawn</span>(<span class="hljs-keyword">async</span> <span class="hljs-keyword">move</span> {
<span class="hljs-comment">// Establish a connection to the server</span>
<span class="hljs-keyword">let</span> <span class="hljs-keyword">mut </span><span class="hljs-variable">client</span> = client::<span class="hljs-title function_ invoke__">connect</span>(<span class="hljs-string">"127.0.0.1:6379"</span>).<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
<span class="hljs-comment">// Start receiving messages</span>
<span class="hljs-keyword">while</span> <span class="hljs-keyword">let</span> <span class="hljs-variable">Some</span>(cmd) = rx.<span class="hljs-title function_ invoke__">recv</span>().<span class="hljs-keyword">await</span> {
<span class="hljs-keyword">use</span> Command::*;
<span class="hljs-keyword">match</span> cmd {
Get { key } =&gt; {
client.<span class="hljs-title function_ invoke__">get</span>(&amp;key).<span class="hljs-keyword">await</span>;
}
Set { key, val } =&gt; {
client.<span class="hljs-title function_ invoke__">set</span>(&amp;key, val).<span class="hljs-keyword">await</span>;
}
}
}
});
</code></pre>
<p>Now, update the two tasks to send commands over the channel instead of issuing
them directly on the Redis connection.</p>
<pre><code class="hljs language-rust"><span class="hljs-comment">// The `Sender` handles are moved into the tasks. As there are two</span>
<span class="hljs-comment">// tasks, we need a second `Sender`.</span>
<span class="hljs-keyword">let</span> <span class="hljs-variable">tx2</span> = tx.<span class="hljs-title function_ invoke__">clone</span>();
<span class="hljs-comment">// Spawn two tasks, one gets a key, the other sets a key</span>
<span class="hljs-keyword">let</span> <span class="hljs-variable">t1</span> = tokio::<span class="hljs-title function_ invoke__">spawn</span>(<span class="hljs-keyword">async</span> <span class="hljs-keyword">move</span> {
<span class="hljs-keyword">let</span> <span class="hljs-variable">cmd</span> = Command::Get {
key: <span class="hljs-string">"foo"</span>.<span class="hljs-title function_ invoke__">to_string</span>(),
};
tx.<span class="hljs-title function_ invoke__">send</span>(cmd).<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
});
<span class="hljs-keyword">let</span> <span class="hljs-variable">t2</span> = tokio::<span class="hljs-title function_ invoke__">spawn</span>(<span class="hljs-keyword">async</span> <span class="hljs-keyword">move</span> {
<span class="hljs-keyword">let</span> <span class="hljs-variable">cmd</span> = Command::Set {
key: <span class="hljs-string">"foo"</span>.<span class="hljs-title function_ invoke__">to_string</span>(),
val: <span class="hljs-string">"bar"</span>.<span class="hljs-title function_ invoke__">into</span>(),
};
tx2.<span class="hljs-title function_ invoke__">send</span>(cmd).<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
});
</code></pre>
<p>At the bottom of the <code>main</code> function, we <code>.await</code> the join handles to ensure the
commands fully complete before the process exits.</p>
<pre><code class="hljs language-rust">t1.<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
t2.<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
manager.<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
</code></pre>
<h1 id="receive-responses">Receive responses</h1>
<p>The final step is to receive the response back from the manager task. The <code>GET</code>
command needs to get the value and the <code>SET</code> command needs to know if the
operation completed successfully.</p>
<p>To pass the response, a <code>oneshot</code> channel is used. The <code>oneshot</code> channel is a
single-producer, single-consumer channel optimized for sending a single value.
In our case, the single value is the response.</p>
<p>Similar to <code>mpsc</code>, <code>oneshot::channel()</code> returns a sender and receiver handle.</p>
<pre><code class="hljs language-rust"><span class="hljs-keyword">use</span> tokio::sync::oneshot;
<span class="hljs-keyword">let</span> (tx, rx) = oneshot::<span class="hljs-title function_ invoke__">channel</span>();
</code></pre>
<p>Unlike <code>mpsc</code>, no capacity is specified as the capacity is always one.
Additionally, neither handle can be cloned.</p>
<p>To receive responses from the manager task, before sending a command, a <code>oneshot</code>
channel is created. The <code>Sender</code> half of the channel is included in the command
to the manager task. The receive half is used to receive the response.</p>
<p>First, update <code>Command</code> to include the <code>Sender</code>. For convenience, a type alias
is used to reference the <code>Sender</code>.</p>
<pre><code class="hljs language-rust"><span class="hljs-keyword">use</span> tokio::sync::oneshot;
<span class="hljs-keyword">use</span> bytes::Bytes;
<span class="hljs-comment">/// Multiple different commands are multiplexed over a single channel.</span>
<span class="hljs-meta">#[derive(Debug)]</span>
<span class="hljs-keyword">enum</span> <span class="hljs-title class_">Command</span> {
Get {
key: <span class="hljs-type">String</span>,
resp: Responder&lt;<span class="hljs-type">Option</span>&lt;Bytes&gt;&gt;,
},
Set {
key: <span class="hljs-type">String</span>,
val: Bytes,
resp: Responder&lt;()&gt;,
},
}
<span class="hljs-comment">/// Provided by the requester and used by the manager task to send</span>
<span class="hljs-comment">/// the command response back to the requester.</span>
<span class="hljs-keyword">type</span> <span class="hljs-title class_">Responder</span>&lt;T&gt; = oneshot::Sender&lt;mini_redis::<span class="hljs-type">Result</span>&lt;T&gt;&gt;;
</code></pre>
<p>Now, update the tasks issuing the commands to include the <code>oneshot::Sender</code>.</p>
<pre><code class="hljs language-rust"><span class="hljs-keyword">let</span> <span class="hljs-variable">t1</span> = tokio::<span class="hljs-title function_ invoke__">spawn</span>(<span class="hljs-keyword">async</span> <span class="hljs-keyword">move</span> {
<span class="hljs-keyword">let</span> (resp_tx, resp_rx) = oneshot::<span class="hljs-title function_ invoke__">channel</span>();
<span class="hljs-keyword">let</span> <span class="hljs-variable">cmd</span> = Command::Get {
key: <span class="hljs-string">"foo"</span>.<span class="hljs-title function_ invoke__">to_string</span>(),
resp: resp_tx,
};
<span class="hljs-comment">// Send the GET request</span>
tx.<span class="hljs-title function_ invoke__">send</span>(cmd).<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
<span class="hljs-comment">// Await the response</span>
<span class="hljs-keyword">let</span> <span class="hljs-variable">res</span> = resp_rx.<span class="hljs-keyword">await</span>;
<span class="hljs-built_in">println!</span>(<span class="hljs-string">"GOT = {:?}"</span>, res);
});
<span class="hljs-keyword">let</span> <span class="hljs-variable">t2</span> = tokio::<span class="hljs-title function_ invoke__">spawn</span>(<span class="hljs-keyword">async</span> <span class="hljs-keyword">move</span> {
<span class="hljs-keyword">let</span> (resp_tx, resp_rx) = oneshot::<span class="hljs-title function_ invoke__">channel</span>();
<span class="hljs-keyword">let</span> <span class="hljs-variable">cmd</span> = Command::Set {
key: <span class="hljs-string">"foo"</span>.<span class="hljs-title function_ invoke__">to_string</span>(),
val: <span class="hljs-string">"bar"</span>.<span class="hljs-title function_ invoke__">into</span>(),
resp: resp_tx,
};
<span class="hljs-comment">// Send the SET request</span>
tx2.<span class="hljs-title function_ invoke__">send</span>(cmd).<span class="hljs-keyword">await</span>.<span class="hljs-title function_ invoke__">unwrap</span>();
<span class="hljs-comment">// Await the response</span>
<span class="hljs-keyword">let</span> <span class="hljs-variable">res</span> = resp_rx.<span class="hljs-keyword">await</span>;
<span class="hljs-built_in">println!</span>(<span class="hljs-string">"GOT = {:?}"</span>, res);
});
</code></pre>
<p>Finally, update the manager task to send the response over the <code>oneshot</code> channel.</p>
<pre><code class="hljs language-rust"><span class="hljs-keyword">while</span> <span class="hljs-keyword">let</span> <span class="hljs-variable">Some</span>(cmd) = rx.<span class="hljs-title function_ invoke__">recv</span>().<span class="hljs-keyword">await</span> {
<span class="hljs-keyword">match</span> cmd {
Command::Get { key, resp } =&gt; {
<span class="hljs-keyword">let</span> <span class="hljs-variable">res</span> = client.<span class="hljs-title function_ invoke__">get</span>(&amp;key).<span class="hljs-keyword">await</span>;
<span class="hljs-comment">// Ignore errors</span>
<span class="hljs-keyword">let</span> <span class="hljs-variable">_</span> = resp.<span class="hljs-title function_ invoke__">send</span>(res);
}
Command::Set { key, val, resp } =&gt; {
<span class="hljs-keyword">let</span> <span class="hljs-variable">res</span> = client.<span class="hljs-title function_ invoke__">set</span>(&amp;key, val).<span class="hljs-keyword">await</span>;
<span class="hljs-comment">// Ignore errors</span>
<span class="hljs-keyword">let</span> <span class="hljs-variable">_</span> = resp.<span class="hljs-title function_ invoke__">send</span>(res);
}
}
}
</code></pre>
<p>Calling <code>send</code> on <code>oneshot::Sender</code> completes immediately and does <strong>not</strong>
require an <code>.await</code>. This is because <code>send</code> on a <code>oneshot</code> channel will always
fail or succeed immediately without any form of waiting.</p>
<p>Sending a value on a oneshot channel returns <code>Err</code> when the receiver half has
dropped. This indicates the receiver is no longer interested in the response. In
our scenario, the receiver cancelling interest is an acceptable event. The <code>Err</code>
returned by <code>resp.send(...)</code> does not need to be handled.</p>
<p>You can find the entire code <a href="https://github.com/tokio-rs/website/blob/master/tutorial-code/channels/src/main.rs">here</a>.</p>
<h1 id="backpressure-and-bounded-channels">Backpressure and bounded channels</h1>
<p>Whenever concurrency or queuing is introduced, it is important to ensure that the
queing is bounded and the system will gracefully handle the load. Unbounded queues
will eventually fill up all available memory and cause the system to fail in
unpredictable ways.</p>
<p>Tokio takes care to avoid implicit queuing. A big part of this is the fact that
async operations are lazy. Consider the following:</p>
<pre><code class="hljs language-rust"><span class="hljs-keyword">loop</span> {
<span class="hljs-title function_ invoke__">async_op</span>();
}
</code></pre>
<p>If the asynchronous operation runs eagerly, the loop will repeatedly queue a new
<code>async_op</code> to run without ensuring the previous operation completed. This
results in implicit unbounded queuing. Callback based systems and <strong>eager</strong>
future based systems are particularly susceptible to this.</p>
<p>However, with Tokio and asynchronous Rust, the above snippet will <strong>not</strong> result
in <code>async_op</code> running at all. This is because <code>.await</code> is never called. If the
snippet is updated to use <code>.await</code>, then the loop waits for the operation to
complete before starting over.</p>
<pre><code class="hljs language-rust"><span class="hljs-keyword">loop</span> {
<span class="hljs-comment">// Will not repeat until `async_op` completes</span>
<span class="hljs-title function_ invoke__">async_op</span>().<span class="hljs-keyword">await</span>;
}
</code></pre>
<p>Concurrency and queuing must be explicitly introduced. Ways to do this include:</p>
<ul>
<li><code>tokio::spawn</code></li>
<li><code>select!</code></li>
<li><code>join!</code></li>
<li><code>mpsc::channel</code></li>
</ul>
<p>When doing so, take care to ensure the total amount of concurrency is bounded. For
example, when writing a TCP accept loop, ensure that the total number of open
sockets is bounded. When using <code>mpsc::channel</code>, pick a manageable channel
capacity. Specific bound values will be application specific.</p>
<p>Taking care and picking good bounds is a big part of writing reliable Tokio applications.</p></div><div class="tk-doc-footer"><div class="level"><div class="level-left"><div class="level-item tk-prev"><a href="/tokio/tutorial/shared-state" rel="prev"><span class="tk-arrow" style="margin-right:0.5rem"><img src="/img/arrow-left.svg"></span>Shared state</a></div></div><div class="level-right"><div class="level-item tk-next"><a href="/tokio/tutorial/io" rel="next">I/O<span class="tk-arrow" style="margin-left:0.5rem"><img src="/img/arrow-right.svg"></span></a></div></div></div><div class="level"><div class="level-left"><div class="level-item tk-help-links"><p>Get Help:<a href="https://github.com/tokio-rs/tokio/discussions"><span class="icon is-medium"><svg width="45" height="45" viewBox="0 0 1792 1792" xmlns="http://www.w3.org/2000/svg"><path class="tk-svg-path" d="M896 128q209 0 385.5 103t279.5 279.5 103 385.5q0 251-146.5 451.5t-378.5 277.5q-27 5-40-7t-13-30q0-3 .5-76.5t.5-134.5q0-97-52-142 57-6 102.5-18t94-39 81-66.5 53-105 20.5-150.5q0-119-79-206 37-91-8-204-28-9-81 11t-92 44l-38 24q-93-26-192-26t-192 26q-16-11-42.5-27t-83.5-38.5-85-13.5q-45 113-8 204-79 87-79 206 0 85 20.5 150t52.5 105 80.5 67 94 39 102.5 18q-39 36-49 103-21 10-45 15t-57 5-65.5-21.5-55.5-62.5q-19-32-48.5-52t-49.5-24l-20-3q-21 0-29 4.5t-5 11.5 9 14 13 12l7 5q22 10 43.5 38t31.5 51l10 23q13 38 44 61.5t67 30 69.5 7 55.5-3.5l23-4q0 38 .5 88.5t.5 54.5q0 18-13 30t-40 7q-232-77-378.5-277.5t-146.5-451.5q0-209 103-385.5t279.5-279.5 385.5-103zm-477 1103q3-7-7-12-10-3-13 2-3 7 7 12 9 6 13-2zm31 34q7-5-2-16-10-9-16-3-7 5 2 16 10 10 16 3zm30 45q9-7 0-19-8-13-17-6-9 5 0 18t17 7zm42 42q8-8-4-19-12-12-20-3-9 8 4 19 12 12 20 3zm57 25q3-11-13-16-15-4-19 7t13 15q15 6 19-6zm63 5q0-13-17-11-16 0-16 11 0 13 17 11 16 0 16-11zm58-10q-2-11-18-9-16 3-14 15t18 8 14-14z" fill="#fff"></path></svg></span></a><a href="https://discord.gg/tokio"><span class="icon is-medium"><svg width="45" height="45" viewBox="0 0 245 240" xmlns="http://www.w3.org/2000/svg"><path class="tk-svg-path" fill="#FFF" d="M104.4 103.9c-5.7 0-10.2 5-10.2 11.1s4.6 11.1 10.2 11.1c5.7 0 10.2-5 10.2-11.1.1-6.1-4.5-11.1-10.2-11.1zM140.9 103.9c-5.7 0-10.2 5-10.2 11.1s4.6 11.1 10.2 11.1c5.7 0 10.2-5 10.2-11.1s-4.5-11.1-10.2-11.1z"></path><path class="tk-svg-path" fill="#FFF" d="M189.5 20h-134C44.2 20 35 29.2 35 40.6v135.2c0 11.4 9.2 20.6 20.5 20.6h113.4l-5.3-18.5 12.8 11.9 12.1 11.2 21.5 19V40.6c0-11.4-9.2-20.6-20.5-20.6zm-38.6 130.6s-3.6-4.3-6.6-8.1c13.1-3.7 18.1-11.9 18.1-11.9-4.1 2.7-8 4.6-11.5 5.9-5 2.1-9.8 3.5-14.5 4.3-9.6 1.8-18.4 1.3-25.9-.1-5.7-1.1-10.6-2.7-14.7-4.3-2.3-.9-4.8-2-7.3-3.4-.3-.2-.6-.3-.9-.5-.2-.1-.3-.2-.4-.3-1.8-1-2.8-1.7-2.8-1.7s4.8 8 17.5 11.8c-3 3.8-6.7 8.3-6.7 8.3-22.1-.7-30.5-15.2-30.5-15.2 0-32.2 14.4-58.3 14.4-58.3 14.4-10.8 28.1-10.5 28.1-10.5l1 1.2c-18 5.2-26.3 13.1-26.3 13.1s2.2-1.2 5.9-2.9c10.7-4.7 19.2-6 22.7-6.3.6-.1 1.1-.2 1.7-.2 6.1-.8 13-1 20.2-.2 9.5 1.1 19.7 3.9 30.1 9.6 0 0-7.9-7.5-24.9-12.7l1.4-1.6s13.7-.3 28.1 10.5c0 0 14.4 26.1 14.4 58.3 0 0-8.5 14.5-30.6 15.2z"></path></svg></span></a></p></div></div><div class="level-right"><div class="level-item tk-edit-this-page"><a href="https://github.com/tokio-rs/website/edit/master/content/tokio/tutorial/channels.md">Edit this page</a></div></div></div></div></div><aside class="column is-one-third tk-content-summary"><ul class="tk-content-summary-menu"><li><a href="#">Channels</a></li><li><a href="#message-passing">Message passing</a></li><li><a href="#tokios-channel-primitives">Tokio's channel primitives</a></li><li><a href="#define-the-message-type">Define the message type</a></li><li><a href="#create-the-channel">Create the channel</a></li><li><a href="#spawn-manager-task">Spawn manager task</a></li><li><a href="#receive-responses">Receive responses</a></li><li><a href="#backpressure-and-bounded-channels">Backpressure and bounded channels</a></li></ul></aside></div></section></div></div></main></div><script id="__NEXT_DATA__" type="application/json">{"props":{"pageProps":{"page":{"key":"channels","href":"/tokio/tutorial/channels","title":"Channels","menuTitle":"Channels","mdPath":"tokio/tutorial/channels.md","data":{"title":"Channels"},"body":"\u003cp\u003eNow that we have learned a bit about concurrency with Tokio, let's apply this on\nthe client side. Put the server code we wrote before into an explicit binary\nfile:\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-bash\"\u003e$ \u003cspan class=\"hljs-built_in\"\u003emkdir\u003c/span\u003e src/bin\n$ \u003cspan class=\"hljs-built_in\"\u003emv\u003c/span\u003e src/main.rs src/bin/server.rs\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eand create a new binary file that will contain the client code:\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-bash\"\u003e$ \u003cspan class=\"hljs-built_in\"\u003etouch\u003c/span\u003e src/bin/client.rs\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eIn this file you will write this page's code. Whenever you want to run it,\nyou will have to launch the server first in a separate terminal window:\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-bash\"\u003e$ cargo run --bin server\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eAnd then the client, \u003cstrong\u003eseparately\u003c/strong\u003e:\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-bash\"\u003e$ cargo run --bin client\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eThat being said, let's code!\u003c/p\u003e\n\u003cp\u003eSay we want to run two concurrent Redis commands. We can spawn\none task per command. Then the two commands would happen concurrently.\u003c/p\u003e\n\u003cp\u003eAt first, we might try something like:\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust,compile_fail\"\u003e\u003cspan class=\"hljs-keyword\"\u003euse\u003c/span\u003e mini_redis::client;\n\n\u003cspan class=\"hljs-meta\"\u003e#[tokio::main]\u003c/span\u003e\n\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003efn\u003c/span\u003e \u003cspan class=\"hljs-title function_\"\u003emain\u003c/span\u003e() {\n \u003cspan class=\"hljs-comment\"\u003e// Establish a connection to the server\u003c/span\u003e\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003emut \u003c/span\u003e\u003cspan class=\"hljs-variable\"\u003eclient\u003c/span\u003e = client::\u003cspan class=\"hljs-title function_ invoke__\"\u003econnect\u003c/span\u003e(\u003cspan class=\"hljs-string\"\u003e\"127.0.0.1:6379\"\u003c/span\u003e).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\n\n \u003cspan class=\"hljs-comment\"\u003e// Spawn two tasks, one gets a key, the other sets a key\u003c/span\u003e\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003et1\u003c/span\u003e = tokio::\u003cspan class=\"hljs-title function_ invoke__\"\u003espawn\u003c/span\u003e(\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e {\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003eres\u003c/span\u003e = client.\u003cspan class=\"hljs-title function_ invoke__\"\u003eget\u003c/span\u003e(\u003cspan class=\"hljs-string\"\u003e\"foo\"\u003c/span\u003e).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e;\n });\n\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003et2\u003c/span\u003e = tokio::\u003cspan class=\"hljs-title function_ invoke__\"\u003espawn\u003c/span\u003e(\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e {\n client.\u003cspan class=\"hljs-title function_ invoke__\"\u003eset\u003c/span\u003e(\u003cspan class=\"hljs-string\"\u003e\"foo\"\u003c/span\u003e, \u003cspan class=\"hljs-string\"\u003e\"bar\"\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003einto\u003c/span\u003e()).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e;\n });\n\n t1.\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\n t2.\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\n}\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eThis does not compile because both tasks need to access the \u003ccode\u003eclient\u003c/code\u003e somehow.\nAs \u003ccode\u003eClient\u003c/code\u003e does not implement \u003ccode\u003eCopy\u003c/code\u003e, it will not compile without some code to\nfacilitate this sharing. Additionally, \u003ccode\u003eClient::set\u003c/code\u003e takes \u003ccode\u003e\u0026#x26;mut self\u003c/code\u003e, which\nmeans that exclusive access is required to call it. We could open a connection\nper task, but that is not ideal. We cannot use \u003ccode\u003estd::sync::Mutex\u003c/code\u003e as \u003ccode\u003e.await\u003c/code\u003e\nwould need to be called with the lock held. We could use \u003ccode\u003etokio::sync::Mutex\u003c/code\u003e,\nbut that would only allow a single in-flight request. If the client implements\n\u003ca href=\"https://redis.io/topics/pipelining\"\u003epipelining\u003c/a\u003e (in short, sending many commands without waiting for each prior\ncommand's response), an async mutex results in underutilizing\nthe connection.\u003c/p\u003e\n\u003ch1 id=\"message-passing\"\u003eMessage passing\u003c/h1\u003e\n\u003cp\u003eThe answer is to use message passing. The pattern involves spawning a dedicated\ntask to manage the \u003ccode\u003eclient\u003c/code\u003e resource. Any task that wishes to issue a request\nsends a message to the \u003ccode\u003eclient\u003c/code\u003e task. The \u003ccode\u003eclient\u003c/code\u003e task issues the request on\nbehalf of the sender, and the response is sent back to the sender.\u003c/p\u003e\n\u003cp\u003eUsing this strategy, a single connection is established. The task managing the\n\u003ccode\u003eclient\u003c/code\u003e is able to get exclusive access in order to call \u003ccode\u003eget\u003c/code\u003e and \u003ccode\u003eset\u003c/code\u003e.\nAdditionally, the channel works as a buffer. Operations may be sent to the\n\u003ccode\u003eclient\u003c/code\u003e task while the \u003ccode\u003eclient\u003c/code\u003e task is busy. Once the \u003ccode\u003eclient\u003c/code\u003e task is\navailable to process new requests, it pulls the next request from the channel.\nThis can result in better throughput, and be extended to support connection\npooling.\u003c/p\u003e\n\u003ch1 id=\"tokios-channel-primitives\"\u003eTokio's channel primitives\u003c/h1\u003e\n\u003cp\u003eTokio provides a \u003ca href=\"https://docs.rs/tokio/1/tokio/sync/index.html\"\u003enumber of channels\u003c/a\u003e, each serving a different purpose.\u003c/p\u003e\n\u003cul\u003e\n\u003cli\u003e\u003ca href=\"https://docs.rs/tokio/1/tokio/sync/mpsc/index.html\"\u003empsc\u003c/a\u003e: multi-producer, single-consumer channel. Many values can be sent.\u003c/li\u003e\n\u003cli\u003e\u003ca href=\"https://docs.rs/tokio/1/tokio/sync/oneshot/index.html\"\u003eoneshot\u003c/a\u003e: single-producer, single consumer channel. A single value can be sent.\u003c/li\u003e\n\u003cli\u003e\u003ca href=\"https://docs.rs/tokio/1/tokio/sync/broadcast/index.html\"\u003ebroadcast\u003c/a\u003e: multi-producer, multi-consumer. Many values can be sent. Each\nreceiver sees every value.\u003c/li\u003e\n\u003cli\u003e\u003ca href=\"https://docs.rs/tokio/1/tokio/sync/watch/index.html\"\u003ewatch\u003c/a\u003e: multi-producer, multi-consumer. Many values can be sent, but no\nhistory is kept. Receivers only see the most recent value.\u003c/li\u003e\n\u003c/ul\u003e\n\u003cp\u003eIf you need a multi-producer multi-consumer channel where only one consumer sees\neach message, you can use the \u003ca href=\"https://docs.rs/async-channel/\"\u003e\u003ccode\u003easync-channel\u003c/code\u003e\u003c/a\u003e crate. There are also channels\nfor use outside of asynchronous Rust, such as \u003ca href=\"https://doc.rust-lang.org/stable/std/sync/mpsc/index.html\"\u003e\u003ccode\u003estd::sync::mpsc\u003c/code\u003e\u003c/a\u003e and\n\u003ca href=\"https://docs.rs/crossbeam/latest/crossbeam/channel/index.html\"\u003e\u003ccode\u003ecrossbeam::channel\u003c/code\u003e\u003c/a\u003e. These channels wait for messages by blocking the\nthread, which is not allowed in asynchronous code.\u003c/p\u003e\n\u003cp\u003eIn this section, we will use \u003ca href=\"https://docs.rs/tokio/1/tokio/sync/mpsc/index.html\"\u003empsc\u003c/a\u003e and \u003ca href=\"https://docs.rs/tokio/1/tokio/sync/oneshot/index.html\"\u003eoneshot\u003c/a\u003e. The other types of message\npassing channels are explored in later sections. The full code from this section\nis found \u003ca href=\"https://github.com/tokio-rs/website/blob/master/tutorial-code/channels/src/main.rs\"\u003ehere\u003c/a\u003e.\u003c/p\u003e\n\u003ch1 id=\"define-the-message-type\"\u003eDefine the message type\u003c/h1\u003e\n\u003cp\u003eIn most cases, when using message passing, the task receiving the messages\nresponds to more than one command. In our case, the task will respond to \u003ccode\u003eGET\u003c/code\u003e and\n\u003ccode\u003eSET\u003c/code\u003e commands. To model this, we first define a \u003ccode\u003eCommand\u003c/code\u003e enum and include a\nvariant for each command type.\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003e\u003cspan class=\"hljs-keyword\"\u003euse\u003c/span\u003e bytes::Bytes;\n\n\u003cspan class=\"hljs-meta\"\u003e#[derive(Debug)]\u003c/span\u003e\n\u003cspan class=\"hljs-keyword\"\u003eenum\u003c/span\u003e \u003cspan class=\"hljs-title class_\"\u003eCommand\u003c/span\u003e {\n Get {\n key: \u003cspan class=\"hljs-type\"\u003eString\u003c/span\u003e,\n },\n Set {\n key: \u003cspan class=\"hljs-type\"\u003eString\u003c/span\u003e,\n val: Bytes,\n }\n}\n\u003c/code\u003e\u003c/pre\u003e\n\u003ch1 id=\"create-the-channel\"\u003eCreate the channel\u003c/h1\u003e\n\u003cp\u003eIn the \u003ccode\u003emain\u003c/code\u003e function, an \u003ccode\u003empsc\u003c/code\u003e channel is created.\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003e\u003cspan class=\"hljs-keyword\"\u003euse\u003c/span\u003e tokio::sync::mpsc;\n\n\u003cspan class=\"hljs-meta\"\u003e#[tokio::main]\u003c/span\u003e\n\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003efn\u003c/span\u003e \u003cspan class=\"hljs-title function_\"\u003emain\u003c/span\u003e() {\n \u003cspan class=\"hljs-comment\"\u003e// Create a new channel with a capacity of at most 32.\u003c/span\u003e\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e (tx, \u003cspan class=\"hljs-keyword\"\u003emut\u003c/span\u003e rx) = mpsc::\u003cspan class=\"hljs-title function_ invoke__\"\u003echannel\u003c/span\u003e(\u003cspan class=\"hljs-number\"\u003e32\u003c/span\u003e);\n\n \u003cspan class=\"hljs-comment\"\u003e// ... Rest comes here\u003c/span\u003e\n}\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eThe \u003ccode\u003empsc\u003c/code\u003e channel is used to \u003cstrong\u003esend\u003c/strong\u003e commands to the task managing the redis\nconnection. The multi-producer capability allows messages to be sent from many\ntasks. Creating the channel returns two values, a sender and a receiver. The two\nhandles are used separately. They may be moved to different tasks.\u003c/p\u003e\n\u003cp\u003eThe channel is created with a capacity of 32. If messages are sent faster than\nthey are received, the channel will store them. Once the 32 messages are stored\nin the channel, calling \u003ccode\u003esend(...).await\u003c/code\u003e will go to sleep until a message has\nbeen removed by the receiver.\u003c/p\u003e\n\u003cp\u003eSending from multiple tasks is done by \u003cstrong\u003ecloning\u003c/strong\u003e the \u003ccode\u003eSender\u003c/code\u003e. For example:\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003e\u003cspan class=\"hljs-keyword\"\u003euse\u003c/span\u003e tokio::sync::mpsc;\n\n\u003cspan class=\"hljs-meta\"\u003e#[tokio::main]\u003c/span\u003e\n\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003efn\u003c/span\u003e \u003cspan class=\"hljs-title function_\"\u003emain\u003c/span\u003e() {\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e (tx, \u003cspan class=\"hljs-keyword\"\u003emut\u003c/span\u003e rx) = mpsc::\u003cspan class=\"hljs-title function_ invoke__\"\u003echannel\u003c/span\u003e(\u003cspan class=\"hljs-number\"\u003e32\u003c/span\u003e);\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003etx2\u003c/span\u003e = tx.\u003cspan class=\"hljs-title function_ invoke__\"\u003eclone\u003c/span\u003e();\n\n tokio::\u003cspan class=\"hljs-title function_ invoke__\"\u003espawn\u003c/span\u003e(\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003emove\u003c/span\u003e {\n tx.\u003cspan class=\"hljs-title function_ invoke__\"\u003esend\u003c/span\u003e(\u003cspan class=\"hljs-string\"\u003e\"sending from first handle\"\u003c/span\u003e).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\n });\n\n tokio::\u003cspan class=\"hljs-title function_ invoke__\"\u003espawn\u003c/span\u003e(\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003emove\u003c/span\u003e {\n tx2.\u003cspan class=\"hljs-title function_ invoke__\"\u003esend\u003c/span\u003e(\u003cspan class=\"hljs-string\"\u003e\"sending from second handle\"\u003c/span\u003e).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\n });\n\n \u003cspan class=\"hljs-keyword\"\u003ewhile\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003eSome\u003c/span\u003e(message) = rx.\u003cspan class=\"hljs-title function_ invoke__\"\u003erecv\u003c/span\u003e().\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e {\n \u003cspan class=\"hljs-built_in\"\u003eprintln!\u003c/span\u003e(\u003cspan class=\"hljs-string\"\u003e\"GOT = {}\"\u003c/span\u003e, message);\n }\n}\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eBoth messages are sent to the single \u003ccode\u003eReceiver\u003c/code\u003e handle. It is not possible to\nclone the receiver of an \u003ccode\u003empsc\u003c/code\u003e channel.\u003c/p\u003e\n\u003cp\u003eWhen every \u003ccode\u003eSender\u003c/code\u003e has gone out of scope or has otherwise been dropped, it is\nno longer possible to send more messages into the channel. At this point, the\n\u003ccode\u003erecv\u003c/code\u003e call on the \u003ccode\u003eReceiver\u003c/code\u003e will return \u003ccode\u003eNone\u003c/code\u003e, which means that all senders\nare gone and the channel is closed.\u003c/p\u003e\n\u003cp\u003eIn our case of a task that manages the Redis connection, it knows that it\ncan close the Redis connection once the channel is closed, as the connection\nwill not be used again.\u003c/p\u003e\n\u003ch1 id=\"spawn-manager-task\"\u003eSpawn manager task\u003c/h1\u003e\n\u003cp\u003eNext, spawn a task that processes messages from the channel. First, a client\nconnection is established to Redis. Then, received commands are issued via the\nRedis connection.\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003e\u003cspan class=\"hljs-keyword\"\u003euse\u003c/span\u003e mini_redis::client;\n\u003cspan class=\"hljs-comment\"\u003e// The `move` keyword is used to **move** ownership of `rx` into the task.\u003c/span\u003e\n\u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003emanager\u003c/span\u003e = tokio::\u003cspan class=\"hljs-title function_ invoke__\"\u003espawn\u003c/span\u003e(\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003emove\u003c/span\u003e {\n \u003cspan class=\"hljs-comment\"\u003e// Establish a connection to the server\u003c/span\u003e\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003emut \u003c/span\u003e\u003cspan class=\"hljs-variable\"\u003eclient\u003c/span\u003e = client::\u003cspan class=\"hljs-title function_ invoke__\"\u003econnect\u003c/span\u003e(\u003cspan class=\"hljs-string\"\u003e\"127.0.0.1:6379\"\u003c/span\u003e).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\n\n \u003cspan class=\"hljs-comment\"\u003e// Start receiving messages\u003c/span\u003e\n \u003cspan class=\"hljs-keyword\"\u003ewhile\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003eSome\u003c/span\u003e(cmd) = rx.\u003cspan class=\"hljs-title function_ invoke__\"\u003erecv\u003c/span\u003e().\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e {\n \u003cspan class=\"hljs-keyword\"\u003euse\u003c/span\u003e Command::*;\n\n \u003cspan class=\"hljs-keyword\"\u003ematch\u003c/span\u003e cmd {\n Get { key } =\u003e {\n client.\u003cspan class=\"hljs-title function_ invoke__\"\u003eget\u003c/span\u003e(\u0026#x26;key).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e;\n }\n Set { key, val } =\u003e {\n client.\u003cspan class=\"hljs-title function_ invoke__\"\u003eset\u003c/span\u003e(\u0026#x26;key, val).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e;\n }\n }\n }\n});\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eNow, update the two tasks to send commands over the channel instead of issuing\nthem directly on the Redis connection.\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003e\u003cspan class=\"hljs-comment\"\u003e// The `Sender` handles are moved into the tasks. As there are two\u003c/span\u003e\n\u003cspan class=\"hljs-comment\"\u003e// tasks, we need a second `Sender`.\u003c/span\u003e\n\u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003etx2\u003c/span\u003e = tx.\u003cspan class=\"hljs-title function_ invoke__\"\u003eclone\u003c/span\u003e();\n\n\u003cspan class=\"hljs-comment\"\u003e// Spawn two tasks, one gets a key, the other sets a key\u003c/span\u003e\n\u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003et1\u003c/span\u003e = tokio::\u003cspan class=\"hljs-title function_ invoke__\"\u003espawn\u003c/span\u003e(\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003emove\u003c/span\u003e {\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003ecmd\u003c/span\u003e = Command::Get {\n key: \u003cspan class=\"hljs-string\"\u003e\"foo\"\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eto_string\u003c/span\u003e(),\n };\n\n tx.\u003cspan class=\"hljs-title function_ invoke__\"\u003esend\u003c/span\u003e(cmd).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\n});\n\n\u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003et2\u003c/span\u003e = tokio::\u003cspan class=\"hljs-title function_ invoke__\"\u003espawn\u003c/span\u003e(\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003emove\u003c/span\u003e {\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003ecmd\u003c/span\u003e = Command::Set {\n key: \u003cspan class=\"hljs-string\"\u003e\"foo\"\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eto_string\u003c/span\u003e(),\n val: \u003cspan class=\"hljs-string\"\u003e\"bar\"\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003einto\u003c/span\u003e(),\n };\n\n tx2.\u003cspan class=\"hljs-title function_ invoke__\"\u003esend\u003c/span\u003e(cmd).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\n});\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eAt the bottom of the \u003ccode\u003emain\u003c/code\u003e function, we \u003ccode\u003e.await\u003c/code\u003e the join handles to ensure the\ncommands fully complete before the process exits.\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003et1.\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\nt2.\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\nmanager.\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\n\u003c/code\u003e\u003c/pre\u003e\n\u003ch1 id=\"receive-responses\"\u003eReceive responses\u003c/h1\u003e\n\u003cp\u003eThe final step is to receive the response back from the manager task. The \u003ccode\u003eGET\u003c/code\u003e\ncommand needs to get the value and the \u003ccode\u003eSET\u003c/code\u003e command needs to know if the\noperation completed successfully.\u003c/p\u003e\n\u003cp\u003eTo pass the response, a \u003ccode\u003eoneshot\u003c/code\u003e channel is used. The \u003ccode\u003eoneshot\u003c/code\u003e channel is a\nsingle-producer, single-consumer channel optimized for sending a single value.\nIn our case, the single value is the response.\u003c/p\u003e\n\u003cp\u003eSimilar to \u003ccode\u003empsc\u003c/code\u003e, \u003ccode\u003eoneshot::channel()\u003c/code\u003e returns a sender and receiver handle.\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003e\u003cspan class=\"hljs-keyword\"\u003euse\u003c/span\u003e tokio::sync::oneshot;\n\n\u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e (tx, rx) = oneshot::\u003cspan class=\"hljs-title function_ invoke__\"\u003echannel\u003c/span\u003e();\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eUnlike \u003ccode\u003empsc\u003c/code\u003e, no capacity is specified as the capacity is always one.\nAdditionally, neither handle can be cloned.\u003c/p\u003e\n\u003cp\u003eTo receive responses from the manager task, before sending a command, a \u003ccode\u003eoneshot\u003c/code\u003e\nchannel is created. The \u003ccode\u003eSender\u003c/code\u003e half of the channel is included in the command\nto the manager task. The receive half is used to receive the response.\u003c/p\u003e\n\u003cp\u003eFirst, update \u003ccode\u003eCommand\u003c/code\u003e to include the \u003ccode\u003eSender\u003c/code\u003e. For convenience, a type alias\nis used to reference the \u003ccode\u003eSender\u003c/code\u003e.\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003e\u003cspan class=\"hljs-keyword\"\u003euse\u003c/span\u003e tokio::sync::oneshot;\n\u003cspan class=\"hljs-keyword\"\u003euse\u003c/span\u003e bytes::Bytes;\n\n\u003cspan class=\"hljs-comment\"\u003e/// Multiple different commands are multiplexed over a single channel.\u003c/span\u003e\n\u003cspan class=\"hljs-meta\"\u003e#[derive(Debug)]\u003c/span\u003e\n\u003cspan class=\"hljs-keyword\"\u003eenum\u003c/span\u003e \u003cspan class=\"hljs-title class_\"\u003eCommand\u003c/span\u003e {\n Get {\n key: \u003cspan class=\"hljs-type\"\u003eString\u003c/span\u003e,\n resp: Responder\u0026#x3C;\u003cspan class=\"hljs-type\"\u003eOption\u003c/span\u003e\u0026#x3C;Bytes\u003e\u003e,\n },\n Set {\n key: \u003cspan class=\"hljs-type\"\u003eString\u003c/span\u003e,\n val: Bytes,\n resp: Responder\u0026#x3C;()\u003e,\n },\n}\n\n\u003cspan class=\"hljs-comment\"\u003e/// Provided by the requester and used by the manager task to send\u003c/span\u003e\n\u003cspan class=\"hljs-comment\"\u003e/// the command response back to the requester.\u003c/span\u003e\n\u003cspan class=\"hljs-keyword\"\u003etype\u003c/span\u003e \u003cspan class=\"hljs-title class_\"\u003eResponder\u003c/span\u003e\u0026#x3C;T\u003e = oneshot::Sender\u0026#x3C;mini_redis::\u003cspan class=\"hljs-type\"\u003eResult\u003c/span\u003e\u0026#x3C;T\u003e\u003e;\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eNow, update the tasks issuing the commands to include the \u003ccode\u003eoneshot::Sender\u003c/code\u003e.\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003e\u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003et1\u003c/span\u003e = tokio::\u003cspan class=\"hljs-title function_ invoke__\"\u003espawn\u003c/span\u003e(\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003emove\u003c/span\u003e {\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e (resp_tx, resp_rx) = oneshot::\u003cspan class=\"hljs-title function_ invoke__\"\u003echannel\u003c/span\u003e();\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003ecmd\u003c/span\u003e = Command::Get {\n key: \u003cspan class=\"hljs-string\"\u003e\"foo\"\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eto_string\u003c/span\u003e(),\n resp: resp_tx,\n };\n\n \u003cspan class=\"hljs-comment\"\u003e// Send the GET request\u003c/span\u003e\n tx.\u003cspan class=\"hljs-title function_ invoke__\"\u003esend\u003c/span\u003e(cmd).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\n\n \u003cspan class=\"hljs-comment\"\u003e// Await the response\u003c/span\u003e\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003eres\u003c/span\u003e = resp_rx.\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e;\n \u003cspan class=\"hljs-built_in\"\u003eprintln!\u003c/span\u003e(\u003cspan class=\"hljs-string\"\u003e\"GOT = {:?}\"\u003c/span\u003e, res);\n});\n\n\u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003et2\u003c/span\u003e = tokio::\u003cspan class=\"hljs-title function_ invoke__\"\u003espawn\u003c/span\u003e(\u003cspan class=\"hljs-keyword\"\u003easync\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003emove\u003c/span\u003e {\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e (resp_tx, resp_rx) = oneshot::\u003cspan class=\"hljs-title function_ invoke__\"\u003echannel\u003c/span\u003e();\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003ecmd\u003c/span\u003e = Command::Set {\n key: \u003cspan class=\"hljs-string\"\u003e\"foo\"\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eto_string\u003c/span\u003e(),\n val: \u003cspan class=\"hljs-string\"\u003e\"bar\"\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003einto\u003c/span\u003e(),\n resp: resp_tx,\n };\n\n \u003cspan class=\"hljs-comment\"\u003e// Send the SET request\u003c/span\u003e\n tx2.\u003cspan class=\"hljs-title function_ invoke__\"\u003esend\u003c/span\u003e(cmd).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e.\u003cspan class=\"hljs-title function_ invoke__\"\u003eunwrap\u003c/span\u003e();\n\n \u003cspan class=\"hljs-comment\"\u003e// Await the response\u003c/span\u003e\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003eres\u003c/span\u003e = resp_rx.\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e;\n \u003cspan class=\"hljs-built_in\"\u003eprintln!\u003c/span\u003e(\u003cspan class=\"hljs-string\"\u003e\"GOT = {:?}\"\u003c/span\u003e, res);\n});\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eFinally, update the manager task to send the response over the \u003ccode\u003eoneshot\u003c/code\u003e channel.\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003e\u003cspan class=\"hljs-keyword\"\u003ewhile\u003c/span\u003e \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003eSome\u003c/span\u003e(cmd) = rx.\u003cspan class=\"hljs-title function_ invoke__\"\u003erecv\u003c/span\u003e().\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e {\n \u003cspan class=\"hljs-keyword\"\u003ematch\u003c/span\u003e cmd {\n Command::Get { key, resp } =\u003e {\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003eres\u003c/span\u003e = client.\u003cspan class=\"hljs-title function_ invoke__\"\u003eget\u003c/span\u003e(\u0026#x26;key).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e;\n \u003cspan class=\"hljs-comment\"\u003e// Ignore errors\u003c/span\u003e\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003e_\u003c/span\u003e = resp.\u003cspan class=\"hljs-title function_ invoke__\"\u003esend\u003c/span\u003e(res);\n }\n Command::Set { key, val, resp } =\u003e {\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003eres\u003c/span\u003e = client.\u003cspan class=\"hljs-title function_ invoke__\"\u003eset\u003c/span\u003e(\u0026#x26;key, val).\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e;\n \u003cspan class=\"hljs-comment\"\u003e// Ignore errors\u003c/span\u003e\n \u003cspan class=\"hljs-keyword\"\u003elet\u003c/span\u003e \u003cspan class=\"hljs-variable\"\u003e_\u003c/span\u003e = resp.\u003cspan class=\"hljs-title function_ invoke__\"\u003esend\u003c/span\u003e(res);\n }\n }\n}\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eCalling \u003ccode\u003esend\u003c/code\u003e on \u003ccode\u003eoneshot::Sender\u003c/code\u003e completes immediately and does \u003cstrong\u003enot\u003c/strong\u003e\nrequire an \u003ccode\u003e.await\u003c/code\u003e. This is because \u003ccode\u003esend\u003c/code\u003e on a \u003ccode\u003eoneshot\u003c/code\u003e channel will always\nfail or succeed immediately without any form of waiting.\u003c/p\u003e\n\u003cp\u003eSending a value on a oneshot channel returns \u003ccode\u003eErr\u003c/code\u003e when the receiver half has\ndropped. This indicates the receiver is no longer interested in the response. In\nour scenario, the receiver cancelling interest is an acceptable event. The \u003ccode\u003eErr\u003c/code\u003e\nreturned by \u003ccode\u003eresp.send(...)\u003c/code\u003e does not need to be handled.\u003c/p\u003e\n\u003cp\u003eYou can find the entire code \u003ca href=\"https://github.com/tokio-rs/website/blob/master/tutorial-code/channels/src/main.rs\"\u003ehere\u003c/a\u003e.\u003c/p\u003e\n\u003ch1 id=\"backpressure-and-bounded-channels\"\u003eBackpressure and bounded channels\u003c/h1\u003e\n\u003cp\u003eWhenever concurrency or queuing is introduced, it is important to ensure that the\nqueing is bounded and the system will gracefully handle the load. Unbounded queues\nwill eventually fill up all available memory and cause the system to fail in\nunpredictable ways.\u003c/p\u003e\n\u003cp\u003eTokio takes care to avoid implicit queuing. A big part of this is the fact that\nasync operations are lazy. Consider the following:\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003e\u003cspan class=\"hljs-keyword\"\u003eloop\u003c/span\u003e {\n \u003cspan class=\"hljs-title function_ invoke__\"\u003easync_op\u003c/span\u003e();\n}\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eIf the asynchronous operation runs eagerly, the loop will repeatedly queue a new\n\u003ccode\u003easync_op\u003c/code\u003e to run without ensuring the previous operation completed. This\nresults in implicit unbounded queuing. Callback based systems and \u003cstrong\u003eeager\u003c/strong\u003e\nfuture based systems are particularly susceptible to this.\u003c/p\u003e\n\u003cp\u003eHowever, with Tokio and asynchronous Rust, the above snippet will \u003cstrong\u003enot\u003c/strong\u003e result\nin \u003ccode\u003easync_op\u003c/code\u003e running at all. This is because \u003ccode\u003e.await\u003c/code\u003e is never called. If the\nsnippet is updated to use \u003ccode\u003e.await\u003c/code\u003e, then the loop waits for the operation to\ncomplete before starting over.\u003c/p\u003e\n\u003cpre\u003e\u003ccode class=\"hljs language-rust\"\u003e\u003cspan class=\"hljs-keyword\"\u003eloop\u003c/span\u003e {\n \u003cspan class=\"hljs-comment\"\u003e// Will not repeat until `async_op` completes\u003c/span\u003e\n \u003cspan class=\"hljs-title function_ invoke__\"\u003easync_op\u003c/span\u003e().\u003cspan class=\"hljs-keyword\"\u003eawait\u003c/span\u003e;\n}\n\u003c/code\u003e\u003c/pre\u003e\n\u003cp\u003eConcurrency and queuing must be explicitly introduced. Ways to do this include:\u003c/p\u003e\n\u003cul\u003e\n\u003cli\u003e\u003ccode\u003etokio::spawn\u003c/code\u003e\u003c/li\u003e\n\u003cli\u003e\u003ccode\u003eselect!\u003c/code\u003e\u003c/li\u003e\n\u003cli\u003e\u003ccode\u003ejoin!\u003c/code\u003e\u003c/li\u003e\n\u003cli\u003e\u003ccode\u003empsc::channel\u003c/code\u003e\u003c/li\u003e\n\u003c/ul\u003e\n\u003cp\u003eWhen doing so, take care to ensure the total amount of concurrency is bounded. For\nexample, when writing a TCP accept loop, ensure that the total number of open\nsockets is bounded. When using \u003ccode\u003empsc::channel\u003c/code\u003e, pick a manageable channel\ncapacity. Specific bound values will be application specific.\u003c/p\u003e\n\u003cp\u003eTaking care and picking good bounds is a big part of writing reliable Tokio applications.\u003c/p\u003e","description":null,"prev":{"title":"Shared state","href":"/tokio/tutorial/shared-state"},"next":{"title":"I/O","href":"/tokio/tutorial/io"}},"menu":[{"key":"tokio","title":"Tokio","nested":[{"page":{"key":"tutorial","href":"/tokio/tutorial","title":"Tutorial","data":{"title":"Tutorial","subtitle":"Overview"}},"nested":[{"page":{"key":"setup","href":"/tokio/tutorial/setup","title":"Setup","data":{"title":"Setup"}}},{"page":{"key":"hello-tokio","href":"/tokio/tutorial/hello-tokio","title":"Hello Tokio","data":{"title":"Hello Tokio"}}},{"page":{"key":"spawning","href":"/tokio/tutorial/spawning","title":"Spawning","data":{"title":"Spawning"}}},{"page":{"key":"shared-state","href":"/tokio/tutorial/shared-state","title":"Shared state","data":{"title":"Shared state"}}},{"page":{"key":"channels","href":"/tokio/tutorial/channels","title":"Channels","data":{"title":"Channels"}}},{"page":{"key":"io","href":"/tokio/tutorial/io","title":"I/O","data":{"title":"I/O"}}},{"page":{"key":"framing","href":"/tokio/tutorial/framing","title":"Framing","data":{"title":"Framing"}}},{"page":{"key":"async","href":"/tokio/tutorial/async","title":"Async in depth","data":{"title":"Async in depth"}}},{"page":{"key":"select","href":"/tokio/tutorial/select","title":"Select","data":{"title":"Select"}}},{"page":{"key":"streams","href":"/tokio/tutorial/streams","title":"Streams","data":{"title":"Streams"}}}]},{"page":{"key":"topics","href":"/tokio/topics","title":"Topics","data":{"title":"Topics"}},"nested":[{"page":{"key":"bridging","href":"/tokio/topics/bridging","title":"Bridging with sync code","data":{"title":"Bridging with sync code"}}},{"page":{"key":"shutdown","href":"/tokio/topics/shutdown","title":"Graceful Shutdown","data":{"title":"Graceful Shutdown"}}},{"page":{"key":"tracing","href":"/tokio/topics/tracing","title":"Getting started with Tracing","data":{"title":"Getting started with Tracing"}}},{"page":{"key":"tracing-next-steps","href":"/tokio/topics/tracing-next-steps","title":"Next steps with Tracing","data":{"title":"Next steps with Tracing"}}},{"page":{"key":"testing","href":"/tokio/topics/testing","title":"Unit Testing","data":{"title":"Unit Testing"}}}]},{"page":{"key":"glossary","href":"/tokio/glossary","title":"Glossary","data":{"title":"Glossary"}}},{"page":{"key":"api","title":"API documentation","href":"https://docs.rs/tokio"}}]}],"app":{"blog":{"date":1758844800000,"key":"2025-09-26-announcing-tokio-conf-cfp","href":"/blog/2025-09-26-announcing-tokio-conf-cfp","title":"The TokioConf 2026 Call For Talk Proposals is now open","menuTitle":"The TokioConf 2026 Call For Talk Proposals is now open","mdPath":"blog/2025-09-26-announcing-tokio-conf-cfp.md","data":{"date":"2025-09-26","title":"The TokioConf 2026 Call For Talk Proposals is now open","description":"September 26, 2025"},"description":"September 26, 2025"}}},"__N_SSG":true},"page":"/[...slug]","query":{"slug":["tokio","tutorial","channels"]},"buildId":"h5xuILhXLuVXe-k6nNHpb","isFallback":false,"gsp":true,"scriptLoader":[]}</script><next-route-announcer><p aria-live="assertive" id="__next-route-announcer__" role="alert" style="border: 0px; clip: rect(0px, 0px, 0px, 0px); height: 1px; margin: -1px; overflow: hidden; padding: 0px; position: absolute; top: 0px; width: 1px; white-space: nowrap; overflow-wrap: normal;"></p></next-route-announcer></body></html>
Reference in New Issue View Git Blame Copy Permalink