实现 epoll 事件模型，完成事件驱动并发

整体定位

前面我们的 Socket 模型是 单线程全局阻塞等待队列模型：一次只能阻塞等待一个 fd，无法同时监听成千上万个连接，只能串行处理。

epoll 是 Linux 内核主流 IO 多路复用模型，核心目标：

把阻塞等待从单个 fd 的等待队列，迁移到全局事件轮询器，一次等待、监听全部 fd 可读/可写/错误事件，实现单线程高并发事件驱动模型

深入操作系统视角：epoll 本质是内核态的事件通知机制，通过回调驱动而非轮询，将 O(n) 的遍历开销降为 O(1) 的就绪事件获取。本文将从内核数据结构、回调注册、就绪队列唤醒等底层机制，完整复刻 epoll 事件模型。

本文分为：epoll 内核原理、核心数据结构、三大核心 API 实现、水平触发实现、Socket 层回调联动、并发测试、避坑指南。

一、IO多路复用背景 & epoll基础原理

1.1 旧模型痛点

• 原始模型：单个 accept/recv 阻塞等待单个 fd，无法同时监听大量连接
• select/poll 缺点：每次传入全部 fd、全量遍历检查就绪状态，海量 fd 时性能很差（O(n)）
• epoll 核心优势：事件回调 + 就绪列表（O(1) 取出就绪 fd）
  • 内核维护红黑树保存监听 fd
  • 内核在数据就绪时主动把 fd 加入就绪链表，不需要遍历全部 fd
  • 两种触发模式：水平触发 LT（默认）、边缘触发 ET（EPOLLET）

✅ 水平触发 LT（优先复刻学习）

只要 fd 还有未读数据 / 可写缓冲区有空，就持续反复触发可读/可写事件

• 优点：逻辑简单、不易漏读，适合学习原型
• 缺点：可能反复触发多余事件

❗ 边缘触发 ET

仅在状态发生首次变化瞬间触发一次事件（例如数据刚到达那一刻）

• 要求：必须一次性读完所有缓冲区数据，否则不会再次触发事件，极易造成数据卡死、连接僵死
• 学习版优先做 LT 水平触发，后续再研究 ET

1.2 epoll 三件核心系统调用

1. epoll_create(size)：创建 epoll 实例，返回 epoll_fd
2. epoll_ctl(epoll_fd, op, fd, event)：添加/修改/删除监听 fd 和关注事件
   • op：EPOLL_CTL_ADD / EPOLL_CTL_MOD / EPOLL_CTL_DEL
   • 事件类型：EPOLLIN（可读）、EPOLLOUT（可写）、EPOLLERR（错误）、EPOLLHUP（挂起/关闭）
3. epoll_wait(epoll_fd, events, maxevents, timeout)：阻塞等待就绪事件，返回就绪 fd 列表

1.3 前置改造前提

1. Socket 必须支持非阻塞模式（O_NONBLOCK）
   • 非阻塞 recv/send：无数据/缓冲区满时立刻返回 EAGAIN，不阻塞主线程
   • 主线程只阻塞在 epoll_wait，不再阻塞在单个 fd 的 recv/accept
2. 原有单 fd 等待队列转为：仅底层内核/软中断使用，主线程统一走 epoll 事件循环
3. 引用计数 + 锁模型已经就绪：防止事件回调时 fd 被提前销毁（野指针）
4. recv_exact/send_all 适配非阻塞 EAGAIN 重试逻辑

---

1.4 epoll 内核实现深度解析

1.4.1 内核数据结构设计

// Linux 内核实际 epoll 数据结构（简化版）
struct eventpoll {
    spinlock_t lock;                    // 自旋锁保护就绪队列
    struct mutex mtx;                   // 互斥锁保护监控树
    wait_queue_head_t wq;               // 等待队列（epoll_wait 阻塞点）
    struct list_head rdllist;           // 就绪文件描述符链表（双向链表）
    struct rb_root rbr;                 // 红黑树根节点（监控的 fd）
    struct epitem *ovflist;             // 单链表存储就绪事件（避免锁竞争）
    struct wakeup_source *ws;           // 唤醒源（防止睡眠丢失）
    struct user_struct *user;           // 用户引用计数
};

