Golang网络模型netpoll源码解析
0、引言
在学习完了Socket编程的基础知识、Linux系统提供的I/O多路复用的实现以及Golang的GMP调度模型之后,我们进而学习Golang的网络模型——netpoll。本文将从为什么需要使用netpoll模型,以及netpoll的具体流程实现两个主要角度来展开学习。当前使用的Go的版本为1.22.4,Linux系统。
1、为什么要使用netpoll模型?
首先,什么是多路复用?
多路,指的是存在着多个需要服务的对象;复用,指的是重复利用一个单元来为上述的多个目标提供服务。
我们知道,Linux系统为用户提供了三个内核实现的IO多路复用技术的系统调用,用发展时间来排序分别为:select->poll->epoll
。其中,epoll
在当今使用的最为广泛,对比与select
调用,它有以下的优势:
fd
数量灵活:可监听的fd
数量上限灵活,使用方可以在调用epoll_create
操作时自行指定。- 更少的内核拷贝次数:在内核中,使用红黑树的结构来存储需要监听的
fd
,相比与调用select
每次需要将所有的fd
拷贝进内核,监听到事件后再全部拷贝回用户态,epoll
只需要将需要监听的fd
添加到事件表后,即可多次监听。 - 返回结果明确:
epoll
运行将就绪事件添加到就绪事件列表中,当用户调用epoll_wait
操作时,内核只返回就绪事件,而select
返回的是所有的事件,需要用户再进行一次遍历,找到就绪事件再处理。
需要注意的是,在不同的条件环境下,epoll的优势可能反而作用不明显。epoll只适用在监听fd基数较大且活跃度不高的场景,如此epoll事件表的空间复用和epoll_wait操作的精准才能体现出其优势;而当处在fd基数较小且活跃度高的场景下,select反而更加简单有效,构造epoll的红黑树结构的消耗会成为其累赘。
考虑到场景的多样性,我们会选择使用epoll
去完成内核事件监听的操作,那么如何将golang
和epoll
结合起来呢?
在 Go 语言的并发模型中,GMP 框架实现了一种高效的协程调度机制,它屏蔽了操作系统线程的细节,用户可以通过轻量级的 Goroutine 来实现细粒度的并发操作。然而,底层的 IO 多路复用机制(如 Linux 的 epoll)调度的单位仍然是线程(M)。为了将 IO 调度从线程层面提升到协程层面,充分发挥 Goroutine 的高并发优势,netpoll 应运而生。
接下来我们就来学习netpoll
框架的实现。
2、netpoll实现原理
2.1、核心结构
1、pollDesc
为了将IO调度从线程提升到协程层面,netpoll
框架有个重要的核心结构pollDesc
,它有两个,一个为表层,含有指针指向了里层的pollDesc
。本文中讲到的pollDesc
都为里层pollDesc
。
表层pollDesc
定位在internel/poll/fd_poll_runtime.go
文件中:
type pollDesc struct {
runtimeCtx uintptr
}
使用一个runtimeCtx
指针指向其底层实现实例。
里层的位于runtime/netpoll.go
中。
//网络poller描述符
type pollDesc struct {
//next指针,指向在pollCache链表结构中,以下个pollDesc实例。
link *pollDesc
//指向fd
fd uintptr
//读事件状态标识器,状态有四种:
//1、pdReady:表示读操作已就绪,等待处理
//2、pdWait:表示g将要被阻塞等待读操作就绪,此时还未阻塞
//3、g:读操作的g已经被阻塞,rg指向阻塞的g实例
//4、pdNil:空
rg atomic.Uintptr
wg atomic.Uintptr
//...
}
pollDesc
的核心字段是读/写标识器rg/wg
,它用于标识fd的io事件状态,并且持有被阻塞的g实例。当后续需要唤醒这个g处理读写事件的时候,可以通过pollDesc
追溯得到g的实例进行操作。有了pollDesc
这个数据结构,Golang就能将对处理socket的调度单位从线程Thread
转换成协程G
。
2、pollCache
pollCache
缓冲池采用了单向链表的方式存储多个pollDesc
实例。
type pollCache struct {
lock mutex
first *pollDesc
}
其包含了两个核心方法,分别是alloc()
和free()
//从pollCache中分配得到一个pollDesc实例
func (c *pollCache) alloc() *pollDesc {
lock(&c.lock)
//如果链表为空,则进行初始化
if c.first == nil {
//pdSize = 248
const pdSize = unsafe.Sizeof(pollDesc{})
//4096 / 248 = 16
n := pollBlockSize / pdSize
if n == 0 {
n = 1
}
//分配指定大小的内存空间
mem := persistentalloc(n*pdSize, 0, &memstats.other_sys)
//完成指定数量的pollDesc创建
for i := uintptr(0); i < n; i++ {
pd := (*pollDesc)(add(mem, i*pdSize))
pd.link = c.first
c.first = pd
}
}
pd := c.first
c.first = pd.link
lockInit(&pd.lock, lockRankPollDesc)
unlock(&c.lock)
return pd
}
//free用于将一个pollDesc放回pollCache
func (c *pollCache) free(pd *pollDesc) {
//...
lock(&c.lock)
pd.link = c.first
c.first = pd
unlock(&c.lock)
}
2.2、netpoll框架宏观流程
在宏观的角度下,netpoll框架主要涉及了以下的几个流程:
poll_init
:底层调用epoll_create
指令,在内核态中开辟epoll事件表。poll_open
:先构造一个pollDesc实例,然后通过epoll_ctl(ADD)
指令,向内核中添加要监听的socket,并将这一个fd绑定在pollDesc中。pollDesc含有状态标识器rg/wg
,用于标识事件状态以及存储阻塞的g。poll_wait
:当g依赖的事件未就绪时,调用gopark
方法,将g置为阻塞态存放在pollDesc中。net_poll
:GMP调度器会轮询netpoll流程,通常会用非阻塞的方式发起epoll_wait
指令,取出就绪的pollDesc,提前出其内部陷入阻塞态的g然后将其重新添加到GMP的调度队列中。(以及在sysmon流程和gc流程都会触发netpoll)
3、流程源码实现
3.1、流程入口
我们参考以下的简易TCP服务器实现框架,走进netpoll框架的具体源码实现。
// 启动 tcp server 代码示例
func main() {
//创建TCP端口监听器,涉及以下事件:
//1:创建socket fd,调用bind和accept系统接口函数
//2:调用epoll_create,创建eventpool
//3:调用epoll_ctl(ADD),将socket fd注册到epoll事件表
l, _ := net.Listen("tcp", ":8080")
// eventloop reactor 模型
for {
//等待TCP连接到达,涉及以下事件:
//1:循环+非阻塞调用accept
//2:若未就绪,则调用gopark进行阻塞
//3:等待netpoller轮询唤醒
//4:获取到conn fd后注册到eventpool
//5:返回conn
conn, _ := l.Accept()
// goroutine per conn
go serve(conn)
}
}
// 处理一笔到来的 tcp 连接
func serve(conn net.Conn) {
//关闭conn,从eventpool中移除fd
defer conn.Close()
var buf []byte
//读取conn中的数据,涉及以下事件:
//1:循环+非阻塞调用recv(read)
//2:若未就绪,通过gopark阻塞,等待netpoll轮询唤醒
_, _ = conn.Read(buf)
//向conn中写入数据,涉及以下事件:
//1:循环+非阻塞调用writev (write)
//2:若未就绪,通过gopark阻塞,等待netpoll轮询唤醒
_, _ = conn.Write(buf)
}
3.2、Socket创建
以net.Listen
方法为入口,进行创建socket fd
,调用的方法栈如下:
方法 | 文件 |
---|---|
net.Listen() | net/dial.go |
net.ListenConfig.Listen() | net/dial.go |
net.sysListener.listenTCP() | net/tcpsock_posix.go |
net.internetSocket() | net/ipsock_posix.go |
net.socket() | net/sock_posix.go |
核心的调用在net.socket()
方法内,源码核心流程如下:
func socket(ctx context.Context, net string, family, sotype, proto int, ipv6only bool, laddr, raddr sockaddr, ctrlCtxFn func(context.Context, string, string, syscall.RawConn) error) (fd *netFD, err error) {
//进行socket系统调用,创建一个socket
s, err := sysSocket(family, sotype, proto)
//绑定socket fd
fd, err = newFD(s, family, sotype, net);
//...
//进行了以下事件:
//1、通过syscall bind指令绑定socket的监听地址
//2、通过syscall listen指令发起对socket的监听
//3、完成epollEvent表的创建(全局执行一次)
//4、将socket fd注册到epoll事件表中,监听读写就绪事件
err := fd.listenStream(ctx, laddr, listenerBacklog(), ctrlCtxFn);
}
首先先执行了sysSocket
系统调用,创建一个socket
,它是一个整数值,用于标识操作系统中打开的文件或网络套接字;接着调用newFD
方法包装成netFD
对象,以便实现更高效的异步 IO 和 Goroutine 调度。
3.3、poll_init
紧接3.2中的net.socket
方法,在内部还调用了net.netFD.listenStream()
,poll_init
的调用栈如下:
方法 | 文件 |
---|---|
net.netFD.listenStream() | net/sock_posix.go |
net.netFD.init() | net/fd_unix.go |
poll.FD.init() | internal/poll/fd_unix.go |
poll.pollDesc.init() | internal/poll/fd_poll_runtime.go |
runtime.poll_runtime_pollServerInit() | runtime/netpoll.go |
runtime.netpollinit() | runtime/netpoll_epoll.go |
net.netFD.listenStream()
核心步骤如下:
func (fd *netFD) listenStream(ctx context.Context, laddr sockaddr, backlog int, ctrlCtxFn func(context.Context, string, string, syscall.RawConn) error) error {
//....
//通过Bind系统调用绑定监听地址
if err = syscall.Bind(fd.pfd.Sysfd, lsa); err != nil {
return os.NewSyscallError("bind", err)
}
//通过Listen系统调用对socket进行监听
if err = listenFunc(fd.pfd.Sysfd, backlog); err != nil {
return os.NewSyscallError("listen", err)
}
//fd.init()进行了以下操作:
//1、完成eventPool的创建
//2、将socket fd注册到epoll事件表中
if err = fd.init(); err != nil {
return err
}
//...
return nil
}
- 使用
Bind
系统调用绑定需要监听的地址 - 使用
Listen
系统调用监听socket - 调用
fd.init
完成eventpool
的创建以及fd的注册
net.netFD.init()
方法在内部转而调用poll.FD.init()
func (fd *netFD) init() error {
return fd.pfd.Init(fd.net, true)
}
func (fd *FD) Init(net string, pollable bool) error {
fd.SysFile.init()
// We don't actually care about the various network types.
if net == "file" {
fd.isFile = true
}
if !pollable {
fd.isBlocking = 1
return nil
}
err := fd.pd.init(fd)
if err != nil {
// If we could not initialize the runtime poller,
// assume we are using blocking mode.
fd.isBlocking = 1
}
return err
}
然后又转入到poll.pollDesc.init()
的调用中。
func (pd *pollDesc) init(fd *FD) error {
//通过sysOnce结构,完成epoll事件表的唯一一次创建
serverInit.Do(runtime_pollServerInit)
//完成init后,进行poll_open
ctx, errno := runtime_pollOpen(uintptr(fd.Sysfd))
//...
//绑定里层的pollDesc实例
pd.runtimeCtx = ctx
return nil
}
这里的poll.pollDesc
是表层pollDesc
,表层pd的init是poll_init
和poll_open
流程的入口:
- 执行
serverInit.Do(runtime_pollServerInit)
,其中serverInit
是名为sysOnce
的特殊结构,它会保证执行的方法在全局只会被执行一次,然后执行runtime_pollServerInit
,完成poll_init
操作 - 完成
poll_init
后,调用runtime_pollOpen(uintptr(fd.Sysfd))
将fd加入到eventpool
中,完成poll_open
操作 - 绑定里层的
pollDesc
实例
我们先来关注serverInit.Do(runtime_pollServerInit)
中,执行的runtime_pollServerInit
方法,它定位在runtime/netpoll.go
下:
//go:linkname poll_runtime_pollServerInit internal/poll.runtime_pollServerInit
func poll_runtime_pollServerInit() {
netpollGenericInit()
}
func netpollGenericInit() {
if netpollInited.Load() == 0 {
lockInit(&netpollInitLock, lockRankNetpollInit)
lock(&netpollInitLock)
if netpollInited.Load() == 0 {
//进入netpollinit调用
netpollinit()
netpollInited.Store(1)
}
unlock(&netpollInitLock)
}
}
func netpollinit() {
var errno uintptr
//进行epollcreate系统调用,创建epoll事件表
epfd, errno = syscall.EpollCreate1(syscall.EPOLL_CLOEXEC)
//...
//创建pipe管道,接收信号,如程序终止:
//r:信号接收端,会注册对应的read事件到epoll事件表中
//w:信号发送端,有信号到达的时候,会往w发送信号,并对r产生读就绪事件
r, w, errpipe := nonblockingPipe()
//...
//在epollEvent中注册监听r的读就绪事件
ev := syscall.EpollEvent{
Events: syscall.EPOLLIN,
}
*(**uintptr)(unsafe.Pointer(&ev.Data)) = &netpollBreakRd
errno = syscall.EpollCtl(epfd, syscall.EPOLL_CTL_ADD, r, &ev)
//...
//使用全局变量缓存pipe的读写端
netpollBreakRd = uintptr(r)
netpollBreakWr = uintptr(w)
}
在netpollinit()
方法内部,进行了以下操作:
执行
epoll_create
指令创建了epoll事件表,并返回epoll文件描述符epfd
。创建了两个pipe管道,当向w端写入信号的时候,r端会发生读就绪事件。
注册监听r的读就绪事件。
缓存管道。
在这里,我们创建了两个管道r
以及w
,并且在eventpool
中注册了r的读就绪事件的监听,当我们向w管道写入数据的时候,r管道就会产生读就绪事件,从而打破阻塞的epoll_wait操作,进而执行其他的操作。
3.3、poll_open
方法 | 文件 |
---|---|
net.netFD.listenStream() | net/sock_posix.go |
net.netFD.init() | net/fd_unix.go |
poll.FD.init() | internal/poll/fd_unix.go |
poll.pollDesc.init() | internal/poll/fd_poll_runtime.go |
runtime.poll_runtime_pollOpen() | runtime/netpoll.go |
runtime.netpollopen | runtime/netpoll_epoll.go |
在poll.pollDesc.init()
方法中,完成了poll_init
流程后,就会进入到poll_open
流程,执行runtime.poll_runtime_pollOpen()
。
//go:linkname poll_runtime_pollOpen internal/poll.runtime_pollOpen
func poll_runtime_pollOpen(fd uintptr) (*pollDesc, int) {
//获取一个pollDesc实例
pd := pollcache.alloc()
lock(&pd.lock)
wg := pd.wg.Load()
if wg != pdNil && wg != pdReady {
throw("runtime: blocked write on free polldesc")
}
rg := pd.rg.Load()
if rg != pdNil && rg != pdReady {
throw("runtime: blocked read on free polldesc")
}
//绑定socket fd到pollDesc中
pd.fd = fd
//...
//初始化读写状态标识器为无状态
pd.rg.Store(pdNil)
pd.wg.Store(pdNil)
//...
unlock(&pd.lock)
//将fd添加进epoll事件表中
errno := netpollopen(fd, pd)
//...
//返回pollDesc实例
return pd, 0
}
func netpollopen(fd uintptr, pd *pollDesc) uintptr {
var ev syscall.EpollEvent
//通过epollctl操作,在EpollEvent中注册针对fd的监听事件
//操作类型宏指令:EPOLL_CTL_ADD——添加fd并注册监听事件
//事件类型:epollevent.events:
//1、EPOLLIN:监听读就绪事件
//2、EPOLLOUT:监听写就绪事件
//3、EPOLLRDHUP:监听中断事件
//4、EPOLLET:使用边缘触发模式
ev.Events = syscall.EPOLLIN | syscall.EPOLLOUT | syscall.EPOLLRDHUP | syscall.EPOLLET
tp := taggedPointerPack(unsafe.Pointer(pd), pd.fdseq.Load())
*(*taggedPointer)(unsafe.Pointer(&ev.Data)) = tp
return syscall.EpollCtl(epfd, syscall.EPOLL_CTL_ADD, int32(fd), &ev)
}
不仅在net.Listen()
流程中会触发poll open
,在net.Listener.Accept
流程中也会,当我们获取到了连接之后,也需要为这个连接封装成一个pollDesc
实例,然后执行poll_open
流程将其注册到epoll事件表中。
func (fd *netFD) accept()(netfd *netFD, err error){
// 通过 syscall accept 接收到来的 conn fd
d, rsa, errcall, err := fd.pfd.Accept()
// ...
// 封装到来的 conn fd
netfd, err = newFD(d, fd.family, fd.sotype, fd.net)
// 将 conn fd 注册到 epoll 事件表中
err = netfd.init()
// ...
return netfd,nil
}
3.4、poll_close
当连接conn需要关闭的时候,最终会进入到poll_close
流程,执行epoll_ctl(DELETE)
删除对应的fd。
方法 | 文件 |
---|---|
net.conn.Close | net/net.go |
net.netFD.Close | net/fd_posix.go |
poll.FD.Close | internal/poll/fd_unix.go |
poll.FD.decref | internal/poll/fd_mutex.go |
poll.FD.destroy | internal/poll/fd_unix.go |
poll.pollDesc.close | internal/poll/fd_poll_runtime.go |
poll.runtime_pollClose | internal/poll/fd_poll_runtime.go |
runtime.poll_runtime_pollClose | runtime/netpoll.go |
runtime.netpollclose | runtime/netpoll_epoll.go |
syscall.EpollCtl | runtime/netpoll_epoll.go |
//go:linkname poll_runtime_pollClose internal/poll.runtime_pollClose
func poll_runtime_pollClose(pd *pollDesc) {
if !pd.closing {
throw("runtime: close polldesc w/o unblock")
}
wg := pd.wg.Load()
if wg != pdNil && wg != pdReady {
throw("runtime: blocked write on closing polldesc")
}
rg := pd.rg.Load()
if rg != pdNil && rg != pdReady {
throw("runtime: blocked read on closing polldesc")
}
netpollclose(pd.fd)
pollcache.free(pd)
}
func netpollclose(fd uintptr) uintptr {
var ev syscall.EpollEvent
return syscall.EpollCtl(epfd, syscall.EPOLL_CTL_DEL, int32(fd), &ev)
}
3.5、poll_wait
poll_wait
流程最终会执行gopark
将g陷入到用户态阻塞。
方法 | 文件 |
---|---|
poll.pollDesc.wait | internal/poll/fd_poll_runtime.go |
poll.runtime_pollWait | internal/poll/fd_poll_runtime.go |
runtime.poll_runtime_pollWait | runtime/netpoll.go |
runtime.netpollblock | runtime/netpoll.go |
runtime.gopark | runtime/proc.go |
runtime.netpollblockcommit | runtime/netpoll.go |
在表层pollDesc
中,会通过其内部的里层pollDesc
指针,调用到runtime
下的netpollblock
方法。
/*
针对某个 pollDesc 实例,监听指定的mode 就绪事件
- 返回true——已就绪 返回false——因超时或者关闭导致中断
- 其他情况下,会通过 gopark 操作将当前g 阻塞在该方法中
*/
func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {
//针对mode事件,获取相应的状态
gpp := &pd.rg
if mode == 'w' {
gpp = &pd.wg
}
for {
//关心的io事件就绪,直接返回
if gpp.CompareAndSwap(pdReady, pdNil) {
return true
}
//关心的io事件未就绪,则置为等待状态,G将要被阻塞
if gpp.CompareAndSwap(pdNil, pdWait) {
break
}
//...
}
//...
//将G置为阻塞态
gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceBlockNet, 5)
//当前g从阻塞态被唤醒,重置标识器
old := gpp.Swap(pdNil)
if old > pdWait {
throw("runtime: corrupted polldesc")
}
//判断是否是因为所关心的事件触发而唤醒
return old == pdReady
}
在gopark方法中,会闭包调用netpollblockcommit
方法,其中会根据g关心的事件类型,将其实例存储到pollDesc的rg或wg容器
中。
// 将 gpp 状态标识器的值由 pdWait 修改为当前 g
func netpollblockcommit(gp *g, gpp unsafe.Pointer) bool {
r := atomic.Casuintptr((*uintptr)(gpp), pdWait, uintptr(unsafe.Pointer(gp)))
if r {
//增加等待轮询器的例程计数。
//调度器使用它来决定是否阻塞
//如果没有其他事情可做,则等待轮询器。
netpollAdjustWaiters(1)
}
return r
}
接着我们来关注何时会触发poll_wait
流程。
首先是在listener.Accept
流程中,如果当前尚未有连接到达,则执行poll wait
将当前g阻塞挂载在该socket fd对应pollDesc的rg
中。
// Accept wraps the accept network call.
func (fd *FD) Accept() (int, syscall.Sockaddr, string, error) {
//...
for {
//以非阻塞模式发起一次accept,尝试接收conn
s, rsa, errcall, err := accept(fd.Sysfd)
if err == nil {
return s, rsa, "", err
}
switch err {
//忽略中断类错误
case syscall.EINTR:
continue
//尚未有到达的conn
case syscall.EAGAIN:
//进入poll_wait流程,监听fd的读就绪事件,当有conn到达表现为fd可读。
if fd.pd.pollable() {
//假如读操作未就绪,当前g会被阻塞在方法内部,直到因为超时或者就绪被netpoll ready唤醒。
if err = fd.pd.waitRead(fd.isFile); err == nil {
continue
}
}
//...
}
}
// 指定 mode 为 r 标识等待的是读就绪事件,然后走入更底层的 poll_wait 流程
func (pd *pollDesc) waitRead(isFile bool) error {
return pd.wait('r', isFile)
}
其次分别是在conn.Read
/conn.Write
流程中,假若conn fd下读操作未就绪(无数据到达)/写操作未就绪(缓冲区空间不足),则会执行poll wait将g阻塞并挂载在对应的pollDesc中的rg/wg
中。
func (fd *FD) Read(p []byte) (int, error) {
//...
for {
//非阻塞模式进行一次read调用
n, err := ignoringEINTRIO(syscall.Read, fd.Sysfd, p)
if err != nil {
n = 0
//进入poll_wait流程,并标识关心读就绪事件
if err == syscall.EAGAIN && fd.pd.pollable() {
if err = fd.pd.waitRead(fd.isFile); err == nil {
continue
}
}
}
err = fd.eofError(n, err)
return n, err
}
}
func (fd *FD)Write(p []byte)(int,error){
// ...
for{
// ...
// 以非阻塞模式执行一次syscall write操作
n, err := ignoringEINTRIO(syscall.Write, fd.Sysfd, p[nn:max])
if n >0{
nn += n
}
// 缓冲区内容都已写完,直接退出
if nn ==len(p){
return nn, err
}
// 走入 poll_wait 流程,并标识关心的是该 fd 的写就绪事件
if err == syscall.EAGAIN && fd.pd.pollable(){
// 倘若写操作未就绪,当前g 会 park 阻塞在该方法内部,直到因超时或者事件就绪而被 netpoll ready 唤醒
if err = fd.pd.waitWrite(fd.isFile); err ==nil{
continue
}
}
// ...
}
3.6、net_poll
netpoll
流程至关重要,它会在底层调用系统的epoll_wait
操作,找到触发事件的fd,然后再逆向找到绑定fd的pollDesc
实例,返回内部阻塞的g叫给上游处理唤醒。其调用栈如下:
方法 | 文件 |
---|---|
runtime.netpoll | runtime/netpoll_epoll.go |
runtime.netpollready | runtime/netpoll.go |
runtime.netpollunblock | runtime/netpoll.go |
netpoll
具体的源码如下:
//netpoll用于轮询检查是否有就绪的io事件
//若发现了就绪的io事件,检查是否有pollDesc中的g关心其事件
//若找到了关心其io事件就绪的g,添加到list返回给上游处理
func netpoll(delay int64) (gList, int32) {
if epfd == -1 {
return gList{}, 0
}
var waitms int32
//根据传入的delay参数,决定调用epoll_wait的模式:
//delay < 0:设为阻塞模式(在 gmp 调度流程中,如果某个 p 迟迟获取不到可执行的 g 时,会通过该模式,使得 thread 陷入阻塞态,但该情况全局最多仅有一例)
//delay = 0:设为非阻塞模式(通常情况下为此模式,包括 gmp 常规调度流程、gc 以及全局监控线程 sysmon 都是以此模式触发的 netpoll 流程)
//delay > 0:设为超时模式(在 gmp 调度流程中,如果某个 p 迟迟获取不到可执行的 g 时,并且通过 timer 启动了定时任务时,会令 thread 以超时模式执行 epoll_wait 操作)
if delay < 0 {
waitms = -1
} else if delay == 0 {
waitms = 0
} else if delay < 1e6 {
waitms = 1
} else if delay < 1e15 {
waitms = int32(delay / 1e6)
} else {
waitms = 1e9
}
//最多接收128个io就绪事件
var events [128]syscall.EpollEvent
retry:
//以指定模式调用epoll_wait
n, errno := syscall.EpollWait(epfd, events[:], int32(len(events)), waitms)
//...
//存储关心io事件就绪的G实例
var toRun gList
delta := int32(0)
//遍历返回的就绪事件
for i := int32(0); i < n; i++ {
ev := events[i]
if ev.Events == 0 {
continue
}
//pipe接收端的信号处理,检查是否需要退出netpoll
if *(**uintptr)(unsafe.Pointer(&ev.Data)) == &netpollBreakRd {
if ev.Events != syscall.EPOLLIN {
println("runtime: netpoll: break fd ready for", ev.Events)
throw("runtime: netpoll: break fd ready for something unexpected")
}
//...
continue
}
var mode int32
//记录io就绪事件的类型
if ev.Events&(syscall.EPOLLIN|syscall.EPOLLRDHUP|syscall.EPOLLHUP|syscall.EPOLLERR) != 0 {
mode += 'r'
}
if ev.Events&(syscall.EPOLLOUT|syscall.EPOLLHUP|syscall.EPOLLERR) != 0 {
mode += 'w'
}
// 根据 epollevent.data 获取到监听了该事件的 pollDesc 实例
if mode != 0 {
tp := *(*taggedPointer)(unsafe.Pointer(&ev.Data))
pd := (*pollDesc)(tp.pointer())
//...
//检查是否为G所关心的事件
delta += netpollready(&toRun, pd, mode)
}
}
return toRun, delta
}
func netpollready(toRun *gList, pd *pollDesc, mode int32) int32 {
delta := int32(0)
var rg, wg *g
if mode == 'r' || mode == 'r'+'w' {
//就绪事件包含读就绪,尝试唤醒pd内部的rg
rg = netpollunblock(pd, 'r', true, &delta)
}
if mode == 'w' || mode == 'r'+'w' {
//就绪事件包含读就绪,尝试唤醒pd内部的wg
wg = netpollunblock(pd, 'w', true, &delta)
}
//存在G实例,则加入list中
if rg != nil {
toRun.push(rg)
}
if wg != nil {
toRun.push(wg)
}
return delta
}
func netpollunblock(pd *pollDesc, mode int32, ioready bool, delta *int32) *g {
//获取存储的g实例
gpp := &pd.rg
if mode == 'w' {
gpp = &pd.wg
}
for {
old := gpp.Load()
//...
new := pdNil
if ioready {
new = pdReady
}
//将gpp的值从g置换成pdReady
if gpp.CompareAndSwap(old, new) {
if old == pdWait {
old = pdNil
} else if old != pdNil {
*delta -= 1
}
//返回需要唤醒的g实例
return (*g)(unsafe.Pointer(old))
}
}
}
那么,我们也同样需要关注在哪个环节进入了net_poll
流程。
首先,是在GMP调度器中的findRunnable
方法中被调用,用于找到可执行的G实例。具体的实现在之前的GMP调度文章中有讲解,这里只关心涉及到net_poll
方面的源码。
findRunnable
方法定位在runtime/proc.go
中
func findRunnable()(gp *g, inheritTime, tryWakeP bool){
// ..
/*
同时满足下述三个条件,发起一次【非阻塞模式】的 netpoll 流程:
- epoll事件表初始化过
- 有 g 在等待io 就绪事件
- 没有空闲 p 在以【阻塞或超时】模式发起 netpoll 流程
*/
if netpollinited()&& atomic.Load(&netpollWaiters)>0&& atomic.Load64(&sched.lastpoll)!=0{
// 以非阻塞模式发起一轮 netpoll,如果有 g 需要唤醒,一一唤醒之,并返回首个 g 给上层进行调度
if list := netpoll(0);!list.empty(){// non-blocking
// 获取就绪 g 队列中的首个 g
gp := list.pop()
// 将就绪 g 队列中其余 g 一一置为就绪态,并添加到全局队列
injectglist(&list)
// 把首个g 也置为就绪态
casgstatus(gp,_Gwaiting,_Grunnable)
// ...
//返回 g 给当前 p进行调度
return gp,false,false
}
}
// ...
/*
同时满足下述三个条件,发起一次【阻塞或超时模式】的 netpoll 流程:
- epoll事件表初始化过
- 有 g 在等待io 就绪事件
- 没有空闲 p 在以【阻塞或超时】模式发起 netpoll 流程
*/
if netpollinited()&&(atomic.Load(&netpollWaiters)>0|| pollUntil !=0)&& atomic.Xchg64(&sched.lastpoll,0)!=0{
// 默认为阻塞模式
delay :=int64(-1)
// 存在定时时间,则设为超时模式
if pollUntil !=0{
delay = pollUntil - now
// ...
}
// 以【阻塞或超时模式】发起一轮 netpoll
list := netpoll(delay)// block until new work is available
}
// ...
}
其次,是位于同文件下的sysmon
方法中,它会被一个全局监控者G执行,每隔10ms发一次非阻塞的net_poll流程。
// The main goroutine.
func main(){
// ...
// 新建一个 m,直接运行 sysmon 函数
systemstack(func(){
newm(sysmon,nil,-1)
})
// ...
}
// 全局唯一监控线程的执行函数
func sysmon(){
// ...
for{
// ...
/*
同时满足下述三个条件,发起一次【非阻塞模式】的 netpoll 流程:
- epoll事件表初始化过
- 没有空闲 p 在以【阻塞或超时】模式发起 netpoll 流程
- 距离上一次发起 netpoll 流程的时间间隔已超过 10 ms
*/
lastpoll :=int64(atomic.Load64(&sched.lastpoll))
if netpollinited()&& lastpoll !=0&& lastpoll+10*1000*1000< now {
// 以非阻塞模式发起 netpoll
list := netpoll(0)// non-blocking - returns list of goroutines
// 获取到的 g 置为就绪态并添加到全局队列中
if!list.empty(){
// ...
injectglist(&list)
// ...
}
}
// ...
}
}
最后,还会发生在GC流程中。
func pollWork() bool{
// ...
// 若全局队列或 p 的本地队列非空,则提前返回
/*
同时满足下述三个条件,发起一次【非阻塞模式】的 netpoll 流程:
- epoll事件表初始化过
- 有 g 在等待io 就绪事件
- 没有空闲 p 在以【阻塞或超时】模式发起 netpoll 流程
*/
if netpollinited()&& atomic.Load(&netpollWaiters)>0&& sched.lastpoll !=0{
// 所有取得 g 更新为就绪态并添加到全局队列
if list := netpoll(0);!list.empty(){
injectglist(&list)
return true
}
}
// ...
}
4、参考博文
感谢观看,本篇博文参考了小徐先生的文章,非常推荐大家去观看并且进入到源码中学习,链接如下: