golang netpoller揭秘

golang netpoller是网络IO模型的核心部分，利用了操作系统提供的事件通知机制，如Linux的epoll、BSD的kqueue或者windows的IOCP。这些机制允许应用程序监视多个文件描述符（在网络编程中，通常是 socket），并在其中任何一个准备好进行 I/O 操作时接收通知。

数据结构

netFD

网络连接都是基于对netFD结构的操作

// 网络文件描述符
type netFD struct {pfd poll.FD// immutable until Close// 网络协议族。比如AF_INET表示ipv4，AF_INET6表示ipv6family      int// socket类型sotype      int// 握手是否完成isConnected bool // handshake completed or use of association with peer// 网络类型。比如tcp、ipnet         string// 储存网络连接的本地地址laddr       Addr// 储存网络连接的远程地址raddr       Addr
}// 文件描述符。可表示网络连接或者系统文件
type FD struct {// 用于锁定文件描述符并序列化 Read 和 Write 方法的使用。fdmu fdMutex// 系统文件描述符。这个字段的值在 Close 方法调用之前是不变的。Sysfd int// 文件描述符的平台相关状态。SysFile// I/O 轮询器pd pollDesc// 文件关闭时发出的信号量csema uint32// 非0代表处于堵塞模式isBlocking uint32// 是否是一个流式描述符，而不是一个基于数据包的描述符，如 UDP 套接字。这个字段的值是不变的。IsStream bool// 零字节读取是否表示 EOF。对于基于消息的套接字连接，这个字段的值是 false。ZeroReadIsEOF bool// 表示这是否是一个文件，而不是一个网络套接字。isFile bool
}

pollDesc

pollDesc为底层轮询器封装

type pollDesc struct {runtimeCtx uintptr
}type pollDesc struct {_     sys.NotInHeap// 下一个pollDesclink  *pollDesc      // in pollcache, protected by pollcache.lock// 文件描述符fd    uintptr        // constant for pollDesc usage lifetime// 保护pollDesc不受过时的的影响fdseq atomic.Uintptr // protects against stale pollDesc// 保存从 closing、rd 和 wd 中获取的位，这些位只在持有锁的情况下写入，以供 netpollcheckerr 使用，netpollcheckerr 不能获取锁。在可能改变摘要的方式下锁定这些字段后，代码必须在释放锁之前调用 publishInfo。atomicInfo atomic.Uint32 // atomic pollInfo// 读取和写入的G 指针，它们是原子访问的rg atomic.Uintptr // pdReady, pdWait, G waiting for read or pdNilwg atomic.Uintptr // pdReady, pdWait, G waiting for write or pdNillock    mutex // protects the following fields// 表示是否正在关闭closing bool// 用户可设置的cookieuser    uint32    // 保护读写计时器不受过时的影响rseq    uintptr   // protects from stale read timerswseq    uintptr   // protects from stale write timers// 读截止timerrt      timer     // read deadline timer (set if rt.f != nil)// 读截止时间rd      int64     // read deadline (a nanotime in the future, -1 when expired)// 写截止timerwt      timer     // write deadline timer// 写截止时间wd      int64     // write deadline (a nanotime in the future, -1 when expired)// 指向pollDesc自身的指针，用于间接接口的储存self    *pollDesc
}

pollCache

pollCache用于缓存pollDesc结构，可以避免在每次网络IO操作时都创建新的pollDesc

type pollCache struct {// 互斥锁，用于在多个 goroutine 之间同步对 pollCache 的访问lock  mutex// 指向 pollDesc 的指针，表示缓存中的第一个 pollDesc 结构。如果缓存为空，那么这个字段的值就是nilfirst *pollDesc
}

PollDesc 对象必须是类型稳定的，因为在描述符关闭或重用后，我们可能会从 epoll/kqueue 中获取到就绪通知。这是因为在网络编程中，文件描述符（File Descriptor，简称 FD）是一个重要的资源，它在关闭后可能会被立即重用。如果一个 FD 关闭后，另一个 FD 立即重用了这个数字，那么原来的 FD 上的任何未决的事件可能会错误地通知到新的 FD 上。

