Explanation of waitgroup usage in go language learning

Catalog

Preface

A profound

The overview

Underlying implementation

Structure

Add

Done

Wait

It's easy to get wrong

summary

Preface

In the previous article , We used it  WaitGroup  Arrange tasks ,Go In language  WaitGroup  and Java Medium  CyclicBarrier、CountDownLatch  Very similar . For example, we have a main task in execution , At a certain point, three subtasks need to be executed in parallel , And you need to wait until all three subtasks are completed , Continue to perform the main task . Then we need to set up a checkpoint , Keep the main task blocked here , Wait until the three subtasks are completed .

explain ： The example in this article , All based on Go1.17 64 Bit machine

A profound

Let's start with a simple example , look down  WaitGroup  How is it used . Use... In the example  Add(5)  We have 5 individual The subtasks , Then I got up 5 individual Cooperate to complete the task , Use of main coordination process  Wait()  Method wait The execution of the subprocess is completed , Output the total waiting time .

func main() { var waitGroup sync.WaitGroup start := time.Now() waitGroup.Add(5) for i := 0; i < 5; i++ { go func() { defer waitGroup.Done() time.Sleep(time.Second) fmt.Println("done") }() } waitGroup.Wait() fmt.Println(time.Now().Sub(start).Seconds())}/*donedonedonedonedone1.000306089*/ The overview

WaitGroup There are three ways ：

(wg *WaitGroup) Add(delta int)(wg *WaitGroup) Done()(wg *WaitGroup) Wait()

Add  Method for setting WaitGroup The count of , It can be understood as the number of subtasks

Done  Method is used to WaitGroup Subtract one from the count of , It can be understood as completing a subtask

Wait  Method is used to block the caller , until WaitGroup The count value of is 0, That is, all subtasks are completed

Normally , When we use it , The number of subtasks needs to be determined first , And then call Add() Method to pass in the corresponding quantity , In the coordination of each subtask , call Done(), The co process call that needs to wait Wait() Method , The status flow is shown in the following figure ：

