dewey/docs-rust
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
docs-rust/ch21/ch21-02-multithreaded.html
Aditya Gupta 4581eea409 feat: added cleanscript
2026-06-22 21:27:36 +05:30

1266 lines
67 KiB
HTML
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>From Single-Threaded to Multithreaded Server</title>
</head>
<body>
<!-- Old headings. Do not remove or links may break. -->
<p><a id="turning-our-single-threaded-server-into-a-multithreaded-server"></a>
<a id="from-single-threaded-to-multithreaded-server"></a></p>
<h2 id="from-a-single-threaded-to-a-multithreaded-server"><a class="header" href="#from-a-single-threaded-to-a-multithreaded-server">From a Single-Threaded to a Multithreaded Server</a></h2>
<p>Right now, the server will process each request in turn, meaning it won’t
process a second connection until the first connection is finished processing.
If the server received more and more requests, this serial execution would be
less and less optimal. If the server receives a request that takes a long time
to process, subsequent requests will have to wait until the long request is
finished, even if the new requests can be processed quickly. We’ll need to fix
this, but first we’ll look at the problem in action.</p>
<!-- Old headings. Do not remove or links may break. -->
<p><a id="simulating-a-slow-request-in-the-current-server-implementation"></a></p>
<h3 id="simulating-a-slow-request"><a class="header" href="#simulating-a-slow-request">Simulating a Slow Request</a></h3>
<p>We’ll look at how a slowly processing request can affect other requests made to
our current server implementation. Listing 21-10 implements handling a request
to <em>/sleep</em> with a simulated slow response that will cause the server to sleep
for five seconds before responding.</p>
<figure class="listing" id="listing-21-10">
<span class="file-name">Filename: src/main.rs</span>
<pre class="playground"><code class="language-rust no_run edition2024">use std::{
fs,
io::{BufReader, prelude::*},
net::{TcpListener, TcpStream},
thread,
time::Duration,
};
// --snip--
<span class="boring">
</span><span class="boring">fn main() {
</span><span class="boring"> let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
</span><span class="boring">
</span><span class="boring"> for stream in listener.incoming() {
</span><span class="boring"> let stream = stream.unwrap();
</span><span class="boring">
</span><span class="boring"> handle_connection(stream);
</span><span class="boring"> }
</span><span class="boring">}
</span>
fn handle_connection(mut stream: TcpStream) {
// --snip--
<span class="boring"> let buf_reader = BufReader::new(&amp;stream);
</span><span class="boring"> let request_line = buf_reader.lines().next().unwrap().unwrap();
</span><span class="boring">
</span> let (status_line, filename) = match &amp;request_line[..] {
"GET / HTTP/1.1" =&gt; ("HTTP/1.1 200 OK", "hello.html"),
"GET /sleep HTTP/1.1" =&gt; {
thread::sleep(Duration::from_secs(5));
("HTTP/1.1 200 OK", "hello.html")
}
_ =&gt; ("HTTP/1.1 404 NOT FOUND", "404.html"),
};
// --snip--
<span class="boring">
</span><span class="boring"> let contents = fs::read_to_string(filename).unwrap();
</span><span class="boring"> let length = contents.len();
</span><span class="boring">
</span><span class="boring"> let response =
</span><span class="boring"> format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
</span><span class="boring">
</span><span class="boring"> stream.write_all(response.as_bytes()).unwrap();
</span>}</code></pre>
<figcaption><a href="#listing-21-10">Listing 21-10</a>: Simulating a slow request by sleeping for five seconds</figcaption>
</figure>
<p>We switched from <code>if</code> to <code>match</code> now that we have three cases. We need to
explicitly match on a slice of <code>request_line</code> to pattern-match against the
string literal values; <code>match</code> doesn’t do automatic referencing and
dereferencing, like the equality method does.</p>
<p>The first arm is the same as the <code>if</code> block from Listing 21-9. The second arm
matches a request to <em>/sleep</em>. When that request is received, the server will
sleep for five seconds before rendering the successful HTML page. The third arm
is the same as the <code>else</code> block from Listing 21-9.</p>
<p>You can see how primitive our server is: Real libraries would handle the
recognition of multiple requests in a much less verbose way!</p>
<p>Start the server using <code>cargo run</code>. Then, open two browser windows: one for
<em>http://127.0.0.1:7878</em> and the other for <em>http://127.0.0.1:7878/sleep</em>. If you
enter the <em>/</em> URI a few times, as before, you’ll see it respond quickly. But if
you enter <em>/sleep</em> and then load <em>/</em>, you’ll see that <em>/</em> waits until <code>sleep</code>
has slept for its full five seconds before loading.</p>
<p>There are multiple techniques we could use to avoid requests backing up behind
a slow request, including using async as we did Chapter 17; the one we’ll
implement is a thread pool.</p>
<h3 id="improving-throughput-with-a-thread-pool"><a class="header" href="#improving-throughput-with-a-thread-pool">Improving Throughput with a Thread Pool</a></h3>
<p>A <em>thread pool</em> is a group of spawned threads that are ready and waiting to
handle a task. When the program receives a new task, it assigns one of the
threads in the pool to the task, and that thread will process the task. The
remaining threads in the pool are available to handle any other tasks that come
in while the first thread is processing. When the first thread is done
processing its task, it’s returned to the pool of idle threads, ready to handle
a new task. A thread pool allows you to process connections concurrently,
increasing the throughput of your server.</p>
<p>We’ll limit the number of threads in the pool to a small number to protect us
from DoS attacks; if we had our program create a new thread for each request as
it came in, someone making 10 million requests to our server could wreak havoc
by using up all our server’s resources and grinding the processing of requests
to a halt.</p>
<p>Rather than spawning unlimited threads, then, we’ll have a fixed number of
threads waiting in the pool. Requests that come in are sent to the pool for
processing. The pool will maintain a queue of incoming requests. Each of the
threads in the pool will pop off a request from this queue, handle the request,
and then ask the queue for another request. With this design, we can process up
to <em><code>N</code></em> requests concurrently, where <em><code>N</code></em> is the number of threads. If each
thread is responding to a long-running request, subsequent requests can still
back up in the queue, but we’ve increased the number of long-running requests
we can handle before reaching that point.</p>
<p>This technique is just one of many ways to improve the throughput of a web
server. Other options you might explore are the fork/join model, the
single-threaded async I/O model, and the multithreaded async I/O model. If
you’re interested in this topic, you can read more about other solutions and
try to implement them; with a low-level language like Rust, all of these
options are possible.</p>
<p>Before we begin implementing a thread pool, let’s talk about what using the
pool should look like. When you’re trying to design code, writing the client
interface first can help guide your design. Write the API of the code so that
it’s structured in the way you want to call it; then, implement the
functionality within that structure rather than implementing the functionality
and then designing the public API.</p>
<p>Similar to how we used test-driven development in the project in Chapter 12,
we’ll use compiler-driven development here. We’ll write the code that calls the
functions we want, and then we’ll look at errors from the compiler to determine
what we should change next to get the code to work. Before we do that, however,
we’ll explore the technique we’re not going to use as a starting point.</p>
<!-- Old headings. Do not remove or links may break. -->
<p><a id="code-structure-if-we-could-spawn-a-thread-for-each-request"></a></p>
<h4 id="spawning-a-thread-for-each-request"><a class="header" href="#spawning-a-thread-for-each-request">Spawning a Thread for Each Request</a></h4>
<p>First, let’s explore how our code might look if it did create a new thread for
every connection. As mentioned earlier, this isn’t our final plan due to the
problems with potentially spawning an unlimited number of threads, but it is a
starting point to get a working multithreaded server first. Then, we’ll add the
thread pool as an improvement, and contrasting the two solutions will be easier.</p>
<p>Listing 21-11 shows the changes to make to <code>main</code> to spawn a new thread to
handle each stream within the <code>for</code> loop.</p>
<figure class="listing" id="listing-21-11">
<span class="file-name">Filename: src/main.rs</span>
<pre class="playground"><code class="language-rust no_run edition2024"><span class="boring">use std::{
</span><span class="boring"> fs,
</span><span class="boring"> io::{BufReader, prelude::*},
</span><span class="boring"> net::{TcpListener, TcpStream},
</span><span class="boring"> thread,
</span><span class="boring"> time::Duration,
</span><span class="boring">};
</span><span class="boring">
</span>fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
for stream in listener.incoming() {
let stream = stream.unwrap();
thread::spawn(|| {
handle_connection(stream);
});
}
}
<span class="boring">
</span><span class="boring">fn handle_connection(mut stream: TcpStream) {
</span><span class="boring"> let buf_reader = BufReader::new(&amp;stream);
</span><span class="boring"> let request_line = buf_reader.lines().next().unwrap().unwrap();
</span><span class="boring">
</span><span class="boring"> let (status_line, filename) = match &amp;request_line[..] {
</span><span class="boring"> "GET / HTTP/1.1" =&gt; ("HTTP/1.1 200 OK", "hello.html"),
</span><span class="boring"> "GET /sleep HTTP/1.1" =&gt; {
</span><span class="boring"> thread::sleep(Duration::from_secs(5));
</span><span class="boring"> ("HTTP/1.1 200 OK", "hello.html")
</span><span class="boring"> }
</span><span class="boring"> _ =&gt; ("HTTP/1.1 404 NOT FOUND", "404.html"),
</span><span class="boring"> };
</span><span class="boring">
</span><span class="boring"> let contents = fs::read_to_string(filename).unwrap();
</span><span class="boring"> let length = contents.len();
</span><span class="boring">
</span><span class="boring"> let response =
</span><span class="boring"> format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
</span><span class="boring">
</span><span class="boring"> stream.write_all(response.as_bytes()).unwrap();
</span><span class="boring">}</span></code></pre>
<figcaption><a href="#listing-21-11">Listing 21-11</a>: Spawning a new thread for each stream</figcaption>
</figure>
<p>As you learned in Chapter 16, <code>thread::spawn</code> will create a new thread and then
run the code in the closure in the new thread. If you run this code and load
<em>/sleep</em> in your browser, then <em>/</em> in two more browser tabs, you’ll indeed see
that the requests to <em>/</em> don’t have to wait for <em>/sleep</em> to finish. However, as
we mentioned, this will eventually overwhelm the system because you’d be making
new threads without any limit.</p>
<p>You may also recall from Chapter 17 that this is exactly the kind of situation
where async and await really shine! Keep that in mind as we build the thread
pool and think about how things would look different or the same with async.</p>
<!-- Old headings. Do not remove or links may break. -->
<p><a id="creating-a-similar-interface-for-a-finite-number-of-threads"></a></p>
<h4 id="creating-a-finite-number-of-threads"><a class="header" href="#creating-a-finite-number-of-threads">Creating a Finite Number of Threads</a></h4>
<p>We want our thread pool to work in a similar, familiar way so that switching
from threads to a thread pool doesn’t require large changes to the code that
uses our API. Listing 21-12 shows the hypothetical interface for a <code>ThreadPool</code>
struct we want to use instead of <code>thread::spawn</code>.</p>
<figure class="listing" id="listing-21-12">
<span class="file-name">Filename: src/main.rs</span>
<pre><code class="language-rust ignore does_not_compile"><span class="boring">use std::{
</span><span class="boring"> fs,
</span><span class="boring"> io::{BufReader, prelude::*},
</span><span class="boring"> net::{TcpListener, TcpStream},
</span><span class="boring"> thread,
</span><span class="boring"> time::Duration,
</span><span class="boring">};
</span><span class="boring">
</span>fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
let pool = ThreadPool::new(4);
for stream in listener.incoming() {
let stream = stream.unwrap();
pool.execute(|| {
handle_connection(stream);
});
}
}
<span class="boring">
</span><span class="boring">fn handle_connection(mut stream: TcpStream) {
</span><span class="boring"> let buf_reader = BufReader::new(&amp;stream);
</span><span class="boring"> let request_line = buf_reader.lines().next().unwrap().unwrap();
</span><span class="boring">
</span><span class="boring"> let (status_line, filename) = match &amp;request_line[..] {
</span><span class="boring"> "GET / HTTP/1.1" =&gt; ("HTTP/1.1 200 OK", "hello.html"),
</span><span class="boring"> "GET /sleep HTTP/1.1" =&gt; {
</span><span class="boring"> thread::sleep(Duration::from_secs(5));
</span><span class="boring"> ("HTTP/1.1 200 OK", "hello.html")
</span><span class="boring"> }
</span><span class="boring"> _ =&gt; ("HTTP/1.1 404 NOT FOUND", "404.html"),
</span><span class="boring"> };
</span><span class="boring">
</span><span class="boring"> let contents = fs::read_to_string(filename).unwrap();
</span><span class="boring"> let length = contents.len();
</span><span class="boring">
</span><span class="boring"> let response =
</span><span class="boring"> format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
</span><span class="boring">
</span><span class="boring"> stream.write_all(response.as_bytes()).unwrap();
</span><span class="boring">}</span></code></pre>
<figcaption><a href="#listing-21-12">Listing 21-12</a>: Our ideal <code>ThreadPool</code> interface</figcaption>
</figure>
<p>We use <code>ThreadPool::new</code> to create a new thread pool with a configurable number
of threads, in this case four. Then, in the <code>for</code> loop, <code>pool.execute</code> has a
similar interface as <code>thread::spawn</code> in that it takes a closure that the pool
should run for each stream. We need to implement <code>pool.execute</code> so that it
takes the closure and gives it to a thread in the pool to run. This code won’t
yet compile, but we’ll try so that the compiler can guide us in how to fix it.</p>
<!-- Old headings. Do not remove or links may break. -->
<p><a id="building-the-threadpool-struct-using-compiler-driven-development"></a></p>
<h4 id="building-threadpool-using-compiler-driven-development"><a class="header" href="#building-threadpool-using-compiler-driven-development">Building <code>ThreadPool</code> Using Compiler-Driven Development</a></h4>
<p>Make the changes in Listing 21-12 to <em>src/main.rs</em>, and then let’s use the
compiler errors from <code>cargo check</code> to drive our development. Here is the first
error we get:</p>
<pre><code class="language-console">$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0433]: failed to resolve: use of undeclared type `ThreadPool`
--&gt; src/main.rs:11:16
|
11 | let pool = ThreadPool::new(4);
| ^^^^^^^^^^ use of undeclared type `ThreadPool`
For more information about this error, try `rustc --explain E0433`.
error: could not compile `hello` (bin "hello") due to 1 previous error
</code></pre>
<p>Great! This error tells us we need a <code>ThreadPool</code> type or module, so we’ll
build one now. Our <code>ThreadPool</code> implementation will be independent of the kind
of work our web server is doing. So, let’s switch the <code>hello</code> crate from a
binary crate to a library crate to hold our <code>ThreadPool</code> implementation. After
we change to a library crate, we could also use the separate thread pool
library for any work we want to do using a thread pool, not just for serving
web requests.</p>
<p>Create a <em>src/lib.rs</em> file that contains the following, which is the simplest
definition of a <code>ThreadPool</code> struct that we can have for now:</p>
<figure class="listing">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust noplayground">pub struct ThreadPool;</code></pre>
</figure>
<p>Then, edit the <em>main.rs</em> file to bring <code>ThreadPool</code> into scope from the library
crate by adding the following code to the top of <em>src/main.rs</em>:</p>
<figure class="listing">
<span class="file-name">Filename: src/main.rs</span>
<pre><code class="language-rust ignore">use hello::ThreadPool;
<span class="boring">use std::{
</span><span class="boring"> fs,
</span><span class="boring"> io::{BufReader, prelude::*},
</span><span class="boring"> net::{TcpListener, TcpStream},
</span><span class="boring"> thread,
</span><span class="boring"> time::Duration,
</span><span class="boring">};
</span><span class="boring">
</span><span class="boring">fn main() {
</span><span class="boring"> let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
</span><span class="boring"> let pool = ThreadPool::new(4);
</span><span class="boring">
</span><span class="boring"> for stream in listener.incoming() {
</span><span class="boring"> let stream = stream.unwrap();
</span><span class="boring">
</span><span class="boring"> pool.execute(|| {
</span><span class="boring"> handle_connection(stream);
</span><span class="boring"> });
</span><span class="boring"> }
</span><span class="boring">}
</span><span class="boring">
</span><span class="boring">fn handle_connection(mut stream: TcpStream) {
</span><span class="boring"> let buf_reader = BufReader::new(&amp;stream);
</span><span class="boring"> let request_line = buf_reader.lines().next().unwrap().unwrap();
</span><span class="boring">
</span><span class="boring"> let (status_line, filename) = match &amp;request_line[..] {
</span><span class="boring"> "GET / HTTP/1.1" =&gt; ("HTTP/1.1 200 OK", "hello.html"),
</span><span class="boring"> "GET /sleep HTTP/1.1" =&gt; {
</span><span class="boring"> thread::sleep(Duration::from_secs(5));
</span><span class="boring"> ("HTTP/1.1 200 OK", "hello.html")
</span><span class="boring"> }
</span><span class="boring"> _ =&gt; ("HTTP/1.1 404 NOT FOUND", "404.html"),
</span><span class="boring"> };
</span><span class="boring">
</span><span class="boring"> let contents = fs::read_to_string(filename).unwrap();
</span><span class="boring"> let length = contents.len();
</span><span class="boring">
</span><span class="boring"> let response =
</span><span class="boring"> format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
</span><span class="boring">
</span><span class="boring"> stream.write_all(response.as_bytes()).unwrap();
</span><span class="boring">}</span></code></pre>
</figure>
<p>This code still won’t work, but let’s check it again to get the next error that
we need to address:</p>
<pre><code class="language-console">$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0599]: no function or associated item named `new` found for struct `ThreadPool` in the current scope
--&gt; src/main.rs:12:28
|
12 | let pool = ThreadPool::new(4);
| ^^^ function or associated item not found in `ThreadPool`
For more information about this error, try `rustc --explain E0599`.
error: could not compile `hello` (bin "hello") due to 1 previous error
</code></pre>
<p>This error indicates that next we need to create an associated function named
<code>new</code> for <code>ThreadPool</code>. We also know that <code>new</code> needs to have one parameter
that can accept <code>4</code> as an argument and should return a <code>ThreadPool</code> instance.
Let’s implement the simplest <code>new</code> function that will have those
characteristics:</p>
<figure class="listing">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust noplayground">pub struct ThreadPool;
impl ThreadPool {
pub fn new(size: usize) -&gt; ThreadPool {
ThreadPool
}
}</code></pre>
</figure>
<p>We chose <code>usize</code> as the type of the <code>size</code> parameter because we know that a
negative number of threads doesn’t make any sense. We also know we’ll use this
<code>4</code> as the number of elements in a collection of threads, which is what the
<code>usize</code> type is for, as discussed in the <a href="../ch03/ch03-02-data-types.html#integer-types">“Integer Types”</a><!--
ignore --> section in Chapter 3.</p>
<p>Let’s check the code again:</p>
<pre><code class="language-console">$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0599]: no method named `execute` found for struct `ThreadPool` in the current scope
--&gt; src/main.rs:17:14
|
17 | pool.execute(|| {
| -----^^^^^^^ method not found in `ThreadPool`
For more information about this error, try `rustc --explain E0599`.
error: could not compile `hello` (bin "hello") due to 1 previous error
</code></pre>
<p>Now the error occurs because we don’t have an <code>execute</code> method on <code>ThreadPool</code>.
Recall from the <a href="#creating-a-finite-number-of-threads">“Creating a Finite Number of
Threads”</a><!-- ignore --> section that we
decided our thread pool should have an interface similar to <code>thread::spawn</code>. In
addition, we’ll implement the <code>execute</code> function so that it takes the closure
it’s given and gives it to an idle thread in the pool to run.</p>
<p>We’ll define the <code>execute</code> method on <code>ThreadPool</code> to take a closure as a
parameter. Recall from the <a href="../ch13/ch13-01-closures.html#moving-captured-values-out-of-closures">“Moving Captured Values Out of
Closures”</a><!-- ignore --> in Chapter 13 that we can
take closures as parameters with three different traits: <code>Fn</code>, <code>FnMut</code>, and
<code>FnOnce</code>. We need to decide which kind of closure to use here. We know we’ll
end up doing something similar to the standard library <code>thread::spawn</code>
implementation, so we can look at what bounds the signature of <code>thread::spawn</code>
has on its parameter. The documentation shows us the following:</p>
<pre><code class="language-rust ignore">pub fn spawn&lt;F, T&gt;(f: F) -&gt; JoinHandle&lt;T&gt;
where
F: FnOnce() -&gt; T,
F: Send + 'static,
T: Send + 'static,</code></pre>
<p>The <code>F</code> type parameter is the one we’re concerned with here; the <code>T</code> type
parameter is related to the return value, and we’re not concerned with that. We
can see that <code>spawn</code> uses <code>FnOnce</code> as the trait bound on <code>F</code>. This is probably
what we want as well, because we’ll eventually pass the argument we get in
<code>execute</code> to <code>spawn</code>. We can be further confident that <code>FnOnce</code> is the trait we
want to use because the thread for running a request will only execute that
request’s closure one time, which matches the <code>Once</code> in <code>FnOnce</code>.</p>
<p>The <code>F</code> type parameter also has the trait bound <code>Send</code> and the lifetime bound
<code>'static</code>, which are useful in our situation: We need <code>Send</code> to transfer the
closure from one thread to another and <code>'static</code> because we don’t know how long
the thread will take to execute. Let’s create an <code>execute</code> method on
<code>ThreadPool</code> that will take a generic parameter of type <code>F</code> with these bounds:</p>
<figure class="listing">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust noplayground"><span class="boring">pub struct ThreadPool;
</span><span class="boring">
</span>impl ThreadPool {
// --snip--
<span class="boring"> pub fn new(size: usize) -&gt; ThreadPool {
</span><span class="boring"> ThreadPool
</span><span class="boring"> }
</span><span class="boring">
</span> pub fn execute&lt;F&gt;(&amp;self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}</code></pre>
</figure>
<p>We still use the <code>()</code> after <code>FnOnce</code> because this <code>FnOnce</code> represents a closure
that takes no parameters and returns the unit type <code>()</code>. Just like function
definitions, the return type can be omitted from the signature, but even if we
have no parameters, we still need the parentheses.</p>
<p>Again, this is the simplest implementation of the <code>execute</code> method: It does
nothing, but we’re only trying to make our code compile. Let’s check it again:</p>
<pre><code class="language-console">$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.24s
</code></pre>
<p>It compiles! But note that if you try <code>cargo run</code> and make a request in the
browser, you’ll see the errors in the browser that we saw at the beginning of
the chapter. Our library isn’t actually calling the closure passed to <code>execute</code>
yet!</p>
<section class="note" aria-role="note">
<p>Note: A saying you might hear about languages with strict compilers, such as
Haskell and Rust, is “If the code compiles, it works.” But this saying is not
universally true. Our project compiles, but it does absolutely nothing! If we
were building a real, complete project, this would be a good time to start
writing unit tests to check that the code compiles <em>and</em> has the behavior we
want.</p>
</section>
<p>Consider: What would be different here if we were going to execute a future
instead of a closure?</p>
<h4 id="validating-the-number-of-threads-in-new"><a class="header" href="#validating-the-number-of-threads-in-new">Validating the Number of Threads in <code>new</code></a></h4>
<p>We aren’t doing anything with the parameters to <code>new</code> and <code>execute</code>. Let’s
implement the bodies of these functions with the behavior we want. To start,
let’s think about <code>new</code>. Earlier we chose an unsigned type for the <code>size</code>
parameter because a pool with a negative number of threads makes no sense.
However, a pool with zero threads also makes no sense, yet zero is a perfectly
valid <code>usize</code>. We’ll add code to check that <code>size</code> is greater than zero before
we return a <code>ThreadPool</code> instance, and we’ll have the program panic if it
receives a zero by using the <code>assert!</code> macro, as shown in Listing 21-13.</p>
<figure class="listing" id="listing-21-13">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust noplayground"><span class="boring">pub struct ThreadPool;
</span><span class="boring">
</span>impl ThreadPool {
/// Create a new ThreadPool.
///
/// The size is the number of threads in the pool.
///
/// # Panics
///
/// The `new` function will panic if the size is zero.
pub fn new(size: usize) -&gt; ThreadPool {
assert!(size &gt; 0);
ThreadPool
}
// --snip--
<span class="boring"> pub fn execute&lt;F&gt;(&amp;self, f: F)
</span><span class="boring"> where
</span><span class="boring"> F: FnOnce() + Send + 'static,
</span><span class="boring"> {
</span><span class="boring"> }
</span>}</code></pre>
<figcaption><a href="#listing-21-13">Listing 21-13</a>: Implementing <code>ThreadPool::new</code> to panic if <code>size</code> is zero</figcaption>
</figure>
<p>We’ve also added some documentation for our <code>ThreadPool</code> with doc comments.
Note that we followed good documentation practices by adding a section that
calls out the situations in which our function can panic, as discussed in
Chapter 14. Try running <code>cargo doc --open</code> and clicking the <code>ThreadPool</code> struct
to see what the generated docs for <code>new</code> look like!</p>
<p>Instead of adding the <code>assert!</code> macro as we’ve done here, we could change <code>new</code>
into <code>build</code> and return a <code>Result</code> like we did with <code>Config::build</code> in the I/O
project in Listing 12-9. But we’ve decided in this case that trying to create a
thread pool without any threads should be an unrecoverable error. If you’re
feeling ambitious, try to write a function named <code>build</code> with the following
signature to compare with the <code>new</code> function:</p>
<pre><code class="language-rust ignore">pub fn build(size: usize) -&gt; Result&lt;ThreadPool, PoolCreationError&gt; {</code></pre>
<h4 id="creating-space-to-store-the-threads"><a class="header" href="#creating-space-to-store-the-threads">Creating Space to Store the Threads</a></h4>
<p>Now that we have a way to know we have a valid number of threads to store in
the pool, we can create those threads and store them in the <code>ThreadPool</code> struct
before returning the struct. But how do we “store” a thread? Let’s take another
look at the <code>thread::spawn</code> signature:</p>
<pre><code class="language-rust ignore">pub fn spawn&lt;F, T&gt;(f: F) -&gt; JoinHandle&lt;T&gt;
where
F: FnOnce() -&gt; T,
F: Send + 'static,
T: Send + 'static,</code></pre>
<p>The <code>spawn</code> function returns a <code>JoinHandle&lt;T&gt;</code>, where <code>T</code> is the type that the
closure returns. Let’s try using <code>JoinHandle</code> too and see what happens. In our
case, the closures we’re passing to the thread pool will handle the connection
and not return anything, so <code>T</code> will be the unit type <code>()</code>.</p>
<p>The code in Listing 21-14 will compile, but it doesn’t create any threads yet.
We’ve changed the definition of <code>ThreadPool</code> to hold a vector of
<code>thread::JoinHandle&lt;()&gt;</code> instances, initialized the vector with a capacity of
<code>size</code>, set up a <code>for</code> loop that will run some code to create the threads, and
returned a <code>ThreadPool</code> instance containing them.</p>
<figure class="listing" id="listing-21-14">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust ignore not_desired_behavior">use std::thread;
pub struct ThreadPool {
threads: Vec&lt;thread::JoinHandle&lt;()&gt;&gt;,
}
impl ThreadPool {
// --snip--
<span class="boring"> /// Create a new ThreadPool.
</span><span class="boring"> ///
</span><span class="boring"> /// The size is the number of threads in the pool.
</span><span class="boring"> ///
</span><span class="boring"> /// # Panics
</span><span class="boring"> ///
</span><span class="boring"> /// The `new` function will panic if the size is zero.
</span> pub fn new(size: usize) -&gt; ThreadPool {
assert!(size &gt; 0);
let mut threads = Vec::with_capacity(size);
for _ in 0..size {
// create some threads and store them in the vector
}
ThreadPool { threads }
}
// --snip--
<span class="boring">
</span><span class="boring"> pub fn execute&lt;F&gt;(&amp;self, f: F)
</span><span class="boring"> where
</span><span class="boring"> F: FnOnce() + Send + 'static,
</span><span class="boring"> {
</span><span class="boring"> }
</span>}</code></pre>
<figcaption><a href="#listing-21-14">Listing 21-14</a>: Creating a vector for <code>ThreadPool</code> to hold the threads</figcaption>
</figure>
<p>We’ve brought <code>std::thread</code> into scope in the library crate because we’re
using <code>thread::JoinHandle</code> as the type of the items in the vector in
<code>ThreadPool</code>.</p>
<p>Once a valid size is received, our <code>ThreadPool</code> creates a new vector that can
hold <code>size</code> items. The <code>with_capacity</code> function performs the same task as
<code>Vec::new</code> but with an important difference: It pre-allocates space in the
vector. Because we know we need to store <code>size</code> elements in the vector, doing
this allocation up front is slightly more efficient than using <code>Vec::new</code>,
which resizes itself as elements are inserted.</p>
<p>When you run <code>cargo check</code> again, it should succeed.</p>
<!-- Old headings. Do not remove or links may break. -->
<p><a id="a-worker-struct-responsible-for-sending-code-from-the-threadpool-to-a-thread"></a></p>
<h4 id="sending-code-from-the-threadpool-to-a-thread"><a class="header" href="#sending-code-from-the-threadpool-to-a-thread">Sending Code from the <code>ThreadPool</code> to a Thread</a></h4>
<p>We left a comment in the <code>for</code> loop in Listing 21-14 regarding the creation of
threads. Here, we’ll look at how we actually create threads. The standard
library provides <code>thread::spawn</code> as a way to create threads, and
<code>thread::spawn</code> expects to get some code the thread should run as soon as the
thread is created. However, in our case, we want to create the threads and have
them <em>wait</em> for code that we’ll send later. The standard library’s
implementation of threads doesn’t include any way to do that; we have to
implement it manually.</p>
<p>We’ll implement this behavior by introducing a new data structure between the
<code>ThreadPool</code> and the threads that will manage this new behavior. We’ll call
this data structure <em>Worker</em>, which is a common term in pooling
implementations. The <code>Worker</code> picks up code that needs to be run and runs the
code in its thread.</p>
<p>Think of people working in the kitchen at a restaurant: The workers wait until
orders come in from customers, and then they’re responsible for taking those
orders and filling them.</p>
<p>Instead of storing a vector of <code>JoinHandle&lt;()&gt;</code> instances in the thread pool,
we’ll store instances of the <code>Worker</code> struct. Each <code>Worker</code> will store a single
<code>JoinHandle&lt;()&gt;</code> instance. Then, we’ll implement a method on <code>Worker</code> that will
take a closure of code to run and send it to the already running thread for
execution. We’ll also give each <code>Worker</code> an <code>id</code> so that we can distinguish
between the different instances of <code>Worker</code> in the pool when logging or
debugging.</p>
<p>Here is the new process that will happen when we create a <code>ThreadPool</code>. We’ll
implement the code that sends the closure to the thread after we have <code>Worker</code>
set up in this way:</p>
<ol>
<li>Define a <code>Worker</code> struct that holds an <code>id</code> and a <code>JoinHandle&lt;()&gt;</code>.</li>
<li>Change <code>ThreadPool</code> to hold a vector of <code>Worker</code> instances.</li>
<li>Define a <code>Worker::new</code> function that takes an <code>id</code> number and returns a
<code>Worker</code> instance that holds the <code>id</code> and a thread spawned with an empty
closure.</li>
<li>In <code>ThreadPool::new</code>, use the <code>for</code> loop counter to generate an <code>id</code>, create
a new <code>Worker</code> with that <code>id</code>, and store the <code>Worker</code> in the vector.</li>
</ol>
<p>If you’re up for a challenge, try implementing these changes on your own before
looking at the code in Listing 21-15.</p>
<p>Ready? Here is Listing 21-15 with one way to make the preceding modifications.</p>
<figure class="listing" id="listing-21-15">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust noplayground">use std::thread;
pub struct ThreadPool {
workers: Vec&lt;Worker&gt;,
}
impl ThreadPool {
// --snip--
<span class="boring"> /// Create a new ThreadPool.
</span><span class="boring"> ///
</span><span class="boring"> /// The size is the number of threads in the pool.
</span><span class="boring"> ///
</span><span class="boring"> /// # Panics
</span><span class="boring"> ///
</span><span class="boring"> /// The `new` function will panic if the size is zero.
</span> pub fn new(size: usize) -&gt; ThreadPool {
assert!(size &gt; 0);
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id));
}
ThreadPool { workers }
}
// --snip--
<span class="boring">
</span><span class="boring"> pub fn execute&lt;F&gt;(&amp;self, f: F)
</span><span class="boring"> where
</span><span class="boring"> F: FnOnce() + Send + 'static,
</span><span class="boring"> {
</span><span class="boring"> }
</span>}
struct Worker {
id: usize,
thread: thread::JoinHandle&lt;()&gt;,
}
impl Worker {
fn new(id: usize) -&gt; Worker {
let thread = thread::spawn(|| {});
Worker { id, thread }
}
}</code></pre>
<figcaption><a href="#listing-21-15">Listing 21-15</a>: Modifying <code>ThreadPool</code> to hold <code>Worker</code> instances instead of holding threads directly</figcaption>
</figure>
<p>We’ve changed the name of the field on <code>ThreadPool</code> from <code>threads</code> to <code>workers</code>
because it’s now holding <code>Worker</code> instances instead of <code>JoinHandle&lt;()&gt;</code>
instances. We use the counter in the <code>for</code> loop as an argument to
<code>Worker::new</code>, and we store each new <code>Worker</code> in the vector named <code>workers</code>.</p>
<p>External code (like our server in <em>src/main.rs</em>) doesn’t need to know the
implementation details regarding using a <code>Worker</code> struct within <code>ThreadPool</code>,
so we make the <code>Worker</code> struct and its <code>new</code> function private. The
<code>Worker::new</code> function uses the <code>id</code> we give it and stores a <code>JoinHandle&lt;()&gt;</code>
instance that is created by spawning a new thread using an empty closure.</p>
<section class="note" aria-role="note">
<p>Note: If the operating system can’t create a thread because there aren’t
enough system resources, <code>thread::spawn</code> will panic. That will cause our
whole server to panic, even though the creation of some threads might
succeed. For simplicity’s sake, this behavior is fine, but in a production
thread pool implementation, you’d likely want to use
<a href="../std/thread/struct.Builder.html"><code>std::thread::Builder</code></a><!-- ignore --> and its
<a href="../std/thread/struct.Builder.html#method.spawn"><code>spawn</code></a><!-- ignore --> method that returns <code>Result</code> instead.</p>
</section>
<p>This code will compile and will store the number of <code>Worker</code> instances we
specified as an argument to <code>ThreadPool::new</code>. But we’re <em>still</em> not processing
the closure that we get in <code>execute</code>. Let’s look at how to do that next.</p>
<h4 id="sending-requests-to-threads-via-channels"><a class="header" href="#sending-requests-to-threads-via-channels">Sending Requests to Threads via Channels</a></h4>
<p>The next problem we’ll tackle is that the closures given to <code>thread::spawn</code> do
absolutely nothing. Currently, we get the closure we want to execute in the
<code>execute</code> method. But we need to give <code>thread::spawn</code> a closure to run when we
create each <code>Worker</code> during the creation of the <code>ThreadPool</code>.</p>
<p>We want the <code>Worker</code> structs that we just created to fetch the code to run from
a queue held in the <code>ThreadPool</code> and send that code to its thread to run.</p>
<p>The channels we learned about in Chapter 16—a simple way to communicate between
two threads—would be perfect for this use case. We’ll use a channel to function
as the queue of jobs, and <code>execute</code> will send a job from the <code>ThreadPool</code> to
the <code>Worker</code> instances, which will send the job to its thread. Here is the plan:</p>
<ol>
<li>The <code>ThreadPool</code> will create a channel and hold on to the sender.</li>
<li>Each <code>Worker</code> will hold on to the receiver.</li>
<li>We’ll create a new <code>Job</code> struct that will hold the closures we want to send
down the channel.</li>
<li>The <code>execute</code> method will send the job it wants to execute through the
sender.</li>
<li>In its thread, the <code>Worker</code> will loop over its receiver and execute the
closures of any jobs it receives.</li>
</ol>
<p>Let’s start by creating a channel in <code>ThreadPool::new</code> and holding the sender
in the <code>ThreadPool</code> instance, as shown in Listing 21-16. The <code>Job</code> struct
doesn’t hold anything for now but will be the type of item we’re sending down
the channel.</p>
<figure class="listing" id="listing-21-16">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust noplayground">use std::{sync::mpsc, thread};
pub struct ThreadPool {
workers: Vec&lt;Worker&gt;,
sender: mpsc::Sender&lt;Job&gt;,
}
struct Job;
impl ThreadPool {
// --snip--
<span class="boring"> /// Create a new ThreadPool.
</span><span class="boring"> ///
</span><span class="boring"> /// The size is the number of threads in the pool.
</span><span class="boring"> ///
</span><span class="boring"> /// # Panics
</span><span class="boring"> ///
</span><span class="boring"> /// The `new` function will panic if the size is zero.
</span> pub fn new(size: usize) -&gt; ThreadPool {
assert!(size &gt; 0);
let (sender, receiver) = mpsc::channel();
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id));
}
ThreadPool { workers, sender }
}
// --snip--
<span class="boring">
</span><span class="boring"> pub fn execute&lt;F&gt;(&amp;self, f: F)
</span><span class="boring"> where
</span><span class="boring"> F: FnOnce() + Send + 'static,
</span><span class="boring"> {
</span><span class="boring"> }
</span>}
<span class="boring">
</span><span class="boring">struct Worker {
</span><span class="boring"> id: usize,
</span><span class="boring"> thread: thread::JoinHandle&lt;()&gt;,
</span><span class="boring">}
</span><span class="boring">
</span><span class="boring">impl Worker {
</span><span class="boring"> fn new(id: usize) -&gt; Worker {
</span><span class="boring"> let thread = thread::spawn(|| {});
</span><span class="boring">
</span><span class="boring"> Worker { id, thread }
</span><span class="boring"> }
</span><span class="boring">}</span></code></pre>
<figcaption><a href="#listing-21-16">Listing 21-16</a>: Modifying <code>ThreadPool</code> to store the sender of a channel that transmits <code>Job</code> instances</figcaption>
</figure>
<p>In <code>ThreadPool::new</code>, we create our new channel and have the pool hold the
sender. This will successfully compile.</p>
<p>Let’s try passing a receiver of the channel into each <code>Worker</code> as the thread
pool creates the channel. We know we want to use the receiver in the thread that
the <code>Worker</code> instances spawn, so we’ll reference the <code>receiver</code> parameter in the
closure. The code in Listing 21-17 won’t quite compile yet.</p>
<figure class="listing" id="listing-21-17">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust ignore does_not_compile"><span class="boring">use std::{sync::mpsc, thread};
</span><span class="boring">
</span><span class="boring">pub struct ThreadPool {
</span><span class="boring"> workers: Vec&lt;Worker&gt;,
</span><span class="boring"> sender: mpsc::Sender&lt;Job&gt;,
</span><span class="boring">}
</span><span class="boring">
</span><span class="boring">struct Job;
</span><span class="boring">
</span>impl ThreadPool {
// --snip--
<span class="boring"> /// Create a new ThreadPool.
</span><span class="boring"> ///
</span><span class="boring"> /// The size is the number of threads in the pool.
</span><span class="boring"> ///
</span><span class="boring"> /// # Panics
</span><span class="boring"> ///
</span><span class="boring"> /// The `new` function will panic if the size is zero.
</span> pub fn new(size: usize) -&gt; ThreadPool {
assert!(size &gt; 0);
let (sender, receiver) = mpsc::channel();
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, receiver));
}
ThreadPool { workers, sender }
}
// --snip--
<span class="boring">
</span><span class="boring"> pub fn execute&lt;F&gt;(&amp;self, f: F)
</span><span class="boring"> where
</span><span class="boring"> F: FnOnce() + Send + 'static,
</span><span class="boring"> {
</span><span class="boring"> }
</span>}
// --snip--
<span class="boring">
</span><span class="boring">struct Worker {
</span><span class="boring"> id: usize,
</span><span class="boring"> thread: thread::JoinHandle&lt;()&gt;,
</span><span class="boring">}
</span><span class="boring">
</span>impl Worker {
fn new(id: usize, receiver: mpsc::Receiver&lt;Job&gt;) -&gt; Worker {
let thread = thread::spawn(|| {
receiver;
});
Worker { id, thread }
}
}</code></pre>
<figcaption><a href="#listing-21-17">Listing 21-17</a>: Passing the receiver to each <code>Worker</code></figcaption>
</figure>
<p>We’ve made some small and straightforward changes: We pass the receiver into
<code>Worker::new</code>, and then we use it inside the closure.</p>
<p>When we try to check this code, we get this error:</p>
<pre><code class="language-console">$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0382]: use of moved value: `receiver`
--&gt; src/lib.rs:26:42
|
21 | let (sender, receiver) = mpsc::channel();
| -------- move occurs because `receiver` has type `std::sync::mpsc::Receiver&lt;Job&gt;`, which does not implement the `Copy` trait
...
25 | for id in 0..size {
| ----------------- inside of this loop
26 | workers.push(Worker::new(id, receiver));
| ^^^^^^^^ value moved here, in previous iteration of loop
|
note: consider changing this parameter type in method `new` to borrow instead if owning the value isn't necessary
--&gt; src/lib.rs:47:33
|
47 | fn new(id: usize, receiver: mpsc::Receiver&lt;Job&gt;) -&gt; Worker {
| --- in this method ^^^^^^^^^^^^^^^^^^^ this parameter takes ownership of the value
help: consider moving the expression out of the loop so it is only moved once
|
25 ~ let mut value = Worker::new(id, receiver);
26 ~ for id in 0..size {
27 ~ workers.push(value);
|
For more information about this error, try `rustc --explain E0382`.
error: could not compile `hello` (lib) due to 1 previous error
</code></pre>
<p>The code is trying to pass <code>receiver</code> to multiple <code>Worker</code> instances. This
won’t work, as you’ll recall from Chapter 16: The channel implementation that
Rust provides is multiple <em>producer</em>, single <em>consumer</em>. This means we can’t
just clone the consuming end of the channel to fix this code. We also don’t
want to send a message multiple times to multiple consumers; we want one list
of messages with multiple <code>Worker</code> instances such that each message gets
processed once.</p>
<p>Additionally, taking a job off the channel queue involves mutating the
<code>receiver</code>, so the threads need a safe way to share and modify <code>receiver</code>;
otherwise, we might get race conditions (as covered in Chapter 16).</p>
<p>Recall the thread-safe smart pointers discussed in Chapter 16: To share
ownership across multiple threads and allow the threads to mutate the value, we
need to use <code>Arc&lt;Mutex&lt;T&gt;&gt;</code>. The <code>Arc</code> type will let multiple <code>Worker</code> instances
own the receiver, and <code>Mutex</code> will ensure that only one <code>Worker</code> gets a job from
the receiver at a time. Listing 21-18 shows the changes we need to make.</p>
<figure class="listing" id="listing-21-18">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust noplayground">use std::{
sync::{Arc, Mutex, mpsc},
thread,
};
// --snip--
<span class="boring">pub struct ThreadPool {
</span><span class="boring"> workers: Vec&lt;Worker&gt;,
</span><span class="boring"> sender: mpsc::Sender&lt;Job&gt;,
</span><span class="boring">}
</span><span class="boring">
</span><span class="boring">struct Job;
</span><span class="boring">
</span>impl ThreadPool {
// --snip--
<span class="boring"> /// Create a new ThreadPool.
</span><span class="boring"> ///
</span><span class="boring"> /// The size is the number of threads in the pool.
</span><span class="boring"> ///
</span><span class="boring"> /// # Panics
</span><span class="boring"> ///
</span><span class="boring"> /// The `new` function will panic if the size is zero.
</span> pub fn new(size: usize) -&gt; ThreadPool {
assert!(size &gt; 0);
let (sender, receiver) = mpsc::channel();
let receiver = Arc::new(Mutex::new(receiver));
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, Arc::clone(&amp;receiver)));
}
ThreadPool { workers, sender }
}
// --snip--
<span class="boring">
</span><span class="boring"> pub fn execute&lt;F&gt;(&amp;self, f: F)
</span><span class="boring"> where
</span><span class="boring"> F: FnOnce() + Send + 'static,
</span><span class="boring"> {
</span><span class="boring"> }
</span>}
// --snip--
<span class="boring">struct Worker {
</span><span class="boring"> id: usize,
</span><span class="boring"> thread: thread::JoinHandle&lt;()&gt;,
</span><span class="boring">}
</span><span class="boring">
</span>impl Worker {
fn new(id: usize, receiver: Arc&lt;Mutex&lt;mpsc::Receiver&lt;Job&gt;&gt;&gt;) -&gt; Worker {
// --snip--
<span class="boring"> let thread = thread::spawn(|| {
</span><span class="boring"> receiver;
</span><span class="boring"> });
</span><span class="boring">
</span><span class="boring"> Worker { id, thread }
</span> }
}</code></pre>
<figcaption><a href="#listing-21-18">Listing 21-18</a>: Sharing the receiver among the <code>Worker</code> instances using <code>Arc</code> and <code>Mutex</code></figcaption>
</figure>
<p>In <code>ThreadPool::new</code>, we put the receiver in an <code>Arc</code> and a <code>Mutex</code>. For each
new <code>Worker</code>, we clone the <code>Arc</code> to bump the reference count so that the
<code>Worker</code> instances can share ownership of the receiver.</p>
<p>With these changes, the code compiles! We’re getting there!</p>
<h4 id="implementing-the-execute-method"><a class="header" href="#implementing-the-execute-method">Implementing the <code>execute</code> Method</a></h4>
<p>Let’s finally implement the <code>execute</code> method on <code>ThreadPool</code>. We’ll also change
<code>Job</code> from a struct to a type alias for a trait object that holds the type of
closure that <code>execute</code> receives. As discussed in the <a href="../ch20/ch20-03-advanced-types.html#type-synonyms-and-type-aliases">“Type Synonyms and Type
Aliases”</a><!-- ignore --> section in Chapter 20, type aliases
allow us to make long types shorter for ease of use. Look at Listing 21-19.</p>
<figure class="listing" id="listing-21-19">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust noplayground"><span class="boring">use std::{
</span><span class="boring"> sync::{Arc, Mutex, mpsc},
</span><span class="boring"> thread,
</span><span class="boring">};
</span><span class="boring">
</span><span class="boring">pub struct ThreadPool {
</span><span class="boring"> workers: Vec&lt;Worker&gt;,
</span><span class="boring"> sender: mpsc::Sender&lt;Job&gt;,
</span><span class="boring">}
</span><span class="boring">
</span>// --snip--
type Job = Box&lt;dyn FnOnce() + Send + 'static&gt;;
impl ThreadPool {
// --snip--
<span class="boring"> /// Create a new ThreadPool.
</span><span class="boring"> ///
</span><span class="boring"> /// The size is the number of threads in the pool.
</span><span class="boring"> ///
</span><span class="boring"> /// # Panics
</span><span class="boring"> ///
</span><span class="boring"> /// The `new` function will panic if the size is zero.
</span><span class="boring"> pub fn new(size: usize) -&gt; ThreadPool {
</span><span class="boring"> assert!(size &gt; 0);
</span><span class="boring">
</span><span class="boring"> let (sender, receiver) = mpsc::channel();
</span><span class="boring">
</span><span class="boring"> let receiver = Arc::new(Mutex::new(receiver));
</span><span class="boring">
</span><span class="boring"> let mut workers = Vec::with_capacity(size);
</span><span class="boring">
</span><span class="boring"> for id in 0..size {
</span><span class="boring"> workers.push(Worker::new(id, Arc::clone(&amp;receiver)));
</span><span class="boring"> }
</span><span class="boring">
</span><span class="boring"> ThreadPool { workers, sender }
</span><span class="boring"> }
</span>
pub fn execute&lt;F&gt;(&amp;self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.send(job).unwrap();
}
}
// --snip--
<span class="boring">
</span><span class="boring">struct Worker {
</span><span class="boring"> id: usize,
</span><span class="boring"> thread: thread::JoinHandle&lt;()&gt;,
</span><span class="boring">}
</span><span class="boring">
</span><span class="boring">impl Worker {
</span><span class="boring"> fn new(id: usize, receiver: Arc&lt;Mutex&lt;mpsc::Receiver&lt;Job&gt;&gt;&gt;) -&gt; Worker {
</span><span class="boring"> let thread = thread::spawn(|| {
</span><span class="boring"> receiver;
</span><span class="boring"> });
</span><span class="boring">
</span><span class="boring"> Worker { id, thread }
</span><span class="boring"> }
</span><span class="boring">}</span></code></pre>
<figcaption><a href="#listing-21-19">Listing 21-19</a>: Creating a <code>Job</code> type alias for a <code>Box</code> that holds each closure and then sending the job down the channel</figcaption>
</figure>
<p>After creating a new <code>Job</code> instance using the closure we get in <code>execute</code>, we
send that job down the sending end of the channel. We’re calling <code>unwrap</code> on
<code>send</code> for the case that sending fails. This might happen if, for example, we
stop all our threads from executing, meaning the receiving end has stopped
receiving new messages. At the moment, we can’t stop our threads from
executing: Our threads continue executing as long as the pool exists. The
reason we use <code>unwrap</code> is that we know the failure case won’t happen, but the
compiler doesn’t know that.</p>
<p>But we’re not quite done yet! In the <code>Worker</code>, our closure being passed to
<code>thread::spawn</code> still only <em>references</em> the receiving end of the channel.
Instead, we need the closure to loop forever, asking the receiving end of the
channel for a job and running the job when it gets one. Let’s make the change
shown in Listing 21-20 to <code>Worker::new</code>.</p>
<figure class="listing" id="listing-21-20">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust noplayground"><span class="boring">use std::{
</span><span class="boring"> sync::{Arc, Mutex, mpsc},
</span><span class="boring"> thread,
</span><span class="boring">};
</span><span class="boring">
</span><span class="boring">pub struct ThreadPool {
</span><span class="boring"> workers: Vec&lt;Worker&gt;,
</span><span class="boring"> sender: mpsc::Sender&lt;Job&gt;,
</span><span class="boring">}
</span><span class="boring">
</span><span class="boring">type Job = Box&lt;dyn FnOnce() + Send + 'static&gt;;
</span><span class="boring">
</span><span class="boring">impl ThreadPool {
</span><span class="boring"> /// Create a new ThreadPool.
</span><span class="boring"> ///
</span><span class="boring"> /// The size is the number of threads in the pool.
</span><span class="boring"> ///
</span><span class="boring"> /// # Panics
</span><span class="boring"> ///
</span><span class="boring"> /// The `new` function will panic if the size is zero.
</span><span class="boring"> pub fn new(size: usize) -&gt; ThreadPool {
</span><span class="boring"> assert!(size &gt; 0);
</span><span class="boring">
</span><span class="boring"> let (sender, receiver) = mpsc::channel();
</span><span class="boring">
</span><span class="boring"> let receiver = Arc::new(Mutex::new(receiver));
</span><span class="boring">
</span><span class="boring"> let mut workers = Vec::with_capacity(size);
</span><span class="boring">
</span><span class="boring"> for id in 0..size {
</span><span class="boring"> workers.push(Worker::new(id, Arc::clone(&amp;receiver)));
</span><span class="boring"> }
</span><span class="boring">
</span><span class="boring"> ThreadPool { workers, sender }
</span><span class="boring"> }
</span><span class="boring">
</span><span class="boring"> pub fn execute&lt;F&gt;(&amp;self, f: F)
</span><span class="boring"> where
</span><span class="boring"> F: FnOnce() + Send + 'static,
</span><span class="boring"> {
</span><span class="boring"> let job = Box::new(f);
</span><span class="boring">
</span><span class="boring"> self.sender.send(job).unwrap();
</span><span class="boring"> }
</span><span class="boring">}
</span><span class="boring">
</span><span class="boring">struct Worker {
</span><span class="boring"> id: usize,
</span><span class="boring"> thread: thread::JoinHandle&lt;()&gt;,
</span><span class="boring">}
</span><span class="boring">
</span>// --snip--
impl Worker {
fn new(id: usize, receiver: Arc&lt;Mutex&lt;mpsc::Receiver&lt;Job&gt;&gt;&gt;) -&gt; Worker {
let thread = thread::spawn(move || {
loop {
let job = receiver.lock().unwrap().recv().unwrap();
println!("Worker {id} got a job; executing.");
job();
}
});
Worker { id, thread }
}
}</code></pre>
<figcaption><a href="#listing-21-20">Listing 21-20</a>: Receiving and executing the jobs in the <code>Worker</code> instance’s thread</figcaption>
</figure>
<p>Here, we first call <code>lock</code> on the <code>receiver</code> to acquire the mutex, and then we
call <code>unwrap</code> to panic on any errors. Acquiring a lock might fail if the mutex
is in a <em>poisoned</em> state, which can happen if some other thread panicked while
holding the lock rather than releasing the lock. In this situation, calling
<code>unwrap</code> to have this thread panic is the correct action to take. Feel free to
change this <code>unwrap</code> to an <code>expect</code> with an error message that is meaningful to
you.</p>
<p>If we get the lock on the mutex, we call <code>recv</code> to receive a <code>Job</code> from the
channel. A final <code>unwrap</code> moves past any errors here as well, which might occur
if the thread holding the sender has shut down, similar to how the <code>send</code>
method returns <code>Err</code> if the receiver shuts down.</p>
<p>The call to <code>recv</code> blocks, so if there is no job yet, the current thread will
wait until a job becomes available. The <code>Mutex&lt;T&gt;</code> ensures that only one
<code>Worker</code> thread at a time is trying to request a job.</p>
<p>Our thread pool is now in a working state! Give it a <code>cargo run</code> and make some
requests:</p>
<!-- manual-regeneration
cd listings/ch21-web-server/listing-21-20
cargo run
make some requests to 127.0.0.1:7878
Can't automate because the output depends on making requests
-->
<pre><code class="language-console">$ cargo run
Compiling hello v0.1.0 (file:///projects/hello)
warning: field `workers` is never read
--&gt; src/lib.rs:7:5
|
6 | pub struct ThreadPool {
| ---------- field in this struct
7 | workers: Vec&lt;Worker&gt;,
| ^^^^^^^
|
= note: `#[warn(dead_code)]` on by default
warning: fields `id` and `thread` are never read
--&gt; src/lib.rs:48:5
|
47 | struct Worker {
| ------ fields in this struct
48 | id: usize,
| ^^
49 | thread: thread::JoinHandle&lt;()&gt;,
| ^^^^^^
warning: `hello` (lib) generated 2 warnings
Finished `dev` profile [unoptimized + debuginfo] target(s) in 4.91s
Running `target/debug/hello`
Worker 0 got a job; executing.
Worker 2 got a job; executing.
Worker 1 got a job; executing.
Worker 3 got a job; executing.
Worker 0 got a job; executing.
Worker 2 got a job; executing.
Worker 1 got a job; executing.
Worker 3 got a job; executing.
Worker 0 got a job; executing.
Worker 2 got a job; executing.
</code></pre>
<p>Success! We now have a thread pool that executes connections asynchronously.
There are never more than four threads created, so our system won’t get
overloaded if the server receives a lot of requests. If we make a request to
<em>/sleep</em>, the server will be able to serve other requests by having another
thread run them.</p>
<section class="note" aria-role="note">
<p>Note: If you open <em>/sleep</em> in multiple browser windows simultaneously, they
might load one at a time in five-second intervals. Some web browsers execute
multiple instances of the same request sequentially for caching reasons. This
limitation is not caused by our web server.</p>
</section>
<p>This is a good time to pause and consider how the code in Listings 21-18, 21-19,
and 21-20 would be different if we were using futures instead of a closure for
the work to be done. What types would change? How would the method signatures be
different, if at all? What parts of the code would stay the same?</p>
<p>After learning about the <code>while let</code> loop in Chapter 17 and Chapter 19, you
might be wondering why we didn’t write the <code>Worker</code> thread code as shown in
Listing 21-21.</p>
<figure class="listing" id="listing-21-21">
<span class="file-name">Filename: src/lib.rs</span>
<pre><code class="language-rust ignore not_desired_behavior"><span class="boring">use std::{
</span><span class="boring"> sync::{Arc, Mutex, mpsc},
</span><span class="boring"> thread,
</span><span class="boring">};
</span><span class="boring">
</span><span class="boring">pub struct ThreadPool {
</span><span class="boring"> workers: Vec&lt;Worker&gt;,
</span><span class="boring"> sender: mpsc::Sender&lt;Job&gt;,
</span><span class="boring">}
</span><span class="boring">
</span><span class="boring">type Job = Box&lt;dyn FnOnce() + Send + 'static&gt;;
</span><span class="boring">
</span><span class="boring">impl ThreadPool {
</span><span class="boring"> /// Create a new ThreadPool.
</span><span class="boring"> ///
</span><span class="boring"> /// The size is the number of threads in the pool.
</span><span class="boring"> ///
</span><span class="boring"> /// # Panics
</span><span class="boring"> ///
</span><span class="boring"> /// The `new` function will panic if the size is zero.
</span><span class="boring"> pub fn new(size: usize) -&gt; ThreadPool {
</span><span class="boring"> assert!(size &gt; 0);
</span><span class="boring">
</span><span class="boring"> let (sender, receiver) = mpsc::channel();
</span><span class="boring">
</span><span class="boring"> let receiver = Arc::new(Mutex::new(receiver));
</span><span class="boring">
</span><span class="boring"> let mut workers = Vec::with_capacity(size);
</span><span class="boring">
</span><span class="boring"> for id in 0..size {
</span><span class="boring"> workers.push(Worker::new(id, Arc::clone(&amp;receiver)));
</span><span class="boring"> }
</span><span class="boring">
</span><span class="boring"> ThreadPool { workers, sender }
</span><span class="boring"> }
</span><span class="boring">
</span><span class="boring"> pub fn execute&lt;F&gt;(&amp;self, f: F)
</span><span class="boring"> where
</span><span class="boring"> F: FnOnce() + Send + 'static,
</span><span class="boring"> {
</span><span class="boring"> let job = Box::new(f);
</span><span class="boring">
</span><span class="boring"> self.sender.send(job).unwrap();
</span><span class="boring"> }
</span><span class="boring">}
</span><span class="boring">
</span><span class="boring">struct Worker {
</span><span class="boring"> id: usize,
</span><span class="boring"> thread: thread::JoinHandle&lt;()&gt;,
</span><span class="boring">}
</span>// --snip--
impl Worker {
fn new(id: usize, receiver: Arc&lt;Mutex&lt;mpsc::Receiver&lt;Job&gt;&gt;&gt;) -&gt; Worker {
let thread = thread::spawn(move || {
while let Ok(job) = receiver.lock().unwrap().recv() {
println!("Worker {id} got a job; executing.");
job();
}
});
Worker { id, thread }
}
}</code></pre>
<figcaption><a href="#listing-21-21">Listing 21-21</a>: An alternative implementation of <code>Worker::new</code> using <code>while let</code></figcaption>
</figure>
<p>This code compiles and runs but doesn’t result in the desired threading
behavior: A slow request will still cause other requests to wait to be
processed. The reason is somewhat subtle: The <code>Mutex</code> struct has no public
<code>unlock</code> method because the ownership of the lock is based on the lifetime of
the <code>MutexGuard&lt;T&gt;</code> within the <code>LockResult&lt;MutexGuard&lt;T&gt;&gt;</code> that the <code>lock</code>
method returns. At compile time, the borrow checker can then enforce the rule
that a resource guarded by a <code>Mutex</code> cannot be accessed unless we hold the
lock. However, this implementation can also result in the lock being held
longer than intended if we aren’t mindful of the lifetime of the
<code>MutexGuard&lt;T&gt;</code>.</p>
<p>The code in Listing 21-20 that uses <code>let job = receiver.lock().unwrap().recv().unwrap();</code> works because with <code>let</code>, any
temporary values used in the expression on the right-hand side of the equal
sign are immediately dropped when the <code>let</code> statement ends. However, <code>while let</code> (and <code>if let</code> and <code>match</code>) does not drop temporary values until the end of
the associated block. In Listing 21-21, the lock remains held for the duration
of the call to <code>job()</code>, meaning other <code>Worker</code> instances cannot receive jobs.</p>
</body>
</html>
Reference in New Issue View Git Blame Copy Permalink