socket编程2：IO复用（select ＆＆ poll ＆＆ epoll）

IO多路复用

IO多路复用是指内核一旦发现进程指定的一个或者多个IO条件准备读取，它就通知该进程。

我们假设这样一个场景去理解复用，假设目前有一个城堡，该城堡前有一个多叉路口，随时都会有从某个岔路出发前往城堡取各自工具的村民，为了能让村民更快地取得工具，城堡主人计划在叉路口蹲守多个守卫来为前往的村民处理事宜，但这样需要的守卫数目前极大，所以主人想到一个办法，让侍卫长蹲守在分叉路汇总的地方，每来一个村民侍卫长直接通知某个侍卫在城堡取得相应工具送还给村民。

用更加专业的语言来解释多路复用：

I/O多路复用就是通过一种机制，一个进程可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知程序进行相应的读写操作。

为什么需要IO复用

IO复用本质上是为了实现同步，若是没有复用机制，我们可以使用阻塞IO来实现同步，若没有数据，等待数据的线程就会被挂起，直到收到数据。

当服务器处理1000个连接，但是只有很少连接执行IO操作，那么需要1000个线程或进程来处理1000个连接，而1000个线程大部分是被挂起的，而且为了实现并发，线程切换即那会十分频繁。

也就是说，在服务器高并发的需求下，它不仅需要同时创建数量庞大的线程，还需要进行频繁的线程切换。

而线程是有内存开销的，1个线程可能需要512K（或2M）存放栈，那么1000个线程就要512M（或2G）内存。线程的切换，或者说上下文切换是有CPU开销的，当大量时间花在上下文切换的时候，分配给真正的操作的CPU就要少很多。

这显然都很浪费资源，所以才需要进行IO复用，在处理1000个连接时，只需要1个线程监控就绪状态，就绪的连接开一个线程处理就可以了，这样需要的线程数大大减少，减少了内存开销和上下文切换的CPU开销。

IO多路复用主要分为3类：select、poll以及epoll，具体相关内容如下文。

select

select目前几乎在所有的平台上支持，其良好跨平台支持也是它的一个优点。

函数定义

我们首先直接了解一下select函数定义：
int select(int maxfd,fd_set *rdset,fd_set *wrset,fd_set *exset,struct timeval *timeout)

其中的参数意义如下：

返回值意义如下：

select函数调用后会阻塞，直到有描述符就绪（有数据 可读、可写、或者有except），或者超时（timeout指定等待时间，如果立即返回设为null即可），函数返回。当select函数返回后，可以通过遍历select函数所监视的文件描述符集合，来找到就绪的描述符。

初始化（宏定义）

这组宏定义函数存在的主要原因是当我们设置好最初监听的描述符后，当数据到达后，这些描述符就会发生改变，所以每处理一次数据就需要重新进行初始化。

编码流程

1. 定义描述符集
2. 清空描述符集
3. 设置指定的描述符并获取最大的描述符值+1
4. 等待描述符就绪
5. 判断已就绪的描述符，并做对应处理。

select优缺点分析

select本质上是通过设置或者检查存放fd标志位的数据结构来进行下一步处理。所带来的缺点是：

1. select最大的缺陷就是单个进程所打开的FD是有一定限制的，它由FD_SETSIZE设置，默认值是1024(32位)。
2. 需要修改传入的参数数组，原因前文已经解释过啦。
3. 对socket进行扫描时是线性扫描，并不能确切指定有数据的socket，采用轮询的方法，效率较低。
4. 需要维护一个用来存放大量fd的数据结构，这样会使得用户空间和内核空间在传递该结构时复制开销大。

C++代码示例

客户端实现

