C++仿muduo库高并发服务器项目：EventLoop模块
前言
本篇文章所讲的是本人的个人项目仿muduo库高并发服务器中的EventLoop模块实现部分。
EventLoop模块
-
功能：
- 事件监控管理模块，是“one thread one loop”里的loop、reactor
- 一个模块对应一个线程
-
意义：
- 负责服务器所有事件
- 每个Connection连接绑定一个EventLoop模块和线程，连接操作需在对应线程执行
-
思想：
- 监控所有连接事件，事件触发后调用回调函数处理
- 连接操作放到EventLoop线程执行
-
功能设计：
- 连接操作任务入队
- 定时任务增、刷、删
事件通知机制
eventfd
eventfd 是 Linux 提供的一种轻量级进程间通信（IPC）机制，专门用于事件通知。它创建一个文件描述符，用于在进程或线程之间传递事件计数。
#include - 创建一个 eventfd 对象，返回对应的文件描述符
initval: 初始计数值（通常为0）flags: 控制行为的标志位
flags标志位：
// 常用标志组合
int fd;
// 1. 默认阻塞模式
fd = eventfd(0, 0);
// 2. 非阻塞模式
fd = eventfd(0, EFD_NONBLOCK);
// 3. 信号量语义（每次-1）
fd = eventfd(0, EFD_SEMAPHORE);
// 4. 使用 EFD_CLOEXEC 创建 eventfd
int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
EFD_CLOEXEC
- 设置该标志后，当进程执行
exec系列函数时，文件描述符会自动关闭 - 否则，文件描述符会被新的程序继承
没有 CLOEXEC 可能导致资源泄漏:
void leaky_daemon() {
int efd = eventfd(0, 0);
// 假设这是一个长期运行的守护进程
while (1) {
pid_t pid = fork();
if (pid == 0) {
// 子进程执行某个任务
execl("/bin/some_tool", "some_tool", NULL);
// efd 被每个子进程继承，直到子进程显式关闭
// 如果 some_tool 不处理这个 fd，就造成泄漏
}
wait(NULL);
}
}
子进程可能不需要这些文件描述符，但它们会一直保持打开状态，占用系统资源（每个进程可打开的文件描述符数量是有限的）。如果父进程不断创建子进程而不关闭不必要的文件描述符，可能会导致资源耗尽。
主要特性：
- 计数器机制：内核维护一个64位无符号整数计数器
- 等待/通知机制：可配合
select/poll/epoll使用 - 零拷贝：完全在内核空间操作，无需用户空间拷贝
工作模式：
写操作（通知事件）
uint64_t val = 1;
write(fd, &val, sizeof(val)); // 增加计数器值
- 写操作将提供的值加到计数器上
- 如果计数器从0变为非0，等待该 fd 的线程/进程会被唤醒
读操作（消费事件）
uint64_t val;
read(fd, &val, sizeof(val)); // 读取当前计数值
- 读操作返回当前计数值并将其重置为0
- 如果计数器为0，读操作会阻塞（除非设为非阻塞），非阻塞模式下会返回-1并且设置错误码EAGAIN，表示没有数据可读
one thread one loop
一个线程绑定一个EventLoop
线程安全问题：
一个连接触发事件后，A线程执行操作，执行过程中该连接又在B线程中触发了事件，就会造成线程安全问题。
如何解决？
需要将一个连接的监控与一个线程绑定起来，使得该连接的事件只能在对于线程中执行。
保证操作在对应线程的方案：给 EventLoop 模块添加任务队列，对连接的所有操作进行封装，当作任务添加到任务队列而非直接执行。
应对使用者使用线程池分发处理任务，我们通过包装任务，将任务具体操作存入EventLoop中的任务队列中，当线程处理完就绪事件，再将任务从任务队列中取出，统一执行任务。
EventLoop 处理流程：
- 在线程中对描述符进行事件监控。
- 若有描述符就绪，对其进行事件处理（需保证处理回调函数中的操作在线程中）。
- 所有就绪事件处理完后，执行任务队列中的所有任务。
这样能保证连接的所有操作在一个线程中进行，不涉及线程安全问题。但是任务队列的操作存在线程安全问题，我们只需要给任务（task）的操作加一把锁即可。
源代码：
class EventLoop
{
private:
using Functor = std::function;
int _event_fd; //eventfd唤醒IO事件监控可能导致的阻塞
std::unique_ptr _event_channel;
std::thread::id _thread_id; //线程ID
Poller _poller; //进行所有描述符的事件监控
std::vector _tasks; //任务池
std::mutex _mutex; //实现任务池操作的线程安全
TimerWheel _timer_wheel; //定时器模块
private:
void RunAllTask()
{
//O（1）地拿到任务队列，执行过程中仍可以向_tasks任务队列中加入任务，互不影响
std::vector tmp;
{
std::unique_lock _lock(_mutex);
_tasks.swap(tmp);
}
for(auto& f : tmp)
{
f();
}
}
//创建eventfd
static int CreateEventFd()
{
//非阻塞模式，并且设置执行exec系列函数后关闭，防止资源泄露
int efd = eventfd(0,EFD_CLOEXEC | EFD_NONBLOCK);
if(efd < 0)
{
ERR_LOG("CREATE EVENTFD FAILED!!");
abort();
}
return efd;
}
//读取eventfd
void ReadEventFd()
{
uint64_t res = 0;
int ret = read(_event_fd,&res,sizeof(res));
if(ret < 0)
{
//被信号打断或者没读到数据
if(errno == EINTR || errno == EAGAIN)
{
return;
}
ERR_LOG("READ EVENTFD FAILED!!");
abort();
}
}
//唤醒事件
void WeakUpEventFd()
{
uint64_t val = 1;
int ret = write(_event_fd,&val,sizeof(val));
if(ret < 0)
{
//被信号打断
if(errno == EINTR)
{
return;
}
ERR_LOG("WRITE EVENTED FAILED!!");
abort();
}
}
public:
EventLoop()
:_thread_id(std::this_thread::get_id())
,_event_fd(CreateEventFd())
,_event_channel(new Channel(this,_event_fd))
,_timer_wheel(this)
{
//给eventfd添加可读回调函数，读取eventfd事件通知次数
_event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd,this));
//启动evetnfd的读事件监控
_event_channel->EnableRead();
}
//判断将要执行的任务是否处于当前线程，是则执行，不是则压入队列
void RunInLoop(const Functor& cb)
{
if(IsInLoop())
{
cb();
return;
}
QueueInLoop(cb);
return;
}
//断言当前线程和Loop是否绑定
void AssertInLoop()
{
assert(_thread_id == std::this_thread::get_id());
}
//将操作压入任务池
void QueueInLoop(const Functor& cb)
{
{
std::unique_lock _lock(_mutex);
_tasks.push_back(cb);
}
//唤醒可能因为没有事件就绪而导致的epoll阻塞
//给eventfd写入一个数据
WeakUpEventFd();
}
//判断当前线程是否是EventLoop的对应线程
bool IsInLoop()
{
return _thread_id == std::this_thread::get_id();
}
//添加/修改描述符的事件监控
void UpdateEvent(Channel* channel)
{
return _poller.UpdateEvent(channel);
}
//删除描述符的事件
void RemoveEvent(Channel* channel)
{
return _poller.RemoveEvent(channel);
}
//事件监控 -----》就绪事件处理-----》执行任务
void Start()
{
while(1)
{
//事件监控
std::vector actives;
_poller.Poll(&actives);
//事件处理
for(auto& channel : actives)
{
channel->HandleEvent();
}
//执行任务
RunAllTask();
}
}
void TimerAdd(uint64_t id,uint32_t timeout,const TaskFunc& cb)
{
_timer_wheel.TimerAdd(id,timeout,cb);
}
void TimerRefresh(uint64_t id)
{
_timer_wheel.TimerRefresh(id);
}
void TimerCancel(uint64_t id)
{
_timer_wheel.TimerCancel(id);
}
bool HasTimer(uint64_t id)
{
return _timer_wheel.HasTimer(id);
}
};
class LoopThread
{
private:
//这两个用于实现_loop获取同步关系,避免因为线程创建了,在_loop实例化之前去获取_loop
std::mutex _mutex; //互斥锁
std::condition_variable _cond; //条件变量
EventLoop* _loop; //EventLoop指针变量，这个对象需要在线程内部进行实例化
std::thread _thread; //EventLoop对应的线程
private:
//实例化EventLoop对象,唤醒_cond上可能阻塞的线程，并且开始运行EventLoop模块的功能
void ThreadEntry()
{
EventLoop loop;
{
//加锁
std::unique_lock lock(_mutex);
_loop = &loop;
_cond.notify_all();
}
loop.Start();
}
public:
//创建线程，设定线程入口函数
LoopThread()
:_loop(nullptr)
,_thread(std::thread(&LoopThread::ThreadEntry,this))
{}
//返回当前线程关联的EventLoop对象指针
EventLoop* GetLoop()
{
EventLoop* loop = nullptr;
{
std::unique_lock lock(_mutex);
_cond.wait(lock,[&](){ return _loop != nullptr; });
loop = _loop;
}
return loop;
}
};
class LoopThreadPool
{
private:
int _thread_count; //线程数量
int _next_idx; //下一个loop的下标
EventLoop* _baseloop; //主Loop，当_thread_count为0，就只使用它
std::vector _threads; //保存所有LoopThread对象
std::vector _loops; //保存所有EventLoop对象,baseloop除外
public:
LoopThreadPool(EventLoop* baseloop)
:_thread_count(0)
,_next_idx(0)
,_baseloop(baseloop)
{}
//设置线程数量
void SetThreadCount(int count)
{
_thread_count = count;
}
//创建所有的从属线程
void Create()
{
if(_thread_count > 0)
{
_threads.resize(_thread_count);
_loops.resize(_thread_count);
}
for(int i = 0;i < _thread_count;i++)
{
_threads[i] = new LoopThread();
_loops[i] = _threads[i]->GetLoop();
}
}
//下一个Loop
EventLoop* NextLoop()
{
if(_thread_count == 0)
{
return _baseloop;
}
_next_idx = (_next_idx + 1) % _thread_count;
return _loops[_next_idx];
}
};
