libevent源码学习_event_test

小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

对应的sample文件中提供了event_test.c，里面就是关于事件的简单示例，具体如下：

/*
* Compile with:
* cc -I/usr/local/include -o event-test event-test.c -L/usr/local/lib -levent
*/ #ifdef HAVE_CONFIG_H
#include "config.h"
#endif #include <sys/types.h>
#include <sys/stat.h>
#include <sys/queue.h>
#include <unistd.h>
#include <sys/time.h>
#include <fcntl.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h> #include <event.h> static void
fifo_read(int fd, short event, void *arg)
{
char buf[];
int len;
struct event *ev = arg; /* Reschedule this event */
event_add(ev, NULL); fprintf(stderr, "fifo_read called with fd: %d, event: %d, arg: %p\n",
fd, event, arg); len = read(fd, buf, sizeof(buf) - ); if (len == -) {
perror("read");
return;
} else if (len == ) {
fprintf(stderr, "Connection closed\n");
return;
} buf[len] = '\0'; fprintf(stdout, "Read: %s\n", buf);
} int
main (int argc, char **argv)
{
struct event evfifo; struct stat st;
const char *fifo = "event.fifo";
int socket; if (lstat (fifo, &st) == ) {
if ((st.st_mode & S_IFMT) == S_IFREG) {
errno = EEXIST;
perror("lstat");
exit ();
}
} unlink (fifo);
if (mkfifo (fifo, ) == -) {
perror("mkfifo");
exit ();
} socket = open (fifo, O_RDWR | O_NONBLOCK, ); if (socket == -) {
perror("open");
exit ();
} fprintf(stderr, "Write data to %s\n", fifo); /* Initalize the event library */
event_init(); /* Initalize one event */
event_set(&evfifo, socket, EV_READ, fifo_read, &evfifo); /* Add it to the active events, without a timeout */
event_add(&evfifo, NULL); event_dispatch(); return ();
}

从这个例子中，我们可以看到使用libevent的基本步骤：

event_init --> event_set --> event_add --> event_dispatch

下面分步来解析这些函数。

1、函数event_init()

位于：event.c

struct event_base *
event_init(void)
{
struct event_base *base = event_base_new(); if (base != NULL)
current_base = base; return (base);
}

比较简单，就是调用了函数event_base_new初始化一个event_base类型的变量，并且将这个变量赋值给一个全局变量current_base。

2、函数event_base_new()

位于：event.c

struct event_base *
event_base_new(void)
{
int i;
struct event_base *base; /*
* 在堆上分配内存存储event_base，所有字段初始化为0
*/
if ((base = calloc(, sizeof(struct event_base))) == NULL)
event_err(, "%s: calloc", __func__); /*
* 设置use_monotonic变量
*/
detect_monotonic(); /*
* 得到当前时间
*/
gettime(base, &base->event_tv); /*
* 初始化小根堆
*/
min_heap_ctor(&base->timeheap); /*
* 初始化注册时间队列
*/
TAILQ_INIT(&base->eventqueue); /*
* 初始化socketpair
*/
base->sig.ev_signal_pair[] = -;
base->sig.ev_signal_pair[] = -; /*
* C语言实现多态
* 根据所支持的系统调用进行对应的初始化
*/
base->evbase = NULL;
for (i = ; eventops[i] && !base->evbase; i++) {
base->evsel = eventops[i]; base->evbase = base->evsel->init(base);
} if (base->evbase == NULL)
event_errx(, "%s: no event mechanism available", __func__); if (evutil_getenv("EVENT_SHOW_METHOD"))
event_msgx("libevent using: %s\n",
base->evsel->name); /*
* 设置优先级base->nactivequeues，分配数组base->activequeues
* 数组大小和优先级相同
*/
/* allocate a single active event queue */
event_base_priority_init(base, ); return (base);
}

主要做了以下事情：

（1）给event_base类型变量分配空间

