难以驾驭的 Go timer，一文带你参透计时器的奥秘

难以驾驭的 Go timer，一文带你参透计时器的奥秘

什么是 timer

可以控制时间，确保应用程序中的某段代码在某个时刻运行。在 Go 语言中可以单次执行，也可以循环执行。

最常见的方式就是引用标准库 time 去做一些事情，普通开发者经常使用到的标准库代码是：

time.Now().Unix()

上述代码可用于获取当前时间的 Unix 时间戳，而在内部的具体实现上提供了Time、Timer 以及 Ticker 的各类配套方法。

timer 基本特性

Timer

演示代码：

func main() {
 timer := time.NewTimer(2 * time.Second)
 <-timer.C
 fmt.Println("我的脑子真的进煎鱼了！")
}

输出结果：

// 等待两秒...
我的脑子真的进煎鱼了！

我们可以通过 time.NewTimer 方法定时在 2 秒进行程序的执行。而其还有个变种的用法，在做 channel 的源码剖析时有发现

func main() {
 v := make(chan struct{})
 timer := time.AfterFunc(2*time.Second, func() {
  fmt.Println("我想在这个点吃煎鱼！")
  v <- struct{}{}
 })
 defer timer.Stop()
 <-v
}

在等待 2 秒后，会立即调用 time.AfterFunc 所对应的匿名方法。在时间上我们也可以指定对应的具体时间，达到异步的定时执行等诉求。

Ticker

演示代码：

func main() {
 ticker := time.NewTicker(time.Second)
 defer ticker.Stop()
 done := make(chan bool)
 go func() {
  time.Sleep(10 * time.Second)
  done <- true
 }()
 for {
  select {
  case <-done:
   fmt.Println("Done!")
   return
  case t := <-ticker.C:
   fmt.Println("炸煎鱼: ", t.Unix())
  }
 }
}

输出结果：

// 每隔一秒输出一次
炸煎鱼:  1611666168
炸煎鱼:  1611666169
炸煎鱼:  1611666170
炸煎鱼:  1611666171
...

我们通过 time.NewTicker 方法设定每 1 秒执行一次方法，因此在 for-select 中，我们会每 1 秒就可以自动 “炸一条煎鱼”，真是快乐极了。

而由于我们在 goroutine 中通过 sleep 方法的设定了 done 变量的输入，因此在 10 秒后就会结束炸煎鱼的循环输出，最终退出。

最小堆：四叉堆

在 Go 语言中，内置计时器的数据结构都会涉及到最小四叉堆，如下图所示：

整体来讲就是父节点一定比其子节点小，子节点之间没有任何关系和大小的要求。

数据结构

在 Go 语言中每个计时器运行时的基本单元是 runtime.timer：

type timer struct {
 pp puintptr
 when   int64
 period int64
 f      func(interface{}, uintptr)
 arg    interface{}
 seq    uintptr
 nextwhen int64
 status uint32
}

• pp：计时器所在的处理器 P 的指针地址。
• when：计时器被唤醒的时间。
• period：计时器再次被唤醒的时间（when+period）。
• f：回调函数，每次在计时器被唤醒时都会调用。
• arg：回调函数的参数，每次在计时器被唤醒时会将该参数项传入回调函数 f 中。
• seq：回调函数的参数，该参数仅在 netpoll 的应用场景下使用。
• nextwhen：当计时器状态为 timerModifiedXX 时，将会使用 nextwhen 的值设置到 where字段上。
• status：计时器的当前状态值，计时器本身包含大量的枚举标识，这块会在后面介绍。

但这类基本单元都不会是对用户端暴露的结构体，在对外上我们直观见的最多的是 time.NewTimer 所创建的 Timer 结构体：

type Timer struct {
 C <-chan Time
 r runtimeTimer
}

• C：用于接收 Timer 所触发的事件，当计时器的消息事件（例如：到期）发生时，该 channel 会接收到通知。
• r：与 runtime.timer 作用类似，内在属性保持一致。