// epoll 条目（红黑树节点）
struct epitem {
    struct rb_node rbn;                 // 红黑树节点
    struct list_head rdllink;           // 就绪链表节点
    struct epoll_filefd ffd;            // 文件描述符 + 文件指针
    struct eventpoll *ep;               // 所属 eventpoll
    struct epoll_event event;           // 用户关注的事件
    struct list_head pwqlist;           // poll 等待队列链表
    struct file *file;                  // 关联的文件对象
};

// 就绪事件包装
struct epoll_event {
    __u32 events;                       // 事件掩码
    __u64 data;                         // 用户数据（联合体）
};

1.4.2 回调注册机制（底层原理）

1. 文件操作回调表：每个文件描述符对应一个 file_operations 结构，包含 poll 方法
2. 回调链注册：epoll_ctl(ADD) 时，内核调用 ep_item_poll() → vfs_poll() → file->f_op->poll()
3. 等待队列注册：将当前 epoll 实例的等待队列项注册到文件描述符的等待队列头
4. 事件触发路径：
   • 数据到达网卡 → 硬中断 → 软中断（NET_RX_SOFTIRQ）
   • 协议栈处理 → sock_def_readable() → wake_up_interruptible_sync_poll()
   • 唤醒 epoll 等待队列 → 将 epitem 加入就绪链表

1.4.3 水平触发 vs 边缘触发内核实现差异

// 水平触发（LT）实现逻辑
static int ep_send_events_proc(struct eventpoll *ep, 
                               struct list_head *head, void *priv) {
    // 检查事件是否仍然有效
    if (epi->event.events & EPOLLONESHOT)
        epi->event.events &= EP_PRIVATE_BITS;
    
    // LT模式：只要条件满足，持续报告事件
    if (!(epi->event.events & EPOLLET)) {
        // 重新加入就绪列表，等待下次检查
        if (ep_item_poll(epi, &pt, 1) != 0) {
            if (!ep_is_linked(&epi->rdllink))
                list_add_tail(&epi->rdllink, &ep->rdllist);
        }
    }
    return 0;
}

// 边缘触发（ET）实现逻辑
static inline int ep_is_et(struct epitem *epi) {
    return epi->event.events & EPOLLET;
}

// ET模式：只在状态变化时报告一次
// 内核通过 ep_poll_callback() 中的 edge_triggered 标志控制

二、复刻 epoll 核心数据结构

2.1 epoll 全局容器

// ==================== epoll 全局容器 ====================
// epoll 监听事件类型
#define EPOLLIN    0x001   // 可读事件
#define EPOLLOUT   0x004   // 可写事件
#define EPOLLERR   0x008   // 错误事件
#define EPOLLHUP   0x010   // 连接挂起/关闭
#define EPOLLET    0x80000000  // 边缘触发模式

// epoll 事件结构体
typedef struct epoll_event {
    uint32_t events;    // 关注的事件类型
    void* data;         // 用户数据指针
} epoll_event_t;

// epoll 实例结构体
typedef struct epoll_instance {
    int epoll_fd;                       // epoll 文件描述符
    pthread_mutex_t lock;               // 互斥锁
    struct rb_root monitor_tree;        // 红黑树：fd -> 关注事件
    struct list_head ready_list;        // 就绪事件链表
    struct wait_queue_head wait_queue;  // 等待队列
} epoll_instance_t;

// 全局 epoll 实例表
static struct rb_root epoll_table = RB_ROOT;
static pthread_mutex_t epoll_table_lock = PTHREAD_MUTEX_INITIALIZER;
static int next_epoll_fd = 100;

// ==================== 核心回调函数 ====================
// 添加就绪事件（由底层Socket/传输层回调触发）
static void epoll_add_ready_event(epoll_instance_t* ep, int fd, uint32_t events) {
    pthread_mutex_lock(&ep->lock);
    
    // 去重：LT模式防止重复加入
    struct epoll_ready_event *event;
    list_for_each_entry(event, &ep->ready_list, list) {
        if (event->fd == fd) {
            event->events |= events;
            goto wakeup;
        }
    }
    
    // 创建新就绪事件
    event = kmalloc(sizeof(*event), GFP_KERNEL);
    event->fd = fd;
    event->events = events;
    list_add_tail(&event->list, &ep->ready_list);
    
wakeup:
    // 唤醒 epoll_wait 的阻塞线程
    wake_up(&ep->wait_queue);
    pthread_mutex_unlock(&ep->lock);
}

