diff --git a/src/common/task_queue.cpp b/src/common/task_queue.cpp index 82294fd39..ab771f65d 100644 --- a/src/common/task_queue.cpp +++ b/src/common/task_queue.cpp @@ -4,15 +4,19 @@ #include "task_queue.h" #include "assert.h" +#include "common/log.h" + +LOG_CHANNEL(Threading); + TaskQueue::TaskQueue() = default; TaskQueue::~TaskQueue() { - SetWorkerCount(0); + SetWorkerCount(0, 0); Assert(m_tasks.empty()); } -void TaskQueue::SetWorkerCount(u32 count) +void TaskQueue::SetWorkerCount(u16 count, u16 max_threads) { std::unique_lock lock(m_mutex); @@ -38,6 +42,8 @@ void TaskQueue::SetWorkerCount(u32 count) for (u32 i = 0; i < count; i++) m_threads.emplace_back(&TaskQueue::WorkerThreadEntryPoint, this); } + + m_max_threads = max_threads; } size_t TaskQueue::GetOutstandingTasks() @@ -50,7 +56,7 @@ void TaskQueue::SubmitTask(TaskFunctionType func) { std::unique_lock lock(m_mutex); - if (m_threads.empty()) [[unlikely]] + if (m_max_threads == 0) [[unlikely]] { lock.unlock(); func(); @@ -59,6 +65,14 @@ void TaskQueue::SubmitTask(TaskFunctionType func) m_tasks.push_back(std::move(func)); m_tasks_outstanding++; + + // If we're under pressure and all threads are busy, spin up another one. + if (m_threads_busy == m_threads.size() && m_threads.size() < m_max_threads) + { + m_threads.emplace_back(&TaskQueue::WorkerThreadEntryPoint, this); + DEV_LOG("Spawning TaskQueue worker thread, now {} threads", m_threads.size()); + } + m_task_wait_cv.notify_one(); } @@ -96,9 +110,12 @@ void TaskQueue::ExecuteOneTask(std::unique_lock& lock) { TaskFunctionType func = std::move(m_tasks.front()); m_tasks.pop_front(); + m_threads_busy++; lock.unlock(); func(); lock.lock(); + DebugAssert(m_threads_busy > 0); + m_threads_busy--; m_tasks_outstanding--; if (m_tasks_outstanding == 0) m_tasks_done_cv.notify_all(); diff --git a/src/common/task_queue.h b/src/common/task_queue.h index fa5762fc3..ff112d43a 100644 --- a/src/common/task_queue.h +++ b/src/common/task_queue.h @@ -26,7 +26,8 @@ public: /// Sets the number of worker threads to be used by the task queue. /// Setting this to zero threads completes tasks on the calling thread. /// @param count The desired number of worker threads. - void SetWorkerCount(u32 count); + /// @param max_threads The maximum number of worker threads. + void SetWorkerCount(u16 count, u16 max_threads); /// Returns the number of tasks remaining. size_t GetOutstandingTasks(); @@ -59,8 +60,10 @@ private: std::mutex m_mutex; std::deque m_tasks; size_t m_tasks_outstanding = 0; + u32 m_threads_busy = 0; + u16 m_max_threads = 0; + bool m_threads_done = false; std::condition_variable m_task_wait_cv; std::condition_variable m_tasks_done_cv; std::vector m_threads; - bool m_threads_done = false; }; diff --git a/src/core/core.cpp b/src/core/core.cpp index ed4248e9f..6836a06d9 100644 --- a/src/core/core.cpp +++ b/src/core/core.cpp @@ -55,8 +55,8 @@ LOG_CHANNEL(Core); namespace Core { -/// Use two async worker threads, should be enough for most tasks. -static constexpr u32 NUM_ASYNC_WORKER_THREADS = 2; +/// Worker threads are dynamically allocated, maximum 8. +static constexpr u32 MAX_ASYNC_WORKER_THREADS = 8; static bool SetAppRootAndResources(Error* error); static bool SetDataRoot(Error* error); @@ -738,7 +738,7 @@ bool Core::CoreThreadInitialize(bool disable_worker_threads, Error* error) s_locals.core_thread_handle = Threading::ThreadHandle::GetForCallingThread(); if (!disable_worker_threads) - s_locals.async_task_queue.SetWorkerCount(NUM_ASYNC_WORKER_THREADS); + s_locals.async_task_queue.SetWorkerCount(0, MAX_ASYNC_WORKER_THREADS); System::LoadSettings(false); @@ -758,7 +758,7 @@ bool Core::CoreThreadInitialize(bool disable_worker_threads, Error* error) void Core::CoreThreadShutdown() { - s_locals.async_task_queue.SetWorkerCount(0); + s_locals.async_task_queue.SetWorkerCount(0, 0); #ifdef ENABLE_DISCORD_PRESENCE DiscordPresence::Shutdown();