自旋锁的实现及优化

自旋锁也是一种互斥锁，和mutex锁相比，它的特点是不会阻塞，如果申请不到锁，就会不断地循环检测锁变量的状态，也就是自旋。自旋锁的实现算法大多使用TAS算法或者CAS算法。

TAS算法
即Test And Set，顾名思义，就是测试并设置，也就是先对目标值进行检查，如果目标值符合预期的结果则把它修改为所需要的值，并返回设置前的原值。它的语义原型，可以使用下面的伪代码表示：

<< atomic >>
function tas(p : pointer to bool) returns bool {
    bool value = *p
    if !*p {
        *p ←true
    }
    
    return value
}

如果指针p指向的变量原值为false，就设置为true，并返回false；如果为true，就直接返回true。这个tas操作过程要求是一个原子操作，中间不允许被打断，也就是保证了比较操作（*p是否为false）和修改操作（*p置为true）是一个原子操作，在X86环境下，使用xchg指令来实现。

CAS 算法
即Compare And Swap，把当前值与一个期望值作比较，如果相等，则更新为新值，并返回比较结果。它的语义原型，可以使用下面的伪代码表示：

<< atomic >>
function cas(p : pointer to int, old : int, new : int) returns bool {
    if *p ≠ old {
        return false
    }
    *p ← new
    return true
}

它的输入参数一共有三个，分别是：
p: 指向要修改的变量，old: 旧值，new: 新值。返回值是布尔类型，表示赋值是否成功。
CAS 实现逻辑也非常简单，就是先检查p所指位置的值是不是等于期望值old，如果等于，那就把该位置的值更新为new，并返回 true，否则就不做任何操作，返回false。同样要保证比较和修改是一个原子操作，在X86环境下，使用lock cmpxchg指令。

下面以TAS算法为例，介绍一下自旋锁的实现及其优化要点（C++11、X86环境）。

atomic_flag
在C++11版中原子操作类atomic_flag提供了TAS算法，下面使用atomic_flag实现一个自旋锁类splin_lock：

class spin_lock final
{
public:
	void lock() {
		while (flag.test_and_set(memory_order_acquire));
	}

	void unlock() {
		flag.clear(memory_order_release);
	}

private:
	atomic_flag flag;
};

使用atomic_flag类型的变量flag作为锁同步变量，在加锁函数lock()中，test_and_set()实现了TAS原语操作。当flag为true时，返回true，说明这个锁已经被占有，其它的线程无法获取了；在解锁函数unlock()中，使用clear()让flag复位0，表示释放锁。

此外，使用自旋锁时，为了保证临界区内共享变量的可见性和顺序性，还要指定内存序（memory order）保证内存一致性，即在解锁时是释放（release）语义，在加锁时是获取（acquire）语义。这样，能够保证解锁时在临界区修改的数据，在获得锁后进入临界区可见，内存操作不产生乱序现象。

实现自旋锁的关键之处是test_and_set函数，前面说过，要保证整个函数的执行是原子的，在X86环境下，它是使用xchg指令来实现的。

xchg指令
进行编译后，lock函数的汇编代码如下：（略有调整）

.L2:
        mov     eax, 1
        xchg    al, BYTE PTR sync_var
        test    al, al
        jne     .L2
        ret

lock函数里面的主要指令是“xchg al, BYTE PTR sync_var”，它是把寄存器al中的值（初始化为1），与同步变量sync_var的值进行交换。然后检查交换后al中的值，如果其值为1，说明同步变量sync_var的原值是1，应该是被别的线程成功申请锁后写入的，说明该锁已经被占用了，那就继续循环，直到交换后al中的值为0，此时sync_var同时也被置为1了，从lock函数中返回。

查看intel指令集手册，会发现xchg指令的功能是交换寄存器和内存变量中的值，显然是一个“读取-修改-写回”（RMW）的操作。尽管没有在这个指令前面加上lock前缀，但CPU在执行这个指令期间，自动实现了锁协议，执行时会有一个#lock信号产生，用来锁定内存总线，直到xchg指令执行完成，可见，该条指令是一个原子操作。我们知道，在锁内存总线期间，意味着别的CPU无法使用访问内存，显然这是为了实现硬件原子操作所付出的代价。

