关于golang:手摸手Go-深入理解syncCond

Today that you are wasting is the unattainable tomorrow to someone who expired yesterday. This very moment that you detest is the unreturnable experience to your future self.

sync.Cond实现了一个条件变量，用于期待一个或一组goroutines满足条件后唤醒的场景。每个Cond关联一个Locker通常是一个*Mutex或RWMutex\`依据需要初始化不同的锁。

根本用法

老规矩正式分析源码前，先来看看sync.Cond如何应用。比方咱们实现一个FIFO的队列

`package main`
`import (`
 `"fmt"`
 `"math/rand"`
 `"os"`
 `"os/signal"`
 `"sync"`
 `"time"`
`)`
`type FIFO struct {`
 `lock  sync.Mutex`
 `cond  *sync.Cond`
 `queue []int`
`}`
`type Queue interface {`
 `Pop() int`
 `Offer(num int) error`
`}`
`func (f *FIFO) Offer(num int) error {`
 `f.lock.Lock()`
 `defer f.lock.Unlock()`
 `f.queue = append(f.queue, num)`
 `f.cond.Broadcast()`
 `return nil`
`}`
`func (f *FIFO) Pop() int {`
 `f.lock.Lock()`
 `defer f.lock.Unlock()`
 `for {`
 `for len(f.queue) == 0 {`
 `f.cond.Wait()`
 `}`
 `item := f.queue[0]`
 `f.queue = f.queue[1:]`
 `return item`
 `}`
`}`
`func main() {`
 `l := sync.Mutex{}`
 `fifo := &FIFO{`
 `lock:  l,`
 `cond:  sync.NewCond(&l),`
 `queue: []int{},`
 `}`
 `go func() {`
 `for {`
 `fifo.Offer(rand.Int())`
 `}`
 `}()`
 `time.Sleep(time.Second)`
 `go func() {`
 `for {`
 `fmt.Println(fmt.Sprintf("goroutine1 pop-->%d", fifo.Pop()))`
 `}`
 `}()`
 `go func() {`
 `for {`
 `fmt.Println(fmt.Sprintf("goroutine2 pop-->%d", fifo.Pop()))`
 `}`
 `}()`
 `ch := make(chan os.Signal, 1)`
 `signal.Notify(ch, os.Interrupt)`
 `<-ch`
`}`

咱们定一个FIFO 队列有Offer和Pop两个操作，咱们起一个gorountine一直向队列投放数据，另外两个gorountine一直取拿数据。

1. Pop操作会判断如果队列里没有数据len(f.queue) == 0则调用f.cond.Wait()将goroutine挂起。
2. 等到Offer操作投放数据胜利，外面调用f.cond.Broadcast()来唤醒所有挂起在这个mutex上的goroutine。当然sync.Cond也提供了一个Signal(),有点儿相似Java中的notify()和notifyAll()的意思 次要是唤醒一个和唤醒全副的区别。

总结一下sync.Mutex的大抵用法

1. 首先申明一个mutex，这里sync.Mutex/sync.RWMutex可依据理论状况选用
2. 调用sync.NewCond(l Locker) *Cond 应用1中的mutex作为入参 留神 这里传入的是指针 为了防止c.L.Lock()、c.L.Unlock()调用频繁复制锁 导致死锁
3. 依据业务条件 满足则调用cond.Wait()挂起goroutine
4. cond.Broadcast()唤起所有挂起的gorotune 另一个办法cond.Signal()唤醒一个最先挂起的goroutine

须要留神的是cond.wait()的应用须要参照如下模版 具体为啥咱们后续剖析

 `c.L.Lock()`
 `for !condition() {`
 `c.Wait()`
 `}`
 `... make use of condition ...`
 `c.L.Unlock()`

源码剖析

数据结构

剖析具体方法前，咱们先来理解下sync.Cond的数据结构。具体源码如下：

`type Cond struct {`
 `noCopy noCopy // Cond应用后不容许拷贝`
 `// L is held while observing or changing the condition`
 `L Locker`
 `//告诉列表调用wait()办法的goroutine会被放到notifyList中`
 `notify  notifyList`
 `checker copyChecker //查看Cond实例是否被复制`
`}`

noCopy之前讲过 不分明的能够看下《你真的理解mutex吗》，除此之外，Locker是咱们刚刚谈到的mutex，copyChecker是用来查看Cond实例是否被复制的,就有一个办法 :

`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")`
 `}`
`}`

大抵意思是说，初始type copyChecker uintptr默认为0，当第一次调用check()会将copyChecker本身的地址复制给本人，至于为什么uintptr(*c) != uintptr(unsafe.Pointer(c))会被调用2次，因为期间goroutine可能曾经扭转copyChecker。二次调用如果不相等，则阐明sync.Cond被复制，重新分配了内存地址。

sync.Cond比拟有意思的是notifyList

`type notifyList struct {`
 `// wait is the ticket number of the next waiter. It is atomically`
 `// incremented outside the lock.`
 `wait uint32 // 期待goroutine操作的数量`
 `// notify is the ticket number of the next waiter to be notified. It can`
 `// be read outside the lock, but is only written to with lock held.`
 `//`
 `// Both wait & notify can wrap around, and such cases will be correctly`
 `// handled as long as their "unwrapped" difference is bounded by 2^31.`
 `// For this not to be the case, we'd need to have 2^31+ goroutines`
 `// blocked on the same condvar, which is currently not possible.`
 `notify uint32 // 唤醒goroutine操作的数量`
 `// List of parked waiters.`
 `lock mutex`
 `head *sudog`
 `tail *sudog`
`}`

蕴含了3类字段：

• wait和notify两个无符号整型，别离示意了Wait()操作的次数和goroutine被唤醒的次数，wait应该是恒大于等于notify
• lock mutex 这个跟sync.Mutex咱们剖析信号量阻塞队列时semaRoot里的mutex一样，并不是Go提供开发者应用的sync.Mutex,而是零碎外部运行时实现的一个简略版本的互斥锁。
• head和tail看名字，咱们就能脑补出跟链表很像 没错这里就是保护了阻塞在以后sync.Cond上的goroutine形成的链表

整体来讲sync.Cond大体构造为:

cond architecture

操作方法

Wait()操作

`func (c *Cond) Wait() {`
 `//1. 查看cond是否被拷贝`
 `c.checker.check()`
 `//2. notifyList.wait+1`
 `t := runtime_notifyListAdd(&c.notify)`
 `//3. 开释锁 让出资源给其余goroutine`
 `c.L.Unlock()`
 `//4. 挂起goroutine`
 `runtime_notifyListWait(&c.notify, t)`
 `//5. 尝试取得锁`
 `c.L.Lock()`
`}`

从Wait()办法源码很容易看出它的操作大略分了5步：

1. 调用copyChecker.check()保障sync.Cond不会被拷贝
2. 每次调用Wait()会将sync.Cond.notifyList.wait属性进行加一操作，这也是它实现FIFO的基石，依据wait来判断\`goroutine1期待的程序

`//go:linkname notifyListAdd sync.runtime_notifyListAdd`
`func notifyListAdd(l *notifyList) uint32 {`
 `// This may be called concurrently, for example, when called from`
 `// sync.Cond.Wait while holding a RWMutex in read mode.`
 `return atomic.Xadd(&l.wait, 1) - 1`
`}`

3. 调用c.L.Unlock()开释锁，因为以后goroutine行将被gopark，让出锁给其余goroutine防止死锁
4. 调用runtime_notifyListWait(&c.notify, t)可能略微简单一点儿

`// notifyListWait waits for a notification. If one has been sent since`
`// notifyListAdd was called, it returns immediately. Otherwise, it blocks.`
`//go:linkname notifyListWait sync.runtime_notifyListWait`
`func notifyListWait(l *notifyList, t uint32) {`
 `lockWithRank(&l.lock, lockRankNotifyList)`
 `// 如果曾经被唤醒 则立刻返回`
 `if less(t, l.notify) {`
 `unlock(&l.lock)`
 `return`
 `}`
 `// Enqueue itself.`
 `s := acquireSudog()`
 `s.g = getg()`
 `// 把期待递增序号赋值给s.ticket 为FIFO打基础`
 `s.ticket = t`
 `s.releasetime = 0`
 `t0 := int64(0)`
 `if blockprofilerate > 0 {`
 `t0 = cputicks()`
 `s.releasetime = -1`
 `}`
 `// 将以后goroutine插入到notifyList链表中`
 `if l.tail == nil {`
 `l.head = s`
 `} else {`
 `l.tail.next = s`
 `}`
 `l.tail = s`
 `// 最终调用gopark挂起以后goroutine`
 `goparkunlock(&l.lock, waitReasonSyncCondWait, traceEvGoBlockCond, 3)`
 `if t0 != 0 {`
 `blockevent(s.releasetime-t0, 2)`
 `}`
 `// goroutine被唤醒后开释sudog`
 `releaseSudog(s)`
`}`

次要实现两个工作：

• 将以后goroutine插入到notifyList链表中
• 调用gopark将以后goroutine挂起

5. 当其余goroutine调用了Signal或Broadcast办法，以后goroutine被唤醒后 再次尝试取得锁

Signal操作

Signal唤醒一个等待时间最长的goroutine,调用时不要求持有锁。

`func (c *Cond) Signal() {`
 `c.checker.check()`
 `runtime_notifyListNotifyOne(&c.notify)`
`}`

具体实现也不简单，先判断sync.Cond是否被复制，而后调用runtime_notifyListNotifyOne

`//go:linkname notifyListNotifyOne sync.runtime_notifyListNotifyOne`
`func notifyListNotifyOne(l *notifyList) {`
 `// wait==notify 阐明没有期待的goroutine了`
 `if atomic.Load(&l.wait) == atomic.Load(&l.notify) {`
 `return`
 `}`
 `lockWithRank(&l.lock, lockRankNotifyList)`
 `// 锁下二次查看`
 `t := l.notify`
 `if t == atomic.Load(&l.wait) {`
 `unlock(&l.lock)`
 `return`
 `}`
 `// 更新下一个须要被唤醒的ticket number`
 `atomic.Store(&l.notify, t+1)`
 `// Try to find the g that needs to be notified.`
 `// If it hasn't made it to the list yet we won't find it,`
 `// but it won't park itself once it sees the new notify number.`
 `//`
 `// This scan looks linear but essentially always stops quickly.`
 `// Because g's queue separately from taking numbers,`
 `// there may be minor reorderings in the list, but we`
 `// expect the g we're looking for to be near the front.`
 `// The g has others in front of it on the list only to the`
 `// extent that it lost the race, so the iteration will not`
 `// be too long. This applies even when the g is missing:`
 `// it hasn't yet gotten to sleep and has lost the race to`
 `// the (few) other g's that we find on the list.`
 `//这里是FIFO实现的外围 其实就是遍历链表 sudog.ticket查找指定须要唤醒的节点`
 `for p, s := (*sudog)(nil), l.head; s != nil; p, s = s, s.next {`
 `if s.ticket == t {`
 `n := s.next`
 `if p != nil {`
 `p.next = n`
 `} else {`
 `l.head = n`
 `}`
 `if n == nil {`
 `l.tail = p`
 `}`
 `unlock(&l.lock)`
 `s.next = nil`
 `readyWithTime(s, 4)`
 `return`
 `}`
 `}`
 `unlock(&l.lock)`
`}`

次要逻辑：

1. 判断是否存在期待须要被唤醒的goroutine 没有间接返回
2. 递增notify属性，因为是依据notify和sudog.ticket匹配来查找须要唤醒的goroutine,因为其是递增生成的，故而有了FIFO语义。
3. 遍历notifyList持有的链表，从head开始根据next指针顺次遍历。这个过程是线性的，故而工夫复杂度为O(n),不过官网说法这个过程理论比拟快This scan looks linear but essentially always stops quickly.

有个小细节:还记得咱们Wait()操作中，wait属性原子更新和goroutine插入期待链表是两个独自的步骤，所以存在竞争的状况下，链表中的节点可能会轻微的乱序产生。然而不要放心，因为ticket是原子递增的 所以唤醒程序不会乱。

Broadcast操作

Broadcast()与Singal()区别次要是它能够唤醒全副期待的goroutine，并间接将wait属性的值赋值给notify。

`func (c *Cond) Broadcast() {`
 `c.checker.check()`
 `runtime_notifyListNotifyAll(&c.notify)`
`}`
`// notifyListNotifyAll notifies all entries in the list.`
`//go:linkname notifyListNotifyAll sync.runtime_notifyListNotifyAll`
`func notifyListNotifyAll(l *notifyList) {`
 `// Fast-path 无期待goroutine间接返回`
 `if atomic.Load(&l.wait) == atomic.Load(&l.notify) {`
 `return`
 `}`
 `lockWithRank(&l.lock, lockRankNotifyList)`
 `s := l.head`
 `l.head = nil`
 `l.tail = nil`
 `// 间接更新notify=wait`
 `atomic.Store(&l.notify, atomic.Load(&l.wait))`
 `unlock(&l.lock)`
 `// 顺次调用goready唤醒goroutine`
 `for s != nil {`
 `next := s.next`
 `s.next = nil`
 `readyWithTime(s, 4)`
 `s = next`
 `}`
`}`

逻辑比较简单不再赘述

总结

1. sync.Cond一旦创立应用 不容许被拷贝，由noCopy和copyChecker来限度爱护。
2. Wait()操作先是递增notifyList.wait属性 而后将goroutine封装进sudog，将notifyList.wait赋值给sudog.ticket,而后将sudog插入notifyList链表中
3. Singal()理论是依照notifyList.notify跟notifyList链表中节点的ticket匹配 来确定唤醒的goroutine，因为notifyList.notify和notifyList.wait都是原子递增的，故而有了FIFO的语义
4. Broadcast()绝对简略 就是唤醒全副期待的goroutine

如果浏览过程中发现本文存疑或谬误的中央，能够关注公众号留言。如果感觉还能够 帮忙点个在看😁