IO复用技术（1）——select/poll/epoll原理介绍及使用案例

文章目录

1.Select
- 1.1 工作流程
- 1.2 fd_set函数
- 1.3 select函数
- 1.4 例程
2.poll
- 2.1 poll函数
- 2.2 例程
3.epoll
3.1 工作流程
- 3.2 相关函数
- 3.3 epoll的两种工作模式
- 3.4 示例代码
4.总结

原理：使用一个线程来检查多个文件描述符，委托内核进行检查，如果有一个文件描述符就绪，则返回，否则阻塞直到超时，大大减少需要的线程数量、内存开销和上下文切换的CPU开销（比如一个事用1000个线程去做，但如果使用IO复用，可以只用一个线程）。

1.Select

1.1 工作流程

需要进行IO操作的socket 添加到socket
阻塞直到select系统调用返回（委托内核进行操作）
用户线程发起read请求
内核进行数据拷贝，给用户线程，完成read

有一张图非常的形象：

1.2 fd_set函数

#define __FD_SETSIZE 1024typedef struct {
unsigned long fds_bits[__FD_SETSIZE / (8 * sizeof(long))];
} __kernel_fd_set;// 将文件描述符fd从set集合中删除 
void FD_CLR(int fd, fd_set *set); // 判断文件描述符fd是否在set集合中 
int  FD_ISSET(int fd, fd_set *set); // 将文件描述符fd添加到set集合中 
void FD_SET(int fd, fd_set *set); // 将set集合中, 所有文件描述符对应的标志位设置为0
void FD_ZERO(fd_set *set);

主要用于将文件描述符fd与fd_set集合进行关联

1.3 select函数

int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

readfds：内核检测该集合中的IO是否可读。
writefds：内核检测该集合中的IO是否可写
exceptfds：内核检测该集合中的IO是否异常
nfds：以上三个集合中最大的文件描述符数值 + 1，例如集合是{0,1,4}，那么 maxfd 就是 5
timeout：用户线程调用select的超时时长
timeout = NULL，等待无限长时间
timeout = 0，不等待，立刻返回
timeout>0，等待指定时间
返回值：
大于0：成功，返回集合中已就绪的IO总个数
等于-1：调用失败
等于0：没有就绪的IO

1.4 例程

先用FD_ZERO将位置0，然后使用FD_SET设置所监听的文件描述符到fd_set，select函数进行监听，当select返回大于0，则使用FD_ISSET遍历所有fd到maxfd，如果可操作，则去操作，操作完后需要用FD_CLR清除已产生的事件。
假设fd = 1，fd = 2上发生事件，则select返回时，rset的值为0x0003，当处理完fd = 1的事件，调用FD_CLR后，则值变为0x0002，以此类推
服务端代码：

#include <stdio.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <sys/select.h>int main() 
{// 创建socketint lFd = socket(PF_INET, SOCK_STREAM, 0);if (lFd < 0) {printf("socket error\n");return -1;}struct sockaddr_in saddr;saddr.sin_port = htons(9999);saddr.sin_family = AF_INET;saddr.sin_addr.s_addr = INADDR_ANY;int iRet = 0;// 绑定iRet = bind(lFd, (struct sockaddr *)&saddr, sizeof(saddr));if (iRet < 0) {printf("bind error\n");return -1;}// 监听iRet = listen(lFd, 8);if (iRet < 0) {printf("listen error\n");return -1;}int maxFd = lFd;fd_set allFdSets, tmpFdSets;FD_ZERO(&allFdSets);FD_SET(lFd, &allFdSets);while(1) {memcpy(&tmpFdSets, &allFdSets, sizeof(tmpFdSets));iRet = select(maxFd + 1, &tmpFdSets, NULL, NULL, 0);if (iRet == -1){perror("select error:");continue;}else if (iRet == 0){printf("select return no results\n");continue;}else{for (int i = lFd; i < maxFd + 1; i++){if (i == lFd){/// new clientif (FD_ISSET(lFd, &tmpFdSets)){struct sockaddr_in addr = {0};int iLen = sizeof(addr);int clientFd = accept(lFd, (struct sockaddr *)&addr, &iLen);FD_SET(clientFd, &allFdSets);maxFd = clientFd > maxFd ? clientFd : maxFd;FD_CLR(lFd, &tmpFdSets);}}else{/// msgif (FD_ISSET(i, &tmpFdSets)){char acBuf[1024] = {0};int iLen = read(i, acBuf, sizeof(acBuf));if (iLen == -1){printf("fd:%d error read ret:%d\n", i, iLen);continue;}else if (iLen == 0){printf("fd %d closed\n", i);}else{printf("fd %d, recv buf :%s, return ok\n", i, acBuf);write(i, "ok", strlen("ok"));}}}FD_CLR(i, &tmpFdSets);}}}return 0;
}

