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