int main(int argc,char* argv[]) {
    
    // 创建套接字
    int fd = socket(AF_INET,SOCK_STREAM,0);
    if(-1 == fd) {
    
        perror("socket error");
        return 1;
    }

    // 连接服务器
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = inet_addr(argv[1]);
    addr.sin_port = htons(atoi(argv[2]));
    int res = connect(fd,reinterpret_cast<sockaddr*>(&addr),sizeof(addr));
    if(-1 == res) {
    
        perror("connect error");
        return 1;
    }

    fd_set fds;
    FD_ZERO(&fds);
    FD_SET(0,&fds);
    FD_SET(fd,&fds);
    int maxfd = fd+1;

    for(;;) {
    
        select(maxfd,&fds,NULL,NULL,NULL); // 代替标准输入和连接的阻塞

        if(FD_ISSET(0,&fds)) {
    
            string msg;
            if(cin >> msg) {
     // 阻塞
                write(fd,msg.c_str(),msg.size()+1);
            } else {
    
                break;
            }
        }
        if(FD_ISSET(fd,&fds)) {
    
            char buff[200];
            int n = read(fd,buff,200); // 阻塞
            if(0 == n) break; // 对方close()
            cout << buff <<endl;
        }
        FD_SET(0,&fds);
        FD_SET(fd,&fds);
        maxfd = fd+1;
    }

    close(fd);
}

服务端实现

int main(int argc,char* argv[]) {
    
    // 创建套接字
    int listenfd = socket(AF_INET,SOCK_STREAM,0);
    if(-1 == listenfd) {
    
        perror("socket error");
        return 1;
    }

    // 绑定IP和端口
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = inet_addr(argv[1]);
    addr.sin_port = htons(atoi(argv[2]));
    int res = bind(listenfd,reinterpret_cast<sockaddr*>(&addr),sizeof(addr));
    if(-1 == res) {
    
        perror("bind error");
        return 1;
    }

    // 设定连接个数
    res = listen(listenfd,3);
    if(-1 == res) {
    
        perror("listen error");
        return 1;
    }

    // 文件描述符集合
    fd_set fds;

    // 初始化清空集合
    FD_ZERO(&fds);

    // 设置需要监听的描述符
    FD_SET(listenfd,&fds); // 监听文件描述符
    FD_SET(0,&fds); // 标准输入
    int maxfd = max(listenfd,0)+1;

    list<int> connfds;
    for(;;) {
    
        select(maxfd,&fds,NULL,NULL,NULL);// 阻塞,监听所有需要阻塞处理的fd

        if(FD_ISSET(listenfd,&fds)) {
     // 处理连接
            int connfd = accept(listenfd,NULL,NULL);
            connfds.push_back(connfd);
        }
        if(FD_ISSET(0,&fds)) {
     // 处理标准输入
            string msg;
            cin >> msg;
            for(int connfd:connfds) {
    
                write(connfd,msg.c_str(),msg.size()+1);
            }
        }
        for(int connfd:connfds) {
    
            if(FD_ISSET(connfd,&fds)) {
    
                char buff[200];
                int n = read(connfd,buff,200); // 阻塞
                // if(0 == n){ // 客户端断开连接
                cout << buff <<endl;
            }
        }
        // 重新注册fd
        FD_SET(listenfd,&fds); // 监听文件描述符
        FD_SET(0,&fds); // 标准输入
        for(int connfd:connfds) {
    
            FD_SET(connfd,&fds);
        }
        maxfd = max(listenfd,
                        *max_element(connfds.begin(),connfds.end()) // 找出链表中最大的fd
                       )+1;
    }

    close(listenfd);
}

poll

poll本质上和select没有区别，它将用户传入的数组拷贝到内核空间，然后查询每个fd对应的设备状态，如果设备就绪则在设备等待队列中加入一项并继续遍历，如果遍历完所有fd后没有发现就绪设备，则挂起当前进程，直到设备就绪或者主动超时，被唤醒后它又要再次遍历fd。

函数定义

poll函数的定义如下：
int poll(struct pollfd *fdarray, unsigned long nfds, int timeout)

参数的意义如下：

返回值意义：

其中的struct poll结构体中的成员为：

• fd:描述符
• events:监听事件，主要用于设置监听事件，监视事件可以是输入事件，可以是输出事件。
• revents:实际触发的事件，用于判断触发的事件，监视事件可以是输入事件，可以是输出事件，还可以是出错事件。

首先我们需要知道我们一般将正规的TCP数据以及所有的UDP数据称作普通数据，所有的UDP带外数据称为优先级数据。接下来我们就可以解释以上所谈及的

fd输入事件：

fd输出事件：

fd出错事件：

编码流程

