[计算机网络]--五种IO模型和select

前言

作者：小蜗牛向前冲

名言：我可以接受失败，但我不能接受放弃

如果觉的博主的文章还不错的话，还请点赞，收藏，关注👀支持博主。如果发现有问题的地方欢迎❀大家在评论区指正

一、五种IO模型

1、什么是IO

2、感性的理解五种IO模型

3、理解五种IO模型

4、高级IO重要概念

二、I/O多路转接之select

1、select的基本概念和接口介绍

2、对select的理解

三、select服务器的编写

1、err.hpp和log.hpp

2、makefile和main.cc

3、 selectServer.hpp和sock.hpp

4、测试

本期学习：IO五层模型的理解，select的接口常识及其多路转接的理解，编写select服务器。

一、五种IO模型

1、什么是IO

"IO" 通常指的是输入/输出（Input/Output）。

在计算机科学和编程中，输入/输出是指程序与外部世界、外部设备或其他程序之间进行数据交换的过程。
这些外部设备可以包括磁盘驱动器、网络连接、键盘、鼠标、显示器等。输入是指程序接收来自外部环境的数据，输出是指程序将数据发送到外部环境。

IO的操作

从文件中读取数据、向文件写入数据、从网络接收数据、向网络发送数据，以及与硬件设备进行交互等。
IO 操作通常是相对较慢的，因为它们涉及到与外部设备或网络通信，而这些通信可能涉及到物理设备的限制或网络延迟。

本文主要讨论文件上的IO。

我们在文件上写入或者是读取数据，在系统层面上就是调用read/recv这些函数借口，前面我们也谈论过调用这些函数的本质其实在拷贝数据。

对于read/recv无非存在二种情况：

没有数据，就会进行阻塞等待。
有数据就会进行拷贝，完成后返回。

这也就说明IO的本质是拷贝+等待

那我们如何做到高效IO呢？

本质上我们只要减少等待的时间就可以。

下面我们通过一个故事感性的理解五种IO模型。

2、感性的理解五种IO模型

有这么几个人，他们非常喜欢钓鱼。

1号张三用一根钓鱼竿钓鱼，他喜欢一直盯这鱼竿看鱼有没有上钩。

2号李四也是一根钓鱼竿钓鱼，但是他就比较休闲，他是每隔一定时间看一下鱼竿动了没，没动就去做别的事情。

3号王五也是一根钓鱼竿钓鱼，但他就比较有意思，他在鱼竿上寄了一个铃铛，要是鱼竿动了他就拉杆看有没鱼，没声音响就一直忙自己的时候。

4号赵六他觉的用一根鱼竿钓鱼的效率太慢了，于是就弄了一排鱼竿，来会的在这一排鱼竿旁边走，看那个鱼竿动了就拉起来。

5号小王他是个大老板，他喜欢吃这里钓的鱼吃，自己时间又忙，于是他就让他的属下田七来这里钓鱼。

在上面故事中的钓鱼其实就分为等+钓。

那上面谁钓鱼的效率高呢？我们知道等的比重越低，单位时间内钓鱼的越高。

那肯定是赵六的效率是最高的，因为他等的比例是最低的。

在程序员看来我们可以认为：

鱼就是数据，鱼塘就是内核空间，鱼竿发生动作鱼就绪是数据就绪的事情，鱼竿我们就认为是文件描述符，钓鱼的动作：recv/read系统接口的调用。

五号任务就代表五种IO模型：

张三----------->阻塞式IO

李四----------->非阻塞式IO

王五----------->信号驱动式IO

赵六----------->多路转接/多路复用

田七----------->异步IO（2这里的老板赵六相当鱼操作系统，田七相当进程/线程）

3、理解五种IO模型

阻塞IO是最常见的IO模型

在阻塞 I/O 中，当应用程序发起一个 I/O 操作（比如读取文件或者从网络接收数据），程序会被阻塞（暂停执行），直到操作完成并且数据准备好被应用程序处理。

