Concurrencia segura en Rust con Mutex y channel

1. Descripción general

Rust llama a su capacidad “fearless concurrency” (concurrencia sin miedo). El compilador bloquea las carreras de datos en tiempo de compilación, antes de que el programa pueda ejecutarse. Este artículo presenta dos herramientas principales de la biblioteca estándar: Mutex (junto con Arc) para compartir estado, y mpsc channel para pasar mensajes entre threads. Después de leerlo, sabrás cómo usar cada uno, escribir pruebas para ellos y evitar trampas comunes como deadlock o mantener un lock demasiado tiempo.

Todos los ejemplos solo usan la biblioteca estándar, sin dependencias externas. Puedes crear un proyecto con cargo new orders-demo y pegar el código en src/main.rs.

2. Problema: compartir datos entre threads

Supongamos que tienes varios threads procesando órdenes y quieres contar el número total de órdenes. Una forma ingenua sería así:

use std::thread;

fn main() {
    let mut total_orders = 0;

    let handles: Vec<_> = (0..4)
        .map(|_| thread::spawn(|| total_orders += 1))
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

Este código no compila:

error[E0373]: closure may outlive the current function, but it borrows `total_orders`, which is owned by the current function

El compilador da dos razones:

• thread::spawn requiere que el closure sea 'static. El thread hijo puede vivir más que main, así que no puede tomar prestada una variable local.
• Incluso si agregamos move, cada thread recibirá su propia copia de total_orders. Nadie está realmente sumando a una variable compartida.

Para resolver esto, necesitas dos cosas: compartir la propiedad con Arc y sincronizar el acceso con Mutex.

3. Compartir estado con Arc<Mutex<T>>

3.1. Contar órdenes desde varios threads

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

fn main() {
    let total_orders = Arc::new(Mutex::new(0u32));
    let mut handles = Vec::new();

    for _ in 0..4 {
        let counter = Arc::clone(&total_orders);
        handles.push(thread::spawn(move || {
            for _ in 0..1_000 {
                *counter.lock().unwrap() += 1;
            }
        }));
    }

    for h in handles {
        h.join().unwrap();
    }

    println!("Total de órdenes: {}", *total_orders.lock().unwrap());
}

Total de órdenes: 4000

Explicación de cada parte:

• Arc::new(Mutex::new(0u32)): Arc (Atomic Reference Counted - Contador de referencias atómico) permite que varios threads tengan la propiedad de un valor. Mutex asegura que en cada momento solo un thread accede al valor interior.
• Arc::clone(&total_orders) solo incrementa el contador de referencias, no copia datos. Cada thread recibe su propio Arc mediante move.
• counter.lock() bloquea el thread actual hasta que obtiene el lock. La función devuelve Result<MutexGuard<u32>, PoisonError<…>>.
• MutexGuard se desreferencia a &mut u32, así que *... += 1 modifica el valor directamente. El guard se descarta al final de la sentencia, y el lock se libera automáticamente allí.

No puedes “olvidar desbloquear” en Rust. Tampoco puedes leer el valor pasando por alto el lock, porque los datos están dentro de Mutex.

Nota: No puedes reemplazar Arc con Rc. Rc no implementa Send, y el compilador reportará un error en cuanto intentes pasarlo a thread::spawn. De igual forma, no puedes reemplazar Mutex con RefCell, porque RefCell no es Sync.

3.2. Alcance del MutexGuard

El lock se mantiene hasta que el guard se descarta. Si asignas el guard a una variable, el lock vive hasta el final del alcance de esa variable:

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

fn reserve(inventory: &Mutex<HashMap<String, u32>>, sku: &str) -> bool {
    let reserved = {
        let mut stock = inventory.lock().unwrap();
        match stock.get_mut(sku) {
            Some(qty) if *qty > 0 => {
                *qty -= 1;
                true
            }
            _ => false,
        }
    }; // el guard se descarta aquí, el lock se libera

    if reserved {
        // Simulamos llamar a PaymentService lentamente, SIN mantener el lock
        thread::sleep(Duration::from_millis(50));
    }
    reserved
}

fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("AO-THUN".to_string(), 2)])));

    let handles: Vec<_> = (0..3)
        .map(|i| {
            let inv = Arc::clone(&inventory);
            thread::spawn(move || println!("Cliente {i}: {}", reserve(&inv, "AO-THUN")))
        })
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

Cliente 0: true
Cliente 1: true
Cliente 2: false

El orden de las líneas puede variar entre ejecuciones. El resultado siempre es exactamente dos true y un false.

El bloque { ... } limita el tiempo de retención del lock al mínimo. La parte lenta (llamada de pago) se ejecuta después de que el lock se haya liberado, así que otros threads no tienen que esperar. Si necesitas liberar más temprano, puedes llamar a drop(stock) explícitamente.

