【原版的】无锁编程与实现-白红宇

【原版的】无锁编程与实现

阅读量：5292 次

发布时间：2019-06-14

本文共 9897 字，大约阅读时间需要 32 分钟。

无锁编程技术及实现

作者：jx （360电商技术组）

1.基于锁的编程的缺点

多线程编程是多CPU系统在中应用最广泛的一种编程方式，在传统的多线程编程中，多线程之间一般用各种锁的机制来保证正确的对共享资源（share resources）进行訪问和操作。

在多线程编程中仅仅要须要共享某些数据，就应当将对它的訪问串行化。比方像++count(count是整型变量)这种简单操作也得加锁，由于即便是增量操作这种操作。，实际上也是分三步进行的：读、改、写（回）。

movl x, %eax

addl $1, %eax

movl %eax, x

更进一步，甚至内存变量的赋值操作都不能保证是原子的，比方在32位环境下运行这种函数

void setValue()

{

value = 0x100000006;

}

运行的过程中，这两条指令之间也是能够被打断的。而不是一条原子操作。（也就是所谓的写撕裂）

所以改动共享数据的操作必须以原子操作的形式出现，这样才干保证没有其他线程能在中途插一脚来破坏相应数据。

而在使用锁机制的过程中。即便在锁的粒度（granularity）。负载（overhead），竞争（contention），死锁（deadlock）等须要重点控制的方面解决的非常好，也无法彻底避免这种机制的例如以下一些缺点：

1，锁机制会引起线程的堵塞（block），对于没有能占用到锁的线程或者进程，将一直等待到锁的占有者释放锁资源后才干继续运行。而等待时间理论上是不可设置和预估的。

2。申请和释放锁的操作，添加了非常多訪问共享资源的消耗。尤其是在锁竞争(lock-contention)非常严重的时候。比方这篇文章所说：

3，现有实现的各种锁机制。都不能非常好的避免编程开发人员设计实现的程序出现死锁或者活锁的可能

4。优先级反转（prorithy inversion）和锁护送(Convoying)的现象

5，难以调试

无锁编程（Lock-Free）就是在某些应用场景和领域下解决以上基于锁机制的并发编程的一种方案。

2.无锁编程(LOCK-FREE)的定义

提到无锁编程（lock-free）。按字面最直观的理解是不使用锁的情况下实现多线程之间对变量同步和訪问的一种程序设计实现方案。严格的说这个理解是不正确的。Lock-Free的程序肯定是不包含锁机制的，而不包含锁机制的程序不一定是lock-free的。

更准确的说。在并发编程上依照同步的维护划分，能够分为堵塞的编程方式（Block）和非堵塞的编程方式（Non-blocking Synchronization）。堵塞的编程方式基本是基于锁的（lock-based）。当中无锁编程（Lock-free）属于非堵塞同步（Non-blocking Synchronization）中的一种情况，实现非堵塞同步的算法方案依照效果要求不同能够粗略的分为：

Wait-free: 满足等待无关的程序，不论什么线程能够在有限步之内结束，无论其他线程的运行速度和进度怎样

Lock-free：锁无关的程序，一个锁无关的程序能够确保它全部线程中至少有一个能够继续往下运行。而有些线程可能会被的延迟。然而在总体上，在某个时刻至少有一个线程能够运行下去。作为总体进程总是在前进的，虽然有些线程的进度可能没有其他线程进行的快。

Obstruction-free：在不论什么时间点。一个孤立运行线程的每个操作能够在有限步之内结束。仅仅要没有竞争，线程就能够持续运行。一旦共享数据被改动，Obstruction-free 要求中止已经完毕的部分操作进行回滚。

更仔细的还能够把并发编程按效果划分为:

Blocking

1. Blocking

2. Starvation-Free

Obstruction-Free

3. Obstruction-Free

Lock-Free

4. Lock-Free (LF)

Wait-Free

5. Wait-Free (WF)

6. Wait-Free Bounded (WFB)

7. Wait-Free Population Oblivious (WFPO)

