IO模型介绍

一、理解IO

网络通信的本质就是进程间通信，进程间通信本质就是IO

TCP中的IO接口：read / write / send / recv，本质都是：等 + 拷贝

所以IO的本质就是：等 + 拷贝

那么如何高效的IO？

减少“等”在单位时间的比重

二、5种IO模型

同步IO就是亲自参与IO过程

1、阻塞IO

等到数据就绪才会拷贝。所有套接字（文件描述符）一开始都是默认阻塞IO

2、非阻塞IO

非阻塞 IO: 如果内核还未将数据准备好 , 系统调用仍然会直接返回 , 并且返回EWOULDBLOCK错误码。但是其实不是数据错误，只是数据没就绪。

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

3、信号驱动IO

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

4、多路转接IO

用select解放recvfrom，把等和拷贝完全分开，能够同时等待多个文件描述符的就绪状态。

5、异步IO

应用进程让OS自己等和拷贝，OS做完之后告诉进程，应用进程再对数据进行处理。

三、非阻塞IO

1、介绍函数

一开始文件描述符都是默认阻塞，用函数实现非阻塞

功能：

（1）复制一个现有的描述符（ cmd=F_DUPFD）

（2）获得/ 设置文件描述符标记 (cmd=F_GETFD 或 F_SETFD)

（3）获得/ 设置文件状态标记 (cmd=F_GETFL 或 F_SETFL)

（4）获得/ 设置异步 I/O 所有权 (cmd=F_GETOWN 或 F_SETOWN)

（5）获得/ 设置记录锁 (cmd=F_GETLK,F_SETLK 或 F_SETLKW)

2、函数实现非阻塞

#include<iostream>
#include<unistd.h>
#include<fcntl.h>
using namespace std;

void SetNonBlock(int fd)
{
    // 获取文件状态标记
    int fl = fcntl(fd, F_GETFL);
    if(fl < 0)
    {
        cout << "fcntl error" << endl;
        return;
    }
    // 文件状态标记加上非阻塞等待
    fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}

3、实现非阻塞轮询

#include<iostream>
#include<cstdio>
#include<unistd.h>
#include"Command.hpp"
using namespace std;

int main()
{
    char buffer[1024];
    // 设置非阻塞文件描述符
    SetNonBlock(0);
    while(1)
    {
        printf("Enter#: ");
        // 立即刷新到显示器
        fflush(stdout);
        ssize_t n = read(0, buffer, sizeof buffer);
        if(n > 0)
        {
            buffer[n] = 0;
            printf("echo: %s", buffer);
        }
        else if(n == 0)
        {
            printf("read done\n");
            break;
        }
        else
        {
            // 非阻塞轮询虽然数据没准备好不是错误，但是会以错误返回
            // 所以判断错误码确认是底层不就绪还是出错
            if(errno == EWOULDBLOCK)
            {
                sleep(1);
                cout << "数据没就绪" << endl;
                // 非阻塞时做其他事情
                continue;
            }

            else if(errno == EINTR)
            {
                continue;
            }

            else
            {
                perror("read");
                break;
            }

        }
    }
    return 0;
}

收到中断信号错误码是EINTR，继续轮询即可。

四、多路转接

1、多路转接思路

只聚焦于等，等待fd中新事件就绪，通知程序员时间就绪，进行IO拷贝。

事件：只有OS才知道到文件描述符的缓冲区是否有读写的能力。OS底层有数据，读事件就绪。OS底层有空间，写事件就绪。

2、方案一：select

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

nfds：传入的文件描述符最大值 + 1，不是传入的文件描述符个数。

timeout：结构体 timeval 里面有秒和毫秒，精度是毫秒，输入输出型参数，微秒级别时间戳结构体指针。

timeval timeout = {3, 0}; 3秒内阻塞等待fd新事件，有就绪就返回timeout剩余时间，3秒没有新事件就非阻塞轮询。

timeval timeout = {0, 0}; 直接非阻塞轮询

timeval timeout = nullptr; 直接阻塞等待

返回值：大于0，有几个文件描述符就绪。等于0，等待超时。小于0，select出错。

fd_set：文件描述符集，位图结构，比特位位置表示几号文件描述符，比特位内容表示是否关心fd事件 / fd事件是否发生。

readfds：输入输出型，只关心读事件。输入：用户告诉OS，要关心fd_set里面的读事件。输出：OS告诉用户，有哪些fd读事件就绪。

writefds：输入输出型，只关心写事件。输入：用户告诉OS，要关心fd_set里面的写事件。输出：OS告诉用户，有哪些fd写事件就绪。

exceptfds：输入输出型，只关心异常事件。输入：用户告诉OS，要关心fd_set里面的异常事件。输出：OS告诉用户，有哪些fd异常事件就绪。

对fd_set类型位图操作

