I/O多路复用select/poll/epoll

常见I/O模型

堵塞I/O

执行read或write等系统调用时，应用程序会被堵塞

应用程序由用户态陷入内核态，内核检查文件描述符是否可读，存在数据时，OS内核将准备好的数据拷贝给应用程序并交回控制权

一旦执行I/O操作，应用程序会陷入堵塞状态等待I/O操作的结束

非堵塞I/O

进程把一个文件描述符设置为非堵塞时，执行read或write等I/O操作会立刻返回

第一次从文件描述符读取数据会触发系统调用并返回EAGAIN错误，意味着文件描述符还在等待缓冲区中的数据，应用程序轮询read直到返回值大于0

信号驱动I/O

工作原理

允许程序在I/O操作准备就绪时通过操作系统的信号通知应用程序。程序首先注册一个信号处理程序，并将文件描述符设置为信号驱动模式。当文件描述符准备好进行I/O操作（如可读、可写或发生错误）时，操作系统会向程序发送一个信号（通常是SIGIO），通知程序执行相应的I/O操作

特点

• 异步通知：信号驱动I/O提供了一种异步通知机制，程序不需要阻塞等待I/O操作，可以继续执行其他任务，直到收到信号。
• 复杂性高：信号驱动I/O的编程模型较为复杂，涉及信号处理程序的注册和处理。此外，信号处理程序的执行是异步的，可能导致调试和错误处理的难度增加。
• 实时性：适用于实时性要求高的场景，但需要开发者对信号处理有深入理解。

异步I/O

工作原理

异步I/O（Asynchronous I/O，AIO）允许程序发起I/O操作后立即返回，程序无需等待I/O操作完成。当I/O操作完成时，操作系统会通知程序（通过回调函数、信号或其他机制），并由程序处理结果。异步I/O完全由操作系统管理，程序不需要手动轮询或等待I/O事件。

特点

• 完全异步：异步I/O是真正的异步操作，程序发起I/O请求后可以立即继续处理其他任务，当I/O完成时再处理结果。
• 高效处理并发I/O：由于异步I/O不阻塞程序的执行，它非常适合处理大量并发I/O操作，特别是在需要处理大量I/O密集型任务时表现优异。
• 编程模型复杂：虽然异步I/O的使用使得程序更加高效，但它的编程模型较为复杂，尤其是在处理多个并发异步操作时，代码的可读性和可维护性可能会受到影响。

I/O多路复用

在操作系统中处理多个I/O操作的技术，允许一个进程在同时等待多个文件描述符的I/O操作准备就绪时高效地进行管理，而不是为每个I/O操作分别堵塞和等待，特别适合网络服务器等需要处理大量并发连接的场景

在其中任意一个文件描述符就绪时，通知进程来处理

I/O多路复用

select

工作原理

使用fd_set位图结构来管理文件描述符集合，程序需要手动将需要监控的文件描述符添加到fd_set中，并调用select函数来等待这些文件描述符的状态变化

特点

• 最早的I/O多路复用机制之一，几乎所有Unix/Linux都可用
• 允许程序通过多个文件描述符来监听I/O事件，并在描述符里任意一个变成可读可写错误时返回
• 需要监视大量文件描述符时效率低，且受限于文件描述符的最大数量
• select每次调用时都要线性扫描遍历所有文件描述符，即使只有少数文件描述符发生了变化

poll

工作原理

• 使用pollfd结构数组来管理文件描述符及其对应事件类型
• 使用动态数组，消除了文件描述符数量的限制

特点

• 无文件描述符数量的限制
• 灵活性，允许程序指定每个文件描述符的事件类型，并返回一个包含所有发生了状态变化的文件描述符及其事件列表
• 性能问题，在处理大量文件描述符时，poll仍然需要遍历整个文件描述符数组
• 每次调用poll都需要重新构建pollfd结构数组，消耗了额外的系统资源

epoll

基本概念

• 通过事件驱动的方式管理文件描述符
• 没有固定上限，可以根据系统资源监控大量的文件描述符

事件驱动模型Event Driven