1. 定义pollfd结构体数组
2. 初始化pollfd结构体数组
3. 设置监听poll事件
4. 等待poll事件
5. 判断触发事件的描述符，并做对应处理。

poll优缺点分析

poll没有最大连接数的限制，原因是它是基于链表来存储的，它的缺点如下：

1. 大量的fd的数组被整体复制于用户态和内核地址空间之间(这下复制不一定都是有用的)；
2. poll还有一个特点是“水平触发”，如果报告了fd后，没有被处理，那么下次poll时会再次报告该fd。
3. 和select都是通过遍历文件描述符来获取已经就绪的socket。事实上，同时连接的大量客户端在一时刻可能只有很少的处于就绪状态，因此随着监视的描述符数量的增长，其效率也会线性下降。

C++代码示例

客户端实现

#include <iostream>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <poll.h>
using namespace std;

int main(int argc,char* argv[]) {
    
    // 创建套接字
    int fd = socket(AF_INET,SOCK_STREAM,0);
    if(-1 == fd) {
    
        perror("socket error");
        return 1;
    }

    // 连接服务器
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = inet_addr(argv[1]);
    addr.sin_port = htons(atoi(argv[2]));
    int res = connect(fd,reinterpret_cast<sockaddr*>(&addr),sizeof(addr));
    if(-1 == res) {
    
        perror("connect error");
        return 1;
    }

    struct pollfd pollfds[2];
    pollfds[0].fd = 0;
    pollfds[0].events = POLLIN; // 监听事件：读入
    pollfds[1].fd = fd;
    pollfds[1].events = POLLIN;
    for(;;){
    
        poll(pollfds, 2, -1);	  // 2指的是pollfds第几个
	                          // -1 表示一直等待
	if(pollfds[0].revents&POLLIN){
       // poll的优势就是将输入和输出分离开，从而避免了select使用时要进行初始化的缺点
	    string msg;
	    if(cin >> msg){
    
	        write(fd, msg.c_str(), msg.size()+1);
	    }else{
    
	        break;
	    }
	}
	if(pollfds[1].revents&POLLIN){
    
	    char buff[200];
	    int n = read(fd, buff, 200);
	    if(0 == n){
    
	        break;
	    }else{
    
	        cout << buff << endl;
	    }
	}
    }

    close(fd);
}

服务端实现

#include <iostream>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <thread>
#include <vector>
#include <map>
#include <list>
#include <algorithm>
#include <poll.h>
#include <cstring>
using namespace std;

int main(int argc,char* argv[]) {
    
    // 创建套接字
    int listenfd = socket(AF_INET,SOCK_STREAM,0);
    if(-1 == listenfd) {
    
        perror("socket error");
        return 1;
    }

    // 绑定IP和端口
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = inet_addr(argv[1]);
    addr.sin_port = htons(atoi(argv[2]));
    int res = bind(listenfd,reinterpret_cast<sockaddr*>(&addr),sizeof(addr));
    if(-1 == res) {
    
        perror("bind error");
        return 1;
    }

    // 设定连接个数
    res = listen(listenfd,3);
    if(-1 == res) {
    
        perror("listen error");
        return 1;
    }
    

    vector<pollfd> fds;
    pollfd fdstdin;
    fdstdin.fd = 0;
    fdstdin.events = POLLIN;
    fds.push_back(fdstdin);

    pollfd fdlisten;
    fdlisten.fd = listenfd;
    fdlisten.events = POLLIN;
    fds.push_back(fdlisten);

    for(;;){
    
        poll(fds.data(), fds.size(), -1);
	for(auto fd:fds){
    
	    if(fd.fd == 0 && fd.revents == POLLIN){
    
	        string msg;
		if(cin >> msg){
    
	            for(int i=2; i<fds.size(); i++)
		        write(fds[i].fd, msg.c_str(), msg.size()+1);
		}else{
    
		    close(fd.fd);
		}
	    }
	    if(fd.fd == listenfd && fd.revents == POLLIN){
    
	        int connfd = accept(listenfd, NULL, NULL);
		pollfd fdconn;
		fdconn.fd = connfd;
		fdconn.events = POLLIN;
		fds.push_back(fdconn);
	    }
	    if(fd.fd != 0 &&fd.fd != listenfd && fd.revents == POLLIN){
    
	        char buff[200];
		int n = read(fd.fd, buff, 200);
		if(n == 0){
    
		    break;
		}else{
    
		    cout << buff << endl;
		    for(int i=2; i<fds.size(); ++i){
    
		        if(fds[i].fd != fd.fd){
    
			    write(fds[i].fd, buff, strlen(buff)+1);
			}
		    }
		}
	    }
	}
    }

    close(listenfd);
}

