IO多路复用

一、概述

IO多路复用：进程同时检查多个文件描述符，以找出他们中的任何一个是否可执行IO操作。

核心：同时检查多个文件描述符，看他们是否准备好了执行IO操作。文件描述符就绪状态的转化是通过一些IO事件来触发。

二、水平触发和边缘触发

水平触发：文件描述符可以非阻塞地执行IO系统调用，此时认为它已经就绪。

边缘触发：文件描述符自上次检查以来有了新IO活动。

水平触发通知时，可以在任意时刻检查文件描述符的就绪状态（即可以多次重复检查，文件描述符一直处于就绪状态），因此，没必要每次当文件描述符处于就绪状态尽可能多执行IO操作。

边缘触发通知时，只有当IO事件发生时我们才会收到通知。因此尽可能多的执行IO操作。一般设计思路：用循环操作IO，IO操作函数应设置为非阻塞状态。

三、select系统调用

1、接口

#include <sys/select.h>/* According to earlier standards */
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>// 系统调用会一直阻塞，直到一个或多个文件描述符集合成为就绪态
// 返回值：-1 - 有错误发生、0 - 无就绪文件描述符、正整数 - 就绪的文件描述符个数
int select(int nfds, fd_set *readfds, fd_set *writefds,fd_set *exceptfds, struct timeval *timeout);// 将文件描述符fd从fdset所指向的集合中移除
void FD_CLR(int fd, fd_set *set);
// 判断fd是否是fdset所指向的集合中的成员
int  FD_ISSET(int fd, fd_set *set);
// 将文件描述符fd添加到由fdset所指向的集合中
void FD_SET(int fd, fd_set *set);
// 将fdset指向的集合初始化为空
void FD_ZERO(fd_set *set);#include <sys/select.h>int pselect(int nfds, fd_set *readfds, fd_set *writefds,fd_set *exceptfds, const struct timespec *timeout,const sigset_t *sigmask);

2、文件描述符说明

readfds：用来检查输入是否就绪的文件描述符集合。

writefds：用来检查输出是否就绪的文件描述符集合。

exceptfds：用来检查异常情况是否发生的文件描述符集合。

四、poll 系统调用

1、接口

struct pollfd {int   fd;         /* file descriptor */short events;     /* requested events */short revents;    /* returned events */
};#include <poll.h>
// 返回值：-1 - 错误、0 - 无就绪文件描述符、正整数 - 就绪文件描述符个数
int poll(struct pollfd *fds, nfds_t nfds, int timeout);#define _GNU_SOURCE         /* See feature_test_macros(7) */
#include <signal.h>
#include <poll.h>int ppoll(struct pollfd *fds, nfds_t nfds,const struct timespec *tmo_p, const sigset_t *sigmask);

2、pollfd结构体中掩码位

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-kXMkb6yE-1678192420784)(E:\note\Linux 应用开发\picture\pollfd结构体中掩码.png)]

五、select和poll存在问题

1、当检查大量处于密集范围内的文件描述符时，该操作耗时的时间将大大超过接下来操作。

2、需要检查fd集合需要在用户空间和内核空间来回拷贝。

3、select和poll返回后，程序需要检查返回的数据结构中的每个元素，以查明哪个文件描述符处于就绪状态。

六、信号驱动IO

七、epoll 编程接口

1、epoll接口概述

epoll API 由以下 3 个系统调用组成。

1、epoll_create() 创建一个epoll实例，返回代表该实例的文件描述符。

2、epoll_ctl() 操作epoll实例相关联的兴趣列表。

3、epoll_wait() 返回与epoll实例相关链的就绪列表成员。

2、创建epoll实例

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

3、修改epoll的兴趣列表

#include <sys/epoll.h>int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

4、事件等待

#include <sys/epoll.h>int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);
int epoll_pwait(int epfd, struct epoll_event *events,int maxevents, int timeout,const sigset_t *sigmask);

附录一：select实现高并发socket服务器

