http: replace WorkQueue and single threads handling for ThreadPool #33689

 0-----BEGIN PGP SIGNED MESSAGE-----
 1Hash: SHA256
 2
 3ACK 195a96258f970c384ce180d57e73616904ef5fa1
 4-----BEGIN PGP SIGNATURE-----
 5
 6iQIzBAEBCAAdFiEE5hdzzW4BBA4vG9eM5+KYS2KJyToFAmkCMdwACgkQ5+KYS2KJ
 7yTriXA//RRnzezUHdmzRlKmoSDA+ZBHz0RY3z2LGk63izb/YdYaJY9JBZL2Y9BA8
 8K2nyexdSDC/DFOm4H56ddEe6ChlB7w+uZc92SgFSLSvavInpZ80KEJRk07vgoIL7
 9hwuyyevWyOOU32iz1NE3q316TMaJmzVsPhRGwbdmTXNwJLtUX6g4czfh28ajW1DC
10Y9ULKwT36rFHRcKwC1YZYuBJUNBZWQgVBcydcmS1UEykY4mBnCW9knrATwn/29b7
112AYPV+yuaiy9OpDEOJ8iKtZOPGR36NrIUleMUqruq4Sy2/TnJtm3AKNK0336/Fxu
12MqVyPKSyusg7kBA6f2h/2+NHpbyLoboYhjZew+HvED/aFfi+Jla+nxybkUYXfciL
13pzbND22TTuRGB1fKU7AwPD0TO9JwOTU385iEdpoGq8rbT3EpgPr31N4TeDQDJx5t
14jPzzWZYj43JMuIc3bm/K5S2HYSdFZUEDQC81kbND+jOLF6YkJwS9794anLO38tvi
15fip/eLK8Nw4pmWnW63/9lc+Y/1gLpgLDMxxhA1NKJyytk7z2IRo7vJKck6TqAjcZ
16nU8Wv9/ful5ndDJfLIKuYT9jqRk9ORohVwv+P+ppuO8jhFjhuswxPlFJKkMXTeZn
17hGi8QCrAUuibvVuLfLKVExJqSmmeUAkTVKp5ZTWLwNB7IkZrSoo=
18=hZYm
19-----END PGP SIGNATURE-----

 0-----BEGIN PGP SIGNED MESSAGE-----
 1Hash: SHA256
 2
 3ACK f5eca8d5252a68f0fbc8b5efec021c75744555b6
 4-----BEGIN PGP SIGNATURE-----
 5
 6iQIzBAEBCAAdFiEE5hdzzW4BBA4vG9eM5+KYS2KJyToFAmkCWuYACgkQ5+KYS2KJ
 7yTpb9RAAgxNlDQlqkVfnIK2Xt8BQmRsiG+1KHQGJA2d3RHoxjFg2129fj47y5i+t
 8GlatlcsXEcibI2D0F22sKSzY1u0LWy4GYLI1d12JAIewA99n/lRr1ktDk3v64pp8
 9kldvYpd8UBs7DCHSJhPs4HbOPgwIILPASdSbQTb+6m48X5jg+cu+yMBuANVq3sbB
