深刻理解GO语言之并发机制

时间 2019-12-04

标签深刻理解语言并发机制繁體版

原文原文链接

前言：能够说GO真正吸引到个人就是并发这块了，深刻理解这个机制后让我收益匪浅，接下来就用本身薄弱的认知来谈谈GO的并发机制。编程

一，初始化过程

在这以前，先看下asm_arm64.s中的汇编代码关于启动这块的逻辑bootstrap

CALL    runtime·args(SB)
CALL    runtime·osinit(SB)
CALL    runtime·hashinit(SB)
CALL    runtime·schedinit(SB)

// create a new goroutine to start program
PUSHQ    $runtime·main·f(SB)        // entry
PUSHQ    $0            // arg size
CALL    runtime·newproc(SB)
POPQ    AX
POPQ    AX

// start this M
CALL    runtime·mstart(SB)复制代码

接下来就进入分析环节数组

1，经过osinit函数还获取cpu个数和page的大小，这块挺简单的
2，接下来看看schedinit函数（跟本节相关的重要代码）bash

func schedinit() {
    //获取当前的G
    _g_ := getg()
    if raceenabled {
        _g_.racectx, raceprocctx0 = raceinit()
    }
    //设置M的最大数量
    sched.maxmcount = 10000
    //初始化栈空间
    stackinit()
    //内存空间初始化操做
    mallocinit()
    //初始化当前的M
    mcommoninit(_g_.m)

    //将P的数量调整为CPU数量
    procs := ncpu
    if n, ok := atoi32(gogetenv("GOMAXPROCS")); ok && n > 0 {
        procs = n
    }
    if procs > _MaxGomaxprocs {
        procs = _MaxGomaxprocs
    }
    //初始化P
    if procresize(procs) != nil {
        throw("unknown runnable goroutine during bootstrap")
    }

}复制代码

3，上面咱们能够看到调用了procresize函数来初始化P，那么咱们来看下procresize函数。这块代码过长，分几个部分解析（只贴重要的代码）
(1) 初始化新的P多线程

for i := int32(0); i < nprocs; i++ {
        pp := allp[i]
        if pp == nil {
            //新建一个P对象
            pp = new(p)
            pp.id = i
            pp.status = _Pgcstop
            //保存到allp数组（负责存储P的数组）
            atomicstorep(unsafe.Pointer(&allp[i]), unsafe.Pointer(pp))
        }
        //若是P尚未cache，那么进行分配
        if pp.mcache == nil {
            if old == 0 && i == 0 {
                if getg().m.mcache == nil {
                    throw("missing mcache?")
                }
                pp.mcache = getg().m.mcache // bootstrap
            } else {
                pp.mcache = allocmcache()//分配cache
            }
        }
    }复制代码

(2) 释放没被使用的P并发

