go语言select

使用

func main() {
	ch1 := make(chan int, 1)
	ch2 := make(chan int, 1)

	go func() {
		time.Sleep(1 * time.Second)
		ch1 <- 1
	}()

	go func() {
		time.Sleep(2 * time.Second)
		ch2 <- 1
	}()

	select {
	case <-ch1:
		fmt.Println("ch 1 out")
	case <-ch2:
		fmt.Println("ch 2 out")
	default:
		fmt.Println("default")
	}

}

C 语言的 select 系统调用可以同时监听多个文件描述符的可读或者可写的状态，Go 语言中的 select 也能够让 Goroutine 同时等待多个 Channel 可读或者可写，在多个文件或者 Channel状态改变之前，select 会一直阻塞当前线程或者 Goroutine。

当我们在 Go 语言中使用 select 控制结构时，会遇到两个有趣的现象：

1. select 能在 Channel 上进行非阻塞的收发操作；
2. select 在遇到多个 Channel 同时响应时，随机的引入就是为了避免饥饿问题的发生；

数据结构

select 在 Go 语言的源代码中不存在对应的结构体，但是我们使用 runtime.scase 结构体表示 select 控制结构中的 case：

type scase struct {
	c    *hchan         // chan
	elem unsafe.Pointer // data element
}

因为非默认的 case 中都与 Channel 的发送和接收有关，所以 runtime.scase 结构体中也包含一个 runtime.hchan 类型的字段存储 case 中使用的 Channel。

实现原理

在默认的情况下，编译器会使用如下的流程处理select语句：

1. 将所有的case转换成包含channel以及类型等信息的runtime.scase结构体
2. 调用运行时函数runtime.selectgo从多个准备就绪的channel中选择一个可执行的runtime.scase结构体
3. 通过for循环生成一组if语句，在语句中判断自己是不是被选中的case

selv := [3]scase{}
order := [6]uint16
for i, cas := range cases {
    c := scase{}
    c.kind = ...
    c.elem = ...
    c.c = ...
}
chosen, revcOK := selectgo(selv, order, 3)
if chosen == 0 {
    ...
    break
}
if chosen == 1 {
    ...
    break
}
if chosen == 2 {
    ...
    break
}

展开后的代码片段中最重要的就是用于选择待执行case的运行时函数runtime.selectgo，这也是我们要关注的重点。因为这个函数的实现比较复杂，所以这里分两部分分析它的执行过程：

1. 执行一些必要的初始化操作并确定case的处理顺序
2. 在循环中根据case的类型做出不同的处理

初始化

runtime.selectgo函数首先会进行执行必要的初始化操作并决定处理case的两个顺序-轮询顺序pollOrder和加锁顺序lockOrder

func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool) {
	cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0))
	order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0))
	
	ncases := nsends + nrecvs
	scases := cas1[:ncases:ncases]
	pollorder := order1[:ncases:ncases]
	lockorder := order1[ncases:][:ncases:ncases]

	norder := 0
	for i := range scases {
		cas := &scases[i]
	}

	for i := 1; i < ncases; i++ {
		j := fastrandn(uint32(i + 1))
		pollorder[norder] = pollorder[j]
		pollorder[j] = uint16(i)
		norder++
	}
	pollorder = pollorder[:norder]
	lockorder = lockorder[:norder]

	// 根据 Channel 的地址排序确定加锁顺序
	...
	sellock(scases, lockorder)
	...
}

轮询顺序pollOrder和加锁顺序lockOrder分别是通过以下方式确认的:

• 轮询顺序：通过runtime.fastrandn函数引入随机性
• 加锁顺序：按照channel的地址排序后确定加锁顺序

随机的轮询顺序可以避免channel的饥饿问题，保证公平性；而根据channel的地址顺序确定加锁顺序能够避免死锁的发生。这段代码最后调用的runtime.sellock会按照之前生成的加锁顺序锁定select语句中包含所有的channel

循环

当我们为select语句锁定了所有channel之后就会进入runtime.selectgo函数的主循环，它会分三个阶段查找或者等待某个channel准备就绪：

1. 查找是否已经存在准备就绪的channel，即可以执行收发操作
2. 将当前goroutine加入channel对应的收发队列上并等待其他goroutine的唤醒
3. 当前goroutine被唤醒之后找到满足条件的channel并进行处理

runtime.selectgo函数会根据不同情况通过goto语句跳转到函数内部不同标签执行相应的逻辑，包括：

• bufrecv：可以从缓冲区读取数据
• bufsend：可以向缓冲区写入数据
• recv：可以从休眠的发送方获取数据
• send：可以向休眠的接收方发送数据
• rclose：可以从关闭的channel读取EOF
• sclose：向关闭的channel发送数据
• retc：结束调用并返回

