Golang package sync 剖析(三):sync.Cond


一、前言

Go语言在设计上对同步(Synchronization,数据同步和线程同步)提供大量的支持,比如 goroutine和channel同步原语,库层面有

1. sync:提供基本的同步原语(比如Mutex、RWMutex、Locker)和 工具类(Once、WaitGroup、Cond、Pool、Map)

2. sync/atomic:提供原子操作(基于硬件指令compare-and-swap)

引用自 。

上一期中,我们介绍了如何使用 sync.WaitGroup 提高程序的并行度。本期文章我们介绍 package sync 下的另一个工具类:sync.Cond。

sync.Cond 对标 同步原语“条件变量”,它可以阻塞一个,或同时阻塞多个线程,直到另一个线程 1) 修改了条件变量; 2)通知一个(或所有)等待的线程。

注:在使用层面上,Go语言里没有线程,只有更轻量级的协程。本文中,“线程”均代指“协程”(goroutine)。

相对于 sync.Once 和 sync.WaitGroup, sync.Cond 比较难以理解,使用门槛也很高,在 Google 上搜一下,排名前10结果中有这样几个:

关键词“golang sync.Cond”部分Google搜索结果

非常神奇的是:一篇名为 “如何正确使用sync.Cond” 的帖子竟然有 16k 的浏览量!

如何正确使用sync.Cond

究竟是条件变量这个概念难以理解,还是 sync.Cond 的设计太反人类,我们一探究竟。

二、sync.Cond 怎么用

开篇我们就提到了条件变量的应用场景,我们回顾一下:

sync.Cond 对标 同步原语“条件变量”,它可以阻塞一个,或同时阻塞多个线程,直到另一个线程 1) 修改了条件变量; 2)通知一个(或所有)等待的线程。

首先,我们把概念搞清楚,条件变量的作用是控制多个线程对一个共享变量的读写。我们有三类主体:

共享变量:条件变量控制多个线程对该变量的读写;等待线程:被条件变量阻塞的线程,有一个或多个;更新线程:更新共享变量,并唤起一个或多个等待线程。

其次,我们看看 sync.Cond 的说明书:

<code>// 创建一个 sync.Cond 对象func NewCond(l Locker) *Cond// 阻塞当前线程,并等待条件触发func (c *Cond) Wait()// 唤醒所有等待线程func (c *Cond) Broadcast()// 唤起一个等待线程// 没有等待线程也不会报错func (c *Cond) Signal()/<code>

大家看完这段代码,脑子里第一个问题大概是:NewCond 要一把锁是干嘛用的?为了便于理解,我们以 kubernetes 源码里 FIFO 队列为例,一步一步说 sync.Cond 的用法:

