Discussion :

[Sapinerie]

Bonjour,

Pour un projet personnel, j'ai été amené à créer un classe gérant la répartition de callback sur plusieurs threads (nommé Task ou Callback selon le context d'éxécution voulu, n'importe quel thread ou principal respectivement).
Quand je configure mon programme pour qu'il utilise 2 threads (principal inclut), il plante (le lieu du plantage sera indiqué dans le code). Quand je ne met que un seul thread, tout fonctionne.
Visual Studio m'indique de plus avec une assertion que le problème est dû à un déférencement impossible d'un itérateur. J'ai essayé de changer le conteneur (queue au début, puis list, puis deque), cela ne change rien.

Le code :

TaskManager.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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
#pragma once
 
#include <functional>
#include <vector>
#include <mutex>
#include <thread>
#include <deque>
 
#include "Semaphore.hpp"
 
class TaskManager
{
public:
 
	TaskManager ();
	~TaskManager ();
 
	void pollCallback (unsigned int taskTimeLimit);
 
	void setThreadNumber (unsigned int number);
 
	inline void addCallback (const std::function <void (void)>& callback)
	{
		m_mutex.lock ();
		m_callbackQueue.push_back (callback);
		m_mutex.unlock ();
	}
 
	inline void addTask (const std::function <void (void)>& task)
	{
		m_mutex.lock ();
		m_taskQueue.push_back (task);
		m_semaphore.post ();
		m_mutex.unlock ();
	}
 
private:
 
	std::deque <std::function <void (void)>> m_callbackQueue;
	std::deque <std::function <void (void)> > m_taskQueue;
	std::mutex m_mutex;
 
	struct ThreadData
	{
		std::thread* thread;
		bool quit;
	};
 
	std::vector <ThreadData> m_threadPool;
	Semaphore m_semaphore;
};

TaskManager.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
101
102
103
104
 
#include "TaskManager.hpp"
 
#include <exception>
#include <ctime>
 
TaskManager::TaskManager ()
{
 
}
 
TaskManager::~TaskManager ()
{
	setThreadNumber (1);
}
 
void TaskManager::pollCallback (unsigned int taskTimeLimit)
{
	clock_t c = clock ();
 
	m_mutex.lock ();
	while (true)
	{
		while (!m_callbackQueue.empty ()) /* ### <<< Celui VS, l'erreur est ici quelque soit la méthode de débugage (pas à pas, arrêt lors de l'assertion ...) */
		{
			std::function <void (void)> f = m_callbackQueue.front ();
			m_callbackQueue.pop_front ();
 
			m_mutex.unlock ();
			f ();
			m_mutex.lock ();
		}
 
		if (m_taskQueue.empty () || (unsigned int) (clock () - c) / (CLOCKS_PER_SEC / 1000) > taskTimeLimit)
			break;
 
		if (m_semaphore.trywait ())
		{
			std::function <void (void)> f = m_taskQueue.front ();
			m_taskQueue.pop_front ();
 
			m_mutex.unlock ();
			f ();
			m_mutex.lock ();
		}
	}
	m_mutex.unlock ();
}
 
void TaskManager::setThreadNumber (unsigned int number)
{
#ifdef _DEBUG
	if (number == 0)
	{
		throw std::invalid_argument ("Can not set the program to have 0 threads");
	}
#endif
 
	number --;
 
	if (number == m_threadPool.size ())
		return;
	else if (number > m_threadPool.size ())
	{
		const unsigned int last = m_threadPool.size ();
		m_threadPool.resize (number);
 
		for (unsigned int i = last ; i < number ; i++)
		{
			ThreadData& current = m_threadPool [i];
			current.quit = false;
			current.thread = new std::thread ([this, &current] ()
			{
				while (!current.quit)
				{
					m_mutex.lock ();
					while (m_semaphore.trywait ())
					{
						std::function <void (void)> f = m_taskQueue.front ();
						m_taskQueue.pop_front ();
 
						m_mutex.unlock ();
						f ();
						m_mutex.lock ();
					}
					m_mutex.unlock ();
 
					std::unique_lock <std::mutex> lock (m_mutex);
					m_semaphore.wait (lock);
				}
			});
		}
	}
	else
	{
		for (unsigned int i = number ; i < m_threadPool.size () ; i++)
		{
			ThreadData& current = m_threadPool [i];
			current.quit = true;
			delete current.thread;
		}
		m_threadPool.resize (number);
	}
}

