快捷導(dǎo)航

詳解Go如何實現(xiàn)協(xié)程并發(fā)執(zhí)行

更新時間：2023年08月28日 11:13:40 作者：unitiny

線程是通過本地隊列,全局隊列或者偷其它線程的方式來獲取協(xié)程的,目前看來,線程運行完一個協(xié)程后再從隊列中獲取下一個協(xié)程執(zhí)行,還只是順序執(zhí)行協(xié)程的,而多個線程一起這么運行也能達到并發(fā)的效果,接下來就給給大家詳細介紹一下Go如何實現(xiàn)協(xié)程并發(fā)執(zhí)行

順序執(zhí)行有什么問題

很明顯，順序執(zhí)行會造成協(xié)程的饑餓問題。如果某個大協(xié)程掛在線程中運行了十分鐘，那么隊列中其它協(xié)程就一直處于休眠中無法運行，這不公平。如果讓某些實時性強的協(xié)程饑餓，得不到cpu運行，會影響業(yè)務(wù)。比如視頻彈幕，用戶發(fā)出一條彈幕，得盡快顯示在視頻中。若此時協(xié)程饑餓，得不到處理，用戶體驗就差了。

該如何解決呢？簡單，讓大協(xié)程切換出去就可以了。

協(xié)程切換

回到線程循環(huán)這張圖中（在深入考究協(xié)程一文中有解釋），業(yè)務(wù)方法這塊即線程執(zhí)行的協(xié)程。如果業(yè)務(wù)方法運行時間過長，則觸發(fā)協(xié)程切換。

對協(xié)程：保存該協(xié)程運行的情況，然后將該協(xié)程放入本地隊列隊尾，休眠該協(xié)程。
對線程：從業(yè)務(wù)方法中跳出，重新執(zhí)行 schedule 方法，之后會從本地隊列中獲取一個新的協(xié)程運行。

但這樣只是本地隊列的協(xié)程切換，全局隊列的協(xié)程仍會饑餓，該如何解決呢？

隨機抽取全局協(xié)程

在線程循環(huán)的 shedule 的 findRunnable 函數(shù)中，每隔一段時間就會從全局隊列中獲取一個協(xié)程放到本地隊列，再通過本地隊列的協(xié)程切換，使得來自全局隊列的協(xié)程有機會運行，從而解決全局隊列協(xié)程的饑餓問題。來看下源碼：

if pp.schedtick%61 == 0 && sched.runqsize > 0 {
   lock(&sched.lock)
   gp := globrunqget(pp, 1)
   unlock(&sched.lock)
   if gp != nil {
      return gp, false, false
   }
}

pp.schedtick 表示線程循環(huán)的次數(shù)，如果達到61的倍數(shù)，就執(zhí)行 globrunqget ，從全局隊列中獲取協(xié)程。

協(xié)程如何并發(fā)執(zhí)行

從以上可得知，線程通過切換協(xié)程的方式，不再順序的執(zhí)行協(xié)程了，從而達到并發(fā)執(zhí)行協(xié)程的效果。這關(guān)鍵在于協(xié)程的切換，那協(xié)程在什么時候會切換呢？

協(xié)程切換時機

協(xié)程的切換時機如下：

主動掛起，調(diào)用 gopark 函數(shù)，使協(xié)程主動休眠等待
系統(tǒng)調(diào)用完成后，io操作耗時，因此切換協(xié)程
基于協(xié)作的搶占式調(diào)度，協(xié)程在跳轉(zhuǎn)到其它方法時，就把自己切換出去
基于信號的搶占式調(diào)度，通過發(fā)送信號，觸發(fā)線程的調(diào)度方法

主動掛起

協(xié)程可以調(diào)用 runtime.gopark 方法，使自己陷入休眠。

源碼如下：

// 將當前協(xié)程置于等待狀態(tài)
func gopark(unlockf func(*g, unsafe.Pointer) bool, lock unsafe.Pointer, reason waitReason, traceEv byte, traceskip int) {
   if reason != waitReasonSleep {
      checkTimeouts() // timeouts may expire while two goroutines keep the scheduler busy
   }
   mp := acquirem()
   gp := mp.curg
   status := readgstatus(gp)
   if status != _Grunning && status != _Gscanrunning {
      throw("gopark: bad g status")
   }
   mp.waitlock = lock
   mp.waitunlockf = unlockf
   gp.waitreason = reason
   mp.waittraceev = traceEv
   mp.waittraceskip = traceskip
   releasem(mp)
   // can't do anything that might move the G between Ms here.
   mcall(park_m)
}

可以看到：

gopark 中通過 acquirem 獲取到當前的線程指針mp
通過mp獲取到當前運行的協(xié)程指針gp
給mp，gp的一些字段賦值，修改狀態(tài)
然后調(diào)用 mcall ， mcall 是一個匯編方法，作用時切換到g0棧，并執(zhí)行傳入的函數(shù)。這里執(zhí)行 park_m 函數(shù)，最終跳轉(zhuǎn)到 schedule 方法，也就是線程循環(huán)的開頭，實現(xiàn)了協(xié)程的主動切換。

