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 <MTAllocator.h> 33 #include <MTTaskPool.h> 34 35 36 namespace MT 37 { 38 39 template<typename CLASS_TYPE, typename MACRO_TYPE> 40 struct CheckType 41 { 42 static_assert(std::is_same<CLASS_TYPE, MACRO_TYPE>::value, "Invalid type in MT_DECLARE_TASK macro. See CheckType template instantiation params to details."); 43 }; 44 45 struct TypeChecker 46 { 47 template <typename T> 48 static T QueryThisType(T thisPtr) 49 { 50 return (T)nullptr; 51 } 52 }; 53 54 55 template <typename T> 56 inline void CallDtor(T * p) 57 { 58 #if _MSC_VER 59 p; 60 #endif 61 p->~T(); 62 } 63 64 } 65 66 #define MT_COLOR_DEFAULT (0) 67 #define MT_COLOR_BLUE (1) 68 #define MT_COLOR_RED (2) 69 #define MT_COLOR_YELLOW (3) 70 71 72 #define MT_DECLARE_TASK_IMPL(TYPE) \ 73 void CompileTimeCheckMethod() \ 74 { \ 75 MT::CheckType< std::remove_pointer< decltype(MT::TypeChecker::QueryThisType(this)) >::type, typename TYPE > compileTypeTypesCheck; \ 76 compileTypeTypesCheck; \ 77 } \ 78 \ 79 static void TaskEntryPoint(MT::FiberContext& fiberContext, void* userData) \ 80 { \ 81 TYPE * task = static_cast< TYPE *>(userData); \ 82 task->Do(fiberContext); \ 83 } \ 84 \ 85 static void PoolTaskDestroy(void* userData) \ 86 { \ 87 TYPE * task = static_cast< TYPE *>(userData); \ 88 MT::CallDtor( task ); \ 89 /* Find task pool header */ \ 90 MT::PoolElementHeader * poolHeader = (MT::PoolElementHeader *)((char*)userData - sizeof(MT::PoolElementHeader)); \ 91 /* Fixup pool header, mark task as unused */ \ 92 poolHeader->id.Store(MT::TaskID::UNUSED); \ 93 } \ 94 95 96 97 #ifdef MT_INSTRUMENTED_BUILD 98 #include <MTMicroWebSrv.h> 99 #include <MTProfilerEventListener.h> 100 101 #define MT_DECLARE_TASK(TYPE, colorID) \ 102 static const char * GetDebugID() \ 103 { \ 104 return #TYPE; \ 105 } \ 106 \ 107 static int GetDebugColorIndex() \ 108 { \ 109 return colorID; \ 110 } \ 111 \ 112 MT_DECLARE_TASK_IMPL(TYPE); 113 114 115 #else 116 117 #define MT_DECLARE_TASK(TYPE, colorID) \ 118 MT_DECLARE_TASK_IMPL(TYPE); 119 120 #endif 121 122 123 124 125 126 127 namespace MT 128 { 129 const uint32 MT_MAX_THREAD_COUNT = 64; 130 const uint32 MT_MAX_FIBERS_COUNT = 256; 131 const uint32 MT_SCHEDULER_STACK_SIZE = 1048576; 132 const uint32 MT_FIBER_STACK_SIZE = 65536; 133 134 namespace internal 135 { 136 struct ThreadContext; 137 } 138 139 //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// 140 // Task scheduler 141 //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// 142 class TaskScheduler 143 { 144 friend class FiberContext; 145 friend struct internal::ThreadContext; 146 147 148 149 //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// 150 // Task group description 151 //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// 152 // Application can assign task group to task and later wait until group was finished. 153 class TaskGroupDescription 154 { 155 AtomicInt32 inProgressTaskCount; 156 Event allDoneEvent; 157 158 //Tasks awaiting group through FiberContext::WaitGroupAndYield call 159 ConcurrentQueueLIFO<FiberContext*> waitTasksQueue; 160 161 public: 162 163 bool debugIsFree; 164 165 166 private: 167 168 TaskGroupDescription(TaskGroupDescription& ) {} 169 void operator=(const TaskGroupDescription&) {} 170 171 public: 172 173 TaskGroupDescription() 174 { 175 inProgressTaskCount.Store(0); 176 allDoneEvent.Create( EventReset::MANUAL, true ); 177 debugIsFree = true; 178 } 179 180 int GetTaskCount() const 181 { 182 return inProgressTaskCount.Load(); 183 } 184 185 ConcurrentQueueLIFO<FiberContext*> & GetWaitQueue() 186 { 187 return waitTasksQueue; 188 } 189 190 int Dec() 191 { 192 return inProgressTaskCount.DecFetch(); 193 } 194 195 int Inc() 196 { 197 return inProgressTaskCount.IncFetch(); 198 } 199 200 int Add(int sum) 201 { 202 return inProgressTaskCount.AddFetch(sum); 203 } 204 205 void Signal() 206 { 207 allDoneEvent.Signal(); 208 } 209 210 void Reset() 211 { 212 allDoneEvent.Reset(); 213 } 214 215 bool Wait(uint32 milliseconds) 216 { 217 return allDoneEvent.Wait(milliseconds); 218 } 219 }; 220 221 222 // Thread index for new task 223 AtomicInt32 roundRobinThreadIndex; 224 225 // Started threads count 226 AtomicInt32 startedThreadsCount; 227 228 // Threads created by task manager 229 volatile uint32 threadsCount; 230 internal::ThreadContext threadContext[MT_MAX_THREAD_COUNT]; 231 232 // All groups task statistic 233 TaskGroupDescription allGroups; 234 235 // Groups pool 236 ConcurrentQueueLIFO<TaskGroup> availableGroups; 237 238 // 239 TaskGroupDescription groupStats[TaskGroup::MT_MAX_GROUPS_COUNT]; 240 241 // Fibers pool 242 ConcurrentQueueLIFO<FiberContext*> availableFibers; 243 244 // Fibers context 245 FiberContext fiberContext[MT_MAX_FIBERS_COUNT]; 246 247 #ifdef MT_INSTRUMENTED_BUILD 248 IProfilerEventListener * profilerEventListener; 249 int64 startTime; 250 profile::MicroWebServer profilerWebServer; 251 int32 webServerPort; 252 #endif 253 254 FiberContext* RequestFiberContext(internal::GroupedTask& task); 255 void ReleaseFiberContext(FiberContext* fiberExecutionContext); 256 void RunTasksImpl(ArrayView<internal::TaskBucket>& buckets, FiberContext * parentFiber, bool restoredFromAwaitState); 257 TaskGroupDescription & GetGroupDesc(TaskGroup group); 258 259 static void ThreadMain( void* userData ); 260 static void FiberMain( void* userData ); 261 static bool TryStealTask(internal::ThreadContext& threadContext, internal::GroupedTask & task, uint32 workersCount); 262 263 static FiberContext* ExecuteTask (internal::ThreadContext& threadContext, FiberContext* fiberContext); 264 265 public: 266 267 /// \brief Initializes a new instance of the TaskScheduler class. 268 /// \param workerThreadsCount Worker threads count. Automatically determines the required number of threads if workerThreadsCount set to 0 269 #ifdef MT_INSTRUMENTED_BUILD 270 TaskScheduler(uint32 workerThreadsCount = 0, IProfilerEventListener* listener = nullptr); 271 #else 272 TaskScheduler(uint32 workerThreadsCount = 0); 273 #endif 274 275 276 ~TaskScheduler(); 277 278 template<class TTask> 279 void RunAsync(TaskGroup group, TTask* taskArray, uint32 taskCount); 280 281 void RunAsync(TaskGroup group, TaskHandle* taskHandleArray, uint32 taskHandleCount); 282 283 284 bool WaitGroup(TaskGroup group, uint32 milliseconds); 285 bool WaitAll(uint32 milliseconds); 286 287 TaskGroup CreateGroup(); 288 void ReleaseGroup(TaskGroup group); 289 290 bool IsEmpty(); 291 292 uint32 GetWorkerCount() const; 293 294 bool IsWorkerThread() const; 295 296 #ifdef MT_INSTRUMENTED_BUILD 297 298 size_t GetProfilerEvents(uint32 workerIndex, MT::ProfileEventDesc * dstBuffer, size_t dstBufferSize); 299 void UpdateProfiler(); 300 int32 GetWebServerPort() const; 301 302 inline int64 GetStartTime() const 303 { 304 return startTime; 305 } 306 307 inline uint64 GetTimeStamp() const 308 { 309 return MT::GetTimeMicroSeconds() - startTime; 310 } 311 312 inline IProfilerEventListener* GetProfilerEventListener() 313 { 314 return profilerEventListener; 315 } 316 317 #endif 318 }; 319 } 320 321 #include "MTScheduler.inl" 322 #include "MTFiberContext.inl" 323