非阻塞IO

非阻塞IO: 如果内核还未将数据准备好, 系统调用仍然会直接返回, 并且返回EWOULDBLOCK错误码

非阻塞IO往往需要程序员循环的方式反复尝试读写文件描述符, 这个过程称为轮询. 这对CPU来说是较大的浪费, 一般只有特定场景下才使用.

信号驱动IO

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

IO多路转接:

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

小结

任何IO过程中, 都包含两个步骤. 第一是等待, 第二是拷贝.
而且在实际的应用场景中, 等待消耗的时间往往都远远高于拷贝的时间. 让IO更高效, 最核心的办法就是让等待的时间尽量少

4、高级IO重要概念

同步通信 vs 异步通信

所谓同步，就是在发出一个调用时，在没有得到结果之前，该调用就不返回. 但是一旦调用返回，就得到返回值了; 换句话说，就是由调用者主动等待这个调用的结果;
异步则是相反，调用在发出之后，这个调用就直接返回了，所以没有返回结果; 换句话说，当一个异步过程调用发出后，调用者不会立刻得到结果; 而是在调用发出后，被调用者通过状态、通知来通知调用者，或通过回调函数处理这个调用

另外, 我们回忆在讲多进程多线程的时候, 也提到同步和互斥. 这里的同步通信和进程之间的同步是完全不想干的概念.

进程/线程同步也是进程/线程之间直接的制约关系
是为完成某种任务而建立的两个或多个线程，这个线程需要在某些位置上协调他们的工作次序而等待、传递信息所产生的制约关系. 尤其是在访问临界资源的时候

同学们以后在看到 "同步" 这个词, 一定要先搞清楚大背景是什么. 这个同步, 是同步通信异步通信的同步, 还是进程同步与互斥的同步

阻塞 vs 非阻塞

阻塞和非阻塞关注的是程序在等待调用结果（消息，返回值）时的状态

阻塞调用是指调用结果返回之前，当前线程会被挂起. 调用线程只有在得到结果之后才会返回.
非阻塞调用指在不能立刻得到结果之前，该调用不会阻塞当前线程

二、I/O多路转接之select

1、select的基本概念和接口介绍

这里我们先一起达成一个公识：IO的本质=等+拷贝。

select是一个系统调用只负责等，可以等待多个fd,select本身没有数据拷贝的能力，拷贝还是要read,write来完成。

系统提供select函数来实现多路复用输入/输出模型.

select系统调用是用来让我们的程序监视多个文件描述符的状态变化的;
程序会停在select这里等待，直到被监视的文件描述符有一个或多个发生了状态改变;

select函数原型

包含的头文件

   #include <sys/time.h>
   #include <sys/types.h>
   #include <unistd.h>

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

参数说明

nfds：监视的文件描述符中最大的文件描述符值加一。
readfds：指向一个 fd_set 结构的指针，用于指定一组待检查是否可读的文件描述符。
writefds：指向一个 fd_set 结构的指针，用于指定一组待检查是否可写的文件描述符。
exceptfds：指向一个 fd_set 结构的指针，用于指定一组待检查是否异常的文件描述符。
timeout：指向 struct timeval 结构的指针，用于设置 select() 调用的超时时间，如果为 NULL 则表示不设置超时，会一直阻塞直到有文件描述符就绪或者被信号中断。

参数timeout取值

NULL：则表示select（）没有timeout，select将一直被阻塞，直到某个文件描述符上发生了事件;
0：非阻塞仅检测描述符集合的状态，然后立即返回，并不等待外部事件的发生。
特定的时间值：如果在指定的时间段里没有事件发生，select将超时返回。
struct timeval timeout{5,0}:表示5秒以内阻塞，超过5秒，非阻塞返回一次。

返回值

ret>0 告诉系统育多少个fd就绪
ret==0调用超时，返回
ret<0调用失败

fd_set:位图结构，表示文件描述符的集合

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

