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

Code before:
};

// Test Cases Overview
// 0) Submit task to a non-started pool.
// 1) Submit tasks and verify completion.
// 2) Maintain all threads busy except one.
// 3) Wait for work to finish.
// 4) Wait for result object.
// 5) The task throws an exception, catch must be done in the consumer side.
// 6) Busy workers, help them by processing tasks externally.
// 7) Recursive submission of tasks.
// 8) Submit task when all threads are busy, stop pool and verify task gets executed.
// 9) Congestion test; create more workers than available cores.
// 10) Ensure Interrupt() prevents further submissions.
BOOST_FIXTURE_TEST_SUITE(threadpool_tests, ThreadPoolFixture)

#define WAIT_FOR(futures)                                                         \
    do {                                                                          \
        for (const auto& f : futures) {                                           \
            BOOST_REQUIRE(f.wait_for(WAIT_TIMEOUT) == std::future_status::ready); \
        }                                                                         \
    } 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)
{
    // Per-thread ready promises to ensure all workers are actually blocked
    std::vector<std::promise<void>> ready_promises(num_of_threads_to_block);
    std::vector<std::future<void>> ready_futures;
    ready_futures.reserve(num_of_threads_to_block);
    for (auto& p : ready_promises) ready_futures.emplace_back(p.get_future());

    // Fill all workers with blocking tasks
    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(Submit(threadPool, [blocker_future, &ready]() {
            ready.set_value();
            blocker_future.wait();
        }));
    }

    // Wait until all threads are actually blocked

Review comment (bug):
Nice helper! Was actually experimenting with changing `ThreadPool::Submit()` to return `util::Expected` yesterday (suffering from lack of knowledge of `std::invoke_result_t<F>` though). It was *so* cumbersome to adapt these test without realizing a helper would be useful. :)
 
nit: The going style seems to be `[[nodiscard]]` on the same line as the return type:

```C++
template <typename F>
[[nodiscard]] auto Submit(ThreadPool& pool, F&& fn)
```

Examples from master:
https://github.com/bitcoin/bitcoin/blob/07b924775e4ff5dbfb9f2e7549fd13db8ec5c61b/src/util/overflow.h#L14-L15
https://github.com/bitcoin/bitcoin/blob/07b924775e4ff5dbfb9f2e7549fd13db8ec5c61b/src/test/fuzz/util.h#L84-L85

Same style nit goes for `ThreadPool::Submit()`.


Code after:
};

// Test Cases Overview
// 0) Submit task to a non-started pool.
// 1) Submit tasks and verify completion.
// 2) Maintain all threads busy except one.
// 3) Wait for work to finish.
// 4) Wait for result object.
// 5) The task throws an exception, catch must be done in the consumer side.
// 6) Busy workers, help them by processing tasks externally.
// 7) Recursive submission of tasks.
// 8) Submit task when all threads are busy, stop pool and verify task gets executed.
// 9) Congestion test; create more workers than available cores.
// 10) Ensure Interrupt() prevents further submissions.
BOOST_FIXTURE_TEST_SUITE(threadpool_tests, ThreadPoolFixture)

#define WAIT_FOR(futures)                                                         \
    do {                                                                          \
        for (const auto& f : futures) {                                           \
            BOOST_REQUIRE(f.wait_for(WAIT_TIMEOUT) == std::future_status::ready); \
        }                                                                         \
    } 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)
{
    // Per-thread ready promises to ensure all workers are actually blocked
    std::vector<std::promise<void>> ready_promises(num_of_threads_to_block);
    std::vector<std::future<void>> ready_futures;
    ready_futures.reserve(num_of_threads_to_block);
    for (auto& p : ready_promises) ready_futures.emplace_back(p.get_future());

    // Fill all workers with blocking tasks
    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(Submit(threadPool, [blocker_future, &ready]() {
            ready.set_value();
            blocker_future.wait();
        }));
    }

    // Wait until all threads are actually blocked