（2）初始化小根堆【struct min_heap timeheap，位于位于结构体event_base】

（3）初始化注册事件队列【struct event_list eventqueue，位于位于结构体event_base】

（4）根据系统支持的系统调用初始化后面真正干活的eventop实例对象【void *evbase，位于结构体event_base】

（5）调用函数event_base_priority_init() 初始化优先队列，确认的说应该是活跃事件的队列，它是带优先级的，因为这里是最开始的初始化，所以就初始化一个队列，并且它的优先级为1，这个优先级就作为初始化队列的数量。具体见下面函数event_base_priority_init() 的分析

上面的init以epoll为例，位于：epoll.c

static void *
epoll_init(struct event_base *base)
{
int epfd;
struct epollop *epollop; /* Disable epollueue when this environment variable is set */
if (evutil_getenv("EVENT_NOEPOLL"))
return (NULL); /* Initalize the kernel queue */
if ((epfd = epoll_create()) == -) {
if (errno != ENOSYS)
event_warn("epoll_create");
return (NULL);
} FD_CLOSEONEXEC(epfd); if (!(epollop = calloc(, sizeof(struct epollop))))
return (NULL); epollop->epfd = epfd; /* Initalize fields */
epollop->events = malloc(INITIAL_NEVENTS * sizeof(struct epoll_event));
if (epollop->events == NULL) {
free(epollop);
return (NULL);
}
epollop->nevents = INITIAL_NEVENTS; epollop->fds = calloc(INITIAL_NFILES, sizeof(struct evepoll));
if (epollop->fds == NULL) {
free(epollop->events);
free(epollop);
return (NULL);
}
epollop->nfds = INITIAL_NFILES; evsignal_init(base); return (epollop);
}

调用了系统调用epoll_create，创建出ep_fd，然后初始化了结构体epollop的成员变量。

struct epollop {
struct evepoll *fds;
int nfds;
struct epoll_event *events;
int nevents;
int epfd;
};

struct evepoll {
struct event *evread;
struct event *evwrite;
};

3、函数event_base_priority_init()

位于：event.c

int
event_base_priority_init(struct event_base *base, int npriorities)
{
int i; /*
* 当前base上有活跃的events则不能设置优先级，返回
*/
if (base->event_count_active)
return (-); /*
* 不同，则先释放原先的activequeues数组
*/
if (base->nactivequeues && npriorities != base->nactivequeues) {
for (i = ; i < base->nactivequeues; ++i) {
free(base->activequeues[i]);
}
free(base->activequeues);
} /*
* 设置新的优先级
* 设置和优先级值相同大小的event_list数组
*/
/* Allocate our priority queues */
base->nactivequeues = npriorities;
base->activequeues = (struct event_list **)
calloc(base->nactivequeues, sizeof(struct event_list *)); if (base->activequeues == NULL)
event_err(, "%s: calloc", __func__); /*
* 初始化activequeues数组中每个元素
*/
for (i = ; i < base->nactivequeues; ++i) {
base->activequeues[i] = malloc(sizeof(struct event_list));
if (base->activequeues[i] == NULL)
event_err(, "%s: malloc", __func__);
TAILQ_INIT(base->activequeues[i]);
} return ();
}

初始化结构体event_base里面的struct event_list **activequeues成员，这是个2维数组，其中的元素activequeues[priority]是一个链表，这个链表里面对应的是相同优先级的事件。

为了更清晰上面的初始化，附上UML图如下：

4、函数event_set()

位于：event.c

void
event_set(struct event *ev, int fd, short events,
void (*callback)(int, short, void *), void *arg)
{
/* Take the current base - caller needs to set the real base later */
ev->ev_base = current_base; ev->ev_callback = callback;
ev->ev_arg = arg;
ev->ev_fd = fd;
ev->ev_events = events;
ev->ev_res = ;
ev->ev_flags = EVLIST_INIT;
ev->ev_ncalls = ;
ev->ev_pncalls = NULL; min_heap_elem_init(ev); /* by default, we put new events into the middle priority */
if(current_base)
ev->ev_pri = current_base->nactivequeues/;
}

