folly SpinLock源码分析

概述

SpinLock是对自旋锁的简单实现，因为SpinLock类仅仅是对MicroSpinLock的简单包装，所以本文主要讲述MicroSpinLock的实现。

后者的实现文件是folly/synchronization/MicroSpinLock.h。

设计-MicroSpinLock

唯一的数据成员就是一个uint8_t，它被作为一个值或为FREE或为LOCKED的原子变量使用：

struct MicroSpinLock { 
    enum { 
    FREE = 0, LOCKED = 1 }; // lock_ can't be std::atomic<> to preserve POD-ness. uint8_t lock_; // Initialize this MSL. It is unnecessary to call this if you // zero-initialize the MicroSpinLock. void init() noexcept { 
    payload()->store(FREE); } private: std::atomic<uint8_t>* payload() noexcept { 
    return reinterpret_cast<std::atomic<uint8_t>*>(&this->lock_); } bool cas(uint8_t compare, uint8_t newVal) noexcept { 
    return std::atomic_compare_exchange_strong_explicit( payload(), &compare, newVal, std::memory_order_acquire, std::memory_order_relaxed); } }; 

lock和unlock的实现：

struct MicroSpinLock { 
    // ... bool try_lock() noexcept { 
    bool ret = cas(FREE, LOCKED); annotate_rwlock_try_acquired( this, annotate_rwlock_level::wrlock, ret, __FILE__, __LINE__); return ret; } void lock() noexcept { 
    detail::Sleeper sleeper; while (!cas(FREE, LOCKED)) { 
    do { 
    sleeper.wait(); } while (payload()->load(std::memory_order_relaxed) == LOCKED); } assert(payload()->load() == LOCKED); annotate_rwlock_acquired( this, annotate_rwlock_level::wrlock, __FILE__, __LINE__); } void unlock() noexcept { 
    assert(payload()->load() == LOCKED); annotate_rwlock_released( this, annotate_rwlock_level::wrlock, __FILE__, __LINE__); payload()->store(FREE, std::memory_order_release); } // ... } 

可以看出来，MicroSpinLock的实现与预期一致，就是采用cas指令来切换锁状态。

• lock()，在忙等待中调用cas(FREE, LOCKED)指令直到加锁成功。
• unlock()，只能在LOCKED状态下被调用，直接将lock_置为FREE。

重点分析以下几点：

sleeper.wait()

实现如下：

