Linux网络编程 --- 高级IO

前言 IO Input&&Output

read && write

1、在应用层read && write的时候，本质把数据从用户层写给OS --- 本质就是拷贝函数

2、IO = 等待 + 拷贝。

等的是：要进行拷贝，必须先判断读写事件成立。读写事件缓冲区空间满了没有，read的时候需要缓冲区有数据。

什么叫做高效的IO呢？

单位事件内我们拷贝的数据越多，越有效率。单位事件内，IO过程中，等的比重越小，IO效率就越高。几乎所有的提高IO效率的策略，本质就是减少等的时间。

一、五种IO模型

1、张三：钓鱼界的新手，一直盯着鱼钩。谁都不理，有鱼上钩，提起鱼竿。阻塞式钓鱼。

2、李四：每隔一小时定盯着鱼漂。不会卡在那里。会看手机什么的。非阻塞式钓鱼。非阻塞轮询

3、王五：鱼竿上放一个铃铛，不会卡在那里，会看手机什么的。鱼钩上有鱼了铃铛响了。信号驱动式IO

以上鱼上钩的概率不如下面的。

4、赵六：一卡车鱼竿。轮询每一个鱼竿。多路复用，多路转接。

以上的IO称为异步IO

5、田七：小王，桶，一个电话，鱼竿。田七就走了，小王开始钓鱼。田七就是钓鱼行为的发起者。田七的钓鱼方式称为异步IO。小王为操作系统。

阻塞IO vs 非阻塞IO IO = 等待 + 拷贝。非阻塞不用一直等。他们等的方式不同！

同步IO vs 异步IO 张三、李四，王五，赵六都在等待，同步就是参不参与IO。参与就是同步，不参与只是发起IO就是异步。

同步IO 和线程同步？不一样。老婆和老婆饼的关系，没有任何关系。

其中多路复用式最有效率的。异步IO的代码逻辑比较混乱，已经有新的技术代替了，比如协程。

非阻塞接口

1、设置称为非阻塞，如果底层fd数据没有就绪，recv/read/write/send，返回值会以出错的形式返回。

2、a、真的出错了， b、底层没有就绪！

3、我们通过errno区分

如果errno == EWOULDBLOCK，当前不是错了，而是fd中的数据没有就绪。

IO多路转接至select

IO = 等待 + 拷贝

select：只负责等待，而且可以等待多个文件描述符。

返回值：n > 0 有n个fd就绪了

n == 0 超时，没有错误，但是也没有fd就绪。

n < 0 等待出错。

第一个参数 nfds maxfd + 1。最大的等待文件描述符的值 + 1。

struct timeval timeout 给select设置等待方式。就是你设置的是5秒超时，那么就是进程会在select函数阻塞等待5秒，要是5秒中没有事件就绪，他会返回要是5秒中，有事件就绪了，他也会立即返回

timeout = {5，0}；每隔5s，timeout一次。在5s之内没有文件描述符就绪，返回。

= {0，0}；立马返回，非阻塞的一种。

NULL：阻塞等待。

上面的timeout是输入输出型参数。等待了2秒，如果我们设置的是3秒。select会把这个三秒返回。

fd_set 内核提供的一种数据类型。我们目前关心的fd上面的事件。读事件成立，写事件成立，异常事件。

fd_set* readfds 输入输出型参数，

输入时：用户告诉内核，我给你的一个或者多个fd，你要帮我关心fd上面的读事件，如果读时间就绪了，你要告诉我。

输出时：内核告诉用户，用户你让我关心的多个fd中，有哪些已经就绪了，用户赶紧读取吧。

从右向左比特位的位置，标识文件描述符编号，比特位的内容0，1是否需要内核关心。

其他的位图也是这样的。

返回时：比特位的内容0 or 1，用户关心的哪些fd，上面的读时间已经就绪了。

fd_set 是一张位图，让用户和内核传递fd是否就绪的信息的!

