程式師世界 >> 編程語言 >> JAVA編程 >> JAVA綜合教程 >> Linux系統中的知名內核線程(1)——ksoftirqd和events

Linux系統中的知名內核線程(1)——ksoftirqd和events

編輯：JAVA綜合教程

Linux系統中的知名內核線程(1)——ksoftirqd和events

Linux系統中的知名內核線程(1)——ksoftirqd和events
——lvyilong316

我們知道linux系統中有很多系統創建的內核線程（kthread），這些內核線程是系統正常工作的保證。這裡我們看下其中比較知名的兩個：ksoftirqd和events。

1.ksoftirqd

提到ksoftirqd就不得不說下“軟中斷（softirq）”，因為這個線程正是用來執行軟中斷的（准確的說應該是執行過多的軟中斷）。我們知道按照優先級來說，中斷>軟中斷>用戶進行，也就是說中斷可以打斷軟中斷，而軟中斷又可以打斷用戶進程。

而對於軟中斷，內核會在幾個特殊的時機執行（注意執行和調度的區別，調度軟中斷只是對軟中斷打上待執行的標記，並沒有真正執行），而在中斷處理程序返回時處理是最常見的。軟中斷的觸發頻率有時可能會很高（例如進行大流量網絡通信期間）。更不利的是，軟中斷的執行函數有時還會調度自身，所以如果軟中斷本身出現的頻率較高，再加上他們又有將自己重新設置為可執行狀態的能力，那麼就會導致用戶空間的進程無法獲得足夠的處理時間，因而處於饑餓狀態。為了避免用戶進程的饑餓。內核開發者做了一些折中，最終在內核的實現方案中是不會立即處理由軟中斷自身重新觸發的軟中斷（不允許軟中斷嵌套）。而作為改進，內核會喚醒一組內核線程來處理這些過多的軟中斷，這些內核線程在最低優先級上運行（nice值是19），這能避免它們跟其他重要的任務搶奪資源，但它們最終肯定會被執行，所以這個方案能夠保證軟中斷負載很重的時候，用戶進程不會因為得不到處理時間而處於饑餓狀態，相應的，也能保證過量的軟中斷終究會得到處理。

每個處理器都有一個這樣的線程。所有的線程的名字都叫做ksoftirq/n，區別在於n，他對應的是處理器的編號。

下面我們詳細的看下軟中斷是如何被ksoftirqd執行的。首先看下軟中斷的處理調度過程。一個軟中斷在執行之前必須被調度（激活），術語稱為"raisethesoftirq"。被激活的softirq通常並不會立即執行，一般會在之後的某個時刻檢查當前系統中是否有被pending的softirq，如果有就去執行，linux執行軟中斷的函數是do_softirq()，而這個函數會再兩個地方被調用，一個是中斷返回時，另一個就是我們討論的ksoftirqd內核線程。我們先來看中斷返回的情況。

1.1irq_exit

//do_IRQ函數執行完硬件ISR後退出時調用此函數。

void irq_exit(void) 
{ 
account_system_vtime(current); 
trace_hardirq_exit(); 
sub_preempt_count(IRQ_EXIT_OFFSET); //這個位置修改preempt_count
// 判斷當前是否有硬件中斷嵌套，並且是否有軟中斷在 pending 狀態，注意：這裡只有兩個條件同時滿足時，才有可能調用 do_softirq() 進入軟中斷。也就是說確認當前所有硬件中斷處理完成，且有硬件中斷安裝了軟中斷處理時理時才會進入。關於in_interrupt()後面會詳細分析。
if (!in_interrupt() && local_softirq_pending()) 
// 其實這裡就是調用 do_softirq() 執行 
invoke_softirq(); 
preempt_enable_no_resched(); 
}

1.2in_interrupt

這裡需要重點分析一下in_interrupt()函數的含義。在linux內核中，為了方便判斷當前執行路徑在哪個上下文環境中，定義了幾個接口：

