1. IO协程调度封装

构思

使用协程开发（一）中已经完成的协程调度模块，封装出具体的调度子类IOManager。不仅有基本的协程调度分配任务、执行任务的功能，而且由于是服务器框架，一定绕不开网络IO的管理，用于管理各种IO接口以及上面发生的读写事件，根据读写事件触发相应的异步回调函数（自动装入任务队列执行）

· 类关系

协程开发 （二） - 图1

基本IO管理思路

协程开发 （二） - 图2

1.1 成员变量

  1. /**
  2.  * @brief IO调度器类
  3.  */
  4. class IOManager: public Scheduler, public TimerManager
  5. {
  6.     ...
  7.     ...
  9. public:
  10.     /**
  11.      * @brief 事件枚举类型 直接对标epoll里的事件赋值
  12.      */
  13.     enum Event{
  14.         NONE =  0x0,    //无事件
  15.         READ =  0x1,    //读事件   EPOLLIN
  16.         WRITE = 0x4     //写事件   EPOLLOUT
  17.     };
  18. private:
  19.     ///epoll句柄
  20.     int m_epfd; 
  21.     ///管道描述符 作为唤醒工具
  22.     int m_tickleFds[2];
  23.     ///待处理事件数量
  24.     std::atomic<size_t> m_pendingEventCount = {0};
  25.     ///读写锁
  26.     MutexType m_mutex;
  27.     ///句柄对象指针数组
  28.     std::vector<FdContext*> m_fdContexts;
  29. };

* 封装套接字句柄对象+事件对象

目的：方便归纳句柄、事件所具有的属性。句柄带有事件，事件依附于句柄。只有句柄上才有事件触发

注意：和HOOK模块的文件句柄类做一个区分，这里的套接字句柄专门针对套接字上的事件做回调管理。

  1. class IOManager: public Scheduler
  2. {
  3.     ...
  4.     ...
  6. private:
  7.     /**
  8.      * @brief 句柄对象结构体
  9.      */
  10.     struct FdContext
  11.     {
  12.         typedef Mutex MutexType;
  14.         /**
  15.          * @brief 事件对象结构体
  16.          */
  17.         struct EventContext
  18.         {
  19.             ///被调度的调度器
  20.             Scheduler *scheduler = nullptr;
  21.             //事件绑定的协程
  22.             Coroutine::ptr coroutine;
  23.             //事件绑定的函数
  24.             std::function<void()> cb;
  26.         };
  28.         ///句柄/文件描述符
  29.         int fd = 0; 
  30.         ///句柄上的读事件对象
  31.         EventContext read_event;
  32.         ///句柄上的写事件对象
  33.         EventContext write_event;
  34.         ///句柄上注册好的事件
  35.         Event events = NONE;
  36.         ///互斥锁
  37.         MutexType mutex;
  39.         /**
  40.          * @brief 获取句柄对象上的事件对象
  41.          * @param[in] event 读/写事件
  42.          * @return EventContext& 
  43.          */
  44.         EventContext& getEventContext(Event event);
  46.         /**
  47.          * @brief 清空句柄对象上的事件对象
  48.          * @param[in] event_ctx 读/写事件对象
  49.          */
  50.         void resetEventContext(struct EventContext& event_ctx);
  52.         /**
  53.          * @brief 主动触发事件，执行事件对象上的回调函数
  54.          * @param[in] event 读/写事件 
  55.          */
  56.         void triggerEvent(Event event);
  58.     };
  59.     ...
  60.     ...
  61. }

接口

1). getEventContext()

功能：获取句柄对象上的事件对象

  1. /**
  2. * @brief 获取句柄对象上的事件对象
  3. * @param[in] event 读/写事件
  4. * @return EventContext& 
  5. */
  6. EventContext& getEventContext(Event event);
  8. IOManager::FdContext::EventContext& IOManager::FdContext::getEventContext(IOManager::Event event)
  9. {
  10.     switch(event)
  11.     {
  12.         case IOManager::READ: return read_event;break;
  13.         case IOManager::WRITE: return write_event;break;
  14.         default:
  15.             KIT_ASSERT2(false, "getContext error");
  16.     }
  18.     throw std::invalid_argument("getContext invalid event!!");
  19. }

2). resetEventContext()

功能：清空句柄对象上的事件对象

  1. /**
  2.  * @brief 清空句柄对象上的事件对象
  3.  * @param[in] event_ctx 读/写事件对象
  4.  */
  5. void resetEventContext(struct EventContext& event_ctx);

3). triggerEvent()

功能：

  1. /**
  2.  * @brief 主动触发事件，执行事件对象上的回调函数
  3.  * @param[in] event 读/写事件 
  4.  */
  5. void triggerEvent(Event event);
  7. void IOManager::FdContext::triggerEvent(IOManager::Event event)
  8. {
  9.     KIT_ASSERT(events & event);
  11.     //把触发后的事件去除掉
  12.     events = (Event)(events & ~event);
  14.     EventContext& ctx = getEventContext(event);
  16.     KIT_ASSERT(ctx.cb || ctx.coroutine);
  18.     if(ctx.cb)
  19.     {
  20.         ctx.scheduler->schedule(&ctx.cb);
  21.     }
  22.     else if(ctx.coroutine)
  23.     {
  24.         ctx.scheduler->schedule(&ctx.coroutine);
  25.     }
  28.     ctx.scheduler = nullptr;
  29. }

1.2 接口

1.2.1 构造函数

  1. /**
  2.  * @brief IO调度器类构造函数
  3.  * @param[in] name 调度器名称
  4.  * @param[in] threads_size 初始线程数量
  5.  * @param[in] use_caller 当前线程是否作为调度线程
  6.  */
  7. IOManager(const std::string& name = "", size_t threads_size = 1, bool use_caller = true);
  9. IOManager::IOManager(const std::string& name, size_t threads_size, bool use_caller)
  10.     :Scheduler(name, threads_size, use_caller)
  11. {
  12.     //创建epoll句柄
  13.     m_epfd = epoll_create(1);
  14.     if(m_epfd < 0)
  15.     {
  16.         KIT_LOG_ERROR(g_logger) << "IOManager: epoll_create error";
  17.         KIT_ASSERT2(false, "epoll_create error");
  18.     }
  20.     //创建管道句柄
  21.     int ret = pipe(m_tickleFds);
  22.     if(ret < 0)
  23.     {
  24.         KIT_LOG_ERROR(g_logger) << "IOManager: pipe create error";
  25.         KIT_ASSERT2(false, "pipe create error");
  26.     }
  28.     //初始化epoll事件
  29.     struct epoll_event event;
  30.     memset(&event, 0, sizeof(struct epoll_event));
  32.     //设置为 读事件触发 以及 边缘触发
  33.     event.events = EPOLLIN | EPOLLET;
  34.     // [0]为读管道 [1]为写管道
  35.     event.data.fd = m_tickleFds[0];    
  37.     //设置读管道句柄属性 将读fd 设置为非阻塞
  38.     ret = fcntl(m_tickleFds[0], F_SETFL, O_NONBLOCK);
  39.     if(ret < 0)
  40.     {
  41.         KIT_LOG_ERROR(g_logger) << "IOManager: fcntl error";
  42.         KIT_ASSERT2(false, "fcntl error");
  43.     }
  45.     //将当前的事件添加到epoll中
  46.     ret = epoll_ctl(m_epfd, EPOLL_CTL_ADD, m_tickleFds[0], &event);
  47.     if(ret < 0)
  48.     {
  49.         KIT_LOG_ERROR(g_logger) << "IOManager: epoll_ctl";
  50.         KIT_ASSERT2(false, "epoll_ctl error");
  51.     }
  53.     //默认为个事件信息
  54.     contextResize(256);
  56.     //启动IO调度器
  57.     start();
  58. }

