中斷（IRQ），尤其是軟中斷（softirq）的重要使用場景之一是網(wǎng)絡(luò)收發(fā)包， 但并未唯一場景。本文整理 IRQ/softirq 的通用基礎(chǔ)，這些東西和網(wǎng)絡(luò)收發(fā)包沒有直接關(guān)系， 雖然整理本文的直接目的是為了更好地理解網(wǎng)絡(luò)收發(fā)包。

什么是中斷？

CPU 通過時分復(fù)用來處理很多任務(wù)，這其中包括一些硬件任務(wù)，例如磁盤讀寫、鍵盤輸入，也包括一些軟件任務(wù)，例如網(wǎng)絡(luò)包處理。在任意時刻，一個 CPU 只能處理一個任務(wù)。當(dāng)某個硬件或軟件任務(wù)此刻沒有被執(zhí)行，但它希望 CPU 來立即處理時，就會給 CPU 發(fā)送一個中斷請求 —— 希望 CPU 停下手頭的工作，優(yōu)先服務(wù)“我”。中斷是以事件的方式通知 CPU 的，因此我們?？吹?“XX 條件下會觸發(fā) XX 中斷事件” 的表述。

兩種類型：

外部或硬件產(chǎn)生的中斷，例如鍵盤按鍵。

軟件產(chǎn)生的中斷，異常事件產(chǎn)生的中斷，例如除以零 。

管理中斷的設(shè)備：Advanced Programmable Interrupt Controller（APIC）。

硬中斷

中斷處理流程

中斷隨時可能發(fā)生，發(fā)生之后必須馬上得到處理。收到中斷事件后的處理流程：

搶占當(dāng)前任務(wù)：內(nèi)核必須暫停正在執(zhí)行的進程；

執(zhí)行中斷處理函數(shù)：找到對應(yīng)的中斷處理函數(shù)，將 CPU 交給它（執(zhí)行）；

中斷處理完成之后：第 1 步被搶占的進程恢復(fù)執(zhí)行。

Maskable and non-maskable

Maskable interrupts 在 x64_64 上可以用 sti/cli 兩個指令來屏蔽（關(guān)閉）和恢復(fù)：

staticinlinevoidnative_irq_disable(void){
asmvolatile("cli":::"memory");//清除IF標(biāo)志位
}
staticinlinevoidnative_irq_enable(void){
asmvolatile("sti":::"memory");//設(shè)置IF標(biāo)志位
}

在屏蔽期間，這種類型的中斷不會再觸發(fā)新的中斷事件。大部分 IRQ 都屬于這種類型。例子：網(wǎng)卡的收發(fā)包硬件中斷。

Non-maskable interrupts 不可屏蔽，所以在效果上屬于更緊急的類型。

問題：執(zhí)行足夠快 vs 邏輯比較復(fù)雜

IRQ handler 的兩個特點：

執(zhí)行要非常快，否則會導(dǎo)致事件（和數(shù)據(jù)）丟失；

需要做的事情可能非常多，邏輯很復(fù)雜，例如收包

這里就有了內(nèi)在矛盾。

解決方式：延后中斷處理（deferred interrupt handling）

傳統(tǒng)上，解決這個內(nèi)在矛盾的方式是將中斷處理分為兩部分：

top half

bottom half

這種方式稱為中斷的推遲處理或延后處理。以前這是唯一的推遲方式，但現(xiàn)在不是了。現(xiàn)在已經(jīng)是個通用術(shù)語，泛指各種推遲執(zhí)行中斷處理的方式。按這種方式，中斷會分為兩部分：

第一部分：只進行最重要、必須得在硬中斷上下文中執(zhí)行的部分；剩下的處理作為第二部分，放入一個待處理隊列；

第二部分：一般是調(diào)度器根據(jù)輕重緩急來調(diào)度執(zhí)行，不在硬中斷上下文中執(zhí)行。

Linux 中的三種推遲中斷（deferred interrupts）：

softirq

tasklet

workqueue

后面會具體介紹。

軟中斷

軟中斷子系統(tǒng)

軟中斷是一個內(nèi)核子系統(tǒng)：

1、每個 CPU 上會初始化一個 ksoftirqd 內(nèi)核線程，負責(zé)處理各種類型的 softirq 中斷事件；

用 cgroup ls 或者 ps -ef 都能看到：

