Spawning

Tokio based applications are organized in terms of Tasks. A task is a small unitof logic that executes independently from other tasks. It is similar to Go’sgoroutine and Erlang’s process, but asynchronous. In other words, tasks areasynchronous green threads. Tasks are spawned for similar reasons that threadsare spawned in synchronous code, but spawning a task with Tokio is extremelylightweight.

Previous examples defined a future and passed that future to tokio::run. Thisresulted in a task being spawned onto Tokio’s runtime to execute the providedfuture. Additional tasks may be spawned by calling tokio::spawn, but only fromcode that is already running on a Tokio task. One way to think about it is thefuture passed to tokio::run is the “main function”.

In the following example, four tasks are spawned.

  1. extern crate tokio;
  2. extern crate futures;
  4. use futures::future::lazy;
  6. tokio::run(lazy(|| {
  7.     for i in 0..4 {
  8.         tokio::spawn(lazy(move || {
  9.             println!("Hello from task {}", i);
  10.             Ok(())
  11.         }));
  12.     }
  14.     Ok(())
  15. }));

The tokio::run function will block until the the future passed to runteriminates as well as all other spawned tasks. In this case, tokio::runblocks until all four tasks output to STDOUT and terminate.

The lazy function runs the closure the first time the future is polled. Itis used here to ensure that tokio::spawn is called from a task. Withoutlazy, tokio::spawn would be called from outside the context of a task,which results in an error.

Communicating with tasks

Just as with Go and Erlang, tasks can communicate using message passing. Infact, it will be very common to use message passing to coordinate multipletasks. This allows independent tasks to still interact.

The futures crate provides a sync module which contains some channeltypes that are ideal for message passing across tasks.

  • oneshot is a channel for sending exactly one value.
  • mpsc is a channel for sending many (zero or more) values.
    A oneshot is ideal for getting the result from a spawned task:
  1. extern crate tokio;
  2. extern crate futures;
  4. use futures::Future;
  5. use futures::future::lazy;
  6. use futures::sync::oneshot;
  8. tokio::run(lazy(|| {
  9.     let (tx, rx) = oneshot::channel();
  11.     tokio::spawn(lazy(|| {
  12.         tx.send("hello from spawned task");
  13.         Ok(())
  14.     }));
  16.     rx.and_then(|msg| {
  17.         println!("Got `{}`", msg);
  18.         Ok(())
  19.     })
  20.     .map_err(|e| println!("error = {:?}", e))
  21. }));

And mpsc is good for sending a stream of values to another task:

  1. extern crate tokio;
  2. extern crate futures;
  4. use futures::{stream, Future, Stream, Sink};
  5. use futures::future::lazy;
  6. use futures::sync::mpsc;
  8. tokio::run(lazy(|| {
  9.     let (tx, rx) = mpsc::channel(1_024);
  11.     tokio::spawn({
  12.         stream::iter_ok(0..10).fold(tx, |tx, i| {
  13.             tx.send(format!("Message {} from spawned task", i))
  14.                 .map_err(|e| println!("error = {:?}", e))
  15.         })
  16.         .map(|_| ()) // Drop tx handle
  17.     });
  19.     rx.for_each(|msg| {
  20.         println!("Got `{}`", msg);
  21.         Ok(())
  22.     })
  23. }));

These two message passing primitives will also be used in the examples below tocoordinate and communicate between tasks.

Multi threaded

While it is possible to introduce concurrency with futures without spawningtasks, this concurrency will be limited to running on a single thread. Spawningtasks allows the Tokio runtime to schedule these tasks on multiple threads.

The multi-threaded Tokio runtime manages multiple OS threads internally.It multiplexes many tasks across a few physical threads. When a Tokioapplication spawns its tasks, these tasks are submitted to the runtime and theruntime handles scheduling.

When to spawn tasks

As all things software related, the answer is that it depends. Generally, theanswer is spawn a new task whenever you can. The more available tasks, thegreater the ability to run the tasks in parallel. However, keep in mind that ifmultiple tasks do require communication, this will involve channel overhead.

The following examples will help illustrate cases for spawning new tasks.

Processing inbound sockets

