How to run async tasks in several threads in Rust-CodePudding

I want to run some async tasks in several threads. I tried to use tokio's Runtime and new_multi_thread, but I've got a panic

thread 'main' panicked at 'Cannot drop a runtime in a context where blocking is not allowed. This happens when a runtime is dropped from within an asynchronous context.'

Here's the code.

async fn routine(millis: u64) {
    tokio::time::sleep(std::time::Duration::from_millis(millis)).await;
}

#[tokio::main]
async fn main() {
    let rt = tokio::runtime::Builder::new_multi_thread().worker_threads(2).enable_all().build().unwrap();

    let handles: Vec<tokio::task::JoinHandle<_>> = (1..10_u64).map(|i| {
        rt.spawn(routine(i))
    }).collect();

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

Could you please tell me what is wrong with the code?

CodePudding user response：

You don't need to initialize a runtime as that's what the annotation #[tokio::main] does (it's a macro).

This...

#[tokio::main]
async fn main() {
    println!("hello");
}

...de-sugars to:

fn main() {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            println!("hello");
        })
}

Note that #[tokio::main] is equivalent to #[tokio::main(flavor = "multi_thread")] (defaults to multi-threaded runtime), as opposed to the single-threaded runtime #[tokio::main(flavor = "current_thread")]. In your case you can use this form #[tokio::main(worker_threads = 2)].

CodePudding user response：

#[tokio::main] creates a Runtime and runs async fn main() on it. Your main then creates another Runtime. Normally, when we create our own runtime, we wouldn't use the #[tokio::main] macro, we'd just write a normal main.

We can use block_on to wait for the tasks to complete, then we can drop the Runtime. With a normal main, we're not dropping the Runtime in an asynchronous context, so the program no longer panics.

async fn routine(millis: u64) {
    println!("in routine({millis})");
    tokio::time::sleep(std::time::Duration::from_millis(millis)).await;
}

fn main() {
    let rt = tokio::runtime::Builder::new_multi_thread().worker_threads(2).enable_all().build().unwrap();

    let handles: Vec<tokio::task::JoinHandle<_>> = (1..10_u64).map(|i| {
        rt.spawn(routine(i))
    }).collect();

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