Go源码学习: 从channel中接收数据的内部实现是什么样的?

1v <- ch
2v, ok <- ch

1runtime.chanrecv1() /usr/local/Cellar/go/1.17.2/libexec/src/runtime/chan.go:439 (PC: 0x100448e)
2Warning: debugging optimized function
3   434: }
4   435:
5   436: // entry points for <- c from compiled code
6   437: //go:nosplit
7   438: func chanrecv1(c *hchan, elem unsafe.Pointer) {
8=> 439:         chanrecv(c, elem, true)
9   440: }

 1> runtime.chanrecv2() /usr/local/Cellar/go/1.17.2/libexec/src/runtime/chan.go:444 (PC: 0x1004e0e)
 2Warning: debugging optimized function
 3   439:         chanrecv(c, elem, true)
 4   440: }
 5   441:
 6   442: //go:nosplit
 7   443: func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {
 8=> 444:         _, received = chanrecv(c, elem, true)
 9   445:         return
10   446: }

1 runtime.chanrecv() /usr/local/Cellar/go/1.17.2/libexec/src/runtime/chan.go:454 (PC: 0x1004e4f)
2Warning: debugging optimized function
3   449: // ep may be nil, in which case received data is ignored.
4   450: // If block == false and no elements are available, returns (false, false).
5   451: // Otherwise, if c is closed, zeros *ep and returns (true, false).
6   452: // Otherwise, fills in *ep with an element and returns (true, true).
7   453: // A non-nil ep must point to the heap or the caller's stack.
8=> 454: func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {

1	if c == nil {
2		if !block {
3			return
4		}
5		gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
6		throw("unreachable")
7	}

 1	if !block && empty(c) {
 2		if atomic.Load(&c.closed) == 0 {
 3			return
 4		}
 5
 6		if empty(c) {
 7			if raceenabled {
 8				raceacquire(c.raceaddr())
 9			}
10			if ep != nil {
11				typedmemclr(c.elemtype, ep)
12			}
13			return true, false
14		}
15	}

1func empty(c *hchan) bool {
2	// c.dataqsiz is immutable.
3	if c.dataqsiz == 0 {
4		return atomic.Loadp(unsafe.Pointer(&c.sendq.first)) == nil
5	}
6	return atomic.Loaduint(&c.qcount) == 0
7}

 1	lock(&c.lock)
 2
 3	if c.closed != 0 && c.qcount == 0 {
 4		if raceenabled {
 5			raceacquire(c.raceaddr())
 6		}
 7		unlock(&c.lock)
 8		if ep != nil {
 9			typedmemclr(c.elemtype, ep)
10		}
11		return true, false
12	}

1	if sg := c.sendq.dequeue(); sg != nil {
2		// Found a waiting sender. If buffer is size 0, receive value
3		// directly from sender. Otherwise, receive from head of queue
4		// and add sender's value to the tail of the queue (both map to
5		// the same buffer slot because the queue is full).
6		recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
7		return true, true
8	}

 1func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
 2	if c.dataqsiz == 0 {
 3		if raceenabled {
 4			racesync(c, sg)
 5		}
 6		if ep != nil {
 7			// copy data from sender
 8			recvDirect(c.elemtype, sg, ep)
 9		}
10	} else {
11		// Queue is full. Take the item at the
12		// head of the queue. Make the sender enqueue
13		// its item at the tail of the queue. Since the
14		// queue is full, those are both the same slot.
15		qp := chanbuf(c, c.recvx)
16		if raceenabled {
17			racenotify(c, c.recvx, nil)
18			racenotify(c, c.recvx, sg)
19		}
20		// copy data from queue to receiver
21		if ep != nil {
22			typedmemmove(c.elemtype, ep, qp)
23		}
24		// copy data from sender to queue
25		typedmemmove(c.elemtype, qp, sg.elem)
26		c.recvx++
27		if c.recvx == c.dataqsiz {
28			c.recvx = 0
29		}
30		c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
31	}
32	sg.elem = nil
33	gp := sg.g
34	unlockf()
35	gp.param = unsafe.Pointer(sg)
36	sg.success = true
37	if sg.releasetime != 0 {
38		sg.releasetime = cputicks()
39	}
40	goready(gp, skip+1)
41}

 1	if c.qcount > 0 {
 2		// Receive directly from queue
 3		qp := chanbuf(c, c.recvx)
 4		if raceenabled {
 5			racenotify(c, c.recvx, nil)
 6		}
 7		if ep != nil {
 8			typedmemmove(c.elemtype, ep, qp)
 9		}
10		typedmemclr(c.elemtype, qp)
11		c.recvx++
12		if c.recvx == c.dataqsiz {
13			c.recvx = 0
14		}
15		c.qcount--
16		unlock(&c.lock)
17		return true, true
18	}

1	if !block {
2		unlock(&c.lock)
3		return false, false
4	}

 1	// no sender available: block on this channel.
 2	gp := getg()
 3	mysg := acquireSudog()
 4	mysg.releasetime = 0
 5	if t0 != 0 {
 6		mysg.releasetime = -1
 7	}
 8	// No stack splits between assigning elem and enqueuing mysg
 9	// on gp.waiting where copystack can find it.
10	mysg.elem = ep
11	mysg.waitlink = nil
12	gp.waiting = mysg
13	mysg.g = gp
14	mysg.isSelect = false
15	mysg.c = c
16	gp.param = nil
17	c.recvq.enqueue(mysg)
18	// Signal to anyone trying to shrink our stack that we're about
19	// to park on a channel. The window between when this G's status
20	// changes and when we set gp.activeStackChans is not safe for
21	// stack shrinking.
22	atomic.Store8(&gp.parkingOnChan, 1)
23	gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)
24
25	// someone woke us up
26	if mysg != gp.waiting {
27		throw("G waiting list is corrupted")
28	}
29	gp.waiting = nil
30	gp.activeStackChans = false
31	if mysg.releasetime > 0 {
32		blockevent(mysg.releasetime-t0, 2)
33	}
34	success := mysg.success
35	gp.param = nil
36	mysg.c = nil
37	releaseSudog(mysg)
38	return true, success