From be4f287b07987f4aead4119dc53ba7fa21676489 Mon Sep 17 00:00:00 2001 From: Mathias Agopian Date: Wed, 24 Jul 2024 15:34:36 -0700 Subject: [PATCH] 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;