正确使用 boost::wait_for_any

Question

我有一个简单的任务调度程序：调用执行方法，packaged_task 返回指向我的Task 的指针。任务完成后，我想显示调试数据（涉及 GUI，因此我需要在主线程中执行此操作）。我想为此使用boost::wait_for_any，但j->get() 有时会抛出异常boost::exception_detail::clone_impl<boost::exception_detail::error_info_injector<boost::promise_already_satisfied> >。这让我有两个想法。要么是由于异常类指示的复制，但我看不到这种情况发生的位置，或者是由于 get 已经被调用，这不可能发生，因为期货仅在此方法块中可见并且我已经转换了它们到 shared_futrues 所以它应该可以工作。

那么在wait_for_any 部分，我将如何取回指向已完成的Task 实例的指针？

编辑使用future而不是shared_future。

原来异常是在我的一项任务的 execute 函数中引发的，futures 会将这些异常传递给 get 调用。代码本身很好（除了缺少的异常处理程序）。但是使用 Boost Signals（见下面的答案）可能是更好的方法。

std::vector<boost::future<Task*>> futures;
std::vector<Task*> tasks = get_tasks();
for (Task* t : tasks) {
    typedef boost::packaged_task<Task*()> task_t;
    task_t task([t]() -> Task* {
        t->execute();
        return t;
    });

    auto fut = task.get_future();
    futures.push_back(std::move(fut));

    impl->io_service.post(boost::bind(&task_t::operator(), boost::make_shared<task_t>(std::move(task))));
}


for (Task* t : tasks) {
    auto j = boost::wait_for_any(futures.begin(), futures.end());
    Task* task = j->get();
    task->display_debug();
    futures.erase(j);
}

Answer 1

嗯。我在这里有点迷失了方向。看来您做的事情比要求的要复杂一些（为什么不使用 Boost Signals2 而不是通过期货对特定“事件”进行“轮询”？看来您无论如何都不期望它们以任何特定的顺序出现？）。

不管怎样，这里有一个适合我的固定版本。如果我以后有更多时间，我可能会比较笔记，看看是什么解释了差异。

看Live On Coliru

#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/make_shared.hpp>
#include <boost/thread/future.hpp>
#include <iostream>
#include <string>
#include <set>

struct Task
{
    virtual ~Task() = default;
    virtual void execute() {}
    virtual void display_debug() { std::cout << __FUNCTION__ << static_cast<void*>(this) << "\n"; }
};

std::set<Task*> get_tasks()
{
    static std::set<Task*> data{ new Task, new Task };

    return data;
}

int main() {
    struct { boost::asio::io_service io_service; } impl_instance;
    auto impl = &impl_instance;

    std::vector<boost::shared_future<Task*>> futures;
    std::set<Task*> tasks = get_tasks();

    for (Task* t : tasks) {
        typedef boost::packaged_task<Task*> task_t;
        task_t wrap([t]() -> Task* {
            t->execute();
            return t;
        });

        auto fut = wrap.get_future();
        futures.push_back(std::move(fut));

        impl->io_service.post(boost::bind(&task_t::operator(), boost::make_shared<task_t>(std::move(wrap))));
    }

    boost::thread th([&]{ impl->io_service.run(); });

    while (!futures.empty()) {
        auto j = boost::wait_for_any(futures.begin(), futures.end());
        auto fut = *j;
        futures.erase(j);

        Task* task = fut.get();
        task->display_debug();

        // optionally:
        // tasks.erase(task);
    }

    th.join();
}