前言今天继续推进仿 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:usingFunctorstd::functionvoid();std::thread::id _thread_id;// 线程IDint_event_fd;// eventfd唤醒IO事件监控有可能导致的阻塞std::unique_ptrChannel_event_channel;Poller _poller;// 进行所有描述符的事件监控std::vectorFunctor_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 threadmuduo 中有一个很重要的思想one loop per thread也就是一个 EventLoop 对应一个固定线程。当前实现中构造 EventLoop 时记录线程 IDEventLoop():_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_idstd::this_thread::get_id());}voidAssertInLoop(){assert(_thread_idstd::this_thread::get_id());}IsInLoop()用来判断当前调用者是不是 EventLoop 对应的线程。AssertInLoop()则可以在某些只能由 EventLoop 线程执行的接口中进行断言。这种设计的意义是后面即使服务器使用多个线程也尽量让一个连接的事件和状态修改都在它所属的 EventLoop 线程中完成从而减少多个线程同时修改同一份数据的情况。四、EventLoop 的三步主循环EventLoop 最核心的接口是Start()// 三步走--事件监控-》就绪事件处理-》执行任务voidStart(){while(1){// 1. 事件监控std::vectorChannel*actives;_poller.Poll(actives);// 2. 事件处理for(autochannel:actives){channel-HandleEvent();}// 3. 执行任务RunAllTask();}}整个循环分为三步。1. 事件监控_poller.Poll(actives);Poller 内部调用epoll_wait等待已经注册的 fd 产生事件。如果暂时没有事件发生EventLoop 会阻塞在这里不会不停空转消耗 CPU。当事件到来后Poller 会把活跃的 Channel 放入actives。2. 就绪事件处理for(autochannel:actives){channel-HandleEvent();}EventLoop 不需要判断当前活跃的是 socket、eventfd 还是 timerfd。它只需要调用Channel::HandleEvent()再由 Channel 根据实际触发的事件执行对应回调。3. 执行任务池RunAllTask();所有活跃 Channel 处理结束以后EventLoop 会继续执行其他线程提交到任务池中的任务。因此这个主循环既能处理 IO 事件也能处理跨线程投递的普通任务。五、RunInLoop决定直接执行还是加入任务池如果一个任务本来就是在 EventLoop 对应线程中产生的可以直接执行。如果任务来自其他线程就先放入任务池等待 EventLoop 线程执行。对应代码如下// 判断将要执行的任务是否处于当前线程中如果是则执行不是则压入队列。voidRunInLoop(constFunctorcb){if(IsInLoop()){returncb();}returnQueueInLoop(cb);}执行流程可以表示为调用 RunInLoop | v 当前是否为 EventLoop 线程 | -- 是直接执行任务 | -- 否调用 QueueInLoop这样设计以后其他模块不需要自己反复判断线程。它们只需要把希望执行的操作交给 EventLoopEventLoop 会决定立即执行还是延后执行。六、QueueInLoop线程安全地提交任务任务池可能同时被 EventLoop 线程和其他工作线程访问。因此向任务池添加任务时需要加锁// 将操作压入任务池voidQueueInLoop(constFunctorcb){{std::unique_lockstd::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::vectorFunctorfunctor;{std::unique_lockstd::mutex_lock(_mutex);_tasks.swap(functor);}for(autof:functor){f();}return;}这里没有直接在_tasks中逐个执行任务而是先创建一个局部数组std::vectorFunctorfunctor;然后在锁内交换两个数组_tasks.swap(functor);交换完成后马上释放锁再执行局部数组中的所有任务。这样做有两个很重要的好处。1. 缩短持锁时间任务回调可能执行很久。如果一直拿着_mutex执行回调其他线程在这段时间里就无法继续提交任务。现在锁内只进行一次swap真正的任务执行发生在锁外。2. 避免回调再次提交任务时产生问题某个任务在执行过程中可能再次调用QueueInLoop()。因为当前已经释放_mutex新任务可以正常加入_tasks并留到下一轮 EventLoop 中执行。所以这里的核心思想是锁内交换任务 锁外执行任务八、为什么需要 eventfdEventLoop 的 Poller 会调用epoll_wait。当没有任何 fd 就绪时EventLoop 线程可能一直阻塞。这时如果其他线程提交了任务需要有一种方式主动唤醒 EventLoop。Linux 的eventfd正好可以解决这个问题。eventfd 内部维护了一个 64 位计数器同时它本身也是一个文件描述符。只要向 eventfd 写入数据它就会变成可读状态从而触发 epoll 的读事件。创建 eventfd 的代码如下staticintCreateEventFd(){intefdeventfd(0,EFD_CLOEXEC|EFD_NONBLOCK);if(efd0){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_tres0;intretread(_event_fd,res,sizeof(res));if(ret0){// EINTR -- 被信号打断EAGAIN -- 表示无数据可读if(errnoEINTR||errnoEAGAIN){return;}ERR_LOG(READ EVENTFD FAILED!);abort();}return;}其他线程唤醒 EventLoop 时向 eventfd 写入一个数值voidWeakUpEventFd(){uint64_tval1;intretwrite(_event_fd,val,sizeof(val));if(ret0){if(errnoEINTR){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。当前实现如下usingTaskFuncstd::functionvoid();usingReleaseFuncstd::functionvoid();classTimerTask{private:uint64_t_id;// 定时器任务对象IDuint32_t_timeout;// 定时任务的超时时间bool_canceled;// false-表示没有被取消true-表示被取消TaskFunc _task_cb;// 定时器对象要执行的定时任务ReleaseFunc _release;// 用于删除TimerWheel中保存的定时器对象信息public:TimerTask(uint64_tid,uint32_tdelay,constTaskFunccb):_id(id),_timeout(delay),_task_cb(cb),_canceled(false){}~TimerTask(){if(_canceledfalse)_task_cb();_release();}voidCancel(){_canceledtrue;}voidSetRelease(constReleaseFunccb){_releasecb;}uint32_tDelayTime(){return_timeout;}};这个类保存了五类信息。1. 任务 IDuint64_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_ptrChannel_timer_channel;其中_loop指向所属的 EventLoop_timerfd每秒产生一次可读事件_timer_channel把 timerfd 封装成 ChannelTimerWheel 构造函数如下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 驱动回调。十二、创建周期 timerfdTimerWheel 内部创建 timerfd 的代码如下staticintCreateTimerfd(){inttimerfdtimerfd_create(CLOCK_MONOTONIC,0);if(timerfd0){ERR_LOG(TIMERFD CREATE FAILED!);abort();}// int timerfd_settime(int fd, int flags,// struct itimerspec *new,// struct itimerspec *old);structitimerspecitime;itime.it_value.tv_sec1;itime.it_value.tv_nsec0;// 第一次超时时间为1s后itime.it_interval.tv_sec1;itime.it_interval.tv_nsec0;// 第一次超时后每次超时的间隔timerfd_settime(timerfd,0,itime,NULL);returntimerfd;}这里使用CLOCK_MONOTONIC创建定时器。它表示单调递增时间不受系统时间被手动修改的影响因此适合服务器中的相对超时。当前 timerfd 的设置是1 秒后第一次触发第一次触发后每隔 1 秒继续触发这样 timerfd 就像时间轮的秒针每秒通知 TimerWheel 前进一步。十三、读取累计超时次数timerfd 可读时读取到的是从上一次读取以后累计发生的超时次数。当前读取函数如下intReadTimefd(){uint64_ttimes;// 有可能因为其他描述符的事件处理花费时间比较长// 在处理定时器描述符事件的时候可能已经超时很多次// read读取到的数据times就是从上一次read之后超时的次数intretread(_timerfd,times,8);if(ret0){ERR_LOG(READ TIMEFD FAILED!);abort();}returntimes;}如果 EventLoop 中某个事件处理时间比较长timerfd 可能已经连续超时多次。因此OnTime()会根据读取到的次数多次推动时间轮voidOnTime(){// 根据实际超时的次数执行对应的超时任务inttimesReadTimefd();for(inti0;itimes;i){RunTimerTask();}}每推动一次时间轮就把秒针向后移动一格// 这个函数应该每秒钟被执行一次相当于秒针向后走了一步voidRunTimerTask(){_tick(_tick1)%_capacity;// 清空指定位置的数组就会把数组中保存的// 所有管理定时器对象的shared_ptr释放掉_wheel[_tick].clear();}如果槽位中的某个 TimerTask 没有其他shared_ptr继续持有它就会进入析构函数。没有取消的任务执行_task_cb随后通过_release清理_timers中的记录。十四、添加、刷新和取消定时任务1. 添加任务真正向时间轮添加任务的函数是TimerAddInLoop()voidTimerAddInLoop(uint64_tid,uint32_tdelay,constTaskFunccb){PtrTaskpt(newTimerTask(id,delay,cb));pt-SetRelease(std::bind(TimerWheel::RemoveTimer,this,id));intpos(_tickdelay)%_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 ptit-second.lock();intdelaypt-DelayTime();intpos(_tickdelay)%_capacity;_wheel[pos].push_back(pt);}刷新并不是把旧槽位中的指针立即删除而是把同一个 TimerTask 的shared_ptr再放到新的槽位中。旧槽位被清空时只会减少一个引用计数。因为新槽位仍然持有任务所以 TimerTask 不会立即析构。等最后一个引用在新的到期槽位被释放时任务才真正触发。3. 取消任务取消接口如下voidTimerCancelInLoop(uint64_tid){autoit_timers.find(id);if(it_timers.end()){return;// 没找着定时任务没法刷新没法延迟}PtrTask ptit-second.lock();if(pt)pt-Cancel();}取消任务并不会立即从时间轮中删除所有引用只是把_canceled设置为true。任务最终析构时由于已经取消就不会执行_task_cb但仍然会调用_release清理索引。十五、定时器操作为什么要交给 EventLoopTimerWheel 内部包含std::vectorstd::vectorPtrTask_wheel;std::unordered_mapuint64_t,WeakTask_timers;如果多个线程直接添加、刷新和取消定时任务就可能同时修改这两个容器。当前设计思路不是给时间轮的每一个操作都加锁而是把真正的修改放到 EventLoop 对应线程中执行。TimerWheel 当前提供了下面几个对外接口声明/*定时器中有个_timers成员定时器信息的操作有可能在多线程中进行因此需要考虑线程安全问题*//*如果不想加锁那就把对定期的所有操作都放到一个线程中进行*/voidTimerAdd(uint64_tid,uint32_tdelay,constTaskFunccb);// 刷新/延迟定时任务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 注册到同一个 Pollersocket 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 驱动 TimerWheelEventLoop 与 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 的退出、资源回收再开始实现新连接接收与连接管理模块。