Linux内核中断机制深度解析：从硬件中断到软IRQ全链路

4.5 ksoftirqd 内核线程

当软中断触发频率过高（例如大量网络数据包到达），硬中断返回时软中断处理不完，内核会唤醒 ksoftirqd 线程：

// kernel/softirq.c
static void run_ksoftirqd(unsigned int cpu)
{
    ksoftirqd_enter(cpu);
    local_bh_disable();
    
    // 检查并处理软中断
    if (local_softirq_pending()) {
        do_softirq();
        local_bh_enable();
        cond_resched();
        return;
    }
    
    local_bh_enable();
    ksoftirqd_exit(cpu);
}

// 典型优化：运行 ksoftirqd（防止单核过载）
// ps aux | grep ksoftirqd
// root      1234  0.5  0.0      0     0 ?        S    10:00   0:00 [ksoftirqd/0]

五、Tasklet：基于软中断的下半部

Tasklet 是基于 HI_SOFTIRQ 和 TASKLET_SOFTIRQ 的通用下半部机制，大多数驱动程序使用 Tasklet。

5.1 Tasklet 数据结构

// include/linux/interrupt.h
struct tasklet_struct {
    struct tasklet_struct *next;    // 链表指针
    unsigned long state;            // 状态位图
    atomic_t count;                 // 引用计数（0=启用，!=0=禁用）
    void (*func)(unsigned long);    // 处理函数
    unsigned long data;             // 传递给 func 的参数
};

// Tasklet 状态
enum {
    TASKLET_STATE_SCHED = 0,    // 已被调度（挂入 executing 链表）
    TASKLET_STATE_RUN,          // 正在执行
};

5.2 Tasklet 调度与执行

// kernel/softirq.c
void tasklet_schedule(struct tasklet_struct *t)
{
    unsigned long flags;
    
    local_irq_save(flags);
    // 检查是否已经在调度队列中
    if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state)) {
        // 挂入当前 CPU 的 tasklet_vec 链表
        t->next = NULL;
        *__this_cpu_read(tasklet_vec.head) = t;
        __this_cpu_write(tasklet_vec.tail, &(t->next));
        // 触发软中断
        raise_softirq_irqoff(TASKLET_SOFTIRQ);
    }
    local_irq_restore(flags);
}

// Tasklet 执行函数
static void tasklet_action(struct softirq_action *a)
{
    struct tasklet_struct *list;
    
    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(tasklet_vec.head));
    local_irq_enable();
    
    while (list) {
        struct tasklet_struct *t = list;
        list = list->next;
        
        if (tasklet_trylock(t)) {
            if (!atomic_read(&t->count)) {
                // 清除 SCHED 状态后执行
                if (!test_and_clear_bit(TASKLET_STATE_SCHED, &t->state))
                    BUG();
                t->func(t->data);
                tasklet_unlock(t);
                continue;
            }
            tasklet_unlock(t);
        }
        // 如果被禁用，重新挂入队列
        local_irq_disable();
        t->next = NULL;
        *__this_cpu_read(tasklet_vec.head) = t;
        __this_cpu_write(tasklet_vec.tail, &(t->next));
        raise_softirq_irqoff(TASKLET_SOFTIRQ);
        local_irq_enable();
    }
}

5.3 Tasklet 关键特性

• 串行执行：同一个 Tasklet 在任意时刻只能在某一个 CPU 上执行（通过 TASKLET_STATE_RUN 状态保证）
• Per-CPU 调度队列：Tasklet 总是在调度它的 CPU 上执行，避免缓存失效
• 不能睡眠：Tasklet 在软中断上下文中执行，不能调用可能睡眠的函数
• 禁止类型限制：Tasklet 执行时不能持有信号架、不能调用 copy_to_user/copy_from_user

六、工作队列（Workqueue）：可睡眠的下半部

工作队列是将工作推迟到内核线程（worker）中执行的机制，可以睡眠、可以执行耗时操作。

