Productor y consumidor con semáforos

El productor escribe datos en el buffer hasta que alcanza el 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.

Los semáforos hacen posible un mayor nivel de concurrencia que los mutex. Si los accesos al buffer estuvieran protegidos por un QMutex, el hilo consumidor no podría acceder al buffer al mismo tiempo que el hilo productor. Sin embargo, no hay inconveniente en que ambos hilos trabajen 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 los semáforos que lo protegen son variables globales.

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

Variables globales

Empecemos revisando el buffer circular y los semáforos asociados:

constexpr int DataSize = 100000;

constexpr int BufferSize = 8192;
char buffer[BufferSize];

QSemaphore freeBytes(BufferSize);
QSemaphore usedBytes;

DataSize es la cantidad de datos que generará el productor. Para mantener el ejemplo lo más simple posible, lo hacemos 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 semáforos. El semáforo freeBytes controla el área "libre" del buffer (el área que el productor aún no ha llenado con datos o que el consumidor ya ha leído). El semáforo usedBytes controla el área "utilizada" del búfer (el área que el productor ha llenado pero que el consumidor aún no ha leído).

Juntos, los semáforos garantizan que el productor nunca esté más de BufferSize bytes por delante del consumidor, y que el consumidor nunca lea datos que el productor aún no haya generado.

El semáforo freeBytes se inicializa con BufferSize, porque inicialmente todo el búfer está vacío. El semáforo usedBytes se inicializa con 0 (el valor por defecto si no se especifica ninguno).

Clase Productora

Revisemos el código de la clase Producer:

class Producer : public QThread
{
public:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            freeBytes.acquire();
            buffer[i % BufferSize] = "ACGT"[QRandomGenerator::global()->bounded(4)];
            usedBytes.release();
        }
    }
};

El productor genera DataSize bytes de datos. Antes de escribir un byte en el buffer circular, debe adquirir un byte "libre" utilizando el semáforo freeBytes. La llamada a QSemaphore::acquire() puede bloquearse si el consumidor no ha seguido el ritmo del productor.

Al final, el productor libera un byte utilizando el semáforo usedBytes. El byte "libre" se ha transformado con éxito en un byte "usado", listo para ser leído por el consumidor.

Clase Consumidor

Pasemos ahora a la clase Consumer:

class Consumer : public QThread
{
public:
    void run() override
    {
        for (int i = 0; i < DataSize; ++i) {
            usedBytes.acquire();
            fprintf(stderr, "%c", buffer[i % BufferSize]);
            freeBytes.release();
        }
        fprintf(stderr, "\n");
    }
};

El código es muy similar al del productor, salvo que esta vez adquirimos un byte "usado" y liberamos un byte "libre", en lugar de lo contrario.

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 libere el semáforo usedBytes (su cuenta inicial available() es 0). Una vez que el productor ha puesto un byte en el buffer, freeBytes.available() es BufferSize - 1 y usedBytes.available() es 1. En ese momento, pueden pasar dos cosas: O bien el hilo consumidor toma el relevo y lee ese byte, o bien el hilo 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. Adquirir y liberar un QSemaphore 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