1.2.2 析构函数

  1. /**
  2.  * @brief IO调度器类析构函数
  3.  */
  4. ~IOManager();
  6. IOManager::~IOManager()
  7. {
  8.     //停止调度器
  9.     stop();
  11.     close(m_epfd);          //关闭epoll句柄
  12.     close(m_tickleFds[0]);  //关闭读管道句柄
  13.     close(m_tickleFds[1]);  //关闭写管道句柄
  15.     //删除事件对象分配的空间
  16.     //不使用智能指针的原因：要把空间释放集中到持有调度器的这个线程中
  17.     for(size_t i = 0;i < m_fdContexts.size();++i)
  18.     {
  19.         if(m_fdContexts[i])
  20.             delete m_fdContexts[i];
  21.     }
  22. }

1.2.3 IOManager大概几个核心的接口

下面几个接口已经实现在基类Scheduler中，派生类中仍然使用：

  • start() 开启调度器运作
  • stop() 停止调度器运作
  • run() 这个函数真正执行调度逻辑
  • schduleNoLock() 将任务加进队列

下面几个接口在派生类中IOManager为核心实现：

  • tickle() 线程唤醒
  • stopping() 判断调度器是否可以停止
  • idle() 协程没有任务可做时的处理，借助epoll_wait来唤醒有任务可执行

下文展开细说：

1). tickle()

功能：通过pipe管道特性，将管道读端加入到epoll的管理中。通过往管道写端写入数据（写入什么不重要）就能触发读端。假设现在只有那个管道读IO活跃，退出epoll_wait()循环监测后，没有其他网络IO需要处理，就会swapOut()切回到run()函数中，去查看任务队列的情况，完成一次线程的”唤醒”。

几处用到**tickle()**地方：

  • schedule()添加任务，判断队列在本次添加前是否空：空就tickle()；不空不唤醒。
  • run()在轮询任务队列的时候，发现有任务不属于自己执行，并且也没有指定任何线程执行的时候要调用tickle()去唤醒线程执行这些任务
  • stop()准备关闭调度器时候，要把所有创建的线程唤醒，去让他们退出：

协程开发 （二） - 图3

2). stopping()

功能：和基类几乎一样。多判断一下待处理事件数量m_pendingEventCount和定时器队列是否空isTimersEmpty()

  1. /**
  2.  * @brief 调度器停止的判断条件
  3.  * @return true 停止成功
  4.  * @return false 停止失败
  5.  */
  6. bool stopping() override;
  9. bool IOManager::stopping()
  10. {
  11.     return m_pendingEventCount == 0  && 
  12.         isTimersEmpty() && Scheduler::stopping();
  13. }

3). idle()（核心）

功能：
  1. 当前协程分配到的任务完成之后，epoll里检查一下是否有别的IO活动，调度对应的回调函数/协程
  2. 当前协程没有任务可做，IO也没有活动的迹象，陷入到epoll_wait的循环里面，阻塞等待。

核心逻辑：

while(1)
{

  1. 检查调度器是否已经停止。已经停止就不能往下执行逻辑，需要退出函数。
  2. 陷入~~epoll_wait~~，等待IO事件返回，默认超时时间为~~5000ms~~
  3. 如果有就绪IO，就带回对应IO：
    1. 网络IO，依次轮询触发注册好的异步读/写回调函数作为任务加入队列供调度；
    2. 管道读IO，由~~tickle()~~函数发送消息导致的IO活动，用于唤醒，将其忽略（不绑定回调函数作为任务去调度仅仅是唤醒作用）。
  4. 如果在超时时间内没有就绪IO，停止监测，让出当前协程执行权~~swapOut()~~。

}

  • 加入定时器后，结合**epoll_wait**有改动

while(1)
{

  1. 检查调度器是否已经停止。已经停止就不能往下执行逻辑，需要退出函数。
  2. 陷入epoll_wait，等待IO事件返回，超时时间需要考虑最近快到期的定时任务和默认超时时间取min选其中的小值。
  3. 退出epoll_wait之后首先检查，是否是因为有到时定时器需要处理。
    1. 有超时定时器，就把所有的超时任务加入调度
    2. 没有就处理就绪IO
  4. 如果有就绪IO，就带回对应IO：
    1. 网络IO，依次轮询触发注册好的异步读/写回调函数作为任务加入队列供调度；
    2. 管道读IO，由tickle()函数发送消息导致的IO活动，用于唤醒，将其忽略（不绑定回调函数作为任务去调度仅仅是唤醒作用）。
  5. 如果在超时时间内没有就绪IO/到时的定时器，停止监测，让出当前协程执行权swapOut()，回到Scheduler::run()函数中去。

}

