135-1821-9792

聊一聊Go协作与抢占

本文转载自微信公众号「 码农桃花源」，转载本文请联系 码农桃花源公众号。

创新互联坚持“要么做到，要么别承诺”的工作理念，服务领域包括：成都做网站、网站建设、企业官网、英文网站、手机端网站、网站推广等服务，满足客户于互联网时代的乌拉特后网站设计、移动媒体设计的需求，帮助企业找到有效的互联网解决方案。努力成为您成熟可靠的网络建设合作伙伴！

  • 协作式调度
    • 主动用户让权：Gosched
    • 主动调度弃权：栈扩张与抢占标记
  • 抢占式调度
    • P 抢占
    • M 抢占
  • 小结
  • 进一步阅读的参考文献

我们在分析调度循环[1]的时候总结过一个问题：如果某个 G 执行时间过长，其他的 G 如何才能被正常地调度?这便涉及到有关调度的两个理念：协作式调度与抢占式调度。

协作式和抢占式这两个理念解释起来很简单：协作式调度依靠被调度方主动弃权;抢占式调度则依靠调度器强制将被调度方被动中断。这两个概念其实描述了调度的两种截然不同的策略，这两种决策模式，在调度理论中其实已经研究得很透彻了。

Go 的运行时并不具备操作系统内核级的硬件中断能力，基于工作窃取的调度器实现，本质上属于先来先服务的协作式调度，为了解决响应时间可能较高的问题，目前运行时实现了两种不同的调度策略、每种策略各两个形式。保证在大部分情况下，不同的 G 能够获得均匀的时间片：

  • 同步协作式调度
  1. 主动用户让权：通过 runtime.Gosched 调用主动让出执行机会;
  2. 主动调度弃权：当发生执行栈分段时，检查自身的抢占标记，决定是否继续执行;
  • 异步抢占式调度
  1. 被动监控抢占：当 G 阻塞在 M 上时(系统调用、channel 等)，系统监控会将 P 从 M 上抢夺并分配给其他的 M 来执行其他的 G，而位于被抢夺 P 的本地调度队列中的 G 则可能会被偷取到其他 M 执行。
  2. 被动 GC 抢占：当需要进行垃圾回收时，为了保证不具备主动抢占处理的函数执行时间过长，导致垃圾回收迟迟不能执行而导致的高延迟，而强制停止 G 并转为执行垃圾回收。

协作式调度

主动用户让权：Gosched

Gosched 是一种主动放弃执行的手段，用户态代码通过调用此接口来出让执行机会，使其他“人”也能在密集的执行过程中获得被调度的机会。

Gosched 的实现非常简单：

 
 
 
    
  
  
  
  1. // Gosched 会让出当前的 P，并允许其他 Goroutine 运行。 
    2. 
  
  
  
  2. // 它不会推迟当前的 Goroutine，因此执行会被自动恢复 
    3. 
  
  
  
  3. func Gosched() { 
    4. 
  
  
  
  4.   checkTimeouts() 
    5. 
  
  
  
  5.   mcall(gosched_m) 
    6. 
  
  
  
  6. } 
    7. 
  
  
  
  7.  
    8. 
  
  
  
  8. // Gosched 在 g0 上继续执行 
    9. 
  
  
  
  9. func gosched_m(gp *g) { 
    10. 
  
  
  
  10.   ... 
    11. 
  
  
  
  11.   goschedImpl(gp) 
    12. 
  
  
  
  12. } 
    13.

