高效IO——IO多路转接select

Posted 2021-11-10 两片空白

目录

一.概念

二.select函数

        2.1 函数原型

        2.2参数详细介绍

                2.2.1 nfd

                2.2.2readfds，writefds，errorfds

                2.2.3 timeout

        2.3. 网络中读/写/异常的就绪条件

        2.4 select特点

        2.5 select缺点

三.select的使用

---

一.概念

        IO主要有两个动作，等待条件就绪和进行数据拷贝。高效IO就是将等待时间比重减小。

        IO多路转接是高效IO的一种。通过调用select，poll，epoll在同一时刻等待多个文件描述符。当至少一个文件描述符准备就绪，再来进行IO操作时，就不需要等待了。

        这样一次性等待多个文件描述符，条件就绪的概率增加了，等待的时间也会减少。

        下文主要介绍select。

二.select函数

        2.1 函数原型

#include <sys/select.h>

 int select(int nfds, fd_set *restrict readfds,fd_set *restrict writefds,
             fd_set *restrict errorfds,struct timeval *restrict timeout);

作用：select可以监视多个文件描述符的状态，程序会停在select这里等待，直到一个或者多个文件描述符状态发生变化。

参数：

参数	作用
nfd	需要监视最大文件描述符值加1
readfds	类型为fd_set，可读文件文件描述符的集合，输入输出型参数
writefds	类型为fd_set，可写文件文件描述符的集合，输入输出型参数
errorfds	类型为fd_set，异常文件文件描述符的集合，输入输出型参数
timeout	结构为timeval，用来设置select等待时间，输入输出型参数

返回值：

• 执行成功返回文件描述符状态改变个数
• 返回0，说明等待时间超过timeout。
• 当有错误发生时返回-1，错误原因存于errno，此时参数readfds，writefds，errorfds和timeout的值变得不可预测。
  • 错误码可能是：
    • EBADF：文件描述词无效或者该文件已经关闭
    • EINTR：此调用被信号中断
    • EINVAL：参数n为负值
    • ENOMEM：核心内存不足

        2.2参数详细介绍

                2.2.1 nfd

        nfd：需要监视最大文件描述符值加1。

如果需要监视的文件描述符为1，2，3，4，nfd等于5。如果逍遥监视的文件描述符为1，5，nfd等于6。

                2.2.2readfds，writefds，errorfds

        readfds，writefds，errorfds是主要和类型fd_set有关。并且它们是类似的。

关于fd_set结构：

typedef struct
{
/*XPG4.2requiresthismembername.Otherwiseavoidthename
fromtheglobalnamespace.*/
#ifdef__USE_XOPEN
__fd_maskfds_bits[__FD_SETSIZE/__NFDBITS];
#define__FDS_BITS(set)((set)->fds_bits)
#else
__fd_mask__fds_bits[__FD_SETSIZE/__NFDBITS];
#define__FDS_BITS(set)((set)->__fds_bits)
#endif
}fd_set;

        fd_set是文件描述符集，结构实际是一个位图。

        readfds，writefds，errorfds是输入输出参数，输入时是用户想告诉内核需要监视哪些文件描述符，当作为输出时，是内核想告诉用户，那些文件描述符已经就绪。

        位图的对应位代表着要监视的文件描述符，比特位的内容，作为输入时，内容代表需要监视的文件，作为输出时，内容代表那些文件条件已经就绪。

        比如：readfds：拿8位举例，作为输入时，当输入1001 0101时，是用户想告诉内核，需要监视文件描述符等于0，2，4，7的文件的读事件的状态。作为输出时，输出为1000 0001时，是内核想告诉用户，文件描述符为0，7的文件读事件一ing就绪，可以进行读操作。

        内核监视文件的个数是确定的，说明内核监视文件的个数是有限的。内核监视多个文件描述符，采用监视的方法是轮询监视。

由于不同的系统fd_set实现方式可能不同，可能是数组，可能是结构体，所以提供了一组操作fd_set的接口，来对位图进行设置。

void FD_CLR(int fd, fd_set *fdset);  //用来清除fd_set中相关fd的位
int FD_ISSET(int fd, fd_set *fdset); //用来测试fd_set中相关fd的位是否为真
void FD_SET(int fd, fd_set *fdset);  //用来设置fd_set中相关fd的位
void FD_ZERO(fd_set *fdset);         //用来清除fd_set中的全部位，相当于初始化

        注意：用户输入了监视那些为你文件描述符，内核输出条件就绪的文件，一定只会是这些文件描述符里的子集。 

                2.2.3 timeout

        timeout，结构是timeval。用来设置select等待时间。

关于timeval结构：

