Async a fundo em Rust: select!, timeouts, streams e Pin

Na página de Async/Await você aprendeu o básico: uma async fn devolve uma future preguiçosa, o .await executa essa future, o #[tokio::main] liga o runtime, e tokio::join! e tokio::spawn rodam coisas ao mesmo tempo. Se algo disso ainda parece novo, leia aquela página primeiro. Esta aqui parte dela.

Programas async de verdade precisam de mais algumas ferramentas. Pense de novo numa cozinha movimentada. Às vezes você espera todos os pratos ficarem prontos (join!). Às vezes só importa a primeira coisa que acontecer: a comida fica pronta ou o cliente vai embora (select!). Às vezes você desiste depois de cinco minutos (timeout). E às vezes os pedidos não param de chegar, um por um, a noite inteira (streams). Esta página cobre tudo isso e ainda o Pin, aquela palavra estranha que mais cedo ou mais tarde aparece nas mensagens de erro, explicado em linguagem simples.

Todos os exemplos usam o tokio com features = ["full"], igual à página do básico. Os botões “Experimente você mesmo” funcionam, porque o Rust Playground já vem com o tokio.

Rodando muitas coisas ao mesmo tempo

Você já conhece o tokio::join!: ele espera todas as suas futures e devolve os resultados numa tupla. O irmão dele, tokio::try_join!, é para futures que devolvem um Result. Ele te dá um Ok com todos os valores ou o primeiro Err, e para de esperar pelas outras assim que uma falha:

use std::time::Duration;
use tokio::time::sleep;

async fn load_profile(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(20)).await;
    Ok(format!("perfil #{id}"))
}

async fn load_settings(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(10)).await;
    if id == 2 {
        return Err(format!("sem configurações para o usuário {id}"));
    }
    Ok(String::from("modo escuro"))
}

#[tokio::main]
async fn main() {
    for id in [1, 2] {
        match tokio::try_join!(load_profile(id), load_settings(id)) {
            Ok((profile, settings)) => println!("usuário {id}: {profile}, {settings}"),
            Err(e) => println!("usuário {id} falhou: {e}"),
        }
    }
}

Saída:

usuário 1: perfil #1, modo escuro
usuário 2 falhou: sem configurações para o usuário 2

▶ Experimente você mesmo

O join! e o try_join! rodam as futures dentro da tarefa atual, revezando entre elas. Isso é perfeito para um punhado de futures que você conhece de antemão. Quando você tem uma lista de trabalhos cujo tamanho só se sabe em tempo de execução, crie tarefas separadas para eles. Daria para guardar cada JoinHandle num Vec, como na página do básico. Mas o tokio tem uma ferramenta melhor para isso: o JoinSet, um saco de tarefas que te entrega cada resultado assim que ele termina:

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::sleep;

async fn check_site(site: &'static str, ms: u64) -> (&'static str, u64) {
    sleep(Duration::from_millis(ms)).await; // finja que isto é uma chamada de rede
    (site, ms)
}

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();
    for (site, ms) in [("rust-lang.org", 30), ("docs.rs", 10), ("crates.io", 20)] {
        set.spawn(check_site(site, ms));
    }

    let mut results = Vec::new();
    while let Some(res) = set.join_next().await {
        results.push(res.unwrap()); // só dá Err se a tarefa entrou em panic
    }

    results.sort(); // a ordem de término pode mudar de uma execução para outra
    for (site, ms) in results {
        println!("{site} respondeu em {ms}ms");
    }
}

Saída:

crates.io respondeu em 20ms
docs.rs respondeu em 10ms
rust-lang.org respondeu em 30ms

▶ Experimente você mesmo

O join_next() devolve None quando o set fica vazio, então o laço while let termina sozinho. Os resultados chegam na ordem em que terminam, não na ordem em que foram criados. Quando as tarefas levam um tempo parecido, essa ordem pode mudar entre execuções, por isso ordenamos antes de imprimir. Mais uma coisa boa: se você descarta (drop) um JoinSet, toda tarefa que ainda estiver dentro dele é cancelada. Nada fica rodando por acidente.

