每个条件唤醒多个线程工作一次

Question

// g++ -Wall -o threadtest -pthread threadtest.cpp
#include <iostream>
#include <condition_variable>
#include <mutex>
#include <thread>
#include <chrono>

std::mutex condMutex;
std::condition_variable condVar;
bool dataReady = false;

void state_change_worker(int id)
{
    while (1)
    {
        {
            std::unique_lock<std::mutex> lck(condMutex);
            condVar.wait(lck, [] { return dataReady; });
            // Do work only once.
            std::cout << "thread " << id << " working\n";
        }
    }
}

int main()
{
    // Create some worker threads.
    std::thread threads[5];
    for (int i = 0; i < 5; ++i)
        threads[i] = std::thread(state_change_worker, i);

    while (1)
    {
        // Signal to the worker threads to work.
        {
            std::cout << "Notifying threads.\n";
            std::unique_lock<std::mutex> lck(condMutex);
            dataReady = true;
            condVar.notify_all();
        }
        // It would be really great if I could wait() on all of the 
        // worker threads being done with their work here, but it's 
        // not strictly necessary.
        std::cout << "Sleep for a bit.\n";
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    }
}

// g++ -Wall -o threadtest -pthread threadtest.cpp
#include <iostream>
#include <condition_variable>
#include <mutex>
#include <thread>
#include <chrono>

class SquadLock
{
public:
    void waitForLeader()
    {
        {
            // Increment count to show that we are waiting in queue.
            // Also, if we are the thread that reached the target, signal
            // to the leader that everything is ready.
            std::unique_lock<std::mutex> count_lock(count_mutex_);
            std::unique_lock<std::mutex> target_lock(target_mutex_);
            if (++count_ >= target_)
                count_cond_.notify_one();
        }
        // Wait for leader to signal done.
        std::unique_lock<std::mutex> lck(done_mutex_);
        done_cond_.wait(lck, [&] { return done_; });
        {
            // Decrement count to show that we are no longer waiting.
            // If we are the last thread set done to false.
            std::unique_lock<std::mutex> lck(count_mutex_);
            if (--count_ == 0)
            {
                done_ = false;
            }
        }
    }

    void waitForHerd()
    {
        std::unique_lock<std::mutex> lck(count_mutex_);
        count_cond_.wait(lck, [&] { return count_ >= target_; });
    }
    void leaderDone()
    {
        std::unique_lock<std::mutex> lck(done_mutex_);
        done_ = true;
        done_cond_.notify_all();
    }
    void incrementTarget()
    {
        std::unique_lock<std::mutex> lck(target_mutex_);
        ++target_;
    }
    void decrementTarget()
    {
        std::unique_lock<std::mutex> lck(target_mutex_);
        --target_;
    }
    void setTarget(int target)
    {
        std::unique_lock<std::mutex> lck(target_mutex_);
        target_ = target;
    }

private:
    // Condition variable to indicate that the leader is done.
    std::mutex done_mutex_;
    std::condition_variable done_cond_;
    bool done_ = false;

    // Count of currently waiting tasks.
    std::mutex count_mutex_;
    std::condition_variable count_cond_;
    int count_ = 0;

    // Target number of tasks ready for the leader.
    std::mutex target_mutex_;
    int target_ = 0;
};

SquadLock squad_lock;
std::mutex print_mutex;
void state_change_worker(int id)
{
    while (1)
    {
        // Wait for the leader to signal that we are ready to work.
        squad_lock.waitForLeader();
        {
            // Adding just a bit of sleep here makes it so that every thread wakes up, but that isn't the right way.
            // std::this_thread::sleep_for(std::chrono::milliseconds(100));
            std::unique_lock<std::mutex> lck(print_mutex);
            std::cout << "thread " << id << " working\n";
        }
    }
}

int main()
{

    // Create some worker threads and increment target for each one
    // since we want to wait until all threads are finished.
    std::thread threads[5];
    for (int i = 0; i < 5; ++i)
    {
        squad_lock.incrementTarget();
        threads[i] = std::thread(state_change_worker, i);
    }
    while (1)
    {
        // Signal to the worker threads to work.
        std::cout << "Starting threads.\n";
        squad_lock.leaderDone();
        // Wait for the worked threads to be done.
        squad_lock.waitForHerd();
        // Wait until next time, processing results.
        std::cout << "Tasks done, waiting for next time.\n";
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    }
}

Answer 1

下面是我创建的一个关于并发设计模式的博客的节选。模式是使用Ada语言表达的，但概念是可以翻译到C++的。

摘要

监视器

有几种创建和控制共享内存的理论方法。其中最灵活和最健壮的是监视器，如C.A.R.首先描述的。霍尔。监视器是一个具有三种不同操作的数据对象。

班锁

班组锁定允许特殊任务（班长）监视一群或一组工人任务的进度。当所有（或足够数量的）工作者任务都完成了他们工作的某个方面，并且领导者准备继续工作时，整个任务集就可以通过一个障碍，继续他们的下一个活动序列。其目的是允许任务异步执行，同时通过一组复杂的活动来协调它们的进度。

package Barriers is
   protected type Barrier(Trigger : Positive) is
      entry Wait_For_Leader; 
      entry Wait_For_Herd; 
      procedure Leader_Done; 
   private
      Done : Boolean := False;
   end Barrier;

   protected type Autobarrier(Trigger : Positive) is
      entry Wait_For_Leader; 
      entry Wait_For_Herd; 
   private
      Done : Boolean := False;
   end Autobarrier;
end Barriers;

这个包显示了两种班锁。屏障保护类型演示了一个基本的小队锁定。herd调用Wait_For_Leader，leader调用Wait_For_Herd，然后调用leader_done。Autobarrier演示了一个更简单的界面。herd调用Wait_For_Leader，leader调用wait_for_herd。在创建任何一种类型的屏障的实例时都使用触发器参数。它设置了在进行之前领导者必须等待的羊群任务的最小数量。

package body Barriers is
   protected body Barrier is
      entry Wait_For_Herd when Wait_For_Leader'Count >= Trigger is
      begin
         null;
      end Wait_For_Herd;

      entry Wait_For_Leader when Done is
      begin
         if Wait_For_Leader'Count = 0 then
            Done := False;
         end if;
      end Wait_For_Leader;

      procedure Leader_Done is
      begin
         Done := True;
      end Leader_Done;
   end Barrier;

   protected body Autobarrier is
      entry Wait_For_Herd when Wait_For_Leader'Count >= Trigger is
      begin
         Done := True;
      end Wait_For_Herd;

      entry Wait_For_Leader when Done is
      begin
         if Wait_For_Leader'Count = 0 then
            Done := False;
         end if;
      end Wait_For_Leader;
   end Autobarrier;
end Barriers;