<code>type FIFO struct { // lock 控制对象读写 lock sync.RWMutex // 阻塞Pop操作,Add成功后激活被阻塞线程 cond sync.Cond // items 存储数据 items map[string]interface{} // queue 存储key queue []string // keyFunc是hash函数 keyFunc KeyFunc // 维护items和queue同步 populated bool initialPopulationCount int // 队列状态:是否已经关闭 closed bool closedLock sync.Mutex}/<code>

首先,这是一个 FIFO 队列,问题又来了:go 内置的 channel 不香吗?还真的是不够香。

真香警告

FIFO 具备一些额外的特性:

支持自定义处理函数,并保障每个元素只被处理一次(exactly once);支持元素去重,版本更新,并只处理最新版本,而不是每次更新都处理一次;支持元素删除,删除的元素不进行处理;支持 list 所有元素。

FIFO 的成员函数有:

<code>// 从队头取一个元素,没有则会被阻塞Pop(PopProcessFunc) (interface{}, error)// 向队尾加一个元素,如果已经存在,则不做任何操作Add(obj interface{}) errorAddIfNotPresent(interface{}) error// 更新元素Update(obj interface{}) error// 删除元素Delete(obj interface{}) error// 关闭队列Close()// 读取所有元素List() []interface{}// 读取所有 keyListKeys() []string// 通过元素读取元素(通过 keyFunc 映射到同样的 key)Get(obj interface{}) (item interface{}, exists bool, err error)// 通过key读取元素GetByKey(key string) (item interface{}, exists bool, err error)// 用传入的数组替换队列内容Replace([]interface{}, string) error// 同步items和queueResync() error// items和queue是否同步HasSynced() bool/<code>

回到本文的主题 sync.Cond, 在上面这个例子中

一个 FIFO 实例就是一个共享变量;调用 Pop 的线程是等待线程;调用 Add 的线程是更新线程

lock sync.RWMutex 用于控制对共享变量的并发访问,本质上是控制对 queue 和 items 两个字段的并发访问。由于条件变量 cond sync.Cond 在实现 Wait 时,把锁操作也包含进去了,所以初始化时需要传入一个锁变量。在使用时,是这样的:

<code>// 初始化一个 FIFOfunc NewFIFO(keyFunc KeyFunc) *FIFO { // lock 和 cond 均是默认值 f := &FIFO{ items: map[string]interface{}{}, queue: []string{}, keyFunc: keyFunc, } // 将 lock 共享给 cond f.cond.L = &f.lock return f}// Pop 操作func (f *FIFO) Pop(process PopProcessFunc) (interface{}, error) { // 锁住共享变量 f.lock.Lock() defer f.lock.Unlock() for { for len(f.queue) == 0 { // 队列已关闭 if f.IsClosed() { return nil, ErrFIFOClosed } // 队列为空,等待数据 f.cond.Wait() } // 此处省略一段代码... // 从 items 和 queue 删除元素 }}// Add 操作func (f *FIFO) Add(obj interface{}) error { id, err := f.keyFunc(obj) if err != nil { return KeyError{obj, err} } // 锁住共享变量 f.lock.Lock() defer f.lock.Unlock() // 此处省略一段代码 ... // 添加元素到 items 和 queue // 通知等待线程 f.cond.Broadcast() return nil}/<code>

上面的代码中,等待线程做的是:

给共享变量加锁有数据,就返回数据;没有数据就调用 Wait 等数据

更新线程做的是:

给共享变量加锁写入数据,调用 Broadcast

看起来很简单,Ok? 但是你品一品,你细品,发现事情没那么简单。

周冬雨的凝视

等待线程 加锁以后,更新线程 要更新共享变量,怎么会取到锁呢?

我们先看看官方文档对 Wait 的解释:

Wait atomically unlocks c.L and suspends execution of the calling goroutine. After later resuming execution, Wait locks c.L before returning.

https://golang.org/pkg/sync/#Cond.Wait

大概意思是: Wait 首先会解锁 c.L,然后阻塞当前的协程;后续协程被 Broadcast/Signal 唤醒以后,在对 c.L 加锁,然后 return。

所以,cond sync.Cond 的初始化需要一把锁,并且和 FIFO 实例用同一把锁。

三、sync.Cond 实现

如果不考虑 runtime 如何实现阻塞和激活,sync.Cond 本身的实现逻辑还是比较简单的。我们看下源码(删减版):

<code>type Cond struct { noCopy noCopy // 共享变量被访问前,必须取到锁 L L Locker notify notifyList checker copyChecker}// Waitfunc (c *Cond) Wait() { // 给当前协程分配一张船票 t := runtime_notifyListAdd(&c.notify) // 解锁 c.L.Unlock() // 暂定当前协程的执行,等通知 runtime_notifyListWait(&c.notify, t) // 加锁 c.L.Lock()}// Signal 唤醒被 c 阻塞的一个协程(如果有)func (c *Cond) Signal() { runtime_notifyListNotifyOne(&c.notify)}// Broadcast 唤醒所有被 c 阻塞的协程func (c *Cond) Broadcast() { runtime_notifyListNotifyAll(&c.notify)}/<code>

这里着重说下 runtime_* 函数的功能:

runtime_notifyListAdd 将当前线程添加到通知列表,以能够接收通知;runtime_notifyListWait 将当前协程休眠,接收到通知以后才会被唤醒;runtime_notifyListNotifyOne 发送通知,唤醒 notify 列表里一个协程runtime_notifyListNotifyAll 发送通知,唤醒 notify 列表里所有协程

四、总结

sync.Cond 是Go语言对条件变量的一个实现方式,但不是唯一的方式。本质上,sync.Once 和 channel 也是条件变量的实现。

sync.Once 里锁和原子操作用于控制共享变量的读写;channel 通过 close(ch) 可以通知其他协程读取数据;

但 sync.Once 和 channel 有一个明显的缺点是:它们都只能保证第一次满足条件变量,而 sync.Cond 可以提供持续的保障。

由于 sync.Cond 的复杂性(我认为是 godoc 写的太差了),且应用场景相对较少,其出现频次低于 sync.Once 和 sync.WaitGroup。不过在合适的应用场景出现时,它就会展示出自己的不可替代性。

References

C++ std::condition_variable https://en.cppreference.com/w/cpp/thread/condition_variablekubernetes FIFO queue https://github.com/kubernetes/client-go/blob/master/tools/cache/fifo.go