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给大家认识，敬请期待。