Linux中的IO复用机制——Epoll

在Linux中IO复用机制主要目的是为了实现在单进程环境下，同时监控多个描述符（文件描述符，套接字描述符等）的目的。一旦监控的描述符就绪，就会通知相应进程并解除该进程的阻塞状态，使进程能够对就绪描述符进行处理。

一、什么是IO复用

UNIX有五大IO模型：

  • 1、异步IO（asynchronous IO）
  • 2、阻塞IO（blocking IO）
  • 3、非阻塞IO（nonblocking IO）
  • 4、IO多路复用（IO multiplexing）
  • 5、信号驱动的IO（signal drivened IO）

1、异步IO

其中前四种IO统称为同步IO。那么同步IO和异步IO的区别是什么呢？
在同步文件IO中，线程启动一个IO操作然后就立即进入等待状态，直到IO操作完成后才醒来继续执行。而异步文件IO方式中，线程发送一个IO请求到内核，然后继续处理其他的事情，内核完成IO请求后，将会通知线程IO操作完成了。
如果某进程有大量IO需要处理，则选择异步IO方式可以显著提高效率。因为选择异步IO之后，进程不必等待IO完成，即可继续处理其他的事情，直到IO事件结束，操作系统会通知该进程IO已经结束。

2、阻塞IO

阻塞IO
从上图可以直观的看出：进程在使用recvfrom()系统调用后，进程由用户态转为核心态，同时内核检测recvfrom()函数对应fd的数据集中没有新数据传入，并开始等待。等待过程中整个进程是阻塞的，不占用任何CPU资源。直到内核发现有新的数据写入对应数据集，此时内核就唤醒处于就绪队列的该进程。进程唤醒后仍处于核心态，将数据集中的数据拷贝并通过recvfrom()函数返回后，此进程才由核心态转化为用户态，并继续之后的工作。
由上述过程可知，如果使用阻塞IO，那么在等待IO就绪的过程中，整个进程是无法进行任何操作且不占用CPU资源的。直到IO结束，内核唤醒该进程，该进程才能继续执行。

3、非阻塞IO

非阻塞IO
从上图可知：在非阻塞IO的情况下，用户进程调用recvfrom()系统调用后，如果IO没有就绪，则不会等待IO就绪直接返回。因此如果想要等到IO结束，就需要不断的向内核询问IO是否完成。
上述过程可以看出，非阻塞IO的特点就是需要不断向服务器询问IO是否就绪。

4、IO多路复用

IO多路复用
上图用select机制作为例子，如果一个进程调用了select，那么整个进程都会被阻塞，直到select所监听的所有fd中，出现完成IO的情况，进程就会解除阻塞。
一个进程通过调用select函数监听多个fd的IO情况，就是一个典型的IO多路复用的例子。

二、Select与Poll机制

1、Select机制

我们先分析一下select函数

1
/**
2
 * 该函数是select函数的声明
3
 * @parameter: maxfdp1 指定待测试的文件描述字个数，它的值是待测试的最大描述字加1。
4
 * @parameter: *readset/*writeset/*exceptset 均为fd_set类型，可以将fd_set理解为一个集合，这个集合中存放的是文件描述符。中间的三个参数指定我们要让内核测试读、写和异常条件的文件描述符集合。如果对某一个的条件不感兴趣，就可以把它设为空指针。
5
 * @parameter: *timeout timeout告知内核等待所指定文件描述符集合中的任何一个就绪可花多少时间。其timeval结构用于指定这段时间的秒数和微秒数。
6
 * @return: int 若有就绪描述符返回其数目，若超时则为0，若出错则为-1
7
 */
