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;