从源码剖析Go语言基于信号抢占式调度

package main

import (
	"fmt"
	"os"
	"runtime"
	"runtime/trace"
	"sync"
)

func main() {
	runtime.GOMAXPROCS(1)
	f, _ := os.Create("trace.output")
	defer f.Close()
	_ = trace.Start(f)
	defer trace.Stop()
	var wg sync.WaitGroup
	for i := 0; i < 30; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			t := 0
			for i:=0;i<1e8;i++ {
				t+=2
			}
			fmt.Println("total:", t)
		}()
	}
	wg.Wait()
}

$ go build -gcflags "-N -l" main.go 
-N表示禁用优化
-l禁用内联

$ ./main 

$ go tool trace -http=":6060" ./trace.output

func initsig(preinit bool) {
	// 预初始化
	if !preinit { 
		signalsOK = true
	} 
	//遍历信号数组
	for i := uint32(0); i < _NSIG; i++ {
		t := &sigtable[i]
		//略过信号：SIGKILL、SIGSTOP、SIGTSTP、SIGCONT、SIGTTIN、SIGTTOU
		if t.flags == 0 || t.flags&_SigDefault != 0 {
			continue
		} 
		...  
		setsig(i, funcPC(sighandler))
	}
}
 

var sigtable = [...]sigTabT{
	/* 0 */ {0, "SIGNONE: no trap"},
	/* 1 */ {_SigNotify + _SigKill, "SIGHUP: terminal line hangup"},
	/* 2 */ {_SigNotify + _SigKill, "SIGINT: interrupt"},
	/* 3 */ {_SigNotify + _SigThrow, "SIGQUIT: quit"},
	/* 4 */ {_SigThrow + _SigUnblock, "SIGILL: illegal instruction"},
	/* 5 */ {_SigThrow + _SigUnblock, "SIGTRAP: trace trap"},
	/* 6 */ {_SigNotify + _SigThrow, "SIGABRT: abort"},
	/* 7 */ {_SigPanic + _SigUnblock, "SIGBUS: bus error"},
	/* 8 */ {_SigPanic + _SigUnblock, "SIGFPE: floating-point exception"},
	/* 9 */ {0, "SIGKILL: kill"},
	/* 10 */ {_SigNotify, "SIGUSR1: user-defined signal 1"},
	/* 11 */ {_SigPanic + _SigUnblock, "SIGSEGV: segmentation violation"},
	/* 12 */ {_SigNotify, "SIGUSR2: user-defined signal 2"},
	/* 13 */ {_SigNotify, "SIGPIPE: write to broken pipe"},
	/* 14 */ {_SigNotify, "SIGALRM: alarm clock"},
	/* 15 */ {_SigNotify + _SigKill, "SIGTERM: termination"},
	/* 16 */ {_SigThrow + _SigUnblock, "SIGSTKFLT: stack fault"},
	/* 17 */ {_SigNotify + _SigUnblock + _SigIgn, "SIGCHLD: child status has changed"},
	/* 18 */ {_SigNotify + _SigDefault + _SigIgn, "SIGCONT: continue"},
	/* 19 */ {0, "SIGSTOP: stop, unblockable"},
	/* 20 */ {_SigNotify + _SigDefault + _SigIgn, "SIGTSTP: keyboard stop"},
	/* 21 */ {_SigNotify + _SigDefault + _SigIgn, "SIGTTIN: background read from tty"},
	/* 22 */ {_SigNotify + _SigDefault + _SigIgn, "SIGTTOU: background write to tty"},
  				 
	/* 23 */ {_SigNotify + _SigIgn, "SIGURG: urgent condition on socket"},
	/* 24 */ {_SigNotify, "SIGXCPU: cpu limit exceeded"},
	/* 25 */ {_SigNotify, "SIGXFSZ: file size limit exceeded"},
	/* 26 */ {_SigNotify, "SIGVTALRM: virtual alarm clock"},
	/* 27 */ {_SigNotify + _SigUnblock, "SIGPROF: profiling alarm clock"},
	/* 28 */ {_SigNotify + _SigIgn, "SIGWINCH: window size change"},
	/* 29 */ {_SigNotify, "SIGIO: i/o now possible"},
	/* 30 */ {_SigNotify, "SIGPWR: power failure restart"},
	/* 31 */ {_SigThrow, "SIGSYS: bad system call"},
	/* 32 */ {_SigSetStack + _SigUnblock, "signal 32"}, /* SIGCANCEL; see issue 6997 */
	/* 33 */ {_SigSetStack + _SigUnblock, "signal 33"}, /* SIGSETXID; see issues 3871, 9400, 12498 */
	...
}

{_SigNotify + _SigIgn, "SIGURG: urgent condition on socket"}

func setsig(i uint32, fn uintptr) {
	var sa sigactiont
	sa.sa_flags = _SA_SIGINFO | _SA_ONSTACK | _SA_RESTORER | _SA_RESTART
	sigfillset(&sa.sa_mask)
	...
	if fn == funcPC(sighandler) {
        // CGO 相关
		if iscgo {
			fn = funcPC(cgoSigtramp)
		} else {
            // 替换为调用 sigtramp
			fn = funcPC(sigtramp)
		}
	}
	sa.sa_handler = fn
	sigaction(i, &sa, nil)
}

