fuzz: add test case for threadpool

Co-authored-by: furszy <[email protected]>

Author: TheCharlatan

src/test/fuzz/CMakeLists.txt

@@ -119,6 +119,7 @@ add_executable(fuzz
   string.cpp
   strprintf.cpp
   system.cpp
+  threadpool.cpp
   timeoffsets.cpp
   torcontrol.cpp
   transaction.cpp

src/test/fuzz/threadpool.cpp

@@ -0,0 +1,109 @@
+// Copyright (c) The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#include <logging.h>
+#include <util/threadpool.h>
+
+#include <test/fuzz/FuzzedDataProvider.h>
+#include <test/fuzz/fuzz.h>
+
+#include <atomic>
+#include <future>
+#include <queue>
+
+struct ExpectedException : std::runtime_error {
+    explicit ExpectedException(const std::string& msg) : std::runtime_error(msg) {}
+};
+
+struct ThrowTask {
+    void operator()() const { throw ExpectedException("fail"); }
+};
+
+struct CounterTask {
+    std::atomic_uint32_t& m_counter;
+    explicit CounterTask(std::atomic_uint32_t& counter) : m_counter{counter} {}
+    void operator()() const { m_counter.fetch_add(1, std::memory_order_relaxed); }
+};
+
+// Waits for a future to complete. Increments 'fail_counter' if the expected exception is thrown.
+static void GetFuture(std::future<void>& future, uint32_t& fail_counter)
+{
+    try {
+        future.get();
+    } catch (const ExpectedException&) {
+        fail_counter++;
+    } catch (...) {
+        assert(false && "Unexpected exception type");
+    }
+}
+
+// Global thread pool for fuzzing. Persisting it across iterations prevents
+// the excessive thread creation/destruction overhead that can lead to
+// instability in the fuzzing environment.
+// This is also how we use it in the app's lifecycle.
+ThreadPool g_pool{"fuzz"};
+// Global to verify we always have the same number of threads.
+size_t g_num_workers = 3;
+
+static void setup_threadpool_test()
+{
+    // Disable logging entirely. It seems to cause memory leaks.
+    LogInstance().DisableLogging();
+    // Ensure the pool is started only once.
+    static std::once_flag started;
+    std::call_once(started, []() {
+        g_pool.Start(g_num_workers);
+    });
+}
+
+FUZZ_TARGET(threadpool, .init = setup_threadpool_test)
+{
+    FuzzedDataProvider fuzzed_data_provider(buffer.data(), buffer.size());
+
+    const uint32_t num_tasks = fuzzed_data_provider.ConsumeIntegralInRange<uint32_t>(0, 1024);
+    assert(g_pool.WorkersCount() == g_num_workers);
+    assert(g_pool.WorkQueueSize() == 0);
+
+    // Counters
+    std::atomic_uint32_t task_counter{0};
+    uint32_t fail_counter{0};
+    uint32_t expected_task_counter{0};
+    uint32_t expected_fail_tasks{0};
+
+    std::queue<std::future<void>> futures;
+    for (uint32_t i = 0; i < num_tasks; ++i) {
+        const bool will_throw = fuzzed_data_provider.ConsumeBool();
+        const bool wait_immediately = fuzzed_data_provider.ConsumeBool();
+
+        std::future<void> fut;
+        if (will_throw) {
+            expected_fail_tasks++;
+            fut = g_pool.Submit(ThrowTask{});
+        } else {
+            expected_task_counter++;
+            fut = g_pool.Submit(CounterTask{task_counter});
+        }
+
+        // If caller wants to wait immediately, consume the future here (safe).
+        if (wait_immediately) {
+            // Waits for this task to complete immediately; prior queued tasks may also complete
+            // as they were queued earlier.
+            GetFuture(fut, fail_counter);
+        } else {
+            // Store task for a posterior check
+            futures.emplace(std::move(fut));
+        }
+    }
+
+    // Drain remaining futures
+    while (!futures.empty()) {
+        auto fut = std::move(futures.front());
+        futures.pop();
+        GetFuture(fut, fail_counter);
+    }
+
+    assert(g_pool.WorkQueueSize() == 0);
+    assert(task_counter.load() == expected_task_counter);
+    assert(fail_counter == expected_fail_tasks);
+}