golang runtime源码阅读 channal实现

Posted 2023-02-21 ythunder

概念及使用场景

通道(channal)是Golang实现CSP并发模型的关键，分为 有缓冲通道 和无缓冲通道。

• 有缓冲管道： channal持有一个固定大小的队列，队列满时发送者将阻塞(反之亦然)。多用于数据共享。
• 无缓冲管道： 发送和接收数据同时完成，如果没有goroutine读取channal，则发送者阻塞(反之亦然)。多用于协程同步。
• 有缓冲+select： +for实现对多个chan的监听操作 +time实现超时控制

基本使用

简单的 无缓冲channal 使用

func main() 
 ch := make(chan int)  //创建无缓冲管道
 
 go func()           //发起goroutine,向管道写入数据
 	ch <- 1
 	close(ch)                    //关闭管道
 () 

 t := <- ch                     //主goroutine等待读取到管道数据后退出
 fmt.Println(t)
 return

需要注意的是：
1. 无缓冲在写入时，必须有其他线程在对端接受，否则线程将阻塞。
2. 向关闭的chan写入数据将Panic
3. 可以从关闭的chan中读取数据
4. 重复关闭chann会导致panic

基本结构及内存布局

chan结构体定义在 ./src/runtime/chan.go文件，type hchan struct

type hchan struct 
	qcount   uint           // 队列数据长度len
	dataqsiz uint           // 缓冲区长度cap c:=make(chan int, 10)即此值为10
	buf      unsafe.Pointer // 指向dataqsiz数组的指针
	elemsize uint16   		//元素类型大小 sizeof(struct)
	closed   uint32  		//关闭标志
	elemtype *_type 		// chan接收的元素类型
	sendx    uint   		// 写入chan索引,写入1个数据时,sendx+1
	recvx    uint   		// 读取chan索引,被读走1个数据时,recvx+1
	recvq    waitq  		// 阻塞在chann上的读等待队列
	sendq    waitq  		// 阻塞在chann上的写等待队列
	lock 	 mutex  		//互斥锁 保证hchan的数据读写安全，包括被阻塞的waitq中的sudog

waitq类型链表，sudog 为封装的goroutine，包含等待 读/写的被阻塞的goroutine信息。

type waitq struct 
	first *sudog
	last  *sudog


type sudog struct 
	g *g  					// 阻塞的goroutine
	isSelect bool  			// select中使用chan是特殊处理的，本章暂时不论
	next     *sudog		
	prev     *sudog
	elem     unsafe.Pointer // 数据指针(指向堆内存或栈空间)
	acquiretime int64
	releasetime int64
	ticket      uint32
	parent      *sudog // semaRoot binary tree
	waitlink    *sudog // g.waiting list or semaRoot
	waittail    *sudog // semaRoot
	c           *hchan // channel

结构中大约了解下 channal 的实现

1. channal 有缓存时通过读写索引读取数据，并保存了可容纳元素数量(dataqsize) 及当前元素量(qcount)。recvq和sendq为两个队列，遵循先进先出原则，队列元素sudog为封装后的goroutine，包含goroutine等待的chan信息和数据地址，一般为阻塞状态。互斥锁保证hchan中的数据读写安全。

2. 对于无缓冲channal，recvq 和 sendq至少有一个为空，且 dataqsiz 和 qcount 都为0。

初始化 channal 过程如下