6.1 工作队列架构

// include/linux/workqueue.h
struct workqueue_struct {
    struct list_head        pwqs;          // 挂载的 pwq 链表
    struct list_head        list;          // 全局工作队列链表
    struct mutex            mutex;
    int                     work_color;    // 颜色标记（调试用）
    int                     flush_color;
    atomic_t                nr_pwqs_to_flush;
    struct wq_device        *wq_dev;
    char                    name[WQ_NAME_LEN];
    ...
};

struct pool_workqueue {
    struct worker_pool      *pool;         // 关联的线程池
    struct workqueue_struct *wq;           // 所属工作队列
    int                     work_color;
    int                     flush_color;
    struct list_head        delayed_works; // 延迟执行的 work
    ...
};

struct worker_pool {
    spinlock_t              lock;
    int                     cpu;
    int                     node;          // NUMA 节点
    int                     id;
    unsigned int            flags;
    struct list_head        worklist;      // 工作链表
    int                     nr_workers;    // 工作线程数
    ...
};

6.2 创建与调度工作

// 方式1: 使用共享的全局工作队列
schedule_work(&work);           // 立即调度
schedule_delayed_work(&work, delay);  // 延迟调度（单位：jiffies）

// 方式2: 创建私有工作队列
struct workqueue_struct *my_wq = create_workqueue("my_wq");
queue_work(my_wq, &work);
queue_delayed_work(my_wq, &dwork, msecs_to_jiffies(100));

// 方式3: 使用 alloc_workqueue（推荐）
struct workqueue_struct *wq = alloc_workqueue("my_wq", WQ_UNBOUND | WQ_HIGHPRI, 0);

6.3 Worker 线程生命周期

// kernel/workqueue.c
static int worker_thread(void *__worker)
{
    struct worker *worker = __worker;
    struct worker_pool *pool = worker->pool;
    
    // 设置 worker 线程状态
    worker->task->flags |= PF_WQ_WORKER;
    
recheck:
    // 如果没有待处理的工作，睡眠等待
    if (likely(!need_more_worker(pool)))
        schedule();
    
    // 醒来后检查工作列表
    while (!list_empty(&pool->worklist)) {
        struct work_struct *work = list_first_entry(
            &pool->worklist, struct work_struct, entry);
        
        // 从链表中取出，执行
        worker->current_work = work;
        work->func(work);
        worker->current_work = NULL;
    }
    goto recheck;
}

6.4 Workqueue 各类型比较

• WQ_UNBOUND：不绑定 CPU，适用于跨 NUMA 节点的工作
• WQ_HIGHPRI：高优先级 worker 线程（nice -20），适用于延迟敏感任务
• WQ_CPU_INTENSIVE：CPU 密集型工作，不计入 workqueue 并发限制
• WQ_MEM_RECLAIM：可参与内存回收的工作（用于文件系统 Journal 等）
• WQ_FREEZABLE：系统休眠时冻结的工作
• max_active：最大并发工作数（默认 256，WQ_UNBOUND 默认 使用 nr_cpu_ids * 4）

七、线程化中断（Threaded IRQ）

Linux 3.x 引入了线程化中断机制，将硬中断处理的大部分工作从 HardIRQ 上下文移到内核线程中执行，减少关中断时间。

7.1 线程化中断的优势

• 减少硬中断延迟：HardIRQ 处理函数只做最少的操作（如确认中断来源）
• 可睡眠：线程化 handler 中可以调用可能睡眠的函数（mutex、alloc、I/O）
• 实时优先级：线程化 handler 可以设置为 SCHED_FIFO 优先级
• 可被抢占：线程化 handler 执行期间可被更高优先级的中断/进程抢占

7.2 注册线程化中断

// 方法1: 使用 request_threaded_irq
int ret = request_threaded_irq(irq,
                    my_hard_handler,    // 上半部（可为 NULL）
                    my_thread_handler,  // 线程化处理
                    IRQF_ONESHOT,       // 推荐：处理完成前屏蔽中断
                    "my_device", dev);

