定时器实现:红黑树、timefd、多级时间轮比较
0 2024-02-23

一、定时器的应用

定时器就像闹钟,可以设定一个时间,然后进入倒计时,到点了提醒我们。同样,应用开发也需要一个定时器,通过设定时间,到点了唤醒程序去执行某项任务。

常见的应用场景有:

1)心跳检测

2)游戏技能冷却

3)倒计时

4) 其他需要延时处理的功能

定时器由两部分组成:容器 + 检测触发机制

1)容器:负责组织大量定时任务

2)检测触发机制:负责检测最近要触发的定时任务

二、定时器的触发方式

对服务端来说,驱动服务端业务逻辑的事件,包括:网络事件、定时事件、以及信号事件。

通常,网络事件和定时事件会进行协同处理。

定时器触发形式通常有两种:

1)利用I/O多路复用系统调用的最后一个参数(超时时间),来触发检测定时器。

2)利用timefd,将定时检测作为I/O多路复用当中的事件进行处理。

2.1 网络事件和定时事件在一个线程中处理

网络事件和定时事件可以进行协同处理,即网络事件和定时事件在一个线程中处理。以epoll多路复用器为例子,通过epoll_wait()的第四个参数timeout作为延时触发定时器,业务逻辑的执行也在同一个线程中。

// 网络事件和定时事件在一个线程中处理,协同处理 while (!quit) { // 最近定时任务的触发时间 = 最近要被触任务的触发时间 - 当前时间 int timeout = get_nearest_timer() - now(); if (timeout < 0) timeout = -1; // 定时任务都过期 // 最近定时任务的触发时间作为 timeout 参数,timetout 定时任务触发 // 1、若没有网络事件,先去处理定时任务 // 2、若收到网络事件,先处理网络事件,再处理定时任务 int nevent = epoll_wait(epfd, ev, nev, timeout); for (int i = 0; i < nevent; i++) { // ... 处理网络事件 } // 轮询处理定时事件 update_timer(); }

1)为什么网络事件和定时事件可以协同处理?

因为reactor是基于事件的网络IO模型,IO的处理是同步的,事件的处理是异步的,而定时任务的处理也是异步的,所以事件的处理和定时任务的处理可以在一个线程中一起处理。

2)如何进行协同处理?

以 io 多路复用作为定时器驱动,“阻塞”地收集就绪事件,timeout 参数用于设置定时。

3)使用场景

  • redis(单reactor)

  • memcached、nginx(多reactor)

4)容器的数据结构

数据结构通常选择红黑树、跳表、最小堆等来实现定时器。

2.2

定时任务在通过一个单独的线程检测,以 sleep(time)作为定时器驱动,time 参数用于设置定时。定时器事件的处理由其他线程或运行队列执行。这种触发方式通常用于处理大量定时任务。

// 网络事件和定时事件在不同线程中处理 void * thread_timer(void *thread_param) { init_timer(); //初始化定时器 while (!quit) { update_timer(); //更新定时器状态 sleep(t); //线程休眠时间t } clear_timer(); return NULL; } pthread_create(&pid, NULL, thread_timer, &thread_param);

该代码创建了一个单独的线程来处理定时事件。在循环中,定时器状态会被更新,并根据需要触发相应的事件。通过调用 sleep 函数,线程可以暂停一段时间,等待下一个定时事件的到来。最后,在线程结束时,定时器资源将被清理和释放。

1)数据结构

使用时间轮数据结构,在一个线程中利用sleep(time)负责检测(time要小于最小时间精度)。时间到达时,通过信号或插入运行队列让其他线程运行业务逻辑。

时间轮只负责检测。这种方式加锁粒度小。

相关视频推荐

C++11,80行代码实现高效灵活的定时器

海量定时器设计,到底应该选择哪种实现?

分布式定时器如何实现?

需要C/C++ Linux服务器架构师学习资料加qun579733396获取(资料包括C/C++,Linux,golang技术,Nginx,ZeroMQ,MySQL,Redis,fastdfs,MongoDB,ZK,流媒体,CDN,P2P,K8S,Docker,TCP/IP,协程,DPDK,ffmpeg等),免费分享

二、定时器的设计

2.1 接口设计

基础接口有

1)初始化定时器

2)添加定时器 —— 添加定时任务

3)删除定时器 —— 删除定时任务

4)更新定时器 —— 到期任务的处理

另外,在协同处理的方案中,即网络事件和定时事件在一个线程中处理的触发方式。此时还需要额外添加接口,来查找最近定时任务的触发时间。

5)查找最近定时任务的触发时间

// 初始化定时器 void init_timer(); // 定时器的添加 Node* add_timer(int expire, callback cb); // 定时器的删除 bool del_timer(Node* node); // 定时器的更新 void update_timer(); // 返回最近定时任务的触发时间,用于协同处理 Node* find_nearest_timer();

2.2 数据结构设计

对定时任务的组织本质是要对定时任务优先级的处理。所谓优先级,就是先触发的定时任务放在最前面。由此产生两类数据结构;

1)按触发时间进行顺序组织,要求数据结构有序,或者相对有序。并且,能快速查找最近触发的定时任务,以及需要考虑怎么处理相同时间触发的定时任务。

  • 红黑树(绝对有序): nginx

  • 跳表(绝对有序):redis(将来引入)

  • 最小堆(相对有序): libevent, libev, go

2)按照执行顺序组织:时间轮

2.2.1 红黑树

红黑树的中序遍历是绝对有序的

2.2.3 最小堆

了解最小堆之前,需要先介绍满二叉树和完全二叉树

1)满二叉树:所有的层节点数都是该层所能容纳节点的最大数量,即满足2^n

2)完全二叉树:若二叉树的深度为h,去掉了h层的节点,就是一个满二叉树;并且h层都集中在最左侧排序。

3)最小堆:

是一颗完全二叉树;

某一个节点的值总是小于等于它的子节点的值;

堆中任意一个节点的子树都是最小堆;

4)最小堆添加节点

