🦀 Curso de Rust — Nivel 5: Concurrencia y programación asíncrona

Objetivo: que tus estudiantes puedan escribir programas que hacen varias cosas a la vez (con hilos y con async), entiendan por qué el compilador de Rust les impide las carreras de datos, y sepan controlar esa concurrencia: cancelarla, ponerle plazos, frenarla cuando hay demasiada carga y limitarla.

Requisitos: Niveles 1 a 4. Este nivel se apoya mucho en ownership, borrowing y lifetimes (Nivel 3), en traits (Nivel 2) y en Result/Option (Nivel 4).

Idea central del nivel: En Rust, el compilador convierte los errores de concurrencia más peligrosos (las carreras de datos) en errores de compilación. Si compila, no tiene carreras de datos.

Índice

0. Puente: closures, lo mínimo
1. Mapa mental: concurrencia, paralelismo y async
2. Threads y std::thread
3. Send y Sync
4. Arc, Mutex y RwLock
5. Canales para comunicar tareas (std::sync::mpsc)
6. async y .await
7. Futures y runtime asíncrono
8. Tokio: tareas
9. Tokio: canales y sincronización
10. Tokio: cancelación
11. Timeouts, backpressure y límites de concurrencia
12. Proyecto integrador: ingesta de eventos de pedidos en tiempo real
13. Ejercicios con soluciones
14. Hoja de referencia rápida
15. Errores frecuentes
16. Plan de clases sugerido

Leyenda:

• ✅ = se puede / es correcto
• ❌ = NO se puede / no compila
• ⚠️ = compila, pero cuidado (puede fallar, bloquearse o dar resultados engañosos)

Aviso sobre la salida de los programas concurrentes: cuando varios hilos o tareas imprimen a la vez, el orden de las líneas puede cambiar de una ejecución a otra. En este curso indicamos cuándo una salida es estable y cuándo es solo “típica”. Es una buena oportunidad para que tus estudiantes vean la no-determinación con sus propios ojos: que ejecuten cada ejemplo varias veces.

Aviso sobre el alcance: los hilos y Tokio exigen closures (|| ...), que no habíamos visto. Por eso el Módulo 0 es un puente mínimo, como el que hicimos con ownership en el Nivel 2. Los iteradores y los combinadores (map, and_then…) siguen fuera de este nivel.

Módulo 0: Puente: closures, lo mínimo

Un closure es una función anónima que puede capturar variables de su entorno. Se escribe con barras verticales: |parámetros| cuerpo.

0.1 Sintaxis

fn main() {
    let sumar = |a: i32, b: i32| a + b;          // con tipos
    let doble = |x| x * 2;                       // tipos inferidos por el primer uso
    let saludar = || println!("hola");           // sin parámetros
    let compleja = |x: i32| {                    // con bloque
        let y = x * 10;
        y + 1
    };

    println!("{} {}", sumar(2, 3), doble(4));    // 5 8
    saludar();
    println!("{}", compleja(2));                 // 21
}

0.2 Capturar variables (lo que una función normal no puede)

fn main() {
    // 1) Captura por referencia compartida (&): solo lee
    let factor = 3;
    let multiplicar = |x| x * factor;
    println!("{}", multiplicar(5));              // 15

    // 2) Captura por referencia mutable (&mut): modifica
    let mut contador = 0;
    let mut incrementar = || contador += 1;      // el closure debe declararse `mut`
    incrementar();
    incrementar();
    println!("{contador}");                      // 2 (usable porque el closure ya no se usa)

    // 3) Captura por valor con `move`: se lleva la propiedad
    let nombre = String::from("Ana");
    let saludar = move || println!("hola, {nombre}");
    saludar();
    // println!("{nombre}");                     // ❌ E0382: `nombre` se movió al closure
}

move es la palabra más importante de este nivel. Hace que el closure sea dueño de lo que captura. Los hilos y las tareas la necesitan porque pueden vivir más que la función que las creó.

0.3 move con tipos Copy: se copia

fn main() {
    let mut n = 10;
    let mut sumar_uno = move || {
        n += 1;                                   // modifica SU copia
        n
    };
    println!("{}", sumar_uno());                  // 11
    println!("{}", sumar_uno());                  // 12
    println!("{n}");                              // 10  ← el original no cambió
}

(Esto vuelve a aparecer en el Módulo 2 como una trampa con hilos.)

0.4 Los tres traits de los closures

El compilador clasifica cada closure según cómo usa lo que captura:

Los parámetros que reciben closures usan estos traits como restricciones (Nivel 2):

fn aplicar<F: Fn(i32) -> i32>(f: F, valor: i32) -> i32 {
    f(valor)
}

fn ejecutar_una_vez<F: FnOnce() -> String>(f: F) -> String {
    f()
}

fn main() {
    let sumar_diez = |x| x + 10;
    println!("{}", aplicar(sumar_diez, 5));                 // 15

    let texto = String::from("consumido");
    let consumir = move || texto;                           // devuelve (mueve) el String
    println!("{}", ejecutar_una_vez(consumir));
}

Para este nivel basta recordar: thread::spawn y tokio::spawn piden un closure (o bloque async) que sea dueño de sus datos (move) y que se pueda enviar a otro hilo (Send).

0.5 Closures y errores comunes

Módulo 1: Mapa mental: concurrencia, paralelismo y async

1.1 Tres palabras que se confunden

1.2 ¿Qué problema se resuelve con cada herramienta?

1.3 Hilos vs. tareas async

1.4 Qué garantiza Rust (y qué no)

Rust llama a esto “concurrencia sin miedo” (fearless concurrency). Es importante ser honestos sobre su alcance:

Frase para clase: “Rust te garantiza que no corromperás la memoria con varios hilos. Que el programa haga lo correcto sigue siendo trabajo tuyo.”

1.5 El error que Rust evita: una carrera de datos

use std::thread;

fn main() {
    let mut contador = 0;

    // ❌ E0373: closure may outlive the current function, but it borrows `contador`
    // let h = thread::spawn(|| {
    //     contador += 1;
    // });
    // h.join().unwrap();

    // ⚠️ Con `move` compila... pero el hilo modifica SU COPIA (contador es Copy):
    let h = thread::spawn(move || {
        contador += 1;
        println!("[hilo] contador = {contador}");     // 1
    });
    h.join().expect("el hilo no debe hacer panic");
    println!("[main] contador = {contador}");         // 0  ← no cambió
}

En C o C++, esto compilaría y fallaría de formas impredecibles. En Rust te obliga a elegir explícitamente cómo compartir (módulo 4) o comunicar (módulo 5).

Módulo 2: Threads y std::thread

2.1 Crear un hilo con thread::spawn

use std::thread;
use std::time::Duration;

fn main() {
    let manejador = thread::spawn(|| {
        for i in 1..=3 {
            println!("[hilo] vuelta {i}");
            thread::sleep(Duration::from_millis(50));
        }
    });

    for i in 1..=3 {
        println!("[main] vuelta {i}");
        thread::sleep(Duration::from_millis(50));
    }

    manejador.join().expect("el hilo no debe hacer panic");
}

Salida típica (el orden entre [hilo] y [main] puede variar):

[main] vuelta 1
[hilo] vuelta 1
[main] vuelta 2
[hilo] vuelta 2
[main] vuelta 3
[hilo] vuelta 3

• thread::spawn recibe un closure y devuelve un JoinHandle<T>.
• join() espera a que el hilo termine y devuelve Result<T, _>: Ok(valor) con lo que devolvió el closure, o Err(_) si el hilo hizo panic.

