libevent源码浅析: http库



libevent自带了一个http库,用它可以很简单的实现一个http服务器,本文非常简单地分析之.evhttp

evhttp库有几个主要的结构体,它们之间的联系非常龌龊:

其中,结构体event, min_heap, evsignal_info, eventop, event_base在前面几篇文章中已经介绍过了,这里不再啰嗦.

evbuffer

evbuffer用于读或写缓冲,图示为:

和evbuffer有关的外露接口主要是:

1.//从文件读数据到缓冲,读取量为max(howmuch, 4096)
2.int evbuffer_read(struct evbuffer *buf, int fd, int howmuch);
3.
4.//把缓冲写出文件
5.int evbuffer_write(struct evbuffer *buffer, int fd)

evbuffer比较简单,不多介绍.

evhttp, evhttp_connection, evhttp_request

libevent对成员的命名不太在意,其实evhttp可以看做是echttpsever,它绑定到某个特定端口和地址(socket(), bind()),保存访问该server的连接(通过成员connections,).evhttp_connection是保存连接信息的结构体, evhttp_request表示请求.

看看http库的使用流程:

01.void http_handler(struct evhttp_request *req, void *arg)
02.{
03.struct evbuffer *buf;
04.buf = evbuffer_new();
05.
06.// 分析请求
07.char *decode_uri = strdup((char*) evhttp_request_uri(req));
08.struct evkeyvalq http_query;
09.evhttp_parse_query(decode_uri, &http_query);
10.free(decode_uri);
11.
12.// 从http头中获取参数
13.const char *request_value = evhttp_find_header(&http_query, "data");
14.
15.// 返回HTTP头部
16.evhttp_add_header(req->output_headers, "Content-Type", "text/html; charset=UTF-8");
17.evhttp_add_header(req->output_headers, "Server", "my_httpd");
18.//evhttp_add_header(req->output_headers, "Connection", "keep-alive");
19.
20.evhttp_add_header(req->output_headers, "Connection", "close");
21.
22.// 将要输出的值写入输出缓存
23.if(request_value != NULL) {
24.evbuffer_add_printf(buf, "%s", request_value);
25.} else {
26.evbuffer_add_printf(buf, "%s", "no error.");
27.}
28.
29.// 输出
30.evhttp_send_reply(req, HTTP_OK, "OK", buf);
31.
32.// 内存释放
33.evhttp_clear_headers(&http_query);
34.evbuffer_free(buf);
35.}
36.
37.int main(int argc, char **argv)
38.{
39.char *host_ip = "0.0.0.0";
40.int host_port = 8080;
41.int timeout = 3;
42.
43.struct evhttp *httpd;
44.
45.event_init();
46.
47.//根据host_ip和host_port创建一个addrinfo结构体,然后创建一个socket,绑定到这个socket后,
48.//根据这些信息得到得到一个event(回调函数设置为accept_socket),然后将这个event关联到对应的event_base,
49.//之后插入到&http->sockets队列中,然后返回&http
50.httpd = evhttp_start(host_ip, host_port);
51.
52.if (httpd == NULL) {
53.fprintf(stderr, "Error: Unable to listen on %s:%d\n\n", host_ip, host_port);
54.exit(1);
55.}
56.
57.// 设置请求超时时间
58.evhttp_set_timeout(httpd, timeout);
59.
60.// 设置请求的处理函数
61.evhttp_set_gencb(httpd, http_handler, NULL);
62.
63.event_dispatch();
64.
65.evhttp_free(httpd);
66.
67.return 0;
68.}

[1] 首先看看evhttp_start():

1.//创建一个evhttp,绑定到端口和地址
2.struct evhttp * evhttp_start(const char *address, u_short port)
3.{
4.struct evhttp *http = evhttp_new_object();
5.evhttp_bind_socket(http, address, port);
6.return (http);
7.}

函数evhttp_bind_socket()代码如下:

01.//根据address和port创建一个非阻塞的socket,
02.//将其bind后的fd创建一个event(在这里设置好回调函数)后添加到&http->sockets
03.int evhttp_bind_socket(struct evhttp *http, const char *address, u_short port)
04.{
05.int fd;
06.int res;
07.
08.//绑定一个socket
09.fd = bind_socket(address, port, 1 /*reuse*/);
10.
11.//根据fd创建一个event,设置好回调函数,
12.//然后将这个event关联到对应的event_base,并将它插入到&http->sockets中.
13.res = evhttp_accept_socket(http, fd);
14.
15.return (res);
16.}

在这里,函数bing_socket()的作用是根据地址和端口创建一个socket,返回bind()后的文件描述符.函数evhttp_accept_socket()的作用在注释中也说明了,其代码如下:

