Rust Concurrent Execution with Futures and Tokio

Question

I've got some Rust code that currently looks like this

fn read_stdin(mut tx: mpsc::Sender<String>) {
    loop {
        // read from stdin and send value over tx.
    }
}

fn sleep_for(n: u64) -> impl Future<Item = (), Error = ()> {
    thread::sleep(time::Duration::from_millis(n));
    println!("[{}] slept for {} ms", Local::now().format("%T%.3f"), n);
    future::ok(())
}

fn main() {
    let (stdin_tx, stdin_rx) = mpsc::channel(0);
    thread::spawn(move || read_stdin(stdin_tx));

    let server = stdin_rx
        .map(|data| data.trim().parse::<u64>().unwrap_or(0))
        .for_each(|n| tokio::spawn(sleep_for(n * 100)));
    tokio::run(server);
}

It uses tokio and futures, with the aim of running some "cpu heavy" work (emulated by the sleep_for function) and then outputting some stuff to stdout.

When I run it, things seems to work fine and I get this output

2
[00:00:00.800] slept for 200 ms
10
1
[00:00:01.800] slept for 1000 ms
[00:00:01.900] slept for 100 ms

The first output with the value 2 is exactly as expected, and I see the timestamp printed after 200ms. But for the next inputs, it becomes clear that the sleep_for function is being executed sequentially, and not concurrently.

The output that I want to see is

2
[00:00:00.800] slept for 200 ms
10
1
[00:00:00.900] slept for 100 ms
[00:00:01.900] slept for 1000 ms

It seems that to get the output I'm looking for I want to execute sleep_for(10) and sleep_for(1) concurrently. How would I go about doing this in Rust with futures and tokio?

(Note: the actual values of the timestamps aren't important I'm using them more to show the ordering of execution within the program)

Josh Leeb-du Toit Josh Leeb-du Toit · Accepted Answer · 2018-11-18T09:48:16

Found a solution with the use of the futures-timer crate.

use chrono::Local;
use futures::{future, sync::mpsc, Future, Sink, Stream};
use futures_timer::Delay;
use std::{io::stdin, thread, time::Duration};

fn read_stdin(mut tx: mpsc::Sender<String>) {
    let stdin = stdin();
    loop {
        let mut buf = String::new();
        stdin.read_line(&mut buf).unwrap();
        tx = tx.send(buf).wait().unwrap()
    }
}

fn main() {
    let (stdin_tx, stdin_rx) = mpsc::channel(0);
    thread::spawn(move || read_stdin(stdin_tx));

    let server = stdin_rx
        .map(|data| data.trim().parse::<u64>().unwrap_or(0) * 100)
        .for_each(|delay| {
            println!("[{}] {} ms -> start", Local::now().format("%T%.3f"), delay);
            tokio::spawn({
                Delay::new(Duration::from_millis(delay))
                    .and_then(move |_| {
                        println!("[{}] {} ms -> done", Local::now().format("%T%.3f"), delay);
                        future::ok(())
                    })
                    .map_err(|e| panic!(e))
            })
        });

    tokio::run(server);
}

The issue is that the rather letting the future to become parked and then notifying the current task, the code presented in the question was just sleeping the thread and so no progress could be made.

Update: Now I've just come across tokio-timer which seems like the standard way of doing this.

Rust Concurrent Execution with Futures and Tokio

1 Answers