无锁队列的实现

无锁队列的实现

关于无锁队列的实现,网上有很多文章,虽然本文可能和那些文章有所重复,但是我还是想以我自己的方式把这些文章中的重要的知识点串起来和大家讲一讲这个技术。下面开始正文。

关于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吧)

注:在实际的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* obj,
                                   T* expected, T desired );
template< class T >
bool atomic_compare_exchange_weak( volatile std::atomic* 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-Loop。就是说,很有可能我在准备在队列尾加入结点时,别的线程已经加成功了,于是tail指针就变了,于是我的CAS返回了false,于是程序再试,直到试成功为止。这个很像我们的抢电话热线的不停重播的情况。

你会看到,为什么我们的“置尾结点”的操作(第12行)不判断是否成功,因为:

  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) != TRUE );
    return p->next->value;
}

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

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

CAS的ABA问题

所谓ABA(见维基百科的ABA词条),问题基本是这个样子:

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

虽然P1以为变量值没有改变,继续执行了,但是这个会引发一些潜在的问题。ABA问题最容易发生在lock free 的算法中的,CAS首当其冲,因为CAS判断的是指针的地址。如果这个地址被重用了呢,问题就很大了。(地址被重用是很经常发生的,一个内存分配后释放了,再分配,很有可能还是原来的地址)

比如上述的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的文章你可以去看看:

注:我配了一张look-free的自行车,寓意为——如果不用专门的车锁,那么自行得自己锁自己!

 (全文完)


关注CoolShell微信公众账号可以在手机端搜索文章

(转载本站文章请注明作者和出处 酷 壳 – CoolShell ,请勿用于任何商业用途)

——=== 访问 酷壳404页面 寻找遗失儿童。 ===——
好烂啊有点差凑合看看还不错很精彩 (18 人打了分,平均分: 4.67 )
Loading...

