1f25ce65dSSergey Makeev // The MIT License (MIT) 2f25ce65dSSergey Makeev // 3f25ce65dSSergey Makeev // Copyright (c) 2015 Sergey Makeev, Vadim Slyusarev 4f25ce65dSSergey Makeev // 5f25ce65dSSergey Makeev // Permission is hereby granted, free of charge, to any person obtaining a copy 6f25ce65dSSergey Makeev // of this software and associated documentation files (the "Software"), to deal 7f25ce65dSSergey Makeev // in the Software without restriction, including without limitation the rights 8f25ce65dSSergey Makeev // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 9f25ce65dSSergey Makeev // copies of the Software, and to permit persons to whom the Software is 10f25ce65dSSergey Makeev // furnished to do so, subject to the following conditions: 11f25ce65dSSergey Makeev // 12f25ce65dSSergey Makeev // The above copyright notice and this permission notice shall be included in 13f25ce65dSSergey Makeev // all copies or substantial portions of the Software. 14f25ce65dSSergey Makeev // 15f25ce65dSSergey Makeev // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 16f25ce65dSSergey Makeev // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 17f25ce65dSSergey Makeev // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 18f25ce65dSSergey Makeev // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 19f25ce65dSSergey Makeev // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 20f25ce65dSSergey Makeev // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN 21f25ce65dSSergey Makeev // THE SOFTWARE. 22f25ce65dSSergey Makeev 2358964d0bSSergey Makeev #include <MTScheduler.h> 24*b8503d38Ss.makeev_local #include <MTStaticVector.h> 25c73d1551SSergey Makeev #include <string.h> // for memset 2658964d0bSSergey Makeev 2758964d0bSSergey Makeev namespace MT 2858964d0bSSergey Makeev { 2958964d0bSSergey Makeev 300727c1f9SSergey Makeev #ifdef MT_INSTRUMENTED_BUILD 31b23bdf5aSs.makeev_local TaskScheduler::TaskScheduler(uint32 workerThreadsCount, WorkerThreadParams* workerParameters, IProfilerEventListener* listener, TaskStealingMode::Type stealMode) 320727c1f9SSergey Makeev #else 33b23bdf5aSs.makeev_local TaskScheduler::TaskScheduler(uint32 workerThreadsCount, WorkerThreadParams* workerParameters, TaskStealingMode::Type stealMode) 340727c1f9SSergey Makeev #endif 3558964d0bSSergey Makeev : roundRobinThreadIndex(0) 3643c41d5cSSergey Makeev , startedThreadsCount(0) 37b23bdf5aSs.makeev_local , taskStealingDisabled(stealMode == TaskStealingMode::DISABLED) 3858964d0bSSergey Makeev { 3951901c6bSSergey Makeev 40ff5c86b0SSergey Makeev #ifdef MT_INSTRUMENTED_BUILD 410727c1f9SSergey Makeev profilerEventListener = listener; 42ff5c86b0SSergey Makeev #endif 43ff5c86b0SSergey Makeev 4434a394c3SSergey Makeev if (workerThreadsCount != 0) 4534a394c3SSergey Makeev { 46a23fd933SSergey Makeev threadsCount.StoreRelaxed( MT::Clamp(workerThreadsCount, (uint32)1, (uint32)MT_MAX_THREAD_COUNT) ); 4734a394c3SSergey Makeev } else 4834a394c3SSergey Makeev { 4934a394c3SSergey Makeev //query number of processor 50d7cf17b1Ss.makeev_local threadsCount.StoreRelaxed( (uint32)MT::Clamp(Thread::GetNumberOfHardwareThreads() - 1, 1, (int)MT_MAX_THREAD_COUNT) ); 5134a394c3SSergey Makeev } 5234a394c3SSergey Makeev 53f4db079dSs.makeev // create fiber pool (fibers with standard stack size) 54f4db079dSs.makeev for (uint32 i = 0; i < MT_MAX_STANDART_FIBERS_COUNT; i++) 5558964d0bSSergey Makeev { 56f4db079dSs.makeev FiberContext& context = standartFiberContexts[i]; 57f4db079dSs.makeev context.fiber.Create(MT_STANDART_FIBER_STACK_SIZE, FiberMain, &context); 5860ac17fbSs.makeev_local bool res = standartFibersAvailable.TryPush( &context ); 5960ac17fbSs.makeev_local MT_USED_IN_ASSERT(res); 6060ac17fbSs.makeev_local MT_ASSERT(res == true, "Can't add fiber to storage"); 6158964d0bSSergey Makeev } 6258964d0bSSergey Makeev 63f4db079dSs.makeev // create fiber pool (fibers with extended stack size) 64f4db079dSs.makeev for (uint32 i = 0; i < MT_MAX_EXTENDED_FIBERS_COUNT; i++) 65f4db079dSs.makeev { 66f4db079dSs.makeev FiberContext& context = extendedFiberContexts[i]; 67f4db079dSs.makeev context.fiber.Create(MT_EXTENDED_FIBER_STACK_SIZE, FiberMain, &context); 6860ac17fbSs.makeev_local bool res = extendedFibersAvailable.TryPush( &context ); 6960ac17fbSs.makeev_local MT_USED_IN_ASSERT(res); 7060ac17fbSs.makeev_local MT_ASSERT(res == true, "Can't add fiber to storage"); 71f4db079dSs.makeev } 72f4db079dSs.makeev 73f4db079dSs.makeev 742e846c40SSergey Makeev for (int16 i = 0; i < TaskGroup::MT_MAX_GROUPS_COUNT; i++) 75806ef292SSergey Makeev { 76754cb997SSergey Makeev if (i != TaskGroup::DEFAULT) 77806ef292SSergey Makeev { 7860ac17fbSs.makeev_local bool res = availableGroups.TryPush( TaskGroup(i) ); 7960ac17fbSs.makeev_local MT_USED_IN_ASSERT(res); 8060ac17fbSs.makeev_local MT_ASSERT(res == true, "Can't add group to storage"); 81806ef292SSergey Makeev } 82806ef292SSergey Makeev } 83806ef292SSergey Makeev 849d94a278Ss.makeev_local #if MT_GROUP_DEBUG 852e846c40SSergey Makeev groupStats[TaskGroup::DEFAULT].SetDebugIsFree(false); 869d94a278Ss.makeev_local #endif 87806ef292SSergey Makeev 8858964d0bSSergey Makeev // create worker thread pool 89a23fd933SSergey Makeev int32 totalThreadsCount = GetWorkersCount(); 90a23fd933SSergey Makeev for (int32 i = 0; i < totalThreadsCount; i++) 9158964d0bSSergey Makeev { 922b4df162SSergey Makeev threadContext[i].SetThreadIndex(i); 9358964d0bSSergey Makeev threadContext[i].taskScheduler = this; 94d7cf17b1Ss.makeev_local 95d7cf17b1Ss.makeev_local uint32 threadCore = i; 96d7cf17b1Ss.makeev_local ThreadPriority::Type priority = ThreadPriority::DEFAULT; 97d7cf17b1Ss.makeev_local if (workerParameters != nullptr) 98d7cf17b1Ss.makeev_local { 99d7cf17b1Ss.makeev_local const WorkerThreadParams& params = workerParameters[i]; 100d7cf17b1Ss.makeev_local 101d7cf17b1Ss.makeev_local threadCore = params.core; 102d7cf17b1Ss.makeev_local priority = params.priority; 103d7cf17b1Ss.makeev_local } 104d7cf17b1Ss.makeev_local 105d7cf17b1Ss.makeev_local threadContext[i].thread.Start( MT_SCHEDULER_STACK_SIZE, WorkerThreadMain, &threadContext[i], threadCore, priority); 10658964d0bSSergey Makeev } 10758964d0bSSergey Makeev } 10858964d0bSSergey Makeev 109d7cf17b1Ss.makeev_local 11058964d0bSSergey Makeev TaskScheduler::~TaskScheduler() 11158964d0bSSergey Makeev { 112a23fd933SSergey Makeev int32 totalThreadsCount = GetWorkersCount(); 113a23fd933SSergey Makeev for (int32 i = 0; i < totalThreadsCount; i++) 11458964d0bSSergey Makeev { 11581ec7369SSergey Makeev threadContext[i].state.Store(internal::ThreadState::EXIT); 11658964d0bSSergey Makeev threadContext[i].hasNewTasksEvent.Signal(); 11758964d0bSSergey Makeev } 11858964d0bSSergey Makeev 119a23fd933SSergey Makeev for (int32 i = 0; i < totalThreadsCount; i++) 12058964d0bSSergey Makeev { 121c7362320Ss.makeev_local threadContext[i].thread.Join(); 12258964d0bSSergey Makeev } 12358964d0bSSergey Makeev } 12458964d0bSSergey Makeev 1254965e494Ss.makeev_local FiberContext* TaskScheduler::RequestFiberContext(internal::GroupedTask& task) 126f4db079dSs.makeev { 1274965e494Ss.makeev_local FiberContext *fiberContext = task.awaitingFiber; 1284965e494Ss.makeev_local if (fiberContext) 1294965e494Ss.makeev_local { 1304965e494Ss.makeev_local task.awaitingFiber = nullptr; 1314965e494Ss.makeev_local return fiberContext; 1324965e494Ss.makeev_local } 13360ac17fbSs.makeev_local 1344965e494Ss.makeev_local MT::StackRequirements::Type stackRequirements = task.desc.stackRequirements; 1354965e494Ss.makeev_local 1364965e494Ss.makeev_local fiberContext = nullptr; 13760ac17fbSs.makeev_local bool res = false; 13860ac17fbSs.makeev_local MT_USED_IN_ASSERT(res); 139f4db079dSs.makeev switch(stackRequirements) 140f4db079dSs.makeev { 141f4db079dSs.makeev case MT::StackRequirements::STANDARD: 14260ac17fbSs.makeev_local res = standartFibersAvailable.TryPop(fiberContext); 14360ac17fbSs.makeev_local MT_ASSERT(res, "Can't get more standard fibers!"); 144f4db079dSs.makeev break; 145f4db079dSs.makeev case MT::StackRequirements::EXTENDED: 14660ac17fbSs.makeev_local res = extendedFibersAvailable.TryPop(fiberContext); 14760ac17fbSs.makeev_local MT_ASSERT(res, "Can't get more extended fibers!"); 148f4db079dSs.makeev break; 149f4db079dSs.makeev default: 150f4db079dSs.makeev MT_REPORT_ASSERT("Unknown stack requrements"); 151f4db079dSs.makeev } 152f4db079dSs.makeev 15360ac17fbSs.makeev_local MT_ASSERT(fiberContext != nullptr, "Can't get more fibers. Too many tasks in flight simultaneously?"); 15458964d0bSSergey Makeev 15558964d0bSSergey Makeev fiberContext->currentTask = task.desc; 15658964d0bSSergey Makeev fiberContext->currentGroup = task.group; 15758964d0bSSergey Makeev fiberContext->parentFiber = task.parentFiber; 158f4db079dSs.makeev fiberContext->stackRequirements = stackRequirements; 15958964d0bSSergey Makeev return fiberContext; 16058964d0bSSergey Makeev } 16158964d0bSSergey Makeev 16260ac17fbSs.makeev_local void TaskScheduler::ReleaseFiberContext(FiberContext*&& fiberContext) 16358964d0bSSergey Makeev { 164721f8c0bSs.makeev_local MT_ASSERT(fiberContext, "Can't release nullptr Fiber. fiberContext is nullptr"); 165f4db079dSs.makeev 166f4db079dSs.makeev MT::StackRequirements::Type stackRequirements = fiberContext->stackRequirements; 16758964d0bSSergey Makeev fiberContext->Reset(); 168f4db079dSs.makeev 1694965e494Ss.makeev_local MT_ASSERT(fiberContext != nullptr, "Fiber context can't be nullptr"); 1704965e494Ss.makeev_local 1714965e494Ss.makeev_local bool res = false; 1724965e494Ss.makeev_local MT_USED_IN_ASSERT(res); 1734965e494Ss.makeev_local switch(stackRequirements) 1744965e494Ss.makeev_local { 1754965e494Ss.makeev_local case MT::StackRequirements::STANDARD: 1764965e494Ss.makeev_local res = standartFibersAvailable.TryPush(std::move(fiberContext)); 1774965e494Ss.makeev_local break; 1784965e494Ss.makeev_local case MT::StackRequirements::EXTENDED: 1794965e494Ss.makeev_local res = extendedFibersAvailable.TryPush(std::move(fiberContext)); 1804965e494Ss.makeev_local break; 1814965e494Ss.makeev_local default: 1824965e494Ss.makeev_local MT_REPORT_ASSERT("Unknown stack requrements"); 1834965e494Ss.makeev_local } 1844965e494Ss.makeev_local 18560ac17fbSs.makeev_local MT_USED_IN_ASSERT(res); 18660ac17fbSs.makeev_local MT_ASSERT(res != false, "Can't return fiber to storage"); 18758964d0bSSergey Makeev } 18858964d0bSSergey Makeev 18958964d0bSSergey Makeev FiberContext* TaskScheduler::ExecuteTask(internal::ThreadContext& threadContext, FiberContext* fiberContext) 19058964d0bSSergey Makeev { 19134a394c3SSergey Makeev MT_ASSERT(threadContext.thread.IsCurrentThread(), "Thread context sanity check failed"); 19258964d0bSSergey Makeev 19334a394c3SSergey Makeev MT_ASSERT(fiberContext, "Invalid fiber context"); 19434a394c3SSergey Makeev MT_ASSERT(fiberContext->currentTask.IsValid(), "Invalid task"); 19558964d0bSSergey Makeev 19658964d0bSSergey Makeev // Set actual thread context to fiber 19758964d0bSSergey Makeev fiberContext->SetThreadContext(&threadContext); 19858964d0bSSergey Makeev 19958964d0bSSergey Makeev // Update task status 20058964d0bSSergey Makeev fiberContext->SetStatus(FiberTaskStatus::RUNNED); 20158964d0bSSergey Makeev 20234a394c3SSergey Makeev MT_ASSERT(fiberContext->GetThreadContext()->thread.IsCurrentThread(), "Thread context sanity check failed"); 20358964d0bSSergey Makeev 204f4db079dSs.makeev const void* poolUserData = fiberContext->currentTask.userData; 20511d67161SSergey Makeev TPoolTaskDestroy poolDestroyFunc = fiberContext->currentTask.poolDestroyFunc; 20611d67161SSergey Makeev 20760ac17fbSs.makeev_local #ifdef MT_INSTRUMENTED_BUILD 2086e90b535Ss.makeev_local //threadContext.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::SUSPEND); 2096e90b535Ss.makeev_local threadContext.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::STOP); 21060ac17fbSs.makeev_local #endif 21160ac17fbSs.makeev_local 21258964d0bSSergey Makeev // Run current task code 21358964d0bSSergey Makeev Fiber::SwitchTo(threadContext.schedulerFiber, fiberContext->fiber); 21458964d0bSSergey Makeev 21560ac17fbSs.makeev_local #ifdef MT_INSTRUMENTED_BUILD 2166e90b535Ss.makeev_local //threadContext.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::RESUME); 2176e90b535Ss.makeev_local threadContext.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::START); 21860ac17fbSs.makeev_local #endif 21960ac17fbSs.makeev_local 22058964d0bSSergey Makeev // If task was done 22158964d0bSSergey Makeev FiberTaskStatus::Type taskStatus = fiberContext->GetStatus(); 22258964d0bSSergey Makeev if (taskStatus == FiberTaskStatus::FINISHED) 22358964d0bSSergey Makeev { 22411d67161SSergey Makeev //destroy task (call dtor) for "fire and forget" type of task from TaskPool 22511d67161SSergey Makeev if (poolDestroyFunc != nullptr) 22611d67161SSergey Makeev { 22711d67161SSergey Makeev poolDestroyFunc(poolUserData); 22811d67161SSergey Makeev } 22911d67161SSergey Makeev 230806ef292SSergey Makeev TaskGroup taskGroup = fiberContext->currentGroup; 23131bebe6fSSergey Makeev 232806ef292SSergey Makeev TaskScheduler::TaskGroupDescription & groupDesc = threadContext.taskScheduler->GetGroupDesc(taskGroup); 23358964d0bSSergey Makeev 23458964d0bSSergey Makeev // Update group status 235806ef292SSergey Makeev int groupTaskCount = groupDesc.Dec(); 23634a394c3SSergey Makeev MT_ASSERT(groupTaskCount >= 0, "Sanity check failed!"); 23758964d0bSSergey Makeev if (groupTaskCount == 0) 23858964d0bSSergey Makeev { 239806ef292SSergey Makeev // Signal pending threads that group work is finished. Group can be destroyed after this call. 240806ef292SSergey Makeev groupDesc.Signal(); 241806ef292SSergey Makeev 242754cb997SSergey Makeev fiberContext->currentGroup = TaskGroup::INVALID; 24358964d0bSSergey Makeev } 24458964d0bSSergey Makeev 24558964d0bSSergey Makeev // Update total task count 246806ef292SSergey Makeev int allGroupTaskCount = threadContext.taskScheduler->allGroups.Dec(); 247806ef292SSergey Makeev MT_ASSERT(allGroupTaskCount >= 0, "Sanity check failed!"); 248806ef292SSergey Makeev if (allGroupTaskCount == 0) 24958964d0bSSergey Makeev { 25058964d0bSSergey Makeev // Notify all tasks in all group finished 2512a2464ccSSergey Makeev threadContext.taskScheduler->allGroups.Signal(); 25258964d0bSSergey Makeev } 25358964d0bSSergey Makeev 25458964d0bSSergey Makeev FiberContext* parentFiberContext = fiberContext->parentFiber; 25558964d0bSSergey Makeev if (parentFiberContext != nullptr) 25658964d0bSSergey Makeev { 25781ec7369SSergey Makeev int childrenFibersCount = parentFiberContext->childrenFibersCount.DecFetch(); 25834a394c3SSergey Makeev MT_ASSERT(childrenFibersCount >= 0, "Sanity check failed!"); 25958964d0bSSergey Makeev 26058964d0bSSergey Makeev if (childrenFibersCount == 0) 26158964d0bSSergey Makeev { 26258964d0bSSergey Makeev // This is a last subtask. Restore parent task 26334a394c3SSergey Makeev MT_ASSERT(threadContext.thread.IsCurrentThread(), "Thread context sanity check failed"); 26434a394c3SSergey Makeev MT_ASSERT(parentFiberContext->GetThreadContext() == nullptr, "Inactive parent should not have a valid thread context"); 26558964d0bSSergey Makeev 26658964d0bSSergey Makeev // WARNING!! Thread context can changed here! Set actual current thread context. 26758964d0bSSergey Makeev parentFiberContext->SetThreadContext(&threadContext); 26858964d0bSSergey Makeev 26934a394c3SSergey Makeev MT_ASSERT(parentFiberContext->GetThreadContext()->thread.IsCurrentThread(), "Thread context sanity check failed"); 27058964d0bSSergey Makeev 27158964d0bSSergey Makeev // All subtasks is done. 27258964d0bSSergey Makeev // Exiting and return parent fiber to scheduler 27358964d0bSSergey Makeev return parentFiberContext; 27458964d0bSSergey Makeev } else 27558964d0bSSergey Makeev { 27658964d0bSSergey Makeev // Other subtasks still exist 27758964d0bSSergey Makeev // Exiting 27858964d0bSSergey Makeev return nullptr; 27958964d0bSSergey Makeev } 28058964d0bSSergey Makeev } else 28158964d0bSSergey Makeev { 28258964d0bSSergey Makeev // Task is finished and no parent task 28358964d0bSSergey Makeev // Exiting 28458964d0bSSergey Makeev return nullptr; 28558964d0bSSergey Makeev } 28658964d0bSSergey Makeev } 28758964d0bSSergey Makeev 28834a394c3SSergey Makeev MT_ASSERT(taskStatus != FiberTaskStatus::RUNNED, "Incorrect task status") 28958964d0bSSergey Makeev return nullptr; 29058964d0bSSergey Makeev } 29158964d0bSSergey Makeev 29258964d0bSSergey Makeev 29358964d0bSSergey Makeev void TaskScheduler::FiberMain(void* userData) 29458964d0bSSergey Makeev { 29558964d0bSSergey Makeev FiberContext& fiberContext = *(FiberContext*)(userData); 29658964d0bSSergey Makeev for(;;) 29758964d0bSSergey Makeev { 29834a394c3SSergey Makeev MT_ASSERT(fiberContext.currentTask.IsValid(), "Invalid task in fiber context"); 29934a394c3SSergey Makeev MT_ASSERT(fiberContext.GetThreadContext(), "Invalid thread context"); 30034a394c3SSergey Makeev MT_ASSERT(fiberContext.GetThreadContext()->thread.IsCurrentThread(), "Thread context sanity check failed"); 30158964d0bSSergey Makeev 30260ac17fbSs.makeev_local #ifdef MT_INSTRUMENTED_BUILD 30360ac17fbSs.makeev_local fiberContext.fiber.SetName( MT_SYSTEM_TASK_FIBER_NAME ); 3046e90b535Ss.makeev_local fiberContext.GetThreadContext()->NotifyTaskExecuteStateChanged( fiberContext.currentTask.debugColor, fiberContext.currentTask.debugID, TaskExecuteState::START ); 30560ac17fbSs.makeev_local #endif 30658964d0bSSergey Makeev 30760ac17fbSs.makeev_local fiberContext.currentTask.taskFunc( fiberContext, fiberContext.currentTask.userData ); 30858964d0bSSergey Makeev fiberContext.SetStatus(FiberTaskStatus::FINISHED); 30958964d0bSSergey Makeev 3108112dedfSSergey Makeev #ifdef MT_INSTRUMENTED_BUILD 31160ac17fbSs.makeev_local fiberContext.fiber.SetName( MT_SYSTEM_TASK_FIBER_NAME ); 3126e90b535Ss.makeev_local fiberContext.GetThreadContext()->NotifyTaskExecuteStateChanged( fiberContext.currentTask.debugColor, fiberContext.currentTask.debugID, TaskExecuteState::STOP ); 3138112dedfSSergey Makeev #endif 3148112dedfSSergey Makeev 31558964d0bSSergey Makeev Fiber::SwitchTo(fiberContext.fiber, fiberContext.GetThreadContext()->schedulerFiber); 31658964d0bSSergey Makeev } 31758964d0bSSergey Makeev 31858964d0bSSergey Makeev } 31958964d0bSSergey Makeev 32058964d0bSSergey Makeev 321b23bdf5aSs.makeev_local bool TaskScheduler::TryStealTask(internal::ThreadContext& threadContext, internal::GroupedTask & task, uint32 workersCount, bool taskStealingDisabled) 3222b4df162SSergey Makeev { 323b23bdf5aSs.makeev_local if (workersCount <= 1 || taskStealingDisabled ) 3242b4df162SSergey Makeev { 3252b4df162SSergey Makeev return false; 3262b4df162SSergey Makeev } 3272b4df162SSergey Makeev 328a1626843SSergey Makeev uint32 victimIndex = threadContext.random.Get(); 329a1626843SSergey Makeev 330a1626843SSergey Makeev for (uint32 attempt = 0; attempt < workersCount; attempt++) 3312b4df162SSergey Makeev { 332a1626843SSergey Makeev uint32 index = victimIndex % workersCount; 333a1626843SSergey Makeev if (index == threadContext.workerIndex) 334a1626843SSergey Makeev { 335a1626843SSergey Makeev victimIndex++; 336a1626843SSergey Makeev index = victimIndex % workersCount; 3372b4df162SSergey Makeev } 3382b4df162SSergey Makeev 339a1626843SSergey Makeev internal::ThreadContext& victimContext = threadContext.taskScheduler->threadContext[index]; 340d7cf17b1Ss.makeev_local if (victimContext.queue.TryPopNewest(task)) 341a1626843SSergey Makeev { 342a1626843SSergey Makeev return true; 343a1626843SSergey Makeev } 344a1626843SSergey Makeev 345a1626843SSergey Makeev victimIndex++; 346a1626843SSergey Makeev } 347a1626843SSergey Makeev return false; 3482b4df162SSergey Makeev } 3492b4df162SSergey Makeev 35002d170cfSs.makeev_local 351c88507a8Ss.makeev_local void TaskScheduler::WorkerThreadMain( void* userData ) 35258964d0bSSergey Makeev { 35358964d0bSSergey Makeev internal::ThreadContext& context = *(internal::ThreadContext*)(userData); 35434a394c3SSergey Makeev MT_ASSERT(context.taskScheduler, "Task scheduler must be not null!"); 355c88507a8Ss.makeev_local 356d7cf17b1Ss.makeev_local #ifdef MT_INSTRUMENTED_BUILD 357c88507a8Ss.makeev_local const char* threadNames[] = {"worker0","worker1","worker2","worker3","worker4","worker5","worker6","worker7","worker8","worker9","worker10","worker11","worker12"}; 358c88507a8Ss.makeev_local if (context.workerIndex < MT_ARRAY_SIZE(threadNames)) 359c88507a8Ss.makeev_local { 360d7cf17b1Ss.makeev_local Thread::SetThreadName(threadNames[context.workerIndex]); 361c88507a8Ss.makeev_local } else 362c88507a8Ss.makeev_local { 363d7cf17b1Ss.makeev_local Thread::SetThreadName("worker_thread"); 364c88507a8Ss.makeev_local } 365d7cf17b1Ss.makeev_local #endif 366c88507a8Ss.makeev_local 367ae5bbefbSs.makeev_local context.schedulerFiber.CreateFromCurrentThreadAndRun(SchedulerFiberMain, userData); 36802d170cfSs.makeev_local } 36902d170cfSs.makeev_local 37002d170cfSs.makeev_local 37171198a5eSs.makeev_local void TaskScheduler::SchedulerFiberMain( void* userData ) 37202d170cfSs.makeev_local { 37302d170cfSs.makeev_local internal::ThreadContext& context = *(internal::ThreadContext*)(userData); 37402d170cfSs.makeev_local MT_ASSERT(context.taskScheduler, "Task scheduler must be not null!"); 3750727c1f9SSergey Makeev 3760727c1f9SSergey Makeev #ifdef MT_INSTRUMENTED_BUILD 3770727c1f9SSergey Makeev context.NotifyThreadCreate(context.workerIndex); 3780727c1f9SSergey Makeev #endif 3790727c1f9SSergey Makeev 380b23bdf5aSs.makeev_local bool taskStealingDisabled = context.taskScheduler->IsTaskStealingDisabled(); 381a23fd933SSergey Makeev uint32 workersCount = context.taskScheduler->GetWorkersCount(); 382a23fd933SSergey Makeev int32 totalThreadsCount = context.taskScheduler->threadsCount.LoadRelaxed(); 383a1626843SSergey Makeev 38481ec7369SSergey Makeev context.taskScheduler->startedThreadsCount.IncFetch(); 38543c41d5cSSergey Makeev 386a23fd933SSergey Makeev //Simple spinlock until all threads is started and initialized 3876bbc0262SSergey Makeev for(;;) 38843c41d5cSSergey Makeev { 389a23fd933SSergey Makeev int32 initializedThreadsCount = context.taskScheduler->startedThreadsCount.Load(); 390a23fd933SSergey Makeev if (initializedThreadsCount == totalThreadsCount) 39143c41d5cSSergey Makeev { 39243c41d5cSSergey Makeev break; 39343c41d5cSSergey Makeev } 394d7cf17b1Ss.makeev_local 395d7cf17b1Ss.makeev_local // sleep some time until all other thread initialized 39643c41d5cSSergey Makeev Thread::Sleep(1); 39743c41d5cSSergey Makeev } 39843c41d5cSSergey Makeev 399f4db079dSs.makeev HardwareFullMemoryBarrier(); 400f4db079dSs.makeev 4010727c1f9SSergey Makeev #ifdef MT_INSTRUMENTED_BUILD 4020727c1f9SSergey Makeev context.NotifyThreadStart(context.workerIndex); 4036e90b535Ss.makeev_local context.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::START); 4040727c1f9SSergey Makeev #endif 4050727c1f9SSergey Makeev 40681ec7369SSergey Makeev while(context.state.Load() != internal::ThreadState::EXIT) 40758964d0bSSergey Makeev { 40858964d0bSSergey Makeev internal::GroupedTask task; 409b23bdf5aSs.makeev_local if (context.queue.TryPopOldest(task) || TryStealTask(context, task, workersCount, taskStealingDisabled) ) 41058964d0bSSergey Makeev { 41160ac17fbSs.makeev_local #ifdef MT_INSTRUMENTED_BUILD 41260ac17fbSs.makeev_local bool isNewTask = (task.awaitingFiber == nullptr); 41360ac17fbSs.makeev_local #endif 41460ac17fbSs.makeev_local 41558964d0bSSergey Makeev // There is a new task 41658964d0bSSergey Makeev FiberContext* fiberContext = context.taskScheduler->RequestFiberContext(task); 41734a394c3SSergey Makeev MT_ASSERT(fiberContext, "Can't get execution context from pool"); 41834a394c3SSergey Makeev MT_ASSERT(fiberContext->currentTask.IsValid(), "Sanity check failed"); 419f4db079dSs.makeev MT_ASSERT(fiberContext->stackRequirements == task.desc.stackRequirements, "Sanity check failed"); 42058964d0bSSergey Makeev 42158964d0bSSergey Makeev while(fiberContext) 42258964d0bSSergey Makeev { 4238112dedfSSergey Makeev #ifdef MT_INSTRUMENTED_BUILD 42460ac17fbSs.makeev_local if (isNewTask) 42560ac17fbSs.makeev_local { 42660ac17fbSs.makeev_local //TODO: 42760ac17fbSs.makeev_local isNewTask = false; 42860ac17fbSs.makeev_local } 4298112dedfSSergey Makeev #endif 43058964d0bSSergey Makeev // prevent invalid fiber resume from child tasks, before ExecuteTask is done 43181ec7369SSergey Makeev fiberContext->childrenFibersCount.IncFetch(); 43258964d0bSSergey Makeev 43358964d0bSSergey Makeev FiberContext* parentFiber = ExecuteTask(context, fiberContext); 43458964d0bSSergey Makeev 43558964d0bSSergey Makeev FiberTaskStatus::Type taskStatus = fiberContext->GetStatus(); 43658964d0bSSergey Makeev 43758964d0bSSergey Makeev //release guard 43881ec7369SSergey Makeev int childrenFibersCount = fiberContext->childrenFibersCount.DecFetch(); 43958964d0bSSergey Makeev 44058964d0bSSergey Makeev // Can drop fiber context - task is finished 44158964d0bSSergey Makeev if (taskStatus == FiberTaskStatus::FINISHED) 44258964d0bSSergey Makeev { 44334a394c3SSergey Makeev MT_ASSERT( childrenFibersCount == 0, "Sanity check failed"); 44460ac17fbSs.makeev_local context.taskScheduler->ReleaseFiberContext(std::move(fiberContext)); 44558964d0bSSergey Makeev 44658964d0bSSergey Makeev // If parent fiber is exist transfer flow control to parent fiber, if parent fiber is null, exit 44758964d0bSSergey Makeev fiberContext = parentFiber; 44858964d0bSSergey Makeev } else 44958964d0bSSergey Makeev { 45034a394c3SSergey Makeev MT_ASSERT( childrenFibersCount >= 0, "Sanity check failed"); 45158964d0bSSergey Makeev 45258964d0bSSergey Makeev // No subtasks here and status is not finished, this mean all subtasks already finished before parent return from ExecuteTask 45358964d0bSSergey Makeev if (childrenFibersCount == 0) 45458964d0bSSergey Makeev { 45534a394c3SSergey Makeev MT_ASSERT(parentFiber == nullptr, "Sanity check failed"); 45658964d0bSSergey Makeev } else 45758964d0bSSergey Makeev { 45858964d0bSSergey Makeev // If subtasks still exist, drop current task execution. task will be resumed when last subtask finished 45958964d0bSSergey Makeev break; 46058964d0bSSergey Makeev } 46158964d0bSSergey Makeev 462*b8503d38Ss.makeev_local // If task is yielded execution, get another task from queue. 4633a3d248dSs.makeev_local if (taskStatus == FiberTaskStatus::YIELDED) 46458964d0bSSergey Makeev { 465*b8503d38Ss.makeev_local // Task is yielded, add to tasks queue 466*b8503d38Ss.makeev_local ArrayView<internal::GroupedTask> buffer(context.descBuffer, 1); 467*b8503d38Ss.makeev_local ArrayView<internal::TaskBucket> buckets( MT_ALLOCATE_ON_STACK(sizeof(internal::TaskBucket)), 1 ); 468*b8503d38Ss.makeev_local 469*b8503d38Ss.makeev_local FiberContext* yieldedTask = fiberContext; 470*b8503d38Ss.makeev_local StaticVector<FiberContext*, 1> yieldedTasksQueue(1, yieldedTask); 471*b8503d38Ss.makeev_local internal::DistibuteDescriptions( TaskGroup(TaskGroup::ASSIGN_FROM_CONTEXT), yieldedTasksQueue.Begin(), buffer, buckets ); 472*b8503d38Ss.makeev_local 473*b8503d38Ss.makeev_local // add yielded task to scheduler 474*b8503d38Ss.makeev_local context.taskScheduler->RunTasksImpl(buckets, nullptr, true); 475*b8503d38Ss.makeev_local 476*b8503d38Ss.makeev_local // ATENTION! yielded task can be already completed at this point 477*b8503d38Ss.makeev_local 47858964d0bSSergey Makeev break; 47958964d0bSSergey Makeev } 48058964d0bSSergey Makeev } 48158964d0bSSergey Makeev } //while(fiberContext) 48258964d0bSSergey Makeev 48358964d0bSSergey Makeev } else 48458964d0bSSergey Makeev { 485a1626843SSergey Makeev #ifdef MT_INSTRUMENTED_BUILD 4864a90b4a6SSergey Makeev context.NotifyThreadIdleBegin(context.workerIndex); 487a1626843SSergey Makeev #endif 488a1626843SSergey Makeev 48960ac17fbSs.makeev_local // Queue if empty and stealing attempt failed 49060ac17fbSs.makeev_local // Wait for new events 49158964d0bSSergey Makeev context.hasNewTasksEvent.Wait(2000); 492a1626843SSergey Makeev 493a1626843SSergey Makeev #ifdef MT_INSTRUMENTED_BUILD 4944a90b4a6SSergey Makeev context.NotifyThreadIdleEnd(context.workerIndex); 495a1626843SSergey Makeev #endif 496a1626843SSergey Makeev 49758964d0bSSergey Makeev } 49858964d0bSSergey Makeev 49958964d0bSSergey Makeev } // main thread loop 5000727c1f9SSergey Makeev 5010727c1f9SSergey Makeev #ifdef MT_INSTRUMENTED_BUILD 5026e90b535Ss.makeev_local context.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::STOP); 5030727c1f9SSergey Makeev context.NotifyThreadStop(context.workerIndex); 5040727c1f9SSergey Makeev #endif 5050727c1f9SSergey Makeev 50658964d0bSSergey Makeev } 50758964d0bSSergey Makeev 50856aa031bSSergey Makeev void TaskScheduler::RunTasksImpl(ArrayView<internal::TaskBucket>& buckets, FiberContext * parentFiber, bool restoredFromAwaitState) 50958964d0bSSergey Makeev { 510806ef292SSergey Makeev // This storage is necessary to calculate how many tasks we add to different groups 511754cb997SSergey Makeev int newTaskCountInGroup[TaskGroup::MT_MAX_GROUPS_COUNT]; 512806ef292SSergey Makeev 513806ef292SSergey Makeev // Default value is 0 514806ef292SSergey Makeev memset(&newTaskCountInGroup[0], 0, sizeof(newTaskCountInGroup)); 515806ef292SSergey Makeev 51658964d0bSSergey Makeev // Set parent fiber pointer 51758964d0bSSergey Makeev // Calculate the number of tasks per group 51858964d0bSSergey Makeev // Calculate total number of tasks 51958964d0bSSergey Makeev size_t count = 0; 520a101e543SSergey Makeev for (size_t i = 0; i < buckets.Size(); ++i) 52158964d0bSSergey Makeev { 52258964d0bSSergey Makeev internal::TaskBucket& bucket = buckets[i]; 52358964d0bSSergey Makeev for (size_t taskIndex = 0; taskIndex < bucket.count; taskIndex++) 52458964d0bSSergey Makeev { 52558964d0bSSergey Makeev internal::GroupedTask & task = bucket.tasks[taskIndex]; 52658964d0bSSergey Makeev 52758964d0bSSergey Makeev task.parentFiber = parentFiber; 528754cb997SSergey Makeev 529754cb997SSergey Makeev int idx = task.group.GetValidIndex(); 530754cb997SSergey Makeev MT_ASSERT(idx >= 0 && idx < TaskGroup::MT_MAX_GROUPS_COUNT, "Invalid index"); 531754cb997SSergey Makeev newTaskCountInGroup[idx]++; 53231bebe6fSSergey Makeev } 53331bebe6fSSergey Makeev 53458964d0bSSergey Makeev count += bucket.count; 53558964d0bSSergey Makeev } 53658964d0bSSergey Makeev 53758964d0bSSergey Makeev // Increments child fibers count on parent fiber 53858964d0bSSergey Makeev if (parentFiber) 53958964d0bSSergey Makeev { 54081ec7369SSergey Makeev parentFiber->childrenFibersCount.AddFetch((int)count); 54158964d0bSSergey Makeev } 54258964d0bSSergey Makeev 54358964d0bSSergey Makeev if (restoredFromAwaitState == false) 54458964d0bSSergey Makeev { 545806ef292SSergey Makeev // Increase the number of active tasks in the group using data from temporary storage 546754cb997SSergey Makeev for (size_t i = 0; i < TaskGroup::MT_MAX_GROUPS_COUNT; i++) 547806ef292SSergey Makeev { 548806ef292SSergey Makeev int groupNewTaskCount = newTaskCountInGroup[i]; 549806ef292SSergey Makeev if (groupNewTaskCount > 0) 550806ef292SSergey Makeev { 551806ef292SSergey Makeev groupStats[i].Reset(); 552806ef292SSergey Makeev groupStats[i].Add((uint32)groupNewTaskCount); 553806ef292SSergey Makeev } 554806ef292SSergey Makeev } 555806ef292SSergey Makeev 55658964d0bSSergey Makeev // Increments all task in progress counter 5572a2464ccSSergey Makeev allGroups.Reset(); 5582a2464ccSSergey Makeev allGroups.Add((uint32)count); 55958964d0bSSergey Makeev } else 56058964d0bSSergey Makeev { 56158964d0bSSergey Makeev // If task's restored from await state, counters already in correct state 56258964d0bSSergey Makeev } 56358964d0bSSergey Makeev 56458964d0bSSergey Makeev // Add to thread queue 565a101e543SSergey Makeev for (size_t i = 0; i < buckets.Size(); ++i) 56658964d0bSSergey Makeev { 567a23fd933SSergey Makeev int bucketIndex = roundRobinThreadIndex.IncFetch() % threadsCount.LoadRelaxed(); 56858964d0bSSergey Makeev internal::ThreadContext & context = threadContext[bucketIndex]; 56958964d0bSSergey Makeev 57058964d0bSSergey Makeev internal::TaskBucket& bucket = buckets[i]; 57158964d0bSSergey Makeev 572d7cf17b1Ss.makeev_local for(;;) 573d7cf17b1Ss.makeev_local { 574d7cf17b1Ss.makeev_local bool res = context.queue.Add(bucket.tasks, bucket.count); 575d7cf17b1Ss.makeev_local if (res == true) 576d7cf17b1Ss.makeev_local { 577d7cf17b1Ss.makeev_local break; 578d7cf17b1Ss.makeev_local } 579d7cf17b1Ss.makeev_local 580d7cf17b1Ss.makeev_local //Can't add new tasks onto the queue. Look like the job system is overloaded. Wait some time and try again. 581d7cf17b1Ss.makeev_local //TODO: implement waiting until workers done using events. 582d7cf17b1Ss.makeev_local Thread::Sleep(10); 583d7cf17b1Ss.makeev_local } 584d7cf17b1Ss.makeev_local 58558964d0bSSergey Makeev context.hasNewTasksEvent.Signal(); 58658964d0bSSergey Makeev } 58758964d0bSSergey Makeev } 58858964d0bSSergey Makeev 589f4db079dSs.makeev void TaskScheduler::RunAsync(TaskGroup group, const TaskHandle* taskHandleArray, uint32 taskHandleCount) 59058d12dadSSergey Makeev { 59158d12dadSSergey Makeev MT_ASSERT(!IsWorkerThread(), "Can't use RunAsync inside Task. Use FiberContext.RunAsync() instead."); 59258d12dadSSergey Makeev 59358d12dadSSergey Makeev ArrayView<internal::GroupedTask> buffer(MT_ALLOCATE_ON_STACK(sizeof(internal::GroupedTask) * taskHandleCount), taskHandleCount); 59458d12dadSSergey Makeev 595a23fd933SSergey Makeev uint32 bucketCount = MT::Min((uint32)GetWorkersCount(), taskHandleCount); 59658d12dadSSergey Makeev ArrayView<internal::TaskBucket> buckets(MT_ALLOCATE_ON_STACK(sizeof(internal::TaskBucket) * bucketCount), bucketCount); 59758d12dadSSergey Makeev 59858d12dadSSergey Makeev internal::DistibuteDescriptions(group, taskHandleArray, buffer, buckets); 59958d12dadSSergey Makeev RunTasksImpl(buckets, nullptr, false); 60058d12dadSSergey Makeev } 60158d12dadSSergey Makeev 602806ef292SSergey Makeev bool TaskScheduler::WaitGroup(TaskGroup group, uint32 milliseconds) 60358964d0bSSergey Makeev { 60434a394c3SSergey Makeev MT_VERIFY(IsWorkerThread() == false, "Can't use WaitGroup inside Task. Use FiberContext.WaitGroupAndYield() instead.", return false); 60558964d0bSSergey Makeev 606806ef292SSergey Makeev TaskScheduler::TaskGroupDescription & groupDesc = GetGroupDesc(group); 6072f083884Ss.makeev_local 608806ef292SSergey Makeev return groupDesc.Wait(milliseconds); 60958964d0bSSergey Makeev } 61058964d0bSSergey Makeev 61158964d0bSSergey Makeev bool TaskScheduler::WaitAll(uint32 milliseconds) 61258964d0bSSergey Makeev { 61334a394c3SSergey Makeev MT_VERIFY(IsWorkerThread() == false, "Can't use WaitAll inside Task.", return false); 61458964d0bSSergey Makeev 6152a2464ccSSergey Makeev return allGroups.Wait(milliseconds); 61658964d0bSSergey Makeev } 61758964d0bSSergey Makeev 618b23bdf5aSs.makeev_local bool TaskScheduler::IsTaskStealingDisabled() const 61958964d0bSSergey Makeev { 620b23bdf5aSs.makeev_local return taskStealingDisabled; 62158964d0bSSergey Makeev } 62258964d0bSSergey Makeev 623a23fd933SSergey Makeev int32 TaskScheduler::GetWorkersCount() const 62458964d0bSSergey Makeev { 625a23fd933SSergey Makeev return threadsCount.LoadRelaxed(); 62658964d0bSSergey Makeev } 62758964d0bSSergey Makeev 62858964d0bSSergey Makeev bool TaskScheduler::IsWorkerThread() const 62958964d0bSSergey Makeev { 63058964d0bSSergey Makeev for (uint32 i = 0; i < MT_MAX_THREAD_COUNT; i++) 63158964d0bSSergey Makeev { 63258964d0bSSergey Makeev if (threadContext[i].thread.IsCurrentThread()) 63358964d0bSSergey Makeev { 63458964d0bSSergey Makeev return true; 63558964d0bSSergey Makeev } 63658964d0bSSergey Makeev } 63758964d0bSSergey Makeev return false; 63858964d0bSSergey Makeev } 63958964d0bSSergey Makeev 640806ef292SSergey Makeev TaskGroup TaskScheduler::CreateGroup() 641806ef292SSergey Makeev { 642806ef292SSergey Makeev MT_ASSERT(IsWorkerThread() == false, "Can't use CreateGroup inside Task."); 643806ef292SSergey Makeev 644754cb997SSergey Makeev TaskGroup group; 64560ac17fbSs.makeev_local if (!availableGroups.TryPop(group)) 646806ef292SSergey Makeev { 6476bbc0262SSergey Makeev MT_REPORT_ASSERT("Group pool is empty"); 648806ef292SSergey Makeev } 649806ef292SSergey Makeev 650754cb997SSergey Makeev int idx = group.GetValidIndex(); 6514965e494Ss.makeev_local MT_USED_IN_ASSERT(idx); 6522e846c40SSergey Makeev MT_ASSERT(groupStats[idx].GetDebugIsFree() == true, "Bad logic!"); 6539d94a278Ss.makeev_local #if MT_GROUP_DEBUG 6542e846c40SSergey Makeev groupStats[idx].SetDebugIsFree(false); 6559d94a278Ss.makeev_local #endif 656806ef292SSergey Makeev 657806ef292SSergey Makeev return group; 658806ef292SSergey Makeev } 659806ef292SSergey Makeev 660806ef292SSergey Makeev void TaskScheduler::ReleaseGroup(TaskGroup group) 661806ef292SSergey Makeev { 662806ef292SSergey Makeev MT_ASSERT(IsWorkerThread() == false, "Can't use ReleaseGroup inside Task."); 663754cb997SSergey Makeev MT_ASSERT(group.IsValid(), "Invalid group ID"); 664806ef292SSergey Makeev 665754cb997SSergey Makeev int idx = group.GetValidIndex(); 6664965e494Ss.makeev_local MT_USED_IN_ASSERT(idx); 6672e846c40SSergey Makeev MT_ASSERT(groupStats[idx].GetDebugIsFree() == false, "Group already released"); 6689d94a278Ss.makeev_local #if MT_GROUP_DEBUG 6692e846c40SSergey Makeev groupStats[idx].SetDebugIsFree(true); 6709d94a278Ss.makeev_local #endif 671806ef292SSergey Makeev 67260ac17fbSs.makeev_local bool res = availableGroups.TryPush(std::move(group)); 67360ac17fbSs.makeev_local MT_USED_IN_ASSERT(res); 67460ac17fbSs.makeev_local MT_ASSERT(res, "Can't return group to pool"); 675806ef292SSergey Makeev } 676806ef292SSergey Makeev 677806ef292SSergey Makeev TaskScheduler::TaskGroupDescription & TaskScheduler::GetGroupDesc(TaskGroup group) 678806ef292SSergey Makeev { 679754cb997SSergey Makeev MT_ASSERT(group.IsValid(), "Invalid group ID"); 680806ef292SSergey Makeev 681754cb997SSergey Makeev int idx = group.GetValidIndex(); 682754cb997SSergey Makeev TaskScheduler::TaskGroupDescription & groupDesc = groupStats[idx]; 683806ef292SSergey Makeev 6842e846c40SSergey Makeev MT_ASSERT(groupDesc.GetDebugIsFree() == false, "Invalid group"); 685806ef292SSergey Makeev return groupDesc; 686806ef292SSergey Makeev } 68758964d0bSSergey Makeev } 68802d170cfSs.makeev_local 68902d170cfSs.makeev_local 690