3.3. Mutex envenenado

Si un thread entra en pánico mientras mantiene un lock, Rust marca el Mutex como envenenado. La razón es que los datos podrían estar en un estado incompleto:

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

fn main() {
    let balance = Arc::new(Mutex::new(100i64));

    let b = Arc::clone(&balance);
    let _ = thread::spawn(move || {
        let mut guard = b.lock().unwrap();
        *guard -= 30;
        panic!("PaymentService falló a mitad de camino");
    })
    .join();

    match balance.lock() {
        Ok(v) => println!("Saldo: {}", *v),
        Err(poisoned) => {
            let v = poisoned.into_inner();
            println!("Mutex envenenado, saldo actual: {}", *v);
        }
    }
}

thread '<unnamed>' panicked at src/main.rs:10:9:
PaymentService falló a mitad de camino
Mutex envenenado, saldo actual: 70

Por lo tanto, .unwrap() después de lock() significa efectivamente “si otro thread entró en pánico mientras mantenía el lock, entonces entra en pánico también”. Para la mayoría de las aplicaciones, esa es una opción razonable. Si sabes cómo recuperar los datos, into_inner() devuelve el guard para que puedas inspeccionar y reparar.

4. Pasar mensajes con canales

Mutex responde la pregunta “varios threads editando datos compartidos”. Los canales van en la dirección opuesta: cada dato solo tiene un propietario en cada momento, y la propiedad se transfiere a través del canal. En muchos problemas, este enfoque es más simple y menos propenso a errores.

4.1. Múltiples productores, un consumidor

std::sync::mpsc es la abreviatura de multiple producer, single consumer (múltiples productores, un único consumidor):

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

#[derive(Debug)]
struct Order {
    id: u32,
    customer: String,
    amount: u64,
}

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

    for shop in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let order = Order {
                id: shop * 100,
                customer: format!("customer-{shop}"),
                amount: 150_000 * shop as u64,
            };
            tx.send(order).unwrap();
        });
    }
    drop(tx); // importante: libera el sender original

    let mut revenue = 0;
    for order in rx {
        println!("Orden recibida #{} de {}", order.id, order.customer);
        revenue += order.amount;
    }
    println!("Ingresos: {revenue}");
}

Orden recibida #100 de customer-1
Orden recibida #300 de customer-3
Orden recibida #200 de customer-2
Ingresos: 900000

Explicación:

• mpsc::channel() devuelve un par (Sender, Receiver). Cada productor recibe su propio tx.clone().
• tx.send(order) mueve order al canal. Después de esto, el thread emisor no puede usar más order, y el compilador reportará un error si lo intentas usar. Gracias a esto, dos threads nunca editan la misma Order.
• for order in rx repite hasta que todos los Sender se hayan descartado. Si falta drop(tx), el sender original aún vive en main, y el bucle esperará para siempre.

El orden de recepción depende del cronograma de los threads. El total de ingresos siempre es 900000.

4.2. Canales acotados y backpressure

mpsc::channel() tiene un buffer ilimitado. Si el productor es más rápido que el consumidor, la memoria se agrandará gradualmente. sync_channel(n) limita el buffer a n elementos, y send se bloqueará cuando el buffer esté lleno:

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

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

    let producer = thread::spawn(move || {
        for id in 1..=5 {
            tx.send(id).unwrap();
            println!("Orden enviada {id}");
        }
    });

    thread::sleep(Duration::from_millis(100));
    for id in rx {
        println!("Procesando orden {id}");
        thread::sleep(Duration::from_millis(20));
    }
    producer.join().unwrap();
}

Orden enviada 1
Orden enviada 2
Procesando orden 1
Orden enviada 3
Procesando orden 2
...

El productor envía dos órdenes y luego debe esperar a que el consumidor saque algunas. Este es un mecanismo de backpressure natural, muy útil para pipelines de procesamiento de datos. Con sync_channel(0), cada send debe esperar hasta que alguien haga recv (rendezvous).

5. Cuándo no necesitas Arc: threads con alcance

Si los threads seguramente terminan antes de que la función devuelva, thread::scope te permite tomar prestados los datos directamente, sin Arc:

use std::sync::Mutex;
use std::thread;

fn main() {
    let processed = Mutex::new(Vec::new());
    let orders = vec![101, 102, 103, 104];

    thread::scope(|s| {
        for chunk in orders.chunks(2) {
            s.spawn(|| processed.lock().unwrap().extend_from_slice(chunk));
        }
    }); // todos los threads en el alcance se han unido aquí

    let mut result = processed.into_inner().unwrap();
    result.sort();
    println!("{result:?}");
}

[101, 102, 103, 104]

Gracias a scope, el compilador sabe que los threads no viven más que processed y orders. Por lo tanto, tomar prestado con & es válido. Después de scope, into_inner() extrae los datos del Mutex sin necesidad de lock.

