L'article manquant sur le multithreading Qt en C++

Les développeurs C++ s'efforcent de créer des applications Qt multithread robustes, mais le multithreading n'a jamais été facile avec toutes ces conditions de concurrence, synchronisation, interblocages et livelocks. À votre crédit, vous n'abandonnez pas et vous vous retrouvez à parcourir StackOverflow. Néanmoins, choisir la bonne solution parmi une douzaine de réponses différentes n'est pas une mince affaire, d'autant plus que chaque solution présente ses propres inconvénients.

Le multithreading est un modèle de programmation et d'exécution répandu qui permet à plusieurs threads d'exister dans le contexte d'un processus. Ces threads partagent les ressources du processus mais peuvent s'exécuter indépendamment. Le modèle de programmation par threads fournit aux développeurs une abstraction utile de l'exécution simultanée. Le multithreading peut également être appliqué à un processus pour permettre une exécution parallèle sur un système multiprocesseur.

- Wikipédia

Le but de cet article est d'agréger les connaissances essentielles sur la programmation concurrente avec le framework Qt, en particulier les sujets les plus mal compris. Un lecteur doit avoir des connaissances préalables en Qt et C++ pour comprendre le contenu.

Choisir entre utiliser QThreadPool et QThread

Le framework Qt offre de nombreux outils pour le multithreading. Choisir le bon outil peut être difficile au début, mais en fait, l'arbre de décision se compose de seulement deux options : soit vous voulez que Qt gère les threads pour vous, soit vous voulez gérer les threads par vous-même. Cependant, il existe d'autres critères importants:

1. Tâches qui n'ont pas besoin de la boucle d'événements. Plus précisément, les tâches qui n'utilisent pas le mécanisme signal/slot lors de l'exécution de la tâche.
   Utilisation : QtConcurrent et QThreadPool + QRunnable.

2. Tâches qui utilisent des signaux/slots et qui ont donc besoin de la boucle d'événements.
   Utilisation : Objets de travail déplacés vers + QThread.

