sync包 waitgroup源码阅读

借鉴于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

添加或者减少等待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)

相当于Add(-1)。

func (wg *WaitGroup) Done() {
    // 计数器减一
    wg.Add(-1)
// 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