Go中定时器实现原理及源码解析

转载请声明出处哦~，本篇文章发布于luozhiyun的博客：https://www.luozhiyun.com

本文使用的go的源码15.7，需要注意的是由于timer是1.14版本进行改版，但是1.14和1.15版本的timer并无很大区别

我在春节期间写了一篇文章有关时间轮的：https://www.luozhiyun.com/archives/444。后来有同学建议我去看看 1.14版本之后的 timer 优化。然后我就自己就时间轮和 timer 也做了个 benchmark：

goos: darwin

goarch: amd64

pkg: gin-test/api/main

BenchmarkTimingWheel_StartStop/N-1m-12   4582120               254 ns/op              85 B/op          2 allocs/op

BenchmarkTimingWheel_StartStop/N-5m-12   3356630               427 ns/op              46 B/op          1 allocs/op

BenchmarkTimingWheel_StartStop/N-10m-12                  2474842               483 ns/op              60 B/op          1 allocs/op

BenchmarkStandardTimer_StartStop/N-1m-12                 6777975               179 ns/op              84 B/op          1 allocs/op

BenchmarkStandardTimer_StartStop/N-5m-12                 6431217               231 ns/op              85 B/op          1 allocs/op

BenchmarkStandardTimer_StartStop/N-10m-12                5374492               266 ns/op              83 B/op          1 allocs/op

PASS

ok      gin-test/api/main       60.414s

从上面可以直接看出，在添加了一千万个定时器后，时间轮的单次调用时间有明显的上涨，但是 timer 却依然很稳。

从官方的一个数据显示：

name                      old time/op  new time/op  delta

AfterFunc-12              1.57ms ± 1%  0.07ms ± 1%  -95.42%  (p=0.000 n=10+8)

After-12                  1.63ms ± 3%  0.11ms ± 1%  -93.54%  (p=0.000 n=9+10)

Stop-12                   78.3µs ± 3%  73.6µs ± 3%   -6.01%  (p=0.000 n=9+10)

SimultaneousAfterFunc-12   138µs ± 1%   111µs ± 1%  -19.57%  (p=0.000 n=10+9)

StartStop-12              28.7µs ± 1%  31.5µs ± 5%   +9.64%  (p=0.000 n=10+7)

Reset-12                  6.78µs ± 1%  4.24µs ± 7%  -37.45%  (p=0.000 n=9+10)

Sleep-12                   183µs ± 1%   125µs ± 1%  -31.67%  (p=0.000 n=10+9)

Ticker-12                 5.40ms ± 2%  0.03ms ± 1%  -99.43%  (p=0.000 n=10+10)

...

在很多项测试中，性能确实得到了很大的增强。下面也就一起看看性能暴涨的原因。

介绍

1.13 版本的 timer

Go 在1.14版本之前是使用 64 个最小堆，运行时创建的所有计时器都会加入到最小堆中，每个处理器（P）创建的计时器会由对应的最小堆维护。

下面是1.13版本 runtime.time源码：

const timersLen = 64

var timers [timersLen]struct {

	timersBucket

	// padding， 防止false sharing

	pad [sys.CacheLineSize - unsafe.Sizeof(timersBucket{})%sys.CacheLineSize]byte

}

// 获取 P 对应的 Bucket

func (t *timer) assignBucket() *timersBucket {

	id := uint8(getg().m.p.ptr().id) % timersLen

	t.tb = &timers[id].timersBucket

	return t.tb

}

type timersBucket struct {

	lock         mutex

	gp           *g

	created      bool

	sleeping     bool

	rescheduling bool

	sleepUntil   int64

	waitnote     note

    // timer 列表

	t            []*timer

}

通过上面的 assignBucket 方法可以知道，如果当前机器上的处理器 P 的个数超过了 64，多个处理器上的计时器就可能存储在同一个桶 timersBucket 中。

每个桶负责管理一堆这样有序的 timer，同时每个桶都会有一个对应的 timerproc 异步任务来负责不断调度这些 timer。t

imerproc 会从 timersBucket 不断取堆顶元素，如果堆顶的 timer 已到期则执行，没有任务到期则 sleep，所有任务都消耗完了，那么调用 gopark 挂起，直到有新的 timer 被添加到桶中时，才会被重新唤醒。

