Linux时间管理涉及数据结构和传统低分辨率时钟的实现

Posted 2020-09-25 太初有道，道与神同在，道就是神……

上篇文章大致描述了Linux时间管理的基本情况，看了一些大牛们的博客感觉自己写的内容很匮乏，但是没办法，只能通过这种方式提升自己……闲话不说，本节介绍下时间管理下重要的数据结构

设备相关数据结构

//时钟源结构 

struct clocksource{}

//时钟设备结构 

struct tick_device {

struct clock_event_device *evtdev;

enum tick_device_mode mode;//记录对应时钟事件设备的模式

};

 

enum tick_device_mode {

TICKDEV_MODE_PERIODIC,//周期模式

TICKDEV_MODE_ONESHOT,//单点触发模式

};

//时钟事件设备结构 

struct clock_event_device {}

 

定时器相关数据结构

低分辨率定时器

struct timer_list{}

struct tvec_base{}

struct timerqueue_head {}

struct timerqueue_node {}

 

高分辨率定时器 

 struct hrtimer_cpu_base{}

 struct hrtimer_clock_base{}

时间相关定义 

union ktime {

s64 tv64;

#if BITS_PER_LONG != 64 && !defined(CONFIG_KTIME_SCALAR)

struct {

# ifdef __BIG_ENDIAN

s32 sec, nsec;

# else

s32 nsec, sec;

# endif

} tv;

#endif

};

 

 

低分辨率 时钟实现

在低分辨率模式下，可以实现周期时钟和动态时钟（在支持单点触发模式状态下）。但是目前低分辨率下的动态时钟并入了高分辨率的处理框架下，所以本节仅仅描述低分辨率下的周期时钟实现。

如前所述，当设备处于TICKDEV_MODE_PERIODIC模式时，其运行在周期模式。基于此实现的定时器成为低分辨率定时器。此模式下事件定期发生，每秒HZ次。HZ一般取250，即两个中断之间的间隔为4ms。这个频率对于计算机而言的确有些低了。当然在编译时，通过配置选项CONFIG_HZ设置。HZ越大表示一秒内发生时钟中断的次数越多，更多的任务可以得到更及时的处理，对于交互性要求较高的系统比较适用。但是中断次数的 增加同样意味着CPU被打断的次数过多，需要处理更多的内核事件，对于性能也是不小的开销。由于由时钟设备直接周期性的提供中断，且不需要手动设置下一次的事件触发时间，故基于低分辨率时钟的低分辨率定时器的实现较为简单。

低分辨率模式下，时钟中断的处理函数为timer_interrupt(IA 32架构下).该函数更新全局信息主要是jiffies以及更新进程时间信息。见函数xtime_update(nticks);和函数update_process_times。

xtime_update

void xtime_update(unsigned long ticks)

{

write_seqlock(&jiffies_lock);

do_timer(ticks);

write_sequnlock(&jiffies_lock);

}

 函数调用了do_timer，其中ticks是更新的滴答计数

do_timer

void do_timer(unsigned long ticks)

{ /*更新jiffies*/

jiffies_64 += ticks;

/*更新墙上时间*/

update_wall_time();

/*计算全局负载*/

calc_global_load(ticks);

}

 

 在do_timer中不仅更新了jiffies，还更新了墙上时间。关于jiffies 和墙上时间，后续还会详细介绍。在最后还计算了全局负载。在update_process_times函数中，会更新当前进程的时间，处理本地定时器、并会通过scheduler_tick调用周期性调度器。代码如下

update_process_times

void update_process_times(int user_tick)

{

struct task_struct *p = current;

int cpu = smp_processor_id();

/* Note: this timer irq context must be accounted for as well. */

account_process_tick(p, user_tick);//更新进程时间

run_local_timers();//处理本地定时器

rcu_check_callbacks(cpu, user_tick);

#ifdef CONFIG_IRQ_WORK

if (in_irq())

irq_work_run();

#endif

scheduler_tick();

run_posix_cpu_timers(p);

}

 本地定时器的处理时通过软中断实现的，即定时器的处理时机位于处理软中断的时候，由于软中断并不是硬件中断，不能任意的触发执行，需要接受系统的安排，所以定时器的执行可能会有所延迟，但是绝对不会提前。回想之前关于软中断的文章，在软中段的类型中有TIMER_SOFTIRQ，便是对应普通的定时器处理。而这里对定时器的处理也很简单，看下代码

 

run_local_timers

void run_local_timers(void)

