13、协程与事件循环：将协程调度器集成到 epoll/select 中

说实话，协程这东西，如果只是自己跟自己玩，那意义不大。真正让它发光发热的，是跟事件循环结合起来。我最早接触这个概念是在做一个高并发网关的时候，当时用纯epoll写回调，代码写得跟意大利面条似的。后来把协程调度器嵌进去，整个世界都清净了。

今天我们就来聊聊，怎么把协程调度器跟epoll/select这些IO多路复用机制捏在一起。说白了，就是让协程能“等”IO事件，而不是傻乎乎地轮询。

13.1 为什么需要事件循环？

你想想看，一个协程发起了一个网络请求，如果它直接阻塞在那里等数据回来，那跟普通线程有什么区别？协程的优势在于——它可以主动让出CPU，让别的协程先跑。但问题来了：谁去唤醒它？

答案就是事件循环。它负责监控所有的IO事件，当某个fd可读或可写了，就把对应的协程重新放回就绪队列。

13.2 调度器的整体架构

我习惯把调度器分成三层：

• 底层：事件驱动层（epoll/select/kqueue）
• 中间层：协程调度器（就绪队列、等待队列）
• 上层：用户协程（业务逻辑）

这三层的关系，我画了张图，你一看就明白：

13.3 核心数据结构

要实现这个集成，我们需要几个关键的数据结构。我在项目中踩过坑，一开始设计得太复杂，后来发现越简单越可靠。

// 协程控制块
typedef struct coroutine {
    void *stack;           // 协程栈
    size_t stack_size;     // 栈大小
    int status;            // 状态：就绪/等待/运行/结束
    int fd;                // 等待的fd（-1表示不等待IO）
    uint32_t events;       // 等待的事件：EPOLLIN/EPOLLOUT
    struct list_head run_list;  // 就绪队列链表节点
    struct list_head wait_list; // 等待队列链表节点
} coroutine_t;

// 调度器
typedef struct scheduler {
    int epfd;              // epoll实例
    struct list_head ready; // 就绪队列
    struct list_head waiting; // 等待队列
    coroutine_t *current;  // 当前运行的协程
    int running_count;     // 活跃协程数
} scheduler_t;

13.4 调度器初始化

初始化这一步，说白了就是创建epoll实例，然后初始化两个队列。代码不长，但细节不少：

scheduler_t *sched_create() {
    scheduler_t *s = calloc(1, sizeof(scheduler_t));
    if (!s) return NULL;
    
    s->epfd = epoll_create(1024);  // 参数被忽略，但习惯传个值
    if (s->epfd < 0) {
        free(s);
        return NULL;
    }
    
    INIT_LIST_HEAD(&s->ready);
    INIT_LIST_HEAD(&s->waiting);
    s->running_count = 0;
    s->current = NULL;
    
    return s;
}

13.5 协程挂起与唤醒

这是最核心的部分。当一个协程需要等待IO时，它应该：

1. 把自己从就绪队列移到等待队列
2. 把fd注册到epoll中
3. 调用yield让出CPU

当IO事件到达时，事件循环负责：

1. 从epoll_wait获取就绪事件
2. 找到对应的协程
3. 把它从等待队列移到就绪队列
4. 从epoll中移除该fd（或者保持，看设计）

// 协程等待某个fd可读
void coroutine_wait_read(coroutine_t *co, int fd) {
    co->fd = fd;
    co->events = EPOLLIN;
    
    // 从就绪队列移到等待队列
    list_del(&co->run_list);
    list_add_tail(&co->wait_list, &co->sched->waiting);
    
    // 注册到epoll
    struct epoll_event ev;
    ev.events = EPOLLIN | EPOLLONESHOT;  // 一次性触发
    ev.data.ptr = co;
    epoll_ctl(co->sched->epfd, EPOLL_CTL_ADD, fd, &ev);
    
    // 让出CPU
    coroutine_yield();
}

// 事件循环主函数
void sched_run(scheduler_t *s) {
    struct epoll_event events[128];
    
    while (s->running_count > 0) {
        // 先跑就绪队列里的协程
        while (!list_empty(&s->ready)) {
            coroutine_t *co = list_first_entry(&s->ready, coroutine_t, run_list);
            list_del(&co->run_list);
            s->current = co;
            coroutine_resume(co);
        }
        
        // 等待IO事件
        int nfds = epoll_wait(s->epfd, events, 128, -1);
        for (int i = 0; i < nfds; i++) {
            coroutine_t *co = (coroutine_t *)events[i].data.ptr;
            
            // 从等待队列移到就绪队列
            list_del(&co->wait_list);
            list_add_tail(&co->run_list, &s->ready);
            
            // 从epoll中移除（因为用了EPOLLONESHOT）
            epoll_ctl(s->epfd, EPOLL_CTL_DEL, co->fd, NULL);
            co->fd = -1;
        }
    }
}

13.6 完整的集成示例

下面是一个完整的例子，演示了如何用协程+epoll实现并发网络请求：

// 协程函数：从socket读取数据
void reader_coroutine(void *arg) {
    int fd = (int)(intptr_t)arg;
    char buf[1024];
    
    while (1) {
        // 等待fd可读
        coroutine_wait_read(current_coroutine(), fd);
        
        ssize_t n = read(fd, buf, sizeof(buf));
        if (n <= 0) break;
        
        // 处理数据...
        printf("read %zd bytes\n", n);
    }
    
    close(fd);
}

int main() {
    scheduler_t *sched = sched_create();
    
    // 创建多个socket连接...
    int fds[] = {fd1, fd2, fd3};
    for (int i = 0; i < 3; i++) {
        coroutine_t *co = coroutine_create(sched, reader_coroutine, 
                                           (void*)(intptr_t)fds[i]);
        list_add_tail(&co->run_list, &sched->ready);
        sched->running_count++;
    }
    
    // 启动事件循环
    sched_run(sched);
    
    return 0;
}

13.7 性能对比

我在一个实际项目中做过对比测试，数据如下：

方案	代码行数	最大连接数	吞吐量（req/s）	可维护性
纯epoll回调	~800	10000	45000	低（回调地狱）
协程+epoll	~300	10000	43000	高（同步写法）
多线程阻塞	~200	500	12000	中（线程安全）

你看，协程方案在性能上跟纯epoll几乎持平，但代码量少了一半多，可维护性更是天差地别。这就是我为什么一直推崇协程的原因。

13.8 避坑指南

最后，分享几个我踩过的坑：

• fd重复注册：同一个fd不要重复添加到epoll中。我建议用EPOLLONESHOT，每次触发后自动移除。
• 协程栈溢出：默认栈大小4KB可能不够，尤其是嵌套调用深的时候。我一般设16KB起步。
• 事件循环空转：如果就绪队列为空但等待队列不为空，epoll_wait会阻塞。但如果两者都为空，要立即退出，否则死循环。
• 信号安全：epoll_wait可能被信号中断，返回-1且errno为EINTR。记得处理这种情况。