libevent源码浅析: http库-白红宇

libevent源码浅析: http库

阅读量：2224 次

发布时间：2019-05-08

本文共 25490 字，大约阅读时间需要 84 分钟。

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(在这里设置好回调函数)后添加到&amp;amp;amp;http-&amp;amp;gt;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,并将它插入到&amp;amp;amp;http-&amp;amp;gt;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来说就太恶心了..

11.02.28

3,436 views

上一篇文章介绍了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处理信号事件的逻辑.

11.02.28

1,796 views

本文将从一个使用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. 
}