Thread Pool et futures – exemple de programmation par promesses

Ce qui suit détaille un exemple complet et opérationnel de programme implémentant un calcul séquentiel (la comptabilité du nombre de mots dans un fichier, appliqué à plusieurs fichiers; probablement le problème canonique dans le genre) et son implémentation parallèle, entre autres dans une optique de cibler intelligemment le moment le plus opportun pour passer d'une stratégie à l'autre. Nous explorerons au passage le schéma de conception Regroupement de threads (Thread Pool) et la métaphore de la programmation par promesses, à l'aide de futures.

Le problème que nous atttaquerons sera un classique dans le genre, c'est-à-dire compter les mots dans un fichier. C'est un problème qui, pour un ensemble de fichiers donné, se résout bien séquentiellement (on ouvre un fichier, on compte les mots qui y apparaissent, on le ferme, on passe au suivant), mais qui est tel qu'il est fortement parallélisable (le traitement de chaque fichier est indépendant de celui des autres; seul les moments où sont colligés les résultats est critique).

L'exemple proposé est complet et comprend plusieurs modules. Dans certains cas, les concepts proposés sont détaillés ailleurs sur le site de votre chic prof, alors soyez tolérant(e)s si vous rencontrez des renvois ici et là. J'ai essayé de présenter les modules en ordre croissant de dépendances, pour que les trucs les plus indépendants (les moins fortement couplés) apparaissent d'abord. J'espère que cela vous conviendra. Entre autres, nous aurons recours à l'idiome d'objets incopiables, à des mutex portables, des sections critiques portables et à des événements portables (que je laisse en exercice). Nous utiliserons aussi des assertions statiques, une technique de métaprogrammation bien connue.

Vous pourrez ainsi naviguer rapidement les sections qui sont moins essentielles au propos, à moins que vous n'ayez un intérêt pour les manoeuvres qui s'y trouvent (et qui sont, en soi, parfois assez amusantes et assez instructives). En escamotant des sections, évidemment, soyez conscient(e)s qu'il est possible que vous deviez faire un acte de foi ici et là en lien avec la technique escamotée au passage.

Survol conceptuel