timerproc 在 sleep 的时候会调用 notetsleepg ，继而引发entersyscallblock调用，该方法会主动调用 handoffp ，解绑 M 和 P。当下一个定时时间到来时，又会进行 M 和 P 的绑定，处理器 P 和线程 M 之间频繁的上下文切换也是 timer 的首要性能影响因素之一。

1.14 版本后 timer 的变化

在Go 在1.14版本之后，移除了timersBucket，所有的计时器都以最小四叉堆的形式存储 P 中。

type p struct {

	...

	// 互斥锁

	timersLock mutex

	// 存储计时器的最小四叉堆

	timers []*timer

	// 计时器数量

	numTimers uint32

	// 处于 timerModifiedEarlier 状态的计时器数量

	adjustTimers uint32

	// 处于 timerDeleted 状态的计时器数量

	deletedTimers uint32

	...

}

timer 不再使用 timerproc 异步任务来调度，而是改用调度循环或系统监控调度的时候进行触发执行，减少了线程之间上下文切换带来的性能损失，并且通过使用 netpoll 阻塞唤醒机制可以让 timer 更加及时的得到执行。

timer的使用

time.Timer计时器必须通过time.NewTimer、time.AfterFunc或者 time.After 函数创建。

如下time.NewTimer：

通过定时器的字段C,我们可以及时得知定时器到期的这个事件来临，C是一个chan time.Time类型的缓冲通道，一旦触及到期时间，定时器就会向自己的C字段发送一个time.Time类型的元素值

func main() {

	//初始化定时器

	t := time.NewTimer(2 * time.Second)

	//当前时间

	now := time.Now()

	fmt.Printf("Now time : %v.\n", now)

	expire := <- t.C

	fmt.Printf("Expiration time: %v.\n", expire)

}

time.After一般是配合select来使用：

func main() {

	ch1 := make(chan int, 1)

	select {

	case e1 := <-ch1:

		//如果ch1通道成功读取数据，则执行该case处理语句

		fmt.Printf("1th case is selected. e1=%v",e1)

	case <- time.After(2 * time.Second):

		fmt.Println("Timed out")

	}

}

time.Afterfunc可以在设置时间过后执行一个函数：

func main() {

	f := func(){

		fmt.Printf("Expiration time : %v.\n", time.Now())

	}

	time.AfterFunc(1*time.Second, f)

 	time.Sleep(2 * time.Second)

}

分析

初始化&Timer结构体

我们先看看NewTimer方法是如何创建一个Timer的：

type Timer struct {

    C <-chan Time

    r runtimeTimer

}

func NewTimer(d Duration) *Timer {

	// 初始化一个channel，用于返回

	c := make(chan Time, 1)

	t := &Timer{

		C: c,

		r: runtimeTimer{

			when: when(d),

			f:    sendTime,

			arg:  c,

		},

	}

	// 调用runtime.time的startTimer方法

	startTimer(&t.r)

	return t

}

func startTimer(*runtimeTimer)

NewTimer方法主要是初始化一个Timer，然后调用startTimer方法，并返回Timer。startTimer方法的真正逻辑并不在time包里面，我们可以使用到上一节提到的使用dlv调试汇编代码：

sleep.go:94     0xd8ea09        e872c7faff              call $time.startTimer

得知startTimer实际上调用的是runtime.time.startTimer方法。也就是说time.Timer只是对runtime包中timer的一层wrap。这层自身实现的最核心功能是将底层的超时回调转换为发送channel消息。

下面我们看看runtime.startTimer：

func startTimer(t *timer) {

	...

	addtimer(t)

}

startTimer方法会将传入的runtimeTimer转为timer，然后调用addtimer方法。

在NewTimer方法中会初始化一个runtimeTimer结构体，这个结构体实际上会被当做runtime.time中的timer结构体传入到startTimer方法中，所以下面我们来看看timer：

type timer struct {

	// 对应处理器P的指针

	pp puintptr

	// 定时器被唤醒的时间

	when   int64

    // 唤醒的间隔时间

	period int64

	// 唤醒时被调用的函数

	f      func(interface{}, uintptr)

	// 被调用的函数的参数

	arg    interface{}

	seq    uintptr

	// 处于timerModifiedXX状态时用于设置when字段

	nextwhen int64

	// 定时器的状态

	status uint32

}