⚠️ Si main termina, los hilos mueren

use std::thread;
use std::time::Duration;

fn main() {
    thread::spawn(|| {
        thread::sleep(Duration::from_millis(100));
        println!("esto probablemente NUNCA se imprime");
    });
    println!("main termina ya");
    // Sin join(), al terminar main el programa se cierra y mata el hilo.
}

Regla: si necesitas que un hilo termine, guarda su JoinHandle y llama a join().

2.2 move y la propiedad de los datos

El closure del hilo puede vivir más que la función que lo creó, así que no puede tomar prestadas variables locales:

use std::thread;

fn main() {
    let v = vec![1, 2, 3];

    // ❌ E0373: closure may outlive the current function, but it borrows `v`
    // let h = thread::spawn(|| {
    //     println!("{:?}", v);
    // });

    // ✅ `move`: el hilo se vuelve dueño de `v`
    let h = thread::spawn(move || {
        println!("{:?}", v);
    });

    // println!("{:?}", v);         // ❌ E0382: `v` se movió al hilo
    h.join().expect("el hilo no debe hacer panic");
}

La firma de spawn lo explica todo (la leerás mejor tras el Módulo 3):

pub fn spawn<F, T>(f: F) -> JoinHandle<T>
where
    F: FnOnce() -> T + Send + 'static,    // closure que se ejecuta una vez, enviable a otro hilo, sin préstamos locales
    T: Send + 'static,                    // el resultado también debe poder viajar y no depender de préstamos

• FnOnce: el closure se llama una vez.
• Send: se puede mover a otro hilo.
• 'static: no contiene referencias a datos locales que puedan desaparecer (Nivel 3: lifetimes).

2.3 Devolver valores desde un hilo

use std::thread;

fn main() {
    let h = thread::spawn(|| {
        let mut suma: u64 = 0;
        for i in 1..=1_000_000u64 {
            suma += i;
        }
        suma                                   // el valor de retorno del closure
    });

    let resultado = h.join().expect("el hilo no debe hacer panic");
    println!("{resultado}");                   // 500000500000
}

2.4 Varios hilos

use std::thread;

fn main() {
    let mut manejadores = Vec::new();

    for id in 0..4 {
        manejadores.push(thread::spawn(move || {
            id * id                            // cada hilo calcula su propio resultado
        }));
    }

    let mut resultados = Vec::new();
    for h in manejadores {
        resultados.push(h.join().expect("el hilo no debe hacer panic"));
    }
    println!("{:?}", resultados);              // [0, 1, 4, 9]   (en el orden de los manejadores)
}

Aunque los hilos corran en cualquier orden, recoger los resultados en el orden de los JoinHandle da una salida estable.

2.5 Si un hilo hace panic

El panic mata solo a ese hilo; los demás siguen. join() te avisa con Err:

use std::thread;

fn main() {
    let h = thread::spawn(|| {
        panic!("fallo dentro del hilo");
    });

    match h.join() {
        Ok(_) => println!("el hilo terminó bien"),
        Err(_) => println!("el hilo hizo panic, pero main sigue vivo"),
    }
}

2.6 Nombrar hilos y consultar el entorno

use std::thread;

fn main() {
    let h = thread::Builder::new()
        .name(String::from("trabajador-1"))
        .spawn(|| {
            println!("soy {:?}", thread::current().name());      // Some("trabajador-1")
        })
        .expect("no se pudo crear el hilo");          // Builder::spawn devuelve io::Result
    h.join().expect("el hilo no debe hacer panic");

    println!("{:?}", thread::current().name());        // Some("main")

    match thread::available_parallelism() {
        Ok(n) => println!("núcleos disponibles: {n}"),
        Err(e) => println!("no se pudo consultar: {e}"),
    }
}

Un hilo con nombre aparece en los mensajes de panic y en depuradores: muy útil.

2.7 Hilos con alcance: thread::scope

Los hilos normales exigen 'static (por eso move y Arc). Con thread::scope, los hilos garantizan terminar antes de que salga el bloque, así que pueden tomar prestados datos locales:

use std::thread;

fn main() {
    let datos = vec![1, 2, 3, 4, 5, 6];

    thread::scope(|s| {
        s.spawn(|| {
            println!("hilo A ve {} elementos", datos.len());      // préstamo & de `datos`
        });
        s.spawn(|| {
            println!("hilo B suma {}", datos.iter().sum::<i32>());   // otro préstamo &
        });
    });                                                           // aquí se hace join automático de A y B

    println!("{:?}", datos);                                      // ✅ `datos` sigue siendo nuestro
}

Modificar partes distintas a la vez (aliasing XOR mutabilidad, del Nivel 3)

use std::thread;

fn main() {
    let mut datos = vec![1, 2, 3, 4, 5, 6];
    let (izq, der) = datos.split_at_mut(3);          // dos &mut DISJUNTOS

    thread::scope(|s| {
        s.spawn(move || {
            for x in izq.iter_mut() {
                *x *= 2;
            }
        });
        s.spawn(move || {
            for x in der.iter_mut() {
                *x *= 10;
            }
        });
    });

    println!("{:?}", datos);                         // [2, 4, 6, 40, 50, 60]
}

Aquí no hace falta Arc ni Mutex: como cada hilo tiene su parte exclusiva, no hay nada que sincronizar. Esta es la forma más barata de paralelismo.

Dividir un cálculo entre hilos

use std::thread;

fn main() {
    let datos: Vec<u64> = (1..=1_000_000).collect();
    let hilos: usize = 4;
    let tamano = datos.len().div_ceil(hilos);

    let total: u64 = thread::scope(|s| {
        let mut manejadores = Vec::new();
        for trozo in datos.chunks(tamano) {                       // porciones (slices) de `datos`
            manejadores.push(s.spawn(move || trozo.iter().sum::<u64>()));
        }

        let mut suma = 0;
        for m in manejadores {
            suma += m.join().expect("el hilo no debe hacer panic");
        }
        suma
    });

    println!("{total}");                                           // 500000500000
}

Lo que scope NO permite

use std::thread;

fn main() {
    let mut x = 0;
    thread::scope(|s| {
        // ❌ E0499: dos hilos con &mut a la misma variable
        // s.spawn(|| x += 1);
        // s.spawn(|| x += 1);
        s.spawn(|| x += 1);                 // ✅ uno solo
    });
    println!("{x}");                        // 1
}

2.8 Cuándo usar cada forma

2.9 Coste de los hilos

• Cada hilo reserva una pila (en Rust, 2 MiB por defecto para hilos creados con spawn) y consume recursos del SO.
• Crear miles de hilos es mala idea: el rendimiento se degrada.
• No es gratis repartir trabajo: crear hilos, sincronizar y juntar resultados cuesta. Mide antes de asumir que paralelizar acelera.

Módulo 3: Send y Sync

Estos dos traits son el mecanismo con el que el compilador sabe qué es seguro usar entre hilos.

3.1 Definiciones

Relación: T es Sync si y solo si &T es Send.

Propiedades importantes:

• Son traits marcadores: no tienen métodos.
• Son traits automáticos (auto traits): el compilador los deriva solo. Un struct es Send si todos sus campos lo son; si un campo no lo es, el struct tampoco.
• Los tipos básicos (i32, String, Vec<T>…) son Send y Sync (si lo es su contenido).

3.2 Comprobarlo con el compilador

use std::cell::Cell;
use std::rc::Rc;
use std::sync::{Arc, Mutex};