客户端代码：

#include <stdio.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>int main() {// 创建socketint fd = socket(PF_INET, SOCK_STREAM, 0);if(fd == -1) {perror("socket");return -1;}struct sockaddr_in seraddr;inet_pton(AF_INET, "127.0.0.1", &seraddr.sin_addr.s_addr);seraddr.sin_family = AF_INET;seraddr.sin_port = htons(9999);// 连接服务器int ret = connect(fd, (struct sockaddr *)&seraddr, sizeof(seraddr));if(ret == -1){perror("connect");return -1;}int num = 0;while(1) {char sendBuf[1024] = {0};sprintf(sendBuf, "data%d", num++);printf("write buf:%s\n", sendBuf);write(fd, sendBuf, strlen(sendBuf) + 1);char recvBuf[1024] = {0};// 接收int len = read(fd, recvBuf, sizeof(recvBuf));if(len == -1) {perror("read");return -1;}else if(len > 0) {printf("read buf = %s\n", recvBuf);} else {printf("服务器已经断开连接...\n");break;}sleep(1);}close(fd);return 0;
}

2.poll

2.1 poll函数

pol和select原理基本相同，使用起来稍微有点差别，它没有最大1024文件描述符限制，也不需要每次重置fd_set数组的值。

struct pollfd {int fd; /* file descriptor */short events; /* events to look for */short revents; /* events returned */
};int poll(struct pollfd *fds, unsigned long nfds, int timeout);

fds：struct pollfd类型的数组, 存储了待检测的文件描述符，struct pollfd有三个成员
fd：委托内核检测的文件描述符
events：委托内核检测的fd事件（输入、输出、错误），每一个事件有多个取值
revents：这是一个传出参数，数据由内核写入，存储内核检测之后的结果
其中event的取值如下，不同事件对应不同值

nfds：描述的是数组 fds 的大小
timeout: 指定poll函数的阻塞时长，-1代表无限等待
返回值：
-1：失败，并设置errno，可以用perror打印
大于0：检测的集合中已就绪的文件描述符的总个数

2.2 例程

