【网络】第七章-典型IO模型

发表于 | 分类于
字数统计: 4.5k

典型IO模型

IO的种类

  IO模型根据特性可以分为以下几个种类：阻塞IO，非阻塞IO，信号驱动IO，异步IO，多路转接IO。

阻塞IO

  为了IO发起IO调用，若IO条件不满足则一直等待，直到条件具备。

非阻塞IO

  为了IO发起IO调用，若条件不满足则直接报错返回，执行其他指令。之后再次发起IO调用，条件不满足则继续报错返回，条件满足则直接进行数据拷贝后调用返回。
  阻塞与非阻塞的区别在与发起一个调用是否能够立即返回。

信号驱动IO

  自定义IO信号，如果IO条件具备则发送IO信号，收到信号后则打断其他操作进行信号处理，执行IO操作进行数据拷贝，结束后调用返回。

异步IO

  自定义IO信号，发起IO调用，然后让操作系统进行等待条件满足，满足后操作系统进行数据拷贝，拷贝完后通知进程，进程收到后直接处理数据。
  与之对应的是同步的操作同步与异步的区别在于功能的完成是否由自身完成。
  那么是同步好还是异步好呢？答案是视使用场景而定。同步的流程控制更加简单，但是不管是否阻塞都会浪费CPU资源，因此对CPU的利用率不足。而异步对CPU的利用率更高，但是流程控制更加复杂，并且IO调用越多，同一时间占用的空间资源越多。
  从以上IO种类来看，IO效率越来越高，但是流程控制越来越复杂，资源占用也越来越多。

多路转接IO

  多路转接IO对大量描述符进行事件监控，能够让用户只对事件就绪的描述符进行操作。在网络通信中，如果用户仅仅对就绪的描述符进行操作，则流程在一个执行流中就不会阻塞，可以实现在一个执行流中对多个描述符进行并发操作。多路转接IO的实现主要是通过几种多路转接模型实现：select/poll/epoll。

select模型

工作原理

  select模型对大量描述符进行几种事件监控，让用户能够仅仅针对事件就绪的描述符进行操作，对就绪事件的判断主要有以下几个标准：
  1、可读事件：接收缓冲区中数据大小大于等于低水位标记（默认一个字节）。
  2、可写事件：发送缓冲区中空闲空间的大小大于等于低水位标记（默认一个字节）。
  3、异常事件：描述符是否发生了某些异常。

实现流程

  1、用户首先定义事件集合，一共三种事件集合可读/可写/异常，每一个集合实际是一个位图，将某个事件集合中描述符对应的位置1用于标记用户关心该描述符的某些事件。
  2、将集合拷贝到内核进行监控。对集合中所有描述符进行遍历判断，判断是否事件就绪。
  3、若有某个描述符就绪了用户关心的事件，则该返回给用户结果了。返回的时候分别从各个事件集合中将没有就绪该事件的描述符对应的位置0，返回给用户三种表示就绪的描述符事件集合。
  4、用户拿到就绪描述符事件集合后，通过分别遍历三种事件集合找到哪些描述符还在集合中，来判断哪些描述符已经就绪了指定事件，然后进行操作。

接口

1
2
3
4
5
6
7
8
9
10
11
12
int select(int nfds, fd_set *readfds, fd_set *writefds,
           fd_set *exceptfds, struct timeval *timeout);
    //nfds:集合中最大的描述符+1
    //readfds:可读事件集合
    //writefds:可写事件集合
    //exceptfds:异常事件集合
    //timeout:NULL表示永久阻塞，timeout.tv_sec表示限时阻塞
    //返回值:>0表示就绪的描述符个数;==0表示没有描述符就绪;<0监控出错
void FD_CLR(int fd, fd_set *set);//将指定的fd描述符从set集合中删除
int  FD_ISSET(int fd, fd_set *set);//判断某个fd描述符是否在set集合中
void FD_SET(int fd, fd_set *set);//将指定的fd描述符添加到set集合中
void FD_ZERO(fd_set *set);//清空set集合内容

用select模型实现tcp服务器

  按照以下步骤实现：
  1、搭建tcp服务器。
  2、在accept之前创建select监控集合，并且将监听socket添加到可读事件集合中。
  3、进行select监控，当select返回并且有就绪描述符。
  4、若有事件就绪，判断就绪的描述符是否是监听socket，如果是则accept新的socket添加监控，否则recv数据。
  每处理完一遍之后，都要再次对所有描述符进行监控，而每次系统都会将没有就绪的描述符剔除，因此每次监控前我们都得先重新添加一遍所有要监控的描述符。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
//用select模型实现并发服务器                                                                     
//对select进行封装
#include <iostream>
#include <vector>
#include <unistd.h>
#include <string>
#include <stdlib.h>
#include <sys/select.h>
#include "tcp_socket.hpp"
class Select
{
  public:
    Select():_maxfd(-1)
    {
      FD_ZERO(&_rfds);
    }
    //添加要监听的套接字
    bool Add(TcpSocket& sock)
    {
      int fd = sock.GetFd();
      FD_SET(fd, &_rfds);
      _maxfd = _maxfd > fd ? _maxfd : fd;
      return true;
    }
    //监听，list返回就绪的描述符，sec为默认等待时间
    bool Wait(std::vector<TcpSocket>& list, int sec = 3)
    {
      struct timeval tv;
      tv.tv_sec = sec;
      tv.tv_usec = 0;
      fd_set tmp_set = _rfds;//每次定义新的集合进行监控，为了避免原有的监控集合被修改
      int count = select(_maxfd + 1, &tmp_set, NULL, NULL, &tv);
      if(count < 0)
      {
        std::cout << "select error" << std::endl;
        return false;
      }
      else if(count == 0)
      {
        std::cout << "wait timeout" << std::endl;
        return false;
      }
      for(int i = 0; i <= _maxfd; i++)
      {
        if(FD_ISSET(i, &tmp_set))
        {
          TcpSocket sock;
          sock.SetFd(i);
          list.push_back(sock);
        }
      }
      return true;
    }
    //删除不用再监听的套接字
    bool Del(TcpSocket& sock)
    {
      int fd = sock.GetFd();
      FD_CLR(fd, &_rfds);
      for(int i = _maxfd; i >= 0; i--)
      {
        if(FD_ISSET(i, &_rfds))
        {
          _maxfd = i;
          return true;
        }
      }
      _maxfd = -1;
      return true;
    }
  private:
    fd_set _rfds;//读事件描述符集合
    int _maxfd;//最大描述符
};
int main(int argc, char* argv[])
{
  if(argc != 3)
  {
    std::cout << "./tcpselect ip port" << std::endl;
    return -1;
  }
  TcpSocket sock;
  std::string srv_ip = argv[1];
  uint16_t srv_port = atoi(argv[2]);                     
  CHECK_RET(sock.Socket());
  CHECK_RET(sock.Bind(srv_ip, srv_port));
  CHECK_RET(sock.Listen());
  Select s;
  s.Add(sock);
  while(1)
  {
    std::vector<TcpSocket> list;
    if(!s.Wait(list))
    {
      sleep(1);
      continue;
    }
    for(auto clisock : list)
    {
      if(clisock.GetFd() == sock.GetFd())//监听套接字
      {
        TcpSocket socktmp;
        if(sock.Accept(socktmp) == false)
        {
          continue;
        }
        s.Add(socktmp);
      }
      else //通信套接字
      {
        std::string buf;
        if(clisock.Recv(buf) == false)
        {
          s.Del(clisock);
          clisock.Close();
          continue;
        }
        std::cout << "client say:" << buf << std::endl;
        buf.clear();
        std::cin >> buf;
        if(clisock.Send(buf) == false)
        {
          s.Del(clisock);
          clisock.Close();
          continue;
        }
      }
    }
  }
  sock.Close();
}

  以上的实现用到了我们之前实现的tcp_socket.hpp，并将其类中的析构函数自动关闭套接字去掉。之后再用之前实现的tcp_cli.cpp即tcp客户端进行连接测试。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
