Go sync. WaitGroup

1 Preface

WaitGroup yes Golang Concurrency control technology often used in application development .

WaitGroup, Can be understood as Wait-Goroutine-Group, That is, wait for a group goroutine end . For example, a certain goroutine Need to wait for a few more goroutine Complete , So use WaitGroup Can be easily implemented .

Here are 1 individual goroutine Wait for another 2 individual goroutine End example ：

package main
 
import (
    "fmt"
    "sync"
)
 
func main() {
    
    var wg sync.WaitGroup  // Create a synchronization waiting group object 
 
    wg.Add(2)  // Set the counter , The value is goroutine The number of ,2 individual 
    go func() {
    
        //Do some work1
        fmt.Println("Goroutine 1 finished!")
        wg.Done()  //goroutine The counter is decremented after execution 1, Wait for the group object counter to be left 2-1=1
    }()
 
    go func() {
    
        //Do some work2
        fmt.Println("Goroutine 2 finished!")
        wg.Done()  //goroutine The counter is decremented after execution 1,, Wait for the group object counter to be left 1-1=0
    }()
 
    wg.Wait()  // Lord goroutine The block wait counter becomes 0
    fmt.Printf("All Goroutine finished!")
}

To put it simply , In the above procedure wg Internal maintenance of a counter ：

start-up goroutine front Pass the counter through Add(2) Set the counter to be started goroutine Number .
start-up goroutine after , Use Wait() Methods block yourself , Wait until the counter changes to 0.
Every goroutine At the end of execution, pass Done() Method to decrement the counter 1.
Counter changes to 0 after , Obstructed goroutine Awakened .

2 Basic knowledge of

2.1 Semaphore
The semaphore is Unix A mechanism provided by the system to protect shared resources , Used to prevent multiple threads from accessing a resource at the same time .

It can be simply understood that the semaphore is a numerical value ：

When the semaphore >0 when , Indicates that the resource is available , When acquiring the semaphore, the system automatically reduces the semaphore 1;
When the semaphore ==0 when , Indicates that the resource is temporarily unavailable , When getting the semaphore , The current thread goes to sleep , Wakes up when the semaphore is positive ;
because WaitGroup Semaphores are also used in the implementation , Here's a brief introduction .

3 WaitGroup Data structure and usage

Source package src/sync/waitgroup.go:WaitGroup Defines its data structure ：

type WaitGroup struct {
    
	state1 [3]uint32
}

state1 The length is 3 Array of , Which includes state And a semaphore , and state It's actually two counters ：

counter： Currently not executed goroutine Counter
waiter count: wait for goroutine-group The end of the goroutine Number , That is, how many people are waiting
semaphore: Semaphore
Consider whether the bytes are aligned , The three appear in different positions , For the sake of simplicity , By byte aligned , The locations of the three in memory are as follows ：

WaitGroup Three external interfaces are provided ：

• func (*WaitGroup) Add： take delta Value added to counter in
  Add() Did two things , One is to put the delta The value is added to counter in , because delta It can be negative , in other words counter It's possible to become 0 Or negative value , So the second thing is to be counter Value to 0 when , Follow up waiter The value releases the same amount of semaphore , Put the waiting goroutine All wake up , If counter It becomes negative , be panic.

Add() The pseudocode is as follows ：

func (wg *WaitGroup) Add(delta int) {
    
    statep, semap := wg.state() // obtain state and semaphore Address of a pointer 
    
    state := atomic.AddUint64(statep, uint64(delta)<<32) // hold delta Move left 32 Bit accumulation state, That is, add to counter in 
    v := int32(state >> 32) // obtain counter value 
    w := uint32(state)      // obtain waiter value 
    
    if v < 0 {
                  // After accumulation counter The value becomes negative ,panic
        panic("sync: negative WaitGroup counter")
    }
 
    // After accumulation , here ,counter >= 0
    // If counter Being positive , Indicates that the semaphore does not need to be released , immediate withdrawal 
    // If waiter zero , It means there are no waiting people , There is no need to release semaphores , immediate withdrawal 
    if v > 0 || w == 0 {
    
        return
    }
 
    // here ,counter It must be equal to 0, and waiter Must be greater than 0（ Internal maintenance waiter, There won't be less than 0 The situation of ）,
    // The first counter Set as 0, Re release waiter The number of semaphores 
    *statep = 0
    for ; w != 0; w-- {
    
        runtime_Semrelease(semap, false) // Release semaphore , Perform one release at a time , Wake up a waiting person 
    }
}

• func (wg *WaitGroup) Wait()：waiter Increasing 1, And block the waiting semaphore semaphore
  Wait() The method also does two things , One is accumulation waiter, The second is to block the waiting semaphore

func (wg *WaitGroup) Wait() {
    
    statep, semap := wg.state() // obtain state and semaphore Address of a pointer 
    for {
    
        state := atomic.LoadUint64(statep) // obtain state value 
        v := int32(state >> 32)            // obtain counter value 
        w := uint32(state)                 // obtain waiter value 
        if v == 0 {
                            // If counter The value is 0, Explain all goroutine All out of the , There is no need to wait , Go straight back to 
            return
        }
        
        //  Use CAS（ Comparison exchange algorithm ） Add up waiter, Accumulation may fail , Pass after failure for loop Try again next time 
        if atomic.CompareAndSwapUint64(statep, state, state+1) {
    
            runtime_Semacquire(semap) // After successful accumulation , Wait for the semaphore to wake you up 
            return
        }
    }
}

It's used here CAS The algorithm guarantees multiple goroutine At the same time Wait() Can also correctly accumulate waiter.

• func (wg *WaitGroup) Done()： counter Decline 1, according to waiter The value releases the semaphore for the corresponding number of times
  Done() Do one thing , Namely the counter reduce 1, We know Add() Negative values are acceptable , therefore Done It's actually just called Add(-1).

Source code is as follows ：

func (wg *WaitGroup) Done() {
    
	wg.Add(-1)
}

Done() The execution logic goes to Add(), In fact, it is the last one to complete goroutine Wake up the waiting person .

4 Summary

• Add() Operation must be earlier than Wait(), Otherwise panic
• Add() The set value must match the actual waiting value goroutine The number is the same , Otherwise panic
• it is to be noted that Add and Done Functions must be paired , Otherwise, deadlock may occur , The error message reported is as follows ： fatal error: all goroutines are asleep - deadlock!