8
int select(int maxfdp1,fd_set *readset,fd_set *writeset,fd_set *exceptset,const struct timeval *timeout);
9
10
//以下是select的具体使用方法：
11
FD_ZERO(&fds);              //每次循环都必须清空FD_Set
12
FD_SET(sock_fd, &fds);      //将sock_fd加入集合
13
14
//此select设定为对整个集合内的fd写监听，监听的最长时间为timeout
15
int n = select(maxfd, NULL, &fds, NULL, &timeout);
16
switch(n) {
17
    case -1:
18
        fprintf(stderr, "Select error:%s \n\a", strerror(errno));
19
        exit(1);
20
    case 0:
21
        printf("select time out, lost packet!\n");
22
        ......
23
        break;
24
    default:
25
        //判断sock_fd是否还在集合中
26
        if(FD_ISSET(sock_fd, &fds)) {
27
            //还在集合中则说明对该fd监听到了写操作
28
            ......
29
        } else{
30
            //没有对该fd监听到写操作
31
            ......
32
        }
33
}

select的FD_SET是通过一组宏函数进行实现的，支持的最大监听数不超过1024个。select函数底层是通过轮询机制实现的，因此对CPU占用很高。同时，每次调用select都需要把fd_set集合从用户态拷贝到内核态，因此该函数的效率很低。

2、Poll机制

poll的机制与select类似，与select在本质上没有多大差别，管理多个描述符也是进行轮询，根据描述符的状态进行处理。poll机制相对于select机制，解决了select的最大文件描述符支持为1024的问题（支持任意大小的描述符集），并没有解决性能开销问题。
下面是pll的函数原型：

1
//poll改变了文件描述符集合的描述方式，使用了pollfd结构而不是select的fd_set结构，
2
//使得poll支持的文件描述符集合限制远大于select的1024
3
typedef struct pollfd {
4
        int fd;                         // 需要被检测或选择的文件描述符
5
        short events;                   // 对文件描述符fd上感兴趣的事件
6
        short revents;                  // 文件描述符fd上当前实际发生的事件
7
} pollfd_t;
8
9
/**
10
 * 实现IO多路复用的poll函数
11
 * @parameter: fds fds是一个struct pollfd类型的数组，用于存放需要检测其状态的socket描述符，
12
 * 并且调用poll函数之后fds数组不会被清空；一个pollfd结构体表示一个被监视的文件描述符，
13
 * 通过传递fds指示 poll() 监视多个文件描述符。其中，结构体的events域是监视该文件描述符的事件掩码，
14
 * 由用户来设置这个域，结构体的revents域是文件描述符的操作结果事件掩码，内核在调用返回时设置这个域。
15
 * @parameter: nfds 记录数组fds中描述符的总数量
16
 * @parameter: timeout 最长阻塞时间
17
 */
18
int poll(struct pollfd *fds, nfds_t nfds, int timeout);

三、Epoll机制

epoll在Linux2.6内核正式提出，是基于事件驱动的IO方式，相对于select来说，epoll没有描述符个数限制，使用一个文件描述符管理多个描述符，将用户关心的文件描述符的事件存放到内核的一个事件表中，这样在用户空间和内核空间的copy只需一次。
Linux中提供的epoll相关函数如下：

1
/**
2
 * epoll_create 函数创建一个epoll句柄。
3
 * @parameter: size 参数size表明内核要监听的描述符数量。
4
 * @return: 调用成功时返回一个epoll句柄描述符，失败时返回-1。
5
 */
6
int epoll_create(int size);
7
8
9
//epoll_event 结构体定义如下：
10
struct epoll_event {
11
    __uint32_t events;  /* Epoll events */
12
    epoll_data_t data;  /* User data variable */
13
};
14
typedef union epoll_data {
15
    void *ptr;
16
    int fd;
17
    __uint32_t u32;
18
    __uint64_t u64;
19
} epoll_data_t;
20
21
/**
22
 * epoll_ctl 函数注册要监听的事件类型。
23
 * @parameter: epfd 表示epoll句柄。由上述 epoll_create() 函数申请得到。
24
 * @parameter: op 表示fd操作类型，有3种:EPOLL_CTL_ADD 注册新的fd到epfd中;EPOLL_CTL_MOD
25
 * 修改已注册的fd的监听事件;EPOLL_CTL_DEL 从epfd中删除一个fd。
26
 * @parameter: fd 是要监听的描述符。
27
 * @parameter: event 表示要监听的事件。
28
 * @return: 调用成功时返回一个epoll句柄描述符，失败时返回-1。
29
 */
