Q:CAS的实现
A:gcc提供了两个函数
bool __sync_bool_compare_and_swap (type *ptr, type oldval, type newval, ...) //type 的类型有限制 只能是 1,2,4,8字节的整形 或者是指针类型
type __sync_val_compare_and_swap (type *ptr, type oldval, type newval, ...)
这两个函数提供原子的比较和交换，如果*ptr == oldval,就将newval写入*ptr,
第一个函数在相等并写入的情况下返回true，这个函数比第二个好在，返回bool值可以知道有没有更新成功.
第二个函数在返回操作之前的值。
第二个函数可以用c语言描述成：
type __sync_val_compare_and_swap (type *ptr, type oldval, type newval, ...)
{
type cur = *ptr;
if (cur == oldval)
{
*ptr = newval;
}
return cur;// 返回操作之前的值
}
官方文档里有一段：GCC will allow any integral scalar or pointer type that is 1, 2, 4 or 8 bytes in length.
意思是GCC允许任何长度为1, 2, 4或8字节的整型和指针
不然就会有下面这种错误
Q:简单介绍一下无锁队列
A:
之所以有无锁队列，是因为在多线程环境下有以下情景,
对同一链队列进行入队操作时 一号线程正在将 新的队列节点A 挂载到 队尾节点的next上 可是还没来的及更新队尾节点 但同一时刻二号线程也在进行入队操作 将 新的队列节点B 也挂载到了 没更新的队尾节点的next上 那么一号线程挂载的节点A就丢失了
为了解决多线程环境下的这类问题
我们第一时间肯定想到了加上互斥锁 控制同一时刻只能有一个线程可以对队列进行写操作
但是加锁的操作太消耗系统资源了 很繁重
因为对临界区的操作只有一步 就是对队列的尾节点进行更新
只要让这一步进行的是原子操作就可以了
所以使用到了CAS操作 也就是实现无锁队列
在博客的最后有无锁队列的源代码
Q: 操作系统级别是如何实现的
A: X86中有一个CMPXCHG的汇编指令
Q: CAS指令有什么缺点
A:
1.存在ABA问题因为CAS需要在操作值的时候检查下值有没有发生变化，如果没有发生变化则更新，但是如果一个值原来是A，变成了B，又变成了A，那么使用CAS进行检查时会发现它的值没有发生变化，但是实际上却变化了。ABA问题的解决思路就是使用版本号。在变量前面追加上版本号，每次变量更新的时候把版本号加一，那么A－B－A 就会变成1A-2B－3A。
2.循环时间长开销大自旋CAS如果长时间不成功，会给CPU带来非常大的执行开销。
3. 只能保证一个共享变量的原子操作对一个共享变量执行操作时，我们可以使用循环CAS的方式来保证原子操作，但是对多个共享变量操作时，循环CAS就无法保证操作的原子性，这个时候就可以用锁，或者有一个取巧的办法，就是把多个共享变量合并成一个共享变量来操作。比如有两个共享变量i＝2,j=a，合并一下ij=2a，然后用CAS来操作ij。
与CAS相关的一些原子操作
gcc从4.1.2提供了__sync_*系列的built-in函数，用于提供加减和逻辑运算的原子操作。
其声明如下：
原子操作的后置加加type __sync_fetch_and_add (type *ptr, type
value, ...)
原子操作的前置加加type __sync_add_and_fetch (type
*ptr, type value, ...)
其他类比
type __sync_fetch_and_sub (type *ptr, type
value, ...)
type __sync_fetch_and_or (type *ptr, type value, ...)
type __sync_fetch_and_and (type *ptr, type value, ...)
type __sync_fetch_and_xor (type *ptr, type value, ...)
type __sync_fetch_and_nand (type *ptr, type value, ...)
type __sync_sub_and_fetch (type *ptr, type value, ...)
type __sync_or_and_fetch (type *ptr, type value, ...)
type __sync_and_and_fetch (type *ptr, type value, ...)
type __sync_xor_and_fetch (type *ptr, type value, ...)
type __sync_nand_and_fetch (type *ptr, type value, ...)
这两组函数的区别在于第一组返回更新前的值，第二组返回更新后的值。
无锁队列源代码
为了有一个对比
写了一份thread_queue.c是用锁对临界区进行控制访问的,也就是有锁队列
另一份是lock_free_queue.c是用CAS确保对临界区的操作是原子操作,也就是无锁队列
依赖关系
lock_free_queue.c -> queue.h -> queue.c
thread_queue.c -> queue.h -> queue.c
gcc lock_free_queue.c queue.c -lpthread -o lock_free_queue
gcc thread_queue.c queue.c -lpthread -o thread_queue
#ifndef QUEUE_H_
#define QUEUE_H_ #include <stdio.h>
#include <stdlib.h> /*
普通的
链式队列
*/
typedef struct QNode
{
int data;
struct QNode *next;
}QNode, *QueuePtr; typedef struct LinkQueue
{
QueuePtr front;
QueuePtr rear;
}LinkQueue; void init_Queue(LinkQueue *q);//初始化队列
void push_Queue(LinkQueue *q, int e);//队尾入队
int pop_Queue(LinkQueue *q, int *e);//队头出队
int is_Empty(LinkQueue *q);
void show(LinkQueue *q); #endif /* QUEUE_H_ */
queue.h
#include "queue.h" /*
初始化
为队列构建一个头结点
让front和rear都指向这个头结点
*/
void init_Queue(LinkQueue *q)
{
q->front = q->rear = (QNode *)malloc(sizeof(QNode));
q->front->next = NULL;
} /*
普通的入队操作
*/
void push_Queue(LinkQueue *q, int e)
{
QueuePtr newNode = (QueuePtr)malloc(sizeof(QNode));
newNode->data = e;
newNode->next = NULL;
q->rear->next = newNode;
q->rear = newNode;
} /*
cas的入队操作
和普通的入队操作一样
新建节点后
要将新节点挂在队尾时需要进行cas操作
*/
void cas_push(LinkQueue *q, int e)
{
QueuePtr newNode = (QueuePtr)malloc(sizeof(QNode));
newNode->data = e;
newNode->next = NULL; QueuePtr tmp;
do
{
tmp = q->rear;
}while (!__sync_bool_compare_and_swap((&(tmp->next)), NULL, (newNode)); q->rear = newNode;
} /*
以前的判空是 q->front == q->rear
但是这样子会增加出队的操作 当出的是最后一个元素时, q->rear需要指向 q->front
我把这一步省了 暂时没有发现有什么副作用
所以我改成了 q->front->next == NULL
*/
int is_Empty(LinkQueue *q)
{
if (q->front->next == NULL)
{
return();
}
return();
} /*
普通的出队操作
如果队空 返回0 也就是false
e作为接受元素的缓冲
*/
int pop_Queue(LinkQueue *q, int *e)
{
if (is_Empty(q))
{
return();
}
QueuePtr tmp;
tmp = q->front->next;
q->front->next = tmp->next; *e = tmp->data;
free(tmp);
return();
} /*
cas的出队操作
每一次都要判断这个队列是不是空
然后执行cas的出队操作：
（1）tmp = q->front 把旧的队头存起来
（2）执行原子操作：看 旧的队头 是否等于 现在的队头 tmp == *(&(q->front->next)) 如果相等执行 *(&(q->front->next)) = tmp->next 返回true
否则，即执行这一步原子操作的时候，别的线程修改了队列，导致队尾指向改变了，返回false ,while(!false)回到第一步重新执行
*/
int cas_pop(LinkQueue *q, int *e)
{
QueuePtr tmp;
do {
if (is_Empty(q))
{
return();
}
tmp = q->front;
} while (!__sync_bool_compare_and_swap(&(q->front->next), tmp, tmp->next->next); *e = tmp->data;
free(tmp);
return();
} /*
遍历队列 打印里面的元素 为了求证队列里面的元素
*/
void show(LinkQueue *q)
{
printf("void show(LinkQueue *q)\n");
QueuePtr tmp = q->front->next;
while (tmp)
{
printf("%d ", tmp->data);
tmp = tmp->next;
}
printf("\n");
}
queue.c
#include "queue.h"
#include <pthread.h>
#include <unistd.h>
#include <assert.h> #define THREAD_NUMBER 4//开启的线程数，电脑是4核，所以用4 void *thread_push(void *arg);
void *thread_pop(void *arg); /*
初始化空队列 为了模拟线程对资源的抢占
开启4个线程 每个线程push 20个元素 0~19
等待4个线程结束
打印队列元素 验证push
开启四个线程 每个线程都对队列进行 pop操作
*/
int main()
{
LinkQueue que;
init_Queue(&que); int i;
/*
创造四个新线程 每个线程都执行 thread_push(&que)
*/
pthread_t threadArr[THREAD_NUMBER];
for (i = ; i < THREAD_NUMBER; ++i)
{
pthread_create(&threadArr[i], NULL, thread_push, (void *)&que);
} /*
等待四个线程都执行完
要不然主线程一下子就跑完了 程序就结束了
还有就是 为了show函数 可以验证元素是不是都push进去了
*/
for (i = ; i < THREAD_NUMBER; ++i)
{
pthread_join(threadArr[i], NULL);
} show(&que); /*
创造四个新线程 每个线程都执行 thread_pop(&que)
*/
for (i = ; i < THREAD_NUMBER; ++i)
{
pthread_create(&threadArr[i], NULL, thread_pop, (void *)&que);
} for (i = ; i < THREAD_NUMBER; ++i)
{
pthread_join(threadArr[i], NULL);
} exit(EXIT_SUCCESS);
} void *thread_push(void *arg)
{
printf("start push\n");
LinkQueue * quePtr = (LinkQueue *)arg;
int i;
for (i = ; i < ; ++i)
{
cas_push(quePtr, i);
}
printf("finish push\n");
pthread_exit(NULL);
} void *thread_pop(void *arg)
{
printf("start pop\n");
LinkQueue * quePtr = (LinkQueue *)arg;
int tmp;
int res;
while ()
{
res = cas_pop(quePtr, &tmp);
if (!res)
{
break;
}
printf("%d ", tmp);
}
printf("finish pop\n");
pthread_exit(NULL);
}
lock_free_queue.c
#include "queue.h"
#include <pthread.h>
#include <unistd.h>
#include <semaphore.h>
#include <assert.h> #define THREAD_NUMBER 4 sem_t queue_sem;//同步锁 void *thread_push(void *arg);
void *thread_pop(void *arg); int main()
{
LinkQueue que;
init_Queue(&que); /*初始化二进制信号量 初始值为1 代表每一次只有1个线程可以访问
本来更加应该用互斥量 比较贴合情景 但是不太熟 就用了信号量
*/
int res = sem_init(&queue_sem, , );
assert(res != -); int i;
pthread_t threadArr[THREAD_NUMBER];
for (i = ; i < THREAD_NUMBER; ++i)
{
pthread_create(&threadArr[i], NULL, thread_push, (void *)&que);
} for (i = ; i < THREAD_NUMBER; ++i)
{
pthread_join(threadArr[i], NULL);
} show(&que); for (i = ; i < THREAD_NUMBER; ++i)
{
pthread_create(&threadArr[i], NULL, thread_pop, (void *)&que);
} for (i = ; i < THREAD_NUMBER; ++i)
{
pthread_join(threadArr[i], NULL);
} sem_destroy(&queue_sem); exit(EXIT_SUCCESS);
} void *thread_push(void *arg)
{
printf("start push\n");
LinkQueue * quePtr = (LinkQueue *)arg;
int i;
for (i = ; i < ; ++i)
{
sem_wait(&queue_sem);
push_Queue(quePtr, i);
sem_post(&queue_sem);
}
printf("finish push\n");
pthread_exit(NULL);
} void *thread_pop(void *arg)
{
printf("start pop\n");
LinkQueue * quePtr = (LinkQueue *)arg;
int tmp;
int res;
while ()
{
sem_wait(&queue_sem);
res = pop_Queue(quePtr, &tmp);
sem_post(&queue_sem);
if (!res)
{
break;
}
printf("%d ", tmp);
}
printf("finish pop\n");
pthread_exit(NULL);
}
thread_queue.c
博客，参考：http://blog.csdn.net/syzcch/article/details/8075830
libcds库，参考：http://blog.jobbole.com/90810/
CAS缺点，参考：https://www.cnblogs.com/zhuawang/p/4196904.html
CAS函数，参考: http://blog.csdn.net/youfuchen/article/details/23179799
gcc Atomic Builtins官方：https://gcc.gnu.org/onlinedocs/gcc-4.1.1/gcc/Atomic-Builtins.html