109I8rUlbUgRva5voy3EGRkXsGTuwcaCoLDNHnnjrqQkJMYTym47rMF/xTZJJ11isF
11QrpzFu+P2tFy1oj0bGd4e5EhqNk++qfRCuw9sGLgLL6YEHzC/ihVH/L5NAxoeJX4
12L0yX09BG5wDrbUQvZn4w1uaQz39Uor5mb8+tZyDUyRmO9MrG5AHomUtfdbiYLwSO
13007bLD+WX4Hxode47xCdhUhqflXqbHD2mhkf5ESyUgGu3smSHSQQosuzIJiVmTn9
14b2UJG6see/tckFvart6YLn1AIu9uCyUMmB+hZIcSasZv95oEHv1YB3E5dcKUmq0d
15A1cUHIhNvU4i/hNy2xgijgSjDdpVQMFbLYUt9y4ZnzpJXFd7mnbpHVUVM1P+Onvp
16ScAtgvosXfbd5PjnAQWsMWgSmVHUtSc4t9GPjGlRYaVpHFkEbqWpWeKrYqMkU3id
179YiFP9BueWNbNV7OBlB0LcfCpwO2WbvfJOW93/jxDovc0tgohbU=
18=c8aD
19-----END PGP SIGNATURE-----

  0diff --git a/src/test/threadpool_tests.cpp b/src/test/threadpool_tests.cpp
  1index e8200533cd..052784db37 100644
  2--- a/src/test/threadpool_tests.cpp
  3+++ b/src/test/threadpool_tests.cpp
  4@@ -2,270 +2,208 @@
  5 // Distributed under the MIT software license, see the accompanying
  6 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
  7 
  8+#include <common/system.h>
  9+#include <test/util/setup_common.h>
 10 #include <util/string.h>
 11 #include <util/threadpool.h>
 12+#include <util/time.h>
 13 
 14 #include <boost/test/unit_test.hpp>
 15 
 16-BOOST_AUTO_TEST_SUITE(threadpool_tests)
 17+#include <latch>
 18+#include <semaphore>
 19 
 20-constexpr auto TIMEOUT_SECS = std::chrono::seconds(120);
 21+using namespace std::chrono;
 22+constexpr auto TIMEOUT = seconds(120);
 23 
 24-template <typename T>
 25-void WaitFor(const std::vector<std::future<T>>& futures, const std::string& context)
 26+void WaitFor(std::span<const std::future<void>> futures)
 27 {
 28-    for (size_t i = 0; i < futures.size(); ++i) {
 29-        if (futures[i].wait_for(TIMEOUT_SECS) != std::future_status::ready) {
 30-            throw std::runtime_error("Timeout waiting for: " + context + ", task index " + util::ToString(i));
 31-        }
 32+    for (const auto& f : futures) {
 33+        BOOST_REQUIRE(f.wait_for(TIMEOUT) == std::future_status::ready);
 34     }
 35 }
 36 
 37 // Block a number of worker threads by submitting tasks that wait on `blocker_future`.
 38 // Returns the futures of the blocking tasks, ensuring all have started and are waiting.
 39-std::vector<std::future<void>> BlockWorkers(ThreadPool& threadPool, std::shared_future<void>& blocker_future, int num_of_threads_to_block, const std::string& context)
 40+std::vector<std::future<void>> BlockWorkers(ThreadPool& threadPool, std::counting_semaphore<>& release_sem, size_t num_of_threads_to_block)
 41 {
 42-    // Per-thread ready promises to ensure all workers are actually blocked
 43-    std::vector<std::promise<void>> ready_promises(num_of_threads_to_block);
 44-    std::vector<std::future<void>> ready_futures;
 45-    ready_futures.reserve(num_of_threads_to_block);
 46-    for (auto& p : ready_promises) ready_futures.emplace_back(p.get_future());
 47-
 48-    // Fill all workers with blocking tasks
 49-    std::vector<std::future<void>> blocking_tasks;
 50-    for (int i = 0; i < num_of_threads_to_block; i++) {
 51-        std::promise<void>& ready = ready_promises[i];
 52-        blocking_tasks.emplace_back(threadPool.Submit([blocker_future, &ready]() {
 53-            ready.set_value();
 54-            blocker_future.wait();
 55-        }));
 56-    }
 57+    assert(threadPool.WorkersCount() >= num_of_threads_to_block);
 58+    std::latch ready{std::ptrdiff_t(num_of_threads_to_block)};
 59 
 60-    // Wait until all threads are actually blocked
 61-    WaitFor(ready_futures, context);
 62+    std::vector<std::future<void>> blocking_tasks(num_of_threads_to_block);
 63+    for (auto& f : blocking_tasks) f = threadPool.Submit([&] {
 64+        ready.count_down();
 65+        release_sem.acquire();
 66+    });
 67+
 68+    ready.wait();
 69     return blocking_tasks;
 70 }
 71 
 72-BOOST_AUTO_TEST_CASE(threadpool_basic)
 73+BOOST_FIXTURE_TEST_SUITE(threadpool_tests, BasicTestingSetup)
 74+
 75+const size_t NUM_WORKERS_DEFAULT{size_t(GetNumCores()) + 1}; // we need to make sure there's *some* contention
 76+
 77+BOOST_AUTO_TEST_CASE(submit_to_non_started_pool_throws)
 78 {
 79-    // Test Cases
 80-    // 0) Submit task to a non-started pool.
 81-    // 1) Submit tasks and verify completion.
 82-    // 2) Maintain all threads busy except one.
 83-    // 3) Wait for work to finish.
 84-    // 4) Wait for result object.
 85-    // 5) The task throws an exception, catch must be done in the consumer side.
 86-    // 6) Busy workers, help them by processing tasks from outside.
 87-    // 7) Recursive submission of tasks.
 88-    // 8) Submit task when all threads are busy, stop pool and verify the task gets executed.
 89-
 90-    const int NUM_WORKERS_DEFAULT = 3;
 91-    const std::string POOL_NAME = "test";
 92-
 93-    // Test case 0, submit task to a non-started pool
 94-    {
 95-        ThreadPool threadPool(POOL_NAME);
 96-        bool err = false;
 97-        try {
 98-            threadPool.Submit([]() { return false; });
 99-        } catch (const std::runtime_error&) {
100-            err = true;
101-        }
102-        BOOST_CHECK(err);
103+    ThreadPool threadPool{"not_started"};
104+    BOOST_CHECK_EXCEPTION(threadPool.Submit([] { return 0; }), std::runtime_error, HasReason{"No active workers"});
105+}
106+
107+BOOST_AUTO_TEST_CASE(submit_and_verify_completion)
108+{
109+    ThreadPool threadPool{"completion"};
110+    threadPool.Start(NUM_WORKERS_DEFAULT);
111+
112+    const auto num_tasks{1 + m_rng.randrange<size_t>(50)};
113+    std::atomic_size_t counter{0};
114+
115+    std::vector<std::future<void>> futures(num_tasks);
116+    for (size_t i{0}; i < num_tasks; ++i) {
117+        futures[i] = threadPool.Submit([&counter, i] { counter.fetch_add(i, std::memory_order_relaxed); });
118     }
119 
120-    // Test case 1, submit tasks and verify completion.
121-    {
122-        int num_tasks = 50;
123+    WaitFor(futures);
124+    BOOST_CHECK_EQUAL(counter.load(std::memory_order_relaxed), (num_tasks - 1) * num_tasks / 2);
125+    BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 0);
126+}
127 
128-        ThreadPool threadPool(POOL_NAME);
129-        threadPool.Start(NUM_WORKERS_DEFAULT);
130-        std::atomic<int> counter = 0;
131+BOOST_AUTO_TEST_CASE(limited_free_workers_processes_all_task)
132+{
133+    ThreadPool threadPool{"block_counts"};
134+    threadPool.Start(NUM_WORKERS_DEFAULT);
135+    const auto num_tasks{1 + m_rng.randrange<size_t>(20)};
136 
137-        // Store futures to ensure completion before checking counter.
138-        std::vector<std::future<void>> futures;
139-        futures.reserve(num_tasks);
140+    for (size_t free{1}; free < NUM_WORKERS_DEFAULT; ++free) {
141+        BOOST_TEST_MESSAGE("Testing with " << free << " available workers");
142+        std::counting_semaphore sem{0};
143+        const auto blocking_tasks{BlockWorkers(threadPool, sem, free)};
144 
145-        for (int i = 1; i <= num_tasks; i++) {
146-            futures.emplace_back(threadPool.Submit([&counter, i]() {
147-                counter.fetch_add(i);
148-            }));
149-        }
150+        size_t counter{0};
151+        std::vector<std::future<void>> futures(num_tasks);
152+        for (auto& f : futures) f = threadPool.Submit([&counter] { ++counter; });
153 
154-        // Wait for all tasks to finish
155-        WaitFor(futures, /*context=*/"test1 task");
156-        int expected_value = (num_tasks * (num_tasks + 1)) / 2; // Gauss sum.
157-        BOOST_CHECK_EQUAL(counter.load(), expected_value);
158+        WaitFor(futures);
159         BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 0);
160-    }
161 
162-    // Test case 2, maintain all threads busy except one.
163-    {
164-        ThreadPool threadPool(POOL_NAME);
165-        threadPool.Start(NUM_WORKERS_DEFAULT);
166-        // Single blocking future for all threads
167-        std::promise<void> blocker;
168-        std::shared_future<void> blocker_future(blocker.get_future());
169-        const auto& blocking_tasks = BlockWorkers(threadPool, blocker_future, NUM_WORKERS_DEFAULT - 1, /*context=*/"test2 blocking tasks enabled");
170-
171-        // Now execute tasks on the single available worker
172-        // and check that all the tasks are executed.
173-        int num_tasks = 15;
174-        int counter = 0;
175-
176-        // Store futures to wait on
177-        std::vector<std::future<void>> futures;
178-        futures.reserve(num_tasks);
179-        for (int i = 0; i < num_tasks; i++) {
180-            futures.emplace_back(threadPool.Submit([&counter]() {
181-                counter += 1;
182-            }));
183+        if (free == 1) {
184+            BOOST_CHECK_EQUAL(counter, num_tasks);
185+        } else {
186+            BOOST_CHECK_LE(counter, num_tasks); // unsynchronized update from multiple threads doesn't guarantee consistency
187         }
188 
189-        WaitFor(futures, /*context=*/"test2 tasks");
190-        BOOST_CHECK_EQUAL(counter, num_tasks);
191-
192-        blocker.set_value();
193-        WaitFor(blocking_tasks, /*context=*/"test2 blocking tasks disabled");
194-        threadPool.Stop();
195-        BOOST_CHECK_EQUAL(threadPool.WorkersCount(), 0);
196+        sem.release(free);
197+        WaitFor(blocking_tasks);
198     }
199 
200-    // Test case 3, wait for work to finish.
201-    {
202-        ThreadPool threadPool(POOL_NAME);
203-        threadPool.Start(NUM_WORKERS_DEFAULT);
204-        std::atomic<bool> flag = false;
205-        std::future<void> future = threadPool.Submit([&flag]() {
206-            std::this_thread::sleep_for(std::chrono::milliseconds{200});
207-            flag.store(true);
208-        });
209-        future.wait();
210-        BOOST_CHECK(flag.load());
211-    }
212+    threadPool.Stop();
213+}
214 
215-    // Test case 4, obtain result object.
216-    {
217-        ThreadPool threadPool(POOL_NAME);
218-        threadPool.Start(NUM_WORKERS_DEFAULT);
219-        std::future<bool> future_bool = threadPool.Submit([]() {
220-            return true;
221-        });
222-        BOOST_CHECK(future_bool.get());
223+BOOST_AUTO_TEST_CASE(future_wait_blocks_until_task_completes)
224+{
225+    ThreadPool threadPool{"wait_test"};
226+    threadPool.Start(NUM_WORKERS_DEFAULT);
227+    const auto num_tasks{1 + m_rng.randrange<size_t>(20)};
228 
229-        std::future<std::string> future_str = threadPool.Submit([]() {
230-            return std::string("true");
231-        });
232-        std::string result = future_str.get();
233-        BOOST_CHECK_EQUAL(result, "true");
234+    const auto start{steady_clock::now()};
235+
236+    std::vector<std::future<void>> futures(num_tasks + 1);
237+    for (size_t i{0}; i <= num_tasks; ++i) {
238+        futures[i] = threadPool.Submit([i] { UninterruptibleSleep(milliseconds{i}); });
239     }
240+    WaitFor(futures);
241 
242-    // Test case 5, throw exception and catch it on the consumer side.
243-    {
244-        ThreadPool threadPool(POOL_NAME);
245-        threadPool.Start(NUM_WORKERS_DEFAULT);
246-
247-        int ROUNDS = 5;
248-        std::string err_msg{"something wrong happened"};
249-        std::vector<std::future<void>> futures;
250-        futures.reserve(ROUNDS);
251-        for (int i = 0; i < ROUNDS; i++) {
252-            futures.emplace_back(threadPool.Submit([err_msg, i]() {
253-                throw std::runtime_error(err_msg + util::ToString(i));
254-            }));
255-        }
256+    const size_t elapsed_ms{size_t(duration_cast<milliseconds>(steady_clock::now() - start).count())};
257+    BOOST_CHECK(elapsed_ms >= num_tasks);
258+}
259 
260-        for (int i = 0; i < ROUNDS; i++) {
261-            try {
262-                futures.at(i).get();
263-                BOOST_FAIL("Expected exception not thrown");
264-            } catch (const std::runtime_error& e) {
265-                BOOST_CHECK_EQUAL(e.what(), err_msg + util::ToString(i));
266-            }
267-        }
268-    }
269+BOOST_AUTO_TEST_CASE(future_get_returns_task_result)
270+{
271+    ThreadPool threadPool{"result_test"};
272+    threadPool.Start(NUM_WORKERS_DEFAULT);
273 
274-    // Test case 6, all workers are busy, help them by processing tasks from outside.
275-    {
276-        ThreadPool threadPool(POOL_NAME);
277-        threadPool.Start(NUM_WORKERS_DEFAULT);
278-
279-        std::promise<void> blocker;
280-        std::shared_future<void> blocker_future(blocker.get_future());
281-        const auto& blocking_tasks = BlockWorkers(threadPool, blocker_future, NUM_WORKERS_DEFAULT, /*context=*/"test6 blocking tasks enabled");
282-
283-        // Now submit tasks and check that none of them are executed.
284-        int num_tasks = 20;
285-        std::atomic<int> counter = 0;
286-        for (int i = 0; i < num_tasks; i++) {
287-            threadPool.Submit([&counter]() {
288-                counter.fetch_add(1);
289-            });
290-        }
291-        std::this_thread::sleep_for(std::chrono::milliseconds{100});
292-        BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 20);
293+    BOOST_CHECK_EQUAL(threadPool.Submit([] { return true; }).get(), true);
294+    BOOST_CHECK_EQUAL(threadPool.Submit([] { return 42; }).get(), 42);
295+    BOOST_CHECK_EQUAL(threadPool.Submit([] { return std::string{"true"}; }).get(), "true");
296+}
297 
298-        // Now process manually
299-        for (int i = 0; i < num_tasks; i++) {
300-            threadPool.ProcessTask();
301-        }
302-        BOOST_CHECK_EQUAL(counter.load(), num_tasks);
303-        BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 0);
304-        blocker.set_value();
305-        threadPool.Stop();
306-        WaitFor(blocking_tasks, "Failure waiting for test6 blocking task futures");
307+BOOST_AUTO_TEST_CASE(task_exception_propagated_to_future)
308+{
309+    ThreadPool threadPool{"exception_test"};
310+    threadPool.Start(NUM_WORKERS_DEFAULT);
311+
312+    const auto err{[&](size_t n) { return strprintf("error on thread #%s", n); }};
313+
314+    const auto num_tasks{1 + m_rng.randrange<size_t>(20)};
315+    for (size_t i{0}; i < num_tasks; ++i) {
316+        BOOST_CHECK_EXCEPTION(threadPool.Submit([&] { throw std::runtime_error(err(i)); }).get(), std::runtime_error, HasReason{err(i)});
317     }
318+}
319+
320+BOOST_AUTO_TEST_CASE(process_task_manually_when_workers_busy)
321+{
322+    ThreadPool threadPool{"manual_process"};
323+    threadPool.Start(NUM_WORKERS_DEFAULT);
324+    const auto num_tasks{1 + m_rng.randrange<size_t>(20)};
325 
326-    // Test case 7, recursive submission of tasks.
327-    {
328-        ThreadPool threadPool(POOL_NAME);
329-        threadPool.Start(NUM_WORKERS_DEFAULT);
330+    std::counting_semaphore sem{0};
331+    const auto blocking_tasks{BlockWorkers(threadPool, sem, NUM_WORKERS_DEFAULT)};
332 
333-        std::promise<void> signal;
334-        threadPool.Submit([&]() {
335-            threadPool.Submit([&]() {
336-                signal.set_value();
337-            });
338-        });
339+    std::atomic_size_t counter{0};
340+    std::vector<std::future<void>> futures(num_tasks);
341+    for (auto& f : futures) f = threadPool.Submit([&counter] { counter.fetch_add(1, std::memory_order_relaxed); });
342 
343-        signal.get_future().wait();
344-        threadPool.Stop();
345-    }
346+    UninterruptibleSleep(milliseconds{100});
347+    BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), num_tasks);
348 
349-    // Test case 8, submit a task when all threads are busy and then stop the pool.
350-    {
351-        ThreadPool threadPool(POOL_NAME);
352-        threadPool.Start(NUM_WORKERS_DEFAULT);
353+    for (size_t i{0}; i < num_tasks; ++i) {
354+        threadPool.ProcessTask();
355+    }
356 
357-        std::promise<void> blocker;
358-        std::shared_future<void> blocker_future(blocker.get_future());
359-        const auto& blocking_tasks = BlockWorkers(threadPool, blocker_future, NUM_WORKERS_DEFAULT, /*context=*/"test8 blocking tasks enabled");
360+    BOOST_CHECK_EQUAL(counter.load(), num_tasks);
361+    BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 0);
362 
363-        // Submit an extra task that should execute once a worker is free
364-        std::future<bool> future = threadPool.Submit([]() { return true; });
365+    sem.release(NUM_WORKERS_DEFAULT);
366+    WaitFor(blocking_tasks);
367+}
368 
369-        // At this point, all workers are blocked, and the extra task is queued
370-        BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 1);
371+BOOST_AUTO_TEST_CASE(recursive_task_submission)
372+{
373+    ThreadPool threadPool{"recursive"};
374+    threadPool.Start(NUM_WORKERS_DEFAULT);
375 
376-        // Wait a short moment before unblocking the threads to mimic a concurrent shutdown
377-        std::thread thread_unblocker([&blocker]() {
378-            std::this_thread::sleep_for(std::chrono::milliseconds{300});
379-            blocker.set_value();
380+    std::promise<void> signal;
381+    threadPool.Submit([&threadPool, &signal] {
382+        threadPool.Submit([&signal] {
383+            signal.set_value();
384         });
385+    });
386 
387-        // Stop the pool while the workers are still blocked
388-        threadPool.Stop();
389+    signal.get_future().wait();
390+}
391 
392-        // Expect the submitted task to complete
393-        BOOST_CHECK(future.get());
394-        thread_unblocker.join();
395+BOOST_AUTO_TEST_CASE(stop_completes_queued_tasks_gracefully)
396+{
397+    ThreadPool threadPool{"graceful_stop"};
398+    threadPool.Start(NUM_WORKERS_DEFAULT);
399 
400-        // Obviously all the previously blocking tasks should be completed at this point too
401-        WaitFor(blocking_tasks, "Failure waiting for test8 blocking task futures");
402+    std::counting_semaphore sem{0};
403+    const auto blocking_tasks{BlockWorkers(threadPool, sem, NUM_WORKERS_DEFAULT)};
404 
405-        // Pool should be stopped and no workers remaining
406-        BOOST_CHECK_EQUAL(threadPool.WorkersCount(), 0);
407-    }
408+    auto future{threadPool.Submit([] { return true; })};
409+    BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 1);
410+
411+    std::thread thread_unblocker{[&sem] {
412+        std::this_thread::sleep_for(milliseconds{300});
413+        sem.release(NUM_WORKERS_DEFAULT);
414+    }};
415+
416+    threadPool.Stop();
417+
418+    BOOST_CHECK(future.get());
419+    thread_unblocker.join();
420+    WaitFor(blocking_tasks);
421+    BOOST_CHECK_EQUAL(threadPool.WorkersCount(), 0);
422 }
423 
424 BOOST_AUTO_TEST_SUITE_END()
425diff --git a/src/util/threadpool.h b/src/util/threadpool.h
426index 5d9884086e..c89fda37c2 100644
427--- a/src/util/threadpool.h
428+++ b/src/util/threadpool.h
429@@ -24,6 +24,8 @@
430 #include <utility>
431 #include <vector>
432 
433+#include <tinyformat.h>
434+
435 /**
436  * [@brief](/bitcoin-bitcoin/contributor/brief/) Fixed-size thread pool for running arbitrary tasks concurrently.
437  *
438@@ -62,16 +64,9 @@ private:
439         for (;;) {
440             std::packaged_task<void()> task;
441             {
442-                // Wait only if needed; avoid sleeping when a new task was submitted while we were processing another one.
443-                if (!m_interrupt && m_work_queue.empty()) {
444-                    // Block until the pool is interrupted or a task is available.
445-                    m_cv.wait(wait_lock, [&]() EXCLUSIVE_LOCKS_REQUIRED(m_mutex) { return m_interrupt || !m_work_queue.empty(); });
446-                }
447-
448-                // If stopped and no work left, exit worker
449-                if (m_interrupt && m_work_queue.empty()) {
450-                    return;
451-                }
452+                // Block until the pool is interrupted or a task is available.
453+                m_cv.wait(wait_lock, [&]() EXCLUSIVE_LOCKS_REQUIRED(m_mutex) { return m_interrupt || !m_work_queue.empty(); });
454+                if (m_interrupt && m_work_queue.empty()) return;
455 
456                 task = std::move(m_work_queue.front());
457                 m_work_queue.pop();
458@@ -101,17 +96,16 @@ public:
459      *
460      * Must be called from a controller (non-worker) thread.
461      */
462-    void Start(int num_workers) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
463+    void Start(size_t num_workers) EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
464     {
465         assert(num_workers > 0);
466         LOCK(m_mutex);
467         if (!m_workers.empty()) throw std::runtime_error("Thread pool already started");
468         m_interrupt = false; // Reset
469 
470-        // Create workers
471         m_workers.reserve(num_workers);
472-        for (int i = 0; i < num_workers; i++) {
473-            m_workers.emplace_back(&util::TraceThread, m_name + "_pool_" + util::ToString(i), [this] { WorkerThread(); });
474+        for (size_t i{0}; i < num_workers; i++) {
475+            m_workers.emplace_back(&util::TraceThread, strprintf("%s_pool_%d", m_name, i), [this] { WorkerThread(); });
476         }
477     }
478 
479@@ -179,12 +173,6 @@ public:
480         task();
481     }
482 
483-    void Interrupt() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
484-    {
485-        WITH_LOCK(m_mutex, m_interrupt = true);
486-        m_cv.notify_all();
487-    }
488-
489     size_t WorkQueueSize() EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
490     {
491         return WITH_LOCK(m_mutex, return m_work_queue.size());