计算机网络（10） --- 高级IO

计算机网络（9） --- 数据链路层与MAC帧_哈里沃克的博客-CSDN博客数据链路层与MAC帧https://blog.csdn.net/m0_63488627/article/details/132178583?spm=1001.2014.3001.5501

1.IO介绍

1.IO本质

1.如果数据没有出现，那么读取文件其实会被阻塞住，以等待资源的就绪；或者数据还在网络上传输，并没有到来，需要等待数据到来

2.而操作系统给我们的读取接口，其实是对数据的拷贝

本质：IO=等数据到来+数据拷贝

其实拷贝数据是两个硬件之间的传输，对于软件层的我们而言无法进行进一步优化；又因为等待的时间其实比拷贝时间要来的多。所以拷贝在IO中的效率占比不是很大

高效IO本质：减少等待的时间带来的成本

2.IO模型

1.阻塞IO: 在内核将数据准备好之前, 系统调用会一直等待. 所有的套接字, 默认都是阻塞方式

2.非阻塞IO: 如果内核还未将数据准备好, 系统调用仍然会直接返回, 并且返回EWOULDBLOCK错误码。往往需要程序员循环的方式反复尝试读写文件描述符, 这个过程称为轮询.

3.信号驱动IO: 内核将数据准备好的时候, 使用SIGIO信号通知应用程序进行IO操作

4.IO多路转接: 虽然从流程图上看起来和阻塞IO类似. 实际上最核心在于IO多路转接能够同时等待多个文件描述符的就绪状态

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

2.非阻塞

void setNoBlock(int fd)
{int fl = fcntl(fd, F_GETFL); //得到原先文件描述符的状态if (fl < 0){std::cerr << "fcntl : " << strerror(errno) << std::endl;}fcntl(fd, F_SETFL, fl | O_NONBLOCK); //追加文件描述符的状态信息
}int main()
{char buffer[1024];while (1){setNonBlock(0);std::cout << ">>>> ";fflush(stdout);ssize_t s = read(0, buffer, sizeof(buffer) - 1);if (s > 0){buffer[s] = 0;std::cout << "echo# " << buffer << std::endl;}else if (s == 0){std::cout << "read end" << std::endl;break;}else{if (errno == EAGAIN){std::cout << "没有错，只是没有数据" << std::endl;}else if (errno == EINTR){std::cout << "系统调用被中断" << std::endl;continue;}else{std::cout << "出错" << std::endl;break;}}}return 0;
}

F_GETFL：得到文件描述符的状态

F_SETFL：追加文件描述符的状态信息

O_NONBLOCK：非阻塞模式

3.IO多路转接

1.select

1.select表现为等待数据

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

1.nfds：监视的多个文件描述符中，最大的文件描述符+1为输入值

2.timeout：等待多个文件描述符时，等待的方式。输入nullptr则表示阻塞式等待；设置传入的数据结构timeval = {0，0}表示非阻塞等待；timeval = {x，0}表示x秒内为阻塞式等待，超过5秒为非阻塞等待

3.返回值：多少文件描述符就绪，则返回多少个文件描述符的数；返回值为0，表示返回超时了；返回值小于0，表示select调用失败

4.select关心的时间只有三类：读、写、异常。fd_set是一个位图，用于表示文件描述符的集合。

5.fd_set：输入的位图参数为自己的需要进行管理的文件描述符置为1；返回则是内核告诉用户哪些文件描述符已经就绪了

编写代码

1.listen套接字也需要被select连接，将其归类为读事件

2.检测事件只有select有这个功能设计，所以需要将连接交给select进行处理

3.操作系统提供的位图大小为1024bite，所以我们需要拿出一个数组fdarray大小也为1024进行管理。