fn exige_send<T: Send>() {}
fn exige_sync<T: Sync>() {}

fn main() {
    exige_send::<i32>();
    exige_send::<String>();
    exige_send::<Vec<i32>>();
    exige_send::<Arc<Mutex<i32>>>();
    exige_sync::<Arc<Mutex<i32>>>();
    exige_send::<Cell<i32>>();                      // ✅ Cell es Send...

    // exige_sync::<Cell<i32>>();                    // ❌ E0277: `Cell<i32>` cannot be shared between threads safely
    // exige_send::<Rc<i32>>();                      // ❌ E0277: `Rc<i32>` cannot be sent between threads safely
    // exige_sync::<Rc<i32>>();                      // ❌
    // exige_send::<*const i32>();                   // ❌ los punteros crudos no son Send
    // exige_send::<std::sync::MutexGuard<'static, i32>>();   // ❌ un guardia de Mutex no es Send
}

Es un buen ejercicio de clase: que prueben tipos y descubran cuáles pasan.

3.3 Tabla de tipos comunes

3.4 Por qué Rc no es Send

use std::rc::Rc;
use std::thread;

fn main() {
    let datos = Rc::new(5);

    // ❌ E0277: `Rc<i32>` cannot be sent between threads safely
    // let h = thread::spawn(move || {
    //     println!("{datos}");
    // });
    // h.join().unwrap();

    println!("{datos}");
}

Rc actualiza su contador de referencias con operaciones normales. Si dos hilos hicieran clone a la vez, el contador se corrompería y podría liberarse memoria dos veces o demasiado pronto. Por eso el compilador no permite enviarlo. La versión segura es Arc (Módulo 4).

El mensaje de error del compilador nombra exactamente el trait: “the trait Send is not implemented for Rc<i32>”. Los errores de Send/Sync son pistas, no obstáculos: te dicen qué tipo no sirve entre hilos.

3.5 Un struct hereda la propiedad de sus campos

use std::rc::Rc;

fn exige_send<T: Send>() {}

#[allow(dead_code)]
struct Bueno {
    nombre: String,
    edad: u32,
}

#[allow(dead_code)]
struct Malo {
    nombre: String,
    compartido: Rc<i32>,       // un solo campo !Send contagia al struct entero
}

fn main() {
    exige_send::<Bueno>();                 // ✅
    // exige_send::<Malo>();               // ❌ E0277: `Rc<i32>` cannot be sent between threads safely
}

3.6 unsafe impl Send/Sync: una promesa

Es posible escribir unsafe impl Send for MiTipo {}: le prometes al compilador que es seguro. Si te equivocas, tú eres responsable de la carrera de datos. Está fuera del alcance de este curso: la inmensa mayoría de los programas nunca lo necesitan.

3.7 Resumen del módulo

1. Send = se puede mover a otro hilo. Sync = se puede compartir por referencia entre hilos.
2. Son automáticos: se derivan de los campos.
3. Rc, Cell/RefCell (para Sync), punteros crudos y MutexGuard (para Send) no pasan.
4. Los errores de Send/Sync te señalan qué tipo rompe la regla.

Módulo 4: Arc, Mutex y RwLock

Hay dos grandes formas de coordinar hilos:

1. Compartir memoria (este módulo): varios hilos acceden al mismo dato, de forma sincronizada.
2. Pasar mensajes (módulo 5): los hilos se envían valores.

4.1 Arc<T>: propiedad compartida entre hilos

Arc = Atomic Reference Counted. Es como Rc, pero con contador atómico: puede cruzar hilos.

use std::sync::Arc;
use std::thread;

fn main() {
    let datos = Arc::new(vec![1, 2, 3]);
    let mut manejadores = Vec::new();

    for i in 0..3 {
        let datos = Arc::clone(&datos);               // nueva referencia (solo incrementa el contador)
        manejadores.push(thread::spawn(move || {
            let suma: i32 = datos.iter().sum();
            println!("hilo {i}: suma = {suma}");
        }));
    }

    for h in manejadores {
        h.join().expect("el hilo no debe hacer panic");
    }
    println!("referencias vivas: {}", Arc::strong_count(&datos));   // 1
}

• Arc::clone(&a) no copia los datos: solo suma 1 al contador. Los datos se liberan cuando el contador llega a 0.
• Se prefiere escribir Arc::clone(&a) en lugar de a.clone() para dejar claro que es barato.
• Un Arc<T> da acceso de solo lectura (&T). Para modificar, hace falta algo más.

❌ No se puede modificar el contenido de un Arc directamente

use std::sync::Arc;

fn main() {
    let a = Arc::new(vec![1, 2, 3]);
    // a.push(4);        // ❌ E0596: cannot borrow data in an `Arc` as mutable
    println!("{:?}", a);
}

Esto es aliasing XOR mutabilidad (Nivel 3): un Arc implica muchos dueños, así que no puede dar &mut. Para mutar, usamos mutabilidad interior con un Mutex.

4.2 Mutex<T>: exclusión mutua

Un Mutex (mutual exclusion) permite que un solo hilo a la vez acceda al dato protegido. Para usarlo, se pide el lock (cerrojo) y se obtiene un guardia (MutexGuard) que da acceso al dato.

use std::sync::Mutex;

fn main() {
    let m = Mutex::new(5);

    {
        let mut guardia = m.lock().expect("mutex envenenado");
        *guardia += 1;                         // el guardia se comporta como un &mut i32
    }                                          // ← el guardia se libera AQUÍ (Drop) y el lock se abre

    println!("{}", *m.lock().expect("mutex envenenado"));    // 6
}

Puntos clave:

• lock() bloquea el hilo hasta conseguir el cerrojo.
• Devuelve un Result (por el envenenamiento, ver 4.4).
• El cerrojo se libera automáticamente cuando el guardia sale de ámbito (ownership + Drop, Nivel 3).
• Un Mutex protege datos, no código: el dato vive dentro del Mutex, y es imposible acceder a él sin pedir el lock. (En C/C++, nada te obliga a ello.)

4.3 El patrón fundamental: Arc<Mutex<T>>

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let contador = Arc::new(Mutex::new(0));
    let mut manejadores = Vec::new();

    for _ in 0..10 {
        let contador = Arc::clone(&contador);
        manejadores.push(thread::spawn(move || {
            for _ in 0..1000 {
                let mut guardia = contador.lock().expect("mutex envenenado");
                *guardia += 1;
            }                                              // cada guardia se libera en cada vuelta
        }));
    }

    for h in manejadores {
        h.join().expect("el hilo no debe hacer panic");
    }

    println!("total = {}", *contador.lock().expect("mutex envenenado"));   // total = 10000
}

Siempre 10000. Sin el Mutex sería una carrera de datos; y Rust ni siquiera te deja compilarlo.

Qué pasa si te olvidas del Arc

use std::sync::Mutex;

fn main() {
    let contador = Mutex::new(0);

    for _ in 0..2 {
        // ❌ E0382: value moved into closure here, in previous iteration of loop
        // std::thread::spawn(move || {
        //     *contador.lock().unwrap() += 1;
        // });
    }

    println!("{:?}", contador.lock().expect("mutex envenenado"));
}

Con move, el primer hilo se lleva el Mutex; no queda nada para el segundo. Arc da un dueño por hilo.