会有很多的位图操作。所以内核提供了很多的文件描述位图的操作。

直接写代码：

#include <iostream>
#include "log.hpp"
#include "Sock.hpp"
#include <sys/select.h>
#include <ctime>
static const int fd_num_max = sizeof(fd_set) * 8;
static const uint16_t defaultport = 8080;
int defaultfd = -1;
class selectserver
{
public:
    selectserver(uint16_t port = defaultport) : _port(port)
    {
        // 初始化数组
        for (int i = 0; i < fd_num_max; i++)
        {
            fd_array[i] = defaultfd;
        }
    }

    // 初始化服务器，这里应该绑定监听端口，创建套接字
    void Init()
    {
        // 绑定监听套接字。
        _listensock.Socket();
        _listensock.Bind(_port);
        _listensock.Listen();
    }
    void Accepter()
    {
        // 接收套接字
        std::string clientip;
        uint16_t clientport = 0;
        int sockfd = _listensock.Accept(&clientip, &clientport); // 这里不会被阻塞
        if (sockfd < 0)
            return;
        lg(Info, "accept success, %s: %d, sock fd: %d", clientip.c_str(), clientport, sockfd);

        // 如果大于0 把新接收的套接字放入到辅助数组中
        int pos = 1; // fd_array[0] = listensock,所以要从1开始
        for (; pos < fd_num_max; pos++)
        {
            // 寻找-1位置
            if (fd_array[pos] != defaultfd)
                continue;
            else
                break;
        }
        if (pos == fd_num_max)
        {
            lg(Warning, "server is full, close %d now!", sockfd);
            close(sockfd);
        }
        else
        {
            fd_array[pos] = sockfd;
            printFd();
        }
    }
    void Recvr(int fd, int pos)
    {
        // demo
        char buffer[1024];
        ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug?
        if (n > 0)
        {
            buffer[n] = 0;
            std::cout << "get a messge: " << buffer << std::endl;
        }
        else if (n == 0)
        {
            lg(Info, "client quit, me too, close fd is : %d", fd);
            close(fd);
            fd_array[pos] = defaultfd; // 这里本质是从select中移除
        }
        else
        {
            lg(Warning, "recv error: fd is : %d", fd);
            close(fd);
            fd_array[pos] = defaultfd; // 这里本质是从select中移除
        }
    }
    void Dispatcher(fd_set &rfds)
    {
        // 在这里我们需要对事件进行处理，从listensock套接字里连接的文件描述符也需要被我们的select等待，我们需要一个辅助
        // 数组。
        for (int i = 0; i < fd_num_max; i++)
        {
            int fd = fd_array[i];
            //std::cout << fd << std::endl;
            if (fd == defaultfd)
                continue;
            if (FD_ISSET(fd, &rfds))
            {
                if (fd == _listensock.Fd())
                {
                    Accepter();
                    std::cout << "新文件描述符完毕" << std::endl;
                }
                else
                {
                    // 不是listen套接字那就是普通的套接字进行读。
                    std::cout << "开始交互" << std::endl;
                    Recvr(fd, i);
                    
                }
            }
        }
    }
    void Start()
    {
        // 这里应该为服务器提供服务。这里我们使用select接口，让文件描述符等待连接。
        // 服务器是一个死循环。
        // 在这里我们需要把listensock的文件描述符放入到我们的辅助数组里。
        int listensock = _listensock.Fd();
        fd_array[0] = listensock;
        std::cout << fd_array[0] << std::endl;
        for (;;)
        {
            // 这里需要对listensock套接字进行等待。等待是否有链接，有连接说明该文件描述符上有事件就绪。
            // 传入的参数需要有一个位图，这个位图是输入输出型参数，用户设置需要等待的文件描述符，内核把
            // 事件就绪的文件描述符放入到其中。每次事件就绪的文件描述符，就会覆盖原来的文件描述符的问题，所以我们
            // 需要在select之前对位图重新设置。
            fd_set rfds;
            FD_ZERO(&rfds);
            // 边循环边设置我们的文件描述符到位图中，然后求出文件描述符中最大的值。
            int maxfd = fd_array[0];
            for (int i = 0; i < fd_num_max; i++)
            {
                if (fd_array[i] == defaultfd)
                    continue;
                // 找到文件描述符设置到我们的描述符位图中
                FD_SET(fd_array[i], &rfds);
                if (maxfd < fd_array[i])
                {
                    maxfd = fd_array[i];
                    lg(Info, "max fd update, max fd is: %d", maxfd);
                }
            }

            // 把listensock文件描述符设置到位图中
            // FD_SET(_listensock.Fd(), &rfds);
            // 设置等待的事件
            // struct timeval timeout = {5, 0};
            struct timeval timeout = {0, 0};
            int n = select(/*文件描述符中值最大的一个 + 1 _listensock.Fd() + 1*/ maxfd + 1, &rfds, /*不关心写事件，只关心读事件*/ nullptr, nullptr, /*&timeout*/ nullptr);
            // 判断返回值 = 0代表没有事件就绪
            //  -1 等待错误
            //> 0 有n个文件描述符的写事件就绪
            switch (n)
            {
            case 0:
                std::cout << "time out,timeout: " << timeout.tv_sec << timeout.tv_usec << std::endl;
                break;
            case -1:
                std::cerr << "select err" << std::endl;
                break;
            default:
                // 等待成功需要做什么。
                std::cout << "进入事件派发器" << std::endl;
                Dispatcher(rfds);
                break;
            }
        }
    }
    void printFd()
    {
        std::cout << "online list:";
        for (int i = 0; i < fd_num_max; i++)
        {
            if (fd_array[i] == defaultfd)
            {
                continue;
            }
            std::cout << fd_array[i] << " ";
        }
        std::cout << std::endl;
    }
    ~selectserver()
    {
        _listensock.Close();
    }

private:
    Sock _listensock;
    uint16_t _port;
    int fd_array[fd_num_max];
};