Disputando futures com select!

O join! espera tudo. O tokio::select! espera a primeira. Cada ramo tem um padrão, uma future e um bloco de código: padrão = future => código. A future que terminar primeiro tem o seu código executado, e os outros ramos são cancelados.

O que “cancelado” quer dizer em Rust? A future perdedora é simplesmente descartada (dropped), como qualquer outro valor que sai do escopo. Ela nunca mais roda. As variáveis locais dela também são descartadas, então o código de limpeza delas é executado. Este exemplo usa um pequeno tipo com Drop para deixar isso visível:

use std::time::Duration;
use tokio::time::sleep;

struct Download(&'static str);

impl Drop for Download {
    fn drop(&mut self) {
        println!("limpando o download da {}", self.0);
    }
}

async fn download(mirror: &'static str, ms: u64) -> &'static str {
    let _download = Download(mirror);
    sleep(Duration::from_millis(ms)).await;
    println!("download da {mirror} concluído");
    mirror
}

#[tokio::main]
async fn main() {
    tokio::select! {
        winner = download("Europa", 50) => println!("vencedor: {winner}"),
        winner = download("Ásia", 1000) => println!("vencedor: {winner}"),
    }
    println!("main continua");
}

Saída:

download da Europa concluído
limpando o download da Europa
limpando o download da Ásia
vencedor: Europa
main continua

▶ Experimente você mesmo

Repare no que falta: “download da Ásia concluído” nunca aparece. A future da Ásia foi parada no seu .await e descartada, e o valor Download dela foi limpo na hora. É assim que o cancelamento funciona em todo o Rust async: pare de fazer poll de uma future e descarte-a. Não precisa de nenhuma chamada especial de “cancelar”.

Um uso clássico do select! é o timeout: você coloca o seu trabalho para disputar uma corrida contra um timer.

tokio::select! {
    report = build_report() => println!("{report}"),
    _ = tokio::time::sleep(Duration::from_secs(5)) => println!("lento demais, desistindo"),
}

Funciona bem, mas existe um atalho para isso, o tokio::time::timeout, que aparece na próxima seção.

Trabalhando até um sinal de desligamento

O select! brilha de verdade dentro de um loop. Aqui um worker (trabalhador) pega tarefas de um canal, mas também fica de ouvido num segundo canal, esperando uma mensagem de “por favor, pare”. Um canal oneshot carrega exatamente uma mensagem, o que faz dele um sinal de desligamento perfeito:

use std::time::Duration;
use tokio::sync::{mpsc, oneshot};
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (job_tx, mut job_rx) = mpsc::channel::<String>(10);
    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();

    let worker = tokio::spawn(async move {
        let mut done = 0;
        loop {
            tokio::select! {
                Some(job) = job_rx.recv() => {
                    println!("trabalhando em: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("sinal de desligamento recebido");
                    break;
                }
            }
        }
        done
    });

    for job in ["redimensionar foto", "enviar e-mail", "atualizar índice de busca"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    sleep(Duration::from_millis(100)).await; // dá tempo para o worker terminar
    stop_tx.send(()).unwrap();

    let done = worker.await.unwrap();
    println!("o worker concluiu {done} tarefas");
}

Saída:

trabalhando em: redimensionar foto
trabalhando em: enviar e-mail
trabalhando em: atualizar índice de busca
sinal de desligamento recebido
o worker concluiu 3 tarefas

▶ Experimente você mesmo

Alguns detalhes que vale a pena saber:

• Escrevemos &mut stop_rx, e não stop_rx. A cada volta do laço, o select! precisa do mesmo receptor de novo, então emprestamos o receptor em vez de movê-lo para dentro da primeira rodada.
• A cada volta do laço, job_rx.recv() cria uma future nova, e a perdedora é descartada. Tudo bem, porque recv() é cancel-safe (seguro para cancelar): descartá-la nunca perde uma mensagem.
• Se o padrão não casar (aqui, quando recv() devolve None porque todos os remetentes sumiram), o select! desliga aquele ramo e continua esperando os outros.
• Se vários ramos ficarem prontos no mesmo instante, o select! escolhe um ao acaso, para que nenhum ramo monopolize o laço. Escreva biased; como primeira linha dentro do select! para que ele confira os ramos de cima para baixo.

Cancel safety, em um parágrafo. Como o ramo perdedor é descartado, qualquer trabalho que ele tenha feito pela metade vai para o lixo. Para recv() num canal ou para sleep() isso não faz mal: nada tinha sido retirado ainda. Mas imagine uma future que lê 100 bytes de um socket de rede e já leu 60 quando é descartada. Esses 60 bytes se perderam. Uma future assim não é cancel-safe. A documentação do tokio diz, para cada método, se ele é cancel-safe ou não. Quando você usar select! num laço, confira os métodos que chama, ou crie a future uma vez só fora do laço e continue fazendo poll dessa mesma future (você vai ver como na seção sobre Pin).

Timeouts e timers

tokio::time::timeout(duration, future) envolve qualquer future. Se a future terminar a tempo, você recebe Ok(valor). Se o tempo acabar antes, a future é descartada (cancelada, igualzinho a um ramo perdedor do select!) e você recebe Err(Elapsed):

use std::time::Duration;
use tokio::time::{sleep, timeout};

async fn fetch_price(server_delay_ms: u64) -> u32 {
    sleep(Duration::from_millis(server_delay_ms)).await;
    42
}

#[tokio::main]
async fn main() {
    let limit = Duration::from_millis(200);

    for (server, delay) in [("servidor rápido", 20), ("servidor lento", 2000)] {
        match timeout(limit, fetch_price(delay)).await {
            Ok(price) => println!("{server}: o preço é {price}"),
            Err(e) => println!("{server}: {e}"),
        }
    }
}

Saída:

servidor rápido: o preço é 42
servidor lento: deadline has elapsed

▶ Experimente você mesmo

A mensagem “deadline has elapsed” (“o prazo se esgotou”) vem do próprio tokio, por isso aparece em inglês. Coloque um timeout em tudo que conversa com o mundo lá fora: chamadas HTTP, consultas ao banco de dados, leituras de um socket. Um servidor que nunca responde não deve congelar o seu programa para sempre.

sleep ou interval? O sleep espera uma vez. O interval te dá um ritmo constante, como um metrônomo. Você chama .tick().await num laço, e cada tick chega no horário certo. O primeiro tick dispara na hora:

use std::time::Duration;
use tokio::time::interval;

#[tokio::main]
async fn main() {
    let mut heartbeat = interval(Duration::from_millis(50));

    for beat in 1..=3 {
        heartbeat.tick().await; // o primeiro tick termina imediatamente
        println!("batida {beat}");
    }
}

Saída:

batida 1
batida 2
batida 3

▶ Experimente você mesmo

Por que não usar simplesmente um sleep de 50ms num laço? Porque aí cada rodada leva “o seu trabalho + 50ms”, e o horário vai escorregando aos poucos. Um interval mede a partir do início planejado de cada tick: se o seu trabalho leva 10ms, ele só espera os 40ms que faltam. Use interval para heartbeats, polling e tarefas do tipo “salvar a cada minuto”. Use sleep para uma pausa única.

Streams: iteradores async

Um iterador te entrega valores um por um com next(). Uma stream é a versão async: next().await te dá o próximo valor quando ele estiver pronto, e None quando não houver mais nenhum. Pense num painel de cotações da bolsa, em linhas chegando de um banco de dados ou em mensagens de um servidor de chat.

Na verdade você já usou algo parecido com uma stream: o receptor de um canal do tokio. rx.recv().await espera a próxima mensagem e devolve None quando todos os remetentes sumiram:

use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

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

    tokio::spawn(async move {
        for temp in [21, 22, 24, 23] {
            tx.send(temp).await.unwrap();
            sleep(Duration::from_millis(10)).await; // uma leitura a cada 10ms
        }
        // tx é descartado aqui, e isso fecha o canal
    });

    while let Some(temp) = rx.recv().await {
        println!("temperatura: {temp}°C");
    }
    println!("o sensor parou");
}

Saída:

temperatura: 21°C
temperatura: 22°C
temperatura: 24°C
temperatura: 23°C
o sensor parou

▶ Experimente você mesmo

A biblioteca padrão ainda não tem uma trait Stream (ela continua sendo desenhada). O ecossistema usa a do crate futures, e o crate auxiliar do tokio, tokio-stream, adiciona uma trait StreamExt com os métodos que você já conhece dos iteradores: next(), map(), filter(), take() e outros. Adicione-o ao seu projeto:

[dependencies]
tokio = { version = "1", features = ["full"] }
tokio-stream = "0.1"

use tokio_stream::{self as stream, StreamExt};

#[tokio::main]
async fn main() {
    let prices = stream::iter(vec![100, 102, 98, 105, 110]);

    let mut doubled = prices.map(|p| p * 2).take(3);

    while let Some(p) = doubled.next().await {
        println!("{p}");
    }
}

Isso imprime 200, 204 e 196. Fica igualzinho a código de iterador, a não ser pelo .await no next(). Para transformar o receptor de um canal do tokio numa stream de verdade, envolva-o assim: tokio_stream::wrappers::ReceiverStream::new(rx). (O crate futures tem o seu próprio StreamExt, com a mesma ideia. Escolha um por arquivo, porque as duas traits têm um método next() e elas entram em conflito.)

Pin, em palavras simples

Mais cedo ou mais tarde você vai ver Pin num tipo ou numa mensagem de erro. Aqui vai a ideia, sem a teoria pesada.

Quando o Rust compila uma async fn, ele a transforma numa struct escondida, uma máquina de estados. Cada variável local que ainda é necessária depois de um .await vira um campo dessa struct. Agora imagine que uma variável local pega emprestada outra, como let name = String::from("Ana"); let r = &name; com um .await no meio. Aí a struct guarda um ponteiro para um campo dela mesma. Se você movesse a struct para outro lugar da memória, esse ponteiro ia apontar para o lugar antigo, agora vazio.

O Pin é a solução. Um valor “pinado” (fixado) é uma promessa: “este valor não vai mais se mover na memória.” Depois que uma future está pinada, é seguro fazer poll dela. Na maior parte do tempo você nem percebe, porque o .await pina a future para você.

Você vai encontrar o Pin em três situações:

1. Fazer poll da mesma future várias vezes, por exemplo um prazo final num laço com select!.
2. Guardar futures num Vec ou num campo de struct, normalmente como Pin<Box<dyn Future<Output = T>>>.
3. Recursão numa async fn.

Para a primeira, veja o que acontece se você tenta reaproveitar um sleep num laço sem pinar:

let deadline = sleep(Duration::from_millis(300));

loop {
    tokio::select! {
        _ = &mut deadline => break,
        Some(n) = rx.recv() => println!("{n}"),
    }
}

error[E0277]: `PhantomPinned` cannot be unpinned
   |
   = note: consider using the `pin!` macro
           consider using `Box::pin` if you need to access the pinned value outside of the current scope
   = note: required for `Sleep` to implement `Unpin`

O compilador diz que Sleep não pode ser “despinado” (Unpin) e até sugere a solução. O std::pin::pin! (ou o tokio::pin!) pina a future ali mesmo, na pilha. Agora o mesmo prazo é compartilhado por todas as rodadas do laço:

use std::pin::pin;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(8);
    tokio::spawn(async move {
        for order in 1..=3 {
            tx.send(order).await.unwrap();
            sleep(Duration::from_millis(10)).await;
        }
        sleep(Duration::from_secs(5)).await; // depois a loja fica em silêncio
    });

    let mut deadline = pin!(sleep(Duration::from_millis(300)));

    loop {
        tokio::select! {
            _ = &mut deadline => {
                println!("hora de fechar");
                break;
            }
            Some(order) = rx.recv() => println!("recebi o pedido #{order}"),
        }
    }
}

Saída:

recebi o pedido #1
recebi o pedido #2
recebi o pedido #3
hora de fechar

▶ Experimente você mesmo

Se escrevêssemos sleep(...) dentro do select!, cada pedido novo começaria um timer novo de 300ms, e uma loja movimentada nunca fecharia. Pinar um único prazo fora do laço resolve isso.

Na recursão, o compilador precisa saber o tamanho da future, e uma future que contém a si mesma teria um tamanho infinito. Chamar countdown(n - 1).await dentro de async fn countdown dá:

error[E0733]: recursion in an async fn requires boxing
  |
  = note: a recursive `async fn` call must introduce indirection such as `Box::pin` to avoid an infinitely sized future

Ou seja: recursão numa async fn exige um Box. O Box::pin move a future interna para o heap e a pina lá. A future de fora agora só guarda um ponteiro, que tem tamanho fixo:

async fn countdown(n: u32) {
    if n == 0 {
        println!("decolar!");
        return;
    }
    println!("{n}...");
    Box::pin(countdown(n - 1)).await;
}

#[tokio::main]
async fn main() {
    countdown(3).await;
}

Saída:

3...
2...
1...
decolar!

▶ Experimente você mesmo

Sinceramente, isso é tudo de que a maioria das pessoas precisa: pin! para uma future da qual você faz poll várias vezes, Box::pin para guardar futures ou para recursão. Se quiser a história completa, o capítulo do Rust Book na lista “Para saber mais” explica passo a passo.

async fn em traits e closures async

Desde o Rust 1.75 você pode escrever async fn direto numa trait. Desde o Rust 1.85 você também tem closures async, escritas como async |x| ..., com as traits correspondentes AsyncFn, AsyncFnMut e AsyncFnOnce (versões async das traits de closures). Aqui estão as duas coisas num exemplo só:

use std::time::Duration;
use tokio::time::sleep;

trait Notifier {
    async fn notify(&self, user: &str) -> String;
}

struct Email;

impl Notifier for Email {
    async fn notify(&self, user: &str) -> String {
        sleep(Duration::from_millis(10)).await; // finja que está falando com um servidor de e-mail
        format!("e-mail enviado para {user}")
    }
}

async fn for_each_user(users: &[&str], job: impl AsyncFn(&str) -> String) {
    for user in users {
        println!("{}", job(user).await);
    }
}

#[tokio::main]
async fn main() {
    let email = Email;
    let users = ["ana", "ben"];

    for_each_user(&users, async |user: &str| email.notify(user).await).await;
}

Saída:

e-mail enviado para ana
e-mail enviado para ben

▶ Experimente você mesmo

Tem um porém com async fn em traits. A trait não promete que a future devolvida é Send, e o tokio::spawn precisa de futures Send (elas podem mudar de thread). Por isso, criar uma tarefa que chama o método através de um genérico N: Notifier falha:

fn send_in_background<N: Notifier + Send + Sync + 'static>(notifier: N) {
    tokio::spawn(async move {
        let msg = notifier.notify("ana").await;
        println!("{msg}");
    });
}

