La necesidad de la programación concurrente

Los procesadores multinúcleo son el estándar actual, y el software que no aprovecha múltiples hilos de ejecución deja capacidad de cómputo sin utilizar. A partir de C++11, la biblioteca estándar ofrece herramientas nativas para crear y coordinar hilos, lo que permite exprimir el hardware sin depender de bibliotecas externas. Este artículo se centra en el patrón productor-consumidor, uno de los problemas clásicos de sincronización, y muestra cómo implementarlo de forma segura y eficiente con las primitivas de C++11.

Componentes fundamentales de concurrencia en C++11

La cabecera <thread> permite lanzar y gestionar hilos. <mutex> proporciona cerrojos y envoltorios RAII como std::lock_guard y std::unique_lock. <condition_variable> permite que un hilo espere hasta que se cumpla una condición. <atomic> ofrece operaciones indivisibles sobre tipos compartidos, y <future> facilita la ejecución asíncrona con std::async y std::future.

Un hilo mínimo se crea así:

#include <iostream>
#include <thread>

void tarea() {
    std::cout << "Hilo " << std::this_thread::get_id() << '\n';
}

int main() {
    std::thread h(tarea);
    h.join();
}

Compilar con g++ -std=c++11 -pthread programa.cpp -o programa.

El modelo productor-consumidor

Un productor genera datos y los deposita en un búfer compartido; un consumidor los retira. El desafío es coordinar el acceso sin corromper el búfer y sin que los hilos esperen indefinidamente. Veremos cuatro variantes con complejidad creciente.

1. Un productor, un consumidor

Usamos un búfer circular con un cerrojo y dos variables de condición. La clave es esperar mientras el búfer esté lleno (productor) o vacío (consumidor).

#include <mutex>
#include <condition_variable>

constexpr size_t TAM = 8;

struct BufCircular {
    int datos[TAM];
    size_t idx_lec = 0, idx_esc = 0;
    size_t ocupados = 0;
    std::mutex mtx;
    std::condition_variable no_lleno, no_vacio;
};

void producir(BufCircular& b, int valor) {
    std::unique_lock<std::mutex> cerrojo(b.mtx);
    b.no_lleno.wait(cerrojo, [&b] { return b.ocupados < TAM; });
    b.datos[b.idx_esc] = valor;
    b.idx_esc = (b.idx_esc + 1) % TAM;
    ++b.ocupados;
    b.no_vacio.notify_one();
}

int consumir(BufCircular& b) {
    std::unique_lock<std::mutex> cerrojo(b.mtx);
    b.no_vacio.wait(cerrojo, [&b] { return b.ocupados > 0; });
    int valor = b.datos[b.idx_lec];
    b.idx_lec = (b.idx_lec + 1) % TAM;
    --b.ocupados;
    b.no_lleno.notify_one();
    return valor;
}

El uso de lambdas en wait protege contra despertares espurios y expresa la condición de forma explícita.

2. Un productor, múltiples consumidores

Cuando varios consumidores compiten, hay que coordinar cuántos elementos procesa cada uno y cuándo deben terminar. Una solución es mantener un contador atómico de elementos consumidos y cerrar el búfer cuando se alcanza el total.

#include <atomic>

std::atomic<size_t> consumidos{0};
constexpr size_t TOTAL = 100;

void consumidor(BufCircular& buf, int id) {
    while (consumidos.load() < TOTAL) {
        int dato = consumir(buf);
        ++consumidos;
        std::cout << "Consumidor " << id << " procesa " << dato << '\n';
    }
}

El productor genera exactamente TOTAL elementos y luego señala el cierre del búfer para que los consumidores salgan de la espera. Esto exige modificar BufCircular añadiendo un método cerrar() que notifique a todas las condiciones.

3. Múltiples productores, un consumidor

Varios productores pueden usar un contador atómico para repartirse los elementos que deben generar. Cada productor obtiene un valor único del contador y lo inserta en el búfer. El consumidor simplemente extrae mientras no se haya cerrado el búfer.

std::atomic<size_t> producidos{0};

void productor(BufCircular& buf, int id) {
    while (true) {
        size_t mi_valor = producidos.fetch_add(1);
        if (mi_valor >= TOTAL) break;
        producir(buf, static_cast<int>(mi_valor));
        std::cout << "Productor " << id << " genera " << mi_valor << '\n';
    }
}

De nuevo, al terminar todos los productores se cierra el búfer para que el consumidor deje de esperar.

4. Múltiples productores y múltiples consumidores

Combinando las técnicas anteriores, un búfer con cierre controlado y contadores atómicos permite que varios productores y varios consumidores trabajen simultáneamante. El búfer se cierra cuando los productores han generado el total, y los consumidores salen al encontrar el búfer cerrado y vacío.

class BufCircularCierre {
    std::vector<int> datos;
    size_t idx_lec = 0, idx_esc = 0, ocupados = 0;
    bool cerrado = false;
    std::mutex mtx;
    std::condition_variable no_lleno, no_vacio;

public:
    explicit BufCircularCierre(size_t cap) : datos(cap) {}

    void producir(int valor) {
        std::unique_lock<std::mutex> lk(mtx);
        no_lleno.wait(lk, [this] { return ocupados < datos.size() || cerrado; });
        if (cerrado) return;
        datos[idx_esc] = valor;
        idx_esc = (idx_esc + 1) % datos.size();
        ++ocupados;
        no_vacio.notify_one();
    }

    std::optional<int> consumir() {
        std::unique_lock<std::mutex> lk(mtx);
        no_vacio.wait(lk, [this] { return ocupados > 0 || cerrado; });
        if (cerrado && ocupados == 0) return std::nullopt;
        int valor = datos[idx_lec];
        idx_lec = (idx_lec + 1) % datos.size();
        --ocupados;
        no_lleno.notify_one();
        return valor;
    }

