C++ 线程与 boost asio

Question

我有一个使用 boosts 库的线程池，并且我在下面的示例中设置了运行两个可以重新运行 4 次的线程。在继续编写代码之前，我可以检查is_service 以查看所有子线程执行是否已完成的最佳方法是什么？其余代码取决于所有子线程在程序继续之前完成。如果我调用Sleep(1000)，我可以获得所需的行为，但这是不可取的，我查看了io_service_.stopped() 的检查，但总是返回0。任何想法将不胜感激。

#include <iostream>                   
#include <boost/asio/io_service.hpp>
#include <boost/bind.hpp>
#include <boost/thread/thread.hpp>
#include <vector>               
#include <string>

using namespace std;

class Model {
  public:
    // Constructor
    Model() {
        work_ctrl_ = new boost::asio::io_service::work(io_service_);

        for (int i = 0; i < 2; ++i) {
            threads_.create_thread(
                    boost::bind(&boost::asio::io_service::run, &io_service_));
        }
    }
    // Deconstructor
    ~Model() {
        delete work_ctrl_;
    }

    // Function I want to thread
    void manipulate_vector(unsigned start, unsigned last) {
        cout << "entering manipulate vector(), from thread " << boost::this_thread::get_id() << endl;
        for(unsigned k = start; k <= last; ++k)
            my_vector_[k] *= sqrt(32);
        cout << "exit manipulate vector()" << endl;
        Sleep(500); // Add a sleep to mimic a long algorithm being executed
    }

    void update() {
        // Do otherstuff that can't be threaded
        cout << "entering update" << endl;

        // run manipulate_vector() across multiple threads
        // - start thread
        // - execute function call.
        // - stop thread
        io_service_.post(boost::bind(manipulate_vector, this, 0, mid_point_));
        io_service_.post(boost::bind(manipulate_vector, this, mid_point_, my_vector_.size()));

        cout << io_service_.stopped() << endl;

        // keep doing otherstuff that can't be threaded
        cout << "hopefully the threads are finished and I can take that information and continue." << endl;
    }

    void run(void) {
        // call update 10 times
        for(unsigned i = 0; i < 4; ++i) {
            update();
            //Sleep(2000);
        }
    }

    void initialise() {
        // initialise vector
        for(unsigned j = 0; j < 100000000; ++j)
            my_vector_.push_back(j);
        mid_point_ = 49999999;
    }

  private:
    boost::asio::io_service io_service_;
    boost::thread_group threads_;
    boost::asio::io_service::work *work_ctrl_;
    unsigned n_threads_;
    vector<double>  my_vector_;
    unsigned mid_point_;
};

int main() {
    std::cout << "----------Enter Main----------" << std::endl;
    Model model;
    model.initialise();
    model.run();
    std::cout << "----------Exit Main----------" << std::endl;
    system("PAUSE");
}

Answer 1

大概您想知道您的工作何时完成，而不是线程何时停止执行（直到您调用io_service.stop() 或删除work-ctrl_，它们才会知道）。

标准解决方案是使用条件变量：

std::mutex mutex;
std::condition_variable condition;
int workCount;

void manipulate_vector(unsigned start, unsigned last) {
    cout << "entering manipulate vector(), from thread " << boost::this_thread::get_id() << endl;
    for(unsigned k = start; k <= last; ++k)
        my_vector_[k] *= sqrt(32);
    cout << "exit manipulate vector()" << endl;
    Sleep(500); // Add a sleep to mimic a long algorithm being executed
    std::unique_lock<std::mutex> lock(mutex);
    workCount--;
    condition.notify_one();
}

void update() {
    // Do otherstuff that can't be threaded
    cout << "entering update" << endl;

    // run manipulate_vector() across multiple threads
    // - start thread
    // - execute function call.
    // - stop thread
    workCount = 2;
    io_service_.post(boost::bind(manipulate_vector, this, 0, mid_point_));
    io_service_.post(boost::bind(manipulate_vector, this, mid_point_, my_vector_.size()));
    {
       std::unique_lock<std::mutex> lock(mutex);
       condition.wait(lock, [&]{ return workCount == 0; });
    }
}

Answer 2

如果您对当前线程中正在运行的某些工作感到满意，您可以在删除work_ctrl_ 后调用io_service_.run() 等待。当它返回时，没有剩余的工作，您可以安全地继续。您可以将线程组中的并发级别降低 1 以获得相同的整体并发。

一个简单的替代方法是在所有帖子都已排队并且您已删除 work_ctrl_ 之后调用 threads_.join_all()。