本文介绍自旋锁的概念、实现以及优化

什么是自旋锁

自旋锁（spin lock）与互斥锁（mutex）类似，任时刻只有一个线程能够获得锁。当一个线程在获取锁的时候，如果锁已经被其它线程获取，那么该线程将循环等待，然后不断的判断锁是否能够被成功获取，直到获取到锁才会退出循环。

在获取锁的过程中，线程一直处于活跃状态。因此与mutex不同，spinlock不会导致线程的状态切换(用户态->内核态)，一直处于用户态，即线程一直都是active的；不会使线程进入阻塞状态，减少了不必要的上下文切换，执行速度快。

由于自旋时不释放CPU，如果持有自旋锁的线程一直不释放自旋锁，那么等待该自旋锁的线程会一直浪费CPU时间。因此，自旋锁主要适用于被持有时间短，线程不希望在重新调度上花过多时间的情况。

实现

根据自旋锁的原理，我们先尝试写出伪代码实现：

//循环检查锁状态，并尝试获取，直到成功
while (true):
  locked = get_lock()
  if locked == false:
    locked = true
    break

//上锁后执行相关任务
do_something

//执行完毕，释放锁
locked = false

细心的同学可以发现上面的逻辑在并发场景会遇到问题：两个线程可能会同时进入if语句，同时获取锁，导致锁不生效。

如何解决这个问题？我们可以把查询锁（get_lock）和设置锁（locked=true）组合成一个原子操作，保证同一时间只有一个线程在执行。如下所示：

//这里get_and_set(locked)就是一个原子操作，执行成功后把locked设置成true。
while (get_and_set(locked) == false):
  continue

do_something

locked = false

如此，就实现了一个简单的自旋锁。

那么现在的问题是如何实现get_and_set(locked)这个原子操作？这里有两个常用的方法可以使用：TAS（test and set）和CAS （compare and swap）。

TAS：一个TAS指令包括两个子步骤，把给定的内存地址设置为1，然后返回之前的旧值。
CAS：CAS指令需要三个参数，一个内存位置(V)、一个期望旧值(A)、一个新值(B)。过程如下：

a. 比较内存V的值是否与A相等？ b. 如果相等，则用新值B替换内存位置V的旧值 c. 如果不相等，不做任何操作。 d. 无论哪个情况，CAS都会把内存V原来的值返回。

很多语言都提供了封装后的TAS和CAS调用方法。

以C++ 11为例，atomic标准库提供了相关方法：std::atomic_flag::test_and_set和std::atomic::compare_exchange_strong
GCC编译器也内置了相关方法：__atomic_test_and_set和__atomic_compare_exchange_n.
Java也提供了例如java.util.concurrent.atomic.AtomicReference.compareAndSet等方法。

使用这些方法替换伪代码中的get_and_set(locked)，就能快速实现自旋锁。

参考实现

下面我们看看一些顶级开源项目中是如何实现自旋锁的。注意，以下每个软件的代码可能在很多地方都实现/使用了自旋锁，这里只选取了其中某一些加以分析。

MariaDB 10.4

代码路径：/storage/innobase/include/ib0mutex.h

struct TTASMutex {

    ...

    /** Try and lock the mutex.
    @return true on success */
    bool try_lock() UNIV_NOTHROW
    {
        uint32_t oldval = MUTEX_STATE_UNLOCKED;
        return m_lock_word.compare_exchange_strong(
            oldval,
            MUTEX_STATE_LOCKED,
            std::memory_order_acquire,
            std::memory_order_relaxed);
    }

    /** Acquire the mutex.
    @param max_spins max number of spins
    @param max_delay max delay per spin */
    void enter(uint32_t max_spins, uint32_t max_delay,
            const char*, uint32_t) UNIV_NOTHROW
    {
        const uint32_t step = max_spins;
        uint32_t n_spins = 0;

        while (!try_lock()) {
            ut_delay(max_delay);
            if (++n_spins == max_spins) {
                os_thread_yield();
                max_spins+= step;
            }
        }

        m_policy.add(n_spins, 0);
    }

    ...
}

如上所示，在TTASMutex结构里，有个enter方法，里面实现了上锁+自旋的功能。其中上锁动作调用了try_lock方法，里面使用了CAS原子操作。

在我们之前写的简单伪代码中，while循环内什么都没做（直接continue），即每次自旋之间无停顿、无其他操作。而mariaDB这里却做了一些动作，在每次while循环中：

执行ut_delay。即每次上锁失败后，会等待一段时间，然后再去尝试上锁。
判断自旋次数，当自旋次数达到某个阈值，不再自旋，直接线程挂起。

这样做可以防止某些自旋锁无限空转、浪费CPU资源的情况。

PostgreSQL

代码路径：src/include/storage/s_lock.h

/*
 * s_lock(lock) - platform-independent portion of waiting for a spinlock.
 */
int
s_lock(volatile slock_t *lock, const char *file, int line, const char *func)
{
    SpinDelayStatus delayStatus;

    init_spin_delay(&delayStatus, file, line, func);

    while (TAS_SPIN(lock))
    {
        perform_spin_delay(&delayStatus);
    }

    finish_spin_delay(&delayStatus);

    return delayStatus.delays;
}

我们可以看到，与MariaDB类似，while的判断中不断获取锁，while语句中加入delay。TAS_SPIN的实现如下：

static __inline__ int
tas(volatile slock_t *lock)
{
    return __sync_lock_test_and_set(lock, 1);
}

