From a60fe416813edc75946905d26fc7af34392b7791 Mon Sep 17 00:00:00 2001 From: Mathias Agopian Date: Tue, 9 Jul 2024 18:12:36 -0700 Subject: [PATCH 1/5] performance improvements to JobSystem - reduce the number of calls to notify_one() and notify_all(). notify_one() is not only called when running a new job, and notify_all() only when a job finishes. - don't hold the condition lock while calling notify_*(), as it is not strictly needed, and because notify_*() can be very slow, there can be a lot of contention on this lock as a result; blocking the whole jobsystem thread pool. - add a new version of run() that takes an opaque thread id that can be retrieved from a job's execute function; this is especially intended to be used by parallel_for(); it's just a more efficient version of run() that avoids a hashmap lookup. Overall these change yield a significant performance boost: - running + waiting a job: +200% - running many jobs: +150% - running many jobs in parallel: +50% --- filament/src/Froxelizer.cpp | 1 + filament/src/RenderPass.h | 3 +- filament/src/components/LightManager.cpp | 9 +-- libs/ibl/src/CubemapSH.cpp | 1 + libs/utils/include/utils/JobSystem.h | 45 +++++++++++--- libs/utils/src/JobSystem.cpp | 79 +++++++++++++----------- tools/matc/src/matc/MaterialCompiler.cpp | 1 + 7 files changed, 88 insertions(+), 51 deletions(-) diff --git a/filament/src/Froxelizer.cpp b/filament/src/Froxelizer.cpp index 47bd0d343d..560891ad5a 100644 --- a/filament/src/Froxelizer.cpp +++ b/filament/src/Froxelizer.cpp @@ -26,6 +26,7 @@ #include #include +#include #include #include diff --git a/filament/src/RenderPass.h b/filament/src/RenderPass.h index e1c7b3c181..23c4cf8249 100644 --- a/filament/src/RenderPass.h +++ b/filament/src/RenderPass.h @@ -417,8 +417,7 @@ private: void instanceify(FEngine& engine, Arena& arena, int32_t eyeCount) noexcept; // We choose the command count per job to minimize JobSystem overhead. - // On a Pixel 4, 2048 commands is about half a millisecond of processing. - static constexpr size_t JOBS_PARALLEL_FOR_COMMANDS_COUNT = 2048; + static constexpr size_t JOBS_PARALLEL_FOR_COMMANDS_COUNT = 1024; static constexpr size_t JOBS_PARALLEL_FOR_COMMANDS_SIZE = sizeof(Command) * JOBS_PARALLEL_FOR_COMMANDS_COUNT; diff --git a/filament/src/components/LightManager.cpp b/filament/src/components/LightManager.cpp index 6393caa10a..ad642f5942 100644 --- a/filament/src/components/LightManager.cpp +++ b/filament/src/components/LightManager.cpp @@ -20,12 +20,13 @@ #include "details/Engine.h" -#include -#include - -#include #include +#include +#include + +#include +#include using namespace filament::math; using namespace utils; diff --git a/libs/ibl/src/CubemapSH.cpp b/libs/ibl/src/CubemapSH.cpp index fc8b00c782..5d22cb804b 100644 --- a/libs/ibl/src/CubemapSH.cpp +++ b/libs/ibl/src/CubemapSH.cpp @@ -22,6 +22,7 @@ #include "CubemapUtilsImpl.h" +#include #include #include diff --git a/libs/utils/include/utils/JobSystem.h b/libs/utils/include/utils/JobSystem.h index a96fd48857..e3786dc322 100644 --- a/libs/utils/include/utils/JobSystem.h +++ b/libs/utils/include/utils/JobSystem.h @@ -21,7 +21,7 @@ #include #include #include -#include +#include #include #include #include @@ -47,12 +47,18 @@ class JobSystem { static constexpr size_t MAX_JOB_COUNT = 16384; static_assert(MAX_JOB_COUNT <= 0x7FFE, "MAX_JOB_COUNT must be <= 0x7FFE"); using WorkQueue = WorkStealingDequeue; + using Mutex = utils::Mutex; + using Condition = utils::Condition; public: class Job; + using ThreadId = uint8_t; + using JobFunc = void(*)(void*, JobSystem&, Job*); + static constexpr ThreadId invalidThreadId = 0xff; + class alignas(CACHELINE_SIZE) Job { public: Job() noexcept {} /* = default; */ /* clang bug */ // NOLINT(modernize-use-equals-default,cppcoreguidelines-pro-type-member-init) @@ -77,7 +83,9 @@ public: uint16_t parent; // 2 | 2 std::atomic runningJobCount = { 1 }; // 2 | 2 mutable std::atomic refCount = { 1 }; // 2 | 2 - // 6 | 2 (padding) + mutable ThreadId id = invalidThreadId; // 1 | 1 + // 1 | 1 (padding) + // 4 | 0 (padding) // 64 | 64 }; @@ -244,7 +252,7 @@ public: } /* - * Add job to this thread's execution queue. It's reference will drop automatically. + * Add job to this thread's execution queue. Its reference will drop automatically. * The current thread must be owned by JobSystem's thread pool. See adopt(). * * The job can't be used after this call. @@ -255,7 +263,19 @@ public: run(p); } - void signal() noexcept; + /* + * Add job to this thread's execution queue. Its reference will drop automatically. + * The current thread must be owned by JobSystem's thread pool. See adopt(). + * id must be the current thread id obtained with JobSystem::getThreadId(Job*). This + * API is more efficient than the methods above. + * + * The job can't be used after this call. + */ + void run(Job*& job, ThreadId id) noexcept; + void run(Job*&& job, ThreadId id) noexcept { // allows run(createJob(...)); + Job* p = job; + run(p, id); + } /* * Add job to this thread's execution queue and keep a reference to it. @@ -312,6 +332,13 @@ public: size_t getThreadCount() const { return mThreadCount; } + // returns the current ThreadId, which can be used with run(). This method can only be + // called from a job's function. + static ThreadId getThreadId(Job const* job) noexcept { + assert_invariant(job->id != invalidThreadId); + return job->id; + } + private: // this is just to avoid using std::default_random_engine, since we're in a public header. class default_random_engine { @@ -363,12 +390,12 @@ private: Job* steal(WorkQueue& workQueue) noexcept; void wait(std::unique_lock& lock, Job* job = nullptr) noexcept; - void wake(size_t hint) noexcept; void wakeAll() noexcept; + void wakeOne() noexcept; // these have thread contention, keep them together - utils::Mutex mWaiterLock; - utils::Condition mWaiterCondition; + Mutex mWaiterLock; + Condition mWaiterCondition; std::atomic mActiveJobs = { 0 }; utils::Arena, LockingPolicy::NoLock> mJobPool; @@ -468,7 +495,7 @@ right_side: // start the left side before attempting the right side, so we parallelize in case // of job creation failure -- rare, but still. - js.run(l); + js.run(l, JobSystem::getThreadId(parent)); // don't spawn a job for the right side, just reuse us -- spawning jobs is more // costly than we'd like. @@ -524,7 +551,7 @@ JobSystem::Job* parallel_for(JobSystem& js, JobSystem::Job* parent, } -template +template class CountSplitter { public: bool split(size_t splits, size_t count) const noexcept { diff --git a/libs/utils/src/JobSystem.cpp b/libs/utils/src/JobSystem.cpp index de972ccb46..b0e199fcef 100644 --- a/libs/utils/src/JobSystem.cpp +++ b/libs/utils/src/JobSystem.cpp @@ -29,14 +29,15 @@ static constexpr bool DEBUG_FINISH_HANGS = false; #include #include +#include #include -#include #include #include #include #include #include +#include #include #include #include @@ -252,6 +253,7 @@ inline bool JobSystem::hasJobCompleted(JobSystem::Job const* job) noexcept { } void JobSystem::wait(std::unique_lock& lock, Job* job) noexcept { + HEAVY_SYSTRACE_CALL(); if constexpr (!DEBUG_FINISH_HANGS) { mWaiterCondition.wait(lock); } else { @@ -289,27 +291,29 @@ void JobSystem::wait(std::unique_lock& lock, Job* job) noexcept { } void JobSystem::wakeAll() noexcept { + // wakeAll() is called when a job finishes (to wake up any thread that might be waiting on it) HEAVY_SYSTRACE_CALL(); - std::lock_guard const lock(mWaiterLock); + mWaiterLock.lock(); // this empty critical section is needed -- it guarantees that notify_all() happens // after the condition's variables are set. + mWaiterLock.unlock(); + // notify_all() can be pretty slow, and it doesn't need to be inside the lock. mWaiterCondition.notify_all(); } -void JobSystem::wake(size_t hint) noexcept { +void JobSystem::wakeOne() noexcept { + // wakeOne() is called when a new job is added to a queue HEAVY_SYSTRACE_CALL(); - std::lock_guard const lock(mWaiterLock); - // this empty critical section is needed -- it guarantees that notify_all() happens + mWaiterLock.lock(); + // this empty critical section is needed -- it guarantees that notify_one() happens // after the condition's variables are set. - if (hint == 1) { - mWaiterCondition.notify_one(); - } else { - mWaiterCondition.notify_all(); - } + mWaiterLock.unlock(); + // notify_one() can be pretty slow, and it doesn't need to be inside the lock. + mWaiterCondition.notify_one(); } inline JobSystem::ThreadState& JobSystem::getState() noexcept { - std::lock_guard const lock(mThreadMapLock); + std::lock_guard const lock(mThreadMapLock); auto iter = mThreadMap.find(std::this_thread::get_id()); FILAMENT_CHECK_PRECONDITION(iter != mThreadMap.end()) << "This thread has not been adopted."; return *iter->second; @@ -331,7 +335,7 @@ void JobSystem::put(WorkQueue& workQueue, Job* job) noexcept { // But it's possible that the job has already been picked-up, so oldActiveJobs could be // negative for instance. We signal only if that's not the case. if (oldActiveJobs >= 0) { - wake(oldActiveJobs + 1); // wake-up a thread if needed... + wakeOne(); // wake-up a thread if needed... } } @@ -347,13 +351,9 @@ JobSystem::Job* JobSystem::pop(WorkQueue& workQueue) noexcept { // If our guess was wrong, i.e. we couldn't pick up a job (b/c our queue was empty), we // need to correct mActiveJobs. if (!job) { - int32_t const oldActiveJobs = mActiveJobs.fetch_add(1, std::memory_order_relaxed); - if (oldActiveJobs >= 0) { - // And if there are some active jobs, then we need to wake someone up. We know it - // can't be us, because we failed taking a job and we know another thread can't - // have added one in our queue. - wake(oldActiveJobs + 1); // wake-up a thread if needed... - } + // no need to wake someone else up because, we will go into job-stealing mode + // immediately after this + mActiveJobs.fetch_add(1, std::memory_order_relaxed); } return job; } @@ -364,19 +364,14 @@ JobSystem::Job* JobSystem::steal(WorkQueue& workQueue) noexcept { mActiveJobs.fetch_sub(1, std::memory_order_relaxed); size_t const index = workQueue.steal(); - assert(index <= MAX_JOB_COUNT); + assert_invariant(index <= MAX_JOB_COUNT); Job* const job = !index ? nullptr : &mJobStorageBase[index - 1]; - // If we failed taking a job, we need to correct mActiveJobs. if (!job) { - int32_t const oldActiveJobs = mActiveJobs.fetch_add(1, std::memory_order_relaxed); - if (oldActiveJobs >= 0) { - // And if there are some active jobs, then we need to wake someone up. We know it - // can't be us, because we failed taking a job and we know another thread can't - // have added one in our queue. - wake(oldActiveJobs + 1); // wake-up a thread if needed... - } + // If we failed taking a job, we need to correct mActiveJobs. + mActiveJobs.fetch_add(1, std::memory_order_relaxed); } + return job; } @@ -430,7 +425,9 @@ bool JobSystem::execute(JobSystem::ThreadState& state) noexcept { if (UTILS_LIKELY(job->function)) { HEAVY_SYSTRACE_NAME("job->function"); + job->id = std::distance(mThreadStates.data(), &state); job->function(job->storage, *this, job); + job->id = invalidThreadId; } finish(job); } @@ -442,9 +439,10 @@ void JobSystem::loop(ThreadState* state) noexcept { setThreadPriority(Priority::DISPLAY); // record our work queue - mThreadMapLock.lock(); + std::unique_lock lock(mThreadMapLock); bool const inserted = mThreadMap.emplace(std::this_thread::get_id(), state).second; - mThreadMapLock.unlock(); + lock.unlock(); + FILAMENT_CHECK_PRECONDITION(inserted) << "This thread is already in a loop."; // run our main loop... @@ -532,10 +530,6 @@ void JobSystem::release(JobSystem::Job*& job) noexcept { job = nullptr; } -void JobSystem::signal() noexcept { - wakeAll(); -} - void JobSystem::run(Job*& job) noexcept { HEAVY_SYSTRACE_CALL(); @@ -547,6 +541,18 @@ void JobSystem::run(Job*& job) noexcept { job = nullptr; } +void JobSystem::run(Job*& job, uint8_t id) noexcept { + HEAVY_SYSTRACE_CALL(); + + ThreadState& state = mThreadStates[id]; + assert_invariant(&state == &getState()); + + put(state.workQueue, job); + + // after run() returns, the job is virtually invalid (it'll die on its own) + job = nullptr; +} + JobSystem::Job* JobSystem::runAndRetain(Job* job) noexcept { JobSystem::Job* retained = retain(job); run(job); @@ -591,6 +597,7 @@ void JobSystem::waitAndRelease(Job*& job) noexcept { } void JobSystem::runAndWait(JobSystem::Job*& job) noexcept { + SYSTRACE_CALL(); runAndRetain(job); waitAndRelease(job); } @@ -598,7 +605,7 @@ void JobSystem::runAndWait(JobSystem::Job*& job) noexcept { void JobSystem::adopt() { const auto tid = std::this_thread::get_id(); - std::unique_lock lock(mThreadMapLock); + std::unique_lock lock(mThreadMapLock); auto iter = mThreadMap.find(tid); ThreadState* const state = iter == mThreadMap.end() ? nullptr : iter->second; lock.unlock(); @@ -631,7 +638,7 @@ void JobSystem::adopt() { void JobSystem::emancipate() { const auto tid = std::this_thread::get_id(); - std::lock_guard const lock(mThreadMapLock); + std::unique_lock const lock(mThreadMapLock); auto iter = mThreadMap.find(tid); ThreadState* const state = iter == mThreadMap.end() ? nullptr : iter->second; FILAMENT_CHECK_PRECONDITION(state) << "this thread is not an adopted thread"; diff --git a/tools/matc/src/matc/MaterialCompiler.cpp b/tools/matc/src/matc/MaterialCompiler.cpp index e1b22f97e6..93d0fedc0e 100644 --- a/tools/matc/src/matc/MaterialCompiler.cpp +++ b/tools/matc/src/matc/MaterialCompiler.cpp @@ -24,6 +24,7 @@ #include +#include #include #include "DirIncluder.h" From 4a5081388dd240e4a0ad0d0bf334ecb902c99292 Mon Sep 17 00:00:00 2001 From: Mathias Agopian Date: Fri, 19 Jul 2024 10:31:27 -0700 Subject: [PATCH 2/5] RenderPass batch size was too small The batch size was calculated incorrectly, which in theory could lead to command buffer overflows (albeit unlikely). --- filament/src/RenderPass.cpp | 31 ++++++++++++++++++++++++------- 1 file changed, 24 insertions(+), 7 deletions(-) diff --git a/filament/src/RenderPass.cpp b/filament/src/RenderPass.cpp index cc892b910d..a428d29d46 100644 --- a/filament/src/RenderPass.cpp +++ b/filament/src/RenderPass.cpp @@ -38,6 +38,7 @@ #include #include "private/backend/CircularBuffer.h" +#include "private/backend/CommandStream.h" #include #include @@ -153,7 +154,7 @@ RenderPass::RenderPass(FEngine& engine, RenderPassBuilder const& builder) noexce sortCommands(builder.mArena); if (engine.isAutomaticInstancingEnabled()) { - uint32_t stereoscopicEyeCount = 1; + int32_t stereoscopicEyeCount = 1; if (builder.mFlags & IS_INSTANCED_STEREOSCOPIC) { stereoscopicEyeCount *= engine.getConfig().stereoscopicEyeCount; } @@ -162,8 +163,7 @@ RenderPass::RenderPass(FEngine& engine, RenderPassBuilder const& builder) noexce } // this destructor is actually heavy because it inlines ~vector<> -RenderPass::~RenderPass() noexcept { -} +RenderPass::~RenderPass() noexcept = default; void RenderPass::resize(Arena& arena, size_t count) noexcept { if (mCommandBegin) { @@ -885,14 +885,31 @@ void RenderPass::Executor::execute(FEngine& engine, // Maximum space occupied in the CircularBuffer by a single `Command`. This must be // reevaluated when the inner loop below adds DriverApi commands or when we change the - // CommandStream protocol. Currently, the maximum is 240 bytes, and we use 256 to be on - // the safer side. - size_t const maxCommandSizeInBytes = 256; + // CommandStream protocol. Currently, the maximum is 320 bytes. + // The batch size is calculated by adding the size of all commands that can possibly be + // emitted per draw call: + constexpr size_t const maxCommandSizeInBytes = + sizeof(CustomCommand) + + sizeof(COMMAND_TYPE(scissor)) + + sizeof(COMMAND_TYPE(bindUniformBuffer)) + + sizeof(COMMAND_TYPE(bindSamplers)) + + sizeof(COMMAND_TYPE(bindBufferRange)) + + sizeof(COMMAND_TYPE(bindBufferRange)) + + sizeof(COMMAND_TYPE(bindSamplers)) + + sizeof(COMMAND_TYPE(bindSamplers)) + + sizeof(COMMAND_TYPE(bindUniformBuffer)) + + sizeof(COMMAND_TYPE(bindSamplers)) + + sizeof(COMMAND_TYPE(bindSamplers)) + + sizeof(COMMAND_TYPE(bindPipeline)) + + sizeof(COMMAND_TYPE(setPushConstant)) + + sizeof(COMMAND_TYPE(bindRenderPrimitive)) + + sizeof(COMMAND_TYPE(draw2)); + // Number of Commands that can be issued and guaranteed to fit in the current // CircularBuffer allocation. In practice, we'll have tons of headroom especially if // skinning and morphing aren't used. With a 2 MiB buffer (the default) a batch is - // 8192 commands (i.e. draw calls). + // 6553 commands (i.e. draw calls). size_t const batchCommandCount = capacity / maxCommandSizeInBytes; while(first != last) { Command const* const batchLast = std::min(first + batchCommandCount, last); From 9057c47a566dbcbba96bcfa3ec0a51eb1d5e1736 Mon Sep 17 00:00:00 2001 From: Sungun Park Date: Tue, 23 Jul 2024 01:23:57 +0000 Subject: [PATCH 3/5] Release Filament 1.53.2 --- NEW_RELEASE_NOTES.md | 2 -- README.md | 4 ++-- RELEASE_NOTES.md | 4 ++++ android/gradle.properties | 2 +- ios/CocoaPods/Filament.podspec | 4 ++-- web/filament-js/package.json | 2 +- 6 files changed, 10 insertions(+), 8 deletions(-) diff --git a/NEW_RELEASE_NOTES.md b/NEW_RELEASE_NOTES.md index 228c34804a..4a1a9c7fa7 100644 --- a/NEW_RELEASE_NOTES.md +++ b/NEW_RELEASE_NOTES.md @@ -7,5 +7,3 @@ for next branch cut* header. appropriate header in [RELEASE_NOTES.md](./RELEASE_NOTES.md). ## Release notes for next branch cut - -- Add drag and drop support for IBL files for desktop gltf_viewer. diff --git a/README.md b/README.md index b368b37c95..97e81b5e08 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,7 @@ repositories { } dependencies { - implementation 'com.google.android.filament:filament-android:1.53.1' + implementation 'com.google.android.filament:filament-android:1.53.2' } ``` @@ -51,7 +51,7 @@ Here are all the libraries available in the group `com.google.android.filament`: iOS projects can use CocoaPods to install the latest release: ```shell -pod 'Filament', '~> 1.53.1' +pod 'Filament', '~> 1.53.2' ``` ### Snapshots diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 3591acf61e..d1392c5a60 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -7,6 +7,10 @@ A new header is inserted each time a *tag* is created. Instead, if you are authoring a PR for the main branch, add your release note to [NEW_RELEASE_NOTES.md](./NEW_RELEASE_NOTES.md). +## v1.53.3 + +- Add drag and drop support for IBL files for desktop gltf_viewer. + ## v1.53.2 diff --git a/android/gradle.properties b/android/gradle.properties index cca037f8f7..7994c61d4a 100644 --- a/android/gradle.properties +++ b/android/gradle.properties @@ -1,5 +1,5 @@ GROUP=com.google.android.filament -VERSION_NAME=1.53.1 +VERSION_NAME=1.53.2 POM_DESCRIPTION=Real-time physically based rendering engine for Android. diff --git a/ios/CocoaPods/Filament.podspec b/ios/CocoaPods/Filament.podspec index fc42e6ee98..f1f1105fa3 100644 --- a/ios/CocoaPods/Filament.podspec +++ b/ios/CocoaPods/Filament.podspec @@ -1,12 +1,12 @@ Pod::Spec.new do |spec| spec.name = "Filament" - spec.version = "1.53.1" + spec.version = "1.53.2" spec.license = { :type => "Apache 2.0", :file => "LICENSE" } spec.homepage = "https://google.github.io/filament" spec.authors = "Google LLC." spec.summary = "Filament is a real-time physically based rendering engine for Android, iOS, Windows, Linux, macOS, and WASM/WebGL." spec.platform = :ios, "11.0" - spec.source = { :http => "https://github.com/google/filament/releases/download/v1.53.1/filament-v1.53.1-ios.tgz" } + spec.source = { :http => "https://github.com/google/filament/releases/download/v1.53.2/filament-v1.53.2-ios.tgz" } # Fix linking error with Xcode 12; we do not yet support the simulator on Apple silicon. spec.pod_target_xcconfig = { diff --git a/web/filament-js/package.json b/web/filament-js/package.json index 7cea353927..c381eef571 100644 --- a/web/filament-js/package.json +++ b/web/filament-js/package.json @@ -1,6 +1,6 @@ { "name": "filament", - "version": "1.53.1", + "version": "1.53.2", "description": "Real-time physically based rendering engine", "main": "filament.js", "module": "filament.js", From d3a35de38667a7f82384dff4773ddf75a3735e17 Mon Sep 17 00:00:00 2001 From: Mathias Agopian Date: Tue, 23 Jul 2024 23:31:20 -0700 Subject: [PATCH 4/5] fix a crash with bloom when screen dimension smaller than 16px --- filament/src/PostProcessManager.cpp | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/filament/src/PostProcessManager.cpp b/filament/src/PostProcessManager.cpp index 85e21d6832..da5c120b8e 100644 --- a/filament/src/PostProcessManager.cpp +++ b/filament/src/PostProcessManager.cpp @@ -1965,8 +1965,8 @@ PostProcessManager::BloomPassOutput PostProcessManager::bloom(FrameGraph& fg, // - visible bloom size changes with dynamic resolution in non-homogenous mode // This allows us to use the 9 sample downsampling filter (instead of 13) // for at least 4 levels. - uint32_t width = std::max(1u, uint32_t(std::floor(bloomWidth))); - uint32_t height = std::max(1u, uint32_t(std::floor(bloomHeight))); + uint32_t width = std::max(16u, uint32_t(std::floor(bloomWidth))); + uint32_t height = std::max(16u, uint32_t(std::floor(bloomHeight))); width &= ~((1 << 4) - 1); // at least 4 levels height &= ~((1 << 4) - 1); bloomWidth = float(width); @@ -1978,6 +1978,8 @@ PostProcessManager::BloomPassOutput PostProcessManager::bloom(FrameGraph& fg, // we don't need to do the fireflies reduction if we have TAA (it already does it) bool fireflies = threshold && !taaOptions.enabled; + assert_invariant(bloomWidth && bloomHeight); + while (2 * bloomWidth < float(desc.width) || 2 * bloomHeight < float(desc.height)) { if (inoutBloomOptions.quality == QualityLevel::LOW || inoutBloomOptions.quality == QualityLevel::MEDIUM) { From be4f287b07987f4aead4119dc53ba7fa21676489 Mon Sep 17 00:00:00 2001 From: Mathias Agopian Date: Wed, 24 Jul 2024 15:34:36 -0700 Subject: [PATCH 5/5] More improvements to the JobSystem (#7988) * improve parallel_for a bit We get about 40% performance increase. The gain comes from not having to copy the JobData structure each time we create a job, by using a new emplaceJob() method, we can create the structure directly into its destination. * avoid calling wakeAll() when possible wakeAll() is very expensive and not always needed when a job finishes because there may not be anyone waiting on that job. We now maintain a waiter count per job, and use that to determine if we need to notify or not. And now that the JobSystem overhead is lower, we can decrease the size of the jobs, which improves the load balancing. * mActiveJobs fixes some comments claimed mActiveJobs needed to be modified before or after accessing the WorkQueue; this couldn't be correct because there were no guaranteed global ordering with the workQueue. --- filament/src/RenderPass.cpp | 2 +- filament/src/RenderPass.h | 2 +- filament/src/details/Scene.cpp | 2 +- libs/utils/include/utils/JobSystem.h | 86 +++++++-- .../utils/include/utils/WorkStealingDequeue.h | 20 ++- libs/utils/src/JobSystem.cpp | 166 +++++++++--------- 6 files changed, 167 insertions(+), 111 deletions(-) diff --git a/filament/src/RenderPass.cpp b/filament/src/RenderPass.cpp index a428d29d46..63811f8b17 100644 --- a/filament/src/RenderPass.cpp +++ b/filament/src/RenderPass.cpp @@ -221,7 +221,7 @@ void RenderPass::appendCommands(FEngine& engine, work(vr.first, vr.size()); } else { auto* jobCommandsParallel = jobs::parallel_for(js, nullptr, vr.first, (uint32_t)vr.size(), - std::cref(work), jobs::CountSplitter()); + std::cref(work), jobs::CountSplitter()); js.runAndWait(jobCommandsParallel); } diff --git a/filament/src/RenderPass.h b/filament/src/RenderPass.h index 23c4cf8249..ebe4f37642 100644 --- a/filament/src/RenderPass.h +++ b/filament/src/RenderPass.h @@ -417,7 +417,7 @@ private: void instanceify(FEngine& engine, Arena& arena, int32_t eyeCount) noexcept; // We choose the command count per job to minimize JobSystem overhead. - static constexpr size_t JOBS_PARALLEL_FOR_COMMANDS_COUNT = 1024; + static constexpr size_t JOBS_PARALLEL_FOR_COMMANDS_COUNT = 128; static constexpr size_t JOBS_PARALLEL_FOR_COMMANDS_SIZE = sizeof(Command) * JOBS_PARALLEL_FOR_COMMANDS_COUNT; diff --git a/filament/src/details/Scene.cpp b/filament/src/details/Scene.cpp index c5b0d1e19e..ece2e99f93 100644 --- a/filament/src/details/Scene.cpp +++ b/filament/src/details/Scene.cpp @@ -248,7 +248,7 @@ void FScene::prepare(utils::JobSystem& js, auto* renderableJob = jobs::parallel_for(js, rootJob, renderableInstances.data(), renderableInstances.size(), - std::cref(renderableWork), jobs::CountSplitter<128, 5>()); + std::cref(renderableWork), jobs::CountSplitter<64>()); auto* lightJob = jobs::parallel_for(js, rootJob, lightInstances.data(), lightInstances.size(), diff --git a/libs/utils/include/utils/JobSystem.h b/libs/utils/include/utils/JobSystem.h index e3786dc322..7ea711661a 100644 --- a/libs/utils/include/utils/JobSystem.h +++ b/libs/utils/include/utils/JobSystem.h @@ -44,7 +44,9 @@ namespace utils { class JobSystem { - static constexpr size_t MAX_JOB_COUNT = 16384; + static constexpr size_t MAX_JOB_COUNT = 1 << 14; // 16384 + static constexpr uint32_t JOB_COUNT_MASK = MAX_JOB_COUNT - 1; + static constexpr uint32_t WAITER_COUNT_SHIFT = 24; static_assert(MAX_JOB_COUNT <= 0x7FFE, "MAX_JOB_COUNT must be <= 0x7FFE"); using WorkQueue = WorkStealingDequeue; using Mutex = utils::Mutex; @@ -81,14 +83,18 @@ public: void* storage[JOB_STORAGE_SIZE_WORDS]; // 48 | 48 JobFunc function; // 4 | 8 uint16_t parent; // 2 | 2 - std::atomic runningJobCount = { 1 }; // 2 | 2 - mutable std::atomic refCount = { 1 }; // 2 | 2 mutable ThreadId id = invalidThreadId; // 1 | 1 - // 1 | 1 (padding) + mutable std::atomic refCount = { 1 }; // 1 | 1 + std::atomic runningJobCount = { 1 }; // 4 | 4 // 4 | 0 (padding) // 64 | 64 }; +#ifndef WIN32 + // on windows std::function is bigger and forces the whole structure to be larger + static_assert(sizeof(Job) == 64); +#endif + explicit JobSystem(size_t threadCount = 0, size_t adoptableThreadsCount = 1) noexcept; ~JobSystem(); @@ -207,6 +213,21 @@ public: return job; } + // creates a job from a KNOWN method pointer w/ object passed by value + template + Job* emplaceJob(Job* parent, ARGS&& ... args) noexcept { + static_assert(sizeof(T) <= sizeof(Job::storage), "user data too large"); + Job* job = create(parent, [](void* storage, JobSystem& js, Job* job) { + T* const that = static_cast(storage); + (that->*method)(js, job); + that->~T(); + }); + if (job) { + new(job->storage) T(std::forward(args)...); + } + return job; + } + // creates a job from a functor passed by value template Job* createJob(Job* parent, T functor) noexcept { @@ -222,6 +243,21 @@ public: return job; } + // creates a job from a functor passed by value + template + Job* emplaceJob(Job* parent, ARGS&& ... args) noexcept { + static_assert(sizeof(T) <= sizeof(Job::storage), "functor too large"); + Job* job = create(parent, [](void* storage, JobSystem& js, Job* job){ + T* const that = static_cast(storage); + that->operator()(js, job); + that->~T(); + }); + if (job) { + new(job->storage) T(std::forward(args)...); + } + return job; + } + /* * Jobs are normally finished automatically, this can be used to cancel a job before it is run. @@ -345,6 +381,16 @@ private: static constexpr uint32_t m = 0x7fffffffu; uint32_t mState; // must be 0 < seed < 0x7fffffff public: + using result_type = uint32_t; + + static constexpr result_type min() noexcept { + return 1; + } + + static constexpr result_type max() noexcept { + return m - 1; + } + inline constexpr explicit default_random_engine(uint32_t seed = 1u) noexcept : mState(((seed % m) == 0u) ? 1u : seed % m) { } @@ -389,7 +435,9 @@ private: Job* pop(WorkQueue& workQueue) noexcept; Job* steal(WorkQueue& workQueue) noexcept; - void wait(std::unique_lock& lock, Job* job = nullptr) noexcept; + [[nodiscard]] + uint32_t wait(std::unique_lock& lock, Job* job) noexcept; + void wait(std::unique_lock& lock) noexcept; void wakeAll() noexcept; void wakeOne() noexcept; @@ -418,7 +466,7 @@ private: uint8_t mParallelSplitCount = 0; // # of split allowable in parallel_for Job* mRootJob = nullptr; - utils::Mutex mThreadMapLock; // this should have very little contention + Mutex mThreadMapLock; // this should have very little contention tsl::robin_map mThreadMap; }; @@ -436,12 +484,13 @@ template JobSystem::Job* createJob(JobSystem& js, JobSystem::Job* parent, CALLABLE&& func, ARGS&&... args) noexcept { struct Data { + explicit Data(std::function f) noexcept: f(std::move(f)) {} std::function f; // Renaming the method below could cause an Arrested Development. void gob(JobSystem&, JobSystem::Job*) noexcept { f(); } - } user{ std::bind(std::forward(func), - std::forward(args)...) }; - return js.createJob(parent, std::move(user)); + }; + return js.emplaceJob(parent, + std::bind(std::forward(func), std::forward(args)...)); } template f) noexcept: f(std::move(f)) {} std::function f; // Renaming the method below could cause an Arrested Development. void gob(JobSystem&, JobSystem::Job*) noexcept { f(); } - } user{ std::bind(std::forward(func), std::forward(o), - std::forward(args)...) }; - return js.createJob(parent, std::move(user)); + }; + return js.emplaceJob(parent, + std::bind(std::forward(func), std::forward(o), std::forward(args)...)); } @@ -486,8 +536,8 @@ struct ParallelForJobData { right_side: if (splitter.split(splits, count)) { const size_type lc = count / 2; - JobData ld(start, lc, splits + uint8_t(1), functor, splitter); - JobSystem::Job* l = js.createJob(parent, std::move(ld)); + JobSystem::Job* l = js.emplaceJob(parent, + start, lc, splits + uint8_t(1), functor, splitter); if (UTILS_UNLIKELY(l == nullptr)) { // couldn't create a job, just pretend we're done splitting goto execute; @@ -527,8 +577,8 @@ template JobSystem::Job* parallel_for(JobSystem& js, JobSystem::Job* parent, uint32_t start, uint32_t count, F functor, const S& splitter) noexcept { using JobData = details::ParallelForJobData; - JobData jobData(start, count, 0, std::move(functor), splitter); - return js.createJob(parent, std::move(jobData)); + return js.emplaceJob(parent, + start, count, 0, std::move(functor), splitter); } // parallel jobs with pointer/count @@ -539,8 +589,8 @@ JobSystem::Job* parallel_for(JobSystem& js, JobSystem::Job* parent, f(data + s, c); }; using JobData = details::ParallelForJobData; - JobData jobData(0, count, 0, std::move(user), splitter); - return js.createJob(parent, std::move(jobData)); + return js.emplaceJob(parent, + 0, count, 0, std::move(user), splitter); } // parallel jobs on a Slice<> diff --git a/libs/utils/include/utils/WorkStealingDequeue.h b/libs/utils/include/utils/WorkStealingDequeue.h index 9e737d0aa7..f9f5b5fbc2 100644 --- a/libs/utils/include/utils/WorkStealingDequeue.h +++ b/libs/utils/include/utils/WorkStealingDequeue.h @@ -35,7 +35,13 @@ namespace utils { * steal() push(), pop() * any thread main thread * - * + * References: + * - This code is largely inspired from + * https://blog.molecular-matters.com/2015/09/25/job-system-2-0-lock-free-work-stealing-part-3-going-lock-free/ + * - other implementations + * https://github.com/ConorWilliams/ConcurrentDeque/blob/main/include/riften/deque.hpp + * https://github.com/ssbl/concurrent-deque/blob/master/include/deque.hpp + * https://github.com/taskflow/work-stealing-queue/blob/master/wsq.hpp */ template class WorkStealingDequeue { @@ -117,7 +123,7 @@ TYPE WorkStealingDequeue::pop() noexcept { index_t top = mTop.load(std::memory_order_seq_cst); if (top < bottom) { - // Queue isn't empty and it's not the last item, just return it, this is the common case. + // Queue isn't empty, and it's not the last item, just return it, this is the common case. return getItemAt(bottom); } @@ -132,13 +138,13 @@ TYPE WorkStealingDequeue::pop() noexcept { if (mTop.compare_exchange_strong(top, top + 1, std::memory_order_seq_cst, std::memory_order_relaxed)) { - // success: we stole our last item from ourself, meaning that a concurrent steal() + // Success: we stole our last item from ourselves, meaning that a concurrent steal() // would have failed. // mTop now equals top + 1, we adjust top to make the queue empty. top++; } else { - // failure: mTop was not equal to top, which means the item was stolen under our feet. - // top now equals to mTop. Simply discard the item we just popped. + // Failure: mTop was not equal to top, which means the item was stolen under our feet. + // `top` now equals to mTop. Simply discard the item we just popped. // The queue is now empty. item = TYPE(); } @@ -149,7 +155,7 @@ TYPE WorkStealingDequeue::pop() noexcept { } // std::memory_order_relaxed used because we're not publishing any data. - // no concurrent writes to mBottom possible, it's always safe to write mBottom. + // No concurrent writes to mBottom possible, it's always safe to write mBottom. mBottom.store(top, std::memory_order_relaxed); return item; } @@ -194,6 +200,8 @@ TYPE WorkStealingDequeue::steal() noexcept { } // failure: the item we just tried to steal was pop()'ed under our feet, // simply discard it; nothing to do -- it's okay to try again. + // However, item might be corrupted, so it must be trivially destructible + static_assert(std::is_trivially_destructible_v); } } diff --git a/libs/utils/src/JobSystem.cpp b/libs/utils/src/JobSystem.cpp index b0e199fcef..57640cd6c9 100644 --- a/libs/utils/src/JobSystem.cpp +++ b/libs/utils/src/JobSystem.cpp @@ -23,9 +23,6 @@ // when SYSTRACE_TAG_JOBSYSTEM is used, enables even heavier systraces #define HEAVY_SYSTRACE 0 -// enable for catching hangs waiting on a job to finish -static constexpr bool DEBUG_FINISH_HANGS = false; - #include #include @@ -249,53 +246,44 @@ inline bool JobSystem::hasActiveJobs() const noexcept { } inline bool JobSystem::hasJobCompleted(JobSystem::Job const* job) noexcept { - return job->runningJobCount.load(std::memory_order_acquire) <= 0; + return (job->runningJobCount.load(std::memory_order_acquire) & JOB_COUNT_MASK) == 0; } -void JobSystem::wait(std::unique_lock& lock, Job* job) noexcept { +inline void JobSystem::wait(std::unique_lock& lock) noexcept { HEAVY_SYSTRACE_CALL(); - if constexpr (!DEBUG_FINISH_HANGS) { - mWaiterCondition.wait(lock); - } else { - do { - // we use a pretty long timeout (4s) so we're very confident that the system is hung - // and nothing else is happening. - std::cv_status status = mWaiterCondition.wait_for(lock, - std::chrono::milliseconds(4000)); - if (status == std::cv_status::no_timeout) { - break; - } - - // hang debugging... - - // We check of we had active jobs or if the job we're waiting on had completed already. - // There is the possibility of a race condition, but our long timeout gives us some - // confidence that we're in an incorrect state. - - size_t const id = std::distance(mThreadStates.data(), &getState()); - auto activeJobs = mActiveJobs.load(); - - if (job) { - auto runningJobCount = job->runningJobCount.load(); - FILAMENT_CHECK_POSTCONDITION(runningJobCount > 0) - << "JobSystem(" << this << ", " << unsigned(id) << "): waiting while job " - << job << " has completed and " << activeJobs << " jobs are active!"; - } - - FILAMENT_CHECK_POSTCONDITION(activeJobs <= 0) - << "JobSystem(" << this << ", " << unsigned(id) << "): waiting while " - << activeJobs << " jobs are active!"; - - } while (true); - } + mWaiterCondition.wait(lock); } +inline uint32_t JobSystem::wait(std::unique_lock& lock, Job* const job) noexcept { + HEAVY_SYSTRACE_CALL(); + // signal we are waiting + + if (hasActiveJobs() || exitRequested()) { + return job->runningJobCount.load(std::memory_order_acquire); + } + + uint32_t runningJobCount = + job->runningJobCount.fetch_add(1 << WAITER_COUNT_SHIFT, std::memory_order_relaxed); + + if (runningJobCount & JOB_COUNT_MASK) { + mWaiterCondition.wait(lock); + } + + runningJobCount = + job->runningJobCount.fetch_sub(1 << WAITER_COUNT_SHIFT, std::memory_order_acquire); + + assert_invariant((runningJobCount >> WAITER_COUNT_SHIFT) >= 1); + + return runningJobCount; +} + +UTILS_NOINLINE void JobSystem::wakeAll() noexcept { // wakeAll() is called when a job finishes (to wake up any thread that might be waiting on it) - HEAVY_SYSTRACE_CALL(); + SYSTRACE_CALL(); mWaiterLock.lock(); // this empty critical section is needed -- it guarantees that notify_all() happens - // after the condition's variables are set. + // either before the condition is checked, or after the condition variable sleeps. mWaiterLock.unlock(); // notify_all() can be pretty slow, and it doesn't need to be inside the lock. mWaiterCondition.notify_all(); @@ -306,7 +294,7 @@ void JobSystem::wakeOne() noexcept { HEAVY_SYSTRACE_CALL(); mWaiterLock.lock(); // this empty critical section is needed -- it guarantees that notify_one() happens - // after the condition's variables are set. + // either before the condition is checked, or after the condition variable sleeps. mWaiterLock.unlock(); // notify_one() can be pretty slow, and it doesn't need to be inside the lock. mWaiterCondition.notify_one(); @@ -328,50 +316,37 @@ void JobSystem::put(WorkQueue& workQueue, Job* job) noexcept { size_t const index = job - mJobStorageBase; assert(index >= 0 && index < MAX_JOB_COUNT); - // put the job into the queue first + // put the job into the queue workQueue.push(uint16_t(index + 1)); - // then increase our active job count - int32_t const oldActiveJobs = mActiveJobs.fetch_add(1, std::memory_order_relaxed); - // But it's possible that the job has already been picked-up, so oldActiveJobs could be - // negative for instance. We signal only if that's not the case. - if (oldActiveJobs >= 0) { - wakeOne(); // wake-up a thread if needed... - } + + // increase our active job count (the order in which we're doing this must not matter + // because we're not using std::memory_order_seq_cst (here or in WorkQueue::push()). + mActiveJobs.fetch_add(1, std::memory_order_relaxed); + + // Note: it's absolutely possible for mActiveJobs to be 0 here, because the job could have + // been handled by a zealous worker already. In that case we could avoid calling wakeOne(), + // but that is not the common case. + + wakeOne(); } JobSystem::Job* JobSystem::pop(WorkQueue& workQueue) noexcept { - // decrement mActiveJobs first, this is to ensure that if there is only a single job left - // (and we're about to pick it up), other threads don't loop trying to do the same. - mActiveJobs.fetch_sub(1, std::memory_order_relaxed); - size_t const index = workQueue.pop(); assert(index <= MAX_JOB_COUNT); Job* const job = !index ? nullptr : &mJobStorageBase[index - 1]; - - // If our guess was wrong, i.e. we couldn't pick up a job (b/c our queue was empty), we - // need to correct mActiveJobs. - if (!job) { - // no need to wake someone else up because, we will go into job-stealing mode - // immediately after this - mActiveJobs.fetch_add(1, std::memory_order_relaxed); + if (UTILS_LIKELY(job)) { + mActiveJobs.fetch_sub(1, std::memory_order_relaxed); } return job; } JobSystem::Job* JobSystem::steal(WorkQueue& workQueue) noexcept { - // decrement mActiveJobs first, this is to ensure that if there is only a single job left - // (and we're about to pick it up), other threads don't loop trying to do the same. - mActiveJobs.fetch_sub(1, std::memory_order_relaxed); - size_t const index = workQueue.steal(); assert_invariant(index <= MAX_JOB_COUNT); Job* const job = !index ? nullptr : &mJobStorageBase[index - 1]; - - if (!job) { - // If we failed taking a job, we need to correct mActiveJobs. - mActiveJobs.fetch_add(1, std::memory_order_relaxed); + if (UTILS_LIKELY(job)) { + mActiveJobs.fetch_sub(1, std::memory_order_relaxed); } - return job; } @@ -402,7 +377,7 @@ JobSystem::Job* JobSystem::steal(JobSystem::ThreadState& state) noexcept { Job* job = nullptr; do { ThreadState* const stateToStealFrom = getStateToStealFrom(state); - if (UTILS_LIKELY(stateToStealFrom)) { + if (stateToStealFrom) { job = steal(stateToStealFrom->workQueue); } // nullptr -> nothing to steal in that queue either, if there are active jobs, @@ -415,14 +390,18 @@ bool JobSystem::execute(JobSystem::ThreadState& state) noexcept { HEAVY_SYSTRACE_CALL(); Job* job = pop(state.workQueue); - if (UTILS_UNLIKELY(job == nullptr)) { + + // It is beneficial for some benchmarks to poll on steal() for a bit, because going back to + // sleep and waking up is pretty expensive. However, it is unclear it helps in practice with + // larger jobs or when parallel_for is used. + constexpr size_t const STEAL_TRY_COUNT = 1; + for (size_t i = 0; UTILS_UNLIKELY(!job && i < STEAL_TRY_COUNT); i++) { // our queue is empty, try to steal a job job = steal(state); } - if (job) { - assert(job->runningJobCount.load(std::memory_order_relaxed) >= 1); - + if (UTILS_LIKELY(job)) { + assert((job->runningJobCount.load(std::memory_order_relaxed) & JOB_COUNT_MASK) >= 1); if (UTILS_LIKELY(job->function)) { HEAVY_SYSTRACE_NAME("job->function"); job->id = std::distance(mThreadStates.data(), &state); @@ -467,11 +446,16 @@ void JobSystem::finish(Job* job) noexcept { do { // std::memory_order_release here is needed to synchronize with JobSystem::wait() // which needs to "see" all changes that happened before the job terminated. - auto runningJobCount = job->runningJobCount.fetch_sub(1, std::memory_order_acq_rel); + uint32_t const v = job->runningJobCount.fetch_sub(1, std::memory_order_acq_rel); + uint32_t const runningJobCount = v & JOB_COUNT_MASK; assert(runningJobCount > 0); + if (runningJobCount == 1) { // no more work, destroy this job and notify its parent - notify = true; + uint32_t const waiters = v >> WAITER_COUNT_SHIFT; + if (waiters) { + notify = true; + } Job* const parent = job->parent == 0x7FFF ? nullptr : &storage[job->parent]; decRef(job); job = parent; @@ -482,7 +466,8 @@ void JobSystem::finish(Job* job) noexcept { } while (job); // wake-up all threads that could potentially be waiting on this job finishing - if (notify) { + if (UTILS_UNLIKELY(notify)) { + // but avoid calling notify_all() at all cost, because it's always expensive wakeAll(); } } @@ -500,10 +485,11 @@ JobSystem::Job* JobSystem::create(JobSystem::Job* parent, JobFunc func) noexcept // add a reference to the parent to make sure it can't be terminated. // memory_order_relaxed is safe because no action is taken at this point // (the job is not started yet). - auto parentJobCount = parent->runningJobCount.fetch_add(1, std::memory_order_relaxed); + UTILS_UNUSED_IN_RELEASE auto const parentJobCount = + parent->runningJobCount.fetch_add(1, std::memory_order_relaxed); // can't create a child job of a terminated parent - assert(parentJobCount > 0); + assert((parentJobCount & JOB_COUNT_MASK) > 0); index = parent - mJobStorageBase; assert(index < MAX_JOB_COUNT); @@ -567,7 +553,7 @@ void JobSystem::waitAndRelease(Job*& job) noexcept { ThreadState& state(getState()); do { - if (!execute(state)) { + if (UTILS_UNLIKELY(!execute(state))) { // test if job has completed first, to possibly avoid taking the lock if (hasJobCompleted(job)) { break; @@ -583,11 +569,23 @@ void JobSystem::waitAndRelease(Job*& job) noexcept { // continue to handle more jobs, as they get added. std::unique_lock lock(mWaiterLock); - if (!hasJobCompleted(job) && !hasActiveJobs() && !exitRequested()) { - wait(lock, job); + uint32_t const runningJobCount = wait(lock, job); + // we could be waking up because either: + // - the job we're waiting on has completed + // - more jobs where added to the JobSystem + // - we're asked to exit + if ((runningJobCount & JOB_COUNT_MASK) == 0 || exitRequested()) { + break; } + + // if we get here, it means that + // - the job we're waiting on is still running, and + // - we're not asked to exit, and + // - there were some active jobs + // So we try to handle one. + continue; } - } while (!hasJobCompleted(job) && !exitRequested()); + } while (UTILS_LIKELY(!hasJobCompleted(job) && !exitRequested())); if (job == mRootJob) { mRootJob = nullptr;