thread pool et submit(fonction)

Version imprimable

Bonsoir,

Je me suis inspiré d'un code source pour un thread pool qui est fourni dans "C++ Concurrency in Action: Practical Multithreading" d'Anthony Williams.
MAIS, en voulant ajouter un argument à la fonction qui estmise dans la queue des tâches, le programme bute sur la syntaxe...

Dans le code ci-dessous, imaginons que la fonction XYZ prend en argument l'indice de la boucle pour l'imprimer à l'écran.
Code:

1 2 3 4 5 6 7 8 9 10 11 // -----Fonction XYZ void XYZ(int x){ std::cout<<"x=" << x << std::endl; } // ----- Création du Thread Pool et boucle de mise en queue d'un fonction XYZ threadPool* tPool = new threadPool(); for(int i=0; i<100; ++i) tPool->submit(XYZ, i); delete tPool;
Dans l'objet threadPool, j'essaie plusieurs formules (en commentaire dans le code suivant).
Comme message d'erreur j'obtiens :
NO MATCHING FUNCTION FOR CALL TO INVOKE
ou alors
EXPRESSION CONTAINS UNEXPANDED PARAMETER PACK 'arg1'
Code:

1 2 3 4 5 6 template<typename FunctionType, typename... Args> void submit(FunctionType f, Args&&... arg1){ // workQueue.push(std::bind(f, std::forward<Args>(arg1)...)); // workQueue.push( std::bind(f, std::ref(arg1) )); }
Une suggestion ?

Bonjour,

Sans avoir le prototype de la ou des fonctions push(), c'est difficile de répondre.
On pourrait peut-être écrire:
Code:

1 2 3 4 template<typename FunctionType, typename... Args> void submit( FunctionType&& f, Args&&... arg1 ) { workQueue.push( std::forward<FunctionType>(f), std::forward<Args>(arg1)... ); }

14/01/2021, 23h48
Invité

Bonsoir,

Vu le nom workQueue, je suppose qu'il s'agit d'une std::queue<std::thread>.
Le problème ici est que le constructeur de la classe std::thread est marqué explicit. Il ne peut donc pas y avoir de conversion implicite depuis le callable retourné par std::bind() vers le std::thread attendu par std::queue::push().

Tu peux utiliser std::queue::emplace() à la place qui va construire le std::thread avec les paramètres que tu lui passeras.

Ah pardon, le prototype de push :

Code:

1
2
3
4
5
6
 
		void push(T newValue){
			std::lock_guard<std::mutex> lk(mut);
			dataQueue.push(std::move(newValue));
			dataCond.notify_one();
			}

15/01/2021, 01h03
dalfab

On progresse, push() attend un T. Et T c'est quoi?
Il faudrait peut-être aussi l'erreur complète, elle doit indiquer la ligne qui a provoqué l'erreur.

Bonjour,
Le header pour mon thread queue est le suivant

Code:

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
 
 
template<typename T>
class threadSafeQueue{
	private:
		mutable std::mutex mut;
		std::queue<T> dataQueue;
		std::condition_variable dataCond;
	public:
		threadSafeQueue(){}
		void push(T newValue){
			std::lock_guard<std::mutex> lk(mut);
			dataQueue.push(std::move(newValue));
			dataCond.notify_one();
			}
		void wait_and_pop(T& value){
			std::unique_lock<std::mutex> lk(mut);
			dataCond.wait(lk, [this]{return !dataQueue.empty();});
			value=std::move(dataQueue.front());
			dataQueue.pop();
			}
		std::shared_ptr<T> wait_and_pop(){
			std::unique_lock<std::mutex> lk(mut);
			dataCond.wait(lk, [this]{return !dataQueue.empty();});
			std::shared_ptr<T> res(std::make_shared<T>(std::move(dataQueue.front())));
			dataQueue.pop();
			return res;
			}
		bool try_pop(T& value){
			std::lock_guard<std::mutex> lk(mut);
			if(dataQueue.empty()) return false;
			value=std::move(dataQueue.front());
			dataQueue.pop();
			return true;
			}
		std::shared_ptr<T> try_pop(){
			std::lock_guard<std::mutex> lk(mut);
			if(dataQueue.empty()) return std::shared_ptr<T>();
			std::shared_ptr<T> res(std::make_shared<T>(std::move(dataQueue.front())));
			dataQueue.pop();
			return res;
			}
		bool empty() const{
			std::lock_guard<std::mutex> lk(mut);
			return dataQueue.empty();
			}
};

15/01/2021, 10h45
devroot

Bonjour WinJerome,
J'essaie avec emplace au lieu de push mais j'obtiens l'erreur suivante :

Citation:

no member named 'emplace' in 'threadSafeQueue<std::function<void ()> >'
15/01/2021, 14h34
Bousk

Comment est déclaré workQueue ?
Pourquoi pas utiliser std::function et lambda ?
15/01/2021, 15h08
devroot

Bonjour Bousk,

workQueue est déclaré comme suit :

Code:

threadSafeQueue<std::function<void()> > workQueue;
16/01/2021, 15h15
Invité

Citation:

Envoyé par devroot

J'essaie avec emplace au lieu de push mais j'obtiens l'erreur suivante :

Normal, ma supposition n'était pas bonne. Tu as codé ta propre queue, à laquelle tu n'as pas ajouté de fonction membre emplace().
Et ce ne sont pas non plus des std::thread que tu y stockes, mais des std::function<void()>.

Donc mis bout à bout, ton code devrait fonctionner avec la première ligne que tu as commenté :

Code:

workQueue.push(std::bind(f, std::forward<Args>(arg1)...));

L'erreur :

Citation:

EXPRESSION CONTAINS UNEXPANDED PARAMETER PACK 'arg1'

peut s'expliquer avec la seconde workQueue.push( std::bind(f, std::ref(arg1) )); où tu ne développes pas le parameter pack arg1 avec les ....
Note au passage que std::ref() n'est utile que si la fonction passée attend une référence en paramètre.

L'autre erreur :

Citation:

NO MATCHING FUNCTION FOR CALL TO INVOKE

par contre se situe à mon avis dans une partie de code que tu n'as pas encore montré.

Je t'invite donc à nous fournir un ECM (Exemple Complet Minimal) qui reproduit le problème.

Bonsoir WinJerome,

Voici un exemple complet minimum SANS argument pour la fonction zPrint qui tourne bien...

Le main()

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
 
#include <tchar.h>
#include <stdio.h>
#include "threadPool.h"
 
 
void zPrint(int im){
	std::cout<<"zPrint " << im << std::flush<< std::endl;
	std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}
 
 
int _tmain(int argc, _TCHAR* argv[]){
		threadPool* tPool = new threadPool();
		for(int i=0; i<20; ++i) {
				tPool->submit(zPrint, i);
				}
		delete tPool;
 
        system("pause");
		return 0;
}

Le threadpool.h avec un joiner.h

Code:

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
 
 
#include <vector>
#include <thread>
#include <mutex>
#include <queue>
#include "threadSafeQueue.h"
 
class join_threads{
    std::vector<std::thread>& threads;
public:
	explicit join_threads(std::vector<std::thread>& threads_):threads(threads_){}
	~join_threads(){
		for(unsigned long i=0;i<threads.size();++i){
			if(threads[i].joinable())
				threads[i].join();
		 }
	}
};
 
class threadPool{
	std::atomic_bool done;
	threadSafeQueue<std::function<void()> > workQueue;
	std::vector<std::thread> threads;
	join_threads joiner;
 
	void workerThread(){
		while(!done){
			std::function<void()> task;
			if(workQueue.try_pop(task)) task();
			else std::this_thread::yield();
			}
	}
 
	public:
		threadPool(): done(false), joiner(threads){
			unsigned const threadCount = std::thread::hardware_concurrency();
			std::cout<<"threadCount=" << threadCount << std::endl;
			try{
				for(unsigned int i =0; i<threadCount; ++i)
				threads.push_back(std::thread(&threadPool::workerThread, this));
				}
			catch(...){
				done=true;
				throw;
				}
		}
 
		~threadPool(){ done=true;}
 
		template<typename FunctionType, typename... Args>
		void submit(FunctionType f, Args&&... arg1){
			workQueue.push(std::bind(f, std::forward<Args>(arg1)...));          
			}
};

Et le thread_queue réduit au minimum :

Code:

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
 
#include <thread>
#include <mutex>
#include <queue>
 
template<typename T>
class threadSafeQueue{
	private:
		mutable std::mutex mut;
		std::queue<T> dataQueue;
		std::condition_variable dataCond;
	public:
		threadSafeQueue(){}
		void push(T newValue){
			std::lock_guard<std::mutex> lk(mut);
			dataQueue.push(std::move(newValue));
			dataCond.notify_one();
			}
		bool try_pop(T& value){
			std::lock_guard<std::mutex> lk(mut);
			if(dataQueue.empty()) return false;
			value=std::move(dataQueue.front());
			dataQueue.pop();
			return true;
			}
};

Si je modifie les lignes A et B suivantes dans threadPool.h pour inclure un argument INT à ma fonction, j'ai bien-sûr une erreur en ligne C lorsque j'appelle la fonction !?!
NO MATCHING FUNCTION FOR CALL TO OBJECT OF TYPE std::function<void(int)>

Code:

1
2
3
4
 
A)  threadSafeQueue<std::function<void(int)> > workQueue;
B)  std::function<void(int)> task;
C) if(workQueue.try_pop(task)) task();

17/01/2021, 10h12
dalfab
Citation:

Envoyé par devroot

Voici un exemple complet minimum SANS argument pour la fonction zPrint qui tourne bien...

J'en conclut que depuis le début tu nous fais chercher le problème dans un code qui marche! Tu nous a bien indiqué que le type de workQueue c'est threadSafeQueue<std::function<void(void)>>.
Citation:
Envoyé par devroot

Si je modifie les lignes A et B suivantes dans threadPool.h pour inclure un argument INT à ma fonction, j'ai bien-sûr une erreur en ligne C lorsque j'appelle la fonction !?!
NO MATCHING FUNCTION FOR CALL TO OBJECT OF TYPE std::function<void(int)>

Code:

1 2 3 4 A) threadSafeQueue<std::function<void(int)> > workQueue; B) std::function<void(int)> task; C) if(workQueue.try_pop(task)) task();
Tu le dis toi-même : bien sûr j'ai une erreur. Et merci de nous donner enfin la ligne (C) en erreur.
Ton premier code fonctionne. Si les fonctions à appeler ont un ou des paramètres, ceux ci doivent être définis au moment du submit. Quand les threads sont lancés ils auront ce/ces paramètres.
Si par contre tu décides de stocker des fonctions nécessitant un paramètre. Il faut fournir ce paramètre au moment du lancement, et il manque ce paramètre dans la commande ligne (C) : task("paramètre manquant"). Mais pourquoi ne pas utiliser le code qui marche?