struct timeval
{
    time_t      tv_sec;     /* seconds 秒*/
    suseconds_t tv_usec;    /* microseconds 微秒*/
};

参数timeout的取值：

1. NULL：表示select在没有文件条件就绪时，会阻塞等待。
2. 0：非阻塞等待，不管条件就没就绪都会返回，用于检测监视的文件的状态。
3. 特定的时间值：等待一段时间，在时间范围内有文件条件就绪，返回，超过时间select返回0。

timeout也是一个输入输出参数。当输入时，用户告诉内核等待时间timeout，当输出时，内核等待完毕，等待时间timeout就为0了。

        注意：在编码时，由于readfds，writefds，errorfds和timeout都是输入输出型参数，当select一次后，重新select时，由于输出时已经改变参数的值，所以需要重新设定readfds，writefds，errorfds和timeout的值。

        2.3. 网络中读/写/异常的就绪条件

读就绪

• 在socket内核中，接收缓存区的字节数，大于等于低水位标记SO_RCVLOWAT，此时可以无阻塞的读取该文件。
• 监听的socket上有了新的连接请求，socket连接请求也是以读的方式获取的。
• socketTCP通信，对端关闭连接，此时对该socket读返回0.
• socket上有未处理的错误。

写就绪

• socket内核中，发送缓冲区中的空闲位置大小，大于或者等于低水位标记。
• socket使用非阻塞connect连接成功或失败之后。
• socket的写操作被关闭，对于写操作被关闭的socket进行写操作，会触发SIGPIPE信号。
• socket上有为读取的错误。

异常就绪：

• socket收到带外数据。

        2.4 select特点

• 内核可监控的文件是有限的，取决于fd_set的位数，即能监控的文件数为sizeof(fd_set)*8。

#include <stdio.h>
#include <fcntl.h>
#include <unistd.h>

int main(){
  printf("%lu\\n",sizeof(fd_set)*8); 

  return 0;
}

不同系统值不同。

• 将fd加入select监控集中，还需要一个数组array来保存select监控的文件描述符。
  • select返回后，需要对数组中保存每一个文件描述符进行判断，是否就绪。FD_ISSET。
  • select返回后会改变fd_set文件描述符集，每次重新开始select时，需要重新设定数组中保存的文件描述符到fd_set中。FD_SET。并且需要保存文件描述符的最大值，用于select参数。

这个在下面编码是可以明显观察到。

        2.5 select缺点

• 每次调用select，都需要程序员重新设定fd_set集合。
• 每次调用select，都需要吧fd集合从用户态拷贝到内核态，这个开销在fd很多会很大。
• 内核在监视所有文件时，都需要不断遍历传进来的所有文件，轮询检测Negev我呢见就绪，这个开销在fd很多时会很大。
• select支持监视的文件数量是有限的。

三.select的使用

用select编写一个单进程echo服务器。

注意点：

1. 需要用数组保存所有要监视的文件描述符。该文件描述符包括进行读的套接字文件描述符和连接套接字。
2. 当没有连接时，accept会阻塞等待，所以连接也需要select等待。
3. 连接是socket返回的套接字做的，有新连接来时，也是读就绪。
4. IO读写是accept返回套接字做的，接收缓冲区的字节数大于低水位标记，读就绪。发送缓冲区剩余空间大小，大于低水位标记，写就绪。
5. 当获取到连接，不是直接进行读写，而是将accept返回值放到数组中，如果直接读写，如果客户端没有发数据，又会阻塞。
6. 注意fd_set位图需要用FD_ZERO初始化

        再程序中有一个BUG，一次性读取并不一定将整个request全部读上来了，可能只读了一部分。我需要将整个数据读上来，再返回response给客户端。而我们定义的收集数据的缓冲区是一个局部变量，每次从调用函数都会被重新建立，之前数据不能保存了。        

        如果定义成全局，其它文件的数据也保存在缓冲区中，就乱了，所以要每一个文件描述符需要一个缓冲区。

        我们可以定义一个map，键值key为文件描述符，value为缓冲区，每次将数据保存到对应文件描述符的缓冲区中。当整个数据时，再返回response。

        但是在epoll中很好的解决了这个问题。

套接字设定： 

#pragma once 

#include <iostream>
#include <string>
#include <stdlib.h>
#include <unistd.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>

#define BLACKLOG 5
using namespace std;

