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