01.int evhttp_accept_socket(struct evhttp *http, int fd)
02.{
03.struct evhttp_bound_socket *bound;
04.struct event *ev;
05.int res;
06.
07.bound = malloc(sizeof(struct evhttp_bound_socket));
08.ev = &bound->bind_ev;
09.
10./* Schedule the socket for accepting */
11.//设置这个ev,回调函数为accept_socket,针对的文件描述符为fd
12.event_set(ev, fd, EV_READ | EV_PERSIST, accept_socket, http);
13.
14.//将ev关联到&http->base
15.EVHTTP_BASE_SET(http, ev);
16.
17.//将ev添加进&http->base
18.res = event_add(ev, NULL);
19.
20.//将bound插入到&http->sockets
21.TAILQ_INSERT_TAIL(&http->sockets, bound, next);
22.}

需要指出的是,在这个函数中,struct event *ev可以看成是服务器struct evhttp的代理,evhttp通过这个ev是否可读来注意到是否有新的连接.(后文会分析.)

[2] 函数evhttp_set_timeout()和evhttp_set_gencb()逻辑比较简单,分别设置超时时间和回调函数.

[3]重头戏来了,函数event_dispatch()负责分发,在前面的文章已经介绍过了,它最终会调用event_base_loop(),分别查看定时器最小堆,信号队列和I/O队列.在http库中,当有一个新的连接时,[1]中已加入到event_base已注册事件队列的事件ev->fd将变成可读,它被移入已就绪事件队列,然后由函数event_process_active()调用ev的回调函数accept_socket()(回调函数在evhttp_accept_socket()函数中设置).

需要说明的是,以下的内容都是在event_base_loop()死循环中被处理的.

现在看一下回调函数accept_socket()的代码:

01.//作为回调函数,accept 一个 socket
02.static void accept_socket(int fd, short what, void *arg)
03.{
04.struct evhttp *http = arg;
05.struct sockaddr_storage ss;
06.socklen_t addrlen = sizeof(ss);
07.int nfd;
08.
09.//获得accept()后的文件描述符
10.nfd = accept(fd, (struct sockaddr *)&ss, &addrlen);
11.
12.//设置为非阻塞
13.evutil_make_socket_nonblocking(nfd);
14.
15.//获得连接
16.evhttp_get_request(http, nfd, (struct sockaddr *)&ss, addrlen);
17.}

代码很好懂,看看evhttp_get_request()函数:

01.//在回调函数accept_socket中被调用.
02.//这里传入的参数fd是accept()后返回的描述符
03.void evhttp_get_request(struct evhttp *http, int fd, struct sockaddr *sa, socklen_t salen)
04.{
05.struct evhttp_connection *evcon;
06.
07.//根据fd和sa创建一个evhttp_connection,并将它关联到http->base.
08.evcon = evhttp_get_request_connection(http, fd, sa, salen);
09.
10.if (http->timeout != -1)
11.//watch out!!!在这里evcon会被设置超时时间.
12.evhttp_connection_set_timeout(evcon, http->timeout);
13.
14.//将evcon关联到http
15.evcon->http_server = http;
16.
17.//将evcon插入到&http->connections
18.TAILQ_INSERT_TAIL(&http->connections, evcon, next);
19.
20.evhttp_associate_new_request_with_connection(evcon);
21.}

跟踪下去看看evhttp_associate_new_request_with_connection()函数:

01.//初始化一个绑定到evcon的evhttp_request
02.static int evhttp_associate_new_request_with_connection(struct evhttp_connection *evcon)
03.{
04.struct evhttp *http = evcon->http_server;
05.struct evhttp_request *req;
06.
07.//在这里会设置该req的回调函数evhttp_handle_request(),此函数很重要..
08.req = evhttp_request_new(evhttp_handle_request, http);
09.
10.req->evcon = evcon;
11.req->flags |= EVHTTP_REQ_OWN_CONNECTION;
12.
13.TAILQ_INSERT_TAIL(&evcon->requests, req, next);
14.
15.req->kind = EVHTTP_REQUEST;
16.
17.req->remote_host = strdup(evcon->address);
18.
19.req->remote_port = evcon->port;
20.
21.evhttp_start_read(evcon);
22.
23.return (0);
24.}

经过这么多层次的函数调用,终于要读数据了,evhttp_start_read()代码:

01.void evhttp_start_read(struct evhttp_connection *evcon)
02.{
03./* Set up an event to read the headers */
04.if (event_initialized(&evcon->ev))
05.event_del(&evcon->ev);
06.
07.//根据这些参数设置好evcon->ev.回调函数为evhttp_read()
08.event_set(&evcon->ev, evcon->fd, EV_READ, evhttp_read, evcon);
09.
10.//关联到event_base中
11.EVHTTP_BASE_SET(evcon, &evcon->ev);
12.
13.//将该ev插入到event_base中
14.
15.//watch out!!!!
16.//在这里会设置这个event的超时时间,它将被加入到定时器最小堆中
17.//超时之后,该事件会被event_active(),插入到就绪队列中,然后执行其回调函数.
18.
19.//evcon->timeout是在evhttp_get_request()被设置的
20.evhttp_add_event(&evcon->ev, evcon->timeout, HTTP_READ_TIMEOUT);
21.evcon->state = EVCON_READING_FIRSTLINE;
22.}

