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.
7 // This file contains the implementation of Go channels.
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.
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.
21 "runtime/internal/atomic"
25 // For gccgo, use go:linkname to rename compiler-called functions to
26 // themselves, so that the compiler will export them.
28 //go:linkname makechan runtime.makechan
29 //go:linkname makechan64 runtime.makechan64
30 //go:linkname chansend1 runtime.chansend1
31 //go:linkname chanrecv1 runtime.chanrecv1
32 //go:linkname chanrecv2 runtime.chanrecv2
33 //go:linkname closechan runtime.closechan
37 hchanSize
= unsafe
.Sizeof(hchan
{}) + uintptr(-int(unsafe
.Sizeof(hchan
{}))&(maxAlign
-1))
42 qcount
uint // total data in the queue
43 dataqsiz
uint // size of the circular queue
44 buf unsafe
.Pointer
// points to an array of dataqsiz elements
47 elemtype
*_type
// element type
48 sendx
uint // send index
49 recvx
uint // receive index
50 recvq waitq
// list of recv waiters
51 sendq waitq
// list of send waiters
53 // lock protects all fields in hchan, as well as several
54 // fields in sudogs blocked on this channel.
56 // Do not change another G's status while holding this lock
57 // (in particular, do not ready a G), as this can deadlock
58 // with stack shrinking.
67 //go:linkname reflect_makechan reflect.makechan
68 func reflect_makechan(t
*chantype
, size
int) *hchan
{
69 return makechan(t
, size
)
72 func makechan64(t
*chantype
, size
int64) *hchan
{
73 if int64(int(size
)) != size
{
74 panic(plainError("makechan: size out of range"))
77 return makechan(t
, int(size
))
80 func makechan(t
*chantype
, size
int) *hchan
{
83 // compiler checks this but be safe.
84 if elem
.size
>= 1<<16 {
85 throw("makechan: invalid channel element type")
87 if hchanSize%maxAlign
!= 0 || elem
.align
> maxAlign
{
88 throw("makechan: bad alignment")
91 if size
< 0 ||
uintptr(size
) > maxSliceCap(elem
.size
) ||
uintptr(size
)*elem
.size
> _MaxMem
-hchanSize
{
92 panic(plainError("makechan: size out of range"))
95 // Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.
96 // buf points into the same allocation, elemtype is persistent.
97 // SudoG's are referenced from their owning thread so they can't be collected.
98 // TODO(dvyukov,rlh): Rethink when collector can move allocated objects.
101 case size
== 0 || elem
.size
== 0:
102 // Queue or element size is zero.
103 c
= (*hchan
)(mallocgc(hchanSize
, nil, true))
104 // Race detector uses this location for synchronization.
105 c
.buf
= unsafe
.Pointer(c
)
106 case elem
.kind
&kindNoPointers
!= 0:
107 // Elements do not contain pointers.
108 // Allocate hchan and buf in one call.
109 c
= (*hchan
)(mallocgc(hchanSize
+uintptr(size
)*elem
.size
, nil, true))
110 c
.buf
= add(unsafe
.Pointer(c
), hchanSize
)
112 // Elements contain pointers.
114 c
.buf
= mallocgc(uintptr(size
)*elem
.size
, elem
, true)
117 c
.elemsize
= uint16(elem
.size
)
119 c
.dataqsiz
= uint(size
)
122 print("makechan: chan=", c
, "; elemsize=", elem
.size
, "; dataqsiz=", size
, "\n")
127 // chanbuf(c, i) is pointer to the i'th slot in the buffer.
128 func chanbuf(c
*hchan
, i
uint) unsafe
.Pointer
{
129 return add(c
.buf
, uintptr(i
)*uintptr(c
.elemsize
))
132 // entry point for c <- x from compiled code
134 func chansend1(c
*hchan
, elem unsafe
.Pointer
) {
135 chansend(c
, elem
, true, getcallerpc())
139 * generic single channel send/recv
140 * If block is not nil,
141 * then the protocol will not
142 * sleep but return if it could
145 * sleep can wake up with g.param == nil
146 * when a channel involved in the sleep has
147 * been closed. it is easiest to loop and re-run
148 * the operation; we'll see that it's now closed.
150 func chansend(c
*hchan
, ep unsafe
.Pointer
, block
bool, callerpc
uintptr) bool {
151 // Check preemption, since unlike gc we don't check on every call.
160 gopark(nil, nil, "chan send (nil chan)", traceEvGoStop
, 2)
165 print("chansend: chan=", c
, "\n")
169 racereadpc(unsafe
.Pointer(c
), callerpc
, funcPC(chansend
))
172 // Fast path: check for failed non-blocking operation without acquiring the lock.
174 // After observing that the channel is not closed, we observe that the channel is
175 // not ready for sending. Each of these observations is a single word-sized read
176 // (first c.closed and second c.recvq.first or c.qcount depending on kind of channel).
177 // Because a closed channel cannot transition from 'ready for sending' to
178 // 'not ready for sending', even if the channel is closed between the two observations,
179 // they imply a moment between the two when the channel was both not yet closed
180 // and not ready for sending. We behave as if we observed the channel at that moment,
181 // and report that the send cannot proceed.
183 // It is okay if the reads are reordered here: if we observe that the channel is not
184 // ready for sending and then observe that it is not closed, that implies that the
185 // channel wasn't closed during the first observation.
186 if !block
&& c
.closed == 0 && ((c
.dataqsiz
== 0 && c
.recvq
.first
== nil) ||
187 (c
.dataqsiz
> 0 && c
.qcount
== c
.dataqsiz
)) {
192 if blockprofilerate
> 0 {
200 panic(plainError("send on closed channel"))
203 if sg
:= c
.recvq
.dequeue(); sg
!= nil {
204 // Found a waiting receiver. We pass the value we want to send
205 // directly to the receiver, bypassing the channel buffer (if any).
206 send(c
, sg
, ep
, func() { unlock(&c
.lock
) }, 3)
210 if c
.qcount
< c
.dataqsiz
{
211 // Space is available in the channel buffer. Enqueue the element to send.
212 qp
:= chanbuf(c
, c
.sendx
)
217 typedmemmove(c
.elemtype
, qp
, ep
)
219 if c
.sendx
== c
.dataqsiz
{
232 // Block on the channel. Some receiver will complete our operation for us.
234 mysg
:= acquireSudog()
237 mysg
.releasetime
= -1
239 // No stack splits between assigning elem and enqueuing mysg
240 // on gp.waiting where copystack can find it.
244 mysg
.isSelect
= false
248 c
.sendq
.enqueue(mysg
)
249 goparkunlock(&c
.lock
, "chan send", traceEvGoBlockSend
, 3)
251 // someone woke us up.
252 if mysg
!= gp
.waiting
{
253 throw("G waiting list is corrupted")
258 throw("chansend: spurious wakeup")
260 panic(plainError("send on closed channel"))
263 if mysg
.releasetime
> 0 {
264 blockevent(mysg
.releasetime
-t0
, 2)
271 // send processes a send operation on an empty channel c.
272 // The value ep sent by the sender is copied to the receiver sg.
273 // The receiver is then woken up to go on its merry way.
274 // Channel c must be empty and locked. send unlocks c with unlockf.
275 // sg must already be dequeued from c.
276 // ep must be non-nil and point to the heap or the caller's stack.
277 func send(c
*hchan
, sg
*sudog
, ep unsafe
.Pointer
, unlockf
func(), skip
int) {
282 // Pretend we go through the buffer, even though
283 // we copy directly. Note that we need to increment
284 // the head/tail locations only when raceenabled.
285 qp
:= chanbuf(c
, c
.recvx
)
288 raceacquireg(sg
.g
, qp
)
289 racereleaseg(sg
.g
, qp
)
291 if c
.recvx
== c
.dataqsiz
{
294 c
.sendx
= c
.recvx
// c.sendx = (c.sendx+1) % c.dataqsiz
298 sendDirect(c
.elemtype
, sg
, ep
)
303 gp
.param
= unsafe
.Pointer(sg
)
304 if sg
.releasetime
!= 0 {
305 sg
.releasetime
= cputicks()
310 // Sends and receives on unbuffered or empty-buffered channels are the
311 // only operations where one running goroutine writes to the stack of
312 // another running goroutine. The GC assumes that stack writes only
313 // happen when the goroutine is running and are only done by that
314 // goroutine. Using a write barrier is sufficient to make up for
315 // violating that assumption, but the write barrier has to work.
316 // typedmemmove will call bulkBarrierPreWrite, but the target bytes
317 // are not in the heap, so that will not help. We arrange to call
318 // memmove and typeBitsBulkBarrier instead.
320 func sendDirect(t
*_type
, sg
*sudog
, src unsafe
.Pointer
) {
321 // src is on our stack, dst is a slot on another stack.
323 // Once we read sg.elem out of sg, it will no longer
324 // be updated if the destination's stack gets copied (shrunk).
325 // So make sure that no preemption points can happen between read & use.
327 typeBitsBulkBarrier(t
, uintptr(dst
), uintptr(src
), t
.size
)
328 memmove(dst
, src
, t
.size
)
331 func recvDirect(t
*_type
, sg
*sudog
, dst unsafe
.Pointer
) {
332 // dst is on our stack or the heap, src is on another stack.
333 // The channel is locked, so src will not move during this
336 typeBitsBulkBarrier(t
, uintptr(dst
), uintptr(src
), t
.size
)
337 memmove(dst
, src
, t
.size
)
340 func closechan(c
*hchan
) {
342 panic(plainError("close of nil channel"))
348 panic(plainError("close of closed channel"))
352 callerpc
:= getcallerpc()
353 racewritepc(unsafe
.Pointer(c
), callerpc
, funcPC(closechan
))
354 racerelease(unsafe
.Pointer(c
))
361 // release all readers
363 sg
:= c
.recvq
.dequeue()
368 typedmemclr(c
.elemtype
, sg
.elem
)
371 if sg
.releasetime
!= 0 {
372 sg
.releasetime
= cputicks()
377 raceacquireg(gp
, unsafe
.Pointer(c
))
379 gp
.schedlink
.set(glist
)
383 // release all writers (they will panic)
385 sg
:= c
.sendq
.dequeue()
390 if sg
.releasetime
!= 0 {
391 sg
.releasetime
= cputicks()
396 raceacquireg(gp
, unsafe
.Pointer(c
))
398 gp
.schedlink
.set(glist
)
403 // Ready all Gs now that we've dropped the channel lock.
406 glist
= glist
.schedlink
.ptr()
412 // entry points for <- c from compiled code
414 func chanrecv1(c
*hchan
, elem unsafe
.Pointer
) {
415 chanrecv(c
, elem
, true)
419 func chanrecv2(c
*hchan
, elem unsafe
.Pointer
) (received
bool) {
420 _
, received
= chanrecv(c
, elem
, true)
424 // chanrecv receives on channel c and writes the received data to ep.
425 // ep may be nil, in which case received data is ignored.
426 // If block == false and no elements are available, returns (false, false).
427 // Otherwise, if c is closed, zeros *ep and returns (true, false).
428 // Otherwise, fills in *ep with an element and returns (true, true).
429 // A non-nil ep must point to the heap or the caller's stack.
430 func chanrecv(c
*hchan
, ep unsafe
.Pointer
, block
bool) (selected
, received
bool) {
431 // raceenabled: don't need to check ep, as it is always on the stack
432 // or is new memory allocated by reflect.
435 print("chanrecv: chan=", c
, "\n")
438 // Check preemption, since unlike gc we don't check on every call.
447 gopark(nil, nil, "chan receive (nil chan)", traceEvGoStop
, 2)
451 // Fast path: check for failed non-blocking operation without acquiring the lock.
453 // After observing that the channel is not ready for receiving, we observe that the
454 // channel is not closed. Each of these observations is a single word-sized read
455 // (first c.sendq.first or c.qcount, and second c.closed).
456 // Because a channel cannot be reopened, the later observation of the channel
457 // being not closed implies that it was also not closed at the moment of the
458 // first observation. We behave as if we observed the channel at that moment
459 // and report that the receive cannot proceed.
461 // The order of operations is important here: reversing the operations can lead to
462 // incorrect behavior when racing with a close.
463 if !block
&& (c
.dataqsiz
== 0 && c
.sendq
.first
== nil ||
464 c
.dataqsiz
> 0 && atomic
.Loaduint(&c
.qcount
) == 0) &&
465 atomic
.Load(&c
.closed) == 0 {
470 if blockprofilerate
> 0 {
476 if c
.closed != 0 && c
.qcount
== 0 {
478 raceacquire(unsafe
.Pointer(c
))
482 typedmemclr(c
.elemtype
, ep
)
487 if sg
:= c
.sendq
.dequeue(); sg
!= nil {
488 // Found a waiting sender. If buffer is size 0, receive value
489 // directly from sender. Otherwise, receive from head of queue
490 // and add sender's value to the tail of the queue (both map to
491 // the same buffer slot because the queue is full).
492 recv(c
, sg
, ep
, func() { unlock(&c
.lock
) }, 3)
497 // Receive directly from queue
498 qp
:= chanbuf(c
, c
.recvx
)
504 typedmemmove(c
.elemtype
, ep
, qp
)
506 typedmemclr(c
.elemtype
, qp
)
508 if c
.recvx
== c
.dataqsiz
{
521 // no sender available: block on this channel.
523 mysg
:= acquireSudog()
526 mysg
.releasetime
= -1
528 // No stack splits between assigning elem and enqueuing mysg
529 // on gp.waiting where copystack can find it.
534 mysg
.isSelect
= false
537 c
.recvq
.enqueue(mysg
)
538 goparkunlock(&c
.lock
, "chan receive", traceEvGoBlockRecv
, 3)
540 // someone woke us up
541 if mysg
!= gp
.waiting
{
542 throw("G waiting list is corrupted")
545 if mysg
.releasetime
> 0 {
546 blockevent(mysg
.releasetime
-t0
, 2)
548 closed := gp
.param
== nil
555 // recv processes a receive operation on a full channel c.
556 // There are 2 parts:
557 // 1) The value sent by the sender sg is put into the channel
558 // and the sender is woken up to go on its merry way.
559 // 2) The value received by the receiver (the current G) is
561 // For synchronous channels, both values are the same.
562 // For asynchronous channels, the receiver gets its data from
563 // the channel buffer and the sender's data is put in the
565 // Channel c must be full and locked. recv unlocks c with unlockf.
566 // sg must already be dequeued from c.
567 // A non-nil ep must point to the heap or the caller's stack.
568 func recv(c
*hchan
, sg
*sudog
, ep unsafe
.Pointer
, unlockf
func(), skip
int) {
574 // copy data from sender
575 recvDirect(c
.elemtype
, sg
, ep
)
578 // Queue is full. Take the item at the
579 // head of the queue. Make the sender enqueue
580 // its item at the tail of the queue. Since the
581 // queue is full, those are both the same slot.
582 qp
:= chanbuf(c
, c
.recvx
)
586 raceacquireg(sg
.g
, qp
)
587 racereleaseg(sg
.g
, qp
)
589 // copy data from queue to receiver
591 typedmemmove(c
.elemtype
, ep
, qp
)
593 // copy data from sender to queue
594 typedmemmove(c
.elemtype
, qp
, sg
.elem
)
596 if c
.recvx
== c
.dataqsiz
{
599 c
.sendx
= c
.recvx
// c.sendx = (c.sendx+1) % c.dataqsiz
604 gp
.param
= unsafe
.Pointer(sg
)
605 if sg
.releasetime
!= 0 {
606 sg
.releasetime
= cputicks()
611 // compiler implements
622 // if selectnbsend(c, v) {
628 func selectnbsend(c
*hchan
, elem unsafe
.Pointer
) (selected
bool) {
629 return chansend(c
, elem
, false, getcallerpc())
632 // compiler implements
643 // if selectnbrecv(&v, c) {
649 func selectnbrecv(elem unsafe
.Pointer
, c
*hchan
) (selected
bool) {
650 selected
, _
= chanrecv(c
, elem
, false)
654 // compiler implements
665 // if c != nil && selectnbrecv2(&v, &ok, c) {
671 func selectnbrecv2(elem unsafe
.Pointer
, received
*bool, c
*hchan
) (selected
bool) {
672 // TODO(khr): just return 2 values from this function, now that it is in Go.
673 selected
, *received
= chanrecv(c
, elem
, false)
677 //go:linkname reflect_chansend reflect.chansend
678 func reflect_chansend(c
*hchan
, elem unsafe
.Pointer
, nb
bool) (selected
bool) {
679 return chansend(c
, elem
, !nb
, getcallerpc())
682 //go:linkname reflect_chanrecv reflect.chanrecv
683 func reflect_chanrecv(c
*hchan
, nb
bool, elem unsafe
.Pointer
) (selected
bool, received
bool) {
684 return chanrecv(c
, elem
, !nb
)
687 //go:linkname reflect_chanlen reflect.chanlen
688 func reflect_chanlen(c
*hchan
) int {
695 //go:linkname reflect_chancap reflect.chancap
696 func reflect_chancap(c
*hchan
) int {
700 return int(c
.dataqsiz
)
703 //go:linkname reflect_chanclose reflect.chanclose
704 func reflect_chanclose(c
*hchan
) {
708 func (q
*waitq
) enqueue(sgp
*sudog
) {
722 func (q
*waitq
) dequeue() *sudog
{
735 sgp
.next
= nil // mark as removed (see dequeueSudog)
738 // if a goroutine was put on this queue because of a
739 // select, there is a small window between the goroutine
740 // being woken up by a different case and it grabbing the
741 // channel locks. Once it has the lock
742 // it removes itself from the queue, so we won't see it after that.
743 // We use a flag in the G struct to tell us when someone
744 // else has won the race to signal this goroutine but the goroutine
745 // hasn't removed itself from the queue yet.
747 if !atomic
.Cas(&sgp
.g
.selectDone
, 0, 1) {
756 func racesync(c
*hchan
, sg
*sudog
) {
757 racerelease(chanbuf(c
, 0))
758 raceacquireg(sg
.g
, chanbuf(c
, 0))
759 racereleaseg(sg
.g
, chanbuf(c
, 0))
760 raceacquire(chanbuf(c
, 0))