• 水平触发（Level-Triggered, LT）
  • epoll的默认工作模式，只要文件描述符仍然满足条件，epoll_wait 每次调用都会返回事件
• 边缘触发（Edge-Triggered, ET）
  • epoll_wait 只会在文件描述符状态发生变化时返回一次事件，要求应用程序必须读取所有数据以避免丢失通知，适合高性能应用

核心函数

epoll_create1

int epoll_create1(int flags);

epoll_ctl

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

• epfd: 由 epoll_create1 返回的 epoll 文件描述符。
• op: 操作类型，可以是 EPOLL_CTL_ADD（添加）、EPOLL_CTL_MOD（修改）、EPOLL_CTL_DEL（删除）。
• fd: 要监控的文件描述符。
• event: 指向 epoll_event 结构体的指针，指定要监控的事件类型（如 EPOLLIN 可读、EPOLLOUT 可写等）。

epoll_wait

用于等待 epoll 实例中的事件，返回已准备好进行 I/O 操作的文件描述符集合

int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

• epfd: epoll 文件描述符。
• events: 用于接收事件的数组。
• maxevents: 数组的大小，即本次最多可以接收的事件数量。
• timeout: 超时时间，单位为毫秒，-1 表示无限期等待。

工作流程

• 创建epoll实例
  • epoll_create1
• 注册文件描述符
  • 使用 epoll_ctl 将需要监控的文件描述符添加到 epoll 实例中，并指定要监控的事件类型（如可读、可写、异常等）
• 等待事件
  • 使用 epoll_wait 等待文件描述符上的事件，epoll 只会返回那些状态发生变化的文件描述符
• 处理事件
  • 遍历 epoll_wait 返回的事件数组，对每个发生事件的文件描述符执行相应的 I/O 操作
• （可选）修改或删除文件描述符
  • 根据需要，可以使用 epoll_ctl 修改或删除已经注册的文件描述符

示例代码

#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#define MAX_EVENTS 10

int main() {
    int epfd, nfds;
    struct epoll_event ev, events[MAX_EVENTS];
    int listen_sock, conn_sock;

    // 假设listen_sock是一个已绑定并监听的套接字

    epfd = epoll_create1(0);
    if (epfd == -1) {
        perror("epoll_create1");
        exit(EXIT_FAILURE);
    }

    ev.events = EPOLLIN;
    ev.data.fd = listen_sock;
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, listen_sock, &ev) == -1) {
        perror("epoll_ctl: listen_sock");
        exit(EXIT_FAILURE);
    }

    for (;;) {
        nfds = epoll_wait(epfd, events, MAX_EVENTS, -1);
        if (nfds == -1) {
            perror("epoll_wait");
            exit(EXIT_FAILURE);
        }

        for (int n = 0; n < nfds; ++n) {
            if (events[n].data.fd == listen_sock) {
                conn_sock = accept(listen_sock, NULL, NULL);
                if (conn_sock == -1) {
                    perror("accept");
                    exit(EXIT_FAILURE);
                }

                ev.events = EPOLLIN;
                ev.data.fd = conn_sock;
                if (epoll_ctl(epfd, EPOLL_CTL_ADD, conn_sock, &ev) == -1) {
                    perror("epoll_ctl: conn_sock");
                    exit(EXIT_FAILURE);
                }
            } else {
                // 处理客户端的读写
                char buffer[512];
                ssize_t count = read(events[n].data.fd, buffer, sizeof(buffer));
                if (count == -1) {
                    perror("read");
                    exit(EXIT_FAILURE);
                } else if (count == 0) {
                    // 对端关闭连接
                    close(events[n].data.fd);
                } else {
                    // 正常读取数据
                    printf("Received: %s", buffer);
                }
            }
        }
    }

    close(epfd);
    return 0;
}

优缺点

优点

• 高效处理大规模连接: epoll 特别适合处理大量并发连接，性能远优于 select 和 poll，尤其是在大量空闲连接的情况下。
• O(1)复杂度: epoll 的事件通知机制使得它能够高效处理文件描述符，即使是监控成千上万个文件描述符，也能保持很高的性能。
• 边缘触发模式: epoll 提供的边缘触发模式可以进一步提高性能，适合高吞吐量场景。

