From: Adeel Kazmi Date: Fri, 10 Nov 2023 17:23:12 +0000 (+0000) Subject: Merge "Implement AsyncTaskManager for toolkit UTC" into devel/master X-Git-Tag: dali_2.2.53~9 X-Git-Url: http://review.tizen.org/git/?a=commitdiff_plain;h=dd80ca79aa5c505d164def71c78d0a50ae3ec19b;hp=4979155c2846b00c77ea836bec6a5db2a880ef08;p=platform%2Fcore%2Fuifw%2Fdali-toolkit.git Merge "Implement AsyncTaskManager for toolkit UTC" into devel/master --- diff --git a/automated-tests/src/dali-physics2d/CMakeLists.txt b/automated-tests/src/dali-physics2d/CMakeLists.txt index e796089..e72db08 100644 --- a/automated-tests/src/dali-physics2d/CMakeLists.txt +++ b/automated-tests/src/dali-physics2d/CMakeLists.txt @@ -16,6 +16,7 @@ SET(TEST_HARNESS_DIR "../dali-toolkit/dali-toolkit-test-utils") SET(TEST_HARNESS_SOURCES ${TEST_HARNESS_DIR}/toolkit-adaptor.cpp ${TEST_HARNESS_DIR}/toolkit-application.cpp + ${TEST_HARNESS_DIR}/toolkit-async-task-manager.cpp ${TEST_HARNESS_DIR}/toolkit-direct-rendering-egl.cpp ${TEST_HARNESS_DIR}/toolkit-event-thread-callback.cpp ${TEST_HARNESS_DIR}/toolkit-environment-variable.cpp diff --git a/automated-tests/src/dali-physics3d/CMakeLists.txt b/automated-tests/src/dali-physics3d/CMakeLists.txt index a863fd8..01ecfc0 100644 --- a/automated-tests/src/dali-physics3d/CMakeLists.txt +++ b/automated-tests/src/dali-physics3d/CMakeLists.txt @@ -15,6 +15,7 @@ SET(TEST_HARNESS_DIR "../dali-toolkit/dali-toolkit-test-utils") SET(TEST_HARNESS_SOURCES ${TEST_HARNESS_DIR}/toolkit-adaptor.cpp ${TEST_HARNESS_DIR}/toolkit-application.cpp + ${TEST_HARNESS_DIR}/toolkit-async-task-manager.cpp ${TEST_HARNESS_DIR}/toolkit-direct-rendering-egl.cpp ${TEST_HARNESS_DIR}/toolkit-event-thread-callback.cpp ${TEST_HARNESS_DIR}/toolkit-environment-variable.cpp diff --git a/automated-tests/src/dali-scene3d-internal/CMakeLists.txt b/automated-tests/src/dali-scene3d-internal/CMakeLists.txt index 561e753..b050314 100755 --- a/automated-tests/src/dali-scene3d-internal/CMakeLists.txt +++ b/automated-tests/src/dali-scene3d-internal/CMakeLists.txt @@ -26,6 +26,7 @@ SET(TEST_HARNESS_DIR "../dali-toolkit/dali-toolkit-test-utils") SET(TEST_HARNESS_SOURCES ${TEST_HARNESS_DIR}/toolkit-adaptor.cpp ${TEST_HARNESS_DIR}/toolkit-application.cpp + ${TEST_HARNESS_DIR}/toolkit-async-task-manager.cpp ${TEST_HARNESS_DIR}/toolkit-event-thread-callback.cpp ${TEST_HARNESS_DIR}/toolkit-environment-variable.cpp ${TEST_HARNESS_DIR}/toolkit-input-method-context.cpp diff --git a/automated-tests/src/dali-scene3d/CMakeLists.txt b/automated-tests/src/dali-scene3d/CMakeLists.txt index 1cb9f3c..1269cb2 100755 --- a/automated-tests/src/dali-scene3d/CMakeLists.txt +++ b/automated-tests/src/dali-scene3d/CMakeLists.txt @@ -50,6 +50,7 @@ SET(TEST_HARNESS_DIR "../dali-toolkit/dali-toolkit-test-utils") SET(TEST_HARNESS_SOURCES ${TEST_HARNESS_DIR}/toolkit-adaptor.cpp ${TEST_HARNESS_DIR}/toolkit-application.cpp + ${TEST_HARNESS_DIR}/toolkit-async-task-manager.cpp ${TEST_HARNESS_DIR}/toolkit-event-thread-callback.cpp ${TEST_HARNESS_DIR}/toolkit-environment-variable.cpp ${TEST_HARNESS_DIR}/toolkit-input-method-context.cpp diff --git a/automated-tests/src/dali-shader-generator/CMakeLists.txt b/automated-tests/src/dali-shader-generator/CMakeLists.txt index 517b7d3..30502d8 100644 --- a/automated-tests/src/dali-shader-generator/CMakeLists.txt +++ b/automated-tests/src/dali-shader-generator/CMakeLists.txt @@ -21,6 +21,7 @@ SET(TEST_HARNESS_DIR "../dali-toolkit/dali-toolkit-test-utils") SET(TEST_HARNESS_SOURCES ${TEST_HARNESS_DIR}/toolkit-adaptor.cpp ${TEST_HARNESS_DIR}/toolkit-application.cpp + ${TEST_HARNESS_DIR}/toolkit-async-task-manager.cpp ${TEST_HARNESS_DIR}/toolkit-event-thread-callback.cpp ${TEST_HARNESS_DIR}/toolkit-environment-variable.cpp ${TEST_HARNESS_DIR}/toolkit-input-method-context.cpp diff --git a/automated-tests/src/dali-toolkit-internal/CMakeLists.txt b/automated-tests/src/dali-toolkit-internal/CMakeLists.txt index 582d39c..05354c0 100755 --- a/automated-tests/src/dali-toolkit-internal/CMakeLists.txt +++ b/automated-tests/src/dali-toolkit-internal/CMakeLists.txt @@ -61,6 +61,7 @@ ENDIF() SET(TEST_HARNESS_SOURCES ../dali-toolkit/dali-toolkit-test-utils/toolkit-adaptor.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-application.cpp + ../dali-toolkit/dali-toolkit-test-utils/toolkit-async-task-manager.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-clipboard.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-event-thread-callback.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-environment-variable.cpp diff --git a/automated-tests/src/dali-toolkit-styling/CMakeLists.txt b/automated-tests/src/dali-toolkit-styling/CMakeLists.txt index 814a4de..dcce7cd 100644 --- a/automated-tests/src/dali-toolkit-styling/CMakeLists.txt +++ b/automated-tests/src/dali-toolkit-styling/CMakeLists.txt @@ -15,6 +15,7 @@ SET(TEST_HARNESS_SOURCES ../dali-toolkit/dali-toolkit-test-utils/test-harness.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-adaptor.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-application.cpp + ../dali-toolkit/dali-toolkit-test-utils/toolkit-async-task-manager.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-clipboard.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-color-controller.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-event-thread-callback.cpp diff --git a/automated-tests/src/dali-toolkit-third-party/CMakeLists.txt b/automated-tests/src/dali-toolkit-third-party/CMakeLists.txt index 994ec67..9633069 100644 --- a/automated-tests/src/dali-toolkit-third-party/CMakeLists.txt +++ b/automated-tests/src/dali-toolkit-third-party/CMakeLists.txt @@ -13,6 +13,7 @@ SET(TC_SOURCES # List of test harness files (Won't get parsed for test cases) SET(TEST_HARNESS_SOURCES ../dali-toolkit/dali-toolkit-test-utils/toolkit-application.cpp + ../dali-toolkit/dali-toolkit-test-utils/toolkit-async-task-manager.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-clipboard.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-event-thread-callback.cpp ../dali-toolkit/dali-toolkit-test-utils/toolkit-environment-variable.cpp diff --git a/automated-tests/src/dali-toolkit/CMakeLists.txt b/automated-tests/src/dali-toolkit/CMakeLists.txt index ec10a33..8d33504 100755 --- a/automated-tests/src/dali-toolkit/CMakeLists.txt +++ b/automated-tests/src/dali-toolkit/CMakeLists.txt @@ -100,6 +100,7 @@ SET(TC_SOURCES SET(TEST_HARNESS_SOURCES dali-toolkit-test-utils/toolkit-adaptor.cpp dali-toolkit-test-utils/toolkit-application.cpp + dali-toolkit-test-utils/toolkit-async-task-manager.cpp dali-toolkit-test-utils/toolkit-canvas-renderer.cpp dali-toolkit-test-utils/toolkit-clipboard.cpp dali-toolkit-test-utils/toolkit-direct-rendering-egl.cpp diff --git a/automated-tests/src/dali-toolkit/dali-toolkit-test-utils/toolkit-adaptor.cpp b/automated-tests/src/dali-toolkit/dali-toolkit-test-utils/toolkit-adaptor.cpp index c675fef..f6b8f8b 100644 --- a/automated-tests/src/dali-toolkit/dali-toolkit-test-utils/toolkit-adaptor.cpp +++ b/automated-tests/src/dali-toolkit/dali-toolkit-test-utils/toolkit-adaptor.cpp @@ -28,6 +28,7 @@ #include #include #include +#include #include #include @@ -65,6 +66,10 @@ Adaptor::Adaptor() Adaptor::~Adaptor() { gAdaptor = nullptr; + + // Ensure all threads and not-excuted tasks are destroyed. + // TODO : we'd better make some singletone service for toolkit UTC in future. + Test::AsyncTaskManager::DestroyAsyncTaskManager(); } void Adaptor::Start(Dali::Window window) diff --git a/automated-tests/src/dali-toolkit/dali-toolkit-test-utils/toolkit-async-task-manager.cpp b/automated-tests/src/dali-toolkit/dali-toolkit-test-utils/toolkit-async-task-manager.cpp new file mode 100644 index 0000000..6678d8d --- /dev/null +++ b/automated-tests/src/dali-toolkit/dali-toolkit-test-utils/toolkit-async-task-manager.cpp @@ -0,0 +1,1092 @@ +/* + * Copyright (c) 2023 Samsung Electronics Co., Ltd. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +// CLASS HEADER +#include + +// EXTERNAL INCLUDE +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// INTERNAL INCLUDE +#include +#include +#include + +namespace Dali +{ +namespace Internal +{ +namespace Adaptor +{ +namespace +{ +std::atomic_uint32_t gThreadId = 0u; +} + +class AsyncTaskThread; + +class AsyncTaskManager : public Dali::BaseObject +{ +public: + static Dali::AsyncTaskManager Get(); + + AsyncTaskManager(); + ~AsyncTaskManager(); + +public: + void AddTask(AsyncTaskPtr task); + void RemoveTask(AsyncTaskPtr task); + + Dali::AsyncTaskManager::TasksCompletedId SetCompletedCallback(CallbackBase* callback, Dali::AsyncTaskManager::CompletedCallbackTraceMask mask); + bool RemoveCompletedCallback(Dali::AsyncTaskManager::TasksCompletedId tasksCompletedId); + +public: + AsyncTaskPtr PopNextCompletedTask(); + void TaskCompleted(); + void TaskAllCompleted(); + +public: // Worker thread called method + AsyncTaskPtr PopNextTaskToProcess(); + void CompleteTask(AsyncTaskPtr&& task); + +private: /** + * @brief Helper class to keep the relation between AsyncTaskThread and corresponding container + */ + class TaskHelper + { + public: + /** + * @brief Create an TaskHelper. + * + * @param[in] asyncTaskManager Reference to the AsyncTaskManager + */ + TaskHelper(AsyncTaskManager& asyncTaskManager); + + /** + * @brief Request the thread to process the task. + * @return True if the request succeeds, otherwise false. + */ + bool Request(); + + public: + TaskHelper(const TaskHelper&) = delete; + TaskHelper& operator=(const TaskHelper&) = delete; + + TaskHelper(TaskHelper&& rhs); + TaskHelper& operator=(TaskHelper&& rhs) = delete; + + private: + /** + * @brief Main constructor that used by all other constructors + */ + TaskHelper(std::unique_ptr processor, AsyncTaskManager& asyncTaskManager); + + private: + std::unique_ptr mProcessor; + AsyncTaskManager& mAsyncTaskManager; + }; + + /** + * @brief State of running task + */ + enum class RunningTaskState + { + RUNNING = 0, ///< Running task + CANCELED = 1, ///< Canceled by user + }; + + /** + * @brief State of complete task + */ + enum class CompletedTaskState + { + REQUIRE_CALLBACK = 0, ///< Need to execute callback when completed task process. + SKIP_CALLBACK = 1, ///< Do not execute callback + }; + +private: + // Undefined + AsyncTaskManager(const AsyncTaskManager& manager); + + // Undefined + AsyncTaskManager& operator=(const AsyncTaskManager& manager); + +private: + // Keep Task as list since we take tasks by FIFO as default. + using AsyncTaskContainer = std::list; + + using AsyncRunningTaskPair = std::pair; + using AsyncRunningTaskContainer = std::list; + + using AsyncCompletedTaskPair = std::pair; + using AsyncCompletedTaskContainer = std::list; + + AsyncTaskContainer mWaitingTasks; ///< The queue of the tasks waiting to async process. Must be locked under mWaitingTasksMutex. + AsyncRunningTaskContainer mRunningTasks; ///< The queue of the running tasks. Must be locked under mRunningTasksMutex. + AsyncCompletedTaskContainer mCompletedTasks; ///< The queue of the tasks with the async process. Must be locked under mCompletedTasksMutex. + + RoundRobinContainerView mTasks; + + Dali::Mutex mWaitingTasksMutex; ///< Mutex for mWaitingTasks. We can lock mRunningTasksMutex and mCompletedTasksMutex under this scope. + Dali::Mutex mRunningTasksMutex; ///< Mutex for mRunningTasks. We can lock mCompletedTasksMutex under this scope. + Dali::Mutex mCompletedTasksMutex; ///< Mutex for mCompletedTasks. We cannot lock any mutex under this scope. + + std::unique_ptr mTrigger; + + struct TasksCompletedImpl; + std::unique_ptr mTasksCompletedImpl; ///< TaskS completed signal interface for AsyncTaskManager. +}; + +inline Internal::Adaptor::AsyncTaskManager& GetImplementation(Dali::AsyncTaskManager& obj) +{ + DALI_ASSERT_ALWAYS(obj && "AsyncTaskManager is empty"); + + Dali::BaseObject& handle = obj.GetBaseObject(); + + return static_cast(handle); +} + +inline const Internal::Adaptor::AsyncTaskManager& GetImplementation(const Dali::AsyncTaskManager& obj) +{ + DALI_ASSERT_ALWAYS(obj && "AsyncTaskManager is empty"); + + const Dali::BaseObject& handle = obj.GetBaseObject(); + + return static_cast(handle); +} + +/********************************************************************************/ +/********************************* INTERNAL CLASS *****************************/ +/********************************************************************************/ + +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"; + +size_t GetNumberOfThreads(const char* environmentVariable, size_t defaultValue) +{ + auto numberString = EnvironmentVariable::GetEnvironmentVariable(environmentVariable); + auto numberOfThreads = numberString ? std::strtoul(numberString, nullptr, 10) : 0; + constexpr auto MAX_NUMBER_OF_THREADS = 16u; + DALI_ASSERT_DEBUG(numberOfThreads <= MAX_NUMBER_OF_THREADS); + return (numberOfThreads > 0 && numberOfThreads <= MAX_NUMBER_OF_THREADS) ? numberOfThreads : defaultValue; +} + +thread_local Dali::AsyncTaskManager gAsyncTaskManager; +} // namespace + +/** + * The worker thread for async process + */ +class AsyncTaskThread : public Thread +{ +public: + /** + * Constructor. + */ + AsyncTaskThread(AsyncTaskManager& asyncTaskManager) + : mConditionalWait(), + mAsyncTaskManager(asyncTaskManager), + mThreadId(++gThreadId), + mDestroyThread(false), + mIsThreadStarted(false), + mIsThreadIdle(true) + { + } + + /** + * Destructor. + */ + ~AsyncTaskThread() override + { + // Stop the thread + { + ConditionalWait::ScopedLock lock(mConditionalWait); + mDestroyThread = true; + mConditionalWait.Notify(lock); + } + + Join(); + } + + /** + * @brief Request the thread to process the task. + * @return True if the request is successed, otherwise false. + */ + bool Request() + { + if(!mIsThreadStarted) + { + Start(); + mIsThreadStarted = true; + } + + { + // Lock while adding task to the queue + ConditionalWait::ScopedLock lock(mConditionalWait); + + if(mIsThreadIdle) + { + mIsThreadIdle = false; + + // wake up the thread + mConditionalWait.Notify(lock); + return true; + } + } + + return false; + } + +protected: + /** + * The entry function of the worker thread. + */ + void Run() override + { + { + char temp[100]; + snprintf(temp, 100, "AsyncTaskThread[%u]", mThreadId); + SetThreadName(temp); + } + + while(!mDestroyThread) + { + AsyncTaskPtr task = mAsyncTaskManager.PopNextTaskToProcess(); + if(!task) + { + ConditionalWait::ScopedLock lock(mConditionalWait); + if(!mDestroyThread) + { + mIsThreadIdle = true; + mConditionalWait.Wait(lock); + } + } + else + { + task->Process(); + if(!mDestroyThread) + { + mAsyncTaskManager.CompleteTask(std::move(task)); + } + } + } + } + +private: + // Undefined + AsyncTaskThread(const AsyncTaskThread& thread) = delete; + + // Undefined + AsyncTaskThread& operator=(const AsyncTaskThread& thread) = delete; + +private: + ConditionalWait mConditionalWait; + AsyncTaskManager& mAsyncTaskManager; + uint32_t mThreadId; + bool mDestroyThread; + bool mIsThreadStarted; + bool mIsThreadIdle; +}; + +// AsyncTaskManager::TasksCompletedImpl + +struct AsyncTaskManager::TasksCompletedImpl +{ + TasksCompletedImpl(AsyncTaskManager& manager, EventThreadCallback* trigger) + : mManager(manager), + mTrigger(trigger), + mEmitCompletedTaskTriggered(false) + { + } + +public: + /** + * @brief Create new tasks completed id and. + * @post AppendTaskTrace or CheckTasksCompletedCallbackCompleted should be called. + * @param[in] callback The callback that want to be executed when we notify that all tasks completed. + */ + Dali::AsyncTaskManager::TasksCompletedId GenerateTasksCompletedId(CallbackBase* callback) + { + // Lock while adding tasks completed callback list to the queue + Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex); + + auto id = mTasksCompletedCount++; + DALI_ASSERT_ALWAYS(mTasksCompletedCallbackList.find(id) == mTasksCompletedCallbackList.end()); + + mTasksCompletedCallbackList.insert({id, CallbackData(callback)}); + return id; + } + + /** + * @brief Append task that will be trace. + * @post RemoveTaskTrace should be called. + * @param[in] id The id of tasks completed. + * @param[in] task The task want to trace. + */ + void AppendTaskTrace(Dali::AsyncTaskManager::TasksCompletedId id, AsyncTaskPtr task) + { + // Lock while adding tasks completed callback list to the queue + Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex); + + auto iter = mTasksCompletedCallbackList.find(id); + if(iter == mTasksCompletedCallbackList.end()) + { + // This task is already erased. Ignore. + return; + } + + auto& callbackData = iter->second; + + auto jter = callbackData.mTasks.find(task.Get()); + + if(jter != callbackData.mTasks.end()) + { + // Increase reference count. + ++(jter->second); + } + else + { + callbackData.mTasks.insert({task.Get(), 1u}); + } + } + + /** + * @brief Remove all task that were traced. + * @param[in] task The task want to remove trace. + * @param[in] taskCount The number of tasks that will be removed. + */ + void RemoveTaskTrace(AsyncTaskPtr task, uint32_t count = 1u) + { + if(count == 0u) + { + return; + } + + // Lock while removing tasks completed callback list to the queue + Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex); + + for(auto iter = mTasksCompletedCallbackList.begin(); iter != mTasksCompletedCallbackList.end();) + { + auto& callbackData = iter->second; + bool eraseCallbackData = false; + + auto jter = callbackData.mTasks.find(task.Get()); + + if(jter != callbackData.mTasks.end()) + { + if(jter->second <= count) + { + callbackData.mTasks.erase(jter); + + if(callbackData.mTasks.empty()) + { + eraseCallbackData = true; + + // Move callback base into list. + // (To avoid task container changed during callback emit) + RegisterTasksCompletedCallback(std::move(callbackData.mCallback), iter->first); + + iter = mTasksCompletedCallbackList.erase(iter); + } + } + else + { + jter->second -= count; + } + } + + if(!eraseCallbackData) + { + ++iter; + } + } + } + + /** + * @brief Check whether current TasksCompletedId completed or not. + * @param[in] id The id of tasks completed. + * @return True if all tasks are completed so we need to execute callback soon. False otherwise. + */ + bool CheckTasksCompletedCallbackCompleted(Dali::AsyncTaskManager::TasksCompletedId id) + { + // Lock while removing tasks completed callback list to the queue + Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex); + + auto iter = mTasksCompletedCallbackList.find(id); + if(iter != mTasksCompletedCallbackList.end()) + { + auto& callbackData = iter->second; + if(callbackData.mTasks.empty()) + { + // Move callback base into list. + // (To avoid task container changed during callback emit) + RegisterTasksCompletedCallback(std::move(callbackData.mCallback), iter->first); + + iter = mTasksCompletedCallbackList.erase(iter); + + return true; + } + } + + return false; + } + + /** + * @brief Remove taskS completed callbacks by id. + * @param[in] id The id of taskS completed. + * @return True if taskS completed id removed. False otherwise. + */ + bool RemoveTasksCompleted(Dali::AsyncTaskManager::TasksCompletedId id) + { + // Lock while removing taskS completed callback list to the queue + Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex); + + auto iter = mTasksCompletedCallbackList.find(id); + if(iter == mTasksCompletedCallbackList.end()) + { + // This task is already erased, or completed. + // Erase from completed excute callback list. + + // Lock while removing excute callback list to the queue + Mutex::ScopedLock lock(mExcuteCallbacksMutex); + + for(auto iter = mExcuteCallbackList.begin(); iter != mExcuteCallbackList.end();) + { + if(iter->second == id) + { + iter = mExcuteCallbackList.erase(iter); + + return true; + } + else + { + ++iter; + } + } + + // This task is alread erased and completed. Ignore. + return false; + } + + mTasksCompletedCallbackList.erase(iter); + + return true; + } + + /** + * @brief Emit all completed callbacks. + * @note This API should be called at event thread. + */ + void EmitCompletedTasks() + { + ExecuteCallbackContainer executeCallbackList; + { + // Lock while removing excute callback list to the queue + Mutex::ScopedLock lock(mExcuteCallbacksMutex); + + mEmitCompletedTaskTriggered = false; + + // Copy callback lists, for let we execute callbacks out of mutex + executeCallbackList = std::move(mExcuteCallbackList); + mExcuteCallbackList.clear(); + } + + if(!executeCallbackList.empty()) + { + // Execute all callbacks + for(auto&& callbackPair : executeCallbackList) + { + auto& callback = callbackPair.first; + auto id = callbackPair.second; + + Dali::CallbackBase::Execute(*callback, id); + } + } + } + + /** + * @brief Check whether there is some completed signal what we need to trace, or not. + * @return True if mTasksCompletedCallbackList is not empty. False otherwise. + */ + bool IsTasksCompletedCallbackExist() + { + Mutex::ScopedLock lock(mTasksCompletedCallbacksMutex); + return !mTasksCompletedCallbackList.empty(); + } + + /** + * @brief Check whether there is some completed signal what we need to execute, or not. + * @return True if mExcuteCallbackList is not empty. False otherwise. + */ + bool IsExecuteCallbackExist() + { + Mutex::ScopedLock lock(mExcuteCallbacksMutex); + return !mExcuteCallbackList.empty(); + } + +private: + void RegisterTasksCompletedCallback(std::unique_ptr callback, Dali::AsyncTaskManager::TasksCompletedId id) + { + // Lock while adding excute callback list to the queue + Mutex::ScopedLock lock(mExcuteCallbacksMutex); + + mExcuteCallbackList.emplace_back(std::move(callback), id); + + if(!mEmitCompletedTaskTriggered) + { + mEmitCompletedTaskTriggered = true; + mTrigger->Trigger(); + } + } + +private: + struct CallbackData + { + public: + CallbackData(CallbackBase* callback) + : mCallback(callback), + mTasks() + { + } + + CallbackData(CallbackData&& rhs) noexcept + : mCallback(std::move(rhs.mCallback)), + mTasks(std::move(rhs.mTasks)) + { + } + + CallbackData& operator=(CallbackData&& rhs) noexcept + { + if(this != &rhs) + { + mCallback = std::move(rhs.mCallback); + mTasks = std::move(rhs.mTasks); + } + + return *this; + } + + private: + // Delete copy operator. + CallbackData(const CallbackData& rhs) = delete; + CallbackData& operator=(const CallbackData& rhs) = delete; + + public: + std::unique_ptr mCallback; + std::unordered_map mTasks; + }; + +private: + AsyncTaskManager& mManager; ///< Owner of this CacheImpl. + EventThreadCallback* mTrigger; ///< EventThread callback trigger. (Not owned.) + + Dali::AsyncTaskManager::TasksCompletedId mTasksCompletedCount{0u}; + + using TasksCompletedContainer = std::unordered_map; + TasksCompletedContainer mTasksCompletedCallbackList; + + using ExecuteCallbackContainer = std::vector, Dali::AsyncTaskManager::TasksCompletedId>>; + ExecuteCallbackContainer mExcuteCallbackList; + + Dali::Mutex mTasksCompletedCallbacksMutex; ///< Mutex for mTasksCompletedCallbackList. We can lock mExcuteCallbacksMutex under this scope. + Dali::Mutex mExcuteCallbacksMutex; ///< Mutex for mExcuteCallbackList. + + bool mEmitCompletedTaskTriggered : 1; +}; + +// AsyncTaskManager + +Dali::AsyncTaskManager AsyncTaskManager::Get() +{ + if(!gAsyncTaskManager) + { + gAsyncTaskManager = Dali::AsyncTaskManager(new AsyncTaskManager()); + } + return gAsyncTaskManager; +} + +AsyncTaskManager::AsyncTaskManager() +: mTasks(GetNumberOfThreads(NUMBER_OF_ASYNC_THREADS_ENV, DEFAULT_NUMBER_OF_ASYNC_THREADS), [&]() { return TaskHelper(*this); }), + mTrigger(new EventThreadCallback(MakeCallback(this, &AsyncTaskManager::TaskCompleted))), + mTasksCompletedImpl(new TasksCompletedImpl(*this, mTrigger.get())) +{ +} + +AsyncTaskManager::~AsyncTaskManager() +{ + // Join all threads. + mTasks.Clear(); + + // Remove task completed impl after all threads are join. + mTasksCompletedImpl.reset(); + + // Remove tasks + mWaitingTasks.clear(); + mRunningTasks.clear(); + mCompletedTasks.clear(); +} + +void AsyncTaskManager::AddTask(AsyncTaskPtr task) +{ + if(task) + { + // Lock while adding task to the queue + Mutex::ScopedLock lock(mWaitingTasksMutex); + + // push back into waiting queue. + mWaitingTasks.insert(mWaitingTasks.end(), task); + + { + // For thread safety + Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex. + + // Finish all Running threads are working + if(mRunningTasks.size() >= mTasks.GetElementCount()) + { + return; + } + } + } + + size_t count = mTasks.GetElementCount(); + size_t index = 0; + while(index++ < count) + { + auto processHelperIt = mTasks.GetNext(); + DALI_ASSERT_ALWAYS(processHelperIt != mTasks.End()); + if(processHelperIt->Request()) + { + break; + } + // If all threads are busy, then it's ok just to push the task because they will try to get the next job. + } +} + +void AsyncTaskManager::RemoveTask(AsyncTaskPtr task) +{ + if(task) + { + uint32_t removedCount = 0u; + + { + // Lock while remove task from the queue + Mutex::ScopedLock lock(mWaitingTasksMutex); + + for(auto iterator = mWaitingTasks.begin(); iterator != mWaitingTasks.end();) + { + if((*iterator) == task) + { + iterator = mWaitingTasks.erase(iterator); + ++removedCount; + } + else + { + ++iterator; + } + } + } + + { + // Lock while remove task from the queue + Mutex::ScopedLock lock(mRunningTasksMutex); + + for(auto iterator = mRunningTasks.begin(); iterator != mRunningTasks.end();) + { + if((*iterator).first == task) + { + // We cannot erase container. Just mark as canceled. + // Note : mAvaliableLowPriorityTaskCounts will be increased after process finished. + (*iterator).second = RunningTaskState::CANCELED; + ++removedCount; + } + ++iterator; + } + } + + { + // Lock while remove task from the queue + Mutex::ScopedLock lock(mCompletedTasksMutex); + + for(auto iterator = mCompletedTasks.begin(); iterator != mCompletedTasks.end();) + { + if((*iterator).first == task) + { + iterator = mCompletedTasks.erase(iterator); + ++removedCount; + } + else + { + ++iterator; + } + } + } + + // Remove TasksCompleted callback trace + if(mTasksCompletedImpl->IsTasksCompletedCallbackExist() && removedCount > 0u) + { + mTasksCompletedImpl->RemoveTaskTrace(task, removedCount); + } + } +} + +Dali::AsyncTaskManager::TasksCompletedId AsyncTaskManager::SetCompletedCallback(CallbackBase* callback, Dali::AsyncTaskManager::CompletedCallbackTraceMask mask) +{ + // mTasksCompletedImpl will take ownership of callback. + Dali::AsyncTaskManager::TasksCompletedId tasksCompletedId = mTasksCompletedImpl->GenerateTasksCompletedId(callback); + + bool taskAdded = false; ///< Flag whether at least one task tracing now. + + // Please be careful the order of mutex, to avoid dead lock. + { + Mutex::ScopedLock lockWait(mWaitingTasksMutex); + { + Mutex::ScopedLock lockRunning(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex. + { + Mutex::ScopedLock lockComplete(mCompletedTasksMutex); // We can lock this mutex under mWaitingTasksMutex and mRunningTasksMutex. + + // Collect all tasks from waiting tasks + for(auto& task : mWaitingTasks) + { + auto checkMask = (task->GetCallbackInvocationThread() == Dali::AsyncTask::ThreadType::MAIN_THREAD ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_MAIN : Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_WORKER) | + (task->GetPriorityType() == Dali::AsyncTask::PriorityType::HIGH ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_HIGH : Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_LOW); + + if((checkMask & mask) == checkMask) + { + taskAdded = true; + mTasksCompletedImpl->AppendTaskTrace(tasksCompletedId, task); + } + } + + // Collect all tasks from running tasks + for(auto& taskPair : mRunningTasks) + { + auto& task = taskPair.first; + auto checkMask = (task->GetCallbackInvocationThread() == Dali::AsyncTask::ThreadType::MAIN_THREAD ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_MAIN : Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_WORKER) | + (task->GetPriorityType() == Dali::AsyncTask::PriorityType::HIGH ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_HIGH : Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_LOW); + + if((checkMask & mask) == checkMask) + { + taskAdded = true; + mTasksCompletedImpl->AppendTaskTrace(tasksCompletedId, task); + } + } + + // Collect all tasks from complete tasks + for(auto& taskPair : mCompletedTasks) + { + auto& task = taskPair.first; + auto checkMask = (task->GetCallbackInvocationThread() == Dali::AsyncTask::ThreadType::MAIN_THREAD ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_MAIN : Dali::AsyncTaskManager::CompletedCallbackTraceMask::THREAD_MASK_WORKER) | + (task->GetPriorityType() == Dali::AsyncTask::PriorityType::HIGH ? Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_HIGH : Dali::AsyncTaskManager::CompletedCallbackTraceMask::PRIORITY_MASK_LOW); + + if((checkMask & mask) == checkMask) + { + taskAdded = true; + mTasksCompletedImpl->AppendTaskTrace(tasksCompletedId, task); + } + } + } + } + } + + // If there is nothing to check task, just excute callback right now. + if(!taskAdded) + { + mTasksCompletedImpl->CheckTasksCompletedCallbackCompleted(tasksCompletedId); + } + return tasksCompletedId; +} + +bool AsyncTaskManager::RemoveCompletedCallback(Dali::AsyncTaskManager::TasksCompletedId tasksCompletedId) +{ + return mTasksCompletedImpl->RemoveTasksCompleted(tasksCompletedId); +} + +AsyncTaskPtr AsyncTaskManager::PopNextCompletedTask() +{ + std::vector ignoredTaskList; ///< To keep asyncTask reference so we can ensure that destructor called out of mutex. + + AsyncTaskPtr nextCompletedTask = nullptr; + { + // Lock while popping task out from the queue + Mutex::ScopedLock lock(mCompletedTasksMutex); + + while(!mCompletedTasks.empty()) + { + auto next = mCompletedTasks.begin(); + AsyncTaskPtr nextTask = next->first; + CompletedTaskState taskState = next->second; + mCompletedTasks.erase(next); + + if(taskState == CompletedTaskState::REQUIRE_CALLBACK) + { + nextCompletedTask = nextTask; + break; + } + + ignoredTaskList.push_back(nextTask); + } + } + + return nextCompletedTask; +} + +void AsyncTaskManager::TaskCompleted() +{ + // For UTC, let we complete only 1 task here. + if(AsyncTaskPtr task = PopNextCompletedTask()) + { + CallbackBase::Execute(*(task->GetCompletedCallback()), task); + + // Remove TasksCompleted callback trace + if(mTasksCompletedImpl->IsTasksCompletedCallbackExist()) + { + mTasksCompletedImpl->RemoveTaskTrace(task); + } + } + + mTasksCompletedImpl->EmitCompletedTasks(); +} + +void AsyncTaskManager::TaskAllCompleted() +{ + while(AsyncTaskPtr task = PopNextCompletedTask()) + { + CallbackBase::Execute(*(task->GetCompletedCallback()), task); + + // Remove TasksCompleted callback trace + if(mTasksCompletedImpl->IsTasksCompletedCallbackExist()) + { + mTasksCompletedImpl->RemoveTaskTrace(task); + } + } + + mTasksCompletedImpl->EmitCompletedTasks(); +} + +/// Worker thread called +AsyncTaskPtr AsyncTaskManager::PopNextTaskToProcess() +{ + // Lock while popping task out from the queue + Mutex::ScopedLock lock(mWaitingTasksMutex); + + // pop out the next task from the queue + AsyncTaskPtr nextTask = nullptr; + + for(auto iter = mWaitingTasks.begin(), endIter = mWaitingTasks.end(); iter != endIter; ++iter) + { + if((*iter)->IsReady()) + { + nextTask = *iter; + + // Add Running queue + { + // Lock while popping task out from the queue + Mutex::ScopedLock lock(mRunningTasksMutex); // We can lock this mutex under mWaitingTasksMutex. + + mRunningTasks.insert(mRunningTasks.end(), std::make_pair(nextTask, RunningTaskState::RUNNING)); + + mWaitingTasks.erase(iter); + } + break; + } + } + + return nextTask; +} + +/// Worker thread called +void AsyncTaskManager::CompleteTask(AsyncTaskPtr&& task) +{ + bool notify = false; + + if(task) + { + bool needTrigger = false; + + // Lock while check validation of task. + { + Mutex::ScopedLock lock(mRunningTasksMutex); + + auto iter = std::find_if(mRunningTasks.begin(), mRunningTasks.end(), [task](const AsyncRunningTaskPair& element) { return element.first == task; }); + if(iter != mRunningTasks.end()) + { + if(iter->second == RunningTaskState::RUNNING) + { + // This task is valid. + notify = true; + } + } + } + + // We should execute this tasks complete callback out of mutex + if(notify && task->GetCallbackInvocationThread() == AsyncTask::ThreadType::WORKER_THREAD) + { + CallbackBase::Execute(*(task->GetCompletedCallback()), task); + + // We need to remove task trace now. + if(mTasksCompletedImpl->IsTasksCompletedCallbackExist()) + { + mTasksCompletedImpl->RemoveTaskTrace(task); + + if(mTasksCompletedImpl->IsExecuteCallbackExist()) + { + // We need to call EmitCompletedTasks(). Trigger main thread. + needTrigger = true; + } + } + } + + // Lock while adding task to the queue + { + Mutex::ScopedLock lock(mRunningTasksMutex); + + auto iter = std::find_if(mRunningTasks.begin(), mRunningTasks.end(), [task](const AsyncRunningTaskPair& element) { return element.first == task; }); + if(iter != mRunningTasks.end()) + { + // 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); + + needTrigger |= callbackRequired; + + mCompletedTasks.insert(mCompletedTasks.end(), std::make_pair(task, callbackRequired ? CompletedTaskState::REQUIRE_CALLBACK : CompletedTaskState::SKIP_CALLBACK)); + + mRunningTasks.erase(iter); + + if(!needTrigger) + { + needTrigger |= (mCompletedTasks.size() >= FORCE_TRIGGER_THRESHOLD); + } + + // Now, task is invalidate. + task.Reset(); + } + } + } + + // Wake up the main thread + if(needTrigger) + { + mTrigger->Trigger(); + } + } +} + +// AsyncTaskManager::TaskHelper + +AsyncTaskManager::TaskHelper::TaskHelper(AsyncTaskManager& asyncTaskManager) +: TaskHelper(std::unique_ptr(new AsyncTaskThread(asyncTaskManager)), asyncTaskManager) +{ +} + +AsyncTaskManager::TaskHelper::TaskHelper(TaskHelper&& rhs) +: TaskHelper(std::move(rhs.mProcessor), rhs.mAsyncTaskManager) +{ +} + +AsyncTaskManager::TaskHelper::TaskHelper(std::unique_ptr processor, AsyncTaskManager& asyncTaskManager) +: mProcessor(std::move(processor)), + mAsyncTaskManager(asyncTaskManager) +{ +} + +bool AsyncTaskManager::TaskHelper::Request() +{ + return mProcessor->Request(); +} + +} // namespace Adaptor + +} // namespace Internal + +/********************************************************************************/ +/********************************* PUBLIC CLASS *******************************/ +/********************************************************************************/ + +AsyncTaskManager::AsyncTaskManager() = default; + +AsyncTaskManager::~AsyncTaskManager() = default; + +AsyncTaskManager AsyncTaskManager::Get() +{ + return Internal::Adaptor::AsyncTaskManager::Get(); +} + +void AsyncTaskManager::AddTask(AsyncTaskPtr task) +{ + Internal::Adaptor::GetImplementation(*this).AddTask(task); +} + +void AsyncTaskManager::RemoveTask(AsyncTaskPtr task) +{ + Internal::Adaptor::GetImplementation(*this).RemoveTask(task); +} + +AsyncTaskManager::TasksCompletedId AsyncTaskManager::SetCompletedCallback(CallbackBase* callback, AsyncTaskManager::CompletedCallbackTraceMask mask) +{ + return Internal::Adaptor::GetImplementation(*this).SetCompletedCallback(callback, mask); +} + +bool AsyncTaskManager::RemoveCompletedCallback(AsyncTaskManager::TasksCompletedId tasksCompletedId) +{ + return Internal::Adaptor::GetImplementation(*this).RemoveCompletedCallback(tasksCompletedId); +} + +AsyncTaskManager::AsyncTaskManager(Internal::Adaptor::AsyncTaskManager* impl) +: BaseHandle(impl) +{ +} + +} // namespace Dali + +namespace Test +{ +namespace AsyncTaskManager +{ +void DestroyAsyncTaskManager() +{ + Dali::Internal::Adaptor::gAsyncTaskManager.Reset(); +} + +void ProcessSingleCompletedTask() +{ + auto asyncTaskManager = Dali::AsyncTaskManager::Get(); + Dali::Internal::Adaptor::GetImplementation(asyncTaskManager).TaskCompleted(); +} + +void ProcessAllCompletedTask() +{ + auto asyncTaskManager = Dali::AsyncTaskManager::Get(); + Dali::Internal::Adaptor::GetImplementation(asyncTaskManager).TaskAllCompleted(); +} +} // namespace AsyncTaskManager +} // namespace Test \ No newline at end of file diff --git a/automated-tests/src/dali-toolkit/dali-toolkit-test-utils/toolkit-async-task-manager.h b/automated-tests/src/dali-toolkit/dali-toolkit-test-utils/toolkit-async-task-manager.h new file mode 100644 index 0000000..875daaf --- /dev/null +++ b/automated-tests/src/dali-toolkit/dali-toolkit-test-utils/toolkit-async-task-manager.h @@ -0,0 +1,39 @@ +#ifndef DALI_TOOLKIT_TOOLKIT_ASYNC_TASK_MANAGER_H +#define DALI_TOOLKIT_TOOLKIT_ASYNC_TASK_MANAGER_H + +/* + * Copyright (c) 2023 Samsung Electronics Co., Ltd. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +// EXTERNAL INCLUDES +#include + +namespace Test +{ +namespace AsyncTaskManager +{ +// Destroy global static AsyncTaskManager, only for ~ToolkitApplication() +void DestroyAsyncTaskManager(); + +// Execute one completed processes synchronously. +void ProcessSingleCompletedTasks(); + +// Execute all completed processes synchronously. +void ProcessAllCompletedTasks(); +} // namespace AsyncTaskManager +} // namespace Test + +#endif // DALI_TOOLKIT_TOOLKIT_ASYNC_TASK_MANAGER_H diff --git a/automated-tests/src/dali-toolkit/utc-Dali-AnimatedImageVisual.cpp b/automated-tests/src/dali-toolkit/utc-Dali-AnimatedImageVisual.cpp index 29abcd2..73f4cf7 100644 --- a/automated-tests/src/dali-toolkit/utc-Dali-AnimatedImageVisual.cpp +++ b/automated-tests/src/dali-toolkit/utc-Dali-AnimatedImageVisual.cpp @@ -1233,10 +1233,9 @@ int UtcDaliAnimatedImageVisualMultiImage01(void) tet_infoline("Test that after 2 ticks that we have 6 textures"); Test::EmitGlobalTimerSignal(); - // TODO : Open this logic if we make AsyncTaskManager for toolkit UTC doesn't execute by SendNotification(). - //application.SendNotification(); - //application.Render(16); - //DALI_TEST_EQUALS(gl.GetNumGeneratedTextures(), 6, TEST_LOCATION); + application.SendNotification(); + application.Render(16); + DALI_TEST_EQUALS(gl.GetNumGeneratedTextures(), 6, TEST_LOCATION); tet_infoline("And that at least 2 textures were requested"); DALI_TEST_EQUALS(Test::WaitForEventThreadTrigger(2), true, TEST_LOCATION); @@ -1246,10 +1245,9 @@ int UtcDaliAnimatedImageVisualMultiImage01(void) tet_infoline("Test that after 3rd tick that we have 7 textures and 1 request"); Test::EmitGlobalTimerSignal(); - // TODO : Open this logic if we make AsyncTaskManager for toolkit UTC doesn't execute by SendNotification(). - //application.SendNotification(); - //application.Render(16); - //DALI_TEST_EQUALS(gl.GetNumGeneratedTextures(), 7, TEST_LOCATION); + application.SendNotification(); + application.Render(16); + DALI_TEST_EQUALS(gl.GetNumGeneratedTextures(), 7, TEST_LOCATION); DALI_TEST_EQUALS(Test::WaitForEventThreadTrigger(1), true, TEST_LOCATION); application.SendNotification(); diff --git a/automated-tests/src/dali-toolkit/utc-Dali-ImageView.cpp b/automated-tests/src/dali-toolkit/utc-Dali-ImageView.cpp index 5ca6052..52d5495 100644 --- a/automated-tests/src/dali-toolkit/utc-Dali-ImageView.cpp +++ b/automated-tests/src/dali-toolkit/utc-Dali-ImageView.cpp @@ -4389,7 +4389,13 @@ int UtcDaliImageViewSetImageOnResourceReadySignal06(void) DALI_TEST_EQUALS(Test::WaitForEventThreadTrigger(2), true, TEST_LOCATION); DALI_TEST_EQUALS(gResourceReadySignalCounter, 2, TEST_LOCATION); - tet_infoline("load done"); + tet_infoline("Note that resource ready should not come now."); + tet_infoline("Try to load remained 2 valid image + apply masking"); + + DALI_TEST_EQUALS(Test::WaitForEventThreadTrigger(4), true, TEST_LOCATION); + DALI_TEST_EQUALS(gResourceReadySignalCounter, 4, TEST_LOCATION); + + tet_infoline("Check all resource ready comes now."); } END_TEST; } diff --git a/dali-toolkit/internal/visuals/animated-vector-image/vector-animation-manager.cpp b/dali-toolkit/internal/visuals/animated-vector-image/vector-animation-manager.cpp index 50e6471..59cd4ba 100644 --- a/dali-toolkit/internal/visuals/animated-vector-image/vector-animation-manager.cpp +++ b/dali-toolkit/internal/visuals/animated-vector-image/vector-animation-manager.cpp @@ -51,7 +51,7 @@ VectorAnimationManager::~VectorAnimationManager() { mEventCallbacks.clear(); - if(mProcessorRegistered) + if(mProcessorRegistered && Adaptor::IsAvailable()) { Adaptor::Get().UnregisterProcessor(*this, true); }