提供了一组操作fd_set的接口, 来比较方便的操作位图.

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的全部位

关于timeval结构

timeval结构用于描述一段时间长度，如果在这个时间内，需要监视的描述符没有事件发生则函数返回，返回值为 0

函数返回值：

执行成功则返回文件描述词状态已改变的个数
如果返回0代表在描述词状态改变前已超过timeout时间，没有返回
当有错误发生时则返回-1，错误原因存于errno，此时参数readfds，writefds, exceptfds和timeout的值变成不可预测。

错误值可能为：

EBADF (ebadf)文件描述词为无效的或该文件已关闭
EINTR(eintr) 此调用被信号所中断
EINVAL(einval) 参数n 为负值。
ENOMEM(enomem) 核心内存不足

2、对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被清空。

socket就绪条件

读就绪

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

写就绪

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

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支持的文件描述符数量太小

select是如何实现多路转接的

准备监视的文件描述符集合：程序通过向内核传递一个文件描述符集合，告诉内核它希望监视哪些文件描述符的状态变化。

调用 select 系统调用：程序调用 select 系统调用，并将准备好进行 I/O 操作的文件描述符集合传递给内核。

内核监视文件描述符状态变化：内核开始监视这些文件描述符的状态变化。如果其中任何一个文件描述符的状态发生变化（例如，变为可读、可写或出现异常），内核将返回给程序。

程序处理返回结果：程序从 select 返回的结果中获取到哪些文件描述符准备好进行 I/O 操作，然后针对这些文件描述符执行相应的 I/O 操作。通常，程序会使用 read、write 等系统调用来实际进行 I/O 操作。

select 的实现通常使用轮询技术，内核会遍历程序提供的所有文件描述符，检查它们的状态是否发生变化。这种方式虽然简单，但效率较低，尤其在文件描述符数量较多时会导致性能下降。

上面我们理解select进行多路转接的原理，下面我们自己写一个select多路转接的服务器加深理解。

三、select服务器的编写

1、err.hpp和log.hpp

err.hpp

#pragma once

#include <iostream>

enum
{
    USAGE_ERR = 1, // usage_err
    SOCKET_ERR,    // socket_err
    BIND_ERR,      // bind_err
    LISTEN_ERR     // listen_err
};

log.hpp

#pragma once

#include <iostream>
#include <string>
#include <cstdarg>
#include <ctime>
#include <unistd.h>

// debug
#define DEBUG 0
// normal
#define NORMAL 1
// warning
#define WARNING 2
// error
#define ERROR 3
// fatal
#define FATAL 4

const char *to_levelstr(int level)
{
    switch (level)
    {
    case DEBUG:
        return "DEBUG";
    case NORMAL:
        return "NORMAL";
    case WARNING:
        return "WARNING";
    case ERROR:
        return "ERROR";
    case FATAL:
        return "FATAL";
    default:
        return nullptr;
    }
}

void logMessage(int level, const char *format, ...)
{
#define NUM 1024
    char logprefix[NUM];
    snprintf(logprefix, sizeof(logprefix), "[%s][%ld][pid: %d]",
             to_levelstr(level), (long int)time(nullptr), getpid());

    char logcontent[NUM];
    va_list arg;
    va_start(arg, format);
    vsnprintf(logcontent, sizeof(logcontent), format, arg);

    std::cout << logprefix << logcontent << std::endl;
}

2、makefile和main.cc

makefile

select_server: main.cc
	g++ -o $@ $^ -std=c++11
.PHONY:clean
clean:
	rm -f select_server

main.cc

#include "selectServer.hpp"
#include "err.hpp"
#include <memory>

using namespace std;
using namespace select_ns;

static void usage(std::string proc)
{
    std::cerr << "Usage:\n\t" << proc << " port"
              << "\n\n";
}

std::string transaction(const std::string &request)
{
    return request;
}

