runtime.main() 函数中,执行 runtime_init() 前,会启动一个 sysmon 的监控线程,执行后台监控任务:

  1. systemstack(func(){
  2. // 创建监控线程,该线程独立于调度器,不需要跟 p 关联即可运行
  3. newm(sysmon,nil)
  4. })

sysmon 函数不依赖 P 直接执行,通过 newm 函数创建一个工作线程:

  1. func newm(fn func(), _p_ *p){
  2. // 创建 m 对象
  3. mp := allocm(_p_, fn)
  4. // 暂存 m
  5. mp.nextp.set(_p_)
  6. mp.sigmask = initSigmask
  7. // ……………………
  8. execLock.rlock()// Prevent process clone.
  9. // 创建系统线程
  10. newosproc(mp,unsafe.Pointer(mp.g0.stack.hi))
  11. execLock.runlock()
  12. }

先调用 allocm 在堆上创建一个 m,接着调用 newosproc 函数启动一个工作线程:

  1. // src/runtime/os_linux.go
  2. //go:nowritebarrier
  3. func newosproc(mp *m, stk unsafe.Pointer){
  4. // ……………………
  5. ret := clone(cloneFlags, stk,unsafe.Pointer(mp),unsafe.Pointer(mp.g0),unsafe.Pointer(funcPC(mstart)))
  6. // ……………………
  7. }

核心就是调用 clone 函数创建系统线程,新线程从 mstart 函数开始执行。clone 函数由汇编语言实现:

  1. // int32 clone(int32 flags, void *stk, M *mp, G *gp, void (*fn)(void));
  2. TEXT runtime·clone(SB),NOSPLIT,$0
  3. // 准备系统调用的参数
  4. MOVL flags+0(FP), DI
  5. MOVQ stk+8(FP), SI
  6. MOVQ $0, DX
  7. MOVQ $0, R10
  8. // 将 mp,gp,fn 拷贝到寄存器,对子线程可见
  9. MOVQ mp+16(FP), R8
  10. MOVQ gp+24(FP), R9
  11. MOVQ fn+32(FP), R12
  12. // 系统调用 clone
  13. MOVL $56, AX
  14. SYSCALL
  15. // In parent, return.
  16. CMPQ AX, $0
  17. JEQ 3(PC)
  18. // 父线程,返回
  19. MOVL AX, ret+40(FP)
  20. RET
  21. // In child, on new stack.
  22. // 在子线程中。设置 CPU 栈顶寄存器指向子线程的栈顶
  23. MOVQ SI, SP
  24. // If g or m are nil, skip Go-related setup.
  25. CMPQ R8, $0 // m
  26. JEQ nog
  27. CMPQ R9, $0 // g
  28. JEQ nog
  29. // Initialize m->procid to Linux tid
  30. // 通过 gettid 系统调用获取线程 ID(tid)
  31. MOVL $186, AX // gettid
  32. SYSCALL
  33. // 设置 m.procid = tid
  34. MOVQ AX, m_procid(R8)
  35. // Set FS to point at m->tls.
  36. // 新线程刚刚创建出来,还未设置线程本地存储,即 m 结构体对象还未与工作线程关联起来,
  37. // 下面的指令负责设置新线程的 TLS,把 m 对象和工作线程关联起来
  38. LEAQ m_tls(R8), DI
  39. CALL runtime·settls(SB)
  40. // In child, set up new stack
  41. get_tls(CX)
  42. MOVQ R8, g_m(R9)// g.m = m
  43. MOVQ R9, g(CX)// tls.g = &m.g0
  44. CALL runtime·stackcheck(SB)
  45. nog:
  46. // Call fn
  47. // 调用 mstart 函数。永不返回
  48. CALL R12
  49. // It shouldn't return. If it does, exit that thread.
  50. MOVL $111, DI
  51. MOVL $60, AX
  52. SYSCALL
  53. JMP -3(PC)// keep exiting

先是为 clone 系统调用准备参数,参数通过寄存器传递。第一个参数指定内核创建线程时的选项,第二个参数指定新线程应该使用的栈,这两个参数都是通过 newosproc 函数传递进来的。

接着将 m, g0, fn 分别保存到寄存器中,待子线程创建好后再拿出来使用。因为这些参数此时是在父线程的栈上,若不保存到寄存器中,子线程就取不出来了。

