Hilos en C++: jthread, mutex, atómicos, async y ThreadSanitizer

Hilos en C++ moderno: repartir un cálculo con std::jthread, ver una carrera de datos y cazarla con ThreadSanitizer, arreglarla con un std::mutex o un atómico, comunicar hilos con una cola y condition_variable, lanzar tareas con std::async y parar un hilo de forma ordenada con stop_token.

También te puede interesar

Repartir un cálculo entre hilos con std::jthread

Un hilo es una función que se ejecuta a la vez que el resto del programa. En C++20 la forma recomendada de crearlos es std::jthread: igual que std::thread, pero al destruirse espera a que el hilo termine (hace join) en lugar de abortar el programa. Este ejemplo suma un millón de números repartiendo el trabajo en cuatro hilos; cada uno escribe su resultado en su propia casilla, así que no comparten nada que se modifique:

#include <array>
#include <numeric>
#include <print>
#include <thread>
#include <vector>

int main()
{
    std::vector<long long> datos(1'000'000);
    std::iota(datos.begin(), datos.end(), 1);              // 1, 2, 3... 1,000,000

    constexpr int N = 4;
    std::array<long long, N> parcial{};
    {
        std::vector<std::jthread> hilos;
        const std::size_t trozo = datos.size() / N;
        for (int h = 0; h < N; ++h) {
            auto ini = datos.begin() + h * trozo;
            auto fin = (h == N - 1) ? datos.end() : ini + trozo;
            hilos.emplace_back([ini, fin, &parcial, h] { parcial[h] = std::accumulate(ini, fin, 0LL); });
        }
    }   // aquí los cuatro jthread hacen join automáticamente

    for (int h = 0; h < N; ++h) std::println("hilo {}: {}", h, parcial[h]);
    std::println("total: {}", std::accumulate(parcial.begin(), parcial.end(), 0LL));
}
salidahilo 0: 31250125000
hilo 1: 93750125000
hilo 2: 156250125000
hilo 3: 218750125000
total: 500000500000

La llave que cierra el bloque es la clave: ahí se destruye el vector de hilos, cada std::jthread espera a que el suyo acabe y solo entonces se leen los resultados. En Linux con GCC conviene compilar con -pthread.

El error típico: una carrera de datos

Si varios hilos modifican la misma variable sin protección, el resultado es una carrera de datos, que en C++ es comportamiento indefinido. ++contador parece una sola operación, pero son tres: leer el valor, sumarle uno y guardarlo. Si dos hilos leen a la vez el mismo valor, una de las dos sumas se pierde:

#include <print>
#include <thread>
#include <vector>

int contador = 0;                        // compartido y sin protección

void sumar(int veces)
{
    for (int i = 0; i < veces; ++i) ++contador;
}

