lockfree 的队列的实现

大家好，又见面了，我是你们的朋友全栈君。

一个高速无锁循环队列的实现。

需要注意的是:

(1)队列的大小（m_lMaxQueueSize）应该足够的大，避免处理不过来时，找半天找不到空位置。

(2)还有一点是这种队列在push数据足够快时效率高点，不然pop时就阻塞了，改善的方式就是使用费阻塞的方式，判断几次没有就跳出去，还有这种队列也就在push数据足够快时效率高点，不然判断的次数就多了。

(3)使用了原子操作的锁

(4)需要优化的的是数值开始与结束之间长度小于阈值就不取。

代码如下：

// 临界锁，线程安全  
// 必须要有一个不会用的空值
typedef unsigned int var_4;
typedef unsigned long var_u8;

template <class T_Key>  //队列里的值
class CQueue_Lockfree  
{  
public:  
    CQueue_Lockfree()  
    {  
        m_lMaxQueueSize = -1;  
        m_tpQueue = NULL;  //队列的内存
        m_lBegPos = 0;  //队列的开始位置
        m_lEndPos = 0;  //队列的结束位置
    };  
    ~CQueue_Lockfree()  
    {  
        if(m_tpQueue)  
            delete m_tpQueue;  
    };  
    var_4 InitQueue(var_4 lMaxQueueSize, T_Key null)//null为空的值（目前为0）
    {  
        m_lMaxQueueSize = lMaxQueueSize;  
        m_tpQueue = new T_Key[m_lMaxQueueSize];  
        if(m_tpQueue == NULL)  
            return -1;  
        m_lBegPos = 0;  
        m_lEndPos = 0;  
        m_null = null;  
        for( var_4 i = 0; i < lMaxQueueSize; i++ )  
        {  
            m_tpQueue[i] = null;  
        }  
        return 0;  
    };  
  
    void ResetQueue()  
    {  
        m_lBegPos = 0;  
        m_lEndPos = 0;    
    };  
  
    void ClearQueue()  
    {  
        if(m_tpQueue)  
        {  
            delete m_tpQueue;  
            m_tpQueue = NULL;  
        }  
        m_lMaxQueueSize = -1;  
        m_lBegPos = 0;  
        m_lEndPos = 0;  
    };  
    void PushData(T_Key tKey)  
    {  
        register var_u8 cnt = 1;  
        var_u8 pos = fetch_and_add(&m_lEndPos, cnt)%m_lMaxQueueSize;  //队列的m_lEndPos位置下一个位置
        while( m_null != m_tpQueue[pos] )  //等到该位置为空才能放数值
            cp_sleep(1);  
        m_tpQueue[pos] = tKey;  
    };  
  
    T_Key PopData()  
    {  
        register var_u8 cnt =1;  
        var_u8 pos = fetch_and_add(&m_lBegPos, cnt)%m_lMaxQueueSize;  //队列的m_lBegPos位置下一个位置
        T_Key *p = m_tpQueue + pos;  
        while( m_null == *p )  //等到该位置不为空才能取数值
            cp_sleep(1);  
  
        T_Key ret = *p;  
        *p = m_null;  
        return ret;  
    };  
private:  
    var_4 m_lMaxQueueSize;  
    T_Key* m_tpQueue;  
    T_Key m_null;  
  
    var_u4 m_lBegPos;  
    var_u4 m_lEndPos;  
};  

别人的测试结果，这种方式比用锁少了40%。

10亿次放入和读取，10线程花费200秒。

测试说加了阈值判断后说是8核1.5GHz花了88秒，没加前是

24核机器200秒

vendor_id : GenuineIntel cpu family : 6 model : 45 model name : Intel(R) Xeon(R) CPU E5-2630 0 @ 2.30GHz stepping : 7 cpu MHz : 2300.469 cache size : 15360 KB

个人觉得主要是因为两个sleep 阻塞了，还是换成非阻塞的方法吧，push的数据加一个等待队列，pop的方法不要用sleep，要非阻塞的，没有要取的数据就跳出去干别的活去。

上面是用两个数来记录开始和结束的下标，我觉得用两个指针也可以。另外还有一个改进是使用引用计数，管理下内存，免得一个节点被重复释放或加入，对于引用计数的加减就只能用下原子函数来保证原子性了，下文提到的ABA 的问题，个人觉得也就只有在内存被重复释放或添加时有影响，所以引用计数是个比较好的方案，也可以避免重复内容的拷贝，效率还是不错的。还有提一下的是如果使用C++11中的 STL 中的 atomic 类的函数和可以跨下平台，不用的人自己加些宏定义也可以，没什么特别的，说是无锁其实也就是应用层的无锁，使用了下原子操作来进行很小颗粒的锁操作。

关于原子操作：

关于 CAS 等原子操作

　　在开始说无锁队列之前，我们需要知道一个很重要的技术就是 CAS 操作——Compare & Set，或是 Compare & Swap，现在几乎所有的 CPU 指令都支持 CAS 的原子操作，X86下对应的是 CMPXCHG 汇编指令。有了这个原子操作，我们就可以用其来实现各种无锁（lock free）的数据结构。

