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