#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <sys/select.h>
#include <sys/time.h>#define BUF_LEN 1024int main(int argc, char *argv[])
{pid_t dPid = 0;struct sockaddr_in addr;struct sockaddr_in cliaddr;socklen_t clilen = sizeof(struct sockaddr_in);int cfd=0;  // 客户端 fdint sfd=0;  // 服务端 fdint i = 0, n = 0;fd_set tmpfds;  // 向select传递参数，并从select返回结果fd_set rdfds;   // 系统对 fd_set 扫描的 fd 管理变量。int maxfd;      // 最大的文件描述符int connfd[FD_SETSIZE]; // 连接描述符管理int maxi = 0;               // connfd有效的文件描述符最大值int nready;     // select 返回值char buf[BUF_LEN] = "";/* 1、初始化select相关变量 */// 文件描述符集初始化FD_ZERO(&tmpfds);FD_ZERO(&rdfds);// 初始化有效的文件描述符集，为-1表示不可用for(i=0;i<FD_SETSIZE;i++){connfd[i] = -1;}/* 2、socket编程 */// 1、创建服务器 socketsfd = socket(AF_INET,SOCK_STREAM,0);if(sfd == -1){return -1;}// 2、初始化 sockaddr_un 结构体memset(&addr,0,sizeof(struct sockaddr_in));addr.sin_family = AF_INET;inet_pton(AF_INET,"127.0.0.1",&addr.sin_addr);addr.sin_port = htons(8888);bzero(&(addr.sin_zero),8);// 3、绑定 IP 地址和端口号if(bind(sfd,(struct sockaddr*)&addr,sizeof(struct sockaddr_in)) == -1){goto ERROTR_RETURN;}// 4、监听客户端连接if(listen(sfd,10) == -1){goto ERROTR_RETURN;}/* 3、添加监听fd到select扫描集合 */FD_SET(sfd,&rdfds);maxfd = sfd;/* 4、调用select扫描文件描述符，并处理相关返回 */while(1){tmpfds = rdfds;nready = select(maxfd+1,&tmpfds,NULL,NULL,NULL);if(nready > 0){// 有就绪文件描述符if(FD_ISSET(sfd,&tmpfds)){// 监听文件描述符就绪cfd = accept(sfd,(struct sockaddr *)&cliaddr,&clilen);if(cfd == -1){printf("accept errno!\r\n");exit(-1);}// 保存当前文件描述符for(i=0;i<FD_SETSIZE;i++){if(connfd[i]==-1){connfd[i] = cfd;break;}}// 若连接总数达到的最大值，关闭但前连接if(i==FD_SETSIZE){close(cfd);printf("too many clients,i==[%d]\n",i);continue;}//确保connfd中maxi保存的是最后一个文件描述符的下标if(i>maxi){maxi = i;}//打印客户端的IP和PORTchar sIP[16];memset(sIP,0x00,sizeof(sIP));printf("receive from client ---->IP[%s],PORT=[%d]\n",inet_ntop(AF_INET,&cliaddr.sin_addr.s_addr,sIP,sizeof(sIP)),htons(cliaddr.sin_port));//将新的文件描述符加入到select监控的文件描述符中FD_SET(cfd,&rdfds);// 调整最大fdif(maxfd<cfd){maxfd = cfd;}//如果没有变化的文件描述符，则无需执行后续代码if(--nready<=0){continue;}}//通信文件描述符就绪for(i=0;i<=maxi;i++){int sockfd = connfd[i];//数组内的文件描述符如果被释放，有可能变为-1if(sockfd==-1){continue;}if(FD_ISSET(sockfd,&tmpfds)){memset(buf,0x00,sizeof(buf));n = read(sockfd,buf,sizeof(buf));if(n<0){// socket 异常perror("read over");close(sockfd);FD_CLR(sockfd,&rdfds);connfd[i] = -1;//将connfd[0]置为-1，表示位置可用}else if(n==0){// socket 关闭printf("client is closed\n");close(sockfd);FD_CLR(sockfd,&rdfds);connfd[i] = -1;//将connfd[0]置为-1，表示位置可用}else{// 数据接收printf("[%d]:[%s]\n",n,buf);write(sockfd,buf,n);}if(--nready<=0){break;//注意这里是break，而不是continue，应该是从最外层的while继续循环}}}}}return 0;
ERROTR_RETURN:close(sfd);return -1;
}

附录二：poll实现高并发socket服务器

#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <poll.h>#define BUF_LEN 1024
#define OPEN_MAX 1024int main(int argc, char *argv[])
{pid_t dPid = 0;struct sockaddr_in addr;struct sockaddr_in cliaddr;socklen_t clilen = sizeof(struct sockaddr_in);int cfd=0;  // 客户端 fdint sfd=0;  // 服务端 fdstruct pollfd client[OPEN_MAX]; // 文件描述符集管理数组int nfds = 0;                   // client最大有效下标int nready = 0;                 // poll 返回值int i = 0;char buf[BUF_LEN] = "";/* 1、poll相关配置 */// 1、初始化文件描述符集管理数组for(int i = 0;i < OPEN_MAX; i++){client[i].fd = -1;          // 文件描述符都初始化为-1client[i].events = POLLIN;  // 默认全部只监听读事件}/* 2、socket编程 */// 1、创建服务器 socketsfd = socket(AF_INET,SOCK_STREAM,0);if(sfd == -1){return -1;}// 2、初始化 sockaddr_un 结构体memset(&addr,0,sizeof(struct sockaddr_in));addr.sin_family = AF_INET;inet_pton(AF_INET,"127.0.0.1",&addr.sin_addr);addr.sin_port = htons(8888);bzero(&(addr.sin_zero),8);// 3、绑定 IP 地址和端口号if(bind(sfd,(struct sockaddr*)&addr,sizeof(struct sockaddr_in)) == -1){goto ERROTR_RETURN;}// 4、监听客户端连接if(listen(sfd,10) == -1){goto ERROTR_RETURN;}/* 3、将监听socket添加到poll扫描集合中 */client[0].fd = sfd; // 要监听的第一个文件描述符是lfdclient[0].events = POLLIN; // 要监听lfd的读事件/* 4、调用poll扫描文件描述符集 */while(1){nready = poll(client,nfds+1,-1);if(client[0].revents & POLLIN){// 有客户端连接int cfd = accept(sfd, (struct sockaddr*)&cliaddr, &clilen);char sIP[16];memset(sIP,0x00,sizeof(sIP));printf("receive from client ---->IP[%s],PORT=[%d]\n",inet_ntop(AF_INET,&cliaddr.sin_addr.s_addr,sIP,sizeof(sIP)),htons(cliaddr.sin_port));for(i = 1; i < OPEN_MAX; i++){if(client[i].fd < 0){client[i].fd = cfd;break;}}if(i == OPEN_MAX){// 客户端连接数操作上限close(cfd);printf("too many clients,i==[%d]\n",i);continue;}// 更新扫描fd集合个数if(i> nfds){nfds = i; }// 判断是否处理完成if(--nready == 0){continue; // 就一个事件，而且是新连接，没必要往下执行了}}// 通信socket状态有变化for(i = 0; i < OPEN_MAX ; i++){if(client[i].fd < 0){// 当前client无效continue;}if(client[i].revents & POLLIN){// 接收到数据int n = read(client[i].fd, buf, sizeof(buf));if(n < 0){perror("read err");close(client[i].fd); // 该文件描述符无用了，关闭client[i].fd = -1;}if(n == 0){printf("client[%d] closed\n",i);close(client[i].fd);client[i].fd = -1;}else{write(client[i].fd, buf, n);printf("[%d]:[%s]\n",n,buf);}if(--nready == 0){break; // 为了防止循环1024}}}}return 0;
ERROTR_RETURN:close(sfd);return -1;
}