　　这个操作用C语言来描述就是下面这个样子：（代码来自 Wikipedia 的 Compare And Swap 词条）意思就是说，看一看内存*reg 里的值是不是 oldval，如果是的话，则对其赋值 newval。

int compare_and_swap (int* reg, int oldval, int newval)
{
  int old_reg_val = *reg;
  if (old_reg_val == oldval)
     *reg = newval;
  return old_reg_val;
}

　这个操作可以变种为返回 bool 值的形式（返回 bool 值的好处在于，可以调用者知道有没有更新成功）：

bool compare_and_swap (int *accum, int *dest, int newval)
{
  if ( *accum == *dest ) {
      *dest = newval;
      return true;
  }
  return false;
}

与 CAS 相似的还有下面的原子操作：（这些东西大家自己看 Wikipedia 吧）

• Fetch And Add，一般用来对变量做 +1 的原子操作
• Test-and-set，写值到某个内存位置并传回其旧值。汇编指令 BST
• Test and Test-and-set，用来低低 Test-and-Set 的资源争夺情况

　　注：在实际的C/C++程序中，CAS 的各种实现版本如下：

　　1）GCC 的 CAS

　　GCC4.1+ 版本中支持 CAS 的原子操作（完整的原子操作可参看 GCC Atomic Builtins）

bool __sync_bool_compare_and_swap (type *ptr, type oldval type newval, ...)
type __sync_val_compare_and_swap (type *ptr, type oldval type newval, ...)

　2）Windows 的 CAS

　　在 Windows 下，你可以使用下面的 Windows API 来完成 CAS：（完整的 Windows 原子操作可参看 MSDN 的 InterLocked Functions）

InterlockedCompareExchange ( __inout LONG volatile *Target,
                                __in LONG Exchange,
                                __in LONG Comperand);

　　3) C++11 中的 CAS

　　C++11中的 STL 中的 atomic 类的函数可以让你跨平台。（完整的C++11的原子操作可参看Atomic Operation Library）

template< class T >
bool atomic_compare_exchange_weak ( std::atomic<T>* obj,
                                   T* expected, T desired );
template< class T >
bool atomic_compare_exchange_weak ( volatile std::atomic<T>* obj,
                                   T* expected, T desired );

无锁队列的链表实现

　　下面的东西主要来自 John D. Valois 1994 年 10 月在拉斯维加斯的并行和分布系统系统国际大会上的一篇论文——《Implementing Lock-Free Queues》。

　　我们先来看一下进队列用 CAS 实现的方式：

EnQueue (x) //进队列
{
    //准备新加入的结点数据
    q = new record ();
    q->value = x;
    q->next = NULL;
    do {
        p = tail; //取链表尾指针的快照
    } while( CAS (p->next, NULL, q) != TRUE); //如果没有把结点链上，再试
 
    CAS (tail, p, q); //置尾结点
}

　　我们可以看到，程序中的那个 do- while 的 Re-Try-Loo。就是说，很有可能我在准备在队列尾加入结点时，别的线程已经加成功了，于是 tail 指针就变了，于是我的 CAS 返回了 false，于是程序再试，直到试成功为止。这个很像我们的抢电话热的不停重播的情况。

　　你会看到，为什么我们的“置尾结点”的操作不判断是否成功，因为：

1. 如果有一个线程 T1，它的 while 中的 CAS 如果成功的话，那么其它所有的随后线程的 CAS 都会失败，然后就会再循环，
2. 此时，如果 T1 线程还没有更新 tail 指针，其它的线程继续失败，因为 tail->next 不是 NULL 了。
3. 直到 T1 线程更新完 tail 指针，于是其它的线程中的某个线程就可以得到新的 tail 指针，继续往下走了。

　　这里有一个潜在的问题——如果 T1 线程在用 CAS 更新 tail 指针的之前，线程停掉了，那么其它线程就进入死循环了。下面是改良版的 EnQueue ()

EnQueue (x) //进队列改良版
{
    q = new record ();
    q->value = x;
    q->next = NULL;
 
    p = tail;
    oldp = p
    do {
        while (p->next != NULL)
            p = p->next;
    } while( CAS (p.next, NULL, q) != TRUE); //如果没有把结点链上，再试
 
    CAS (tail, oldp, q); //置尾结点
}

　　我们让每个线程，自己 fetch 指针 p 到链表尾。但是这样的 fetch 会很影响性能。而通实际情况看下来，99.9% 的情况不会有线程停转的情况，所以，更好的做法是，你可以接合上述的这两个版本，如果 retry 的次数超了一个值的话（比如说 3 次），那么，就自己 fetch 指针。

　　好了，我们解决了 EnQueue，我们再来看看 DeQueue 的代码：（很简单，我就不解释了）

