Linux网络编程 使用epoll实现一个高性能TCP Echo服务器

https://my.oschina.net/u/989096/blog/1189269

#零、前言 在Linux网络编程中，使用I/O复用来处理大规模并发连接是一种常见的方法，常用的有

select

、

poll

、

epoll

，本文主要讲述

epoll

的使用方法，并使用

epoll

实现一个高性能的Echo服务器，即把收到的数据原样返回给客户端。最后分析与

select

和

poll

的区别，并介绍一些常见问题。

#一、epoll介绍 ##1、相关的函数和数据结构

#include <sys/epoll.h>
int epoll_create(int size);

epoll_create()函数来创建一个epoll实例。参数_size_用来指定epoll要管理的socket的个数。在Linux 2.6.8以后的版本中，参数 size 已被忽略，但是必须大于0。

返回值是一个代表新创建的epoll实例的文件描述符。这个描述符将用于后续的一系列epoll的操作。当这个描述符不再需要时，应该使用

close()

函数将它关闭，系统会销毁实例并释放相关资源。其他细节请参考Linux man page。

#include <sys/epoll.h>

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 */
};

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

epoll_ctl()函数用于注册socket对应的事件。第一个参数_epfd_是epoll实例对应的fd，第二个参数_op_表操作的类型：

EPOLL_CTL_ADD：注册新的fd到epfd中
EPOLL_CTL_MOD：修改已经注册的fd的监听事件
EPOLL_CTL_DEL：从epfd中删除一个fd

第三参数_fd_表示要监听的socket，第四个参数_event_表示监听事件的类型。

struct epoll_event

中的events可以是以下宏的集合：

EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；
EPOLLOUT：表示对应的文件描述符可以写；
EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；
EPOLLERR：表示对应的文件描述符发生错误；
EPOLLHUP：表示对应的文件描述符被挂断；
EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式（下一节会介绍），这是相对于水平触发(Level Triggered)来说的。
EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里

如果epoll_ctl的第二个参数是EPOLL_CTL_DEL，那么第三个参数的内容被忽略。在Kernel 2.6.9以前的版本中，第三个参数不能设置为NULL，虽然它被忽略了。2.6.9以后的版本中，第三个参数可以设置为NULL。但是考虑到可移植性，还是应该传入一个不是NULL的指针。

int epoll_wait(int epfd, struct epoll_event *events,
int maxevents, int timeout);

epoll_wait的功能与select()类似，它等待_epfd_所代表的epoll实例中监听的事件发生。_events_指针返回已经准备好的事件，最多有_maxevents_个，参数_maxevent_必须大于零。_timeout_参数指定epoll_wait函数阻塞的毫秒数的最小值（精度和系统时钟有关，内核调度也会对此造成一些影响），设置_timeout_为-1则epoll_wait()会一直阻塞，设置为0则会立即返回。

如果函数调用成功，epoll_wait()函数返回已经准备好进行所要求的I/O操作的文件描述符的数量，如果在_timeout_时间内没有描述符准备好则返回0。出错时，epoll_wait()返回-1并且把errno设置为对应的值。

##2、水平触发（Level Triggered）和边沿触发（Edge Triggered） 水平触发和边沿触发类似于电路中的概念，先说一说水平触发。在水平触发模式下，内核检测的是是否有数据。如下图所示：



最初，没有任何数据，也就是处于红圈1的位置，过了一小段时间，收到数据后，处于红圈2的位置。此时内核检测到了这个状态，epoll_wait()则会返回这一可读事件。如果在读取数据的时候没有一次性全部读完，比如收到了2KB的数据而缓冲区只有1KB，那么在下一次调用epoll_wai()t时，还会继续返回这个事件，因为它还是处于有数据的状态。

而在边沿触发模式下，内核检测的是数据从没有到有的变化，也就是“上升沿”，如下图所示：



最初，没有任何数据，过了一小段时间，收到数据后，从没有数据的状态变成了有数据的状态，也就是说出现了图中红圈1标出的上升沿，此时内核检测到了从没有数据到有数据的变化，epoll_wait()会返回这个事件。但是，如果epoll_wait()返回这个可读事件后，没有读取完全部数据，那么以后不论有没有收到数据，epoll_wait()都不会再返回这个可读事件了。因为如果数据没有读取完的话，会一直处于有数据的状态，就不会产生从没数据到有数据的变化，也就是不会有红圈2标出的上升沿产生，所以就没有可读事件了。

