點(diǎn)擊上方藍(lán)色“Go語言中文網(wǎng)”關(guān)注我們，領(lǐng)全套Go資料，每天學(xué)習(xí)?Go?語言

本文作者：葉不聞

原文鏈接：https://juejin.im/post/5dafc241f265da5ba95c465d

golang 調(diào)度模型

模型總攬

核心實(shí)體

Goroutines (G)

golang 調(diào)度單元，golang 可以開啟成千上萬個(gè) g，每個(gè) g 可以理解為一個(gè)任務(wù)，等待被調(diào)度。其存儲(chǔ)了 goroutine 的執(zhí)行 stack 信息、goroutine 狀態(tài)以及 goroutine 的任務(wù)函數(shù)等。g 只能感知到 p，下文說的 m 對(duì)其透明的。

OSThread (M)

系統(tǒng)線程，實(shí)際執(zhí)行 g 的狠角色，但 m 并不維護(hù) g 的狀態(tài)，一切都是由幕后黑手 p 來控制。

Processor (P)

維護(hù) m 執(zhí)行時(shí)所需要的上下文，p 的個(gè)數(shù)通常和 cpu 核數(shù)一致(可以設(shè)置)，代表 gorotine 的并發(fā)度。其維護(hù)了 g 的隊(duì)列。

實(shí)體間的關(guān)系

一圖勝千言，直接看這個(gè)經(jīng)典的圖

調(diào)度本質(zhì)

即 schedule 函數(shù)，通過調(diào)度，放棄目前執(zhí)行的 g，選擇一個(gè) g 來執(zhí)行。選擇算法不是本文重點(diǎn)，這里不做過多講述。

切換時(shí)機(jī)

• 會(huì)阻塞的系統(tǒng)調(diào)用，比如文件 io，網(wǎng)絡(luò) io；
• time 系列定時(shí)操作；
• go func 的時(shí)候, func 執(zhí)行完的時(shí)候；
• 管道讀寫阻塞的情況；
• 垃圾回收之后。
• 主動(dòng)調(diào)用 runtime.Gosched()

調(diào)度時(shí)機(jī)分析

阻塞性系統(tǒng)調(diào)用

系統(tǒng)調(diào)用，如 read，golang 重寫了所有系統(tǒng)調(diào)用，在系統(tǒng)調(diào)用加入了調(diào)度邏輯。

拿 read 舉例

/usr/local/go/src/os/file.go:97

// Read reads up to len(b) bytes from the File.
// It returns the number of bytes read and an error, if any.
// EOF is signaled by a zero count with err set to io.EOF.
func (f *File) Read(b []byte) (n int, err error) {
if f == nil {
return 0, ErrInvalid
}
n, e := f.read(b)
if n == 0 && len(b) > 0 && e == nil {
return 0, io.EOF
}
if e != nil {
err = &PathError{"read", f.name, e}
}
return n, err
}

嵌套到幾層，就不全部貼出來，跟到底是如下函數(shù)：

func read(fd int, p []byte) (n int, err error) {
var _p0 unsafe.Pointer
if len(p) > 0 {
_p0 = unsafe.Pointer(&p[0])
} else {
_p0 = unsafe.Pointer(&_zero)
}
r0, _, e1 := Syscall(SYS_READ, uintptr(fd), uintptr(_p0), uintptr(len(p)))
n = int(r0)
if e1 != 0 {
err = errnoErr(e1)
}
return
}

func Syscall(trap, a1, a2, a3 uintptr) (r1, r2 uintptr, err Errno)

Syscall 是匯編實(shí)現(xiàn)

TEXT ·Syscall(SB),NOSPLIT,$0-56
BL runtime·entersyscall(SB)
MOVD a1+8(FP), R3
MOVD a2+16(FP), R4
MOVD a3+24(FP), R5
MOVD R0, R6
MOVD R0, R7
MOVD R0, R8
MOVD trap+0(FP), R9 // syscall entry
SYSCALL R9
BVC ok
MOVD $-1, R4
MOVD R4, r1+32(FP) // r1
MOVD R0, r2+40(FP) // r2
MOVD R3, err+48(FP) // errno
BL runtime·exitsyscall(SB)
RET
ok:
MOVD R3, r1+32(FP) // r1
MOVD R4, r2+40(FP) // r2
MOVD R0, err+48(FP) // errno
BL runtime·exitsyscall(SB)
RET

可以看到，進(jìn)入系統(tǒng)調(diào)用時(shí)，是調(diào)用 entersyscall，當(dāng)離開系統(tǒng)調(diào)用，會(huì)運(yùn)行 exitsyscall

// Standard syscall entry used by the go syscall library and normal cgo calls.
//go:nosplit
func entersyscall(dummy int32) {
reentersyscall(getcallerpc(unsafe.Pointer(&dummy)), getcallersp(unsafe.Pointer(&dummy)))
}

