02 IO多路转接

小鱼儿 2022-03-21 15:14 146阅读 0赞

多路IO转接服务器也叫做多任务IO服务器。该类服务器实现的**主旨思想：不再由应用程序自己监视客户端连接，取而代之由内核替应用程序监视文件。**

## 一. select ##

**1. select函数介绍：**

#include <sys/select.h>
    /* According to earlier standards */
    #include <sys/time.h>
    #include <sys/types.h>
    #include <unistd.h>
    
    int select(int nfds, fd_set *readfds, fd_set *writefds, 
            fd_set *exceptfds, struct timeval *timeout);

nfds: 监控的文件描述符集里最大文件描述符加1，因为此参数会告诉内核检测前多少个文件描述符的状态。  
readfds： 监控读数据到达文件描述符集合，传入传出参数  
writefds： 监控写数据到达文件描述符集合，传入传出参数  
exceptfds： 监控异常发生到达文件描述符集合，如带外数据到达异常，传入传出参数  
timeout： 定时阻塞监控时间，3种情况  
1.NULL，永远等下去  
2.设置timeval，等待固定时间  
3.设置timeval里时间均为0，检查描述字后立即返回，轮询

--------------------

**2. select内核操作原理：**

内核中使用位图机制来实现数据的集合，如果将对应位设置为1，表示通知内核监视对应的文件描述符。然而定义的位图并不允许我们直接使用位操作进行修改，系统专门为此提供了一套函数进行操作。

void FD_CLR(int fd, fd_set *set);   //把文件描述符集合里fd位清0
    int FD_ISSET(int fd, fd_set *set);  //测试文件描述符集合里fd是否置1
    void FD_SET(int fd, fd_set *set);   //把文件描述符集合里fd位置1
    void FD_ZERO(fd_set *set);          //把文件描述符集合里所有位清0

socket -> setsockopt -> bind -> listen -> while(  
temps = reads;  
//1. temp是一个传入传出参数，传入需要监听的文件描述符，传出有读事件的文件描述符。  
//2. 所以此处需要设置一个临时变量，以保留设置的需要监听的文件描述符的集合。  
//3. 如果需要监听写事件，那么可以设置select的参数三，将需要监听写事件的文件描述符添加的该集合中。  
ret = select(maxfd+1, &temps, NULL, NULL, NULL);//  
if (FD\_ISSET(lfd, &temps)) \{ ... \} //判断是否有新连接。有则添加到reads数组中。  
for(i=lfd+1; i<=maxfd; ++i) \{ ... \} //遍历检测的文件描述符是否有读操作。  
);  
因为系统返回的文件描述符均符合当前系统**最小可用**，所以文件描述符本身的数值即为位图中的编号。

**3. select 总结：**

优点：  
1. 跨平台。  
缺点：  
1. 调用select ，集合拷贝：用户态 -> 内核态 -> 用户态。  
2. 遍历文件描述符的集合。  
3. 支持的文件描述符少，只有1024个。