所以如果要使用边沿触发模式的话，对应的socket就一定要用非阻塞模式。当epoll_wait()返回可读事件后，一直使用read()函数度数据，直到read()返回EAGAIN这个错误时，表明内核收到的数据已经全部读完了。如果不使用非阻塞的话，数据读完了以后，read()函数就阻塞住了，没有办法再去处理其他的socket。

##3、使用边沿触发模式时的饥饿问题（Starvation） 如果某个socket源源不断地收到非常多的数据，那么在试图读取完所有数据的过程中，有可能会造成其他的socket得不到处理，从而造成饥饿（这个问题不只针对epoll）。解决的办法是为每个已经准备好的描述符维护一个队列，这样程序就可以知道哪些描述符已经准备好了但是还在轮询等待。当然而简单的方法是使用水平触发模式。

#二、用epoll实现Echo服务器 Echo服务器的逻辑非常简单，只需要把用户发来的数据原样返回就可以了。Echo服务器的主要流程可以用下面的伪代码来描述：

创建服务器的socket
创建epoll实例
使用epoll_ctl设置监听服务器socket是否有新的客户端建立连接（即是否可读）
while（true） {
epoll_wait();
for 每个事件 {
if（服务器socket可读）{
accept()
把新的客户端socket加入epoll监听事件
}
else {
read()
send()
}
}
}

下面是Echo服务器的源代码。在发送数据的时候，直接调用到了send()函数。更好的做法应该是使用epoll_wait()等待socket可写了以后再发送数据，否则可能造成阻塞。这样做需要为每个socket都维护一个数据结构，其中包含socket对应的读写buf和读写到的位置指针等信息。这里为了简便就没有实现。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>

#define ECHO_SERVER_PORT        60000
#define LISTEN_BACKLOG          16
#define MAX_EVENT_COUNT         32
#define BUF_SIZE                2048

int main() {
int ret, i;
int server_fd, client_fd, epoll_fd;
int ready_count;
struct sockaddr_in server_addr;
struct sockaddr_in client_addr;
socklen_t addr_len;
struct epoll_event event;
struct epoll_event* event_array;
char* buf;

event_array = (struct epoll_event*)
malloc(sizeof(struct epoll_event)*MAX_EVENT_COUNT);
buf = (char*)malloc(sizeof(char)*BUF_SIZE);

memset(&server_addr, 0, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
server_addr.sin_port = htons(ECHO_SERVER_PORT);

server_fd = socket(AF_INET, SOCK_STREAM, 0);
if(server_fd == -1) {
perror("create socket failed.\n");
return 1;
}
ret = bind(server_fd, (struct sockaddr*)&server_addr, sizeof(server_addr));
if(ret == -1) {
perror("bind failed.\n");
return 1;
}
ret = listen(server_fd, LISTEN_BACKLOG);
if(ret == -1) {
perror("listen failed.\n");
return 1;
}

epoll_fd = epoll_create(1);
if(epoll_fd == -1) {
perror("epoll_create failed.\n");
return 1;
}
event.events = EPOLLIN;
event.data.fd = server_fd;
ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_fd, &event);
if(ret == -1) {
perror("epoll_ctl failed.\n");
return 1;
}

while(1) {
ready_count = epoll_wait(epoll_fd, event_array, MAX_EVENT_COUNT, -1);
if(ready_count == -1) {
perror("epoll_wait failed.\n");
return 1;
}
for(i = 0; i < ready_count; i++) {
if(event_array[i].data.fd == server_fd) {
client_fd = accept(server_fd,
(struct sockaddr*)&client_addr, &addr_len);
if(client_fd == -1) {
perror("accept failed.\n");
return 1;
}
event.events = EPOLLIN;
event.data.fd = client_fd;
ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client_fd, &event);
if(ret == -1) {
perror("epoll_ctl failed.\n");
return 1;
}
}
else {
ret = recv(event_array[i].data.fd, buf, BUF_SIZE, 0);
if(ret <= 0) {
close(event_array[i].data.fd);
epoll_ctl(epoll_fd, EPOLL_CTL_DEL,
event_array[i].data.fd, &event);
continue;
}
ret = send(event_array[i].data.fd, buf, (size_t)ret, 0);
if(ret == -1) {
perror("send failed.\n");
}
}
} // for each event
} // while(1)

