标签搜索

目 录CONTENT

文章目录

I_O多路转接之select.md

小小城
2021-08-22 / 0 评论 / 0 点赞 / 8 阅读 / 5,355 字 / 正在检测是否收录...
温馨提示:
本文最后更新于 2022-05-02,若内容或图片失效,请留言反馈。部分素材来自网络,若不小心影响到您的利益,请联系我们删除。

I/O多路转接之select

@[toc]

一、五种IO模型

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

在这里插入图片描述

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

在这里插入图片描述

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

在这里插入图片描述

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

在这里插入图片描述

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

在这里插入图片描述

高级IO相关概念:

  • ==同步通信 vs 异步通信==

  • ==同步和异步关注的是消息通信机制==(注意和线程同步区分)

  • 所谓同步,就是在发出一个调用时,在没有得到结果之前,该调用就不返回. 但是一旦调用返回,就得 到返回值了;换句话说,就是由==调用者主动等待==这个调用的结果;

  • 异步则是相反,调用在发出之后,这个调用就直接返回了,所以没有返回结果; 换句话说,当一个异步过程调用发出后,==调用者不会立刻得到结果==; 而是在调用发出后,被调用者通过状态、通知来通知调用者,或通过回调函数处理这个调用

  • ==阻塞 vs 非阻塞==

  • 阻塞和非阻塞关注的是程序在等待调用结果(消息,返回值)时的状态.

  • 阻塞调用是指调用结果返回之前,当前线程会被挂起. 调用线程只有在得到结果之后才会返回.

  • 非阻塞调用指在不能立刻得到结果之前,该调用不会阻塞当前线程

二、I/O多路转接之select原理

  •  1.函数原型
int select(int nfds, fd_set *readfds, fd_set *writefds,
                  fd_set *exceptfds, struct timeval *timeout);
参数作用
nfds是需要监视的最大文件描述值+1
fd_set *readfds需要检测的可读文件描述符的集合,
fd_set *writefds需要检测的可写文件描述符的集合
fd_set *exceptfds需要检测的异常文件描述符的集合;
struct timeval *timeout为结构timeval,用来设置select()的等待时间
  • 函数返回值
1.执行成功则返回文件描述符状态已经改变的个数;

2.如果返回0代表在描述词状态改变之前已经超过 timeout时间,没有返回;

3.当有错误发生时则返回-1,错误原因存于error,此时参数readfd、writefd、exceptfd和timeout的值变成不可预测。
  •  2.理解fd_set结构

  • 其实这个结构就是一个整型数组,更严格的说,是一个“位图”,使用位图中对应的位表示要监视的文件描述符

  • 用来消除描述词组 set中相关fd的位

void FD_CLR(int fd,fd_set *set);
  • 用来测试描述词组set中的相关fd的位是否为真;
void FD_ISSET(int fd,fd_set *set);
  • 用来设置描述词组set中相关的位;
void FD_SET(int fd,fd_set *set);
  • 用来消除描述词组set的全部位;
void FD_ZERO(int fd,fd_set *set);
  •  3.socket就绪条件

  • 读就绪:

      1.socket内核中,接收缓冲区中的字节数,大于等于低水位标记SO_RCVLOWAT。此时就可以无阻塞的读该文件描述符,并且返回值大于0;
    
      2.socket TCP通信中,对端关闭连接,此时对该socket读,则返回0(四次挥手);
    
      3.监听的socket上有新的连接请求;
    
      4.socket上有未处理的错误。
    
  • 写就绪

      1.socket内核中,发送缓冲区中的可用字节数(发送缓冲区的空闲位置大小),大于等于低水位标记SO_RCVLOWAT。此时可以无阻塞的写,并且返回值大于0;
    
      2.socket的写操作被关闭。对一个写操作被关闭的socket进行写操作,会触发SIGPIPE信号;
    
      3.socket使用非阻塞connect连接成功或失败之后;
    
      4.socket上有未读取的错误;
    
  • 异常就绪:

      socket上带外数据。关于带外数据,和TCP紧急模式相关(TCP协议头中的紧急指针字段)
    
  •  4.理解select执行过程

理解select模型的关键在于理解fd_set,为说明方便,取fd_set长度为1字节,fd_set中的每一bit可以对应一个文件描述符fd。则1字节长的fd_set最大可以对应8个fd.

  • (1)执行fd_set set; FD_ZERO(&set);则set用位表示0000,0000。
  • (2)若fd=5,执行FD_SET(fd,&set);后set变为0001,0000(第5位置为1)
  • (3)若再加入fd=2,fd=1,则set变为0001,0011
  • (4)执行select(6,&set,0,0,0)阻塞等待
  • (5)若fd=1,fd=2上都发生可读事件,则select返回,此时set变为0000,0011。注意:没有事件发生的fd=5被清空
  •  5.select优缺点

select的特点

  • 可监控的文件描述符个数取决与sizeof(fd_set)的值. 我这边服务器上sizeof(fd_set)=512,每bit表示一个文件描述符,则我服务器上支持的最大文件描述符是512*8=4096.
  • 将fd加入select监控集的同时,还要再使用一个数据结构array保存放到select监控集中的fd,
  • 一是用于再select 返回后,array作为源数据和fd_set进行FD_ISSET判断。
  • 二是select返回后会把以前加入的但并无事件发生的fd清空,则每次开始select前都要重新从array取得 fd逐一加入(FD_ZERO最先),扫描array的同时取得fd最大值maxfd,用于select的第一个参数

