【Linux】IO多路转接技术Epoll的使用

news/2024/12/22 19:40:48/

【Linux】IO多路转接技术Epoll的使用

文章目录

  • 【Linux】IO多路转接技术Epoll的使用
    • 前言
    • 正文
      • 接口介绍
      • 工作原理
      • LT模式与ET模式
        • 边缘触发（ET）
        • 水平触发（LT）
      • 理解ET模式和非阻塞文件描述符
      • ET模式`epoll`实现TCP服务器
        • 简单地封装`epoll`系统调用
        • 封装网络套接字接口
        • 编写TCP服务器

前言

​ 在学习epoll之前，我们首先了解一下Linux中的多路复用技术：

在Linux系统中，IO多路复用是一种重要的技术，它允许一个进程同时监视多个文件描述符，一旦某个描述符准备好进行读取（通常是读就绪或写就绪），内核会通知该进程进行相应的读写操作。这样，我们可以有效地处理多个I/O事件而不需要创建多个线程或进程，从而减小系统开销。这种可以同时监视多个文件描述符的技术经常用于会维护很多文件描述符的高并发网络编程。其中多路复用共有三种方案，分别是select、poll、epoll，而epoll是目前的多路复用技术中最先进且常用的技术。

正文

接口介绍

​ epoll 是 Linux 下的一种 I/O 事件通知机制，用于高效地处理大量的文件描述符（sockets、文件等）上的 I/O 事件。epoll 提供了三个主要的接口函数：

  1. epoll_create：创建一个 epoll 实例。

    • 功能：创建一个 epoll 实例，返回一个文件描述符，用于标识该 epoll 实例。
    • 参数：无参数或者一个整数，指定要返回的文件描述符的数量（在新的内核版本中该参数已经被忽略）。
    • 返回值：返回一个指向 epoll 实例的文件描述符，如果出错，返回 -1。

  2. epoll_ctl：控制 epoll 实例上的事件。

    • 功能：向 epoll 实例中添加、修改或删除感兴趣的事件。
    • 参数：
      • epfd：epoll 实例的文件描述符。
      • op：要执行的操作，可以是 EPOLL_CTL_ADD、EPOLL_CTL_MOD 或 EPOLL_CTL_DEL。
      • fd：需要添加、修改或删除的文件描述符。
      • event：指向 epoll_event 结构体数组的指针，描述了关联于文件描述符 fd 的事件。
    • 返回值：成功时返回 0，失败时返回 -1。

  3. epoll_wait：等待 epoll 实例上的事件发生。

    • 功能：阻塞等待 epoll 实例上注册的文件描述符上的事件发生。
    • 参数：
      • epfd：epoll 实例的文件描述符。
      • events：指向 epoll_event 结构体数组的指针，用于存储发生的事件。
      • maxevents：events 数组的大小，指定最多可以存储多少个事件。
      • timeout：等待的超时时间，单位为毫秒；如果传入 -1，则表示永远等待直到有事件发生。
    • 返回值：返回发生的事件的数量，如果超时则返回 0，如果出错则返回 -1。

工作原理

​ 在大致了解了epoll的基础知识以及它的系统调用之后，我们继续来了解epoll的底层工作原理。

​ 为了获取更高的性能以及支持更大的并发连接，epoll的底层采用了一颗红黑树以及一个就绪队列对事件进行管理。让我们画图进行表示：
在这里插入图片描述

​ 每当我们调用epoll_create，系统就会在内核中帮我们创建出一个这样的结构体对应着上面的结构（进程使用这个数据结构的方式是将它的指针放入到进程控制块(task_struct)中的文件描述符所对应的struct file中）：

struct eventpoll
{	..../*红黑树的根节点，这颗树中存储着所有添加到epoll中的需要监控的事件*/ struct rb_root rbr;/*双链表中则存放着将要通过epoll_wait返回给用户的满足条件的事件*/ struct list_head rdlist;.... 
};

