Go-Channel

Channel 是 Go 的核心数据结构和 Goroutine 之间的通信方式，支持 Go 的高性能并发编程模型。

设计原理

不通过共享内存的方式通信，而是通过通信的方式共享内存，采用的并发模式为：CSP（通信顺序进程）。Goroutine 和 Channel 分别对应 CSP 中的实体和传递信息的媒介，Goroutine 之间会通过 Channel 传递数据。

怎么理解“不通过共享内存的方式通信，而是通过通信的方式共享内存”这句话呢？

前半句指通过 sync 包里的一些组件进行并发编程；后半句指使用 channel 进行并发编程。实际上，channel 的底层就是通过 mutex 来控制并发的。只是 channel 是更高一层次的并发编程原语，封装了更多的功能。

上图中的两个 Goroutine，一个会向 Channel 中发送数据，另一个会从 Channel 中接收数据，它们两者能够独立运行并不存在直接关联，但是能通过 Channel 间接完成通信。

使用通信来共享内存是 Go 的设计哲学，那么为什么是这样的呢？

  1. 使用发送/接受消息来同步信息，相较于直接使用共享内存和互斥锁，是一种更高级的抽象，提供更好的封装；
  2. 便于将线程的职责分为生产者和消费者，通过消息传递的方式解耦；
  3. 消息发送的方式，保证同一时间只有一个活跃的线程能访问数据，从设计上天然地避免线程竞争和数据冲突。

FIFO

Channel的收发操作遵循“先进先出”：

  • 先从 Channel 读取数据的 Goroutine 会先接收到数据；
  • 先向 Channel 发送数据的 Goroutine 会得到先发送数据的权利。

无锁管道

锁作为一种并发控制技术，通常分为：乐观锁和悲观锁。无锁队列对应使用乐观锁的队列。

  • 乐观锁：本质是基于验证的协议，使用原子指令CAS在多线程之间同步数据。无锁队列的实现也依赖这一原子指令。

Channel在运行时的内部表示是runtime.hchan，包含互斥锁，因此是一个用于同步和通信的有锁队列。

然而，锁导致的休眠和唤醒将带来额外的上下文切换；如果临界区过大，加锁解锁导致的额外开销，将成为性能瓶颈。Go社区曾提出了无锁Channel实现方案，分为以下3种类型：

  1. 同步Channel（不带缓冲）：，发送方直接将数据交给Handoff接收方；
    • 同步模式下，当且仅当发送方和接收方都ready的情况下，数据才能传输（本质上是内存拷贝）；否则，任意一方先行进行发送或接收操作，都会被挂起，等待另一方的出现才能被唤醒。
  2. 异步Channel（带缓冲）：基于环形缓存的传统生产者消费者模型；
    • 在缓冲槽有剩余容量的情况下，发送和接收操作都可以顺利进行。否则，操作的一方（如写入）同样会被挂起，直到出现相反操作（如接收）才会被唤醒。
  3. chan struct{}类型的异步Channel：struct{}类型不占用内存空间，无需实现缓冲区和直接发送Handoff的语义。

这个提案的目的也不是实现完全无锁的队列，只是在一些关键路径上通过无锁提升 Channel 的性能。

数据结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
type hchan struct {
	qcount   uint           // Channel中的元素个数
	dataqsiz uint           // Channel中的循环队列长度
	buf      unsafe.Pointer // Channel的缓冲区数据指针
	elemsize uint16
	closed   uint32
	timer    *timer // Channel的定时器
	elemtype *_type 
	sendx    uint   // Channel 的已发送元素在缓冲区循环数组中的索引
	recvx    uint   // Channel 的已接收元素在缓冲区循环数组中的索引
	recvq    waitq  // 等待接收的 Goroutine 列表
	sendq    waitq  // 等待发送的 Goroutine 列表
	bubble   *synctestBubble

  /*
  	lock 保护 hchan 中的所有字段，以及在该通道上被阻塞的多个 sudog 字段。
  	在持有此锁的情况下，切勿更改其他 G 的状态（特别是不要将一个 G 设置为就绪状态），因为这可能导致与堆栈收缩相关的死锁
  */
	lock mutex		// 用于保证每个读/写Channel的操作都是原子的
}