#include <stdio.h>
    #include <unistd.h>
    #include <stdlib.h>
    #include <sys/types.h>
    #include <string.h>
    #include <sys/socket.h>
    #include <arpa/inet.h>
    #include <ctype.h>
    #include <sys/select.h>
    
    #define SERV_PORT 8989
    
    int main(int argc, const char* argv[])
    {
        int lfd, cfd;
        struct sockaddr_in serv_addr, clien_addr;
        int serv_len, clien_len;
    
        // 创建套接字
        lfd = socket(AF_INET, SOCK_STREAM, 0);
        // 初始化服务器 sockaddr_in 
        memset(&serv_addr, 0, sizeof(serv_addr));
        serv_addr.sin_family = AF_INET;                   // 地址族 
        serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);    // 监听本机所有的IP
        serv_addr.sin_port = htons(SERV_PORT);            // 设置端口 
        serv_len = sizeof(serv_addr);
        // 绑定IP和端口
        bind(lfd, (struct sockaddr*)&serv_addr, serv_len);
    
        // 设置同时监听的最大个数
        listen(lfd, 36);
        printf("Start accept ......\n");
    
        int ret;
        int maxfd = lfd;
        // reads 实时更新，temps　内核检测
        fd_set reads, temps;
        
        /*===============================================================*/
        // 记录要检测的文件描述符的数组
        int allfd[FD_SETSIZE];   // 1024
        // 记录数组中最后一个元素的下标
        int last_index = 0;
        // 初始化数组
        for(int i=0; i<FD_SETSIZE; ++i)
        {
            allfd[i] = -1;  // 无效文件描述符值
        }
        allfd[0] = lfd;     // 监听的文件描述符添加到数组中
        /*===============================================================*/
    
        // 初始化监听的读集合
        FD_ZERO(&reads);
        FD_SET(lfd, &reads);
    
        while(1)
        {
            // 每次都需要更新，否则select不会重新检测
            temps = reads;
            ret = select(maxfd+1, &temps, NULL, NULL, NULL);
            if(ret == -1)
            {
                perror("select error");
                exit(1);
            }
    
            int i = 0;
            char bufip[64];
            // 判断是否有新连接
            if(FD_ISSET(lfd, &temps))
            {
                // 接受连接请求
                clien_len = sizeof(clien_len);
                int cfd = accept(lfd, (struct sockaddr*)&clien_addr, &clien_len);
                printf("client ip: %s, port: %d\n",
                       inet_ntop(AF_INET, &clien_addr.sin_addr.s_addr, bufip, sizeof(bufip)),
                       ntohs(clien_addr.sin_port));
    
                // 文件描述符放入检测集合
                FD_SET(cfd, &reads);
                // 更新最大文件描述符
                maxfd = maxfd < cfd ? cfd : maxfd;
                // cfd添加到检测数组中
                for(i=0; i<FD_SETSIZE; ++i)
                {
                    if(allfd[i] == -1)
                    {
                        allfd[i] = cfd;
                        break;
                    }
                }
                // 更新数组最后一个有效值下标
                last_index = last_index < i ? i : last_index; 
            }
    
            // 遍历检测的文件描述符是否有读操作
            for(i=lfd+1; i<=maxfd; ++i)
            {
                if(FD_ISSET(i, &temps))
                {
                    // 读数据
                    char buf[1024] = {0};
                    int len = read(i, buf, sizeof(buf));
                    if(len  == -1)
                    {
                        perror("read error");
                        exit(1);
                    }
                    else if(len == 0)
                    {
                        // 对方关闭了连接
                        FD_CLR(i, &reads);
                        close(i);
                        if(maxfd == i)
                        {
                            maxfd--;
                        }
                        allfd[i] = -1;
                        printf("对方已经关闭了连接。。。。。。\n");
                    }
                    else
                    {
                        printf("read buf = %s\n", buf);
                        for(int j=0; j<len; ++j)
                        {
                            buf[j] = toupper(buf[j]);
                        }
                        printf("--buf toupper: %s\n", buf);
                        write(i, buf, strlen(buf)+1);
                    }
                }
            }
        }
    
        close(lfd);
        return 0;
    }

##  ##

## 二. pselect ##

#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);

pselect 函数是一个防止信号干扰的增强型select 函数，pselect从形式向看相对于select多了一个参数，而具体区别如下：

1. pselect 使用timespec 结构，能指定到纳秒级(旧结构只能指定到微秒级)  
2. pselect 增加了指向信号集的指针sigmask，表示信号屏蔽集。若sigmask为空，那么在与信号有关的方面，pselect的运行状况和select相同；否则，sigmask指向一信号屏蔽字，**在调用pselect时，以原子操作的方式安装该信号屏蔽字。在返回时恢复以前的信号屏蔽字。**

