使用 Boost 线程和 io_service 创建线程池答案

【问题标题】：Using Boost threads and io_service to create a threadpool使用 Boost 线程和 io_service 创建线程池
【发布时间】：2014-04-25 01:55:58
【问题描述】：

我环顾了 Stack Overflow，对此有一些非常好的答案，（我的代码实际上基于 this answer here）但由于某种原因，我的行为很奇怪——应该调用 ls1 次 thread_func ，但在线程退出之前它只运行了 0 到 2 次。似乎 ioService.stop() 在完成之前切断了排队的作业，但据我所知，这不应该发生。下面是相关代码sn-p：

boost::asio::io_service ioService;
boost::asio::io_service::work work(ioService);

boost::thread_group threadpool;

for (unsigned t = 0; t < num_threads; t++)
{   
    threadpool.create_thread(boost::bind(&boost::asio::io_service::run, &ioService));
}   

//Iterate over the dimensions of the matrices
for (unsigned i = 0; i < ls1; i++)
{   
    ioService.post(boost::bind(&thread_func,i, rs1, rs2, ls2, arr, left_arr, &result));
}   

ioService.stop();
threadpool.join_all();

任何帮助将不胜感激，谢谢！

【问题讨论】：

标签： c++ boost boost-asio threadpool boost-thread

【解决方案1】：

io_service::stop() 使run() 或run_one() 的所有调用尽快返回。它不会删除任何已在io_service 中排队的未完成处理程序。当io_service::stop()被调用时，threadpool中的线程会尽快返回，导致每个线程执行完毕。

由于io_service::post() 将在请求io_service 调用处理程序后立即返回，因此在io_service 停止之前threadpool 中的线程将调用多少个已发布处理程序是不确定的。

如果您希望 thread_func 被调用 ls1 次，那么一种简单的替代方法是重新组织代码，以便在 threadpool 服务之前将工作添加到 io_service，然后应用程序让io_service 运行完成。

boost::asio::io_service ioService;

// Add work to ioService.
for (unsigned i = 0; i < ls1; i++)
{   
  ioService.post(boost::bind(
      &thread_func,i, rs1, rs2, ls2, arr, left_arr, &result));
}   

// Now that the ioService has work, use a pool of threads to service it.
boost::thread_group threadpool;    
for (unsigned t = 0; t < num_threads; t++)
{   
  threadpool.create_thread(boost::bind(
      &boost::asio::io_service::run, &ioService));
}   

// Once all work has been completed (thread_func invoked ls1 times), the
// threads in the threadpool will be completed and can be joined.
threadpool.join_all();

【讨论】：

run 是否总是必须在 post() 之后调用？

【解决方案2】：

如果您希望 thread_func 被调用 ls1 次，那么您应该等到它实际被调用 ls1 次后再停止 io_service。正如所写，stop() 可能会在任何线程有机会被调度之前被调用。

有很多方法可以等待这种情况。例如，您可以使用条件变量：

#include <boost/asio.hpp>
#include <boost/thread.hpp>
unsigned num_threads = 10, ls1=11;
int result = 0;
boost::mutex m;
boost::condition_variable cv;
void thread_func(unsigned , int* result) {
    /* do stuff */
    {
        boost::lock_guard<boost::mutex> lk(m);
        ++*result;
    }
    cv.notify_one();
}
int main()
{
    boost::asio::io_service ioService;
    boost::asio::io_service::work work(ioService);
    boost::thread_group threadpool;
    for (unsigned t = 0; t < num_threads; t++)
        threadpool.create_thread(boost::bind(&boost::asio::io_service::run,
                                             &ioService));
    for (unsigned i = 0; i < ls1; i++)
        ioService.post(boost::bind(&thread_func,i,&result));

    {
        boost::unique_lock<boost::mutex> lk(m);
        cv.wait(lk, []{return result == ls1; });
    }
    ioService.stop();
    threadpool.join_all();
    std::cout << "result = " << result << '\n';
}

【讨论】：