The most straightforward example for spawning tasks is a network server.The primary task listens for inbound sockets on a TCP listener. When anew connection arrives, the listener task spawns a new task forprocessing the socket.

  1. extern crate tokio;
  2. extern crate futures;
  4. use tokio::io;
  5. use tokio::net::TcpListener;
  6. use futures::{Future, Stream};
  8. let addr = "127.0.0.1:0".parse().unwrap();
  9. let listener = TcpListener::bind(&addr).unwrap();
  11. # if false {
  12. tokio::run({
  13.     listener.incoming().for_each(|socket| {
  14.         // An inbound socket has been received.
  15.         //
  16.         // Spawn a new task to process the socket
  17.         tokio::spawn({
  18.             // In this example, "hello world" will be written to the
  19.             // socket followed by the socket being closed.
  20.             io::write_all(socket, "hello world")
  21.                 // Drop the socket
  22.                 .map(|_| ())
  23.                 // Write any error to STDOUT
  24.                 .map_err(|e| println!("socket error = {:?}", e))
  25.         });
  27.         // Receive the next inbound socket
  28.         Ok(())
  29.     })
  30.     .map_err(|e| println!("listener error = {:?}", e))
  31. });
  32. # }

The listener task and the tasks that process each socket are completelyunrelated. They do not communicate and either can terminate withoutimpacting the others. This is a perfect use case for spawning tasks.

Background processing

Another case is to spawn a task that runs background computations inservice of other tasks. The primary tasks send data to the backgroundtask for processing but do not care about if and when the data getsprocessed. This also allows a single background task to coalesce datafrom multiple primary tasks.

This requires communication between the primary tasks and the backgroundtask. This is usually handled with an mpsc channel.

The following example is a TCP server that reads data from the remotepeer and tracks the number of received bytes. It then sends the numberof received bytes to a background task. This background task writes thetotal number of bytes read from all socket tasks every 30 seconds.

  1. extern crate tokio;
  2. extern crate futures;
  4. use tokio::io;
  5. use tokio::net::TcpListener;
  6. use tokio::timer::Interval;
  7. use futures::{future, stream, Future, Stream, Sink};
  8. use futures::future::lazy;
  9. use futures::sync::mpsc;
  10. use std::time::Duration;
  12. // Defines the background task. The `rx` argument is the channel receive
  13. // handle. The task will pull `usize` values (which represent number of
  14. // bytes read by a socket) off the channel and sum it internally. Every
  15. // 30 seconds, the current sum is written to STDOUT and the sum is reset
  16. // to zero.
  17. fn bg_task(rx: mpsc::Receiver<usize>)
  18. -> impl Future<Item = (), Error = ()>
  19. {
  20.     // The stream of received `usize` values will be merged with a 30
  21.     // second interval stream. The value types of each stream must
  22.     // match. This enum is used to track the various values.
  23.     #[derive(Eq, PartialEq)]
  24.     enum Item {
  25.         Value(usize),
  26.         Tick,
  27.         Done,
  28.     }
  30.     // Interval at which the current sum is written to STDOUT.
  31.     let tick_dur = Duration::from_secs(30);
  33.     let interval = Interval::new_interval(tick_dur)
  34.         .map(|_| Item::Tick)
  35.         .map_err(|_| ());
  37.     // Turn the stream into a sequence of:
  38.     // Item(num), Item(num), ... Done
  39.     //
  40.     let items = rx.map(Item::Value)
  41.       .chain(stream::once(Ok(Item::Done)))
  42.       // Merge in the stream of intervals
  43.       .select(interval)
  44.       // Terminate the stream once `Done` is received. This is necessary
  45.       // because `Interval` is an infinite stream and `select` will keep
  46.       // selecting on it.
  47.       .take_while(|item| future::ok(*item != Item::Done));
  49.     // With the stream of `Item` values, start our logic.
  50.     //
  51.     // Using `fold` allows the state to be maintained across iterations.
  52.     // In this case, the state is the number of read bytes between tick.
  53.     items.fold(0, |num, item| {
  54.         match item {
  55.             // Sum the number of bytes with the state.
  56.             Item::Value(v) => future::ok(num + v),
  57.             Item::Tick => {
  58.                 println!("bytes read = {}", num);
  60.                 // Reset the byte counter
  61.                 future::ok(0)
  62.             }
  63.             _ => unreachable!(),
  64.         }
  65.     })
  66.     .map(|_| ())
  67. }
  69. # if false {
  70. // Start the application
  71. tokio::run(lazy(|| {
  72.     let addr = "127.0.0.1:0".parse().unwrap();
  73.     let listener = TcpListener::bind(&addr).unwrap();
  75.     // Create the channel that is used to communicate with the
  76.     // background task.
  77.     let (tx, rx) = mpsc::channel(1_024);
  79.     // Spawn the background task:
  80.     tokio::spawn(bg_task(rx));
  82.     listener.incoming().for_each(move |socket| {
  83.         // An inbound socket has been received.
  84.         //
  85.         // Spawn a new task to process the socket
  86.         tokio::spawn({
  87.             // Each spawned task will have a clone of the sender handle.
  88.             let tx = tx.clone();
  90.             // In this example, "hello world" will be written to the
  91.             // socket followed by the socket being closed.
  92.             io::read_to_end(socket, vec![])
  93.                 // Drop the socket
  94.                 .and_then(move |(_, buf)| {
  95.                     tx.send(buf.len())
  96.                         .map_err(|_| io::ErrorKind::Other.into())
  97.                 })
  98.                 .map(|_| ())
  99.                 // Write any error to STDOUT
  100.                 .map_err(|e| println!("socket error = {:?}", e))
  101.         });
  103.         // Receive the next inbound socket
  104.         Ok(())
  105.     })
  106.     .map_err(|e| println!("listener error = {:?}", e))
  107. }));
  108. # }