它首先会通过 note 机制通知那些等待被 ready 的 Goroutine：

 
 
 
    
  
  
  
  1. // checkTimeouts 恢复那些在等待一个 note 且已经触发其 deadline 时的 Goroutine。 
    2. 
  
  
  
  2. func checkTimeouts() { 
    3. 
  
  
  
  3.   now := nanotime() 
    4. 
  
  
  
  4.   for n, nt := range notesWithTimeout { 
    5. 
  
  
  
  5.     if n.key == note_cleared && now > nt.deadline { 
    6. 
  
  
  
  6.       n.key = note_timeout 
    7. 
  
  
  
  7.       goready(nt.gp, 1) 
    8. 
  
  
  
  8.     } 
    9. 
  
  
  
  9.   } 
    10. 
  
  
  
  10. } 
    11. 
  
  
  
  11.  
    12. 
  
  
  
  12. func goready(gp *g, traceskip int) { 
    13. 
  
  
  
  13.   systemstack(func() { 
    14. 
  
  
  
  14.     ready(gp, traceskip, true) 
    15. 
  
  
  
  15.   }) 
    16. 
  
  
  
  16. } 
    17. 
  
  
  
  17.  
    18. 
  
  
  
  18. // 将 gp 标记为 ready 来运行 
    19. 
  
  
  
  19. func ready(gp *g, traceskip int, next bool) { 
    20. 
  
  
  
  20.   if trace.enabled { 
    21. 
  
  
  
  21.     traceGoUnpark(gp, traceskip) 
    22. 
  
  
  
  22.   } 
    23. 
  
  
  
  23.  
    24. 
  
  
  
  24.   status := readgstatus(gp) 
    25. 
  
  
  
  25.  
    26. 
  
  
  
  26.   // 标记为 runnable. 
    27. 
  
  
  
  27.   _g_ := getg() 
    28. 
  
  
  
  28.   _g_.m.locks++ // 禁止抢占，因为它可以在局部变量中保存 p 
    29. 
  
  
  
  29.   if status&^_Gscan != _Gwaiting { 
    30. 
  
  
  
  30.     dumpgstatus(gp) 
    31. 
  
  
  
  31.     throw("bad g->status in ready") 
    32. 
  
  
  
  32.   } 
    33. 
  
  
  
  33.  
    34. 
  
  
  
  34.   // 状态为 Gwaiting 或 Gscanwaiting, 标记 Grunnable 并将其放入运行队列 runq 
    35. 
  
  
  
  35.   casgstatus(gp, _Gwaiting, _Grunnable) 
    36. 
  
  
  
  36.   runqput(_g_.m.p.ptr(), gp, next) 
    37. 
  
  
  
  37.   if atomic.Load(&sched.npidle) != 0 && atomic.Load(&sched.nmspinning) == 0 { 
    38. 
  
  
  
  38.     wakep() 
    39. 
  
  
  
  39.   } 
    40. 
  
  
  
  40.   _g_.m.locks-- 
    41. 
  
  
  
  41.   if _g_.m.locks == 0 && _g_.preempt { // 在 newstack 中已经清除它的情况下恢复抢占请求 
    42. 
  
  
  
  42.     _g_.stackguard0 = stackPreempt 
    43. 
  
  
  
  43.   } 
    44. 
  
  
  
  44. } 
    45. 
  
  
  
  45.  
    46. 
  
  
  
  46. func notetsleepg(n *note, ns int64) bool { 
    47. 
  
  
  
  47.   gp := getg() 
    48. 
  
  
  
  48.   ... 
    49. 
  
  
  
  49.  
    50. 
  
  
  
  50.   if ns >= 0 { 
    51. 
  
  
  
  51.     deadline := nanotime() + ns 
    52. 
  
  
  
  52.     ... 
    53. 
  
  
  
  53.     notesWithTimeout[n] = noteWithTimeout{gp: gp, deadline: deadline} 
    54. 
  
  
  
  54.     ... 
    55. 
  
  
  
  55.     gopark(nil, nil, waitReasonSleep, traceEvNone, 1) 
    56. 
  
  
  
  56.     ... 
    57. 
  
  
  
  57.     delete(notesWithTimeout, n) 
    58. 
  
  
  
  58.     ... 
    59. 
  
  
  
  59.   } 
    60. 
  
  
  
  60.  
    61. 
  
  
  
  61.   ... 
    62. 
  
  
  
  62. } 
    63.

而后通过 mcall 调用 gosched_m 在 g0 上继续执行并让出 P，实质上是让 G 放弃当前在 M 上的执行权利，M 转去执行其他的 G，并在上下文切换时候，将自身放入全局队列等待后续调度：

 
 
 
    
  
  
  
  1. func goschedImpl(gp *g) { 
    2. 
  
  
  
  2.   // 放弃当前 g 的运行状态 
    3. 
  
  
  
  3.   status := readgstatus(gp) 
    4. 
  
  
  
  4.   ... 
    5. 
  
  
  
  5.   casgstatus(gp, _Grunning, _Grunnable) 
    6. 
  
  
  
  6.   // 使当前 m 放弃 g 
    7. 
  
  
  
  7.   dropg() 
    8. 
  
  
  
  8.   // 并将 g 放回全局队列中 
    9. 
  
  
  
  9.   lock(&sched.lock) 
    10. 
  
  
  
  10.   globrunqput(gp) 
    11. 
  
  
  
  11.   unlock(&sched.lock) 
    12. 
  
  
  
  12.  
    13. 
  
  
  
  13.   // 重新进入调度循环 
    14. 
  
  
  
  14.   schedule() 
    15. 
  
  
  
  15. } 
    16.

当然，尽管具有主动弃权的能力，但它对 Go 语言的用户要求比较高，因为用户在编写并发逻辑的时候需要自行甄别是否需要让出时间片，这并非用户友好的，而且很多 Go 的新用户并不会了解到这个问题的存在，我们在随后的抢占式调度中再进一步展开讨论。

主动调度弃权：栈扩张与抢占标记

另一种主动放弃的方式是通过抢占标记的方式实现的。基本想法是在每个函数调用的序言(函数调用的最前方)插入抢占检测指令，当检测到当前 Goroutine 被标记为应该被抢占时，则主动中断执行，让出执行权利。表面上看起来想法很简单，但实施起来就比较复杂了。

在 6.6 执行栈管理[2] 一节中我们已经了解到，函数调用的序言部分会检查 SP 寄存器与 stackguard0 之间的大小，如果 SP 小于 stackguard0 则会触发 morestack_noctxt，触发栈分段操作。换言之，如果抢占标记将 stackgard0 设为比所有可能的 SP 都要大(即 stackPreempt)，则会触发 morestack，进而调用 newstack：

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

