转载：http://blog.csdn.net/u010657219/article/details/44061629
在linux 没有实现epoll事件驱动机制之前，我们一般选择用select或者poll等IO多路复用的方法来实现并发服务程序。在大数据、高并发、集群等一些名词唱得火热之年代，select和poll的用武之地越来越有限，风头已经被epoll占尽。
本文便来介绍epoll的实现机制，并附带讲解一下select和poll。通过对比其不同的实现机制，真正理解为何epoll能实现高并发。
select()和poll() IO多路复用模型
select的缺点：
- 单个进程能够监视的文件描述符的数量存在最大限制，通常是1024，当然可以更改数量，但由于select采用轮询的方式扫描文件描述符，文件描述符数量越多，性能越差；(在linux内核头文件中，有这样的定义：#define __FD_SETSIZE 1024)
- 内核 / 用户空间内存拷贝问题，select需要复制大量的句柄数据结构，产生巨大的开销；
- select返回的是含有整个句柄的数组，应用程序需要遍历整个数组才能发现哪些句柄发生了事件；
- select的触发方式是水平触发，应用程序如果没有完成对一个已经就绪的文件描述符进行IO操作，那么之后每次select调用还是会将这些文件描述符通知进程。
相比select模型，poll使用链表保存文件描述符，因此没有了监视文件数量的限制，但其他三个缺点依然存在。
拿select模型为例，假设我们的服务器需要支持100万的并发连接，则在__FD_SETSIZE 为1024的情况下，则我们至少需要开辟1k个进程才能实现100万的并发连接。除了进程间上下文切换的时间消耗外，从内核/用户空间大量的无脑内存拷贝、数组轮询等，是系统难以承受的。因此，基于select模型的服务器程序，要达到10万级别的并发访问，是一个很难完成的任务。
因此，该epoll上场了。
epoll IO多路复用模型实现机制
由于epoll的实现机制与select/poll机制完全不同，上面所说的 select的缺点在epoll上不复存在。
设想一下如下场景：有100万个客户端同时与一个服务器进程保持着TCP连接。而每一时刻，通常只有几百上千个TCP连接是活跃的(事实上大部分场景都是这种情况)。如何实现这样的高并发？
在select/poll时代，服务器进程每次都把这100万个连接告诉操作系统(从用户态复制句柄数据结构到内核态)，让操作系统内核去查询这些套接字上是否有事件发生，轮询完后，再将句柄数据复制到用户态，让服务器应用程序轮询处理已发生的网络事件，这一过程资源消耗较大，因此，select/poll一般只能处理几千的并发连接。
epoll的设计和实现与select完全不同。epoll通过在Linux内核中申请一个简易的文件系统(文件系统一般用什么数据结构实现？B+树)。把原先的select/poll调用分成了3个部分：
1）调用epoll_create()建立一个epoll对象(在epoll文件系统中为这个句柄对象分配资源)
2）调用epoll_ctl向epoll对象中添加这100万个连接的套接字
3）调用epoll_wait收集发生的事件的连接
如此一来，要实现上面说是的场景，只需要在进程启动时建立一个epoll对象，然后在需要的时候向这个epoll对象中添加或者删除连接。同时，epoll_wait的效率也非常高，因为调用epoll_wait时，并没有一股脑的向操作系统复制这100万个连接的句柄数据，内核也不需要去遍历全部的连接。
下面来看看Linux内核具体的epoll机制实现思路。
当某一进程调用epoll_create方法时，Linux内核会创建一个eventpoll结构体，这个结构体中有两个成员与epoll的使用方式密切相关。eventpoll结构体如下所示：
每一个epoll对象都有一个独立的eventpoll结构体，用于存放通过epoll_ctl方法向epoll对象中添加进来的事件。这些事件都会挂载在红黑树中，如此，重复添加的事件就可以通过红黑树而高效的识别出来(红黑树的插入时间效率是lgn，其中n为树的高度)。
而所有添加到epoll中的事件都会与设备(网卡)驱动程序建立回调关系，也就是说，当相应的事件发生时会调用这个回调方法。这个回调方法在内核中叫ep_poll_callback,它会将发生的事件添加到rdlist双链表中。
在epoll中，对于每一个事件，都会建立一个epitem结构体，如下所示：
当调用epoll_wait检查是否有事件发生时，只需要检查eventpoll对象中的rdlist双链表中是否有epitem元素即可。如果rdlist不为空，则把发生的事件复制到用户态，同时将事件数量返回给用户。
epoll数据结构示意图
从上面的讲解可知：通过红黑树和双链表数据结构，并结合回调机制，造就了epoll的高效。
OK，讲解完了Epoll的机理，我们便能很容易掌握epoll的用法了。一句话描述就是：三步曲。
第一步：epoll_create()系统调用。此调用返回一个句柄，之后所有的使用都依靠这个句柄来标识。
第二步：epoll_ctl()系统调用。通过此调用向epoll对象中添加、删除、修改感兴趣的事件，返回0标识成功，返回-1表示失败。
第三部：epoll_wait()系统调用。通过此调用收集收集在epoll监控中已经发生的事件。
最后，附上一个epoll编程实例。(作者为sparkliang)
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
|
//
// a simple echo server using epoll
in linux
//
// 2009-11-05
// 2013-03-22:修改了几个问题，1是/n格式问题，2是去掉了原代码不小心加上的ET模式;
// 本来只是简单的示意程序，决定还是加上 recv/send时的buffer偏移
// by sparkling
//
#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 <iostream>
usingnamespacestd;
#define MAX_EVENTS 500
structmyevent_s
{
intfd;
void(*call_back)(intfd,intevents,void*arg);
intevents;
void*arg;
intstatus;//
1: in epoll wait list, 0 not in
charbuff[128];//
recv data buffer
intlen,s_offset;
longlast_active;//
last active time
};
// set event
voidEventSet(myevent_s*ev,intfd,void(*call_back)(int,int,void*),void*arg)
{
ev->fd=fd;
ev->call_back=call_back;
ev->events=0;
ev->arg=arg;
ev->status=0;
bzero(ev->buff,sizeof(ev->buff));
ev->s_offset=0;
ev->len=0;
ev->last_active=time(NULL);
}
// add/mod an event to epoll
voidEventAdd(intepollFd,intevents,myevent_s*ev)
{
structepoll_eventepv={0,{0}};
intop;
epv.data.ptr=ev;
epv.events=ev->events=events;
if(ev->status==1){
op=EPOLL_CTL_MOD;
}
else{
op=EPOLL_CTL_ADD;
ev->status=1;
}
if(epoll_ctl(epollFd,op,ev->fd,&epv)<0)
printf("Event
Add failed[fd=%d], evnets[%d]
",ev->fd,events);
else
printf("Event
Add OK[fd=%d], op=%d, evnets[%0X]
",ev->fd,op,events);
}
// delete an event from epoll
voidEventDel(intepollFd,myevent_s*ev)
{
structepoll_eventepv={0,{0}};
if(ev->status!=1)return;
epv.data.ptr=ev;
ev->status=0;
epoll_ctl(epollFd,EPOLL_CTL_DEL,ev->fd,&epv);
}
intg_epollFd;
myevent_sg_Events[MAX_EVENTS+1];//
g_Events[MAX_EVENTS] is used by listen fd
voidRecvData(intfd,intevents,void*arg);
voidSendData(intfd,intevents,void*arg);
// accept new connections from clients
voidAcceptConn(intfd,intevents,void*arg)
{
structsockaddr_insin;
socklen_tlen=sizeof(structsockaddr_in);
intnfd,i;
//
accept
if((nfd=accept(fd,(structsockaddr*)&sin,&len))==-1)
{
if(errno!=EAGAIN&&errno!=EINTR)
{
}
printf("%s:
accept, %d",__func__,errno);
return;
}
do
{
for(i=0;i<MAX_EVENTS;i++)
{
if(g_Events[i].status==0)
{
break;
}
}
if(i==MAX_EVENTS)
{
printf("%s:max
connection limit[%d].",__func__,MAX_EVENTS);
break;
}
//
set nonblocking
intiret=0;
if((iret=fcntl(nfd,F_SETFL,O_NONBLOCK))<0)
{
printf("%s:
fcntl nonblocking failed:%d",__func__,iret);
break;
}
//
add a read event for receive data
EventSet(&g_Events[i],nfd,RecvData,&g_Events[i]);
EventAdd(g_epollFd,EPOLLIN,&g_Events[i]);
}while(0);
printf("new
conn[%s:%d][time:%d], pos[%d]
",inet_ntoa(sin.sin_addr),
ntohs(sin.sin_port),g_Events[i].last_active,i);
}
// receive data
voidRecvData(intfd,intevents,void*arg)
{
structmyevent_s*ev=(structmyevent_s*)arg;
intlen;
//
receive data
len=recv(fd,ev->buff+ev->len,sizeof(ev->buff)-1-ev->len,0);
EventDel(g_epollFd,ev);
if(len>0)
{
ev->len+=len;
ev->buff[len]=' ';
printf("C[%d]:%s
",fd,ev->buff);
//
change to send event
EventSet(ev,fd,SendData,ev);
EventAdd(g_epollFd,EPOLLOUT,ev);
}
elseif(len==0)
{
close(ev->fd);
printf("[fd=%d]
pos[%d], closed gracefully.
",fd,ev-g_Events);
}
else
{
close(ev->fd);
printf("recv[fd=%d]
error[%d]:%s
",fd,errno,strerror(errno));
}
}
// send data
voidSendData(intfd,intevents,void*arg)
{
structmyevent_s*ev=(structmyevent_s*)arg;
intlen;
//
send data
len=send(fd,ev->buff+ev->s_offset,ev->len-ev->s_offset,0);
if(len>0)
{
printf("send[fd=%d],
[%d<->%d]%s
",fd,len,ev->len,ev->buff);
ev->s_offset+=len;
if(ev->s_offset==ev->len)
{
//
change to receive event
EventDel(g_epollFd,ev);
EventSet(ev,fd,RecvData,ev);
EventAdd(g_epollFd,EPOLLIN,ev);
}
}
else
{
close(ev->fd);
EventDel(g_epollFd,ev);
printf("send[fd=%d]
error[%d]
",fd,errno);
}
}
voidInitListenSocket(intepollFd,shortport)
{
intlistenFd=socket(AF_INET,SOCK_STREAM,0);
fcntl(listenFd,F_SETFL,O_NONBLOCK);//
set non-blocking
printf("server
listen fd=%d
",listenFd);
EventSet(&g_Events[MAX_EVENTS],listenFd,AcceptConn,&g_Events[MAX_EVENTS]);
//
add listen socket
EventAdd(epollFd,EPOLLIN,&g_Events[MAX_EVENTS]);
//
bind & listen
sockaddr_insin;
bzero(&sin,sizeof(sin));
sin.sin_family=AF_INET;
sin.sin_addr.s_addr=INADDR_ANY;
sin.sin_port=htons(port);
bind(listenFd,(constsockaddr*)&sin,sizeof(sin));
listen(listenFd,5);
}
intmain(intargc,char**argv)
{
unsignedshortport=12345;//
default port
if(argc==2){
port=atoi(argv[1]);
}
//
create epoll
g_epollFd=epoll_create(MAX_EVENTS);
if(g_epollFd<=0)printf("create
epoll failed.%d
",g_epollFd);
//
create & bind listen socket, and add to epoll, set non-blocking
InitListenSocket(g_epollFd,port);
//
event loop
structepoll_eventevents[MAX_EVENTS];
printf("server
running:port[%d]
",port);
intcheckPos=0;
while(1){
//
a simple timeout check here, every time 100, better to use a mini-heap, and add timer event
longnow=time(NULL);
for(inti=0;i<100;i++,checkPos++)//
doesn't check listen fd
{
if(checkPos==MAX_EVENTS)checkPos=0;//
recycle
if(g_Events[checkPos].status!=1)continue;
longduration=now-g_Events[checkPos].last_active;
if(duration>=60)//
60s timeout
{
close(g_Events[checkPos].fd);
printf("[fd=%d]
timeout[%d--%d].
",g_Events[checkPos].fd,g_Events[checkPos].last_active,now);
EventDel(g_epollFd,&g_Events[checkPos]);
}
}
//
wait for events to happen
intfds=epoll_wait(g_epollFd,events,MAX_EVENTS,1000);
if(fds<0){
printf("epoll_wait
error, exit
");
break;
}
for(inti=0;i<fds;i++){
myevent_s*ev=(structmyevent_s*)events[i].data.ptr;
if((events[i].events&EPOLLIN)&&(ev->events&EPOLLIN))//
read event
{
ev->call_back(ev->fd,events[i].events,ev->arg);
}
if((events[i].events&EPOLLOUT)&&(ev->events&EPOLLOUT))//
write event
{
ev->call_back(ev->fd,events[i].events,ev->arg);
}
}
}
//
free resource
return0;
}
|