sync.Cond
前言
本次的代碼是基於go version go1.13.15 darwin/amd64
什么是sync.Cond
Go語言標准庫中的條件變量sync.Cond，它可以讓一組的Goroutine都在滿足特定條件時被喚醒。
每個Cond都會關聯一個Lock（*sync.Mutex or *sync.RWMutex）
var (
locker = new(sync.Mutex)
cond = sync.NewCond(locker)
)
func listen(x int) {
// 獲取鎖
cond.L.Lock()
// 等待通知 暫時阻塞
cond.Wait()
fmt.Println(x)
// 釋放鎖
cond.L.Unlock()
}
func main() {
// 啟動60個被cond阻塞的線程
for i := 1; i <= 60; i++ {
go listen(i)
}
fmt.Println("start all")
// 3秒之后 下發一個通知給已經獲取鎖的goroutine time.Sleep(time.Second * 3)
fmt.Println("++++++++++++++++++++one Signal")
cond.Signal()
// 3秒之后 下發一個通知給已經獲取鎖的goroutine
time.Sleep(time.Second * 3)
fmt.Println("++++++++++++++++++++one Signal")
cond.Signal()
// 3秒之后 下發廣播給所有等待的goroutine
time.Sleep(time.Second * 3)
fmt.Println("++++++++++++++++++++begin broadcast")
cond.Broadcast()
// 阻塞直到所有的全部輸出
time.Sleep(time.Second * 60)
}
上面是個簡單的例子，我們啟動了60個線程，然后都被cond阻塞，主函數通過Signal()通知一個goroutine接觸阻塞，通過Broadcast()通知所有被阻塞的全部解除阻塞。
看下源碼
// Wait 原子式的 unlock c.L， 並暫停執行調用的 goroutine。
// 在稍后執行后，Wait 會在返回前 lock c.L. 與其他系統不同，
// 除非被 Broadcast 或 Signal 喚醒，否則等待無法返回。
//
// 因為等待第一次 resume 時 c.L 沒有被鎖定，所以當 Wait 返回時，
// 調用者通常不能認為條件為真。相反，調用者應該在循環中使用 Wait()：
//
// c.L.Lock()
// for !condition() {
// c.Wait()
// }
// ... make use of condition ...
// c.L.Unlock()
//
type Cond struct {
// 用於保證結構體不會在編譯期間拷貝
noCopy noCopy
// 鎖
L Locker
// goroutine鏈表，維護等待喚醒的goroutine隊列
notify notifyList
// 保證運行期間不會發生copy
checker copyChecker
}
重點分析下：notifyList和copyChecker
- notify
type notifyList struct {
// 總共需要等待的數量
wait uint32
// 已經通知的數量
notify uint32
// 鎖
lock uintptr
// 指向鏈表頭部
head *sudog
// 指向鏈表尾部
tail *sudog
}
這個是核心，所有wait的goroutine都會被加入到這個鏈表中，然后在通知的時候再從這個鏈表中獲取。
- copyChecker
保證運行期間不會發生copy
type copyChecker uintptr
// copyChecker holds back pointer to itself to detect object copying
func (c *copyChecker) check() {
if uintptr(*c) != uintptr(unsafe.Pointer(c)) &&
!atomic.CompareAndSwapUintptr((*uintptr)(c), 0, uintptr(unsafe.Pointer(c))) &&
uintptr(*c) != uintptr(unsafe.Pointer(c)) {
panic("sync.Cond is copied")
}
}
Wait
func (c *Cond) Wait() {
// 監測是否復制
c.checker.check()
// 更新 notifyList中需要等待的wait的數量
// 返回當前需要插入鏈表節點ticket
t := runtime_notifyListAdd(&c.notify)
c.L.Unlock()
// 為當前的加入的waiter構建一個鏈表的節點，插入鏈表的尾部
runtime_notifyListWait(&c.notify, t)
c.L.Lock()
}
// go/src/runtime/sema.go
// 更新 notifyList中需要等待的wait的數量
// 同時返回當前的加入的 waiter 的 ticket 編號，從0開始
//go:linkname notifyListAdd sync.runtime_notifyListAdd
func notifyListAdd(l *notifyList) uint32 {
// 使用atomic原子的對wait字段進行加一操作
return atomic.Xadd(&l.wait, 1) - 1
}
// go/src/runtime/sema.go
// 為當前的加入的waiter構建一個鏈表的節點，插入鏈表的尾部
//go:linkname notifyListWait sync.runtime_notifyListWait
func notifyListWait(l *notifyList, t uint32) {
lock(&l.lock)
// 當t小於notifyList中的notify，說明當前節點已經被通知了
if less(t, l.notify) {
unlock(&l.lock)
return
}
// 構建當前節點
s := acquireSudog()
s.g = getg()
s.ticket = t
s.releasetime = 0
t0 := int64(0)
if blockprofilerate > 0 {
t0 = cputicks()
s.releasetime = -1
}
// 頭結點沒構建，插入頭結點
if l.tail == nil {
l.head = s
} else {
// 插入到尾節點
l.tail.next = s
}
l.tail = s
// 將當前goroutine置於等待狀態並解鎖
// 通過調用goready（gp），可以使goroutine再次可運行。
// 也就是將 M/P/G 解綁，並將 G 調整為等待狀態，放入 sudog 等待隊列中
goparkunlock(&l.lock, waitReasonSyncCondWait, traceEvGoBlockCond, 3)
if t0 != 0 {
blockevent(s.releasetime-t0, 2)
}
releaseSudog(s)
}
梳理流程
1、首先檢測對象的復制行為，如果有復制發生直接拋出panic；
2、然后調用runtime_notifyListAdd對notifynotifyListList中的wait(需要等待的數量)進行加一操作，同時返回一個ticket，用來作為當前wait的編號，這個編號，會和notifyList中的notify對應起來；
3、然后調用runtime_notifyListWait把當前的wait封裝成鏈表的一個節點，插入到notifyList維護的鏈表的尾部。
Signal
// 喚醒一個被wait的goroutine
func (c *Cond) Signal() {
// 監測是否復制
c.checker.check()
runtime_notifyListNotifyOne(&c.notify)
}
// go/src/runtime/sema.go
// 通知鏈表中的第一個
//go:linkname notifyListNotifyOne sync.runtime_notifyListNotifyOne
func notifyListNotifyOne(l *notifyList) {
// wait和notify，說明已經全部通知到了
if atomic.Load(&l.wait) == atomic.Load(&l.notify) {
return
}
lock(&l.lock)
// 這里做了二次的確認
// wait和notify，說明已經全部通知到了
t := l.notify
if t == atomic.Load(&l.wait) {
unlock(&l.lock)
return
}
// 原子的對notify執行+1操作
atomic.Store(&l.notify, t+1)
// 嘗試找到需要被通知的 g
// 如果目前還沒來得及入隊，是無法找到的
// 但是，當它看到通知編號已經發生改變是不會被 park 的
//
// 這個查找過程看起來是線性復雜度，但實際上很快就停了
// 因為 g 的隊列與獲取編號不同，因而隊列中會出現少量重排，但我們希望找到靠前的 g
// 而 g 只有在不再 race 后才會排在靠前的位置，因此這個迭代也不會太久，
// 同時，即便找不到 g，這個情況也成立：
// 它還沒有休眠，並且已經失去了我們在隊列上找到的（少數）其他 g 的 race。
for p, s := (*sudog)(nil), l.head; s != nil; p, s = s, s.next {
// 順序拿到一個節點的ticket，會和上面會和notifyList中的notify做比較，相同才進行后續的操作
// 這個我們分析了，notifyList中的notify和鏈表節點中的ticket是一一對應的
if s.ticket == t {
n := s.next
if p != nil {
p.next = n
} else {
l.head = n
}
if n == nil {
l.tail = p
}
unlock(&l.lock)
s.next = nil
// 通過goready掉起在上面通過goparkunlock掛起的goroutine
readyWithTime(s, 4)
return
}
}
unlock(&l.lock)
}
梳理下流程：
1、首先檢測對象的復制行為，如果有復制發生直接拋出panic；
2、判斷wait和notify，如果兩者相同說明已經已經全部通知到了；
3、調用notifyListNotifyOne，通過for循環，依次遍歷這個鏈表，直到找到和notifyList中的notify，相匹配的ticket的節點；
4、掉起goroutine，完成通知。
Broadcast
// 喚醒所有被wait的goroutine
func (c *Cond) Broadcast() {
c.checker.check()
runtime_notifyListNotifyAll(&c.notify)
}
// go/src/runtime/sema.go
// notifyListNotifyAll notifies all entries in the list.
//go:linkname notifyListNotifyAll sync.runtime_notifyListNotifyAll
func notifyListNotifyAll(l *notifyList) {
// wait和notify，說明已經全部通知到了
if atomic.Load(&l.wait) == atomic.Load(&l.notify) {
return
}
// 加鎖
lock(&l.lock)
s := l.head
l.head = nil
l.tail = nil
// 這個很粗暴，直接將notify的值置換成wait
atomic.Store(&l.notify, atomic.Load(&l.wait))
unlock(&l.lock)
// 循環鏈表，一個個喚醒goroutine
for s != nil {
next := s.next
s.next = nil
readyWithTime(s, 4)
s = next
}
}
梳理下流程：
1、首先檢測對象的復制行為，如果有復制發生直接拋出panic；
2、判斷wait和notify，如果兩者相同說明已經已經全部通知到了；
3、notifyListNotifyAll，就相對簡單了，直接將notify的值置為wait，標注這個已經全部通知了；
4、循環鏈表，一個個喚醒goroutine。
總結
sync.Cond不是一個常用的同步機制，但是在條件長時間無法滿足時，與使用for {}進行忙碌等待相比,sync.Cond能夠讓出處理器的使用權，提供CPU的利用率。使用時我們也需要注意以下問題：
1、sync.Cond.Wait在調用之前一定要使用獲取互斥鎖，否則會觸發程序崩潰；
2、sync.Cond.Signal 喚醒的 Goroutine都是隊列最前面、等待最久的Goroutine；
3、sync.Cond.Broadcast會按照一定順序廣播通知等待的全部 Goroutine。
版權聲明：本文為博主原創文章，遵循 CC 4.0 BY-SA 版權協議，轉載請附上原文出處鏈接和本聲明。
本文鏈接：https://www.cnblogs.com/ricklz/p/14515563.html