除此之外，timer还有一些标志位来表示 status 状态：

const (

	// 初始化状态

	timerNoStatus = iota

	// 等待被调用

	// timer 已在 P 的列表中

	timerWaiting

	// 表示 timer 在运行中

	timerRunning

	// timer 已被删除

	timerDeleted

	// timer 正在被移除

	timerRemoving

	// timer 已被移除，并停止运行

	timerRemoved

	// timer 被修改了

	timerModifying

  	// 被修改到了更早的时间

	timerModifiedEarlier 

  	// 被修改到了更晚的时间

	timerModifiedLater

  // 已经被修改，并且正在被移动

	timerMoving

)

addtimer 新增 timer

runtime.addtimer

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

	// 获取当前 P

	pp := getg().m.p.ptr()

	lock(&pp.timersLock)

	// 清理 P 的 timer 列表头中的 timer

	cleantimers(pp)

	// 将 timer 加入到 P 的最小堆中

	doaddtimer(pp, t)

	unlock(&pp.timersLock)

	// 唤醒 netpoller 中休眠的线程

	wakeNetPoller(when)

}

addtimer 会对 timer 被唤醒的时间 when 进行校验，以及校验 status 必须是新出初始化的 timer；
接着会在加锁后调用 cleantimers 对 P 中对应的 timer 列表的头节点进行清理工作，清理完后调用 doaddtimer 将 timer 加入到 P 的最小堆中，并释放锁；
调用 wakeNetPoller 唤醒 netpoller 中休眠的线程。

下面分别来看看 addtimer 中几个重要函数的具体实现：

runtime.cleantimers

func cleantimers(pp *p) {

	gp := getg()

	for {

		// 调度器列表为空，直接返回

		if len(pp.timers) == 0 {

			return

		}

		// 如果当前 G 被抢占了，直接返回

		if gp.preemptStop {

			return

		}

		// 获取第一个 timer

		t := pp.timers[0]

		if t.pp.ptr() != pp {

			throw("cleantimers: bad p")

		}

		switch s := atomic.Load(&t.status); s {

		case timerDeleted:

			// 设置 timer 的状态

			if !atomic.Cas(&t.status, s, timerRemoving) {

				continue

			}

			// 删除第一个 timer

			dodeltimer0(pp)

			// 删除完毕后重置状态为 timerRemoved

			if !atomic.Cas(&t.status, timerRemoving, timerRemoved) {

				badTimer()

			}

			atomic.Xadd(&pp.deletedTimers, -1)

		// timer 被修改到了更早或更晚的时间

		case timerModifiedEarlier, timerModifiedLater:

			// 将 timer 状态设置为 timerMoving

			if !atomic.Cas(&t.status, s, timerMoving) {

				continue

			}

			// 重新设置 when 字段

			t.when = t.nextwhen

			// 在列表中删除后重新加入

			dodeltimer0(pp)

			doaddtimer(pp, t)

			if s == timerModifiedEarlier {

				atomic.Xadd(&pp.adjustTimers, -1)

			}

			// 设置状态为 timerWaiting

			if !atomic.Cas(&t.status, timerMoving, timerWaiting) {

				badTimer()

			}

		default:

			return

		}

	}

}

cleantimers 函数中使用了一个无限循环来获取头节点。如果头节点的状态是 timerDeleted ，那么需要从 timer 列表中删除；如果头节点的状态是 timerModifiedEarlier 或 timerModifiedLater ，表示头节点的触发的时间被修改到了更早或更晚的时间，那么就先从 timer队列中删除再重新添加。

runtime.doaddtimer