5、函数event_add()

位于：event.c

int
event_add(struct event *ev, const struct timeval *tv)
{
/*
* 要注册到的event_base
* 得到ev对应的反应堆实例event_base
*/
struct event_base *base = ev->ev_base;
const struct eventop *evsel = base->evsel;
void *evbase = base->evbase;
int res = ; event_debug((
"event_add: event: %p, %s%s%scall %p",
ev,
ev->ev_events & EV_READ ? "EV_READ " : " ",
ev->ev_events & EV_WRITE ? "EV_WRITE " : " ",
tv ? "EV_TIMEOUT " : " ",
ev->ev_callback)); assert(!(ev->ev_flags & ~EVLIST_ALL)); /*
* ev->ev_events表示事件类型
* 如果ev->ev_events是读/写/信号事件,而且ev不在已注册队列或已就绪队列
* 那么调用evbase注册ev事件
*/
if ((ev->ev_events & (EV_READ|EV_WRITE|EV_SIGNAL)) &&
!(ev->ev_flags & (EVLIST_INSERTED|EVLIST_ACTIVE))) {
res = evsel->add(evbase, ev);
if (res != -)
/*
* 注册成功，插入event到已注册链表中
*/
event_queue_insert(base, ev, EVLIST_INSERTED);
} return (res);
}

只留下了关键逻辑，精简了和定时器相关的代码。

上面的add以epoll为例，位于：epoll.c

static int
epoll_add(void *arg, struct event *ev)
{
struct epollop *epollop = arg;
struct epoll_event epev = {, {}};
struct evepoll *evep;
int fd, op, events; if (ev->ev_events & EV_SIGNAL)
return (evsignal_add(ev)); fd = ev->ev_fd;
if (fd >= epollop->nfds) {
/* Extent the file descriptor array as necessary */
if (epoll_recalc(ev->ev_base, epollop, fd) == -)
return (-);
} /*
* 获得地址，后面给它赋值
*/
evep = &epollop->fds[fd];
op = EPOLL_CTL_ADD;
events = ; /*
* 如果原先存在就EPOLL_CTL_MOD而不是EPOLL_CTL_ADD
*/
if (evep->evread != NULL) {
events |= EPOLLIN;
op = EPOLL_CTL_MOD;
}
if (evep->evwrite != NULL) {
events |= EPOLLOUT;
op = EPOLL_CTL_MOD;
} /*
* 设置关注的事件
*/
if (ev->ev_events & EV_READ)
events |= EPOLLIN;
if (ev->ev_events & EV_WRITE)
events |= EPOLLOUT; epev.data.fd = fd;
epev.events = events;
if (epoll_ctl(epollop->epfd, op, ev->ev_fd, &epev) == -)
return (-); /* Update events responsible */
if (ev->ev_events & EV_READ)
evep->evread = ev;
if (ev->ev_events & EV_WRITE)
evep->evwrite = ev; return ();
}

如果新加入的fd大小大于了之前分配的fd最大个数，则需要调用函数epoll_recalc()重新分配空间，否则就是更新相关的结构体变量，并调用系统调用epoll_ctl来EPOLL_CTL_ADD或EPOLL_CTL_MOD对应的事件。

函数epoll_recalc()解析如下：

static int
epoll_recalc(struct event_base *base, void *arg, int max)
{
struct epollop *epollop = arg; /*
* 当前的fd大于了之前根据最大fd分配的结构体evepoll个数，重新分配，否则直接返回
*/
if (max >= epollop->nfds) {
struct evepoll *fds;
int nfds; /*
* 每次以2倍大小扩充
*/
nfds = epollop->nfds;
while (nfds <= max)
nfds <<= ; /*
* 扩充
*/
fds = realloc(epollop->fds, nfds * sizeof(struct evepoll));
if (fds == NULL) {
event_warn("realloc");
return (-);
} /*
* 更新成员变量的值，并且把新扩充的内存清空
*/
epollop->fds = fds;
memset(fds + epollop->nfds, ,
(nfds - epollop->nfds) * sizeof(struct evepoll));
epollop->nfds = nfds;
} return ();
}