原版
  1. //协程没有任务可做 要执行epoll监听
  2. void IOManager::idle()
  3. {
  4.     KIT_LOG_DEBUG(g_logger) << "idle start";
  6.     //4个一组 取出已经就绪的IO
  7.     struct epoll_event *events = new epoll_event[4];
  9.     //小技巧：借助智能指针的指定析构函数  自动释放数组
  10.     std::shared_ptr<struct epoll_event> evs_sp(events, [=](struct epoll_event* p){
  11.         delete[] events;
  12.     });
  14.     while(1)
  15.     {
  16.         /*1.如果调度器关闭了 就退出该函数*/
  17.         if(stopping())
  18.         {
  19.             KIT_LOG_INFO(g_logger) << "iomanager name=" << getName() << " is stopping, idle func exit";
  21.             return;
  22.         }
  24.         /*2.通过epoll_wait 带回已经就绪的IO*/
  25.         int n_ready;
  26.         do{
  27.             //最大超时时间 5000ms
  28.             static const int MAXTIMEOUT = 5000;
  30.             n_ready = epoll_wait(m_epfd, events, 32, MAXTIMEOUT);
  31.             KIT_LOG_DEBUG(g_logger) << "epoll_wait n_ready=" <<  n_ready;
  32.             if(n_ready < 0 && errno == EINTR)
  33.                 continue;       //重新尝试等待wait
  34.             else            //拿到了返回epoll_event 或者 已经超时就break
  35.                 break;
  38.         }while(1);
  40.         /*3.依次处理已经就绪的IO*/
  41.         for(int i = 0;i < n_ready;++i)
  42.         {
  43.             struct epoll_event& event = events[i];
  45.             //过滤外部发消息唤醒的IO  跳过
  46.             if(event.data.fd == m_tickleFds[0])
  47.             {
  48.                 KIT_LOG_DEBUG(g_logger) << "读管道活跃";
  49.                 uint8_t temp;
  50.                 //循环的目的 把缓冲区全部读干净
  51.                 while(read(m_tickleFds[0], &temp, 1) == 1);
  53.                 continue;
  54.             }
  56.             //处理剩下的真正就绪的IO
  57.             FdContext* fd_ctx = (FdContext*)event.data.ptr;
  59.             FdContext::MutexType::Lock lock(fd_ctx->mutex);
  61.             //如果是错误或者中断 导致的活动  重置一下
  62.             if(event.events & (EPOLLERR | EPOLLHUP))
  63.                 event.events |= EPOLLIN | EPOLLOUT;
  65.             //开一个变量转换 从epoll_event的事件----->自定义的事件Event
  66.             int real_events = NONE;
  67.             if(event.events & EPOLLIN)
  68.             {
  69.                 real_events |= READ;
  70.             }
  72.             if(event.events & EPOLLOUT)
  73.             {
  74.                 real_events |= WRITE;
  75.             }
  77.             //和当前IO上的事件对比 为0x0没有事件触发 跳过
  78.             if((fd_ctx->events & real_events) == Event::NONE)
  79.                 continue;
  81.             //把下面准备主动触发的事件 去除掉 剩余的事件放回epoll中
  82.             //因为不知道是单有读 单有写  读写都有 需要涵盖这三种情况
  83.             int left_events = (fd_ctx->events & ~real_events);
  85.             int op = left_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
  87.             //复用event
  88.             event.events = EPOLLET | left_events;
  90.             int ret = epoll_ctl(m_epfd, op, fd_ctx->fd, &event);
  91.             if(ret < 0)
  92.             {
  93.                 KIT_LOG_ERROR(g_logger) << "\nidel: epoll_ctl(" << m_epfd << ", " << op << ", " << fd_ctx->fd << ", " << fd_ctx->events << ");\n"
  94.                 << " error:" << ret << "(" << errno << "," << strerror(errno) << ")"; 
  96.                 continue;
  97.             }
  99.             //把剩余没有触发的读事件 主动触发
  100.             if(real_events & READ)
  101.             {
  102.                 KIT_LOG_DEBUG(g_logger) << "idle 读事件触发";
  103.                 fd_ctx->triggerEvent(READ);
  104.                 --m_pendingEventCount;
  105.             }
  107.             //把剩余没有触发的写事件 主动触发
  108.             if(real_events & WRITE)
  109.             {
  110.                 KIT_LOG_DEBUG(g_logger) << "idle 写事件触发";
  111.                 fd_ctx->triggerEvent(WRITE);
  112.                 --m_pendingEventCount;
  113.             }
  115.         }
  118.         /*4.处理完就绪的IO  让出当前协程的执行权 到Scheduler::run中去*/
  119.         Coroutine::ptr cur = Coroutine::GetThis();
  120.         auto p = cur.get();
  121.         cur.reset();
  123.         p->swapOut();
  125.     }
  127. }

改动（加入Timer）
  1. //协程没有任务可做 要执行epoll监听
  2. void IOManager::idle()
  3. {
  4.     KIT_LOG_DEBUG(g_logger) << "idle start";
  6.     //4个一组 取出已经就绪的IO
  7.     struct epoll_event *events = new epoll_event[4];
  9.     //小技巧：借助智能指针的指定析构函数  自动释放数组
  10.     std::shared_ptr<struct epoll_event> evs_sp(events, [=](struct epoll_event* p){
  11.         delete[] events;
  12.     });
  14.     while(1)
  15.     {
  16.         KIT_LOG_DEBUG(g_logger) << "while(1) again";
  18.         /*1.如果调度器关闭了 就退出该函数*/
  19.         if(stopping())
  20.         {
  22.             //如果定时器队列为空就可以退出了
  23.             if(isTimersEmpty())
  24.             {
  25.                 KIT_LOG_INFO(g_logger) << "iomanager name= " << getName() << " is stopping, idle func exit";
  27.                 return;
  28.             }
  29.         }
  31.         //获取下一次定时器的执行时间
  32.         uint64_t next_timeout = getNextTime();
  35.         /*2.通过epoll_wait 带回已经就绪的IO*/
  36.         int n_ready;
  37.         do{
  38.             //最大超时时间 5000ms
  39.             static const int MAXTIMEOUT = 1000;
  41.             //下一次定时器执行时间
  42.             if(next_timeout != ~0ull)
  43.             {
  44.                 next_timeout = next_timeout > MAXTIMEOUT ? MAXTIMEOUT : next_timeout;
  45.             }
  46.             else
  47.             {
  48.                 next_timeout = MAXTIMEOUT;
  49.             }
  51.             n_ready = epoll_wait(m_epfd, events, 32, (int)next_timeout);
  52.             KIT_LOG_DEBUG(g_logger) << "epoll_wait n_ready=" <<  n_ready;
  53.             if(n_ready < 0 && errno == EINTR)
  54.                 continue;       //重新尝试等待wait
  55.             else            //拿到了返回epoll_event 或者 已经超时就break
  56.                 break;
  59.         }while(1);
  61.         /*3. 检查定时器队列 将所有到时定时器任务进行调度*/
  62.         std::vector<std::function<void()> > cbs;
  63.         listExpiredCb(cbs);
  64.         if(cbs.size())
  65.         {
  66.             schedule(cbs.begin(), cbs.end());
  67.             cbs.clear();
  68.         }
  71.         /*4.依次处理已经就绪的IO*/
  72.         for(int i = 0;i < n_ready;++i)
  73.         {
  74.             struct epoll_event& event = events[i];
  76.             //过滤外部发消息唤醒的IO  跳过
  77.             if(event.data.fd == m_tickleFds[0])
  78.             {
  79.                 KIT_LOG_DEBUG(g_logger) << "读管道活跃";
  80.                 uint8_t temp;
  81.                 //循环的目的 把缓冲区全部读干净
  82.                 while(read(m_tickleFds[0], &temp, 1) == 1);
  84.                 continue;
  85.             }
  87.             //处理剩下的真正就绪的IO
  88.             FdContext* fd_ctx = (FdContext*)event.data.ptr;
  90.             FdContext::MutexType::Lock lock(fd_ctx->mutex);
  92.             //如果是错误或者中断 导致的活动  重置一下
  93.             if(event.events & (EPOLLERR | EPOLLHUP))
  94.                 event.events |= EPOLLIN | EPOLLOUT;
  96.             //开一个变量转换 从epoll_event的事件----->自定义的事件Event
  97.             int real_events = 0;
  98.             if(event.events & EPOLLIN)
  99.             {
  100.                 real_events |= READ;
  101.             }
  103.             if(event.events & EPOLLOUT)
  104.             {
  105.                 real_events |= WRITE;
  106.             }
  108.             //和当前IO上的事件对比 不符合就说明没有事件触发 跳过
  109.             if((fd_ctx->events & real_events) == Event::NONE)
  110.                 continue;
  112.             //把下面准备主动触发的事件 去除掉 剩余的事件放回epoll中
  113.             //BUG点:自定义的Event 和 
  114.             int left_events = (fd_ctx->events & ~real_events);
  116.             int op = left_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
  118.             //复用event
  119.             event.events = EPOLLET | left_events;
  121.             int ret = epoll_ctl(m_epfd, op, fd_ctx->fd, &event);
  122.             if(ret < 0)
  123.             {
  124.                 KIT_LOG_ERROR(g_logger) << "\nidel: epoll_ctl(" << m_epfd << ", " << op << ", " << fd_ctx->fd << ", " << fd_ctx->events << ");\n"
  125.                 << " error:" << ret << "(" << errno << "," << strerror(errno) << ")"; 
  127.                 continue;
  128.             }
  130.             //把需要触发的读事件 主动触发
  131.             if(real_events & READ)
  132.             {
  133.                 KIT_LOG_DEBUG(g_logger) << "idle 读事件触发";
  134.                 fd_ctx->triggerEvent(READ);
  135.                 --m_pendingEventCount;
  136.             }
  138.             //把需要触发的写事件 主动触发
  139.             if(real_events & WRITE)
  140.             {
  141.                 KIT_LOG_DEBUG(g_logger) << "idle 写事件触发";
  142.                 fd_ctx->triggerEvent(WRITE);
  143.                 --m_pendingEventCount;
  144.             }
  148.         }
  151.         /*4.处理完就绪的IO  让出当前协程的执行权 到Scheduler::run中去*/
  152.         Coroutine::ptr cur = Coroutine::GetThis();
  153.         auto p = cur.get();
  154.         cur.reset();
  156.         //从这切回 又会 从这切进
  157.         p->swapOut();
  159.     }
  161. }

