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