epoll方法實現(xiàn)non-blocking socket
event-based方法和epoll
epoll是event-based的方法實現(xiàn)異步io/non-blocking io。從Linux kernel 2.5.44之后epoll加入Linux kernel中，代替loop style方法的select和poll，比后者更加高效更適用于高并發(fā)多client的應(yīng)用。loop style方法的時間復(fù)雜度為O(n)(因為需要線性地檢測指定的file descriptor)，而epoll等event-based方法的時間復(fù)雜度為O(1)。event-based通過為不同的events設(shè)置callback函數(shù)，在該event發(fā)生的時候自動執(zhí)行相應(yīng)函數(shù)(epoll uses callbacks in the kernel file structure)。
和epoll類似的其他event-based的方法還有：kqueue(FreeBSD/NetBSD/OpenBSD/Darwin), /dev/poll(Solaris/HPUX), pollset(AIX), Event Completion(Solaris 10), I/O Completion Ports(Windows)等。因此如果程序運行目標(biāo)平臺是Linux(Kernel > 2.5.44)可以使用epoll即可，如果其他Unix平臺可以考慮相應(yīng)的方法(如在FreeBSD等UNIX上使用kqueue)。如果希望使用跨平臺可移植的event-based可以使用libevent庫，它支持多種方法（select/poll, epoll, kqueue, /dev/poll等）。
epoll提供的系統(tǒng)調(diào)用
epoll包含了3個系統(tǒng)調(diào)用：epoll_create, epoll_ctl和epoll_wait(需#include <sys/epoll.h>)。具體步驟是：首先使用epoll_create創(chuàng)建一個epoll file descriptor；然后使用epoll_ctl添加要監(jiān)聽的IO file descriptor到epoll中；最后循環(huán)地調(diào)用epoll_wait檢測各IO fd相關(guān)events的變化，然后采取相應(yīng)的措施。
epoll_create
int epoll_create(int size); //創(chuàng)建epoll
其中size告知kernel需要為之后添加的IO file descriptor準(zhǔn)備的event backing store的大小。Open an epoll file descriptor by requesting the kernel allocate an event backing store dimensioned for size descriptors. The size is not the maximum size of the backing store but just a hint to the kernel about how to dimension internal structures.
從Linux 2.6.8開始size值不再被使用，但是其賦值需要> 0。參考: http://www.kernel.org/doc/man-pages/online/pages/man2/epoll_create.2.html
使用epoll_create創(chuàng)建的epoll file descriptor在程序結(jié)束的時候需要使用close()將其關(guān)閉，例如：
int epfd;
if ( (epfd = epoll_create(1)) == -1 ) {
perror("epoll_create failed");
exit(EXIT_FAILURE);
}
...
close(epfd);
epoll_ctl
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event); // add/update/del IO file descriptor to be watched on the epoll instance
//epfd: 即是如上epoll_create創(chuàng)建的epoll file descriptor
//op: 指定要對指定的fd進行何種操作，支持的操作包括：
EPOLL_CTL_ADD 將指定的file descriptor添加到epoll中
EPOLL_CTL_MOD 修改指定file descriptor的event，相當(dāng)于update
EPOLL_CTL_DEL 將指定的file descriptor從epoll中清除，the event is ignored and can be NULL
//fd: 要操作的IO file descriptor
//event: 表示the event linked to this file descriptor
其中struct epoll_event的定義如下：
struct epoll_event {
__uint32_t events; //epoll events
epoll_data_t data; //user data variable
};
typedef union epoll_data {
void *ptr;
int fd;
__uint32_t u32;
__uint64_t u64;
} epoll_data_t;
其中的struct epoll_event的events域是一個bit set(可以通過|操作符進行多賦值)，支持的event type有：
EPOLLIN //ready for read
EPOLLOUT //ready for write
EPOLLPRI //urgent data available for read
EPOLLERR //error condition happened, epoll_wait會默認檢測該event不需要設(shè)置
EPOLLHUP //hang up happended on the fd, epoll_wait會默認檢測該event不需要設(shè)置
EPOLLET //設(shè)置使用Edge Triggered模式，epoll模式使用Level Triggered
更多參看：http://linux.die.net/man/2/epoll_ctl
epoll_ctl的返回值：成功返回0, 發(fā)生錯誤返回-1
epoll_wait
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
//epfd 為如上定義的epoll file descriptor
//events 返回發(fā)生改變的events
//maxevents 最多返回events的個數(shù)，必須 > 0
//timeout 等待的milliseconds；和poll類似，如果timeout設(shè)置為-1則epoll_wait將持續(xù)等待下去；如果timeout設(shè)置為0，則epoll_wait將立即返回
該函數(shù)類似select和poll函數(shù)，執(zhí)行的時候會等待直到epfd定義的指定IO fd的events發(fā)生變化或timeout參數(shù)指定的milliseconds時間到期才返回。wait and block until events on the watched set happens or timeout expires。
返回值：成功返回number of fd ready for requested io; 0表示在timeout以后沒有ready的fd; -1表示發(fā)生錯誤。
epoll檢測events改變的兩種模式：edge-triggered和level-triggered
調(diào)用epoll_wait會返回events發(fā)生變化的IO fd，epoll支持兩種模式：
只要發(fā)生的events沒有結(jié)束，每次調(diào)用epoll_wait都顯示該events存在。例如：當(dāng)一個IO fd的狀態(tài)變?yōu)閍vailable for reading的時候，調(diào)用epoll_wait會將該event返回；如果下次調(diào)用epoll_wait的時候該read過程還沒有完成，則epoll_wait仍舊會返回該event。
和level triggered不同，它只在event產(chǎn)生的時候發(fā)出event信息，之后即使event沒有結(jié)束不再發(fā)送此信息。例如：當(dāng)一個IO fd狀態(tài)變?yōu)閍vailable for reading的時候，調(diào)用epoll_wait會將該event返回；如果下次調(diào)用epoll_wait的時候該read過程還沒有完成，epoll_wait不會立即返回而是需要等待新的events或直到timetout的時間。Edge Triggered event distribution delivers events only when events happens on the monitored file.
epoll默認采用Level Triggered模式，如果需要對某個IO fd采用Edge Triggered模式，在調(diào)用epoll_ctl的時候指定其struct epoll_event的events的時候添加EPOLLET。
epoll實現(xiàn)non-blocking socket實例：
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include <fcntl.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <errno.h>
#define DEFAULT_PORT 1984 //默認端口
#define BUFF_SIZE 1024 //buffer大小
#define EPOLL_MAXEVENTS 64 //epoll_wait的最多返回的events個數(shù)
#define EPOLL_TIMEOUT 5000 //epoll_wait的timeout milliseconds
//函數(shù)：設(shè)置sock為non-blocking mode
void setSockNonBlock(int sock) {
int flags;
flags = fcntl(sock, F_GETFL, 0);
if (flags < 0) {
perror("fcntl(F_GETFL) failed");
exit(EXIT_FAILURE);
}
if (fcntl(sock, F_SETFL, flags | O_NONBLOCK) < 0) {
perror("fcntl(F_SETFL) failed");
exit(EXIT_FAILURE);
}
}
int main(int argc, char *argv[]) {
//獲取自定義端口
unsigned short int port;
if (argc == 2) {
port = atoi(argv[1]);
} else if (argc < 2) {
port = DEFAULT_PORT;
} else {
fprintf(stderr, "USAGE: %s [port]\n", argv[0]);
exit(EXIT_FAILURE);
}
//創(chuàng)建socket
int sock;
if ( (sock = socket(PF_INET, SOCK_STREAM, 0)) == -1 ) {
perror("socket failed");
exit(EXIT_FAILURE);
}
printf("socket done\n");
//in case of 'address already in use' error message
int yes = 1;
if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int))) {
perror("setsockopt failed");
exit(EXIT_FAILURE);
}
//設(shè)置sock為non-blocking
setSockNonBlock(sock);
//創(chuàng)建要bind的socket address
struct sockaddr_in bind_addr;
memset(&bind_addr, 0, sizeof(bind_addr));
bind_addr.sin_family = AF_INET;
bind_addr.sin_addr.s_addr = htonl(INADDR_ANY); //設(shè)置接受任意地址
bind_addr.sin_port = htons(port); //將host byte order轉(zhuǎn)換為network byte order
//bind sock到創(chuàng)建的socket address上
if ( bind(sock, (struct sockaddr *) &bind_addr, sizeof(bind_addr)) == -1 ) {
perror("bind failed");
exit(EXIT_FAILURE);
}
printf("bind done\n");
//listen
if ( listen(sock, 5) == -1) {
perror("listen failed");
exit(EXIT_FAILURE);
}
printf("listen done\n");
//創(chuàng)建epoll (epoll file descriptor)
int epfd;
if ( (epfd = epoll_create(1)) == -1 ) {
perror("epoll_create failed");
exit(EXIT_FAILURE);
}
//將sock添加到epoll中
struct epoll_event event;
event.events = EPOLLIN;
event.data.fd = sock;
if ( epoll_ctl(epfd, EPOLL_CTL_ADD, sock, &event) == -1 ) {
perror("epoll_ctl");
exit(EXIT_FAILURE);
}
//初始化epoll_wait的參數(shù)
struct epoll_event events[EPOLL_MAXEVENTS];
memset(events, 0, sizeof(events));
//循環(huán)偵聽
int conn_sock;
struct sockaddr_in client_addr;
socklen_t client_addr_len;
char client_ip_str[INET_ADDRSTRLEN];
int res;
int i;
char buffer[BUFF_SIZE];
int recv_size;
while (1) {
//每次循環(huán)調(diào)用依次epoll_wait偵聽
res = epoll_wait(epfd, events, EPOLL_MAXEVENTS, EPOLL_TIMEOUT);
if (res < 0) {
perror("epoll_wait failed");
exit(EXIT_FAILURE);
} else if (res == 0) {
fprintf(stderr, "no socket ready for read within %d secs\n", EPOLL_TIMEOUT / 1000);
continue;
}
//檢測到res個IO file descriptor的events，loop各個fd進行響應(yīng)
for (i = 0; i < res; i++) {
//events[i]即為檢測到的event，域events[i].events表示具體哪些events，域events[i].data.fd即對應(yīng)的IO fd
if ( (events[i].events & EPOLLERR) ||
(events[i].events & EPOLLHUP) ||
(!(events[i].events & EPOLLIN)) ) {
//由于events[i].events使用每個bit表示event，因此判斷是否包含某個具體事件可以使用`&`操作符
//這里判斷是否存在EPOLLERR, EPOLLHUP等event
fprintf (stderr, "epoll error\n");
close (events[i].data.fd);
continue;
}
//對檢測到event的各IO fd進行響應(yīng)
if (events[i].data.fd == sock) {
//當(dāng)前fd是server的socket，不進行讀而是accept所有client連接請求
while (1) {
client_addr_len = sizeof(client_addr);
conn_sock = accept(sock, (struct sockaddr *) &client_addr, &client_addr_len);
if (conn_sock == -1) {
if ( (errno == EAGAIN) || (errno == EWOULDBLOCK) ) {
//non-blocking模式下無新connection請求，跳出while (1)
break;
} else {
perror("accept failed");
exit(EXIT_FAILURE);
}
}
if (!inet_ntop(AF_INET, &(client_addr.sin_addr), client_ip_str, sizeof(client_ip_str))) {
perror("inet_ntop failed");
exit(EXIT_FAILURE);
}
printf("accept a client from: %s\n", client_ip_str);
//設(shè)置conn_sock為non-blocking
setSockNonBlock(conn_sock);
//把conn_sock添加到epoll的偵聽中
event.events = EPOLLIN;
event.data.fd = conn_sock;
if ( epoll_ctl(epfd, EPOLL_CTL_ADD, conn_sock, &event) == -1 ) {
perror("epoll_ctl(EPOLL_CTL_ADD) failed");
exit(EXIT_FAILURE);
}
}
} else {
//當(dāng)前fd是client連接的socket，可以讀(read from client)
conn_sock = events[i].data.fd;
memset(buffer, 0, sizeof(buffer));
if ( (recv_size = recv(conn_sock, buffer, sizeof(buffer), 0)) == -1 && (errno != EAGAIN) ) {
//recv在non-blocking模式下，返回-1且errno為EAGAIN表示當(dāng)前無可讀數(shù)據(jù)，并不表示錯誤
perror("recv failed");
exit(EXIT_FAILURE);
}
printf("recved from conn_sock=%d : %s(%d length string)\n", conn_sock, buffer, recv_size);
//立即將收到的內(nèi)容寫回去
if ( send(conn_sock, buffer, recv_size, 0) == -1 && (errno != EAGAIN) && (errno != EWOULDBLOCK) ) {
//send在non-blocking模式下，返回-1且errno為EAGAIN或EWOULDBLOCK表示當(dāng)前無可寫數(shù)據(jù)，并不表示錯誤
perror("send failed");
exit(EXIT_FAILURE);
}
printf("send to conn_sock=%d done\n", conn_sock);
//將當(dāng)前socket從epoll的偵聽中移除(有文章說：關(guān)閉con_sock之后，其會自動從epoll中刪除，因此此段代碼可以省略)
if ( epoll_ctl(epfd, EPOLL_CTL_DEL, conn_sock, NULL) == -1 ) {
perror("epoll_ctl(EPOLL_CTL_DEL) failed");
exit(EXIT_FAILURE);
}
//關(guān)閉連接
if ( close(conn_sock) == -1 ) {
perror("close failed");
exit(EXIT_FAILURE);
}
printf("close conn_sock=%d done\n", conn_sock);
}
}
}
close(sock); //關(guān)閉server的listening socket
close(epfd); //關(guān)閉epoll file descriptor
return 0;
}
測試：編譯并運行程序；然后嘗試運行多個telnet localhost 1984和server進行通信。
注意：epoll在使用epoll_ctl為file descriptor指定events的時候，默認采用Level Triggered，即如果events未完成調(diào)用epoll_wait的話每次都會返回該事件；通過如下方式：
struct epoll_event event;
event.data.fd = sock;
event.events = EPOLLIN | EPOLLET;
if ( epoll_ctl(epfd, EPOLL_CTL_ADD, sock, &event) == -1 ) {
perror("epoll_ctl(EPOLL_CTL_ADD) failed");
exit(EXIT_FAILURE);
}
可以指定該fd的event采用Edge Triggered模型，如果采用該模型，epoll_wait檢測到每次事件變化只通知一次，因此在epoll_wait之后的處理的時候需要注意(例如有可讀的event的時候，注意數(shù)據(jù)讀取完整)。可參考該文章介紹的代碼。
小結(jié)：
epoll這種event-based的方法比較1。select/poll等loop style方法；2。多進程(forking)/多線程(threading)方法(每個進程或線程對應(yīng)一個connection socket)，在多client高并發(fā)下性能更優(yōu)越。因此推薦在實際中應(yīng)用。例如：Nginx, Lighttpd, Memcached等都采用有該event-based的異步IO模型。
另外，event-handling庫libevent也支持epoll方法(還支持kqueue(FreeBSD/NetBSD/OpenBSD/Darwin), /dev/poll(Solaris/HPUX), select, poll等方法)在實際中也可使用該庫編寫高性能的Server，方便實現(xiàn)跨平臺可移植。