func doaddtimer(pp *p, t *timer) {

	// Timers 依赖于 netpoller

	// 所以如果 netpoller 没有启动，需要启动一下

	if netpollInited == 0 {

		netpollGenericInit()

	}

	// 校验是否早已在 timer 列表中

	if t.pp != 0 {

		throw("doaddtimer: P already set in timer")

	}

	// 设置 timer 与 P 的关联

	t.pp.set(pp)

	i := len(pp.timers)

	// 将 timer 加入到 P 的 timer 列表中

	pp.timers = append(pp.timers, t)

	// 维护 timer 在 最小堆中的位置

	siftupTimer(pp.timers, i)

	// 如果 timer 是列表中头节点，需要设置一下 timer0When

	if t == pp.timers[0] {

		atomic.Store64(&pp.timer0When, uint64(t.when))

	}

	atomic.Xadd(&pp.numTimers, 1)

}

doaddtimer 函数实际上很简单，主要是将 timer 与 P 设置关联关系，并将 timer 加入到 P 的 timer 列表中，并维护 timer 列表最小堆的顺序。

runtime.wakeNetPoller

func wakeNetPoller(when int64) {

	if atomic.Load64(&sched.lastpoll) == 0 {

		pollerPollUntil := int64(atomic.Load64(&sched.pollUntil))

		// 如果计时器的触发时间小于netpoller的下一次轮询时间

		if pollerPollUntil == 0 || pollerPollUntil > when {

			// 向netpollBreakWr里面写入数据，立即中断netpoll

			netpollBreak()

		}

	}

}

func netpollBreak() {

	if atomic.Cas(&netpollWakeSig, 0, 1) {

		for {

			var b byte

            // 向 netpollBreakWr 里面写入数据

			n := write(netpollBreakWr, unsafe.Pointer(&b), 1)

			if n == 1 {

				break

			}

			if n == -_EINTR {

				continue

			}

			if n == -_EAGAIN {

				return

			}

			println("runtime: netpollBreak write failed with", -n)

			throw("runtime: netpollBreak write failed")

		}

	}

}

wakeNetPoller 主要是将 timer 下次调度的时间和 netpoller 的下一次轮询时间相比，如果小于的话，调用 netpollBreak 向 netpollBreakWr 里面写入数据，立即中断netpoll。具体如何中断的，我们下面再聊。

stopTimer 终止 timer

终止 timer 的逻辑主要是 timer 的状态的变更：

如果该timer处于 timerWaiting 或 timerModifiedLater 或 timerModifiedEarlier：

timerModifying -> timerDeleted

如果该timer处于其他状态:

待状态改变或者直接返回

所以在终止 timer 的过程中不会去删除 timer，而是标记一个状态，等待被删除。

modTimer 修改 timer

