在 C++11 中实现 boost::barrier

Posted 2023-03-17

【中文标题】在 C++11 中实现 boost::barrier

【英文标题】：Implementing boost::barrier in C++11

【发布时间】：2014-08-19 08:57:06

【问题描述】：

我一直在尝试让项目摆脱所有 boost 引用并切换到纯 C++11。

在某一时刻，线程工作者被创建，等待屏障发出“go”命令，完成工作（通过 N 个线程传播）并在所有线程完成时同步。基本思路是主循环给出go命令（boost::barrier.wait()），用同样的函数等待结果。

我在另一个项目中实现了一个基于 Boost 版本的定制屏障，并且一切正常。实现如下：

Barrier.h：

class Barrier 
public:
    Barrier(unsigned int n);
    void Wait(void);
private:
    std::mutex counterMutex;
    std::mutex waitMutex;

    unsigned int expectedN;
    unsigned int currentN;
;

Barrier.cpp

Barrier::Barrier(unsigned int n) 
    expectedN = n;
    currentN = expectedN;


void Barrier::Wait(void) 
    counterMutex.lock();

    // If we're the first thread, we want an extra lock at our disposal

    if (currentN == expectedN) 
        waitMutex.lock();
    

    // Decrease thread counter

    --currentN;

    if (currentN == 0) 
        currentN = expectedN;
        waitMutex.unlock();

        currentN = expectedN;
        counterMutex.unlock();
     else 
        counterMutex.unlock();

        waitMutex.lock();
        waitMutex.unlock();
    

此代码已在 ios 和 android 的 NDK 上使用，没有任何问题，但在 Visual Studio 2013 项目上尝试时，似乎只有锁定互斥锁的线程才能解锁它（断言：解锁无主互斥锁）。

是否有任何可用于 C++11 的非旋转（阻塞，例如这个）屏障版本？我只能找到使用忙等待的障碍，这是我想阻止的（除非真的没有理由这样做）。

【问题讨论】：

您是否使用waitMutex 作为信号量？

【参考方案1】：

class Barrier 
public:
    explicit Barrier(std::size_t iCount) : 
      mThreshold(iCount), 
      mCount(iCount), 
      mGeneration(0) 
    

    void Wait() 
        std::unique_lock<std::mutex> lLockmMutex;
        auto lGen = mGeneration;
        if (!--mCount) 
            mGeneration++;
            mCount = mThreshold;
            mCond.notify_all();
         else 
            mCond.wait(lLock, [this, lGen]  return lGen != mGeneration; );
        
    

private:
    std::mutex mMutex;
    std::condition_variable mCond;
    std::size_t mThreshold;
    std::size_t mCount;
    std::size_t mGeneration;
;

【讨论】：

我喜欢它可以持续几代人的事实。

这个实现是可重用的，并且不会被虚假唤醒。

我相信有一个错字，应该是 lLock(mMutex) 而不是 lLockmMutex。

@jamshid 不，这不是错字，请查看list initialization上的示例

如果你好奇为什么std::unique_lock 而不是std::lock_guard，那是因为它是可重新锁定的。我们需要在等待条件变量时释放锁。 See this SO answer

【参考方案2】：

使用std::condition_variable 而不是std::mutex 来阻塞所有线程，直到最后一个线程到达屏障。

class Barrier

private:
    std::mutex _mutex;
    std::condition_variable _cv;
    std::size_t _count;
public:
    explicit Barrier(std::size_t count) : _count(count)  
    void Wait()
    
        std::unique_lock<std::mutex> lock(_mutex);
        if (--_count == 0) 
            _cv.notify_all();
         else 
            _cv.wait(lock, [this]  return _count == 0; );
        
    
;

【讨论】：

请注意，此实现仅适用于一次性使用屏障，但可以避免虚假唤醒。

假设我有三个线程需要在屏障上等待，我将屏障初始化为屏障屏障（2）；此代码似乎不起作用。

【参考方案3】：

这是我上面接受的答案的版本，带有重复使用的自动重置行为；这是通过交替向上和向下计数来实现的。

    /**
    * @brief Represents a CPU thread barrier
    * @note The barrier automatically resets after all threads are synced
    */
    class Barrier
    
    private:
        std::mutex m_mutex;
        std::condition_variable m_cv;

        size_t m_count;
        const size_t m_initial;

        enum State : unsigned char 
            Up, Down
        ;
        State m_state;

    public:
        explicit Barrier(std::size_t count) : m_count count , m_initial count , m_state State::Down   

        /// Blocks until all N threads reach here
        void Sync()
        
            std::unique_lock<std::mutex> lock m_mutex ;

            if (m_state == State::Down)
            
                // Counting down the number of syncing threads
                if (--m_count == 0) 
                    m_state = State::Up;
                    m_cv.notify_all();
                
                else 
                    m_cv.wait(lock, [this]  return m_state == State::Up; );
                
            

            else // (m_state == State::Up)
            
                // Counting back up for Auto reset
                if (++m_count == m_initial) 
                    m_state = State::Down;
                    m_cv.notify_all();
                
                else 
                    m_cv.wait(lock, [this]  return m_state == State::Down; );
                
            
        
    ;