Spawning

We are going to shift gears and start working on the Redis server.

First, move the client SET/GET code from the previous section to an example file. This way, we can run it against our server.

  1. $ mkdir -p examples
  2. $ mv src/main.rs examples/hello-redis.rs

Then create a new, empty src/main.rs and continue.

Accepting sockets

The first thing our Redis server needs to do is to accept inbound TCP sockets. This is done with tokio::net::TcpListener.

Many of Tokio’s types are named the same as their synchronous equivalent in the Rust standard library. When it makes sense, Tokio exposes the same APIs as std but using async fn.

A TcpListener is bound to port 6379, then sockets are accepted in a loop. Each socket is processed then closed. For now, we will read the command, print it to stdout and respond with an error.

  1. use tokio::net::{TcpListener, TcpStream};
  2. use mini_redis::{Connection, Frame};
  4. #[tokio::main]
  5. async fn main() {
  6.     // Bind the listener to the address
  7.     let listener = TcpListener::bind("127.0.0.1:6379").await.unwrap();
  9.     loop {
  10.         // The second item contains the IP and port of the new connection.
  11.         let (socket, _) = listener.accept().await.unwrap();
  12.         process(socket).await;
  13.     }
  14. }
  16. async fn process(socket: TcpStream) {
  17.     // The `Connection` lets us read/write redis **frames** instead of
  18.     // byte streams. The `Connection` type is defined by mini-redis.
  19.     let mut connection = Connection::new(socket);
  21.     if let Some(frame) = connection.read_frame().await.unwrap() {
  22.         println!("GOT: {:?}", frame);
  24.         // Respond with an error
  25.         let response = Frame::Error("unimplemented".to_string());
  26.         connection.write_frame(&response).await.unwrap();
  27.     }
  28. }

Now, run this accept loop:

  1. $ cargo run

In a separate terminal window, run the hello-redis example (the SET/GET command from the previous section):

  1. $ cargo run --example hello-redis

The output should be:

  1. Error: "unimplemented"

In the server terminal, the output is:

  1. GOT: Array([Bulk(b"set"), Bulk(b"hello"), Bulk(b"world")])

Concurrency

Our server has a slight problem (besides only responding with errors). It processes inbound requests one at a time. When a connection is accepted, the server stays inside the accept loop block until the response is fully written to the socket.

We want our Redis server to process many concurrent requests. To do this, we need to add some concurrency.

Concurrency and parallelism is not the same thing. If you alternate between two tasks, then you are working on both tasks concurrently, but not in parallel. For it to qualify as parallel, you would need two people, one dedicated to each task.

One of the advantages of using Tokio is that asynchronous code allows you to work on many tasks concurrently, without having to work on them in parallel using ordinary threads. In fact, Tokio can run many tasks concurrently on a single thread!

To process connections concurrently, a new task is spawned for each inbound connection. The connection is processed on this task.

The accept loop becomes:

  1. use tokio::net::TcpListener;
  3. #[tokio::main]
  4. async fn main() {
  5.     let listener = TcpListener::bind("127.0.0.1:6379").await.unwrap();
  7.     loop {
  8.         let (socket, _) = listener.accept().await.unwrap();
  9.         // A new task is spawned for each inbound socket. The socket is
  10.         // moved to the new task and processed there.
  11.         tokio::spawn(async move {
  12.             process(socket).await;
  13.         });
  14.     }
  15. }

Tasks

A Tokio task is an asynchronous green thread. They are created by passing an async block to tokio::spawn. The tokio::spawn function returns a JoinHandle, which the caller may use to interact with the spawned task. The async block may have a return value. The caller may obtain the return value using .await on the JoinHandle.

For example:

  1. #[tokio::main]
  2. async fn main() {
  3.     let handle = tokio::spawn(async {
  4.         // Do some async work
  5.         "return value"
  6.     });
  8.     // Do some other work
  10.     let out = handle.await.unwrap();
  11.     println!("GOT {}", out);
  12. }

Awaiting on JoinHandle returns a Result. When a task encounters an error during execution, the JoinHandle will return an Err. This happens when the task either panics, or if the task is forcefully cancelled by the runtime shutting down.