使用 seq 变量来检测过时的通知。在改变截止日期或者描述符被重用时，seq 会增加。这是为了防止因为 FD 的重用导致的错误通知。通过在每次改变 FD 的状态时增加 seq，可以确保即使 FD 被重用，也能正确地识别出哪些通知是过时的

监听

net.Listen

调用net.Listen之后会通过系统调用socket方法创建网络socket分配给listener

// socket returns a network file descriptor that is ready for
// asynchronous I/O using the network poller.
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) {// 创建系统sockets, err := sysSocket(family, sotype, proto)if err != nil {return nil, err}// 设置默认的socket选项if err = setDefaultSockopts(s, family, sotype, ipv6only); err != nil {poll.CloseFunc(s)return nil, err}// 创建新的网络文件描述符if fd, err = newFD(s, family, sotype, net); err != nil {poll.CloseFunc(s)return nil, err}// 若本地地址存在，远程地址不存在，则为监听。根据socket类型，调用网络文件描述符方法进行监听if laddr != nil && raddr == nil {switch sotype {case syscall.SOCK_STREAM, syscall.SOCK_SEQPACKET:if err := fd.listenStream(ctx, laddr, listenerBacklog(), ctrlCtxFn); err != nil {fd.Close()return nil, err}return fd, nilcase syscall.SOCK_DGRAM:if err := fd.listenDatagram(ctx, laddr, ctrlCtxFn); err != nil {fd.Close()return nil, err}return fd, nil}}// 否则，为dial请求if err := fd.dial(ctx, laddr, raddr, ctrlCtxFn); err != nil {fd.Close()return nil, err}return fd, nil
}

netFD.listenStream

netFD.listenStream创建socket，并绑定监听地址进行监听

func (fd *netFD) listenStream(ctx context.Context, laddr sockaddr, backlog int, ctrlCtxFn func(context.Context, string, string, syscall.RawConn) error) error {// 设置默认的监听socket选项var err errorif err = setDefaultListenerSockopts(fd.pfd.Sysfd); err != nil {return err}// 获取socket地址var lsa syscall.Sockaddrif lsa, err = laddr.sockaddr(fd.family); err != nil {return err}// 执行控制操作if ctrlCtxFn != nil {c := newRawConn(fd)if err := ctrlCtxFn(ctx, fd.ctrlNetwork(), laddr.String(), c); err != nil {return err}}// 将socket绑定到指定地址if err = syscall.Bind(fd.pfd.Sysfd, lsa); err != nil {return os.NewSyscallError("bind", err)}// 开始监听if err = listenFunc(fd.pfd.Sysfd, backlog); err != nil {return os.NewSyscallError("listen", err)}// 网络描述符初始化，注册到netpoller中if err = fd.init(); err != nil {return err}// 获取socket本地网络地址，并设置到fd本地地址中lsa, _ = syscall.Getsockname(fd.pfd.Sysfd)fd.setAddr(fd.addrFunc()(lsa), nil)return nil
}

poll_runtime_pollOpen

在网络描述符初始化中，会调用poll_runtime_pollOpen打开文件描述符，并注册到netpoller中

func poll_runtime_pollOpen(fd uintptr) (*pollDesc, int) {// 从缓存中分配到pollDescpd := pollcache.alloc()lock(&pd.lock)// 初始化分配到的pollDescwg := 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")}pd.fd = fdif pd.fdseq.Load() == 0 {// The value 0 is special in setEventErr, so don't use it.pd.fdseq.Store(1)}pd.closing = falsepd.setEventErr(false, 0)pd.rseq++pd.rg.Store(pdNil)pd.rd = 0pd.wseq++pd.wg.Store(pdNil)pd.wd = 0pd.self = pd// 更新pollDesc的原子信息pd.publishInfo()unlock(&pd.lock)// 将文件描述符注册到netpollererrno := netpollopen(fd, pd)if errno != 0 {pollcache.free(pd)return nil, int(errno)}return pd, 0
}