为了满足完全二叉树的定义,往二叉树最高层沿着最左侧添加一个节点;然后考虑是否能上升操作;

如果此时添加值为 4 的节点,4 节点是 5 节点的左子树;4 比 5 小,4 和 5 需要交换值;


5)最小堆删除节点

删除操作需要先查找是否包含这个节点;确定存在后,交换最后一个节点,先考虑能否执行下降操作,否则执行上升操作;最后删除最后一个节点;

例如:删除 1 号节点,则需要下沉操作;


删除 9 号节点,则需要上升操作;


2.2.4 时间轮

时间轮是根据时钟运行规律而来的。时间精度为1s,时间范围为12h。定义三个数组分别存 秒、分、时;一个指针一秒钟移动一次,只需关注最近一分钟内要触发的定时任务。


1)添加任务

根据定时任务的间隔时间time,判断将其放在哪一层。

比如当前时间tick是65s,即秒针指向5,分针指向1。

现在要添加间隔时间112s的定时任务。那么根据((65+112)/60)%60 = 2,得到该任务添加在分针层级2的任务队列里。

2)重新映射

我们给出的时间轮的时间精度是秒,也就是只执行秒层的任务。所以秒针每转一圈,需要把下一分钟的任务重新映射到秒层。比如原来分针指向1,秒针转了一圈,下一轮需要把分针2里的任务,重新映射到秒层。

比如,原来添加的任务,过了55s后,当前时间120s。(65+112-120)%60 = 57,也就是刚才添加的任务映射在秒层57的位置。

3)删除节点

时间轮删除节点不方便,一般节点不能删除,因为tick一直在移动,会出现重新映射,节点位置可能改变。那么可以添加一个标记字段cancel,当任务触发时检查这个字段,如果cancel=true则不执行具体任务。

三、利用红黑树实现定时器

3.1 数据结构

在C++中,setmapmultisetmultimap容器使用的是红黑树管理数据。这里选择 set来存储定时器任务。

使用set设计定时器,需要考虑一个关键问题:相同触发事件的定时任务如何处理?

举个例子,任务A到来时的时间 tick = 10,间隔 10s 后触发执行,其触发时间 expire = 20s ;任务B到来时的时间 tick = 15s,间隔 5s 后触发执行,其触发时间也是 expire = 20s。那到点了,应该先执行哪个任务呢?

我们根据插入顺序来决定执行顺序,先插入的先执行,放在红黑树的zuoce。后插入的后执行,放在红黑树的youce。通过 id 属性描述任务到来的先后顺序。

因此定时器结点的结构为

// 定义定时结点的基类,存储唯一标识的元素 struct TimerNodeBase { time_t expire; // 任务触发时间 int64_t id; // 用来描述插入先后顺序 }; // 定时结点,包含定时任务等 struct TimerNode : public TimerNodeBase { // 定时器任务回调函数 // C++ 11特性,使用函数对象。降低拷贝消耗,提高效率 // 使用 using 关键字定义了一个 Callback 类型的别名, // 别名指向一个接受 const TimerNode &node 参数、返回值为 void 的函数对象。 using Callback = std::function; Callback func; // 构造函数,容器内部只构造一次 TimerNode(int64_t id, time_t expire, Callback func) : func(func) { this->expire = expire; this->id = id; } };

在代码中,我们把函数对象剥离出来。这是因为红黑树在改变的时候,为了保持平衡,需要频繁的对比。而对比就涉及到复制、移动等操作。注意到,对比我们只需要对比触发时间和id,不需要函数对象。

函数对象作为类,占用大量的空间,复制和移动代价高,因此,我们拆分成基类和派生类,基类存储标识,用于复制和移动;子类存储函数对象等,在容器内构造后,不再复制和移动。

因此,实现比较函数,采用基类引用比较。

// 按触发时间的先后顺序对结点进行排序 bool operator < (const TimerNodeBase &lhd, const TimerNodeBase &rhd) { // 先比较触发时间 if (lhd.expire < rhd.expire) return true; else if (lhd.expire > rhd.expire) return false; // 触发时间相同,比较插入的先后顺序 // 比较id大小,先插入的结点id小,先执行 return lhd.id < rhd.id; }

3.2 接口实现

3.2.1 初始化定时器

获取当前时间

static time_t GetTick() { auto sc = chrono::time_point_cast(chrono::steady_clock::now()); auto temp = chrono::duration_cast(sc.time_since_epoch()); return temp.count(); }

1)chrono::steady_clock::now():获取系统启动到当前的稳定时间。

2)chrono::time_point_cast(chrono::steady_clock::now()):将当前的稳定时钟时间点转换为毫秒精度的时间点(time_point)。

3)sc.time_since_epoch():计算时间间隔(duration)。

4)chrono::duration_cast(sc.time_since_epoch()):将计算得到的时间间隔转换为毫秒精度(milliseconds)。

5)temp.count():获取转换后的时间间隔的值,即毫秒数。

3.2.2 添加定时器

// 参数:msec 任务触发时间间隔,func 任务执行的回调函数 TimerNodeBase AddTimer(time_t msec, TimerNode::Callback func) { time_t expire = GetTick() + msec; // 相对于insert,emplace 避免了额外的拷贝或移动操作 auto ele = timermap.emplace(GenID(), expire, func); return static_cast(*ele.first); }

上面代码每次插入,需要根据红黑树的插入,重新调整set容器。其实红黑树是有序的,如果是插入一个大于红黑树最右边结点的元素,直接在这个结点,也就是容器末尾插入即可,时间复杂度O(1)。

