被遗弃在角落里的 sync.Cond

本是古典何须时尚 2022-09-03 06:15 135阅读 0赞

Go 语言通过 go 关键字开启 goroutine 让开发者可以轻松地实现并发编程，而并发程序的有效运行，往往离不开 sync 包的保驾护航。目前，sync 包的赋能列表包括：sync.atomic 下的原子操作、sync.Map 并发安全 map、sync.Mutex 与 sync.RWMutex 提供的互斥锁与读写锁、sync.Pool 复用对象池、sync.Once 单例模式、 sync.Waitgroup 的多任务协作模式、sync.Cond 的监视器模式。当然，除了 sync 包，还有封装层面更高的 channel 与 context。

要想写出合格的 Go 程序，以上的这些并发原语是必须要掌握的。对于大多数 Gopher 而言，`sync.Cond` 应该是最为陌生，本文将一探究竟。

**初识 sync.Cond**

##  ##

`sync.Cond` 字面意义就是同步条件变量，它实现的是一种监视器（Monitor）模式。

*In concurrent programming(also known as parallel programming), a **monitor** is a synchronization construct that allows threads to have both mutual exclusion and the ability to wait (block) for a certain condition to become false.*

对于 Cond 而言，它实现一个条件变量，是 goroutine 间等待和通知的点。条件变量与共享的数据隔离，它可以同时阻塞多个 goroutine，直到另外的 goroutine 更改了条件变量，并通知唤醒阻塞着的一个或多个 goroutine。

初次接触的读者，可能会不太明白，那么下面我们看一下 GopherCon 2018 上《Rethinking Classical Concurrency Patterns》 中的演示代码例子。

1type Item = int
     2
     3type Queue struct {
     4   items     []Item
     5   itemAdded sync.Cond
     6}
     7
     8func NewQueue() *Queue {
     9   q := new(Queue)
    10   q.itemAdded.L = &sync.Mutex{} // 为 Cond 绑定锁
    11   return q
    12}
    13
    14func (q *Queue) Put(item Item) {
    15   q.itemAdded.L.Lock()
    16   defer q.itemAdded.L.Unlock()
    17   q.items = append(q.items, item)
    18   q.itemAdded.Signal()        // 当 Queue 中加入数据成功，调用 Singal 发送通知
    19}
    20
    21func (q *Queue) GetMany(n int) []Item {
    22   q.itemAdded.L.Lock()
    23   defer q.itemAdded.L.Unlock()
    24   for len(q.items) < n {     // 等待 Queue 中有 n 个数据
    25      q.itemAdded.Wait()      // 阻塞等待 Singal 发送通知
    26   }
    27   items := q.items[:n:n]
    28   q.items = q.items[n:]
    29   return items
    30}
    31
    32func main() {
    33   q := NewQueue()
    34
    35   var wg sync.WaitGroup
    36   for n := 10; n > 0; n-- {
    37      wg.Add(1)
    38      go func(n int) {
    39         items := q.GetMany(n)
    40         fmt.Printf("%2d: %2d\n", n, items)
    41         wg.Done()
    42      }(n)
    43   }
    44
    45   for i := 0; i < 100; i++ {
    46      q.Put(i)
    47   }
    48
    49   wg.Wait()
    50}

在这个例子中，`Queue` 是存储数据 `Item` 的结构体，它通过 `Cond` 类型的 `itemAdded` 来控制数据的输入与输出。可以注意到，这里通过 10 个 goroutine 来消费数据，但它们所需的数据量并不相等，我们可以称之为 batch，依次在 1-10 之间。之后，逐步添加 100 个数据至 `Queue` 中。最后，我们能够看到 10 个 gotoutine 都能被唤醒，得到它想要的数据。

程序运行结果如下

1 6: [ 7  8  9 10 11 12]
     2 5: [50 51 52 53 54]
     3 9: [14 15 16 17 18 19 20 21 22]
     4 1: [13]
     5 2: [33 34]
     6 4: [35 36 37 38]
     7 3: [39 40 41]
     8 7: [ 0  1  2  3  4  5  6]
     9 8: [42 43 44 45 46 47 48 49]
    1010: [23 24 25 26 27 28 29 30 31 32]

当然，程序每次运行结果都不会相同，以上输出只是某一种情况。

**sync.Cond 实现**

##  ##