func modtimer(t *timer, when, period int64, f func(interface{}, uintptr), arg interface{}, seq uintptr) bool {

	if when < 0 {

		when = maxWhen

	}

	status := uint32(timerNoStatus)

	wasRemoved := false

	var pending bool

	var mp *m

loop:

	for {

		// 修改 timer 状态

		switch status = atomic.Load(&t.status); status {

		...

	}

	t.period = period

	t.f = f

	t.arg = arg

	t.seq = seq

	// 如果 timer 已被删除，那么需要重新添加到 timer 列表中

	if wasRemoved {

		t.when = when

		pp := getg().m.p.ptr()

		lock(&pp.timersLock)

		doaddtimer(pp, t)

		unlock(&pp.timersLock)

		if !atomic.Cas(&t.status, timerModifying, timerWaiting) {

			badTimer()

		}

		releasem(mp)

		wakeNetPoller(when)

	} else {

		t.nextwhen = when

		newStatus := uint32(timerModifiedLater)

		// 如果修改后的时间小于修改前的时间，将状态设置为 timerModifiedEarlier

		if when < t.when {

			newStatus = timerModifiedEarlier

		}

		...

		if !atomic.Cas(&t.status, timerModifying, newStatus) {

			badTimer()

		}

		releasem(mp)

		// 如果修改时间提前，那么触发 netpoll 中断

		if newStatus == timerModifiedEarlier {

			wakeNetPoller(when)

		}

	}

	return pending

}

modtimer 进入到 for 循环后会根据不同的状态做状态设置以及必要字段的处理；如果是 timer 已被删除，那么需要重新添加到 timer 列表中；如果 timer 修改后的时间小于修改前的时间，将状态设置为 timerModifiedEarlier，修改时间提前，还需要触发 netpoll 中断。

timer 的运行

聊完了如何添加 timer，下面我们来看看 timer 是如何运行的。timer 的运行是交给 runtime.runtimer函数执行的，这个函数会检查 P 上最小堆的最顶上的 timer 的状态，根据状态做不同的处理。

func runtimer(pp *p, now int64) int64 {

	for {

		// 获取最小堆的第一个元素

		t := pp.timers[0]

		if t.pp.ptr() != pp {

			throw("runtimer: bad p")

		}

		// 获取 timer 状态

		switch s := atomic.Load(&t.status); s {

		// timerWaiting

		case timerWaiting:

			// 还没到时间，返回下次执行时间

			if t.when > now {

				// Not ready to run.

				return t.when

			}

			// 修改状态为 timerRunning

			if !atomic.Cas(&t.status, s, timerRunning) {

				continue

			}

			// 运行该 timer

			runOneTimer(pp, t, now)

			return 0

		// timerDeleted

		case timerDeleted:

			if !atomic.Cas(&t.status, s, timerRemoving) {

				continue

			}

			// 删除最小堆的第一个 timer

			dodeltimer0(pp)

			if !atomic.Cas(&t.status, timerRemoving, timerRemoved) {

				badTimer()

			}

			atomic.Xadd(&pp.deletedTimers, -1)

			if len(pp.timers) == 0 {

				return -1

			}

		// 需要重新移动位置的 timer

		case timerModifiedEarlier, timerModifiedLater:

			if !atomic.Cas(&t.status, s, timerMoving) {

				continue

			}

			t.when = t.nextwhen

			// 删除最小堆的第一个 timer

			dodeltimer0(pp)

			// 将该 timer 重新添加到最小堆

			doaddtimer(pp, t)

			if s == timerModifiedEarlier {

				atomic.Xadd(&pp.adjustTimers, -1)

			}

			if !atomic.Cas(&t.status, timerMoving, timerWaiting) {

				badTimer()

			}

		case timerModifying:

			osyield()

		case timerNoStatus, timerRemoved:

			badTimer()

		case timerRunning, timerRemoving, timerMoving:

			badTimer()

		default:

			badTimer()

		}

	}

}

runtimer 里面会启动一个 for 循环，不停的检查 P 的 timer 列表的第一个元素的状态。

如果该 timer 处于 timerWaiting，那么判断当前的时间大于 timer 执行的时间，则调用 runOneTimer 执行；
如果该 timer 处于 timerDeleted，表示该 timer 是需要被删除的，那么调用 dodeltimer0 删除最小堆的第一个 timer ，并修改其状态；
如果该 timer 状态是 timerModifiedEarlier 、timerModifiedLater，那么表示该 timer 的执行时间被修改过，需要重新调整它在最小堆中的位置，所以先调用 dodeltimer0 删除该 timer，再调用 doaddtimer 将该 timer 重新添加到最小堆。

runtime.runOneTimer

func runOneTimer(pp *p, t *timer, now int64) {

	...

	// 需要被执行的函数

	f := t.f

	// 被执行函数的参数

	arg := t.arg

	seq := t.seq

	// 表示该 timer 为 ticker，需要再次触发

	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)

	// 一次性 timer

	} else {

		// 删除该 timer.

		dodeltimer0(pp)

		if !atomic.Cas(&t.status, timerRunning, timerNoStatus) {

			badTimer()

		}

	}

	unlock(&pp.timersLock)

	// 运行该函数

	f(arg, seq)

	lock(&pp.timersLock)

	...

}

runOneTimer 会根据 period 是否大于0判断该 timer 是否需要反复执行，如果是的话需要重新调整 when 下次执行时间后重新调整该 timer 在堆中的位置。一次性 timer 的话会执行 dodeltimer0 删除该 timer ，最后运行 timer 中的函数；

timer 的触发

下面这里是我觉得比较有意思的地方，timer 的触发有两种：

从调度循环中直接触发；
另一种是Go语言的后台系统监控中会定时触发；

调度循环触发

调度循环，我在这篇文章 https://www.luozhiyun.com/archives/448 已经讲的很清楚了，不明白的同学可以自己再去看看。

整个调度循环会有三个地方去检查是否有可执行的 timer：

