semaphore(Semaphore的原理和實現方法)

semaphore

數據結構

//Go語言中暴露的semaphore實現//具體的用法是提供sleep和wakeup原語//以使其能夠在其它同步原語中的競爭情況下使用//因此這里的semaphore和Linux中的futex目標是一致的//只不過語義上更簡單一些////也就是說,不要認為這些是信號量//把這里的東西看作sleep和wakeup實現的一種方式//每一個sleep都會和一個wakeup配對//即使在發生race時,wakeup在sleep之前時也是如此////SeeMullenderandCox,``SemaphoresinPlan9,''//http://swtch.com/semaphore.pdf//為sync.Mutex準備的異步信號量//semaRoot持有一棵地址各不相同的sudog(s.elem)的平衡樹//每一個sudog都反過來指向(通過s.waitlink)一個在同一個地址上等待的其它sudog們//同一地址的sudog的內部列表上的操作時間復雜度都是O(1)。頂層semaRoot列表的掃描//的時間復雜度是O(logn),n是被哈希到同一個semaRoot的不同地址的總數,每一個地址上都會有一些goroutine被阻塞。//訪問golang.org/issue/17953來查看一個在引入二級列表之前性能較差的程序樣例,test/locklinear.go//中有一個復現這個樣例的測試typesemaRootstruct{lockmutextreap*sudog//rootofbalancedtreeofuniquewaiters.nwaituint32//Numberofwaiters.Readw/othelock.}//Primetonotcorrelatewithanyuserpatterns.constsemTabSize=251varsemtable[semTabSize]struct{rootsemaRootpad[sys.CacheLineSize-unsafe.Sizeof(semaRoot{})]byte}funcsemroot(addr*uint32)*semaRoot{return&semtable[(uintptr(unsafe.Pointer(addr))>>3)%semTabSize].root}

┌─────┬─────┬─────┬─────┬─────┬────────────────────────┬─────┐│0│1│2│3│4│.....│250│└─────┴─────┴─────┴─────┴─────┴────────────────────────┴─────┘││││└──┐└─┐││││▼▼┌─────────┐┌─────────┐│struct││struct│├─────────┴─────────┐├─────────┴─────────┐│rootsemaRoot│──┐│rootsemaRoot│──┐├───────────────────┤│├───────────────────┤││pad│││pad││└───────────────────┘│└───────────────────┘│││┌────────────────┘┌────────────────┘││││▼▼┌──────────┐┌──────────┐│semaRoot││semaRoot│├──────────┴────────┐├──────────┴────────┐│lockmutex││lockmutex│├───────────────────┤├───────────────────┤│treap*sudog││treap*sudog│├───────────────────┤├───────────────────┤│nwaituint32││nwaituint32│└───────────────────┘└───────────────────┘

treap 結構:

┌──────────┐┌─┬─?│sudog│││├──────────┴────────────┐┌─────────────────────┼─┼──│prev*sudog││││├───────────────────────┤││││next*sudog│────┐│││├───────────────────────┤│││││parent*sudog│││││├───────────────────────┤│││││elemunsafe.Pointer│││││├───────────────────────┤│││││ticketuint32│││││└───────────────────────┘│││││││││││││││││││││││││▼││▼┌──────────┐││┌──────────┐│sudog││││sudog│├──────────┴────────────┐││├──────────┴────────────┐│prev*sudog││││prev*sudog│├───────────────────────┤││├───────────────────────┤│next*sudog││││next*sudog│├───────────────────────┤││├───────────────────────┤│parent*sudog│───┘└─────────────────────────│parent*sudog│├───────────────────────┤├───────────────────────┤│elemunsafe.Pointer││elemunsafe.Pointer│├───────────────────────┤├───────────────────────┤│ticketuint32││ticketuint32│└───────────────────────┘└───────────────────────┘

在這個 treap 結構里,從 elem 的視角(其實就是 lock 的 addr)來看,這個結構是個二叉搜索樹。從 ticket 的角度來看,整個結構就是一個小頂堆。

所以才叫樹堆(treap)。

相同 addr,即對同一個 mutex 上鎖的 g,會阻塞在同一個地址上。這些阻塞在同一個地址上的 goroutine 會被打包成 sudog,組成一個鏈表。用 sudog 的 waitlink 相連:

┌──────────┐┌──────────┐┌──────────┐│sudog│┌─────?│sudog│┌─────?│sudog│├──────────┴────────────┐│├──────────┴────────────┐│├──────────┴────────────┐│waitlink*sudog│─────┘│waitlink*sudog│──────┘│waitlink*sudog│├───────────────────────┤├───────────────────────┤├───────────────────────┤│waittail*sudog││waittail*sudog││waittail*sudog│└───────────────────────┘└───────────────────────┘└───────────────────────┘

