c++ - 将参数转发给工作队列函数时出现编译错误

Question

事先，我从 here 获得了基本代码，感谢他们在这里的工作。基本想法是有一个工作池，您可以向其提交任意函数

int main(){
    WorkerQueue q;
    auto job = [](){std::cout << "yay\n";};
    q.submit(job)
    q.wait_for_completion();
    return 0;
}

该实现无法处理job 的输入参数。因此，我想进行扩展，使其按如下方式工作:

int
main()
{

  auto job2 = [](int yay) {
    std::cout << yay << '\n';
    return;
  };

  WorkQueue q;

  int arg = 3;
  q.submit(job2, arg);
  q.wait_for_completion();

  return 0;
}

这是我的实现:

#include <iostream>

#include <cassert>
#include <deque>
#include <functional>
#include <future>
#include <thread>
#include <vector>


// Source:
// https://codereview.stackexchange.com/questions/60363/thread-pool-worker-implementation

template<typename ReturnType, typename... Args>
using PromiseFunctionPair =
  std::pair<std::promise<ReturnType>, std::function<ReturnType(Args...)>>;

template<typename ReturnType, typename... Args>
using DataPointer = std::shared_ptr<PromiseFunctionPair<ReturnType, Args...>>;

/** A typical thread worker queue that can execute arbitrary jobs.
 *
 */
class WorkQueue
{

private:
  std::deque<std::function<void()>> m_work;
  std::mutex m_mutex;
  std::condition_variable m_signal;
  std::atomic<bool> m_exit{ false };
  std::atomic<bool> m_finish_work{ true };
  std::vector<std::thread> m_workers;

  void do_work();
  void join_all();

  void operator=(const WorkQueue&) = delete;
  WorkQueue(const WorkQueue&) = delete;

public:
  explicit WorkQueue(int64_t numWorkers = -1);
  virtual ~WorkQueue();
  void abort();
  void stop();
  void wait_for_completion();

  // template<typename ReturnType, typename ...Args>
  // std::future<ReturnType> submit(std::function<ReturnType(Args...)>&&
  // function, Args && ...args);
  template<typename FunctionObject, typename... Args>
  auto submit(FunctionObject&& function, Args&&... args)
    -> std::future<decltype(function(args...))>;
};

/** Executes the given function asynchronously.
 *
 * @param function : the function to execute
 * @return result : future of result that will be generated by
 *                  the function argument. Exceptions from the
 *                  function will be thrown by get() on the future.
 *
 * @throw std::runtime_error : if attempting to submit a job
 *                             to a work queue that is terminating
 */
/*
template<typename ReturnType, typename ...Args>
std::future<ReturnType> WorkQueue::submit(std::function<ReturnType(Args...)>&&
function, Args && ...args)
*/
template<typename FunctionObject, typename... Args>
auto
WorkQueue::submit(FunctionObject&& function, Args&&... args)
  -> std::future<decltype(function(args...))>
{
  using ReturnType = decltype(function(args...));

  if (m_exit) {
    throw std::runtime_error(
      "Caught work submission to work queue that is desisting.");
  }

  // Workaround for lack of lambda move capture
  // auto data =
  //  std::make_shared<PromiseFunctionPair<ReturnType,
  //  Args...>>(std::promise<ReturnType>(), std::move(function));

  auto data = std::make_shared<
    std::pair<std::promise<ReturnType>, std::function<ReturnType(Args...)>>>(
    std::promise<ReturnType>(), std::move(function));

  std::future<ReturnType> future = data->first.get_future();

  {
    std::lock_guard<std::mutex> lg(m_mutex);
    m_work.emplace_back([data, args...]() {
      try {

        if constexpr (std::is_same<ReturnType, void>::value) {
          data->second(std::forward<Args>(args)...);
          data->first.set_value();
        } else {
          data->first.set_value(data->second(std::forward<Args>(args)...));
        }

      } catch (...) {
        data->first.set_exception(std::current_exception());
      }
    });
  }
  m_signal.notify_one();
  return std::move(future);
}

