Thread Pool

Thread Pool - C++ - Programmation

Marsh Posté le 16-07-2012 à 12:42:18    

Bonjour,

 

J'essaie de faire un thread pool pour voir comment ça marche.
http://en.wikipedia.org/wiki/Thread_pool_pattern

 

Est ce que vous pourriez me dire si mon implémentation est correcte ...?

 

merci par avance

 

les tâches font rien de spécial, j'ai juste crée ça :

 
Code :
  1. class Task
  2. {
  3. public:
  4.     void execute() const
  5.     {
  6.         cout << "task does job ... " << endl;
  7.     }
  8. };
 

Pour le thread pool, j'ai crée une classe avec une queue qui contient les taches, et un vecteur pour les n threads :

 
Code :
  1. class ThreadPool
  2. {
  3. public:
  4.     ThreadPool(unsigned nbThreads  = 2) : _nbThreads( nbThreads )
  5.     {
  6.         pthread_mutex_init( & _mtx, 0 );
  7.         pthread_cond_init( & _cond, 0);
  8.     }
  9.    
  10.     void push( const Task & task );
  11.    
  12.     Task pop();
  14.     void initialize();
  16. private:
  17.    
  18.     unsigned _nbThreads;
  20.     pthread_mutex_t _mtx;
  21.     pthread_cond_t _cond;
  23.     queue<Task> _tasks;
  25.     vector<pthread_t> _threads;
  26. };
 


La fonction associée aux threads dépile les taches et execute la tâche courante :

Code :
  1. void * treat( void * data )
  2. {
  3.     ThreadPool * threadPool = (ThreadPool*)data;
  5.     while ( true )
  6.     {
  7.         Task t = threadPool->pop();
  9.         t.execute();
  10.     }
  12.     return (void*)0;
  13. }
 


voici le code complet :

 
Code :
  1. #include "pthread.h"
  3. #include <iostream>
  4. #include <queue>
  5. #include <vector>
  6. using namespace std;
  8. void * treat( void * data );
  10. class Task
  11. {
  12. public:
  13.     void execute() const
  14.     {
  15.         cout << "task does job ... " << endl;
  16.     }
  17.     const std::string operator << (int)
  18.     {
  19.         char tmp[16];
  20.         sprintf(tmp,"%s",this);
  21.         return string(tmp);
  22.     }
  23. };
  25. class ThreadPool
  26. {
  27. public:
  28.     ThreadPool(unsigned nbThreads  = 2) : _nbThreads( nbThreads )
  29.     {
  30.         pthread_mutex_init( & _mtx, 0 );
  31.         pthread_cond_init( & _cond, 0);
  32.     }
  33.    
  34.     void push( const Task & task );
  35.    
  36.     Task pop();
  38.     void initialize();
  40. private:
  41.    
  42.     unsigned _nbThreads;
  44.     pthread_mutex_t _mtx;
  45.     pthread_cond_t _cond;
  47.     queue<Task> _tasks;
  49.     vector<pthread_t> _threads;
  50. };
  52. void ThreadPool::push( const Task & task )
  53. {
  54.     pthread_mutex_lock( & _mtx );
  56.     cout << "push new data on queue" << endl;
  58.     _tasks.push( task );
  60.     pthread_cond_signal( & _cond );
  62.     pthread_mutex_unlock( & _mtx );
  63. }
  65. Task ThreadPool::pop()
  66. {
  67.     pthread_mutex_lock( & _mtx );
  69.     cout << "ThreadPool::pop(), tid:" << *(int*)&pthread_self() << endl;
  71.     while ( _tasks.empty() )
  72.         pthread_cond_wait( & _cond , & _mtx );
  74.     Task t = _tasks.front();
  75.     _tasks.pop();
  77.     pthread_mutex_unlock( & _mtx );
  78.    
  79.     return t;
  80. }
  82. void ThreadPool::initialize()
  83. {
  84.     cout << "initialize " << _nbThreads << " threads" << endl;
  86.     for ( int i = 0 ; i < _nbThreads ; ++i )
  87.     {
  88.         pthread_t thread;
  89.         _threads.push_back( thread );
  90.     }
  92.     vector<pthread_t>::iterator it = _threads.begin();
  94.     for ( ; it != _threads.end() ; ++ it)
  95.     {
  96.         pthread_create( &(*it), 0 , treat , this );
  97.     }
  98. }
  100. void * treat( void * data )
  101. {
  102.     ThreadPool * threadPool = (ThreadPool*)data;
  104.     while ( true )
  105.     {
  106.         Task t = threadPool->pop();
  108.         t.execute();
  109.     }
  111.     return (void*)0;
  112. }
  115. int main()
  116. {
  117.      ThreadPool tp(2);
  119.      tp.initialize();
  121.      vector<Task> someTasks(10);
  122.      vector<Task>::iterator it = someTasks.begin();
  123.      for ( ; it != someTasks.end() ; ++ it )
  124.      {
  125.          tp.push( *it );
  126.      }
  128.      cin.get();
  129.      return 0;
  130. }
 


