Linux网络编程 --- 高级IO

前言 IO Input&&Output

read && write

1、在应用层read && write的时候，本质把数据从用户层写给OS --- 本质就是拷贝函数

2、IO = 等待  + 拷贝。

等的是：要进行拷贝，必须先判断读写事件成立。读写事件缓冲区空间满了没有，read的时候需要缓冲区有数据。

什么叫做高效的IO呢？

单位事件内我们拷贝的数据越多，越有效率。单位事件内，IO过程中，等的比重越小，IO效率就越高。几乎所有的提高IO效率的策略，本质就是减少等的时间。

一、五种IO模型

1、张三：钓鱼界的新手，一直盯着鱼钩。谁都不理，有鱼上钩，提起鱼竿。阻塞式钓鱼。

2、李四：每隔一小时定盯着鱼漂。不会卡在那里。会看手机什么的。非阻塞式钓鱼。非阻塞轮询

3、王五：鱼竿上放一个铃铛，不会卡在那里，会看手机什么的。鱼钩上有鱼了铃铛响了。信号驱动式IO

以上鱼上钩的概率不如下面的。

4、赵六：一卡车鱼竿。轮询每一个鱼竿。多路复用，多路转接。

以上的IO称为异步IO

5、田七：小王，桶，一个电话，鱼竿。田七就走了，小王开始钓鱼。田七就是钓鱼行为的发起者。田七的钓鱼方式称为异步IO。小王为操作系统。

阻塞IO vs 非阻塞IO  IO = 等待 + 拷贝。非阻塞不用一直等。他们等的方式不同！

同步IO vs 异步IO 张三、李四，王五，赵六都在等待，同步就是参不参与IO。参与就是同步，不参与只是发起IO就是异步。

同步IO 和线程同步？不一样。老婆和老婆饼的关系，没有任何关系。

其中多路复用式最有效率的。异步IO的代码逻辑比较混乱，已经有新的技术代替了，比如协程。

非阻塞接口

1、设置称为非阻塞，如果底层fd数据没有就绪，recv/read/write/send，返回值会以出错的形式返回。

2、a、真的出错了， b、底层没有就绪！

3、我们通过errno区分

如果errno == EWOULDBLOCK，当前不是错了， 而是fd中的数据没有就绪。
 

IO多路转接至select 

IO = 等待 + 拷贝

select：只负责等待，而且可以等待多个文件描述符。 

返回值：n  > 0 有n个fd就绪了

               n == 0 超时，没有错误，但是也没有fd就绪。

               n < 0 等待出错。 

第一个参数 nfds maxfd + 1。最大的等待文件描述符的值 + 1。

struct timeval timeout 给select设置等待方式。就是你设置的是5秒超时，那么就是进程会在select函数阻塞等待5秒，要是5秒中没有事件就绪，他会返回要是5秒中，有事件就绪了，他也会立即返回

timeout = {5，0}；每隔5s，timeout一次。在5s之内没有文件描述符就绪，返回。

             = {0，0}；立马返回，非阻塞的一种。

NULL：阻塞等待。

上面的timeout是输入输出型参数。等待了2秒，如果我们设置的是3秒。select会把这个三秒返回。

  fd_set 内核提供的一种数据类型。我们目前关心的fd上面的事件。读事件成立，写事件成立，异常事件。

fd_set* readfds 输入输出型参数，

输入时：用户告诉内核，我给你的一个或者多个fd，你要帮我关心fd上面的读事件，如果读时间就绪了，你要告诉我。

输出时：内核告诉用户，用户你让我关心的多个fd中，有哪些已经就绪了，用户赶紧读取吧。

从右向左比特位的位置，标识文件描述符编号，比特位的内容0，1是否需要内核关心。

其他的位图也是这样的。

返回时：比特位的内容0 or 1，用户关心的哪些fd，上面的读时间已经就绪了。

fd_set 是一张位图，让用户和内核传递fd是否就绪的信息的!

会有很多的位图操作。所以内核提供了很多的文件描述位图的操作。

 
直接写代码：

