By

文章目录
  1. 1. 自旋锁(Spinlock)
  2. 2. 自旋锁的原理
  3. 3. 使用C++11的原子操作实现自旋锁
    1. 3.1. 指定内存序提高性能
    2. 3.2. 不使用CAS(compare-and-swap)的实现
    3. 3.3. 使用std::atomic_flag的实现

自旋锁(Spinlock)

自旋锁是一种用于保护多线程共享资源的锁，与一般的互斥锁(mutex)不同之处在于当自旋锁尝试获取锁的所有权时会以忙等待(busy waiting)的形式不断的循环检查锁是否可用。在多处理器环境中对持有锁时间较短的程序来说使用自旋锁代替一般的互斥锁往往能提高程序的性能。

自旋锁的原理

自旋锁有两种基本状态：

  1. 锁定状态
    锁定状态又称不可用状态，当自旋锁被某个线程持有时就是锁定状态，在自旋锁被释放之前其他线程不能获得锁的所有权。
  2. 可用状态
    当自选锁未被任何线程持有时的状态就是可用状态。

假设某自旋锁内部使用bool类型的flag变量来标识自旋锁的状态。当flag为true表示锁定状态，为false表示可用状态。
获取自旋锁的一种可行的流程如下

其中灰色方框中的3步应是一个不可分割的原子操作。
释放自旋锁时只需以原子操作的形式将flag置为false。

使用C++11的原子操作实现自旋锁

C++11提供了对原子操作的支持，其中std::atomic是标准库提供的一个原子类模板。可以这样来声明一个自旋锁互斥对象spin_mutex。

1
2
3
4
5
6
7
8
9
10
11
#include <atomic>

class spin_mutex {
  std::atomic<bool> flag = ATOMIC_VAR_INIT(false);
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock(); // 获取自旋锁
  void unlock(); // 释放自旋锁
};

对于lock函数，若根据前面的流程图来实现，需要CAS(compare and swap)的原子操作，可以使用std::atomic类模板的成员函数compare_exchange_strong。该成员函数的4个重载(overload)声明如下

1
2
3
4
5
6
7
8
9
10
bool compare_exchange_strong(T& expected, T desired,
                              std::memory_order success,
                              std::memory_order failure); // overload 1
bool compare_exchange_strong(T& expected, T desired,
                              std::memory_order success,
                              std::memory_order failure) volatile; // overload 2
bool compare_exchange_strong(T& expected, T desired,
                              std::memory_order order = std::memory_order_seq_cst); // overload 3
bool compare_exchange_strong(T& expected, T desired,
                              std::memory_order order = std::memory_order_seq_cst) volatile; // overload 4