Sacar el valor al final

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let total = Arc::new(Mutex::new(0));
    let t = Arc::clone(&total);
    thread::spawn(move || {
        *t.lock().expect("mutex envenenado") += 5;
    })
    .join()
    .expect("el hilo no debe hacer panic");

    // Cuando solo queda UN dueño, se puede desenvolver todo:
    let mutex = Arc::try_unwrap(total).expect("todavía hay otros dueños");
    let valor = mutex.into_inner().expect("mutex envenenado");
    println!("{valor}");                                      // 5
}

4.4 Envenenamiento (poisoning)

Si un hilo hace panic mientras tiene el lock, el dato podría haber quedado a medias. Por eso Rust marca el Mutex como envenenado, y los siguientes lock() devuelven Err:

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let m = Arc::new(Mutex::new(0));
    let m2 = Arc::clone(&m);

    let h = thread::spawn(move || {
        let _guardia = m2.lock().expect("mutex envenenado");
        panic!("fallo mientras se tenía el lock");
    });
    let _ = h.join();                                   // Err: el hilo hizo panic

    println!("envenenado: {}", m.is_poisoned());        // true

    match m.lock() {
        Ok(g) => println!("ok: {}", *g),
        Err(envenenado) => {
            let g = envenenado.into_inner();            // recuperar el guardia a pesar de todo
            println!("recuperado con valor {}", *g);    // recuperado con valor 0
        }
    }
}

Qué hacer en la práctica:

• Si no esperas que ocurra: .expect("mutex envenenado") (un panic aquí es razonable: el estado puede estar corrupto).
• Si el dato es simple y se puede recuperar: into_inner() como arriba.
• Desde Rust 1.77 existe clear_poison() para “limpiar” la marca.

4.5 Deadlocks: lo que Rust no previene

use std::sync::Mutex;

fn main() {
    let a = Mutex::new(1);
    let _g1 = a.lock().expect("mutex envenenado");

    // ⚠️ Pedir el MISMO lock otra vez en el mismo hilo: se queda esperando para siempre
    // (o hace panic; el comportamiento no está especificado).
    // let _g2 = a.lock().expect("mutex envenenado");
}

El deadlock clásico con dos locks:

Hilo 1: toma A ... y quiere B  ──┐
                                 ├── ninguno puede avanzar: esperan eternamente
Hilo 2: toma B ... y quiere A  ──┘

Reglas para evitar deadlocks:

1. Mantén las secciones críticas cortas: toma el lock, haz lo mínimo, suéltalo.
2. No hagas trabajo lento con el lock tomado (E/S, sleep, cálculos largos).
3. Orden global de locks: si necesitas dos, tómalos siempre en el mismo orden (por ejemplo por id).
4. No llames a código desconocido (callbacks) con un lock tomado.
5. Considera canales (módulo 5) en vez de varios locks.
6. Para soltar antes de tiempo: drop(guardia); o un bloque { ... } más pequeño.

use std::sync::Mutex;

fn main() {
    let m = Mutex::new(vec![1, 2, 3]);

    // ✅ Copia lo necesario y suelta el lock antes de trabajar
    let copia = {
        let guardia = m.lock().expect("mutex envenenado");
        guardia.clone()
    };                                                       // lock liberado aquí
    println!("{}", copia.iter().sum::<i32>());               // trabajo "largo" SIN lock
}

try_lock: intentar sin bloquear

use std::sync::Mutex;

fn main() {
    let m = Mutex::new(0);
    let _g = m.lock().expect("mutex envenenado");

    match m.try_lock() {
        Ok(_) => println!("lo conseguí"),
        Err(_) => println!("está ocupado, hago otra cosa"),      // este camino
    }
}

4.6 RwLock<T>: muchos lectores o un escritor

Es el principio aliasing XOR mutabilidad del Nivel 3, aplicado en ejecución:

• read() → muchos lectores a la vez.
• write() → un único escritor, sin lectores.

use std::sync::{Arc, RwLock};
use std::thread;

fn main() {
    let configuracion = Arc::new(RwLock::new(String::from("v1")));

    // Varios lectores a la vez
    let mut lectores = Vec::new();
    for i in 0..3 {
        let c = Arc::clone(&configuracion);
        lectores.push(thread::spawn(move || {
            let lectura = c.read().expect("rwlock envenenado");
            println!("lector {i} ve {}", *lectura);                  // lector N ve v1
        }));
    }
    for h in lectores {
        h.join().expect("el hilo no debe hacer panic");
    }

    // Un escritor
    {
        let mut escritura = configuracion.write().expect("rwlock envenenado");
        *escritura = String::from("v2");
    }

    println!("ahora: {}", *configuracion.read().expect("rwlock envenenado"));   // ahora: v2
}

Cuidados:

• ⚠️ No pidas write() teniendo ya un read() en el mismo hilo (deadlock). Suelta el read primero.
• Un RwLock es más costoso que un Mutex por operación: solo vale la pena con muchas lecturas y pocas escrituras.
• Admite envenenamiento igual que Mutex.

¿Mutex o RwLock?

4.7 Atómicos: contadores sin lock (mención)

Para contadores y banderas simples existen los tipos atómicos: operaciones indivisibles, sin Mutex.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;

fn main() {
    let contador = Arc::new(AtomicUsize::new(0));
    let mut manejadores = Vec::new();

    for _ in 0..8 {
        let c = Arc::clone(&contador);
        manejadores.push(thread::spawn(move || {
            for _ in 0..1000 {
                c.fetch_add(1, Ordering::SeqCst);
            }
        }));
    }
    for h in manejadores {
        h.join().expect("el hilo no debe hacer panic");
    }
    println!("{}", contador.load(Ordering::SeqCst));            // 8000
}

Ordering controla qué garantías de orden de memoria se dan. Para empezar, usa siempre SeqCst (la más estricta y simple de razonar).

4.8 Guía para elegir cómo compartir

4.9 Resumen del módulo

1. Arc<T> comparte propiedad entre hilos (solo lectura).
2. Mutex<T> / RwLock<T> aportan mutabilidad interior sincronizada.
3. Arc<Mutex<T>> es el patrón estándar para estado compartido modificable.
4. El guardia libera el lock al salir de ámbito: mantén ese ámbito pequeño.
5. Rust previene carreras de datos, pero no deadlocks.

Módulo 5: Canales para comunicar tareas (std::sync::mpsc)

“No te comuniques compartiendo memoria; comparte memoria comunicándote.” (lema popularizado por Go)

Un canal es una tubería: unos hilos envían mensajes y otros los reciben. El nombre mpsc significa multiple producer, single consumer: muchos productores, un solo consumidor.

5.1 Primer canal

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel();           // tx = transmisor (Sender), rx = receptor (Receiver)

    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(i).expect("el receptor sigue vivo");
            thread::sleep(Duration::from_millis(50));
        }
    });                                       // al terminar el hilo, `tx` se destruye

    for recibido in rx {                      // el bucle termina cuando TODOS los transmisores se destruyen
        println!("recibí {recibido}");
    }
    println!("canal cerrado");
}

Salida estable:

recibí 1
recibí 2
recibí 3
canal cerrado

5.2 Cómo funciona

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

fn main() {
    let (tx, rx) = mpsc::channel::<i32>();

    println!("{:?}", rx.try_recv());                              // Err(Empty)
    println!("{:?}", rx.recv_timeout(Duration::from_millis(50))); // Err(Timeout)

    tx.send(7).expect("receptor vivo");
    println!("{:?}", rx.recv());                                  // Ok(7)

    drop(tx);                                                    // se cierra el lado emisor
    println!("{:?}", rx.recv());                                  // Err(RecvError)
    println!("{:?}", rx.try_recv());                              // Err(Disconnected)

    // El otro lado: enviar cuando ya no hay receptor
    let (tx2, rx2) = mpsc::channel::<i32>();
    drop(rx2);
    match tx2.send(1) {
        Ok(()) => println!("enviado"),
        Err(e) => println!("falló: {e}"),                        // falló: sending on a closed channel
    }
}

