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