// 方法2: 使用 devm_request_threaded_irq（自动释放）
ret = devm_request_threaded_irq(dev, irq, NULL, my_handler,
                                IRQF_ONESHOT, "my_device", dev);

// 方法3: 使用 request_any_context_irq（自动选择）
ret = request_any_context_irq(irq, handler, flags, name, dev);
// 返回值：
//   - 非负：成功
//   - -EOPTNOTSUPP：不支持线程化
//   - ENOSYS：没有 hardirq handler 且不支持线程化

7.3 IRQF_ONESHOT 标志

这是线程化中断最重要的标志，它保证在线程handler函数执行完成之前，中断线始终被屏蔽：

// kernel/irq/manage.c
static irqreturn_t irq_thread_fn(void *data)
{
    struct irqaction *action = data;
    irqreturn_t ret;
    
    while (true) {
        // 执行 handler 前清掉中断
        if (!IRQ_HANDLED_WITHOUT_MASKING)
            unmask_threaded_irq(desc);  // 解屏蔽中断线
        
        ret = action->thread_fn(action-> irq, action->dev_id);
        
        // 再次确认是否有新的中断到达（级联中断场景）
        if (!(desc->istate & IRQS_PENDING))
            break;
    }
    
    // handler 返回 IRQ_HANDLED 后，屏蔽中断并唤醒等待者
    return ret;
}

八、中断亲和性（IRQ Affinity）

中断亲和性允许将特定的 IRQ 绑定到指定的 CPU 核心处理，是实现网卡多队列、隔离中断负载的关键技术。

8.1 设置中断亲和性

// 查看当前 IRQ 的 CPU 绑定（IRQ 号 123 为例）
cat /proc/irq/123/smp_affinity
cat /proc/irq/123/smp_affinity_list

// 将 IRQ 123 绑定到 CPU 0-1（十六进制掩码 0x3）
echo 3 > /proc/irq/123/smp_affinity

// 将 IRQ 123 绑定到 CPU 2-3（十六进制掩码 0xc）
echo c > /proc/irq/123/smp_affinity

// 通过 irqbalance 服务自动管理
systemctl stop irqbalance    # 停止手动管理

8.2 网卡多队列与 RSS

现代网卡通常支持 16-128 个 TX/RX 队列，每个队列有独立的中断：

# 查看网卡队列
ethtool -l eth0
#   Combined: 8       # 当前使用 8 个队列
#   RX: 0
#   TX: 0

# 设置网卡队列数
ethtool -L eth0 combined 16

# 配置网卡 Ring Buffer
ethtool -G eth0 rx 4096 tx 4096

# 查看各队列的 IRQ 号
cat /proc/interrupts | grep eth0

# 手动分配中断到不同 CPU
echo "1" > /proc/irq/123/smp_affinity   # CPU0 - eth0-rx-0
echo "2" > /proc/irq/124/smp_affinity   # CPU1 - eth0-rx-1
echo "4" > /proc/irq/125/smp_affinity   # CPU2 - eth0-rx-2
echo "8" > /proc/irq/126/smp_affinity   # CPU3 - eth0-rx-3

8.3 irqbalance 服务

irqbalance 是用户空间的中断负载均衡守护进程：

• 工作模式：周期性采样各 CPU 的中断负载，动态调整 IRQ 与 CPU 的绑定关系
• SNC（Sub-Numa Clustering）：在 NUMA 架构下，优先将中断绑定到设备所在的 NUMA 节点
• 禁止规则：可配置特定 IRQ 不被 irqbalance 管理（如手机、实时系统）
• 性能建议：在高性能网络场景建议禁用 irqbalance，手动配置亲和性

九、中断处理中的并发与同步

9.1 中断上下文 vs 进程上下文