netpollopen将pollDesc注册到Linux epoll中

func netpollopen(fd uintptr, pd *pollDesc) uintptr {var ev syscall.EpollEvent// 设置事件为输入、输出、连接被对方关闭或者半关闭事件以及设置边缘触发模式ev.Events = syscall.EPOLLIN | syscall.EPOLLOUT | syscall.EPOLLRDHUP | syscall.EPOLLET// 将 pd 和 pd.fdseq.Load() 打包成一个标记指针 tp。这个标记指针被存储在 ev.Data 中tp := taggedPointerPack(unsafe.Pointer(pd), pd.fdseq.Load())*(*taggedPointer)(unsafe.Pointer(&ev.Data)) = tp// 将 fd 添加到 epoll 的实例中，并返回 syscall.EpollCtl 的结果。syscall.EpollCtl 是一个系统调用，用于控制 epoll 的行为。在这里，它的行为是添加一个新的文件描述符到 epoll 的实例中return syscall.EpollCtl(epfd, syscall.EPOLL_CTL_ADD, int32(fd), &ev)
}

接收

listener.Accept

listener.Accept实质上就是调用其内部的netFD的accept方法，接受网络连接，并以建立的新连接创建网络文件描述符

func (fd *netFD) accept() (netfd *netFD, err error) {// 接受新的网络连接，并返回新的系统文件描述符d和远程socket地址 rsad, rsa, errcall, err := fd.pfd.Accept()if err != nil {if errcall != "" {err = wrapSyscallError(errcall, err)}return nil, err}// 根据建立连接的系统文件描述符，创建新的网络文件描述符if netfd, err = newFD(d, fd.family, fd.sotype, fd.net); err != nil {poll.CloseFunc(d)return nil, err}// 初始化新网络文件描述符if err = netfd.init(); err != nil {netfd.Close()return nil, err}// 获取新网络文件描述符的本地socket地址，并设置到fd本地地址中lsa, _ := syscall.Getsockname(netfd.pfd.Sysfd)netfd.setAddr(netfd.addrFunc()(lsa), netfd.addrFunc()(rsa))return netfd, nil
}

fd.Accept

fd.Accept与下面fd.Read大致流程一致，只是循环调用的是accept

