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