error: future cannot be sent between threads safely
  |
7 |         let msg = notifier.notify("ana").await;
  |                   ^^^^^^^^^^^^^^^^^^^^^^ await occurs here on type `impl Future<Output = String>`, which is not `Send`
  |
help: `Send` can be made part of the associated future's guarantees for all implementations of `Notifier::notify`
  |
2 -     async fn notify(&self, user: &str) -> String;
2 +     fn notify(&self, user: &str) -> impl std::future::Future<Output = String> + Send;

O erro diz que a future não pode ser enviada para outra thread com segurança, e a sugestão do compilador é a solução de sempre: escreva o método da trait como fn ... -> impl Future<Output = String> + Send. Quem implementa a trait ainda pode escrever uma async fn normal. Repare também que traits com async fn ainda não podem ser usadas como trait objects dyn Notifier. Se você precisar disso, o popular crate async-trait coloca as futures num Box para você.

Erros comuns

Bloquear o runtime. std::thread::sleep, cálculos pesados ou leituras grandes com std::fs dentro de uma tarefa async congelam a thread de trabalho inteira, e todas as outras tarefas nela ficam esperando. A página do básico mostrou a versão com sleep. Para trabalho pesado de CPU ou que bloqueia, passe-o para o tokio::task::spawn_blocking, que o roda num pool separado de threads feito para isso:

