0


[计算机网络]--I/O多路转接之poll和epoll

前言

作者:小蜗牛向前冲

名言:我可以接受失败,但我不能接受放弃

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

本期学习:poll函数的相关接口,poll函数是如何实现多路转接的,epoll函数的学习,epoll函数的工作原理,poll函数和epoll函数服务器的实现。

在学习poll和npoll之前,我们先来回顾一下select的特点

  1. select能同时等待的文件fd是有上限的,除非重新改内核,否则无法解决

2.必须借助第三方数组,来维护合法的fd
3. select的大部分参数是输入输出型的,调用select前,要重新设置所有的fd,调用之后,我们还有检查更新所有的fd.这带来的就是遍历的成本―--用户
4. select为什么第一个参数是最大fd+1呢?确定遍历范围--内核层面
5. select采用位图,用户->内核,内核->用户,来回的进行数据拷贝,拷贝成本的问题

更加详细的回顾:传送门

为了解决select在IO时会fd上限和每次调用都要重新设定关心fd,所以我们的poll函数就上亮登场了。

一、poll函数基础知识

1、poll函数接口

头文件

 #include <poll.h>

**函数接口 **

 int poll(struct pollfd *fds, nfds_t nfds, int timeout);

** 参数说明**

  • fds是一个poll函数监听的结构列表. 每一个元素中, 包含了三部分内容: 文件描述符, 监听的事件集合, 返 回的事件集合.
  • nfds表示fds数组的长度.
  • timeout表示poll函数的超时时间, 单位是毫秒(ms)

**

pollfd

结构体的定义:**

struct pollfd {
    int fd;           // 文件描述符
    short events;     // 要监视的事件(输入)
    short revents;    // 实际发生的事件(输出)
};

**

events

字段是要监视的事件**。它是一个位掩码,可以包括以下常量之一或多个:

  • OLLIN:文件描述符可读。
  • POLLOUT:文件描述符可写。
  • POLLPRI:文件描述符有紧急数据可读。
  • POLLERR:发生错误。
  • POLLHUP:文件描述符挂起(连接关闭)。
  • POLLNVAL:文件描述符不是一个打开的文件

**

revents

字段是实际发生的事件**。它是

poll()

函数返回后被填充的字段,表示文件描述符上实际发生的事件。

返回结果

  • 返回值小于0, 表示出错;
  • 返回值等于0, 表示poll函数等待超时;
  • 返回值大于0, 表示poll由于监听的文件描述符就绪而返回

2、poll函数多路转接的实现

多路转接的实现

  • 使用 poll() 函数时,通常需要创建一个**pollfd 数组**,每个元素描述一个要监视的文件描述符及其关注的事件。
  • 然后,将该数组传递给 poll() 函数,并指定超时时间(或者设置为 -1 表示永远等待),poll() 函数会阻塞直到有事件发生或超时。
  • 返回后,程序可以检查每个 pollfd 结构体的 revents 字段来判断每个文件描述符上实际发生的事件。

** poll的优点**

不同与select使用三个位图来表示三个fdset的方式,poll使用一个pollfd的指针实现.

  • pollfd结构包含了要监视的event和发生的event,不再使用select“参数-值”传递的方式. 接口使用比 select更方便.
  • poll并没有最大数量限制 (但是数量过大后性能也是会下降).

**poll的缺点 **

poll中监听的文件描述符数目增多时

  • 和select函数一样,poll返回后,需要轮询pollfd来获取就绪的描述符.
  • 每次调用poll都需要把大量的pollfd结构从用户态拷贝到内核中.
  • 同时连接的大量客户端在一时刻可能只有很少的处于就绪状态, 因此随着监视的描述符数量的增长, 其效率也会线性下降.

二、poll服务器的实现

pollServer.hpp

#pragma once

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

namespace poll_ns
{
    static const int defaultport = 8081;
    static const int num = 2048;
    static const int defaultfd = -1;

    using func_t = std::function<std::string(const std::string &)>;
    class PollServer
    {
    public:
        PollServer(func_t f, int port = defaultport) : _func(f), _port(port), _listensock(-1), _rfds(nullptr)
        {
        }
        void ResetItem(int i)
        {
            _rfds[i].fd = defaultfd;
            _rfds[i].events = 0;
            _rfds[i].revents = 0;
        }
        void initServer()
        {
            _listensock = Sock::Socket();
            Sock::Bind(_listensock, _port);
            Sock::Listen(_listensock);
            _rfds = new struct pollfd[num];
            for (int i = 0; i < num; i++)
            {
                ResetItem(i);
            }
            _rfds[0].fd = _listensock;
            _rfds[0].events = POLLIN;
        }

