std::condition_variable - 等待多个线程通知观察者

Question

我的问题是这样的：

我有一个观察者，它持有一个 std::condition_variable 和一个 std::mutex，我的工作线程对象有一个指向观察者的指针。每次工作线程完成其工作时，它都会调用 m_Observer->NotifyOne()，然后调用 condition_variable 的 notify_one() 函数。现在我想要做的是，启动一堆工作线程，每个线程都有不同的工作和不同的（独立）数据，并等待所有线程向观察者发出信号（使用 m_Observer->NotifyOne()），以便我能够根据所有工作线程的结果继续工作。

我的观察者看起来像这样：

class IAsyncObserver
{
private:
    std::condition_variable m_ObserverCV;
    bool m_bNotified;
    std::mutex m_Mutex;

public:
    IAsyncObserver()
    {
        m_bNotified = false;
    }

    ~IAsyncObserver()
    {
        /*m_bNotified = true;
        m_ObserverCV.notify_all();*/
    }

    void NotifyOne()
    {
        std::unique_lock<std::mutex> Lock(m_Mutex);
        m_bNotified = true;
        m_ObserverCV.notify_one();
    }

    void NotifyAll()
    {
        std::unique_lock<std::mutex> Lock(m_Mutex);
        m_bNotified = true;
        m_ObserverCV.notify_all();
    }

    void WaitForNotifications(uint32_t _uNumOfNotifications = 1)
    {
        uint32_t uNotifyCount = 0;
        while (uNotifyCount < _uNumOfNotifications)
        {
            std::unique_lock<std::mutex> Lock(m_Mutex);
            m_bNotified = false;
            m_ObserverCV.wait(Lock);

            if (m_bNotified)
                ++uNotifyCount;
        }
    }

}; // IAsyncObserver

其中 _uNumOfNotifications 是我要等待的工作线程数。

现在每个工作线程都应该运行一个模拟函数，该函数为一个时间步长/数据垃圾执行实际工作，然后暂停/等待，直到观察者通知工作线程继续。

worker 的线程函数可能如下所示：

do{
    //suspend simulation
    while (m_PauseSimulation.load())
    {
        std::unique_lock<std::mutex> wait(m_WaitMutex);
        m_CV.wait(wait);
        if (m_RunSimulation.load() == false)
        {
            SignalObserver();
            return;
        }
    }

    //lock the data while simulating
    {
        std::lock_guard<std::mutex> lock(m_LockMutex);

        //update simulation 
        Simulate(static_cast<float>(m_fDeltaTime));

        m_PauseSimulation.store(true);
    }

    //notify the physics manager that work is done here
    SignalObserver();       

} while (m_RunSimulation.load());

SignalObserver() 只是调用 m_Observer->NotifyOne()。

现在的问题是，一段时间后，线程在某处陷入死锁，并且观察者没有通知它们继续下一个时间步。问题可能出在 WaitForNotifications() 函数中，但我不确定。 Atm 我只有一个工作线程，所以 uNumOfNotifications = 1，但它仍然遇到在 m_ObserverCV.wait(Lock) 和 m_CV.wait(wait) 等待的问题，我什至不确定它是否真的是死锁或什么使用 condition_variable，因为我尝试从多个线程访问它。

在这一点上，我想引用 Ned Flanders 父亲的话：“我们什么都没尝试，而且都没有想法！”

感谢您的帮助。任何小费表示赞赏。

法比安

编辑：

感谢所有有用的信息和建议。我最终实现了迈克尔的第二个想法，因为我没有找到任何关于 std::barriers 的信息。所以这就是我所做的：

class IAsyncObserver
{
private:
    std::condition_variable m_ObserverCV;
    bool m_bNotified;
    std::mutex m_Mutex;

    uint32_t m_uNumOfNotifications;
    uint32_t m_uNotificationCount;

public:
    IAsyncObserver()
    {
        m_bNotified = false;
        m_uNumOfNotifications = 0;
        m_uNotificationCount = 0;
    }

    ~IAsyncObserver()
    {
        /*m_bNotified = true;
        m_ObserverCV.notify_all();*/
    }

    void SetBarrier(uint32_t _uNumOfNotifications = 1)
    {
        m_uNumOfNotifications = _uNumOfNotifications;
    }

    void NotifyBarrier()
    {
        std::unique_lock<std::mutex> Lock(m_Mutex);
        if (++m_uNotificationCount >= m_uNumOfNotifications)
        {
            m_bNotified = true;
            m_ObserverCV.notify_one();
        }
    }

    void WaitForNotifications()
    {
        std::unique_lock<std::mutex> Lock(m_Mutex);
        while (m_bNotified == false)
        {
            m_ObserverCV.wait(Lock);
        }
        m_uNotificationCount = 0;
    }

    void NotifyOne()
    {
        std::unique_lock<std::mutex> Lock(m_Mutex);
        m_bNotified = true;
        m_ObserverCV.notify_one();
    }

    void NotifyAll()
    {
        std::unique_lock<std::mutex> Lock(m_Mutex);
        m_bNotified = true;
        m_ObserverCV.notify_all();
    }

}; // IAsyncObserver

在我的“主要”功能中：MassSpringSystem 和 RigidBodySystem 是我的工人 atm

    //update systems here:
    {
        SetBarrier(m_uTotalNotifyCount);

        {   //start MassSpringSystems
            std::lock_guard<std::mutex> lock(m_LockMutex);
            for (std::shared_ptr<MassSpringSystem> MSS : m_MassSpringSystems)
            {
                MSS->SetDeltaTime(fDeltaTime);
                MSS->Continue();
            }
        }

        //ATTENTION this system works directly on the m_OctreeEntities!
        {   //start RigidBodySystems
            m_RigidBodySystem.SetDeltaTime(fDeltaTime);
            m_RigidBodySystem.AddData(m_RigidBodies);
            m_RigidBodySystem.Continue();
        }

        //wait for all systems to finish -> till they call SignalObserver
        WaitForNotifications();
    }

和上面一样在worker的线程函数中，但是这次SignalObserver调用了NotifyBarrier()

现在一切正常。一个简单但功能强大的解决方案，谢谢！

Answer 1

您尝试以不应该使用的方式使用条件变量 - 在这种情况下，您假设您可以计算通知。你不能。您可能会因此丢失通知，并且您正在计算标准允许的虚假唤醒。

相反，您应该使用在互斥体下递增的计数器，并仅在计数器达到工作人员数量时才向条件变量发出信号。 （最后在每个工人中执行此操作）。主线程一直在条件变量上休眠，直到计数器达到预期值。 （当然，计数器的验证也必须通过持有用于递增的互斥锁来完成）。据我所见，用原子替换互斥计数器（不互斥）似乎是不可能的，因为你不能原子地检查计数器并在 condvar 上休眠，所以你会在不互斥计数器的情况下获得竞争条件。

另一个从 boost 线程已知的同步原语是屏障，它没有进入 C++11。您构造一个屏障，并将工作线程的数量加一作为构造函数参数传递给它。所有工作线程都应在其末尾等待条件变量，主线程应在构建工作人员后等待。所有线程都将阻塞在该屏障上，直到所有工作线程和主线程都阻塞，并且在那一刻被释放。所以如果主线程被释放，你就知道所有的worker都完成了。但这有一个问题：在所有工作线程完成之前，没有工作线程完成（并释放相关的管理资源），这对您来说可能是也可能不是问题。 This question presents an implementation of boost::barrier using C++11 threading facilities.