void wait() noexcept { 
    if (spinCount < kMaxActiveSpin) { 
    // static constexpr uint32_t kMaxActiveSpin = 4000 ++spinCount; asm_volatile_pause(); // x86下实现同asm volatile("pause"); } else { 
    /* sleep override */ std::this_thread::sleep_for(delta); // delta的默认值为500微妙 } } 

为了避免lock()函数中while循环消耗大量CPU资源，folly采取了一种“巧妙”的算法：

1. 循环次数未达到4000次时，使用asm(“pause”)减缓CPU速度。
2. 超过4000次后，每次循环都要sleep 500微妙。

cas指令

这里folly直接使用了C++11出品的std::atomic_compare_exchange_strong_explicit()方法，而不用编写针对不同平台的汇编代码，这也显示了一个有着全面功能的标准库的好处。

其原型如下：

template< class T > bool atomic_compare_exchange_strong_explicit( std::atomic<T>* obj, typename std::atomic<T>::value_type* expected, typename std::atomic<T>::value_type desired, std::memory_order succ, std::memory_order fail ) noexcept; 

据cppreference描述，这个函数将会原子地比较obj和expected，如果相等，就用desired的值替换obj；不相等则把obj原来的值赋给expected。参数succ和fail则描述了替换成功发生和未成功发生时使用的memory_order，它们一般使用std::memory_order_acquire和std::memory_order_required。

所以cas(compare, newVal)函数中：

std::atomic_compare_exchange_strong_explicit( payload(), &compare, newVal, std::memory_order_acquire, std::memory_order_relaxed); 

就是比较compare和原子变量lock_，如果相等则设置lock_ = newVal，不相等则do nothing。

因为std::atomic_compare_exchange_strong_explicit()在成功替换时返回true，失败时返回false，这和cas()函数一致。

而且cas()函数的返回值false = 0、 true = 1和FREE = 0、LOCKED = 1保持一致，这种统一使得相关的lock代码非常优雅。

can’t be atomic_int?

为什么MicroSpinLock不直接使用std::atomic

lock_
而是uint8_t lock_ + payload()呢？

其实答案正如注释中所说lock_ can't be std::atomic<> to preserve POD-ness，如果使用了std::atomic字段，那么MIcroSpinLock将不再是一个POD，这可能导致某些优化（例如对POD对象数组以memcpy代替逐个拷贝构造）无法进行。

当然，这种操作的前提是std::atomic
的实现本身就只有一个uint8_t数据成员（而没有其他非POD数据成员）。

这也为我们提供了一种将非POD类型转换为POD类型的范例。

设计-SpinLockArray

紧随MicroSpinLock类之后，folly又提供了一个SpinLockArray类，后者可以提供一个消除了false sharing的spinlocks数组。

伪共享懂得都懂https://en.wikipedia.org/wiki/False_sharing。

SpinLockArray的实现非常简单：

template <class T, size_t N> struct alignas(max_align_v) SpinLockArray { 
    T& operator[](size_t i) noexcept { 
    return data_[i].lock; } const T& operator[](size_t i) const noexcept { 
    return data_[i].lock; } constexpr size_t size() const noexcept { 
    return N; } private: struct PaddedSpinLock { 
    PaddedSpinLock() : lock() { 
   } T lock; char padding[hardware_destructive_interference_size - sizeof(T)]; }; static_assert( sizeof(PaddedSpinLock) == hardware_destructive_interference_size, "Invalid size of PaddedSpinLock"); // Check if T can theoretically cross a cache line. static_assert( max_align_v > 0 && hardware_destructive_interference_size % max_align_v == 0 && sizeof(T) <= max_align_v, "T can cross cache line boundaries"); char padding_[hardware_destructive_interference_size]; std::array<PaddedSpinLock, N> data_; }; 

大致就是一个普通的std::array

，只是对MicroSpinLock进行了padding化包装，成为PaddedSpinLock类。

struct PaddedSpinLock { 
    PaddedSpinLock() : lock() { 
   } T lock; char padding[hardware_destructive_interference_size - sizeof(T)]; }; 

这个类就是消除false sharing问题的关键。

其中：

• struct alignas(max_align_v) SpinLockArray，要求整个数组类根据max_align_v对齐，后者的解释见下文。
• hardware_destructive_interference_size，即 Minimum offset between two objects to avoid false sharing. Guaranteed to be at least alignof([std::max_align_t](dfile:///Users/peter/Library/Application Support/Dash/DocSets/C++/C++.docset/Contents/Resources/Documents/en.cppreference.com/w/cpp/types/max_align_t.html))，可以理解为一个cache line的大小。
• char padding_[hardware_destructive_interference_size]，整个数组最前的padding，避免数组和之前的数据产生false sharing。

所以简单地添加填充字节，使得每一个SpinLock独享一个cache line的空间（典型值为64字节），就可以消除false sharing问题。

什么是max_align_v呢？这是folly中自定义的一个值：

constexpr std::size_t max_align_v = detail::max_align_v_::value(); struct alignas(max_align_v) max_align_t { 
   }; template <typename... Ts> struct max_align_t_ { 
    // 此函数将返回下面的类型列表max_align_v_（long double, double...）中的sizeof的最大值 static constexpr std::size_t value() { 
    std::size_t const values[] = { 
   0u, alignof(Ts)...}; std::size_t r = 0u; for (auto const v : values) { 
    r = r < v ? v : r; } return r; } }; using max_align_v_ = max_align_t_< long double, double, float, long long int, long int, int, short int, bool, char, char16_t, char32_t, wchar_t, void*, std::max_align_t>; } // namespace detail 

max_align_v就是max_align_t的alignment（即对齐量，此值一般是sizeof的因数），而max_align_t是当前平台的最大默认内存对齐类型，所以可以定义 struct alignas(max_align_v) max_align_t {};

其实标准C++都有相应的std::max_align_t，folly这里强化实现了它，通过对上述代码的观察也可看出max_align_v的值是如何算出的：

通过value()函数，依次取常见的基本类型的alignment和std::max_align_t，取其中的最大值就是max_align_v。

有关C++11定义的alignof、alignas、std::align、std::max_align_t等，可以参考我的另一篇文章《内存对齐和alignof》。