TimerNodeBase AddTimer(time_t msec, TimerNode::Callback func) { time_t expire = GetTick() + msec; // 如果timermap为空,或者触发时间<=timermap的最后一个(最大的)结点的时间,正常插入 if (timermap.empty() || expire <= timermap.crbegin()->expire){ auto pairs = timermap.emplace(GenID(), expire, std::move(func)); return static_cast(*pairs.first); } // 否则直接插入最后,emplace_hint插入时间复杂度是O(1) auto ele = timermap.emplace_hint(timermap.crbegin().base(), GenID(), expire, std::move(func)); return static_cast(*ele); }

3.2.3 删除定时器

bool DelTimer(TimerNodeBase &node) { // 代替子类结点,避免函数对象复制控制和移动 auto iter = timermap.find(node); // 若存在,则删除该结点 if (iter != timermap.end()) { timermap.erase(iter); return true; } return false; }

3.2.4 更新定时器

bool CheckTimer() { auto iter = timermap.begin(); if (iter != timermap.end() && iter->expire <= GetTick()) { // 定时任务被触发,则执行对应的定时任务 iter->func(*iter); // 删除执行完毕的定时任务 timermap.erase(iter); return true; } return false; }

3.2.5 返回最近定时任务的触发时间

time_t TimeToSleep() { auto iter = timermap.begin(); if (iter == timermap.end()) { return -1; } // 最近任务的触发时间 = 最近任务初始设置的触发时间 - 当前时间 time_t diss = iter->expire - GetTick(); // 最近要触发的任务时间 > 0,继续等待;= 0,立即处理任务 (对应epoll_wait 的 timeout) return diss > 0 ? diss : 0; }

3.3 定时器的驱动

定时器驱动的方式,这里选择 epoll 来实现,通过参数 timeout 设置定时。

while (true) { // 最近任务的触发时间接口:TimeToSlee,作为 timeout 参数 int n = epoll_wait(epfd, ev, 64, timer->TimeToSleep()); for (int i = 0; i < n; i++) { /* 处理网络事件 */ } // 处理定时事件 while(timer->CheckTimer()); }

3.4 代码实现

// g++ timer.cc -o timer -std=c++14 #include #include#include#include#include#includeusing namespace std; // 定时结点的基类,存储唯一标识的元素,轻量级,用于比较 struct TimerNodeBase { time_t expire; // 任务触发时间 int64_t id; // 用来描述插入先后顺序,int64_t,能记录5000多年 }; // 定时结点,包含定时任务等 struct TimerNode : public TimerNodeBase { // 定时器任务回调函数 // 函数对象拷贝代价高,在容器内拷贝构造后不会再去移动 using Callback = std::function; Callback func; // 构造函数,容器内部就地拷贝构造调用一次,此后不会再去调用 TimerNode(int64_t id, time_t expire, Callback func) : func(func) { this->expire = expire; this->id = id; } }; // 根据触发时间对结点进行排序 // 基类引用,多态特性,基类代替timerNode结点,避免拷贝构造子类 bool operator < (const TimerNodeBase &lhd, const TimerNodeBase &rhd) { // 先比较触发时间 if (lhd.expire < rhd.expire) return true; else if (lhd.expire > rhd.expire) return false; // 触发时间相同,比较插入的先后顺序 // 比较id大小,先插入的结点id小,先执行 return lhd.id < rhd.id; } // 定时器类的实现 class Timer { public: // 获取当前时间 static time_t GetTick() { // 获取系统时间戳,系统启动到当前的时间 auto sc = chrono::time_point_cast(chrono::steady_clock::now()); // 获取到时间戳的时间段 auto temp = chrono::duration_cast(sc.time_since_epoch()); return temp.count(); } // 2、添加定时任务 // 参数:msec 任务触发时间间隔,func 任务执行的回调函数 TimerNodeBase AddTimer(time_t msec, TimerNode::Callback func) { time_t expire = GetTick() + msec; // 如果timermap为空,或者触发时间<=timermap的最后一个(最大的)结点的时间,正常插入 if (timermap.empty() || expire <= timermap.crbegin()->expire){ auto pairs = timermap.emplace(GenID(), expire, std::move(func)); return static_cast(*pairs.first); } // 否则直接插入最后,emplace_hint插入时间复杂度是O(1) auto ele = timermap.emplace_hint(timermap.crbegin().base(), GenID(), expire, std::move(func)); return static_cast(*ele); } // 3、删除/取消定时任务 bool DelTimer(TimerNodeBase &node) { // C++14的新特性:只需传递等价 key 比较,无需创建 key 对象比较, // 代替子类结点,避免函数对象复制控制和移动 auto iter = timermap.find(node); // 若存在,则删除该结点 if (iter != timermap.end()) { timermap.erase(iter); return true; } return false; } // 4、检测定时任务是否被触发,触发则执行定时任务 bool CheckTimer() { auto iter = timermap.begin(); if (iter != timermap.end() && iter->expire <= GetTick()) { // 定时任务被触发,则执行对应的定时任务 iter->func(*iter); // 删除执行完毕的定时任务 timermap.erase(iter); return true; } return false; } // 5、返回最近定时任务触发时间,作为timeout的参数 time_t TimeToSleep() { auto iter = timermap.begin(); if (iter == timermap.end()) { return -1; } // 最近任务的触发时间 = 最近任务初始设置的触发时间 - 当前时间 time_t diss = iter->expire - GetTick(); // 最近要触发的任务时间 > 0,继续等待;= 0,立即处理任务 return diss > 0 ? diss : 0; } private: // 产生 id 的方法 static int64_t GenID() { return gid++; } static int64_t gid; // 利用 set 排序快速查找要到期的任务 set> timermap; }; int64_t Timer::gid = 0; int main() { // 定时器驱动 int epfd = epoll_create(1); // 创建定时器 unique_ptrtimer = make_unique(); int i = 0; timer->AddTimer(1000, [&](const TimerNode &node) { cout << Timer::GetTick() << "node id:" << node.id << " revoked times:" << ++i << endl; }); timer->AddTimer(1000, [&](const TimerNode &node) { cout << Timer::GetTick() << "node id:" << node.id << " revoked times:" << ++i << endl; }); timer->AddTimer(3000, [&](const TimerNode &node) { cout << Timer::GetTick() << "node id:" << node.id << " revoked times:" << ++i << endl; }); auto node = timer->AddTimer(2100, [&](const TimerNode &node) { cout << Timer::GetTick() << "node id:" << node.id << " revoked times:" << ++i << endl; }); timer->DelTimer(node); cout << "now time:" << Timer::GetTick() << endl; epoll_event ev[64] = {0}; while (true) { // 最近任务的触发时间接口:TimeToSlee,作为 timeout 参数 int n = epoll_wait(epfd, ev, 64, timer->TimeToSleep()); for (int i = 0; i < n; i++) { /*... 处理网络事件 ...*/ } // 处理定时事件 while(timer->CheckTimer()); } return 0; }

四、利用timefd实现定时器

timefd实现方式跟红黑树很相似,主要区别在于驱动方式。

4.1 流程

1)创建一个 timerfd 文件描述符。可以使用 timerfd_create 函数来创建它。

2)设置定时器参数。构建一个 struct itimerspec 结构体,指定定时器的起始时间和间隔时间。

3)使用 timerfd_settime 函数将定时器参数应用到 timerfd 文件描述符上

4)等待定时器事件触发。可以使用 select、poll、epoll 等函数来等待文件描述符上的可读事件。当定时器事件触发,timerfd 文件描述符会变为可读,你可以在相应的事件处理逻辑中进行处理。

5)当不再需要定时器时,关闭 timerfd 文件描述符。使用 close 函数来关闭它。

4.2 代码实现

#include #include #include // for timespec itimerspec #include // for close #include#include#include#include#includeusing namespace std; struct TimerNodeBase { time_t expire; uint64_t id; }; struct TimerNode : public TimerNodeBase { using Callback = std::function; Callback func; TimerNode(int64_t id, time_t expire, Callback func) : func(func) { this->expire = expire; this->id = id; } }; bool operator < (const TimerNodeBase &lhd, const TimerNodeBase &rhd) { if (lhd.expire < rhd.expire) { return true; } else if (lhd.expire > rhd.expire) { return false; } else return lhd.id < rhd.id; } class Timer { public: static inline time_t GetTick() { return chrono::duration_cast(chrono::steady_clock::now().time_since_epoch()).count(); } TimerNodeBase AddTimer(int msec, TimerNode::Callback func) { time_t expire = GetTick() + msec; if (timeouts.empty() || expire <= timeouts.crbegin()->expire) { auto pairs = timeouts.emplace(GenID(), expire, std::move(func)); return static_cast(*pairs.first); } auto ele = timeouts.emplace_hint(timeouts.crbegin().base(), GenID(), expire, std::move(func)); return static_cast(*ele); } void DelTimer(TimerNodeBase &node) { auto iter = timeouts.find(node); if (iter != timeouts.end()) timeouts.erase(iter); } void HandleTimer(time_t now) { auto iter = timeouts.begin(); while (iter != timeouts.end() && iter->expire <= now) { iter->func(*iter); iter = timeouts.erase(iter); } } public: virtual void UpdateTimerfd(const int fd) { struct timespec abstime; // 获取最小触发时间 auto iter = timeouts.begin(); if (iter != timeouts.end()) { abstime.tv_sec = iter->expire / 1000; abstime.tv_nsec = (iter->expire % 1000) * 1000000; } else { abstime.tv_sec = 0; abstime.tv_nsec = 0; } struct itimerspec its = { .it_interval = {}, .it_value = abstime }; // 通过文件描述符 fd 设置定时器的参数 timerfd_settime(fd, TFD_TIMER_ABSTIME, &its, nullptr); } private: static inline uint64_t GenID() { return gid++; } static uint64_t gid; set> timeouts; }; uint64_t Timer::gid = 0; int main() { int epfd = epoll_create(1); // 创建一个定时器文件描述符 int timerfd = timerfd_create(CLOCK_MONOTONIC, 0); struct epoll_event ev = {.events=EPOLLIN | EPOLLET}; epoll_ctl(epfd, EPOLL_CTL_ADD, timerfd, &ev); unique_ptrtimer = make_unique(); int i = 0; timer->AddTimer(1000, [&](const TimerNode &node) { cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl; }); timer->AddTimer(1000, [&](const TimerNode &node) { cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl; }); timer->AddTimer(3000, [&](const TimerNode &node) { cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl; }); auto node = timer->AddTimer(2100, [&](const TimerNode &node) { cout << Timer::GetTick() << " node id:" << node.id << " revoked times:" << ++i << endl; }); timer->DelTimer(node); cout << "now time:" << Timer::GetTick() << endl; struct epoll_event evs[64] = {0}; while (true) { //设置定时器的参数 timer->UpdateTimerfd(timerfd); int n = epoll_wait(epfd, evs, 64, -1); time_t now = Timer::GetTick(); for (int i = 0; i < n; i++) { // for network event handle } timer->HandleTimer(now); } epoll_ctl(epfd, EPOLL_CTL_DEL, timerfd, &ev); close(timerfd); close(epfd); return 0; }

五、利用多级时间轮实现定时器

时间轮 timewheel 是一个环形结构,使用 hash + list 实现。高层级的每一个格子,存储底层级的一个list。

根据轮子的类型,可以分为主动轮(层级1)和从动轮。

  • 主动轮:当刻度指针指向当前槽的时候,槽内的任务被顺序执行。

  • 从动轮:当对应轮的刻度指针指向当前槽位的时候,槽内的任务链依次向低级轮(序号较高的轮)转移,从动轮没有执行任务权限,只是对任务进行记录与缓存。

以 skynet 为例,skynet 作为单 reactor模型,适用于 cpu 密集型的场景。timer 由 timer 线程管理,当有定时任务时将任务派发给其他线程执行。

5.1 数据结构

如上图,定义了5个链表数组,每个数组里包含多个定时器链表,near 数组大小为256 =2^8,其余数组大小为64 = 2^8,表示的时间范围2^32。

我们只关注主动轮near数组,因为里面的任务是最近要触发执行的。

注意到 2^32 −1刚好是uint32_t的最大值。因此我们只需要一个time指针,就可以根据位运算,得到任务触发时间在各层级的位置。当 time 溢出时,32位无符号循环,再次从0开始计数。


举个例子,如果触发时间是562568,对应二进制是100010 010101 10001000。near位的十进制是136,t[0]位的十进制是41,t[1]位的十进制是67。

1)定时器的结构