Nous mettrons en place un système capable de comptabiliser le nombre de mots total apparaissant dans un ensemble de fichiers. Il est possible de comptabiliser le même fichier plusieurs fois si le code client le souhaite (filtrer la liste des noms de fichiers pour en éliminer la redondance n'est pas le propos de cet exercice), et on considère un mot comme étant une séquence de symboles séparés par des blancs, ce qui permettra de consommer des mots aussi simplement qu'on consommerait des std::string sur un std::istream à l'aide de l'opérateur >>.

Nous utiliserons aussi un regroupement de threads (en anglais : Thread Pool). Plutôt que de lancer des threads à la pièce, nous en construirons un certain nombre au préalable, et nous leur donnerons tous une seule et même responsabilité : piger une tâche parmi celles qui sont en attente de traitement, la prendre en charge, puis recommencer. Ces threads seront des ouvriers, ou Worker Threads.

Ceci permettra l'implémentation de futures de manière quelque peu simplifiée, et limitera le coût du démarrage des threads à celui encouru au moment initial (mis à part les moments où le système sera tellement sollicité par des tâches en cours qu'il devra accroître le nombre d'ouvriers à l'oeuvre. Les regroupements de threads foisonnent sur les plateformes contemporaines; vous pouvez remplacer celui proposé ici par celui de votre plateforme de prédilection dans la mesure où vous aurez pris soin d'adapter le code quelque peu.

Reste à voir comment, concrètement, nous compterons les mots de manière séquentielle (le code montrant le traitement en version parallèle a été placé dans la section sur les futures, un peu plus bas).

Le programme principal et son rôle

Le programme principal réalisera un ensemble de tests et exposera, dans un fichier de sortie, le fruit des calculs réalisés. Pour l'invoquer, il suffira de lancer le programme avec une liste de noms de fichiers, et de le laisser travailler. Les fruits de ce programme permettront entre autres de trouver un bon seuil de transition séquentiel/ parallèle pour un système donné.

Pour l'essentiel, donc, ce code lance des tâches de traitement séquentiel ou parallèle, tire des métriques, et entrepose le tout dans un fichier de sortie pour consultation ultérieure, à tête reposée. Ce programme fait la démonstration qu'il est possible d'utiliser le regroupement de threads et les futures, mais sert surtout à déterminer les conditions idéales utilisation de ces outils dans un contexte appliqué.

#include "statistiques.h"
#include "compter_mots.h"
#include "compter_mots_seriel.h"
#include "compter_mots_map_reduce.h"
#include "minuterie.h"
#include <sstream>
#include <iostream>
#include <iterator>
#include <algorithm>
#include <string>
#include <sstream>
using namespace std;
const char *terminaison(bool pluriel)
{
   static const char * TERMINAISONS [] = { "", "s" };
   return TERMINAISONS[pluriel? 1 : 0];
}
const char *config()
{
#ifdef _DEBUG
   return "(DEBUG)";
#else
   return "(RELEASE)";
#endif
}
template <class It>
   double tester_seriel(It debut, It fin, ostream &sortie, int n_essai)
   {
      stringstream sstr;
      sortie << "Approche sérielle, essai " << n_essai << endl;
      int n_seriel;
      double ecoule = 0.0;
      {
         minuterie minu{sstr, &ecoule};
         n_seriel = compter_mots_multidocuments_seriel(debut, fin);
      }
      sortie << '\t' << n_seriel << " mots au total" << endl;
      sortie << '\t' << sstr.str() << endl;
      return ecoule;
   }
template <class It>
   double tester_parallele(It debut, It fin, ostream &sortie, int n_essai, int seuil = CompterMots<It>::SEUIL_DEFAUT)
   {
      stringstream sstr;
      sortie << "Approche parallele, essai " << n_essai <<  ", seuil sequentiel " << seuil << endl;
      int n_map_reduce;
      double ecoule = 0.0;
      {
         minuterie minu{sstr, &ecoule};
         n_map_reduce = compter_mots_multidocuments_map_reduce(debut, fin, seuil);
      }
      sortie << '\t' << n_map_reduce << " mots au total" << endl;
      sortie << '\t' << sstr.str() << endl;
      return ecoule;
   }
template <class It>
   void production_statistiques(const std::string &titre, It debut, It fin, std::ostream &os)
   {
      os << "Version " << titre << ":\n"
         << "*\tmeilleur resultat: " << *min_element(debut, fin) << " ms\n"
         << "*\tpire resultat: " << *max_element(debut, fin) << " ms\n"
         << "*\ttemps moyen: " << moyenne(debut, fin) << " ms\n"
         << "*\tecart type: " << ecart_type(debut, fin) << " ms\n"
         << endl;
   }
int main(int argc, char *argv[])
{
   ofstream sortie{"out.txt"};
   const int NDOCS = argc - 1;
   enum { NTESTS = 10 };
   sortie << "Compter les mots, " << NDOCS
          << " document" << terminaison(argc > 2)
          << ", " << NTESTS << " test" << terminaison(NTESTS > 2)
          << ' ' << config() << "\n\t";
   copy (argv + 1, argv + argc, ostream_iterator<char*>{sortie, "\n\t"});
   sortie << endl;
   enum { NSEUILS = 20 };
   double resultats_sequentiel[NTESTS] = { 0.0 };
   double resultats_parallele[NTESTS] = { 0.0 };
   double resultats_parallele_seuil[NSEUILS][NTESTS] = { {0.0} };
   for (int i = 0; i < NTESTS; ++i)
   {
      resultats_sequentiel[i] = tester_seriel(argv + 1, argv + argc, sortie, i + 1);
      resultats_parallele[i] = tester_parallele(argv + 1, argv + argc, sortie, i + 1);
      for (int j = 0; j < NSEUILS; ++j)
         resultats_parallele_seuil[j][i] = tester_parallele(argv + 1, argv + argc, sortie, i + 1, j + 1);
      sortie << endl;
      cout << '.' << flush;
   }
   cout << endl;
   //
   // Production de statistiques simples
   //
   production_statistiques("sequentielle", resultats_sequentiel, resultats_sequentiel + NTESTS, sortie);
   production_statistiques("parallele", resultats_parallele, resultats_parallele + NTESTS, sortie);
   for (int i = 0; i < NSEUILS; ++i)
      production_statistiques("parallele avec seuil sequentiel " + to_string(i+1),
                              resultats_parallele_seuil[i],
                              resultats_parallele_seuil[i] + NTESTS,
                              sortie);
   double moyennes[NSEUILS];
   for (int i = 0; i < NSEUILS; ++i)
      moyennes[i] = moyenne(begin(resultats_parallele_seuil[i]), end(resultats_parallele_seuil[i]));
   sortie << "\n\nMeilleur temps moyen: ";
   double *p = min_element(begin(moyennes), end(moyennes));
   sortie << *p << " ms. avec un seuil sequentiel de " << static_cast<int>(p - moyennes + 1) << endl;
   thread_pool_::get().log_stats(sortie);
}

Pour comprendre les futures

Tel que mentionné plus haut, une future encapsule la promesse du résultat d'un calcul. Les futures que nous utiliserons ici seront couplées avec l'implémentation du regroupement de threads dont nous discuterons plus bas, mais c'est plus un choix d'implémentation de la part de votre humble serviteur qu'une condition technique ou conceptuelle à proprement dit.

Pour comprendre le regroupement de threads

Le regroupement de threads sera un singleton, mais c'est plus une question de simplicité que de nécessité car il serait possible d'envisager un système à plusieurs regroupements de threads distincts. Une partie importante du code que vous trouverez dans cette section tire des statistiques d'utilisation; le code du regroupement en soi est plus léger qu'il n'y paraît.

Quelques utilitaires légers

Nous utilisons plusieurs petits utilitaires (souvent des foncteurs) dans le code présenté ici.

Le code plus près des délégués

Depuis C++ 14

Le code qui précède est lourd mais fonctionne. Cependant, depuis C++ 14, le tout devient nettement plus simple. Je ne reprendrai pas ici toutes les explications proposées plus haut, me limitant à expliquer les simplifications, disparitions, ajouts et autres ajustements.

Notez que j'ai évacué le volet « statistiques » pour cette version, dans le but de ne pas obscurcir le propos (si vous souhaitez le réintégrer, il est sensiblement comme celui des exemples précédents). J'ai strictement gardé le volet décrivant le temps écoulé en fonction de la charge de travail et des seuils de séparations des tâches.

Voilà : une solution nettement plus simple, plus compréhensible, et probablement plus efficace.

Le Thread Pool lancera plusieurs threads ouvriers (ce qu'on nomme typiquement des Worker Threads), qui prendront en charge des tâches, représentées par la classe Tache visible à droite. Une Tache possèdera un identifiant entier (pour faciliter le débogage). Elle pourra être prise en charge et exécutée (méthode abstraite faire()), testée de manière non bloquante pour vérifier si elle a été complétée, et permettra qu'on se mette en attente bloquante de la fin de son exécution. D'autres services (attente de la complétion jusqu'à concurrence d'un certain délai maximal) pourraient être ajoutés à cette classe si vous le jugez nécessaire.	#ifndef TACHE_H #define TACHE_H #include "Evenement.h" struct Tache { virtual void faire() = 0; virtual ~Tache() = default; Tache() : complete_{}, no_(++g_no_), evenement_{Evenement::ManualReset{}} { } bool complete() const { return complete_; } int no() const { return no_; } void attendre() const { evenement_.attendre(); } private: Evenement evenement_; static int g_no_; int no_; bool complete_; void fini() { complete_ = true; evenement_.provoquer(); } // // ICI: cacher ce détail technique quelque part // friend unsigned long __stdcall ze_thread(void *); }; #endif
Pour des raisons techniques, nous définissons la variable représentant l'identifiant de Tache le plus récemment attribué dans un fichier source. En C++, il est possible de définir les constantes de classe entières dans un fichier d'en-tête, mais les constantes de classe qui ne sont pas entière et les attributs de classe non-constants doivent être définis dans une unité de traduction qui assurera leur unicité (donc, typiquement, dans un fichier source, puisque personne ne les inclura).	`#include "Tache.h" int Tache::g_no_ = 0;`
Les threads ouvriers seront capables de prendre une Tache en charge, de l'exécuter et de s'en débarrasser. Ce sont ces ouvriers qui assureront la dynamique du système. Là où chaque Tache représentera ce qui doit être fait, chaque Worker représentera l'action de faire cette chose. La méthode clé ici est prendre_en_charge(), qui assure la prise en charge d'une tâche par un ouvrier. Elle est décrite plus bas. Notez que, bien qu'un Worker ait pour rôle de manipuler des instances de Tache, l'interface entière de la classe Worker n'a aucunement besoin de connaître le détail de la classe Tache ou la taille des instances de cette classe (dans l'interface de Worker, le seul endroit où le nom Tache est utilisé, il l'est à travers un pointeur). Pour cette raison, seul le nom de la classe Tache est introduit ici (instruction struct Tache;), limitant ainsi le couplage entre les fichiers.	#ifndef WORKER_H #define WORKER_H #include "Incopiable.h" #include "Evenement.h" struct Tache; // // Un Worker traite une Tache* sans se préoccuper de considérations // de synchronisation ou de détails propres au domain d'application // class Worker : Incopiable { // // ICI: cacher cette signature dans un .cpp pour réduire // le couplage, quand j'aurai une minute... // friend unsigned long __stdcall ze_thread(void ); bool actif_; bool demande_arret_; Evenement evenement_; void fin() { actif_ = {}; evenement_.provoquer(); } public: bool demande_arret() const { return demande_arret_; } Worker() : actif_{true}, demande_arret_{}, evenement_{Evenement::ManualReset{}} { } void prendre_en_charge(Tache ); bool actif() const { return actif_; } void stop() { demande_arret_ = true; evenement_.attendre(); } }; #endif
Le code de prise en charge d'une Tache par un ouvrier est dissimulé dans un fichier source pour fins de réduction de couplage. Sans grande surprise, quand un ouvrier prend en charge une Tache, il lui dit de se faire, et c'est tout.	`#include "Worker.h" #include "Tache.h" void Worker::prendre_en_charge(Tache *t) { t->faire(); }`
Pour faciliter la jonction entre les instances de future et les instances de Tache, une classe générique TacheHolder sera utilisée. Tache est une racine polymorphique, ce qui permet de placer des pointeurs sur toutes les tâches dans un seul et même conteneur et d'exprimer le regroupement de threads sur la base de tâches au sens large, mais lorsqu'il vient le temps d'implémenter des tâches spécifiques, nous instancierons un TacheHolder<F> pour chaque tâche de type F. L'abstraction des détails techniques de F passera par un délégué nullaire, ce qui est d'ailleurs un excellent rôle pour des délégués. Cette classe dépose l'opération à faire dans un délégué, et implémente le service Tache::faire() à l'aide d'une invocation du délégué en question. Ce faisant, l'opération encapsulée par une Tache donnée pourra être une fonction nullaire ou un foncteur nullaire. Le TacheHolder déduira le type de donnée retourné par la tâche qu'encapsulera le délégué, et saura retenir une copie de cette donnée dans un attribut d'instance, permettant ainsi le découplage temporel entre le travail fait par la future et les besoins du code client. On trouve des idiomes analogues dans d'autres types de données exigeant à la fois une interface non-générique (pensez à Tache) mais une implémentation générique (pensez à TacheHolder). Pour un exemple, voir la technique de l'effacement de types.	`#ifndef TACHE_HOLDER_H #define TACHE_HOLDER_H #include "Tache.h" #include "functional_ext.h" #include "delegue.h" template <class F> class TacheHolder : public Tache { public: using result_type = typename nullary_function_traits<F>::result_type; private: delegue<result_type> fct_; result_type res_; public: TacheHolder(F fct) : fct_{fct} { } void faire() { res_ = fct_(); } result_type result() const { return res_; } }; #endif`

La comptabilité sérielle du nombre de mots dans un fichier demeurera telle quelle, à ceci près que l'amusant type whatever devient redondant maintenant que nous avons des expressions λ génériques.	#ifndef COMPTER_MOTS_SERIEL #define COMPTER_MOTS_SERIEL #include <algorithm> #include <numeric> #include <string> #include <iterator> #include <fstream> int compter_mots_document(const std::string &nom) { using namespace std; return count_if( istream_iterator<string>{ ifstream{ nom } }, istream_iterator<string>{}, [](auto &&) { return true; } ); } template <class It> int compter_mots_multidocuments_seriel(It debut, It fin) { using namespace std; return accumulate(debut, fin, 0, [](int so_far, const string &nom) { return so_far + compter_mots_document(nom); }); } #endif
Le volet Map/ Reduce est simplifié du fait que nous avons recours à std:future et à std::async().	#ifndef COMPTER_MOTS_MAP_REDUCE_H #define COMPTER_MOTS_MAP_REDUCE_H #include <thread> #include <future> #include <iterator> template <class It> int compter_mots_multidocuments_map_reduce (It debut, It fin, int seuil = std::thread::hardware_concurrency()) { using namespace std; auto n = distance(debut, fin); if (n < seuil) return compter_mots_multidocuments_seriel(debut, fin); auto f_ = async(compter_mots_multidocuments_map_reduce<It>, debut, next(debut, n / 2), seuil), g_ = async(compter_mots_multidocuments_map_reduce<It>, next(debut, n / 2), fin, seuil); return f_.get() + g_.get(); } #endif
J'ai écrit une version simplifiée de l'algorithme moyenne() mais acceptant une fonction de transformation des données cumulées. La raison pour ce changement est que dans cette version, le code de test (plus bas) cumulera le temps non pas sur des double, mais bien sur des high_resolution_clock::duration, ce qui impliquera éventuellement d'aller chercher le nombre de tics cumulés (méthode count()) pour réaliser la division qu'implique ce calcul. Dans les cas où aucune transformation du cumul n'est requise, le foncteur noop sera utilisé en lieu et place de cette transformation.	#ifndef MOYENNE_H #define MOYENNE_H #include <numeric> #include <cassert> #include <iterator> template <class It, class C = typename std::iterator_traits<It>::value_type> auto somme(It debut, It fin, C init = {}) { return std::accumulate(debut, fin, init); } auto noop = [](auto && arg) -> auto && { return std::forward<decltype(arg)>(arg); }; template <class R, class It, class SumT> R moyenne(It debut, It fin, SumT sumT) { using namespace std; assert(debut != fin); return sumT(somme(debut, fin)) / static_cast<R>(distance(debut, fin)); } template <class R, class It> R moyenne(It debut, It fin) { return moyenne<R>(debut, fin, noop); } #endif
Le programme de test suit. Remarquez d'office à quel point l'échafaudage a été simplifié dans cet exemple plus contemporain (vous pouvez comparer avec l'exemple précédent pour faire le constat vous-mêmes). La fonction terminaison(), décorative, peut maintenant être résolue à la compilation, et il en va de même pour la fonction config().	`// ... // en-têtes maison, ci-dessus, omis pour fins de simplicité // ... #include <sstream> #include <iostream> #include <iterator> #include <algorithm> #include <string> #include <chrono> using namespace std; using namespace std::chrono; static constexpr const char terminaison(bool pluriel) { return pluriel ? "s" : ""; } static constexpr const char config() { #ifdef _DEBUG return "(DEBUG)"; #else return "(RELEASE)"; #endif }`
J'ai écrit une fonction générique de test, retournant une paire faire du résultat d'un calcul et du temps écoulé pour l'effectuer. Ceci remplace avantageusement la classe de minuterie des exemples précédents.	`template <class F, class ... Args> auto tester(F f, Args && ... args) { auto avant = high_resolution_clock::now(); auto res = f(std::forward<Args>(args)...); auto apres = high_resolution_clock::now(); return make_pair(res, apres - avant); }`
Les tests sériel et parallèle n'ont que peu changé, outre le fait qu'ils sont plus simples dû au mécanisme de test dont nous venons de discuter. Les expressions λ sont un outil merveilleux pour simplifier l'écriture des programmes.	template <class It> auto tester_seriel(It debut, It fin, ostream &sortie, int n_essai) { sortie << "Approche sérielle, essai " << n_essai << endl; auto res = tester([debut, fin]() { return compter_mots_multidocuments_seriel(debut, fin); }); sortie << '\t' << res.first << " mots au total\n" << '\t' < duration_cast<milliseconds>(res.second).count() << " ms" << endl; return res.second; } template <class It> auto tester_parallele(It debut, It fin, ostream &sortie, int n_essai, int seuil) { sortie << "Approche parallele, essai " << n_essai << ", seuil sequentiel " << seuil << endl; auto res = tester([debut, fin, seuil]() { return compter_mots_multidocuments_map_reduce(debut, fin, seuil); }); sortie << '\t' << res.first << " mots au total\n" << '\t' << duration_cast<milliseconds>(res.second).count() << " ms" << endl; return res.second; }
J'ai modifié le programme en soi sur quelques aspects : Les temps sont mesurés en fonction des outils de chrono, plus rapides et plus rigoureux Les seuils de séparation du travail sont plus pertinents Par défaut, les tests parallèles utilisent un seuil qui utilise tous les coeurs (ici, je triche un peu en faisant semblant que main() n'est pas un thread pour que l'exemple demeure simple) Le calcul de la moyenne est adapté pour tenir compte des changements décrits plus haut	int main(int argc, char argv[]) { auto debut = &argv[1], fin = &argv[argc]; ofstream sortie{ "sortie.txt" }; const int NDOCS = argc - 1; enum { NTESTS = 10 }; sortie << "Compter les mots, " << NDOCS << " document" << terminaison(argc > 2) << ", " << NTESTS << " test" << terminaison(NTESTS > 2) << ' ' << config() << "\n\t"; copy(debut, fin, ostream_iterator<string>{sortie, "\n\t"}); sortie << endl; int seuils[] { 64, 128, 256, 512, 1024 }; high_resolution_clock::duration resultats_sequentiel[NTESTS] = { }, resultats_parallele[NTESTS] = { }, resultats_parallele_seuil[size(seuils)][NTESTS] = { { } }; for (int i = 0; i < NTESTS; ++i) { resultats_sequentiel[i] = tester_seriel(debut, fin, sortie, i + 1); resultats_parallele[i] = tester_parallele( debut, fin, sortie, i + 1, distance(debut, fin) / thread::hardware_concurrency() ); for (size_t j = 0; j != size(seuils); ++j) resultats_parallele_seuil[j][i] = tester_parallele( debut, fin, sortie, i + 1, seuils[j] ); sortie << endl; cout << '.' << flush; } cout << endl; double moyennes[size(seuils)]; for (size_t i = 0; i < size(seuils); ++i) moyennes[i] = moyenne<double>( begin(resultats_parallele_seuil[i]), end(resultats_parallele_seuil[i]), [](high_resolution_clock::duration t) { return static_cast<double>(t.count()); } ); sortie << "\n\nMeilleur temps moyen: "; double p = min_element(begin(moyennes), end(moyennes)); sortie << *p << " ms. avec un seuil sequentiel de " << static_cast<int>(p - moyennes + 1) << endl; }