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