Go 源码之 Chan

Go 源码之 chan

go源码之chan - Jxy 博客

目录

  • Go 源码之 chan
    • 一、总结
    • 二、源码
      • (一)hchan
      • (二)创建
      • (三)发送
      • (四)接收
      • (五)关闭
    • 三、常见问题
      • 1.为什么要使用环形队列
      • 2. 关于chan的操作
      • 有劳各位看官 `点赞、关注➕收藏 `,你们的支持是我最大的动力!!!
      • 接下来会不断更新 `golang` 的一些`底层源码及个人开发经验`(个人见解)!!!
      • 同时也欢迎大家在评论区提问、分享您的经验和见解!!!

一、总结

chan 提供了一种在 goroutine 之间进行数据交换和同步的方式。通道可以用于控制并发访问和共享数据,从而减少竞态条件和死锁问题,并且可以自然地处理异步事件和信号。如果你的应用程序需要在 goroutine 之间传递数据或消息,那么通道是一个不错的选择

  • 内部是一个 hchan 结构(字段见源码),环形队列 + 发送者双向链表 + 接收者双向链表 + 锁
  • channel 与 select 语句结合使用时,底层调用的还是 chansendchanrecv 函数
  • channel
    • 结构:环形缓存、sendq、recvq;
    • 流程:上锁/解锁,阻塞/非阻塞,缓冲/非缓冲,缓存入队出队,sudog 入队出队,协程休眠/唤醒

二、源码

/src/runtime/chan.go

  • 一个环形队列
  • 两个双向列表

image-20230323100951975.png

(一)hchan

buf + sendx + recvx 形成环形队列


type hchan struct {
	qcount   uint           // 队列中现存元素数量
	dataqsiz uint           // 队列容量(缓冲区)
	buf      unsafe.Pointer // 队列,指向一个动态分配的数组,用于存储 channel 中的元素
	elemsize uint16         // 队列中元素大小
	closed   uint32         // 0 正常	,1 关闭
	elemtype *_type     		// 队列中元素类型,
  
	sendx    uint   				// 队列(buf)已发送位置,当(sendx++)==dataqsiz,则从头开始发,sendx=0
	recvx    uint   				// 队列(buf)已接收位置;
  												// 当 `sendx` 和 `recvx` 相等时,channel 中无元素,发送 / 接收 操作阻塞
  
	recvq    waitq  				// 双向链表 ,FIFO 由 recv 行为(也就是读 <-ch)阻塞在 channel 上的 goroutine 队列
	sendq    waitq 					// 双向链表 ,FIFO 由 send 行为 (也就是写 ch<-) 阻塞在 channel 上的 goroutine 队列

	
	lock mutex 							// 读写锁,保护hchan中的所有字段,以及waitq中所有的字段
}

// 双向链表,存储了g
type waitq struct {
	first *sudog  // 链表头部,协程 g 的数据结构
	last  *sudog  // 链表尾部,协程 g 的数据结构
}

(二)创建

ch1 := make(chan int)
ch2 := make(chan int,2)

底层都是调用了runtime.makechan()

  • 合法性校验
    • 数据类型大小校验
    • 内存溢出校验
  • 初始化 hchan
    • 初始化 无缓冲 hchan
    • 初始化 有缓冲 && 无指针元素 hchan
    • 初始化 无缓冲 && 有指针元素 hchan
    • 初始化 hchan 其他元素:如 dataqsize、elemsize、elemtype、lock