ça vous parait bien ou y a t-il une erreur de design ou de conception ?

 

merki


Message édité par in_your_phion le 16-07-2012 à 14:59:31
Reply

Marsh Posté le 16-07-2012 à 12:42:18   

Reply

Marsh Posté le 16-07-2012 à 17:07:03    

Tes locks devraient être faits en RAII, tu verras, ca allège pas mal le code dès que tu as des fonctions un peu complexes.
 
Je ne suis pas fan du while(true) dans la fonction treat. Globalement, killer des threads, c'est pas très joli, et avec la manière dont c'est architecturé, là, t'as pas trop le choix.
 

Message cité 1 fois

---------------
last.fm
Reply

Marsh Posté le 16-07-2012 à 19:08:35    

theshockwave a écrit :

Tes locks devraient être faits en RAII, tu verras, ca allège pas mal le code dès que tu as des fonctions un peu complexes.

 

Je ne suis pas fan du while(true) dans la fonction treat. Globalement, killer des threads, c'est pas très joli, et avec la manière dont c'est architecturé, là, t'as pas trop le choix.

 


 

hello !
merci pour tes réponses, pour les locks/unlocks en fait j'ai eu un peu la flemme de les faire en RAII  :jap:

 

sinon, comment je pourrais architecturer différement pour ne pas avoir à killer mes threads ...? il y a une technique pour ça ?

 

autrement, comment on choisis le nombre de threads ? J'ai essayé avec plusieurs valeurs, même jusqu'à 512 lol. ça doit être combien généralement ..?

 

merci  ;)

 

ps : par ailleurs, pourquoi les "cout" s'affichent souvent un peu n'importe comment, alors que pas les printf ? (pour les mêmes messages)

Message cité 1 fois
Message édité par in_your_phion le 16-07-2012 à 19:10:33
Reply

Marsh Posté le 16-07-2012 à 19:20:38    

Ca dépend complètement de tes besoins.
Si tu vois ton système comme un job manager (Edit : micro tâches pour faire une distributions en découpant au max tes lourdes tâches), tu vas probablement vouloir avoir autant de threads à dispositions que tu as de threads hardware sur ta machine.
Si tu vois ca comme juste la possibilité de faire du "Fire and Forget" sur tes tâches (Edit : tâches lourdes) pour un max de client, peut-être que tu en voudras plus quitte à ralentir ta machine juste pour que le boulot pour chaque client commence au plus tôt ...
 
Dans le premier cas, tu voudras aussi probablement une interface qui te permette d'ajouter N tâches d'un coup. Dans l'autre, probablement pas.
 
Edit : les cas que je liste ne sont pas nécessairement les seuls que tu rencontreras, évidemment.

Message cité 1 fois
Message édité par theshockwave le 16-07-2012 à 19:22:19

---------------
last.fm
Reply

Marsh Posté le 17-07-2012 à 10:28:31    

theshockwave a écrit :

Ca dépend complètement de tes besoins.
Si tu vois ton système comme un job manager (Edit : micro tâches pour faire une distributions en découpant au max tes lourdes tâches), tu vas probablement vouloir avoir autant de threads à dispositions que tu as de threads hardware sur ta machine.
Si tu vois ca comme juste la possibilité de faire du "Fire and Forget" sur tes tâches (Edit : tâches lourdes) pour un max de client, peut-être que tu en voudras plus quitte à ralentir ta machine juste pour que le boulot pour chaque client commence au plus tôt ...
 
