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

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