// 主要逻辑:合法性验证 和 分配地址空间
// t 是指向 chantype 的指针,size 表示缓冲区大小,0表无缓冲
func makechan(t *chantype, size int) *hchan {
	elem := t.elem // 元素的类型

  // ----------- 1. 合法性验证 ----------
	// 数据类型大小验证,大于1<<16时异常
	if elem.size >= 1<<16 {
		throw("makechan: invalid channel element type")
	}
  // 内存对齐(降低寻址次数),大于最大内存(8字节数)时异常
	if hchanSize%maxAlign != 0 || elem.align > maxAlign {
		throw("makechan: bad alignment")
	}

  // 传入的size大于堆可分配的最大内存时:内存溢出异常
	mem, overflow := math.MulUintptr(elem.size, uintptr(size))
	if overflow || mem > maxAlloc-hchanSize || size < 0 {
		panic(plainError("makechan: size out of range"))
	}

  // ----------- 2. 分配地址空间 ----------
  // hchanSize 为 hchan 结构大小
  // mem 为缓存区大小
  /* 根据 channel 中收发元素的类型和缓冲区的大小初始化 runtime.hchan 和 缓冲区,分为三种情况:
  		* 如果不存在缓冲区,分配 hchan 结构体空间,即无缓存 channel
  		* 如果 channel 存储的类型不是指针类型,分配连续地址空间,包括 hchan 结构体 + 数据
  		* 默认情况包括指针,为 hchan 和 buf 单独分配数据地址空间
  	更新 hchan 结构体的数据,包括 elemsize、elemtype 和 dataqsiz
	*/
	var c *hchan
	switch {
    case mem == 0:
    	 // 创建无缓冲的 chan ,buf==0 ,初始化 hchan
       c = (*hchan)(mallocgc(hchanSize, nil, true)) // hchanSize表示空hchan需要占用的字节
       c.buf = c.raceaddr()  //  raceaddr内部实现为:return unsafe.Pointer(&c.buf)
    case elem.ptrdata == 0:
       // 有缓存区,并且队列中不存在指针,分配连续地址空间,大小为 hchanSize + mem
       c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
       // buf指针指向空hchan占用空间的末尾
       c.buf = add(unsafe.Pointer(c), hchanSize)
    default:
       // 队列包含指针类型
       // 为buf单独开辟mem大小的空间,用来保存所有的数据
       c = new(hchan)
       c.buf = mallocgc(mem, elem, true)
    }

	c.elemsize = uint16(elem.size)  			// 元素大小
	c.elemtype = elem 										// 元素类型
	c.dataqsiz = uint(size) 							// chan 缓存区大小
	lockInit(&c.lock, lockRankHchan) 			// 初始化锁

	if debugChan {
		print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
	}
	return c
}

(三)发送

ch <- 1

执行 runtime.chansend1(SB)

  • 异常检查
    • 发送到 nil chan 中,阻塞挂起
    • 往 closed chan 发送(写),则 panic
    • 当前 chan 是否可以发送
  • 同步发送:recvq 中存在等待接收者,则直接唤醒并发送数据
  • 异步发送:c.qcount < c.dataqsiz 缓存区空闲,则数据发送到缓存区
  • 阻塞发送:当前面都不满足时 且 block = true 时:发送操作 线程阻塞 挂起,并且添加到 sendq 等待队列,直到有接收者接收才释放
