c++ - 如何查看io_service中的任务是否完成？

Question

我有一个关于 boost::io_service 的问题。

我有一组可以同时运行的任务。运行所有这些之后，我需要同时运行另一组任务。但是，必须在开始运行第二组之前完成第一组。这意味着我需要确保提交到 io_service 的所有作业都已完成，然后才能开始安排到第二组。

我可以通过保留某种计数器并添加一个繁忙的循环来实现它，但它看起来效率不高。所以，我想检查一下是否有人有更好的主意。以下是我用来进行实验的虚拟代码。

提前致谢!

 #include <cstdio>
 #include <iostream>
 #include <unistd.h>
 #include <boost/asio/io_service.hpp>
 #include <boost/bind.hpp>
 #include <boost/thread/thread.hpp>

 const size_t numTasks = 100000;

 void print_counter(const size_t id)
 {
   if (id + 1 == numTasks) {
     printf("sleeping for %ld\n", id);
     sleep(15);
   }
   printf("%ld\n", id);
 }

 int main(int argc, char** argv)
 {
   using namespace std;
   using namespace boost;

   asio::io_service io_service;
   asio::io_service::work work(io_service);

   const size_t numWorker = 4;
   boost::thread_group workers;
   for(size_t i = 0; i < numWorker; ++i) {
     workers.create_thread(boost::bind(&asio::io_service::run, &io_service));
   }

   for(size_t i = 0; i < numTasks; ++i) {
     io_service.post(boost::bind(print_counter, i));
   }

   // TODO: wait until all the tasks are done above


   for(size_t i = 0; i < numTasks; ++i) {
     io_service.post(boost::bind(print_counter, i));
   }

   // TODO: wait until all the tasks are done above

   // ...

   // Finally stop the service
   io_service.stop();
   workers.join_all();
   return 0;
 }

Answer 1

您的主要问题是您的所有任务集都由 io_service 的同一个实例处理。函数 io_service::run 返回没有要处理的任务的地方。 io_service::work 的析构函数通知 io_service 对象 run 可以返回队列中没有待执行任务的地方。您可以发布第一组的所有任务，然后销毁工作并等待 io_service::run 返回，然后再次创建 work 对象，发布下一组的任务并删除工作， 等等。要做到这一点，只需编写如下所示的辅助类:

class TasksWaiter 
{
public:
    TasksWaiter(int numOfThreads) 
    {
        work = std::make_unique<boost::asio::io_service::work>(io_service);
        for(size_t i = 0; i < numOfThreads; ++i) {
            workers.create_thread(boost::bind(&boost::asio::io_service::run, &io_service));
        }
    }

    ~TasksWaiter() {
        work.reset();
        workers.join_all();
    }

    template<class F>
    void post(F f) {
        io_service.post(f);
    }

    boost::thread_group workers;
    boost::asio::io_service io_service;
    std::unique_ptr<boost::asio::io_service::work> work;
};

int main()
{
    {
        TasksWaiter w1{4};
        for (int i = 0; i < numTasks; ++i)
            w1.post(boost::bind(print_counter,i));
        // work in w1 is destroyed, then io_service::run ends 
        // when there are no tasks to be performed
    }
    printf("wait here");
    {
        TasksWaiter w1{4};
        for (int i = 0; i < numTasks; ++i)
            w1.post(boost::bind(print_counter,i));
    }
}

几点说明:

1. 在构造函数中创建线程池

2. 在析构函数中工作被删除，因此 io_service::run 仅在没有待处理任务时返回

3. 析构函数的功能可以包装到成员函数中 - 例如wait，那么您就不必使用 {} 范围来等待您的任务。