Async a fondo en Rust: select!, timeouts, streams y Pin

En la página de Async/Await aprendiste lo básico: una async fn devuelve un future perezoso, .await lo ejecuta, #[tokio::main] arranca el runtime, y tokio::join! y tokio::spawn ejecutan cosas al mismo tiempo. Si algo de eso te suena nuevo, lee primero esa página. Esta página continúa a partir de ahí.

Los programas async de verdad necesitan unas cuantas herramientas más. Vuelve a pensar en una cocina con mucho trabajo. A veces esperas a que salgan todos los platos (join!). A veces solo te importa lo primero que pase: la comida está lista o el cliente se va (select!). A veces te rindes después de cinco minutos (timeout). Y a veces los pedidos no dejan de llegar, uno tras otro, toda la noche (streams). Esta página cubre todo eso y, además, Pin, esa palabra rara que tarde o temprano verás en los mensajes de error, explicada con palabras sencillas.

Todos los ejemplos usan tokio con features = ["full"], igual que en la página básica. Los botones “Pruébalo tú mismo” funcionan, porque el Playground de Rust trae tokio incluido.

Ejecutar muchas cosas a la vez

Ya conoces tokio::join!: espera a todos sus futures y te devuelve sus resultados en una tupla. Su hermano tokio::try_join! es para futures que devuelven un Result. Te da Ok con todos los valores, o el primer Err, y deja de esperar a los demás en cuanto uno falla:

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!("el usuario {id} no tiene ajustes"));
    }
    Ok(String::from("modo oscuro"))
}

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

Salida:

usuario 1: perfil #1, modo oscuro
usuario 2 falló: el usuario 2 no tiene ajustes

▶ Pruébalo tú mismo

join! y try_join! ejecutan los futures dentro de la tarea actual, por turnos. Eso es perfecto para un puñado de futures que conoces de antemano. Cuando tienes una lista de trabajos cuyo tamaño solo sabrás en tiempo de ejecución, lánzalos como tareas separadas. Podrías meter cada JoinHandle en un Vec, como en la página básica. Pero tokio tiene una herramienta más cómoda para esto: JoinSet, una bolsa de tareas que te entrega cada resultado en cuanto 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; // imagina que esto es una llamada de red
    (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()); // Err solo si la tarea hizo panic
    }

    results.sort(); // el orden de llegada puede cambiar de una ejecución a otra
    for (site, ms) in results {
        println!("{site} respondió en {ms}ms");
    }
}

Salida:

crates.io respondió en 20ms
docs.rs respondió en 10ms
rust-lang.org respondió en 30ms

▶ Pruébalo tú mismo

join_next() devuelve None cuando el conjunto está vacío, así que el bucle while let termina solo. Los resultados llegan en orden de finalización, no en el orden en que lanzaste las tareas. Cuando las tareas tardan más o menos lo mismo, ese orden puede cambiar entre ejecuciones, por eso ordenamos antes de imprimir. Otra cosa buena: si haces drop de un JoinSet (lo descartas), se cancelan todas las tareas que sigan dentro. Nada se queda ejecutándose por accidente.

Carreras de futures con `select!`

join! espera a todos. tokio::select! espera al primero. Cada rama tiene un patrón, un future y un bloque de código: patrón = future => código. El future que termine primero ejecuta su código, y las demás ramas se cancelan.

¿Qué significa “cancelar” en Rust? Al future que pierde simplemente se le hace drop, como a cualquier otro valor que sale de su ámbito. No vuelve a ejecutarse nunca. Sus variables locales también se descartan, así que se ejecuta su código de limpieza. Este ejemplo usa un pequeño tipo con Drop para que lo veas:

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