从上面的汇编代码可以看到，在调用lock函数申请锁时，只要lock_var变量不为0，就会无条件的循环执行xchg指令，xchg指令执行的太频繁了，以至于其它CPU的运行都要受到影响。因为一个CPU在xchg执行期间要独占内存总线，其它CPU要访问内存就得和这个CPU竞争内存总线，如果竞争不到，就得停下来等待内存总线的释放，造成CPU一定程度上处于“失速”状态，如果几个线程同时在申请锁，进一步加剧了内存总线的竞争程度。也就是在一个CPU竞争自旋锁期间，其它CPU尽管没有参与竞争，但是也受到影响，导致性能下降，对其它CPU来说，这不是一个高效的实现方案，因为它降低了系统的整体性能。
如何进行优化呢？

双检查锁机制
我们知道在实现单例模式时，有一种针对互斥锁的优化方案，叫做双检查锁机制，它的基本思想是先使用执行成本低的代码来检查条件是否满足，如果条件满足，再使用有锁的逻辑继续检查条件并进行设置，也即TAS操作。

基于这个思想，也可以使用双检查锁机制来进行优化，下面是修改后的lock函数代码，不过atomic_flag提供的成员函数test()在C++20版本才提供，需要支持C++20版本的编译器才能编译。

void lock() {
    while (true) {
        while (flag.test(memory_order_relaxed)); // 第一次检查：Test
        if (!flag.test_and_set(memory_order_acquire)) { // 第二次检查并设置：TAS操作
            break;
        }
    }
}

编译后生成的汇编代码如下（略有调整）：

        mov     edx, 1
.L2:
        movzx   eax, BYTE PTR sync_var // 从内存中读取同步变量的值
        test    al, al // 判断，第一次检查 Test
        jne     .L2 // 条件不符合，回退
        mov     eax, edx
        xchg    al, BYTE PTR sync_var // 第二次检查并设置：TAS操作
        test    al, al
        jne     .L2
        ret

先看一下第一次检查，在检查时，每次都要从内存中读取变量sync_var，然后进行检查是否为0，如果不为0，则继续读取再检查，一直循环执行“读取-判断-回退”的逻辑，直到读出的sync_var为0为止。注意，这里的读取指令是“movzx eax, BYTE PTR sync_var”，是非常简单的指令，它从内存中加载一个字节的数据到寄存器al中，也是一条原子操作指令，但该指令不会锁总线。根据MESI缓存一致性协议，如果sync_var的值没有被别的CPU修改过，那么它就在当前CPU的L1 cache缓存中会一直有效，也就是只要内存中的sync_var的值没有被更新，该指令执行时就会一直从L1 cache中读取，即没有访问内存，也就不会占用内存总线，所以第一次的循环检查也被称为“本地自旋”（见汇编语句3（读取）、4（判断）、5（回退））。

当本地自旋检测到sync_var值变为0后，就退出自旋，立刻进入第二次检查，这时为什么还要进行检查呢？因为第一次检查不能原子地设置sync_var为1，所以还得要使用原子指令xchg再一次进行检查并设置，由于已经检测到sync_var为0了，这次执行xchg指令成功的几率非常高。

可见，双检查锁在自旋锁中的实现逻辑是Test-Test-And-Set，所以有时候也简称为TTAS算法。同自旋锁的TAS算法相比，TTAS算法大大减轻了对内存总线的竞争程度，也降低了xchg执行时为了维持cache一致性而产生的内存总线流量，不过，因为要进行两次检查判断，申请锁的速度没有TAS算法快，也就是TTAS算法申请锁时的延迟比TAS算法大。

pause指令
再看TTAS算法的代码实现，当第一次检查时，如果一个CPU从本地自旋中退出，说明此时已经有别的CPU把这个sync_var变量置为0了，即别的线程已经把锁释放了。当接着使用TAS操作进行第二次检查时，如果没有设置成功，说明在第一次检查结束和第二次检查开始之间出现了状况：第一种可能是执行被中断了，即有优先级更高的线程把它抢占了；第二种可能是和别的CPU在执行时间上有重叠，让别的CPU抢先获得了锁。
如果发生了第二种情况，别的CPU获得锁后要进入临界区进行业务处理，这也暗示着自旋锁不会马上被释放，那么本CPU肯定会再次进入“本地自旋”并且会持续一段时间。我们不妨估算一下大概需要自旋多长时间：如果想让CPU从本地自旋中退出，至少要等到另一个CPU执行完临界区代码，调用unlock时，把0值写入内存变量sync_var，本CPU读取sync_var变量时发现缓存和内存中的值不一致，重新从内存中加载；假设临界区没有代码，它的执行时间为0（实际上是不可能的），也就是说最少也要等待CPU一次写内存和一次读内存的时间，小于这个时间是不会从循环中退出的，这段时间CPU一直处于空转状态。同时，我们知道“本地自旋”实质上是一个“死循环”，指令流非常短，全部在L1 cache中，而且数据也在L1 cache中，执行期间几乎不访问主内存，执行速度极快，也就是CPU处于高速运转中，电量消耗非常大，显然这也是一大不足。
既然能够预见到这种情况肯定会发生，能否针对此情况进行改进呢？intel的CPU专门针对自旋等待的场景提供了一个指令：pause，该条指令的功能是给CPU一个提示：当前正在进行自旋等待，可以暂停一下，比如等待一个或几个内存总线操作周期的时间之后，然后再继续运行（因为在这期间，自旋锁是大概率获取不到的，与其让CPU自旋忙等，不如停下来休息一下）。

