Concurrencia sin miedo

Los ordenadores actuales tienen varios núcleos, y para aprovecharlos un programa tiene que hacer varias cosas a la vez usando hilos (threads). El problema es que la programación concurrente es famosa por sus errores difíciles de reproducir: dos hilos que modifican el mismo dato a la vez, o que se esperan mutuamente para siempre.

Rust lo llama concurrencia sin miedo porque las mismas reglas de ownership y préstamos que ya conoces impiden, al compilar, la mayoría de esos errores. Si tu código concurrente compila, no tiene condiciones de carrera sobre los datos.

Crear hilos con thread::spawn

thread::spawn recibe una closure y la ejecuta en un hilo nuevo, en paralelo con el hilo principal:

let handle = thread::spawn(|| {
    for i in 1..=3 {
        println!("hilo secundario: {i}");
    }
});

for i in 1..=3 {
    println!("hilo principal: {i}");
}

handle.join().unwrap();

El orden de las líneas depende de cómo el sistema operativo reparta el tiempo entre los hilos, así que cambia de una ejecución a otra. Una salida posible:

hilo principal: 1
hilo secundario: 1
hilo secundario: 2
hilo principal: 2
hilo principal: 3
hilo secundario: 3

Esperar a un hilo con join

thread::spawn devuelve un JoinHandle. Al llamar a su método join, el hilo actual espera a que el otro termine. Es importante: cuando main acaba, el programa termina y se cierran todos los hilos, hayan acabado o no.

join devuelve además lo que devuelva la closure, envuelto en un Result (es Err si el hilo hizo panic):

use std::thread;

fn main() {
    let handle = thread::spawn(|| {
        let suma: u64 = (1..=1_000).sum();
        suma
    });

    let resultado = handle.join().unwrap();
    println!("La suma es {resultado}");
}

Closures move en hilos

Un hilo puede seguir ejecutándose cuando la función que lo creó ya ha terminado. Por eso Rust no le deja tomar prestadas variables locales:

use std::thread;

fn main() {
    let nombres = vec!["Ana", "Luis"];
    let handle = thread::spawn(|| {
        println!("{nombres:?}");
    });
    handle.join().unwrap();
}

El error (closure may outlive the current function) sugiere la solución: con move, el hilo se queda con la propiedad de lo que usa.

use std::thread;

fn main() {
    let nombres = vec!["Ana", "Luis"];
    let handle = thread::spawn(move || {
        println!("{nombres:?}");
    });
    handle.join().unwrap();
}

Es habitual lanzar varios hilos, guardar sus JoinHandle en un vector y recoger los resultados después. Al hacer join en orden, los resultados salen siempre en el mismo orden:

use std::thread;

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

    for id in 1..=4 {
        handles.push(thread::spawn(move || id * id));
    }

    for handle in handles {
        println!("{}", handle.join().unwrap());
    }
}

Canales: enviar mensajes entre hilos

Una forma segura de comunicar hilos es pasarse mensajes en lugar de compartir memoria. mpsc::channel crea un canal con dos extremos: el transmisor tx y el receptor rx. Su nombre viene de multiple producer, single consumer: muchos pueden enviar, uno recibe.

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

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

    thread::spawn(move || {
        for palabra in ["hola", "desde", "otro", "hilo"] {
            tx.send(palabra).unwrap();
        }
    });

    for recibido in rx {
        println!("Recibido: {recibido}");
    }
}

• send mueve el valor al canal: después ya no puedes usarlo en el hilo que lo envió.
• recv() espera hasta que llegue un mensaje. Devuelve Err cuando todos los transmisores se han cerrado.
• Recorrer rx con for recibe mensajes hasta que se cierran todos los transmisores.

Para tener varios productores, clona el transmisor con tx.clone(), uno por hilo:

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

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

    for id in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(format!("trabajador {id} listo")).unwrap();
        });
    }
    drop(tx); // el original no envía nada: lo cerramos para que el bucle termine

    let mut mensajes: Vec<String> = rx.iter().collect();
    mensajes.sort();
    for mensaje in &mensajes {
        println!("{mensaje}");
    }
}

Los mensajes llegan en el orden en que terminen los hilos, que no es fijo. Por eso aquí se ordenan antes de imprimirlos.

Cuidado: Si olvidas el drop(tx), el transmisor original sigue abierto y el bucle sobre rx espera para siempre un mensaje que nunca llegará.

Estado compartido: Arc y Mutex

La otra opción es que varios hilos compartan un dato. Para modificarlo sin que se pisen, se protege con un Mutex (mutual exclusion): solo un hilo a la vez puede acceder a su interior.

use std::sync::Mutex;

fn main() {
    let contador = Mutex::new(0);
    {
        let mut valor = contador.lock().unwrap();
        *valor += 1;
    } // aquí se libera el bloqueo
    println!("contador = {}", contador.lock().unwrap());
}

lock() espera hasta conseguir el acceso y devuelve un guardián que se comporta como &mut T. El bloqueo se libera solo cuando el guardián sale de su ámbito, gracias a Drop. Devuelve un Result porque falla si otro hilo hizo panic mientras tenía el bloqueo; de ahí el unwrap().

Para que varios hilos sean dueños del mismo Mutex hace falta Arc<T>, la versión de Rc segura entre hilos (atomic reference counted):

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

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

    for _ in 0..10 {
        let contador = Arc::clone(&contador);
        handles.push(thread::spawn(move || {
            let mut valor = contador.lock().unwrap();
            *valor += 1;
        }));
    }

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

    println!("Resultado: {}", contador.lock().unwrap());
}

Arc<Mutex<T>> es el equivalente entre hilos de Rc<RefCell<T>>.

¿Por qué no Rc? Send y Sync

Si intentas enviar un Rc a otro hilo, el código no compila:

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

fn main() {
    let datos = Rc::new(5);
    let copia = Rc::clone(&datos);
    let handle = thread::spawn(move || {
        println!("{copia}");
    });
    handle.join().unwrap();
}

El contador de Rc no está protegido: si dos hilos lo actualizaran a la vez, podría corromperse. Rust lo sabe gracias a dos traits que se implementan automáticamente:

• Send: el valor se puede mover a otro hilo. Casi todos los tipos lo son; Rc no.
• Sync: el valor se puede compartir por referencia entre hilos. Mutex lo es; RefCell no.

Hilos con ámbito: thread::scope

thread::scope crea hilos que tienen garantizado terminar antes de que acabe el bloque. Por eso sí pueden tomar prestadas variables locales, sin move ni Arc:

use std::thread;

fn main() {
    let numeros = vec![1, 2, 3, 4, 5, 6];
    let (izquierda, derecha) = numeros.split_at(3);

    let (a, b) = thread::scope(|s| {
        let h1 = s.spawn(|| izquierda.iter().sum::<i32>());
        let h2 = s.spawn(|| derecha.iter().sum::<i32>());
        (h1.join().unwrap(), h2.join().unwrap())
    });

    println!("{a} + {b} = {}", a + b);
}

Hola, Ferris, desde otro hilo