c++ - C++ 队列中的线程池

Question

我一直在尝试同时解决一个问题，这非常适合线程池模式。在这里我将尝试提供一个最小的代表性示例:

假设我们有一个像这样的伪程序:

Q : collection<int>
while (!Q.empty()) {
    for each q in Q {
        // perform some computation
    }
    // assign a new value to Q
    Q = something_completely_new();
}

我正在尝试以并行方式实现这一点，使用 n-1 个工作线程和一个主线程。工作人员将通过从 Q 中获取元素来在内循环中执行计算。

我尝试使用两个条件变量来解决这个问题:work，主线程在其中通知工作线程 Q 已分配给它；另一个 work_done，其中工作人员通知主人整个计算可能已完成。

这是我的 C++ 代码:

#include <iostream>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <thread>

using namespace std;

std::queue<int> Q;
std::mutex mut;
std::condition_variable work;
std::condition_variable work_done;

void run_thread() {
    for (;;) {
        std::unique_lock<std::mutex> lock(mut);
        work.wait(lock, [&] { return Q.size() > 0; });

        // there is work to be done - pretend we're working on something
        int x = Q.front(); Q.pop();
        std::cout << "Working on " << x << std::endl;

        work_done.notify_one();
    }
}

int main() {
    // your code goes here
    std::vector<std::thread *> workers(3);

    for (size_t i = 0; i < 3; i++) {
        workers[i] = new std::thread{
            [&] { run_thread(); }
        };
    }

    for (int i = 4; i > 0; --i) {
        std::unique_lock<std::mutex> lock(mut);
        Q = std::queue<int>();
        for (int k = 0; k < i; k++) {
            Q.push(k);
        }
        work.notify_all();
        work_done.wait(lock, [&] { return Q.size() == 0; });
    }

    for (size_t i = 0; i < 3; i++) {
        delete workers[i];
    }

    return 0;
}

不幸的是，在 OS X 上使用 g++ -std=c++11 -Wall -o main main.cpp 编译后，我得到以下输出:

Working on 0
Working on 1
Working on 2
Working on 3
Working on 0
Working on 1
Working on 2
Working on 0
Working on 1
Working on 0
libc++abi.dylib: terminating
Abort trap: 6

经过一段时间的谷歌搜索后，它看起来像是一个段错误。这可能与我滥用条件变量有关。我希望能得到一些见解，包括架构上的(关于如何解决此类问题的)和具体的见解，就像我在这里做错了什么一样。

感谢您的帮助

Answer 1

您的应用程序已被 std::terminate 终止。

线程函数体是无限循环的，因此当执行这些行时

for (size_t i = 0; i < 3; i++) {
    delete workers[i];
}

您想要删除仍在运行的线程(每个线程都处于可连接状态)。当您调用处于可连接状态的线程的析构函数时，会发生以下情况(来自 http://www.cplusplus.com/reference/thread/thread/~thread/ )

If the thread is joinable when destroyed, terminate() is called.

因此，如果您希望不调用terminate，则应在创建线程后调用detach()方法。

  for (size_t i = 0; i < 3; i++) {
    workers[i] = new std::thread{
        [&] { run_thread(); }
    };
    workers[i]->detach();  // <---
}