#define hardirq_count() (preempt_count() & HARDIRQ_MASK)
#define softirq_count() (preempt_count() & SOFTIRQ_MASK)
#define irq_count() (preempt_count() & (HARDIRQ_MASK | SOFTIRQ_MASK | NMI_MASK))
/*
* Are we doing bottom half or hardware interrupt processing?
* Are we in a softirq context? Interrupt context?
*/
#define in_irq() (hardirq_count())
#define in_softirq() (softirq_count())
#define in_interrupt() (irq_count())
/*
* Are we in NMI context?
*/
#define in_nmi() (preempt_count() & NMI_MASK)

從注釋可以看出包括：硬件中斷上下文，軟件中斷上下文，不可屏蔽上下文等。在這些宏中，都涉及到了preempt_count()這個宏，這個宏是一個比較重要的宏，在Linux源碼中對其做了詳細的注釋：

/*
* We put the hardirq and softirq counter into the preemption
* counter. The bitmask has the following meaning:
*
* - bits 0-7 are the preemption count (max preemption depth: 256)
* - bits 8-15 are the softirq count (max # of softirqs: 256)
*
* The hardirq count can in theory reach the same as NR_IRQS.
* In reality, the number of nested IRQS is limited to the stack
* size as well. For archs with over 1000 IRQS it is not practical
* to expect that they will all nest. We give a max of 10 bits for
* hardirq nesting. An arch may choose to give less than 10 bits.
* m68k expects it to be 8.
*
* - bits 16-25 are the hardirq count (max # of nested hardirqs: 1024)
* - bit 26 is the NMI_MASK
* - bit 28 is the PREEMPT_ACTIVE flag
*
* PREEMPT_MASK: 0x000000ff
* SOFTIRQ_MASK: 0x0000ff00
* HARDIRQ_MASK: 0x03ff0000
* NMI_MASK: 0x04000000
*/

從注釋可以看出，preempt_count各個bit位的含義：

（1）bit0~7位表示搶占計數，即支持最大的搶占深度為256

（2）bit8~15位表示軟中斷計數，即支持最大的軟中斷的個數為256，需要注意的是，由於軟中斷還受制於pending狀態，一個32位的變量，因此實際最大只能支持32個軟中斷。

（3）bit16~25位表示硬件中斷嵌套層數，即最大可支持的嵌套層次為1024，實際情況下這是不可能的，因為中斷的嵌套層數還受制於中斷處理的棧空間的大小。

介紹了這麼多，現在了重點分析下上面提到的in_interrupt到底表示什麼意思？

#define in_interrupt() (irq_count())
#define irq_count() (preempt_count() & (HARDIRQ_MASK | SOFTIRQ_MASK \
| NMI_MASK))

從其宏定義可以看出，in_interrupt宏的值是硬件中斷嵌套層數，軟中斷計數以及可屏蔽中斷三者之和。所以如果in_interrupt()的值大於0，就不會處理軟中斷，意思是(a)當有硬件中斷嵌套，(b)或軟中斷被禁止(c)不可屏蔽中斷的情況下，不會去處理軟中斷。有人會問軟中斷不是從中斷處理後irq_exit中進入的嗎？那軟中斷執行時preempt_count的硬中斷位不是還沒有修改掉嗎？其實已經做了修改，就在irq_exit函數中的sub_preempt_count中進行。其實執行過sub_preempt_count就算退出中斷處理程序了。

l注：軟中斷被禁止會增加軟中斷的計數；

__local_bh_disable((unsigned long)__builtin_return_address(0));
static inline void __local_bh_disable(unsigned long ip)
{
add_preempt_count(SOFTIRQ_OFFSET);
barrier();
}
# define add_preempt_count(val) do { preempt_count() += (val); } while (0)

1.3do_softirq

下面重點分析以下do_softirq()，了解Linux內核到底是怎麼來處理softirq的。

