阻塞队列实现：竞争条件在哪里？答案

【问题标题】：Blocking Queue implementation: Where's the race condition?阻塞队列实现：竞争条件在哪里？
【发布时间】：2023-03-23 13:05:01
【问题描述】：

又是我和我的 BlockingQueue...我根据this article 和this question 重写了它。它发送一些项目，然后因访问冲突而崩溃。代码如下：

template <typename T>
bool DRA::CommonCpp::CTBlockingQueue<T>::Push( T pNewValue ){
    volatile long oldSize;
    ::InterlockedExchange( &oldSize, m_Size );
    CTNode* pNewNode = new CTNode();
    pNewNode->m_pValue = pNewValue;
    {//RAII block
        CGuard g( m_TailCriticalSection );
        m_pTailNode->m_pNext = pNewNode;
        m_pTailNode = pNewNode;
        ::InterlockedIncrement( &m_Size );
    }
    if( oldSize == 0 )
        m_eAtLeastOneElement.set();
    return true;
}

template <typename T>
bool DRA::CommonCpp::CTBlockingQueue<T>::Pop( T& pValue ){
    CTNode* pCurrentNode;
    {//RAII block
        CGuard g( m_HeadCriticalSection );
        pCurrentNode = m_pHeadNode;
        CTNode* pNewHeadNode = m_pHeadNode->m_pNext;
        if( pNewHeadNode == NULL ){
            CEvent* pSignaledEvent;
            CEvent::waitForPair( m_eAtLeastOneElement, m_eFinished, pSignaledEvent );
            if( pSignaledEvent == &m_eFinished )
                return false;
            pNewHeadNode = m_pHeadNode->m_pNext;
        }
        pValue = pNewHeadNode->m_pValue;
        m_pHeadNode = pNewHeadNode;
        ::InterlockedDecrement( &m_Size );
    }
    delete pCurrentNode;
    return true;
}

它总是在调用 Pop() 时崩溃，在 if 之后的行中，即：

pValue = pNewHeadNode->m_pValue

因为 pNewHeadNode 为 NULL，所以它炸毁了。但是怎么会这样呢？

编辑：忘记初始化代码：

template <typename T>
DRA::CommonCpp::CTBlockingQueue<T>::CTBlockingQueue():
        m_HeadCriticalSection("CTBlockingQueue<T>::m_Head"),
        m_TailCriticalSection("CTBlockingQueue<T>::m_Tail"){
    CTNode* pDummyNode = new CTNode();
    m_pHeadNode = pDummyNode;
    m_pTailNode = pDummyNode;
    m_Size = 0; //Dummy node doesn't count
}

【问题讨论】：

当你有一个互斥体时，为什么你需要互锁的递增/递减？
询问队列大小。我没有发布它，但是有一个 getter 方法
等等，还有另一个更重要的原因：唤醒在空队列中阻塞的待处理 Pop 呼叫。查看 Push() 的最后几行
你不需要那种连锁魔法。无论如何，您都可以在您命中的关键部分内获得该大小而无需互锁操作。
... 你崩溃了，因为你没有检查 pNewHeadNode 是否不是 NULL 以防你的队列为空:)

标签： c++ concurrency race-condition blockingqueue

【解决方案1】：

我的假设必须是事件设置在临界区之外，这意味着推送可能会两次通知事件。您是否尝试过在临界区设置事件？

【讨论】：

我在回答中添加了初始化代码：我创建了一个虚拟节点来避免这种情况
我明白了，我已经更新了我的答案，说明为什么 m_pHeadNode 会导致异常。
pCurrentNode 在 m_pHeadNode 更改为指向其他位置后被删除，我看不出有问题
你是对的，这不是问题所在。我的回答不正确。
再次编辑，这次是在黑暗中刺伤:(

【解决方案2】：

最后，我回到了我最初的效率较低的实现，即我发布的here，添加了一个 Finish() 方法，以便生产者可以向消费者发出优雅结束的信号，以及一个 Restart () 方法在不破坏和重新创建队列的情况下再次开始生产：

//Template definitions
template<class Element>
DRA::CommonCpp::CTBlockingQueue<Element>::CTBlockingQueue( unsigned int maxSize ):
    m_csFinished( "CTBlockingQueue::m_csFinished" ),
    m_csQueue( "CTBlockingQueue::m_csQueue" ),
    m_semElementCount( 0, maxSize ),
    m_bFinished(false){
}

template<class Element>
DRA::CommonCpp::CTBlockingQueue<Element>::~CTBlockingQueue(){
    Finish();
}

template<class Element>
void DRA::CommonCpp::CTBlockingQueue<Element>::Push( Element newElement ){
    {//RAII block
        CGuard g( m_csQueue );
        m_Queue.push( newElement );
    }
    m_semElementCount.Signal();
}

template<class Element>
bool DRA::CommonCpp::CTBlockingQueue<Element>::Pop( Element& element ){
    m_semElementCount.Wait();
    {//RAII block
        CGuard g( m_csFinished );
        if( m_bFinished ){
            CGuard g( m_csQueue );
            if ( m_Queue.size() == 0 )
                return false;
        }
    }
    {//RAII block
        CGuard g( m_csQueue );
        element = m_Queue.front();
        m_Queue.pop();
    }
    return true;
}

template<class Element>
void DRA::CommonCpp::CTBlockingQueue<Element>::Finish(){
    {//RAII block
        CGuard g( m_csFinished );
        m_bFinished = true;
    }
    {//RAII block
        CGuard g( m_csQueue );
        m_semElementCount.Signal();
    }
}

template<class Element>
void DRA::CommonCpp::CTBlockingQueue<Element>::Restart(){
    {//RAII block
        CGuard g( m_csFinished );
        m_bFinished = false;
    }
}

这不是最快的方法，但对我来说足够快。

【讨论】：