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

   1 /*
   2  * Copyright (c) 2024 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 // INTERNAL INCLUDES
  29 #include <dali/internal/system/common/environment-variables.h>
  30
  31 #include <unordered_map>
  32
  33 namespace Dali
  34 {
  35 namespace Internal
  36 {
  37 namespace Adaptor
  38 {
  39 namespace
  40 {
  41 constexpr auto FORCE_TRIGGER_THRESHOLD = 128u; ///< Trigger TasksCompleted() forcely if the number of completed task contain too much.
  42
  43 constexpr auto DEFAULT_NUMBER_OF_ASYNC_THREADS = size_t{8u};
  44
  45 // The number of threads for low priority task.
  46 constexpr auto DEFAULT_NUMBER_OF_LOW_PRIORITY_THREADS = size_t{6u};
  47
  48 size_t GetNumberOfThreads(size_t defaultValue)
  49 {
  50   auto           numberString          = EnvironmentVariable::GetEnvironmentVariable(DALI_ENV_ASYNC_MANAGER_THREAD_POOL_SIZE);
  51   auto           numberOfThreads       = numberString ? std::strtoul(numberString, nullptr, 10) : 0;
  52   constexpr auto MAX_NUMBER_OF_THREADS = 16u;
  53   DALI_ASSERT_DEBUG(numberOfThreads <= MAX_NUMBER_OF_THREADS);
  54   return (numberOfThreads > 0 && numberOfThreads <= MAX_NUMBER_OF_THREADS) ? numberOfThreads : defaultValue;
  55 }
  56
  57 size_t GetNumberOfLowPriorityThreads(size_t defaultValue, size_t maxValue)
  58 {
  59   auto numberString    = EnvironmentVariable::GetEnvironmentVariable(DALI_ENV_ASYNC_MANAGER_LOW_PRIORITY_THREAD_POOL_SIZE);
  60   auto numberOfThreads = numberString ? std::strtoul(numberString, nullptr, 10) : 0;
  61   DALI_ASSERT_DEBUG(numberOfThreads <= maxValue);
  62   return (numberOfThreads > 0 && numberOfThreads <= maxValue) ? numberOfThreads : std::min(defaultValue, maxValue);
  63 }
  64
  65 #if defined(DEBUG_ENABLED)
  66 Debug::Filter* gAsyncTasksManagerLogFilter = Debug::Filter::New(Debug::NoLogging, false, "LOG_ASYNC_TASK_MANAGER");
  67
  68 uint32_t gThreadId = 0u; // Only for debug
  69 #endif
  70
  71 /**
  72  * @brief Get the Task Name.
  73  * Note that we can get const char* from std::string_view as data() since it will be const class.
  74  *
  75  * @param task The task what we want to get the name.
  76  * @return The name of task, or (nil) if task is invalid.
  77  */
  78 const char* GetTaskName(AsyncTaskPtr task)
  79 {
  80   // Note
  81   return task ? task->GetTaskName().data() : "(nil)";
  82 }
  83
  84 } // unnamed namespace
  85
  86 // AsyncTaskThread
  87
  88 AsyncTaskThread::AsyncTaskThread(AsyncTaskManager& asyncTaskManager)
  89 : mConditionalWait(),
  90   mAsyncTaskManager(asyncTaskManager),
  91   mLogFactory(Dali::Adaptor::Get().GetLogFactory()),
  92   mTraceFactory(Dali::Adaptor::Get().GetTraceFactory()),
  93   mDestroyThread(false),
  94   mIsThreadStarted(false),
  95   mIsThreadIdle(true)
  96 {
  97 }
  98
  99 AsyncTaskThread::~AsyncTaskThread()
 100 {
 101   // Stop the thread
 102   {
 103     ConditionalWait::ScopedLock lock(mConditionalWait);
 104     mDestroyThread = true;
 105     mConditionalWait.Notify(lock);
 106   }
 107
 108   Join();
 109 }
 110
 111 bool AsyncTaskThread::Request()
 112 {
 113   if(!mIsThreadStarted)
 114   {
 115     Start();
 116     mIsThreadStarted = true;
 117   }
 118
 119   {
 120     // Lock while adding task to the queue
 121     ConditionalWait::ScopedLock lock(mConditionalWait);
 122
 123     if(mIsThreadIdle)
 124     {
 125       mIsThreadIdle = false;
 126
 127       // wake up the thread
 128       mConditionalWait.Notify(lock);
 129       return true;
 130     }
 131   }
 132
 133   return false;
 134 }
 135
 136 void AsyncTaskThread::Run()
 137 {
 138 #if defined(DEBUG_ENABLED)
 139   uint32_t threadId = gThreadId++;
 140   {
 141     char temp[100];
 142     snprintf(temp, 100, "AsyncTaskThread[%u]", threadId);
 143     SetThreadName(temp);
 144   }
 145 #else
 146   SetThreadName("AsyncTaskThread");
 147 #endif
 148   mLogFactory.InstallLogFunction();
 149   mTraceFactory.InstallTraceFunction();
 150
 151   while(!mDestroyThread)
 152   {
 153     AsyncTaskPtr task = mAsyncTaskManager.PopNextTaskToProcess();
 154     if(!task)
 155     {
 156       ConditionalWait::ScopedLock lock(mConditionalWait);
 157       if(!mDestroyThread)
 158       {
 159         mIsThreadIdle = true;
 160         DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Thread[%u] wait\n", threadId);
 161         mConditionalWait.Wait(lock);
 162         DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Thread[%u] awake\n", threadId);
 163       }
 164     }
 165     else
 166     {
 167       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Thread[%u] Process task [%p][%s]\n", threadId, task.Get(), GetTaskName(task));
 168       task->Process();
 169       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Thread[%u] Complete task [%p][%s]\n", threadId, task.Get(), GetTaskName(task));
 170       if(!mDestroyThread)
 171       {
 172         mAsyncTaskManager.CompleteTask(std::move(task));
 173       }
 174     }
 175   }
 176 }
 177
 178 // AsyncTaskManager::TasksCompletedImpl
 179
 180 struct AsyncTaskManager::TasksCompletedImpl
 181 {
 182   TasksCompletedImpl(AsyncTaskManager& manager, EventThreadCallback* trigger)
 183   : mManager(manager),
 184     mTrigger(trigger),
 185     mEmitCompletedTaskTriggered(false)
 186   {
 187   }
 188
 189 public:
 190   /**
 191    * @brief Create new tasks completed id and.
 192    * @post AppendTaskTrace or CheckTasksCompletedCallbackCompleted should be called.
 193    * @param[in] callback The callback that want to be executed when we notify that all tasks completed.
 194    */
 195   Dali::AsyncTaskManager::TasksCompletedId GenerateTasksCompletedId(CallbackBase* callback)
 196   {
 197     // Lock while adding tasks completed callback list to the queue
 198     Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex);
 199
 200     auto id = mTasksCompletedCount++;
 201     DALI_ASSERT_ALWAYS(mTasksCompletedCallbackList.find(id) == mTasksCompletedCallbackList.end());
 202
 203     mTasksCompletedCallbackList.insert({id, CallbackData(callback)});
 204
 205     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "GenerateTasksCompletedId id[%u] callback[%p]\n", id, callback);
 206     return id;
 207   }
 208
 209   /**
 210    * @brief Append task that will be trace.
 211    * @post RemoveTaskTrace should be called.
 212    * @param[in] id The id of tasks completed.
 213    * @param[in] task The task want to trace.
 214    */
 215   void AppendTaskTrace(Dali::AsyncTaskManager::TasksCompletedId id, AsyncTaskPtr task)
 216   {
 217     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "AppendTaskTrace id[%u] task[%p][%s]\n", id, task.Get(), GetTaskName(task));
 218
 219     // Lock while adding tasks completed callback list to the queue
 220     Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex);
 221
 222     auto iter = mTasksCompletedCallbackList.find(id);
 223     if(iter == mTasksCompletedCallbackList.end())
 224     {
 225       // This task is already erased. Ignore.
 226       return;
 227     }
 228
 229     auto& callbackData = iter->second;
 230
 231     auto jter = callbackData.mTasks.find(task.Get());
 232
 233     if(jter != callbackData.mTasks.end())
 234     {
 235       // Increase reference count.
 236       ++(jter->second);
 237     }
 238     else
 239     {
 240       callbackData.mTasks.insert({task.Get(), 1u});
 241     }
 242   }
 243
 244   /**
 245    * @brief Remove all task that were traced.
 246    * @param[in] task The task want to remove trace.
 247    * @param[in] taskCount The number of tasks that will be removed.
 248    */
 249   void RemoveTaskTrace(AsyncTaskPtr task, uint32_t count = 1u)
 250   {
 251     if(count == 0u)
 252     {
 253       return;
 254     }
 255     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "RemoveTaskTrace task[%p][%s] remove count[%u]\n", task.Get(), GetTaskName(task), count);
 256
 257     // Lock while removing tasks completed callback list to the queue
 258     Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex);
 259
 260     for(auto iter = mTasksCompletedCallbackList.begin(); iter != mTasksCompletedCallbackList.end();)
 261     {
 262       auto& callbackData      = iter->second;
 263       bool  eraseCallbackData = false;
 264
 265       auto jter = callbackData.mTasks.find(task.Get());
 266
 267       if(jter != callbackData.mTasks.end())
 268       {
 269         DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "RemoveTaskTrace id[%u] task[%p][%s], current refcount[%u]\n", iter->first, task.Get(), GetTaskName(task), (jter->second));
 270
 271         if(jter->second <= count)
 272         {
 273           callbackData.mTasks.erase(jter);
 274
 275           DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "RemoveTaskTrace id[%u] task erased. remained tasks[%zu]", iter->first, callbackData.mTasks.size());
 276
 277           if(callbackData.mTasks.empty())
 278           {
 279             eraseCallbackData = true;
 280
 281             // Move callback base into list.
 282             // (To avoid task container changed during callback emit)
 283             RegisterTasksCompletedCallback(std::move(callbackData.mCallback), iter->first);
 284
 285             DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "id[%u] completed!\n", iter->first);
 286
 287             iter = mTasksCompletedCallbackList.erase(iter);
 288           }
 289         }
 290         else
 291         {
 292           jter->second -= count;
 293         }
 294       }
 295
 296       if(!eraseCallbackData)
 297       {
 298         ++iter;
 299       }
 300     }
 301   }
 302
 303   /**
 304    * @brief Check whether current TasksCompletedId completed or not.
 305    * @param[in] id The id of tasks completed.
 306    * @return True if all tasks are completed so we need to execute callback soon. False otherwise.
 307    */
 308   bool CheckTasksCompletedCallbackCompleted(Dali::AsyncTaskManager::TasksCompletedId id)
 309   {
 310     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "CheckTasksCompletedCallbackCompleted[%u]\n", id);
 311
 312     // Lock while removing tasks completed callback list to the queue
 313     Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex);
 314
 315     auto iter = mTasksCompletedCallbackList.find(id);
 316     if(iter != mTasksCompletedCallbackList.end())
 317     {
 318       auto& callbackData = iter->second;
 319       if(callbackData.mTasks.empty())
 320       {
 321         // Move callback base into list.
 322         // (To avoid task container changed during callback emit)
 323         RegisterTasksCompletedCallback(std::move(callbackData.mCallback), iter->first);
 324
 325         DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "id[%u] completed!\n", iter->first);
 326
 327         iter = mTasksCompletedCallbackList.erase(iter);
 328
 329         return true;
 330       }
 331     }
 332
 333     return false;
 334   }
 335
 336   /**
 337    * @brief Remove taskS completed callbacks by id.
 338    * @param[in] id The id of taskS completed.
 339    * @return True if taskS completed id removed. False otherwise.
 340    */
 341   bool RemoveTasksCompleted(Dali::AsyncTaskManager::TasksCompletedId id)
 342   {
 343     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "RemoveTasksCompleted[%u]\n", id);
 344
 345     // Lock while removing taskS completed callback list to the queue
 346     Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex);
 347
 348     auto iter = mTasksCompletedCallbackList.find(id);
 349     if(iter == mTasksCompletedCallbackList.end())
 350     {
 351       // This task is already erased, or completed.
 352       // Erase from completed excute callback list.
 353
 354       // Lock while removing excute callback list to the queue
 355       Mutex::ScopedLock lock(mExcuteCallbacksMutex);
 356
 357       for(auto iter = mExcuteCallbackList.begin(); iter != mExcuteCallbackList.end();)
 358       {
 359         if(iter->second == id)
 360         {
 361           iter = mExcuteCallbackList.erase(iter);
 362
 363           return true;
 364         }
 365         else
 366         {
 367           ++iter;
 368         }
 369       }
 370
 371       // This task is alread erased and completed. Ignore.
 372       return false;
 373     }
 374
 375     mTasksCompletedCallbackList.erase(iter);
 376
 377     return true;
 378   }
 379
 380   /**
 381    * @brief Emit all completed callbacks.
 382    * @note This API should be called at event thread.
 383    */
 384   void EmitCompletedTasks()
 385   {
 386     ExecuteCallbackContainer executeCallbackList;
 387     {
 388       // Lock while removing excute callback list to the queue
 389       Mutex::ScopedLock lock(mExcuteCallbacksMutex);
 390
 391       mEmitCompletedTaskTriggered = false;
 392
 393       // Copy callback lists, for let we execute callbacks out of mutex
 394       executeCallbackList = std::move(mExcuteCallbackList);
 395       mExcuteCallbackList.clear();
 396     }
 397
 398     if(!executeCallbackList.empty())
 399     {
 400       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Excute callback count[%zu]\n", executeCallbackList.size());
 401       // Execute all callbacks
 402       for(auto&& callbackPair : executeCallbackList)
 403       {
 404         auto& callback = callbackPair.first;
 405         auto  id       = callbackPair.second;
 406
 407         DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Excute taskS completed callback[%p] for id[%u]\n", callback.get(), id);
 408
 409         Dali::CallbackBase::Execute(*callback, id);
 410       }
 411
 412       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Excute callback end\n");
 413     }
 414   }
 415
 416   /**
 417    * @brief Check whether there is some completed signal what we need to trace, or not.
 418    * @return True if mTasksCompletedCallbackList is not empty. False otherwise.
 419    */
 420   bool IsTasksCompletedCallbackExist()
 421   {
 422     Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex);
 423     return !mTasksCompletedCallbackList.empty();
 424   }
 425
 426   /**
 427    * @brief Check whether there is some completed signal what we need to execute, or not.
 428    * @return True if mExcuteCallbackList is not empty. False otherwise.
 429    */
 430   bool IsExecuteCallbackExist()
 431   {
 432     Mutex::ScopedLock lock(mExcuteCallbacksMutex);
 433     return !mExcuteCallbackList.empty();
 434   }
 435
 436 private:
 437   void RegisterTasksCompletedCallback(std::unique_ptr<CallbackBase> callback, Dali::AsyncTaskManager::TasksCompletedId id)
 438   {
 439     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "TasksCompleted[%u] need to be execute with callback[%p]\n", id, callback.get());
 440
 441     // Lock while adding excute callback list to the queue
 442     Mutex::ScopedLock lock(mExcuteCallbacksMutex);
 443
 444     mExcuteCallbackList.emplace_back(std::move(callback), id);
 445
 446     if(!mEmitCompletedTaskTriggered)
 447     {
 448       mEmitCompletedTaskTriggered = true;
 449
 450       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Trigger processor\n");
 451       mTrigger->Trigger();
 452     }
 453   }
 454
 455 private:
 456   struct CallbackData
 457   {
 458   public:
 459     CallbackData(CallbackBase* callback)
 460     : mCallback(callback),
 461       mTasks()
 462     {
 463     }
 464
 465     CallbackData(CallbackData&& rhs) noexcept
 466     : mCallback(std::move(rhs.mCallback)),
 467       mTasks(std::move(rhs.mTasks))
 468     {
 469     }
 470
 471     CallbackData& operator=(CallbackData&& rhs) noexcept
 472     {
 473       if(this != &rhs)
 474       {
 475         mCallback = std::move(rhs.mCallback);
 476         mTasks    = std::move(rhs.mTasks);
 477       }
 478
 479       return *this;
 480     }
 481
 482   private:
 483     // Delete copy operator.
 484     CallbackData(const CallbackData& rhs) = delete;
 485     CallbackData& operator=(const CallbackData& rhs) = delete;
 486
 487   public:
 488     std::unique_ptr<CallbackBase>                  mCallback;
 489     std::unordered_map<const AsyncTask*, uint32_t> mTasks;
 490   };
 491
 492 private:
 493   AsyncTaskManager&    mManager; ///< Owner of this CacheImpl.
 494   EventThreadCallback* mTrigger; ///< EventThread callback trigger. (Not owned.)
 495
 496   Dali::AsyncTaskManager::TasksCompletedId mTasksCompletedCount{0u};
 497
 498   using TasksCompletedContainer = std::unordered_map<Dali::AsyncTaskManager::TasksCompletedId, CallbackData>;
 499   TasksCompletedContainer mTasksCompletedCallbackList;
 500
 501   using ExecuteCallbackContainer = std::vector<std::pair<std::unique_ptr<CallbackBase>, Dali::AsyncTaskManager::TasksCompletedId>>;
 502   ExecuteCallbackContainer mExcuteCallbackList;
 503
 504   Dali::Mutex mTasksCompletedCallbacksMutex; ///< Mutex for mTasksCompletedCallbackList. We can lock mExcuteCallbacksMutex under this scope.
 505   Dali::Mutex mExcuteCallbacksMutex;         ///< Mutex for mExcuteCallbackList.
 506
 507   bool mEmitCompletedTaskTriggered : 1;
 508 };
 509
 510 // AsyncTaskManager::CacheImpl
 511
 512 struct AsyncTaskManager::CacheImpl
 513 {
 514   CacheImpl(AsyncTaskManager& manager)
 515   : mManager(manager)
 516   {
 517   }
 518
 519 public:
 520   // Insert / Erase task cache API.
 521
 522   /**
 523    * @brief Insert cache that input task.
 524    * @pre Mutex be locked.
 525    */
 526   template<typename CacheContainer, typename Iterator>
 527   static void InsertTaskCache(CacheContainer& cacheMap, AsyncTaskPtr task, Iterator iterator)
 528   {
 529     auto& cacheContainer = cacheMap[task.Get()]; // Get or Create cache container.
 530     cacheContainer.insert(cacheContainer.end(), iterator);
 531   }
 532
 533   /**
 534    * @brief Erase cache that input task.
 535    * @pre Mutex be locked.
 536    */
 537   template<typename CacheContainer, typename Iterator>
 538   static void EraseTaskCache(CacheContainer& cacheMap, AsyncTaskPtr task, Iterator iterator)
 539   {
 540     auto mapIter = cacheMap.find(task.Get());
 541     if(mapIter != cacheMap.end())
 542     {
 543       auto& cacheContainer = (*mapIter).second;
 544       auto  cacheIter      = std::find(cacheContainer.begin(), cacheContainer.end(), iterator);
 545
 546       if(cacheIter != cacheContainer.end())
 547       {
 548         cacheContainer.erase(cacheIter);
 549         if(cacheContainer.empty())
 550         {
 551           cacheMap.erase(mapIter);
 552         }
 553       }
 554     }
 555   }
 556
 557   /**
 558    * @brief Erase all cache that input task.
 559    * @pre Mutex be locked.
 560    */
 561   template<typename CacheContainer>
 562   static void EraseAllTaskCache(CacheContainer& cacheMap, AsyncTaskPtr task)
 563   {
 564     auto mapIter = cacheMap.find(task.Get());
 565     if(mapIter != cacheMap.end())
 566     {
 567       cacheMap.erase(mapIter);
 568     }
 569   }
 570
 571 public:
 572   AsyncTaskManager& mManager; ///< Owner of this CacheImpl.
 573
 574   // Keep cache iterators as list since we take tasks by FIFO as default.
 575   using TaskCacheContainer          = std::unordered_map<const AsyncTask*, std::list<AsyncTaskContainer::iterator>>;
 576   using RunningTaskCacheContainer   = std::unordered_map<const AsyncTask*, std::list<AsyncRunningTaskContainer::iterator>>;
 577   using CompletedTaskCacheContainer = std::unordered_map<const AsyncTask*, std::list<AsyncCompletedTaskContainer::iterator>>;
 578
 579   TaskCacheContainer          mWaitingTasksCache;   ///< The cache of tasks and iterator for waiting to async process. Must be locked under mWaitingTasksMutex.
 580   RunningTaskCacheContainer   mRunningTasksCache;   ///< The cache of tasks and iterator for running tasks. Must be locked under mRunningTasksMutex.
 581   CompletedTaskCacheContainer mCompletedTasksCache; ///< The cache of tasks and iterator for completed async process. Must be locked under mCompletedTasksMutex.
 582 };
 583
 584 // AsyncTaskManager
 585
 586 Dali::AsyncTaskManager AsyncTaskManager::Get()
 587 {
 588   Dali::AsyncTaskManager manager;
 589   SingletonService       singletonService(SingletonService::Get());
 590   if(singletonService)
 591   {
 592     // Check whether the async task manager is already created
 593     Dali::BaseHandle handle = singletonService.GetSingleton(typeid(Dali::AsyncTaskManager));
 594     if(handle)
 595     {
 596       // If so, downcast the handle of singleton
 597       manager = Dali::AsyncTaskManager(dynamic_cast<Internal::Adaptor::AsyncTaskManager*>(handle.GetObjectPtr()));
 598     }
 599
 600     if(!manager)
 601     {
 602       // If not, create the async task manager and register it as a singleton
 603       Internal::Adaptor::AsyncTaskManager* internalAsyncTaskManager = new Internal::Adaptor::AsyncTaskManager();
 604       manager                                                       = Dali::AsyncTaskManager(internalAsyncTaskManager);
 605       singletonService.Register(typeid(manager), manager);
 606     }
 607   }
 608   return manager;
 609 }
 610
 611 AsyncTaskManager::AsyncTaskManager()
 612 : mTasks(GetNumberOfThreads(DEFAULT_NUMBER_OF_ASYNC_THREADS), [&]() { return TaskHelper(*this); }),
 613   mAvaliableLowPriorityTaskCounts(GetNumberOfLowPriorityThreads(DEFAULT_NUMBER_OF_LOW_PRIORITY_THREADS, mTasks.GetElementCount())),
 614   mWaitingHighProirityTaskCounts(0u),
 615   mTrigger(new EventThreadCallback(MakeCallback(this, &AsyncTaskManager::TasksCompleted))),
 616   mTasksCompletedImpl(new TasksCompletedImpl(*this, mTrigger.get())),
 617   mCacheImpl(new CacheImpl(*this)),
 618   mProcessorRegistered(false)
 619 {
 620 }
 621
 622 AsyncTaskManager::~AsyncTaskManager()
 623 {
 624   if(mProcessorRegistered && Dali::Adaptor::IsAvailable())
 625   {
 626     mProcessorRegistered = false;
 627     Dali::Adaptor::Get().UnregisterProcessor(*this);
 628   }
 629
 630   // Join all threads.
 631   mTasks.Clear();
 632
 633   // Remove task completed impl and cache impl after all threads are join.
 634   mTasksCompletedImpl.reset();
 635   mCacheImpl.reset();
 636
 637   // Remove tasks after CacheImpl removed
 638   mWaitingTasks.clear();
 639   mRunningTasks.clear();
 640   mCompletedTasks.clear();
 641 }
 642
 643 void AsyncTaskManager::AddTask(AsyncTaskPtr task)
 644 {
 645   if(task)
 646   {
 647     // Lock while adding task to the queue
 648     Mutex::ScopedLock lock(mWaitingTasksMutex);
 649
 650     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "AddTask [%p][%s]\n", task.Get(), GetTaskName(task));
 651
 652     // push back into waiting queue.
 653     auto waitingIter = mWaitingTasks.insert(mWaitingTasks.end(), task);
 654     CacheImpl::InsertTaskCache(mCacheImpl->mWaitingTasksCache, task, waitingIter);
 655
 656     if(task->GetPriorityType() == AsyncTask::PriorityType::HIGH)
 657     {
 658       // Increase the number of waiting tasks for high priority.
 659       ++mWaitingHighProirityTaskCounts;
 660     }
 661
 662     {
 663       // For thread safety
 664       Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
 665
 666       // Finish all Running threads are working
 667       if(mRunningTasks.size() >= mTasks.GetElementCount())
 668       {
 669         return;
 670       }
 671     }
 672   }
 673
 674   {
 675     Mutex::ScopedLock lock(mTasksMutex);
 676     size_t            count = mTasks.GetElementCount();
 677     size_t            index = 0;
 678     while(index++ < count)
 679     {
 680       auto processHelperIt = mTasks.GetNext();
 681       DALI_ASSERT_ALWAYS(processHelperIt != mTasks.End());
 682       if(processHelperIt->Request())
 683       {
 684         break;
 685       }
 686       // If all threads are busy, then it's ok just to push the task because they will try to get the next job.
 687     }
 688   }
 689
 690   // Register Process (Since mTrigger execute too late timing if event thread running a lots of events.)
 691   RegisterProcessor();
 692
 693   return;
 694 }
 695
 696 void AsyncTaskManager::RemoveTask(AsyncTaskPtr task)
 697 {
 698   if(task)
 699   {
 700     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "RemoveTask [%p][%s]\n", task.Get(), GetTaskName(task));
 701
 702     // Check whether we need to unregister processor.
 703     // If there is some non-empty queue exist, we don't need to unregister processor.
 704     bool needCheckUnregisterProcessor = true;
 705
 706     uint32_t removedCount = 0u;
 707
 708     {
 709       // Lock while remove task from the queue
 710       Mutex::ScopedLock lock(mWaitingTasksMutex);
 711
 712       auto mapIter = mCacheImpl->mWaitingTasksCache.find(task.Get());
 713       if(mapIter != mCacheImpl->mWaitingTasksCache.end())
 714       {
 715         for(auto& iterator : mapIter->second)
 716         {
 717           DALI_ASSERT_DEBUG((*iterator) == task);
 718           if((*iterator)->GetPriorityType() == AsyncTask::PriorityType::HIGH && mWaitingHighProirityTaskCounts > 0u)
 719           {
 720             // Decrease the number of waiting tasks for high priority.
 721             --mWaitingHighProirityTaskCounts;
 722           }
 723           mWaitingTasks.erase(iterator);
 724           ++removedCount;
 725         }
 726         CacheImpl::EraseAllTaskCache(mCacheImpl->mWaitingTasksCache, task);
 727       }
 728
 729       if(!mWaitingTasks.empty())
 730       {
 731         needCheckUnregisterProcessor = false;
 732       }
 733     }
 734
 735     {
 736       // Lock while remove task from the queue
 737       Mutex::ScopedLock lock(mRunningTasksMutex);
 738
 739       auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
 740       if(mapIter != mCacheImpl->mRunningTasksCache.end())
 741       {
 742         for(auto& iterator : mapIter->second)
 743         {
 744           DALI_ASSERT_DEBUG((*iterator).first == task);
 745           // We cannot erase container. Just mark as canceled.
 746           // Note : mAvaliableLowPriorityTaskCounts will be increased after process finished.
 747           if((*iterator).second == RunningTaskState::RUNNING)
 748           {
 749             (*iterator).second = RunningTaskState::CANCELED;
 750             ++removedCount;
 751           }
 752         }
 753       }
 754
 755       if(!mRunningTasks.empty())
 756       {
 757         needCheckUnregisterProcessor = false;
 758       }
 759     }
 760
 761     {
 762       // Lock while remove task from the queue
 763       Mutex::ScopedLock lock(mCompletedTasksMutex);
 764
 765       auto mapIter = mCacheImpl->mCompletedTasksCache.find(task.Get());
 766       if(mapIter != mCacheImpl->mCompletedTasksCache.end())
 767       {
 768         for(auto& iterator : mapIter->second)
 769         {
 770           DALI_ASSERT_DEBUG((*iterator).first == task);
 771           if((*iterator).second == CompletedTaskState::REQUIRE_CALLBACK)
 772           {
 773             ++removedCount;
 774           }
 775           mCompletedTasks.erase(iterator);
 776         }
 777         CacheImpl::EraseAllTaskCache(mCacheImpl->mCompletedTasksCache, task);
 778       }
 779
 780       if(!mCompletedTasks.empty())
 781       {
 782         needCheckUnregisterProcessor = false;
 783       }
 784     }
 785
 786     // Remove TasksCompleted callback trace
 787     if(removedCount > 0u && mTasksCompletedImpl->IsTasksCompletedCallbackExist())
 788     {
 789       mTasksCompletedImpl->RemoveTaskTrace(task, removedCount);
 790     }
 791
 792     // UnregisterProcessor required to lock mutex. Call this API only if required.
 793     if(needCheckUnregisterProcessor)
 794     {
 795       UnregisterProcessor();
 796     }
 797   }
 798 }
 799
 800 Dali::AsyncTaskManager::TasksCompletedId AsyncTaskManager::SetCompletedCallback(CallbackBase* callback, Dali::AsyncTaskManager::CompletedCallbackTraceMask mask)
 801 {
 802   // mTasksCompletedImpl will take ownership of callback.
 803   Dali::AsyncTaskManager::TasksCompletedId tasksCompletedId = mTasksCompletedImpl->GenerateTasksCompletedId(callback);
 804
 805   bool taskAdded = false; ///< Flag whether at least one task tracing now.
 806
 807   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "SetCompletedCallback id : %u, mask : %d\n", tasksCompletedId, static_cast<int32_t>(mask));
 808
 809   // Please be careful the order of mutex, to avoid dead lock.
 810   {
 811     Mutex::ScopedLock lockWait(mWaitingTasksMutex);
 812     {
 813       Mutex::ScopedLock lockRunning(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
 814       {
 815         Mutex::ScopedLock lockComplete(mCompletedTasksMutex); // We can lock this mutex under mWaitingTasksMutex and mRunningTasksMutex.
 816
 817         // Collect all tasks from waiting tasks
 818         for(auto& task : mWaitingTasks)
 819         {
 820           auto checkMask = (task->GetCallbackInvocationThread() == Dali::AsyncTask::ThreadType::MAIN_THREAD ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_MAIN : Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_WORKER) |
 821                            (task->GetPriorityType() == Dali::AsyncTask::PriorityType::HIGH ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_HIGH : Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_LOW);
 822
 823           if((checkMask & mask) == checkMask)
 824           {
 825             taskAdded = true;
 826             mTasksCompletedImpl->AppendTaskTrace(tasksCompletedId, task);
 827           }
 828         }
 829
 830         // Collect all tasks from running tasks
 831         for(auto& taskPair : mRunningTasks)
 832         {
 833           // Trace only if it is running now.
 834           if(taskPair.second == RunningTaskState::RUNNING)
 835           {
 836             auto& task      = taskPair.first;
 837             auto  checkMask = (task->GetCallbackInvocationThread() == Dali::AsyncTask::ThreadType::MAIN_THREAD ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_MAIN : Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_WORKER) |
 838                              (task->GetPriorityType() == Dali::AsyncTask::PriorityType::HIGH ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_HIGH : Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_LOW);
 839
 840             if((checkMask & mask) == checkMask)
 841             {
 842               taskAdded = true;
 843               mTasksCompletedImpl->AppendTaskTrace(tasksCompletedId, task);
 844             }
 845           }
 846         }
 847
 848         // Collect all tasks from complete tasks
 849         for(auto& taskPair : mCompletedTasks)
 850         {
 851           // Trace only if it is need callback.
 852           // Note : There are two CompletedTaskState::SKIP_CALLBACK cases, worker thread invocation and canceled cases.
 853           //        If worker thread invocation, than it already remove trace at completed timing.
 854           //        If canceled cases, we don't append trace at running tasks already.
 855           //        So, we don't need to trace for SKIP_CALLBACK cases.
 856           if(taskPair.second == CompletedTaskState::REQUIRE_CALLBACK)
 857           {
 858             auto& task      = taskPair.first;
 859             auto  checkMask = (task->GetCallbackInvocationThread() == Dali::AsyncTask::ThreadType::MAIN_THREAD ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_MAIN : Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_WORKER) |
 860                              (task->GetPriorityType() == Dali::AsyncTask::PriorityType::HIGH ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_HIGH : Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_LOW);
 861
 862             if((checkMask & mask) == checkMask)
 863             {
 864               taskAdded = true;
 865               mTasksCompletedImpl->AppendTaskTrace(tasksCompletedId, task);
 866             }
 867           }
 868         }
 869       }
 870     }
 871   }
 872
 873   // If there is nothing to check task, just excute callback right now.
 874   if(!taskAdded)
 875   {
 876     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "CompletedCallback id[%u] executed now due to no task exist\n", tasksCompletedId);
 877
 878     mTasksCompletedImpl->CheckTasksCompletedCallbackCompleted(tasksCompletedId);
 879   }
 880   return tasksCompletedId;
 881 }
 882
 883 bool AsyncTaskManager::RemoveCompletedCallback(Dali::AsyncTaskManager::TasksCompletedId tasksCompletedId)
 884 {
 885   return mTasksCompletedImpl->RemoveTasksCompleted(tasksCompletedId);
 886 }
 887
 888 AsyncTaskPtr AsyncTaskManager::PopNextCompletedTask()
 889 {
 890   std::vector<AsyncTaskPtr> ignoredTaskList; ///< To keep asyncTask reference so we can ensure that destructor called out of mutex.
 891
 892   AsyncTaskPtr nextCompletedTask = nullptr;
 893   {
 894     // Lock while popping task out from the queue
 895     Mutex::ScopedLock lock(mCompletedTasksMutex);
 896
 897     while(!mCompletedTasks.empty())
 898     {
 899       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "PopNextCompletedTask, completed task count : [%zu]\n", mCompletedTasks.size());
 900
 901       auto               next      = mCompletedTasks.begin();
 902       AsyncTaskPtr       nextTask  = next->first;
 903       CompletedTaskState taskState = next->second;
 904       CacheImpl::EraseTaskCache(mCacheImpl->mCompletedTasksCache, nextTask, next);
 905       mCompletedTasks.erase(next);
 906
 907       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Completed task [%p][%s] (callback required? : %d)\n", nextTask.Get(), GetTaskName(nextTask), taskState == CompletedTaskState::REQUIRE_CALLBACK);
 908
 909       if(taskState == CompletedTaskState::REQUIRE_CALLBACK)
 910       {
 911         nextCompletedTask = nextTask;
 912         break;
 913       }
 914
 915       ignoredTaskList.push_back(nextTask);
 916     }
 917
 918     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Pickup completed [%p][%s]\n", nextCompletedTask.Get(), GetTaskName(nextCompletedTask));
 919   }
 920
 921   return nextCompletedTask;
 922 }
 923
 924 void AsyncTaskManager::RegisterProcessor()
 925 {
 926   if(!mProcessorRegistered && Dali::Adaptor::IsAvailable())
 927   {
 928     Dali::Adaptor::Get().RegisterProcessor(*this);
 929     mProcessorRegistered = true;
 930   }
 931 }
 932
 933 void AsyncTaskManager::UnregisterProcessor()
 934 {
 935   if(mProcessorRegistered && Dali::Adaptor::IsAvailable())
 936   {
 937     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "UnregisterProcessor begin\n");
 938     // Keep processor at least 1 task exist.
 939     // Please be careful the order of mutex, to avoid dead lock.
 940     // TODO : Should we lock all mutex rightnow?
 941     Mutex::ScopedLock lockWait(mWaitingTasksMutex);
 942     if(mWaitingTasks.empty())
 943     {
 944       Mutex::ScopedLock lockRunning(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
 945       if(mRunningTasks.empty())
 946       {
 947         Mutex::ScopedLock lockComplete(mCompletedTasksMutex); // We can lock this mutex under mWaitingTasksMutex and mRunningTasksMutex.
 948         if(mCompletedTasks.empty())
 949         {
 950           mProcessorRegistered = false;
 951           Dali::Adaptor::Get().UnregisterProcessor(*this);
 952         }
 953       }
 954     }
 955     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "UnregisterProcessor end (registed? %d)\n", mProcessorRegistered);
 956   }
 957 }
 958
 959 void AsyncTaskManager::TasksCompleted()
 960 {
 961   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "TasksCompleted begin\n");
 962   while(AsyncTaskPtr task = PopNextCompletedTask())
 963   {
 964     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Execute callback [%p][%s]\n", task.Get(), GetTaskName(task));
 965     CallbackBase::Execute(*(task->GetCompletedCallback()), task);
 966
 967     // Remove TasksCompleted callback trace
 968     if(mTasksCompletedImpl->IsTasksCompletedCallbackExist())
 969     {
 970       mTasksCompletedImpl->RemoveTaskTrace(task);
 971     }
 972   }
 973
 974   UnregisterProcessor();
 975   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "TasksCompleted end\n");
 976
 977   mTasksCompletedImpl->EmitCompletedTasks();
 978 }
 979
 980 void AsyncTaskManager::Process(bool postProcessor)
 981 {
 982   TasksCompleted();
 983 }
 984
 985 /// Worker thread called
 986 AsyncTaskPtr AsyncTaskManager::PopNextTaskToProcess()
 987 {
 988   // Lock while popping task out from the queue
 989   Mutex::ScopedLock lock(mWaitingTasksMutex);
 990
 991   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "PopNextTaskToProcess, waiting task count : [%zu]\n", mWaitingTasks.size());
 992
 993   // pop out the next task from the queue
 994   AsyncTaskPtr nextTask = nullptr;
 995
 996   // Fast cut if all waiting tasks are LOW priority, and we cannot excute low task anymore.
 997   if(mWaitingHighProirityTaskCounts == 0u && !mWaitingTasks.empty())
 998   {
 999     // For thread safety
1000     Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
1001
1002     if(mAvaliableLowPriorityTaskCounts == 0u)
1003     {
1004       // There are no avaliabe tasks to run now. Return nullptr.
1005       return nextTask;
1006     }
1007   }
1008
1009   for(auto iter = mWaitingTasks.begin(), endIter = mWaitingTasks.end(); iter != endIter; ++iter)
1010   {
1011     if((*iter)->IsReady())
1012     {
1013       const auto priorityType  = (*iter)->GetPriorityType();
1014       bool       taskAvaliable = priorityType == AsyncTask::PriorityType::HIGH; // Task always valid if it's priority is high
1015       if(!taskAvaliable)
1016       {
1017         // For thread safety
1018         Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
1019
1020         taskAvaliable = (mAvaliableLowPriorityTaskCounts > 0u); // priority is low, but we can use it.
1021       }
1022
1023       // Check whether we try to running same task at multiple threads.
1024       if(taskAvaliable)
1025       {
1026         Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
1027         auto              mapIter = mCacheImpl->mRunningTasksCache.find((*iter).Get());
1028         if(mapIter != mCacheImpl->mRunningTasksCache.end())
1029         {
1030           if(!mapIter->second.empty())
1031           {
1032             // Some other thread running this tasks now. Ignore it.
1033             DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Some other thread running this task [%p][%s]\n", (*iter).Get(), GetTaskName(*iter));
1034             taskAvaliable = false;
1035           }
1036         }
1037       }
1038
1039       if(taskAvaliable)
1040       {
1041         nextTask = *iter;
1042
1043         // Add Running queue
1044         {
1045           // Lock while popping task out from the queue
1046           Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
1047
1048           DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Waiting -> Running [%p][%s]\n", nextTask.Get(), GetTaskName(nextTask));
1049
1050           auto runningIter = mRunningTasks.insert(mRunningTasks.end(), std::make_pair(nextTask, RunningTaskState::RUNNING));
1051           CacheImpl::InsertTaskCache(mCacheImpl->mRunningTasksCache, nextTask, runningIter);
1052
1053           CacheImpl::EraseTaskCache(mCacheImpl->mWaitingTasksCache, nextTask, iter);
1054           mWaitingTasks.erase(iter);
1055
1056           // Decrease avaliable task counts if it is low priority
1057           if(priorityType == AsyncTask::PriorityType::LOW)
1058           {
1059             // We are under running task mutex. We can decrease it.
1060             --mAvaliableLowPriorityTaskCounts;
1061           }
1062         }
1063
1064         if(priorityType == AsyncTask::PriorityType::HIGH && mWaitingHighProirityTaskCounts > 0u)
1065         {
1066           // Decrease the number of waiting tasks for high priority.
1067           --mWaitingHighProirityTaskCounts;
1068         }
1069         break;
1070       }
1071     }
1072   }
1073
1074   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Pickup process [%p][%s]\n", nextTask.Get(), GetTaskName(nextTask));
1075
1076   return nextTask;
1077 }
1078
1079 /// Worker thread called
1080 void AsyncTaskManager::CompleteTask(AsyncTaskPtr&& task)
1081 {
1082   if(task)
1083   {
1084     bool needTrigger = false;
1085
1086     // Check now whether we need to execute callback or not, for worker thread cases.
1087     if(task->GetCallbackInvocationThread() == AsyncTask::ThreadType::WORKER_THREAD)
1088     {
1089       bool notify = false;
1090
1091       // Lock while check validation of task.
1092       {
1093         Mutex::ScopedLock lock(mRunningTasksMutex);
1094
1095         auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
1096         if(mapIter != mCacheImpl->mRunningTasksCache.end())
1097         {
1098           const auto cacheIter = mapIter->second.begin();
1099           DALI_ASSERT_ALWAYS(cacheIter != mapIter->second.end());
1100
1101           const auto iter = *cacheIter;
1102           DALI_ASSERT_DEBUG(iter->first == task);
1103           if(iter->second == RunningTaskState::RUNNING)
1104           {
1105             // This task is valid.
1106             notify = true;
1107           }
1108         }
1109
1110         DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "CompleteTask [%p][%s] (is notify? : %d)\n", task.Get(), GetTaskName(task), notify);
1111       }
1112
1113       // We should execute this tasks complete callback out of mutex
1114       if(notify)
1115       {
1116         DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Execute callback on worker thread [%p][%s]\n", task.Get(), GetTaskName(task));
1117         CallbackBase::Execute(*(task->GetCompletedCallback()), task);
1118
1119         // We need to remove task trace now.
1120         if(mTasksCompletedImpl->IsTasksCompletedCallbackExist())
1121         {
1122           mTasksCompletedImpl->RemoveTaskTrace(task);
1123
1124           if(mTasksCompletedImpl->IsExecuteCallbackExist())
1125           {
1126             // We need to call EmitCompletedTasks(). Trigger main thread.
1127             needTrigger = true;
1128           }
1129         }
1130       }
1131     }
1132
1133     // Lock while adding task to the queue
1134     {
1135       bool notify = false;
1136
1137       Mutex::ScopedLock lock(mRunningTasksMutex);
1138
1139       auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
1140       if(mapIter != mCacheImpl->mRunningTasksCache.end())
1141       {
1142         const auto cacheIter = mapIter->second.begin();
1143         DALI_ASSERT_ALWAYS(cacheIter != mapIter->second.end());
1144
1145         const auto iter = *cacheIter;
1146
1147         DALI_ASSERT_DEBUG(iter->first == task);
1148         if(iter->second == RunningTaskState::RUNNING)
1149         {
1150           // This task is valid.
1151           notify = true;
1152         }
1153
1154         const auto priorityType = iter->first->GetPriorityType();
1155         // Increase avaliable task counts if it is low priority
1156         if(priorityType == AsyncTask::PriorityType::LOW)
1157         {
1158           // We are under running task mutex. We can increase it.
1159           ++mAvaliableLowPriorityTaskCounts;
1160         }
1161
1162         // Move task into completed, for ensure that AsyncTask destroy at main thread.
1163         {
1164           Mutex::ScopedLock lock(mCompletedTasksMutex); // We can lock this mutex under mRunningTasksMutex.
1165
1166           const bool callbackRequired = notify && (task->GetCallbackInvocationThread() == AsyncTask::ThreadType::MAIN_THREAD);
1167
1168           needTrigger |= callbackRequired;
1169
1170           DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Running -> Completed [%p][%s] (callback required? : %d)\n", task.Get(), GetTaskName(task), callbackRequired);
1171
1172           auto completedIter = mCompletedTasks.insert(mCompletedTasks.end(), std::make_pair(task, callbackRequired ? CompletedTaskState::REQUIRE_CALLBACK : CompletedTaskState::SKIP_CALLBACK));
1173           CacheImpl::InsertTaskCache(mCacheImpl->mCompletedTasksCache, task, completedIter);
1174
1175           CacheImpl::EraseTaskCache(mCacheImpl->mRunningTasksCache, task, iter);
1176           mRunningTasks.erase(iter);
1177
1178           if(!needTrigger)
1179           {
1180             needTrigger |= (mCompletedTasks.size() >= FORCE_TRIGGER_THRESHOLD);
1181           }
1182
1183           // Now, task is invalidate.
1184           task.Reset();
1185         }
1186       }
1187     }
1188
1189     // Wake up the main thread
1190     if(needTrigger)
1191     {
1192       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Trigger main thread\n");
1193       mTrigger->Trigger();
1194     }
1195   }
1196 }
1197
1198 // AsyncTaskManager::TaskHelper
1199
1200 AsyncTaskManager::TaskHelper::TaskHelper(AsyncTaskManager& asyncTaskManager)
1201 : TaskHelper(std::unique_ptr<AsyncTaskThread>(new AsyncTaskThread(asyncTaskManager)), asyncTaskManager)
1202 {
1203 }
1204
1205 AsyncTaskManager::TaskHelper::TaskHelper(TaskHelper&& rhs)
1206 : TaskHelper(std::move(rhs.mProcessor), rhs.mAsyncTaskManager)
1207 {
1208 }
1209
1210 AsyncTaskManager::TaskHelper::TaskHelper(std::unique_ptr<AsyncTaskThread> processor, AsyncTaskManager& asyncTaskManager)
1211 : mProcessor(std::move(processor)),
1212   mAsyncTaskManager(asyncTaskManager)
1213 {
1214 }
1215
1216 bool AsyncTaskManager::TaskHelper::Request()
1217 {
1218   return mProcessor->Request();
1219 }
1220 } // namespace Adaptor
1221
1222 } // namespace Internal
1223
1224 } // namespace Dali