6. Escribir pruebas para código concurrente

Debes separar la lógica en funciones puras para hacerla fácil de probar. Pon lo siguiente en src/lib.rs:

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

pub fn count_orders(workers: u32, per_worker: u32) -> u32 {
    let total = Arc::new(Mutex::new(0));
    let handles: Vec<_> = (0..workers)
        .map(|_| {
            let total = Arc::clone(&total);
            thread::spawn(move || {
                for _ in 0..per_worker {
                    *total.lock().unwrap() += 1;
                }
            })
        })
        .collect();
    handles.into_iter().for_each(|h| h.join().unwrap());
    let result = *total.lock().unwrap();
    result
}

pub fn sum_payments(payments: Vec<u64>) -> u64 {
    let (tx, rx) = mpsc::channel();
    for amount in payments {
        let tx = tx.clone();
        thread::spawn(move || tx.send(amount).unwrap());
    }
    drop(tx);
    rx.iter().sum()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn counts_every_order_exactly_once() {
        assert_eq!(count_orders(8, 10_000), 80_000);
    }

    #[test]
    fn sums_all_payments_from_channel() {
        assert_eq!(sum_payments(vec![50_000, 120_000, 30_000]), 200_000);
    }

    #[test]
    fn empty_payments_returns_zero() {
        assert_eq!(sum_payments(vec![]), 0);
    }
}

cargo test

running 3 tests
test tests::empty_payments_returns_zero ... ok
test tests::sums_all_payments_from_channel ... ok
test tests::counts_every_order_exactly_once ... ok

test result: ok. 3 passed; 0 failed

La prueba empty_payments_returns_zero verifica un caso límite: si olvidas drop(tx), esta prueba se colgará indefinidamente en lugar de devolver 0. Esa es una señal clara para detectar el error.

Hay un pequeño punto en count_orders: el valor se asigna a la variable result antes de devolver. La razón es que el guard temporal en la expresión final de la función puede vivir más que total, y el compilador reportará un error de borrow.

7. Errores comunes

Deadlock por orden inverso de bloqueo. El thread A bloquea orders y luego customers, mientras que el thread B bloquea customers y luego orders. Ambos esperarán entre sí indefinidamente. Rust solo previene las carreras de datos, no previene el deadlock. Debes establecer un orden fijo de bloqueo, o combinar los dos datos en un Mutex<(A, B)> único.

El guard temporal vive más de lo que crees. Mira el siguiente código:

while let Some(order) = queue.lock().unwrap().pop() {
    process(order); // ¡el lock aún se está manteniendo aquí!
}

El temporary en la expresión del while let (y match) vive hasta el final del cuerpo del bucle. Cualquier otro thread que intente agregar órdenes a queue se bloqueará durante todo el tiempo que process se ejecute. La solución es separar: let next = queue.lock().unwrap().pop(); y luego procesar.

Llamar a lock() de nuevo en el mismo thread. std::sync::Mutex no es reentrante. Llamar a lock() dos veces en el mismo thread resultará en deadlock o panic.

Mantener el lock mientras esperas I/O o recv(). Cualquier operación bloqueante mientras mantienes el guard convierte a Mutex en un cuello de botella. Debes copiar o clonar los datos necesarios afuera, liberar el lock, y luego hacer el trabajo lento.

Olvidar descartar Sender. Como se vio arriba, el consumidor esperará indefinidamente.

Nota: En código async (Tokio), no mantengas std::sync::MutexGuard a través de .await. Usa tokio::sync::Mutex o los canales de Tokio. Además, para un simple entero como un contador, AtomicU32 es mucho más rápido que Mutex<u32>.

¿Cuándo elegir cuál?

• Usa Mutex cuando varios threads necesitan leer/modificar el mismo estado durante mucho tiempo, por ejemplo, caché o almacén. Si la lectura es mucho más frecuente que la escritura, considera RwLock.
• Usa canal cuando los datos fluyen de un lugar a otro, por ejemplo, colas de órdenes, pipelines o grupos de trabajadores. El canal ayuda a que cada valor tenga un solo propietario, así que la lógica es más fácil de razonar.

8. Conclusión

Rust convierte los errores de concurrencia comunes en errores de compilación. No puedes compartir datos sin Arc, no puedes modificar datos sin Mutex, y no puedes reutilizar un valor que ya se ha enviado a través de un canal. Arc<Mutex<T>> es adecuado para estado compartido, mientras que los canales mpsc son adecuados para transferir propiedad entre threads. sync_channel agrega backpressure, y thread::scope ayuda a prescindir de Arc cuando el ciclo de vida de los threads es claro. Lo que el compilador no puede proteger es el deadlock, guards que viven demasiado, y senders que no se descartan. Esos son los lugares donde necesitas revisar cuidadosamente y escribir pruebas con cuidado.