/**
* @Description: 
 chansend函数主要可以归纳为四部分:
	 异常检查、同步发送、异步发送、阻塞发送:
* @Param:c:hchan结构;ep:发送的元素;block:是否阻塞;callerpc:
* @return: true发送成功,false发送失败
*/
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  // ------------------------------------ 1.异常检查 ------------------------------------
	if c == nil { 
		if !block {
			return false
		}
		gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2) 	// 发送到 nil chan 中,阻塞挂起
		throw("unreachable")
	}
	..........

 																							 // 当channel不为nil,此时检查channel是否做好接收发送操作的准备,
  if !block && c.closed == 0 && full(c) { 	
		return false															// 非阻塞且未关闭: 1. 无缓存区,recvq为空 2. 有缓冲区,但是buffer已满
	}


	lock(&c.lock) // 先上锁

	if c.closed != 0 { // chan已经关闭,则解锁
		unlock(&c.lock)
		panic(plainError("send on closed channel")) 				// 往 closed chan 发送(写),则 panic
	}

  // ------------------------------------  2.同步发送 ------------------------------------
  																										// recvq 中存在等待接收者,则直接唤醒并发送数据
	if sg := c.recvq.dequeue(); sg != nil { 
		send(c, sg, ep, func() { unlock(&c.lock) }, 3)		// recvq 等待队列取取出 sg(sudog)并唤醒并发送数据 ep
		return true
	}
   
  // ------------------------------------  3.异步发送 ------------------------------------
  // (有缓存区,没有等待接收者,先发到缓冲区中,等有接收者再去读)
	if c.qcount < c.dataqsiz {  					// 存在的元素个数< 缓冲区:说明缓存区可以继续写数据
		qp := chanbuf(c, c.sendx) 					// 获取缓存区index地址
		typedmemmove(c.elemtype, qp, ep)		// 数据写入buffer
		c.sendx++ 													// 发送数据的下标++
		if c.sendx == c.dataqsiz { 					// 当发送数据的下标等于缓冲区,表数据发送完毕,从头开始
			c.sendx = 0
		}
		c.qcount++ 													// 元素数量++
		unlock(&c.lock) 										// 解锁
		return true 												// 返回结果
	}

	if !block {
		unlock(&c.lock) 										// 解锁
		return false
	}

  // ------------------------------------ 4. 阻塞发送 ------------------------------------
  // 当前面都不满足时(没有等待接收者,没有空闲缓冲区) 且 block = true 时,发送操作 线程阻塞 挂起,直到有接收者接收才释放:
	gp := getg()
	..........
  c.sendq.enqueue(mysg)  											// 将发送 的 sg 添加到 sendq 等待队列中
	return true
}
func full(c *hchan) bool {
	if c.dataqsiz == 0 { // 无缓冲
		return c.recvq.first == nil
	}
	// 有缓冲,现有元素的个数 是否等于 缓冲区容量时(缓冲区满)
	return c.qcount == c.dataqsiz
}
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
   ......
   if sg.elem != nil {
      sendDirect(c.elemtype, sg, ep) //  将数据拷贝到接收变量的内存地址上
      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) // 唤醒sudog协程;下一轮调度时会唤醒这个接收的 goroutine。
}
// 实现了等待队列的入队操作。它将一个元素添加到等待队列的末尾,并更新队列的 first 和 last 指针
func (q *waitq) enqueue(sgp *sudog) {
	sgp.next = nil // 表示该元素是队列的最后一个
	x := q.last // 将等待队列 q 中的最后一个元素(如果存在)赋值给变量 x。
	if x == nil { // 如果队列中最后一个都没有,则队列无元素,即 x 为 nil,
		sgp.prev = nil // 则将 sgp 元素的 prev 指针设为 nil,表示该元素是队列中的第一个元素,
		q.first = sgp //  然后将队列的 first 和 last 指针都指向该元素,表示该元素是队列中唯一的元素。
		q.last = sgp  // 然后直接返回,结束入队操作。
		return
	}
	sgp.prev = x // sgp 是新加的最后元素,需要关联前一个元素(x为原队列中最后一个元素)
	x.next = sgp // 设置x的的下一个元素为新加的元素
	q.last = sgp // 设置q队列的最后一个元素
}

(四)接收

i <- ch i, ok <- ch

执行 runtime.chanrecv1(SB) 都是调用的chanrecv()

  • 异常检查
    • 从 nil chan 中读取,阻塞挂起
    • 从 closed chan 接收(读),返回零值
    • 当前 chan 是否可以接收
  • 同步接收:sendq 中存在发送者,则直接唤醒并接收数据
  • 异步接收:c.qcount 队列中有元素,则则从 buf 中读取数据
  • 阻塞接收:当前面都不满足时 且 block = true 时:接收操作 线程阻塞 挂起,并且添加到 recvq 等待队列,直到有发送者才释放
chanrecv 函数的逻辑和 chansend 的逻辑基本一致

(五)关闭

close(ch)

closechan(c *hchan)

主要逻辑:

  • 异常检查:
    • 关闭 nil chan ,panic
    • 关闭 closed chan,panic
  • 标记 chan 为关闭状态
  • 释放等待的 sudog: 唤醒并调度等待队列 recvq、sendq 中的 sudog,所有接收者收到零值

func closechan(c *hchan) {
  // ------------------------------------ 1. 异常检查 ------------------------------------
	if c == nil {
		panic(plainError("close of nil channel")) 				// 关闭 nil chan ,panic
	}

	lock(&c.lock)																			// 上锁
	if c.closed != 0 {
		unlock(&c.lock)
		panic(plainError("close of closed channel"))		// 关闭 closed chan,panic
	}

	c.closed = 1																			// 标识chan已经关闭

   // ------------------------------------ 2. 释放等待的 sudog ------------------------------------
	var glist gList 																	// 存储 recvq、sendq 等待队列中的 sg(sudog)
	for {
		sg := c.recvq.dequeue()													// 将 recvq 等待队列中的  sg(sudog) 添加到 glist
		......
		glist.push(gp)
	}
	for {
		sg := c.sendq.dequeue()													// 将 sendq 等待队列中的  sg(sudog) 添加到 glist
		......
		glist.push(gp)
	}
	unlock(&c.lock)																		// 解锁

  for !glist.empty() {															//依次从 glist 中弹出 sg(sudog)并唤醒执行,所有接收者收到零值
		gp := glist.pop()
		gp.schedlink = 0
		goready(gp, 3)
	}
}

