Go并发调度进阶-GMP初始化，最难啃的有时候耐心看完还是很简单

2022-01-07 本文已影响0人 stackfuture

Go并发调度进阶-【公粽号：堆栈future】

2. GMP初始化

1. M的初始化

图片

M 只有自旋和非自旋两种状态。自旋的时候，会努力找工作；找不到的时候会进入非自旋状态，之后会休眠，直到有工作需要处理时，被其他工作线程唤醒，又进入自旋状态。

<pre data-tool="mdnice编辑器" style="box-sizing: border-box !important; margin: 10px 0px; padding: 0px; outline: 0px; font-family: SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace; font-size: 14px; overflow: auto; display: block; color: rgb(33, 37, 41); max-width: 100%; overflow-wrap: break-word !important; font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: 0.544px; orphans: 2; text-align: left; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; background-color: rgb(255, 255, 255); text-decoration-thickness: initial; text-decoration-style: initial; text-decoration-color: initial; border-radius: 5px; box-shadow: rgba(0, 0, 0, 0.55) 0px 2px 10px;">`// src/runtime/proc.go

func mcommoninit(mp *m, id int64) {
g := getg()
...
lock(&sched.lock)
...
//random初始化，用于窃取 G
mp.fastrand[0] = uint32(int64Hash(uint64(mp.id), fastrandseed))
mp.fastrand[1] = uint32(int64Hash(uint64(cputicks()), ^fastrandseed))
if mp.fastrand[0]|mp.fastrand[1] == 0 {
mp.fastrand[1] = 1
}

// 创建用于信号处理的gsignal，只是简单的从堆上分配一个g结构体对象,然后把栈设置好就返回了
mpreinit(mp)
if mp.gsignal != nil {
mp.gsignal.stackguard1 = mp.gsignal.stack.lo + _StackGuard
}

// 把 M 挂入全局链表allm之中
mp.alllink = allm
...
}` </pre>

这里传入的 id 是-1，初次调用会将 id 设置为 0，这里并未对M0做什么关于调度相关的初始化，所以可以简单的认为这个函数只是把M0放入全局链表allm之中就返回了。当然这个M0就是主M。

2. P的初始化

image.gif

通常情况下（在程序运行时不调整 P 的个数），P 只会在上图中的四种状态下进行切换。当程序刚开始运行进行初始化时，所有的 P 都处于 _Pgcstop 状态，随着 P 的初始化（runtime.procresize），会被置于 _Pidle。
当 M 需要运行时，会 runtime.acquirep 来使 P 变成 Prunning 状态，并通过 runtime.releasep 来释放。
当 G 执行时需要进入系统调用，P 会被设置为 _Psyscall，如果这个时候被系统监控抢夺（runtime.retake），则 P 会被重新修改为 _Pidle。
如果在程序运行中发生 GC，则 P 会被设置为 _Pgcstop，并在 runtime.startTheWorld 时重新调整为 _Prunning。