$systemd-cgls-k|grepsoftirq#-k:includekernelthreadsintheoutput
├─12[ksoftirqd/0]
├─19[ksoftirqd/1]
├─24[ksoftirqd/2]
...

2、軟中斷事件的 handler 提前注冊到 softirq 子系統(tǒng)， 注冊方式 open_softirq(softirq_id, handler)

例如，注冊網(wǎng)卡收發(fā)包（RX/TX）軟中斷處理函數(shù)：

//net/core/dev.c

open_softirq(NET_TX_SOFTIRQ,net_tx_action);
open_softirq(NET_RX_SOFTIRQ,net_rx_action);

3、軟中斷占 CPU 的總開銷：可以用 top 查看，里面 si 字段就是系統(tǒng)的軟中斷開銷（第三行倒數(shù)第二個指標(biāo)）：

$top-n1|head-n3
top-1805up86days,23:45,2users,loadaverage:5.01,5.56,6.26
Tasks:969total,2running,733sleeping,0stopped,2zombie
%Cpu(s):13.9us,3.2sy,0.0ni,82.7id,0.0wa,0.0hi,0.1si,0.0st

主處理

smpboot.c 類似于一個事件驅(qū)動的循環(huán)，里面會調(diào)度到 ksoftirqd 線程，執(zhí)行 pending 的軟中斷。ksoftirqd 里面會進一步調(diào)用到 __do_softirq，

判斷哪些 softirq 需要處理，

執(zhí)行 softirq handler

避免軟中斷占用過多 CPU

軟中斷方式的潛在影響：推遲執(zhí)行部分（比如 softirq）可能會占用較長的時間，在這個時間段內(nèi)， 用戶空間線程只能等待。反映在 top 里面，就是 si 占比。

不過 softirq 調(diào)度循環(huán)對此也有改進，通過 budget 機制來避免 softirq 占用過久的 CPU 時間。

unsignedlongend=jiffies+MAX_SOFTIRQ_TIME;
...
restart:
while((softirq_bit=ffs(pending))){
...
h->action(h);//這里面其實也有機制，避免softirq占用太多CPU
...
}
...
pending=local_softirq_pending();
if(pending){
if(time_before(jiffies,end)&&!need_resched()&&--max_restart)//避免softirq占用太多CPU
gotorestart;
}
...

硬中斷 -> 軟中斷 調(diào)用棧

前面提到，softirq 是一種推遲中斷處理機制，將 IRQ 的大部分處理邏輯推遲到了這里執(zhí)行。兩條路徑都會執(zhí)行到 softirq 主處理邏輯 __do_softirq()，

1、CPU 調(diào)度到 ksoftirqd 線程時，會執(zhí)行到 __do_softirq()；

2、每次 IRQ handler 退出時：do_IRQ() -> ...。

do_IRQ() 是內(nèi)核中最主要的 IRQ 處理方式。它執(zhí)行結(jié)束時，會調(diào)用 exiting_irq()，這會展開成 irq_exit()。后者會檢查是pending 的 softirq，有的話就喚醒：

//arch/x86/kernel/irq.c

if(!in_interrupt()&&local_softirq_pending())
invoke_softirq();

進而會使 CPU 執(zhí)行到 __do_softirq()。

軟中斷觸發(fā)執(zhí)行的步驟

To summarize, each softirq goes through the following stages: 每個軟中斷會經(jīng)過下面幾個階段：

通過 open_softirq() 注冊軟中斷處理函數(shù)；

通過 raise_softirq() 將一個軟中斷標(biāo)記為 deferred interrupt，這會喚醒改軟中斷（但還沒有開始處理）；

內(nèi)核調(diào)度器調(diào)度到 ksoftirqd 內(nèi)核線程時，會將所有等待處理的 deferred interrupt（也就是 softirq）拿出來，執(zhí)行對應(yīng)的處理方法（softirq handler）；

以收包軟中斷為例， IRQ handler 并不執(zhí)行 NAPI，只是觸發(fā)它，在里面會執(zhí)行到 raise NET_RX_SOFTIRQ；真正的執(zhí)行在 softirq，里面會調(diào)用網(wǎng)卡的 poll() 方法收包。IRQ handler 中會調(diào)用 napi_schedule()，然后啟動 NAPI poll()，

