Linux下IO多路复用

今天药忘吃喽~ 2022-10-02 14:48 157阅读 0赞

### 文章目录 ###

*   *  一、IO多路复用处理数据报文
     *  二、select
     *   *  1. 简介
         *  2. 函数原型
         *   *  2.1 参数说明
             *  2.2 fd\_set结构说明
             *  2.3 timeval结构说明
             *  2.4 返回值说明
         *  3. 就绪条件
         *   *  3.1 读就绪
             *  3.2 写就绪
         *  4. 函数使用
         *  5. 函数特点
         *  6. 函数缺点
     *  三、poll
     *   *  1. 函数原型
         *   *  1.1 参数说明
             *  1.2 pollfd结构说明
             *  1.3 返回值说明
         *  2. 就绪条件
         *  3. 函数使用
         *  4. 优点
         *  5. 缺点
     *  四、epoll
     *   *  1. 函数原型
         *   *  1.1 epoll\_create说明、
             *   *  1.1.1 参数说明
                 *  1.1.2 返回值说明
             *  1.2 epoll\_ctl说明
             *   *  1.2.1 参数说明
                 *  1.2.2 epoll\_event结构说明
             *  1.3 epoll\_wait说明
             *   *  1.3.1 参数说明
                 *  1.3.2 返回值说明
                 *  1.3.3 函数使用示例
         *  2. 工作原理
         *  3. 工作模式
         *   *  3.1 LT模式
             *  3.2 ET模式
         *  4. 函数使用
         *  5. 惊群效应
         *  6. 优点
         *  7. 使用场景

## 一、IO多路复用处理数据报文 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3phbmRhXw_size_16_color_FFFFFF_t_70]

## 二、select ##

### 1. 简介 ###

select系统调用是用来让我们的程序监视多个文件描述符的状态变化的。程序会阻塞到select这里，直到被监听的文件描述符一个或者多个发生了状态变化。

### 2. 函数原型 ###

int select(int nfds, fd_set* read_fds, fd_set* write_fds, 
               fd_set* except_fds, struct timeval* timeout);

#### 2.1 参数说明 ####

*  nfds：是需要监视的最大的文件描述符值+1
 *  read\_sets：对应于需要检测的可读文件描述符的集合
 *  write\_sets：对应于需要检测的可写文件描述符的集合
 *  except\_sets：对应于需要检测的异常文件描述符的集合
 *  timeout：用来设置select的等待时间
    
     *  NULL：表示不设置select的等待时间，select在监听到描述符状态变化之前将一直阻塞
     *  0：表示仅检测描述符集合的状态，然后立即返回，并不等待外部事件的发生
     *  特定的时间值：如果在这个设置的时间值内没有事件发生，select将超时返回

#### 2.2 fd\_set结构说明 ####

首先，我们可以看一下这个结构的定义：

/* The fd_set member is required to be an array of longs. */
    typedef long int __fd_mask;
    
    typedef struct{ 
     	/* something */
    	__fd_mask fds_bits[__FD_SETSIZE / __NFDBITS];
        /* something */
    };

> 注：上面代码节选自<sys/select.h>中，其中我只保留了便于理解的部分。

从定义中我们可以看出，其实`fd_set`结构就是一个long型数组，或者说，它代表一种数据结构----“位图”。使用位图中对应的位来表示要监视的文件描述符。

select提供了一组操纵位图的接口：

void FD_CLR(int fd, fd_set *set);  // 用来清除描述位图set中相关fd的位
    int FD_ISSET(int fd, fd_set *set); // 用来测试描述位图set中相关fd的位是否为真
    void FD_SET(int fd, fd_set *set);  // 用来设置描述位图set中相关fd的位
    void FD_ZERO(fd_set* set);         // 用来清空位图set中的所有位

#### 2.3 timeval结构说明 ####

这个结构的定义：

/* A time value that is accurate to the nearest microsecond but also has a range of years. */
    struct timeval{ 
        __time_t tv_sec; /* Second. */
        __suseconds_t tv_usec; /* Microseconds. */
    };

> 注：上面代码节选自<time.h>中，其中我只保留了便于理解的部分。

#### 2.4 返回值说明 ####

执行成功

*  返回文件描述符中状态已经改变的描述符个数

其他结果

