是否将std::packaged_task添加到现有线程

Add a std::packaged_task to an existing thread?

本文关键字：线程 添加 std packaged 是否 task      更新时间：2023-10-16

是否有标准的方法将std::packaged_task添加到现有线程？在运行任务之前，必须发生大量的开销，所以我想做一次，然后保持线程运行并等待任务执行。我希望能够使用futures，这样我就可以有选择地获得任务的结果并捕获异常。

我的C++11之前的实现要求我的任务从一个抽象基类继承，该基类具有Run()方法(有点麻烦，不能使用lambdas(，并且有一个std::deque集合，我将这些集合添加到主线程中，并从工作线程中出列。我必须保护该集合不被同时访问，并向工作线程提供一个信号，表明有什么事情要做，这样它就不会旋转或休眠。Enqueing返回一个"结果"对象和一个等待任务完成的同步对象，以及一个结果值。这一切都很好，但如果有更好的东西，是时候升级了。

这里有一个玩具线程池：

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;
};

复制自我的另一个答案。

一个有1个线程的thread_pool与您的描述非常匹配。

上面只是一个玩具，一个真正的线程池，我会用move_only_function<void()>代替std::packaged_task<void()>，这就是我使用它的全部目的。(packaged_task<void()>可以有趣地保持packaged_task<R()>，如果没有效率的话(。

你将不得不对关机进行推理并制定计划。如果您试图在不首先清除线程的情况下关闭上面的代码，那么它就会锁定。

相关文章：