go调度器的源码级分析

本文主要是介绍go调度器的源码级分析，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

 

基于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)
}

实际上定时器比这更加复杂，它也有复杂的状态机，但是和本文主题相差太远，就在此略过了。

这篇关于go调度器的源码级分析的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！