Source file src/runtime/select.go

Documentation: runtime

  // Copyright 2009 The Go Authors. All rights reserved.
  // Use of this source code is governed by a BSD-style
  // license that can be found in the LICENSE file.
  
  package runtime
  
  // This file contains the implementation of Go select statements.
  
  import (
  	"unsafe"
  )
  
  const debugSelect = false
  
  // scase.kind values.
  // Known to compiler.
  // Changes here must also be made in src/cmd/compile/internal/gc/select.go's walkselect.
  const (
  	caseNil = iota
  	caseRecv
  	caseSend
  	caseDefault
  )
  
  // Select case descriptor.
  // Known to compiler.
  // Changes here must also be made in src/cmd/internal/gc/select.go's scasetype.
  type scase struct {
  	c           *hchan         // chan
  	elem        unsafe.Pointer // data element
  	kind        uint16
  	pc          uintptr // race pc (for race detector / msan)
  	releasetime int64
  }
  
  var (
  	chansendpc = funcPC(chansend)
  	chanrecvpc = funcPC(chanrecv)
  )
  
  func selectsetpc(cas *scase) {
  	cas.pc = getcallerpc()
  }
  
  func sellock(scases []scase, lockorder []uint16) {
  	var c *hchan
  	for _, o := range lockorder {
  		c0 := scases[o].c
  		if c0 != nil && c0 != c {
  			c = c0
  			lock(&c.lock)
  		}
  	}
  }
  
  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(scases) - 1; i >= 0; i-- {
  		c := scases[lockorder[i]].c
  		if c == nil {
  			break
  		}
  		if i > 0 && c == scases[lockorder[i-1]].c {
  			continue // will unlock it on the next iteration
  		}
  		unlock(&c.lock)
  	}
  }
  
  func selparkcommit(gp *g, _ unsafe.Pointer) bool {
  	// 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 *hchan
  	for 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
  }
  
  func block() {
  	gopark(nil, nil, waitReasonSelectNoCases, traceEvGoStop, 1) // forever
  }
  
  // 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. Both reside on the goroutine's
  // stack (regardless of any escaping in selectgo).
  //
  // 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 returns whether
  // a value was received.
  func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool) {
  	if debugSelect {
  		print("select: cas0=", cas0, "\n")
  	}
  
  	cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0))
  	order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0))
  
  	scases := cas1[:ncases:ncases]
  	pollorder := order1[:ncases:ncases]
  	lockorder := order1[ncases:][:ncases: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
  		}
  	}
  
  	// 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 order
  	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 i := 0; i < ncases; i++ {
  		j := i
  		// Start with the pollorder to permute cases on the same channel.
  		c := scases[pollorder[i]].c
  		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
  	var dfli int
  	var dfl *scase
  	var casi int
  	var cas *scase
  	var recvOK bool
  	for i := 0; i < ncases; i++ {
  		casi = int(pollorder[i])
  		cas = &scases[casi]
  		c = cas.c
  
  		switch cas.kind {
  		case caseNil:
  			continue
  
  		case caseRecv:
  			sg = c.sendq.dequeue()
  			if sg != nil {
  				goto recv
  			}
  			if c.qcount > 0 {
  				goto bufrecv
  			}
  			if c.closed != 0 {
  				goto rclose
  			}
  
  		case caseSend:
  			if raceenabled {
  				racereadpc(unsafe.Pointer(c), cas.pc, chansendpc)
  			}
  			if c.closed != 0 {
  				goto sclose
  			}
  			sg = c.recvq.dequeue()
  			if sg != nil {
  				goto send
  			}
  			if c.qcount < c.dataqsiz {
  				goto bufsend
  			}
  
  		case caseDefault:
  			dfli = casi
  			dfl = cas
  		}
  	}
  
  	if dfl != nil {
  		selunlock(scases, lockorder)
  		casi = dfli
  		cas = dfl
  		goto retc
  	}
  
  	// pass 2 - enqueue on all chans
  	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:
  			c.recvq.enqueue(sg)
  
  		case caseSend:
  			c.sendq.enqueue(sg)
  		}
  	}
  
  	// wait for someone to wake us up
  	gp.param = nil
  	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.
  	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
  		}
  		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
  	}
  
  	if cas == nil {
  		// We can wake up with gp.param == nil (so cas == nil)
  		// when a channel involved in the select has been closed.
  		// It is easiest to loop and re-run the operation;
  		// we'll see that it's now closed.
  		// Maybe some day we can signal the close explicitly,
  		// but we'd have to distinguish close-on-reader from close-on-writer.
  		// It's easiest not to duplicate the code and just recheck above.
  		// We know that something closed, and things never un-close,
  		// so we won't block again.
  		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 {
  		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)
  	}
  	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(unsafe.Pointer(c))
  	}
  	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"))
  }
  
  func (c *hchan) sortkey() uintptr {
  	// TODO(khr): if we have a moving garbage collector, we'll need to
  	// change this function.
  	return uintptr(unsafe.Pointer(c))
  }
  
  // A runtimeSelect is a single case passed to rselect.
  // This must match ../reflect/value.go:/runtimeSelect
  type runtimeSelect struct {
  	dir selectDir
  	typ unsafe.Pointer // channel type (not used here)
  	ch  *hchan         // channel
  	val unsafe.Pointer // ptr to data (SendDir) or ptr to receive buffer (RecvDir)
  }
  
  // These values must match ../reflect/value.go:/SelectDir.
  type selectDir int
  
  const (
  	_             selectDir = iota
  	selectSend              // case Chan <- Send
  	selectRecv              // 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))
  	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))
  }
  
  func (q *waitq) dequeueSudoG(sgp *sudog) {
  	x := sgp.prev
  	y := sgp.next
  	if x != nil {
  		if y != nil {
  			// middle of queue
  			x.next = y
  			y.prev = x
  			sgp.next = nil
  			sgp.prev = nil
  			return
  		}
  		// end of queue
  		x.next = nil
  		q.last = x
  		sgp.prev = nil
  		return
  	}
  	if y != nil {
  		// start of queue
  		y.prev = nil
  		q.first = y
  		sgp.next = nil
  		return
  	}
  
  	// 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 = nil
  		q.last = nil
  	}
  }
  

View as plain text