不能直接accept，检测并获取listensock上面的事件，新连接到来，等价于读事件就绪。

select，如果事件就绪，如果上层不处理，select会一直通知。select告诉你就绪了，接下来的一次读取，我们读取fd的时候，不会阻塞。

位图的大小为1024个比特位，所以一次可以等待1024个文件描述符。

select的优点是：多路转接的方案。一个进程处理多个用户的连接。

select的缺点是：

1、等待的fd是有上限的。

2、输入输出型参数比较多，数据拷贝的频率比较高。

3、输入输出型参数比较多，每次都要对关心的fd进行事件重置。

4、使用第三方数组管理用户fd，用户层需要很多次遍历，内核中检测fd事件就绪，也要遍历。

IO多路转接之poll

#include <iostream>
#include "log.hpp"
#include "Sock.hpp"
#include <poll.h>
#include <ctime>
static const int fd_num_max = 64;
static const uint16_t defaultport = 8080;
int defaultfd = -1;
int non_event = 0;
class pollserver
{
public:
    pollserver(uint16_t port = defaultport) : _port(port)
    {
        // 初始化数组
        for (int i = 0; i < fd_num_max; i++)
        {
            _fd_events[i].fd = defaultfd;
            _fd_events[i].events = non_event;
            _fd_events[i].events = non_event;
        }
    }