Dans le premier cas, tu voudras aussi probablement une interface qui te permette d'ajouter N tâches d'un coup. Dans l'autre, probablement pas.
 
Edit : les cas que je liste ne sont pas nécessairement les seuls que tu rencontreras, évidemment.


 
hello !
ok merci pour tes réponses. Quand tu parles de "threads hardware", ce sont en fait les coeur du CPU non ?
sinon, à partir de quand une tache deviens "lourde" ?  [:tinostar]

Message cité 1 fois
Reply

Marsh Posté le 17-07-2012 à 11:20:48    

in_your_phion a écrit :


 
hello !
ok merci pour tes réponses. Quand tu parles de "threads hardware", ce sont en fait les coeur du CPU non ?
sinon, à partir de quand une tache deviens "lourde" ?  [:tinostar]


 
 
Threads hardware, c'est pas tout à fait coeur, mais presque. Sur du intel, avec hyperthreading, tu as deux threads hardware par coeur.
Encore une fois, la "lourdeur" de la tâche dépend du contexte. Suivant les contraintes que tu as en terme de temps (obligation de terminer tes tâches dans un délais imparti) ou d'architecture (genre rentabiliser les SPUs d'un cell qui n'ont pas d'accès direct à la mémoire) ta notion de "lourdeur" va varier.
 
J'ai tendance à penser qu'un système de jobs "légers", ca va être quand tu as besoin que tous tes jobs aient terminé pour pouvoir en lancer de nouveaux (à peu de choses près) alors que le système de tâches "lourdes", ca va plus être du fire and forget ... Ca termine quand ca peut et tu en tiens compte à ce moment là.
 
Exemple : dans un jeu vidéo, tu vas sans doute avoir une phase de mise à jour des positions de tes objets que tu vas pouvoir paralléliser, et tu vas avoir une attente forte dessus pour pouvoir afficher le résultat à l'écran. Là, clairement, on est dans un contexte de tâches "légères" (c'est critique si ca prend plus de 2 ou 3 millisecondes). Par contre, une requête d'IA, si c'est pas immédiat, c'est peut-être pas si grave, du coup c'est "lourd" (quelques dizaines de millisecondes)
 
Dans le cas des tâches "légères", tu veux clairement faire correspondre tes threads disponibles à tes ressources matérielles parce que c'est critique.
Dans le cas des tâches "lourdes", tu veux juste avoir un pool de threads sous le coude pour ne pas payer le coût de la création d'un thread à chaque lancement de tâche.

Message cité 1 fois

---------------
last.fm
Reply

Marsh Posté le 17-07-2012 à 11:32:44    

Autre détail : tu as des mutex pour protéger ta file de tâches. Normalement, tu dois pouvoir faire une "lock-free queue". En terme de performance, ca peut changer la vie.
Si tu vires ce mutex, la condition ne sera plus utilisable, tu n'auras plus besoin que d'un système de signal basique et tes threads attendront ce signal quand la liste de tâches est vide. Ton manager pourra réveiller tous les threads ou juste ceux qu'ils choisi en fonction de ce qui arrive à traiter.

Message cité 1 fois

---------------
last.fm
Reply

Marsh Posté le 17-07-2012 à 12:07:27    

theshockwave a écrit :

 


Threads hardware, c'est pas tout à fait coeur, mais presque. Sur du intel, avec hyperthreading, tu as deux threads hardware par coeur.
Encore une fois, la "lourdeur" de la tâche dépend du contexte. Suivant les contraintes que tu as en terme de temps (obligation de terminer tes tâches dans un délais imparti) ou d'architecture (genre rentabiliser les SPUs d'un cell qui n'ont pas d'accès direct à la mémoire) ta notion de "lourdeur" va varier.

 

J'ai tendance à penser qu'un système de jobs "légers", ca va être quand tu as besoin que tous tes jobs aient terminé pour pouvoir en lancer de nouveaux (à peu de choses près) alors que le système de tâches "lourdes", ca va plus être du fire and forget ... Ca termine quand ca peut et tu en tiens compte à ce moment là.

 

