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