Producteur et consommateur utilisant des sémaphores

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

Les sémaphores permettent d'obtenir un niveau de concurrence plus élevé que les mutex. Si les accès au tampon étaient 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 différentes parties 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 sémaphores qui le protègent sont des variables globales.

Une alternative à l'utilisation de QSemaphore pour résoudre le problème producteur-consommateur est d'utiliser QWaitCondition et QMutex. C'est ce que fait l'exemple du producteur et du consommateur utilisant des conditions d'attente.

Variables globales

Commençons par examiner le tampon circulaire et les sémaphores associés :

constexpr int DataSize = 100000;

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

QSemaphore freeBytes(BufferSize);
QSemaphore usedBytes;

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 sémaphores. Le sémaphore freeBytes contrôle la zone "libre" du tampon (la zone que le producteur n'a pas encore remplie de données ou que le consommateur a déjà lue). Le sémaphore usedBytes contrôle la zone "utilisée" de la mémoire tampon (la zone que le producteur a remplie mais que le consommateur n'a pas encore lue).

Ensemble, ces sémaphores 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.

Le sémaphore freeBytes est initialisé avec BufferSize, car le tampon est initialement vide. Le sémaphore usedBytes est initialisé à 0 (la valeur par défaut si aucune valeur n'est spécifiée).

Classe de producteur

Examinons le code de la classe 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();
        }
    }
};

Le producteur génère DataSize octets de données. Avant d'écrire un octet dans le tampon circulaire, il doit acquérir un octet "libre" à l'aide du sémaphore freeBytes. L'appel à QSemaphore::acquire() peut bloquer si le consommateur n'a pas suivi le rythme du producteur.

À la fin, le producteur libère un octet en utilisant le sémaphore usedBytes. L'octet "libre" a été transformé avec succès en octet "utilisé", prêt à être lu par le consommateur.

Classe de consommateur

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

Le code est très similaire à celui du producteur, sauf que cette fois nous acquérons un octet "utilisé" et libérons un octet "libre", au lieu de l'inverse.

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 ne 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 le sémaphore usedBytes soit libéré (son compte initial available() est 0). Une fois que le producteur a placé un octet dans le tampon, freeBytes.available() est BufferSize - 1 et usedBytes.available() est 1. À ce stade, deux choses peuvent se produire : Soit le thread consommateur prend le relais et lit cet octet, soit le thread 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 les 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. L'acquisition et la libération d'un QSemaphore 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