Libevent学习——Echo Server based on libevent

首先给出官方文档吧： 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事件的下一次执行。

    以上步骤比较晦涩，具体代码可参考官方文档里面的【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行，多抄几遍，就能彻底弄懂啦！更复杂的例子参见官方文档里面的

#include <event2/listener.h>
#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <arpa/inet.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
static void
echo_read_cb(struct bufferevent *bev, void *ctx) {
	struct evbuffer *input = bufferevent_get_input(bev);
	struct evbuffer *output = bufferevent_get_output(bev);
	evbuffer_add_buffer(output, input);
}
static void
echo_event_cb(struct bufferevent *bev, short events, void *ctx) {
	if (events & BEV_EVENT_ERROR)
		perror("Error from bufferevent");
	if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR))  {
		bufferevent_free(bev);
	}
}
static void
accept_conn_cb(struct evconnlistener *listener,
               evutil_socket_t fd, struct sockaddr *address, int socklen,
               void *ctx) {
	struct event_base *base = evconnlistener_get_base(listener);
	struct bufferevent *bev = bufferevent_socket_new(
	                             base, fd, BEV_OPT_CLOSE_ON_FREE);
	bufferevent_setcb(bev, echo_read_cb, NULL, echo_event_cb, NULL);
	bufferevent_enable(bev, EV_READ|EV_WRITE);
}
static void
accept_error_cb(struct evconnlistener *listener, void *ctx) {
	struct event_base *base = evconnlistener_get_base(listener);
	int err = EVUTIL_SOCKET_ERROR();
	fprintf(stderr, "Got an error %d (%s) on the listener. "
	        "Shutting down.\n", err, evutil_socket_error_to_string(err));
	event_base_loopexit(base, NULL);
}

int main(int argc, char **argv) {
	struct event_base *base;
	struct evconnlistener *listener;
	struct sockaddr_in sin;
	int port = 9876;
	if (argc > 1)  {
		port = atoi(argv[1]);
	}
	if (port<=0 || port>65535)  {
		puts("Invalid port");
		return 1;
	}
	base = event_base_new();
	if (!base)  {
		puts("Couldn't open event base");
		return 1;
	}
	memset(&sin, 0, sizeof(sin));
	sin.sin_family = AF_INET;
	sin.sin_addr.s_addr = htonl(0);
	sin.sin_port = htons(port);
	listener = evconnlistener_new_bind(base, accept_conn_cb, NULL,
	                                   LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE, -1,
	                                   (struct sockaddr*)&sin, sizeof(sin));
	if (!listener)  {
		perror("Couldn't create listener");
		return 1;
	}
	evconnlistener_set_error_cb(listener, accept_error_cb);
	event_base_dispatch(base);
	return 0;
}