锁、CAS操作和无锁队列的实现

 https://blog.csdn.net/yishizuofei/article/details/78353722

锁的机制

锁和人很像，有的人乐观，总会想到好的一方面，所以只要越努力，就会越幸运；有的人悲观，总会想到不好的一方面，患得患失，所以经常会做不好事。我一直把前一个当作为我前进的动力和方向，快乐充实的过好每一天。

常用的锁机制也有两种： 
1、乐观锁：假设不会发生并发冲突，每次不加锁而去完成某项操作，只在提交操作时，检查是否违反数据完整性。如果因为冲突失败就继续重试，直到成功为止。而乐观锁用到的机制就是CAS。

乐观锁大多是基于数据版本记录机制实现。为数据增加一个版本标识，比如在基于数据库表的版本解决方案中，一般是通过微数据库表增加一个“version”字段来实现。读取数据时，将此版本号一同读出，之后更新时，对此版本号加一。此时，将提交数据的版本数据与数据库表对应记录的当前版本信息进行比对，如果提交的数据版本号大于数据库表当前版本号，则予以更新，否则认为是过期数据。*乐观锁的缺点是不能解决脏读的问题*。

注意：在实际生产环境里边，如果并发量不大且不允许脏读，可以使用悲观锁解决并发问题。但如果系统的并发非常大的话，悲观锁定会带来非常大的性能问题，所以我们就要选择乐观锁。

2、悲观锁：假定会发生并发冲突，屏蔽一切可能违反数据完整性的操作。悲观锁的实现，往往依靠底层提供的锁机制。悲观锁会导致其他所有需要锁的线程挂起，等待持有锁的线程释放锁。如果所有线程都在等待其他线程释放锁，而不能主动释放锁资源，那么也会造成死锁问题。

锁的机制存在以下问题： 
（1）在多线程竞争下，加锁、释放锁会导致比较多的上下文切换和调度延时，引起性能问题。 
（2）一个线程持有锁会导致其他所有需要次所的线程挂起。 
（3）如果一个优先级搞得线程等待一个优先级低的线程释放锁会导致优先级倒置，引起性能风险。

CAS操作

Compare And Set（或Compare And Swap），CAS是解决多线程并行情况下使用锁造成性能损耗的一种机制，CAS操作包含三个操作数——内存位置（V）、预期原值（A）、新值（B）。如果内存位置的值与预期原值相同，那么处理器会自动将内存的值更新为新值。否则，处理器不做任何操作。无论哪种情况，处理器都会在CAS指令之前返回该位置的值。CAS有效地说明了“我认为位置V应该包含值A；如果包含该值，则将B放到这个位置；否则，不要更新该位置，只告诉我这个位置现在的值即可。” 
现在几乎所有的CPU指令都支持CAS的原子操作，X86下对应的是CMPXCHG汇编指令。有了这个操作，我们就可以用其来实现各种无锁的数据结构。

这个操作可以用以下的例子来描述： 
意思是，看一看内存*reg里的值是不是oldval，如果是的话，则对其赋值newval，并返回true，表示更新成功，如果返回false，则表示修改失败。

bool compare_and_swap(int *reg,int oldval,int newval)
{
    int reg_val = *reg;
    if(reg_val == oldval)
    {
        *reg = newval;
        return true;
    }
    return false;
}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10

CAS操作无锁队列的实现（参考）

Q:CAS的实现 
A:gcc提供了两个函数

bool __sync_bool_compare_and_swap (type *ptr, type oldval, 
                                    type newval, ...)；
type __sync_val_compare_and_swap (type *ptr, type oldval, 
                            type newval, ...)；

• 1
• 2
• 3
• 4

这两个函数提供原子的比较和交换，如果*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;// 返回操作之前的值
}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10

type只能是1,2,4或8字节长度的int类型，否则会发生下面的错误 
 
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。

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, …) 
这两组函数的区别在于第一组返回更新前的值，第二组返回更新后的值。

关于CAS函数，参考: http://blog.csdn.net/youfuchen/article/details/23179799

在多线程环境下有以下情景：

比如对同一链队列进行入队操作时一个线程正在将新的队列节点 挂载到 队尾节点的next上，可是还没来的及更新队尾节点 但同一时刻另一个线程也在进行入队操作将新的队列节点也挂在到了没更新的队尾节点那么先挂载的节点就丢失了。

为了解决多线程环境下的这些问题，我们第一时间肯定想到了加上互斥锁控制同一时刻只能有一个线程可以对队列进行写操作，但是加锁的操作太消耗系统资源了很繁重 。

因为对临界区的操作只有一步 就是对队列的尾节点进行更新，只要让这一步进行的是原子操作就可以了，所以使用到了CAS操作。

