如何创建一个保存 boost::packaged_task<> 的队列使用返回任意类型的函数？

Question

我正在尝试构建一个需要由一个线程执行并且可以由多个线程提供的函数的工作队列。为了实现这一目标，我计划使用 boost::packaged_task 和 boost::unique_future。这个想法是你会这样做：

Foo value =queue.add(myFunc).get();

这将阻塞，直到该函数被执行。所以queue.add(...)接受一个boost::function，并返回一个boost::unique_future。然后，它在内部使用 boost::function 作为其构造函数创建一个 boost::packaged_task 。

我遇到的问题是 boost::function<...>每次都不会一样。具体来说，它的返回值将会改变（但是，函数永远不会采用任何参数）。因此，我必须有一个看起来像这样的 add 函数：

template <typename ResultType>
boost::unique_future<ResultType> add(boost::function<ResultType ()> f) {
   boost::packaged_task<boost::function<ResultType ()> > task(f);
   queue.push_back(task);
   return task.get_future();
}

好吧，这看起来不太糟糕，但后来我遇到了如何定义“队列”的问题。我想我别无选择，只能使用 boost::any，因为类型不会是恒定的：

std::list<boost::any> queue; // note: I'm not concerned with thread-safety yet

但是当我尝试实现我的executeSingle（仅从队列中取出一个项目来执行）时，我遇到了问题

void executeSingle() {
    boost::any value = queue.back();
    boost::packaged_task<?> task = boost::packaged_task<?>(boost::move(value));
    // actually execute task
    task();
    queue.pop_back();
}

： '？'表示我不确定的事情。我无法使用模板调用executeSingle，因为它是从单独的线程调用的。我尝试使用 boost::any，但出现错误：

  conversion from 'boost::any' to non-scalar type  boost::detail::thread_move_t<boost:thread>' requested.

有趣的是，我实际上并不关心 packaged_task 的返回类型，我只想执行它，但我可以弄清楚模板详细信息。

任何见解将不胜感激！

Answer 1

您应该存储 boost::function 的。请注意，boost::packaged_task::operator() 不会返回任何内容；它填充关联的 boost::future。事实上，即使它返回了一些东西，你仍然可以使用 boost::function 因为你仍然对返回的值不感兴趣：你所关心的只是调用 <代码>queue.back()()。如果是这种情况，boost::function::operator() 会为您丢弃返回的值。

需要注意的是，您可能需要更改 add 方法的签名，以在泛型类型 Functor 而不是 boost::function，并使用 boost::result_of 获取 boost::packaged_task 的结果类型。

我的整体建议：

template<typename Functor>
boost::future<typename boost::result_of<Functor()>::type>
queue::add(Functor functor) // assuming your class is named queue
{
    typedef typename boost::result_of<Functor()>::type result_type;
    boost::packaged_task<result_type> task(functor);
    boost::unique_future<result_type> future = task.get_future();
    internal_queue.push_back(boost::move(task)); // assuming internal_queue member
    return boost::move(future);
}

void
queue::executeSingle()
{
    // Note: do you really want LIFO here?
    queue.back()();
    queue.pop_back();
}

编辑

如何处理queue::add内的移动语义

typedef typename boost::result_of<Functor()>::type result_type;
typedef boost::packaged_task<result_type> task_type;
boost::shared_ptr<task_type> task = boost::make_shared<task_type>(functor);
boost::unique_future<result_type> future = task->get_future();

/* boost::shared_ptr is possibly move-enabled so you can try moving it */
internal_queue.push_back( boost::bind(dereference_functor(), task) );

return boost::move(future);

，其中dereference_functor可能是：

struct dereference_functor {
    template<typename Pointer>
    void
    operator()(Pointer const& p) const
    {
        (*p)();
    }
};

您也可以替换bind 表达式更加清晰

boost::bind(&task_type::operator(), task)

，也不需要自定义函子。但是，如果 task_type::operator() 有多个重载，则可能需要消除歧义；如果 Boost.Thread 中的未来更改引入重载，代码也可能会中断。

Answer 2

您使用老式的虚拟函数。使用virtual execute 方法定义基类task_base，然后定义一个保存特定任务实例的模板派生类。大致意思是：

struct task_base {
  virtual void execute() = 0;
};
template<typename ResultType>
struct task_holder : task_base {
  task_holder(boost::packaged_task<boost::function<ResultType ()> >&& task)
    : m_task(task) { }
  void execute() {
    m_task();
  }
private:
  boost::packaged_task<boost::function<ResultType ()> > m_task;
};

并定义您的队列来保存unique_ptr。这本质上就是 boost::any 的作用，只是您会使用特定的函数，即 execute。

注意：未经测试的代码！而且我仍然不太熟悉右值引用。这只是为了让您了解代码的外观。

Answer 3

虽然有点晚了，但您可能需要考虑使用 Boost.Asio，而不是滚动自己的队列运行器解决方案。

虽然它是作为一个 I/O 库发展起来的，但它确实支持像这样的异步调用。只需在某处定义一个 io_service ，在线程内运行它，然后在该线程上调用 post 函子即可。