#include <stdio.h>
    #include <string.h>
    #include <stdlib.h>
    #include <sys/types.h>
    #include <sys/stat.h>
    #include <errno.h>
    #include <fcntl.h>
    #include <unistd.h>
    #include <sys/socket.h>
    #include <arpa/inet.h>
    #include <sys/select.h>
    
    #define SIZE 128
    
    //服务端  select
    int main(void)
    {
        int ret = -1;
        int sockfd = -1;
        int connfd = -1;
    
        char buf[SIZE];
    
        struct sockaddr_in addr;
        struct sockaddr_in from;
    
        socklen_t len = sizeof(from);
    
        //select相关的参数
        int maxfd = -1;
        fd_set readfds;
        struct timespec tmo;
    
        //1. 创建套接字
        sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if (-1 == sockfd)
        {
            perror("sockfd"); 
            goto err0;
        }
    
        //2. 绑定
        //指定IP和端口
        memset(&addr, 0, sizeof(addr)); 
        addr.sin_family = AF_INET;      //指定协议族
        addr.sin_port = htons(10086);   //指定端口
        inet_pton(AF_INET, "192.168.73.42", (void *)&addr.sin_addr);
        
        ret = bind(sockfd, (void*)&addr, sizeof(addr));
        if (-1 == ret)
        {
            perror("bind"); 
            goto err1;
        }
        
        //3. 监听
        ret = listen(sockfd, 10); 
        if (-1 == ret)
        {
            perror("listen"); 
            goto err1;
        }
        printf("服务端正在监听客户端的连接...\n");
    
        //4. 接受客户端的连接
        connfd = accept(sockfd, (void*)&from, &len);
        if (-1 == connfd)
        {
            perror("accept");
            goto err1;
        }
    
        //输出连接服务端的客户端的IP和Port
        printf("\033[32mclient IP: %s port: %d\033[0m\n", inet_ntop(AF_INET, (void*)&from.sin_addr, buf, SIZE), ntohs(from.sin_port));
    
        //5. 循环的接收和发送数据
        while(1)
        {
            //设置超时的时间
            tmo.tv_sec = 3;
            tmo.tv_nsec = 0;
    
            //清空读集合
            FD_ZERO(&readfds); 
            //将标准输入和套接字加入到监听读集合中
            FD_SET(STDIN_FILENO, &readfds);  
            FD_SET(connfd, &readfds);
    
            //比较文件描述符
            maxfd = STDIN_FILENO > connfd ? STDIN_FILENO : connfd;
    
            //由内核监视对应的文件描述符
            ret = pselect(maxfd + 1, &readfds, NULL, NULL, &tmo, NULL);
            if (-1 == ret)
            {
                perror("pselect"); 
                break;
            }
            else if (0 == ret)
            {
                printf("3 seconds timeout....\n"); 
                continue;
            }
            else
            {
                //有数据可以读 标准输入有数据
                if (FD_ISSET(STDIN_FILENO, &readfds)) 
                {
                    memset(buf, 0, SIZE);
                    fgets(buf, SIZE, stdin);
                    //去掉最后一个换行
                    if ('\n' == buf[strlen(buf) - 1])
                        buf[strlen(buf) - 1] = '\0';
                
                    //发送数据
                    ret = send(connfd, buf, strlen(buf), 0);
                    if (ret <= 0)
                    {
                        perror("send"); 
                        break;
                    }
                    printf("server send %d bytes\n", ret);
                }
    
                //有数据可以读  套接字上有数据
                if (FD_ISSET(connfd, &readfds)) 
                {
                    memset(buf, 0, SIZE);
                    ret = recv(connfd, buf, SIZE, 0);  
                    if (ret <= 0)
                    {
                        perror("recv"); 
                        break;
                    }
                    printf("\033[31mrecv: %s\033[0m\n", buf);
                }
            }
        }
    
        //6. 关闭连接
    
        close(sockfd);
        close(connfd);
    
        return 0;
    err1:
        close(sockfd);
    err0:
        return 1;
    }

###  ###

### 三. poll ###

int poll(struct pollfd *fd, nfds_t nfds, int timeout);
    struct pollfd {
    	int   fd;         /* 文件描述符 */
    	short events;     /* 等待的事件 */
    	short revents;    /* 实际发生的事件 */
    };

**1. poll参数介绍：**

