一文讀懂go中semaphore(信號量)源碼
運行時信號量機制 semaphore
前言
最近在看源碼,發(fā)現(xiàn)好多地方用到了這個semaphore。
本文是在go version go1.13.15 darwin/amd64上進行的
作用是什么
下面是官方的描述
// Semaphore implementation exposed to Go. // Intended use is provide a sleep and wakeup // primitive that can be used in the contended case // of other synchronization primitives. // Thus it targets the same goal as Linux's futex, // but it has much simpler semantics. // // That is, don't think of these as semaphores. // Think of them as a way to implement sleep and wakeup // such that every sleep is paired with a single wakeup, // even if, due to races, the wakeup happens before the sleep. // 具體的用法是提供 sleep 和 wakeup 原語 // 以使其能夠在其它同步原語中的競爭情況下使用 // 因此這里的 semaphore 和 Linux 中的 futex 目標(biāo)是一致的 // 只不過語義上更簡單一些 // // 也就是說,不要認為這些是信號量 // 把這里的東西看作 sleep 和 wakeup 實現(xiàn)的一種方式 // 每一個 sleep 都會和一個 wakeup 配對 // 即使在發(fā)生 race 時,wakeup 在 sleep 之前時也是如此
上面提到了和futex作用一樣,關(guān)于futex
futex(快速用戶區(qū)互斥的簡稱)是一個在Linux上實現(xiàn)鎖定和構(gòu)建高級抽象鎖如信號量和POSIX互斥的基本工具
Futex 由一塊能夠被多個進程共享的內(nèi)存空間(一個對齊后的整型變量)組成;這個整型變量的值能夠通過匯編語言調(diào)用CPU提供的原子操作指令來增加或減少,并且一個進程可以等待直到那個值變成正數(shù)。Futex 的操作幾乎全部在用戶空間完成;只有當(dāng)操作結(jié)果不一致從而需要仲裁時,才需要進入操作系統(tǒng)內(nèi)核空間執(zhí)行。這種機制允許使用 futex 的鎖定原語有非常高的執(zhí)行效率:由于絕大多數(shù)的操作并不需要在多個進程之間進行仲裁,所以絕大多數(shù)操作都可以在應(yīng)用程序空間執(zhí)行,而不需要使用(相對高代價的)內(nèi)核系統(tǒng)調(diào)用。
go中的semaphore作用和futex目標(biāo)一樣,提供sleep和wakeup原語,使其能夠在其它同步原語中的競爭情況下使用。當(dāng)一個goroutine需要休眠時,將其進行集中存放,當(dāng)需要wakeup時,再將其取出,重新放入調(diào)度器中。
例如在讀寫鎖的實現(xiàn)中,讀鎖和寫鎖之前的相互阻塞喚醒,就是通過sleep和wakeup實現(xiàn),當(dāng)有讀鎖存在的時候,新加入的寫鎖通過semaphore阻塞自己,當(dāng)前面的讀鎖完成,在通過semaphore喚醒被阻塞的寫鎖。
寫鎖
// 獲取互斥鎖
// 阻塞等待所有讀操作結(jié)束(如果有的話)
func (rw *RWMutex) Lock() {
...
// 原子的修改readerCount的值,直接將readerCount減去rwmutexMaxReaders
// 說明,有寫鎖進來了,這在上面的讀鎖中也有體現(xiàn)
r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
// 當(dāng)r不為0說明,當(dāng)前寫鎖之前有讀鎖的存在
// 修改下readerWait,也就是當(dāng)前寫鎖需要等待的讀鎖的個數(shù)
if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
// 阻塞當(dāng)前寫鎖
runtime_SemacquireMutex(&rw.writerSem, false, 0)
}
...
}
通過runtime_SemacquireMutex對當(dāng)前寫鎖進行sleep
讀鎖釋放
// 減少讀操作計數(shù),即readerCount--
// 喚醒等待寫操作的協(xié)程(如果有的話)
func (rw *RWMutex) RUnlock() {
...
// 首先通過atomic的原子性使readerCount-1
// 1.若readerCount大于0, 證明當(dāng)前還有讀鎖, 直接結(jié)束本次操作
// 2.若readerCount小于0, 證明已經(jīng)沒有讀鎖, 但是還有因為讀鎖被阻塞的寫鎖存在
if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
// 嘗試喚醒被阻塞的寫鎖
rw.rUnlockSlow(r)
}
...
}
func (rw *RWMutex) rUnlockSlow(r int32) {
...
// readerWait--操作,如果readerWait--操作之后的值為0,說明,寫鎖之前,已經(jīng)沒有讀鎖了
// 通過writerSem信號量,喚醒隊列中第一個阻塞的寫鎖
if atomic.AddInt32(&rw.readerWait, -1) == 0 {
// 喚醒一個寫鎖
runtime_Semrelease(&rw.writerSem, false, 1)
}
}
寫鎖處理完之后,調(diào)用runtime_Semrelease來喚醒sleep的寫鎖
幾個主要的方法
在go/src/sync/runtime.go中,定義了這幾個方法
// Semacquire等待*s > 0,然后原子遞減它。 // 它是一個簡單的睡眠原語,用于同步 // library and不應(yīng)該直接使用。 func runtime_Semacquire(s *uint32) // SemacquireMutex類似于Semacquire,用來阻塞互斥的對象 // 如果lifo為true,waiter將會被插入到隊列的頭部 // skipframes是跟蹤過程中要省略的幀數(shù),從這里開始計算 // runtime_SemacquireMutex's caller. func runtime_SemacquireMutex(s *uint32, lifo bool, skipframes int) // Semrelease會自動增加*s并通知一個被Semacquire阻塞的等待的goroutine // 它是一個簡單的喚醒原語,用于同步 // library and不應(yīng)該直接使用。 // 如果handoff為true, 傳遞信號到隊列頭部的waiter // skipframes是跟蹤過程中要省略的幀數(shù),從這里開始計算 // runtime_Semrelease's caller. func runtime_Semrelease(s *uint32, handoff bool, skipframes int)
具體的實現(xiàn)是在go/src/runtime/sema.go中
//go:linkname sync_runtime_Semacquire sync.runtime_Semacquire
func sync_runtime_Semacquire(addr *uint32) {
semacquire1(addr, false, semaBlockProfile, 0)
}
//go:linkname sync_runtime_Semrelease sync.runtime_Semrelease
func sync_runtime_Semrelease(addr *uint32, handoff bool, skipframes int) {
semrelease1(addr, handoff, skipframes)
}
//go:linkname sync_runtime_SemacquireMutex sync.runtime_SemacquireMutex
func sync_runtime_SemacquireMutex(addr *uint32, lifo bool, skipframes int) {
semacquire1(addr, lifo, semaBlockProfile|semaMutexProfile, skipframes)
}
如何實現(xiàn)
sudog 緩存
semaphore的實現(xiàn)使用到了sudog,我們先來看下
sudog 是運行時用來存放處于阻塞狀態(tài)的goroutine的一個上層抽象,是用來實現(xiàn)用戶態(tài)信號量的主要機制之一。 例如當(dāng)一個goroutine因為等待channel的數(shù)據(jù)需要進行阻塞時,sudog會將goroutine及其用于等待數(shù)據(jù)的位置進行記錄, 并進而串聯(lián)成一個等待隊列,或二叉平衡樹。
// sudogs are allocated from a special pool. Use acquireSudog and
// releaseSudog to allocate and free them.
type sudog struct {
// 以下字段受hchan保護
g *g
// isSelect 表示 g 正在參與一個 select, so
// 因此 g.selectDone 必須以 CAS 的方式來獲取wake-up race.
isSelect bool
next *sudog
prev *sudog
elem unsafe.Pointer // 數(shù)據(jù)元素(可能指向棧)
// 以下字段不會并發(fā)訪問。
// 對于通道,waitlink只被g訪問。
// 對于信號量,所有字段(包括上面的字段)
// 只有當(dāng)持有一個semroot鎖時才被訪問。
acquiretime int64
releasetime int64
ticket uint32
parent *sudog //semaRoot 二叉樹
waitlink *sudog // g.waiting 列表或 semaRoot
waittail *sudog // semaRoot
c *hchan // channel
}
sudog的獲取和歸還,遵循以下策略:
1、獲取,首先從per-P緩存獲取,對于per-P緩存,如果per-P緩存為空,則從全局池抓取一半,然后取出per-P緩存中的最后一個;
2、歸還,歸還到per-P緩存,如果per-P緩存滿了,就把per-P緩存的一半歸還到全局緩存中,然后歸還sudog到per-P緩存中。
acquireSudog
1、如果per-P緩存的內(nèi)容沒達到長度的一般,則會從全局額緩存中抓取一半;
2、然后返回把per-P緩存中最后一個sudog返回,并且置空;
// go/src/runtime/proc.go
//go:nosplit
func acquireSudog() *sudog {
// Delicate dance: 信號量的實現(xiàn)調(diào)用acquireSudog,然后acquireSudog調(diào)用new(sudog)
// new調(diào)用malloc, malloc調(diào)用垃圾收集器,垃圾收集器在stopTheWorld調(diào)用信號量
// 通過在new(sudog)周圍執(zhí)行acquirem/releasem來打破循環(huán)
// acquirem/releasem在new(sudog)期間增加m.locks,防止垃圾收集器被調(diào)用。
// 獲取當(dāng)前 g 所在的 m
mp := acquirem()
// 獲取p的指針
pp := mp.p.ptr()
if len(pp.sudogcache) == 0 {
lock(&sched.sudoglock)
// 首先,嘗試從中央緩存獲取一批數(shù)據(jù)。
for len(pp.sudogcache) < cap(pp.sudogcache)/2 && sched.sudogcache != nil {
s := sched.sudogcache
sched.sudogcache = s.next
s.next = nil
pp.sudogcache = append(pp.sudogcache, s)
}
unlock(&sched.sudoglock)
// 如果中央緩存中沒有,新分配
if len(pp.sudogcache) == 0 {
pp.sudogcache = append(pp.sudogcache, new(sudog))
}
}
// 取緩存中最后一個
n := len(pp.sudogcache)
s := pp.sudogcache[n-1]
pp.sudogcache[n-1] = nil
// 將剛?cè)〕龅脑诰彺嬷幸瞥?
pp.sudogcache = pp.sudogcache[:n-1]
if s.elem != nil {
throw("acquireSudog: found s.elem != nil in cache")
}
releasem(mp)
return s
}
releaseSudog
1、如果per-P緩存滿了,就歸還per-P緩存一般的內(nèi)容到全局緩存;
2、然后將回收的sudog放到per-P緩存中。
// go/src/runtime/proc.go
//go:nosplit
func releaseSudog(s *sudog) {
if s.elem != nil {
throw("runtime: sudog with non-nil elem")
}
if s.isSelect {
throw("runtime: sudog with non-false isSelect")
}
if s.next != nil {
throw("runtime: sudog with non-nil next")
}
if s.prev != nil {
throw("runtime: sudog with non-nil prev")
}
if s.waitlink != nil {
throw("runtime: sudog with non-nil waitlink")
}
if s.c != nil {
throw("runtime: sudog with non-nil c")
}
gp := getg()
if gp.param != nil {
throw("runtime: releaseSudog with non-nil gp.param")
}
// 避免重新安排到另一個P
mp := acquirem() // avoid rescheduling to another P
pp := mp.p.ptr()
// 如果緩存滿了
if len(pp.sudogcache) == cap(pp.sudogcache) {
// 將本地高速緩存的一半傳輸?shù)街醒敫咚倬彺?
var first, last *sudog
for len(pp.sudogcache) > cap(pp.sudogcache)/2 {
n := len(pp.sudogcache)
p := pp.sudogcache[n-1]
pp.sudogcache[n-1] = nil
pp.sudogcache = pp.sudogcache[:n-1]
if first == nil {
first = p
} else {
last.next = p
}
last = p
}
lock(&sched.sudoglock)
last.next = sched.sudogcache
sched.sudogcache = first
unlock(&sched.sudoglock)
}
// 歸還sudog到`per-P`緩存中
pp.sudogcache = append(pp.sudogcache, s)
releasem(mp)
}
semaphore
// go/src/runtime/sema.go
// 用于sync.Mutex的異步信號量。
// semaRoot擁有一個具有不同地址(s.elem)的sudog平衡樹。
// 每個sudog都可以依次(通過s.waitlink)指向一個列表,在相同地址上等待的其他sudog。
// 對具有相同地址的sudog內(nèi)部列表進行的操作全部為O(1)。頂層semaRoot列表的掃描為O(log n),
// 其中,n是阻止goroutines的不同地址的數(shù)量,通過他們散列到給定的semaRoot。
type semaRoot struct {
lock mutex
// waiters的平衡樹的根節(jié)點
treap *sudog
// waiters的數(shù)量,讀取的時候無所
nwait uint32
}
// Prime to not correlate with any user patterns.
const semTabSize = 251
var semtable [semTabSize]struct {
root semaRoot
pad [cpu.CacheLinePadSize - unsafe.Sizeof(semaRoot{})]byte
}
poll_runtime_Semacquire/sync_runtime_SemacquireMutex
// go/src/runtime/sema.go
//go:linkname poll_runtime_Semacquire internal/poll.runtime_Semacquire
func poll_runtime_Semacquire(addr *uint32) {
semacquire1(addr, false, semaBlockProfile, 0)
}
//go:linkname sync_runtime_SemacquireMutex sync.runtime_SemacquireMutex
func sync_runtime_SemacquireMutex(addr *uint32, lifo bool, skipframes int) {
semacquire1(addr, lifo, semaBlockProfile|semaMutexProfile, skipframes)
}
func semacquire1(addr *uint32, lifo bool, profile semaProfileFlags, skipframes int) {
// 判斷這個goroutine,是否是m上正在運行的那個
gp := getg()
if gp != gp.m.curg {
throw("semacquire not on the G stack")
}
// *addr -= 1
if cansemacquire(addr) {
return
}
// 增加等待計數(shù)
// 再試一次 cansemacquire 如果成功則直接返回
// 將自己作為等待者入隊
// 休眠
// (等待器描述符由出隊信號產(chǎn)生出隊行為)
// 獲取一個sudog
s := acquireSudog()
root := semroot(addr)
t0 := int64(0)
s.releasetime = 0
s.acquiretime = 0
s.ticket = 0
if profile&semaBlockProfile != 0 && blockprofilerate > 0 {
t0 = cputicks()
s.releasetime = -1
}
if profile&semaMutexProfile != 0 && mutexprofilerate > 0 {
if t0 == 0 {
t0 = cputicks()
}
s.acquiretime = t0
}
for {
lock(&root.lock)
// 添加我們自己到nwait來禁用semrelease中的"easy case"
atomic.Xadd(&root.nwait, 1)
// 檢查cansemacquire避免錯過喚醒
if cansemacquire(addr) {
atomic.Xadd(&root.nwait, -1)
unlock(&root.lock)
break
}
// 任何在 cansemacquire 之后的 semrelease 都知道我們在等待(因為設(shè)置了 nwait),因此休眠
// 隊列將s添加到semaRoot中被阻止的goroutine中
root.queue(addr, s, lifo)
// 將當(dāng)前goroutine置于等待狀態(tài)并解鎖鎖。
// 通過調(diào)用goready(gp),可以使goroutine再次可運行。
goparkunlock(&root.lock, waitReasonSemacquire, traceEvGoBlockSync, 4+skipframes)
if s.ticket != 0 || cansemacquire(addr) {
break
}
}
if s.releasetime > 0 {
blockevent(s.releasetime-t0, 3+skipframes)
}
// 歸還sudog
releaseSudog(s)
}
func cansemacquire(addr *uint32) bool {
for {
v := atomic.Load(addr)
if v == 0 {
return false
}
if atomic.Cas(addr, v, v-1) {
return true
}
}
}
sync_runtime_Semrelease
// go/src/runtime/sema.go
//go:linkname sync_runtime_Semrelease sync.runtime_Semrelease
func sync_runtime_Semrelease(addr *uint32, handoff bool, skipframes int) {
semrelease1(addr, handoff, skipframes)
}
func semrelease1(addr *uint32, handoff bool, skipframes int) {
root := semroot(addr)
atomic.Xadd(addr, 1)
// Easy case:沒有等待者
// 這個檢查必須發(fā)生在xadd之后,以避免錯過喚醒
if atomic.Load(&root.nwait) == 0 {
return
}
// Harder case: 找到等待者,并且喚醒
lock(&root.lock)
if atomic.Load(&root.nwait) == 0 {
// 該計數(shù)已被另一個goroutine占用,
// 因此無需喚醒其他goroutine。
unlock(&root.lock)
return
}
// 搜索一個等待著然后將其喚醒
s, t0 := root.dequeue(addr)
if s != nil {
atomic.Xadd(&root.nwait, -1)
}
unlock(&root.lock)
if s != nil { // 可能會很慢,因此先解鎖
acquiretime := s.acquiretime
if acquiretime != 0 {
mutexevent(t0-acquiretime, 3+skipframes)
}
if s.ticket != 0 {
throw("corrupted semaphore ticket")
}
if handoff && cansemacquire(addr) {
s.ticket = 1
}
// goready(s.g, 5)
// 標(biāo)記 runnable,等待被重新調(diào)度
readyWithTime(s, 5+skipframes)
}
}
摘自"同步原語"的一段總結(jié)
這一對 semacquire 和 semrelease 理解上可能不太直觀。 首先,我們必須意識到這兩個函數(shù)一定是在兩個不同的 M(線程)上得到執(zhí)行,否則不會出現(xiàn)并發(fā),我們不妨設(shè)為 M1 和 M2。 當(dāng) M1 上的 G1 執(zhí)行到 semacquire1 時,如果快速路徑成功,則說明 G1 搶到鎖,能夠繼續(xù)執(zhí)行。但一旦失敗且在慢速路徑下 依然搶不到鎖,則會進入 goparkunlock,將當(dāng)前的 G1 放到等待隊列中,進而讓 M1 切換并執(zhí)行其他 G。 當(dāng) M2 上的 G2 開始調(diào)用 semrelease1 時,只是單純的將等待隊列的 G1 重新放到調(diào)度隊列中,而當(dāng) G1 重新被調(diào)度時(假設(shè)運氣好又在 M1 上被調(diào)度),代碼仍然會從 goparkunlock 之后開始執(zhí)行,并再次嘗試競爭信號量,如果成功,則會歸還 sudog。
參考
【同步原語】https://golang.design/under-the-hood/zh-cn/part2runtime/ch06sched/sync/
【Go并發(fā)編程實戰(zhàn)--信號量的使用方法和其實現(xiàn)原理】https://juejin.cn/post/6906677772479889422
【Semaphore】https://github.com/cch123/golang-notes/blob/master/semaphore.md
【進程同步之信號量機制(pv操作)及三個經(jīng)典同步問題】https://blog.csdn.net/SpeedMe/article/details/17597373
到此這篇關(guān)于go中semaphore(信號量)源碼解讀的文章就介紹到這了,更多相關(guān)go中semaphore源碼內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Golang回調(diào)函數(shù)與閉包和接口函數(shù)的定義及使用介紹
這篇文章主要介紹了Golang回調(diào)函數(shù)與閉包和接口函數(shù)的定義及使用,文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)吧2023-05-05

