WaitGroup usage scenarios and methods
When we have many tasks to be carried out at the same time, if we do not need to care about the execution progress of each task, then just use the go keyword.
If we need to care about all tasks before running down, we need WaitGroup to block and wait for these concurrent tasks.
WaitGroup, as it literally means waiting for a group of goroutines to complete, consisting of three main methods:
- Add(delta int) : Add the number of tasks
- Wait(): Blocking and waiting for all tasks to complete
- Done(): Complete the task
Here are their specific usages, and their specific functions are all in the comments:
package main
import (
"fmt"
"sync"
"time"
)
func worker(wg *) {
doSomething()
() // 2.1. Complete the task}
func main() {
var wg
(5) // 1. Add 5 tasks for i := 1; i <= 5; i++ {
go worker(&wg) // 2. Each task is executed concurrently }
() // 3. Block and wait for all tasks to complete}WaitGroup Source Code Analysis
The use of WaitGroup above is very simple. Next, we go to src/sync/ to analyze its source code. First, there is the structure of WaitGroup:
type WaitGroup struct {
noCopy noCopy
state1 [3]uint32
}
noCopy
Among them, noCopy means that WaitGroup is not replicable. So what does not mean to be replicated?
For example, when we define this uncopyable type for a function parameter, the developer can only pass the function parameter through a pointer. What are the benefits of using pointers to pass?
The advantage is that if multiple functions define this uncopyable parameter, then the multiple function parameters can share the same pointer variable to synchronize the execution results. And WaitGroup requires such constraints.
state1 field
Next, let's take a look at the state1 field of WaitGroup. state1 is an uint32 array containing the total number of counters, waiter waits, and sema sema.
Whenever a goroutine calls the Wait() method to block the wait, it will +1 for the waiter number and then wait for the semaphore to be called to notify the call.
When we call the Add() method, we will + 1 for the counter count of state1.
The number of counters is -1 when the Done() method is called.
Until counter == 0, the corresponding number of goroutines can be evoked through the semaphore, that is, the goroutines that have just blocked and waited.
For the explanation of semaphore, please refer to the followinggolang Important knowledge: mutexRelated introductions in:
PV primitive explanation:
The synchronization and mutual exclusion problems between processes are handled by operating the semaphore S.
S>0: means that S resources are available; S=0 means that no resources are available; S<0 absolute value indicates the number of processes in the waiting queue or linked list. The initial value of the semaphore S should be greater than or equal to 0.
P primitive: It means applying for a resource and decrement of S atomicity by 1. If S>=0 is still after decrement of 1, the process continues to execute; if S<0 is reduced by 1, it means that no resources are available, and you need to block yourself and put it on the waiting queue.
V primitive: means to release a resource and add 1 to S atomicity; if 1 is added after S>0, the process continues to execute; if 1 is added after S<=0, it means that there is a waiting process on the waiting queue, and the first waiting process needs to be awakened.
Hereoperating systemIt can be understood as GoRuntime，processIt can be understood asCoroutine。
Method explanation
Finally, let’s dive into the three WaitGroup methods and conduct source code analysis. If you are interested, you can continue reading, mainly the analysis and comments on the source code.
Add(delta int) method
func (wg *WaitGroup) Add(delta int) {
statep, semap := ()
if { // This is the competition detection of go, so you don't have to worry about it _ = *statep
if delta < 0 {
((wg))
}
()
defer ()
}
state := atomic.AddUint64(statep, uint64(delta)<<32)
v := int32(state >> 32) // Get counter w := uint32(state) // Get waiter if && delta > 0 && v == int32(delta) { // Go competition detection, you don't have to worry about it ((semap))
}
if v < 0 {
panic("sync: negative WaitGroup counter")
}
if w != 0 && delta > 0 && v == int32(delta) {
panic("sync: WaitGroup misuse: Add called concurrently with Wait")
}
if v > 0 || w == 0 { // counter > 0: There is still task being executed; waiter == 0 means that there is no goroutine blocking and waiting return
}
if *statep != state {
panic("sync: WaitGroup misuse: Add called concurrently with Wait")
}
// Execution here is equivalent to countr = 0, that is, all tasks have been executed and you need to call up the waiting goroutine *statep = 0
for ; w != 0; w-- {
runtime_Semrelease(semap, false, 0)
}
}Done Method
func (wg *WaitGroup) Done() {
(-1) // Directly call the Add method to counter -1}
Wait Method
func (wg *WaitGroup) Wait() {
statep, semap := ()
if { // Go competition detection, you don't have to worry about it _ = *statep
()
}
for {
state := atomic.LoadUint64(statep)
v := int32(state >> 32)
w := uint32(state)
if v == 0 {
// counter is 0, no need to wait any longer. if {
()
((wg))
}
return
}
// Number of waiters +1. if atomic.CompareAndSwapUint64(statep, state, state+1) {
if && w == 0 {
((semap)) // Go competition detection, you don't have to worry about it }
runtime_Semacquire(semap) // Blocking waiting to be called if *statep != 0 {
panic("sync: WaitGroup is reused before previous Wait has returned")
}
if {
()
((wg))
}
return
}
}
}From the source code of these methods, we can see that Go does not use mutex and other locks to modify the field value, but uses atomic atomic operation to modify it. This is supported on the underlying hardware, so the performance is better.
Summarize
WaitGroup is relatively simple, which is the maintenance of some count values and the blocking and evocation of goroutines. It is also simple to use, and the three methods Add, Done, and Wait often appear at the same time. I believe that everyone can see a rough idea when you go deep into the source code. For more information about golang waitgroup concurrency control, please follow my other related articles!