fn count_primes(limit: u64) -> usize {
    (2..limit)
        .filter(|&n| (2..).take_while(|d| d * d <= n).all(|d| n % d != 0))
        .count()
}

#[tokio::main]
async fn main() {
    let handle = tokio::task::spawn_blocking(|| count_primes(50_000));

    println!("ainda respondendo requisições enquanto contamos...");

    let primes = handle.await.unwrap();
    println!("existem {primes} números primos abaixo de 50000");
}

Saída:

ainda respondendo requisições enquanto contamos...
existem 5133 números primos abaixo de 50000

▶ Experimente você mesmo

Segurar o guard de um std::sync::Mutex através de um .await. Isto parece inofensivo:

let handle = tokio::spawn(async move {
    let mut count = visits.lock().unwrap();
    *count += 1;
    save_to_disk(*count).await;
});

error: future cannot be sent between threads safely
   |
   = help: within `{async block@src/main.rs:13:31: 13:41}`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, u32>`
note: future is not `Send` as this value is used across an await
   |
14 |         let mut count = visits.lock().unwrap();
   |             --------- has type `std::sync::MutexGuard<'_, u32>` which is not `Send`
15 |         *count += 1;
16 |         save_to_disk(*count).await;
   |                              ^^^^^ await occurs here, with `mut count` maybe used later