5.3 send mueve el valor (ownership)

use std::sync::mpsc;

fn main() {
    let (tx, rx) = mpsc::channel();
    let texto = String::from("hola");

    tx.send(texto).expect("receptor vivo");
    // println!("{texto}");               // ❌ E0382: `texto` se movió al canal

    println!("{}", rx.recv().expect("emisor vivo"));
}

Es la forma más segura de pasar datos entre hilos: después de enviar, el emisor ya no puede tocarlos. No hay dos hilos con acceso al mismo dato: no hay carrera posible.

Requisito: el tipo enviado debe ser Send (Módulo 3). Enviar un Rc por un canal no compila.

5.4 Varios productores

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    for id in 0..3 {
        let tx = tx.clone();                              // un transmisor por hilo
        thread::spawn(move || {
            tx.send((id, id * 10)).expect("receptor vivo");
        });
    }
    drop(tx);                                             // ⚠️ IMPRESCINDIBLE: suelta el transmisor original

    let mut recibidos: Vec<(i32, i32)> = rx.iter().collect();
    recibidos.sort();                                     // el orden de llegada varía: ordenamos
    println!("{:?}", recibidos);                          // [(0, 0), (1, 10), (2, 20)]
}

⚠️ El error clásico: olvidar drop(tx)

Si conservas el tx original vivo, el canal nunca se cierra y for x in rx espera para siempre: queda un transmisor (el tuyo) que podría enviar más.

Regla: el canal se cierra cuando se destruyen todos los transmisores. Si repartes clones, destruye el original.

5.5 Mensajes tipados: enum como protocolo

Un enum permite definir qué órdenes entiende un hilo (patrón actor). Incluso puedes enviar dentro del mensaje un Sender para recibir la respuesta:

use std::sync::mpsc;
use std::thread;

enum Orden {
    Sumar(i32),
    Reportar(mpsc::Sender<i32>),       // el mensaje trae el canal de respuesta
    Terminar,
}

fn main() {
    let (tx, rx) = mpsc::channel::<Orden>();

    let actor = thread::spawn(move || {
        let mut total = 0;
        for orden in rx {
            match orden {
                Orden::Sumar(n) => total += n,
                Orden::Reportar(respuesta) => {
                    respuesta.send(total).expect("quien preguntó sigue esperando");
                }
                Orden::Terminar => break,
            }
        }
        total
    });

    tx.send(Orden::Sumar(5)).expect("actor vivo");
    tx.send(Orden::Sumar(7)).expect("actor vivo");

    let (tx_resp, rx_resp) = mpsc::channel();
    tx.send(Orden::Reportar(tx_resp)).expect("actor vivo");
    println!("total parcial: {}", rx_resp.recv().expect("el actor respondió"));   // 12

    tx.send(Orden::Terminar).expect("actor vivo");
    let final_ = actor.join().expect("el actor no debe hacer panic");
    println!("total final: {final_}");                                           // 12
}

Ventaja enorme: el estado (total) pertenece a un solo hilo: no necesita Mutex.

5.6 Canales acotados: sync_channel y backpressure

mpsc::channel() es no acotado: si el productor es más rápido que el consumidor, la cola crece sin límite (y puede agotar la memoria). mpsc::sync_channel(n) crea un canal con capacidad n: cuando está lleno, send bloquea hasta que haya espacio.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::sync_channel::<u32>(2);          // capacidad 2

    let productor = thread::spawn(move || {
        for i in 1..=5 {
            println!("[prod] enviando {i}");
            tx.send(i).expect("receptor vivo");           // se BLOQUEA cuando la cola está llena
            println!("[prod] enviado   {i}");
        }
    });

    thread::sleep(Duration::from_millis(300));            // consumidor "lento": tarda en empezar
    for v in rx {
        println!("            [cons] recibí {v}");
        thread::sleep(Duration::from_millis(100));
    }
    productor.join().expect("el productor no debe hacer panic");
}

Salida típica: el productor envía 1 y 2, se queda esperando en el 3 hasta que el consumidor empieza a vaciar la cola; a partir de ahí avanzan al ritmo del consumidor. Esto es backpressure (contrapresión): el consumidor lento frena al productor rápido. Lo retomamos a fondo en el Módulo 11.

sync_channel(0) es un encuentro (rendezvous): cada send espera a que alguien haga recv.

5.7 Varios consumidores: un pool de trabajadores

Receiver no es Clone (es “single consumer”). Para repartir trabajo entre varios hilos, se comparte el receptor con Arc<Mutex<_>>:

use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let (tx_trabajo, rx_trabajo) = mpsc::channel::<u32>();
    let rx_trabajo = Arc::new(Mutex::new(rx_trabajo));
    let (tx_resultado, rx_resultado) = mpsc::channel::<(u32, u32)>();

    let mut trabajadores = Vec::new();
    for _ in 0..3 {
        let rx = Arc::clone(&rx_trabajo);
        let tx_res = tx_resultado.clone();
        trabajadores.push(thread::spawn(move || loop {
            // El guardia temporal se libera al final de ESTA sentencia, antes de procesar:
            let mensaje = rx.lock().expect("mutex envenenado").recv();
            match mensaje {
                Ok(n) => tx_res.send((n, n * n)).expect("recolector vivo"),
                Err(_) => break,                          // el canal se cerró: no hay más trabajo
            }
        }));
    }
    drop(tx_resultado);                                   // soltamos nuestro transmisor de resultados

    for n in 1..=6 {
        tx_trabajo.send(n).expect("hay trabajadores vivos");
    }
    drop(tx_trabajo);                                     // señal de "no hay más trabajo"

    for t in trabajadores {
        t.join().expect("el trabajador no debe hacer panic");
    }

    let mut resultados: Vec<(u32, u32)> = rx_resultado.iter().collect();
    resultados.sort();
    for (n, cuadrado) in resultados {
        println!("{n} → {cuadrado}");
    }
}

Salida estable (porque ordenamos): 1 → 1, 2 → 4, 3 → 9, 4 → 16, 5 → 25, 6 → 36.

⚠️ Trampa famosa: while let y el guardia

// ❌ El guardia temporal vive durante TODO el cuerpo del `while let`:
// while let Ok(n) = rx.lock().expect("...").recv() {
//     procesar(n);          // ← los demás trabajadores esperan el lock mientras éste procesa
// }
//
// ✅ Toma el mensaje en una sentencia aparte (como arriba), de modo que el lock se libere antes de procesar.

5.8 Cierre ordenado: cerrar el canal

La forma más limpia de decirle a los trabajadores “se acabó” es cerrar el canal (destruir los Sender): los recv() devuelven Err y los hilos salen del bucle. Es más simple y robusto que mandar un mensaje especial Terminar a cada hilo.

5.9 ¿Canales o Mutex?

5.10 Resumen del módulo

1. mpsc::channel() → (Sender, Receiver): muchos productores, un consumidor.
2. send mueve el valor; el tipo debe ser Send.
3. El canal se cierra cuando se destruyen todos los Sender: ¡recuerda drop del original!
4. sync_channel(n) acota la cola y aplica backpressure.
5. Un enum de mensajes + un hilo “actor” = estado sin locks.

