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