从抢占调度的角度来看，这种发生在函数序言部分的抢占的一个重要目的就是能够简单且安全的记录执行现场(随后的抢占式调度我们会看到记录执行现场给采用信号方式中断线程执行的调度带来多大的困难)。事实也是如此，在 morestack 调用中：

 
 
 
    
  
  
  
  1. TEXT runtime·morestack(SB),NOSPLIT,$0-0 
    2. 
  
  
  
  2.     ... 
    3. 
  
  
  
  3.     MOVQ    0(SP), AX // f's PC 
    4. 
  
  
  
  4.     MOVQ    AX, (g_sched+gobuf_pc)(SI) 
    5. 
  
  
  
  5.     MOVQ    SI, (g_sched+gobuf_g)(SI) 
    6. 
  
  
  
  6.     LEAQ    8(SP), AX // f's SP 
    7. 
  
  
  
  7.     MOVQ    AX, (g_sched+gobuf_sp)(SI) 
    8. 
  
  
  
  8.     MOVQ    BP, (g_sched+gobuf_bp)(SI) 
    9. 
  
  
  
  9.     MOVQ    DX, (g_sched+gobuf_ctxt)(SI) 
    10. 
  
  
  
  10.     ... 
    11. 
  
  
  
  11.     CALL    runtime·newstack(SB) 
    12.

是有记录 Goroutine 的 PC 和 SP 寄存器，而后才开始调用 newstack 的：

 
 
 
    
  
  
  
  1. //go:nowritebarrierrec 
    2. 
  
  
  
  2. func newstack() { 
    3. 
  
  
  
  3.   thisg := getg() 
    4. 
  
  
  
  4.   ... 
    5. 
  
  
  
  5.  
    6. 
  
  
  
  6.   gp := thisg.m.curg 
    7. 
  
  
  
  7.   ... 
    8. 
  
  
  
  8.  
    9. 
  
  
  
  9.   morebuf := thisg.m.morebuf 
    10. 
  
  
  
  10.   thisg.m.morebuf.pc = 0 
    11. 
  
  
  
  11.   thisg.m.morebuf.lr = 0 
    12. 
  
  
  
  12.   thisg.m.morebuf.sp = 0 
    13. 
  
  
  
  13.   thisg.m.morebuf.g = 0 
    14. 
  
  
  
  14.  
    15. 
  
  
  
  15.   // 如果是发起的抢占请求而非真正的栈分段 
    16. 
  
  
  
  16.   preempt := atomic.Loaduintptr(&gp.stackguard0) == stackPreempt 
    17. 
  
  
  
  17.  
    18. 
  
  
  
  18.   // 保守的对用户态代码进行抢占，而非抢占运行时代码 
    19. 
  
  
  
  19.   // 如果正持有锁、分配内存或抢占被禁用，则不发生抢占 
    20. 
  
  
  
  20.   if preempt { 
    21. 
  
  
  
  21.     if !canPreemptM(thisg.m) { 
    22. 
  
  
  
  22.       // 不发生抢占，继续调度 
    23. 
  
  
  
  23.       gp.stackguard0 = gp.stack.lo + _StackGuard 
    24. 
  
  
  
  24.       gogo(&gp.sched) // 重新进入调度循环 
    25. 
  
  
  
  25.     } 
    26. 
  
  
  
  26.   } 
    27. 
  
  
  
  27.   ... 
    28. 
  
  
  
  28.   // 如果需要对栈进行调整 
    29. 
  
  
  
  29.   if preempt { 
    30. 
  
  
  
  30.     ... 
    31. 
  
  
  
  31.     if gp.preemptShrink { 
    32. 
  
  
  
  32.       // 我们正在一个同步安全点，因此等待栈收缩 
    33. 
  
  
  
  33.       gp.preemptShrink = false 
    34. 
  
  
  
  34.       shrinkstack(gp) 
    35. 
  
  
  
  35.     } 
    36. 
  
  
  
  36.     if gp.preemptStop { 
    37. 
  
  
  
  37.       preemptPark(gp) // 永不返回 
    38. 
  
  
  
  38.     } 
    39. 
  
  
  
  39.     ... 
    40. 
  
  
  
  40.     // 表现得像是调用了 runtime.Gosched，主动让权 
    41. 
  
  
  
  41.     gopreempt_m(gp) // 重新进入调度循环 
    42. 
  
  
  
  42.   } 
    43. 
  
  
  
  43.   ... 
    44. 
  
  
  
  44. } 
    45. 
  
  
  
  45. // 与 gosched_m 一致 
    46. 
  
  
  
  46. func gopreempt_m(gp *g) { 
    47. 
  
  
  
  47.   ... 
    48. 
  
  
  
  48.   goschedImpl(gp) 
    49. 
  
  
  
  49. } 
    50.