typedef struct timer { link_list_t near[TIME_NEAR]; // 最低级的时间轮,主动轮 link_list_t t[4][TIME_LEVEL]; // 其他层级的时间轮,从动轮 struct spinlock lock; // 自旋锁,O(1) uint32_t time; // tick 指针,当前时间片 uint64_t current; // 从系统开始时刻到现在的时长,timer运行时间,时间精度10ms uint64_t current_point; // 系统启动时长,时间精度10ms }s_timer_t;

2)任务结点的设计

任务结点使用链表存储,链表中存储同一时间触发的任务结点

struct timer_node { struct timer_node *next; // 指向的下一个任务 uint32_t expire; // 任务触发时间 handler_pt callback; // 任务回调函数 uint8_t cancel; // 删除任务的标记,取消任务的执行 int id; // 执行该任务的线程 id };

5.2 接口设计

5.2.1 初始化定时器

// 初始化定时器 void init_timer(void) { TI = timer_create_timer(); // 创建定时器 TI->current_point = gettime(); // 获取系统当前运行时间 } // 创建定时器 s_timer_t* timer_create_timer() { s_timer_t *r = (s_timer_t *)malloc(sizeof(s_timer_t)); memset(r, 0, sizeof(*r)); int i, j; // 创建主动轮,最低级时间轮 for (i = 0; i < TIME_NEAR; ++i) { //清除指定链表,并返回指向链表的第一个节点的指针 link_clear(&r->near[i]); } // 创建从动轮,高层级时间轮 for (i = 0; i < 4; ++i) { for (j = 0;j < TIME_LEVEL; ++j) { link_clear(&r->t[i][j]); } } // 初始化自旋锁 spinlock_init(&r->lock); r->current = 0; return r; } // 获得从系统启动开始计时的时间,不受系统时间被用户改变的影响,精确到1/100秒 uint64_t gettime() { uint64_t t; #if !defined(__APPLE__) || defined(AVAILABLE_MAC_OS_X_VERSION_10_12_AND_LATER) struct timespec ti; clock_gettime(CLOCK_MONOTONIC, &ti); // CLOCK_MONOTONIC t = (uint64_t)ti.tv_sec * 1000; t += ti.tv_nsec / 1000000; #else struct timeval tv; gettimeofday(&tv, NULL); t = (uint64_t)tv.tv_sec * 100; t += tv.tv_usec / 10000; #endif return t; }

5.2.2 添加定时器

32位无符号整数time记录时间片分别对应数组near[256]t[4][64],每次添加节点时,如果expire - time < 256则将节点添加到near数组对应元素的链表中,否则从高位往低位依次比较expire的第i个6位二进制的值n与time的第i个6位二进制的值m,哪个不相等则将节点添加到数组t[4-i][n]对应的元素链表中)

