定時器的實現原理

定時器的實現依賴的是CPU時鐘中斷，時鐘中斷的精度就決定定時器精度的極限。一個時鐘中斷源如何實現多個定時器呢？對於核心，簡單來說就是用特定的資料結構管理眾多的定時器，在時鐘中斷處理中判斷哪些定時器超時，然後執行超時處理動作。而使用者空間程式不直接感知CPU時鐘中斷，通過感知核心的訊號、IO事件、排程，間接依賴時鐘中斷。用軟體來實現動態定時器常用資料結構有：時間輪、最小堆和紅黑樹。下面就是一些知名的實現：

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 使用者空間程式開發中，常用的定期器可以分為兩類：

1. 執行一次的單次定時器 single-short；
2. 迴圈執行的週期定時器 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;
}

在實際專案中，一個常用的做法是新起一個執行緒，專門管理定時器，定時來源使用rtc、select等比較精確的來源，定時器超時後向主要的work執行緒發訊息即可，或者使用timefd介面。

參考：