Why all the Tokio spawn, Tokio spawn + spawn_blocking, and Tokio spawn + Rayon spawn finish at the same time?

use std::{
  time::Instant,
  io::{self, Write},
  sync::Arc
};
use tokio::{fs, sync::oneshot};

fn fib(n: u64) -> u64 {
    match n {
        0 => 0,
        1 => 1,
        _ => fib(n - 1) + fib(n - 2),
    }
}

async fn fib_rayon(i: u64, val: String, data_ref: Arc<whirlwind::ShardMap<String, String>>) {
  let (s, r) = oneshot::channel();
  rayon::spawn(move || {
    let result = fib(i);
    let entry = format!("val: {}, fib: {}", val.trim(), result);
    
    let _ = s.send(entry);
  });
  match r.await {
    Ok(val) => {
    data_ref.insert(i.to_string(), val).await;
    },
    Err(_) => {}
  }
  
}

async fn spawn(n: u64){
  let data = Arc::new(whirlwind::ShardMap::new());

    let mut handles = Vec::new();
    
    let time = Instant::now();

    for i in 0..n {
        let data_ref = data.clone();
        let handle = tokio::spawn(async move {
        
            let mut val = fs::read_to_string("data.txt").await.unwrap();
            let result = fib(i);
            
            val = format!("val: {}, fib: {}", val, result.to_string());
            data_ref.insert(i.to_string(), val).await;
            
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.await.unwrap();
    }
    
    let time2 = time.elapsed();
    
    println!("\ntotal items: {}", data.len().await);
    println!("\n\nspawn: {:?}\n\n", time2);
}

async fn spawn_blocking(n: u64){
  let data = Arc::new(whirlwind::ShardMap::new());

    let mut handles = Vec::new();
    
    let time = Instant::now();

    for i in 0..n {
        let data_ref = data.clone();
        let handle = tokio::spawn(async move {
        
            let mut val = fs::read_to_string("data.txt").await.unwrap();
            
          let (key, val) = tokio::task::spawn_blocking(move || { 
            let result = fib(i);
            val = format!("val: {}, fib: {}", val, result);
            (i.to_string(), val)
          })
          .await
          .unwrap();

        data_ref.insert(key, val).await;
            
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.await.unwrap();
    }
    
    let time2 = time.elapsed();
    
    println!("\ntotal items: {}", data.len().await);
    println!("\n\nspawn blocking: {:?}\n\n", time2);
}

async fn rayon(n: u64) {
    let data = Arc::new(whirlwind::ShardMap::new());
    let mut handles = Vec::new();
    let time = Instant::now();

    for i in 0..n {
        let data_ref = data.clone();

        let handle = tokio::spawn(async move {

            let val = fs::read_to_string("data.txt").await.unwrap();

            fib_rayon(i, val, data_ref).await;

        });

        handles.push(handle);
    }

    for handle in handles {
        handle.await.unwrap();
    }

    let duration = time.elapsed();
    println!("\ntotal items: {}", data.len().await);
    println!("\n\nrayon: {:?}\n\n", duration);
}

#[tokio::main]
async fn main() {
    if !std::path::Path::new("data.txt").exists() {
        fs::write("data.txt", "hello").await.unwrap();
    }
    
    loop {
        print!("\n1 - spawn\n2 - spawn_blocking\n3 - rayon\n\ninput: ");
        io::stdout().flush().unwrap();

        let mut input = String::new();
        io::stdin().read_line(&mut input).unwrap();
        let input = input.trim(); 

        match input {
            "1" => spawn(45).await,
            "2" => spawn_blocking(45).await,
            "3" => rayon(45).await,
            "exit" => {
                println!("Exiting program...");
                break;
            }
            _ => println!("Input is not valid"),
        }
    }
}

5 posts - 3 participants

Read full topic