Go:关于 Channel

写在前面

本篇主要是通过 Channel 的模型图，对 Channel 的原理做一个基本的概述

内容

模型图与代码

我们先来看下 Channel 的模型图：

以上的图是一个简要的模型图，意味着丢失一些细节，我们再结合源码来看下：

type hchan struct {
    // 以下是环形缓冲区
	qcount   uint           // total data in the queue
	dataqsiz uint           // size of the circular queue
	buf      unsafe.Pointer // points to an array of dataqsiz elements
	elemsize uint16
	elemtype *_type // element type
	// Channel 的关闭状态
	closed   uint32
	// 以下是等待队列和接收队列
	sendx    uint   // send index
	recvx    uint   // receive index
	recvq    waitq  // list of recv waiters
	sendq    waitq  // list of send waiters
	// 以下是 Channel 的锁
	lock mutex
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

可以看到，模型图相比于源码，主要是少了 closed 和 lock 这两个属性。
从模型图里我们可以看到，Channel 的构造主要有三部分：

• 发送队列
• 接收队列
• 缓冲区

因此我们经常遇到的问题就出现在这三部分是否有无的排列组合，例如发送队列里有等待协程，接收队列里有等待协程，无缓冲区这样。

接下来我们结合模型图，并分别从发送流程和接收流程来看 Channel 的一个基本运作流程。至于模型图里没有包括的部分（锁和关闭状态）就作为补充处理。
（注：Channel 的源码位于 runtime 包下的 chan.go 文件）

发送流程

发送流程里，主要涉及到的方法是

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool
1

假设我们现在要发送一个数据，那么这个数据就可能有这么几种状态

• 在发送等待队列里
• 在缓冲区里
• 被接收队列给拿走了

在源码里，发送数据的时候，总体流程上是

我们可以看下：

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
    ...

	lock(&c.lock)

    // 向一个已经关闭的通道发送数据，会 panic
	if c.closed != 0 {
		unlock(&c.lock)
		panic(plainError("send on closed channel"))
	}
    
    // 会先检查接收队列里是否有等待接收的协程，如果有等待协程，我们就可以忽略缓冲区的两种情况：
    // - 没有缓冲区
    // - 缓冲区满
    // 那么意味着这里我们的数据是直接跟接收队列对接，既然接收队列里有等待协程，那就把数据给他就行了
	if sg := c.recvq.dequeue(); sg != nil {
		send(c, sg, ep, func() { unlock(&c.lock) }, 3)
		return true
	}

    // 到了这一步，就意味着不考虑接收队列了，因为没有等待协程可以用
    // 于是就看下是否有缓冲区了，有缓冲区的话，就把数据放到缓冲区即可
	if c.qcount < c.dataqsiz {
		// Space is available in the channel buffer. Enqueue the element to send.
		qp := chanbuf(c, c.sendx)
		if raceenabled {
			racenotify(c, c.sendx, nil)
		}
		typedmemmove(c.elemtype, qp, ep)
		c.sendx++
		if c.sendx == c.dataqsiz {
			c.sendx = 0
		}
		c.qcount++
		unlock(&c.lock)
		return true
	}
    ...
	
	// 到了这里，说明不考虑接收队列和缓冲区了，那就是要进入发送等待队列了
	gp := getg()
	mysg := acquireSudog()
	mysg.releasetime = 0
	if t0 != 0 {
		mysg.releasetime = -1
	}
	mysg.elem = ep
	mysg.waitlink = nil
	mysg.g = gp
	mysg.isSelect = false
	mysg.c = c
	gp.waiting = mysg
	gp.param = nil
	// 加到发送等待队列里，然后就进入阻塞状态
	c.sendq.enqueue(mysg)

	atomic.Store8(&gp.parkingOnChan, 1)
	gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)

	KeepAlive(ep)

	// someone woke us up.
	if mysg != gp.waiting {
		throw("G waiting list is corrupted")
	}
	gp.waiting = nil
	gp.activeStackChans = false
	closed := !mysg.success
	gp.param = nil
	if mysg.releasetime > 0 {
		blockevent(mysg.releasetime-t0, 2)
	}
	mysg.c = nil
	releaseSudog(mysg)
	// 再检查一下通道是否关闭了，向一个已经关闭的通道发送数据，会 panic
	if closed {
		if c.closed == 0 {
			throw("chansend: spurious wakeup")
		}
		panic(plainError("send on closed channel"))
	}
	return true
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

这里面关于 closed 和 lock 有一些细节：

• 在发送开始的时候，我们需要上锁，等数据要么被拿走，要么进入缓冲区了，要么进入等待队列了，再解锁
  • 也就是操作 channel 的发送是需要加锁的
• 如果一个 channel 被关闭了，此时还向它发送数据，会发生 panic

接收流程

接收数据里，主要涉及的方法是：

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool)
1

那么从接收数据的角度来看，要取一个数据，就会出现几种情况

• 取到数据
  • 从缓冲区取
  • 从等待发送队列里取
• 取不到数据
  • 进入等待接收队列

在源码里，它的总体流程是：

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
	... 
	lock(&c.lock)

     // 如果 channel 被关闭
	if c.closed != 0 {
	    // 如果缓冲区里也没有数据，就直接 return 即可
		if c.qcount == 0 {
			... 
			unlock(&c.lock)
			if ep != nil {
				typedmemclr(c.elemtype, ep)
			}
			return true, false
		}
	} else {
	    // 如果 channel 没关闭
		// 看下等待发送队列里是否有等待协程
		// 如果有等待发送的协程，那就再去看下缓冲区
		// 如果没有缓冲区，或是缓冲区没数据，就直接从发送等待队列里拿数据
		// 如果有缓冲区且有数据，就从缓冲区里拿数据，再把等待队列里的数据追加到缓冲队列的末尾
		if sg := c.sendq.dequeue(); sg != nil {
			recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
			return true, true
		}
	}

    // 到这里就认为不去看接收队列了，直接看缓冲区
    // 缓冲区有数据，就直接取
	if c.qcount > 0 {
		// Receive directly from queue
		qp := chanbuf(c, c.recvx)
		if raceenabled {
			racenotify(c, c.recvx, nil)
		}
		if ep != nil {
			typedmemmove(c.elemtype, ep, qp)
		}
		typedmemclr(c.elemtype, qp)
		c.recvx++
		if c.recvx == c.dataqsiz {
			c.recvx = 0
		}
		c.qcount--
		unlock(&c.lock)
		return true, true
	}

	...

	// 想取数据却没地方可取，那就加入接收等待队列，然后就进入阻塞状态了
	gp := getg()
	mysg := acquireSudog()
	mysg.releasetime = 0
	if t0 != 0 {
		mysg.releasetime = -1
	}
	
	mysg.elem = ep
	mysg.waitlink = nil
	gp.waiting = mysg
	mysg.g = gp
	mysg.isSelect = false
	mysg.c = c
	gp.param = nil
	c.recvq.enqueue(mysg)
	
	atomic.Store8(&gp.parkingOnChan, 1)
	gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)

	// someone woke us up
	if mysg != gp.waiting {
		throw("G waiting list is corrupted")
	}
	gp.waiting = nil
	gp.activeStackChans = false
	if mysg.releasetime > 0 {
		blockevent(mysg.releasetime-t0, 2)
	}
	success := mysg.success
	gp.param = nil
	mysg.c = nil
	releaseSudog(mysg)
	return true, success
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

同样的，接收的过程里，也有 closed 和 lock 的一些细节：

• 在接收过程前，需要加锁，处理完后再解锁
• 从一个已经关闭的 channel 里取数据，不会造成 panic