Discussion :

[Nekkro]

Bonjour,

Je rencontre quelques problèmes avec une threadpool codée moi-même.
Je n'utilise pas de threadpool déjà existante car j'ai besoin d'un système de Task un peu spécial.

Ma threadpool gère mes classes Executor et ConnectionListener.
Mon executor gère les threads workers et les connectionListener gèrent les connexions au programme (deux types de connexions dont 2 listeners).

Allons-y pour le code :

Thread.hpp

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
class Thread
{
	public:
		Thread () : _thread (NULL) {}
		virtual ~Thread () {}
		virtual void run () = 0;
		void terminate () { if (_thread != NULL) _thread->interrupt(); }
		void set_thread (boost::thread * t) { _thread = t; }
 
	private:
		boost::thread * _thread;
};

ThreadPool.hpp

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class ThreadPoolManager
{
	public:
		void start_up ();
		void shutdown ();
		bool running () const { return is_running; }
		void schedule (Task * t);
		void launch (Thread * t, bool delete_t);
		unsigned garbage_collector ();
 
	private:
		ThreadPoolManager ();
		~ThreadPoolManager () { }
 
		bool is_running;
		set< pair<Thread *, bool> > _threads;
		boost::shared_ptr< boost::asio::io_service > _io_service;
		boost::shared_ptr< boost::asio::io_service::work > _work;
		ThreadPoolExecutor _executor;
};

ThreadPool.cpp

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
ThreadPoolManager::ThreadPoolManager () :
_io_service (new boost::asio::io_service),
_work (new boost::asio::io_service::work (*_io_service)),
_executor (_io_service)
{
	launch (&_executor, false);
}
 
void ThreadPoolManager::start_up ()
{
	if (!is_running)
		is_running = true;
}
 
void ThreadPoolManager::shutdown ()
{
	if (!is_running)
		return;
	is_running = false;
 
	for (set<pair<Thread *, bool> >::iterator it = _threads.begin() ; it != _threads.end() ; ++it)
	{
		delete it->first->_thread;
		if (it->second) // si on doit delete le Thread *
			delete it->first;
	}
	_threads.clear();
}
 
void ThreadPoolManager::schedule (Task * t)
{
	if (!is_running)
		return;
 
	_io_service->post (boost::bind (&Task::execute, t, t->_target, t->_data)); // Là j'ai un problème
}
 
void ThreadPoolManager::launch (Thread * t, bool delete_t)
{
	if (!is_running)
		return;
 
	boost::thread * new_t = new boost::thread (boost::bind (&Thread::run, t));
	t->set_thread (new_t);
	_threads.insert (make_pair (t, delete_t));
}
 
unsigned ThreadPoolManager::garbage_collector ()
{
	list<set<pair<Thread *, bool> >::iterator> to_erase;
	boost::posix_time::time_duration td = boost::posix_time::milliseconds(1);
	for (set<pair<Thread *, bool> >::iterator it = _threads.begin() ; it != _threads.end() ; ++it)
	{
		if (it->first->_thread->timed_join (td))
		{
			delete it->first->_thread;
			if (it->second)
				delete it->first;
			to_erase.push_back (it); // si on faisait direct un erase, on aurait le droit à un beau plantage "invalid iterator etc" au prochain tour de boucle
		}
	}
	unsigned count = to_erase.size();
	for (list<set<pair<Thread *, bool> >::iterator>::iterator it = to_erase.begin() ; it != to_erase.end() ; ++it)
		_threads.erase (*it);
	return count;
}

Executor.hpp

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
namespace
{
	void worker_thread (boost::shared_ptr< boost::asio::io_service > io_service)
	{
		while (true)
		{
			io_service->run();
		}
	}
}
 
 
class ThreadPoolExecutor : public Thread
{
	public:
		ThreadPoolExecutor (boost::shared_ptr< boost::asio::io_service > io_service)
		{
			for (int x = 0 ; x < 2 ; ++x) // Pour le moment, seulement 2 workers
				_workers.create_thread (boost::bind (&worker_thread, io_service));
		}
		~ThreadPoolExecutor () { _workers.interrupt_all(); }
		void run ()
		{
			while (true)
			{
				_workers.join_all(); // Quel est réellement l'impact de cette instruction ?
				boost::this_thread::sleep(boost::posix_time::milliseconds (50));
			}
		}
 