这个几个参数保存在父线程的寄存器中,创建子线程时,操作系统内核会把父线程所有的寄存器帮我们复制一份给子线程,所以当子线程开始运行时就能拿到父线程保存在寄存器中的值,从而拿到这几个参数。

之后,调用 clone 系统调用,内核帮我们创建出了一个子线程。相当于原来的一个执行分支现在变成了两个执行分支,于是会有两个返回。这和著名的 fork 系统调用类似,根据返回值来判断现在是处于父线程还是子线程。

如果是父线程,就直接返回了。如果是子线程,接着还要执行一堆操作,例如设置 tls,设置 m.procid 等等。

最后执行 mstart 函数,这是在 newosproc 函数传递进来的。mstart 函数再调用 mstart1,在 mstart1 里会执行这一行:

  1. // 执行启动函数。初始化过程中,fn == nil
  2. if fn := _g_.m.mstartfn; fn !=nil{
  3. fn()
  4. }

之前我们在讲初始化的时候,这里的 fn 是空,会跳过的。但在这里,fn 就是最开始在 runtime.main 里设置的 sysmon 函数,因此这里会执行 sysmon,而它又是一个无限循环,永不返回。

所以,这里不会执行到 mstart1 函数后面的 schedule 函数,也就不会进入 schedule 循环。因此这是一个不用和 p 结合的 m,它直接在后台执行,默默地执行监控任务。

接下来,我们就来看 sysmon 函数到底做了什么?

sysmon 执行一个无限循环,一开始每次循环休眠 20us,之后(1 ms 后)每次休眠时间倍增,最终每一轮都会休眠 10ms。

sysmon 中会进行 netpool(获取 fd 事件)、retake(抢占)、forcegc(按时间强制执行 gc),scavenge heap(释放自由列表中多余的项减少内存占用)等处理。

和调度相关的,我们只关心 retake 函数:

  1. func retake(now int64) uint32 {
  2. n :=0
  3. // 遍历所有的 p
  4. for i := int32(0); i < gomaxprocs; i++{
  5. _p_ := allp[i]
  6. if _p_ ==nil{
  7. continue
  8. }
  9. // 用于 sysmon 线程记录被监控 p 的系统调用时间和运行时间
  10. pd :=&_p_.sysmontick
  11. // p 的状态
  12. s := _p_.status
  13. if s ==_Psyscall{
  14. // P 处于系统调用之中,需要检查是否需要抢占
  15. // Retake P from syscall if it's there for more than 1 sysmon tick (at least 20us).
  16. // _p_.syscalltick 用于记录系统调用的次数,在完成系统调用之后加 1
  17. t := int64(_p_.syscalltick)
  18. if int64(pd.syscalltick)!= t {
  19. // pd.syscalltick != _p_.syscalltick,说明已经不是上次观察到的系统调用了,
  20. // 而是另外一次系统调用,所以需要重新记录 tick 和 when 值
  21. pd.syscalltick = uint32(t)
  22. pd.syscallwhen = now
  23. continue
  24. }
  25. // 只要满足下面三个条件中的任意一个,则抢占该 p,否则不抢占
  26. // 1. p 的运行队列里面有等待运行的 goroutine
  27. // 2. 没有无所事事的 p
  28. // 3. 从上一次监控线程观察到 p 对应的 m 处于系统调用之中到现在已经超过 10 毫秒
  29. if runqempty(_p_)&& atomic.Load(&sched.nmspinning)+atomic.Load(&sched.npidle)>0&& pd.syscallwhen+10*1000*1000> now {
  30. continue
  31. }
  32. incidlelocked(-1)
  33. if atomic.Cas(&_p_.status, s,_Pidle){
  34. // ……………………
  35. n++
  36. _p_.syscalltick++
  37. // 寻找一新的 m 接管 p
  38. handoffp(_p_)
  39. }
  40. incidlelocked(1)
  41. }elseif s ==_Prunning{
  42. // P 处于运行状态,检查是否运行得太久了
  43. // Preempt G if it's running for too long.
  44. // 每发生一次调度,调度器 ++ 该值
  45. t := int64(_p_.schedtick)
  46. if int64(pd.schedtick)!= t {
  47. pd.schedtick = uint32(t)
  48. pd.schedwhen = now
  49. continue
  50. }
  51. //pd.schedtick == t 说明(pd.schedwhen ~ now)这段时间未发生过调度
  52. // 这段时间是同一个goroutine一直在运行,检查是否连续运行超过了 10 毫秒
  53. if pd.schedwhen+forcePreemptNS > now {
  54. continue
  55. }
  56. // 连续运行超过 10 毫秒了,发起抢占请求
  57. preemptone(_p_)
  58. }
  59. }
  60. return uint32(n)
  61. }

