chan.go

Documentation: runtime

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