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.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()中心过程如下：

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

在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。

//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堕入到用户态堵塞。

在表层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.