        void Print()
        {
            std::cout << "fd list: ";
            for (int i = 0; i < num; i++)
            {
                if (_rfds[i].fd != defaultfd)
                    std::cout << _rfds[i].fd << " ";
            }
            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 吗?不能,整个代码,只有poll有资格检测事件是否就绪
            // 将新的sock 托管给poll!
            // 将新的sock托管给poll本质,其实就是将sock,添加到fdarray数组中即可!

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

        void Recver(int pos)
        {
            logMessage(DEBUG, "in Recver");

            // 读取request
            char buffer[1024];
            ssize_t s = recv(_rfds[pos].fd, buffer, sizeof(buffer) - 1, 0);

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

            // 2. 处理request
            std::string response = _func(buffer);

            // 3. 返回response
            // write bug
            write(_rfds[pos].fd, response.c_str(), response.size());

            logMessage(DEBUG, "out Recver");
        }
        // 1. handler event rfds 中,不仅仅是有一个fd是就绪的,可能存在多个
        // 2. 我们的poll目前只处理了read事件
        void HandlerReadEvent()
        {
            // 遍历fdarray数组
            for (int i = 0; i < num; i++)
            {
                // 过滤非法的fd
                if (_rfds[i].fd == defaultfd)
                    continue;
                if (!(_rfds[i].events & POLLIN))
                    continue;
                if (_rfds[i].fd == _listensock && (_rfds[i].revents & POLLIN))
                    Accepter(_listensock);
                else if (_rfds[i].revents & POLLIN)
                    Recver(i);
                else
                {
                }
            }
        }

        void start()
        {
            int timeout = -1;
            for (;;)
            {
                int n = poll(_rfds, num, timeout);
                switch (n)
                {
                case 0:
                    logMessage(NORMAL, "timeout...");
                    break;
                case -1:
                    logMessage(WARNING, "poll error, code: %d, err string: %s", errno, strerror(errno));
                    break;
                default:
                    // 说明已经有事情就绪了
                    logMessage(NORMAL, "have event ready!");
                    HandlerReadEvent();
                    break;
                }
            }
        }
        ~PollServer()
        {
            if (_listensock < 0)
                close(_listensock);
            if (_rfds)
                delete[] _rfds;
        }

    private:
        int _port;
        int _listensock;
        struct pollfd *_rfds;
        func_t _func;
    };
}

为了解决pool的缺点,程序员们又设计出了npoll

三、epoll函数的基础知识

按照man手册的说法: 是为处理大批量句柄而作了改进的poll. 它是在2.5.44内核中被引进的(epoll(4) is a new API introduced in Linux kernel 2.5.44) 它几乎具备了之前所说的一切优点,被公认为Linux2.6下性能最好的多路I/O就绪通知方法

1、epoll的相关系统调用

epoll 有3个相关的系统调用.

epoll_create 创建一个epoll的句柄(创建了epoll模型)

int epoll_create(int size);
  • 自从linux2.6.8之后,size参数是被忽略的.
  • 用完之后, 必须调用close()关闭

** epoll_ct :**epoll的事件注册函数

int epoll_ctl(int epfd, int op, int fd, struct epoll_event* event)
  • 它不同于select()是在监听事件时告诉内核要监听什么类型的事件, 而是在这里先注册要监听的事件类型.
  • 第一个参数是epoll_create()的返回值(epoll的句柄).
  • 第二个参数表示动作,用三个宏来表示.
  • 第三个参数是需要监听的fd.
  • 第四个参数是告诉内核需要监听什么事件

** 第二个参数的取值:(增改删)**

  • EPOLL_CTL_ADD :注册新的fd到epfd中;
  • EPOLL_CTL_MOD :修改已经注册的fd的监听事件;
  • EPOLL_CTL_DEL :从epfd中删除一个fd;

struct epoll_event结构如下

typedef union epoll_data {
    void    *ptr;
    int      fd;
    uint32_t u32;
    uint64_t u64;
} epoll_data_t;

struct epoll_event {
    uint32_t events;   // 事件类型(输入)
    epoll_data_t data; // 用户数据(输出)
};
poll_data_t

是一个联合体,用于在

struct epoll_event

中传递用户数据

events可以是以下几个宏的集合:

  • EPOLLIN : 表示对应的文件描述符可以读 (包括对端SOCKET正常关闭);
  • EPOLLOUT : 表示对应的文件描述符可以写;
  • EPOLLPRI : 表示对应的文件描述符有紧急的数据可读 (这里应该表示有带外数据到来); EPOLLERR : 表示对应的文件描述符发生错误;
  • EPOLLHUP : 表示对应的文件描述符被挂断;
  • EPOLLET : 将EPOLL设为边缘触发(Edge Triggered)模式, 这是相对于水平触发(Level Triggered)来说的.
  • EPOLLONESHOT:只监听一次事件, 当监听完这次事件之后, 如果还需要继续监听这个socket的话, 需要 再次把这个socket加入到EPOLL队列里.

epoll_wait:收集在epoll监控的事件中已经发送的事件

int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
  • 参数events是分配好的epoll_event结构体数组.
  • epoll将会把发生的事件赋值到events数组中 (events不可以是空指针,内核只负责把数据复制到这个 events数组中,不会去帮助我们在用户态中分配内存).
  • maxevents告之内核这个events有多大,这个 maxevents的值不能大于创建epoll_create()时的size.
  • 参数timeout是超时时间 (毫秒,0会立即返回,-1是永久阻塞).
  • 如果函数调用成功,返回对应I/O上已准备好的文件描述符数目,如返回0表示已超时,返回小于0表示函 数失败

2、epoll工作原理

当某一进程调用epoll_create方法时,Linux内核会创建一个eventpoll结构体,这个结构体中有两个成员与epoll的使用方式密切相关

  • 每一个epoll对象都有一个独立的eventpoll结构体,用于存放通过epoll_ctl方法向epoll对象中添加进来的事件
  • 这些事件都会挂载在红黑树中,如此,重复添加的事件就可以通过红黑树而高效的识别出来(红黑树的插入时间效率是lgn,其中n为树的高度)
  • 而所有添加到epoll中的事件都会与设备(网卡)驱动程序建立回调关系,也就是说,当响应的事件发生时 会调用这个回调方法
  • 这个回调方法在内核中叫ep_poll_callback,它会将发生的事件添加到rdlist双链表中
struct eventpoll{ 
 .... 
 /*红黑树的根节点,这颗树中存储着所有添加到epoll中的需要监控的事件*/ 
 struct rb_root rbr; 
 /*双链表中则存放着将要通过epoll_wait返回给用户的满足条件的事件*/ 
 struct list_head rdlist; 
 .... 
}

在epoll中,对于每一个事件,都会**建立一个epitem结构体. **

struct epitem{ 
 struct rb_node rbn;//红黑树节点 
 struct list_head rdllink;//双向链表节点 
 struct epoll_filefd ffd; //事件句柄信息 
 struct eventpoll *ep; //指向其所属的eventpoll对象 
 struct epoll_event event; //期待发生的事件类型 
}
  • 当调用epoll_wait检查是否有事件发生时,只需要检查eventpoll对象中的rdlist双链表中是否有epitem 元素即可.
  • 如果rdlist不为空,则把发生的事件复制到用户态,同时将事件数量返回给用户. 这个操作的时间复杂度 是O(1).

总结一下, epoll的使用过程就是三部曲:

  • 调用epoll_create创建一个epoll句柄;
  • 调用epoll_ctl, 将要监控的文件描述符进行注册;
  • 调用epoll_wait, 等待文件描述符就绪

**epoll的优点(和 select 的缺点对应) **