具体来说:

  • 首先检查节点的expiretime的高24位是否相等,相等则将该节点添加到expire低8位值对应数组near的元素的链表中,不相等则进行下一步。

  • 检查expire与time的高18位是否相等,相等则将该节点添加到expire低第9位到第14位对应的6位二进制值对应数组t[0]的元素的链表中,如果不相等则进行下一步。

  • 检查expire与time的高12位是否相等,相等则将该节点添加到expire低第15位到第20位对应的6位二进制值对应数组t[1]的元素的链表中,如果不相等则进行下一步。

  • 检查expiretime的高6位是否相等,相等则将该节点添加到expire低第21位到第26位对应的6位二进制值对应数组t[2]的元素的链表中,如果不相等则进行下一步。

  • 将该节点添加到expire低第27位到第32位对应的6位二进制值对应数组t[3]的元素的链表中

先看一下位操作:

1 << n:表示将二进制数 1 向左移动 n 位,即 2^n。(time>>TIME_NEAR_SHIFT) & TIME_LEVEL_MASK):表示time右移8位,然后与TIME_LEVEL_MASK相与,得到的是t[0]处的位置。

// 添加任务结点到定时器中 // 根据 msec 判断结点应该放入时间轮的层级 void add_node(s_timer_t *T, timer_node_t *node) { uint32_t time = node->expire; // 过期时间 uint32_t current_time=T->time; // 当前时间 uint32_t msec = time - current_time; // 剩余时间 //根据 expire-time 的差值将结点放入相应的层级 //[0, 2^8) if (msec < TIME_NEAR) { link(&T->near[time&TIME_NEAR_MASK],node); } //[2^8, 2^14) else if (msec < (1 << (TIME_NEAR_SHIFT+TIME_LEVEL_SHIFT))) { link(&T->t[0][((time>>TIME_NEAR_SHIFT) & TIME_LEVEL_MASK)],node); } //[2^14, 2^20) else if (msec < (1 << (TIME_NEAR_SHIFT+2*TIME_LEVEL_SHIFT))) { link(&T->t[1][((time>>(TIME_NEAR_SHIFT + TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node); } //[2^20, 2^26) else if (msec < (1 << (TIME_NEAR_SHIFT+3*TIME_LEVEL_SHIFT))) { link(&T->t[2][((time>>(TIME_NEAR_SHIFT + 2*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node); } //[2^26, 2^32) else { link(&T->t[3][((time>>(TIME_NEAR_SHIFT + 3*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node); } } // 添加定时任务 timer_node_t* add_timer(int time, handler_pt func, int threadid) { timer_node_t *node = (timer_node_t *)malloc(sizeof(*node)); spinlock_lock(&TI->lock); // 设置定时任务结点的属性 node->expire = time + TI->time; // 添加触发时间 = 触发时间间隔 + 当前时间 node->callback = func; // 添加任务回调函数 node->id = threadid; // 添加执行该任务的线程id // 判断是否需要立即执行任务 if (time <= 0) { spinlock_unlock(&TI->lock); node->callback(node); free(node); return NULL; } // 添加任务结点到定时器中 add_node(TI, node); spinlock_unlock(&TI->lock); return node; }

