CVPR/LIVE
34
1#include "parallel.h"2#include <list>3#include <thread>4#include <condition_variable>5#include <vector>6#include <cassert>7 8// From https://github.com/mmp/pbrt-v3/blob/master/src/core/parallel.cpp9 10static std::vector<std::thread> threads;11static bool shutdownThreads = false;12struct ParallelForLoop;13static ParallelForLoop *workList = nullptr;14static std::mutex workListMutex;15 16struct ParallelForLoop {17 ParallelForLoop(std::function<void(int64_t)> func1D, int64_t maxIndex, int chunkSize)18 : func1D(std::move(func1D)), maxIndex(maxIndex), chunkSize(chunkSize) {19 }20 ParallelForLoop(const std::function<void(Vector2i)> &f, const Vector2i count)21 : func2D(f), maxIndex(count[0] * count[1]), chunkSize(1) {22 nX = count[0];23 }24 25 std::function<void(int64_t)> func1D;26 std::function<void(Vector2i)> func2D;27 const int64_t maxIndex;28 const int chunkSize;29 int64_t nextIndex = 0;30 int activeWorkers = 0;31 ParallelForLoop *next = nullptr;32 int nX = -1;33 34 bool Finished() const {35 return nextIndex >= maxIndex && activeWorkers == 0;36 }37};38 39void Barrier::Wait() {40 std::unique_lock<std::mutex> lock(mutex);41 assert(count > 0);42 if (--count == 0) {43 // This is the last thread to reach the barrier; wake up all of the44 // other ones before exiting.45 cv.notify_all();46 } else {47 // Otherwise there are still threads that haven't reached it. Give48 // up the lock and wait to be notified.49 cv.wait(lock, [this] { return count == 0; });50 }51}52 53static std::condition_variable workListCondition;54 55static void worker_thread_func(const int tIndex, std::shared_ptr<Barrier> barrier) {56 ThreadIndex = tIndex;57 58 // The main thread sets up a barrier so that it can be sure that all59 // workers have called ProfilerWorkerThreadInit() before it continues60 // (and actually starts the profiling system).61 barrier->Wait();62 63 // Release our reference to the Barrier so that it's freed once all of64 // the threads have cleared it.65 barrier.reset();66 67 std::unique_lock<std::mutex> lock(workListMutex);68 while (!shutdownThreads) {69 if (!workList) {70 // Sleep until there are more tasks to run71 workListCondition.wait(lock);72 } else {73 // Get work from _workList_ and run loop iterations74 ParallelForLoop &loop = *workList;75 76 // Run a chunk of loop iterations for _loop_77 78 // Find the set of loop iterations to run next79 int64_t indexStart = loop.nextIndex;80 int64_t indexEnd = std::min(indexStart + loop.chunkSize, loop.maxIndex);81 82 // Update _loop_ to reflect iterations this thread will run83 loop.nextIndex = indexEnd;84 if (loop.nextIndex == loop.maxIndex)85 workList = loop.next;86 loop.activeWorkers++;87 88 // Run loop indices in _[indexStart, indexEnd)_89 lock.unlock();90 for (int64_t index = indexStart; index < indexEnd; ++index) {91 if (loop.func1D) {92 loop.func1D(index);93 }94 // Handle other types of loops95 else {96 assert(loop.func2D != nullptr);97 loop.func2D(Vector2i{int(index % loop.nX),98 int(index / loop.nX)});99 }100 }101 lock.lock();102 103 // Update _loop_ to reflect completion of iterations104 loop.activeWorkers--;105 if (loop.Finished()) {106 workListCondition.notify_all();107 }108 }109 }110}111 112void parallel_for_host(const std::function<void(int64_t)> &func,113 int64_t count,114 int chunkSize) {115 // Run iterations immediately if not using threads or if _count_ is small116 if (threads.empty() || count < chunkSize) {117 for (int64_t i = 0; i < count; ++i) {118 func(i);119 }120 return;121 }122 123 // Create and enqueue _ParallelForLoop_ for this loop124 ParallelForLoop loop(func, count, chunkSize);125 workListMutex.lock();126 loop.next = workList;127 workList = &loop;128 workListMutex.unlock();129 130 // Notify worker threads of work to be done131 std::unique_lock<std::mutex> lock(workListMutex);132 workListCondition.notify_all();133 134 // Help out with parallel loop iterations in the current thread135 while (!loop.Finished()) {136 // Run a chunk of loop iterations for _loop_137 138 // Find the set of loop iterations to run next139 int64_t indexStart = loop.nextIndex;140 int64_t indexEnd = std::min(indexStart + loop.chunkSize, loop.maxIndex);141 142 // Update _loop_ to reflect iterations this thread will run143 loop.nextIndex = indexEnd;144 if (loop.nextIndex == loop.maxIndex) {145 workList = loop.next;146 }147 loop.activeWorkers++;148 149 // Run loop indices in _[indexStart, indexEnd)_150 lock.unlock();151 for (int64_t index = indexStart; index < indexEnd; ++index) {152 if (loop.func1D) {153 loop.func1D(index);154 }155 // Handle other types of loops156 else {157 assert(loop.func2D != nullptr);158 loop.func2D(Vector2i{int(index % loop.nX),159 int(index / loop.nX)});160 }161 }162 lock.lock();163 164 // Update _loop_ to reflect completion of iterations165 loop.activeWorkers--;166 }167}168 169thread_local int ThreadIndex;170 171void parallel_for_host(172 std::function<void(Vector2i)> func, const Vector2i count) {173 // Launch worker threads if needed174 if (threads.empty() || count.x * count.y <= 1) {175 for (int y = 0; y < count.y; ++y) {176 for (int x = 0; x < count.x; ++x) {177 func(Vector2i{x, y});178 }179 }180 return;181 }182 183 ParallelForLoop loop(std::move(func), count);184 {185 std::lock_guard<std::mutex> lock(workListMutex);186 loop.next = workList;187 workList = &loop;188 }189 190 std::unique_lock<std::mutex> lock(workListMutex);191 workListCondition.notify_all();192 193 // Help out with parallel loop iterations in the current thread194 while (!loop.Finished()) {195 // Run a chunk of loop iterations for _loop_196 197 // Find the set of loop iterations to run next198 int64_t indexStart = loop.nextIndex;199 int64_t indexEnd = std::min(indexStart + loop.chunkSize, loop.maxIndex);200 201 // Update _loop_ to reflect iterations this thread will run202 loop.nextIndex = indexEnd;203 if (loop.nextIndex == loop.maxIndex) {204 workList = loop.next;205 }206 loop.activeWorkers++;207 208 // Run loop indices in _[indexStart, indexEnd)_209 lock.unlock();210 for (int64_t index = indexStart; index < indexEnd; ++index) {211 if (loop.func1D) {212 loop.func1D(index);213 }214 // Handle other types of loops215 else {216 assert(loop.func2D != nullptr);217 loop.func2D(Vector2i{int(index % loop.nX),218 int(index / loop.nX)});219 }220 }221 lock.lock();222 223 // Update _loop_ to reflect completion of iterations224 loop.activeWorkers--;225 }226}227 228int num_system_cores() {229 // return 1;230 int ret = std::thread::hardware_concurrency();231 if (ret == 0) {232 return 16;233 }234 return ret;235}236 237void parallel_init() {238 assert(threads.size() == 0);239 int nThreads = num_system_cores();240 ThreadIndex = 0;241 242 // Create a barrier so that we can be sure all worker threads get past243 // their call to ProfilerWorkerThreadInit() before we return from this244 // function. In turn, we can be sure that the profiling system isn't245 // started until after all worker threads have done that.246 std::shared_ptr<Barrier> barrier = std::make_shared<Barrier>(nThreads);247 248 // Launch one fewer worker thread than the total number we want doing249 // work, since the main thread helps out, too.250 for (int i = 0; i < nThreads - 1; ++i) {251 threads.push_back(std::thread(worker_thread_func, i + 1, barrier));252 }253 254 barrier->Wait();255}256 257void parallel_cleanup() {258 if (threads.empty()) {259 return;260 }261 262 {263 std::lock_guard<std::mutex> lock(workListMutex);264 shutdownThreads = true;265 workListCondition.notify_all();266 }267 268 for (std::thread &thread : threads) {269 thread.join();270 }271 threads.erase(threads.begin(), threads.end());272 shutdownThreads = false;273}274