4). onTimerInsertedAtFront()

功能：定时器队列队头插入对象后进行epoll_wait超时更新

  1. /**
  2.  * @brief 定时器队列队头插入对象后进行epoll_wait超时更新
  3.  */
  4. void onTimerInsertedAtFront() override;
  6. void IOManager::onTimerInsertedAtFront()
  7. {
  8.     //唤醒一下 在epoll_wait的线程
  9.     tickle();
  10. }

1.2.4 addEvent()

功能：为句柄添加/修改事件，创建对应的事件对象

  1. /**
  2.  * @brief 为句柄添加事件 0成功  -1出错
  3.  * @param[in] fd 给哪一个句柄fd添加
  4.  * @param[in] event 读/写事件
  5.  * @param[in] cb 要执行的回调函数
  6.  * @return 
  7.  *      @retval 0  添加成功
  8.  *      @retval -1 添加失败
  9.  * 
  10.  */
  11. int addEvent(int fd, Event event, std::function<void()> cb = nullptr);
  13. int IOManager::addEvent(int fd, Event event, std::function<void()> cb)
  14. {
  15.     FdContext *fd_ctx = nullptr;
  17.     /*拿到对应的 句柄对象  没有就创建*/
  18.     //给句柄队列加读锁
  19.     MutexType::ReadLock lock(m_mutex); 
  21.     if((int)m_fdContexts.size() > fd)
  22.     {
  23.         //KIT_LOG_DEBUG(g_logger) << "存在并取出 fd = " << fd;
  24.         fd_ctx = m_fdContexts[fd];
  25.         lock.unlock();  //解读锁
  26.     }
  27.     else  //事件对象扩容
  28.     {
  29.         //解读锁
  30.         lock.unlock();
  31.         //给句柄队列加写锁
  32.         MutexType::WriteLock lock2(m_mutex);
  34.         KIT_LOG_DEBUG(g_logger) << "fd不存在并扩容,fd=" << fd;
  35.         contextResize(fd * 1.5);
  36.         fd_ctx = m_fdContexts[fd];
  37.     }
  39.     //给句柄资源加互斥锁
  40.     FdContext::MutexType::Lock _lock(fd_ctx->mutex);
  42.     /*设置句柄对象的信息*/
  43.     //同一个句柄上面不能加加相同的事件
  44.     //如果有这种情况出现 说明有多个线程在操作同一个句柄
  45.     if(KIT_UNLIKELY(fd_ctx->events & event))   
  46.     {
  47.         KIT_LOG_ERROR(g_logger) << "addEvent: event exists, fd= " << fd
  48.             << ", event=" << event
  49.             << ";exist event=" << fd_ctx->events;
  50.         KIT_ASSERT(!(fd_ctx->events & event));
  51.     }
  53.     //判断本次事件修改还是新增
  54.     int op = fd_ctx->events ? EPOLL_CTL_MOD : EPOLL_CTL_ADD;
  56.     struct epoll_event ev;
  57.     ev.events = EPOLLET | fd_ctx->events | event;
  58.     ev.data.ptr = fd_ctx;
  60.     //将事件添加/修改到epoll
  61.     int ret = epoll_ctl(m_epfd, op, fd, &ev);
  62.     if(ret < 0)
  63.     {
  64.         KIT_LOG_ERROR(g_logger) << "\naddEvent: epoll_ctl(" << m_epfd << ", " << op << ", " << fd << ", " << ev.events << ");\n"
  65.         << " error:" << ret << "(" << errno << "," << strerror(errno) << ")"; 
  67.         return -1;
  68.     }
  70.     //待处理事件自增
  71.     ++m_pendingEventCount;
  73.     //将句柄上的事件叠加
  74.     fd_ctx->events = (Event)(fd_ctx->events | event);
  76.     //构建对应的添加的读/写 事件对象 设置相关信息
  77.     //要加 读事件就返回 read_event;要加写事件 就返回write_event
  78.     FdContext::EventContext& event_contex = fd_ctx->getEventContext(event);
  80.     KIT_ASSERT(!event_contex.scheduler && !event_contex.coroutine && !event_contex.cb);
  82.     //设置调度器
  83.     event_contex.scheduler = Scheduler::GetThis();
  85.     //设置回调函数
  86.     if(cb)
  87.         event_contex.cb.swap(cb);
  88.     else    //没有设置回调  下一次就继续执行当前协程
  89.     {
  91.         event_contex.coroutine = Coroutine::GetThis();
  93.         //给事件对象绑定协程时候 协程应该是运行的
  94.         KIT_ASSERT2(event_contex.coroutine->getState() == Coroutine::State::EXEC,  "thread id=" << GetThreadId() << ",coroutine id=" << event_contex.coroutine->getID() << ",state=" << event_contex.coroutine->getState());
  95.     }
  97.     return 0;
  99. }

1.2.5 delEvent()

