I/O多路转接之select

承接上文：I/O模型之非阻塞IO-CSDN博客

简介

select函数原型介绍使用

一个select简单的服务器的代码书写

select的缺点

初识select

系统提供select函数来实现多路复用输入/输出模型

select系统调用是用来让我们的程序监视多个文件描述符的状态变化的;
程序会停在select这里等待，直到被监视的文件描述符有一个或多个发生了状态改变

首先要知道的一点，是 select 只负责等待，并且可以一次等待多个fd，select本身没有数据拷贝的能力，拷贝要read，write来完成

就好像有一群人等着吃饭，有一个人负责给他们通知饭好了吗，当有人的饭好了，就叫那一个人过来吃饭，饭本身表示负责等待的人的，而是别人的，同样的其他人相当于不用等，只负责吃就好了

select函数原型

select的函数原型如下: #include <sys/select.h>

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

参数解释

参数nfds是需要监视的最大的文件描述符值+1；
rdset,wrset,exset分别对应于需要检测的可读文件描述符的集合，可写文件描述符的集 合及异常文件描述符的集合;
参数timeout为结构timeval，用来设置select()的等待时间

参数timeout取值

NULL：则表示select（）没有timeout， select将一直被阻塞，直到某个文件描述符上发生了事件;
0：仅检测描述符集合的状态，然后立即返回，并不等待外部事件的发生。
特定的时间值：如果在指定的时间段里没有事件发生， select将超时返回

关于fd_set结构

其实这个结构就是一个整数数组, 更严格的说, 是一个 "位图". 使用位图中对应的位来表示要监视的文件描述符.

提供了一组操作fd_set的接口, 来比较方便的操作位图.

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的全部位

关于timeval结构

timeval结构用于描述一段时间长度，如果在这个时间内，需要监视的描述符没有事件发生则函数返回，返回值为0

函数返回值：

执行成功则返回文件描述词状态已改变的个数
如果返回0代表在描述词状态改变前已超过timeout时间，没有返回
当有错误发生时则返回-1，错误原因存于errno，此时参数readfds， writefds, exceptfds和timeout的值变成不可预测

错误值可能为：

EBADF 文件描述词为无效的或该文件已关闭
EINTR 此调用被信号所中断
EINVAL 参数n 为负值。
ENOMEM 核心内存不足

 

编写代码

准备工作代码

这是只是一个服务器基本构架，现在什么功能都没有，关于里面的细节，可以参考我前面的文章，这里的代码是直接从前面文章中提取出来，简单处理过后的一个服务器

err.hpp

#pragma once// 错误信息
enum
{USAGE_ERR = 1,SOCKET_ERR,BIND_ERR,LISTEN_ERR
};

log.hpp

#pragma once#include <iostream>
#include <string>
#include <cstdarg>
#include <ctime>
#include <unistd.h>#define DEBUG   0
#define NORMAL  1
#define WARNING 2
#define ERROR   3
#define FATAL   4const char * to_levelstr(int level)
{switch(level){case DEBUG : return "DEBUG";case NORMAL: return "NORMAL";case WARNING: return "WARNING";case ERROR: return "ERROR";case FATAL: return "FATAL";default : return nullptr;}
}void logMessage(int level, const char *format, ...)
{
#define NUM 1024char logprefix[NUM];snprintf(logprefix, sizeof(logprefix), "[%s][%ld][pid: %d]",to_levelstr(level), (long int)time(nullptr), getpid());char logcontent[NUM];va_list arg;va_start(arg, format);vsnprintf(logcontent, sizeof(logcontent), format, arg);std::cout << logprefix << logcontent << std::endl;
}

main.cc

#include "selectServer.hpp"
#include "err.hpp"
#include <memory>using namespace std;
using namespace select_ns;static void usage(std::string proc)
{std::cerr << "Usage:\n\t" << proc << " prot" << "\n\n";
}// ./select_server 8081
int main(int argc, char* argv[])
{   if(argc != 2){usage(argv[0]);exit(USAGE_ERR);}unique_ptr<SelectServer> svr(new SelectServer(atoi(argv[1])));svr->initServer();svr->start();return 0;
}

makefile

select_server:main.ccg++ -o $@ $^ -std=c++11
.PHONY:clean
clean:rm -f select_server

