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