[IO复用] IO复用问答

记录一下IO复用相关的基础知识。

文章目录

- 阻塞和非阻塞
- 同步和异步
- 为什么使用IO复用
- 什么是IO复用
- IO复用有哪些方式
- - select IO复用
  - poll IO复用
  - epoll IO复用
  - 什么时候用select 或者 epoll？
- select、poll、epoll的区别
- Windows中的IO复用
- Reactor模式
- - C10K问题
  - C10M问题
- 面试题
- 参考

阻塞和非阻塞

如果当前调用的函数，会立即返回，他就是非阻塞的。
入股哦当前函数，不能立即返回，必须等待，就是阻塞的。

我们可以用socket中的recv()函数来举例子，recv通过设置可以设置为阻塞和非阻塞。
当recv()被设置为阻塞时，如果没有收到信息，就会一直等待，直到超时（如果设置了超时），不会继续执行程序。
让recv()被设置为非阻塞时，缓冲区中没有未读取的消息，recv会立刻返回。

关于recv和send的具体情况，可以参考：
[IO复用] recv()和send()的阻塞和非阻塞、返回值、超时

同步和异步

同步和异步是一个相对的概念。
我们需要等待一个函数执行完，再执行下一个函数，这个就是同步。
我们调用一个函数，不用等待它执行完，就继续执行，这个就是异步。
多线程就是实现异步的一种方法。

为什么使用IO复用

想在一个线程中，对多个IO进行管理，就出现了IO复用的方式。

什么是IO复用

IO多路复用，多路指的是可以管理多个网络连接，复用是指的再一个线程中实现。
其功能是就是，通过IO多路复用，实现在一个线程中，监视多个网络socket。
由系统来查询哪些socket有活动，如果有，IO多路复用函数就返回它，用户层进行对应操作。
如果没有活动的socket，就阻塞，让出cpu。

在一个线程中就实现了多个socket的管理，就避免了一个连接一个线程，
数量多了以后导致的资源浪费的情况。

常见的IO模型分为：阻塞IO、非阻塞IO、IO复用、异步IO。
IO复用可以实现reactor模型。
异步IO可以实现proactor模型。

异步IO，Linux中有信号IO、IO uring（Linux 5.1以后）等。
windows有 overlapped IO和IOCP等。

IO复用有哪些方式

IO复用通常有select、poll、epoll，他们都是同步IO。其中epoll是linux中的，windows是没有的。

select，poll，epoll本质上都是同步I/O，因为他们都需要在读写事件就绪后自己负责进行读写，也就是说这个读写过程是阻塞的，而异步I/O则无需自己负责进行读写，异步I/O的实现会负责把数据从内核拷贝到用户空间。
来源：IO多路复用——深入浅出理解select、poll、epoll的实现

select IO复用

select的函数

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

void FD_CLR(int fd, fd_set *set);
int  FD_ISSET(int fd, fd_set *set);
void FD_SET(int fd, fd_set *set);
void FD_ZERO(fd_set *set);

select流程

socket()
bind()
listen()
把listen的fd添加到fd_set数组。
select()。
对于select()返回的fd_set数组进行遍历，找到可读可写的fd，进行accept、read、send。
示例代码

#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h> //socket()
#include <unistd.h>  //close()
#include <netinet/in.h> //struct sockaddr
#include <arpa/inet.h> //inet_ntoa()
#include <sys/select.h>