30
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
31
32
/**
33
 * epoll_wait 函数等待事件的就绪。
34
 * @parameter: epfd 表示epoll句柄。由上述 epoll_create() 函数申请得到。
35
 * @parameter: events 表示从内核得到的就绪事件集合。
36
 * @parameter: maxevents 告诉内核events的大小。
37
 * @parameter: timeout 表示等待的超时事件。
38
 * @return: 成功时返回就绪的事件数目，调用失败时返回 -1，等待超时返回 0。
39
 */
40
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

epoll是Linux内核为处理大批量文件描述符而作了改进的poll，是Linux下多路复用IO接口select/poll的增强版本，它能显著提高程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率。原因就是获取事件的时候，它无须遍历整个被侦听的描述符集，只要遍历那些被内核IO事件异步唤醒而加入Ready队列的描述符集合就行了。
epoll除了提供select/poll那种IO事件的水平触发（Level Triggered）外，还提供了边缘触发（Edge Triggered），这就使得用户空间程序有可能缓存IO状态，减少epoll_wait/epoll_pwait的调用，提高应用程序效率。
水平触发（LT）：默认工作模式，即当epoll_wait检测到某描述符事件就绪并通知应用程序时，应用程序可以不立即处理该事件；下次调用epoll_wait时，会再次通知此事件
边缘触发（ET）： 当epoll_wait检测到某描述符事件就绪并通知应用程序时，应用程序必须立即处理该事件。如果不处理，下次调用epoll_wait时，不会再次通知此事件。（直到你做了某些操作导致该描述符变成未就绪状态了，也就是说边缘触发只在状态由未就绪变为就绪时只通知一次）。
LT和ET原本应该是用于脉冲信号的，可能用它来解释更加形象。Level和Edge指的就是触发点，Level为只要处于水平，那么就一直触发，而Edge则为上升沿和下降沿的时候触发。比如：0->1 就是Edge，1->1 就是Level。
ET模式很大程度上减少了epoll事件的触发次数，因此效率比LT模式下高。
select、poll、epoll三者机制的区别如下图所示：
select、poll、epoll的比较
上述对比图存在错误：epoll的底层实现应该是红黑树！
epoll是Linux目前大规模网络并发程序开发的首选模型。在绝大多数情况下性能远超select和poll。目前流行的高性能web服务器Nginx正式依赖于epoll提供的高效网络套接字轮询服务。但是，在并发连接不高的情况下，多线程+阻塞I/O方式可能性能更好。
以下是一个简单的基于epoll的C/S网络通信示例：
服务器代码（只看Server类的定义和main函数即可）：

1
//
2
// Created by mylord on 2019/10/26.
3
//
4
5
#ifndef FILETRANSER_SERVER_H
6
#define FILETRANSER_SERVER_H
7
8
#include <iostream>
9
#include <stdio.h>
10
#include <stdlib.h>
11
#include <errno.h>
12
#include <string>
13
#include <netdb.h>
14
#include <sys/types.h>
15
#include <netinet/in.h>
16
#include <sys/socket.h>
17
#include <unistd.h>
18
#include <arpa/inet.h>
19
#include <vector>
20
#include <string.h>
21
#include <algorithm>
22
#include <sys/epoll.h>
23
24
#define MAX_BUFF_SIZE 1024
25
26
using namespace std;
27
28
typedef struct {
29
    sockaddr_in client_sock;
30
    int client_fd;
31
    string client_ip;
32
}ClientInfo;
33
34
class Server {
35
private:
36
    vector<ClientInfo> client_info;
37
38
    int listen_fd, listen_port, listen_size;
39
    sockaddr_in server_addr;
40
public:
41
    Server(int port = 5555, int size = 10);
42
43
    void Init();
44
45
    void Listen();
46
47
    int AcceptConnection();
48
49
    int Write(int sock_fd, char buff[]);
50
51
    int Read(int sock_fd, char buff[]);
52
53
    int Close(int client_fd);
54
55
    int getListenFd() const;
56
};
57
58
59
#endif //FILETRANSER_SERVER_H
60
61
62
int main() {
63
    Server server(5555, 10);
64
    char buffer[1204];
65
    struct epoll_event ev, events[10];   //声明epoll_event结构体的变量,ev用于注册事件,数组用于回传要处理的事件
66
    int epoll_fd = epoll_create(10);    //创建一个epoll的句柄，并告诉内核这个监听的数目为10  
67
    int listen_fd = server.getListenFd();
68
    int nfds = 0;                       //记录需要处理的事件数
69
70
    ev.data.fd = listen_fd;
71
    ev.events = EPOLLIN | EPOLLET;      //linsten_fd可读，边缘触发
72
    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ev);