附录三：epoll实现高并发socket服务器

#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <errno.h>#define BUF_LEN 1024
#define OPEN_MAX 1024int main(int argc, char *argv[])
{pid_t dPid = 0;struct sockaddr_in addr;struct sockaddr_in cliaddr;socklen_t clilen = sizeof(struct sockaddr_in);int cfd=0;  // 客户端 fdint sfd=0;  // 服务端 fdint clientfd=0;struct epoll_event event;   // 告诉内核要监听什么事件  struct epoll_event wait_event[OPEN_MAX]; //内核监听完的结果int i = 0;int nready = 0;char buf[BUF_LEN] = "";/* 1、socket编程 */// 1、创建服务器 socketsfd = socket(AF_INET,SOCK_STREAM,0);if(sfd == -1){return -1;}// 2、初始化 sockaddr_un 结构体memset(&addr,0,sizeof(struct sockaddr_in));addr.sin_family = AF_INET;inet_pton(AF_INET,"127.0.0.1",&addr.sin_addr);addr.sin_port = htons(8888);bzero(&(addr.sin_zero),8);// 3、绑定 IP 地址和端口号if(bind(sfd,(struct sockaddr*)&addr,sizeof(struct sockaddr_in)) == -1){goto ERROTR_RETURN;}// 4、监听客户端连接if(listen(sfd,10) == -1){goto ERROTR_RETURN;}/* 2、epoll 配置 */// 创建一个 epoll 的句柄，参数要大于 0， 没有太大意义int epfd = epoll_create(10);  if( -1 == epfd ){  perror ("epoll_create");  return -1;  }// 将sfd添加到扫描fd集合event.data.fd = sfd;     //监听套接字  event.events = EPOLLIN; // 表示对应的文件描述符可以读int ret = epoll_ctl(epfd, EPOLL_CTL_ADD, sfd, &event);  if(-1 == ret){  perror("epoll_ctl");  return -1;  } while(1){nready = epoll_wait(epfd, wait_event, OPEN_MAX, -1);if(nready == -1){exit(EXIT_FAILURE);}for(int i = 0 ; i < nready; i++){if(wait_event[i].data.fd == sfd){// 有客户端连接cfd = accept(sfd, (struct sockaddr *)&cliaddr, &clilen);if(cfd == -1){//连接建立失败continue;}event.events = EPOLLIN;event.data.fd = cfd;epoll_ctl(epfd, EPOLL_CTL_ADD, cfd, &event);}else{// 通信socket准备就绪clientfd = wait_event[i].data.fd;ret = recv(clientfd, buf, 1024, 0);if(ret < 0){//现在是不可能-1的，但是如果一直while读，读空了就会返回-1if(errno == EAGAIN || errno == EWOULDBLOCK){continue; //资源暂不可用， 再尝试一次}else{//出错close(clientfd);event.events = EPOLLIN;event.data.fd = clientfd;epoll_ctl(clientfd, EPOLL_CTL_DEL,clientfd, &event);}}else if(ret == 0){ // recv 返回0代表连接已经断开printf("连接被关闭%d\n",clientfd);close(clientfd);event.events = EPOLLIN;event.data.fd = clientfd;epoll_ctl(clientfd, EPOLL_CTL_DEL,clientfd, &event);                }else{printf("[%d]:[%s]\n",ret,buf);}}}}return 0;
ERROTR_RETURN:close(sfd);return -1;
}