    void cerrar() {
        std::lock_guard<std::mutex> lk(mtx);
        cerrado = true;
        no_lleno.notify_all();
        no_vacio.notify_all();
    }
};

Los productores usan un contador atómico para distribuirse el trabajo; los consumidores extraen mientras el búfer no esté cerrado y vacío. Al finalizar, el hilo principal cierra el búfer y espera a los consumidores.

Prácticas recomendadas

Evitar interbloqueos

Cuando se adquieren varios cerrojos, se debe usar std::lock para evitar dependencias cíclicas.

void transferir(Cuenta& origen, Cuenta& destino, double monto) {
    std::unique_lock<std::mutex> lk1(origen.mtx, std::defer_lock);
    std::unique_lock<std::mutex> lk2(destino.mtx, std::defer_lock);
    std::lock(lk1, lk2);
    origen.saldo -= monto;
    destino.saldo += monto;
}

RAII para gestión de recursos

Envolver el búfer en una clase que encapsule el cerrojo y las condiciones garantiza que los bloqueos se liberen automáticamente, incluso ante excepciones. El método consumir que devuelve std::optional es un ejemplo de interfaz segura.

Reducir la contención

Dividir un recurso en particiones independientes reduce la disputa por un único cerrojo. Por ejemplo, un contador concurrente puede usar varios mutexes, cada uno protegiendo una región distinta.

class ContadorParticionado {
    std::vector<std::mutex> mutexes;
    std::vector<int> contadores;
public:
    ContadorParticionado(size_t particiones) : mutexes(particiones), contadores(particiones) {}

    void incrementar(size_t indice) {
        std::lock_guard<std::mutex> lk(mutexes[indice]);
        ++contadores[indice];
    }

    int total() const {
        int suma = 0;
        for (size_t i = 0; i < contadores.size(); ++i) {
            std::lock_guard<std::mutex> lk(mutexes[i]);
            suma += contadores[i];
        }
        return suma;
    }
};

Errores frecuentes y cómo evitarlos

Despertares espurios

Las variables de condición pueden despertar sin una notificación real. Por eso siempre se debe usar una condición de guarda en wait, como se ha hecho en todos los ejemplos con lambdas.

Inversión de prioridad

Ocurre cuando un hilo de alta prioridad espera un recurso bloqueado por uno de baja prioridad. En sistemas que soportan herencia de prioridad se puede mitigar; en C++ estándar no hay solución directa, pero se puede reducir el tiempo que los hilos de baja prioridad mantienen los cerrojos.

Fugas de recursos

Si un hilo termina abruptamente, los recursos deben liberarse. Las clases con destructores que limpian correctamente y el uso de std::thread con join() en todos los caminos ayudan a evitar fugas.

Implementación completa multiproductor‑multiconsumidor

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <vector>
#include <atomic>
#include <optional>
#include <chrono>

constexpr size_t CAPACIDAD = 8;
constexpr int ITEMS_TOTALES = 50;

class BufCircular {
    std::vector<int> datos;
    size_t idx_lec = 0, idx_esc = 0, ocupados = 0;
    bool cerrado = false;
    std::mutex mtx;
    std::condition_variable no_lleno, no_vacio;
public:
    explicit BufCircular(size_t cap) : datos(cap) {}

    void producir(int valor) {
        std::unique_lock<std::mutex> lk(mtx);
        no_lleno.wait(lk, [this] { return ocupados < datos.size() || cerrado; });
        if (cerrado) return;
        datos[idx_esc] = valor;
        idx_esc = (idx_esc + 1) % datos.size();
        ++ocupados;
        no_vacio.notify_one();
    }

    std::optional<int> consumir() {
        std::unique_lock<std::mutex> lk(mtx);
        no_vacio.wait(lk, [this] { return ocupados > 0 || cerrado; });
        if (cerrado && ocupados == 0) return std::nullopt;
        int valor = datos[idx_lec];
        idx_lec = (idx_lec + 1) % datos.size();
        --ocupados;
        no_lleno.notify_one();
        return valor;
    }

    void cerrar() {
        std::lock_guard<std::mutex> lk(mtx);
        cerrado = true;
        no_lleno.notify_all();
        no_vacio.notify_all();
    }
};

std::atomic<int> producidos{0}, consumidos{0};

void productor(BufCircular& buf, int id) {
    while (true) {
        int mi_valor = producidos.fetch_add(1);
        if (mi_valor >= ITEMS_TOTALES) break;
        buf.producir(mi_valor);
        std::cout << "P" << id << " generó " << mi_valor << '\n';
        std::this_thread::sleep_for(std::chrono::milliseconds(20));
    }
}

void consumidor(BufCircular& buf, int id) {
    while (true) {
        auto item = buf.consumir();
        if (!item) break;
        std::cout << "C" << id << " procesó " << *item << '\n';
        consumidos.fetch_add(1);
        std::this_thread::sleep_for(std::chrono::milliseconds(40));
    }
}

int main() {
    BufCircular buf(CAPACIDAD);
    std::vector<std::thread> hilos;

    for (int i = 0; i < 3; ++i)
        hilos.emplace_back(productor, std::ref(buf), i);
    for (int i = 0; i < 3; ++i)
        hilos.emplace_back(consumidor, std::ref(buf), i);

    // Esperar a que todos los productores terminen
    for (size_t i = 0; i < 3; ++i)
        hilos[i].join();

    buf.cerrar();

    // Esperar a los consumidores
    for (size_t i = 3; i < hilos.size(); ++i)
        hilos[i].join();

    std::cout << "Producidos: " << producidos.load()
              << ", Consumidos: " << consumidos.load() << '\n';
}