Productor y consumidor con condiciones de espera

El productor escribe datos en el buffer hasta que llega al final del buffer, momento en el que reinicia desde el principio, sobrescribiendo los datos existentes. El hilo consumidor lee los datos a medida que se producen y los escribe en el error estándar.

Las condiciones de espera hacen posible tener un mayor nivel de concurrencia de lo que es posible sólo con mutexes. Si los accesos al buffer estuvieran simplemente protegidos por un QMutex, el hilo consumidor no podría acceder al buffer al mismo tiempo que el hilo productor. Sin embargo, no hay nada de malo en tener ambos hilos trabajando en diferentes partes del buffer al mismo tiempo.

El ejemplo consta de dos clases: Producer y Consumer. Ambas heredan de QThread. El buffer circular utilizado para la comunicación entre estas dos clases y las herramientas de sincronización que lo protegen son variables globales.

Una alternativa al uso de QWaitCondition y QMutex para resolver el problema productor-consumidor es utilizar QSemaphore. Esto es lo que hace el ejemplo Productor y Consumidor usando Semáforos.

Variables globales

Empecemos revisando el buffer circular y las herramientas de sincronización asociadas:

constexpr int DataSize = 100000;
constexpr int BufferSize = 8192;

QMutex mutex; // protects the buffer and the counter
char buffer[BufferSize];
int numUsedBytes;

QWaitCondition bufferNotEmpty;
QWaitCondition bufferNotFull;

DataSize es la cantidad de datos que generará el productor. Para mantener el ejemplo lo más simple posible, lo convertimos en una constante. BufferSize es el tamaño del buffer circular. Es menor que DataSize, lo que significa que en algún momento el productor llegará al final del búfer y volverá a empezar desde el principio.

Para sincronizar el productor y el consumidor, necesitamos dos condiciones de espera y un mutex. La condición bufferNotEmpty se señala cuando el productor ha generado algunos datos, indicando al consumidor que puede empezar a leerlos. La condición bufferNotFull se activa cuando el consumidor ha leído algunos datos, indicando al productor que puede generar más. numUsedBytes es el número de bytes del búfer que contienen datos.

Juntas, las condiciones de espera, el mutex y el contador numUsedBytes aseguran que el productor nunca esté más de BufferSize bytes por delante del consumidor, y que el consumidor nunca lea datos que el productor no haya generado todavía.

Clase Productor

Revisemos el código de la clase Producer:

class Producer : public QThread
{
public:
    explicit Producer(QObject *parent = nullptr)
        : QThread(parent)
    {
    }

private:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            {
                const QMutexLocker locker(&mutex);
                while (numUsedBytes == BufferSize)
                    bufferNotFull.wait(&mutex);
            }

            buffer[i % BufferSize] = "ACGT"[QRandomGenerator::global()->bounded(4)];

            {
                const QMutexLocker locker(&mutex);
                ++numUsedBytes;
                bufferNotEmpty.wakeAll();
            }
        }
    }
};

El productor genera DataSize bytes de datos. Antes de escribir un byte en el búfer circular, debe comprobar si el búfer está lleno (es decir, numUsedBytes es igual a BufferSize). Si el búfer está lleno, el hilo espera en la condición bufferNotFull.

Al final, el productor incrementa numUsedBytes y señala que la condición bufferNotEmpty es verdadera, ya que numUsedBytes es necesariamente mayor que 0.

Protegemos todos los accesos a la variable numUsedBytes con un mutex. Además, la función QWaitCondition::wait() acepta un mutex como argumento. Este mutex se desbloquea antes de que el hilo se ponga a dormir y se bloquea cuando el hilo se despierta. Además, la transición del estado bloqueado al estado de espera es atómica, para evitar que se produzcan condiciones de carrera.

Clase Consumidor

Pasemos a la clase Consumer:

class Consumer : public QThread
{
public:
    explicit Consumer(QObject *parent = nullptr)
        : QThread(parent)
    {
    }

private:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            {
                const QMutexLocker locker(&mutex);
                while (numUsedBytes == 0)
                    bufferNotEmpty.wait(&mutex);
            }

            fprintf(stderr, "%c", buffer[i % BufferSize]);

            {
                const QMutexLocker locker(&mutex);
                --numUsedBytes;
                bufferNotFull.wakeAll();
            }
        }
        fprintf(stderr, "\n");
    }
};

El código es muy similar al del productor. Antes de leer el byte, comprobamos si el buffer está vacío (numUsedBytes es 0) en lugar de si está lleno y esperamos en la condición bufferNotEmpty si está vacío. Después de leer el byte, decrementamos numUsedBytes (en lugar de incrementarlo), y señalamos la condición bufferNotFull (en lugar de la condición bufferNotEmpty ).

La función main()

En main(), creamos los dos hilos y llamamos a QThread::wait() para asegurarnos de que ambos hilos tienen tiempo de terminar antes de que salgamos:

int main(int argc, char *argv[])
{
    QCoreApplication app(argc, argv);
    Producer producer;
    Consumer consumer;
    producer.start();
    consumer.start();
    producer.wait();
    consumer.wait();
    return 0;
}

¿Qué ocurre cuando ejecutamos el programa? Inicialmente, el hilo productor es el único que puede hacer algo; el consumidor está bloqueado esperando a que se señale la condición bufferNotEmpty (numUsedBytes es 0). Una vez que el productor ha puesto un byte en el buffer, numUsedBytes es estrictamente mayor que 0, y la condición bufferNotEmpty es señalada. En ese momento, pueden ocurrir dos cosas: O bien el hilo consumidor toma el relevo y lee ese byte, o bien el productor llega a producir un segundo byte.

El modelo productor-consumidor presentado en este ejemplo hace posible escribir aplicaciones multihilo altamente concurrentes. En una máquina multiprocesador, el programa es potencialmente hasta dos veces más rápido que el programa equivalente basado en mutex, ya que los dos hilos pueden estar activos al mismo tiempo en diferentes partes del buffer.

Sin embargo, ten en cuenta que estas ventajas no siempre se materializan. Bloquear y desbloquear un QMutex tiene un coste. En la práctica, probablemente valdría la pena dividir el buffer en trozos y operar sobre trozos en lugar de bytes individuales. El tamaño del búfer también es un parámetro que debe seleccionarse cuidadosamente, basándose en la experimentación.

Proyecto de ejemplo @ code.qt.io