其中的 canPreemptM 验证了可以被抢占的条件：

  1. 运行时没有禁止抢占(m.locks == 0)
  2. 运行时没有在执行内存分配(m.mallocing == 0)
  3. 运行时没有关闭抢占机制(m.preemptoff == "")
  4. M 与 P 绑定且没有进入系统调用(p.status == _Prunning)
 
 
 
    
  
  
  
  1. // canPreemptM 报告 mp 是否处于可抢占的安全状态。 
    2. 
  
  
  
  2. //go:nosplit 
    3. 
  
  
  
  3. func canPreemptM(mp *m) bool { 
    4. 
  
  
  
  4.   return mp.locks == 0 && mp.mallocing == 0 && mp.preemptoff == "" && mp.p.ptr().status == _Prunning 
    5. 
  
  
  
  5. } 
    6.

从可被抢占的条件来看，能够对一个 G 进行抢占其实是呈保守状态的。这一保守体现在抢占对很多运行时所需的条件进行了判断，这也理所当然是因为运行时优先级更高，不应该轻易发生抢占，但与此同时由于又需要对用户态代码进行抢占，于是先作出一次不需要抢占的判断(快速路径)，确定不能抢占时返回并继续调度，如果真的需要进行抢占，则转入调用 gopreempt_m，放弃当前 G 的执行权，将其加入全局队列，重新进入调度循环。

什么时候会给 stackguard0 设置抢占标记 stackPreempt 呢?一共有以下几种情况：

  1. 进入系统调用时(runtime.reentersyscall，注意这种情况是为了保证不会发生栈分裂，真正的抢占是异步地通过系统监控进行的)
  2. 任何运行时不再持有锁的时候(m.locks == 0)
  3. 当垃圾回收器需要停止所有用户 Goroutine 时

抢占式调度

从上面提到的两种协作式调度逻辑我们可以看出，这种需要用户代码来主动配合的调度方式存在一些致命的缺陷：一个没有主动放弃执行权、且不参与任何函数调用的函数，直到执行完毕之前，是不会被抢占的。

那么这种不会被抢占的函数会导致什么严重的问题呢?回答是，由于运行时无法停止该用户代码，则当需要进行垃圾回收时，无法及时进行;对于一些实时性要求较高的用户态 Goroutine 而言，也久久得不到调度。我们这里不去深入讨论垃圾回收的具体细节，读者将在垃圾回收器[3]一章中详细看到这类问题导致的后果。单从调度的角度而言，我们直接来看一个非常简单的例子：

 
 
 
    
  
  
  
  1. // 此程序在 Go 1.14 之前的版本不会输出 OK 
    2. 
  
  
  
  2. package main 
    3. 
  
  
  
  3. import ( 
    4. 
  
  
  
  4.   "runtime" 
    5. 
  
  
  
  5.   "time" 
    6. 
  
  
  
  6. ) 
    7. 
  
  
  
  7. func main() { 
    8. 
  
  
  
  8.   runtime.GOMAXPROCS(1) 
    9. 
  
  
  
  9.   go func() { 
    10. 
  
  
  
  10.     for { 
    11. 
  
  
  
  11.     } 
    12. 
  
  
  
  12.   }() 
    13. 
  
  
  
  13.   time.Sleep(time.Millisecond) 
    14. 
  
  
  
  14.   println("OK") 
    15. 
  
  
  
  15. } 
    16.

这段代码中处于死循环的 Goroutine 永远无法被抢占，其中创建的 Goroutine 会执行一个不产生任何调用、不主动放弃执行权的死循环。由于主 Goroutine 优先调用了休眠，此时唯一的 P 会转去执行 for 循环所创建的 Goroutine。进而主 Goroutine 永远不会再被调度，进而程序彻底阻塞在了这个 Goroutine 上，永远无法退出。这样的例子非常多，但追根溯源，均为此问题导致。

Go 团队其实很早(1.0 以前)就已经意识到了这个问题，但在 Go 1.2 时增加了上文提到的在函数序言部分增加抢占标记后，此问题便被搁置，直到越来越多的用户提交并报告此问题。在 Go 1.5 前后，Austin Clements 希望仅解决这种由密集循环导致的无法抢占的问题 [Clements, 2015]，于是尝试通过协作式 loop 循环抢占，通过编译器辅助的方式，插入抢占检查指令，与流程图回边(指节点被访问过但其子节点尚未访问完毕)安全点(在一个线程执行中，垃圾回收器能够识别所有对象引用状态的一个状态)的方式进行解决。

尽管此举能为抢占带来显著的提升，但是在一个循环中引入分支显然会降低性能。尽管随后 David Chase 对这个方法进行了改进，仅在插入了一条 TESTB 指令 [Chase, 2017]，在完全没有分支以及寄存器压力的情况下，仍然造成了几何平均 7.8% 的性能损失。这种结果其实是情理之中的，很多需要进行密集循环的计算时间都是在运行时才能确定的，直接由编译器检测这类密集循环而插入额外的指令可想而知是欠妥的做法。

终于在 Go 1.10 后 [Clements, 2019]，Austin 进一步提出的解决方案，希望使用每个指令与执行栈和寄存器的映射关系，通过记录足够多的信息，并通过异步线程来发送抢占信号的方式来支持异步抢占式调度。

