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

上篇文章大致描述了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 {
s64tv64;
#if BITS_PER_LONG != 64 && !defined(CONFIG_KTIME_SCALAR)
struct {
# ifdef __BIG_ENDIAN
s32sec, nsec;
# else
s32nsec, 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类型的软中断,如果没在中断上下文还会唤醒软中断守护进程ksoftirqd对其进程处理,否则在下个软中断处理时机,会处理该定时器。

到最后调用了周期性的调度器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内核架构》

上一篇:【转】10 条提升 Android 性能的建议


下一篇:製程能力介紹(SPC introduction) ─ Cp之製程能力解釋