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

improve JobSystem::parallel_for + minor optimizations

Browse Source 
- parallel_for doesn't use recursion anymore to create the "leaf"
jobs, this is now done linearly on N thread (one thread per CPU).
This uses less stack space, and reduces miss-predicted branches.

- remove almost all SYSTRACE calls because they have a huge impact
on things like parallel_for() and are misleading. They can be
enabled again by setting HEAVY_SYSTRACE to true.
This commit is contained in:
Pixelflinger
2019-07-09 16:12:15 -07:00
committed by Mathias Agopian
parent 2df639133b
commit 8170ca7cd1
3 changed files with 66 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/utils/benchmark/benchmark_JobSystem.cpp
+3 -3
View File
@@ -51,7 +51,7 @@ static void BM_JobSystemAsChildren4k(benchmark::State& state) {
for (auto _ : state) {
auto root = js.create(nullptr, &emptyJob);
for (size_t i = 0; i < 4095; i++) {
js.run(js.create(root, &emptyJob));
js.run(js.create(root, &emptyJob), JobSystem::DONT_SIGNAL);
}
js.runAndWait(root);
}
@@ -68,8 +68,8 @@ static void BM_JobSystemParallelFor(benchmark::State& state) {
{
PerformanceCounters pc(state);
for (auto _ : state) {
auto job = jobs::parallel_for(js, nullptr, 0, 4096, [](uint32_t start, uint32_t count) {
}, jobs::CountSplitter<1>());
auto job = jobs::parallel_for(js, nullptr, 0, 4096,
[](uint32_t start, uint32_t count) { }, jobs::CountSplitter<1>());
js.runAndWait(job);
}
}
libs/utils/include/utils/JobSystem.h
+43 -38
View File
@@ -374,7 +374,6 @@ private:
return !index ? nullptr : &mJobStorageBase[index - 1];
}
template <typename Mutex>
void wait(std::unique_lock<Mutex>& lock) noexcept;
void wake() noexcept;
@@ -468,20 +467,20 @@ struct ParallelForJobData {
// We first split about the number of threads we have, and only then we split the rest
// in a single thread (but execute the final cut in new jobs, see parallel() below),
// this way we save a lot of copies of JobData.
// this way we save a lot of copies of JobData and miss-predicted branches
if (splits == js.getParallelSplitCount()) {
parallel(js, parent);
return;
}
// this branch is often miss-predicted (it both sides happen 50% of the calls)
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<JobData, &JobData::parallelWithJobs>(parent, std::move(ld));
if (UTILS_UNLIKELY(l == nullptr)) {
// couldn't create a job, just pretend we're done splitting
goto done;
goto execute;
}
// start the left side before attempting the right side, so we parallelize in case
@@ -491,56 +490,62 @@ struct ParallelForJobData {
const size_type rc = count - lc;
JobData rd(start + lc, rc, splits + uint8_t(1), functor, splitter);
JobSystem::Job* r = js.createJob<JobData, &JobData::parallelWithJobs>(parent, std::move(rd));
if (UTILS_UNLIKELY(r == nullptr)) {
// couldn't allocate right side job, execute it right now
functor(start + lc, rc);
return;
start += lc;
count = rc;
goto execute;
}
// All good, execute the right side, but don't signal it,
// so it's more likely to be executed next on the same thread
js.run(r, JobSystem::DONT_SIGNAL);
} else {
done:
execute:
// we're done splitting, do the real work here!
functor(start, count);
}
}
private:
void parallel(JobSystem& js, JobSystem::Job* parent) noexcept {
// here we split the data ona single thread, and launch jobs once we're completely
// done splitting
if (splitter.split(splits, count)) {
auto lc = count / 2;
auto rc = count - lc;
auto rd = start + lc;
auto s = ++splits;
// left-side
count = lc;
parallel(js, parent);
// note: in practice the compiler is able to optimize out the call to parallel() below
// right-side
start = rd;
count = rc;
splits = s;
parallel(js, parent);
} else {
// only capture what we need
auto job = js.createJob(parent,
[f = functor, s = start, c = count](JobSystem&, JobSystem::Job*) {
// we're done splitting, do the real work here!
f(s, c);
});
if (UTILS_LIKELY(job)) {
js.run(job);
} else {
// oops, no more job available
functor(start, count);
}
// figure out how many splits we need
size_type c = count;
uint8_t s = splits;
while (splitter.split(s, c)) {
c /= 2u;
++s;
}
// then linearly create all jobs with number of elements required by the splitter
JobSystem::Job* job = nullptr;
size_type curr = start;
size_type end = start + count;
while (curr + c < end) {
job = js.createJob(parent, [f = functor, curr, c](JobSystem&, JobSystem::Job*) {
f(curr, c);
});
if (UTILS_UNLIKELY(!job)) {
goto error; // oops, no more job available
}
js.run(job, JobSystem::DONT_SIGNAL);
curr += c;
}
// and create the finishing job, which is guaranteed to have more elements that required
job = js.createJob(parent,
[f = functor, curr, c = end - curr](JobSystem&, JobSystem::Job*) {
f(curr, c);
});
if (UTILS_UNLIKELY(!job)) {
goto error; // oops, no more job available
}
js.run(job);
return;
error:
functor(curr, end - curr);
}
size_type start; // 4
libs/utils/src/JobSystem.cpp
+20 -15
View File
@@ -21,7 +21,6 @@
// when SYSTRACE_TAG_JOBSYSTEM is used, enables even heavier systraces
#define HEAVY_SYSTRACE 0
#include <utils/JobSystem.h>
#include <cmath>
@@ -55,6 +54,16 @@
# define gettid() syscall(SYS_gettid)
#endif
#if HEAVY_SYSTRACE
# define HEAVY_SYSTRACE_CALL() SYSTRACE_CALL()
# define HEAVY_SYSTRACE_NAME(name) SYSTRACE_NAME(name)
# define HEAVY_SYSTRACE_VALUE32(name, v) SYSTRACE_VALUE32(name, v)
#else
# define HEAVY_SYSTRACE_CALL()
# define HEAVY_SYSTRACE_NAME(name)
# define HEAVY_SYSTRACE_VALUE32(name, v)
#endif
namespace utils {
UTILS_DEFINE_TLS(JobSystem::ThreadState *) JobSystem::sThreadState(nullptr);
@@ -202,7 +211,6 @@ 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);
@@ -210,9 +218,10 @@ void JobSystem::wait(std::unique_lock<Mutex>& lock) noexcept {
}
void JobSystem::wake() noexcept {
mWaiterLock.lock();
Mutex& lock = mWaiterLock;
lock.lock();
const uint32_t waiterCount = mWaiterCount;
mWaiterLock.unlock();
lock.unlock();
mWaiterCondition.notify_n(waiterCount);
}
@@ -245,7 +254,7 @@ inline JobSystem::ThreadState* JobSystem::getStateToStealFrom(JobSystem::ThreadS
}
JobSystem::Job* JobSystem::steal(JobSystem::ThreadState& state) noexcept {
SYSTRACE_CALL();
HEAVY_SYSTRACE_CALL();
Job* job = nullptr;
do {
ThreadState* const stateToStealFrom = getStateToStealFrom(state);
@@ -257,7 +266,7 @@ JobSystem::Job* JobSystem::steal(JobSystem::ThreadState& state) noexcept {
}
bool JobSystem::execute(JobSystem::ThreadState& state) noexcept {
SYSTRACE_CALL();
HEAVY_SYSTRACE_CALL();
Job* job = pop(state.workQueue);
if (UTILS_UNLIKELY(job == nullptr)) {
@@ -269,10 +278,10 @@ bool JobSystem::execute(JobSystem::ThreadState& state) noexcept {
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);
HEAVY_SYSTRACE_VALUE32("JobSystem::activeJobs", activeJobs - 1);
if (UTILS_LIKELY(job->function)) {
SYSTRACE_NAME("job->function");
HEAVY_SYSTRACE_NAME("job->function");
job->function(job->storage, *this, job);
}
finish(job);
@@ -305,7 +314,7 @@ void JobSystem::loop(ThreadState* state) noexcept {
UTILS_NOINLINE
void JobSystem::finish(Job* job) noexcept {
SYSTRACE_CALL();
HEAVY_SYSTRACE_CALL();
bool notify = false;
@@ -378,11 +387,7 @@ void JobSystem::release(JobSystem::Job*& job) noexcept {
}
void JobSystem::run(JobSystem::Job*& job, uint32_t flags) noexcept {
#if HEAVY_SYSTRACE
SYSTRACE_CALL();
#else
SYSTRACE_CONTEXT();
#endif
HEAVY_SYSTRACE_CALL();
ThreadState& state(getState());
@@ -393,7 +398,7 @@ void JobSystem::run(JobSystem::Job*& job, uint32_t flags) noexcept {
put(state.workQueue, job);
SYSTRACE_VALUE32("JobSystem::activeJobs", activeJobs + 1);
HEAVY_SYSTRACE_VALUE32("JobSystem::activeJobs", activeJobs + 1);
// wake-up a thread if needed...
if (!(flags & DONT_SIGNAL)) {