{

hrtimer_run_queues();

raise_softirq(TIMER_SOFTIRQ);

}

 hrtimer_run_queues是为了在低精度模式下处理高精度的定时器，主要用在高精度模式未启动的时期，待高精度模式启动之后，该函数就为空。之后触发了一个TIMER_SOFTIRQ类型的软中断。这样在下个软中断处理时机，会处理该定时器。

 

到最后调用了周期性的调度器scheduler_tick，该函数最终会调用到具体调度类如CFS的周期调度器，周期调度器会更新当前调度实体的运行时间、更新当前调度实体以及队列的虚拟运行时间vruntime。如果调度实体是进程，还需要更新其所在的组的时间信息，cgroup相关，暂不深入。最后计算下当前队列的时间是否还充足，如果不足就需要尝试扩展runtime，如果扩展runtime失败并且当前任务不为空，就设置重调度位。在不考虑高分辨率时钟的情况下回 检查是否有其他等待运行的进程，如果有，则检查抢占。

普通定时器的处理

函数run_timer_softirq 为定时器软中断的处理函数。

 

run_timer_softirq

static void run_timer_softirq(struct softirq_action *h)

{

struct tvec_base *base = __this_cpu_read(tvec_bases);

hrtimer_run_pending();

if (time_after_eq(jiffies, base->timer_jiffies))

__run_timers(base);

}

 

 hrtimer_run_pending是在处理一般定时器的时候不断的检查是否可以转成高分辨率模式，如果可以则进行转换。然后判断当前时间和定时器时间，在介绍具体的处理之前先介绍下普通定时器的组织。

 普通定时器的组织

 由于定时器是局部于CPU的，所以每个CPU维护一个定时器的管理结构

static DEFINE_PER_CPU(struct tvec_base *, tvec_bases) = &boot_tvec_bases;

//该结构描述如下 

struct tvec_base {

spinlock_t lock;

struct timer_list *running_timer;//记录当前正在处理的定时器

unsigned long timer_jiffies;//在此之前的定时器均已经处理，所以每次处理一个定时器要递增该值

unsigned long next_timer;

unsigned long active_timers;

/*保存CPU 上的定时器*/

struct tvec_root tv1;

struct tvec tv2;

struct tvec tv3;

struct tvec tv4;

struct tvec tv5;

} ____cacheline_aligned;

struct tvec {

struct list_head vec[TVN_SIZE];

};

struct tvec_root {

struct list_head vec[TVR_SIZE];

}; 

 

 有两个重要的结构tvec_root和tvec记录定时器。系统主要从第一个结构提取处理，后者就做备用存储。可以看到，tvec_root和tvec均是一个链表头数组，前者有TVR_SIZE 项一般是256，对应0-255个时钟周期内到期的定时器，如果有多个定时器对应的时间相同，则使用链表维护。从2-5都是后备存储，对于这几个组的容量说明见下表

组	时间间隔/时钟周期	单项容量
Tv1	0~255	1
Tv2	256~214-1	256
Tv3	214~220-1	214
Tv4	220~226-1	220
Tv5	226~232-1	226

 由此可见，后继组的一项对应的时钟间隔就是整个前驱组的整个间隔，在填充的时候，从后继组中取出一项便可以填充整个前驱组，比如当TV1处理完，则可以从Tv2取出第一项，对TV1 进行填充。以此类推。tvec_base中还有一个timer_jiffies字段表示在此之前的定时器均已经得到处理，所以每次处理完Tv1中的一个表项，就需要递增该值。而普通定时器结构为timer_list，我们只关注几个字段

struct timer_list {

/*

* All fields that change during normal runtime grouped to the

* same cacheline

*/

struct list_head entry;

unsigned long expires;

struct tvec_base *base;

void (*function)(unsigned long);

unsigned long data;

…… 

 } 

 首个字段entry作为一个节点维护其在双链表中存在。expires记录到期时间，单位是jiffies，base指向其所属的tvec_base,接下来是一个函数指针和一个data字段，这就是定时器注册的回调函数，data为参数。OK，下面看具体处理流程，见__run_timers函数

 

static inline void __run_timers(struct tvec_base *base)

