muduo源码剖析之TimerQueue类

news2024/11/24 0:18:59

简介

TimerQueue

​ 通过timerfd实现的定时器功能，为EventLoop扩展了一系列runAt，runEvery，runEvery等函数TimerQueue中通过std::set维护所有的Timer，也可以使用优先队列实现

muduo的TimerQueue是基于timerfd_create实现，这样超时很容易和epoll结合起来。等待超时事件保存在set集合中，注意set集合的有序性，从小到大排列，整个对TimerQueue的处理也就是对set集合的操作。实现TimerQueue用了3个set，分别是等待超时事件set，活跃事件set，被撤销定时set。主要是STL的一些操作。

主要成员及属性解析

主要接口

  • addTimer

向定时器中添加Timer
Timer是一个封装了回调函数和时间的类
通过内部实现addTimerInLoop保证线程安全

  • cancel

从定时器中移除某个Timer

  • 核心实现：getExpired

从timers_集合中移除已经到期的Timer

  • 核心实现：handleRead

向timerfdChannel注册的回调函数
在timerfd触发可读时间时，也就是定时器到期的时候执行
会调用getExpired，并依次执行返回的所有Timer中的回调

主要成员

  • timerfdChannel_

用来唤醒计时器的Channel，维护了一个timerfd，注册了TimerQueue::handleRead回调

  • std::set<std::pair<Timestamp, Timer*>> timers_

使用std::set容器来取得最近将要超时的Timer，从而决定是否resetTimerfd

TimerQueue执行用户回调的时序图：

img

源码剖析

TimerQueue.h

// Copyright 2010, Shuo Chen.  All rights reserved.
// http://code.google.com/p/muduo/
//
// Use of this source code is governed by a BSD-style license
// that can be found in the License file.

// Author: Shuo Chen (chenshuo at chenshuo dot com)
//
// This is an internal header file, you should not include this.

#ifndef MUDUO_NET_TIMERQUEUE_H
#define MUDUO_NET_TIMERQUEUE_H

#include <set>
#include <vector>

#include "muduo/base/Mutex.h"
#include "muduo/base/Timestamp.h"
#include "muduo/net/Callbacks.h"
#include "muduo/net/Channel.h"

namespace muduo
{
namespace net
{

class EventLoop;
class Timer;
class TimerId;

///
/// A best efforts timer queue.
/// No guarantee that the callback will be on time.
///
class TimerQueue : noncopyable
{
 public:
  explicit TimerQueue(EventLoop* loop);
  ~TimerQueue();

  ///
  /// Schedules the callback to be run at given time,
  /// repeats if @c interval > 0.0.
  ///
  /// Must be thread safe. Usually be called from other threads.
  //添加定时事件
  TimerId addTimer(TimerCallback cb,
                   Timestamp when,
                   double interval);
  //取消定时事件
  void cancel(TimerId timerId);

 private:

  // FIXME: use unique_ptr<Timer> instead of raw pointers.
  // This requires heterogeneous comparison lookup (N3465) from C++14
  // so that we can find an T* in a set<unique_ptr<T>>.
  typedef std::pair<Timestamp, Timer*> Entry;
  typedef std::set<Entry> TimerList;
  typedef std::pair<Timer*, int64_t> ActiveTimer;
  typedef std::set<ActiveTimer> ActiveTimerSet;

  //在workloop中执行实际添加的定时事件的回调函数
  void addTimerInLoop(Timer* timer);
  
  //在workloop中执行实际删除定时事件的回调函数
  void cancelInLoop(TimerId timerId);
  
  // called when timerfd alarms
  //定时器超时时候被调用处理事件的回调函数
  void handleRead();
  
  // move out all expired timers
  //获取所有的超时事件
  std::vector<Entry> getExpired(Timestamp now);
  
  //重置刷新定时器列表
  void reset(const std::vector<Entry>& expired, Timestamp now);

  //将一个定时事件加入到定时列表中
  bool insert(Timer* timer);

  //所属loop的指针
  EventLoop* loop_;
  