func reentersyscall(pc, sp uintptr) {
_g_ := getg()

// Disable preemption because during this function g is in Gsyscall status,
// but can have inconsistent g->sched, do not let GC observe it.
_g_.m.locks++

// Entersyscall must not call any function that might split/grow the stack.
// (See details in comment above.)
// Catch calls that might, by replacing the stack guard with something that
// will trip any stack check and leaving a flag to tell newstack to die.
_g_.stackguard0 = stackPreempt
_g_.throwsplit = true

// Leave SP around for GC and traceback.
save(pc, sp)
_g_.syscallsp = sp
_g_.syscallpc = pc
casgstatus(_g_, _Grunning, _Gsyscall)
if _g_.syscallsp < _g_.stack.lo || _g_.stack.hi < _g_.syscallsp {
systemstack(func() {
print("entersyscall inconsistent ", hex(_g_.syscallsp), " [", hex(_g_.stack.lo), ",", hex(_g_.stack.hi), "]\n")
throw("entersyscall")
})
}

if trace.enabled {
systemstack(traceGoSysCall)
// systemstack itself clobbers g.sched.{pc,sp} and we might
// need them later when the G is genuinely blocked in a
// syscall
save(pc, sp)
}

if atomic.Load(&sched.sysmonwait) != 0 { // TODO: fast atomic
systemstack(entersyscall_sysmon)
save(pc, sp)
}

if _g_.m.p.ptr().runSafePointFn != 0 {
// runSafePointFn may stack split if run on this stack
systemstack(runSafePointFn)
save(pc, sp)
}

_g_.m.syscalltick = _g_.m.p.ptr().syscalltick
_g_.sysblocktraced = true
_g_.m.mcache = nil
_g_.m.p.ptr().m = 0
atomic.Store(&_g_.m.p.ptr().status, _Psyscall)
if sched.gcwaiting != 0 {
systemstack(entersyscall_gcwait)
save(pc, sp)
}

// Goroutines must not split stacks in Gsyscall status (it would corrupt g->sched).
// We set _StackGuard to StackPreempt so that first split stack check calls morestack.
// Morestack detects this case and throws.
_g_.stackguard0 = stackPreempt
_g_.m.locks--
}

進(jìn)入系統(tǒng)調(diào)用時(shí)，p 和 m 分離，當(dāng)前運(yùn)行的 g 狀態(tài)變?yōu)?_Gsyscall。

_Gsyscall 恢復(fù)時(shí)機(jī)：

當(dāng) m 執(zhí)行完，調(diào)用 exitsyscall 重新和之前的 p 綁定，其中調(diào)度的還是 schedule 函數(shù)；

sysmon 線程，發(fā)現(xiàn)該 p 一定時(shí)間沒有執(zhí)行，會(huì)其分配一個(gè)新的 m。此時(shí)進(jìn)入調(diào)度。

time 定時(shí)類操作

都拿 time.Sleep 舉例

// Sleep pauses the current goroutine for at least the duration d.
// A negative or zero duration causes Sleep to return immediately.
func Sleep(d Duration)
實(shí)際定義在runtime
// timeSleep puts the current goroutine to sleep for at least ns nanoseconds.
//go:linkname timeSleep time.Sleepfunc timeSleep(ns int64) {
if ns <= 0 {
return
}

t := getg().timer
if t == nil {
t = new(timer)
getg().timer = t
}
*t = timer{}
t.when = nanotime() + ns
t.f = goroutineReady
t.arg = getg()
lock(&timers.lock)
addtimerLocked(t)
goparkunlock(&timers.lock, "sleep", traceEvGoSleep, 2)
}

goparkunlock 最終調(diào)用 gopark

func gopark(unlockf func(*g, unsafe.Pointer) bool, lock unsafe.Pointer, reason string, traceEv byte, traceskip int) {
mp := acquirem()
gp := mp.curg
status := readgstatus(gp)
if status != _Grunning && status != _Gscanrunning {
throw("gopark: bad g status")
}
mp.waitlock = lock
mp.waitunlockf = *(*unsafe.Pointer)(unsafe.Pointer(&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)
}

mcall(fn) 是切換到 g0，讓 g0 來調(diào)用 fn，這里我們看下 park_m 定義 park_m

func park_m(gp *g) {mcall(park_m)
_g_ := getg()

if trace.enabled {
traceGoPark(_g_.m.waittraceev, _g_.m.waittraceskip)
}

casgstatus(gp, _Grunning, _Gwaiting)
dropg()

if _g_.m.waitunlockf != nil {
fn := *(*func(*g, unsafe.Pointer) bool)(unsafe.Pointer(&_g_.m.waitunlockf))ok := fn(gp, _g_.m.waitlock)
_g_.m.waitunlockf = nil
_g_.m.waitlock = nilif !ok {
if trace.enabled {
traceGoUnpark(gp, 2)
}
casgstatus(gp, _Gwaiting, _Grunnable)
execute(gp, true) // Schedule it back, never returns.
}
}
schedule()
}

可以看到，先把狀態(tài)轉(zhuǎn)化為 _Gwaiting, 再進(jìn)行了一次 schedule 針對(duì) _Gwaiting 的 g，需要調(diào)用 goready,才能恢復(fù)。

新起一個(gè)協(xié)程和退出

新開一個(gè)協(xié)程，g 狀態(tài)會(huì)變?yōu)?_GIdle，觸發(fā)調(diào)度。當(dāng)協(xié)程執(zhí)行完，會(huì)調(diào)用 goexit1 此時(shí)狀態(tài)變?yōu)?_GDead，_Gdead 可以被復(fù)用，或者被 gc 清除。