// 等待就绪事件
static int epoll_wait_internal(epoll_instance_t* ep, 
                              struct epoll_event* events, 
                              int maxevents, 
                              int timeout_ms) {
    int count = 0;
    
    // 若无就绪事件，阻塞等待
    if (list_empty(&ep->ready_list)) {
        wait_event_interruptible_timeout(ep->wait_queue, 
                                        !list_empty(&ep->ready_list), 
                                        msecs_to_jiffies(timeout_ms));
    }
    
    pthread_mutex_lock(&ep->lock);
    
    // 返回就绪事件
    struct epoll_ready_event *event, *tmp;
    list_for_each_entry_safe(event, tmp, &ep->ready_list, list) {
        if (count >= maxevents) break;
        
        events[count].events = event->events;
        events[count].data.fd = event->fd;
        list_del(&event->list);
        kfree(event);
        count++;
    }
    
    pthread_mutex_unlock(&ep->lock);
    return count;
}

---

📊 核心设计要点

设计要点	说明
事件驱动机制	事件不是 epoll 主动轮询，而是底层 Socket/传输层收到数据后主动回调 add_ready_event()，把 fd 加入就绪队列，唤醒 epoll_wait
内核唤醒路径	数据到达 → 协议栈处理 → sock_def_readable() → 唤醒 epoll 等待队列 → 将 epitem 加入就绪链表
性能优势	就绪事件 O(1) 获取，监控 fd 数量不影响性能，红黑树查找 O(log n)
内存布局	每个 epoll 实例约 1KB + 每个监控 fd 约 200 字节，万级连接内存开销约 2MB
锁优化策略	就绪队列使用自旋锁，监控树使用互斥锁，避免锁竞争

2.2 非阻塞模式改造

新增 fcntl_set_nonblock(fd, flag=True) API，修改 SocketSk 标记：

// Socket结构体扩展非阻塞标志
struct socket_sk {
    // ...原有字段
    unsigned int non_block:1;  // 是否非阻塞模式
    unsigned int epoll_registered:1;  // 是否已注册到epoll
    struct list_head epoll_list;      // 注册的epoll实例链表
};

// 设置非阻塞模式
int fcntl_set_nonblock(int fd, int flag) {
    struct socket_sk *sk = sockfd_lookup(fd);
    if (!sk) return -EBADF;
    
    sk->non_block = flag ? 1 : 0;
    
    // 更新底层文件标志
    struct file *filp = fget(fd);
    if (filp) {
        if (flag) {
            filp->f_flags |= O_NONBLOCK;
        } else {
            filp->f_flags &= ~O_NONBLOCK;
        }
        fput(filp);
    }
    
    sockfd_put(sk);
    return 0;
}

• 非阻塞模式下：原生 recv/send 立刻返回，无数据/缓冲区满返回 EAGAIN，不再调用本 Socket 的 wait_queue.block()
• 主线程阻塞点从单个 fd wait_queue → 全局 epoll wait_queue

三、三大 epoll API 实现

3.1 epoll_create

// ==================== epoll_create 实现 ====================
int epoll_create(int size) {
    (void)size; // size参数在Linux中已废弃，仅用于兼容
    
    epoll_instance_t *ep = kmalloc(sizeof(*ep), GFP_KERNEL);
    if (!ep) return -ENOMEM;
    
    // 初始化epoll实例
    ep->epoll_fd = atomic_inc_return(&next_epoll_fd);
    pthread_mutex_init(&ep->lock, NULL);
    ep->monitor_tree = RB_ROOT;
    INIT_LIST_HEAD(&ep->ready_list);
    init_waitqueue_head(&ep->wait_queue);
    
    // 添加到全局表
    pthread_mutex_lock(&epoll_table_lock);
    struct epoll_table_entry *entry = kmalloc(sizeof(*entry), GFP_KERNEL);
    entry->fd = ep->epoll_fd;
    entry->ep = ep;
    rb_insert(&epoll_table, entry);
    pthread_mutex_unlock(&epoll_table_lock);
    
    return ep->epoll_fd;
}

3.2 epoll_ctl 监听管理

// ==================== epoll_ctl 监听管理 ====================
// epoll_ctl 操作类型
#define EPOLL_CTL_ADD 1
#define EPOLL_CTL_MOD 2
#define EPOLL_CTL_DEL 3