• 中断上下文：没有关联的 task_struct，不能睡眠（没有调度实体），执行时间短
• 进程上下文：内核线程或用户进程，可以睡眠，可重入
• 判断方法：in_interrupt() 返回 true 表示在中断/软中断上下文
• 危险操作：在中断上下文调用 copy_to_user()、kmalloc(GFP_KERNEL)、mutex_lock() 等会导致系统崩溃

9.2 自旋锁与中断屏蔽

// 场景1: 仅进程上下文共享数据（同一 CPU 不同时间点执行）
spin_lock(&lock);
// 访问临界区
spin_unlock(&lock);

// 场景2: 进程上下文 + 硬件中断共享数据
unsigned long flags;
spin_lock_irqsave(&lock, flags);
// 访问临界区（中断已屏蔽，不会被打断）
spin_unlock_irqrestore(&lock, flags);

// 场景3: 进程上下文 + 软中断共享数据
spin_lock_bh(&lock);
// 访问临界区（软中断被屏蔽）
spin_unlock_bh(&lock);

// 场景4: 与 Tasklet 共享数据（Tasklet 不能睡眠）
spin_lock_bh(&lock);  // 或 spin_lock(&lock)
// 访问临界区
spin_unlock_bh(&lock);

9.3 Per-CPU 变量

Per-CPU 变量是中断上下文中最常用的无锁数据访问方式：

// 声明
DEFINE_PER_CPU(unsigned long, my_counter);

// 读写（不需要锁，因为每个 CPU 有自己的副本）
get_cpu_var(my_counter)++;
__get_cpu_var(my_counter) += 10;
put_cpu_var(my_counter);

// API 形式
this_cpu_inc(my_counter);       // 原子 +1
this_cpu_add(my_counter, 5);    // 原子 +5
raw_cpu_ptr(&my_counter);       // 获取原始指针

9.4 RCU 与中断上下文

RCU（Read-Copy-Update）是读多写少的并发机制，在 INTERRUPT 上下文中：

• 读侧：rcu_read_lock() / rcu_read_unlock() 可在中断上下文中使用（不可睡眠）
• 写侧：synchronize_rcu() 会睡眠，不能在中断上下文使用
• 替代方案：使用 call_rcu() 将回调函数放到 RCU_SOFTIRQ 中异步执行
• 中断延迟： rcu_nocbs 参数可将特定 CPU 移出 RCU 回调，降低中断延迟

十、中断性能调优与监控

10.1 关键性能参数

# /etc/sysctl.conf

# 软中断处理包数上限（每轮）
net.core.netdev_budget = 600         # NAPI 单次 poll 最大包数（默认 300）
net.core.netdev_budget_usecs = 8000  # NAPI 单次 poll 最大耗时（微秒）

# ksoftirqd 优先级（负值提升优先级）
# /etc/systemd/system/ksoftirqd.service
# Nice=-20
# CPUSchedulingPolicy=fifo

# 中断处理 Workers 数量
# 内核编译参数 CONFIG_RPS_RFS_ENTRIES=32768

# PCI 总线相关
# 确保 PCIe Max_P_Read_Request_Size 合理（通常为 4096 字节）

10.2 中断监控命令

# 查看所有中断在各 CPU 上的分布
cat /proc/interrupts

# 输出示例：
#            CPU0   CPU1   CPU2   CPU3
#   9:    12345    678    ...   ...   PCI-MSI-edge eth0-rx-0
#  10:      678  12345    ...   ...   PCI-MSI-edge eth0-rx-1
#  11:      ...    ...  12345   ...   PCI-MSI-edge eth0-rx-2

# 软中断统计（包括 NET_RX、NET_TX、BLOCK 等）
cat /proc/softirqs

# 中断处理耗时分析（使用 perf）
perf record -g -e irq:irq_handler_entry -ag sleep 10

# NAPI/softirq 监控
cat /proc/net/softnet_stat  # 每列：处理数、时延、budget用完次数、CPU碰撞

# ksoftirqd 线程 CPU 占用（如果某 CPU 上此线程占用高，说明软中断过载）
top -p $(pgrep ksoftirqd)