可以看到,对于这个连接,evhttp_connection结构体evcon是通过内部成员event *ev来处理的.函数evhttp_start_read()对&evcon->ev设置好超时时间和回调函数后将它插入到event_base中.

直到这里,回调函数accept_socket()的功能终于完成了.

(3.2) 上一段提到accept_socket()函数最终会调用evhttp_start_read()来设置连接对应的event(&evcon->ev)的超时时间和回调函数,并将它插入已**事件队列进行schedule.

在&evcon->ev超时之后,它会被函数timeout_process()从已**事件队列移入已就绪事件队列,然后由函数event_process_active()调用它的回调函数,也即是evhttp_read()(此回调函数在函数evhttp_start_read()中设置).代码如下:

01.//读数据
02.void evhttp_read(int fd, short what, void *arg)
03.{
04.struct evhttp_connection *evcon = arg;
05.//拿到第一个req
06.struct evhttp_request *req = TAILQ_FIRST(&evcon->requests);
07.struct evbuffer *buf = evcon->input_buffer;
08.int n, len;
09.
10.if (what == EV_TIMEOUT) {
11.evhttp_connection_fail(evcon, EVCON_HTTP_TIMEOUT);
12.return;
13.}
14.
15.//从fd读数据到buf
16.n = evbuffer_read(buf, fd, -1);
17.len = EVBUFFER_LENGTH(buf);
18.
19.if (n == -1) {
20.if (errno != EINTR && errno != EAGAIN) {
21.event_debug(("%s: evbuffer_read", __func__));
22.evhttp_connection_fail(evcon, EVCON_HTTP_EOF);
23.} else {
24.evhttp_add_event(&evcon->ev, evcon->timeout,
25.HTTP_READ_TIMEOUT);
26.}
27.return;
28.} else if (n == 0) {
29./* Connection closed */
30.evhttp_connection_done(evcon);
31.return;
32.}
33.
34.switch (evcon->state) {
35.case EVCON_READING_FIRSTLINE:
36.evhttp_read_firstline(evcon, req);
37.break;
38.case EVCON_READING_HEADERS:
39.evhttp_read_header(evcon, req);
40.break;
41.case EVCON_READING_BODY:
42.evhttp_read_body(evcon, req);
43.break;
44.case EVCON_READING_TRAILER:
45.evhttp_read_trailer(evcon, req);
46.break;
47.case EVCON_DISCONNECTED:
48.case EVCON_CONNECTING:
49.case EVCON_IDLE:
50.case EVCON_WRITING:
51.default:
52.event_errx(1, "%s: illegal connection state %d",
53.__func__, evcon->state);
54.}
55.}

代码中的fd其实是evcon->fd,也就是accept()后返回的文件描述符..

函数evhttp_read()就这么一直读数据下去(可能经过了多次循环,因为在evhttp_accept_socket()函数中被设置了EV_PERSIST标志,所以它不会从已注册时间队列中被移除,而是不断的超时,不断地被调用其回调函数),直到数据读完了(这里经过了好多状态,非常让人不爽的是,libevent官网上连个FSM图都没有,这种体力活我也不会干的,哈哈~),就调用evhttp_connection_done(),代码如下:

01.//累个半死终于读完啦
02.static void evhttp_connection_done(struct evhttp_connection *evcon)
03.{
04....//省略
05.
06.//调用req的回调函数
07.(*req->cb)(req, req->cb_arg);
08.
09.}

在这里,会调用req的回调函数,也就是在函数evhttp_associate_new_request_with_connection()中设置的evhttp_handle_request(),此回调函数代码为:

01.//处理请求,在这里会调用http的回调函数http->gencb
02.static void evhttp_handle_request(struct evhttp_request *req, void *arg)
03.{
04....//一堆无用的噪音
05.
06.//由用户指定的回调函数终于显灵了.
07.if (http->gencb) {
08.(*http->gencb)(req, http->gencbarg);
09.return;
10.}
11.
12.}

在数据全都读入后,libevent终于终于终于调用了用户指定的回调函数(*http->gencb).在本文一开始的小例子中,也就是函数http_handler(),要达到这一步可真不容易啊,撒花..

由上文提到的种种的繁琐的过程可以看出,libevent对于user来说是很友善的,几句代码就可以实现一个httpd,可以对于developer来说就太恶心了..

libevent源码浅析: 定时器和信号

ripwu Programming >> libevent Permalink 11.02.28 3,436 views 7 comments

上一篇文章介绍了libevent下基本的I/O事件,这篇文章将讲讲libevent对定时器和信号事件的处理.

Timer事件

反应堆event_base包含了一个最小堆min_heap结构体的实例,以此维护注册到这个反应堆实例的定时器事件:

1.struct event_base {
2.//其他成员
3.struct min_heap timeheap;
4.};

回顾一下最小堆min_heap:

1.typedef struct min_heap
2.{
3.//p指向一个动态分配的数组,数组元素是event指针.
4.struct event** p;
5.unsigned n, a; // n表示目前保存了多少元素,a表示p指向的内存能够存储event指针的个数
6.} min_heap_t;

