util/parallel.cpp - TerrainGen coverage

util/parallel.cpp

Line	Branch	Exec	Source
1			#include "terraingen/util/parallel.h"
2
3			#include <algorithm>
4			#include <atomic>
5			#include <condition_variable>
6			#include <deque>
7			#include <exception>
8			#include <memory>
9			#include <mutex>
10			#include <thread>
11			#include <vector>
12
13			namespace terraingen {
14
15			namespace {
16
17			/// One `ParallelFor` call: its tasks, handed out in order to the caller and its helpers.
18			struct Job {
19			std::size_t count = 0;
20			const std::function<void(std::size_t)>* task = nullptr;
21			std::atomic<std::size_t> next{0};
22			std::atomic<int> slots{0};
23			std::atomic<bool> failed{false};
24			std::mutex mutex;
25			std::condition_variable idle;
26			int active = 0;
27			std::exception_ptr error;
28
29			/// Runs tasks until none are left or one has failed.
30		2940	void Work() {
31	2/2 ✓ Branch 1 taken 205328 times. ✓ Branch 2 taken 3 times.	208271	while (!failed.load()) {
32		205328	const std::size_t i = next.fetch_add(1);
33	2/2 ✓ Branch 0 taken 2937 times. ✓ Branch 1 taken 202391 times.	205328	if (i >= count) {
34		2937	return;
35			}
36			try {
37	1/1 ✓ Branch 1 taken 202389 times.	202391	(*task)(i);
38		2	} catch (...) {
39	1/1 ✓ Branch 1 taken 2 times.	2	const std::lock_guard lock(mutex);
40	1/2 ✓ Branch 1 taken 2 times. ✗ Branch 2 not taken.	2	if (!error) {
41		2	error = std::current_exception();
42			}
43		2	failed.store(true);
44		2	}
45			}
46			}
47
48			/// Whether helpers have nothing left to take.
49		2936	bool Exhausted() const {
50	4/4 ✓ Branch 1 taken 2934 times. ✓ Branch 2 taken 2 times. ✓ Branch 4 taken 965 times. ✓ Branch 5 taken 1969 times.	5870	return failed.load() \|\| next.load() >= count;
51			}
52			};
53
54			/// Threads kept for the life of the program, helping whichever jobs are waiting. The pool is
55			/// never destroyed: joining threads while statics are torn down can deadlock, as when a library is
56			/// unloaded on Windows, and the threads end with the process.
57			class Pool {
58			public:
59	1/1 ✓ Branch 3 taken 218 times.	218	Pool() {
60		218	const unsigned size = std::max(1u, std::thread::hardware_concurrency()) - 1;
61	2/2 ✓ Branch 0 taken 654 times. ✓ Branch 1 taken 218 times.	872	for (unsigned t = 0; t < size; ++t) {
62	2/2 ✓ Branch 1 taken 654 times. ✓ Branch 4 taken 654 times.	1296	std::thread([this] { Loop(); }).detach();
63			}
64		218	}
65
66			Pool(const Pool&) = delete;
67			Pool& operator=(const Pool&) = delete;
68
69			/// Offers `job` to the pool's threads.
70		1040	void Post(const std::shared_ptr<Job>& job) {
71			{
72	1/1 ✓ Branch 1 taken 1040 times.	1040	const std::lock_guard lock(mutex_);
73	1/1 ✓ Branch 1 taken 1040 times.	1040	jobs_.push_back(job);
74		1040	}
75		1040	wake_.notify_all();
76		1040	}
77
78			private:
79		642	void Loop() {
80	1/1 ✓ Branch 1 taken 642 times.	642	std::unique_lock lock(mutex_);
81			while (true) {
82	1/1 ✓ Branch 1 taken 2936 times.	9130	wake_.wait(lock, [this] { return !jobs_.empty(); });
83		2936	std::shared_ptr<Job> job = jobs_.front();
84	6/6 ✓ Branch 2 taken 1969 times. ✓ Branch 3 taken 967 times. ✓ Branch 5 taken 69 times. ✓ Branch 6 taken 1900 times. ✓ Branch 7 taken 1036 times. ✓ Branch 8 taken 1900 times.	4905	if (job->Exhausted() \|\| job->slots.fetch_sub(1) <= 0) {
85		1036	jobs_.pop_front();
86		1036	continue;
87			}
88	1/1 ✓ Branch 1 taken 1900 times.	1900	lock.unlock();
89			{
90	1/1 ✓ Branch 2 taken 1900 times.	1900	const std::lock_guard jobLock(job->mutex);
91		1900	++job->active;
92		1900	}
93	1/1 ✓ Branch 2 taken 1900 times.	1900	job->Work();
94			{
95	1/1 ✓ Branch 2 taken 1900 times.	1900	const std::lock_guard jobLock(job->mutex);
96		1900	--job->active;
97		1900	}
98		1900	job->idle.notify_all();
99	1/1 ✓ Branch 1 taken 1895 times.	1895	lock.lock();
100	2/2 ✓ Branch 1 taken 1895 times. ✓ Branch 2 taken 1036 times.	5862	}
101			}
102
103			std::mutex mutex_;
104			std::condition_variable wake_;
105			std::deque<std::shared_ptr<Job>> jobs_;
106			};
107
108		1040	Pool& SharedPool() {
109	5/10 ✓ Branch 0 taken 218 times. ✓ Branch 1 taken 822 times. ✓ Branch 3 taken 218 times. ✗ Branch 4 not taken. ✓ Branch 6 taken 218 times. ✓ Branch 9 taken 218 times. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 15 not taken. ✗ Branch 16 not taken.	1040	static Pool* const pool = new Pool;
110		1040	return *pool;
111			}
112
113			} // namespace
114
115		149837	unsigned ResolveThreads(unsigned threads) {
116	2/2 ✓ Branch 0 taken 103331 times. ✓ Branch 1 taken 46506 times.	149837	if (threads > 0) {
117		103331	return threads;
118			}
119		46506	return std::max(1u, std::thread::hardware_concurrency());
120			}
121
122		149835	void ParallelFor(std::size_t count, unsigned threads,
123			const std::function<void(std::size_t index)>& task) {
124	1/1 ✓ Branch 1 taken 149835 times.	149835	const std::size_t workers = std::min<std::size_t>(ResolveThreads(threads), count);
125	2/2 ✓ Branch 0 taken 148795 times. ✓ Branch 1 taken 1040 times.	149835	if (workers <= 1) {
126	2/2 ✓ Branch 0 taken 162297 times. ✓ Branch 1 taken 148794 times.	311091	for (std::size_t i = 0; i < count; ++i) {
127	1/1 ✓ Branch 1 taken 162296 times.	162297	task(i);
128			}
129		148794	return;
130			}
131	1/1 ✓ Branch 1 taken 1040 times.	1040	auto job = std::make_shared<Job>();
132		1040	job->count = count;
133		1040	job->task = &task;
134		1040	job->slots.store(static_cast<int>(workers - 1));
135	2/2 ✓ Branch 1 taken 1040 times. ✓ Branch 4 taken 1040 times.	1040	SharedPool().Post(job);
136	1/1 ✓ Branch 2 taken 1040 times.	1040	job->Work();
137			{
138	1/1 ✓ Branch 2 taken 1040 times.	1040	std::unique_lock lock(job->mutex);
139	1/1 ✓ Branch 2 taken 1040 times.	2884	job->idle.wait(lock, [&] { return job->active == 0; });
140		1040	}
141		1040	job->slots.store(0);
142	2/2 ✓ Branch 2 taken 2 times. ✓ Branch 3 taken 1038 times.	1040	if (job->error) {
143		4	std::rethrow_exception(job->error);
144			}
145		1040	}
146
147			} // namespace terraingen
148