Exemple : dans un jeu vidéo, tu vas sans doute avoir une phase de mise à jour des positions de tes objets que tu vas pouvoir paralléliser, et tu vas avoir une attente forte dessus pour pouvoir afficher le résultat à l'écran. Là, clairement, on est dans un contexte de tâches "légères" (c'est critique si ca prend plus de 2 ou 3 millisecondes). Par contre, une requête d'IA, si c'est pas immédiat, c'est peut-être pas si grave, du coup c'est "lourd" (quelques dizaines de millisecondes)

 

Dans le cas des tâches "légères", tu veux clairement faire correspondre tes threads disponibles à tes ressources matérielles parce que c'est critique.
Dans le cas des tâches "lourdes", tu veux juste avoir un pool de threads sous le coude pour ne pas payer le coût de la création d'un thread à chaque lancement de tâche.

 


merci bien pour toutes ces explications ! pourtant il m'avait semblé lire que le thread pool devait être utilisé pour des tâches qui ne durent pas très longtemps, mais peut être que c'est relatif ?

 

Par exemple, si tu as une matrice géante et que tu veux paralléliser les calculs (un thread par calcul de produit scalaire ligne x colonne), est ce qu'un thread pool est approprié ?

 
theshockwave a écrit :

Autre détail : tu as des mutex pour protéger ta file de tâches. Normalement, tu dois pouvoir faire une "lock-free queue". En terme de performance, ca peut changer la vie.
Si tu vires ce mutex, la condition ne sera plus utilisable, tu n'auras plus besoin que d'un système de signal basique et tes threads attendront ce signal quand la liste de tâches est vide. Ton manager pourra réveiller tous les threads ou juste ceux qu'ils choisi en fonction de ce qui arrive à traiter.

 

merci je vais essayer ça  :jap: dans ce cas le signal est juste une variable partagée par les threads ? (genre un booléen ?)

Message cité 1 fois
Message édité par in_your_phion le 17-07-2012 à 12:08:05
Reply

Marsh Posté le 17-07-2012 à 12:38:39    

in_your_phion a écrit :


 
 
merci bien pour toutes ces explications ! pourtant il m'avait semblé lire que le thread pool devait être utilisé pour des tâches qui ne durent pas très longtemps, mais peut être que c'est relatif ?  
 
Par exemple, si tu as une matrice géante et que tu veux paralléliser les calculs (un thread par calcul de produit scalaire ligne x colonne), est ce qu'un thread pool est approprié ?
 


 
houlà, c'est toujours des threads pools, ce sont juste deux cas de figures différents qui font que tu vas pas forcément programmer ton pool de la même manière. Quoiqu'il arrive, oui, quand tu veux paralléliser c'est une bonne idée d'avoir un pool de threads, quelles que soient tes contraintes. Ton exemple de multiplication de matrices me fait plus penser à un process léger, justement, où tu vas distribuer ton calcul de manière à ce que tous tes threads hardwares tournent à 100% mais pas plus. En gros, sur ce genre de tâches, on cherche aussi à minimiser les changements de tâches des CPUs. Passer d'un contexte de thread à un autre n'est pas gratuit pour le matériel. Quand tu compares ca à une série de multiplications et additions même en flotttants, ca reste conséquent.
 
 

in_your_phion a écrit :

merci je vais essayer ça  :jap: dans ce cas le signal est juste une variable partagée par les threads ? (genre un booléen ?)


Dans l'idée, c'est un booléen, oui, mais tu dois avoir un système de signal dans les pthreads, j'imagine, qui te permettra de le faire correctement. (attention à ne pas utiliser juste un signal "instantané" que tes threads risqueraient de rater mais bien quelque chose qui place un flag jusqu'à ce que ce soit resetté soit manuellement, soit implicitement par une lecture)
 


---------------
last.fm
Reply

Marsh Posté le 17-07-2012 à 20:43:41    

in_your_phion a écrit :


 
hello !
merci pour tes réponses, pour les locks/unlocks en fait j'ai eu un peu la flemme de les faire en RAII  :jap:  
 
sinon, comment je pourrais architecturer différement pour ne pas avoir à killer mes threads ...? il y a une technique pour ça ?
 
autrement, comment on choisis le nombre de threads ? J'ai essayé avec plusieurs valeurs, même jusqu'à 512 lol. ça doit être combien généralement ..?
 
merci  ;)  
 
