Branch data Line data Source code
1 : : // Copyright (c) The Bitcoin Core developers
2 : : // Distributed under the MIT software license, see the accompanying
3 : : // file COPYING or http://www.opensource.org/licenses/mit-license.php.
4 : :
5 : : #include <common/system.h>
6 : : #include <logging.h>
7 : : #include <random.h>
8 : : #include <util/string.h>
9 : : #include <util/threadpool.h>
10 : : #include <util/time.h>
11 : :
12 : : #include <boost/test/unit_test.hpp>
13 : :
14 : : // General test values
15 : : int NUM_WORKERS_DEFAULT = 0;
16 : : constexpr char POOL_NAME[] = "test";
17 : : constexpr auto WAIT_TIMEOUT = 120s;
18 : :
19 : : struct ThreadPoolFixture {
20 : 11 : ThreadPoolFixture() {
21 [ + - ]: 11 : NUM_WORKERS_DEFAULT = FastRandomContext().randrange(GetNumCores()) + 1;
22 : 11 : LogInfo("thread pool workers count: %d", NUM_WORKERS_DEFAULT);
23 : 11 : }
24 : : };
25 : :
26 : : // Test Cases Overview
27 : : // 0) Submit task to a non-started pool.
28 : : // 1) Submit tasks and verify completion.
29 : : // 2) Maintain all threads busy except one.
30 : : // 3) Wait for work to finish.
31 : : // 4) Wait for result object.
32 : : // 5) The task throws an exception, catch must be done in the consumer side.
33 : : // 6) Busy workers, help them by processing tasks externally.
34 : : // 7) Recursive submission of tasks.
35 : : // 8) Submit task when all threads are busy, stop pool and verify task gets executed.
36 : : // 9) Congestion test; create more workers than available cores.
37 : : // 10) Ensure Interrupt() prevents further submissions.
38 : : BOOST_FIXTURE_TEST_SUITE(threadpool_tests, ThreadPoolFixture)
39 : :
40 : : #define WAIT_FOR(futures) \
41 : : do { \
42 : : for (const auto& f : futures) { \
43 : : BOOST_REQUIRE(f.wait_for(WAIT_TIMEOUT) == std::future_status::ready); \
44 : : } \
45 : : } while (0)
46 : :
47 : : // Helper to unwrap a valid pool submission
48 : : template <typename F>
49 : 332 : [[nodiscard]] auto Submit(ThreadPool& pool, F&& fn)
50 : : {
51 [ - + ]: 332 : return std::move(*Assert(pool.Submit(std::forward<F>(fn))));
52 : : }
53 : :
54 : : // Block a number of worker threads by submitting tasks that wait on `blocker_future`.
55 : : // Returns the futures of the blocking tasks, ensuring all have started and are waiting.
56 : 4 : std::vector<std::future<void>> BlockWorkers(ThreadPool& threadPool, const std::shared_future<void>& blocker_future, int num_of_threads_to_block)
57 : : {
58 : : // Per-thread ready promises to ensure all workers are actually blocked
59 : 4 : std::vector<std::promise<void>> ready_promises(num_of_threads_to_block);
60 : 4 : std::vector<std::future<void>> ready_futures;
61 [ + - ]: 4 : ready_futures.reserve(num_of_threads_to_block);
62 [ + - + - : 74 : for (auto& p : ready_promises) ready_futures.emplace_back(p.get_future());
+ + ]
63 : :
64 : : // Fill all workers with blocking tasks
65 : 4 : std::vector<std::future<void>> blocking_tasks;
66 [ + + ]: 39 : for (int i = 0; i < num_of_threads_to_block; i++) {
67 [ + - ]: 35 : std::promise<void>& ready = ready_promises[i];
68 [ + - + - : 140 : blocking_tasks.emplace_back(Submit(threadPool, [blocker_future, &ready]() {
+ - + - -
- ]
69 : 35 : ready.set_value();
70 : 35 : blocker_future.wait();
71 : 35 : }));
72 : : }
73 : :
74 : : // Wait until all threads are actually blocked
75 [ + - + - : 39 : WAIT_FOR(ready_futures);
+ - + + ]
76 : 4 : return blocking_tasks;
77 : 4 : }
78 : :
79 : : // Test 0, submit task to a non-started pool
80 [ + - + - : 7 : BOOST_AUTO_TEST_CASE(submit_task_before_start_fails)
+ - + - -
+ + - + -
+ - + - +
- - + + -
+ - + - +
- + - - +
+ - + - +
- + - + -
- + + - +
- + - + -
+ - - + +
- ]
81 : : {
82 [ + - ]: 1 : ThreadPool threadPool(POOL_NAME);
83 : 1 : auto res = threadPool.Submit([]{ return false; });
84 [ + - + - : 2 : BOOST_CHECK(!res);
+ - ]
85 [ + - + - ]: 1 : BOOST_CHECK_EQUAL(SubmitErrorString(res.error()), "No active workers");
86 : 1 : }
87 : :
88 : : // Test 1, submit tasks and verify completion
89 [ + - + - : 7 : BOOST_AUTO_TEST_CASE(submit_tasks_complete_successfully)
+ - + - -
+ + - + -
+ - + - +
- - + + -
+ - + - +
- + - - +
+ - + - +
- + - + -
- + + - +
- + - + -
+ - - + +
- ]
90 : : {
91 : 1 : int num_tasks = 50;
92 : :
93 [ + - ]: 1 : ThreadPool threadPool(POOL_NAME);
94 [ + - ]: 1 : threadPool.Start(NUM_WORKERS_DEFAULT);
95 : 1 : std::atomic<int> counter = 0;
96 : :
97 : : // Store futures to ensure completion before checking counter.
98 : 1 : std::vector<std::future<void>> futures;
99 [ + - ]: 1 : futures.reserve(num_tasks);
100 [ + + ]: 51 : for (int i = 1; i <= num_tasks; i++) {
101 [ + - + - ]: 150 : futures.emplace_back(Submit(threadPool, [&counter, i]() {
102 : 50 : counter.fetch_add(i, std::memory_order_relaxed);
103 : : }));
104 : : }
105 : :
106 : : // Wait for all tasks to finish
107 [ + - + - : 51 : WAIT_FOR(futures);
+ - + + ]
108 : 1 : int expected_value = (num_tasks * (num_tasks + 1)) / 2; // Gauss sum.
109 [ + - + - ]: 1 : BOOST_CHECK_EQUAL(counter.load(), expected_value);
110 [ + - + - : 1 : BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 0);
+ - ]
111 : 1 : }
112 : :
113 : : // Test 2, maintain all threads busy except one
114 [ + - + - : 7 : BOOST_AUTO_TEST_CASE(single_available_worker_executes_all_tasks)
+ - + - -
+ + - + -
+ - + - +
- - + + -
+ - + - +
- + - - +
+ - + - +
- + - + -
- + + - +
- + - + -
+ - - + +
- ]
115 : : {
116 [ + - ]: 1 : ThreadPool threadPool(POOL_NAME);
117 [ + - ]: 1 : threadPool.Start(NUM_WORKERS_DEFAULT);
118 : : // Single blocking future for all threads
119 [ + - ]: 1 : std::promise<void> blocker;
120 [ + - + - ]: 2 : std::shared_future<void> blocker_future(blocker.get_future());
121 [ + - ]: 1 : const auto blocking_tasks = BlockWorkers(threadPool, blocker_future, NUM_WORKERS_DEFAULT - 1);
122 : :
123 : : // Now execute tasks on the single available worker
124 : : // and check that all the tasks are executed.
125 : 1 : int num_tasks = 15;
126 : 1 : int counter = 0;
127 : :
128 : : // Store futures to wait on
129 [ + - ]: 1 : std::vector<std::future<void>> futures(num_tasks);
130 [ + - - + : 31 : for (auto& f : futures) f = Submit(threadPool, [&counter]{ counter++; });
+ + ]
131 : :
132 [ + - + - : 16 : WAIT_FOR(futures);
+ - + + ]
133 [ + - + - ]: 1 : BOOST_CHECK_EQUAL(counter, num_tasks);
134 : :
135 [ + - ]: 1 : blocker.set_value();
136 [ + - + - : 10 : WAIT_FOR(blocking_tasks);
+ - + + ]
137 [ + - ]: 1 : threadPool.Stop();
138 [ + - + - : 1 : BOOST_CHECK_EQUAL(threadPool.WorkersCount(), 0);
+ - ]
139 [ + - ]: 2 : }
140 : :
141 : : // Test 3, wait for work to finish
142 [ + - + - : 7 : BOOST_AUTO_TEST_CASE(wait_for_task_to_finish)
+ - + - -
+ + - + -
+ - + - +
- - + + -
+ - + - +
- + - - +
+ - + - +
- + - + -
- + + - +
- + - + -
+ - - + +
- ]
143 : : {
144 [ + - ]: 1 : ThreadPool threadPool(POOL_NAME);
145 [ + - ]: 1 : threadPool.Start(NUM_WORKERS_DEFAULT);
146 : 1 : std::atomic<bool> flag = false;
147 : 2 : std::future<void> future = Submit(threadPool, [&flag]() {
148 : 1 : UninterruptibleSleep(200ms);
149 : 1 : flag.store(true, std::memory_order_release);
150 [ + - ]: 2 : });
151 [ + - + - : 2 : BOOST_CHECK(future.wait_for(WAIT_TIMEOUT) == std::future_status::ready);
+ - + - ]
152 [ + - + - : 2 : BOOST_CHECK(flag.load(std::memory_order_acquire));
+ - ]
153 : 1 : }
154 : :
155 : : // Test 4, obtain result object
156 [ + - + - : 7 : BOOST_AUTO_TEST_CASE(get_result_from_completed_task)
+ - + - -
+ + - + -
+ - + - +
- - + + -
+ - + - +
- + - - +
+ - + - +
- + - + -
- + + - +
- + - + -
+ - - + +
- ]
157 : : {
158 [ + - ]: 1 : ThreadPool threadPool(POOL_NAME);
159 [ + - ]: 1 : threadPool.Start(NUM_WORKERS_DEFAULT);
160 [ + - ]: 1 : std::future<bool> future_bool = Submit(threadPool, []() { return true; });
161 [ + - + - : 2 : BOOST_CHECK(future_bool.get());
+ - + - ]
162 : :
163 [ + - + - ]: 2 : std::future<std::string> future_str = Submit(threadPool, []() { return std::string("true"); });
164 [ + - ]: 1 : std::string result = future_str.get();
165 [ + - + - ]: 1 : BOOST_CHECK_EQUAL(result, "true");
166 [ - + - + ]: 1 : }
167 : :
168 : : // Test 5, throw exception and catch it on the consumer side
169 [ + - + - : 7 : BOOST_AUTO_TEST_CASE(task_exception_propagates_to_future)
+ - + - -
+ + - + -
+ - + - +
- - + + -
+ - + - +
- + - - +
+ - + - +
- + - + -
- + + - +
- + - + -
+ - - + +
- ]
170 : : {
171 [ + - ]: 1 : ThreadPool threadPool(POOL_NAME);
172 [ + - ]: 1 : threadPool.Start(NUM_WORKERS_DEFAULT);
173 : :
174 : 1 : int num_tasks = 5;
175 [ + - ]: 1 : std::string err_msg{"something wrong happened"};
176 : 1 : std::vector<std::future<void>> futures;
177 [ + - ]: 1 : futures.reserve(num_tasks);
178 [ + + ]: 6 : for (int i = 0; i < num_tasks; i++) {
179 [ - + + - : 20 : futures.emplace_back(Submit(threadPool, [err_msg, i]() {
+ - ]
180 [ + - + - : 10 : throw std::runtime_error(err_msg + util::ToString(i));
+ - ]
181 : : }));
182 : : }
183 : :
184 [ + + ]: 6 : for (int i = 0; i < num_tasks; i++) {
185 [ + - + - : 15 : BOOST_CHECK_EXCEPTION(futures.at(i).get(), std::runtime_error, [&](const std::runtime_error& e) {
- + - - -
- - + + -
+ - + - ]
186 : : BOOST_CHECK_EQUAL(e.what(), err_msg + util::ToString(i));
187 : : return true;
188 : : });
189 : : }
190 : 1 : }
191 : :
192 : : // Test 6, all workers are busy, help them by processing tasks from outside
193 [ + - + - : 7 : BOOST_AUTO_TEST_CASE(process_tasks_manually_when_workers_busy)
+ - + - -
+ + - + -
+ - + - +
- - + + -
+ - + - +
- + - - +
+ - + - +
- + - + -
- + + - +
- + - + -
+ - - + +
- ]
194 : : {
195 [ + - ]: 1 : ThreadPool threadPool(POOL_NAME);
196 [ + - ]: 1 : threadPool.Start(NUM_WORKERS_DEFAULT);
197 : :
198 [ + - ]: 1 : std::promise<void> blocker;
199 [ + - + - ]: 2 : std::shared_future<void> blocker_future(blocker.get_future());
200 [ + - ]: 1 : const auto& blocking_tasks = BlockWorkers(threadPool, blocker_future, NUM_WORKERS_DEFAULT);
201 : :
202 : : // Now submit tasks and check that none of them are executed.
203 : 1 : int num_tasks = 20;
204 : 1 : std::atomic<int> counter = 0;
205 [ + + ]: 21 : for (int i = 0; i < num_tasks; i++) {
206 [ + - ]: 60 : (void)Submit(threadPool, [&counter]() {
207 : 20 : counter.fetch_add(1, std::memory_order_relaxed);
208 : : });
209 : : }
210 [ + - ]: 1 : UninterruptibleSleep(100ms);
211 [ + - + - : 1 : BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), num_tasks);
+ - ]
212 : :
213 : : // Now process manually
214 [ + + ]: 21 : for (int i = 0; i < num_tasks; i++) {
215 [ + - ]: 20 : threadPool.ProcessTask();
216 : : }
217 [ + - + - ]: 1 : BOOST_CHECK_EQUAL(counter.load(), num_tasks);
218 [ + - + - : 1 : BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 0);
+ - ]
219 [ + - ]: 1 : blocker.set_value();
220 [ + - ]: 1 : threadPool.Stop();
221 [ + - + - : 17 : WAIT_FOR(blocking_tasks);
+ - + + ]
222 [ + - ]: 2 : }
223 : :
224 : : // Test 7, submit tasks from other tasks
225 [ + - + - : 7 : BOOST_AUTO_TEST_CASE(recursive_task_submission)
+ - + - -
+ + - + -
+ - + - +
- - + + -
+ - + - +
- + - - +
+ - + - +
- + - + -
- + + - +
- + - + -
+ - - + +
- ]
226 : : {
227 [ + - ]: 1 : ThreadPool threadPool(POOL_NAME);
228 [ + - ]: 1 : threadPool.Start(NUM_WORKERS_DEFAULT);
229 : :
230 [ + - ]: 1 : std::promise<void> signal;
231 [ + - ]: 2 : (void)Submit(threadPool, [&]() {
232 [ + - ]: 1 : (void)Submit(threadPool, [&]() {
233 [ - - + - ]: 1 : signal.set_value();
234 : : });
235 : 1 : });
236 : :
237 [ + - + - ]: 2 : signal.get_future().wait();
238 [ + - ]: 1 : threadPool.Stop();
239 : 1 : }
240 : :
241 : : // Test 8, submit task when all threads are busy and then stop the pool
242 [ + - + - : 7 : BOOST_AUTO_TEST_CASE(task_submitted_while_busy_completes)
+ - + - -
+ + - + -
+ - + - +
- - + + -
+ - + - +
- + - - +
+ - + - +
- + - + -
- + + - +
- + - + -
+ - - + +
- ]
243 : : {
244 [ + - ]: 1 : ThreadPool threadPool(POOL_NAME);
245 [ + - ]: 1 : threadPool.Start(NUM_WORKERS_DEFAULT);
246 : :
247 [ + - ]: 1 : std::promise<void> blocker;
248 [ + - + - ]: 2 : std::shared_future<void> blocker_future(blocker.get_future());
249 [ + - ]: 1 : const auto& blocking_tasks = BlockWorkers(threadPool, blocker_future, NUM_WORKERS_DEFAULT);
250 : :
251 : : // Submit an extra task that should execute once a worker is free
252 [ + - ]: 1 : std::future<bool> future = Submit(threadPool, []() { return true; });
253 : :
254 : : // At this point, all workers are blocked, and the extra task is queued
255 [ + - + - : 1 : BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 1);
+ - ]
256 : :
257 : : // Wait a short moment before unblocking the threads to mimic a concurrent shutdown
258 : 2 : std::thread thread_unblocker([&blocker]() {
259 : 1 : UninterruptibleSleep(300ms);
260 : 1 : blocker.set_value();
261 [ + - ]: 2 : });
262 : :
263 : : // Stop the pool while the workers are still blocked
264 [ + - ]: 1 : threadPool.Stop();
265 : :
266 : : // Expect the submitted task to complete
267 [ + - + - : 2 : BOOST_CHECK(future.get());
+ - + - ]
268 [ + - ]: 1 : thread_unblocker.join();
269 : :
270 : : // Obviously all the previously blocking tasks should be completed at this point too
271 [ + - + - : 10 : WAIT_FOR(blocking_tasks);
+ - + + ]
272 : :
273 : : // Pool should be stopped and no workers remaining
274 [ + - + - : 1 : BOOST_CHECK_EQUAL(threadPool.WorkersCount(), 0);
+ - ]
275 [ - + + - ]: 2 : }
276 : :
277 : : // Test 9, more workers than available cores (congestion test)
278 [ + - + - : 7 : BOOST_AUTO_TEST_CASE(congestion_more_workers_than_cores)
+ - + - -
+ + - + -
+ - + - +
- - + + -
+ - + - +
- + - - +
+ - + - +
- + - + -
- + + - +
- + - + -
+ - - + +
- ]
279 : : {
280 [ + - ]: 1 : ThreadPool threadPool(POOL_NAME);
281 [ + - - + : 1 : threadPool.Start(std::max(1, GetNumCores() * 2)); // Oversubscribe by 2×
+ - ]
282 : :
283 : 1 : int num_tasks = 200;
284 : 1 : std::atomic<int> counter{0};
285 : :
286 : 1 : std::vector<std::future<void>> futures;
287 [ + - ]: 1 : futures.reserve(num_tasks);
288 [ + + ]: 201 : for (int i = 0; i < num_tasks; i++) {
289 [ + - + - ]: 600 : futures.emplace_back(Submit(threadPool, [&counter] {
290 : 200 : counter.fetch_add(1, std::memory_order_relaxed);
291 : : }));
292 : : }
293 : :
294 [ + - + - : 201 : WAIT_FOR(futures);
+ - + + ]
295 [ + - + - ]: 1 : BOOST_CHECK_EQUAL(counter.load(), num_tasks);
296 : 1 : }
297 : :
298 : : // Test 10, Interrupt() prevents further submissions
299 [ + - + - : 7 : BOOST_AUTO_TEST_CASE(interrupt_blocks_new_submissions)
+ - + - -
+ + - + -
+ - + - +
- - + + -
+ - + - +
- + - - +
+ - + - +
- + - + -
- + + - +
- + - + -
+ - - + +
- ]
300 : : {
301 : : // 1) Interrupt from main thread
302 [ + - ]: 1 : ThreadPool threadPool(POOL_NAME);
303 [ + - ]: 1 : threadPool.Start(NUM_WORKERS_DEFAULT);
304 [ + - ]: 1 : threadPool.Interrupt();
305 : :
306 : 1 : auto res = threadPool.Submit([]{});
307 [ + - + - : 2 : BOOST_CHECK(!res);
+ - ]
308 [ + - + - ]: 1 : BOOST_CHECK_EQUAL(SubmitErrorString(res.error()), "Interrupted");
309 : :
310 : : // Reset pool
311 [ + - ]: 1 : threadPool.Stop();
312 : :
313 : : // 2) Interrupt() from a worker thread
314 : : // One worker is blocked, another calls Interrupt(), and the remaining one waits for tasks.
315 [ + - ]: 1 : threadPool.Start(/*num_workers=*/3);
316 : 1 : std::atomic<int> counter{0};
317 [ + - ]: 1 : std::promise<void> blocker;
318 [ + - + - : 3 : const auto blocking_tasks = BlockWorkers(threadPool, blocker.get_future().share(), 1);
- + ]
319 [ + - ]: 2 : Submit(threadPool, [&threadPool, &counter]{
320 : 1 : threadPool.Interrupt();
321 : 1 : counter.fetch_add(1, std::memory_order_relaxed);
322 [ + - ]: 2 : }).get();
323 [ + - ]: 1 : blocker.set_value(); // unblock worker
324 : :
325 [ + - + - ]: 1 : BOOST_CHECK_EQUAL(counter.load(), 1);
326 [ + - ]: 1 : threadPool.Stop();
327 [ + - + - : 2 : WAIT_FOR(blocking_tasks);
+ - + + ]
328 [ + - + - : 1 : BOOST_CHECK_EQUAL(threadPool.WorkersCount(), 0);
+ - ]
329 : 1 : }
330 : :
331 : : BOOST_AUTO_TEST_SUITE_END()
|