func procresize(nprocs int32) p {
// 获取先前的 P 个数
old := gomaxprocs
// 更新统计信息
now := nanotime()
if sched.procresizetime != 0 {
sched.totaltime += int64(old) * (now - sched.procresizetime)
}
sched.procresizetime = now
// 根据 runtime.MAXGOPROCS 调整 p 的数量,因为 runtime.MAXGOPROCS 用户可以自行设定
if nprocs > int32(len(allp)) {
lock(&allpLock)
if nprocs <= int32(cap(allp)) {
allp = allp[:nprocs]
} else {
nallp := make([]p, nprocs)
copy(nallp, allp[:cap(allp)])
allp = nallp
}
unlock(&allpLock)
}

// 初始化新的 P
for i := old; i < nprocs; i++ {
pp := allp[i]
// 为空,则申请新的 P 对象
if pp == nil {
pp = new(p)
}
pp.init(i)
atomicstorep(unsafe.Pointer(&allp[i]), unsafe.Pointer(pp))
}

g := getg()
// P 不为空,并且 id 小于 nprocs ,那么可以继续使用当前 P
if g.m.p != 0 && g.m.p.ptr().id < nprocs {
// continue to use the current P
g.m.p.ptr().status = _Prunning
g.m.p.ptr().mcache.prepareForSweep()
} else {
// 释放当前 P，因为已失效
if g.m.p != 0 {
g.m.p.ptr().m = 0
}
g.m.p = 0
p := allp[0]
p.m = 0
p.status = _Pidle
// P0 绑定到当前的 M0
acquirep(p)
}
// 从未使用的 P 释放资源
for i := nprocs; i < old; i++ {
p := allp[i]
p.destroy()
// 不能释放 p 本身，因为他可能在 m 进入系统调用时被引用
}
// 释放完 P 之后重置allp的长度
if int32(len(allp)) != nprocs {
lock(&allpLock)
allp = allp[:nprocs]
unlock(&allpLock)
}
var runnablePs *p
// 将没有本地任务的 P 放到空闲链表中
for i := nprocs - 1; i >= 0; i-- {
p := allp[i]
// 当前正在使用的 P 略过
if g.m.p.ptr() == p {
continue
}
// 设置状态为 _Pidle
p.status = _Pidle
// P 的任务列表是否为空
if runqempty(p) {
// 放入到空闲列表中
pidleput(p)
} else {
// 获取空闲 M 绑定到 P 上
p.m.set(mget())
//
p.link.set(runnablePs)
runnablePs = p
}
}
stealOrder.reset(uint32(nprocs))
var int32p int32 = &gomaxprocs // make compiler check that gomaxprocs is an int32
atomic.Store((uint32)(unsafe.Pointer(int32p)), uint32(nprocs))
return runnablePs
}` </pre>

procresize方法的执行过程如下：

allp 是全局变量 P 的资源池，如果 allp 的切片中的处理器数量少于期望数量，会对切片进行扩容；
扩容的时候会使用 new 申请一个新的 P ，然后使用 init 初始化，需要注意的是初始化的 P 的 id 就是传入的 i 的值，状态为 _Pgcstop；
然后通过 g.m.p 获取 M0，如果 M0 已与有效的 P 绑定上，则将被绑定的 P 的状态修改为 _Prunning。否则获取 allp[0] 作为 P0 调用 runtime.acquirep 与 M0 进行绑定；
超过处理器个数的 P 通过p.destroy释放资源，p.destroy会将与 P 相关的资源释放，并将 P 状态设置为 _Pdead；
通过截断改变全局变量 allp 的长度保证与期望处理器数量相等；
遍历 allp 检查 P 的是否处于空闲状态，是的话放入到空闲列表中；

3. G的初始化

image.gif

这是G的状态流转图，G的初始化相当复杂，需要大家下去对照源码再看一遍。

// 用 g0 系统栈创建 Goroutine 对象
// 传递的参数包括 fn 函数入口地址, argp 参数起始地址, siz 参数长度, gp（g0），调用方 pc（goroutine）
systemstack(func() {
newg := newproc1(fn, argp, siz, gp, pc)

p := getg().m.p.ptr() //获取p
runqput(p, newg, true) //并将其放入 P 本地队列的队头或全局队列

//检查空闲的 P，将其唤醒，准备执行 G，但我们目前处于初始化阶段，主 Goroutine 尚未开始执行，因此这里不会唤醒 P。
if mainStarted {
wakep()
}
})
}

// 创建一个运行 fn 的新 g，具有 narg 字节大小的参数，从 argp 开始。
// callerps 是 go 语句的起始地址。新创建的 g 会被放入 g 的队列中等待运行。
func newproc1(fn *funcval, argp unsafe.Pointer, narg int32, callergp *g, callerpc uintptr) *g {
g := getg()//因为是在系统栈运行所以此时的 g 为 g0
acquirem() //禁止抢占
if fn == nil {
g.m.throwing = -1 // do not dump full stacks
throw("go of nil func value")
}

...
siz := narg
siz = (siz + 7) &^ 7

...

// 当前工作线程所绑定的 p
// 初始化时 p = g0.m.p，也就是 p = allp[0]
p := g.m.p.ptr()
// 从 p 的本地缓冲里获取一个没有使用的 g，初始化时为空，返回 nil
newg := gfget(p)
if newg == nil {
// 创建一个拥有 _StackMin 大小的栈的 g
newg = malg(_StackMin)
// 将新创建的 g 从 _Gidle 更新为 _Gdead 状态
casgstatus(newg, _Gidle, _Gdead)
// 将 Gdead 状态的 g 添加到 allg，这样 GC 不会扫描未初始化的栈
allgadd(newg)
}
if newg.stack.hi == 0 {
throw("newproc1: newg missing stack")
}

if readgstatus(newg) != _Gdead {
throw("newproc1: new g is not Gdead")
}

...
// 确定 sp 位置
sp := newg.stack.hi - totalSize
// 确定参数入栈位置
spArg := sp

...

if narg > 0 {
// 将参数从执行 newproc 函数的栈拷贝到新 g 的栈
memmove(unsafe.Pointer(spArg), argp, uintptr(narg))
...
}

// 设置newg的调度相关信息
memclrNoHeapPointers(unsafe.Pointer(&newg.sched), unsafe.Sizeof(newg.sched))
newg.sched.sp = sp
newg.stktopsp = sp
newg.sched.pc = funcPC(goexit) + sys.PCQuantum
newg.sched.g = guintptr(unsafe.Pointer(newg))
gostartcallfn(&newg.sched, fn)
newg.gopc = callerpc
newg.ancestors = saveAncestors(callergp)
newg.startpc = fn.fn
if g.m.curg != nil {
newg.labels = g.m.curg.labels
}
if isSystemGoroutine(newg, false) {
atomic.Xadd(&sched.ngsys, +1)
}

// 设置 g 的状态为 _Grunnable，可以运行了
casgstatus(newg, _Gdead, _Grunnable)

// 设置goid
newg.goid = int64(p.goidcache)
p.goidcache++

...
releasem(g.m) //恢复抢占本质上是加锁
return newg
}` </pre>

创建 G 的过程也是相对比较复杂的，我们来总结一下这个过程：

首先尝试从 P 本地 gfree 链表或全局 gfree 队列获取已经执行过的 g
初始化过程中程序无论是本地队列还是全局队列都不可能获取到 g，因此创建一个新的 g，并为其分配运行线程（执行栈），这时 g 处于 _Gidle 状态
创建完成后，g 被更改为 _Gdead 状态，并根据要执行函数的入口地址和参数，初始化执行栈的 SP 和参数的入栈位置，并将需要的参数拷贝一份存入执行栈中
根据 SP、参数，在 g.sched 中保存 SP 和 PC 指针来初始化 g 的运行现场
将调用方、要执行的函数的入口 PC 进行保存，并将 g 的状态更改为 _Grunnable
给 Goroutine 分配 id，并将其放入 P 本地队列的队头或全局队列（初始化阶段队列肯定不是满的，因此不可能放入全局队列）
检查空闲的 P，将其唤醒，准备执行 G，但我们目前处于初始化阶段，主 Goroutine 尚未开始执行，因此这里不会唤醒 P。

4. 小结

结合上篇GMP结构和这篇的初始化，那么对于GMP的调度我相信会有一定的理解，起码你知道了底层的一些点点滴滴，后续文章还会给大家继续讲解GMP调度，你坚持学习到最后发现，最难啃的GMP也就那么回事！