【Linux后端服务器开发】select多路转接IO服务器

目录

一、高级IO

二、fcntl

三、select函数接口

四、select实现多路转接IO服务器

---

一、高级IO

在介绍五种IO模型之前，我们先讲解一个钓鱼例子。

• 有一条大河，河里有很多鱼，分布均匀。
• 张三是一个钓鱼新手，他钓鱼的时候很紧张，一刻也不敢放松，于是就死死的盯住鱼线，只要鱼线颤动就说明有鱼咬钩了，他便提竿将鱼放入鱼桶中，再重新钓鱼。
• 李四是一个钓鱼老手，他钓鱼的时候很放松，一边闭目养神一边听着音乐，只是用手感受鱼竿的震动，一旦鱼竿震动就说明有鱼咬钩了，他便提竿将鱼放入鱼桶中，再重新钓鱼。
• 王五也是一个钓鱼老手，也是一边闭目养神一边听着音乐，但是他怕自己不能清楚的感受到鱼竿的震动，于是他在鱼竿上系了一个铃铛，一旦有鱼咬钩了，铃铛就会响动提醒王五，于是王五就提竿将鱼放入鱼桶中，再重新钓鱼。
• 赵六是一个卖鱼的，他是做生意的，并不是为了享受钓鱼的过程，于是他开了一个大卡车，上面固定有很多个鱼竿，他就在车上等待并且循环的查看所有鱼竿，发现那个鱼竿震动或鱼线颤动他就将哪个杆上钓到的鱼取下来，再重新将杆放入水中继续钓鱼。
• 田七是一个有钱的老板，他只是想吃河里的鱼，并不想自己钓鱼，于是他就雇了一个员工小王，让他帮自己钓鱼，一旦钓到鱼将鱼桶装满了，就给自己打电话，自己就开车来取鱼。

好了，例子结束了，以上五个人有五种不同的钓鱼方式，那么谁的钓鱼效率最高呢？答案毫无疑问就是赵六，在相同的时间里，赵六能钓到最多的鱼。

钓鱼的过程就类似于IO过程，钓鱼的过程 = 等 + 钓，IO的过程 = 等 + 读/写

• 鱼是数据
• 大河是内核空间，鱼线颤动、鱼竿震动、铃铛响就是数据就绪的事件
• 鱼竿是文件描述符
• 提竿的动作就是recv/read的调用
• 张三、李四、王五、赵六、田七是不同的进程或线程，员工小王是操作系统

从钓鱼策略角度，张三是阻塞式IO，李四是非阻塞IO，王五是信号驱动式IO，赵六是多路转接(多路复用)IO，田七是异步IO。

从效率上看，张三、李四、王五、田七钓鱼的效率是一样的，因为他们都是只有一个鱼竿，而鱼咬钩的概率是一样的，即阻塞式IO、非阻塞IO、异步IO的效率是一样的。

张三、李四、王五、赵六都亲自参与了钓鱼，即阻塞式IO、非阻塞IO、信号驱动式IO、多路转接IO都亲自参与了IO，称为同步IO。

田七并没有亲自参与钓鱼，即异步IO没有亲自参与IO的任何一个阶段。

• 阻塞式IO：在内核将数据准备好之前，系统调用会一直等待。所有套接字默认都是阻塞IO。
• 非阻塞IO：如果内核还没将数据准备好，系统调用仍然会直接返回，并且返回EWOULDBLOCK错误码。非阻塞IO需要程序员循环的方式尝试读写文件描述符（轮询），这对CPU来说是较大的浪费，一般只有特定场景下才使用。

• 信号驱动IO：内核将数据准备好的时候，使用SIGO信号通知应用程序进行IO操作。
• 多路转接IO：虽然从流程图上看起来和阻塞IO类似，实际上最核心的在于IO多路转接能够同时等待多个文件描述符的就绪状态，并且多路转接将等待事件就绪与处理就绪事件做了分离。

• 异步IO：由内核在数据拷贝完成时，通知应用程序(而信号驱动IO是告诉应用程序何时可以开始拷贝数据)

在任何IO过程中，都包含两个步骤：第一是等待，第二是拷贝。

