死鎖問題分析
死鎖就是多個進程(線程)因為等待別的進程已占有的自己所需要的資源而陷入阻塞的一種狀態,死鎖狀態一旦形成,進程本身是解決不了的,需要外在的推動,才能解決,最重要的是死鎖不僅僅影響進程業務,而且還會占用系統資源,影響其他進程。所以內核中設計了內核死鎖檢測機制,一旦發現死鎖進程,就重啟OS,快刀斬亂麻解決問題。之所以使用重啟招數,還是在於分布式系統中可以容忍單點崩潰,不能容忍單點進程計算異常,否則進行死鎖檢測重啟OS就得不償失了。
內核提供自旋鎖、信號量等鎖形式的工具,具體不再贅述。
Linux內核死鎖主要分為分為兩種:D狀態死鎖和R狀態死鎖。
一、D狀態死鎖檢測
D狀態死鎖:進程長時間處於TASK_UNINTERRUPTIBLE而不恢復的狀態。進程處於TASK_UNINTERRUPTIBLE狀態,不響應其他信號(kill -9),保證一些內核原子操作不被意外中斷。但這種狀態時間長就表示進程異常了,需要處理。
內核D狀態死鎖檢測就是hung_task機制,主要代碼就在kernel/hung_task.c文件。
具體實現原理:
1.創建Normal級別的khungtaskd內核線程,在死循環中每隔sysctl_hung_task_timeout_secs時間后check一下,用schedule_timeout定時(節約定時器浪費的CPU)。
2.調用do_each_thread,while_each_thread宏遍歷所有的進程信息,如果有D狀態進程,則檢查最近切換次數和task計算是否一致,即最近是否有調度切換,如果一致,則沒有切換,打印相關信息,並根據sysctl_hung_task_panic開關決定是否重啟。
對應用戶態控制的proc接口有:
/proc/sys/kernel/hung_task_timeout_secs,hung_task_panic等。
二、R狀態死鎖檢測
R狀態死鎖:進程長時間處於TASK_RUNNING 狀態搶占CPU而不發生切換,一般是,進程關搶占后一直執行任務,或者進程關搶占后處於死循環或者睡眠,此時往往會導致多個CPU互鎖,整個系統異常。
補充:lockdep不是所謂的死鎖。
內核R狀態死鎖檢測機制就是lockdep機制,入口即是lockup_detector_init函數。
1.通過cpu_callback函數調用watchdog_enable,在每個CPU core上創建SCHED_FIFO級別的實時線程watchdog,其中使用了hrtimer定時器,控制檢查周期。
2.hrtimer定時器調用watchdog_timer_fn進行清狗的時間檢查,而線程則每次重置清狗時間,如果watchdog_timer_fn發現狗的重置時間已經和當前時間差出危險值,則根據開關進行panic處理。
對應用戶態控制的proc接口有:
/proc/sys/kernel/watchdog_thresh,softlockup_panic等。
整個死鎖檢測機制比較簡單,但cpu_callback函數結構性設計巧妙,可以在很多地方參考使用。
static int __cpuinit
cpu_callback(struct notifier_block *nfb, unsigned long action, void *hcpu)
{
int hotcpu = (unsigned long)hcpu;
switch (action) {
case CPU_UP_PREPARE:
case CPU_UP_PREPARE_FROZEN:
watchdog_prepare_cpu(hotcpu);
break;
case CPU_ONLINE:
case CPU_ONLINE_FROZEN:
if (watchdog_enabled)
watchdog_enable(hotcpu);
break;
#ifdef CONFIG_HOTPLUG_CPU
case CPU_UP_CANCELED:
case CPU_UP_CANCELED_FROZEN:
watchdog_disable(hotcpu);
break;
case CPU_DEAD:
case CPU_DEAD_FROZEN:
watchdog_disable(hotcpu);
break;
#endif /* CONFIG_HOTPLUG_CPU */
}
/*
* hardlockup and softlockup are not important enough
* to block cpu bring up. Just always succeed and
* rely on printk output to flag problems.
*/
return NOTIFY_OK;
}
死鎖:就是多個進程(≥2)因為爭奪資源而相互等待的一種現象,若無外力推動,將無法繼續運行下去。
注意,只有在多進程或者多線程之間或者他們與中斷之間相互通訊或者共享資源才有可能發生死鎖,單線程或者進程之間沒有聯系的話,一般不會發生死鎖。鎖的種類比較多,這里主要說自旋鎖和信號量。兩者的差別就在於前者獲得不到資源時的動作是不斷的資源(即忙轉浪費cpu的cycles)而后者則表現為睡眠等待。
死鎖的基本情況如下:
(一)、自旋鎖
1 遞歸使用:同一個進程或線程中,申請自旋鎖,但沒有釋放之前又再次申請,一定產生死鎖。
2 進程得到自旋鎖后阻塞,睡眠:在獲得自旋鎖之后調用copy_from_user()、copy_to_ser()、和kmalloc()等有可能引起阻塞的函數。
3 中斷中沒有關中斷,或着因為申請未釋放的自旋鎖:在中斷中使用自旋鎖是可以的,應該在進入中斷的時候關閉中斷,不然中斷再次進入的時候,中斷處理函數會自旋等待自旋鎖可以再次使用。或者在進程中申請了自旋鎖,釋放前進入中斷處理函數,中斷處理函數又申請同樣的自旋鎖,這將導致死鎖。
4 中斷與中斷下半部共享資源和中斷與進程共享資源死鎖出現的情況類似。
5 中斷下半部與進程共享資源和中斷與進程共享資源死鎖出現的情況類似。
自旋鎖三種狀態:
自旋鎖保持期間是搶占失效的(內核不允許被搶占)。
1 單CPU且內核不可搶占:
自旋鎖的所有操作都是空。不會引起死鎖,內核進程間不存在並發操作進程,進程與中斷仍然可能共享數據,存在並發操作,此時內核自旋鎖已經失去效果。
2 單CPU且內核可搶占:
當獲得自旋鎖的時候,禁止內核搶占直到釋放鎖為止。此時可能存在死鎖的情況是參考自旋鎖可能死鎖的一般情況。
禁止內核搶占並不代表不會進行內核調度,如果在獲得自旋鎖后阻塞或者主動調度,內核會調度其他進程運行,被調度的內核進程返回用戶空間時,會進行用戶搶占,此時調用的進程再次申請上次未釋放的自旋鎖時,會一直自旋。但是內核被禁止搶占,從而造成死鎖。
內核被禁止搶占,但此時中斷並沒被禁止,內核進程可能因為中斷申請自旋鎖而死鎖。
3 多CPU且內核可搶占:
這才是是真正的SMP的情況。當獲得自旋鎖的時候,禁止內核搶占直到釋放鎖為止。
(二)信號量
1 遞歸使用: 同理,在同一個進程或線程中,申請了信號量,但沒有釋放之前又再次申請,進程會一直睡眠,這種情況一定死鎖。
2 進程得到信號量后阻塞,睡眠:由於獲取到信號量的進程阻塞或者隨眠,其他在獲取不到后信號量也會進入睡眠等待,這種情況可能造成死鎖。
3 中斷中申請信號量:由於信號量在獲取不到自旋鎖后會進入睡眠等待,中斷處理函數不允許睡眠,如果睡眠,中斷將無法返回。
4 中斷下半部申請信號量:中斷下半部允許睡眠,這種情況不會造成死鎖。
5 倆個進程相互等待資源:進程1獲得信號量A,需要信號量B,在進程1需要信號量B之前進程2獲得信號量B,需要信號量A。進程1、2因相互等待資源而死鎖。
上面是死鎖的基本情況和類型。linux本身就提供了檢測死鎖的機制,如下:
(一)D狀態死鎖檢測
所謂D狀態死鎖:進程長時間(系統默認配置120秒)處於TASK_UNINTERRUPTIBLE 睡眠狀態,這種狀態下進程不響應異步信號。如:進程與外設硬件的交互(如read),通常使用這種狀態來保證進程與設備的交互過程不被打斷,否則設備可能處於不可控的狀態。
對於這種死鎖的檢測linux提供的是hungtask機制,主要內容集中在Hung_task.c文件中。具體實現原理如下:
1)、系統創建normal級別的khungtaskd內核線程,內核線程每120秒檢查一次,檢查的內容:遍歷所有的線程鏈表,發現D狀態的任務,就判斷自最近一次切換以來是否還有切換發生,若是有,則返回。若沒有發生切換,則把任務的所有調用棧等信息打印出來。
2)、具體實現過程如下:
首先,hung_task_init創建一個名為khungtaskd的內核線程,內核線程的工作由watchdog來完成。
點擊(此處)折疊或打開
static int __init hung_task_init(void)
{
atomic_notifier_chain_register(&panic_notifier_list, &panic_block);
watchdog_task = kthread_run(watchdog, NULL, "khungtaskd");
return 0;
}
其次,我們看watchdog的實現:
點擊(此處)折疊或打開
static int watchdog(void *dummy)
{
//將內核線程設置為normal級別
set_user_nice(current, 0);
for ( ; ; ) {
//設置hungtask的校驗時間間隔,用戶可以修改這個時間,默認為120秒
unsigned long timeout = sysctl_hung_task_timeout_secs;
while (schedule_timeout_interruptible(timeout_jiffies(timeout)))
timeout = sysctl_hung_task_timeout_secs;
//核心的檢查代碼在下面的函數中實現。 check_hung_uninterruptible_tasks(timeout);
}
return 0;
}
最后,我們分析一下hungtask的核心實現check_hung_uninterruptible_tasks,如下:
點擊(此處)折疊或打開
static void check_hung_uninterruptible_tasks(unsigned long timeout)
{
int max_count = sysctl_hung_task_check_count;
int batch_count = HUNG_TASK_BATCHING;
struct task_struct *g, *t;
/*
* If the system crashed already then all bets are off,
* do not report extra hung tasks:
*/
//判斷系統是否已經die、oops或者panic了,若是系統已經crash了,就無需再做hungtask了。
if (test_taint(TAINT_DIE) || did_panic)
return;
rcu_read_lock();
//檢查進程的列表,尋找D狀態的任務
do_each_thread(g, t) {
//判斷用戶是否設置了檢查進程的數量,若是已經達到用戶設置的限制,就跳出循環。
if (!max_count--)
goto unlock;
//判斷是否到達批處理的個數,做這個批處理的目的就是因為整個檢查是在關搶占的前提下進行的,可能進程列表的進程數很多,為了防止hungtask壟斷cpu,所以,做了一個批處理的限制,到達批處理的數量后,就放一下權,給其他的進程運行的機會。
if (!--batch_count) {
batch_count = HUNG_TASK_BATCHING;
rcu_lock_break(g, t);
/* Exit if t or g was unhashed during refresh. */
if (t->state == TASK_DEAD || g->state == TASK_DEAD)
goto unlock;
}
/* use "==" to skip the TASK_KILLABLE tasks waiting on NFS */
//如果進程處於D狀態,就開始把相關的信息顯示出來了。
if (t->state == TASK_UNINTERRUPTIBLE)
check_hung_task(t, timeout);
} while_each_thread(g, t);
unlock:
rcu_read_unlock();
}
點擊(此處)折疊或打開
static void check_hung_task(struct task_struct *t, unsigned long timeout)
{
//統計進程的切換次數=主動切換次數+被動切換次數
unsigned long switch_count = t->nvcsw + t->nivcsw;
/*
* Ensure the task is not frozen.
* Also, when a freshly created task is scheduled once, changes
* its state to TASK_UNINTERRUPTIBLE without having ever been
* switched out once, it musn
怎么樣,經過上面的分析,可以發現其實原理很簡單。
(二)R狀態死鎖檢測
所謂R狀態死鎖:進程長時間(系統默認配置60秒)處於TASK_RUNNING 狀態壟斷cpu而不發生切換,一般情況下是進程關搶占后長時候干活,有時候可能進程關搶占后處於死循環或者睡眠后,這樣就造成系統異常。
對於這種死鎖的檢測機制linux提供的機制是softlockup。主要集中在softlockup.c文件中。
1)、系統創建一個fifo的進程,此進程周期性的清一下時間戳(per cpu),而系統的時鍾中斷中會被softlockup掛入一個鈎子(softlockup_tick),這個鈎子就是每個時鍾中斷到來的時候都檢查是否每cpu的時間戳被touch了,若在閥值60秒內都沒有被touch,系統就打印調試信息。
2)、讓我們分析一下具體的實現:
首先,系統初始化的時候為每個cpu創建一個watchdog線程,這個線程是fifo的。具體實現如下:
點擊(此處)折疊或打開
static int __init spawn_softlockup_task(void)
{
void *cpu = (void *)(long)smp_processor_id();
int err;
//可以通過啟動參數禁止softlockup
if (nosoftlockup)
return 0;
//下面兩個回調函數就是為每個cpu創建一個watchdog線程
err = cpu_callback(&cpu_nfb, CPU_UP_PREPARE, cpu);
if (err == NOTIFY_BAD) {
BUG();
return 1;
}
cpu_callback(&cpu_nfb, CPU_ONLINE, cpu);
register_cpu_notifier(&cpu_nfb);
atomic_notifier_chain_register(&panic_notifier_list, &panic_block);
return 0;
}
點擊(此處)折疊或打開
static int __cpuinit
cpu_callback(struct notifier_block *nfb, unsigned long action, void *hcpu)
{
int hotcpu = (unsigned long)hcpu;
struct task_struct *p;
switch (action) {
case CPU_UP_PREPARE:
case CPU_UP_PREPARE_FROZEN:
BUG_ON(per_cpu(softlockup_watchdog, hotcpu));
//創建watchdog內核線程
p = kthread_create(watchdog, hcpu, "watchdog/%d", hotcpu);
if (IS_ERR(p)) {
printk(KERN_ERR "watchdog for %i failed\n", hotcpu);
return NOTIFY_BAD;
}
//將時間戳清零
per_cpu(softlockup_touch_ts, hotcpu) = 0;
//設置watchdog
per_cpu(softlockup_watchdog, hotcpu) = p;
//綁定cpu
kthread_bind(p, hotcpu);
break;
case CPU_ONLINE:
case CPU_ONLINE_FROZEN:
//喚醒watchdog這個內核線程 wake_up_process(per_cpu(softlockup_watchdog, hotcpu));
break;
......
return NOTIFY_OK;
}
其次,我們看一下內核線程watchdog的實現,如下:
點擊(此處)折疊或打開
static int watchdog(void *__bind_cpu)
{
//將watchdog設置為fifo
struct sched_param param = { .sched_priority = MAX_RT_PRIO-1 };
sched_setscheduler(current, SCHED_FIFO, ?m);
//清狗,即touch時間戳
/* initialize timestamp */
__touch_softlockup_watchdog();
//當前進程可以被信號打斷
set_current_state(TASK_INTERRUPTIBLE);
/*
* Run briefly once per second to reset the softlockup timestamp.
* If this gets delayed for more than 60 seconds then the
* debug-printout triggers in softlockup_tick().
*/
while (!kthread_should_stop()) {
//核心實現就是touch時間戳,讓出cpu
__touch_softlockup_watchdog();
schedule();
if (kthread_should_stop())
break;
set_current_state(TASK_INTERRUPTIBLE);
}
__set_current_state(TASK_RUNNING);
return 0;
}
最后,softlockup在時鍾中斷中掛上一個鈎子softlockup_tick,每個時鍾中斷都來檢查watchdog這個fifo進程是否有touch過時間戳,若是60秒都沒有touch過,就向系統上報異常信息了,如下:
void softlockup_tick(void)
{
//取得當前的cpu
int this_cpu = smp_processor_id();
//獲得當前cpu的時間戳
unsigned long touch_ts = per_cpu(softlockup_touch_ts, this_cpu);
unsigned long print_ts;
//獲得進程當前的寄存器組
struct pt_regs *regs = get_irq_regs();
unsigned long now;
//如果沒有打開softlockup,就將這個cpu對應的時間戳清零
/* Is detection switched off? */
if (!per_cpu(softlockup_watchdog, this_cpu) || softlockup_thresh <= 0) {
/* Be sure we don
怎么樣,說白了,也不難,就是兩條線並行工作:一條線是fifo級別的內核線程負責清時間戳,另一條線是時鍾中斷定期檢查時間戳是否有被清過,若是到了閥值都沒有被請過,則打印softlockup的信息。
(三)長時間關中斷檢測
長時間關中斷檢測可以有幾種實現機制,而利用nmi watchdog來檢查這種長時間關中斷情況,是比較簡單的。其原理是需要軟硬件配合,硬件通常提供一個計數器(可以遞增也可以遞減),當記數到某個值得時候,系統就硬件復位。而nmi watchdog就定期(小於這個計數到達系統復位的時間)的去清一下系統的計數,若是某個進程長時間關中斷,則可能導致nmi watchdog得不到清,最終系統復位。