定時器的實現原理
定時器的實現依賴的是CPU時鐘中斷，時鐘中斷的精度就決定定時器精度的極限。一個時鐘中斷源如何實現多個定時器呢？對於內核，簡單來說就是用特定的數據結構管理眾多的定時器，在時鐘中斷處理中判斷哪些定時器超時，然後執行超時處理動作。而用戶空間程序不直接感知CPU時鐘中斷，通過感知內核的信號、IO事件、調度，間接依賴時鐘中斷。用軟件來實現動態定時器常用數據結構有：時間輪、最小堆和紅黑樹。下面就是一些知名的實現：
Linux內核的 Hierarchy 時間輪算法
Asio C++ Library最小堆定時器實現
nginx 使用紅黑樹結構管理定時器事件
Linux內核定時器相關(Linux v4.9.7, x86體系架構)的一些相關代碼：
內核啟動註冊時鐘中斷
// @file: arch/x86/kernel/time.c - Linux 4.9.7
// 內核init階段註冊時鐘中斷處理函數
static struct irqaction irq0 = {
.handler = timer_interrupt,
.flags = IRQF_NOBALANCING | IRQF_IRQPOLL | IRQF_TIMER,
.name = "timer"
};
void __init setup_default_timer_irq(void)
{
if (!nr_legacy_irqs())
return;
setup_irq(0, &irq0);
}
// Default timer interrupt handler for PIT/HPET
static irqreturn_t timer_interrupt(int irq, void *dev_id)
{
// 調用體系架構無關的時鐘處理流程
global_clock_event->event_handler(global_clock_event);
return IRQ_HANDLED;
}
內核時鐘中斷處理流程
// @file: kernel/time/timer.c - Linux 4.9.7
/*
* Called from the timer interrupt handler to charge one tick to the current
* process. user_tick is 1 if the tick is user time, 0 for system.
*/
void update_process_times(int user_tick)
{
struct task_struct *p = current;
/* Note: this timer irq context must be accounted for as well. */
account_process_tick(p, user_tick);
run_local_timers();
rcu_check_callbacks(user_tick);
#ifdef CONFIG_IRQ_WORK
if (in_irq())
irq_work_tick();
#endif
scheduler_tick();
run_posix_cpu_timers(p);
}
/*
* Called by the local, per-CPU timer interrupt on SMP.
*/
void run_local_timers(void)
{
struct timer_base *base = this_cpu_ptr(&timer_bases[BASE_STD]);
hrtimer_run_queues();
/* Raise the softirq only if required. */
if (time_before(jiffies, base->clk)) {
if (!IS_ENABLED(CONFIG_NO_HZ_COMMON) || !base->nohz_active)
return;
/* CPU is awake, so check the deferrable base. */
base++;
if (time_before(jiffies, base->clk))
return;
}
raise_softirq(TIMER_SOFTIRQ); // 標記一個軟中斷去處理所有到期的定時器
}
內核定時器時間輪算法
單層時間輪算法的原理比較簡單：用一個數組表示時間輪，每個時鐘週期，時間輪 current 往後走一個格，並處理掛在這個格子的定時器鏈表，如果超時則進行超時動作處理，然後刪除定時器，沒有則剩餘輪數減一。原理如圖：
Linux 內核則採用的是 Hierarchy 時間輪算法，Hierarchy 時間輪將單一的 bucket 數組分成了幾個不同的數組，每個數組表示不同的時間精度，Linux 內核中用 jiffies 記錄時間，jiffies記錄了系統啟動以來經過了多少tick。下面是Linux 4.9的一些代碼：
// @file: kernel/time/timer.c - Linux 4.9.7
/*
* The timer wheel has LVL_DEPTH array levels. Each level provides an array of
* LVL_SIZE buckets. Each level is driven by its own clock and therefor each
* level has a different granularity.
*/
/* Size of each clock level */
#define LVL_BITS 6
#define LVL_SIZE (1UL << LVL_BITS)
/* Level depth */
#if HZ > 100
# define LVL_DEPTH 9
# else
# define LVL_DEPTH 8
#endif
#define WHEEL_SIZE (LVL_SIZE * LVL_DEPTH)
struct timer_base {
spinlock_t lock;
struct timer_list *running_timer;
unsigned long clk;
unsigned long next_expiry;
unsigned int cpu;
bool migration_enabled;
bool nohz_active;
bool is_idle;
DECLARE_BITMAP(pending_map, WHEEL_SIZE);
struct hlist_head vectors[WHEEL_SIZE];
} ____cacheline_aligned;
Hierarchy 時間輪的原理大致如下，下面是一個時分秒的Hierarchy時間輪，不同於Linux內核的實現，但原理類似。對於時分秒三級時間輪，每個時間輪都維護一個cursor，新建一個timer時，要掛在合適的格子，剩餘輪數以及時間都要記錄，到期判斷超時並調整位置。原理圖大致如下:
定時器的使用方法
在Linux 用戶空間程序開發中，常用的定期器可以分為兩類：
執行一次的單次定時器 single-short；
循環執行的週期定時器 Repeating Timer；
其中，Repeating Timer 可以通過在Single-Shot Timer 終止之後，重新再註冊到定時器系統裡來實現。當一個進程需要使用大量定時器時，同樣利用時間輪、最小堆或紅黑樹等結構來管理定時器。而時鐘週期來源則需要藉助系統調用，最終還是從時鐘中斷。Linux用戶空間程序的定時器可用下面方法來實現：
通過alarm()或setitimer()系統調用，非阻塞異步，配合SIGALRM信號處理；
通過select()或nanosleep()系統調用，阻塞調用，往往需要新建一個線程；
通過timefd()調用，基於文件描述符，可以被用於 select/poll 的應用場景；
通過RTC機制, 利用系統硬件提供的Real Time Clock機制, 計時非常精確；
上面方法沒提sleep()，因為Linux中並沒有系統調用sleep()，sleep()是在庫函數中實現，是通過調用alarm()來設定報警時間，調用sigsuspend()將進程掛起在信號SIGALARM上，而且sleep()也只能精確到秒級上，精度不行。當使用阻塞調用作為定時週期來源時，可以單獨啟一個線程用來管理所有定時器，當定時器超時的時候，向業務線程發送定時器消息即可。
一個基於時間輪的定時器簡單實現
#include <stdio.h>
#include <signal.h>
#include <stdlib.h>
#include <unistd.h>
#define TIME_WHEEL_SIZE 8
typedef void (*func)(int data);
struct timer_node {
struct timer_node *next;
int rotation;
func proc;
int data;
};
struct timer_wheel {
struct timer_node *slot[TIME_WHEEL_SIZE];
int current;
};
struct timer_wheel timer = {{0}, 0};
void tick(int signo)
{
// 使用二級指針刪進行單鏈表的刪除
struct timer_node **cur = &timer.slot[timer.current];
while (*cur) {
struct timer_node *curr = *cur;
if (curr->rotation > 0) {
curr->rotation--;
cur = &curr->next;
} else {
curr->proc(curr->data);
*cur = curr->next;
free(curr);
}
}
timer.current = (timer.current + 1) % TIME_WHEEL_SIZE;
alarm(1);
}
void add_timer(int len, func action)
{
int pos = (len + timer.current) % TIME_WHEEL_SIZE;
struct timer_node *node = malloc(sizeof(struct timer_node));
// 插入到對應格子的鏈表頭部即可, O(1)複雜度
node->next = timer.slot[pos];
timer.slot[pos] = node;
node->rotation = len / TIME_WHEEL_SIZE;
node->data = 0;
node->proc = action;
}
// test case1: 1s循環定時器
int g_sec = 0;
void do_time1(int data)
{
printf("timer %s, %d\\n", __FUNCTION__, g_sec++);
add_timer(1, do_time1);
}
// test case2: 2s單次定時器
void do_time2(int data)
{
printf("timer %s\\n", __FUNCTION__);
}
// test case3: 9s循環定時器
void do_time9(int data)
{
printf("timer %s\\n", __FUNCTION__);
add_timer(9, do_time9);
}
int main()
{
signal(SIGALRM, tick);
alarm(1); // 1s的週期心跳
// test
add_timer(1, do_time1);
add_timer(2, do_time2);
add_timer(9, do_time9);
while(1) pause();
return 0;
}/<unistd.h>/<stdlib.h>/<signal.h>/<stdio.h>
在實際項目中，一個常用的做法是新起一個線程，專門管理定時器，定時來源使用rtc、select等比較精確的來源，定時器超時後向主要的work線程發消息即可，或者使用timefd接口。
最後，如果你想學C/C++可以私信小編“01”獲取素材資料以及開發工具和聽課權限哦！
閱讀更多 編程小新 的文章