在实际的应用场景中，等待消耗的时间往往都远远高于拷贝的时间，让IO更高效，最核心的办法就是让等待的时间尽量减少。

二、fcntl

基于fcntl，我们实现一个Set_Nonblock函数，将文件描述符设置为非阻塞。

#include <unistd.h>
#include <fcntl.h>int fcntl(int fd, int cmd, ...);// 复制一个现有的描述符 （cmd = F_DUPFD）
// 获得 / 设置文件描述符标记 （cmd = F_GETFD 或 cmd = F_SETFD）
// 获得 / 设置文件状态标记 （cmd = F_GETFL 或 cmd = F_SETFL）
// 获得 / 设置异IO所有权 （cmd = F_GETOWN 或 cmd = F_SETOWN）
// 获得 / 设置记录锁 （cmd = F_GETLK 或  cmd = SETLK）

#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>void Set_Nonblock(int fd)
{int f1 = fcntl(fd, F_GETFL);if (f1 < 0){perror("fcntl");return;}fcntl(fd, F_SETFL, f1 | O_NONBLOCK);
}int main()
{Set_Nonblock(0);while (1){char buf[1024];ssize_t read_size = read(0, buf, sizeof(buf) - 1);if (read_size < 0){perror("read");sleep(1);continue;}printf("input: %s\n", buf);}return 0;
}

• 我们通过获取/设置文件状态标记，便可以将一个文件描述符设置为非阻塞
• 使用F_GETFL将当前的文件描述符的属性取出来（一个位图结构）
• 再使用F_SETFL将文件描述符设置回去，设置回去的同时，加上一个O_NONBOLOCK参数
• 轮询的方式读取标准输入

三、select函数接口

系统提供select函数来实现多路转接IO模型

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

#include <sys/select.h>int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *excptfds, struct timeval* timeout);# 参数解释：
# 参数nfds是需要监视的最大文件描述符值+1
# rdset、wrset、exset分别对应于需要检测的可读文件描述符的集合、可写文件描述符的集合、异常文件描述符集合
# 参数timeout结构为timeval，用来设置select()的等待时间# 参数timeout的取值：
# nullptr：表示select()没有timeout，select将一直被阻塞，直到某个文件描述符上发生事件
# 0：仅检测描述符集合的状态，然后立即返回，并不等待外部事件的发生
# 特定的时间值：如果在指定的时间段里没有时间发生，select将超时返回

fd_set结构：一个整数结构（位图结构），使用位图中的位来表示需要监视的文件描述符

/* The fd_set member is required to be an array of longs.  */
typedef long int __fd_mask;typedef struct
{/* XPG4.2 requires this member name.  Otherwise avoid the namefrom the global namespace.  */
#ifdef __USE_XOPEN__fd_mask fds_bits[__FD_SETSIZE / __NFDBITS];
#define __FDS_BITS(set) ((set)->fds_bits)
#else__fd_mask __fds_bits[__FD_SETSIZE / __NFDBITS];
#define __FDS_BITS(set) ((set)->__fds_bits)
#endif
} 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结构：用于描述一段时间长度，如果在这个时间内，需要监视的描述符没有事件发生或函数返回，返回值为0。

struct timeval
{__time_t tv_sec;       /* Seconds.  */__suseconds_t tv_usec; /* Microseconds.  */
};

select函数返回值

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

select的执行过程

理解select的关键在于理解fd_set，为方便说明，取fd_set长度为1字节，fd_set中的每一位bit可以对应一个文件描述符fd，1字节长度的fd_set最大可以对应8个fd

1. 执行fd_set set; FD_ZERO(&set); 则set用位表示是0000 0000
2. 若fd=5，执行FD_SET(fd, &set); 后set变为0001 0000（第5位置1）
3. 若再加入fd=2，fd=1，则set变为0001 0011
4. 执行select(6, &set, nullptr, nullptr, nullptr)阻塞等待
5. 若fd=1，fd=2上都发生可读事件，则select返回，此时select变为0000 0011（注意：没有发生事件的fd=5被清空）

socket就绪条件