我们知道现代操作系统的调度器多为抢占式调度，其实现方式通过硬件中断来支持线程的切换，进而能安全的保存运行上下文。在 Go 运行时实现抢占式调度同样也可以使用类似的方式，通过向线程发送系统信号的方式来中断 M 的执行，进而达到抢占的目的。但与操作系统的不同之处在于，由于运行时诸多机制的存在(例如垃圾回收器)，还必须能够在 Goroutine 被停止时，保存充足的上下文信息(见 8.9 安全点分析[4])。这就给中断信号带来了麻烦，如果中断信号恰好发生在一些关键阶段(例如写屏障期间)，则无法保证程序的正确性。这也就要求我们需要严格考虑触发异步抢占的时机。

异步抢占式调度的一种方式就与运行时系统监控有关，监控循环会将发生阻塞的 Goroutine 抢占，解绑 P 与 M，从而让其他的线程能够获得 P 继续执行其他的 Goroutine。这得益于 sysmon中调用的 retake 方法。这个方法处理了两种抢占情况，一是抢占阻塞在系统调用上的 P，二是抢占运行时间过长的 G。其中抢占运行时间过长的 G 这一方式还会出现在垃圾回收需要进入 STW 时。

P 抢占

我们先来看抢占阻塞在系统调用上的 G 这种情况。这种抢占的实现方法非常的自然，因为 Goroutine 已经阻塞在了系统调用上，我们可以非常安全的将 M 与 P 进行解绑，即便是 Goroutine 从阻塞中恢复，也会检查自身所在的 M 是否仍然持有 P，如果没有 P 则重新考虑与可用的 P 进行绑定。这种异步抢占的本质是：抢占 P。

 
 
 
    
  
  
  
  1. unc retake(now int64) uint32 { 
    2. 
  
  
  
  2.   n := 0 
    3. 
  
  
  
  3.   // 防止 allp 数组发生变化，除非我们已经 STW，此锁将完全没有人竞争 
    4. 
  
  
  
  4.   lock(&allpLock) 
    5. 
  
  
  
  5.   for i := 0; i < len(allp); i++ { 
    6. 
  
  
  
  6.     _p_ := allp[i] 
    7. 
  
  
  
  7.     ... 
    8. 
  
  
  
  8.     pd := &_p_.sysmontick 
    9. 
  
  
  
  9.     s := _p_.status 
    10. 
  
  
  
  10.     sysretake := false 
    11. 
  
  
  
  11.     if s == _Prunning || s == _Psyscall { 
    12. 
  
  
  
  12.       // 如果 G 运行时时间太长则进行抢占 
    13. 
  
  
  
  13.       t := int64(_p_.schedtick) 
    14. 
  
  
  
  14.       if int64(pd.schedtick) != t { 
    15. 
  
  
  
  15.         pd.schedtick = uint32(t) 
    16. 
  
  
  
  16.         pd.schedwhen = now 
    17. 
  
  
  
  17.       } else if pd.schedwhen+forcePreemptNS <= now { 
    18. 
  
  
  
  18.         ... 
    19. 
  
  
  
  19.         sysretake = true 
    20. 
  
  
  
  20.       } 
    21. 
  
  
  
  21.     } 
    22. 
  
  
  
  22.     // 对阻塞在系统调用上的 P 进行抢占 
    23. 
  
  
  
  23.     if s == _Psyscall { 
    24. 
  
  
  
  24.       // 如果已经超过了一个系统监控的 tick（20us），则从系统调用中抢占 P 
    25. 
  
  
  
  25.       t := int64(_p_.syscalltick) 
    26. 
  
  
  
  26.       if !sysretake && int64(pd.syscalltick) != t { 
    27. 
  
  
  
  27.         pd.syscalltick = uint32(t) 
    28. 
  
  
  
  28.         pd.syscallwhen = now 
    29. 
  
  
  
  29.         continue 
    30. 
  
  
  
  30.       } 
    31. 
  
  
  
  31.       // 一方面，在没有其他 work 的情况下，我们不希望抢夺 P 
    32. 
  
  
  
  32.       // 另一方面，因为它可能阻止 sysmon 线程从深度睡眠中唤醒，所以最终我们仍希望抢夺 P 
    33. 
  
  
  
  33.       if runqempty(_p_) && atomic.Load(&sched.nmspinning)+atomic.Load(&sched.npidle) > 0 && pd.syscallwhen+10*1000*1000 > now { 
    34. 
  
  
  
  34.         continue 
    35. 
  
  
  
  35.       } 
    36. 
  
  
  
  36.       // 解除 allpLock，从而可以获取 sched.lock 
    37. 
  
  
  
  37.       unlock(&allpLock) 
    38. 
  
  
  
  38.       // 在 CAS 之前需要减少空闲 M 的数量（假装某个还在运行） 
    39. 
  
  
  
  39.       // 否则发生抢夺的 M 可能退出 syscall 然后再增加 nmidle ，进而发生死锁 
    40. 
  
  
  
  40.       // 这个过程发生在 stoplockedm 中 
    41. 
  
  
  
  41.       incidlelocked(-1) 
    42. 
  
  
  
  42.       if atomic.Cas(&_p_.status, s, _Pidle) { // 将 P 设为 idle，从而交与其他 M 使用 
    43. 
  
  
  
  43.         ... 
    44. 
  
  
  
  44.         n++ 
    45. 
  
  
  
  45.         _p_.syscalltick++ 
    46. 
  
  
  
  46.         handoffp(_p_) 
    47. 
  
  
  
  47.       } 
    48. 
  
  
  
  48.       incidlelocked(1) 
    49. 
  
  
  
  49.       lock(&allpLock) 
    50. 
  
  
  
  50.     } 
    51. 
  
  
  
  51.   } 
    52. 
  
  
  
  52.   unlock(&allpLock) 
    53. 
  
  
  
  53.   return uint32(n) 
    54. 
  
  
  
  54. } 
    55.

