目录
一、高级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 name
from 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
- 执行fd_set set; FD_ZERO(&set); 则set用位表示是0000 0000
- 若fd=5，执行FD_SET(fd, &set); 后set变为0001 0000（第5位置1）
- 若再加入fd=2，fd=1，则set变为0001 0011
- 执行select(6, &set, nullptr, nullptr, nullptr)阻塞等待
- 若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 1024
const 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;
else
break;
}
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. 读取request
char 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. 处理request
string response = _func(buffer);
// 3. 返回response
write(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)
{
// 过滤掉非法的fd
if (_fdarray[i] == g_defaultfd)
continue;
// 正常的fd
if (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中最大的fd
if (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连接服务器，向服务器发送数据并得到响应