// ./select_server 8081
int main(int argc, char *argv[])
{
    // if(argc != 2)
    // {
    //     usage(argv[0]);
    //     exit(USAGE_ERR);
    // }

    // unique_ptr<SelectServer> svr(new SelectServer(atoi(argv[1])));

    // std::cout << "test: " << sizeof(fd_set) * 8 << std::endl;
    unique_ptr<SelectServer> svr(new SelectServer(transaction));

    svr->initServer();

    svr->start();

    return 0;
}

3、 selectServer.hpp和sock.hpp

selectServer.hpp

#pragma once

#include <iostream>
#include <string>
#include <functional>
#include "sock.hpp"

namespace select_ns
{
    static const int defaultport = 8081;
    static const int fdnum = sizeof(fd_set) * 8;
    static const int defaultfd = -1;

    using func_t = std::function<std::string(const std::string &)>;
    class SelectServer
    {
    public:
        SelectServer(func_t f, int port = defaultport) : func(f), _port(port), _listensock(-1), fdarray(nullptr)
        {
        }
        void initServer()
        {
            _listensock = Sock::Socket();
            Sock::Bind(_listensock, _port);
            Sock::Listen(_listensock);
            fdarray = new int[fdnum];
            for (int i = 0; i < fdnum; i++)
            {
                fdarray[i] = defaultfd;
            }
            fdarray[0] = _listensock;
        }

        void Print()
        {
            std::cout << "fd list: ";
            for (int i = 0; i < fdnum; i++)
            {
                if (fdarray[i] != defaultfd)
                    std::cout << fdarray[i] << " ";
            }
            std::cout << std::endl;
        }

        void Accepter(int listensock)
        {
            logMessage(DEBUG, "Accepter in");
            // select 告诉我， listensock读事件就绪了
            std::string clientip;
            uint16_t clientport = 0;
            int sock = Sock::Accept(listensock, &clientip, &clientport);
            if (sock < 0)
                return;
            logMessage(NORMAL, "accept success [%s:%d]", clientip.c_str(), clientport);
            // sock我们能直接recv/read 吗？不能，整个代码，只有select有资格检测事件是否就绪
            // 将新的sock 托管给select！
            // 将新的sock托管给select的本质，其实就是将sock，添加到fdarray数组中即可！

            int i = 0;
            // 找fdarray字符集中没有被占用的位置
            for (; i < fdnum; i++)
            {
                if (fdarray[i] != defaultfd)
                    continue;
                else
                    break;
            }
            if (i == fdnum)
            {
                logMessage(WARNING, "server if full, please wait");
                close(sock);
            }
            else
            {
                fdarray[i] = sock;
            }
            Print();
            logMessage(DEBUG, "Accepter out");
        }

        void Recver(int sock, int pos)
        {
            logMessage(DEBUG, "in Recver");
            // 1. 读取request
            // 这样读取是有问题的！
            char buffer[1024];
            ssize_t s = recv(sock, buffer, sizeof(buffer) - 1, 0);

            if (s > 0)
            {
                buffer[s] = 0;
                logMessage(NORMAL, "client# %s", buffer);
            }
            else if (s == 0)
            {
                close(sock);
                fdarray[pos] = defaultfd;
                logMessage(NORMAL, "client quit");
                return;
            }
            else
            {
                close(sock);
                fdarray[pos] = defaultfd;
                logMessage(ERROR, "client quit: %s", strerror(errno));
                return;
            }

            // 2、处理request
            std::string response = func(buffer);

            // 3、返回response
            write(sock, response.c_str(), response.size());
            logMessage(DEBUG, "out Recver");
        }
        // 1. handler event rfds 中，不仅仅是有一个fd是就绪的，可能存在多个
        // 2. 我们的select目前只处理了read事件
        void HandlerReadEvent(fd_set &rfds)
        {
            // 遍历fdarray数组
            for (int i = 0; i < fdnum; i++)
            {
                // 过滤掉非法的fd
                if (fdarray[i] == defaultfd)
                    continue;
                // 正常的fd,不一定就绪了
                if (FD_ISSET(fdarray[i], &rfds) && fdarray[i] == _listensock)
                    Accepter(_listensock);
                else if (FD_ISSET(fdarray[i], &rfds))
                    Recver(fdarray[i], i);
                else
                {
                }
            }
        }

