图解｜低精度定时器原理

用户7686797

发布于 2023-09-02 17:23:07

1990

发布于 2023-09-02 17:23:07

Linux 内核通常会使用 定时器 来做一些延时的操作，比如常用的 sleep() 系统调用就是使用定时器来实现的。

在 Linux 内核中，有两种类型的定时器：高精度定时器 与 低精度定时器。低精度定时器基于硬件的时钟中断实现的，其定时周期的粒度为 1/HZ ms。例如，内核 HZ 为 1000（也就是说 1 秒能够产生 1000 次时钟中断），那么低精度定时器的最小定时间隔为1ms；而高精度定时器可以实现纳秒级别的定时（实际的定时周期粒度与 CPU 的主频有关）。

可能有读者会问，既然有了高精度定时器，那么低精度定时器是否可以废弃呢？答案是否定的，主要原因是使用高精度定时器的成本比低精度定时器要高。所以，如果对时间精度不是非常敏感的话，使用低精度定时器更合适。

本文主要介绍 Linux 内核中的低精度定时器的原理与实现。

时间轮

低精度定时器是基于时钟中断实现的，而时钟中断的触发频率是可以配置的，Linux 内核一般设置为每秒触发 1000 次，也就是说每隔 1 毫秒就会触发一次时钟中断。

一般来说，内核中可能会存在成千上万个定时器，那么内核如何能够快速找到将要到期的定时器呢？

在学习数据结构课程时，我们知道用于快速查找有序数据的数据结构有如何几种：

平衡二叉树
最大堆/最小堆
跳跃表
...

由于这些数据结构的时间复杂度都是 log(n)，对性能要求非常高的内核来说是不能接受的，所以内核使用了一种性能更高的数据结构：时间轮。

时间轮能够保证在时间复杂度为 log(1) 的情况下找到将要到期的定时器，下面我们将会介绍时间轮的原理。

时间轮的基本思想是通过数组来保存定时器，而数组的索引就是定时器的过期时间。如下图所示：

如上图所示的数组中，索引为 1 的槽位存放的是 1 毫秒后超时的定时器列表，索引为 2 的槽位存放的是 2 毫秒后超时的定时器列表，如此类推...

此时，我们可以使用一个指针来指向超时的定时器列表，如下图所示：

每当时钟中断被触发一次，指针向下移动一位，这样就能在时间复杂度为 log(1) 的情况下获取到期的定时器。

这样虽然能够在时间复杂度为 log(1) 的情况下找到到期的定时器，但如果超时时间非常大的话，那么用于存储定时器的数组也会非常巨大，如：定时器的超时时间为 4294967295 毫秒，那么将需要一个大小为 4294967296 的数组。

1. 存储定时器

为了解决这个问题，内核使用 层级 的概念来减少数组占用的内存空间。其原理如下图所示：

由于超时时间是一个整数（32 位整型），所以可以将其划分为 5 个等级，每个级别使用一个数组来表示。例如第一级数组占用 8 个位，所以其大小为 256。而其他级别的数组都占用 6 个位，所以大小都为 64。

一个定时器被存放到哪个数组，是由其超时时间决定的，算法也非常简单：如果第五级的值不为零，那么将会被存放到第五级数组中，而存放的位置以第五级的值作为索引。

例如，一个定时器的超时时间其第五级的值为 32，那么此定时器将会被存放到第五级数组的第 32 个槽位上。如下图所示：

如果第五级的值为零，而第四级的值不为零，那么定时器将会被存放在第四级数组中，存放的位置以第四级的值作为索引，如此类推。

从上面的分析可以看出，定时器的超时时间越小，其存放的数组层级就越小。所以：

第一级数组存放的是超时时间范围为 [0, 256) 毫秒的定时器。
第二级数组存放的是超时时间范围为 [256, 16384) 毫秒的定时器（16384 = 256 * 64）。
第三级数组存放的是超时时间范围为 [16384, 1048576) 毫秒的定时器（1048576 = 256 * 64 * 64）。
第四级数组存放的是超时时间范围为 [1048576, 67108864) 毫秒的定时器（67108864 = 256 * 64 * 64 * 64）。
第五级数组存放的是超时时间范围为 [67108864, 4294967296) 毫秒的定时器（4294967296 = 256 * 64 * 64 * 64 * 64）。

2. 执行定时器

接下来，我们将要分析内核是如何选择到期的定时器来执行的。

如果所有定时器只存储在一级数组中，那么选择到期的定时器就非常简单：由于数组每个槽位的索引对应着定时器的超时时间，所以只需要在时钟中断发生时，执行到期指针指向的定时器列表。执行完毕后，将到期指针移动到下一个位置即可。如下图所示：

但对于定时器存储在多级数组的情况，算法就变得复杂很多。