5.2.3 删除定时器

由于结点位置可能发生变化(重新映射),不能找到任务结点的位置,无法删除。

在结点中添加一个 cancel 字段,任务触发碰到该标记则不执行任务,之后统一释放空间。

void del_timer(timer_node_t *node) { node->cancel = 1; }

5.2.4 更新定时器

主要包括对到期任务的处理和对从动轮任务(time高24位对应的链表)的重新映射。

void timer_update(s_timer_t *T) { spinlock_lock(&T->lock); // shift time first, and then dispatch timer message timer_execute(T); //检查当前的时间片的低8位对应的数组元素的链表是否为空,不为空则取出 // shift time first, and then dispatch timer message timer_shift(T); //检查当前的时间片的低8位对应的数组元素的链表是否为空,不为空则取出 // 若发生重新映射,若time的指向有任务,则需要执行 timer_execute(T); spinlock_unlock(&T->lock); }

5.2.5 到期任务的处理

以当前 tick 值的低8位作为索引,取出 near 数组中对应的 list。list 里面包含了所有在该 tick 到期的定时器任务

//检查当前的时间片的低8位对应的数组元素的链表是否为空,不为空则取出 void timer_execute(s_timer_t *T) { // 取出time低8位对应的索引值 int idx = T->time & TIME_NEAR_MASK; // 如果低8位值对应的near数组元素有链表,则取出 while (T->near[idx].head.next) { // 取出对应的定时器list timer_node_t *current = link_clear(&T->near[idx]); spinlock_unlock(&T->lock); // 将链表各结点的任务派发出去 dispatch_list(current); spinlock_lock(&T->lock); } } // 任务派发 void dispatch_list(timer_node_t *current) { do { timer_node_t *temp = current; current=current->next; // cancel 标记为0,执行任务回调函数;否则,不执行任务回调 if (temp->cancel == 0) temp->callback(temp); free(temp); } while (current); }

5.2.6 刷新时间片,重新映射

只有主动轮的结点要执行,从动轮只是存储结点,主动轮结点执行完后,需要从动轮补充。

具体操作是

  • 检查time是否溢出,如果溢出则将t[3][0]这个链表取出并依次将该链表中的节点添加(即实现该链表的移动操作),如果time未溢出,则进行下一步。

  • 检查time低8位是否溢出产生进位,没有则结束,有则检查time的低第9位到第14位是否产生溢出,没有则将time的低第9位到第14位对应的值对应数组t[0]中的链表取出,并依次将该链表中的节点添加(即实现该链表的移动操作),如果有溢出,则进行下一步。

  • 检查time低14位是否溢出产生进位,没有则结束,有则检查time的低第15位到第20位是否产生溢出,没有则将time的低第15位到第20位对应的值对应数组t[1]中的链表取出,并依次将该链表中的节点添加(即实现该链表的移动操作),如果有溢出,则进行下一步。

  • 检查time低20位是否溢出产生进位,没有则结束,有则检查time的低第21位到第26位是否产生溢出,没有则将time的低第21位到第26位对应的值对应数组t[2]中的链表取出,并依次将该链表中的节点添加(即实现该链表的移动操作),如果有溢出,则进行下一步。

  • 检查time低26位是否溢出产生进位,没有则结束,有则检查time的低第27位到第32位是否产生溢出,没有则将time的低第27位到第32位对应的值对应数组t[3]中的链表取出,并依次将该链表中的节点添加(即实现该链表的移动操作)。

所谓溢出,就是移动一轮了,就跟秒针转动一圈,重新计数一样。重新计数,分针也需要移动。也就是主动轮移动一轮,现在要移动下一轮,此时需要把从动轮的任务映射过去。

比如。低八位11111111对应是255,加1就是1 00000000对应是256,低八位都是0溢出。

// 重新映射,判断是否需要重新映射 // 时间片time自加1,将高24位对应的4个6位的数组中的各个元素的链表往低位移 void timer_shift(s_timer_t *T) { int mask = TIME_NEAR; // 时间片+1 uint32_t ct = ++T->time; // 时间片溢出,无符号整数,循环,time重置0 if (ct == 0) { // 将对应的t[3][0]链表取出,重新移动到定时器中 move_list(T, 3, 0); } else { // ct右移8位,进入到从动轮 uint32_t time = ct >> TIME_NEAR_SHIFT; // 第 i 层时间轮 int i = 0; // 判断是否需要重新映射? // 即循环判断当前层级对应的数位是否全0,即溢出产生进位 while ((ct & (mask-1))==0) { // 取当前层级的索引值 int idx = time & TIME_LEVEL_MASK; // idx=0 说明当前层级溢出,继续循环判断直至当前层级不溢出 if (idx != 0) { // 将对应的t[i][idx]链表取出,依次移动到定时器中 move_list(T, i, idx); break; } mask <<= TIME_LEVEL_SHIFT; // mask 右移 time >>= TIME_LEVEL_SHIFT; // time 左移 ++i; // 时间轮层级增加 } } }

5.3 定时器的驱动

// timer 线程中,每过1/4时间精度,即2.5ms,执行一次定时器的检测 while (!ctx.quit) { expire_timer(); usleep(2500); }

刷新定时器,每过1/4时间精度执行一次

// 原因是 dispatch 分发任务花费时间,影响精度 void expire_timer(void) { // 获取当前系统运行时间,不受用户的影响 uint64_t cp = gettime(); // 当前系统启动时间与定时器记录的系统启动时间不相等 if (cp != TI->current_point) { // 获取上述两者的差值 uint32_t diff = (uint32_t)(cp - TI->current_point); // 更新定时器记录的系统运行时间 TI->current_point = cp; // 更新timer的运行时间 TI->current += diff; // 更新定时器的时间(time的值),并执行对应的过期任务 int i; for (i=0; i

5.4 代码实现

1)timer_wheel.h