    // 初始化服务器，这里应该绑定监听端口，创建套接字
    void Init()
    {
        // 绑定监听套接字。
        _listensock.Socket();
        _listensock.Bind(_port);
        _listensock.Listen();
    }
    void Accepter()
    {
        // 接收套接字
        std::string clientip;
        uint16_t clientport = 0;
        int sockfd = _listensock.Accept(&clientip, &clientport); // 这里不会被阻塞
        if (sockfd < 0)
            return;
        lg(Info, "accept success, %s: %d, sock fd: %d", clientip.c_str(), clientport, sockfd);

        // 如果大于0 把新接收的套接字放入到辅助数组中
        int pos = 1; // fd_array[0] = listensock,所以要从1开始
        for (; pos < fd_num_max; pos++)
        {
            // 寻找-1位置
            if (_fd_events[pos].fd != defaultfd)
                continue;
            else
                break;
        }
        if (pos == fd_num_max)
        {
            lg(Warning, "server is full, close %d now!", sockfd);
            close(sockfd);
        }
        else
        {
            _fd_events[pos].fd = sockfd;
            _fd_events[pos].events = POLLIN;
            _fd_events[pos].revents = non_event;
            printFd();
        }
    }
    void Recvr(int fd, int pos)
    {
        // demo
        char buffer[1024];
        ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug?
        if (n > 0)
        {
            buffer[n] = 0;
            std::cout << "get a messge: " << buffer << std::endl;
        }
        else if (n == 0)
        {
            lg(Info, "client quit, me too, close fd is : %d", fd);
            close(fd);
            _fd_events[pos].fd= defaultfd; // 这里本质是从select中移除
        }
        else
        {
            lg(Warning, "recv error: fd is : %d", fd);
            close(fd);
             _fd_events[pos].fd= defaultfd; // 这里本质是从select中移除
        }
    }
    void Dispatcher()
    {
        // 在这里我们需要对事件进行处理，从listensock套接字里连接的文件描述符也需要被我们的select等待，我们需要一个辅助
        // 数组。
        for (int i = 0; i < fd_num_max; i++)
        {
            int fd = _fd_events[i].fd;
            //std::cout << fd << std::endl;
            if (fd == defaultfd)
                continue;
            //读事件就绪，且是监听sock
            if (_fd_events[i].revents & POLLIN)
            {
                if (fd == _listensock.Fd())
                {
                    Accepter();
                    std::cout << "新文件描述符完毕" << std::endl;
                }
                else
                {
                    // 不是listen套接字那就是普通的套接字进行读。
                    std::cout << "开始交互" << std::endl;
                    Recvr(fd, i);
                    
                }
            }
        }
    }
    void Start()
    {
        // 这里应该为服务器提供服务。这里我们使用select接口，让文件描述符等待连接。
        // 服务器是一个死循环。
        // 在这里我们需要把listensock的文件描述符放入到我们的辅助数组里。
        int listensock = _listensock.Fd();
        for (;;)
        {
            //把listenfd交给poll管理
            _fd_events[0].fd = listensock;
            _fd_events[0].events = POLLIN;
            _fd_events[0].revents = non_event;

            //设置超时事件3000毫秒
            int timeout = 3000;
            ssize_t n = poll(_fd_events,fd_num_max,timeout);
            switch (n)
            {
            case 0:
                std::cout << "timeout" << std::endl;
                break;
            case -1:
                std::cerr << "poll err" << std::endl;
                break;
            default:
                // 等待成功需要做什么。
                std::cout << "进入事件派发器" << std::endl;
                Dispatcher();
                break;
            }
        }
    }
    void printFd()
    {
        std::cout << "online list:";
        for (int i = 0; i < fd_num_max; i++)
        {
            if ( _fd_events[i].fd == defaultfd)
            {
                continue;
            }
            std::cout << _fd_events[i].fd << " ";
        }
        std::cout << std::endl;
    }
    ~pollserver()
    {
        _listensock.Close();
    }

private:
    Sock _listensock;
    uint16_t _port;
    struct pollfd _fd_events[fd_num_max];
};

poll只负责等待。

timeout 证书毫秒。

struct pollfd

nfds_t struct pollfd的数组大小。

将输入输出事件进行了分离。

poll的缺点：

1、遍历，用户层和内核也得遍历。效率问题。

IO多路转接之epoll

1、快速认识epoll的接口

epoll_create：

参数被忽略，设置成大于0就可以。返回值是一个文件描述符。

对epoll新增描述特定的读写事件

epoll_wait：