功能：为句柄删除单个事件

  1. /**
  2.  * @brief 为句柄删除单个事件
  3.  * @param[in] fd 给哪一个句柄fd删除
  4.  * @param[in] event 读/写事件
  5.  * @return true 删除成功
  6.  * @return false 删除失败
  7.  */
  8. bool delEvent(int fd, Event event);
  10. bool IOManager::delEvent(int fd, Event event)
  11. {
  12.     MutexType::ReadLock lock(m_mutex);
  13.     //句柄对象不存在不用删除
  14.     if((int)m_fdContexts.size() <= fd)
  15.     {
  16.         return false;
  17.     }   
  19.     FdContext* fd_ctx = m_fdContexts[fd];
  20.     lock.unlock();
  22.     //给句柄资源加互斥锁
  23.     FdContext::MutexType::Lock lock2(fd_ctx->mutex);
  24.     //该句柄上没有对应事件 不用删除
  25.     if(!(fd_ctx->events & event))
  26.     {
  27.         return false;
  28.     }
  30.     //取反运算+与运算 就是去掉该事件event
  31.     Event left_events = (Event)(fd_ctx->events & ~event);
  33.     //去掉之后看句柄上还是否有剩余的事件  有就修改epoll 没有了就从epoll删除
  34.     int op = left_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
  36.     //构造epoll_event事件
  37.     struct epoll_event ev;
  38.     ev.events = EPOLLET | left_events;
  39.     ev.data.ptr = fd_ctx;
  41.     //将事件添加/修改到epoll
  42.     int ret = epoll_ctl(m_epfd, op, fd, &ev);
  43.     if(ret < 0)
  44.     {
  45.         KIT_LOG_ERROR(g_logger) << "\ndelEvent: epoll_ctl(" << m_epfd << ", " << op << ", " << fd << ", " << ev.events << ");\n"
  46.         << " error:" << ret << "(" << errno << "," << strerror(errno) << ")"; 
  48.         return false;
  49.     }
  52.     //更新句柄上的事件
  53.     fd_ctx->events = left_events;
  55.     //待处理事件对象自减
  56.     --m_pendingEventCount;
  58.     //把fdContext句柄对象中对应的读/写事件对象EventContext拿出来清空
  59.     FdContext::EventContext& event_context = fd_ctx->getEventContext(event);
  60.     fd_ctx->resetEventContext(event_context);
  62.     return true;
  64. }

1.2.6 cancelEvent()

功能：为句柄取消单个事件，找到对应事件强制触发执行，不等待句柄上的事件触发

  1. /**
  2.  * @brief 为句柄取消单个事件，找到对应事件强制触发执行，不等待条件满足
  3.  * @param[in] fd 给哪一个句柄fd取消
  4.  * @param[in] event 读/写事件
  5.  * @return true 取消并触发成功
  6.  * @return false 取消并触发失败
  7.  */
  8. bool cancelEvent(int fd, Event event);
  9. bool IOManager::cancelEvent(int fd, Event event)
  10. {
  11.     MutexType::ReadLock lock(m_mutex);
  12.     //句柄不存在不用触发
  13.     if((int)m_fdContexts.size() <= fd)
  14.     {
  15.         return false;
  16.     }   
  18.     FdContext* fd_ctx = m_fdContexts[fd];
  19.     lock.unlock();
  21.     FdContext::MutexType::Lock lock2(fd_ctx->mutex);
  22.     //该句柄上没有对应事件 不用触发
  23.     if(!(fd_ctx->events & event))
  24.     {
  25.         return false;
  26.     }
  28.     //取反运算 + 与运算 就是去掉该事件
  29.     Event left_events = (Event)(fd_ctx->events & ~event);
  31.     //去掉之后看句柄上还是否有剩余的事件  有就修改epoll 没有了就从epoll删除
  32.     int op = left_events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
  34.     //构造epoll_event事件
  35.     struct epoll_event ev;
  36.     ev.events = EPOLLET | left_events;
  37.     ev.data.ptr = fd_ctx;
  39.     //将事件添加/修改到epoll
  40.     int ret = epoll_ctl(m_epfd, op, fd, &ev);
  41.     if(ret < 0)
  42.     {
  43.         KIT_LOG_ERROR(g_logger) << "\ndelEvent: epoll_ctl(" << m_epfd << ", " << op << ", " << fd << ", " << ev.events << ");\n"
  44.         << " error:" << ret << "(" << errno << "," << strerror(errno) << ")"; 
  46.         return false;
  47.     }
  50.     //主动触发句柄上事件绑定的回调函数 重新加入到任务队列里
  51.     fd_ctx->triggerEvent(event);
  53.     //待处理事件对象自减
  54.     --m_pendingEventCount;
  56.     return true;
  59. }

1.2.7 cancelAll()

功能：为句柄取消所有事件，所有事件强制触发执行，不等待条件满足

  1. /**
  2.  * @brief 为句柄取消所有事件，所有事件强制触发执行，不等待条件满足
  3.  * @param[in] fd 给哪一个句柄fd取消
  4.  * @return true 取消并触发成功
  5.  * @return false 取消并触发失败
  6.  */
  7. bool cancelAll(int fd);
  9. bool IOManager::cancelAll(int fd)
  10. {
  11.     MutexType::ReadLock lock(m_mutex);
  12.     //句柄不存在不用删除
  13.     if((int)m_fdContexts.size() <= fd)
  14.     {
  15.         return false;
  16.     }   
  18.     FdContext* fd_ctx = m_fdContexts[fd];
  19.     lock.unlock();
  21.     FdContext::MutexType::Lock lock2(fd_ctx->mutex);
  22.     //该句柄上没有对应事件 不用删除
  23.     if(!fd_ctx->events)
  24.     {
  25.         return false;
  26.     }
  28.     //直接从epoll里移除该事件
  29.     int op = EPOLL_CTL_DEL;
  31.     //构造epoll_event事件
  32.     struct epoll_event ev;
  33.     ev.events = 0;
  34.     ev.data.ptr = fd_ctx;
  36.     //将事件删除到epoll
  37.     int ret = epoll_ctl(m_epfd, op, fd, &ev);
  38.     if(ret < 0)
  39.     {
  40.         KIT_LOG_ERROR(g_logger) << "\ndelEvent: epoll_ctl(" << m_epfd << ", " << op << ", " << fd << ", " << ev.events << ");\n"
  41.         << " error:" << ret << "(" << errno << "," << strerror(errno) << ")"; 
  43.         return false;
  44.     }
  48.     if(fd_ctx->events & READ)
  49.     {
  50.         //事件对象 主动触发读事件对象上的回调
  51.         fd_ctx->triggerEvent(READ);
  53.         //待处理事件对象自减
  54.         --m_pendingEventCount;
  55.     }
  57.     if(fd_ctx->events & WRITE)
  58.     {
  60.         //事件对象 主动触发写事件对象上的回调
  61.         fd_ctx->triggerEvent(WRITE);
  63.         //待处理事件对象自减
  64.         --m_pendingEventCount;
  66.     }
  68.     //句柄对象上的注册事件应该为NONE = 0
  69.     KIT_ASSERT(fd_ctx->events == 0);
  71.     return true;
  74. }

1.2.8 contextResize()