struct Download(&'static str);

impl Drop for Download {
    fn drop(&mut self) {
        println!("limpiando la descarga de {}", self.0);
    }
}

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

#[tokio::main]
async fn main() {
    tokio::select! {
        winner = download("Europa", 50) => println!("ganador: {winner}"),
        winner = download("Asia", 1000) => println!("ganador: {winner}"),
    }
    println!("main sigue adelante");
}

Salida:

descarga de Europa terminada
limpiando la descarga de Europa
limpiando la descarga de Asia
ganador: Europa
main sigue adelante

▶ Pruébalo tú mismo

Fíjate en lo que falta: “descarga de Asia terminada” nunca se imprime. El future de Asia se detuvo en su .await y se descartó, y su valor Download se limpió en ese mismo momento. Así funciona la cancelación en todo el código async de Rust: dejas de hacer poll a un future y lo descartas. No hace falta ninguna llamada especial para “cancelar”.

Un uso clásico de select! es un timeout (tiempo límite): pones tu trabajo a competir contra un temporizador.

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

Esto funciona bien, pero hay un atajo para ello, tokio::time::timeout, que verás en la siguiente sección.

Trabajar hasta recibir una señal de apagado

select! brilla de verdad dentro de un loop. Aquí un worker (trabajador) toma tareas de un canal, pero también escucha un mensaje de “por favor, detente” en un segundo canal. Un canal oneshot lleva exactamente un mensaje, y eso lo convierte en una señal de apagado perfecta:

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!("trabajando en: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("señal de apagado recibida");
                    break;
                }
            }
        }
        done
    });

    for job in ["redimensionar foto", "enviar correo", "actualizar índice de búsqueda"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    sleep(Duration::from_millis(100)).await; // le damos tiempo al worker para terminar
    stop_tx.send(()).unwrap();

    let done = worker.await.unwrap();
    println!("el worker terminó {done} tareas");
}

Salida:

trabajando en: redimensionar foto
trabajando en: enviar correo
trabajando en: actualizar índice de búsqueda
señal de apagado recibida
el worker terminó 3 tareas

▶ Pruébalo tú mismo

Algunos detalles que vale la pena conocer:

Escribimos &mut stop_rx, no stop_rx. En cada vuelta del bucle, select! necesita otra vez el mismo receptor, así que lo prestamos en lugar de moverlo a la primera ronda.
En cada vuelta del bucle, job_rx.recv() crea un future nuevo, y el que pierde se descarta. No pasa nada, porque recv() es cancel-safe (seguro ante cancelación): descartarlo nunca hace que se pierda un mensaje.
Si el patrón no coincide (aquí, cuando recv() devuelve None porque ya no queda ningún emisor), select! desactiva esa rama y sigue esperando a las demás.
Si varias ramas están listas en el mismo momento, select! elige una al azar, para que ninguna rama acapare el bucle. Escribe biased; como primera línea dentro de select! para que revise las ramas de arriba abajo.

La seguridad ante cancelación, en un párrafo. Como a la rama perdedora se le hace drop, cualquier trabajo que tuviera a medias se pierde. Para recv() en un canal o para sleep() eso no hace daño: todavía no se había tomado nada. Pero imagina un future que lee 100 bytes de un socket de red y ya ha leído 60 cuando lo descartan. Esos 60 bytes se pierden. Un future así no es cancel-safe. La documentación de tokio indica en cada método si es cancel-safe o no. Cuando uses select! dentro de un bucle, revisa los métodos que llamas, o crea el future una sola vez fuera del bucle y sigue haciendo poll a ese mismo future (verás cómo en la sección de Pin).

Timeouts y temporizadores

tokio::time::timeout(duration, future) envuelve cualquier future. Si el future termina a tiempo, obtienes Ok(value). Si el tiempo se acaba antes, el future se descarta (se cancela, igual que una rama perdedora de select!) y obtienes 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}: el precio es {price}"),
            Err(e) => println!("{server}: {e}"),
        }
    }
}

Salida:

servidor rápido: el precio es 42
servidor lento: deadline has elapsed

▶ Pruébalo tú mismo

El texto “deadline has elapsed” (“se acabó el tiempo límite”) sale en inglés porque es el mensaje del propio error Elapsed de tokio.

Ponle un timeout a todo lo que hable con el mundo exterior: llamadas HTTP, consultas a la base de datos, lecturas de un socket. Un servidor que nunca responde no debería congelar tu programa para siempre.

sleep frente a interval. sleep espera una sola vez. interval te da un ritmo constante, como un metrónomo. Llamas a .tick().await en un bucle, y cada tick llega a su hora. El primer tick se dispara de inmediato:

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; // el primer tick se completa de inmediato
        println!("latido {beat}");
    }
}

Salida:

latido 1
latido 2
latido 3

▶ Pruébalo tú mismo

¿Por qué no hacer simplemente sleep de 50ms en un bucle? Porque entonces cada vuelta tarda “tu trabajo + 50ms”, y el horario se va desfasando poco a poco. Un interval mide desde el inicio previsto de cada tick, así que si tu trabajo tarda 10ms, solo espera los 40ms que faltan. Usa interval para latidos (heartbeats), consultas periódicas y tareas tipo “guardar cada minuto”. Usa sleep para una sola pausa.

Streams: iteradores async