同时在计时器运行模式上自 Go1.14 起发生了变更，runtime.timer 改为将每个 timer 均存储在对应的处理器 P 中

type p struct {
 ...
 timersLock mutex
 timers []*timer
 ...
}

在处理器 P 上，timers 字段就是一个以最小四叉堆形式存储的媒介。在时序上，需要立刻执行，或说需要越早执行的，就越排在堆的越上面：

实现原理

在了解了计时器的基本特性和数据结构后，我们进一步展开，一层层剖析其原理，看看其是何物。在 Go 语言中，计时器在运行时涉及十种状态处理，分别涉及增、删、改以及重置等操作。

计时器所包含的状态如下：

这时候可能就会有小伙伴疑惑，各种启动、删除、停止、启动是指代的是什么意思？为什么会涉及到 P 的管理？

创建计时器

接下来我们依然是从 NewTimer 和 NewTicker 方法开始入手：

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 对象，主要是 C 和 r 属性，含义与前面所表述的一致。
• 调用 startTimer 方法，启动计时器。

NewTicker 方法与 NewTimer 类似，主要是增加了 period 字段：

func NewTicker(d Duration) *Ticker {
 c := make(chan Time, 1)
 t := &Ticker{
  C: c,
  r: runtimeTimer{
   when:   when(d),
   period: int64(d),
   f:      sendTime,
   arg:    c,
  },
 }
 startTimer(&t.r)
 return t
}

在 Ticker 结构体中，period 字段用于表示计时器再次被唤醒的时间，可以便于做轮询触发。

启动计时器

在前面调用 NewTimer、NewTicker 方法时，会将新创建的新计时器 timer 加入到创建 timer 的 P 的最小堆中：

func addtimer(t *timer) {
 if t.when < 0 {
  t.when = maxWhen
 }
 if t.status != timerNoStatus {
  throw("addtimer called with initialized timer")
 }
 t.status = timerWaiting
 when := t.when
 pp := getg().m.p.ptr()
 lock(&pp.timersLock)
 cleantimers(pp)
 doaddtimer(pp, t)
 unlock(&pp.timersLock)
 wakeNetPoller(when)
}

• 检查是否满足基本条件：新增计时器的边界处理，timerNoStatus 状态判断排除。
• 调用 cleantimers 方法：清理处理器 P 中的计时器队列，可以加快创建和删除计时器的程序的速度。
• 调用 doaddtimer 方法：将当前所新创建的 timer 新增到当前处理器 P 的堆中。
• 调用 wakeNetPoller 方法：唤醒网络轮询器中休眠的线程，检查计时器被唤醒的时间（when）是否在当前轮询预期运行的时间（pollerPollUntil）内，若是唤醒。

停止计时器

在计时器的运转中，一般会调用 timer.Stop() 方法来停止/终止/删除计时器。虽然说法多样。但大家的真实目的是一样的，就是让这个 timer 从轮询器中消失，也就是从处理器 P 的堆中移除 timer：

func deltimer(t *timer) bool {
 for {
  switch s := atomic.Load(&t.status); s {
  case timerWaiting, timerModifiedLater:
   // timerWaiting/timerModifiedLater -> timerDeleted
   ...
  case timerModifiedEarlier:
      // timerModifiedEarlier -> timerModifying -> timerDeleted
   ...
  case timerDeleted, timerRemoving, timerRemoved:
      // timerDeleted/timerRemoving/timerRemoved 
   return false
  case timerRunning, timerMoving:
      // timerRunning/timerMoving
   osyield()
  case timerNoStatus:
   return false
  case timerModifying:
   osyield()
  default:
   badTimer()
  }
 }
}

但移除也不是直接一个 delete 就完事的，其在真正的删除方法 deltimer 中遵循了基本的规则处理：

1. timerWaiting/timerModifiedLater -> timerDeleted。
2. timerModifiedEarlier -> timerModifying -> timerDeleted。
3. timerDeleted/timerRemoving/timerRemoved -> 无需变更，已经满足条件。
4. timerRunning/timerMoving/timerModifying -> 正在执行、移动中，无法停止，等待下一次状态检查再处理。
5. timerNoStatus -> 无法停止，不满足条件。