pollfd：数组的地址  
nfds：数组的最大长度, 数组中最后一个使用的元素下标+1  
1. 内核会轮询检测fd数组的每个文件描述符  
timeout:  
1. == -1: 永久阻塞  
2. == 0: 调用完成立即返回  
3. >0: 等待的时长毫秒

**2. poll评价：**

**优点：**

1. 传入、传出事件分离。无需每次调用时，重新设定监听事件。  
2. 文件描述符上限，可突破1024限制。能监控的最大上限数可使用配置文件调整。

缺点：(同select)

1. 每次**调用**都会出现一次从用户空间到内核空间的拷贝。  
2. 每次**返回**都会出现一次从内核空间到用户空间的拷贝。  
3. 返回后需要用户依次扫描fds数组，因此会做很多没必要的检查。  
4. 不能跨平台。

#include <stdio.h>
    #include <unistd.h>
    #include <stdlib.h>
    #include <sys/types.h>
    #include <string.h>
    #include <sys/socket.h>
    #include <arpa/inet.h>
    #include <ctype.h>
    #include <poll.h>
    
    #define SERV_PORT 8989
    
    int main(int argc, const char* argv[])
    {
        int lfd, cfd;
        struct sockaddr_in serv_addr, clien_addr;
        int serv_len, clien_len;
    
        // 创建套接字
        lfd = socket(AF_INET, SOCK_STREAM, 0);
        // 初始化服务器 sockaddr_in 
        memset(&serv_addr, 0, sizeof(serv_addr));
        serv_addr.sin_family = AF_INET;                   // 地址族 
        serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);    // 监听本机所有的IP
        serv_addr.sin_port = htons(SERV_PORT);            // 设置端口 
        serv_len = sizeof(serv_addr);
        // 绑定IP和端口
        bind(lfd, (struct sockaddr*)&serv_addr, serv_len);
    
        // 设置同时监听的最大个数
        listen(lfd, 36);
        printf("Start accept ......\n");
    
        // poll结构体
        struct pollfd allfd[1024];
        int max_index = 0;
        // init
        for(int i=0; i<1024; ++i)
        {
            allfd[i].fd = -1;
            allfd[i].events = POLLIN;
        }
        allfd[0].fd = lfd;
    
        while(1)
        {
            int i = 0;
            int ret = poll(allfd, max_index+1, -1); 
            if(ret == -1)
            {
                perror("poll error");
                exit(1);
            }
    
            // 判断是否有连接请求
            if(allfd[0].revents & POLLIN)
            {
                clien_len = sizeof(clien_addr);
                // 接受连接请求
                int cfd = accept(lfd, (struct sockaddr*)&clien_addr, &clien_len);
                printf("============\n");
    
                // cfd添加到ｐｏｌｌ数组
                for(i=0; i<1024; ++i)
                {
                    if(allfd[i].fd == -1)
                    {
                        allfd[i].fd = cfd;
                        break;
                    }
                }
                // 更新最后一个元素的下标
                max_index = max_index < i ? i : max_index;
            }
    
            // 遍历数组
            for(i=1; i<=max_index; ++i)
            {
                int fd = allfd[i].fd;
                if(fd == -1)
                {
                    continue;
                }
                if(allfd[i].revents & POLLIN)
                {
                    // 接受数据
                    char buf[1024] = {0};
                    int len = recv(fd, buf, sizeof(buf), 0);
                    if(len == -1)
                    {
                        perror("recv error");
                        exit(1);
                    }
                    else if(len == 0)
                    {
                        allfd[i].fd = -1;
                        close(fd);
                        printf("客户端已经主动断开连接。。。\n");
                    }
                    else
                    {
                        printf("recv buf = %s\n", buf);
                        for(int k=0; k<len; ++k)
                        {
                            buf[k] = toupper(buf[k]);
                        }
                        printf("buf toupper: %s\n", buf);
                        send(fd, buf, strlen(buf)+1, 0);
                    }
    
                }
    
            }
        }
    
        close(lfd);
        return 0;
    }

### 四. epoll ###

epoll可以显著提高在**大量并发连接中只有少量活跃**的情况下的系统CPU利用率，它会复用文件描述符集合，内部维护了红黑树，只需要遍历被内核IO事件异步唤醒而加入的文件描述符集合即可。

