Go语言channel与select原理

本文会尝试解释 go runtime 中 channel 和 select 的具体实现,部分内容来自 gophercon2017。Go版本为1.8.3

channel

第一部分讲述一下 channel 的用法。channel 可以看做一个队列,用于多个goroutine之间的通信,例如下面的例子,一个goroutine发送msg,另一个msg接受消息。channel 分为带缓冲和不带缓冲,差别不是很大,具体请自行google。看一个简单的例子,了解一下channel的使用。

package mainimport "fmt"func main() {// Create a new channel with `make(chan val-type)`.// Channels are typed by the values they convey.messages := make(chan string)// Send a value into a channel using the `channel <-`// syntax. Here we send `"ping"`  to the `messages`// channel we made above, from a new goroutine.go func() { messages <- "ping" }()// The `<-channel` syntax receives a value from the// channel. Here we'll receive the `"ping"` message// we sent above and print it out.msg := <-messagesfmt.Println(msg)
}

channel的功能点:

  1. 队列
  2. 阻塞
  3. 当一端阻塞,可以被另一个端唤醒

我们围绕这3点功能展开,讲讲具体的实现。

channel结构

注释标注了几个重要的变量,从功能上大致可以分为两个功能单元,一个是 ring buffer,用于存数据; 一个是存放 goroutine 的队列。

type hchan struct {qcount   uint           // 当前队列中的元素个数dataqsiz uint           // 缓冲队列的固定大小buf      unsafe.Pointer // 缓冲数组elemsize uint16closed   uint32elemtype *_type // element typesendx    uint   // 下一次发送的 indexrecvx    uint   // 下一次接收的 indexrecvq    waitq  // 接受者队列sendq    waitq  // 发送者队列// 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
}

Ring Buffer

主要是以下变量组成的功能, 一个 buf 存储实际数据,两个指针分别代表发送,接收的索引位置,配合 size, count 在数组大小范围内来回滑动。

qcount   uint           // 当前队列中的元素个数
dataqsiz uint           // 缓冲队列的固定大小
buf      unsafe.Pointer // 缓冲数组
sendx    uint   // 下一次发送的 index
recvx    uint   // 下一次接收的 index

举个例子,假设我们初始化了一个带缓冲的channel, ch := make(chan int, 3), 那么它初始状态的值为:

qcount   = 0
dataqsiz = 3
buf      = [3]int{0, 0, 0} // 表示长度为3的数组
sendx    = 0
recvx    = 0

第一步,向 channel 里 send 一个值, ch <- 1, 因为现在缓冲还没满,所以操作后状态如下:

qcount   = 1
dataqsiz = 3
buf      = [3]int{1, 0, 0} // 表示长度为3的数组
sendx    = 1
recvx    = 0

快进两部,连续向 channel 里 send 两个值 (2, 3),状态如下:

qcount   = 3
dataqsiz = 3
buf      = [3]int{1, 2, 3} // 表示长度为3的数组
sendx    = 0 // 下一个发送的 index 回到了0
recvx    = 0

从 channel 中 receive 一个值, <- ch, 状态如下:

qcount   = 2
dataqsiz = 3
buf      = [3]int{1, 2, 3} // 表示长度为3的数组
sendx    = 0 // 下一个发送的 index 回到了0
recvx    = 1 // 下一个接收的 index

阻塞

我们看下,如果 receive channel 时,channel 的 buffer中没有数据是怎么处理的。逻辑在 chanrecv 这个方法中,它的大致流程如下,仅保留了阻塞操作的代码。