  //timerfd文件描述符
  const int timerfd_;
  
  //存储timerfd事件的channel
  Channel timerfdChannel_;
  
  // Timer list sorted by expiration
  //按过期时间排序的定时器列表
  TimerList timers_;

  // for cancel()
  //存储活跃定时事件的容器
  ActiveTimerSet activeTimers_;

  //正在处理超时事件的标志
  bool callingExpiredTimers_; /* atomic */
  
  //存储所有需要取消定时事件的容器
  ActiveTimerSet cancelingTimers_;
};

}  // namespace net
}  // namespace muduo
#endif  // MUDUO_NET_TIMERQUEUE_H

TimerQueue.cc

// Copyright 2010, Shuo Chen.  All rights reserved.
// http://code.google.com/p/muduo/
//
// Use of this source code is governed by a BSD-style license
// that can be found in the License file.

// Author: Shuo Chen (chenshuo at chenshuo dot com)

#ifndef __STDC_LIMIT_MACROS
#define __STDC_LIMIT_MACROS
#endif

#include "muduo/net/TimerQueue.h"

#include "muduo/base/Logging.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/Timer.h"
#include "muduo/net/TimerId.h"

#include <sys/timerfd.h>
#include <unistd.h>

namespace muduo
{
namespace net
{
namespace detail
{

//创建一个timerfd
int createTimerfd()
{
  //创建一个timerFd,设置为单调,非阻塞,fork+exec关闭fd
  int timerfd = ::timerfd_create(CLOCK_MONOTONIC,
                                 TFD_NONBLOCK | TFD_CLOEXEC);
  if (timerfd < 0)
  {
    LOG_SYSFATAL << "Failed in timerfd_create";
  }
  return timerfd;
}

//计算定时器触发还有多长
struct timespec howMuchTimeFromNow(Timestamp when)
{
  //当前时间与when的时间间隔
  int64_t microseconds = when.microSecondsSinceEpoch()
                         - Timestamp::now().microSecondsSinceEpoch();
  if (microseconds < 100)//100起步
  {
    microseconds = 100;
  }
  struct timespec ts;
  //秒
  ts.tv_sec = static_cast<time_t>(
      microseconds / Timestamp::kMicroSecondsPerSecond);
  //纳秒
  ts.tv_nsec = static_cast<long>(
      (microseconds % Timestamp::kMicroSecondsPerSecond) * 1000);
  return ts;
}

void readTimerfd(int timerfd, Timestamp now)
{
  //在timerfd在读取8字节
  uint64_t howmany;
  ssize_t n = ::read(timerfd, &howmany, sizeof howmany);
  LOG_TRACE << "TimerQueue::handleRead() " << howmany << " at " << now.toString();
  if (n != sizeof howmany)
  {
    LOG_ERROR << "TimerQueue::handleRead() reads " << n << " bytes instead of 8";
  }
}

//重置定时器超时时间戳
void resetTimerfd(int timerfd, Timestamp expiration)
{
  // wake up loop by timerfd_settime()
  struct itimerspec newValue;
  struct itimerspec oldValue;
  memZero(&newValue, sizeof newValue);
  memZero(&oldValue, sizeof oldValue);
  
  newValue.it_value = howMuchTimeFromNow(expiration);
  //设置下一次定时事件的到达时间
  int ret = ::timerfd_settime(timerfd, 0, &newValue, &oldValue);
  if (ret)
  {
    LOG_SYSERR << "timerfd_settime()";
  }
}

}  // namespace detail
}  // namespace net
}  // namespace muduo

using namespace muduo;
using namespace muduo::net;
using namespace muduo::net::detail;

TimerQueue::TimerQueue(EventLoop* loop)
  : loop_(loop),
    timerfd_(createTimerfd()),
    timerfdChannel_(loop, timerfd_),
    timers_(),
    callingExpiredTimers_(false)
{
  //设置读事件的回调函数
  timerfdChannel_.setReadCallback(
      std::bind(&TimerQueue::handleRead, this));
  // we are always reading the timerfd, we disarm it with timerfd_settime.
  //监听读事件
  timerfdChannel_.enableReading();
}

