golang源码学习之select

ihornet · · 650 次点击 · · 开始浏览

这是一个创建于的文章，其中的信息可能已经有所发展或是发生改变。

先上结论吧

select 是针对chan类型的，所以case 只有default和chan（读/写）两种
遍历case的时候顺序不确定，但chan的优先级比default高。当有default和可执行的chan时，总是执行chan
当没有default,且无可执行的chan时，阻塞
select{}, 阻塞

开始看源码吧

scase

// case 的几种类型
const (
    caseNil = iota
    caseRecv
    caseSend
    caseDefault
)

type scase struct {
    c           *hchan         // chan
    elem        unsafe.Pointer // data element 数据元素
    kind        uint16  // 对应const的那几种类型
    pc          uintptr // race pc (for race detector / msan)
    releasetime int64
}

入口

func reflect_rselect(cases []runtimeSelect) (int, bool) {
    // 没有case 时直接阻塞， 所以我们在demo时 main(){ .... select{} }
    if len(cases) == 0 {
        block()
    }
    sel := make([]scase, len(cases))
    // 为什么是 2 倍呢？ 后面的pollorder和lockorder会用到
    order := make([]uint16, 2*len(cases))
    for i := range cases {
        rc := &cases[i]
        switch rc.dir {
        case selectDefault:
            sel[i] = scase{kind: caseDefault}
        case selectSend:
            sel[i] = scase{kind: caseSend, c: rc.ch, elem: rc.val}
        case selectRecv:
            sel[i] = scase{kind: caseRecv, c: rc.ch, elem: rc.val}
        }
        if raceenabled || msanenabled {
            selectsetpc(&sel[i])
        }
    }

    return selectgo(&sel[0], &order[0], len(cases))
}

这里主要是初始化case数组。重点在selectgo中

func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool) {
    if debugSelect {
        print("select: cas0=", cas0, "\n")
    }

    // 指向case数组首地址
    cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0))
       // order1 长度是 cas1 的两倍
    order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0))

       // slice里面有两个冒号什么意思呢？ a[x:y:z] 切片长度: y-x 切片容量:z-x
    scases := cas1[:ncases:ncases]
    // 轮询顺序
    pollorder := order1[:ncases:ncases]
    // chan 加锁顺序
    lockorder := order1[ncases:][:ncases:ncases] //赋值完之后，其实只用了order1的 2*ncases长度，pollorder占了前面ncases， lockorder占了后面ncases

    // Replace send/receive cases involving nil channels with
    // caseNil so logic below can assume non-nil channel.
    for i := range scases {
        cas := &scases[i]
        if cas.c == nil && cas.kind != caseDefault {
            *cas = scase{}
        }
    }

    var t0 int64
    if blockprofilerate > 0 {
        t0 = cputicks()
        for i := 0; i < ncases; i++ {
            scases[i].releasetime = -1
        }
    }

    // 洗牌 打乱pollorder顺序
    for i := 1; i < ncases; i++ {
        j := fastrandn(uint32(i + 1))
        pollorder[i] = pollorder[j]
        pollorder[j] = uint16(i)
    }

    // sort the cases by Hchan address to get the locking order.
    // simple heap sort, to guarantee n log n time and constant stack footprint.
    // 下面一堆for循环，根据 hchan的地址排序， 生成lockorder加锁顺序
    for i := 0; i < ncases; i++ {
        j := i
        // Start with the pollorder to permute cases on the same channel.
        c := scases[pollorder[i]].c
        // 这里的sortkey() 其实只是返回内存地址
        for j > 0 && scases[lockorder[(j-1)/2]].c.sortkey() < c.sortkey() {
            k := (j - 1) / 2
            lockorder[j] = lockorder[k]
            j = k
        }
        lockorder[j] = pollorder[i]
    }
    for i := ncases - 1; i >= 0; i-- {
        o := lockorder[i]
        c := scases[o].c
        lockorder[i] = lockorder[0]
        j := 0
        for {
            k := j*2 + 1
            if k >= i {
                break
            }
            if k+1 < i && scases[lockorder[k]].c.sortkey() < scases[lockorder[k+1]].c.sortkey() {
                k++
            }
            if c.sortkey() < scases[lockorder[k]].c.sortkey() {
                lockorder[j] = lockorder[k]
                j = k
                continue
            }
            break
        }
        lockorder[j] = o
    }

    if debugSelect {
        for i := 0; i+1 < ncases; i++ {
            if scases[lockorder[i]].c.sortkey() > scases[lockorder[i+1]].c.sortkey() {
                print("i=", i, " x=", lockorder[i], " y=", lockorder[i+1], "\n")
                throw("select: broken sort")
            }
        }
    }

    // lock all the channels involved in the select
    sellock(scases, lockorder)

    var (
        gp     *g
        sg     *sudog
        c      *hchan
        k      *scase
        sglist *sudog
        sgnext *sudog
        qp     unsafe.Pointer
        nextp  **sudog
    )