无锁队列的实现》的相关评论

  1. @vingc @aa
    如果只有两个线程,那么楼主给的 solution 是正确的,只有第二个线程需要更新 tail,已经 hang 住的线程无须完成此操作
    @Jumbal @陈皓
    但是改良版的 EnQueue 存在如下 bug:
    1)thread A 成功入队元素 a(此时队列中仅有a),hang 在 13行,未完成 tail 更新
    2)thread B 和 thread C 进入 EnQueue,获得相同的 oldp
    3)thread B 成功入队元素 b(此时队列:a->b),并更新 tail 指向 b
    4)thread C 成功入队元素 c(此时队列:a->b->c),但由于其 oldp 与当前 tail(指向b)不符,故更新 tail 操作失败,而这种行为是不符合预期的

    这里的本质在于,任何需要保证一致性的并行系统都需要某种形式的同步,只是粒度不同。
    CAS也好,mutex也好,都只是具体的同步方式
    原版中的 EnQueue 操作 do … while 循环处的 CAS 就是用来完成同步的,而改良版破坏了这种同步

  2. 是啊,并发算法很重要,感觉这个东西未来很可能重定义OS;
    不过引用的例子不是非常恰当,目前最为稳定的算法似乎是:
    A practical nonblocking queue algorithm using compare_and_swap

  3. 这博文早就收藏,最近开始究Lock-Free编程,于是翻出来细读。
    发现ABA问题还是不太清楚,所以发帖提问一下。

    个人觉得DeQueue不会发生ABA,改良版EnQueue才发生ABA。
    因为DeQueue的
    8行:while( CAS(head, p, p->next) != TRUE );
    即使head被pop出去后又push回来,但由于它赋的值是head的next指针,所以本身没有问题,但
    9行:return p->next->value;
    这句有问题,因为可能return前被强占,p->next被free()掉,所以pop出来的值不是CAS时的值。
    另一个方案是把p->next->value在while循环里取出放到局部变量过后再return,但可能由于ABA问题可能导致返回值和内存的不一样。这个局面是由自己的内存由他人free导致的。
    而改良版EnQueue的ABA问题,
    14行:CAS(tail, oldp, q);
    例如,线程T1执行push(B)时tail指向A,在CAS(tail, A, B)前被线程T2强占,T2把所有元素pop,然后push(A),接着在push(c)时也是在执行CAS(tail, A, C)前被T1强占回来,这时tail也是指向A,于是T1执行CAS(tail, A, B),这导致tail指向错误的内存。而原始版EnQueue可以把
    12行:CAS(tail, p, q);
    改成CAS(tail, p, p->next);避免ABA问题。

    不知我的想法有没有问题。

  4. 改进型EnQueue()如果在进入do{ … }while循环就被阻塞,此时DeQueue()不断推出head并释放节点,岂不是EnQueue会出现非法指针?感觉来个CAS(p, tail, tail)比较稳当点。

  5. 其实enQ/deQ里面的dowhile检测就是在枷锁,其实感觉所谓的无锁只不过是降低了锁的粒度。

  6. if ( *accum == *dest ) {
    *dest = newval;
    return true;
    }
    如果一个线程T1刚通过了IF判断。CPU时间片给了另一个线程,并且另一个线程把accum改变了。这样还是有问题的啊。

  7. 改良版的Enque不能保证 最后一行CAS(tail, oldp, q); 正确
    假设线程t1先抢到tail,插入一个节点 然后在最一行被执行前被停住。
    这时t2线程往最后加入一个节点,因为你改良的代码使得它能在t1更新tail前做到插入新节点。
    然后t2也来到最后一行,这时如果t1更新了tail,t2本来又更新tail,显然就出错了。

    John D. Valois的论文中不是这样做的。

    他的做法是直接在while中也更新tail,这样能保证tail的正确性,也能处理文中所说的异常。

    EnQueue(x)
    {
    q = new record();
    q->value = x;
    q->next = NULL;

    p = tail;
    oldp = p

    do {
    p = tail;
    suc = CAS(p.next, NULL, q);
    if suc == false
    CAS(tail, p, p->next);
    } while( suc == false); //如果没有把结点链在尾上,再试

    CAS(tail, p, q); //置尾结点
    }

  8. stack :
    改良版的Enque不能保证 最后一行CAS(tail, oldp, q); 正确
    假设线程t1先抢到tail,插入一个节点 然后在最一行被执行前被停住。
    这时t2线程往最后加入一个节点,因为你改良的代码使得它能在t1更新tail前做到插入新节点。
    然后t2也来到最后一行,这时如果t2先更新了tail,t1之后更新tail,显然就出错了。
    John D. Valois的论文中不是这样做的。
    他的做法是直接在while中也更新tail,这样能保证tail的正确性,也能处理文中所说的异常。
    EnQueue(x)
    {
    q = new record();
    q->value = x;
    q->next = NULL;
    p = tail;
    oldp = p
    do {
    p = tail;
    suc = CAS(p.next, NULL, q);
    if suc == false
    CAS(tail, p, p->next);
    } while( suc == false); //如果没有把结点链在尾上,再试
    CAS(tail, p, q); //置尾结点
    }

  9. 我也觉得改良版存在问题。

    进程1 插入数据q1, 循环执行后,tail->next = q1. 然后被进程2抢占。
    进程2 插入数据q2, 开始执行, while(p->next != NULL)成立, p等于q1, 然后q1->next = q2. 被进程1被抢占。

    进程1 判断tail == oldp, 将tail改为q1。
    进程2 判断tail != oldp, 不改变tail。

    这时候tail不就错了吗?

    是不是我那个地方忽略了?

  10. @stack
    如果每一次tail都会从内存读,那么程序是没有问题的,包括陈浩的。但如果考虑到Cache,你的也有同样的问题。tail为老的值,自然与你的临时变量p的值相同。

  11. 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); //置尾结点
    }
    最后这个CAS是不是没必要,直接 tail = q就行了??

  12. DeQueue() //出队列
    {
    do{
    p = head;
    if (p->next == NULL){
    return ERR_EMPTY_QUEUE;
    }
    while( CAS(head, p, p->next) != TRUE );
    return p->next->value;
    }
    出队列的这个,通过head->next是否为null来判断队列是否为空,最后return的也是head->next的data,说明head应该是个dummy node. 但是下面的逻辑好像又把head当作正常的数据节点处理。如果head对应的是dummy节点,代码是否应该为:
    DeQueue() //出队列
    {
    do{
    next = head->next;
    if (next == NULL){
    return ERR_EMPTY_QUEUE;
    }
    while( CAS(head->next, next, next->next) != TRUE );
    return p->next->value;
    }

  13. 请问下耗子老师,执行ASM LOCK前缀锁总线时,在这个时候,是锁住整个内存不被其它CPU访问?还是只锁住内存中被访问的某一段内存空间不被其它CPU访问?

  14. stack说的是对的,但valois的实现中还存有其他的问题,并继续存在于其后一篇引用更广泛的论文中《Lock-Free Linked Lists Using Compare-and-Swap》中。后来分别由micheal《Correction of a Memory Management Method for Lock-Free Data Structures》和本人《ERRATA_Lock-Free_Linked_Lists_Using_Compare-and-Swap》做出修订。
    虽然似乎valois此后就销声匿迹,但不能否认他的论文第一次给出了工程级别的细粒度无锁并行化的实现,并就此开创了应用无锁并行化的崭新局面,其中很多技巧在今天仍然具有借鉴意义。

    另外补充一句,细粒度无锁并行化的核心在于帮助系统的存在,stack的代码中循环内部CAS的存在就是保证自己继续的前提是别人应该完成的工作必须已经完成,或者代为完成。无论lock-free还是wait-free,就是要实现这样的一种系统。

  15. 通过pthread_mutex_lock 和 cas在8核CPU,8G内存的服务器上做了测试,发现cas只是有微弱的优势,跟预期有些差距…
    另外,之前以为pthread_mutex_lock的底层实现应该也是通过lock指令,不过看汇编代码,pthread_mutex_lock 实际并未用lock指令,不知道具体是什么实现机制,耗子哥是否可以解下惑?

发表评论

电子邮件地址不会被公开。 必填项已用*标注

*