GO GMP協程排程實現原理 5w字長文史上最全

 1 type schedt struct {
 2    // 用來為goroutine生成唯一id，需要以原子存取形式進行存取
 3    // 放在struct頂部，以便在32位元系統上可以進行對齊
 4    goidgen   uint64
 5 ...................
 6    lock mutex
 7     // 空閒的m組成的連結串列
 8    midle        muintptr // idle m's waiting for work
 9    // 空閒的工作執行緒數量
10    nmidle       int32    // number of idle m's waiting for work
11    // 空閒的且被lock的m的數量
12    nmidlelocked int32    // number of locked m's waiting for work
13    mnext        int64    // number of m's that have been created and next M ID
14    // 表示最多能夠建立的工作執行緒數量
15    maxmcount    int32    // maximum number of m's allowed (or die)
16    nmsys        int32    // number of system m's not counted for deadlock
17    nmfreed      int64    // cumulative number of freed m's
18     // 整個goroutine的數量，能夠自動保持更新
19    ngsys uint32 // number of system goroutines; updated atomically
20     // 由空閒的p結構體物件組成的連結串列
21    pidle      puintptr // idle p's
22    // 空閒的p結構體物件的數量
23    npidle     uint32
24    // 自旋的m的數量
25    nmspinning uint32 // See "Worker thread parking/unparking" comment in proc.go.
26 
27    // Global runnable queue.
28    // 全域性的的g的佇列
29    runq     gQueue
30    runqsize int32
31 
32    disable struct {
33       // user disables scheduling of user goroutines.
34       user     bool
35       runnable gQueue // pending runnable Gs
36       n        int32  // length of runnable
37    }
38 
39    // Global cache of dead G's.
40    // 空閒的g佇列，這裡面g的狀態為_Gdead
41    gFree struct {
42       lock    mutex
43       stack   gList // Gs with stacks
44       noStack gList // Gs without stacks
45       n       int32
46    }
47 .................
48     // 空閒的m佇列
49    freem *m
50 .....................
51     // 上次修改gomaxprocs的時間
52    procresizetime int64 // nanotime() of last change to gomaxprocs
53 ......................
54 }

 1 // runtime/runtime2.go
 2 var (
 3     // 儲存所有的m
 4    allm       *m
 5    // p的最大個數，預設等於cpu核數
 6    gomaxprocs int32
 7    // cpu的核數，程式啟動時會呼叫osinit獲取ncpu值
 8    ncpu       int32
 9    // 排程器結構體物件，記錄了排程器的工作狀態
10    sched      schedt
11    newprocs   int32
12 
13    allpLock mutex
14    // 全域性的p佇列
15    allp []*p
16 )
17 
18 // runtime/proc.go
19 var (
20     // 代表程序主執行緒的m0物件
21    m0           m
22    // m0的g0
23    g0           g
24    // 全域性的mcache物件，管理各種型別的span佇列
25    mcache0      *mcache
26    raceprocctx0 uintptr
27 )

 1 // runtime·rt0_go
 2 
 3 // 程式剛啟動的時候必定有一個執行緒啟動（主執行緒）
 4 // 將當前的棧和資源儲存在g0
 5 // 將該執行緒儲存在m0
 6 // tls: Thread Local Storage
 7 // set the per-goroutine and per-mach "registers"
 8 get_tls(BX)
 9 LEAQ    runtime·g0(SB), CX