读就绪：

①socket内核中，接收缓冲区的字节数，大于等于低水位标记SO_RECVLOWAT。此时可以无阻塞的读该文件描述符，并且返回值大于0；

②socket TCP通信中，对端关闭连接，此时对socket读返回0；

③监听的socket上有新的连接请求；

④socket上有未处理的错误。

写就绪：

①socket内核中，发送缓冲区的可用字节数(发送缓冲区的闲置空间大小)大于等于低水位标记SO_SNDLOWAT，此时可以无阻塞的写，并且返回值大于0；

②socket的写操作被关闭，对一个写操作被关闭的文件描述符进行写操作，会触发SIGPIPE信号；

③socket使用非阻塞connect连接成功或失败之后；

④socket上有未读取的错误。

select的特点

• 可监控的文件描述符个数取决于与sizeof(fd_set)的值，不同的系统的fd_set值不同，通常情况下服务器支持可监控的最大文件描述符个数是数千个。
• 将fd加入select监控集的同时，还要再使用一个数组数据结构array保存放到select中的fd。一是用于在select返回后，array作为源数据和fd_set进行FD_ISSET判断；二是select返回后会把以前加入的但是无事发生的fd清空，则每次开始select前都需要重新从array取得fd逐一加入，扫描array的同时取得fd最大值fdmax，用于select的第一个参数。
• fd_set的大小可调整，涉及到重新编译内核。

select的缺陷

• 每次调用select，都需要手动设置fd集合，从接口使用角度来说并不方便
• 每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大
• 每次调用select，都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也很大
• select支持的文件描述符数量上限太小

四、select实现多路转接IO服务器

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 4#define NUM 1024const 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 Log_Message(int level, const char *format, ...)
{char 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;
}

Sock.hpp