在 `$GOPATH/src/sync/cond.go` 中，`Cond` 的结构体定义如下

1type Cond struct {
    2   noCopy noCopy
    3   L Locker
    4   notify  notifyList
    5   checker copyChecker
    6}

其中，`noCopy` 与 `checker` 字段均是为了避免 `Cond` 在使用过程中被复制，详见小菜刀的 [《no copy 机制》][no copy] 一文。

`L` 是 Locker 接口，一般该字段的实际对象是 `*RWmutex` 或者 `*Mutex`。

1type Locker interface {
    2   Lock()
    3   Unlock()
    4}

`notifyList` 记录的是一个基于票号的通知列表，这里初次看注释看不懂没关系，和下文来回连贯着看。

1type notifyList struct {
    2   wait   uint32         // 用于记录下一个等待者 waiter 的票号
    3   notify uint32         // 用于记录下一个应该被通知的 waiter 的票号
    4   lock   uintptr        // 内部锁
    5   head   unsafe.Pointer // 指向等待者 waiter 的队列队头
    6   tail   unsafe.Pointer // 指向等待者 waiter 的队列队尾
    7}

其中，`head` 与 `tail` 是指向 sudog 结构体的指针，sudog 是代表的处于等待列表的 goroutine，它本身就是双向链表。值得一提的是，在 sudog 中有一个字段 `ticket` 就是用于给当前 goroutine 记录票号使用的。

Cond 实现的核心模式为**票务系统（ticket system）**，每一个想要来买票的 goroutine （调用Cond.Wait()）我们称之为 waiter，票务系统会给每个 waiter 分配一个取票码，等供票方有该取票码的号时，就会唤醒 waiter。卖票的 goroutine 有两种，第一种是调用 Cond.Signal() 的，它会按照票号唤醒一个买票的 waiter （如果有的话），第二种是调用 Cond.Broadcast() 的，它会通知唤醒所有的阻塞 waiter。为了方便读者能够比较轻松地理解票务系统，下面我们给出图解示例。

![dc87885716af462007cb4eed37799544.png][]

在 上文中，我们知道 Cond 字段中 notifyList 结构体是一个记录票号的通知列表。这里将 notifyList 比作排队取票买电影票，当 G1 通过 Wait 来买票时，发现此时并没有票可买，因此他只能阻塞等待有票之后的通知，此时他手上已经取得了专属取票码 0。同样的，G2 和 G3 也同样无票可买，它们分别取到了自己的取票码 1和 2。而 G4 是电影票提供商，它是卖票的，它通过两次 Signal 先后带来了两张票，按照票号顺序依次通知了 G1 和 G2 来取票，并把 notify 更新为了最新的 1。G5 也是买票的，它发现此时已经无票可买了，拿了自己的取票码 3 ，就阻塞等待了。G6 是个大票商，它通过 Broadcast 可以满足所有正在等待的买票者都买到票，此时等待的是 G3 和 G5，因此他直接唤醒了 G3 和 G5，并将 notify 更新到和 wait 值相等。

理解了上述取票系统的运作原理后，我们下面来看 Cond 包下四个实际对外方法函数的实现。

*  **NewCond 方法**

1func NewCond(l Locker) *Cond {
    2   return &Cond{L: l}
    3}

用于初始化 Cond  对象，就是初始化控制锁。

*  **Cond.Wait 方法**

1func (c *Cond) Wait() {
    2   c.checker.check()
    3   t := runtime_notifyListAdd(&c.notify)
    4   c.L.Unlock()
    5   runtime_notifyListWait(&c.notify, t)
    6   c.L.Lock()
    7}

runtime\_notifyListAdd  的实现在 runtime/sema.go 的 notifyListAdd ，它用于原子性地增加等待者的 waiter 票号，并返回当前 goroutine 应该取的票号值 `t` 。runtime\_notifyListWait 的实现在runtime/sema.go 的 notifyListWait，它会尝试去比较此时 goroutine 的应取票号 `t` 与 `notify` 中记录的当前应该被通知的票号。如果 `t` 小于当前票号，那么直接能得到返回，否则将会则塞等待，通知取号。

同时，这里需要注意的是，**由于在进入 runtime\_notifyListWait 时，当前 goroutine 通过** **`c.L.Unlock()` 将锁解了，这就意味着有可能会有多个 goroutine 来让条件发生变化。那么，当前 goroutine 是不能保证在 runtime\_notifyListWait 返回后，条件就一定是真的，因此需要循环判断条件**。正确的 Wait 使用姿势如下：