DeQueue () //出队列
{
    do{
        p = head;
        if (p->next == NULL){
            return ERR_EMPTY_QUEUE;
        }
    while( CAS (head, p, p->next);
    return p->next->value;
}

　我们可以看到，DeQueue 的代码操作的是 head->next，而不是 head 本身。这样考虑是因为一个边界条件，我们需要一个 dummy 的头指针来解决链表中如果只有一个元素，head 和 tail 都指向同一个结点的问题，这样 EnQueue 和 DeQueue 要互相排斥了。

注：上图的 tail 正处于更新之前的装态。

DAS 的 ABA 问题

　　所谓 ABA（见维基百科的 ABA 词条），问题基本是这个样子：

1. 进程 P1 在共享变量中读到值为A
2. P1被抢占了，进程 P2 执行
3. P2把共享变量里的值从A改成了B，再改回到A，此时被 P1 抢占。
4. P1回来看到共享变量里的值没有被改变，于是继续执行。

　　虽然 P1 以为变量值没有改变，继续执行了，但是这个会引发一些潜在的问题。ABA 问题最容易发生在 lock free 的算法中的，DAS 首当其冲，因为 DAS 判断的是指针的地址。如果这个地址被重用了呢，问题就很大了。

　　比如上述的 DeQueue ()函数，因为我们要让 head 和 tail 分开，所以我们引入了一个 dummy 指针给 head，当我们做 CAS 的之前，如果 head 的那块内存被回收并被重用了，而重用的内存又被 EnQueue ()进来了，这会有很大的问题。（内存管理中重用内存基本上是一种很常见的行为）

　　这个例子你可能没有看懂，维基百科上给了一个活生生的例子——

你拿着一个装满钱的手提箱在飞机场，此时过来了一个火辣性感的美女，然后她很暖昧地挑逗着你，并趁你不注意的时候，把用一个一模一样的手提箱和你那装满钱的箱子调了个包，然后就离开了，你看到你的手提箱还在那，于是就提着手提箱去赶飞机去了。

　　这就是 ABA 的问题。

　　解决 ABA 的问题

　　维基百科上给了一个解——使用 double-CAS（双保险的 CAS），例如，在 32 位系统上，我们要检查 64 位的内容

　　1）一次用 CAS 检查双倍长度的值，前半部是指针，后半部分是一个计数器。

　　2）只有这两个都一样，才算通过检查，要吧赋新的值。并把计数器累加1。

　　这样一来，ABA 发生时，虽然值一样，但是计数器就不一样（但是在 32 位的系统上，这个计数器会溢出回来又从 1 开始的，这还是会有 ABA 的问题）

　　当然，我们这个队列的问题就是不想让那个内存重用，这样明确的业务问题比较好解决，论文《Implementing Lock-Free Queues》给出一这么一个方法——使用结点内存引用计数 refcnt！

SafeRead (q)
{
    loop:
        p = q->next;
        if (p == NULL){
            return p;
        }
 
        Fetch&Add (p->refcnt, 1);
 
        if (p == q->next){
            return p;
        }else{
            Release (p);
        }
    goto loop
}

　　其中的 Fetch&Add 和 Release 分是是加引用计数和减引用计数，都是原子操作，这样就可以阻止内存被回收了。

　　用数组实现无锁队列

　　本实现来自论文《Implementing Lock-Free Queues》

　　使用数组来实现队列是很常见的方法，因为没有内存的分部和释放，一切都会变得简单，实现的思路如下：

　　1）数组队列应该是一个 ring buffer 形式的数组（环形数组）

　　2）数组的元素应该有三个可能的值：HEAD，TAIL，EMPTY（当然，还有实际的数据）

　　3）数组一开始全部初始化成 EMPTY，有两个相邻的元素要初始化成 HEAD 和 TAIL，这代表空队列。

　　4）EnQueue 操作。假设数据x要入队列，定位 TAIL 的位置，使用 double-CAS 方法把(TAIL, EMPTY) 更新成 (x， TAIL)。需要注意，如果找不到(TAIL, EMPTY)，则说明队列满了。

　　5）DeQueue 操作。定位 HEAD 的位置，把(HEAD, x)更新成(EMPTY, HEAD)，并把x返回。同样需要注意，如果x是 TAIL，则说明队列为空。

　　算法的一个关键是——如何定位 HEAD 或 TAIL？

　　1）我们可以声明两个计数器，一个用来计数 EnQueue 的次数，一个用来计数 DeQueue 的次数。

　　2）这两个计算器使用使用 Fetch&ADD 来进行原子累加，在 EnQueue 或 DeQueue 完成的时候累加就好了。

　　3）累加后求个模什么的就可以知道 TAIL 和 HEAD 的位置了。

　　如下图所示：

　 小结

　　以上基本上就是所有的无锁队列的技术细节，这些技术都可以用在其它的无锁数据结构上。

　　1）无锁队列主要是通过 CAS、FAA 这些原子操作，和 Retry-Loop 实现。

　　2）对于 Retry-Loop，我个人感觉其实和锁什么什么两样。只是这种“锁”的粒度变小了，主要是“锁”HEAD 和 TAIL 这两个关键资源。而不是整个数据结构。

　　还有一些和 Lock Free 的文章你可以去看看：

• Code Project 上的雄文 《Yet another implementation of a lock-free circular array queue》
• Herb Sutter 的《Writing Lock-Free Code: A Corrected Queue》– 用C++11的 std::atomic 模板。

发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/161685.html原文链接：https://javaforall.cn