中間的元素的 waittail 都會指向最后一個元素:

┌──────────┐│sudog│├──────────┴────────────┐│waitlink*sudog│├───────────────────────┤│waittail*sudog│───────────────────────────────────────────────────────────┐└───────────────────────┘│┌──────────┐││sudog││├──────────┴────────────┐││waitlink*sudog││├───────────────────────┤││waittail*sudog│─────────────────────────┤└───────────────────────┘▼┌──────────┐│sudog│├──────────┴────────────┐│waitlink*sudog│├───────────────────────┤│waittail*sudog│└───────────────────────┘

對外封裝

在 sema.go 里實現的內容,用 go:linkname 導出給 sync、poll 庫來使用,也是在鏈接期做了些手腳:

//go:linknamesync_runtime_Semacquiresync.runtime_Semacquirefuncsync_runtime_Semacquire(addr*uint32){semacquire1(addr,false,semaBlockProfile)}//go:linknamepoll_runtime_Semacquireinternal/poll.runtime_Semacquirefuncpoll_runtime_Semacquire(addr*uint32){semacquire1(addr,false,semaBlockProfile)}//go:linknamesync_runtime_Semreleasesync.runtime_Semreleasefuncsync_runtime_Semrelease(addr*uint32,handoffbool){semrelease1(addr,handoff)}//go:linknamesync_runtime_SemacquireMutexsync.runtime_SemacquireMutexfuncsync_runtime_SemacquireMutex(addr*uint32,lifobool){semacquire1(addr,lifo,semaBlockProfile|semaMutexProfile)}//go:linknamepoll_runtime_Semreleaseinternal/poll.runtime_Semreleasefuncpoll_runtime_Semrelease(addr*uint32){semrelease(addr)}

實現

sem 本身支持 acquire 和 release,其實就是 OS 里常說的 P 操作和 V 操作。

公共部分

funccansemacquire(addr*uint32)bool{for{v:=atomic.Load(addr)ifv==0{returnfalse}ifatomic.Cas(addr,v,v-1){returntrue}}}

acquire 過程

typesemaProfileFlagsintconst(semaBlockProfilesemaProfileFlags=1<<iotasemaMutexProfile)//Calledfromruntime.funcsemacquire(addr*uint32){semacquire1(addr,false,0)}funcsemacquire1(addr*uint32,lifobool,profilesemaProfileFlags){gp:=getg()ifgp!=gp.m.curg{throw("semacquirenotontheGstack")}//低成本的情況ifcansemacquire(addr){return}//高成本的情況://增加waitercount的值//再嘗試調用一次cansemacquire,成本了就直接返回//沒成功就把自己作為一個waiter入隊//sleep//(之后waiter的descriptor被signaler用dequeue踢出)s:=acquireSudog()root:=semroot(addr)t0:=int64(0)s.releasetime=0s.acquiretime=0s.ticket=0for{lock(&root.lock)//給nwait加一,這樣后來的就不會在semrelease中進低成本的路徑了atomic.Xadd(&root.nwait,1)//檢查cansemacquire避免錯過了喚醒ifcansemacquire(addr){atomic.Xadd(&root.nwait,-1)unlock(&root.lock)break}//在cansemacquire之后的semrelease都可以知道我們正在等待//(上面設置了nwait),所以會直接進入sleep//注:這里說的sleep其實就是goparkunlockroot.queue(addr,s,lifo)goparkunlock(&root.lock,"semacquire",traceEvGoBlockSync,4)ifs.ticket!=0||cansemacquire(addr){break}}ifs.releasetime>0{blockevent(s.releasetime-t0,3)}releaseSudog(s)}

release 過程

