Синхронизация производителя / потребителя с барьером

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) до того, как поток производителя начнет следующую итерацию:

1. Все 4 темы достигли барьера. Барьер освобождается и нити могут продолжаться.
2. Поток производителя блокирует мьютекс до того, как все потоки потребителя достигнут cond.wait(lock). Таким образом, по крайней мере один потребительский поток заблокирован lock.lock().
3. Поток производителя начинает следующую итерацию, создает работу и уведомляет потребителей. Поскольку по крайней мере один потребительский поток еще не достиг cond.wait(lock), notify_all() будет пропущен по крайней мере одним потребительским потоком. Эти потоки теперь ждут следующего notify_all() - который никогда не прибудет.
4. В следующий раз, когда будет достигнут барьер, по крайней мере один потребительский поток все еще ждет следующего notify_all(). Таким образом, барьер не будет разблокирован и возникнет тупик.

Как мне разрешить эту ситуацию?

Answer 1

Условие_ переменная должно использоваться вместе с флагом, чтобы помочь предотвратить ложные пробуждения. Этот же флаг можно также использовать для проверки того, должен ли поток вообще ждать или просто приступить к работе.

Добавьте bool go_to_work=false;, затем мы просто добавим его в качестве предиката при вызове к wait и убедимся, что мы установили / удалили его из основного потока.

В главном потоке перед вызовом 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...

Теперь, если поток достигает вызова wait после того, как основной поток установил go_to_work=true, он вообще не будет ждать и просто продолжит работу. В качестве бонуса это также защищает от ложных пробуждений.