#include <iostream>
#include "log.hpp"
#include "Sock.hpp"
#include <sys/select.h>
#include <ctime>
static const int fd_num_max = sizeof(fd_set) * 8;
static const uint16_t defaultport = 8080;
int defaultfd = -1;
class selectserver
{
public:selectserver(uint16_t port = defaultport) : _port(port){// 初始化数组for (int i = 0; i < fd_num_max; i++){fd_array[i] = defaultfd;}}// 初始化服务器，这里应该绑定监听端口，创建套接字void Init(){// 绑定监听套接字。_listensock.Socket();_listensock.Bind(_port);_listensock.Listen();}void Accepter(){// 接收套接字std::string clientip;uint16_t clientport = 0;int sockfd = _listensock.Accept(&clientip, &clientport); // 这里不会被阻塞if (sockfd < 0)return;lg(Info, "accept success, %s: %d, sock fd: %d", clientip.c_str(), clientport, sockfd);// 如果大于0 把新接收的套接字放入到辅助数组中int pos = 1; // fd_array[0] = listensock,所以要从1开始for (; pos < fd_num_max; pos++){// 寻找-1位置if (fd_array[pos] != defaultfd)continue;elsebreak;}if (pos == fd_num_max){lg(Warning, "server is full, close %d now!", sockfd);close(sockfd);}else{fd_array[pos] = sockfd;printFd();}}void Recvr(int fd, int pos){// demochar buffer[1024];ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug?if (n > 0){buffer[n] = 0;std::cout << "get a messge: " << buffer << std::endl;}else if (n == 0){lg(Info, "client quit, me too, close fd is : %d", fd);close(fd);fd_array[pos] = defaultfd; // 这里本质是从select中移除}else{lg(Warning, "recv error: fd is : %d", fd);close(fd);fd_array[pos] = defaultfd; // 这里本质是从select中移除}}void Dispatcher(fd_set &rfds){// 在这里我们需要对事件进行处理，从listensock套接字里连接的文件描述符也需要被我们的select等待，我们需要一个辅助// 数组。for (int i = 0; i < fd_num_max; i++){int fd = fd_array[i];//std::cout << fd << std::endl;if (fd == defaultfd)continue;if (FD_ISSET(fd, &rfds)){if (fd == _listensock.Fd()){Accepter();std::cout << "新文件描述符完毕" << std::endl;}else{// 不是listen套接字那就是普通的套接字进行读。std::cout << "开始交互" << std::endl;Recvr(fd, i);}}}}void Start(){// 这里应该为服务器提供服务。这里我们使用select接口，让文件描述符等待连接。// 服务器是一个死循环。// 在这里我们需要把listensock的文件描述符放入到我们的辅助数组里。int listensock = _listensock.Fd();fd_array[0] = listensock;std::cout << fd_array[0] << std::endl;for (;;){// 这里需要对listensock套接字进行等待。等待是否有链接，有连接说明该文件描述符上有事件就绪。// 传入的参数需要有一个位图，这个位图是输入输出型参数，用户设置需要等待的文件描述符，内核把// 事件就绪的文件描述符放入到其中。每次事件就绪的文件描述符，就会覆盖原来的文件描述符的问题，所以我们// 需要在select之前对位图重新设置。fd_set rfds;FD_ZERO(&rfds);// 边循环边设置我们的文件描述符到位图中，然后求出文件描述符中最大的值。int maxfd = fd_array[0];for (int i = 0; i < fd_num_max; i++){if (fd_array[i] == defaultfd)continue;// 找到文件描述符设置到我们的描述符位图中FD_SET(fd_array[i], &rfds);if (maxfd < fd_array[i]){maxfd = fd_array[i];lg(Info, "max fd update, max fd is: %d", maxfd);}}// 把listensock文件描述符设置到位图中// FD_SET(_listensock.Fd(), &rfds);// 设置等待的事件// struct timeval timeout = {5, 0};struct timeval timeout = {0, 0};int n = select(/*文件描述符中值最大的一个 + 1 _listensock.Fd() + 1*/ maxfd + 1, &rfds, /*不关心写事件，只关心读事件*/ nullptr, nullptr, /*&timeout*/ nullptr);// 判断返回值 = 0代表没有事件就绪//  -1 等待错误//> 0 有n个文件描述符的写事件就绪switch (n){case 0:std::cout << "time out,timeout: " << timeout.tv_sec << timeout.tv_usec << std::endl;break;case -1:std::cerr << "select err" << std::endl;break;default:// 等待成功需要做什么。std::cout << "进入事件派发器" << std::endl;Dispatcher(rfds);break;}}}void printFd(){std::cout << "online list:";for (int i = 0; i < fd_num_max; i++){if (fd_array[i] == defaultfd){continue;}std::cout << fd_array[i] << " ";}std::cout << std::endl;}~selectserver(){_listensock.Close();}private:Sock _listensock;uint16_t _port;int fd_array[fd_num_max];
};

不能直接accept，检测并获取listensock上面的事件，新连接到来，等价于读事件就绪。

select，如果事件就绪，如果上层不处理，select会一直通知。select告诉你就绪了，接下来的一次读取，我们读取fd的时候，不会阻塞。

位图的大小为1024个比特位，所以一次可以等待1024个文件描述符。 

select的优点是：多路转接的方案。一个进程处理多个用户的连接。

select的缺点是：

1、等待的fd是有上限的。

2、输入输出型参数比较多，数据拷贝的频率比较高。

3、输入输出型参数比较多，每次都要对关心的fd进行事件重置。

4、使用第三方数组管理用户fd，用户层需要很多次遍历，内核中检测fd事件就绪，也要遍历。

IO多路转接之poll

#include <iostream>
#include "log.hpp"
#include "Sock.hpp"
#include <poll.h>
#include <ctime>
static const int fd_num_max = 64;
static const uint16_t defaultport = 8080;
int defaultfd = -1;
int non_event = 0;
class pollserver
{
public:pollserver(uint16_t port = defaultport) : _port(port){// 初始化数组for (int i = 0; i < fd_num_max; i++){_fd_events[i].fd = defaultfd;_fd_events[i].events = non_event;_fd_events[i].events = non_event;}}// 初始化服务器，这里应该绑定监听端口，创建套接字void Init(){// 绑定监听套接字。_listensock.Socket();_listensock.Bind(_port);_listensock.Listen();}void Accepter(){// 接收套接字std::string clientip;uint16_t clientport = 0;int sockfd = _listensock.Accept(&clientip, &clientport); // 这里不会被阻塞if (sockfd < 0)return;lg(Info, "accept success, %s: %d, sock fd: %d", clientip.c_str(), clientport, sockfd);// 如果大于0 把新接收的套接字放入到辅助数组中int pos = 1; // fd_array[0] = listensock,所以要从1开始for (; pos < fd_num_max; pos++){// 寻找-1位置if (_fd_events[pos].fd != defaultfd)continue;elsebreak;}if (pos == fd_num_max){lg(Warning, "server is full, close %d now!", sockfd);close(sockfd);}else{_fd_events[pos].fd = sockfd;_fd_events[pos].events = POLLIN;_fd_events[pos].revents = non_event;printFd();}}void Recvr(int fd, int pos){// demochar buffer[1024];ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug?if (n > 0){buffer[n] = 0;std::cout << "get a messge: " << buffer << std::endl;}else if (n == 0){lg(Info, "client quit, me too, close fd is : %d", fd);close(fd);_fd_events[pos].fd= defaultfd; // 这里本质是从select中移除}else{lg(Warning, "recv error: fd is : %d", fd);close(fd);_fd_events[pos].fd= defaultfd; // 这里本质是从select中移除}}void Dispatcher(){// 在这里我们需要对事件进行处理，从listensock套接字里连接的文件描述符也需要被我们的select等待，我们需要一个辅助// 数组。for (int i = 0; i < fd_num_max; i++){int fd = _fd_events[i].fd;//std::cout << fd << std::endl;if (fd == defaultfd)continue;//读事件就绪，且是监听sockif (_fd_events[i].revents & POLLIN){if (fd == _listensock.Fd()){Accepter();std::cout << "新文件描述符完毕" << std::endl;}else{// 不是listen套接字那就是普通的套接字进行读。std::cout << "开始交互" << std::endl;Recvr(fd, i);}}}}void Start(){// 这里应该为服务器提供服务。这里我们使用select接口，让文件描述符等待连接。// 服务器是一个死循环。// 在这里我们需要把listensock的文件描述符放入到我们的辅助数组里。int listensock = _listensock.Fd();for (;;){//把listenfd交给poll管理_fd_events[0].fd = listensock;_fd_events[0].events = POLLIN;_fd_events[0].revents = non_event;//设置超时事件3000毫秒int timeout = 3000;ssize_t n = poll(_fd_events,fd_num_max,timeout);switch (n){case 0:std::cout << "timeout" << std::endl;break;case -1:std::cerr << "poll err" << std::endl;break;default:// 等待成功需要做什么。std::cout << "进入事件派发器" << std::endl;Dispatcher();break;}}}void printFd(){std::cout << "online list:";for (int i = 0; i < fd_num_max; i++){if ( _fd_events[i].fd == defaultfd){continue;}std::cout << _fd_events[i].fd << " ";}std::cout << std::endl;}~pollserver(){_listensock.Close();}private:Sock _listensock;uint16_t _port;struct pollfd _fd_events[fd_num_max];
};

poll只负责等待。

timeout 证书毫秒。

struct pollfd

nfds_t  struct pollfd的数组大小。

将输入输出事件进行了分离。

poll的缺点：

1、遍历，用户层和内核也得遍历。效率问题。

IO多路转接之epoll

1、快速认识epoll的接口

epoll_create：

参数被忽略，设置成大于0就可以。返回值是一个文件描述符。

对epoll新增描述特定的读写事件

epoll_wait：

参数返回已经就绪的fd和事件。

已经就绪的fd的个数。

epoll_event结构体：

epoll_ctl

第一个参数epoll_create的返回值，第二个参数三个选项，第三个参数需要被控制的fd，第四个参数哪些事件被设置。

它不同于 select() 是在监听事件时告诉内核要监听什么类型的事件 , 而是在这里先注册要监听的事件类型 .

第一个参数是 epoll_create() 的返回值 (epoll 的句柄 ).

第二个参数表示动作，用三个宏来表示 .

第三个参数是需要监听的 fd.

第四个参数是告诉内核需要监听什么事 .

2、epoll的原理

用户只需要从就绪队列中获取就绪节点即可。上面的三套机制称为epoll模型。红黑树，就绪队列，回调。epoll模型被统一接入到了文件描述符表里，所以epoll_create的返回值是文件描述符。

epoll_create就是在创建struct_file，也就是创建epoll模型。

epoll_ctl 修改红黑树。

epoll_wait 中的events参数是输出型参数，把就绪队列中的节点，一个一个的放入到我们的epoll_event数组里。

优势：1、检测就绪O(1)，获取就绪O(N)

