select 负责监听channel

其核心函数selectgo中使用了对地址的堆排序（大根堆），以及洗牌算法，保证case被执行到的概率接近均等。后续补充。

select中case的结构

const debugSelect = false //调试开关// Select case descriptor.
// Known to compiler.
// Changes here must also be made in src/cmd/compile/internal/walk/select.go's scasetype.
type scase struct {c    *hchan         // chan   通道elem unsafe.Pointer // data element   成员
}

sellock 对case依次上锁

func sellock(scases []scase, lockorder []uint16) {var c *hchanfor _, o := range lockorder {c0 := scases[o].cif c0 != c {c = c0lock(&c.lock)}}
}

selunlock 对case依次解锁

func selunlock(scases []scase, lockorder []uint16) {// We must be very careful here to not touch sel after we have unlocked// the last lock, because sel can be freed right after the last unlock.// Consider the following situation.// First M calls runtime·park() in runtime·selectgo() passing the sel.// Once runtime·park() has unlocked the last lock, another M makes// the G that calls select runnable again and schedules it for execution.// When the G runs on another M, it locks all the locks and frees sel.// Now if the first M touches sel, it will access freed memory.for i := len(lockorder) - 1; i >= 0; i-- {c := scases[lockorder[i]].cif i > 0 && c == scases[lockorder[i-1]].c {continue // will unlock it on the next iteration}unlock(&c.lock)}
}

selparkcommit 挂起

func selparkcommit(gp *g, _ unsafe.Pointer) bool {// There are unlocked sudogs that point into gp's stack. Stack// copying must lock the channels of those sudogs.// Set activeStackChans here instead of before we try parking// because we could self-deadlock in stack growth on a// channel lock.gp.activeStackChans = true// Mark that it's safe for stack shrinking to occur now,// because any thread acquiring this G's stack for shrinking// is guaranteed to observe activeStackChans after this store.atomic.Store8(&gp.parkingOnChan, 0)// Make sure we unlock after setting activeStackChans and// unsetting parkingOnChan. The moment we unlock any of the// channel locks we risk gp getting readied by a channel operation// and so gp could continue running before everything before the// unlock is visible (even to gp itself).// This must not access gp's stack (see gopark). In// particular, it must not access the *hselect. That's okay,// because by the time this is called, gp.waiting has all// channels in lock order.var lastc *hchanfor sg := gp.waiting; sg != nil; sg = sg.waitlink {if sg.c != lastc && lastc != nil {// As soon as we unlock the channel, fields in// any sudog with that channel may change,// including c and waitlink. Since multiple// sudogs may have the same channel, we unlock// only after we've passed the last instance// of a channel.unlock(&lastc.lock)}lastc = sg.c}if lastc != nil {unlock(&lastc.lock)}return true
}

block 阻塞 select{}

func block() {gopark(nil, nil, waitReasonSelectNoCases, traceEvGoStop, 1) // forever
}

selectgo 多路select