三、常见问题

1.为什么要使用环形队列

chan的内部使用环形队列来存取元素,每次发/收元素时,会根据sendx/recvx记录的位置从队列buf中存取元素,

所以环形队列:buf+sendx+recvx实现的,

使用环形数组实现的好处:

  • 避免对数组进行复制或者移动操作

    比如数组【1,2,3】,现在添加4,变为【2,3,4】,数组就需要进行复制拷贝操作,如果是环形队列,则直接将4添加到队列的尾部即可,

  • 避免内存分配和拷贝的开销,从而提高程序的性能

    重复利用,避免重新分配内存

2. 关于chan的操作

image-20230328143228214.png

有劳各位看官 点赞、关注➕收藏 ,你们的支持是我最大的动力!!!

接下来会不断更新 golang 的一些底层源码及个人开发经验(个人见解)!!!

同时也欢迎大家在评论区提问、分享您的经验和见解!!!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/510941.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

SV学习笔记(一)

SV&#xff1a;SystemVerilog 开启SV之路 数据类型 內建数据类型 四状态与双状态 &#xff1a; 四状态指0、1、X、Z&#xff0c;包括logic、integer、 reg、 wire。双状态指0、1&#xff0c;包括bit、byte、 shortint、int、longint。 有符号与无符号 &#xff1a; 有符号&am…

12313124

c语言中的小小白-CSDN博客c语言中的小小白关注算法,c,c语言,贪心算法,链表,mysql,动态规划,后端,线性回归,数据结构,排序算法领域.https://blog.csdn.net/bhbcdxb123?spm1001.2014.3001.5343 给大家分享一句我很喜欢我话&#xff1a; 知不足而奋进&#xff0c;望远山而前行&am…

【Linux】进程管理(2):进程控制

一、进程创建&#xff1a;fork函数 我们在命令行中输入man fork 即可得到fork函数的函数接口的函数的使用方法。 我们可以看到&#xff0c;fork函数位于man手册的第2部分&#xff0c;由于第2部分通常是用于描述系统调用和库函数&#xff0c;所以我们可以了解到fork函数实际是一…

【总结】在嵌入式设备上可以离线运行的LLM--Llama

文章目录 Llama 简介运用另一种&#xff1a;MLC-LLM 一个令人沮丧的结论在资源受限的嵌入式设备上无法运行LLM&#xff08;大语言模型&#xff09;。 一丝曙光&#xff1a;tinyLlama-1.1b&#xff08;10亿参数&#xff0c;需要至少2.98GB的RAM&#xff09; Llama 简介 LLaMA…

自动驾驶的世界模型:综述

自动驾驶的世界模型&#xff1a;综述 附赠自动驾驶学习资料和量产经验&#xff1a;链接 24年3月澳门大学和夏威夷大学的论文“World Models for Autonomous Driving: An Initial Survey”。 在快速发展的自动驾驶领域&#xff0c;准确预测未来事件并评估其影响的能力对安全性…

ssm017网上花店设计+vue

网上花店的设计与实现 摘 要 网络技术和计算机技术发展至今&#xff0c;已经拥有了深厚的理论基础&#xff0c;并在现实中进行了充分运用&#xff0c;尤其是基于计算机运行的软件更是受到各界的关注。加上现在人们已经步入信息时代&#xff0c;所以对于信息的宣传和管理就很关…

[云呐]固定资产盘点报告哪个部门写

固定资产盘点报告通常由哪个部门来完成和签发呢?总体来说,固定资产盘点报告主要由资产管理部门或核算部门具体组织拟定并与财务部门共同签发。个别重大报告还需要上级领导或委员会研讨通过。  资产管理部门:  资产管理部门是直接负责公司固定资产管理工作的核心部门,它主导…

超市销售数据-python数据分析项目

Python数据分析项目-基于Python的销售数据分析项目 文章目录 Python数据分析项目-基于Python的销售数据分析项目项目介绍数据分析结果导出数据查阅 数据分析内容哪些类别比较畅销?哪些商品比较畅销?不同门店的销售额占比哪个时间段是超市的客流高封期?查看源数据类型计算本月…

浅谈iOS开发中的自动引用计数ARC

1.ARC是什么 我们知道&#xff0c;在C语言中&#xff0c;创建对象时必须手动分配和释放适量的内存。然而&#xff0c;在 Swift 中&#xff0c;当不再需要类实例时&#xff0c;ARC 会自动释放这些实例的内存。 Swift 使用 ARC 来跟踪和管理应用程序的内存&#xff0c;其主要是由…

