C++ 無鎖隊列實現

上源碼

#ifndef __GLOBAL_LOCK_FREE_QUEUE_H__
#define __GLOBAL_LOCK_FREE_QUEUE_H__

#include <atomic>
#include <list>

#ifdef _WINDOWS
#include <windows.h>
//PVOID __cdecl InterlockedCompareExchangePointer(
//    _Inout_ PVOID volatile *Destination,
//    _In_    PVOID          Exchange,
//    _In_    PVOID          Comparand
//    );
//
//Parameters
//    Destination [in, out]
//A pointer to a pointer to the destination value.
//    Exchange [in]
//The exchange value.
//    Comparand [in]
//The value to compare to Destination.
//    Return value
//    The function returns the initial value of the Destination parameter.
//    Remarks
//    The function compares the Destination value with the Comparand value. If the Destination value is equal to the Comparand value, the Exchange value is stored in the address specified by Destination. Otherwise, no operation is performed.
//    On a 64-bit system, the parameters are 64 bits and must be aligned on 64-bit boundaries; otherwise, the function will behave unpredictably. On a 32-bit system, the parameters are 32 bits and must be aligned on 32-bit boundaries.
//    The interlocked functions provide a simple mechanism for synchronizing access to a variable that is shared by multiple threads. This function is atomic with respect to calls to other interlocked functions.
#define __sync_bool_compare_and_swap(a,b,c) (InterlockedCompareExchangePointer((void*volatile*)a,c,b), (*a)==(c))
#endif

namespace DataCache
{

    template <typename T>
    class LinkList
    {
    public:
        T data;
        LinkList<T> *next;
    };// class LinkeList<T>

    template <typename T>
    class LockFreeQueue
    {
    public:
        LockFreeQueue();

        void push_back(T t);

        T pop_front(void);

        bool isEmpty(void);

        int GetLength();

    private:
        LinkList<T> *head_;
        LinkList<T> *tail_;
        std::_Atomic_integral_t elementNumbers_;
    }; // class LockFreeQueue

    template <typename T>
    LockFreeQueue<T>::LockFreeQueue()
        :head_(NULL),
        tail_(new LinkList<T>),
        elementNumbers_(0)
    {
        head_ = tail_;
        tail_->next = NULL;
    }

    template <typename T>
    void LockFreeQueue<T>::push_back(T t)
    {
        auto newVal = new LinkList<T>;
        newVal->data = t;
        newVal->next = NULL;

        LinkList<T> *p;
        do
        {
            p = tail_;
        } while (!__sync_bool_compare_and_swap(&tail_->next, NULL, newVal));

        //move tail_
        __sync_bool_compare_and_swap(&tail_, tail_, newVal);
        elementNumbers_++;
    }

    template <typename T>
    T LockFreeQueue<T>::pop_front()
    {
        LinkList<T> *p;

        do
        {
            //record the first node.
            p = head_->next;
            if (!p)
            {
                return 0;
            }
        } while (!__sync_bool_compare_and_swap(&head_->next, p, p->next));

        if (elementNumbers_ > 0) elementNumbers_--;
        if (elementNumbers_ == 0)
        {
            // if the queue is empty then the tail to header.
            do
            {

            } while (!__sync_bool_compare_and_swap(&tail_, p, head_));
        }

        return p->data;
    }

    template <typename T>
    bool LockFreeQueue<T>::isEmpty()
    {
        if (elementNumbers_ == 0)
        {
            return true;
        }
        else
        {
            return false;
        }
    }

    template <typename T>
    int LockFreeQueue<T>::GetLength()
    {
        return elementNumbers_;
    }

}// namespace DataCache

#endif

 

 

源碼解析：

重點在函數InterlockedCompareExchangePointer(a,c,b),其作用為如果a==b,則修改a的值為c，這是微軟提供的一個原子操作的API，這里面通過定義這個宏，來進行queue入隊列和出隊列時的指針修改，達到無鎖操作的目的，可以大幅提高隊列的操作性能，避免過程中進行加鎖操作。在GCC中已經原生支持了__sync_bool_compare_and_swap，所以這個宏的作用是針對MSVC版本提供的類似函數。

 

測試代碼：

#include"lockFreeQueue.h"

DataCache::LockFreeQueue<int> queue;

void func1(void)
{
    for(int i = 0; i <100000 ; i++)
    {
        queue.push_back(i);
    }
}

void func2(void)
{
    for(int i = 100000; i < 200000; i++)
    {
        queue.push_back(i);
    }
}

void func5(void)
{
    for(int i = 200000; i < 300000; i++)
    {
        queue.push_back(i);
    }
}

void func4(void)
{
    for(int i = 400000; i < 500000; i++)
    {
        queue.push_back(i);
    }
}

void func3(void)
{
    while (!queue.isEmpty())
    {
        std::cout<<queue.pop_front()<<" ";
    }

    std::cout<<std::endl;
}

int main(int argc,char **argv)
{
    std::thread t1(func1);
    std::thread t2(func2);
    std::thread t5(func5);
    std::thread t6(func4);

    t1.join();
    t2.join();
    t5.join();
    t6.join();

    int length=queue.GetLength();
    assert(length==500000);
    return 0;
}