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 挂载到队尾节点的next上可是还没来的及更新队尾节点但同一时刻二号线程也在进行入队操作将新的队列节点B 也挂载到了没更新的队尾节点的next上那么一号线程挂载的节点A就丢失了

为了解决多线程环境下的这类问题

我们第一时间肯定想到了加上互斥锁控制同一时刻只能有一个线程可以对队列进行写操作
但是加锁的操作太消耗系统资源了很繁重
因为对临界区的操作只有一步就是对队列的尾节点进行更新
只要让这一步进行的是原子操作就可以了
所以使用到了CAS操作也就是实现无锁队列

在博客的最后有无锁队列的源代码

Q: 操作系统级别是如何实现的

A: X86中有一个CMPXCHG的汇编指令

Q: CAS指令有什么缺点

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

 1 #ifndef QUEUE_H_
 2 #define QUEUE_H_
 3
 4 #include <stdio.h>
 5 #include <stdlib.h>
 6
 7 /*
 8 普通的
 9 链式队列
10 */
11 typedef struct QNode
12 {
13     int data;
14     struct QNode *next;
15 }QNode, *QueuePtr;
16
17 typedef struct LinkQueue
18 {
19     QueuePtr front;
20     QueuePtr rear;
21 }LinkQueue;
22
23
24 void init_Queue(LinkQueue *q);//初始化队列
25 void push_Queue(LinkQueue *q, int e);//队尾入队
26 int pop_Queue(LinkQueue *q, int *e);//队头出队
27 int is_Empty(LinkQueue *q);
28 void show(LinkQueue *q);
29
30
31 #endif /* QUEUE_H_ */

