前言
今天继续推进仿 muduo 的 C++ 高并发服务器项目。
前面已经陆续实现了几个基础模块:
Channel:封装文件描述符、关注事件和事件回调Poller:封装 epoll,负责事件的添加、修改、删除和等待TimerWheel:使用时间轮管理定时任务timerfd:把定时事件转换成可以被 epoll 监听的文件描述符
不过这些模块之前还比较分散。
Channel知道事件发生后应该执行什么回调,Poller知道哪些 fd 已经就绪,TimerWheel知道什么时候释放定时任务,但是还缺少一个对象把它们持续驱动起来。
今天主要完成了两个工作:
- 实现
EventLoop,建立 Reactor 模型中的事件循环 - 将之前的
TimerWheel通过timerfd + Channel接入 EventLoop
目前代码已经完成编译,并且实际测试运行正常。
本文记录的是我今天写下的当前版本。代码部分保持原实现,不在文章中替换函数名、调整错误处理或者补充尚未实现的逻辑;发现的问题统一放在后面的“当前版本思考”中。
一、EventLoop 是 Reactor 的驱动核心
在 Reactor 模型中,几个核心对象的职责可以这样划分:
Channel:一个 fd 关注什么事件,事件发生后调用什么回调 Poller:封装 epoll,等待事件并返回活跃的 Channel EventLoop:不断驱动事件监听、事件处理和任务执行也就是说,Poller负责“等”,Channel负责“处理”,而EventLoop负责让这套流程不断循环。
今天实现的 EventLoop 主流程可以概括为:
事件监控 | v 处理活跃 Channel | v 执行任务池中的任务 | v 进入下一轮事件监控有了 EventLoop 以后,前面实现的 Channel 和 Poller 才真正组成了一个可以持续运行的 Reactor 骨架。
二、EventLoop 的核心成员
EventLoop 中保存了下面这些成员:
classEventLoop{private:usingFunctor=std::function<void()>;std::thread::id _thread_id;// 线程IDint_event_fd;// eventfd唤醒IO事件监控有可能导致的阻塞std::unique_ptr<Channel>_event_channel;Poller _poller;// 进行所有描述符的事件监控std::vector<Functor>_tasks;// 任务池std::mutex _mutex;// 实现任务池操作的线程安全TimerWheel _timer_wheel;// 定时器模块};这些成员可以分成五部分。
1._thread_id
记录 EventLoop 创建时所在的线程。
后续可以通过线程 ID 判断某个操作当前是否就在 EventLoop 所属线程中执行。
2._event_fd
这是通过 Linuxeventfd创建的文件描述符,用来唤醒可能阻塞在epoll_wait中的 EventLoop。
3._event_channel
把_event_fd封装成 Channel。
这样 eventfd 也可以像 socket 一样注册到 Poller 中,由 epoll 统一监控。
4._tasks和_mutex
_tasks保存其他线程提交给 EventLoop 的任务。
因为任务可能从其他线程加入,所以需要使用_mutex保证任务池操作的线程安全。
5._poller和_timer_wheel
_poller负责所有 fd 的事件监控,_timer_wheel负责定时任务。
EventLoop 把二者组合起来以后,网络 IO 事件和定时事件就能在同一个循环中处理。
三、one loop per thread
muduo 中有一个很重要的思想:
one loop per thread也就是一个 EventLoop 对应一个固定线程。
当前实现中,构造 EventLoop 时记录线程 ID:
EventLoop():_thread_id(std::this_thread::get_id()),_event_fd(CreateEventFd()),_event_channel(newChannel(this,_event_fd)),_timer_wheel(this){// 给eventfd添加可读事件回调函数,读取eventfd事件通知次数_event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd,this));// 启动eventfd的读事件监控_event_channel->EnableRead();}线程判断接口如下:
boolIsInLoop(){return(_thread_id==std::this_thread::get_id());}voidAssertInLoop(){assert(_thread_id==std::this_thread::get_id());}IsInLoop()用来判断当前调用者是不是 EventLoop 对应的线程。
AssertInLoop()则可以在某些只能由 EventLoop 线程执行的接口中进行断言。
这种设计的意义是:后面即使服务器使用多个线程,也尽量让一个连接的事件和状态修改都在它所属的 EventLoop 线程中完成,从而减少多个线程同时修改同一份数据的情况。
四、EventLoop 的三步主循环
EventLoop 最核心的接口是Start():
// 三步走--事件监控-》就绪事件处理-》执行任务voidStart(){while(1){// 1. 事件监控std::vector<Channel*>actives;_poller.Poll(&actives);// 2. 事件处理for(auto&channel:actives){channel->HandleEvent();}// 3. 执行任务RunAllTask();}}整个循环分为三步。
1. 事件监控
_poller.Poll(&actives);Poller 内部调用epoll_wait,等待已经注册的 fd 产生事件。
如果暂时没有事件发生,EventLoop 会阻塞在这里,不会不停空转消耗 CPU。
当事件到来后,Poller 会把活跃的 Channel 放入actives。
2. 就绪事件处理
for(auto&channel:actives){channel->HandleEvent();}EventLoop 不需要判断当前活跃的是 socket、eventfd 还是 timerfd。
它只需要调用Channel::HandleEvent(),再由 Channel 根据实际触发的事件执行对应回调。
3. 执行任务池
RunAllTask();所有活跃 Channel 处理结束以后,EventLoop 会继续执行其他线程提交到任务池中的任务。
因此,这个主循环既能处理 IO 事件,也能处理跨线程投递的普通任务。
五、RunInLoop:决定直接执行还是加入任务池
如果一个任务本来就是在 EventLoop 对应线程中产生的,可以直接执行。
如果任务来自其他线程,就先放入任务池,等待 EventLoop 线程执行。
对应代码如下:
// 判断将要执行的任务是否处于当前线程中,如果是则执行,不是则压入队列。voidRunInLoop(constFunctor&cb){if(IsInLoop()){returncb();}returnQueueInLoop(cb);}执行流程可以表示为:
调用 RunInLoop | v 当前是否为 EventLoop 线程? | +-- 是:直接执行任务 | +-- 否:调用 QueueInLoop这样设计以后,其他模块不需要自己反复判断线程。
它们只需要把希望执行的操作交给 EventLoop,EventLoop 会决定立即执行还是延后执行。
六、QueueInLoop:线程安全地提交任务
任务池可能同时被 EventLoop 线程和其他工作线程访问。
因此,向任务池添加任务时需要加锁:
// 将操作压入任务池voidQueueInLoop(constFunctor&cb){{std::unique_lock<std::mutex>_lock(_mutex);_tasks.push_back(cb);}// 唤醒有可能因为没有事件就绪,而导致的epoll阻塞;// 其实就是给eventfd写入一个数据,eventfd就会触发可读事件WeakUpEventFd();}这里的额外大括号让互斥锁在任务入队后立即释放。
也就是说,加锁范围只包含:
_tasks.push_back(cb);后面的 eventfd 写入不需要继续持有任务队列的锁。
任务加入队列以后,代码会调用WeakUpEventFd()。
这是因为 EventLoop 可能正阻塞在epoll_wait中。如果只是把任务加入_tasks,却没有任何 fd 产生事件,EventLoop 就无法及时发现新任务。
七、RunAllTask:交换任务池后在锁外执行
执行任务池的代码如下:
// 执行任务池中的所有任务voidRunAllTask(){std::vector<Functor>functor;{std::unique_lock<std::mutex>_lock(_mutex);_tasks.swap(functor);}for(auto&f:functor){f();}return;}这里没有直接在_tasks中逐个执行任务,而是先创建一个局部数组:
std::vector<Functor>functor;然后在锁内交换两个数组:
_tasks.swap(functor);交换完成后马上释放锁,再执行局部数组中的所有任务。
这样做有两个很重要的好处。
1. 缩短持锁时间
任务回调可能执行很久。
如果一直拿着_mutex执行回调,其他线程在这段时间里就无法继续提交任务。
现在锁内只进行一次swap,真正的任务执行发生在锁外。
2. 避免回调再次提交任务时产生问题
某个任务在执行过程中,可能再次调用QueueInLoop()。
因为当前已经释放_mutex,新任务可以正常加入_tasks,并留到下一轮 EventLoop 中执行。
所以这里的核心思想是:
锁内交换任务 锁外执行任务八、为什么需要 eventfd
EventLoop 的 Poller 会调用epoll_wait。
当没有任何 fd 就绪时,EventLoop 线程可能一直阻塞。
这时如果其他线程提交了任务,需要有一种方式主动唤醒 EventLoop。
Linux 的eventfd正好可以解决这个问题。
eventfd 内部维护了一个 64 位计数器,同时它本身也是一个文件描述符。
只要向 eventfd 写入数据,它就会变成可读状态,从而触发 epoll 的读事件。
创建 eventfd 的代码如下:
staticintCreateEventFd(){intefd=eventfd(0,EFD_CLOEXEC|EFD_NONBLOCK);if(efd<0){ERR_LOG("CREATE EVENTFD FAILED!!");abort();// 让程序异常退出}returnefd;}这里使用了两个标志:
EFD_CLOEXEC:执行新程序时自动关闭该描述符EFD_NONBLOCK:把 eventfd 设置成非阻塞模式
EventLoop 构造时,会把 eventfd 封装成 Channel,并开启读事件监控。
因此 eventfd 的注册流程仍然是前面已经实现的链路:
_event_channel->EnableRead() | v Channel::Update() | v EventLoop::UpdateEvent() | v Poller::UpdateEvent() | v epoll_ctl()九、eventfd 的读取和唤醒
eventfd 的读取函数如下:
voidReadEventfd(){uint64_tres=0;intret=read(_event_fd,&res,sizeof(res));if(ret<0){// EINTR -- 被信号打断;EAGAIN -- 表示无数据可读if(errno==EINTR||errno==EAGAIN){return;}ERR_LOG("READ EVENTFD FAILED!");abort();}return;}其他线程唤醒 EventLoop 时,向 eventfd 写入一个数值:
voidWeakUpEventFd(){uint64_tval=1;intret=write(_event_fd,&val,sizeof(val));if(ret<0){if(errno==EINTR){return;}ERR_LOG("READ EVENTFD FAILED!");abort();}return;}在当前实现中,eventfd 只负责通知 EventLoop“任务池中已经有新任务”。
具体任务仍然保存在_tasks中。
因此 eventfd 和任务池的职责是分开的:
_tasks:保存需要执行的具体任务 eventfd:唤醒阻塞中的 EventLoop完整唤醒流程如下:
其他线程调用 QueueInLoop | v 任务加入 _tasks | v 向 eventfd 写入数据 | v eventfd 变为可读 | v epoll_wait 返回 | v Channel 调用 ReadEventfd | v EventLoop 调用 RunAllTask | v 任务在 EventLoop 线程中执行十、TimerTask:一个具体的定时任务
时间轮中保存的具体任务对象是TimerTask。
当前实现如下:
usingTaskFunc=std::function<void()>;usingReleaseFunc=std::function<void()>;classTimerTask{private:uint64_t_id;// 定时器任务对象IDuint32_t_timeout;// 定时任务的超时时间bool_canceled;// false-表示没有被取消,true-表示被取消TaskFunc _task_cb;// 定时器对象要执行的定时任务ReleaseFunc _release;// 用于删除TimerWheel中保存的定时器对象信息public:TimerTask(uint64_tid,uint32_tdelay,constTaskFunc&cb):_id(id),_timeout(delay),_task_cb(cb),_canceled(false){}~TimerTask(){if(_canceled==false)_task_cb();_release();}voidCancel(){_canceled=true;}voidSetRelease(constReleaseFunc&cb){_release=cb;}uint32_tDelayTime(){return_timeout;}};这个类保存了五类信息。
1. 任务 ID
uint64_t_id;用于在 TimerWheel 的哈希表中查找任务。
2. 超时时间
uint32_t_timeout;保存任务需要延迟多少秒执行。
3. 取消状态
bool_canceled;调用Cancel()后,任务对象依然可以留在时间轮槽位中,但析构时不会再执行_task_cb。
4. 定时任务回调
TaskFunc _task_cb;任务正常超时并释放时执行。
5. 释放回调
ReleaseFunc _release;任务对象析构以后,通过_release删除 TimerWheel 中保存的任务索引。
当前 TimerTask 把任务触发和对象析构关联起来:
时间轮槽位被清空 | v shared_ptr 被释放 | v TimerTask 最后一个引用消失 | v 进入 TimerTask 析构函数 | +-- 没有取消:执行 _task_cb | +-- 执行 _release 清理索引这种设计让 TimerWheel 不需要额外遍历到期任务并逐个调用回调,而是通过智能指针的生命周期完成触发。
十一、TimerWheel 与 EventLoop 的整合
TimerWheel 新增了三个与事件循环有关的成员:
EventLoop*_loop;int_timerfd;// 定时器描述符--可读事件回调就是读取计数器,执行定时任务std::unique_ptr<Channel>_timer_channel;其中:
_loop指向所属的 EventLoop_timerfd每秒产生一次可读事件_timer_channel把 timerfd 封装成 Channel
TimerWheel 构造函数如下:
TimerWheel(EventLoop*loop):_capacity(60),_tick(0),_wheel(_capacity),_loop(loop),_timerfd(CreateTimerfd()),_timer_channel(newChannel(_loop,_timerfd)){_timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime,this));_timer_channel->EnableRead();// 启动读事件监控}和 eventfd 一样,timerfd 也被封装成 Channel。
eventfd | v Channel -> ReadEventfd timerfd | v Channel -> TimerWheel::OnTime对于 Poller 来说,socket、eventfd 和 timerfd 都只是普通文件描述符。
它们统一通过 Channel 注册事件,再由 EventLoop 驱动回调。
十二、创建周期 timerfd
TimerWheel 内部创建 timerfd 的代码如下:
staticintCreateTimerfd(){inttimerfd=timerfd_create(CLOCK_MONOTONIC,0);if(timerfd<0){ERR_LOG("TIMERFD CREATE FAILED!");abort();}// int timerfd_settime(int fd, int flags,// struct itimerspec *new,// struct itimerspec *old);structitimerspecitime;itime.it_value.tv_sec=1;itime.it_value.tv_nsec=0;// 第一次超时时间为1s后itime.it_interval.tv_sec=1;itime.it_interval.tv_nsec=0;// 第一次超时后,每次超时的间隔timerfd_settime(timerfd,0,&itime,NULL);returntimerfd;}这里使用CLOCK_MONOTONIC创建定时器。
它表示单调递增时间,不受系统时间被手动修改的影响,因此适合服务器中的相对超时。
当前 timerfd 的设置是:
- 1 秒后第一次触发
- 第一次触发后,每隔 1 秒继续触发
这样 timerfd 就像时间轮的秒针,每秒通知 TimerWheel 前进一步。
十三、读取累计超时次数
timerfd 可读时,读取到的是从上一次读取以后累计发生的超时次数。
当前读取函数如下:
intReadTimefd(){uint64_ttimes;// 有可能因为其他描述符的事件处理花费时间比较长,// 在处理定时器描述符事件的时候,可能已经超时很多次// read读取到的数据times就是从上一次read之后超时的次数intret=read(_timerfd,×,8);if(ret<0){ERR_LOG("READ TIMEFD FAILED!");abort();}returntimes;}如果 EventLoop 中某个事件处理时间比较长,timerfd 可能已经连续超时多次。
因此OnTime()会根据读取到的次数多次推动时间轮:
voidOnTime(){// 根据实际超时的次数,执行对应的超时任务inttimes=ReadTimefd();for(inti=0;i<times;i++){RunTimerTask();}}每推动一次时间轮,就把秒针向后移动一格:
// 这个函数应该每秒钟被执行一次,相当于秒针向后走了一步voidRunTimerTask(){_tick=(_tick+1)%_capacity;// 清空指定位置的数组,就会把数组中保存的// 所有管理定时器对象的shared_ptr释放掉_wheel[_tick].clear();}如果槽位中的某个 TimerTask 没有其他shared_ptr继续持有,它就会进入析构函数。
没有取消的任务执行_task_cb,随后通过_release清理_timers中的记录。
十四、添加、刷新和取消定时任务
1. 添加任务
真正向时间轮添加任务的函数是TimerAddInLoop():
voidTimerAddInLoop(uint64_tid,uint32_tdelay,constTaskFunc&cb){PtrTaskpt(newTimerTask(id,delay,cb));pt->SetRelease(std::bind(&TimerWheel::RemoveTimer,this,id));intpos=(_tick+delay)%_capacity;_wheel[pos].push_back(pt);_timers[id]=WeakTask(pt);}添加任务时会完成三件事:
- 创建 TimerTask
- 根据当前秒针和延迟时间计算槽位
- 使用
weak_ptr在_timers中保存任务索引
_wheel中保存shared_ptr,负责管理任务生命周期;_timers中保存weak_ptr,负责通过 ID 查找任务,但不会额外延长任务生命周期。
2. 刷新任务
刷新接口如下:
voidTimerRefreshInLoop(uint64_tid){// 通过保存的定时器对象的weak_ptr构造一个shared_ptr出来,添加到轮子中autoit=_timers.find(id);if(it==_timers.end()){return;// 没找着定时任务,没法刷新,没法延迟}PtrTask pt=it->second.lock();intdelay=pt->DelayTime();intpos=(_tick+delay)%_capacity;_wheel[pos].push_back(pt);}刷新并不是把旧槽位中的指针立即删除,而是把同一个 TimerTask 的shared_ptr再放到新的槽位中。
旧槽位被清空时,只会减少一个引用计数。
因为新槽位仍然持有任务,所以 TimerTask 不会立即析构。等最后一个引用在新的到期槽位被释放时,任务才真正触发。
3. 取消任务
取消接口如下:
voidTimerCancelInLoop(uint64_tid){autoit=_timers.find(id);if(it==_timers.end()){return;// 没找着定时任务,没法刷新,没法延迟}PtrTask pt=it->second.lock();if(pt)pt->Cancel();}取消任务并不会立即从时间轮中删除所有引用,只是把_canceled设置为true。
任务最终析构时,由于已经取消,就不会执行_task_cb,但仍然会调用_release清理索引。
十五、定时器操作为什么要交给 EventLoop
TimerWheel 内部包含:
std::vector<std::vector<PtrTask>>_wheel;std::unordered_map<uint64_t,WeakTask>_timers;如果多个线程直接添加、刷新和取消定时任务,就可能同时修改这两个容器。
当前设计思路不是给时间轮的每一个操作都加锁,而是把真正的修改放到 EventLoop 对应线程中执行。
TimerWheel 当前提供了下面几个对外接口声明:
/*定时器中有个_timers成员,定时器信息的操作有可能在多线程中进行,因此需要考虑线程安全问题*//*如果不想加锁,那就把对定期的所有操作,都放到一个线程中进行*/voidTimerAdd(uint64_tid,uint32_tdelay,constTaskFunc&cb);// 刷新/延迟定时任务voidTimerRefresh(uint64_tid);voidTimerCancel(uint64_tid);今天的版本中,这三个公开接口还只有声明,具体的类外定义还没有完成。
后面的目标是通过EventLoop::RunInLoop(),把操作投递到 EventLoop 线程,然后再调用对应的TimerAddInLoop、TimerRefreshInLoop和TimerCancelInLoop。
目标链路如下:
其他线程操作定时器 | v TimerAdd / TimerRefresh / TimerCancel | v EventLoop::RunInLoop | v TimerAddInLoop / TimerRefreshInLoop / TimerCancelInLoop | v 只在 EventLoop 线程修改时间轮当前代码已经准备好了 EventLoop 的任务调度能力和三个*InLoop内部操作,但公开接口的线程切换封装仍然是下一步工作。
十六、EventLoop 对事件操作的封装
EventLoop 还提供了 Channel 事件操作的转发接口:
// 添加/修改描述符的事件监控voidUpdateEvent(Channel*channel){return_poller.UpdateEvent(channel);}// 移除描述符的监控voidRemoveEvent(Channel*channel){return_poller.RemoveEvent(channel);}Channel 修改关注事件时,通过 EventLoop 把请求转发给 Poller。
完整注册链路如下:
Channel::EnableRead / EnableWrite | v Channel::Update | v EventLoop::UpdateEvent | v Poller::UpdateEvent | v epoll_ctl这样 Channel 不需要知道 epoll 的实现细节,Poller 也不需要处理上层业务。
十七、整合后的完整 Reactor 流程
现在项目中的事件来源主要有三类:
socket fd:处理网络读写 eventfd:处理跨线程唤醒 timerfd:推动时间轮它们都通过 Channel 注册到同一个 Poller:
socket fd -----> Channel ---+ eventfd -----> Channel ---+ timerfd -----> Channel ---+ | v Poller / epoll | v EventLoop完整运行流程如下:
EventLoop::Start | v Poller::Poll | v epoll_wait 等待事件 | v 返回活跃 Channel | v EventLoop 遍历活跃 Channel | v Channel::HandleEvent | +-- socket 可读/可写:执行网络回调 | +-- eventfd 可读:读取唤醒计数 | +-- timerfd 可读:推动 TimerWheel | v EventLoop::RunAllTask | v 进入下一轮 epoll_wait到这里,Channel、Poller、EventLoop、eventfd、timerfd 和 TimerWheel 已经组成了一条完整的 Reactor 主链路。
十八、当前编译和运行情况
今天已经完成:
- EventLoop 三步主循环
- EventLoop 所属线程判断
RunInLoop和QueueInLoop- 任务池的加锁、交换和执行
- eventfd 的创建、读取和写入
- eventfd Channel 与 Poller 的整合
- TimerTask 的任务、取消和释放逻辑
- timerfd Channel 与 Poller 的整合
- timerfd 驱动 TimerWheel
- EventLoop 与 TimerWheel 的整体编译
十九、总结
今天完成了仿 muduo 项目中的核心事件循环EventLoop,并把之前实现的 TimerWheel 接入了 Reactor 主流程。
EventLoop 的核心循环可以总结为:
Poll -> HandleEvent -> RunAllTask跨线程任务通过 eventfd 唤醒 EventLoop:
任务入队 -> 写 eventfd -> 唤醒 epoll_wait -> EventLoop 执行任务定时任务则通过 timerfd 推动时间轮:
timerfd 可读 -> OnTime -> 时间轮前进 -> 释放到期 TimerTaskTimerTask 使用任务回调、取消标志和释放回调,把定时任务的执行与智能指针生命周期关联起来。
目前 EventLoop、Channel、Poller、eventfd、timerfd 和 TimerWheel 已经完成整合,代码编译通过,实际测试运行正常。
到这里,项目已经具备了 Reactor 事件循环的基本骨架。下一步继续补齐定时器的跨线程接口和 EventLoop 的退出、资源回收,再开始实现新连接接收与连接管理模块。