读写自旋锁详解，第 2 部分 基于简单共享变量的实现（1）

look_w

读者优先的读写自旋锁我们先不考虑性能，搞出一个可用的实现再说。首先，用一个整型变量 status 来记录当前状态；另一个整型变量 nr_readers 来记录同时持有锁的读者数量，只有当 nr_readers 为 0 的时候，锁才被读者彻底释放。此外不需要额外变量。
其次，我们使用高级互斥原语－普通的自旋锁，决定线程的执行顺序。读写自旋锁居然在内部使用普通自旋锁，这看起来有点古怪，还能够提高读者的并发性么？
我们需要留心的是，从合适状态出现到取得自旋锁之间可能发生状态转换，所以取得自旋锁之后还需检查一下当前状态。
清单 1. 基于自旋锁的读者优先实现

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

#define STATUS_AVAILABE 0 #define STATUS_READER 1 #define STATUS_WRITER 2 typedef struct {     volatile int status;     volatile int nr_readers;     spinlock_t sl; } rwlock_t; void init_lock(rwlock_t *lock) {     lock->status = STATUS_AVAILABE;     lock->nr_readers = 0;     spin_lock_init(&lock->sl); } void reader_lock(rwlock_t *lock) {     while (TRUE) {         while (lock->status == STATUS_WRITER)             cpu_relax();         spin_lock(&lock->sl);         if (lock->status != STATUS_WRITER) {             if (lock->status == STATUS_AVAILABE)                 lock->status = STATUS_READER;             lock->nr_readers++;             spin_unlock(&lock->sl);             return;         }         spin_unlock(&lock->sl);     } } void reader_unlock(rwlock_t *lock) {     spin_lock(&lock->sl);     if (--lock->nr_readers == 0)         lock->status = STATUS_AVAILABE;     spin_unlock(&lock->sl); } void writer_lock(rwlock_t *lock) {     while (TRUE) {         while (lock->status != STATUS_AVAILABE)             cpu_relax();         spin_lock(&lock->sl);         if (lock->status == STATUS_ AVAILABE) {             lock->status = STATUS_WRITER;             spin_unlock(&lock->sl);             return;         }         spin_unlock(&lock->sl);     } } void writer_unlock(rwlock_t *lock) {     spin_lock(&lock->sl);     lock->status = STATUS_AVAILABE;              //(a)     spin_unlock(&lock->sl); }

如果底层体系结构能原子地执行代码 (a)，那么无需先取得内部的自旋锁。上述实现内部使用了一把大锁，故而正确性容易得到保证，下面我们分析一下不足之处：

• 锁数据结构用到的 3 个域：lock，status 和 nr_readers，即使它们可以放到同一缓存行（Cache Line）中，多条非连续的写指令也可能带来较多的缓存无效化开销。
• 如果线程访问共享资源的操作相对短小，那么锁自身的开销会比较大。此时对于读者而言，其并发性基本被内部的自旋锁限制。
• 至少 reader_unlock() 要先获得内部的自旋锁，所以无法保证在较短的时间内结束（理论上可能永远无法结束），导致整体吞吐量降低。

上述代码的主要问题是变量和生成的指令太多，无法高效地执行。一个很自然的改进想法是把多个变量合并成单一变量，这样就有可能用一条原子指令更新状态 [3]，而无需使用高级同步原语。同时也能使得锁的释放操作在有限步骤内完成，于是保证获得锁的线程必然在有限时间内将锁释放掉（后文列出的代码均满足这一特性）。我们观察到：

• status 只需要 2 个可能值，因为 nr_readers 大于 0 即可表示锁被读者持有，因此 status 用一个 bit 即可。
• status 和 nr_readers 可以合并成一个变量。
• 锁被写者持有时，nr_readers 也可以用于记录等待的读者数目。

基于上述 3 点，我们将锁的数据结构简化为一个整型成员 rdr_cnt_and_flag。rdr_cnt_and_flag 最低位代表 status，其余位代表 nr_readers（当然也可以用最高位代表 status）：

• rdr_cnt_and_flag 等于 0，表示锁无人持有。
• rdr_cnt_and_flag 大于 0 且最低位为 0，表示有 rdr_cnt_and_flag / 2 个读者同时持有锁。
• rdr_cnt_and_flag 等于 1，表示锁为写者持有且无等待读者。
• rdr_cnt_and_flag 大于 0 且最低位为 1，表示锁为写者持有且有 (rdr_cnt_and_flag - 1) / 2 个等待读者。

清单 2. 基于简单共享变量的读者优先实现

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

#define WAFLAG 1 #define RC_INCR 2 typedef struct {     atomic_t rdr_cnt_and_flag; } rwlock_t; void init_lock(rwlock_t *lock) {     lock->rdr_cnt_and_flag = ATOMIC_INIT(0);                 //(a) } void reader_lock(rwlock_t *lock) {     atomic_add(RC_INCR, &lock->rdr_cnt_and_flag);            //(b)     while ((atomic_read(&lock->rdr_cnt_and_flag) & WAFLAG) != 0)     //(c)         cpu_relax(); } void reader_unlock(rwlock_t *lock) {     atomic_sub(RC_INCR, &lock->rdr_cnt_and_flag);            //(d) } void writer_lock(rwlock_t *lock) {     while (atomic_cmpxchg(&lock->rdr_cnt_and_flag, 0, WAFLAG) != 0)      //(e)         while (atomic_read(&lock->rdr_cnt_and_flag) != 0)        //(f)             cpu_relax(); } void writer_unlock(rwlock_t *lock) {     atomic_dec(&lock->rdr_cnt_and_flag);                     //(g) }

