将 std::packaged_task 添加到现有线程?
Add a std::packaged_task to an existing thread?
是否有标准方法可以将 std::packaged_task
添加到现有线程?在任务 运行 之前必须发生大量的开销,所以我想这样做一次,然后保持线程 运行ning 并等待任务执行。我希望能够使用 futures,这样我就可以有选择地获取任务的结果并捕获异常。
我的 C++11 之前的实现要求我的任务使用 Run()
方法从抽象基础 class 继承(有点痛苦,不能使用 lambda),并且有一个 std::deque
集合,其中包含我在主线程中添加并从工作线程中出队的那些。我必须保护该集合不被同时访问,并向工作线程提供一个信号,表明有事情要做,这样它就不会旋转或休眠。 Enqueing something returns 一个 "result" 对象和一个等待任务完成的同步对象,以及一个结果值。一切正常,但如果有更好的东西,是时候升级了。
这是一个玩具线程池:
template<class T>
struct threaded_queue {
using lock = std::unique_lock<std::mutex>;
void push_back( T t ) {
{
lock l(m);
data.push_back(std::move(t));
}
cv.notify_one();
}
boost::optional<T> pop_front() {
lock l(m);
cv.wait(l, [this]{ return abort || !data.empty(); } );
if (abort) return {};
auto r = std::move(data.back());
data.pop_back();
return std::move(r);
}
void terminate() {
{
lock l(m);
abort = true;
data.clear();
}
cv.notify_all();
}
~threaded_queue()
{
terminate();
}
private:
std::mutex m;
std::deque<T> data;
std::condition_variable cv;
bool abort = false;
};
struct thread_pool {
thread_pool( std::size_t n = 1 ) { start_thread(n); }
thread_pool( thread_pool&& ) = delete;
thread_pool& operator=( thread_pool&& ) = delete;
~thread_pool() = default; // or `{ terminate(); }` if you want to abandon some tasks
template<class F, class R=std::result_of_t<F&()>>
std::future<R> queue_task( F task ) {
std::packaged_task<R()> p(std::move(task));
auto r = p.get_future();
tasks.push_back( std::move(p) );
return r;
}
template<class F, class R=std::result_of_t<F&()>>
std::future<R> run_task( F task ) {
if (threads_active() >= total_threads()) {
start_thread();
}
return queue_task( std::move(task) );
}
void terminate() {
tasks.terminate();
}
std::size_t threads_active() const {
return active;
}
std::size_t total_threads() const {
return threads.size();
}
void clear_threads() {
terminate();
threads.clear();
}
void start_thread( std::size_t n = 1 ) {
while(n-->0) {
threads.push_back(
std::async( std::launch::async,
[this]{
while(auto task = tasks.pop_front()) {
++active;
try{
(*task)();
} catch(...) {
--active;
throw;
}
--active;
}
}
)
);
}
}
private:
std::vector<std::future<void>> threads;
threaded_queue<std::packaged_task<void()>> tasks;
std::atomic<std::size_t> active;
};
从我的 another answer 复制而来。
有 1 个线程的 thread_pool
与您的描述非常匹配。
以上只是一个玩具,一个真正的线程池我会用move_only_function<void()>
替换std::packaged_task<void()>
,这就是我使用它的全部。 (如果效率低下,packaged_task<void()>
可以有趣地容纳 packaged_task<R()>
)。
您将不得不对关机进行推理并制定计划。如果您在未先清除线程的情况下尝试将其关闭,则上述代码将被锁定。
是否有标准方法可以将 std::packaged_task
添加到现有线程?在任务 运行 之前必须发生大量的开销,所以我想这样做一次,然后保持线程 运行ning 并等待任务执行。我希望能够使用 futures,这样我就可以有选择地获取任务的结果并捕获异常。
我的 C++11 之前的实现要求我的任务使用 Run()
方法从抽象基础 class 继承(有点痛苦,不能使用 lambda),并且有一个 std::deque
集合,其中包含我在主线程中添加并从工作线程中出队的那些。我必须保护该集合不被同时访问,并向工作线程提供一个信号,表明有事情要做,这样它就不会旋转或休眠。 Enqueing something returns 一个 "result" 对象和一个等待任务完成的同步对象,以及一个结果值。一切正常,但如果有更好的东西,是时候升级了。
这是一个玩具线程池:
template<class T>
struct threaded_queue {
using lock = std::unique_lock<std::mutex>;
void push_back( T t ) {
{
lock l(m);
data.push_back(std::move(t));
}
cv.notify_one();
}
boost::optional<T> pop_front() {
lock l(m);
cv.wait(l, [this]{ return abort || !data.empty(); } );
if (abort) return {};
auto r = std::move(data.back());
data.pop_back();
return std::move(r);
}
void terminate() {
{
lock l(m);
abort = true;
data.clear();
}
cv.notify_all();
}
~threaded_queue()
{
terminate();
}
private:
std::mutex m;
std::deque<T> data;
std::condition_variable cv;
bool abort = false;
};
struct thread_pool {
thread_pool( std::size_t n = 1 ) { start_thread(n); }
thread_pool( thread_pool&& ) = delete;
thread_pool& operator=( thread_pool&& ) = delete;
~thread_pool() = default; // or `{ terminate(); }` if you want to abandon some tasks
template<class F, class R=std::result_of_t<F&()>>
std::future<R> queue_task( F task ) {
std::packaged_task<R()> p(std::move(task));
auto r = p.get_future();
tasks.push_back( std::move(p) );
return r;
}
template<class F, class R=std::result_of_t<F&()>>
std::future<R> run_task( F task ) {
if (threads_active() >= total_threads()) {
start_thread();
}
return queue_task( std::move(task) );
}
void terminate() {
tasks.terminate();
}
std::size_t threads_active() const {
return active;
}
std::size_t total_threads() const {
return threads.size();
}
void clear_threads() {
terminate();
threads.clear();
}
void start_thread( std::size_t n = 1 ) {
while(n-->0) {
threads.push_back(
std::async( std::launch::async,
[this]{
while(auto task = tasks.pop_front()) {
++active;
try{
(*task)();
} catch(...) {
--active;
throw;
}
--active;
}
}
)
);
}
}
private:
std::vector<std::future<void>> threads;
threaded_queue<std::packaged_task<void()>> tasks;
std::atomic<std::size_t> active;
};
从我的 another answer 复制而来。
有 1 个线程的 thread_pool
与您的描述非常匹配。
以上只是一个玩具,一个真正的线程池我会用move_only_function<void()>
替换std::packaged_task<void()>
,这就是我使用它的全部。 (如果效率低下,packaged_task<void()>
可以有趣地容纳 packaged_task<R()>
)。
您将不得不对关机进行推理并制定计划。如果您在未先清除线程的情况下尝试将其关闭,则上述代码将被锁定。