带线程超时的Boost ASIO线程池

这是我想出的一个快速解决方案。

它要求你提交的函数对象接受一个类型为exec_context的参数。

在io_service中运行的任务可以查询.canceled()访问器(这是原子的)，以确定是否应该提前取消。

然后，它可以抛出异常或返回任何它想返回的值。

主叫通过submit函数提交。该函数用context对象包装worker函数，并将其返回值和/或异常封送到std::future中。

调用者可以根据需要查询或等待这个未来(或忽略它)。

调用者获得一个句柄对象，该对象具有cancel()方法。使用此句柄，调用者可以取消、查询或等待提交的任务。

希望有帮助。写起来很有趣。

#include <boost/asio.hpp>
#include <iostream>
#include <atomic>
#include <thread>
#include <chrono>
#include <future>
#include <stdexcept>
#include <exception>
#include <utility>
#include <type_traits>

//
// an object to allow the caller to communicate a cancellation request to the
// submitted task
//
struct exec_controller
{
    /// @returns previous cancellation request state;
    bool notify_cancel()
    {
        return _should_cancel.exchange(true);
    }
    bool should_cancel() const {
        return _should_cancel;
    }
private:
    std::atomic<bool> _should_cancel = { false };
};
template<class Ret>
struct exec_state : std::enable_shared_from_this<exec_state<Ret>>
{
    using return_type = Ret;
    bool notify_cancel() {
        return _controller.notify_cancel();
    }
    std::shared_ptr<exec_controller>
    get_controller_ptr() {
        return std::shared_ptr<exec_controller>(this->shared_from_this(),
                                                std::addressof(_controller));
    }
    std::promise<return_type>& promise() { return _promise; }
private:
    std::promise<return_type> _promise;
    exec_controller _controller;
};
struct applyer;
struct exec_context
{
    exec_context(std::shared_ptr<exec_controller> impl)
    : _impl(impl)
    {}
    bool canceled() const {
        return _impl->should_cancel();
    }
private:
    friend applyer;
    std::shared_ptr<exec_controller> _impl;
};
struct applyer
{
    template<class F, class Ret>
    void operator()(F& f, std::shared_ptr<exec_state<Ret>> const& p) const
    {
        try {
            p->promise().set_value(f(exec_context { p->get_controller_ptr() }));
        }
        catch(...) {
            p->promise().set_exception(std::current_exception());
        }
    }
    template<class F>
    void operator()(F& f, std::shared_ptr<exec_state<void>> const& p) const
    {
        try {
            f(exec_context { p->get_controller_ptr() });
            p->promise().set_value();
        }
        catch(...) {
            p->promise().set_exception(std::current_exception());
        }
    }
};
template<class Ret>
struct exec_result
{
    using return_type = Ret;
    exec_result(std::shared_ptr<exec_state<return_type>> p)
    : _impl(p)
    {}
    bool cancel() {
        return _impl->notify_cancel();
    }
    std::future<Ret>& get_future()
    {
        return _future;
    }
private:
    std::shared_ptr<exec_state<return_type>> _impl;
    std::future<return_type> _future { _impl->promise().get_future() };
};

template<class Executor, class F>
auto submit(Executor& exec, F&& f)
{
    using function_type = std::decay_t<F>;
    using result_type = std::result_of_t<function_type(exec_context)>;
    using state_type = exec_state<result_type>;
    auto shared_state = std::make_shared<state_type>();
    exec.post([shared_state, f = std::forward<F>(f)]
              {
                  applyer()(f, shared_state);
              });
    return exec_result<result_type>(std::move(shared_state));
}

int main()
{
    using namespace std::literals;
    boost::asio::io_service ios;
    boost::asio::io_service::strand strand(ios);
    boost::asio::io_service::work work(ios);
    std::thread runner([&] { ios.run(); });
    std::thread runner2([&] { ios.run(); });
    auto func = [](auto context)
    {
        for(int i = 0 ; i < 1000 ; ++i)
        {
            if (context.canceled())
                throw std::runtime_error("canceled");
            std::this_thread::sleep_for(100ms);
        }
    };
    auto handle = submit(strand, func);
    auto handle2 = submit(ios, [](auto context) { return 2 + 2; });
    // cancel the handle, or wait on it as you wish
    std::this_thread::sleep_for(1s);
    handle.cancel();
    handle2.cancel();   // prove that late cancellation is a nop
    try {
        std::cout << "2 + 2 is " << handle2.get_future().get() << std::endl;
    }
    catch(std::exception& e)
    {
        std::cerr << "failed to add 2 + 2 : " << e.what() << std::endl;
    }
    try {
        handle.get_future().get();
        std::cout << "task completed" << std::endl;
    }
    catch(std::exception const& e) {
        std::cout << "task threw exception: " << e.what() << std::endl;
    }
    ios.stop();
    runner.join();
    runner2.join();
}

update: v2为类添加了一些隐私保护，演示了2个并发任务。

2 + 2 is 4
task threw exception: canceled