{
namespace
{
+constexpr auto FORCE_TRIGGER_THRESHOLD = 128u; ///< Trigger TasksCompleted() forcely if the number of completed task contain too much.
+
constexpr auto DEFAULT_NUMBER_OF_ASYNC_THREADS = size_t{8u};
constexpr auto NUMBER_OF_ASYNC_THREADS_ENV = "DALI_ASYNC_MANAGER_THREAD_POOL_SIZE";
: mConditionalWait(),
mAsyncTaskManager(asyncTaskManager),
mLogFactory(Dali::Adaptor::Get().GetLogFactory()),
+ mTraceFactory(Dali::Adaptor::Get().GetTraceFactory()),
mDestroyThread(false),
mIsThreadStarted(false),
mIsThreadIdle(true)
SetThreadName("AsyncTaskThread");
#endif
mLogFactory.InstallLogFunction();
+ mTraceFactory.InstallTraceFunction();
while(!mDestroyThread)
{
DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Thread[%u] Complete task [%p]\n", threadId, task.Get());
if(!mDestroyThread)
{
- mAsyncTaskManager.CompleteTask(task);
+ mAsyncTaskManager.CompleteTask(std::move(task));
}
}
}
AsyncTaskManager& mManager; ///< Owner of this CacheImpl.
// Keep cache iterators as list since we take tasks by FIFO as default.
- using TaskCacheContainer = std::unordered_map<const AsyncTask*, std::list<AsyncTaskContainer::iterator>>;
- using RunningTaskCacheContainer = std::unordered_map<const AsyncTask*, std::list<AsyncRunningTaskContainer::iterator>>;
+ using TaskCacheContainer = std::unordered_map<const AsyncTask*, std::list<AsyncTaskContainer::iterator>>;
+ using RunningTaskCacheContainer = std::unordered_map<const AsyncTask*, std::list<AsyncRunningTaskContainer::iterator>>;
+ using CompletedTaskCacheContainer = std::unordered_map<const AsyncTask*, std::list<AsyncCompletedTaskContainer::iterator>>;
- TaskCacheContainer mWaitingTasksCache; ///< The cache of tasks and iterator for waiting to async process. Must be locked under mWaitingTasksMutex.
- RunningTaskCacheContainer mRunningTasksCache; ///< The cache of tasks and iterator for running tasks. Must be locked under mRunningTasksMutex.
- TaskCacheContainer mCompletedTasksCache; ///< The cache of tasks and iterator for completed async process. Must be locked under mCompletedTasksMutex.
+ TaskCacheContainer mWaitingTasksCache; ///< The cache of tasks and iterator for waiting to async process. Must be locked under mWaitingTasksMutex.
+ RunningTaskCacheContainer mRunningTasksCache; ///< The cache of tasks and iterator for running tasks. Must be locked under mRunningTasksMutex.
+ CompletedTaskCacheContainer mCompletedTasksCache; ///< The cache of tasks and iterator for completed async process. Must be locked under mCompletedTasksMutex.
};
// AsyncTaskManager
Dali::Adaptor::Get().UnregisterProcessor(*this);
}
+ // Join all threads.
mTasks.Clear();
+
+ // Remove cache impl after all threads are join.
+ mCacheImpl.reset();
+
+ // Remove tasks after CacheImpl removed
+ mWaitingTasks.clear();
+ mRunningTasks.clear();
+ mCompletedTasks.clear();
}
void AsyncTaskManager::AddTask(AsyncTaskPtr task)
auto mapIter = mCacheImpl->mWaitingTasksCache.find(task.Get());
if(mapIter != mCacheImpl->mWaitingTasksCache.end())
{
- for(auto iterator : mapIter->second)
+ for(auto& iterator : mapIter->second)
{
DALI_ASSERT_DEBUG((*iterator) == task);
if((*iterator)->GetPriorityType() == AsyncTask::PriorityType::HIGH)
auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
if(mapIter != mCacheImpl->mRunningTasksCache.end())
{
- for(auto iterator : mapIter->second)
+ for(auto& iterator : mapIter->second)
{
DALI_ASSERT_DEBUG((*iterator).first == task);
// We cannot erase container. Just mark as canceled.
auto mapIter = mCacheImpl->mCompletedTasksCache.find(task.Get());
if(mapIter != mCacheImpl->mCompletedTasksCache.end())
{
- for(auto iterator : mapIter->second)
+ for(auto& iterator : mapIter->second)
{
- DALI_ASSERT_DEBUG((*iterator) == task);
+ DALI_ASSERT_DEBUG(iterator->first == task);
mCompletedTasks.erase(iterator);
}
CacheImpl::EraseAllTaskCache(mCacheImpl->mCompletedTasksCache, task);
// Lock while popping task out from the queue
Mutex::ScopedLock lock(mCompletedTasksMutex);
- if(mCompletedTasks.empty())
+ AsyncTaskPtr nextCompletedTask = nullptr;
+
+ while(!mCompletedTasks.empty())
{
- return AsyncTaskPtr();
- }
+ DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "PopNextCompletedTask, completed task count : [%zu]\n", mCompletedTasks.size());
- DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "PopNextCompletedTask, completed task count : [%zu]\n", mCompletedTasks.size());
+ auto next = mCompletedTasks.begin();
+ AsyncTaskPtr nextTask = next->first;
+ CompletedTaskState taskState = next->second;
+ CacheImpl::EraseTaskCache(mCacheImpl->mCompletedTasksCache, nextTask, next);
+ mCompletedTasks.erase(next);
- auto next = mCompletedTasks.begin();
- AsyncTaskPtr nextTask = *next;
- CacheImpl::EraseTaskCache(mCacheImpl->mCompletedTasksCache, nextTask, next);
- mCompletedTasks.erase(next);
+ DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Completed task [%p] (callback required? : %d)\n", nextTask.Get(), taskState == CompletedTaskState::REQUIRE_CALLBACK);
- DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Pickup completed [%p]\n", nextTask.Get());
+ if(taskState == CompletedTaskState::REQUIRE_CALLBACK)
+ {
+ nextCompletedTask = nextTask;
+ break;
+ }
+ }
- return nextTask;
+ DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::General, "Pickup completed [%p]\n", nextCompletedTask.Get());
+
+ return nextCompletedTask;
}
void AsyncTaskManager::UnregisterProcessor()
auto runningIter = mRunningTasks.insert(mRunningTasks.end(), std::make_pair(nextTask, RunningTaskState::RUNNING));
CacheImpl::InsertTaskCache(mCacheImpl->mRunningTasksCache, nextTask, runningIter);
+ CacheImpl::EraseTaskCache(mCacheImpl->mWaitingTasksCache, nextTask, iter);
+ mWaitingTasks.erase(iter);
+
// Decrease avaliable task counts if it is low priority
if(priorityType == AsyncTask::PriorityType::LOW)
{
// Decrease the number of waiting tasks for high priority.
--mWaitingHighProirityTaskCounts;
}
-
- CacheImpl::EraseTaskCache(mCacheImpl->mWaitingTasksCache, nextTask, iter);
- mWaitingTasks.erase(iter);
break;
}
}
}
/// Worker thread called
-void AsyncTaskManager::CompleteTask(AsyncTaskPtr task)
+void AsyncTaskManager::CompleteTask(AsyncTaskPtr&& task)
{
bool notify = false;
- // Lock while adding task to the queue
if(task)
{
- Mutex::ScopedLock lock(mRunningTasksMutex);
+ bool needTrigger = (task->GetCallbackInvocationThread() == AsyncTask::ThreadType::MAIN_THREAD);
- auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
- if(mapIter != mCacheImpl->mRunningTasksCache.end())
+ // Lock while check validation of task.
{
- const auto cacheIter = mapIter->second.begin();
- DALI_ASSERT_ALWAYS(cacheIter != mapIter->second.end());
+ Mutex::ScopedLock lock(mRunningTasksMutex);
- const auto iter = *cacheIter;
- DALI_ASSERT_DEBUG(iter->first == task);
- if(iter->second == RunningTaskState::RUNNING)
+ auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
+ if(mapIter != mCacheImpl->mRunningTasksCache.end())
{
- // This task is valid.
- notify = true;
- }
+ const auto cacheIter = mapIter->second.begin();
+ DALI_ASSERT_ALWAYS(cacheIter != mapIter->second.end());
- const auto priorityType = iter->first->GetPriorityType();
- // Increase avaliable task counts if it is low priority
- if(priorityType == AsyncTask::PriorityType::LOW)
- {
- // We are under running task mutex. We can increase it.
- ++mAvaliableLowPriorityTaskCounts;
+ const auto iter = *cacheIter;
+ DALI_ASSERT_DEBUG(iter->first == task);
+ if(iter->second == RunningTaskState::RUNNING)
+ {
+ // This task is valid.
+ notify = true;
+ }
}
- CacheImpl::EraseTaskCache(mCacheImpl->mRunningTasksCache, task, iter);
- mRunningTasks.erase(iter);
+
+ DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "CompleteTask [%p] (is notify? : %d)\n", task.Get(), notify);
}
- DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "CompleteTask [%p] (is notify? : %d)\n", task.Get(), notify);
+ // We should execute this tasks complete callback out of mutex
+ if(notify && task->GetCallbackInvocationThread() == AsyncTask::ThreadType::WORKER_THREAD)
+ {
+ DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Execute callback on worker thread [%p]\n", task.Get());
+ CallbackBase::Execute(*(task->GetCompletedCallback()), task);
+ }
- // We should move the task to compeleted task under mRunningTaskMutex.
- if(notify && task->GetCallbackInvocationThread() == AsyncTask::ThreadType::MAIN_THREAD)
+ // Lock while adding task to the queue
{
- Mutex::ScopedLock lock(mCompletedTasksMutex); // We can lock this mutex under mRunningTasksMutex.
+ Mutex::ScopedLock lock(mRunningTasksMutex);
+
+ auto mapIter = mCacheImpl->mRunningTasksCache.find(task.Get());
+ if(mapIter != mCacheImpl->mRunningTasksCache.end())
+ {
+ const auto cacheIter = mapIter->second.begin();
+ DALI_ASSERT_ALWAYS(cacheIter != mapIter->second.end());
+
+ const auto iter = *cacheIter;
+ const auto priorityType = iter->first->GetPriorityType();
+ // Increase avaliable task counts if it is low priority
+ if(priorityType == AsyncTask::PriorityType::LOW)
+ {
+ // We are under running task mutex. We can increase it.
+ ++mAvaliableLowPriorityTaskCounts;
+ }
+
+ // Move task into completed, for ensure that AsyncTask destroy at main thread.
+ {
+ Mutex::ScopedLock lock(mCompletedTasksMutex); // We can lock this mutex under mRunningTasksMutex.
+
+ const bool callbackRequired = notify && (task->GetCallbackInvocationThread() == AsyncTask::ThreadType::MAIN_THREAD);
+
+ DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Running -> Completed [%p] (callback required? : %d)\n", task.Get(), callbackRequired);
- DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Running -> Completed [%p]\n", task.Get());
+ auto completedIter = mCompletedTasks.insert(mCompletedTasks.end(), std::make_pair(task, callbackRequired ? CompletedTaskState::REQUIRE_CALLBACK : CompletedTaskState::SKIP_CALLBACK));
+ CacheImpl::InsertTaskCache(mCacheImpl->mCompletedTasksCache, task, completedIter);
- auto completedIter = mCompletedTasks.insert(mCompletedTasks.end(), task);
- CacheImpl::InsertTaskCache(mCacheImpl->mCompletedTasksCache, task, completedIter);
+ CacheImpl::EraseTaskCache(mCacheImpl->mRunningTasksCache, task, iter);
+ mRunningTasks.erase(iter);
+
+ if(!needTrigger)
+ {
+ needTrigger |= (mCompletedTasks.size() >= FORCE_TRIGGER_THRESHOLD);
+ }
+
+ // Now, task is invalidate.
+ task.Reset();
+ }
+ }
}
- }
- // We should execute this tasks complete callback out of mutex
- if(notify)
- {
- if(task->GetCallbackInvocationThread() == AsyncTask::ThreadType::MAIN_THREAD)
+ // Wake up the main thread
+ if(needTrigger)
{
- // wake up the main thread
DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Trigger main thread\n");
mTrigger->Trigger();
}
- else // task->GetCallbackInvocationThread() == AsyncTask::ThreadType::WORKER_THREAD
- {
- DALI_LOG_INFO(gAsyncTasksManagerLogFilter, Debug::Verbose, "Execute callback on worker thread [%p]\n", task.Get());
- CallbackBase::Execute(*(task->GetCompletedCallback()), task);
- }
}
}
projects / platform / core / uifw / dali-adaptor.git / blobdiff