功能：保护类型接口，修改句柄对象数量

  1. /**
  2.  * @brief 修改句柄对象数量
  3.  * @param[in] size 要修改到的数量
  4.  */
  6. void contextResize(size_t size);
  8. void IOManager::contextResize(size_t size)
  9. {
  10.     m_fdContexts.resize(size);
  12.     for(size_t i = 0;i < m_fdContexts.size();++i)
  13.     {
  14.         if(!m_fdContexts[i])
  15.         {
  16.             m_fdContexts[i] = new FdContext;
  17.             m_fdContexts[i]->fd = i;
  18.         }
  19.     }
  21. }

2. 定时器Timer封装

功能：在间隔一个指定时间后，会自动的往任务队列中添加一个函数/协程，执行我们需要的操作。支持一次性定时器触发和循环定时器触发

使用场景：在特定时间段或者某一段时间内需要执行一些任务等场景

2.1 Timer类

Timer类不能直接创建其对象，构造函数置为私有属性，将其对象创建交由TimerManager类负责管理

Timer和TimerManager互相置为友元类。方便调用对方的接口以及访问对方的成员变量（主要为了能够调用Timer的构造函数，不能显式调用）；但是造成两个类的封装性被破坏，两个类的耦合度较高

  1. class Timer
  2. {
  3.     ...
  4.     ...
  5. private:
  6.     //Timer的构造函数为私有意味着不能显示创建对象  必须由TimerManager来创建
  7.     Timer(uint64_t ms, std::function<void()> cb, bool recurring, TimerManager *manager);
  9.     Timer(uint64_t next);
  10.     ...
  11.     ...
  12. }

2.1.1 成员变量

  2. class Timer
  3. {
  4.     ...
  5.     ...
  6. private:
  7.     /// 定时器间隔时长
  8.     uint64_t m_ms = 0;
  9.     /// 执行的回调函数
  10.     std::function<void()> m_cb;
  11.     /// 当前定时器是否是循环定时器
  12.     bool m_recurring = false;
  13.     /// 精确的定时器超时时间点
  14.     uint64_t m_next = 0;
  15.     /// 定时器管理类指针
  16.     TimerManager* m_manager = nullptr;
  18. };

2.1.2 接口

2.1.2.1 构造函数（私有）

功能：意味着不能显式创建对象 必须由TimerManager来创建

  1. /**
  2.  * @brief 定时器类构造函数 不能显式创建对象
  3.  * @param[in] ms 间隔时长
  4.  * @param[in] cb 执行的回调函数
  5.  * @param[in] recurring 是否是循环定时器
  6.  * @param[in] manager 哪一个定时器管理类
  7.  */
  8. Timer(uint64_t ms, std::function<void()> cb, bool recurring, TimerManager *manager);
  9. Timer::Timer(uint64_t ms, std::function<void()> cb, bool recurring, TimerManager *manager)
  10.     :m_ms(ms), m_cb(cb), m_recurring(recurring), m_manager(manager)
  11. {
  12.     //计算到期时间点
  13.     m_next = GetCurrentMs() + m_ms;
  14. }
  17. /**
  18.  * @brief 定时器类构造函数 不能显式创建对象
  19.  * @param[in] next 到期时间点
  20.  */
  21. Timer(uint64_t next);
  22. Timer::Timer(uint64_t next)
  23.     :m_next(next)
  24. {
  26. }

2.1.2.1 cancel()（核心）

功能：取消当前定时器的定时任务。
通过操作TimerManger的定时器队列，通过比较定时器的内存地址以及到期时间点，找到对应的定时器对象，从定时器队列中移除。

  1. /**
  2.  * @brief 取消定时器
  3.  * @return true 取消成功
  4.  * @return false 取消失败
  5.  */
  6. bool cancel();
  7. bool Timer::cancel()
  8. {
  9.     TimerManager::MutexType::WriteLock lock(m_manager->m_mutex);
  10.     if(m_cb)
  11.     {
  12.         m_cb = nullptr;
  13.         auto it = m_manager->m_timers.find(shared_from_this());
  14.         if(it == m_manager->m_timers.end())
  15.         {
  16.             KIT_LOG_ERROR(g_logger) << "Timer::cancel get timer error";
  17.             return false;
  18.         }
  19.         m_manager->m_timers.erase(it);
  21.         return true;
  22.     }
  24.     return false;
  26. }

2.1.2.2 refresh()

功能：重新刷新定时器的间隔时间，让其重新开始新的计时间隔，但不改变原来设定好的时间间隔量。

注意：由于使用的容器set<Timer::ptr>底层是红黑树，是一个有序结构，必须先将原来的定时器从这个有序序列中移除，再重新加入才能合法，否则直接原地修改会打乱有序规则，发生未知错误。

  1. /**
  2.  * @brief 刷新定时器时间
  3.  * @return true 刷新成功
  4.  * @return false 刷新失败
  5.  */
  6. bool refresh();
  7. bool Timer::refresh()
  8. {
  9.     TimerManager::MutexType::WriteLock lock(m_manager->m_mutex);
  10.     if(!m_cb)
  11.         return false;
  13.     auto it = m_manager->m_timers.find(shared_from_this());
  15.     if(it == m_manager->m_timers.end())
  16.     {
  17.         return false;
  18.     }
  20.     //注意：由于使用的set 要先移除再进行重置加入 不能直接在原来的定时器上直接修改时间
  21.     m_manager->m_timers.erase(it);
  22.     m_next = GetCurrentMs() + m_ms;
  23.     m_manager->m_timers.insert(shared_from_this());
  25.     return true;
  27. }

2.1.2.3 reset() (核心)

功能：重新设定当前定时器的时间间隔量。并且还要判断是从即刻生效还是继续上一次旧的间隔量基础上继续等待触发。默认情况下让其从此刻重新开始计时触发。

注意：重新设置时间间隔量，可能存在比原来的间隔量大的可能性，也存在比原来的间隔量小的可能性，都需要考虑处理。

  • 核心逻辑：
  1. 判断重新设定的时间间隔是否和原来相等，且重新设定是否从即刻生效。（如果和原来相等，并且即刻生效，其功能就会变成refresh()。）
  2. 获取TimerManager的读写锁资源RWMutex，加上写锁
  3. 从TimerManaer的定时器队列里找出对应的定时器移除，并且重新计算定时器的到期时间点：
    1. 如果from_now = true，即：即刻生效。获取当前时间GetCurrentMs()，将其作为新的时间起点，然后又加上新的时间间隔量。
    2. 如果from_now = false，即：不需要即刻生效。计算得到原来的时间起点原时间起点 = 原触发时间点 - 原时间间隔，在原时间起点的基础上加上新的时间间隔来完成触发。也就是说这种重置定时器的做法，设置完之后的定时器已经过了一部分时间，计划等待时间 >= 实际等待时间
  4. 写锁解锁
  5. 调用TimerManager::addTimer()将定时器重新加入队列，不直接使用set.insert()的原因在于：当前定时器的到期时间点 < 队头定时器(队列到期时间点最近的定时器)到期时间点，在加入队列时存在这种可能，需要tickle()去唤醒epoll_wait去修改其超时时间。
  1. /**
  2.  * @brief 重新设定定时器间隔时间
  3.  * @param[in] ms 间隔时间
  4.  * @param[in] from_now 是否从当前时间点开始重新计时 默认是
  5.  * @return true 重置成功
  6.  * @return false 重置失败
  7.  */
  8. bool reset(uint64_t ms, bool from_now = true);
  10. bool Timer::reset(uint64_t ms, bool from_now)
  11. {
  12.     if(m_ms == ms && !from_now)
  13.         return true;
  15.     TimerManager::MutexType::WriteLock lock(m_manager->m_mutex);
  17.     if(!m_cb)   //没有任务就要返回
  18.         return false;
  21.     auto it = m_manager->m_timers.find(shared_from_this());
  23.     if(it == m_manager->m_timers.end())
  24.     {
  25.         return false;
  26.     }
  27.     //注意：由于使用的set 要先移除再进行重置加入
  28.     m_manager->m_timers.erase(it);
  30.     uint64_t start = 0;
  31.     //重新从现在开始计时
  32.     if(from_now)
  33.         start = GetCurrentMs();
  34.     else //继续上一次的计时
  35.         start = m_next - m_ms;
  37.     //设置新的计时间隔
  38.     m_ms = ms;
  39.     //设置新的触发时间点
  40.     m_next = start + m_ms;
  42.     //写锁解锁
  43.     lock.unlock();
  45.     //因为reset可能出现执行时间变成最小的可能(放在队头) 会有一次唤醒
  46.     m_manager->addTimer(shared_from_this());
  48.     return true;
  49. }

