网站标签的作用,门头沟做网站,免费门户网站开发,深圳网站建设李天亮概念背景 IO的本质就是输入输出 刚开始学网络的时候#xff0c;我们简单的写过一些网络服务#xff0c;其中用到了read#xff0c;write这样的接口#xff0c;当时我们用的就是基础IO#xff0c;高级IO主要就是效率问题。 我们在应用层调用readwrite的时候…概念背景 IO的本质就是输入输出 刚开始学网络的时候我们简单的写过一些网络服务其中用到了readwrite这样的接口当时我们用的就是基础IO高级IO主要就是效率问题。 我们在应用层调用readwrite的时候本质就是把数据由用户层写给OS所以这些函数本质就是拷贝函数。 在应用层尤其是在网络通信的时候我们在IO的时候其实大部分的时间都是在做 “等待” 的操作只有数据来了才进行拷贝所以IO 等待 拷贝。但是大部分时间都花在了 等 上面所以IO本质就是 等 拷贝。 因此要进行拷贝我们必须先判断条件是否成立在网络IO这里我们叫做读写事件是否就绪。 在了解了IO的本质之后我们就理解什么是高效的IO了高效的IO在IO过程中等所占的比重越小IO的效率越高。 所以几乎所有提高IO效率的策略都是为了降低等待的时间。 我们之前写过简单的线程池为什么线程池效率高呢很大的原因都是线程们的等待是并行的减少了等待所占的时间比重所以效率高。
五种IO模型
介绍 1.阻塞IO也是最常见的IO模型在内核数据准备好之前系统调用一直会阻塞式的等待。所有的套接字默认都是阻塞方式。
2.非阻塞IO如果内核数据尚未准备好那么系统调用会直接返回返回EWOULDBLOCK错误码。非阻塞IO往往需要程序猿通过轮询的方式不断尝试读写文件描述符对CPU来说是比较大的浪费。
3.信号驱动IO内核数据准备好的时候使用SIGIO的信号通知应用程序进行IO操作。
4.多路转接/复用IO以阻塞等待的方式同时等待多个文件描述符的就绪状态。
以上的IO都是同步IO。
5.异步IO由内核在数据拷贝完成时通知应用程序。注意不是数据准备好而是拷贝完成可以理解为用户给OS一个缓冲区OS一有数据就自动将数据拷贝到缓冲区里等缓冲区满了就通知应用程序。也就是进程并不参与IO具体的过程。
阻塞IOvs非阻塞IO
因为IO 等 拷贝所以阻塞IO和非阻塞IO的IO效率是一样的但是因为非阻塞IO可以在等待失败返回后做一下其他事情所以效率会比阻塞IO的高一些。因此阻塞IO和非阻塞IO的区别只是等的方法不同而已。
同步IOvs异步IO
是否是同步IO就看特定的某个进程是否参与了IO这个参与可以是参与了等也可以是参与了拷贝。只要参与了就是同步IO。
而异步IO的本质是不参与IO它只是发起IO最后拿结果就可以了。
另外同步IO中的同步与线程同步里的同步是两个完全不同的东西。同步IO指的是IO层面上的概念而线程同步的同步是指线程之间谁先运行谁要等待的问题。
这些IO模型里面效率最高的就是多路复用IO。异步IO往往会带来IO混乱。
非阻塞IO/fcntl
先看看recvread也可以 我们看到recv前三个参数跟read一模一样但是多了个flags 看看read的 以前flags我们默认都是设置为0代表是阻塞式读取我们可以进行设置flags来设置非阻塞。
除此之外我们还可以用fcntl 来直接对文件描述符进行设置 其中cmd可以传入以下 复制一个现有的描述符cmdF_DUPFD. 获得/设置文件描述符标记(cmdF_GETFD或F_SETFD). 获得/设置文件状态标记(cmdF_GETFL或F_SETFL). 获得/设置异步I/O所有权(cmdF_GETOWN或F_SETOWN). 获得/设置记录锁(cmdF_GETLK,F_SETLK或F_SETLKW) 只要文件描述符是非阻塞的那么将来通过这个文件描述符来调用read/recv/write/send这样的接口就都是非阻塞的。 比如我们可以设置一个函数它可以将传入的文件描述符设置成非阻塞的 void SetNoBlock(int fd)
{ int fl fcntl(fd, F_GETFL); if (fl 0) { perror(fcntl);return; }fcntl(fd, F_SETFL, fl | O_NONBLOCK);
} 如果我们设置成为非阻塞如果底层fd数据没有就绪recv/read/write/send就会以出错的形式返回。所以如果如果有错误形式返回了它有两种情况 1.真的出错了。 2.底层没有就绪。 可以通过errnor进行区分。底层没有就绪的错误码是11EWOULDBLOCK。 例当返回值小于0时 多路转接之select 多路转接也是五种IO模型中的一种。
以前我们学到的所有接口都是既做等又做拷贝的所以效率就很难提高。 而select只可以做等的工作函数原型
int select(int nfds, fd_set *readfds, fd_set *writefds,fd_set *exceptfds, struct timeval *timeout); 新标准只要包一个头文件就可以了就是
#include sys/select.h
select只负责等并且可以一次等待多个fd。
程序会在select这里等待直到监视的文件描述符有一个或多个发生了状态的改变。
参数解释
nfds就是需要监视的最大文件描述符 1。
后面的readfdswritefdsexceptfds其实本质都是位图所对应的就是需要检测的文件描述符的可读/可写/是否出错的集合。虽然fd_set也是一个结构体但是里面是位图。它们都是输入输出型参数我们也可以同时传入多个fd_set表示我们关心多个状态的就绪。
对于输入输出理解
1.输入时是用户告诉内核要内核关心哪些fd哪些fd就绪了要及时告诉用户。
2.输出时是内核告诉用户哪些fd就绪了要用户赶快处理。
因为是位图结构的那么 比特位的位置代表的是文件描述符的编号。
比特位的内容代表是否需要内核关心。
所以fd_set本质是一张位图它是让用户与内核之前传递fd是否就绪的信息的。
所以我们在使用select的时候会先有大量的位图操作。 位图的操作接口
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的全部位
timval是一个结构体用来设置select的等待时间。
它的定义如下
struct timeval
{
__time_t tv_sec; /* Seconds. */
__suseconds_t tv_usec; /* Microseconds. */
};
这个tv_sec也就是秒tv_usec就是微秒。
我们在设置 timeout的时候一般是 timeout {5,0}其中这个5代表的含义就是5秒timeout一次如果是{0.0}那么就是立马返回没有阻塞。或者我们也可以直接将timeout设置成为NULL表示阻塞等待。这个timeval也是输入输出型参数。
即便还在等待时间里只要有fd就绪了就立马返回假设是在2秒的时候就绪了然后返回那么timeout里面的值就变成了{5 - 2 3,0}也就是{3,0}就是剩余的返回时间。如果一个fd都没有就绪那么会在timeout被减到0的时候自动返回一次那么此时的timeout已经是{0,0}了也就是非阻塞了。因此timeout可能要进行周期的重复设置。
关于select的返回值
1.n 0有n个fd就绪了。
2.n 0超时了没有出错但是也没有fd就绪。
3.n 0出错了。
所以现在我们写网络服务的时候就不能直接accept了因此accept也是检测并获取listensock上面的事件期间也是在阻塞式等待新连接到来等价于读事件就绪了。
所以我们可以先用select进行等待如果select告诉用户就绪了那么接下来的一次读取fd的时候就不会被阻塞了因为已经不需要等了只要读了。
如果事件就绪了上层不处理那么select会一直通知用户。
简单实现基于select的多路转接的echo服务器
SelectServer.hpp
#pragma once#include iostream
#include sys/select.h
#include sys/time.h
#include Socket.hpp
#include Log.hppusing namespace std;static const int defaultport 8080;
// *8是因为一字节有8个比特位一个比特位就可以表示一个fd
static const int fd_num_max (sizeof(fd_set) * 8);
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;}}bool Init(){_listensock.Socket();_listensock.Bind(_port);_listensock.Listen();return true;}void Accepter(){// 说明连接事件就绪了std::string clientip;uint16_t clientport 0;;int sock _listensock.Accept(clientip,clientport); // 这里已经不会再阻塞了select已经帮我们等完了if(sock 0) return;lg(Info,accept success,%s : %d,sock fd : %d,clientip.c_str(),clientport,sock);//这里下标是从1开始的因为0号下标是listensockfd的。int pos 1;for(;pos fd_num_max; pos) // 这里是第二个循环是为了将新的fd插入到用户维护的数组中{if(fd_array[pos] ! defaultfd)continue;else break;}if(pos fd_num_max){lg(Warning,server is full,close %d now!,sock);close(sock);}else {fd_array[pos] sock;PrintFd();}}void Recver(int fd,int pos){// 这里只是一个demo我们并没有处理都上来的数据是否是完整的报文char buffer[1024];ssize_t n read(fd,buffer,sizeof(buffer) - 1); //数据不一定完整 ,但是这里不会阻塞了if(n 0){buffer[n] 0;cout get a messge : buffer 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) // 任务派发器{for(int i 0; i fd_num_max; i) // 这里是第三个循环了{int fd fd_array[i];if(fd defaultfd)continue;if(FD_ISSET(fd,rfds)){if(fd _listensock.Fd()) // 说明是新链接就绪了{Accepter(); // 连接管理器}else // 说明某个链接的读事件就绪了{Recver(fd,i);}}}}void Start(){int listensock _listensock.Fd();fd_array[0] listensock;while(true){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,maxfd updata,maxfd is : %d,maxfd);}}// 这里不能直接accept了read也是因为是单进程的如果直接用// 这里直接就因为一个链接可能就阻塞了struct timeval timeout {1,0}; // 因为是输入输出型参数所以要周期的重复设置int n select(maxfd 1,rfds,nullptr,nullptr,timeout);if(n 0){// 说明有事件就绪了cout get a new link endl;// 我们也不知道是一个就绪了还是多个就绪了// 而且这里也不知道是新链接来了还是读事件就绪了Dispatcher(rfds); }else if(n 0){cout time out ... endl;}else if(n -1){cout select error!!! endl;}}}void PrintFd(){cout online fd list : ;for(int i 0; i fd_num_max; i){if(fd_array[i] defaultfd)continue;cout fd_array[i] ;}cout endl;}~SelectServer(){_listensock.Close();}
private:Sock _listensock;uint16_t _port;int fd_array[fd_num_max]; // 这是用户维护的数组。
};
select那里rfds也是输入输出型参数它的输出就是告诉我们哪些fd就绪了将位图对应的位置修改成1没有就绪的修改成0。 比如一开始传入的是1111 1111假如只有一个fd就绪了那么它返回的就是0000 0001可见其他的只是会被置为0的如果我们还要对其他的fd继续关心就需要我们在用户层维护一个数组表示我们要关心哪些fd。
因为这里我们写的比较简单只关心读事件所以也只维护了一个读事件的数组将来如果要关心写或者异常事件那么还需要相应的增加数组。
我们每次在select前都要重新对位图进行设置这个时候就要用循环来遍历我们维护的数组将它设置进位图当中另外select传入的fd表示的是我们需要关心的最大fd 1那么也就是需要我们每次循环都要知道当前的maxfd是多少所以在每次在select前还要顺便在这个循环中更新出maxfd。
另外fd_set位图的比特位最多是1024个也就是最多能表示1024个fd。
timeval参数那里我们也可以传入nullptr这样就是阻塞式等待了。
在任务派发器那里因为就绪的不一定只有1个fd因此我们又加上了一个循环以便将这次就绪的fd全部处理。
Main.cc
#include memory
#include SelectServer.hppint main()
{std::unique_ptrSelectServer svr(new SelectServer());svr-Init();svr-Start();return 0;
}
之间一直使用的Socket.hpp简单加入了setsockopt顺便复习一下
Socket.hpp
#pragma once#include iostream
#include string
#include unistd.h
#include cstring
#include sys/types.h
#include sys/stat.h
#include sys/socket.h
#include arpa/inet.h
#include netinet/in.h
#include Log.hppenum
{SocketErr 2,BindErr,ListenErr
};const int backlog 10;class Sock
{
public:Sock(){}~Sock(){}public:void Socket(){sockfd_ socket(AF_INET,SOCK_STREAM,0);if(sockfd_ 0){lg(Fatal,socker error,%s: %d,strerror(errno),errno);exit(SocketErr);}int opt 1;setsockopt(sockfd_,SOL_SOCKET,SO_REUSEADDR | SO_REUSEPORT,opt,sizeof(opt)); // 防止服务器挂掉后不能立即重启。}void Bind(uint16_t port){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(sockfd_,(struct sockaddr*)local,sizeof(local)) 0){lg(Fatal,bind error, %s: %d,strerror(errno),errno);exit(BindErr);}}void Listen(){if(listen(sockfd_,backlog) 0){lg(Fatal,listen error,%s: %d,strerror(errno),errno);exit(ListenErr);}}int Accept(std::string *clientip,uint16_t *clientport) // 两个输出型参数{struct sockaddr_in peer;socklen_t len sizeof(peer);int newfd accept(sockfd_,(struct sockaddr*)peer,len);if(newfd 0){lg(Warning,accept error,%s: %d,strerror(errno),errno);return -1;}// 开始准备输出ip和端口号char ipstr[64];inet_ntop(AF_INET,peer.sin_addr,ipstr,sizeof(ipstr));*clientip ipstr;*clientport ntohs(peer.sin_port);return newfd;}bool Connect(const std::string ip,const uint16_t port){struct sockaddr_in peer;memset(peer,0,sizeof(peer));peer.sin_family AF_INET;peer.sin_port htons(port);inet_pton(AF_INET,ip.c_str(),(peer.sin_addr));int n connect(sockfd_,(struct sockaddr*)peer,sizeof(peer));if(n -1){std::cout connect to ip : port error std::endl;return false;}return true;}void Close(){close(sockfd_);}int Fd(){return sockfd_;}
private:int sockfd_;
};
select总结
优点
1.就如实现的那样select已经是多路转接的一种方案了我们可以把所有的fd交给select去等待就绪后通知我们去处理这样的话即便是单进程也可以高效的处理多个fd的读写事件。
缺点
1.等待的fd数量是有上限的centos中是1024个。这不是操作系统的限制而是它的接口设计就是这样的。这也是最大的缺点
2.输入输出型参数比较多用户到内核内核到用户数据拷贝频率较高。
3.也是因为输入输出型参数比较多每次都要对关心的fd进行事件重置。
4.对于用户层使用第三方数组来管理用户的fd用户层需要很多次遍历。对于内核层检测fd事件就绪也要遍历。
所以总的看来select在所有的IO中效率算高的毕竟它是多路转接的一种方案但是它实际上会有很多的拷贝和遍历这也是比较影响效率的。
一般select在小型的场景下可以简单使用。 多路转接之poll 学完select之后我们明白了select有很多的短板我们看poll是如何解决的。 这里的timeout跟select不一样这里是一个整形代表的单位是毫秒。
其中这个pollfd的结构体 看描述就可以知道events 是我们需要内核关心哪些事件而revents则是内核告诉用户哪些事件就绪了而fd则是我们告诉内核是哪个文件描述符需要被关心。
所以poll这里我们第一个参数传入的是一个结构体数组第二个参数是结构体数组的大小。
因此poll最大的特点的是将输入和输出事件进行了分离
关于events的传参这里也设计成了位图的样子 通过传入不同的参来表示关心的事件。这些大写的都是宏。这里我们可以重点看POLLIN和POLLOUT分别代表的读和写。
简单讲代码替换成基于poll实现的服务器
PollServer.hpp
#pragma once#include iostream
#include poll.h
#include time.h
#include Socket.hpp
#include Log.hppusing namespace std;static const int defaultport 8080;
static const int fd_num_max 128; // 这里直接设置成65535也没问题
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){_event_fds[i].fd defaultfd;_event_fds[i].events non_event;_event_fds[i].revents non_event;}}bool Init(){_listensock.Socket();_listensock.Bind(_port);_listensock.Listen();return true;}void Accepter(){// 说明连接事件就绪了std::string clientip;uint16_t clientport 0;;int sock _listensock.Accept(clientip,clientport); // 这里已经不会再阻塞了select已经帮我们等完了if(sock 0) return;lg(Info,accept success,%s : %d,sock fd : %d,clientip.c_str(),clientport,sock);int pos 1;for(;pos fd_num_max; pos) // 这里是第1个循环是为了将新的fd插入到用户维护的数组中{if(_event_fds[pos].fd ! defaultfd)continue;else break;}if(pos fd_num_max){lg(Warning,server is full,close %d now!,sock);close(sock);}else {_event_fds[pos].fd sock;_event_fds[pos].events POLLIN; // 如果要同时关心读写 POLLIN | POLLOUT_event_fds[pos].revents non_event;PrintFd();}}void Recver(int fd,int pos){// 这里只是一个demo我们并没有处理都上来的数据是否是完整的报文char buffer[1024];ssize_t n read(fd,buffer,sizeof(buffer) - 1); //数据不一定完整 ,但是这里不会阻塞了if(n 0){buffer[n] 0;cout get a messge : buffer endl;}else if(n 0) // 对方关闭连接了{lg(Info,client quit , me too, close fd is : %d,fd);close(fd);_event_fds[pos].fd defaultfd; // 这里本质是从select中移除}else {// 读出错了也要关闭lg(Warning,recv error : fd is : %d,fd);close(fd);_event_fds[pos].fd defaultfd; // 本质还是从select中移除}}void Dispatcher() // 任务派发器{for(int i 0; i fd_num_max; i) // 这里是第二个循环了{int fd _event_fds[i].fd;if(fd defaultfd)continue;if(_event_fds[i].revents POLLIN) {if(fd _listensock.Fd()) // 说明是新链接就绪了{Accepter(); // 连接管理器}else // 说明某个链接的读事件就绪了{Recver(fd,i);}}}}void Start(){_event_fds[0].fd _listensock.Fd();_event_fds[0].events POLLIN;int timeout 3000; // 代表的是3swhile(true){int n poll(_event_fds,fd_num_max,timeout);if(n 0){// 说明有事件就绪了cout get a new link endl;// 我们也不知道是一个就绪了还是多个就绪了// 而且这里也不知道是新链接来了还是读事件就绪了Dispatcher(); }else if(n 0){cout time out ... endl;}else if(n -1){cout select error!!! endl;}}}void PrintFd(){cout online fd list : ;for(int i 0; i fd_num_max; i){if(_event_fds[i].fd defaultfd)continue;cout _event_fds[i].fd ;}cout endl;}~PollServer(){_listensock.Close();}
private:Sock _listensock;uint16_t _port;struct pollfd _event_fds[fd_num_max]; // 依旧是由用户维护的数组//int fd_array[fd_num_max]; // 这是用户维护的数组。
};
Main.cc和makefile简单改改就好了。
这样实现的话我们发现不仅接口变简单了而且每次poll前相比select少了一次循环。
更重要的是fd的数量上限不再受接口限制了。
select已经比较古老了如果要在select和poll二选一建议还是选择poll除非平台也很古老只支持select。
我们很轻易的就将select替换成poll了说明两者还是有很多相似的地方了。
主要还是遍历的问题虽然少了一次但是依旧需要在用户层遍历在内核层也是。如果数组很大每次遍历的消耗也大效率就不一定像原本想的那样高。
这个已经不是功能方面的问题了这是效率方面的问题。
