土法炼钢兴趣小组的算法知识备份

IO 多路复用层 (The Backend)

2025-11-27 · Libevent · #C #epoll #kqueue #IO Multiplexing

目录

Libevent 之所以能在 Linux、BSD、Windows 上都能跑出高性能，归功于其精妙的后端抽象层。本篇我们将深入源码，看看它是如何封装 epoll 等系统调用的。

1. 统一接口: struct eventop

Libevent 定义了一套统一的接口 struct eventop (位于 event-internal.h)，所有的后端（epoll, kqueue, select 等）都必须实现这套接口。这是一种典型的 多态 (Polymorphism) 设计（虽然是用 C 语言实现的）。

struct eventop {
    const char *name; // 后端名称，如 "epoll"
    
    // 初始化
    void *(*init)(struct event_base *);
    
    // 注册/注销事件
    // fd: 文件描述符
    // old: 旧的事件掩码
    // events: 新的事件掩码 (EV_READ | EV_WRITE | EV_ET)
    // fdinfo: 后端特定的辅助数据
    int (*add)(struct event_base *, evutil_socket_t fd, short old, short events, void *fdinfo);
    int (*del)(struct event_base *, evutil_socket_t fd, short old, short events, void *fdinfo);
    
    // 核心分发函数 (对应 epoll_wait)
    int (*dispatch)(struct event_base *, struct timeval *);
    
    // 清理
    void (*dealloc)(struct event_base *);
    
    // ... 其他特性标志 (need_reinit, features)
};

2. 源码分析: epoll.c

以 Linux 下最常用的 epoll 为例，看看它是如何实现上述接口的。源码位于 epoll.c。

2.1. 初始化 (epoll_init)

static void *epoll_init(struct event_base *base) {
    int epfd = -1;
    struct epollop *epollop;

    // 1. 创建 epoll 句柄
    if ((epfd = epoll_create(32000)) == -1) {
        return NULL;
    }

    // 2. 设置 FD_CLOEXEC，防止 fork 后泄露
    evutil_make_socket_closeonexec(epfd);

    // 3. 分配后端数据结构
    epollop = mm_calloc(1, sizeof(struct epollop));
    epollop->epfd = epfd;

    // 4. 预分配 events 数组 (用于 epoll_wait 接收结果)
    epollop->events = mm_calloc(INITIAL_NEVENT, sizeof(struct epoll_event));
    epollop->nevents = INITIAL_NEVENT;

    return epollop;
}

2.2. 注册事件 (epoll_nochangelist_add)

当用户调用 event_add 时，最终会调用到后端的 add 方法。

static int epoll_nochangelist_add(struct event_base *base, evutil_socket_t fd,
                       short old, short events, void *p) {
    struct epollop *epollop = base->evbase;
    struct epoll_event epev;
    int op, events_to_set = 0;

    // 1. 将 Libevent 的事件标志转换为 epoll 的标志
    if (events & EV_READ) events_to_set |= EPOLLIN;
    if (events & EV_WRITE) events_to_set |= EPOLLOUT;
    if (events & EV_ET) events_to_set |= EPOLLET;

    // 2. 决定是 EPOLL_CTL_ADD 还是 EPOLL_CTL_MOD
    if (old == 0) {
        op = EPOLL_CTL_ADD;
    } else {
        op = EPOLL_CTL_MOD;
    }

    epev.data.fd = fd;
    epev.events = events_to_set;

    // 3. 调用系统调用
    if (epoll_ctl(epollop->epfd, op, fd, &epev) == -1) {
        return -1;
    }
    return 0;
}

优化细节: Libevent 实现了 changelist 机制（默认开启）。它不会每次 event_add 都立即调用 epoll_ctl，而是先在用户态缓存操作，等到 dispatch 时一次性批量应用。这大大减少了系统调用次数。

2.3. 事件分发 (epoll_dispatch)

这是 Reactor 的心脏跳动。

static int epoll_dispatch(struct event_base *base, struct timeval *tv) {
    struct epollop *epollop = base->evbase;
    struct epoll_event *events = epollop->events;
    int i, res;
    long timeout = -1;

    // 1. 将 timeval 转换为毫秒
    if (tv != NULL) {
        timeout = evutil_tvto_msec(tv);
    }

    // 2. 阻塞等待 (epoll_wait)
    res = epoll_wait(epollop->epfd, events, epollop->nevents, timeout);

    if (res == -1) {
        // 处理 EINTR 等错误
        return 0;
    }

    // 3. 遍历结果，激活事件
    for (i = 0; i < res; i++) {
        int what = events[i].events;
        short ev = 0;

        // 将 epoll 标志转回 Libevent 标志
        if (what & (EPOLLHUP|EPOLLERR)) {
            ev = EV_READ | EV_WRITE;
        } else {
            if (what & EPOLLIN) ev |= EV_READ;
            if (what & EPOLLOUT) ev |= EV_WRITE;
        }

        // 激活事件 (插入 active 队列)
        // 注意：这里不执行回调，只负责激活
        evmap_io_active(base, events[i].data.fd, ev);
    }

    // 4. 动态扩容
    // 如果这次 epoll_wait 填满了所有 events 槽位，说明负载很高
    // 下次循环前扩大 events 数组，避免频繁系统调用
    if (res == epollop->nevents && epollop->nevents < MAX_NEVENT) {
        // realloc events array...
    }

    return 0;
}

3. ET (Edge Trigger) vs LT (Level Trigger)

Libevent 默认使用 LT (水平触发) 模式。如果你在 event_new 时指定了 EV_ET，它会透传给 epoll。

Libevent 对 ET 的支持主要体现在 epoll_nochangelist_add 中设置 EPOLLET 标志。对于 bufferevent，它内部处理了 ET 模式下的循环读取逻辑。

4. 后端选择机制

Libevent 定义了一个全局数组 eventops，按优先级列出了所有支持的后端：

static const struct eventop *eventops[] = {
#ifdef HAVE_EVENT_PORTS
    &evportops,
#endif
#ifdef HAVE_WORKING_KQUEUE
    &kqops,
#endif
#ifdef HAVE_EPOLL
    &epollops,
#endif
#ifdef HAVE_DEVPOLL
    &devpollops,
#endif
#ifdef HAVE_POLL
    &pollops,
#endif
#ifdef HAVE_SELECT
    &selectops,
#endif
    NULL
};

在 event_base_new 时，它会遍历这个数组，尝试调用 init 方法。第一个初始化成功的后端将被选中。这就是为什么在 Linux 上它自动用 epoll，在 macOS 上自动用 kqueue 的原因。

5. 总结

Libevent 的后端层通过 struct eventop 实现了优雅的抽象。它不仅封装了 epoll_ctl/epoll_wait，还做了很多工程上的优化（如 Changelist 缓存、动态扩容）。

理解了这一层，我们就能明白为什么 Libevent 在高并发下依然能保持低延迟。下一篇，我们将对比不同操作系统后端的差异，探讨跨平台开发的深坑。

上一篇: 01-core/event-base-loop.md - Event Base 与 Event Loop 下一篇: 01-core/cross-platform.md - 跨平台后端对比

返回 Libevent 专题索引

By .