Compare And Swap（CAS）实现无锁多生产者

1、CAS 原理

compare and swap，解决多线程并行情况下使用锁造成性能损耗的一种机制，CAS操作包含三个操作数——内存位置（V）、预期原值（A）和新值(B)。如果内存位置的值与预期原值相匹配，那么处理器会自动将该位置值更新为新值。否则，处理器不做任何操作。无论哪种情况，它都会在CAS指令之前返回该位置的值。CAS有效地说明了“我认为位置V应该包含值A；如果包含该值，则将B放到这个位置；否则，不要更改该位置，只告诉我这个位置现在的值即可。
当同时存在读写线程时，默认情况下是不保证线程安全的，因而需要利用信号量来进行线程同步（Synchronization），如关键代码段、互斥体等， 同时操作系统也提供了相应的API。然而同步并不总是满足条件的且有效率的，比如陷入内核时会有性能损失、死锁、活锁以及资源浪费等。

于是Lock-Free和Wait-Free的思想出现了，由于此时不存在读写线程的同步，因而在写线程运行时，读线程也在运行（多核中两个线程在不同的核上被调度运行），而且代码量减少，程序运行更快。而这一思想是通过CAS机制来实现，如下

template<typename T>
bool CAS(T* ptr, T expected, T fresh)
{
    if(*ptr != expected)
         return false;
    *ptr = fresh;
    return true;
}

CAS的原理是，将旧值与一个期望值进行比较，如果相等，则更新旧值，类型T = {char, short, int, __int64, …}等，以及指针（pointer to any type）。
注意CAS这里只是说明了原理，并不是真实的源代码实现，具体实现请参考操作系统。
在Windows API中，提供了很多原子操作（Atomic Operatoration），如InterlockedCompareExchange等一系列InterLocked函数，从汇编的角度来 讲，intel的XCHG指令即可以一个时钟周期内完成数据的交换（寄存器和内存的数据交换），使用方法可参考 InterlockedCompareExchange的反汇编代码。
考虑这样一种情况：存在多个读线程和一个写线程，在使用同步方法时， 很可能写线程并不能立即获得锁，最坏的情况下是写线程永远得不到锁，即进入活锁状态。但是使用CAS的方法时，便可以让读写线程并行运行，当写线程一旦更 新为新的共享数据时，读线程便能即时读出更新后的数据。

class Widget
{
    Data* p_;
    ...
    void Use() { ... use p_ ... }
    void Update() {
    Data * pOld, * pNew = new Data;
    do
    {
    pOld = p_;
    ...
    }while (!CAS(&p_, pOld, pNew));
    }
};

但随之而来会有一个疑问，Update函数中该何时删除旧数据呢，由于很有可能有别的读线程在使用旧数据。对于JAVA等有自动内存回收（GC）机制的语言环境而言，这不是问题，但对于C/C++等无GC机制的环境而言，旧数据的回收就比较棘手的问题了。
当然也存在很多的解决方法，这也成为CAS机制中最有趣最受讨论的问题，而且在不同条件下方法也不同。

2、实现无锁多生产者

struct node{
struct node *next;
int data;
}

struct node *queue;//队列头

多个消费者（多线程）都需要向这个queue插入数据

为了说明问题的复杂性，先看看只有一个消费者时的情况，插入队列的操作非常简单：
Step1) new_head->next = queue->head;
Step2) queue->head = new_head;
加入了多线程，问题变得复杂，以step 2为例，多个线程可能会同时进行这个操作，因此结果是不可预知的。

解决办法1）任何线程在进行step1之前先获取锁，得到step 2完成后再释放锁，这种办法是最简单的，但锁的性能开销较大，还可以考虑改进的办法。

一个比较妙的思路是：

每次操作之前先确认别的生产者没有在改变队列的头部，如果没有别的生产者正在操作，当前生产者就可以操作了。

do{                             
  old_head = queue->head;        
  new_head->next = old_head;     
  if (old_head == queue->head){  
    queue->head = new_head;     
  }                              
}while(queue->head != new_head)

意循环终止条件：

当queue->head等于new_head时，说明本生产者已经成功操作了队列

否则，说明本轮有其他生产者操作了队列，下轮再做尝试，直到成功为止

这样看起来可以保证只有一个生产者来操作队列了（其他的生产者），现在的问题是第4行和第5行无法保证原子执行，也就是说存在多个线程的4条件都成立，紧接着又都执行了，这样还是会出现错误。

这个问题如何解决呢？4和5步如果能原子性的执行，问题就很大程度上得到了解决。幸运的是不同架构的cpu都提供了类似cas/cmpxchg的指令，保证操作的原子性

下面的c代码说明了cas的含义（这里的代码是示意性的，实际的指令是原子性的）

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;
}

有了这个指令之后，上面的代码就可以改写成多生产者安全的

do{                                              
old_head = queue->head;                        
new_head->next = old_head;                     
val=cmpxchg(&queue->head, old_head, new_head); 
}while(val!=old_head)

注意循环终止的判断条件：
当val == old_head时，说明3,4步之间没有生产者更改过队列头，操作已经成功
当val != old_head时，说明已经3,4步之间已经有其他生产者操作过队列，此时当前的生产者需要重新尝试操作队列
为何将2,3步放到循环内部呢？为了说明这个，可以假设将2,3放到循环外面会如何？假设第一轮其他生产者操作了队列，我们需要重新来过，重新来过时，queue->head已经是其他线程更新过的了，如果放到循环外面，old_head无法更新，而val则会返回新的head，此时判断条件会永远失败，导致死循环。