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   size_t count = mTasks.GetElementCount();
 674   size_t index = 0;
 675   while(index++ < count)
 676   {
 677     auto processHelperIt = mTasks.GetNext();
 678     DALI_ASSERT_ALWAYS(processHelperIt != mTasks.End());
 679     if(processHelperIt->Request())
 680     {
 681       break;
 682     }
 683     // If all threads are busy, then it's ok just to push the task because they will try to get the next job.
 684   }
 685
 686   // Register Process (Since mTrigger execute too late timing if event thread running a lots of events.)
 687   RegisterProcessor();
 688
 689   return;
 690 }
 691
 692 void AsyncTaskManager::RemoveTask(AsyncTaskPtr task)
 693 {
 694   if(task)
 695   {
 696     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "RemoveTask [%p][%s]\n", task.Get(), GetTaskName(task));
 697
 698     // Check whether we need to unregister processor.
 699     // If there is some non-empty queue exist, we don't need to unregister processor.
 700     bool needCheckUnregisterProcessor = true;
 701
 702     uint32_t removedCount = 0u;
 703
 704     {
 705       // Lock while remove task from the queue
 706       Mutex::ScopedLock lock(mWaitingTasksMutex);
 707
 708       auto mapIter = mCacheImpl->mWaitingTasksCache.find(task.Get());
 709       if(mapIter != mCacheImpl->mWaitingTasksCache.end())
 710       {
 711         for(auto& iterator : mapIter->second)
 712         {
 713           DALI_ASSERT_DEBUG((*iterator) == task);
 714           if((*iterator)->GetPriorityType() == AsyncTask::PriorityType::HIGH && mWaitingHighProirityTaskCounts > 0u)
 715           {
 716             // Decrease the number of waiting tasks for high priority.
 717             --mWaitingHighProirityTaskCounts;
 718           }
 719           mWaitingTasks.erase(iterator);
 720           ++removedCount;
 721         }
 722         CacheImpl::EraseAllTaskCache(mCacheImpl->mWaitingTasksCache, task);
 723       }
 724
 725       if(!mWaitingTasks.empty())
 726       {
 727         needCheckUnregisterProcessor = false;
 728       }
 729     }
 730
 731     {
 732       // Lock while remove task from the queue
 733       Mutex::ScopedLock lock(mRunningTasksMutex);
 734
 735       auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
 736       if(mapIter != mCacheImpl->mRunningTasksCache.end())
 737       {
 738         for(auto& iterator : mapIter->second)
 739         {
 740           DALI_ASSERT_DEBUG((*iterator).first == task);
 741           // We cannot erase container. Just mark as canceled.
 742           // Note : mAvaliableLowPriorityTaskCounts will be increased after process finished.
 743           if((*iterator).second == RunningTaskState::RUNNING)
 744           {
 745             (*iterator).second = RunningTaskState::CANCELED;
 746             ++removedCount;
 747           }
 748         }
 749       }
 750
 751       if(!mRunningTasks.empty())
 752       {
 753         needCheckUnregisterProcessor = false;
 754       }
 755     }
 756
 757     {
 758       // Lock while remove task from the queue
 759       Mutex::ScopedLock lock(mCompletedTasksMutex);
 760
 761       auto mapIter = mCacheImpl->mCompletedTasksCache.find(task.Get());
 762       if(mapIter != mCacheImpl->mCompletedTasksCache.end())
 763       {
 764         for(auto& iterator : mapIter->second)
 765         {
 766           DALI_ASSERT_DEBUG((*iterator).first == task);
 767           if((*iterator).second == CompletedTaskState::REQUIRE_CALLBACK)
 768           {
 769             ++removedCount;
 770           }
 771           mCompletedTasks.erase(iterator);
 772         }
 773         CacheImpl::EraseAllTaskCache(mCacheImpl->mCompletedTasksCache, task);
 774       }
 775
 776       if(!mCompletedTasks.empty())
 777       {
 778         needCheckUnregisterProcessor = false;
 779       }
 780     }
 781
 782     // Remove TasksCompleted callback trace
 783     if(removedCount > 0u && mTasksCompletedImpl->IsTasksCompletedCallbackExist())
 784     {
 785       mTasksCompletedImpl->RemoveTaskTrace(task, removedCount);
 786     }
 787
 788     // UnregisterProcessor required to lock mutex. Call this API only if required.
 789     if(needCheckUnregisterProcessor)
 790     {
 791       UnregisterProcessor();
 792     }
 793   }
 794 }
 795
 796 Dali::AsyncTaskManager::TasksCompletedId AsyncTaskManager::SetCompletedCallback(CallbackBase* callback, Dali::AsyncTaskManager::CompletedCallbackTraceMask mask)
 797 {
 798   // mTasksCompletedImpl will take ownership of callback.
 799   Dali::AsyncTaskManager::TasksCompletedId tasksCompletedId = mTasksCompletedImpl->GenerateTasksCompletedId(callback);
 800
 801   bool taskAdded = false; ///< Flag whether at least one task tracing now.
 802
 803   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "SetCompletedCallback id : %u, mask : %d\n", tasksCompletedId, static_cast<int32_t>(mask));
 804
 805   // Please be careful the order of mutex, to avoid dead lock.
 806   {
 807     Mutex::ScopedLock lockWait(mWaitingTasksMutex);
 808     {
 809       Mutex::ScopedLock lockRunning(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
 810       {
 811         Mutex::ScopedLock lockComplete(mCompletedTasksMutex); // We can lock this mutex under mWaitingTasksMutex and mRunningTasksMutex.
 812
 813         // Collect all tasks from waiting tasks
 814         for(auto& task : mWaitingTasks)
 815         {
 816           auto checkMask = (task->GetCallbackInvocationThread() == Dali::AsyncTask::ThreadType::MAIN_THREAD ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_MAIN : Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_WORKER) |
 817                            (task->GetPriorityType() == Dali::AsyncTask::PriorityType::HIGH ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_HIGH : Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_LOW);
 818
 819           if((checkMask & mask) == checkMask)
 820           {
 821             taskAdded = true;
 822             mTasksCompletedImpl->AppendTaskTrace(tasksCompletedId, task);
 823           }
 824         }
 825
 826         // Collect all tasks from running tasks
 827         for(auto& taskPair : mRunningTasks)
 828         {
 829           // Trace only if it is running now.
 830           if(taskPair.second == RunningTaskState::RUNNING)
 831           {
 832             auto& task      = taskPair.first;
 833             auto  checkMask = (task->GetCallbackInvocationThread() == Dali::AsyncTask::ThreadType::MAIN_THREAD ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_MAIN : Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_WORKER) |
 834                              (task->GetPriorityType() == Dali::AsyncTask::PriorityType::HIGH ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_HIGH : Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_LOW);
 835
 836             if((checkMask & mask) == checkMask)
 837             {
 838               taskAdded = true;
 839               mTasksCompletedImpl->AppendTaskTrace(tasksCompletedId, task);
 840             }
 841           }
 842         }
 843
 844         // Collect all tasks from complete tasks
 845         for(auto& taskPair : mCompletedTasks)
 846         {
 847           // Trace only if it is need callback.
 848           // Note : There are two CompletedTaskState::SKIP_CALLBACK cases, worker thread invocation and canceled cases.
 849           //        If worker thread invocation, than it already remove trace at completed timing.
 850           //        If canceled cases, we don't append trace at running tasks already.
 851           //        So, we don't need to trace for SKIP_CALLBACK cases.
 852           if(taskPair.second == CompletedTaskState::REQUIRE_CALLBACK)
 853           {
 854             auto& task      = taskPair.first;
 855             auto  checkMask = (task->GetCallbackInvocationThread() == Dali::AsyncTask::ThreadType::MAIN_THREAD ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_MAIN : Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_WORKER) |
 856                              (task->GetPriorityType() == Dali::AsyncTask::PriorityType::HIGH ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_HIGH : Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_LOW);
 857
 858             if((checkMask & mask) == checkMask)
 859             {
 860               taskAdded = true;
 861               mTasksCompletedImpl->AppendTaskTrace(tasksCompletedId, task);
 862             }
 863           }
 864         }
 865       }
 866     }
 867   }
 868
 869   // If there is nothing to check task, just excute callback right now.
 870   if(!taskAdded)
 871   {
 872     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "CompletedCallback id[%u] executed now due to no task exist\n", tasksCompletedId);
 873
 874     mTasksCompletedImpl->CheckTasksCompletedCallbackCompleted(tasksCompletedId);
 875   }
 876   return tasksCompletedId;
 877 }
 878
 879 bool AsyncTaskManager::RemoveCompletedCallback(Dali::AsyncTaskManager::TasksCompletedId tasksCompletedId)
 880 {
 881   return mTasksCompletedImpl->RemoveTasksCompleted(tasksCompletedId);
 882 }
 883
 884 AsyncTaskPtr AsyncTaskManager::PopNextCompletedTask()
 885 {
 886   std::vector<AsyncTaskPtr> ignoredTaskList; ///< To keep asyncTask reference so we can ensure that destructor called out of mutex.
 887
 888   AsyncTaskPtr nextCompletedTask = nullptr;
 889   {
 890     // Lock while popping task out from the queue
 891     Mutex::ScopedLock lock(mCompletedTasksMutex);
 892
 893     while(!mCompletedTasks.empty())
 894     {
 895       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "PopNextCompletedTask, completed task count : [%zu]\n", mCompletedTasks.size());
 896
 897       auto               next      = mCompletedTasks.begin();
 898       AsyncTaskPtr       nextTask  = next->first;
 899       CompletedTaskState taskState = next->second;
 900       CacheImpl::EraseTaskCache(mCacheImpl->mCompletedTasksCache, nextTask, next);
 901       mCompletedTasks.erase(next);
 902
 903       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Completed task [%p][%s] (callback required? : %d)\n", nextTask.Get(), GetTaskName(nextTask), taskState == CompletedTaskState::REQUIRE_CALLBACK);
 904
 905       if(taskState == CompletedTaskState::REQUIRE_CALLBACK)
 906       {
 907         nextCompletedTask = nextTask;
 908         break;
 909       }
 910
 911       ignoredTaskList.push_back(nextTask);
 912     }
 913
 914     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Pickup completed [%p][%s]\n", nextCompletedTask.Get(), GetTaskName(nextCompletedTask));
 915   }
 916
 917   return nextCompletedTask;
 918 }
 919
 920 void AsyncTaskManager::RegisterProcessor()
 921 {
 922   if(!mProcessorRegistered && Dali::Adaptor::IsAvailable())
 923   {
 924     Dali::Adaptor::Get().RegisterProcessor(*this);
 925     mProcessorRegistered = true;
 926   }
 927 }
 928
 929 void AsyncTaskManager::UnregisterProcessor()
 930 {
 931   if(mProcessorRegistered && Dali::Adaptor::IsAvailable())
 932   {
 933     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "UnregisterProcessor begin\n");
 934     // Keep processor at least 1 task exist.
 935     // Please be careful the order of mutex, to avoid dead lock.
 936     // TODO : Should we lock all mutex rightnow?
 937     Mutex::ScopedLock lockWait(mWaitingTasksMutex);
 938     if(mWaitingTasks.empty())
 939     {
 940       Mutex::ScopedLock lockRunning(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
 941       if(mRunningTasks.empty())
 942       {
 943         Mutex::ScopedLock lockComplete(mCompletedTasksMutex); // We can lock this mutex under mWaitingTasksMutex and mRunningTasksMutex.
 944         if(mCompletedTasks.empty())
 945         {
 946           mProcessorRegistered = false;
 947           Dali::Adaptor::Get().UnregisterProcessor(*this);
 948         }
 949       }
 950     }
 951     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "UnregisterProcessor end (registed? %d)\n", mProcessorRegistered);
 952   }
 953 }
 954
 955 void AsyncTaskManager::TasksCompleted()
 956 {
 957   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "TasksCompleted begin\n");
 958   while(AsyncTaskPtr task = PopNextCompletedTask())
 959   {
 960     DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Execute callback [%p][%s]\n", task.Get(), GetTaskName(task));
 961     CallbackBase::Execute(*(task->GetCompletedCallback()), task);
 962
 963     // Remove TasksCompleted callback trace
 964     if(mTasksCompletedImpl->IsTasksCompletedCallbackExist())
 965     {
 966       mTasksCompletedImpl->RemoveTaskTrace(task);
 967     }
 968   }
 969
 970   UnregisterProcessor();
 971   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "TasksCompleted end\n");
 972
 973   mTasksCompletedImpl->EmitCompletedTasks();
 974 }
 975
 976 void AsyncTaskManager::Process(bool postProcessor)
 977 {
 978   TasksCompleted();
 979 }
 980
 981 /// Worker thread called
 982 AsyncTaskPtr AsyncTaskManager::PopNextTaskToProcess()
 983 {
 984   // Lock while popping task out from the queue
 985   Mutex::ScopedLock lock(mWaitingTasksMutex);
 986
 987   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "PopNextTaskToProcess, waiting task count : [%zu]\n", mWaitingTasks.size());
 988
 989   // pop out the next task from the queue
 990   AsyncTaskPtr nextTask = nullptr;
 991
 992   // Fast cut if all waiting tasks are LOW priority, and we cannot excute low task anymore.
 993   if(mWaitingHighProirityTaskCounts == 0u && !mWaitingTasks.empty())
 994   {
 995     // For thread safety
 996     Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
 997
 998     if(mAvaliableLowPriorityTaskCounts == 0u)
 999     {
1000       // There are no avaliabe tasks to run now. Return nullptr.
1001       return nextTask;
1002     }
1003   }
1004
1005   for(auto iter = mWaitingTasks.begin(), endIter = mWaitingTasks.end(); iter != endIter; ++iter)
1006   {
1007     if((*iter)->IsReady())
1008     {
1009       const auto priorityType  = (*iter)->GetPriorityType();
1010       bool       taskAvaliable = priorityType == AsyncTask::PriorityType::HIGH; // Task always valid if it's priority is high
1011       if(!taskAvaliable)
1012       {
1013         // For thread safety
1014         Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
1015
1016         taskAvaliable = (mAvaliableLowPriorityTaskCounts > 0u); // priority is low, but we can use it.
1017       }
1018
1019       // Check whether we try to running same task at multiple threads.
1020       if(taskAvaliable)
1021       {
1022         Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
1023         auto              mapIter = mCacheImpl->mRunningTasksCache.find((*iter).Get());
1024         if(mapIter != mCacheImpl->mRunningTasksCache.end())
1025         {
1026           if(!mapIter->second.empty())
1027           {
1028             // Some other thread running this tasks now. Ignore it.
1029             DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Some other thread running this task [%p][%s]\n", (*iter).Get(), GetTaskName(*iter));
1030             taskAvaliable = false;
1031           }
1032         }
1033       }
1034
1035       if(taskAvaliable)
1036       {
1037         nextTask = *iter;
1038
1039         // Add Running queue
1040         {
1041           // Lock while popping task out from the queue
1042           Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex.
1043
1044           DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Waiting -> Running [%p][%s]\n", nextTask.Get(), GetTaskName(nextTask));
1045
1046           auto runningIter = mRunningTasks.insert(mRunningTasks.end(), std::make_pair(nextTask, RunningTaskState::RUNNING));
1047           CacheImpl::InsertTaskCache(mCacheImpl->mRunningTasksCache, nextTask, runningIter);
1048
1049           CacheImpl::EraseTaskCache(mCacheImpl->mWaitingTasksCache, nextTask, iter);
1050           mWaitingTasks.erase(iter);
1051
1052           // Decrease avaliable task counts if it is low priority
1053           if(priorityType == AsyncTask::PriorityType::LOW)
1054           {
1055             // We are under running task mutex. We can decrease it.
1056             --mAvaliableLowPriorityTaskCounts;
1057           }
1058         }
1059
1060         if(priorityType == AsyncTask::PriorityType::HIGH && mWaitingHighProirityTaskCounts > 0u)
1061         {
1062           // Decrease the number of waiting tasks for high priority.
1063           --mWaitingHighProirityTaskCounts;
1064         }
1065         break;
1066       }
1067     }
1068   }
1069
1070   DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Pickup process [%p][%s]\n", nextTask.Get(), GetTaskName(nextTask));
1071
1072   return nextTask;
1073 }
1074
1075 /// Worker thread called
1076 void AsyncTaskManager::CompleteTask(AsyncTaskPtr&& task)
1077 {
1078   if(task)
1079   {
1080     bool needTrigger = false;
1081
1082     // Check now whether we need to execute callback or not, for worker thread cases.
1083     if(task->GetCallbackInvocationThread() == AsyncTask::ThreadType::WORKER_THREAD)
1084     {
1085       bool notify = false;
1086
1087       // Lock while check validation of task.
1088       {
1089         Mutex::ScopedLock lock(mRunningTasksMutex);
1090
1091         auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
1092         if(mapIter != mCacheImpl->mRunningTasksCache.end())
1093         {
1094           const auto cacheIter = mapIter->second.begin();
1095           DALI_ASSERT_ALWAYS(cacheIter != mapIter->second.end());
1096
1097           const auto iter = *cacheIter;
1098           DALI_ASSERT_DEBUG(iter->first == task);
1099           if(iter->second == RunningTaskState::RUNNING)
1100           {
1101             // This task is valid.
1102             notify = true;
1103           }
1104         }
1105
1106         DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "CompleteTask [%p][%s] (is notify? : %d)\n", task.Get(), GetTaskName(task), notify);
1107       }
1108
1109       // We should execute this tasks complete callback out of mutex
1110       if(notify)
1111       {
1112         DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Execute callback on worker thread [%p][%s]\n", task.Get(), GetTaskName(task));
1113         CallbackBase::Execute(*(task->GetCompletedCallback()), task);
1114
1115         // We need to remove task trace now.
1116         if(mTasksCompletedImpl->IsTasksCompletedCallbackExist())
1117         {
1118           mTasksCompletedImpl->RemoveTaskTrace(task);
1119
1120           if(mTasksCompletedImpl->IsExecuteCallbackExist())
1121           {
1122             // We need to call EmitCompletedTasks(). Trigger main thread.
1123             needTrigger = true;
1124           }
1125         }
1126       }
1127     }
1128
1129     // Lock while adding task to the queue
1130     {
1131       bool notify = false;
1132
1133       Mutex::ScopedLock lock(mRunningTasksMutex);
1134
1135       auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
1136       if(mapIter != mCacheImpl->mRunningTasksCache.end())
1137       {
1138         const auto cacheIter = mapIter->second.begin();
1139         DALI_ASSERT_ALWAYS(cacheIter != mapIter->second.end());
1140
1141         const auto iter = *cacheIter;
1142
1143         DALI_ASSERT_DEBUG(iter->first == task);
1144         if(iter->second == RunningTaskState::RUNNING)
1145         {
1146           // This task is valid.
1147           notify = true;
1148         }
1149
1150         const auto priorityType = iter->first->GetPriorityType();
1151         // Increase avaliable task counts if it is low priority
1152         if(priorityType == AsyncTask::PriorityType::LOW)
1153         {
1154           // We are under running task mutex. We can increase it.
1155           ++mAvaliableLowPriorityTaskCounts;
1156         }
1157
1158         // Move task into completed, for ensure that AsyncTask destroy at main thread.
1159         {
1160           Mutex::ScopedLock lock(mCompletedTasksMutex); // We can lock this mutex under mRunningTasksMutex.
1161
1162           const bool callbackRequired = notify && (task->GetCallbackInvocationThread() == AsyncTask::ThreadType::MAIN_THREAD);
1163
1164           needTrigger |= callbackRequired;
1165
1166           DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Running -> Completed [%p][%s] (callback required? : %d)\n", task.Get(), GetTaskName(task), callbackRequired);
1167
1168           auto completedIter = mCompletedTasks.insert(mCompletedTasks.end(), std::make_pair(task, callbackRequired ? CompletedTaskState::REQUIRE_CALLBACK : CompletedTaskState::SKIP_CALLBACK));
1169           CacheImpl::InsertTaskCache(mCacheImpl->mCompletedTasksCache, task, completedIter);
1170
1171           CacheImpl::EraseTaskCache(mCacheImpl->mRunningTasksCache, task, iter);
1172           mRunningTasks.erase(iter);
1173
1174           if(!needTrigger)
1175           {
1176             needTrigger |= (mCompletedTasks.size() >= FORCE_TRIGGER_THRESHOLD);
1177           }
1178
1179           // Now, task is invalidate.
1180           task.Reset();
1181         }
1182       }
1183     }
1184
1185     // Wake up the main thread
1186     if(needTrigger)
1187     {
1188       DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Trigger main thread\n");
1189       mTrigger->Trigger();
1190     }
1191   }
1192 }
1193
1194 // AsyncTaskManager::TaskHelper
1195
1196 AsyncTaskManager::TaskHelper::TaskHelper(AsyncTaskManager& asyncTaskManager)
1197 : TaskHelper(std::unique_ptr<AsyncTaskThread>(new AsyncTaskThread(asyncTaskManager)), asyncTaskManager)
1198 {
1199 }
1200
1201 AsyncTaskManager::TaskHelper::TaskHelper(TaskHelper&& rhs)
1202 : TaskHelper(std::move(rhs.mProcessor), rhs.mAsyncTaskManager)
1203 {
1204 }
1205
1206 AsyncTaskManager::TaskHelper::TaskHelper(std::unique_ptr<AsyncTaskThread> processor, AsyncTaskManager& asyncTaskManager)
1207 : mProcessor(std::move(processor)),
1208   mAsyncTaskManager(asyncTaskManager)
1209 {
1210 }
1211
1212 bool AsyncTaskManager::TaskHelper::Request()
1213 {
1214   return mProcessor->Request();
1215 }
1216 } // namespace Adaptor
1217
1218 } // namespace Internal
1219
1220 } // namespace Dali