threadpool: make Submit return Expected instead of throwing
What changed, and why it matters
This change refactors Bitcoin Core's internal thread pool so that submitting a task returns an error code instead of throwing an exception. The main practical effect is fixing an unclean shutdown bug (#34573) where an unhandled exception could be thrown from an HTTP event callback. It is a defensive reliability improvement rather than a direct exploit for stealing funds or remote code execution.
Treat as a routine reliability/defensive fix. Review that all production callers of ThreadPool::Submit() now handle the returned Expected value correctly; the diff only shows test callers being updated, so inspect other call sites in the codebase for proper error handling.
Security signals we found
Change from exception-throwing to error-returning API
Commit message explicitly links change to fixing unclean shutdown bug #34573
HTTP callback context mentioned as a caller that previously could not handle exceptions cleanly
No change to network protocol, consensus, cryptography, or wallet logic
Evidence from the diff
ThreadPool::Submit() now returns util::Expected<std::future<…>, SubmitError> instead of throwing std::runtime_error when the pool is inactive or interrupted. Callers must explicitly handle the error. The patch also updates tests and fuzz harnesses to unwrap the Expected result. The commit message states this fixes an unclean shutdown bug (#34573) because http_request_cb() no longer receives an exception from Submit().
Changed components
src/util/threadpool.hsrc/test/threadpool_tests.cppsrc/test/fuzz/threadpool.cppInspect captured patch +64 / −35
diff --git a/src/test/fuzz/threadpool.cpp b/src/test/fuzz/threadpool.cpp
index 293aa63e..a5b01db1 100644
--- a/src/test/fuzz/threadpool.cpp
+++ b/src/test/fuzz/threadpool.cpp
@@ -87,10 +87,10 @@ FUZZ_TARGET(threadpool, .init = setup_threadpool_test) EXCLUSIVE_LOCKS_REQUIRED(
std::future<void> fut;
if (will_throw) {
expected_fail_tasks++;
- fut = g_pool.Submit(ThrowTask{});
+ fut = *Assert(g_pool.Submit(ThrowTask{}));
} else {
expected_task_counter++;
- fut = g_pool.Submit(CounterTask{task_counter});
+ fut = *Assert(g_pool.Submit(CounterTask{task_counter}));
}
// If caller wants to wait immediately, consume the future here (safe).
diff --git a/src/test/threadpool_tests.cpp b/src/test/threadpool_tests.cpp
index 46acf1d6..850cd166 100644
--- a/src/test/threadpool_tests.cpp
+++ b/src/test/threadpool_tests.cpp
@@ -44,6 +44,13 @@ BOOST_FIXTURE_TEST_SUITE(threadpool_tests, ThreadPoolFixture)
} \
} while (0)
+// Helper to unwrap a valid pool submission
+template <typename F>
+[[nodiscard]] auto Submit(ThreadPool& pool, F&& fn)
+{
+ return std::move(*Assert(pool.Submit(std::forward<F>(fn))));
+}
+
// Block a number of worker threads by submitting tasks that wait on `blocker_future`.
// Returns the futures of the blocking tasks, ensuring all have started and are waiting.
std::vector<std::future<void>> BlockWorkers(ThreadPool& threadPool, const std::shared_future<void>& blocker_future, int num_of_threads_to_block)
@@ -58,7 +65,7 @@ std::vector<std::future<void>> BlockWorkers(ThreadPool& threadPool, const std::s
std::vector<std::future<void>> blocking_tasks;
for (int i = 0; i < num_of_threads_to_block; i++) {
std::promise<void>& ready = ready_promises[i];
- blocking_tasks.emplace_back(threadPool.Submit([blocker_future, &ready]() {
+ blocking_tasks.emplace_back(Submit(threadPool, [blocker_future, &ready]() {
ready.set_value();
blocker_future.wait();
}));
@@ -73,10 +80,9 @@ std::vector<std::future<void>> BlockWorkers(ThreadPool& threadPool, const std::s
BOOST_AUTO_TEST_CASE(submit_task_before_start_fails)
{
ThreadPool threadPool(POOL_NAME);
- BOOST_CHECK_EXCEPTION((void)threadPool.Submit([]{ return false; }), std::runtime_error, [&](const std::runtime_error& e) {
- BOOST_CHECK_EQUAL(e.what(), "No active workers; cannot accept new tasks");
- return true;
- });
+ auto res = threadPool.Submit([]{ return false; });
+ BOOST_CHECK(!res);
+ BOOST_CHECK_EQUAL(SubmitErrorString(res.error()), "No active workers");
}
// Test 1, submit tasks and verify completion
@@ -92,7 +98,7 @@ BOOST_AUTO_TEST_CASE(submit_tasks_complete_successfully)
std::vector<std::future<void>> futures;
futures.reserve(num_tasks);
for (int i = 1; i <= num_tasks; i++) {
- futures.emplace_back(threadPool.Submit([&counter, i]() {
+ futures.emplace_back(Submit(threadPool, [&counter, i]() {
counter.fetch_add(i, std::memory_order_relaxed);
}));
}
@@ -121,7 +127,7 @@ BOOST_AUTO_TEST_CASE(single_available_worker_executes_all_tasks)
// Store futures to wait on
std::vector<std::future<void>> futures(num_tasks);
- for (auto& f : futures) f = threadPool.Submit([&counter]{ counter++; });
+ for (auto& f : futures) f = Submit(threadPool, [&counter]{ counter++; });
WAIT_FOR(futures);
BOOST_CHECK_EQUAL(counter, num_tasks);
@@ -138,7 +144,7 @@ BOOST_AUTO_TEST_CASE(wait_for_task_to_finish)
ThreadPool threadPool(POOL_NAME);
threadPool.Start(NUM_WORKERS_DEFAULT);
std::atomic<bool> flag = false;
- std::future<void> future = threadPool.Submit([&flag]() {
+ std::future<void> future = Submit(threadPool, [&flag]() {
UninterruptibleSleep(200ms);
flag.store(true, std::memory_order_release);
});
@@ -151,10 +157,10 @@ BOOST_AUTO_TEST_CASE(get_result_from_completed_task)
{
ThreadPool threadPool(POOL_NAME);
threadPool.Start(NUM_WORKERS_DEFAULT);
- std::future<bool> future_bool = threadPool.Submit([]() { return true; });
+ std::future<bool> future_bool = Submit(threadPool, []() { return true; });
BOOST_CHECK(future_bool.get());
- std::future<std::string> future_str = threadPool.Submit([]() { return std::string("true"); });
+ std::future<std::string> future_str = Submit(threadPool, []() { return std::string("true"); });
std::string result = future_str.get();
BOOST_CHECK_EQUAL(result, "true");
}
@@ -170,7 +176,7 @@ BOOST_AUTO_TEST_CASE(task_exception_propagates_to_future)
std::vector<std::future<void>> futures;
futures.reserve(num_tasks);
for (int i = 0; i < num_tasks; i++) {
- futures.emplace_back(threadPool.Submit([err_msg, i]() {
+ futures.emplace_back(Submit(threadPool, [err_msg, i]() {
throw std::runtime_error(err_msg + util::ToString(i));
}));
}
@@ -197,7 +203,7 @@ BOOST_AUTO_TEST_CASE(process_tasks_manually_when_workers_busy)
int num_tasks = 20;
std::atomic<int> counter = 0;
for (int i = 0; i < num_tasks; i++) {
- (void)threadPool.Submit([&counter]() {
+ (void)Submit(threadPool, [&counter]() {
counter.fetch_add(1, std::memory_order_relaxed);
});
}
@@ -222,8 +228,8 @@ BOOST_AUTO_TEST_CASE(recursive_task_submission)
threadPool.Start(NUM_WORKERS_DEFAULT);
std::promise<void> signal;
- (void)threadPool.Submit([&]() {
- (void)threadPool.Submit([&]() {
+ (void)Submit(threadPool, [&]() {
+ (void)Submit(threadPool, [&]() {
signal.set_value();
});
});
@@ -243,7 +249,7 @@ BOOST_AUTO_TEST_CASE(task_submitted_while_busy_completes)
const auto& blocking_tasks = BlockWorkers(threadPool, blocker_future, NUM_WORKERS_DEFAULT);
// Submit an extra task that should execute once a worker is free
- std::future<bool> future = threadPool.Submit([]() { return true; });
+ std::future<bool> future = Submit(threadPool, []() { return true; });
// At this point, all workers are blocked, and the extra task is queued
BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 1);
@@ -280,7 +286,7 @@ BOOST_AUTO_TEST_CASE(congestion_more_workers_than_cores)
std::vector<std::future<void>> futures;
futures.reserve(num_tasks);
for (int i = 0; i < num_tasks; i++) {
- futures.emplace_back(threadPool.Submit([&counter] {
+ futures.emplace_back(Submit(threadPool, [&counter] {
counter.fetch_add(1, std::memory_order_relaxed);
}));
}
@@ -296,10 +302,10 @@ BOOST_AUTO_TEST_CASE(interrupt_blocks_new_submissions)
ThreadPool threadPool(POOL_NAME);
threadPool.Start(NUM_WORKERS_DEFAULT);
threadPool.Interrupt();
- BOOST_CHECK_EXCEPTION((void)threadPool.Submit([]{}), std::runtime_error, [&](const std::runtime_error& e) {
- BOOST_CHECK_EQUAL(e.what(), "No active workers; cannot accept new tasks");
- return true;
- });
+
+ auto res = threadPool.Submit([]{});
+ BOOST_CHECK(!res);
+ BOOST_CHECK_EQUAL(SubmitErrorString(res.error()), "Interrupted");
// Reset pool
threadPool.Stop();
@@ -310,7 +316,7 @@ BOOST_AUTO_TEST_CASE(interrupt_blocks_new_submissions)
std::atomic<int> counter{0};
std::promise<void> blocker;
const auto blocking_tasks = BlockWorkers(threadPool, blocker.get_future().share(), 1);
- threadPool.Submit([&threadPool, &counter]{
+ Submit(threadPool, [&threadPool, &counter]{
threadPool.Interrupt();
counter.fetch_add(1, std::memory_order_relaxed);
}).get();
diff --git a/src/util/threadpool.h b/src/util/threadpool.h
index b75a9415..c039b59c 100644
--- a/src/util/threadpool.h
+++ b/src/util/threadpool.h
@@ -7,6 +7,7 @@
#include <sync.h>
#include <tinyformat.h>
+#include <util/expected.h>
#include <util/check.h>
#include <util/thread.h>
@@ -15,8 +16,8 @@
#include <functional>
#include <future>
#include <queue>
-#include <stdexcept>
#include <thread>
+#include <type_traits>
#include <utility>
#include <vector>
@@ -143,28 +144,39 @@ public:
// Note: m_interrupt is left true until next Start()
}
+ enum class SubmitError {
+ Inactive,
+ Interrupted,
+ };
+
/**
* @brief Enqueues a new task for asynchronous execution.
*
- * Returns a `std::future` that provides the task's result or propagates
- * any exception it throws.
- * Note: Ignoring the returned future requires guarding the task against
- * uncaught exceptions, as they would otherwise be silently discarded.
+ * @param fn Callable to execute asynchronously.
+ * @return On success, a future containing fn's result.
+ * On failure, an error indicating why the task was rejected:
+ * - SubmitError::Inactive: Pool has no workers (never started or already stopped).
+ * - SubmitError::Interrupted: Pool task acceptance has been interrupted.
+ *
+ * Thread-safe: Can be called from any thread, including within the provided 'fn' callable.
+ *
+ * @warning Ignoring the returned future requires guarding the task against
+ * uncaught exceptions, as they would otherwise be silently discarded.
*/
- template <class F> [[nodiscard]] EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
- auto Submit(F&& fn)
+ template <class F>
+ [[nodiscard]] util::Expected<std::future<std::invoke_result_t<F>>, SubmitError> Submit(F&& fn) noexcept EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
{
- std::packaged_task task{std::forward<F>(fn)};
+ std::packaged_task<std::invoke_result_t<F>()> task{std::forward<F>(fn)};
auto future{task.get_future()};
{
LOCK(m_mutex);
- if (m_interrupt || m_workers.empty()) {
- throw std::runtime_error("No active workers; cannot accept new tasks");
- }
+ if (m_workers.empty()) return util::Unexpected{SubmitError::Inactive};
+ if (m_interrupt) return util::Unexpected{SubmitError::Interrupted};
+
m_work_queue.emplace(std::move(task));
}
m_cv.notify_one();
- return future;
+ return {std::move(future)};
}
/**
@@ -208,4 +220,15 @@ public:
}
};
+constexpr std::string_view SubmitErrorString(const ThreadPool::SubmitError err) noexcept {
+ switch (err) {
+ case ThreadPool::SubmitError::Inactive:
+ return "No active workers";
+ case ThreadPool::SubmitError::Interrupted:
+ return "Interrupted";
+ }
+ Assume(false); // Unreachable
+ return "Unknown error";
+}
+
#endif // BITCOIN_UTIL_THREADPOOL_H
Why this scored 31/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.