Repository: bitcoin/bitcoin
File: src/test/threadpool_tests.cpp (C++)

Code before:
    WAIT_FOR(futures);
    BOOST_CHECK_EQUAL(counter.load(), num_tasks);

    threadPool.Stop();
    WAIT_FOR(blocking_tasks);
}

// Test 13, ensure the Stop() calling thread helps drain the queue
BOOST_AUTO_TEST_CASE(stop_active_wait_drains_queue)
{
    ThreadPool threadPool(POOL_NAME);
    threadPool.Start(NUM_WORKERS_DEFAULT);

    std::counting_semaphore<> blocker(0);
    const auto blocking_tasks = BlockWorkers(threadPool, blocker, NUM_WORKERS_DEFAULT);

    auto main_thread_id = std::this_thread::get_id();
    std::atomic<int> main_thread_tasks{0};
    const int num_tasks = 20;
    for (int i = 0; i < num_tasks; i++) {
        (void)Submit(threadPool, [&main_thread_tasks, main_thread_id]() {
            if (std::this_thread::get_id() == main_thread_id)
                main_thread_tasks.fetch_add(1, std::memory_order_relaxed);
        });
    }
    BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), static_cast<size_t>(num_tasks));

    // Delay release so Stop() drain all tasks from the calling thread
    std::thread unblocker([&blocker, &threadPool]() {
        while (threadPool.WorkQueueSize() > 0) {
            UninterruptibleSleep(300ms);
        }
        blocker.release(NUM_WORKERS_DEFAULT);
    });

    threadPool.Stop();
    unblocker.join();

    // Check the main thread processed all tasks
    BOOST_CHECK_EQUAL(main_thread_tasks.load(), num_tasks);
    WAIT_FOR(blocking_tasks);
}

BOOST_AUTO_TEST_SUITE_END()


Review comment (suggestion):
nit: Could avoid cast:

```suggestion
    const size_t num_tasks = 20;
    for (size_t i = 0; i < num_tasks; i++) {
        (void)Submit(threadPool, [&main_thread_tasks, main_thread_id]() {
            if (std::this_thread::get_id() == main_thread_id)
                main_thread_tasks.fetch_add(1, std::memory_order_relaxed);
        });
    }
    BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), num_tasks);
```

Code after:
    stopper_thread.join();
}

// Test 12, queued tasks complete after Interrupt()
BOOST_AUTO_TEST_CASE(queued_tasks_complete_after_interrupt)
{
    ThreadPool threadPool(POOL_NAME);
    threadPool.Start(NUM_WORKERS_DEFAULT);

    std::counting_semaphore<> blocker(0);
    const auto blocking_tasks = BlockWorkers(threadPool, blocker, NUM_WORKERS_DEFAULT);

    // Queue tasks while all workers are busy, then interrupt
    std::atomic<int> counter{0};
    const int num_tasks = 10;
    std::vector<std::future<void>> futures;
    futures.reserve(num_tasks);
    for (int i = 0; i < num_tasks; i++) {
        futures.emplace_back(Submit(threadPool, [&counter]{ counter.fetch_add(1, std::memory_order_relaxed); }));
    }
    threadPool.Interrupt();

    // Queued tasks must still complete despite the interrupt
    blocker.release(NUM_WORKERS_DEFAULT);
    WAIT_FOR(futures);
    BOOST_CHECK_EQUAL(counter.load(), num_tasks);

    threadPool.Stop();
    WAIT_FOR(blocking_tasks);
}

// Test 13, ensure the Stop() calling thread helps drain the queue
BOOST_AUTO_TEST_CASE(stop_active_wait_drains_queue)
{
    ThreadPool threadPool(POOL_NAME);
    threadPool.Start(NUM_WORKERS_DEFAULT);

    std::counting_semaphore<> blocker(0);
    const auto blocking_tasks = BlockWorkers(threadPool, blocker, NUM_WORKERS_DEFAULT);

    auto main_thread_id = std::this_thread::get_id();
    std::atomic<int> main_thread_tasks{0};
    const size_t num_tasks = 20;
    for (size_t i = 0; i < num_tasks; i++) {
        (void)Submit(threadPool, [&main_thread_tasks, main_thread_id]() {
            if (std::this_thread::get_id() == main_thread_id)
                main_thread_tasks.fetch_add(1, std::memory_order_relaxed);
        });
    }
    BOOST_CHECK_EQUAL(threadPool.WorkQueueSize(), num_tasks);