缺点

• 仅限于Linux: epoll 是Linux特有的I/O多路复用机制，无法跨平台使用。
• 编程复杂度: epoll 的使用相对复杂，特别是在处理边缘触发模式时，开发者需要更加小心地处理可能的数据丢失问题。

go代码分析

环境

• go：1.21.0

相关特点

编译器会根据平台选择树中特定分支进行编译，如果是linux，则是使用epoll

相关源码

基本函数

func netpollinit()
func netpollopen(fd uintptr, pd *pollDesc) int32
func netpoll(delta int64) gList
func netpollBreak()
func netpollIsPollDescriptor(fd uintptr) bool

初始化

func netpollinit() {
	var errno uintptr
	epfd, errno = syscall.EpollCreate1(syscall.EPOLL_CLOEXEC)
	if errno != 0 {
		println("runtime: epollcreate failed with", errno)
		throw("runtime: netpollinit failed")
	}
	r, w, errpipe := nonblockingPipe()
	if errpipe != 0 {
		println("runtime: pipe failed with", -errpipe)
		throw("runtime: pipe failed")
	}
	ev := syscall.EpollEvent{
		Events: syscall.EPOLLIN,
	}
	*(**uintptr)(unsafe.Pointer(&ev.Data)) = &netpollBreakRd
	errno = syscall.EpollCtl(epfd, syscall.EPOLL_CTL_ADD, r, &ev)
	if errno != 0 {
		println("runtime: epollctl failed with", errno)
		throw("runtime: epollctl failed")
	}
	netpollBreakRd = uintptr(r)
	netpollBreakWr = uintptr(w)
}

这个流程用到了上述提到的系统调用

• 执行syscall.EpollCreate1创建epoll实例
• 创建非阻塞管道，返回两个描述符读r和写w
• 将读端文件描述符r注册到epoll
  • 创建一个EpollEvent结构体实例ev，设置Events字段为EPOLLIN，表示希望监听r上的可读事件
  • ev.Data字段被设置为指向netpollBreakRd的指针，这样在事件发生时，可以通过这个字段识别对应的事件
  • 使用syscall.EpollCtl将r文件描述符注册到epoll实例中，并监控它的可读事件。如果注册失败，会输出错误信息并终止程序
• 保存管道的读写端文件描述符，以便在后续的网络事件处理中使用

监听边缘触发事件

func netpollopen(fd uintptr, pd *pollDesc) uintptr {
	var ev syscall.EpollEvent
	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)
}

将文件描述符fd注册到epoll实例的一个函数 netpollopen

• 入参
  • fd文件描述符
  • pd指向 pollDesc 结构的指针，用于描述与文件描述符相关的事件和状态
• 初始化EpollEvent结构
  • EPOLLIN：监控文件描述符上的可读事件。
  • EPOLLOUT：监控文件描述符上的可写事件。
  • EPOLLRDHUP：监控远端关闭连接的事件，常用于检测对方关闭TCP连接的情况。
  • EPOLLET：设置为边缘触发模式（Edge Triggered），提高性能，减少重复通知。
• 设置事件数据
  • taggedPointerPack 是一个将指针和标识信息打包成一个值的函数，在这里，它将 pd 指针和文件描述符的序列号打包成一个 taggedPointer 类型的值
  • ev.Data 字段被设置为 tp，这个字段将在 epoll_wait 返回事件时提供给程序，以便识别哪个 pollDesc 结构与该事件关联
• 注册文件描述符到epoll

轮询网络并返回一组已经准备就绪的 goroutine

