读写自旋锁详解，第 2 部分

 引用：http://www.ibm.com/developerworks/cn/linux/l-cn-rwspinlock2/?ca=drs-

 

 

读者优先的读写自旋锁

我们先不考虑性能，搞出一个可用的实现再说。首先，用一个整型变量 status 来记录当前状态；另一个整型变量 nr_readers 来记录同时持有锁的读者数量，只有当 nr_readers 为 0 的时候，锁才被读者彻底释放。此外不需要额外变量。

其次，我们使用高级互斥原语－普通的自旋锁，决定线程的执行顺序。读写自旋锁居然在内部使用普通自旋锁，这看起来有点古怪，还能够提高读者的并发性么？

我们需要留心的是，从合适状态出现到取得自旋锁之间可能发生状态转换，所以取得自旋锁之后还需检查一下当前状态。

清单 1. 基于自旋锁的读者优先实现

				
 #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)，那么无需先取得内部的自旋锁。上述实现内部使用了一把大锁，故而正确性容易得到保证，下面我们分析一下不足之处：

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

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

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

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

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

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

				
 #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 内核实现的读写自旋锁）。

我们给出代码正确性的简要证明：

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

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

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

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

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

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

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

				
 #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 内联函数中，用汇编指令实现。

回页首

写者优先读写自旋锁

我们在清单 2 的代码基础上实现写者优先的读写自旋锁，关键之处是用号码分配的方式确定写者的到来顺序 [3]。使用 2 个整型变量，一个用于存放下一个分配的写者号码 writer_requests，一个用于存放下一个允许执行的写者号码 writer_completions。初始化的时候，将二者均置为 0。

有兴趣的读者朋友也可以在清单 3 的代码基础上实现写者优先的读写自旋锁。

写者到来的时候以当前的 writer_requests 值作为自己的号码 id，并原子地递增 writer_requests。当 id == writer_completions 时，表明先来的写者已经全部离开，但是可能有读者持有锁，因此写者还得检查持有锁的读者数目 nr_readers 是否为 0。写者释放锁的时候，增加 writer_completions，通知下一个等待写者。对写者 W 而言，所谓在 W 后面到来的申请线程是指在 W 获得号码之后开始执行 reader_lock() 和 writer_lock() 的线程。

对读者而言，当 writer_requests == writer_completions 时，表明当前已经没有写者，即无写者持有锁，也无等待写者，但这并不表示读者可以马上获得锁，因为可能在准备尝试获取锁的时候又有新的写者到来并获得锁。

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

				
 #define WAFLAG 1 
 #define RC_INCR 2 

 typedef struct { 
	 atomic_t rdr_cnt_and_flag __cacheline_aligned_in_smp; 
	 atomic_t writer_requests __cacheline_aligned_in_smp; 
	 atomic_t writer_completions __cacheline_aligned_in_smp; 
 } rwlock_t; 

 void init_lock(rwlock_t *lock) 
 { 
	 lock->rdr_cnt_and_flag = ATOMIC_INIT(0); 
	 lock->writer_requests = ATOMIC_INIT(0); 
	 lock->writer_completions = ATOMIC_INIT(0); 
 } 

 void reader_lock(rwlock_t *lock) 
 { 
	 while (atomic_read(&lock->writer_completions) != 		 //(a) 
 atomic_read(&lock->writer_requests)) 
		 cpu_relax(); 
	 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) 
 { 
	 int id = atomic_inc_return(&lock->writer_requests); 		 //(e) 
	 while (atomic_read(&lock->writer_completions) != id) 		 //(f) 
		 cpu_relax(); 
	 while (atomic_cmpxchg(&lock->rdr_cnt_and_flag, 0, WAFLAG) != 0)//(g) 
		 while (atomic_read(&lock->rdr_cnt_and_flag) != 0) 
			 cpu_relax(); 
 } 

 void writer_unlock(rwlock_t *lock) 
 { 
	 atomic_dec(&lock->rdr_cnt_and_flag); 				  //(h) 
	 lock->writer_completions.counter++; 				  //(i) 
 } 

代码 (i) 不用写成原子递增操作 atomic_inc(&lock->writer_completions)，因为此时有且仅有一个线程对 writer_completions 赋值。atomic_t 结构中的 counter 域有 volatile 关键字修饰，所以 (i) 一旦执行完毕，其它处理器就能感知 writer_completions 的新值。代码 (h) 是个原子操作，隐含了一个内存屏障，所以 (h) 的执行效果必定发生在 (i) 之前。