从上面的分析可知，第一级数组存放的是 0 ～ 255 毫秒后到期的定时器列表，而数组的索引对应的就是定时器的超时时间。如下图所示：

而其他等级的数组，每个槽位存放的定时器其超时时间并不是一个固定的值，而是一个范围，范围与数组的等级和槽位的索引值有关，其计算方式为：

256 * 64^n * 槽位索引 <= 超时时间 < 256 * 64^n * (槽位索引+1)

在上面的公式中，n 的值等于 数组的等级 减去 2。所以对于第二级数组来说，其公式如下：

256 * 槽位索引 <= 超时时间 < 256 * (槽位索引+1)

第三级数组公式如下：

256 * 64 * 槽位索引 <= 超时时间 < 256 * 64 * (槽位索引+1)

第四和第五级数组如此类推。

由于内核不会使用索引为 0 的槽位，所以第二、第三级数组的定时器如下图所示：

内核只会执行第一级数组中的定时器，每当时钟中断触发时，会执行第一级数组 到期指针 指向的定时器列表，执行完毕后会将 到期指针 向下移动一位。如下图所示：

当到期指针执行完最后一个槽位的定时器列表后，会重新移动到第一个槽位。

那么其他级别数组的定时器在什么时候才会被执行呢？其实对于其他级别的数组也有一个 到期指针，每当前一级别的数组执行完一轮后，当前级别数组的 到期指针 将会移动到下一个槽位，如：当第一级数组执行一轮后，第二级数组的 到期指针 将会移动到下一个槽位。

其他级别的数组（非第一级数组）移动 到期指针 时，会将指针指向的定时器列表从数组中删除，并且重新添加到内核中。如下图所示：

如上图所示，第一级数组执行一轮后，内核将会把第二级数组的到期指针指向的定时器列表删除，并且重新添加到内核中。然后，将会把到期指针移动到下一个槽位。

第三级数组也会在第二级数组执行一轮后，将其到期指针指向的定时器列表删除，并且重新添加到内核中。接着将到期指针移动到下一个槽位，其他级别的数组如此类推。

源码实现

接下来，我们将会分析 Linux 内核是如何实现低精度定时器的。由于高版本的内核其实现与上面介绍的原理有些区别，但基本原理是一致的，这里我们将使用 2.4.37 版本作为分析的对象。

1. 五个等级数组

如上面分析一致，在 Linux 内核中定义了 5 个数组来存放系统中的定时器，如下代码所示：

struct timer_vec {
 int index;     // 到期指针
 struct list_head vec[64];
};

struct timer_vec_root {
 int index;     // 到期指针
 struct list_head vec[256];
};

static struct timer_vec tv5;
static struct timer_vec tv4;
static struct timer_vec tv3;
static struct timer_vec tv2;
static struct timer_vec_root tv1;

上面代码中，tv1、tv2、tv3、tv4、tv5 分别对应第一级、二级、三级、四级和五级数组。

通过代码可知，数组元素的类型为链表，用于存放不同到期时间的定时器。另外，除了第一级数组的元素个数是 256 个外，其他级别的数组的元素个数都是 64 个。每个级别的数组都有一个到期指针，用于指向当前正在执行的定时器列表。

我们接着来看看内核怎么初始化这些数组的，内核调用 init_timervecs() 函数来初始化各级数组。代码如下：

void init_timervecs(void)
{
    int i;

    for (i = 0; i < TVN_SIZE; i++) {
        INIT_LIST_HEAD(tv5.vec + i);
        INIT_LIST_HEAD(tv4.vec + i);
        INIT_LIST_HEAD(tv3.vec + i);
        INIT_LIST_HEAD(tv2.vec + i);
    }
  
    for (i = 0; i < TVR_SIZE; i++)
        INIT_LIST_HEAD(tv1.vec + i);
}

init_timervecs() 主要通过 INIT_LIST_HEAD 宏来初始化各级数组的元素。

2. 定时器对象

在内核中，定时器使用 timer_list 对象来表示，其定义如下：

struct timer_list {
    struct list_head list;
    unsigned long expires;
    unsigned long data;
    void (*function)(unsigned long);
};

下面介绍一下 timer_list 对象各个字段的作用：

list：用于连接到期时候相同的定时器。
expires：定时器的到期时间。
data：传给回调函数的数据。
function：定时器到期后，将会调用的回调函数。

我们要向内核添加一个定时器时，需要先创建一个 timer_list 对象，并且设置其到期时间和回调函数。

3. 添加定时器

在内核中，可以使用 add_timer() 函数来添加一个定时器。其代码如下所示：

void add_timer(struct timer_list *timer)
{
    unsigned long flags;

    // 上锁
    spin_lock_irqsave(&timerlist_lock, flags);
    ...
    // 向内核添加定时器
    internal_add_timer(timer);
    // 解锁
    spin_unlock_irqrestore(&timerlist_lock, flags);
    return;
}