TimerQueue::~TimerQueue()
{
  //取消所有的监听事件
  timerfdChannel_.disableAll();
  //将自己从polller中移除
  timerfdChannel_.remove();
  //关闭fd
  ::close(timerfd_);
  // do not remove channel, since we're in EventLoop::dtor();
  //释放定时器列表中所有的定时器
  for (const Entry& timer : timers_)
  {
    delete timer.second;
  }
}

//添加定时事件
TimerId TimerQueue::addTimer(TimerCallback cb,//超时回调
                             Timestamp when,//时间戳
                             double interval)//时间间隔
{
  //创建一个定时器
  Timer* timer = new Timer(std::move(cb), when, interval);
  //添加定时器
  loop_->runInLoop(
      std::bind(&TimerQueue::addTimerInLoop, this, timer));
  
  return TimerId(timer, timer->sequence());
}

//取消一个定时事件
void TimerQueue::cancel(TimerId timerId)
{
  loop_->runInLoop(
      std::bind(&TimerQueue::cancelInLoop, this, timerId));
}

void TimerQueue::addTimerInLoop(Timer* timer)
{
  loop_->assertInLoopThread();
  //将定时事件插入定时器列表
  bool earliestChanged = insert(timer);

  //如果这个定时事件将会是最先被触发的,那么就重新设置定时器超时时间戳
  if (earliestChanged)
  {
    resetTimerfd(timerfd_, timer->expiration());
  }
}

//取消一个定时器
void TimerQueue::cancelInLoop(TimerId timerId)
{
  loop_->assertInLoopThread();
  assert(timers_.size() == activeTimers_.size());
  //寻找这个定时器
  ActiveTimer timer(timerId.timer_, timerId.sequence_);
  ActiveTimerSet::iterator it = activeTimers_.find(timer);
  if (it != activeTimers_.end())//找到了
  {
    //删除+释放
    size_t n = timers_.erase(Entry(it->first->expiration(), it->first));
    assert(n == 1); (void)n;
    delete it->first; // FIXME: no delete please
    activeTimers_.erase(it);
  }
  //如果此时正在处理过期事件,那么就将该定时器加入到过期事件列表
  else if (callingExpiredTimers_)
  {
    cancelingTimers_.insert(timer);
  }
  assert(timers_.size() == activeTimers_.size());
}

//timer读事件的回调函数
void TimerQueue::handleRead()
{
  loop_->assertInLoopThread();
  Timestamp now(Timestamp::now());
  //read timer
  readTimerfd(timerfd_, now);

  //获取所有超时的定时事件
  std::vector<Entry> expired = getExpired(now);

  callingExpiredTimers_ = true;
  //释放所有过期事件
  cancelingTimers_.clear();
  // safe to callback outside critical section
  //执行所有超时事件的回调函数
  for (const Entry& it : expired)
  {
    it.second->run();
  }
  callingExpiredTimers_ = false;

  //重置定时器列表
  reset(expired, now);
}

//获取所有超时的定时事件
std::vector<TimerQueue::Entry> TimerQueue::getExpired(Timestamp now)
{
  assert(timers_.size() == activeTimers_.size());
  std::vector<Entry> expired;
  //获取当前时间戳
  Entry sentry(now, reinterpret_cast<Timer*>(UINTPTR_MAX));
  //返回指向第一个不小于给定值的元素的迭代器
  //也就是通过二分搜索找到第一个时间戳大于当前时间点的迭代器
  TimerList::iterator end = timers_.lower_bound(sentry);
  assert(end == timers_.end() || now < end->first);
  //back_inserter(expired)将获取vector尾部可插入元素的迭代器
  //将所有超时事件都复制到vector中
  std::copy(timers_.begin(), end, back_inserter(expired));
  //删除所有超时事件
  timers_.erase(timers_.begin(), end);

  //将所有超时事件都从活跃事件列表中摘除
  for (const Entry& it : expired)
  {
    ActiveTimer timer(it.second, it.second->sequence());
    size_t n = activeTimers_.erase(timer);
    assert(n == 1); (void)n;
  }

  assert(timers_.size() == activeTimers_.size());
  return expired;
}

