我们知道libevent是一个Reactor模式的事件驱动的网络库。
到目前为止，我们已经看了核心的event和event_base结构体的源码，看了初始化这两个结构体的源码，看了注册event的源码，也将event注册到I/O多路复用监听的事件上了。现在准备工作都做好了，下面就是看运行时的主循环了，在这个主循环中，是如何检测事件、分发事件、调用事件的回调函数的。这一步就是libevent的核心框架流程了。
Reactor模式中的Event、Event Handler、Reactor目前都完成了，下面就剩Event Demultiplexer了。
这一步通过event_base_dispatch()完成
int
event_base_dispatch(struct event_base *event_base)
{
return (event_base_loop(event_base, 0)); //调用event_base_loop()
}
可以看到，该函数只是做了调用event_base_loop()这一个动作，所以工作实际是在函数event_base_loop()内完成的。
event_base_loop()
该函数完成以下工作：
1.信号标记被设置，则调用信号的回调函数
2.根据定时器最小时间，设置I/O多路复用的最大等待时间，这样即使没有I/O事件发生，也能在最小定时器超时时返回。
3.调用I/O多路复用，监听事件，将活跃事件添加到活跃事件链表中
4.检查定时事件，将就绪的定时事件从小根堆中删除，插入到活跃事件链表中
5.对event_base的活跃事件链表中的事件，调用event_process_active(）函数，在该函数内调用event的回调函数，优先级高的event先处理。
该函数内部调用了eventop.dispatch()监听事件，event_sigcb函数指针处理信号事件，timeout_process()将超时的定时事件加入到活跃事件链表中，event_process_active()处理活跃事件链表中的事件，调用相应的回调函数。
int
event_base_loop(struct event_base *base, int flags)
{
const struct eventop *evsel = base->evsel;
void *evbase = base->evbase; //event_base的I/O多路复用
struct timeval tv;
struct timeval *tv_p;
int res, done;
/* clear time cache */
//清空时间缓存
base->tv_cache.tv_sec = 0;
//处理Signal事件时，指定信号所属的event_base
if (base->sig.ev_signal_added)
evsignal_base = base;
done = 0;
while (!done) { //进入事件主循环
/* Terminate the loop if we have been asked to */
//设置event_base的标记，以表明是否需要跳出循环
if (base->event_gotterm) { //event_loopexit_cb()可设置
base->event_gotterm = 0;
break;
}
if (base->event_break) { //event_base_loopbreak()可设置
base->event_break = 0;
break;
}
/* You cannot use this interface for multi-threaded apps */
//当event_gotsig被设置时，则event_sigcb就是信号处理的回调函数
while (event_gotsig) {
event_gotsig = 0;
if (event_sigcb) {
res = (*event_sigcb)(); //调用信号处理的回调函数
if (res == -1) {
errno = EINTR;
return (-1);
}
}
}
timeout_correct(base, &tv); //校准时间
tv_p = &tv;
//根据定时器堆中最小超时时间计算I/O多路复用的最大等待时间tv_p
if (!base->event_count_active && !(flags & EVLOOP_NONBLOCK)) {
timeout_next(base, &tv_p);
} else {
/*
* if we have active events, we just poll new events
* without waiting.
*/
evutil_timerclear(&tv);
}
/* If we have no events, we just exit */
//没有注册事件，则退出
if (!event_haveevents(base)) {
event_debug(("%s: no events registered.", __func__));
return (1);
}
/* update last old time */
gettime(base, &base->event_tv);
/* clear time cache */
base->tv_cache.tv_sec = 0;
//调用I/O多路复用，监听事件
res = evsel->dispatch(base, evbase, tv_p);
if (res == -1)
return (-1);
//将time cache赋值为当前系统时间
gettime(base, &base->tv_cache);
//检查定时事件，将就绪的定时事件从小根堆中删除，插入到活跃事件链表中
timeout_process(base);
if (base->event_count_active) {
//处理event_base的活跃链表中的事件
//调用event的回调函数，优先级高的event先处理
event_process_active(base);
if (!base->event_count_active && (flags & EVLOOP_ONCE))
done = 1;
} else if (flags & EVLOOP_NONBLOCK)
done = 1;
}
/* clear time cache */
//循环结束，清空时间缓存
base->tv_cache.tv_sec = 0;
event_debug(("%s: asked to terminate loop.", __func__));
return (0);
}
epoll_dispatch()
在上面我们看到，event_base_loop()中通过I/O多路复用的dispatch()函数完成监听事件功能。在之前的event_init()中我们看到，通过遍历eventops数组，从中选择一个I/O多路复用机制，所以不同的I/O多路复用机制有不同的eventop结构体，相应的也就有不同的dispatch()函数。下面，再次看下eventop结构体(event-internal.h)
struct eventop {
const char *name;
void *(*init)(struct event_base *); //初始化
int (*add)(void *, struct event *); //注册事件
int (*del)(void *, struct event *); //删除事件
int (*dispatch)(struct event_base *, void *, struct timeval *); //事件分发
void (*dealloc)(struct event_base *, void *); //注销，释放资源
/* set if we need to reinitialize the event base */
int need_reinit;
};
在event_add()中通过add()成员函数注册event到监听事件中，现在在event_base_loop()中通过dispatch()成员函数监听事件。
libevent支持多种I/O多路复用机制，下面先看下epoll的eventop结构体(epoll.c)
const struct eventop epollops = {
"epoll",
epoll_init,
epoll_add,
epoll_del,
epoll_dispatch,
epoll_dealloc,
1 /* need reinit */
};
然后看下epoll的dispatch()函数(epoll.c)
从下面源码可见，epoll_dispatch()的工作主要有：
1.调用epoll_wait()监听事件
2.如果有信号发生，调用evsignal_process()处理信号
3.将活跃的event根据其活跃的类型注册到活跃事件链表上
4.如果events数组大小不够，则重新分配为原来2倍大小
static int
epoll_dispatch(struct event_base *base, void *arg, struct timeval *tv)
{
struct epollop *epollop = arg;
struct epoll_event *events = epollop->events;
struct evepoll *evep;
int i, res, timeout = -1;
if (tv != NULL)
timeout = tv->tv_sec * 1000 + (tv->tv_usec + 999) / 1000; //转换为微米
if (timeout > MAX_EPOLL_TIMEOUT_MSEC) { //设置最大超时时间
/* Linux kernels can wait forever if the timeout is too big;
* see comment on MAX_EPOLL_TIMEOUT_MSEC. */
timeout = MAX_EPOLL_TIMEOUT_MSEC;
}
res = epoll_wait(epollop->epfd, events, epollop->nevents, timeout); //监听事件发生
if (res == -1) {
if (errno != EINTR) {
event_warn("epoll_wait");
return (-1);
}
evsignal_process(base); //由于Signal事件发生中断，处理Signal事件
return (0);
} else if (base->sig.evsignal_caught) {
evsignal_process(base); //有Signal事件发生，处理Signal事件
}
event_debug(("%s: epoll_wait reports %d", __func__, res));
for (i = 0; i < res; i++) { //处理活跃事件
int what = events[i].events; //活跃类型
struct event *evread = NULL, *evwrite = NULL;
int fd = events[i].data.fd; //event的文件描述符
if (fd < 0 || fd >= epollop->nfds)
continue;
evep = &epollop->fds[fd];
if (what & (EPOLLHUP|EPOLLERR)) { //判断epoll的events类型，并找到注册的event
evread = evep->evread;
evwrite = evep->evwrite;
} else {
if (what & EPOLLIN) {
evread = evep->evread;
}
if (what & EPOLLOUT) {
evwrite = evep->evwrite;
}
}
if (!(evread||evwrite))
continue;
//添加event到活跃事件链表中
if (evread != NULL)
event_active(evread, EV_READ, 1);
if (evwrite != NULL)
event_active(evwrite, EV_WRITE, 1);
}
//如果注册的事件全部变为活跃，则增大events数组为原来两倍
if (res == epollop->nevents && epollop->nevents < MAX_NEVENTS) {
/* We used all of the event space this time. We should
be ready for more events next time. */
int new_nevents = epollop->nevents * 2;
struct epoll_event *new_events;
new_events = realloc(epollop->events,
new_nevents * sizeof(struct epoll_event));
if (new_events) {
epollop->events = new_events;
epollop->nevents = new_nevents;
}
}
return (0);
}
event_process_active()
好了，现在活跃的I/O事件、定时器事件已经全部添加到活跃事件链表中了。下面就开始调用这些event的回调函数进行处理了，这步是在event_base_loop()中调用event_process_active()来完成的。
该函数从event_base的activequeueus链表数组上取出一个链表；对该链表上的event调用回调函数；优先调用优先级值最小的event
/*
* Active events are stored in priority queues. Lower priorities are always
* process before higher priorities. Low priority events can starve high
* priority ones.
*/
static void
event_process_active(struct event_base *base)
{
struct event *ev;
struct event_list *activeq = NULL;
int i;
short ncalls;
for (i = 0; i < base->nactivequeues; ++i) { //取出第一个活跃链表
if (TAILQ_FIRST(base->activequeues[i]) != NULL) {
activeq = base->activequeues[i];
break;
}
}
assert(activeq != NULL);
//优先处理优先级值最小的event
for (ev = TAILQ_FIRST(activeq); ev; ev = TAILQ_FIRST(activeq)) {
if (ev->ev_events & EV_PERSIST)
event_queue_remove(base, ev, EVLIST_ACTIVE); //是持久事件，则从活跃链表移除
else
event_del(ev); //不是持久事件，则直接删除该事件
/* Allows deletes to work */
ncalls = ev->ev_ncalls;
ev->ev_pncalls = &ncalls;
while (ncalls) {
ncalls--;
ev->ev_ncalls = ncalls;
//调用该event的回调函数，event.ev_res保存返回值
(*ev->ev_callback)((int)ev->ev_fd, ev->ev_res, ev->ev_arg);
if (event_gotsig || base->event_break) {
ev->ev_pncalls = NULL;
return;
}
}
ev->ev_pncalls = NULL;
}
}