基于GMP模型的调度器是go实现其引以为傲的用户态线程的核心。本文就以GMP调度器为核心分析一下调度流程，顺便分析一下定时器Timer的实现，它和调度器息息相关。
本文的大纲如下：
1.GMP的关键数据结构
2.goroutine的生命周期
3.系统线程的生命周期
4.触发shedule()的时机
5.网络轮询netpoll
6.sysmon守护线程
7.定时器实现
1.GMP的关键数据结构
调度器相关的数据结构都在runtime/runtime2.go中，主要有g(goroutine，协程)，m(machine，操作系统线程)，p(processor，调度器的资源抽象，m争抢p来获取执行权)，schedt(scheduler_type，调度器的全局抽象)
以下都是只给出一些关键字段，实际上每个完整数据结构都有几十个字段！
1.1 g
type g struct {
stack stack
_panic *_panic
_defer *_defer
m *m
sched gobuf
atomicstatus uint32
}
stack: 协程栈的上下界，由lo和hi两个字段组成
_panic: 协程的panic链表，头部的是最内层的panic(也就是外层的panic可能会被abort)
_defer: 协程的defer链表，头部的是最内层defer。不过本文不关心defer和panic，只是想说明defer和panic都是goroutine级别的
m: 当前g对应的m
sched: 记录了协程的运行时状态，内部包含sp、pc、bp等伪寄存器
atomicstatus: 协程的状态机，共有近十种状态，如idle,dead,runnable,running,syscall,waiting,preempted等
1.2 m
type m struct {
g0 *g
gsignal *g
curg *g
p puintptr
nextp puintptr
oldp puintptr
spinning bool
}
g0: 运行在系统栈上的协程，即负责调度协程的协程
gsignal: 负责响应信号的协程。由上可知一个空闲的m也自带两个g
curg: m当前运行的g，可能为空表示空闲
p: m当前绑定的p。要运行用户代码，必须绑定上一个p。一般p远少于m。
nextp: m刚刚创建或从阻塞转为就绪时，预先绑定的p。真正运行时，就调用acquirep绑定这个p
oldp: m陷入系统调用时会出让p，这里保存这个p，系统调用返回时试图恢复这个p。这是希望利用局部性。
spinning: 是否自旋。如果当前的m是自旋的，而又从本地队列runq和全局队列globalrunq都获取不到可运行的g时，就会从其他p的本地队列尝试窃取g
1.3 p
type p struct {
status uint32
m muintptr
deferpool [5][]*_defer
runq [256]guintptr
gFree struct {
gList
n int32
}
sudogcache []*sudog
timer0When uint64
timers []*timer
}
status: p的状态机
m: p的m。与m的p双向绑定
deferpool: _defer结构体的per-P缓存池。在deferproc创建新_defer时(本文不分析，详见runtime/panic.go)会优先从中分配内存，按内存大小分为五个链表。这也是利用局部性
runq: 当前p的可运行g队列
gFree: 当前p的空闲g队列(即状态为dead的g。g执行完毕后不会销毁，而是等待被复用)
sudogcache: 创建sudog的缓存池，作用同deferpool。sudog代表一个阻塞状态的g，其记录了g所在的阻塞队列等信息，本文不介绍。
timers,timer0When: timers保存了当前p全部定时器，timer0When保存了所有定时器中最近的即将到达的时间。本文会介绍定时器的实现
1.4 schedt
type schedt struct {
lastpoll uint64
midle muintptr
nmidle int32
nmidlelocked int32
maxmcount int32
nmsys int32
nmfreed int64
pidle puintptr
npidle uint32
nmspinning uint32
runq gQueue
gFree struct {
lock mutex
stack gList
noStack gList
n int32
}
sudogcache *sudog
deferpool [5]*_defer
}
可见，schedt的这些字段部分是状态统计信息，一些(最后4个)是p中的某些结构的全局版本。
以及第一个lastpoll是上一次进行网络轮询(netpoll)的时间。在讲定时器时会说到
2. goroutine的生命周期
对于用户来说，GMP调度器中距离自己最近的当然是g。那么当我们轻轻松松地敲下go来启动一个g时，golang为我们做了什么呢？
在编译阶段，编译器会把go关键字转换成一个runtime.newproc调用。newproc有两个参数，初始栈大小和go出来的函数指针，这两个都是在编译时就能确定的。接下来详细看一下newproc
注意！本文的代码都是高度简化版，源代码太复杂了，作者太强了，顶礼膜拜！
func newproc(siz int32, fn *funcval) {
argp := add(unsafe.Pointer(&fn), sys.PtrSize)
gp := getg()
pc := getcallerpc()
systemstack(func() {
newg := newproc1(fn, argp, siz, gp, pc) //初始化一个g结构体
_p_ := getg().m.p.ptr()
runqput(_p_, newg, true) //将g加入当前p的本地队列。当前p也就是使用go关键字的那个g所在的m绑定的p，有点绕..
if mainStarted {
wakep() //尝试启动一个空闲的p，这个函数之后会说到
}
})
}
先看newproc1具体怎么初始化g
func newproc1(fn *funcval, argp unsafe.Pointer, narg int32, callergp *g, callerpc uintptr) *g {
_g_ := getg()
siz := narg
siz = (siz + 7) &^ 7
_p_ := _g_.m.p.ptr()
newg := gfget(_p_) //gfget(p)从p的空闲列表中获取一个g。如果p的空闲列表为空，从全局空闲列表中将最多32个g放入p的空闲列表并返回。
if newg == nil {
newg = malg(_StackMin) //memory_alloc_g，新创建一个g，这个g只初始化了栈空间的相关信息
casgstatus(newg, _Gidle, _Gdead) //状态转移。dead表示分配了栈，但未在执行代码，这名字有点迷惑性，刚初始化完和执行结束都是dead
allgadd(newg)
}
totalSize := 4*sys.PtrSize + uintptr(siz) + sys.MinFrameSize
totalSize += -totalSize & (sys.StackAlign - 1)
sp := newg.stack.hi - totalSize
spArg := sp
//接下来就是初始化sched字段等运行状态信息了
memclrNoHeapPointers(unsafe.Pointer(&newg.sched), unsafe.Sizeof(newg.sched))
newg.sched.sp = sp
newg.stktopsp = sp
newg.sched.pc = abi.FuncPCABI0(goexit) + sys.PCQuantum //*1
newg.sched.g = guintptr(unsafe.Pointer(newg))
gostartcallfn(&newg.sched, fn)
newg.gopc = callerpc
newg.ancestors = saveAncestors(callergp)
newg.startpc = fn.fn
casgstatus(newg, _Gdead, _Grunnable)
return newg
}
*1 所有新创建的g的pc都从goexit这个ABI(Application Binary Interface)函数所在的位置开始。这样做等效于：当这个g执行完毕时，就会返回到goexit，执行goexit函数。goexit在本节的最后讲
现在g已经加入p的本地队列了，接下来就是等待被调度了！触发调度的方式有多种，之后会讲到。这边先说明一下，调度的主函数是schedule()函数。它的作用是进行一轮调度：尝试获取一个g，并执行它
func schedule() {
_g_ := getg()
top:
pp := _g_.m.p.ptr()
checkTimers(pp, 0) //检查最近的定时器有没有到达，如果到了就运行它。具体逻辑更加复杂，后面会说
var gp *g
var inheritTime bool
if gp == nil {
if _g_.m.p.ptr().schedtick%61 == 0 && sched.runqsize > 0 { //约有1/61的调度会从全局队列中拿，这是为了减少饥饿
lock(&sched.lock)
gp = globrunqget(_g_.m.p.ptr(), 1)
unlock(&sched.lock)
}
}
if gp == nil {
gp, inheritTime = runqget(_g_.m.p.ptr()) //从p的本地队列中拿
}
if gp == nil {
gp, inheritTime = findrunnable() // 一个非常复杂的方法，它是阻塞的，直到获取到一个可运行的g
}
execute(gp, inheritTime) //执行goroutine的真正入口
}
findrunnable是一个非常复杂的方法，它阻塞直到获取到一个可运行的g。期间还会顺便做一下网络轮询(之后会说)。工作窃取(work-stealing)也是在这个函数里完成的
func findrunnable() (gp *g, inheritTime bool) {
_g_ := getg()
top:
_p_ := _g_.m.p.ptr()
now, pollUntil, _ := checkTimers(_p_, 0) //检查定时器。返回值pollUntil表示下一个定时器到达的时间，后续步骤有用到
if gp, inheritTime := runqget(_p_); gp != nil { //从p的本地队列中取
return gp, inheritTime
}
if sched.runqsize != 0 {
lock(&sched.lock)
gp := globrunqget(_p_, 0) //从全局队列中取
unlock(&sched.lock)
if gp != nil {
return gp, false
}
}
//轮询网络(后面会说)！netpoll(0)表示进行非阻塞轮询，返回一组就绪的g，返回第一个，调用injectglist将剩下的加入p的本地队列
if netpollinited() && atomic.Load(&netpollWaiters) > 0 && atomic.Load64(&sched.lastpoll) != 0 {
if list := netpoll(0); !list.empty() { // non-blocking
gp := list.pop()
injectglist(&list) //将一组g加入当前p的本地队列，如果当前没有绑定p，就加入全局队列
casgstatus(gp, _Gwaiting, _Grunnable) //状态转移：阻塞态->就绪态
return gp, false
}
}
//如果当前的m处于自旋状态(或者忙碌的p数量比自旋的m数量更多)，就尝试从其它p的本地队列中窃取g！
procs := uint32(gomaxprocs)
if _g_.m.spinning || 2*atomic.Load(&sched.nmspinning) < procs-atomic.Load(&sched.npidle) {
if !_g_.m.spinning {
_g_.m.spinning = true
atomic.Xadd(&sched.nmspinning, 1)
}
gp, inheritTime, tnow, w, newWork := stealWork(now)
now = tnow
if gp != nil {
return gp, inheritTime
}
if newWork {
goto top
}
if w != 0 && (pollUntil == 0 || w < pollUntil) {
pollUntil = w
}
}
lock(&sched.lock)
if sched.gcwaiting != 0 || _p_.runSafePointFn != 0 {
unlock(&sched.lock)
goto top
}
if sched.runqsize != 0 {
gp := globrunqget(_p_, 0)
unlock(&sched.lock)
return gp, false
}
//实在无事可做了，就把当前的p加入schedt的空闲p列表
pidleput(_p_)
unlock(&sched.lock)
// ***关键！阻塞直到有可用的g
if netpollinited() && (atomic.Load(&netpollWaiters) > 0 || pollUntil != 0) && atomic.Xchg64(&sched.lastpoll, 0) != 0 {
atomic.Store64(&sched.pollUntil, uint64(pollUntil))
delay := int64(-1)
if pollUntil != 0 {
if now == 0 {
now = nanotime()
}
delay = pollUntil - now
if delay < 0 {
delay = 0
}
}
list := netpoll(delay) //阻塞至多delay时间。如果没有定时器，则无限阻塞，直到被netpollBreak唤起。返回就绪的一组g
atomic.Store64(&sched.pollUntil, 0)
atomic.Store64(&sched.lastpoll, uint64(nanotime()))
_p_ = pidleget() //获取空闲p
if _p_ == nil {
injectglist(&list) //获取失败，将所有g注入（本地或全局）队列
} else {
acquirep(_p_) //获取成功，将p和m绑定，将第一个以后的g注入（本地或全局）队列，第一个g被返回
if !list.empty() {
gp := list.pop()
injectglist(&list)
casgstatus(gp, _Gwaiting, _Grunnable)
return gp, false
}
goto top
}
} else if pollUntil != 0 && netpollinited() {
pollerPollUntil := int64(atomic.Load64(&sched.pollUntil))
if pollerPollUntil == 0 || pollerPollUntil > pollUntil {
netpollBreak()
}
}
stopm()
goto top
}
上面标有***代码是关键，完成了阻塞获取g的逻辑。netpoll和定时器部分后面会讲到
execute是goroutine的真正入口，它的核心逻辑非常简单，就是调用gogo函数，定位到g.sched中保存的运行位置
func execute(gp *g, inheritTime bool) {
_g_ := getg()
_g_.m.curg = gp
gp.m = _g_.m //把当前的m(实际上是g0的m)和要执行的g绑定起来
casgstatus(gp, _Grunnable, _Grunning) //状态转移：就绪态->运行态
gp.waitsince = 0
gp.preempt = false
gp.stackguard0 = gp.stack.lo + _StackGuard
gogo(&gp.sched) //gogo!
}
gogo是一个汇编(go assembly，和普通汇编略有不同)实现的函数，用g.sched里保存的pc、sp等寄存器信息进行跳转
TEXT runtime·gogo(SB), NOSPLIT, $0-8
MOVQ buf+0(FP), BX // 读取g.sched参数放入BX
MOVQ gobuf_g(BX), DX
MOVQ 0(DX), CX
JMP gogo<>(SB)
TEXT gogo<>(SB), NOSPLIT, $0
get_tls(CX)
MOVQ DX, g(CX)
MOVQ DX, R14
MOVQ gobuf_sp(BX), SP
MOVQ gobuf_ret(BX), AX
MOVQ gobuf_ctxt(BX), DX
MOVQ gobuf_bp(BX), BP
MOVQ gobuf_pc(BX), BX
JMP BX //就是在这里跳转到了g.sched里保存的pc位置！开始正式执行用户写的func
现在终于开始执行用户代码了！但是这个g依然不能高枕无忧了！有多种方式可以让这个g失去执行权，即让g.m与p解绑，之后会讨论。现在先假设g顺利执行完了，那就会按之前说的，调用goexit
goexit也是个汇编函数，但逻辑比较简单，总之最后调用到goexit0函数
func goexit0(gp *g) {
_g_ := getg()
casgstatus(gp, _Grunning, _Gdead) //状态转移
gp.m = nil //将当前g的一些状态清空
locked := gp.lockedm != 0
gp.lockedm = 0
_g_.m.lockedg = 0
gp.preemptStop = false
gp.paniconfault = false
gp._defer = nil
gp._panic = nil
gp.writebuf = nil
gp.waitreason = 0
gp.param = nil
gp.labels = nil
gp.timer = nil
dropg() //里面完成了m.curg与当前g的解绑
gfput(_g_.m.p.ptr(), gp) //把当前g加入全局的空闲队列schedt.gFree
schedule() //关键！
}
可以看出，g退出时会先进行一些清理工作，最终会再次调用schedule()，开始新一轮调度！这就是调度器能持续不断地运行的原因
此外，执行完毕的g不会被销毁，而是加入了全局的空闲队列。还记得这个空闲队列哪里用到吗？newproc创建g的时候！以此实现g的复用
3. 系统线程的生命周期
系统线程(m)是g的运行载体，一个m只能在同一时刻运行一个g。m为了获取代码执行权必须绑定一个p。如果m不在执行用户代码（如进行系统调用），则和p解绑。
启动一个m的入口是startm函数。
func startm(_p_ *p, spinning bool) {
mp := acquirem()
lock(&sched.lock)
if _p_ == nil { //startm的第一个参数决定了该函数的语义：是让m被动等待可用的p，还是让p主动把所有权移交给m
_p_ = pidleget() //从空闲p列表中获取p
if _p_ == nil { //如果没有，就返回，启动m失败！因为m的运行依附于p
unlock(&sched.lock)
releasem(mp)
return
}
}
nmp := mget() //从空闲m列表中获取m
if nmp == nil {
id := mReserveID()
unlock(&sched.lock)
var fn func()
if spinning {
fn = mspinning
}
newm(fn, _p_, id) //如果获取不到，就新建一个m
releasem(mp)
return
}
unlock(&sched.lock)
nmp.spinning = spinning
nmp.nextp.set(_p_) //把新建出来的m的nextp设为当前p，等m真正运行时，就会绑定这个p
notewakeup(&nmp.park)
releasem(mp)
}
以下总结了调用startm的时机：
1. 有新的g可用时。一般是调用newproc和ready(标记一个g为就绪)函数时。前面提到newproc会调用wakep，其中就调的startm，第一个参数为nil，也就是不指定p，而是让m等待空闲的p
2. p移交所有权时。典型的情况就是sysmon守护线程（后面会讲）发现一个p处于syscall状态的时间太长了，就会调用handoffp函数将它的所有权移交，具体在后面讲sysmon的时候会说。其中也会调startm，第一个参数就是这个p本身，表示让m和这个p绑定，也就是p的所有者被移交(handoff)了。
startm中提到，如果在空闲m列表中获取不到m，就会调用newm创建一个。
func newm(fn func(), _p_ *p, id int64) {
mp := allocm(_p_, fn, id)
mp.doesPark = (_p_ != nil)
mp.nextp.set(_p_)
mp.sigmask = initSigmask
newm1(mp)
}
allocm分配了一个m的空间，进行一些初始化，但还没有和任何系统线程绑定
func allocm(_p_ *p, fn func(), id int64) *m {
_g_ := getg()
acquirem()
if _g_.m.p == 0 {
acquirep(_p_)
}
mp := new(m)
mp.mstartfn = fn
mcommoninit(mp, id)
mp.g0 = malg(-1) //为这个m创建g0
mp.g0.m = mp
if _p_ == _g_.m.p.ptr() {
releasep()
}
releasem(_g_.m)
return mp
}
newm1将m真正和系统线程绑定
func newm1(mp *m) {
newosproc(mp) //其实这个函数不止这么点，但核心就这一句，新建操作系统线程，并与m绑定
}
在linux环境下，newosproc会使用clone系统调用创建新线程
func newosproc(mp *m) {
//也是省了一堆无关代码
ret := clone(cloneFlags, stk, unsafe.Pointer(mp), unsafe.Pointer(mp.g0), unsafe.Pointer(funcPC(mstart)))
}
clone的第四个参数指定线程的入口，是mstart函数的地址，于是目光转向mstart函数，它是真正创建系统线程的入口
mstart先是进行一些栈的初始化，最终进入mstart1函数
func mstart0() {
_g_ := getg()
//初始化一大堆栈参数
mstart1()
mexit(osStack)
}
func mstart1() {
_g_ := getg()
//保存当前g(因为是新创建的m，所以只能是g0)的sched状态
_g_.sched.g = guintptr(unsafe.Pointer(_g_))
_g_.sched.pc = getcallerpc()
_g_.sched.sp = getcallersp()
asminit()
minit()
if _g_.m == &m0 {
mstartm0() //在这里面初始化gsignal信号处理协程
}
if fn := _g_.m.mstartfn; fn != nil {
fn()
}
if _g_.m != &m0 {
acquirep(_g_.m.nextp.ptr()) //acquirep(p)将p和当前m绑定，也就是这个m终于获得了代码的执行权了！还记得nextp吗，是在newm函数里设置的
_g_.m.nextp = 0
}
schedule()
}
mstart1将之前设置的nextp与m绑定，使m终于获得了代码执行权！并且立即进入下一轮调度schedule()
注意，这个函数一开始还在g0.sched中保存了运行状态。别忘了schedule()会跳转到下一个g的pc而不会返回，所以，为了能让mstart1返回，必须有类似gogo(&_g_.m.g0.sched)的调用，跳转到mstart1的调用者，才能返回。返回后，进入mexit函数销毁线程。
mexit函数主要就是调用handoffp函数，让即将销毁的m的p转交所有权。然后调用一些系统调用来摧毁线程，没什么好说的。
那么什么时候会有gogo(&_g_.m.g0.sched)的调用呢？我看了半天只找到一处，就是在某个g退出时(调用goexit)，会检查这个g有没有锁定（锁定，一般指的是调用公有函数runtime.LockOsThread，使当前g独占某个操作系统线程）过当前m，有的话就会调用这个函数。换句话说，如果g锁定了所在的m，那么就认为这个m不干净了，等g结束后就会跳转到g0.sched从而销毁m。这个g真是坑爹啊！正常情况下，m是不会被销毁的。
最后，在startm中提到会现在空闲m列表中获取m，获取不到再进入newm。这个空闲m列表，其实就是保存了休眠状态下的m。触发垃圾回收时当前m会休眠。此外，当g从系统调用中返回时，g.m会寻找可用的p以继续执行代码，如果没有可用的p，也只能进入休眠了。
4. 触发schedule()的时机
之前看到，在启动g或m的时候会立即开始一轮调度。除此之外，触发调度还有一些时机
4.1 g让出执行权
主要是运行态的g主动成为阻塞态或就绪态。
成为阻塞态就是调用gopark函数挂起自身。gopark函数最终调用park_m函数
func park_m(gp *g) { //gp是当前的g
casgstatus(gp, _Grunning, _Gwaiting)
dropg() //将g.m.curg和g解绑
schedule()
}
成为就绪态就是调用公有函数runtime.GoSched()出让当前g的执行权，有点像java里的Thread.yield()。GoSched最终调用goschedImpl函数
func goschedImpl(gp *g) {
status := readgstatus(gp)
casgstatus(gp, _Grunning, _Grunnable)
dropg()
globrunqput(gp) //将g放入全局队列
schedule()
}
两者最终都调用schedule()开启新一轮调度
4.2 系统调用
在进行系统调用时，编译器会在这个系统调用前后插入reentersyscall和exitsyscall两个钩子函数，在这两个函数里进行GMP状态的保存和恢复
func reentersyscall(pc, sp uintptr) {
_g_ := getg()
save(pc, sp) //保存寄存器状态
_g_.syscallsp = sp
_g_.syscallpc = pc
casgstatus(_g_, _Grunning, _Gsyscall) //g的状态转移：运行态->系统调用中
_g_.m.syscalltick = _g_.m.p.ptr().syscalltick
_g_.sysblocktraced = true
pp := _g_.m.p.ptr()
pp.m = 0 //将m和p解绑
_g_.m.oldp.set(pp) //在m.oldp中保存p，为了在系统调用返回时尝试重新获得p
_g_.m.p = 0
atomic.Store(&pp.status, _Psyscall) //p的状态转移: 变为系统调用中。或许是p的状态比g简单，所以不用cas？
}
func exitsyscall() {
_g_ := getg()
oldp := _g_.m.oldp.ptr()
_g_.m.oldp = 0
if exitsyscallfast(oldp) { //尝试重新获取oldp
if trace.enabled {
if oldp != _g_.m.p.ptr() || _g_.m.syscalltick != _g_.m.p.ptr().syscalltick {
systemstack(traceGoStart)
}
}
_g_.m.p.ptr().syscalltick++
casgstatus(_g_, _Gsyscall, _Grunning)
_g_.syscallsp = 0
_g_.m.locks--
_g_.throwsplit = false
if sched.disable.user && !schedEnabled(_g_) {
Gosched()
}
return
}
mcall(exitsyscall0) //尝试失败，走完整流程
}
reentersyscall中在解绑m与p时，会将p保存到m.oldp中，在exitsyscall中尝试恢复p，这样做是希望利用局部性。
func exitsyscallfast(oldp *p) bool {
_g_ := getg()
//尝试绑定原来的p，前提是p没有移情别恋，例如被sysmon监控到p处于syscall状态太久而移交所有权
if oldp != nil && oldp.status == _Psyscall && atomic.Cas(&oldp.status, _Psyscall, _Pidle) {
wirep(oldp) //将oldp与m绑定
exitsyscallfast_reacquired()
return true
}
if sched.pidle != 0 { //如果能找到空闲的p，也能接受，至少m不用休眠了！
var ok bool
systemstack(func() {
ok = exitsyscallfast_pidle() //里面会把空闲的p与m绑定
})
if ok {
return true
}
}
return false
}
func exitsyscall0(gp *g) { //快速绑定失败，走完整流程
casgstatus(gp, _Gsyscall, _Grunnable)
dropg()
var _p_ *p
if schedEnabled(gp) {
_p_ = pidleget() //获取空闲p
}
var locked bool
if _p_ == nil {
globrunqput(gp) //获取失败，将g加入全局队列
stopm() //将m加入空闲m列表
}
if _p_ != nil { //获取成功，将p与m绑定，开始执行g
acquirep(_p_)
execute(gp, false)
}
schedule()
}
exitsyscall0是慢路径，因为它由mcall调用，会造成协程切换(g切换到g0)。如果最终获取不到p，就使g加入全局队列，m休眠，开启新一轮调度。
4.3 p被sysmon线程发起抢占
这个留在后面和sysmon一起说
5. 网络轮询netpoll
netpoll是go对多路复用的实现。在linux系统上，主要依赖epoll系列系统调用。
在调度过程中，主要是在finerunnable函数中进行netpoll, 返回一组就绪的g
netpoll接口需要实现netpollinit, netpoll, netpollBreak, netpollopen, netpollclose, netpollisPollDescriptor六个函数。与本文主题相关的是前三个
func netpollinit() {
epfd = epollcreate1(1024)
r, w, errno := nonblockingPipe()
ev := epollevent{
events: _EPOLLIN,
}
*(**uintptr)(unsafe.Pointer(&ev.data)) = &netpollBreakRd
errno = epollctl(epfd, _EPOLL_CTL_ADD, r, &ev)
netpollBreakRd = uintptr(r)
netpollBreakWr = uintptr(w)
}
看到这几个函数我不禁留下了感动的眼泪，太熟悉了，就是epoll的套皮。
epollinit先调用epoll_create，再创建epollevent结构体，再创建一个管道进行读方向的监听
func netpoll(delay int64) gList {
if epfd == -1 {
return gList{}
}
var waitms int32
if delay < 0 { //参数<0表示无限阻塞
waitms = -1
} else if delay == 0 { //参数==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)
if n < 0 {
goto retry
}
var toRun gList
for i := int32(0); i < n; i++ {
ev := &events[i]
if ev.events == 0 {
continue
}
if mode != 0 {
pd := *(**pollDesc)(unsafe.Pointer(&ev.data))
netpollready(&toRun, pd, mode)
}
}
return toRun
}
netpoll根据传入的参数不同有阻塞、非阻塞、超时阻塞三种模式。调用epoll_wait返回就绪事件，并用netpollready工具函数把这些事件转换成g链表后返回
func netpollBreak() {
if atomic.Cas(&netpollWakeSig, 0, 1) {
for {
var b byte
n := write(netpollBreakWr, unsafe.Pointer(&b), 1)
if n == 1 {
break
}
}
}
}
netpollBreak向管道中写入内容，导致管道的读端事件被触发，这可以中止netpoll的阻塞状态。
所以，现在可以回顾以下findrunnable中所做的关于网络轮询的事情：
1.调用checkTimers运行已经到达的定时器，如果没有定时器到达，就返回一个pollUntil表示下一个定时器的到达时间。如果没有下一个定时器则返回0
2.调用非阻塞的netpoll先尝试获取一波g，这些g可能是之前执行了read或connect等调用而进入阻塞
3.所有能做的事都做完了之后就进行阻塞的netpoll调用。如果pollUntil不为0，则进行带超时的netpoll调用，使得届时能从netpoll中及时返回以响应定时器到达事件
6. sysmon守护线程
sysmon(system monitor)守护线程有点像linux守护进程或者mysql后台线程，总之都是打杂的。sysmon和其他守护线程一样，也是在一个大循环里进行轮询监控，并对一些异常事件进行处理。
sysmon随着全局g0的启动而启动。
func main() {
...省略
if GOARCH != "wasm" {
atomic.Store(&sched.sysmonStarting, 1)
systemstack(func() {
newm(sysmon, nil, -1)
})
}
...省略
}
全局g0在main函数中通过newm创建了一个不依赖p的m(第二个参数为nil)，并把sysmon函数作为m的底层线程的入口，即创建了sysmon守护线程。
func sysmon() {
...省略
idle := 0 // 记录sysmon已经连续多少个循环没被唤醒，也没唤醒别人了，即一片祥和
delay := uint32(0) //大循环的间隔
for {
if idle == 0 { // 至少间隔20微秒
delay = 20
} else if idle > 50 { // 如果连续没被唤醒的次数过多，就可以多睡一会了
delay *= 2
}
if delay > 10*1000 { // 最多间隔不超过10毫秒
delay = 10 * 1000
}
now := nanotime()
if debug.schedtrace <= 0 && (sched.gcwaiting != 0 || atomic.Load(&sched.npidle) == uint32(gomaxprocs)) {
lock(&sched.lock)
if atomic.Load(&sched.gcwaiting) != 0 || atomic.Load(&sched.npidle) == uint32(gomaxprocs) {
syscallWake := false
next, _ := timeSleepUntil() //返回下一个定时器的到达时间
//如果sysmon没什么活干(无须追踪调度、不在等待gc或p全部空闲)，定时器也还没到，就可以休息了
if next > now {
atomic.Store(&sched.sysmonwait, 1)
unlock(&sched.lock)
// Make wake-up period small enough
// for the sampling to be correct.
sleep := forcegcperiod / 2
if next-now < sleep {
sleep = next - now
}
shouldRelax := sleep >= osRelaxMinNS
if shouldRelax {
osRelax(true)
}
syscallWake = notetsleep(&sched.sysmonnote, sleep) //休息了。进行系统调用之类的会唤醒sysmon，那就会把idle清零，delay重置
mDoFixup()
if shouldRelax {
osRelax(false)
}
lock(&sched.lock)
atomic.Store(&sched.sysmonwait, 0)
noteclear(&sched.sysmonnote)
}
if syscallWake {
idle = 0
delay = 20
}
}
unlock(&sched.lock)
}
}
...省略
}
sysmon也会想偷懒，没什么活干就休息了，详见上面的注释
sysmon()函数很长，但功能划分得很清楚，主要就四个部分：检查死锁、网络轮询兜底、对p发起抢占、垃圾收集兜底
6.1 检查死锁
核心逻辑是checkdead函数。去除了cgo相关的逻辑，还是蛮简单的
func checkdead() {
run := mcount() - sched.nmidle - sched.nmidlelocked - sched.nmsys //运行中的m
if run > 0 {
return
}
grunning := 0 //若没有m在运行，统计不在运行的g数量。这变量名很奇怪，grunning统计的是不在运行的g
forEachG(func(gp *g) {
if isSystemGoroutine(gp, false) {
return
}
s := readgstatus(gp)
switch s &^ _Gscan {
case _Gwaiting,
_Gpreempted:
grunning++
case _Grunnable,
_Grunning,
_Gsyscall:
throw("checkdead: runnable g")
}
})
if grunning == 0 { //如果不在运行的g数量为0，那压根就没g了，说明所有g都退出了，这是由于主协程调用了runtime.Goexit引起的
throw("no goroutines (main called runtime.Goexit) - deadlock!")
}
// 如果不在运行的g的数量大于0，即有阻塞态的g，就检查有没有定时器，如果有的话，所有g阻塞也是合理的
for _, _p_ := range allp {
if len(_p_.timers) > 0 {
return
}
}
throw("all goroutines are asleep - deadlock!") //如果也没有定时器，那就是死锁了
}
6.2 网络轮询兜底
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)
if !list.empty() {
incidlelocked(-1)
injectglist(&list)
incidlelocked(1)
}
}
和findrunnable里的逻辑几乎一样。作用是每10ms会进行一次netpoll兜底，将就绪的g注入队列
6.3 对p发起抢占
核心逻辑是retake函数。
func retake(now int64) uint32 {
n := 0
// Prevent allp slice changes. This lock will be completely
// uncontended unless we're already stopping the world.
lock(&allpLock)
// We can't use a range loop over allp because we may
// temporarily drop the allpLock. Hence, we need to re-fetch
// allp each time around the loop.
for i := 0; i < len(allp); i++ {
_p_ := allp[i]
pd := &_p_.sysmontick
s := _p_.status
sysretake := false
if s == _Prunning || s == _Psyscall {
// 如果p的运行时间太长（超过10ms），调用preemptone将其抢占！
t := int64(_p_.schedtick)
if int64(pd.schedtick) != t {
pd.schedtick = uint32(t)
pd.schedwhen = now
} else if pd.schedwhen+forcePreemptNS <= now {
preemptone(_p_)
sysretake = true
}
}
if s == _Psyscall {
// p处于系统调用状态时，
t := int64(_p_.syscalltick)
if !sysretake && int64(pd.syscalltick) != t {
pd.syscalltick = uint32(t)
pd.syscallwhen = now
continue
}
//若p的本地队列不为空，或者不存在自旋的m与空闲的p，或者距离系统调用发起已经超过10ms了，就调用handoffp移交p的所有权！也就是尽量不让p闲着
if runqempty(_p_) && atomic.Load(&sched.nmspinning)+atomic.Load(&sched.npidle) > 0 && pd.syscallwhen+10*1000*1000 > now {
continue
}
// Drop allpLock so we can take sched.lock.
unlock(&allpLock)
incidlelocked(-1)
if atomic.Cas(&_p_.status, s, _Pidle) {
if trace.enabled {
traceGoSysBlock(_p_)
traceProcStop(_p_)
}
n++
_p_.syscalltick++
handoffp(_p_)
}
incidlelocked(1)
lock(&allpLock)
}
}
unlock(&allpLock)
return uint32(n)
}
preemptone(p)主要是依赖preemptM系统调用，让当前的g停止在p上运行
handoffp(p)通过调用startm函数来启动一个m与p绑定。handoffp秉持着尽量启动m的原则，如果实在太闲了，则不会启动m，而是将p加入空闲列表
如果retake没能抢占任何p，则sysmon的idle自增，表明又白跑一轮。
6.4 垃圾回收兜底
这不是本文的内容，简单带过一下就是每120秒或触发兜底的ix强制垃圾回收，没什么好说的
7. 定时器实现
最后讲一下定时器的实现，在前面的小节里也出现过很多次了。相比GMP，定时器模块就比较简单了
定时器的数据结构是timer，主要字段如下
type timer struct {
pp puintptr
when int64
period int64
f func(interface{}, uintptr)
arg interface{}
seq uintptr
}
pp: 指向定时器所在的p。回忆一下p结构体里也有一个[]*timer保存了所有定时器
when: 下次触发时间
period: 周期定时器(Ticker)的周期
f,arg: 定时器到达的回调函数及其参数
seq: 触发序号
p结构体里的timers其实不是一个简单的切片，而是一个小顶四叉堆（4-ary heap，二叉堆的扩展，也是用数组表示树形结构），堆顶元素就是when最小的，也就是最快到达的定时器。
不信的话看看这段源码：
func siftupTimer(t []*timer, i int) int {
if i >= len(t) {
badTimer()
}
when := t[i].when
if when <= 0 {
badTimer()
}
tmp := t[i]
for i > 0 {
p := (i - 1) / 4
if when >= t[p].when {
break
}
t[i] = t[p]
i = p
}
if tmp != t[i] {
t[i] = tmp
}
return i
}
一行没改！这就是一个经典的堆实现，只不过是四叉的，取父亲下标要除以4。这是向上筛选的函数，向下筛选的函数类似，不展示了
调用addtimer新增定时器，最终进入doaddtimer函数
func doaddtimer(pp *p, t *timer) {
if netpollInited == 0 { //定时器依赖netpoll模块(之前展示过)，所以没初始化要先初始化
netpollGenericInit()
}
t.pp.set(pp) //pp就是当前的p
i := len(pp.timers)
pp.timers = append(pp.timers, t)
siftupTimer(pp.timers, i) //往p.timers堆中插入一个定时器
if t == pp.timers[0] {
atomic.Store64(&pp.timer0When, uint64(t.when))
}
atomic.Xadd(&pp.numTimers, 1)
}
定时器的触发被放在调度器的调度周期里进行，触发定时器的入口是checkTimers(在schedule()函数内，之前展示过)
如果g的计算量较大，执行时间很长，或者进行频繁大量的系统调用，schedule函数的主动触发频率就会降低，不得不依靠sysmon线程的抢占式调度。在最坏情况下，调度周期可达到10ms，这也就是定时器的最高保证精度。
checkTimers在成功获取到已到达的定时器后，最终会走进runOneTimer
func runOneTimer(pp *p, t *timer, now int64) {
f := t.f
arg := t.arg
seq := t.seq
if t.period > 0 { //如果是周期定时器，计算下一轮触发事件并调整小顶堆
delta := t.when - now
t.when += t.period * (1 + -delta/t.period)
if t.when < 0 { // check for overflow.
t.when = maxWhen
}
siftdownTimer(pp.timers, 0)
if !atomic.Cas(&t.status, timerRunning, timerWaiting) {
badTimer()
}
updateTimer0When(pp)
} else { //否则这个定时器没用了，删除
dodeltimer0(pp)
if !atomic.Cas(&t.status, timerRunning, timerNoStatus) {
badTimer()
}
}
unlock(&pp.timersLock)
f(arg, seq) //触发回调函数
lock(&pp.timersLock)
}
实际上定时器比这更加复杂，它也有复杂的状态机，但是和本文主题相差太远，就在此略过了。