#ifndef _MARK_TIMEWHEEL_ #define _MARK_TIMEWHEEL_ #include #define TIME_NEAR_SHIFT 8 #define TIME_NEAR (1 << TIME_NEAR_SHIFT) // 1 << 8 表示将二进制数 1 向左移动 8 位,即 2^8 = 256 #define TIME_LEVEL_SHIFT 6 #define TIME_LEVEL (1 << TIME_LEVEL_SHIFT) // 1 << 6 表示将二进制数 1 向左移动 6 位,即 2^6 = 64 #define TIME_NEAR_MASK (TIME_NEAR-1) // 255 #define TIME_LEVEL_MASK (TIME_LEVEL-1) // 63 typedef struct timer_node timer_node_t; typedef void (*handler_pt) (struct timer_node *node); // 任务结点 struct timer_node { struct timer_node *next; // 相同过期时间的待执行的下一个任务 uint32_t expire; // 任务过期时间 handler_pt callback; // 任务回调函数 uint8_t cancel; // 删除任务,遇到该标记则取消任务的执行 int id; // 此时携带参数 }; timer_node_t* add_timer(int time, handler_pt func, int threadid); void expire_timer(void); void del_timer(timer_node_t* node); void init_timer(void); void clear_timer(); #endif

2)timer_wheel.c

// timer_wheel.c #include "spinlock.h" #include "timewheel.h" #include #include #include #if defined(__APPLE__) #include #include #include #include #else #include #endif typedef struct link_list { timer_node_t head; timer_node_t *tail; }link_list_t; // 定时器的数据结构 typedef struct timer { link_list_t near[TIME_NEAR]; // 最低级的时间轮,主动轮 link_list_t t[4][TIME_LEVEL]; // 其他层级的时间轮,从动轮 struct spinlock lock; // 自旋锁,O(1) uint32_t time; // tick 指针,当前时间片 uint64_t current; // timer运行时间,精度10ms uint64_t current_point; // 系统运行时间,精度10ms }s_timer_t; static s_timer_t * TI = NULL; // 清空链表 // 并返回指向链表的第一个结点的指针 timer_node_t* link_clear(link_list_t *list) { // 指向头指针的下一个位置 timer_node_t * ret = list->head.next; // 头结点断链 list->head.next = 0; // 尾指针指向头结点 list->tail = &(list->head); return ret; } // 尾插法 void link(link_list_t *list, timer_node_t *node) { list->tail->next = node; list->tail = node; node->next=0; } // 添加任务结点到定时器中 // 根据 time 判断结点应该放入时间轮的层级 void add_node(s_timer_t *T, timer_node_t *node) { uint32_t time = node->expire; // 过期时间 uint32_t current_time=T->time; // 当前时间 uint32_t msec = time - current_time; // 剩余时间 //根据 expire-time 的差值将结点放入相应的层级 //[0, 2^8) if (msec < TIME_NEAR) { link(&T->near[time&TIME_NEAR_MASK],node); } //[2^8, 2^14) else if (msec < (1 << (TIME_NEAR_SHIFT+TIME_LEVEL_SHIFT))) { link(&T->t[0][((time>>TIME_NEAR_SHIFT) & TIME_LEVEL_MASK)],node); } //[2^14, 2^20) else if (msec < (1 << (TIME_NEAR_SHIFT+2*TIME_LEVEL_SHIFT))) { link(&T->t[1][((time>>(TIME_NEAR_SHIFT + TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node); } //[2^20, 2^26) else if (msec < (1 << (TIME_NEAR_SHIFT+3*TIME_LEVEL_SHIFT))) { link(&T->t[2][((time>>(TIME_NEAR_SHIFT + 2*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node); } //[2^26, 2^32) else { link(&T->t[3][((time>>(TIME_NEAR_SHIFT + 3*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node); } } // 添加定时任务 timer_node_t* add_timer(int time, handler_pt func, int threadid) { timer_node_t *node = (timer_node_t *)malloc(sizeof(*node)); spinlock_lock(&TI->lock); // 设置定时任务结点的属性 node->expire = time + TI->time; // 添加任务触发时间 expire = time + tick node->callback = func; // 添加任务回调函数 node->id = threadid; // 添加执行该任务的线程id // 判断是否需要立即执行任务 if (time <= 0) { spinlock_unlock(&TI->lock); node->callback(node); free(node); return NULL; } // 添加任务结点到定时器中 add_node(TI, node); spinlock_unlock(&TI->lock); return node; } // 移动链表 // 第level层第idx个位置的链表结点重新添加到定时器T中 void move_list(s_timer_t *T, int level, int idx) { timer_node_t *current = link_clear(&T->t[level][idx]); while (current) { timer_node_t *temp = current->next; add_node(T, current); current = temp; } } // 重新映射,判断是否需要重新映射 // 时间片time自加1,将高24位对应的4个6位的数组中的各个元素的链表往低位移 void timer_shift(s_timer_t *T) { int mask = TIME_NEAR; // 时间片+1 uint32_t ct = ++T->time; // 时间片溢出,无符号整数,循环,time重置0 if (ct == 0) { // 将对应的t[3][0]链表取出,重新移动到定时器中 move_list(T, 3, 0); } else { // ct右移8位,进入到从动轮 uint32_t time = ct >> TIME_NEAR_SHIFT; // 第 i 层时间轮 int i = 0; // 判断是否需要重新映射? // 即循环判断当前层级对应的数位是否全0,即溢出产生进位 while ((ct & (mask-1))==0) { // 取当前层级的索引值 int idx = time & TIME_LEVEL_MASK; // idx=0 说明当前层级溢出,继续循环判断直至当前层级不溢出 if (idx != 0) { // 将对应的t[i][idx]链表取出,依次移动到定时器中 move_list(T, i, idx); break; } mask <<= TIME_LEVEL_SHIFT; // mask 右移 time >>= TIME_LEVEL_SHIFT; // time 左移 ++i; // 时间轮层级增加 } } } // 任务派发给其他线程执行 void dispatch_list(timer_node_t *current) { do { timer_node_t *temp = current; current=current->next; // cancel 标记为0,执行任务回调函数 if (temp->cancel == 0) temp->callback(temp); free(temp); } while (current); } // 执行任务 // 以time的低8位对应的near数组的索引,取出该位置对应的list void timer_execute(s_timer_t *T) { // 取出time低8位对应的值 int idx = T->time & TIME_NEAR_MASK; // 如果低8位值对应的near数组元素有链表,则取出 while (T->near[idx].head.next) { // 取出对应的定时器list timer_node_t *current = link_clear(&T->near[idx]); spinlock_unlock(&T->lock); // 将链表各结点的任务派发出去 dispatch_list(current); spinlock_lock(&T->lock); } } // 定时器更新 void timer_update(s_timer_t *T) { spinlock_lock(&T->lock); // 执行任务 timer_execute(T); /// time+1,并判断是否进行重新映射 timer_shift(T); // 若发生重新映射,若time的指向有任务,则需要执行 timer_execute(T); spinlock_unlock(&T->lock); } // 删除定时器任务 void del_timer(timer_node_t *node) { node->cancel = 1; } // 创建定时器 s_timer_t * timer_create_timer() { s_timer_t *r = (s_timer_t *)malloc(sizeof(s_timer_t)); memset(r, 0, sizeof(*r)); int i, j; // 创建主动轮,最低级时间轮 for (i = 0; i < TIME_NEAR; ++i) { link_clear(&r->near[i]); } // 创建从动轮,高层级时间轮 for (i = 0; i < 4; ++i) { for (j = 0;j < TIME_LEVEL; ++j) { link_clear(&r->t[i][j]); } } // 初始化自旋锁 spinlock_init(&r->lock); r->current = 0; return r; } // 获取当前时间,时间精度10ms uint64_t gettime() { uint64_t t; #if !defined(__APPLE__) || defined(AVAILABLE_MAC_OS_X_VERSION_10_12_AND_LATER) struct timespec ti; clock_gettime(CLOCK_MONOTONIC, &ti);// CLOCK_MONOTONIC,从系统启动这一刻起开始计时,不受系统时间被用户改变的影响 t = (uint64_t)ti.tv_sec * 1000; t += ti.tv_nsec / 1000000; #else struct timeval tv; gettimeofday(&tv, NULL); t = (uint64_t)tv.tv_sec * 100; t += tv.tv_usec / 10000; #endif return t; } // 检测定时器,时间精度10ms,每过1/4时间精度2.5ms执行1次 // 原因是dispatch分发任务花费时间,影响精度 void expire_timer(void) { // 获取当前系统运行时间,不受系统时间被用户的影响 uint64_t cp = gettime(); // 当前系统启动时间与定时器记录的系统启动时间不相等 if (cp != TI->current_point) { // 获取上述两者的差值 uint32_t diff = (uint32_t)(cp - TI->current_point); // 更新定时器记录的系统运行时间 TI->current_point = cp; // 更新timer的运行时间 TI->current += diff; // 更新定时器的时间(time的值),并执行对应的过期任务 int i; for (i=0; icurrent_point = gettime(); // 获取当前时间 } void clear_timer() { int i,j; for (i=0;inear[i]; timer_node_t* current = list->head.next; while(current) { timer_node_t * temp = current; current = current->next; free(temp); } link_clear(&TI->near[i]); } for (i=0;i<4;i++) { for (j=0;jt[i][j]; timer_node_t* current = list->head.next; while (current) { timer_node_t * temp = current; current = current->next; free(temp); } link_clear(&TI->t[i][j]); } } }