Módulo 6: async y .await

6.1 Por qué existe async

Un servidor web con 10 000 conexiones simultáneas casi siempre está esperando (a la red, al disco, a la base de datos). Con hilos, serían 10 000 hilos dormidos y mucha memoria. Con async, son 10 000 tareas pequeñas que un pequeño número de hilos va atendiendo solo cuando tienen algo que hacer.

async no hace las cosas más rápidas. Permite esperar muchas cosas a la vez sin gastar un hilo por espera.

6.2 async fn y .await

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

async fn saludar(nombre: &str) -> String {
    sleep(Duration::from_millis(100)).await;      // "espero 100 ms SIN bloquear el hilo"
    format!("hola, {nombre}")
}

#[tokio::main]
async fn main() {
    let futuro = saludar("Ana");                  // ⚠️ no ejecuta NADA todavía
    println!("el futuro está creado, pero aún no corrió");

    let mensaje = futuro.await;                   // aquí SÍ se ejecuta
    println!("{mensaje}");
}

(Para ejecutar este ejemplo necesitas Tokio: ver Módulo 8.1. Antes, en el Módulo 7, verás cómo funciona todo esto por dentro sin Tokio.)

Reglas básicas:

1. async fn no devuelve el valor directamente: devuelve un futuro (impl Future<Output = T>).
2. Un futuro es perezoso (lazy): no hace nada hasta que se le hace .await (o se le da a un runtime). Crear el futuro solo describe el trabajo.
3. .await significa: “suspende esta función hasta que el futuro termine; mientras tanto, el hilo puede atender otras tareas”.
4. .await solo se puede usar dentro de un contexto async.

async fn calcular() -> i32 {
    42
}

fn funcion_normal() {
    // let x = calcular().await;
    // ❌ E0728: `await` is only allowed inside `async` functions and blocks
}

// ❌ E0752: `main` function is not allowed to be `async`
// async fn main() { }
//    → main necesita un runtime: por eso se usa #[tokio::main]

fn main() {
    funcion_normal();
}

6.3 Olvidar .await: el error más común

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

async fn tarea() {
    sleep(Duration::from_millis(10)).await;
    println!("tarea ejecutada");
}

#[tokio::main]
async fn main() {
    tarea();             // ⚠️ warning: unused implementer of `Future` that must be used → NO se ejecuta
    tarea().await;       // ✅ ahora sí
}

6.4 Bloques async

También puedes crear un futuro anónimo con un bloque:

#[tokio::main]
async fn main() {
    let nombre = String::from("Ana");

    let futuro = async {
        1 + 1
    };

    let futuro_con_datos = async move {            // `move` se lleva `nombre` al futuro
        format!("hola, {nombre}")
    };

    println!("{}", futuro.await);                  // 2
    println!("{}", futuro_con_datos.await);        // hola, Ana
}

Igual que con los closures, async move es lo que usarás para tareas que viven más que la función que las crea.

6.5 Secuencial vs. concurrente

await una cosa tras otra es secuencial: las esperas se suman. Para esperar varias a la vez, usa join!:

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

async fn tarea(nombre: &str, ms: u64) -> String {
    sleep(Duration::from_millis(ms)).await;
    format!("{nombre} listo")
}

#[tokio::main]
async fn main() {
    // SECUENCIAL: 300 + 200 = ~500 ms
    let inicio = Instant::now();
    let a = tarea("A", 300).await;
    let b = tarea("B", 200).await;
    println!("{a}, {b} en ~{} ms", inicio.elapsed().as_millis());       // ~500 ms

    // CONCURRENTE: el máximo de los dos = ~300 ms
    let inicio = Instant::now();
    let (a, b) = tokio::join!(tarea("A", 300), tarea("B", 200));
    println!("{a}, {b} en ~{} ms", inicio.elapsed().as_millis());       // ~300 ms
}

Lección clave: .await no significa “en paralelo”. Dos .await seguidos son secuenciales. La concurrencia aparece al combinar futuros (join!, select!) o al crear tareas (spawn).

6.6 Lo que async NO es

6.7 El pecado capital: bloquear dentro de async

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

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let t1 = tokio::spawn(async {
        // ❌ BLOQUEA el único hilo del runtime: ninguna otra tarea puede avanzar mientras tanto
        std::thread::sleep(Duration::from_millis(500));
        println!("t1 terminó");
    });

    let t2 = tokio::spawn(async {
        // ✅ Cede el hilo mientras espera
        sleep(Duration::from_millis(100)).await;
        println!("t2 terminó");
    });

    t1.await.expect("t1 no debe hacer panic");
    t2.await.expect("t2 no debe hacer panic");
}

Con el std::thread::sleep, t2 no puede avanzar hasta que t1 termine, aunque solo necesitara 100 ms. Cambia el std::thread::sleep por tokio::time::sleep(...).await y t2 termina primero.

6.8 async y los tipos Send

Cuando un futuro se envía a otro hilo (con tokio::spawn), el futuro entero debe ser Send. Eso incluye todo lo que viva a través de un .await:

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

#[tokio::main]
async fn main() {
    let m = std::sync::Arc::new(Mutex::new(0));

    // ❌ error: future cannot be sent between threads safely
    //    (`MutexGuard<'_, i32>` no es Send y vive a través del .await)
    // tokio::spawn(async move {
    //     let mut g = m.lock().unwrap();
    //     *g += 1;
    //     sleep(Duration::from_millis(10)).await;      // ← el guardia sigue vivo aquí
    // });

    // ✅ Suelta el guardia ANTES del await
    let m2 = std::sync::Arc::clone(&m);
    tokio::spawn(async move {
        {
            let mut g = m2.lock().expect("mutex envenenado");
            *g += 1;
        }                                                  // guardia liberado
        sleep(Duration::from_millis(10)).await;
    })
    .await
    .expect("la tarea no debe hacer panic");

    println!("{}", *m.lock().expect("mutex envenenado"));  // 1
}

Los errores de “future cannot be sent between threads safely” casi siempre se resuelven acortando la vida de algo (Rc, un MutexGuard…) para que no cruce un .await. (Tokio ofrece además un Mutex async: Módulo 9.)

6.9 Resumen del módulo

1. async fn devuelve un futuro perezoso; .await lo ejecuta y cede el hilo mientras espera.
2. .await solo funciona en contextos async.
3. .await consecutivos son secuenciales; join!/spawn dan concurrencia.
4. Nunca bloquees dentro de async.
5. Lo que vive a través de un .await debe ser Send si la tarea se envía a otro hilo.

Módulo 7: Futures y runtime asíncrono

Este módulo abre la caja negra. No lo necesitas para el día a día, pero entenderlo elimina el 80 % de la confusión sobre async.

7.1 El trait Future

pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

pub enum Poll<T> {
    Ready(T),       // terminé, aquí está el resultado
    Pending,        // aún no puedo avanzar; te aviso cuando pueda
}

Un futuro es un objeto con una única operación, poll: “intenta avanzar; dime si ya terminaste o si todavía no”.

• Poll::Ready(valor) → el futuro terminó.
• Poll::Pending → todavía no; antes de devolverlo, el futuro debe haber dejado un aviso (el Waker) para que el runtime sepa cuándo volver a llamar a poll.
• Pin garantiza que el futuro no se mueve de lugar en memoria mientras se evalúa (necesario porque async puede crear referencias internas). Para este curso: “Pin = no me muevas”.

7.2 Un async fn es una máquina de estados