​ 我们由此可以发现epoll的优势：

  • epoll 使用基于事件驱动的模型，可以有效地处理大量并发连接，是更高效的事件通知机制，避免了 poll 中遍历线性数组的性能瓶颈。在大规模并发连接的场景下，epoll 的性能表现更为出色。
  • epoll 在内核中利用红黑树（Red-Black Tree）来管理关心的事件和文件描述符，而不是通过传统的遍历线性数组的方式进行管理。这意味着 epoll 可以轻松处理数以万计的连接，而不会随着连接数量的增加而有较大的性能下降。
  • 不同于select和poll,在 epoll 中，内核直接将数据从内核缓冲区拷贝到用户空间缓冲区，避免了额外的数据拷贝过程，提高了数据传输的效率。

LT模式与ET模式

​ epoll 支持两种事件触发方式：边缘触发（Edge-Triggered，简称 ET）和水平触发（Level-Triggered，简称 LT）。它们之间有一些区别，下面我会详细介绍它们以及各自的优缺点。

边缘触发（ET）
  • 触发条件：只有当事件状态发生变化，由不可读/不可写变为可读/可写时，epoll 才会通知用户空间。
  • 优点：
    1. 对于大量并发连接，ET 触发模式下的事件通知更高效，因为它只在状态变化时通知，避免了重复通知。
    2. ET的触发条件倒逼程序员每次通知时都需要一次性读完本轮的所有数据，于是其更适用于非阻塞IO，能够充分地利用系统资源。
  • 缺点：
    1. ET触发模式的代码复杂度更大。
    2. 用户需要及时处理完整个数据流，否则可能会错过部分数据，因为下次的触发条件是状态的变化。
    3. 对于文件描述符的读写操作，必须一直读取/写入直至返回 EAGAIN 错误，否则可能会错过数据。
水平触发（LT）
  • 触发条件：只要文件描述符处于可读/可写状态，epoll 就会不断通知用户空间。
  • 优点：
    1. 对于普通的 I/O 操作，LT 触发模式下更容易使用，因为不需要像 ET 模式那样严格控制数据的读写。
    2. 用户处理数据时可以按照自己的节奏进行，不必担心错过部分数据。
  • 缺点：
    1. 在大量并发连接的情况下，LT 模式可能会导致频繁的事件通知，增加了系统开销。
    2. 容易出现事件饥饿（Event Starvation）问题，即某些事件一直处于就绪状态但得不到及时处理。

​ 总结来说：

  • ET 触发模式适用于高性能、高并发的场景，能够更精确地通知事件状态的变化，但要求用户处理数据时要及时且完整。
  • LT 触发模式更适合普通的 I/O 操作，用户可以按照自己的节奏进行数据处理，但可能会出现频繁的事件通知和事件饥饿问题。

理解ET模式和非阻塞文件描述符

​ 使用 ET 模式的 epoll, 需要将文件描述设置为非阻塞。 这个不是接口上的要求, 而是 “工程实践” 上的要求。 假设这样的场景: 服务器接受到一个10k的请求, 会向客户端返回一个应答数据. 如果客户端收不到应答, 不会发送第 二个10k请求。

在这里插入图片描述

​ 如果服务端写的代码是阻塞式的read, 并且一次只 read 1k 数据的话(read不能保证一次就把所有的数据都读出来, 参考 man 手册的说明, 可能被信号打断), 剩下的9k数据就会待在缓冲区中。

在这里插入图片描述

​ 此时由于 epoll 是ET模式, 并不会认为文件描述符读就绪。epoll_wait 就不会再次返回. 剩下的 9k 数据会一直在缓 冲区中。直到下一次客户端再给服务器写数据，epoll_wait 才能返回。

​ 但是问题来了。

  • 服务器只读到1k个数据, 要10k读完才会给客户端返回响应数据。
  • 客户端要读到服务器的响应 。
  • 客户端发送了下一个请求, epoll_wait 才会返回, 才能去读缓冲区中剩余的数据。

在这里插入图片描述

​ 所以, 为了解决上述问题(阻塞read不一定能一下把完整的请求读完), 于是就可以使用非阻塞轮询的方式来读缓冲区, 保证一定能把完整的请求都读出来。

​ 而如果是LT就没这个问题. 只要缓冲区中的数据没读完, 就能够让 epoll_wait 返回文件描述符读就绪.

​ 由此我们也可以发现，其实LT模式其实未必比ET模式效率低，如果我们在LT模式中也每一次通知都直接取走所有数据，那么效率就与ET模式相同了。