asmlinkage void do_softirq(void)
{
__u32 pending;
 unsigned long flags;
// 這個函數判斷，如果當前有硬件中斷嵌套，或者軟中斷被禁止時，則馬上返回。在這個入口判斷主要是為了與 ksoftirqd 互斥。
if (in_interrupt())
   return;
// 關中斷執行以下代碼
 local_irq_save(flags); 
// 判斷是否有 pending 的軟中斷需要處理。
 pending = local_softirq_pending();
// 如果有則調用 __do_softirq() 進行實際處理
if (pending)
   __do_softirq();
// 開中斷繼續執行
 local_irq_restore(flags);
}

注意調用local_softirq_pending()獲取pending和將pending請0這兩個操作一定要位於關中斷情況，否則兩個操作間可能發生中斷，中斷再次調度軟中段的置位標記會丟失。

真正的軟中斷處理再__do_softirq中。

1.4__do_softirq

// 最大軟中斷調用次數為 10 次。
#define MAX_SOFTIRQ_RESTART 10
asmlinkage void __do_softirq(void)
{
// 軟件中斷處理結構，此結構中包括軟中斷回調函數。
  struct softirq_action *h;
  __u32 pending;
int max_restart = MAX_SOFTIRQ_RESTART;
int cpu;
// 得到當前所有 pending 的軟中斷。
  pending = local_softirq_pending();
  account_system_vtime(current);
// 執行到這裡要禁止其他軟中斷，這裡也就證明了每個 CPU 上同時運行的軟中斷只能有一個。
  __local_bh_disable((unsigned long)__builtin_return_address(0));
  trace_softirq_enter();
// 針對 SMP 得到當前正在處理的 CPU
  cpu = smp_processor_id();
restart:
// 每次循環在允許硬件 中斷強占前，首先重置軟中斷的標志位。
/* Reset the pending bitmask before enabling irqs */
  set_softirq_pending(0); //要在關中斷情況下才能調用
 // 到這裡才開中斷運行，注意：以前運行狀態一直是關中斷運行，這時當前處理軟中斷才可能被硬件中斷搶占。也就是說在進入軟中斷時不是一開始就會被硬件中斷搶占。只有在這裡以後的代碼才可能被硬件中斷搶占。
  local_irq_enable();
 // 這裡要注意，以下代碼運行時可以被硬件中斷搶占，但這個硬件中斷執行完成後，它的所注冊的軟中斷無法馬上運行，別忘了，現在雖是開硬件中斷執行，但前面的 __local_bh_disable()函數屏蔽了軟中斷。所以這種環境下只能被硬件中斷搶占，但這個硬中斷注冊的軟中斷回調函數無法運行。要問為什麼，那是因為__local_bh_disable() 函數設置了一個標志當作互斥量，而這個標志正是上面的 irq_exit() 和 do_softirq() 函數中的in_interrupt() 函數判斷的條件之一，也就是說 in_interrupt() 函數不僅檢測硬中斷而且還判斷了軟中斷。所以在這個環境下觸發硬中斷時注冊的軟中斷，根本無法重新進入到這個函數中來，只能是做一個標志，等待下面的重復循環（最大 MAX_SOFTIRQ_RESTART）才可能處理到這個時候觸發的硬件中斷所注冊的軟中斷。得到軟中斷向量表。
  h = softirq_vec;
 // 循環處理所有 softirq 軟中斷注冊函數。
 do {
// 如果對應的軟中斷設置 pending 標志則表明需要進一步處理它所注冊的函數。
if (pending & 1) {
// 在這裡執行了這個軟中斷所注冊的回調函數。
      h->action(h);
      rcu_bh_qsctr_inc(cpu);
}
// 繼續找，直到把軟中斷向量表中所有 pending 的軟中斷處理完成。
     h++;
     // 從代碼裡可以看出按位操作，表明一次循環只處理 32 個軟中斷的回調函數。
      pending >>= 1; 
} while (pending);
// 關中斷執行以下代碼。注意：這裡又關中斷了，下面的代碼執行過程中硬件中斷無法搶占。
  local_irq_disable();
// 前面提到過，在剛才開硬件中斷執行環境時只能被硬件中斷搶占 ，在這個時候是無法處理軟中斷的，因為剛才開中斷執行過程中可能多次被硬件中斷搶占，每搶占一次就有可能注冊一個軟中斷，所以要再重新取一次所有的軟中斷。以便下面的代碼進行處理後跳回到 restart 處重復執行。 
  pending = local_softirq_pending(); 
// 如果在上面的開中斷執行環境中觸發了硬件中斷，且注冊了一個軟中斷的話，這個軟中斷會設置 pending 位，但在當前一直屏蔽軟中斷的環境下無法得到執行，前面提到過，因為 irq_exit() 和 do_softirq() 根本無法進入到這個處理過程中來。這個在上面周詳的記錄過了。那麼在這裡又有了一個執行的機會。注意：雖然當前環境一直是處於屏蔽軟中斷執行的環境中，但在這裡又給出了一個執行剛才在開中斷環境過程中觸發硬件中斷時所注冊的軟中斷的機會，其實只要理解了軟中斷機制就會知道，無非是在一些特定環境下調用 ISR 注冊到軟中斷向量表裡的函數而已。如果剛才觸發的硬件中斷注冊了軟中斷，並且重復執行次數沒有到 10 次的話，那麼則跳轉到 restart 標志處重復以上所介紹的所有步驟：設置軟中斷標志位，重新開中斷執行... 
// 注意：這裡是要兩個條件都滿足的情況下才可能重復以上步驟。 
if (pending && --max_restart) 
    goto restart; 
// 如果以上步驟重復了 10 次後還有 pending 的軟中斷的話，那麼系統在一定時間內可能達到了一個峰值，為了平衡這點。系統專門建立了一個 ksoftirqd 線程來處理，這樣避免在一 定時間內負荷太大。這個 ksoftirqd 線程本身是個大循環，在某些條件下為了不負載過重，他是能被其他進程搶占的，但注意，他是顯示的調用了 preempt_xxx() 和 schedule()才會被搶占和轉換的。這麼做的原因是因為在他一旦調用 local_softirq_pending() 函數檢測到有 pending 的軟中斷需要處理的時候，則會顯示的調用 do_softirq() 來處理軟中 斷。也就是說，下面代碼喚醒的 ksoftirqd 線程有可能會回到這個函數當中來，尤其是在系統需要響應非常多軟中斷的情況下，他的調用入口是 do_softirq()，這也就是為什麼在 do_softirq() 的入口處也會用 in_interrupt() 函數來判斷是否有軟中斷正在處理的原因了，目的還是為了防止重入。ksoftirqd 實現看下面對 ksoftirqd() 函數的分析。 
if (pending) 
// 此函數實際是調用 wake_up_process() 來喚醒 ksoftirqd 
   wakeup_softirqd(); 
trace_softirq_exit(); 
account_system_vtime(current); 
// 到最後才開軟中斷執行環境，允許軟中斷執行。注意：這裡使用的不是 local_bh_enable()，不會再次觸發 do_softirq()的調用。 
_local_bh_enable(); 
}