上述五个基本流转逻辑就覆盖了 runtimer.deltimer 方法了，若有进一步需求的可通过传送门详细阅读。

修改/重置计时器

在应用程序的调度中，有时候因为逻辑产生了变更，我们需要重置计时器。这时候一般会调用timer.Reset() 方法来重新设置 Duration 值。

其表面对应的是 resetTimer 方法，但实际与修改计时器的 modtimer 方法是共用的：

func resettimer(t *timer, when int64) bool {
 return modtimer(t, when, t.period, t.f, t.arg, t.seq)
}

因此在这节中我们可以将重置和修改计时器放在一起分析。修改计时器，本质上是需要变更现有计时器，而在 Go 语言的计时器中是需要遵循基本规则，因此 modtimer 遵循下述规则处理：

1. timerWaiting -> timerModifying -> timerModifiedXX
2. timerModifiedXX -> timerModifying -> timerModifiedYY
3. timerNoStatus -> timerModifying -> timerWaiting
4. timerRemoved -> timerModifying -> timerWaiting
5. timerDeleted -> timerModifying -> timerModifiedXX
6. timerRunning -> 等待状态改变，才可以进行下一步
7. timerMoving -> 等待状态改变，才可以进行下一步
8. timerRemoving -> 等待状态改变，才可以进行下一步
9. timerModifying -> 等待状态改变，才可以进行下一步

func modtimer(t *timer, when, period int64, f func(interface{}, uintptr), arg interface{}, seq uintptr) bool {
 ...
 if wasRemoved {
  t.when = when
  pp := getg().m.p.ptr()
  lock(&pp.timersLock)
  doaddtimer(pp, t)
  unlock(&pp.timersLock)
  releasem(mp)
  wakeNetPoller(when)
 } else {
  t.nextwhen = when
  newStatus := uint32(timerModifiedLater)
  if when < t.when {
   newStatus = timerModifiedEarlier
  }
  ...
  releasem(mp)
  if newStatus == timerModifiedEarlier {
   wakeNetPoller(when)
  }
 }
 return pending
}

在完成了计时器的状态处理后，会分为两种情况处理：

• 待修改的计时器已经被删除：由于既有的计时器已经没有了，因此会调用 doaddtimer 方法创建一个新的计时器，并将原本的 timer 属性赋值过去，再调用 wakeNetPoller 方法在预定时间唤醒网络轮询。
• 正常逻辑处理：如果修改后的计时器的触发时间小于原本的触发时间，则修改该计时器的状态为 timerModifiedEarlier，并且调用 wakeNetPoller 方法在预定时间唤醒网络轮询。

触发计时器

在前面有提到 Go1.14 后，Go Timer 都已经归属到各个处理器 P 中去了，因此计时器的触发分为了两个部分：

• 通过调度器在调度时进行计时器的触发。
• 通过系统监控检查并触发计时器（到期未执行）。

调度器触发

调度器的触发一共分两种情况，一种是在调度循环的时候调用 checkTimers 方法进行计时器的触发：

func schedule() {
 _g_ := getg()
top:
 pp := _g_.m.p.ptr()
 pp.preempt = false
 // 处理调度时的计时器触发
 checkTimers(pp, 0)
 ...
 execute(gp, inheritTime)
}

另外一种是当前处理器 P 没有可执行的 Timer，且没有可执行的 G。那么按照调度模型，就会去窃取其他计时器和 G：

func findrunnable() (gp *g, inheritTime bool) {
 _g_ := getg()
top:
 _p_ := _g_.m.p.ptr()
 ...
 now, pollUntil, _ := checkTimers(_p_, 0)
 ...
}

调度系统在计时器处不深究，我们进一步剖析具体触发计时器的 checkTimers 方法：