// selectgo implements the select statement.
//
// cas0 points to an array of type [ncases]scase, and order0 points to
// an array of type [2*ncases]uint16 where ncases must be <= 65536.
// Both reside on the goroutine's stack (regardless of any escaping in
// selectgo).
//
// For race detector builds, pc0 points to an array of type
// [ncases]uintptr (also on the stack); for other builds, it's set to
// nil.
//
// selectgo returns the index of the chosen scase, which matches the
// ordinal position of its respective select{recv,send,default} call.
// Also, if the chosen scase was a receive operation, it reports whether
// a value was received.
func selectgo(cas0 *scase, order0 *uint16, pc0 *uintptr, nsends, nrecvs int, block bool) (int, bool) {  // cas0指向case数组，order0指向2倍cas0长度的数组if debugSelect { //调试模式print("select: cas0=", cas0, "\n")}// NOTE: In order to maintain a lean stack size, the number of scases// is capped at 65536.cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0)) //cas1指向cas0指向的数组，最大支持65536个case，其为了保证“堆”的大小合理order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0)) //order1指向order0指向的数组，保证case执行的随机性，其被分为两部分，pollorder和lockorderncases := nsends + nrecvs // 发送和接收的case总数scases := cas1[:ncases:ncases] // 取出所有casepollorder := order1[:ncases:ncases] // 轮训顺序lockorder := order1[ncases:][:ncases:ncases] // 上锁顺序// NOTE: pollorder/lockorder's underlying array was not zero-initialized by compiler.// Even when raceenabled is true, there might be select// statements in packages compiled without -race (e.g.,// ensureSigM in runtime/signal_unix.go).var pcs []uintptrif raceenabled && pc0 != nil { // 竞态检测相关pc1 := (*[1 << 16]uintptr)(unsafe.Pointer(pc0))pcs = pc1[:ncases:ncases]}casePC := func(casi int) uintptr {if pcs == nil {return 0}return pcs[casi]}var t0 int64if blockprofilerate > 0 {t0 = cputicks()}// The compiler rewrites selects that statically have// only 0 or 1 cases plus default into simpler constructs.// The only way we can end up with such small sel.ncase// values here is for a larger select in which most channels// have been nilled out. The general code handles those// cases correctly, and they are rare enough not to bother// optimizing (and needing to test).// generate permuted ordernorder := 0for i := range scases { //对pollorder的len(scases)长度的数据进行洗牌，使pollorder乱序cas := &scases[i]// Omit cases without channels from the poll and lock orders.if cas.c == nil {cas.elem = nil // allow GCcontinue}j := fastrandn(uint32(norder + 1))pollorder[norder] = pollorder[j]pollorder[j] = uint16(i)norder++}pollorder = pollorder[:norder] //确定case个数lockorder = lockorder[:norder] //同上// 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 i := range lockorder {   //根据地址对lockorder进行堆排序j := i// Start with the pollorder to permute cases on the same channel.c := scases[pollorder[i]].cfor j > 0 && scases[lockorder[(j-1)/2]].c.sortkey() < c.sortkey() {k := (j - 1) / 2lockorder[j] = lockorder[k]j = k}lockorder[j] = pollorder[i]}for i := len(lockorder) - 1; i >= 0; i-- {o := lockorder[i]c := scases[o].clockorder[i] = lockorder[0]j := 0for {k := j*2 + 1if 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 = kcontinue}break}lockorder[j] = o}if debugSelect { //调试模式for i := 0; i+1 < len(lockorder); 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 selectsellock(scases, lockorder) //对scases 根据 lockorder次序上锁var (gp     *gsg     *sudogc      *hchank      *scasesglist *sudogsgnext *sudogqp     unsafe.Pointernextp  **sudog)// pass 1 - look for something already waitingvar casi intvar cas *scasevar caseSuccess boolvar caseReleaseTime int64 = -1var recvOK boolfor _, casei := range pollorder { // 轮询pollorder，走一个分支casi = int(casei)cas = &scases[casi]c = cas.cif casi >= nsends { //执行系列操作sg = c.sendq.dequeue()if sg != nil {goto recv}if c.qcount > 0 {goto bufrecv}if c.closed != 0 {goto rclose}} else {if raceenabled {racereadpc(c.raceaddr(), casePC(casi), chansendpc)}if c.closed != 0 {goto sclose}sg = c.recvq.dequeue()if sg != nil {goto send}if c.qcount < c.dataqsiz {goto bufsend}}}if !block { //非阻塞，解锁selunlock(scases, lockorder)casi = -1goto retc}// pass 2 - enqueue on all chansgp = getg()if gp.waiting != nil {throw("gp.waiting != nil")}nextp = &gp.waitingfor _, casei := range lockorder { //将sg入队casi = int(casei)cas = &scases[casi]c = cas.csg := acquireSudog()sg.g = gpsg.isSelect = true// No stack splits between assigning elem and enqueuing// sg on gp.waiting where copystack can find it.sg.elem = cas.elemsg.releasetime = 0if t0 != 0 {sg.releasetime = -1}sg.c = c// Construct waiting list in lock order.*nextp = sgnextp = &sg.waitlinkif casi < nsends {c.sendq.enqueue(sg)} else {c.recvq.enqueue(sg)}}// wait for someone to wake us upgp.param = nil// Signal to anyone trying to shrink our stack that we're about// to park on a channel. The window between when this G's status// changes and when we set gp.activeStackChans is not safe for// stack shrinking.atomic.Store8(&gp.parkingOnChan, 1)gopark(selparkcommit, nil, waitReasonSelect, traceEvGoBlockSelect, 1)//挂起，等待调度gp.activeStackChans = falsesellock(scases, lockorder) //加锁，根据lockordergp.selectDone = 0sg = (*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.casi = -1cas = nilcaseSuccess = falsesglist = gp.waiting// Clear all elem before unlinking from gp.waiting.for sg1 := gp.waiting; sg1 != nil; sg1 = sg1.waitlink {sg1.isSelect = falsesg1.elem = nilsg1.c = nil}gp.waiting = nilfor _, casei := range lockorder {k = &scases[casei]if sg == sglist {// sg has already been dequeued by the G that woke us up.casi = int(casei)cas = kcaseSuccess = sglist.successif sglist.releasetime > 0 {caseReleaseTime = sglist.releasetime}} else {c = k.cif int(casei) < nsends {c.sendq.dequeueSudoG(sglist)} else {c.recvq.dequeueSudoG(sglist)}}sgnext = sglist.waitlinksglist.waitlink = nilreleaseSudog(sglist)sglist = sgnext}if cas == nil {throw("selectgo: bad wakeup")}c = cas.cif debugSelect {print("wait-return: cas0=", cas0, " c=", c, " cas=", cas, " send=", casi < nsends, "\n")}if casi < nsends {if !caseSuccess {goto sclose}} else {recvOK = caseSuccess}if raceenabled {if casi < nsends {raceReadObjectPC(c.elemtype, cas.elem, casePC(casi), chansendpc)} else if cas.elem != nil {raceWriteObjectPC(c.elemtype, cas.elem, casePC(casi), chanrecvpc)}}if msanenabled {if casi < nsends {msanread(cas.elem, c.elemtype.size)} else if cas.elem != nil {msanwrite(cas.elem, c.elemtype.size)}}if asanenabled {if casi < nsends {asanread(cas.elem, c.elemtype.size)} else if cas.elem != nil {asanwrite(cas.elem, c.elemtype.size)}}selunlock(scases, lockorder)goto retcbufrecv:// can receive from bufferif raceenabled {if cas.elem != nil {raceWriteObjectPC(c.elemtype, cas.elem, casePC(casi), chanrecvpc)}racenotify(c, c.recvx, nil)}if msanenabled && cas.elem != nil {msanwrite(cas.elem, c.elemtype.size)}if asanenabled && cas.elem != nil {asanwrite(cas.elem, c.elemtype.size)}recvOK = trueqp = chanbuf(c, c.recvx)if cas.elem != nil {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 retcbufsend:// can send to bufferif raceenabled {racenotify(c, c.sendx, nil)raceReadObjectPC(c.elemtype, cas.elem, casePC(casi), chansendpc)}if msanenabled {msanread(cas.elem, c.elemtype.size)}if asanenabled {asanread(cas.elem, c.elemtype.size)}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 retcrecv:// 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 = truegoto retcrclose:// read at end of closed channelselunlock(scases, lockorder)recvOK = falseif cas.elem != nil {typedmemclr(c.elemtype, cas.elem)}if raceenabled {raceacquire(c.raceaddr())}goto retcsend:// can send to a sleeping receiver (sg)if raceenabled {raceReadObjectPC(c.elemtype, cas.elem, casePC(casi), chansendpc)}if msanenabled {msanread(cas.elem, c.elemtype.size)}if asanenabled {asanread(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 retcretc:if caseReleaseTime > 0 {blockevent(caseReleaseTime-t0, 1)}return casi, recvOKsclose:// send on closed channelselunlock(scases, lockorder)panic(plainError("send on closed channel"))
}