#pragma once#include <iostream>
#include <string>
#include <cstring>
#include <unistd.h>#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>#include "Log.hpp"enum
{USAGE_ERR = 1,SOCKET_ERR,BIND_ERR,LISTEN_ERR
};class Sock
{const static int backlog = 32;public:static int Socket(){// 1. 创建socket文件套接字对象int sock = socket(AF_INET, SOCK_STREAM, 0);if (sock < 0){Log_Message(FATAL, "create socket error");exit(SOCKET_ERR);}Log_Message(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){Log_Message(FATAL, "bind socket error");exit(BIND_ERR);}Log_Message(NORMAL, "bind socket success");}static void Listen(int sock){// 3. 设置socket 为监听状态if (listen(sock, backlog) < 0) // 第二个参数backlog后面在填这个坑{Log_Message(FATAL, "listen socket error");exit(LISTEN_ERR);}Log_Message(NORMAL, "listen socket success");}static int Accept(int listensock, std::string *clientip, uint16_t *clientport){struct sockaddr_in peer;socklen_t len = sizeof(peer);int sock = accept(listensock, (struct sockaddr *)&peer, &len);if (sock < 0)Log_Message(ERROR, "accept error, next");else{Log_Message(NORMAL, "accept a new link success, get new sock: %d", sock); // ?*clientip = inet_ntoa(peer.sin_addr);*clientport = ntohs(peer.sin_port);}return sock;}
};

SelectServer.hpp

#pragma once#include <iostream>
#include <string>
#include <functional>#include "Sock.hpp"using namespace std;static const int g_defaultport = 8080;
static const int g_fdnum = sizeof(fd_set) - 1;
static const int g_defaultfd = -1;using func_t = function<string(const string)>;class SelectServer
{
public:SelectServer(func_t func, int port = g_defaultport): _func(func), _port(port), _listensock(g_defaultfd), _fdarray(nullptr){}void Init(){_listensock = Sock::Socket();Sock::Bind(_listensock, _port);Sock::Listen(_listensock);_fdarray = new int[g_fdnum];for (int i = 0; i < g_fdnum; ++i)_fdarray[i] = g_defaultfd;_fdarray[0] = _listensock;    // 不变}void Print_FD_List(){   cout << "fd list: ";for (int i = 0; i < g_fdnum; ++i)if (_fdarray[i] != g_defaultfd)cout << _fdarray[i] << " ";cout << endl;}void Accepter(int listensock){Log_Message(DEBUG, "Accept in");string clientip;uint16_t clientport = 0;int sock = Sock::Accept(listensock, &clientip, &clientport);    // accept = 等 + 获取连接if (sock < 0)return;Log_Message(NORMAL, "accept success [%s: %d]", clientip.c_str(), clientport);// sock我们能直接recv/read吗？不能，只有select有资格检测事件是否就绪// 将新的sock托管给select：将新的sock添加到_fdarray数组中int i = 0;for (; i < g_fdnum; ++i){if (_fdarray[i] != g_defaultfd)continue;elsebreak;}if (i == g_fdnum){Log_Message(WARNING, "server if full, please wait");close(sock);}else{_fdarray[i] = sock;}Print_FD_List();Log_Message(DEBUG, "Accept out");}void Recver(int sock, int pos){Log_Message(DEBUG, "in Recver");// 1. 读取requestchar buffer[1024];ssize_t s = recv(sock, buffer, sizeof(buffer) - 1, 0);if (s > 0){buffer[s] = 0;Log_Message(NORMAL, "client# %s", buffer);}else if (s == 0){close(sock);_fdarray[pos] = g_defaultfd;Log_Message(NORMAL, "client quit");return;}else{close(sock);_fdarray[pos] = g_defaultfd;Log_Message(ERROR, "client quit: %s", strerror(errno));return;}// 2. 处理requeststring response = _func(buffer);// 3. 返回responsewrite(sock, response.c_str(), response.size());Log_Message(DEBUG, "out Recver");}// 1. handler event rfds中，不仅仅是有一个fd是就绪的，可能存在多个// 2. 我么你的select目前只处理read事件void Handler_Read_Envent(fd_set& rfds){for (int i = 0; i < g_fdnum; ++i){// 过滤掉非法的fdif (_fdarray[i] == g_defaultfd)continue;// 正常的fdif (FD_ISSET(_fdarray[i], &rfds) && _fdarray[i] == _listensock)Accepter(_listensock);else if (FD_ISSET(_fdarray[i], &rfds))Recver(_fdarray[i], i);}}void Start(){while (1){fd_set rfds;FD_ZERO(&rfds);int maxfd = _fdarray[0];for (int i = 0; i < g_fdnum; ++i){// 将全部合法的fd添加到读文件描述符中if (_fdarray[i] == g_defaultfd)continue;FD_SET(_fdarray[i], &rfds);// 更新所有的fd中最大的fdif (maxfd < _fdarray[i])maxfd = _fdarray[i];}Log_Message(NORMAL, "maxfd is: %d", maxfd);// 一般而言，要是用select，需要程序员自己维护一个保存所有合法fd的数组int n = select(maxfd + 1, &rfds, nullptr, nullptr, nullptr);switch (n){case 0:Log_Message(NORMAL, "timeout ...");break;case -1:Log_Message(WARNING, "select error, code: %d, err string: %s", errno, strerror(errno));break;default:// 有事件就绪Log_Message(NORMAL, "have event ready!");Handler_Read_Envent(rfds);break;}}}~SelectServer(){if (_listensock < 0)close(_listensock);if (_fdarray)delete[] _fdarray;}private:int _port;int _listensock;int* _fdarray;func_t _func;
};

main.cc

#include "SelectServer.hpp"
#include <memory>using namespace std;static void Usage(std::string proc)
{std::cerr << "Usage:\n\t" << proc << " port" << "\n\n";exit(USAGE_ERR);
}std::string Transaction(const std::string &request)
{return request;
}// ./select_server 8081
int main(int argc, char *argv[])
{if(argc != 2)Usage(argv[0]);unique_ptr<SelectServer> svr(new SelectServer(Transaction, atoi(argv[1])));// std::cout << "test: " << sizeof(fd_set) * 8 << std::endl;// unique_ptr<SelectServer> svr(new SelectServer(Transaction));svr->Init();svr->Start();return 0;
}

执行效果：运行服务器之后，通过telnet连接服务器，向服务器发送数据并得到响应