Golang的Channel源码阅读、工作流程分析。

Channel整体结构

源码位置

位于src/runtime下的chan.go中。

在这里插入图片描述

Channel整体结构图

图源:https://i6448038.github.io/2019/04/11/go-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
	closed   uint32
	elemtype *_type // element type
	sendx    uint   // send index
	recvx    uint   // receive index
	recvq    waitq  // list of recv waiters
	sendq    waitq  // list of send waiters

	// lock protects all fields in hchan, as well as several
	// fields in sudogs blocked on this channel.
	//
	// Do not change another G's status while holding this lock
	// (in particular, do not ready a G), as this can deadlock
	// with stack shrinking.
	lock mutex
}

我们可以看到,其中有一个buf空间,这个对应的是我们生成的有缓冲通道、无缓冲通道。recvqsendq对应的是waitq类型,其中主要存储的是发送、接受方的Goroutine。

waitq&&sudog

waitq


type waitq struct {
	first *sudog
	last  *sudog
}

sudog

// sudogs are allocated from a special pool. Use acquireSudog and
// releaseSudog to allocate and free them.
type sudog struct {
	// The following fields are protected by the hchan.lock of the
	// channel this sudog is blocking on. shrinkstack depends on
	// this for sudogs involved in channel ops.

	g *g

	next *sudog
	prev *sudog
	elem unsafe.Pointer // data element (may point to stack)

	// The following fields are never accessed concurrently.
	// For channels, waitlink is only accessed by g.
	// For semaphores, all fields (including the ones above)
	// are only accessed when holding a semaRoot lock.

	acquiretime int64
	releasetime int64
	ticket      uint32

	// isSelect indicates g is participating in a select, so
	// g.selectDone must be CAS'd to win the wake-up race.
	isSelect bool

	// success indicates whether communication over channel c
	// succeeded. It is true if the goroutine was awoken because a
	// value was delivered over channel c, and false if awoken
	// because c was closed.
	success bool

	parent   *sudog // semaRoot binary tree
	waitlink *sudog // g.waiting list or semaRoot
	waittail *sudog // semaRoot
	c        *hchan // channel
}

Channel工作流程

创建管道

先在创建阶段:会根据缓冲大小对buf进行初始化,无缓冲通道的buf为0。具体见

发送数据

发送数据前:
首先会进行加锁(因此-“一个通道同时只能进行一个收/发操作”)。如果Channel已关闭,则会报panic。

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

	if c.closed != 0 {
		unlock(&c.lock)
		panic(plainError("send on closed channel"))
	}

发送数据时,会分为多种情况:
1、有等待的接收者——直接发给阻塞的接收者。
2、无等待 但是缓冲区有空间——写入Channel的缓冲区。
3、无等待 无空间——等待其他Goroutine接受数据。

update:刚才脑子转不过来了一下。。疑惑:情况1的时候,那缓存内的东西先进去,不应该排队放后面吗?为什么直接丢给goroutine了。一下反应过来:如果缓存有内容,那接收者就直接拿了啊!!阻塞,就说明他已经缓存拿不到东西,才会去阻塞等待的。

工作流程:(由于GPT4.0解读源码总结完成)

1、检查通道是否为nil:如果尝试向一个nil的通道发送数据,如果是非阻塞的(block为false),则直接返回false;如果是阻塞的,则该goroutine会被挂起,直到被唤醒(实际上,向nil通道发送数据会导致永久阻塞,这里的唤醒仅是理论上的,因为后面紧接着会调用throw("unreachable")抛出异常,表示这个代码路径不应该被执行)。

2、快速路径检查:在尝试获取锁之前,先检查通道是否已关闭并且是否已满,以避免在这些明显无法发送成功的情况下还获取锁,提高效率。

3、获取锁:为了保证对通道状态的修改是安全的,需要先获取通道的锁。

4、检查通道是否已关闭:如果通道已经关闭,则抛出“send on closed channel”的异常。