服务端：
[misaki@localhost AdvancedIO]$ ./tcpselect 192.168.239.128 9000
wait timeout
wait timeout
wait timeout
wait timeout
client say:nihao
wohenhao
wait timeout
client say:heihei
haha
wait timeout
wait timeout
wait timeout
peer shutdown
wait timeout

客户端：
[misaki@localhost AdvancedIO]$ ./tcpclient 192.168.239.128 9000
nihao    
server say: wohenhao
heihei
server say: haha

select优缺点分析

  缺点：
  1、select所能监控的描述符数量是有上线的，FD_SETSIZE = 1024。
  2、每次监控都需要将监控集合拷贝到内核中。
  3、在内核中进行轮询遍历查询，随着描述符的增多性能降低。
  4、返回的是就绪集合，需要用户自己进行判断才能对描述符进行操作。
  5、每次返回都会清空未就绪描述符，因此每次监控都需要重新添加到集合中。
  优点：
  1、POSIX标准，支持跨平台。

poll模型

工作原理

  poll采用事件结构的方式对描述符进行事件监控，只需要一个事件结构数组，将要响应的描述符提交到数组的每一个结构节点的fd中，以及将用户关心的事件添加到响应节点events中。即可进行监控。

接口

1
2
3
4
5
6
7
8
9
10
11
12
int poll(struct pollfd *fds, nfds_t nfds, int timeout);
//以下这个结构体用于告诉系统需要监控哪些描述符以及所需监控的状态
struct pollfd
{
    int   fd;         /* file descriptor */
    short events;     //事件，POLLIN-可读/POLLOUT-可写
    short revents等    //这次监控返回之后这个描述符就绪的状态
}
//fds：struct pollfd数组
//nfds：数组的大小
//timeout：超时时间，单位ms
//返回值：小于0出错，等于0超时，大于零有描述符就绪

使用流程

  1、用户定义一个事件结构数组，然后将关心的描述符以及事件添加到数组中。
  2、将数组拷贝到内核进行监控。在内核中会轮询遍历监控。
  3、当事件数组中有描述符事件就绪/等待超时则poll返回，poll返回时将每个描述符就虚的事件添加到相应节点revents中。
  4、用户对数组中的每个节点的revents进行判断之后进行操作。

优缺点分析

  缺点：
  1、无法跨平台。
  2、依然需要将监控的描述符事件数组拷贝到内核中。
  3、
在内核中同样是轮询遍历监控，性能会随着描述符的增多而下降。
  4、只是将就绪的事件放到了结构的revents中，需要用户轮询查找就绪的描述符。
  优点：
  1、没有描述符监控的数量上限。
  2、采用事件结构进行监控，简化了select三种事件集合的操作流程。
  3、不需要每次重新向集合中添加事件数组。

epoll模型

  epoll模型是为了处理大批量句柄而做了改进的poll，它几乎具备了所需的一切优点，是Linux下性能最好的多路I/O模型。

接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
int epoll_create(int size);
//在内核中创建eventpoll结构，结构中主要信息：双向链表，红黑树，具体作用看下文。返回一个操作句柄描述符
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
//向内核中的eventpoll结构中的红黑树添加用户关心的描述符事件结构
//epfd:epoll操作句柄
//op:工作模式，EPOLL_CTL_ADD/EPOLL_CTL_MOD/EPOLL_CTL_DEL
//fd:用户关心的描述符
//event:对于fd描述符所要监控的事件，结构体参考如下
struct epoll_event {
    uint32_t     events; //要监听的事件EPOLLIN可读/EPOLLOUT可写
    epoll_data_t data; //就绪的描述符，这个值通常与fd一致
};

