Linu下写者优先的读写锁的设计

1、Linux 下写者优先的读写锁的设计在 IBM Bluemix 云平台上开发并部署您的下一个应用。现在就开始免费试用一、本文的目的在 linux 下有两种实现数据互斥的基本机制，包括了 semaphore（信号量） ,spinlock （自旋锁）。这里要说的读写锁 (read write lock) 是自旋锁的一个变种，与一般的自旋锁的区别是，自旋锁一次只能有一个进程进入临界区，而对读写锁而言，如果进程是读的话，那就可以有多个进程同时进入临界区，而如果是写的话，则只有一个可以。就现在的 linux 内核源代码的发行版本而言，已经实现了读写锁的一个类型，就是读者优先的读写锁。 （在这个设计中，作

2、为读的请求可以更容易的进入临界区，而写的请求的请求往往容易受阻，这个我在后面会分析） ，而我要设计的读写锁，则是以写进程为优先的考虑的对象，如果有写的请求发出，则它会在被允许的最快时间内得到响应。这样的好处是在一个由很多客户端以读的权限访问的服务器（如一般的公众服务器），如果管理员对服务器的某些内容或配置进行修改的话，那它的及时性就有可能无法满足。这有时是不可以被接受的。回页首二、 linux 现有的读写锁状况我先来分析现在linux 内核源代码中的读写锁的实现方式，这样就可以很容易的理解后面的写者优先的读写锁的设计。先介绍一个数据结构，这是在读写锁起到重要作用。(注：下面所有的内核源代码均来

3、自 linux 2.4.17 ，如果与你的现有的内核源代码不同，请你作一些相应的改变就可以了，原理部分没有变化)typedefstruct volatile unsigned int lock;#if SPINLOCK_DEBUGunsigned magic;#endif rwlock_t; 这里的 magic 是用于调试的，而 lock 就是允许可以加的读锁数。这个代码在 linux/include/asm-i386/spinlock.h中定义了 read_lock 和 write_lockstatic inline void read_lock(rwlock_t *rw)#if SPINL

4、OCK_DEBUGif (rw->magic != RWLOCK_MAGIC)BUG();#endif_build_read_lock(rw, "_read_lock_failed");static inline void write_lock(rwlock_t *rw)#if SPINLOCK_DEBUGif (rw->magic != RWLOCK_MAGIC)BUG();#endif_build_write_lock(rw, "_write_lock_failed"); 注意这里有两个参数，一个是 rw 就是允许的读锁数，而后面一个参数

5、是如果加锁失败的话，处理失败的函数。在这里真正调用的对 write_lock 是 _build_write_lock_const 或_build_write_lock_ptr ，对 read_lock 中调用的是_build_read_lock_const 或 _build_read_lock_ptr ，这里的取决因素是调用参数的操作指令寻址方式。我们这里只看const类#define _build_write_lock_const(rw, helper)asmvolatile(LOCK "subl $" RW_LOCK_BIAS_STR ",(%0)nt&quo

6、t;"jnz 2fn""1:n"".section .text.lock,"ax"n""2:tpushl %eaxnt""leal %0,%eaxnt""call "helper "nt""popl %eaxnt""jmp 1bn"".previous":"=m" (*(volatile int *)rw) : : "memory")这里

7、的 RW_LOCK_BIAS_STR 就是 "0x01000000" ，取这个值的原因是这个值足够大，可以使满足读的请求足够多。在".section .text.lock,"ax"n"".previous"中的内容是把这一段的代码汇编到一个叫.text.lock 的节中，并且这个节的属性是可重定位和可执行的，这样在代码的执行过程中，因为不同的节会被加载到不同的页面中，所以如果在前面不出现jmp ，就在 1:处结束了。而call 的是在前面的 write_lock 中所写入的 _write_lock_failed ，

8、这个函数在arch/asm-i386/kernel/semaphore.c 中定义.align.globl _write_lock_failed_write_lock_failed:" LOCK "addl$" RW_LOCK_BIAS_STR ",(%eax)1: rep; nopcmpl $" RW_LOCK_BIAS_STR ",(%eax) jne 1b" LOCK "subl$" RW_LOCK_BIAS_STR ",(%eax)jnz _write_lock_failedret 这里

9、的 LOCK 是一个在 SMP 体系结构中锁住总线， 不让其它的 CPU 访问内存。在这里先" LOCK "addl $"RW_LOCK_BIAS_STR ",(%eax)是为了防止在后面的自旋等待中，不会让后面的读请求受阻，要不然的话，就会出现死锁，这是致命的。而1: rep; nopcmpl$" RW_LOCK_BIAS_STR ",(%eax)jne 1b 就是不断的检查锁是否可以得到，得不到就会nop，这种方法可以在不改变lock 的值的情况下实现等待，这就不用 LOCK ，这样的锁住总线了。最后如果得到锁就" LO

10、CK"subl $" RW_LOCK_BIAS_STR ",(%eax) 这样就实现了write_lock的功能。对读锁也是类似#define_build_read_lock_const(rw, helper)asm volatile(LOCK"subl $1,%0nt""js 2fn""1:n"".section .text.lock,"ax"n""2:tpushl %eaxnt""leal %0,%eaxnt""