// netpoll checks for ready network connections.
// Returns list of goroutines that become runnable.
// delay < 0: blocks indefinitely
// delay == 0: does not block, just polls
// delay > 0: block for up to that many nanoseconds
func netpoll(delay int64) gList {
	if epfd == -1 {
		return gList{}
	}
	var waitms int32
	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 {
		// An arbitrary cap on how long to wait for a timer.
		// 1e9 ms == ~11.5 days.
		waitms = 1e9
	}
	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{}
		}
		goto retry
	}
	var toRun gList
	for i := int32(0); i < n; i++ {
		ev := events[i]
		if ev.Events == 0 {
			continue
		}

		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")
			}
			if delay != 0 {
				// netpollBreak could be picked up by a
				// nonblocking poll. Only read the byte
				// if blocking.
				var tmp [16]byte
				read(int32(netpollBreakRd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))
				netpollWakeSig.Store(0)
			}
			continue
		}

		var mode int32
		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'
		}
		if mode != 0 {
			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)
				netpollready(&toRun, pd, mode)
			}
		}
	}
	return toRun
}

它用于处理网络轮询，检查哪些文件描述符已准备好进行I/O操作，并将相应的goroutine标记为可运行（runnable）

• 入参delay，根据具体值的范围有修正
  • < 0，函数无限期阻塞，直到至少一个事件准备好
  • = 0，不阻塞，只进行一次轮询
  • > 0，阻塞指定时间，单位纳秒
• 创建一个syscall.EpollEvent数组，接收epoll_wait事件
  • 阻塞指定时间等待事件的发生，填充数组events并返回事件数目n
• 遍历所有发生事件的描述符events
  • 事件为0，则跳过
  • 事件是netpollBreakRd（用于中断epoll_wait），并且事件类型不是EPOLLIN，则抛出异常
  • 如果是阻塞调用，读取中断信号，并清除netpollWakeSig标志
• 处理网络时间并标记可运行
  • 根据事件类型设置mode，表示文件描述符上发生的事件类型：
    • r（可读）：EPOLLIN、EPOLLRDHUP、EPOLLHUP、EPOLLERR。
    • w（可写）： EPOLLOUT、EPOLLHUP、EPOLLERR。
  • 从ev.Data中获取与事件关联的pollDesc结构，通过检查pollDesc的文件描述符序列号来验证事件是否有效
  • 调用netpollready函数将准备好的pollDesc添加到toRun列表中，表示这些goroutine可以运行

唤醒网络轮询器

// netpollBreak interrupts an epollwait.
func netpollBreak() {
	// Failing to cas indicates there is an in-flight wakeup, so we're done here.
	if !netpollWakeSig.CompareAndSwap(0, 1) {
		return
	}

	for {
		var b byte
		n := write(netpollBreakWr, unsafe.Pointer(&b), 1)
		if n == 1 {
			break
		}
		if n == -_EINTR {
			continue
		}
		if n == -_EAGAIN {
			return
		}
		println("runtime: netpollBreak write failed with", -n)
		throw("runtime: netpollBreak write failed")
	}
}

用于中断正在进行的 epoll_wait 调用。这在需要唤醒正在阻塞等待 I/O 事件的 Go runtime 时非常有用，通常用于处理定时事件、关闭操作或其他需要立即停止阻塞等待的情况

• 检查是否需要中断
  • 使用了原子操作CAS来检查并设置一个标志，表示是否需要进行唤醒操作
  • 如果 netpollWakeSig 当前值为 0，则将其置为 1，并继续执行唤醒操作
  • 如果 netpollWakeSig 已经被置为 1（表示唤醒操作已经在进行中），那么这个 CAS 操作将失败，函数直接返回，不再进行任何操作。这避免了重复的唤醒操作
• 写入管道
  • 写入成功 (n == 1)：表示字节成功写入管道，epoll_wait 会因为该管道的可读事件被唤醒。成功写入后，直接跳出循环，结束函数
  • 写入失败 (n == -_EINTR)：如果写操作被系统信号中断 (_EINTR)，则重新尝试写入
  • 写入失败 (n == -_EAGAIN)：如果管道已经满了（表示唤醒请求已经发送），则直接返回，不再继续写入。_EAGAIN 表示资源暂时不可用
  • 其他错误：如果出现其他写入错误，程序输出错误信息并抛出异常

判断文件描述符是否被轮询器使用

func netpollIsPollDescriptor(fd uintptr) bool {
	return fd == uintptr(epfd) || fd == netpollBreakRd || fd == netpollBreakWr
}

赏

谢谢你请我吃糖果