sendq 和 recvq 存储了当前 Channel 由于缓冲区空间不足而阻塞的 Goroutine 列表，这些等待队列使用双向链表 runtime.waitq 表示，链表中所有的元素都是 runtime.sudog 结构（表示一个在等待列表中的 Goroutine）：

1
2
3
4
type waitq struct {
	first *sudog
	last  *sudog
}

创建Channel

Channel 是一个引用类型，所以在被初始化之前，它的值为nil；所有 Channel 的创建都会使用 make 关键字进行初始化。例如：make(chan int, 10)。传递的int值代表Channel缓冲区的容量，构造的是一个缓冲Channel；不传int值或者传0的，构造的是一个非缓冲Channel。

1
2
3
4
// 无缓冲通道
ch1 := make(chan int)
// 有缓冲通道
ch2 := make(chan int, 10)

经过一系列编译步骤（为啥不写，因为我半天没弄懂），用于运行时创建chan的函数是makechan：

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
func makechan(t *chantype, size int) *hchan {
	elem := t.Elem
  mem, overflow := math.MulUintptr(elem.Size_, uintptr(size))

	// 省略一些元素大小、内存对齐、缓冲区大小的检查

	// Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.
	// buf points into the same allocation, elemtype is persistent.
	// SudoG's are referenced from their owning thread so they can't be collected.
	// TODO(dvyukov,rlh): Rethink when collector can move allocated objects.
	var c *hchan
	switch {
	case mem == 0:
		// 当前Channel不存在缓冲区：只会为 runtime.hchan 分配一段内存空间
		c = (*hchan)(mallocgc(hchanSize, nil, true))
		c.buf = c.raceaddr()
	case !elem.Pointers():
    // 如果当前 Channel 中存储的类型不是指针类型：会为当前的 Channel 和底层的数组分配一块连续的内存空间；
		c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
		c.buf = add(unsafe.Pointer(c), hchanSize)
	default:
    // 默认情况：单独为 runtime.hchan 和缓冲区分配内存
		c = new(hchan)
		c.buf = mallocgc(mem, elem, true)
	}
	// 更新字段
	c.elemsize = uint16(elem.Size_)
	c.elemtype = elem
	c.dataqsiz = uint(size)
	......
	return c
}

发送数据

当需要向Channel发送数据时，使用ch <- i语句，实际上调用runtime.chansend。如果在调用时将 block 参数设置成 true，那么表示当前发送操作是阻塞的：

1
2
3
4
5
6
7
8
9
10
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  // 在发送数据前为当前Channel加锁，防止多个线程并发修改数据
	lock(&c.lock)
	// 检查Channel是否已关闭
	if c.closed != 0 {
		unlock(&c.lock)
		panic(plainError("send on closed channel"))
	}
  ......
}

分为3个部分：（针对接收方状态）

  1. 当接收队列存在等待的Goroutine 时：通过 runtime.send 直接将数据发送给阻塞的接收者；
  2. 第1条不满足，且缓冲区存在剩余空间时：将发送的数据写入 Channel 的缓冲区；
  3. 第1，2条不满足，且不存在缓冲区或者缓冲区已满时：等待其他 Goroutine 从 Channel 接收数据。

直接发送

如果目标Channel没有被关闭，且已经有处于读等待的 Goroutine：从接收队列 recvq 中取出最先陷入等待的 Goroutine ，并直接向它发送数据：

1
2
3
4
5
6
7
8
9
10
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  ......
  if sg := c.recvq.dequeue(); sg != nil {
		// Found a waiting receiver. We pass the value we want to send
		// directly to the receiver, bypassing the channel buffer (if any).
		send(c, sg, ep, func() { unlock(&c.lock) }, 3)
		return true
	}
  ......
}