*  返回0，表示在参数传入的timeout时间内没有文件描述符状态发生变化
 *  返回-1，表示在执行中发生错误，错误原因存储于errno中，errno可能的结果有以下几种：
    
     *  EBADF：文件描述符无效或文件已关闭
     *  EINTR：此次select调用被信号打断
     *  EINVAL：参数n为负值
     *  ENOMEM：内存不足

### 3. 就绪条件 ###

#### 3.1 读就绪 ####

*  socket内核中, 接收缓冲区中的字节数, 大于等于低水位标记SO\_RCVLOWAT；
 *  socket TCP通信中, 对端关闭连接, 此时对该socket读, 则返回0；
 *  监听的socket上有新的连接请求；
 *  socket上有未处理的错误。

#### 3.2 写就绪 ####

*  socket内核中, 发送缓冲区中的可用字节数(发送缓冲区的空闲位置大小), 大于等于低水位标记
    
    SO\_SNDLOWAT；
 *  socket的写操作被关闭(close或者shutdown). 对一个写操作被关闭的socket进行写操作, 会触发SIGPIPE
    
    信号；
 *  socket使用非阻塞connect连接成功或失败之后；
 *  socket上有未处理的错误。

### 4. 函数使用 ###

使用select实现一个本地回显程序。

#include <stdio.h>
    #include <stdlib.h>
    #include <unistd.h>
    #include <sys/select.h>
    
    int main() { 
      fd_set read_fds;
      FD_ZERO(&read_fds); //初始化fd_set结构
      FD_SET(STDIN_FILENO, &read_fds); //监听标准输入
      while(1){ 
        printf("> ");
        fflush(stdout);
        int ret = select(STDIN_FILENO+1, &read_fds, NULL, NULL, NULL);
        if(ret < 0){ 
          perror("select");
          continue;
        }
        if(FD_ISSET(STDIN_FILENO, &read_fds)){ 
          char buf[1024] = { 0}; 
          read(STDIN_FILENO, buf, sizeof(buf) - 1); //读取键盘输入
          printf("Echo: %s\n", buf);
        } else { 
          printf("error! invalid fd\n");
          continue;
        }
        FD_ZERO(&read_fds);
        FD_SET(STDIN_FILENO, &read_fds);
      }
      return 0;
    }

### 5. 函数特点 ###

*  可监控的文件描述符个数取决与sizeof(fd\_set)的值。
 *  将fd加入select监控集的同时，还要再使用一个数据结构array保存放到select监控集中的fd，
    
     *  一是用于再select返回后，array作为源数据和fd\_set进行FD\_ISSET判断。
     *  二是select返回后会把以前加入的但并无事件发生的fd清空，则每次开始select前都要重新从array取得fd逐一加入(FD\_ZERO最先)，扫描array的同时取得fd最大值maxfd，用于select的第一个参数。

### 6. 函数缺点 ###

*  每次调用select, 都需要手动设置fd集合, 从接口使用角度来说也非常不便.
 *  每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大
 *  同时每次调用select都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也很大
 *  select支持的文件描述符数量太小

## 三、poll ##

### 1. 函数原型 ###

int poll(struct pollfd* fds, nfds_t nfds, int timeout);

#### 1.1 参数说明 ####

*  fds：fds是一个poll函数监听的结构列表
 *  nfds：表示fds数组的长度
 *  timeout：表示poll函数的超时时间, 单位是毫秒(ms)

#### 1.2 pollfd结构说明 ####

struct pollfd{ 
    	int fd;        	   /* File descriptor to poll. */
    	short int events;  /* Types of events poller cares about. */
    	short int revents; /* Types of events that actually occurred. */
    };

> 注：上面代码节选自<time.h>中

events和revents的取值列表：

<table> 
 <thead> 
  <tr> 
   <th>事件</th> 
   <th>描述</th> 
   <th>是否可作为输入</th> 
   <th>是否可作为输出</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>POLLIN</td> 
   <td>数据可读（包括普通数据和优先数据）</td> 
   <td>是</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>POLLRDNORM</td> 
   <td>普通数据可读</td> 
   <td>是</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>POLLPRI</td> 
   <td>高优先级数据可读，eg: TCP带外数据</td> 
   <td>是</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>POLLOUT</td> 
   <td>数据可写（包括普通数据和优先数据）</td> 
   <td>是</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>POLLWRNORM</td> 
   <td>普通数据可写</td> 
   <td>是</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>POLLWRBAND</td> 
   <td>优先级带数据可写</td> 
   <td>是</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>POLLRDHUP</td> 
   <td>TCP连接被对方关闭，或对方关闭了写操作</td> 
   <td>是</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>POLLERR</td> 
   <td>错误</td> 
   <td>否</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>POLLHUP</td> 
   <td>挂起</td> 
   <td>否</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>POLLNVAL</td> 
   <td>文件描述符未打开</td> 
   <td>否</td> 
   <td>是</td> 
  </tr> 
 </tbody> 