11、call " helper "nt""popl %eaxnt""jmp 1bn"".previous":"=m" (*(volatile int *)rw) : : "memory")但这里要注意一点， 就是对 read_lock 而言，只要减1 并且只要这个值不为负的话，就可以得到锁了，但rw.lock的值在被初始化的时候就被赋值成了0x01000000 ，这个值足够大， 而要减的很小，所以要获得读锁是很容易的，从理论上说比得到写锁要容易0x01000000 倍，

12、这就是我前面说现在在linux 内部实现的读写锁是读者优先的。而这个数也让我们容易理解在要获得写锁时，要对这个lock 值 减去 0x01000000，就是如果有一个读或者写请求在临界区内的话，第二个写请求就无法得到写锁了。而如果得不到读锁，所要跳的是在read_lock 所指明的 _read_lock_failed.align 4.globl _read_lock_failed_read_lock_failed:lock ; incl(%eax)1: rep; nopcmpl $1,(%eax) js 1block ; decl (%eax) js _read_lock_failedret

13、这里的道理与前面的_write_lock_failed中的道理是相似的。回页首三、写者优先的读写锁的实现那既然要实现以写者为优先的读写锁，很自然，我们就想到了在读的请求发生时，不先去试图获得读锁，而是去检查有没有写的请求正在等待，如果有写的请求正在等待，则读的请求必须先处于等待状态。让写的请求完成之后，发现已经没有写的请求在等待了，才去试图获得读的锁。这里我们先来设计rwlock_t这个数据结构，typedef struct volatile unsigned int lock;#if WLOCK_PRIORITYvolatile unsigned int wlock_waiting;#end

14、if#if SPINLOCK_DEBUGunsigned magic;#endif rwlock_t; 这里所增加的这个wlock_waiting就是作为检测是否有写的请求在等待的标志数。如果这个数不等于0 则说明已经有写的请求在等待。它的负值的大小决定了写请求等待的个数。这里我们先修改_build_write_lock_const中#define _build_write_lock_const(rw,wlock_wait,helper,helper1)asm volatile("cmpl $0,(%1)nt" "jnz 3fn" "1:t&q

15、uot; LOCK "subl $"RW_LOCK_BIAS_STR ",(%0)nt""jnz 4fn""2:n"".section .text.lock,"ax"n""3:tpushl %ebxnt" "leal %1,%ebxnt" "call " helper1"nt" "popl %ebxnt" "jmp 1bn" "4:tpushl %

16、eaxnt"pushl %ebxnt""leal %0,%eaxnt""leal %1,%ebxnt""call " helper "nt""popl %ebxnt" "popl %eaxnt""jmp 2bn"".previous":"=m" (*(volatile int *)rw) ,"=m"(*(volatile int *)wlock_waiting): : &quo

17、t;memory") 这里的新增加的 wlock_waiting 是表示前面所定义的 wlock_waiting 的地址，这个是地址，而不是值本身，它的原因是指令寻址的方式决定的，为了保证指令的操作在直接对内存，而不是把内存中的数据 load 到寄存器中， 再进行处理后放回到内存中， 如果不是这样的话，就有可能使这个变量的在寄存器内的处理时被其它的 cpu 在它们的寄存器中被改变。 而 helper1 则是知道已经有了写请求在等待得到锁，而跳转的处理地址。这里我取的名字是 _read_lock_failed_wlock_wait .align4.globl _read_lock_fa

18、iled_wlock_wait_read_lock_failed_wlock_wait:1: rep; nop cmpl $0,(%ebx) jnz 1bjs _read_lock_failed_wlock_waitret 这里就是不断的检查wlock_waiting的数值是否为0，如果不是 0 就要执行空转指令。而helper 就是取前面的_read_lock_failed 的名字，但有一点的变化。.align 4.globl _read_lock_failed_read_lock_failed:lock ; incl(%eax)1: rep; nopcmpl$0,(%ebx)jnz 1br

19、ep; nopcmpl$1,(%eax)js 1block ; decl(%eax)js _read_lock_failedret 这里的还是要不断的检查是否有写请求在等待，因为如果不是这样的话，在前面的指令的跳转过程中就有可能有写的请求到来，而我们是要严格的执行写者优先的读写规则。如果在试图得到读锁过程中失败，也要跳转到检查写请求的地址，也是这个原因。对于写锁的获得也要修改。#define _build_write_lock_const(rw,wlock_wait, helper)asm volatile(LOCK "subl $1,(%1)nt" LOCK "

20、subl $"RW_LOCK_BIAS_STR ",(%0)nt""jnz 2fn""1:n"LOCK "addl $1,(%1)nt"".section .text.lock,"ax"n""2:tpushl %eaxnt""leal %0,%eaxnt""call " helper "nt""popl %eaxnt" "jmp 1bn" ".previous" :"=m" (*(volatile int *)rw) ,"=m" (*(volatile int *)wlock_waiting) : : "memory") 这里与在 _build_read_lock_const 中的原理类似，但有一点