3)tw-timer.c

// tw-timer.c // gcc tw-timer.c timewheel.c -o tw -I./ -lpthread #include #include #include #include #include #include "timewheel.h" struct context { int quit; int thread; }; struct thread_param { struct context *ctx; int id; }; static struct context ctx = {0}; void do_timer(timer_node_t *node) { printf("do_timer expired:%d - thread-id:%d\n", node->expire, node->id); add_timer(100, do_timer, node->id); } void do_clock(timer_node_t *node) { static int time; time ++; printf("---time = %d ---\n", time); add_timer(100, do_clock, node->id); } void* thread_worker(void *p) { struct thread_param *tp = p; int id = tp->id; struct context *ctx = tp->ctx; int expire = rand() % 200; add_timer(expire, do_timer, id); while (!ctx->quit) { usleep(1000); } printf("thread_worker:%d exit!\n", id); return NULL; } void do_quit(timer_node_t * node) { ctx.quit = 1; } int main() { srand(time(NULL)); ctx.thread = 2; pthread_t pid[ctx.thread]; init_timer(); add_timer(6000, do_quit, 100); add_timer(0, do_clock, 100); struct thread_param task_thread_p[ctx.thread]; int i; for (i = 0; i < ctx.thread; i++) { task_thread_p[i].id = i; task_thread_p[i].ctx = &ctx; if (pthread_create(&pid[i], NULL, thread_worker, &task_thread_p[i])) { fprintf(stderr, "create thread failed\n"); exit(1); } } while (!ctx.quit) { expire_timer(); usleep(2500); // 2.5ms } clear_timer(); for (i = 0; i < ctx.thread; i++) { pthread_join(pid[i], NULL); } printf("all thread is closed\n"); return 0; }

5.5 总结

skynet 是怎么样运转定时器的?

skynet的 timer 线程会不断触发 expire_timer函数,在该函数中会不断执行timer_execute对 near 中的时器执行超时操作。执行完毕后,调用 timer_shift 从t[0] ~ t[3]中选择合适的定时器节点加入到 near 中,这过程就相当于提高了定时器节点的紧急程度(因为随着时间的流逝,定时器节点的紧急程度会越来越向near逼近)。



声明: 本文转载自其它媒体或授权刊载,目的在于信息传递,并不代表本站赞同其观点和对其真实性负责,如有新闻稿件和图片作品的内容、版权以及其它问题的,请联系我们及时删除。(联系我们,邮箱:evan.li@aspencore.com )
0
评论
  • 相关技术文库
  • 工业
  • 安防
  • 航空
  • CAN
下载排行榜
更多
评测报告
更多
EE直播间
更多
广告