深入理解 go sync-Con
在 go 的標準庫中,提供了 sync.Cond
這個併發原語,讓我們可以實現多個 goroutine
等待某一條件滿足之後再繼續執行。 它需要配合 sync.Mutex
一起使用,因爲 Cond
的 Wait
方法需要在 Mutex
的保護下才能正常工作。 對於條件變量,可能大多數人只是知道它的存在,但是用到它的估計寥寥無幾,因爲很多併發場景的處理都能使用 chan
來實現, 而且 chan
的使用也更加簡單。 但是在某些場景下,Cond
可能是最好的選擇,本文就來探討一下 Cond
的使用場景,基本用法,以及它的實現原理。
sync.Cond 是什麼?
sync.Cond
表示的是條件變量,它是一種同步機制,用來協調多個 goroutine
之間的同步,當共享資源的狀態發生變化的時候, 可以通過條件變量來通知所有等待的 goroutine
去重新獲取共享資源。
適用場景
在實際使用中,我們可能會有多個 goroutine
在執行的過程中,由於某一條件不滿足而阻塞的情況。 這個時候,我們就可以使用條件變量來實現 goroutine
之間的同步。比如,我們有一個 goroutine
用來獲取數據, 但是可能會比較耗時,這個時候,我們就可以使用條件變量來實現 goroutine
之間的同步, 當數據準備好之後,就可以通過條件變量來通知所有等待的 goroutine
去重新獲取共享資源。
sync.Cond
條件變量用來協調想要訪問共享資源的那些goroutine
,當共享資源的狀態發生變化的時候, 它可以用來通知所有等待的goroutine
去重新獲取共享資源。
sync.Cond 的基本用法
sync.Cond
的基本用法非常簡單,我們只需要通過 sync.NewCond
方法來創建一個 Cond
實例, 然後通過 Wait
方法來等待條件滿足,通過 Signal
或者 Broadcast
方法來通知所有等待的 goroutine
去重新獲取共享資源。
NewCond 創建實例
sync.NewCond
方法用來創建一個 Cond
實例,它的參數是一個 Locker
接口,我們可以傳入一個 Mutex
或者 RWMutex
實例。 這個條件變量的 Locker
接口就是用來保護共享資源的。
Wait 等待條件滿足
Wait
方法用來等待條件滿足,它會先釋放 Cond
的鎖 (Cond.L
),然後阻塞當前 goroutine
(實際調用的是 goparkunlock
),直到被 Signal
或者 Broadcast
喚醒。
它做了如下幾件事情:
-
釋放
Cond
的鎖 (Cond.L
),然後阻塞當前goroutine
。(所以,使用之前需要先鎖定) -
被
Signal
或者Broadcast
喚醒之後,會重新獲取Cond
的鎖 (Cond.L
)。 -
之後,就返回到
goroutine
阻塞的地方繼續執行。
Signal 通知一個等待的 goroutine
Signal
方法用來通知一個等待的 goroutine
,它會喚醒一個等待的 goroutine
,然後繼續執行當前 goroutine
。 如果沒有等待的 goroutine
,則不會有任何操作。
Broadcast 通知所有等待的 goroutine
Broadcast
方法用來通知所有等待的 goroutine
,它會喚醒所有等待的 goroutine
,然後繼續執行當前 goroutine
。 如果沒有等待的 goroutine
,則不會有任何操作。
sync.Cond 使用實例
下面我們通過一個實例來看一下 sync.Cond
的使用方法。
package cond
import (
"fmt"
"sync"
"testing"
"time"
)
var done bool
var data string
func write(c *sync.Cond) {
fmt.Println("writing.")
// 讓 reader 先獲取鎖,模擬條件不滿足然後 wait 的情況
time.Sleep(time.Millisecond * 10)
c.L.Lock()
// 模擬耗時的寫操作
time.Sleep(time.Millisecond * 50)
data = "hello world"
done = true
fmt.Println("writing done.")
c.L.Unlock()
c.Broadcast()
}
func read(c *sync.Cond) {
fmt.Println("reading")
c.L.Lock()
for !done {
fmt.Println("reader wait.")
c.Wait()
}
fmt.Println("read done.")
fmt.Println("data:", data)
defer c.L.Unlock()
}
func TestCond(t *testing.T) {
var c = sync.NewCond(&sync.Mutex{})
go read(c) // 讀操作
go read(c) // 讀操作
go write(c) // 寫操作
time.Sleep(time.Millisecond * 100) // 等待操作完成
}
輸出:
reading
reader wait. // 還沒獲取完數據,需要等待
writing.
reading
reader wait.
writing done. // 獲取完數據了,通知所有等待的 reader
read done. // 讀取到數據了
data: hello world // 輸出讀取到的數據
read done.
data: hello world
這個例子可以粗略地用下圖來表示:
說明:
-
read1
和reader2
表示兩個goroutine
,它們都會調用read
函數。 -
在
done
爲false
的時候,reader1
和reader2
都會調用c.Wait()
函數,然後阻塞等待。 -
write
表示一個goroutine
,它會調用write
函數。 -
在
write
函數中,獲取完數據之後,會將done
設置爲true
,然後調用c.Broadcast()
函數,通知所有等待的reader
去重新獲取共享資源。 -
reader1
和reader2
在解除阻塞狀態後,都會重新獲取共享資源,然後輸出讀取到的數據。
在這個例子中,done
的功能是標記,用來表示共享資源是否已經獲取完畢,如果沒有獲取完畢,那麼 reader
就會阻塞等待。
爲什麼要用 sync.Cond?
在文章開頭,我們說了,很多併發編程的問題都可以通過 channel
來解決。 同樣的,在上面提到的 sync.Cond
的使用場景,使用 channel
也是可以實現的, 我們只要 close(ch)
來關閉 channel
就可以實現通知多個等待的協程了。
那麼爲什麼還要用 sync.Cond
呢? 主要原因是,sync.Cond
可以重複地進行 Wait()
和 Signal()
、Broadcast()
操作, 但是,如果想通過關閉 chan
來實現這個功能的話,那就只能通知一次了。 因爲 channel
只能關閉一次,關閉一個已經關閉的 channel
會導致程序 panic。
使用 channel
的另外一種方式是,記錄 reader
的數量,然後通過往 channel
中發送多次數據來實現通知多個 reader
。 但是這樣一來代碼就會複雜很多,從另一個角度說,出錯的概率大了很多。
close channel 廣播實例
下面的例子模擬了使用 close(chan)
來實現 sync.Cond
中那種廣播功能,但是隻能通知一次。
package close_chan
import (
"fmt"
"testing"
"time"
)
var data string
func read(c <-chan struct{}) {
fmt.Println("reading.")
// 從 chan 接收數據,如果 chan 中沒有數據,會阻塞。
// 如果能接收到數據,或者 chan 被關閉,會解除阻塞狀態。
<-c
fmt.Println("data:", data)
}
func write(c chan struct{}) {
fmt.Println("writing.")
// 模擬耗時的寫操作
time.Sleep(time.Millisecond * 10)
data = "hello world"
fmt.Println("write done.")
// 關閉 chan 的時候,會通知所有的 reader
// 所有等待從 chan 接收數據的 goroutine 都會被喚醒
close(c)
}
func TestCloseChan(t *testing.T) {
ch := make(chan struct{})
go read(ch)
go read(ch)
go write(ch)
// 不能關閉已經關閉的 chan
time.Sleep(time.Millisecond * 20)
// panic: close of closed channel
// 下面這行代碼會導致 panic
//go write(ch)
time.Sleep(time.Millisecond * 100)
}
輸出:
writing.
reading. // 會阻塞直到寫完
reading. // 會阻塞直到寫完
write done. // 寫完之後,才能讀
data: hello world
data: hello world
上面例子的 write
不能多次調用,否則會導致 panic。
sync.Cond 基本原理
go 的 sync.Cond
中維護了一個鏈表,這個鏈表記錄了所有阻塞的 goroutine
,也就是由於調用了 Wait
而阻塞的 goroutine
。 而 Signal
和 Broadcast
方法就是用來喚醒這個鏈表中的 goroutine
的。Signal
方法只會喚醒鏈表中的第一個 goroutine
,而 Broadcast
方法會喚醒鏈表中的所有 goroutine
。
下圖是 Signal
方法的效果,可以看到,Signal
方法只會喚醒鏈表中的第一個 goroutine
:
說明:
-
notifyList
是sync.Cond
中維護的一個鏈表,這個鏈表記錄了所有阻塞的goroutine
。 -
head
是鏈表的頭節點,tail
是鏈表的尾節點。 -
Signal
方法只會喚醒鏈表中的第一個goroutine
。
而 Broadcast
方法會喚醒 notifyList
中的所有 goroutine
。
sync.Cond 的設計與實現
最後,我們來看一下 sync.Cond
的設計與實現。
sync.Cond 模型
sync.Cond
的模型如下所示:
type Cond struct {
noCopy noCopy
// L is held while observing or changing the condition
L Locker // L 在觀察或改變條件時被持有
notify notifyList
checker copyChecker
}
屬性說明:
-
noCopy
是一個空結構體,用來檢查sync.Cond
是否被複制。(在編譯前通過go vet
命令來檢查) -
L
是一個Locker
接口,用來保護條件變量。 -
notify
是一個notifyList
類型,用來記錄所有阻塞的goroutine
。 -
checker
是一個copyChecker
類型,用來檢查sync.Cond
是否被複制。(如果在運行時被複制,會導致panic
)
notifyList 結構體
notifyList
是 sync.Cond
中維護的一個鏈表,這個鏈表記錄了所有因爲共享資源還沒準備好而阻塞的 goroutine
。它的定義如下所示:
type notifyList struct {
wait atomic.Uint32
notify uint32
// 阻塞的 waiter 名單。
lock mutex // 鎖
head *sudog // 阻塞的 goroutine 鏈表(鏈表頭)
tail *sudog // 阻塞的 goroutine 鏈表(鏈表尾)
}
屬性說明:
-
wait
是下一個waiter
的編號。它在鎖外自動遞增。 -
notify
是下一個要通知的waiter
的編號。它可以在鎖外讀取,但只能在持有鎖的情況下寫入。 -
lock
是一個mutex
類型,用來保護notifyList
。 -
head
是一個sudog
類型,用來記錄阻塞的goroutine
鏈表的頭節點。 -
tail
是一個sudog
類型,用來記錄阻塞的goroutine
鏈表的尾節點。
notifyList
的方法說明:
notifyList
中包含了幾個操作阻塞的goroutine
鏈表的方法。
-
notifyListAdd
方法將waiter
的編號加 1。 -
notifyListWait
方法將當前的goroutine
加入到notifyList
中。(也就是將當前協程掛起) -
notifyListNotifyOne
方法將notifyList
中的第一個goroutine
喚醒。 -
notifyListNotifyAll
方法將notifyList
中的所有goroutine
喚醒。 -
notifyListCheck
方法檢查 notifyList 的大小是否正確。
sync.Cond 的方法
notifyList
就不細說了,本文重點講解一下 sync.Cond
的實現。
Wait 方法
Wait
方法用在當條件不滿足的時候,將當前運行的協程掛起。
func (c *Cond) Wait() {
// 檢查是否被複制
c.checker.check()
// 更新 notifyList 中需要等待的 waiter 的數量
// 返回當前需要插入 notifyList 的編號
t := runtime_notifyListAdd(&c.notify)
// 解鎖
c.L.Unlock()
// 掛起當前 g,直到被喚醒
runtime_notifyListWait(&c.notify, t)
// 喚醒之後,重新加鎖。
// 因爲阻塞之前解鎖了。
c.L.Lock()
}
對於 Wait
方法,我們需要注意的是,使用之前,我們需要先調用 L.Lock()
方法加鎖,然後再調用 Wait
方法,否則會報錯。
文檔裏面的例子:
c.L.Lock()
for !condition() {
c.Wait()
}
// ...使用條件...
// 這裏是我們在條件滿足之後,需要執行的代碼。
c.L.Unlock()
好了,問題來了,調用 Wait
方法之前爲什麼要先加鎖呢?
這是因爲在我們使用共享資源的時候,可能一些代碼是互斥的,所以我們需要加鎖。 這樣我們就可以保證在我們使用共享資源的時候,不會被其他協程修改。 但是如果因爲條件不滿足,我們需要等待的話,我們不可能在持有鎖的情況下等待, 因爲在修改條件的時候,可能也需要加鎖,這樣就會造成死鎖。
另外一個問題是,爲什麼要使用 for
來檢查條件是否滿足,而不是使用 if
呢?
這是因爲在我們調用 Wait
方法之後,可能會有其他協程喚醒我們,但是條件並沒有滿足, 這個時候依然是需要繼續 Wait
的。
Signal 方法
Signal
方法用在當條件滿足的時候,將 notifyList
中的第一個 goroutine
喚醒。
func (c *Cond) Signal() {
// 檢查 sync.Cond 是否被複制了
c.checker.check()
// 喚醒 notifyList 中的第一個 goroutine
runtime_notifyListNotifyOne(&c.notify)
}
Broadcast 方法
Broadcast
方法用在當條件滿足的時候,將 notifyList
中的所有 goroutine
喚醒。
func (c *Cond) Broadcast() {
// 檢查 sync.Cond 是否被複制了
c.checker.check()
// 喚醒 notifyList 中的所有 goroutine
runtime_notifyListNotifyAll(&c.notify)
}
copyChecker 結構體
copyChecker
結構體用來檢查 sync.Cond
是否被複制。它實際上只是一個 uintptr
類型的值。
type copyChecker uintptr
// check 方法檢查 copyChecker 是否被複制了。
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")
}
}
copyChecker
的值只有兩種可能:
-
0
,表示還沒有調用過Wait
,Signal
或Broadcast
方法。 -
uintptr(unsafe.Pointer(©Checker))
,表示已經調用過Wait
,Signal
或Broadcast
方法。在這幾個方法裏面會調用check
方法,所以copyChecker
的值會被修改。
所以如果 copyChecker
的值不是 0
,也不是 uintptr(unsafe.Pointer(©Checker))
(也就是最初的 copyChecker
的內存地址),則表示 copyChecker
被複制了。
需要注意的是,這個方法在調用
CompareAndSwapUintptr
還會檢查一下,這是因爲有可能會併發調用CompareAndSwapUintptr
, 如果另外一個協程調用了CompareAndSwapUintptr
並且成功了,那麼當前協程的這個CompareAndSwapUintptr
調用會返回false
, 這個時候就需要檢查是否是因爲另外一個協程調用了CompareAndSwapUintptr
而導致的,如果是的話,就不會panic
。
爲什麼 sync.Cond 不能被複制?
從上一小節中我們可以看到,sync.Cond
其實是不允許被複制的,但是如果是在調用 Wait
, Signal
或 Broadcast
方法之前複製,那倒是沒關係。
這是因爲 sync.Cond
中維護了一個阻塞的 goroutine
列表。如果 sync.Cond
被複制了,那麼這個列表就會被複制,這樣就會導致兩個 sync.Cond
都包含了這個列表;但是我們喚醒的時候,只會有其中一個 sync.Cond
被喚醒,另外一個 sync.Cond
就會一直阻塞。 所以 go 直接從語言層面限制了這種情況,不允許 sync.Cond
被複制。
總結
-
sync.Cond
是一個條件變量,它可以用來協調多個goroutine
之間的同步,當條件滿足的時候,去通知那些因爲條件不滿足被阻塞的goroutine
繼續執行。 -
sync.Cond
的接口比較簡單,只有Wait
,Signal
和Broadcast
三個方法。 -
Wait
方法用來阻塞當前goroutine
,直到條件滿足。調用Wait
方法之前,需要先調用L.Lock
方法加鎖。 -
Signal
方法用來喚醒notifyList
中的第一個goroutine
。 -
Broadcast
方法用來喚醒notifyList
中的所有goroutine
。 -
sync.Cond
的實現也比較簡單,它的核心就是notifyList
,它是一個鏈表,用來保存所有因爲條件不滿足而被阻塞的goroutine
。 -
用關閉
channel
的方式也可以實現類似的廣播功能,但是有個問題是channel
不能被重複關閉,所以這種方式無法被多次使用。也就是說使用這種方式無法多次廣播。 -
使用
channel
發送通知的方式也是可以的,但是這樣實現起來就複雜很多了,就更容易出錯了。 -
sync.Cond
中使用copyChecker
來檢查sync.Cond
是否被複制,如果被複制了,就會panic
。需要注意的是,這裏的複製是指調用了Wait
,Signal
或Broadcast
方法之後,sync.Cond
被複制了。在調用這幾個方法之前進行復制是沒有影響的。
本文由 Readfog 進行 AMP 轉碼,版權歸原作者所有。
來源:https://mp.weixin.qq.com/s/Nzh6vycaOR4kthdqjf3I-g