1. 程式人生 > >Linux時間管理涉及數據結構和傳統低分辨率時鐘的實現

Linux時間管理涉及數據結構和傳統低分辨率時鐘的實現

load fin 手動 span div current lds 其中 context

上篇文章大致描述了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內核架構》

Linux時間管理涉及數據結構和傳統低分辨率時鐘的實現