C ++多个消费者线程卡在条件变量上

Question

我正在用 C++ 编写一个生产者、多个消费者的程序。我首先调用消费者线程，然后将元素添加到数组中。 一切正常，但最终消费者线程没有加入，因为它们被卡在等待条件变量并且程序冻结。

我认为问题在于线程在循环中不断被调用，因为 currentSize 不受保护并且它们无法退出条件变量，但我不知道如何解决它。

struct Item {
public:
    string name;
    int time;
    double height;
};

struct Monitor {
private:
    Item items[12];
    int currentSize;
    bool finished;
    mutex lock;
    condition_variable cv;
public:
    Monitor() {
        finished = false;
        currentSize = 0;
    }
    void put(Item item) {
        unique_lock<mutex> guard(lock);
        cv.wait(guard, [&] { return (currentSize < 12); });
        items[currentSize] = item;
        currentSize++;
        cv.notify_all();
    }

    Item get() {
        unique_lock<mutex> guard(lock);
        cv.wait(guard, [&] { return (currentSize > 0); });
        Item item = items[currentSize - 1];
        currentSize--;
        return item;
    }
    bool get_finished() {
        return finished;
    }
    void set_finished() {
        finished = true;
    }
    int get_size() {
        return currentSize;
    }
};

int main() {
    vector<Item> items = read_file(file);

    Monitor monitor;
    vector<thread> threads;
    vector<Item> results;

    for (int i = 0; i < 4; i++) {
        threads.emplace_back([&] {
            while (!monitor.get_finished()) {
                if (monitor.get_size() > 0) {
                    Item item = monitor.get();
                    results.push_back(item);
                }
            }
        });
    }

    for (int i = 0; i < items.size(); i++) {
        monitor.put(items[i]);
    }
    monitor.set_finished();

    for_each(threads.begin(), threads.end(), mem_fn(&thread::join));

    return 0;
}

Answer 1

为什么消费者线程会阻塞？

我已经测试了你的代码，结果证明是生产者线程阻塞了put() 方法。为什么？

想象以下场景：向量items中有13个项目。

主线程（生产者）愉快地加载前 12 个项目，并在 cv 上等待 currentSize 低于 12。

通知消费者线程，愉快地消费前12个项目，然后在cv上等待currentSize变得大于0。

但是等等！现在每个人都在等着什么，没有人通知。因此，所有线程都会阻塞。当currentSize 低于 12 时，需要通知生产者。

Answer 2

我注意到了一些问题。使成员变量原子化，notify_all 在 get api 中。但是也存在逻辑错误。想象一下，您当前有 4 个线程正在运行，并且有 5 个项目在队列中。此时假设每个线程都能够从队列中取出一个，现在有 4 个线程，队列中只有一个项目。其中一个线程取出最后一个，现在那里有 0 个项目，但其他三个线程仍在等待条件变量。因此，一个解决方案是，如果最后一项出现，应该通知每个线程，如果没有其他 elemnet 从 API 返回。

#include <iostream>
#include <vector>
#include <condition_variable>
#include <thread>
#include <algorithm>
#include <atomic>

using namespace std;

using Item = int;

struct Monitor {
private:
    Item items[12];
    std::atomic<int> currentSize;
    std::atomic<bool> finished;
    mutex lock;
    condition_variable cv;
public:
    Monitor() {
        finished = false;
        currentSize = 0;
    }
    void put(Item item) {
        unique_lock<mutex> guard(lock);
        cv.wait(guard, [&] { return (currentSize < 12); });
        items[currentSize] = item;
        currentSize++;
        cv.notify_all();

        std::cerr << "+ " << currentSize << std::endl ;
    }

    Item get() {
        unique_lock<mutex> guard(lock);
        cv.wait(guard, [&] { return (currentSize >= 0 ); });

        Item item;
        if (currentSize > 0 ){      
            currentSize--;
            item = items[currentSize];
            cv.notify_all();
            std::cerr << "- " << currentSize << std::endl ;
        }
        return item;
    }
    bool get_finished() {
        return finished;
    }
    void set_finished() {
        finished = true;
    }
    int get_size() {
        return currentSize;
    }
};

int main() {
    vector<Item> items(200);
    std::fill ( items.begin() , items.end(), 100);

    Monitor monitor;
    vector<thread> threads;
    vector<Item> results;

    for (int i = 0; i < 10; i++) {
        threads.emplace_back([&] {
            while ( !monitor.get_finished() ) {
                if (monitor.get_size() > 0) {
                    Item item = monitor.get();
                    results.push_back(item);
                }
            }
        });
    }

    for (int i = 0; i < items.size(); i++) {
        monitor.put(items[i]);
    }
    monitor.set_finished();

    for_each(threads.begin(), threads.end(), mem_fn(&thread::join));

    return 0;
}