多核多線程已經成為當下一個時髦的話題,，而無鎖編程更是這個時髦話題中的熱點話題。Linux內核可能是當今最大最復雜的并行程序之一,，為我們分析多核多線程提供了絕佳的范例,。內核設計者已經將最新的無鎖編程技術帶進了2。6系統(tǒng)內核中，本文以2,。6,。10版本為藍本，帶領您領略多核多線程編程的真諦,，窺探無鎖編程的奧秘,，體味大師們的高雅設計！
非阻塞型同步(Non-blockingSynchronization)簡介
如何正確有效的保護共享數(shù)據(jù)是編寫并行程序必須面臨的一個難題,，通常的手段就是同步,。同步可分為阻塞型同步（BlockingSynchronization）和非阻塞型同步（Non-blockingSynchronization）。
阻塞型同步是指當一個線程到達臨界區(qū)時,，因另外一個線程已經持有訪問該共享數(shù)據(jù)的鎖,，從而不能獲取鎖資源而阻塞，直到另外一個線程釋放鎖,。常見的同步原語有mutex,、semaphore等。如果同步方案采用不當,，就會造成死鎖（deadlock）,，活鎖（livelock）和優(yōu)先級反轉（priorityinversion），以及效率低下等現(xiàn)象,。
為了降低風險程度和提高程序運行效率,，業(yè)界提出了不采用鎖的同步方案，依照這種設計思路設計的算法稱為非阻塞型算法,，其本質特征就是停止一個線程的執(zhí)行不會阻礙系統(tǒng)中其他執(zhí)行實體的運行,。
當今比較流行的Non-blockingSynchronization實現(xiàn)方案有三種：
Wait-free
Wait-free是指任意線程的任何操作都可以在有限步之內結束，而不用關心其它線程的執(zhí)行速度,。Wait-free是基于per-thread的,，可以認為是starvation-free的。非常遺憾的是實際情況并非如此,，采用Wait-free的程序并不能保證starvation-free,，同時內存消耗也隨線程數(shù)量而線性增長。目前只有極少數(shù)的非阻塞算法實現(xiàn)了這一點,。
Lock-free
Lock-Free是指能夠確保執(zhí)行它的所有線程中至少有一個能夠繼續(xù)往下執(zhí)行,。由于每個線程不是starvation-free的，即有些線程可能會被任意地延遲,，然而在每一步都至少有一個線程能夠往下執(zhí)行,，因此系統(tǒng)作為一個整體是在持續(xù)執(zhí)行的，可以認為是system-wide的,。所有Wait-free的算法都是Lock-Free的,。
Obstruction-free
Obstruction-free是指在任何時間點,，一個孤立運行線程的每一個操作可以在有限步之內結束。只要沒有競爭,，線程就可以持續(xù)運行,。一旦共享數(shù)據(jù)被修改，Obstruction-free要求中止已經完成的部分操作,，并進行回滾,。所有Lock-Free的算法都是Obstruction-free的。
綜上所述,，不難得出Obstruction-free是Non-blockingsynchronization中性能最差的,，而Wait-free性能是最好的，但實現(xiàn)難度也是最大的,，因此Lock-free算法開始被重視,，并廣泛運用于當今正在運行的程序中，比如linux內核,。
一般采用原子級的read-modify-write原語來實現(xiàn)Lock-Free算法,，其中LL和SC是Lock-Free理論研究領域的理想原語，但實現(xiàn)這些原語需要CPU指令的支持,，非常遺憾的是目前沒有任何CPU直接實現(xiàn)了SC原語,。根據(jù)此理論，業(yè)界在原子操作的基礎上提出了著名的CAS（Compare-And-Swap）操作來實現(xiàn)Lock-Free算法,，Intel實現(xiàn)了一條類似該操作的指令：cmpxchg8,。
CAS原語負責將某處內存地址的值（1個字節(jié)）與一個期望值進行比較，如果相等,，則將該內存地址處的值替換為新值，CAS操作偽碼描述如下：
清單1,。CAS偽碼
BoolCAS(T*addr,，Texpected，TnewValue)
{
if(*addr==expected)
{
*addr=newValue;
returntrue;
}
else
returnfalse;
}
在實際開發(fā)過程中,，利用CAS進行同步,，代碼如下所示：
清單2。CAS實際操作
do{
備份舊數(shù)據(jù),；
基于舊數(shù)據(jù)構造新數(shù)據(jù),；
}while(！CAS(內存地址,，備份的舊數(shù)據(jù),，新數(shù)據(jù)))
就是指當兩者進行比較時，如果相等,，則證明共享數(shù)據(jù)沒有被修改,，替換成新值,，然后繼續(xù)往下運行；如果不相等,，說明共享數(shù)據(jù)已經被修改,，放棄已經所做的操作，然后重新執(zhí)行剛才的操作,。容易看出CAS操作是基于共享數(shù)據(jù)不會被修改的假設,，采用了類似于數(shù)據(jù)庫的commit-retry的模式。當同步沖突出現(xiàn)的機會很少時,，這種假設能帶來較大的性能提升,。
加鎖的層級
根據(jù)復雜程度、加鎖粒度及運行速度,，可以得出如下圖所示的鎖層級：
圖1,。加鎖層級
其中標注為紅色字體的方案為Blockingsynchronization，黑色字體為Non-blockingsynchronization,。Lock-based和Lockless-based兩者之間的區(qū)別僅僅是加鎖粒度的不同,。圖中最底層的方案就是大家經常使用的mutex和semaphore等方案，代碼復雜度低,，但運行效率也最低,。
Linux內核中的無鎖分析
Linux內核可能是當今最大最復雜的并行程序之一，它的并行主要來至于中斷,、內核搶占及SMP等,。內核設計者們?yōu)榱瞬粩嗵岣週inux內核的效率，從全局著眼,，逐步廢棄了大內核鎖來降低鎖的粒度,；從細處下手，不斷對局部代碼進行優(yōu)化,，用無鎖編程替代基于鎖的方案,，如seqlock及RCU等；不斷減少鎖沖突程度,、降低等待時間,，如Double-checkedlocking和原子鎖等。
無論什么時候當臨界區(qū)中的代碼僅僅需要加鎖一次,，同時當其獲取鎖的時候必須是線程安全的,，此時就可以利用Double-checkedLocking模式來減少鎖競爭和加鎖載荷。目前Double-checkedLocking已經廣泛應用于單例(Singleton)模式中,。內核設計者基于此思想,，巧妙的將Double-checkedLocking方法運用于內核代碼中。
當一個進程已經僵死,，即進程處于TASK_ZOMBIE狀態(tài),，如果父進程調用waitpid()系統(tǒng)調用時,，父進程需要為子進程做一些清理性的工作，代碼如下所示：
清單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行之前,，鎖的范圍過大,，鎖的負載也會加重，影響效率,；如果將加鎖的代碼放到判斷里面,，且沒有1106行的代碼，程序會正確嗎,？在單核情況下是正確的,，但在雙核情況下問題就出現(xiàn)了。一個非主進程在一個CPU上運行,，正準備調用exit退出,，此時主進程在另外一個CPU上運行，在子進程調用release_task函數(shù)之前調用上述代碼,。子進程在exit_notify函數(shù)中,，先持有讀寫鎖tasklist_lock，調用forget_original_parent,。主進程運行到1104處,，由于此時子進程先持有該鎖，所以父進程只好等待,。在forget_original_parent函數(shù)中，如果該子進程還有子進程,，則會調用reparent_thread(),，將執(zhí)行p->parent=p->real_parent;語句，導致兩者相等,，等非主進程釋放讀寫鎖tasklist_lock時,，另外一個CPU上的主進程被喚醒，一旦開始執(zhí)行,，繼續(xù)運行將會導致bug,。
嚴格的說,，Double-checkedlocking不屬于無鎖編程的范疇，但由原來的每次加鎖訪問到大多數(shù)情況下無須加鎖,，就是一個巨大的進步,。同時從這里也可以看出一點端倪，內核開發(fā)者為了降低鎖沖突率,，減少等待時間,，提高運行效率，一直在持續(xù)不斷的進行改進,。
原子操作可以保證指令以原子的方式執(zhí)行——執(zhí)行過程不被打斷,。內核提供了兩組原子操作接口：一組針對于整數(shù)進行操作，另外一組針對于單獨的位進行操作,。內核中的原子操作通常是內聯(lián)函數(shù),，一般是通過內嵌匯編指令來完成。對于一些簡單的需求,，例如全局統(tǒng)計,、引用計數(shù)等等，可以歸結為是對整數(shù)的原子計算,。
1,。Lock-free應用場景一——SpinLock
SpinLock是一種輕量級的同步方法，一種非阻塞鎖,。當lock操作被阻塞時,，并不是把自己掛到一個等待隊列，而是死循環(huán)CPU空轉等待其他線程釋放鎖,。Spinlock鎖實現(xiàn)代碼如下：
清單4,。spinlock實現(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;
}
匯編語言指令xchgb原子性的交換8位oldval(存0)和lock->lock的值，如果oldval為1(lock初始值為1),，則獲取鎖成功,，反之，則繼續(xù)循環(huán),，接著relax休息一會兒,，然后繼續(xù)周而復始，直到成功,。
對于應用程序來說,，希望任何時候都能獲取到鎖，也就是期望lock->lock為1,，那么用CAS原語來描述_raw_spin_trylock(lock)就是CAS(lock->lock,，1,，0);
如果同步操作總是能在數(shù)條指令內完成，那么使用SpinLock會比傳統(tǒng)的mutexlock快一個數(shù)量級,。SpinLock多用于多核系統(tǒng)中,，適合于鎖持有時間小于將一個線程阻塞和喚醒所需時間的場合。
pthread庫已經提供了對spinlock的支持,，所以用戶態(tài)程序也能很方便的使用spinlock了,，需要包含pthread。h,。在某些場景下,，pthread_spin_lock效率是pthread_mutex_lock效率的一倍多。美中不足的是,，內核實現(xiàn)了讀寫spinlock鎖,，但pthread未能實現(xiàn)。
2,。Lock-free應用場景二——Seqlock
手表最主要最常用的功能是讀時間,，而不是校正時間，一旦后者成了最常用的功能,，消費者肯定不會買賬,。計算機的時鐘也是這個功能，修改時間是小概率事件,，而讀時間是經常發(fā)生的行為,。以下代碼摘自2。4,。34內核：
清單5,。2。4,。34seqlock實現(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)獲取時間和修改時間采用的是spinlock讀寫鎖，讀鎖和寫鎖具有相同的優(yōu)先級,，只要讀持有鎖,，寫鎖就必須等待，反之亦然,。
Linux2。6內核中引入一種新型鎖——順序鎖(seqlock),，它與spinlock讀寫鎖非常相似,，只是它為寫者賦予了較高的優(yōu)先級,。也就是說，即使讀者正在讀的時候也允許寫者繼續(xù)運行,。當存在多個讀者和少數(shù)寫者共享一把鎖時,，seqlock便有了用武之地，因為seqlock對寫者更有利,，只要沒有其他寫者,，寫鎖總能獲取成功。根據(jù)lock-free和時鐘功能的思想,，內核開發(fā)者在2,。6內核中，將上述讀寫鎖修改成了順序鎖seqlock,，代碼如下：
清單6,。2。6,。10seqlock實現(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實現(xiàn)原理是依賴一個序列計數(shù)器,，當寫者寫入數(shù)據(jù)時,，會得到一把鎖，并且將序列值加1,。當讀者讀取數(shù)據(jù)之前和之后,，該序列號都會被讀取，如果讀取的序列號值都相同,，則表明寫沒有發(fā)生,。反之，表明發(fā)生過寫事件,，則放棄已進行的操作,，重新循環(huán)一次，直至成功,。不難看出,，do_gettimeofday函數(shù)里面的while循環(huán)和接下來的兩行賦值操作就是CAS操作。
采用順序鎖seqlock好處就是寫者永遠不會等待,，缺點就是有些時候讀者不得不反復多次讀相同的數(shù)據(jù)直到它獲得有效的副本,。當要保護的臨界區(qū)很小，很簡單，頻繁讀取而寫入很少發(fā)生（WRRM---WriteRarelyReadMostly）且必須快速時,，就可以使用seqlock,。但seqlock不能保護包含有指針的數(shù)據(jù)結構，因為當寫者修改數(shù)據(jù)結構時,，讀者可能會訪問一個無效的指針,。
3。Lock-free應用場景三——RCU
在2,。6內核中,，開發(fā)者還引入了一種新的無鎖機制-RCU(Read-Copy-Update)，允許多個讀者和寫者并發(fā)執(zhí)行,。RCU技術的核心是寫操作分為寫和更新兩步,，允許讀操作在任何時候無阻礙的運行，換句話說,，就是通過延遲寫來提高同步性能,。RCU主要應用于WRRM場景，但它對可保護的數(shù)據(jù)結構做了一些限定：RCU只保護被動態(tài)分配并通過指針引用的數(shù)據(jù)結構,，同時讀寫控制路徑不能有睡眠,。以下數(shù)組動態(tài)增長代碼摘自2。4,。34內核：
清單7,。2。4,。34RCU實現(xiàn)代碼
其中ipc_lock是讀者,，grow_ary是寫者，不論是讀或者寫,，都需要加spinlock對被保護的數(shù)據(jù)結構進行訪問,。改變數(shù)組大小是小概率事件，而讀取是大概率事件,，同時被保護的數(shù)據(jù)結構是指針,，滿足RCU運用場景。以下代碼摘自2,。6,。10內核：
清單8。2,。6,。10RCU實現(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;
}
縱觀整個流程，寫者除內核屏障外,，幾乎沒有一把鎖。當寫者需要更新數(shù)據(jù)結構時,，首先復制該數(shù)據(jù)結構,，申請new內存，然后對副本進行修改,，調用memcpy將原數(shù)組的內容拷貝到new中,，同時對擴大的那部分賦新值，修改完畢后,，寫者調用rcu_assign_pointer修改相關數(shù)據(jù)結構的指針,，使之指向被修改后的新副本，整個寫操作一氣呵成,，其中修改指針值的操作屬于原子操作,。在數(shù)據(jù)結構被寫者修改后，需要調用內存屏障smp_wmb,，讓其他CPU知曉已更新的指針值,，否則會導致SMP環(huán)境下的bug。當所有潛在的讀者都執(zhí)行完成后,，調用call_rcu釋放舊副本,。同Spinlock一樣，RCU同步技術主要適用于SMP環(huán)境,。
環(huán)形緩沖區(qū)是生產者和消費者模型中常用的數(shù)據(jù)結構,。生產者將數(shù)據(jù)放入數(shù)組的尾端，而消費者從數(shù)組的另一端移走數(shù)據(jù),，當達到數(shù)組的尾部時,，生產者繞回到數(shù)組的頭部。
如果只有一個生產者和一個消費者,，那么就可以做到免鎖訪問環(huán)形緩沖區(qū)（RingBuffer）,。寫入索引只允許生產者訪問并修改，只要寫入者在更新索引之前將新的值保存到緩沖區(qū)中,，則讀者將始終看到一致的數(shù)據(jù)結構,。同理，讀取索引也只允許消費者訪問并修改,。
圖2,。環(huán)形緩沖區(qū)實現(xiàn)原理圖
如圖所示，當讀者和寫者指針相等時，表明緩沖區(qū)是空的,，而只要寫入指針在讀取指針后面時,，表明緩沖區(qū)已滿。
清單9,。2,。6。10環(huán)形緩沖區(qū)實現(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內核,，通過代碼的注釋（斜體部分）可以看出,，當只有一個消費者和一個生產者時，可以不用添加任何額外的鎖,，就能達到對共享數(shù)據(jù)的訪問,。
總結
通過對比2。4和2,。6內核代碼,，不得不佩服內核開發(fā)者的智慧,，為了提高內核性能，一直不斷的進行各種優(yōu)化,，并將業(yè)界最新的lock-free理念運用到內核中,。
在實際開發(fā)過程中，進行無鎖設計時,，首先進行場景分析,，因為每種無鎖方案都有特定的應用場景，接著根據(jù)場景分析進行數(shù)據(jù)結構的初步設計,，然后根據(jù)先前的分析結果進行并發(fā)模型建模,，最后在調整數(shù)據(jù)結構的設計，以便達到最優(yōu),。