调用 runtime.schedule 执行调度时；
调用runtime.findrunnable获取可执行函数时；
调用runtime.findrunnable执行抢占时；

runtime.schedule

func schedule() {

	_g_ := getg()

	...

top:

	pp := _g_.m.p.ptr()

	...

	// 检查是否有可执行 timer 并执行

	checkTimers(pp, 0)

	var gp *g

	...

	if gp == nil {

		gp, inheritTime = findrunnable() // blocks until work is available

	}

	...

	execute(gp, inheritTime)

}

下面我们看看 checkTimers 做了什么：

runtime.checkTimers

func checkTimers(pp *p, now int64) (rnow, pollUntil int64, ran bool) {

	// 如果没有需要调整的 timer

	if atomic.Load(&pp.adjustTimers) == 0 {

		// 获取 timer0 的执行时间

		next := int64(atomic.Load64(&pp.timer0When))

		if next == 0 {

			return now, 0, false

		}

		if now == 0 {

			now = nanotime()

		}

		// 下次执行大于当前时间，

		if now < next {

			// 需要删除的 timer 个数小于 timer列表个数的4分之1，直接返回

			if pp != getg().m.p.ptr() || int(atomic.Load(&pp.deletedTimers)) <= int(atomic.Load(&pp.numTimers)/4) {

				return now, next, false

			}

		}

	}

	lock(&pp.timersLock)

	// 进行调整 timer

	adjusttimers(pp)

	rnow = now

	if len(pp.timers) > 0 {

		if rnow == 0 {

			rnow = nanotime()

		}

		for len(pp.timers) > 0 {

			// 查找堆中是否存在需要执行的 timer

			if tw := runtimer(pp, rnow); tw != 0 {

				if tw > 0 {

					pollUntil = tw

				}

				break

			}

			ran = true

		}

	}

	// 如果需要删除的 timer 超过了 timer 列表数量的四分之一，那么清理需要删除的 timer

	if pp == getg().m.p.ptr() && int(atomic.Load(&pp.deletedTimers)) > len(pp.timers)/4 {

		clearDeletedTimers(pp)

	}

	unlock(&pp.timersLock)

	return rnow, pollUntil, ran

}

checkTimers 中主要做了这么几件事：

检查是否有需要进行调整的 timer，如果没有需要执行的计时器时，直接返回；如果下一个要执行的 timer 没有到期并且需要删除的计时器较少（四分之一）时也会直接返回；
调用 adjusttimers 进行 timer 列表的调整，主要是维护 timer 列表的最小堆的顺序；
调用 runtime.runtimer查找堆中是否存在需要执行的timer， runtime.runtimer上面已经讲过了，这里不再赘述；
如果当前 Goroutine 的 P 和传入的 P 相同，并且需要删除的 timer 超过了 timer 列表数量的四分之一，那么调用 clearDeletedTimers 清理需要删除的 timer；

runtime.findrunnable