这个实现明显优于前者，每个函数的原子指令数和总指令数都非常少。reader_lock() 只执行一条原子加法指令，系统开销相当之小，而且不必像某些实现那样在尝试失败的情况下需要执行一个回滚操作（例如 Linux 内核实现的读写自旋锁）。
我们给出代码正确性的简要证明：

• 互斥。我们以 reader_lock() 和 writer_lock() 成功前最后一条原子操作（atomic_read(&lock->rdr_cnt_and_flag) 和 atomic_cmpxchg(&lock->rdr_cnt_and_flag, 0, 1)）的执行顺序作为锁的获得顺序。假定在时刻 t，有 A1，A2，…，An这 n（> = 1） 个线程同时持有锁，满足：在 A1之前获得锁的线程此时都已释放锁；对于 1 <= k <= (n – 1)，Ak在 A(k+1)前获得锁。可能存在线程 B，B 在某个 Ai和 A(i+1)之间获得锁，但在时刻 t，B 已经释放锁。分情况讨论：
  • A1是读者。如果 A2– An中有写者，假定 Am是第一个写者，由于锁释放后线程对 rdr_cnt_and_flag 的总贡献为 0，Am执行到代码 (e) 时因为尚有读者持有锁，rdr_cnt_and_flag 必定大于 0 且最低位是 0，因此 Am无法跳出该处的循环。可见 A2– An中没有写者。
  • A1是写者。A1获得锁的前提条件是 rdr_cnt_and_flag 等于 0，也就是说在 A1之前获得锁的线程已经释放了锁。如果 n > 1，不论 A2角色如何，A2都无法通过代码 (c) 或 (f)，因为此时 rdr_cnt_and_flag 必然等于 1。可见 n 只能等于 1。

综上可知，读者和写者不可能同时持有锁，任何时刻至多只有一个写者持有锁。

• 读者并发。从代码 (b) - (c) 可看出，只要没有写者持有锁，多个读者都能结束循环，从而获得锁。
• 无死锁。从锁申请角度来证明，假定申请线程为 A，分情况讨论：
  • A 是读者，如果在代码 (c) 处循环，这说明某个写者持有锁，但是写者必然在有限时间内将锁释放，届时 rdr_cnt_and_flag 的最低位必然为 0，那么 A 将立即获得锁。
  • A 是写者，如果无法获得锁，说明 A 在代码 (e) 或 (f) 处循环。这表明总是有读者或写者持有锁，但是持有者迟早得释放锁，只能说明某个或某些线程必定无限次地获得锁。
• 读者优先。新来的读者并不检查是否有等待的写者，当读者持有锁时，显然能够通过 (c) 的条件，马上获得锁；锁被写者持有或未被持有时，新来的读者通过代码 (b) 实现了“加塞”，能够抢占先来的等待写者；

这个实现的不足之处有 4 点：

• 读者在任何情况下都能“加塞”到等待写者之前，如果读者持续到来，写者很难有机会获得锁。
• 如果几乎“同时”申请锁的读者的数目要远多于写者，执行完代码 (b) 后读者获得锁的概率总是比较大，那么随后检查共享变量 rdr_cnt_and_flag 的代价将比较大，因为该变量被连续改写，需等到缓存更新后才能取到最新值。
• 如果在写者持有锁期间，读者持续到来，那么 rdr_cnt_and_flag 会被不断修改，增加读者间的缓存同步开销。
• 读者必须执行额外的逻辑与操作才能知道是否有写者持有锁。

如果我们用 rdr_cnt_and_flag 的最高位表示 status，其余位代表 nr_readers，那么有写者持有锁时，rdr_cnt_and_flag 必然是个负数（因为不可能同时有 0x8000000 个或更多的读者“同时”申请锁），检查起来比较快捷，于是上述 2、4 不足之处可以得到改进。代码如下：
清单 3. 基于简单共享变量的读者优先实现 2

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

#define WAFLAG (int)0x80000000 #define RC_INCR 1 typedef struct {     atomic_t rdr_cnt_and_flag; } rwlock_t; void init_lock(rwlock_t *lock) {     lock->rdr_cnt_and_flag = ATOMIC_INIT(0);                 //(a) } static inline int my_atomic_inc_negative(atomic_t *v) {     unsigned char c;     asm volatile(LOCK_PREFIX "incl %0; sets %1"             : "+m" (v->counter), "=qm" (c)             : : "memory");     return c; } void reader_lock(rwlock_t *lock) {     int sign = my_atomic_inc_negative(&lock->rdr_cnt_and_flag);          //(b)     if (sign)                               //(c)         while (atomic_read(&lock->rdr_cnt_and_flag) < 0)          //(d)             cpu_relax(); } void reader_unlock(rwlock_t *lock) {     atomic_dec(&lock->rdr_cnt_and_flag);                     //(e) } void writer_lock(rwlock_t *lock) {     while (atomic_cmpxchg(&lock->rdr_cnt_and_flag, 0, WAFLAG) != 0)      //(f)         while (atomic_read(&lock->rdr_cnt_and_flag) != 0)        //(g)             cpu_relax(); } void writer_unlock(rwlock_t *lock) {     atomic_add(-WAFLAG, &lock->rdr_cnt_and_flag);            //(h) }

因为 nr_readers 在低 31 位，读者到来时使用原子递增指令即可，比原子加法指令要快。执行完毕后我们可以先观察一下 EFLAGS 或 RFLAGS 寄存器的 SF 位，如果为 0，说明 rdr_cnt_and_flag 的新值是非负数（只能是正数，参见前面的描述），即说明没有写者持有锁，这比再次检查 rdr_cnt_and_flag 更高效。我们把这 2 个操作合并在一个 my_atomic_inc_negative 内联函数中，用汇编指令实现。