namespace select_ns
{static const int defaultport = 8081;static const int fdnum = sizeof(fd_set) * 8;static const int defaultfd = -1;class SelectServer{public:SelectServer(int port = defaultport) : _port(port), _listensock(-1), fdarray(nullptr){}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;}void HandlerEvent(fd_set &rfds){//? 目前一定是listensock，只有这一个if (FD_ISSET(_listensock, &rfds)){// 走到这里，accept 函数，会不会阻塞？？？1 0// 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有资格检测事件是否就绪// 将新的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 if full, please wait");close(sock);}else{fdarray[i] = sock;}Print();}}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;fdarray[0] = _listensock; // 不变了}void start(){for (;;){fd_set rfds;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}// struct timeval timeout = {1, 0};// int n = select(_listensock + 1, &rfds, nullptr, nullptr, &timeout); // ??// 一般而言，要是用select，需要程序员自己维护一个保存所有合法fd的数组！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));break;default:logMessage(NORMAL, "get a new link...");HandlerEvent(rfds);break;}}}~SelectServer(){if (_listensock < 0)close(_listensock);if (fdarray)delete[] fdarray;}private:int _port;int _listensock;int *fdarray;};
}

优缺点

1.select等待的文件描述符是有上限的，除非重新改内核能提高上限，否则无法解决。

2.需要借助第三方数组对select的文件描述符进行管理

3.需要不断检查不同的位图，进行循环管理，时间成本高

4.select的第一个参数为最大fd+1的目的是：用于select遍历合法文件描述符的范围

2.poll

1.poll解决了select的fd有上限问题

2.解决select需要反复设置fd问题

1.fds：为一个动态数组

2.nfds：fds数组的长度

3.timeout：ms为单位，当数>0在timeout内阻塞，超过时间非阻塞方式进行等待；=0以非阻塞方式进行等待；<0以阻塞方式进行等待

4.pollfd：为一个结构体表示fd和对应的events事件。event表示内核告诉用户哪些事件准备就绪；revent则是输出

特点：输入输出分离，大小可设置

编写代码

namespace poll_ns
{static const int defaultport = 8081;static const int num = 2048;static const int defaultfd = -1;using func_t = std::function<std::string (const std::string&)>;class PollServer{public:PollServer(func_t f, int port = defaultport) : _func(f), _port(port), _listensock(-1), _rfds(nullptr){}void initServer(){_listensock = Sock::Socket();Sock::Bind(_listensock, _port);Sock::Listen(_listensock);_rfds = new struct pollfd[num];for (int i = 0; i < num; i++) ResetItem(i);_rfds[0].fd = _listensock; // 不变了_rfds[0].events = POLLIN;}void Print(){std::cout << "fd list: ";for (int i = 0; i < num; i++){if (_rfds[i].fd != defaultfd)std::cout << _rfds[i].fd << " ";}std::cout << std::endl;}void ResetItem(int i){_rfds[i].fd = defaultfd;_rfds[i].events = 0;_rfds[i].revents = 0;}void Accepter(int listensock){logMessage(DEBUG, "Accepter in");// 走到这里，accept 函数，会不会阻塞？？？1 0// 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有资格检测事件是否就绪// 将新的sock 托管给select！// 将新的sock托管给select的本质，其实就是将sock，添加到fdarray数组中即可！int i = 0;for (; i < num; i++){if (_rfds[i].fd != defaultfd)continue;elsebreak;}if (i == num){logMessage(WARNING, "server if full, please wait");close(sock);}else{_rfds[i].fd = sock;_rfds[i].events = POLLIN;_rfds[i].revents = 0;}Print();logMessage(DEBUG, "Accepter out");}void Recver(int pos){logMessage(DEBUG, "in Recver");// 1. 读取request// 这样读取是有问题的！char buffer[1024];ssize_t s = recv(_rfds[pos].fd, buffer, sizeof(buffer) - 1, 0); // 这里在进行读取的时候，会不会被阻塞？1， 0if (s > 0){buffer[s] = 0;logMessage(NORMAL, "client# %s", buffer);}else if (s == 0){close(_rfds[pos].fd);ResetItem(pos);logMessage(NORMAL, "client quit");return;}else{close(_rfds[pos].fd);ResetItem(pos);logMessage(ERROR, "client quit: %s", strerror(errno));return;}// 2. 处理requeststd::string response = _func(buffer);// 3. 返回response// write bugwrite(_rfds[pos].fd, response.c_str(), response.size());logMessage(DEBUG, "out Recver");}// 1. handler event rfds 中，不仅仅是有一个fd是就绪的，可能存在多个// 2. 我们的select目前只处理了read事件void HandlerReadEvent(){for (int i = 0; i < num; i++){// 过滤掉非法的fdif (_rfds[i].fd == defaultfd)continue;if (!(_rfds[i].events & POLLIN)) continue;// 正常的fd// 正常的fd不一定就绪了// 目前一定是listensock，只有这一个if (_rfds[i].fd== _listensock && (_rfds[i].revents & POLLIN))Accepter(_listensock);else if(_rfds[i].revents & POLLIN)Recver(i);else{}}}void start(){int timeout = -1;for (;;){int n = poll(_rfds, num, timeout);switch (n){case 0:logMessage(NORMAL, "timeout...");break;case -1:logMessage(WARNING, "poll error, code: %d, err string: %s", errno, strerror(errno));break;default:logMessage(NORMAL, "have event ready!");HandlerReadEvent();break;}}}~PollServer(){if (_listensock < 0)close(_listensock);if (_rfds)delete[] _rfds;}private:int _port;int _listensock;struct pollfd *_rfds;func_t _func;};
}

3.epoll

1.接口

epoll_create：创建一个epoll

epoll_ctl：加入准备好的文件描述符

epoll_event：为一个结构体，其中的events表示文件描述符的事件；epoll_data_t为一个联合体。

epfd：表示添加的epoll文件描述符

op：表示添加epoll结构的文件描述符需要进行什么操作

fd：为文件描述符

epoll_wait：捞取准备好的文件描述符进行执行，返回值为可以处理的文件描述符数量

2.实现原理

1.数据一定会从驱动层发送到此操作系统中。

2.先通过epoll_create创建epoll的文件描述符，该文件描述符指向所谓的epoll模型

3.epoll模型中，一旦需要关注某个文件描述符的从套接字处接收，那么通过epoll_ctl能对文件描述符和需要处理的事件一起放入epoll结构体中。由于需要管理，epoll_ctl的过程一并将epoll结构收录到操作系统的epoll模型的红黑树中进行管理。

4.红黑树中的文件描述符如果准备就绪，那么就会通过epoll_wait将epoll的结构插入到准备队列中，那么当启动epoll_wait，就会一连串的进行所加载的文件

3.编程

namespace epoll_ns
{static const int defaultport = 8888;static const int size = 128;static const int defaultvalue = -1;static const int defalultnum = 64;class EpollServer{public:EpollServer(uint16_t port = defaultport, int num = defalultnum): _num(num), _revs(nullptr), _port(port), _listensock(defaultvalue), _epfd(defaultvalue){}void initServer(){// 1. 创建socket_listensock = Sock::Socket();Sock::Bind(_listensock, _port);Sock::Listen(_listensock);// 2. 创建epoll模型_epfd = epoll_create(size);if (_epfd < 0){logMessage(FATAL, "epoll create error: %s", strerror(errno));exit(EPOLL_CREATE_ERR);}// 3. 添加listensock到epoll中！struct epoll_event ev;ev.events = EPOLLIN;ev.data.fd = _listensock;epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock, &ev);// 4.申请就绪事件的空间_revs = new struct epoll_event[_num];logMessage(NORMAL, "init server success");}void HandlerEvent(int readyNum){logMessage(DEBUG, "HandlerEvent in");for (int i = 0; i < readyNum; i++){uint32_t events = _revs[i].events;int sock = _revs[i].data.fd;if (sock == _listensock && (events & EPOLLIN)){//_listensock读事件就绪, 获取新连接std::string clientip;uint16_t clientport;int fd = Sock::Accept(sock, &clientip, &clientport);if (fd < 0){logMessage(WARNING, "accept error");continue;}// 获取fd成功，可以直接读取吗？？不可以，放入epollstruct epoll_event ev;ev.events = EPOLLIN;ev.data.fd = fd;epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &ev);}else if (events & EPOLLIN){// 普通的读事件就绪}else{// 其他事件不进行操作}}logMessage(DEBUG, "HandlerEvent out");}void start(){int timeout = -1;for (;;){int n = epoll_wait(_epfd, _revs, _num, timeout);switch (n){case 0:logMessage(NORMAL, "timeout ...");break;case -1:logMessage(WARNING, "epoll_wait failed, code: %d, errstring: %s", errno, strerror(errno));break;default:logMessage(NORMAL, "have event ready");HandlerEvent(n);break;}}}~EpollServer(){if (_listensock != defaultvalue)close(_listensock);if (_epfd != defaultvalue)close(_epfd);if (_revs)delete[] _revs;}private:uint16_t _port;int _listensock;int _epfd;struct epoll_event *_revs;int _num;};
}