El compilador transforma cada async fn en una máquina de estados: cada .await es un punto donde puede pausarse y reanudarse.

async fn ejemplo() -> u32 {
    let a = paso_uno().await;      // estado 0 → espera paso_uno
    let b = paso_dos(a).await;     // estado 1 → espera paso_dos
    a + b                          // estado final
}

Por eso las tareas async son tan baratas: no hay una pila por tarea, solo los datos que sobreviven entre .await (el estado de la máquina). Y por eso no hay “magia”: es un objeto normal con un poll.

7.3 Escribir un futuro a mano

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct CuentaRegresiva {
    restante: u32,
}

impl Future for CuentaRegresiva {
    type Output = &'static str;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if self.restante == 0 {
            Poll::Ready("¡despegue!")
        } else {
            println!("  poll: faltan {}", self.restante);
            self.restante -= 1;
            cx.waker().wake_by_ref();       // "vuelve a llamarme pronto"
            Poll::Pending
        }
    }
}

7.4 Un runtime mínimo: block_on en 20 líneas

Un runtime (o executor) es, en esencia, un bucle que llama a poll. Aquí tienes uno funcional solo con la librería estándar:

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll, Wake, Waker};
use std::thread::{self, Thread};

// --- El futuro de la sección anterior ---
struct CuentaRegresiva {
    restante: u32,
}

