入队列
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，于是程序再试，直到试成功为止。这个很像我们的抢电话热线的不停重播的情况。
这里有一个潜在的问题——如果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); //置尾结点
}
DeQueue() //出队列
{
do{
p = head;
if (p->next == NULL){
return ERR_EMPTY_QUEUE;
}
while( CAS(head, p, p->next) != TRUE );
return p->next->value;
}
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问题。
/*lock_free.h*/
#ifndef _LOCK_FREE_H_
#define _LOCK_FREE_H_
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
typedef struct node_s node_t;
struct node_s {
node_t *next;
void *data;
};
node_t *create_queue();
node_t *enqueue(void * d);
void* dequeue();
#endif
/*lock_free.c*/
#include "lock_free.h"
node_t *head = NULL;
node_t *tail = NULL;
node_t *create_queue()
{
if (head != NULL)
return head;
head = (node_t*)malloc(sizeof(node_t));
if (!head) {
fprintf(stderr, "malloc error\n");
return NULL;
}
head->next = NULL;
head->data = NULL;
tail = head;
return head;
}
node_t* enqueue(void *d)
{
node_t *p = (node_t*)malloc(sizeof(node_t));
if (!p) {
fprintf(stderr, "malloc error\n");
return NULL;
}
p->next = NULL;
p->data = d;
node_t *q;
do {
q = tail;
} while (!__sync_bool_compare_and_swap(&(q->next), NULL, p));
__sync_bool_compare_and_swap(&tail, q, p);
return p;
}
void* dequeue()
{
node_t *p;
void *res;
do {
p = head;
if (p->next == NULL)
return NULL;
res = p->next->data;
} while(!__sync_bool_compare_and_swap(&head, p, p->next));
/*
在释放头节点之前保存返回的结果，
如果没有保存这个值，当有线程1都更新了head值后并且还没有返回，
此时另一个线程却删除这个值，会导致线程1返回的结果为一个无效指针
*/
if (p) {
free(p);
p = NULL;
}
return res;
}
/*main.c*/
#include "lock_free.h"
#include <pthread.h>
#include <string.h>
void *entry(void *data);
int main()
{
if (!create_queue()) {
fprintf(stderr, "create queue error\n");
exit(1);
}
int i;
pthread_t pid;
for (i = 0; i < 4; i ++) {
if (0 != pthread_create(&pid, NULL, entry, NULL)) {
perror("pthread create error\n");
}
}
getchar();
return 0;
}
void *entry(void *data)
{
char str[64] = {0};
char *dst = NULL;
sprintf(str, "%lu", pthread_self());
int i;
for (i = 0; i < 4; i ++) {
dst = (char*)malloc(128);
memset(dst, 0, 128);
sprintf(dst, "%s-%c", str, i + 65);
if (!enqueue((void *)(dst))) {
fprintf(stderr, "enqueue error\n");
break;
}
}
void *res = dequeue();
for (; res ;) {
fprintf(stdout, "%s\n", (char*)(res));
free(res);
res = NULL;
res = dequeue();
}
return NULL;
}
