前言

简单来讲I/O多路复用就是用一个进程来监听多个文件描述符(fd)，我们将监听的fd通过系统调用注册到内核中，如果有一个或多个fd可读或可写，内核会通知应用程序来对这些fd做读写操作，select、poll、epoll都是用于处理此类问题的系统API，只不过注册和调用的方式略有不同。
例如telnet命令的操作，telnet命令从shell读入数据然后写到socket fd上，同时也需要从socket fd上读数据写到shell上。telnet server需要从socket读出命令并发送给shell，再将命令执行结果返回给telnet客户端。此时对于telnet命令来说，需要接收用户输入和sockfd的输入，也需要输出给用户和socket fd，这两种输入和输出是无序的，不能单纯的阻塞某一个读操作，如何处理这种场景？

  1. 将两个read fd设置为非阻塞，然后轮询两个read fd，如果第一个收到数据，则处理，之后再看第二个read fd是否有数据需要读取，如此往复。
  2. 使用多进程或者多线程，将用户输入和输出到sockfd作为一条通道。将sockfd输入和输出给用户作为一条通道。

这样父进程读入用户数据后会发送给socketfd到telenetd，子进程读入telnetd数据后发送给用。当用户终止父进程时，需要发送信号给子进程。当子进I/O结束终止时，父进程也需要接收子进程的结束信号。使用多线程同样需要一些复杂的线程间同步操作。

  1. 异步I/O的方式，对两个read fd使用不同的信号，使用不同的处理函数处理。

以上三种方法在读写连接少的时候没什么问题，当一个server进程需要维护成千上万条通信连接时就会出问题。第1种会无端浪费cpu，第2种就算使用线程进程池来避免上下文切换的开销，当连接数量过多的时候，会占用大量的内存，第3种使用异步I/O显然信号类型肯定是不够用的。所以为了应对此类问题，有了I/O多路复用的技术。

  1. 使用select、poll、epoll，将两个read fd注册到内核，I/O多路复用会阻塞直到有read请求过来，然后返回通知应用，应用针对不同的描述符进行不同的操作。这样可以做到在一个进程中监听并处理多个描述符，再搭配线程池使用，则可以尽量的减少cpu和内存的使用，自然可以维护更多的连接。

select

先看一下select的创建函数

#include 

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

// 监听描述符数目
// readfds、writefds、exceptfds表示可读、可写、异常事件对应的fd
// timeout表示select阻塞多长时间后返回，NULL为一直阻塞、0为立即返回、或指定超时时间

/* 
    返回值：
    0表示超时时间内没有就绪的fds
    成功时返回就绪fds总数（读、写、异常）
    失败返回-1并设置errno，如果select等待期间被信号中断则立即返回-1并设置errno为EINTR
*/
  • fd_set是一个字节数组，每一位标识一个fd。所以通常nfds设置为最大的fd的值+1，在sys/selct.h中可以找到/* Number of descriptors that can fit in an fd_set'. */值为#define __FD_SETSIZE 1024，系统默认单个进程打开最大fd数量ulimit -n`为1024，所以select默认最大只能监听1024个fd。

select通过以下四个宏来对fd_set置位：

void FD_CLR(int fd, fd_set *set); // 清除fd_set中的fd位
int  FD_ISSET(int fd, fd_set *set); // 确认fd是否在fd_set中开启，非0值为开启，0为关闭
void FD_SET(int fd, fd_set *set); // 开启fd在fd_set中的位
void FD_ZERO(fd_set *set); // 清除fd_set的所有位

demo

我们可以使用select的read_fds和exception_fds来接收普通数据和带外数据

#include 
#include 
#include 
#include 
#include 

#include 
#include 
#include 
#define BUFFERSIZE 1024
using namespace std;

