From e974989a95f186fba4d18b7310150e1daed52217 Mon Sep 17 00:00:00 2001 From: Mathias Agopian Date: Wed, 1 Oct 2025 14:54:34 -0700 Subject: [PATCH] new utility AsyncJobQueue (#9278) * new utility AsyncJobQueue this is a very simple job queue, it spawns a thread and runs the jobs pushed to the queue in sequence. * use AsyncJobQueue in OpenGLTimerQuery --- .../backend/src/opengl/OpenGLTimerQuery.cpp | 44 ++---------- .../backend/src/opengl/OpenGLTimerQuery.h | 14 +--- libs/utils/CMakeLists.txt | 2 + libs/utils/include/utils/AsyncJobQueue.h | 62 ++++++++++++++++ libs/utils/src/AsyncJobQueue.cpp | 72 +++++++++++++++++++ 5 files changed, 143 insertions(+), 51 deletions(-) create mode 100644 libs/utils/include/utils/AsyncJobQueue.h create mode 100644 libs/utils/src/AsyncJobQueue.cpp diff --git a/filament/backend/src/opengl/OpenGLTimerQuery.cpp b/filament/backend/src/opengl/OpenGLTimerQuery.cpp index 04d7711fe1..16ee07c3b2 100644 --- a/filament/backend/src/opengl/OpenGLTimerQuery.cpp +++ b/filament/backend/src/opengl/OpenGLTimerQuery.cpp @@ -172,46 +172,12 @@ void TimerQueryNativeFactory::endTimeElapsedQuery(OpenGLDriver& driver, GLTimerQ // ------------------------------------------------------------------------------------------------ TimerQueryFenceFactory::TimerQueryFenceFactory(OpenGLPlatform& platform) - : mPlatform(platform) { - mQueue.reserve(2); - mThread = std::thread([this]() { - utils::JobSystem::setThreadName("OpenGLTimerQueryFence"); - utils::JobSystem::setThreadPriority(utils::JobSystem::Priority::URGENT_DISPLAY); - auto& queue = mQueue; - bool exitRequested; - do { - std::unique_lock lock(mLock); - mCondition.wait(lock, [this, &queue]() -> bool { - return mExitRequested || !queue.empty(); - }); - exitRequested = mExitRequested; - if (!queue.empty()) { - Job const job(queue.front()); - queue.erase(queue.begin()); - lock.unlock(); - job(); - } - } while (!exitRequested); - }); + : mPlatform(platform), + mJobQueue("OpenGLTimerQueryFence", utils::AsyncJobQueue::Priority::URGENT_DISPLAY) { } TimerQueryFenceFactory::~TimerQueryFenceFactory() { - assert_invariant(mQueue.empty()); - if (mThread.joinable()) { - std::unique_lock lock(mLock); - mExitRequested = true; - mCondition.notify_one(); - lock.unlock(); - if (mThread.joinable()) { - mThread.join(); - } - } -} - -void TimerQueryFenceFactory::push(TimerQueryFenceFactory::Job&& job) { - std::unique_lock const lock(mLock); - mQueue.push_back(std::move(job)); - mCondition.notify_one(); + mJobQueue.drainAndExit(); } void TimerQueryFenceFactory::createTimerQuery(GLTimerQuery* tq) { @@ -236,7 +202,7 @@ void TimerQueryFenceFactory::beginTimeElapsedQuery(GLTimerQuery* tq) { // on a dummy target for instance, or somehow latch the begin time at the next renderpass // start. - push([&platform = mPlatform, fence = mPlatform.createFence(), weak]() { + mJobQueue.push([&platform = mPlatform, fence = mPlatform.createFence(), weak]() { auto state = weak.lock(); if (state) { platform.waitFence(fence, FENCE_WAIT_FOR_EVER); @@ -252,7 +218,7 @@ void TimerQueryFenceFactory::endTimeElapsedQuery(OpenGLDriver&, GLTimerQuery* tq assert_invariant(tq->state); std::weak_ptr const weak = tq->state; - push([&platform = mPlatform, fence = mPlatform.createFence(), weak]() { + mJobQueue.push([&platform = mPlatform, fence = mPlatform.createFence(), weak]() { auto state = weak.lock(); if (state) { platform.waitFence(fence, FENCE_WAIT_FOR_EVER); diff --git a/filament/backend/src/opengl/OpenGLTimerQuery.h b/filament/backend/src/opengl/OpenGLTimerQuery.h index 5941c87d2b..85c005ea84 100644 --- a/filament/backend/src/opengl/OpenGLTimerQuery.h +++ b/filament/backend/src/opengl/OpenGLTimerQuery.h @@ -21,8 +21,7 @@ #include "DriverBase.h" -#include -#include +#include #include "gl_headers.h" @@ -108,22 +107,13 @@ public: explicit TimerQueryFenceFactory(OpenGLPlatform& platform); ~TimerQueryFenceFactory() override; private: - using Job = std::function; - using Container = std::vector; - void createTimerQuery(GLTimerQuery* query) override; void destroyTimerQuery(GLTimerQuery* query) override; void beginTimeElapsedQuery(GLTimerQuery* tq) override; void endTimeElapsedQuery(OpenGLDriver& driver, GLTimerQuery* tq) override; - void push(Job&& job); - OpenGLPlatform& mPlatform; - std::thread mThread; - mutable utils::Mutex mLock; - mutable utils::Condition mCondition; - Container mQueue; - bool mExitRequested = false; + utils::AsyncJobQueue mJobQueue; }; class TimerQueryFallbackFactory final : public TimerQueryFactoryInterface { diff --git a/libs/utils/CMakeLists.txt b/libs/utils/CMakeLists.txt index f0aecf566b..8a9a6edd9f 100644 --- a/libs/utils/CMakeLists.txt +++ b/libs/utils/CMakeLists.txt @@ -17,6 +17,7 @@ set(DIST_HDRS ${PUBLIC_HDR_DIR}/${TARGET}/CallStack.h ${PUBLIC_HDR_DIR}/${TARGET}/debug.h ${PUBLIC_HDR_DIR}/${TARGET}/Allocator.h + ${PUBLIC_HDR_DIR}/${TARGET}/AsyncJobQueue.h ${PUBLIC_HDR_DIR}/${TARGET}/BitmaskEnum.h ${PUBLIC_HDR_DIR}/${TARGET}/compiler.h ${PUBLIC_HDR_DIR}/${TARGET}/compressed_pair.h @@ -62,6 +63,7 @@ set(SRCS src/ashmem.cpp src/debug.cpp src/Allocator.cpp + src/AsyncJobQueue.cpp src/CallStack.cpp src/CString.cpp src/CountDownLatch.cpp diff --git a/libs/utils/include/utils/AsyncJobQueue.h b/libs/utils/include/utils/AsyncJobQueue.h new file mode 100644 index 0000000000..a67dd0217c --- /dev/null +++ b/libs/utils/include/utils/AsyncJobQueue.h @@ -0,0 +1,62 @@ +/* + * Copyright (C) 2025 The Android Open Source Project + * + * 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. + */ + +#ifndef TNT_ASYNCJOBQUEUE_H +#define TNT_ASYNCJOBQUEUE_H + +#include +#include +#include // NOLINT(*-include-cleaner) +#include // NOLINT(*-include-cleaner) + +#include +#include + +namespace utils { + +/** + * Simple asynchronous job queue. This manages a thread that executes jobs submitted to it + * in order. + */ +class AsyncJobQueue { +public: + using Job = Invocable; + using Priority = JobSystem::Priority; + + // create the job queue with a name and desired priority + AsyncJobQueue(const char* name, Priority priority); + + // drainAndExit() must be called first + ~AsyncJobQueue() noexcept; + + // blocks until all jobs are executed and quits the thread + void drainAndExit(); + + // adds a job to the queue. no-op if drainAndExit() was called. + void push(Job&& job); + +private: + using Container = std::vector; + std::thread mThread; + Mutex mLock; // NOLINT(*-include-cleaner) + Condition mCondition; // NOLINT(*-include-cleaner) + Container mQueue; + bool mExitRequested = false; +}; + +} // namespace utils + +#endif //TNT_ASYNCJOBQUEUE_H diff --git a/libs/utils/src/AsyncJobQueue.cpp b/libs/utils/src/AsyncJobQueue.cpp new file mode 100644 index 0000000000..c0ec92ad2c --- /dev/null +++ b/libs/utils/src/AsyncJobQueue.cpp @@ -0,0 +1,72 @@ +/* + * Copyright (C) 2025 The Android Open Source Project + * + * 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. + */ + +#include +#include +#include + +#include +#include + +namespace utils { + +AsyncJobQueue::AsyncJobQueue(const char* name, Priority priority) { + mQueue.reserve(2); + mThread = std::thread([this, name, priority]() { + JobSystem::setThreadName(name); + JobSystem::setThreadPriority(priority); + auto& queue = mQueue; + bool exitRequested; + do { + std::unique_lock lock(mLock); + mCondition.wait(lock, [this]() -> bool { return mExitRequested || !mQueue.empty(); }); + exitRequested = mExitRequested; + if (!queue.empty()) { + Job const job(std::move(queue.front())); + queue.erase(queue.begin()); + lock.unlock(); + + job(); + } + } while (!exitRequested); + }); +} + +AsyncJobQueue::~AsyncJobQueue() noexcept { + assert_invariant(mQueue.empty()); +} + +void AsyncJobQueue::push(Job&& job) { + std::unique_lock lock(mLock); + if (!mExitRequested) { + mQueue.push_back(std::move(job)); + lock.unlock(); + mCondition.notify_one(); + } +} + +void AsyncJobQueue::drainAndExit() { + std::unique_lock lock(mLock); + mCondition.wait(lock, [this] { return mQueue.empty(); }); + mExitRequested = true; + lock.unlock(); + mCondition.notify_one(); + if (mThread.joinable()) { + mThread.join(); + } +} + +} // namespace utils