在抢占 P 的过程中，有两个非常小心的处理方式：

  1. 如果此时队列为空，那么完全没有必要进行抢占，这时候似乎可以继续遍历其他的 P，但必须在调度器中自旋的 M 和 空闲的 P 同时存在时、且系统调用阻塞时间非常长的情况下才能这么做。否则，这个 retake 过程可能返回 0，进而系统监控可能看起来像是什么事情也没做的情况下调整自己的步调进入深度睡眠。
  2. 在将 P 设置为空闲状态前，必须先将 M 的数量减少，否则当 M 退出系统调用时，会在 exitsyscall0 中调用 stoplockedm 从而增加空闲 M 的数量，进而发生死锁。

M 抢占

在上面我们没有展现一个细节，那就是在检查 P 的状态时，P 如果是运行状态会调用preemptone，来通过系统信号来完成抢占，之所以没有在之前提及的原因在于该调用在 M 不与 P 绑定的情况下是不起任何作用直接返回的。这种异步抢占的本质是：抢占 M。我们不妨继续从系统监控产生的抢占谈起：

 
 
 
    
  
  
  
  1. func retake(now int64) uint32 { 
    2. 
  
  
  
  2.   ... 
    3. 
  
  
  
  3.   for i := 0; i < len(allp); i++ { 
    4. 
  
  
  
  4.     _p_ := allp[i] 
    5. 
  
  
  
  5.     ... 
    6. 
  
  
  
  6.     if s == _Prunning || s == _Psyscall { 
    7. 
  
  
  
  7.       ... 
    8. 
  
  
  
  8.       } else if pd.schedwhen+forcePreemptNS <= now { 
    9. 
  
  
  
  9.         // 对于 syscall 的情况，因为 M 没有与 P 绑定， 
    10. 
  
  
  
  10.         // preemptone() 不工作 
    11. 
  
  
  
  11.         preemptone(_p_) 
    12. 
  
  
  
  12.         sysretake = true 
    13. 
  
  
  
  13.       } 
    14. 
  
  
  
  14.     } 
    15. 
  
  
  
  15.     ... 
    16. 
  
  
  
  16.   } 
    17. 
  
  
  
  17.   ... 
    18. 
  
  
  
  18. } 
    19. 
  
  
  
  19. func preemptone(_p_ *p) bool { 
    20. 
  
  
  
  20.   // 检查 M 与 P 是否绑定 
    21. 
  
  
  
  21.   mp := _p_.m.ptr() 
    22. 
  
  
  
  22.   if mp == nil || mp == getg().m { 
    23. 
  
  
  
  23.     return false 
    24. 
  
  
  
  24.   } 
    25. 
  
  
  
  25.   gp := mp.curg 
    26. 
  
  
  
  26.   if gp == nil || gp == mp.g0 { 
    27. 
  
  
  
  27.     return false 
    28. 
  
  
  
  28.   } 
    29. 
  
  
  
  29.  
    30. 
  
  
  
  30.   // 将 G 标记为抢占 
    31. 
  
  
  
  31.   gp.preempt = true 
    32. 
  
  
  
  32.  
    33. 
  
  
  
  33.   // 一个 Goroutine 中的每个调用都会通过比较当前栈指针和 gp.stackgard0 
    34. 
  
  
  
  34.   // 来检查栈是否溢出。 
    35. 
  
  
  
  35.   // 设置 gp.stackgard0 为 StackPreempt 来将抢占转换为正常的栈溢出检查。 
    36. 
  
  
  
  36.   gp.stackguard0 = stackPreempt 
    37. 
  
  
  
  37.  
    38. 
  
  
  
  38.   // 请求该 P 的异步抢占 
    39. 
  
  
  
  39.   if preemptMSupported && debug.asyncpreemptoff == 0 { 
    40. 
  
  
  
  40.     _p_.preempt = true 
    41. 
  
  
  
  41.     preemptM(mp) 
    42. 
  
  
  
  42.   } 
    43. 
  
  
  
  43.  
    44. 
  
  
  
  44.   return true 
    45. 
  
  
  
  45. } 
    46.