        void start()
        {
            for (;;)
            {
                fd_set rfds;
                FD_ZERO(&rfds);
                int maxfd = fdarray[0];

                for (int i = 0; i < fdnum; i++)
                {
                    if (fdarray[i] == defaultfd)
                        continue;
                    FD_SET(fdarray[i], &rfds); // 将合法的fd全部添加到读文件描述符中

                    // 更新最大的maxfd
                    if (maxfd < fdarray[i])
                        maxfd = fdarray[i];
                }
                logMessage(NORMAL, "max fd is: %d", maxfd);

                int n = select(maxfd + 1, &rfds, nullptr, nullptr, nullptr);
                switch (n)
                {
                case 0:
                    logMessage(NORMAL, "timeout...");
                    break;
                case -1:
                    logMessage(WARNING, "select error, code: %d, err string: %s", errno, strerror(errno));
                    break;
                default:
                    // 说明有事件就绪了,目前只有一个监听事件就绪了
                    logMessage(NORMAL, "have event ready!");
                    HandlerReadEvent(rfds);
                    // HandlerWriteEvent(wfds);
                    break;
                }
            }
        }
        ~SelectServer()
        {
            if (_listensock < 0)
                close(_listensock);
            if (fdarray)
                delete fdarray;
        }

    private:
        int _port;
        int _listensock;
        int *fdarray;
        func_t func;
    };
}

sock.hpp

#pragma once

#include <iostream>
#include <string>
#include <cstring>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "log.hpp"
#include "err.hpp"

class Sock
{
    const static int backlog = 32; // sokc listen的数量

public:
    static int Socket()
    {
        // 1创建套接字
        // int sock = socket(AF_FILE, SOCK_STREAM, 0); // af_file,sock_stream errror
        int sock = socket(AF_INET, SOCK_STREAM, 0);//af_inet
        if (sock < 0)
        {
            logMessage(FATAL, "create socket error");
            exit(SOCKET_ERR);
        }
        logMessage(NORMAL, "create socket success: %d", sock);
        int opt = 1;
        setsockopt(sock, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)); // sol_socket,so_reuseaddr,so_reuseport//服务器重启后可快速复用地址和端口
        return sock;
    }

    static void Bind(int sock, int port)
    {
        // 2bind绑定网络信息
        struct sockaddr_in local;
        memset(&local, 0, sizeof(local));
        local.sin_family = AF_INET; // afinet
        local.sin_port = htons(port);
        local.sin_addr.s_addr = INADDR_ANY; // inaddr_any
        if (bind(sock, (struct sockaddr *)&local, sizeof(local)) < 0)
        {
            logMessage(FATAL, "bind socket error");
            exit(BIND_ERR);
        }
        logMessage(NORMAL, "bind socket success");
    }

    static void Listen(int sock)
    {
        // 3设置sock为监听
        if (listen(sock, backlog) < 0)
        {
            logMessage(FATAL, "listen socket error");
            exit(LISTEN_ERR);
        }
        logMessage(NORMAL, "listen socket success");
    }

    static int Accept(int listensock, std::string *clientip, uint16_t *clientport)
    {
        struct sockaddr_in peer;
        socklen_t len = sizeof(peer);
        int sock = accept(listensock, (struct sockaddr *)&peer, &len);
        if (sock < 0)
            logMessage(ERROR, "accept error, next");
        else
        {
            logMessage(NORMAL, "accept a new link success, get new sock: %d", sock); // ?
            *clientip = inet_ntoa(peer.sin_addr);
            *clientport = ntohs(peer.sin_port);
        }
        return sock;
    }
};