可以看到,它包含一个连续的内存块用于存储定时器事件.针对min_heap的操作主要有:

1.static inline int min_heap_push(min_heap_t* s, struct event* e);
2.static inline struct event*  min_heap_pop(min_heap_t* s);

其中,min_heap_push()用于插入节点,min_heap_pop()用于弹出节点.其内部逻辑很简单,不必描述了.

现在看看libevent处理定时器事件的例子:

01.static void timeout_cb(int fd, short event, void *arg) {...}
02.
03.int main (int argc, char **argv)
04.{
05.struct event timeout;
06.struct timeval tv;
07.
08.event_init();
09.
10.evtimer_set(&timeout, timeout_cb, &timeout);
11.
12.evutil_timerclear(&tv);
13.tv.tv_sec = 2;
14.event_add(&timeout, &tv);
15.
16.lasttime = time(NULL);
17.
18.event_dispatch();
19.}

首先,和上篇文章例子一样的,event_init()初始化一个event_base(反应堆实例),然后由evtimer_set()设置定时器事件的回调函数,接着event_add()把定时器事件加入反应堆实例中.最后进入event_dispatch()主循环.

在这里,evtimer_set定义如下:

1.#define evtimer_set(ev, cb, arg)    event_set(ev, -1, 0, cb, arg)

至于event_set(),没有什么好说的,就是对一个event结构体做初始化罢了.

上一篇文章已经从I/O事件的角度介绍了event_add(),这里看看它是如何处理定时器事件的:

01.int event_add(struct event *ev, const struct timeval *tv)
02.{
03.struct event_base *base = ev->ev_base;
04.
05.....//处理IO事件或者信号事件的逻辑.
06.
07.//如果tv不为0
08.if (tv != NULL)
09.{
10.event_queue_insert(base, ev, EVLIST_TIMEOUT);
11.}
12.}

可以看到,event_add()会把一个定时器事件压入到其对应的反应堆实例下的定时器最小堆timeheap中(&ev->base.timeheap).

回到event_dispatch(),它会调用event_base_loop(),此函数对定时器事件处理如下:

01.//事件主循环
02.int
03.event_base_loop(struct event_base *base, int flags)
04.{
05....//不必多虑的其他代码
06.
07.done = 0;
08.while (!done)
09.{
10.//检查heap中的timer events,将就绪的timer event从heap上删除,并插入到**链表中,
11.//这意味着base->event_count_active会增加
12.timeout_process(base);
13.
14.//有就绪事件了
15.if (base->event_count_active)
16.{
17.//处理就绪事件吧.
18.event_process_active(base);
19.}
20.}
21.}

其中,timeout_process()会将已超时的定时器事件插入到反应堆实例下的已就绪事件队列中,接着由event_process_active()处理已就绪事件.event_process_active()代码在上一篇文章中已经介绍过了,这里看一下timeout_process():

01./时间到~~~
02.//开始处理base里面的定时器堆里的事件鸟.
03.//检查heap中的timer events,将就绪的timer event从heap上删除,并插入到**链表中
04.void timeout_process(struct event_base *base)
05.{
06.struct timeval now;
07.struct event *ev;
08.
09.while ((ev = min_heap_top(&base->timeheap)))
10.{
11.//ev超时时间的比现在的时间大,也就是说,这个ev还没有超时,那么while循环结束
12.if (evutil_timercmp(&ev->ev_timeout, &now, >))
13.break;
14.
15.//else 意味着 evutil_timercmp(&ev->ev_timeout, &now, <=)为真
16.//也就说明定时器最小堆的根超时了
17.
18.//从定时器堆删除
19.event_del(ev);
20.
21.//把它插到**链表吧.
22.event_active(ev, EV_TIMEOUT, 1);
23.}
24.}

Signal事件

signal事件的处理时libevent中比较难懂的地方,前人之述不详,本文重点讲解之.

反应堆event_base包含了一个evsignal_info结构体的实例,来维护注册到这个反应堆实例的信号事件:

1.struct event_base {
2.//其他成员
3.struct evsignal_info sig;
4.};

这里仔细研究一下evsignal_info结构体的定义:

01.struct evsignal_info {
02.
03.//为 socket pair 的读 socket向 event_base 注册读事件时使用的 event 结构体
04.//这个是所有信号事件共用的.
05.struct event ev_signal;
06.
07.//这个也是所有信号事件共用的.
08.int ev_signal_pair[2];
09.
10.//记录ev_signal 事件是否已经注册了
11.int ev_signal_added;
12.
13.//是否有信号发生的标记
14.//只在evsignal_handler()中被修改为1
15.volatile sig_atomic_tsig_atomic_t evsignal_caught;
16.
17.//evsigevents[signo]表示注册到信号 signo 的事件链表
18.struct event_list evsigevents[NSIG];
19.
20.//具体记录每个信号触发的次数,evsigcaught[signo]是记录信号signo被触发的次数
21.sig_atomic_tsig_atomic_t evsigcaught[NSIG];
22.
23.//记录了原来的signal处理函数指针,当信号signo注册的event被清空时,需要重新设置其处理函数
24.struct sigaction **sh_old;
25.};