1.5ksoftirqd

這個函數就是ksoftirqd內核線程對應的執行函數。只要有待處理的軟中斷（由softirq_pending()函數負責發現），ksoftirq就會調用do_softirq()去處理它們。通過重復執行這樣的操作，重新觸發的軟中斷也會被執行。如果有必要的話，每次迭代後都會調用schedule()以便讓更重要的進程得到處理機會。當所有需要執行的操作都完成以後，該內核線程將自己設置為TASK_INTERTUPTIBLE狀態，喚起調度程序選擇其他可執行的進程投入運行。

static int ksoftirqd(void * __bind_cpu)
{
// 顯示調用此函數設置當前進程的靜態優先級。當然，這個優先級會隨調度器策略而變化。
set_user_nice(current, 19);
// 設置當前進程不允許被掛啟
current->flags |= PF_NOFREEZE;
//設置當前進程狀態為可中斷的狀態，這種睡眠狀態可響應信號處理等。
set_current_state(TASK_INTERRUPTIBLE);
// 下面是個大循環，循環判斷當前進程是否會停止，不會則繼續判斷當前是否有 pending 的軟中斷需要處理。
while (!kthread_should_stop()) {
   // 如果能進行處理，那麼在此處理期間內禁止當前進程被搶占。
   preempt_disable();
   // 首先判斷系統當前沒有需要處理的 pending 狀態的軟中斷
   if (!local_softirq_pending()) {
// 沒有的話在主動放棄 CPU 前先要允許搶占，因為一直是在不允許搶占狀態下執行的代碼。
preempt_enable_no_resched();
// 顯示調用此函數主動放棄 CPU 將當前進程放入睡眠隊列，並轉換新的進程執行（調度器相關不記錄在此）
schedule();
// 注意：如果當前顯示調用 schedule() 函數主動轉換的進程再次被調度執行的話，那麼將從調用這個函數的下一條語句開始執行。也就是說，在這裡當前進程再次被執行的話，將會執行下面的 preempt_disable() 函數。當進程再度被調度時，在以下處理期間內禁止當前進程被搶占。
preempt_disable();
}
/*設置當前進程為運行狀態。注意：已設置了當前進程不可搶占在進入循環後，以上兩個分支不論走哪個都會執行到這裡。一是進入循環時就有 pending 的軟中斷需要執行時。二是進入循環時沒有 pending 的軟中斷，當前進程再次被調度獲得 CPU 時繼續執行時。*/
__set_current_state(TASK_RUNNING);
/* 循環判斷是否有 pending 的軟中斷，如果有則調用 do_softirq()來做具體處理。注意：這裡又是個 do_softirq() 的入口點，那麼在 __do_softirq() 當中循環處理 10 次軟中斷的回調函數後，如果更有 pending 的話，會又調用到這裡。那麼在這裡則又會有可能去調用 __do_softirq() 來處理軟中斷回調函數。在前面介紹 __do_softirq() 時已提到過，處理 10 次還處理不完的話說明系統正處於繁忙狀態。根據以上分析，我們能試想如果在系統非常繁忙時，這個進程將會和 do_softirq() 相互交替執行，這時此進程占用 CPU 應該會非常高，雖然下面的 cond_resched()函數做了一些處理，他在處理完一輪軟中斷後當前處理進程可能會因被調度而減少 CPU 負荷，不過在非常繁忙時這個進程仍然有可能大量占用 CPU。*/
while (local_softirq_pending()) {
/* Preempt disable stops cpu going offline. If already offline, we’ll be on wrong CPU: don’t process */
if (cpu_is_offline((long)__bind_cpu))
/*如果當前被關聯的 CPU 無法繼續處理則跳轉到 wait_to_die 標記出，等待結束並退出。*/
goto wait_to_die;
/*執行 do_softirq() 來處理具體的軟中斷回調函數。注意：如果此時有一個正在處理的軟中斷的話，則會馬上返回，還記得前面介紹的 in_interrupt() 函數麼。*/
do_softirq();
/*允許當前進程被搶占。*/
preempt_enable_no_resched();
/*這個函數有可能間接的調用 schedule() 來轉換當前進程，而且上面已允許當前進程可被搶占。也就是說在處理完一輪軟中斷回調函數時，有可能會轉換到其他進程。我認為這樣做的目的一是為了在某些負載超標的情況下不至於讓這個進程長時間大量的占用 CPU，二是讓在有非常多軟中斷需要處理時不至於讓其他進程得不到響應。*/
        cond_resched();
       /* 禁止當前進程被搶占。*/
preempt_disable();
/* 處理完所有軟中斷了嗎？沒有的話繼續循環以上步驟*/
     }
/*待一切都處理完成後，允許當前進程被搶占，並設置當前進程狀態為可中斷狀態，繼續循環以上所有過程。*/
     preempt_enable();
     set_current_state(TASK_INTERRUPTIBLE);
}
/*如果將會停止則設置當前進程為運行狀態後直接返回。調度器會根據優先級來使當前進程運行。*/
__set_current_state(TASK_RUNNING);
return 0;
/*一直等待到當前進程被停止*/
wait_to_die:
/*允許當前進程被搶占。*/
preempt_enable();
/* Wait for kthread_stop */
/*設置當前進程狀態為可中斷的狀態，這種睡眠狀態可響應信號處理等。*/
set_current_state(TASK_INTERRUPTIBLE);
/*判斷當前進程是否會被停止，如果不是的話則設置進程狀態為可中斷狀態並放棄當前 CPU主動轉換。也就是說這裡將一直等待當前進程將被停止時候才結束。*/
while (!kthread_should_stop()) {
schedule();
set_current_state(TASK_INTERRUPTIBLE);
}
/*如果將會停止則設置當前進程為運行狀態後直接返回。調度器會根據優先級來使當前進程運行。*/
__set_current_state(TASK_RUNNING);
return 0;
}