loop:
    // pass 1 - look for something already waiting
    // CASE 1: case中有可执行的chan， 或者存在default case
    var dfli int
    var dfl *scase
    var casi int
    var cas *scase
    var recvOK bool
    //开始遍历case数组了
    for i := 0; i < ncases; i++ {
        casi = int(pollorder[i])
        cas = &scases[casi]
        c = cas.c

        switch cas.kind {
        // chan 为空 下一轮循环
        case caseNil:
            continue

        case caseRecv: // 接收chan

            sg = c.sendq.dequeue()
            //当chan的send队列存在 G 时
            if sg != nil {
                goto recv
            }
            // 当chan的缓存队列存在元素时
            if c.qcount > 0 {
                goto bufrecv
            }

            // 当chan关闭时
            if c.closed != 0 {
                goto rclose
            }

        case caseSend: // 发送队列
            if raceenabled {
                racereadpc(c.raceaddr(), cas.pc, chansendpc)
            }

            // chan关闭时
            if c.closed != 0 {
                goto sclose
            }

            sg = c.recvq.dequeue()
            // chan的接收队列存在 G 时
            if sg != nil {
                goto send
            }

            // chan的缓存队列的元素少于缓存容量时
            if c.qcount < c.dataqsiz {
                goto bufsend
            }

        case caseDefault: // case default, 你看 default的情况并没有结束循环，说明 chan的优先级比default高
            dfli = casi
            dfl = cas
        }
    }

    if dfl != nil {
        selunlock(scases, lockorder)
        casi = dfli
        cas = dfl
        goto retc
    }

    // pass 2 - enqueue on all chans
    // CASE 2: 将当前的 G 加入的 chan 的等待队列中
    gp = getg()
    if gp.waiting != nil {
        throw("gp.waiting != nil")
    }
    nextp = &gp.waiting
    for _, casei := range lockorder {
        casi = int(casei)
        cas = &scases[casi]
        if cas.kind == caseNil {
            continue
        }
        c = cas.c
        sg := acquireSudog()
        sg.g = gp
        sg.isSelect = true
        // No stack splits between assigning elem and enqueuing
        // sg on gp.waiting where copystack can find it.
        sg.elem = cas.elem
        sg.releasetime = 0
        if t0 != 0 {
            sg.releasetime = -1
        }
        sg.c = c
        // Construct waiting list in lock order.
        *nextp = sg
        nextp = &sg.waitlink

        switch cas.kind {
        case caseRecv:
            // 加入等待接收队列
            // 不断的循环是不是会导致队列边长呢？ 其实不是的， 因为在CASE 1 的时候有做出栈操作
            c.recvq.enqueue(sg)

        case caseSend:
            // 加入等待发送队列
            c.sendq.enqueue(sg)
        }
    }

    // wait for someone to wake us up
    gp.param = nil
    // 当前 G 进入休眠
    gopark(selparkcommit, nil, waitReasonSelect, traceEvGoBlockSelect, 1)

    sellock(scases, lockorder)

    gp.selectDone = 0
    sg = (*sudog)(gp.param)
    gp.param = nil

    // pass 3 - dequeue from unsuccessful chans
    // otherwise they stack up on quiet channels
    // record the successful case, if any.
    // We singly-linked up the SudoGs in lock order.

    // CASE 3: 被唤醒， 这种情况是不存在default的时候
    casi = -1
    cas = nil
    sglist = gp.waiting
    // Clear all elem before unlinking from gp.waiting.
    for sg1 := gp.waiting; sg1 != nil; sg1 = sg1.waitlink {
        sg1.isSelect = false
        sg1.elem = nil
        sg1.c = nil
    }
    gp.waiting = nil

    for _, casei := range lockorder {
        k = &scases[casei]
        if k.kind == caseNil {
            continue
        }
        if sglist.releasetime > 0 {
            k.releasetime = sglist.releasetime
        }

        // 这段代码一直没想明白，直到我回想到chan的send()方法时，才有些明白了。
        // sg = (*sudog)(gp.param), gp.param其实就是sudog，也就是加入等待队列时的sudog。
        // 当被唤醒时，唤醒的是gp.param，所以遍历等待队列 判断sudog相等就可以确定是哪个case了
        if sg == sglist {
            // sg has already been dequeued by the G that woke us up.
            casi = int(casei)
            cas = k
        } else {
            c = k.c
            if k.kind == caseSend {
                c.sendq.dequeueSudoG(sglist)
            } else {
                c.recvq.dequeueSudoG(sglist)
            }
        }
        sgnext = sglist.waitlink
        sglist.waitlink = nil
        releaseSudog(sglist)
        sglist = sgnext
    }

    // 没找到对应的case, 重新进入loop
    if cas == nil {
        goto loop
    }

    c = cas.c

    if debugSelect {
        print("wait-return: cas0=", cas0, " c=", c, " cas=", cas, " kind=", cas.kind, "\n")
    }

    if cas.kind == caseRecv {
        recvOK = true
    }

    if raceenabled {
        if cas.kind == caseRecv && cas.elem != nil {
            raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc)
        } else if cas.kind == caseSend {
            raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc)
        }
    }
    if msanenabled {
        if cas.kind == caseRecv && cas.elem != nil {
            msanwrite(cas.elem, c.elemtype.size)
        } else if cas.kind == caseSend {
            msanread(cas.elem, c.elemtype.size)
        }
    }

    selunlock(scases, lockorder)
    goto retc

