网卡每秒收百万个包、磁盘每次 I/O 完成都要通知内核——如果每次都现场完整处理，CPU 会被中断吞没。Linux 的解法是把中断切成两半：上半部只做最紧急的应答，真正的重活推给下半部延迟执行。本文用一个 120 行的 Python 模拟器复现上半部、三种下半部（软中断/tasklet、工作队列、线程化 IRQ）、共享中断线与 NAPI 风暴抑制的完整行为，并用真实运行输出验证每个机制。

一、为什么要切成两半

中断处理函数运行在中断上下文，受到两个硬约束：当前 CPU 的同类中断被屏蔽（甚至全部中断关闭），且绝对不能睡眠——没有进程上下文可切走。如果在中断里做协议解析、内存分配、磁盘 I/O 这类慢活，整个系统的中断响应都会瘫痪。于是一切中断处理都遵循同一模式：上半部（top half）只应答硬件、搬最少的数据，几微秒内返回；下半部（bottom half）在稍后安全的环境完成剩余工作。

二、上半部：最紧急的事

上半部由 request_irq() 注册，触发时按 IRQ 号查到处理链执行。它的职责清单极短：读设备状态寄存器确认中断源、给硬件写应答位（否则设备会持续拉高中断线）、把数据从设备寄存器/DMA 缓冲区搬到内存、标记下半部待执行。耗时通常 1~5 微秒。

三、三种下半部：按场景选

四、完整可运行模拟器（Python 3，约 120 行）

下面的程序复现四个场景：网卡收包（tasklet）、磁盘 I/O 完成（工作队列）、共享中断线遍历、NAPI 中断风暴抑制：

#!/usr/bin/env python3
"""中断处理模型模拟：上半部应答、下半部延迟执行、
   共享 IRQ 遍历、NAPI 风暴抑制"""

log = []
T = [0.0]          # 虚拟时钟（微秒）

def tick(us):
    T[0] += us

class SoftIRQQueue:
    def __init__(self):
        self.pending = []
    def raise_softirq(self, fn, tag):
        self.pending.append((fn, tag))
        log.append('  [t=%.1f] 上半部标记 %s 待执行' % (T[0], tag))
    def run_pending(self):
        """上半部返回后、恢复用户态前，内核处理软中断"""
        while self.pending:
            fn, tag = self.pending.pop(0)
            log.append('  [t=%.1f] 下半部执行 %s'
                       '（软中断上下文，不可睡眠）' % (T[0], tag))
            fn()

softirq = SoftIRQQueue()
processed_packets = []

# ---------- 网卡中断：上半部 + tasklet 下半部 ----------
def nic_top_half(pkt):
    log.append('[t=%.1f] IRQ 11 触发：网卡收到包 %s' % (T[0], pkt))
    log.append('  [t=%.1f] 上半部：读 NIC 状态寄存器、'
               '应答中断（关中断，耗时 2us）' % T[0])
    tick(2)
    def bottom():
        processed_packets.append(pkt)
        log.append('    [t=%.1f] 协议栈处理 %s 完成，'
                   '移交 socket 缓冲' % (T[0], pkt))
    softirq.raise_softirq(bottom, 'NET_RX_SOFTIRQ(tasklet)')
    log.append('  [t=%.1f] 上半部返回，恢复中断' % T[0])

# ---------- 磁盘中断：上半部 + 工作队列（可睡眠） ----------
class WorkQueue:
    def __init__(self):
        self.q = []
    def schedule(self, fn, tag):
        self.q.append((fn, tag))
        log.append('  [t=%.1f] 上半部投递工作项 %s' % (T[0], tag))
    def run(self):
        while self.q:
            fn, tag = self.q.pop(0)
            log.append('  [t=%.1f] kworker 线程执行 %s'
                       '（进程上下文，可睡眠/可取锁）' % (T[0], tag))
            fn()

wq = WorkQueue()
disk_done = []

def disk_top_half(req):
    log.append('[t=%.1f] IRQ 14 触发：磁盘完成请求 %s' % (T[0], req))
    tick(1.5)
    def work():
        disk_done.append(req)
        log.append('    [t=%.1f] 完成 %s 的页缓存回写收尾'
                   '（含可能睡眠的 I/O 操作）' % (T[0], req))
    wq.schedule(work, 'blk-complete-work')

# ---------- 共享中断线 ----------
def shared_irq_handler(devices, irq=11):
    """共享 IRQ：逐个询问设备是否是自己的中断"""
    log.append('[t=%.1f] 共享 IRQ %d 触发，遍历设备链：' % (T[0], irq))
    for name, pending in devices:
        if pending:
            log.append('  %s: 是我的中断 -> 处理' % name)
            return name
        else:
            log.append('  %s: 状态寄存器无挂起 -> 跳过' % name)
    return None

# ---------- NAPI 风暴抑制 ----------
def napi_demo(n_packets):
    log.append('=== NAPI 中断风暴抑制（%d 个包连续到达） ==='
               % n_packets)
    log.append('第 1 个包：触发硬中断，上半部关闭 RX 中断、'
               '切换轮询模式')
    budget = 8
    napi_runs = (n_packets + budget - 1) // budget
    for i in range(napi_runs):
        got = min(budget, n_packets - i * budget)
        log.append('  NAPI 轮询第 %d 轮：收包 %d 个'
                   '（budget=%d），无中断开销' % (i + 1, got, budget))
    log.append('队列清空：重新开启 RX 中断。'
               '总硬中断次数 = 1（而非 %d）' % n_packets)