从代码来看,主要会对处于 _Psyscall_Prunning 状态的 p 进行抢占。

抢占进行系统调用的 P

当 P 处于 _Psyscall 状态时,表明对应的 goroutine 正在进行系统调用。如果抢占 p,需要满足几个条件:

  1. p 的本地运行队列里面有等待运行的 goroutine。这时 p 绑定的 g 正在进行系统调用,无法去执行其他的 g,因此需要接管 p 来执行其他的 g。

  2. 没有“无所事事”的 p。sched.nmspinningsched.npidle 都为 0,这就意味着没有“找工作”的 m,也没有空闲的 p,大家都在“忙”,可能有很多工作要做。因此要抢占当前的 p,让它来承担一部分工作。

  3. 从上一次监控线程观察到 p 对应的 m 处于系统调用之中到现在已经超过 10 毫秒。这说明系统调用所花费的时间较长,需要对其进行抢占,以此来使得 retake 函数返回值不为 0,这样,会保持 sysmon 线程 20 us 的检查周期,提高 sysmon 监控的实时性。

注意,原代码是用的三个与条件,三者都要满足才会执行下面的 continue,也就是不进行抢占。因此要想进行抢占的话,只需要三个条件有一个不满足就行了。于是就有了上述三种情况。

确定要抢占当前 p 后,先使用原子操作将 p 的状态修改为 _Pidle,最后调用 handoffp 进行抢占。

  1. func handoffp(_p_ *p){
  2. // 如果 p 本地有工作或者全局有工作,需要绑定一个 m
  3. if!runqempty(_p_)|| sched.runqsize !=0{
  4. startm(_p_,false)
  5. return
  6. }
  7. // ……………………
  8. // 所有其它 p 都在运行 goroutine,说明系统比较忙,需要启动 m
  9. if atomic.Load(&sched.nmspinning)+atomic.Load(&sched.npidle)==0&& atomic.Cas(&sched.nmspinning,0,1){// TODO: fast atomic
  10. // p 没有本地工作,启动一个自旋 m 来找工作
  11. startm(_p_,true)
  12. return
  13. }
  14. lock(&sched.lock)
  15. // ……………………
  16. // 全局队列有工作
  17. if sched.runqsize !=0{
  18. unlock(&sched.lock)
  19. startm(_p_,false)
  20. return
  21. }
  22. // ……………………
  23. // 没有工作要处理,把 p 放入全局空闲队列
  24. pidleput(_p_)
  25. unlock(&sched.lock)
  26. }

handoffp 再次进行场景判断,以调用 startm 启动一个工作线程来绑定 p,使得整体工作继续推进。

当 p 的本地运行队列或全局运行队列里面有待运行的 goroutine,说明还有很多工作要做,调用 startm(_p_, false) 启动一个 m 来结合 p,继续工作。

当除了当前的 p 外,其他所有的 p 都在运行 goroutine,说明天下太平,每个人都有自己的事做,唯独自己没有。为了全局更快地完成工作,需要启动一个 m,且要使得 m 处于自旋状态,和 p 结合之后,尽快找到工作。

最后,如果实在没有工作要处理,就将 p 放入全局空闲队列里。

