Golang原始碼學習：監控執行緒
阿新 • • 發佈：2020-05-28
監控執行緒是在runtime.main執行的時候在系統棧中建立的，監控執行緒與普通的工作執行緒區別在於，監控執行緒不需要繫結p來執行。
## 監控執行緒的建立與啟動
### 簡單的呼叫圖
先給出個簡單的呼叫圖，好心裡有數，逐個分析完後做個小結。
### 主體程式碼
以下會合並小篇幅且易懂的程式碼段，個人認為重點的會單獨摘出來。
#### main->newm->newm1->newosproc
```
func main() {
......
if GOARCH != "wasm" { // no threads on wasm yet, so no sysmon
systemstack(func() {
newm(sysmon, nil)
})
}
......
}
func newm(fn func(), _p_ *p) {
mp := allocm(_p_, fn) // 分配一個m
mp.nextp.set(_p_)
mp.sigmask = initSigmask
......
newm1(mp)
}
func newm1(mp *m) {
......
execLock.rlock() // Prevent process clone.
newosproc(mp)
execLock.runlock()
}
cloneFlags = _CLONE_VM | /* share memory */
_CLONE_FS | /* share cwd, etc */
_CLONE_FILES | /* share fd table */
_CLONE_SIGHAND | /* share sig handler table */
_CLONE_SYSVSEM | /* share SysV semaphore undo lists (see issue #20763) */
_CLONE_THREAD /* revisit - okay for now */
func newosproc(mp *m) {
stk := unsafe.Pointer(mp.g0.stack.hi)
......
sigprocmask(_SIG_SETMASK, &sigset_all, &oset)
// 這裡注意一下，mstart會被作為工作執行緒的開始，在runtime.clone中會被呼叫。
ret := clone(cloneFlags, stk, unsafe.Pointer(mp), unsafe.Pointer(mp.g0), unsafe.Pointer(funcPC(mstart)))
sigprocmask(_SIG_SETMASK, &oset, nil)
......
}
```
#### allocm
在此場景中其工作是new一個m，m.mstartfn = sysmon。
分配一個g與mp相互繫結，這個g就是mp的g0。但不是全域性變數的那個g0，**全域性變數g0是m0的m.g0。**
```
func allocm(_p_ *p, fn func()) *m {
_g_ := getg()
acquirem() // disable GC because it can be called from sysmon
// 忽略sysmon不會執行的程式碼
mp := new(m) // 新建一個m
mp.mstartfn = fn // fn 指向 sysmon
mcommoninit(mp) // 之前的文章有分析過，做一些初始化工作。
if iscgo || GOOS == "solaris" || GOOS == "illumos" || GOOS == "windows" || GOOS == "plan9" || GOOS == "darwin" {
mp.g0 = malg(-1)
} else {
mp.g0 = malg(8192 * sys.StackGuardMultiplier) // 分配一個g
}
mp.g0.m = mp
......
releasem(_g_.m)
return mp
}
```
#### runtime.clone
呼叫clone，核心會創建出一個子執行緒，返回兩次。返回0是子執行緒，否則是父執行緒。
效果與fork類似，其實是fork封裝了clone。
```
// int32 clone(int32 flags, void *stk, M *mp, G *gp, void (*fn)(void));
TEXT runtime·clone(SB),NOSPLIT,$0
// 準備clone系統呼叫的引數
MOVL flags+0(FP), DI
MOVQ stk+8(FP), SI
MOVQ $0, DX
MOVQ $0, R10
// 從父程序棧複製mp, gp, fn。子執行緒會用到。
MOVQ mp+16(FP), R8
MOVQ gp+24(FP), R9
MOVQ fn+32(FP), R12
// 呼叫clone
MOVL $SYS_clone, AX
SYSCALL
// 父執行緒，返回.
CMPQ AX, $0
JEQ 3(PC)
MOVL AX, ret+40(FP)
RET
// 子執行緒，設定棧頂
MOVQ SI, SP
// If g or m are nil, skip Go-related setup.
CMPQ R8, $0 // m
JEQ nog
CMPQ R9, $0 // g
JEQ nog
// 呼叫系統呼叫 gettid 獲取執行緒id初始化 mp.procid
MOVL $SYS_gettid, AX
SYSCALL
MOVQ AX, m_procid(R8)
// 設定執行緒tls
LEAQ m_tls(R8), DI
CALL runtime·settls(SB)
// In child, set up new stack
get_tls(CX)
MOVQ R8, g_m(R9) // gp.m = mp
MOVQ R9, g(CX) // mp.tls[0] = gp
CALL runtime·stackcheck(SB)
nog:
// Call fn
CALL R12 // 呼叫fn，此處是mstart，永不返回。
// It shouldn't return. If it does, exit that thread.
MOVL $111, DI
MOVL $SYS_exit, AX
SYSCALL
JMP -3(PC) // keep exiting
```
總結一下clone的工作：
- 準備系統呼叫clone的引數
- 將mp，gp，fn從父執行緒棧複製到暫存器中，給子執行緒用
- 呼叫clone
- 父執行緒返回
- 子執行緒設定 m.procid、tls、gp，mp互相繫結、呼叫fn
#### 呼叫sysmon
在newosproc中呼叫clone，並將 mstart 的地址傳入。也就是整個執行緒開始執行。
mstart 與 mstart1 在之前的文章有分析過，現在來看一下與本文有關的段落。
```
func mstart1() {
_g_ := getg()
save(getcallerpc(), getcallersp())
asminit()
minit()
// 之前初始化時的呼叫邏輯是 rt0_go->mstart->mstart1，當時這裡的fn == nil。所以會繼續向下走，進入排程迴圈。
// 現在呼叫邏輯是通過 newm(sysmon, nil)->allocm 中設定了 mp.mstartfn 為 sysmon的指標。所以下面的 fn 就不是 nil 了
// fn != nil 呼叫 sysmon，並且sysmon永不會返回。也就是說不會走到下面schedule中。
if fn := _g_.m.mstartfn; fn != nil {
fn()
}
......
schedule()
}
```
### 小結
監控執行緒通過在runtime.main中呼叫**newm(sysmon, nil)**建立。
- **newm**：呼叫了allocm 獲得了mp。
- **allocm**：new了一個m，也就是前面的mp。並且將 **mp.mstartfn 賦值為 sysmon的指標**，這很重要，後面會用。
- **newm->newm1->newosproc->runtime.clone**(cloneFlags, stk, unsafe.Pointer(mp), unsafe.Pointer(mp.g0), unsafe.Pointer(funcPC(mstart)))
- **runtime.clone**：準備系統呼叫clone的引數；從父執行緒棧複製mp，gp，fn到暫存器；呼叫clone；父執行緒返回；子執行緒設定sp，m.procid，tls，互相繫結mp與gp。呼叫mstart作為子執行緒的開始執行。
- **mstart->mstart1**：呼叫 \_g\_.m.mstartfn 指向的函式，也就是sysmon，此時監控工作正式開始。
## 搶佔排程
### 主體程式碼
sysmon開始會檢查死鎖，接下來是函式主體，一個無限迴圈，每隔一個短時間執行一次。其工作包含網路輪詢、搶佔排程、垃圾回收。
#### sysmon中搶佔排程程式碼
```
func sysmon() {
lock(&sched.lock)
sched.nmsys++
checkdead()
unlock(&sched.lock)
lasttrace := int64(0)
idle := 0 // how many cycles in succession we had not wokeup somebody
delay := uint32(0) // 睡眠時間，開始是20微秒；idle大於50後，翻倍增長；但最大為10毫秒
for {
if idle == 0 { // start with 20us sleep...
delay = 20
} else if idle > 50 { // start doubling the sleep after 1ms...
delay *= 2
}
if delay > 10*1000 { // up to 10ms
delay = 10 * 1000
}
usleep(delay)
now := nanotime()
......
// retake P's blocked in syscalls
// and preempt long running G's
if retake(now) != 0 {
idle = 0
} else {
idle++
}
......
}
}
```
#### retake
- **preemptone**：搶佔執行時間過長的G。
- **handoffp**：嘗試為過長時間處在_Psyscall的P關聯一個M繼續排程。
```
func retake(now int64) uint32 {
n := 0
lock(&allpLock)
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 {
// 如果執行時間太長，則搶佔g
t := int64(_p_.schedtick)
if int64(pd.schedtick) != t {
pd.schedtick = uint32(t)
pd.schedwhen = now
} else if pd.schedwhen+forcePreemptNS <= now {
preemptone(_p_) // 在系統呼叫的情況下，preemptone() 不會工作，因為P沒有與之關聯的M。
sysretake = true
}
}
// 因為此時P的狀態是 _Psyscall，所以是呼叫過了Syscall(或者Syscall6)開頭的 entersyscall 函式，而此函式會解綁P和M，所以 p.m = 0；m.p=0。
if s == _Psyscall {
......
// p的local佇列為空 && （存在自旋的m || 存在空閒的p） && 距離上次系統呼叫不超過10ms ==> # 如果rsp較小,表示棧有溢位風險,呼叫runtime.morestack_noctxt
// 此處省略具體函式彙編程式碼
0x00000000004583b0 <+80>: callq 0x451b30
0x00000000004583b5 <+85>: jmp 0x458360
```
假設上面的彙編程式碼是屬於一個叫 caller 的函式的（實際上確實是的）。
當執行caller的G（暫且稱其為gp）由於執行時間過長，被監控執行緒sysmon通過preemptone函式標記其 gp.preempt = true；gp.stackguard0 = stackPreempt。
當caller被呼叫時，會先進行棧的檢查，因為 stackPreempt 是一個大於任何真實sp的值，所以jbe指令跳轉呼叫 runtime.morestack_noctxt 。
#### goschedImpl
goschedImpl是搶佔排程的關鍵邏輯，從 morestack_noctxt 到 goschedImpl 的呼叫鏈如下：
morestack_noctxt->morestack->newstack->gopreempt_m->goschedImpl。其中 morestack_noctxt 和 morestack 由彙編編寫。
**goschedImpl** 的主要邏輯是：
- **更改gp的狀態為_Grunable，dropg解綁G和M**
- **globrunqput放入全域性佇列**
- **schedule重新排程**
```
func newstack() {
thisg := getg()
......
gp := thisg.m.curg
preempt := atomic.Loaduintptr(&gp.stackguard0) == stackPreempt
......
if preempt {
......
// Act like goroutine called runtime.Gosched.
gopreempt_m(gp) // never return
}
}
func gopreempt_m(gp *g) {
if trace.enabled {
traceGoPreempt()
}
goschedImpl(gp)
}
func goschedImpl(gp *g) {
status := readgstatus(gp)
if status&^_Gscan != _Grunning {
dumpgstatus(gp)
throw("bad g status")
}
casgstatus(gp, _Grunning, _Grunnable) // 狀態從 _Grunning 改為 _Grunnable。你執行的太久了，下來吧你。
dropg() // 解綁G和M
lock(&sched.lock)
globrunqput(gp) // 放入全域性佇列
unlock(&sched.lock)
schedule() // 重新排程，進入排程迴圈。
}
```
### 基於訊號的非同步搶佔
上述的 preemptone函式會呼叫preemptM函式，並且最終會呼叫tgkill系統呼叫，向需要被搶佔的G所在的工作執行緒傳送 _SIGURG 訊號。
#### 傳送訊號
```
func preemptM(mp *m) {
......
signalM(mp, sigPreempt)
}
// signalM sends a signal to mp.
func signalM(mp *m, sig int) {
tgkill(getpid(), int(mp.procid), sig)
}
TEXT ·tgkill(SB),NOSPLIT,$0
MOVQ tgid+0(FP), DI
MOVQ tid+8(FP), SI
MOVQ sig+16(FP), DX
MOVL $SYS_tgkill, AX
SYSCALL
RET
```
#### 執行搶佔
核心在收到 _SIGURG 訊號後，會呼叫該執行緒註冊的訊號處理程式，最終會執行到以下程式。
因為註冊邏輯不是問題的關注核心，所以就放在後面有介紹。
```
func sighandler(sig uint32, info *siginfo, ctxt unsafe.Pointer, gp *g) {
_g_ := getg()
c := &sigctxt{info, ctxt}
......
if sig == sigPreempt { // const sigPreempt
doSigPreempt(gp, c)
}
......
}
func doSigPreempt(gp *g, ctxt *sigctxt) {
// Check if this G wants to be preempted and is safe to
// preempt.
if wantAsyncPreempt(gp) && isAsyncSafePoint(gp, ctxt.sigpc(), ctxt.sigsp(), ctxt.siglr()) {
// Inject a call to asyncPreempt.
ctxt.pushCall(funcPC(asyncPreempt))
}
// Acknowledge the preemption.
atomic.Xadd(&gp.m.preemptGen, 1)
}
func (c *sigctxt) pushCall(targetPC uintptr) {
// Make it look like the signaled instruction called target.
pc := uintptr(c.rip())
sp := uintptr(c.rsp())
sp -= sys.PtrSize
*(*uintptr)(unsafe.Pointer(sp)) = pc
c.set_rsp(uint64(sp))
c.set_rip(uint64(targetPC)) // pc指向asyncPreempt
}
// asyncPreempt->asyncPreempt2
func asyncPreempt2() {
gp := getg()
gp.asyncSafePoint = true
if gp.preemptStop {
mcall(preemptPark)
} else {
mcall(gopreempt_m)
}
gp.asyncSafePoint = false
}
// gopreempt_m裡呼叫了goschedImpl，這個函式上面分析過，是完成搶佔的關鍵。此時也就是完成了搶佔，進入排程迴圈。
func gopreempt_m(gp *g) {
if trace.enabled {
traceGoPreempt()
}
goschedImpl(gp)
}
```
### 訊號處理程式的註冊與執行
#### 註冊
m0的訊號處理程式是在整個程式一開始就在 mstart1 中開始註冊的。
而其他M所屬執行緒因為在clone的時候指定了 _CLONE_SIGHAND 標記，共享了訊號handler table。所以一出生就有了。
註冊邏輯如下：
```
// 省略了一些無關程式碼
func mstart1() {
if _g_.m == &m0 {
mstartm0()
}
}
func mstartm0() {
initsig(false)
}
// 迴圈註冊訊號處理程式
func initsig(preinit bool) {
for i := uint32(0); i < _NSIG; i++ {
......
setsig(i, funcPC(sighandler))
}
}
// sigtramp註冊為處理程式
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 GOARCH == "386" || GOARCH == "amd64" {
sa.sa_restorer = funcPC(sigreturn)
}
if fn == funcPC(sighandler) {
if iscgo {
fn = funcPC(cgoSigtramp)
} else {
fn = funcPC(sigtramp)
}
}
sa.sa_handler = fn
sigaction(i, &sa, nil)
}
// sigaction->