	private:
		boost::thread_group _workers;
};

Task.hpp

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
enum RequiredData { /* ... */ };
 
struct Task
{
		Task () : _target (NULL), _flag (NONE) {}
 
		void (*execute) (Session *, string);
		Session * _target;
		string _data;
		RequiredData _flag;
};

Pour le moment, deux problèmes se posent à moi :
- Mon bind n'est pas accepté dans ThreadPool::schedule (), j'ai les erreurs suivantes :

In instantiation of 'boost::_bi::result_traits<boost::_bi::unspecified, void (* Task::*)(Session*, std::string)>':
instantiated from 'boost::_bi::bind_t<boost::_bi::unspecified, void (* Task::*)(Session*, std::string), boost::_bi::list3<boost::_bi::value<Task*>, boost::_bi::value<Session*>, boost::_bi::value<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > >'
error: 'void (* Task::*)(Session*, std::string)' is not a class, struct, or union type

In member function 'void boost::asio::io_service::post(const CompletionHandler&) [with CompletionHandler = boost::_bi::bind_t<boost::_bi::unspecified, void (* Task::*)(Session*, std::string), boost::_bi::list3<boost::_bi::value<Task*>, boost::_bi::value<Session*>, boost::_bi::value<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > >]':
error: no match for call to '(boost::_bi::bind_t<boost::_bi::unspecified, void (* Task::*)(Session*, std::string), boost::_bi::list3<boost::_bi::value<Task*>, boost::_bi::value<Session*>, boost::_bi::value<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > >) ()'

In function 'void boost::asio::asio_handler_invoke(Function, ...) [with Function = boost::_bi::bind_t<boost::_bi::unspecified, void (* Task::*)(Session*, std::string), boost::_bi::list3<boost::_bi::value<Task*>, boost::_bi::value<Session*>, boost::_bi::value<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > >]':

C:\boost\boost_1_47_0/boost/asio/detail/handler_invoke_helpers.hpp:39: instantiated from 'void boost_asio_handler_invoke_helpers::invoke(Function&, Context&) [with Function = boost::_bi::bind_t<boost::_bi::unspecified, void (* Task::*)(Session*, std::string), boost::_bi::list3<boost::_bi::value<Task*>, boost::_bi::value<Session*>, boost::_bi::value<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > >, Context = boost::_bi::bind_t<boost::_bi::unspecified, void (* Task::*)(Session*, std::string), boost::_bi::list3<boost::_bi::value<Task*>, boost::_bi::value<Session*>, boost::_bi::value<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > >]'

C:\boost\boost_1_47_0/boost/asio/detail/completion_handler.hpp:66: instantiated from 'static void boost::asio::detail::completion_handler<Handler>::do_complete(boost::asio::detail::io_service_impl*, boost::asio::detail::operation*, boost::system::error_code, size_t) [with Handler = boost::_bi::bind_t<boost::_bi::unspecified, void (* Task::*)(Session*, std::string), boost::_bi::list3<boost::_bi::value<Task*>, boost::_bi::value<Session*>, boost::_bi::value<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > >]'

C:\boost\boost_1_47_0/boost/asio/detail/completion_handler.hpp:38: instantiated from 'boost::asio::detail::completion_handler<Handler>::completion_handler(Handler&) [with Handler = boost::_bi::bind_t<boost::_bi::unspecified, void (* Task::*)(Session*, std::string), boost::_bi::list3<boost::_bi::value<Task*>, boost::_bi::value<Session*>, boost::_bi::value<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > >]'

C:\boost\boost_1_47_0/boost/asio/detail/impl/win_iocp_io_service.hpp:66: instantiated from 'void boost::asio::detail::win_iocp_io_service::post(Handler) [with Handler = boost::_bi::bind_t<boost::_bi::unspecified, void (* Task::*)(Session*, std::string), boost::_bi::list3<boost::_bi::value<Task*>, boost::_bi::value<Session*>, boost::_bi::value<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > >]'