/**  Constructors a new work queue object
 *
 * @param num_workers : number of workers. If <1 all cores are used.
 */
WorkQueue::WorkQueue(int64_t num_workers)
{
  if (num_workers < 1) {
    num_workers = std::thread::hardware_concurrency() + 1;
  }
  while (num_workers--) {
    m_workers.emplace_back(std::thread(&WorkQueue::do_work, this));
  }
}

/** Will abort all pending jobs and run any in-progress jobs to completion
 * upon destruction.
 *
 */
WorkQueue::~WorkQueue()
{
  abort();
}

/** worker thread function, for picking jobs.
 *
 */
void
WorkQueue::do_work()
{
  std::unique_lock<std::mutex> ul(m_mutex);
  while (!m_exit || (m_finish_work && !m_work.empty())) {
    if (!m_work.empty()) {
      std::function<void()> work(std::move(m_work.front()));
      m_work.pop_front();
      ul.unlock();
      work();
      ul.lock();
    } else {
      m_signal.wait(ul);
    }
  }
}

/** Waits for all threads to finish
 *
 */
void
WorkQueue::join_all()
{
  for (auto& thread : m_workers) {
    thread.join();
  }
  m_workers.clear();
}

/** Stops queue and jobs
 *
 * Stops work queue and finishes jobs currently being executed.
 * Queued jobs that have not begun execution will have their promises
 * broken.
 *
 */
void
WorkQueue::abort()
{
  m_exit = true;
  m_finish_work = false;
  m_signal.notify_all();
  join_all();
  {
    std::lock_guard<std::mutex> lg(m_mutex);
    m_work.clear();
  }
}

/** Stops new work from being submitted to this work queue
 *
 */
void
WorkQueue::stop()
{
  m_exit = true;
  m_finish_work = true;
  m_signal.notify_all();
}

/**  Wait for completion of all running jobs. No more work will done.
 *
 */
void
WorkQueue::wait_for_completion()
{
  stop();
  join_all();
  assert(m_work.empty());
}

问题出在这一行:

data->second(std::forward<Args>(args)...);

使用 g++-7.3 我得到了漂亮而经典的模板错误输出:

<source>: In instantiation of 'WorkQueue::submit(FunctionObject&&, Args&& ...)::<lambda()> [with FunctionObject = main()::<lambda(int)>&; Args = {int&}]':
<source>:98:40:   required from 'struct WorkQueue::submit(FunctionObject&&, Args&& ...) [with FunctionObject = main()::<lambda(int)>&; Args = {int&}; decltype (function(WorkQueue::submit::args ...)) = void]::<lambda()>'
<source>:98:9:   required from 'std::future<decltype (function(WorkQueue::submit::args ...))> WorkQueue::submit(FunctionObject&&, Args&& ...) [with FunctionObject = main()::<lambda(int)>&; Args = {int&}; decltype (function(WorkQueue::submit::args ...)) = void]'
<source>:232:21:   required from here
<source>:102:46: error: no matching function for call to 'forward<int&>(const int&)'
               data->second(std::forward<Args>(args)...);
                            ~~~~~~~~~~~~~~~~~~^~~~~~
In file included from /opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/bits/nested_exception.h:40:0,
                 from /opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/exception:143,
                 from /opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/ios:39,
                 from /opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/ostream:38,
                 from /opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/iostream:39,
                 from <source>:3:
/opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/bits/move.h:73:5: note: candidate: template<class _Tp> constexpr _Tp&& std::forward(typename std::remove_reference<_From>::type&)
     forward(typename std::remove_reference<_Tp>::type& __t) noexcept
     ^~~~~~~
/opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/bits/move.h:73:5: note:   template argument deduction/substitution failed:
<source>:102:46: note:   cannot convert '((const WorkQueue::submit(FunctionObject&&, Args&& ...) [with FunctionObject = main()::<lambda(int)>&; Args = {int&}; decltype (function(WorkQueue::submit::args ...)) = void]::<lambda()>*)__closure)->WorkQueue::submit(FunctionObject&&, Args&& ...) [with FunctionObject = main()::<lambda(int)>&; Args = {int&}; decltype (function(WorkQueue::submit::args ...)) = void]::<lambda()>::<args#0 capture>' (type 'const int') to type 'std::remove_reference<int&>::type& {aka int&}'
               data->second(std::forward<Args>(args)...);
                            ~~~~~~~~~~~~~~~~~~^~~~~~