O compilador está te protegendo: o MutexGuard não é Send, e ele continua vivo durante o .await. Enquanto essa tarefa dorme no .await, ela ainda estaria segurando o lock, e qualquer outra tarefa que quisesse o lock bloquearia a thread inteira. A solução é descartar o guard antes do .await, colocando o lock num bloco próprio:

let count = {
    let mut count = visits.lock().unwrap();
    *count += 1;
    *count
}; // o guard é descartado aqui
save_to_disk(count).await;

Se você realmente precisa manter um lock enquanto espera, use o tokio::sync::Mutex, cujo lock().await espera sem bloquear a thread. (Para saber mais sobre locks, veja Estado compartilhado.)

Esquecer o .await. Futures são preguiçosas, então save_file(); sem .await não faz absolutamente nada. O compilador avisa com “futures do nothing unless you .await or poll them” (futures não fazem nada a menos que você use .await ou faça poll delas). Veja Async/Await para a história completa, e não ignore esse aviso.

Criar uma tarefa e nunca esperar por ela. Quando a main retorna, o runtime é desligado e toda tarefa que não terminou é descartada:

use std::time::Duration;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    tokio::spawn(async {
        sleep(Duration::from_millis(100)).await;
        println!("relatório salvo"); // nunca é impresso!
    });

    println!("main terminou");
}