close(epoll_fd);
close(server_fd);
free(event_array);
free(buf);

return 0;
}

#三、与

select

和

poll

的区别 select()的用法和epoll类似，只不过它使用了3个fd_set来表明哪些socket要监听可读事件，哪些监听可写事件和异常事件。select()主要的缺点是有：

每次调用select()时，都需要把fd_set从用户态拷贝到内核态，开销比较大。
当select()返回后，需要依次检查fd_set中的每个描述符，看看是否发生了期望的事件，当描述符比较多时，开销也很大。
select()默认只能支持最多1024个描述符。这个限制是一个宏，虽然可以通过重新编译内核来修改，但是还是很不方便并且不太实用。
每次调用select()前，都需要重新对每个fd_set进行赋值。
poll()与select()相比，除了接口上的区别以外，它改进了select()的第3个和第缺点。poll()没有1024个描述符的限制，理论上可以支持很多描述符，只要不超过系统限制的每个进程最多的文件描述符并且内存够用就行。而且poll使用的pollfd结构体中，把期待的事件和返回的事件分开了，这样使得每次调用poll()之前不用对pollfd重新复制。但是它依然避免不了用户态到内核态的拷贝，以及poll()返回后检查每个文件描述符的状态。

epoll克服了select()的所有的缺点，并且还增加了边沿触发模式，使得性能大大提高。

#四、FAQ

问：用什么来区分注册在一个epoll集合里面的文件描述符？ 答：根据文件描述符（file descriptors）和打开文件描述符（open file descriptors）（在内核内部代表一个打开的文件）。
问：如果在一个epoll实例中两次注册同一个文件描述符会发生什么？ 答：你会得到EEXIST。然而，你可以是用通过dup()，dup2()，fcntl() F_DUPFD这些方法复制一个文件描述符，添加到相同的epoll实例中。把复制的文件描述符使用不同的events掩码注册到同一个epoll事例中，这是一种非常实用的区分事件的方法。
两个epoll实例可以等待同一个文件描述符吗？如果可以，两个epoll实例都会报告这一事件吗？答：是的，两个实例都会报告这个事件。然而，细心的程序员需要正确地处理这种情况。
epoll文件描述符本身可以用来select/poll/epoll吗？答：是的，如果epoll有监听的事件发生了，那么它是可读的。
如果把epoll文件描述符添加到它自己监听的描述符集中，会发生什么？答：epoll_ctl()函数会失败（EINVAL)，然而，你可以把epoll文件描述符添加到其他的epoll文件描述符集中。
我可以把epoll文件描述符通过UNIX域套接字发送到其他进程中吗？答：可以，但是这样做并没有什么意义。因为对方进程没有这个epoll实例的文件描述符集的副本。
关闭一个文件描述符会导致它自动从所有的epoll集合中移除吗？答：是的。*
如果在两次epoll_wait()调用之间，有多个事件发生，那么它们会集中一起通知还是分开通知？答：一起通知。
对一个文件描述符的操作会影响到已经收集但还没有报告的事件吗？答：你可以对现有的文件描述符做两种操作，在这种情况下移除没有什么意义，修改则会重新读取可用的I/O。*
在使用EPOLLET标志（边沿触发）时，我需要对一个文件描述符持续地read/write直到出现EAGAIN吗？答：从epoll_wait()收到一个事件则表明这个文件描述符已经准备好做对应的I/O操作了。直到下一次read/write出现EAGAIN之前，你必须认为它是已经准备好了的。至于什么时候和怎样使用这个文件描述符就完全看你自己了。对于基于数据包类型的文件描述符，比如UDP socket和canonical模式下的terminal，检测read/write的I/O空间是否用尽的唯一方法就是持续地read/write直到出现EAGAIN。对于基于数据流的文件描述符，比如pipe,
FIFO, TCP socket，还可以用检测向目标文件描述符发送/接收数据的总量的方法可以检测read/write的I/O空间是否用尽。*
标有*的问答翻译的不准确，请参考原文。