#include <stdio.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <sys/select.h>
#include <poll.h>
#include <signal.h>int main() 
{// 创建socketint lFd = socket(PF_INET, SOCK_STREAM, 0);if (lFd < 0) {printf("socket error\n");return -1;}struct sockaddr_in saddr;saddr.sin_port = htons(9999);saddr.sin_family = AF_INET;saddr.sin_addr.s_addr = INADDR_ANY;int iRet = 0;// 绑定iRet = bind(lFd, (struct sockaddr *)&saddr, sizeof(saddr));if (iRet < 0) {printf("bind error\n");return -1;}// 监听iRet = listen(lFd, 8);if (iRet < 0) {printf("listen error\n");return -1;}int nFds = 0;struct pollfd fds[512] = {0};int maxFds = sizeof(fds)/ sizeof(fds[0]);for (int i = 0; i < maxFds; i++){fds[i].fd = -1;fds[i].events = POLLIN;}fds[0].fd = lFd;fds[0].events = POLLIN;while(1) {iRet = poll(fds, nFds + 1, -1);if (iRet == -1){perror("poll error:");continue;}else if (iRet == 0){printf("poll return no results\n");continue;}else{/// new clientif (fds[0].revents & POLLIN){struct sockaddr_in cliaddr;int len = sizeof(cliaddr);int cfd = accept(lFd, (struct sockaddr *)&cliaddr, &len);printf("new client connect\n");for (int i = 1; i < maxFds; i++){if (fds[i].fd == -1){fds[i].fd = cfd;fds[i].events = POLLIN;nFds = i > nFds ? i : nFds;printf("new client connect success, fd:%d, nFds:%d\n", fds[i].fd, nFds);break;}}}/// client have datafor (int i = 1; i <= nFds; i++){if (fds[i].revents & POLLIN){char acBuf[1024] = {0};int iLen = read(fds[i].fd, acBuf, sizeof(acBuf));if (iLen == -1){printf("fd:%d error read ret:%d\n", i, iLen);continue;}else if (iLen == 0){if (i == nFds){nFds--;}printf("fd %d closed, relase source, nFds:%d\n", fds[i].fd, nFds);fds[i].fd = -1;fds[i].events = POLLIN;}else{printf("fd %d, recv buf :%s, return ok\n", fds[i].fd, acBuf);write(fds[i].fd, "ok", strlen("ok"));}}}}}return 0;
}

#include <stdio.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>int main() {// 创建socketint fd = socket(PF_INET, SOCK_STREAM, 0);if(fd == -1) {perror("socket");return -1;}struct sockaddr_in seraddr;inet_pton(AF_INET, "127.0.0.1", &seraddr.sin_addr.s_addr);seraddr.sin_family = AF_INET;seraddr.sin_port = htons(9999);// 连接服务器int ret = connect(fd, (struct sockaddr *)&seraddr, sizeof(seraddr));if(ret == -1){perror("connect");return -1;}int num = 0;while(1) {char sendBuf[1024] = {0};sprintf(sendBuf, "data%d", num++);printf("write buf:%s\n", sendBuf);write(fd, sendBuf, strlen(sendBuf) + 1);char recvBuf[1024] = {0};// 接收int len = read(fd, recvBuf, sizeof(recvBuf));if(len == -1) {perror("read");return -1;}else if(len > 0) {printf("read buf = %s\n", recvBuf);} else {printf("server disconnect...\n");break;}sleep(1);}close(fd);return 0;
}

3.epoll

Epoll 没有最大并发连接的限制，上限是最大可以打开文件的数目，且epoll采用红黑树管理文件描述符，效率会更高

3.1 工作流程

正如这个图一样，epoll相比较select和poll一个比较大的优点就在于，它能够准确告知应用层是哪一个事件来了，而不需要去一个个遍历，减少很大一部分开销

epoll整体流程如下：
在内核中创建了一个数据，这个数据中有两个比较重要的数据，一个是需检测文件描述符信息（红黑树），还有一个是就绪列表，存放已改变的文件描述符信息（双向链表）

3.2 相关函数

创建epoll句柄

int epoll_create(int size);  
int epoll_create1(int flags);

控制epoll实例，主要是增加或删除需要监控的IO事件

struct epoll_event {__uint32_t events;epoll_data_t data;
};union epoll_data {void     *ptr;int       fd;uint32_t  u32;uint64_t  u64;
};
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

epfd：epoll句柄
op：操作选项
EPOLL_CTL_ADD：向 epoll 句柄注册文件描述符对应的事件
EPOLL_CTL_DEL：向 epoll 句柄删除文件描述符对应的事件
fd：操作的文件描述符
event：注册的事件类型，并且可以通过这个结构体设置用户自定义数据
events：注册的事件类型
data：用户自定义数据

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

等待I/O事件就绪

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

**epfd：**epoll句柄
events：出参，代表发生变化的文件描述符信息，可能是多个
maxevents：events的结构体数组大小
timeout：
-1，一直阻塞，直到有事件就绪后返回
0，不阻塞，函数马上返回
大于0：等待指定时间后返回

