select.go

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