  • 接口使用方便: 虽然拆分成了三个函数, 但是反而使用起来更方便高效. 不需要每次循环都设置关注的文件描述符, 也做到了输入输出参数分离开
  • 数据拷贝轻量: 只在合适的时候调用 EPOLL_CTL_ADD 将文件描述符结构拷贝到内核中, 这个操作并不频 繁(而select/poll都是每次循环都要进行拷贝)
  • 事件回调机制: 避免使用遍历, 而是使用回调函数的方式, 将就绪的文件描述符结构加入到就绪队列中, epoll_wait 返回直接访问就绪队列就知道哪些文件描述符就绪. 这个操作时间复杂度O(1). 即使文件描述 符数目很多, 效率也不会受到影响.
  • 没有数量限制: 文件描述符数目无上限

**epoll工作方式 **

感性理解

你是一个网瘾少年, 天天就喜欢呆在自己的房间玩游戏,你妈饭做好了, 喊你吃饭的时候有两种方式:

    1. 如果你妈喊你一次, 你没动, 那么你妈会继续喊你第二次, 第三次...,还有一种可能是,你吃了一口,又继续去玩,你妈过一会又会开始喊你吃饭,直到你下来把饭吃完(** 水平触发)**
    1. 如果早上你妈喊你一次, 你没动你妈就不管你了,到了下午又吃饭了,你妈又会叫你一次,没来你妈也不管你,到了晚上...(边缘触发)

epoll有2种工作方式-水平触发(LT)和边缘触发(ET)

水平触发Level Triggered 工作模式

epoll默认状态下就是LT工作模式:

  • 当epoll检测到socket上事件就绪的时候, 可以不立刻进行处理. 或者只处理一部分.
  • 如上面的例子(你妈喊你吃饭类似), 由于只读了1K数据, 缓冲区中还剩1K数据, 在第二次调用 epoll_wait 时, epoll_wait 仍然会立刻返回并通知socket读事件就绪.
  • 直到缓冲区上所有的数据都被处理完, epoll_wait 才会立刻返回.
  • 支持阻塞读写和非阻塞读

简单点来说只要底层数据没有读完就,epoll就会一直通知用户要读取数据LT

**边缘触发Edge Triggered工作模式 **

如果我们在第1步将socket添加到epoll描述符的时候使用了EPOLLET标志, epoll进入ET工作模式

  • 当epoll检测到socket上事件就绪时, 必须立刻处理.
  • 如上面的例子(你妈喊你吃饭类似), 虽然只读了1K的数据, 缓冲区还剩1K的数据, 在第二次调用 epoll_wait 的时候, epoll_wait 不会再返回了.
  • 也就是说, ET模式下, 文件描述符上的事件就绪后, 只有一次处理机会.
  • ET的性能比LT性能更高( epoll_wait 返回的次数少了很多). Nginx默认采用ET模式使用epoll.
  • 只支持非阻塞的读写

ET就是底层数据没有读完,epoll也不会通知用户在去读取数据,除非底层数据变化的时候(增多),才会在通知用户一次。

对于ET模式的效率是非常高的,因为对于epoll在此模式下只有底层数据变化了才会通知用户去读数据,但是我们不知道数据读完了,所以就会倒逼着用户将本轮就绪的数据全部读取到上层(循环读取),所说,一般的fd是阻塞式的,但是在ET模式下的fd必须是非阻塞式的。

倒逼着用户将本轮就绪的数据全部读取到上层体现:

  • 不仅仅在通知机制上,尽快让上层把数据都读走。
  • 也让T CP可以给对方发生提供了一个更大的窗口大小,让对方更新出更大的滑动窗口
  • 让底层的数据发送效率更好,其中TCP6位标志位中的PUH提示接收端应用程序立刻从TCP缓冲区把数据读走 。

** select和poll其实也是工作在LT模式下.**epoll既可以支持LT, 也可以支持ET。

四、epoll服务器

epollServer.hpp

#pragma once

#include <iostream>
#include <string>
#include <cstring>
#include <functional>
#include <sys/epoll.h>
#include "err.hpp"
#include "log.hpp"
#include "sock.hpp"

namespace epoll_ns
{
    static const int defaultport = 8888;
    static const int size = 128;
    static const int defaultvalue = -1;
    static const int defalultnum = 64;

    using func_t = std::function<std::string(const std::string &)>;
    class EpollServer
    {
    public:
        EpollServer(func_t f, uint16_t port = defaultport, int num = defalultnum)
            : func_(f), _num(num), _revs(nullptr), _port(port), _listensock(defaultvalue), _epfd(defaultvalue)
        {
        }