ET模式epoll实现TCP服务器

​ 了解完关于epoll的理论，我们来尝试使用ET方式编写一个简单的TCP服务器。

简单地封装epoll系统调用

​ 首先我们对epoll接口进行简单的封装，封装成一个叫做Epoller的类：

// 使用此类继承得到的类无法被拷贝
class nocopy
{
public:nocopy(){}nocopy(const nocopy&) = delete;nocopy& operator=(const nocopy&) = delete;
};class Epoller : public nocopy
{static const int size = 128;
public:Epoller(){_epfd = epoll_create(size);if(_epfd == -1){// lg是独立编写的打印日志模块lg(Warning, "epoll create fail!, strerr: %s", strerror(errno));}else{lg(Info, "epoll create success! _epfd: %d", _epfd);}}int EpollerWait(struct epoll_event* revs, int num, int timeout = -1){//                                    timeout  非阻塞  一直阻塞int n = epoll_wait(_epfd, revs, num, timeout);return n;}void EpollerUpdate(int oper, int sockfd, uint32_t event){int n;if(oper == EPOLL_CTL_DEL){n = epoll_ctl(_epfd, oper, sockfd, nullptr);if(n < 0)lg(Error, "epoll event delete fail!");lg(Debug, "epoll delete fd : %d", sockfd);}else{struct epoll_event ev;ev.events = event;ev.data.fd = sockfd;n = epoll_ctl(_epfd, oper, sockfd, &ev);if(n < 0)lg(Error, "epollctl fail!");}}~Epoller(){if(_epfd > 0)close(_epfd);}
private:int _epfd;
};
封装网络套接字接口

​ 然后对套接字进行简单的封装，便于对listen套接字的使用：

enum{SocketError = 1,BindError,ListenError
};const int backlog = 10;class Sock
{
public:Sock(){}void Socket(){sockfd_ = socket(AF_INET, SOCK_STREAM, 0);if(sockfd_ < 0){lg(Fatal, "socket error! errno:%d errstr:%s", errno, strerror(errno));exit(SocketError);}int opt = 1;setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));}void Bind(uint16_t port){struct sockaddr_in local;memset(&local, 0, sizeof(local));local.sin_family = AF_INET;local.sin_port = htons(port);local.sin_addr.s_addr = INADDR_ANY;if(bind(sockfd_, (struct sockaddr*)&local, sizeof(local)) < 0){lg(Fatal, "bind error! errno:%d errstr:%s", errno, strerror(errno));exit(BindError);}}void Listen(){if(listen(sockfd_, backlog) < 0){lg(Fatal, "listen error, errno:%d, errorstring:%s", errno, strerror(errno));exit(ListenError);}}int Accept(std::string *clientip, uint16_t *clientport){struct sockaddr_in peer;socklen_t len = sizeof(peer);int newfd = accept(sockfd_, (struct sockaddr*)&peer, &len);if(newfd < 0){lg(Warning , "listen error, errno:%d, errorstring:%s", errno, strerror(errno));return -1;}char ipstr[64];inet_ntop(AF_INET, &(peer.sin_addr), ipstr, sizeof(ipstr));*clientip = ipstr;*clientport = ntohs(peer.sin_port);return newfd;}bool Connect(const std::string& serverip, const uint16_t serverport){struct sockaddr_in server;memset(&server, 0, sizeof(server));server.sin_family = AF_INET;server.sin_port = htons(serverport);inet_pton(AF_INET, serverip.c_str(), &(server.sin_addr));int n = connect(sockfd_, (struct sockaddr*)&server, sizeof(server));if(n < 0){lg(Warning , "connect error, errno:%d, errorstring:%s", errno, strerror(errno));return false;}return true;}void Close(){close(sockfd_);}int Fd(){return sockfd_;}~Sock(){}
private:int sockfd_;
};
编写TCP服务器

​ 最后我们对TCP服务器进行编写，其中封装了两个类，分别是：

Connection类：用于表示TCP服务器接收到的链接，其中封装了读写缓冲区以及读写异常方法。方便TCP服务器进行管理。

TcpServer类：用于表示TCP服务器，封装了Epoller以及Connection类与服务器接收到的sockfd的映射关系，以及一些服务器的方法。底层使用epoll的ET模式对事件进行关心。

