owned this note
owned this note
Published
Linked with GitHub
# Building a Concurrent Web Server with Async Rust
In this chapter, we'll use asynchronous Rust to modify the Rust book's
[single-threaded web server](https://doc.rust-lang.org/book/ch20-01-single-threaded.html)
to serve requests concurrently.
Here's what the code looked like at the end of the lesson:
`src/main.rs`:
```rust
use std::fs;
use std::io::prelude::*;
use std::net::TcpListener;
use std::net::TcpStream;
fn main() {
// Listen for incoming TCP connections on localhost port 7878
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
// Block forever, handling each request that arrives at this IP address
for stream in listener.incoming() {
let stream = stream.unwrap();
handle_connection(stream);
}
}
fn handle_connection(mut stream: TcpStream) {
// Read the first 1024 bytes of data from the stream
let mut buffer = [0; 1024];
stream.read(&mut buffer).unwrap();
let get = b"GET / HTTP/1.1\r\n";
// Respond with greetings or a 404,
// depending on the data in the request
let (status_line, filename) = if buffer.starts_with(get) {
("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
} else {
("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html")
};
let contents = fs::read_to_string(filename).unwrap();
// Write response back to the stream,
// and flush the stream to ensure the response is sent back to the client
let response = format!("{}{}", status_line, contents);
stream.write(response.as_bytes()).unwrap();
stream.flush().unwrap();
}
```
`hello.html`:
```html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>Hello!</title>
</head>
<body>
<h1>Hello!</h1>
<p>Hi from Rust</p>
</body>
</html>
```
`404.html`:
```html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>Hello!</title>
</head>
<body>
<h1>Oops!</h1>
<p>Sorry, I don't know what you're asking for.</p>
</body>
</html>
```
If you run the server with `cargo run` and visit `127.0.0.1:7878` in your browser,
you'll be greeted with a friendly message from Ferris!
## Running Asynchronous Code
As [the book explains](https://doc.rust-lang.org/book/ch20-02-multithreaded.html#turning-our-single-threaded-server-into-a-multithreaded-server),
we don't want our web server to wait for each request to finish before handling the next,
as some requests could be very slow.
Instead of improving throughput by adding threads,
we'll use asynchronous code to process requests concurrently.
Let's modify `handle_connection` to return a future by declaring it an `async fn`:
```rust
async fn handle_connection(mut stream: TcpStream) {
// <-- snip -->
}
```
Adding `async` to the function declaration changes its return type
from the unit type `()` to a type that implements `Future<Output=()>`.
If we try to compile this, the compiler warns us that it will not work:
```
$ cargo check
Checking async-rust v0.1.0 (file:///projects/async-rust)
warning: unused implementer of `std::future::Future` that must be used
--> src/main.rs:12:9
|
12 | handle_connection(stream);
| ^^^^^^^^^^^^^^^^^^^^^^^^^^
|
= note: `#[warn(unused_must_use)]` on by default
= note: futures do nothing unless you `.await` or poll them
```
Because we haven't `await`ed or `poll`ed the result of `handle_connection`,
it'll never run. If you run the server and visit `127.0.0.1:7878` in a browser,
you'll see that the connection is refused; our server is not handling requests.
We can't `await` or `poll` futures within synchronous code by itself.
We'll need an executor to handle scheduling and running futures to completion.
Please consult the section [Choosing an Executor](./relative_link) for more information on executors.
Here, we'll use the `run` executor from the `smol` crate.
It might be tempting to write something like this:
```rust
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
smol::run(async {
for stream in listener.incoming() {
let stream = stream.unwrap();
// Warning: This is not concurrent!
handle_connection(stream).await;
}
})
}
async fn handle_connection(mut stream: TcpStream) {
// <-- snip -->
}
```
However, just because this program uses an asynchronous connection handler
doesn't mean that it handles connections concurrently.
To illustrate this, try out the
[simulation of a slow request](https://doc.rust-lang.org/book/ch20-02-multithreaded.html#simulating-a-slow-request-in-the-current-server-implementation)
from the Book. You'll see that one slow request will block any other incoming requests!
This is because there are no other concurrent tasks that can make progress
while we are `await`ing the result of `handle_connection`.
## Handling Connections Concurrently
The problem with our code so far is that `listener.incoming()` is a blocking iterator;
we can't read a new request from this stream until we're done with the previous one.
One strategy to work around this is to spawn a new Task to handle each connection in the background:
```rust
extern crate smol;
use smol::Task;
smol::run(async {
for stream in listener.incoming() {
let stream = stream.unwrap();
Task::spawn(handle_connection(stream)).detach();
}
})
```
This works because under the hood, the `smol` executor runs `handle_connection` on a separate thread.
However, this doesn't completely solve our problem: `listener.incoming()` still blocks the executor.
Even if connections are handled in separate threads, futures running on the main thread
are blocked while `listener` waits on incoming connections.
In order to fix this, we can use the `smol::Async` trait to transform our `TcpListener`
into an asynchronous `TcpListener`:
```rust
let listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 7878)).unwrap();
```
This change prevents `listener.incoming()` from blocking the executor
by allowing us to `await` the next TCP connection on this port.
Now, the executor can yield to other futures running on the main thread
while there are no incoming TCP connections to be processed.
(Note that this trait still does *not* allow the stream to emit items concurrently.
We still need to process a stream or spawn a task to handle it before moving on to the next one.)
Let's update our example to make use of the `Async<TcpListener>`.
First, we'll need to update our code to `await` the next incoming connection,
rather than iterating over `listener.incoming()`:
```rust
fn main() {
smol::run(async {
let listener = Async::<TcpListener>::bind(([127, 0, 0, 1], 7878)).unwrap();
loop {
let (stream, _) = listener.accept().await.unwrap();
Task::spawn(async { handle_connection(stream).await }).detach();
}
})
}
```
Lastly, we'll have to update our connection handler to accept an `Async::<TcpStream>`:
```rust
async fn handle_connection(mut stream: Async<TcpStream>) {
let mut buffer = [0; 1024];
stream.read(&mut buffer).await.unwrap();
// <-- snip -->
stream.write(response.as_bytes()).await.unwrap();
stream.flush().await.unwrap();
```
## Testing Async Code
Let's move on to testing our `handle_connection` function.
First, we need a `TcpStream` to work with, but we don't want to make a real TCP connection in test code.
We could work around this in a few ways.
One strategy could be to refactor the code to be more modular,
and only test that the correct responses are returned for the respective inputs.
Another strategy is to connect to `localhost` on port 0.
Port 0 isn't a valid UNIX port, but it'll work for testing.
The operating system will return a connection on any open TCP port.
```rust
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
let mut stream = TcpStream::connect(addr).unwrap();
```
Instead of those strategies, we'll change the signature of `handle_connection` to make it easier to test.
`handle_connection` doesn't actually require an `Async<TcpStream>`;
it requires any struct that implements `AsyncRead`, `AsyncWrite`, and `Unpin`.
Changing the type signature to reflect this allows us to pass a mock for testing instead of a TcpStream.
```rust
async fn handle_connection(mut stream: impl AsyncRead + AsyncWrite + Unpin) {
// <-- snip -->
}
```
Next, let's create a mock TcpStream with an underlying `Vec` representing its contents,
and implement `AsyncRead`, `AsyncWrite`, and `Unpin`.
(For more information on pinning and the `Unpin` trait, see the [section on pinning](insert link).)
```rust
struct MockTcpStream {
contents: Vec<u8>,
}
impl AsyncRead for MockTcpStream {
fn poll_read(self: Pin<&mut Self>, _: &mut Context, buf: &mut [u8]) -> Poll<Result<usize, Error>> {
let size: usize;
if self.contents.len() >= buf.len() {
buf.clone_from_slice(&self.contents[..buf.len()]);
size = buf.len();
} else {
buf[0..self.contents.len()].clone_from_slice(&self.contents[..]);
size = self.contents.len();
}
Poll::Ready(Ok(size))
}
}
impl AsyncWrite for MockTcpStream {
fn poll_write(mut self: Pin<&mut Self>, _: &mut Context, buf: &[u8]) -> Poll<Result<usize, Error>> {
let size: usize;
if self.contents.len() >= buf.len() {
size = buf.len();
self.contents[0..size].clone_from_slice(buf);
} else {
size = self.contents.len();
self.contents.clone_from_slice(&buf[..size]);
}
return Poll::Ready(Ok(size));
}
fn poll_flush(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Error>> {
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Error>> {
Poll::Ready(Ok(()))
}
}
impl Unpin for MockTcpStream {}
```
Now we're ready to test the `handle_connection` function.
After setting up the `MockTcpStream` containing some initial data,
we can run `handle_connection` using the `smol` executor, exactly as we did in the main method.
```rust
#[cfg(test)]
mod tests {
// <-- snip -->
struct MockTcpStream {
contents: Vec<u8>,
}
impl AsyncRead for MockTcpStream {
// <-- snip -->
}
impl AsyncWrite for MockTcpStream {
// <-- snip -->
}
impl Unpin for MockTcpStream {}
#[test]
fn test_handle_connection() {
let input_bytes = b"GET / HTTP/1.1\r\n";
let mut contents = vec![0u8; 1024];
contents[..input_bytes.len()].clone_from_slice(input_bytes);
let mut stream = MockTcpStream { contents };
smol::run(async {
handle_connection(&mut stream).await;
let mut buf = [0u8; 1024];
stream.read(&mut buf).await.unwrap();
});
let expected_contents = fs::read_to_string("hello.html").unwrap();
let expected_response = format!("HTTP/1.1 200 OK\r\n\r\n{}", expected_contents);
assert!(stream.contents.starts_with(expected_response.as_bytes()));
}
}
```
Google Drive
Gist
Clipboard
Sign in with Wallet
