Go：有了 Sync 為什么還有 Atomic？
Go 是一種擅長并發的語言，啟動新的 goroutine 就像輸入 “go” 一樣簡單。隨著你發現自己構建的系統越來越復雜，正確保護對共享資源的訪問以防止競爭條件變得極其重要。此類資源可能包括可即時更新的配置(例如功能標志)、內部狀態(例如斷路器狀態)等。
01 什么是競態條件?
對于大多數讀者來說，這可能是基礎知識，但由于本文的其余部分取決于對競態條件的理解，因此有必要進行簡短的復習。競態條件是一種情況，在這種情況下，程序的行為取決于其他不可控事件的順序或時間。在大多數情況下，這種情況是一個錯誤，因為可能會發生不希望的結果。
舉個具體的例子或許更容易理解：
- // race_condition_test.go
- package main
- import (
- "fmt"
- "sort"
- "sync"
- "testing"
- )
- func Test_RaceCondition(t *testing.T) {
- var s = make([]int, 0)
- wg := sync.WaitGroup{}
- // spawn 10 goroutines to modify the slice in parallel
- for i := 0; i < 10; i++ {
- wg.Add(1)
- go func(i int) {
- defer wg.Done()
- s = append(s, i) //add a new item to the slice
- }(i)
- }
- wg.Wait()
- sort.Ints(s) //sort the response to have comparable results
- fmt.Println(s)
- }
執行一：
- $ go test -v race_condition_test.go
- === RUN Test_RaceCondition
- [0 1 2 3 4 5 6 7 8 9]
- --- PASS: Test_RaceCondition (0.00s)
這里看起來一切都很好。這是我們預期的輸出。該程序迭代了 10 次，并在每次迭代時將索引添加到切片中。
執行二：
- === RUN Test_RaceCondition
- [0 3]
- --- PASS: Test_RaceCondition (0.00s)
等等，這里發生了什么?這次我們的響應切片中只有兩個元素。這是因為切片的內容 s 在加載和修改之間發生了變化，導致程序覆蓋了一些結果。這種特殊的競態條件是由數據競爭引起的，在這種情況下，多個 goroutine 嘗試同時訪問特定的共享變量，并且這些 goroutine 中的至少一個嘗試修改它。(注意，以上結果并非一定如此，每次運行結果可能都不相同)
如果你使用 -race 標志執行測試，go 甚至會告訴你存在數據競爭并幫助你準確定位：
- $ go test race_condition_test.go -race
- ==================
- WARNING: DATA RACE
- Read at 0x00c000132048 by goroutine 9:
- command-line-arguments.Test_RaceCondition.func1()
- /home/sfinlay/go/src/benchmarks/race_condition_test.go:20 +0xb4
- command-line-arguments.Test_RaceCondition·dwrap·1()
- /home/sfinlay/go/src/benchmarks/race_condition_test.go:21 +0x47
- Previous write at 0x00c000132048 by goroutine 8:
- command-line-arguments.Test_RaceCondition.func1()
- /home/sfinlay/go/src/benchmarks/race_condition_test.go:20 +0x136
- command-line-arguments.Test_RaceCondition·dwrap·1()
- /home/sfinlay/go/src/benchmarks/race_condition_test.go:21 +0x47
- Goroutine 9 (running) created at:
- command-line-arguments.Test_RaceCondition()
- /home/sfinlay/go/src/benchmarks/race_condition_test.go:18 +0xc5
- testing.tRunner()
- /usr/local/go/src/testing/testing.go:1259 +0x22f
- testing.(*T).Run·dwrap·21()
- /usr/local/go/src/testing/testing.go:1306 +0x47
- Goroutine 8 (finished) created at:
- command-line-arguments.Test_RaceCondition()
- /home/sfinlay/go/src/benchmarks/race_condition_test.go:18 +0xc5
- testing.tRunner()
- /usr/local/go/src/testing/testing.go:1259 +0x22f
- testing.(*T).Run·dwrap·21()
- /usr/local/go/src/testing/testing.go:1306 +0x47
- ==================
02 并發控制
保護對這些共享資源的訪問通常涉及常見的內存同步機制，例如通道或互斥鎖。
這是將競態條件調整為使用互斥鎖的相同測試用例：
- func Test_NoRaceCondition(t *testing.T) {
- var s = make([]int, 0)
- m := sync.Mutex{}
- wg := sync.WaitGroup{}
- // spawn 10 goroutines to modify the slice in parallel
- for i := 0; i < 10; i++ {
- wg.Add(1)
- go func(i int) {
- m.Lock()
- defer wg.Done()
- defer m.Unlock()
- s = append(s, i)
- }(i)
- }
- wg.Wait()
- sort.Ints(s) //sort the response to have comparable results
- fmt.Println(s)
- }
這次它始終返回所有 10 個整數，因為它確保每個 goroutine 僅在沒有其他人執行時才讀寫切片。如果第二個 goroutine 同時嘗試獲取鎖，它必須等到前一個 goroutine 完成(即直到它解鎖)。
然而，對于高吞吐量系統，性能變得非常重要，因此減少鎖爭用(即一個進程或線程試圖獲取另一個進程或線程持有的鎖的情況)變得更加重要。執行此操作的最基本方法之一是使用讀寫鎖 ( sync.RWMutex) 而不是標準 sync.Mutex，但是 Go 還提供了一些原子內存原語即 atomic 包。
03 原子
Go 的 atomic 包提供了用于實現同步算法的低級原子內存原語。這聽起來像是我們需要的東西，所以讓我們嘗試用 atomic 重寫該測試：
- import "sync/atomic"
- func Test_RaceCondition_Atomic(t *testing.T) {
- var s = atomic.Value{}
- s.Store([]int{}) // store empty slice as the base
- wg := sync.WaitGroup{}
- // spawn 10 goroutines to modify the slice in parallel
- for i := 0; i < 10; i++ {
- wg.Add(1)
- go func(i int) {
- defer wg.Done()
- s1 := s.Load().([]int)
- s.Store(append(s1, i)) //replace the slice with a new one containing the new item
- }(i)
- }
- wg.Wait()
- s1 := s.Load().([]int)
- sort.Ints(s1) //sort the response to have comparable results
- fmt.Println(s1)
- }
執行結果：
- === RUN Test_RaceCondition_Atomic
- [1 3]
- --- PASS: Test_RaceCondition_Atomic (0.00s)
什么?這和我們之前遇到的問題完全一樣，那么這個包有什么好處呢?
04 讀取-復制-更新
atomic 不是靈丹妙藥，它顯然不能替代互斥鎖，但是當涉及到可以使用讀取-復制-更新[1]模式管理的共享資源時，它非常出色。在這種技術中，我們通過引用獲取當前值，當我們想要更新它時，我們不修改原始值，而是替換指針(因此沒有人訪問另一個線程可能訪問的相同資源)。前面的示例無法使用此模式實現，因為它應該隨著時間的推移擴展現有資源而不是完全替換其內容，但在許多情況下，讀取-復制-更新是完美的。
這是一個基本示例，我們可以在其中獲取和存儲布爾值(例如，對于功能標志很有用)。在這個例子中，我們正在執行一個并行基準測試，比較原子和讀寫互斥：
- package main
- import (
- "sync"
- "sync/atomic"
- "testing"
- )
- type AtomicValue struct{
- value atomic.Value
- }
- func (b *AtomicValue) Get() bool {
- return b.value.Load().(bool)
- }
- func (b *AtomicValue) Set(value bool) {
- b.value.Store(value)
- }
- func BenchmarkAtomicValue_Get(b *testing.B) {
- atomB := AtomicValue{}
- atomB.value.Store(false)
- b.RunParallel(func(pb *testing.PB) {
- for pb.Next() {
- atomB.Get()
- }
- })
- }
- /************/
- type MutexBool struct {
- mutex sync.RWMutex
- flag bool
- }
- func (mb *MutexBool) Get() bool {
- mb.mutex.RLock()
- defer mb.mutex.RUnlock()
- return mb.flag
- }
- func BenchmarkMutexBool_Get(b *testing.B) {
- mb := MutexBool{flag: true}
- b.RunParallel(func(pb *testing.PB) {
- for pb.Next() {
- mb.Get()
- }
- })
- }
結果：
- cpu: Intel(R) Core(TM) i7-8650U CPU @ 1.90GHz
- BenchmarkAtomicValue_Get
- BenchmarkAtomicValue_Get-8 1000000000 0.5472 ns/op
- BenchmarkMutexBool_Get
- BenchmarkMutexBool_Get-8 24966127 48.80 ns/op
結果很清楚。atomic 的速度提高了 89 倍以上。并且可以通過使用更原始的類型來進一步改進：
- type AtomicBool struct{ flag int32 }
- func (b *AtomicBool) Get() bool {
- return atomic.LoadInt32(&(b.flag)) != 0
- }
- func (b *AtomicBool) Set(value bool) {
- var i int32 = 0
- if value {
- i = 1
- }
- atomic.StoreInt32(&(b.flag), int32(i))
- }
- func BenchmarkAtomicBool_Get(b *testing.B) {
- atomB := AtomicBool{flag: 1}
- b.RunParallel(func(pb *testing.PB) {
- for pb.Next() {
- atomB.Get()
- }
- })
- }
- cpu: Intel(R) Core(TM) i7-8650U CPU @ 1.90GHz
- BenchmarkAtomicBool_Get
- BenchmarkAtomicBool_Get-8 1000000000 0.3161 ns/op
此版本比互斥鎖版本快 154 倍以上。
寫操作也顯示出明顯的差異(盡管規模并不那么令人印象深刻)：
- func BenchmarkAtomicBool_Set(b *testing.B) {
- atomB := AtomicBool{flag: 1}
- b.RunParallel(func(pb *testing.PB) {
- for pb.Next() {
- atomB.Set(true)
- }
- })
- }
- /************/
- func BenchmarkAtomicValue_Set(b *testing.B) {
- atomB := AtomicValue{}
- atomB.value.Store(false)
- b.RunParallel(func(pb *testing.PB) {
- for pb.Next() {
- atomB.Set(true)
- }
- })
- }
- /************/
- func BenchmarkMutexBool_Set(b *testing.B) {
- mb := MutexBool{flag: true}
- b.RunParallel(func(pb *testing.PB) {
- for pb.Next() {
- mb.Set(true)
- }
- })
- }
結果：
- cpu: Intel(R) Core(TM) i7-8650U CPU @ 1.90GHz
- BenchmarkAtomicBool_Set
- BenchmarkAtomicBool_Set-8 64624705 16.79 ns/op
- BenchmarkAtomicValue_Set
- BenchmarkAtomicValue_Set-8 47654121 26.43 ns/op
- BenchmarkMutexBool_Set
- BenchmarkMutexBool_Set-8 20124637 66.50 ns/op
在這里我們可以看到 atomic 在寫入時比在讀取時慢得多，但仍然比互斥鎖快得多。有趣的是，我們可以看到互斥鎖讀取和寫入之間的差異不是很明顯(慢 30%)。盡管如此， atomic 仍然表現得更好(比互斥鎖快 2-4 倍)。
05 為什么 atomic 這么快?
簡而言之，原子操作很快，因為它們依賴于原子 CPU 指令而不是依賴外部鎖。使用互斥鎖時，每次獲得鎖時，goroutine 都會短暫暫?；蛑袛?，這種阻塞占使用互斥鎖所花費時間的很大一部分。原子操作可以在沒有任何中斷的情況下執行。
06 atomic 總是答案嗎?
正如我們在一個早期示例中已經證明的那樣，atomic 無法解決所有問題，某些操作只能使用互斥鎖來解決。
考慮以下示例，該示例演示了我們使用 map 作為內存緩存的常見模式：
- package main
- import (
- "sync"
- "sync/atomic"
- "testing"
- )
- //Don't use this implementation!
- type AtomicCacheMap struct {
- value atomic.Value //map[int]int
- }
- func (b *AtomicCacheMap) Get(key int) int {
- return b.value.Load().(map[int]int)[key]
- }
- func (b *AtomicCacheMap) Set(key, value int) {
- oldMap := b.value.Load().(map[int]int)
- newMap := make(map[int]int, len(oldMap)+1)
- for k, v := range oldMap {
- newMap[k] = v
- }
- newMap[key] = value
- b.value.Store(newMap)
- }
- func BenchmarkAtomicCacheMap_Get(b *testing.B) {
- atomM := AtomicCacheMap{}
- atomM.value.Store(testMap)
- b.RunParallel(func(pb *testing.PB) {
- for pb.Next() {
- atomM.Get(0)
- }
- })
- }
- func BenchmarkAtomicCacheMap_Set(b *testing.B) {
- atomM := AtomicCacheMap{}
- atomM.value.Store(testMap)
- var i = 0
- b.RunParallel(func(pb *testing.PB) {
- for pb.Next() {
- atomM.Set(i, i)
- i++
- }
- })
- }
- /************/
- type MutexCacheMap struct {
- mutex sync.RWMutex
- value map[int]int
- }
- func (mm *MutexCacheMap) Get(key int) int {
- mm.mutex.RLock()
- defer mm.mutex.RUnlock()
- return mm.value[key]
- }
- func (mm *MutexCacheMap) Set(key, value int) {
- mm.mutex.Lock()
- defer mm.mutex.Unlock()
- mm.value[key] = value
- }
- func BenchmarkMutexCacheMap_Get(b *testing.B) {
- mb := MutexCacheMap{value: testMap}
- b.RunParallel(func(pb *testing.PB) {
- for pb.Next() {
- mb.Get(0)
- }
- })
- }
- func BenchmarkMutexCacheMap_Set(b *testing.B) {
- mb := MutexCacheMap{value: testMap}
- var i = 0
- b.RunParallel(func(pb *testing.PB) {
- for pb.Next() {
- mb.Set(i, i)
- i++
- }
- })
- }
結果：
- cpu: Intel(R) Core(TM) i7-8650U CPU @ 1.90GHz
- BenchmarkAtomicCacheMap_Get
- BenchmarkAtomicCacheMap_Get-8 301664540 4.194 ns/op
- BenchmarkAtomicCacheMap_Set
- BenchmarkAtomicCacheMap_Set-8 87637 95889 ns/op
- BenchmarkMutexCacheMap_Get
- BenchmarkMutexCacheMap_Get-8 20000959 54.63 ns/op
- BenchmarkMutexCacheMap_Set
- BenchmarkMutexCacheMap_Set-8 5012434 267.2 ns/op
哎呀，這種表現是痛苦的。這意味著，當必須復制大型結構時，atomic 的性能非常差。不僅如此，此代碼還包含競態條件。就像本文開頭的切片案例一樣，原子緩存示例具有競態條件，其中可能會在復制 map 和存儲 map 的時間之間添加新的緩存條目，在這種情況下，新條目將丟失。在這種情況下，該 -race 標志不會檢測到任何數據競爭，因為沒有對同一 map 的并發訪問。
07 注意事項
Go 的文檔[2]警告了 atomic 包的潛在誤用：
這些函數需要非常小心才能正確使用。除了特殊的低級應用程序，同步最好使用通道或 sync 包的工具來完成。通過通信共享內存;不要通過共享內存進行通信。
開始使用 atomic 包時，你可能會遇到的第一個問題是：
- panic: sync/atomic: store of inconsistently typed value into Value
使用 atomic.Store，確保每次調用方法時都存儲完全相同的類型很重要。這聽起來很容易，但通常并不像聽起來那么簡單：
- package main
- import (
- "fmt"
- "sync/atomic"
- )
- //Our own custom error type which implements the error interface
- type CustomError struct {
- Code int
- Message string
- }
- func (e CustomError) Error() string {
- return fmt.Sprintf("%d: %s", e.Code, e.Message)
- }
- func InternalServerError(msg string) error {
- return CustomError{Code: 500, Message: msg}
- }
- func main() {
- var (
- err1 error = fmt.Errorf("error happened")
- err2 error = InternalServerError("another error happened")
- )
- errVal := atomic.Value{}
- errVal.Store(err1)
- errVal.Store(err2) //panics here
- }
兩個值都是 error 類型是不夠的，因為它們只是實現了錯誤接口。它們的具體類型仍然不同，因此 atomic 不喜歡它。
08 總結
競態條件很糟糕，應該保護對共享資源的訪問?；コ怏w很酷，但由于鎖爭用而趨于緩慢。對于某些讀取-復制-更新模式有意義的情況(這往往是動態配置之類的東西，例如特性標志、日志級別或 map 或結構體，一次填充例如通過 JSON 解析等)，尤其是當讀取次數比寫入次數多時。atomic 通常不應用于其他用例(例如，隨時間增長的變量，如緩存)，并且該特性的使用需要非常小心。
可能最重要的方法是將鎖保持在最低限度，如果你在在考慮原子等替代方案，請務必在投入生產之前對其進行廣泛的測試和試驗。
原文鏈接：https://www.sixt.tech/golangs-atomic
參考資料
[1]讀取-復制-更新: https://en.wikipedia.org/wiki/Read-copy-update
[2]文檔: https://pkg.go.dev/sync/atomic
本文轉載自微信公眾號「幽鬼」，可以通過以下二維碼關注。轉載本文請聯系幽鬼公眾號。