我们不能自己对位图操作，要调用接口。

注意：

首先我们要维护多个文件描述符，所以就要有一个数组来保存合法fd。

而且三个参数都是输入输出型参数，这就意味着用户告诉内核要关心的事件和内核给用户返回关心事件的情况这两种情况都是由同一个 struct fd_set 传达的，这就要每次使用时重置。

事件就绪后要用循环检测处理事件。

代码实现（先只展示读文件描述符）

#pragma once
#include <iostream>
using namespace std;
#include "Socket.hpp"
using namespace socket_ns;
#include "log.hpp"
using namespace log_ns;
#include "inetAddr.hpp"
class SelectServer
{
    const static int gnum = sizeof(fd_set) * 8;
    const static int gdefaultfd = -1;

public:
    SelectServer(uint16_t port)
        : _port(port), _listensockfd(make_unique<TcpSocket>())
    {
        _listensockfd->BuildListenSocket(_port);
    }

    void InitServer()
    {
        for (int i = 0; i < gnum; i++)
        {
            fd_array[i] = gdefaultfd;
        }
        // 默认直接添加
        fd_array[0] = _listensockfd->Sockfd();
    }

    // listen套接字获取到新连接，即读事件就绪
    void Accepter()
    {
        // listen套接字读事件就绪
        // 已经就绪，绝对不会接收阻塞
        InetAddr addr;
        int connfd = _listensockfd->Accepter(&addr);
        if (connfd > 0)
        {
            LOG(DEBUG, "get a new link, client info %s:%d\n", addr.Ip().c_str(), addr.Port());
            // 新连接已经来了，但是不能直接读，可能会阻塞
            // OS清楚底层connfd数据是否就绪，要select
            // 把新的connfd添加给select，即添加到fd_array
            bool flag = false;
            for (int pos = 1; pos < gnum; pos++)
            {
                if (fd_array[pos] == gdefaultfd)
                {
                    fd_array[pos] = connfd;
                    flag = true;
                    LOG(INFO, "add %d to fd_array success!\n", connfd);
                    break;
                }
            }
            // 没有空余的空间存储新的文件描述符
            if (flag == false)
            {
                LOG(WARNING, "select is full\n");
                close(connfd);
            }
        }
        else
        {
            return;
        }
    }

    // 普通套接字读事件就绪，进行IO
    void HandlerIO(int connfd)
    {
        // 普通套接字读事件就绪
        char buffer[1024];
        // 不会阻塞
        ssize_t n = recv(connfd, buffer, sizeof buffer - 1, 0);
        if (n > 0)
        {
            buffer[n] = 0;
            cout << "client say# " << buffer << endl;
            string echo_str = "[server echo info] ";
            echo_str += buffer;
            // 可以直接写，任何一个文件描述符一开始获取时两个缓冲区都是空的
            // 这就意味着，一开始读事件一定不就绪，但是写事件一定就绪
            send(connfd, echo_str.c_str(), echo_str.size(), 0);
        }
        else if (n == 0)
        {
            LOG(INFO, "client quit....\n");
            // 关闭fd
            close(connfd);
            // select不要再关心fd，即移除fd
            connfd = gdefaultfd;
        }
        else
        {
            LOG(ERROR, "recv error\n");
            // 关闭fd
            close(connfd);
            // select不要再关心fd，即移除fd
            connfd = gdefaultfd;
        }
    }

    // 事件就绪，开始处理
    void HandlerEvent(fd_set &rfds)
    {
        // 事件派发
        for (int i = 0; i < gnum; i++)
        {
            if (fd_array[i] == gdefaultfd)
                continue;
            // fd合法，但不知道是不是就绪
            if (FD_ISSET(fd_array[i], &rfds))
            {
                // 读事件就绪
                if (_listensockfd->Sockfd() == fd_array[i])
                {
                    Accepter();
                }

                else
                {
                    HandlerIO(fd_array[i]);
                }
            }
        }
    }

    void Loop()
    {
        while (1)
        {
            // 1.文件描述符集初始化
            fd_set rfds;
            FD_ZERO(&rfds);
            int max_fd = gdefaultfd;

            // 2.合法fd添加到rfds
            for (int i = 0; i < gnum; i++)
            {
                if (fd_array[i] != gdefaultfd)
                {
                    FD_SET(fd_array[i], &rfds);
                }
                // 获取最大文件描述符值
                if (max_fd < fd_array[i])
                    max_fd = fd_array[i];
            }

            // 3.开始select所有合法fd
            struct timeval timeout = {3, 0};

            // 当前不能直接accept listensocket 因为函数的本质是把套接字看成文件描述符
            // 今天里面的函数没有改造，本质就是只关心新连接的到来，是读事件就绪，阻塞等待
            int n = select(max_fd + 1, &rfds, nullptr, nullptr, nullptr /*&timeout*/);

            switch (n)
            {
            case 0:
                LOG(DEBUG, "time out, %d.%d\n", timeout.tv_sec, timeout.tv_usec);
                break;
            case -1:
                LOG(ERROR, "select error\n");
                break;
            default:
                LOG(INFO, "eventr eady, n = %d\n", n);
                HandlerEvent(rfds);
                PrintDebug();
                break;
            }
        }
    }

