lockfreequeue.h

Go to the documentation of this file.

 
#ifndef _LOCK_FREE_QUEUE_CLASS_HEADER_
#define _LOCK_FREE_QUEUE_CLASS_HEADER_
 
#ifndef A_UTILS_TAGGED_POINTER_AVALIABLE
    #include <queue>
#endif
 
namespace A_UTILS_NS
{
 
 
#ifdef A_UTILS_TAGGED_POINTER_AVALIABLE
 
template <class T>
class A_UTILS_CAS_ALIGNMENT lock_free_queue
{
    protected:
 
        struct A_UTILS_CAS_ALIGNMENT cNode
        {
            tagged_ptr<cNode> m_pNext;  
            T m_oData;                  
        };
 
        typedef tagged_ptr<cNode> tTaggedNodePtr;
 
        tTaggedNodePtr m_pHead;
 
        tTaggedNodePtr m_pTail;
 
        free_list<cNode, T> m_oFree;
 
    public:
 
        lock_free_queue(tUInt nReserve = 0, tBool bGrow = tTrue): m_pHead(nullptr), m_pTail(nullptr), m_oFree(nReserve, bGrow)
        {
            m_pHead.Set(new cNode);
            m_pTail = m_pHead;
        }
 
        virtual ~lock_free_queue()
        {
            // clear stack
            T oTmp;
            while ((ERR_NOERROR) == Pop(&oTmp));
            delete m_pHead.GetPtr();
        }
 
        tResult Push(const T& oValue)
        {
            // get a node from the free list
            cNode* pTmp = m_oFree.GetFreeNode();
            if (!pTmp)
            {
                return (ERR_MEMORY);
            }
 
            // store the value
            pTmp->m_oData = oValue;
            // increase tag to avoid ABA problem
            pTmp->m_pNext.Set(nullptr, pTmp->m_pNext.GetTag() + 1);
 
            // and insert it into the list
            for (;;)
            {
                tTaggedNodePtr pTail = m_pTail;
 
                memory_barrier();
 
                tTaggedNodePtr pNext = pTail->m_pNext;
 
                if (pTail == m_pTail)
                {
                    if (!pNext)
                    {
                        if (pTail->m_pNext.CompareAndSwap(pNext, pTmp))
                        {
                            m_pTail.CompareAndSwap(pTail, pTmp);
                            break;
                        }
                    }
                    else
                    {
                        m_pTail.CompareAndSwap(pTail, pNext.GetPtr());
                    }
                }
            }
 
            return (ERR_NOERROR);
        }
 
        tResult Pop(T* pValue)
        {
            for (;;)
            {
                tTaggedNodePtr pHead = m_pHead;
 
                memory_barrier();
 
                tTaggedNodePtr pTail = m_pTail;
 
                memory_barrier();
 
                cNode* pNext = pHead->m_pNext.GetPtr();
 
                if (pHead == m_pHead)
                {
                    if (pHead.GetPtr() == pTail.GetPtr())
                    {
                        if (pNext == nullptr)
                        {
                            return (ERR_EMPTY);
                        }
 
                        m_pTail.CompareAndSwap(pTail, pNext);
                    }
                    else
                    {
                        if (pNext == nullptr)
                        {
                            continue;
                        }
 
                        *pValue = pNext->m_oData;
                        if (m_pHead.CompareAndSwap(pHead, pNext))
                        {
                            m_oFree.AddFreeNode(pHead.GetPtr());
                            break;
                        }
                    }
                }
            }
 
            return (ERR_NOERROR);
        }
 
    private:
        lock_free_queue(const lock_free_queue& oOther);
};
 
#else
 
template <class T>
class lock_free_queue
{
    protected:
        std::queue<T>    m_oQueue;
        std::recursive_mutex m_oMutex;
    public:
 
        lock_free_queue(tUInt nReserve = 0, tBool bGrow = tTrue)
        {
        }
 
        virtual ~lock_free_queue()
        {
        }
 
        tResult Push(const T& oValue)
        {
            std::lock_guard<std::recursive_mutex> oLck(m_oMutex);
            m_oQueue.push(oValue);
            return (ERR_NOERROR);
        }
 
        tResult Pop(T* pValue)
        {
            std::lock_guard<std::recursive_mutex> oLck(m_oMutex);
            if (m_oQueue.empty())
            {
                return (ERR_EMPTY);
            }
            *pValue = m_oQueue.front();
            m_oQueue.pop();
            return (ERR_NOERROR);
        }
 
    private:
        lock_free_queue(const lock_free_queue& oOther);
};
 
#endif
 
}
 
#endif