</table>

#### 1.3 返回值说明 ####

执行成功

*  返回文件描述符中状态已经改变的描述符个数

其他结果

*  返回0，表示在参数传入的timeout时间内没有文件描述符状态发生变化
 *  返回-1，表示在执行中发生错误

### 2. 就绪条件 ###

同select

### 3. 函数使用 ###

使用poll实现一个本地回显程序。

#include <sys/poll.h>
    #include <unistd.h>
    #include <stdio.h>
    
    int main(){ 
        struct pollfd poll_fd;
        poll_fd.fd = STDIN_FILEIN;
        poll_fd.events = POLLIN;
        while(1){ 
            printf("> ");
            fflush(stdout);
            int ret = poll(&poll_fd, 1, -1);
            if(0 == ret){ 
                printf("poll timeout\n");
                continue;
            }else if(ret < 0){ 
                perror("poll");
                continue;
            }
            if(POLLIN == poll_fd.revents){ 
                char buf[1024] = { 0};
                read(STDIN_FILENO, buf, sizeof(buf) - 1);
                printf("stdin: %s", buf);
            }
        }
        return 0;
    }

### 4. 优点 ###

*  poll的接口使用比select更方便了一些
 *  poll监听的文件描述符的数目理论上没有上限，但是由于poll底层是采用遍历检测文件描述符是否就绪的方式，所以数量过大后性能还是会下降的

### 5. 缺点 ###

*  和select函数一样，poll返回后，需要轮询pollfd来获取就绪的描述符，监视的文件描述符超过一定数量，轮询的时间开销会线性增长

## 四、epoll ##

### 1. 函数原型 ###

int epoll_create(int size);

#### 1.1 epoll\_create说明、 ####

##### 1.1.1 参数说明 #####

在Linux2.6.8版本后，这个参数会被忽略。但是还是要注意，这个参数不可以传入小于0的数。

##### 1.1.2 返回值说明 #####

epoll\_create返回一个操作epoll的句柄。

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

#### 1.2 epoll\_ctl说明 ####

##### 1.2.1 参数说明 #####

*  epfd：是epoll\_create的返回值，即epoll的句柄。
 *  op：表示epoll\_ctl函数的动作，有三个取值：
    
     *  EPOLL\_CTL\_ADD：注册新的文件描述符fd到epfd中
     *  EPOLL\_CTL\_MOD：修改已经注册到epfd的监听事件
     *  EPOLL\_CTL\_DEL：从epfd中注销一个文件描述符fd
 *  fd：需要监听的文件描述符
 *  event：告诉内核要监听什么类型的事件

##### 1.2.2 epoll\_event结构说明 #####

这个结构体的定义是这样的：

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

> 上面代码节选自<sys/epoll.h>中，我只保留了便于理解的部分。

其中events可以是以下几个宏的集合：

*  EPOLLIN：表示对应的文件描述符可以读（包括对端socket正常关闭）
 *  EPOLLOUT：表示对应的文件描述符可以写
 *  EPOLLPRI：表示对应的文件描述符有紧急数据可读（带外数据）
 *  EPOLLERR：表示对应的文件描述符发生错误
 *  EPOLLHUP：表示对应的文件描述符被挂断
 *  EPOLLET：将EPOLL设置为边缘触发（ET）模式，EPOLL默认为水平触发（LT）模式
 *  EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还要继续监听这个事件，需要重新把这个事件添加到EPOLL队列中

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

#### 1.3 epoll\_wait说明 ####

##### 1.3.1 参数说明 #####

*  epfd：是epoll\_create的返回值，即epoll的句柄。
 *  events：epoll会把已经发生的事件拷贝到events数组中（events不可以是空指针，内核只负责把数据拷贝到这个数组中，不会主动在用户态分配内存存储）
 *  max\_events：events的大小，这个参数不能大于调用epoll\_create时的传入的size
 *  timeout：超时时间，单位是毫秒
    
     *  这个参数填0，epoll\_wait会立即返回
     *  这个参数填-1，epoll\_wait会永久阻塞，直到有事件发生

