Producteur et consommateur utilisant des conditions d'attente

Le producteur écrit des données dans le tampon jusqu’à ce qu’il atteigne la fin de celui-ci, après quoi il recommence depuis le début, écrasant les données existantes. Le thread consommateur lit les données au fur et à mesure qu’elles sont produites et les écrit dans le flux d’erreur standard.

Les conditions d’attente permettent d’atteindre un niveau de concurrence plus élevé que celui rendu possible par les mutex seuls. Si les accès au tampon étaient simplement protégés par un QMutex, le thread consommateur ne pourrait pas accéder au tampon en même temps que le thread producteur. Pourtant, il n’y a aucun inconvénient à ce que les deux threads travaillent simultanément sur différentes parties du tampon.

L'exemple comprend deux classes : ` Producer ` et ` Consumer`. Toutes deux héritent de ` QThread`. Le tampon circulaire utilisé pour la communication entre ces deux classes et les outils de synchronisation qui le protègent sont des variables globales.

Une alternative à l’utilisation de ` QWaitCondition ` et ` QMutex ` pour résoudre le problème producteur-consommateur consiste à utiliser ` QSemaphore`. C’est ce que fait l’exemple « Producteur et consommateur utilisant des sémaphores ».

Variables globales

Commençons par passer en revue le tampon circulaire et les outils de synchronisation associés :

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 est la quantité de données que le producteur va générer. Pour que l'exemple reste aussi simple que possible, nous en faisons une constante. ` BufferSize ` est la taille du tampon circulaire. Elle est inférieure à ` DataSize`, ce qui signifie qu'à un moment donné, le producteur atteindra la fin du tampon et recommencera depuis le début.

Pour synchroniser le producteur et le consommateur, nous avons besoin de deux conditions d’attente et d’un mutex. La condition bufferNotEmpty est signalée lorsque le producteur a généré des données, indiquant au consommateur qu’il peut commencer à les lire. La condition bufferNotFull est signalée lorsque le consommateur a lu des données, indiquant au producteur qu’il peut en générer davantage. Le compteur « numUsedBytes » correspond au nombre d’octets du tampon contenant des données.

Ensemble, les conditions d’attente, le mutex et le compteur « numUsedBytes » garantissent que le producteur n’a jamais plus de « BufferSize » octets d’avance sur le consommateur, et que le consommateur ne lit jamais de données que le producteur n’a pas encore générées.

Classe Producteur

Passons en revue le code de la classe ` 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();
            }
        }
    }
};

Le producteur génère DataSize octets de données. Avant d’écrire un octet dans le tampon circulaire, il doit d’abord vérifier si le tampon est plein (c’est-à-dire si numUsedBytes est égal à BufferSize). Si le tampon est plein, le thread attend sur la condition bufferNotFull.

À la fin, le producteur incrémente numUsedBytes et signale que la condition bufferNotEmpty est vraie, puisque numUsedBytes est nécessairement supérieur à 0.

Nous protégeons tous les accès à la variable numUsedBytes à l’aide d’un mutex. De plus, la fonction QWaitCondition::wait() accepte un mutex comme argument. Ce mutex est déverrouillé avant que le thread ne soit mis en veille et verrouillé lorsque le thread se réveille. Par ailleurs, la transition de l’état verrouillé à l’état d’attente est atomique, afin d’empêcher toute condition de concurrence.

Classe Consumer

Passons maintenant à la classe 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");
    }
};

Le code est très similaire à celui du producteur. Avant de lire l’octet, nous vérifions si le tampon est vide (numUsedBytes vaut 0) plutôt que s’il est plein, et nous attendons la condition bufferNotEmpty s’il est vide. Une fois l’octet lu, nous décrémentons ` numUsedBytes ` (au lieu de l’incrémenter), et nous signalons la condition ` bufferNotFull ` (au lieu de la condition ` bufferNotEmpty `).

La fonction main()

Dans ` main()`, nous créons les deux threads et appelons ` QThread::wait()` pour nous assurer que les deux threads aient le temps de se terminer avant de quitter le programme :

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

Que se passe-t-il donc lorsque nous exécutons le programme ? Au départ, le thread producteur est le seul à pouvoir agir ; le thread consommateur est bloqué, attendant que la condition « bufferNotEmpty » soit signalée (la valeur de `numUsedBytes ` est égale à 0). Une fois que le producteur a placé un octet dans le tampon, la valeur de ` numUsedBytes ` est strictement supérieure à 0, et la condition « bufferNotEmpty » est signalée. À ce stade, deux scénarios sont possibles : soit le thread consommateur prend le relais et lit cet octet, soit le producteur parvient à produire un deuxième octet.

Le modèle producteur-consommateur présenté dans cet exemple permet d’écrire des applications multithread à forte concurrence. Sur une machine multiprocesseur, le programme peut potentiellement être jusqu’à deux fois plus rapide qu’un programme équivalent basé sur des mutex, puisque les deux threads peuvent être actifs simultanément sur différentes parties du tampon.

Sachez toutefois que ces avantages ne se concrétisent pas toujours. Le verrouillage et le déverrouillage d’un « QMutex » ont un coût. En pratique, il serait probablement judicieux de diviser le tampon en blocs et d’opérer sur ces blocs plutôt que sur des octets individuels. La taille du tampon est également un paramètre qui doit être choisi avec soin, à l’issue d’expérimentations.

Exemple de projet sur code.qt.io