3.3 epoll的两种工作模式

epoll由两种工作模式，分别为LT模式（条件触发）、ET模式（边缘触发），默认为条件触发
条件触发：只要输入缓冲有数据，便一直触发事件
a. 用户不读数据，数据一直在缓冲区，epoll 会一直通知
b. 用户只读了一部分数据，epoll会通知
c. 缓冲区的数据读完了，不通知
边缘触发：只有描述符从未就绪变为就绪时，才会为文件描述符发送一次就绪通知，之后不再通知
减少了事件被重复触发的次数，效率比LT模式高，且可以分离接收数据和处理数据的时间点，工作在该模式必须要使用非阻塞等待
a. 用户不读数据，数据一直在缓冲区中，epoll下次检测的时候就不再次通知了
b. 用户只读了一部分数据，epoll不再次通知
c. 缓冲区的数据读完了，不再次通知

3.4 示例代码

条件触发模式

#include <stdio.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>int main() {// 创建socketint lfd = socket(PF_INET, SOCK_STREAM, 0);struct sockaddr_in saddr;saddr.sin_port = htons(9999);saddr.sin_family = AF_INET;saddr.sin_addr.s_addr = INADDR_ANY;// 绑定bind(lfd, (struct sockaddr *)&saddr, sizeof(saddr));// 监听listen(lfd, 8);// 调用epoll_create()创建一个epoll实例int epfd = epoll_create(100);// 将监听的文件描述符相关的检测信息添加到epoll实例中struct epoll_event epev;epev.events = EPOLLIN;epev.data.fd = lfd;epoll_ctl(epfd, EPOLL_CTL_ADD, lfd, &epev);struct epoll_event epevs[1024];while(1) {int ret = epoll_wait(epfd, epevs, 1024, -1);if(ret == -1) {perror("epoll_wait");exit(-1);}printf("ret = %d\n", ret);for(int i = 0; i < ret; i++) {int curfd = epevs[i].data.fd;if(curfd == lfd) {// 监听的文件描述符有数据达到，有客户端连接struct sockaddr_in cliaddr;int len = sizeof(cliaddr);int cfd = accept(lfd, (struct sockaddr *)&cliaddr, &len);epev.events = EPOLLIN;epev.data.fd = cfd;epoll_ctl(epfd, EPOLL_CTL_ADD, cfd, &epev);} else {if(epevs[i].events & EPOLLOUT) {continue;}   // 有数据到达，需要通信char buf[1024] = {0};int len = read(curfd, buf, sizeof(buf));if(len == -1) {perror("read");exit(-1);} else if(len == 0) {printf("client closed...\n");epoll_ctl(epfd, EPOLL_CTL_DEL, curfd, NULL);close(curfd);} else if(len > 0) {printf("read buf = %s\n", buf);write(curfd, buf, strlen(buf) + 1);}}}}close(lfd);close(epfd);return 0;
}

#include <stdio.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>int main() {// 创建socketint fd = socket(PF_INET, SOCK_STREAM, 0);if(fd == -1) {perror("socket");return -1;}struct sockaddr_in seraddr;inet_pton(AF_INET, "127.0.0.1", &seraddr.sin_addr.s_addr);seraddr.sin_family = AF_INET;seraddr.sin_port = htons(9999);// 连接服务器int ret = connect(fd, (struct sockaddr *)&seraddr, sizeof(seraddr));if(ret == -1){perror("connect");return -1;}int num = 0;while(1) {char sendBuf[1024] = {0};// sprintf(sendBuf, "send data %d", num++);fgets(sendBuf, sizeof(sendBuf), stdin);write(fd, sendBuf, strlen(sendBuf) + 1);// 接收int len = read(fd, sendBuf, sizeof(sendBuf));if(len == -1) {perror("read");return -1;}else if(len > 0) {printf("read buf = %s\n", sendBuf);} else {printf("服务器已经断开连接...\n");break;}}close(fd);return 0;
}