int main(int argc, char *argv[]) {
  if (argc

客户端截取部分发送内容

const char *oob_data = "abc";
const char *normal_data = "123";
send(sockfd, normal_data, strlen(normal_data), 0);
send(sockfd, oob_data, strlen(oob_data), MSG_OOB);
send(sockfd, normal_data, strlen(normal_data), 0);
send(sockfd, normal_data, strlen(normal_data), 0);
send(sockfd, normal_data, strlen(normal_data), 0);

运行结果如下，成功的接收到带外数据并处理：

socket与I/O事件触发

socket fd可读事件

  1. 内核接收缓冲区中字节数大于等于SO_RCVLOWAT值（通过getsockopt和setsockopt获取设置），socket可读，recv大于0。对端关闭连接，recv等于0。如果没有资源这次读取不成功recv返回小于0，并且错误码为EAGIN或EWOULDBLOCK errno，这种不算是错误，或许下次读取就可以成功。
  2. socket listenfd有新的连接请求
  3. socket上有未处理的错误，通过getsockopt读取和清除错误

socket fd可写事件

  1. 内核发送缓冲区空间大于等于SO_SNDLOWAT可无阻塞写，send返回大于0
  2. 如果该socket fd已经关闭，再执行写会触发SIGPIPE信号
  3. connect连接成功或超时失败
  4. socket上有未处理的错误，通过getsockopt读取和清除错误

socket fd异常事件

  1. socket上接收到带外数据

poll

poll较select做出了改进，select使用bitmap来监视fds，而poll使用pollfd结构的数组来监视fds，突破了fds数量的限制，通过结构体将fd与events绑定，可以监视更多类型的事件

struct pollfd {
   int   fd;         /* file descriptor */
   short events;     /* requested events 注册的事件*/
   short revents;    /* returned events 实际发生的事件*/
};

常用事件类型

  • POLLIN：数据可读
  • POLLOUT：数据可写
  • POLLRDHUP：TCP连接被对端关闭，或者对端关闭了写操作
  • POLLERR：poll发生错误
  • POLLHUP：管道写端关闭，读端fd收到POLLHUP事件
  • POLLINVAL：fd没有打开

poll的创建函数

int poll(struct pollfd *fds, nfds_t nfds, int timeout)
// fds 是pollfd结构类型的数组
// nfds 指定fds的大小
// timeout 超时时间，-1阻塞，0立即返回

/* 
    返回值：
    0表示超时时间内没有就绪的fds
    成功时返回就绪fds总数（读、写、异常）
    失败返回-1并设置errno，如果select等待期间被信号中断则立即返回-1并设置errno为EINTR
*/

demo

监听两个文件的写入，输出到标准输出

#include 
#include 
#include 

#include 
#include 
#include 
#define BUFFERSIZE 1024
using namespace std;
// 存放pollfd结构数组
pollfd fds[2];

void setnonblocking(int fd) {
  int old_fd_option = fcntl(fd, F_GETFL);
  int new_fd_option = O_NONBLOCK | old_fd_option;
  fcntl(fd, F_SETFL, new_fd_option);
}

int main(int argc, char *argv[]) {
  if (argc  0) {
          cout
  1. 新建文件1.txt和2.txt
  2. 运行server，另起终端随机在1.txt和2.txt上使用echo追加写入内容

server端输出

epoll

epoll与select和poll有很大的差异，epoll将需要监视的fd放入内核的红黑树表中，通过epoll_ctl函数来添加或删除该表中需要监视的fd，只复制已经就绪的fd集合返回给应用。

  • 一方面无需像使用select/poll每次调用都将整个fd集传递给它们。
  • 另一方面在使用的时候应用遍历的都是事件就绪的fd。

创建epoll：

int epoll_create(int size);
// size：提示内核事件表的大小，不是硬限制
// 返回一个fd，所有其他的函数都操作该fd

操作事件：

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
// epfd：epoll_create返回的fd
/* op:
	EPOLL_CTL_ADD 添加fd到epfd，事件集合为event
    EPOLL_CTL_MOD 修改epfd中的fd事件，事件集合为event
    EPOLL_CTL_DEL 从epfd中删除fd，忽略event参数，一般设为NULL
*/
// 返回值：成功返回0，失败返回-1设置errno

获取就绪的事件集

int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
// epfd：epoll_create返回的fd
// events：就绪的事件数组，应用遍历它
// maxevents：指定最大监听的事件数目
// timeout：超时时间，-1阻塞，0立即返回
// 返回值：成功返回就绪fd的数目，失败返回-1设置errno

LT和ET模式

epoll支持两个模式LT（Level Trigger）和ET(Edge Trigger)

  • LT模式可以认为是高效一点的poll，只要fd上有事件发生就会不断的唤醒通知，拿读来说，应用不需要每次都将fd的缓存读完，epoll会不断的通知应用来读取
  • ET模式当触发事件时，只进行一次唤醒通知，不管此次应用是否将fd缓存读完，后续都不会再唤醒，直到新的事件被触发，这样大大减少了同一个事件触发唤醒的次数，减少了epoll_wait系统调用的次数（上下文切换），所以这种模式也被称为高效的epoll模式

EPOLLONESHOT事件

我们说ET模式对于一个事件只会触发一次，如果是多线程的并发场景下，当前线程在读完socket上的数据后开始处理这些数据，在处理期间有新的数据到来，此时唤醒新的线程来处理新到来的数据，出现了两个线程操作同一fd的情况，可能会出现未知错误。EPOLLONESHOT事件可以保证，操作系统对该fd只触发一种事件，并且只触发一次，这样任何时刻只能有一个线程操作该fd。这样也会导致下次该事件无法触发，所以线程处理完毕后应当使用epoll_ctl重置EPOLLONESHOT。

demo

server的主线程与客户端建立TCP连接，建立好连接后将连接fd注册到epoll，如果该链接有请求数据就启动新的线程来处理。使用telnet作为客户端对比不使用EPOLLONESHOT和使用EPOLLONESHOT后server的行为

#include 
#include 
#include 
#include 
#include 
#include 
#include 

#include 
#include 
#include 

using namespace std;
#define MAX_EVENT_NUMBER 1024
#define BUFFERSIZE 1024

static int epollfd = 0;

void setnonblocking(int fd) {
  int old_fd_option = fcntl(fd, F_GETFL);
  int new_fd_option = old_fd_option | O_NONBLOCK;
  fcntl(fd, F_SETFL, new_fd_option);
}

void register_epoll(int epollfd, int fd, bool newfd = false,
                    bool oneshot = false) {
  epoll_event events;
  events.data.fd = fd;
  events.events = EPOLLIN | EPOLLET;  // 读事件、ET工作模式
  if (oneshot) {
    events.events |= EPOLLONESHOT;  // 使用EPOLLONESHOT
  }
  if (newfd) {
    epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &events);
  } else {
    epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &events);
  }
  setnonblocking(fd);
}

void *handle_connect(void *arg) {
  pid_t tid = gettid();
  int connfd = *((int *)arg);
  cout

使用EPOLLONESHOT事件：

  1. telnet1发送c1 h1, 发送c1 h2
  2. telnet2发送c2 h1
  3. telnet3发送c1 h3

server使用线程102108逐个处理 connect5的请求，对于connect6使用线程102109单独处理

不使用EPOLLONESHOT事件：

修改代码

注释掉
// register_epoll(epollfd, connfd, false,
//                true);  // 重置该连接fd的EPOLLONESHOT

不给connfd使用EPOLLONESHOT

// 新的连接使用EPOLLONESHOT属性
// register_epoll(epollfd, connfd, true, true);

// 新的连接不使用EPOLLONESHOT属性
register_epoll(epollfd, connfd, true, false);

编译运行

  1. telnet1发送c1 h1, 发送c1 h2
  2. telnet2发送c2 h1
  3. telnet3发送c1 h3

线程102137处理connfd 5，sleep的期间内，connfd5有新的请求到来，可以看到新起了线程来处理connfd5的新消息

对比总结

select

  • 事件集的传入与使用：select没有fd与event的绑定结构，只是给可读、可写、异常传递一个fd集合，不能处理更多的事件类型，将fd_set拷贝到内核中，内核遍历fd_set，如果有事件发生，内核对fd_set直接修改，将没有事件的fd位置空，拷贝到应用，因此每次调用select都需要重新设置fd_set。应用需要再次完全遍历fd_set，通过FD_ISSET判断事件是否就绪。（两次fd_set拷贝，两次fd_set遍历）
  • 效率：内核处理事件集时间复杂度为O(n)，应用索引就绪文件描述符的时间复杂度为O(n)
  • 工作模式：LT
  • 最大可监视fd数：受限于__FD_SETSIZE 1024宏，可修改该值重新编译内核来增加select可监视fd的数目
  • 可移植性：支持windows、linux

poll

  • 事件集的传入与使用：poll将fd与event绑定在pollfd结构中，将pollfd数组复制到内核，触发事件时内核会修改revents，再将数组复制回用户态，因此无需重置需要监视的成员。但是用户使用遍历的时候仍然需要遍历整个数组成员，判断传入的events是否与返回的revents相同
  • 效率：内核处理事件集时间复杂度为O(n)，应用索引就绪文件描述符的时间复杂度为O(n)
  • 工作模式：LT
  • 最大可监视fd数：系统支持的最大fd数目，/proc/sys/fs/file-max/
  • 可移植性：支持windows、linux

epoll

  • 事件集的传入与使用：epoll在内核中维护一个红黑树结构的事件表，绑定fd与events，这个事件表通过epoll_create创建，返回一个fd来使用，维护着所有需要监视的fd。通过系统调用epoll_ctl对fd对应的事件进行增、删、改。应用代码调用epoll_wait来获取已经触发事件的fd，epoll会将就绪的epoll_event结构的fd放入数组并拷贝到用户态，应用直接遍历该数组即可拿到每一个触发事件的fd
  • 效率：内核处理事件集时间复杂度为O(logn)（操作红黑树），应用索引就绪文件描述符的时间复杂度为O(1)
  • 工作模式：LT或者ET
  • 最大可监视fd数：系统支持的最大fd数目，/proc/sys/fs/file-max/
  • 可移植性：仅支持linux

学习自：
《Linux高性能服务器编程》
《UNIX环境高级编程》
《UNIX系统编程》

文章来源于互联网:
I/O多路复用与socket

上一篇 Pytest系列（31） – config.cache 使用
下一篇 计算机组成原理

相关文章