• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    公众号

go学习笔记 sync/once源码 和 sync/WaitGroup源码

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

sync.Once.Do(f func())是一个挺有趣的东西,能保证once只执行一次,无论你是否更换once.Do(xx)这里的方法,这个sync.Once块只会执行一次。

package sync
 
import (
    "sync/atomic"
)
 
// Once is an object that will perform exactly one action.
type Once struct {
    // done indicates whether the action has been performed.
    // It is first in the struct because it is used in the hot path.
    // The hot path is inlined at every call site.
    // Placing done first allows more compact instructions on some architectures (amd64/x86),
    // and fewer instructions (to calculate offset) on other architectures.
    done uint32 // 初始值为0表示还未执行过,1表示已经执行过
    m    Mutex
}
 
// Do calls the function f if and only if Do is being called for the
// first time for this instance of Once. In other words, given
//     var once Once
// if once.Do(f) is called multiple times, only the first call will invoke f,
// even if f has a different value in each invocation. A new instance of
// Once is required for each function to execute.
//
// Do is intended for initialization that must be run exactly once. Since f
// is niladic, it may be necessary to use a function literal to capture the
// arguments to a function to be invoked by Do:
//     config.once.Do(func() { config.init(filename) })
//
// Because no call to Do returns until the one call to f returns, if f causes
// Do to be called, it will deadlock.
//
// If f panics, Do considers it to have returned; future calls of Do return
// without calling f.
//
func (o *Once) Do(f func()) {
    // Note: Here is an incorrect implementation of Do:
    //
    //    if atomic.CompareAndSwapUint32(&o.done, 0, 1) {
    //        f()
    //    }
    //
    // Do guarantees that when it returns, f has finished.
    // This implementation would not implement that guarantee:
    // given two simultaneous calls, the winner of the cas would
    // call f, and the second would return immediately, without
    // waiting for the first's call to f to complete.
    // This is why the slow path falls back to a mutex, and why
    // the atomic.StoreUint32 must be delayed until after f returns.
    // 每次一进来先读标识位 0 标识没有被执行过,1 标识已经被执行过
    if atomic.LoadUint32(&o.done) == 0 {
        // Outlined slow-path to allow inlining of the fast-path.
        o.doSlow(f)
    }
}
 
func (o *Once) doSlow(f func()) {
    o.m.Lock() // 施加互斥锁
    defer o.m.Unlock()
    if o.done == 0 {
        defer atomic.StoreUint32(&o.done, 1)
        f()
    }
}

从上面我们可以看出,once只有一个 Do 方法;once的结构体中只定义了两个字段:一个mutex的m,一个代表标识位的done。

下面我们来看看Do方法的流程:

WaitGroup用于等待一组线程的结束。父线程调用Add 方法来设定应等待的线程数量。每个被等待的线程在结束时应调用Done方法。同时,主线程里可以调用wait方法阻塞至所有线程结束。 注意:Add和创建协程的数量一定要匹配,否则会产出panic
主要函数:
func (wg *WaitGroup) Add(delta int):等待协程的数量。
func (wg *WaitGroup) Done(): 减少waitgroup线程等待线程数量的值,一般在协程完成之后执行。
func (wg *WaitGroup) Wait():wait方法一般在主线程调用,阻塞直到group计数减少为0。

 
package sync
 
import (
    "internal/race"
    "sync/atomic"
    "unsafe"
)
 
// A WaitGroup waits for a collection of goroutines to finish.
// The main goroutine calls Add to set the number of
// goroutines to wait for. Then each of the goroutines
// runs and calls Done when finished. At the same time,
// Wait can be used to block until all goroutines have finished.
//
// A WaitGroup must not be copied after first use.
type WaitGroup struct {
    noCopy 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 do not ensure it. So we allocate 12 bytes and then use
    // the aligned 8 bytes in them as state, and the other 4 as storage
    // for the sema.
    // 64 bit:高32 bit是计数器,低32位是 阻塞的goroutine计数。
    // 64位的原子操作需要64位的对齐,但是32位。
    // 编译器不能确保它,所以分配了12个byte对齐的8个byte作为状态。其他4个作为信号量
    state1 [3]uint32
}
 