2.2 TimerManager类

功能：管理Timer定时器队列，让其能够有序的执行到期后的定时任务。

置为一个虚基类，让其继承类通过纯虚函数virtual void onTimerInsertedAtFront() = 0去加入相关的其他业务拓展，TimerManger虚基类的基本功能就是负责管理一个定时器队列。

小技巧：Timer内部构建仿函数作为set容器比较的函数调用

目的：由于原本的set容器根据存储对象类型，默认比较存储类型的值。（存int比较int，存指针比较指针）。但是存储定时器队列我们不希望是通过比较指针值来排列，我们希望通过比较到期时间来进行排列，故需要重置比较函数。

  1. class Timer
  2. {
  3.     ....
  5. private:
  7.     /**
  8.      * @brief 构建仿函数 用于set比较
  9.      */
  10.     struct Comparator
  11.     {
  12.         bool operator()(const Timer::ptr& lhs, const Timer::ptr& rhs) const;
  13.     };
  15.     ...
  16. };
  18. bool Timer::Comparator::operator()(const Timer::ptr& lhs, const Timer::ptr& rhs) const
  19. {
  20.     /*判断地址*/
  21.     if(!lhs && !rhs)
  22.         return false;
  24.     if(!lhs)
  25.         return true;
  27.     if(!rhs)
  28.         return false;
  30.     /*都有值 判断到期时间*/
  31.     if(lhs->m_next < rhs->m_next)
  32.         return true;
  34.     if(lhs->m_next > rhs->m_next)
  35.         return false;
  37.     //到期时间相等的话 地址小的优先调度
  38.     return lhs.get() < rhs.get();
  40. }

2.2.1 成员变量

  1. class TimerManager
  2. {
  3.     ...
  5. private:
  6.     //读写锁
  7.     MutexType m_mutex;
  8.     //定时器队列
  9.     std::set<Timer::ptr, Timer::Comparator> m_timers; 
  10.     //避免频繁修改的一个ticked标记
  11.     bool m_tickled = false;
  12.     //旧的服务器的时间
  13.     uint64_t m_previousTime = 0;
  14. };

2.2.2 接口

2.2.2.1 构造函数

  1. /**
  2.  * @brief 定时器管理类构造函数
  3.  */
  4. TimerManager();
  6. TimerManager::TimerManager()
  7. {
  8.     //初始化时候 获取一次服务器时间
  9.     m_previousTime = GetCurrentMs();
  10. }

2.2.2.2 addTimer()

功能：通过传参，创建并往定时器队列中添加一个定时器
注意：在这个调用接口中创建定时器，又因为Timer构造函数已经置为私有属性，必须将TimerManager置为Timer的友元类方可调用。

  1. //创建并添加定时器
  2. Timer::ptr TimerManager::addTimer(uint64_t ms, std::function<void()> cb, bool recurring)
  3. {
  4.     //在TimerManager 中构造Timer
  5.     Timer::ptr timer(new Timer(ms, cb, recurring, this));
  7.     addTimer(timer);
  9.     return timer;
  10. }

  • 如果不考虑外部访问的安全性，在构造函数私有情况下，也有不需要置为友元类也能创建对象的方法，但是该方法没有到达我们想要杜绝单独创建一个Timer对象的目的： ```cpp class Timer { public:

    static Timer::ptr getNewTimer() {

    1.   Timer::ptr timer(new Timer(ms, cb, recurring, IOMager::GetThis()));
    3.   ...
    5.   reutrn timer;

    }

private: //Timer的构造函数为私有意味着不能显示创建对象 必须由TimerManager来创建 Timer(uint64_t ms, std::function cb, bool recurring, TimerManager *manager);

  1. Timer(uint64_t next);
  2. ...

};

  2. <a name="S5mMu"></a>
  3. #### 2.2.2.3 `addTimer(Timer::ptr p)` （核心）
  4. 功能：通过传入定时器，往定时器队列中添加一个定时器
  6. - **核心逻辑：**
  7. 1. 加写锁
  8. 1. 调用`set.insert()`插入定时器队列，并获取插入后的位置
  9. 1. 如果此时定时器被插入到队头，且唤醒标志`m_tickled == false`，认为此时队头的最近到期时间点发生改变`is_front = true`，需要去唤醒线程修改对应的`epoll_wait`的超时时间
  10. 1. 写锁解锁
  11. 1. 根据插入情况，选择唤醒
  12. ```cpp
  13. /**
  14.  * @brief 以智能指针形式 添加条件定时器
  15.  * @param[in] p 定时器智能指针
  16.  */
  17. void addTimer(Timer::ptr p);   
  19. void TimerManager::addTimer(Timer::ptr p)
  20. {
  21.     MutexType::WriteLock lock(m_mutex);
  23.     //insert返回两个值 插入后的迭代器位置/插入是否成功
  24.     //判断插入后是否是在队头 最小的到时时间
  25.     auto it = m_timers.insert(p).first;
  26.     bool is_front = (it == m_timers.begin()) && !m_tickled;
  27.     //频繁修改时候 避免总是去唤醒
  28.     if(is_front)
  29.         m_tickled = true;
  31.     lock.unlock();
  33.     //当有比之前更小的 定时器任务插入  就要通知epoll_wait那边去修改为更小的等待时间
  34.     //实际上：将挂起的协程唤醒，重新走一遍流程idle()---->run()--->idle()
  35.     if(is_front)
  36.         onTimerInsertedAtFront();
  38. }

2.2.2.4 addConditionTimer()