queue.h

  1 #include "queue.h"
  2
  3 /*
  4 初始化
  5 为队列构建一个头结点
  6 让front和rear都指向这个头结点
  7 */
  8 void init_Queue(LinkQueue *q)
  9 {
 10     q->front = q->rear = (QNode *)malloc(sizeof(QNode));
 11     q->front->next = NULL;
 12 }
 13
 14 /*
 15 普通的入队操作
 16 */
 17 void push_Queue(LinkQueue *q, int e)
 18 {
 19     QueuePtr newNode = (QueuePtr)malloc(sizeof(QNode));
 20     newNode->data = e;
 21     newNode->next = NULL;
 22     q->rear->next = newNode;
 23     q->rear = newNode;
 24 }
 25
 26 /*
 27 cas的入队操作
 28 和普通的入队操作一样
 29 新建节点后
 30 要将新节点挂在队尾时需要进行cas操作
 31 */
 32 void cas_push(LinkQueue *q, int e)
 33 {
 34     QueuePtr newNode = (QueuePtr)malloc(sizeof(QNode));
 35     newNode->data = e;
 36     newNode->next = NULL;
 37
 38     QueuePtr tmp;
 39     do
 40     {
 41         tmp = q->rear;
 42     }while (!__sync_bool_compare_and_swap((&(tmp->next)), NULL, (newNode));
 43
 44     q->rear = newNode;
 45 }
 46
 47 /*
 48 以前的判空是 q->front == q->rear
 49 但是这样子会增加出队的操作 当出的是最后一个元素时, q->rear需要指向 q->front
 50 我把这一步省了 暂时没有发现有什么副作用
 51 所以我改成了 q->front->next == NULL
 52 */
 53 int is_Empty(LinkQueue *q)
 54 {
 55     if (q->front->next == NULL)
 56     {
 57         return(1);
 58     }
 59     return(0);
 60 }
 61
 62 /*
 63 普通的出队操作
 64 如果队空 返回0 也就是false
 65 e作为接受元素的缓冲
 66 */
 67 int pop_Queue(LinkQueue *q, int *e)
 68 {
 69     if (is_Empty(q))
 70     {
 71         return(0);
 72     }
 73     QueuePtr tmp;
 74     tmp = q->front->next;
 75     q->front->next = tmp->next;
 76
 77     *e = tmp->data;
 78     free(tmp);
 79     return(1);
 80 }
 81
 82 /*
 83 cas的出队操作
 84 每一次都要判断这个队列是不是空
 85 然后执行cas的出队操作：
 86 （1）tmp = q->front 把旧的队头存起来
 87 （2）执行原子操作：看 旧的队头 是否等于 现在的队头 tmp == *(&(q->front->next)) 如果相等执行 *(&(q->front->next)) = tmp->next 返回true
 88     否则，即执行这一步原子操作的时候，别的线程修改了队列，导致队尾指向改变了，返回false ,while(!false)回到第一步重新执行
 89 */
 90 int cas_pop(LinkQueue *q, int *e)
 91 {
 92     QueuePtr tmp;
 93     do {
 94         if (is_Empty(q))
 95         {
 96             return(0);
 97         }
 98         tmp = q->front;
 99     } while (!__sync_bool_compare_and_swap(&(q->front->next), tmp, tmp->next->next);
100
101     *e = tmp->data;
102     free(tmp);
103     return(1);
104 }
105
106 /*
107 遍历队列 打印里面的元素 为了求证队列里面的元素
108 */
109 void show(LinkQueue *q)
110 {
111     printf("void show(LinkQueue *q)\n");
112     QueuePtr tmp = q->front->next;
113     while (tmp)
114     {
115         printf("%d ", tmp->data);
116         tmp = tmp->next;
117     }
118     printf("\n");
119 }

queue.c

#include "queue.h"
#include <pthread.h>
#include <unistd.h>
#include <assert.h>#define THREAD_NUMBER 4//开启的线程数，电脑是4核，所以用4void *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 = 0; i < THREAD_NUMBER; ++i){pthread_create(&threadArr[i], NULL, thread_push, (void *)&que);}/*等待四个线程都执行完要不然主线程一下子就跑完了 程序就结束了还有就是 为了show函数 可以验证元素是不是都push进去了*/for (i = 0; i < THREAD_NUMBER; ++i){pthread_join(threadArr[i], NULL);}show(&que);/*创造四个新线程 每个线程都执行 thread_pop(&que)*/for (i = 0; i < THREAD_NUMBER; ++i){pthread_create(&threadArr[i], NULL, thread_pop, (void *)&que);}for (i = 0; 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 = 0; i < 20; ++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 (1){res = cas_pop(quePtr, &tmp);if (!res){break;}printf("%d ", tmp);}printf("finish pop\n");pthread_exit(NULL);
}

lock_free_queue.c

 1 #include "queue.h"
 2 #include <pthread.h>
 3 #include <unistd.h>
 4 #include <semaphore.h>
 5 #include <assert.h>
 6
 7 #define THREAD_NUMBER 4
 8
 9 sem_t queue_sem;//同步锁
10
11 void *thread_push(void *arg);
12 void *thread_pop(void *arg);
13
14 int main()
15 {
16     LinkQueue que;
17     init_Queue(&que);
18
19     /*初始化二进制信号量 初始值为1 代表每一次只有1个线程可以访问
20     本来更加应该用互斥量 比较贴合情景 但是不太熟 就用了信号量
21     */
22     int res = sem_init(&queue_sem, 0, 1);
23     assert(res != -1);
24
25     int i;
26     pthread_t threadArr[THREAD_NUMBER];
27     for (i = 0; i < THREAD_NUMBER; ++i)
28     {
29         pthread_create(&threadArr[i], NULL, thread_push, (void *)&que);
30     }
31
32     for (i = 0; i < THREAD_NUMBER; ++i)
33     {
34         pthread_join(threadArr[i], NULL);
35     }
36
37     show(&que);
38
39     for (i = 0; i < THREAD_NUMBER; ++i)
40     {
41         pthread_create(&threadArr[i], NULL, thread_pop, (void *)&que);
42     }
43
44     for (i = 0; i < THREAD_NUMBER; ++i)
45     {
46         pthread_join(threadArr[i], NULL);
47     }
48
49     sem_destroy(&queue_sem);
50
51     exit(EXIT_SUCCESS);
52 }
53
54 void *thread_push(void *arg)
55 {
56     printf("start push\n");
57     LinkQueue * quePtr = (LinkQueue *)arg;
58     int i;
59     for (i = 0; i < 20; ++i)
60     {
61         sem_wait(&queue_sem);
62         push_Queue(quePtr, i);
63         sem_post(&queue_sem);
64     }
65     printf("finish push\n");
66     pthread_exit(NULL);
67 }
68
69 void *thread_pop(void *arg)
70 {
71     printf("start pop\n");
72     LinkQueue * quePtr = (LinkQueue *)arg;
73     int tmp;
74     int res;
75     while (1)
76     {
77         sem_wait(&queue_sem);
78         res = pop_Queue(quePtr, &tmp);
79         sem_post(&queue_sem);
80         if (!res)
81         {
82             break;
83         }
84         printf("%d ", tmp);
85     }
86     printf("finish pop\n");
87     pthread_exit(NULL);
88 }

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

转载于:https://www.cnblogs.com/hanhuihanhui/p/8017232.html

CAS简介和无锁队列的实现相关推荐

CAS操作与无锁队列
在多线程编程中,为了保证内存的可见性,我们加入了一些锁的机制,例如信号量,互斥锁,条件变量等等,但是锁的机制不是一个简单的机制,需要加入很多的控制,所以在使用中又有了一些轻量级的同步机制,例如vola ...
.net 延时操作_锁、CAS操作和无锁队列的实现
(给算法爱好者加星标,修炼编程内功) 来源:yishizuofei blog.csdn.net/yishizuofei/article/details/78353722 锁的机制锁和人很像,有的人乐 ...
锁、CAS操作和无锁队列的实现
锁的机制锁和人很像,有的人乐观,总会想到好的一方面,所以只要越努力,就会越幸运:有的人悲观,总会想到不好的一方面,患得患失,所以经常会做不好事.我一直把前一个当作为我前进的动力和方向,快乐充实的过好 ...
CAS无锁队列的实现
文章目录 1. 基本原理 2. 代码实现 2.1 使用链表实现无锁队列 2.2 使用数组实现环形无锁队列 3. ABA 问题及解决 4. 参考资料 1. 基本原理源于1994年10月发表在国际并行与 ...
基于数组的无锁队列(译)
2019独角兽企业重金招聘Python工程师标准>>> 1 引言最近对于注重性能的应用程序,我们有了一种能显著提高程序性能的选择:多线程.线程的概念实际上已经存在了很长时间.在过去 ...
无锁CAS及无锁队列实现
CAS ⽐较并交换(compare and swap, CAS),是原⼦操作的⼀种,可⽤于在多线程编程中实现不被打断的数据交换操作,从⽽避免多线程同时改写某⼀数据时由于执⾏顺序不确定性以及中断的不可 ...
无锁CAS/无锁队列
高并发,读写十分频繁,会使用CAS 1 互斥锁自旋锁原子操作锁住的代码耗时短:counter++操作,自旋锁有优势锁住的代码耗时长:for_add操作,自旋锁无优势[因为在等待的时候消耗的CP ...
深入理解高并发技术dpdk无锁队列
前两周给大家直播分享,并发技术全景(从硬件,操作系统,虚拟机/标准库,编程语言等) 上半场(5个小时):并发/并行技术全景指南下半场(5个小时):人生的下半场,你准备好了吗最后我上周还布置了一个作 ...
(Erlang语言)运行时中的无锁队列及其在异步线程中的应用
本文首先介绍 Erlang 运行时中需要使用无锁队列的场合,然后介绍无锁队列的基本原理及会遇到的问题,接下来介绍 Erlang 运行时中如何通过"线程进度"机制解决无锁队列的问题, ...
Erlang运行时中的无锁队列及其在异步线程中的应用
本文首先介绍 Erlang 运行时中需要使用无锁队列的场合,然后介绍无锁队列的基本原理及会遇到的问题,接下来介绍 Erlang 运行时中如何通过"线程进度"机制解决无锁队列的问题, ...

CAS简介和无锁队列的实现