1//    c.L.Lock()
    2//    for !condition() {
    3//        c.Wait()
    4//    }
    5//    ... make use of condition ...
    6//    c.L.Unlock()

*  **Cond.Signal 方法**

1func (c *Cond) Signal() {
    2   c.checker.check()
    3   runtime_notifyListNotifyOne(&c.notify)
    4}

runtime\_notifyListNotifyOne 的详细实现在 runtime/sema.go 的 notifyListNotifyOne，它的目的就是通知 waiter 取票。具体操作是：如果在上一次通知取票之后没有新的 waiter 取票者，那么该函数会直接返回。否则，它会将取票号 +1，并通知唤醒等待取票的 waiter。

需要注意的是，调用 Signal 方法时，并不需要持有 c.L 锁。

*  **Cond.Broadcast 方法**

1func (c *Cond) Broadcast() {
    2   c.checker.check()
    3   runtime_notifyListNotifyAll(&c.notify)
    4}

runtime\_notifyListNotifyAll 的详细实现在 runtime/sema.go 的 notifyListNotifyAll，它会通知唤醒所有的 waiter，并将 `notify` 值置为 和 `wait` 值相等。调用 Broadcast 方法时，也不需要持有 c.L 锁。

**讨论**

在 `$GOPATH/src/sync/cond.go` 下，我们可以发现其代码量非常之少，但它呈现的只是核心逻辑，其实现细节位于 `runtime/sema.go` 之中，**依赖的是 runtime 层的调度原语**，对细节感兴趣的读者可以深入学习。

问题来了，为什么在日常开发中，我们很少会使用到 sync.Cond ？

*  **无效唤醒**

前文中我们提到，使用 Cond.Wait 正确姿势如下

1    c.L.Lock()
    2    for !condition() {
    3        c.Wait()
    4    }
    5    ... make use of condition ...
    6    c.L.Unlock()

以文章开头的例子而言，如果在每次调用 Put 方法时，使用 Broadcast 方法唤醒所有的 waiter，那么很大概率上被唤醒的 waiter 醒来发现条件并不满足，又会重新进入等待。尽管是调用 Signal 方法唤醒指定的 waiter，但是它也不能保证唤醒的 waiter 条件一定满足。因此，在实际的使用中，我们需要尽量保证唤醒操作是有效地，为了做到这点，代码的复杂度难免会增加。

*  **饥饿问题**

还是以文章开头例子为例，如果同时有多个 goroutine 执行 GetMany(3) 和 GetMany(3000)，执行 GetMany(3) 与执行 GetMany(3000) 的 goroutine 被唤醒的概率是一样的，但是由于 GetMany(3) 只需要 3个数据就能满足条件，那么如果一直存在 GetMany(3) 的 goroutine，执行 GetMany(3000) 的 goroutine 将永远拿不到数据，一直被无效唤醒。

*  **不能响应其他事件**

条件变量的意义在于让 goroutine 等待某种条件发生时进入睡眠状态。但是这会让 goroutine 在等待条件时，可能会错过一些需要注意的其他事件。例如，调用 Cond.Wait 的函数中包含了 context 上下文，当 context 传来取消信号时，它并不能像我们期望的一样，获取到取消信号并退出。Cond 的使用，让我们不能同时选择（select）条件和其他事件。

*  **可替代性**

通过对 sync.Cond 几个对外方法的分析，我们不难看到，它的使用场景是可以被 channel 所代替的，但是这也会增加代码的复杂性。上文中的例子，可以使用 channel 改写如下。