sortkey 获取chan的地址作为排序的依据

func (c *hchan) sortkey() uintptr {return uintptr(unsafe.Pointer(c))
}

reflect相关

// A runtimeSelect is a single case passed to rselect.
// This must match ../reflect/value.go:/runtimeSelect
type runtimeSelect struct {dir selectDirtyp unsafe.Pointer // channel type (not used here)ch  *hchan         // channelval unsafe.Pointer // ptr to data (SendDir) or ptr to receive buffer (RecvDir)
}// These values must match ../reflect/value.go:/SelectDir.
type selectDir intconst (_             selectDir = iotaselectSend              // case Chan <- SendselectRecv              // case <-Chan:selectDefault           // default
)//go:linkname reflect_rselect reflect.rselect
func reflect_rselect(cases []runtimeSelect) (int, bool) {if len(cases) == 0 {block()}sel := make([]scase, len(cases))orig := make([]int, len(cases))nsends, nrecvs := 0, 0dflt := -1for i, rc := range cases {var j intswitch rc.dir {case selectDefault:dflt = icontinuecase selectSend:j = nsendsnsends++case selectRecv:nrecvs++j = len(cases) - nrecvs}sel[j] = scase{c: rc.ch, elem: rc.val}orig[j] = i}// Only a default case.if nsends+nrecvs == 0 {return dflt, false}// Compact sel and orig if necessary.if nsends+nrecvs < len(cases) {copy(sel[nsends:], sel[len(cases)-nrecvs:])copy(orig[nsends:], orig[len(cases)-nrecvs:])}order := make([]uint16, 2*(nsends+nrecvs))var pc0 *uintptrif raceenabled {pcs := make([]uintptr, nsends+nrecvs)for i := range pcs {selectsetpc(&pcs[i])}pc0 = &pcs[0]}chosen, recvOK := selectgo(&sel[0], &order[0], pc0, nsends, nrecvs, dflt == -1)// Translate chosen back to caller's ordering.if chosen < 0 {chosen = dflt} else {chosen = orig[chosen]}return chosen, recvOK
}

