dali/internal/system/common/async-task-manager-impl.cpp

   1 /*
   2  * Copyright (c) 2023 Samsung Electronics Co., Ltd.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  *
  16  */
  17
  18 // CLASS HEADER
  19 #include <dali/internal/system/common/async-task-manager-impl.h>
  20
  21 // EXTERNAL INCLUDES
  22 #include <dali/devel-api/adaptor-framework/environment-variable.h>
  23 #include <dali/devel-api/adaptor-framework/thread-settings.h>
  24 #include <dali/devel-api/common/singleton-service.h>
  25 #include <dali/integration-api/adaptor-framework/adaptor.h>
  26 #include <dali/integration-api/debug.h>
  27
  28 #include <unordered_map>
  29
  30 namespace Dali
  31 {
  32 namespace Internal
  33 {
  34 namespace Adaptor
  35 {
  36 namespace
  37 {
  38 constexpr auto FORCE_TRIGGER_THRESHOLD = 128u; ///< Trigger TasksCompleted() forcely if the number of completed task contain too much.
  39
  40 constexpr auto DEFAULT_NUMBER_OF_ASYNC_THREADS = size_t{8u};
  41 constexpr auto NUMBER_OF_ASYNC_THREADS_ENV     = "DALI_ASYNC_MANAGER_THREAD_POOL_SIZE";
  42
  43 // The number of threads for low priority task.
  44 constexpr auto DEFAULT_NUMBER_OF_LOW_PRIORITY_THREADS = size_t{6u};
  45 constexpr auto NUMBER_OF_LOW_PRIORITY_THREADS_ENV     = "DALI_ASYNC_MANAGER_LOW_PRIORITY_THREAD_POOL_SIZE";
  46
  47 size_t GetNumberOfThreads(const char* environmentVariable, size_t defaultValue)
  48 {
  49   auto           numberString          = EnvironmentVariable::GetEnvironmentVariable(environmentVariable);
  50   auto           numberOfThreads       = numberString ? std::strtoul(numberString, nullptr, 10) : 0;
  51   constexpr auto MAX_NUMBER_OF_THREADS = 16u;
  52   DALI_ASSERT_DEBUG(numberOfThreads <= MAX_NUMBER_OF_THREADS);
  53   return (numberOfThreads > 0 && numberOfThreads <= MAX_NUMBER_OF_THREADS) ? numberOfThreads : defaultValue;
  54 }
  55
  56 size_t GetNumberOfLowPriorityThreads(const char* environmentVariable, size_t defaultValue, size_t maxValue)
  57 {
  58   auto numberString    = EnvironmentVariable::GetEnvironmentVariable(environmentVariable);
  59   auto numberOfThreads = numberString ? std::strtoul(numberString, nullptr, 10) : 0;
  60   DALI_ASSERT_DEBUG(numberOfThreads <= maxValue);
  61   return (numberOfThreads > 0 && numberOfThreads <= maxValue) ? numberOfThreads : std::min(defaultValue, maxValue);
  62 }
  63
  64 #if defined(DEBUG_ENABLED)
  65 Debug::Filter* gAsyncTasksManagerLogFilter = Debug::Filter::New(Debug::NoLogging, false, "LOG_ASYNC_TASK_MANAGER");
  66
  67 uint32_t gThreadId = 0u; // Only for debug
  68 #endif
  69
  70 } // unnamed namespace
  71
  72 // AsyncTaskThread
  73
  74 AsyncTaskThread::AsyncTaskThread(AsyncTaskManager& asyncTaskManager)
  75 : mConditionalWait(),
  76   mAsyncTaskManager(asyncTaskManager),
  77   mLogFactory(Dali::Adaptor::Get().GetLogFactory()),
  78   mTraceFactory(Dali::Adaptor::Get().GetTraceFactory()),
  79   mDestroyThread(false),
  80   mIsThreadStarted(false),
  81   mIsThreadIdle(true)
  82 {
  83 }
  84
  85 AsyncTaskThread::~AsyncTaskThread()
  86 {
  87   // Stop the thread
  88   {
  89     ConditionalWait::ScopedLock lock(mConditionalWait);
  90     mDestroyThread = true;
  91     mConditionalWait.Notify(lock);
  92   }
  93
  94   Join();
  95 }
  96
  97 bool AsyncTaskThread::Request()
  98 {
  99   if(!mIsThreadStarted)
 100   {
 101     Start();
 102     mIsThreadStarted = true;
 103   }
 104
 105   {
 106     // Lock while adding task to the queue
 107     ConditionalWait::ScopedLock lock(mConditionalWait);
 108
 109     if(mIsThreadIdle)
 110     {
 111       mIsThreadIdle = false;
 112
 113       // wake up the thread
 114       mConditionalWait.Notify(lock);
 115       return true;
 116     }
 117   }
 118
 119   return false;
 120 }
 121
 122 void AsyncTaskThread::Run()
 123 {
 124 #if defined(DEBUG_ENABLED)
 125   uint32_t threadId = gThreadId++;
 126   {
 127     char temp[100];
 128     snprintf(temp, 100, "AsyncTaskThread[%u]", threadId);
 129     SetThreadName(temp);
 130   }
 131 #else
 132   SetThreadName("AsyncTaskThread");
 133 #endif
 134   mLogFactory.InstallLogFunction();
 135   mTraceFactory.InstallTraceFunction();
 136
 137   while(!mDestroyThread)
 138   {
 139     AsyncTaskPtr task = mAsyncTaskManager.PopNextTaskToProcess();
 140     if(!task)
 141     {
 142       ConditionalWait::ScopedLock lock(mConditionalWait);
 143       if(!mDestroyThread)
 144       {
 145         mIsThreadIdle = true;
 146         DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Thread[%u] wait\n", threadId);
 147         mConditionalWait.Wait(lock);
 148         DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Thread[%u] awake\n", threadId);
 149       }
 150     }
 151     else
 152     {
 153       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Thread[%u] Process task [%p]\n", threadId, task.Get());
 154       task->Process();
 155       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Thread[%u] Complete task [%p]\n", threadId, task.Get());
 156       if(!mDestroyThread)
 157       {
 158         mAsyncTaskManager.CompleteTask(std::move(task));
 159       }
 160     }
 161   }
 162 }
 163
 164 // AsyncTaskManager::CacheImpl
 165
 166 struct AsyncTaskManager::CacheImpl
 167 {
 168   CacheImpl(AsyncTaskManager& manager)
 169   : mManager(manager)
 170   {
 171   }
 172
 173 public:
 174   // Insert / Erase task cache API.
 175
 176   /**
 177    * @brief Insert cache that input task.
 178    * @pre Mutex be locked.
 179    */
 180   template<typename CacheContainer, typename Iterator>
 181   static void InsertTaskCache(CacheContainer& cacheMap, AsyncTaskPtr task, Iterator iterator)
 182   {
 183     auto& cacheContainer = cacheMap[task.Get()]; // Get or Create cache container.
 184     cacheContainer.insert(cacheContainer.end(), iterator);
 185   }
 186
 187   /**
 188    * @brief Erase cache that input task.
 189    * @pre Mutex be locked.
 190    */
 191   template<typename CacheContainer, typename Iterator>
 192   static void EraseTaskCache(CacheContainer& cacheMap, AsyncTaskPtr task, Iterator iterator)
 193   {
 194     auto mapIter = cacheMap.find(task.Get());
 195     if(mapIter != cacheMap.end())
 196     {
 197       auto& cacheContainer = (*mapIter).second;
 198       auto  cacheIter      = std::find(cacheContainer.begin(), cacheContainer.end(), iterator);
 199
 200       if(cacheIter != cacheContainer.end())
 201       {
 202         cacheContainer.erase(cacheIter);
 203         if(cacheContainer.empty())
 204         {
 205           cacheMap.erase(mapIter);
 206         }
 207       }
 208     }
 209   }
 210
 211   /**
 212    * @brief Erase all cache that input task.
 213    * @pre Mutex be locked.
 214    */
 215   template<typename CacheContainer>
 216   static void EraseAllTaskCache(CacheContainer& cacheMap, AsyncTaskPtr task)
 217   {
 218     auto mapIter = cacheMap.find(task.Get());
 219     if(mapIter != cacheMap.end())
 220     {
 221       cacheMap.erase(mapIter);
 222     }
 223   }
 224
 225 public:
 226   AsyncTaskManager& mManager; ///< Owner of this CacheImpl.
 227
 228   // Keep cache iterators as list since we take tasks by FIFO as default.
 229   using TaskCacheContainer          = std::unordered_map<const AsyncTask*, std::list<AsyncTaskContainer::iterator>>;
 230   using RunningTaskCacheContainer   = std::unordered_map<const AsyncTask*, std::list<AsyncRunningTaskContainer::iterator>>;
 231   using CompletedTaskCacheContainer = std::unordered_map<const AsyncTask*, std::list<AsyncCompletedTaskContainer::iterator>>;
 232
 233   TaskCacheContainer          mWaitingTasksCache;   ///< The cache of tasks and iterator for waiting to async process. Must be locked under mWaitingTasksMutex.
 234   RunningTaskCacheContainer   mRunningTasksCache;   ///< The cache of tasks and iterator for running tasks. Must be locked under mRunningTasksMutex.
 235   CompletedTaskCacheContainer mCompletedTasksCache; ///< The cache of tasks and iterator for completed async process. Must be locked under mCompletedTasksMutex.
 236 };
 237
 238 // AsyncTaskManager
 239
 240 Dali::AsyncTaskManager AsyncTaskManager::Get()
 241 {
 242   Dali::AsyncTaskManager manager;
 243   SingletonService       singletonService(SingletonService::Get());
 244   if(singletonService)
 245   {
 246     // Check whether the async task manager is already created
 247     Dali::BaseHandle handle = singletonService.GetSingleton(typeid(Dali::AsyncTaskManager));
 248     if(handle)
 249     {
 250       // If so, downcast the handle of singleton
 251       manager = Dali::AsyncTaskManager(dynamic_cast<Internal::Adaptor::AsyncTaskManager*>(handle.GetObjectPtr()));
 252     }
 253
 254     if(!manager)
 255     {
 256       // If not, create the async task manager and register it as a singleton
 257       Internal::Adaptor::AsyncTaskManager* internalAsyncTaskManager = new Internal::Adaptor::AsyncTaskManager();
 258       manager                                                       = Dali::AsyncTaskManager(internalAsyncTaskManager);
 259       singletonService.Register(typeid(manager), manager);
 260     }
 261   }
 262   return manager;
 263 }
 264
 265 AsyncTaskManager::AsyncTaskManager()
 266 : mTasks(GetNumberOfThreads(NUMBER_OF_ASYNC_THREADS_ENV, DEFAULT_NUMBER_OF_ASYNC_THREADS), [&]() { return TaskHelper(*this); }),
 267   mAvaliableLowPriorityTaskCounts(GetNumberOfLowPriorityThreads(NUMBER_OF_LOW_PRIORITY_THREADS_ENV, DEFAULT_NUMBER_OF_LOW_PRIORITY_THREADS, mTasks.GetElementCount())),
 268   mWaitingHighProirityTaskCounts(0u),
 269   mCacheImpl(new CacheImpl(*this)),
 270   mTrigger(new EventThreadCallback(MakeCallback(this, &AsyncTaskManager::TasksCompleted))),
 271   mProcessorRegistered(false)
 272 {
 273 }
 274
 275 AsyncTaskManager::~AsyncTaskManager()
 276 {
 277   if(mProcessorRegistered && Dali::Adaptor::IsAvailable())
 278   {
 279     mProcessorRegistered = false;
 280     Dali::Adaptor::Get().UnregisterProcessor(*this);
 281   }
 282
 283   // Join all threads.
 284   mTasks.Clear();
 285
 286   // Remove cache impl after all threads are join.
 287   mCacheImpl.reset();
 288
 289   // Remove tasks after CacheImpl removed
 290   mWaitingTasks.clear();
 291   mRunningTasks.clear();
 292   mCompletedTasks.clear();
 293 }
 294
 295 void AsyncTaskManager::AddTask(AsyncTaskPtr task)
 296 {
 297   if(task)
 298   {
 299     // Lock while adding task to the queue
 300     Mutex::ScopedLock lock(mWaitingTasksMutex);
 301
 302     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "AddTask [%p]\n", task.Get());
 303
 304     // push back into waiting queue.
 305     auto waitingIter = mWaitingTasks.insert(mWaitingTasks.end(), task);
 306     CacheImpl::InsertTaskCache(mCacheImpl->mWaitingTasksCache, task, waitingIter);
 307
 308     if(task->GetPriorityType() == AsyncTask::PriorityType::HIGH)
 309     {
 310       // Increase the number of waiting tasks for high priority.
 311       ++mWaitingHighProirityTaskCounts;
 312     }
 313
 314     {
 315       // For thread safety
 316       Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
 317
 318       // Finish all Running threads are working
 319       if(mRunningTasks.size() >= mTasks.GetElementCount())
 320       {
 321         return;
 322       }
 323     }
 324   }
 325
 326   size_t count = mTasks.GetElementCount();
 327   size_t index = 0;
 328   while(index++ < count)
 329   {
 330     auto processHelperIt = mTasks.GetNext();
 331     DALI_ASSERT_ALWAYS(processHelperIt != mTasks.End());
 332     if(processHelperIt->Request())
 333     {
 334       break;
 335     }
 336     // If all threads are busy, then it's ok just to push the task because they will try to get the next job.
 337   }
 338
 339   // Register Process (Since mTrigger execute too late timing if event thread running a lots of events.)
 340   if(!mProcessorRegistered && Dali::Adaptor::IsAvailable())
 341   {
 342     Dali::Adaptor::Get().RegisterProcessor(*this);
 343     mProcessorRegistered = true;
 344   }
 345
 346   return;
 347 }
 348
 349 void AsyncTaskManager::RemoveTask(AsyncTaskPtr task)
 350 {
 351   if(task)
 352   {
 353     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "RemoveTask [%p]\n", task.Get());
 354
 355     // Check whether we need to unregister processor.
 356     // If there is some non-empty queue exist, we don't need to unregister processor.
 357     bool needCheckUnregisterProcessor = true;
 358
 359     {
 360       // Lock while remove task from the queue
 361       Mutex::ScopedLock lock(mWaitingTasksMutex);
 362
 363       auto mapIter = mCacheImpl->mWaitingTasksCache.find(task.Get());
 364       if(mapIter != mCacheImpl->mWaitingTasksCache.end())
 365       {
 366         for(auto& iterator : mapIter->second)
 367         {
 368           DALI_ASSERT_DEBUG((*iterator) == task);
 369           if((*iterator)->GetPriorityType() == AsyncTask::PriorityType::HIGH)
 370           {
 371             // Decrease the number of waiting tasks for high priority.
 372             --mWaitingHighProirityTaskCounts;
 373           }
 374           mWaitingTasks.erase(iterator);
 375         }
 376         CacheImpl::EraseAllTaskCache(mCacheImpl->mWaitingTasksCache, task);
 377       }
 378
 379       if(!mWaitingTasks.empty())
 380       {
 381         needCheckUnregisterProcessor = false;
 382       }
 383     }
 384
 385     {
 386       // Lock while remove task from the queue
 387       Mutex::ScopedLock lock(mRunningTasksMutex);
 388
 389       auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
 390       if(mapIter != mCacheImpl->mRunningTasksCache.end())
 391       {
 392         for(auto& iterator : mapIter->second)
 393         {
 394           DALI_ASSERT_DEBUG((*iterator).first == task);
 395           // We cannot erase container. Just mark as canceled.
 396           // Note : mAvaliableLowPriorityTaskCounts will be increased after process finished.
 397           (*iterator).second = RunningTaskState::CANCELED;
 398         }
 399       }
 400
 401       if(!mRunningTasks.empty())
 402       {
 403         needCheckUnregisterProcessor = false;
 404       }
 405     }
 406
 407     {
 408       // Lock while remove task from the queue
 409       Mutex::ScopedLock lock(mCompletedTasksMutex);
 410
 411       auto mapIter = mCacheImpl->mCompletedTasksCache.find(task.Get());
 412       if(mapIter != mCacheImpl->mCompletedTasksCache.end())
 413       {
 414         for(auto& iterator : mapIter->second)
 415         {
 416           DALI_ASSERT_DEBUG(iterator->first == task);
 417           mCompletedTasks.erase(iterator);
 418         }
 419         CacheImpl::EraseAllTaskCache(mCacheImpl->mCompletedTasksCache, task);
 420       }
 421
 422       if(!mCompletedTasks.empty())
 423       {
 424         needCheckUnregisterProcessor = false;
 425       }
 426     }
 427
 428     // UnregisterProcessor required to lock mutex. Call this API only if required.
 429     if(needCheckUnregisterProcessor)
 430     {
 431       UnregisterProcessor();
 432     }
 433   }
 434 }
 435
 436 AsyncTaskPtr AsyncTaskManager::PopNextCompletedTask()
 437 {
 438   // Lock while popping task out from the queue
 439   Mutex::ScopedLock lock(mCompletedTasksMutex);
 440
 441   AsyncTaskPtr nextCompletedTask = nullptr;
 442
 443   while(!mCompletedTasks.empty())
 444   {
 445     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "PopNextCompletedTask, completed task count : [%zu]\n", mCompletedTasks.size());
 446
 447     auto               next      = mCompletedTasks.begin();
 448     AsyncTaskPtr       nextTask  = next->first;
 449     CompletedTaskState taskState = next->second;
 450     CacheImpl::EraseTaskCache(mCacheImpl->mCompletedTasksCache, nextTask, next);
 451     mCompletedTasks.erase(next);
 452
 453     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Completed task [%p] (callback required? : %d)\n", nextTask.Get(), taskState == CompletedTaskState::REQUIRE_CALLBACK);
 454
 455     if(taskState == CompletedTaskState::REQUIRE_CALLBACK)
 456     {
 457       nextCompletedTask = nextTask;
 458       break;
 459     }
 460   }
 461
 462   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Pickup completed [%p]\n", nextCompletedTask.Get());
 463
 464   return nextCompletedTask;
 465 }
 466
 467 void AsyncTaskManager::UnregisterProcessor()
 468 {
 469   if(mProcessorRegistered && Dali::Adaptor::IsAvailable())
 470   {
 471     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "UnregisterProcessor begin\n");
 472     // Keep processor at least 1 task exist.
 473     // Please be careful the order of mutex, to avoid dead lock.
 474     // TODO : Should we lock all mutex rightnow?
 475     Mutex::ScopedLock lockWait(mWaitingTasksMutex);
 476     if(mWaitingTasks.empty())
 477     {
 478       Mutex::ScopedLock lockRunning(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
 479       if(mRunningTasks.empty())
 480       {
 481         Mutex::ScopedLock lockComplete(mCompletedTasksMutex); // We can lock this mutex under mWaitingTasksMutex and mRunningTasksMutex.
 482         if(mCompletedTasks.empty())
 483         {
 484           mProcessorRegistered = false;
 485           Dali::Adaptor::Get().UnregisterProcessor(*this);
 486         }
 487       }
 488     }
 489     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "UnregisterProcessor end (registed? %d)\n", mProcessorRegistered);
 490   }
 491 }
 492
 493 void AsyncTaskManager::TasksCompleted()
 494 {
 495   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "TasksCompleted begin\n");
 496   while(AsyncTaskPtr task = PopNextCompletedTask())
 497   {
 498     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Execute callback [%p]\n", task.Get());
 499     CallbackBase::Execute(*(task->GetCompletedCallback()), task);
 500   }
 501
 502   UnregisterProcessor();
 503   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "TasksCompleted end\n");
 504 }
 505
 506 void AsyncTaskManager::Process(bool postProcessor)
 507 {
 508   TasksCompleted();
 509 }
 510
 511 /// Worker thread called
 512 AsyncTaskPtr AsyncTaskManager::PopNextTaskToProcess()
 513 {
 514   // Lock while popping task out from the queue
 515   Mutex::ScopedLock lock(mWaitingTasksMutex);
 516
 517   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "PopNextTaskToProcess, waiting task count : [%zu]\n", mWaitingTasks.size());
 518
 519   // pop out the next task from the queue
 520   AsyncTaskPtr nextTask = nullptr;
 521
 522   // Fast cut if all waiting tasks are LOW priority, and we cannot excute low task anymore.
 523   if(mWaitingHighProirityTaskCounts == 0u && !mWaitingTasks.empty())
 524   {
 525     // For thread safety
 526     Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
 527
 528     if(mAvaliableLowPriorityTaskCounts == 0u)
 529     {
 530       // There are no avaliabe tasks to run now. Return nullptr.
 531       return nextTask;
 532     }
 533   }
 534
 535   for(auto iter = mWaitingTasks.begin(), endIter = mWaitingTasks.end(); iter != endIter; ++iter)
 536   {
 537     if((*iter)->IsReady())
 538     {
 539       const auto priorityType  = (*iter)->GetPriorityType();
 540       bool       taskAvaliable = priorityType == AsyncTask::PriorityType::HIGH; // Task always valid if it's priority is high
 541       if(!taskAvaliable)
 542       {
 543         // For thread safety
 544         Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
 545
 546         taskAvaliable = (mAvaliableLowPriorityTaskCounts > 0u); // priority is low, but we can use it.
 547       }
 548
 549       if(taskAvaliable)
 550       {
 551         nextTask = *iter;
 552
 553         // Add Running queue
 554         {
 555           // Lock while popping task out from the queue
 556           Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
 557
 558           DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Waiting -> Running [%p]\n", nextTask.Get());
 559
 560           auto runningIter = mRunningTasks.insert(mRunningTasks.end(), std::make_pair(nextTask, RunningTaskState::RUNNING));
 561           CacheImpl::InsertTaskCache(mCacheImpl->mRunningTasksCache, nextTask, runningIter);
 562
 563           CacheImpl::EraseTaskCache(mCacheImpl->mWaitingTasksCache, nextTask, iter);
 564           mWaitingTasks.erase(iter);
 565
 566           // Decrease avaliable task counts if it is low priority
 567           if(priorityType == AsyncTask::PriorityType::LOW)
 568           {
 569             // We are under running task mutex. We can decrease it.
 570             --mAvaliableLowPriorityTaskCounts;
 571           }
 572         }
 573
 574         if(priorityType == AsyncTask::PriorityType::HIGH)
 575         {
 576           // Decrease the number of waiting tasks for high priority.
 577           --mWaitingHighProirityTaskCounts;
 578         }
 579         break;
 580       }
 581     }
 582   }
 583
 584   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Pickup process [%p]\n", nextTask.Get());
 585
 586   return nextTask;
 587 }
 588
 589 /// Worker thread called
 590 void AsyncTaskManager::CompleteTask(AsyncTaskPtr&& task)
 591 {
 592   bool notify = false;
 593
 594   if(task)
 595   {
 596     bool needTrigger = (task->GetCallbackInvocationThread() == AsyncTask::ThreadType::MAIN_THREAD);
 597
 598     // Lock while check validation of task.
 599     {
 600       Mutex::ScopedLock lock(mRunningTasksMutex);
 601
 602       auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
 603       if(mapIter != mCacheImpl->mRunningTasksCache.end())
 604       {
 605         const auto cacheIter = mapIter->second.begin();
 606         DALI_ASSERT_ALWAYS(cacheIter != mapIter->second.end());
 607
 608         const auto iter = *cacheIter;
 609         DALI_ASSERT_DEBUG(iter->first == task);
 610         if(iter->second == RunningTaskState::RUNNING)
 611         {
 612           // This task is valid.
 613           notify = true;
 614         }
 615       }
 616
 617       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "CompleteTask [%p] (is notify? : %d)\n", task.Get(), notify);
 618     }
 619
 620     // We should execute this tasks complete callback out of mutex
 621     if(notify && task->GetCallbackInvocationThread() == AsyncTask::ThreadType::WORKER_THREAD)
 622     {
 623       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Execute callback on worker thread [%p]\n", task.Get());
 624       CallbackBase::Execute(*(task->GetCompletedCallback()), task);
 625     }
 626
 627     // Lock while adding task to the queue
 628     {
 629       Mutex::ScopedLock lock(mRunningTasksMutex);
 630
 631       auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
 632       if(mapIter != mCacheImpl->mRunningTasksCache.end())
 633       {
 634         const auto cacheIter = mapIter->second.begin();
 635         DALI_ASSERT_ALWAYS(cacheIter != mapIter->second.end());
 636
 637         const auto iter         = *cacheIter;
 638         const auto priorityType = iter->first->GetPriorityType();
 639         // Increase avaliable task counts if it is low priority
 640         if(priorityType == AsyncTask::PriorityType::LOW)
 641         {
 642           // We are under running task mutex. We can increase it.
 643           ++mAvaliableLowPriorityTaskCounts;
 644         }
 645
 646         // Move task into completed, for ensure that AsyncTask destroy at main thread.
 647         {
 648           Mutex::ScopedLock lock(mCompletedTasksMutex); // We can lock this mutex under mRunningTasksMutex.
 649
 650           const bool callbackRequired = notify && (task->GetCallbackInvocationThread() == AsyncTask::ThreadType::MAIN_THREAD);
 651
 652           DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Running -> Completed [%p] (callback required? : %d)\n", task.Get(), callbackRequired);
 653
 654           auto completedIter = mCompletedTasks.insert(mCompletedTasks.end(), std::make_pair(task, callbackRequired ? CompletedTaskState::REQUIRE_CALLBACK : CompletedTaskState::SKIP_CALLBACK));
 655           CacheImpl::InsertTaskCache(mCacheImpl->mCompletedTasksCache, task, completedIter);
 656
 657           CacheImpl::EraseTaskCache(mCacheImpl->mRunningTasksCache, task, iter);
 658           mRunningTasks.erase(iter);
 659
 660           if(!needTrigger)
 661           {
 662             needTrigger |= (mCompletedTasks.size() >= FORCE_TRIGGER_THRESHOLD);
 663           }
 664
 665           // Now, task is invalidate.
 666           task.Reset();
 667         }
 668       }
 669     }
 670
 671     // Wake up the main thread
 672     if(needTrigger)
 673     {
 674       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Trigger main thread\n");
 675       mTrigger->Trigger();
 676     }
 677   }
 678 }
 679
 680 // AsyncTaskManager::TaskHelper
 681
 682 AsyncTaskManager::TaskHelper::TaskHelper(AsyncTaskManager& asyncTaskManager)
 683 : TaskHelper(std::unique_ptr<AsyncTaskThread>(new AsyncTaskThread(asyncTaskManager)), asyncTaskManager)
 684 {
 685 }
 686
 687 AsyncTaskManager::TaskHelper::TaskHelper(TaskHelper&& rhs)
 688 : TaskHelper(std::move(rhs.mProcessor), rhs.mAsyncTaskManager)
 689 {
 690 }
 691
 692 AsyncTaskManager::TaskHelper::TaskHelper(std::unique_ptr<AsyncTaskThread> processor, AsyncTaskManager& asyncTaskManager)
 693 : mProcessor(std::move(processor)),
 694   mAsyncTaskManager(asyncTaskManager)
 695 {
 696 }
 697
 698 bool AsyncTaskManager::TaskHelper::Request()
 699 {
 700   return mProcessor->Request();
 701 }
 702 } // namespace Adaptor
 703
 704 } // namespace Internal
 705
 706 } // namespace Dali