1 // The MIT License (MIT)
2 //
3 // 	Copyright (c) 2015 Sergey Makeev, Vadim Slyusarev
4 //
5 // 	Permission is hereby granted, free of charge, to any person obtaining a copy
6 // 	of this software and associated documentation files (the "Software"), to deal
7 // 	in the Software without restriction, including without limitation the rights
8 // 	to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 // 	copies of the Software, and to permit persons to whom the Software is
10 // 	furnished to do so, subject to the following conditions:
11 //
12 //  The above copyright notice and this permission notice shall be included in
13 // 	all copies or substantial portions of the Software.
14 //
15 // 	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 // 	IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 // 	FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 // 	AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 // 	LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 // 	OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
21 // 	THE SOFTWARE.
22 
23 #pragma once
24 
25 #include <MTTools.h>
26 #include <MTPlatform.h>
27 #include <MTConcurrentQueueLIFO.h>
28 #include <MTStackArray.h>
29 #include <MTArrayView.h>
30 #include <MTThreadContext.h>
31 #include <MTFiberContext.h>
32 #include <MTTaskBase.h>
33 #include <MTAllocator.h>
34 #include <MTTaskPool.h>
35 
36 #ifdef MT_INSTRUMENTED_BUILD
37 #include <MTMicroWebSrv.h>
38 #include <MTProfilerEventListener.h>
39 #endif
40 
41 namespace MT
42 {
43 	const uint32 MT_MAX_THREAD_COUNT = 64;
44 	const uint32 MT_MAX_FIBERS_COUNT = 256;
45 	const uint32 MT_SCHEDULER_STACK_SIZE = 1048576;
46 	const uint32 MT_FIBER_STACK_SIZE = 65536;
47 
48 	namespace internal
49 	{
50 		struct ThreadContext;
51 	}
52 
53 	////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
54 	// Task scheduler
55 	////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
56 	class TaskScheduler
57 	{
58 		friend class FiberContext;
59 		friend struct internal::ThreadContext;
60 
61 
62 
63 		////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
64 		// Task group description
65 		////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
66 		// Application can assign task group to task and later wait until group was finished.
67 		class TaskGroupDescription
68 		{
69 			AtomicInt32 inProgressTaskCount;
70 			Event allDoneEvent;
71 
72 			//Tasks awaiting group through FiberContext::WaitGroupAndYield call
73 			ConcurrentQueueLIFO<FiberContext*> waitTasksQueue;
74 
75 		public:
76 
77 			bool debugIsFree;
78 
79 
80 		private:
81 
82 			TaskGroupDescription(TaskGroupDescription& ) {}
83 			void operator=(const TaskGroupDescription&) {}
84 
85 		public:
86 
87 			TaskGroupDescription()
88 			{
89 				inProgressTaskCount.Store(0);
90 				allDoneEvent.Create( EventReset::MANUAL, true );
91 				debugIsFree = true;
92 			}
93 
94 			int GetTaskCount() const
95 			{
96 				return inProgressTaskCount.Load();
97 			}
98 
99 			ConcurrentQueueLIFO<FiberContext*> & GetWaitQueue()
100 			{
101 				return waitTasksQueue;
102 			}
103 
104 			int Dec()
105 			{
106 				return inProgressTaskCount.DecFetch();
107 			}
108 
109 			int Inc()
110 			{
111 				return inProgressTaskCount.IncFetch();
112 			}
113 
114 			int Add(int sum)
115 			{
116 				return inProgressTaskCount.AddFetch(sum);
117 			}
118 
119 			void Signal()
120 			{
121 				allDoneEvent.Signal();
122 			}
123 
124 			void Reset()
125 			{
126 				allDoneEvent.Reset();
127 			}
128 
129 			bool Wait(uint32 milliseconds)
130 			{
131 				return allDoneEvent.Wait(milliseconds);
132 			}
133 		};
134 
135 
136 		// Thread index for new task
137 		AtomicInt32 roundRobinThreadIndex;
138 
139 		// Started threads count
140 		AtomicInt32 startedThreadsCount;
141 
142 		// Threads created by task manager
143 		volatile uint32 threadsCount;
144 		internal::ThreadContext threadContext[MT_MAX_THREAD_COUNT];
145 
146 		// All groups task statistic
147 		TaskGroupDescription allGroups;
148 
149 		// Groups pool
150 		ConcurrentQueueLIFO<TaskGroup> availableGroups;
151 
152 		//
153 		TaskGroupDescription groupStats[TaskGroup::MT_MAX_GROUPS_COUNT];
154 
155 		// Fibers pool
156 		ConcurrentQueueLIFO<FiberContext*> availableFibers;
157 
158 		// Fibers context
159 		FiberContext fiberContext[MT_MAX_FIBERS_COUNT];
160 
161 #ifdef MT_INSTRUMENTED_BUILD
162 		IProfilerEventListener * profilerEventListener;
163 		int64 startTime;
164 		profile::MicroWebServer profilerWebServer;
165 		int32 webServerPort;
166 #endif
167 
168 		FiberContext* RequestFiberContext(internal::GroupedTask& task);
169 		void ReleaseFiberContext(FiberContext* fiberExecutionContext);
170 		void RunTasksImpl(ArrayView<internal::TaskBucket>& buckets, FiberContext * parentFiber, bool restoredFromAwaitState);
171 		TaskGroupDescription & GetGroupDesc(TaskGroup group);
172 
173 		static void ThreadMain( void* userData );
174 		static void FiberMain( void* userData );
175 		static bool TryStealTask(internal::ThreadContext& threadContext, internal::GroupedTask & task, uint32 workersCount);
176 
177 		static FiberContext* ExecuteTask (internal::ThreadContext& threadContext, FiberContext* fiberContext);
178 
179 	public:
180 
181 		/// \brief Initializes a new instance of the TaskScheduler class.
182 		/// \param workerThreadsCount Worker threads count. Automatically determines the required number of threads if workerThreadsCount set to 0
183 #ifdef MT_INSTRUMENTED_BUILD
184 		TaskScheduler(uint32 workerThreadsCount = 0, IProfilerEventListener* listener = nullptr);
185 #else
186 		TaskScheduler(uint32 workerThreadsCount = 0);
187 #endif
188 
189 
190 		~TaskScheduler();
191 
192 		template<class TTask>
193 		void RunAsync(TaskGroup group, TTask* taskArray, uint32 taskCount);
194 
195 		void RunAsync(TaskGroup group, TaskHandle* taskHandleArray, uint32 taskHandleCount);
196 
197 
198 		bool WaitGroup(TaskGroup group, uint32 milliseconds);
199 		bool WaitAll(uint32 milliseconds);
200 
201 		TaskGroup CreateGroup();
202 		void ReleaseGroup(TaskGroup group);
203 
204 		bool IsEmpty();
205 
206 		uint32 GetWorkerCount() const;
207 
208 		bool IsWorkerThread() const;
209 
210 #ifdef MT_INSTRUMENTED_BUILD
211 
212 		size_t GetProfilerEvents(uint32 workerIndex, MT::ProfileEventDesc * dstBuffer, size_t dstBufferSize);
213 		void UpdateProfiler();
214 		int32 GetWebServerPort() const;
215 
216 		inline int64 GetStartTime() const
217 		{
218 			return startTime;
219 		}
220 
221 		inline uint64 GetTimeStamp() const
222 		{
223 			return MT::GetTimeMicroSeconds() - startTime;
224 		}
225 
226 		inline IProfilerEventListener* GetProfilerEventListener()
227 		{
228 			return profilerEventListener;
229 		}
230 
231 #endif
232 	};
233 }
234 
235 #include "MTScheduler.inl"
236 #include "MTFiberContext.inl"
237