class Connection;
class TcpServer;using func_t = function<void(std::weak_ptr<Connection>)>;
using excp_func_t = function<void(std::weak_ptr<Connection>)>;class Connection
{
public:Connection(int sockfd, const std::weak_ptr<TcpServer> svr_ptr): _sockfd(sockfd),_tcpsvr_ptr(svr_ptr){}void SetHandler(func_t recv_cb, func_t send_cb, excp_func_t excp_cb){_recv_cb = recv_cb;_send_cb = send_cb;_excp_cb = excp_cb;}void AppendInbuffer(const std::string &info){_inbuf += info;}void AppendOutbuffer(const std::string &info){_outbuf += info;}int Sockfd(){return _sockfd;}string &Inbuffer(){return _inbuf;}string &Outbuffer(){return _outbuf;}~Connection(){}private:// TCP链接的文件描述符以及读写缓冲区int _sockfd;std::string _inbuf;std::string _outbuf;public:// 链接的读写方法func_t _recv_cb;func_t _send_cb;excp_func_t _excp_cb;std::weak_ptr<TcpServer> _tcpsvr_ptr;std::string _ip;uint16_t _port;
};uint32_t EVENT_IN = (EPOLLIN | EPOLLET);
uint32_t EVENT_OUT = (EPOLLOUT | EPOLLET);void SetNonBlock(int fd)
{int fl = fcntl(fd, F_GETFL);if(fl < 0){perror("fcntl");return;}fcntl(fd, F_SETFL, fl | O_NONBLOCK);std::cout << "set " << fd << " nonblock." << std::endl;
}class TcpServer : public nocopy, public enable_shared_from_this<TcpServer>
{
private:// 一次接收事件的最大数量static const int num = 64;// 接收缓冲区的默认大小static const int g_buffer_size = 128;private:// Epoller的智能指针std::shared_ptr<Epoller> _epoller_ptr;// epoll事件的返回数组struct epoll_event revs[num];// listen套接字的智能指针std::shared_ptr<Sock> _listensock_ptr;// TCP链接的sockfd与对应Connection类的映射关系std::unordered_map<int, std::shared_ptr<Connection>> _connections;// 服务器的端口号uint16_t _port;bool _quit;func_t _OnMessage;public:TcpServer(uint16_t port, func_t OnMessage): _port(port),_epoller_ptr(new Epoller),_listensock_ptr(new Sock),_OnMessage(OnMessage){}void Init(){_listensock_ptr->Socket();// ET模式下每一个sockfd都需要设置非阻塞SetNonBlock(_listensock_ptr->Fd());_listensock_ptr->Bind(_port);_listensock_ptr->Listen();lg(Info, "listensock create successfully!");AddConnection(_listensock_ptr->Fd(), EVENT_IN,std::bind(&TcpServer::Accepter, this, std::placeholders::_1), nullptr, nullptr);}void Loop(){_quit = false;while (!_quit){Dispatcher(-1);// PrintConnection();}_quit = true;}void PrintConnection(){std::cout << "_connections fd list: ";for (auto &connection : _connections){std::cout << connection.second->Sockfd() << ", ";std::cout << "inbuffer: " << connection.second->Inbuffer().c_str();}std::cout << std::endl;}~TcpServer(){}public:bool IsConnectionSafe(int sockfd){auto iter = _connections.find(sockfd);if (iter == _connections.end())return false;return true;}void AddConnection(int sockfd, uint32_t event, func_t recv_cb, func_t send_cb, excp_func_t excp_cb,const string &ip = "0.0.0.0", uint16_t port = 0){// 1.首先创建出Connection对象std::shared_ptr<Connection> connection(new Connection(sockfd, shared_from_this()));connection->SetHandler(recv_cb, send_cb, excp_cb);connection->_ip = ip;connection->_port = port;// 2.然后将对应的sockfd加入进epoll内核中_epoller_ptr->EpollerUpdate(EPOLL_CTL_ADD, sockfd, event);// 3.最后将connection加入到映射中_connections[sockfd] = connection;lg(Info, "a connection create successfully! sockfd is : %d", sockfd);}void Dispatcher(int timeout){int n = _epoller_ptr->EpollerWait(revs, num, timeout);for (int i = 0; i < n; ++i){uint32_t events = revs[i].events;int sockfd = revs[i].data.fd;bool flag = IsConnectionSafe(sockfd);// 只需要处理读写事件，如果中途出现问题则直接在读写函数内进行处理if ((events & EPOLLIN) && flag)if (_connections[sockfd]->_recv_cb)_connections[sockfd]->_recv_cb(_connections[sockfd]);if ((events & EPOLLOUT) && flag)if (_connections[sockfd]->_send_cb)_connections[sockfd]->_send_cb(_connections[sockfd]);}}void Accepter(std::weak_ptr<Connection> conn){auto connection = conn.lock();while (true){struct sockaddr_in peer;socklen_t len = sizeof(peer);int sockfd = ::accept(connection->Sockfd(), (struct sockaddr *)&peer, &len);if (sockfd > 0){uint16_t peerport = ntohs(peer.sin_port);char ipbuf[128];inet_ntop(AF_INET, &(peer.sin_addr), ipbuf, sizeof(ipbuf));// ET模式下每一个sockfd都需要设置非阻塞SetNonBlock(sockfd);AddConnection(sockfd, EVENT_IN,std::bind(&TcpServer::Recver, this, std::placeholders::_1),std::bind(&TcpServer::Sender, this, std::placeholders::_1),std::bind(&TcpServer::Excepter, this, std::placeholders::_1),ipbuf, peerport);lg(Info, "get a new client link, get info-> [%s:%d], sockfd : %d", ipbuf, peerport, sockfd);}else{// 事件不就绪，直接结束轮询if (errno == EWOULDBLOCK)break;// 被信号中断，重新开始轮询else if (errno == EINTR)continue;elsebreak;}}}void Recver(std::weak_ptr<Connection> conn){auto connection = conn.lock();int sockfd = connection->Sockfd();while (true){char buff[g_buffer_size];memset(buff, 0, sizeof(buff));ssize_t n = recv(sockfd, buff, sizeof(buff), 0);if (n > 0){connection->AppendInbuffer(buff);}else if (n == 0){lg(Info, "client info [%s:%d] close the link, me too.", connection->_ip.c_str(), connection->_port);connection->_excp_cb(connection);break;}else{if (errno == EWOULDBLOCK)break;else if (errno == EINTR)continue;else{lg(Warning, "sockfd: %d, client info [%s:%d] recv error...", sockfd, connection->_ip, connection->_port);connection->_excp_cb(connection);break;}}}_OnMessage(connection);}void EnableEvent(int sockfd, bool readable, bool writeable){uint32_t events;events |= (readable ? EPOLLIN : 0) | (writeable ? EPOLLOUT : 0) | EPOLLET;_epoller_ptr->EpollerUpdate(EPOLL_CTL_MOD, sockfd, events);}void Sender(std::weak_ptr<Connection> conn){auto connection = conn.lock();auto &outbuffer = connection->Outbuffer();int sockfd = connection->Sockfd();while (true){ssize_t n = send(sockfd, outbuffer.c_str(), outbuffer.size(), 0);if (n > 0){outbuffer.erase(0, n);if (outbuffer.empty())break;}else if (n == 0){return;}else{if (errno == EWOULDBLOCK)break;else if (errno == EINTR)break;else{lg(Warning, "sockfd: %d, client info [%s:%d] recv error...", sockfd, connection->_ip, connection->_port);connection->_excp_cb(connection);break;}}}if(outbuffer.empty()){// 关闭对写事件的关心EnableEvent(sockfd, true, false);}else{// 开启对写事件的关心EnableEvent(sockfd, true, true);}}void Excepter(std::weak_ptr<Connection> conn){auto connection = conn.lock();int sockfd = connection->Sockfd();if(!IsConnectionSafe(sockfd)) return;// 1.从epoll内核中删除当前链接的fd_epoller_ptr->EpollerUpdate(EPOLL_CTL_DEL, sockfd, 0);// 2.关闭当前文件描述符close(sockfd);lg(Debug, "sockfd : %d closed.", sockfd);// 3.从映射中删除当前sockfd对应的链接_connections.erase(sockfd);lg(Debug, "remove sockfd : %d, from unordered_map.", sockfd);}
};