详细细节能够參考这篇文章。有对堵塞非堵塞的等定义的详细描写叙述，这里不详细论述。

3.无锁编程中涉及的一些技术原理

无锁编程详细使用和考虑到的技术方法包含：原子操作（atomic operations）, 内存栅栏（memory barriers）, 内存顺序冲突（memory order）。指令序列一致性（sequential consistency）和顺ABA现象等等。这方面借用一篇资料的总结的图概况：

在这当中最基础最重要的是操作的原子性或说原子操作。原子操作能够理解为在运行完毕之前不会被不论什么其他任务或事件中断的一系列操作。原子操作是非堵塞编程最核心主要的部分，没有原子操作的话，操作会由于中断异常等各种原因引起数据状态的不一致从而影响到程序的正确。

对于原子操作的实现机制。在硬件层面上CPU处理器会默认保证主要的内存操作的原子性。CPU保证从系统内存当中读取或者写入一个字节的行为肯定是原子的，当一个处理器读取一个字节时，其他CPU处理器不能訪问这个字节的。可是对于复杂的内存操作CPU处理器不能自己主动保证其原子性，比方跨总线宽度或者跨多个缓存行（Cache Line），跨页表的訪问等。这个时候就须要用到CPU指令集中设计的原子操作指令，如今大部分CPU指令集都会支持一系列的原子操作。而在无锁编程中经经常使用到的原子操作是Read-Modify-Write （RMW）这种类型的，这当中最经常使用的原子操作又是 COMPARE AND SWAP（CAS），差点儿全部的CPU指令集都支持CAS的原子操作，比方X86平台下中的是 CMPXCHG。

继续说一下CAS，CAS操作行为是比較某个内存地址处的内容是否和期望值一致。假设一致则将该地址处的数值替换为一个新值。CAS能够操作的位数越多，使用它来实现锁无关的数据结构就越easy（细节能够在intel手冊中查看）。CAS操作详细的实现原理主要是两种方式：总线锁定和缓存锁定。

所谓总线锁定，就是CPU运行某条指令的时候先锁住数据总线的，使用同一条数据总线的CPU就无法訪问内存了，在指令运行完毕后再释放锁住的数据总线。锁住数据总线的方式系统开销非常大，限制了訪问内存的效率。所以又有了基于CPU缓存一致性来保持操作原子性作的方法作为补充。简单来说就是用CPU的缓存一致性的机制来防止内存区域的数据被两个以上的处理器改动（可详见CPU缓存的MESI协议）。

在操作系统的层面，Linux系统提供了软件级的原子操作，包含两大类系统调用，一类是基于对整数进行操作的atomic_set/and/inc，一类是针对单独的位进行操作的set/clear/change_bit，它们大部分都是基于硬件层面的CAS的指令实现的。

在各种开发语言中（c,c++,java）基于操作系统提供的接口也都封装实现了相应的原子操作api。所以开发人员全然能够直接调用各个开发语言提供的接口实现无锁程序。

除了使用原子操作保证操作的原子性。还要考虑在不用的语言和内存模型下，怎样保证，操作的顺序性，编译时和运行时的指令重排序。还有由假共享引起内存顺序冲突（Memory order violation）等等细节。原子性确保指令运行期间不被打断，要么全部运行，要么根本不运行。而顺序性确保即使两条或多条指令出如今独立的运行线程中或者独立的处理器上时。保持它们本该运行的顺序。

假共享是指多个CPU同一时候改动同一个缓存行的不同部分而引起当中一个CPU的操作无效，当出现这个内存顺序冲突时。CPU必须清空流水线。这些事实上在多核编程的环境下的locked-base的实现中也有相似的问题所以这里就不多展开。

4.几个样例

John D. Valois 《》中提到的一个基于链表的无锁队列链表的实现，能够作为了解lock-free一个样例

EnQueue(x) { //入队列方法

q = new record();

q->value = x; //队列节点的值

q->next = NULL;//下一个节点

p = tail; //保存尾节点指针

oldp = p;

do { //開始 loop cas

while (p->next != NULL) //用来防止进行cas（tail,oldp,q）操作的线程挂掉引起死循环

p = p->next;

} while( CAS(p.next, NULL, q) != TRUE);

CAS(tail, oldp, q);

}

