Inside Bitcoin
Opening the selected file…
Inside Bitcoin
Opening the selected file…
The code behind the network, with room to understand it.
Release published 2026-10-28 · Snapshot fetched 2026-11-01. Latest stable release at fetch time; development branches and Git history are excluded. Only the selected code window and visible folders load here.
251// Test 7, submit tasks from other tasks252BOOST_AUTO_TEST_CASE(recursive_task_submission)253{254ThreadPool threadPool(POOL_NAME);255threadPool.Start(NUM_WORKERS_DEFAULT);257std::promise<void> signal;258(void)Submit(threadPool, [&]() {259(void)Submit(threadPool, [&]() {260signal.set_value();261});262});264signal.get_future().wait();265threadPool.Stop();266}268// Test 8, submit task when all threads are busy and then stop the pool269BOOST_AUTO_TEST_CASE(task_submitted_while_busy_completes)270{271ThreadPool threadPool(POOL_NAME);272threadPool.Start(NUM_WORKERS_DEFAULT);274std::counting_semaphore<> blocker(0);275const auto& blocking_tasks = BlockWorkers(threadPool, blocker, NUM_WORKERS_DEFAULT);277// Submit an extra task that should execute once a worker is free278std::future<bool> future = Submit(threadPool, []() { return true; });280// At this point, all workers are blocked, and the extra task is queued281BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 1);283// Wait a short moment before unblocking the threads to mimic a concurrent shutdown284std::thread thread_unblocker([&blocker]() {285UninterruptibleSleep(300ms);286blocker.release(NUM_WORKERS_DEFAULT);287});289// Stop the pool while the workers are still blocked290threadPool.Stop();292// Expect the submitted task to complete293BOOST_CHECK(future.get());294thread_unblocker.join();296// Obviously all the previously blocking tasks should be completed at this point too297WAIT_FOR(blocking_tasks);299// Pool should be stopped and no workers remaining300BOOST_CHECK_EQUAL(threadPool.WorkersCount(), 0);301}303// Test 9, more workers than available cores (congestion test)304BOOST_AUTO_TEST_CASE(congestion_more_workers_than_cores)305{306ThreadPool threadPool(POOL_NAME);307threadPool.Start(std::max(1, GetNumCores() * 2)); // Oversubscribe by 2×309int num_tasks = 200;310std::atomic<int> counter{0};312std::vector<std::future<void>> futures;313futures.reserve(num_tasks);314for (int i = 0; i < num_tasks; i++) {315futures.emplace_back(Submit(threadPool, [&counter] {316counter.fetch_add(1, std::memory_order_relaxed);317}));318}320WAIT_FOR(futures);321BOOST_CHECK_EQUAL(counter.load(), num_tasks);322}324// Test 10, Interrupt() prevents further submissions325BOOST_AUTO_TEST_CASE(interrupt_blocks_new_submissions)326{327// 1) Interrupt from main thread328ThreadPool threadPool(POOL_NAME);329threadPool.Start(NUM_WORKERS_DEFAULT);330threadPool.Interrupt();332auto res = threadPool.Submit([]{});333BOOST_CHECK(!res);334BOOST_CHECK_EQUAL(SubmitErrorString(res.error()), "Interrupted");336std::vector<std::function<void()>> tasks;337const auto range_res{threadPool.Submit(std::move(tasks))};338BOOST_CHECK(!range_res);339BOOST_CHECK_EQUAL(SubmitErrorString(range_res.error()), "Interrupted");341// Reset pool342threadPool.Stop();344// 2) Interrupt() from a worker thread345// One worker is blocked, another calls Interrupt(), and the remaining one waits for tasks.346threadPool.Start(/*num_workers=*/3);347std::atomic<int> counter{0};348std::counting_semaphore<> blocker(0);349const auto blocking_tasks = BlockWorkers(threadPool, blocker, 1);350Submit(threadPool, [&threadPool, &counter]{351threadPool.Interrupt();352counter.fetch_add(1, std::memory_order_relaxed);353}).get();354blocker.release(1); // unblock worker356BOOST_CHECK_EQUAL(counter.load(), 1);357threadPool.Stop();358WAIT_FOR(blocking_tasks);359BOOST_CHECK_EQUAL(threadPool.WorkersCount(), 0);360}362// Test 11, Start() must not cause a deadlock when called during Stop()363BOOST_AUTO_TEST_CASE(start_mid_stop_does_not_deadlock)364{365ThreadPool threadPool(POOL_NAME);366threadPool.Start(NUM_WORKERS_DEFAULT);368// Keep all workers busy so Stop() gets stuck waiting for them to finish during join()369std::counting_semaphore<> workers_blocker(0);370const auto blocking_tasks = BlockWorkers(threadPool, workers_blocker, NUM_WORKERS_DEFAULT);372std::thread stopper_thread([&threadPool] { threadPool.Stop(); });374// Stop() takes ownership of the workers before joining them, so WorkersCount()375// hits 0 the moment Stop() is waiting for them to join. That is our signal376// to call Start() right into the middle of the joining phase.377while (threadPool.WorkersCount() != 0) {378std::this_thread::yield(); // let the OS breathe so it can switch context379}380// Now we know for sure the stopper thread is hanging while workers are still alive.381// Restart the pool and resume workers so the stopper thread can proceed.382// This will throw an exception only if the pool handles Start-Stop race properly,383// otherwise it will proceed and hang the stopper_thread.384BOOST_CHECK_EXCEPTION(threadPool.Start(NUM_WORKERS_DEFAULT), std::runtime_error, HasReason{"Thread pool has been interrupted or is stopping"});385workers_blocker.release(NUM_WORKERS_DEFAULT);386WAIT_FOR(blocking_tasks);388// If Stop() is stuck, joining the stopper thread will deadlock389stopper_thread.join();390}392// Test 12, queued tasks complete after Interrupt()393BOOST_AUTO_TEST_CASE(queued_tasks_complete_after_interrupt)394{395ThreadPool threadPool(POOL_NAME);396threadPool.Start(NUM_WORKERS_DEFAULT);398std::counting_semaphore<> blocker(0);399const auto blocking_tasks = BlockWorkers(threadPool, blocker, NUM_WORKERS_DEFAULT);401// Queue tasks while all workers are busy, then interrupt402std::atomic<int> counter{0};403const int num_tasks = 10;404std::vector<std::future<void>> futures;405futures.reserve(num_tasks);406for (int i = 0; i < num_tasks; i++) {407futures.emplace_back(Submit(threadPool, [&counter]{ counter.fetch_add(1, std::memory_order_relaxed); }));408}409threadPool.Interrupt();411// Queued tasks must still complete despite the interrupt412blocker.release(NUM_WORKERS_DEFAULT);413WAIT_FOR(futures);414BOOST_CHECK_EQUAL(counter.load(), num_tasks);416threadPool.Stop();417WAIT_FOR(blocking_tasks);418}420// Test 13, ensure the Stop() calling thread helps drain the queue421BOOST_AUTO_TEST_CASE(stop_active_wait_drains_queue)422{423ThreadPool threadPool(POOL_NAME);424threadPool.Start(NUM_WORKERS_DEFAULT);426std::counting_semaphore<> blocker(0);427const auto blocking_tasks = BlockWorkers(threadPool, blocker, NUM_WORKERS_DEFAULT);429auto main_thread_id = std::this_thread::get_id();430std::atomic<int> main_thread_tasks{0};431const size_t num_tasks = 20;432for (size_t i = 0; i < num_tasks; i++) {433(void)Submit(threadPool, [&main_thread_tasks, main_thread_id]() {434if (std::this_thread::get_id() == main_thread_id)435main_thread_tasks.fetch_add(1, std::memory_order_relaxed);436});437}438BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), num_tasks);440// Delay release so Stop() drains all tasks from the calling thread441std::thread unblocker([&blocker, &threadPool]() {442while (threadPool.WorkQueueSize() > 0) {443std::this_thread::yield();444}445blocker.release(NUM_WORKERS_DEFAULT);446});448threadPool.Stop();449unblocker.join();451// Check the main thread processed all tasks452BOOST_CHECK_EQUAL(main_thread_tasks.load(), num_tasks);453WAIT_FOR(blocking_tasks);454}456// Test 14, submit range of tasks in one lock acquisition457BOOST_AUTO_TEST_CASE(submit_range_of_tasks_complete_successfully)458{459constexpr int32_t num_tasks{50};461ThreadPool threadPool{POOL_NAME};462threadPool.Start(NUM_WORKERS_DEFAULT);463std::atomic_int32_t sum{0};464const auto square{[&sum](int32_t i) {465sum.fetch_add(i, std::memory_order_relaxed);466return i * i;467}};469std::array<std::function<int32_t()>, static_cast<size_t>(num_tasks)> array_tasks;470std::vector<std::function<int32_t()>> vector_tasks;471vector_tasks.reserve(static_cast<size_t>(num_tasks));472for (const auto i : std::views::iota(int32_t{1}, num_tasks + 1)) {473array_tasks.at(static_cast<size_t>(i - 1)) = [i, square] { return square(i); };474vector_tasks.emplace_back([i, square] { return square(i); });475}477auto futures{std::move(*Assert(threadPool.Submit(std::move(array_tasks))))};478BOOST_CHECK_EQUAL(futures.size(), static_cast<size_t>(num_tasks));479std::ranges::move(*Assert(threadPool.Submit(std::move(vector_tasks))), std::back_inserter(futures));480BOOST_CHECK_EQUAL(futures.size(), static_cast<size_t>(num_tasks * 2));482auto squares_sum{0};483for (auto& future : futures) {484squares_sum += future.get();485}487// 2x Gauss sum.488const auto expected_sum{2 * ((num_tasks * (num_tasks + 1)) / 2)};489const auto expected_squares_sum{2 * ((num_tasks * (num_tasks + 1) * ((num_tasks * 2) + 1)) / 6)};490BOOST_CHECK_EQUAL(sum, expected_sum);491BOOST_CHECK_EQUAL(squares_sum, expected_squares_sum);492BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), 0);493}495BOOST_AUTO_TEST_SUITE_END()
Independent explanations by MyBitcoinExplorer. Not affiliated with or endorsed by Bitcoin Core. Guides are educational and do not replace the source or its tests.
Bitcoin Core source © its contributors, distributed under the MIT license. Original notices and bundled third-party licenses are retained in the source tree.