C:\boost\boost_1_47_0/boost/asio/impl/io_service.hpp:90: instantiated from 'void boost::asio::io_service::post(const CompletionHandler&) [with CompletionHandler = boost::_bi::bind_t<boost::_bi::unspecified, void (* Task::*)(Session*, std::string), boost::_bi::list3<boost::_bi::value<Task*>, boost::_bi::value<Session*>, boost::_bi::value<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > >]'

C:\boost\boost_1_47_0/boost/asio/handler_invoke_hook.hpp:64: error: no match for call to '(boost::_bi::bind_t<boost::_bi::unspecified, void (* Task::*)(Session*, std::string), boost::_bi::list3<boost::_bi::value<Task*>, boost::_bi::value<Session*>, boost::_bi::value<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > >) ()'

C'est assez imbuvable et surtout incompréhensible ...

Mon deuxième problème est que j'ai une fuite mémoire après destruction de tout ce système ... Mais ça je verrais quand ça marchera correctement.

Pour le boost::bind, j'ai pourtant effectué pas mal de recherche pour avoir la bonne syntaxe, mais apparemment il ne reconnait pas mon pointeur sur fonction membre ...

Devrais-je utiliser boost::function ? Ou changer ma méthode ?

[Arzar]

Bonjour,

Tu fais peut être une petite confusion sur cette ligne :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
struct Task
{
//..
   void (*execute) (Session *, string);
};

Dans ce cas, la structure Task possède un membre de type "pointeur de fonction prenant en argument un Session* et un string"

C'est à dire que c'est un code beaucoup plus proche de:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
struct Task
{
//..
   int* execute;
};

que de :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
struct Task
{
//..
  void execute(Session*, string);
};

C'est pour ça que la ligne _io_service->post (boost::bind (&Task::execute, t, t->_target, t->_data)); ne marche pas, car dans ce cas tu essayes de binder une hypothétique fonction membre "execute" appartenant à Task, fonction membre qui n'existe pas.

Si tu veux binder un pointeur de fonction avec des arguments alors tu peux le faire directement avec la syntaxe std::bind(monPointeurDeFonction, arg1, arg2,...);.


Un exemple complet montrant la différence :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
#include <string>
#include <iostream>
#include <functional>
 
struct Session{};
 
void fonction_libre(Session* s, std::string str){std::cout << "fonction libre\n";}
 
struct Task
{
   void fonction_membre(Session* s , std::string str ){std::cout << "fonction membre\n";}
   void (*execute) (Session* , std::string);
};
 
 
void func(std::function<void(void)> f)
{
   f();
}
 
int main()
{
   Task t;
   t.execute = &fonction_libre;
   Session s;
   std::string str;
   func(std::bind(&Task::fonction_membre, t, &s, str));
   func(std::bind(t.execute, &s, str));
}

[Nekkro]

C'était effectivement ça ... Alors, toutes ces lignes ... pour ça ? Impressionnant.
Je te remercie !

Premier problème résolu !

Quelques questions :

- Lorsque je détruis la threadpool, je passe par la méthode shutdown(), qui va détruire tous les Thread * et donc leurs boost::thread * associés. En détruisant mon Executor, je dois également détruire le thread_group, mais quelle est la façon la plus propre pour ce faire ? Attendre que les workers aient terminé leur appel à io_service::run() ? Y a-t-il un destructeur particulier à appeler pour faire ça sans fuite mémoire ? (Actuellement je fais un simple interrupt_all()).

- Est-ce que la boucle de Executor::run() est vraiment utile ? J'ai cru comprendre qu'il n'y avait pas forcément nécessité de join() un thread si on lui laisse le temps de s'exécuter. J'ai faux ?

[Nekkro]

J'ai finalement changé un grosse partie de la threadpool :
- Plus d'Executor
- Les blocages que je rencontrais étaient liés au fait que lorsque je join_all(), je bloquais complètement le système puisque il n'y avait à ce moment là rien que io_service::run() puisse faire ...

J'ai simplement un problème de comportement bizarre à la destruction de la ThreadPool :