Underlying implementation Structure type WaitGroup struct { noCopy noCopy // noCopy Field identification , because WaitGroup Can't copy , Convenient tool detection state1 [3]uint32 // 12 Bytes ,8 Byte identification Count value and waiting quantity ,4 Bytes are used to identify semaphores }

state1  Is a compound field , Will be split into two parts ： 64 position （8 Bytes ） Of  statep  As a whole for atomic operations , In front of it 4 Bytes represent the count value , The last four bytes indicate the number of waits ; The remaining  32 position （4 Bytes ）semap  Used to identify semaphores .

Go In language, for 64 Bit variables to perform atomic operations , You need to ensure that the variable is 64 Bit alignment Of , That is to ensure that 8 Bytes The first address of 8 Integer multiple . So when  state1  The first address of 8 The integral times of , take front 8 individual Bytes as  statep , after 4 individual Bytes as  semap; When  state1  Your first address is not 8 The integral times of , take after 8 individual Bytes as  statep , front 4 individual Bytes as  semap.

func (wg *WaitGroup) state() (statep *uint64, semap *uint32) { // The first address is 8 A multiple of the , front 8 Bytes to statep, The last four bytes are semap if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 { return (*uint64)(unsafe.Pointer(&wg.state1)), &wg.state1[2] } else { // after 8 Bytes to statep, The first four bytes are semap return (*uint64)(unsafe.Pointer(&wg.state1[1])), &wg.state1[0] }}Add

Add  Method is used to add a count value （ A negative number is equal to minus ）, When the count changes to 0 after , Wait  All waiters blocked by the method will be released

It is illegal for the count value to become negative , produce  panic

When the count is 0 when （ The initial state ）,Add  Methods cannot be compared with  Wait  Method is called concurrently , Need assurance  Add  Method in  Wait  Method Before call , Otherwise  panic

func (wg *WaitGroup) Add(delta int) { // Get the count value and wait for the variable statep and Semaphore semap statep, semap := wg.state() // Count value plus delta: statep The first four bytes of are count values , So it will delta Move forward 32 position state := atomic.AddUint64(statep, uint64(delta)<<32) // Count value v := int32(state >> 32) // The number of people waiting w := uint32(state) // If you add delta after , The count value becomes negative , illegal ,panic if v < 0 { panic("sync: negative WaitGroup counter") } // delta > 0 && v == int32(delta) : From 0 Start adding count values // w!=0 ： It means that there are already waiting people // Explain that when adding the count value , At the same time, a waiting person is added , Illegal operation . To add a waiting person, you need to add the count value if w != 0 && delta > 0 && v == int32(delta) { panic("sync: WaitGroup misuse: Add called concurrently with Wait") } // v>0 : The count value is not equal to 0, There is no need to wake up the waiters , Go straight back to // w==0: There are no waiters , There is no need to wake up , Go straight back to if v > 0 || w == 0 { return } // Check the data again for consistency if *statep != state { panic("sync: WaitGroup misuse: Add called concurrently with Wait") } // Here, the count value is 0, And the waiting person is greater than 0, It is necessary to wake up all the waiting people , And set the system to the initial state （0 state ） // Set both the count value and the number of waiters to 0 *statep = 0 // Wake up the waiting for ; w != 0; w-- { runtime_Semrelease(semap, false, 0) }}Done// Complete a task , Subtract one from the count , When the count value is reduced to 0 when , It is necessary to wake up all the waiting people func (wg *WaitGroup) Done() { wg.Add(-1)}Wait// call Wait Methods will be blocked , until Count value Turn into 0func (wg *WaitGroup) Wait() { // Get count 、 Waiting number and semaphore statep, semap := wg.state() for { state := atomic.LoadUint64(statep) // Count value v := int32(state >> 32) // The number of people waiting w := uint32(state) // The number of counts is 0, Go straight back to , No need to wait if v == 0 { return } // This indicates that the count value is greater than 0 // Increase the number of people waiting ： There will be competition , Like multiple Wait call , Or call at the same time Add Method , If the increase is unsuccessful, it will continue for loop if atomic.CompareAndSwapUint64(statep, state, state+1) { // After adding success , Block here in semaphore , Waiting to be awakened runtime_Semacquire(semap) // When awakened , Should be 0 state . If you reuse WaitGroup, Need to wait Wait return if *statep != 0 { panic("sync: WaitGroup is reused before previous Wait has returned") } return } }} It's easy to get wrong

From the above analysis of the source code, we can see that several will be generated  panic  The point of , This is what we use  WaitGroup  What to pay attention to

1. The count value becomes negative

call Add The parameter value is transferred to a negative number

func main() {var wg sync.WaitGroupwg.Add(1)wg.Add(-1)wg.Add(-1)}

Multiple calls Done Method

func main() { var wg sync.WaitGroup wg.Add(1) go func() { fmt.Println("test") wg.Done() wg.Done() }() time.Sleep(time.Second) wg.Wait()}

2.Add and Wait Concurrent invocations

Add  and  Wait  Concurrent invocations , It may not achieve the desired effect , even to the extent that  panic. In the following example , We want to wait 3 Execute the main task after all the sub tasks are executed , But the actual situation may be that the subtask hasn't been up yet , The main task continues to be executed .

func doSomething(wg *sync.WaitGroup) { wg.Add(1) fmt.Println("do something") defer wg.Done()}func main() { var wg sync.WaitGroup for i := 0; i < 3; i++ { go doSomething(&wg) } wg.Wait() fmt.Println("main")}//main//do something//do something

The right way to use , Should be calling  Wait  Call... First  Add

func doSomething(wg *sync.WaitGroup) { defer wg.Done() fmt.Println("do something")}func main() { var wg sync.WaitGroup wg.Add(3) for i := 0; i < 3; i++ { go doSomething(&wg) } wg.Wait() fmt.Println("main")}//do something//do something//do something//main

3. No wait Wait return , Just reuse it WaitGroup

func main() { var wg sync.WaitGroup wg.Add(1) go func() { fmt.Println("do something") wg.Done() wg.Add(1) }() wg.Wait()}

4. Copy use

We know Go Parameter passing in language , It's all value transfer , A copy operation will occur . So you're passing... To the function WaitGroup when , Operate with a pointer .

// Wrong usage , Pointer not used func doSomething(wg sync.WaitGroup) { fmt.Println("do something") defer wg.Done()}func main() { var wg sync.WaitGroup wg.Add(3) for i := 0; i < 3; i++ { // No pointers are used here ,wg The state never changes , Lead to Wait Keep blocking go doSomething(wg) } wg.Wait() fmt.Println("main")} summary

We use the source code + Example way , We learned together  sync.WaitGroup  Implementation logic , Some points for attention are also given , Just do the following , There's no problem ：

Guarantee Add stay Wait Pre invocation

Add Negative numbers are not passed in

Don't forget to call after the task is completed Done Method , It is recommended to use defer wg.Done()

Do not copy and use WaitGroup, Function is passed with a pointer

Try not to reuse WaigGroup, Reduce the risk of problems

This is about Go Language learning WaitGroup This is the end of the detailed usage article , More about Go Language WaitGroup Please search the previous articles of software development network or continue to browse the relevant articles below. I hope you will support software development network more in the future ！