我们接着来看 startm 函数都做了些什么:

  1. // runtime/proc.go
  2. //
  3. // 调用 m 来绑定 p,如果没有 m,那就新建一个
  4. // 如果 p 为空,那就尝试获取一个处于空闲状态的 p,如果找到 p,那就什么都不做
  5. func startm(_p_ *p, spinning bool){
  6. lock(&sched.lock)
  7. if _p_ ==nil{
  8. // 没有指定 p 则需要从全局空闲队列中获取一个 p
  9. _p_ = pidleget()
  10. if _p_ ==nil{
  11. unlock(&sched.lock)
  12. if spinning {
  13. // 如果找到 p,放弃。还原全局处于自旋状态的 m 的数量
  14. if int32(atomic.Xadd(&sched.nmspinning,-1))<0{
  15. throw("startm: negative nmspinning")
  16. }
  17. }
  18. // 没有空闲的 p,直接返回
  19. return
  20. }
  21. }
  22. // 从 m 空闲队列中获取正处于睡眠之中的工作线程,
  23. // 所有处于睡眠状态的 m 都在此队列中
  24. mp := mget()
  25. unlock(&sched.lock)
  26. if mp ==nil{
  27. // 如果没有找到 m
  28. var fn func()
  29. if spinning {
  30. // The caller incremented nmspinning, so set m.spinning in the new M.
  31. fn = mspinning
  32. }
  33. // 创建新的工作线程
  34. newm(fn, _p_)
  35. return
  36. }
  37. if mp.spinning {
  38. throw("startm: m is spinning")
  39. }
  40. if mp.nextp !=0{
  41. throw("startm: m has p")
  42. }
  43. if spinning &&!runqempty(_p_){
  44. throw("startm: p has runnable gs")
  45. }
  46. // The caller incremented nmspinning, so set m.spinning in the new M.
  47. mp.spinning = spinning
  48. // 设置 m 马上要结合的 p
  49. mp.nextp.set(_p_)
  50. // 唤醒 m
  51. notewakeup(&mp.park)
  52. }

首先处理 p 为空的情况,直接从全局空闲 p 队列里找,如果没找到,则直接返回。如果设置了 spinning 为 true 的话,还需要还原全局的处于自旋状态的 m 的数值:&sched.nmspinning

搞定了 p,接下来看 m。先调用 mget 函数从全局空闲的 m 队列里获取一个 m,如果没找到 m,则要调用 newm 新创建一个 m,并且如果设置了 spinning 为 true 的话,先要设置好 mstartfn:

  1. func mspinning(){
  2. // startm's caller incremented nmspinning. Set the new M's spinning.
  3. getg().m.spinning =true
  4. }

这样,启动 m 后,在 mstart1 函数里,进入 schedule 循环前,执行 mstartfn 函数,使得 m 处于自旋状态。

接下来是正常情况下(找到了 p 和 m)的处理:

  1. mp.spinning = spinning
  2. // 设置 m 马上要结合的 p
  3. mp.nextp.set(_p_)
  4. // 唤醒 m
  5. notewakeup(&mp.park)

设置 nextp 为找到的 p,调用 notewakeup 唤醒 m。之前我们讲 findrunnable 函数的时候,对于最后没有找到工作的 m,我们调用 notesleep(&_g_.m.park),使得 m 进入睡眠状态。现在终于有工作了,需要老将出山,将其唤醒:

  1. // src/runtime/lock_futex.go
  2. func notewakeup(n *note){
  3. // 设置 n.key = 1, 被唤醒的线程通过查看该值是否等于 1
  4. // 来确定是被其它线程唤醒还是意外从睡眠中苏醒
  5. old := atomic.Xchg(key32(&n.key),1)
  6. if old !=0{
  7. print("notewakeup - double wakeup (", old,")\n")
  8. throw("notewakeup - double wakeup")
  9. }
  10. futexwakeup(key32(&n.key),1)
  11. }

notewakeup 函数首先使用 atomic.Xchg 设置 note.key 值为 1,这是为了使被唤醒的线程可以通过查看该值是否等于 1 来确定是被其它线程唤醒还是意外从睡眠中苏醒了过来。

如果该值为 1 则表示是被唤醒的,可以继续工作,但如果该值为 0 则表示是意外苏醒,需要再次进入睡眠。

调用 futexwakeup 来唤醒工作线程,它和 futexsleep 是相对的。

  1. func futexwakeup(addr *uint32, cnt uint32){
  2. // 调用 futex 函数唤醒工作线程
  3. ret := futex(unsafe.Pointer(addr), _FUTEX_WAKE, cnt,nil,nil,0)
  4. if ret >=0{
  5. return
  6. }
  7. // ……………………
  8. }

futex 由汇编语言实现,前面已经分析过,这里就不重复了。主要内容就是先准备好参数,然后进行系统调用,由内核唤醒线程。

内核在完成唤醒工作之后当前工作线程从内核返回到 futex 函数继续执行 SYSCALL 指令之后的代码并按函数调用链原路返回,继续执行其它代码。

而被唤醒的工作线程则由内核负责在适当的时候调度到 CPU 上运行。

抢占长时间运行的 P

我们知道,Go scheduler 采用的是一种称为协作式的抢占式调度,就是说并不强制调度,大家保持协作关系,互相信任。对于长时间运行的 P,或者说绑定在 P 上的长时间运行的 goroutine,sysmon 会检测到这种情况,然后设置一些标志,表示 goroutine 自己让出 CPU 的执行权,给其他 goroutine 一些机会。

