关于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;
}
我们可以看到，old_reg_val 总是返回，于是，我们可以在 compare_and_swap 操作之后对其进行测试，以查看它是否与 oldval相匹配，因为它可能有所不同，这意味着另一个并发线程已成功地竞争到 compare_and_swap 并成功将 reg 值从 oldval 更改为别的值了。
这个操作可以变种为返回bool值的形式（返回 bool值的好处在于，可以调用者知道有没有更新成功）：
bool compare_and_swap (int *addr, int oldval, int newval)
{
if ( *addr != oldval ) {
return false;
}
*addr = newval;
return true;
}
与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* 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》
- 美国纽约罗切斯特大学 maged m. michael 和 michael l. scott 在1996年3月发表的一篇论文 《simple, fast, and practical non-blocking and blocking concurrentqueue algorithms》
- （注：下面的代码并不完全与这篇论文相同）
初始化一个队列的代码很简，初始化一个dummy结点（注：在链表操作中，使用一个dummy结点，可以少掉很多边界条件的判断），如下所示：
initqueue(q)
{
node = new node()
node->next = null;
q->head = q->tail = node;
}
我们先来看一下进队列用cas实现的方式，基本上来说就是链表的两步操作：
第一步，把tail指针的next指向要加入的结点。 tail->next = p;
第二步，把tail指针移到队尾。 tail = p;
enqueue(q, data) //进队列
{
//准备新加入的结点数据
n = new node();
n->value = data;
n->next = null;
do {
p = q->tail; //取链表尾指针的快照
} while( cas(p->next, null, n) != true);
//while条件注释：如果没有把结点链在尾指针上，再试
cas(q->tail, p, n); //置尾结点 tail = n;
}
我们可以看到，程序中的那个 do-while 的 retry-loop 中的 cas 操作：如果 p->next 是 null，那么，把新结点 n 加到队尾。如果不成功，则重新再来一次！
就是说，很有可能我在准备在队列尾加入结点时，别的线程已经加成功了，于是tail指针就变了，于是我的cas返回了false，于是程序再试，直到试成功为止。这个很像我们的抢电话热线的不停重播的情况。
但是你会看到，为什么我们的“置尾结点”的操作（第13行）不判断是否成功，因为：
- 如果有一个线程t1，它的while中的cas如果成功的话，那么其它所有的 随后线程的cas都会失败，然后就会再循环，
- 此时，如果t1 线程还没有更新tail指针，其它的线程继续失败，因为tail->next不是null了。
- 直到t1线程更新完 tail 指针，于是其它的线程中的某个线程就可以得到新的 tail 指针，继续往下走了。
- 所以，只要线程能从 while 循环中退出来，意味着，它已经“独占”了，tail 指针必然可以被更新。
这里有一个潜在的问题——如果t1线程在用cas更新tail指针的之前，线程停掉或是挂掉了，那么其它线程就进入死循环了。下面是改良版的enqueue()
enqueue(q, data) //进队列改良版 v1
{
n = new node();
n->value = data;
n->next = null;
p = q->tail;
oldp = p
do {
while (p->next != null)
p = p->next;
} while( cas(p.next, null, n) != true); //如果没有把结点链在尾上，再试
cas(q->tail, oldp, n); //置尾结点
}
我们让每个线程，自己fetch 指针 p 到链表尾。但是这样的fetch会很影响性能。而且，如果一个线程不断的enqueue，会导致所有的其它线程都去 fetch 他们的 p 指针到队尾，能不能不要所有的线程都干同一个事？这样可以节省整体的时间？
比如：直接 fetch q->tail 到队尾？因为，所有的线程都共享着 q->tail，所以，一旦有人动了它后，相当于其它的线程也跟着动了，于是，我们的代码可以改进成如下的实现：
enqueue(q, data) //进队列改良版 v2
{
n = new node();
n->value = data;
n->next = null;
while(true) {
//先取一下尾指针和尾指针的next
tail = q->tail;
next = tail->next;
//如果尾指针已经被移动了，则重新开始
if ( tail != q->tail ) continue;
//如果尾指针的 next 不为null，则 fetch 全局尾指针到next
if ( next != null ) {
cas(q->tail, tail, next);
continue;
}
//如果加入结点成功，则退出
if ( cas(tail->next, next, n) == true ) break;
}
cas(q->tail, tail, n); //置尾结点
}
上述的代码还是很清楚的，相信你一定能看懂，而且，这也是 java 中的 concurrentlinkedqueue 的实现逻辑，当然，我上面的这个版本比 java 的好一点，因为没有 if 嵌套，嘿嘿。
好了，我们解决了enqueue，我们再来看看dequeue的代码：（很简单，我就不解释了）
dequeue(q) //出队列
{
do{
p = q->head;
if (p->next == null){
return err_empty_queue;
}
while( cas(q->head, p, p->next) != true );
return p->next->value;
}
我们可以看到，dequeue的代码操作的是 head->next，而不是 head 本身。这样考虑是因为一个边界条件，我们需要一个dummy的头指针来解决链表中如果只有一个元素，head 和 tail 都指向同一个结点的问题，这样 enqueue 和 dequeue 要互相排斥了。
但是，如果 head 和 tail 都指向同一个结点，这意味着队列为空，应该返回 err_empty_queue，但是，在判断 p->next == null 时，另外一个enqueue操作做了一半，此时的 p->next 不为 null了，但是 tail 指针还差最后一步，没有更新到新加的结点，这个时候就会出现，在 enqueue 并没有完成的时候， dequeue 已经把新增加的结点给取走了，此时，队列为空，但是，head 与 tail 并没有指向同一个结点。如下所示：
虽然，enqueue的函数会把 tail 指针置对，但是，这种情况可能还是会导致一些并发问题，所以，严谨来说，我们需要避免这种情况。于是，我们需要加入更多的判断条件，还确保这个问题。下面是相关的改进代码：
dequeue(q) //出队列，改进版
{
while(true) {
//取出头指针，尾指针，和第一个元素的指针
head = q->head;
tail = q->tail;
next = head->next;
// q->head 指针已移动，重新取 head指针
if ( head != q->head ) continue;
// 如果是空队列
if ( head == tail && next == null ) {
return err_empty_queue;
}
//如果 tail 指针落后了
if ( head == tail && next == null ) {
cas(q->tail, tail, next);
continue;
}
//移动 head 指针成功后，取出数据
if ( cas( q->head, head, next) == true){
value = next->value;
break;
}
}
free(head); //释放老的dummy结点
return value;
}
上面这段代码的逻辑和 java 的 concurrentlinkedqueue 的 poll 方法很一致了。也是《simple, fast, and practical non-blocking and blocking concurrentqueue algorithms》这篇论文中的实现。
cas的aba问题
所谓aba（见维基百科的aba词条），问题基本是这个样子：
- 进程p1在共享变量中读到值为a
- p1被抢占了，进程p2执行
- p2把共享变量里的值从a改成了b，再改回到a，此时被p1抢占。
- 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！（论文《simple, fast, and practical non-blocking and blocking concurrentqueue algorithms》中的实现方法也基本上是一样的，用到的是增加一个计数，可以理解为版本号）
）
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模板。
ibm developerworks的《设计不使用互斥锁的并发数据结构》