抢占信号的选取

preemptM 完成了信号的发送，其实现也非常直接，直接向需要进行抢占的 M 发送 SIGURG 信号即可。但是真正的重要的问题是，为什么是 SIGURG 信号而不是其他的信号?如何才能保证该信号不与用户态产生的信号产生冲突?这里面有几个原因：

  1. 默认情况下，SIGURG 已经用于调试器传递信号。
  2. SIGURG 可以不加选择地虚假发生的信号。例如，我们不能选择 SIGALRM，因为信号处理程序无法分辨它是否是由实际过程引起的(可以说这意味着信号已损坏)。而常见的用户自定义信号 SIGUSR1 和 SIGUSR2 也不够好，因为用户态代码可能会将其进行使用。
  3. 需要处理没有实时信号的平台(例如 macOS)。

考虑以上的观点，SIGURG 其实是一个很好的、满足所有这些条件、且极不可能因被用户态代码进行使用的一种信号。

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

抢占调用的注入

我们在信号处理一节[5]中已经知道，每个运行的 M 都会设置一个系统信号的处理的回调，当出现系统信号时，操作系统将负责将运行代码进行中断，并安全的保护其执行现场，进而 Go 运行时能将针对信号的类型进行处理，当信号处理函数执行结束后，程序会再次进入内核空间，进而恢复到被中断的位置。