class Sock{
  public:
  static int Socket(){
    int sock = 0;
    sock = socket(AF_INET, SOCK_STREAM, 0);
    if(sock < 0){
      cerr << "socket error"<<endl;
      exit(1);
    }
    return sock;
  }
  static void Bind(int sock, int port){
    struct sockaddr_in local;
    
    local.sin_family = AF_INET;
    local.sin_port = htons(port);
    local.sin_addr.s_addr = htonl(INADDR_ANY);
    if(bind(sock, (struct sockaddr *)&local, sizeof(local)) < 0){
      cerr << "bind error" <<endl;
      exit(3);
    }
  }
  static void Listen(int sock){
    if(listen(sock, BLACKLOG) < 0){
      cerr << "listen error"<<endl;
      exit(4);
    }

  }
  static int Accept(int lsock){
    struct sockaddr_in peer;
    socklen_t len = sizeof(peer);
    return  accept(lsock, (struct sockaddr *)&peer, &len);
  }


};

select服务器主体：

#pragma once 

#include "Sock.hpp"

#define NUM sizeof(fd_set)*8//数组大小=最多能监视文件个数
#define DET_FD -1//数组默认文件描述符 

class SelectServer{
  private:
    int _lsock;//套接字
    int _port;//端口号
    int array[NUM];//保存要监视的文件描述符
  public:
    SelectServer(int lsock = -1, int port = 8080)
    :_lsock(lsock)
    ,_port(port)
    {}
    void InitServer(){
      for(size_t i =  0; i < NUM; i++){
        array[i] = DET_FD;
      }
      _lsock = Sock::Socket();
      //端口复用                                                                                                                                         
      int opt = 1;                                                    
      setsockopt(_lsock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); 
      Sock::Bind(_lsock, _port);
      Sock::Listen(_lsock);
      array[0] = _lsock;

    }
    void AddtoArray(int index){
      //找到没有数组没有占用的位置
      size_t i = 0;
      for(; i < NUM; i++){
        if(array[i] == DET_FD){
          break;
        }
      }
      //满了
      if(i >= NUM){
        cout<<"select is full, close fd"<<endl;
        close(index);
      }
      else{
        array[i] = index;
      }
    }
    void Delete(size_t index){
      if(index >= 0 && index < NUM){
        array[index] = DET_FD;
        
      }

    }
    void Handle(int i){
      //IO条件就绪
      char buf[10240];
      ssize_t n = recv(array[i], buf, sizeof(buf), 0);
      if(n > 0){
        buf[n] = 0;
        cout<<buf<<endl;
      }
      else if(n == 0){
        //对端关闭
        cout<<"client close..."<<endl;
        close(array[i]);
        //文件已经关闭，还需要将数组文件描述符删除
        Delete(i);
      }
      else{
        cerr << "read error"<<endl;
        close(array[i]);
        Delete(i);
      }


    }
    void Start(){
      while(1){
        int maxfd = DET_FD;
        //重新设定，需要等待的文件
        fd_set readfds;
        //初始化fd_set
        FD_ZERO(&readfds);
        //找文件描述符，将要监视的fd_set对应位设置为1
        for(size_t i =0; i < NUM; i++){
          if(array[i] == DET_FD){
            continue;
          }
          cout <<array[i];
          FD_SET(array[i], &readfds);
          //找文件描述符最大值
          if(maxfd < array[i]){
            maxfd = array[i];
          }
        }
        cout<<endl;
        //struct timeval timeout = {5, 0};
        //调用 select 等待多个文件
        //阻塞等待
        int fdn = select(maxfd+1, &readfds, nullptr, nullptr, nullptr);
        if(fdn > 0){
          //有文件就绪
          //找哪个文件就绪
          for(size_t i =0; i < NUM; i++){
            if(array[i] != DET_FD && FD_ISSET(array[i] , &readfds)){
              if(array[i] == _lsock){
                //有新连接
                int sock = Sock::Accept(array[i]);
                if(sock >= 0){
                  cout << "get a link...."<<endl;
                  //加入到数组中
                  AddtoArray(sock);
                }
                
              }
              else{
                //进行IO操作
                Handle(i);
              }
            }
          }

        }
		//超时
        else if(fdn == 0){
          cerr << "select timeout..."<<endl;

        }
		//异常
        else{
          cerr <<"fdn:"<<fdn<< "select error"<<endl;
        }


      }
    }

    ~SelectServer(){
      for(size_t i = 0; i < NUM; i++){
        if(array[i] != DET_FD){
          close(array[i]);
        }
      }
    }


};

#include"selectServer.hpp"

void Notice(string str){
  cout<<"Notice\\n\\t"<<"please enter port"<<endl;
}

int main(int argc, char *argv[]){
  if(argc != 2){
    Notice(argv[0]);
    exit(1);
  }

  SelectServer *sser = new SelectServer(atoi(argv[1]));
  sser->InitServer();
  sser->Start();
  delete sser;

  return 0;
}