接下来我们就来分析当 P 处于 _Prunning 状态的情况。sysmon 扫描每个 p 时,都会记录下当前调度器调度的次数和当前时间,数据记录在结构体:

  1. type sysmontick struct{
  2. schedtick uint32
  3. schedwhen int64
  4. syscalltick uint32
  5. syscallwhen int64
  6. }

前面两个字段记录调度器调度的次数和时间,后面两个字段记录系统调用的次数和时间。

在下一次扫描时,对比 sysmon 记录下的 p 的调度次数和时间,与当前 p 自己记录下的调度次数和时间对比,如果一致。说明 P 在这一段时间内一直在运行同一个 goroutine。那就来计算一下运行时间是否太长了。

如果发现运行时间超过了 10 ms,则要调用 preemptone(_p_) 发起抢占的请求:

  1. func preemptone(_p_ *p)bool{
  2. mp := _p_.m.ptr()
  3. if mp ==nil|| mp == getg().m {
  4. returnfalse
  5. }
  6. // 被抢占的 goroutine
  7. gp := mp.curg
  8. if gp ==nil|| gp == mp.g0 {
  9. returnfalse
  10. }
  11. // 设置抢占标志
  12. gp.preempt =true
  13. // 在 goroutine 内部的每次调用都会比较栈顶指针和 g.stackguard0,
  14. // 来判断是否发生了栈溢出。stackPreempt 非常大的一个数,比任何栈都大
  15. // stackPreempt = 0xfffffade
  16. gp.stackguard0 = stackPreempt
  17. returntrue
  18. }

基本上只是将 stackguard0 设置了一个很大的值,而检查 stackguard0 的地方在函数调用前的一段汇编代码里进行。

举一个简单的例子:

  1. package main
  2. import"fmt"
  3. func main(){
  4. fmt.Println("hello qcrao.com!")
  5. }

执行命令:

  1. go tool compile -S main.go

得到汇编代码:

  1. "".main STEXT size=120 args=0x0 locals=0x48
  2. 0x000000000(test26.go:5) TEXT "".main(SB), $72-0
  3. 0x000000000(test26.go:5) MOVQ (TLS), CX
  4. 0x000900009(test26.go:5) CMPQ SP,16(CX)
  5. 0x000d00013(test26.go:5) JLS 113
  6. 0x000f00015(test26.go:5) SUBQ $72, SP
  7. 0x001300019(test26.go:5) MOVQ BP,64(SP)
  8. 0x001800024(test26.go:5) LEAQ 64(SP), BP
  9. 0x001d00029(test26.go:5) FUNCDATA $0, gclocals·69c1753bd5f81501d95132d08af04464(SB)
  10. 0x001d00029(test26.go:5) FUNCDATA $1, gclocals·e226d4ae4a7cad8835311c6a4683c14f(SB)
  11. 0x001d00029(test26.go:6) MOVQ $0,""..autotmp_0+48(SP)
  12. 0x002600038(test26.go:6) MOVQ $0,""..autotmp_0+56(SP)
  13. 0x002f00047(test26.go:6) LEAQ type.string(SB), AX
  14. 0x003600054(test26.go:6) MOVQ AX,""..autotmp_0+48(SP)
  15. 0x003b00059(test26.go:6) LEAQ "".statictmp_0(SB), AX
  16. 0x004200066(test26.go:6) MOVQ AX,""..autotmp_0+56(SP)
  17. 0x004700071(test26.go:6) LEAQ ""..autotmp_0+48(SP), AX
  18. 0x004c00076(test26.go:6) MOVQ AX,(SP)
  19. 0x005000080(test26.go:6) MOVQ $1,8(SP)
  20. 0x005900089(test26.go:6) MOVQ $1,16(SP)
  21. 0x006200098(test26.go:6) PCDATA $0, $1
  22. 0x006200098(test26.go:6) CALL fmt.Println(SB)
  23. 0x006700103(test26.go:7) MOVQ 64(SP), BP
  24. 0x006c00108(test26.go:7) ADDQ $72, SP
  25. 0x007000112(test26.go:7) RET
  26. 0x007100113(test26.go:7) NOP
  27. 0x007100113(test26.go:5) PCDATA $0, $-1
  28. 0x007100113(test26.go:5) CALL runtime.morestack_noctxt(SB)
  29. 0x007600118(test26.go:5) JMP 0