{

struct timer_list *timer;

spin_lock_irq(&base->lock);

while (time_after_eq(jiffies, base->timer_jiffies)) {

struct list_head work_list;

struct list_head *head = &work_list;

int index = base->timer_jiffies & TVR_MASK;

/*

* Cascade timers:

*/

if (!index &&

(!cascade(base, &base->tv2, INDEX(0))) &&

(!cascade(base, &base->tv3, INDEX(1))) &&

!cascade(base, &base->tv4, INDEX(2)))

cascade(base, &base->tv5, INDEX(3));

++base->timer_jiffies;

/*得到某个jiffies的定时器链表*/

list_replace_init(base->tv1.vec + index, &work_list);

while (!list_empty(head)) {

void (*fn)(unsigned long);

unsigned long data;

bool irqsafe;

/*获取定时器*/

timer = list_first_entry(head, struct timer_list,entry);

/*得到定时器的回调函数*/

fn = timer->function;

/*得到定时器的参数*/

data = timer->data;

irqsafe = tbase_get_irqsafe(timer->base);

timer_stats_account_timer(timer);

base->running_timer = timer;

detach_expired_timer(timer, base);

if (irqsafe) {

spin_unlock(&base->lock);

/*处理该定时器*/

call_timer_fn(timer, fn, data);

spin_lock(&base->lock);

} else {

spin_unlock_irq(&base->lock);

call_timer_fn(timer, fn, data);

spin_lock_irq(&base->lock);

}

}

}

base->running_timer = NULL;

spin_unlock_irq(&base->lock);

}

 函数主体是一个大的while循环，循环条件就是当前时间大于base->timer_jiffies，这段时间内的定时器还没有处理，这段时间内很可能没有定时器，但是总是需要检查下。前面已经介绍，tv1数组的项对应0-255个时钟周期，每个周期对应一个，故这里通过base->timer_jiffies & TVR_MASK来获取下标，接下来的if是对那几个数组做填充，注意初次执行时一般是不会填充的，因为base->timer_jiffies在自增到256的倍数的时候正好大于了当前jiffies的时候并不多。这点后续在讨论。没什么异常情况就自增base->timer_jiffies，然后根据index获取链表，接下来又是一个循环，用以处理这个时间上的所有定时器。这里就没什么特殊的，后去定时器结构timer_list，然后获取其回调函数和参数，然后就通过call_timer_fn执行回调函数了。在处理之前已经把该定时器从链表中摘下。（这里我有个疑问，为何不在处理完成后再摘下呢？）

 

定时器向量的填充问题

 再次参考下代码

int index = base->timer_jiffies & TVR_MASK;

/*貌似是每处理256个jiffies就填充一次*/

/*

* Cascade timers:

*/

if (!index &&

(!cascade(base, &base->tv2, INDEX(0))) &&

(!cascade(base, &base->tv3, INDEX(1))) &&

!cascade(base, &base->tv4, INDEX(2)))

cascade(base, &base->tv5, INDEX(3));

…… 

 

 系统启动后，base->timer_jiffies是一直递增的，这里每次递增256个时钟周期就对定时器向量填充一次。256个时钟周期有可能是经过分批处理才完成的。也可能是好长一段时间没有处理定时器了，累计的定时器比较多，一次性就处理好多。这里并不重要。重要的是每次 base->timer_jiffies递增了

 256后，index就为0，然后就从下一级的向量组中填充。以此类推。

#define INDEX(N) ((base->timer_jiffies >> (TVR_BITS + (N) * TVN_BITS)) & TVN_MASK)

 

 INDEX宏用以计算源向量组中的下标，为何这么整不太容易理解，举个例子分析

 现在 base->timer_jiffies递增到了0x0000c300,此时触发了填充首个向量组，首个向量组的容量为256，因此INDEX宏的参数为0，这里就右移8位以256个时钟周期为单位进行处理；类似的，当填充第二个向量组时，其容量为2^(8+6)，这里就需要右移8+6=14位，依次类推；我们可以知道，上一轮处理的 base->timer_jiffies必定为0x0000c2**,处理完成后才递增到了0x0000c300，按照上述公式计算0x0000c300>>8&0x1F，得到3，即从源向量组的第三项开始填充。因为在此之前的项肯定已经填充到了上一级且已经处理过了，每当index循环到0时，就触发下一级的填充，有一点需要注意，因为jiffies在不断递增，而向量组中的安排是按照时间线安排的，比如Tv4的首个表项肯定为空，因为其内容离散分布在前TV3-TV1中，TV3的首个表项也为空，其内容离散分布在TV1-TV2中，所以每次填充柄没有指定填充到固定的TV，而是采用统一的函数__internal_add_timer，根据各个定时器的到期时间进行添加。当从后一个向量组添加时，会添加到前面所有的向量组。

 

 

以马内利！

参考资料：

linux3.10.1源码

深入linux内核架构》