EFPN代码解读

论文 Extended Feature Pyramid Network for Small Object Detection python3 D:/Project/EFPN-detectron2-master/tools/train_net.py --config-file configs/InstanceSegmentation/pointrend_rcnn_R_50_FPN_1x_coco.yaml --num-gpus 1 训练脚本 cfg 中的配置 先获取配置…

JavaWeb 项目运行配置

JavaWeb 项目运行配置

保持ssh断开后,程序不会停止执行

保持ssh断开后&#xff0c;程序不会停止执行 一、前言 笔者做远程部署搞了一阵子&#xff0c;快结项时发现一旦我关闭了ssh连接窗口&#xff0c;远程服务器会自动杀掉我在ssh连接状态下运行的程序。 这怎么行&#xff0c;岂不是想要它一直运行还得要一台电脑一直打开ssh连接咯…

【优选算法专栏】专题十六:BFS解决最短路问题---前言

本专栏内容为&#xff1a;算法学习专栏&#xff0c;分为优选算法专栏&#xff0c;贪心算法专栏&#xff0c;动态规划专栏以及递归&#xff0c;搜索与回溯算法专栏四部分。 通过本专栏的深入学习&#xff0c;你可以了解并掌握算法。 &#x1f493;博主csdn个人主页&#xff1a;小…

【QingHub】企业级应用开发管理

QingHub 企业级应用开发设计器是QingHub Studio的一个核心模块&#xff0c;它可以实现应用搭建、团队管理&#xff0c;共享开发&#xff0c;可以快速接入API接口&#xff0c;复杂功能可以通过自定义脚本快速实现业务逻辑。打通前端开发与后台业务逻辑一体化。通过可视化的方式&…

使用 PDManer 对数据库表建模(建表语句生成,代码生成)

目录 前言 基本使用教程 新建项目 创建表 关系图 建表语句 生成代码 导入 前言 在软件开发中过程中&#xff0c;一般分为几个过程&#xff1a;需求分析、概要设计、详细设计、编码实现、软件测试和软件交付。 在概要设计和详细设计过程中&#xff0c;则需要对业务进…

苍穹外卖06(HttpClient,微信小程序开发,微信登录流程,获取授权码从微信平台获取用户信息)

目录 一、HttpClient 1. 介绍 2. 入门案例 1 导入依赖(已有) 2 GET方式请求 2 POST方式请求 二、微信小程序开发 1. 介绍 2. 准备工作 1 注册小程序获取AppID 注册小程序 完善小程序信息 2 下载并安装开发者工具 3 设置小程序开发者工具(必做) 3. 入门案例 1 小…

CentOS 7 下离线安装RabbitMQ教程

CentOS 7 下安装RabbitMQ教程一、做准备&#xff08;VMWare 虚拟机上的 CentOS 7 镜像 上安装的&#xff09; &#xff08;1&#xff09;准备RabbitMQ的安装包&#xff08;rabbitmq-server-3.8.5-1.el7.noarch&#xff09;下载地址mq https://github.com/rabbitmq/rabbitmq-se…

基于51单片机的简易计算器设计

1、任务 本课题模拟计算器设计硬件电路采用三部分电路模块构成&#xff0c; 第一部分是键盘模块电路&#xff0c;采用4*4矩阵式键盘作为输入电路&#xff1b; 第二部分是LCD1602液晶显示模块&#xff1b; 第三部分是以51单片机作为控制核心。 软件程序主要由三部分组成&am…

AWS-EKS 给其他IAM赋予集群管理权限

AWS EKS 设计了权限管理系统&#xff0c;A用户创建的集群 B用户是看不到并且不能管理和使用kubectl的&#xff0c;所以我们需要共同管理集群时就需要操场共享集群访问给其他IAM用户。 两种方式添加集群控制权限&#xff08;前提&#xff1a;使用有管理权限的用户操作&#xff…

子集与全排列问题(力扣78,90,46,47)

系列文章目录 子集和全排列问题与下面的组合都是属于回溯方法里的&#xff0c;相信结合前两期&#xff0c;再看这篇笔记&#xff0c;更有助于大家对本系列的理解 一、组合回溯问题 二、组合总和问题 文章目录 系列文章目录题目子集一、思路二、解题方法三、Code 子集II一、思…