epoll

epoll是之前的select和poll的增强版本。相对于select和poll来说，epoll更加灵活，没有描述符限制。epoll使用一个文件描述符管理多个描述符，将用户关系的文件描述符的事件存放到内核的一个事件表中，这样在用户空间和内核空间的copy只需一次。

函数定义

创建epoll：

int epoll_create(int size)，其中size为监听的数目;

控制epoll

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

参数及意义：

【操作】具体：

【返回值】具体如下：

轮询I/O事件

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

参数如下：

返回值意义：

其中我们提到epoll使用的结构体epoll_event有两个成员：

• fd：描述符
• events：设置/获取epoll事件

具体定义如下：

struct epoll_event {
    
     __uint32_t events;      /* epoll event */
     epoll_data_t data;      /* User data variable */
};

其中的data类型具体如下：

    typedef union epoll_data {
    
        void *ptr;
         int fd;
         __uint32_t u32;
         __uint64_t u64;
     } epoll_data_t;//保存触发事件的某个文件描述符相关的数据

其中events表示感兴趣的事件和被触发的事件，可能的取值为：

编码流程

1. 创建epoll描述符
2. 注册epoll事件
3. 等待epoll事件
4. 判断触发epoll事件的描述符和事件
5. 关闭epoll描述符

epoll优缺点分析

epoll是通过内核与用户空间mmap同一块内存实现的。mmap将用户空间的一块地址和内核空间的一块地址同时映射到相同的一块物理内存地址（不管是用户空间还是内核空间都是虚拟地址，最终要通过地址映射映射到物理地址），使得这块物理内存对内核和对用户均可见，减少用户态和内核态之间的数据交换。内核可以直接看到epoll监听的句柄，效率高。

红黑树将存储epoll所监听的套接字。上面mmap出来的内存如何保存epoll所监听的套接字，必然也得有一套数据结构，epoll在实现上采用红黑树去存储所有套接字，当添加或者删除一个套接字时（epoll_ctl），都在红黑树上去处理，红黑树本身插入和删除性能比较好，时间复杂度O(logN)。

而通过epoll_ctl函数添加进来的事件都会被放在红黑树的某个节点内，所以，重复添加是没有用的。当把事件添加进来的时候时候会完成关键的一步，那就是该事件都会与相应的设备（网卡）驱动程序建立回调关系，当相应的事件发生后，就会调用这个回调函数，该回调函数在内核中被称为：ep_poll_callback,这个回调函数其实就所把这个事件添加到rdllist这个双向链表中。**一旦有事件发生，epoll就会将该事件添加到双向链表中。**那么当我们调用epoll_wait时，epoll_wait只需要检查rdlist双向链表中是否有存在注册的事件，效率非常可观。这里也需要将发生了的事件复制到用户态内存中即可。

优点：

1. 没有最大并发连接的限制，能打开的FD的上限远大于1024（1G的内存上能监听约10万个端口）。
2. 效率提升，不是轮询的方式，不会随着FD数目的增加效率下降。只有活跃可用的FD才会调用callback函数；即Epoll最大的优点就在于它只管你“活跃”的连接，而跟连接总数无关，因此在实际的网络环境中，Epoll的效率就会远远高于select和poll。
3. 内存拷贝，利用mmap()文件映射内存加速与内核空间的消息传递；即epoll使用mmap减少复制开销。

C++代码示例

客户端实现

#include <iostream>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/epoll.h>
using namespace std;

