Sync.WaitGroup

A WaitGroup waits for a collection of goroutines to finish.

可以透過內建的sync WaitGroup來等待線程結束，

就像一群學生在休息，等到大家集合完畢才能開始上課。

package main

import (
	"fmt"
	"time"
	"sync"
)

func main() {
	fmt.Println("下課休息3秒鐘！")

	wg := sync.WaitGroup{} 
	wg.Add(2)	

	go rest(&wg)
	go rest(&wg)

	fmt.Println("開始休息")
	wg.Wait()
	fmt.Println("休息完畢準備上課")
}

func rest(wg *sync.WaitGroup) {
	time.Sleep(time.Second * 3)
	fmt.Println("學生休息完畢。")
	wg.Done()
}

WaitGroup拿計數器(Counter)來當作任務數量，若counter < 0會發生panic。

• WaitGroup.Add(n)：計數器+n
• WaitGroup.Done()：任務完成，從計數器中減去1，可搭配defer使用
• WaitGroup.Wait()：阻塞(Block)住，直到計數器歸0

此外需要注意以下幾點:

• 如果計數器大於線程數就會發生死結(Deadlock)。啊兵就只有兩隻，等到死還是只有這麼多隻，永遠沒辦法集合完畢。
• 因為是針對該鎖的物件操作，記得是要傳入func指針(Pointer)與位址(Address)。

sync.WaitGroup雖然好用，但也會面臨著零零種種的問題，也因此我們必須時時刻刻的讓它保持原子性

Mutex

為了讓為其保持原子性，我們必須通過snyc.Mutex確保該語句在同一時間只被單一線程goroutine所訪問。

package main

import (
	"fmt"
	"sync"
)

var total struct {
	sync.Mutex
	value int
}

func worker(wg *sync.WaitGroup) {
	defer wg.Done()
	
	for i := 0; i<= 10; i++ {
		total.Lock()
		total.value += i
		total.Unlock()

	}
}

func main() {
	var wg sync.WaitGroup
  start := time.Now()
	wg.Add(2)
	go worker(&wg)
	go worker(&wg)
	wg.Wait()
	elapsed := time.Since(start)
	fmt.Println(total.value)
	fmt.Println("executing time: ", elapsed)
}

運行後可得結果

110

上述程式碼表示有兩個worker再不爭奪資源的情況下累加0~10，但雖然sync.Mutex使得goroutine能夠很安全的

Atomic

然而使用互斥鎖共享資源會使得效率低下，因此我們可以使用sync/atomic來解決這問題。

package main

import (
        "fmt"
	"sync"
	"sync/atomic"
)

var total uint64

func worker(wg *sync.WaitGroup) {
	defer wg.Done()
	
	var i uint64
	for i = 0; i <= 10; i++ {
		atomic.AddUint64(&total, i)
	}
}

func main() {
	var wg sync.WaitGroup
	wg.Add(2)
	go worker(&wg)
	go worker(&wg)
	wg.Wait()

	fmt.Println(total)

}

運行後可得結果

110

atomic.AddUint64() 保證了total的CURD是個原子操作，因此在多線程訪問也是安全的。

由於互斥鎖的代價比原子讀寫高得多，在性能敏感地方可以增加一個數字型標誌，通過原子檢測標誌狀態來降低互斥鎖的使用次數來提高性能。

Sync.Map

由於之前所介紹的資料結構map在併發時只保證read是線程安全，但write並非線程安全，也因此官方在go1.19時加入了sync.Map來確保併發的安全性與高效性。

併發使用map所產生的問題

我們先來試試看使用一般的map要如何安全併發

package main

import (
        "fmt"
	"time"
)

func main() {
	m := map[int]int {1:1}
	go do(m)
	go do(m)
	time.Sleep(1*time.Second)
	fmt.Println(m)
}

func do (m map[int]int) {
	i := 0
	for i < 10000 {
		m[1]=1
		i++
	}
}

運行後可得以下結果

fatal error: concurrent map writes

goroutine 6 [running]:
runtime.throw({0x4974fc, 0x0})
	/usr/local/go-faketime/src/runtime/panic.go:1198 +0x71 fp=0xc000036758 sp=0xc000036728 pc=0x42fa91