TEXT runtime·sigtramp<ABIInternal>(SB),NOSPLIT,$72
	...
	// We don't save mxcsr or the x87 control word because sigtrampgo doesn't
	// modify them.

	MOVQ	DX, ctx-56(SP)
	MOVQ	SI, info-64(SP)
	MOVQ	DI, signum-72(SP)
	MOVQ	$runtime·sigtrampgo(SB), AX
	CALL AX

	...
	RET

func sigtrampgo(sig uint32, info *siginfo, ctx unsafe.Pointer) {
	if sigfwdgo(sig, info, ctx) {
		return
	}
	c := &sigctxt{info, ctx}
	g := sigFetchG(c)
	... 
	sighandler(sig, info, ctx, g)
	setg(g)
	if setStack {
		restoreGsignalStack(&gsignalStack)
	}
}


func sighandler(sig uint32, info *siginfo, ctxt unsafe.Pointer, gp *g) {
	_g_ := getg()
	c := &sigctxt{info, ctxt}
	... 
  // 如果是一个抢占信号
	if sig == sigPreempt && debug.asyncpreemptoff == 0 { 
   		// 处理抢占信号
		doSigPreempt(gp, c) 
	}

	...
}

func doSigPreempt(gp *g, ctxt *sigctxt) { 
	// 检查此 G 是否要被抢占并且可以安全地抢占
	if wantAsyncPreempt(gp) { 
		// 检查是否能安全的进行抢占
		if ok, newpc := isAsyncSafePoint(gp, ctxt.sigpc(), ctxt.sigsp(), ctxt.siglr()); ok {
			// 修改寄存器,并执行抢占调用
			ctxt.pushCall(funcPC(asyncPreempt), newpc)
		}
	}
 
	// 更新一下抢占相关字段
	atomic.Xadd(&gp.m.preemptGen, 1)
	atomic.Store(&gp.m.signalPending, 0) 
}

// 保存用户态寄存器后调用asyncPreempt2
func asyncPreempt()

func asyncPreempt2() {
	gp := getg()
	gp.asyncSafePoint = true
	// 该 G 是否可以被抢占 
	if gp.preemptStop { 
		mcall(preemptPark)
	} else { 
    	// 让 G 放弃当前在 M 上的执行权利,将 G 放入全局队列等待后续调度
		mcall(gopreempt_m)
	}
	gp.asyncSafePoint = false
}

func preemptPark(gp *g) {
	
	status := readgstatus(gp)
	if status&^_Gscan != _Grunning {
		dumpgstatus(gp)
		throw("bad g status")
	}
	gp.waitreason = waitReasonPreempted 
	casGToPreemptScan(gp, _Grunning, _Gscan|_Gpreempted)
    // 使当前 m 放弃 g，让出线程
	dropg()
    // 修改当前 Goroutine 的状态到 _Gpreempted
	casfrom_Gscanstatus(gp, _Gscan|_Gpreempted, _Gpreempted)
    // 并继续执行调度
	schedule()
}

func gopreempt_m(gp *g) { 
	goschedImpl(gp)
}

func goschedImpl(gp *g) {
	status := readgstatus(gp)
	...
  // 更新状态为 _Grunnable
	casgstatus(gp, _Grunning, _Grunnable)
  // 使当前 m 放弃 g，让出线程
	dropg()
	lock(&sched.lock)
  // 重新加入到全局执行队列中
	globrunqput(gp)
	unlock(&sched.lock)
	// 并继续执行调度
	schedule()
}

const sigPreempt = _SIGURG

func preemptM(mp *m) {
	...
	if atomic.Cas(&mp.signalPending, 0, 1) { 
		
		// preemptM 向 M 发送抢占请求。
		// 接收到该请求后，如果正在运行的 G 或 P 被标记为抢占，并且 Goroutine 处于异步安全点，
		// 它将抢占 Goroutine。
		signalM(mp, sigPreempt)
	}
}

func retake(now int64) uint32 {
	n := 0
	 
	lock(&allpLock) 
	// 遍历 allp 数组
	for i := 0; i < len(allp); i++ {
		_p_ := allp[i]
		if _p_ == nil { 
			continue
		}
		pd := &_p_.sysmontick
		s := _p_.status
		sysretake := false
		if s == _Prunning || s == _Psyscall {
			// 调度次数
			t := int64(_p_.schedtick)
			if int64(pd.schedtick) != t {
				pd.schedtick = uint32(t)
				// 处理器上次调度时间
				pd.schedwhen = now
			// 抢占 G 的执行，如果上一次触发调度的时间已经过去了 10ms
			} else if pd.schedwhen+forcePreemptNS <= now {
				preemptone(_p_)
				sysretake = true
			}
		}
		...
	}
	unlock(&allpLock)
	return uint32(n)
}