int epoll_wait(int epfd, struct epoll_event *events,
               int maxevents, int timeout);
//epfd:epoll操作句柄
//events:就绪事件数组，返回就绪的描述符和对应的描述符
//maxevents:最大节点数
//timeout:超时事件，单位ms，-1阻塞，0非阻塞

工作原理

  epoll监控是一个异步阻塞操作。对描述符的监控由操作系统完成，当描述符就绪之后，则将就绪的描述符对应的epoll_event结构添加到双向链表list中，而当前进程只是每隔一段时间判断以下list是否为空，即可知道是否有描述符就绪。
  操作系统完成监控，对于每一个描述符所关心的事件都定义了一个事件回调，当描述符就绪事件的时候就会调用回调函数，这个回调函数负责将事件结构信息即struct epoll_event添加到双向链表中。
  epoll_wait会自动检测list双向链表，检测到链表list不为空，表示有就绪事件，则将这个链表中这些epoll_event放到用户的events数组中返回出去。

用epoll模型实现tcp服务器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
#include <iostream>                                                      
#include <vector>                            
#include <string>              
#include <sys/epoll.h>
#include <unistd.h>   
#include "tcp_socket.hpp"    
#define MAX_EPOLL 1024
class Epoll
{                               
  public:                    
    Epoll()
      :_epfd(-1)
    {          
     
    }                                      
    ~Epoll()
    {             
                                             
    }  
    bool Init()            
    {                                         
      _epfd = epoll_create(MAX_EPOLL);
      if(_epfd < 0)
      {  
        std::cerr << "create epoll error" << std::endl;
        return false;
      }   
      return true;
    }                   
    bool Add(TcpSocket& sock)         
    {    
      struct epoll_event ev;
      int fd = sock.GetFd();
      ev.events = EPOLLIN;
      ev.data.fd = fd;                                 
      int ret = epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &ev);
      if(ret < 0)      
      {                               
        std::cerr << "append monitor error" << std::endl;
        return false;     
      }
      return true;
    }  
    bool Del(TcpSocket& sock)
    {
      int fd = sock.GetFd();
      int ret = epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, NULL);
      if(ret < 0)
      {
        std::cerr << "remove monitor error" << std::endl;
        return false;
      }
      return true;
    }
    bool Wait(std::vector<TcpSocket>& list, int timeout = 3000)
    {
      struct epoll_event evs[MAX_EPOLL];
      int nfds = epoll_wait(_epfd, evs, MAX_EPOLL, timeout);
      if(nfds < 0)
      {
        std::cerr << "epoll monitor error" << std::endl;
        return false;
      }
      else if(nfds == 0)
      {
        std::cerr << "epoll monitor timeout" << std::endl;
        return false;
      }
      for(int i = 0; i < nfds; i++)
      {
        int fd = evs[i].data.fd;
        TcpSocket sock;
        sock.SetFd(fd);
        list.push_back(sock);
      }
      return true;
    }
  private:
    int _epfd;
};
int main(int argc, char* argv[])
{
  if(argc != 3)
  {                                                                         
    std::cout << "./tcp_epoll ip port" << std::endl;
    return -1;
  }
  std::string srv_ip = argv[1];
  uint16_t srv_port = atoi(argv[2]);
  TcpSocket lst_sock;
  CHECK_RET(lst_sock.Socket());
  CHECK_RET(lst_sock.Bind(srv_ip, srv_port));
  CHECK_RET(lst_sock.Listen());
  Epoll e;
  CHECK_RET(e.Init());
  CHECK_RET(e.Add(lst_sock));
  while(1)
  {
    std::vector<TcpSocket> list;
    if(e.Wait(list) == false)
    {
      sleep(1);
      continue;
    }
    for(size_t i = 0; i < list.size(); i++)
    {
      //监听套接字
      if(list[i].GetFd() == lst_sock.GetFd())
      {
        TcpSocket cli_sock;
        if(lst_sock.Accept(cli_sock) == false)
        {
          continue;
        }
        CHECK_RET(e.Add(cli_sock));
      }
      else 
      {
        std::string buf;
        if(list[i].Recv(buf) == false)
        {
          list[i].Close();
          continue;
        }
        std::cout << "Client Say:" << buf << std::endl;
        buf.clear();                                                       
        std::cin>> buf;
        if(list[i].Send(buf) == false)
        {
          list[i].Close();
          continue;
        }
      }
    }
  }
  lst_sock.Close();
}      