以前看这段代码的时候会直接跳过前面的几行代码,看不懂。这次能看懂了!所以,那些暂时看不懂的,先放一放,没关系,让子弹飞一会儿,很多东西回过头再来看就会豁然开朗,这就是一个很好的例子。

  1. 0x000000000(test26.go:5) MOVQ (TLS), CX

将本地存储 tls 保存到 CX 寄存器中,(TLS)表示它所关联的 g,这里就是前面所讲到的 main gouroutine。

  1. 0x000900009(test26.go:5) CMPQ SP,16(CX)

比较 SP 寄存器(代表当前 main goroutine 的栈顶寄存器)和 16(CX),我们看下 g 结构体:

  1. type g struct{
  2. // goroutine 使用的栈
  3. stack stack // offset known to runtime/cgo
  4. // 用于栈的扩张和收缩检查
  5. stackguard0 uintptr // offset known to liblink
  6. // ……………………
  7. }

对象 g 的第一个字段是 stack 结构体:

  1. type stack struct{
  2. lo uintptr
  3. hi uintptr
  4. }

共 16 字节。而 16(CX) 表示 g 对象的第 16 个字节,跳过了 g 的第一个字段,也就是 g.stackguard0 字段。

如果 SP 小于 g.stackguard0,这是必然的,因为前面已经把 g.stackguard0 设置成了一个非常大的值,因此跳转到了 113 行。

  1. 0x007100113(test26.go:7) NOP
  2. 0x007100113(test26.go:5) PCDATA $0, $-1
  3. 0x007100113(test26.go:5) CALL runtime.morestack_noctxt(SB)
  4. 0x007600118(test26.go:5) JMP 0

调用 runtime.morestack_noctxt 函数:

  1. // src/runtime/asm_amd64.s
  2. TEXT runtime·morestack_noctxt(SB),NOSPLIT,$0
  3. MOVL $0, DX
  4. JMP runtime·morestack(SB)

直接跳转到 morestack 函数:

  1. TEXT runtime·morestack(SB),NOSPLIT,$0-0
  2. // Cannot grow scheduler stack (m->g0).
  3. get_tls(CX)
  4. // BX = g,g 表示 main goroutine
  5. MOVQ g(CX), BX
  6. // BX = g.m
  7. MOVQ g_m(BX), BX
  8. // SI = g.m.g0
  9. MOVQ m_g0(BX), SI
  10. CMPQ g(CX), SI
  11. JNE 3(PC)
  12. CALL runtime·badmorestackg0(SB)
  13. INT $3
  14. // ……………………
  15. // Set g->sched to context in f.
  16. // 将函数的返回地址保存到 AX 寄存器
  17. MOVQ 0(SP), AX // f's PC
  18. // 将函数的返回地址保存到 g.sched.pc
  19. MOVQ AX,(g_sched+gobuf_pc)(SI)
  20. // g.sched.g = g
  21. MOVQ SI,(g_sched+gobuf_g)(SI)
  22. // 取地址操作符,调用 morestack_noctxt 之前的 rsp
  23. LEAQ 8(SP), AX // f's SP
  24. // 将 main 函数的栈顶地址保存到 g.sched.sp
  25. MOVQ AX,(g_sched+gobuf_sp)(SI)
  26. // 将 BP 寄存器保存到 g.sched.bp
  27. MOVQ BP,(g_sched+gobuf_bp)(SI)
  28. // newstack will fill gobuf.ctxt.
  29. // Call newstack on m->g0's stack.
  30. // BX = g.m.g0
  31. MOVQ m_g0(BX), BX
  32. // 将 g0 保存到本地存储 tls
  33. MOVQ BX, g(CX)
  34. // 把 g0 栈的栈顶寄存器的值恢复到 CPU 的寄存器 SP,达到切换栈的目的,下面这一条指令执行之前,
  35. // CPU 还是使用的调用此函数的 g 的栈,执行之后 CPU 就开始使用 g0 的栈了
  36. MOVQ (g_sched+gobuf_sp)(BX), SP
  37. // 准备参数
  38. PUSHQ DX // ctxt argument
  39. // 不返回
  40. CALL runtime·newstack(SB)
  41. MOVQ $0,0x1003// crash if newstack returns
  42. POPQ DX // keep balance check happy
  43. RET

