從原始碼剖析Go語言基於訊號搶佔式排程
阿新 • • 發佈：2021-03-28
> 轉載請宣告出處哦~，本篇文章釋出於luozhiyun的部落格：https://www.luozhiyun.com/archives/485
>
> 本文使用的go的原始碼15.7
這一次來講講基於訊號式搶佔式排程。
## 介紹
在 Go 的 1.14 版本之前搶佔試排程都是基於協作的，需要自己主動的讓出執行，但是這樣是無法處理一些無法被搶佔的邊緣情況。例如：for 迴圈或者垃圾回收長時間佔用執行緒，這些問題中的一部分直到 1.14 才被基於訊號的搶佔式排程解決。
下面我們通過一個例子來驗證一下1.14 版本和 1.13 版本之間的搶佔差異：
```go
package main
import (
"fmt"
"os"
"runtime"
"runtime/trace"
"sync"
)
func main() {
runtime.GOMAXPROCS(1)
f, _ := os.Create("trace.output")
defer f.Close()
_ = trace.Start(f)
defer trace.Stop()
var wg sync.WaitGroup
for i := 0; i < 30; i++ {
wg.Add(1)
go func() {
defer wg.Done()
t := 0
for i:=0;i<1e8;i++ {
t+=2
}
fmt.Println("total:", t)
}()
}
wg.Wait()
}
```
這個例子中會通過 go trace 來進行執行過程的呼叫跟蹤。在程式碼中指定 `runtime.GOMAXPROCS(1)`設定最大的可同時使用的 CPU 核數為1，只用一個 P（處理器），這樣就確保是單處理器的場景。然後呼叫一個 for 迴圈開啟 10 個 goroutines 來執行 func 函式，這是一個純計算且耗時的函式，防止 goroutines 空閒讓出執行。
下面我們編譯程式分析 trace 輸出：
```
$ go build -gcflags "-N -l" main.go
-N表示禁用優化
-l禁用內聯
$ ./main
```
然後我們獲取到 trace.output 檔案後進行視覺化展示：
```
$ go tool trace -http=":6060" ./trace.output
```
### Go1.13 trace 分析
從上面的這個圖可以看出：
1. 因為我們限定了只有一個 P，所以在 PROCS 這一欄裡面只有一個 Proc0；
2. 我們在 for 迴圈裡面啟動了 30 個 goroutines ，所以我們可以數一下 Proc0 裡面的顏色框框，剛好30 個；
3. 30 個 goroutines 在 Proc0 裡面是序列執行的，一個執行完再執行另一個，沒有進行搶佔；
4. 隨便點選一個 goroutines 的詳情欄可以看到 Wall Duration 為 0.23s 左右，表示這個 goroutines 持續執行了 0.23s，總共 10 個 goroutines 執行時間是 7s 左右；
5. 切入呼叫棧 Start Stack Trace 是 main.main.func1:20，在程式碼上面是 func 函式執行頭： `go func() `；
6. 切走呼叫棧 End Stack Trace 是 main.main.func1:26，在程式碼上是 func 函式最後執行列印：`fmt.Println("total:", t)`；
從上面的 trace 分析可以知道，Go 的協作式排程對 calcSum 函式是毫無作用的，一旦執行開始，只能等執行結束。每個 goroutine 耗費了 0.23s 這麼長的時間，也無法搶佔它的執行權。
### Go 1.14 以上 trace 分析
在 Go 1.14 之後引入了基於訊號的搶佔式排程，從上面的圖可以看到 Proc0 這一欄中密密麻麻都是 goroutines 在切換時的呼叫情況，不會再出現 goroutines 一旦執行開始，只能等執行結束這種情況。
上面跑動的時間是 4s 左右這個情況可以忽略，因為我是在兩臺配置不同的機器上跑的（主要是我閒麻煩要找兩臺一樣的機器）。
下面我們拉近了看一下明細情況：
通過這個明細可以看出：
1. 這個 goroutine 運行了 0.025s 就讓出執行了；
2. 切入呼叫棧 Start Stack Trace 是 main.main.func1:21，和上面一樣；
3. 切走呼叫棧 End Stack Trace 是 runtime.asyncPreempt:50 ，這個函式是收到搶佔訊號時執行的函式，從這個地方也能明確的知道，被非同步搶佔了；
## 分析
### 搶佔訊號的安裝
runtime/signal_unix.go
程式啟動時，在`runtime.sighandler`中註冊 `SIGURG` 訊號的處理函式`runtime.doSigPreempt`。
**initsig**
```go
func initsig(preinit bool) {
// 預初始化
if !preinit {
signalsOK = true
}
//遍歷訊號陣列
for i := uint32(0); i < _NSIG; i++ {
t := &sigtable[i]
//略過訊號：SIGKILL、SIGSTOP、SIGTSTP、SIGCONT、SIGTTIN、SIGTTOU
if t.flags == 0 || t.flags&_SigDefault != 0 {
continue
}
...
setsig(i, funcPC(sighandler))
}
}
```
在 initsig 函式裡面會遍歷所有的訊號量，然後呼叫 setsig 函式進行註冊。我們可以檢視 sigtable 這個全域性變數看看有什麼資訊：
```go
var sigtable = [...]sigTabT{
/* 0 */ {0, "SIGNONE: no trap"},
/* 1 */ {_SigNotify + _SigKill, "SIGHUP: terminal line hangup"},
/* 2 */ {_SigNotify + _SigKill, "SIGINT: interrupt"},
/* 3 */ {_SigNotify + _SigThrow, "SIGQUIT: quit"},
/* 4 */ {_SigThrow + _SigUnblock, "SIGILL: illegal instruction"},
/* 5 */ {_SigThrow + _SigUnblock, "SIGTRAP: trace trap"},
/* 6 */ {_SigNotify + _SigThrow, "SIGABRT: abort"},
/* 7 */ {_SigPanic + _SigUnblock, "SIGBUS: bus error"},
/* 8 */ {_SigPanic + _SigUnblock, "SIGFPE: floating-point exception"},
/* 9 */ {0, "SIGKILL: kill"},
/* 10 */ {_SigNotify, "SIGUSR1: user-defined signal 1"},
/* 11 */ {_SigPanic + _SigUnblock, "SIGSEGV: segmentation violation"},
/* 12 */ {_SigNotify, "SIGUSR2: user-defined signal 2"},
/* 13 */ {_SigNotify, "SIGPIPE: write to broken pipe"},
/* 14 */ {_SigNotify, "SIGALRM: alarm clock"},
/* 15 */ {_SigNotify + _SigKill, "SIGTERM: termination"},
/* 16 */ {_SigThrow + _SigUnblock, "SIGSTKFLT: stack fault"},
/* 17 */ {_SigNotify + _SigUnblock + _SigIgn, "SIGCHLD: child status has changed"},
/* 18 */ {_SigNotify + _SigDefault + _SigIgn, "SIGCONT: continue"},
/* 19 */ {0, "SIGSTOP: stop, unblockable"},
/* 20 */ {_SigNotify + _SigDefault + _SigIgn, "SIGTSTP: keyboard stop"},
/* 21 */ {_SigNotify + _SigDefault + _SigIgn, "SIGTTIN: background read from tty"},
/* 22 */ {_SigNotify + _SigDefault + _SigIgn, "SIGTTOU: background write to tty"},
/* 23 */ {_SigNotify + _SigIgn, "SIGURG: urgent condition on socket"},
/* 24 */ {_SigNotify, "SIGXCPU: cpu limit exceeded"},
/* 25 */ {_SigNotify, "SIGXFSZ: file size limit exceeded"},
/* 26 */ {_SigNotify, "SIGVTALRM: virtual alarm clock"},
/* 27 */ {_SigNotify + _SigUnblock, "SIGPROF: profiling alarm clock"},
/* 28 */ {_SigNotify + _SigIgn, "SIGWINCH: window size change"},
/* 29 */ {_SigNotify, "SIGIO: i/o now possible"},
/* 30 */ {_SigNotify, "SIGPWR: power failure restart"},
/* 31 */ {_SigThrow, "SIGSYS: bad system call"},
/* 32 */ {_SigSetStack + _SigUnblock, "signal 32"}, /* SIGCANCEL; see issue 6997 */
/* 33 */ {_SigSetStack + _SigUnblock, "signal 33"}, /* SIGSETXID; see issues 3871, 9400, 12498 */
...
}
```
具體的訊號含義可以看這個介紹：Unix訊號 https://zh.wikipedia.org/wiki/Unix%E4%BF%A1%E5%8F%B7。需要注意的是，搶佔訊號在這裡是 ` _SigNotify + _SigIgn` 如下：
```
{_SigNotify + _SigIgn, "SIGURG: urgent condition on socket"}
```
下面我們看一下 setsig 函式，這個函式是在 `runtime/os_linux.go`檔案裡面：
**setsig**
```go
func setsig(i uint32, fn uintptr) {
var sa sigactiont
sa.sa_flags = _SA_SIGINFO | _SA_ONSTACK | _SA_RESTORER | _SA_RESTART
sigfillset(&sa.sa_mask)
...
if fn == funcPC(sighandler) {
// CGO 相關
if iscgo {
fn = funcPC(cgoSigtramp)
} else {
// 替換為呼叫 sigtramp
fn = funcPC(sigtramp)
}
}
sa.sa_handler = fn
sigaction(i, &sa, nil)
}
```
這裡需要注意的是，當 fn 等於 sighandler 的時候，呼叫的函式會被替換成 sigtramp。sigaction 函式在 Linux 下會呼叫系統呼叫函式 sys_signal 以及 sys_rt_sigaction 實現安裝訊號。
### 執行搶佔訊號
到了這裡是訊號發生的時候進行訊號的處理，原本應該是在傳送搶佔訊號之後，但是這裡我先順著安裝訊號往下先講了。大家可以跳到傳送搶佔訊號後再回來。
上面分析可以看到當 fn 等於 sighandler 的時候，呼叫的函式會被替換成 sigtramp，sigtramp是彙編實現，下面我們看看。
`src/runtime/sys_linux_amd64.s`:
```
TEXT runtime·sigtramp(SB),NOSPLIT,$72
...
// We don't save mxcsr or the x87 control word because sigtrampgo doesn't
// modify them.
MOVQ DX, ctx-56(SP)
MOVQ SI, info-64(SP)
MOVQ DI, signum-72(SP)
MOVQ $runtime·sigtrampgo(SB), AX
CALL AX
...
RET
```
這裡會被呼叫說明訊號已經發送響應了，`runtime·sigtramp`會進行訊號的處理。`runtime·sigtramp`會繼續呼叫 `runtime·sigtrampgo` 。
這個函式在` runtime/signal_unix.go`檔案中：
**sigtrampgo&sighandler**
```go
func sigtrampgo(sig uint32, info *siginfo, ctx unsafe.Pointer) {
if sigfwdgo(sig, info, ctx) {
return
}
c := &sigctxt{info, ctx}
g := sigFetchG(c)
...
sighandler(sig, info, ctx, g)
setg(g)
if setStack {
restoreGsignalStack(&gsignalStack)
}
}
func sighandler(sig uint32, info *siginfo, ctxt unsafe.Pointer, gp *g) {
_g_ := getg()
c := &sigctxt{info, ctxt}
...
// 如果是一個搶佔訊號
if sig == sigPreempt && debug.asyncpreemptoff == 0 {
// 處理搶佔訊號
doSigPreempt(gp, c)
}
...
}
```
sighandler 方法裡面做了很多其他訊號的處理工作，我們只關心搶佔部分的程式碼，這裡最終會通過 doSigPreempt 方法執行搶佔。
這個函式在` runtime/signal_unix.go`檔案中：
**doSigPreempt**
```go
func doSigPreempt(gp *g, ctxt *sigctxt) {
// 檢查此 G 是否要被搶佔並且可以安全地搶佔
if wantAsyncPreempt(gp) {
// 檢查是否能安全的進行搶佔
if ok, newpc := isAsyncSafePoint(gp, ctxt.sigpc(), ctxt.sigsp(), ctxt.siglr()); ok {
// 修改暫存器,並執行搶佔呼叫
ctxt.pushCall(funcPC(asyncPreempt), newpc)
}
}
// 更新一下搶佔相關欄位
atomic.Xadd(&gp.m.preemptGen, 1)
atomic.Store(&gp.m.signalPending, 0)
}
```
函式會處理搶佔訊號，獲取當前的 SP 和 PC 暫存器並呼叫 `ctxt.pushCall`修改暫存器，並呼叫 `runtime/preempt.go` 的 asyncPreempt 函式。
```go
// 儲存使用者態暫存器後呼叫asyncPreempt2
func asyncPreempt()
```
asyncPreempt 的彙編程式碼在` src/runtime/preempt_amd64.s`中，該函式會儲存使用者態暫存器後呼叫 `runtime/preempt.go` 的 asyncPreempt2 函式中：
**asyncPreempt2**
```go
func asyncPreempt2() {
gp := getg()
gp.asyncSafePoint = true
// 該 G 是否可以被搶佔
if gp.preemptStop {
mcall(preemptPark)
} else {
// 讓 G 放棄當前在 M 上的執行權利,將 G 放入全域性佇列等待後續排程
mcall(gopreempt_m)
}
gp.asyncSafePoint = false
}
```
該函式會獲取當前 G ，然後判斷 G 的 preemptStop 值，preemptStop 會在呼叫 `runtime/preempt.go`的 suspendG 函式的時候將 `_Grunning` 狀態的 Goroutine 標記成可以被搶佔 `gp.preemptStop = true`，表示該 G 可以被搶佔。
下面我們看一下執行搶佔任務會呼叫的 `runtime/proc.go`的 preemptPark函式：
**preemptPark**
```go
func preemptPark(gp *g) {
status := readgstatus(gp)
if status&^_Gscan != _Grunning {
dumpgstatus(gp)
throw("bad g status")
}
gp.waitreason = waitReasonPreempted
casGToPreemptScan(gp, _Grunning, _Gscan|_Gpreempted)
// 使當前 m 放棄 g，讓出執行緒
dropg()
// 修改當前 Goroutine 的狀態到 _Gpreempted
casfrom_Gscanstatus(gp, _Gscan|_Gpreempted, _Gpreempted)
// 並繼續執行排程
schedule()
}
```
preemptPark 會修改當前 Goroutine 的狀態到 `_Gpreempted` ，呼叫 dropg 讓出執行緒，最後呼叫 schedule 函式繼續執行其他 Goroutine 的任務迴圈排程。
**gopreempt_m**
gopreempt_m 方法比起搶佔更像是主動讓權，然後重新加入到執行佇列中等待排程。
```go
func gopreempt_m(gp *g) {
goschedImpl(gp)
}
func goschedImpl(gp *g) {
status := readgstatus(gp)
...
// 更新狀態為 _Grunnable
casgstatus(gp, _Grunning, _Grunnable)
// 使當前 m 放棄 g，讓出執行緒
dropg()
lock(&sched.lock)
// 重新加入到全域性執行佇列中
globrunqput(gp)
unlock(&sched.lock)
// 並繼續執行排程
schedule()
}
```
### 搶佔訊號傳送
搶佔訊號的傳送是由 preemptM 進行的。
這個函式在`runtime/signal_unix.go`檔案中：
**preemptM**
```go
const sigPreempt = _SIGURG
func preemptM(mp *m) {
...
if atomic.Cas(&mp.signalPending, 0, 1) {
// preemptM 向 M 傳送搶佔請求。
// 接收到該請求後，如果正在執行的 G 或 P 被標記為搶佔，並且 Goroutine 處於非同步安全點，
// 它將搶佔 Goroutine。
signalM(mp, sigPreempt)
}
}
```
preemptM 這個函式會呼叫 signalM 將在初始化的安裝的 `_SIGURG` 訊號傳送到指定的 M 上。
使用 preemptM 傳送搶佔訊號的地方主要有下面幾個：
1. Go 後臺監控 runtime.sysmon 檢測超時傳送搶佔訊號；
2. Go GC 棧掃描傳送搶佔訊號；
3. Go GC STW 的時候呼叫 preemptall 搶佔所有 P，讓其暫停；
#### Go 後臺監控執行搶佔
系統監控 `runtime.sysmon` 會在迴圈中呼叫 `runtime.retake`搶佔處於執行或者系統呼叫中的處理器，該函式會遍歷執行時的全域性處理器。
系統監控通過在迴圈中搶佔主要是為了避免 G 佔用 M 的時間過長造成飢餓。
`runtime.retake`主要分為兩部分：
1. 呼叫 preemptone 搶佔當前處理器；
2. 呼叫 handoffp 讓出處理器的使用權；
**搶佔當前處理器**
```go
func retake(now int64) uint32 {
n := 0
lock(&allpLock)
// 遍歷 allp 陣列
for i := 0; i < len(allp); i++ {
_p_ := allp[i]
if _p_ == nil {
continue
}
pd := &_p_.sysmontick
s := _p_.status
sysretake := false
if s == _Prunning || s == _Psyscall {
// 排程次數
t := int64(_p_.schedtick)
if int64(pd.schedtick) != t {
pd.schedtick = uint32(t)
// 處理器上次排程時間
pd.schedwhen = now
// 搶佔 G 的執行，如果上一次觸發排程的時間已經過去了 10ms
} else if pd.schedwhen+forcePreemptNS <= now {
preemptone(_p_)
sysretake = true
}
}
...
}
unlock(&allpLock)
return uint32(n)
}
```
這一過程會獲取當前 P 的狀態，如果處於 `_Prunning` 或者 `_Psyscall` 狀態時，並且上一次觸發排程的時間已經過去了 10ms，那麼會呼叫 preemptone 進行搶佔訊號的傳送，preemptone 在上面我們已經講過了，這裡就不再複述。
**呼叫 handoffp 讓出處理器的使用權**
```go
func retake(now int64) uint32 {
n := 0
lock(&allpLock)
// 遍歷 allp 陣列
for i := 0; i < len(allp); i++ {
_p_ := allp[i]
if _p_ == nil {
continue
}
pd := &_p_.sysmontick
s := _p_.status
sysretake := false
...
if s == _Psyscall {
// 系統呼叫的次數
t := int64(_p_.syscalltick)
if !sysretake && int64(pd.syscalltick) != t {
pd.syscalltick = uint32(t)
// 系統呼叫的時間
pd.syscallwhen = now
continue
}
if runqempty(_p_) && atomic.Load(&sched.nmspinning)+atomic.Load(&sched.npidle) >