功能：添加一个条件定时器，和普通定时器相比，多了一个用weak_ptr<>弱指针指向shared_ptr<>共享指针管理的”条件”，定时器触发时检查”条件”是否还存在，不存在将不执行所持有的任务

  1. /**
  2.  * @brief 添加条件定时器辅助函数
  3.  * @param[in] weak_cond 判断条件是否存在的弱指针 
  4.  * @param[in] cb 触发的回调函数 
  5.  */
  6. static void OnTimer(std::weak_ptr<void> weak_cond, std::function<void()> cb)
  7. {   
  8.     //利用弱指针weak_ptr 来判断条件是否存在
  9.     std::shared_ptr<void> temp = weak_cond.lock();
  10.     if(temp)
  11.     {
  12.         cb();
  13.     }
  15. }
  17. /**
  18.  * @brief 添加条件定时器
  19.  * @param[in] ms 间隔时长
  20.  * @param[in] cb 执行的回调函数
  21.  * @param[in] weak_cond 判断条件是否还存在的弱指针
  22.  * @param[in] recurring 是否是循环定时器 默认不是 
  23.  * @return Timer::ptr 添加完成后会将创建的定时器返回
  24.  */
  25. Timer::ptr addConditionTimer(uint64_t ms, std::function<void()> cb, std::weak_ptr<void> weak_cond, bool recurring = false);
  27. Timer::ptr TimerManager::addConditionTimer(uint64_t ms, std::function<void()> cb, std::weak_ptr<void> weak_cond, bool recurring)
  28. {
  29.     return addTimer(ms, std::bind(&OnTimer, weak_cond, cb), recurring);
  30. }

2.2.2.5 getNextTime() （核心）

功能：获取当前定时器队列队头的到期时间点的值。如果已经超出了触发时间点，返回0

  1. /**
  2.  * @brief 获取队头定时器的到期时间点
  3.  * @return uint64_t 
  4.  */
  5. uint64_t getNextTime();
  7. uint64_t TimerManager::getNextTime()
  8. {
  9.     MutexType::ReadLock lock(m_mutex);
  10.     //获取队头定时器时间一次 就可以去唤醒一次
  11.     m_tickled = false;
  13.      //定时器队列为空返回一个极大值
  14.     if(!m_timers.size())    
  15.         return ~0ull; 
  17.     const Timer::ptr& next = *m_timers.begin();
  18.     uint64_t now_ms = GetCurrentMs();
  19.     if(now_ms >= next->m_next)  //现在获取的时间 已经晚于预计要触发的时间点 马上执行
  20.         return 0;
  21.     else    //还没到预定时间就返回剩余时间间隔
  22.         return next->m_next - now_ms;
  23. }

2.2.2.6 listExpiredCb() (核心)

功能：检查当前的定时队列，找出所有已经到时需要触发的定时器，从定时器队列中移除，并且将这些已经要触发的定时器中的函数/协程全部取出放入一个目标容器内。

注意：处理服务器时间被修改的情况，会对定时器触发有一定的影响

  • **checkClockChange()** （私有）

功能：探测当前服务器时间是否被往前调小了。被调小的判定：当前时间 < 上一次记录的时间 && 当前时间 < 上一次记录的时间的一个小时前的时间 ， 重点检查时间被调小的情况因为可能会导致定时器队列里所有任务都触发。

  1. /**
  2.  * @brief 探测当前服务器时间是否改变 
  3.  * @param[in] now_ms 当前的时间点 
  4.  * @return true 已经被修改
  5.  * @return false 没有被修改
  6.  */
  7. bool checkClockChange(uint64_t now_ms);
  9. bool TimerManager::checkClockChange(uint64_t now_ms)
  10. {
  11.     bool changed = false;
  12.     //当前时间小于上一次时间且比上一次时间一小时前还要小 就认为服务器时间被修改
  13.     if(now_ms < m_previousTime && now_ms < (m_previousTime - 60 * 60 * 1000))
  14.         changed = true;
  16.     m_previousTime = now_ms;
  17.     return changed;
  18. }
  1. /**
  2.  * @brief 找出已经到期的定时任务集合
  3.  * @param[out] cbs 存储全部已经到期的任务
  4.  */
  5. void listExpiredCb(std::vector<std::function<void()> >& cbs);
  7. void TimerManager::listExpiredCb(std::vector<std::function<void()> >& cbs)
  8. {
  9.     uint64_t now_ms = GetCurrentMs();
  10.     std::vector<std::shared_ptr<Timer> > expired;
  12.     {
  13.         MutexType::ReadLock lock(m_mutex);
  14.         if(!m_timers.size())
  15.             return;
  16.     }
  18.     MutexType::WriteLock lock(m_mutex);
  19.     if(!m_timers.size())
  20.         return;
  22.     bool changed = checkClockChange(now_ms);
  23.     //服务器时间没有发生改变 且不存在超时定时器 就退出函数
  24.     if(!changed && (*m_timers.begin())->m_next < now_ms)
  25.         return;
  27.     //构造一个装有当前时间的Timer为了应用lower_bound算法函数
  28.     Timer::ptr now_timer(new Timer(now_ms));
  30.     //如果服务器时间发生了变动 就返回end() 会将整个m_timers全部清理 重新加入队列
  31.     //否则 将找出大于等于当前时间的第一个Timer
  32.     auto it = changed ? m_timers.end() : m_timers.lower_bound(now_timer);
  34.     //找到还没到时的第一个定时器的位置后 停下
  35.     while(it != m_timers.end() && (*it)->m_next == now_ms)
  36.         ++it;
  38.     //将队头~当前位置之前 所有已经到时的定时器全部拿出 到expired队列中
  39.     expired.insert(expired.begin(), m_timers.begin(), it);
  40.     //将原队列前部到期的定时器全部移除
  41.     m_timers.erase(m_timers.begin(), it);
  43.     //扩充可执行任务队列空间
  44.     cbs.reserve(expired.size());
  45.     for(auto &x : expired)
  46.     {
  47.         cbs.emplace_back(x->m_cb);
  48.         //循环定时器就放回 set中
  49.         if(x->m_recurring)
  50.         {
  51.             x->m_next = now_ms + x->m_ms;
  52.             m_timers.insert(x);
  53.         }
  54.         else
  55.             x->m_cb = nullptr;
  57.     }
  59. }

C\C++知识点补充复习：fcntl函数

出处：《UNIX环境高级编程 第3版》P66
功能：修改已经打开文件的各种属性

  1. #include <unistd.h>
  2. #include <fcntl.h>
  4. int fcntl(int fd, int cmd, ... /* arg */ );

具体功能：

  1. 复制已有的文件描述符F_DUPFD 或 F_DUPFD_CLOEXEC
  2. 获取/设置文件描述符标志F_GETFD 和 F_SETFD
  3. 获取/设置文件描述符状态F_GETFL 和 F_SETFL
  4. 获取/设置异步IO所有权F_GETTOWN 和 F_SETTOWN
  5. 获取/设置记录数 F_GETLK 和F_SETLK、F_SETLKW
  • 文件状态标志位：

image.png

C\C++知识点补充复习：算法lower_bound函数

头文件<algorithm>
功能：对于有序容器，返回一个迭代器，指向大于等于key的第一个元素。
触类旁通：upper_bound函数返回的迭代器，指向小于等于key的第一个元素