**1. epoll操作函数：**与其他IO复用不同的一点，epoll需要三个操作函数来完成。

1. 创建一个红黑树的根节点：  
int epoll\_create(int size);  
size参数用来告诉内核监听的数目，但这仅仅是一个建议，如今的系统中只要大于0即可。

2. 事件注册函数：  
int epoll\_ctl(int epfd, int op, int fd, struct epoll\_event \*event);

// 感兴趣的事件和被触发的事件
        struct epoll_event {
            __uint32_t events; /* Epoll events */
            epoll_data_t data; /* User data variable */
        };
    
        // 保存触发事件的某个文件描述符相关的数据（与具体使用方式有关）
        typedef union epoll_data {
            void *ptr;
            int fd;
            __uint32_t u32;
            __uint64_t u64;
        } epoll_data_t;

op：表示动作，用三个宏来表示：  
EPOLL\_CTL\_ADD：注册新的 fd 到 epfd 中；  
EPOLL\_CTL\_MOD：修改已经注册的fd的监听事件；  
EPOLL\_CTL\_DEL：从 epfd 中删除一个 fd；  
fd：需要监听的文件描述符。  
event：告诉内核要监听什么事件，struct epoll\_event 结构。

events 可以是以下几个宏的集合：需要使用位或( | )进行操作。  
EPOLLIN ：表示对应的文件描述符可以读（包括对端 SOCKET 正常关闭）；  
EPOLLOUT：表示对应的文件描述符可以写；  
EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；  
EPOLLERR：表示对应的文件描述符发生错误；  
EPOLLHUP：表示对应的文件描述符被挂断；  
EPOLLET ：将 EPOLL 设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的。

3. 等待事件的产生  
int epoll\_wait(int epfd, struct epoll\_event \* events, int maxevents, int timeout );  
1. epfd: epoll 专用的文件描述符，epoll\_create()的返回值  
2. events: 分配好的 epoll\_event 结构体数组，epoll 将会把发生的事件赋值到events 数组中(**events 不可以是空指针，内核只负责把数据复制到这个 events 数组中，不会去帮助我们在用户态中分配内存**)。  
3. maxevents: maxevents 告诉内核这个 events 有多大 。  
4. timeout: 超时时间，单位为毫秒，为 -1 时，函数为阻塞。

**2. epoll的应用：**epoll默认为边沿触发，当没有即使处理事件时会继续触发，所以为了**减少触发的次数以提高效率**，应该使用水平触发，使用循环读取一次性读完所有数据。但是数据读完就进程就会阻塞，所以需要将与客户端进行通讯的文件描述符设置为非阻塞。即最终选定的模式为epoll非阻塞边沿触发，**触发一次，读完所有数据**。

