chan.go

Documentation: runtime

     1  // Copyright 2014 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 channels.
     8  
     9  // Invariants:
    10  //  At least one of c.sendq and c.recvq is empty,
    11  //  except for the case of an unbuffered channel with a single goroutine
    12  //  blocked on it for both sending and receiving using a select statement,
    13  //  in which case the length of c.sendq and c.recvq is limited only by the
    14  //  size of the select statement.
    15  //
    16  // For buffered channels, also:
    17  //  c.qcount > 0 implies that c.recvq is empty.
    18  //  c.qcount < c.dataqsiz implies that c.sendq is empty.
    19  
    20  import (
    21  	"runtime/internal/atomic"
    22  	"runtime/internal/math"
    23  	"unsafe"
    24  )
    25  
    26  const (
    27  	maxAlign  = 8
    28  	hchanSize = unsafe.Sizeof(hchan{}) + uintptr(-int(unsafe.Sizeof(hchan{}))&(maxAlign-1))
    29  	debugChan = false
    30  )
    31  
    32  type hchan struct {
    33  	qcount   uint           // total data in the queue
    34  	dataqsiz uint           // size of the circular queue
    35  	buf      unsafe.Pointer // points to an array of dataqsiz elements
    36  	elemsize uint16
    37  	closed   uint32
    38  	elemtype *_type // element type
    39  	sendx    uint   // send index
    40  	recvx    uint   // receive index
    41  	recvq    waitq  // list of recv waiters
    42  	sendq    waitq  // list of send waiters
    43  
    44  	// lock protects all fields in hchan, as well as several
    45  	// fields in sudogs blocked on this channel.
    46  	//
    47  	// Do not change another G's status while holding this lock
    48  	// (in particular, do not ready a G), as this can deadlock
    49  	// with stack shrinking.
    50  	lock mutex
    51  }
    52  
    53  type waitq struct {
    54  	first *sudog
    55  	last  *sudog
    56  }
    57  
    58  //go:linkname reflect_makechan reflect.makechan
    59  func reflect_makechan(t *chantype, size int) *hchan {
    60  	return makechan(t, size)
    61  }
    62  
    63  func makechan64(t *chantype, size int64) *hchan {
    64  	if int64(int(size)) != size {
    65  		panic(plainError("makechan: size out of range"))
    66  	}
    67  
    68  	return makechan(t, int(size))
    69  }
    70  
    71  func makechan(t *chantype, size int) *hchan {
    72  	elem := t.elem
    73  
    74  	// compiler checks this but be safe.
    75  	if elem.size >= 1<<16 {
    76  		throw("makechan: invalid channel element type")
    77  	}
    78  	if hchanSize%maxAlign != 0 || elem.align > maxAlign {
    79  		throw("makechan: bad alignment")
    80  	}
    81  
    82  	mem, overflow := math.MulUintptr(elem.size, uintptr(size))
    83  	if overflow || mem > maxAlloc-hchanSize || size < 0 {
    84  		panic(plainError("makechan: size out of range"))
    85  	}
    86  
    87  	// Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.
    88  	// buf points into the same allocation, elemtype is persistent.
    89  	// SudoG's are referenced from their owning thread so they can't be collected.
    90  	// TODO(dvyukov,rlh): Rethink when collector can move allocated objects.
    91  	var c *hchan
    92  	switch {
    93  	case mem == 0:
    94  		// Queue or element size is zero.
    95  		c = (*hchan)(mallocgc(hchanSize, nil, true))
    96  		// Race detector uses this location for synchronization.
    97  		c.buf = c.raceaddr()
    98  	case elem.ptrdata == 0:
    99  		// Elements do not contain pointers.
   100  		// Allocate hchan and buf in one call.
   101  		c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
   102  		c.buf = add(unsafe.Pointer(c), hchanSize)
   103  	default:
   104  		// Elements contain pointers.
   105  		c = new(hchan)
   106  		c.buf = mallocgc(mem, elem, true)
   107  	}
   108  
   109  	c.elemsize = uint16(elem.size)
   110  	c.elemtype = elem
   111  	c.dataqsiz = uint(size)
   112  	lockInit(&c.lock, lockRankHchan)
   113  
   114  	if debugChan {
   115  		print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
   116  	}
   117  	return c
   118  }
   119  
   120  // chanbuf(c, i) is pointer to the i'th slot in the buffer.
   121  func chanbuf(c *hchan, i uint) unsafe.Pointer {
   122  	return add(c.buf, uintptr(i)*uintptr(c.elemsize))
   123  }
   124  
   125  // full reports whether a send on c would block (that is, the channel is full).
   126  // It uses a single word-sized read of mutable state, so although
   127  // the answer is instantaneously true, the correct answer may have changed
   128  // by the time the calling function receives the return value.
   129  func full(c *hchan) bool {
   130  	// c.dataqsiz is immutable (never written after the channel is created)
   131  	// so it is safe to read at any time during channel operation.
   132  	if c.dataqsiz == 0 {
   133  		// Assumes that a pointer read is relaxed-atomic.
   134  		return c.recvq.first == nil
   135  	}
   136  	// Assumes that a uint read is relaxed-atomic.
   137  	return c.qcount == c.dataqsiz
   138  }
   139  
   140  // entry point for c <- x from compiled code
   141  //go:nosplit
   142  func chansend1(c *hchan, elem unsafe.Pointer) {
   143  	chansend(c, elem, true, getcallerpc())
   144  }
   145  
   146  /*
   147   * generic single channel send/recv
   148   * If block is not nil,
   149   * then the protocol will not
   150   * sleep but return if it could
   151   * not complete.
   152   *
   153   * sleep can wake up with g.param == nil
   154   * when a channel involved in the sleep has
   155   * been closed.  it is easiest to loop and re-run
   156   * the operation; we'll see that it's now closed.
   157   */
   158  func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
   159  	if c == nil {
   160  		if !block {
   161  			return false
   162  		}
   163  		gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
   164  		throw("unreachable")
   165  	}
   166  
   167  	if debugChan {
   168  		print("chansend: chan=", c, "\n")
   169  	}
   170  
   171  	if raceenabled {
   172  		racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
   173  	}
   174  
   175  	// Fast path: check for failed non-blocking operation without acquiring the lock.
   176  	//
   177  	// After observing that the channel is not closed, we observe that the channel is
   178  	// not ready for sending. Each of these observations is a single word-sized read
   179  	// (first c.closed and second full()).
   180  	// Because a closed channel cannot transition from 'ready for sending' to
   181  	// 'not ready for sending', even if the channel is closed between the two observations,
   182  	// they imply a moment between the two when the channel was both not yet closed
   183  	// and not ready for sending. We behave as if we observed the channel at that moment,
   184  	// and report that the send cannot proceed.
   185  	//
   186  	// It is okay if the reads are reordered here: if we observe that the channel is not
   187  	// ready for sending and then observe that it is not closed, that implies that the
   188  	// channel wasn't closed during the first observation. However, nothing here
   189  	// guarantees forward progress. We rely on the side effects of lock release in
   190  	// chanrecv() and closechan() to update this thread's view of c.closed and full().
   191  	if !block && c.closed == 0 && full(c) {
   192  		return false
   193  	}
   194  
   195  	var t0 int64
   196  	if blockprofilerate > 0 {
   197  		t0 = cputicks()
   198  	}
   199  
   200  	lock(&c.lock)
   201  
   202  	if c.closed != 0 {
   203  		unlock(&c.lock)
   204  		panic(plainError("send on closed channel"))
   205  	}
   206  
   207  	if sg := c.recvq.dequeue(); sg != nil {
   208  		// Found a waiting receiver. We pass the value we want to send
   209  		// directly to the receiver, bypassing the channel buffer (if any).
   210  		send(c, sg, ep, func() { unlock(&c.lock) }, 3)
   211  		return true
   212  	}
   213  
   214  	if c.qcount < c.dataqsiz {
   215  		// Space is available in the channel buffer. Enqueue the element to send.
   216  		qp := chanbuf(c, c.sendx)
   217  		if raceenabled {
   218  			racenotify(c, c.sendx, nil)
   219  		}
   220  		typedmemmove(c.elemtype, qp, ep)
   221  		c.sendx++
   222  		if c.sendx == c.dataqsiz {
   223  			c.sendx = 0
   224  		}
   225  		c.qcount++
   226  		unlock(&c.lock)
   227  		return true
   228  	}
   229  
   230  	if !block {
   231  		unlock(&c.lock)
   232  		return false
   233  	}
   234  
   235  	// Block on the channel. Some receiver will complete our operation for us.
   236  	gp := getg()
   237  	mysg := acquireSudog()
   238  	mysg.releasetime = 0
   239  	if t0 != 0 {
   240  		mysg.releasetime = -1
   241  	}
   242  	// No stack splits between assigning elem and enqueuing mysg
   243  	// on gp.waiting where copystack can find it.
   244  	mysg.elem = ep
   245  	mysg.waitlink = nil
   246  	mysg.g = gp
   247  	mysg.isSelect = false
   248  	mysg.c = c
   249  	gp.waiting = mysg
   250  	gp.param = nil
   251  	c.sendq.enqueue(mysg)
   252  	// Signal to anyone trying to shrink our stack that we're about
   253  	// to park on a channel. The window between when this G's status
   254  	// changes and when we set gp.activeStackChans is not safe for
   255  	// stack shrinking.
   256  	atomic.Store8(&gp.parkingOnChan, 1)
   257  	gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
   258  	// Ensure the value being sent is kept alive until the
   259  	// receiver copies it out. The sudog has a pointer to the
   260  	// stack object, but sudogs aren't considered as roots of the
   261  	// stack tracer.
   262  	KeepAlive(ep)
   263  
   264  	// someone woke us up.
   265  	if mysg != gp.waiting {
   266  		throw("G waiting list is corrupted")
   267  	}
   268  	gp.waiting = nil
   269  	gp.activeStackChans = false
   270  	closed := !mysg.success
   271  	gp.param = nil
   272  	if mysg.releasetime > 0 {
   273  		blockevent(mysg.releasetime-t0, 2)
   274  	}
   275  	mysg.c = nil
   276  	releaseSudog(mysg)
   277  	if closed {
   278  		if c.closed == 0 {
   279  			throw("chansend: spurious wakeup")
   280  		}
   281  		panic(plainError("send on closed channel"))
   282  	}
   283  	return true
   284  }
   285  
   286  // send processes a send operation on an empty channel c.
   287  // The value ep sent by the sender is copied to the receiver sg.
   288  // The receiver is then woken up to go on its merry way.
   289  // Channel c must be empty and locked.  send unlocks c with unlockf.
   290  // sg must already be dequeued from c.
   291  // ep must be non-nil and point to the heap or the caller's stack.
   292  func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
   293  	if raceenabled {
   294  		if c.dataqsiz == 0 {
   295  			racesync(c, sg)
   296  		} else {
   297  			// Pretend we go through the buffer, even though
   298  			// we copy directly. Note that we need to increment
   299  			// the head/tail locations only when raceenabled.
   300  			racenotify(c, c.recvx, nil)
   301  			racenotify(c, c.recvx, sg)
   302  			c.recvx++
   303  			if c.recvx == c.dataqsiz {
   304  				c.recvx = 0
   305  			}
   306  			c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
   307  		}
   308  	}
   309  	if sg.elem != nil {
   310  		sendDirect(c.elemtype, sg, ep)
   311  		sg.elem = nil
   312  	}
   313  	gp := sg.g
   314  	unlockf()
   315  	gp.param = unsafe.Pointer(sg)
   316  	sg.success = true
   317  	if sg.releasetime != 0 {
   318  		sg.releasetime = cputicks()
   319  	}
   320  	goready(gp, skip+1)
   321  }
   322  
   323  // Sends and receives on unbuffered or empty-buffered channels are the
   324  // only operations where one running goroutine writes to the stack of
   325  // another running goroutine. The GC assumes that stack writes only
   326  // happen when the goroutine is running and are only done by that
   327  // goroutine. Using a write barrier is sufficient to make up for
   328  // violating that assumption, but the write barrier has to work.
   329  // typedmemmove will call bulkBarrierPreWrite, but the target bytes
   330  // are not in the heap, so that will not help. We arrange to call
   331  // memmove and typeBitsBulkBarrier instead.
   332  
   333  func sendDirect(t *_type, sg *sudog, src unsafe.Pointer) {
   334  	// src is on our stack, dst is a slot on another stack.
   335  
   336  	// Once we read sg.elem out of sg, it will no longer
   337  	// be updated if the destination's stack gets copied (shrunk).
   338  	// So make sure that no preemption points can happen between read & use.
   339  	dst := sg.elem
   340  	typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.size)
   341  	// No need for cgo write barrier checks because dst is always
   342  	// Go memory.
   343  	memmove(dst, src, t.size)
   344  }
   345  
   346  func recvDirect(t *_type, sg *sudog, dst unsafe.Pointer) {
   347  	// dst is on our stack or the heap, src is on another stack.
   348  	// The channel is locked, so src will not move during this
   349  	// operation.
   350  	src := sg.elem
   351  	typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.size)
   352  	memmove(dst, src, t.size)
   353  }
   354  
   355  func closechan(c *hchan) {
   356  	if c == nil {
   357  		panic(plainError("close of nil channel"))
   358  	}
   359  
   360  	lock(&c.lock)
   361  	if c.closed != 0 {
   362  		unlock(&c.lock)
   363  		panic(plainError("close of closed channel"))
   364  	}
   365  
   366  	if raceenabled {
   367  		callerpc := getcallerpc()
   368  		racewritepc(c.raceaddr(), callerpc, funcPC(closechan))
   369  		racerelease(c.raceaddr())
   370  	}
   371  
   372  	c.closed = 1
   373  
   374  	var glist gList
   375  
   376  	// release all readers
   377  	for {
   378  		sg := c.recvq.dequeue()
   379  		if sg == nil {
   380  			break
   381  		}
   382  		if sg.elem != nil {
   383  			typedmemclr(c.elemtype, sg.elem)
   384  			sg.elem = nil
   385  		}
   386  		if sg.releasetime != 0 {
   387  			sg.releasetime = cputicks()
   388  		}
   389  		gp := sg.g
   390  		gp.param = unsafe.Pointer(sg)
   391  		sg.success = false
   392  		if raceenabled {
   393  			raceacquireg(gp, c.raceaddr())
   394  		}
   395  		glist.push(gp)
   396  	}
   397  
   398  	// release all writers (they will panic)
   399  	for {
   400  		sg := c.sendq.dequeue()
   401  		if sg == nil {
   402  			break
   403  		}
   404  		sg.elem = nil
   405  		if sg.releasetime != 0 {
   406  			sg.releasetime = cputicks()
   407  		}
   408  		gp := sg.g
   409  		gp.param = unsafe.Pointer(sg)
   410  		sg.success = false
   411  		if raceenabled {
   412  			raceacquireg(gp, c.raceaddr())
   413  		}
   414  		glist.push(gp)
   415  	}
   416  	unlock(&c.lock)
   417  
   418  	// Ready all Gs now that we've dropped the channel lock.
   419  	for !glist.empty() {
   420  		gp := glist.pop()
   421  		gp.schedlink = 0
   422  		goready(gp, 3)
   423  	}
   424  }
   425  
   426  // empty reports whether a read from c would block (that is, the channel is
   427  // empty).  It uses a single atomic read of mutable state.
   428  func empty(c *hchan) bool {
   429  	// c.dataqsiz is immutable.
   430  	if c.dataqsiz == 0 {
   431  		return atomic.Loadp(unsafe.Pointer(&c.sendq.first)) == nil
   432  	}
   433  	return atomic.Loaduint(&c.qcount) == 0
   434  }
   435  
   436  // entry points for <- c from compiled code
   437  //go:nosplit
   438  func chanrecv1(c *hchan, elem unsafe.Pointer) {
   439  	chanrecv(c, elem, true)
   440  }
   441  
   442  //go:nosplit
   443  func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {
   444  	_, received = chanrecv(c, elem, true)
   445  	return
   446  }
   447  
   448  // chanrecv receives on channel c and writes the received data to ep.
   449  // ep may be nil, in which case received data is ignored.
   450  // If block == false and no elements are available, returns (false, false).
   451  // Otherwise, if c is closed, zeros *ep and returns (true, false).
   452  // Otherwise, fills in *ep with an element and returns (true, true).
   453  // A non-nil ep must point to the heap or the caller's stack.
   454  func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
   455  	// raceenabled: don't need to check ep, as it is always on the stack
   456  	// or is new memory allocated by reflect.
   457  
   458  	if debugChan {
   459  		print("chanrecv: chan=", c, "\n")
   460  	}
   461  
   462  	if c == nil {
   463  		if !block {
   464  			return
   465  		}
   466  		gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
   467  		throw("unreachable")
   468  	}
   469  
   470  	// Fast path: check for failed non-blocking operation without acquiring the lock.
   471  	if !block && empty(c) {
   472  		// After observing that the channel is not ready for receiving, we observe whether the
   473  		// channel is closed.
   474  		//
   475  		// Reordering of these checks could lead to incorrect behavior when racing with a close.
   476  		// For example, if the channel was open and not empty, was closed, and then drained,
   477  		// reordered reads could incorrectly indicate "open and empty". To prevent reordering,
   478  		// we use atomic loads for both checks, and rely on emptying and closing to happen in
   479  		// separate critical sections under the same lock.  This assumption fails when closing
   480  		// an unbuffered channel with a blocked send, but that is an error condition anyway.
   481  		if atomic.Load(&c.closed) == 0 {
   482  			// Because a channel cannot be reopened, the later observation of the channel
   483  			// being not closed implies that it was also not closed at the moment of the
   484  			// first observation. We behave as if we observed the channel at that moment
   485  			// and report that the receive cannot proceed.
   486  			return
   487  		}
   488  		// The channel is irreversibly closed. Re-check whether the channel has any pending data
   489  		// to receive, which could have arrived between the empty and closed checks above.
   490  		// Sequential consistency is also required here, when racing with such a send.
   491  		if empty(c) {
   492  			// The channel is irreversibly closed and empty.
   493  			if raceenabled {
   494  				raceacquire(c.raceaddr())
   495  			}
   496  			if ep != nil {
   497  				typedmemclr(c.elemtype, ep)
   498  			}
   499  			return true, false
   500  		}
   501  	}
   502  
   503  	var t0 int64
   504  	if blockprofilerate > 0 {
   505  		t0 = cputicks()
   506  	}
   507  
   508  	lock(&c.lock)
   509  
   510  	if c.closed != 0 && c.qcount == 0 {
   511  		if raceenabled {
   512  			raceacquire(c.raceaddr())
   513  		}
   514  		unlock(&c.lock)
   515  		if ep != nil {
   516  			typedmemclr(c.elemtype, ep)
   517  		}
   518  		return true, false
   519  	}
   520  
   521  	if sg := c.sendq.dequeue(); sg != nil {
   522  		// Found a waiting sender. If buffer is size 0, receive value
   523  		// directly from sender. Otherwise, receive from head of queue
   524  		// and add sender's value to the tail of the queue (both map to
   525  		// the same buffer slot because the queue is full).
   526  		recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
   527  		return true, true
   528  	}
   529  
   530  	if c.qcount > 0 {
   531  		// Receive directly from queue
   532  		qp := chanbuf(c, c.recvx)
   533  		if raceenabled {
   534  			racenotify(c, c.recvx, nil)
   535  		}
   536  		if ep != nil {
   537  			typedmemmove(c.elemtype, ep, qp)
   538  		}
   539  		typedmemclr(c.elemtype, qp)
   540  		c.recvx++
   541  		if c.recvx == c.dataqsiz {
   542  			c.recvx = 0
   543  		}
   544  		c.qcount--
   545  		unlock(&c.lock)
   546  		return true, true
   547  	}
   548  
   549  	if !block {
   550  		unlock(&c.lock)
   551  		return false, false
   552  	}
   553  
   554  	// no sender available: block on this channel.
   555  	gp := getg()
   556  	mysg := acquireSudog()
   557  	mysg.releasetime = 0
   558  	if t0 != 0 {
   559  		mysg.releasetime = -1
   560  	}
   561  	// No stack splits between assigning elem and enqueuing mysg
   562  	// on gp.waiting where copystack can find it.
   563  	mysg.elem = ep
   564  	mysg.waitlink = nil
   565  	gp.waiting = mysg
   566  	mysg.g = gp
   567  	mysg.isSelect = false
   568  	mysg.c = c
   569  	gp.param = nil
   570  	c.recvq.enqueue(mysg)
   571  	// Signal to anyone trying to shrink our stack that we're about
   572  	// to park on a channel. The window between when this G's status
   573  	// changes and when we set gp.activeStackChans is not safe for
   574  	// stack shrinking.
   575  	atomic.Store8(&gp.parkingOnChan, 1)
   576  	gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)
   577  
   578  	// someone woke us up
   579  	if mysg != gp.waiting {
   580  		throw("G waiting list is corrupted")
   581  	}
   582  	gp.waiting = nil
   583  	gp.activeStackChans = false
   584  	if mysg.releasetime > 0 {
   585  		blockevent(mysg.releasetime-t0, 2)
   586  	}
   587  	success := mysg.success
   588  	gp.param = nil
   589  	mysg.c = nil
   590  	releaseSudog(mysg)
   591  	return true, success
   592  }
   593  
   594  // recv processes a receive operation on a full channel c.
   595  // There are 2 parts:
   596  // 1) The value sent by the sender sg is put into the channel
   597  //    and the sender is woken up to go on its merry way.
   598  // 2) The value received by the receiver (the current G) is
   599  //    written to ep.
   600  // For synchronous channels, both values are the same.
   601  // For asynchronous channels, the receiver gets its data from
   602  // the channel buffer and the sender's data is put in the
   603  // channel buffer.
   604  // Channel c must be full and locked. recv unlocks c with unlockf.
   605  // sg must already be dequeued from c.
   606  // A non-nil ep must point to the heap or the caller's stack.
   607  func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
   608  	if c.dataqsiz == 0 {
   609  		if raceenabled {
   610  			racesync(c, sg)
   611  		}
   612  		if ep != nil {
   613  			// copy data from sender
   614  			recvDirect(c.elemtype, sg, ep)
   615  		}
   616  	} else {
   617  		// Queue is full. Take the item at the
   618  		// head of the queue. Make the sender enqueue
   619  		// its item at the tail of the queue. Since the
   620  		// queue is full, those are both the same slot.
   621  		qp := chanbuf(c, c.recvx)
   622  		if raceenabled {
   623  			racenotify(c, c.recvx, nil)
   624  			racenotify(c, c.recvx, sg)
   625  		}
   626  		// copy data from queue to receiver
   627  		if ep != nil {
   628  			typedmemmove(c.elemtype, ep, qp)
   629  		}
   630  		// copy data from sender to queue
   631  		typedmemmove(c.elemtype, qp, sg.elem)
   632  		c.recvx++
   633  		if c.recvx == c.dataqsiz {
   634  			c.recvx = 0
   635  		}
   636  		c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
   637  	}
   638  	sg.elem = nil
   639  	gp := sg.g
   640  	unlockf()
   641  	gp.param = unsafe.Pointer(sg)
   642  	sg.success = true
   643  	if sg.releasetime != 0 {
   644  		sg.releasetime = cputicks()
   645  	}
   646  	goready(gp, skip+1)
   647  }
   648  
   649  func chanparkcommit(gp *g, chanLock unsafe.Pointer) bool {
   650  	// There are unlocked sudogs that point into gp's stack. Stack
   651  	// copying must lock the channels of those sudogs.
   652  	// Set activeStackChans here instead of before we try parking
   653  	// because we could self-deadlock in stack growth on the
   654  	// channel lock.
   655  	gp.activeStackChans = true
   656  	// Mark that it's safe for stack shrinking to occur now,
   657  	// because any thread acquiring this G's stack for shrinking
   658  	// is guaranteed to observe activeStackChans after this store.
   659  	atomic.Store8(&gp.parkingOnChan, 0)
   660  	// Make sure we unlock after setting activeStackChans and
   661  	// unsetting parkingOnChan. The moment we unlock chanLock
   662  	// we risk gp getting readied by a channel operation and
   663  	// so gp could continue running before everything before
   664  	// the unlock is visible (even to gp itself).
   665  	unlock((*mutex)(chanLock))
   666  	return true
   667  }
   668  
   669  // compiler implements
   670  //
   671  //	select {
   672  //	case c <- v:
   673  //		... foo
   674  //	default:
   675  //		... bar
   676  //	}
   677  //
   678  // as
   679  //
   680  //	if selectnbsend(c, v) {
   681  //		... foo
   682  //	} else {
   683  //		... bar
   684  //	}
   685  //
   686  func selectnbsend(c *hchan, elem unsafe.Pointer) (selected bool) {
   687  	return chansend(c, elem, false, getcallerpc())
   688  }
   689  
   690  // compiler implements
   691  //
   692  //	select {
   693  //	case v = <-c:
   694  //		... foo
   695  //	default:
   696  //		... bar
   697  //	}
   698  //
   699  // as
   700  //
   701  //	if selectnbrecv(&v, c) {
   702  //		... foo
   703  //	} else {
   704  //		... bar
   705  //	}
   706  //
   707  func selectnbrecv(elem unsafe.Pointer, c *hchan) (selected bool) {
   708  	selected, _ = chanrecv(c, elem, false)
   709  	return
   710  }
   711  
   712  // compiler implements
   713  //
   714  //	select {
   715  //	case v, ok = <-c:
   716  //		... foo
   717  //	default:
   718  //		... bar
   719  //	}
   720  //
   721  // as
   722  //
   723  //	if c != nil && selectnbrecv2(&v, &ok, c) {
   724  //		... foo
   725  //	} else {
   726  //		... bar
   727  //	}
   728  //
   729  func selectnbrecv2(elem unsafe.Pointer, received *bool, c *hchan) (selected bool) {
   730  	// TODO(khr): just return 2 values from this function, now that it is in Go.
   731  	selected, *received = chanrecv(c, elem, false)
   732  	return
   733  }
   734  
   735  //go:linkname reflect_chansend reflect.chansend
   736  func reflect_chansend(c *hchan, elem unsafe.Pointer, nb bool) (selected bool) {
   737  	return chansend(c, elem, !nb, getcallerpc())
   738  }
   739  
   740  //go:linkname reflect_chanrecv reflect.chanrecv
   741  func reflect_chanrecv(c *hchan, nb bool, elem unsafe.Pointer) (selected bool, received bool) {
   742  	return chanrecv(c, elem, !nb)
   743  }
   744  
   745  //go:linkname reflect_chanlen reflect.chanlen
   746  func reflect_chanlen(c *hchan) int {
   747  	if c == nil {
   748  		return 0
   749  	}
   750  	return int(c.qcount)
   751  }
   752  
   753  //go:linkname reflectlite_chanlen internal/reflectlite.chanlen
   754  func reflectlite_chanlen(c *hchan) int {
   755  	if c == nil {
   756  		return 0
   757  	}
   758  	return int(c.qcount)
   759  }
   760  
   761  //go:linkname reflect_chancap reflect.chancap
   762  func reflect_chancap(c *hchan) int {
   763  	if c == nil {
   764  		return 0
   765  	}
   766  	return int(c.dataqsiz)
   767  }
   768  
   769  //go:linkname reflect_chanclose reflect.chanclose
   770  func reflect_chanclose(c *hchan) {
   771  	closechan(c)
   772  }
   773  
   774  func (q *waitq) enqueue(sgp *sudog) {
   775  	sgp.next = nil
   776  	x := q.last
   777  	if x == nil {
   778  		sgp.prev = nil
   779  		q.first = sgp
   780  		q.last = sgp
   781  		return
   782  	}
   783  	sgp.prev = x
   784  	x.next = sgp
   785  	q.last = sgp
   786  }
   787  
   788  func (q *waitq) dequeue() *sudog {
   789  	for {
   790  		sgp := q.first
   791  		if sgp == nil {
   792  			return nil
   793  		}
   794  		y := sgp.next
   795  		if y == nil {
   796  			q.first = nil
   797  			q.last = nil
   798  		} else {
   799  			y.prev = nil
   800  			q.first = y
   801  			sgp.next = nil // mark as removed (see dequeueSudog)
   802  		}
   803  
   804  		// if a goroutine was put on this queue because of a
   805  		// select, there is a small window between the goroutine
   806  		// being woken up by a different case and it grabbing the
   807  		// channel locks. Once it has the lock
   808  		// it removes itself from the queue, so we won't see it after that.
   809  		// We use a flag in the G struct to tell us when someone
   810  		// else has won the race to signal this goroutine but the goroutine
   811  		// hasn't removed itself from the queue yet.
   812  		if sgp.isSelect && !atomic.Cas(&sgp.g.selectDone, 0, 1) {
   813  			continue
   814  		}
   815  
   816  		return sgp
   817  	}
   818  }
   819  
   820  func (c *hchan) raceaddr() unsafe.Pointer {
   821  	// Treat read-like and write-like operations on the channel to
   822  	// happen at this address. Avoid using the address of qcount
   823  	// or dataqsiz, because the len() and cap() builtins read
   824  	// those addresses, and we don't want them racing with
   825  	// operations like close().
   826  	return unsafe.Pointer(&c.buf)
   827  }
   828  
   829  func racesync(c *hchan, sg *sudog) {
   830  	racerelease(chanbuf(c, 0))
   831  	raceacquireg(sg.g, chanbuf(c, 0))
   832  	racereleaseg(sg.g, chanbuf(c, 0))
   833  	raceacquire(chanbuf(c, 0))
   834  }
   835  
   836  // Notify the race detector of a send or receive involving buffer entry idx
   837  // and a channel c or its communicating partner sg.
   838  // This function handles the special case of c.elemsize==0.
   839  func racenotify(c *hchan, idx uint, sg *sudog) {
   840  	// We could have passed the unsafe.Pointer corresponding to entry idx
   841  	// instead of idx itself.  However, in a future version of this function,
   842  	// we can use idx to better handle the case of elemsize==0.
   843  	// A future improvement to the detector is to call TSan with c and idx:
   844  	// this way, Go will continue to not allocating buffer entries for channels
   845  	// of elemsize==0, yet the race detector can be made to handle multiple
   846  	// sync objects underneath the hood (one sync object per idx)
   847  	qp := chanbuf(c, idx)
   848  	// When elemsize==0, we don't allocate a full buffer for the channel.
   849  	// Instead of individual buffer entries, the race detector uses the
   850  	// c.buf as the only buffer entry.  This simplification prevents us from
   851  	// following the memory model's happens-before rules (rules that are
   852  	// implemented in racereleaseacquire).  Instead, we accumulate happens-before
   853  	// information in the synchronization object associated with c.buf.
   854  	if c.elemsize == 0 {
   855  		if sg == nil {
   856  			raceacquire(qp)
   857  			racerelease(qp)
   858  		} else {
   859  			raceacquireg(sg.g, qp)
   860  			racereleaseg(sg.g, qp)
   861  		}
   862  	} else {
   863  		if sg == nil {
   864  			racereleaseacquire(qp)
   865  		} else {
   866  			racereleaseacquireg(sg.g, qp)
   867  		}
   868  	}
   869  }
   870
View as plain text