性能篇答疑：epoll内核实现剖析
在性能篇中，我们围绕 C10K 问题进行了深入剖析，最终引出了事件分发机制与多线程模型。基于 epoll 的事件分发能力，是 Linux 下高性能网络编程的核心技术。本文通过梳理 epoll 的内核源码实现，从数据结构、核心函数、触发模式等维度深入理解其工作原理。
说明：以下源码分析基于 Linux 内核当前稳定版本的结构。不同内核版本在实现细节上可能存在差异，但核心设计思想保持一致。
基本数据结构
epoll 的内核实现围绕三个核心数据结构展开：eventpoll、epitem 和 eppoll_entry。
eventpoll 结构体
eventpoll 是调用 epoll_create 后内核侧创建的核心数据结构，代表一个 epoll 实例。后续 epoll_ctl 和 epoll_wait 等操作均针对该结构进行。该结构保存在 epoll_create 创建的匿名文件的 private_data 字段中。
/*
* This structure is stored inside the "private_data" member of the file
* structure and represents the main data structure for the eventpoll
* interface.
*/
struct eventpoll {
/* Protect the access to this structure */
spinlock_t lock;
/*
* This mutex is used to ensure that files are not removed
* while epoll is using them. This is held during the event
* collection loop, the file cleanup path, the epoll file exit
* code and the ctl operations.
*/
struct mutex mtx;
/* Wait queue used by sys_epoll_wait() */
// 此队列存放执行 epoll_wait 而等待的进程
wait_queue_head_t wq;
/* Wait queue used by file->poll() */
// 此队列存放 eventpoll 作为 poll 对象实例加入的等待队列
// eventpoll 本身也是一个 file，因此也会有 poll 操作
wait_queue_head_t poll_wait;
/* List of ready file descriptors */
// 事件就绪的 fd 列表，链表元素为 epitem
struct list_head rdllist;
/* RB tree root used to store monitored fd structs */
// 用于快速查找 fd 的红黑树
struct rb_root_cached rbr;
/*
* This is a single linked list that chains all the "struct epitem" that
* happened while transferring ready events to userspace w/out
* holding ->lock.
*/
struct epitem *ovflist;
/* wakeup_source used when ep_scan_ready_list is running */
struct wakeup_source *ws;
/* The user that created the eventpoll descriptor */
struct user_struct *user;
// eventpoll 对应的匿名文件，体现 Linux 一切皆文件的设计哲学
struct file *file;
/* used to optimize loop detection check */
int visited;
struct list_head visited_list_link;
#ifdef CONFIG_NET_RX_BUSY_POLL
/* used to track busy poll napi_id */
unsigned int napi_id;
#endif
};epitem 结构体
每当调用 epoll_ctl 增加一个 fd 时，内核创建一个 epitem 实例，并将其作为红黑树节点插入 eventpoll 的 rbr 字段。此后对该 fd 的事件检测均通过红黑树上的 epitem 进行。
/*
* Each file descriptor added to the eventpoll interface will
* have an entry of this type linked to the "rbr" RB tree.
* Avoid increasing the size of this struct, there can be many thousands
* of these on a server and we do not want this to take another cache line.
*/
struct epitem {
union {
/* RB tree node links this structure to the eventpoll RB tree */
struct rb_node rbn;
/* Used to free the struct epitem */
struct rcu_head rcu;
};
/* List header used to link this structure to the eventpoll ready list */
// 将此 epitem 连接到 eventpoll 的 rdllist
struct list_head rdllink;
/*
* Works together "struct eventpoll"->ovflist in keeping the
* single linked chain of items.
*/
struct epitem *next;
/* The file descriptor information this item refers to */
// epoll 监听的 fd
struct epoll_filefd ffd;
/* Number of active wait queue attached to poll operations */
// 一个文件可被多个 epoll 实例监听，此处记录被监听次数
int nwait;
/* List containing poll wait queues */
struct list_head pwqlist;
/* The "container" of this item */
// 当前 epitem 所属的 eventpoll
struct eventpoll *ep;
/* List header used to link this item to the "struct file" items list */
struct list_head fllink;
/* wakeup_source used when EPOLLWAKEUP is set */
struct wakeup_source __rcu *ws;
/* The structure that describe the interested events and the source fd */
struct epoll_event event;
};eppoll_entry 结构体
每当一个 fd 关联到 epoll 实例时，就会产生一个 eppoll_entry，其核心作用是将 epitem 与目标文件的等待队列关联起来。
/* Wait structure used by the poll hooks */
struct eppoll_entry {
/* List header used to link this structure to the "struct epitem" */
struct list_head llink;
/* The "base" pointer is set to the container "struct epitem" */
struct epitem *base;
/*
* Wait queue item that will be linked to the target file wait
* queue head.
*/
wait_queue_entry_t wait;
/* The wait queue head that linked the "wait" wait queue item */
wait_queue_head_t *whead;
};数据结构关系总览
epoll_create
使用 epoll 时，首先调用 epoll_create（或 epoll_create1）创建 epoll 实例。
内核版本注记：
epoll_create自 Linux 2.5.44 引入，epoll_create1自 Linux 2.6.27 引入。epoll_create1支持EPOLL_CLOEXEC标志。
参数验证
/* Check the EPOLL_* constant for consistency. */
BUILD_BUG_ON(EPOLL_CLOEXEC != O_CLOEXEC);
if (flags & ~EPOLL_CLOEXEC)
return -EINVAL;分配 eventpoll 内存
/* Create the internal data structure ("struct eventpoll"). */
error = ep_alloc(&ep);
if (error < 0)
return error;创建匿名文件与文件描述符
epoll_create 为 epoll 实例分配匿名文件和文件描述符，体现了 UNIX 一切皆文件的设计哲学。eventpoll 实例通过 anon_inode_getfile 保存为匿名文件的 private_data，后续通过 epoll 实例的文件描述符可快速定位 eventpoll 对象。
/*
* Creates all the items needed to setup an eventpoll file. That is,
* a file structure and a free file descriptor.
*/
fd = get_unused_fd_flags(O_RDWR | (flags & O_CLOEXEC));
if (fd < 0) {
error = fd;
goto out_free_ep;
}
file = anon_inode_getfile("[eventpoll]", &eventpoll_fops, ep,
O_RDWR | (flags & O_CLOEXEC));
if (IS_ERR(file)) {
error = PTR_ERR(file);
goto out_free_fd;
}
ep->file = file;
fd_install(fd, file);
return fd;epoll_ctl
epoll_ctl 负责将套接字添加到 epoll 实例中，或修改/删除已注册的套接字。
查找 epoll 实例
首先通过 epoll 实例句柄获取对应的匿名文件：
// 获取 epoll 实例对应的匿名文件
f = fdget(epfd);
if (!f.file)
goto error_return;获取目标套接字对应的文件（tf = target file）：
/* Get the "struct file *" for the target file */
// 获取真正的文件，如监听套接字、读写套接字
tf = fdget(fd);
if (!tf.file)
goto error_fput;进行参数合法性验证，确保 epfd 确实是 epoll 实例句柄：
/* The target file descriptor must support poll */
// 不支持 poll 的文件描述符无效
error = -EPERM;
if (!tf.file->f_op->poll)
goto error_tgt_fput;
...通过 private_data 获取 eventpoll 实例：
/*
* At this point it is safe to assume that the "private_data" contains
* our own data structure.
*/
ep = f.file->private_data;红黑树查找
epoll_ctl 通过目标文件和描述符在红黑树中查找是否已存在该套接字。红黑树（RB-tree）是 epoll 高效的关键数据结构，eventpoll 通过红黑树跟踪所有监听的文件描述符，树的根保存在 eventpoll 的 rbr 字段中。
/* RB tree root used to store monitored fd structs */
struct rb_root_cached rbr;查找操作：
/*
* Try to lookup the file inside our RB tree, Since we grabbed "mtx"
* above, we can be sure to be able to use the item looked up by
* ep_find() till we release the mutex.
*/
epi = ep_find(ep, tf.file, fd);红黑树节点的排序能力由 epoll_filefd 结构体提供：
struct epoll_filefd {
struct file *file; // pointer to the target file struct corresponding to the fd
int fd; // target file descriptor number
} __packed;
/* Compare RB tree keys */
static inline int ep_cmp_ffd(struct epoll_filefd *p1,
struct epoll_filefd *p2)
{
return (p1->file > p2->file ? +1:
(p1->file < p2->file ? -1 : p1->fd - p2->fd));
}排序规则：先按文件地址大小排序，若相同则按文件描述符排序。
ep_insert
若为 ADD 操作且红黑树中不存在对应节点，则调用 ep_insert 插入新节点：
case EPOLL_CTL_ADD:
if (!epi) {
epds.events |= POLLERR | POLLHUP;
error = ep_insert(ep, &epds, tf.file, fd, full_check);
} else
error = -EEXIST;
if (full_check)
clear_tfile_check_list();
break;ep_insert 首先检查当前监控的文件数是否超过 /proc/sys/fs/epoll/max_user_watches 的预设最大值：
user_watches = atomic_long_read(&ep->user->epoll_watches);
if (unlikely(user_watches >= max_user_watches))
return -ENOSPC;分配资源并初始化：
if (!(epi = kmem_cache_alloc(epi_cache, GFP_KERNEL)))
return -ENOMEM;
/* Item initialization follow here ... */
INIT_LIST_HEAD(&epi->rdllink);
INIT_LIST_HEAD(&epi->fllink);
INIT_LIST_HEAD(&epi->pwqlist);
epi->ep = ep;
ep_set_ffd(&epi->ffd, tfile, fd);
epi->event = *event;
epi->nwait = 0;
epi->next = EP_UNACTIVE_PTR;回调函数注册
ep_insert 为每个文件描述符设置回调函数，这是 epoll 事件通知机制的核心。回调函数通过 ep_ptable_queue_proc 设置，当文件描述符上有事件发生时（如套接字缓冲区有数据到达），内核调用 ep_poll_callback。内核设计同样遵循事件回调原理。
/*
* This is the callback that is used to add our wait queue to the
* target file wakeup lists.
*/
static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead, poll_table *pt)
{
struct epitem *epi = ep_item_from_epqueue(pt);
struct eppoll_entry *pwq;
if (epi->nwait >= 0 && (pwq = kmem_cache_alloc(pwq_cache, GFP_KERNEL))) {
init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);
pwq->whead = whead;
pwq->base = epi;
if (epi->event.events & EPOLLEXCLUSIVE)
add_wait_queue_exclusive(whead, &pwq->wait);
else
add_wait_queue(whead, &pwq->wait);
list_add_tail(&pwq->llink, &epi->pwqlist);
epi->nwait++;
} else {
/* We have to signal that an error occurred */
epi->nwait = -1;
}
}内核版本注记：
EPOLLEXCLUSIVE标志自 Linux 4.5 引入，用于避免多个 epoll 实例监听同一 fd 时的惊群效应（thundering herd problem）。
ep_poll_callback
ep_poll_callback 将内核事件与 epoll 对象关联起来，其工作流程如下：
- 通过
wait_queue_entry_t对象找到对应的epitem对象（ep_item_from_wait函数完成地址计算） - 获取
epitem后，进一步定位eventpoll实例
static int ep_poll_callback(wait_queue_entry_t *wait, unsigned mode, int sync, void *key)
{
int pwake = 0;
unsigned long flags;
struct epitem *epi = ep_item_from_wait(wait);
struct eventpoll *ep = epi->ep;加锁后进行事件过滤——为性能考虑，ep_insert 向监控文件注册的是所有事件，而用户侧可能仅订阅了部分事件：
if (key && !((unsigned long) key & epi->event.events))
goto out_unlock;判断是否需要将事件传递给用户空间，若需要且该事件不在已完成队列中，则将其加入：
/* If this file is already in the ready list we exit soon */
if (!ep_is_linked(&epi->rdllink)) {
list_add_tail(&epi->rdllink, &ep->rdllist);
ep_pm_stay_awake_rcu(epi);
}唤醒因 epoll_wait 而休眠的进程：
if (waitqueue_active(&ep->wq)) {
if ((epi->event.events & EPOLLEXCLUSIVE) &&
!((unsigned long)key & POLLFREE)) {
switch ((unsigned long)key & EPOLLINOUT_BITS) {
case POLLIN:
if (epi->event.events & POLLIN)
ewake = 1;
break;
case POLLOUT:
if (epi->event.events & POLLOUT)
ewake = 1;
break;
case 0:
ewake = 1;
break;
}
}
wake_up_locked(&ep->wq);
}epoll_wait
参数验证
/* The maximum number of event must be greater than zero */
if (maxevents <= 0 || maxevents > EP_MAX_EVENTS)
return -EINVAL;
/* Verify that the area passed by the user is writeable */
if (!access_ok(VERIFY_WRITE, events, maxevents * sizeof(struct epoll_event)))
return -EFAULT;查找 eventpoll 实例
与 epoll_ctl 类似，通过 epoll 实例找到对应的匿名文件和 eventpoll 实例：
/* Get the "struct file *" for the eventpoll file */
f = fdget(epfd);
if (!f.file)
return -EBADF;
error = -EINVAL;
if (!is_file_epoll(f.file))
goto error_fput;
ep = f.file->private_data;调用 ep_poll 完成事件收集并传递到用户空间：
/* Time to fish for events ... */
error = ep_poll(ep, events, maxevents, timeout);ep_poll
ep_poll 根据 timeout 值进行不同处理：大于 0 设置超时时间，等于 0 立即检查是否有事件发生。
static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
int maxevents, long timeout)
{
int res = 0, eavail, timed_out = 0;
unsigned long flags;
u64 slack = 0;
wait_queue_entry_t wait;
ktime_t expires, *to = NULL;
if (timeout > 0) {
struct timespec64 end_time = ep_set_mstimeout(timeout);
slack = select_estimate_accuracy(&end_time);
to = &expires;
*to = timespec64_to_ktime(end_time);
} else if (timeout == 0) {
timed_out = 1;
spin_lock_irqsave(&ep->lock, flags);
goto check_events;
}若无事件发生，将当前进程加入 eventpoll 的等待队列 wq，以便 ep_poll_callback 唤醒：
if (!ep_events_available(ep)) {
ep_reset_busy_poll_napi_id(ep);
init_waitqueue_entry(&wait, current);
__add_wait_queue_exclusive(&ep->wq, &wait);进入无限循环，通过 schedule_hrtimeout_range 将当前进程休眠。唤醒条件有四种：
- 当前进程超时
- 当前进程收到 signal 信号
- 某个描述符上有事件发生
- CPU 重新调度，重新判断后若无前三个条件则再次休眠
条件 1、2、3 通过 break 跳出循环。
for (;;) {
set_current_state(TASK_INTERRUPTIBLE);
if (fatal_signal_pending(current)) {
res = -EINTR;
break;
}
if (ep_events_available(ep) || timed_out)
break;
if (signal_pending(current)) {
res = -EINTR;
break;
}
spin_unlock_irqrestore(&ep->lock, flags);
if (!schedule_hrtimeout_range(to, slack, HRTIMER_MODE_ABS))
timed_out = 1;
spin_lock_irqsave(&ep->lock, flags);
}进程从休眠返回后，从等待队列中删除并恢复 TASK_RUNNING 状态：
__remove_wait_queue(&ep->wq, &wait);
__set_current_state(TASK_RUNNING);最后调用 ep_send_events 将事件拷贝到用户空间：
if (!res && eavail &&
!(res = ep_send_events(ep, events, maxevents)) && !timed_out)
goto fetch_events;
return res;ep_send_events
ep_send_events 将 ep_send_events_proc 作为回调函数调用 ep_scan_ready_list，后者对每个就绪事件循环处理。
ep_send_events_proc 处理就绪事件时，会再次调用每个文件描述符的 poll 方法确认事件确实有效。尽管如此，仍存在极小概率：当 ep_send_events_proc 再次调用 poll 函数后，用户空间获得的事件通知可能已不再有效（可能已被用户空间处理或其他原因）。这正是将非阻塞套接字配合 epoll 使用作为最佳实践的原因——若套接字为阻塞模式，read 调用可能阻塞整个进程。
确认事件有效后，通过 __put_user 将事件结构体拷贝到用户空间：
// 对 fd 再次进行 poll 操作以确认事件
revents = ep_item_poll(epi, &pt);
if (revents) {
if (__put_user(revents, &uevent->events) ||
__put_user(epi->event.data, &uevent->data)) {
list_add(&epi->rdllink, head);
ep_pm_stay_awake(epi);
return eventcnt ? eventcnt : -EFAULT;
}
eventcnt++;
uevent++;Level-Triggered VS Edge-Triggered
从实现角度看，条件触发（Level-Triggered，LT）与边缘触发（Edge-Triggered，ET）的区别非常简洁。
在 ep_send_events_proc 函数末尾，针对 LT 模式，当前 epitem 被重新加入 eventpoll 的就绪列表，这样下一次 epoll_wait 调用时该 epitem 会被重新处理：
// Level-Triggered 处理：事件被重新加回 ready list
// 下一轮 epoll_wait 时会重新检查
else if (!(epi->event.events & EPOLLET)) {
list_add_tail(&epi->rdllink, &ep->rdllist);
ep_pm_stay_awake(epi);
}由于在拷贝到用户空间之前会再次调用 poll 方法确认事件是否仍然有效，因此：
- 若用户空间已处理该事件，不会再次通知
- 若用户空间未处理（事件仍然有效），会再次通知
epoll VS poll/select
从实现角度分析 epoll 效率远高于 poll/select 的原因。
原因一：避免大量内存分配与拷贝
poll/select 每次调用需将监听的 fd 集合从用户空间拷贝到内核空间，处理后再拷贝回用户空间，涉及内核空间的内存申请与释放。在大量 fd 场景下，该开销极为显著。
epoll 通过红黑树持久维护 fd 集合，避免了每次调用的内存申请和释放，且查找速度为 O(log n)。
以下是 select 在内核空间申请内存的实现，先尝试栈上分配，fd 较多时转为堆分配：
int core_sys_select(int n, fd_set __user *inp, fd_set __user *outp,
fd_set __user *exp, struct timespec64 *end_time)
{
fd_set_bits fds;
void *bits;
int ret, max_fds;
size_t size, alloc_size;
struct fdtable *fdt;
/* Allocate small arguments on the stack to save memory and be faster */
long stack_fds[SELECT_STACK_ALLOC/sizeof(long)];
ret = -EINVAL;
if (n < 0)
goto out_nofds;
rcu_read_lock();
fdt = files_fdtable(current->files);
max_fds = fdt->max_fds;
rcu_read_unlock();
if (n > max_fds)
n = max_fds;
size = FDS_BYTES(n);
bits = stack_fds;
if (size > sizeof(stack_fds) / 6) {
/* Not enough space in on-stack array; must use kmalloc */
ret = -ENOMEM;
if (size > (SIZE_MAX / 6))
goto out_nofds;
alloc_size = 6 * size;
bits = kvmalloc(alloc_size, GFP_KERNEL);
if (!bits)
goto out_nofds;
}
fds.in = bits;
fds.out = bits + size;
fds.ex = bits + 2*size;
fds.res_in = bits + 3*size;
fds.res_out = bits + 4*size;
fds.res_ex = bits + 5*size;
...原因二：避免遍历全部 fd
select/poll 从休眠中被唤醒时，若监听多个 fd，只要其中有一个 fd 有事件发生，内核就需遍历内部 list 检查具体是哪个事件到达。epoll 则通过 fd 直接关联 eventpoll 对象，将 fd 快速加入就绪列表，无需遍历。
static int do_select(int n, fd_set_bits *fds, struct timespec64 *end_time)
{
...
retval = 0;
for (;;) {
unsigned long *rinp, *routp, *rexp, *inp, *outp, *exp;
bool can_busy_loop = false;
inp = fds->in; outp = fds->out; exp = fds->ex;
rinp = fds->res_in; routp = fds->res_out; rexp = fds->res_ex;
for (i = 0; i < n; ++rinp, ++routp, ++rexp) {
unsigned long in, out, ex, all_bits, bit = 1, mask, j;
unsigned long res_in = 0, res_out = 0, res_ex = 0;
in = *inp++; out = *outp++; ex = *exp++;
all_bits = in | out | ex;
if (all_bits == 0) {
i += BITS_PER_LONG;
continue;
}
if (!poll_schedule_timeout(&table, TASK_INTERRUPTIBLE,
to, slack))
timed_out = 1;
...总结
本文通过深度分析 epoll 的内核源码实现，揭示了其高效工作的原理：
-
红黑树持久维护 fd 集合：减少了内核与用户空间大量的数据拷贝和内存分配，查找效率为 O(log n)。
-
就绪链表记录活跃事件：内核在每个文件有事件发生时将自身登记到就绪列表，通过文件与 eventpoll 之间的回调和唤醒机制，避免了对全部描述符的遍历，大幅加速了事件通知和检测效率。
-
LT/ET 双模式支持：LT 模式通过将未处理完的 epitem 重新加入就绪链表实现持续通知；ET 模式仅通知一次，效率更高但需更谨慎的编程。
-
对比 poll/select：epoll 从内存管理和事件检测两个维度克服了 poll/select 的固有弊端，是 Linux 下高性能网络编程的核心基础设施。
版本信息
- 更新日期：2026-06-09
- 目标内核：Linux 7.0
- 关键 API 版本：epoll 自 Linux 2.5.44 引入；
epoll_create1自 Linux 2.6.27 引入；EPOLLEXCLUSIVE自 Linux 4.5 引入；EPOLLRDHUP自 Linux 2.6.17 引入；CONFIG_NET_RX_BUSY_POLL自 Linux 3.11 引入