Go语言——goroutine并发模型

参考：

Goroutine并发调度模型深度解析&手撸一个协程池

Golang 的 goroutine 是如何实现的？

Golang - 调度剖析【第二部分】

简介

stack

OS线程初始栈为2MB。Go语言中，每个goroutine采用动态扩容方式，初始2KB，按需增长，最大1G。此外GC会收缩栈空间。

BTW，增长扩容都是有代价的，需要copy数据到新的stack，所以初始2KB可能有些性能问题。

管理

用户线程的调度以及生命周期管理都是用户层面，Go语言自己实现的，不借助OS系统调用，减少系统资源消耗。

G-M-P

Go语言采用两级线程模型，即用户线程与内核线程KSE（kernel scheduling entity）是M:N的。最终goroutine还是会交给OS线程执行，但是需要一个中介，提供上下文。这就是G-M-P模型

• G: goroutine, 类似进程控制块，保存栈，状态，id，函数等信息。G只有绑定到P才可以被调度。
• M: machine, OS线程，绑定有效的P之后，进入调度。
• P: 逻辑处理器，保存各种队列G。对于G而言，P就是cpu core。对于M而言，P就是上下文。P的数量由GOMAXPROCS设置，最大256。
• sched: 调度程序，保存GRQ，midle M空闲队列，pidle P空闲队列以及lock等信息

G-M-P模型

队列

Go调度器有两个不同的运行队列：

• GRQ，全局运行队列，尚未分配给P的G
• LRQ，本地运行队列，每个P都有一个LRQ，用于管理分配给P执行的G

状态

go1.10\src\runtime\runtime2.go

• _Gidle: 分配了G，但是没有初始化
• _Grunnable: 在run queue运行队列中，LRQ或者GRQ
• _Grunning: 正在运行指令，有自己的stack。不在runq运行队列中，分配给M和P
• _Gsyscall: 正在执行syscall，而非用户指令，不在runq，分给M，P给找idle的M
• _Gwaiting: block。不在RQ，但是可能会在channel的wait queue等待队列
• _Gdead: unused。在P的gfree list中，不在runq。
• _Gcopystack: stack扩容？

上下文切换

Go调度器根据事件进行上下文切换。

• go关键字，创建goroutine
• gc垃圾回收，gc也是goroutine，所以需要时间片
• system call系统调用，block当前G
• sync同步，block当前G

调度

调度的目的就是防止M堵塞，空闲，系统进程切换。

详见 Golang - 调度剖析【第二部分】

异步调用

Linux可以通过epoll实现网络调用，统称网络轮询器N（Net Poller）。

1. G1在M上运行，P的LRQ有其他3个G，N空闲；
2. G1进行网络IO，因此被移动到N，M继续从LRQ取其他的G执行。比如G2就被上下文切换到M上；
3. G1结束网络请求，收到响应，G1被移回LRQ，等待切换到M执行。

同步调用

文件IO操作

1. G1在M1上运行，P的LRQ有其他3个G；
2. G1进行同步调用，堵塞M；
3. 调度器将M1与P分离，此时M1下只有G1，没有P。
4. 将P与空闲M2绑定，并从LRQ选择G2切换
5. G1结束堵塞操作，移回LRQ。M1空闲备用。

任务窃取

上面都是防止M堵塞，任务窃取是防止M空闲

1. 两个P，P1，P2
2. 如果P1的G都执行完了，LRQ空，P1就开始任务窃取。
3. 第一种情况，P2 LRQ还有G，则P1从P2窃取了LRQ中一半的G
4. 第二种情况，P2也没有LRQ，P1从GRQ窃取。

code

go1.10\src\runtime\proc.go

new

// The minimum size of stack used by Go code
    var _StackMin = 2048

func newproc1(fn *funcval, argp *uint8, narg int32, callerpc uintptr) {
    _g_ := getg()
    _p_ := _g_.m.p.ptr()
    
    newg := gfget(_p_)
    if newg == nil {
        newg = malg(_StackMin)
    }    
    
    newg.startpc = fn.fn
    
    runqput(_p_, newg, true)
    
    if atomic.Load(&sched.npidle) != 0 && atomic.Load(&sched.nmspinning) == 0 && mainStarted {
        wakep()
    }
}

1. 获取当前G
2. 获取当前G的P
3. 从P的gfree中获取G，避免重新创建，有点池化的意思
4. 如果没有可复用的G，就重新创建，参数表示stack大小，起始2KB，支持动态扩容
5. 将G入队，放入P的LRQ中；由于有工作窃取机制，其他P可以从这个P窃取G
6. 如果runq满了（长度256），就放入GRQ中，在sched中
7. 尝试加入额外的P去执行G

start

G没办法自己运行，必须通过M运行

func mstart() {
    mstart1(0)
    
    mexit(osStack)
}

func mstart1(dummy int32) {
    _g_ := getg()

    if _g_ != _g_.m.g0 {
        throw("bad runtime·mstart")
    }    
    
    schedule()    
}

M通过通过调度，执行G

schdule

// One round of scheduler: find a runnable goroutine and execute it.
// Never returns.
func schedule() {
    _g_ := getg()
    
    var gp *g
    
    gp, inheritTime = runqget(_g_.m.p.ptr())
    
    execute(gp, inheritTime)
}

从M挂载P的runq中找到G，执行G