大多數(shù)程序員從阻塞IO調(diào)用開(kāi)始學(xué)習(xí)。如果調(diào)用在操作完成之前，或者足夠的時(shí)間已經(jīng)流逝使得網(wǎng)絡(luò)棧放棄操作之前，不會(huì)返回，那么就是異步的。比如說(shuō)，在TCP連接上調(diào)用connect（）時(shí)，操作系統(tǒng)將一個(gè)SYN分組排隊(duì)到TCP連接的另一端主機(jī)中。在收到來(lái)自對(duì)方主機(jī)的SYN ACK分組之前，或者直到足夠的時(shí)間已經(jīng)流逝而決定放棄操作之前，控制不會(huì)返回到應(yīng)用程序。
這里有一個(gè)使用阻塞網(wǎng)絡(luò)調(diào)用的簡(jiǎn)單客戶端示例。它打開(kāi)到www.google.com的連接，發(fā)送一個(gè)簡(jiǎn)單的HTTP請(qǐng)求，將響應(yīng)打印到stdout。
/* For sockaddr_in */
#include <netinet/in.h>
/* For socket functions */
#include <sys/socket.h>
/* For gethostbyname */
#include <netdb.h>
#include <unistd.h>
#include <string.h>
#include <stdio.h>
int main(int c, char **v)
{
const char query[] =
"GET / HTTP/1.0\r\n"
"Host: www.google.com\r\n"
"\r\n";
const char hostname[] = "www.google.com";
struct sockaddr_in sin;
struct hostent *h;
const char *cp;
int fd;
ssize_t n_written, remaining;
char buf[1024];
/* Look up the IP address for the hostname. Watch out; this isn't
threadsafe on most platforms. */
h = gethostbyname(hostname);
if (!h) {
fprintf(stderr, "Couldn't lookup %s: %s", hostname, hstrerror(h_errno));
return 1;
}
if (h->h_addrtype != AF_INET) {
fprintf(stderr, "No ipv6 support, sorry.");
return 1;
}
/* Allocate a new socket */
fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd < 0) {
perror("socket");
return 1;
}
/* Connect to the remote host. */
sin.sin_family = AF_INET;
sin.sin_port = htons(80);
sin.sin_addr = *(struct in_addr*)h->h_addr;
if (connect(fd, (struct sockaddr*) &sin, sizeof(sin))) {
perror("connect");
close(fd);
return 1;
}
/* Write the query. */
/* XXX Can send succeed partially? */
cp = query;
remaining = strlen(query);
while (remaining) {
n_written = send(fd, cp, remaining, 0);
if (n_written <= 0) {
perror("send");
return 1;
}
remaining -= n_written;
cp += n_written;
}
/* Get an answer back. */
while (1) {
ssize_t result = recv(fd, buf, sizeof(buf), 0);
if (result == 0) {
break;
} else if (result < 0) {
perror("recv");
close(fd);
return 1;
}
fwrite(buf, 1, result, stdout);
}
close(fd);
return 0;
}
上述代碼中的所有網(wǎng)絡(luò)調(diào)用都是阻塞的：在成功解析www.google.com，或者解析失敗之前，gethostbyname不會(huì)返回；連接建立之前connect不會(huì)返回；收到數(shù)據(jù)或者關(guān)閉之前recv調(diào)用不會(huì)返回；至少在清空輸出緩沖區(qū)到內(nèi)核的寫(xiě)緩沖區(qū)之前，send調(diào)用不會(huì)返回。
這里，阻塞IO沒(méi)有什么不好的。如果沒(méi)有其他事情需要同時(shí)進(jìn)行，阻塞IO會(huì)工作得很好。但是考慮需要同時(shí)處理多個(gè)連接的情形。考慮一個(gè)具體的例子：需要從兩個(gè)連接讀取輸入，但是不知道哪個(gè)連接將先收到輸入。程序可能是這樣的：
/* This won't work. */
char buf[1024];
int i, n;
while (i_still_want_to_read()) {
for (i=0; i<n_sockets; ++i) {
n = recv(fd[i], buf, sizeof(buf), 0);
if (n==0)
handle_close(fd[i]);
else if (n<0)
handle_error(fd[i], errno);
else
handle_input(fd[i], buf, n);
}
}
即使fd[2]上最先有數(shù)據(jù)到達(dá)，對(duì)fd[0]和fd[1]的讀取操作取得一些數(shù)據(jù)并且完成之前，程序不會(huì)試圖從fd[2]進(jìn)行讀取。
有時(shí)候用多線程或者多進(jìn)程服務(wù)器來(lái)解決此問(wèn)題。最簡(jiǎn)單的方式是用一個(gè)單獨(dú)的進(jìn)程（或者線程）處理每個(gè)連接。因?yàn)槊總€(gè)連接擁有獨(dú)立的進(jìn)程，一個(gè)連接上阻塞的IO調(diào)用不會(huì)阻塞其他任何連接的進(jìn)程。
這里有另一個(gè)示例程序。它是一個(gè)簡(jiǎn)單的服務(wù)器，在端口47013上監(jiān)聽(tīng)TCP連接，每次從其輸入緩沖區(qū)讀取一行，寫(xiě)回其ROT13混淆結(jié)果。程序使用fork（）調(diào)用為每個(gè)進(jìn)入的連接創(chuàng)建一個(gè)新的進(jìn)程。
/* For sockaddr_in */
#include <netinet/in.h>
/* For socket functions */
#include <sys/socket.h>
#include <unistd.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#define MAX_LINE 16384
char
rot13_char(char c)
{
/* We don't want to use isalpha here; setting the locale would change
* which characters are considered alphabetical. */
if ((c >= 'a' && c <= 'm') || (c >= 'A' && c <= 'M'))
return c + 13;
else if ((c >= 'n' && c <= 'z') || (c >= 'N' && c <= 'Z'))
return c - 13;
else
return c;
}
void
child(int fd)
{
char outbuf[MAX_LINE+1];
size_t outbuf_used = 0;
ssize_t result;
while (1) {
char ch;
result = recv(fd, &ch, 1, 0);
if (result == 0) {
break;
} else if (result == -1) {
perror("read");
break;
}
/* We do this test to keep the user from overflowing the buffer. */
if (outbuf_used < sizeof(outbuf)) {
outbuf[outbuf_used++] = rot13_char(ch);
}
if (ch == '\n') {
send(fd, outbuf, outbuf_used, 0);
outbuf_used = 0;
continue;
}
}
}
void
run(void)
{
int listener;
struct sockaddr_in sin;
sin.sin_family = AF_INET;
sin.sin_addr.s_addr = 0;
sin.sin_port = htons(40713);
listener = socket(AF_INET, SOCK_STREAM, 0);
#ifndef WIN32
{
int one = 1;
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
}
#endif
if (bind(listener, (struct sockaddr*)&sin, sizeof(sin)) < 0) {
perror("bind");
return;
}
if (listen(listener, 16)<0) {
perror("listen");
return;
}
while (1) {
struct sockaddr_storage ss;
socklen_t slen = sizeof(ss);
int fd = accept(listener, (struct sockaddr*)&ss, &slen);
if (fd < 0) {
perror("accept");
} else {
if (fork() == 0) {
child(fd);
exit(0);
}
}
}
}
int
main(int c, char **v)
{
run();
return 0;
}
是否有同時(shí)處理多個(gè)連接的完美解決方案？我可以停止編寫(xiě)本書(shū)，去做其他事情嗎？不可以。首先，一些平臺(tái)上進(jìn)程創(chuàng)建（甚至線程創(chuàng)建）的開(kāi)銷是很大的。現(xiàn)實(shí)中你可能想用線程池代替創(chuàng)建新進(jìn)程。然而，線程的擴(kuò)展性根本達(dá)不到期望。如果需要同時(shí)處理成千上萬(wàn)個(gè)連接，處理上萬(wàn)個(gè)線程的效率并不比在每個(gè)CPU上使用少量線程高。
如果線程不是處理多個(gè)連接的答案，那么什么是呢？在Unix世界中，可以使用非阻塞套接字：
fcntl(fd, F_SETFL, O_NONBLOCK);
這里fd是套接字的文件描述符。將fd（套接字）設(shè)置為非阻塞之后，對(duì)fd進(jìn)行網(wǎng)絡(luò)調(diào)用時(shí)，調(diào)用要么立即完成操作，要么返回一個(gè)特定的錯(cuò)誤號(hào)，指示“現(xiàn)在不能進(jìn)行操作，請(qǐng)重試”。這樣，示例程序可以寫(xiě)作：
/* This will work, but the performance will be unforgivably bad. */
int i, n;
char buf[1024];
for (i=0; i < n_sockets; ++i)
fcntl(fd[i], F_SETFL, O_NONBLOCK);
while (i_still_want_to_read()) {
for (i=0; i < n_sockets; ++i) {
n = recv(fd[i], buf, sizeof(buf), 0);
if (n == 0) {
handle_close(fd[i]);
} else if (n < 0) {
if (errno == EAGAIN)
; /* The kernel didn't have any data for us to read. */
else
handle_error(fd[i], errno);
} else {
handle_input(fd[i], buf, n);
}
}
}
使用非阻塞套接字，上述代碼可以工作，但只是在很少的情況下。程序性能將很糟糕，原因有兩個(gè)。首先，如果任何連接上都沒(méi)有數(shù)據(jù)可讀，循環(huán)還是會(huì)無(wú)限進(jìn)行，消耗CPU時(shí)間。第二，如果用這種方式處理多于一兩個(gè)連接，程序?qū)槊總€(gè)連接進(jìn)行內(nèi)核調(diào)用，不論連接上是否有數(shù)據(jù)。我們需要的是一種可以告訴內(nèi)核“等待這些套接字中的某一個(gè)有數(shù)據(jù)可讀，并且告知是哪一個(gè)”。
對(duì)于此問(wèn)題，現(xiàn)在仍然使用的最老的解決方案是select（）。select（）調(diào)用要求三個(gè)fd集合（作為位數(shù)組實(shí)現(xiàn)）：一個(gè)用于讀取，一個(gè)用于寫(xiě)入，一個(gè)用于異常。select（）將等待集合中的某個(gè)套接字就緒，并且修改集合，使之僅包含已經(jīng)就緒的套接字。
這是使用select的相同示例：
/* If you only have a couple dozen fds, this version won't be awful */
fd_set readset;
int i, n;
char buf[1024];
while (i_still_want_to_read()) {
int maxfd = -1;
FD_ZERO(&readset);
/* Add all of the interesting fds to readset */
for (i=0; i < n_sockets; ++i) {
if (fd[i]>maxfd) maxfd = fd[i];
FD_SET(fd[i], &readset);
}
/* Wait until one or more fds are ready to read */
select(maxfd+1, &readset, NULL, NULL, NULL);
/* Process all of the fds that are still set in readset */
for (i=0; i < n_sockets; ++i) {
if (FD_ISSET(fd[i], &readset)) {
n = recv(fd[i], buf, sizeof(buf), 0);
if (n == 0) {
handle_close(fd[i]);
} else if (n < 0) {
if (errno == EAGAIN)
; /* The kernel didn't have any data for us to read. */
else
handle_error(fd[i], errno);
} else {
handle_input(fd[i], buf, n);
}
}
}
}
這里是使用select重新實(shí)現(xiàn)的ROT13服務(wù)器：
/* For sockaddr_in */
#include <netinet/in.h>
/* For socket functions */
#include <sys/socket.h>
/* For fcntl */
#include <fcntl.h>
/* for select */
#include <sys/select.h>
#include <assert.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#define MAX_LINE 16384
char
rot13_char(char c)
{
/* We don't want to use isalpha here; setting the locale would change
* which characters are considered alphabetical. */
if ((c >= 'a' && c <= 'm') || (c >= 'A' && c <= 'M'))
return c + 13;
else if ((c >= 'n' && c <= 'z') || (c >= 'N' && c <= 'Z'))
return c - 13;
else
return c;
}
struct fd_state {
char buffer[MAX_LINE];
size_t buffer_used;
int writing;
size_t n_written;
size_t write_upto;
};
struct fd_state *
alloc_fd_state(void)
{
struct fd_state *state = malloc(sizeof(struct fd_state));
if (!state)
return NULL;
state->buffer_used = state->n_written = state->writing =
state->write_upto = 0;
return state;
}
void
free_fd_state(struct fd_state *state)
{
free(state);
}
void
make_nonblocking(int fd)
{
fcntl(fd, F_SETFL, O_NONBLOCK);
}
int
do_read(int fd, struct fd_state *state)
{
char buf[1024];
int i;
ssize_t result;
while (1) {
result = recv(fd, buf, sizeof(buf), 0);
if (result <= 0)
break;
for (i=0; i < result; ++i) {
if (state->buffer_used < sizeof(state->buffer))
state->buffer[state->buffer_used++] = rot13_char(buf[i]);
if (buf[i] == '\n') {
state->writing = 1;
state->write_upto = state->buffer_used;
}
}
}
if (result == 0) {
return 1;
} else if (result < 0) {
if (errno == EAGAIN)
return 0;
return -1;
}
return 0;
}
int
do_write(int fd, struct fd_state *state)
{
while (state->n_written < state->write_upto) {
ssize_t result = send(fd, state->buffer + state->n_written,
state->write_upto - state->n_written, 0);
if (result < 0) {
if (errno == EAGAIN)
return 0;
return -1;
}
assert(result != 0);
state->n_written += result;
}
if (state->n_written == state->buffer_used)
state->n_written = state->write_upto = state->buffer_used = 0;
state->writing = 0;
return 0;
}
void
run(void)
{
int listener;
struct fd_state *state[FD_SETSIZE];
struct sockaddr_in sin;
int i, maxfd;
fd_set readset, writeset, exset;
sin.sin_family = AF_INET;
sin.sin_addr.s_addr = 0;
sin.sin_port = htons(40713);
for (i = 0; i < FD_SETSIZE; ++i)
state[i] = NULL;
listener = socket(AF_INET, SOCK_STREAM, 0);
make_nonblocking(listener);
#ifndef WIN32
{
int one = 1;
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
}
#endif
if (bind(listener, (struct sockaddr*)&sin, sizeof(sin)) < 0) {
perror("bind");
return;
}
if (listen(listener, 16)<0) {
perror("listen");
return;
}
FD_ZERO(&readset);
FD_ZERO(&writeset);
FD_ZERO(&exset);
while (1) {
maxfd = listener;
FD_ZERO(&readset);
FD_ZERO(&writeset);
FD_ZERO(&exset);
FD_SET(listener, &readset);
for (i=0; i < FD_SETSIZE; ++i) {
if (state[i]) {
if (i > maxfd)
maxfd = i;
FD_SET(i, &readset);
if (state[i]->writing) {
FD_SET(i, &writeset);
}
}
}
if (select(maxfd+1, &readset, &writeset, &exset, NULL) < 0) {
perror("select");
return;
}
if (FD_ISSET(listener, &readset)) {
struct sockaddr_storage ss;
socklen_t slen = sizeof(ss);
int fd = accept(listener, (struct sockaddr*)&ss, &slen);
if (fd < 0) {
perror("accept");
} else if (fd > FD_SETSIZE) {
close(fd);
} else {
make_nonblocking(fd);
state[fd] = alloc_fd_state();
assert(state[fd]);/*XXX*/
}
}
for (i=0; i < maxfd+1; ++i) {
int r = 0;
if (i == listener)
continue;
if (FD_ISSET(i, &readset)) {
r = do_read(i, state[i]);
}
if (r == 0 && FD_ISSET(i, &writeset)) {
r = do_write(i, state[i]);
}
if (r) {
free_fd_state(state[i]);
state[i] = NULL;
close(i);
}
}
}
}
int
main(int c, char **v)
{
setvbuf(stdout, NULL, _IONBF, 0);
run();
return 0;
}
事情還沒(méi)完。因?yàn)樯珊妥x取select位數(shù)組所需的時(shí)間與用于select的最大fd成比例，所以當(dāng)套接字個(gè)數(shù)增加時(shí)，select調(diào)用的開(kāi)銷將急劇增加。
不同的操作系統(tǒng)為select提供了不同的替代功能，包括poll、epoll、kqueue、evports和/dev/poll。這些函數(shù)的性能都比select高，而且除了poll之外，添加、刪除套接字和通知套接字已經(jīng)準(zhǔn)備好IO的性能都是O（1）。
不幸的是，這些接口都不是標(biāo)準(zhǔn)的。Linux有epoll、BSD（包括Darwin）有kqueue、Solaris有evports和/dev/poll……，然而沒(méi)有哪個(gè)操作系統(tǒng)有其他系統(tǒng)所擁有的調(diào)用。所以，如果想編寫(xiě)可移植的高性能異步應(yīng)用，就需要一個(gè)封裝所有這些接口的抽象，提供這些調(diào)用中性能最高的一個(gè)供使用。
這就是Libevent API最底層所做的事情。Libevent為各種select替代提供了一致的接口，使用所運(yùn)行在的計(jì)算機(jī)上的最高效版本。
下面是另一個(gè)版本的異步ROT13服務(wù)器。這次用Libevent 2代替了select。注意fd_sets已經(jīng)被拋棄：替代的是，將事件與結(jié)構(gòu)體event_base關(guān)聯(lián)或者斷開(kāi)關(guān)聯(lián)，這可能是用select、poll、epoll或者kqueue實(shí)現(xiàn)的。
/* For sockaddr_in */
#include <netinet/in.h>
/* For socket functions */
#include <sys/socket.h>
/* For fcntl */
#include <fcntl.h>
#include <event2/event.h>
#include <assert.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#define MAX_LINE 16384
void do_read(evutil_socket_t fd, short events, void *arg);
void do_write(evutil_socket_t fd, short events, void *arg);
char
rot13_char(char c)
{
/* We don't want to use isalpha here; setting the locale would change
* which characters are considered alphabetical. */
if ((c >= 'a' && c <= 'm') || (c >= 'A' && c <= 'M'))
return c + 13;
else if ((c >= 'n' && c <= 'z') || (c >= 'N' && c <= 'Z'))
return c - 13;
else
return c;
}
struct fd_state {
char buffer[MAX_LINE];
size_t buffer_used;
size_t n_written;
size_t write_upto;
struct event *read_event;
struct event *write_event;
};
struct fd_state *
alloc_fd_state(struct event_base *base, evutil_socket_t fd)
{
struct fd_state *state = malloc(sizeof(struct fd_state));
if (!state)
return NULL;
state->read_event = event_new(base, fd, EV_READ|EV_PERSIST, do_read, state);
if (!state->read_event) {
free(state);
return NULL;
}
state->write_event =
event_new(base, fd, EV_WRITE|EV_PERSIST, do_write, state);
if (!state->write_event) {
event_free(state->read_event);
free(state);
return NULL;
}
state->buffer_used = state->n_written = state->write_upto = 0;
assert(state->write_event);
return state;
}
void
free_fd_state(struct fd_state *state)
{
event_free(state->read_event);
event_free(state->write_event);
free(state);
}
void
do_read(evutil_socket_t fd, short events, void *arg)
{
struct fd_state *state = arg;
char buf[1024];
int i;
ssize_t result;
while (1) {
assert(state->write_event);
result = recv(fd, buf, sizeof(buf), 0);
if (result <= 0)
break;
for (i=0; i < result; ++i) {
if (state->buffer_used < sizeof(state->buffer))
state->buffer[state->buffer_used++] = rot13_char(buf[i]);
if (buf[i] == '\n') {
assert(state->write_event);
event_add(state->write_event, NULL);
state->write_upto = state->buffer_used;
}
}
}
if (result == 0) {
free_fd_state(state);
} else if (result < 0) {
if (errno == EAGAIN) // XXXX use evutil macro
return;
perror("recv");
free_fd_state(state);
}
}
void
do_write(evutil_socket_t fd, short events, void *arg)
{
struct fd_state *state = arg;
while (state->n_written < state->write_upto) {
ssize_t result = send(fd, state->buffer + state->n_written,
state->write_upto - state->n_written, 0);
if (result < 0) {
if (errno == EAGAIN) // XXX use evutil macro
return;
free_fd_state(state);
return;
}
assert(result != 0);
state->n_written += result;
}
if (state->n_written == state->buffer_used)
state->n_written = state->write_upto = state->buffer_used = 1;
event_del(state->write_event);
}
void
do_accept(evutil_socket_t listener, short event, void *arg)
{
struct event_base *base = arg;
struct sockaddr_storage ss;
socklen_t slen = sizeof(ss);
int fd = accept(listener, (struct sockaddr*)&ss, &slen);
if (fd < 0) { // XXXX eagain??
perror("accept");
} else if (fd > FD_SETSIZE) {
close(fd); // XXX replace all closes with EVUTIL_CLOSESOCKET */
} else {
struct fd_state *state;
evutil_make_socket_nonblocking(fd);
state = alloc_fd_state(base, fd);
assert(state); /*XXX err*/
assert(state->write_event);
event_add(state->read_event, NULL);
}
}
void
run(void)
{
evutil_socket_t listener;
struct sockaddr_in sin;
struct event_base *base;
struct event *listener_event;
base = event_base_new();
if (!base)
return; /*XXXerr*/
sin.sin_family = AF_INET;
sin.sin_addr.s_addr = 0;
sin.sin_port = htons(40713);
listener = socket(AF_INET, SOCK_STREAM, 0);
evutil_make_socket_nonblocking(listener);
#ifndef WIN32
{
int one = 1;
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
}
#endif
if (bind(listener, (struct sockaddr*)&sin, sizeof(sin)) < 0) {
perror("bind");
return;
}
if (listen(listener, 16)<0) {
perror("listen");
return;
}
listener_event = event_new(base, listener, EV_READ|EV_PERSIST, do_accept, (void*)base);
/*XXX check it */
event_add(listener_event, NULL);
event_base_dispatch(base);
}
int
main(int c, char **v)
{
setvbuf(stdout, NULL, _IONBF, 0);
run();
return 0;
}
（代碼需要注意的其他地方：使用evutil_socket_t代替int來(lái)代表套接字；調(diào)用evutil_make_socket_nonblocking來(lái)將套接字設(shè)置為異步的，而不是調(diào)用fcntl（O_NONBLOCK）。這使得代碼兼容于Win32網(wǎng)絡(luò)API）
使用是否便捷？（還有Windows呢？）
你可能注意到代碼效率更高了，但是也更復(fù)雜了。使用fork的時(shí)候，（1）不需要為每個(gè)連接管理緩沖區(qū)：僅對(duì)每個(gè)進(jìn)程使用一個(gè)單獨(dú)的在棧上分配的緩沖區(qū)。（2）不需要顯式跟蹤每個(gè)套接字是否在讀取或者寫(xiě)入：這隱藏在代碼中了。（3）也不需要跟蹤每個(gè)操作是否完成的結(jié)構(gòu)體：只需要循環(huán)和棧變量。
此外，如果對(duì)Windows網(wǎng)絡(luò)有很深的體驗(yàn)，你將認(rèn)識(shí)到用于上述示例的時(shí)候，Libevent并不能取得優(yōu)化的性能。在Windows上進(jìn)行快速異步IO的方法不是使用select接口：而是使用IOCP。與其他快速網(wǎng)絡(luò)API不同的是，IOCP不是在套接字已經(jīng)準(zhǔn)備好某種操作時(shí)通知程序，然后程序可以進(jìn)行相應(yīng)的操作。替代的是，程序告知Windows網(wǎng)絡(luò)棧啟動(dòng)某網(wǎng)絡(luò)操作，IOCP在操作完成時(shí)通知程序。
幸運(yùn)的是，Libevent 2 的“bufferevent”接口解決了所有這些問(wèn)題：它提供了讓Libevent在Windows和Unix上都能夠有效實(shí)現(xiàn)的接口，讓程序編寫(xiě)更簡(jiǎn)單。
這是最后一個(gè)版本的ROT13，使用bufferevent API：
/* For sockaddr_in */
#include <netinet/in.h>
/* For socket functions */
#include <sys/socket.h>
/* For fcntl */
#include <fcntl.h>
#include <event2/event.h>
#include <event2/buffer.h>
#include <event2/bufferevent.h>
#include <assert.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#define MAX_LINE 16384
void do_read(evutil_socket_t fd, short events, void *arg);
void do_write(evutil_socket_t fd, short events, void *arg);
char
rot13_char(char c)
{
/* We don't want to use isalpha here; setting the locale would change
* which characters are considered alphabetical. */
if ((c >= 'a' && c <= 'm') || (c >= 'A' && c <= 'M'))
return c + 13;
else if ((c >= 'n' && c <= 'z') || (c >= 'N' && c <= 'Z'))
return c - 13;
else
return c;
}
void
readcb(struct bufferevent *bev, void *ctx)
{
struct evbuffer *input, *output;
char *line;
size_t n;
int i;
input = bufferevent_get_input(bev);
output = bufferevent_get_output(bev);
while ((line = evbuffer_readln(input, &n, EVBUFFER_EOL_LF))) {
for (i = 0; i < n; ++i)
line[i] = rot13_char(line[i]);
evbuffer_add(output, line, n);
evbuffer_add(output, "\n", 1);
free(line);
}
if (evbuffer_get_length(input) >= MAX_LINE) {
/* Too long; just process what there is and go on so that the buffer
* doesn't grow infinitely long. */
char buf[1024];
while (evbuffer_get_length(input)) {
int n = evbuffer_remove(input, buf, sizeof(buf));
for (i = 0; i < n; ++i)
buf[i] = rot13_char(buf[i]);
evbuffer_add(output, buf, n);
}
evbuffer_add(output, "\n", 1);
}
}
void
errorcb(struct bufferevent *bev, short error, void *ctx)
{
if (error & BEV_EVENT_EOF) {
/* connection has been closed, do any clean up here */
/* 
*/
} else if (error & BEV_EVENT_ERROR) {
/* check errno to see what error occurred */
/* */
} else if (error & BEV_EVENT_TIMEOUT) {
/* must be a timeout event handle, handle it */
/* */
}
bufferevent_free(bev);
}
void
do_accept(evutil_socket_t listener, short event, void *arg)
{
struct event_base *base = arg;
struct sockaddr_storage ss;
socklen_t slen = sizeof(ss);
int fd = accept(listener, (struct sockaddr*)&ss, &slen);
if (fd < 0) {
perror("accept");
} else if (fd > FD_SETSIZE) {
close(fd);
} else {
struct bufferevent *bev;
evutil_make_socket_nonblocking(fd);
bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
bufferevent_setcb(bev, readcb, NULL, errorcb, NULL);
bufferevent_setwatermark(bev, EV_READ, 0, MAX_LINE);
bufferevent_enable(bev, EV_READ|EV_WRITE);
}
}
void
run(void)
{
evutil_socket_t listener;
struct sockaddr_in sin;
struct event_base *base;
struct event *listener_event;
base = event_base_new();
if (!base)
return; /*XXXerr*/
sin.sin_family = AF_INET;
sin.sin_addr.s_addr = 0;
sin.sin_port = htons(40713);
listener = socket(AF_INET, SOCK_STREAM, 0);
evutil_make_socket_nonblocking(listener);
#ifndef WIN32
{
int one = 1;
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
}
#endif
if (bind(listener, (struct sockaddr*)&sin, sizeof(sin)) < 0) {
perror("bind");
return;
}
if (listen(listener, 16)<0) {
perror("listen");
return;
}
listener_event = event_new(base, listener, EV_READ|EV_PERSIST, do_accept, (void*)base);
/*XXX check it */
event_add(listener_event, NULL);
event_base_dispatch(base);
}
int
main(int c, char **v)
{
setvbuf(stdout, NULL, _IONBF, 0);
run();
return 0;
} 這些真的很高效嗎？
XXXX將在此寫(xiě)有效的一節(jié)。libevent網(wǎng)頁(yè)上的benchmarks數(shù)據(jù)真的過(guò)期了。