這里需要注意，雖然 IRQ handler 做的事情非常少，但是接下來處理這個包的 softirq 和 IRQ 在同一個 CPU 運行。這就是說，如果大量的包都放到了同一個 RX queue，那雖然 IRQ 的開銷可能并不多，但這個 CPU 仍然會非常繁忙，都花在 softirq 上了。解決方式：RPS。它并不會降低延遲，只是將包重新分發(fā)：RXQ -> CPU。

三種推遲執(zhí)行方式（softirq/tasklet/workqueue）

前面提到，Linux 中的三種推遲中斷執(zhí)行的方式：

softirq

tasklet

workqueue

其中，

softirq 和 tasklet 依賴軟中斷子系統(tǒng)，運行在軟中斷上下文中；

workqueue 不依賴軟中斷子系統(tǒng)，運行在進程上下文中。

softirq

前面已經(jīng)看到， Linux 在每個 CPU 上會創(chuàng)建一個 ksoftirqd 內(nèi)核線程。

softirqs 是在 Linux 內(nèi)核編譯時就確定好的，例外網(wǎng)絡(luò)收包對應(yīng)的 NET_RX_SOFTIRQ 軟中斷。因此是一種靜態(tài)機制。如果想加一種新 softirq 類型，就需要修改并重新編譯內(nèi)核。

內(nèi)部組織

在內(nèi)部是用一個數(shù)組（或稱向量）來管理的，每個軟中斷號對應(yīng)一個 softirq handler。數(shù)組和注冊：

//kernel/softirq.c

//NR_SOFTIRQS是enumsoftirqtype的最大值，在5.10中是10，見下面
staticstructsoftirq_actionsoftirq_vec[NR_SOFTIRQS]__cacheline_aligned_in_smp;

voidopen_softirq(intnr,void(*action)(structsoftirq_action*)){
softirq_vec[nr].action=action;
}

5.10 中所有類型的 softirq：

//include/linux/interrupt.h

enum{
HI_SOFTIRQ=0,//tasklet
TIMER_SOFTIRQ,//timer
NET_TX_SOFTIRQ,//networking
NET_RX_SOFTIRQ,//networking
BLOCK_SOFTIRQ,//IO
IRQ_POLL_SOFTIRQ,
TASKLET_SOFTIRQ,//tasklet
SCHED_SOFTIRQ,//schedule
HRTIMER_SOFTIRQ,//timer
RCU_SOFTIRQ,//lock
NR_SOFTIRQS
};

也就是在 cat /proc/softirqs 看到的哪些。

$cat/proc/softirqs
CPU0CPU1...CPU46CPU47
HI:20...01
TIMER:443727467971...313696270110
NET_TX:5791965998...4228754840
NET_RX:287285262341...8110655244
BLOCK:2611564...268986463918
IRQ_POLL:00...00
TASKLET:98207...129122
SCHED:18544271124268...51548045332269
HRTIMER:1222468926...2549724272
RCU:1469356972856...59617375917455

觸發(fā)（喚醒）softirq

voidraise_softirq(unsignedintnr){
local_irq_save(flags);//關(guān)閉IRQ
raise_softirq_irqoff(nr);//喚醒ksoftirqd線程（但執(zhí)行不在這里，在ksoftirqd線程中）
local_irq_restore(flags);//打開IRQ
}
if(!in_interrupt())
wakeup_softirqd();

staticvoidwakeup_softirqd(void){
structtask_struct*tsk=__this_cpu_read(ksoftirqd);

if(tsk&&tsk->state!=TASK_RUNNING)
wake_up_process(tsk);
}

以收包軟中斷為例， IRQ handler 并不執(zhí)行 NAPI，只是觸發(fā)它，在里面會執(zhí)行到 raise NET_RX_SOFTIRQ；真正的執(zhí)行在 softirq，里面會調(diào)用網(wǎng)卡的 poll() 方法收包。IRQ handler 中會調(diào)用 napi_schedule()，然后啟動 NAPI poll()。

tasklet

如果對內(nèi)核源碼有一定了解就會發(fā)現(xiàn)，softirq 用到的地方非常少，原因之一就是上面提到的，它是靜態(tài)編譯的， 靠內(nèi)置的 ksoftirqd 線程來調(diào)度內(nèi)置的那 9 種 softirq。如果想新加一種，就得修改并重新編譯內(nèi)核， 所以開發(fā)成本非常高。