要了解evsignal_info为何是这样设计的,首先需要明白int ev_signal_pair[2];的作用.它实际上表示两个文件描述符,在libevent中一个用于写,一个用于读,它们在event_init()是被初始化.好吧,其实更确切点说,event_init()会调用event_base_new(),而event_base_new()调用封装好I/O多路复用技术的结构体eventop实例(&event_base->evsel)的init函数(&event_base->evsel),这个init函数会初始化eventop实例的内部数据结构,然后调用evsignal_init()对evsignal_info结构体实例(&event_base->sig)做初始化.而在初始化实例的过程中,对其内部的ev_signal_pair[2]数组的初始化是通过调用evutil_socketpair()函数来实现的.够了,上面这段话已经够恶心了,图示如下:

看看evutil_socketpair()代码:

1.int
2.evutil_socketpair(int family, int type, int protocol, int fd[2])
3.{
4.#ifndef WIN32
5.return socketpair(family, type, protocol, fd);
6.#else
7....//山寨一个socketpair函数
8.}

它使用socketpair系统调用创建一对全双工管道(如果有时间的话,可以读一下evutil_socketpair()后半部分的代码,它在WIN32环境下如何山寨了一个socketpair函数,熟悉之可以加深不少理解.).这个全双工管道有什么用呢? 这里先卖个关子,我们看看evsignal_info结构体下的成员struct event ev_signal是如何被初始化的.

evsignal_init()调用event_set()函数,event_set()将&event_base->sig.ev_signal.ev_fd设置为&event_base->sig.ev_signal_pair[1],其回调函数为evsignal_cb(). ([1]).

至此,铺垫基本上做好了.我们看一个使用libevent处理信号事件的例子吧:

01.static void signal_cb(int fd, short event, void *arg) {...}
02.
03.int main (int argc, char **argv)
04.{
05./* Initalize the event library */
06.event_init();
07.
08.struct event signal_int;
09.event_set(&signal_int, SIGINT, EV_SIGNAL|EV_PERSIST, signal_cb, &signal_int);
10.
11.event_add(&signal_int, NULL);
12.
13.event_dispatch();
14.}

首先是由event_init()创建一个反应堆实例(在此背后,对维护信号事件的结构体evsigal_info的实例(&event_base.sig)如何被初始化在上文已经做了介绍了.),然后由event_set()设置一个事件,将其标志&signal_int.events设为EV_SIGNAL|EV_PERSIST,文件描述符&signal_int.ev_fd设置对应的信号(在例子中是SIGIN,即中断信号,中断下可以用ctrl-c触发).然后设置好这个信号事件对应的回调函数 ([2]注意,回调函数对应的是信号事件,而非信号.注意与[3]的不同.).

之后,调用event_add()将信号事件注册到反应堆实例中,event_add()对信号事件的处理如下:

01.int event_add(struct event *ev, const struct timeval *tv)
02.{
03.struct event_base *base = ev->ev_base;
04.const struct eventop *evsel = base->evsel;
05.void *evbase = base->evbase;
06.int res = 0;
07.
08.//ev->ev_events表示事件类型
09.//如果ev->ev_events是 读/写/信号 事件,而且ev不在 已注册链表 或 已**链表,那么调用evbase注册ev事件
10.if ((ev->ev_events & (EV_READ|EV_WRITE|EV_SIGNAL)) &&
11.!(ev->ev_flags & (EVLIST_INSERTED|EVLIST_ACTIVE)))
12.{
13.//实际执行操作的是evbase
14.res = evsel->add(evbase, ev);
15.
16.if (res != -1) //注册成功,把事件ev插入 已注册链表 中
17.event_queue_insert(base, ev, EVLIST_INSERTED);
18.}
19.}

为了描述方便,我们假定libevent使用的I/O多路复用技术是select,看看select_add()代码吧:

1.static int select_add(void *arg, struct event *ev)
2.{
3.if (ev->ev_events & EV_SIGNAL)
4.return (evsignal_add(ev));
5.}

对于信号事件,它转手给evsignal_add()函数处理,evsignal_add()代码如下:

01.//将信号事件ev下的描述符ev_fd(也就是信号)添加到&ev->ev_base->sig->evsigevents[ev_fd]队列中
02.int evsignal_add(struct event *ev)
03.{
04.int evsignal;
05.struct event_base *base = ev->ev_base;
06.struct evsignal_info *sig = &ev->ev_base->sig;
07.
08.//拿到event下的信号标号
09.evsignal = EVENT_SIGNAL(ev);
10.
11.if (TAILQ_EMPTY(&sig->evsigevents[evsignal]))
12.{
13.//设置这个事件对应的信号对应的处理函数
14.
15.//watch out!!!!针对的是信号,不是事件
16.if (_evsignal_set_handler(
17.base, evsignal, evsignal_handler) == -1)
18.return (-1);
19.
20.//这里注册的sig本身,而不是信号事件
21.//也就是就是说,sig是在真正有信号事件时才注册的.
22.if (!sig->ev_signal_added)
23.{
24.//注册这个信号对应的事件
25.if (event_add(&sig->ev_signal, NULL))
26.return (-1);
27.sig->ev_signal_added = 1;
28.}
29.}
30.
31.//多个事件可能对应同一信号
32.TAILQ_INSERT_TAIL(&sig->evsigevents[evsignal], ev, ev_signal_next);
33.}