int main()
{
    {
        std::vector<std::jthread> hilos;
        for (int h = 0; h < 4; ++h) hilos.emplace_back(sumar, 100'000);
    }
    std::println("esperado: 400000   obtenido: {}", contador);
}
salida (cambia en cada ejecución)esperado: 400000   obtenido: 173131

En nuestra ejecución salieron 173,131 en lugar de 400,000: se perdieron más de la mitad de las sumas. Cada ejecución da un número distinto, así que si lo pruebas tú saldrá otro. Lo peor de estos errores es que pueden no aparecer en las pruebas y sí en producción. Para encontrarlos está ThreadSanitizer, incluido en GCC y Clang, que vigila los accesos a memoria mientras el programa se ejecuta:

#include <print>
#include <thread>
#include <vector>

int contador = 0;

void sumar(int veces)
{
    for (int i = 0; i < veces; ++i) ++contador;
}

int main()
{
    {
        std::vector<std::jthread> hilos;
        for (int h = 0; h < 4; ++h) hilos.emplace_back(sumar, 100'000);
    }
    std::println("obtenido: {}", contador);
}
g++ -std=c++23 -g -fsanitize=thread -O1 carrera.cpp -o carrera && ./carrera
salida con errorWARNING: ThreadSanitizer: data race (pid=95)
  Read of size 4 at 0x0000004232a4 by thread T2:
    #0 sumar(int) /curso/carrera_tsan.cpp:9 (carrera_tsan+0x4025ea)
…
  Previous write of size 4 at 0x0000004232a4 by thread T1:
    #0 sumar(int) /curso/carrera_tsan.cpp:9 (carrera_tsan+0x402612)
…
  Location is global 'contador' of size 4 at 0x0000004232a4 (carrera_tsan+0x4232a4)
…
SUMMARY: ThreadSanitizer: data race /curso/carrera_tsan.cpp:9 in sumar(int)
…
WARNING: ThreadSanitizer: data race (pid=95)
  Write of size 4 at 0x0000004232a4 by thread T3:
    #0 sumar(int) /curso/carrera_tsan.cpp:9 (carrera_tsan+0x402612)
…
  Previous write of size 4 at 0x0000004232a4 by thread T2:
    #0 sumar(int) /curso/carrera_tsan.cpp:9 (carrera_tsan+0x402612)
…
  Location is global 'contador' of size 4 at 0x0000004232a4 (carrera_tsan+0x4232a4)
…
SUMMARY: ThreadSanitizer: data race /curso/carrera_tsan.cpp:9 in sumar(int)
…
ThreadSanitizer: reported 2 warnings

El informe señala la línea exacta (++contador, la 9 del archivo), qué dos hilos chocaron en ella y que la variable es la global contador. ThreadSanitizer hace el programa entre 5 y 15 veces más lento, así que se usa en las pruebas, no en producción. No se puede combinar con AddressSanitizer en la misma compilación.

Arreglarlo con un mutex o con un atómico

Hay dos soluciones. Un mutex deja pasar a un solo hilo a la vez por la parte protegida; std::scoped_lock lo bloquea al crearse y lo libera al salir del bloque, aunque haya una excepción. Para un contador simple es más eficiente un atómico, una variable cuyas operaciones el procesador hace de una sola vez:

#include <atomic>
#include <mutex>
#include <print>
#include <thread>
#include <vector>

int con_mutex = 0;
std::mutex m;
std::atomic<int> atomico{0};

void sumar(int veces)
{
    for (int i = 0; i < veces; ++i) {
        {
            std::scoped_lock bloqueo(m);   // solo un hilo a la vez dentro de estas llaves
            ++con_mutex;
        }
        atomico.fetch_add(1, std::memory_order_relaxed);
    }
}

int main()
{
    {
        std::vector<std::jthread> hilos;
        for (int h = 0; h < 4; ++h) hilos.emplace_back(sumar, 100'000);
    }
    std::println("con mutex:   {}", con_mutex);
    std::println("con atómico: {}", atomico.load());
}
salidacon mutex:   400000
con atómico: 400000

Este programa está compilado también con ThreadSanitizer, y no informa de nada: ya no hay carrera. memory_order_relaxed basta para un contador, en el que solo importa el total; si el atómico se usa para avisar a otro hilo de que unos datos están listos, deja el orden por omisión.

Esperar un aviso: una cola con condition_variable

Muchas veces un hilo produce trabajo y otro lo consume. El consumidor no debe dar vueltas comprobando la cola sin parar, sino dormir hasta que haya algo. Para eso está std::condition_variable:

#include <condition_variable>
#include <mutex>
#include <print>
#include <queue>
#include <string>
#include <thread>

std::queue<std::string> cola;
std::mutex m;
std::condition_variable hay_trabajo;
bool terminado = false;

void consumidor()
{
    while (true) {
        std::unique_lock bloqueo(m);
        hay_trabajo.wait(bloqueo, [] { return !cola.empty() || terminado; });   // duerme hasta que haya algo
        if (cola.empty()) return;                                               // terminado y sin trabajo
        std::string tarea = std::move(cola.front());
        cola.pop();
        bloqueo.unlock();                                // procesar sin tener el mutex
        std::println("procesado: {}", tarea);
    }
}

int main()
{
    std::jthread trabajador(consumidor);
    for (std::string t : {"factura-001", "factura-002", "factura-003", "informe-octubre"}) {
        {
            std::scoped_lock bloqueo(m);
            cola.push(t);
        }
        hay_trabajo.notify_one();
    }
    {
        std::scoped_lock bloqueo(m);
        terminado = true;
    }
    hay_trabajo.notify_one();
}   // el jthread espera aquí a que el consumidor vacíe la cola
salidaprocesado: factura-001
procesado: factura-002
procesado: factura-003
procesado: informe-octubre

La condición que se pasa a wait es imprescindible: el hilo puede despertarse sin que nadie lo avise (un «despertar espurio»), y además el aviso puede llegar antes de que el consumidor empiece a esperar. Con la condición, wait comprueba el estado real antes de dormirse y al despertar.

Tareas con resultado: std::async y std::future

Si solo quieres lanzar un cálculo y recoger su resultado más tarde, std::async es más sencillo que manejar hilos: devuelve un std::future, y su get() espera al resultado. Si la tarea lanzó una excepción, get() la vuelve a lanzar en el hilo que llama:

#include <future>
#include <print>
#include <stdexcept>
#include <vector>

int primos_hasta(int n)
{
    if (n < 0) throw std::invalid_argument("n no puede ser negativo");
    int cuenta = 0;
    for (int i = 2; i <= n; ++i) {
        bool primo = true;
        for (int d = 2; d * d <= i; ++d)
            if (i % d == 0) { primo = false; break; }
        cuenta += primo;
    }
    return cuenta;
}

int main()
{
    const std::vector<int> limites{10'000, 100'000, 1'000'000};
    std::vector<std::future<int>> tareas;
    for (int n : limites)
        tareas.push_back(std::async(std::launch::async, primos_hasta, n));   // empiezan ya, en paralelo

    for (std::size_t i = 0; i < limites.size(); ++i)
        std::println("primos hasta {:>9}: {}", limites[i], tareas[i].get());   // get() espera a cada una

    auto mala = std::async(std::launch::async, primos_hasta, -5);
    try {
        mala.get();
    } catch (const std::invalid_argument& e) {
        std::println("la tarea falló: {}", e.what());
    }
}
salidaprimos hasta     10000: 1229
primos hasta    100000: 9592
primos hasta   1000000: 78498
la tarea falló: n no puede ser negativo

Pasa siempre std::launch::async: sin él, la implementación puede decidir no crear ningún hilo y ejecutar la tarea más tarde, dentro de get(), y entonces no hay paralelismo. Otro detalle: el std::future que devuelve std::async espera a que la tarea termine al destruirse, así que std::async(...); sin guardar el resultado no lanza nada en segundo plano, sino que espera ahí mismo.

Parar un hilo de forma ordenada: stop_token

C++ no permite matar un hilo desde fuera, y es mejor así: lo dejaría a medias, quizá con un mutex bloqueado. Lo correcto es pedirle que pare. std::jthread trae ese mecanismo: si la función del hilo recibe un std::stop_token, puede preguntar si alguien le ha pedido parar, y el destructor del jthread lo pide automáticamente antes de esperar:

#include <chrono>
#include <print>
#include <stop_token>
#include <thread>

using namespace std::chrono_literals;

void vigilar(std::stop_token parar, int intervalo_ms)
{
    int revisiones = 0;
    while (!parar.stop_requested()) {
        ++revisiones;                                     // aquí iría el trabajo de verdad
        std::this_thread::sleep_for(std::chrono::milliseconds(intervalo_ms));
    }
    std::println("vigilante: me pidieron parar; cierro ordenadamente ({})",
                 revisiones > 0 ? "hice al menos una revisión" : "no llegué a revisar");
}

int main()
{
    std::jthread vigilante(vigilar, 10);                  // el stop_token lo pone jthread
    std::this_thread::sleep_for(50ms);
    std::println("main: pido al vigilante que pare");
    vigilante.request_stop();
    vigilante.join();
    std::println("main: el vigilante ya terminó");
}
salidamain: pido al vigilante que pare
vigilante: me pidieron parar; cierro ordenadamente (hice al menos una revisión)
main: el vigilante ya terminó

Si un hilo pasa mucho tiempo dormido esperando algo, conviene que la espera también se despierte al pedir que pare: std::condition_variable_any tiene versiones de wait que reciben el stop_token.

Otro error típico: un std::thread sin join

Con el std::thread de C++11, olvidar el join() (o el detach()) no es un descuido menor: si el objeto se destruye mientras el hilo sigue unido a él, el programa entero se aborta.

#include <print>
#include <thread>

void tarea()
{
    std::println("trabajando...");
}

int main()
{
    std::thread t(tarea);
    // MAL: falta t.join(). Al salir de main se destruye t y C++ llama a std::terminate
}
salida con errorterminate called without an active exception

Por eso en código nuevo es mejor std::jthread, que hace el join solo. detach() tampoco suele ser buena idea: el hilo queda suelto, nadie espera a que termine y, si sigue usando variables del programa cuando este acaba, el resultado es indefinido.

  • ParaUsa
  • Lanzar un hilostd::jthread (C++20); std::thread con join() si tu compilador es antiguo
  • Proteger datos compartidosstd::mutex con std::scoped_lock
  • Un contador o un indicador compartidostd::atomic
  • Esperar a que haya trabajostd::condition_variable con una condición
  • Un cálculo con resultadostd::async(std::launch::async, ...) y std::future
  • Pedir a un hilo que parestd::stop_token y request_stop()
  • Encontrar carreras de datos-fsanitize=thread (ThreadSanitizer)

Todo el código de esta página se compiló con GCC 15.3 (C++23, -std=c++23 -Wall -Wextra, sin avisos) y se ejecutó en Linux, en un contenedor limpio sin acceso a la red, antes de publicarla; las salidas están copiadas de esa ejecución.

decodigo.com

Código listo para usar: ejemplos en Java, Python, Go, Rust y JavaScript, cursos, guías, herramientas en línea y noticias de tecnología.