ThreadPoolManager.cpp

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
namespace
{
	void worker_thread (boost::shared_ptr< boost::asio::io_service > io_service)
	{
		{
			ostringstream oss;
			oss << boost::this_thread::get_id();
			sLog->print_header (TYELLOW, true, "[TPoolManager]", "Worker <%s> started.\n", oss.str().c_str());
		}
		while (sThreadPool->running())
		{
			io_service->run();
			sLog->print_header (TYELLOW, true, "[TPoolManager]", "Task complete.\n");
		}
	}
}
 
ThreadPoolManager::ThreadPoolManager () :
is_running (true),
_io_service (new boost::asio::io_service),
_work (new boost::asio::io_service::work (*_io_service))
{
	sLog->print (TBLUE, true, "Starting up the Thread Pool ...");
	{
		ProgressBar bar (1);
		bar.step();
	}
	for (int x = 0 ; x < 2 ; ++x)
		_workers.create_thread (boost::bind (&worker_thread, _io_service));
}
 
ThreadPoolManager::~ThreadPoolManager ()
{
	is_running = false;
	_workers.interrupt_all();
	shutdown();
	sLog->print_header (TYELLOW, true, "[TPoolManager]", "Workers destructed.\n");
	_work.reset();
	_io_service.reset();
	cout << "Fin du TPM destructeur atteint" << endl;
}
 
void ThreadPoolManager::start_up ()
{
}
 
void ThreadPoolManager::shutdown ()
{
	if (!_threads.empty())
	{
		for (set<pair<Thread *, bool> >::iterator it = _threads.begin() ; it != _threads.end() ; ++it)
		{
			sLog->print_header (TYELLOW, true, "[TPoolManager]", "Deletion of thread <%s>\n", typeid(it->first).name());
			delete it->first->_thread;
			if (it->second) // si on doit delete le ThreadContext*
				delete it->first;
		}
		_threads.clear();
	}
}
 
void ThreadPoolManager::schedule (Task * t)
{
	if (!is_running)
		return;
 
	_io_service->post (boost::bind (t->execute, t->_target, t->_data));
}
 
void ThreadPoolManager::launch (Thread * t, bool delete_t)
{
	if (!is_running)
		return;
 
	sLog->print_header (TYELLOW, true, "[TPoolManager]", "Launched Thread <%s> !\n", typeid(t).name());
	boost::thread * new_t = new boost::thread (boost::bind (&Thread::run, t));
	t->set_thread (new_t);
	_threads.insert (make_pair (t, delete_t));
}
 
unsigned ThreadPoolManager::garbage_collector ()
{
	list<set<pair<Thread *, bool> >::iterator> to_erase;
	for (set<pair<Thread *, bool> >::iterator it = _threads.begin() ; it != _threads.end() ; ++it)
	{
		boost::posix_time::time_duration td = boost::posix_time::milliseconds(1);
		if (it->first->_thread->timed_join (td))
		{
			sLog->print_header (TYELLOW, true, "[TPoolManager]", "Deletion of dead thread <%s>\n", typeid(it->first).name());
			delete it->first->_thread;
			if (it->second)
				delete it->first; // idem
			to_erase.push_back (it); // si on faisait direct un erase, on aurait le droit à un beau plantage "invalid iterator etc" au prochain tour de boucle
		}
	}
	unsigned count = to_erase.size();
	for (list<set<pair<Thread *, bool> >::iterator>::iterator it = to_erase.begin() ; it != to_erase.end() ; ++it)
		_threads.erase (*it);
	return count;
}

Lorsque je fais _workers.interrupt_all(), mes workers se mettent à tourner en rond et m'affiche en boucle et à toute vitesse "Task complete.\n" ...

interrupt_all() n'est-elle pas censé couper complètement les threads du thread_groups ?

Vu que le thread_group est un membre de ma classe, ça veut dire que je dois attendre la fin du destructeur de ma threadpool pour qu'il soit détruit, malheureusement vu que je fais un appel a io_service::reset() avant afin de libérer la mémoire, cela provoque un comportement inattendu ...