但是这里面有一个很巧妙的用法，因为 sighandler 能够获得操作系统所提供的执行上下文参数(例如寄存器 rip, rep 等)，如果在 sighandler 中修改了这个上下文参数，OS 会根据就该的寄存器进行恢复，这也就为抢占提供了机会。

 
 
 
    
  
  
  
  1. //go:nowritebarrierrec 
    2. 
  
  
  
  2. func sighandler(sig uint32, info *siginfo, ctxt unsafe.Pointer, gp *g) { 
    3. 
  
  
  
  3.   ... 
    4. 
  
  
  
  4.   c := &sigctxt{info, ctxt} 
    5. 
  
  
  
  5.   ... 
    6. 
  
  
  
  6.   if sig == sigPreempt { 
    7. 
  
  
  
  7.     // 可能是一个抢占信号 
    8. 
  
  
  
  8.     doSigPreempt(gp, c) 
    9. 
  
  
  
  9.     // 即便这是一个抢占信号，它也可能与其他信号进行混合，因此我们 
    10. 
  
  
  
  10.     // 继续进行处理。 
    11. 
  
  
  
  11.   } 
    12. 
  
  
  
  12.   ... 
    13. 
  
  
  
  13. } 
    14. 
  
  
  
  14. // doSigPreempt 处理了 gp 上的抢占信号 
    15. 
  
  
  
  15. func doSigPreempt(gp *g, ctxt *sigctxt) { 
    16. 
  
  
  
  16.   // 检查 G 是否需要被抢占、抢占是否安全 
    17. 
  
  
  
  17.   if wantAsyncPreempt(gp) && isAsyncSafePoint(gp, ctxt.sigpc(), ctxt.sigsp(), ctxt.siglr()) { 
    18. 
  
  
  
  18.     // 插入抢占调用 
    19. 
  
  
  
  19.     ctxt.pushCall(funcPC(asyncPreempt)) 
    20. 
  
  
  
  20.   } 
    21. 
  
  
  
  21.  
    22. 
  
  
  
  22.   // 记录抢占 
    23. 
  
  
  
  23.   atomic.Xadd(&gp.m.preemptGen, 1) 
    24.

在 ctxt.pushCall 之前，ctxt.rip() 和 ctxt.rep() 都保存了被中断的 Goroutine 所在的位置，但是 pushCall 直接修改了这些寄存器，进而当从 sighandler 返回用户态 Goroutine 时，能够从注入的 asyncPreempt 开始执行：

 
 
 
    
  
  
  
  1. func (c *sigctxt) pushCall(targetPC uintptr) { 
    2. 
  
  
  
  2.   pc := uintptr(c.rip()) 
    3. 
  
  
  
  3.   sp := uintptr(c.rsp()) 
    4. 
  
  
  
  4.   sp -= sys.PtrSize 
    5. 
  
  
  
  5.   *(*uintptr)(unsafe.Pointer(sp)) = pc 
    6. 
  
  
  
  6.   c.set_rsp(uint64(sp)) 
    7. 
  
  
  
  7.   c.set_rip(uint64(targetPC)) 
    8. 
  
  
  
  8. } 
    9.

完成 sighandler 之，我们成功恢复到 asyncPreempt 调用：

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

该函数的主要目的是保存用户态寄存器，并且在调用完毕前恢复所有的寄存器上下文就好像什么事情都没有发生过一样：

 
 
 
    
  
  
  
  1. TEXT ·asyncPreempt(SB),NOSPLIT|NOFRAME,$0-0 
    2. 
  
  
  
  2.     ... 
    3. 
  
  
  
  3.     MOVQ AX, 0(SP) 
    4. 
  
  
  
  4.     ... 
    5. 
  
  
  
  5.     MOVUPS X15, 352(SP) 
    6. 
  
  
  
  6.     CALL ·asyncPreempt2(SB) 
    7. 
  
  
  
  7.     MOVUPS 352(SP), X15 
    8. 
  
  
  
  8.     ... 
    9. 
  
  
  
  9.     MOVQ 0(SP), AX 
    10. 
  
  
  
  10.     ... 
    11. 
  
  
  
  11.     RET 
    12.

当调用 asyncPreempt2 时，会根据 preemptPark 或者 gopreempt_m 重新切换回调度循环，从而打断密集循环的继续执行。

 
 
 
    
  
  
  
  1. //go:nosplit 
    2. 
  
  
  
  2. func asyncPreempt2() { 
    3. 
  
  
  
  3.   gp := getg() 
    4. 
  
  
  
  4.   gp.asyncSafePoint = true 
    5. 
  
  
  
  5.   if gp.preemptStop { 
    6. 
  
  
  
  6.     mcall(preemptPark) 
    7. 
  
  
  
  7.   } else { 
    8. 
  
  
  
  8.     mcall(gopreempt_m) 
    9. 
  
  
  
  9.   } 
    10. 
  
  
  
  10.   // 异步抢占过程结束 
    11. 
  
  
  
  11.   gp.asyncSafePoint = false 
    12. 
  
  
  
  12. } 
    13.

至此，异步抢占过程结束。我们总结一下抢占调用的整体逻辑：

  1. M1 发送中断信号(signalM(mp, sigPreempt))
  2. M2 收到信号，操作系统中断其执行代码，并切换到信号处理函数(sighandler(signum, info, ctxt, gp))
  3. M2 修改执行的上下文，并恢复到修改后的位置(asyncPreempt)
  4. 重新进入调度循环进而调度其他 Goroutine(preemptPark 和 gopreempt_m)

上述的异步抢占流程我们是通过系统监控来说明的，正如前面所提及的，异步抢占的本质是在为垃圾回收器服务，由于我们还没有讨论过 Go 语言垃圾回收的具体细节，这里便不做过多展开，读者只需理解，在垃圾回收周期开始时，垃圾回收器将通过上述异步抢占的逻辑，停止所有用户 Goroutine，进而转去执行垃圾回收。

小结

总的来说，应用层的调度策略不易实现，因此实现上也并不是特别紧急。我们回顾 Go 语言调度策略的调整过程不难发现，实现它们的动力是从实际需求出发的。Go 语言从设计之初并没有刻意地去考虑对 Goroutine 的抢占机制。从早期无法对 Goroutine 进行抢占的原始时代，到现在的协作与抢占同时配合的调度策略，其问题的核心是垃圾回收的需要。

运行时需要执行垃圾回收时，协作式调度能够保证具备函数调用的用户 Goroutine 正常停止;抢占式调度则能避免由于死循环导致的任意时间的垃圾回收延迟。至此，Go 语言的用户可以放心地写出各种形式的代码逻辑，运行时垃圾回收也能够在适当的时候及时中断用户代码，不至于导致整个系统进入不可预测的停顿。

进一步阅读的参考文献[Clements, 2019] Austin Clements. Proposal: Non-cooperative goroutine preemption. January 18, 2019. https://github.com/golang/proposal/blob/master/design/24543-non-cooperative-preemption.md

[Clements, 2015] Austin Clements. runtime: tight loops should be preemptible](https://golang.org/issue/10958

[Chase, 2017] David Chase. cmd/compile: loop preemption with "fault branch" on amd64. May 09, 2019. https://golang.org/cl/43050

参考资料

[1]分析调度循环: https://changkun.de/golang/zh-cn/part2runtime/ch06sched/exec[2]6.6 执行栈管理: https://changkun.de/golang/zh-cn/part2runtime/ch06sched/stack/[3]垃圾回收器: https://github.com/qcrao/Go-Questions/blob/master/GC/GC.md[4]8.9 安全点分析: https://changkun.de/golang/zh-cn/part2runtime/ch08gc/safe[5]信号处理一节: https://changkun.de/golang/zh-cn/part2runtime/ch06sched/signal/

本文作者欧长坤，德国慕尼黑大学在读博士，Go/etcd/Tensorflow contributor，开源书籍《Go 语言原本》作者，《Go 夜读》SIG 成员/讲师，对 Go 有很深的研究。Github：@changkun，https://changkun.de。


名称栏目：聊一聊Go协作与抢占
当前网址：http://wtcwzsj.com/article/cdoghgh.html

其他资讯


关于我们
我们的服务
网站建设案例
新闻动态
联系我们

135-1821-9792

公司服务热线
友情链接： 服务器租用   成都小程序   崇州威斯达宇   成都橡塑保温管   成都电梯广告设计   成都服装定制   泸州柴油发电机组保养   雷运建站   德阳柴油发电机   温江做网站   

Copyright © 2009-2022 www.wtcwzsj.com 青羊区广皓图文设计工作室（个体工商户） 版权所有　蜀ICP备19037934号