多核多線程已經(jīng)成為當(dāng)下一個(gè)時(shí)髦的話題,,而無(wú)鎖編程更是這個(gè)時(shí)髦話題中的熱點(diǎn)話題,。Linux內(nèi)核可能是當(dāng)今最大最復(fù)雜的并行程序之一,為我們分析多核多線程提供了絕佳的范例,。內(nèi)核設(shè)計(jì)者已經(jīng)將最新的無(wú)鎖編程技術(shù)帶進(jìn)了2,。6系統(tǒng)內(nèi)核中,本文以2,。6,。10版本為藍(lán)本,,帶領(lǐng)您領(lǐng)略多核多線程編程的真諦,窺探無(wú)鎖編程的奧秘,,體味大師們的高雅設(shè)計(jì),!
非阻塞型同步(Non-blockingSynchronization)簡(jiǎn)介
如何正確有效的保護(hù)共享數(shù)據(jù)是編寫(xiě)并行程序必須面臨的一個(gè)難題,通常的手段就是同步,。同步可分為阻塞型同步(BlockingSynchronization)和非阻塞型同步(Non-blockingSynchronization),。
阻塞型同步是指當(dāng)一個(gè)線程到達(dá)臨界區(qū)時(shí),因另外一個(gè)線程已經(jīng)持有訪問(wèn)該共享數(shù)據(jù)的鎖,,從而不能獲取鎖資源而阻塞,,直到另外一個(gè)線程釋放鎖。常見(jiàn)的同步原語(yǔ)有mutex,、semaphore等,。如果同步方案采用不當(dāng),就會(huì)造成死鎖(deadlock),,活鎖(livelock)和優(yōu)先級(jí)反轉(zhuǎn)(priorityinversion),,以及效率低下等現(xiàn)象。
為了降低風(fēng)險(xiǎn)程度和提高程序運(yùn)行效率,,業(yè)界提出了不采用鎖的同步方案,,依照這種設(shè)計(jì)思路設(shè)計(jì)的算法稱(chēng)為非阻塞型算法,其本質(zhì)特征就是停止一個(gè)線程的執(zhí)行不會(huì)阻礙系統(tǒng)中其他執(zhí)行實(shí)體的運(yùn)行,。
當(dāng)今比較流行的Non-blockingSynchronization實(shí)現(xiàn)方案有三種:
Wait-free
Wait-free是指任意線程的任何操作都可以在有限步之內(nèi)結(jié)束,,而不用關(guān)心其它線程的執(zhí)行速度。Wait-free是基于per-thread的,,可以認(rèn)為是starvation-free的,。非常遺憾的是實(shí)際情況并非如此,采用Wait-free的程序并不能保證starvation-free,,同時(shí)內(nèi)存消耗也隨線程數(shù)量而線性增長(zhǎng)。目前只有極少數(shù)的非阻塞算法實(shí)現(xiàn)了這一點(diǎn),。
Lock-free
Lock-Free是指能夠確保執(zhí)行它的所有線程中至少有一個(gè)能夠繼續(xù)往下執(zhí)行,。由于每個(gè)線程不是starvation-free的,即有些線程可能會(huì)被任意地延遲,,然而在每一步都至少有一個(gè)線程能夠往下執(zhí)行,,因此系統(tǒng)作為一個(gè)整體是在持續(xù)執(zhí)行的,可以認(rèn)為是system-wide的,。所有Wait-free的算法都是Lock-Free的,。
Obstruction-free
Obstruction-free是指在任何時(shí)間點(diǎn),一個(gè)孤立運(yùn)行線程的每一個(gè)操作可以在有限步之內(nèi)結(jié)束,。只要沒(méi)有競(jìng)爭(zhēng),,線程就可以持續(xù)運(yùn)行,。一旦共享數(shù)據(jù)被修改,Obstruction-free要求中止已經(jīng)完成的部分操作,,并進(jìn)行回滾,。所有Lock-Free的算法都是Obstruction-free的。
綜上所述,,不難得出Obstruction-free是Non-blockingsynchronization中性能最差的,,而Wait-free性能是最好的,但實(shí)現(xiàn)難度也是最大的,,因此Lock-free算法開(kāi)始被重視,,并廣泛運(yùn)用于當(dāng)今正在運(yùn)行的程序中,比如linux內(nèi)核,。
一般采用原子級(jí)的read-modify-write原語(yǔ)來(lái)實(shí)現(xiàn)Lock-Free算法,,其中LL和SC是Lock-Free理論研究領(lǐng)域的理想原語(yǔ),但實(shí)現(xiàn)這些原語(yǔ)需要CPU指令的支持,,非常遺憾的是目前沒(méi)有任何CPU直接實(shí)現(xiàn)了SC原語(yǔ),。根據(jù)此理論,業(yè)界在原子操作的基礎(chǔ)上提出了著名的CAS(Compare-And-Swap)操作來(lái)實(shí)現(xiàn)Lock-Free算法,,Intel實(shí)現(xiàn)了一條類(lèi)似該操作的指令:cmpxchg8,。
CAS原語(yǔ)負(fù)責(zé)將某處內(nèi)存地址的值(1個(gè)字節(jié))與一個(gè)期望值進(jìn)行比較,如果相等,,則將該內(nèi)存地址處的值替換為新值,,CAS操作偽碼描述如下:
清單1。CAS偽碼
BoolCAS(T*addr,,Texpected,,TnewValue)
{
if(*addr==expected)
{
*addr=newValue;
returntrue;
}
else
returnfalse;
}
在實(shí)際開(kāi)發(fā)過(guò)程中,利用CAS進(jìn)行同步,,代碼如下所示:
清單2,。CAS實(shí)際操作
do{
備份舊數(shù)據(jù);
基于舊數(shù)據(jù)構(gòu)造新數(shù)據(jù),;
}while(,!CAS(內(nèi)存地址,備份的舊數(shù)據(jù),,新數(shù)據(jù)))
就是指當(dāng)兩者進(jìn)行比較時(shí),,如果相等,則證明共享數(shù)據(jù)沒(méi)有被修改,,替換成新值,,然后繼續(xù)往下運(yùn)行;如果不相等,,說(shuō)明共享數(shù)據(jù)已經(jīng)被修改,,放棄已經(jīng)所做的操作,,然后重新執(zhí)行剛才的操作。容易看出CAS操作是基于共享數(shù)據(jù)不會(huì)被修改的假設(shè),,采用了類(lèi)似于數(shù)據(jù)庫(kù)的commit-retry的模式,。當(dāng)同步?jīng)_突出現(xiàn)的機(jī)會(huì)很少時(shí),這種假設(shè)能帶來(lái)較大的性能提升,。
加鎖的層級(jí)
根據(jù)復(fù)雜程度,、加鎖粒度及運(yùn)行速度,可以得出如下圖所示的鎖層級(jí):
圖1,。加鎖層級(jí)
其中標(biāo)注為紅色字體的方案為Blockingsynchronization,,黑色字體為Non-blockingsynchronization。Lock-based和Lockless-based兩者之間的區(qū)別僅僅是加鎖粒度的不同,。圖中最底層的方案就是大家經(jīng)常使用的mutex和semaphore等方案,,代碼復(fù)雜度低,但運(yùn)行效率也最低,。
Linux內(nèi)核中的無(wú)鎖分析
Linux內(nèi)核可能是當(dāng)今最大最復(fù)雜的并行程序之一,,它的并行主要來(lái)至于中斷、內(nèi)核搶占及SMP等,。內(nèi)核設(shè)計(jì)者們?yōu)榱瞬粩嗵岣週inux內(nèi)核的效率,,從全局著眼,逐步廢棄了大內(nèi)核鎖來(lái)降低鎖的粒度,;從細(xì)處下手,,不斷對(duì)局部代碼進(jìn)行優(yōu)化,用無(wú)鎖編程替代基于鎖的方案,,如seqlock及RCU等,;不斷減少鎖沖突程度、降低等待時(shí)間,,如Double-checkedlocking和原子鎖等,。
無(wú)論什么時(shí)候當(dāng)臨界區(qū)中的代碼僅僅需要加鎖一次,同時(shí)當(dāng)其獲取鎖的時(shí)候必須是線程安全的,,此時(shí)就可以利用Double-checkedLocking模式來(lái)減少鎖競(jìng)爭(zhēng)和加鎖載荷,。目前Double-checkedLocking已經(jīng)廣泛應(yīng)用于單例(Singleton)模式中。內(nèi)核設(shè)計(jì)者基于此思想,,巧妙的將Double-checkedLocking方法運(yùn)用于內(nèi)核代碼中,。
當(dāng)一個(gè)進(jìn)程已經(jīng)僵死,,即進(jìn)程處于TASK_ZOMBIE狀態(tài),,如果父進(jìn)程調(diào)用waitpid()系統(tǒng)調(diào)用時(shí),父進(jìn)程需要為子進(jìn)程做一些清理性的工作,,代碼如下所示:
清單3,。少鎖操作
984staticintwait_task_zombie(task_t*p,,intnoreap,
985structsiginfo__user*infop,,
986int__user*stat_addr,,structrusage__user*ru)
987{
……
1103if(p->real_parent!=p->parent){
1104write_lock_irq(&tasklist_lock);
1105/*Double-checkwithlockheld,。*/
1106if(p->real_parent,!=p->parent){
1107__ptrace_unlink(p);
1108//TODO:isthissafe?
1109p->exit_state=EXIT_ZOMBIE;
……
1120}
1121write_unlock_irq(&tasklist_lock);
1122}
……
1127}
如果將write_lock_irq放置于1103行之前,,鎖的范圍過(guò)大,,鎖的負(fù)載也會(huì)加重,影響效率,;如果將加鎖的代碼放到判斷里面,,且沒(méi)有1106行的代碼,程序會(huì)正確嗎,?在單核情況下是正確的,,但在雙核情況下問(wèn)題就出現(xiàn)了。一個(gè)非主進(jìn)程在一個(gè)CPU上運(yùn)行,,正準(zhǔn)備調(diào)用exit退出,,此時(shí)主進(jìn)程在另外一個(gè)CPU上運(yùn)行,在子進(jìn)程調(diào)用release_task函數(shù)之前調(diào)用上述代碼,。子進(jìn)程在exit_notify函數(shù)中,,先持有讀寫(xiě)鎖tasklist_lock,調(diào)用forget_original_parent,。主進(jìn)程運(yùn)行到1104處,,由于此時(shí)子進(jìn)程先持有該鎖,所以父進(jìn)程只好等待,。在forget_original_parent函數(shù)中,,如果該子進(jìn)程還有子進(jìn)程,則會(huì)調(diào)用reparent_thread(),,將執(zhí)行p->parent=p->real_parent;語(yǔ)句,,導(dǎo)致兩者相等,等非主進(jìn)程釋放讀寫(xiě)鎖tasklist_lock時(shí),,另外一個(gè)CPU上的主進(jìn)程被喚醒,,一旦開(kāi)始執(zhí)行,繼續(xù)運(yùn)行將會(huì)導(dǎo)致bug,。
嚴(yán)格的說(shuō),,Double-checkedlocking不屬于無(wú)鎖編程的范疇,但由原來(lái)的每次加鎖訪問(wèn)到大多數(shù)情況下無(wú)須加鎖,,就是一個(gè)巨大的進(jìn)步,。同時(shí)從這里也可以看出一點(diǎn)端倪,,內(nèi)核開(kāi)發(fā)者為了降低鎖沖突率,減少等待時(shí)間,,提高運(yùn)行效率,,一直在持續(xù)不斷的進(jìn)行改進(jìn)。
原子操作可以保證指令以原子的方式執(zhí)行——執(zhí)行過(guò)程不被打斷,。內(nèi)核提供了兩組原子操作接口:一組針對(duì)于整數(shù)進(jìn)行操作,,另外一組針對(duì)于單獨(dú)的位進(jìn)行操作。內(nèi)核中的原子操作通常是內(nèi)聯(lián)函數(shù),,一般是通過(guò)內(nèi)嵌匯編指令來(lái)完成,。對(duì)于一些簡(jiǎn)單的需求,例如全局統(tǒng)計(jì),、引用計(jì)數(shù)等等,,可以歸結(jié)為是對(duì)整數(shù)的原子計(jì)算。
1,。Lock-free應(yīng)用場(chǎng)景一——SpinLock
SpinLock是一種輕量級(jí)的同步方法,,一種非阻塞鎖。當(dāng)lock操作被阻塞時(shí),,并不是把自己掛到一個(gè)等待隊(duì)列,,而是死循環(huán)CPU空轉(zhuǎn)等待其他線程釋放鎖。Spinlock鎖實(shí)現(xiàn)代碼如下:
清單4,。spinlock實(shí)現(xiàn)代碼
staticinlinevoid__preempt_spin_lock(spinlock_t*lock)
{
……
do{
preempt_enable();
while(spin_is_locked(lock))
cpu_relax();
preempt_disable();
}while(,!_raw_spin_trylock(lock));
}
staticinlineint_raw_spin_trylock(spinlock_t*lock)
{
charoldval;
__asm____volatile__(
"xchgb%b0,%1"
:"=q"(oldval),,"=m"(lock->lock)
:"0"(0):"memory");
returnoldval>0;
}
匯編語(yǔ)言指令xchgb原子性的交換8位oldval(存0)和lock->lock的值,,如果oldval為1(lock初始值為1),則獲取鎖成功,,反之,,則繼續(xù)循環(huán),接著relax休息一會(huì)兒,,然后繼續(xù)周而復(fù)始,,直到成功。
對(duì)于應(yīng)用程序來(lái)說(shuō),,希望任何時(shí)候都能獲取到鎖,,也就是期望lock->lock為1,那么用CAS原語(yǔ)來(lái)描述_raw_spin_trylock(lock)就是CAS(lock->lock,,1,,0);
如果同步操作總是能在數(shù)條指令內(nèi)完成,那么使用SpinLock會(huì)比傳統(tǒng)的mutexlock快一個(gè)數(shù)量級(jí)。SpinLock多用于多核系統(tǒng)中,,適合于鎖持有時(shí)間小于將一個(gè)線程阻塞和喚醒所需時(shí)間的場(chǎng)合。
pthread庫(kù)已經(jīng)提供了對(duì)spinlock的支持,,所以用戶(hù)態(tài)程序也能很方便的使用spinlock了,,需要包含pthread。h,。在某些場(chǎng)景下,,pthread_spin_lock效率是pthread_mutex_lock效率的一倍多。美中不足的是,,內(nèi)核實(shí)現(xiàn)了讀寫(xiě)spinlock鎖,,但pthread未能實(shí)現(xiàn)。
2,。Lock-free應(yīng)用場(chǎng)景二——Seqlock
手表最主要最常用的功能是讀時(shí)間,,而不是校正時(shí)間,一旦后者成了最常用的功能,,消費(fèi)者肯定不會(huì)買(mǎi)賬,。計(jì)算機(jī)的時(shí)鐘也是這個(gè)功能,修改時(shí)間是小概率事件,,而讀時(shí)間是經(jīng)常發(fā)生的行為,。以下代碼摘自2。4,。34內(nèi)核:
清單5,。2。4,。34seqlock實(shí)現(xiàn)代碼
443voiddo_gettimeofday(structtimeval*tv)
444{
……
448read_lock_irqsave(&xtime_lock,,flags);
……
455sec=xtime。tv_sec;
456usec+=xtime,。tv_usec;
457read_unlock_irqrestore(&xtime_lock,,flags);
……
466}
468voiddo_settimeofday(structtimeval*tv)
469{
470write_lock_irq(&xtime_lock);
……
490write_unlock_irq(&xtime_lock);
491}
不難發(fā)現(xiàn)獲取時(shí)間和修改時(shí)間采用的是spinlock讀寫(xiě)鎖,讀鎖和寫(xiě)鎖具有相同的優(yōu)先級(jí),,只要讀持有鎖,,寫(xiě)鎖就必須等待,反之亦然,。
Linux2,。6內(nèi)核中引入一種新型鎖——順序鎖(seqlock),它與spinlock讀寫(xiě)鎖非常相似,,只是它為寫(xiě)者賦予了較高的優(yōu)先級(jí),。也就是說(shuō),即使讀者正在讀的時(shí)候也允許寫(xiě)者繼續(xù)運(yùn)行。當(dāng)存在多個(gè)讀者和少數(shù)寫(xiě)者共享一把鎖時(shí),,seqlock便有了用武之地,,因?yàn)閟eqlock對(duì)寫(xiě)者更有利,只要沒(méi)有其他寫(xiě)者,,寫(xiě)鎖總能獲取成功,。根據(jù)lock-free和時(shí)鐘功能的思想,內(nèi)核開(kāi)發(fā)者在2,。6內(nèi)核中,,將上述讀寫(xiě)鎖修改成了順序鎖seqlock,代碼如下:
清單6,。2,。6。10seqlock實(shí)現(xiàn)代碼
staticinlineunsignedread_seqbegin(constseqlock_t*sl)
{
unsignedret=sl->sequence;
smp_rmb();
returnret;
}
staticinlineintread_seqretry(constseqlock_t*sl,,unsignediv)
{
smp_rmb();
return(iv&1)|(sl->sequence^iv);
}
staticinlinevoidwrite_seqlock(seqlock_t*sl)
{
spin_lock(&sl->lock);
++sl->sequence;
smp_wmb();
}
voiddo_gettimeofday(structtimeval*tv)
{
unsignedlongseq;
unsignedlongusec,,sec;
unsignedlongmax_ntp_tick;
……
do{
unsignedlonglost;
seq=read_seqbegin(&xtime_lock);
……
sec=xtime。tv_sec;
usec+=(xtime,。tv_nsec/1000);
}while(read_seqretry(&xtime_lock,,seq));
……
tv->tv_sec=sec;
tv->tv_usec=usec;
}
intdo_settimeofday(structtimespec*tv)
{
……
write_seqlock_irq(&xtime_lock);
……
write_sequnlock_irq(&xtime_lock);
clock_was_set();
return0;
}
Seqlock實(shí)現(xiàn)原理是依賴(lài)一個(gè)序列計(jì)數(shù)器,當(dāng)寫(xiě)者寫(xiě)入數(shù)據(jù)時(shí),,會(huì)得到一把鎖,,并且將序列值加1。當(dāng)讀者讀取數(shù)據(jù)之前和之后,,該序列號(hào)都會(huì)被讀取,,如果讀取的序列號(hào)值都相同,則表明寫(xiě)沒(méi)有發(fā)生,。反之,,表明發(fā)生過(guò)寫(xiě)事件,則放棄已進(jìn)行的操作,,重新循環(huán)一次,,直至成功。不難看出,,do_gettimeofday函數(shù)里面的while循環(huán)和接下來(lái)的兩行賦值操作就是CAS操作,。
采用順序鎖seqlock好處就是寫(xiě)者永遠(yuǎn)不會(huì)等待,缺點(diǎn)就是有些時(shí)候讀者不得不反復(fù)多次讀相同的數(shù)據(jù)直到它獲得有效的副本,。當(dāng)要保護(hù)的臨界區(qū)很小,,很簡(jiǎn)單,頻繁讀取而寫(xiě)入很少發(fā)生(WRRM---WriteRarelyReadMostly)且必須快速時(shí),,就可以使用seqlock,。但seqlock不能保護(hù)包含有指針的數(shù)據(jù)結(jié)構(gòu),,因?yàn)楫?dāng)寫(xiě)者修改數(shù)據(jù)結(jié)構(gòu)時(shí),讀者可能會(huì)訪問(wèn)一個(gè)無(wú)效的指針,。
3,。Lock-free應(yīng)用場(chǎng)景三——RCU
在2。6內(nèi)核中,,開(kāi)發(fā)者還引入了一種新的無(wú)鎖機(jī)制-RCU(Read-Copy-Update),,允許多個(gè)讀者和寫(xiě)者并發(fā)執(zhí)行。RCU技術(shù)的核心是寫(xiě)操作分為寫(xiě)和更新兩步,,允許讀操作在任何時(shí)候無(wú)阻礙的運(yùn)行,換句話說(shuō),,就是通過(guò)延遲寫(xiě)來(lái)提高同步性能,。RCU主要應(yīng)用于WRRM場(chǎng)景,但它對(duì)可保護(hù)的數(shù)據(jù)結(jié)構(gòu)做了一些限定:RCU只保護(hù)被動(dòng)態(tài)分配并通過(guò)指針引用的數(shù)據(jù)結(jié)構(gòu),,同時(shí)讀寫(xiě)控制路徑不能有睡眠,。以下數(shù)組動(dòng)態(tài)增長(zhǎng)代碼摘自2。4,。34內(nèi)核:
清單7,。2。4,。34RCU實(shí)現(xiàn)代碼
其中ipc_lock是讀者,,grow_ary是寫(xiě)者,不論是讀或者寫(xiě),,都需要加spinlock對(duì)被保護(hù)的數(shù)據(jù)結(jié)構(gòu)進(jìn)行訪問(wèn),。改變數(shù)組大小是小概率事件,而讀取是大概率事件,,同時(shí)被保護(hù)的數(shù)據(jù)結(jié)構(gòu)是指針,,滿足RCU運(yùn)用場(chǎng)景。以下代碼摘自2,。6,。10內(nèi)核:
清單8。2,。6,。10RCU實(shí)現(xiàn)代碼
#definercu_read_lock()preempt_disable()
#definercu_read_unlock()preempt_enable()
#definercu_assign_pointer(p,v)({
smp_wmb();
(p)=(v);
})
structkern_ipc_perm*ipc_lock(structipc_ids*ids,,intid)
{
……
rcu_read_lock();
entries=rcu_dereference(ids->entries);
if(lid>=entries->size){
rcu_read_unlock();
returnNULL;
}
out=entries->p[lid];
if(out==NULL){
rcu_read_unlock();
returnNULL;
}
……
returnout;
}
staticintgrow_ary(structipc_ids*ids,,intnewsize)
{
structipc_id_ary*new;
structipc_id_ary*old;
……
new=ipc_rcu_alloc(sizeof(structkern_ipc_perm*)*newsize+
sizeof(structipc_id_ary));
if(new==NULL)
returnsize;
new->size=newsize;
memcpy(new->p,ids->entries->p,,sizeof(structkern_ipc_perm*)*size
+sizeof(structipc_id_ary));
for(i=size;inew->p[i]=NULL;
}
old=ids->entries;
/*
*Usercu_assign_pointer()tomakesurethememcpyedcontents
*ofthenewarrayarevisiblebeforethenewarraybecomesvisible,。
*/
rcu_assign_pointer(ids->entries,new);
ipc_rcu_putref(old);
returnnewsize;
}
縱觀整個(gè)流程,寫(xiě)者除內(nèi)核屏障外,,幾乎沒(méi)有一把鎖,。當(dāng)寫(xiě)者需要更新數(shù)據(jù)結(jié)構(gòu)時(shí),首先復(fù)制該數(shù)據(jù)結(jié)構(gòu),,申請(qǐng)new內(nèi)存,,然后對(duì)副本進(jìn)行修改,調(diào)用memcpy將原數(shù)組的內(nèi)容拷貝到new中,,同時(shí)對(duì)擴(kuò)大的那部分賦新值,,修改完畢后,寫(xiě)者調(diào)用rcu_assign_pointer修改相關(guān)數(shù)據(jù)結(jié)構(gòu)的指針,,使之指向被修改后的新副本,,整個(gè)寫(xiě)操作一氣呵成,其中修改指針值的操作屬于原子操作,。在數(shù)據(jù)結(jié)構(gòu)被寫(xiě)者修改后,,需要調(diào)用內(nèi)存屏障smp_wmb,讓其他CPU知曉已更新的指針值,,否則會(huì)導(dǎo)致SMP環(huán)境下的bug,。當(dāng)所有潛在的讀者都執(zhí)行完成后,調(diào)用call_rcu釋放舊副本,。同Spinlock一樣,,RCU同步技術(shù)主要適用于SMP環(huán)境。
環(huán)形緩沖區(qū)是生產(chǎn)者和消費(fèi)者模型中常用的數(shù)據(jù)結(jié)構(gòu),。生產(chǎn)者將數(shù)據(jù)放入數(shù)組的尾端,,而消費(fèi)者從數(shù)組的另一端移走數(shù)據(jù),當(dāng)達(dá)到數(shù)組的尾部時(shí),,生產(chǎn)者繞回到數(shù)組的頭部,。
如果只有一個(gè)生產(chǎn)者和一個(gè)消費(fèi)者,那么就可以做到免鎖訪問(wèn)環(huán)形緩沖區(qū)(RingBuffer),。寫(xiě)入索引只允許生產(chǎn)者訪問(wèn)并修改,,只要寫(xiě)入者在更新索引之前將新的值保存到緩沖區(qū)中,則讀者將始終看到一致的數(shù)據(jù)結(jié)構(gòu),。同理,,讀取索引也只允許消費(fèi)者訪問(wèn)并修改。
圖2,。環(huán)形緩沖區(qū)實(shí)現(xiàn)原理圖
如圖所示,,當(dāng)讀者和寫(xiě)者指針相等時(shí),表明緩沖區(qū)是空的,,而只要寫(xiě)入指針在讀取指針后面時(shí),,表明緩沖區(qū)已滿,。
清單9。2,。6,。10環(huán)形緩沖區(qū)實(shí)現(xiàn)代碼
/*
*__kfifo_put-putssomedataintotheFIFO,nolockingversion
*Notethatwithonlyoneconcurrentreaderandoneconcurrent
*writer,,youdon'tneedextralockingtousethesefunctions,。
*/
unsignedint__kfifo_put(structkfifo*fifo,
unsignedchar*buffer,,unsignedintlen)
{
unsignedintl;
len=min(len,,fifo->size-fifo->in+fifo->out);
/*firstputthedatastartingfromfifo->intobufferend*/
l=min(len,fifo->size-(fifo->in&(fifo->size-1)));
memcpy(fifo->buffer+(fifo->in&(fifo->size-1)),,buffer,,l);
/*thenputtherest(ifany)atthebeginningofthebuffer*/
memcpy(fifo->buffer,buffer+l,,len-l);
fifo->in+=len;
returnlen;
}
/*
*__kfifo_get-getssomedatafromtheFIFO,,nolockingversion
*Notethatwithonlyoneconcurrentreaderandoneconcurrent
*writer,,youdon'tneedextralockingtousethesefunctions,。
*/
unsignedint__kfifo_get(structkfifo*fifo,
unsignedchar*buffer,,unsignedintlen)
{
unsignedintl;
len=min(len,,fifo->in-fifo->out);
/*firstgetthedatafromfifo->outuntiltheendofthebuffer*/
l=min(len,fifo->size-(fifo->out&(fifo->size-1)));
memcpy(buffer,,fifo->buffer+(fifo->out&(fifo->size-1)),,l);
/*thengettherest(ifany)fromthebeginningofthebuffer*/
memcpy(buffer+l,fifo->buffer,,len-l);
fifo->out+=len;
returnlen;
}
以上代碼摘自2,。6。10內(nèi)核,,通過(guò)代碼的注釋?zhuān)ㄐ斌w部分)可以看出,,當(dāng)只有一個(gè)消費(fèi)者和一個(gè)生產(chǎn)者時(shí),可以不用添加任何額外的鎖,,就能達(dá)到對(duì)共享數(shù)據(jù)的訪問(wèn),。
總結(jié)
通過(guò)對(duì)比2。4和2,。6內(nèi)核代碼,,不得不佩服內(nèi)核開(kāi)發(fā)者的智慧,為了提高內(nèi)核性能,,一直不斷的進(jìn)行各種優(yōu)化,,并將業(yè)界最新的lock-free理念運(yùn)用到內(nèi)核中,。
在實(shí)際開(kāi)發(fā)過(guò)程中,進(jìn)行無(wú)鎖設(shè)計(jì)時(shí),,首先進(jìn)行場(chǎng)景分析,,因?yàn)槊糠N無(wú)鎖方案都有特定的應(yīng)用場(chǎng)景,接著根據(jù)場(chǎng)景分析進(jìn)行數(shù)據(jù)結(jié)構(gòu)的初步設(shè)計(jì),,然后根據(jù)先前的分析結(jié)果進(jìn)行并發(fā)模型建模,,最后在調(diào)整數(shù)據(jù)結(jié)構(gòu)的設(shè)計(jì),以便達(dá)到最優(yōu),。