10 MOVQ    CX, g(BX)
11 LEAQ    runtime·m0(SB), AX
12 
13 // m0和g0互相繫結
14 // save m->g0 = g0
15 MOVQ    CX, m_g0(AX)
16 // save m0 to g0->m
17 MOVQ    AX, g_m(CX)
18 // 處理args
19 CALL    runtime·args(SB) 
20 // os初始化， os_linux.go
21 CALL    runtime·osinit(SB) 
22 // 排程系統初始化, proc.go
23 CALL    runtime·schedinit(SB) 
24 
25 // 建立一個goroutine，然後開啟執行程式
26 // create a new goroutine to start program
27 MOVQ    $runtime·mainPC(SB), AX        // entry
28 PUSHQ    AX
29 PUSHQ    $0            // arg size
30 CALL    runtime·newproc(SB)
31 POPQ    AX
32 POPQ    AX
33 
34 // start this M
35 // 啟動執行緒，並且啟動排程系統
36 CALL    runtime·mstart(SB)

 1 func schedinit() {
 2 ................
 3     // g0
 4    _g_ := getg()
 5    if raceenabled {
 6       _g_.racectx, raceprocctx0 = raceinit()
 7    }
 8     // 最多啟動10000個工作執行緒
 9    sched.maxmcount = 10000
10 
11    // The world starts stopped.
12    worldStopped()
13 
14    moduledataverify()
15    // 初始化協程堆疊，包括專門分配小棧的stackpool和分配大棧的stackLarge
16    stackinit()
17    // 整個堆記憶體的初始化分配
18    mallocinit()
19    fastrandinit() // must run before mcommoninit
20    // 初始化m0
21    mcommoninit(_g_.m, -1)
22    cpuinit()       // must run before alginit
23    alginit()       // maps must not be used before this call
24    modulesinit()   // provides activeModules
25    typelinksinit() // uses maps, activeModules
26    itabsinit()     // uses activeModules
27 
28    sigsave(&_g_.m.sigmask)
29    initSigmask = _g_.m.sigmask
30 
31    if offset := unsafe.Offsetof(sched.timeToRun); offset%8 != 0 {
32       println(offset)
33       throw("sched.timeToRun not aligned to 8 bytes")
34    }
35 
36    goargs()
37    goenvs()
38    parsedebugvars()
39    gcinit()
40    // 這部分是初始化p，
41    // cpu有多少個核數就初始化多少個p
42    lock(&sched.lock)
43    sched.lastpoll = uint64(nanotime())
44    procs := ncpu
45    if n, ok := atoi32(gogetenv("GOMAXPROCS")); ok && n > 0 {
46       procs = n
47    }
48    if procresize(procs) != nil {
49       throw("unknown runnable goroutine during bootstrap")
50    }
51    unlock(&sched.lock)
52 
53    // World is effectively started now, as P's can run.
54    worldStarted()
55 }

 1 func mstart0() {
 2     // 這裡獲取的是g0在系統棧上執行
 3    _g_ := getg()
 4 .............
 5    mstart1()
 6 .............
 7 }
 8 
 9 func mstart1(dummy int32) {
10     _g_ := getg()
11     // 確保g是系統棧上的g0
12     // 排程器只在g0上執行
13     if _g_ != _g_.m.g0 {
14         throw("bad runtime·mstart")
15     }
16     ...
17     // 初始化m，主要是設定執行緒的備用訊號堆疊和訊號掩碼
18     minit()
19     // 如果當前g的m是初始m0，執行mstartm0()
20     if _g_.m == &m0 {
21         // 對於初始m，需要一些特殊處理，主要是設定系統號誌的處理常式
22         mstartm0()
23     }
24     // 如果有m的起始任務函數，則執行，比如 sysmon 函數
25     // 對於m0來說，是沒有 mstartfn 的
26     if fn := _g_.m.mstartfn; fn != nil {
27         fn()
28     }
29     if _g_.m.helpgc != 0 {
30         _g_.m.helpgc = 0
31         stopm()
32     } else if _g_.m != &m0 { // 如果不是m0，需要繫結p
33         // 繫結p
34         acquirep(_g_.m.nextp.ptr())
35         _g_.m.nextp = 0
36     }
37     // 進入排程迴圈，永不返回
38     schedule()
39 }

 1 // The main goroutine.
 2 func main() {
 3     // 獲取 main goroutine
 4     g := getg()
 5     ...
 6     // 在系統棧上執行 sysmon
 7     systemstack(func() {
 8         // 分配一個新的m，執行sysmon系統後臺監控
 9         // （定期垃圾回收和排程搶佔）
10         newm(sysmon, nil)
11     })
12     ...
13     // 確保是主執行緒
14     if g.m != &m0 {
15         throw("runtime.main not on m0")
16     }
17     // runtime 內部 init 函數的執行，編譯器動態生成的。
18     runtime_init() // must be before defer
19     ...
20     // gc 啟動一個goroutine進行gc清掃
21     gcenable()
22     ...
23     // 執行init函數，編譯器動態生成的，
24     // 包括使用者定義的所有的init函數。
25     // make an indirect call,
26     // as the linker doesn't know the address of
27     // the main package when laying down the runtime
28     fn := main_init 
29     fn()
30     ...
31     // 真正的執行main func in package main
32     // make an indirect call,
33     // as the linker doesn't know the address of
34     // the main package when laying down the runtime
35     fn = main_main 
36     fn()
37     ...
38     // 退出程式
39     exit(0)
40     // 為何這裡還需要for迴圈？
41     // 下面的for迴圈一定會導致程式崩掉，這樣就確保了程式一定會退出
42     for {
43         var x *int32
44         *x = 0
45     }
46 }

 1 func schedule() {
 2     _g_ := getg()
 3     ...
 4 top:
 5     // 如果當前GC需要停止整個世界（STW), 則呼叫gcstopm休眠當前的M
 6     if sched.gcwaiting != 0 {
 7         // 為了STW，停止當前的M
 8         gcstopm()
 9         // STW結束後回到 top
10         goto top
11     }
12     ...
13     var gp *g
14     var inheritTime bool
15     ...
16     if gp == nil {
17         // 為了防止全域性佇列中的g永遠得不到執行，所以go語言讓p每執行61排程，
18                 // 就需要優先從全域性佇列中獲取一個G到當前P中，並執行下一個要執行的G
19         if _g_.m.p.ptr().schedtick%61 == 0 && sched.runqsize > 0 {
20             lock(&sched.lock)
21             gp = globrunqget(_g_.m.p.ptr(), 1)
22             unlock(&sched.lock)
23         }
24     }
25     if gp == nil {
26         // 從p的本地佇列中獲取
27         gp, inheritTime = runqget(_g_.m.p.ptr())
28         if gp != nil && _g_.m.spinning {
29             throw("schedule: spinning with local work")
30         }
31     }
32     if gp == nil {
33         // 想盡辦法找到可執行的G，找不到就不用返回了
34         gp, inheritTime = findrunnable() // blocks until work is available
35     }
36     ...
37     // println("execute goroutine", gp.goid)
38     // 找到了g，那就執行g上的任務函數
39     execute(gp, inheritTime)
40 }

 1 func findrunnable() (gp *g, inheritTime bool) {
 2    _g_ := getg()
 3 
 4 top:
 5 ............................
 6    // 本地佇列中檢查
 7    if gp, inheritTime := runqget(_p_); gp != nil {
 8       return gp, inheritTime
 9    }
10    // 從全域性佇列中尋找
11    if sched.runqsize != 0 {
12       lock(&sched.lock)
13       gp := globrunqget(_p_, 0)
14       unlock(&sched.lock)
15       if gp != nil {
16          return gp, false
17       }
18    }
19    // 從就緒的網路協程中查詢
20    if netpollinited() && atomic.Load(&netpollWaiters) > 0 && atomic.Load64(&sched.lastpoll) != 0 {
21       if list := netpoll(0); !list.empty() { // non-blocking
22          gp := list.pop()
23          injectglist(&list)
24          casgstatus(gp, _Gwaiting, _Grunnable)
25          if trace.enabled {
26             traceGoUnpark(gp, 0)
27          }
28          return gp, false
29       }
30    }
31 
32     // 進入自旋狀態
33     procs := uint32(gomaxprocs)
34     if _g_.m.spinning || 2*atomic.Load(&sched.nmspinning) < procs-atomic.Load(&sched.npidle) {
35        if !_g_.m.spinning {
36           _g_.m.spinning = true
37           atomic.Xadd(&sched.nmspinning, 1)
38        }
39         // 從其他p的本地佇列中偷工作
40        gp, inheritTime, tnow, w, newWork := stealWork(now)
41 ..............................
42     }
43 }

 1 func Gosched() {
 2    checkTimeouts()
 3    mcall(gosched_m)
 4 }
 5 
 6 // Gosched continuation on g0.
 7 func gosched_m(gp *g) {
 8    if trace.enabled {
 9       traceGoSched()
10    }
11    goschedImpl(gp)
12 }
13 
14 func goschedImpl(gp *g) {
15    status := readgstatus(gp)
16    if status&^_Gscan != _Grunning {
17       dumpgstatus(gp)
18       throw("bad g status")
19    }
20    // 更改g的執行狀態
21    casgstatus(gp, _Grunning, _Grunnable)
22    // 接觸g和m的繫結關係
23    dropg()
24    // 將g放入全域性佇列中
25    lock(&sched.lock)
26    globrunqput(gp)
27    unlock(&sched.lock)
28 
29    schedule()
30 }

 1 func gopark(unlockf func(*g, unsafe.Pointer) bool, lock unsafe.Pointer, reason waitReason, traceEv byte, traceskip int) {
 2    if reason != waitReasonSleep {
 3       checkTimeouts() // timeouts may expire while two goroutines keep the scheduler busy
 4    }
 5    // 禁止搶佔和GC
 6    mp := acquirem()
 7    gp := mp.curg // 過去當前m上執行的g
 8    status := readgstatus(gp)
 9    if status != _Grunning && status != _Gscanrunning {
10       throw("gopark: bad g status")
11    }
12    // 設定相關的wait欄位
13    mp.waitlock = lock
14    mp.waitunlockf = unlockf
15    gp.waitreason = reason
16    mp.waittraceev = traceEv
17    mp.waittraceskip = traceskip
18    releasem(mp)
19    // can't do anything that might move the G between Ms here.
20    mcall(park_m)
21 }
22 
23 // park continuation on g0.
24 func park_m(gp *g) {
25    _g_ := getg()
26     // 變更g的狀態
27    casgstatus(gp, _Grunning, _Gwaiting)
28    // 接觸g與m的繫結關係
29    dropg()
30     // 根據被動排程不同原因，執行不同的waitunlockf函數
31    if fn := _g_.m.waitunlockf; fn != nil {
32       ok := fn(gp, _g_.m.waitlock)
33       _g_.m.waitunlockf = nil
34       _g_.m.waitlock = nil
35       if !ok {
36          if trace.enabled {
37             traceGoUnpark(gp, 2)
38          }
39          casgstatus(gp, _Gwaiting, _Grunnable)
40          execute(gp, true) // Schedule it back, never returns.
41       }
42    }
43    // 進入下一輪排程
44    schedule()
45 }

 1 func ready(gp *g, traceskip int, next bool) {
 2 ..............
 3    status := readgstatus(gp)
 4 
 5    // Mark runnable.
 6    _g_ := getg()
 7    mp := acquirem() // disable preemption because it can be holding p in a local var
 8 ...............
 9    // 變更狀態之後，放入p的區域性執行佇列中
10    casgstatus(gp, _Gwaiting, _Grunnable)
11    runqput(_g_.m.p.ptr(), gp, next)
12    wakep()
13    releasem(mp)
14 }

 1 func retake(now int64) uint32 {
 2    n := 0
 3    lock(&allpLock)
 4    // 遍歷所有的P
 5    for i := 0; i < len(allp); i++ {
 6       _p_ := allp[i]
 7       if _p_ == nil {
 8          // This can happen if procresize has grown
 9          // allp but not yet created new Ps.
10          continue
11       }
12       pd := &_p_.sysmontick
13       s := _p_.status
14       sysretake := false
15       
16       if s == _Prunning || s == _Psyscall {
17          // 判斷如果g的執行時間過長則搶佔
18          t := int64(_p_.schedtick)
19          if int64(pd.schedtick) != t {
20             pd.schedtick = uint32(t)
21             pd.schedwhen = now
22          } else if pd.schedwhen+forcePreemptNS <= now {
23              // 如果連續執行10ms則進行搶佔
24             preemptone(_p_)
25             sysretake = true
26          }
27       }
28       // 針對系統呼叫情況進行搶佔
29       // 如果p的執行佇列中有等待執行的g則搶佔
30       // 如果沒有空閒的p則進行搶佔
31       // 系統呼叫時間超過10ms則進行搶佔
32       if s == _Psyscall {
33          // Retake P from syscall if it's there for more than 1 sysmon tick (at least 20us).
34          t := int64(_p_.syscalltick)
35          if !sysretake && int64(pd.syscalltick) != t {
36             pd.syscalltick = uint32(t)
37             pd.syscallwhen = now
38             continue
39          }
40          // On the one hand we don't want to retake Ps if there is no other work to do,
41          // but on the other hand we want to retake them eventually
42          // because they can prevent the sysmon thread from deep sleep.
43          if runqempty(_p_) && atomic.Load(&sched.nmspinning)+atomic.Load(&sched.npidle) > 0 && pd.syscallwhen+10*1000*1000 > now {
44             continue
45          }
46          // Drop allpLock so we can take sched.lock.
47          unlock(&allpLock)
48          // Need to decrement number of idle locked M's
49          // (pretending that one more is running) before the CAS.
50          // Otherwise the M from which we retake can exit the syscall,
51          // increment nmidle and report deadlock.
52          incidlelocked(-1)
53          if atomic.Cas(&_p_.status, s, _Pidle) {
54             if trace.enabled {
55                traceGoSysBlock(_p_)
56                traceProcStop(_p_)
57             }
58             n++
59             _p_.syscalltick++
60             handoffp(_p_)
61          }
62          incidlelocked(1)
63          lock(&allpLock)
64       }
65    }
66    unlock(&allpLock)
67    return uint32(n)
68 }