Home >Backend Development >Golang >Let's talk in depth about sync.Cond in Golang
This article will introduce the sync.Cond
concurrency primitive in the Go language, including the basic usage and implementation principles of sync.Cond
, Precautions for use and common usage scenarios. Be able to better understand and apply Cond to achieve synchronization between goroutines.
sync.Cond
is the Go language standard library A type in that represents a condition variable. Condition variables are a mechanism for synchronization and mutual exclusion between multiple goroutines. sync.Cond
can be used to wait and notify goroutines so that they can wait or continue execution under specific conditions.
sync.Cond
is defined as follows, providing Wait
,Singal
,Broadcast
and NewCond
methods
type Cond struct { noCopy noCopy // L is held while observing or changing the condition L Locker notify notifyList checker copyChecker } func NewCond(l Locker) *Cond {} func (c *Cond) Wait() {} func (c *Cond) Signal() {} func (c *Cond) Broadcast() {}
NewCond
methods: Provide methods to create Cond
instancesWait
Method: Put the current thread into blocking state, waiting for other coroutines to wake up Singal
Method: Wake up a thread waiting for the condition variable, If no threads are waiting, the method returns immediately. Broadcast
Method: Wake up all threads waiting for the condition variable. If no thread is waiting, the method will return immediately. When using sync.Cond
, the following steps are usually required:
sync.Cond
object and associate this mutex; Wait
method to wait for the condition variable to be notified; Signal
or Broadcast
method notifies the waiting coroutine. The following is a simple example using sync.Cond to implement a producer-consumer model:
var ( // 1. 定义一个互斥锁 mu sync.Mutex cond *sync.Cond count int ) func init() { // 2.将互斥锁和sync.Cond进行关联 cond = sync.NewCond(&mu) } func worker(id int) { // 消费者 for { // 3. 在需要等待的地方,获取互斥锁,调用Wait方法等待被通知 mu.Lock() // 这里会不断循环判断 是否有待消费的任务 for count == 0 { cond.Wait() // 等待任务 } count-- fmt.Printf("worker %d: 处理了一个任务\n", id) // 5. 最后释放锁 mu.Unlock() } } func main() { // 启动5个消费者 for i := 1; i <= 5; i++ { go worker(i) } for { // 生产者 time.Sleep(1 * time.Second) mu.Lock() count++ // 4. 在需要等待的地方,获取互斥锁,调用BroadCast/Singal方法进行通知 cond.Broadcast() mu.Unlock() } }
In this example, a producer is created to produce tasks, and five consumers are created to consume tasks. When the number of tasks is 0, the consumer will call the Wait
method to enter the blocking state and wait for notification from the producer.
When the producer generates a task, it uses the Broadcast
method to notify all consumers, wake up the blocked consumers, and start consuming the task. sync.Cond
is used here to achieve communication and synchronization between multiple coroutines.
The reason here is that calling Wait
If you do not lock before the method, race conditions may occur.
It is assumed here that multiple coroutines are in a waiting state, and then a coroutine calls Broadcast to wake up one or more of the coroutines. At this time, these coroutines will be awakened.
is as follows, assuming that there is no lock before calling the Wait
method, then all coroutines will call the condition
method to determine whether the conditions are met, and then all pass verification. Perform subsequent actions.
for !condition() { c.Wait() } c.L.Lock() // 满足条件情况下,执行的逻辑 c.L.Unlock()
What will happen at this time is that the operation can only be performed if the condition
method is met. Now there is a possible effect. When the previous part of the coroutine is executed, it still satisfies the condition
condition; but the subsequent coroutine, although it does not meet the condition
condition, still performs subsequent operations. May cause program errors.
The correct usage should be to lock before calling the Wait
method. Then even if multiple coroutines are awakened, only one coroutine will judge whether it is satisfied at a time condition
condition, and then perform subsequent operations. In this way, multiple coroutines will not be judged at the same time, resulting in the situation that the conditions are not met and subsequent operations are performed.
c.L.Lock() for !condition() { c.Wait() } // 满足条件情况下,执行的逻辑 c.L.Unlock()
sync.Cond
is for coordination Designed for access to shared data between multiple coroutines. The scenarios for using sync.Cond
usually involve the operation of shared data. If there is no operation of shared data, then it is not necessary to use sync.Cond
to coordinate. Of course, if there is a scenario of repeated wake-ups, sync.Cond
can be used for coordination even if there is no operation on shared data.
Usually, the scenario of using sync.Cond
is: multiple coroutines need to access the same shared data, and they need to wait for a certain condition to be met before they can access or modify the shared data. .
在这些场景下,使用sync.Cond
可以方便地实现对共享数据的协调,避免了多个协程之间的竞争和冲突,保证了共享数据的正确性和一致性。因此,如果没有涉及到共享数据的操作,就没有必要使用sync.Cond
来进行协调。
下面举一个使用 sync.Cond
的例子,用它来实现生产者-消费者模型。生产者往items
放置元素,当items
满了之后,便进入等待状态,等待消费者唤醒。消费者从items
中取数据,当items
空了之后,便进入等待状态,等待生产者唤醒。
这里多个协程对同一份数据进行操作,且需要基于该数据判断是否唤醒其他协程或进入阻塞状态,来实现多个协程的同步和协调。sync.Cond
就适合在这种场景下使用,其正是为这种场景设计的。
package main import ( "fmt" "sync" "time" ) type Queue struct { items []int cap int lock sync.Mutex cond *sync.Cond } func NewQueue(cap int) *Queue { q := &Queue{ items: make([]int, 0), cap: cap, } q.cond = sync.NewCond(&q.lock) return q } func (q *Queue) Put(item int) { q.lock.Lock() defer q.lock.Unlock() for len(q.items) == q.cap { q.cond.Wait() } q.items = append(q.items, item) q.cond.Broadcast() } func (q *Queue) Get() int { q.lock.Lock() defer q.lock.Unlock() for len(q.items) == 0 { q.cond.Wait() } item := q.items[0] q.items = q.items[1:] q.cond.Broadcast() return item } func main() { q := NewQueue(10) // Producer go func() { for { q.Put(i) fmt.Printf("Producer: Put %d\n", i) time.Sleep(100 * time.Millisecond) } }() // Consumer go func() { for { item := q.Get() fmt.Printf("Consumer: Get %d\n", item) time.Sleep(200 * time.Millisecond) } }() wg.Wait() }
在某些场景中,由于不满足某种条件,此时协程进入阻塞状态,等待条件满足后,由其他协程唤醒,再继续执行。在整个流程中,可能会多次进入阻塞状态,多次被唤醒的情况。
比如上面生产者和消费者模型的例子,生产者可能会产生一批任务,然后唤醒消费者,消费者消费完之后,会进入阻塞状态,等待下一批任务的到来。所以这个流程中,协程可能多次进入阻塞状态,然后再多次被唤醒。
sync.Cond
能够实现即使协程多次进入阻塞状态,也能重复唤醒该协程。所以,当出现需要实现重复唤醒的场景时,使用sync.Cond
也是非常合适的。
在Sync.Cond
存在一个通知队列,保存了所有处于等待状态的协程。通知队列定义如下:
type notifyList struct { wait uint32 notify uint32 lock uintptr // key field of the mutex head unsafe.Pointer tail unsafe.Pointer }
当调用Wait
方法时,此时Wait
方法会释放所持有的锁,然后将自己放到notifyList
等待队列中等待。此时会将当前协程加入到等待队列的尾部,然后进入阻塞状态。
当调用Signal
时,此时会唤醒等待队列中的第一个协程,其他继续等待。如果此时没有处于等待状态的协程,调用Signal
不会有其他作用,直接返回。当调用BoradCast
方法时,则会唤醒notfiyList
中所有处于等待状态的协程。
sync.Cond
的代码实现比较简单,协程的唤醒和阻塞已经由运行时包实现了,sync.Cond
的实现直接调用了运行时包提供的API。
Wait
方法首先调用runtime_notifyListAd
方法,将自己加入到等待队列中,然后释放锁,等待其他协程的唤醒。
func (c *Cond) Wait() { // 将自己放到等待队列中 t := runtime_notifyListAdd(&c.notify) // 释放锁 c.L.Unlock() // 等待唤醒 runtime_notifyListWait(&c.notify, t) // 重新获取锁 c.L.Lock() }
Singal
方法调用runtime_notifyListNotifyOne
唤醒等待队列中的一个协程。
func (c *Cond) Signal() { // 唤醒等待队列中的一个协程 runtime_notifyListNotifyOne(&c.notify) }
Broadcast
方法调用runtime_notifyListNotifyAll
唤醒所有处于等待状态的协程。
func (c *Cond) Broadcast() { // 唤醒等待队列中所有的协程 runtime_notifyListNotifyAll(&c.notify) }
在上面2.5已经说明了,调用Sync.Cond
方法前需要加锁,否则有可能出现竞态条件。而且,现有的sync.Cond
的实现,如果在调用Wait
方法前未加锁,此时会直接panic
,下面是一个简单例子的说明:
package main import ( "fmt" "sync" "time" ) var ( count int cond *sync.Cond lk sync.Mutex ) func main() { cond = sync.NewCond(&lk) wg := sync.WaitGroup{} wg.Add(2) go func() { defer wg.Done() for { time.Sleep(time.Second) count++ cond.Broadcast() } }() go func() { defer wg.Done() for { time.Sleep(time.Millisecond * 500) //cond.L.Lock() for count%10 != 0 { cond.Wait() } t.Logf("count = %d", count) //cond.L.Unlock() } }() wg.Wait() }
上面代码中,协程一每隔1s,将count字段的值自增1,然后唤醒所有处于等待状态的协程。协程二执行的条件为count的值为10的倍数,此时满足执行条件,唤醒后将会继续往下执行。
但是这里在调用sync.Wait
方法前,没有先获取锁,下面是其执行结果,会抛出 fatal error: sync: unlock of unlocked mutex 错误,结果如下:
count = 0 fatal error: sync: unlock of unlocked mutex
因此,在调用Wait
方法前,需要先获取到与sync.Cond
关联的锁,否则会直接抛出异常。
调用sync.Wait
方法,协程进入阻塞状态后被唤醒,没有重新检查条件变量,此时有可能仍然处于不满足条件变量的场景下。然后直接执行后续操作,有可能会导致程序出错。下面举一个简单的例子:
package main import ( "fmt" "sync" "time" ) var ( count int cond *sync.Cond lk sync.Mutex ) func main() { cond = sync.NewCond(&lk) wg := sync.WaitGroup{} wg.Add(3) go func() { defer wg.Done() for { time.Sleep(time.Second) cond.L.Lock() // 将flag 设置为true flag = true // 唤醒所有处于等待状态的协程 cond.Broadcast() cond.L.Unlock() } }() for i := 0; i < 2; i++ { go func(i int) { defer wg.Done() for { time.Sleep(time.Millisecond * 500) cond.L.Lock() // 不满足条件,此时进入等待状态 if !flag { cond.Wait() } // 被唤醒后,此时可能仍然不满足条件 fmt.Printf("协程 %d flag = %t", i, flag) flag = false cond.L.Unlock() } }(i) } wg.Wait() }
在这个例子,我们启动了一个协程,定时将flag
设置为true,相当于每隔一段时间,便满足执行条件,然后唤醒所有处于等待状态的协程。
然后又启动了两个协程,在满足条件的前提下,开始执行后续操作,但是这里协程被唤醒后,没有重新检查条件变量,具体看第39行。这里会出现的场景是,第一个协程被唤醒后,此时执行后续操作,然后将flag
重新设置为false,此时已经不满足条件了。之后第二个协程唤醒后,获取到锁,没有重新检查此时是否满足执行条件,直接向下执行,这个就和我们预期不符,可能会导致程序出错,代码执行效果如下:
协程 1 flag = true 协程 0 flag = false 协程 1 flag = true 协程 0 flag = false
可以看到,此时协程0执行时,flag
的值均为false
,说明此时其实并不符合执行条件,可能会导致程序出错。因此正确用法应该像下面这样子,被唤醒后,需要重新检查条件变量,满足条件之后才能继续向下执行。
c.L.Lock() // 唤醒后,重新检查条件变量是否满足条件 for !condition() { c.Wait() } // 满足条件情况下,执行的逻辑 c.L.Unlock()
本文介绍了 Go 语言中的 sync.Cond 并发原语,它是用于实现 goroutine 之间的同步的重要工具。我们首先学习了 sync.Cond
的基本使用方法,包括创建和使用条件变量、使用Wait
和Signal
/Broadcast
方法等。
接着,我们对 sync.Cond
的使用场景进行了说明,如同步和协调多个协程之间共享资源等。
在接下来的部分中,我们介绍了 sync.Cond
的实现原理,主要是对等待队列的使用,从而sync.Cond
有更好的理解,能够更好得使用它。同时,我们也讲述了使用sync.Cond
的注意事项,如调用Wait
方法前需要加锁等。
基于以上内容,本文完成了对 sync.Cond
的介绍,希望能够帮助大家更好地理解和使用Go语言中的并发原语。
推荐学习:Golang教程
The above is the detailed content of Let's talk in depth about sync.Cond in Golang. For more information, please follow other related articles on the PHP Chinese website!