Scarfski/filament
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

improvements to JobSystem

Browse Source 
- we simplify the waiting code by using only a single
condition variable instead of two.

- wait() now behaves just like a looper, it will process jobs until
the one it's waiting for finishes -- before it could just sit there
(the idea was that the job would finish quickly, but that's not always
the case).

- we also make sure to never call notify_n() when it's not needed.
We track how many waiters we have and use that to decide if we need
to notify().

notify is pretty slow on all architectures, even on linux it's always
a syscall, so it's better to avoid it.

- don't use stand-alone fences, makes things ugly for no real benefit

- refactored the code a bit, hopefully it's more clear.
This commit is contained in:
Mathias Agopian
2019-07-01 18:42:37 -07:00
committed by Mathias Agopian
parent fc1d39334f
commit 2df639133b
2 changed files with 79 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

137
libs/utils/src/JobSystem.cpp
View File

@@ -87,22 +87,6 @@ void JobSystem::setThreadPriority(Priority priority) noexcept {
#endif
}
void JobSystem::setThreadAffinity(uint32_t mask) noexcept {
#if defined(__linux__)
int bit = 0;
cpu_set_t set;
CPU_ZERO(&set);
while (mask) {
if (mask & 1) {
CPU_SET(bit, &set);
}
bit++;
mask >>= 1;
}
sched_setaffinity(gettid(), sizeof(set), &set);
#endif
}
void JobSystem::setThreadAffinityById(size_t id) noexcept {
#if defined(__linux__)
cpu_set_t set;
@@ -185,16 +169,10 @@ void JobSystem::decRef(Job const* job) noexcept {
// Similarly, we need to guarantee that no read/write are reordered before the last decref,
// or some other thread could see a destroyed object before the ref-count is 0. This is done
// with memory_order_acquire.
auto c = job->refCount.fetch_sub(1, __has_feature(thread_sanitizer) ?
std::memory_order_acq_rel :
std::memory_order_release);
auto c = job->refCount.fetch_sub(1, std::memory_order_acq_rel);
assert(c > 0);
if (c == 1) {
// This was the last reference, it's safe to destroy the job.
#if !__has_feature(thread_sanitizer)
// TSAN doesn't handle standalone fences, we use memory_order_acq_rel instead
std::atomic_thread_fence(std::memory_order_acquire);
#endif
mJobPool.destroy(job);
}
}
@@ -206,12 +184,9 @@ JobSystem* JobSystem::getJobSystem() noexcept {
void JobSystem::requestExit() noexcept {
mExitRequested.store(true);
{ std::lock_guard<Mutex> lock(mLooperLock); }
mLooperCondition.notify_all();
{ std::lock_guard<Mutex> lock(mWaiterLock); }
mWaiterCondition.notify_all();
}
inline bool JobSystem::exitRequested() const noexcept {
@@ -219,10 +194,28 @@ inline bool JobSystem::exitRequested() const noexcept {
return mExitRequested.load(std::memory_order_relaxed);
}
inline bool JobSystem::hasActiveJobs() const noexcept {
return mActiveJobs.load(std::memory_order_relaxed) > 0;
}
inline bool JobSystem::hasJobCompleted(JobSystem::Job const* job) noexcept {
return job->runningJobCount.load(std::memory_order_relaxed) <= 0;
}
template <typename Mutex>
void JobSystem::wait(std::unique_lock<Mutex>& lock) noexcept {
++mWaiterCount;
mWaiterCondition.wait(lock);
--mWaiterCount;
}
void JobSystem::wake() noexcept {
mWaiterLock.lock();
const uint32_t waiterCount = mWaiterCount;
mWaiterLock.unlock();
mWaiterCondition.notify_n(waiterCount);
}
inline JobSystem::ThreadState& JobSystem::getState() noexcept {
// check we're not using a thread not owned by the thread pool
assert(sThreadState);
@@ -234,36 +227,47 @@ JobSystem::Job* JobSystem::allocateJob() noexcept {
}
inline JobSystem::ThreadState* JobSystem::getStateToStealFrom(JobSystem::ThreadState& state) noexcept {
auto& threadStates = mThreadStates;
// memory_order_relaxed is okay because we don't take any action that has data dependency
// on this value (in particular mThreadStates, is always initialized properly).
uint16_t adopted = mAdoptedThreads.load(std::memory_order_relaxed);
// this is biased, but frankly, we don't care. it's fast.
uint16_t index = uint16_t(state.rndGen() % (mThreadCount + adopted));
assert(index < mThreadStates.size());
return &mThreadStates[index];
uint16_t const threadCount = mThreadCount + adopted;
JobSystem::ThreadState* stateToStealFrom = nullptr;
do {
// this is biased, but frankly, we don't care. it's fast.
uint16_t index = uint16_t(state.rndGen() % threadCount);
assert(index < threadStates.size());
stateToStealFrom = &threadStates[index];
// don't steal from our own queue
} while (stateToStealFrom == &state);
return stateToStealFrom;
}
JobSystem::Job* JobSystem::steal(JobSystem::ThreadState& state) noexcept {
SYSTRACE_CALL();
Job* job = nullptr;
do {
ThreadState* const stateToStealFrom = getStateToStealFrom(state);
job = steal(stateToStealFrom->workQueue);
// nullptr -> nothing to steal in that queue either, if there are active jobs,
// continue to try stealing one.
} while (!job && hasActiveJobs());
return job;
}
bool JobSystem::execute(JobSystem::ThreadState& state) noexcept {
SYSTRACE_CALL();
Job* job = pop(state.workQueue);
if (job == nullptr) {
if (UTILS_UNLIKELY(job == nullptr)) {
// our queue is empty, try to steal a job
do {
ThreadState* stateToStealFrom = nullptr;
do {
stateToStealFrom = getStateToStealFrom(state);
// don't steal from our own queue
} while (stateToStealFrom == &state);
job = steal(stateToStealFrom->workQueue);
// nullptr -> nothing to steal in that queue either, if there are active jobs,
// continue to try stealing one.
} while (!job && mActiveJobs.load(std::memory_order_relaxed) && !exitRequested());
job = steal(state);
}
if (job) {
SYSTRACE_CALL();
UTILS_UNUSED_IN_RELEASE uint32_t activeJobs = mActiveJobs.fetch_sub(1, std::memory_order_relaxed);
UTILS_UNUSED_IN_RELEASE
uint32_t activeJobs = mActiveJobs.fetch_sub(1, std::memory_order_relaxed);
assert(activeJobs); // whoops, we were already at 0
SYSTRACE_VALUE32("JobSystem::activeJobs", activeJobs - 1);
@@ -290,9 +294,9 @@ void JobSystem::loop(ThreadState* state) noexcept {
// run our main loop...
do {
if (!execute(*state)) {
std::unique_lock<Mutex> lock(mLooperLock);
while (!exitRequested() && !(mActiveJobs.load(std::memory_order_relaxed))) {
mLooperCondition.wait(lock);
std::unique_lock<Mutex> lock(mWaiterLock);
while (!exitRequested() && !hasActiveJobs()) {
wait(lock);
setThreadAffinityById(state->id);
}
}
@@ -310,16 +314,10 @@ 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,
__has_feature(thread_sanitizer) ?
std::memory_order_acq_rel :
std::memory_order_release);
auto runningJobCount = job->runningJobCount.fetch_sub(1, std::memory_order_acq_rel);
assert(runningJobCount > 0);
if (runningJobCount == 1) {
#if !__has_feature(thread_sanitizer)
std::atomic_thread_fence(std::memory_order_acquire);
#endif
// no more work, destroy this job and notify its the parent
// no more work, destroy this job and notify its parent
notify = true;
Job* const parent = job->parent == 0x7FFF ? nullptr : &storage[job->parent];
decRef(job);
@@ -332,8 +330,7 @@ void JobSystem::finish(Job* job) noexcept {
// wake-up all threads that could potentially be waiting on this job finishing
if (notify) {
{ std::lock_guard<Mutex> lock(mWaiterLock); }
mWaiterCondition.notify_all();
wake();
}
}
@@ -402,8 +399,7 @@ void JobSystem::run(JobSystem::Job*& job, uint32_t flags) noexcept {
if (!(flags & DONT_SIGNAL)) {
// wake-up multiple queues because there could be multiple jobs queued
// especially if DONT_SIGNAL was used
{ std::lock_guard<Mutex> lock(mLooperLock); }
mLooperCondition.notify_all();
wake();
}
// after run() returns, the job is virtually invalid (it'll die on its own)
@@ -426,11 +422,22 @@ void JobSystem::waitAndRelease(Job*& job) noexcept {
do {
if (!execute(state)) {
// test if job has completed first, to possibly avoid taking the lock
if (!hasJobCompleted(job)) {
std::unique_lock<Mutex> lock(mWaiterLock);
if (!hasJobCompleted(job) && !exitRequested()) {
mWaiterCondition.wait(lock);
}
if (hasJobCompleted(job)) {
break;
}
// the only way we can be here is if the job we're waiting on it being handled
// by another thread:
// - we returned from execute() which means all queues are empty
// - yet our job hasn't completed yet
// ergo, it's being run in another thread
//
// this could take time however, so we will wait with a condition, and
// continue to handle more jobs, as they get added.
std::unique_lock<Mutex> lock(mWaiterLock);
if (!hasJobCompleted(job) && !hasActiveJobs() && !exitRequested()) {
wait(lock);
}
}
} while (!hasJobCompleted(job) && !exitRequested());