发送数据时调用 runtime.send，分为2个部分：

  1. 调用 runtime.sendDirect 将发送的数据直接拷贝到 x = <-c 表达式中变量 x 所在的内存地址上；
  2. 调用 runtime.goready 将等待接收数据的 Goroutine 标记成可运行状态 Grunnable， 并把该 Goroutine 放到发送方所在的处理器的 runnext 上等待执行，该处理器在下一次调度时会立刻唤醒数据的接收方；
    • 注：发送数据的过程只是将接收方的 Goroutine 放到了处理器的 runnext 中，程序没有立刻执行该 Goroutine。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
  /* sg 是接收方的协程；ep 是发送方传递的数据指针；*/
	if sg.elem != nil {
		sendDirect(c.elemtype, sg, ep)
		sg.elem = nil
	}
	gp := sg.g		// 获取与 sg（接收方协程）关联的 Goroutine
	unlockf()			// 解锁通道
	gp.param = unsafe.Pointer(sg)		// 将接收方的 sg 传递给该 Goroutine
	sg.success = true								// 标记发送操作成功
	if sg.releasetime != 0 {
		sg.releasetime = cputicks()
	}
	goready(gp, skip+1)							// 将接收方协程 gp 设置为就绪状态，准备继续执行
}

缓冲区

如果创建的 Channel 包含缓冲区，且 Channel 中的数据没有装满：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
	......
	if c.qcount < c.dataqsiz {
		// 计算缓冲区中下一个可以存储数据的地址 qp
		qp := chanbuf(c, c.sendx)
    // 将待发送的数据 ep 拷贝到缓冲区地址 qp；并增加 sendx 索引和 qcount 计数器
		typedmemmove(c.elemtype, qp, ep)
		c.sendx++
		if c.sendx == c.dataqsiz {
			c.sendx = 0	// buf 是一个循环数组；所以当 sendx 等于 dataqsiz 时：会重新回到数组开始的位置
		}
		c.qcount++
		unlock(&c.lock)
		return true
	}
	......
}

下图展示相应流程：当前sendx=6，recvx=4：将待发送的数据拷贝到sendx=6的位置；移动sendx到下一个位置。

阻塞发送

当 Channel 没有接收者能够处理数据时，向 Channel 发送数据会被下游阻塞。

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
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
	...... 
	if !block {
		unlock(&c.lock)
		return false
	}

	// 1. 获取发送数据使用的 Goroutine
	gp := getg()
  // 2. 获取 runtime.sudog 结构并设置这一次阻塞发送的相关信息：例如发送的 Channel、是否在 select 中和待发送数据的内存地址等
	mysg := acquireSudog()
	// 3. 将刚刚创建并初始化的 runtime.sudog 加入发送等待队列 sendq，并设置到当前 Goroutine 的 waiting 上，表示 Goroutine 正在等待该 sudog 准备就绪
	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)
	......
  // 使得当前的发送者 Goroutine 陷入沉睡，等待唤醒，让出处理器的使用权
	gopark(chanparkcommit, unsafe.Pointer(&c.lock), reason, traceBlockChanSend, 2)
	KeepAlive(ep)
	if mysg != gp.waiting {
		throw("G waiting list is corrupted")
	}
	gp.waiting = nil
	gp.activeStackChans = false
	closed := !mysg.success
	gp.param = nil
	......
	releaseSudog(mysg)
	......
	return true
}

小结一下：向Channel发送数据的3种情况：

  1. 如果当前 Channel 的 recvq 上存在已经被阻塞的 Goroutine，那么会直接将数据发送给当前 Goroutine 并将其设置成下一个运行的 Goroutine；
  2. 如果 Channel 存在缓冲区并且其中还有空闲的容量，我们会直接将数据存储到缓冲区 sendx 所在的位置上；
  3. 如果不满足上面的两种情况，会创建一个 runtime.sudog 结构并将其加入 Channel 的 sendq 队列中，当前 Goroutine 也会陷入阻塞等待其他的协程从 Channel 接收数据；