管道阻塞

chansend 即 c 的實(shí)現(xiàn)

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
if c == nil {
if !block {
returnfalse
}
gopark(nil, nil, "chan send (nil chan)", traceEvGoStop, 2)
throw("unreachable")
}

if debugChan {
print("chansend: chan=", c, "\n")
}

if raceenabled {
racereadpc(unsafe.Pointer(c), callerpc, funcPC(chansend))
}

........
// 省略無關(guān)代碼
........

// Block on the channel. Some receiver will complete our operation for us.
gp := getg()
mysg := acquireSudog()
mysg.releasetime = 0
if t0 != 0 {
mysg.releasetime = -1
}
// No stack splits between assigning elem and enqueuing mysg
// on gp.waiting where copystack can find it.
mysg.elem = ep
mysg.waitlink = nil
mysg.g = gp
mysg.selectdone = nil
mysg.c = c
gp.waiting = mysg
gp.param = nil
c.sendq.enqueue(mysg)
goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)

// someone woke us up.
if mysg != gp.waiting {
throw("G waiting list is corrupted")
}
gp.waiting = nil
if gp.param == nil {
if c.closed == 0 {
throw("chansend: spurious wakeup")
}
panic(plainError("send on closed channel"))
}
gp.param = nil
if mysg.releasetime > 0 {
blockevent(mysg.releasetime-t0, 2)
}
mysg.c = nil
releaseSudog(mysg)
returntrue
}

可以看到，實(shí)際還是調(diào)用 goparkunlock->gopark，來進(jìn)行調(diào)度。

gc 之后

stw 之后，會(huì)重新選擇 g 開始執(zhí)行。此處不對(duì)垃圾回收做過多擴(kuò)展。

主動(dòng)調(diào)用 runtime.Gosched()

沒有找到非要調(diào)用 runtime.Gosched 的場(chǎng)景，主要作用還是為了調(diào)試，學(xué)習(xí) runtime 吧

// Gosched yields the processor, allowing other goroutines to run. It does not
// suspend the current goroutine, so execution resumes automatically.
//go:nosplit
func Gosched() {
mcall(gosched_m)
}

第一步就將環(huán)境切換到 g0，然后執(zhí)行一個(gè)叫 gosched_m 的函數(shù)

// Gosched continuation on g0.
func gosched_m(gp *g) {
if trace.enabled {
traceGoSched()
}
goschedImpl(gp)
}

func goschedImpl(gp *g) {
status := readgstatus(gp)
if status&^_Gscan != _Grunning {
dumpgstatus(gp)
throw("bad g status")
}
casgstatus(gp, _Grunning, _Grunnable)
dropg()
lock(&sched.lock)
globrunqput(gp)
unlock(&sched.lock)

schedule()
}

可以看到，當(dāng)前 g 被設(shè)置為 _Grunnable，放入執(zhí)行隊(duì)列。然后調(diào)用 schedule，選擇一個(gè)合適的 g 進(jìn)行執(zhí)行。

總結(jié)

golang 協(xié)程調(diào)度時(shí)機(jī)主要是阻塞性操作開始，結(jié)束。研究每個(gè)場(chǎng)景相關(guān)代碼，即可對(duì) golang 有更深的理解。這里也分享一個(gè)閱讀源碼的小經(jīng)驗(yàn)，每次帶著一個(gè)特定問題去尋找答案，比如本文的調(diào)度時(shí)機(jī)，后面再看調(diào)度算法，垃圾回收，這樣每次能忽略無關(guān)因素，通過多個(gè)不同的主題，整個(gè)框架會(huì)越來越完善。

參考文章

A complete journey with Goroutines: https://medium.com/@riteeksrivastava/a-complete-journey-with-goroutines-8472630c7f5c

Go's work-stealing scheduler: https://rakyll.org/scheduler/

推薦閱讀

• xxx

---

喜歡本文的朋友，歡迎關(guān)注“Go語言中文網(wǎng)”：

Go語言中文網(wǎng)啟用微信學(xué)習(xí)交流群，歡迎加微信：274768166

總結(jié)

以上是生活随笔為你收集整理的golang var 初始化时机_你应该知道的 Go 调度器知识：Go 核心原理 — 协程调度时机...的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。