Can't get spawn to work in parallel?

[moh-eulith]

Here is a simplified test case:

use std::time::Duration;
use async_trait::async_trait;
use anyhow::Error;

pub struct Race {}

#[derive(Debug)]
pub enum Work {
    DoIt,
    Quit
}

#[async_trait]
pub trait Worker {
    async fn work(&mut self);
}

impl Race {

    pub async fn race<T>(racer1: T, iterations: u32, interval_millis: u64)
        where T: Worker + Send + Sync + 'static {
        let (s1, r1) = tokio::sync::mpsc::unbounded_channel();
        let handle1 = tokio::task::spawn(async move { Race::coordinate(racer1, r1) } );
        for _i in 0..iterations {
            tokio::time::sleep(Duration::from_millis(interval_millis)).await;
            s1.send(Work::DoIt).unwrap();
        }
        s1.send(Work::Quit).unwrap();
        handle1.await.unwrap().await.unwrap();
    }

    async fn coordinate<T>(mut racer: T, mut receiver: tokio::sync::mpsc::UnboundedReceiver<Work>)  -> Result<(), Error>
        where T: Worker + Send + Sync + 'static {
        loop {
            let work = receiver.recv().await.unwrap();
            match work {
                Work::Quit => break,
                Work::DoIt => racer.work().await
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use async_trait::async_trait;
    use std::time::{Instant};
    use crate::{Race, Worker};

    struct Printer {
        name: String,
        count: u64,
        last: Instant,
    }

    impl Printer {
        pub fn new(name: String) -> Self {
            Printer { name, count: 0 , last: Instant::now() }
        }
    }

    #[async_trait]
    impl Worker for Printer {
        async fn work(&mut self) {
            self.count += 1;
            if self.last.elapsed().as_millis() < 1000 {
                println!("ERROR: Printer {} count {} didn't sleep long enough: {:?}", self.name, self.count, self.last.elapsed());
            } else if self.last.elapsed().as_millis() > 2000 {
                println!("ERROR: Printer {} count {} very late: {:?}", self.name, self.count, self.last.elapsed());
            } else {
                println!("Printer {} count {} elapsed {:?}", self.name, self.count, self.last.elapsed());
            }
            self.last = Instant::now();
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    pub async fn test_race() {
        let p1 = Printer::new("First".to_string());
        Race::race(p1, 10, 1500).await;
    }
}

Running this prints the ERROR lines, as if the main thread runs and the spawned task is not running at all until the joinHandle is awaited. Given the generous timings, I was expecting it to work with 1 worker_thread, but it doesn't work even with 4. What am I doing wrong?

Edit: removed use of async_channel in favor of tokio's mpsc, but it didn't make a difference.

[Noah-Kennedy]

Ah, you seem to have forgotten to await coordinate inside the task that you spawned!

The task you spawned does nothing but return the future, as you never await it within the block. In other words, the future that you spawned just returns a future. This is why you await twice when you await the handle. You are first awaiting the spawned task, and then awaiting the future that you had meant to run within it, but didn't.

If you await coordinate method within the async block, that should fix the issue. You could also simply get rid of the async block entirely, and simply tokio::task::spawn(Race::coordinate(racer1, r1)), which would be a bit more succinct.