Início / coding / Perguntas / 79150081
Accepted
yushizhao
Asked: 2024-11-02 14:23:56 +0800 CST

Por que o trabalhador bloqueado do Tokio ainda pode trabalhar simultaneamente?

  • 772

Quando eu corro

use tokio::sync::mpsc;

#[tokio::main(flavor = "multi_thread", worker_threads = 1)]
async fn main() {
    let (tx, mut rx) = mpsc::channel(1);

    tokio::spawn(async move {
        while let Some(i) = rx.recv().await {
            println!("got = {}", i);
        } 
    });

    for i in 0..5 {

        // busy calculation
        std::thread::sleep(std::time::Duration::from_millis(10));

        match tx.try_send(i) {
            Ok(_) => {
                println!("sent = {}", i);
            },
            Err(err) => {
                println!("{}", err);
            }
        };

    };
 
}

Eu obtive

sent = 0
got = 0
sent = 1
got = 1
sent = 2
got = 2
sent = 3
got = 3
sent = 4

Pelo que entendi, o único trabalhador está trabalhando no loop for porque ele nunca rende. O único trabalhador não deve ter chance de trabalhar no recebimento. Portanto, o canal deve estar cheio após o primeiro envio. Acontece que estou errado. O que estou perdendo?

asynchronous

1 respostas

  1. Best Answer

    O código em uma #[tokio::main]função não é executado de fato em um thread worker. Portanto, a tarefa gerada é enviada para o único thread worker, enquanto o forloop é executado no thread principal do programa.

    Sob o capô, tokio::maineleva o corpo da função para um asyncbloco, constrói um runtime e então passa o future para o bloco async gerado para Runtime::block_on. De acordo com a documentação deste método:

    Note que o future requerido por esta função não é executado como um worker. A expectativa é que outras tarefas sejam geradas pelo future aqui. Aguardar outros futures do future fornecido aqui não executará tão rápido quanto aqueles gerados como workers.

    Você pode concretizar o que está esperando de duas maneiras.

    A primeira maneira é elevar o corpo da sua mainfunção para uma nova tarefa:

    use tokio::sync::mpsc;

#[tokio::main(flavor = "multi_thread", worker_threads = 1)]
async fn main() {
    tokio::spawn(async move {
        let (tx, mut rx) = mpsc::channel(1);

        tokio::spawn(async move {
            while let Some(i) = rx.recv().await {
                println!("got = {}", i);
            }
        });

        for i in 0..5 {
            // busy calculation
            std::thread::sleep(std::time::Duration::from_millis(10));

            match tx.try_send(i) {
                Ok(_) => {
                    println!("sent = {}", i);
                }
                Err(err) => {
                    println!("{}", err);
                }
            };
        }
    })
    .await
    .unwrap();
}

    A segunda é usar o runtime "current thread" em vez do runtime multithreaded. Isso executa todas as tarefas no thread principal.

    use tokio::sync::mpsc;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, mut rx) = mpsc::channel(1);

    tokio::spawn(async move {
        while let Some(i) = rx.recv().await {
            println!("got = {}", i);
        }
    });

    for i in 0..5 {
        // busy calculation
        std::thread::sleep(std::time::Duration::from_millis(10));

        match tx.try_send(i) {
            Ok(_) => {
                println!("sent = {}", i);
            }
            Err(err) => {
                println!("{}", err);
            }
        };
    }
}

    Ambos mostrarão "enviado = 0" e try_sendfalharão com "nenhuma capacidade disponível".

    • 2

relate perguntas