ps : par ailleurs, pourquoi les "cout" s'affichent souvent un peu n'importe comment, alors que pas les printf ? (pour les mêmes messages)


 
 
Ce que je fais en général, c'est une boucle de travail dans mon thread avec une variable de controle , et je suspend mon thread pendant un temps donné s'il n'a rien à faire.  
 
Pour le détruire, il suffit de changer la valeur de la variable de controle, de resumer le thread et de le join.
 

Code :
  1. void * ma_fonction_thread( void * param )
  2. {
  3. while ( !sortie )
  4. {
  5.   // faire le travail s'il y a
  7.   // puis suspendre le thread un temps donné
  9.    struct timespec timeout; // mettre le temps voulu
  11.    pthread_mutex_lock(&m_hMutex);
  12.    pthread_cond_timedwait(&m_hEvent,&m_hMutex, &timeout);
  13.    pthread_mutex_unlock(&m_hMutex);
  16. }
  18. return NULL;
  19. }
  22. // la fonction qui resume  
  24. void resume( )
  25. {
  26. pthread_mutex_lock(&m_hMutex);
  27. pthread_cond_signal(&m_hEvent);
  28. pthread_mutex_unlock(&m_hMutex);
  29. }
  31. // il faut en plus initialiser de cette facon la le thread
  33. void init()
  34. {
  35. pthread_cond_init(&m_hEvent,NULL);
  36. pthread_mutex_init(&m_hMutex,NULL);
  37. }
  39. // et detruire
  41. void shut()
  42. {
  43. pthread_cond_destroy(&m_hEvent);
  44. pthread_mutex_destroy(&m_hMutex);
  45. }
  47. // pour terminer proprement le thread  
  49. void terminate()
  50. {
  51. // on change la valeur de la variable de controle
  53. sortie = true;
  55. // on resume pour eviter d'attendre un tour de boucle
  56. resume();
  58. // et on join
  60. pthread_join(m_hThread, NULL);
  61. }


 
En espérant que ca t'aide. (j 'ai juste fait du copier coller de code pour expliquer le principe, il faut bien sur tester les valeurs retournées des appels système )

Message cité 2 fois
Reply

Marsh Posté le 17-07-2012 à 20:43:41   

Reply

Marsh Posté le 18-07-2012 à 10:20:20    

xilebo a écrit :

Ce que je fais en général, c'est une boucle de travail dans mon thread avec une variable de controle , et je suspend mon thread pendant un temps donné s'il n'a rien à faire.  
 
Pour le détruire, il suffit de changer la valeur de la variable de controle, de resumer le thread et de le join.


 
 :jap:


---------------
last.fm
Reply

Marsh Posté le 18-07-2012 à 11:03:09    

xilebo a écrit :

En espérant que ca t'aide. (j 'ai juste fait du copier coller de code pour expliquer le principe, il faut bien sur tester les valeurs retournées des appels système )

 

salut,
merci beaucoup pour ton code :) je vois mieux là ^^

 

en fait j'avais essayé avec ça mais ça ne marchais pas terrible pour mon code, car j'avais mis un wait sans timeout, mais avec un timeout ça devrait plus le faire

 

ps : j'aurais encore une autre question, est ce que faire une boucle avec des try_lock() peut être vu comme un spin lock ?

 

merci

Message cité 1 fois
Message édité par in_your_phion le 18-07-2012 à 11:03:27
Reply

Marsh Posté le 18-07-2012 à 11:09:35    

in_your_phion a écrit :


 
salut,
merci beaucoup pour ton code :) je vois mieux là ^^  
 
en fait j'avais essayé avec ça mais ça ne marchais pas terrible pour mon code, car j'avais mis un wait sans timeout, mais avec un timeout ça devrait plus le faire
 
ps : j'aurais encore une autre question, est ce que faire une boucle avec des try_lock() peut être vu comme un spin lock ?
 
merci


 
Pas besoin de timeout, il faut juste signaler ta condition. Du point de vue du threads, tu dois traiter les deux cas, soit il y a effectivement quelque chose à consommer, soit il n'y a rien et c'est peut-être qu'il faut quitter (consulte le flag correspondant)


---------------
last.fm
Reply

Sujets relatifs:

Leave a Replay

Make sure you enter the(*)required information where indicate.HTML code is not allowed