在上面的add完毕后，要把对应的事件放到已注册事件的链表里面。

void
event_queue_insert(struct event_base *base, struct event *ev, int queue)
{
/*
* ev可能已经在激活列表中了，避免重复插入
*/
if (ev->ev_flags & queue) {
/* Double insertion is possible for active events */
if (queue & EVLIST_ACTIVE)
return; event_errx(, "%s: %p(fd %d) already on queue %x", __func__,
ev, ev->ev_fd, queue);
} if (~ev->ev_flags & EVLIST_INTERNAL)
base->event_count++; /*
* 记录queue标记
*/
ev->ev_flags |= queue;
switch (queue) {
/*
* I/O或Signal事件，加入已注册事件链表
*/
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: {
min_heap_push(&base->timeheap, ev);
break;
}
default:
event_errx(, "%s: unknown queue %x", __func__, queue);
}
}

6、函数event_dispatch()

位于：event.c

int
event_dispatch(void)
{
return (event_loop());
}

/* not thread safe */ int
event_loop(int flags)
{
return event_base_loop(current_base, flags);
}

可以看到，用了全局变量current_base，所以它并不是线程安全的。

int
event_base_loop(struct event_base *base, int flags)
{
const struct eventop *evsel = base->evsel;
void *evbase = base->evbase;
struct timeval tv;
struct timeval *tv_p;
int res, done; /* clear time cache */
base->tv_cache.tv_sec = ; done = ;
while (!done) {
/*
* 校正系统时间，如果系统使用的是非MONOTONIC时间，用户可能会向后调整了系统时间
* 在timeout_correct函数里，比较last wait time和当前时间
* 如果当前时间< last wait time 表明时间有问题
* 这时需要更新timer_heap中所有定时事件的超时时间。
*/
timeout_correct(base, &tv); /*
* 根据timer heap中事件的最小超时时间，计算系统I/O demultiplexer的最大等待时间
*/
tv_p = &tv;
if (!base->event_count_active && !(flags & EVLOOP_NONBLOCK)) {
timeout_next(base, &tv_p);
} else {
/*
* 依然有未处理的就绪事件，就让I/O demultiplexer立即返回，不必等待
* 下面会提到，在libevent中，低优先级的就绪事件可能不能立即被处理
*/
/*
* 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 ();
} /* update last old time */
gettime(base, &base->event_tv); /* clear time cache */
base->tv_cache.tv_sec = ; /*
* 调用系统I/O demultiplexer等待就绪I/O events，可能是epoll_wait，或者select等；
* 在evsel->dispatch()中，会把就绪signal event、I/O event插入到激活链表中
*/
res = evsel->dispatch(base, evbase, tv_p); if (res == -)
return (-); /*
* 将time cache赋值为当前系统时间
*/
gettime(base, &base->tv_cache); /*
* 检查heap中的timer events
* 将就绪的timer event从heap上删除，并插入到激活链表中
*/
timeout_process(base); /*
* 调用event_process_active()处理激活链表中的就绪event，调用其回调函数执行事件处理
* 该函数会寻找最高优先级（priority值越小优先级越高）的激活事件链表，
* 然后处理该链表中的所有就绪事件；
* 因此低优先级的就绪事件可能得不到及时处理
*
* 见函数event_process_active()的注释:
* Active events are stored in priority queues. Lower priorities are always
* process before higher priorities. Low priority events can starve high
* priority ones.
*
* 在函数event_process_active()里面就是寻找priority值最小的已就绪事件队列
* 找到一个就开始处理里面所有的事件回调了，其他的队列根本就不管了......
* 所以原作者用的Low priority events can starve high priority ones非常贴切
*/
if (base->event_count_active) {
/*
* 处理event_base的活跃链表中的事件
* 调用event的回调函数，优先级高的event先处理
*/
event_process_active(base); if (!base->event_count_active && (flags & EVLOOP_ONCE)) {
done = ;
}
} else if (flags & EVLOOP_NONBLOCK) {
done = ;
}
} /* clear time cache */
base->tv_cache.tv_sec = ; event_debug(("%s: asked to terminate loop.", __func__));
return ();
}

1、去掉了一些无用的业务逻辑代码。

2、调用epoll_dispatch来进行事件的分发，激活就绪的事件都弄到激活的链表里面去。

3、调用timeout_process把超时的事件也弄到激活的链表里面去。

4、调用event_process_active开始处理，调用对应事件的回调函数；需要注意的是：就是寻找priority值最小的已就绪事件队列，找到一个就开始处理里面所有的事件回调了，其他的队列根本就不管了......所以原作者用的Low priority events can starve high priority ones非常贴切。

5、关于定时器、时间相关的函数未仔细看。

函数epoll_dispatch()，位于epoll.c

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 = -; if (tv != NULL)
timeout = tv->tv_sec * + (tv->tv_usec + ) / ; 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 == -) {
if (errno != EINTR) {
event_warn("epoll_wait");
return (-);
} evsignal_process(base);
return ();
} else if (base->sig.evsignal_caught) {
evsignal_process(base);
} event_debug(("%s: epoll_wait reports %d", __func__, res)); for (i = ; i < res; i++) {
int what = events[i].events;
struct event *evread = NULL, *evwrite = NULL;
int fd = events[i].data.fd; if (fd < || fd >= epollop->nfds)
continue;
evep = &epollop->fds[fd]; if (what & (EPOLLHUP|EPOLLERR)) {
evread = evep->evread;
evwrite = evep->evwrite;
} else {
if (what & EPOLLIN) {
evread = evep->evread;
} if (what & EPOLLOUT) {
evwrite = evep->evwrite;
}
} if (!(evread||evwrite))
continue; if (evread != NULL)
event_active(evread, EV_READ, );
if (evwrite != NULL)
event_active(evwrite, EV_WRITE, );
} 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 * ;
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 ();
}

