将 producer/consumer 与屏障同步

Question

我有一个生产者线程，它为三个消费者线程生产工作。产生工作后，生产者线程会等待，直到消费者线程完成对工作的处理。然后生产者线程继续处理结果。

#include <condition_variable>
#include <mutex>
#include <boost/thread/barrier.hpp>
#include <vector>
#include <queue>


std::condition_variable cond;
std::mutex mutex;
boost::barrier barrier(4);

std::vector<std::thread> workers;
std::queue<unsigned int> work;
std::queue<unsigned int> results;

void worker();

int main()
{
    // 1 producer and 3 consumers
    for(unsigned int i = 0; i < 3; i++)
        workers.push_back(std::thread(worker));

    // Wait here so the three workers can get to cond.wait();
    barrier.wait();

    std::unique_lock<std::mutex> lock(mutex);
    while(true)
    {
        // Generate work
        std::cout << "gen" << std::endl;
        for(unsigned int i = 0; i < 10; i++)
            work.push(i);

        cond.notify_all();

        lock.unlock();
        barrier.wait();

        // Handle the results
        while(results.size() > 0)
            results.pop();

        lock.lock();
    }

    return 0;
}

void worker()
{
    while(true)
    {
        std::unique_lock<std::mutex> lock(mutex);
        while(results.size() == 0)
        {
            lock.unlock();
            barrier.wait();
            lock.lock();
            cond.wait(lock);
        }

        // Get work
        unsigned int next = work.front();
        work.pop();

        // Store the result
        results.push(next);

        lock.unlock();


    }
}

问题是我需要确保在生产者线程开始其下一次迭代之前所有消费者线程都已进入cond.wait(lock)：

所有 4 个线程都已到达屏障。屏障被释放，线程可以继续。
生产者线程在所有消费者线程到达 cond.wait(lock) 之前锁定互斥量。因此至少有一个消费者线程被 lock.lock().
生产者线程开始下一次迭代，创建工作并通知消费者。由于至少一个消费者线程尚未达到 cond.wait(lock)，因此至少一个消费者线程将错过 notify_all()。这些线程现在等待下一个 notify_all() - 它永远不会到达。
下一次到达屏障时，至少有一个消费者线程仍在等待下一个notify_all()。这样barrier就不会解锁，就出现了死锁。

我该如何解决这种情况？

Answer 1

A condition_variable 应与标志一起使用，以帮助防止虚假唤醒。这个相同的标志也可以用来检查线程是否应该等待或者直接去工作。

添加一个 bool go_to_work=false;，然后我们只需将其添加为对 wait 的调用中的谓词，并确保我们从主线程 set/unset 它。

在调用 notify_all 之前的主线程中，我们设置了 bool

go_to_work=true;
cond.notify_all();

在我们的工作线程中，我们将谓词添加到我们的 wait 调用中

cond.wait(lock, [](){ return go_to_work; });

最后，在我们的主线程中，我们希望在完成所有工作后将标志设置回 false。

barrier.wait();
lock.lock();  // We need to lock the mutex before modifying the bool
go_to_work=false;
lock.unlock();

//Handle result...

现在，如果线程在主线程设置 go_to_work=true 后到达 wait 调用，它根本不会等待，直接继续执行工作。作为奖励，这还可以防止虚假唤醒。

将 producer/consumer 与屏障同步

Synchronizing producer/consumer with a barrier

c++

multithreading

mutex

producer-consumer

barrier