func findrunnable() (gp *g, inheritTime bool) {

	_g_ := getg()

top:

	_p_ := _g_.m.p.ptr()

	...

	// 检查 P 中可执行的 timer

	now, pollUntil, _ := checkTimers(_p_, 0)

	...

    // 如果 netpoll 已被初始化，并且 Waiters 大于零，并且 lastpoll 不为0

	if netpollinited() && atomic.Load(&netpollWaiters) > 0 && atomic.Load64(&sched.lastpoll) != 0 {

		// 尝试从netpoller获取Glist

		if list := netpoll(0); !list.empty() { // 无阻塞

			gp := list.pop()

			//将其余队列放入 P 的可运行G队列

			injectglist(&list)

			casgstatus(gp, _Gwaiting, _Grunnable)

			if trace.enabled {

				traceGoUnpark(gp, 0)

			}

			return gp, false

		}

	}

	...

	// 开始窃取

	for i := 0; i < 4; i++ {

		for enum := stealOrder.start(fastrand()); !enum.done(); enum.next() {

			if sched.gcwaiting != 0 {

				goto top

			}

			// 如果 i>2 表示如果其他 P 运行队列中没有 G ，将要从其他队列的 runnext 中获取

			stealRunNextG := i > 2 // first look for ready queues with more than 1 g

			// 随机获取一个 P

			p2 := allp[enum.position()]

			if _p_ == p2 {

				continue

			}

			// 从其他 P 的运行队列中获取一般的 G 到当前队列中

			if gp := runqsteal(_p_, p2, stealRunNextG); gp != nil {

				return gp, false

			}

			// 如果运行队列中没有 G，那么从 timers 中获取可执行的 timer

			if i > 2 || (i > 1 && shouldStealTimers(p2)) {

				// ran 为 true 表示有执行过 timer

				tnow, w, ran := checkTimers(p2, now)

				now = tnow

				if w != 0 && (pollUntil == 0 || w < pollUntil) {

					pollUntil = w

				}

				if ran {

					// 因为已经运行过 timer 了，说不定已经有准备就绪的 G 了

					// 再次检查本地队列尝试获取 G

					if gp, inheritTime := runqget(_p_); gp != nil {

						return gp, inheritTime

					}

					ranTimer = true

				}

			}

		}

	} 

	if ranTimer {

		// 执行完一个 timer 后可能存在已经就绪的 G

		goto top

	}

stop:

	...

	delta := int64(-1)

	if pollUntil != 0 {

		// checkTimers ensures that polluntil > now.

		delta = pollUntil - now

	}

	...

	// poll network

	// 休眠前再次检查 poll 网络

	if netpollinited() && (atomic.Load(&netpollWaiters) > 0 || pollUntil != 0) && atomic.Xchg64(&sched.lastpoll, 0) != 0 {

		...

		list := netpoll(delta) // 阻塞调用

		lock(&sched.lock)

		_p_ = pidleget()

		unlock(&sched.lock)

		if _p_ == nil {

			injectglist(&list)

		} else {

			acquirep(_p_)

			if !list.empty() {

				gp := list.pop()

				injectglist(&list)

				casgstatus(gp, _Gwaiting, _Grunnable)

				if trace.enabled {

					traceGoUnpark(gp, 0)

				}

				return gp, false

			}

			goto top

		}

	} else if pollUntil != 0 && netpollinited() {

		pollerPollUntil := int64(atomic.Load64(&sched.pollUntil))

		if pollerPollUntil == 0 || pollerPollUntil > pollUntil {

			netpollBreak()

		}

	}

	// 休眠当前 M

	stopm()

	goto top

}

findrunnable 我在这篇文章 https://www.luozhiyun.com/archives/448 已经讲的很清楚了，这里提取 timer 相关的代码分析一下：

findrunnable 在窃取前先会调用 checkTimers 检查 P 中可执行的 timer；
如果 netpoll 中有等待的 waiter，那么会调用 netpoll 尝试无阻塞的从netpoller获取Glist；
如果获取不到可执行的 G，那么就会开始执行窃取。窃取的时候会调用 checkTimers 随机从其他的 P 中获取 timer；
窃取完毕后也没有可执行的 timer，那么会继续往下，休眠前再次检查 netpoll 网络，调用 netpoll(delta) 函数进行阻塞调用。

系统监控触发

系统监控其实就是 Go 语言的守护进程，它们能够在后台监控系统的运行状态，在出现意外情况时及时响应。它会每隔一段时间检查 Go 语言运行时状态，确保没有异常发生。我们这里不主要去讲系统监控，只抽离出其中的和 timer 相关的代码进行讲解。

runtime.sysmon

func sysmon() {

	...

	for {

		...

		now := nanotime()

		// 返回下次需要调度 timer 到期时间

		next, _ := timeSleepUntil()

		...

		// 如果超过 10ms 没有 poll，则 poll 一下网络

		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) // 非阻塞，返回 G 列表

			// G 列表不为空

			if !list.empty() {

				incidlelocked(-1)

				// 将获取到的 G 列表插入到空闲的 P 中或全局列表中

				injectglist(&list)

				incidlelocked(1)

			}

		}

		// 如果有 timer 到期

		if next < now {

			// 启动新的 M 处理 timer

			startm(nil, false)

		}

		...

	}

}

sysmon 会通过 timeSleepUntil 遍历所有的 P 的 timer 列表，找到下一个需要执行的 timer；
如果超过 10ms 没有 poll，则 poll 一下网络；
如果有 timer 到期，这个时候直接启动新的 M 处理 timer；