5、尝试直接发送给等待接收的goroutine:如果有goroutine正在等待接收(即接收队列不为空),则直接将值传递给它,并唤醒该goroutine。

6、检查通道缓冲区是否有空间:如果通道的缓冲区还有空间,则将值放入缓冲区,并更新相关指标。

7、非阻塞发送失败:如果是非阻塞发送且到达这一步,说明无法立即发送,释放锁并返回false8、准备阻塞发送:如果是阻塞发送,则创建一个sudog对象表示当前goroutine,将其加入到发送队列中,并挂起当前goroutine等待被唤醒(通常是接收方接收到值或通道被关闭时唤醒)。

9、唤醒后的处理:被唤醒后,检查发送是否成功(通过检查sudog的success字段)。如果通道在等待期间被关闭,则抛出“send on closed channel”的异常。

10、资源清理和返回:最后,释放sudog资源,返回发送是否成功。

详细源码工作流程,见此

接收数据

当已被关闭&&缓冲区没有数据,会返回。

接收的三种情况:
1、存在发送者时,直接从发送者或缓冲区数据。
2、缓冲区存在数据,从缓冲区接收。
3、都不存在时,等待其他Goroutine发送。

源码阅读(chanrecv函数):

1、检查通道是否为空:如果尝试从一个nil的通道接收数据,根据block参数的不同,可能会导致goroutine挂起或者直接返回。

2、快速路径检查:在不阻塞的情况下,如果通道为空,则尝试检查通道是否关闭。如果通道已关闭且为空,则清空指针ep指向的内存(如果ep不为nil)并返回。

3、加锁:为了修改通道状态,需要先获取通道的锁。

4、通道已关闭且无数据:如果通道已关闭并且没有数据,清空ep指向的内存并返回。

5、从等待发送的goroutine接收数据:如果通道未关闭且有等待发送的goroutine,直接从发送方接收数据。

6、从通道缓冲区接收数据:如果通道有数据(qcount > 0),则从通道的缓冲区接收数据到ep指向的位置,并清空缓冲区中该数据的位置。

7、非阻塞情况下无数据可接:如果是非阻塞接收且到达这一步,说明无法立即接收数据,释放锁并返回。

8、准备阻塞接收:如果是阻塞接收,则挂起当前goroutine,直到有数据可接收或通道被关闭。

9、唤醒后的处理:被唤醒后,检查接收是否成功。如果接收成功,则ep指向的位置已被填充。

10、资源清理和返回:最后,释放相关资源,返回操作结果。

关闭管道

closechan函数:

1、检查通道是否为nil:如果尝试关闭一个nil的通道,会引发panic2、加锁:为了保证对通道状态的修改是并发安全的,需要先获取通道的锁。

3、检查通道是否已经关闭:如果通道已经被关闭(c.closed != 0),则释放锁并panic。这防止了通道被多次关闭导致的未定义行为。

4、设置通道为关闭状态:将通道的closed标志设置为1,表示该通道已经关闭。

5、处理等待接收的goroutine:遍历接收队列recvq,对于队列中的每个等待接收的goroutine(通过sudog表示),清空它们等待接收的元素指针(如果有),并将它们标记为操作未成功(success = false)。这些goroutine将会被唤醒,但是接收操作会因为通道已关闭而失败。

6、处理等待发送的goroutine:遍历发送队列sendq,对于队列中的每个等待发送的goroutine,清空它们准备发送的元素指针(如果有),并将它们标记为操作未成功。这些goroutine在被唤醒后会感知到通道已经关闭,并可能引发panic7、释放锁:完成上述操作后,释放通道的锁。

8、唤醒所有goroutine:最后,对于通过上述步骤收集到的所有goroutine(存储在glist中),将它们标记为就绪状态(goready),这样它们就可以被调度执行了。这确保了所有因为该通道操作而阻塞的goroutine都能继续执行,无论是因为等待发送还是接收。

本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
THE END
分享
二维码
< <上一篇
下一篇>>