1. Introduction
This article will introduce the Go language Concurrent primitives, includingBasic usage methods, implementation principles, usage precautions and common usage scenarios. Better understand and apply Cond to achieve synchronization between goroutines.
2. Basic use
2.1 Definition
It is a type in the Go language standard library, representing condition variables. Conditional variables are a mechanism for synchronization and mutual exclusion between multiple goroutines.Can be used to wait and notify goroutines so that they can wait or continue to execute under certain conditions.
2.2 Method Description
The definition ofWait ,Singal,Broadcastas well asNewCondmethod
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() {}
-
NewCondMethod: Provide creationCondExample method -
WaitMethod: Make the current thread enter a blocking state and wait for other coroutines to wake up -
SingalMethod: Wake up a thread waiting for the condition variable. If no thread is waiting, the method will return immediately. -
BroadcastMethod: Wake up all threads waiting for this condition variable. If no thread is waiting, the method will return immediately.
2.3 How to use
When usingWhen you are in the process of , the following steps are usually required:
- Define a mutex to protect shared data;
- Create a
Object, associate this mutex; - Where you need to wait for the condition variable, get this mutex and use
WaitThe method waits for the condition variable to be notified; - When you need to notify the waiting coroutine, use
SignalorBroadcastMethod notifies waiting coroutines. - Finally, release this mutex.
2.4 Use examples
Here is a simple example of usage that implements a producer-consumer model:
var (
// 1. Define a mutex lock mu
cond *
count int
)
func init() {
// 2. Associate the mutex and cond = (&mu)
}
func worker(id int) {
// Consumer for {
// 3. Get the mutex lock where you need to wait, call the Wait method and wait for notification ()
// Here we will continue to judge whether there are tasks to be consumed. for count == 0 {
() // Wait for the task }
count--
("worker %d: handled a task\n", id)
// 5. Release the lock at last ()
}
}
func main() {
// Start 5 consumers for i := 1; i <= 5; i++ {
go worker(i)
}
for {
// Producer (1 * )
()
count++
// 4. Get the mutex lock where you need to wait and call the BroadCast/Singal method to notify ()
()
}
}
In this example, a producer is created to produce tasks while five consumers are created to consume tasks. When the number of tasks is 0, the consumer will callWaitThe method enters a blocking state, waiting for the producer's notification.
When the producer generates a task, useBroadcastMethod notifies all consumers, wake up the blocked consumers, and start the consumption task. Used hereImplement communication and synchronization between multiple coroutines.
2.5 Why do you need to associate a lock and then you need to obtain the lock before calling the Wait method
The reason here is the callWaitIf no lock is added before the method, race conditions may occur.
Here, it is assumed that multiple coroutines are in a waiting state, and then a coroutine calls Broadcast to wake up one or more of the coroutines, and these coroutines will be awakened.
Assume that the call isWaitIf there is no lock before the method, then all coroutines will be called.conditionThe method determines whether the conditions are met, and then passes verification and performs subsequent operations.
for !condition() {
()
}
()
// Execution logic when conditions are met()
What will happen at this time is that it originally needs to be metconditionOperations that can only be performed under the premise of a method. Now there is a possible effect, and it is still satisfactory when the previous part of the coroutine is executed.conditionconditional; but the subsequent coroutines are not satisfiedconditionConditions, subsequent operations are still performed, which may cause program errors.
The correct usage should be, invokingWaitLock is added before the method, so even if multiple coroutines are awakened, only one coroutine will determine whether it is satisfied at a time.conditioncondition, and then perform subsequent operations. In this way, there will be no multiple coroutines to make judgments at the same time, resulting in the failure to meet the conditions and perform subsequent operations.
()
for !condition() {
()
}
// Execution logic when conditions are met()
3. Usage scenarios
3.1 Basic description
It is designed to coordinate access to shared data between multiple coroutines. useScenarios usually involve operations on shared data. If there is no operation on shared data, then there is no need to use it.to coordinate. Of course, if there is a repeated wake-up scenario, even if there is no operation on shared data, it can be usedto coordinate.
Normally, useThe scenario is: multiple coroutines need to access the same shared data, and they need to wait until a certain condition is met before accessing or modifying the shared data.
In these scenarios, useCoordination of shared data can be conveniently achieved, avoiding competition and conflict between multiple coroutines, and ensuring the correctness and consistency of shared data. Therefore, if there is no operation involving sharing data, there is no need to use itto coordinate.
3.2 Scene Description
3.2.1 Synchronize and coordinate sharing resources between multiple coroutines
Here is one to useExamples of using it to implement the producer-consumer model. ProducersitemsPlace elements whenitemsAfter it is full, it enters a waiting state, waiting for the consumer to wake up. Consumers fromitemsfetch data whenitemsAfter being empty, it enters a waiting state, waiting for the producer to wake up.
Here, multiple coroutines operate on the same data, and it is necessary to determine whether other coroutines are awakened or entered a blocking state based on the data to realize synchronization and coordination of multiple coroutines.It is suitable for use in this scenario, and it is designed for this scenario.
package main
import (
"fmt"
"sync"
"time"
)
type Queue struct {
items []int
cap int
lock
cond *
}
func NewQueue(cap int) *Queue {
q := &Queue{
items: make([]int, 0),
cap: cap,
}
= (&)
return q
}
func (q *Queue) Put(item int) {
()
defer ()
for len() == {
()
}
= append(, item)
()
}
func (q *Queue) Get() int {
()
defer ()
for len() == 0 {
()
}
item := [0]
= [1:]
()
return item
}
func main() {
q := NewQueue(10)
// Producer
go func() {
for {
(i)
("Producer: Put %d\n", i)
(100 * )
}
}()
// Consumer
go func() {
for {
item := ()
("Consumer: Get %d\n", item)
(200 * )
}
}()
()
}
3.2.2 Used in scenarios that require repeated wake-up
In some scenarios, since certain conditions are not met, the coroutine enters a blocking state at this time. After the conditions are met, the other coroutines will wake up and continue to execute. During the entire process, it may enter a blocked state multiple times and be awakened multiple times.
For example, in the example of the producer and consumer model above, the producer may generate a batch of tasks and then awaken the consumer. After the consumer spends, it will enter a blocked state and wait for the next batch of tasks to arrive. So in this process, the coroutine may enter the blocking state many times and then be awakened multiple times.
It can realize that even if the coroutine enters the blocking state multiple times, the coroutine can be repeatedly awakened. Therefore, when a scene that requires repeated wake-up occurs, useIt's also very suitable.
4. Principle
4.1 Basic Principles
existThere is a notification queue that saves all coroutines in a waiting state. The notification queue is defined as follows:
type notifyList struct {
wait uint32
notify uint32
lock uintptr // key field of the mutex
head
tail
}
When calledWaitWhen the method isWaitThe method releases the lock held and then places itself innotifyListWait in the queue. At this time, the current coroutine will be added to the tail of the waiting queue and then enter a blocking state.
When calledSignalWhen , the first coroutine in the waiting queue will be awakened, and the others will continue to wait. If there is no waiting coroutine at this time, callSignalThere will be no other effect, please return directly. When calledBoradCastWhen the method is used, it will wake upnotfiyListAll coroutines in waiting state.
The code implementation is relatively simple, and the wake-up and blocking of coroutines have been implemented by the runtime package.The implementation directly calls the API provided by the runtime package.
4.2 Implementation
4.2.1 Wait method implementation
WaitThe method is called firstruntime_notifyListAdMethod: add yourself to the waiting queue, then release the lock, and wait for other coroutines to wake up.
func (c *Cond) Wait() {
// Put yourself in the waiting queue t := runtime_notifyListAdd(&)
// Release the lock ()
// Wait for wake up runtime_notifyListWait(&, t)
// Reacquire the lock ()
}
4.2.2 Singal method implementation
SingalMethod callruntime_notifyListNotifyOneWake up a coroutine in the waiting queue.
func (c *Cond) Signal() {
// Wake up a coroutine in the waiting queue runtime_notifyListNotifyOne(&)
}
4.2.3 Broadcast method implementation
BroadcastMethod callruntime_notifyListNotifyAllWake up all waiting coroutines.
func (c *Cond) Broadcast() {
// Wake up all coroutines in the waiting queue runtime_notifyListNotifyAll(&)
}
5. Precautions for use
5.1 Unlocked before calling Wait method
As already stated in 2.5 above, callLocking is required before the method, otherwise race conditions may occur. And, the existing onesImplementation, if invokedWaitThe method has not been locked before, and it will be directlypanic, Here is a brief example:
package main
import (
"fmt"
"sync"
"time"
)
var (
count int
cond *
lk
)
func main() {
cond = (&lk)
wg := {}
(2)
go func() {
defer ()
for {
()
count++
()
}
}()
go func() {
defer ()
for {
( * 500)
//()
for count%10 != 0 {
()
}
("count = %d", count)
//()
}
}()
()
}
In the above code, every 1s, the coroutine increments the value of the count field by 1, and then wakes up all coroutines in the waiting state. The execution condition of coroutine 2 is a multiple of count value of 10. At this time, the execution condition is met and the execution will continue to be executed after wake-up.
But here is callingBefore the method, the lock is not acquired first. The following is its execution result. Fatal error: sync: unlock of unlocked mutex error will be thrown, and the result is as follows:
count = 0 fatal error: sync: unlock of unlocked mutex
Therefore, invokingWaitBefore the method, you need to obtain theThe associated lock will otherwise an exception will be thrown directly.
5.2 After the Wait method receives the notification, the condition variable is not checked again.
CallMethod: The coroutine is awakened after entering the blocking state and the condition variable is not rechecked. At this time, it may still be in a scenario where the condition variable does not meet it. Then perform subsequent operations directly, which may cause program errors. Here is a simple example:
package main
import (
"fmt"
"sync"
"time"
)
var (
count int
cond *
lk
)
func main() {
cond = (&lk)
wg := {}
(3)
go func() {
defer ()
for {
()
()
// Set flag to true flag = true
// Wake up all waiting coroutines ()
()
}
}()
for i := 0; i < 2; i++ {
go func(i int) {
defer ()
for {
( * 500)
()
// The condition is not met, and the waiting state is entered at this time if !flag {
()
}
// After being awakened, the conditions may still not be met at this time ("Coroutine %d flag = %t", i, flag)
flag = false
()
}
}(i)
}
()
}
In this example, we start a coroutine, and the timing willflagSet to true, which is equivalent to satisfying the execution conditions every once in a while, and then awakening all coroutines in the waiting state.
Then two coroutines were started, and subsequent operations were performed on the premise that the conditions were met. However, after the coroutines were awakened, the condition variables were not rechecked. See Line 39 for details. The scenario that will appear here is that after the first coroutine is awakened, the subsequent operation is performed, and then theflagReset to false, the condition is no longer met at this time. After the second coroutine wakes up, the lock is obtained, and the execution conditions are not checked again at this time, and it is executed directly downward. This is inconsistent with our expectations and may cause program errors. The code execution effect is as follows:
Coroutine 1 flag = true
Coroutine 0 flag = false
Coroutine 1 flag = true
Coroutine 0 flag = false
You can see that when coroutine 0 is executed,flagAll values arefalse, which means that the execution conditions are not met at this time, which may cause program errors. Therefore, the correct usage should be like the following. After being awakened, the condition variable needs to be rechecked and the conditions can only be executed downwards after the conditions are met.
()
// After wake up, recheck whether the condition variable meets the conditionfor !condition() {
()
}
// Execution logic when conditions are met()
6. Summary
This article introduces the concurrent primitives in the Go language, which are an important tool for synchronizing goroutines. We first learnedThe basic usage methods include creating and using condition variables and usingWaitandSignal/BroadcastMethods, etc.
Next, weThe usage scenarios are explained, such as synchronizing and coordinating the sharing of resources between multiple coroutines.
In the following sections, we introduceThe implementation principle is mainly to use the waiting queue, soHave a better understanding and be able to use it better. At the same time, we also talk about usingNotes, such as callingWaitLocking is required before the method.
Based on the above content, this article has completed theI hope that this can help you better understand and use concurrent primitives in Go.
For more information about the principles of Go, please follow my other related articles!