主要做的工作就是将当前 goroutine,也就是 main goroutine 的和调度相关的信息保存到 g.sched 中,以便在调度到它执行时,可以恢复。

最后,将 g0 的地址保存到 tls 本地存储,并且切到 g0 栈执行之后的代码。继续调用 newstack 函数:

  1. func newstack(ctxt unsafe.Pointer){
  2. // thisg = g0
  3. thisg := getg()
  4. // ……………………
  5. // gp = main goroutine
  6. gp := thisg.m.curg
  7. // Write ctxt to gp.sched. We do this here instead of in
  8. // morestack so it has the necessary write barrier.
  9. gp.sched.ctxt = ctxt
  10. // ……………………
  11. morebuf := thisg.m.morebuf
  12. thisg.m.morebuf.pc =0
  13. thisg.m.morebuf.lr =0
  14. thisg.m.morebuf.sp =0
  15. thisg.m.morebuf.g =0
  16. // 检查 g.stackguard0 是否被设置成抢占标志
  17. preempt := atomic.Loaduintptr(&gp.stackguard0)== stackPreempt
  18. if preempt {
  19. if thisg.m.locks !=0|| thisg.m.mallocing !=0|| thisg.m.preemptoff !=""|| thisg.m.p.ptr().status !=_Prunning{
  20. // 还原 stackguard0 为正常值,表示我们已经处理过抢占请求了
  21. gp.stackguard0 = gp.stack.lo +_StackGuard
  22. // 不抢占,调用 gogo 继续运行当前这个 g,不需要调用 schedule 函数去挑选另一个 goroutine
  23. gogo(&gp.sched)// never return
  24. }
  25. }
  26. // ……………………
  27. if preempt {
  28. if gp == thisg.m.g0 {
  29. throw("runtime: preempt g0")
  30. }
  31. if thisg.m.p ==0&& thisg.m.locks ==0{
  32. throw("runtime: g is running but p is not")
  33. }
  34. // Synchronize with scang.
  35. casgstatus(gp,_Grunning,_Gwaiting)
  36. // ……………………
  37. // Act like goroutine called runtime.Gosched.
  38. // 修改为 running,调度起来运行
  39. casgstatus(gp,_Gwaiting,_Grunning)
  40. // 调用 gopreempt_m 把 gp 切换出去
  41. gopreempt_m(gp)// never return
  42. }
  43. // ……………………
  44. }

去掉了很多暂时还看不懂的地方,留到后面再研究。只关注有关抢占相关的。第一次判断 preempt 标志是 true 时,检查了 g 的状态,发现不能抢占,例如它所绑定的 P 的状态不是 _Prunning,那就恢复它的 stackguard0 字段,下次就不会走这一套流程了。然后,调用 gogo(&gp.sched) 继续执行当前的 goroutine。

中间又处理了很多判断流程,再次判断 preempt 标志是 true 时,调用 gopreempt_m(gp) 将 gp 切换出去。

  1. func gopreempt_m(gp *g){
  2. if trace.enabled {
  3. traceGoPreempt()
  4. }
  5. goschedImpl(gp)
  6. }

最终调用 goschedImpl 函数:

  1. func goschedImpl(gp *g){
  2. status := readgstatus(gp)
  3. if status&^_Gscan!=_Grunning{
  4. dumpgstatus(gp)
  5. throw("bad g status")
  6. }
  7. // 更改 gp 的状态
  8. casgstatus(gp,_Grunning,_Grunnable)
  9. // 解除 m 和 g 的关系
  10. dropg()
  11. lock(&sched.lock)
  12. // 将 gp 放入全局可运行队列
  13. globrunqput(gp)
  14. unlock(&sched.lock)
  15. // 进入新一轮的调度循环
  16. schedule()
  17. }

将 gp 的状态改为 _Grunnable,放入全局可运行队列,等待下次有 m 来全局队列找工作时才能继续运行,毕竟你已经运行这么长时间了,给别人一点机会嘛。

最后,调用 schedule() 函数进入新一轮的调度循环,会找出一个 goroutine 来运行,永不返回。

这样,关于 sysmon 线程在关于调度这块到底做了啥,我们已经回答完了。总结一下:

  1. 抢占处于系统调用的 P,让其他 m 接管它,以运行其他的 goroutine。
  2. 将运行时间过长的 goroutine 调度出去,给其他 goroutine 运行的机会。