##### 1.3.2 返回值说明 #####

执行成功

*  返回文件描述符中状态已经改变的描述符个数

其他结果

*  返回0，表示在参数传入的timeout时间内没有文件描述符状态发生变化
 *  返回<0，表示函数执行过程中发生错误

##### 1.3.3 函数使用示例 #####

while(true){ 
    	struct epoll_event epoll_events[1024];
    	int n = epoll_wait(epollfd, epoll_events, 1024, 1000);
    	if(n < 0){ 
    		if(EINTR == errno) //被信号中断
    			continue;
    		break; //出错，直接退出
    	} else if(0 == n) {  //超时
    		continue;
    	}
    	for(size_t i = 0; i != n; ++i){ 
    		if(epoll_events[i].events & POLLIN){ 
    			//TODO:处理可读事件
    		} else if(epoll_events[i].events & POLLOUT) { 
    			//TODO:处理可写事件
    		} else if(epoll_events[i].events & POLLERR) { 
    			//TODO:处理异常事件
    		}
    	}
    }

### 2. 工作原理 ###

当进程调用epoll\_create时，Linux内核会创建一个eventpoll结构体，这个结构体中有两个成员和epoll的使用密切相关：

struct eventpoll{ 
    	/*something*/
        /* 红黑树的根节点，这棵树中存储着所有添加到epoll中需要监控的事件。 */
        struct rb_root rbr;
        /* 双链表的头节点，双链表中存储着要通过epoll_wait返回给用户的满足条件的事件。 */
        struct list_head rblist;
        /*something*/
    };

每一个epoll对象都会在内核中创建一个eventpoll结构体，通过epoll\_ctl方法向epoll对象添加进来的事件，这些事件都会挂载到eventpoll中的红黑树上，事件的结构类型是这样的：

struct epitem{ 
    	struct rb_node rbn; /* 红黑树节点 */
    	struct list_head rdllink; /* 双向链表节点 */
    	struct epoll_filefd ffd; /* 事件句柄信息 */
    	struct eventpoll* ep; /* 指向其所属的 eventpoll 对象 */
    	struct epoll_event event; /* 期待发生的事件类型 */
    };

所有添加到epoll对象中的事件都会与设备驱动程序建立回调关系，当事件发生响应，就会调用这个回调方法。这个回调方法在内核中叫ep\_poll\_callback，它主要的作用是将发生响应的事件添加到eventpoll中的双链表上。

当调用epoll\_wait检查事件是否发生的时候，只需要检查eventpoll对象中的rdlist双链表中是否存在epitem(即事件)元素即可。如果存在，则把rdlist中的事件拷贝到用户态，同时通过返回值将rdlist中的元素数目(事件发生的个数)返回给用户。

### 3. 工作模式 ###

> 先看一下这个例子：
> 
> 已经把一个tcp socket添加到epoll描述符中，这个时候socket的客户端写入了2KB数据，服务器端调用epoll\_wait，epoll\_wait返回，然后调用read，可在服务器端分配的缓冲区只有1KB，所以一次只读取了1KB数据，继续调用epoll\_wait…

#### 3.1 LT模式 ####

当epoll检测到socket上事件就绪的时候，可以选择只处理一部分数据，或者不立即进行处理。例如上面的例子，由于第一次调用epoll\_wait，服务器端只读取了1KB数据，在第二次调用epoll\_wait时，epoll\_wait仍然会立即返回并通知socket读事件就绪，直到socket缓冲区中所有的数据都读取，epoll\_wait才不会因为这个socket而立即返回。

LT模式支持阻塞读写和非阻塞读写。

> 注：select和poll其实就相当于是epoll的LT模式。

#### 3.2 ET模式 ####

当epoll\_wait返回通知socket读时间就绪时，必须立即处理，而且要一次处理完。例如上面的例子，第一次调用epoll\_wait时，只读取了1KB数据，那么第二次调用epoll\_wait时，epoll\_wait就不会再返回。

ET模式支持非阻塞读写。注意：使用ET模式的时候，需要将监听的文件描述符设置为非阻塞。

> 注：Nginx默认用的就是epoll的ET模式。

### 4. 函数使用 ###

