Try   HackMD

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 to serve requests concurrently. Here's what the code looked like at the end of the lesson:

src/main.rs:

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:

<!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:

<!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, 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:

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 awaited or polled 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 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:

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 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 awaiting 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:

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:

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():

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>:

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.

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.

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).)

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.

#[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()));
    }
}
Last changed by 
 
lbernick
0
3407
Published on HackMD