Un iterador te da valores uno a uno con next(). Un stream (flujo) es la versión async: next().await te da el siguiente valor cuando está listo, y None cuando ya no hay más. Piensa en un panel con precios de acciones, filas que llegan de una base de datos o mensajes de un servidor de chat.

En realidad ya has usado algo parecido a un stream: el receptor de un canal de tokio. rx.recv().await espera el siguiente mensaje y devuelve None cuando todos los emisores han desaparecido:

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; // una lectura cada 10ms
        }
        // aquí se descarta tx, y eso cierra el canal
    });

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

Salida:

temperatura: 21°C
temperatura: 22°C
temperatura: 24°C
temperatura: 23°C
sensor detenido

▶ Pruébalo tú mismo

La biblioteca estándar todavía no tiene un trait Stream (aún se está diseñando). El ecosistema usa el del crate futures, y el crate auxiliar de tokio, tokio-stream, añade un trait StreamExt con los métodos que ya conoces de los iteradores: next(), map(), filter(), take() y más. Añádelo a tu proyecto:

[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}");
    }
}

Esto imprime 200, 204 y 196. Se lee igual que el código con iteradores, salvo por el .await en next(). Para convertir el receptor de un canal de tokio en un stream de verdad, envuélvelo: tokio_stream::wrappers::ReceiverStream::new(rx). (El crate futures tiene su propio StreamExt con la misma idea. Elige uno por archivo, porque los dos traits tienen un método next() y chocan entre sí.)

Pin, en palabras sencillas

Tarde o temprano verás Pin en un tipo o en un mensaje de error. Esta es la idea, sin la teoría profunda.

Cuando Rust compila una async fn, la convierte en un struct oculto, una máquina de estados. Cada variable local que todavía se necesita después de un .await se convierte en un campo de ese struct. Ahora imagina que una variable local toma prestada otra, como let name = String::from("Ana"); let r = &name; con un .await en medio. Entonces el struct guarda un puntero a uno de sus propios campos. Si movieras el struct a otro lugar de la memoria, ese puntero apuntaría al sitio viejo, ya vacío.

Pin es la solución. Un valor fijado (pinned) es una promesa: “este valor ya no se moverá en memoria”. Una vez que un future está fijado, es seguro hacerle poll. Casi siempre ni lo notas, porque .await fija el future por ti.

Te encontrarás con Pin en tres situaciones:

Hacer poll al mismo future muchas veces, por ejemplo, una fecha límite en un bucle con select!.
Guardar futures en un Vec o en un campo de un struct, normalmente como Pin<Box<dyn Future<Output = T>>>.
Recursión en una async fn.

