golang 源码学习之timer/ticker

源码目录 time/time.go (1.1.4.1)

数据结构

/// time/sleep.go
type Timer struct {
 C <-chan Time // 时间到达时的通道
 r runtimeTimer
}
/// time/time.go
type Ticker struct {
 C <-chan Time 
 r runtimeTimer
}
/// time/sleep.go
type runtimeTimer struct {
 pp uintptr
 when int64 // 到期时间
 period int64 // ticker的周期。timer只会触发一次,所有为空
 f func(interface{}, uintptr) // 时间到达时触发的方法
 arg interface{} 
 seq uintptr
 nextwhen int64
 status uint32
}
/// runtime/runtime2.go
type p struct {
 .....
 timersLock mutex
 timers []*timer // 每个P都有个timer集合。
 
 ....
}

从数据结构上看,timer和ticker其实是一样的。每个P都维护一个timer的最小堆

创建

func NewTicker(d Duration) *Ticker {
 if d <= 0 { // 时间必须大于0
 panic(errors.New("non-positive interval for NewTicker"))
 }
 c := make(chan Time, 1) // 通道容量为1,那ticker是否会堵塞呢?
 t := &Ticker{
 C: c,
 r: runtimeTimer{
 when: when(d), // 到期时间
 period: int64(d), // 周期
 f: sendTime, // 时间到达时往通道发送消息
 arg: c,
 },
 }
 startTimer(&t.r)
 return t
}
// 时间到达时往通道发送消息
func sendTime(c interface{}, seq uintptr) {
 select {
 case c.(chan Time) <- Now():
 default: // default 说明通道不会堵塞
 }
}
// runtimeTimer的参数比tiker少了个period。因为timer只触发一次
func NewTimer(d Duration) *Timer {
 c := make(chan Time, 1)
 t := &Timer{
 C: c,
 r: runtimeTimer{
 when: when(d),
 f: sendTime,
 arg: c,
 },
 }
 startTimer(&t.r)
 return t
}

timer的创建和ticker的创建基本一致,初始化通道和runtimeTimer。

添加timer到P

/// runtime/time.go
func startTimer(t *timer) {
 addtimer(t)
}
func addtimer(t *timer) {
 ....
 
 t.status = timerWaiting
 addInitializedTimer(t)
}
func addInitializedTimer(t *timer) {
 when := t.when
 pp := getg().m.p.ptr() // 获取当前P
 lock(&pp.timersLock)
 cleantimers(pp) // 清理P中无用的timer
 doaddtimer(pp, t) // 添加timer到p
 unlock(&pp.timersLock)
 wakeNetPoller(when)
}
func doaddtimer(pp *p, t *timer) {
 ...
 
 t.pp.set(pp) // timer 和 p关联
 i := len(pp.timers)
 pp.timers = append(pp.timers, t) // timer加入P的timer集合
 siftupTimer(pp.timers, i) // 调整最小堆,最先到达的时间在最前面
 if t == pp.timers[0] { // 如果当前timer就是最小堆的一个,更新最先的到达时间
 atomic.Store64(&pp.timer0When, uint64(t.when))
 }
 atomic.Xadd(&pp.numTimers, 1) // 更新timer数
}

将timer添加到当前P的最小堆中

调度

/// runtime/proc.go
func schedule() {
 ...
 
 pp := _g_.m.p.ptr() // 获取当前P
 
 ...
 
 checkTimers(pp, 0) 
}
func checkTimers(pp *p, now int64) (rnow, pollUntil int64, ran bool) {
 
 ...
 lock(&pp.timersLock)
 adjusttimers(pp)
 rnow = now
 if len(pp.timers) > 0 {
 if rnow == 0 {
 rnow = nanotime()
 }
 for len(pp.timers) > 0 { // timers集合大于0,循环
 if tw := runtimer(pp, rnow); tw != 0 { 
 if tw > 0 {
 pollUntil = tw
 }
 break // checkTimers结束
 }
 ran = true
 }
 }
 
 ...
 
 unlock(&pp.timersLock)
 return rnow, pollUntil, ran
}
func runtimer(pp *p, now int64) int64 {
 for {
 t := pp.timers[0] // 最小堆的第一个
 switch s := atomic.Load(&t.status); s {
 case timerWaiting:
 if t.when > now { // 时间还没到,返回到期时间,checkTimers的for循环也会终止
 return t.when
 }
 if !atomic.Cas(&t.status, s, timerRunning) {
 continue
 }
 runOneTimer(pp, t, now) // 时间到期
 return 0
 
 .....
 }
 }
}
func runOneTimer(pp *p, t *timer, now int64) {
 f := t.f // timer的回调方法,创建时候设置的sendTime
 arg := t.arg // 创建时候设置的通道
 seq := t.seq
 if t.period > 0 { // t.period > 0说明是ticker
 delta := t.when - now
 t.when += t.period * (1 + -delta/t.period) // 设置下一轮的到达时间
 siftdownTimer(pp.timers, 0) // 调整最小堆
 if !atomic.Cas(&t.status, timerRunning, timerWaiting) {
 badTimer()
 }
 updateTimer0When(pp) // 更新P的到达时间
 } else { // 如果不是ticker 则移除出最小堆
 dodeltimer0(pp)
 if !atomic.Cas(&t.status, timerRunning, timerNoStatus) {
 badTimer()
 }
 }
 unlock(&pp.timersLock)
 f(arg, seq) // 执行sendTime
 lock(&pp.timersLock)
}

schedule() 是goroutine调度中的方法(参考我的另一篇博客), 获取当前p的timers最小堆中的一个timer,如果timer时间到期,则执行回调向通道发送消息,通知上层程序。如果是ticker则重新设置下次到达的时间,更新最小堆。

小结

timer和ticker的数据结构相同。每个P都有一个timer的最小堆。在runtime调度时,会获取当前P最小堆的一个timer,如果时间到期,则调用回调方法通过通道通知上层程序。如果timer是ticker则重新设置到期时间。

已离职,求工作

本文来自:简书

感谢作者:ihornet

查看原文:golang 源码学习之timer/ticker

• 编辑
• 预览