Excuse-moi,
Je me suis mal expliqué alors. Je reformule ma question :
Dans ma question initiale figure le problème unique que j'ai depuis le début malgré les corrections, à savoir:
le code - avec une fonction qui n'est pas prévue pour prendre un argument et qui n'en prend pas - fonctionne.
Le code tournait parfaitement avant que je pose la question.

MAIS si je souhaite lui faire prendre un argument, ça bute sur la ligne suivante:
Code:

1 2 3 4 5 6 template<typename FunctionType, typename... Args> void submit(FunctionType f, Args&&... arg1){ // workQueue.push(std::bind(f, std::forward<Args>(arg1)...)); // workQueue.push( std::bind(f, std::ref(arg1) )); }
J'ai bien compris que les lignes suivantes doivent mentionner le traitement d'un integer :
Code:

1 2 3 A) threadSafeQueue<std::function<void(int)> > workQueue; B) std::function<void(int)> task;
Mais il reste que cette ligne plante car il manque l'argument. Mais je ne sais pas quoi ajouter ici :
Code:

1 2 C) if(workQueue.try_pop(task)) task();

17/01/2021, 14h59
Invité
Eh bien… l'argument ;).

Repartons simplement de ta fonction zPrint :
Code:

1 2 3 4 void zPrint(int im){ std::cout<<"zPrint " << im << std::flush<< std::endl; std::this_thread::sleep_for(std::chrono::milliseconds(1000)); }
Tu peux :
1. soit stocker des std::funtion<void()> : il te faudra alors fournir ton argument en amont :
  
  Code:
  
  std::function<void()> f = std::bind(XYZ, 42);
  
  l'appel de f s'effectuera alors sans paramètre :
  
  Code:
  
  f(); // zPrint(42)
