g0 栈何用户栈如何切换
上一讲讲完了 main goroutine 的诞生,它不是第一个,算上 g0,它要算第二个了。不过,我们要考虑的就是这个 goroutine,它会真正执行用户代码。
g0 栈用于执行调度器的代码,执行完之后,要跳转到执行用户代码的地方,如何跳转?这中间涉及到栈和寄存器的切换。要知道,函数调用和返回主要靠的也是 CPU 寄存器的切换。goroutine 的切换和此类似。
继续看 proc1 函数的代码。中间有一段调整运行空间的代码,计算出的结果一般为 0,也就是一般不会调整 SP 的位置,忽略好了。
1
// 确定参数入栈位置
2
spArg := sp
Copied!
参数的入参位置也是从 SP 处开始,通过:
1
// 将参数从执行 newproc 函数的栈拷贝到新 g 的栈
2
memmove(unsafe.Pointer(spArg), unsafe.Pointer(argp), uintptr(narg))
Copied!
将 fn 的参数从 g0 栈上拷贝到 newg 的栈上,memmove 函数需要传入源地址、目的地址、参数大小。由于 main 函数在这里没有参数需要拷贝,因此这里相当于没做什么。
接着,初始化 newg 的各种字段,而且涉及到最重要的 pc,sp 等字段:
1
// 把 newg.sched 结构体成员的所有成员设置为 0
2
memclrNoHeapPointers(unsafe.Pointer(&newg.sched), unsafe.Sizeof(newg.sched))
3
// 设置 newg 的 sched 成员,调度器需要依靠这些字段才能把 goroutine 调度到 CPU 上运行
4
newg.sched.sp = sp
5
newg.stktopsp = sp
6
// newg.sched.pc 表示当 newg 被调度起来运行时从这个地址开始执行指令
7
newg.sched.pc = funcPC(goexit) + sys.PCQuantum // +PCQuantum so that previous instruction is in same function
8
newg.sched.g = guintptr(unsafe.Pointer(newg))
9
gostartcallfn(&newg.sched, fn)
10
newg.gopc = callerpc
11
// 设置 newg 的 startpc 为 fn.fn,该成员主要用于函数调用栈的 traceback 和栈收缩
12
// newg 真正从哪里开始执行并不依赖于这个成员,而是 sched.pc
13
newg.startpc = fn.fn
14
if _g_.m.curg != nil {
15
newg.labels = _g_.m.curg.labels
16
}
Copied!
首先,memclrNoHeapPointers 将 newg.sched 的内存全部清零。接着,设置 sched 的 sp 字段,当 goroutine 被调度到 m 上运行时,需要通过 sp 字段来指示栈顶的位置,这里设置的就是新栈的栈顶位置。
最关键的一行来了:
1
// newg.sched.pc 表示当 newg 被调度起来运行时从这个地址开始执行指令
2
newg.sched.pc = funcPC(goexit) + sys.PCQuantum // +PCQuantum so that previous instruction is in same function
Copied!
设置 pc 字段为函数 goexit 的地址加 1,也说是 goexit 函数的第二条指令,goexit 函数是 goroutine 退出后的一些清理工作。有点奇怪,这是要干嘛?接着往后看。
1
newg.sched.g = guintptr(unsafe.Pointer(newg))
Copied!
设置 g 字段为 newg 的地址。插一句,sched 是 g 结构体的一个字段,它本身也是一个结构体,保存调度信息。复习一下:
1
type gobuf struct {
2
// 存储 rsp 寄存器的值
3
sp uintptr
4
// 存储 rip 寄存器的值
5
pc uintptr
6
// 指向 goroutine
7
g guintptr
8
ctxt unsafe.Pointer // this has to be a pointer so that gc scans it
9
// 保存系统调用的返回值
10
ret sys.Uintreg
11
lr uintptr
12
bp uintptr // for GOEXPERIMENT=framepointer
13
}
Copied!
接下来的这个函数非常重要,可以解释之前为什么要那样设置 pc 字段的值。调用 gostartcallfn
1
gostartcallfn(&newg.sched, fn) //调整sched成员和newg的栈
Copied!
传入 newg.sched 和 fn。
1
func gostartcallfn(gobuf *gobuf, fv *funcval) {
2
var fn unsafe.Pointer
3
if fv != nil {
4
// fn: gorotine 的入口地址,初始化时对应的是 runtime.main
5
fn = unsafe.Pointer(fv.fn)
6
} else {
7
fn = unsafe.Pointer(funcPC(nilfunc))
8
}
9
gostartcall(gobuf, fn, unsafe.Pointer(fv))
10
}
11
12
func gostartcall(buf *gobuf, fn, ctxt unsafe.Pointer) {
13
// newg 的栈顶,目前 newg 栈上只有 fn 函数的参数,sp 指向的是 fn 的第一参数
14
sp := buf.sp
15
16
// …………………………
17
18
// 为返回地址预留空间
19
sp -= sys.PtrSize
20
// 这里填的是 newproc1 函数里设置的 goexit 函数的第二条指令
21
// 伪装 fn 是被 goexit 函数调用的,使得 fn 执行完后返回到 goexit 继续执行,从而完成清理工作
22
*(*uintptr)(unsafe.Pointer(sp)) = buf.pc
23
// 重新设置 buf.sp
24
buf.sp = sp
25
// 当 goroutine 被调度起来执行时,会从这里的 pc 值开始执行,初始化时就是 runtime.main
26
buf.pc = uintptr(fn)
27
buf.ctxt = ctxt
28
}
Copied!
函数 gostartcallfn 只是拆解出了包含在 funcval 结构体里的函数指针,转过头就调用 gostartcall。将 sp 减小了一个指针的位置,这是给返回地址留空间。果然接着就把 buf.pc 填入了栈顶的位置:
1
*(*uintptr)(unsafe.Pointer(sp)) = buf.pc
Copied!
原来 buf.pc 只是做了一个搬运工,搞什么啊。重新设置 buf.sp 为送减掉一个指针位置之后的值,设置 buf.pc 为 fn,指向要执行的函数,这里就是指的 runtime.main 函数。
对嘛,这才是应有的操作。之后,当调度器“光顾”此 goroutine 时,取出 buf.sp 和 buf.pc,恢复 CPU 相应的寄存器,就可以构造出 goroutine 的运行环境。
而 goexit 函数也通过“偷天换日”将自己的地址“强行”放到 newg 的栈顶,达到自己不可告人的目的:每个 goroutine 执行完之后,都要经过我的一些清理工作,才能“放行”。这样一说,goexit 函数还真是无私,默默地做一些“扫尾”的工作。
设置完 newg.sched 这后,我们的图又可以前进一步:
设置 newg.sched
上图中,newg 新增了 sched.pc 指向 runtime.main 函数,当它被调度起来执行时,就从这里开始;新增了 sched.sp 指向了 newg 栈顶位置,同时,newg 栈顶位置的内容是一个跳转地址,指向 runtime.goexit 的第二条指令,当 goroutine 退出时,这条地址会载入 CPU 的 PC 寄存器,跳转到这里执行“扫尾”工作。
之后,将 newg 的状态改为 runnable,设置 goroutine 的 id:
1
// 设置 g 的状态为 _Grunnable,可以运行了
2
casgstatus(newg, _Gdead, _Grunnable)
3
newg.goid = int64(_p_.goidcache)
Copied!
每个 P 每次会批量(16个)申请 id,每次调用 newproc 函数,新创建一个 goroutine,id 加 1。因此 g0 的 id 是 0,而 main goroutine 的 id 就是 1。
newg 的状态变成可执行后(Runnable),就可以将它加入到 P 的本地运行队列里,等待调度。所以,goroutine 何时被执行,用户代码决定不了。来看源码:
1
// 将 G 放入 _p_ 的本地待运行队列
2
runqput(_p_, newg, true)
3
4
// runqput 尝试将 g 放到本地可执行队列里。
5
// 如果 next 为假,runqput 将 g 添加到可运行队列的尾部
6
// 如果 next 为真,runqput 将 g 添加到 p.runnext 字段
7
// 如果 run queue 满了,runnext 将 g 放到全局队列里
8
//
9
// runnext 成员中的 goroutine 会被优先调度起来运行
10
func runqput(_p_ *p, gp *g, next bool) {
11
// ……………………
12
13
if next {
14
retryNext:
15
oldnext := _p_.runnext
16
if !_p_.runnext.cas(oldnext, guintptr(unsafe.Pointer(gp))) {
17
// 有其它线程在操作 runnext 成员,需要重试
18
goto retryNext
19
}
20
// 老的 runnext 为 nil,不用管了
21
if oldnext == 0 {
22
return
23
}
24
// 把之前的 runnext 踢到正常的 runq 中
25
// 原本存放在 runnext 的 gp 放入 runq 的尾部
26
gp = oldnext.ptr()
27
}
28
29
retry:
30
h := atomic.Load(&_p_.runqhead) // load-acquire, synchronize with consumers
31
t := _p_.runqtail
32
// 如果 P 的本地队列没有满,入队
33
if t-h < uint32(len(_p_.runq)) {
34
_p_.runq[t%uint32(len(_p_.runq))].set(gp)
35
// 原子写入
36
atomic.Store(&_p_.runqtail, t+1) // store-release, makes the item available for consumption
37
return
38
}
39
// 可运行队列已经满了,放入全局队列了
40
if runqputslow(_p_, gp, h, t) {
41
return
42
}
43
// the queue is not full, now the put above must succeed
44
// 没有成功放入全局队列,说明本地队列没满,重试一下
45
goto retry
46
}
Copied!
runqput 函数的主要作用就是将新创建的 goroutine 加入到 P 的可运行队列,如果本地队列满了,则加入到全局可运行队列。前两个参数都好理解,最后一个参数 next 的作用是,当它为 true 时,会将 newg 加入到 P 的 runnext 字段,具有最高优先级,将先于普通队列中的 goroutine 得到执行。
先将 P 老的 runnext 成员取出,接着用一个原子操作 cas 来试图将 runnext 成员设置成 newg,目的是防止其他线程在同时修改 runnext 字段。
设置成功之后,相当于 newg “挤掉” 了原来老的处于 runnext 的 goroutine,还得给人遣散费,安顿好人家嘛,不然和强盗有何区别?
“安顿”的动作在 retry 代码段中执行。先通过 headtaillen(_p_.runq) 来判断队列是否已满,如果没满,则直接写到队列尾部,同时修改队列尾部的指针。
1
// store-release, makes it available for consumption
2
atomic.Store(&_p_.runqtail, t+1)
Copied!
这里使用原子操作写入 runtail,防止编译器和 CPU 指令重排,保证上一行代码对 runq 的修改发生在修改 runqtail 之前,并且保证当前线程对队列的修改对其它线程立即可见。
如果本地队列满了,那就只能试图将 newg 添加到全局可运行队列中了。调用 runqputslow(_p_, gp, h, t) 完成。
1
// 将 g 和 _p_ 本地队列的一半 goroutine 放入全局队列。
2
// 因为要获取锁,所以会慢
3
func runqputslow(_p_ *p, gp *g, h, t uint32) bool {
4
var batch [len(_p_.runq)/2 + 1]*g
5
6
// First, grab a batch from local queue.
7
n := t - h
8
n = n / 2
9
if n != uint32(len(_p_.runq)/2) {
10
throw("runqputslow: queue is not full")
11
}
12
for i := uint32(0); i < n; i++ {
13
batch[i] = _p_.runq[(h+i)%uint32(len(_p_.runq))].ptr()
14
}
15
// 如果 cas 操作失败,说明本地队列不满了,直接返回
16
if !atomic.Cas(&_p_.runqhead, h, h+n) { // cas-release, commits consume
17
return false
18
}
19
batch[n] = gp
20
21
// …………………………
22
23
// Link the goroutines.
24
// 全局运行队列是一个链表,这里首先把所有需要放入全局运行队列的 g 链接起来,
25
// 减小锁粒度,从而降低锁冲突,提升性能
26
for i := uint32(0); i < n; i++ {
27
batch[i].schedlink.set(batch[i+1])
28
}
29
30
// Now put the batch on global queue.
31
lock(&sched.lock)
32
globrunqputbatch(batch[0], batch[n], int32(n+1))
33
unlock(&sched.lock)
34
return true
35
}
Copied!
先将 P 本地队列里所有的 goroutine 加入到一个数组中,数组长度为 len(_p_.runq)/2 + 1,也就是 runq 的一半加上 newg。
接着,将从 runq 的头部开始的前一半 goroutine 存入 bacth 数组。然后,使用原子操作尝试修改 P 的队列头,因为出队了一半 goroutine,所以 head 要向后移动 1/2 的长度。如果修改失败,说明 runq 的本地队列被其他线程修改了,因此后面的操作就不进行了,直接返回 false,表示 newg 没被添加进来。
1
batch[n] = gp
Copied!
将 newg 本身添加到数组。
通过循环将 batch 数组里的所有 g 串成链表:
1
for i := uint32(0); i < n; i++ {
2
batch[i].schedlink.set(batch[i+1])
3
}
Copied!
批量 goroutine 连接成链表
最后,将链表添加到全局队列中。由于操作的是全局队列,因此需要获取锁,因为存在竞争,所以代价较高。这也是本地可运行队列存在的原因。调用 globrunqputbatch(batch[0], batch[n], int32(n+1))
1
// Put a batch of runnable goroutines on the global runnable queue.
2
// Sched must be locked.
3
func globrunqputbatch(ghead *g, gtail *g, n int32) {
4
gtail.schedlink = 0
5
if sched.runqtail != 0 {
6
sched.runqtail.ptr().schedlink.set(ghead)
7
} else {
8
sched.runqhead.set(ghead)
9
}
10
sched.runqtail.set(gtail)
11
sched.runqsize += n
12
}
Copied!
如果全局的队列尾 sched.runqtail 不为空,则直接将其和前面生成的链表头相接,否则说明全局的可运行列队为空,那就直接将前面生成的链表头设置到 sched.runqhead。
最后,再设置好队列尾,增加 runqsize。
设置完成之后:
放到全局可运行队列
再回到 runqput 函数,如果将 newg 添加到全局队列失败了,说明本地队列在此过程中发生了变化,又有了位置可以添加 newg,因此重试 retry 代码段。我们也可以发现,P 的本地可运行队列的长度为 256,它是一个循环队列,因此最多只能放下 256 个 goroutine。
因为本文还是处于初始化的场景,所以 newg 被成功放入 p0 的本地可运行队列,等待被调度。
将我们的图再完善一下:
newg 添加到本地 runq

参考资料

【阿波张 Go语言调度器之调度 main 】https://mp.weixin.qq.com/s/8eJm5hjwKXya85VnT4y8Cw
Copy link
Contents
参考资料