runtime.mapassign_fast64(0x0, 0x0, 0x1)
	/usr/local/go-faketime/src/runtime/map_fast64.go:101 +0x2c5 fp=0xc000036790 sp=0xc000036758 pc=0x40f485
main.do(0x0)
	/tmp/sandbox1088286762/prog.go:19 +0x36 fp=0xc0000367c8 sp=0xc000036790 pc=0x47e616
main.main·dwrap·1()
	/tmp/sandbox1088286762/prog.go:10 +0x26 fp=0xc0000367e0 sp=0xc0000367c8 pc=0x47e5a6
runtime.goexit()
	/usr/local/go-faketime/src/runtime/asm_amd64.s:1581 +0x1 fp=0xc0000367e8 sp=0xc0000367e0 pc=0x45ad21
created by main.main
	/tmp/sandbox1088286762/prog.go:10 +0x7f

goroutine 1 [sleep]:
time.Sleep(0x3b9aca00)
	/usr/local/go-faketime/src/runtime/time.go:193 +0x111
main.main()
	/tmp/sandbox1088286762/prog.go:12 +0xcf

goroutine 7 [runnable]:
main.do(0x0)
	/tmp/sandbox1088286762/prog.go:19 +0x4b
created by main.main
	/tmp/sandbox1088286762/prog.go:11 +0xc5

這邊可以很清楚得知，無法併發的同時對map寫入。

也因此下面我們會加入mutex試試看能否解決

package main

import (
        "fmt"
	"time"
	"sync"
)

var s sync.Mutex

func main() {
	m := map[int]int {1:1}
	go do(m)
	go do(m)
	time.Sleep(1*time.Second)
	fmt.Println(m)
}

func do (m map[int]int) {
	i := 0
	for i < 10000 {
		s.Lock()
		m[1]=1
		i++
		s.Unlock()
	}
}

運行後可得以下結果

map[1:1]

加入鎖之後避免Race Condition果然就能解決這問題，但加鎖始終並不是最佳解，因為他會產生效率問題。

也因此我們必須想辦法減少加解鎖的時間：

1. 透過空間換取時間。
2. 降低影響範圍減少效能的退減。

使用sync.Map

package main

import (
        "fmt"
	"time"
	"sync"
)

func main() {
	m := sync.Map{}
	m.Store(1,1)
	go do(m)
	go do(m)
	time.Sleep(1*time.Second)
	fmt.Println(m.Load(1))
}

func do (m sync.Map) {
	i := 0
	for i < 10000 {
		m.Store(1,1)
		i++
	}
}

運行後可得以下結果

1 true

Sync.Map 思路分析

• 空間換取時間，透過讀寫分離(read & dirty兩個資料結構)來降低鎖時間進而提升效率
• 動態調整資料，miss次數多了會將dirty的資料migrate至read
• 優先從read進行RUD，因為對read操作不需要鎖，性能較好
• 但不適用於大量寫入場景，這樣會使read map讀不到數據而進一步加鎖讀取，同時dirty map也會一直晉升為read map，整體性能較差。

接下來看一下標準庫吧

src/sync/map.go

type Map struct {
    mu Mutex
    read atomic.Value
    dirty map[interface{}]*entry
    misses int
}

• mu：當涉及dirty操作時，會需要使用該鎖保持原子性
• read: 對read進行讀取並不需要鎖，只需要一個atomic持續記錄最新的pointer
• dirty: 包含大部分map的key-value data，對dirty操作會需要用到mu，且dirty的資料會持續的更新至read當中。
• misses: 用來記錄read讀取不到資料，而在dirty讀取得到的次數，當misses的count等於dirty長度時，就會進行一次migration(dirty → read)

type readOnly struct {
    m       map[interface{}]*entry
    
    amended bool
}

• m: read當中的資料，不會對任何內容資料進行增加或刪除，但可以改變entry的pointer。
• amended: 當其顯示為true，表示read資料並不完整，有部分資料需要從dirty migrate過來。

所以結論如下：

• 大量讀少量寫的併發場景：Sync.Map
• 大量寫入的併發場景： map + Mutex

Summary

這章節讓我們知道了Sync.WaitGroup與Sync.Map的使用情境與方式，在下個章節則會介紹channel給大家認識，敬請期待。