libevent入门教程:Echo Server based on libevent
来自
felix021( http://www.felix021.com/blog/read.php?2068) 大侠花了两天的时间在libevent上,想总结下,就以写简单tutorial的方式吧,貌似没有一篇简单的说明,让人马上就能上手用的。
首先给出官方文档吧: http://libevent.org ,首页有个Programming with Libevent,里面是一节一节的介绍libevent,但是感觉信息量太大了,而且还是英文的-。-(当然,如果想好好用libevent,看看还是很有必要的),还有个Reference,大致就是对各个版本的libevent使用doxgen生成的文档,用来查函数原型和基本用法什么的。
下面假定已经学习过基本的socket编程(socket,bind,listen,accept,connect,recv,send,close),并且对异步/callback有基本认识。
基本的socket编程是阻塞/同步的,每个操作除非已经完成或者出错才会返回,这样对于每一个请求,要使用一个线程或者单独的进程去处理,系统资源没法支撑大量的请求(所谓c10k problem?),例如内存:默认情况下每个线程需要占用2~8M的栈空间。posix定义了可以使用异步的select系统调用,但是因为其采用了轮询的方式来判断某个fd是否变成active,效率不高[O(n)],连接数一多,也还是撑不住。于是各系统分别提出了基于异步/callback的系统调用,例如Linux的epoll,BSD的kqueue,Windows的IOCP。由于在内核层面做了支持,所以可以用O(1)的效率查找到active的fd。基本上,libevent就是对这些高效IO的封装,提供统一的API,简化开发。
libevent大概是这样的:
默认情况下是单线程的(可以配置成多线程,如果有需要的话),每个线程有且只有一个event\_base,对应一个struct event\_base结构体(以及附于其上的事件管理器),用来schedule托管给它的一系列event,可以和操作系统的进程管理类比,当然,要更简单一点。当一个事件发生后,event\_base会在合适的时间(不一定是立即)去调用绑定在这个事件上的函数(传入一些预定义的参数,以及在绑定时指定的一个参数),直到这个函数执行完,再返回schedule其他事件。
//创建一个event_base
struct event_base *base = event_base_new();
assert(base != NULL);
event\_base内部有一个循环,循环阻塞在epoll/kqueue等系统调用上,直到有一个/一些事件发生,然后去处理这些事件。当然,这些事件要被绑定在这个event\_base上。每个事件对应一个struct event,可以是监听一个fd或者POSIX信号量之类(这里只讲fd了,其他的看manual吧)。struct event使用event\_new来创建和绑定,使用event\_add来启用:
//创建并绑定一个event
struct event *listen_event;
//参数:event_base, 监听的fd,事件类型及属性,绑定的回调函数,给回调函数的参数
listen_event = event_new(base, listener, EV_READ|EV_PERSIST, callback_func, ( void*)base);
//参数:event,超时时间( struct timeval *类型的, NULL表示无超时设置)
event_add(listen_event, NULL);
注:libevent支持的事件及属性包括(使用bitfield实现,所以要用 | 来让它们合体)
(a) EV\_TIMEOUT: 超时
(b) EV\_READ: 只要网络缓冲中还有数据,回调函数就会被触发
(c) EV\_WRITE: 只要塞给网络缓冲的数据被写完,回调函数就会被触发
(d) EV\_SIGNAL: POSIX信号量,参考manual吧
(e) EV\_PERSIST: 不指定这个属性的话,回调函数被触发后事件会被删除
(f) EV\_ET: Edge-Trigger边缘触发,参考EPOLL\_ET
然后需要启动event\_base的循环,这样才能开始处理发生的事件。循环的启动使用event\_base\_dispatch,循环将一直持续,直到不再有需要关注的事件,或者是遇到event\_loopbreak()/event\_loopexit()函数。
//启动事件循环
event_base_dispatch(base);
接下来关注下绑定到event的回调函数callback\_func:传递给它的是一个socket fd、一个event类型及属性bit\_field、以及传递给event\_new的最后一个参数(去上面几行回顾一下,把event\_base给传进来了,实际上更多地是分配一个结构体,把相关的数据都撂进去,然后丢给event\_new,在这里就能取得到了)。其原型是:
typedef void(* event_callback_fn)(evutil_socket_t sockfd, short event_type, void *arg)
对于一个服务器而言,上面的流程大概是这样组合的:
1. listener = socket(),bind(),listen(),设置nonblocking(POSIX系统中可使用fcntl设置,windows不需要设置,实际上libevent提供了统一的包装evutil\_make\_socket\_nonblocking)
2. 创建一个event\_base
3. 创建一个event,将该socket托管给event\_base,指定要监听的事件类型,并绑定上相应的回调函数(及需要给它的参数)。对于listener socket来说,只需要监听EV\_READ|EV\_PERSIST
4. 启用该事件
5. 进入事件循环
---------------
6. (异步) 当有client发起请求的时候,调用该回调函数,进行处理。
问题:为什么不在listen完马上调用accept,获得客户端连接以后再丢给event\_base呢?这个问题先想想噢。
回调函数要做什么事情呢?当然是处理client的请求了。首先要accept,获得一个可以与client通信的sockfd,然后……调用recv/send吗?错!大错特错!如果直接调用recv/send的话,这个线程就阻塞在这个地方了,如果这个客户端非常的阴险(比如一直不发消息,或者网络不好,老是丢包),libevent就只能等它,没法处理其他的请求了——所以应该创建一个新的event来托管这个sockfd。
在老版本libevent上的实现,比较罗嗦\[如果不想详细了解的话,看下一部分\]。
对于服务器希望先从client获取数据的情况,大致流程是这样的:
1. 将这个sockfd设置为nonblocking
2. 创建2个event:
event\_read,绑上sockfd的EV\_READ|EV\_PERSIST,设置回调函数和参数(后面提到的struct)
event\_write,绑上sockfd的EV\_WRITE|EV\_PERSIST,设置回调函数和参数(后面提到的struct)
3. 启用event\_read事件
------
4. (异步) 等待event\_read事件的发生, 调用相应的回调函数。这里麻烦来了:回调函数用recv读入的数据,不能直接用send丢给sockfd了事——因为sockfd是nonblocking的,丢给它的话,不能保证正确(为什么呢?)。所以需要一个自己管理的缓存用来保存读入的数据中(在accept以后就创建一个struct,作为第2步回调函数的arg传进来),在合适的时间(比如遇到换行符)启用event\_write事件【event\_add(event\_write, NULL)】,等待EV\_WRITE事件的触发
------
5. (异步) 当event\_write事件的回调函数被调用的时候,往sockfd写入数据,然后删除event\_write事件【event\_del(event\_write)】,等待event\_read事件的下一次执行。
以上步骤比较晦涩,具体代码可参考 [官方文档][Link 1] 里面的【Example: A low-level ROT13 server with Libevent】
由于需要自己管理缓冲区,且过程晦涩难懂,并且不兼容于Windows的IOCP,所以libevent2开始,提供了bufferevent这个神器,用来提供更加优雅、易用的API。struct bufferevent内建了两个event(read/write)和对应的缓冲区【struct evbuffer \*input, \*output】,并提供相应的函数用来操作缓冲区(或者直接操作bufferevent)。每当有数据被读入input的时候,read\_cb函数被调用;每当output被输出完的时候,write\_cb被调用;在网络IO操作出现错误的情况(连接中断、超时、其他错误),error\_cb被调用。于是上一部分的步骤被简化为:
1. 设置sockfd为nonblocking
2. 使用bufferevent\_socket\_new创建一个struct bufferevent \*bev,关联该sockfd,托管给event\_base
3. 使用bufferevent\_setcb(bev, read\_cb, write\_cb, error\_cb, (void \*)arg)将EV\_READ/EV\_WRITE对应的函数
4. 使用bufferevent\_enable(bev, EV\_READ|EV\_WRITE|EV\_PERSIST)来启用read/write事件
------
5. (异步)
在read\_cb里面从input读取数据,处理完毕后塞到output里(会被自动写入到sockfd)
在write\_cb里面(需要做什么吗?对于一个echo server来说,read\_cb就足够了)
在error\_cb里面处理遇到的错误
\*. 可以使用bufferevent\_set\_timeouts(bev, struct timeval \*READ, struct timeval \*WRITE)来设置读写超时, 在error\_cb里面处理超时。
\*. read\_cb和write\_cb的原型是
void read\_or\_write\_callback(struct bufferevent \*bev, void \*arg)
error\_cb的原型是
void error\_cb(struct bufferevent \*bev, short error, void \*arg) //这个是event的标准回调函数原型
可以从bev中用libevent的API提取出event\_base、sockfd、input/output等相关数据,详情RTFM~
于是代码简化到只需要几行的read\_cb和error\_cb函数即可:
void read_cb( struct bufferevent *bev, void *arg) {
char line[256];
int n;
evutil_socket_t fd = bufferevent_getfd(bev);
while (n = bufferevent_read(bev, line, 256), n > 0)
bufferevent_write(bev, line, n);
}
void error_cb( struct bufferevent *bev, short event, void *arg) {
bufferevent_free(bev);
}
于是一个支持大并发量的echo server就成型了!下面附上无注释的echo server源码,110行,多抄几遍,就能完全弄懂啦!更复杂的例子参见 [官方文档][Link 1] 里面的【Example: A simpler ROT13 server with Libevent】
# include
# include
# include
# include
# include
# include
# define LISTEN_PORT 9999
# define LISTEN_BACKLOG 32
void do_accept(evutil_socket_t listener, short event, void *arg);
void read_cb( struct bufferevent *bev, void *arg);
void error_cb( struct bufferevent *bev, short event, void *arg);
void write_cb( struct bufferevent *bev, void *arg);
int main( int argc, char *argv[])
{
int ret;
evutil_socket_t listener;
listener = socket(AF_INET, SOCK_STREAM, 0);
assert(listener > 0);
evutil_make_listen_socket_reuseable(listener);
struct sockaddr\_in sin;
sin.sin\_family = AF\_INET;
sin.sin\_addr.s\_addr = 0;
sin.sin\_port = htons(LISTEN\_PORT);
if (bind(listener, ( struct sockaddr \*)&sin, sizeof(sin)) < 0) \{
perror("bind");
return 1;
\}
if (listen(listener, LISTEN\_BACKLOG) < 0) \{
perror("listen");
return 1;
\}
printf ("Listening...\\n");
evutil\_make\_socket\_nonblocking(listener);
struct event\_base \*base = event\_base\_new();
assert(base != NULL);
struct event \*listen\_event;
listen\_event = event\_new(base, listener, EV\_READ|EV\_PERSIST, do\_accept, ( void\*)base);
event\_add(listen\_event, NULL);
event\_base\_dispatch(base);
printf("The End.");
return 0;
}
void do_accept(evutil_socket_t listener, short event, void *arg)
{
struct event_base *base = ( struct event_base *)arg;
evutil_socket_t fd;
struct sockaddr_in sin;
socklen_t slen;
fd = accept(listener, ( struct sockaddr *)&sin, &slen);
if (fd < 0) {
perror(“accept”);
return;
}
if (fd > FD_SETSIZE) {
perror(“fd > FD_SETSIZE\n”);
return;
}
printf("ACCEPT: fd = %u\\n", fd);
struct bufferevent \*bev = bufferevent\_socket\_new(base, fd, BEV\_OPT\_CLOSE\_ON\_FREE);
bufferevent\_setcb(bev, read\_cb, NULL, error\_cb, arg);
bufferevent\_enable(bev, EV\_READ|EV\_WRITE|EV\_PERSIST);
}
void read_cb( struct bufferevent *bev, void *arg)
{
# define MAX_LINE 256
char line[MAX_LINE+1];
int n;
evutil_socket_t fd = bufferevent_getfd(bev);
while (n = bufferevent\_read(bev, line, MAX\_LINE), n > 0) \{
line\[n\] = '\\0';
printf("fd=%u, read line: %s\\n", fd, line);
bufferevent\_write(bev, line, n);
\}
}
void write_cb( struct bufferevent *bev, void *arg) {}
void error_cb( struct bufferevent *bev, short event, void *arg)
{
evutil_socket_t fd = bufferevent_getfd(bev);
printf(“fd = %u, “, fd);
if (event & BEV_EVENT_TIMEOUT) {
printf(“Timed out\n”); // if bufferevent_set_timeouts() called
}
else if (event & BEV_EVENT_EOF) {
printf(“connection closed\n”);
}
else if (event & BEV_EVENT_ERROR) {
printf(“some other error\n”);
}
bufferevent_free(bev);
}
还没有评论,来说两句吧...