用原子操作实现无锁编程[通俗易懂]

大家好，又见面了，我是你们的朋友全栈君。

假设我们要维护一个全局的线程安全的 int 类型变量 count, 下面这两行代码都是很危险的:

count ++;

count += n;

我们知道, 高级语言中的一条语句, 并不是一个原子操作. 比如一个最简单的自增操作就分为三步:

1. 从缓存取到寄存器 2. 在寄存器加1 3. 存入缓存。

多个线程访问同一块内存时, 需要加锁来保证访问操作是互斥的.

所以, 我们可以在操作 count 的时候加一个互斥锁. 如下面的代码：

pthread_mutex_t count_lock = PTHREAD_MUTEX_INITIALIZER;

pthread_mutex_lock(&count_lock); count++; pthread_mutex_unlock(&count_lock);

另一个办法就是, 让 count++ 和 count+=n 这样的语句变成原子操作. 一个原子操作必然是线程安全的. 有两种使用原子操作的方式:

1. 使用 gcc 的原子操作

2. 使用 c++11中STL中的 stomic 类的函数

在这里我只介绍 gcc 里的原子操作, 这些函数分成以下几组:

type __sync_fetch_and_add (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_add_and_fetch (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, …)

返回更新后的值

bool __sync_bool_compare_and_swap (type *ptr, type oldval type newval, …)

type __sync_val_compare_and_swap (type *ptr, type oldval type newval, …)

这两个函数提供原子的比较和交换，如果*ptr == oldval,就将newval写入*ptr,

第一个函数在相等并写入的情况下返回true.

第二个函数在返回操作之前的值。

type __sync_lock_test_and_set (type *ptr, type value, …)

将*ptr设为value并返回*ptr操作之前的值。

void __sync_lock_release (type *ptr, …)

将*ptr置0

因为 gcc 具体实现的问题, 后面的可扩展参数 (…) 没有什么用, 可以省略掉.

gcc 保证了这些接口都是原子的. 调用这些接口时, 前端串行总线会被锁住, 锁住了它, 其它 cpu 就不能从存储器获取数据. 从而保证对内存操作的互斥. 当然, 这种操作是有不小代价, 所以只能在操作小的内存才可以这么做. 上面的接口使用的 type 只能是 1, 2, 4 或 8 字节的整形, 即：

int8_t / uint8_t

int16_t / uint16_t

int32_t / uint32_t

int64_t / uint64_t

性能上原子操作的速度是互斥锁的6～7倍。

有了这些函数, 就可以很方便的进行原子操作了, 以 count++ 为例,

count 初始值为0, 可以这么写

__sync_fetch_and_add(&count, 1);//返回0, count现在等于1, 类似 count ++

count 初始值为0, 或者这么写

__sync_add_and_fetch(&count, 1);//返回1, count现在等于1, 类似 ++ count

原子操作也可以用来实现互斥锁:

int a = 0;

#define LOCK(a) while (__sync_lock_test_and_set(&a,1)) {sched_yield();}

#define UNLOCK(a) __sync_lock_release(&a);

sched_yield()这个函数可以使用另一个级别等于或高于当前线程的线程先运行。如果没有符合条件的线程，那么这个函数将会立刻返回然后继续执行当前线程的程序。

如果去掉 sched_yield(), 这个锁就会一直自旋.

下面我们利用原子操作来实现一个无锁并发堆栈;

struct Node{

void* data;

Node* next

Node(void* d):data(d),next(NULL){}

};

class Stack{

public:

Stack():top(NULL){}

void Push(void* d);

void* Pop();

private:

Node *top;

};

void Stack::Push(void* d){

Node* n = new Node(d);

for (;;){

n->next = top;

if (__sync_bool_compare_and_swap(&top, n->next, n)){

break;

}

}

}

压栈操作首先创建了一个新节点，它的 next 指针指向堆栈的顶部。然后用原子操作把新的节点复制到 top 位置。 从多个线程的角度来看，完全可能有两个或更多线程同时试图把数据压入堆栈。假设线程 A 试图把 pA 压入堆栈，线程 B 试图压入 pB，线程 A 先获得了时间片。在 n->next = top 指令结束之后，调度程序暂停了线程 A。现在，线程 B 获得了时间片，它能够完成原子操作，把 pB 压入堆栈后结束。接下来，线程 A 恢复执行，显然对于这个线程 *top 和 n->next 不匹配，因为线程 B 修改了 top 位置的内容。因此，代码回到循环的开头，指向正确的 top 指针（线程 B 修改后的），调用原子操作，把 pA 压入堆栈后结束。

void* Stack::Pop(){

for (;;){

Node* n = top;

if (n == NULL){

return NULL;

}

if (top != NULL && __sync_bool_compare_and_swap(&top, n, n->next)){

void* p = n->data;

delete n;

return p;

}

}

}

出栈操作的原理和压栈类似. 即使线程 B 在线程 A 试图弹出数据的同时修改了堆栈顶，也可以确保不会跳过堆栈中的元素。

发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/140637.html原文链接：https://javaforall.cn