tooIbox/ThreadPool.h at master · devtoi/tooIbox · GitHub

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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
#pragma once
#include <thread>
#include <vector>
#include <future>
#include <functional>
#include <deque>
#include <memory/Alloc.h>
#include "ThreadDefinitions.h"
#include "UtilityLibraryDefine.h"
#include "WorkStealQueue.h"
//#define THREAD_EXECUTION_TIME_TRACKING

class ThreadPool {
public:
	UTILITY_API ThreadPool();
	UTILITY_API ThreadPool( const ThreadPool& rhs ) = delete;
	UTILITY_API ~ThreadPool();

	UTILITY_API bool operator == ( const ThreadPool& rhs ) = delete;

	UTILITY_API void Shutdown ();

	UTILITY_API ThreadIdentifier CreateThread ( ThreadType threadType, const pString& name = "Unnamed thread" );

	template <typename ReturnFunction, typename ... Parameters>
	std::future<ReturnFunction> EnqueueJob( const pString& taskName, ThreadIdentifier threadIdentifier, std::function<ReturnFunction(
													Parameters ... )> function, Parameters ... parameters ) {
		std::atomic_bool* ready = tNew( std::atomic_bool );

		ready->store( false );
		std::promise<ReturnFunction>* promise = tNew( std::promise<ReturnFunction> );

		// TODO: Investigate why VC++ doesn't allow these as move parameters
		auto taskWrapper = [promise, ready] ( std::function<ReturnFunction( Parameters ... )> function, Parameters&& ... parameters ) {
							   promise->set_value( function( parameters ... ) );
							   ready->store( true );
						   };

		auto returnWrapper = [promise, ready] () -> ReturnFunction {
								 while ( !ready->load() ) // TODO: Make not busy wait?
									 std::this_thread::yield();
								 auto ret = promise->get_future().get();

								 tDelete( promise );
								 tDelete( ready );
								 return ret;
							 };

		PerThreadQueue& targetQueue = GetEditablePerThreadQueue( threadIdentifier );
		targetQueue.Queue.Push(new std::future<void>(std::async( std::launch::deferred, taskWrapper, function, parameters ... )));
#ifdef THREAD_EXECUTION_TIME_TRACKING
		targetQueue.TaskNames.emplace_back( taskName );
#else
		(void)taskName;
#endif
		targetQueue.EmptyCV.notify_one();

		return std::async( std::launch::deferred, returnWrapper );
	}

	template <typename ReturnFunction, typename ... Parameters>
	std::future<ReturnFunction> EnqueueAnyJob( const pString& taskName, std::function<ReturnFunction( Parameters ... )> function, Parameters ... parameters ) {
		return EnqueueJob( taskName, ThreadIdentifier::invalid(), function, parameters ... );
	}

	void ResetTimes ();
	void PrintTimes ();

	void StartTrackingFrame ();
	void EndTrackingFrame ();

	std::chrono::high_resolution_clock::time_point GetTrackingFrameStart() const;
	std::chrono::high_resolution_clock::time_point GetTrackingFrameEnd() const;

	void StartMainThreadTaskTracking( );
	void StopMainThreadTaskTracking( const pString& taskName );

	struct TrackedThreadInfo;
	const pVector<TrackedThreadInfo>& GetTrackedThreadInfos () const;

	struct TaskExecutionInfo {
		pString	 TaskName;
		std::chrono::high_resolution_clock::time_point Start;
		std::chrono::high_resolution_clock::time_point End;
	};

	struct TrackedThreadInfo {
		pString ThreadName;
		pVector<TaskExecutionInfo> TaskExecutionInfos;
	};

private:
	struct PerThreadQueue;
	struct ThreadInfo;

	void			ThreadFunction ( ThreadInfo* threadInfo );
	void			NamedThreadFunction ( ThreadInfo* threadInfo );
	UTILITY_API PerThreadQueue& GetEditablePerThreadQueue ( ThreadIdentifier threadIdentifier );

	struct ThreadInfo {
		ThreadType		 Type = ThreadType::Any;
		ThreadIdentifier QueueIndex;
		int				 Index = -1;
		pString			 Name  = "Unnamed thread";
#ifdef THREAD_EXECUTION_TIME_TRACKING
		std::mutex TaskTimesLock;
		pVector<TaskExecutionInfo> TaskTimes;
#endif
	};

	struct PerThreadQueue {
		std::mutex EmptyMutex;
		std::condition_variable EmptyCV;
		bool Joining	  = false;
		tooibox::WorkStealQueue<std::future<void>, 4096> Queue;
#ifdef THREAD_EXECUTION_TIME_TRACKING
		pDeque<pString> TaskNames;
#endif
	};

	static const int		  INVALID_QUEUE_INDEX		 = -1;
	static const int		  INVALID_THREAD_ID			 = -1;
	static const unsigned int MAX_NR_OF_SPECIFIC_THREADS = 32;

	pVector<std::thread> m_Pool;
	pVector<ThreadInfo*> m_ThreadInfos;
	pVector<TrackedThreadInfo> m_TrackedThreadInfos;

	PerThreadQueue m_TasksAny;
	PerThreadQueue m_TasksSpecificThread[MAX_NR_OF_SPECIFIC_THREADS];

	int m_NextSpecificThreadQueueIndex = 0;

	std::chrono::high_resolution_clock::time_point m_FrameStart;
	std::chrono::high_resolution_clock::time_point m_FrameEnd;

	std::chrono::high_resolution_clock::time_point m_TrackedFrameStart;
	std::chrono::high_resolution_clock::time_point m_TrackedFrameEnd;

	std::chrono::high_resolution_clock::time_point m_MainThreadStart;
	std::chrono::high_resolution_clock::time_point m_MainThreadEnd;
	pVector<TaskExecutionInfo> m_MainThreadTaskTimes;

	bool m_hasShutDown = false;
};