compare_exchange_strong能够自动比较*this与expected的值，如果二者相等,会将*this的值修改为desired的值(执行read-modify-write操作)，否则将expected的值修改为*this的值。
当*this被改变时compare_exchange_strong返回true,否则返回false。
这4个重载中还有3个类型为std::memory_order的参数，这些参数用于指定内存序(memory order)，对内存序的介绍超出了本文的范围，之后会在《C++11之多线程(五、内存序(Memory Order)）》中较为详细的介绍，这里仅做简单介绍。
简单的讲就是程序的内存访问顺序在编译期(代码优化)和运行期(CPU的乱序执行)可能会被重新排序，这种重新排序往往是为了提高程序的执行速度，并且在单处理器的情况下这种重新排序不会对程序的正确性产生影响，但在多处理器的多线程环境中就可能得到非预期的结果。memory order就是用来指定原子操作周围(指前后)的非原子操作的内存访问如何被排序和同步。std::memory_order_seq_cst是标准库中原子操作内存序的默认值，这是最为严格的memory order对性能有一定的损害，若不苛求性能可以总是使用这个值。
简单介绍完了memory order再回到compare_exchange_strong的函数声明中来。
在重载1和2中success参数用于指定当比较成功(*this与expected值相等时)执行read-modify-write操作的内存序，failure参数用于指定当比较失败时执行load操作的内存序。
在重载3和4中order参数同时指定sccuess和failure时的内存序。
下面两种调用方式是等价的。

1
2
flag.compare_exchange_strong(expected, desired, order);
flag.compare_exchange_strong(expected, desired, order, order);

对于unlock函数，可以使用std::atomic类模板的成员函数store来以原子操作的方式将flag置false。

1
2
void store(T desired, memory_order = std::memory_order_seq_cst);
void store(T desired, memory_order = std::memory_order_seq_cst) volatile;

使用默认内存序的自旋锁完整的实现如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class spin_mutex {
  std::atomic<bool> flag = ATOMIC_VAR_INIT(false);
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock() {
    bool expected = false;
    while(!flag.compare_exchange_strong(expected, true))
      expected = false;
  }
  void unlock() {
    flag.store(false);
  }
};

下面的代码演示了使用自旋锁保护变量num

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
int num = 0;
spin_mutex sm;

void thread_proc()
{

  for(int i = 0; i < 100000; ++i) {
    sm.lock();
    ++num;
    sm.unlock();
  }
}

int main()
{
  std::thread td1(thread_proc), td2(thread_proc);
  td1.join();
  td2.join();
  std::cout << num << std::endl;
  return 0;
}

指定内存序提高性能

通过指定内存序代替默认的std::memory_order_seq_cst可以提高性能。C++11的std::memory_order有6个枚举值。

Value 描述
std::memory_order_relaxed 没有任何同步和排序限制，只需保证操作是原子的。
std::memory_order_consume 一个指定了此值的load操作在受影响的内存位置上执行consume操作，此操作使得另一个在同一内存位置执行了release操作的线程在此之前对数据依赖(data-dependent)的内存位置的写操作为当前线程可见。
std::memory_order_acquire 一个指定了此值的load操作在受影响的内存位置上执行acquire操作，此操作使得另一个在同一内存位置执行了release操作的线程在此之前对任意的内存位置的写操作为当前线程可见。
std::memory_order_release 一个指定了此值的store操作在受影响的内存位置上执行release操作，此操作使得此线程之前对数据依赖(data-dependent)的内存位置的写操作为另一个在此之后通过对同一内存位置执行consume操作的线程可见，使得此线程之前对任意的内存位置的写操作为另一个在此之后通过对同一个内存位置执行acquire操作的线程可见。
std::memory_order_acq_rel 一个指定了此值的read-modify-write操作，在读阶段(相当于load)对受影响的内存位置执行acquire操作，在写阶段(相当于store)对同一内存位置执行release操作。
std::memory_order_seq_cst 顺序一致性，所有的线程观察到的整个程序中内存修改顺序是一致的。

在指定memory order前需要明确自旋锁的责任:自旋锁除了要避免多线程重入，还要保证一个线程在持有自旋锁期间对内存的写操作要能够被另一个线程在获得自旋锁的时候可观测到(可见)。
对照前面memory order枚举值的表，优先考虑性能。可以在unlock中对flag调用store置false的执行release操作。而在lock中对flag调用compare_exchange_strong的时候执行acquire操作。
使用Rlease-Acquire内存序的自旋锁的完整实现如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class spin_mutex {
  std::atomic<bool> flag = ATOMIC_VAR_INIT(false);
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock() {
    bool expected = false;
    while(!flag.compare_exchange_strong(expected, true, std::memory_order_acquire))
      expected = false;
  }
  void unlock() {
    flag.store(false, std::memory_order_release);
  }
};

其中flag.compare_exchange_strong(expected, true, std::memory_order_acquire)等价于flag.compare_exchange_strong(expected, true, std::memory_order_acquire, std::memory_order_acquire)。获得自旋锁的所有权的充分必要条件是当且仅当compare_exchange_strong比较成功且成功改变flag的值为true，也就是说当比较失败时是没有必要执行acquire操作的。
于是lock函数可以再修改成

1
2
3
4
5
void lock() {
  bool expected = false;
  while(!flag.compare_exchange_strong(expected, true, std::memory_order_acquire, std::memory_order_relaxed))
    expected = false;
}

不使用CAS(compare-and-swap)的实现

前面的CAS实现中lock函数中的原子操作需要3个步骤:

  1. 取flag的值；
  2. 比较flag的值；
  3. 置flag一个新值。

由于自旋锁只有两个状态，事实上获取自旋锁可以使用下面的流程


其中灰色矩形中的2步应是一个不可分割的原子操作:

  1. 取flag的值;
  2. 置flag一个新值。

这样就将原子操作中需要做的3步减少到2步了。
std::atomic类模板的exchange成员函数能够以原子的方式对其进行赋值并返回旧值。

1
2
T exchange(T desired, memory_order = std::memory_order_seq_cst);
T exchange(T desired, memory_order = std::memory_order_seq_cst) volatile;

使用exchange实现的自旋锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class spin_mutex {
  std::atomic<bool> flag = ATOMIC_VAR_INIT(false);
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock() {
    while(flag.exchange(true, std::memory_order_acquire))
      ;
  }
  void unlock() {
    flag.store(false, std::memory_order_release);
  }
};

使用std::atomic_flag的实现

C++11并不要求std::atomic的实现必须是无锁的(lock-free)，可以通过使用std::atomic类模板的成员函数is_lock_free来检查atomic对象是不是无锁的。如果自旋锁内部的flag不是无锁的类型那么这个自旋锁就没有存在的意义了。所幸C++11提供了一个无锁的二值(bool)原子类型std::atomic_flag。使用std::atomic_flag就可以实现一个真正有用的自旋锁了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class spin_mutex {
  std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock() {
    while(flag.test_and_set(std::memory_order_acquire))
      ;
  }
  void unlock() {
    flag.clear(std::memory_order_release);
  }
};

与标准库的std::mutex一样这里实现的spin_mutex同样不建议直接去调用lock和unlock函数，而推荐使用std::lock_guard来自动管理自旋锁。完整的实现和使用代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
#include <atomic>
#include <thread>
#include <mutex>
#include <iostream>

class spin_mutex {
  std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock() {
    while(flag.test_and_set(std::memory_order_acquire))
      ;
  }
  void unlock() {
    flag.clear(std::memory_order_release);
  }
};

int num = 0;
spin_mutex sm;

void thread_proc()
{

  for(int i = 0; i < 100000; ++i) {
    std::lock_guard<spin_mutex> lock(sm);
    ++num;
  }
}

int main()
{
  std::thread td1(thread_proc), td2(thread_proc);
  td1.join();
  td2.join();
  std::cout << num << std::endl;
  return 0;
}
文章目录
  1. 1. 自旋锁(Spinlock)
  2. 2. 自旋锁的原理
  3. 3. 使用C++11的原子操作实现自旋锁
    1. 3.1. 指定内存序提高性能
    2. 3.2. 不使用CAS(compare-and-swap)的实现
    3. 3.3. 使用std::atomic_flag的实现