Tasks are the unit of execution managed by the scheduler. Spawning the task submits it to the Tokio scheduler, which then ensures that the task executes when it has work to do. The spawned task may be executed on the same thread as where it was spawned, or it may execute on a different runtime thread. The task can also be moved between threads after being spawned.

Tasks in Tokio are very lightweight. Under the hood, they require only a single allocation and 64 bytes of memory. Applications should feel free to spawn thousands, if not millions of tasks.

'static bound

When you spawn a task on the Tokio runtime, its type must be 'static. This means that the spawned task must not contain any references to data owned outside the task.

It is a common misconception that 'static always means “lives forever”, but this is not the case. Just because a value is 'static does not mean that you have a memory leak. You can read more in Common Rust Lifetime Misconceptions.

For example, the following will not compile:

  1. use tokio::task;
  3. #[tokio::main]
  4. async fn main() {
  5.     let v = vec![1, 2, 3];
  7.     task::spawn(async {
  8.         println!("Here's a vec: {:?}", v);
  9.     });
  10. }

Attempting to compile this results in the following error:

  1. error[E0373]: async block may outlive the current function, but
  2.               it borrows `v`, which is owned by the current function
  3.  --> src/main.rs:7:23
  4.   |
  5. 7 |       task::spawn(async {
  6.   |  _______________________^
  7. 8 | |         println!("Here's a vec: {:?}", v);
  8.   | |                                        - `v` is borrowed here
  9. 9 | |     });
  10.   | |_____^ may outlive borrowed value `v`
  11.   |
  12. note: function requires argument type to outlive `'static`
  13.  --> src/main.rs:7:17
  14.   |
  15. 7 |       task::spawn(async {
  16.   |  _________________^
  17. 8 | |         println!("Here's a vector: {:?}", v);
  18. 9 | |     });
  19.   | |_____^
  20. help: to force the async block to take ownership of `v` (and any other
  21.       referenced variables), use the `move` keyword
  22.   |
  23. 7 |     task::spawn(async move {
  24. 8 |         println!("Here's a vec: {:?}", v);
  25. 9 |     });
  26.   |