我们先来分析循环执行的第一个阶段，查找已经准备就绪的 Channel。循环会遍历所有的 case 并找到需要被唤起的 runtime.sudog 结构，在这个阶段，我们会根据 case 的四种类型分别处理：

1. 当 case 不包含 Channel 时；

• 这种 case 会被跳过；

2. 当 case 会从 Channel 中接收数据时；
   • 如果当前 Channel 的 sendq 上有等待的 Goroutine，就会跳到 recv 标签并从缓冲区读取数据后将等待 Goroutine 中的数据放入到缓冲区中相同的位置；
   • 如果当前 Channel 的缓冲区不为空，就会跳到 bufrecv 标签处从缓冲区获取数据；
   • 如果当前 Channel 已经被关闭，就会跳到 rclose 做一些清除的收尾工作；
3. 当 case 会向 Channel 发送数据时；
   • 如果当前 Channel 已经被关，闭就会直接跳到 sclose 标签，触发 panic 尝试中止程序；
   • 如果当前 Channel 的 recvq 上有等待的 Goroutine，就会跳到 send 标签向 Channel 发送数据；
   • 如果当前 Channel 的缓冲区存在空闲位置，就会将待发送的数据存入缓冲区；
4. 当 select 语句中包含 default 时；
   • 表示前面的所有 case 都没有被执行，这里会解锁所有 Channel 并返回，意味着当前 select 结构中的收发都是非阻塞的；

第一阶段的主要职责是查找所有 case 中是否有可以立刻被处理的 Channel。无论是在等待的 Goroutine 上还是缓冲区中，只要存在数据满足条件就会立刻处理，如果不能立刻找到活跃的 Channel 就会进入循环的下一阶段，按照需要将当前 Goroutine 加入到 Channel 的 sendq 或者 recvq 队列中：

func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool) {
	...
	gp = getg()
	nextp = &gp.waiting
	for _, casei := range lockorder {
		casi = int(casei)
		cas = &scases[casi]
		c = cas.c
		sg := acquireSudog()
		sg.g = gp
		sg.c = c

		if casi < nsends {
			c.sendq.enqueue(sg)
		} else {
			c.recvq.enqueue(sg)
		}
	}

	gopark(selparkcommit, nil, waitReasonSelect, traceEvGoBlockSelect, 1)
	...
}

除了将当前 Goroutine 对应的 runtime.sudog 结构体加入队列之外，这些结构体都会被串成链表附着在 Goroutine 上。在入队之后会调用 runtime.gopark 挂起当前 Goroutine 等待调度器的唤醒。

等到 select 中的一些 Channel 准备就绪之后，当前 Goroutine 就会被调度器唤醒。这时会继续执行 runtime.selectgo 函数的第三部分，从 runtime.sudog 中读取数据：

func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool) {
	...
	sg = (*sudog)(gp.param)
	gp.param = nil

	casi = -1
	cas = nil
	sglist = gp.waiting
	for _, casei := range lockorder {
		k = &scases[casei]
		if sg == sglist {
			casi = int(casei)
			cas = k
		} else {
			c = k.c
			if int(casei) < nsends {
				c.sendq.dequeueSudoG(sglist)
			} else {
				c.recvq.dequeueSudoG(sglist)
			}
		}
		sgnext = sglist.waitlink
		sglist.waitlink = nil
		releaseSudog(sglist)
		sglist = sgnext
	}

	c = cas.c
	goto retc
	...
}

第三次遍历全部 case 时，我们会先获取当前 Goroutine 接收到的参数 sudog 结构，我们会依次对比所有 case 对应的 sudog 结构找到被唤醒的 case，获取该 case 对应的索引并返回。

由于当前的 select 结构找到了一个 case 执行，那么剩下 case 中没有被用到的 sudog 就会被忽略并且释放掉。为了不影响 Channel 的正常使用，我们还是需要将这些废弃的 sudog 从 Channel 中出队。

selectgo 会根据变量 cas 的值来判断是收发操作唤醒还是关闭操作唤醒关闭操作唤醒的话 gp.param 会被置为 nil，那么就不会赋值 cas 变量

关闭操作唤醒 selectgo，并不一定会选择该 case对于关闭操作唤醒，逻辑会回到 loop 中再次执行 scases 的检查操作

关闭操作唤醒 selectgo 后，在完成所有 channel 加锁前又有 channel 准备好收发操作了，那么在 loop 查询时，按照 pollorder 随机顺序，可能会选中刚刚准备好的 channel，而不是唤醒 selectgo 的 case

收发操作唤醒 selectgo, 必定会选择该 case，而对于收发操作，已经完成了值的拷贝，必然会选择这个 case，而不会再次去查询