Go 协作与抢占

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

// 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)    ...  }
  ...}

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()}

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

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)

//go:nowritebarrierrecfunc 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)}

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

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

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

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}

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

//go:nowritebarrierrecfunc 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)

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

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

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

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