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