如果写者的数目和到来的频率较大，那么 writer_requests 和 writer_completions 这 2 个共享变量也会被频繁修改，如果它们和 rdr_cnt_and_flag 被放置在同一缓存行中，将增加处理器间缓存的同步开销。解决这种伪共享（False Sharing）问题的一种简单优化方法是用 __cacheline_aligned_in_smp 宏将锁结构中的 3 个共享变量放置在不同的缓存行中。

从代码 (e) 可知，每个写者都会取得一个号码 id，id 从 0 开始，中间显然不会遗漏，申请的顺序就是取得 id 的顺序，也就是说 id 小的写者较早申请。我们可以假定 W_n是 id 为 n 的写者，不失一般性，可以认为每个写者都是不同的。我们以 reader_lock() 和 writer_lock() 成功前最后一条原子操作（atomic_read(&lock->rdr_cnt_and_flag) 和 atomic_cmpxchg(&lock->rdr_cnt_and_flag, 0, 1)）的执行顺序作为锁的获得顺序。

后面的证明还依赖这个事实：对于任意 n >= 0，W_n必定在有限时间内执行一次对 writer_completions 的递增操作，将其从 n 变为 (n + 1)，且当 n > 0 时，W_{(n – 1)}在 W_n完成该操作。证明请参阅下节关于公平读写自旋锁的论述。

我们给出代码正确性的简要证明：

1. 互斥。和读者优先的读写自旋锁的证明一样。
2. 读者并发。从代码 (b) - (c) 可看出，只要先到达的写者都已离开且当前没有写者持有锁，多个读者都能结束循环，从而获得锁。
3. 无死锁。从锁申请角度来证明，假定申请线程为 A，分情况讨论：
   1. A 是读者，通过了代码 (a) 处循环。这说明通过时已无写者，但当执行代码 (b) 前，可能有新的写者到来并获得锁，不过一旦其释放，A 必定能获得锁。
   2. A 是读者，无等待写者，锁被写者持有，导致无法通过代码 (a) 处循环。不过锁迟早被释放，因此 A 在有限时间内能通过 (a) 处循环，与上种情况相同。
   3. A 是读者，至少有一个等待写者，导致无法通过代码 (a) 处循环，等待写者可能是比 A 晚来的。假定号码最小的等待写者是 B，我们证明 B 或别的线程能获得锁，请参考 d。
   4. A 是写者，前面无等待写者。如果锁被写者持有，锁一旦释放后，A 就能跳出代码 (f) 处的循环。锁处于其它 2 种状态时，A 当然能通过 (f)。不过此时 A 不一定能通过代码 (g) 处的循环，因为可能已经有读者持有锁或者比 A 早到的等待读者抢先获得了锁。但是这些持有锁的读者数目是有限的，且必定在有限时间内释放锁，因此 A 一定能获得锁。
   5. A 是写者，前面有至少一个等待写者。假定号码最小的等待写者是 B，我们证明 B 或别的线程能获得锁，请参考 d。
4. 写者优先。一旦写者 W 完成代码 (e) 而获得分配的号码，那么在释放锁之前，writer_completions 必定小于 writer_requests，那么后到的读者不能退出代码 (a) 处的循环，可见 W 一定在后到的读者之前获得锁。同理，后到的写者会在代码 (f) 处不断循环，W 一定在后到的写者之前获得锁。

回页首

公平读写自旋锁

现实中，公平读写自旋锁可以保证当线程提交一个申请操作后，其等待时间有个可控的上界。这个优良特性，往往受到用户的青睐。

笔者在上节基础上实现了这个公平读写自旋锁。还是使用号码分配的方式为所有的读者 / 写者定序，但是这有个缺陷：线程无法知道直接后继的角色。如果读者 A 的直接后继 B 是个读者，那么 A 获得锁后，B 应该马上获得锁，而不是等到 A 释放锁之后。A 如何及时通知 B 呢？我们采用“事不关己，高高挂起”的态度，即读者获得锁后，立刻增加 completions 值，剩下的事丢给后继者自行处理。如果直接后继是写者咋办？写者此时还应判断是否仍有读者持有锁，即检查 nr_readers 是否大于 0。对写者而言，释放锁的时候才增加 completions 值。