funcsemrelease(addr*uint32){semrelease1(addr,false)}funcsemrelease1(addr*uint32,handoffbool){root:=semroot(addr)atomic.Xadd(addr,1)//低成本情況:沒有waiter?//這個atomic的檢查必須發生在xadd之前,以避免錯誤喚醒//(具體參見semacquire中的循環)ifatomic.Load(&root.nwait)==0{return}//高成本情況:搜索waiter并喚醒它lock(&root.lock)ifatomic.Load(&root.nwait)==0{//count值已經被另一個goroutine消費了//所以我們不需要喚醒其它goroutine了unlock(&root.lock)return}s,t0:=root.dequeue(addr)ifs!=nil{atomic.Xadd(&root.nwait,-1)}unlock(&root.lock)ifs!=nil{//可能會很慢,所以先解鎖acquiretime:=s.acquiretimeifacquiretime!=0{mutexevent(t0-acquiretime,3)}ifs.ticket!=0{throw("corruptedsemaphoreticket")}ifhandoff&&cansemacquire(addr){s.ticket=1}readyWithTime(s,5)}}funcreadyWithTime(s*sudog,traceskipint){ifs.releasetime!=0{s.releasetime=cputicks()}goready(s.g,traceskip)}

treap 結構

sudog 按照地址 hash 到 251 個 bucket 中的其中一個,每一個 bucket 都是一棵 treap。而相同 addr 上的 sudog 會形成一個鏈表。

為啥同一個地址的 sudog 不需要展開放在 treap 中呢?顯然,sudog 喚醒的時候,block 在同一個 addr 上的 goroutine,說明都是加的同一把鎖,這些 goroutine 被喚醒肯定是一起被喚醒的,相同地址的 g 并不需要查找才能找到,只要決定是先進隊列的被喚醒(fifo)還是后進隊列的被喚醒(lifo)就可以了。

//queue函數會把s添加到semaRoot上阻塞的goroutine們中//實際上就是把s添加到其地址對應的treap上func(root*semaRoot)queue(addr*uint32,s*sudog,lifobool){s.g=getg()s.elem=unsafe.Pointer(addr)s.next=nils.prev=nilvarlast*sudogpt:=&root.treapfort:=*pt;t!=nil;t=*pt{ift.elem==unsafe.Pointer(addr){//Alreadyhaveaddrinlist.iflifo{//treap中在t的位置用s覆蓋掉t*pt=ss.ticket=t.tickets.acquiretime=t.acquiretimes.parent=t.parents.prev=t.prevs.next=t.nextifs.prev!=nil{s.prev.parent=s}ifs.next!=nil{s.next.parent=s}//把t放在s的waitlist的第一個位置s.waitlink=ts.waittail=t.waittailifs.waittail==nil{s.waittail=t}t.parent=nilt.prev=nilt.next=nilt.waittail=nil}else{//把s添加到t的等待列表的末尾ift.waittail==nil{t.waitlink=s}else{t.waittail.waitlink=s}t.waittail=ss.waitlink=nil}return}last=tifuintptr(unsafe.Pointer(addr))<uintptr(t.elem){pt=&t.prev}else{pt=&t.next}}//把s作為樹的新的葉子插入進去//平衡樹使用ticket作為堆的權重值,這個ticket是隨機生成的//也就是說,這個結構以元素地址來看的話,是一個二叉搜索樹//同時用ticket值使其同時又是一個小頂堆,滿足//s.ticket<=boths.prev.ticketands.next.ticket.//https://en.wikipedia.org/wiki/Treap//http://faculty.washington.edu/aragon/pubs/rst89.pdf////s.ticket會在一些地方和0相比,因此只設置最低位的bit//這樣不會明顯地影響treap的質量?s.ticket=fastrand()|1s.parent=last*pt=s//按照ticket來進行旋轉,以滿足treap的性質fors.parent!=nil&&s.parent.ticket>s.ticket{ifs.parent.prev==s{root.rotateRight(s.parent)}else{ifs.parent.next!=s{panic("semaRootqueue")}root.rotateLeft(s.parent)}}}//dequeue會搜索到阻塞在addr地址的semaRoot中的第一個goroutine//如果這個sudog需要進行profile,dequeue會返回它被喚醒的時間(now),否則的話now為0func(root*semaRoot)dequeue(addr*uint32)(found*sudog,nowint64){ps:=&root.treaps:=*psfor;s!=nil;s=*ps{ifs.elem==unsafe.Pointer(addr){gotoFound}ifuintptr(unsafe.Pointer(addr))<uintptr(s.elem){ps=&s.prev}else{ps=&s.next}}returnnil,0Found:now=int64(0)ifs.acquiretime!=0{now=cputicks()}ift:=s.waitlink;t!=nil{//替換掉同樣在addr上等待的t。*ps=tt.ticket=s.tickett.parent=s.parentt.prev=s.previft.prev!=nil{t.prev.parent=t}t.next=s.nextift.next!=nil{t.next.parent=t}ift.waitlink!=nil{t.waittail=s.waittail}else{t.waittail=nil}t.acquiretime=nows.waitlink=nils.waittail=nil}else{//向下旋轉s到葉節點,以進行刪除,同時要考慮優先級fors.next!=nil||s.prev!=nil{ifs.next==nil||s.prev!=nil&&s.prev.ticket<s.next.ticket{root.rotateRight(s)}else{root.rotateLeft(s)}}//Removes,nowaleaf.//刪除s,現在是葉子節點了ifs.parent!=nil{ifs.parent.prev==s{s.parent.prev=nil}else{s.parent.next=nil}}else{root.treap=nil}}s.parent=nils.elem=nils.next=nils.prev=nils.ticket=0returns,now}

“Semaphore的原理和實現方法”的內容就介紹到這里了,感謝大家的閱讀。如果想了解更多行業相關的知識可以關注本站網站,小編將為大家輸出更多高質量的實用文章!