// uintptr和unsafe.Pointer的区别就是:unsafe.Pointer只是单纯的通用指针类型,用于转换不同类型指针,它不可以参与指针运算;
// 而uintptr是用于指针运算的,GC 不把 uintptr 当指针,也就是说 uintptr 无法持有对象,uintptr类型的目标会被回收。
// state()函数可以获取到wg.state1数组中元素组成的二进制对应的十进制的值 和信号量
// 根据编译器位数,获得标志位和等待次数的数据域
// state returns pointers to the state and sema fields stored within wg.state1.
func (wg *WaitGroup) state() (statep *uint64, semap *uint32) {
    if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 {
       // 是否是 64位机器:因为64位机器站高8位 信号量在后面
        return (*uint64)(unsafe.Pointer(&wg.state1)), &wg.state1[2]
    } else {
        // 如果是 32位机器,型号量在最前面
        return (*uint64)(unsafe.Pointer(&wg.state1[1])), &wg.state1[0]
    }
}
 
// Add adds delta, which may be negative, to the WaitGroup counter.
// If the counter becomes zero, all goroutines blocked on Wait are released.
// If the counter goes negative, Add panics.
//
// Note that calls with a positive delta that occur when the counter is zero
// must happen before a Wait. Calls with a negative delta, or calls with a
// positive delta that start when the counter is greater than zero, may happen
// at any time.
// Typically this means the calls to Add should execute before the statement
// creating the goroutine or other event to be waited for.
// If a WaitGroup is reused to wait for several independent sets of events,
// new Add calls must happen after all previous Wait calls have returned.
// See the WaitGroup example.
func (wg *WaitGroup) Add(delta int) {
    // 获取到wg.state1数组中元素组成的二进制对应的十进制的值的指针 和信号量
    statep, semap := wg.state()
    if race.Enabled {
        _ = *statep // trigger nil deref early
        if delta < 0 {
            // Synchronize decrements with Wait.
            race.ReleaseMerge(unsafe.Pointer(wg))
        }
        race.Disable()
        defer race.Enable()
    }
    // 将标记为加delta 因为高32位是计数器 所以把 delta的值左移32位,并从数组的首元素处开始赋值
    state := atomic.AddUint64(statep, uint64(delta)<<32)
    v := int32(state >> 32) // 获取计数器的值:转int32
    //获得调用 wait()等待次数:转uint32
    w := uint32(state)
    if race.Enabled && delta > 0 && v == int32(delta) {
        // The first increment must be synchronized with Wait.
        // Need to model this as a read, because there can be
        // several concurrent wg.counter transitions from 0.
        race.Read(unsafe.Pointer(semap))
    }
    // 计数器为负数,报panic
    //标记位不能小于0(done过多或者Add()负值太多)
    if v < 0 {
        panic("sync: negative WaitGroup counter")
    }
    // 不能Add 与Wait 同时调用
    if w != 0 && delta > 0 && v == int32(delta) {
        panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }
    // Add 完毕
    if 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 misuse.
    // 当等待计数器> 0时,而goroutine将设置为0。
    // 此时不可能有同时发生的状态突变:
    // - Add()不能与 Wait() 同时发生,
    // - 如果计数器counter == 0,不再增加等待计数器
    // 不能Add 与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, 0)
    }
}
 
// Done decrements the WaitGroup counter by one.
// Done方法其实就是Add(-1)
func (wg *WaitGroup) Done() {
    wg.Add(-1)
}
 
// Wait blocks until the WaitGroup counter is zero.
// Wait 会一直阻塞到 计数器值为0为止
func (wg *WaitGroup) Wait() {
    statep, semap := wg.state()
    if race.Enabled {
        _ = *statep // trigger nil deref early
        race.Disable()
    }
    //循环检查计数器V啥时候等于0
    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.
        // 尚有未执行完的go程,等待标志位+1(直接在低位处理,无需移位)
        // 增加等待goroution计数,对低32位加1,不需要移位
        if atomic.CompareAndSwapUint64(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,
                // otherwise concurrent Waits will race with each other.
                race.Write(unsafe.Pointer(semap))
            }
            // 目的是作为一个简单的sleep原语,以供同步使用
            runtime_Semacquire(semap)
            // 在上一次Wait返回之前重新使用WaitGroup,即在之前的Done 中没有清空 计数量就会有问题
            if *statep != 0 {
                panic("sync: WaitGroup is reused before previous Wait has returned")
            }
            if race.Enabled {
                race.Enable()
                race.Acquire(unsafe.Pointer(wg))
            }
            return
        }
    }
}

Add:

Wait:

 


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Go语言初始化变量的几种方法发布时间:2022-07-10
下一篇:
gopanic发布时间:2022-07-10
热门推荐
热门话题
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap