Linux下寫者優(yōu)先的讀寫鎖的設(shè)計(jì)

1、Linux 下寫者優(yōu)先的讀寫鎖的設(shè)計(jì)在 IBM Bluemix 云平臺(tái)上開發(fā)并部署您的下一個(gè)應(yīng)用。現(xiàn)在就開始免費(fèi)試用、本文的目的在 linux 下有兩種實(shí)現(xiàn)數(shù)據(jù)互斥的基本機(jī)制， 包括了 semaphore （信號(hào)量），spinlock （自旋鎖）。這里要說 的讀寫鎖 （read write lock） 是自旋鎖的一個(gè)變種，與一般的自 旋鎖的區(qū)別是，自旋鎖一次只能有一個(gè)進(jìn)程進(jìn)入臨界區(qū)，而 對讀寫鎖而言，如果進(jìn)程是讀的話，那就可以有多個(gè)進(jìn)程同 時(shí)進(jìn)入臨界區(qū)，而如果是寫的話，則只有一個(gè)可以。就現(xiàn)在的 linux 內(nèi)核源代碼的發(fā)行版本而言，已經(jīng)實(shí)現(xiàn)了讀寫鎖的個(gè)類型，就是讀者優(yōu)先的讀寫鎖。在這個(gè)設(shè)計(jì)讀的

2、請求可以更容易的進(jìn)入臨界區(qū)，而寫的請求的請求往往容易受阻，這個(gè)我在后面會(huì)分析），而我要設(shè)計(jì)的讀寫鎖，則是以寫進(jìn)程為優(yōu)先的考慮的對象，如果有寫的請求發(fā)出，則它會(huì)在被允許的最快時(shí)間內(nèi)得到響應(yīng)。這樣的好處是在 個(gè)由很多客戶端以讀的權(quán)限訪問的服務(wù)器（如一般的公眾服 務(wù)器），如果管理員對服務(wù)器的某些內(nèi)容或配置進(jìn)行修改的 話，那它的及時(shí)性就有可能無法滿足。這有時(shí)是不可以被接 受的?；仨撌锥?、 linux 現(xiàn)有的讀寫鎖狀況我先來分析現(xiàn)在linux 內(nèi)核源代碼中的讀寫鎖的實(shí)現(xiàn)方式， 這樣就可以很容易 的理解后面的寫者優(yōu)先的讀寫鎖的設(shè)計(jì)。先介紹一個(gè)數(shù)據(jù)結(jié)構(gòu)，這是在讀寫鎖起到重要作用。 (注： 下面所有的內(nèi)核源代

3、 碼均來自 linux 2.4.17 ，如果與你的現(xiàn)有的內(nèi)核源代碼不同， 請你作一些相應(yīng)的改變就可以了， 原理部分沒有變化 )typedef struct volatile unsigned int lock;#if SPINLOCK_DEBUGunsigned magic;#endif rwlock_t; 這里的 magic 是用于調(diào)試的，而 lock 就是允許可 以加的讀鎖數(shù)。這個(gè)代碼在 linux/include/asm-i386/spinlock.h中定義了 read_lock 和 write_lockstatic inline void read_lock(rwlock_t *rw)

4、#if SPINLOCK_DEBUGif (rw->magic != RWLOCK_MAGIC)BUG();#endifbuild_read_lock(rw, "_read_lock_failed");static inline void write_lock(rwlock_t *rw)#if SPINLOCK_DEBUGif (rw->magic != RWLOCK_MAGIC)BUG();#endifbuild_write_lock(rw, "_write_lock_failed");注意這里有兩個(gè)參數(shù)，一個(gè)是 rw 就是允許的讀鎖數(shù)，而

5、后 面一個(gè)參數(shù)是如果加鎖失敗的話，處理失敗的函數(shù)。在這里 真正調(diào)用的對 write_lock 是 _build_write_lock_const 或build_write_lock_ptr ，對 read_lock 中調(diào)用的是build_read_lock_const 或 _build_read_lock_ptr ，這里的取constasm決因素是調(diào)用參數(shù)的操作指令尋址方式。 我們這里只看#define _build_write_lock_const(rw, helper)volatile(LOCK "subl $" RW_LOCK_BIAS_STR ",(%0)

6、nt""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&qu

7、ot;)這里的 RW_LOCK_BIAS_STR 就是 "0x01000000" ，取這個(gè)值的 原因是這個(gè)值足夠大，可以使?jié)M足讀的請求足夠多。在".section .text.lock,"ax"n"".previous"中的內(nèi)容是把這一段的代碼匯編到一個(gè)叫.text.lock 的節(jié)中，并且這個(gè)節(jié)的屬性是可重定位和可執(zhí)行的，這樣在代碼的執(zhí) 行過程中，因?yàn)椴煌墓?jié)會(huì)被加載到不同的頁面中，所以如 果在前面不出現(xiàn) jmp ，就在 1:處結(jié)束了。而 call 的是在前面 的 write_lock 中所寫入的 _write_l

8、ock_failed ，這個(gè)函數(shù)在arch/asm-i386/kernel/semaphore.c 中定義.align .globl _write_lock_failedwrite_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_loc

9、k_failedret 這里的 LOCK 是一個(gè)在 SMP 體系結(jié)構(gòu)中鎖住總線， 不 讓其它的 CPU 訪問內(nèi)存。在這里先 " LOCK "addl $"RW_LOCK_BIAS_STR ",(%eax) 是為了防止在后面的自旋等 待中，不會(huì)讓后面的讀請求受阻，要不然的話，就會(huì)出現(xiàn)死 鎖，這是致命的。而 1: rep; nopcmpl $" RW_LOCK_BIAS_STR ",(%eax)jne 1b 就是不斷的檢查鎖是否可以得到， 得不到就會(huì) nop，這種方法可以在不改變 lock 的值的情況下實(shí)現(xiàn)等待， 這就不" LO