Saída:

main terminou

▶ Experimente você mesmo

O relatório nunca é salvo. Se o trabalho importa, guarde o JoinHandle e faça .await nele (ou coloque as tarefas num JoinSet e esvazie o set) antes que a main termine.

Prática

1. Consulta de preços com prazo

Quatro lojas respondem a um pedido de preço depois de 20, 40, 300 e 60 milissegundos. Pergunte às quatro ao mesmo tempo com um JoinSet, mas dê a cada loja só 150ms. Imprima os preços que chegaram (ordenados pelo nome da loja) e depois quantas lojas estouraram o tempo. Dica: coloque timeout(limit, ask_shop(name, delay)) no set. Cada tarefa terminada vai te dar um Ok ou um Err(Elapsed), por isso devolva o nome da loja junto com esse resultado.

Ver solução

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::{sleep, timeout};

async fn ask_shop(delay_ms: u64) -> u32 {
    sleep(Duration::from_millis(delay_ms)).await;
    100 + delay_ms as u32 // um preço inventado
}

#[tokio::main]
async fn main() {
    let shops = [("Lojinha da Esquina", 20), ("MegaMart", 40), ("SlowCo", 300), ("Bazar do Zé", 60)];
    let limit = Duration::from_millis(150);

    let mut set = JoinSet::new();
    for (name, delay) in shops {
        set.spawn(async move { (name, timeout(limit, ask_shop(delay)).await) });
    }

    let mut prices = Vec::new();
    let mut timed_out = 0;
    while let Some(res) = set.join_next().await {
        match res.unwrap() {
            (name, Ok(price)) => prices.push((name, price)),
            (_, Err(_)) => timed_out += 1,
        }
    }

    prices.sort();
    for (name, price) in prices {
        println!("{name}: {price}");
    }
    println!("lojas que estouraram o tempo: {timed_out}");
}

