golang系列之-sync.Cond

发表于 更新于 本文字数： 1.3k 阅读时长 ≈ 5 分钟

sync.Cond经常用在多个 goroutine 等待，一个 goroutine 通知（事件发生）的场景。如果是一个通知，一个等待，使用互斥锁或 channel 就能搞定了。底层实现基于信号量semaphore

当前go版本：1.24

快速上手

以下展示一个sync.Cond的使用案例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
package main

import (
    "fmt"
    "sync"
)

var shared = make(map[string]interface{})

func main() {
    var wg sync.WaitGroup
    wg.Add(2)

    cond := sync.NewCond(&sync.Mutex{})

    // reader 打印shared[key]
    reader := func(key string) {
        cond.L.Lock()
        // 等待，直到shared有数据
        for len(shared) == 0 {
            cond.Wait() // Wait内部会暂时解锁/加锁
        }
        fmt.Println(shared[key])
        cond.L.Unlock()
        wg.Done()
    }

    // 创建两个goroutine
    go reader("rsc1")
    go reader("rsc2")

    // writer
    cond.L.Lock()
    // 写入
    shared["rsc1"] = "foo"
    shared["rsc2"] = "bar"
    // 通知所有goroutine
    cond.Broadcast()
    cond.L.Unlock()

    wg.Wait()
}

数据结构

todo：文章图片待补充

Cond的数据结构如下

  • L - 只要实现Locker接口的类型都可以，比如Mutex
  • notify - 计数器也是ticket生成器，可实现等待队列的先进先出-FIFO，由内部sema使用
  • checker - 用来检测是否被复制
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// sync.Cond结构体，看notify就行了
type Cond struct {
    noCopy noCopy

    // L is held while observing or changing the condition
    L Locker                // mutex，由外部传入

    notify  notifyList      // 计数器+sudog链表，内部sema使用
    checker copyChecker     // 检查Cond是否被复制
}

type notifyList struct {
    wait atomic.Uint32      // 等待g数量，只增不减

    notify uint32           // 唤醒g数量，范围[0,wait]，保证waiter先进先出

    lock mutex              // 锁
	// 下面是等待g的链表的头部跟尾部
    head *sudog             // 指向第一个sudog
    tail *sudog             // 指向最后一个sudog
}

核心方法

NewCond

创建一个condition variable，没什么好讲的

1
2
3
4
// 没什么好说的
func NewCond(l Locker) *Cond {
    return &Cond{L: l}
}

Wait

大概逻辑如下

  1. 更新wait计数器，获得ticket，临时解锁cond，让其他goroutine可以获得ticket并挂起
  2. 将当前goroutine和ticket打包进sudog，放进队列，当前goroutine挂起陷入等待
  3. goroutine被唤醒，加锁cond
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
// 使用示例
//    c.L.Lock()
//    for !condition() {
//        c.Wait()
//    }
//    ... make use of condition ...
//    c.L.Unlock()
func (c *Cond) Wait() {
    // 上层业务逻辑已加锁 => cond.L.Lock()

	// 检查是否被复制
    c.checker.check()

	// wait计数器更新
	// t = (c.notify.wait+=1) - 1
    t := runtime_notifyListAdd(&c.notify)

    // 解锁，其他g现在可以调用Wait了
    c.L.Unlock()

    // 将当前g挂起，等待唤醒（函数看下方notifyListWait）
    runtime_notifyListWait(&c.notify, t)
	// 被唤醒

    // 恢复加锁，由上层业务逻辑负责继续解锁 => cond.L.Unlock()
    c.L.Lock()
}

// runtime_notifyListWait函数
// sudog纪录g放入链表，g让出CPU挂起等待唤醒
func notifyListWait(l *notifyList, t uint32) {
	// 加锁
	lockWithRank(&l.lock, lockRankNotifyList)

	// t是等待计数器wait，每次唤醒时notify计数器+1，任何wait<notify的id说明该g已被/应被唤醒
	if less(t, l.notify) {
		unlock(&l.lock)
		return
	}

	// 获取sudog
	s := acquireSudog()
	// 封装当前g
	s.g = getg()
	// ticket可以纪录一个uint32的数
	s.ticket = t
	// 纪录g阻塞耗时
	s.releasetime = 0
	t0 := int64(0)
	if blockprofilerate > 0 {
		t0 = cputicks()
		s.releasetime = -1
	}
	// 把sudog加到链表
	if l.tail == nil {
		l.head = s
	} else {
		l.tail.next = s
	}
	l.tail = s
	// 当前g让出CPU，g0执行调度运行其他g
	// 在内部g、m解除绑定后会解锁l.lock
	goparkunlock(&l.lock, waitReasonSyncCondWait, traceBlockCondWait, 3)
	if t0 != 0 {
		// 纪录阻塞事件
		blockevent(s.releasetime-t0, 2)
	}
	// 释放sudog
	releaseSudog(s)
}