int main(int argc,char* argv[]) {
    
    // 创建套接字
    int fd = socket(AF_INET,SOCK_STREAM,0);
    if(-1 == fd) {
    
        perror("socket error");
        return 1;
    }

    // 连接服务器
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = inet_addr(argv[1]);
    addr.sin_port = htons(atoi(argv[2]));
    int res = connect(fd,reinterpret_cast<sockaddr*>(&addr),sizeof(addr));
    if(-1 == res) {
    
        perror("connect error");
        return 1;
    }

    // select使用数组的方式来解决
    // poll使用结构体数组的方式解决
    // epoll使用面向对象的方式解决

    int epollfd = epoll_create(1024);
    epoll_event evstd;
    evstd.data.fd = 0;
    evstd.events = EPOLLIN;
    epoll_ctl(epollfd, EPOLL_CTL_ADD, 0, &evstd);
    epoll_event evconn;
    evconn.data.fd = fd;
    evconn.events = EPOLLIN;
    epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &evconn);

    for(;;){
    
        epoll_event retev[2];  // 接受返回值
	int retcnt = epoll_wait(epollfd, retev, 2, -1);
	for(int i=0; i<retcnt; ++i){
    
	    if(retev[i].data.fd == 0 && retev[i].events&EPOLLIN) {
    
	        string msg;
		if(cin >> msg){
    
		    write(fd, msg.c_str(), msg.size()+1);
		}else{
    
		    break;
		}
	    }
	    if(retev[i].data.fd == fd && retev[i].events&EPOLLIN) {
    
	        char buff[200];
		int n = read(fd, buff, 200);
		if(0 == n){
    
		    break;
		}else{
    
		    cout << buff << endl;
		}
	    }
	}
    }

    close(fd);
}

服务端实现

#include <iostream>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <thread>
#include <vector>
#include <map>
#include <list>
#include <algorithm>
#include <sys/epoll.h>
#include <cstring>
using namespace std;

int main(int argc,char* argv[]) {
    
    // 创建套接字
    int listenfd = socket(AF_INET,SOCK_STREAM,0);
    if(-1 == listenfd) {
    
        perror("socket error");
        return 1;
    }

    // 绑定IP和端口
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = inet_addr(argv[1]);
    addr.sin_port = htons(atoi(argv[2]));
    int res = bind(listenfd,reinterpret_cast<sockaddr*>(&addr),sizeof(addr));
    if(-1 == res) {
    
        perror("bind error");
        return 1;
    }

    // 设定连接个数
    res = listen(listenfd,3);
    if(-1 == res) {
    
        perror("listen error");
        return 1;
    }
    
    int epollfd = epoll_create(1024);

    epoll_event evstd;
    evstd.data.fd = 0;
    evstd.events = EPOLLIN;
    epoll_ctl(epollfd, EPOLL_CTL_ADD, 0, &evstd);

    epoll_event evlisten;
    evlisten.data.fd = listenfd;
    evlisten.events = EPOLLIN;
    epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &evlisten);

    vector<int> fds;
    int n = 2;   // 目前epoll里面有两个fd
    for(;;){
    
        epoll_event retevs[n];
	int revcnt = epoll_wait(epollfd, retevs, n, -1);
	for(int i=0; i<revcnt; ++i){
    
	    if(retevs[i].data.fd == 0 && retevs[i].events&EPOLLIN) {
    
	        string msg;
		if(cin >> msg){
    
	            for(int fd:fds){
    
		         write(fd, msg.c_str(), msg.size()+1);
		    }
		}else{
    
		    break;
		}
	    }
	    if(retevs[i].data.fd == listenfd && retevs[i].events&EPOLLIN) {
    
	        int connfd = accept(listenfd, NULL, NULL);
		epoll_event evconn;
		evconn.data.fd = connfd;
		evconn.events = EPOLLIN;
		epoll_ctl(epollfd, EPOLL_CTL_ADD, connfd, &evconn);
		n++;
		fds.push_back(connfd);

	    }
	    if(0!=retevs[i].data.fd && listenfd!=retevs[i].data.fd && retevs[i].events&EPOLLIN) {
    
	        char buff[200];
		int n = read(retevs[i].data.fd, buff,200);
		if(0 == n){
    
		    break;
		}else{
    
	  	    cout << buff << endl;
	            for(int fd:fds){
    
		         write(fd, buff, strlen(buff)+1);
		    }
		
		}
	    }
	}
    }

    close(listenfd);
}

select、poll、epoll的区别