Saída:

Bazar do Zé: 160
Lojinha da Esquina: 120
MegaMart: 140
lojas que estouraram o tempo: 1

▶ Experimente você mesmo

Todas as lojas esperam ao mesmo tempo, então a consulta inteira leva uns 150ms, não importa o quanto a SlowCo seja lenta. Quando o timeout dela dispara, a future dela é descartada, então ela não tem como continuar rodando em segundo plano.

2. Tamanho de pastas com recursão async

Uma Folder (pasta) tem um nome, uma lista de tamanhos de arquivos em KB e uma lista de subpastas. Escreva async fn total_size(folder: &Folder) -> u64, que soma os arquivos da própria pasta mais o tamanho de cada subpasta, chamando a si mesma para cada uma. Finja que cada pasta precisa primeiro de uma rápida “leitura de disco” async, com tokio::task::yield_now().await. Dica: a chamada recursiva precisa de um Box::pin(...) em volta, senão você vai receber o erro E0733.

Ver solução

struct Folder {
    name: &'static str,
    files_kb: Vec<u64>,
    subfolders: Vec<Folder>,
}

async fn total_size(folder: &Folder) -> u64 {
    tokio::task::yield_now().await; // finja que está lendo o disco

    let mut total: u64 = folder.files_kb.iter().sum();
    for sub in &folder.subfolders {
        total += Box::pin(total_size(sub)).await;
    }
    println!("{}: {} KB", folder.name, total);
    total
}

#[tokio::main]
async fn main() {
    let home = Folder {
        name: "home",
        files_kb: vec![4],
        subfolders: vec![
            Folder { name: "fotos", files_kb: vec![2048, 1024], subfolders: vec![] },
            Folder {
                name: "código",
                files_kb: vec![12, 8],
                subfolders: vec![Folder { name: "target", files_kb: vec![500], subfolders: vec![] }],
            },
        ],
    };

    let total = total_size(&home).await;
    println!("total: {total} KB");
}

Saída:

fotos: 3072 KB
target: 500 KB
código: 520 KB
home: 3596 KB
total: 3596 KB

▶ Experimente você mesmo

Cada pasta só imprime depois de todas as suas subpastas, porque primeiro ela precisa esperar os tamanhos delas. O Box::pin coloca a future de cada chamada aninhada no heap, e assim a future de fora continua com tamanho fixo.

3. Esvaziar a fila e depois parar