selectServer.hpp

#pragma once#include <iostream>#include "Sock.hpp"namespace select_ns
{   static const int defaultport = 8081;    // 默认启动端口号class SelectServer{public:SelectServer(int port = defaultport):_port(port), _listensock(-1){}void initServer(){_listensock = Sock::Socket();Sock::Bind(_listensock, _port);Sock::Listen(_listensock);}void start(){for(;;){   std::string clientip;uint16_t clientport = 0;int sock = Sock::Accept(_listensock, &clientip, &clientport);if(sock<0) continue;// 开始进行服务器的处理逻辑}}~SelectServer(){if(_listensock < 0) close(_listensock);}private:int _port;int _listensock;};
}

Sock.hpp

#pragma once#include <iostream>
#include <cstring>
#include <string>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>#include "log.hpp"
#include "err.hpp"class Sock
{const static int backlog = 32; // 全连接长度，为了方便演示，直接使用默认的
public:// 创建一个套接字static int Socket(){// 1. 创建socket文件套接字对象int sock = socket(AF_INET, SOCK_STREAM, 0);if (sock < 0){logMessage(FATAL, "create socket error");exit(SOCKET_ERR);}logMessage(NORMAL, "create socket success: %d", sock);int opt = 1;setsockopt(sock, SOL_SOCKET, SO_REUSEADDR|SO_REUSEPORT , &opt, sizeof(opt));    // 端口复用return sock;}// 特定的套接字和特定的端口号进行绑定 -- 因为这里是演示代码，所以这里的端口号直接使用默认的static void Bind(int sock, int port){// 2. bind绑定自己的网络信息struct sockaddr_in local;memset(&local, 0, sizeof(local));local.sin_family = AF_INET;local.sin_port = htons(port);local.sin_addr.s_addr = INADDR_ANY;if (bind(sock, (struct sockaddr *)&local, sizeof(local)) < 0){logMessage(FATAL, "bind socket error");exit(BIND_ERR);}logMessage(NORMAL, "bind socket success");}static void Listen(int sock){// 3. 设置socket 为监听状态if (listen(sock, backlog) < 0) // 第二个参数全连接长度，为了方便演示，直接使用默认的{logMessage(FATAL, "listen socket error");exit(LISTEN_ERR);}logMessage(NORMAL, "listen socket success");}// 监听套接字，待提取的对端的ip，待提取的对端的端口号 -- 提取后放到参数中去让调用者使用static int Accept(int listensock, std::string *clientip, uint16_t *clientport){// 4. server 获取新链接// sock, 和client进行通信的fdstruct sockaddr_in peer;socklen_t len = sizeof(peer);int sock = accept(listensock, (struct sockaddr *)&peer, &len);if (sock < 0)logMessage(ERROR, "accept error, next");else{logMessage(NORMAL, "accept a new link success, get new sock: %d", sock);*clientip = inet_ntoa(peer.sin_addr); //将一个网络字节序的IP地址（也就是结构体in_addr类型变量,表示一个32位的IPv4地址）转化为点分十进制的IP地址（字符串）*clientport = ntohs(peer.sin_port);}return sock;}
};

有了一个基础的服务器，我们现在可以将其改成为select多路转接的形式

1.监听套接字也应该放入select中去

2.start修正

运行结果

每隔3秒一次非阻塞式，当我们有链接到来的时候就会触发下面的情况

图中我们会发现这货会一直在打印default的日志信息，这是因为我们虽然读取到了，但是我们并没有将其进行处理，拿走已经响应好的套接字，这样就会导致一直有链接到来，因为原来的链接一直没有被拿走，监听套接字一直有消息到来(原来的没被取走)

当我们单方面的关闭链接的时候，这时候服务器还是会一直打印日志，因为原来的链接没有被拿走，这种状态被称为半状态链接

上面我们已经利用select监听到了链接，现在进行链接的处理行为

3.读取链接

走到这里，accept 函数，会不会阻塞???

select 告诉我，listensock读事件就绪了，于是一定不会阻塞

但是同时有一个问题，我们能够直接读取吗？