dequeueSudoG 将g出队

func (q *waitq) dequeueSudoG(sgp *sudog) {x := sgp.prevy := sgp.nextif x != nil {if y != nil {// middle of queuex.next = yy.prev = xsgp.next = nilsgp.prev = nilreturn}// end of queuex.next = nilq.last = xsgp.prev = nilreturn}if y != nil {// start of queuey.prev = nilq.first = ysgp.next = nilreturn}// x==y==nil. Either sgp is the only element in the queue,// or it has already been removed. Use q.first to disambiguate.if q.first == sgp {q.first = nilq.last = nil}
}

Go 1.19.3 select原理简析相关推荐

Go 1.19.3 channel原理简析
channel channel和goroutine是Go语言的核心命脉.这篇文章来简单介绍一下Go chan的原理,源码并不好读,应结合gmp调度模型来理解,后续补充吧. 由上图可见,chan的底层结 ...
Go 1.19.3 error原理简析
Go error是一个很痛的话题(真心难用) 标准库 error 的定义 // The error built-in interface type is the conventional interf ...
Android Handler与Looper原理简析
一直感觉自己简直就是一个弱智,最近越来越感觉是这样了,真的希望自己有一天能够认同自己,认同自己．本文转载于:https://juejin.im/post/59083d7fda2f60005d14ef ...
Mysql锁机制及原理简析
Mysql锁机制及原理简析一.前言 1.什么是锁? 锁是计算机协调多个进程或线程并发访问某一资源的机制. 锁保证数据并发访问的一致性.有效性: 锁冲突也是影响数据库并发访问性能的一个重要因素. 锁是 ...
Webpack模块化原理简析
webpack模块化原理简析 1.webpack的核心原理一切皆模块:在webpack中,css,html.js,静态资源文件等都可以视作模块:便于管理,利于重复利用: 按需加载:进行代码分割,实现 ...
grpc通信原理_gRPC原理简析
gRPC原理简析 gRPC是由谷歌提出并开发的RPC协议,gRPC提供了一套机制,使得应用程序之间可以进行通信. 降级开发者的使用门槛,屏蔽网络协议,调用对端的接口就像是调用本地的函数一样.而gRPC ...
Android V1及V2签名原理简析
Android为了保证系统及应用的安全性,在安装APK的时候需要校验包的完整性,同时,对于覆盖安装的场景还要校验新旧是否匹配,这两者都是通过Android签名机制来进行保证的,本文就简单看下Andro ...
CRC原理简析——史上最清新脱俗简单易懂的CRC解析
CRC原理简析 1. CRC校验原理 CRC校验原理根本思想就是先在要发送的帧后面附加一个数(这个就是用来校验的校验码,但要注意,这里的数也是二进制序列的,下同),生成一个新帧发送给接收端.当然,这个 ...
Java的定时器Timer和定时任务TimerTask应用以及原理简析
记录:272 场景:Java JDK自带的定时器Timer和定时任务TimerTask应用以及原理简析.在JDK工具包:java.util中可以找到源码,即java.util.Timer和java.u ...

Go 1.19.3 select原理简析

select 负责监听channel

select中case的结构

sellock 对case依次上锁

selunlock 对case依次解锁

selparkcommit 挂起

block 阻塞 select{}

selectgo 多路select

sortkey 获取chan的地址作为排序的依据

reflect相关

dequeueSudoG 将g出队

Go 1.19.3 select原理简析相关推荐

最新文章

热门文章