func (fd *FD) Accept() (int, syscall.Sockaddr, string, error) {if err := fd.readLock(); err != nil {return -1, nil, "", err}defer fd.readUnlock()if err := fd.pd.prepareRead(fd.isFile); err != nil {return -1, nil, "", err}for {s, rsa, errcall, err := accept(fd.Sysfd)if err == nil {return s, rsa, "", err}switch err {case syscall.EINTR:continuecase syscall.EAGAIN:if fd.pd.pollable() {if err = fd.pd.waitRead(fd.isFile); err == nil {continue}}case syscall.ECONNABORTED:// This means that a socket on the listen// queue was closed before we Accept()ed it;// it's a silly error, so try again.continue}return -1, nil, errcall, err}
}

读取

fd.Read

主要就是加锁，尝试从文件描述符中读取数据，若读取不成功，就进行读取等待

func (fd *FD) Read(p []byte) (int, error) {// 获取读锁if err := fd.readLock(); err != nil {return 0, err}defer fd.readUnlock()// 若读取字节长度为0，直接返回if len(p) == 0 {return 0, nil}// 准备读取if err := fd.pd.prepareRead(fd.isFile); err != nil {return 0, err}// 若fd是流，且读取字节长度大于1<<30，则截断if fd.IsStream && len(p) > maxRW {p = p[:maxRW]}for {// 系统调用从文件描述符中读取数据n, err := ignoringEINTRIO(syscall.Read, fd.Sysfd, p)if err != nil {n = 0// 若读未就绪时，等待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}
}

在读取信息未到达时，waitRead方法会调用到runtime_pollWait方法，而runtime_pollWait又会调用netpollblock方法

netpollblock

netpollblock用于堵塞当前goroutine，等待网络IO事件的发生

// 对于同一个模式，不允许并发调用netpollblock，因为 pollDesc 只能为每种模式持有一个等待的 goroutine
func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {// 根据读写模式，获取状态gpp := &pd.rgif mode == 'w' {gpp = &pd.wg}// set the gpp semaphore to pdWaitfor {// 若期待的IO事件准备好了，则解除堵塞if gpp.CompareAndSwap(pdReady, pdNil) {return true}// 若没有期待的IO事件发生，则设置为等待，退出循环if gpp.CompareAndSwap(pdNil, pdWait) {break}// 防止出现意外状态，导致无限循环if v := gpp.Load(); v != pdReady && v != pdNil {throw("runtime: double wait")}}// 若设置了忽略错误等待或者无错误，则堵塞// 在将 gpp 设置为 pdWait 后需要重新检查错误状态。因为 runtime_pollUnblock、runtime_pollSetDeadline 和 deadlineimpl 这几个函数的操作顺序与此相反：它们先将状态存储到 closing/rd/wd，然后发布信息，最后加载 rg/wg。所以，为了保证状态的正确性，需要在设置 gpp 为 pdWait 后重新检查错误状态if waitio || netpollcheckerr(pd, mode) == pollNoError {gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceBlockNet, 5)}// 堵塞结束后，设置pdNil状态old := gpp.Swap(pdNil)if old > pdWait {throw("runtime: corrupted polldesc")}return old == pdReady
}

netpoll

netpoll检测所有就绪的网络连接，并返回所有可允许的goroutine。因此，通过该方法可以唤醒在网络读取、写入等待中的goroutine

func netpoll(delay int64) (gList, int32) {if epfd == -1 {return gList{}, 0}// 根据输入的延迟时间，计算epoll等待时间// 等待时间delay < 0，永远堵塞下去// 等待时间delay = 0，非堵塞，仅轮询// 等待时间delay > 0，堵塞若干时间var waitms int32if 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 {// An arbitrary cap on how long to wait for a timer.// 1e9 ms == ~11.5 days.waitms = 1e9}// 调用epollWait，等待epoll事件var events [128]syscall.EpollEvent
retry:n, errno := syscall.EpollWait(epfd, events[:], int32(len(events)), waitms)if errno != 0 {if errno != _EINTR {println("runtime: epollwait on fd", epfd, "failed with", errno)throw("runtime: netpoll failed")}// If a timed sleep was interrupted, just return to// recalculate how long we should sleep now.if waitms > 0 {return gList{}, 0}goto retry}// 遍历事件，获取需要恢复的goroutinevar toRun gListdelta := int32(0)for i := int32(0); i < n; i++ {ev := events[i]if ev.Events == 0 {continue}// 如果设置了提前唤醒，则会跳出当前轮询。因为break事件只是表示有其他事情需要处理，并不是真正的网络事件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")}// 如果 delay 不为 0，这表示这个 "break" 事件是在一个阻塞的轮询中被检测到的。在这种情况下，代码会从 netpollBreakRd 读取数据，并将 netpollWakeSig 设置为 0。这样可以确保下一次 "break" 事件能够被正确处理。if delay != 0 {var tmp [16]byteread(int32(netpollBreakRd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))netpollWakeSig.Store(0)}continue}// 判断事件发生的类型，以决定从pollDesc中wg还是rg取goroutinevar mode int32if 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'}if mode != 0 {// 取出保存在事件数据中的pollDes，并添加到就绪goroutine链表tp := *(*taggedPointer)(unsafe.Pointer(&ev.Data))pd := (*pollDesc)(tp.pointer())tag := tp.tag()if pd.fdseq.Load() == tag {pd.setEventErr(ev.Events == syscall.EPOLLERR, tag)delta += netpollready(&toRun, pd, mode)}}}return toRun, delta
}