最後說明一下，因為tasklet也是通過軟中斷實現的，所以tasklet過多也會導致ksoftirqd線程的調度，進而再進程上下文中執行tasklet。（ksoftirqd執行軟中斷處理程序，tasklet對應的軟中斷處理程序執行所有調度的tasklet）

2.events

下面看events線程，提到這個線程就不得不說道“工作隊列（workqueue）”了，這個線程是就是工作隊了用來執行隊列中的工作的。

2.1什麼是workqueue？

Workqueue也是linux下半部（包括軟中斷、tasklet、工作隊列）實現的一種方式。Linux中的Workqueue機制就是為了簡化內核線程的創建。通過調用workqueue的接口就能創建內核線程。並且可以根據當前系統CPU的個數創建線程的數量，使得線程處理的事務能夠並行化。

Workqueue是內核中實現簡單而有效的機制，他顯然簡化了內核daemon的創建，方便了用戶的編程。

2.2Workqueue機制的實現

Workqueue機制中定義了兩個重要的數據結構，分析如下：

1．cpu_workqueue_struct結構。該結構將CPU和內核線程進行了綁定。在創建workqueue的過程中，Linux根據當前系統CPU的個數創建cpu_workqueue_struct。在該結構主要維護了一個任務（work_struct）隊列，以及內核線程需要睡眠的等待隊列，另外還維護了一個任務上下文，即task_struct。

2.work_struct結構是對任務的抽象。在該結構中需要維護具體的任務方法，需要處理的數據，以及任務處理的時間。該結構定義如下：

struct work_struct {
 unsigned long pending;
  struct list_head entry; /* 將任務掛載到queue的掛載點 */
  void (*func)(void *); /* 任務方法 */
  void *data; /* 任務處理的數據*/
  void *wq_data; /* work的屬主 */
  strut timer_list timer; /* 任務延時處理定時器 */
};

當用戶調用workqueue的初始化接口create_workqueue或者create_singlethread_workqueue對workqueue隊列進行初始化時，內核就開始為用戶分配一個workqueue對象，並且將其鏈到一個全局的workqueue隊列中。然後Linux根據當前CPU的情況，為workqueue對象分配與CPU個數相同的cpu_workqueue_struct對象，每個cpu_workqueue_struct對象都會存在一條任務隊列。緊接著，Linux為每個cpu_workqueue_struct對象分配一個內核thread，即內核daemon去處理每個隊列中的任務。至此，用戶調用初始化接口將workqueue初始化完畢，返回workqueue的指針。

在初始化workqueue過程中，內核需要初始化內核線程，注冊的內核線程工作比較簡單，就是不斷的掃描對應cpu_workqueue_struct中的任務隊列，從中獲取一個有效任務，然後執行該任務。所以如果任務隊列為空，那麼內核daemon就在cpu_workqueue_struct中的等待隊列上睡眠，直到有人喚醒daemon去處理任務隊列。

Workqueue初始化完畢之後，將任務運行的上下文環境構建起來了，但是具體還沒有可執行的任務，所以，需要定義具體的work_struct對象。然後將work_struct加入到任務隊列中，Linux會喚醒daemon去處理任務。

上述描述的workqueue內核實現原理可以描述如下：