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