Mude o exemplo do sinal de desligamento para que ele não precise do sleep(100ms) antes de mandar o sinal de parada. Envie as três tarefas e o sinal de parada de uma vez, mas garanta que o worker ainda termine todas as tarefas da fila antes de parar. Dica: sem ajuda, o select! escolhe um ramo ao acaso quando os dois estão prontos. Adicione biased; e coloque o ramo das tarefas primeiro.

Ver solução

use tokio::sync::{mpsc, oneshot};

#[tokio::main]
async fn main() {
    let (job_tx, mut job_rx) = mpsc::channel::<String>(10);
    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();

    for job in ["redimensionar foto", "enviar e-mail", "atualizar índice de busca"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    stop_tx.send(()).unwrap();

    let worker = tokio::spawn(async move {
        let mut done = 0;
        loop {
            tokio::select! {
                biased;
                Some(job) = job_rx.recv() => {
                    println!("trabalhando em: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("a fila está vazia, parando");
                    break;
                }
            }
        }
        done
    });

    let done = worker.await.unwrap();
    println!("o worker concluiu {done} tarefas");
}

Saída:

trabalhando em: redimensionar foto
trabalhando em: enviar e-mail
trabalhando em: atualizar índice de busca
a fila está vazia, parando
o worker concluiu 3 tarefas

▶ Experimente você mesmo

Com biased;, o select! sempre tenta primeiro o ramo das tarefas, então só olha para o sinal de parada quando o recv() não tem nada pronto. Sem isso, o worker poderia parar depois de uma ou duas tarefas, um bug que só apareceria de vez em quando.

Resumo

• O join! espera todas as futures, o try_join! para no primeiro Err, e o JoinSet roda muitas tarefas criadas com spawn e devolve os resultados conforme elas terminam (ordene se a ordem importar).
• O select! espera a primeira future terminar. As outras são descartadas, e é assim que o cancelamento funciona no Rust async. Dentro de um laço, use futures cancel-safe e biased; quando a ordem importar.
• O timeout desiste de uma future lenta. O interval dá ticks num ritmo constante, o sleep espera uma vez.
• Uma stream é um iterador async. O receptor de um canal funciona como uma, e o tokio-stream adiciona map, take e companhia.
• O Pin promete que uma future não vai se mover na memória. Use pin! para fazer poll de uma future várias vezes e Box::pin para futures guardadas e recursão async.
• Não bloqueie o runtime (spawn_blocking), não segure um lock do std através de um .await e espere pelas tarefas que você cria.

Teste o que aprendeu

Teste o que aprendeu

1. No tokio::select!, o que acontece com os ramos que não terminam primeiro?

Continuam rodando em segundo plano São descartados, então param e fazem a limpeza Rodam de novo na próxima volta do laço

2. O que o tokio::time::timeout devolve quando o tempo acaba?

Ele entra em panic Ok com um valor padrão Err(Elapsed), e a future interna é descartada

3. Por que você precisa de Box::pin quando uma async fn chama a si mesma?

Uma future que contém a si mesma teria tamanho infinito Recursão não é permitida em código async de jeito nenhum Para fazer a função rodar em outra thread

4. Você precisa fazer 2 segundos de trabalho pesado de CPU dentro de um programa tokio. Qual é a melhor ferramenta?

std::thread::sleep em volta dele tokio::select! Um JoinSet com tokio::spawn tokio::task::spawn_blocking

Pontuação: 0 / 4

Para saber mais

• Select — o tutorial do Tokio (em inglês): o select! em detalhes: como os ramos funcionam, o que significa cancelar e como usá-lo num laço.
• Streams — o tutorial do Tokio (em inglês): como usar, adaptar e construir streams com o tokio-stream.
• A Closer Look at the Traits for Async — The Rust Book (em inglês): um olhar calmo, passo a passo, sobre Future, Pin, Unpin e Stream.
• std::pin — biblioteca padrão do Rust (em inglês): a explicação oficial de pinning, Pin, Unpin e da macro pin!.
• tokio::time no docs.rs (em inglês): sleep, timeout, interval e o que acontece com ticks perdidos.