int main()
{
	int sockfd = socket(AF_INET,SOCK_STREAM,0);
	printf("sockfd created.\n");
	
	struct sockaddr_in svraddr = { 0 };
	svraddr.sin_family = AF_INET;
	svraddr.sin_addr.s_addr = htonl(INADDR_ANY);
	svraddr.sin_port =  htons(2048);

	if(0 > bind(sockfd, (struct sockaddr*)&svraddr, sizeof(svraddr))){
		return -1;
	}
	printf("bind sockfd.\n");
	
	listen(sockfd,10);	
	printf("listen sockfd.\n");
	int maxfd = sockfd;
	
	fd_set fds,rset; //创建fd_set数组，fds是源数组，rset是用于select函数返回的监控可写状态的数组。
	FD_ZERO(&fds);//清空fds数组
	FD_SET(sockfd,&fds); //把listen的fd添加到fds数组中
	while(1)
	{			
		rset = fds; //把fds赋值给rset，把rset传入select()，让select监控rset中的fd
		int nready = select(maxfd + 1,&rset,NULL,NULL,NULL);
		if(FD_ISSET(sockfd,&rset)) {			//检查是否是listen的fd有事件
			struct sockaddr cliaddr = {0};
			socklen_t len = sizeof(cliaddr);
			int clientfd = accept(sockfd,(struct sockaddr*)&cliaddr,&len);
			maxfd = clientfd;  //fd的最大值需要更新
			FD_SET(clientfd,&fds); //把accept的fd添加到fds中
			
			struct sockaddr_in* ptr = (struct  sockaddr_in*)&cliaddr;
			printf("accept,clientfd=%d,clientaddr=%s,clientport=%d\n",
				clientfd,inet_ntoa(ptr->sin_addr),
				ntohs(ptr->sin_port));
		}
		for(int i = sockfd + 1; i < maxfd + 1; ++i)	{ //检查listen fd以外的，监控的fd是否有事件
			if(FD_ISSET(i,&fds)){
				int clientfd = i;
				char buff[128] = { 0x00 };
				int msg_count = recv(clientfd,buff,sizeof(buff),0);
				if(msg_count == 0){
					printf("clientfd=%d disconnect.\n",clientfd);
					FD_CLR(clientfd,&fds);
					close(clientfd);
					continue;
				}
				printf("recv,clientfd=%d,msg_count=%d,msg=%s\n",clientfd,msg_count,buff);
				send(clientfd,buff,msg_count,0);			
			}
		}
	}
}

select的缺点

fd_set是一个位图，FD_SETSIZE是1024，fd_set最大就能容纳1024位，也就是select能监控的fd是有数量上限1024的。（windows select的默认FD_SETSIZE 是64，可以手动修改为最大1024）。
select的效率不高，因为传入select的fd_set数组，需要从用户态拷贝到内核态，内核态是遍历来检查是否有fd就绪，然后结果传回用户态。而到了用户态，也需要遍历一遍。
在用户态把fd_set几个数组传入select函数，就有从用户态到内核态的拷贝，高并发的时候，这样开销比较大。

poll IO复用

poll没用过，只是略微了解，这里直接搬运文章中的代码示例。
一网打尽：面试中的 IO 多路复用高频题！

poll函数

#include <poll.h>
// 数据结构
struct pollfd {
    int fd;                         // 需要监视的文件描述符
    short events;                   // 需要内核监视的事件
    short revents;                  // 实际发生的事件
};
 
// API
int poll(struct pollfd fds[], nfds_t nfds, int timeout);

poll示例

// 先宏定义长度
#define MAX_POLLFD_LEN 4096  
 