10.3 中断优化策略

• 合并中断：ethtool -C eth0 rx-usecs 100（100微秒合并间隔）减少中断次数
• 多队列分流：配置多队列网卡 RSS + 手动 IRQ affinity，分散中断负载
• NAPI 机制：网卡数据包接收切换到轮询模式，降低高速率场景下的中断风暴
• XDP 绕过：在网卡驱动层直接处理/丢弃数据包，零协议栈开销
• CPU 隔离：通过 isolcpus 内核参数将特定 CPU 从调度器中隔离，专用于中断处理

十一、实战案例：网卡中断风暴处理

11.1 问题现象

某视频流媒体服务器（10GbE 网卡）在高峰期出现：

• TCP 重传率上升至 15%
• ksoftirqd/3 CPU 占用持续 100%
• top 显示 si（soft interrupt）占用比 sy 高出 10 倍

11.2 诊断分析

# 查看 CPU 3 处理的软中断
watch -n 1 'cat /proc/softirqs | head -6'
# NET_RX_SOFTIRQ 在 CPU 3 上每秒处理 30000+ 次

# 检查网卡中断分布
cat /proc/interrupts | grep eth0
# 发现所有网卡中断都落在 CPU 3 上（亲和性配置错误）

# 查看网卡队列数
ethtool -l eth0
# Combined: 1  # 仅启用 1 个队列！

# 查看 /proc/net/softnet_stat 第 3 列（budget耗尽次数）
# 该值持续增长，表示 NAPI poll 配额不够

11.3 解决方案

# 步骤1: 启用多队列
ethtool -L eth0 combined 8

# 步骤2: 手动配置多队列 RSS
ethtool -X eth0 equal 8

# 步骤3: 手动绑定中断到不同 CPU
cat /proc/interrupts | grep eth0
#  eth0-rx-0 -> IRQ 120 -> CPU 0
#  eth0-rx-1 -> IRQ 121 -> CPU 1
#  ...

echo 1 > /proc/irq/120/smp_affinity
echo 2 > /proc/irq/121/smp_affinity
echo 4 > /proc/irq/122/smp_affinity
echo 8 > /proc/irq/123/smp_affinity

# 步骤4: 增大 NAPI budget
sysctl -w net.core.netdev_budget=600

# 步骤5: 停止 irqbalance（手动管理更精确）
systemctl stop irqbalance

# 步骤6: 启用 XDP 处理高吞吐模式
ip link set dev eth0 xdp obj xdp_pass.o

优化后：

• TCP 重传率从 15% 降至 0.1% 以下
• ksoftirqd 从单核 100% 降至 8 核各 12%
• 网卡吞吐量从 2.1Gbps 提升至 9.4Gbps（接近线速）

十二、总结

Linux 内核中断处理机制的核心设计原则是：快速响应、延迟处理。通过将中断分为上半部（HardIRQ）和下半部（SoftIRQ / Tasklet / Workqueue），在保证硬件及时响应的同时，将耗时操作推迟到更安全的上下文中执行。

关键知识点回顾：

1. IDT + IRQ 向量：中断的入口路由机制
2. 上半部/下半部分离：HardIRQ 做最少工作，Bottom Half 延迟处理
3. SoftIRQ：10 种类型、Per-CPU 位图、串行执行、最高优先级下半部
4. Tasklet：基于 SoftIRQ，不可睡眠，适用于驱动下半部
5. Workqueue：可睡眠、可重入，适用于文件系统、块设备等复杂场景
6. Threaded IRQ：IRQF_ONESHOT，将中断大部分工作转移到线程中
7. IRQ Affinity：多队列网卡 + CPU 绑定，是实现高性能网络的基础
8. 同步原语：spin_lock_irqsave（关中断）、spin_lock_bh（关软中断）、Per-CPU 变量

掌握这些机制，不仅能写出更高效的内核驱动代码，也能在系统性能调优中精准定位中断相关的瓶颈。