Producteur et consommateur utilisant des sémaphores

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'erreurs standard.

Les sémaphores permettent d’atteindre 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. Or, 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 sémaphores qui le protègent sont des variables globales.

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

Variables globales

Commençons par passer en revue 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 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 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 « occupée » du 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é à ` BufferSize`, car au départ, l’intégralité du tampon est vide. Le sémaphore ` usedBytes ` est initialisé à 0 (valeur par défaut si aucune n’est spécifiée).

Classe « Producer »

Passons en revue 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 se bloquer si le consommateur n’a pas suivi le rythme du producteur.

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

Classe 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-ci, 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 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 consommateur est bloqué, attendant que le sémaphore usedBytes soit libéré (sa valeur initiale de available() est 0). Une fois que le producteur a placé un octet dans le tampon, ` freeBytes.available() ` vaut ` BufferSize - 1` et ` usedBytes.available() ` vaut 1. À ce stade, deux scénarios sont possibles : soit le thread consommateur prend le relais et lit cet octet, soit le thread producteur peut 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. L'acquisition et la libération d'un « QSemaphore » 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