Discussion :

[devroot]

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 :

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

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 :

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

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 ?

[dalfab]

Bonjour,

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

Code :

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

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)... );
}

[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.

[devroot]

Ah pardon, le prototype de push :

Code :

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

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();
			}

[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.

[devroot]

Bonjour,
Le header pour mon thread queue est le suivant

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
 
 
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();
			}
};

[devroot]

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

no member named 'emplace' in 'threadSafeQueue<std::function<void ()> >'

[Bousk]

Comment est déclaré workQueue ?
Pourquoi pas utiliser std::function et lambda ?

[devroot]

Bonjour Bousk,

workQueue est déclaré comme suit :

Code :

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

threadSafeQueue<std::function<void()> > workQueue;

[Invité]

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 :

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

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

L'erreur :

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 :

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.

[devroot]

Bonsoir WinJerome,

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

Le main()

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 <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 :

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
 
 
#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 :

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
 
#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 :

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

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

[dalfab]

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)>>.

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 :

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

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?

[devroot]

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 :

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

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 :

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

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 :

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

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

[Invité]

Eh bien… l'argument .

Repartons simplement de ta fonction zPrint :

Code :

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

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 :

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

   std::function<void()> f = std::bind(XYZ, 42);

   l'appel de f s'effectuera alors sans paramètre :
   

   Code :

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

   f(); // zPrint(42)

2. soit stocker des std::function<void(int)> :
   

   Code :

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

   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 :

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

   g(42); // zPrint(42)

[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 :

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

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 :

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

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

[devroot]

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

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
#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;
}