函数event_process_active()

static void
event_process_active(struct event_base *base)
{
struct event *ev;
struct event_list *activeq = NULL;
int i;
short ncalls; /*
* 寻找最高优先级(priority值越小优先级越高)的已就绪事件队列
*/
for (i = ; i < base->nactivequeues; ++i) {
if (TAILQ_FIRST(base->activequeues[i]) != NULL) {
activeq = base->activequeues[i];
break;
}
} assert(activeq != NULL); for (ev = TAILQ_FIRST(activeq); ev; ev = TAILQ_FIRST(activeq)) {
/*
* 如果有persist标志,则只从激活队列中移除此事件
*/
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;
/*
* 回调
*/
(*ev->ev_callback)((int)ev->ev_fd, ev->ev_res, ev->ev_arg);
if (base->event_break)
return;
}
}
}

至此，看了2个下午的libevent事件处理流程收官！

event-test.c例子中使用一个命名管道（也被称为FIFO文件），它通过读的方式打开一个命名管道，并且监听这个命名管道是否有数据可读，当有数据可读时会执行fifo_read函数，把读取的内容打印出来。

可以搞一个往这个命名管道写内容的简单的程序，进行测试：

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/queue.h>
#include <sys/time.h>
#include <fcntl.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <errno.h> int main(int argc, char **argv)
{
char *input = argv[];
if (argc != )
{
input = "hello";
}
int fd ;
fd = open("event.fifo",O_WRONLY);
if(fd == -){
perror("open error");
exit(EXIT_FAILURE);
} write(fd, input, strlen(input));
close(fd);
printf("write success\n");
return ;
}