This happens because, by default, variables are not moved into async blocks. The v vector remains owned by the main function. The println! line borrows v. The rust compiler helpfully explains this to us and even suggests the fix! Changing line 7 to task::spawn(async move { will instruct the compiler to move v into the spawned task. Now, the task owns all of its data, making it 'static.

If a single piece of data must be accessible from more than one task concurrently, then it must be shared using synchronization primitives such as Arc.

Note that the error message talks about the argument type outliving the 'static lifetime. This terminology can be rather confusing because the 'static lifetime lasts until the end of the program, so if it outlives it, don’t you have a memory leak? The explanation is that it is the type, not the value that must outlive the 'static lifetime, and the value may be destroyed before its type is no longer valid.

When we say that a value is 'static, all that means is that it would not be incorrect to keep that value around forever. This is important because the compiler is unable to reason about how long a newly spawned task stays around, so the only way it can be sure that the task doesn’t live too long is to make sure it may live forever.

The link in the info-box above uses the terminology “bounded by 'static“ rather than “its type outlives 'static“ or “the value is 'static“ for T: 'static. These all mean the same thing, and are different from “annotated with 'static“ as in &'static T.

Send bound

Tasks spawned by tokio::spawn must implement Send. This allows the Tokio runtime to move the tasks between threads while they are suspended at an .await.

Tasks are Send when all data that is held across .await calls is Send. This is a bit subtle. When .await is called, the task yields back to the scheduler. The next time the task is executed, it resumes from the point it last yielded. To make this work, all state that is used after .await must be saved by the task. If this state is Send, i.e. can be moved across threads, then the task itself can be moved across threads. Conversely, if the state is not Send, then neither is the task.

For example, this works:

  1. use tokio::task::yield_now;
  2. use std::rc::Rc;
  4. #[tokio::main]
  5. async fn main() {
  6.     tokio::spawn(async {
  7.         // The scope forces `rc` to drop before `.await`.
  8.         {
  9.             let rc = Rc::new("hello");
  10.             println!("{}", rc);
  11.         }
  13.         // `rc` is no longer used. It is **not** persisted when
  14.         // the task yields to the scheduler
  15.         yield_now().await;
  16.     });
  17. }

This does not:

  1. use tokio::task::yield_now;
  2. use std::rc::Rc;
  4. #[tokio::main]
  5. async fn main() {
  6.     tokio::spawn(async {
  7.         let rc = Rc::new("hello");
  9.         // `rc` is used after `.await`. It must be persisted to
  10.         // the task's state.
  11.         yield_now().await;
  13.         println!("{}", rc);
  14.     });
  15. }

Attempting to compile the snippet results in:

  1. error: future cannot be sent between threads safely
  2.    --> src/main.rs:6:5
  3.     |
  4. 6   |     tokio::spawn(async {
  5.     |     ^^^^^^^^^^^^ future created by async block is not `Send`
  6.     | 
  7.    ::: [..]spawn.rs:127:21
  8.     |
  9. 127 |         T: Future + Send + 'static,
  10.     |                     ---- required by this bound in
  11.     |                          `tokio::task::spawn::spawn`
  12.     |
  13.     = help: within `impl std::future::Future`, the trait
  14.     |       `std::marker::Send` is not  implemented for
  15.     |       `std::rc::Rc<&str>`
  16. note: future is not `Send` as this value is used across an await
  17.    --> src/main.rs:10:9
  18.     |
  19. 7   |         let rc = Rc::new("hello");
  20.     |             -- has type `std::rc::Rc<&str>` which is not `Send`
  21. ...
  22. 10  |         yield_now().await;
  23.     |         ^^^^^^^^^^^^^^^^^ await occurs here, with `rc` maybe
  24.     |                           used later
  25. 11  |         println!("{}", rc);
  26. 12  |     });
  27.     |     - `rc` is later dropped here

We will discuss a special case of this error in more depth in the next chapter.

Store values

We will now implement the process function to handle incoming commands. We will use a HashMap to store values. SET commands will insert into the HashMap and GET values will load them. Additionally, we will use a loop to accept more than one command per connection.

  1. use tokio::net::TcpStream;
  2. use mini_redis::{Connection, Frame};
  4. async fn process(socket: TcpStream) {
  5.     use mini_redis::Command::{self, Get, Set};
  6.     use std::collections::HashMap;
  8.     // A hashmap is used to store data
  9.     let mut db = HashMap::new();
  11.     // Connection, provided by `mini-redis`, handles parsing frames from
  12.     // the socket
  13.     let mut connection = Connection::new(socket);
  15.     // Use `read_frame` to receive a command from the connection.
  16.     while let Some(frame) = connection.read_frame().await.unwrap() {
  17.         let response = match Command::from_frame(frame).unwrap() {
  18.             Set(cmd) => {
  19.                 // The value is stored as `Vec<u8>`
  20.                 db.insert(cmd.key().to_string(), cmd.value().to_vec());
  21.                 Frame::Simple("OK".to_string())
  22.             }
  23.             Get(cmd) => {
  24.                 if let Some(value) = db.get(cmd.key()) {
  25.                     // `Frame::Bulk` expects data to be of type `Bytes`. This
  26.                     // type will be covered later in the tutorial. For now,
  27.                     // `&Vec<u8>` is converted to `Bytes` using `into()`.
  28.                     Frame::Bulk(value.clone().into())
  29.                 } else {
  30.                     Frame::Null
  31.                 }
  32.             }
  33.             cmd => panic!("unimplemented {:?}", cmd),
  34.         };
  36.         // Write the response to the client
  37.         connection.write_frame(&response).await.unwrap();
  38.     }
  39. }

Now, start the server:

  1. $ cargo run

and in a separate terminal window, run the hello-redis example:

  1. $ cargo run --example hello-redis

Now, the output will be:

  1. got value from the server; result=Some(b"world")

We can now get and set values, but there is a problem: The values are not shared between connections. If another socket connects and tries to GET the hello key, it will not find anything.

You can find the full code here.

In the next section, we will implement persisting data for all sockets.