Producteur et consommateur utilisant les conditions d'attente

Le producteur écrit des données dans la mémoire tampon jusqu'à ce qu'il atteigne la fin de la mémoire tampon, auquel cas il recommence depuis le début, en écrasant les données existantes. Le thread consommateur lit les données au fur et à mesure qu'elles sont produites et les écrit sur l'erreur standard.

Les conditions d'attente permettent d'obtenir un niveau de concurrence plus élevé que ce qui est possible avec les seuls mutex. 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 en même temps sur des parties différentes de la mémoire 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 est d'utiliser QSemaphore. C'est ce que fait l'exemple Producer and Consumer using Semaphores.

Variables globales

Commençons par examiner 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 soit 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 d'autres. numUsedBytes est le nombre d'octets de la mémoire tampon qui contiennent 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 de producteur

Examinons 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 que numUsedBytes est égal à BufferSize). Si la mémoire tampon est pleine, le thread attend 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 avec 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 sommeil et verrouillé lorsque le thread se réveille. En outre, la transition de l'état verrouillé à l'état d'attente est atomique, afin d'éviter les conditions de course.

Classe de consommateur

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 de la classe producer. Avant de lire l'octet, nous vérifions si le tampon est vide (numUsedBytes vaut 0) au lieu de vérifier s'il est plein et nous attendons la condition bufferNotEmpty s'il est vide. Après avoir lu l'octet, 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 ont le temps de se terminer avant que nous quittions 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 lorsque nous exécutons le programme ? Au départ, le thread du producteur est le seul à pouvoir faire quelque chose ; le consommateur est bloqué en attendant que la condition bufferNotEmpty soit signalée (numUsedBytes est 0). Une fois que le producteur a mis un octet dans le tampon, numUsedBytes est strictement supérieur à 0, et la condition bufferNotEmpty est signalée. À ce moment-là, deux choses peuvent se produire : Soit le thread consommateur prend le relais et lit cet octet, soit le producteur produit un deuxième octet.

Le modèle producteur-consommateur présenté dans cet exemple permet d'écrire des applications multithreads hautement concurrentes. Sur une machine multiprocesseur, le programme est potentiellement jusqu'à deux fois plus rapide que le programme équivalent basé sur le mutex, puisque les deux threads peuvent être actifs en même temps sur différentes parties du tampon.

Il faut cependant savoir que ces avantages ne sont pas toujours réalisés. Le verrouillage et le déverrouillage d'un site QMutex ont un coût. Dans la pratique, il serait probablement utile de diviser le tampon en morceaux et d'opérer sur des morceaux plutôt que sur des octets individuels. La taille de la mémoire tampon est également un paramètre qui doit être choisi avec soin, sur la base de l'expérimentation.

Exemple de projet @ code.qt.io