The basic concept

Easy concurrency is one of Golang’s defining strengths, and concurrency is no stranger to The WaitGroup of the Sync package. The WaitGroup is used to wait for Golang concurrent instances (Goroutine). When using GO to start multiple concurrent programs, WaitGroup can wait for all go programs to finish before executing the following code logic, such as:

func Main() {
    wg := sync.WaitGroup{}
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {defer wg.done () time.sleep (10 * time.second)}()} wg.wait () {defer wg.done ()}Copy the code

Realize the principle of

WaitGroup provides three methods: Add(int),Done(), and Wait(). Done() calls Add(-1).

WaitGroup mainly maintains two counters, a request counter V and a wait counter W, which form a 64-bit value with a high request counter 32bit and a low wait counter 32bit.

Each time Add executes, the request counter v increments by 1, the Done method executes, the request counter decreases by 1, and the semaphore wakes Wait() with v equal to 0.

So why wait for the counter? This is because the Wait() method of the same instance supports multiple calls. Each time a Wait() is executed, the Wait counter W is incremented by 1. When a request counter V is 0 to trigger a Wait(), w semaphores are sent to trigger all Wait() correctly. But in some cases where it is useful (such as when multiple concurrent WaitGroup instances depend on the end signal for the next action), the demo code looks like this:

func main() {
  wg := sync.WaitGroup{}
  for i := 0; i < 10; i++ {
    wg.Add(1)
    go func() {
      defer wg.Done()
​    }()
  }
  time.Sleep(2 * time.Second)
  forj := 0; j < 3; J++ {go func(I int) {// call Wait(), wg.wait () fmt.println () wg.wait () fmt.println ("wait done now ", i)
    }(j)
  }
  time.Sleep(10 * time.Second)
  return} /* The output is as follows, with the numbers appearing in a random orderwait done now  1
wait done now  0
wait done now  2
*/
Copy the code

There are also strict checks on the use logic in WaitGroup, such as not adding () once Wait() starts.

Here is the annotated code, with the trace part removed that does not affect the code logic:

Func (wg *WaitGroup) Add(delta int) {statep := wg.state(waitState := atomy.adDUINT64 (STATEP, uint64(delta)<<32) V := INT32 (state >> 32) W := uint32(state)if v < 0 {
        panic("sync: negative WaitGroup counter")}ifw ! = 0 && delta > 0 && v == int32(delta) { //waitIf the value does not equal 0, Wait has been executed. Add Panic ("sync: WaitGroup misuse: Add called concurrently with Wait"} // Normally, Add will increase v, Done will decrease V, if not all Done, v will always be greater than 0, until v is 0, and w represents how many Goruntine are waitingdoneThe signal,waitIn compareAndSwap increments the w by 1if v > 0 || w == 0 {
        return
    }
    // This goroutine has set counter to 0 when waiters > 0.
    // Now there can't be concurrent mutations of state: // - Adds must not happen concurrently with Wait, // - Wait does not increment waiters if it sees counter == 0. // Still do a cheap sanity check to detect WaitGroup // The misuse of v is 0(" Done ") or w is not 0(" w "), but there is an Add in the process that causes the statep to change, panic if *statep! = state { panic("sync: WaitGroup misuse: Run the following command to restore the partitioned state: // Waiters count to 0. *statep = 0 // The semaphore is protected by the Waitgroup. w ! = 0; w-- { runtime_Semrelease(&wg.sema, false) } } // Done decrements the WaitGroup counter by one. func (wg *WaitGroup) Done() { wg.Add(-1) } // Wait blocks until the WaitGroup counter is zero. func (wg *WaitGroup) Wait() { statep := wg.state() for { state := atomic.LoadUint64(statep) v := int32(state >> 32) w := uint32(state) if v == 0 { // Counter is 0, no need to wait. if race.Enabled { race.Enable() race.Acquire(unsafe.Pointer(wg)) } return } // Increment waiters count. // If statep is equal to state, then the Wait count is increased and the semaphores are added. // If statep is equal to state, the semaphores are added and the semaphores are added. Statep was overwritten by another goroutine between the time it read and the time it compared CAS, so instead of going into if, go back and read it again, so that you don't have to lock it, To be more efficient if atomic.Com pareAndSwapUint64 (statep, state, state+1) { if race.Enabled && w == 0 { // Wait must be synchronized with the first Add. // Need to model this is as a write to race with the read in Add. // As a consequence, can do the write only for the first waiter, // An otherwise concurrent Waits will race with each other.race.write (unsafe.Pointer(&wg.sema))} // an otherwise concurrent Waits will race with each other.race.write (unsafe Runtime_Semacquire (&Wg.sema) // Add is Done if *statep! Add panic("sync: "); // Add panic("sync: "); WaitGroup is reused before previous Wait has returned") } return } } }Copy the code