現在有這么一個場景：我是一個很忙的大老板，我有100個手機，手機來信息了，我的秘書就會告訴我“老板，你的手機來信息了。”我很生氣，我的秘書就是這樣子，每次手機來信息就只告訴我來信息了，老板趕緊去看。但是她從來不把話說清楚：到底是哪個手機來信息啊！我可有100個手機啊！於是，我只能一個一個手機去查看，來確定到底是哪幾個手機來信息了。這就是IO復用中select模型的缺點！老板心想，要是秘書能把來信息的手機直接拿到我桌子上就好了，那么我的效率肯定大增（這就是epoll模型）。
那我們先來總結一下select模型的缺點：
- 單個進程能夠監視的文件描述符的數量存在最大限制，通常是1024，當然可以更改數量，但由於select采用輪詢的方式掃描文件描述符，文件描述符數量越多，性能越差；(在linux內核頭文件中，有這樣的定義：#define __FD_SETSIZE 1024)
- 內核 / 用戶空間內存拷貝問題，select需要復制大量的句柄數據結構，產生巨大的開銷；
select返回的是含有整個句柄的數組，應用程序需要遍歷整個數組才能發現哪些句柄發生了事件； - select的觸發方式是水平觸發，應用程序如果沒有完成對一個已經就緒的文件描述符進行IO操作，那么之后每次select調用還是會將這些文件描述符通知進程。
設想一下如下場景：有100萬個客戶端同時與一個服務器進程保持着TCP連接。而每一時刻，通常只有幾百上千個TCP連接是活躍的(事實上大部分場景都是這種情況)。如何實現這樣的高並發？
粗略計算一下，一個進程最多有1024個文件描述符，那么我們需要開1000個進程來處理100萬個客戶連接。如果我們使用select模型，這1000個進程里某一段時間內只有數個客戶連接需要數據的接收，那么我們就不得不輪詢1024個文件描述符以確定究竟是哪個客戶有數據可讀，想想如果1000個進程都有類似的行為，那系統資源消耗可有多大啊！
針對select模型的缺點，epoll模型被提出來了！
epoll模型的優點
- 支持一個進程打開大數目的socket描述符
- IO效率不隨FD數目增加而線性下降
- 使用mmap加速內核與用戶空間的消息傳遞
epoll的兩種工作模式
-
LT(level triggered，水平觸發模式)是缺省的工作方式，並且同時支持 block 和 non-block socket。在這種做法中，內核告訴你一個文件描述符是否就緒了，然后你可以對這個就緒的fd進行IO操作。如果你不作任何操作，內核還是會繼續通知你的，所以，這種模式編程出錯誤可能性要小一點。比如內核通知你其中一個fd可以讀數據了，你趕緊去讀。你還是懶懶散散，不去讀這個數據，下一次循環的時候內核發現你還沒讀剛才的數據，就又通知你趕緊把剛才的數據讀了。這種機制可以比較好的保證每個數據用戶都處理掉了。
-
ET(edge-triggered，邊緣觸發模式)是高速工作方式，只支持no-block socket。在這種模式下，當描述符從未就緒變為就緒時，內核通過epoll告訴你。然后它會假設你知道文件描述符已經就緒，並且不會再為那個文件描述符發送更多的就緒通知，等到下次有新的數據進來的時候才會再次出發就緒事件。簡而言之，就是內核通知過的事情不會再說第二遍，數據錯過沒讀，你自己負責。這種機制確實速度提高了，但是風險相伴而行。
epoll模型API
#include <sys/epoll.h>
/* 創建一個epoll的句柄，size用來告訴內核需要監聽的數目一共有多大。當創建好epoll句柄后，
它就是會占用一個fd值，所以在使用完epoll后，必須調用close()關閉，否則可能導致fd被耗盡。*/
int epoll_create(int size);
/*epoll的事件注冊函數*/
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
/*等待事件的到來，如果檢測到事件，就將所有就緒的事件從內核事件表中復制到它的第二個參數events指向的數組*/
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
epoll的事件注冊函數epoll_ctl，第一個參數是 epoll_create() 的返回值，第二個參數表示動作，使用如下三個宏來表示：
POLL_CTL_ADD //注冊新的fd到epfd中；
EPOLL_CTL_MOD //修改已經注冊的fd的監聽事件；
EPOLL_CTL_DEL //從epfd中刪除一個fd；
struct epoll_event 結構如下：
typedef union epoll_data
{
void *ptr;
int fd;
__uint32_t u32;
__uint64_t u64;
} epoll_data_t;
struct epoll_event
{
__uint32_t events; /* Epoll events */
epoll_data_t data; /* User data variable */
};
epoll_event結構體中的events 可以是以下幾個宏的集合：
EPOLLIN //表示對應的文件描述符可以讀（包括對端SOCKET正常關閉）；
EPOLLOUT //表示對應的文件描述符可以寫；
EPOLLPRI //表示對應的文件描述符有緊急的數據可讀（這里應該表示有帶外數據到來）；
EPOLLERR //表示對應的文件描述符發生錯誤；
EPOLLHUP //表示對應的文件描述符被掛斷；
EPOLLET //將EPOLL設為邊緣觸發(Edge Triggered)模式，這是相對於水平觸發(Level Triggered)來說的。
EPOLLONESHOT//只監聽一次事件，當監聽完這次事件之后，如果還需要繼續監聽這個socket的話，需要再次把這個socket加入到EPOLL隊列里。
epoll的一個簡單使用范例
#include <sys/socket.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#define MAXLINE 5
#define OPEN_MAX 100
#define LISTENQ 20
#define SERV_PORT 5000
#define INFTIM 1000
void setnonblocking(int sock)
{
int opts;
opts=fcntl(sock,F_GETFL);
if(opts<0)
{
perror("fcntl(sock,GETFL)");
exit(1);
}
opts = opts|O_NONBLOCK;
if(fcntl(sock,F_SETFL,opts)<0)
{
perror("fcntl(sock,SETFL,opts)");
exit(1);
}
}
int main(int argc, char* argv[])
{
int i, maxi, listenfd, connfd, sockfd,epfd,nfds, portnumber;
ssize_t n;
char line[MAXLINE];
socklen_t clilen;
if ( 2 == argc )
{
if( (portnumber = atoi(argv[1])) < 0 )
{
fprintf(stderr,"Usage:%s portnumber/a/n",argv[0]);
return 1;
}
}
else
{
fprintf(stderr,"Usage:%s portnumber/a/n",argv[0]);
return 1;
}
//聲明epoll_event結構體的變量,ev用於注冊事件,數組用於回傳要處理的事件
struct epoll_event ev,events[20];
//生成用於處理accept的epoll專用的文件描述符
epfd=epoll_create(256);
struct sockaddr_in clientaddr;
struct sockaddr_in serveraddr;
listenfd = socket(AF_INET, SOCK_STREAM, 0);
//把socket設置為非阻塞方式
//setnonblocking(listenfd);
//設置與要處理的事件相關的文件描述符
ev.data.fd=listenfd;
//設置要處理的事件類型
ev.events=EPOLLIN|EPOLLET;
//ev.events=EPOLLIN;
//注冊epoll事件
epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev);
bzero(&serveraddr, sizeof(serveraddr));
serveraddr.sin_family = AF_INET;
char *local_addr="127.0.0.1";
inet_aton(local_addr,&(serveraddr.sin_addr));//htons(portnumber);
serveraddr.sin_port=htons(portnumber);
bind(listenfd,(struct sockaddr *)&serveraddr, sizeof(serveraddr));
listen(listenfd, LISTENQ);
maxi = 0;
for ( ; ; ) {
//等待epoll事件的發生
nfds=epoll_wait(epfd,events,20,500);
//處理所發生的所有事件
for(i=0;i<nfds;++i)
{
if(events[i].data.fd==listenfd)//如果新監測到一個SOCKET用戶連接到了綁定的SOCKET端口，建立新的連接。
{
connfd = accept(listenfd,(struct sockaddr *)&clientaddr, &clilen);
if(connfd<0){
perror("connfd<0");
exit(1);
}
//setnonblocking(connfd);
char *str = inet_ntoa(clientaddr.sin_addr);
printf("accapt a connection from\n ");
//設置用於讀操作的文件描述符
ev.data.fd=connfd;
//設置用於注測的讀操作事件
ev.events=EPOLLIN|EPOLLET;
//ev.events=EPOLLIN;
//注冊ev
epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);
}
else if(events[i].events&EPOLLIN)//如果是已經連接的用戶，並且收到數據，那么進行讀入。
{
printf("EPOLLIN\n");
if ( (sockfd = events[i].data.fd) < 0)
continue;
if ( (n = read(sockfd, line, MAXLINE)) < 0) {
if (errno == ECONNRESET) {
close(sockfd);
events[i].data.fd = -1;
} else
printf("readline error\n");
} else if (n == 0) {
close(sockfd);
events[i].data.fd = -1;
}
if(n<MAXLINE-2)
line[n] = '\0';
//設置用於寫操作的文件描述符
ev.data.fd=sockfd;
//設置用於注測的寫操作事件
ev.events=EPOLLOUT|EPOLLET;
//修改sockfd上要處理的事件為EPOLLOUT
//epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);
}
else if(events[i].events&EPOLLOUT) // 如果有數據發送
{
sockfd = events[i].data.fd;
write(sockfd, line, n);
//設置用於讀操作的文件描述符
ev.data.fd=sockfd;
//設置用於注測的讀操作事件
ev.events=EPOLLIN|EPOLLET;
//修改sockfd上要處理的事件為EPOLIN
epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);
}
}
}
return 0;
}
帶ET和LT雙模式的epoll服務器
#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <pthread.h>
#include <errno.h>
#include <stdbool.h>
#define MAX_EVENT_NUMBER 1024 //event的最大數量
#define BUFFER_SIZE 10 //緩沖區大小
#define ENABLE_ET 1 //是否啟用ET模式
/* 將文件描述符設置為非擁塞的 */
int SetNonblocking(int fd)
{
int old_option = fcntl(fd, F_GETFL);
int new_option = old_option | O_NONBLOCK;
fcntl(fd, F_SETFL, new_option);
return old_option;
}
/* 將文件描述符fd上的EPOLLIN注冊到epoll_fd指示的epoll內核事件表中，參數enable_et指定是否對fd啟用et模式 */
void AddFd(int epoll_fd, int fd, bool enable_et)
{
struct epoll_event event;
event.data.fd = fd;
event.events = EPOLLIN; //注冊該fd是可讀的
if(enable_et)
{
event.events |= EPOLLET;
}
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &event); //向epoll內核事件表注冊該fd
SetNonblocking(fd);
}
/* LT工作模式特點：穩健但效率低 */
void lt_process(struct epoll_event* events, int number, int epoll_fd, int listen_fd)
{
char buf[BUFFER_SIZE];
int i;
for(i = 0; i < number; i++) //number: 就緒的事件數目
{
int sockfd = events[i].data.fd;
if(sockfd == listen_fd) //如果是listen的文件描述符，表明有新的客戶連接到來
{
struct sockaddr_in client_address;
socklen_t client_addrlength = sizeof(client_address);
int connfd = accept(listen_fd, (struct sockaddr*)&client_address, &client_addrlength);
AddFd(epoll_fd, connfd, false); //將新的客戶連接fd注冊到epoll事件表,使用lt模式
}
else if(events[i].events & EPOLLIN) //有客戶端數據可讀
{
// 只要緩沖區的數據還沒讀完，這段代碼就會被觸發。這就是LT模式的特點：反復通知，直至處理完成
printf("lt mode: event trigger once!\n");
memset(buf, 0, BUFFER_SIZE);
int ret = recv(sockfd, buf, BUFFER_SIZE - 1, 0);
if(ret <= 0) //讀完數據了，記得關閉fd
{
close(sockfd);
continue;
}
printf("get %d bytes of content: %s\n", ret, buf);
}
else
{
printf("something unexpected happened!\n");
}
}
}
/* ET工作模式特點：高效但潛在危險 */
void et_process(struct epoll_event* events, int number, int epoll_fd, int listen_fd)
{
char buf[BUFFER_SIZE];
int i;
for(i = 0; i < number; i++)
{
int sockfd = events[i].data.fd;
if(sockfd == listen_fd)
{
struct sockaddr_in client_address;
socklen_t client_addrlength = sizeof(client_address);
int connfd = accept(listen_fd, (struct sockaddr*)&client_address, &client_addrlength);
AddFd(epoll_fd, connfd, true); //使用et模式
}
else if(events[i].events & EPOLLIN)
{
/* 這段代碼不會被重復觸發，所以我么循環讀取數據，以確保把socket讀緩存的所有數據讀出。這就是我們消除ET模式潛在危險的手段 */
printf("et mode: event trigger once!\n");
while(1)
{
memset(buf, 0, BUFFER_SIZE);
int ret = recv(sockfd, buf, BUFFER_SIZE - 1, 0);
if(ret < 0)
{
/* 對於非擁塞的IO，下面的條件成立表示數據已經全部讀取完畢，此后epoll就能再次觸發sockfd上的EPOLLIN事件，以驅動下一次讀操作 */
if(errno == EAGAIN || errno == EWOULDBLOCK)
{
printf("read later!\n");
break;
}
close(sockfd);
break;
}
else if(ret == 0)
{
close(sockfd);
}
else //沒讀完，繼續循環讀取
{
printf("get %d bytes of content: %s\n", ret, buf);
}
}
}
else
{
printf("something unexpected happened!\n");
}
}
}
int main(int argc, char* argv[])
{
if(argc <= 2)
{
printf("usage: ip_address + port_number\n");
return -1;
}
const char* ip = argv[1];
int port = atoi(argv[2]);
int ret = -1;
struct sockaddr_in address;
bzero(&address, sizeof(address));
address.sin_family = AF_INET;
inet_pton(AF_INET, ip, &address.sin_addr);
address.sin_port = htons(port);
int listen_fd = socket(PF_INET, SOCK_STREAM, 0);
if(listen_fd < 0)
{
printf("fail to create socket!\n");
return -1;
}
ret = bind(listen_fd, (struct sockaddr*)&address, sizeof(address));
if(ret == -1)
{
printf("fail to bind socket!\n");
return -1;
}
ret = listen(listen_fd, 5);
if(ret == -1)
{
printf("fail to listen socket!\n");
return -1;
}
struct epoll_event events[MAX_EVENT_NUMBER];
int epoll_fd = epoll_create(5); //事件表大小為5
if(epoll_fd == -1)
{
printf("fail to create epoll!\n");
return -1;
}
AddFd(epoll_fd, listen_fd, true); //使用ET模式epoll,將listen文件描述符加入事件表
while(1)
{
int ret = epoll_wait(epoll_fd, events, MAX_EVENT_NUMBER, -1);
if(ret < 0)
{
printf("epoll failure!\n");
break;
}
if(ENABLE_ET)
{
et_process(events, ret, epoll_fd, listen_fd);
}
else
{
lt_process(events, ret, epoll_fd, listen_fd);
}
}
close(listen_fd);
return 0;
}
然后再寫一個簡單的TCP客戶端來測試一下：
//客戶端
#include <sys/types.h>
#include <sys/socket.h>
#include <stdio.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/time.h>
int main()
{
int client_sockfd;
int len;
struct sockaddr_in address;//服務器端網絡地址結構體
int result;
char str1[] = "ABCDE";
char str2[] = "ABCDEFGHIJK";
client_sockfd = socket(AF_INET, SOCK_STREAM, 0);//建立客戶端socket
address.sin_family = AF_INET;
address.sin_addr.s_addr = inet_addr("127.0.0.1");
address.sin_port = htons(8888);
len = sizeof(address);
result = connect(client_sockfd, (struct sockaddr *)&address, len);
if(result == -1)
{
perror("oops: client2");
exit(1);
}
//第一次讀寫
write(client_sockfd, str1, sizeof(str1));
sleep(5);
//第二次讀寫
write(client_sockfd, str2, sizeof(str2));
close(client_sockfd);
return 0;
}
TCP客戶端的動作是這樣的：第一次先發送字符串"ABCDE"過去服務器端，5秒后，再發字符串"ABCDEFGHIJK"過去服務端，我們觀察一下ET模式的服務器和LT模式的服務器在讀取數據的方式上到底有什么區別。
ET模式
ET模式現象分析：我們的服務器讀緩沖區大小我們設置了10。第一次接受字符串時，我們的緩沖區有足夠的空間接受它，所以打印出內容"ABCDE"並且打印出"read later"表示數據已經讀完了。第二次接收字符串時，我們的緩沖區空間不足以接收所有的字符，所以分了兩次接收。但是總觸發次數僅為2次。
LT模式
LT模式現象分析：
同理，第一次接受字符串有足夠的空間接受，第二次接收字符串緩沖區空間不足，所以第二次接收時分了兩次來接受。同時也注意到，只要你沒有完全接收完上次的數據，內核就會繼續通知你去接收數據！所以事件觸發的次數是3次。
EPOLLONESHOT事件
即使我們使用ET模式，一個socket上的某個事件還是可能被觸發多次，這在並發程序中就會引發一些問題。比如一個縣城在讀取完某個socket上的數據后開始處理這些數據，而在數據的出來過程中該socket上又有新數據可讀（EPOLLIN再次被觸發），此時另一個縣城被喚醒來讀取這些新數據。於是就出現了兩個線程同時操作一個socket的局面。這當然不是我們所期望的，我們期望的是一個socket連接在任一時刻都只被一個線程處理。這一點可以使用EPOLLONESHOT事件實現。
對於注冊了EPOLLONSHOT事件的文件描述符，操作系統最多觸發其上注冊的一個可讀、可寫或者異常事件，且只觸發一次，除非我們使用epoll_ctl函數重置該文件描述符上注冊的EPOLLONESHOT事件。這樣，當一個線程在處理某個socket時，其他線程是不可能有機會操作該socket的。但反過來思考，注冊了EPOLLONESHOT事件的socket一旦被某個線程處理完畢，該線程就應該立即重置這個socket上的EPOLLONESHOT事件，以確保這個socket下一次可讀時，其EPOLLIN事件能被觸發，進而讓其他工作線程有機會繼續處理這個socket。
下面是一個使用了EPOLLONESHOT的epoll服務器
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <pthread.h>
#include <stdbool.h>
#define MAX_EVENT_NUMBER 1024
#define BUFFER_SIZE 10
struct fds
{
int epollfd;
int sockfd;
};
int SetNonblocking(int fd)
{
int old_option = fcntl(fd, F_GETFL);
int new_option = old_option | O_NONBLOCK;
fcntl(fd, F_SETFL, new_option);
return old_option;
}
void AddFd(int epollfd, int fd, bool oneshot)
{
struct epoll_event event;
event.data.fd = fd;
event.events = EPOLLIN | EPOLLET;
if(oneshot)
{
event.events |= EPOLLONESHOT;
}
epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);
SetNonblocking(fd);
}
/*重置fd上的事件，這操作以后，盡管fd上的EPOLLONESHOT事件被注冊，但是操作系統仍然會觸發fd上的EPOLLIN事件，且只觸發一次*/
void reset_oneshot(int epollfd, int fd)
{
struct epoll_event event;
event.data.fd = fd;
event.events = EPOLLIN | EPOLLET | EPOLLONESHOT;
epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &event);
}
/*工作線程*/
void* worker(void* arg)
{
int sockfd = ((struct fds*)arg)->sockfd;
int epollfd = ((struct fds*)arg)->epollfd;
printf("start new thread to receive data on fd: %d\n", sockfd);
char buf[BUFFER_SIZE];
memset(buf, 0, BUFFER_SIZE);
while(1)
{
int ret = recv(sockfd, buf,BUFFER_SIZE-1, 0);
if(ret == 0)
{
close(sockfd);
printf("foreigner closed the connection\n");
break;
}
else if(ret < 0)
{
if(errno = EAGAIN)
{
reset_oneshot(epollfd, sockfd);
printf("read later\n");
break;
}
}
else
{
printf("get content: %s\n", buf);
//休眠5秒，模擬數據處理過程
printf("worker working...\n");
sleep(5);
}
}
printf("end thread receiving data on fd: %d\n", sockfd);
}
int main(int argc, char* argv[])
{
if(argc <= 2)
{
printf("usage: ip_address + port_number\n");
return -1;
}
const char* ip = argv[1];
int port = atoi(argv[2]);
int ret = -1;
struct sockaddr_in address;
bzero(&address, sizeof(address));
address.sin_family = AF_INET;
inet_pton(AF_INET, ip, &address.sin_addr);
address.sin_port = htons(port);
int listenfd = socket(PF_INET, SOCK_STREAM, 0);
if(listenfd < 0)
{
printf("fail to create socket!\n");
return -1;
}
ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address));
if(ret == -1)
{
printf("fail to bind socket!\n");
return -1;
}
ret = listen(listenfd, 5);
if(ret == -1)
{
printf("fail to listen socket\n");
return -1;
}
struct epoll_event events[MAX_EVENT_NUMBER];
int epollfd = epoll_create(5);
if(epollfd == -1)
{
printf("fail to create epoll\n");
return -1;
}
//注意，監聽socket listenfd上是不能注冊EPOLLONESHOT事件的，否則應用程序只能處理一個客戶連接！因為后續的客戶連接請求將不再觸發listenfd的EPOLLIN事件
AddFd(epollfd, listenfd, false);
while(1)
{
int ret = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1); //永久等待
if(ret < 0)
{
printf("epoll failure!\n");
break;
}
int i;
for(i = 0; i < ret; i++)
{
int sockfd = events[i].data.fd;
if(sockfd == listenfd)
{
struct sockaddr_in client_address;
socklen_t client_addrlength = sizeof(client_address);
int connfd = accept(listenfd, (struct sockaddr*)&client_address, &client_addrlength);
//對每個非監聽文件描述符都注冊EPOLLONESHOT事件
AddFd(epollfd, connfd, true);
}
else if(events[i].events & EPOLLIN)
{
pthread_t thread;
struct fds fds_for_new_worker;
fds_for_new_worker.epollfd = epollfd;
fds_for_new_worker.sockfd = events[i].data.fd;
/*新啟動一個工作線程為sockfd服務*/
pthread_create(&thread, NULL, worker, &fds_for_new_worker);
}
else
{
printf("something unexpected happened!\n");
}
}
}
close(listenfd);
return 0;
}
EPOLLONESHOT模式現象分析：我們繼續使用上面的TCP客戶端來測試，需要修改一下客戶端的sleep時間改為3秒。工作流程就是：客戶端第一次發送數據時服務器的接收緩沖區是有足夠空間的，然后服務器的工作線程進入5秒的處理數據階段；3秒后客戶端繼續發送新數據過來，但是工作線程還在處理數據，沒辦法立即接收新的數據。2秒后，客戶端該線程數據處理完了，開始接收新的數據。可以觀察到，我們客戶端只使用了同一個線程去處理同一個客戶端的請求，符合預期。