De quelle façon dois-je détruire mes workers pour que ce soit fait proprement svp ?

EDIT : Je pense avoir trouvé ce qu'il me faut : io_service::stop();

Ça coupe effectivement tout appel à io_service::run().

Malheureusement pour moi un autre bogue vicieux est apparu

A la fin de la destruction de ma threadpool, le constructeur est à nouveau appelé, 2 fois de suite, et je ne sais ABSOLUMENT pas par qui ...

A la fin du programme, je détruit l'instance singleton, ce qui appelle le destructeur, effectue bien ce que je lui ai demandé, puis le constructeur est relancé de je-ne-sais-où et me fais des trucs moches, en plus de me relancer des workers, -.-'

Voyez vous-même :


Très franchement c'est au-delà de ma limite de compréhension, d'autant que mon constructeur est private ... J'aurais vraiment besoin d'un avis svp.

[Arzar]

J'ai finalement changé un grosse partie de la threadpool

Damn, j'avais noté des commentaires sur l'ancien code et j'ai vraiment la flemme de tout relire et d'adapter. Donc voilà en espérant que ce soient toujours pertinent...

- Lorsque je détruis la threadpool, je passe par la méthode shutdown(), qui va détruire tous les Thread * et donc leurs boost::thread * associés. En détruisant mon Executor, je dois également détruire le thread_group, mais quelle est la façon la plus propre pour ce faire ? Attendre que les workers aient terminé leur appel à io_service::run() ? Y a-t-il un destructeur particulier à appeler pour faire ça sans fuite mémoire ? (Actuellement je fais un simple interrupt_all()).

Et c'est une drole d'idée. Tu devrais vérifier le comportement d'interrupt dans la doc car il est quand même assez spécial. En particulier, dans la plupart des cas ça ne termine pas du tout le thread, et dans les cas ou ça le termine quand même une exception est lancée en prime !
Pour clore proprement un thread il faut appeler join() qui va bloquer jusqu'à ce que l'exécution du thread se termine.

