Redis IO模型与多路复用实现

/* === aeApiCreate: 创建 epoll 实例 === */
static int aeApiCreate(aeEventLoop *eventLoop) {
    aeApiState *state = zmalloc(sizeof(aeApiState));
    
    // 创建 epoll 实例，1024是提示值非限制
    state->epfd = epoll_create(1024);
    anetCloexec(state->epfd);  // 设置O_CLOEXEC
    
    // 分配事件数组
    state->events = zmalloc(sizeof(struct epoll_event) * eventLoop->setsize);
    eventLoop->apidata = state;
    return 0;
}

/* === aeApiPoll: 等待事件 === */
/* 关键：将 epoll 事件转换为 Redis 事件 */
static int aeApiPoll(aeEventLoop *eventLoop, long tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval, numevents = 0;
    
    // 调用 epoll_wait 等待事件
    retval = epoll_wait(state->epfd, state->events, 
                  eventLoop->nevents, tvp);
    
    if (retval > 0) {
        numevents = retval;
        for (int j = 0; j < numevents; j++) {
            int mask = 0;
            struct epoll_event *e = &state->events[j];
            
            // EPOLLIN → AE_READABLE
            if (e->events & EPOLLIN) mask |= AE_READABLE;
            // EPOLLOUT → AE_WRITABLE
            if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
            // 错误/挂起也标记为可读写
            if (e->events & EPOLLERR | e->events & EPOLLHUP)
                mask |= AE_READABLE | AE_WRITABLE;
            
            eventLoop->fired[j].fd = e->data.fd;
            eventLoop->fired[j].mask = mask;
        }
    }
    return numevents;
}

/* === aeApiAddEvent: 添加/修改监听 === */
/* 智能判断: 新fd用ADD，已有fd用MOD */
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
    aeApiState *state = eventLoop->apidata;
    struct epoll_event ee = {0};  // 消除valgrind警告
    
    int op = eventLoop->events[fd].mask == AE_NONE ?
           EPOLL_CTL_ADD : EPOLL_CTL_MOD;
    
    // 合并新旧事件掩码
    int events = mask | eventLoop->events[fd].mask;
    
    // Redis掩码 → epoll事件
    if (mask & AE_READABLE) events |= EPOLLIN;
    if (mask & AE_WRITABLE) events |= EPOLLOUT;
    
    ee.events = events;
    ee.data.fd = fd;
    
    if (epoll_ctl(state->epfd, op, fd, &ee) == -1) return -1;
    return 0;
}

/* === aeApiState: select需要保存两份fd_set === */
/* 原因: select返回后原fd_set会被内核修改 */
typedef struct {
    fd_set rfds, wfds;     // 当前监听的读写集合
    fd_set _rfds, _wfds;    // 备份，select前拷贝
} aeApiState;

/* === aeApiPoll: select 实现 === */
/* 性能痛点: 每次都要memcpy + O(n)遍历 */
static int aeApiPoll(aeEventLoop *eventLoop, long tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval, i, count = 0;
    
    // 拷贝fd_set (用户态↔内核态 数据传输)
    memcpy(&state->_rfds, &state->rfds, sizeof(fd_set));
    memcpy(&state->_wfds, &state->wfds, sizeof(fd_set));
    
    long sec = tvp / 1000;
    long usec = (tvp % 1000) * 1000;
    struct timeval tv = {sec, usec};
    
    // 调用 select (maxfd+1，因为fd从0开始)
    retval = select(eventLoop->maxfd + 1, 
              &state->_rfds, &state->_wfds, NULL, &tv);
    
    if (retval > 0) {
        // O(n) 遍历: 必须检查所有fd
        for (i = 0; i <= eventLoop->maxfd; i++) {
            int mask = 0;
            aeFileEvent *fe = &eventLoop->events[i];
            
            if (fe->mask == AE_NONE) continue;
            
            // FD_ISSET 逐个检查
            if (FD_ISSET(i, &state->_rfds)) mask |= AE_READABLE;
            if (FD_ISSET(i, &state->_wfds)) mask |= AE_WRITABLE;
            
            if (mask) {
                eventLoop->fired[count].fd = i;
                eventLoop->fired[count++].mask = mask;
            }
        }
    }
    return count;
}

/* === 连接数限制 === */
/* select 默认 FD_SETSIZE = 1024 */
static int aeApiResize(aeEventLoop *eventLoop, int setsize) {
    if (setsize >= FD_SETSIZE) return AE_ERR;
    /* ... 重新分配 fd_set ... */
    return 0;
}

/* Redis 服务器主循环 */
void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    serverLog(LL_NOTICE, "Redis ready to accept connections");
    
    /* 事件循环 - 永不退出 */
    while (!eventLoop->stop) {
        /* 处理文件事件和时间事件 */
        aeProcessEvents(eventLoop, 
            AE_FILE_EVENTS |   // 处理客户端命令
            AE_TIME_EVENTS |   // 处理定时任务
            AE_DONT_WAIT);    // 控制阻塞时间
    }
}

/* main() 中初始化事件循环 */
/* aeCreateEventLoop(setsize) - setsize默认10032 */