为了有一个对比 写了一份thread_queue.c是用锁对临界区进行控制访问的 
另一份是lock_free_queue.c是用CAS确保对临界区的操作是原子操作

“queue.h”

#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.c”

#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操作
因为官方文档：The definition given in the Intel documentation allows only for the use of the types int, long, long long as well as their unsigned counterparts
只能用 int, long, long long
所以要把指针类型 QueuePtr 变成 long
用long的另一个原因就是：屏蔽32位和64位的差异 long在32位是4字节 64位是8字节
*/
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((long *)(&(tmp->next)), NULL, (long)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(1);
    }
    return(0);
}

/*
普通的出队操作
如果队空 返回0 也就是false
e作为接受元素的缓冲
*/
int pop_Queue(LinkQueue *q, int *e)
{
    if (is_Empty(q))
    {
        return(0);
    }
    QueuePtr tmp;
    tmp = q->front->next;
    q->front->next = tmp->next;

    *e = tmp->data;
    free(tmp);
    return(1);
}

/*
cas的出队操作
每一次都要判断这个队列是不是空
然后执行cas的出队操作：
（1）tmp = q->rear 把旧的队头存起来
（2）执行原子操作：看 旧的队头 是否等于 现在的队头 tmp == *(&(q->front)) 如果相等执行 *(&(q->front)) = tmp->next 返回true 
    否则，即执行这一步原子操作的时候，别的线程修改了队列，导致队尾指向改变了，返回false ,while(!false)回到第一步重新执行
*/
int cas_pop(LinkQueue *q, int *e)
{
    QueuePtr tmp;
    do {
        if (is_Empty(q))
        {
            return(0);
        }
        //printf("cas_pop...
");
        tmp = q->front->next;
    } while (!__sync_bool_compare_and_swap((long *)(&(q->front->next)), (long)tmp, (long)tmp->next));

    *e = tmp->data;
    free(tmp);
    return(1);
}

/*
遍历队列 打印里面的元素 为了求证队列里面的元素
*/
void show(LinkQueue *q)
{
    printf("void show(LinkQueue *q)
");
    QueuePtr tmp = q->front->next;
    while (tmp)
    {
        printf("%d ", tmp->data);
        tmp = tmp->next;
    }
    printf("
");
}

“thread_queue.c”

#include "queue.h"
#include <pthread.h>
#include <unistd.h>
#include <semaphore.h>
#include <assert.h>

#define THREAD_NUMBER 4//开启的线程数，电脑是4核，所以用4

//sem_t queue_sem;//信号量
pthread_mutex_t mutex;//互斥锁

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, 0, 1);
    //assert(res != -1);

    int i;
    pthread_t threadArr[THREAD_NUMBER];
    for (i = 0; i < THREAD_NUMBER; ++i)
    {
        pthread_create(&threadArr[i], NULL, thread_push, (void *)&que);
    }

    for (i = 0; i < THREAD_NUMBER; ++i)
    {
        pthread_join(threadArr[i], NULL);
    }

    show(&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);
    }

    //sem_destroy(&queue_sem);

    exit(EXIT_SUCCESS);
}

void *thread_push(void *arg)
{
    printf("start push
");
    LinkQueue * quePtr = (LinkQueue *)arg;
    int i;
    for (i = 0; i < 20; ++i)
    {
        //sem_wait(&queue_sem);
        pthread_mutex_lock(&mutex);
        push_Queue(quePtr, i);
        pthread_mutex_unlock(&mutex);
        //sem_post(&queue_sem);
    }
    printf("finish push
");
    pthread_exit(NULL);
}

void *thread_pop(void *arg)
{
    printf("start pop
");
    LinkQueue * quePtr = (LinkQueue *)arg;
    int tmp;
    int res;
    while (1)
    {
        //sem_wait(&queue_sem);
        pthread_mutex_lock(&mutex);
        res = pop_Queue(quePtr, &tmp);
        pthread_mutex_unlock(&mutex);
        //sem_post(&queue_sem);
        if (!res)
        {
            break;
        }
        printf("%d ", tmp);
    }
    printf("finish pop
");
    pthread_exit(NULL);
}

“lock_free_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 = 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
");
    LinkQueue * quePtr = (LinkQueue *)arg;
    int i;
    for (i = 0; i < 20; ++i)
    {
        cas_push(quePtr, i);
    }
    printf("finish push
");
    pthread_exit(NULL);
}

void *thread_pop(void *arg)
{
    printf("start pop
");
    LinkQueue * quePtr = (LinkQueue *)arg;
    int tmp;
    int res;
    while (1)
    {
        res = cas_pop(quePtr, &tmp);
        if (!res)
        {
            break;
        }
        printf("%d ", tmp);
        //sleep(1);
    }
    printf("finish pop
");
    pthread_exit(NULL);
}