Colas

Colas seguras para hilos

La cola bloqueante es el componente de software más importante de una instalación concurrente: es el buffer entre estaciones. Los hilos productores depositan mensajes o piezas y se bloquean si la cola está llena; los consumidores retiran y se bloquean si está vacía. Aquí la construimos desde cero.

La instalación en acción

Un pulmón de tres posiciones separa fabricación y procesado: puro paso de mensajes con push/pop bloqueantes. Arranca la simulación y observa el backpressure: cuando el pulmón se llena, la fabricación se bloquea hasta que el consumidor libera un hueco.

Consejo: cuenta las veces que la fabricación queda parada y relaciónalo con la ocupación media.

Por qué std::queue no basta

std::queue no es thread-safe: si el hilo de la cinta hace push mientras el de la máquina hace pop, hay data race. Y aunque se protegiera con un mutex, faltaría la segunda mitad del problema: bloquear en vez de sondear. Queremos que pop() duerma al consumidor hasta que haya datos, y que push() duerma al productor si no hay hueco.

La cola bloqueante completa

#include <queue>
#include <mutex>
#include <condition_variable>

template <typename T>
class ColaBloqueante {
    std::queue<T> datos;
    mutable std::mutex m;
    std::condition_variable noVacia, noLlena;
    size_t capacidad;

public:
    explicit ColaBloqueante(size_t cap) : capacidad(cap) {}

    void push(T item) {                       // productor
        std::unique_lock<std::mutex> lk(m);
        noLlena.wait(lk, [&] {                // espera hueco
            return datos.size() < capacidad;
        });
        datos.push(std::move(item));
        noVacia.notify_one();                 // hay un elemento más
    }

    T pop() {                                 // consumidor
        std::unique_lock<std::mutex> lk(m);
        noVacia.wait(lk, [&] {                // espera elemento
            return !datos.empty();
        });
        T item = std::move(datos.front());
        datos.pop();
        noLlena.notify_one();                 // hay un hueco más
        return item;
    }

    size_t size() const {
        std::lock_guard<std::mutex> lk(m);
        return datos.size();
    }
};

Observa la simetría con la pareja de semáforos sp/sc: noLlena es el semáforo de huecos y noVacia el de elementos, pero integrados y sin riesgo de olvidar un signal.

Anatomía del push

1. Se toma el mutex (vía unique_lock): nadie puede tocar la cola mientras decidimos.
2. wait con predicado: si está llena, suelta el mutex y duerme; al despertar vuelve a comprobar.
3. Se inserta el elemento (con std::move para evitar copias de mensajes grandes).
4. Se notifica noVacia: si un consumidor dormía esperando datos, despierta.
5. Al salir del ámbito, el mutex se libera solo.

Capacidad: el parámetro de diseño

• Cola ilimitada (capacidad enorme): el productor nunca se bloquea… y la memoria crece sin control si el consumidor es lento. En planta casi nunca interesa.
• Cola limitada: cuando se llena, el productor se bloquea. La presión se propaga hacia atrás por toda la línea (backpressure) y la instalación se autorregula: nadie produce más de lo que se puede consumir. Es la opción por defecto recomendada.

Variantes útiles

pop con timeout (para supervisión)

// devuelve false si pasa el tiempo sin datos; útil para hilos que
// además de consumir deben vigilar alarmas o la parada
bool popDurante(T& item, std::chrono::milliseconds t) {
    std::unique_lock<std::mutex> lk(m);
    if (!noVacia.wait_for(lk, t, [&] { return !datos.empty(); }))
        return false;                    // timeout
    item = std::move(datos.front());
    datos.pop();
    noLlena.notify_one();
    return true;
}

try_push / try_pop (sin bloqueo)

Versiones que devuelven false inmediatamente si no pueden operar, para hilos que no pueden permitirse quedar bloqueados (p. ej. el hilo de seguridad).

Cola bloqueante con semáforos (estilo clásico)

La misma estructura puede expresarse con semáforos explícitos, como en los ejercicios:

// sp = semáforo de huecos (valor inicial = capacidad)
// sc = semáforo de elementos (valor inicial = 0)
// sM = semáforo mutex (valor inicial = 1)

void push(T item) {
    sp.wait();            // espera hueco
    sM.wait();            // región crítica
    buffer.push(item);
    sM.signal();
    sc.signal();          // un elemento más
}

El riesgo de esta versión es hacerlo mal (olvidar el mutex, señalizar en mal orden); por eso en C++ moderno se encapsula en una clase como ColaBloqueante.