se
{
// 开启对写事件的关心
EnableEvent(sockfd, true, true);
}
}

void Excepter(std::weak_ptr<Connection> conn)
{auto connection = conn.lock();int sockfd = connection->Sockfd();if(!IsConnectionSafe(sockfd)) return;// 1.从epoll内核中删除当前链接的fd_epoller_ptr->EpollerUpdate(EPOLL_CTL_DEL, sockfd, 0);// 2.关闭当前文件描述符close(sockfd);lg(Debug, "sockfd : %d closed.", sockfd);// 3.从映射中删除当前sockfd对应的链接_connections.erase(sockfd);lg(Debug, "remove sockfd : %d, from unordered_map.", sockfd);
}

};

http://www.ppmy.cn/news/1438661.html

相关文章

【转载】C#集成JWT快速入门

【转载】C#集成JWT快速入门

一、JWT基本概念 JSON Web Token&#xff08;JWT&#xff09;是一种开放标准&#xff08;RFC 7519&#xff09;&#xff0c;它定义了一种紧凑的、自包含的方式&#xff0c;用于在双方之间安全地传输信息作为JSON对象。这些信息可以被验证、信任&#xff0c;因为它们是数字签名…
阅读更多...
React Hooks(常用)笔记

React Hooks(常用)笔记