func chanrecv(t *chantype, c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {// 检查 channdel 是否为 nil// 当不阻塞时,检查buffer大小,当前大小,检查chennel是否关闭,看看是否能直接返回// 检查发送端是否有等待的goroutine,下部分会提到// 当前buffer中有数据,则尝试取出。// 如果非阻塞,直接返回// 没有sender等待,buffer中没有数据,则阻塞等待。gp := getg()mysg := acquireSudog()mysg.releasetime = 0if t0 != 0 {mysg.releasetime = -1}// No stack splits between assigning elem and enqueuing mysg// on gp.waiting where copystack can find it.mysg.elem = epmysg.waitlink = nilgp.waiting = mysgmysg.g = gpmysg.selectdone = nilmysg.c = cgp.param = nilc.recvq.enqueue(mysg)//关键操作:设置 goroutine 状态为 waiting, 把 G 和 M 分离goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)// someone woke us up// 被唤醒,清理 sudogif mysg != gp.waiting {throw("G waiting list is corrupted")}gp.waiting = nilif mysg.releasetime > 0 {blockevent(mysg.releasetime-t0, 2)}closed := gp.param == nilgp.param = nilmysg.c = nilreleaseSudog(mysg)return true, !closed
}

这里的操作就是 创建一个 当前 goroutine 的 sudog, 然后把这个 sudog 放入 channel 的接受者等待队列;设置当前 G 的状态,和 M分离,到这里当前G就阻塞了,代码不会执行下去。
当被唤醒后,执行sudog的清理操作。这里接受buffer中的值的指针是 ep 这个变量,被唤醒后好像没有向 ep 中赋值的操作。这个我们下部分会讲。

sudog

还剩最后一个疑问,当一个goroutine因为channel阻塞,另一个goroutine是如何唤醒它的。

channel 中有两个 waitq 类型的变量, 看下结构发现,就是sudog的链表,关键是 sudog。sudog中包含了goroutine的引用,注意一下 elem这个变量,注释说可能会指向stack。

type waitq struct {first *sudoglast  *sudog
}type sudog struct {// The following fields are protected by the hchan.lock of the// channel this sudog is blocking on. shrinkstack depends on// this.g          *gselectdone *uint32 // CAS to 1 to win select race (may point to stack)next       *sudogprev       *sudogelem       unsafe.Pointer // data element (may point to stack)// The following fields are never accessed concurrently.// waitlink is only accessed by g.acquiretime int64releasetime int64ticket      uint32waitlink    *sudog // g.waiting listc           *hchan // channel
}

讲阻塞部分的时候,我们看到goroutine被调度之前,有一个 enqueue操作,这时,当前G的sudog已经被存入recvq中,我们看下发送者这时的操作。

这里的操作是,sender发送的值 直接被拷贝到 sudog.elem 了。然后唤醒 sudog.g ,这样对面的receiver goroutine 就被唤醒了。具体请下面的注释。

func chansend(t *chantype, c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {// 检查工作// 如果能从 chennel 的 recvq 弹出 sudog, 那么直接sendif 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) })return true}// buffer有空余空间,返回; 阻塞操作
}func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func()) {// 处理 index// 关键if sg.elem != nil {// 这里是根据 elemtype.size 复制内存sendDirect(c.elemtype, sg, ep)sg.elem = nil}// 一些处理// 重新设置 goroutine 的状态,唤醒它goready(gp, 4)
}func sendDirect(t *_type, sg *sudog, src unsafe.Pointer) {// src is on our stack, dst is a slot on another stack.// Once we read sg.elem out of sg, it will no longer// be updated if the destination's stack gets copied (shrunk).// So make sure that no preemption points can happen between read & use.dst := sg.elemtypeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.size)memmove(dst, src, t.size)
}// memmove copies n bytes from "from" to "to".
// in memmove_*.s
//go:noescape
func memmove(to, from unsafe.Pointer, n uintptr)

select

在看 chanrecv()方法 时,发现了一个 block 参数,代表操作是否阻塞。一般情况下,channel 都是阻塞的(不考虑buffer),那什么时候非阻塞呢?

第一个想到的就是 select, 在写了default case的时候,其他的channel是非阻塞的。

还有一个可能不常用,就是 channel 的反射 value, 可以是非阻塞的,这个方法是public的,我们先看下简单的。

func (v Value) TryRecv() (x Value, ok bool)
func (v Value) TrySend(x Value) bool

select 就复杂一点点,首先在源码中发现一段注释:

// compiler implements
//
//    select {
//    case c <- v:
//        ... foo
//    default:
//        ... bar
//    }
//
// as
//
//    if selectnbsend(c, v) {
//        ... foo
//    } else {
//        ... bar
//    }
//
func selectnbsend(t *chantype, c *hchan, elem unsafe.Pointer) (selected bool) {return chansend(t, c, elem, false, getcallerpc(unsafe.Pointer(&t)))
}// compiler implements
//
//    select {
//    case v = <-c:
//        ... foo
//    default:
//        ... bar
//    }
//
// as
//
//    if selectnbrecv(&v, c) {
//        ... foo
//    } else {
//        ... bar
//    }
//
func selectnbrecv(t *chantype, elem unsafe.Pointer, c *hchan) (selected bool) {selected, _ = chanrecv(t, c, elem, false)return
}

如果是一个 case + default 的模式,那么编译器就调用以上方法来实现。

如果是多个 case + default 的模式呢?select 在runtime到底是如何执行的?写个简单的select编译一下。

package mainfunc main() {var ch chan intselect {case <-ch:case ch <- 1:default:}
}

go tool compile -S -l -N test.go > test.s 结果中找一下关键字,例如:

0x008c 00140 (test.go:5)    CALL    runtime.newselect(SB)
0x00ad 00173 (test.go:6)    CALL    runtime.selectrecv(SB)
0x00ec 00236 (test.go:7)    CALL    runtime.selectsend(SB)
0x0107 00263 (test.go:8)    CALL    runtime.selectdefault(SB)
0x0122 00290 (test.go:5)    CALL    runtime.selectgo(SB)

这里 selectgo 是实际运行的方法,找一下,注意注释。先检查channel是否能操作,如果不能操作,就走 default 逻辑。

loop:// pass 1 - look for something already waitingvar dfl *scasevar cas *scasefor i := 0; i < int(sel.ncase); i++ {cas = &scases[pollorder[i]]c = cas.cswitch cas.kind {// 接受数据case caseRecv:sg = c.sendq.dequeue()// 如果有 sender 在等待if sg != nil {goto recv}// 当前buffer中有数据if c.qcount > 0 {goto bufrecv}// 关闭的channelif c.closed != 0 {goto rclose}case caseSend:if raceenabled {racereadpc(unsafe.Pointer(c), cas.pc, chansendpc)}// 关闭if c.closed != 0 {goto sclose}// 有 receiver 正在等待sg = c.recvq.dequeue()if sg != nil {goto send}// 有空间接受if c.qcount < c.dataqsiz {goto bufsend}// 走defaultcase caseDefault:dfl = cas}}if dfl != nil {selunlock(scases, lockorder)cas = dflgoto retc}

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

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Xadmin添加用户小组件出错

环境&#xff1a; Python 3.5.6 Django 2.1 Xadmin 原因&#xff1a; render函数在django2.1上有变化 解决方案&#xff1a; 1.在Python终端输入命令help(xadmin) 查看xadmin安装位置 得到如下输出 FILE/root/anaconda3/envs/learndjango/lib/python3.5/site-packages/xad…

成本预算的四个步骤_全网推广步骤有哪些?

全网推广的步骤是什么&#xff1f;一般来说&#xff0c;搜索引擎优化是大多数中小企业常用的推广方法。主要是通过对一些搜索引擎的排名来提高网站的曝光率&#xff0c;从而更好的提高自己网站的流量&#xff0c;从而更好的实现互联网层面的销售。接下来&#xff0c;让我们学习…

undefined reference to `std::cout'等错误

&#xff08;1&#xff09;gcc和g都是GNU(组织)的一个编译器。 &#xff08;2&#xff09;后缀名为.c的程序和.cpp的程序g都会当成是c的源程序来处理。而gcc不然&#xff0c;gcc会把.c的程序处理成c程序。 &#xff08;3&#xff09;对于.cpp的程序&#xff0c;编译可以用gcc/g…

FFPLAY的原理(二)

关于包Packets的注释从技术上讲一个包可以包含部分或者其它的数据&#xff0c;但是ffmpeg的解释器保证了我们得到的包Packets包含的要么是完整的要么是多种完整的帧。现在我们需要做的是让SaveFrame函数能把RGB信息定稿到一个PPM格式的文件中。我们将生成一个简单的PPM格式文件…

python生成requirements.txt的两种方法

python项目如何在另一个环境上重新构建项目所需要的运行环境依赖包&#xff1f; 使用的时候边记载是个很麻烦的事情&#xff0c;总会出现遗漏的包的问题&#xff0c;这个时候手动安装也很麻烦&#xff0c;不能确定代码报错的需要安装的包是什么版本。这些问题&#xff0c;requi…

node.js 安装使用http-server

node.js npm全局安装了http-server后我该怎么使用它&#xff1f;我在它的安装目录下创建了inde.html&#xff0c;浏览器localhost:8080可以访问&#xff0c;那我的项目需要放在它的安装目录下&#xff1f;还是需要在我的项目下配置什么或者使用什么指令启动它&#xff1f;我在我…

D - 卿学姐与魔法

卿学姐与魔法 Time Limit: 1200/800MS (Java/Others) Memory Limit: 65535/65535KB (Java/Others) Submit Status“你的膜法也救不了你 在去拯救公主的道路上&#xff0c;卿学姐披荆斩棘&#xff0c;刀刃早已锈迹斑斑。 一日卿学姐正在为武器的问题发愁&#xff0c;碰到了正…

python对excel表统计视频教程_Python实现对excel文件列表值进行统计的方法

本文实例讲述了Python实现对excel文件列表值进行统计的方法。分享给大家供大家参考。具体如下&#xff1a;#!/usr/bin/env python#codinggbk#此PY用来统计一个execl文件中的特定一列的值的分类import win32com.clientfilenameraw_input("请输入要统计文件的详细地址&#…

mooc后台管理系统设计

摘 要 本设计采用Python中的Django框架实现Mooc后台管理界面设计,django是一个完整的开源web开源框架,使用起来能够快速的搭建你想要的网站,由于django自带后台管理系统,本设计中后台管理模板采用功能更加强大的Xadmin实现。数据库部分采用mysql5.7,由于django中有自带封装的数…

DirectShow系统初级指南

流媒体的处理&#xff0c;以其复杂性和技术性&#xff0c;一向广受工业界的关注。特别伴随着因特网的普及&#xff0c;流媒体在网络上的广泛应用&#xff0c;怎样使流媒体的处理变得简单而富有成效逐渐成为了焦点问题。选择一种合适的应用方案&#xff0c;事半功倍。此时&#…

正则正整数含0

^0?$|^([1-9][0-9]*)?$

MySQL 数据库导出导入操作

有时需要将 MySQL 数据库中的数据导入到其它的数据库中&#xff0c;这里以从 Ubuntu 系统的 MySQL 数据库导出 zabbix 这个数据库到 Windows 系统中的MySQL 为例。 导出数据库 导出数据其实非常方便&#xff0c;比如将 MySQL 中的 zabbix 这个数据库导出到当前文件夹&#xff…

您的apple id 暂时不符合使用此应用程序_Mac相机不工作时该怎么办

苹果公司的许多台式机和笔记本电脑都包含一个内置网络摄像头&#xff0c;该公司愉快地将其称为FaceTime相机。但是&#xff0c;如果您的Mac网络摄像头无法正常工作&#xff0c;并且在尝试访问它时显示为断开连接或不可用&#xff0c;则您可能不会感到高兴。您可以尝试以下操作来…

基于DirectShow的流媒体解码和回放

一、 前言  流媒体的定义很广泛&#xff0c;大多数时候指的是把连续的影像和声音信息经过压缩处理后放上网站服务器&#xff0c;让用户一边下载一边观看、收听&#xff0c;而不需要等整个压缩文件下载到自己机器就可以观看的视频/音频传输、压缩技术。流媒体也指代由这种技术…

《知易行难》扩展练习

在学习了《知易行难》后&#xff0c;这个是一个选做的扩展练习&#xff0c;但是里面的问题真的的很好&#xff0c;所以我也将在这里真实的分享&#xff0c;但是有些敏感的人名我就隐去了。 1. 这一年你做了些什么事情&#xff1f; 1&#xff09;团队的整合&#xff0c;将团队…

python 裁判文书网_python - 用selenium模拟登陆裁判文书网,系统报错找不到元素。...

问 题from selenium import webdriverfrom selenium.webdriver.common.desired_capabilities import DesiredCapabilitiesdcap dict(DesiredCapabilities.PHANTOMJS)dcap["phantomjs.page.settings.userAgent"]("Mozilla/5.0 (Windows NT 6.3; WOW64) AppleWeb…

Python 四大主流 Web 编程框架

目前Python的网络编程框架已经多达几十个&#xff0c;逐个学习它们显然不现实。但这些框架在系统架构和运行环境中有很多共通之处&#xff0c;本文带领读者学习基于Python网络框架开发的常用知识,及目前的4种主流Python网络框架&#xff1a;Django、Tornado、Flask、Twisted。 …

汕头市队赛 SRM16 T2

描述 猫和老鼠&#xff0c;看过吧&#xff1f;猫来了&#xff0c;老鼠要躲进洞里。在一条数轴上&#xff0c;一共有n个洞&#xff0c;位置分别在xi&#xff0c;能容纳vi只老鼠。一共有m只老鼠位置分别在Xi&#xff0c;要躲进洞里&#xff0c;问所有老鼠跑进洞里的距离总和最小是…

基于django和vue的xdh官网设计

前言 本项目是使用三段分离的设计 前台 使用materialize框架搭建的前台页面,后端使用的django写的接口 后台 使用Amazon UI 模板搭建的界面,管理各个部分的内容 项目环境 python3.7.2 django2.2.9 vue axios jQuery materialize mysql摘 要 本设计采用前后端分离的设计…

C#调用WebService实例和开发(转)

http://www.cnblogs.com/peterpc/p/4628441.html 一、基本概念 Web Service也叫XML Web Service WebService是一种可以接收从Internet或者Intranet上的其它系统中传递过来的请求&#xff0c;轻量级的独立的通讯技术。是:通过SOAP在Web上提供的软件服务&#xff0c;使用WSDL文件…