Black Lives Matter. Support the Equal Justice Initiative.

Source file src/runtime/select.go

Documentation: runtime

     1  // Copyright 2009 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  package runtime
     6  
     7  // This file contains the implementation of Go select statements.
     8  
     9  import (
    10  	"unsafe"
    11  )
    12  
    13  const debugSelect = false
    14  
    15  // scase.kind values.
    16  // Known to compiler.
    17  // Changes here must also be made in src/cmd/compile/internal/gc/select.go's walkselectcases.
    18  const (
    19  	caseNil = iota
    20  	caseRecv
    21  	caseSend
    22  	caseDefault
    23  )
    24  
    25  // Select case descriptor.
    26  // Known to compiler.
    27  // Changes here must also be made in src/cmd/internal/gc/select.go's scasetype.
    28  type scase struct {
    29  	c           *hchan         // chan
    30  	elem        unsafe.Pointer // data element
    31  	kind        uint16
    32  	pc          uintptr // race pc (for race detector / msan)
    33  	releasetime int64
    34  }
    35  
    36  var (
    37  	chansendpc = funcPC(chansend)
    38  	chanrecvpc = funcPC(chanrecv)
    39  )
    40  
    41  func selectsetpc(cas *scase) {
    42  	cas.pc = getcallerpc()
    43  }
    44  
    45  func sellock(scases []scase, lockorder []uint16) {
    46  	var c *hchan
    47  	for _, o := range lockorder {
    48  		c0 := scases[o].c
    49  		if c0 != nil && c0 != c {
    50  			c = c0
    51  			lock(&c.lock)
    52  		}
    53  	}
    54  }
    55  
    56  func selunlock(scases []scase, lockorder []uint16) {
    57  	// We must be very careful here to not touch sel after we have unlocked
    58  	// the last lock, because sel can be freed right after the last unlock.
    59  	// Consider the following situation.
    60  	// First M calls runtime·park() in runtime·selectgo() passing the sel.
    61  	// Once runtime·park() has unlocked the last lock, another M makes
    62  	// the G that calls select runnable again and schedules it for execution.
    63  	// When the G runs on another M, it locks all the locks and frees sel.
    64  	// Now if the first M touches sel, it will access freed memory.
    65  	for i := len(scases) - 1; i >= 0; i-- {
    66  		c := scases[lockorder[i]].c
    67  		if c == nil {
    68  			break
    69  		}
    70  		if i > 0 && c == scases[lockorder[i-1]].c {
    71  			continue // will unlock it on the next iteration
    72  		}
    73  		unlock(&c.lock)
    74  	}
    75  }
    76  
    77  func selparkcommit(gp *g, _ unsafe.Pointer) bool {
    78  	// There are unlocked sudogs that point into gp's stack. Stack
    79  	// copying must lock the channels of those sudogs.
    80  	gp.activeStackChans = true
    81  	// This must not access gp's stack (see gopark). In
    82  	// particular, it must not access the *hselect. That's okay,
    83  	// because by the time this is called, gp.waiting has all
    84  	// channels in lock order.
    85  	var lastc *hchan
    86  	for sg := gp.waiting; sg != nil; sg = sg.waitlink {
    87  		if sg.c != lastc && lastc != nil {
    88  			// As soon as we unlock the channel, fields in
    89  			// any sudog with that channel may change,
    90  			// including c and waitlink. Since multiple
    91  			// sudogs may have the same channel, we unlock
    92  			// only after we've passed the last instance
    93  			// of a channel.
    94  			unlock(&lastc.lock)
    95  		}
    96  		lastc = sg.c
    97  	}
    98  	if lastc != nil {
    99  		unlock(&lastc.lock)
   100  	}
   101  	return true
   102  }
   103  
   104  func block() {
   105  	gopark(nil, nil, waitReasonSelectNoCases, traceEvGoStop, 1) // forever
   106  }
   107  
   108  // selectgo implements the select statement.
   109  //
   110  // cas0 points to an array of type [ncases]scase, and order0 points to
   111  // an array of type [2*ncases]uint16 where ncases must be <= 65536.
   112  // Both reside on the goroutine's stack (regardless of any escaping in
   113  // selectgo).
   114  //
   115  // selectgo returns the index of the chosen scase, which matches the
   116  // ordinal position of its respective select{recv,send,default} call.
   117  // Also, if the chosen scase was a receive operation, it reports whether
   118  // a value was received.
   119  func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool) {
   120  	if debugSelect {
   121  		print("select: cas0=", cas0, "\n")
   122  	}
   123  
   124  	// NOTE: In order to maintain a lean stack size, the number of scases
   125  	// is capped at 65536.
   126  	cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0))
   127  	order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0))
   128  
   129  	scases := cas1[:ncases:ncases]
   130  	pollorder := order1[:ncases:ncases]
   131  	lockorder := order1[ncases:][:ncases:ncases]
   132  
   133  	// Replace send/receive cases involving nil channels with
   134  	// caseNil so logic below can assume non-nil channel.
   135  	for i := range scases {
   136  		cas := &scases[i]
   137  		if cas.c == nil && cas.kind != caseDefault {
   138  			*cas = scase{}
   139  		}
   140  	}
   141  
   142  	var t0 int64
   143  	if blockprofilerate > 0 {
   144  		t0 = cputicks()
   145  		for i := 0; i < ncases; i++ {
   146  			scases[i].releasetime = -1
   147  		}
   148  	}
   149  
   150  	// The compiler rewrites selects that statically have
   151  	// only 0 or 1 cases plus default into simpler constructs.
   152  	// The only way we can end up with such small sel.ncase
   153  	// values here is for a larger select in which most channels
   154  	// have been nilled out. The general code handles those
   155  	// cases correctly, and they are rare enough not to bother
   156  	// optimizing (and needing to test).
   157  
   158  	// generate permuted order
   159  	for i := 1; i < ncases; i++ {
   160  		j := fastrandn(uint32(i + 1))
   161  		pollorder[i] = pollorder[j]
   162  		pollorder[j] = uint16(i)
   163  	}
   164  
   165  	// sort the cases by Hchan address to get the locking order.
   166  	// simple heap sort, to guarantee n log n time and constant stack footprint.
   167  	for i := 0; i < ncases; i++ {
   168  		j := i
   169  		// Start with the pollorder to permute cases on the same channel.
   170  		c := scases[pollorder[i]].c
   171  		for j > 0 && scases[lockorder[(j-1)/2]].c.sortkey() < c.sortkey() {
   172  			k := (j - 1) / 2
   173  			lockorder[j] = lockorder[k]
   174  			j = k
   175  		}
   176  		lockorder[j] = pollorder[i]
   177  	}
   178  	for i := ncases - 1; i >= 0; i-- {
   179  		o := lockorder[i]
   180  		c := scases[o].c
   181  		lockorder[i] = lockorder[0]
   182  		j := 0
   183  		for {
   184  			k := j*2 + 1
   185  			if k >= i {
   186  				break
   187  			}
   188  			if k+1 < i && scases[lockorder[k]].c.sortkey() < scases[lockorder[k+1]].c.sortkey() {
   189  				k++
   190  			}
   191  			if c.sortkey() < scases[lockorder[k]].c.sortkey() {
   192  				lockorder[j] = lockorder[k]
   193  				j = k
   194  				continue
   195  			}
   196  			break
   197  		}
   198  		lockorder[j] = o
   199  	}
   200  
   201  	if debugSelect {
   202  		for i := 0; i+1 < ncases; i++ {
   203  			if scases[lockorder[i]].c.sortkey() > scases[lockorder[i+1]].c.sortkey() {
   204  				print("i=", i, " x=", lockorder[i], " y=", lockorder[i+1], "\n")
   205  				throw("select: broken sort")
   206  			}
   207  		}
   208  	}
   209  
   210  	// lock all the channels involved in the select
   211  	sellock(scases, lockorder)
   212  
   213  	var (
   214  		gp     *g
   215  		sg     *sudog
   216  		c      *hchan
   217  		k      *scase
   218  		sglist *sudog
   219  		sgnext *sudog
   220  		qp     unsafe.Pointer
   221  		nextp  **sudog
   222  	)
   223  
   224  loop:
   225  	// pass 1 - look for something already waiting
   226  	var dfli int
   227  	var dfl *scase
   228  	var casi int
   229  	var cas *scase
   230  	var recvOK bool
   231  	for i := 0; i < ncases; i++ {
   232  		casi = int(pollorder[i])
   233  		cas = &scases[casi]
   234  		c = cas.c
   235  
   236  		switch cas.kind {
   237  		case caseNil:
   238  			continue
   239  
   240  		case caseRecv:
   241  			sg = c.sendq.dequeue()
   242  			if sg != nil {
   243  				goto recv
   244  			}
   245  			if c.qcount > 0 {
   246  				goto bufrecv
   247  			}
   248  			if c.closed != 0 {
   249  				goto rclose
   250  			}
   251  
   252  		case caseSend:
   253  			if raceenabled {
   254  				racereadpc(c.raceaddr(), cas.pc, chansendpc)
   255  			}
   256  			if c.closed != 0 {
   257  				goto sclose
   258  			}
   259  			sg = c.recvq.dequeue()
   260  			if sg != nil {
   261  				goto send
   262  			}
   263  			if c.qcount < c.dataqsiz {
   264  				goto bufsend
   265  			}
   266  
   267  		case caseDefault:
   268  			dfli = casi
   269  			dfl = cas
   270  		}
   271  	}
   272  
   273  	if dfl != nil {
   274  		selunlock(scases, lockorder)
   275  		casi = dfli
   276  		cas = dfl
   277  		goto retc
   278  	}
   279  
   280  	// pass 2 - enqueue on all chans
   281  	gp = getg()
   282  	if gp.waiting != nil {
   283  		throw("gp.waiting != nil")
   284  	}
   285  	nextp = &gp.waiting
   286  	for _, casei := range lockorder {
   287  		casi = int(casei)
   288  		cas = &scases[casi]
   289  		if cas.kind == caseNil {
   290  			continue
   291  		}
   292  		c = cas.c
   293  		sg := acquireSudog()
   294  		sg.g = gp
   295  		sg.isSelect = true
   296  		// No stack splits between assigning elem and enqueuing
   297  		// sg on gp.waiting where copystack can find it.
   298  		sg.elem = cas.elem
   299  		sg.releasetime = 0
   300  		if t0 != 0 {
   301  			sg.releasetime = -1
   302  		}
   303  		sg.c = c
   304  		// Construct waiting list in lock order.
   305  		*nextp = sg
   306  		nextp = &sg.waitlink
   307  
   308  		switch cas.kind {
   309  		case caseRecv:
   310  			c.recvq.enqueue(sg)
   311  
   312  		case caseSend:
   313  			c.sendq.enqueue(sg)
   314  		}
   315  	}
   316  
   317  	// wait for someone to wake us up
   318  	gp.param = nil
   319  	gopark(selparkcommit, nil, waitReasonSelect, traceEvGoBlockSelect, 1)
   320  	gp.activeStackChans = false
   321  
   322  	sellock(scases, lockorder)
   323  
   324  	gp.selectDone = 0
   325  	sg = (*sudog)(gp.param)
   326  	gp.param = nil
   327  
   328  	// pass 3 - dequeue from unsuccessful chans
   329  	// otherwise they stack up on quiet channels
   330  	// record the successful case, if any.
   331  	// We singly-linked up the SudoGs in lock order.
   332  	casi = -1
   333  	cas = nil
   334  	sglist = gp.waiting
   335  	// Clear all elem before unlinking from gp.waiting.
   336  	for sg1 := gp.waiting; sg1 != nil; sg1 = sg1.waitlink {
   337  		sg1.isSelect = false
   338  		sg1.elem = nil
   339  		sg1.c = nil
   340  	}
   341  	gp.waiting = nil
   342  
   343  	for _, casei := range lockorder {
   344  		k = &scases[casei]
   345  		if k.kind == caseNil {
   346  			continue
   347  		}
   348  		if sglist.releasetime > 0 {
   349  			k.releasetime = sglist.releasetime
   350  		}
   351  		if sg == sglist {
   352  			// sg has already been dequeued by the G that woke us up.
   353  			casi = int(casei)
   354  			cas = k
   355  		} else {
   356  			c = k.c
   357  			if k.kind == caseSend {
   358  				c.sendq.dequeueSudoG(sglist)
   359  			} else {
   360  				c.recvq.dequeueSudoG(sglist)
   361  			}
   362  		}
   363  		sgnext = sglist.waitlink
   364  		sglist.waitlink = nil
   365  		releaseSudog(sglist)
   366  		sglist = sgnext
   367  	}
   368  
   369  	if cas == nil {
   370  		// We can wake up with gp.param == nil (so cas == nil)
   371  		// when a channel involved in the select has been closed.
   372  		// It is easiest to loop and re-run the operation;
   373  		// we'll see that it's now closed.
   374  		// Maybe some day we can signal the close explicitly,
   375  		// but we'd have to distinguish close-on-reader from close-on-writer.
   376  		// It's easiest not to duplicate the code and just recheck above.
   377  		// We know that something closed, and things never un-close,
   378  		// so we won't block again.
   379  		goto loop
   380  	}
   381  
   382  	c = cas.c
   383  
   384  	if debugSelect {
   385  		print("wait-return: cas0=", cas0, " c=", c, " cas=", cas, " kind=", cas.kind, "\n")
   386  	}
   387  
   388  	if cas.kind == caseRecv {
   389  		recvOK = true
   390  	}
   391  
   392  	if raceenabled {
   393  		if cas.kind == caseRecv && cas.elem != nil {
   394  			raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc)
   395  		} else if cas.kind == caseSend {
   396  			raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc)
   397  		}
   398  	}
   399  	if msanenabled {
   400  		if cas.kind == caseRecv && cas.elem != nil {
   401  			msanwrite(cas.elem, c.elemtype.size)
   402  		} else if cas.kind == caseSend {
   403  			msanread(cas.elem, c.elemtype.size)
   404  		}
   405  	}
   406  
   407  	selunlock(scases, lockorder)
   408  	goto retc
   409  
   410  bufrecv:
   411  	// can receive from buffer
   412  	if raceenabled {
   413  		if cas.elem != nil {
   414  			raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc)
   415  		}
   416  		raceacquire(chanbuf(c, c.recvx))
   417  		racerelease(chanbuf(c, c.recvx))
   418  	}
   419  	if msanenabled && cas.elem != nil {
   420  		msanwrite(cas.elem, c.elemtype.size)
   421  	}
   422  	recvOK = true
   423  	qp = chanbuf(c, c.recvx)
   424  	if cas.elem != nil {
   425  		typedmemmove(c.elemtype, cas.elem, qp)
   426  	}
   427  	typedmemclr(c.elemtype, qp)
   428  	c.recvx++
   429  	if c.recvx == c.dataqsiz {
   430  		c.recvx = 0
   431  	}
   432  	c.qcount--
   433  	selunlock(scases, lockorder)
   434  	goto retc
   435  
   436  bufsend:
   437  	// can send to buffer
   438  	if raceenabled {
   439  		raceacquire(chanbuf(c, c.sendx))
   440  		racerelease(chanbuf(c, c.sendx))
   441  		raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc)
   442  	}
   443  	if msanenabled {
   444  		msanread(cas.elem, c.elemtype.size)
   445  	}
   446  	typedmemmove(c.elemtype, chanbuf(c, c.sendx), cas.elem)
   447  	c.sendx++
   448  	if c.sendx == c.dataqsiz {
   449  		c.sendx = 0
   450  	}
   451  	c.qcount++
   452  	selunlock(scases, lockorder)
   453  	goto retc
   454  
   455  recv:
   456  	// can receive from sleeping sender (sg)
   457  	recv(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
   458  	if debugSelect {
   459  		print("syncrecv: cas0=", cas0, " c=", c, "\n")
   460  	}
   461  	recvOK = true
   462  	goto retc
   463  
   464  rclose:
   465  	// read at end of closed channel
   466  	selunlock(scases, lockorder)
   467  	recvOK = false
   468  	if cas.elem != nil {
   469  		typedmemclr(c.elemtype, cas.elem)
   470  	}
   471  	if raceenabled {
   472  		raceacquire(c.raceaddr())
   473  	}
   474  	goto retc
   475  
   476  send:
   477  	// can send to a sleeping receiver (sg)
   478  	if raceenabled {
   479  		raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc)
   480  	}
   481  	if msanenabled {
   482  		msanread(cas.elem, c.elemtype.size)
   483  	}
   484  	send(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
   485  	if debugSelect {
   486  		print("syncsend: cas0=", cas0, " c=", c, "\n")
   487  	}
   488  	goto retc
   489  
   490  retc:
   491  	if cas.releasetime > 0 {
   492  		blockevent(cas.releasetime-t0, 1)
   493  	}
   494  	return casi, recvOK
   495  
   496  sclose:
   497  	// send on closed channel
   498  	selunlock(scases, lockorder)
   499  	panic(plainError("send on closed channel"))
   500  }
   501  
   502  func (c *hchan) sortkey() uintptr {
   503  	return uintptr(unsafe.Pointer(c))
   504  }
   505  
   506  // A runtimeSelect is a single case passed to rselect.
   507  // This must match ../reflect/value.go:/runtimeSelect
   508  type runtimeSelect struct {
   509  	dir selectDir
   510  	typ unsafe.Pointer // channel type (not used here)
   511  	ch  *hchan         // channel
   512  	val unsafe.Pointer // ptr to data (SendDir) or ptr to receive buffer (RecvDir)
   513  }
   514  
   515  // These values must match ../reflect/value.go:/SelectDir.
   516  type selectDir int
   517  
   518  const (
   519  	_             selectDir = iota
   520  	selectSend              // case Chan <- Send
   521  	selectRecv              // case <-Chan:
   522  	selectDefault           // default
   523  )
   524  
   525  //go:linkname reflect_rselect reflect.rselect
   526  func reflect_rselect(cases []runtimeSelect) (int, bool) {
   527  	if len(cases) == 0 {
   528  		block()
   529  	}
   530  	sel := make([]scase, len(cases))
   531  	order := make([]uint16, 2*len(cases))
   532  	for i := range cases {
   533  		rc := &cases[i]
   534  		switch rc.dir {
   535  		case selectDefault:
   536  			sel[i] = scase{kind: caseDefault}
   537  		case selectSend:
   538  			sel[i] = scase{kind: caseSend, c: rc.ch, elem: rc.val}
   539  		case selectRecv:
   540  			sel[i] = scase{kind: caseRecv, c: rc.ch, elem: rc.val}
   541  		}
   542  		if raceenabled || msanenabled {
   543  			selectsetpc(&sel[i])
   544  		}
   545  	}
   546  
   547  	return selectgo(&sel[0], &order[0], len(cases))
   548  }
   549  
   550  func (q *waitq) dequeueSudoG(sgp *sudog) {
   551  	x := sgp.prev
   552  	y := sgp.next
   553  	if x != nil {
   554  		if y != nil {
   555  			// middle of queue
   556  			x.next = y
   557  			y.prev = x
   558  			sgp.next = nil
   559  			sgp.prev = nil
   560  			return
   561  		}
   562  		// end of queue
   563  		x.next = nil
   564  		q.last = x
   565  		sgp.prev = nil
   566  		return
   567  	}
   568  	if y != nil {
   569  		// start of queue
   570  		y.prev = nil
   571  		q.first = y
   572  		sgp.next = nil
   573  		return
   574  	}
   575  
   576  	// x==y==nil. Either sgp is the only element in the queue,
   577  	// or it has already been removed. Use q.first to disambiguate.
   578  	if q.first == sgp {
   579  		q.first = nil
   580  		q.last = nil
   581  	}
   582  }
   583  

View as plain text