evsignal_add()函数先获得信号事件对应的信号,通过_evsignal_set_handler()函数将此信号相应的信号处理函数设置为evsignal_handler(). ([3]注意,[2]设置的回调函数是针对信号事件的,这里设置的处理函数才是针对信号的.) 接着,evsignal_add()判断sig->ev_signal_added是否为0,为0则将&sig->ev_signal事件注册到反应堆实例中,然后将sig->ev_signal_added置1。;如果不为0,那么跳过这段代码.需要指出的是,sig->ev_signal_added唯一一次被置1就是在这段代码中,这保证了&sig->ev_signal事件只被注册到反应堆实例中一次.其实也就是说,只有在第1次有信号事件需要通过event_add()被注册到反应堆实例时,&sig->ev_signal事件才会被一起注册,这是libevent对&event_base->sig的延后处理.

<hr/>

接下来,貌似应该讲讲event_dispatch()对信号事件的处理了.且慢,我们回头把 [1], [2], [3] 整理一下:

(1) 在调用event_init()新建一个反应堆实例(以base表示)时,evsignal_info结构体(libevent用它来管理信号事件集合) base->sig被初始化,base->sig->ev_signal的回调函数总是被设置为evsignal_cb(),而evsignal_cb()是定义在libevent内部的,对libevent用户完全透明,其代码如下:

1.static void
2.evsignal_cb(int fd, short what, void *arg)
3.{
4.recv(fd, signals, sizeof(signals), 0);
5.}

它从一个文件描述符(后文会看到,这个文件描述符总是&event_base->sig.ev_signal_pair[1])读1比特的数据.

(2) 在已经通过调用event_init()获得一个反应堆实例后,通过event_set()设置一个信号事件signal_int的文件描述符signal_int.ev_fd(其实对于信号事件而言,ev_fd也就是此信号事件对应的信号),event_set()还设置了这个信号事件的回调函数.很明显,对于同一个信号,可以有不同的信号事件,这些信号事件的回调函数也可以完全不同.在这里,回调函数是由用户设计的,表示信号被触发时希望作出的反馈函数.

(3) 为了将一个事件(这个事件可以是I/O事件,也可以是定时器事件,也可以是信号事件)注册到反应堆实例中,我们必须调用event_add(),而event_add()通过重重调用,最终由evsignal_add()来完成将信号事件注册.回顾一下evsignal_add():

它通过_evsignal_set_handler总是将信号事件对应的信号的处理函数设置为evsignal_handler(),evsignal_handler()代码如下:

01.//通知event_base有信号发生的技巧,往sig.ev_signal_pair[0]写1字节数据
02.//会设置sig.evsignal_caught = 1,标记有信号产生.
03.static void evsignal_handler(int sig)
04.{
05.evsignal_base->sig.evsigcaught[sig]++;
06.evsignal_base->sig.evsignal_caught = 1; //将信号发生标志至1
07.
08.send(evsignal_base->sig.ev_signal_pair[0], "a", 1, 0);
09.}

它将信号发生标志evsignal_base->sig.evsignal_caught置1,以此通过libevent有信号发生.然后往&event_base->sig.ev_signal_pair[0]写1比特数据.

<hr/>

好吧,现在终于可以看看libevent是如何处理信号事件的了:

libevent先进入event_base_loop()主循环,等待已经准备好(可读可写或异常)的事件(通过select_dispatch找出已准备好的文件描述符).当有一个信号产生时,由于这个信号的信号处理函数(总是evsignal_handler())总是会往&event_base->sig.ev_signal_pair[0]写1比特数据(这是由操作系统调用的,对libevent是透明的,对libevent的用户就更加透明了).此时,根据前面的描述,由于ev_signal_pair[0]与ev_signal_pair[1]是一对全双工管道,所以,ev_signal_pair[1]将变得可读.而&event_base->sig.ev_signal事件的文件描述符正是ev_signal_pair[1],所以libevent可以知道&event_base->sig.ev_signal事件准备好了.为此,&event_base->sig.ev_signal事件被移入反应堆实例下的已就绪事件队列.接着在event_base_loop()的后续部分代码中被处理,通过event_process_active()调用其回调函数,也就是evsignal_cb(),从&event_base->sig.ev_signal_pair[1])读1比特的数据.[4]我们把信号被捕捉到的这个while()循环记为第1次while()循环.

写到这里,仍然有一个疑惑没有解开,上面都是讲libevent内部定义的&event_base->sig.ev_signal如何如何,可是我们希望的是自己定义的信号事件signal_int如何如何啊.

