如何设计一个高性能可扩展的自旋锁

 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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

https://github.com/facebook/folly/blob/main/folly/SpinLock.h
SpinLock--->folly::MicroSpinLock 

/*
 * folly::MicroSpinLock
 * ====================
 * A really, *really* small spinlock for fine-grained locking of lots
 * of teeny-tiny data. 
 * 
 * Designed to be POD (Plain Old Data) so it can be
 * packed into other structures without extra overhead.
 *
 * 特性：
 *  1. 极小自旋锁，适用于超细粒度场景
 *  2. POD 类型，可零初始化，相当于调用 init()
 *  3. 用户态实现，无系统调用开销
 *  4. 支持动态检测工具插桩（ThreadSanitizer 等）
 */

struct MicroSpinLock {
  // FREE 表示锁可用，LOCKED 表示锁已被持有
  enum { FREE = 0, LOCKED = 1 };
  //
  // 用一个字节存储锁状态，避免使用 std::atomic<>，保持 POD 特性
  //为什么用uint8_t表示 std::atomic<>？ 
  //这个和序列化有什么关系？怎么实现的
  typedef unsigned char           uint8_t;
  uint8_t lock_; //locked? 
  

  /**
   * 尝试获取锁，非阻塞
   * @return true: 获得锁；false: 未获得
   */
  bool try_lock() noexcept {
    // 调用 xchg 将新值置为 LOCKED，返回旧值
    bool ret = (xchg(LOCKED) == FREE);
    // 插桩：记录尝试获取锁的结果，便于动态分析
    annotate_rwlock_try_acquired(
        this, annotate_rwlock_level::wrlock, ret, __FILE__, __LINE__);
    return ret;
  }

  /**
   * 获取锁，阻塞自旋
   * 在持锁竞争时，会进行退避等待以降低总线抖动
   */
  void lock() noexcept {
    detail::Sleeper sleeper;
    // 自旋尝试交换，如果旧值不为 FREE，表示竞争失败
    while (xchg(LOCKED) != FREE) {
      // 进入退避等待循环
      do {
        sleeper.wait();  // 调用 pause/yield 或指数退避
      } while (payload()->load(std::memory_order_relaxed) == LOCKED);
    }
    // 加锁成功，断言当前状态为 LOCKED
    assert(payload()->load() == LOCKED);
    // 插桩：记录获取锁的事件
    annotate_rwlock_acquired(
        this, annotate_rwlock_level::wrlock, __FILE__, __LINE__);
  }

  /**
   * 释放锁
   * 将状态设为 FREE，并使用 release 语义保证可见性
   */
  void unlock() noexcept {
    // 断言当前持锁状态
    assert(payload()->load() == LOCKED);
    // 释放锁  write
    payload()->store(FREE, std::memory_order_release);
  }

 private:
  /**
   * 内部：获取指向 lock_ 字段的原子指针
   */
  std::atomic<uint8_t>* payload() noexcept {
    // reinterpret_cast 到 std::atomic<uint8_t>* 类型
    return reinterpret_cast<std::atomic<uint8_t>*>(&this->lock_);
  }

  /**
   * 原子交换操作，将 lock_ 原子地置为 newVal，返回旧值
   */
  uint8_t xchg(uint8_t newVal) noexcept {
    return std::atomic_exchange_explicit(
        payload(), newVal, std::memory_order_acq_rel);
  } //A read-modify-write operation with this memory order is both an _acquire operation_ and a _release operation_
};

 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

参考：https://mfukar.github.io/2017/09/08/ticketspinlock.htm
  now_serving   next_ticket
        |             |
        V             V
... 0 1 2 3 4 5 6 7 0 1 2 ...
         \___________/
           8 threads
           holding one ticket each

struct TicketSpinLock {
    /**
     * Attempt to grab the lock:
     * 1. Get a ticket number
     * 2. Wait for it
     */
    void enter() {
        /* We don't care about a specific ordering with other threads,
         * as long as the increment of the `next_ticket` counter happens atomically.
         * Therefore, std::memory_order_relaxed.
         */
        const auto ticket = next_ticket.fetch_add(1, std::memory_order_relaxed);

        while (now_serving.load(std::memory_order_acquire) != ticket) {
            spin_wait();
        } //没等到我 我就是wait
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

int ObQSyncLock::rdlock()

{
	int ret = common::OB_SUCCESS;
	do {
		if (common::OB_EAGAIN == (ret = try_rdlock())) {
				sched_yield();
			//sched_yield()会主动放弃当前CPU给其他进程使用；
			//但是如果当前CPU上无其他进程等待执行，则直接返回继续执行当前进程
		   // 退避策略：使用sched_yield()让出CPU，而不是无限自旋
		}
	} while (common::OB_EAGAIN == ret); //读锁加锁失败，重试
   return ret;

}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

int ObQSyncLock::try_rdlock()
{
     int ret = OB_SUCCESS;
     if (OB_UNLIKELY(0 != ATOMIC_LOAD(&write_flag_))) {
       ret = OB_EAGAIN;
       //快速检查写标志：首先使用ATOMIC_LOAD原子操作检查write_flag_是否为0，
       //如果不为0表示有写锁存在，直接返回OB_EAGAI
     } else {
       // 找到线程对应的引用槽位并增加计数
       const int64_t idx = qsync_.acquire_ref();
       if (OB_UNLIKELY(0 != ATOMIC_LOAD(&write_flag_))) {
         qsync_.release_ref(idx);
         ret = OB_EAGAIN;
         // 第二次检查写锁状态（双重检查）
       } else {
         // success, do nothing
       }
     }
     return ret;
}

1
2
3

#define ATOMIC_LOAD(x) __atomic_load_n((x), __ATOMIC_SEQ_CST)
#define ATOMIC_LOAD_ACQ(x) __atomic_load_n((x), __ATOMIC_ACQUIRE)
#define ATOMIC_LOAD_RLX(x) __atomic_load_n((x), __ATOMIC_RELAXED