           2、fd_event没有上限。

           3、这颗红黑树，就是select，poll自己维护的数组。

           4、返回值n，表示有几个fd就绪了，就绪事件是连续的！有返回值个。

3、快速写代码 --- echo server

#pragma once
#include <iostream>
#include <sys/epoll.h>
#include "log.hpp"
class Epoller
{static const int size = 128;
public:Epoller(){//创建epoll模型epollfd = epoll_create(/*这里的参数已经可以忽略了*/size);if(epollfd == -1){lg(Error,"epoll create err %d %s",errno,strerror(errno));}else{lg(Info,"epoll create success %d",epollfd);}}//返回epoll中的就绪队列int waitEpoller(struct epoll_event revents[],int num){int n = epoll_wait(epollfd,revents,num,-1);return n;}//通过系统调用把我们的文件描述符放入到内核中的红黑树中int updataEpoller(int sock,int oper,uint32_t event){int n = 0;if(oper == EPOLL_CTL_DEL){n = epoll_ctl(epollfd,oper,sock,nullptr);if(n!=0){lg(Error,"epoll_ctl err");}}else{//EPOLL_CTL_ADD || EPOLL_CTL_MODstruct epoll_event ev;ev.data.fd = sock;ev.events = event;n = epoll_ctl(epollfd,oper,sock,&ev);if(n != 0){lg(Error,"epoll_ctl err %d %s",errno,strerror(errno));}}return n;}~Epoller(){if(epollfd >= 0){close(epollfd);}}
private:int epollfd;int timeout{3000};
};

#include <iostream>
#include "Sock.hpp"
#include "log.hpp"
#include "nocopy.hpp"
#include "Epoller.hpp"
#include <memory>
int EVENT_IN = (EPOLLIN);
int EVENT_OUT = (EPOLLOUT);
static const uint16_t defaultport = 8080;
static const int defaultfd = -1;
class epollserver : public nocopy
{static const int num = 64;public:epollserver(uint16_t port = defaultport) : _port(port), _listensockfd_ptr(new Sock()), _epoller_ptr(new Epoller()){}bool Init(){// 对套接字进行创建绑定监听_listensockfd_ptr->Socket();_listensockfd_ptr->Bind(_port);_listensockfd_ptr->Listen();return true;}void Accepter(){// 接收套接字std::string clientip;uint16_t clientport = 0;int sockfd = _listensockfd_ptr->Accept(&clientip, &clientport); // 这里不会被阻塞if (sockfd < 0)return;lg(Info, "accept success, %s: %d,", clientip.c_str(), clientport);if (sockfd > 0){//std::cout << "Accept 放入内核中" << std::endl;_epoller_ptr->updataEpoller(sockfd, EPOLL_CTL_ADD, EVENT_IN);lg(Info, "get a new link, client info@ %s:%d %d", clientip.c_str(), clientport,sockfd);}}void Recvr(int fd){// demochar buffer[1024];ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug?if (n > 0){buffer[n] = 0;std::cout << "get a messge: " << buffer << std::endl;std::string echo_string = "server say#";echo_string += buffer;write(fd, echo_string.c_str(), echo_string.size());}else if (n == 0){lg(Info, "client quit, me too, close fd is : %d", fd);close(fd);_epoller_ptr->updataEpoller(fd, EPOLL_CTL_DEL,0); // 这里本质是从select中移除}else{lg(Warning, "recv error: fd is : %d", fd);close(fd);_epoller_ptr->updataEpoller(fd, EPOLL_CTL_DEL,0); // 这里本质是从select中移除}}void Dispatcher(struct epoll_event revents[], int num){for (int i = 0; i < num; i++){int fd = revents[i].data.fd;//std::cout << fd << std::endl;uint32_t events = revents[i].events;if (events & EVENT_IN){if (fd == _listensockfd_ptr->Fd()){Accepter();}else{Recvr(fd);}}}}void Start(){// 我们只需要把listenfd放入到红黑树中_epoller_ptr->updataEpoller(_listensockfd_ptr->Fd(), EPOLL_CTL_ADD, EVENT_IN);struct epoll_event revents[num];// for(int i = 0;i<num;i++)// {//     revents[i].data.fd = defaultfd;//     revents[i].events = 0;// }for (;;){// 在开始阶段我们只有listensock,所以我们可以把listensock等待数据的就绪。// 获取内核就绪事件的fd，获取完成后，查看是否是读事件就绪。int n = _epoller_ptr->waitEpoller(revents, num);if (n > 0){lg(Debug, "event happened, fd is : %d", revents[0].data.fd);Dispatcher(revents, n);}else if (n == 0){std::cout << "time out" << std::endl;}else{std::cerr << "epoll err" << std::endl;}}}~epollserver(){}private:std::shared_ptr<Sock> _listensockfd_ptr;std::shared_ptr<Epoller> _epoller_ptr;uint16_t _port;
};

epoll的工作模式：LT和ET。

水平触发模式 LEVEL Triggered

事件到来，但是上层不处理，一直通知

边缘触发模式Edge Triggered

数据或者连接，从无到有，从有到多，变化的时候，才会通知我们一次。ET的通知效率更高。倒逼程序员，每次通知，都必须把本轮数据全部取走。循环读取，读取出错证明缓冲区没有数据了 ->fd默认是阻塞的 -> ET,所有的fd必须是non_block。不仅如此ET的IO效率也是更高的！通知一次必须把缓冲区的数据全部读走，以为着tcp回向对方通告一个更大的窗口，从而从概率上让对方一次给我发送更多的的数据。

ET vs LT ？ LT可不可以将所有的fd设置成为non_block，然后循环读取呢，通知第一次的时候，就全部取走，不就和ET一样了吗？

LT和ET的本质区别是添加就绪队列的方式，LT是次次都添加，而ET只添加一次。