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