// park_m函數(shù)最終跳轉(zhuǎn)到schedule
func park_m(gp *g) {
   mp := getg().m
   ...
   schedule()
}

由于gopark是小寫開頭的，外部無法調(diào)用。我們在使用 time.Sleep ， sync.WaitGroup 時，會間接的使用到gopark，將協(xié)程休眠。

系統(tǒng)調(diào)用完成后

當協(xié)程要執(zhí)行讀寫文件、網(wǎng)絡(luò) IO、進程間通信等系統(tǒng)調(diào)用的操作時，會進入 entersyscall 函數(shù)，將該協(xié)程暫停并放入等待隊列。

當系統(tǒng)調(diào)用完成后，由于io操作都比較耗時，說明該協(xié)程已經(jīng)運行了挺長一段時間了，因此將協(xié)程掛起，切換另一個協(xié)程執(zhí)行很合理。

而 exitsyscall 也位于runtime中，源碼部分如下：

func exitsyscall() {
   gp := getg()
   ...
   mcall(exitsyscall0)
   ...
}

又是熟悉的 mcall ，mcall執(zhí)行了 exitsyscall0 函數(shù)，最終跳轉(zhuǎn)到線程循環(huán)開頭的 schedule 函數(shù)，完成協(xié)程切換。

基于協(xié)作的搶占式調(diào)度

如果協(xié)程既不主動掛起，也沒有進行系統(tǒng)調(diào)用呢，那就一直切換不出去了？該怎么解決呢，如果每個協(xié)程都經(jīng)常調(diào)用同一個方法的話，那就可以在這個方法里加入一個鉤子，讓這個協(xié)程切換出去。

思路有了，具體找哪個方法呢？這里做一個演示。

package main
import (
   "fmt"
   "time"
)
func do1() {
   do2()
}
func do2() {
   do3()
}
func do3() {
   fmt.Println("do3")
}
func main() {
   go do1()
   time.Sleep(time.Hour)
}

以上代碼開啟一個do1協(xié)程，do1調(diào)用do2，do2調(diào)用do3。我們通過 go build -gcflags -S main.go 命令，查看匯編代碼，發(fā)現(xiàn)多次調(diào)用到了 runtime.morestack_noctxt 方法。在函數(shù)跳轉(zhuǎn)的時候，編譯器會插入 runtime.morestack_noctxt 這個方法。目的是檢查函數(shù)棧空間是否足夠。簡略源碼如下：

TEXT runtime·morestack_noctxt(SB),NOSPLIT,$0
   MOVL   $0, DX
   JMP    runtime·morestack(SB)

TEXT runtime·morestack(SB),NOSPLIT|NOFRAME,$0-0
   ...
   BL runtime·newstack(SB)
   ...

最終調(diào)用到 newstack 這個go方法。

現(xiàn)在對于運行時間超過10ms的大協(xié)程，其 g.stackguard0 會被賦值為 stackPreempt ，意味著該協(xié)程要切換出去了。

stackPreempt值為 0xfffffade

// 0xfffffade in hex.
const stackPreempt = uintptrMask & -1314

于是在 newstack 方法中會判斷 g.stackguard0 是否為 stackPreempt ，是則將該協(xié)程切換出去。

func newstack() {
        // 判斷是否有搶占信號
        preempt := stackguard0 == stackPreempt
        ...
	if preempt {
		...
		// Act like goroutine called runtime.Gosched.
		gopreempt_m(gp) // never return
	}
        ...
}

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

func goschedImpl(gp *g) {
   ...
   schedule()
}

以上流程總結(jié)來說：

Go對大協(xié)程會把g.stackguard0標記為stackPreempt。
在大協(xié)程調(diào)用其它函數(shù)時，會調(diào)用newstack判斷棧空間，順便判斷該協(xié)程是否要切換出去。
要切換則進入gopreempt_m -> goschedImpl -> schedule，最終回到線程循環(huán)的開頭。

流程圖如下：

基于信號的搶占式調(diào)度

如果協(xié)程不主動掛起，不系統(tǒng)調(diào)用，不調(diào)用其它函數(shù)，只是純計算的任務(wù)，那該如何切換呢？如下：

go func() {
   i := 0
   for {
      i++
   }
}()

Go就利用了操作系統(tǒng)通信的方式，通過GC的線程向該協(xié)程對應(yīng)的線程發(fā)送信號，觸發(fā)該線程的切換方法。具體步驟為：

注冊 SIGURG 信號的處理函數(shù)
GC線程工作時，向該目標線程發(fā)送信號
線程接收信號后，觸發(fā)調(diào)度方法

流程圖如下：

源碼分析：

線程接收到操作系統(tǒng)信號，進入 sighandler 方法，識別信號為SIGURG，進入 doSigPreempt 方法。之后流程：doSigPreempt -> asyncPreempt -> asyncPreempt2 -> mcall -> gopreempt_m -> goschedImpl。最終調(diào)用schedule方法，回到線程開頭，完成協(xié)程切換。

具體細節(jié)各位可以動手查看下，感悟更多。