触发Goroutine调度的时机：

  1. 发送数据时发现 Channel 上存在等待接收数据的 Goroutine，立刻设置处理器的 runnext 属性，但是并不会立刻触发调度；
  2. 发送数据时并没有找到接收方并且缓冲区已经满了，这时会将自己加入 Channel 的 sendq 队列并调用 runtime.goparkunlock 触发 Goroutine 的调度让出处理器的使用权；

接收数据

Go 语言中可以使用两种不同的方式去接收 Channel 中的数据：

1
2
i <- ch
i, ok <- ch

最终都会调用runtime.chanrecv。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
  // 1. 如果从一个空 Channel 接收数据，会直接调用 runtime.gopark 让出处理器的使用权
	if c == nil {
		if !block {
			return
		}
		gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
		throw("unreachable")
	}

	lock(&c.lock)
	// 2. 如果当前 Channel 已被关闭，且缓冲区不存在任何数据：清除ep指针中的数据，返回（true, false）
	if c.closed != 0 && c.qcount == 0 {
		unlock(&c.lock)
		if ep != nil {
			typedmemclr(c.elemtype, ep)
		}
		return true, false
	}

以上两种为特殊情况，除此之外，还有3种情形：（针对发送方状态）

  1. 当发送队列sendq存在等待的Goroutine 时：通过 runtime.recv 直接从阻塞的发送者或者缓冲区中获取数据；
  2. 第1条不满足，且缓冲区中存在数据时：从 Channel 的缓冲区中接收数据；
  3. 第1，2条不满足，且缓冲区中不存在数据时：等待其他 Goroutine 向 Channel 发送数据。

直接接收

当Channel的sendq队列中包含等待的Goroutine 时：取出队头等待的Goroutine（和直接发送的处理逻辑较像）。

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

调用 runtime.recv ，分为两个部分：

  1. 将发送者的数据sg放入Channel，发送者被唤醒以继续执行；
  2. 被接收者（当前Goroutine）接收到的数据，被写入ep。
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
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
	......
  // 1.1 无缓冲的Channel（同步通道）：
	if c.dataqsiz == 0 {
		if raceenabled {
			racesync(c, sg)
		}
		if ep != nil {
      // 将 Channel 发送队列中 Goroutine 存储的 elem 数据拷贝到目标内存地址中
			recvDirect(c.elemtype, sg, ep)
		}
	} else {	// 1.2 有缓冲的Channel（异步通道）：
    // 从接收队列头部，取出一个元素 qp
		qp := chanbuf(c, c.recvx)
		// 接收端：qp 拷贝到接收方的内存地址 ep
		if ep != nil {
			typedmemmove(c.elemtype, ep, qp)
		}
		// 发送端：将发送者的数据从 sg.elem 拷贝到队列中当前位置 qp
		typedmemmove(c.elemtype, qp, sg.elem)
    // 更新接收指针
		c.recvx++
		if c.recvx == c.dataqsiz {
			c.recvx = 0
		}
		c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
	}
  // 2. 完成接收操作：
	sg.elem = nil
	gp := sg.g
	unlockf()		// 解锁通道
	gp.param = unsafe.Pointer(sg)
	sg.success = true
	if sg.releasetime != 0 {
		sg.releasetime = cputicks()
	}
	goready(gp, skip+1)
}

下图展示相应流程。缓冲区已满，发送队列中存在等待的Gouroutine时：使用发送队列头 runtime.sudog 中的元素替换接收索引 recvx 所在位置的元素；recvx原有元素被拷贝到接收数据的变量对应的内存空间上。

缓冲区

当Channel的缓冲区中已经包含数据时：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
  if c.qcount > 0 {
    // 获取recvx位置
    qp := chanbuf(c, c.recvx)
    if ep != nil {
    // 将缓冲区地址 qp 中的数据拷贝到接收端内存地址 ep；并增加 recv 索引，减少 qcount 计数器
      typedmemmove(c.elemtype, ep, qp)
    }
    typedmemclr(c.elemtype, qp)
    c.recvx++
    if c.recvx == c.dataqsiz {
      c.recvx = 0
    }
    c.qcount--
    unlock(&c.lock)		// 释放Channel持有的锁
    return true, true
  }
}