#include <stdio.h>
    #include <string.h>
    #include <stdlib.h>
    #include <sys/types.h>
    #include <sys/stat.h>
    #include <errno.h>
    #include <fcntl.h>
    #include <unistd.h>
    #include <sys/socket.h>
    #include <arpa/inet.h>
    #include <sys/select.h>
    #include <sys/epoll.h>
    #include <fcntl.h>
    
    #define SIZE 128
    #define SIZE1 6
    
    //服务端  select
    int main(void)
    {
        int ret = -1;
        int sockfd = -1;
        int connfd = -1;
        int i = 0;
        int op = 0;
    
        char buf[SIZE];
        char buf1[SIZE1];
    
        struct sockaddr_in addr;
        struct sockaddr_in from;
    
        socklen_t len = sizeof(from);
    
        //epoll相关参数
        int epoll_fd = -1;
        int timeout = 3000;
        struct epoll_event event;
        struct epoll_event revent[SIZE]; //返回准备好的文件描述符的集合
    
        int count = 0;
    
        //1. 创建套接字
        sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if (-1 == sockfd)
        {
            perror("sockfd"); 
            goto err0;
        }
    
        //设置端口复用
        op = 1;
        ret = setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (void *)&op, sizeof(op));
        if (-1 == ret)
        {
            perror("setsockopt"); 
            goto err1;
        }
    
        //2. 绑定
        //指定IP和端口
        memset(&addr, 0, sizeof(addr)); 
        addr.sin_family = AF_INET;      //指定协议族
        addr.sin_port = htons(10086);   //指定端口
        inet_pton(AF_INET, "192.168.73.42", (void *)&addr.sin_addr);
    
        ret = bind(sockfd, (void*)&addr, sizeof(addr));
        if (-1 == ret)
        {
            perror("bind"); 
            goto err1;
        }
    
        //3. 监听
        ret = listen(sockfd, 10); 
        if (-1 == ret)
        {
            perror("listen"); 
            goto err1;
        }
        printf("服务端正在监听客户端的连接...\n");
    
        //4. 创建epoll文件描述符
        epoll_fd = epoll_create(1024);
        if (-1 == epoll_fd)
        {
            perror("epoll_create"); 
            goto err1;
        }
        printf("epoll_fd = %d\n", epoll_fd);
    
        //5. 上树 添加监听的文件描述符和事件
        event.events = EPOLLIN; //监听读事件 
        event.data.fd = sockfd; //监听的文件描述符
        ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sockfd, &event);
        if (-1 == ret)
        {
            perror("epoll_ctl"); 
            goto err2;
        }
    
        //6. 检测
        while(1)
        {
            //雇佣了一个秘书
            //检测树中节点文件描述符是否有对应的事件发生 revent返回
            ret = epoll_wait(epoll_fd, revent, SIZE, timeout);
            if (-1 == ret)
            {
                perror("epoll_wait"); 
                break;
            }
            else if (0 == ret)
            {
                printf("3 seconds timeout....\n"); 
                continue;
            }
            else
            {
                //如果成功了 秘书就会告诉我们有多少个描述符准备好
                count = ret;
    
                for (i = 0; i < count; i++)
                {
                    //表示有数据可以读
                    if (revent[i].events & EPOLLIN)
                    {
                        //表示有新的客户端连接服务端
                        if (revent[i].data.fd == sockfd)
                        {
                            //4. 接受客户端的连接
                            connfd = accept(sockfd, (void*)&from, &len);
                            if (-1 == connfd)
                            {
                                perror("accept");
                                continue;
                            }
                            //输出连接服务端的客户端的IP和Port
                            printf("\033[32mclient IP: %s port: %d\033[0m\n", inet_ntop(AF_INET, (void*)&from.sin_addr, buf, SIZE), ntohs(from.sin_port));
    
                            //设置文件描述符为非阻塞
                            int val;
                            //获取对应文件描述符的属性
                            val = fcntl(connfd, F_GETFL);
                            val |= O_NONBLOCK; //追加非阻塞属性
                            //设置非阻塞属性
                            fcntl(connfd, F_SETFL, val);
    
                            //新节点上树
                            event.events = EPOLLIN | EPOLLET; //读事件 设置边沿触发
                            event.data.fd = connfd; //文件描述符
                            ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, connfd, &event);
                            if (-1 == ret)
                            {
                                perror("epoll_ctl"); 
                                break;
                            }
    
                        } //sockfd
                        else
                        {
                            while(1)
                            {
                                //接收数据  每次接收5个字节
                                memset(buf1, 0, SIZE1); 
                                ret = recv(revent[i].data.fd, buf1, SIZE1 - 1, 0);
                                //printf("=====>ret:%d\n", ret);
                                if (ret < 0)
                                {
                                    perror("----->recv");
                                    break; 
                                }
    
                                if (0 == ret)
                                    break;
                                printf("\033[31mbuf: %s\033[0m", buf1); 
    
                                send(revent[i].data.fd, buf1, strlen(buf1), 0);
                            }
    
                            printf("\n");
                        }
                    }
                }
            }
    
        }//while(1)
    
        //6. 关闭连接
        close(epoll_fd);
        close(sockfd);
    
        return 0;
    err2:
        close(epoll_fd);
    err1:
        close(sockfd);
    err0:
        return 1;
    }