In file included from /opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/bits/nested_exception.h:40:0,
                 from /opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/exception:143,
                 from /opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/ios:39,
                 from /opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/ostream:38,
                 from /opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/iostream:39,
                 from <source>:3:
/opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/bits/move.h:84:5: note: candidate: template<class _Tp> constexpr _Tp&& std::forward(typename std::remove_reference<_From>::type&&)
     forward(typename std::remove_reference<_Tp>::type&& __t) noexcept
     ^~~~~~~
/opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/bits/move.h:84:5: note:   template argument deduction/substitution failed:
<source>:102:46: note:   cannot convert '((const WorkQueue::submit(FunctionObject&&, Args&& ...) [with FunctionObject = main()::<lambda(int)>&; Args = {int&}; decltype (function(WorkQueue::submit::args ...)) = void]::<lambda()>*)__closure)->WorkQueue::submit(FunctionObject&&, Args&& ...) [with FunctionObject = main()::<lambda(int)>&; Args = {int&}; decltype (function(WorkQueue::submit::args ...)) = void]::<lambda()>::<args#0 capture>' (type 'const int') to type 'std::remove_reference<int&>::type&& {aka int&&}'
               data->second(std::forward<Args>(args)...);
                            ~~~~~~~~~~~~~~~~~~^~~~~~
In file included from /opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/deque:66:0,
                 from <source>:6:
/opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/bits/deque.tcc:161:7: error: 'std::deque<_Tp, _Alloc>::reference std::deque<_Tp, _Alloc>::emplace_back(_Args&& ...) [with _Args = {WorkQueue::submit(FunctionObject&&, Args&& ...) [with FunctionObject = main()::<lambda(int)>&; Args = {int&}; decltype (function(WorkQueue::submit::args ...)) = void]::<lambda()>}; _Tp = std::function<void()>; _Alloc = std::allocator<std::function<void()> >; std::deque<_Tp, _Alloc>::reference = std::function<void()>&]', declared using local type 'WorkQueue::submit(FunctionObject&&, Args&& ...) [with FunctionObject = main()::<lambda(int)>&; Args = {int&}; decltype (function(WorkQueue::submit::args ...)) = void]::<lambda()>', is used but never defined [-fpermissive]
       deque<_Tp, _Alloc>::
       ^~~~~~~~~~~~~~~~~~
/opt/compiler-explorer/gcc-7.3.0/include/c++/7.3.0/bits/deque.tcc:161:7: warning: 'std::deque<_Tp, _Alloc>::reference std::deque<_Tp, _Alloc>::emplace_back(_Args&& ...) [with _Args = {WorkQueue::submit(FunctionObject&&, Args&& ...) [with FunctionObject = main()::<lambda(int)>&; Args = {int&}]::<lambda()>}; _Tp = std::function<void()>; _Alloc = std::allocator<std::function<void()> >]' used but never defined
Compiler returned: 1

经过一段时间的努力，我感觉很接近了，但无法解决最后一个问题。直到现在我都不是可变参数模板专家，我因为无法处理这个而感到非常愚蠢......

Answer 1

问题是当您通过 by copy 您的可变参数 args... 在可变参数模板内的 lambda 中传递参数时 提交() 方法

//  by copy now ...........vvvvvvv
m_work.emplace_back([data, args...]() {

在 lambda 内部你不能再使用完美转发

      data->second(std::forward<Args>(args)...);

因为 args... 不再是通用引用。

据我所知，解决这个问题的正确方法是通过引用将 args... 传递给 lambda

//      add this ----------v
m_work.emplace_back([data, &args...]() {

这种方式将 args... 维护为 lambda 中的通用引用，以便您能够转发它们。