在无锁FIFO缓冲区中删除了节点检测

Question

我一直在处理无锁C 11 FIFO缓冲区。而且我几乎明白了。但是，一个小细节使我变得更好。缓冲区的头指向：

std::shared_ptr<node<T>> m_head;

类型：

    struct node
    {
        node(const T data)
            :
            data(new T(data)),
            next(nullptr)
        {}
        std::shared_ptr<T> data;
        std::shared_ptr<node<T>> next;
    };

然后有农产品：

    void produce(const T &&data)
    {
        //bool indicating whether a notification should be sent after adding
        bool l_notifyUponAdding;
        //the new node to be added at the end of the array
        std::shared_ptr<node<T>> l_newNode(new node<T>(std::forward<const T&&>(data)));
        //pointer to the last node
        std::shared_ptr<node<T>> l_lastNode(std::atomic_load(&m_head));
        //value to compare the next of the last node with
        std::shared_ptr<node<T>> l_expectedNullPointer;
        //notify if this isn't the only node
        l_notifyUponAdding = !l_lastNode;
        if (!l_lastNode)//if there are no nodes, add this as the only node
        if (std::atomic_compare_exchange_strong(&m_head, &l_expectedNullPointer, l_newNode))
            return;
        do
        {
            l_expectedNullPointer.reset();
            while (l_lastNode->next)
            {
                l_lastNode = std::atomic_load(&l_lastNode)->next;
            }
        } while (!std::atomic_compare_exchange_weak(&l_lastNode->next, &l_expectedNullPointer, l_newNode));
        //adding failed since another thread already did this. 
        l_lastNode = l_expectedNullPointer;

        if (l_notifyUponAdding)
            m_newDataWaiter.notify_one();
        }    

和消费：

        std::shared_ptr<T> consume(bool blockingCall = false)
        {
            //Check if the head is null if it is:
            if (!std::atomic_load(&m_head))
            {
                if (blockingCall)//And this is a blocking call,
                {
                    do
                    {
                        m_newDataWaiter.wait(m_newDataWaiterLock, [this]{return std::atomic_load(&(this->m_head)) == nullptr; });//we block until
                    } while (!std::atomic_load(&m_head));// the load yields a head that is not null(to avoid unnecessary calls on spurious wake ups)
                }
                else//And this is not a blocking call we 
                {
                    return nullptr;
                }
            }
        //If we've found a valid head we will now try to make the node pointed to by head the new head. 
        std::shared_ptr<node<T>> l_poppee = atomic_load(&m_head);
        std::shared_ptr<node<T>> l_newHead = atomic_load(&m_head);
        //note that l_poppee gets updated if the compare exchange fails
        while (l_poppee && !std::atomic_compare_exchange_weak(&m_head, &l_poppee, l_poppee->next))
        {
        }
        if (l_poppee)
            return l_poppee->data;
        else
            return std::shared_ptr<T>();
    }

功能。

似乎一切正常。但是我认为存在一个缺陷。如果在执行produce时消耗所有节点。数据将添加到最后一个元素中。即使该元素已经被删除。

如果执行此行，请更加精确：

if (std::atomic_compare_exchange_strong(&m_head, &l_expectedNullPointer, l_newNode))

和加载节点不是零。最后一个节点的下一个元素将更改。无论是否同时删除节点。只要有共同的指示，只要拒绝产生功能，节点就不会被物理删除。

但是，主要指针将设置为null。因此，一旦退出生产功能，新节点将立即删除。

是否有人碰巧知道这个问题的解决方案：）？

Answer 1

通过在列表中保留虚拟节点，始终在无锁列表中解决此情况。头总是指向虚拟节点，这是列表。

当队列变为空时，头部和尾部都指向虚拟节点。

您可以查看http://www.research.ibm.com/people/m/michael/podc-1996.pdf以获取详细信息，这样我就不会歪曲该概念，因为它很容易从文章中挑选出来。