GolangWaitGroup实现原理解析

type WaitGroup struct {
   noCopy noCopy
   // 64-bit value: high 32 bits are counter, low 32 bits are waiter count.
   // 64-bit atomic operations require 64-bit alignment, but 32-bit
   // compilers only guarantee that 64-bit fields are 32-bit aligned.
   // For this reason on 32 bit architectures we need to check in state()
   // if state1 is aligned or not, and dynamically "swap" the field order if
   // needed.
   state1 uint64
   state2 uint32
}

func (wg *WaitGroup) state() (statep *uint64, semap *uint32) {
	if unsafe.Alignof(wg.state1) == 8 || uintptr(unsafe.Pointer(&wg.state1))%8 == 0 {
		// state1 is 64-bit aligned: nothing to do.
		return &wg.state1, &wg.state2
	} else {
		// state1 is 32-bit aligned but not 64-bit aligned: this means that
		// (&state1)+4 is 64-bit aligned.
		state := (*[3]uint32)(unsafe.Pointer(&wg.state1))
		return (*uint64)(unsafe.Pointer(&state[1])), &state[0]
	}
}

func (wg *WaitGroup) Add(delta int) {
    // wg.state()返回的是地址
	statep, semap := wg.state()
    // 原子操作，修改statep高32位的值，即counter的值
	state := atomic.AddUint64(statep, uint64(delta)<<32)
    // 右移32位，使高32位变成了低32，得到counter的值
	v := int32(state >> 32)
    // 直接取低32位，得到waiter的值
	w := uint32(state)
	// 不规范的操作
	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 {
		return
	}
    // 剩下的就是 counter == 0 且 waiter != 0 的情况
    // 在这个情况下，*statep 的值就是 waiter 的值，否则就有问题
    // 在这个情况下，所有的任务都已经完成，可以将 *statep 整个置0
    // 同时向所有的Waiter释放信号量
	// 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 misuse.
	if *statep != state {
		panic("sync: WaitGroup misuse: Add called concurrently with Wait")
	}
	// Reset waiters count to 0.
	*statep = 0
	for ; w != 0; w-- {
		runtime_Semrelease(semap, false, 0)
	}
}

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

func (wg *WaitGroup) Wait() {
    // wg.state()返回的是地址
	statep, semap := wg.state()
    // for循环是配合CAS操作
	for {
		state := atomic.LoadUint64(statep)
		v := int32(state >> 32) // counter
		w := uint32(state) // waiter
        // 如果counter为0，说明所有的任务在调用Wait的时候就已经完成了，直接退出
        // 这就要求，必须在同步的情况下调用Add()，否则Wait可能先退出了
		if v == 0 {
			return
		}
		// waiter++，原子操作
		if atomic.CompareAndSwapUint64(statep, state, state+1) {
            // 如果自增成功，则获取信号量，此处信号量起到了同步的作用
			runtime_Semacquire(semap)
			return
		}
	}
}