10、CK用 LOCK ，這樣的鎖住總線了。最后如果得到鎖就"subl $" RW_LOCK_BIAS_STR ",(%eax) 這樣就實(shí)現(xiàn)了write_lock 的功能。對讀鎖也是類似#definebuild_read_lock_const(rw, helper)asm volatile(LOCK"subl $1,%0nt""js 2fn""1:n"".section .text.lock,"ax"n""2:tpushl %eaxnt""l

11、eal %0,%eaxnt""call " helper "nt""popl %eaxnt""jmp 1bn"".previous":"=m" (*(volatile int *)rw) : : "memory") 但這里要注意一點(diǎn)， 就是對 read_lock 而言， 只要減 1 并且只要這個(gè)值不為負(fù)的話，就可以得到鎖了，但 rw.lock 的值在被初 減的很小，所以要獲得讀鎖是很容易的，從理論上說比得到 寫鎖要容易 0x01000000 倍，

12、這就是我前面說現(xiàn)在在 linux 內(nèi) 部實(shí)現(xiàn)的讀寫鎖是讀者優(yōu)先的。而這個(gè)數(shù)也讓我們?nèi)菀桌斫馐蓟臅r(shí)候就被賦值成了0x01000000 ，這個(gè)值足夠大， 而要在要獲得寫鎖時(shí)，要對這個(gè)lock值 減去0x01000000,就是 如果有一個(gè)讀或者寫請求在臨界區(qū)內(nèi)的話，第二個(gè)寫請求就 無法得到寫鎖了。而如果得不到讀鎖，所要跳的是在read_lock 所指明的 _read_lock_failed.align 4 .globl _read_lock_failedread_lock_failed:lock ; incl (%eax) 1: rep; nopcmpl $1,(%eax)js 1block ;

13、decl (%eax)js _read_lock_failed中的道理是相ret 這里的道理與前面的 _write_lock_failed似的。回頁首三、寫者優(yōu)先的讀寫鎖的實(shí)現(xiàn)那既然要實(shí)現(xiàn)以 寫者為優(yōu)先的讀寫鎖，很自然，我們就想到了在讀的請求發(fā) 生時(shí)，不先去試圖獲得讀鎖，而是去檢查有沒有寫的請求正 在等待，如果有寫的請求正在等待，則讀的請求必須先處于等待狀態(tài)。讓寫的請求完成之后，發(fā)現(xiàn)已經(jīng)沒有寫的請求在 這個(gè)數(shù)據(jù)結(jié)構(gòu)， typedef struct 等待了，才去試圖獲得讀的鎖。這里我們先來設(shè)計(jì)rwlock_tvolatile unsigned int lock;#if WLOCK_PRIORIT

14、Yvolatile unsigned int wlock_waiting;#endif #if SPINLOCK_DEBUGunsigned magic;#endif rwlock_t; 這里所增加的這個(gè) wlock_waiting 就是作為檢測是否有寫的請求在等待的標(biāo)志數(shù)。如果這個(gè)數(shù)不等于0 則說明已經(jīng)有寫的請求在等待。它的負(fù)值的大小決定了寫請求等待 的個(gè)數(shù)。這里我們先修改 _build_write_lock_const 中#define _build_write_lock_const(rw, wlock_wait,helper,helper1) asm volatile("cmp

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

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

17、;=m"(*(volatile int *)wlock_waiting): : "memory") 這里的新增加 的 wlock_waiting 是表示前面所定義的 wlock_waiting 的地址， 這個(gè)是地址，而不是值本身，它的原因是指令尋址的方式?jīng)Q 定的，為了保證指令的操作在直接對內(nèi)存，而不是把內(nèi)存中 的數(shù)據(jù) load 到寄存器中， 再進(jìn)行處理后放回到內(nèi)存中， 如果 不是這樣的話，就有可能使這個(gè)變量的在寄存器內(nèi)的處理時(shí) 被其它的 cpu 在它們的寄存器中被改變。 而 helper1 則是知道 已經(jīng)有了寫請求在等待得到鎖，而跳轉(zhuǎn)的處理地址。這里我 取的名字是

18、 _read_lock_failed_wlock_wait .align 4 .globl _read_lock_failed_wlock_waitread_lock_failed_wlock_wait:1: rep; nopcmpl $0,(%ebx)jnz 1bjs _read_lock_failed_wlock_waitret 這里就是不斷的檢查 wlock_waiting 的數(shù)值是否為 0，如果不是 0 就要執(zhí)行空轉(zhuǎn)指令。而 helper 就是取前面的.align 4read_lock_failed 的名字，但有一點(diǎn)的變化。.globl _read_lock_failedread_lo

19、ck_failed:lock ; incl (%eax) 1: rep; nopcmpl $0,(%ebx) jnz 1brep; nopcmpl $1,(%eax)js 1block ; decl (%eax)js _read_lock_failedret 這里的還是要不斷的檢查是否有寫請求在等待，因?yàn)槿绻皇沁@樣的話，在前面的指令的跳轉(zhuǎn)過程中就有可能有 寫的請求到來，而我們是要嚴(yán)格的執(zhí)行寫者優(yōu)先的讀寫規(guī) 則。如果在試圖得到讀鎖過程中失敗，也要跳轉(zhuǎn)到檢查寫請 求的地址，也是這個(gè)原因。對于寫鎖的獲得也要修改。#define _build_write_lock_const(rw,wlock_wa

20、it, helper)asm volatile(LOCK "subl $1,(%1)nt" LOCK "subl $""jnz 2fn"RW_LOCK_BIAS_STR ",(%0)nt""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但有一點(diǎn)不同，就是一旦要試圖取