// epoll_ctl 实现
int epoll_ctl(int epoll_fd, int op, int fd, struct epoll_event *event) {
    // 查找epoll实例
    pthread_mutex_lock(&epoll_table_lock);
    struct epoll_table_entry *entry = rb_find(&epoll_table, epoll_fd);
    if (!entry) {
        pthread_mutex_unlock(&epoll_table_lock);
        return -EBADF;
    }
    epoll_instance_t *ep = entry->ep;
    pthread_mutex_unlock(&epoll_table_lock);
    
    // 查找socket
    struct socket_sk *sk = sockfd_lookup(fd);
    if (!sk) return -EBADF;
    
    pthread_mutex_lock(&ep->lock);
    
    switch (op) {
    case EPOLL_CTL_ADD:
    case EPOLL_CTL_MOD:
        // 添加到监控树
        struct epoll_monitor_entry *monitor;
        monitor = rb_find(&ep->monitor_tree, fd);
        if (!monitor) {
            monitor = kmalloc(sizeof(*monitor), GFP_KERNEL);
            monitor->fd = fd;
            rb_insert(&ep->monitor_tree, monitor);
        }
        monitor->events = event->events;
        monitor->data = event->data;
        
        // 给Socket注册回调
        if (!sk->epoll_registered) {
            sk->epoll_registered = 1;
            list_add(&sk->epoll_list, &ep->epoll_sockets);
            register_socket_callback(sk, ep);
        }
        break;
        
    case EPOLL_CTL_DEL:
        // 从监控树移除
        struct epoll_monitor_entry *monitor = rb_find(&ep->monitor_tree, fd);
        if (monitor) {
            rb_erase(&ep->monitor_tree, monitor);
            kfree(monitor);
        }
        
        // 从socket的epoll列表移除
        if (sk->epoll_registered) {
            list_del(&sk->epoll_list);
            sk->epoll_registered = 0;
        }
        break;
        
    default:
        pthread_mutex_unlock(&ep->lock);
        sockfd_put(sk);
        return -EINVAL;
    }
    
    pthread_mutex_unlock(&ep->lock);
    sockfd_put(sk);
    return 0;
}

• 核心回调绑定：当 Socket rx_buf 有数据 / tx_buf 可写 / 发生错误时，调用 ep.add_ready_event(fd, mask)
  • TCP：收到报文、全连接队列新增连接、FIN关闭、RST异常
  • UDP：收到新UDP数据报
  • 监听 fd：accept 就绪

3.3 epoll_wait 阻塞等待就绪事件

// epoll_wait 实现
int epoll_wait(int epoll_fd, struct epoll_event *events, 
               int maxevents, int timeout_ms) {
    // 查找epoll实例
    pthread_mutex_lock(&epoll_table_lock);
    struct epoll_table_entry *entry = rb_find(&epoll_table, epoll_fd);
    if (!entry) {
        pthread_mutex_unlock(&epoll_table_lock);
        return -EBADF;
    }
    epoll_instance_t *ep = entry->ep;
    pthread_mutex_unlock(&epoll_table_lock);
    
    // 参数检查
    if (maxevents <= 0) return -EINVAL;
    if (timeout_ms < -1) return -EINVAL;
    
    // 转换为阻塞等待（timeout_ms=-1表示无限等待）
    int actual_timeout = (timeout_ms == -1) ? MAX_SCHEDULE_TIMEOUT : 
                          msecs_to_jiffies(timeout_ms);
    
    return epoll_wait_internal(ep, events, maxevents, actual_timeout);
}

四、Socket层底层事件回调联动（核心联动逻辑）

4.1 注册事件回调

给每个被 epoll 监听的 Socket 绑定回调函数：

1. 可读事件回调 EPOLLIN
   • 监听Socket（listen fd）：accept_queue 出现新连接
   • 普通TCP/UDP Socket：rx_buf 收到新数据
   • 挂起事件：连接关闭、EOF
2. 可写事件回调 EPOLLOUT
   • Socket 发送缓冲区有空余位置，可以继续发送数据
3. 错误事件回调 EPOLLERR/EPOLLHUP
   • TCP RST异常、连接异常断开、错误状态

示例回调伪代码：

// C语言实现的事件回调注册
static void register_socket_callback(struct socket_sk *sk, epoll_instance_t *ep) {
    // 注册可读事件回调
    sk->rx_callback = (socket_callback_t)epoll_rx_callback;
    sk->rx_callback_data = ep;
    
    // 注册可写事件回调  
    sk->tx_callback = (socket_callback_t)epoll_tx_callback;
    sk->tx_callback_data = ep;
    
    // 注册错误事件回调
    sk->err_callback = (socket_callback_t)epoll_err_callback;
    sk->err_callback_data = ep;
}