//重置定时器列表
//expired中存放过期的定时器事件
void TimerQueue::reset(const std::vector<Entry>& expired, Timestamp now)
{
  Timestamp nextExpire;

  for (const Entry& it : expired)
  {
    ActiveTimer timer(it.second, it.second->sequence());
	
    if (it.second->repeat()
        && cancelingTimers_.find(timer) == cancelingTimers_.end())
    {
    //如果这个定时事件设置了重复执行,并且没有被取消,
      it.second->restart(now);//刷新时间戳
      insert(it.second);//加入定时器列表
    }
    else//否则释放掉
    {
      // FIXME move to a free list
      delete it.second; // FIXME: no delete please
    }
  }

  //如果定时器列表非空,就获取首节点的时间戳
  if (!timers_.empty())
  {
    nextExpire = timers_.begin()->second->expiration();
  }
  //刷新重置timer的下一次超时时间戳
  if (nextExpire.valid())
  {
    resetTimerfd(timerfd_, nextExpire);
  }
}

//将一个定时事件加入到定时列表中
bool TimerQueue::insert(Timer* timer)
{
  loop_->assertInLoopThread();
  assert(timers_.size() == activeTimers_.size());
  //此变量用来标记这个待插入的定时事件是不是会最早触发的
  bool earliestChanged = false;
  //获取到该定时器的时间戳
  Timestamp when = timer->expiration();
  
  TimerList::iterator it = timers_.begin();
  if (it == timers_.end() || when < it->first)
  {
    earliestChanged = true;
  }
  {
    //将定时事件插入事件列表
    std::pair<TimerList::iterator, bool> result
      = timers_.insert(Entry(when, timer));
    assert(result.second); (void)result;
  }
  {
    //将定时事件插入活跃事件列表
    std::pair<ActiveTimerSet::iterator, bool> result
      = activeTimers_.insert(ActiveTimer(timer, timer->sequence()));
    assert(result.second); (void)result;
  }

  assert(timers_.size() == activeTimers_.size());
  return earliestChanged;
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1185863.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Jupyter Notebook 内核似乎挂掉了，它很快将自动重启

Jupyter Notebook 内核似乎挂掉了，它很快将自动重启

报错原因&#xff1a; OMP: Error #15: Initializing libiomp5md.dll, but found libiomp5md.dll already initialized. OMP: Hint This means that multiple copies of the OpenMP runtime have been linked into the program. That is dangerous, since it can degrade perfo…
阅读更多...
AD教程 （十）Value值的核对

AD教程 （十）Value值的核对

AD教程 &#xff08;十&#xff09;Value值的核对 填写器件位号 直接根据原理图的原始编号进行更改 通过位号编辑器快速更改 点击工具&#xff0c;选择标注&#xff0c;选择原理图标注&#xff0c;进入位号编辑器 可以在位号编辑器中 设置处理顺序&#xff0c;从上往下还是从…
阅读更多...
推荐一款网络拓扑自动扫描工具

推荐一款网络拓扑自动扫描工具

topology-scanner Topology-Scanner是WeOps团队免费开放的一个网络拓扑自动扫描模块&#xff0c;可以自动发现网络设备的类型、网络设备之间的互联 资源下载地址&#xff1a;https://download.csdn.net/download/XMWS_IT/88510697 或 加 薇|信|号吗&#xff1a;xmws-IT 免…
阅读更多...
C嘎嘎之类和对象下

C嘎嘎之类和对象下

> 作者简介&#xff1a;დ旧言~&#xff0c;目前大二&#xff0c;现在学习Java&#xff0c;c&#xff0c;c&#xff0c;Python等 > 座右铭&#xff1a;松树千年终是朽&#xff0c;槿花一日自为荣。 > 目标&#xff1a;了解顺序表和链表的利弊&#xff0c;能在不同的题…
阅读更多...
MathType安装激活教程安装后点击exe闪退问题解决

MathType安装激活教程安装后点击exe闪退问题解决

MathType安装 官方下载地址&#xff1a;https://mathtype.org/download/ 1.打开下载好的安装包 2.点击我接受 3.等待。 4.默认C盘路径安装&#xff0c;不要修改&#xff0c;点击确定。 5.等待。 6.点击确定。 7.退出安装&#xff0c;安装完成&#xff01; 8.软件安装后&…
阅读更多...
人工智能基础_机器学习020_归一化实战_天池工业蒸汽量项目归一化实战过程_使用sklearn库中的最大最小值归一化和_sklearn库中的Z-score归一化_01---人工智能工作笔记0060

人工智能基础_机器学习020_归一化实战_天池工业蒸汽量项目归一化实战过程_使用sklearn库中的最大最小值归一化和_sklearn库中的Z-score归一化_01---人工智能工作笔记0060

首先我们找到对应的数据,天池蒸汽项目的数据. import pandas as pd 导入文件读取库 import numpy as np 导入数学计算库 df=pd.readcsv(./zhenggi_train.txt,sep=\t) 读取csv文件，这个就是天池的用tab分割的数据 X_train = df . iloc[:,: - 1]，我们我们切片，从0列，切片到…
阅读更多...
Loki | 数据过期自动删除策略设计

Loki | 数据过期自动删除策略设计

最近使用 Loki 碰到一个比较蛋疼问题&#xff0c;配置日志过期时间&#xff0c;配置这种事情&#xff0c;自然是要参照官方文档了&#xff0c;当时就找到了这个文档&#xff0c;地址&#xff1a; https://github.com/grafana/loki/blob/v1.5.0/docs/operations/storage/retenti…
阅读更多...
手写txt模拟数据库简易框架

手写txt模拟数据库简易框架

一.前言 之前学校让我们写控制台饿了么项目&#xff0c;当时进度没有到数据库&#xff0c;所以用的文本文件txt&#xff0c;来模拟数据库的实现&#xff0c;其实本质上就是一个文件读写&#xff0c;但是简单文件读写并不能包含我们想要的功能&#xff0c;例如条件查询&#xff…
阅读更多...
【算法秘籍】藏在0和1之间的秘密，助你码出优秀人生

【算法秘籍】藏在0和1之间的秘密，助你码出优秀人生

《算法秘籍》双十一 5折购书&#xff0c;就在京东商城 数据结构和算法是计算机科学的基石&#xff0c;是计算机的灵魂&#xff0c;要想成为计算机专业人员&#xff0c;学习和掌握算法是十分必要的。不懂数据结构和算法的人不可能写出效率更高的代码。计算机科学的很多新行业都离…
阅读更多...
Discord将切换到临时文件链接以阻止恶意软件传播

Discord将切换到临时文件链接以阻止恶意软件传播

导语 Discord是一个广受欢迎的社交平台&#xff0c;但其服务器长期以来一直是恶意活动的温床。为了解决这个问题&#xff0c;Discord决定采取行动&#xff0c;将切换到临时文件链接&#xff0c;以阻止恶意软件的传播。本文将介绍Discord的最新举措&#xff0c;并探讨其对用户安…
阅读更多...
C盘清理指南（三）——文件目录更改

C盘清理指南（三）——文件目录更改

各位小伙伴你们好&#xff0c;今天的推送是C盘清理系列的第三期——文件路径更改&#xff0c;分为文件夹路径和软件默认路径两个模块。 一&#xff0e;文件夹路径更改 点击进入C盘&#xff0c;依次点击上方“查看——隐藏的项目”&#xff0c;可以看到C盘中各种隐藏目录。 单击…
阅读更多...
kafka微服务学习

kafka微服务学习

消息中间件对比&#xff1a; 1、吞吐、可靠性、性能 Kafka安装 Kafka对于zookeeper是强依赖&#xff0c;保存kafka相关的节点数据&#xff0c;所以安装Kafka之前必须先安装zookeeper Docker安装zookeeper 下载镜像&#xff1a; docker pull zookeeper:3.4.14创建容器 do…
阅读更多...
Linux 下最主流的文件系统格式——ext

Linux 下最主流的文件系统格式——ext

硬盘分成相同大小的单元&#xff0c;我们称为块&#xff08;Block&#xff09;。一块的大小是扇区大小的整数倍&#xff0c;默认是 4K。在格式化的时候&#xff0c;这个值是可以设定的。 一大块硬盘被分成了一个个小的块&#xff0c;用来存放文件的数据部分。这样一来&#xf…
阅读更多...
操作系统 day06（进程控制、原语）

操作系统 day06（进程控制、原语）

进程控制的概念 原语 怎么实现进程控制—用原语实现 如果不能一气呵成&#xff0c;那么会出现操作系统中的某些关键数据结构信息不统一的情况&#xff0c;这会影响操作系统进行别的管理工作&#xff0c;如下图所示&#xff1a; 原语的原子性怎么实现 正常情况下&#xff…
阅读更多...
Source insight 创建工程研读库源码

Source insight 创建工程研读库源码

下载库文件解压 创建工程添加路径 添加文件到工程 同步文件 下载库文件解压 创建工程添加路径 添加文件到工程 同步文件 数据库更新 强制重新分析所有文件仅同步当前源文件 添加和删除文件 自动添加新文件从项目中删除丢失的文件
阅读更多...
程序员想要网上接单？那这几点注意事项你可要记好了！不看后悔！

程序员想要网上接单？那这几点注意事项你可要记好了！不看后悔！

相信网上接单对于程序员来说并不陌生&#xff0c;甚至有些程序员还以此为主业&#xff0c;靠网上接单来增加收入&#xff0c;维持生计&#xff0c;但是你真的确定你懂网上接单的套路吗&#xff1f;你知道网上接单的注意事项吗&#xff1f;这期文章就来盘点一下&#xff0c;无论…
阅读更多...
跟着森老师学React Hooks(1)——使用Vite构建React项目

跟着森老师学React Hooks(1)——使用Vite构建React项目

Vite是一款构建工具&#xff0c;对ts有很好的支持&#xff0c;最近也是在前端越来越流行。 以往的React项目的初始化方式大多是通过脚手架create-react-app(本质是webpack)&#xff0c;其实比起Vite来构建&#xff0c;启动会慢一些。 所以这次跟着B站的一个教程&#xff0c;使用…
阅读更多...
JavaWeb Day05 前后端请求响应与分层解耦

JavaWeb Day05 前后端请求响应与分层解耦

目录 一、请求与响应 &#xff08;一&#xff09;请求的参数接收 ①数组参数 ②集合参数 ③日期参数 ④json参数 ⑤路径参数 总结 &#xff08;二&#xff09;响应 ①简单文本text ②数组 ③列表 ④同一响应数据格式 ⑤总结 二、三层架构与分层解耦 &#xff0…
阅读更多...
Windows下MSYS2下载与安装

Windows下MSYS2下载与安装

下载地址&#xff1a; 官网下载地址 https://www.msys2.org/阿里云镜像下载 https://mirrors.aliyun.com/msys2/distrib/x86_64/https://mirrors.aliyun.com/msys2/distrib/x86_64/msys2-x86_64-20231026.exe?spma2c6h.25603864.0.0.12b92551XW5OSM官网下载 ![官网下载](htt…
阅读更多...
程序员怎样才能学好算法，推荐好书送给大家

程序员怎样才能学好算法，推荐好书送给大家

前言 数据结构和算法是计算机科学的基石&#xff0c;是计算机的灵魂 要想成为计算机专业人员&#xff0c;学习和掌握算法是十分必要的。不懂数据结构和算法的人不可能写出效率更高的代码。计算机科学的很多新行业都离不开数据结构和算法作为基石&#xff0c;比如大数据、人工…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023