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