6.8 协作与抢占

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// Gosched 会让出当前的 P，并允许其他 Goroutine 运行。
// 它不会推迟当前的 Goroutine，因此执行会被自动恢复
func Gosched() {
	checkTimeouts()
	mcall(gosched_m)
}
// Gosched 在 g0 上继续执行
func gosched_m(gp *g) {
	...
	goschedImpl(gp)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

// checkTimeouts 恢复那些在等待一个 note 且已经触发其 deadline 时的 Goroutine。
func checkTimeouts() {
	now := nanotime()
	for n, nt := range notesWithTimeout {
		if n.key == note_cleared && now > nt.deadline {
			n.key = note_timeout
			goready(nt.gp, 1)
		}
	}
}
func goready(gp *g, traceskip int) {
	systemstack(func() {
		ready(gp, traceskip, true)
	})
}
// 将 gp 标记为 ready 来运行
func ready(gp *g, traceskip int, next bool) {
	if trace.enabled {
		traceGoUnpark(gp, traceskip)
	}

	status := readgstatus(gp)

	// 标记为 runnable.
	_g_ := getg()
	_g_.m.locks++ // 禁止抢占，因为它可以在局部变量中保存 p
	if status&^_Gscan != _Gwaiting {
		dumpgstatus(gp)
		throw("bad g->status in ready")
	}

	// 状态为 Gwaiting 或 Gscanwaiting, 标记 Grunnable 并将其放入运行队列 runq
	casgstatus(gp, _Gwaiting, _Grunnable)
	runqput(_g_.m.p.ptr(), gp, next)
	if atomic.Load(&sched.npidle) != 0 && atomic.Load(&sched.nmspinning) == 0 {
		wakep()
	}
	_g_.m.locks--
	if _g_.m.locks == 0 && _g_.preempt { // 在 newstack 中已经清除它的情况下恢复抢占请求
		_g_.stackguard0 = stackPreempt
	}
}

func notetsleepg(n *note, ns int64) bool {
	gp := getg()
	...

	if ns >= 0 {
		deadline := nanotime() + ns
		...
		notesWithTimeout[n] = noteWithTimeout{gp: gp, deadline: deadline}
		...
		gopark(nil, nil, waitReasonSleep, traceEvNone, 1)
		...
		delete(notesWithTimeout, n)
		...
	}

	...
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

func goschedImpl(gp *g) {
	// 放弃当前 g 的运行状态
	status := readgstatus(gp)
	...
	casgstatus(gp, _Grunning, _Grunnable)
	// 使当前 m 放弃 g
	dropg()
	// 并将 g 放回全局队列中
	lock(&sched.lock)
	globrunqput(gp)
	unlock(&sched.lock)

	// 重新进入调度循环
	schedule()
}

1
2
3
4
5

// Goroutine 抢占请求
// 存储到 g.stackguard0 来导致栈分段检查失败
// 必须比任何实际的 SP 都要大
// 十六进制为：0xfffffade
const stackPreempt = (1<<(8*sys.PtrSize) - 1) & -1314

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

TEXT runtime·morestack(SB),NOSPLIT,$0-0
	...
	MOVQ	0(SP), AX // f's PC
	MOVQ	AX, (g_sched+gobuf_pc)(SI)
	MOVQ	SI, (g_sched+gobuf_g)(SI)
	LEAQ	8(SP), AX // f's SP
	MOVQ	AX, (g_sched+gobuf_sp)(SI)
	MOVQ	BP, (g_sched+gobuf_bp)(SI)
	MOVQ	DX, (g_sched+gobuf_ctxt)(SI)
	...
	CALL	runtime·newstack(SB)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

//go:nowritebarrierrec
func newstack() {
	thisg := getg()
	...

	gp := thisg.m.curg
	...

	morebuf := thisg.m.morebuf
	thisg.m.morebuf.pc = 0
	thisg.m.morebuf.lr = 0
	thisg.m.morebuf.sp = 0
	thisg.m.morebuf.g = 0

	// 如果是发起的抢占请求而非真正的栈分段
	preempt := atomic.Loaduintptr(&gp.stackguard0) == stackPreempt

	// 保守的对用户态代码进行抢占，而非抢占运行时代码
	// 如果正持有锁、分配内存或抢占被禁用，则不发生抢占
	if preempt {
		if !canPreemptM(thisg.m) {
			// 不发生抢占，继续调度
			gp.stackguard0 = gp.stack.lo + _StackGuard
			gogo(&gp.sched) // 重新进入调度循环
		}
	}
	...
	// 如果需要对栈进行调整
	if preempt {
		...
		if gp.preemptShrink {
			// 我们正在一个同步安全点，因此等待栈收缩
			gp.preemptShrink = false
			shrinkstack(gp)
		}
		if gp.preemptStop {
			preemptPark(gp) // 永不返回
		}
		...
		// 表现得像是调用了 runtime.Gosched，主动让权
		gopreempt_m(gp) // 重新进入调度循环
	}
	...
}
// 与 gosched_m 一致
func gopreempt_m(gp *g) {
	...
	goschedImpl(gp)
}

1
2
3
4
5

// canPreemptM 报告 mp 是否处于可抢占的安全状态。
//go:nosplit
func canPreemptM(mp *m) bool {
	return mp.locks == 0 && mp.mallocing == 0 && mp.preemptoff == "" && mp.p.ptr().status == _Prunning
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// 此程序在 Go 1.14 之前的版本不会输出 OK
package main
import (
	"runtime"
	"time"
)
func main() {
	runtime.GOMAXPROCS(1)
	go func() {
		for {
		}
	}()
	time.Sleep(time.Millisecond)
	println("OK")
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

func retake(now int64) uint32 {
	n := 0
	// 防止 allp 数组发生变化，除非我们已经 STW，此锁将完全没有人竞争
	lock(&allpLock)
	for i := 0; i < len(allp); i++ {
		_p_ := allp[i]
		...
		pd := &_p_.sysmontick
		s := _p_.status
		sysretake := false
		if s == _Prunning || s == _Psyscall {
			// 如果 G 运行时时间太长则进行抢占
			t := int64(_p_.schedtick)
			if int64(pd.schedtick) != t {
				pd.schedtick = uint32(t)
				pd.schedwhen = now
			} else if pd.schedwhen+forcePreemptNS <= now {
				...
				sysretake = true
			}
		}
		// 对阻塞在系统调用上的 P 进行抢占
		if s == _Psyscall {
			// 如果已经超过了一个系统监控的 tick（20us），则从系统调用中抢占 P
			t := int64(_p_.syscalltick)
			if !sysretake && int64(pd.syscalltick) != t {
				pd.syscalltick = uint32(t)
				pd.syscallwhen = now
				continue
			}
			// 一方面，在没有其他 work 的情况下，我们不希望抢夺 P
			// 另一方面，因为它可能阻止 sysmon 线程从深度睡眠中唤醒，所以最终我们仍希望抢夺 P
			if runqempty(_p_) && atomic.Load(&sched.nmspinning)+atomic.Load(&sched.npidle) > 0 && pd.syscallwhen+10*1000*1000 > now {
				continue
			}
			// 解除 allpLock，从而可以获取 sched.lock
			unlock(&allpLock)
			// 在 CAS 之前需要减少空闲 M 的数量（假装某个还在运行）
			// 否则发生抢夺的 M 可能退出 syscall 然后再增加 nmidle ，进而发生死锁
			// 这个过程发生在 stoplockedm 中
			incidlelocked(-1)
			if atomic.Cas(&_p_.status, s, _Pidle) { // 将 P 设为 idle，从而交于其他 M 使用
				...
				n++
				_p_.syscalltick++
				handoffp(_p_)
			}
			incidlelocked(1)
			lock(&allpLock)
		}
	}
	unlock(&allpLock)
	return uint32(n)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

func retake(now int64) uint32 {
	...
	for i := 0; i < len(allp); i++ {
		_p_ := allp[i]
		...
		if s == _Prunning || s == _Psyscall {
			...
			} else if pd.schedwhen+forcePreemptNS <= now {
				// 对于 syscall 的情况，因为 M 没有与 P 绑定，
				// preemptone() 不工作
				preemptone(_p_)
				sysretake = true
			}
		}
		...
	}
	...
}
func preemptone(_p_ *p) bool {
	// 检查 M 与 P 是否绑定
	mp := _p_.m.ptr()
	if mp == nil || mp == getg().m {
		return false
	}
	gp := mp.curg
	if gp == nil || gp == mp.g0 {
		return false
	}

	// 将 G 标记为抢占
	gp.preempt = true

	// 一个 Goroutine 中的每个调用都会通过比较当前栈指针和 gp.stackgard0
	// 来检查栈是否溢出。
	// 设置 gp.stackgard0 为 StackPreempt 来将抢占转换为正常的栈溢出检查。
	gp.stackguard0 = stackPreempt

	// 请求该 P 的异步抢占
	if preemptMSupported && debug.asyncpreemptoff == 0 {
		_p_.preempt = true
		preemptM(mp)
	}

	return true
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

const sigPreempt = _SIGURG

// preemptM 向 mp 发送抢占请求。该请求可以异步处理，也可以与对 M 的其他请求合并。
// 接收到该请求后，如果正在运行的 G 或 P 被标记为抢占，并且 Goroutine 处于异步安全点，
// 它将抢占 Goroutine。在处理抢占请求后，它始终以原子方式递增 mp.preemptGen。
func preemptM(mp *m) {
	...
	signalM(mp, sigPreempt)
}
func signalM(mp *m, sig int) {
	tgkill(getpid(), int(mp.procid), sig)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

//go:nowritebarrierrec
func sighandler(sig uint32, info *siginfo, ctxt unsafe.Pointer, gp *g) {
	...
	c := &sigctxt{info, ctxt}
	...
	if sig == sigPreempt {
		// 可能是一个抢占信号
		doSigPreempt(gp, c)
		// 即便这是一个抢占信号，它也可能与其他信号进行混合，因此我们
		// 继续进行处理。
	}
	...
}
// doSigPreempt 处理了 gp 上的抢占信号
func doSigPreempt(gp *g, ctxt *sigctxt) {
	// 检查 G 是否需要被抢占、抢占是否安全
	if wantAsyncPreempt(gp) && isAsyncSafePoint(gp, ctxt.sigpc(), ctxt.sigsp(), ctxt.siglr()) {
		// 插入抢占调用
		ctxt.pushCall(funcPC(asyncPreempt))
	}

	// 记录抢占
	atomic.Xadd(&gp.m.preemptGen, 1)

1
2
3
4
5
6
7
8

func (c *sigctxt) pushCall(targetPC uintptr) {
	pc := uintptr(c.rip())
	sp := uintptr(c.rsp())
	sp -= sys.PtrSize
	*(*uintptr)(unsafe.Pointer(sp)) = pc
	c.set_rsp(uint64(sp))
	c.set_rip(uint64(targetPC))
}

1
2
3
4

// asyncPreempt 保存了所有用户寄存器，并调用 asyncPreempt2
//
// 当栈扫描遭遇 asyncPreempt 栈帧时，将会保守的扫描调用方栈帧
func asyncPreempt()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

TEXT ·asyncPreempt(SB),NOSPLIT|NOFRAME,$0-0
	...
	MOVQ AX, 0(SP)
	...
	MOVUPS X15, 352(SP)
	CALL ·asyncPreempt2(SB)
	MOVUPS 352(SP), X15
	...
	MOVQ 0(SP), AX
	...
	RET

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

//go:nosplit
func asyncPreempt2() {
	gp := getg()
	gp.asyncSafePoint = true
	if gp.preemptStop {
		mcall(preemptPark)
	} else {
		mcall(gopreempt_m)
	}
	// 异步抢占过程结束
	gp.asyncSafePoint = false
}