一文读懂 Go sync.Cond 设计

机器铃砍菜刀 - 2021-10-27 13:23:21

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》中的演示代码例子。

type Item = int

type Queue struct {
   items     []Item
   itemAdded sync.Cond
}

func NewQueue() *Queue {
   q := new(Queue)
   q.itemAdded.L = &sync.Mutex{} // 为 Cond 绑定锁
   return q
}

func (q *Queue) Put(item Item) {
   q.itemAdded.L.Lock()
   defer q.itemAdded.L.Unlock()
   q.items = append(q.items, item)
   q.itemAdded.Signal()        // 当 Queue 中加入数据成功，调用 Singal 发送通知
}

func (q *Queue) GetMany(n int) []Item {
   q.itemAdded.L.Lock()
   defer q.itemAdded.L.Unlock()
   for len(q.items) < n {     // 等待 Queue 中有 n 个数据
      q.itemAdded.Wait()      // 阻塞等待 Singal 发送通知
   }
   items := q.items[:n:n]
   q.items = q.items[n:]
   return items
}

func main() {
   q := NewQueue()

   var wg sync.WaitGroup
   for n := 10; n > 0; n-- {
      wg.Add(1)
      go func(n int) {
         items := q.GetMany(n)
         fmt.Printf("%2d: %2d\n", n, items)
         wg.Done()
      }(n)
   }

   for i := 0; i < 100; i++ {
      q.Put(i)
   }

   wg.Wait()
}

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

程序运行结果如下

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

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

sync.Cond 实现

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

type Cond struct {
   noCopy noCopy
   L Locker
   notify  notifyList
   checker copyChecker
}

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

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

type Locker interface {
   Lock()
   Unlock()
}

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

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

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

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

在上文中，我们知道 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 方法

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

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

Cond.Wait 方法

func (c *Cond) Wait() {
   c.checker.check()
   t := runtime_notifyListAdd(&c.notify)
   c.L.Unlock()
   runtime_notifyListWait(&c.notify, t)
   c.L.Lock()
}

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 使用姿势如下：

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

Cond.Signal 方法

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

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

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

Cond.Broadcast 方法

func (c *Cond) Broadcast() {   c.checker.check()   runtime_notifyListNotifyAll(&c.notify)}

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 正确姿势如下

    c.L.Lock()    for !condition() {        c.Wait()    }    ... make use of condition ...    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 改写如下。

type Item = inttype waiter struct {    n int    c chan []Item}type state struct {    items []Item    wait  []waiter}type Queue struct {    s chan state}func NewQueue() *Queue {    s := make(chan state, 1)    s <- state{}    return &Queue{s}}func (q *Queue) Put(item Item) {    s := <-q.s    s.items = append(s.items, item)    for len(s.wait) > 0 {        w := s.wait[0]        if len(s.items) < w.n {            break        }        w.c <- s.items[:w.n:w.n]        s.items = s.items[w.n:]        s.wait = s.wait[1:]    }    q.s <- s}func (q *Queue) GetMany(n int) []Item {    s := <-q.s    if len(s.wait) == 0 && len(s.items) >= n {        items := s.items[:n:n]        s.items = s.items[n:]        q.s <- s        return items    }    c := make(chan []Item)    s.wait = append(s.wait, waiter{n, c})    q.s <- s    return <-c}

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

特别申明：本文内容来源网络，版权归原作者所有，如有侵权请立即与我们联系（cy198701067573@163.com），我们将及时处理。