//服务端
[misaki@localhost AdvancedIO]$ ./tcpepoll 192.168.239.128 9000
epoll monitor timeout
epoll monitor timeout
epoll monitor timeout
Client Say:nihao
heihei
epoll monitor timeout
Client Say:haha
xixi
peer shutdown
epoll monitor timeout

//客户端
[misaki@localhost AdvancedIO]$ ./tcpclient 192.168.239.128 9000
nihao
server say: heihei
haha
server say: xixi

  在事件结构体中events字段还有另外几种事件触发模式EPOLLLT水平触发，EPOLLET边缘触发。
  水平触发对于可读事件来说，只要接收缓冲区中的数据大小高于低水位标记就会触发事件。对于可写事件来说，只要发送缓冲区中剩余空间高于低水位标记时就会触发事件。总结来说就是只要缓冲区满足可读或可写要求就会触发事件。举个例子，如果你在读取缓冲区，一次没有读完，只要缓冲区中还有数据那么下一次epoll会继续触发事件告诉你有东西可读。
  边缘触发对于可读和可写事件来说都是只有当缓冲区中内容改变即新数据接收到接收缓冲区或发送缓冲区数据被发送走时才会触发时间。同样举个例子，如果想要读取接收缓冲区的内容，一次没有读取完，虽然缓冲区中还有数据但是下一次epoll不会触发事件让你继续读取，知道有新的数据接收到接收缓冲区中，此时你可以连着上次的数据和新数据一起读取。
  水平触发和边缘触发没有明确的性能差距，但是边缘触发一次读写只触发一次，确实比水平触发要触发的次数少，系统就不会触发一些你不关心的就绪文件描述符。因此有时使用边缘触发更好一些，但是为了一次触发就能把缓冲区中数据全部读完需要循环读取数据，知道读完位置，又为了不造成阻塞要将描述符设置为非阻塞，然后循环读取数据直到EAGAIN错误就表示一次读完了。其实实际使用中都是非阻塞循环进行数据读取，毕竟在不清楚数据到底有多少的情况下，要想一次读完就只能采用以上方法。

epoll优缺点分析

  优点：
  1、采用事件结构方式监控，简化多个监控集合的操作流程。
  2、没有所能监控的描述符数量上限。
  3、epoll监控的事件只需要向内核拷贝一次，不需要每次都拷贝。
  4、监控采用异步阻塞，在内核中进行事件回调方式监控，因此性能不会随着描述符的增多而降低。
  5、直接返回就绪事件结构，用户可以通过就绪事件结构中的描述符直接操作，不需要进行遍历判断。
  缺点：
  1、不支持跨平台。

多路转接IO适用场景

  多路转接模型适用于对大量描述符进行监控，但是同一时间只有少量活跃的场景。多线程/多进程的并发处理时比较高效且公平的。但是多路转接模型的并发是轮询处理的，一个处理完才会处理下一个，如果活跃描述符很多，则会导致后面的描述符等待时间过长。
  因此使用epoll往往是用其进行活跃判断，当描述符活跃再将其放到线程池中进行公平处理。这样的搭配才是较为完美的。

-------------本文结束感谢您的阅读!-------------
记录学习每一分，感谢您的赞助