Wait until A job (as starkly opposed to ALL jobs) posted to boost::asio::thread_pool completes?

user_1885442 0
Viewed 46

I post multiple jobs to a boost::asio::thread_pool, and I want to process their results as soon as each individual result is available. I would 1000× prefer an "event loop" idiom for this over callback idiom, because the event loop idiom automatically does synchronization for me: only one thread consumes the results and writes them to an aggregate data structure without needing to synchronize access to it. The documentation is unclear on how to do this; the examples given use lots of difficult concepts and seem about 10× more complicated than a typical instance of the event loop idiom should be.

Is the event loop idiom supported by boost::asio at all?

threadpool boost-asio c++
1 Answers

Asio has the proactor model. Basically, a service is "run" by executing new handlers when they are ready. If you view a handler as an event, then you will see that you already have what you are after.

The pool operates on an event loop per thread. So, you have a task, and have it post the continuation event to the pool. Same guarantees that you describe.

Demo

The following code runs taskB that depends on taskA as soon as taskA completes. It runs four of these tasks in parallel:

Live On Coliru

// #define BOOST_ASIO_ENABLE_HANDLER_TRACKING
#include <boost/asio.hpp>
#include <random>
#include <iostream>
#include <iomanip>
using namespace std::literals;
auto now = std::chrono::high_resolution_clock::now;

static auto const start = now();
static void randelay()
{
    thread_local auto gen = std::bind(std::uniform_int_distribution<>(100, 800),
                                      std::mt19937{std::random_device{}()});
    std::this_thread::sleep_for(gen() * 1ms);
}

int main()
{
    auto taskB = [](int resultFromTaskA) {
        randelay();
        static std::mutex mx;
        std::lock_guard lk(mx);
        std::cout << "at " << std::setw(4) << (now() - start) / 1ms
                  << "ms taskB resultFromTaskA: " << resultFromTaskA
                  << std::endl;
    };

    boost::asio::thread_pool ctx;
    auto executor = ctx.get_executor();
    auto taskA    = [=](int payload) {
        randelay();
        post(executor, std::bind(taskB, payload*payload));
    };

    for (auto i = 1; i < 5; ++i)
        post(ctx, std::bind(taskA, i));

    ctx.join();
}

Prints e.g.

at  234ms taskB resultFromTaskA: 16
at  837ms taskB resultFromTaskA: 4
at 1214ms taskB resultFromTaskA: 1
at 1290ms taskB resultFromTaskA: 9

And with enabled handler tracking:

enter image description here

user_853710
Related