源码
- 版本
- 1.14.1
- 相关目录
- runtime/asm_amd64.s
- runtime/proc.go
- runtime/runtime2.go
关键概念
- G - 我们代码写的go func(){ }
- M - 内核线程
- P - M调度G的上下文, P中存储了很多G,M通过调用P来获取并执行G。为了方便,下文中称它为==局部调度器==
- schedt - 全局调度器，主要存储了一些空闲的G、M、P
G、M、P、schedt之间的关系
graph TB;
subgraph schedt
A(空闲G集合)
B(runable G集合)
C(空闲P集合)
D(空闲M集合)
end
graph TB;
subgraph M
A(running G)
end
subgraph P
B(空闲G集合)
C(runable G集合)
end
A -- M执行完G,放入空闲集合 --> B
A -- M从P获取可运行G --> C
D[M] -.-> E[P]
D2[M] -.-> E2[P]
D3[M] -.-> E3[P]
GMP.png
关键数据结构
G
/// runtime/runtime2.go 关键字段
type g struct {
stack stack // g自己的栈
m *m // 执行当前g的m
sched gobuf // 保存了g的现场，goroutine切换时通过它来恢复
atomicstatus uint32 // g的状态Gidle,Grunnable,Grunning,Gsyscall,Gwaiting,Gdead
goid int64
schedlink guintptr // 下一个g, g链表
preempt bool //抢占标记
lockedm muintptr // 锁定的M,g中断恢复指定M执行
gopc uintptr // 创建该goroutine的指令地址
startpc uintptr // goroutine 函数的指令地址
}
这里先介绍下G的各个状态、
- Gidle 被创建但没初始换
- Grunnable 可运行
- Grunning 正在运行
- Gsyscall 正在系统调用
- Gwaiting 正在等待
- Gdead 运行完成
M
/// runtime/runtime2.go 关键字段
type m struct {
g0 *g // g0, 每个M都有自己独有的g0
curg *g // 当前正在运行的g
p puintptr // 当前用于的p
nextp puintptr // 当m被唤醒时，首先拥有这个p
id int64
spinning bool // 是否处于自旋
park note
alllink *m // on allm
schedlink muintptr // 下一个m, m链表
mcache *mcache // 内存分配
lockedg guintptr // 和 G 的lockedm对应
freelink *m // on sched.freem
}
P
/// runtime/runtime2.go 关键字段
type p struct {
id int32
status uint32 // 状态
link puintptr // 下一个P, P链表
m muintptr // 拥有这个P的M
mcache *mcache
// P本地runnable状态的G队列
runqhead uint32
runqtail uint32
runq [256]guintptr
runnext guintptr // 一个比runq优先级更高的runnable G
// 状态为dead的G链表，在获取G时会从这里面获取
gFree struct {
gList
n int32
}
gcBgMarkWorker guintptr // (atomic)
gcw gcWork
}
这里先介绍下P的各个状态
- Pidle：没有关联的M
- Prunning：已和某个M关联
- Psyscall：当前P中的被运行的那个G正在进行系统调用
- Pgcstop： 系统正在GC
- Pdead: 当前P不再使用
schedt
/// runtime/runtime2.go 关键字段
type schedt struct {
lock mutex
midle muintptr // 空闲M链表
nmidle int32 // 空闲M数量
nmidlelocked int32 // 被锁住的M的数量
mnext int64 // 已创建M的数量，以及下一个M ID
maxmcount int32 // 允许创建最大的M数量
nmsys int32 // 不计入死锁的M数量
nmfreed int64 // 累计释放M的数量
pidle puintptr // 空闲的P链表
npidle uint32 // 空闲的P数量
runq gQueue // 全局runnable的G队列
runqsize int32 // 全局runnable的G数量
// Global cache of dead G's.
gFree struct {
lock mutex
stack gList // Gs with stacks
noStack gList // Gs without stacks
n int32
}
// freem is the list of m's waiting to be freed when their
// m.exited is set. Linked through m.freelink.
freem *m
}
启动
/// runtime/asm_amd64.s
get_tls(BX)
LEAQ runtime·g0(SB), CX
MOVQ CX, g(BX)
LEAQ runtime·m0(SB), AX
// save m->g0 = g0
MOVQ CX, m_g0(AX)
// save m0 to g0->m
MOVQ AX, g_m(CX)
CLD // convention is D is always left cleared
CALL runtime·check(SB)
MOVL 16(SP), AX // copy argc
MOVL AX, 0(SP)
MOVQ 24(SP), AX // copy argv
MOVQ AX, 8(SP)
CALL runtime·args(SB) // 命令行初始化,获取参数
CALL runtime·osinit(SB) // OS初始化
CALL runtime·schedinit(SB)
// create a new goroutine to start program
MOVQ $runtime·mainPC(SB), AX // entry 对应runtime·main
PUSHQ AX
PUSHQ $0 // arg size
CALL runtime·newproc(SB)
POPQ AX
POPQ AX
// start this M
CALL runtime·mstart(SB)
CALL runtime·abort(SB) // mstart should never return
....
DATA runtime·mainPC+0(SB)/8,$runtime·main(SB)
汇编语言我也看不懂，看字面意思可知道启动流程：
1.创建g0
2.创建m0
3.m.g0 = g0
4.g0.m = m0
5.命令行初始化，OS初始化
6.schedinit 调度器初始化
7.newproc 将runtime.main作为参数创建goroutine
8.mstart
所以g0和m0是通过汇编指令创建的，并将m0赋值给g0.m
schedinit 调度器初始化
///runtime/proc.go
func schedinit() {
// 获取g0
_g_ := getg()
...
// 设置最大 M 数量
sched.maxmcount = 10000
...
// 栈和内存初始化
stackinit()
mallocinit()
...
// 初始化当前 M
mcommoninit(_g_.m)
...
//参数和环境初始化
goargs()
goenvs()
...
// 设置 P 的数量
procs := ncpu
// 通过环境变量设置P的数量
if n, ok := atoi32(gogetenv("GOMAXPROCS")); ok && n > 0 {
procs = n
}
...
}
调度器初始化的主要工作： 空间申请、M的最大数量设置、P的数量设置、初始化参数和环境
newproc 创建goroutine
func newproc(siz int32, fn *funcval) {
// sys.PtrSize = 8, 表示跳过函数指针， 获取第一个参数的地址
argp := add(unsafe.Pointer(&fn), sys.PtrSize)
gp := getg() // 获取当前g
pc := getcallerpc() // 获取下一条要执行的指令地址
// 用 g0 的栈创建 G
// systemstack 会切换当前的 g 到 g0, 并且使用g0的栈空间
systemstack(func() {
newproc1(fn, argp, siz, gp, pc)
})
}
func newproc1(fn *funcval, argp unsafe.Pointer, narg int32, callergp *g, callerpc uintptr) {
...
_p_ := _g_.m.p.ptr() // 获取P
newg := gfget(_p_) // 在P或者sched中获取空闲的G, 在这里也就是主goroutine
if newg == nil { // 获取失败就创建一个新的
newg = malg(_StackMin)
casgstatus(newg, _Gidle, _Gdead)
allgadd(newg) // publishes with a g->status of Gdead so GC scanner doesn't look at uninitialized stack.
}
...
// 将runtime.main地址存储在主goroutine的sched中
gostartcallfn(&newg.sched, fn)
...
runqput(_p_, newg, true) // 将runnable的newg放入P中
...
}
这里先不分析具体实现，在启动阶段将runtime.main作为入参创建G,也就是创建一个G来运行runtime.main。
runtime.main
func main() {
...
// 64位系统 栈的最大空间为 1G, 32为系统 为 250M
if sys.PtrSize == 8 {
maxstacksize = 1000000000
} else {
maxstacksize = 250000000
}
...
fn := main_main // 这就是我们代码main包的main函数
fn() // 运行我们的main函数
...
}
这里就可以调用我们main.main了，所以上面的newproc是为了给我们的main.main创建一个主goroutine。
现在有了主goroutine，那怎么启动这个goroutine呢？-- mstart
mstart
func mstart() {
...
mstart1()
...
}
func mstart1() {
...
schedule()
}
func schedule() {
_g_ := getg()
...
top:
pp := _g_.m.p.ptr()
...
var gp *g
...
// 从sched或者P或获取G,启动阶段至此一个产生了两个G:g0和main的G。g0不会存在sched和P中，所以这里获取的是main的G
if gp == nil {
if _g_.m.p.ptr().schedtick%61 == 0 && sched.runqsize > 0 {
lock(&sched.lock)
gp = globrunqget(_g_.m.p.ptr(), 1)
unlock(&sched.lock)
}
}
if gp == nil {
gp, inheritTime = runqget(_g_.m.p.ptr())
}
if gp == nil {
gp, inheritTime = findrunnable() // blocks until work is available
}
execute(gp, inheritTime)
}
func execute(gp *g, inheritTime bool) {
...
// 在上面我们已经将runtime.main的地址存在gp.sched中。这里就调用runtime.main。
gogo(&gp.sched)
}
mstart经过一系列的调用，最终通过gogo(&gp.sched)调用了runtime.main。在runtime.main中又调用了mian.main，至此启动结束。
启动小结
graph TB;
A["创建G0、M0, g0.m = m0"] --> C
C[初始化命令行和OS]-->D
D["schedinit:设置M最大数量、P个数、栈和内存初始化"] --> E
E["newproc:为main.main创建一个主goroutine"] --> F
F["mstart:运行主goroutine --> 运行main.main"]
启动.png
==下面开始分析G、M、P==
G 创建
func newproc(siz int32, fn *funcval) {
// sys.PtrSize = 8, 表示跳过函数指针， 获取第一个参数的地址
argp := add(unsafe.Pointer(&fn), sys.PtrSize)
gp := getg() // 获取当前g
pc := getcallerpc() // 获取下一条要执行的指令地址
// 用 g0 的栈创建 G
// systemstack 会切换当前的 g 到 g0, 并且使用g0的栈空间
systemstack(func() {
newproc1(fn, argp, siz, gp, pc)
})
}
func newproc1(fn *funcval, argp unsafe.Pointer, narg int32, callergp *g, callerpc uintptr) {
_g_ := getg() //获取当前g,也就是g0。因为上面的systemstack会切换到g0
if fn == nil { // fn空报错
_g_.m.throwing = -1 // do not dump full stacks
throw("go of nil func value")
}
acquirem() // disable preemption because it can be holding p in a local var
siz := narg
siz = (siz + 7) &^ 7
// 参数大小不能大于 2048 - 4 * 8 - 8 = 2000
if siz >= _StackMin-4*sys.RegSize-sys.RegSize {
throw("newproc: function arguments too large for new goroutine")
}
_p_ := _g_.m.p.ptr()
newg := gfget(_p_) // 在P中获取G
// 如果没获取到则创建一个新的G,并设置成dead状态，加入全局的allgs
if newg == nil {
newg = malg(_StackMin)
casgstatus(newg, _Gidle, _Gdead)
allgadd(newg) // publishes with a g->status of Gdead so GC scanner doesn't look at uninitialized stack.
}
if newg.stack.hi == 0 {
throw("newproc1: newg missing stack")
}
if readgstatus(newg) != _Gdead {
throw("newproc1: new g is not Gdead")
}
totalSize := 4*sys.RegSize + uintptr(siz) + sys.MinFrameSize // extra space in case of reads slightly beyond frame
totalSize += -totalSize & (sys.SpAlign - 1) // align to spAlign
sp := newg.stack.hi - totalSize // 栈顶地址
spArg := sp
if usesLR {
// caller's LR
*(*uintptr)(unsafe.Pointer(sp)) = 0
prepGoExitFrame(sp)
spArg += sys.MinFrameSize
}
if narg > 0 {
// 将参数压入G的栈
memmove(unsafe.Pointer(spArg), argp, uintptr(narg))
if writeBarrier.needed && !_g_.m.curg.gcscandone {
f := findfunc(fn.fn)
stkmap := (*stackmap)(funcdata(f, _FUNCDATA_ArgsPointerMaps))
if stkmap.nbit > 0 {
// We're in the prologue, so it's always stack map index 0.
bv := stackmapdata(stkmap, 0)
bulkBarrierBitmap(spArg, spArg, uintptr(bv.n)*sys.PtrSize, 0, bv.bytedata)
}
}
}
// 清除G的运行现场，因为G有可能是从P中获取的，清除原有的数据
memclrNoHeapPointers(unsafe.Pointer(&newg.sched), unsafe.Sizeof(newg.sched))
// 重新对现场复制
newg.sched.sp = sp
newg.stktopsp = sp
// 将pc指向goexit。这个很重要，G运行完时会执行它，实际将调用goexit1
newg.sched.pc = funcPC(goexit) + sys.PCQuantum
newg.sched.g = guintptr(unsafe.Pointer(newg))
// 其实在这里面真正pc指向的是fn, 而上面的goexit被用于sp,当RET的时候pop出goexit
gostartcallfn(&newg.sched, fn)
newg.gopc = callerpc
newg.ancestors = saveAncestors(callergp)
newg.startpc = fn.fn
if _g_.m.curg != nil {
newg.labels = _g_.m.curg.labels
}
if isSystemGoroutine(newg, false) {
atomic.Xadd(&sched.ngsys, +1)
}
// 状态设为runnable
casgstatus(newg, _Gdead, _Grunnable)
if _p_.goidcache == _p_.goidcacheend {
// Sched.goidgen is the last allocated id,
// this batch must be [sched.goidgen+1, sched.goidgen+GoidCacheBatch].
// At startup sched.goidgen=0, so main goroutine receives goid=1.
_p_.goidcache = atomic.Xadd64(&sched.goidgen, _GoidCacheBatch)
_p_.goidcache -= _GoidCacheBatch - 1
_p_.goidcacheend = _p_.goidcache + _GoidCacheBatch
}
// 设置id,
newg.goid = int64(_p_.goidcache)
_p_.goidcache++
if raceenabled {
newg.racectx = racegostart(callerpc)
}
if trace.enabled {
traceGoCreate(newg, newg.startpc)
}
// 加入P的runable数组
runqput(_p_, newg, true)
if atomic.Load(&sched.npidle) != 0 && atomic.Load(&sched.nmspinning) == 0 && mainStarted {
wakep()
}
releasem(_g_.m)
}
从P获取空闲的G,如果没有则生成新的G。把参数复制到G的栈，pc: fn, sp:goexit,把G设为runable,并加入P的runable数组
如何从P获取G: gfget(p)
func gfget(_p_ *p) *g {
retry:
// 如果P的本地空闲链表为空&全局空闲链表不为空
if _p_.gFree.empty() && (!sched.gFree.stack.empty() || !sched.gFree.noStack.empty()) {
lock(&sched.gFree.lock)
// 从全局移一批到本地空闲链表
for _p_.gFree.n < 32 {
// Prefer Gs with stacks.
gp := sched.gFree.stack.pop()
if gp == nil {
gp = sched.gFree.noStack.pop()
if gp == nil {
break
}
}
sched.gFree.n--
_p_.gFree.push(gp)
_p_.gFree.n++
}
unlock(&sched.gFree.lock)
goto retry
}
// 从本地空闲链表pop一个G
gp := _p_.gFree.pop()
if gp == nil {
return nil
}
_p_.gFree.n--
if gp.stack.lo == 0 {
// Stack was deallocated in gfput. Allocate a new one.
systemstack(func() {
gp.stack = stackalloc(_FixedStack)
})
gp.stackguard0 = gp.stack.lo + _StackGuard
} else {
if raceenabled {
racemalloc(unsafe.Pointer(gp.stack.lo), gp.stack.hi-gp.stack.lo)
}
if msanenabled {
msanmalloc(unsafe.Pointer(gp.stack.lo), gp.stack.hi-gp.stack.lo)
}
}
return gp
}
从P的本地空闲链表获取G。如果本地空闲链表为空，则从全局空闲链表移一批到本地。
加入本地runabled列表，runqput(p, newg, true)
func runqput(_p_ *p, gp *g, next bool) {
if randomizeScheduler && next && fastrand()%2 == 0 {
next = false
}
if next {
retryNext:
// 把g设为runnext, 之前的runnext加入runable队列
oldnext := _p_.runnext
if !_p_.runnext.cas(oldnext, guintptr(unsafe.Pointer(gp))) {
goto retryNext
}
if oldnext == 0 {
return
}
// Kick the old runnext out to the regular run queue.
gp = oldnext.ptr()
}
retry:
h := atomic.LoadAcq(&_p_.runqhead) // load-acquire, synchronize with consumers
t := _p_.runqtail
// 本地runable队列未满，加入本地
if t-h < uint32(len(_p_.runq)) {
_p_.runq[t%uint32(len(_p_.runq))].set(gp)
atomic.StoreRel(&_p_.runqtail, t+1) // store-release, makes the item available for consumption
return
}
// 本地runable队列已满，移一半到全局runable队列
if runqputslow(_p_, gp, h, t) {
return
}
// 移了一部分到全局，所以本地队列未满，再次尝试加入本地队列
goto retry
}
把本地runable队列一半到全局 runqputslow
func runqputslow(_p_ *p, gp *g, h, t uint32) bool {
var batch [len(_p_.runq)/2 + 1]*g
// First, grab a batch from local queue.
n := t - h
n = n / 2
// 如果head - tail 不等于 1/2，说明出问题了
if n != uint32(len(_p_.runq)/2) {
throw("runqputslow: queue is not full")
}
// 从head开始遍历 1/2 加入batch中
for i := uint32(0); i < n; i++ {
batch[i] = _p_.runq[(h+i)%uint32(len(_p_.runq))].ptr()
}
// 并没有从队列中删除，只是修改了head
if !atomic.CasRel(&_p_.runqhead, h, h+n) { // cas-release, commits consume
return false
}
batch[n] = gp
if randomizeScheduler {
// 重新排序
for i := uint32(1); i <= n; i++ {
j := fastrandn(i + 1)
batch[i], batch[j] = batch[j], batch[i]
}
}
// Link the goroutines.
for i := uint32(0); i < n; i++ {
// 各个G连在一起
batch[i].schedlink.set(batch[i+1])
}
var q gQueue
q.head.set(batch[0])
q.tail.set(batch[n])
// 全局队列可能其他P也会操作，所以加锁
lock(&sched.lock)
// 加入全局runable队列
globrunqputbatch(&q, int32(n+1))
unlock(&sched.lock)
return true
}
runable入队列:
- 替换runnext
- 本地队列未满:
- 加入本地队列
- 本地队列已满：
- 本地队列移一半到全局队列
- 再次尝试加入本地队列
G小结
graph TB;
A["go func() {} 我们写的goroutine"] --> B
B["newproc 获取func和参数"] -- "切换到g0,使用g0栈空间" --> C
C[newproc1] --> D[gfget 从当前P获取空闲G]
D --> E{"P空&全局不空"}
E -- Y --> F[全局移32个到P本地]
E -- N --> G
F --> G[本地取出空闲G,初始化栈空间]
G --> H{"获取空闲G成功"}
H -- N --> I[创建G,初始化栈空间, 加入全局G数组]
H -- Y --> J
I --> J["参数复制到栈,清除堆,pc:func,sp:goexit1"]
J --> K["状态设为runable,设置goid"]
K -- runqput 加入当前P的runable队列 --> L[用g替换runnext]
L --> M{ 本地runable队列满 }
M -- Y --> N[本地runbale队列移一半到全局]
M -- N --> O[加入本地runable队列]
N --> M
O --> P{有空闲P&没有自旋的M }
P -- Y --> Q["wakep()"]
G.png
现在有了G,那存放G的P又是怎么来的呢？procresize
在启动的时候有一个环节是schedinit
func schedinit() {
...
// 修改P的个数
if procresize(procs) != nil {
throw("unknown runnable goroutine during bootstrap")
}
...
}
func procresize(nprocs int32) *p {
old := gomaxprocs
// P个数必须大于0
if old < 0 || nprocs <= 0 {
throw("procresize: invalid arg")
}
if trace.enabled {
traceGomaxprocs(nprocs)
}
// update statistics
now := nanotime()
if sched.procresizetime != 0 {
sched.totaltime += int64(old) * (now - sched.procresizetime)
}
sched.procresizetime = now
// 截断或扩容allp, allp全局存了所有的P
if nprocs > int32(len(allp)) {
lock(&allpLock)
if nprocs <= int32(cap(allp)) {
allp = allp[:nprocs]
} else {
nallp := make([]*p, nprocs)
copy(nallp, allp[:cap(allp)])
allp = nallp
}
unlock(&allpLock)
}
// 创建新增的P
for i := old; i < nprocs; i++ {
pp := allp[i]
if pp == nil {
pp = new(p)
}
// 初始化，状态设为Pgcstop
pp.init(i)
// 存入allp
atomicstorep(unsafe.Pointer(&allp[i]), unsafe.Pointer(pp))
}
_g_ := getg()
// 如果当前的P.id < nprocs,说明还在allp里面，继续运行
if _g_.m.p != 0 && _g_.m.p.ptr().id < nprocs {
_g_.m.p.ptr().status = _Prunning
_g_.m.p.ptr().mcache.prepareForSweep()
} else {
if _g_.m.p != 0 {
if trace.enabled {
traceGoSched()
traceProcStop(_g_.m.p.ptr())
}
_g_.m.p.ptr().m = 0
}
//将当前P和M取消关联
_g_.m.p = 0
_g_.m.mcache = nil
p := allp[0]
p.m = 0
p.status = _Pidle
// 再将当前M和allp的第一个P关联,并设置为Prunning
acquirep(p)
if trace.enabled {
traceGoStart()
}
}
// 释放多余的P
for i := nprocs; i < old; i++ {
p := allp[i]
// 这里的释放工作不是直接删除回收p,而是主要把p中的可运行和空闲的G移到全局去
p.destroy()
}
// 如果长度不相等，裁剪allp
if int32(len(allp)) != nprocs {
lock(&allpLock)
allp = allp[:nprocs]
unlock(&allpLock)
}
var runnablePs *p
for i := nprocs - 1; i >= 0; i-- {
p := allp[i]
// 当前P继续
if _g_.m.p.ptr() == p {
continue
}
// 将P设为Pidle,也就是不和M关联
p.status = _Pidle
// 如果P没有可运行的G, 那么将P加入全局空闲P链表
if runqempty(p) {
pidleput(p)
} else {
// 重新设置M
p.m.set(mget())
// 将allp的各个P连起来
p.link.set(runnablePs)
runnablePs = p
}
}
stealOrder.reset(uint32(nprocs))
var int32p *int32 = &gomaxprocs
// 更新gomaxprocs, P个数
atomic.Store((*uint32)(unsafe.Pointer(int32p)), uint32(nprocs))
return runnablePs
}
func (pp *p) destroy() {
// 将p的runable移到全局去
for pp.runqhead != pp.runqtail {
// Pop from tail of local queue
pp.runqtail--
gp := pp.runq[pp.runqtail%uint32(len(pp.runq))].ptr()
// Push onto head of global queue
globrunqputhead(gp)
}
// runnext移到全局
if pp.runnext != 0 {
globrunqputhead(pp.runnext.ptr())
pp.runnext = 0
}
...
// If there's a background worker, make it runnable and put
// it on the global queue so it can clean itself up.
if gp := pp.gcBgMarkWorker.ptr(); gp != nil {
casgstatus(gp, _Gwaiting, _Grunnable)
if trace.enabled {
traceGoUnpark(gp, 0)
}
globrunqput(gp)
// This assignment doesn't race because the
// world is stopped.
pp.gcBgMarkWorker.set(nil)
}
...
// 释放p的内存
freemcache(pp.mcache)
pp.mcache = nil
// P的gFree链表移到全局
gfpurge(pp)
...
pp.status = _Pdead
}
P小结
通过启动时候的schedinit调用procresize生成对应个数的P。因为可以通过runtime.GOMAXPROCS来动态修改P的个数，所以在procresize中会对P数组进行调整，或新增P或减少P。被减少的P会将自身的runable、runnext、gfee移到全局去。
- 如果当前P不在多余的P中，则状态为running
- 如果当前P在多余的P中，则将当前M和P解绑，再将M和P数组的第一P绑定，并设为running
- 除了当前P外；所有P都设为idle，如果P中没有runnable,则将P加入全局空闲P,否则获取全局空闲M和P绑定。
M从何而来？
在创建G的时候会根据情况是否创新M
func newproc1(fn *funcval, argp unsafe.Pointer, narg int32, callergp *g, callerpc uintptr) {
...
// 全局存在空闲P且没有自旋的M
if atomic.Load(&sched.npidle) != 0 && atomic.Load(&sched.nmspinning) == 0 && mainStarted {
wakep()
}
...
}
func wakep() {
// be conservative about spinning threads
if !atomic.Cas(&sched.nmspinning, 0, 1) {
return
}
startm(nil, true)
}
func startm(_p_ *p, spinning bool) {
lock(&sched.lock)
if _p_ == nil {
_p_ = pidleget()
// M与P绑定，没有了P 当然也就不需要生成M了
if _p_ == nil {
unlock(&sched.lock)
if spinning {
if int32(atomic.Xadd(&sched.nmspinning, -1)) < 0 {
throw("startm: negative nmspinning")
}
}
return
}
}
// 全局获取空闲M
mp := mget()
unlock(&sched.lock)
if mp == nil {
var fn func()
if spinning {
// The caller incremented nmspinning, so set m.spinning in the new M.
fn = mspinning
}
// 没有空闲M 就重新生成一个
newm(fn, _p_)
return
}
...
// The caller incremented nmspinning, so set m.spinning in the new M.
mp.spinning = spinning
mp.nextp.set(_p_) // nextP指向P
notewakeup(&mp.park) // 唤醒M
}
创建G时，如果全局存在空闲P且没有自旋的M,则获取M和空闲P绑定。
- 如果全局存在空闲M：获取空闲M，绑定P,唤醒M
- 如果全局不存在空闲M： 新生成M
func newm(fn func(), _p_ *p) {
mp := allocm(_p_, fn) // 新生成M, 并创建M的g0
mp.nextp.set(_p_) // nextp指向p
mp.sigmask = initSigmask
...
newm1(mp)
}
func newm1(mp *m) {
...
execLock.rlock() // Prevent process clone.
newosproc(mp)
execLock.runlock()
}
func newosproc(mp *m) {
...
var oset sigset
sigprocmask(_SIG_SETMASK, &sigset_all, &oset)
// 创建线程和M绑定，最终会调用mstart
err = pthread_create(&attr, funcPC(mstart_stub), unsafe.Pointer(mp))
sigprocmask(_SIG_SETMASK, &oset, nil)
...
}
创建M和P绑定，创建线程和M绑定，最终将调用mstart
func mstart() {
...
mstart1()
...
}
func mstart1() {
...
schedule()
...
}
func schedule() {
...
top:
pp := _g_.m.p.ptr()
pp.preempt = false
...
// 下面就是在各个地方获取runable的G
if gp == nil && gcBlackenEnabled != 0 {
gp = gcController.findRunnableGCWorker(_g_.m.p.ptr())
tryWakeP = tryWakeP || gp != nil
}
if gp == nil {
// 以一定频率从全局获取runable G。 平衡本地和全局
if _g_.m.p.ptr().schedtick%61 == 0 && sched.runqsize > 0 {
lock(&sched.lock)
gp = globrunqget(_g_.m.p.ptr(), 1)
unlock(&sched.lock)
}
}
if gp == nil {
// 从本地获取runable G
gp, inheritTime = runqget(_g_.m.p.ptr())
}
if gp == nil {
// 阻塞直到获取runable。从本地、全局、网络、其他P中获取。
// 如果没有可取的runable,则M进入休眠,
gp, inheritTime = findrunnable()
}
if _g_.m.spinning {
resetspinning()
}
...
execute(gp, inheritTime)
}
func execute(gp *g, inheritTime bool) {
_g_ := getg()
// g和m相互绑定，并设为running
_g_.m.curg = gp
gp.m = _g_.m
casgstatus(gp, _Grunnable, _Grunning)
...
// 运作g,gp.sched的pc是我们写的func, sp是goexit1。
// 执行完func,RET的时候弹出的是goexit1。
gogo(&gp.sched)
}
func goexit1() {
...
// goexit1 调用的是goexit0
mcall(goexit0)
}
func goexit0(gp *g) {
_g_ := getg()
// g执行完了状态改为dead
casgstatus(gp, _Grunning, _Gdead)
...
// 清除g的各种信息
gp.m = nil
locked := gp.lockedm != 0
gp.lockedm = 0
_g_.m.lockedg = 0
gp.preemptStop = false
gp.paniconfault = false
gp._defer = nil // should be true already but just in case.
gp._panic = nil // non-nil for Goexit during panic. points at stack-allocated data.
gp.writebuf = nil
gp.waitreason = 0
gp.param = nil
gp.labels = nil
gp.timer = nil
...
// g 和 m相互解绑
dropg()
...
// 将g放入本地空闲链表。 如果本地空闲个数大于64个，则移一半到全局去
gfput(_g_.m.p.ptr(), gp)
...
// 这里又开始调用上面的schedule了。所以M是在不断获取G，执行G
schedule()
}
M小结
graph TB;
A["有空闲P&没有自旋的M"] --> B["wakep()"]
B -- startm --> B2[全局获取空闲P]
B2--> C[全局获取空闲M]
C --> D{获取成功}
D -- Y --> E[M和P绑定]
D -- N --> F["创建M"]
F --> E
E --> G[唤醒M]
G --> H[mstart / mstart1]
H --> I[schedule]
I --> J[在P本地或全局获取runbale G]
J --> K{获取成功}
K -- Y --> L
K -- N --> M["反复在本地、全局、网络、其他P中获取runable G"]
M --> N{获取成功}
N -- Y --> L["G和M相互绑定，G设为running"]
L --> O["汇编执行G的pc:func。执行完RET弹出sp:goexit1"]
O --> P[goexit1 / goexit0]
P --> Q["将G状态设为dead,清除G的各种信息，G和M相互解绑"]
Q --> R["G放入本地空闲链表。如果本地空闲个数大于64个，则移一半到全局去"]
R --> I
N -- N --> S["MP解绑，P加入全局空闲P,M加入全局空闲M"]
S --> T[M进入睡眠]
M.png
就这么完了？ 不，还有一个独立的M
在启动阶段创建了一个M执行sysmon
func main() {
...
if GOARCH != "wasm" {
systemstack(func() {
newm(sysmon, nil)
})
}
...
}
func sysmon() {
lock(&sched.lock)
sched.nmsys++
checkdead()
unlock(&sched.lock)
lasttrace := int64(0)
idle := 0 // how many cycles in succession we had not wokeup somebody
delay := uint32(0) // 间隔时间 20us ~ 10ms
for {
if idle == 0 { // start with 20us sleep...
delay = 20
} else if idle > 50 { // start doubling the sleep after 1ms...
delay *= 2
}
if delay > 10*1000 { // up to 10ms
delay = 10 * 1000
}
usleep(delay)
now := nanotime()
next, _ := timeSleepUntil() //所有P中timer最先到时的时间
// 正在STW或者所有P都处于空闲时，sysmon休眠一会
if debug.schedtrace <= 0 && (sched.gcwaiting != 0 || atomic.Load(&sched.npidle) == uint32(gomaxprocs)) {
lock(&sched.lock)
if atomic.Load(&sched.gcwaiting) != 0 || atomic.Load(&sched.npidle) == uint32(gomaxprocs) {
if next > now {
atomic.Store(&sched.sysmonwait, 1)
...
// sysmon休眠一会
notetsleep(&sched.sysmonnote, sleep)
...
atomic.Store(&sched.sysmonwait, 0)
noteclear(&sched.sysmonnote)
}
idle = 0
delay = 20
}
unlock(&sched.lock)
}
// trigger libc interceptors if needed
if *cgo_yield != nil {
asmcgocall(*cgo_yield, nil)
}
// poll network if not polled for more than 10ms
lastpoll := int64(atomic.Load64(&sched.lastpoll))
// 如果超过10ms
if netpollinited() && lastpoll != 0 && lastpoll+10*1000*1000 < now {
atomic.Cas64(&sched.lastpoll, uint64(lastpoll), uint64(now))
list := netpoll(0) // 获取网络事件的Gs
if !list.empty() {
incidlelocked(-1)
// 将这些G设为runable,加入全局runable
// 如果存在空闲P,则调用startm运行他们
injectglist(&list)
incidlelocked(1)
}
}
// 有timer到时了，启动M去执行
if next < now {
startm(nil, false)
}
// retake P's blocked in syscalls
// and preempt long running G's
if retake(now) != 0 {
idle = 0
} else {
idle++
}
// 2分钟或GC标记堆达到一定大小，触发垃圾回收
if t := (gcTrigger{kind: gcTriggerTime, now: now}); t.test() && atomic.Load(&forcegc.idle) != 0 {
lock(&forcegc.lock)
forcegc.idle = 0
var list gList
list.push(forcegc.g)
injectglist(&list)
unlock(&forcegc.lock)
}
if debug.schedtrace > 0 && lasttrace+int64(debug.schedtrace)*1000000 <= now {
lasttrace = now
schedtrace(debug.scheddetail > 0)
}
}
}
func retake(now int64) uint32 {
n := 0
lock(&allpLock)
for i := 0; i < len(allp); i++ {
_p_ := allp[i]
if _p_ == nil {
continue
}
pd := &_p_.sysmontick
s := _p_.status
sysretake := false
if s == _Prunning || s == _Psyscall {
// G运行时间超过10ms,进行抢占。
// 其实这只针对于running,因为syscall的P没有M
t := int64(_p_.schedtick)
if int64(pd.schedtick) != t {
pd.schedtick = uint32(t)
pd.schedwhen = now
} else if pd.schedwhen+forcePreemptNS <= now {
preemptone(_p_)
sysretake = true
}
}
if s == _Psyscall {
...
if runqempty(_p_) && atomic.Load(&sched.nmspinning)+atomic.Load(&sched.npidle) > 0 && pd.syscallwhen+10*1000*1000 > now {
continue
}
...
incidlelocked(-1)
if atomic.Cas(&_p_.status, s, _Pidle) {
...
n++
_p_.syscalltick++
// 将P交由其他M
handoffp(_p_)
}
incidlelocked(1)
lock(&allpLock)
}
}
unlock(&allpLock)
return uint32(n)
}
sysmon小结
graph TB;
A["启动阶段创建独立M"] -- sysmon --> B{"STW || 所有P空闲"}
B --Y--> C[睡眠一会儿]
B --N--> D{超过10ms}
C --> D
D --Y-->E["从netpoll获取G,加入全局runable。如有空闲P，startM运行他们"]
D --N--> F["如有timer到时，startM运行"]
E --> F
F --retake--> G[循环所有P]
G -.running:超过10ms.-> H[preempton抢占]
H -.syscall:有work或小于10ms .-> I[handoffp将P交由其他M]
I -.-> G
I --> J["2分钟或GC标记堆达到一定大小，触发垃圾回收"]
J -- "20us~10ms" --> B
sysmon.png