/* epoll LT模式的回显服务器 */
    
    #include <stdio.h>
    #include <string.h>
    #include <stdlib.h>
    #include <unistd.h>
    #include <sys/epoll.h>
    #include <sys/socket.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    
    #define CHECK_ERROR(str, n) do{if(n < 0){ perror(str); exit(EXIT_FAILURE); }}while(0)
    
    void ProcessConnect(int listen_fd, int epoll_fd){ 
      struct sockaddr_in client_addr;
      socklen_t len = sizeof(client_addr);
      int connect_fd = accept(listen_fd, (struct sockaddr*)&client_addr, &len);
      if(connect_fd < 0){ 
        perror("accept");
        exit(EXIT_FAILURE);
      }
      printf("client %s:%d connect\n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));
      struct epoll_event ev;
      ev.data.fd = connect_fd;
      ev.events = EPOLLIN;
      int ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, connect_fd, &ev);
      if(ret < 0){ 
        perror("epoll_ctl");
        exit(EXIT_FAILURE);
      }
    }
    
    void ProcessRequest(int connect_fd, int epoll_fd){ 
      char buf[1024] = { 0};
      ssize_t read_size = read(connect_fd, buf, sizeof(buf) - 1);
      if(read_size < 0){ 
        perror("read");
        return;
      }
      if(0 == read_size){ 
        close(connect_fd);
        epoll_ctl(epoll_fd, EPOLL_CTL_DEL, connect_fd, NULL);
        printf("client say: goodbye\n");
        return;
      }
      printf("client say: %s\n", buf);
      write(connect_fd, buf, strlen(buf));
    }
    
    void CorrectUsage(){ 
      printf("Usage: ./epoll_server [ip] [port]\n");
    }
    
    int main(int argc, char* argv[]) { 
      if(argc != 3){ 
        CorrectUsage();
        exit(EXIT_FAILURE);
      } 
      struct sockaddr_in addr;
      addr.sin_family = AF_INET;
      addr.sin_port = htons(atoi(argv[2]));
      addr.sin_addr.s_addr = inet_addr(argv[1]);
      int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
      CHECK_ERROR("socket", listen_fd);
        
      int ret = bind(listen_fd, (struct sockaddr*)&addr, sizeof(addr));  
      CHECK_ERROR("bind", ret);
        
      ret = listen(listen_fd, 10);
      CHECK_ERROR("listen", ret);
        
      int epoll_fd = epoll_create(5);
      CHECK_ERROR("epoll_create", epoll_fd);
        
      struct epoll_event event;
      event.events = EPOLLIN;
      event.data.fd = listen_fd;
      ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &event);
      CHECK_ERROR("epoll_ctl", ret);
        
      while(1){ 
        struct epoll_event events[10];
        int size = epoll_wait(epoll_fd, events, sizeof(events) / sizeof(events[0]), -1);
        if(size < 0){ 
          perror("epoll_wait");
          continue;
        }
        if(0 == size){ 
          printf("epoll timeout\n");
          continue;
        }
        for(int i = 0; i < size; ++i){ 
          if(!(events[i].events & EPOLLIN)){ 
            continue;
          }
          if(events[i].data.fd == listen_fd){ 
            ProcessConnect(listen_fd, epoll_fd);
          }else{ 
            ProcessRequest(events[i].data.fd, epoll_fd);
          }
        }
      }
      return 0;
    }

### 5. 惊群效应 ###

参考：[Epoll的惊群效应][Epoll]

### 6. 优点 ###

前面介绍了一下epoll的底层原理，可以看出它相对于之前的IO多路复用接口做了很多优化：

*  不用再通过轮询的方式来检测事件是否就绪，而是通过回调机制，大大缩短了轮询的时间开销
 *  不用每次使用检测事件是否就绪接口前都要拷贝之前的事件集合，大大减少了拷贝开销
 *  文件描述符的数目理论上没有上限，不像select，“文件描述符的数目取决于 ’位图‘ 的大小，想要更改 ‘位图‘ 大小甚至还要重新编译内核”
 *  接口使用也方便了很多，不需要每次循环都要重新设置关注的文件描述符（eg：select）

### 7. 使用场景 ###

对于多连接，并且多连接中只有一部分连接比较活跃时，比较适合用epoll。例如：各种互联网APP的入口服务器，就很适合用epoll。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3phbmRhXw_size_16_color_FFFFFF_t_70]: /images/20220109/0fd86f863bf3420bb928309726f4c716.png
[Epoll]: https://blog.csdn.net/zanda_/article/details/93136584