参数返回已经就绪的fd和事件。

已经就绪的fd的个数。

epoll_event结构体：

epoll_ctl

第一个参数epoll_create的返回值，第二个参数三个选项，第三个参数需要被控制的fd，第四个参数哪些事件被设置。

它不同于 select() 是在监听事件时告诉内核要监听什么类型的事件 , 而是在这里先注册要监听的事件类型 .

第一个参数是 epoll_create() 的返回值 (epoll 的句柄 ).

第二个参数表示动作，用三个宏来表示 .

第三个参数是需要监听的 fd.

第四个参数是告诉内核需要监听什么事 .

2、epoll的原理

用户只需要从就绪队列中获取就绪节点即可。上面的三套机制称为epoll模型。红黑树，就绪队列，回调。epoll模型被统一接入到了文件描述符表里，所以epoll_create的返回值是文件描述符。

epoll_create就是在创建struct_file，也就是创建epoll模型。

epoll_ctl 修改红黑树。

epoll_wait 中的events参数是输出型参数，把就绪队列中的节点，一个一个的放入到我们的epoll_event数组里。

优势：1、检测就绪O(1)，获取就绪O(N)

2、fd_event没有上限。

3、这颗红黑树，就是select，poll自己维护的数组。

4、返回值n，表示有几个fd就绪了，就绪事件是连续的！有返回值个。

3、快速写代码 --- echo server

#pragma once
#include <iostream>
#include <sys/epoll.h>
#include "log.hpp"
class Epoller
{
    static const int size = 128;
public:
    Epoller()
    {
        //创建epoll模型
        epollfd = epoll_create(/*这里的参数已经可以忽略了*/size);
        if(epollfd == -1)
        {
            lg(Error,"epoll create err %d %s",errno,strerror(errno));
        }
        else
        {
            lg(Info,"epoll create success %d",epollfd);
        }

    }
    //返回epoll中的就绪队列
    int waitEpoller(struct epoll_event revents[],int num)
    {
        int n = epoll_wait(epollfd,revents,num,-1);
        return n;
    }
    //通过系统调用把我们的文件描述符放入到内核中的红黑树中
    int updataEpoller(int sock,int oper,uint32_t event)
    {
        int n = 0;
        if(oper == EPOLL_CTL_DEL)
        {
            n = epoll_ctl(epollfd,oper,sock,nullptr);
            if(n!=0)
            {
                lg(Error,"epoll_ctl err");
            }
        }
        else
        {
            //EPOLL_CTL_ADD || EPOLL_CTL_MOD
            
            struct epoll_event ev;
            ev.data.fd = sock;
            ev.events = event;
            n = epoll_ctl(epollfd,oper,sock,&ev);
            if(n != 0)
            {
                lg(Error,"epoll_ctl err %d %s",errno,strerror(errno));
            }
        }
        return n;
    }
    ~Epoller()
    {
        if(epollfd >= 0)
        {
            close(epollfd);
        }
    }
private:
    int epollfd;
    int timeout{3000};
};

#include <iostream>
#include "Sock.hpp"
#include "log.hpp"
#include "nocopy.hpp"
#include "Epoller.hpp"
#include <memory>
int EVENT_IN = (EPOLLIN);
int EVENT_OUT = (EPOLLOUT);
static const uint16_t defaultport = 8080;
static const int defaultfd = -1;
class epollserver : public nocopy
{
    static const int num = 64;

public:
    epollserver(uint16_t port = defaultport) : _port(port), _listensockfd_ptr(new Sock()), _epoller_ptr(new Epoller())
    {}