// use example:
ch := make(chan int, 10)
// code
func makechan(t *chantype, size int) *hchan 
	elem := t.elem
	// 检查是否存在size越界问题	
	if size < 0 || uintptr(size) > maxSliceCap(elem.size) || uintptr(size)*elem.size > maxAlloc-hchanSize 
		panic(plainError("makechan: size out of range"))
	
	var c *hchan
	switch 
	// make(chan int, 0) size=0 but elem.size=8
	// make(struct, 2) size=2 but elem.size=0
	case size == 0 || elem.size == 0:  
		//分配hchan大小，无缓冲chan只分配这些，以下是有缓冲chan
		c = (*hchan)(mallocgc(hchanSize, nil, true))
		// race检测，不懂
		c.buf = c.raceaddr()
	// channal类型不包含指针 
	case elem.kind&kindNoPointers != 0:
		// 分配空间大小为(hchan结构大小 单个元素大小*请求分配数量)的堆内存
		c = (*hchan)(mallocgc(hchanSize+uintptr(size)*elem.size, nil, true))
		// c.buf指针指向hchan结构后，即存储元素区
		c.buf = add(unsafe.Pointer(c), hchanSize)
	// channal类型包含指针
	default:
		// 分配一块hchan内存
		// c.buf 指向 分配固定大小的元素存储空间 的堆内存
 		c = new(hchan)
		c.buf = mallocgc(uintptr(size)*elem.size, elem, true)
	

	c.elemsize = uint16(elem.size)
	c.elemtype = elem
	c.dataqsiz = uint(size)

	return c

如上，make channal时内存分配如下图

如上，
无缓冲队列中仅有hchan，即没有空间可用于存储元素，所以如果发生写数据请求，只有 1)阻塞在sendq队列中 2) 对端有阻塞recvq，直接copy到对方goroutine空间
有缓冲chan之所以要分开指针类型缓冲区主要是为了区分gc操作，需要将它设置为flagNoScan。并且指针大小固定，可以跟hchan头部一起分配内存。

然后看下sendq和recvq操作具体如何发生

chan如何工作

写 channal 操作

1. 写入未初始化的chan，程序将永久阻塞。此时go run 报错：<font color=orange> fatal error: all goroutines are asleep - deadlock! </font>，goroutine被强制结束
2. 写入已关闭chan，程序将发生panic。go run报错: <font color=orange> panic: send on closed channel </font>
3. 如果是无缓冲chan，对端有等待goroutine时，复制数据给goroutine
4. 写入有缓冲chan，并且有空间可写时，将消息复制给等待的goroutine
5. 缓冲队列已满, 新建(sudog)G,并加入到sendq阻塞队列

// 编译代码中 c <- x 入口函数
//go:nosplit
func chansend1(c *hchan, elem unsafe.Pointer) 
	chansend(c, elem, true, getcallerpc())

// c为chan, ep为需要写入的变量的地址, block=TRUE, callerpc
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool 
	//CASE 1: chan未初始化时为Nil，send数据将永久阻塞
	if c == nil 
		if !block 
			return false
		
		//gopark函数会使goroutine休眠, 通过unlockf唤醒，但此处unlockf为nil,所以将一直休眠
		gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
		throw("unreachable")
	
	
	// 未被加锁 未关闭 且 1.(队列长度cap为0(无缓冲channal) 且 外部写chan协程队列为空） 或 2.(cap>0 且 cap=len已满)返回false. 但是chansend1调用时block是true，判断跳过
	if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
		(c.dataqsiz > 0 && c.qcount == c.dataqsiz)) 
		return false
	

	// 关于cpu调度
	var t0 int64
	if blockprofilerate > 0 
		t0 = cputicks()
	

	//获取chann的锁
	lock(&c.lock)

	//CASE 2: send已关闭chan会panic
	if c.closed != 0 
		unlock(&c.lock)
		panic(plainError("send on closed channel"))
	

	//CASE 3: c.recvq队列中有等待接收goroutine。调用send函数将ep数据发给sg(recvGoroutine),稍后看send函数
	if sg := c.recvq.dequeue(); sg != nil 
		send(c, sg, ep, func()  unlock(&c.lock) , 3)
		return true
	

	//CASE 4: 有缓冲管道,并且有空间可写。将消息复制到等待goroutine
	if c.qcount < c.dataqsiz 
		// 定位写chan索引位置。c.sendx为写chan的个数位置，qp定位在 c+sendx*数据类型长度
		qp := chanbuf(c, c.sendx)
		//将需要写入的数据ep拷贝到qp指向的地址
		typedmemmove(c.elemtype, qp, ep)
		//chan每次只能写入一个元素
		c.sendx++
		//sendx为写入chann的索引地址，==dataqsiz时缓冲区写满。带缓冲chan使用环形地址，写入索引sendx变为0
		if c.sendx == c.dataqsiz 
			c.sendx = 0
		
		//数据长度++后解锁
		c.qcount++
		unlock(&c.lock)
		return true
	

	if !block 
		unlock(&c.lock)
		return false
	

	// CASE5: 缓冲队列已满,新建(sudog)G,并加入到sendq队列
	//获取当前goroutine
	gp := getg()
	// 创建一个等待队列抽象goroutine(sudog)
	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
	// mySg加入到send队列
	c.sendq.enqueue(mysg)
	//然后此协程阻塞,直到获取到c.lock
	goparkunlock(&c.lock, waitReasonChanSend, traceEvGoBlockSend, 3)