清单 5. 基于简单共享变量的公平实现

				
 typedef struct { 
	 atomic_t requests __cacheline_aligned_in_smp; 
	 atomic_t completions __cacheline_aligned_in_smp; 
	 atomic_t nr_readers __cacheline_aligned_in_smp; 
 } rwlock_t; 

 void init_lock(rwlock_t *lock) 
 { 
	 lock->requests = ATOMIC_INIT(0); 
	 lock->completions = ATOMIC_INIT(0); 
	 lock->nr_readers = ATOMIC_INIT(0); 
 } 

 void reader_lock(rwlock_t *lock) 
 { 
	 int id = atomic_inc_return(&lock->requests); 			 //(a) 
	 while (atomic_read(&lock->completions) != id) 		 //(b) 
		 cpu_relax(); 
	 atomic_inc(&lock->nr_readers); 				 //(c) 
	 lock->completions.counter++; 					 //(d) 
 } 

 void reader_unlock(rwlock_t *lock) 
 { 
	 atomic_dec(&lock->nr_readers); 				 //(e) 
 } 

 void writer_lock(rwlock_t *lock) 
 { 
	 int id = atomic_inc_return(&lock->requests); 			 //(f) 
	 while (atomic_read(&lock->completions) != id) 		 //(g) 
		 cpu_relax(); 
	 while (atomic_read(&lock->nr_readers) > 0) 			 //(h) 
		 cpu_relax(); 
 } 

 void writer_unlock(rwlock_t *lock) 
 { 
	 lock->completions.counter++; 					 //(i) 
 } 

代码 (c) 必须用原子递增操作，因为此时可能有读者正在执行代码 (e) 释放锁。(d) 和 (i) 不用写成原子递增操作，因为此时有且仅有一个线程对 completions 赋值。由于所有的线程均在 completions 这个共享变量上自旋、写者还需在 nr_readers 上自旋且读者不断修改 nr_readers，因此处理器间的同步开销比较大，将这 3 个共享变量放置在不同的缓存行中可以提高性能。

从代码 (a) 和 (f) 可知，每个申请线程都会取得一个号码 id，id 从 0 开始，中间显然不会遗漏，申请的顺序就是取得 id 的顺序，也就是说 id 小的线程较早申请。我们可以假定 A_n是 id 为 n 的申请线程，不失一般性，可以认为每个申请线程都是不同的。

先证明：对于任意 n >= 0，A_n必定在有限时间内执行一次对 completions 的递增操作，将其从 n 变为 (n + 1)，且当 n > 0 时，A_{(n – 1)}在 A_n完成该操作。采用第二数学归纳法：

1. n 等于 0、1 或 2 时，显然成立。
2. 假设 n < k（k >= 2） 时成立。
3. 当 n = k 时，只有当 A_{(k – 1)}执行完对 completions 的递增操作后， completions 才变为 k。此时 A₁，…，A_{(k – 1)}已不能再对该变量进行操作，而 A_k之后的线程要么还没到来，要么在代码 (b) 或 (g) 处循环，只有 A_k才有可能跳出这两处的循环。如果 A_k是读者，那么它很快就可以对 completions 执行递增操作；如果 A_k是写者，它还得先等 nr_readers 为 0。nr_readers 必定在有限时间内变为 0，因为读者迟早要释放锁。然后 A_k在有限时间内执行 writer_lock() 时将 completions 递增。不论哪种情况，A_k必定在有限时间内执行一次对 completions 的递增操作，将其从 k 变为 (k + 1)。显然 A_k的执行在 A_{(k – 1)}之后。

因为线程对 completions 的递增操作完成时意味着已经获得了锁，所以上面这个命题告诉我们申请线程必定在有限时间内获得锁，且获得锁的顺序和取得 id 的顺序一致，即“无死锁”和“公平”。

对于“互斥”性，我们假定在时刻 t，有 A_I1，A_I2，…，A_In这 n（> = 1） 个线程同时持有锁，且在 A_I1之前获得锁的线程此时都已释放锁。因为申请线程依次获得锁，所以 I1 < I2 < … < In。可能存在线程 B，B 在某个 A_i和 A_(i+1)之间获得锁，但在时刻 t，B 已经释放锁。分情况讨论：

1. A_I1是读者。如果 A_I2– A_In中有写者，假定 A_Im是第一个写者，由于锁释放后线程对 nr_readers 的总贡献为 0（实际上写者根本不修改这个变量），A_Im执行到代码 (g) 时因为尚有读者持有锁，nr_readers 必定大于 0，因此 A_Im无法跳出该处的循环。可见 A_I2– A_In中没有写者。
2. A_I1是写者。因为写者持有锁的时候并不对 completions 执行递增操作，所以此时 A_I1之后的申请线程不可能获得锁，于是 n 只能等于 1。

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

一旦读者 A 获得锁，由代码 (d) 知它会递增 completions，如果 A 的直接后继 B 是个读者，那么 B 可以跳出代码 (b) 处的循环，从而获得锁。可见“读者并发”也是成立的。