select缺点

  • 每次调用select, 都需要手动设置fd集合, 从接口使用角度来说也非常不便.
  • 每次调用select,都需要把fd集合从用户态拷贝到内核态,这个开销在fd很多时会很大
  • 同时每次调用select都需要在内核遍历传递进来的所有fd,这个开销在fd很多时也很大
  • select支持的文件描述符数量太小
  •  6.select实现tcp服务器(支持多用户)
#pragma once
#include <vector>
#include <unordered_map>
#include <functional>
#include <sys/select.h>
#include "tcp_socket.hpp"


// 必要的调试函数

inline void PrintFdSet(fd_set* fds, int max_fd) 
{
  printf("select fds: ");
  for (int i = 0; i < max_fd + 1; ++i) 
  {
    if (!FD_ISSET(i, fds)) 
    {
      continue;
    }
    printf("%d ", i);
  }
  printf("\n");
}

typedef std::function<void (const std::string& req, std::string* resp)> Handler;

class Selector 
{
  public:
    Selector() 
    {
      // [注意!] 初始化千万别忘了!!
      max_fd_ = 0;
      FD_ZERO(&read_fds_);
    }

    bool Add( TcpSocket& sock) 
    {
      int fd = sock.GetFd();
      printf("[Selector::Add] %d\n", fd);

      if (fd_map_.find(fd) != fd_map_.end()) 
      {
        printf("Add failed! fd has in Selector!\n");
        return false;
      }

      fd_map_[fd] = sock;
      FD_SET(fd, &read_fds_);
      if (fd > max_fd_) 
      {
        max_fd_ = fd;

      }
      return true;
    }

    bool Del( TcpSocket& sock) 
    {
      int fd = sock.GetFd();
      printf("[Selector::Del] %d\n", fd);
      if (fd_map_.find(fd) == fd_map_.end()) 
      {
        printf("Del failed! fd has not in Selector!\n");
        return false;

      }
      fd_map_.erase(fd);

      FD_CLR(fd, &read_fds_);
      // 重新找最大的文件描述符, 从右往左找比较快
      for (int i = max_fd_; i >= 0; --i) 
      {
        if (!FD_ISSET(i, &read_fds_)) 
        {
          continue;
        }
        max_fd_ = i;
        break;
      }
      return true;
    }
    // 返回读就绪的文件描述符集
    bool Wait(std::vector<TcpSocket>* output) 
    {
      output->clear();
      // [注意] 此处必须要创建一个临时变量, 否则原来的结果会被覆盖掉
      fd_set tmp = read_fds_;
      PrintFdSet(&tmp, max_fd_);
      int nfds = select(max_fd_ + 1, &tmp, NULL, NULL, NULL);
      if (nfds < 0) 
      {
        perror("select");
        return false;

      }
      // [注意!] 此处的循环条件必须是 i < max_fd_ + 1
      for (int i = 0; i < max_fd_ + 1; ++i) 
      {
        if (!FD_ISSET(i, &tmp)) 
        {
          continue;
        }
        output->push_back(fd_map_[i]);
      }
      return true;
    }
  private:
    fd_set read_fds_;
    int max_fd_;
    // 文件描述符和 socket 对象的映射关系
    std::unordered_map<int, TcpSocket> fd_map_;
};

class TcpSelectServer 
{
  public:
    TcpSelectServer(const std::string& ip, uint16_t port) : ip_(ip), port_(port) 
  {}

    bool Start(Handler handler) const 
    {
      //1. 创建 socket
      TcpSocket listen_sock;
      bool ret = listen_sock.Socket();
      if (!ret) 
      {
        return false;
      }
      // 2. 绑定端口号
      ret = listen_sock.Bind(ip_, port_);
      if (!ret) 
      {
        return false;
      }
      // 3. 进行监听
      ret = listen_sock.Listen(5);
      if (!ret) 
      {
        return false;
      }
      // 4. 创建 Selector 对象
      Selector selector;
      selector.Add(listen_sock);
      // 5. 进入事件循环
      for (;;) 
      {
        std::vector<TcpSocket> output;

        bool ret = selector.Wait(&output);
        if (!ret) 
        {
          continue;

        }
        // 6. 根据就绪的文件描述符的差别, 决定后续的处理逻辑
        for (size_t i = 0; i < output.size(); ++i) 
        {
          if (output[i].GetFd() == listen_sock.GetFd()) 
          {
            // 如果就绪的文件描��符是 listen_sock, 就执行 accept, 并加入到 select 中
            TcpSocket new_sock;
            listen_sock.Accept(&new_sock, NULL, NULL);
            selector.Add(new_sock);
          }
          else 
          {
            // 如果就绪的文件描述符是 new_sock, 就进行一次请求的处理
            std::string req, resp;
            bool ret = output[i].Recv(&req);
            if (!ret) 
            {
              selector.Del(output[i]);
              // [注意!] 需要关闭 socket
              output[i].Close();
              continue;
            }
            // 调用业务函数计算响应
            handler(req, &resp);
            // 将结果写回到客户端
            output[i].Send(resp);
          }
        } // end for
      } // end for (;;)
      return true;
    }
  private:
    std::string ip_;
    uint16_t port_;
};

在这里插入图片描述

在这里插入图片描述

0

评论区