从上面代码可以看出，add_timer() 函数主要通过调用 internal_add_timer() 函数来添加定时器。我们继续来分析 internal_add_timer() 函数的实现，代码如下：

static inline void internal_add_timer(struct timer_list *timer)
{
    unsigned long expires = timer->expires;
    unsigned long idx = expires - timer_jiffies; // 多少毫秒数后到期
    struct list_head * vec;

    if (idx < TVR_SIZE) {
        int i = expires & TVR_MASK;
        vec = tv1.vec + i;
    } else if (idx < 1 << (TVR_BITS + TVN_BITS)) {
        int i = (expires >> TVR_BITS) & TVN_MASK;
        vec = tv2.vec + i;
    } else if (idx < 1 << (TVR_BITS + 2 * TVN_BITS)) {
        int i = (expires >> (TVR_BITS + TVN_BITS)) & TVN_MASK;
        vec =  tv3.vec + i;
    } else if (idx < 1 << (TVR_BITS + 3 * TVN_BITS)) {
        int i = (expires >> (TVR_BITS + 2 * TVN_BITS)) & TVN_MASK;
        vec = tv4.vec + i;
    } else if ((signed long) idx < 0) {
        vec = tv1.vec + tv1.index;
    } else if (idx <= 0xffffffffUL) {
        int i = (expires >> (TVR_BITS + 3 * TVN_BITS)) & TVN_MASK;
        vec = tv5.vec + i;
    } else {
        INIT_LIST_HEAD(&timer->list);
        return;
    }

    list_add(&timer->list, vec->prev);
}

internal_add_timer() 函数首先会计算定时器还有多少毫秒到期，然后按照到期的毫秒数来选择对应的级别数组：

如果到期时间小于256毫秒，那么将会添加到第一级数组中。
如果到期时间大于等于256毫秒，并且小于16384毫米，那么将会添加到第二级数组中。
其他等级如此类推。

选择到合适的数组后，内核会调用 list_add() 函数将定时器添加到对应槽位的链表中。

4. 执行到期的定时器

内核会在时钟中断中通过调用 run_timer_list() 函数来执行到期的定时器，其实现如下：

static inline void run_timer_list(void)
{
    ...
    while ((long)(jiffies - timer_jiffies) >= 0) {
        struct list_head *head, *curr;

        // 1. 如果第一级数组已经执行完一轮（到期指针变为0）
        if (!tv1.index) {
            int n = 1;
            do {
                cascade_timers(tvecs[n]);
            } while (tvecs[n]->index == 1 && ++n < NOOF_TVECS);
        }

repeat:
        // 2. 第一级数组当前到期指针指向的定时器列表
        head = tv1.vec + tv1.index;

        // 3. 遍历到期的定时器列表
        curr = head->next;
        if (curr != head) {
            struct timer_list *timer;
            void (*fn)(unsigned long);
            unsigned long data;

            timer = list_entry(curr, struct timer_list, list);
            fn = timer->function;
            data= timer->data;

            // 4. 把定时器从链表中删除
            detach_timer(timer);
            timer->list.next = timer->list.prev = NULL;
            timer_enter(timer);

            spin_unlock_irq(&timerlist_lock);
            // 5. 执行定时器的回调函数
            fn(data);
            spin_lock_irq(&timerlist_lock);

            timer_exit();
            goto repeat;
        }
        ++timer_jiffies;
        // 6. 将到期指针移动一个位置
        tv1.index = (tv1.index + 1) & TVR_MASK;
    }
    ...
}

run_timer_list() 函数主要按照以下步骤来执行到期的定时器：

如果第一级数组已经执行完一轮（也就是说，到期指针变为0时），通过调用 cascade_timers() 函数来计算其他等级当前到期指针指向的定时器列表（重新添加到内核中）。
遍历第一级数组的到期指针指向的定时器列表。
把定时器从链表中删除。
执行定时器的回调函数。
将到期指针移动一个位置。

从时间轮的原理可知，每当某一级数组执行完一轮后，就会移动下一级数组的到期指针，并且将指针指向的定时器列表重新添加到内核中，这个过程由 cascade_timers() 函数完成。代码如下所示：

static inline void cascade_timers(struct timer_vec *tv)
{
    struct list_head *head, *curr, *next;

    head = tv->vec + tv->index;
    curr = head->next;

    // 1. 遍历定时器列表
    while (curr != head) {
        struct timer_list *tmp;

        tmp = list_entry(curr, struct timer_list, list);
        next = curr->next;
        list_del(curr);
        // 2. 将定时器重新添加到内核中
        internal_add_timer(tmp);
        curr = next;
    }
    INIT_LIST_HEAD(head);
    // 3. 将到期指针移动到下一个位置
    tv->index = (tv->index + 1) & TVN_MASK;
}