    bool Init()
    {
        // 对套接字进行创建绑定监听
        _listensockfd_ptr->Socket();
        _listensockfd_ptr->Bind(_port);
        _listensockfd_ptr->Listen();
        return true;
    }
    void Accepter()
    {
        // 接收套接字
        std::string clientip;
        uint16_t clientport = 0;
        int sockfd = _listensockfd_ptr->Accept(&clientip, &clientport); // 这里不会被阻塞
        if (sockfd < 0)
            return;
        lg(Info, "accept success, %s: %d,", clientip.c_str(), clientport);

        if (sockfd > 0)
        {
            //std::cout << "Accept 放入内核中" << std::endl;
            _epoller_ptr->updataEpoller(sockfd, EPOLL_CTL_ADD, EVENT_IN);
            lg(Info, "get a new link, client info@ %s:%d %d", clientip.c_str(), clientport,sockfd);
        }
    }
    void Recvr(int fd)
    {
        // demo
        char buffer[1024];
        
        ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug?
        if (n > 0)
        {
            buffer[n] = 0;
            std::cout << "get a messge: " << buffer << std::endl;


            std::string echo_string = "server say#";
            echo_string += buffer;
            write(fd, echo_string.c_str(), echo_string.size());
        }
        else if (n == 0)
        {
            lg(Info, "client quit, me too, close fd is : %d", fd);
            close(fd);
            _epoller_ptr->updataEpoller(fd, EPOLL_CTL_DEL,0); // 这里本质是从select中移除
        }
        else
        {
            lg(Warning, "recv error: fd is : %d", fd);
            close(fd);
            _epoller_ptr->updataEpoller(fd, EPOLL_CTL_DEL,0); // 这里本质是从select中移除
        }
    }
    void Dispatcher(struct epoll_event revents[], int num)
    {
        for (int i = 0; i < num; i++)
        {
            int fd = revents[i].data.fd;
            //std::cout << fd << std::endl;
            uint32_t events = revents[i].events;
            if (events & EVENT_IN)
            {
                if (fd == _listensockfd_ptr->Fd())
                {
                    Accepter();
                }
                else
                {
                    Recvr(fd);
                }
            }
        }
    }
    void Start()
    {
        // 我们只需要把listenfd放入到红黑树中
        _epoller_ptr->updataEpoller(_listensockfd_ptr->Fd(), EPOLL_CTL_ADD, EVENT_IN);
        struct epoll_event revents[num];
        // for(int i = 0;i<num;i++)
        // {
        //     revents[i].data.fd = defaultfd;
        //     revents[i].events = 0;
        // }
        for (;;)
        {
            // 在开始阶段我们只有listensock,所以我们可以把listensock等待数据的就绪。
            // 获取内核就绪事件的fd，获取完成后，查看是否是读事件就绪。
            int n = _epoller_ptr->waitEpoller(revents, num);
            if (n > 0)
            {
                lg(Debug, "event happened, fd is : %d", revents[0].data.fd);
                Dispatcher(revents, n);
            }
            else if (n == 0)
            {
                std::cout << "time out" << std::endl;
            }
            else
            {
                std::cerr << "epoll err" << std::endl;
            }
        }
    }
    ~epollserver()
    {
    }

private:
    std::shared_ptr<Sock> _listensockfd_ptr;
    std::shared_ptr<Epoller> _epoller_ptr;
    uint16_t _port;
};

epoll的工作模式：LT和ET。

水平触发模式 LEVEL Triggered

事件到来，但是上层不处理，一直通知

边缘触发模式Edge Triggered

数据或者连接，从无到有，从有到多，变化的时候，才会通知我们一次。ET的通知效率更高。倒逼程序员，每次通知，都必须把本轮数据全部取走。循环读取，读取出错证明缓冲区没有数据了 ->fd默认是阻塞的 -> ET,所有的fd必须是non_block。不仅如此ET的IO效率也是更高的！通知一次必须把缓冲区的数据全部读走，以为着tcp回向对方通告一个更大的窗口，从而从概率上让对方一次给我发送更多的的数据。

ET vs LT ？ LT可不可以将所有的fd设置成为non_block，然后循环读取呢，通知第一次的时候，就全部取走，不就和ET一样了吗？

LT和ET的本质区别是添加就绪队列的方式，LT是次次都添加，而ET只添加一次。