Programación multi-hilos usando QWaitCondition y QMutex.

En el artículo anterior mostramos los principales elementos a tener en cuenta para realizar programación concurrente usando el Framework Qt, específicamente empleamos la clase QSemaphore para resolver el problema del productor-consumidor. En este nuevo artículo explicaremos otra alternativa que también es muy eficiente para programar aplicaciones multi-hilos, se trata de la combinación de QWaitCondition y QMutex.

El objetivo del artículo es mostrar cómo utilizar QWaitCondition y QMutex para controlar el acceso a un búfer circular que comparten un hilo productor y otro hilo consumidor [1].

El productor escribe datos en el búfer hasta que se llega al final del búfer, en cuyo punto se reinicia desde el principio, sobrescribiendo los datos existentes. El hilo consumidor lee los datos a medida que se produce y lo escribe en la salida de error estándar.

Las condiciones de espera (Wait Condition) hacen que sea posible tener un mayor nivel de concurrencia que el que lograríamos si usáramos solamente mutex. Si los accesos al búfer simplemente fueron vigilados por un QMutex, el hilo consumidor no podría acceder al búfer al mismo tiempo que el hilo productor. Sin embargo, no hay nada malo en tener dos hilos que trabajan en diferentes partes del búfer al mismo tiempo.

El ejemplo consta de dos clases: Productor (Producer) y Consumidor (Consumer). Ambos heredan de QThread. El búfer circular utilizado para la comunicación entre estas dos clases y las herramientas de sincronización que lo protegen son variables globales.

Variables globales

Vamos a empezar por la revisión del búfer circular y las herramientas de sincronización asociadas:

const int DataSize = 100000;

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

QWaitCondition bufferNotEmpty;
QWaitCondition bufferNotFull;
QMutex mutex;
int numUsedBytes = 0;

DataSize es la cantidad de datos que el productor va a generar. Para mantener el ejemplo tan simple como sea posible, hacemos una constante. BufferSize es el tamaño del búfer circular. Es menos de DataSize, lo que significa que en algún momento el productor alcanzará el final del búfer y reiniciará 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, diciéndole al consumidor que puede empezar a leerlo. La condición bufferNotFull se señala cuando el consumidor haya leído algunos datos, dice la productora que pueda generar mucho más. La variable numUsedBytes indica el número de bytes en el búfer que contiene los datos.

Juntos, las condiciones de espera, la exclusión mutua, y el contador numUsedBytes deben garantizar que el productor nunca está escribiendo bytes en una ubicación mayor que bufferSize por delante del consumidor, y que el consumidor nunca lee los datos que el productor no ha generado todavía.

Clase Productor

Repasemos el código de la clase Producer:

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

    void run()
    {
        qsrand(QTime(0,0,0).secsTo(QTime::currentTime()));

        for (int i = 0; i < DataSize; ++i) {
            mutex.lock();
            if (numUsedBytes == BufferSize)
                bufferNotFull.wait(&mutex);
            mutex.unlock();

            buffer[i % BufferSize] = "ACGT"[(int)qrand() % 4];

            mutex.lock();
            ++numUsedBytes;
            bufferNotEmpty.wakeAll();
            mutex.unlock();
        }
    }
};

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

Al final, el productor incrementa la variable numUsedBytes y envía la señal indicando que la condición bufferNotEmpty es verdadera, ya que numUsedBytes es necesariamente mayor que 0.

Guardamos todos los accesos a la variable numUsedBytes con un mutex. Además, la función QWaitCondition::wait() acepta un mutex como argumento. Este mutex está desbloqueado antes el hilo se pone a dormir y bloqueado cuando el hilo se despierta. Por otra parte, la transición desde el estado de bloqueo al estado de espera es atómica, para evitar que se produzcan las condiciones de carrera.

Clase del Consumidor

Volvamos a la clase de los consumidores:

class Consumer : public QThread
{
    Q_OBJECT
public:
    Consumer(QObject *parent = NULL) : QThread(parent)
    {
    }

    void run()
    {
        for (int i = 0; i < DataSize; ++i) {
            mutex.lock();
            if (numUsedBytes == 0)
                bufferNotEmpty.wait(&mutex);
            mutex.unlock();

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

            mutex.lock();
            --numUsedBytes;
            bufferNotFull.wakeAll();
            mutex.unlock();
        }
        fprintf(stderr, "\n");
    }

signals:
    void stringConsumed(const QString &text);
};

El código es muy similar a la del productor. Antes de leer el byte, comprobamos si el búfer está vacío (numUsedBytes es 0) y esperamos en la condición bufferNotEmpty si está vacío. Después de que hemos leído el byte, hacemos el decremento de la variable numUsedBytes (en lugar de incrementarlo), y señalamos la condición bufferNotFull (en lugar de la condición bufferNotEmpty).

La función main ()

En la función main (), creamos los dos hilos y llamamos a la función QThread::wait() para asegurar de que ambos contextos de ejecución tienen tiempo para terminar antes de la salida:

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

Entonces, ¿qué sucede cuando ejecutamos el programa? Inicialmente, el hilo productor es el único que puede hacer cualquier cosa, el consumidor está bloqueado en espera de que la condición bufferNotEmpty sea señalizada (numUsedBytes es 0). Una vez que el productor ha puesto un byte en el buffer, numUsedBytes es BufferSize - 1 y la condición bufferNotEmpty se marca. En ese momento, pueden ocurrir dos cosas: o bien el hilo consumidor se hace cargo y lee el byte, o el productor genera un segundo byte.

El modelo productor-consumidor que se presenta en este ejemplo permite escribir aplicaciones multiprocesos altamente concurrentes. En un equipo con varios procesadores, el programa es potencialmente hasta el doble de rápido que el programa equivalente basado en exclusión mutua, ya que los dos hilos pueden estar activos al mismo tiempo en diferentes partes del búfer.

Tenga en cuenta que estos beneficios no siempre son efectivos. Los procesos de bloqueo y desbloqueo de un QMutex tiene un costo. En la práctica, probablemente valdría la pena dividir el búfer en trozos, pues es más eficiente operar en trozos en lugar de bytes individuales. El tamaño del búfer es también un parámetro que se debe seleccionar cuidadosamente, basado en la experimentación.

El código fuente de esta aplicación pueden descargarlo desde aquí.

---