netpoll 的作用

我们从一开始调用 runtime.addtimer 添加 timer 的时候，就会 runtime.wakeNetPoller来中断 netpoll ，那么它是如何做到的？我们下面先来看一个官方的例子：

func TestNetpollBreak(t *testing.T) {

	if runtime.GOMAXPROCS(0) == 1 {

		t.Skip("skipping: GOMAXPROCS=1")

	}

	// 初始化 netpoll

	runtime.NetpollGenericInit()

	start := time.Now()

	c := make(chan bool, 2)

	go func() {

		c <- true

		// netpoll 等待时间

		runtime.Netpoll(10 * time.Second.Nanoseconds())

		c <- true

	}()

	<-c

loop:

	for {

		runtime.Usleep(100)

		// 中断netpoll 等待

		runtime.NetpollBreak()

		runtime.NetpollBreak()

		select {

		case <-c:

			break loop

		default:

		}

	}

	if dur := time.Since(start); dur > 5*time.Second {

		t.Errorf("netpollBreak did not interrupt netpoll: slept for: %v", dur)

	}

}

在上面这个例子中，首先会调用 runtime.Netpoll进行阻塞等待，然后循环调度 runtime.NetpollBreak进行中断阻塞。

runtime.netpoll

func netpoll(delay int64) gList {

	if epfd == -1 {

		return gList{}

	}

	var waitms int32

	// 因为传入delay单位是纳秒，下面将纳秒转换成毫秒

	if delay < 0 {

		waitms = -1

	} else if delay == 0 {

		waitms = 0

	} else if delay < 1e6 {

		waitms = 1

	} else if delay < 1e15 {

		waitms = int32(delay / 1e6)

	} else {

		waitms = 1e9

	}

	var events [128]epollevent

retry:

	// 等待文件描述符转换成可读或者可写

	n := epollwait(epfd, &events[0], int32(len(events)), waitms)

	// 返回负值，那么重新调用epollwait进行等待

	if n < 0 {

		...

		goto retry

	}

	var toRun gList

	for i := int32(0); i < n; i++ {

		ev := &events[i]

		if ev.events == 0 {

			continue

		}

		// 如果是 NetpollBreak 中断的，那么执行 continue 跳过

		if *(**uintptr)(unsafe.Pointer(&ev.data)) == &netpollBreakRd {

			if ev.events != _EPOLLIN {

				println("runtime: netpoll: break fd ready for", ev.events)

				throw("runtime: netpoll: break fd ready for something unexpected")

			}

			if delay != 0 {

				var tmp [16]byte

				read(int32(netpollBreakRd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))

				atomic.Store(&netpollWakeSig, 0)

			}

			continue

		}

		...

	}

	return toRun

}

在调用runtime.findrunnable执行抢占时，最后会传入一个时间，超时阻塞调用 netpoll，如果没有事件中断，那么循环调度会一直等待直到 netpoll 超时后才往下进行：

func findrunnable() (gp *g, inheritTime bool) {

	...

	delta := int64(-1)

	if pollUntil != 0 {

		// checkTimers ensures that polluntil > now.

		delta = pollUntil - now

	}

	...

	// poll network

	// 休眠前再次检查 poll 网络

	if netpollinited() && (atomic.Load(&netpollWaiters) > 0 || pollUntil != 0) && atomic.Xchg64(&sched.lastpoll, 0) != 0 {

		...

		// 阻塞调用

		list := netpoll(delta) 

	}

	...

	// 休眠当前 M

	stopm()

	goto top

}

所以在调用 runtime.addtimer 添加 timer 的时候进行 netpoll 的中断操作可以更加灵敏的响应 timer 这类时间敏感的任务。

总结

我们通过 timer 的 1.13版本以及1.14版本后的对比可以发现，即使是一个定时器 go 语言都做了相当多的优化工作。从原来的需要维护 64 个桶，然后每个桶里面跑异步任务，到现在的将 timer列表直接挂到了 P 上面，这不仅减少了上下文切换带来的性能损耗，也减少了在锁之间的争抢问题，通过这些优化后有了可以媲美时间轮的性能表现。