// 可读事件回调函数
static void epoll_rx_callback(struct socket_sk *sk, void *data) {
    epoll_instance_t *ep = (epoll_instance_t *)data;
    uint32_t events = EPOLLIN;
    
    // 检查是否为监听socket（有新连接）
    if (sk->state == SOCK_LISTEN && !list_empty(&sk->accept_queue)) {
        events |= EPOLLIN;
    }
    // 检查普通socket是否有数据
    else if (sk->rx_buf_len > 0) {
        events |= EPOLLIN;
    }
    
    if (events) {
        epoll_add_ready_event(ep, sk->fd, events);
    }
}

// 可写事件回调函数
static void epoll_tx_callback(struct socket_sk *sk, void *data) {
    epoll_instance_t *ep = (epoll_instance_t *)data;
    
    // 发送缓冲区有空闲空间
    if (sk->tx_buf_free > 0) {
        epoll_add_ready_event(ep, sk->fd, EPOLLOUT);
    }
}

// 底层传输层触发回调的时机
// 1. 网卡中断处理完成后，调用 netif_receive_skb()
// 2. 协议栈处理：ip_local_deliver() → tcp_v4_rcv()/udp_rcv()
// 3. Socket层：sock_def_readable() → wake_up_interruptible_sync_poll()
// 4. 最终调用注册的 epoll 回调函数

• 当底层传输层填充 rx_buf / accept_queue 时自动调用回调，唤醒 epoll_wait

4.2 非阻塞事件处理循环（主线程核心主循环）

// C语言实现的事件循环
void event_loop(int epoll_fd) {
    struct epoll_event events[MAX_EVENTS];
    struct socket_sk *sk;
    int nfds, i;
    
    while (1) {
        // 阻塞等待就绪事件（内核态阻塞）
        nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
        if (nfds < 0) {
            if (errno == EINTR) continue; // 被信号中断
            perror("epoll_wait");
            break;
        }
        
        // 处理所有就绪事件
        for (i = 0; i < nfds; i++) {
            int fd = events[i].data.fd;
            uint32_t mask = events[i].events;
            
            // 获取socket结构（带引用计数保护）
            sk = sockfd_lookup(fd);
            if (!sk) {
                // fd已关闭，从epoll移除
                epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
                continue;
            }
            
            // 处理可读事件
            if (mask & EPOLLIN) {
                if (sk->state == SOCK_LISTEN) {
                    // 监听socket：接受新连接
                    int new_fd = tcp_accept(fd);
                    if (new_fd > 0) {
                        // 设置非阻塞
                        fcntl_set_nonblock(new_fd, 1);
                        
                        // 加入epoll监听
                        struct epoll_event ev;
                        ev.events = EPOLLIN | EPOLLOUT | EPOLLET;
                        ev.data.fd = new_fd;
                        epoll_ctl(epoll_fd, EPOLL_CTL_ADD, new_fd, &ev);
                        
                        printf("Accepted new connection: fd=%d\n", new_fd);
                    }
                } else {
                    // 普通socket：读取数据
                    char buffer[4096];
                    ssize_t n = recv_nonblock(fd, buffer, sizeof(buffer));
                    
                    if (n > 0) {
                        // 处理数据（示例：回声）
                        printf("Received %zd bytes from fd=%d\n", n, fd);
                        send_nonblock(fd, buffer, n);
                    } else if (n == 0) {
                        // 对端关闭连接
                        printf("Connection closed by peer: fd=%d\n", fd);
                        epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
                        close(fd);
                        sockfd_put(sk);
                        continue;
                    } else if (n == -EAGAIN) {
                        // 非阻塞无数据，等待下次事件
                    } else {
                        // 错误处理
                        perror("recv");
                    }
                }
            }
            
            // 处理可写事件
            if (mask & EPOLLOUT) {
                // 发送缓冲区可写，继续发送待发送数据
                if (sk->tx_pending_len > 0) {
                    ssize_t sent = send_pending_data(sk);
                    if (sent < 0 && errno != EAGAIN) {
                        // 发送失败，关闭连接
                        epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
                        close(fd);
                    }
                }
            }
            
            // 处理错误/挂起事件
            if (mask & (EPOLLERR | EPOLLHUP)) {
                printf("Error/HUP on fd=%d, closing\n", fd);
                epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
                close(fd);
            }
            
            // 释放socket引用
            sockfd_put(sk);
        }
    }
}