Semaphore.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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
#pragma once
 
#include <condition_variable>
#include <mutex>
 
class Semaphore
{
public:
 
	Semaphore ()
	{
 
	}
 
	~Semaphore ()
	{
		m_cond.notify_all ();
	}
 
	inline void post ()
	{
		m_counter++;
		m_cond.notify_one ();
	}
 
	inline void wait (std::unique_lock <std::mutex>& lock)
	{
		while (!m_counter)
			m_cond.wait (lock);
		m_counter --;
		return;
	}
 
	inline bool trywait ()
	{
		bool ok = m_counter != 0;
 
		if (ok)
			m_counter --;
 
		return ok;
	}
 
private:
 
	unsigned int m_counter;
	std::condition_variable m_cond;
};

Merci d'avance,

[Trademark]

Salut,

On peut voir ta classe principal où tu crées les threads et ajoutent les callbacks/tasks ? Je n'ai pas vu grand chose qui pourrait poser problème dans ta classe TaskManager.

Par contre dans ta classe Semaphore, tu fais plein d'opérations concurrentes non protégées. À moins que son accès soit toujours protégé par le mutex de ta classe TaskManager...

[Sapinerie]

D'abord, merci pour ta réponse

Voici ma classe principale, pour limiter la recherche, j'ai réduit le code à ce qui génère l'erreur, le reste a été commenté (et supprimé de code ci dessous). L'erreur reste.

Game.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
 
#pragma once
 
#include <functional>
 
#include <SFML\Graphics.hpp>
 
class TaskManager;
 
class Game
{
public:
 
	Game ();
	~Game ();
 
	void exec ();
	void stop ();
 
	inline sf::RenderWindow& getWindow ()
	{
		return m_window;
	}
 
	void postCallback (const std::function <void (void)>& callback);
	void postTask (const std::function <void (void)>& task);
 
private:
 
	sf::RenderWindow m_window;
 
	TaskManager* m_taskManager;
};

Game.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
#include "Game.hpp"
 
#include "TaskManager.hpp"
 
Game::Game () :
	m_window (sf::VideoMode (800, 600), "Titre"),
	m_taskManager (new TaskManager)
{
	m_taskManager->setThreadNumber (2);
}
 
Game::~Game ()
{
	delete m_taskManager;
}
 
void Game::exec ()
{
	sf::Clock clock;
 
	while (m_window.isOpen ())
	{
		sf::Event event;
		while (m_window.pollEvent (event))
		{
			if (event.type == sf::Event::Closed)
			{
				stop ();
			}
		}
 
		m_window.clear ();
 
		sf::Uint32 limit = 33 - clock.getElapsedTime ().asMilliseconds ();
		clock.restart ();
 
		m_taskManager->pollCallback (limit);
 
		m_window.display ();
	}
}
 
void Game::stop ()
{
	m_window.close ();
}
 
void Game::postCallback (const std::function <void (void)>& callback)
{
	m_taskManager->addCallback (callback);
}
 
void Game::postTask (const std::function <void (void)>& task)
{
	m_taskManager->addTask (task);
}
 
#include <exception>
#include <iostream>
#include <typeinfo>
 
int main ()
{
	try
	{
		Game game;
		game.exec ();
		return 0;
	}
	catch (std::exception& e)
	{
		std::cerr << "Exception caught [" << typeid (e).name () << "] : " << e.what () << std::endl;
		system ("pause");
	}
}

Je viens aussi de lancer un débogage sur le code ainsi obtenue. Le point d'arrêt étant sur le mutex avant l'erreur : le programme plante dès que je lance un pas à pas principal.

[Iradrille]

Code :

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

sf::Uint32 limit = 33 - clock.getElapsedTime ().asMilliseconds ();

Rien à voir avec ton soucis, mais si une image mets plus de 33 ms à être dessinée (très probable pour la première image par exemple), ta limite vaudra ~4 milliards. Remplaces par

Code :

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

sf::Uint32 limit = 33 - std::min(33, clock.getElapsedTime ().asMilliseconds ());

pour plus de sureté.

[Sapinerie]

J'ai trouvé l'erreur. Cela vient de Semaphore::trywait qui renvoie toujours true car Semaphore::m_counter n'est pas initialisé et vaut potentiellement beaucoup. Ainsi le thread 2 essaye de pop sur un conteneur vide. L'erreur se propage ensuite dans le thread 1 quand celui ci se réveille après le mutex.

Sinon merci Iradrille pour ta remarque