實際上，實現(xiàn)推遲執(zhí)行的更常用方式 tasklet。它構(gòu)建在 softirq 機制之上， 具體來說就是使用了上面提到的兩種 softirq：

HI_SOFTIRQ

TASKLET_SOFTIRQ

換句話說，tasklet 是可以在運行時（runtime）創(chuàng)建和初始化的 softirq，

void__initsoftirq_init(void){
for_each_possible_cpu(cpu){
per_cpu(tasklet_vec,cpu).tail=&per_cpu(tasklet_vec,cpu).head;
per_cpu(tasklet_hi_vec,cpu).tail=&per_cpu(tasklet_hi_vec,cpu).head;
}

open_softirq(TASKLET_SOFTIRQ,tasklet_action);
open_softirq(HI_SOFTIRQ,tasklet_hi_action);
}

內(nèi)核軟中斷子系統(tǒng)初始化了兩個 per-cpu 變量：

tasklet_vec：普通 tasklet，回調(diào) tasklet_action()

tasklet_hi_vec：高優(yōu)先級 tasklet，回調(diào) tasklet_hi_action()

structtasklet_struct{
structtasklet_struct*next;
unsignedlongstate;
atomic_tcount;
void(*func)(unsignedlong);
unsignedlongdata;
};

tasklet 再執(zhí)行針對 list 的循環(huán)：

staticvoidtasklet_action(structsoftirq_action*a)
{
local_irq_disable();
list=__this_cpu_read(tasklet_vec.head);
__this_cpu_write(tasklet_vec.head,NULL);
__this_cpu_write(tasklet_vec.tail,this_cpu_ptr(&tasklet_vec.head));
local_irq_enable();

while(list){
if(tasklet_trylock(t)){
t->func(t->data);
tasklet_unlock(t);
}
...
}
}

tasklet 在內(nèi)核中的使用非常廣泛。不過，后面又出現(xiàn)了第三種方式：workqueue。

workqueue

這也是一種推遲執(zhí)行機制，與 tasklet 有點類似，但也有很大不同。

tasklet 是運行在 softirq 上下文中；

workqueue 運行在內(nèi)核進程上下文中；這意味著 wq 不能像 tasklet 那樣是原子的；

tasklet 永遠運行在指定 CPU，這是初始化時就確定了的；

workqueue 默認行為也是這樣，但是可以通過配置修改這種行為。

使用場景

// Documentation/core-api/workqueue.rst：

Therearemanycaseswhereanasynchronousprocessexecutioncontext
isneededandtheworkqueue(wq)APIisthemostcommonlyused
mechanismforsuchcases.

Whensuchanasynchronousexecutioncontextisneeded,aworkitem
describingwhichfunctiontoexecuteisputonaqueue.An
independentthreadservesastheasynchronousexecutioncontext.The
queueiscalledworkqueueandthethreadiscalledworker.

Whilethereareworkitemsontheworkqueuetheworkerexecutesthe
functionsassociatedwiththeworkitemsoneaftertheother.When
thereisnoworkitemleftontheworkqueuetheworkerbecomesidle.
Whenanewworkitemgetsqueued,theworkerbeginsexecutingagain.

簡單來說，workqueue 子系統(tǒng)提供了一個接口，通過這個接口可以創(chuàng)建內(nèi)核線程來處理從其他地方 enqueue 過來的任務(wù)。這些內(nèi)核線程就稱為 worker threads，內(nèi)置的 per-cpu worker threads：

$systemd-cgls-k|grepkworker
├─5[kworker/0:0H]
├─15[kworker/1:0H]
├─20[kworker/2:0H]
├─25[kworker/3:0H]

結(jié)構(gòu)體

//include/linux/workqueue.h

structworker_pool{
spinlock_tlock;
intcpu;
intnode;
intid;
unsignedintflags;

structlist_headworklist;
intnr_workers;
...

structwork_struct{
atomic_long_tdata;
structlist_headentry;
work_func_tfunc;
structlockdep_maplockdep_map;
};

kworker 線程調(diào)度 workqueues，原理與 ksoftirqd 線程調(diào)度 softirqs 一樣。但是我們可以為 workqueue 創(chuàng)建新的線程，而 softirq 則不行。