Para la primera, esto es lo que pasa si intentas reutilizar un sleep en un bucle sin fijarlo:

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`

El compilador dice que Sleep no se puede “desfijar” (no implementa Unpin) y hasta te sugiere la solución. std::pin::pin! (o tokio::pin!) fija el future en su sitio, en la pila. Ahora todas las vueltas del bucle comparten la misma fecha límite:

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; // y luego la tienda se queda tranquila
    });

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

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

Salida:

pedido recibido #1
pedido recibido #2
pedido recibido #3
hora de cerrar

▶ Pruébalo tú mismo

Si escribiéramos sleep(...) dentro del select!, cada pedido nuevo arrancaría un temporizador nuevo de 300ms, y una tienda con mucho movimiento no cerraría nunca. Fijar una sola fecha límite fuera del bucle lo resuelve.

En la recursión, el compilador necesita saber qué tamaño tiene el future, y un future que se contiene a sí mismo sería infinitamente grande. Llamar a countdown(n - 1).await dentro de async fn countdown produce:

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

Es decir: la recursión en una async fn necesita una indirección como Box::pin para no crear un future de tamaño infinito. Box::pin mueve el future interior al heap y lo fija allí. El future exterior ahora solo guarda un puntero, que tiene un tamaño fijo:

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

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

Salida:

3...
2...
1...
¡despegue!

▶ Pruébalo tú mismo

Sinceramente, eso es todo lo que la mayoría de la gente necesita: pin! para un future al que haces poll muchas veces, y Box::pin para guardar futures o para la recursión. Si quieres la historia completa, el capítulo del libro de Rust de la lista “Para leer más” lo explica paso a paso.

`async fn` en traits y closures async

Desde Rust 1.75 puedes escribir async fn directamente en un trait. Desde Rust 1.85 también tienes closures async, que se escriben async |x| ..., con sus traits correspondientes AsyncFn, AsyncFnMut y AsyncFnOnce (versiones async de los traits de closures). Aquí tienes las dos cosas en un solo ejemplo:

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; // imagina que hablamos con un servidor de correo
        format!("correo enviado a {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;
}

Salida:

correo enviado a ana
correo enviado a ben

▶ Pruébalo tú mismo

Hay una trampa con async fn en traits. El trait no promete que el future que devuelve sea Send, y tokio::spawn necesita futures Send (pueden moverse entre hilos). Por eso falla lanzar una llamada a través de un genérico N: Notifier:

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;

El error dice que el future no se puede enviar entre hilos de forma segura, y la ayuda (help) propone hacer que Send forme parte de la garantía del método. Esa sugerencia del compilador es la solución habitual: escribe el método del trait como fn ... -> impl Future<Output = String> + Send. Quienes implementen el trait pueden seguir escribiendo una async fn normal. Ten en cuenta también que los traits con async fn todavía no se pueden usar como trait objects dyn Notifier. Si lo necesitas, el popular crate async-trait mete los futures en un Box por ti.

Errores comunes

Bloquear el runtime. std::thread::sleep, los cálculos pesados o las lecturas grandes con std::fs dentro de una tarea async congelan todo el hilo worker, y todas las demás tareas de ese hilo se quedan esperando. La página básica mostró la versión con sleep. Para trabajo pesado de CPU o que bloquea, pásalo a tokio::task::spawn_blocking, que lo ejecuta en un grupo aparte de hilos pensado justo para eso:

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!("sigo respondiendo peticiones mientras contamos...");

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

Salida:

sigo respondiendo peticiones mientras contamos...
hay 5133 números primos por debajo de 50000

▶ Pruébalo tú mismo

Mantener el guard de un std::sync::Mutex a través de un .await. Esto 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

El compilador te está protegiendo: el MutexGuard no es Send y se usa a través de un .await. Mientras esta tarea duerme en el .await, seguiría teniendo el lock, y cualquier otra tarea que quisiera el lock bloquearía todo su hilo. La solución es soltar el guard antes del .await, poniendo el bloqueo en su propio bloque:

let count = {
    let mut count = visits.lock().unwrap();
    *count += 1;
    *count
}; // aquí se descarta el guard
save_to_disk(count).await;

Si de verdad necesitas mantener un lock mientras esperas, usa tokio::sync::Mutex, cuyo lock().await espera sin bloquear el hilo. (Para saber más sobre locks, mira Estado compartido.)

Olvidar el .await. Los futures son perezosos, así que save_file(); sin .await no hace absolutamente nada. El compilador te avisa con “futures do nothing unless you .await or poll them” (los futures no hacen nada a menos que les hagas .await o poll). Mira Async/Await para la historia completa, y no ignores ese aviso.

Lanzar una tarea y no esperarla nunca. Cuando main termina, el runtime se apaga y todas las tareas sin terminar se descartan:

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

#[tokio::main]
async fn main() {
    tokio::spawn(async {
        sleep(Duration::from_millis(100)).await;
        println!("informe guardado"); // ¡nunca se imprime!
    });

    println!("main ha terminado");
}

Salida:

main ha terminado

▶ Pruébalo tú mismo

El informe nunca se guarda. Si el trabajo importa, guarda el JoinHandle y hazle .await (o mete las tareas en un JoinSet y vacíalo) antes de que termine main.

Práctica

1. Consulta de precios con fecha límite

Cuatro tiendas responden a una petición de precio después de 20, 40, 300 y 60 milisegundos. Pregunta a las cuatro al mismo tiempo con un JoinSet, pero dale a cada tienda solo 150ms. Imprime los precios que llegaron (ordenados por nombre de tienda) y luego cuántas tiendas no respondieron a tiempo. Pista: lanza timeout(limit, ask_shop(name, delay)) dentro del conjunto. Cada tarea terminada te dará entonces un Ok o un Err(Elapsed), así que devuelve el nombre de la tienda junto con ese resultado.

Ver solución

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 // un precio inventado
}

#[tokio::main]
async fn main() {
    let shops = [("La Esquina", 20), ("MegaMart", 40), ("LentoCo", 300), ("Mil Cosas", 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!("{timed_out} tienda(s) no respondieron a tiempo");
}

Salida:

La Esquina: 120
MegaMart: 140
Mil Cosas: 160
1 tienda(s) no respondieron a tiempo

▶ Pruébalo tú mismo

Todas las tiendas esperan al mismo tiempo, así que la consulta completa tarda unos 150ms, por muy lenta que sea LentoCo. Cuando salta su timeout, su future se descarta, así que no puede seguir ejecutándose en segundo plano.

2. Tamaño de carpetas con recursión async

Una Folder tiene un nombre, una lista de tamaños de archivo en KB y una lista de subcarpetas. Escribe async fn total_size(folder: &Folder) -> u64 que sume sus propios archivos más el tamaño de cada subcarpeta, llamándose a sí misma para cada una. Imagina que cada carpeta necesita primero una breve “lectura de disco” async, con tokio::task::yield_now().await. Pista: la llamada recursiva necesita Box::pin(...) a su alrededor, o te saldrá el error E0733.

Ver solución

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

async fn total_size(folder: &Folder) -> u64 {
    tokio::task::yield_now().await; // imagina que leemos el 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: "codigo",
                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");
}

Salida:

fotos: 3072 KB
target: 500 KB
codigo: 520 KB
home: 3596 KB
total: 3596 KB

▶ Pruébalo tú mismo

Cada carpeta se imprime después de todas sus subcarpetas, porque primero tiene que esperar a conocer sus tamaños. Box::pin pone el future de cada llamada anidada en el heap, así el future exterior mantiene un tamaño fijo.

3. Vaciar la cola y luego parar

Cambia el ejemplo del apagado para que no necesite el sleep(100ms) antes de enviar la señal de parada. Envía las tres tareas y la señal de parada enseguida, pero asegúrate de que el worker siga terminando todas las tareas de la cola antes de detenerse. Pista: si no le ayudas, select! elige una rama al azar cuando las dos están listas. Añade biased; y pon primero la rama de las tareas.

Ver solución

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 correo", "actualizar índice de búsqueda"] {
        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!("trabajando en: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("la cola está vacía, me detengo");
                    break;
                }
            }
        }
        done
    });

    let done = worker.await.unwrap();
    println!("el worker terminó {done} tareas");
}

Salida:

trabajando en: redimensionar foto
trabajando en: enviar correo
trabajando en: actualizar índice de búsqueda
la cola está vacía, me detengo
el worker terminó 3 tareas

▶ Pruébalo tú mismo

Con biased;, select! siempre prueba primero la rama de las tareas, así que solo mira la señal de parada cuando recv() no tiene nada listo. Sin eso, el worker podría detenerse después de una o dos tareas, un bug que solo aparecería de vez en cuando.

Resumen

join! espera a todos los futures, try_join! se detiene en el primer Err, y JoinSet ejecuta muchas tareas lanzadas y te devuelve los resultados a medida que terminan (ordénalos si el orden importa).
select! espera a que termine el primer future. A los demás se les hace drop, y así es como funciona la cancelación en el código async de Rust. Dentro de un bucle, usa futures cancel-safe y biased; cuando el orden importe.
timeout abandona un future lento. interval marca un ritmo constante; sleep espera una sola vez.
Un stream es un iterador async. El receptor de un canal funciona como uno, y tokio-stream añade map, take y compañía.
Pin promete que un future no se moverá en memoria. Usa pin! para hacer poll a un mismo future muchas veces y Box::pin para futures guardados y para la recursión async.
No bloquees el runtime (spawn_blocking), no mantengas un lock de std a través de un .await, y espera a las tareas que lanzas.

Comprueba lo que aprendiste

Comprueba lo aprendido

1. En tokio::select!, ¿qué pasa con las ramas que no terminan primero?

2. ¿Qué devuelve tokio::time::timeout cuando se acaba el tiempo?

3. ¿Por qué necesitas Box::pin cuando una async fn se llama a sí misma?

4. Tienes que hacer 2 segundos de trabajo pesado de CPU dentro de un programa con tokio. ¿Cuál es la mejor herramienta?

Puntuación: 0 / 4

Para leer más

Select — el tutorial de Tokio (en inglés): select! en detalle: cómo funcionan las ramas, qué significa cancelar y cómo usarlo en un bucle.
Streams — el tutorial de Tokio (en inglés): cómo usar, adaptar y construir streams con tokio-stream.
A Closer Look at the Traits for Async — El libro de Rust (en inglés): una explicación amable y paso a paso de Future, Pin, Unpin y Stream.
std::pin — la biblioteca estándar de Rust (en inglés): la explicación oficial del pinning, Pin, Unpin y la macro pin!.
tokio::time en docs.rs (en inglés): sleep, timeout, interval y cómo se manejan los ticks perdidos.