La grande flexibilité du framework Qt permet de contourner le problème de « boucle d'événement manquante » et d'en ajouter un à QRunnable :

 class MyTask : public QObject, public QRunnable { Q_OBJECT public: void MyTask::run() { _loop.exec(); } public slots: // you need a signal connected to this slot to exit the loop, // otherwise the thread running the loop would remain blocked... void finishTask() { _loop.exit(); } private: QEventLoop _loop; }

Essayez d'éviter ces "solutions de contournement", car elles sont dangereuses et inefficaces : si l'un des threads du pool de threads (exécutant MyTask) est bloqué en raison de l'attente d'un signal, il ne peut pas exécuter d'autres tâches du pool.

Vous pouvez également exécuter un QThread sans aucune boucle d'événement en remplaçant la méthode QThread::run() et c'est parfaitement bien tant que vous savez ce que vous faites. Par exemple, ne vous attendez pas à ce que la méthode quit() fonctionne dans ce cas.

Exécuter une instance de tâche à la fois

Imaginez que vous deviez vous assurer qu'une seule instance de tâche à la fois peut être exécutée et que toutes les demandes en attente pour exécuter la même tâche attendent dans une certaine file d'attente. Cela est souvent nécessaire lorsqu'une tâche accède à une ressource exclusive, comme l'écriture dans le même fichier ou l'envoi de paquets à l'aide d'un socket TCP.

Oublions un instant l'informatique et le modèle producteur-consommateur et considérons quelque chose de trivial ; quelque chose qui peut être facilement trouvé dans de vrais projets.

Une solution naïve à ce problème pourrait être d'utiliser un QMutex . Dans la fonction de tâche, vous pouvez simplement acquérir le mutex sérialisant efficacement tous les threads tentant d'exécuter la tâche. Cela garantirait qu'un seul thread à la fois pourrait exécuter la fonction. Cependant, cette solution a un impact sur les performances en introduisant un problème de conflit élevé car tous ces threads seraient bloqués (sur le mutex) avant de pouvoir continuer. Si vous avez de nombreux threads utilisant activement une telle tâche et effectuant un travail utile entre les deux, alors tous ces threads dormiront la plupart du temps.

 void logEvent(const QString & event) { static QMutex lock; QMutexLocker locker(& lock); // high contention! logStream << event; // exclusive resource }

Pour éviter les conflits, nous avons besoin d'une file d'attente et d'un travailleur qui vit dans son propre thread et traite la file d'attente. C'est à peu près le schéma classique producteur-consommateur . Le travailleur ( consommateur ) sélectionnerait les demandes de la file d'attente une par une, et chaque producteur peut simplement ajouter ses demandes dans la file d'attente. Cela semble simple au début et vous pouvez penser à utiliser QQueue et QWaitCondition , mais attendez et voyons si nous pouvons atteindre l'objectif sans ces primitives :

• Nous pouvons utiliser QThreadPool car il a une file d'attente de tâches en attente

Ou

• Nous pouvons utiliser QThread::run() par défaut car il a QEventLoop

La première option consiste à utiliser QThreadPool . Nous pouvons créer une instance QThreadPool et utiliser QThreadPool::setMaxThreadCount(1) . Ensuite, nous pouvons utiliser QtConcurrent::run() pour planifier les requêtes :

 class Logger: public QObject { public: explicit Logger(QObject *parent = nullptr) : QObject(parent) { threadPool.setMaxThreadCount(1); } void logEvent(const QString &event) { QtConcurrent::run(&threadPool, [this, event]{ logEventCore(event); }); } private: void logEventCore(const QString &event) { logStream << event; } QThreadPool threadPool; };

Cette solution a un avantage : QThreadPool::clear() vous permet d' annuler instantanément toutes les requêtes en attente, par exemple lorsque votre application doit s'arrêter rapidement. Cependant, il existe également un inconvénient important lié à l'affinité des threads : la fonction logEventCore s'exécutera probablement dans différents threads d'un appel à l'autre. Et nous savons que Qt a certaines classes qui nécessitent une affinité de thread : QTimer , QTcpSocket et peut-être quelques autres.

Ce que dit la spécification Qt à propos de l'affinité de thread : les minuteurs démarrés dans un thread ne peuvent pas être arrêtés à partir d'un autre thread. Et seul le thread possédant une instance de socket peut utiliser ce socket. Cela implique que vous devez arrêter tous les temporisateurs en cours d'exécution dans le thread qui les a démarrés et vous devez appeler QTcpSocket::close() dans le thread propriétaire du socket. Les deux exemples sont généralement exécutés dans des destructeurs.

La meilleure solution repose sur l'utilisation de QEventLoop fourni par QThread . L'idée est simple : nous utilisons un mécanisme de signal/slot pour émettre des requêtes, et la boucle d'événement exécutée à l'intérieur du thread servira de file d'attente permettant l'exécution d'un seul slot à la fois.

 // the worker that will be moved to a thread class LogWorker: public QObject { Q_OBJECT public: explicit LogWorker(QObject *parent = nullptr); public slots: // this slot will be executed by event loop (one call at a time) void logEvent(const QString &event); };

L'implémentation du constructeur LogWorker et de logEvent est simple et n'est donc pas fournie ici. Nous avons maintenant besoin d'un service qui gérera le thread et l'instance de travail :

 // interface class LogService : public QObject { Q_OBJECT public: explicit LogService(QObject *parent = nullptr); ~LogService(); signals: // to use the service, just call this signal to send a request: // logService->logEvent("event"); void logEvent(const QString &event); private: QThread *thread; LogWorker *worker; }; // implementation LogService::LogService(QObject *parent) : QObject(parent) { thread = new QThread(this); worker = new LogWorker; worker->moveToThread(thread); connect(this, &LogService::logEvent, worker, &LogWorker::logEvent); connect(thread, &QThread::finished, worker, &QObject::deleteLater); thread->start(); } LogService::~LogService() { thread->quit(); thread->wait(); } 

Voyons comment ce code fonctionne :

• Dans le constructeur, nous créons une instance de thread et de travailleur. Notez que le travailleur ne reçoit pas de parent, car il sera déplacé vers le nouveau thread. Pour cette raison, Qt ne pourra pas libérer automatiquement la mémoire du travailleur, et par conséquent, nous devons le faire en connectant le signal QThread::finished au slot deleteLater . Nous connectons également la méthode proxy LogService::logEvent() à LogWorker::logEvent() qui utilisera le mode Qt::QueuedConnection en raison de différents threads.
• Dans le destructeur, nous plaçons l'événement quit dans la file d'attente de la boucle d'événements. Cet événement sera traité après tous les autres événements. Par exemple, si nous avons effectué des centaines d' logEvent() juste avant l'appel du destructeur, le logger les traitera tous avant de récupérer l'événement quit. Cela prend du temps, bien sûr, nous devons donc wait() jusqu'à ce que la boucle d'événements se termine. Il convient de mentionner que toutes les futures demandes de journalisation postées après l'événement de fermeture ne seront jamais traitées.
• La journalisation elle-même ( LogWorker::logEvent ) sera toujours effectuée dans le même thread, donc cette approche fonctionne bien pour les classes nécessitant thread-affinity . En même temps, le constructeur et le destructeur de LogWorker sont exécutés dans le thread principal (en particulier le thread LogService s'exécute), et par conséquent, vous devez faire très attention au code que vous y exécutez. Plus précisément, n'arrêtez pas les temporisateurs ou n'utilisez pas de sockets dans le destructeur du travailleur à moins que vous ne puissiez exécuter le destructeur dans le même thread !

Exécution du destructeur du travailleur dans le même thread

Si votre travailleur a affaire à des temporisateurs ou à des sockets, vous devez vous assurer que le destructeur est exécuté dans le même thread (le thread que vous avez créé pour le travailleur et où vous l'avez déplacé). La manière évidente de prendre en charge cela est de sous- QThread et de delete le travailleur à l'intérieur QThread::run() . Considérez le modèle suivant :

 template <typename TWorker> class Thread : QThread { public: explicit Thread(TWorker *worker, QObject *parent = nullptr) : QThread(parent), _worker(worker) { _worker->moveToThread(this); start(); } ~Thread() { quit(); wait(); } TWorker worker() const { return _worker; } protected: void run() override { QThread::run(); delete _worker; } private: TWorker *_worker; };

À l'aide de ce modèle, nous redéfinissons LogService à partir de l'exemple précédent :

 // interface class LogService : public Thread<LogWorker> { Q_OBJECT public: explicit LogService(QObject *parent = nullptr); signals: void **logEvent**(const QString &event); }; // implementation LogService::**LogService**(QObject *parent) : Thread<LogWorker>(new LogWorker, parent) { connect(this, &LogService::logEvent, worker(), &LogWorker::logEvent); }

Voyons comment cela est censé fonctionner :

• Nous avons fait de LogService l'objet QThread car nous devions implémenter la run() personnalisée. Nous avons utilisé le sous-classement privé pour empêcher l'accès aux fonctions de QThread puisque nous voulons contrôler le cycle de vie du thread en interne.
• Dans la Thread::run() , nous exécutons la boucle d'événements en appelant l'implémentation par défaut QThread::run() et détruisons l'instance de travail juste après la fin de la boucle d'événements. Notez que le destructeur du worker est exécuté dans le même thread.
• LogService::logEvent() est la fonction proxy (signal) qui publiera l'événement de journalisation dans la file d'attente des événements du thread.

Mettre en pause et reprendre les discussions

Une autre opportunité intéressante est de pouvoir suspendre et reprendre nos threads personnalisés. Imaginez que votre application effectue un traitement qui doit être suspendu lorsque l'application est réduite, verrouillée ou qu'elle a simplement perdu la connexion réseau. Ceci peut être réalisé en créant une file d'attente asynchrone personnalisée qui contiendra toutes les demandes en attente jusqu'à ce que le travailleur soit repris. Cependant, puisque nous recherchons des solutions plus simples, nous utiliserons (à nouveau) la file d'attente de la boucle d'événements dans le même but.

Pour suspendre un thread, nous avons clairement besoin qu'il attende une certaine condition d'attente. Si le thread est bloqué de cette façon, sa boucle d'événements ne gère aucun événement et Qt doit mettre keep dans la file d'attente. Une fois reprise, la boucle d'événements traitera toutes les demandes accumulées. Pour la condition d'attente, nous utilisons simplement l'objet QWaitCondition qui nécessite également un QMutex . Pour concevoir une solution générique qui pourrait être réutilisée par n'importe quel travailleur, nous devons mettre toute la logique de suspension/reprise dans une classe de base réutilisable. Appelons-le SuspendableWorker . Une telle classe doit prendre en charge deux méthodes :

• suspend() serait un appel bloquant qui place le thread en attente sur une condition d'attente. Cela se ferait en publiant une demande de suspension dans la file d'attente et en attendant qu'elle soit traitée. Assez similaire à QThread::quit() + wait() .
• resume() signalerait à la condition d'attente de réveiller le thread endormi pour continuer son exécution.

Passons en revue l'interface et l'implémentation :

 // interface class SuspendableWorker : public QObject { Q_OBJECT public: explicit SuspendableWorker(QObject *parent = nullptr); ~SuspendableWorker(); // resume() must be called from the outer thread. void resume(); // suspend() must be called from the outer thread. // the function would block the caller's thread until // the worker thread is suspended. void suspend(); private slots: void suspendImpl(); private: QMutex _waitMutex; QWaitCondition _waitCondition; };

 // implementation SuspendableWorker::SuspendableWorker(QObject *parent) : QObject(parent) { _waitMutex.lock(); } SuspendableWorker::~SuspendableWorker() { _waitCondition.wakeAll(); _waitMutex.unlock(); } void SuspendableWorker::resume() { _waitCondition.wakeAll(); } void SuspendableWorker::suspend() { QMetaObject::invokeMethod(this, &SuspendableWorker::suspendImpl); // acquiring mutex to block the calling thread _waitMutex.lock(); _waitMutex.unlock(); } void SuspendableWorker::suspendImpl() { _waitCondition.wait(&_waitMutex); }

N'oubliez pas qu'un thread suspendu ne recevra jamais d'événement de quit . Pour cette raison, nous ne pouvons pas l'utiliser en toute sécurité avec vanilla QThread à moins de reprendre le fil avant de publier quitter. Intégrons cela dans notre modèle Thread<T> personnalisé pour le rendre à l'épreuve des balles.

 template <typename TWorker> class Thread : QThread { public: explicit Thread(TWorker *worker, QObject *parent = nullptr) : QThread(parent), _worker(worker) { _worker->moveToThread(this); start(); } ~Thread() { resume(); quit(); wait(); } void suspend() { auto worker = qobject_cast<SuspendableWorker*>(_worker); if (worker != nullptr) { worker->suspend(); } } void resume() { auto worker = qobject_cast<SuspendableWorker*>(_worker); if (worker != nullptr) { worker->resume(); } } TWorker worker() const { return _worker; } protected: void run() override { QThread::*run*(); delete _worker; } private: TWorker *_worker; };

Avec ces modifications, nous reprendrons le fil de discussion avant de publier l'événement d'abandon. En outre, Thread<TWorker> permet toujours à tout type de travailleur d'être transmis, qu'il s'agisse d'un SuspendableWorker ou non.

L'utilisation serait la suivante :

 LogService logService; logService.logEvent("processed event"); logService.suspend(); logService.logEvent("queued event"); logService.resume(); // "queued event" is now processed.

volatil vs atomique

C'est un sujet souvent mal compris. La plupart des gens pensent que les variables volatile peuvent être utilisées pour servir certains indicateurs accessibles par plusieurs threads et que cela préserve des conditions de concurrence des données. C'est faux, et les classes QAtomic* (ou std::atomic ) doivent être utilisées à cette fin.

Prenons un exemple réaliste : une classe de connexion TcpConnection qui fonctionne dans un thread dédié, et nous voulons que cette classe exporte une méthode thread-safe : bool isConnected() . En interne, la classe écoutera les événements de socket : connected et disconnected pour maintenir un indicateur booléen interne :

 // pseudo-code, won't compile class TcpConnection : QObject { Q_OBJECT public: // this is not thread-safe! bool isConnected() const { return _connected; } private slots: void handleSocketConnected() { _connected = true; } void handleSocketDisconnected() { _connected = false; } private: bool _connected; }

Rendre volatile le membre _connected ne résoudra pas le problème et ne rendra pas isConnected() thread-safe. Cette solution fonctionnera 99% du temps, mais le 1% restant fera de votre vie un cauchemar. Pour résoudre ce problème, nous devons protéger l'accès aux variables de plusieurs threads. Utilisons QReadWriteLocker à cette fin :

 // pseudo-code, won't compile class TcpConnection : QObject { Q_OBJECT public: bool isConnected() const { QReadLocker locker(&_lock); return _connected; } private slots: void handleSocketConnected() { QWriteLocker locker(&_lock); _connected = true; } void handleSocketDisconnected() { QWriteLocker locker(&_lock); _connected = false; } private: QReadWriteLocker _lock; bool _connected; }

Cela fonctionne de manière fiable, mais pas aussi rapidement que l'utilisation d'opérations atomiques "sans verrouillage". La troisième solution est à la fois rapide et thread-safe (l'exemple utilise std::atomic au lieu de QAtomicInt , mais sémantiquement, ils sont identiques) :

 // pseudo-code, won't compile class TcpConnection : QObject { Q_OBJECT public: bool isConnected() const { return _connected; } private slots: void handleSocketConnected() { _connected = true; } void handleSocketDisconnected() { _connected = false; } private: std::atomic<bool> _connected; }

Conclusion

Dans cet article, nous avons discuté de plusieurs préoccupations importantes concernant la programmation simultanée avec le framework Qt et avons conçu des solutions pour répondre à des cas d'utilisation spécifiques. Nous n'avons pas examiné de nombreux sujets simples tels que l'utilisation de primitives atomiques, les verrous en lecture-écriture et bien d'autres, mais si ceux-ci vous intéressent, laissez votre commentaire ci-dessous et demandez un tel didacticiel.

Si vous souhaitez explorer Qmake, j'ai également récemment publié The Vital Guide to Qmake. C'est une excellente lecture !