impl Future for CuentaRegresiva {
    type Output = &'static str;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if self.restante == 0 {
            Poll::Ready("¡despegue!")
        } else {
            println!("  poll: faltan {}", self.restante);
            self.restante -= 1;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

// --- Un "despertador": cuando alguien lo activa, despierta al hilo que espera ---
struct HiloWaker(Thread);

impl Wake for HiloWaker {
    fn wake(self: Arc<Self>) {
        self.0.unpark();
    }
}

// --- El runtime más pequeño del mundo ---
fn block_on<F: Future>(futuro: F) -> F::Output {
    let mut futuro = std::pin::pin!(futuro);                       // fijar el futuro en la pila
    let waker = Waker::from(Arc::new(HiloWaker(thread::current())));
    let mut cx = Context::from_waker(&waker);

    loop {
        match futuro.as_mut().poll(&mut cx) {
            Poll::Ready(valor) => return valor,                    // terminó
            Poll::Pending => thread::park(),                       // dormir hasta que alguien llame a wake()
        }
    }
}

fn main() {
    let resultado = block_on(async {
        println!("empieza");
        let r = CuentaRegresiva { restante: 3 }.await;
        println!("termina");
        r
    });
    println!("{resultado}");
}

Salida estable:

empieza
  poll: faltan 3
  poll: faltan 2
  poll: faltan 1
termina
¡despegue!

Qué demuestra:

• Un bloque async es un futuro: lo único que hizo block_on fue llamarlo a poll repetidamente.
• CuentaRegresiva dijo Pending tres veces; cada vez avisó con el Waker (wake_by_ref) y el bucle volvió a llamar a poll.
• Un futuro real (por ejemplo, esperar un paquete de red) no llama a wake de inmediato: lo hace más tarde, cuando llega el dato. Entre medias, el hilo queda libre para hacer poll a otros futuros. Eso es la concurrencia asíncrona.

7.5 Lo que hace un runtime real

Tokio y los demás runtimes son este mismo bucle, mucho más sofisticado:

        ┌──────────────────────────────────────────────────────────────┐
        │                         RUNTIME                              │
        │                                                              │
        │   ┌────────────┐   poll()   ┌─────────────────────────┐      │
        │   │ Cola de    │──────────▶ │   Tarea (futuro)        │      │
        │   │ tareas     │            └─────────┬───────────────┘      │
        │   │ listas     │       Ready ◀────────┤ Pending              │
        │   └─────▲──────┘     (se descarta)    │ (registra un Waker)  │
        │         │ wake()                      ▼                      │
        │   ┌─────┴───────────────────────────────────────────┐       │
        │   │ Reactor: espera eventos del SO (red, disco),    │       │
        │   │ temporizadores...  y llama a wake() cuando       │       │
        │   │ algo está listo                                  │       │
        │   └──────────────────────────────────────────────────┘       │
        └──────────────────────────────────────────────────────────────┘

Componentes:

7.6 Rust no trae runtime

La librería estándar define Future, async/await, Waker… pero no incluye un runtime. Se elige uno como dependencia:

Ventaja: el lenguaje no te impone uno. Inconveniente: las bibliotecas async deben ser compatibles con tu runtime (en la práctica, casi todas lo son con Tokio).

7.7 Reglas que se derivan de cómo funciona

1. Cada .await es un posible punto de suspensión. Entre dos .await, tu código se ejecuta sin interrupciones: si tarda mucho, bloquea a las demás tareas del hilo.
2. Un futuro que nunca llega a un .await que ceda no deja avanzar a nadie.
3. Cancelar un futuro = dejar de llamarle a poll (descartarlo). Solo puede ocurrir en un .await (Módulo 10).
4. Lo que vive a través de un .await forma parte del estado del futuro: debe ser Send si la tarea se mueve entre hilos, y ocupa memoria.

7.8 Resumen del módulo

1. Future = “algo que se puede avanzar con poll” → Ready o Pending.
2. async fn = máquina de estados generada por el compilador.
3. Un runtime = ejecutor (bucle de poll) + reactor (eventos del SO) + temporizadores.
4. El Waker es el mecanismo de “avísame cuando puedas avanzar”.
5. Rust no trae runtime: se elige uno (Tokio).

Módulo 8: Tokio: tareas

8.1 Instalación

cargo new async_demo
cd async_demo
cargo add tokio --features full

o en Cargo.toml:

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

full activa todo (runtime, temporizadores, canales, red, archivos, señales…). En proyectos reales conviene activar solo lo necesario (rt-multi-thread, macros, time, sync, net…), para compilar más rápido.

8.2 #[tokio::main]

#[tokio::main]
async fn main() {
    println!("dentro del runtime de Tokio");
}

La macro transforma main en una función normal que construye un runtime y le pasa tu async fn con block_on. Es equivalente a:

fn main() {
    let runtime = tokio::runtime::Runtime::new().expect("no se pudo crear el runtime");
    runtime.block_on(async {
        println!("dentro del runtime de Tokio");
    });
}

Tipos de runtime

use tokio::runtime::Builder;

fn main() {
    let runtime = Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()                       // habilita temporizadores y E/S
        .build()
        .expect("no se pudo crear el runtime");

    runtime.block_on(async {
        println!("runtime personalizado");
    });
}

⚠️ Errores típicos de runtime (panics):

• “there is no reactor running, must be called from the context of a Tokio 1.x runtime” → usaste tokio::spawn/sleep fuera de un runtime.
• “Cannot start a runtime from within a runtime” → llamaste block_on dentro de código async.

8.3 tokio::spawn: crear tareas

Una tarea es un futuro que el runtime ejecuta por su cuenta, concurrentemente con las demás. Es el equivalente async de thread::spawn.

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

#[tokio::main]
async fn main() {
    let manejador = tokio::spawn(async {
        sleep(Duration::from_millis(100)).await;
        42                                           // valor de retorno de la tarea
    });

    println!("mientras tanto, main sigue trabajando...");

    let resultado = manejador.await.expect("la tarea no debe hacer panic");
    println!("la tarea devolvió {resultado}");       // 42
}

Detalles clave:

• tokio::spawn devuelve un JoinHandle<T>.
• handle.await devuelve Result<T, JoinError>: Err si la tarea hizo panic o fue cancelada.
• La tarea empieza a correr de inmediato (no hay que esperar al .await del handle).
• Soltar el JoinHandle no cancela la tarea: la desvincula (detach), sigue corriendo en segundo plano.
• Al terminar main, el runtime se cierra y las tareas pendientes se cancelan.

8.4 Los requisitos de spawn: Send + 'static

Igual que thread::spawn, la tarea puede ejecutarse en otro hilo y vivir más que quien la creó:

use std::rc::Rc;

#[tokio::main]
async fn main() {
    let v = vec![1, 2, 3];

    // ❌ E0373: async block may outlive the current function, but it borrows `v`
    // tokio::spawn(async {
    //     println!("{:?}", v);
    // });

    // ✅ `async move`: la tarea se lleva `v`
    tokio::spawn(async move {
        println!("{:?}", v);
    })
    .await
    .expect("la tarea no debe hacer panic");

    // ❌ error: future cannot be sent between threads safely (Rc no es Send)
    // let r = Rc::new(5);
    // tokio::spawn(async move {
    //     println!("{r}");
    // });
    let _ = Rc::new(5);
}

Los mismos conceptos del Módulo 3 (Send) y del Nivel 3 ('static / propiedad). Para compartir datos entre tareas: Arc, o mejor, canales.

8.5 Muchas tareas, muy poco costo

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

#[tokio::main]
async fn main() {
    let mut tareas = JoinSet::new();

    for i in 0..10_000u32 {
        tareas.spawn(async move {
            sleep(Duration::from_millis(100)).await;
            i
        });
    }

    let mut suma: u64 = 0;
    while let Some(resultado) = tareas.join_next().await {
        suma += resultado.expect("la tarea no debe hacer panic") as u64;
    }
    println!("{suma}");                              // 49995000
}

Diez mil tareas esperando 100 ms a la vez terminan en ~100 ms y casi sin memoria. Con 10 000 hilos del SO, esto sería muy pesado.

8.6 JoinSet: un grupo de tareas

JoinSet agrupa tareas y permite:

• spawn(futuro): añadir una tarea.
• join_next().await → Option<Result<T, JoinError>>: espera la siguiente que termine (en orden de finalización, no de creación). None cuando no quedan.
• abort_all(): cancelar todas.
• Si el JoinSet se destruye, todas sus tareas se cancelan automáticamente. Esto lo hace ideal para concurrencia estructurada: “las tareas hijas no sobreviven a su padre”.

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

#[tokio::main]
async fn main() {
    let mut tareas = JoinSet::new();
    tareas.spawn(async { sleep(Duration::from_millis(300)).await; "lenta" });
    tareas.spawn(async { sleep(Duration::from_millis(100)).await; "rápida" });
    tareas.spawn(async { sleep(Duration::from_millis(200)).await; "media" });

    while let Some(resultado) = tareas.join_next().await {
        println!("terminó: {}", resultado.expect("la tarea no debe hacer panic"));
    }
}

Salida estable (por los tiempos): terminó: rápida, terminó: media, terminó: lenta.

8.7 Errores y panic dentro de las tareas

El panic de una tarea no tumba al runtime ni a las demás: se convierte en Err(JoinError):

#[tokio::main]
async fn main() {
    let h = tokio::spawn(async {
        panic!("¡algo salió mal en la tarea!");
    });

    match h.await {
        Ok(()) => println!("la tarea terminó bien"),
        Err(e) if e.is_panic() => println!("la tarea hizo panic"),
        Err(e) if e.is_cancelled() => println!("la tarea fue cancelada"),
        Err(e) => println!("otro error: {e}"),
    }
}

Y si la tarea devuelve un Result, tienes dos niveles: handle.await (¿terminó la tarea?) y el Result interno (¿tuvo éxito la operación?):

#[tokio::main]
async fn main() {
    let h = tokio::spawn(async {
        let n: Result<i32, String> = Err(String::from("fallo de negocio"));
        n
    });

    match h.await {
        Ok(Ok(valor)) => println!("éxito: {valor}"),
        Ok(Err(e)) => println!("la tarea terminó, pero la operación falló: {e}"),
        Err(e) => println!("la tarea no pudo completarse: {e}"),
    }
}

8.8 Combinar futuros sin crear tareas: join! y try_join!

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

async fn paso(nombre: &str, ms: u64, falla: bool) -> Result<String, String> {
    sleep(Duration::from_millis(ms)).await;
    if falla {
        Err(format!("{nombre} falló"))
    } else {
        Ok(format!("{nombre} ok"))
    }
}

#[tokio::main]
async fn main() {
    // join!: espera a TODOS (aunque alguno falle)
    let (a, b) = tokio::join!(paso("A", 100, false), paso("B", 50, true));
    println!("{:?} / {:?}", a, b);       // Ok("A ok") / Err("B falló")

    // try_join!: si UNO devuelve Err, termina de inmediato y CANCELA a los demás
    match tokio::try_join!(paso("C", 300, false), paso("D", 50, true)) {
        Ok((c, d)) => println!("{c}, {d}"),
        Err(e) => println!("falló la combinación: {e}"),         // falló la combinación: D falló
    }
}

8.9 Trabajo bloqueante o pesado: spawn_blocking

Si debes ejecutar código que bloquea (cálculo pesado, una biblioteca sin async), no lo hagas en una tarea normal. Envíalo a la reserva de hilos bloqueantes:

fn calculo_pesado(n: u64) -> u64 {
    let mut suma = 0;
    for i in 0..n {
        suma += i % 7;
    }
    suma
}

#[tokio::main]
async fn main() {
    let h = tokio::task::spawn_blocking(|| calculo_pesado(50_000_000));
    println!("el runtime sigue libre mientras se calcula...");
    let resultado = h.await.expect("el cálculo no debe hacer panic");
    println!("{resultado}");
}

⚠️ Una tarea de spawn_blocking no se puede cancelar una vez que empezó.

8.10 Ceder el paso voluntariamente

Si una tarea hace un bucle largo sin ningún .await, monopoliza el hilo. tokio::task::yield_now().await cede el turno a las demás:

async fn bucle_largo() {
    for i in 0..1_000_000u32 {
        // ... trabajo ...
        if i % 10_000 == 0 {
            tokio::task::yield_now().await;      // "dejo pasar a otras tareas"
        }
    }
}

8.11 Resumen del módulo

1. #[tokio::main] crea el runtime y ejecuta tu main async.
2. tokio::spawn(async move { ... }) crea una tarea; devuelve JoinHandle; exige Send + 'static.
3. handle.await → Result<T, JoinError>; el panic de una tarea no afecta a las demás.
4. JoinSet agrupa tareas y las cancela al destruirse.
5. join!/try_join! combinan futuros sin crear tareas.
6. Lo bloqueante va a spawn_blocking.

Módulo 9: Tokio: canales y sincronización

Tokio ofrece versiones async de las herramientas del módulo 4 y 5: esperan con .await en lugar de bloquear el hilo.

9.1 mpsc: varios productores, un consumidor

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<String>(8);        // capacidad 8 (ACOTADO)

    let productor = tokio::spawn(async move {
        for i in 1..=3 {
            tx.send(format!("mensaje {i}")).await.expect("el receptor sigue vivo");
        }
    });                                                   // `tx` se destruye al terminar la tarea

    while let Some(mensaje) = rx.recv().await {           // None cuando se cierran todos los Sender
        println!("{mensaje}");
    }
    productor.await.expect("el productor no debe hacer panic");
}