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;
}