    void PrintDebug()
    {
        cout << "fd list: ";
        for (int i = 0; i < gnum; i++)
        {
            if (fd_array[i] != gdefaultfd)
            {
                cout << fd_array[i] << " ";
            }
        }
        cout << endl;
    }

    ~SelectServer()
    {
    }

private:
    uint16_t _port;
    unique_ptr<Socket> _listensockfd;
    int fd_array[gnum]; // 辅助数组,保存合法fd
};

思路：一开始创建 SelectServer 对象时创建监听套接字，InitServer 函数把监听套接字作为要维护的第一个文件描述符。然后 Loop 函数作为入口函数，循环检测每一次读文件描述符集的变化，用于解决事件函数 HandlerEvent，HandlerEvent 函数做事件派发的工作，如果是监听套接字读事件就绪，那就是有新的连接来了，这时调用 Accepter 函数（找到维护文件描述符的数组中空的位置填入新的连接描述符 connfd），如果是普通的连接套接字读事件就绪，就进行普通IO事件，调用HandlerIO 函数（收数据，然后写回应答，可以直接写，任何一个文件描述符一开始获取时两个缓冲区都是空的。这就意味着，一开始读事件一定不就绪，但是写事件一定就绪。）到这里 HandlerEvent 函数结束，Loop 函数也结束了。

select 优缺点：

优点：能等待多个文件描述符。

缺点：每次调用 select 要重新用 fd_array 设置合法文件描述符进文件描述符集

每次调用 select 要把 fd 从用户态发到内核态，开销大（多路转接无法避免）

每次调用 select 要遍历 fd_array ：重新设置 fd_set 时，事件派发时遍历检测 fd 是否就绪，为新连接找到合适的 fd_array 位置

select 存储的文件描述符太少

3、方案二：poll

解决了文件描述符太少和每次都要重新设置 fd_set 问题。

返回值：大于0，有几个文件描述符就绪。等于0，等待超时。小于0，poll出错。

timeout：以毫秒为单位的超时时间，只作输入。等于0，非阻塞等待。小于0，阻塞等待。大于0，先阻塞等待，有新事件就返回，没有就超时阻塞等待。

struct pollfd：

fd：要关心的文件描述符

events：用户到内核，告诉内核你要关心的指定文件描述符的指定事件。

revents：内核到用户，内核返回给用户关心的指定文件描述符的指定事件已经就绪。

因为事件被定义成了宏，所以多个事件的添加用 | 来连接，判断返回的事件里面有无指定事件用 & 判断。

事件介绍：

fds：数组 struct pollfd 的起始位置指针。

nfds：数组元素个数，理论上无限多。

poll 优缺点：

优点：包含了 events 和 revents 分别表示用户到内核和内核到用户，所以解决每次都要重新设置 fd_set 问题。poll没有最大文件描述符的限制，只取决于用户想创建多少个 pollfd

缺点：底层是OS帮我们做的循环检测，还是要遍历。每次调用要把 fd 从用户态拷贝到内核态。

4、方案三：epoll

（1）接口介绍

a、epoll创建

返回值：如果创建成功返回文件描述符，失败返回-1

size：废弃字段，填大于0就行。

b、epoll控制用户 -> 内核

epfd：epoll_create返回的文件描述符

op：操作epoll的选项：

EPOLL_CTL_ADD：增加一个特定描述符fd的特定事件。

EPOLL_CTL_MOD：修改一个特定描述符fd的特定事件。

EPOLL_CTL_DEL：删除一个特定描述符fd的特定事件。

fd：op中的特定描述符，上层关心的文件描述符。

struct epoll_event：

epoll_data是枚举类型，四选一，一般选fd

events可以选择：

event：一般先创建一个struct epoll_event，初始化后取地址做参数。当op == EPOLL_CTL_DEL时，event是nullptr，即无事件。

c、epoll等待内核 -> 用户

events：一般创建一个struct epoll_event数组来接收内核告诉用户有多少个事件就绪。

maxevents：struct epoll_event数组大小

返回值：大于0，有几件事件就绪。等于0超时。小于0出错。

（2）原理

检测底层是否有事件就绪，如果有红黑树中关心的事件就绪，就形成节点了链入到就绪队列，epoll_wait 会将就绪事件一次严格按顺序放入用户定义的缓冲区，上层用户拿到的一定是有序的有效的待处理事件。

（3）内核级理解