DeQueue() //出队列方法

{

do{

p = head;

if (p->next == NULL){

return ERR_EMPTY_QUEUE;

}

while( CAS(head, p, p->next) != TRUE );

return p->next->value;

}

详细实现中，在linux 平台上gcc编译器（内置了支持cas操作的函数，或者直接调用汇编命令（比方CMPXCHG 和 CMPXCHG8B）也能够在c程序里实现cas操作。GCC (GNU Compiler Collection，4.1 和更高版本号）提供几个内置函数，能够使用它们在 x86 和 x86-64 平台上实现 CAS 操作，这些内置函数包含：

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, ...)

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

很多其他细节分析可參看John D. Valois的《》

相同，在java语言中Lock-Free的数据结构和算法事实上更加常见，在java.util.concurrent包中大量实现都是建立在基于CAS实现Lock-Free算法上，没有CAS就不会有此包。Java.util.concurrent.atomic提供了基于CAS实现的若干原语：

AtomicBoolean -- 原子布尔

AtomicInteger -- 原子整型

AtomicIntegerArray -- 原子整型数组

AtomicLong -- 原子长整型

AtomicLongArray -- 原子长整型数组

AtomicReference -- 原子引用

AtomicReferenceArray -- 原子引用数组

AtomicMarkableReference -- 原子标记引用

AtomicStampedReference -- 原子戳记引用

AtomicIntegerFieldUpdater -- 用来包裹对整形 volatile 域的原子操作

AtomicLongFieldUpdater -- 用来包裹对长整型 volatile 域的原子操作

AtomicReferenceFieldUpdater -- 用来包裹对对象 volatile 域的原子操作

引入这些类的主要目的就是为了实现Lock-Free算法和相关数据结构，以incrementAndGet这种方法为例：

public final int incrementAndGet() {

for (;;) {

int current = get();

int next = current + 1;

if (compareAndSet(current, next))

return next;

}

在这里使用CAS原子操作，每次读取数据数值后将此数值和+1后的结果进行CAS操作。假设成功就返回结果，否则重试到成功为止。当中compareAndSet是java中实现的CAS函数，在java语言中的实现，是借助JNI机制来调用汇编实现的：

public final boolean compareAndSet(int expect, int update) {

return unsafe.compareAndSwapInt(this, valueOffset, expect, update);

}

unsafe.compareAndSwapInt是个本地方法调用，相应的x86处理器的jni源代码

#define LOCK_IF_MP(mp) __asm cmp mp, 0 \

__asm je L0 \

__asm _emit 0xF0 \

__asm L0:

inline jint Atomic::cmpxchg (jint exchange_value, volatile jint* dest, jint compare_value) {

// alternative for InterlockedCompareExchange

int mp = os::is_MP();

__asm {

mov edx, dest

mov ecx, exchange_value

mov eax, compare_value

LOCK_IF_MP(mp)//推断是否是多核，是则加入LOCK指令维护顺序一致性

cmpxchg dword ptr [edx], ecx

}

再看看在java里无锁队列的实现作为对照參照：

import java.util.concurrent.atomic.AtomicReference;

public class LockFreeQueue<T> {

private AtomicReference<Node> head; //能够规避ABA现象

private AtomicReference<Node> tail;

/**

* Create a new object of this class.

public LockFreeQueue() {

Node sentinel = new Node(null);

　 head = new AtomicReference<Node>(sentinel);

　 tail = new AtomicReference<Node>(sentinel);

}

/**

* Enqueue an item.

* @param value Item to enqueue.

public void enq(T value) {

　 // try to allocate new node from local pool

　 Node node = new Node(value);

　 while (true) {

　　 Node last = tail.get();　

　　 Node next = last.next.get();

　　 // are they consistent

　　 if (last == tail.get()) {

　　　 if (next == null) {

　　　　 // try to link node to end of list

　　　　 if (laspareAndSet(next, node) {

　　　　　 // enq done, try to advance tail

　　　　　 tail.compareAndSet(last, node);

　　　　　 return;

　　　　 }

　　　 } else {

　　　　 // try to swing tail to next node

　　　　 tail.compareAndSet(last, next);

　　　 }

/**

* Dequeue an item.

* @throws queue.EmptyException The queue is empty.

* @return Item at the head of the queue.

public T deq() throws EmptyException {

　 while (true) {

　　 Node first = head.get();

　　 Node last = tail.get();

　　 Node next = first.next.get();

　　 // are they consistent

　　 if (first == head.get()) {

　　　 if (first == last) {

　　　　 if (next == null) {

　　　　　 throw new EmptyException();

　　　　 }

　　　　 // tail is behind, try to advance

　　　　 tail.compareAndSet(last, next);

　　　 } else {

　　　　 T value = next.value;

　　　　 if (head.compareAndSet(first, next)) {

　　　　　 return value;

　　　　 }

/**

* Items are kept in a list of nodes.

public class Node {

　 /**

　　 * Item kept by this node.

　 public T value;

　 /**

　　 * Next node in the queue.

　 public AtomicReference<Node> next;

　　/**

　　 * Create a new node.

　 public Node(T value) {

　　 this.next = new AtomicReference<Node>(null);

　　 this.value = value;

　 }

最后这里随便说一下CAS操作的ABA的问题。所谓的ABA的问题简要的说就是，线程a先读取了要对照的值v后，被线程b抢占了。线程b对v进行了改动后又改会v原来的值，线程1继续运行运行CAS操作的时候，无法推断出v的值被改过又改回来。

ABA对基于指针实现的算法影响非常大。上面的C语言里实现的程序是有可能被ABA问题影响的，由于它的CAS操作推断的是指针的地址，这个地址被重用的可能性非常大（地址被重用是非常经常发生的。一个内存分配后释放了，再分配。非常有可能还是原来的地址。内存管理中重用内存基本上是一种非经常见的行为）。解决ABA的问题的一种方法是，一次用CAS检查双倍长度的值，前半部是指针。后半部分是一个计数器。或者对CAS的数值加上版本号号。

《》也有提到相关的解决方式。

5.结论和建议

无锁编程方式相对基于锁的编程方式，具备一定的长处。比方不会发生死锁，不会有优先级倒置，进行CAS操作的消耗比加锁操作轻非常多等等。单从这个角度上讲在相应用程序不太复杂。而对操作实时性要求较高时，採用无锁多线程能发挥一定优势。

在性能上基于CAS实现的硬件级的相互排斥。其单次操作性能比相同条件下的应用层的较为高效，但当多个线程并发时，硬件级的相互排斥引入的消耗一样非常高(相似spin_lock)。无锁算法及相关数据结构并不意味在全部的环境下都能带来总体性能的极大提升。循环CAS操作对时会大量占用cpu。对系统时间的开销也是非常大。这也是基于循环CAS实现的各种自旋锁不适合做操作和等待时间太长的并发操作的原因。而通过对有锁程序进行合理的设计和优化，在非常多的场景下更easy使程序实现高度的并发性。

在开发维护的成本和复杂度上，无锁编程难度非常高，相似ABA的问题也比較难直观的探測和解决。而且实现细节和硬件平台相关性非常强。眼下理论和实践上仅仅有少数数据结构能够支持实现无锁编程。比方队列、栈、链表、词典等。眼下要在产品业务开发环境中进行大规模的无锁编程较为困难，很多其他的是用在部分特殊场景解决锁竞争等问题比較合适。比方操作系统中实现metux。semaphare, 一些语言的库的实现（比方 java current lib， lmax disprute）。

若想在应用开发中尝试的Lock-Free的方案。建议能够选择合适的第三方lib实现。

附一. 些常见的相关lib库工具

https://github.com/mthssdrbrg/LockFreeStack

http://www.cse.chalmers.se/research/group/noble/