C++11线程池——带输入参数的任务

Question

我正在尝试使用 Anthony Williams “C++ Concurrency in Action”一书中的一个简单线程池示例。我什至在其中一篇文章中找到了代码（类 thread_pool）： Synchronizing tasks 但我有一个不同的问题。我想用以下签名向队列提交一个任务（一个成员函数）：

class A;
class B;
bool MyClass::Func(A*, B*); 

我需要如何更改 thread_pool 类，或者如何将我的函数打包到一些 void F() 中，假设它被用作任务在这个例子中？ 这是对我来说最相关的课程部分（详细信息请参见上面的链接）：

class thread_pool 
{
  thread_safe_queue<std::function<void()> work_queue; // bool MyClass::Func(a,b) ??

  void worker_thread() {
   while(!done) {         
    std::function<void()> task;
    if(work_queue.try_pop(task)) {
     task();  // how should my function MyClass::Func(a,b) be called here?                    
    }
    else {
     std::this_thread::yield();
    }
   }
  }

  // -- Submit a task to the thread pool
  template <typename FunctionType>
  void submit(FunctionType f) {
  work_queue.push(std::function<void()>(f)); // how should bool MyClassFunc(A*, B*) be submitted here
 }

}

最后，如何在我的代码中调用提交函数？

非常感谢您的帮助（不幸的是，我在使用所有 C++11 功能方面还不是很有经验，这可能也是我在这里需要帮助的原因，但这个问题的答案将是开始:))。

Answer 1

当您将任务插入队列时，您必须将参数绑定到一个值。这意味着您必须为您的函数创建一个包装器，该包装器存储this 的值和两个函数参数的值。有很多方法可以做到这一点，例如lambda 函数或std::bind。

work_queue.push_back( [obj, a, b]() {obj->Func(a,b)} );
work_queue.push_back( std::bind(&MyClass::Func, obj, a, b) );

您的提交函数必须采用这些参数并创建绑定，例如

template<typename F, typename... Args>
void submit(F f, Args&&... args) {
    work_queue.push_back( std::bind(f, std::forward<Args>(args)...) );
}

为成员函数和对象创建一个特殊的重载可能会很方便。

Answer 2

我以前写过一些（非常）类似的东西。我把代码贴在这里，你可以看看。 GenCmd 是函数包装器。队列看起来像这样，并在 Impl 中使用/定义（代码省略）。您只需要查看 GenCmd 的实现，因为它包含必要的工作。

ConcurrentQueue<std::unique_ptr<Cmd>> cqueue_;

我已将 std::function 包装为队列中的多态。 std_utility 包含 make_index_sequence，用于从元组中提取值（谷歌 make_index_sequence 以在某处找到实现，如果这还不是您的 std 库的一部分）。

#include <functional>
#include <memory>
#include <iostream>
#include <utility>
#include <boost/noncopyable.hpp>

class CmdExecutor : public boost::noncopyable
{
  public:
    CmdExecutor(std::ostream& errorOutputStream);
    ~CmdExecutor();

    template <class Receiver, class ... FArgs, class ... CArgs >
      void process(Receiver& receiver, void (Receiver::*f)(FArgs...), CArgs&&... args)
    {
      process(std::unique_ptr<Cmd>(new GenCmd<void(Receiver,FArgs...)>(f, receiver, std::forward<CArgs>(args)...)));
    }

  private:
    class Cmd
    {
      public:
        virtual void execute() = 0;
        virtual ~Cmd(){}
    };

    template <class T> class GenCmd;

    template <class Receiver, class ... Args>
    class GenCmd<void(Receiver, Args...)> : public Cmd
    {
      public:
        template <class FuncT, class ... CArgs>
        GenCmd(FuncT&& f, Receiver& receiver, CArgs&&... args)
          : call_(std::move(f)),
            receiver_(receiver),
            args_(args...)
        {
        }
        //We must convert references to values...

        virtual void execute()
        {
          executeImpl(std::make_index_sequence<sizeof...(Args)>{});
        }

      private:
        template <std::size_t ... Is>
        void executeImpl(std::index_sequence<Is...>)
        {
          // We cast the values in the tuple to the original type (of Args...)
          call_(receiver_, static_cast<Args>(std::get<Is>(args_))...);
        }

        std::function<void(Receiver&, Args...)> call_;
        Receiver& receiver_;
        // NOTE:
        // References converted to values for safety sake, as they are likely
        // to not be around when this is executed in other context.
        std::tuple<typename std::remove_reference<Args>::type...> args_;
    };

    void process(std::unique_ptr<Cmd> command);

    class Impl;
    Impl* pimpl_;
};

基本用法如下：

...
CmdExecutor context_;
...

void MyClass::myFunction()
{
  ArgX x;
  ArgY y;

  context_.process(*this, &MyClass::someFunction, x, y);
}

从中可以看出，进程对成员函数类型进行了包装，并将其转换为底层类型以存储在队列中。这允许多种参数类型。我选择使用运行时多态性来存储函数类型，因此使用 GenCmd 派生。

注意：如果被调用的函数接收到一个右值 (Arg&&)，则存储的类型将被转换为原始类型，因此会导致移动，并将适用的命令参数（只会被调用一次）呈现为空（这就是意图，至少 - 未经测试...）