锁、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...\n");
        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)\n");
    QueuePtr tmp = q->front->next;
    while (tmp)
    {
        printf("%d ", tmp->data);
        tmp = tmp->next;
    }
    printf("\n");
}

“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\n");
    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\n");
    pthread_exit(NULL);
}

void *thread_pop(void *arg)
{
    printf("start pop\n");
    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\n");
    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\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);
        //sleep(1);
    }
    printf("finish pop\n");
    pthread_exit(NULL);
}

原文地址：https://www.cnblogs.com/kelelipeng/p/10671795.html

时间： 2024-10-07 01:10:05

锁、CAS操作和无锁队列的实现的相关文章

具体CAS操作实现(无锁算法)

具体CAS操作 上一篇讲述了CAS机制,这篇讲解CAS具体操作. 什么是悲观锁.乐观锁?在java语言里,总有一些名词看语义跟本不明白是啥玩意儿,也就总有部分面试官拿着这样的词来忽悠面试者,以此来找优越感,其实理解清楚了,这些词也就唬不住人了. synchronized是悲观锁,这种线程一旦得到锁,其他需要锁的线程就挂起的情况就是悲观锁. CAS操作的就是乐观锁,每次不加锁而是假设没有冲突而去完成某项操作,如果因为冲突失败就重试,直到成功为止. 那么问题来了,什么是CAS操作? CAS是Comp

SpinLock 自旋锁, CAS操作（Compare &amp; Set） ABA Problem

SpinLock 自旋锁 spinlock 用于CPU同步, 它的实现是基于CPU锁定数据总线的指令. 当某个CPU锁住数据总线后, 它读一个内存单元(spinlock_t)来判断这个spinlock 是否已经被别的CPU锁住. 如果否, 它写进一个特定值, 表示锁定成功, 然后返回. 如果是, 它会重复以上操作直到成功, 或者spin次数超过一个设定值. 锁定数据总线的指令只能保证一个机器指令内, CPU独占数据总线. 单CPU当然能用spinlock, 但实现上无需锁定数据总线. spinl

CAS原子操作实现无锁及性能分析

Author:Echo Chen(陈斌) Email:[email protected] Blog:Blog.csdn.net/chen19870707 Date:Nov 13th, 2014 最近在研究nginx的自旋锁的时候,又见到了GCC CAS原子操作,于是决定动手分析下CAS实现的无锁到底性能如何,网上关于CAS实现无锁的文章很多,但少有研究这种无锁的性能提升的文章,这里就以实验结果和我自己的理解逐步展开. 1.什么是CAS原子操作 在研究无锁之前,我们需要首先了解一下CAS原子操作-

Compare And Swap（CAS）实现无锁多生产者

1.CAS 原理 compare and swap,解决多线程并行情况下使用锁造成性能损耗的一种机制,CAS操作包含三个操作数--内存位置(V).预期原值(A)和新值(B).如果内存位置的值与预期原值相匹配,那么处理器会自动将该位置值更新为新值.否则,处理器不做任何操作.无论哪种情况,它都会在CAS指令之前返回该位置的值.CAS有效地说明了"我认为位置V应该包含值A:如果包含该值,则将B放到这个位置:否则,不要更改该位置,只告诉我这个位置现在的值即可. 当同时存在读写线程时,默认情况下是不保证线

【DPDK】【ring】从DPDK的ring来看x86无锁队列的实现

[前言] 队列是众多数据结构中最常见的一种之一.曾经有人和我说过这么一句话,叫做“程序等于数据结构+算法”.因此在设计模块.写代码时,队列常常作为一个很常见的结构出现在模块设计中.DPDK不仅是一个加速网络IO的框架,其内部还提供众多的功能组件,rte_ring就是DPDK内部提供的一种无锁队列,本篇文章将从使用的角度出发阐述DPDK的ring怎么用?在怎么用的角度上再来阐述ring无锁的实现,最后将探讨实现无锁队列的关键以及在不通平台上如何实现,本文将会探讨x86平台下无锁队列的实现. 权当抛

多线程编程之无锁队列

关于无锁队列的概念与实现,可以参考博文<无锁队列的实现>,主要涉及到的知识点包括CAS原子操作.无锁队列的链表实现.无锁队列的数组实现以及ABA问题. 下面借鉴了<多线程的那点儿事(之无锁队列)>的代码,说明两个线程(一个添加一个读取数据)之间的无锁队列,可以不借助线程互斥方法就能够达到并行效果.代码如下: #define MAX_NUMBER 1000L #define STATUS int #define OK 0 #define FALSE -1 typedef struct

转 Linux kernel 无锁队列

struct kfifo { unsigned char *buffer;    /* the buffer holding the data */ unsigned int size;    /* the size of the allocated buffer */ unsigned int in;    /* data is added at offset (in % size) */ unsigned int out;    /* data is extracted from off. 

非阻塞同步算法与CAS(Compare and Swap)无锁算法

CAS无锁算法 要实现无锁(lock-free)的非阻塞算法有多种实现方法,其中CAS(比较与交换,Compare and swap)是一种有名的无锁算法.CAS, CPU指令,在大多数处理器架构,包括IA32.Space中采用的都是CAS指令,CAS的语义是"我认为V的值应该为A,如果是,那么将V的值更新为B,否则不修改并告诉V的值实际为多少",CAS是项乐观锁技术,当多个线程尝试使用CAS同时更新同一个变量时,只有其中一个线程能更新变量的值,而其它线程都失败,失败的线程并不会被挂起

深入理解SPDK 之二： 消息和无锁队列

并发理论 期望: 随着硬件的线性增加,性能也线性增加: 传统 锁 的优点: 无锁 到有锁 扩展的方便: 锁直接加在共享数据的前面: 缺点:随着系统内线程的增加,数据和临界段的竞争越来越激烈,而且多个线程可能处于不同物理core,频繁竞争导致剧烈的L1 cache 失效. SPDK的做法:各管各妈:你看我娃 ** 各管各妈 SPDK takes a different approach altogether. Instead of placing shared data in a global l