epoll方法實(shí)現(xiàn)non-blocking socket
event-based方法和epoll
epoll是event-based的方法實(shí)現(xiàn)異步io/non-blocking io。從Linux kernel 2.5.44之后epoll加入Linux kernel中，代替loop style方法的select和poll，比后者更加高效更適用于高并發(fā)多client的應(yīng)用。loop style方法的時(shí)間復(fù)雜度為O(n)(因?yàn)樾枰€性地檢測(cè)指定的file descriptor)，而epoll等event-based方法的時(shí)間復(fù)雜度為O(1)。event-based通過(guò)為不同的events設(shè)置callback函數(shù)，在該event發(fā)生的時(shí)候自動(dòng)執(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)等。因此如果程序運(yùn)行目標(biāo)平臺(tái)是Linux(Kernel > 2.5.44)可以使用epoll即可，如果其他Unix平臺(tái)可以考慮相應(yīng)的方法(如在FreeBSD等UNIX上使用kqueue)。如果希望使用跨平臺(tái)可移植的event-based可以使用libevent庫(kù)，它支持多種方法（select/poll, epoll, kqueue, /dev/poll等）。
epoll提供的系統(tǒng)調(diào)用
epoll包含了3個(gè)系統(tǒng)調(diào)用：epoll_create, epoll_ctl和epoll_wait(需#include <sys/epoll.h>)。具體步驟是：首先使用epoll_create創(chuàng)建一個(gè)epoll file descriptor；然后使用epoll_ctl添加要監(jiān)聽(tīng)的IO file descriptor到epoll中；最后循環(huán)地調(diào)用epoll_wait檢測(cè)各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開(kāi)始size值不再被使用，但是其賦值需要> 0。參考: http://www.kernel.org/doc/man-pages/online/pages/man2/epoll_create.2.html
使用epoll_create創(chuàng)建的epoll file descriptor在程序結(jié)束的時(shí)候需要使用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: 指定要對(duì)指定的fd進(jìn)行何種操作，支持的操作包括：
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域是一個(gè)bit set(可以通過(guò)|操作符進(jìn)行多賦值)，支持的event type有：
EPOLLIN //ready for read
EPOLLOUT //ready for write
EPOLLPRI //urgent data available for read
EPOLLERR //error condition happened, epoll_wait會(huì)默認(rèn)檢測(cè)該event不需要設(shè)置
EPOLLHUP //hang up happended on the fd, epoll_wait會(huì)默認(rèn)檢測(cè)該event不需要設(shè)置
EPOLLET //設(shè)置使用Edge Triggered模式，epoll模式使用Level Triggered
更多參看：http://linux.die.net/man/2/epoll_ctl
epoll_ctl的返回值：成功返回0, 發(fā)生錯(cuò)誤返回-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的個(gè)數(shù)，必須 > 0
//timeout 等待的milliseconds；和poll類似，如果timeout設(shè)置為-1則epoll_wait將持續(xù)等待下去；如果timeout設(shè)置為0，則epoll_wait將立即返回
該函數(shù)類似select和poll函數(shù)，執(zhí)行的時(shí)候會(huì)等待直到epfd定義的指定IO fd的events發(fā)生變化或timeout參數(shù)指定的milliseconds時(shí)間到期才返回。wait and block until events on the watched set happens or timeout expires。
返回值：成功返回number of fd ready for requested io; 0表示在timeout以后沒(méi)有ready的fd; -1表示發(fā)生錯(cuò)誤。
epoll檢測(cè)events改變的兩種模式：edge-triggered和level-triggered
調(diào)用epoll_wait會(huì)返回events發(fā)生變化的IO fd，epoll支持兩種模式：
只要發(fā)生的events沒(méi)有結(jié)束，每次調(diào)用epoll_wait都顯示該events存在。例如：當(dāng)一個(gè)IO fd的狀態(tài)變?yōu)閍vailable for reading的時(shí)候，調(diào)用epoll_wait會(huì)將該event返回；如果下次調(diào)用epoll_wait的時(shí)候該read過(guò)程還沒(méi)有完成，則epoll_wait仍舊會(huì)返回該event。
和level triggered不同，它只在event產(chǎn)生的時(shí)候發(fā)出event信息，之后即使event沒(méi)有結(jié)束不再發(fā)送此信息。例如：當(dāng)一個(gè)IO fd狀態(tài)變?yōu)閍vailable for reading的時(shí)候，調(diào)用epoll_wait會(huì)將該event返回；如果下次調(diào)用epoll_wait的時(shí)候該read過(guò)程還沒(méi)有完成，epoll_wait不會(huì)立即返回而是需要等待新的events或直到timetout的時(shí)間。Edge Triggered event distribution delivers events only when events happens on the monitored file.
epoll默認(rèn)采用Level Triggered模式，如果需要對(duì)某個(gè)IO fd采用Edge Triggered模式，在調(diào)用epoll_ctl的時(shí)候指定其struct epoll_event的events的時(shí)候添加EPOLLET。
epoll實(shí)現(xiàn)non-blocking socket實(shí)例：
#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 //默認(rèn)端口
#define BUFF_SIZE 1024 //buffer大小
#define EPOLL_MAXEVENTS 64 //epoll_wait的最多返回的events個(gè)數(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)偵聽(tīng)
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偵聽(tīng)
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;
}
//檢測(cè)到res個(gè)IO file descriptor的events，loop各個(gè)fd進(jìn)行響應(yīng)
for (i = 0; i < res; i++) {
//events[i]即為檢測(cè)到的event，域events[i].events表示具體哪些events，域events[i].data.fd即對(duì)應(yīng)的IO fd
if ( (events[i].events & EPOLLERR) ||
(events[i].events & EPOLLHUP) ||
(!(events[i].events & EPOLLIN)) ) {
//由于events[i].events使用每個(gè)bit表示event，因此判斷是否包含某個(gè)具體事件可以使用`&`操作符
//這里判斷是否存在EPOLLERR, EPOLLHUP等event
fprintf (stderr, "epoll error\n");
close (events[i].data.fd);
continue;
}
//對(duì)檢測(cè)到event的各IO fd進(jìn)行響應(yīng)
if (events[i].data.fd == sock) {
//當(dāng)前fd是server的socket，不進(jìn)行讀而是accept所有client連接請(qǐng)求
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模式下無(wú)新connection請(qǐng)求，跳出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的偵聽(tīng)中
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)前無(wú)可讀數(shù)據(jù)，并不表示錯(cuò)誤
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)前無(wú)可寫數(shù)據(jù)，并不表示錯(cuò)誤
perror("send failed");
exit(EXIT_FAILURE);
}
printf("send to conn_sock=%d done\n", conn_sock);
//將當(dāng)前socket從epoll的偵聽(tīng)中移除(有文章說(shuō)：關(guān)閉con_sock之后，其會(huì)自動(dòng)從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;
}
測(cè)試：編譯并運(yùn)行程序；然后嘗試運(yùn)行多個(gè)telnet localhost 1984和server進(jìn)行通信。
注意：epoll在使用epoll_ctl為file descriptor指定events的時(shí)候，默認(rèn)采用Level Triggered，即如果events未完成調(diào)用epoll_wait的話每次都會(huì)返回該事件；通過(guò)如下方式：
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檢測(cè)到每次事件變化只通知一次，因此在epoll_wait之后的處理的時(shí)候需要注意(例如有可讀的event的時(shí)候，注意數(shù)據(jù)讀取完整)。可參考該文章介紹的代碼。
小結(jié)：
epoll這種event-based的方法比較1。select/poll等loop style方法；2。多進(jìn)程(forking)/多線程(threading)方法(每個(gè)進(jìn)程或線程對(duì)應(yīng)一個(gè)connection socket)，在多client高并發(fā)下性能更優(yōu)越。因此推薦在實(shí)際中應(yīng)用。例如：Nginx, Lighttpd, Memcached等都采用有該event-based的異步IO模型。
另外，event-handling庫(kù)libevent也支持epoll方法(還支持kqueue(FreeBSD/NetBSD/OpenBSD/Darwin), /dev/poll(Solaris/HPUX), select, poll等方法)在實(shí)際中也可使用該庫(kù)編寫高性能的Server，方便實(shí)現(xiàn)跨平臺(tái)可移植。