func retake(now int64) uint32 {
	n := 0
	lock(&allpLock) 
	// 遍历 allp 数组
	for i := 0; i < len(allp); i++ {
		_p_ := allp[i]
		if _p_ == nil { 
			continue
		}
		pd := &_p_.sysmontick
		s := _p_.status
		sysretake := false
		...
		if s == _Psyscall { 
			// 系统调用的次数
			t := int64(_p_.syscalltick)
			if !sysretake && int64(pd.syscalltick) != t {
				pd.syscalltick = uint32(t)
				// 系统调用的时间
				pd.syscallwhen = now
				continue
			} 
			if runqempty(_p_) && atomic.Load(&sched.nmspinning)+atomic.Load(&sched.npidle) > 0 && pd.syscallwhen+10*1000*1000 > now {
				continue
			} 
			unlock(&allpLock) 
			incidlelocked(-1)
			if atomic.Cas(&_p_.status, s, _Pidle) { 
				n++
				_p_.syscalltick++
				// 让出处理器的使用权
				handoffp(_p_)
			}
			incidlelocked(1)
			lock(&allpLock)
		}
	}
	unlock(&allpLock)
	return uint32(n)
}

func markroot(gcw *gcWork, i uint32) { 
	...
 	switch { 
	...
	// 扫描各个 G 的栈
	default: 
		// 获取需要扫描的 G
		var gp *g
		if baseStacks <= i && i < end {
			gp = allgs[i-baseStacks]
		} else {
			throw("markroot: bad index")
		} 
		...
		// 转交给g0进行扫描
		systemstack(func() {  
			...
			// 挂起 G，让对应的 G 停止运行
			stopped := suspendG(gp)
			if stopped.dead {
				gp.gcscandone = true
				return
			}
			if gp.gcscandone {
				throw("g already scanned")
			}
			// 扫描g的栈
			scanstack(gp, gcw)
			gp.gcscandone = true
			// 恢复该 G 的执行
			resumeG(stopped) 
		})
	}
}

func suspendG(gp *g) suspendGState {
	...
	const yieldDelay = 10 * 1000

	var nextPreemptM int64
	for i := 0; ; i++ {
		switch s := readgstatus(gp); s { 
		... 
		case _Grunning:
			if gp.preemptStop && gp.preempt && gp.stackguard0 == stackPreempt && asyncM == gp.m && atomic.Load(&asyncM.preemptGen) == asyncGen {
				break
			}
			if !castogscanstatus(gp, _Grunning, _Gscanrunning) {
				break
			}
			// 设置抢占字段
			gp.preemptStop = true
			gp.preempt = true
			gp.stackguard0 = stackPreempt
 
			asyncM2 := gp.m
			asyncGen2 := atomic.Load(&asyncM2.preemptGen)
			// asyncM 与 asyncGen 标记的是循环里 上次抢占的信息，用来校验不能重复抢占
			needAsync := asyncM != asyncM2 || asyncGen != asyncGen2
			asyncM = asyncM2
			asyncGen = asyncGen2

			casfrom_Gscanstatus(gp, _Gscanrunning, _Grunning)
 
			if preemptMSupported && debug.asyncpreemptoff == 0 && needAsync { 
				now := nanotime()
				// 限制抢占的频率
				if now >= nextPreemptM {
					nextPreemptM = now + yieldDelay/2
					// 执行抢占信号发送
					preemptM(asyncM)
				}
			}
		}
		...
	}
}

func stopTheWorldWithSema() {
	_g_ := getg() 

	lock(&sched.lock)
	sched.stopwait = gomaxprocs
	// 标记 gcwaiting，调度时看见此标记会进入等待
	atomic.Store(&sched.gcwaiting, 1)
	// 发送抢占信号
	preemptall() 
	// 暂停当前 P
	_g_.m.p.ptr().status = _Pgcstop // Pgcstop is only diagnostic.
	...
	wait := sched.stopwait > 0
	unlock(&sched.lock)
	if wait {
		for {
			//  等待 100 us
			if notetsleep(&sched.stopnote, 100*1000) {
				noteclear(&sched.stopnote)
				break
			}
			// 再次进行发送抢占信号
			preemptall()
		}
	}
	...
}

func preemptall() bool {
   res := false
   // 遍历所有的 P
   for _, _p_ := range allp {
      if _p_.status != _Prunning {
         continue
      }
      // 对正在运行的 P 发送抢占信号
      if preemptone(_p_) {
         res = true
      }
   }
   return res
}

func preemptone(_p_ *p) bool {
	// 获取 P 对应的 M
	mp := _p_.m.ptr()
	if mp == nil || mp == getg().m {
		return false
	}
	// 获取 M 正在执行的 G
	gp := mp.curg
	if gp == nil || gp == mp.g0 {
		return false
	}
	// 将 G 标记为抢占
	gp.preempt = true

	// 在栈扩张的时候会检测是否被抢占
	gp.stackguard0 = stackPreempt

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