__sync_lock_test_and_set是gcc内置的老版本的TAS原子操作，推荐使用__atomic_test_and_set

Linux kernel

Kernel的spin lock很复杂，有多种实现，以arm64为例，在4.16版本之前，使用的是汇编实现。4.16之后使用了通用的qspinlock，qspinlock较复杂，请参考这里和这里。以后另开文章分析。

优化

原子操作优化

首先我们先看看TAS和CAS在汇编层面是什么样的？这里需要特别说明一下硬件架构和编译工具的选择，在不同硬件架构上使用不同版本的编译器，得到的汇编指令是不同的。本文以ARM64为例，使用gcc 8.2编译器, 编译参入为-O2 -std=c++11，分别执行__atomic_test_and_set和__atomic_compare_exchange_n。

`__atomic_test_and_set`

#include <atomic>

int main()
{
   int val = 456;
   __atomic_test_and_set(&val, 1);
}

汇编结果如下：

main:
        sub     sp, sp, #16
        mov     w0, 456
        add     x2, sp, 12
        str     w0, [sp, 12]
        mov     w0, 1
.L2:
        ldaxrb  w1, [x2]
        stxrb   w3, w0, [x2]
        cbnz    w3, .L2
        mov     w0, 0
        add     sp, sp, 16
        ret

`__atomic_compare_exchange_n`

#include <atomic>

int main()
{
   int expected = 123;
   int val = 456;
   __atomic_compare_exchange_n(&val, &expected, 1 , false, __ATOMIC_ACQUIRE, __ATOMIC_ACQUIRE); 
}

汇编结果如下：

main:
        sub     sp, sp, #16
        mov     w0, 456
        add     x2, sp, 12
        str     w0, [sp, 12]
        mov     w0, 1
.L2:
        ldaxr   w1, [x2]
        cmp     w1, 123
        bne     .L3
        stxr    w3, w0, [x2]
        cbnz    w3, .L2
.L3:
        mov     w0, 0
        add     sp, sp, 16
        ret

分析可以发现，TAS是直接写操作，CAS是先比较，满足条件后再写。而写是一个相对耗时的操作，因此在高并发、频繁使用锁的场景，CAS性能会更好。

因此在Postgre中，使用CAS代替TAS就是一个优化点了。详见社区讨论。

锁等待优化

在MariaDB和Postgre的源码中，我们可以看到在不断获取锁的过程中，都有delay的操作。适当的delay操作可以避免很多无意义的锁获取动作。那么当执行delay操作时，数据库到底在干什么？

我们先看MariaDB的ut_delay方法。

/**
  Run a delay loop while waiting for a shared resource to be released.
  @param delay originally, roughly microseconds on 100 MHz Intel Pentium
*/
static inline void ut_delay(unsigned delay)
{
  unsigned i= my_cpu_relax_multiplier / 4 * delay;
  HMT_low();
  while (i--)
    MY_RELAX_CPU();
  HMT_medium();
}

static inline void MY_RELAX_CPU(void)
{
#ifdef _WIN32
  /*
    In the Win32 API, the x86 PAUSE instruction is executed by calling
    the YieldProcessor macro defined in WinNT.h. It is a CPU architecture-
    independent way by using YieldProcessor.
  */
  YieldProcessor();
#elif defined HAVE_PAUSE_INSTRUCTION
  /*
    According to the gcc info page, asm volatile means that the
    instruction has important side-effects and must not be removed.
    Also asm volatile may trigger a memory barrier (spilling all registers
    to memory).
  */
#ifdef __SUNPRO_CC
  asm ("pause" );
#else
  __asm__ __volatile__ ("pause");
#endif
#elif defined(_ARCH_PWR8)
  __ppc_get_timebase();
#elif defined __GNUC__ && (defined __arm__ || defined __aarch64__)
  /* Mainly, prevent the compiler from optimizing away delay loops */
  __asm__ __volatile__ ("":::"memory");
#else
  int32 var, oldval = 0;
  my_atomic_cas32_strong_explicit(&var, &oldval, 1, MY_MEMORY_ORDER_RELAXED,
                                  MY_MEMORY_ORDER_RELAXED);
#endif
}

我们可以看到ut_delay中调用了MY_RELAX_CPU方法，而MY_RELAX_CPU方法中，根据不同架构和平台，执行了不同的命令。在X86架构中，调用了底层汇编指令PAUSE。在ARM64平台执行内存屏障(__asm__ __volatile__ ("":::"memory");)，防止编译优化，保持CPU空转（即不执行任何操作）。注意，ARM64的这个内存屏障代码是在这个优化提交中新增的。

在这个提交没有合入前，在ARM64平台上，默认执行的是my_atomic_cas32_strong_explicit方法，这是一个模拟无效的CAS的方法。为什么会有这样的修改？

最早的时候，社区开发者在48核的Qualcomm Centriq 2400 ARM服务上测试后，发现模拟CAS操作能提高ARM的性能。但随着代码不断迭代以及ARM服务器的不断发展。在MariaDB 10.4分支，开发者发现，在ARM上保持CPU空转的性能更高，有大概12%的性能提升。

所以我们很难在原理上讲清为什么CPU空转比模拟CAS性能高，但事实就是如此。通过这个优化，我也学到了两点，分享给大家：

性能优化有时不是看代码就能发现的，更多的还是需要实际测试。
性能是实时变化的，以前的优化可能放到后来就是负担。性能优化需要持续不断的展开。

kunpengcompute / kunpengcompute.github.io

自旋锁的原理与优化 #63