一、useState&#xff08;保存组件状态&#xff09; 1、基本使用 import { useState } from react;function Example() {const [initialState, setInitialState] useState(default); } useState(保存组件状态) &#xff1a;React hooks是function组件(无状态组件) &#xf…
阅读更多...
07 流量回放实现自动化回归测试

07 流量回放实现自动化回归测试

在本模块的前四讲里&#xff0c;我向你介绍了可以直接落地的、能够支撑百万并发的读服务的系统架构&#xff0c;包含懒加载缓存、全量缓存&#xff0c;以及数据同步等方案的技术细节。 基于上述方案及细节&#xff0c;你可以直接对你所负责的读服务进行架构升级&#xff0c;将…
阅读更多...
采购管理软件：如何快速实现采购申请自动化流转？

采购管理软件：如何快速实现采购申请自动化流转？

在没有采购管理软件的情况下&#xff0c;采购申请完全依赖纸质表格、电子邮件和 excel 表等过时的工具会大大降低效率&#xff0c;甚至影响企业的利润。 但一些企业尚未准备好重塑人工采购申请流程。他们似乎没有意识到&#xff0c;在采购相关活动上花费的资金越多&#xff0c…
阅读更多...
Python浅谈清朝秋海棠叶版图

Python浅谈清朝秋海棠叶版图

1、清朝疆域概述&#xff1a; 清朝是我国最后一个封建王朝&#xff0c;其始于1616年建州女真部努尔哈赤建立后金&#xff0c;此后统一女真各部、东北地区。后又降服漠南蒙古&#xff0c;1644年入关打败农民起义军、灭南明&#xff0c;削三藩&#xff0c;复台湾。后又收外蒙&am…
阅读更多...
paddlepaddle/paddle 命令注入漏洞复现_$1500 CVE-2024-0934

paddlepaddle/paddle 命令注入漏洞复现_$1500 CVE-2024-0934

目录 1.漏洞概述 2.影响版本 3.漏洞等级 4.漏洞复现 4.1 安装漏洞环境
阅读更多...
03-JAVA设计模式

03-JAVA设计模式

设计模式GOF23 GOF23是指由设计模式经典名著《Design Patterns: Elements of Reusable Object-Oriented Software》&#xff08;中译本名为《设计模式——可复用面向对象软件的基础》&#xff09;的四位作者Erich Gamma、Richard Helm、Ralph Johnson、以及John Vlissides提出…
阅读更多...
uniapp制作分页查询功能

uniapp制作分页查询功能

效果 代码 标签中 <uni-pagination change"pageChanged" :current"pageIndex" :pageSize"pageSize" :total"pageTotle" class"pagination" /> data中 pageIndex: 1, //分页器页码 pageSize: 10, //分页器每页显示…
阅读更多...
最新文章

Copyright @ 2022~2023