4.3 非阻塞 recv / send 适配 & recv_exact 改造

• 增加 EAGAIN 错误判断：非阻塞读无数据时返回 EAGAIN，直接跳出本轮处理，等待下次 epoll 可读事件
• 非阻塞 recv_exact：记录半包缓存状态，下次可读事件触发时继续读取剩余字节，不可在单次回调里无限阻塞循环读取
• 关闭时：先调用 epoll_ctl(DEL) 移除监听，再执行 close / refcount 释放，防止 epoll 持续监听无效 fd

五、TCP监听fd + UDP fd 接入epoll完整流程

5.1 TCP服务端接入

1. 创建 listen_fd、bind、listen、设置为非阻塞
2. epoll_ctl(ADD, listen_fd, EPOLLIN) 加入 epoll 监听
3. epoll 触发 listen_fd EPOLLIN → 调用 accept 获取 new_fd
4. new_fd 设置非阻塞，加入 epoll 监听 EPOLLIN/EPOLLOUT
5. 后续由 epoll 循环处理 new_fd 的读写事件

5.2 UDP服务端接入

1. 创建 udp_fd、bind、设置为非阻塞
2. epoll_ctl(ADD, udp_fd, EPOLLIN) 加入 epoll 监听
3. epoll 触发 EPOLLIN → 调用 recvfrom 读取完整UDP报文、回发 sendto
4. UDP无连接模型：无需 accept，直接处理报文并记录远端地址

六、关键坑点 & 并发安全要点

6.1 边缘触发ET坑（初学尽量不用）

• ET模式只触发一次事件，必须一次性读完整个缓冲区所有数据（循环 recv 直到返回 EAGAIN）
• 否则缓冲区残留数据，后续不再触发可读事件 → 连接卡死僵死
• LT水平触发：每次有剩余数据都会持续触发可读事件，原型首选

6.2 fd生命周期 & 引用计数坑

• epoll 内部会持有 fd 引用，必须保证 epoll_ctl(DEL) 完成、epoll 回调处理完毕后，再释放 Socket（依靠 refcount）
• 禁止直接 close 仍被 epoll 监听的 fd，否则 epoll 会持续访问无效 fd，造成脏读、崩溃
• 关闭顺序：epoll_ctl(DEL) → 等待回调完成 → close / sk_release

6.3 惊群问题

• 原版 epoll + 多线程会出现惊群：多个线程同时被同一个 fd 事件唤醒、争抢处理
  • 学习原型：单线程 epoll 事件循环，规避多线程惊群问题
  • 真实内核：EPOLLEXCLUSIVE 独占模式解决惊群
• 主线程事件循环保持单线程，IO处理可交由工作线程池，主线程只做事件分发

6.4 非阻塞半包 & EAGAIN 处理

• 非阻塞 send 缓冲区满返回 EAGAIN：注册 EPOLLOUT 事件，等缓冲区可写后继续发送剩余数据
• 非阻塞 recv 半包：保存临时缓存，等待下次 EPOLLIN 事件继续读取，不可阻塞

6.5 信号/超时/僵死连接

• 增加连接空闲超时定时器，自动关闭长期无数据的僵死长连接
• 处理 EPOLLHUP/EPOLLERR 挂起/异常事件，主动清理无效 fd，避免 epoll 堆积无效监听

七、端到端测试方案

1. 同时建立 100+ TCP客户端 + 若干UDP客户端连接，通过epoll单线程事件循环完成回声通信
2. 验证：并发收发正常、无数据损坏、无内存泄漏、无卡死
3. 验证：正常关闭fd后不再收到无效epoll事件
4. 验证：半关闭shutdown后可正常读完剩余数据，EPOLLIN可正常触发EOF事件

八、整体架构变化对比

✅ 旧架构

• 主线程阻塞在单个 fd 等待队列，串行处理连接
• 并发上限极低，不适合海量长连接

✅ epoll事件驱动架构

• 主线程仅阻塞在 epoll_wait 全局等待队列
• 就绪事件回调驱动读写，单线程可处理上千连接
• 分层模型不变：epoll 属于 Socket 层上层IO调度框架，不改动底层 fd映射、SocketSk、传输层TCP/UDP逻辑