func checkTimers(pp *p, now int64) (rnow, pollUntil int64, ran bool) {
 if atomic.Load(&pp.adjustTimers) == 0 {
  next := int64(atomic.Load64(&pp.timer0When))
  if next == 0 {
   return now, 0, false
  }
  if now == 0 {
   now = nanotime()
  }
  if now < next {
   if pp != getg().m.p.ptr() || int(atomic.Load(&pp.deletedTimers)) <= int(atomic.Load(&pp.numTimers)/4) {
    return now, next, false
   }
  }
 }
 lock(&pp.timersLock)
 adjusttimers(pp)
    ...
}

• 起始先通过 pp.adjustTimers 检查当前处理器 P 中是否有需要处理的计时器。

• • 若无需执行的计时器，则直接返回。
  • 若有，则判断下一个计时器待删除的计时器和处理器 P 上的计时器数量，若前者小于后者 1/4 则直接返回。
• 确定需要处理计时器后，通过调用 adjusttimers 方法重新根据时间将 timers 切片中 timer 的先后顺序重新排列（相当于 resort）。

func checkTimers(pp *p, now int64) (rnow, pollUntil int64, ran bool) {
 ...
 rnow = now
 if len(pp.timers) > 0 {
  if rnow == 0 {
   rnow = nanotime()
  }
  for len(pp.timers) > 0 {
   if tw := runtimer(pp, rnow); tw != 0 {
    if tw > 0 {
     pollUntil = tw
    }
    break
   }
   ran = true
  }
 }
 ...
}

在前面调整了 timers 切片中的最小堆的排序后，将会调用 runtimer 方法去真正运行所需要执行的 timer，完成触计时器的发。

func checkTimers(pp *p, now int64) (rnow, pollUntil int64, ran bool) {
 ...
 if pp == getg().m.p.ptr() && int(atomic.Load(&pp.deletedTimers)) > len(pp.timers)/4 {
  clearDeletedTimers(pp)
 }
 unlock(&pp.timersLock)
 return rnow, pollUntil, ran
}

在最后扫尾阶段，如果当前 G 的处理器与调用 checkTimers 方法所传入的处理器一致，并且处理器中 timerDeleted 状态的计时器数量是处理器 P 堆中的计时器的 1/4 以上，则调用 clearDeletedTimers 方法对已为删除状态的的计时器进行清理。

系统监控触发

即使是通过每次调度器调度和窃取的时候触发，但毕竟是具有一定的随机和不确定性。

因此系统监控触发依然是一个兜底保障，在 Go 语言中 runtime.sysmon 方法承担了这一个责任，存在触发计时器的逻辑：

func sysmon() {
 ...
 for {
  ...
  next, _ := timeSleepUntil()
  if debug.schedtrace <= 0 && (sched.gcwaiting != 0 || atomic.Load(&sched.npidle) == uint32(gomaxprocs)) {
   lock(&sched.lock)
   if atomic.Load(&sched.gcwaiting) != 0 || atomic.Load(&sched.npidle) == uint32(gomaxprocs) {
    if next > now {
     ...
     next, _ = timeSleepUntil()
     lock(&sched.lock)
     atomic.Store(&sched.sysmonwait, 0)
     noteclear(&sched.sysmonnote)
    }
    idle = 0
    delay = 20
   }
   unlock(&sched.lock)
  }
  ...
 }
}

在每次进行系统监控时，都会在流程上调用 timeSleepUntil 方法去获取下一个计时器应触发的时间，以及保存该计时器已打开的计时器堆的 P。

在获取完毕后会马上检查当前是否存在 GC，若是正在 STW 则获取调度互斥锁。若发现下一个计时器的触发时间已经过去，则重新调用 timeSleepUntil 获取下一个计时器的时间和相应 P 的地址。

func sysmon() {
 ...
 for {
  ...
  lock(&sched.sysmonlock)
  {
   now1 := nanotime()
   if now1-now > 50*1000 /* 50µs */ {
    next, _ = timeSleepUntil()
   }
   now = now1
  }
  ...
 }
}

检查 sched.sysmonlock 所花费的时间是否超过 50µs。若是，则有可能前面所获取的下一个计时器触发时间已过期，因此重新调用 timeSleepUntil 方法再次获取。

