如何使用 boost::latch？

Question

我试图在我的程序中使用 boost::latch 来阻止等待，直到所有线程完成或超时。我的代码如下。 ctpl是从https://github.com/vit-vit/CTPL采用的线程池库。

#include <boost/thread/latch.hpp>
#include <CTPL/ctpl.h>
#include <mutex>
#include <iostream>

using namespace std;

int main(int argc, char **argv) {

    ctpl::thread_pool outer_tp(100);
    ctpl::thread_pool inner_tp(5, 5000);

    auto out_func = [&inner_tp](int outer_id, int outer_invoke_idx) {
        int num_batch = 20;
        boost::latch latch_(num_batch);

        auto func = [&latch_, &outer_invoke_idx](int inner_id, int inner_invoke_idx) {
            try {
                std::cout << "outer: " << outer_invoke_idx << ", inner: " << inner_invoke_idx << endl;
            } catch (exception &ex) { cout << "error: " << ex.what() << endl; }
            latch_.count_down();
        };

        for (int i = 0; i < num_batch; ++i) {
            inner_tp.push(func, i);
        }

        latch_.wait_for(boost::chrono::milliseconds(1));
    };

    for (int i = 0; i < 5000; ++i) outer_tp.push(out_func, i);
    outer_tp.stop(true);

    return 0;
}

g++ -std=c++11 test.cpp -lboost_system -lpthread -lboost_chrono -lboost_thread

但是我收到以下错误消息。

bool boost::latch::count_down(boost::unique_lock&): Assertion `count_ > 0' failed.

如果我使用latch_.wait() 而不是latch_.wait_for() 或设置很长的等待时间，则代码可以正常工作。因此我猜“超时”会导致这个错误问题。有谁知道如何解决这个错误。

Answer 1

您的代码似乎没有什么问题。我认为在内部线程中通过引用引用latch_ 就是其中之一。用shared_ptr 替换为boost::latch 可以解决这个问题。另一个问题与outer_invoke_idx 类似。修复这些并等待inner_tp 完成似乎可以让您的测试正常工作。

这是对我有用的修改后的测试用例：

#include <boost/thread/latch.hpp>
#include <memory>
#include <CTPL/ctpl.h>
#include <mutex>
#include <iostream>

using namespace std;

int main(int argc, char **argv) {

    ctpl::thread_pool outer_tp(100);
    ctpl::thread_pool inner_tp(5, 5000);

    auto out_func = [&inner_tp](int outer_id, int outer_invoke_idx) {
        int num_batch = 20;
        auto latch_ = std::make_shared<boost::latch>(num_batch);

        auto func = [latch_, outer_invoke_idx](int inner_id, int inner_invoke_idx) {
            try {
                std::cout << "outer: " << outer_invoke_idx << ", inner: " << inner_invoke_idx << endl;
            } catch (exception &ex) { cout << "error: " << ex.what() << endl; }
            latch_->count_down();
        };

        for (int i = 0; i < num_batch; ++i) {
            inner_tp.push(func, i);
        }

        latch_->wait_for(boost::chrono::milliseconds(1));
    };

    for (int i = 0; i < 5000; ++i) outer_tp.push(out_func, i);

    outer_tp.stop(true);
    inner_tp.stop(true);
    std::cout << "EXITING!!!" << std::endl;
    return 0;
}