关于send过程

// send在空chan上执行写操作。ep值被发送者copy到recv队列中的sg中。chan必须为空并且锁定，send通过unlockf函数。注意调用此函数时，对端有等待recvq阻塞，说明缓冲为空或无缓冲
// 为其解锁. sg必须被退出c的队列，ep值必须非空并指向堆或者调用栈。
// use eg: send(c, recvSg, ep, func()  unlock(&c.lock) , 3)
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) 
   //接收sg的elem在等待队列中时，一般会指向一片内存(栈或堆), 检测到不为空时，直接将数据写入那片内存并将指针置空
   if sg.elem != nil 
   	// 将数据ep数据复制给sg的存储数据地址
   	sendDirect(c.elemtype, sg, ep)
   	sg.elem = nil
   
   gp := sg.g
   // 解锁chan
   unlockf()
   gp.param = unsafe.Pointer(sg)
   if sg.releasetime != 0 
   	sg.releasetime = cputicks()
   
   //gp 置位可运行态
   goready(gp, skip+1)

读channal操作

以下为 “从chan中读数据到ep” 情况：

• CASE 1: 读未初始化的chan，程序将永久阻塞。此时go run 报错： fatal error: all goroutines are asleep - deadlock! ，goroutine被强制结束
• CASE 2: 与写chan不同。读已关闭chan，不会发生panic，如果缓冲区无数据，返回true(表示读取操作成功) 和 false(表示未读到数据)
• CASE 3: 对端有阻塞等待写goroutine 且 无缓冲区，直接将数据写入对端goroutine
• CASE 4: 对端有阻塞等待写goroutine 且 是有缓冲chan时，此时缓冲队列已满，又因为readx和recvx索引为环形队列，所以此时两值相等。 从缓冲区recvx处取出数据复制给ep，再将写goroutine的数据写入缓冲区，过程中队列依旧满。
• CASE 5：对端无阻塞等待写goroutine，且缓冲有数据时，从recvx处复制数据给ep。
  (recv时，ep可为nil，此时相当于清理chan缓冲中一个元素)

// 编译代码 <- c 入口函数 
func chanrecv1(c *hchan, elem unsafe.Pointer) 
   chanrecv(c, elem, true)

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) 
   //CASE 1：读未初始化的chan,永久阻塞
   if c == nil 
   	if !block 
   		return
   	
   	gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
   	throw("unreachable")
   

   if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||
   	c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&
   	atomic.Load(&c.closed) == 0 
   	return
   

   lock(&c.lock)
   //CASE 2：已关闭且数据长度为0.解锁,清理ep的数据指向内存,返回true(chan正常), false(未取到数据)
   if c.closed != 0 && c.qcount == 0 
   	unlock(&c.lock)
   	if ep != nil 
   		// 清理内存
   		typedmemclr(c.elemtype, ep)
   	
   	return true, false
   

   //CASE 3: sendq队列中有阻塞gorotine时，recv数据。分为有缓冲和无缓冲情况，在recv函数中
   if sg := c.sendq.dequeue(); sg != nil 
   	recv(c, sg, ep, func()  unlock(&c.lock) , 3)
   	return true, true
   

   //CASE 4: 无sendSg但chan缓冲中有数据，取数据返回true true
   if c.qcount > 0 
   	// Receive directly from queue 找到读索引位置
   	qp := chanbuf(c, c.recvx)
   	//ep不为空时读走数据
   	if ep != nil 
   		typedmemmove(c.elemtype, ep, qp)
   	
   	//ep清理读索引指向的那一个数据(即不管左端是否有ep接受数据,这个数据都没有了)
   	typedmemclr(c.elemtype, qp)
   	//读索引++
   	c.recvx++
   	if c.recvx == c.dataqsiz 
   		c.recvx = 0
   	
   	//队列中数据量-1
   	c.qcount--
   	unlock(&c.lock)
   	return true, true
   

   if !block 
   	unlock(&c.lock)
   	return false, false
   

   //CASE 4: 缓冲无数据，创建sudog加入到recvqueue中阻塞等待chan有新数据
   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)
   goparkunlock(&c.lock, waitReasonChanReceive, traceEvGoBlockRecv, 3)
   releaseSudog(mysg)
   return true, !closed