下图展示相应流程：当前sendx=4，recvx=6：将待接收的数据从recvx=6的位置拷贝到内存地址；移动recvx到下一个位置。

阻塞接收

当 Channel 没有发送者能够处理数据时，从 Channel 接收数据会被阻塞。（然而不是所有的接收操作都是阻塞的，与 select 语句结合使用时就可能会使用到非阻塞的接收操作）

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
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
  // selected：是否选中；received：是否成功接收到数据
	...
  // 1. 非阻塞模式：直接解锁Channel的锁
	if !block {
		unlock(&c.lock)
		return false, false
	}
	// 2. 阻塞模式：需要阻塞以等待数据
	gp := getg()		// 获取当前goroutine
	mysg := acquireSudog()	// 创建一个新的sudog（代表一个等待的goroutine）
	mysg.elem = ep
	gp.waiting = mysg
	mysg.g = gp
	mysg.c = c
  // 2.1 将 mysg（当前正在等待接收的 goroutine）加入到通道 c 的接收队列中
	c.recvq.enqueue(mysg)
  // 2.2 让出处理器的使用权并等待调度器的调度
	goparkunlock(&c.lock, waitReasonChanReceive, traceEvGoBlockRecv, 3)
  // 2.3 恢复并释放资源
	gp.waiting = nil
	closed := gp.param == nil
	gp.param = nil
	releaseSudog(mysg)
	return true, !closed
}

小结一下：从Channel中接收数据的5种情况：

  1. 如果 Channel 为空，那么会直接调用 runtime.gopark 挂起当前 Goroutine；
  2. 如果 Channel 已经关闭并且缓冲区没有任何数据，runtime.chanrecv 会直接返回；
  3. 如果 Channel 的 sendq 队列中存在挂起的 Goroutine，会将 recvx 索引所在的数据拷贝到接收变量所在的内存空间上并将 sendq 队列中 Goroutine 的数据拷贝到缓冲区；
  4. 如果 Channel 的缓冲区中包含数据，那么直接读取 recvx 索引对应的数据；
  5. 在默认情况下会挂起当前的 Goroutine，将 runtime.sudog 结构加入 recvq 队列并陷入休眠等待调度器的唤醒；

触发Goroutine调度的两个时机：

  1. 当 Channel 为空时；
  2. 当缓冲区中不存在数据并且也不存在数据的发送者时。

关闭Channel

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
func closechan(c *hchan) {
  // 1. 处理Channel是空指针/已被关闭的情况：直接崩溃，抛出异常
	if c == nil {
		panic(plainError("close of nil channel"))
	}
	lock(&c.lock)
	if c.closed != 0 {
		unlock(&c.lock)
		panic(plainError("close of closed channel"))
	}
	c.closed = 1

  // 2. 将 recvq 和 sendq 两个队列中的数据，加入到Goroutine列表gList中，同时清除所有 runtime.sudog 上未被处理的元素：
	var glist gList
	// release all readers
	for {
		sg := c.recvq.dequeue()
		if sg == nil {
			break
		}
		if sg.elem != nil {
			typedmemclr(c.elemtype, sg.elem)
			sg.elem = nil
		}
		if sg.releasetime != 0 {
			sg.releasetime = cputicks()
		}
		gp := sg.g
		gp.param = unsafe.Pointer(sg)
		sg.success = false
		glist.push(gp)
	}

	// release all writers (they will panic)
	for {
		sg := c.sendq.dequeue()
		if sg == nil {
			break
		}
		sg.elem = nil
		if sg.releasetime != 0 {
			sg.releasetime = cputicks()
		}
		gp := sg.g
		gp.param = unsafe.Pointer(sg)
		sg.success = false
		glist.push(gp)
	}
	unlock(&c.lock)

	// 为所有被阻塞的 Goroutine 调用 runtime.goready 触发调度
	for !glist.empty() {
		gp := glist.pop()
		gp.schedlink = 0
		goready(gp, 3)
	}
}

参考

为什么使用通信来共享内存

Go 语言设计与实现