2. soit stocker des std::function<void(int)> :
  
  Code:
  
  std::function<void(int)> g = std::bind(XYZ); // ou plus directement XYZ
  
  et c'est lors de l'appel que tu fourniras ton argument (int) attendu :
  
  Code:
  
  g(42); // zPrint(42)
17/01/2021, 15h31
devroot
Merci WinJerome pour tes suggestions,

Je comprends mieux le fonctionnement du wrapper de fonction.
Je préfère ta première solution qui me permet de conserver mes déclarations simples std::function<void()> dans le code du threadpool.
J'imbrique alors ma fonction zPrint et son argument (l'itérateur de la boucle qui passera de 0 à 19 donc) dans un std:function.
Je ne me soucie plus d'un argument dans la ligne
Code:

1 2 if(workQueue.try_pop(task)) task();
Ca tourne MAIS ça n'éxécute que quatre fois la fonction zPrint malgré la boucle prévue pour 20 passages...
J'ai vérifié à l'impression à l'écran et en mode debug avec arrêt sur la ligne.

De fait, en mode debug, 4 passages et impressions.
Sans mode debug, aucune impression.

Evidemment 4 correspond au nombre renvoyé par std::thread::hardware_concurrency().
Je m'interroge sur cette limitation...

Si je remplace le nombre de threads possible par la valeur que je veux soit 20 arbitrairement , alors ça tourne 20 fois (en debug).
Code:

1 2 3 4 for(unsigned int i =0; i<20; ++i) au lieu de for(unsigned int i =0; i<std::thread::hardware_concurrency(); ++i)
MAIS je ne comprends pas pourquoi ça tournait 20 fois auparavant quand je n'avais pas d'argument pour ma fonction.

Je ne sais pas si mon explication est claire, je peux reformuler si vous voulez...
Merci

J'ai donc testé ce code qui tourne sur le site https://godbolt.org/z/r8jsvx :

Code:

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
#include <iostream>
#include <thread>
#include <queue>
#include <functional>
#include <mutex>
#include <condition_variable>
 
 
template<typename T>
class threadSafeQueue{
	private:
		mutable std::mutex mut;
		std::queue<T> dataQueue;
		std::condition_variable dataCond;
	public:
		threadSafeQueue(){}
		void push(T newValue){
			std::lock_guard<std::mutex> lk(mut);
			dataQueue.push(std::move(newValue));
			dataCond.notify_one();
			}
		bool try_pop(T& value){
			std::lock_guard<std::mutex> lk(mut);
			if(dataQueue.empty()) return false;
			value=std::move(dataQueue.front());
			dataQueue.pop();
			return true;
			}
};
 
class join_threads{
    std::vector<std::thread>& threads;
public:
	explicit join_threads(std::vector<std::thread>& threads_):threads(threads_){}
	~join_threads(){
		for(unsigned long i=0;i<threads.size();++i){
			if(threads[i].joinable())
				threads[i].join();
		 }
	}
};
 
class threadPool{
	std::atomic_bool done;
	threadSafeQueue<std::function<void()> > workQueue;
	std::vector<std::thread> threads;
	join_threads joiner;
 
	void workerThread(){
		while(!done){
			std::function<void()> task;
			if(workQueue.try_pop(task)) task();
			else std::this_thread::yield();
			}
	}
 
	public:
		threadPool(): done(false), joiner(threads){
			unsigned const threadCount = std::thread::hardware_concurrency();
			std::cout<<"threadCount=" << threadCount << std::endl;
			try{
				for(unsigned int i =0; i<20; ++i)                                                    //20 ou threadCount
				threads.push_back(std::thread(&threadPool::workerThread, this));
				}
			catch(...){
				done=true;
				throw;
				}
		}
 
		~threadPool(){ done=true;}
 
		template<typename FunctionType>
		void submit(FunctionType f){
			 workQueue.push(std::function<void()>(f));        						// push 1
		   //	workQueue.push(std::bind(f, std::forward<Args>(arg1)...));          // push 2
			}
};
 
 std::mutex mu;
 
void zPrint(int im){
	mu.lock();std::cout<<"zPrint " << im << std::flush<< std::endl; mu.unlock();
	std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}
 
 
int _tmain(){
		threadPool* tPool = new threadPool();
		for(int i=0; i<20; ++i) {
				std::function<void()> f = std::bind(zPrint, i);
				tPool->submit(f);
				}
		delete tPool;
 
        system("pause");
		return 0;
}