答案是,正如(3)描述的那样,在操作系统调用信号处理函数evsignal_handler()时,它会将信号发生标志置1.然后将evsignal_info结构体中用于记录信号被捕捉次数的evsigcaught[id]++,id也就是这个信号.

在第2此while()循环时(参考[4]),它还是调用select_dispatch(),这时,由于信号发生标志为1,所以select_dispatch()会调用函数evsignal_process().select_dispatch()相关代码如下:

1.static int
2.select_dispatch(struct event_base *base, void *arg, struct timeval *tv)
3.{
4.if (base->sig.evsignal_caught){
5.evsignal_process(base);
6.}

evsignal_process()代码如下:

01.void evsignal_process(struct event_base *base)
02.{
03.struct evsignal_info *sig = &base->sig;
04.struct event *ev, *next_ev;
05.sig_atomic_tsig_atomic_t ncalls;
06.int i;
07.
08.base->sig.evsignal_caught = 0;
09.for (i = 1; i < NSIG; ++i)
10.{
11.ncalls = sig->evsigcaught[i];
12.if (ncalls == 0)
13.continue;
14.sig->evsigcaught[i] -= ncalls;
15.
16.for (ev = TAILQ_FIRST(&sig->evsigevents[i]);
17.ev != NULL; ev = next_ev)
18.{
19.next_ev = TAILQ_NEXT(ev, ev_signal_next);
20.if (!(ev->ev_events & EV_PERSIST))
21.event_del(ev);
22.
23.//移到已就绪事件队列,ncalls回调函数将会被调用多少次
24.event_active(ev, EV_SIGNAL, ncalls);
25.}
26.}
27.}

总结一下,反应堆结构体event_base有一个数据成员evsignal_info结构体,它维护信号事件集.之所以evsignal_info会有一个event事件成员ev_signal,是因为libevent通过socket pair让操作系统通知自己有信号发生,在信号处理函数中将信号发生标志置1,并使该信号被捕捉的次数自增,然后ev_signal被移到已就绪事件队列,接着被处理.然后libevent检查到信号发生标志已经被置1,遍历所有信号事件,找出信号被捕捉次数不为0的那个信号事件集,将它们移到已就绪事件队列,然后处理之.

以上,就是libevent处理信号事件的逻辑.

libevent源码浅析: 事件处理框架

ripwu Programming >> libevent Permalink 11.02.28 1,796 views rfc

本文将从一个使用libevent的小例子出发,解释libevent处理事件的流程.

例子如下:

01.static void fifo_read(int fd, short event, void *arg) {...}
02.
03.int main (int argc, char **argv)
04.{
05.int socket = open ("/tmp/event.fifo", O_RDONLY | O_NONBLOCK, 0);
06.
07.fprintf(stdout, "Please write data to %s\n", fifo);
08.
09.event_init();
10.
11.struct event evfifo;
12.event_set(&evfifo, socket, EV_READ, fifo_read, &evfifo);
13.
14.event_add(&evfifo, NULL);
15.
16.event_dispatch();
17.}

libevent库的使用方法大体上就像例子展示的那样,先由event_init()得到一个event_base实例(也就是反应堆实例),然后由 event_set()初始化一个event,接着用event_add()将event绑定到event_base,最后调用event_dispatch()进入时间主循环.

event_init()和event_set()功能都很简单,它们分别对event_base结构体和event结构体做初始化.我们直接看看event_add():

01.//为了思路清晰,这里分析的是I/O事件,暂不考虑信号和定时器相关处理代码.
02.
03.//函数将ev注册到ev->ev_base上,事件类型由ev->ev_events指明.如果注册成功,ev将被插入到已注册链表中.
04.int event_add(struct event *ev, const struct timeval *tv)
05.{
06.//得到ev对应的反应堆实例event_base
07.struct event_base *base = ev->ev_base;
08.
09.//得到libevent选择的已封装的I/O多路复用技术
10.const struct eventop *evsel = base->evsel;
11.
12.void *evbase = base->evbase;
13.int res = 0;
14.
15.//ev->ev_events表示事件类型
16.//如果ev->ev_events是 读/写/信号 事件,而且ev不在 已注册队列 或 已就绪队列,
17.//那么调用evbase注册ev事件
18.if ((ev->ev_events & (EV_READ|EV_WRITE|EV_SIGNAL)) &&
19.!(ev->ev_flags & (EVLIST_INSERTED|EVLIST_ACTIVE)))
20.{
21.
22.//实际执行操作的是evbase
23.res = evsel->add(evbase, ev);
24.
25.//注册成功,把事件ev插入已注册队列中
26.if (res != -1)
27.event_queue_insert(base, ev, EVLIST_INSERTED);
28.}
29.
30.return (res);
31.}

注释已经很明了了,如果一个事件不在已注册队列或者已**队列,而且它是I/O事件或者信号事件,那么调用select_add()将ev插入到selectop的内部数据结构中(本文以select为例,下文不再说明.).select_add()代码如下:

01.//略去信号处理的相关代码
02.
03.static int select_add(void *arg, struct event *ev)
04.{
05.struct selectop *sop = arg;
06.
07.//如果是读类型事件,把该事件的文件描述符加入到selectop维护的读fd_set集
08.//event_readset_in中,并且把该事件插入到读事件队列.
09.if (ev->ev_events & EV_READ)
10.{
11.FD_SET(ev->ev_fd, sop->event_readset_in);
12.sop->event_r_by_fd[ev->ev_fd] = ev;
13.}
14.
15.//略去对写事件的处理
16.}

小结一下,结合event_add()代码和select_add()代码,可以看出在调用event_add()时,事件将被插入其对应的反应堆实例event_base的已注册事件队列中,而且还会被加入到selectop维护的内部数据结构中进行监视.

现在可以看看event_dispatch()代码了:

01.//略去信号事件和定时器事件处理的相关代码
02.
03.int event_dispatch(void)
04.{
05.return (event_loop(0));
06.}
07.
08.int event_loop(int flags)
09.{
10.return event_base_loop(current_base, flags);
11.}
12.
13.//事件主循环
14.int event_base_loop(struct event_base *base, int flags)
15.{
16.const struct eventop *evsel = base->evsel;
17.void *evbase = base->evbase;
18.struct timeval *tv_p;
19.int res, done;
20.
21.done = 0;
22.while (!done)
23.{
24.//从定时器最小堆中取出根节点,其时间值作为select最大等待时间
25.//如果定时器最小堆没有元素,那么select最大等待时间为0
26.timeout_next(base, &tv_p);
27.
28.//调用select_dispatch(),它会将已经准备好的事件移到已就绪事件队列中
29.res = evsel->dispatch(base, evbase, tv_p);
30.
31.//有就绪事件了,那就处理就绪事件吧.
32.if (base->event_count_active)
33.event_process_active(base);
34.}
35.}

event_base_loop()先从定时器最小堆中取出根节点作为select的最大等待时间,然后调用select_dispatch()将已经准备好的事件移到已就绪事件队列中,最后调用event_process_active()处理已就绪事件队列.

01.//略去信号事件和定时器事件处理的相关代码
02.
03.static int select_dispatch(struct event_base *base, void *arg, struct timeval *tv)
04.{
05.int res, j;
06.struct selectop *sop = arg;
07.
08.//根据前面对select_add()的解释,事件fd已被加入到fd_set集中进行监视.
09.res = select(sop->event_fds + 1, sop->event_readset_out,
10.sop->event_writeset_out, NULL, tv);
11.
12.for (j = 0, res = 0; j <= sop->event_fds; ++j, res = 0)
13.{
14.struct event *r_ev = NULL, *w_ev = NULL;
15.
16.//找出已经准备好读的事件
17.if (FD_ISSET(j, sop->event_readset_out))
18.{
19.r_ev = sop->event_r_by_fd[i];
20.res |= EV_READ;
21.}
22.
23.//将已经准备好读的事件移到已就绪事件队列
24.if (r_ev && (res & r_ev->ev_events))
25.event_active(r_ev, res & r_ev->ev_events, 1);
26.
27.//略去对已经准备好写的事件的处理
28.}
29.}

看看在event_base_loop()中被调用的event_process_active()代码:

01.static void event_process_active(struct event_base *base)
02.{
03.struct event *ev;
04.struct event_list *activeq = NULL;
05.int i;
06.short ncalls;
07.
08.//寻找最高优先级(priority值越小优先级越高)的已就绪事件队列
09.for (i = 0; i < base->nactivequeues; ++i)
10.{
11.if (TAILQ_FIRST(base->activequeues[i]) != NULL)
12.{
13.activeq = base->activequeues[i];
14.break;
15.}
16.}
17.
18.for (ev = TAILQ_FIRST(activeq); ev; ev = TAILQ_FIRST(activeq))
19.{
20.//如果有persist标志,则只从**队列中移除此事件,
21.if (ev->ev_events & EV_PERSIST)
22.event_queue_remove(base, ev, EVLIST_ACTIVE);
23.
24.else //否则则从**事件列表,以及已注册事件中双杀此事件
25.event_del(ev);
26.
27.ncalls = ev->ev_ncalls;
28.ev->ev_pncalls = &ncalls;
29.
30.//每个事件的回调函数的调用次数
31.while (ncalls)
32.{
33.ncalls--;
34.ev->ev_ncalls = ncalls;
35.
36.//调用回调函数
37.(*ev->ev_callback)((int)ev->ev_fd, ev->ev_res, ev->ev_arg);
38.}
39.}
40.}

现在,看看这个被阉割的只考虑I/O事件的libevent主循环框架:

event_base_loop: while() { //从定时器最小堆取出select最大等待时间 //select出已准备事件,将它们移到已就绪事件队列中 //处理已就绪事件 }

这真是篇节能环保的文章啊,哈哈.因为libevent代码太恶心了,描述出来都觉得恶心,有空得拿来重构下..下篇文章会讲讲libevent中非常恶心的信号集成处理.