Donc pour couper proprement la thread pool dans le destructeur de ThreadPoolExecutor il faut d'abord stopper l'io_service (ou sinon les threads ne s'arrêteront jamais, c'est quand même exactement le but d'une thread pool que d'avoir un pack de thread tournant un permanence attendant qu'on leur donne des taches à exécuter), puis ensuite faire un _workers.join_all() qui va en fait faire un join() sur tous les threads du group et donc attendre que tous les threads se terminent correctement.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
 
~ThreadPoolExecutor ()
{
   io_service.stop();
   _workers.join_all() }

Est-ce que la boucle de Executor::run() est vraiment utile ?

Oui, si tu cherches à crée un bug et à bloquer l'appli
Vu que run() appelle join_all() alors que l'io_service n'a pas été arrêté alors les threads ne s'arrêteront pas et join_all va bloquer indéfiniment.

J'ai cru comprendre qu'il n'y avait pas forcément nécessité de join() un thread si on lui laisse le temps de s'exécuter. J'ai faux ?

Oui, c'est faux. Il me semble qu'il n'y aucun moyen avec boost.thread de savoir si un thread s'est vraiment arrêté ou non, donc il faut toujours faire un join() avant de détruire un thread ou de laisser le thread être détruit par son destructeur par sécurité. Si le thread a déjà fini son exécution alors join() retourne immédiatement mais si le thread a encore du travail alors join() va bloquer et attendre que le thread se termine correctement.

Quelques remarques à propos du code :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

class ThreadPoolExecutor : public Thread

Je suppose que c'est une erreur en recopiant car faire dériver THreadPoolExecutor de Thread n'a absolument aucun sens ?

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
void worker_thread (boost::shared_ptr< boost::asio::io_service > io_service)
{
   while (true)
   {
      io_service->run();
   }
}

- while(true){//blabla} la boucle va tourner indéfiniment, ne jamais retourner et pomper tout le CPU disponible. A ne jamais faire (sauf s'il y a un "break" dans le blabla qui s'active sous une certaine condition par exemple.)
- io_service::run ne doit être appelé qu'une seule fois en tout et pour tout dans le programme ! C'est la fonction qui met en branle toute la mécanique interne d'asio , lance la pompe à message etc. Je suis épaté que tu n'es pas subi de crash violent à faire des run() en boucle comme en forcené, peut être que l'auteur d'asio a mis en place des protections contre ce genre de mauvaises utilisations ?

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
if (it->first->_thread->timed_join (td))
{
   delete it->first->_thread;
//...
}

timed_join retourne : soit si le thread a terminé son exécution, soit si le temsp passé en paramètre est écoulé. Donc dans le second cas tu vas faire un delete sur un thread qui tourne encore.

Sinon, plus généralement la classe ThreadPoolManager m'a l'air d'un gros mélange entre :
- un ThreadPoolExecutor + la fonction schedule() qui permet d'exploiter la threadpool offerte par boost.asio. Rien à redire.
- Tout le reste, tout ce mic-mac à base de Thread, startup(), shutdown(), running(), launch(), garbage_collector(), set< pair<Thread *, bool> > _threads... Je n'en comprends pas l'utilité, on dirait que tu veux faire une sorte de registre de thread, qu'est ce que ça vient faire dans une classe nommée ThreadPoolManager ?

Pour finir, un exemple minimale de thread pool basée sur asio :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
 
#include <boost/thread.hpp>
#include <boost/asio.hpp>
#include "Task.h"
 
class ThreadPool
{
   public:
   ThreadPool (int numThread);
   ~ThreadPool ();
   void schedule (Task * t);
 
private:
 
   ThreadPool(const ThreadPool&); // no copy
 
    boost::thread_group _workers;
    boost::asio::io_service _io_service;
    boost::asio::io_service::work _work;
};

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
 
#include "ThreadPool.h"
 
ThreadPool::ThreadPool (int numThread) :
_io_service(),
_work(_io_service)
{
    for (int x = 0 ; x < numThread ; ++x)
        _workers.create_thread (boost::bind (&boost::asio::io_service::run, &_io_service));
}
 
ThreadPool::~ThreadPool()
{
   _io_service.stop();
   _workers.join_all();
}
 
 
void ThreadPool::schedule (Task * t)
{
    _io_service.post (boost::bind (t->execute, t->_data));
}

[Nekkro]

Un grand merci pour ton analyse. Et un merci encore plus grand car tu as résolu mon problème de compréhension pour la destruction de mes threads !

Pour ce qui pouvait manquer pour la compréhension du code :

- J'ai supprimé l'Executor, c'est le manager qui s'occupe lui-même des workers.

- En fait je stocke et utilise "deux types de threads" dans ma threadpool. Les workers, de simples threads qui exécutent en boucle io_service::run(), et mes classes qui héritent de Thread, qui contient elle-même un boost:thread.

ThreadPoolManager::launch() va me servir à démarrer le thread de ces classes qui héritent de Thread et c'est le manager qui va garder un pointeur vers l'instance.

ThreadPoolManager::schedule() va faire exécuter la tâche par les workers.

ThreadPoolManager::shutdown() me permet d'arrêter les threads (là encore je ne pense pas m'y prendre comme il faut) et de détruire ou non le conteneur du thread si je veux le relancer plus tard. Je viens juste de me rendre compte que mon appel à _threads.clear(); va de toute façon tout détruire ...

ThreadPoolManager::start_up() est désormais inutile et a été supprimée.

ThreadPoolManager::running() me sert maintenant pour les workers :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
void worker_thread (boost::shared_ptr< boost::asio::io_service > io_service)
{
	while (sThreadPool->running())
	{
		io_service->run();
	}
}

Par contre je pense améliorer le système en créant une queue de Task. J'ai récemment vu des tutoriels sur les conditions que je pense utiliser pour prévenir les workers qu'une tâche est disponible et lancer un io_service->run().

Après un petit test simple, le système de tâche actuel fonctionne :
Je lance trois tâche qui affiche un message, chacune à 1ms d'intervalle, et elle sont effectuées dans l'ordre, parfois tellement rapidement que le deuxième worker n'as pas le temps de démarrer et de s'occuper d'elle, le premier ayant déjà effectué tout le travail.