显然是不能的，所以在我们实现select服务器的时候，需要自己维护一个套接字的数组，来让我们知道那些描述符是合法的，并且之前我们select的第一个参数是直接使用_listensock+1来暂时填写的，但是实际上是不能的，这是一个动态变化的，通过数组我们就可以解决以上的各种问题，为此我们在类中再添加一个对象

4.文件描述符数组

那么我们首先要知道的一个是，这个数组应该要多大呢？

我们首先要知道的是，fd_set是一个类型，这是一个位图结构，那么它的大小一定是固定的，我们再通过

得到了结果，这是一个16字节的类型，位图结构，一个可以存储128个文件描述符，这下我们就知道了这个数据应该设置为多大了

我们这样设置

注意这一个最大的sock也需要更改 

接下来我们就可以将新的sock，托管给select了，注意将新的sock托管给select的本质，其实就是将sock，添加到fdarray数组中即可！

5.将新的sock添加到数组中去

打印测试

结果

6.整理解耦

Recver

进行业务逻辑处理 -- 处理对应的事件

Accepter

将 listensock 到来的套接字添加到数组中去

测试结果

select优缺点

缺点

有上限

        可监控的文件描述符个数取决与sizeof(fd_set)的值. 我这边服务器上sizeof(fd_set)＝ 512，每bit表示一个文件描述符，则我服务器上支持的最大文件描述符是512*8=4096；

        因为该类型是一个位图结构，是位图结构就有上限

        备注: fd_set的大小可以调整，可能涉及到重新编译内核，可以自行去了解

需要自己创建一个数组

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

 

因为有了以上的种种缺陷，因此就诞生了下一篇我们讲述的话题  I/O多路转接之poll

源码

err.hpp

#pragma once// 错误信息
enum
{USAGE_ERR = 1,SOCKET_ERR,BIND_ERR,LISTEN_ERR
};

log.hpp

#pragma once#include <iostream>
#include <string>
#include <cstdarg>
#include <ctime>
#include <unistd.h>#define DEBUG   0
#define NORMAL  1
#define WARNING 2
#define ERROR   3
#define FATAL   4const char * to_levelstr(int level)
{switch(level){case DEBUG : return "DEBUG";case NORMAL: return "NORMAL";case WARNING: return "WARNING";case ERROR: return "ERROR";case FATAL: return "FATAL";default : return nullptr;}
}void logMessage(int level, const char *format, ...)
{
#define NUM 1024char logprefix[NUM];snprintf(logprefix, sizeof(logprefix), "[%s][%ld][pid: %d]",to_levelstr(level), (long int)time(nullptr), getpid());char logcontent[NUM];va_list arg;va_start(arg, format);vsnprintf(logcontent, sizeof(logcontent), format, arg);std::cout << logprefix << logcontent << std::endl;
}

main.cc

#include "selectServer.hpp"
#include "err.hpp"
#include <memory>using namespace std;
using namespace select_ns;static void usage(std::string proc)
{std::cerr << "Usage:\n\t" << proc << " prot" << "\n\n";
}std::string transaction(const std::string &request)
{return request;
}// ./select_server 8081
int main(int argc, char* argv[])
{   if(argc != 2){usage(argv[0]);exit(USAGE_ERR);}unique_ptr<SelectServer> svr(new SelectServer(transaction));svr->initServer();svr->start();return 0;
}

makefile

select_server:main.ccg++ -o $@ $^ -std=c++11
.PHONY:clean
clean:rm -f select_server

selectServer.hpp

