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 #include <MTScheduler.h> 24 #include <MTStaticVector.h> 25 #include <string.h> // for memset 26 27 28 // Enable low latency experimental wait code path. 29 // Look like low latency hybrid wait is work better for PS4/X1, but a little worse on PC 30 //#define MT_LOW_LATENCY_EXPERIMENTAL_WAIT (1) 31 32 33 namespace MT 34 { 35 mt_thread_local uint32 isWorkerThreadTLS = 0; 36 37 38 #ifdef MT_INSTRUMENTED_BUILD 39 TaskScheduler::TaskScheduler(uint32 workerThreadsCount, WorkerThreadParams* workerParameters, IProfilerEventListener* listener, TaskStealingMode::Type stealMode) 40 #else 41 TaskScheduler::TaskScheduler(uint32 workerThreadsCount, WorkerThreadParams* workerParameters, TaskStealingMode::Type stealMode) 42 #endif 43 : roundRobinThreadIndex(0) 44 , startedThreadsCount(0) 45 , taskStealingDisabled(stealMode == TaskStealingMode::DISABLED) 46 { 47 48 #ifdef MT_INSTRUMENTED_BUILD 49 profilerEventListener = listener; 50 #endif 51 52 if (workerThreadsCount != 0) 53 { 54 threadsCount.StoreRelaxed( MT::Clamp(workerThreadsCount, (uint32)1, (uint32)MT_MAX_THREAD_COUNT) ); 55 } else 56 { 57 //query number of processor 58 threadsCount.StoreRelaxed( (uint32)MT::Clamp(Thread::GetNumberOfHardwareThreads() - 1, 1, (int)MT_MAX_THREAD_COUNT) ); 59 } 60 61 // create fiber pool (fibers with standard stack size) 62 for (uint32 i = 0; i < MT_MAX_STANDART_FIBERS_COUNT; i++) 63 { 64 FiberContext& context = standartFiberContexts[i]; 65 context.fiber.Create(MT_STANDART_FIBER_STACK_SIZE, FiberMain, &context); 66 bool res = standartFibersAvailable.TryPush( &context ); 67 MT_USED_IN_ASSERT(res); 68 MT_ASSERT(res == true, "Can't add fiber to storage"); 69 } 70 71 // create fiber pool (fibers with extended stack size) 72 for (uint32 i = 0; i < MT_MAX_EXTENDED_FIBERS_COUNT; i++) 73 { 74 FiberContext& context = extendedFiberContexts[i]; 75 context.fiber.Create(MT_EXTENDED_FIBER_STACK_SIZE, FiberMain, &context); 76 bool res = extendedFibersAvailable.TryPush( &context ); 77 MT_USED_IN_ASSERT(res); 78 MT_ASSERT(res == true, "Can't add fiber to storage"); 79 } 80 81 82 for (int16 i = 0; i < TaskGroup::MT_MAX_GROUPS_COUNT; i++) 83 { 84 if (i != TaskGroup::DEFAULT) 85 { 86 bool res = availableGroups.TryPush( TaskGroup(i) ); 87 MT_USED_IN_ASSERT(res); 88 MT_ASSERT(res == true, "Can't add group to storage"); 89 } 90 } 91 92 #if MT_GROUP_DEBUG 93 groupStats[TaskGroup::DEFAULT].SetDebugIsFree(false); 94 #endif 95 96 // create worker thread pool 97 int32 totalThreadsCount = GetWorkersCount(); 98 for (int32 i = 0; i < totalThreadsCount; i++) 99 { 100 threadContext[i].SetThreadIndex(i); 101 threadContext[i].taskScheduler = this; 102 103 uint32 threadCore = i; 104 ThreadPriority::Type priority = ThreadPriority::DEFAULT; 105 if (workerParameters != nullptr) 106 { 107 const WorkerThreadParams& params = workerParameters[i]; 108 109 threadCore = params.core; 110 priority = params.priority; 111 } 112 113 threadContext[i].thread.Start( MT_SCHEDULER_STACK_SIZE, WorkerThreadMain, &threadContext[i], threadCore, priority); 114 } 115 } 116 117 118 TaskScheduler::~TaskScheduler() 119 { 120 int32 totalThreadsCount = GetWorkersCount(); 121 for (int32 i = 0; i < totalThreadsCount; i++) 122 { 123 threadContext[i].state.Store(internal::ThreadState::EXIT); 124 threadContext[i].hasNewTasksEvent.Signal(); 125 } 126 127 for (int32 i = 0; i < totalThreadsCount; i++) 128 { 129 threadContext[i].thread.Join(); 130 } 131 } 132 133 FiberContext* TaskScheduler::RequestFiberContext(internal::GroupedTask& task) 134 { 135 FiberContext *fiberContext = task.awaitingFiber; 136 if (fiberContext) 137 { 138 task.awaitingFiber = nullptr; 139 return fiberContext; 140 } 141 142 MT::StackRequirements::Type stackRequirements = task.desc.stackRequirements; 143 144 fiberContext = nullptr; 145 bool res = false; 146 MT_USED_IN_ASSERT(res); 147 switch(stackRequirements) 148 { 149 case MT::StackRequirements::STANDARD: 150 res = standartFibersAvailable.TryPop(fiberContext); 151 MT_ASSERT(res, "Can't get more standard fibers!"); 152 break; 153 case MT::StackRequirements::EXTENDED: 154 res = extendedFibersAvailable.TryPop(fiberContext); 155 MT_ASSERT(res, "Can't get more extended fibers!"); 156 break; 157 default: 158 MT_REPORT_ASSERT("Unknown stack requrements"); 159 } 160 161 MT_ASSERT(fiberContext != nullptr, "Can't get more fibers. Too many tasks in flight simultaneously?"); 162 163 fiberContext->currentTask = task.desc; 164 fiberContext->currentGroup = task.group; 165 fiberContext->parentFiber = task.parentFiber; 166 fiberContext->stackRequirements = stackRequirements; 167 return fiberContext; 168 } 169 170 void TaskScheduler::ReleaseFiberContext(FiberContext*&& fiberContext) 171 { 172 MT_ASSERT(fiberContext, "Can't release nullptr Fiber. fiberContext is nullptr"); 173 174 MT::StackRequirements::Type stackRequirements = fiberContext->stackRequirements; 175 fiberContext->Reset(); 176 177 MT_ASSERT(fiberContext != nullptr, "Fiber context can't be nullptr"); 178 179 bool res = false; 180 MT_USED_IN_ASSERT(res); 181 switch(stackRequirements) 182 { 183 case MT::StackRequirements::STANDARD: 184 res = standartFibersAvailable.TryPush(std::move(fiberContext)); 185 break; 186 case MT::StackRequirements::EXTENDED: 187 res = extendedFibersAvailable.TryPush(std::move(fiberContext)); 188 break; 189 default: 190 MT_REPORT_ASSERT("Unknown stack requrements"); 191 } 192 193 MT_USED_IN_ASSERT(res); 194 MT_ASSERT(res != false, "Can't return fiber to storage"); 195 } 196 197 FiberContext* TaskScheduler::ExecuteTask(internal::ThreadContext& threadContext, FiberContext* fiberContext) 198 { 199 MT_ASSERT(threadContext.threadId.IsCurrentThread(), "Thread context sanity check failed"); 200 201 MT_ASSERT(fiberContext, "Invalid fiber context"); 202 MT_ASSERT(fiberContext->currentTask.IsValid(), "Invalid task"); 203 204 // Set actual thread context to fiber 205 fiberContext->SetThreadContext(&threadContext); 206 207 // Update task status 208 fiberContext->SetStatus(FiberTaskStatus::RUNNED); 209 210 MT_ASSERT(fiberContext->GetThreadContext()->threadId.IsCurrentThread(), "Thread context sanity check failed"); 211 212 const void* poolUserData = fiberContext->currentTask.userData; 213 TPoolTaskDestroy poolDestroyFunc = fiberContext->currentTask.poolDestroyFunc; 214 215 #ifdef MT_INSTRUMENTED_BUILD 216 //threadContext.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::SUSPEND); 217 threadContext.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::STOP); 218 #endif 219 220 // Run current task code 221 Fiber::SwitchTo(threadContext.schedulerFiber, fiberContext->fiber); 222 223 #ifdef MT_INSTRUMENTED_BUILD 224 //threadContext.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::RESUME); 225 threadContext.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::START); 226 #endif 227 228 // If task was done 229 FiberTaskStatus::Type taskStatus = fiberContext->GetStatus(); 230 if (taskStatus == FiberTaskStatus::FINISHED) 231 { 232 //destroy task (call dtor) for "fire and forget" type of task from TaskPool 233 if (poolDestroyFunc != nullptr) 234 { 235 poolDestroyFunc(poolUserData); 236 } 237 238 TaskGroup taskGroup = fiberContext->currentGroup; 239 240 TaskScheduler::TaskGroupDescription & groupDesc = threadContext.taskScheduler->GetGroupDesc(taskGroup); 241 242 // Update group status 243 int groupTaskCount = groupDesc.Dec(); 244 MT_ASSERT(groupTaskCount >= 0, "Sanity check failed!"); 245 if (groupTaskCount == 0) 246 { 247 fiberContext->currentGroup = TaskGroup::INVALID; 248 } 249 250 // Update total task count 251 int allGroupTaskCount = threadContext.taskScheduler->allGroups.Dec(); 252 MT_USED_IN_ASSERT(allGroupTaskCount); 253 MT_ASSERT(allGroupTaskCount >= 0, "Sanity check failed!"); 254 255 FiberContext* parentFiberContext = fiberContext->parentFiber; 256 if (parentFiberContext != nullptr) 257 { 258 int childrenFibersCount = parentFiberContext->childrenFibersCount.DecFetch(); 259 MT_ASSERT(childrenFibersCount >= 0, "Sanity check failed!"); 260 261 if (childrenFibersCount == 0) 262 { 263 // This is a last subtask. Restore parent task 264 MT_ASSERT(threadContext.threadId.IsCurrentThread(), "Thread context sanity check failed"); 265 MT_ASSERT(parentFiberContext->GetThreadContext() == nullptr, "Inactive parent should not have a valid thread context"); 266 267 // WARNING!! Thread context can changed here! Set actual current thread context. 268 parentFiberContext->SetThreadContext(&threadContext); 269 270 MT_ASSERT(parentFiberContext->GetThreadContext()->threadId.IsCurrentThread(), "Thread context sanity check failed"); 271 272 // All subtasks is done. 273 // Exiting and return parent fiber to scheduler 274 return parentFiberContext; 275 } else 276 { 277 // Other subtasks still exist 278 // Exiting 279 return nullptr; 280 } 281 } else 282 { 283 // Task is finished and no parent task 284 // Exiting 285 return nullptr; 286 } 287 } 288 289 MT_ASSERT(taskStatus != FiberTaskStatus::RUNNED, "Incorrect task status") 290 return nullptr; 291 } 292 293 294 void TaskScheduler::FiberMain(void* userData) 295 { 296 FiberContext& fiberContext = *(FiberContext*)(userData); 297 for(;;) 298 { 299 MT_ASSERT(fiberContext.currentTask.IsValid(), "Invalid task in fiber context"); 300 MT_ASSERT(fiberContext.GetThreadContext(), "Invalid thread context"); 301 MT_ASSERT(fiberContext.GetThreadContext()->threadId.IsCurrentThread(), "Thread context sanity check failed"); 302 303 #ifdef MT_INSTRUMENTED_BUILD 304 fiberContext.fiber.SetName( MT_SYSTEM_TASK_FIBER_NAME ); 305 fiberContext.GetThreadContext()->NotifyTaskExecuteStateChanged( fiberContext.currentTask.debugColor, fiberContext.currentTask.debugID, TaskExecuteState::START ); 306 #endif 307 308 fiberContext.currentTask.taskFunc( fiberContext, fiberContext.currentTask.userData ); 309 fiberContext.SetStatus(FiberTaskStatus::FINISHED); 310 311 #ifdef MT_INSTRUMENTED_BUILD 312 fiberContext.fiber.SetName( MT_SYSTEM_TASK_FIBER_NAME ); 313 fiberContext.GetThreadContext()->NotifyTaskExecuteStateChanged( fiberContext.currentTask.debugColor, fiberContext.currentTask.debugID, TaskExecuteState::STOP ); 314 #endif 315 316 Fiber::SwitchTo(fiberContext.fiber, fiberContext.GetThreadContext()->schedulerFiber); 317 } 318 319 } 320 321 322 bool TaskScheduler::TryStealTask(internal::ThreadContext& threadContext, internal::GroupedTask & task) 323 { 324 bool taskStealingDisabled = threadContext.taskScheduler->IsTaskStealingDisabled(); 325 uint32 workersCount = threadContext.taskScheduler->GetWorkersCount(); 326 327 if (workersCount <= 1 || taskStealingDisabled ) 328 { 329 return false; 330 } 331 332 uint32 victimIndex = threadContext.random.Get(); 333 334 for (uint32 attempt = 0; attempt < workersCount; attempt++) 335 { 336 uint32 index = victimIndex % workersCount; 337 if (index == threadContext.workerIndex) 338 { 339 victimIndex++; 340 index = victimIndex % workersCount; 341 } 342 343 internal::ThreadContext& victimContext = threadContext.taskScheduler->threadContext[index]; 344 if (victimContext.queue.TryPopNewest(task)) 345 { 346 return true; 347 } 348 349 victimIndex++; 350 } 351 return false; 352 } 353 354 void TaskScheduler::WorkerThreadMain( void* userData ) 355 { 356 internal::ThreadContext& context = *(internal::ThreadContext*)(userData); 357 MT_ASSERT(context.taskScheduler, "Task scheduler must be not null!"); 358 359 isWorkerThreadTLS = 1; 360 context.threadId.SetAsCurrentThread(); 361 362 #ifdef MT_INSTRUMENTED_BUILD 363 const char* threadNames[] = {"worker0","worker1","worker2","worker3","worker4","worker5","worker6","worker7","worker8","worker9","worker10","worker11","worker12"}; 364 if (context.workerIndex < MT_ARRAY_SIZE(threadNames)) 365 { 366 Thread::SetThreadName(threadNames[context.workerIndex]); 367 } else 368 { 369 Thread::SetThreadName("worker_thread"); 370 } 371 #endif 372 373 context.schedulerFiber.CreateFromCurrentThreadAndRun(SchedulerFiberMain, userData); 374 } 375 376 377 void TaskScheduler::SchedulerFiberWait( void* userData ) 378 { 379 WaitContext& waitContext = *(WaitContext*)(userData); 380 internal::ThreadContext& context = *waitContext.threadContext; 381 MT_ASSERT(context.taskScheduler, "Task scheduler must be not null!"); 382 MT_ASSERT(waitContext.waitCounter, "Wait counter must be not null!"); 383 384 #ifdef MT_INSTRUMENTED_BUILD 385 context.NotifyWaitStarted(); 386 context.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::START); 387 #endif 388 389 bool isTaskStealingDisabled = context.taskScheduler->IsTaskStealingDisabled(); 390 391 int64 timeOut = GetTimeMicroSeconds() + (waitContext.waitTimeMs * 1000); 392 393 SpinWait spinWait; 394 395 for(;;) 396 { 397 if ( SchedulerFiberStep(context, isTaskStealingDisabled) == false ) 398 { 399 spinWait.SpinOnce(); 400 } else 401 { 402 spinWait.Reset(); 403 } 404 405 int32 groupTaskCount = waitContext.waitCounter->Load(); 406 if (groupTaskCount == 0) 407 { 408 waitContext.exitCode = 0; 409 break; 410 } 411 412 int64 timeNow = GetTimeMicroSeconds(); 413 if (timeNow >= timeOut) 414 { 415 waitContext.exitCode = 1; 416 break; 417 } 418 } 419 420 #ifdef MT_INSTRUMENTED_BUILD 421 context.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::STOP); 422 context.NotifyWaitFinished(); 423 #endif 424 425 } 426 427 void TaskScheduler::SchedulerFiberMain( void* userData ) 428 { 429 internal::ThreadContext& context = *(internal::ThreadContext*)(userData); 430 MT_ASSERT(context.taskScheduler, "Task scheduler must be not null!"); 431 432 #ifdef MT_INSTRUMENTED_BUILD 433 context.NotifyThreadCreated(context.workerIndex); 434 #endif 435 436 int32 totalThreadsCount = context.taskScheduler->threadsCount.LoadRelaxed(); 437 context.taskScheduler->startedThreadsCount.IncFetch(); 438 439 //Simple spinlock until all threads is started and initialized 440 for(;;) 441 { 442 int32 initializedThreadsCount = context.taskScheduler->startedThreadsCount.Load(); 443 if (initializedThreadsCount == totalThreadsCount) 444 { 445 break; 446 } 447 448 // sleep some time until all other thread initialized 449 Thread::Sleep(1); 450 } 451 452 HardwareFullMemoryBarrier(); 453 454 #ifdef MT_INSTRUMENTED_BUILD 455 context.NotifyThreadStarted(context.workerIndex); 456 context.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::START); 457 #endif 458 bool isTaskStealingDisabled = context.taskScheduler->IsTaskStealingDisabled(); 459 460 while(context.state.Load() != internal::ThreadState::EXIT) 461 { 462 if ( SchedulerFiberStep(context, isTaskStealingDisabled) == false) 463 { 464 #ifdef MT_INSTRUMENTED_BUILD 465 context.NotifyThreadIdleStarted(context.workerIndex); 466 #endif 467 468 #if MT_LOW_LATENCY_EXPERIMENTAL_WAIT 469 470 SpinWait spinWait; 471 472 for(;;) 473 { 474 // Queue is empty and stealing attempt has failed. 475 // Fast Spin Wait for new tasks 476 if (spinWait.SpinOnce() >= SpinWait::YIELD_SLEEP0_THRESHOLD) 477 { 478 // Fast Spin wait for new tasks has failed. 479 // Wait for new events using events 480 context.hasNewTasksEvent.Wait(20000); 481 482 spinWait.Reset(); 483 484 #ifdef MT_INSTRUMENTED_BUILD 485 context.NotifyThreadIdleFinished(context.workerIndex); 486 #endif 487 488 break; 489 } 490 491 internal::GroupedTask task; 492 if ( context.queue.TryPopOldest(task) ) 493 { 494 #ifdef MT_INSTRUMENTED_BUILD 495 context.NotifyThreadIdleFinished(context.workerIndex); 496 #endif 497 498 SchedulerFiberProcessTask(context, task); 499 500 break; 501 } 502 503 } 504 #else 505 // Queue is empty and stealing attempt has failed. 506 // Wait for new events using events 507 context.hasNewTasksEvent.Wait(20000); 508 509 #ifdef MT_INSTRUMENTED_BUILD 510 context.NotifyThreadIdleFinished(context.workerIndex); 511 #endif 512 513 #endif 514 515 } 516 517 } // main thread loop 518 519 #ifdef MT_INSTRUMENTED_BUILD 520 context.NotifyTaskExecuteStateChanged( MT_SYSTEM_TASK_COLOR, MT_SYSTEM_TASK_NAME, TaskExecuteState::STOP); 521 context.NotifyThreadStoped(context.workerIndex); 522 #endif 523 524 } 525 526 void TaskScheduler::SchedulerFiberProcessTask( internal::ThreadContext& context, internal::GroupedTask& task ) 527 { 528 #ifdef MT_INSTRUMENTED_BUILD 529 bool isNewTask = (task.awaitingFiber == nullptr); 530 #endif 531 532 // There is a new task 533 FiberContext* fiberContext = context.taskScheduler->RequestFiberContext(task); 534 MT_ASSERT(fiberContext, "Can't get execution context from pool"); 535 MT_ASSERT(fiberContext->currentTask.IsValid(), "Sanity check failed"); 536 MT_ASSERT(fiberContext->stackRequirements == task.desc.stackRequirements, "Sanity check failed"); 537 538 while(fiberContext) 539 { 540 #ifdef MT_INSTRUMENTED_BUILD 541 if (isNewTask) 542 { 543 //TODO: 544 isNewTask = false; 545 } 546 #endif 547 // prevent invalid fiber resume from child tasks, before ExecuteTask is done 548 fiberContext->childrenFibersCount.IncFetch(); 549 550 FiberContext* parentFiber = ExecuteTask(context, fiberContext); 551 552 FiberTaskStatus::Type taskStatus = fiberContext->GetStatus(); 553 554 //release guard 555 int childrenFibersCount = fiberContext->childrenFibersCount.DecFetch(); 556 557 // Can drop fiber context - task is finished 558 if (taskStatus == FiberTaskStatus::FINISHED) 559 { 560 MT_ASSERT( childrenFibersCount == 0, "Sanity check failed"); 561 context.taskScheduler->ReleaseFiberContext(std::move(fiberContext)); 562 563 // If parent fiber is exist transfer flow control to parent fiber, if parent fiber is null, exit 564 fiberContext = parentFiber; 565 } else 566 { 567 MT_ASSERT( childrenFibersCount >= 0, "Sanity check failed"); 568 569 // No subtasks here and status is not finished, this mean all subtasks already finished before parent return from ExecuteTask 570 if (childrenFibersCount == 0) 571 { 572 MT_ASSERT(parentFiber == nullptr, "Sanity check failed"); 573 } else 574 { 575 // If subtasks still exist, drop current task execution. task will be resumed when last subtask finished 576 break; 577 } 578 579 // If task is yielded execution, get another task from queue. 580 if (taskStatus == FiberTaskStatus::YIELDED) 581 { 582 // Task is yielded, add to tasks queue 583 ArrayView<internal::GroupedTask> buffer(context.descBuffer, 1); 584 ArrayView<internal::TaskBucket> buckets( MT_ALLOCATE_ON_STACK(sizeof(internal::TaskBucket)), 1 ); 585 586 FiberContext* yieldedTask = fiberContext; 587 StaticVector<FiberContext*, 1> yieldedTasksQueue(1, yieldedTask); 588 internal::DistibuteDescriptions( TaskGroup(TaskGroup::ASSIGN_FROM_CONTEXT), yieldedTasksQueue.Begin(), buffer, buckets ); 589 590 // add yielded task to scheduler 591 context.taskScheduler->RunTasksImpl(buckets, nullptr, true); 592 593 // ATENTION! yielded task can be already completed at this point 594 595 break; 596 } 597 } 598 } //while(fiberContext) 599 } 600 601 bool TaskScheduler::SchedulerFiberStep( internal::ThreadContext& context, bool disableTaskStealing) 602 { 603 internal::GroupedTask task; 604 if ( context.queue.TryPopOldest(task) || (disableTaskStealing == false && TryStealTask(context, task) ) ) 605 { 606 SchedulerFiberProcessTask(context, task); 607 return true; 608 } 609 610 return false; 611 } 612 613 void TaskScheduler::RunTasksImpl(ArrayView<internal::TaskBucket>& buckets, FiberContext * parentFiber, bool restoredFromAwaitState) 614 { 615 616 #if MT_LOW_LATENCY_EXPERIMENTAL_WAIT 617 // Early wakeup worker threads (worker thread spin wait for some time before sleep) 618 int32 roundRobinIndex = roundRobinThreadIndex.LoadRelaxed(); 619 for (size_t i = 0; i < buckets.Size(); ++i) 620 { 621 int bucketIndex = ((roundRobinIndex + i) % threadsCount.LoadRelaxed()); 622 internal::ThreadContext & context = threadContext[bucketIndex]; 623 context.hasNewTasksEvent.Signal(); 624 } 625 #endif 626 627 628 // This storage is necessary to calculate how many tasks we add to different groups 629 int newTaskCountInGroup[TaskGroup::MT_MAX_GROUPS_COUNT]; 630 631 // Default value is 0 632 memset(&newTaskCountInGroup[0], 0, MT_ARRAY_SIZE(newTaskCountInGroup)); 633 634 // Set parent fiber pointer 635 // Calculate the number of tasks per group 636 // Calculate total number of tasks 637 size_t count = 0; 638 for (size_t i = 0; i < buckets.Size(); ++i) 639 { 640 internal::TaskBucket& bucket = buckets[i]; 641 for (size_t taskIndex = 0; taskIndex < bucket.count; taskIndex++) 642 { 643 internal::GroupedTask & task = bucket.tasks[taskIndex]; 644 645 task.parentFiber = parentFiber; 646 647 int idx = task.group.GetValidIndex(); 648 MT_ASSERT(idx >= 0 && idx < TaskGroup::MT_MAX_GROUPS_COUNT, "Invalid index"); 649 newTaskCountInGroup[idx]++; 650 } 651 652 count += bucket.count; 653 } 654 655 // Increments child fibers count on parent fiber 656 if (parentFiber) 657 { 658 parentFiber->childrenFibersCount.AddFetch((int)count); 659 } 660 661 if (restoredFromAwaitState == false) 662 { 663 // Increase the number of active tasks in the group using data from temporary storage 664 for (size_t i = 0; i < TaskGroup::MT_MAX_GROUPS_COUNT; i++) 665 { 666 int groupNewTaskCount = newTaskCountInGroup[i]; 667 if (groupNewTaskCount > 0) 668 { 669 groupStats[i].Add((uint32)groupNewTaskCount); 670 } 671 } 672 673 // Increments all task in progress counter 674 allGroups.Add((uint32)count); 675 } else 676 { 677 // If task's restored from await state, counters already in correct state 678 } 679 680 // Add to thread queue 681 for (size_t i = 0; i < buckets.Size(); ++i) 682 { 683 int bucketIndex = roundRobinThreadIndex.IncFetch() % threadsCount.LoadRelaxed(); 684 internal::ThreadContext & context = threadContext[bucketIndex]; 685 686 internal::TaskBucket& bucket = buckets[i]; 687 688 for(;;) 689 { 690 MT_ASSERT(bucket.count < (internal::TASK_BUFFER_CAPACITY - 1), "Sanity check failed. Too many tasks per one bucket."); 691 692 bool res = context.queue.Add(bucket.tasks, bucket.count); 693 if (res == true) 694 { 695 break; 696 } 697 698 //Can't add new tasks onto the queue. Look like the job system is overloaded. Wait some time and try again. 699 //TODO: implement waiting until workers done using events. 700 Thread::Sleep(10); 701 } 702 703 context.hasNewTasksEvent.Signal(); 704 } 705 } 706 707 void TaskScheduler::RunAsync(TaskGroup group, const TaskHandle* taskHandleArray, uint32 taskHandleCount) 708 { 709 MT_ASSERT(!IsWorkerThread(), "Can't use RunAsync inside Task. Use FiberContext.RunAsync() instead."); 710 711 ArrayView<internal::GroupedTask> buffer(MT_ALLOCATE_ON_STACK(sizeof(internal::GroupedTask) * taskHandleCount), taskHandleCount); 712 713 uint32 bucketCount = MT::Min((uint32)GetWorkersCount(), taskHandleCount); 714 ArrayView<internal::TaskBucket> buckets(MT_ALLOCATE_ON_STACK(sizeof(internal::TaskBucket) * bucketCount), bucketCount); 715 716 internal::DistibuteDescriptions(group, taskHandleArray, buffer, buckets); 717 RunTasksImpl(buckets, nullptr, false); 718 } 719 720 bool TaskScheduler::WaitGroup(TaskGroup group, uint32 milliseconds) 721 { 722 MT_VERIFY(IsWorkerThread() == false, "Can't use WaitGroup inside Task. Use FiberContext.WaitGroupAndYield() instead.", return false); 723 724 TaskScheduler::TaskGroupDescription& groupDesc = GetGroupDesc(group); 725 726 // Early exit if not tasks in group 727 int32 taskCount = groupDesc.GetTaskCount(); 728 if (taskCount == 0) 729 { 730 return true; 731 } 732 733 size_t bytesCountForDescBuffer = internal::ThreadContext::GetMemoryRequrementInBytesForDescBuffer(); 734 void* descBuffer = MT_ALLOCATE_ON_STACK(bytesCountForDescBuffer); 735 736 internal::ThreadContext context(descBuffer); 737 context.taskScheduler = this; 738 context.SetThreadIndex(0xFFFFFFFF); 739 context.threadId.SetAsCurrentThread(); 740 741 WaitContext waitContext; 742 waitContext.threadContext = &context; 743 waitContext.waitCounter = groupDesc.GetWaitCounter(); 744 waitContext.waitTimeMs = milliseconds; 745 waitContext.exitCode = 0; 746 747 isWorkerThreadTLS = 1; 748 749 context.schedulerFiber.CreateFromCurrentThreadAndRun(SchedulerFiberWait, &waitContext); 750 751 isWorkerThreadTLS = 0; 752 return (waitContext.exitCode == 0); 753 } 754 755 bool TaskScheduler::WaitAll(uint32 milliseconds) 756 { 757 MT_VERIFY(IsWorkerThread() == false, "Can't use WaitAll inside Task.", return false); 758 759 // Early exit if not tasks in group 760 int32 taskCount = allGroups.GetTaskCount(); 761 if (taskCount == 0) 762 { 763 return true; 764 } 765 766 size_t bytesCountForDescBuffer = internal::ThreadContext::GetMemoryRequrementInBytesForDescBuffer(); 767 void* descBuffer = MT_ALLOCATE_ON_STACK(bytesCountForDescBuffer); 768 769 internal::ThreadContext context(descBuffer); 770 context.taskScheduler = this; 771 context.SetThreadIndex(0xFFFFFFFF); 772 context.threadId.SetAsCurrentThread(); 773 774 WaitContext waitContext; 775 waitContext.threadContext = &context; 776 waitContext.waitCounter = allGroups.GetWaitCounter(); 777 waitContext.waitTimeMs = milliseconds; 778 waitContext.exitCode = 0; 779 780 isWorkerThreadTLS = 1; 781 782 context.schedulerFiber.CreateFromCurrentThreadAndRun(SchedulerFiberWait, &waitContext); 783 784 isWorkerThreadTLS = 0; 785 return (waitContext.exitCode == 0); 786 } 787 788 bool TaskScheduler::IsTaskStealingDisabled() const 789 { 790 if (threadsCount.LoadRelaxed() <= 1) 791 { 792 return true; 793 } 794 795 return taskStealingDisabled; 796 } 797 798 int32 TaskScheduler::GetWorkersCount() const 799 { 800 return threadsCount.LoadRelaxed(); 801 } 802 803 804 bool TaskScheduler::IsWorkerThread() const 805 { 806 return (isWorkerThreadTLS != 0); 807 } 808 809 TaskGroup TaskScheduler::CreateGroup() 810 { 811 MT_ASSERT(IsWorkerThread() == false, "Can't use CreateGroup inside Task."); 812 813 TaskGroup group; 814 if (!availableGroups.TryPop(group)) 815 { 816 MT_REPORT_ASSERT("Group pool is empty"); 817 } 818 819 int idx = group.GetValidIndex(); 820 MT_USED_IN_ASSERT(idx); 821 MT_ASSERT(groupStats[idx].GetDebugIsFree() == true, "Bad logic!"); 822 #if MT_GROUP_DEBUG 823 groupStats[idx].SetDebugIsFree(false); 824 #endif 825 826 return group; 827 } 828 829 void TaskScheduler::ReleaseGroup(TaskGroup group) 830 { 831 MT_ASSERT(IsWorkerThread() == false, "Can't use ReleaseGroup inside Task."); 832 MT_ASSERT(group.IsValid(), "Invalid group ID"); 833 834 int idx = group.GetValidIndex(); 835 MT_USED_IN_ASSERT(idx); 836 MT_ASSERT(groupStats[idx].GetDebugIsFree() == false, "Group already released"); 837 #if MT_GROUP_DEBUG 838 groupStats[idx].SetDebugIsFree(true); 839 #endif 840 841 bool res = availableGroups.TryPush(std::move(group)); 842 MT_USED_IN_ASSERT(res); 843 MT_ASSERT(res, "Can't return group to pool"); 844 } 845 846 TaskScheduler::TaskGroupDescription & TaskScheduler::GetGroupDesc(TaskGroup group) 847 { 848 MT_ASSERT(group.IsValid(), "Invalid group ID"); 849 850 int idx = group.GetValidIndex(); 851 TaskScheduler::TaskGroupDescription & groupDesc = groupStats[idx]; 852 853 MT_ASSERT(groupDesc.GetDebugIsFree() == false, "Invalid group"); 854 return groupDesc; 855 } 856 } 857 858 859