Signal

大概逻辑如下

  1. 如果waiter数量为0（wait==notify），不处理
  2. 根据notify计算出待唤醒的waiter_id，并更新notify
  3. 在sudog列表查找ticket==waiter_id的goroutine，将其唤醒
  4. 如果goroutine在wait时就接到信号，那么它肯定不在sudog列表，不处理
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
// 唤醒一个g
func (c *Cond) Signal() {
    c.checker.check()
    // 按顺序唤醒一个g（看下方notifyListNotifyOne）
    runtime_notifyListNotifyOne(&c.notify)
}

// runtime_notifyListNotifyOne函数
func notifyListNotifyOne(l *notifyList) {
	// 每次唤醒时notify计数器+1，任何wait<notify的id说明该g已被/应被唤醒
	if l.wait.Load() == atomic.Load(&l.notify) {
		return
	}

	// 加锁
	lockWithRank(&l.lock, lockRankNotifyList)

	// 应被唤醒的g的id
	t := l.notify
	// wait计数器==notify计数器，g已被唤醒
	if t == l.wait.Load() {
		unlock(&l.lock)
		return
	}

	// notify计数器+=1
	atomic.Store(&l.notify, t+1)

	// 遍历sudog链表
	// prev，curr = nil, l.head
	for p, s := (*sudog)(nil), l.head; s != nil; p, s = s, s.next {
		// ticket相同，找到了
		if s.ticket == t {
			// 下一个sudog
			n := s.next
			// prev不为nil，将其与next相连接
			if p != nil {
				p.next = n
			} else {
				l.head = n
			}
			if n == nil {
				l.tail = p
			}
			// 解锁
			unlock(&l.lock)
			// 当前sudog的一些字段都要重置
			s.next = nil
			// 测试，忽略
			if s.g.syncGroup != nil && getg().syncGroup != s.g.syncGroup {
				println("semaphore wake of synctest goroutine", s.g.goid, "from outside bubble")
				panic("semaphore wake of synctest goroutine from outside bubble")
			}
			// 纪录releasetime，调用goready唤醒g
			// 将g放到本地队列，从空闲队列拿一个p和一个m绑定并唤醒（可能拿不到p）
			readyWithTime(s, 4)
			return
		}
	}
	// 没找到sudog
	unlock(&l.lock)
}

Broadcast

大概逻辑如下

  1. 如果waiter数量为0（wait==notify），不处理
  2. 更新notify，使其等于wait
  3. 扫描sudog列表，逐个唤醒goroutine
  4. 如果goroutine在wait时就接到信号，那么它肯定不在sudog列表，不处理
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
// 
func (c *Cond) Broadcast() {
    c.checker.check()
    // 唤醒所有goroutine（看下方notifyListNotifyAll）
    runtime_notifyListNotifyAll(&c.notify)
}

// 
func notifyListNotifyAll(l *notifyList) {
	// 每次唤醒时notify计数器+1，任何wait<notify的id说明该g已被/应被唤醒
	if l.wait.Load() == atomic.Load(&l.notify) {
		return
	}

	// 加锁
	lockWithRank(&l.lock, lockRankNotifyList)
	// 获取整个sudog链表
	s := l.head
	// 重置head和tail
	l.head = nil
	l.tail = nil

	// 使两个计数器数值相等
	atomic.Store(&l.notify, l.wait.Load())
	// 解锁
	unlock(&l.lock)

	// 遍历sudog链表
	for s != nil {
		next := s.next
		// 重置
		s.next = nil
		// 测试
		if s.g.syncGroup != nil && getg().syncGroup != s.g.syncGroup {
			println("semaphore wake of synctest goroutine", s.g.goid, "from outside bubble")
			panic("semaphore wake of synctest goroutine from outside bubble")
		}
		// 纪录releasetime，调用goready唤醒g
		// 将g放到本地队列，从空闲队列拿一个p和一个m绑定并唤醒（可能拿不到p）
		readyWithTime(s, 4)
		s = next
	}
}

参考文档

Go sync.Cond, the Most Overlooked Sync Mechanism