        void initServer()
        {
            _listensock = Sock::Socket();
            Sock::Bind(_listensock, _port);
            Sock::Listen(_listensock);
            // 1  创建epoll模型
            _epfd = epoll_create(size);
            if (_epfd < 0)
            {
                logMessage(FATAL, "epoll create error: %s", strerror(errno));
                exit(EPOLL_CREATE_ERR);
            }

            // 2 添加listensocket到epoll中
            struct epoll_event ev;
            ev.events = EPOLLIN;
            ev.data.fd = _listensock;
            epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock, &ev);

            // 3 申请就绪事情的空间
            _revs = new struct epoll_event[_num];
            logMessage(NORMAL, "init server success!");
        }

        void HandlerEvent(int readyNum)
        {
            logMessage(DEBUG, "HandlerEvent in");
            for (int i = 0; i < readyNum; i++)
            {
                uint32_t events = _revs[i].events;
                int sock = _revs[i].data.fd;
                if (sock == _listensock && (events & EPOLLIN))
                {
                    //_listensock读数据就绪,获取新连接
                    std::string clinetip;
                    uint16_t clinetport;
                    int fd = Sock::Accept(sock, &clinetip, &clinetport);
                    if (fd < 0)
                    {
                        logMessage(WARNING, "accept error");
                        continue;
                    }
                    // 获取fd成功,放入到epoll中等
                    struct epoll_event ev;
                    ev.events = EPOLLIN;
                    ev.data.fd = fd;
                    epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &ev);
                }
                else if (events & EPOLLIN)
                {
                    // 普通事情准备好了
                    char buffer[1024];
                    // 这里是有BUG的这里不能保证读取是完整信息,这里先不解决
                    int n = recv(sock, buffer, sizeof(buffer) - 1, 0);
                    if (n > 0)
                    {
                        buffer[n] = 0;
                        logMessage(DEBUG, "client# %s", buffer);

                        // 应答
                        std::string respose = func_(buffer);
                        send(sock, respose.c_str(), respose.size(), 0);
                    }
                    else if (n == 0)
                    {
                        // 客户端退出了
                        epoll_ctl(_epfd, EPOLL_CTL_DEL, sock, nullptr);
                        close(sock);
                        logMessage(NORMAL, "client quit");
                    }
                    else
                    {
                        epoll_ctl(_epfd, EPOLL_CTL_DEL, sock, nullptr);
                        close(sock);
                        logMessage(ERROR, "recv error, code: %d, errstring: %s", errno, strerror(errno));
                    }
                }
                else
                {
                }
            }
            logMessage(DEBUG, "HandlerEvent out");
        }
        void start()
        {
            int timeout = -1;
            for (;;)
            {
                int n = epoll_wait(_epfd, _revs, _num, timeout);
                switch (n)
                {
                case 0:
                    logMessage(NORMAL, "timeout...");
                case -1:
                    logMessage(WARNING, "epoll_wait failed,code:%d,errstring: %s", errno, strerror(errno));
                // 到这里事件都就绪了
                default:
                    logMessage(NORMAL, "have event ready");
                    HandlerEvent(n);
                    break;
                }
            }
        }
        ~EpollServer()
        {
            if (_listensock != defaultvalue)
                close(_listensock);
            if (_epfd != defaultvalue)
                close(_epfd);

            if (_revs)
                delete[] _revs;
        }

    private:
        uint16_t _port;
        int _listensock;
        int _epfd;
        struct epoll_event *_revs;
        int _num;
        func_t func_;
    };
}

在使用

telnet

命令连接到服务器后,你可以通过几种方式来退出连接:

  1. 发送 Telnet 命令序列:- 在 telnet 连接中,你可以发送一些特殊的 Telnet 命令来结束连接。- 在连接中直接输入 Ctrl+],然后输入 quit 或者 exit,然后按回车键。
  2. 直接关闭 telnet 客户端:- 如果你不介意强制终止连接,可以直接关闭或者终止 telnet 客户端。- 在大多数系统中,你可以使用 Ctrl+C 或者 Ctrl+D 来中断当前运行的命令或程序。在这种情况下,这将关闭 telnet 客户端并且终止连接。

本文转载自: https://blog.csdn.net/qq_61552595/article/details/136421197
版权归原作者 小蜗牛~向前冲 所有, 如有侵权,请联系我们删除。

“[计算机网络]--I/O多路转接之poll和epoll”的评论:

还没有评论