print('=== 中断处理模型模拟 ===')
log.append('--- 场景 1: 网卡收包（tasklet 下半部） ---')
nic_top_half('pkt#1')
softirq.run_pending()

log.append('--- 场景 2: 磁盘 I/O 完成（工作队列下半部） ---')
disk_top_half('req#42')
log.append('[t=%.1f] 上半部返回，kworker 线程随后被调度' % T[0])
tick(5)
wq.run()

log.append('--- 场景 3: 共享中断线 ---')
shared_irq_handler([('eth0', False), ('eth1', True),
                    ('sata', False)])

log.append('--- 场景 4: NAPI ---')
napi_demo(20)

for line in log:
    print(line)
print()
print('结果汇总：协议栈处理包 %s，磁盘完成 %s'
      % (processed_packets, disk_done))

五、真实运行输出与解读

以上代码在 Python 3 下原样运行，输出如下：

=== 中断处理模型模拟 ===
--- 场景 1: 网卡收包（tasklet 下半部） ---
[t=0.0] IRQ 11 触发：网卡收到包 pkt#1
  [t=0.0] 上半部：读 NIC 状态寄存器、应答中断（关中断，耗时 2us）
  [t=2.0] 上半部标记 NET_RX_SOFTIRQ(tasklet) 待执行
  [t=2.0] 上半部返回，恢复中断
  [t=2.0] 下半部执行 NET_RX_SOFTIRQ(tasklet)（软中断上下文，不可睡眠）
    [t=2.0] 协议栈处理 pkt#1 完成，移交 socket 缓冲
--- 场景 2: 磁盘 I/O 完成（工作队列下半部） ---
[t=2.0] IRQ 14 触发：磁盘完成请求 req#42
  [t=3.5] 上半部投递工作项 blk-complete-work
[t=3.5] 上半部返回，kworker 线程随后被调度
  [t=8.5] kworker 线程执行 blk-complete-work（进程上下文，可睡眠/可取锁）
    [t=8.5] 完成 req#42 的页缓存回写收尾（含可能睡眠的 I/O 操作）
--- 场景 3: 共享中断线 ---
[t=8.5] 共享 IRQ 11 触发，遍历设备链：
  eth0: 状态寄存器无挂起 -> 跳过
  eth1: 是我的中断 -> 处理
--- 场景 4: NAPI ---
=== NAPI 中断风暴抑制（20 个包连续到达） ===
第 1 个包：触发硬中断，上半部关闭 RX 中断、切换轮询模式
  NAPI 轮询第 1 轮：收包 8 个（budget=8），无中断开销
  NAPI 轮询第 2 轮：收包 8 个（budget=8），无中断开销
  NAPI 轮询第 3 轮：收包 4 个（budget=8），无中断开销
队列清空：重新开启 RX 中断。总硬中断次数 = 1（而非 20）

逐场景解读：

六、理论分析

6.1 中断上下文为什么不能睡眠

睡眠意味着调用 schedule() 切走，而 schedule() 要保存『当前进程』的上下文——中断上下文里 current 指针借用的是被打断进程的 task_struct，自身没有独立的可调度实体。一旦在中断里睡眠，被打断的进程状态会被破坏，且中断返回路径无处恢复，内核直接 panic（『scheduling while atomic』）。这条红线决定了一切下半部机制的存在意义。

6.2 软中断的执行时机

内核在三个时机检查并执行待处理软中断：硬中断返回时（irq_exit）、ksoftirqd 内核线程被唤醒时、local_bh_enable 等显式开启点。为避免软中断饿死用户态（网络风暴下协议栈软中断反复触发），内核限制连续执行次数与时长（MAX_SOFTIRQ_TIME 约 2ms），超出的部分甩给每 CPU 的 ksoftirqd 线程——这是『延迟与公平』的又一重平衡。

6.3 中断亲和性与多核扩展

单队列网卡的所有中断都砸在一个 CPU 上会成为瓶颈；多队列网卡 + MSI-X 给每个队列分配独立中断向量，配合 irqbalance 或手工 /proc/irq/N/smp_affinity 把队列绑到不同核，中断负载随核数线性扩展。RSS（Receive Side Scaling）用五元组哈希把流散到各队列，保证同一连接固定落在一个核上，保住缓存局部性。

七、工程应用

八、常见误区

九、参考文献

  1. J. Corbet, A. Rubini, G. Kroah-Hartman (2005). Linux Device Drivers (3rd ed.), Chapter 10: Interrupt Handling. O'Reilly.
  2. Linux kernel documentation: Documentation/core-api/genericirq.rst 与 Documentation/IRQ.txt。
  3. J. Corbet (2006/2023). What happened to tasklets? 及 NAPI 相关系列文章, LWN.net。
  4. Red Hat Enterprise Linux 文档: Network Interface Card Performance — RSS, IRQ affinity, and NAPI tuning。
  5. W. Mauerer 著，郭旭译. 专业 Linux 内核架构 第 5 章：中断与下半部机制。
← 返回文章列表