所以我们就理解为什么epoll_create返回的是文件描述符了，通过文件描述符就能找到最后的epoll模型。

但是我们还要深入理解一下就绪队列和红黑树里面的节点关系，因为我们明显发现其实节点里面存储的数据应该不会有大差别（实际上是没有差别）

Linux内核中链表实现

所以其实不同于以前的节点，内核中实现的节点只有连接节点，每一个不同的以链表为基础的存储节点只要带上通用的连接节点就行，数据自己定。

知道link结构体怎么算出task_struct结构体的起始地址？

（4）epoll代码案例

#pragma once
#include <string>
#include <iostream>
using namespace std;
#include "Socket.hpp"
using namespace socket_ns;
#include "log.hpp"
using namespace log_ns;
#include "inetAddr.hpp"
#include <sys/epoll.h>

class EpollServer
{
    const static int num = 128;
    const static int size = 128;

public:
    EpollServer(uint16_t port)
        : _port(port), _listensock(make_unique<TcpSocket>())
    {
        _listensock->BuildListenSocket(port);
        _epfd = ::epoll_create(size);
        if (_epfd < 0)
        {
            LOG(FATAL, "epoll create error\n");
            exit(1);
        }
        LOG(INFO, "epoll create success, epfd:%d\n", _epfd);
    }

    void InitServer()
    {
        // 先添加listen套接字
        struct epoll_event ev;
        // 新连接到来读事件就绪
        ev.events = EPOLLIN;
        ev.data.fd = _listensock->Sockfd();
        int n = epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock->Sockfd(), &ev);
        if (n < 0)
        {
            LOG(FATAL, "epoll create error\n");
            exit(2);
        }
        LOG(INFO, "epoll_ctl success, add new sockfd:%d\n", _listensock->Sockfd());
    }

    string EventsToString(uint32_t events)
    {
        string eventstr;
        if (events & EPOLLIN)
            eventstr = "EPOLLIN";
        if (events & EPOLLOUT)
            eventstr += "|EPOLLOUT";
        return eventstr;
    }

    void Accepter()
    {
        InetAddr addr;
        int connfd = _listensock->Accepter(&addr);
        if (connfd < 0)
        {
            LOG(ERROR, "accept error\n");
            return;
        }
        LOG(INFO, "get a new link, %d, client info:%s:%d\n", connfd, addr.Ip(), addr.Port());
        // 新连接不能读，但是一定能写
        // 两个缓冲区都是空，读事件一定不就绪
        // 新连接connfd放入epoll
        struct epoll_event ev;
        ev.data.fd = connfd;
        ev.events = EPOLLIN;
        ::epoll_ctl(_epfd, EPOLL_CTL_ADD, connfd, &ev);
        LOG(INFO, "epoll_ctl success, add new sockfd:%d\n", connfd);
    }

    void HandlerIO(int fd)
    {
        char buffer[4096];
        int n = recv(fd, buffer, sizeof buffer - 1, 0);
        if (n > 0)
        {
            buffer[n] = 0;
            cout << buffer;
        }
        else if (n == 0)
        {
            LOG(INFO, "client quit, close fd,:%d\n", fd);
            // 从epoll中移除fd，这个fd必须健康合法，所以先移除后关闭
            epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr);
            // 关闭fd
            ::close(fd);
        }
        else
        {
            LOG(ERROR, "recv error, close fd,:%d\n", fd);
            // 从epoll中移除fd，这个fd必须健康合法，所以先移除后关闭
            epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr);
            // 关闭fd
            ::close(fd);
        }
    }

    void HandlerEvent(int n)
    {
        for (int i = 0; i < n; i++)
        {
            int fd = revs[i].data.fd;
            uint32_t revents = revs[i].events;
            LOG(INFO, "%d 有事件就绪，事件是%s\n", EventsToString(revents).c_str());
            if (revents & EPOLLIN)
            {
                // 处理连接套接字
                if (fd == _listensock->Sockfd())
                    Accepter();
                // 处理普通套接字
                else
                    HandlerIO(fd);
            }
        }
    }

    void Loop()
    {
        int timeout = 1000;
        while (1)
        {
            int n = ::epoll_wait(_epfd, revs, num, timeout);
            switch (n)
            {
            case 0:
                LOG(INFO, "epoll time out\n");
                break;
            case -1:
                LOG(ERROR, "epoll error\n");
                break;
            default:
                LOG(INFO, "haved event happend, n = %d\n", n);
                HandlerEvent(n);
                break;
            }
        }
    }

    ~EpollServer()
    {
        if (_epfd >= 0)
            ::close(_epfd);
        _listensock->Close();
    }

private:
    uint16_t _port;
    unique_ptr<Socket> _listensock;
    int _epfd;
    // 定义epoll缓冲区
    struct epoll_event revs[num];
};