#pragma once#include <iostream>
#include <string>
#include <functional>
#include "Sock.hpp"namespace select_ns
{static const int defaultport = 8081;         // 默认启动端口号static const int fdnum = sizeof(fd_set) * 8; // 通过sizeof可以知道这个类型有多大，我们的数组要设置为多大static const int defaultfd = -1;             // 我们把不关注的全部设置为-1using func_t = std::function<std::string(const std::string &)>; // 创建一个函数指针，传递业务逻辑的处理方法class SelectServer{public:SelectServer(func_t f, int port = defaultport) : func(f), _port(port), _listensock(-1){}void initServer(){_listensock = Sock::Socket();Sock::Bind(_listensock, _port);Sock::Listen(_listensock);fdarray = new int[fdnum]; // 创建数组for (int i = 0; i < fdnum; ++i)fdarray[i] = defaultfd; // 全部先初始化为-1fdarray[0] = _listensock;   // 不变了, 这个放置在那都没有问题}// 一个用来检测的打印函数void Print(){std::cout << "fd list: ";for (int i = 0; i < fdnum; ++i){if (fdarray[i] != defaultfd)std::cout << fdarray[i] << " ";}std::cout << std::endl;}// 将 listensock 到来的套接字添加到数组中去void Accepter(int listensock){logMessage(DEBUG, "Accepter in"); // Accepter 进入// 走到这里，accept 函数，会不会阻塞???// select 告诉我，listensock读事件就绪了，于是一定不会阻塞std::string clientip;uint16_t clientport = 0;int sock = Sock::Accept(_listensock, &clientip, &clientport); // accept = 等 + 获取if (sock < 0)return;logMessage(NORMAL, "accept success [%s:%d]", clientip.c_str(), clientport);// sock 我们能直接recv/read 吗？// 不能 -- 不能保证底层有没有数据，可能只是建立链接了还没有发送数据// 那么这个进程(假如是多进程)就会被阻塞// 整个代码，只有select有资格检测事件是否就绪// 所以我们需要把新的套接字再交付给select，来让它帮我们检测是否有数据真的到了// 将新的sock 托管给select！// 将新的sock托管给select的本质，其实就是将sock，添加到fdarray数组中即可！int i = 0;for (; i < fdnum; ++i){if (fdarray[i] != defaultfd)continue;elsebreak;}if (i == fdnum){logMessage(WARNING, "server is full, please wait"); // 等待的sock已经满了，不能再添加了close(sock);                                        // 直接关闭掉这个套接字}else{fdarray[i] = sock;}Print();                           // 打印测试logMessage(DEBUG, "Accepter out"); // 出去Accepter}// 进行业务逻辑处理 -- 处理对应的事件void Recver(int sock, int pos){logMessage(DEBUG, "in Recver"); // 进入了Recver// 1. 读取request// 这样写是有问题的，我们无法保证读取的是一个完整的报文，这一点在我们之前的协议定制才能解决，这里为了演示就不做处理了char buffer[1024];ssize_t s = recv(sock, buffer, sizeof(buffer) - 1, 0); // 在这里进行读取的时候，一定不会阻塞，因为读事件已经就绪if (s > 0){// 读取成功，回显buffer[s] = 0;logMessage(NORMAL, "client# %s", buffer);}else if (s == 0){// 读取完毕，退出并且关闭close(sock);fdarray[pos] = defaultfd; // 将这个sock移除数组中，不再关心这个logMessage(NORMAL, "client quit");return;}else{// 读取失败, 退出并且关闭sockclose(sock);fdarray[pos] = defaultfd; // 将这个sock移除数组中，不再关心这个logMessage(ERROR, "client quit");return;}// 2.处理 requeststd::string response = func(buffer);// 3.返回 response// write bugwrite(sock, response.c_str(), response.size());logMessage(DEBUG, "out Recver"); // 出Recver}// 1. handler even rfds 中, 不仅仅是有一个fd是就绪的，可能存在多个// 2. 我们的select目前只处理了read事件void HandlerReadEvent(fd_set &rfds){for (int i = 0; i < fdnum; ++i){// 目前一定是listensock，因为我们只把listensock放入了select -- 之后再进行处理if (FD_ISSET(fdarray[i], &rfds) && fdarray[i] == _listensock) // 修正一下判断Accepter(_listensock);else if (FD_ISSET(fdarray[i], &rfds)) // 读事件已经就绪的文件描述符进行处理Recver(fdarray[i], i);            // 进行业务逻辑处理 -- 处理对应的事件else                                  // 读事件话还没有就绪的文件描述符进行处理{}}}void start(){for (;;){fd_set rfds;    // 创建一个读文件描述符集 -- 只处理读// fd_set wfds; // 创建一个写文件描述符集 -- 只处理写 -- 未来可以实现FD_ZERO(&rfds); // 进行清空int maxfd = fdarray[0];for (int i = 0; i < fdnum; i++){if (fdarray[i] == defaultfd)continue;FD_SET(fdarray[i], &rfds); // 将合法 fd 全部添加到读文件描述符集中if (maxfd < fdarray[i])maxfd = fdarray[i]; // 更新所有fd中最大的fd}logMessage(NORMAL, "max fd is: %d", maxfd); // 打印最大的fd// 不关心 写 和 异常，只关心读事件// struct timeval timeout = {3, 0};                                    // 每隔一秒回来一次, 并且需要注意的是，这个时间需要每一次循环都重新设定，不然在select中会被修改，然后造成非阻塞式等待// int n = select(_listensock + 1, &rfds, nullptr, nullptr, &timeout); // 暂时这样书写，肯定是错误的写法，因为listensock是一个固定值，但是实际上这个值是变化的int n = select(maxfd + 1, &rfds, nullptr, nullptr, nullptr);switch (n){case 0:logMessage(NORMAL, "timeout..."); // 正常返回break;case -1:logMessage(WARNING, "select error, code: %d, err string: %s", errno, strerror(errno));default:// 说明有事件就绪了，目前只有一个监听事件就绪了logMessage(NORMAL, "have event ready!"); // 正常返回HandlerReadEvent(rfds);                  // 处理读链接的函数// HandlerWriteEvent(rfds);              // 处理写链接的函数 -- 目前没有实现，还要一个写事件集合break;}// 下面的写法因为我们不知道监听套接字什么时候就绪，所以这是一种阻塞式写法，进行修正/*std::string clientip;uint16_t clientport = 0;int sock = Sock::Accept(_listensock, &clientip, &clientport); // accept = 等 + 获取if(sock<0) continue;*/// 开始进行服务器的处理逻辑}}~SelectServer(){if (_listensock < 0)close(_listensock);if (fdarray)delete[] fdarray; // 清空数组}private:int _port;int _listensock;int *fdarray; // 创建一个数据放置文件描述符func_t func;};
}

