From 4d703e38078e8bcf9c35fa555e79c10ae4ea253e Mon Sep 17 00:00:00 2001 From: Ben Doherty Date: Fri, 11 Aug 2023 17:03:36 -0700 Subject: [PATCH] Refactor CompilerThreadPool out of OpenGLDriver (#7067) --- filament/backend/CMakeLists.txt | 2 + filament/backend/src/CompilerThreadPool.cpp | 131 ++++++++++++++++++ filament/backend/src/CompilerThreadPool.h | 68 +++++++++ .../src/opengl/ShaderCompilerService.cpp | 128 ++--------------- .../src/opengl/ShaderCompilerService.h | 28 +--- 5 files changed, 218 insertions(+), 139 deletions(-) create mode 100644 filament/backend/src/CompilerThreadPool.cpp create mode 100644 filament/backend/src/CompilerThreadPool.h diff --git a/filament/backend/CMakeLists.txt b/filament/backend/CMakeLists.txt index 3d95501294..550f569f02 100644 --- a/filament/backend/CMakeLists.txt +++ b/filament/backend/CMakeLists.txt @@ -31,6 +31,7 @@ set(SRCS src/CircularBuffer.cpp src/CommandBufferQueue.cpp src/CommandStream.cpp + src/CompilerThreadPool.cpp src/Driver.cpp src/Handle.cpp src/HandleAllocator.cpp @@ -55,6 +56,7 @@ set(PRIVATE_HDRS include/private/backend/PlatformFactory.h include/private/backend/SamplerGroup.h src/CommandStreamDispatcher.h + src/CompilerThreadPool.h src/DataReshaper.h src/DriverBase.h ) diff --git a/filament/backend/src/CompilerThreadPool.cpp b/filament/backend/src/CompilerThreadPool.cpp new file mode 100644 index 0000000000..008b353aaa --- /dev/null +++ b/filament/backend/src/CompilerThreadPool.cpp @@ -0,0 +1,131 @@ +/* + * Copyright (C) 2023 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 "CompilerThreadPool.h" + +#include + +namespace filament::backend { + +using namespace utils; + +ProgramToken::~ProgramToken() = default; + +CompilerThreadPool::CompilerThreadPool() noexcept = default; + +CompilerThreadPool::~CompilerThreadPool() noexcept { + assert_invariant(mCompilerThreads.empty()); + assert_invariant(mQueues[0].empty()); + assert_invariant(mQueues[1].empty()); +} + +void CompilerThreadPool::init(uint32_t threadCount, JobSystem::Priority priority, + ThreadSetup&& threadSetup) noexcept { + auto setup = std::make_shared(std::move(threadSetup)); + + for (size_t i = 0; i < threadCount; i++) { + mCompilerThreads.emplace_back([this, priority, setup]() { + // give the thread a name + JobSystem::setThreadName("CompilerThreadPool"); + // run at a slightly lower priority than other filament threads + JobSystem::setThreadPriority(priority); + + (*setup)(); + + // process jobs from the queue until we're asked to exit + while (!mExitRequested) { + std::unique_lock lock(mQueueLock); + mQueueCondition.wait(lock, [this]() { + return mExitRequested || + (!std::all_of( std::begin(mQueues), std::end(mQueues), + [](auto&& q) { return q.empty(); })); + }); + if (!mExitRequested) { + Job job; + // use the first queue that's not empty + auto& queue = [this]() -> auto& { + for (auto& q: mQueues) { + if (!q.empty()) { + return q; + } + } + return mQueues[0]; // we should never end-up here. + }(); + assert_invariant(!queue.empty()); + std::swap(job, queue.front().second); + queue.pop_front(); + + // execute the job without holding any locks + lock.unlock(); + job(); + } + } + }); + + } +} + +auto CompilerThreadPool::find(program_token_t const& token) -> std::pair { + for (auto&& q: mQueues) { + auto pos = std::find_if(q.begin(), q.end(), [&token](auto&& item) { + return item.first == token; + }); + if (pos != q.end()) { + return { q, pos }; + } + } + // this can happen if the program is being processed right now + return { mQueues[0], mQueues[0].end() }; +} + +auto CompilerThreadPool::dequeue(program_token_t const& token) -> Job { + std::unique_lock const lock(mQueueLock); + Job job; + auto&& [q, pos] = find(token); + if (pos != q.end()) { + std::swap(job, pos->second); + q.erase(pos); + } + return job; +} + +void CompilerThreadPool::queue(CompilerPriorityQueue priorityQueue, + program_token_t const& token, Job&& job) { + std::unique_lock const lock(mQueueLock); + mQueues[size_t(priorityQueue)].emplace_back(token, std::move(job)); + mQueueCondition.notify_one(); +} + +void CompilerThreadPool::terminate() noexcept { + std::unique_lock lock(mQueueLock); + mExitRequested = true; + mQueueCondition.notify_all(); + lock.unlock(); + + for (auto& thread: mCompilerThreads) { + if (thread.joinable()) { + thread.join(); + } + } + mCompilerThreads.clear(); + + // Clear all the queues, dropping the remaining jobs. This relies on the jobs being cancelable. + for (auto&& q : mQueues) { + q.clear(); + } +} + +} // namespace filament::backend diff --git a/filament/backend/src/CompilerThreadPool.h b/filament/backend/src/CompilerThreadPool.h new file mode 100644 index 0000000000..78ce4c7079 --- /dev/null +++ b/filament/backend/src/CompilerThreadPool.h @@ -0,0 +1,68 @@ +/* + * Copyright (C) 2023 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_FILAMENT_BACKEND_COMPILERTHREADPOOL_H +#define TNT_FILAMENT_BACKEND_COMPILERTHREADPOOL_H + +#include + +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace filament::backend { + +struct ProgramToken { + virtual ~ProgramToken(); +}; + +using program_token_t = std::shared_ptr; + +class Platform; + +class CompilerThreadPool { +public: + CompilerThreadPool() noexcept; + ~CompilerThreadPool() noexcept; + using Job = utils::Invocable; + using ThreadSetup = utils::Invocable; + void init(uint32_t threadCount, utils::JobSystem::Priority priority, + ThreadSetup&& threadSetup) noexcept; + void terminate() noexcept; + void queue(CompilerPriorityQueue priorityQueue, program_token_t const& token, Job&& job); + Job dequeue(program_token_t const& token); + +private: + using Queue = std::deque>; + std::vector mCompilerThreads; + std::atomic_bool mExitRequested{false}; + std::mutex mQueueLock; + std::condition_variable mQueueCondition; + std::array mQueues; + // lock must be held for methods below + std::pair find(program_token_t const& token); +}; + +} // namespace filament::backend + +#endif // TNT_FILAMENT_BACKEND_COMPILERTHREADPOOL_H + diff --git a/filament/backend/src/opengl/ShaderCompilerService.cpp b/filament/backend/src/opengl/ShaderCompilerService.cpp index 541b4ad51d..abdc4ecbdc 100644 --- a/filament/backend/src/opengl/ShaderCompilerService.cpp +++ b/filament/backend/src/opengl/ShaderCompilerService.cpp @@ -63,7 +63,7 @@ static inline std::string to_string(float f) noexcept { // ------------------------------------------------------------------------------------------------ -struct ShaderCompilerService::ProgramToken { +struct ShaderCompilerService::OpenGLProgramToken : ProgramToken { struct ProgramBinary { GLenum format{}; GLuint program{}; @@ -71,7 +71,8 @@ struct ShaderCompilerService::ProgramToken { std::vector blob; }; - ProgramToken(ShaderCompilerService& compiler, utils::CString const& name) noexcept + ~OpenGLProgramToken(); + OpenGLProgramToken(ShaderCompilerService& compiler, utils::CString const& name) noexcept : compiler(compiler), name(name) { } ShaderCompilerService& compiler; @@ -120,6 +121,8 @@ struct ShaderCompilerService::ProgramToken { bool canceled = false; // not part of the signaling }; +ShaderCompilerService::OpenGLProgramToken::~OpenGLProgramToken() = default; + void ShaderCompilerService::setUserData(const program_token_t& token, void* user) noexcept { token->user = user; } @@ -130,114 +133,6 @@ void* ShaderCompilerService::getUserData(const program_token_t& token) noexcept // ------------------------------------------------------------------------------------------------ -ShaderCompilerService::CompilerThreadPool::CompilerThreadPool() noexcept = default; - -ShaderCompilerService::CompilerThreadPool::~CompilerThreadPool() noexcept { - assert_invariant(mCompilerThreads.empty()); - assert_invariant(mQueues[0].empty()); - assert_invariant(mQueues[1].empty()); -} - -void ShaderCompilerService::CompilerThreadPool::init( - bool useSharedContexts, uint32_t threadCount, JobSystem::Priority priority, - OpenGLPlatform& platform) noexcept { - - for (size_t i = 0; i < threadCount; i++) { - mCompilerThreads.emplace_back([this, useSharedContexts, priority, &platform]() { - // give the thread a name - JobSystem::setThreadName("CompilerThreadPool"); - // run at a slightly lower priority than other filament threads - JobSystem::setThreadPriority(priority); - - // create a gl context current to this thread - platform.createContext(useSharedContexts); - - // process jobs from the queue until we're asked to exit - while (!mExitRequested) { - std::unique_lock lock(mQueueLock); - mQueueCondition.wait(lock, [this]() { - return mExitRequested || - (!std::all_of( std::begin(mQueues), std::end(mQueues), - [](auto&& q) { return q.empty(); })); - }); - if (!mExitRequested) { - Job job; - // use the first queue that's not empty - auto& queue = [this]() -> auto& { - for (auto& q: mQueues) { - if (!q.empty()) { - return q; - } - } - return mQueues[0]; // we should never end-up here. - }(); - assert_invariant(!queue.empty()); - std::swap(job, queue.front().second); - queue.pop_front(); - - // execute the job without holding any locks - lock.unlock(); - job(); - } - } - }); - - } -} - -auto ShaderCompilerService::CompilerThreadPool::find( - program_token_t const& token) -> std::pair { - for (auto&& q: mQueues) { - auto pos = std::find_if(q.begin(), q.end(), [&token](auto&& item) { - return item.first == token; - }); - if (pos != q.end()) { - return { q, pos }; - } - } - // this can happen if the program is being processed right now - return { mQueues[0], mQueues[0].end() }; -} - -auto ShaderCompilerService::CompilerThreadPool::dequeue(program_token_t const& token) -> Job { - std::unique_lock const lock(mQueueLock); - Job job; - auto&& [q, pos] = find(token); - if (pos != q.end()) { - std::swap(job, pos->second); - q.erase(pos); - } - return job; -} - -void ShaderCompilerService::CompilerThreadPool::queue(CompilerPriorityQueue priorityQueue, - program_token_t const& token, Job&& job) { - std::unique_lock const lock(mQueueLock); - mQueues[size_t(priorityQueue)].emplace_back(token, std::move(job)); - mQueueCondition.notify_one(); -} - -void ShaderCompilerService::CompilerThreadPool::terminate() noexcept { - std::unique_lock lock(mQueueLock); - mExitRequested = true; - mQueueCondition.notify_all(); - lock.unlock(); - - for (auto& thread: mCompilerThreads) { - if (thread.joinable()) { - thread.join(); - } - } - mCompilerThreads.clear(); - - // Clear all the queues, dropping the remaining jobs. This relies on the jobs being cancelable. - for (auto&& q : mQueues) { - q.clear(); - } -} - -// ------------------------------------------------------------------------------------------------ - ShaderCompilerService::ShaderCompilerService(OpenGLDriver& driver) : mDriver(driver), KHR_parallel_shader_compile(driver.getContext().ext.KHR_parallel_shader_compile) { @@ -274,8 +169,11 @@ void ShaderCompilerService::init() noexcept { } mShaderCompilerThreadCount = poolSize; - mCompilerThreadPool.init(mUseSharedContext, - mShaderCompilerThreadCount, priority, mDriver.mPlatform); + mCompilerThreadPool.init(mShaderCompilerThreadCount, priority, + [platform = &mDriver.mPlatform, sharedContext = mUseSharedContext]() { + // create a gl context current to this thread + platform->createContext(sharedContext); + }); } } } @@ -303,7 +201,7 @@ ShaderCompilerService::program_token_t ShaderCompilerService::createProgram( utils::CString const& name, Program&& program) { auto& gl = mDriver.getContext(); - auto token = std::make_shared(*this, name); + auto token = std::make_shared(*this, name); if (UTILS_UNLIKELY(gl.isES2())) { token->attributes = std::move(program.getAttributes()); @@ -329,7 +227,7 @@ ShaderCompilerService::program_token_t ShaderCompilerService::createProgram( // link the program GLuint const glProgram = linkProgram(gl, shaders, token->attributes); - ProgramToken::ProgramBinary binary; + OpenGLProgramToken::ProgramBinary binary; binary.shaders = shaders; if (UTILS_LIKELY(mUseSharedContext)) { @@ -551,7 +449,7 @@ void ShaderCompilerService::notifyWhenAllProgramsAreReady(CompilerPriorityQueue // ------------------------------------------------------------------------------------------------ void ShaderCompilerService::getProgramFromCompilerPool(program_token_t& token) noexcept { - ProgramToken::ProgramBinary const& binary{ token->get() }; + OpenGLProgramToken::ProgramBinary const& binary{ token->get() }; if (!token->canceled) { token->gl.shaders = binary.shaders; if (UTILS_LIKELY(mUseSharedContext)) { diff --git a/filament/backend/src/opengl/ShaderCompilerService.h b/filament/backend/src/opengl/ShaderCompilerService.h index ffb74dbed0..668d4e31f2 100644 --- a/filament/backend/src/opengl/ShaderCompilerService.h +++ b/filament/backend/src/opengl/ShaderCompilerService.h @@ -19,6 +19,8 @@ #include "gl_headers.h" +#include "CompilerThreadPool.h" + #include #include @@ -49,10 +51,10 @@ class CallbackHandler; * A class handling shader compilation that supports asynchronous compilation. */ class ShaderCompilerService { - struct ProgramToken; + struct OpenGLProgramToken; public: - using program_token_t = std::shared_ptr; + using program_token_t = std::shared_ptr; explicit ShaderCompilerService(OpenGLDriver& driver); @@ -93,28 +95,6 @@ public: CallbackHandler* handler, CallbackHandler::Callback callback, void* user); private: - class CompilerThreadPool { - public: - CompilerThreadPool() noexcept; - ~CompilerThreadPool() noexcept; - using Job = utils::Invocable; - void init(bool useSharedContexts, uint32_t threadCount, utils::JobSystem::Priority priority, - OpenGLPlatform& platform) noexcept; - void terminate() noexcept; - void queue(CompilerPriorityQueue priorityQueue, program_token_t const& token, Job&& job); - Job dequeue(program_token_t const& token); - - private: - using Queue = std::deque>; - std::vector mCompilerThreads; - std::atomic_bool mExitRequested{ false }; - std::mutex mQueueLock; - std::condition_variable mQueueCondition; - std::array mQueues; - // lock must be held for methods below - std::pair find(program_token_t const& token); - }; - OpenGLDriver& mDriver; CompilerThreadPool mCompilerThreadPool;