将参数转发给工作队列函数时出现编译错误
Compilation Error while forwarding arguments to a worker queue function
之前，我已经从 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
经过一段时间的努力，我感觉很接近了，但无法解决最后一个问题。直到现在我都不是可变参数模板专家，我因为无法处理这个而感到非常愚蠢......
问题是当您通过 by copy 可变参数模板 args... 中的 lambda 参数时 submit() 方法
// 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 中的通用引用进行维护，以便您能够转发它们。
之前，我已经从 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
经过一段时间的努力，我感觉很接近了，但无法解决最后一个问题。直到现在我都不是可变参数模板专家，我因为无法处理这个而感到非常愚蠢......
问题是当您通过 by copy 可变参数模板 args... 中的 lambda 参数时 submit() 方法
// 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 中的通用引用进行维护，以便您能够转发它们。