借鉴于Go夜读，加了个人理解：https://reading.developerlearning.cn/articles/sync/sync_waitgroup_source_code_analysis/
go版本：go1.12 windows/amd64
结构体
// A WaitGroup must not be copied after first use.
type WaitGroup struct {
noCopy noCopy // 标记不可复制，使用go vet作为检测使用，并因此只能进行指针传递，从而保证全局唯一
// 64位值：高32位是计数器，低32位是goroutine等待计数
// 不使用64位值是因为32位的编译器不能确保64位原子操作的位对齐
// 一个uit32占4个byte，我们分配了12个byte对齐的8个byte作为状态，剩下4个用于信号量的存储
state1 [3]uint32 // 前8个元素代表状态。转为2进制：0000 0000，0000 0000... ...0000 0000 后4个元素代表信号量
}
函数
以下代码已经去掉了与核心代码无关的race代码。
state
// state returns pointers to the state and sema fields stored within wg.state1.
// 根据结构体中初始化分配的 12bytes 来兼容处理 64位操作系统和 32位操作系统,
// 具体原理是，12bytes 中必定含有一个8bytes，仅仅使用这个含有的8bytes做为数据对齐使用，具体：
// 当指针位置刚好指在 (8byte) 的位置，证明位对齐，使用 前8bytes 作为状态计数；
// 当指针位置不指在 (8byte) ，没对齐，抛弃前 4bytes，使用后8bytes作为位对齐，用于记录状态计数。
func (wg *WaitGroup) state() (statep *uint64, semap *uint32) {
if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 { //（Why?）
// 前8byte为状态
return (*uint64)(unsafe.Pointer(&wg.state1)), &wg.state1[2]
} else {
// 后8byte为状态
return (*uint64)(unsafe.Pointer(&wg.state1[1])), &wg.state1[0]
}
}
有关内存对齐：https://www.e-learn.cn/content/qita/1020837
Add
添加或者减少等待goroutine的数量。
参数delta可能是负的，加到WaitGroup计数器,可能出现如下结果 - 如果计数器变为零，所有被阻塞的goroutines都会被释放。 - 如果计数器变成负数，就panic。
func (wg *WaitGroup) Add(delta int) {
// 获得wg.state1数组中元素组成的二进制对应的十进制的值
statep, semap := wg.state()
// 高32位是计数器,低32位是goroutine等待计数
// 原子操作，如初始状态 statep 为空，且 delta 等于 1, 操作 加 1：
// 00000000 00000000 00000000 00000001 00000000 …… 00000000
// \___________ 前32位 _______________/\__ 后32位均为0 __/
// 若当前状态位存在值 1，则再添加 delta 等于 1， 其结果为：
// 00000000 00000000 00000000 00000010 00000000 …… 00000000
// \___________ 前32位 _______________/\__ 后32位均为0 __/
state := atomic.AddUint64(statep, uint64(delta)<<32)
// 获取计数器
v := int32(state >> 32)
w := uint32(state)
// 计数器为负数，报panic
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 { // w == 0 代表无等待的goroutine
return
}
// Done()函数调用，v==0，唤醒线程
// 不能出现等待计数器>0，但goroutine已经设置为0的情况:
// 1.Wait()跟Add()同时发生；
// 2.如果计数器已经为0了，Wait()不能再增加等待计数器了
if *statep != state {
panic("sync: WaitGroup misuse: Add called concurrently with Wait")
}
// Reset waiters count to 0.
*statep = 0
for ; w != 0; w-- {
// 目的是作为一个简单的wakeup原语，以供同步使用。true为唤醒排在等待队列的第一个goroutine
runtime_Semrelease(semap, false)
}
}
Done
相当于Add(-1)。
func (wg *WaitGroup) Done() {
// 计数器减一
wg.Add(-1)
}
Wait
// Wait blocks until the WaitGroup counter is zero.
func (wg *WaitGroup) Wait() {
statep, semap := wg.state()
// cas算法
for {
state := atomic.LoadUint64(statep)
// 高32位是计数器
v := int32(state >> 32)
w := uint32(state)
// 计数器为0，结束等待
if v == 0 {
// Counter is 0, no need to wait.
if race.Enabled {
race.Enable()
race.Acquire(unsafe.Pointer(wg))
}
return
}
// 增加等待goroutine计数，对低32位加1，不需要移位
if atomic.CompareAndSwapUint64(statep, state, state+1) {
// 目的是作为一个简单的sleep原语，以供同步使用
runtime_Semacquire(semap)
if *statep != 0 {
panic("sync: WaitGroup is reused before previous Wait has returned")
}
return
}
}
}
使用注意事项
- WaitGroup不能保证多个 goroutine 执行次序
- WaitGroup无法指定固定的goroutine数目
