chan.go

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