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