Coordinating access to a resource

When working with futures, the preferred strategy for coordinatingaccess to a shared resource (socket, data, etc…) is by using messagepassing. To do this, a dedicated task is spawned to manage the resourceand other tasks interact with the resource by sending messages.

This pattern is very similar to the previous example, but this time thetasks want to receive a message back once the operation is complete. Toimplement this, both mpsc and oneshot channels are used.

The example coordinates access to a transport over a ping / pongprotocol. Pings are sent into the transport and pongs are received.Primary tasks send a message to the coordinator task to initiate a ping,the coordinator task will respond to the ping request with the roundtrip time. The message sent to the coordinator task over thempsc contains a oneshot::Sender allowing the coordinator task torespond.

  1. extern crate tokio;
  2. extern crate futures;
  4. use tokio::io;
  5. use futures::{future, Future, Stream, Sink};
  6. use futures::future::lazy;
  7. use futures::sync::{mpsc, oneshot};
  8. use std::time::{Duration, Instant};
  10. type Message = oneshot::Sender<Duration>;
  12. struct Transport;
  14. impl Transport {
  15.     fn send_ping(&self) {
  16.         // ...
  17.     }
  19.     fn recv_pong(&self) -> impl Future<Item = (), Error = io::Error> {
  20. #         future::ok(())
  21.         // ...
  22.     }
  23. }
  25. fn coordinator_task(rx: mpsc::Receiver<Message>)
  26. -> impl Future<Item = (), Error = ()>
  27. {
  28.     let transport = Transport;
  30.     rx.for_each(move |pong_tx| {
  31.         let start = Instant::now();
  33.         transport.send_ping();
  35.         transport.recv_pong()
  36.             .map_err(|_| ())
  37.             .and_then(move |_| {
  38.                 let rtt = start.elapsed();
  39.                 pong_tx.send(rtt).unwrap();
  40.                 Ok(())
  41.             })
  42.     })
  43. }
  45. /// Request an rtt.
  46. fn rtt(tx: mpsc::Sender<Message>)
  47. -> impl Future<Item = (Duration, mpsc::Sender<Message>), Error = ()>
  48. {
  49.     let (resp_tx, resp_rx) = oneshot::channel();
  51.     tx.send(resp_tx)
  52.         .map_err(|_| ())
  53.         .and_then(|tx| {
  54.             resp_rx.map(|dur| (dur, tx))
  55.                 .map_err(|_| ())
  56.         })
  57. }
  59. # if false {
  60. // Start the application
  61. tokio::run(lazy(|| {
  62.     // Create the channel that is used to communicate with the
  63.     // background task.
  64.     let (tx, rx) = mpsc::channel(1_024);
  66.     // Spawn the background task:
  67.     tokio::spawn(coordinator_task(rx));
  69.     // Spawn a few tasks that use the coordinator to requst RTTs.
  70.     for _ in 0..4 {
  71.         let tx = tx.clone();
  73.         tokio::spawn(lazy(|| {
  74.             rtt(tx).and_then(|(dur, _)| {
  75.                 println!("duration = {:?}", dur);
  76.                 Ok(())
  77.             })
  78.         }));
  79.     }
  81.     Ok(())
  82. }));
  83. # }

When not to spawn tasks

If the amount of coordination via message passing and synchronization primitivesoutweighs the parallism benefits from spawning tasks, then maintaining a singletask is preferred.

For example, it is generally better to maintain reading from and writing to asingle TCP socket on a single task instead of splitting up reading and writingbetween two tasks.

Next up: Leaf futures