Sock.hpp

#pragma once#include <iostream>
#include <cstring>
#include <string>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>#include "log.hpp"
#include "err.hpp"class Sock
{const static int backlog = 32; // 全连接长度，为了方便演示，直接使用默认的
public:// 创建一个套接字static int Socket(){// 1. 创建socket文件套接字对象int sock = socket(AF_INET, SOCK_STREAM, 0);if (sock < 0){logMessage(FATAL, "create socket error");exit(SOCKET_ERR);}logMessage(NORMAL, "create socket success: %d", sock);int opt = 1;setsockopt(sock, SOL_SOCKET, SO_REUSEADDR|SO_REUSEPORT , &opt, sizeof(opt));    // 端口复用return sock;}// 特定的套接字和特定的端口号进行绑定 -- 因为这里是演示代码，所以这里的端口号直接使用默认的static void Bind(int sock, int port){// 2. bind绑定自己的网络信息struct sockaddr_in local;memset(&local, 0, sizeof(local));local.sin_family = AF_INET;local.sin_port = htons(port);local.sin_addr.s_addr = INADDR_ANY;if (bind(sock, (struct sockaddr *)&local, sizeof(local)) < 0){logMessage(FATAL, "bind socket error");exit(BIND_ERR);}logMessage(NORMAL, "bind socket success");}static void Listen(int sock){// 3. 设置socket 为监听状态if (listen(sock, backlog) < 0) // 第二个参数全连接长度，为了方便演示，直接使用默认的{logMessage(FATAL, "listen socket error");exit(LISTEN_ERR);}logMessage(NORMAL, "listen socket success");}// 监听套接字，待提取的对端的ip，待提取的对端的端口号 -- 提取后放到参数中去让调用者使用static int Accept(int listensock, std::string *clientip, uint16_t *clientport){// 4. server 获取新链接// sock, 和client进行通信的fdstruct sockaddr_in peer;socklen_t len = sizeof(peer);int sock = accept(listensock, (struct sockaddr *)&peer, &len);if (sock < 0)logMessage(ERROR, "accept error, next");else{logMessage(NORMAL, "accept a new link success, get new sock: %d", sock);*clientip = inet_ntoa(peer.sin_addr); //将一个网络字节序的IP地址（也就是结构体in_addr类型变量,表示一个32位的IPv4地址）转化为点分十进制的IP地址（字符串）*clientport = ntohs(peer.sin_port);}return sock;}
};