73
74
    server.Listen();
75
    cout << "server is listening......" << endl;
76
77
    while(true) {
78
        nfds = epoll_wait(epoll_fd, events, 10, -1);
79
80
        for(int i = 0; i < nfds; i++) {
81
            if(events[i].data.fd == listen_fd) {
82
                int client_fd = server.AcceptConnection();
83
84
                ev.data.fd = client_fd;
85
                ev.events = EPOLLIN | EPOLLET;
86
                epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client_fd, &ev);
87
            } else if(events[i].events & EPOLLIN) {  //epoll池中的时间被边缘触发且是因为收到数据待读入
88
                int nbytes = 0;
89
                if ((nbytes = server.Read(events[i].data.fd, buffer)) <= 0) {
90
                    cout << "fd-" << events[i].data.fd << " receive data error!" << endl;
91
                    server.Close(events[i].data.fd);
92
                    continue;
93
                }
94
                cout << "EPOLL receive from " << events[i].data.fd << ":" << buffer << endl;
95
                strcpy(buffer, "received from epoll server!");
96
                server.Write(events[i].data.fd, buffer);
97
            }
98
        }
99
    }
100
}
101
102
103
Server::Server(int port, int size): listen_port(port), listen_size(10) {
104
    this->Init();
105
}
106
107
void Server::Init() {
108
    if((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1)
109
    {   //服务器端开始建立socket描述符
110
        fprintf(stderr, "Socket error:%s \n\a", strerror(errno));
111
        exit(1);
112
    }
113
    //服务器端填充tcp sockaddr结构
114
    bzero(&server_addr, sizeof(struct sockaddr_in));    //先将套接字地址数据结构清零
115
    server_addr.sin_family = AF_INET;
116
    server_addr.sin_addr.s_addr = htons(INADDR_ANY);
117
    server_addr.sin_port = htons(listen_port);
118
119
    if(bind(listen_fd, (struct sockaddr *)(&server_addr), sizeof(struct sockaddr)) == -1)
120
    {
121
        fprintf(stderr, "Bind error:%s\n\a", strerror(errno));
122
        exit(1);
123
    }
124
}
125
126
void Server::Listen() {
127
    if(listen(listen_fd, listen_size) == -1)
128
    {   //端口绑定成功，监听socketfd描述符，同时处理的最大连接请求数为10
129
        fprintf(stderr, "Listen error:%s\n\a", strerror(errno));
130
        exit(1);
131
    }
132
}
133
134
int Server::AcceptConnection() {
135
    int sockaddr_size = sizeof(struct sockaddr_in);
136
    ClientInfo temp;
137
138
    if ((temp.client_fd = accept(listen_fd, (struct sockaddr *)(&temp.client_sock),
139
                                     (socklen_t *) &sockaddr_size)) == -1)
140
    {   //调用accept接受一个连接请求
141
        fprintf(stderr, "Accept error:%s\n\a", strerror(errno));
142
        exit(1);
143
    }
144
145
    temp.client_ip.assign(inet_ntoa(temp.client_sock.sin_addr));
146
147
    cout << "Connected from "<< temp.client_ip << "\tclient fd is " << temp.client_fd << endl;
148
149
    client_info.push_back(temp);
150
151
    return temp.client_fd;
152
}
153
154
int Server::Write(int sock_fd, char buff[]) {
155
    int nbytes = 0;
156
    if((nbytes = write(sock_fd, buff, strlen(buff))) == -1)
157
        fprintf(stderr, "Write Error:%s\n", strerror(errno));
158
    return nbytes;
159
}
160
161
int Server::Read(int sock_fd, char buff[]) {
162
    int nbytes = 0;
163
    if ((nbytes = read(sock_fd, buff, MAX_BUFF_SIZE)) == -1)
164
        fprintf(stderr, "Read Error:%s\n", strerror(errno));
165
166
    buff[nbytes] = '\0';
167
    return nbytes;
168
}
169
170
int Server::Close(int client_fd) {
171
    //先查找client_fd的迭代器client_account_it
172
    auto client_it = find_if(client_info.begin(), client_info.end(),
173
            [client_fd](const ClientInfo &cli){ return cli.client_fd == client_fd; });
174
175
    client_info.erase(client_it);
176
177
    close(client_fd);
178
179
    printf("fd-%d exit!\n", client_fd);
180
}
181
182
int Server::getListenFd() const{
183
    return listen_fd;
184
}

客户端代码（只看Client类的定义和main函数即可）：

1
//
2
// Created by mylord on 2019/10/26.
3
//
4
5
#include <stdio.h>
6
#include <stdlib.h>
7
#include <string.h>
8
#include <netdb.h>
9
#include <netinet/in.h>
10
#include <sys/socket.h>
11
#include <unistd.h>
12
#include <iostream>
13
#include <fstream>
14
15
#define MAX_BUFF_SIZE 1024
16
17
class Client
18
{
19
private:
20
    int sock_fd;
21
    struct sockaddr_in client_addr;
22
    char server_ip[16];
23
    int server_port;
24
25
public:
26
    Client(char * server_ip, int server_port);
27
28
    int Connect();
29
30
    int Write(char * buff);
31
32
    int Read(char * buff);
33
34
    int Close();
35
};
36
37
38
39
Client::Client(char * server_ip, int server_port)
40
{
41
    strncpy(this->server_ip, server_ip, 16);
42
    this->server_port = server_port;
43
44
    struct hostent * host;
45
    if((host = gethostbyname(this->server_ip)) == NULL)
46
    {
47
        fprintf(stderr, "The host name %s is illegal.\n", server_ip);
48
        exit(1);
49
    }
50
51
    if((this->sock_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1)
52
    {
53
        fprintf(stderr, "Init socket fd error!\n");
54
        exit(1);
55
    }
56
57
    bzero(&this->client_addr, sizeof(this->client_addr));
58
    this->client_addr.sin_family = AF_INET;
59
    this->client_addr.sin_port = htons(this->server_port);
60
    this->client_addr.sin_addr = *((struct in_addr *)host->h_addr);
61
62
}
63
64
65
int Client::Connect()
66
{
67
    int nbytes = connect(this->sock_fd, (struct sockaddr *)(&this->client_addr),
68
                         sizeof(struct sockaddr));
69
70
    return nbytes;
71
}
72
73
74
int Client::Write(char * buff)
75
{
76
    int nbytes = write(this->sock_fd, buff, strlen(buff));
77
78
    return nbytes;
79
}
80
81
int Client::Read(char * buff)
82
{
83
    int nbytes = read(this->sock_fd, buff, MAX_BUFF_SIZE);
84
    buff[nbytes] = '\0';
85
86
    return nbytes;
87
}
88
89
int Client::Close()
90
{
91
    close(this->sock_fd);
92
    printf("fd-%d exit!\n", this->sock_fd);
93
}
94
95
int main(int argc, char *argv[]) {
96
    Client client(argv[1], 5555);
97
    char buff[1024];
98
    char send_msg[] = "hello epoll!";
99
100
    client.Connect();
101
102
    while(true) {
103
        client.Write(send_msg);
104
        client.Read(buff);
105
        std::cout << buff << std::endl;
106
        sleep(1);
107
    }
108
}

程序运行结果：
服务器

10个同时运行的客户端

参考文献：
https://www.cnblogs.com/zhangmingda/p/9396994.html
https://www.jianshu.com/p/397449cadc9a

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

<br>并非所有流浪者都迷失了自我</br>