epoll方法實現non-blocking socket
event-based方法和epoll
epoll是event-based的方法實現異步io/non-blocking io。從Linux kernel 2.5.44之后epoll加入Linux kernel中，代替loop style方法的select和poll，比后者更加高效更適用于高并發多client的應用。loop style方法的時間復雜度為O(n)(因為需要線性地檢測指定的file descriptor)，而epoll等event-based方法的時間復雜度為O(1)。event-based通過為不同的events設置callback函數，在該event發生的時候自動執行相應函數(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)等。因此如果程序運行目標平臺是Linux(Kernel > 2.5.44)可以使用epoll即可，如果其他Unix平臺可以考慮相應的方法(如在FreeBSD等UNIX上使用kqueue)。如果希望使用跨平臺可移植的event-based可以使用libevent庫，它支持多種方法（select/poll, epoll, kqueue, /dev/poll等）。
epoll提供的系統調用
epoll包含了3個系統調用：epoll_create, epoll_ctl和epoll_wait(需#include <sys/epoll.h>)。具體步驟是：首先使用epoll_create創建一個epoll file descriptor；然后使用epoll_ctl添加要監聽的IO file descriptor到epoll中；最后循環地調用epoll_wait檢測各IO fd相關events的變化，然后采取相應的措施。
epoll_create
int epoll_create(int size); //創建epoll
其中size告知kernel需要為之后添加的IO file descriptor準備的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創建的epoll file descriptor在程序結束的時候需要使用close()將其關閉，例如：
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創建的epoll file descriptor
//op: 指定要對指定的fd進行何種操作，支持的操作包括：
EPOLL_CTL_ADD 將指定的file descriptor添加到epoll中
EPOLL_CTL_MOD 修改指定file descriptor的event，相當于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不需要設置
EPOLLHUP //hang up happended on the fd, epoll_wait會默認檢測該event不需要設置
EPOLLET //設置使用Edge Triggered模式，epoll模式使用Level Triggered
更多參看：http://linux.die.net/man/2/epoll_ctl
epoll_ctl的返回值：成功返回0, 發生錯誤返回-1
epoll_wait
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
//epfd 為如上定義的epoll file descriptor
//events 返回發生改變的events
//maxevents 最多返回events的個數，必須 > 0
//timeout 等待的milliseconds；和poll類似，如果timeout設置為-1則epoll_wait將持續等待下去；如果timeout設置為0，則epoll_wait將立即返回
該函數類似select和poll函數，執行的時候會等待直到epfd定義的指定IO fd的events發生變化或timeout參數指定的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表示發生錯誤。
epoll檢測events改變的兩種模式：edge-triggered和level-triggered
調用epoll_wait會返回events發生變化的IO fd，epoll支持兩種模式：
只要發生的events沒有結束，每次調用epoll_wait都顯示該events存在。例如：當一個IO fd的狀態變為available for reading的時候，調用epoll_wait會將該event返回；如果下次調用epoll_wait的時候該read過程還沒有完成，則epoll_wait仍舊會返回該event。
和level triggered不同，它只在event產生的時候發出event信息，之后即使event沒有結束不再發送此信息。例如：當一個IO fd狀態變為available for reading的時候，調用epoll_wait會將該event返回；如果下次調用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模式，在調用epoll_ctl的時候指定其struct epoll_event的events的時候添加EPOLLET。
epoll實現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個數
#define EPOLL_TIMEOUT 5000 //epoll_wait的timeout milliseconds
//函數：設置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);
}
//創建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);
}
//設置sock為non-blocking
setSockNonBlock(sock);
//創建要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); //設置接受任意地址
bind_addr.sin_port = htons(port); //將host byte order轉換為network byte order
//bind sock到創建的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");
//創建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的參數
struct epoll_event events[EPOLL_MAXEVENTS];
memset(events, 0, sizeof(events));
//循環偵聽
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) {
//每次循環調用依次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進行響應
for (i = 0; i < res; i++) {
//events[i]即為檢測到的event，域events[i].events表示具體哪些events，域events[i].data.fd即對應的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進行響應
if (events[i].data.fd == sock) {
//當前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);
//設置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 {
//當前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表示當前無可讀數據，并不表示錯誤
perror("recv failed");
exit(EXIT_FAILURE);
}
printf("recved from conn_sock=%d : %s(%d length string)\n", conn_sock, buffer, recv_size);
//立即將收到的內容寫回去
if ( send(conn_sock, buffer, recv_size, 0) == -1 && (errno != EAGAIN) && (errno != EWOULDBLOCK) ) {
//send在non-blocking模式下，返回-1且errno為EAGAIN或EWOULDBLOCK表示當前無可寫數據，并不表示錯誤
perror("send failed");
exit(EXIT_FAILURE);
}
printf("send to conn_sock=%d done\n", conn_sock);
//將當前socket從epoll的偵聽中移除(有文章說：關閉con_sock之后，其會自動從epoll中刪除，因此此段代碼可以省略)
if ( epoll_ctl(epfd, EPOLL_CTL_DEL, conn_sock, NULL) == -1 ) {
perror("epoll_ctl(EPOLL_CTL_DEL) failed");
exit(EXIT_FAILURE);
}
//關閉連接
if ( close(conn_sock) == -1 ) {
perror("close failed");
exit(EXIT_FAILURE);
}
printf("close conn_sock=%d done\n", conn_sock);
}
}
}
close(sock); //關閉server的listening socket
close(epfd); //關閉epoll file descriptor
return 0;
}
測試：編譯并運行程序；然后嘗試運行多個telnet localhost 1984和server進行通信。
注意：epoll在使用epoll_ctl為file descriptor指定events的時候，默認采用Level Triggered，即如果events未完成調用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的時候，注意數據讀取完整)。可參考該文章介紹的代碼。
小結：
epoll這種event-based的方法比較1。select/poll等loop style方法；2。多進程(forking)/多線程(threading)方法(每個進程或線程對應一個connection socket)，在多client高并發下性能更優越。因此推薦在實際中應用。例如：Nginx, Lighttpd, Memcached等都采用有該event-based的異步IO模型。
另外，event-handling庫libevent也支持epoll方法(還支持kqueue(FreeBSD/NetBSD/OpenBSD/Darwin), /dev/poll(Solaris/HPUX), select, poll等方法)在實際中也可使用該庫編寫高性能的Server，方便實現跨平臺可移植。