Go中定時器實現原理及原始碼解析
阿新 • • 發佈：2021-03-07
> 轉載請宣告出處哦~，本篇文章釋出於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`原始碼：
```go
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 中。
```go
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型別的元素值
```go
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來使用：
```go
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`可以在設定時間過後執行一個函式：
```go
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的：
```go
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`：
```go
func startTimer(t *timer) {
...
addtimer(t)
}
```
startTimer方法會將傳入的runtimeTimer轉為timer，然後呼叫addtimer方法。
在NewTimer方法中會初始化一個runtimeTimer結構體，這個結構體實際上會被當做`runtime.time`中的timer結構體傳入到startTimer方法中，所以下面我們來看看timer：
```go
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 狀態：
```go
const (
// 初始化狀態
timerNoStatus = iota
// 等待被呼叫
// timer 已在 P 的列表中
timerWaiting
// 表示 timer 在執行中
timerRunning
// timer 已被刪除
timerDeleted
// timer 正在被移除
timerRemoving
// timer 已被移除，並停止執行
timerRemoved
// timer 被修改了
timerModifying
// 被修改到了更早的時間
timerModifiedEarlier
// 被修改到了更晚的時間
timerModifiedLater
// 已經被修改，並且正在被移動
timerMoving
)
```
### addtimer 新增 timer
**runtime.addtimer**
```go
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)
}
```
1. addtimer 會對 timer 被喚醒的時間 when 進行校驗，以及校驗 status 必須是新出初始化的 timer；
2. 接著會在加鎖後呼叫 cleantimers 對 P 中對應的 timer 列表的頭節點進行清理工作，清理完後呼叫 doaddtimer 將 timer 加入到 P 的最小堆中，並釋放鎖；
3. 呼叫 wakeNetPoller 喚醒 netpoller 中休眠的執行緒。
下面分別來看看 addtimer 中幾個重要函式的具體實現：
**runtime.cleantimers**
```go
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**
```go
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**
```go
func wakeNetPoller(when int64) {
if atomic.Load64(&sched.lastpoll) == 0 {
pollerPollUntil := int64(atomic.Load64(&sched.pollUntil))
// 如果計時器的觸發時間小於netpoller的下一次輪詢時間
if pollerPollUntil == 0 || pollerPollUntil >