高性能 I/O 框架库 Libevent"/>
Linux 网络编程学习笔记——十二、高性能 I/O 框架库 Libevent
目录
在处理 I/O 事件、信号和定时事件时，需要考虑如下三个问题：
- 统一事件源：很明显，统一处理这三类事件既能使代码简单易懂，又能避免一些潜在的逻辑错误。
- 可移植性：不同的操作系统具有不同的 I/O 复用方式，比如 Solaris 的 dev/poll 文件，FreeBSD 的 kqueue 机制，Linux 的 epoll 系列系统调用。
- 对并发编程的支持：在多进程和多线程环境下，我们需要考虑各执行实体如何协同处理客户连接、信号和定时器，以避免竞态条件。
一、I/O 框架库概述
I/O 框架库以库函数的形式，封装了较为底层的系统调用，给应用程序提供了一组更便于使用的接口。这些库函数往往比程序员自己实现的同样功能的函数更合理、更高效，且更健壮。因为它们经受住了真实网络环境下的高压测试，以及时间的考验。各种 I/O 框架库的实现原理基本相似，要么以 Reactor 模式实现，要么以 Proactor 模式实现，要么同时以这两种模式实现。举例来说，基于 Reactor 模式的 I/O 框架库包含如下几个组件：句柄（Handle）、事件多路分发器（EventDemultiplexer）、事件处理器（EventHandler）和具体的事件处理器（ConcreteEventHandler）、Reactor 。这些组件的关系下：
1. 句柄
I/O 框架库要处理的对象，即 I/O 事件、信号和定时事件，统一称为事件源。一个事件源通常和一个句柄绑定在一起。句柄的作用是：当内核检测到就绪事件时，它将通过句柄来通知应用程序这一事件。在 Linux 环境下，I/O 事件对应的句柄是文件描述符，信号事件对应的句柄就是信号值。
2. 事件多路分发器
事件的到来是随机的、异步的。我们无法预知程序何时收到一个客户连接请求，又亦或收到一个暂停信号。所以程序需要循环地等待并处理事件，这就是事件循环。在事件循环中，等待事件一般使用 I/O 复用技术来实现。I/O 框架库一般将系统支持的各种 I/O 复用系统调用封装成统一的接口，称为事件多路分发器。事件多路分发器的 demultiplex 方法是等待事件的核心函数，其内部调用的是 select 、poll 、epoll_wait 等函数。
此外，事件多路分发器还需要实现 register_event 和 remove_event 方法，以供调用者往事件多路分发器中添加事件和从事件多路分发器中删除事件。
3. 事件处理器和具体事件处理器
事件处理器执行事件对应的业务逻辑。它通常包含一个或多个 handle_event 回调函数，这些回调函数在事件循环中被执行。 I/O 框架库提供的事件处理器通常是一个接口，用户需要继承它来实现自己的事件处理器，即具体事件处理器。因此，事件处理器中的回调函数一般被声明为虚函数，以支持用户的扩展。
此外，事件处理器一般还提供一个 get_handle 方法，它返回与该事件处理器关联的句柄。那么，事件处理器和句柄有什么关系？当事件多路分发器检测到有事件发生时，它是通过句柄来通知应用程序的。因此，我们必须将事件处理器和句柄绑定，才能在事件发生时获取到正确的事件处理器。
4. Reactor
Reactor 是 I/O 框架库的核心。它提供的几个主要方法是：
- handle_events：该方法执行事件循环。它重复如下过程：等待事件，然后依次处理所有就绪事件对应的事件处理器；
- register_handler：该方法调用事件多路分发器的 register_event 方法来往事件多路分发器中注册一个事件；
- remove_handler：该方法调用事件多路分发器的 remove_event 方法来删除事件多路分发器中的一个事件。
二、Libevent 源码分析（基于 2012 年 5 月 3 日发布的 2.0.19）
Libevent 是开源社区的一款高性能的 I/O 框架库，其学习者和使用者众多。使用 Libevent 的著名案例有：高性能的分布式内存对象缓存软件 memcached ，Google 浏览器 Chromium 的 Linux 版本。作为一个 I/O 框架库，Libevent 具有如下特点：
- 跨平台支持：Libevent 支持 Linux 、UNIX 和 Windows ；
- 统一事件源：Libevent 对 I/O 事件、信号和定时事件提供统一的处理；
- 线程安全：Libevent 使用 libevent_pthreads 库来提供线程安全支持；
- 基于 Reactor 模式的实现。
1. hello world 实例
#include <sys/signal.h>
#include <event.h>
void signal_cb(int fd, short event, void *argc)
{struct event_base *base = (event_base *)argc;struct timeval delay = {2, 0};printf("Caught an interrupt signal;exiting cleanly in two seconds...\n");event_base_loopexit(base, &delay);
}
void timeout_cb(int fd, short event, void *argc) { printf("timeout\n"); }
int main()
{struct event_base *base = event_init();struct event *signal_event = evsignal_new(base, SIGINT, signal_cb, base);event_add(signal_event, NULL);timeval tv = {1, 0};struct event *timeout_event = evtimer_new(base, timeout_cb, NULL);event_add(timeout_event, &tv);event_base_dispatch(base);event_free(timeout_event);event_free(signal_event);event_base_free(base);
}
由代码可知，调用 Libevent 库的主要逻辑如下：
- 调用 event_init 函数创建 event_base 对象。一个 event_base 相当于一个 Reactor 实例；
- 创建具体的事件处理器，并设置它们所从属的 Reactor 实例。evsignal_new 和 evtimer_new 分别用于创建信号事件处理器和定时事件处理器，它们是定义在 include/event2/event.h 文件中的宏：
可见，它们的统一入口是 event_new 函数，即用于创建通用事件处理器（图 12 - 1 中的 EventHandler）的函数。其定义是：#define evsignal_new(b,x,cb,arg) event_new((b),(x),EV_SIGNAL|EV_PERSIST,(cb),(arg)) #define evtimer_new(b,cb,arg) event_new((b),-1,0,(cb),(arg))struct event*event_new(struct event_base *base, evutil_socket_t fd, short events, void(*cb)(evutil_socket_t, short,void*), void*arg)- base：指定新创建的事件处理器从属的 Reactor ；
- fd：指定与该事件处理器关联的句柄：
- 创建 I/O 事件处理器时，应该给 fd 参数传递文件描述符值；
- 创建信号事件处理器时，应该给 fd 参数传递信号值；
- 创建定时事件处理器时，应该给 fd 参数传递 -1 ；
- events：指定事件类型，其可选值都定义在 include/event2/event.h 文件中：
EV_PERSIST：事件被触发后，自动重新对这个 event 调用 event_add 函数；#define EV_TIMEOUT 0x01/*定时事件*/ #define EV_READ 0x02/*可读事件*/ #define EV_WRITE 0x04/*可写事件*/ #define EV_SIGNAL 0x08/*信号事件*/ #define EV_PERSIST 0x10/*永久事件*/ /*边沿触发事件，需要I/O复用系统调用支持，比如epoll*/ #define EV_ET 0x20 - cb：指定目标事件对应的回调函数，相当于图 12 - 1 中事件处理器的 handle_event 方法；
- arg：是 Reactor 传递给回调函数的参数；
- event_new 函数成功时返回一个 event 类型的对象，也就是 Libevent 的事件处理器；
- Libevent 用单词 event 来描述事件处理器，而不是事件，为防止歧义，约定如下：
- 事件指的是一个句柄上绑定的事件，比如文件描述符 0 上的可读事件；
- 事件处理器，也就是 event 结构体类型的对象，除了包含事件必须具备的两个要素（句柄和事件类型）外，还有很多其他成员，比如回调函数；
- 事件由事件多路分发器管理，事件处理器则由事件队列管理。事件队列包括多种，比如 event_base 中的注册事件队列、活动事件队列和通用定时器队列，以及 evmap 中的 I/O 事件队列、信号事件队列；
- 事件循环对一个被激活事件（就绪事件）的处理，指的是执行该事件对应的事件处理器中的回调函数。
- 调用 event_add 函数，将事件处理器添加到注册事件队列中，并将该事件处理器对应的事件添加到事件多路分发器中。event_add 函数相当于 Reactor 中的 register_handler 方法。
- 调用 event_base_dispatch 函数来执行事件循环。
- 事件循环结束后，使用 *_free 系列函数来释放系统资源。
2. 源代码组织架构
Libevent 源代码中的目录和文件按照功能可划分为如下部分：
- 头文件目录 include/event2：该目录是自 Libevent 主版本升级到 2.0 之后引入的，在 1.4 及更老的版本中并无此目录。该目录中的头文件是 Libevent 提供给应用程序使用的，比如，event.h 头文件提供核心函数，http.h 头文件提供 HTTP 协议相关服务，rpc.h 头文件提供远程过程调用支持。
- 源码根目录下的头文件：这些头文件分为两类：一类是对 include/event2 目录下的部分头文件的包装，另外一类是供 Libevent 内部使用的辅助性头文件，它们的文件名都具有 *-internal.h 的形式。
- 通用数据结构目录 compat/sys：该目录下仅有一个文件 queue.h 。它封装了跨平台的基础数据结构，包括单向链表、双向链表、队列、尾队列和循环队列。
- sample 目录：提供一些示例程序。
- test 目录：提供一些测试代码。
- WIN32-Code 目录：提供 Windows 平台上的一些专用代码。
- event.c 文件：该文件实现 Libevent 的整体框架，主要是 event 和 event_base 两个结构体的相关操作。
- devpoll.c 、kqueue.c 、evport.c 、select.c 、win32select.c、poll.c 和 epoll.c 文件：分别封装了如下 I/O 复用机制：/dev/poll 、kqueue 、event ports 、POSIX select 、Windows select 、poll 和 epoll 。这些文件的主要内容相似，都是针对结构体 eventop 所定义的接口函数的具体实现。
- minheap-internal.h 文件：实现了一个时间堆，以提供对定时事件的支持。
- signal.c 文件：提供对信号的支持。其内容也是针对结构体 eventop 所定义的接口函数的具体实现。
- evmap.c 文件：维护句柄（文件描述符或信号）与事件处理器的映射关系。
- event_tagging.c 文件：提供往缓冲区中添加标记数据（比如一个整数），以及从缓冲区中读取标记数据的函数。
- event_iocp.c 文件：提供对 Windows IOCP（Input/Output Completion Port，输入输出完成端口）的支持。
- buffer*.c 文件：提供对网络 I/O 缓冲的控制，包括：输入输出数据过滤，传输速率限制，使用SSL（Secure Sockets Layer）协议对应用数据进行保护，以及零拷贝文件传输等。
- evthread*.c 文件：提供对多线程的支持。
- listener.c 文件：封装了对监听 socket 的操作，包括监听连接和接受连接。
- logs.c 文件：是 Libevent 的日志系统。
- evutil.c 、evutil_rand.c 、strlcpy.c 和 arc4random.c 文件：提供一些基本操作，比如生成随机数、获取 socket 地址信息、读取文件、设置 socket 属性等。
- evdns.c 、http.c 和 evrpc.c 文件：分别提供了对 DNS 协议、 HTTP 协议和 RPC（Remote Procedure Call ，远程过程调用）协议的支持。
- epoll_sub.c 文件：该文件未见使用。
在整个源码中，event-internal.h 、include/event2/event_struct.h 、event.c 和 evmap.c 等 4 个文件最为重要。它们定义了 event 和 event_base 结构体，并实现了这两个结构体的相关操作。
3. event 结构体
前文提到，Libevent 中的事件处理器是 event 结构类型。event 结构体封装了句柄、事件类型、回调函数，以及其他必要的标志和数据。该结构体在 include/event2/event_struct.h 文件中定义：
struct event { TAILQ_ENTRY(event)ev_active_next; TAILQ_ENTRY(event)ev_next; union{ TAILQ_ENTRY(event)ev_next_with_common_timeout; int min_heap_idx; } ev_timeout_pos; evutil_socket_t ev_fd; struct event_base*ev_base;union{ struct{ TAILQ_ENTRY(event)ev_io_next; struct timeval ev_timeout; }ev_io; struct{ TAILQ_ENTRY(event)ev_signal_next; short ev_ncalls; short*ev_pncalls; }ev_signal; }_ev; short ev_events; short ev_res; short ev_flags; ev_uint8_t ev_pri; ev_uint8_t ev_closure; struct timeval ev_timeout; void(*ev_callback)(evutil_socket_t,short,void*arg); void*ev_arg;
};
- ev_events：代表事件类型。其取值可以是 Libevent 支持的事件类型的标志的按位或（互斥的事件类型除外，比如读写事件和信号事件就不能同时被设置）；
- ev_next：所有已经注册的事件处理器（包括 I/O 事件处理器和信号事件处理器）通过该成员串联成一个尾队列，我们称之为注册事件队列。宏 TAILQ_ENTRY 是尾队列中的节点类型，它定义在 compat/sys/queue.h 文件中：
#define TAILQ_ENTRY(type) struct{struct type*tqe_next; /*下一个元素*/ struct type**tqe_prev; /*前一个元素的地址*/ } - ev_active_next：所有被激活的事件处理器通过该成员串联成一个尾队列，我们称之为活动事件队列。活动事件队列不止一个，不同优先级的事件处理器被激活后将被插入不同的活动事件队列中。在事件循环中，Reactor 将按优先级从高到低遍历所有活动事件队列，并依次处理其中的事件处理器；
- ev_timeout_pos：这是一个联合体，它仅用于定时事件处理器。为了讨论的方便，后面称定时事件处理器为定时器。老版本的 Libevent 中，定时器都是由时间堆来管理的。但开发者认为有时候使用简单的链表来管理定时器将具有更高的效率。因此，新版本的 Libevent 就引入了所谓“通用定时器”的概念。这些定时器不是存储在时间堆中，而是存储在尾队列中，我们称之为通用定时器队列。对于通用定时器而言，ev_timeout_pos 联合体的 ev_next_with_common_timeout 成员指出了该定时器在通用定时器队列中的位置。对于其他定时器而言，ev_timeout_pos 联合体的 min_heap_idx 成员指出了该定时器在时间 堆中的位置；
- _ev：这是一个联合体。所有具有相同文件描述符值的 I/O 事件处理器通过 ev.ev_io.ev_io_next 成员串联成一个尾队列，称之为 I/O 事件队列；所有具有相同信号值的信号事件处理器通过 ev.ev_signal.ev_signal_next 成员串联成一个尾队列，称之为信号事件队列。ev.ev_signal.ev_ncalls 成员指定信号事件发生时，Reactor 需要执行多少次该事件对应的事件处理器中的回调函数。ev.ev_signal.ev_pncalls 指针成员要么是 NULL ，要么指向 ev.ev_signal.ev_ncalls ；在程序中，我们可能针对同一个 socket 文件描述符上的可读/可写事件创建多个事件处理器（它们拥有不同的回调函数）。当该文件描述符上有可读/可写事件发生时，所有这些事件处理器都应该被处理。所以，Libevent 使用 I/O 事件队列将具有相同文件描述符值的事件处理 器组织在一起。这样，当一个文件描述符上有事件发生时，事件多路分发器就能很快地把所有相关的事件处理器添加到活动事件队列中。信号事件队列的存在也是由于相同的原因。可见，I/O 事件队列和信号事件队列并不是注册事件队列的细致分类，而是另有用处；
- ev_fd：对于 I/O 事件处理器，它是文件描述符值；对于信号事件处理器，它是信号值；
- ev_base：该事件处理器从属的 event_base 实例；
- ev_res：记录当前激活事件的类型；
- ev_flags：是一些事件标志。其可选值定义在 include/event2/event_struct.h 文件中：
#define EVLIST_TIMEOUT 0x01/*事件处理器从属于通用定时器队列或时间堆*/ #define EVLIST_INSERTED 0x02/*事件处理器从属于注册事件队列*/ #define EVLIST_SIGNAL 0x04/*没有使用*/ #define EVLIST_ACTIVE 0x08/*事件处理器从属于活动事件队列*/ #define EVLIST_INTERNAL 0x10/*内部使用*/ #define EVLIST_INIT 0x80/*事件处理器已经被初始化*/ #define EVLIST_ALL(0xf000|0x9f)/*定义所有标志*/ - ev_pri：指定事件处理器优先级，值越小则优先级越高；
- ev_closure：指定 event_base 执行事件处理器的回调函数时的行为。其可选值定义于 event-internal.h 文件中：
/*默认行为*/ #define EV_CLOSURE_NONE 0 /*执行信号事件处理器的回调函数时，调用ev.ev_signal.ev_ncalls次该回调函数 */ #define EV_CLOSURE_SIGNAL 1 /*执行完回调函数后，再次将事件处理器加入注册事件队列中*/ #define EV_CLOSURE_PERSIST 2 - ev_timeout：仅对定时器有效，指定定时器的超时值；
- ev_callback：是事件处理器的回调函数，由 event_base 调用。回调函数被调用时，它的 3 个参数分别被传入事件处理器的如下 3 个成员：ev_fd 、ev_res 和 ev_arg ；
- ev_arg：回调函数的参数。
4. 往注册事件队列中添加事件处理器
前面提到，创建一个 event 对象的函数是 event_new（及其变体），它在 event.c 文件中实现。该函数的实现相当简单，主要是给 event 对象分配内存并初始化它的部分成员。event 对象创建好之后，应用程序需要调用 event_add 函数将其添加到注册事件队列中，并将对应的事件注册到事件多路分发器上。event_add 函数在 event.c 文件中实现，主要是调用另外一个内部函数 event_add_internal ：
static inline int event_add_internal(struct event *ev, const struct timeval *tv, int tv_is_absolute)
{struct event_base *base = ev->ev_base;int res = 0;int notify = 0;EVENT_BASE_ASSERT_LOCKED(base);_event_debug_assert_is_setup(ev);event_debug(("event_add:event:%p(fd%d),%s%s%scall%p", ev, (int)ev->ev_fd, ev->ev_events & EV_READ ? "EV_READ" : "", ev->ev_events & EV_WRITE ? "EV_WRITE" : "", tv ? "EV_TIMEOUT" : "", ev->ev_callback));EVUTIL_ASSERT(!(ev->ev_flags & ~EVLIST_ALL)); /*如果新添加的事件处理器是定时器，且它尚未被添加到通用定时器队列或时间堆中， 则为该定时器在时间堆上预留一个位置*/if (tv != NULL && !(ev->ev_flags & EVLIST_TIMEOUT)){if (min_heap_reserve(&base->timeheap, 1 + min_heap_size(&base->timeheap)) == -1)return (-1);} /*如果当前调用者不是主线程（执行事件循环的线程），并且被添加的事件处理器是信 号事件处理器，而且主线程正在执行该信号事件处理器的回调函数，则当前调用者必须等待 主线程完成调用，否则将引起竞态条件（考虑event结构体的ev_ncalls和ev_pncalls成 员）*/
#ifndef_EVENT_DISABLE_THREAD_SUPPORT if (base->current_event == ev && (ev->ev_events & EV_SIGNAL) && !EVBASE_IN_THREAD(base)){++base->current_event_waiters;EVTHREAD_COND_WAIT(base->current_event_cond, base->th_base_lock);}
#endif if ((ev->ev_events & (EV_READ | EV_WRITE | EV_SIGNAL)) && !(ev->ev_flags & (EVLIST_INSERTED | EVLIST_ACTIVE))){if (ev->ev_events & (EV_READ | EV_WRITE)) /*添加I/O事件和I/O事件处理器的映射关系*/res = evmap_io_add(base, ev->ev_fd, ev);else if (ev->ev_events & EV_SIGNAL) /*添加信号事件和信号事件处理器的映射关系*/res = evmap_signal_add(base, (int)ev->ev_fd, ev);if (res != -1) /*将事件处理器插入注册事件队列*/event_queue_insert(base, ev, EVLIST_INSERTED);if (res == 1){/*事件多路分发器中添加了新的事件，所以要通知主线程*/ notify = 1;res = 0;}} /*下面将事件处理器添加至通用定时器队列或时间堆中。对于信号事件处理器和I/O事 件处理器，根据evmap_*_add函数的结果决定是否添加（这是为了给事件设置超时）；而对 于定时器，则始终应该添加之*/if (res != -1 && tv != NULL){struct timeval now;int common_timeout; /*对于永久性事件处理器，如果其超时时间不是绝对时间，则将该事件处理器的超时时 间记录在变量ev->ev_io_timeout中。ev_io_timeout是定义在event-internal.h 文件中的宏：#define ev_io_timeout_ev.ev_io.ev_timeout*/if (ev->ev_closure == EV_CLOSURE_PERSIST && !tv_is_absolute)ev->ev_io_timeout = *tv; /*如果该事件处理器已经被插入通用定时器队列或时间堆中，则先删除它*/if (ev->ev_flags & EVLIST_TIMEOUT){if (min_heap_elt_is_top(ev))notify = 1;event_queue_remove(base, ev, EVLIST_TIMEOUT);} /*如果待添加的事件处理器已经被激活，且原因是超时，则从活动事件队列中删除它， 以避免其回调函数被执行。对于信号事件处理器，必要时还需将其ncalls成员设置为0（注 意，ev_pncalls如果不为NULL，它指向ncalls）。前面提到，信号事件被触发时， ncalls指定其回调函数被执行的次数。将ncalls设置为0，可以干净地终止信号事件的处 理*/if ((ev->ev_flags & EVLIST_ACTIVE) && (ev->ev_res & EV_TIMEOUT)){if (ev->ev_events & EV_SIGNAL){if (ev->ev_ncalls && ev->ev_pncalls){*ev->ev_pncalls = 0;}}event_queue_remove(base, ev, EVLIST_ACTIVE);}gettime(base, &now);common_timeout = is_common_timeout(tv, base);if (tv_is_absolute){ev->ev_timeout = *tv; /*判断应该将定时器插入通用定时器队列，还是插入时间堆*/}else if (common_timeout){struct timeval tmp = *tv;tmp.tv_usec &= MICROSECONDS_MASK;evutil_timeradd(&now, &tmp, &ev->ev_timeout);ev->ev_timeout.tv_usec |= (tv->tv_usec & ~MICROSECONDS_MASK);}else{/*加上当前系统时间，以取得定时器超时的绝对时间*/ evutil_timeradd(&now, tv, &ev->ev_timeout);}event_debug(("event_add:timeout in%d seconds,call%p", (int)tv->tv_sec, ev->ev_callback));event_queue_insert(base, ev, EVLIST_TIMEOUT); /*最后，插入定时器*/ /*如果被插入的事件处理器是通用定时器队列中的第一个元素，则通过调用 common_timeout_schedule函数将其转移到时间堆中。这样，通用定时器链表和时间堆 中的定时器就得到了统一的处理*/if (common_timeout){struct common_timeout_list *ctl = get_common_timeout_list(base, &ev->ev_timeout);if (ev == TAILQ_FIRST(&ctl->events)){common_timeout_schedule(ctl, &now, ev);}}else{if (min_heap_elt_is_top(ev))notify = 1;}} /*如果必要，唤醒主线程*/if (res != -1 && notify && EVBASE_NEED_NOTIFY(base))evthread_notify_base(base);_event_debug_note_add(ev);return (res);
}
该函数内部调用了几个重要的函数：
- evmap_io_add：该函数将 I/O 事件添加到事件多路分发器中，并将对应的事件处理器添加到 I/O 事件队列中，同时建立 I/O 事件和 I/O 事件处理器之间的映射关系；
- evmap_signal_add：该函数将信号事件添加到事件多路分发器中，并将对应的事件处理器添加到信号事件队列中，同时建立信号事件和信号事件处理器之间的映射关系；
- event_queue_insert：该函数将事件处理器添加到各种事件队列中：将 I/O 事件处理器和信号事件处理器插入注册事件队列；将定时器插入通用定时器队列或时间堆；将被激活的事件处理器添加到活动事件队列中，其实现如下：
static void event_queue_insert(struct event_base *base, struct event *ev, int queue) {EVENT_BASE_ASSERT_LOCKED(base); /*避免重复插入*/if (ev->ev_flags & queue){ /*Double insertion is possible for active events*/if (queue & EVLIST_ACTIVE)return;event_errx(1, "%s:%p(fd%d)already on queue%x", __func__, ev, ev->ev_fd, queue);return;}if (~ev->ev_flags & EVLIST_INTERNAL)base->event_count++; /*将event_base拥有的事件处理器总数加1*/ev->ev_flags |= queue; /*标记此事件已被添加过*/switch (queue){ /*将I/O事件处理器或信号事件处理器插入注册事件队列*/case EVLIST_INSERTED:TAILQ_INSERT_TAIL(&base->eventqueue, ev, ev_next);break; /*将就绪事件处理器插入活动事件队列*/case EVLIST_ACTIVE:base->event_count_active++;TAILQ_INSERT_TAIL(&base->activequeues[ev->ev_pri], ev, ev_active_next);break; /*将定时器插入通用定时器队列或时间堆*/case EVLIST_TIMEOUT:{if (is_common_timeout(&ev->ev_timeout, base)){struct common_timeout_list *ctl = get_common_timeout_list(base, &ev->ev_timeout);insert_common_timeout_inorder(ctl, ev);}elsemin_heap_push(&base->timeheap, ev);break;}default:event_errx(1, "%s:unknown queue%x", __func__, queue);} }
5. 往事件多路分发器中注册事件
event_queue_insert 函数所做的仅仅是将一个事件处理器加入 event_base 的某个事件队列中。对于新添加的 I/O 事件处理器和信号事件处理器，我们还需要让事件多路分发器来监听其对应的事件，同时建立文件描述符、信号值与事件处理器之间的映射关系。这就要通过调用 evmap_io_add 和 evmap_signal_add 两个函数来完成。这两个函数相当于事件多路分发器中的 register_event 方法，它们由 evmap.c 文件实现，其中用到了一些重要数据结构：
#ifdef EVMAP_USE_HT
#include "ht-internal.h"
struct event_map_entry;
/*如果定义了EVMAP_USE_HT，则将event_io_map定义为哈希表。该哈希表存储 event_map_entry对象和I/O事件队列（见前文，具有同样文件描述符值的I/O事件处理器构成I/O事件队列）之间的映射关系，实际上也就是存储了文件描述符和I/O事件处理器之间 的映射关系*/
HT_HEAD(event_io_map, event_map_entry);
#else /*否则event_io_map和下面的event_signal_map一样*/
#define event_io_map event_signal_map
#endif /*下面这个结构体中的entries数组成员存储信号值和信号事件处理器之间的映射关系 （用信号值索引数组entries即得到对应的信号事件处理器）*/
struct event_signal_map
{void **entries; /*用于存放evmap_io或evmap_signal的数组*/int nentries; /*entries数组的大小*/
};
/*如果定义了EVMAP_USE_HT，则哈希表event_io_map中的成员具有如下类型*/
struct event_map_entry
{HT_ENTRY(event_map_entry)map_node;evutil_socket_t fd;union{struct evmap_io evmap_io;} ent;
};
/*event_list是由event组成的尾队列，前面讨论的所有事件队列都是这种类型*/
TAILQ_HEAD(event_list, event);
/*I/O事件队列（确切地说，evmap_io.events才是I/O事件队列）*/
struct evmap_io
{struct event_list events;ev_uint16_t nread;ev_uint16_t nwrite;
};
/*信号事件队列（确切地说，evmap_signal.events才是信号事件队列）*/
struct evmap_signal
{struct event_list events;
};
由于 evmap_io_add 和 evmap_signal_add 两个函数的逻辑基本相同，因此仅讨论 evmap_io_add 函数：
int evmap_io_add(struct event_base *base, evutil_socket_t fd, struct event *ev)
{/*获得event_base的后端I/O复用机制实例*/const struct eventop *evsel = base->evsel;/*获得event_base中文件描述符与I/O事件队列的映射表（哈希表或数组）*/struct event_io_map *io = &base->io;/*fd参数对应的I/O事件队列*/struct evmap_io *ctx = NULL;int nread, nwrite, retval = 0;short res = 0, old = 0;struct event *old_ev;EVUTIL_ASSERT(fd == ev->ev_fd);if (fd < 0)return 0;
#ifndef EVMAP_USE_HT/*I/O事件队列数组io.entries中，每个文件描述符占用一项。如果fd大于当前数组 的大小，则增加数组的大小（扩大后的数组的容量要大于fd）*/if (fd > = io->nentries){if (evmap_make_space(io, fd, sizeof(struct evmap_io *)) == -1)return (-1);}
#endif/*下面这个宏根据EVMAP_USE_HT是否被定义而有不同的实现，但目的都是创建ctx， 在映射表io中为fd和ctx添加映射关系*/GET_IO_SLOT_AND_CTOR(ctx, io, fd, evmap_io, evmap_io_init, evsel->fdinfo_len);nread = ctx->nread;nwrite = ctx->nwrite;if (nread)old |= EV_READ;if (nwrite)old |= EV_WRITE;if (ev->ev_events & EV_READ){if (++nread == 1)res |= EV_READ;}if (ev->ev_events & EV_WRITE){if (++nwrite == 1)res |= EV_WRITE;}if (EVUTIL_UNLIKELY(nread > 0xffff || nwrite > 0xffff)){event_warnx("Too many events reading or writing on fd%d", (int)fd);return -1;}if (EVENT_DEBUG_MODE_IS_ON() && (old_ev = TAILQ_FIRST(&ctx->events)) && (old_ev->ev_events & EV_ET) != (ev->ev_events & EV_ET)){event_warnx("Tried to mix edge-triggered and non-edge-triggered""events on fd%d",(int)fd);return -1;}if (res){void *extra = ((char *)ctx) + sizeof(struct evmap_io);/*往事件多路分发器中注册事件。add是事件多路分发器的接口函数之一。对不同的后 端I/O复用机制，这些接口函数有不同的实现。我们将在后面讨论事件多路分发器的接口函 数*/if (evsel->add(base, ev->ev_fd, old, (ev->ev_events & EV_ET) | res, extra) == -1)return (-1);retval = 1;}ctx->nread = (ev_uint16_t)nread;ctx->nwrite = (ev_uint16_t)nwrite;/*将ev插到I/O事件队列ctx的尾部。ev_io_next是定义在event-internal.h文件 中的宏：#define ev_io_next_ev.ev_io.ev_io_next*/TAILQ_INSERT_TAIL(&ctx->events, ev, ev_io_next);return (retval);
}
6. eventop 结构体
eventop 结构体封装了 I/O 复用机制必要的一些操作，比如注册事件、等待事件等。它为 event_base 支持的所有后端 I/O 复用机制提供了一个统一的接口。该结构体定义在 event-internal.h 文件中：
struct eventop
{ /*后端I/O复用技术的名称*/const char *name; /*初始化函数*/void *(*init)(struct event_base *); /*注册事件*/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); /*等待事件*/int (*dispatch)(struct event_base *, struct timeval *); /*释放I/O复用机制使用的资源*/void (*dealloc)(struct event_base *); /*程序调用fork之后是否需要重新初始化event_base*/int need_reinit; /*I/O复用技术支持的一些特性，可选如下3个值的按位或：EV_FEATURE_ET（支持边 沿触发事件EV_ET）、EV_FEATURE_O1（事件检测算法的复杂度是O(1)）和 EV_FEATURE_FDS（不仅能监听socket上的事件，还能监听其他类型的文件描述符上的事 件）*/enum event_method_feature features; /*有的I/O复用机制需要为每个I/O事件队列和信号事件队列分配额外的内存，以避免 同一个文件描述符被重复插入I/O复用机制的事件表中。evmap_io_add（或 evmap_io_del）函数在调用eventop的add（或del）方法时，将这段内存的起始地址作 为第5个参数传递给add（或del）方法。下面这个成员则指定了这段内存的长度*/size_t fdinfo_len;
};
前文提到，devpoll.c 、kqueue.c 、evport.c 、select.c 、win32select.c 、poll.c 和 epoll.c 文件分别针对不同的 I/O 复用技术实现了 eventop 定义的这套接口。那么，在支持多种 I/O 复用技术的系统上，Libevent 将选择使用哪个呢？这取决于这些 I/O 复用技术的优先级。Libevent 支持的后端 I/O 复用技术及它们的优先级在 event.c 文件中定义：
#ifdef_EVENT_HAVE_EVENT_PORTS extern const struct eventop evportops;
#endif #ifdef_EVENT_HAVE_SELECT extern const struct eventop selectops;
#endif #ifdef_EVENT_HAVE_POLL extern const struct eventop pollops;
#endif #ifdef_EVENT_HAVE_EPOLL extern const struct eventop epollops;
#endif #ifdef_EVENT_HAVE_WORKING_KQUEUE extern const struct eventop kqops;
#endif #ifdef_EVENT_HAVE_DEVPOLL extern const struct eventop devpollops;
#endif #ifdef WIN32 extern const struct eventop win32ops;
#endif static const struct eventop *eventops[] = {
#ifdef_EVENT_HAVE_EVENT_PORTS &evportops, #endif
#ifdef_EVENT_HAVE_WORKING_KQUEUE &kqops,
#endif #ifdef_EVENT_HAVE_EPOLL &epollops, #endif
#ifdef_EVENT_HAVE_DEVPOLL &devpollops,
#endif #ifdef_EVENT_HAVE_POLL &pollops,
#endif #ifdef_EVENT_HAVE_SELECT &selectops,
#endif #ifdef WIN32 &win32ops,
#endif NULL
};
Libevent 通过遍历 eventops 数组来选择其后端 I/O 复用技术。遍历的顺序是从数组的第一个元素开始，到最后一个元素结束。所以，在 Linux下 ，Libevent 默认选择的后端 I/O 复用技术是 epoll 。
7. event_base 结构体
结构体 event_base 是Libevent的Reactor 。它定义在 event-internal.h 文件中：
struct event_base
{/*初始化Reactor的时候选择一种后端I/O复用机制，并记录在如下字段中*/const struct eventop *evsel;/*指向I/O复用机制真正存储的数据，它通过evsel成员的init函数来初始化*/void *evbase;/*事件变化队列。其用途是：如果一个文件描述符上注册的事件被多次修改，则可以使 用缓冲来避免重复的系统调用（比如epoll_ctl）。它仅能用于时间复杂度为O(1)的I/O复 用技术*/struct event_changelist changelist;/*指向信号的后端处理机制，目前仅在singal.h文件中定义了一种处理方法*/const struct eventop *evsigsel;/*信号事件处理器使用的数据结构，其中封装了一个由socketpair创建的管道。它用 于信号处理函数和事件多路分发器之间的通信，这和我们在10.4节讨论的统一事件源的思路 是一样的*/struct evsig_info sig;/*添加到该event_base的虚拟事件、所有事件和激活事件的数量*/int virtual_event_count;int event_count;int event_count_active;/*是否执行完活动事件队列上剩余的任务之后就退出事件循环*/int event_gotterm;/*是否立即退出事件循环，而不管是否还有任务需要处理*/int event_break;/*是否应该启动一个新的事件循环*/int event_continue;/*目前正在处理的活动事件队列的优先级*/int event_running_priority;/*事件循环是否已经启动*/int running_loop;/*活动事件队列数组。索引值越小的队列，优先级越高。高优先级的活动事件队列中的 事件处理器将被优先处理*/struct event_list *activequeues;/*活动事件队列数组的大小，即该event_base一共有nactivequeues个不同优先级 的活动事件队列*/int nactivequeues;/*下面3个成员用于管理通用定时器队列*/struct common_timeout_list **common_timeout_queues;int n_common_timeouts;int n_common_timeouts_allocated;/*存放延迟回调函数的链表。事件循环每次成功处理完一个活动事件队列中的所有事件 之后，就调用一次延迟回调函数*/struct deferred_cb_queue defer_queue;/*文件描述符和I/O事件之间的映射关系表*/struct event_io_map io;/*信号值和信号事件之间的映射关系表*/struct event_signal_map sigmap;/*注册事件队列，存放I/O事件处理器和信号事件处理器*/struct event_list eventqueue;/*时间堆*/struct min_heap timeheap;/*管理系统时间的一些成员*/struct timeval event_tv;struct timeval tv_cache;
#if defined(_EVENT_HAVE_CLOCK_GETTIME) && defined(CLOCK_MONOTONIC)struct timeval tv_clock_diff;time_t last_updated_clock_diff;
#endif
/*多线程支持*/
#ifndef_EVENT_DISABLE_THREAD_SUPPORTunsigned long th_owner_id; /*当前运行该event_base的事件循环的线程*/void *th_base_lock; /*对event_base的独占锁*//*当前事件循环正在执行哪个事件处理器的回调函数*/struct event *current_event;/*条件变量（见第14章），用于唤醒正在等待某个事件处理完毕的线程*/void *current_event_cond;int current_event_waiters; /*等待current_event_cond的线程数*/
#endif #ifdef WIN32 struct event_iocp_port *iocp;
#endif /*该event_base的一些配置参数*/enum event_base_config_flag flags;/*下面这组成员变量给工作线程唤醒主线程提供了方法（使用socketpair创建的管 道）*/int is_notify_pending;evutil_socket_t th_notify_fd[2];struct event th_notify;int (*th_notify_fn)(struct event_base *base);
};
8. 事件循环
最后，讨论一下 Libevent 的“动力”，即事件循环。Libevent 中实现事件循环的函数是 event_base_loop 。该函数首先调用 I/O 事件多路分发器的事件监听函数，以等待事件；当有事件发生时，就依次处理之：
int event_base_loop(struct event_base *base, int flags)
{const struct eventop *evsel = base->evsel;struct timeval tv;struct timeval *tv_p;int res, done, retval = 0;EVBASE_ACQUIRE_LOCK(base, th_base_lock);/*一个event_base仅允许运行一个事件循环*/if (base->running_loop){event_warnx("%s:reentrant invocation.Only one event_base_loop""can run on each event_base at once.",__func__);EVBASE_RELEASE_LOCK(base, th_base_lock);return -1;}base->running_loop = 1; /*标记该event_base已经开始运行*/clear_time_cache(base); /*清除event_base的系统时间缓存*//*设置信号事件的event_base实例*/if (base->sig.ev_signal_added && base->sig.ev_n_signals_added)evsig_set_base(base);done = 0;
#ifndef_EVENT_DISABLE_THREAD_SUPPORTbase->th_owner_id = EVTHREAD_GET_ID();
#endifbase->event_gotterm = base->event_break = 0;while (!done){base->event_continue = 0;if (base->event_gotterm){break;}if (base->event_break){break;}timeout_correct(base, &tv); /*校准系统时间*/tv_p = &tv;if (!N_ACTIVE_CALLBACKS(base) && !(flags & EVLOOP_NONBLOCK)){/*获取时间堆上堆顶元素的超时值，即I/O复用系统调用本次应该设置的超时值*/timeout_next(base, &tv_p);}else{/*如果有就绪事件尚未处理，则将I/O复用系统调用的超时时间“置0”。这样I/O复用系 统调用直接返回，程序也就可以立即处理就绪事件了*/evutil_timerclear(&tv);}/*如果event_base中没有注册任何事件，则直接退出事件循环*/if (!event_haveevents(base) && !N_ACTIVE_CALLBACKS(base)){event_debug(("%s:no events registered.", __func__));retval = 1;goto done;}/*更新系统时间，并清空时间缓存*/gettime(base, &base->event_tv);clear_time_cache(base);/*调用事件多路分发器的dispatch方法等待事件，将就绪事件插入活动事件队列*/res = evsel->dispatch(base, tv_p);if (res == -1){event_debug(("%s:dispatch returned unsuccessfully.", __func__));retval = -1;goto done;}update_time_cache(base);/*将时间缓存更新为当前系统时间*/ /*检查时间堆上的到期事件并依次执行之*/timeout_process(base);if (N_ACTIVE_CALLBACKS(base)){/*调用event_process_active函数依次处理就绪的信号事件和I/O事件*/int n = event_process_active(base);if ((flags & EVLOOP_ONCE) && N_ACTIVE_CALLBACKS(base) == 0 && n != 0)done = 1;}else if (flags & EVLOOP_NONBLOCK)done = 1;}event_debug(("%s:asked to terminate loop.", __func__));
done:/*事件循环结束，清空时间缓存，并设置停止循环标志*/clear_time_cache(base);base->running_loop = 0;EVBASE_RELEASE_LOCK(base, th_base_lock);return (retval);
}
更多推荐
Linux 网络编程学习笔记——十二、高性能 I/O 框架库 Libevent
发布评论