for i := nprocs; i < old; i++ {
        p := allp[i]
        // 将本地任务添加到全局队列中
        for p.runqhead != p.runqtail {
            p.runqtail--
            gp := p.runq[p.runqtail%uint32(len(p.runq))].ptr()
            // 插入全局队列的头部
            globrunqputhead(gp)
        }
        //释放P所绑定的cache
        freemcache(p.mcache)
        p.mcache = nil
        //将当前的P的G复用连接到全局
        gfpurge(p)
        p.status = _Pdead
        // can't free P itself because it can be referenced by an M in syscall }复制代码

通过这两个步骤后，那么咱们就建立了一批的P，闲置的P会被放进调度器Sched的空闲链表中函数

二，建立G的过程

从上面的汇编代码能够看出接下来会去调用newproc函数来建立主G，而后用这个主函数去执行runtime.main,而后建立一个线程（这个线程在运行期间专门负责系统监控），接下来就进入GO程序中的main函数去运行了。
先看下newproc代码学习

func newproc(siz int32, fn *funcval) {
    argp := add(unsafe.Pointer(&fn), sys.PtrSize)//获取参数的地址
    pc := getcallerpc(unsafe.Pointer(&siz))//获取调用方的PC支
    systemstack(func() {
        newproc1(fn, (*uint8)(argp), siz, 0, pc)//真正建立G的地方
    })
}复制代码

接下来看下newpro1的主要代码ui

func newproc1(fn *funcval, argp *uint8, narg int32, nret int32, callerpc uintptr) *g {
    //从当前P复用链表来获取G
    _p_ := _g_.m.p.ptr()
    newg := gfget(_p_)
    //若是获取失败，则新建一个
    if newg == nil {
        newg = malg(_StackMin)
        casgstatus(newg, _Gidle, _Gdead)
        allgadd(newg) 
    }
    //将获得的G放入P的运行队列中
    runqput(_p_, newg, true)
    //下面三个条件分别为：是否有空闲的P；M是否处于自旋状态；当前是否建立runteime.main
    if atomic.Load(&sched.npidle) != 0 && atomic.Load(&sched.nmspinning) == 0 && runtimeInitTime != 0 {
        wakep()
    }

}复制代码

这个wakep()函数的代码也是值得一看的，这个思想能够用到平时的代码编程中去this

func wakep() {
    //线程被唤醒后须要绑定一个P，这里使用cas操做，能够避免唤醒过多线程，这里也对应了上面的三个判断条件之一
    if !atomic.Cas(&sched.nmspinning, 0, 1) {
        return
    }
    startm(nil, true)
}复制代码

startm的代码就留给读者本身去看了，否则感受整个博文都是代码，主要的思想是：获取一个空闲的P（若是传入的P为空），而后先尝试获取空闲M(空闲的M被调度器schedt管理，这个结构体也能够去看下)，获取不到再去建立一个M等。

三，Channel

这块就稍微比较简单了，代码也很少，可是看下来收获仍是不少的

1，建立Channel

先看下结构体定义(有删减)

type hchan struct {
    qcount   uint           // 队列中数据个数
    dataqsiz uint           // 缓冲槽大小
    buf      unsafe.Pointer // 指向缓冲槽的指针
    elemsize uint16         // 数据大小
    closed   uint32         // 表示 channel 是否关闭
    elemtype *_type // 数据类型
    sendx    uint   // 发送位置索引
    recvx    uint   // 接收位置索引
    recvq    waitq  // 接收等待列表
    sendq    waitq  // 发送等待列表
    lock mutex      // 锁
}
type sudog struct {
    g          *g
    selectdone *uint32 // CAS to 1 to win select race (may point to stack)
    next       *sudog
    prev       *sudog
    elem       unsafe.Pointer // data element (may point to stack)
    waitlink    *sudog // g.waiting list or semaRoot
    waittail    *sudog // semaRoot
    c           *hchan // channel
}复制代码

上面的recvq实际上是读操做阻塞在channel的G列表，sendq实际上是写操做阻塞在channel的G列表，那么G能够同时阻塞在不一样的channel上，那么如何解决呢？这时候就引入了sudog，它实际上是对G的一个包装，表明在等待队列上的一个G。

接下来看看建立过程

func makechan(t *chantype, size int64) *hchan {
    elem := t.elem

    // 大小不超过64K
    if elem.size >= 1<<16 {
        throw("makechan: invalid channel element type")
    }
    var c *hchan
    // 整个建立过程仍是简单明了的
    if elem.kind&kindNoPointers != 0 || size == 0 {
        //一次性分配内存
        c = (*hchan)(mallocgc(hchanSize+uintptr(size)*elem.size, nil, true))
        if size > 0 && elem.size != 0 {
            c.buf = add(unsafe.Pointer(c), hchanSize)
        } else {
            c.buf = unsafe.Pointer(c)
        }
    } else {
        c = new(hchan)
        c.buf = newarray(elem, int(size))
    }
    //设置数据大小，类型和缓冲槽大小
    c.elemsize = uint16(elem.size)
    c.elemtype = elem
    c.dataqsiz = uint(size)

    return c
}复制代码

2，发送

send函数的代码有点长，接下来就拆分进行说明
(1) 若是recvq有G在阻塞，那么就从该队列取出该G，将数据给该G

if sg := c.recvq.dequeue(); sg != nil {
        // Found a waiting receiver. We pass the value we want to send
        // directly to the receiver, bypassing the channel buffer (if any).
        send(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true
    }复制代码

(2) 若是hchan.buf还有可用的空间，那么就将数据放入

//经过比较qcount和datasiz来判断是否还有可用空间
if c.qcount < c.dataqsiz {
        // 将数据放入buf中
        qp := chanbuf(c, c.sendx)
        if raceenabled {
            raceacquire(qp)
            racerelease(qp)
        }
        typedmemmove(c.elemtype, qp, ep)
        c.sendx++
        if c.sendx == c.dataqsiz {
            c.sendx = 0
        }
        c.qcount++
        unlock(&c.lock)
        return true
    }复制代码

(3) hchan.buf满了，那么就会阻塞住了

// 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
}
//初始化一些参数
mysg.elem = ep         
mysg.waitlink = nil
mysg.g = gp
mysg.selectdone = nil
mysg.c = c
gp.waiting = mysg
gp.param = nil
// 将当前 goroutine加入等待队列
c.sendq.enqueue(mysg)   
goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)复制代码

这里咱们就能够看到了，若是满了，那么sudog就会出现了，经过初始化后表明当前G进入等待队列

3，接收

同理，接收也分为三种状况

(1) 当前有发送goroutine阻塞在channel上,buf满了

if sg := c.sendq.dequeue(); sg != nil {
        recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true, true
    }复制代码

(2) buf中有数据

if c.qcount > 0 {
        // 直接从队列中接收
        qp := chanbuf(c, c.recvx)
        if ep != nil {
            typedmemmove(c.elemtype, ep, qp)
        }
        typedmemclr(c.elemtype, qp)
        c.recvx++
        if c.recvx == c.dataqsiz {
            c.recvx = 0
        }
        c.qcount--
        unlock(&c.lock)
        return true, true
    }复制代码

(3) buf中无数据了，那么则会阻塞住

gp := getg()
    mysg := acquireSudog()
    mysg.releasetime = 0
    if t0 != 0 {
        mysg.releasetime = -1
    }
    // 一样的，由sudog表明G去排队
    mysg.elem = ep
    mysg.waitlink = nil
    gp.waiting = mysg
    mysg.g = gp
    mysg.selectdone = nil
    mysg.c = c
    gp.param = nil
    c.recvq.enqueue(mysg)
    goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)复制代码

总结：虽然这块代码逻辑不复杂，可是设计的东西不少，仍是用了不少时间，如今对M执行G的逻辑是懂了，可是还不清楚细节，后面会继续研究。总的读下来，首先第一是对并发的机制能够说是很了解了，对之后在编写相关代码确定颇有帮助。第二，学习到了一些编程思想，例如cas操做，如何更好的进行封装和抽象等。