1type Item = int
     2
     3type waiter struct {
     4    n int
     5    c chan []Item
     6}
     7
     8type state struct {
     9    items []Item
    10    wait  []waiter
    11}
    12
    13type Queue struct {
    14    s chan state
    15}
    16
    17func NewQueue() *Queue {
    18    s := make(chan state, 1)
    19    s <- state{}
    20    return &Queue{s}
    21}
    22
    23func (q *Queue) Put(item Item) {
    24    s := <-q.s
    25    s.items = append(s.items, item)
    26    for len(s.wait) > 0 {
    27        w := s.wait[0]
    28        if len(s.items) < w.n {
    29            break
    30        }
    31        w.c <- s.items[:w.n:w.n]
    32        s.items = s.items[w.n:]
    33        s.wait = s.wait[1:]
    34    }
    35    q.s <- s
    36}
    37
    38func (q *Queue) GetMany(n int) []Item {
    39    s := <-q.s
    40    if len(s.wait) == 0 && len(s.items) >= n {
    41        items := s.items[:n:n]
    42        s.items = s.items[n:]
    43        q.s <- s
    44        return items
    45    }
    46
    47    c := make(chan []Item)
    48    s.wait = append(s.wait, waiter{n, c})
    49    q.s <- s
    50
    51    return <-c
    52}

最后，虽然在上文的讨论中都是列出的 sync.Cond 潜在问题，但是如果开发者能够在使用中考虑到以上的几点问题，对于监视器模型的实现而言，在代码的语义逻辑上，sync.Cond 的使用会比 channel 的模式更易理解和维护。记住一点，通俗易懂的代码模型总是比深奥的炫技要接地气。

**往期推荐**

*  [同步原语的基石][Link 1]
 *  [详解逃逸分析][Link 2]  
    
 *  [CPU缓存体系对Go程序的影响][CPU_Go]  
    
 *  [Go中看似简单的WaitGroup源码设计，竟然暗含这么多知识？][Go_WaitGroup]  
    
 *  [一个Golang版丝滑的可视化库][Golang]

![7b2f0d23035e1720e5c0825bb52a8bbc.png][]

**机器铃砍菜刀**

欢迎添加小菜刀微信

加入Golang分享群学习交流！

感谢你的**点赞**和**在看**哦~

![7830a7eed951d12cf3af5213984b35ae.gif][]

[no copy]: https://mp.weixin.qq.com/s?__biz=MzkyMzIyNjIxMQ%3D%3D&idx=1&mid=2247484558&scene=21&sn=34b1f3112657c823cbc35e3e35ffee6b#wechat_redirect
[dc87885716af462007cb4eed37799544.png]: /images/20220829/1d19fe4f21a54a7bb79c75adf5269870.png
[Link 1]: http://mp.weixin.qq.com/s?__biz=MzIwNTA4MDAwMQ%3D%3D&chksm=97371030a04099263dfbf16dc24438f409166a053fd9fef802f9f9e4f8fd3c4aba9dd97df8f8&idx=1&mid=2247484574&scene=21&sn=5bbe2668e64455f428c70f00c09f28a2#wechat_redirect
[Link 2]: http://mp.weixin.qq.com/s?__biz=MzIwNTA4MDAwMQ%3D%3D&chksm=973710d5a04099c3c2dde737eb9de31eddb9d3707ea5885abed6a06fa7fc9855fad7de8561f7&idx=1&mid=2247484539&scene=21&sn=be23b39e21067e3017f15afb36271570#wechat_redirect
[CPU_Go]: http://mp.weixin.qq.com/s?__biz=MzIwNTA4MDAwMQ%3D%3D&chksm=973710e8a04099fe6f113c38a9e9940c0ff590877ad7ba0a0f4c383356e91619574f052a6064&idx=1&mid=2247484486&scene=21&sn=83ebd97be71caad3537502e459182f88#wechat_redirect
[Go_WaitGroup]: http://mp.weixin.qq.com/s?__biz=MzkyMzIyNjIxMQ%3D%3D&chksm=c1e91dc9f69e94df73356253f5abddded37f6002de3d40c624e33b61a50aa034be7253eb05d6&idx=1&mid=2247484613&scene=21&sn=3ee68a03ad8d558b3446f8401bd4fa26#wechat_redirect
[Golang]: http://mp.weixin.qq.com/s?__biz=MzIwNTA4MDAwMQ%3D%3D&chksm=973710afa04099b9e02d22b3ca479a7c10aa0343cbbd2e035b28fba90ef866a2ad520fde4e07&idx=1&mid=2247484417&scene=21&sn=f0169faaf62e28ae3bf13b49317acafb#wechat_redirect
[7b2f0d23035e1720e5c0825bb52a8bbc.png]: /images/20220829/8f23b28a1b514239b4291e0e52f2c7ce.png
[7830a7eed951d12cf3af5213984b35ae.gif]: /images/20220829/46ffa2a9b6934a798303467f2e372edd.png