正确使用升压:: wait_for_any

Question

我有一个简单的任务调度：一个execute方法被调用，并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> >。这让我有两个想法。无论是由于复制，其异常类指示，但我不会看到这个occures，或由于get被已经调用，这不能发生，因为期货是只在这个方法块可见的，我已经他们转化为shared_futrues所以它应该工作。

那么在wait_for_any部分，我将如何取回指向已完成的Task实例的指针？使用future而不是shared_future。

原来我的任务之一execute函数中抛出了异常，期货将把这些异常带到get调用中。代码本身很好（除了缺少的异常处理程序）。然而使用升压信号（见下面的答案）可能是更好的方法。

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(); 
}