bufrecv:
    // can receive from buffer
    if raceenabled {
        if cas.elem != nil {
            raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc)
        }
        raceacquire(chanbuf(c, c.recvx))
        racerelease(chanbuf(c, c.recvx))
    }
    if msanenabled && cas.elem != nil {
        msanwrite(cas.elem, c.elemtype.size)
    }
    recvOK = true
    qp = chanbuf(c, c.recvx)
    if cas.elem != nil {
        // 将chan缓存中的数据拷贝到 case.elem。  eg: a := <-ch, a就是case.elem
        typedmemmove(c.elemtype, cas.elem, qp)
    }
    typedmemclr(c.elemtype, qp)
    c.recvx++
    if c.recvx == c.dataqsiz {
        c.recvx = 0
    }
    c.qcount--
    selunlock(scases, lockorder)
    goto retc

bufsend:
    // can send to buffer
    if raceenabled {
        raceacquire(chanbuf(c, c.sendx))
        racerelease(chanbuf(c, c.sendx))
        raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc)
    }
    if msanenabled {
        msanread(cas.elem, c.elemtype.size)
    }
    // 将cas.elem拷贝到chan的缓存中。eg: ch <- a, a 就是 cas.elem
    typedmemmove(c.elemtype, chanbuf(c, c.sendx), cas.elem)
    c.sendx++
    if c.sendx == c.dataqsiz {
        c.sendx = 0
    }
    c.qcount++
    selunlock(scases, lockorder)
    goto retc

recv:
    // can receive from sleeping sender (sg)
    recv(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
    if debugSelect {
        print("syncrecv: cas0=", cas0, " c=", c, "\n")
    }
    recvOK = true
    goto retc

rclose:
    // read at end of closed channel
    selunlock(scases, lockorder)
    recvOK = false
    if cas.elem != nil {
        typedmemclr(c.elemtype, cas.elem)
    }
    if raceenabled {
        raceacquire(c.raceaddr())
    }
    goto retc

send:
    // can send to a sleeping receiver (sg)
    if raceenabled {
        raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc)
    }
    if msanenabled {
        msanread(cas.elem, c.elemtype.size)
    }
    send(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
    if debugSelect {
        print("syncsend: cas0=", cas0, " c=", c, "\n")
    }
    goto retc

retc:
    if cas.releasetime > 0 {
        blockevent(cas.releasetime-t0, 1)
    }
    return casi, recvOK

sclose:
    // send on closed channel
    selunlock(scases, lockorder)
    panic(plainError("send on closed channel"))
}

bufrecv、bufsend、recv、rclose、send最终都会跳转到retc。这里涉及到一些channel的知识，有兴趣的可以看我另一篇关于channel的文章 https://www.jianshu.com/p/9dd5e77469da

有疑问加站长微信联系（非本文作者）