func sysmon() {
 ...
 for {
  ...
  lastpoll := int64(atomic.Load64(&sched.lastpoll))
  if netpollinited() && lastpoll != 0 && lastpoll+10*1000*1000 < now {
   atomic.Cas64(&sched.lastpoll, uint64(lastpoll), uint64(now))
   list := netpoll(0) // non-blocking - returns list of goroutines
   if !list.empty() {
    incidlelocked(-1)
    injectglist(&list)
    incidlelocked(1)
   }
  }
  if next < now {
   startm(nil, false)
  }
 }
}

如果发现超过 10ms 的时间没有进行 netpoll 网络轮询，则主动调用 netpoll 方法触发轮询。

同时如果存在不可抢占的处理器 P，则调用 startm 方法来运行那些应该运行，但没有在运行的计时器。

运行计时器

runtimer 方法主要承担计时器的具体运行，同时也会针对计时器的不同状态（含删除、修改、等待等）都进行了对应的处理，也相当于是个大的集中处理中枢了。例如在timerDeleted 状态下的计时器将会进行删除。

其遵循下述规则处理

1. timerNoStatus -> 恐慌：计时器未初始化
2. timerWaiting -> timerWaiting
3. timerWaiting -> timerRunning -> timerNoStatus
4. timerWaiting -> timerRunning -> timerWaiting
5. timerModifying -> 等待状态改变，才可以进行下一步
6. timerModifiedXX -> timerMoving -> timerWaiting
7. timerDeleted -> timerRemoving -> timerRemoved
8. timerRunning -> 恐慌：并发调用
9. timerRemoved -> 恐慌：计时器堆不一致
10. timerRemoving -> 恐慌：计时器堆不一致
11. timerMoving -> 恐慌：计时器堆不一致

我们再根据时间状态机，去针对性的看看源码是如何实现的：

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 {
    return t.when
   }
   runOneTimer(pp, t, now)
   return 0
  case timerDeleted:
      ...
  case timerModifiedEarlier, timerModifiedLater:
      ...
  case timerModifying:
   osyield()
  case timerNoStatus, timerRemoved:
   badTimer()
  case timerRunning, timerRemoving, timerMoving:
   badTimer()
  default:
   badTimer()
  }
 }
}

我们主要关注运行计时器，也就是 timerWaiting 状态下的处理，其首先会对触发时间（when）进行判定，若大于当前时间则直接返回（因为所需触发的时间未到）。否则将会调用 runOneTimer 方法去执行本次触发：

func runOneTimer(pp *p, t *timer, now int64) {
 f := t.f
 arg := t.arg
 seq := t.seq
 if t.period > 0 {
  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)
 } else {
  dodeltimer0(pp)
 }
 unlock(&pp.timersLock)
 f(arg, seq)
 lock(&pp.timersLock)
}

• 如果 period 大于 0，说明当前是 ticker，需要再次触发，因此还需要调整计时器的状态。

• • 重新计算下一次的触发时间，并且更新其在最小堆的位置。
  • 调用 atomic.Cas 方法该计时器的状态从 timerRunning 原子修改为 timerWaiting 状态。
  • 调用 updateTimer0When 方法设置处理器 P 的 timer0When 字段。
• 如果 period 等于 0，说明当前是 timer，只需要单次触发就可以了。

在完成计时器的运行属性更新后，上互斥锁，调用计时器的回调方法 f，完成本次完整的触发流程。

总结

Go 语言的 Timer 其实已经改过了好几版，在 Go1.14 的正式大改版后。目前来看已经初步的到了一个新的阶段。其设计的模式主要围绕三块：

• 在各个处理器 P 中，Timer 以最小四叉堆的存储方式在 timers 中。
• 在调度器的每轮调度中都会对计时器进行触发和检查。
• 在系统监听上 netpoll 会定时进行计时器的触发和检查。
• 在计时器的处理中，十个状态的流转和对应处理非常重要。