recv 过程

// 此函数用于从sg(写chan的goroutine)读取数据到ep上，注意调用此函数时，说明缓冲区已满，所以对端有goroutine阻塞(sendx=datasiz  recvx=0) 或者无缓冲区
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) 
   // CASE 1: 无缓冲chan
   if c.dataqsiz == 0 
   	//直接从sg数据区 copy数据到ep
   	if ep != nil 
   		recvDirect(c.elemtype, sg, ep)
   	
   	
   //CASE 2: 有缓冲chan
    else 
   	//找到读数据的索引位置
   	qp := chanbuf(c, c.recvx)
       // 从chan缓冲区读索引位置读c.elemtype类型大小的内存到ep中
   	if ep != nil 
   		typedmemmove(c.elemtype, ep, qp)
   	
   	// 然后从sendSg中取数据追加给chan缓冲区，recv++
   	typedmemmove(c.elemtype, qp, sg.elem)
   	// chan写数据索引++.索引地址达到最大时归零
   	c.recvx++
   	if c.recvx == c.dataqsiz 
   		c.recvx = 0
   	
   	// recvx++后，sendx=recvx 将内存区当做环形来用，相当于sendx++ 
   	c.sendx = c.recvx  // c.sendx = (c.sendx+1) % c.dataqsiz
   
   sg.elem = nil
   gp := sg.g
   unlockf()
   gp.param = unsafe.Pointer(sg)
   if sg.releasetime != 0 
   	sg.releasetime = cputicks()
   
   goready(gp, skip+1)

close channal过程

func closechan(c *hchan) 
   //关闭未初始化chan，panic
   if c == nil 
   	panic(plainError("close of nil channel"))
   
   //close已经关闭chan, panic
   lock(&c.lock)
   if c.closed != 0 
   	unlock(&c.lock)
   	panic(plainError("close of closed channel"))
   
   c.closed = 1

   var glist *g

   // 释放所有的阻塞的读goroutine
   for 
   	//获取读数据的队列,清理接收队列中的数据空间等
   	sg := c.recvq.dequeue()
   	if sg == nil 
   		break
   	
   	if sg.elem != nil 
   		//清除sg.elem指向的内存为type的类型,然后指向空
   		typedmemclr(c.elemtype, sg.elem)
   		sg.elem = nil
   	
   	if sg.releasetime != 0 
   		sg.releasetime = cputicks()
   	
   	gp := sg.g
   	gp.param = nil
   	gp.schedlink.set(glist)
   	glist = gp
   

   // 释放所有的阻塞的写goroutine(将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 = nil
   	gp.schedlink.set(glist)
   	glist = gp
   
   unlock(&c.lock)

   // 依次唤醒所有阻塞的goroutine
   for glist != nil 
   	gp := glist
   	glist = glist.schedlink.ptr()
   	gp.schedlink = 0
   	goready(gp, 3)
   

遗留问题：
为什么指针类型的chan make时，hchan要和缓冲区分开分配。而非指针类型是连续的一块存储？？ 好像跟gc有关系
为什么清理读goroutine 未清理写goroutine?? 一般读goroutine的数据区是等待写入数据或为nil的，一般无数据可释放。 写goroutine一般是已经初始化的数据