因为C++中没有相应的函数来执行pause指令，所以嵌入了汇编指令，下面修改后的代码：

void lock() {
    while (true) {
        while (flag.test()) {
            asm("pause");
        }
        if (!flag.test_and_set(memory_order_acquire)) {
            break;
        }
    }
}

显然这情况能够大大降低CPU执行资源和电量的消耗。不过，这种方法进一步加大了获取锁的延迟性，因为每次自旋条件不符合需要就得先pause一段时间，如果在pause期间，锁被释放了，不会及时的获取锁，只能等到CPU从pause中退出时才有可能。而且这个延迟的时间也不固定：有可能CPU刚刚开始pause，锁就释放了，显然此时延迟最大；也有可能CPU从pause退出时，恰好锁就释放了，显然此时延迟最小，这是最理想的场景。如果应用场景是对响应性能敏感的场景，就得要慎重考虑了，能否容忍使用pause指令带来的延迟损失。

同时我们也知道，CPU处于pause状态时，什么也不做，不像操作系统提供的yield命令，可以让CPU执行别的任务。让CPU资源处于闲置状态，可能会感觉有点可惜，不过在支持超线程技术的CPU架构体系中可以利用这个特点，在一定程度上提高超线程的执行速度。现代CPU大多提供了超线程技术，即在一个物理核上有两个逻辑核，它们有自己独享的寄存器，但是需要共享执行单元、系统总线和缓存等硬件资源。如果执行自旋操作的CPU是一个逻辑核，当它进入本地自旋的时候，会加剧同一个物理核中的执行资源和系统总线的竞争程度，从而影响其它逻辑核的执行效率；但是，如果一个逻辑核进入了pause状态，就不再使用系统资源了，同物理核的另一个逻辑核就可以独享物理核的全部资源了。

不过需要指出的是，当自旋锁的临界区的执行时间很短时，比如只是简单的几个内存访问和计算，可以使用自旋锁+PAUSE技术；如果临界区执行时间较为耗时，或者发生了前面说的第一种场景，即线程被抢占后中断执行了，使用pause会让一个CPU长时间处于暂停状态，也是一种资源浪费，不如把CPU分配给别的线程使用。这时可以考虑让操作系统参与进来，比如调用yield，把CPU调度给别的任务（如NGINX实现的自旋锁），或者直接把当前线程挂起，等到解锁时再唤醒它，当然这要涉及到线程上下文的切换，也就是所谓的重量级互斥锁了（如mutex锁）。

常见平台的自旋锁实现
下面是一些平台所实现的自旋锁加锁函数的代码，也采用了上面介绍的实现方法，只是形式有所不同，其中nginx还对pause的次数进行定制，执行一定次数之后还会调用yield，让出CPU的资源，大家不妨分析一下：
1、DPDK

typedef struct {
	volatile int locked; /**< lock status 0 = unlocked, 1 = locked */
} rte_spinlock_t;

static inline void
rte_spinlock_lock(rte_spinlock_t *sl)
{
	while (__sync_lock_test_and_set(&sl->locked, 1))
		while(sl->locked)
			rte_pause();
}

3、NGINX（去掉了一些不相关代码）

void ngx_spinlock(ngx_atomic_t *lock, ngx_atomic_int_t value, ngx_uint_t spin)
{
    ngx_uint_t  i, n;

    for ( ;; ) {
        if (*lock == 0 && ngx_atomic_cmp_set(lock, 0, value)) {
            return;
        }

        if (ngx_ncpu > 1) {
            for (n = 1; n < spin; n <<= 1) {
                for (i = 0; i < n; i++) {
                    ngx_cpu_pause();
                }
                if (*lock == 0 && ngx_atomic_cmp_set(lock, 0, value)) {
                    return;
                }
            }
        }

        ngx_sched_yield();
    }
}