int main() {
  /*
   * 在这里进行一些初始化的操作，
   * 比如初始化数据和socket等。
   */
 
  int nfds = 0;
  pollfd fds[MAX_POLLFD_LEN];
  memset(fds, 0, sizeof(fds));
  fds[0].fd = listenfd;
  fds[0].events = POLLRDNORM;
  int max  = 0;  // 队列的实际长度，是一个随时更新的，也可以自定义其他的
  int timeout = 0;
 
  int current_size = max;
  while (1) {
    // 阻塞获取
    // 每次需要把fd从用户态拷贝到内核态
    nfds = poll(fds, max+1, timeout);
    if (fds[0].revents & POLLRDNORM) {
        // 这里处理accept事件
        connfd = accept(listenfd);
        //将新的描述符添加到读描述符集合中
    }
    // 每次需要遍历所有fd，判断有无读写事件发生
    for (int i = 1; i < max; ++i) {     
      if (fds[i].revents & POLLRDNORM) { 
         sockfd = fds[i].fd
         if ((n = read(sockfd, buf, MAXLINE)) <= 0) {
            // 这里处理read事件
            if (n == 0) {
                close(sockfd);
                fds[i].fd = -1;
            }
         } else {
             // 这里处理write事件     
         }
         if (--nfds <= 0) {
            break;       
         }   
      }
    }
  }

poll的缺点
poll和select唯一不同就是没有监控fd的上限限制（因为他传入的是用户指定长度的pollfd数组）。
但是由于用户态拷贝到内核态的资源浪费、遍历导致的效率低，还是存在。

epoll IO复用

epoll主要函数

int epoll_create(int size);
int epoll_wait(int epfd, struct epoll_event *events,
                      int maxevents, int timeout);
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

epoll示例

#include <stdio.h>
#include <sys/socket.h> //socket()
#include <sys/epoll.h>	//epoll
#include <unistd.h> 	//close()
#include <netinet/in.h> //struct sockaddr

#include <errno.h>		//perror()

int main()
{
	int sockfd = socket(AF_INET,SOCK_STREAM,0);

	struct sockaddr_in svraddr;
	svraddr.sin_family = AF_INET;
	svraddr.sin_addr.s_addr = htonl(INADDR_ANY);
	svraddr.sin_port = htons(2048);
	if(	bind(sockfd,(struct sockaddr*)&svraddr,sizeof(svraddr)) < 0) {
		perror("bind");
		return -1;
	} 

	listen(sockfd,10);

	int epfd = epoll_create(1); //这是一个链表，想要监听的fd是添加在这里面的 //从Linux 2.6.8开始参数大于0就可以
	struct epoll_event ev;
	ev.events = EPOLLIN;
	ev.data.fd = sockfd; //为要监听的fd设置监听事件
	epoll_ctl(epfd,EPOLL_CTL_ADD,sockfd,&ev); //将要监听的socket和事件，添加到epoll_fd中

	struct epoll_event events[1024] = { 0x00 }; //这是用于存储epoll_wait返回的就绪fd的数组
	
	while(1){		
		printf("epoll_wait()\n");
		int nready = epoll_wait(epfd,events,1024,-1); //最后一个参数是timeout，设置成-1就是一直阻塞
		if(nready > 0){
			for(int i = 0; i < 1024; ++i){
				if(events[i].data.fd == sockfd){ //如果就绪的是listen的fd，就accept，并把accept到的fd，添加到epoll_fd中
					struct sockaddr cliaddr;
					socklen_t len = sizeof(cliaddr);
					int clientfd = accept(sockfd,(struct sockaddr*)&cliaddr,&len);
					printf("accept,clientfd=%d\n",clientfd);

					ev.data.fd = clientfd;
					ev.events = EPOLLIN | EPOLLET;
					epoll_ctl(epfd,EPOLL_CTL_ADD,clientfd,&ev);
				}	
				else if(events[i].events & EPOLLIN){
					int clientfd = events[i].data.fd;
					char buff[12] = { 0x00 };
					printf("try clientfd=%d recv.\n",clientfd);
					int msg_count = recv(clientfd,buff,sizeof(buff),0);
					if(msg_count == 0){ //recv 返回0，证明连接断开，需要在epoll_fd中删除监听的fd，并且close fd
						printf("clientfd=%d disconnect.\n",clientfd);
						epoll_ctl(epfd,EPOLL_CTL_DEL,clientfd,NULL);
						close(clientfd);
						continue;
					}
					printf("recv,clientfd=%d,msg_count=%d,msg=%s\n",clientfd,msg_count,buff);
					send(clientfd,buff,msg_count,0);
				}
			}
		}
	}	
}

几个注意点
1.struct epoll_event 中有一个成员data,他是一个联合体，可以使用其中的fd直接存储要监听的fd，
也可以通过void *ptr 来存储一个回调的结构体。
void *ptr 的使用可以参考：
[IO复用]epoll_data_t的void *ptr和int fd的使用区别
要注意的是，不同的fd的void *ptr 如果指向了同一个结构体，那么就都会修改这同一个结构体了。
为了避免这种情况，可以为不同的fd 指向不同的结构体对象。

struct epoll_event
{
  uint32_t events;	/* Epoll events */
  epoll_data_t data;	/* User data variable */
} __EPOLL_PACKED;
typedef union epoll_data
{
  void *ptr;
  int fd;
  uint32_t u32;
  uint64_t u64;
} epoll_data_t;

epoll_create() 的参数，设置一个非零值就可以了。它返回的epoll fd也是个文件描述符，最好使用完close掉。

LT水平触发和边沿触发
epoll 可以设置为水平触发或者边沿触发。
水平触发就是如果可读，读缓冲区中只要有数据，就会一直触发epoll_wait返回fd的EPOLLIN事件。
边沿触发是只在缓冲区有数据可读时候，触发一次epoll_wait返回fd的EPOLLIN事件，即使没有读完，直到下次再有数据流入之前都不会再提示了。

epoll默认就是水平触发，边沿触发需要进行以下设置：

ev.data.fd = acceptfd;
ev.events = EPOLLIN;
ev.events |= EPOLLET; //这里就是对事件设置边沿触发
epoll_ctl(epfd, EPOLL_CTL_ADD, acceptfd, &ev);

伴随边沿触发，往往socket fd也要设置成非阻塞的。

int SetNonBlockFD(int fd) 
{
    int oldflag = fcntl(fd,F_GETFL);
    int newflag = fcntl(fd,F_SETFL, oldflag | O_NONBLOCK);
    if(newflag == -1)
        return -1;
    return oldflag;
}

边缘触发效率更高，减少了事件被重复触发的次数，
这样epoll_wait()函数返回的就绪队列epoll_event *就会更精简。

如果使用边沿模式，必须使用非阻塞 I/O。
在收到一次epoll_wait（）通知后，须循环调用 recv等函数，直到返回 EWOULDBLOCK 为止。
然后再调用 epoll_wait 等待操作系统的下一次通知。
如果在边沿模式，使用阻塞IO，虽然也可以用循环重复读取，但读取最后的数据后，IO函数会阻塞，
这就违背了使用边沿模式的初衷。

参考：
[IO复用]EPOLL 如何实现ET(边缘触发)

epoll的优缺点

select用的位图来监控设备，poll用的数组。epoll用的红黑树，没有最大连接数的上限，并且查找和删除更叫高效。
select是用户态在维护fd的集合，所以需在select()的时候把fd集合传给内核。epoll 只需要在EPOLL_CTL_ADD的时候，从用户态传递到内核态，不需要在执行epoll_wait()时的拷贝。
epoll内核态使用的回调函数，来把有事件的fd写到epoll_wait（）的返回队列中。返回队列也只有，有事件的fd，不需要像seelct一样把所有fd_set的fd都FD_ISSET遍历一遍了。

什么时候用select 或者 epoll？

在我们上面的内容中，epoll肉眼可见的要比select要好。
但是为什么说到IO复用，还是select、poll、epoll并列，而不是只有epoll呢？
因为每种IO复用方式，都有适用的范围。

用户活跃度高，连接量大不的情况下，select 优于epoll.

当连接数较多并且有很多的不活跃连接时，epoll 的效率比其它两者高很多。当连接数较少并且都十分活跃的情况下，由于 epoll 需要很多回调，因此性能可能低于其它两者。
来源：后端面试必问的I/O多路复用，这一篇就够了！

什么情况下使用Epoll：
1.你的程序通过多个线程来处理大量的网络连接。如果你的程序只是单线程的那么将会失去epoll的很多优点。并且很有可能不会比poll更好。
2.你需要监听的套接字数量非常大（至少1000）；如果监听的套接字数量很少则使用epoll不会有任何性能上的优势甚至可能还不如poll。
3.你的网络连接相对来说都是长连接；就像上面提到的epoll处理短连接的性能还不如poll因为epoll需要额外的系统调用来添加描述符到集合中。
4.你的应用程序依赖于Linux上的其他特性
来源：【性能篇】多路复用之 Select，Poll，Epoll 的差异与选择

select、poll、epoll的区别

在这里插入图片描述

来源：IO多路复用——深入浅出理解select、poll、epoll的实现

Windows中的IO复用

Windows中可以使用select 和 poll ，不能使用epoll。

Windows中不但提供了select，还提供了WSAAsyncSelect和WSAEventSelect，但他们是异步IO模型，
后面会记录异步IO。
→[IO复用] Windows select FD_SETSIZE 大小修改

Reactor模式

Reactor 模式也叫 Dispatcher 模式，即IO多路复用监听事件，收到事件后，根据事件类型分配（Dispatch）给某个进程 / 线程。
Reactor模式的思想特点在于，不是应用程序主动调用函数并等待函数执行完成，而是应用程序把回调函数和事件注册在Reactor中，当有事件发生时，自动执行对应的回调函数。

Reactor分为：单Reactor单线程、单Reactor多线程、主从Reactor多线程，3种。
这三种的差异和解释可以看：
Reactor模型

我写过单Reactor多线程模式，我用我的理解，来描述一下：
这个模式主要分为selector、accepter、handler、processor模块。
selector核心是select()，用于监听fd。当监听到listen的fd有事件时，分发给accepter处理。
accepter 中accept到新的fd后，把fd加入fd_set继续监听。
当reactor监听到listen外的fd有事件时，分发给handler模块处理，handler会把相关信息和回调函数，传入线程池处理。

C10K问题

即单个服务器进程如何处理10K个并发连接。
这个问题是早期互联网面临的问题，那时候后并发连接比较少，往往是一连接一线程的处理方式。
解决C10K的方案就是IO复用，尤其是Epoll的出现。之后，linux有Epoll ，windos 有IOCP，可以实现高并发的处理。

C10M问题

C10M问题是C10K问题的升级，到了C10M这个程度，制约程序的是内核进行了太多的切换和调度，要想办法避免线程的切换和调度，就提出了协程的概念。
关于C10K和C10M问题，可以参考：
C10k问题简述

面试题

1、在epoll IO多路复用中，某个socket读到一半，在这个socket上又有读事件来了怎么办？
答：为了避免在同一个socket上再次监听到同一个可读事件，可以在对应的描述符中添加 EPOLL_ONESHOT事件。
其效果是监听到一次事件后就将对应的描述符从监听集合中移除，也就不会再被追踪到了。读操作完成后再把对应的文件描述符重新加入监听集合。
作者：linux
链接：https://www.zhihu.com/question/24200063/answer/2991495235
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

2、LT和ET模式下的阻塞与非阻塞？
答：在LT（水平触发）模式下，也是epoll的默认模式，epoll_wait返回可读事件，表明socket一定收到了数据，我们可以使用read函数来读取数据。如果指定读取的数据大于缓冲区数据，无论socket是阻塞还是非阻塞，read函数不会阻塞，会返回实际读取到的数据大小。在read之后再次调用read，如果socket是阻塞的，read将阻塞，直到接收到数据才返回。此时，如果指定读取的数据小于缓冲区中数据，epoll_wait 会继续被触发，因为还有读缓冲区中还有数据没有被读取完。
在ET（边缘触发）模式下，只有新的数据到来时才会触发。如果指定读取的数据小于缓冲区中的数据，epoll_wait 不会被继续触发。因此，使用ET模式时，有数据到来时，必须循环读取读缓冲区中的数据，直到read返回-1，并且errno错误码为EAGAIN，才算读取完了全部缓冲区中的内容。
对于监听的listen_fd，最好使用LT模式，如果使用ET模式会导致高并发情况下，有的客户端会连接不上。如果非要使用ET模式，可以在while循环中调用accept()函数。
对于读写的conn_fd，LT模式下，阻塞和非阻塞效果都一样，因为在阻塞模式下，如果数据读取不完全则返回继续触发，反之读取完则返回继续等待。建议将文件描述符设置为非阻塞。
对于读写的conn_fd，ET模式下，必须使用非阻塞IO，并要求一次性地完整读写完全部数据。因为如果不一次性读取完缓冲区中的全部数据，缓冲区剩余数据不会被 epoll_wait 再次触发。
作者：linux
链接：https://www.zhihu.com/question/24200063/answer/2991495235
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。