读薄《Linux 内核设计与实现》 - 中断与同步

Posted 2020-08-02 冰水比水冰

这篇文章是《读薄「Linux 内核设计与实现」》系列文章的第 IV 篇，本文主要讲了以下问题：中断和中断处理程序的概念与实现原理、Linux 中的下半部以及内核同步方法。

0x00 中断和中断处理程序

I 中断

• 中断是一种特殊的电信号，由硬件发向处理器，处理器接收到中断时，会马上箱操作系统反映，由操作系统进行处理。中断随时可以产生，因此，内核随时可能因为新到来的中断而被打断。

• 不同的设备对应的中断不同，每个中断通过一个唯一的数字标识，这些中断值通常被称为中断请求（IRQ）线。

II 中断处理程序

• 中断处理程序又成为中断处理例程（ISR），是内核在响应中断时执行的一个函数

• 一个中断处理程序对应一个中断，一个设备可能发出多种中断

• 对于外部设备，中断处理程序是设备驱动程序的一部分

• 在 Linux 中，中断处理程序和 C 函数区别不大，但有自己的规范，主要是运行时需要在中断上下文中

0x01 中断处理机制

I 注册中断处理程序

驱动程序可以通过request_irq()函数注册一个中断处理程序(linux/interrupt.h)

int request_irq(unsigned int irq,
                irqhandler_t handler,
                unsigined long falgs,
                const char *name,
                void *dev)

• irq:表示要分配的中断号

• handler:一个指针，指向处理这个中断的实际中断处理函数

typedef irqhandler_t(*irq_handler_t)(int, void*);

II 释放中断处理程序

卸载驱动程序时，需要注销响应中断处理程序，并释放中断线。

void free_irq(unsigned int irq, void *dev);

如果指定的中断线不是共享的，那么该函数删除处理程序的同时将禁用这条中断线；中断线是共享的，则仅仅删除 dev 对应的处理程序，而这条中断线本身只有在删除了最后一个处理程序时才会被禁用

III 中断的禁止与激活

local_irq_disable();
local_irq_enable();

IV 上半部与下半部

又想中断处理程序运行的快，又想中断处理程序完成的工作多，这两个目的显然有所抵触，所以把中断处理分为两个部分：

• 中断处理程序是上半部，接收到一个中断，它就立刻开始执行，但只做有严格时限的工作，例如一些只有在中断被禁止的状态下才能完成的工作

• 能够被允许稍后完成的工作会推迟2到下半部去，此后，在合适的时机，下半部会被开中断执行

Q1：为什么要分上半部和下半部？ {% endcq %}

• 中断程序以异步方式执行，可能打断重要操作的执行，越快越好
• 中断处理程序会屏蔽其他同级中断，所以执行越快越好
• 中断处理程序往往需要对硬件操作，通常有很高的时限要求
• 中断处理程序不在进程的上下文中运行，所以不能阻塞

Q2：上半部和下半部如何分开？ {% endcq %}

• 如果一个任务对时间非常敏感，将其放到上半部；
• 如果一个任务和硬件相关，将其放到上半部；
• 如果一个任务要保证不被其它中断打断，将其放到上半部；
• 其他的所有任务考虑放到下半部

0x02 下半部

下半部的任务就是执行与终端处理密切相关但中断处理程序本身不执行的工作。我们期望中断处理程序将尽量多的工作放到下半部执行，以快速从中断返回。

I 下半部实现机制

a.软中断

此处的软中断和系统调用使用的 int 80H 不同，是操作系统支持的一种，在编译期间静态分配

软中断的实现

• 定义于 linux/interrupt.h 中：

struct softirq_action{
    void (*action)(struct sfotirq_action*); //待执行的函数
    void *data; //传递的参数
}

• 最多可能32个软中断，定义于 kernel/softirq.c

static struct softirq_action softirq_vec[NR_SOFTIRQS];

软中断处理程序

void softirq_handler(struct softirq_action*); //传递整个结构体

执行软中断

一个注册的软中断必须在被标记后才会执行下列地方，待处理的软中断会被检查和执行：

• 在 ksoftirqd 内核线程中
• 在那些显式检查和和执行待处理的软中断的代码中（如网络子系统）

不管执行的时机，软中断都要在do_softirq 中执行

使用软中断

• 分配索引： 通过 linux/interrupt.h 中的一个枚举类型中声明一个新的软中断

• 注册处理程序：在运行时通过调用open_softirq()注册软中断处理程序，有两个参数，软中断和处理函数

• 触发软中断：raise_softirq()函数可以将一个软中断设置为挂起状态，让它在下次调用do_softirq()函数时投入运行

b.tasklet

基于软中断的实现，但它的接口更简单，锁保护要求更低

tasklet 的实现

• tasklet 结构体（linux/interrupt.h）

struct tasklet_struct{
    struct tasklet_struct *next;    //链表
    unsigned long state;    //tasklet 状态
    atomic_t count; //引用计数器
    void (*funx)(unsigned long);    //taklet 处理函数
    unsigned long data; //给处理器函数传递的参数
}

调度 tasklet

被触发的软中断存放在2个数据结构：tasklet_vec,task_hi_vec,这两个数据结构都是由task_struct构成的链表，由tasklet_schedule()和task_hi_schedule()进行调度，调度步骤如下：

（1）检查 tasklet 状态，如果为TASK_STATE_SCHED则返回
（2）调用_tasklet_schedule()
（3）保存中断状态，禁止本地中断
（4）把需要调度的 tasklet 加到 tasklet_vec或tasklet_hi_vec链表头
（5）唤起TASKLET_SOFTIRQ或HI_SOFTIRQ软中断，下一次调用do_softirq()时会执行该 tasklet
（6）恢复中断

• 软中断处理程序：tasklet_action()或task_hi_action()[tasklet 处理的核心]：
  
  • （1）禁止中断
  • （2）将当前处理器上的该链表头设为 NULL
  • （3）允许中断
  • （4）循环遍历链表上所有待处理的 tasklet
  • （5）如果是多个处理器，检查TASKLET_STATE_RUN判断这个 tasklet 是否在其他处理器上运行，如果是，跳到笑一个 tasklet
  • （6）如果否，设置TASKLET_STATE_RUN
  • （7）检查 count 是否为0，确保 tasklet 没有被禁止；如果被禁止，跳到下一个 tasklet
  • （8）执行 tasklet 处理程序
  • （9）执行完毕，清除TASKLET_STATE_RUN
  • （10）重复执行下一个 tasklet

使用 tasklet

声明自己的 tasklet：

• 静态：linux/interrupt.h 中的2个宏：

DECLARE_TASKLET(name,func,data);
DECLARE_TASKLET_DIASBLED(name,func,data);

• 动态：通过一个指针赋给一个动态创建的 tasklet_struct:

tasklet_init(t, takslet_handler, dev);

编写自己的 tasklet 处理程序

void tasklet_handler(unsigned long data)

注意：不能再 tasklet 中使用信号量或者其他阻塞式函数

调度自己的 tasklet

tasklet_schedule(&my_tasklet);
tasklet_enable(&my_tasklet);
tasklet_disable(&my_tasklet);

ksoftirqd

ksoftirqd 是内核线程，每个处理器都有一个，用于在空闲处理器处理软中断

for(;;){
    if(!softirq_pending(cpu))
        schedule();
    set_current_state(TASK_RUNNING);
    while(softirq_pending(cpu)){
        do_softirq();
        if(need_resched())
        schedule();
    }
    set_current_sdtate(TASK_INTERRUPTIBLE);
}

只要有待处理的软中断，该线程就会处理

c.工作队列

工作队列机制将下半部功能交给内核县城去执行，有线程上下文，可以睡眠

工作队列的实现

• 提供创建工作者线程的接口
• 提供默认的工作者线程处理排到队列里的下半部工作
• 提供吧需要推后执行的任务排到队列里的接口

处理机制

• 线程将自己休眠，并加到等待队列（TASK_INTERRUPTIBLE）
• 如果工作链表为空，线程调用schedule()，休眠
• 如果不为空，将自己设为TASK_RUNNING
• 调用run_workqueue()执行被推后的工作

该函数循环遍历链表上每个待处理的工作：

• 当链表非空，选取下一个节点对象
• 获取要执行的函数和参数
• 从链表上解下该节点，将 pending 位清零
• 调用函数
• 重复执行

工作队列的使用

• 创建推后的工作：

静态：

DECLARE_WORK(name, void(*func)(void*), void *data);

动态：

INIT_WORK(struct work_struct *work, void(*func)(void*), void *data);

• 工作队列处理函数

void work_handler(void *data)

• 对工作的调度

schedule_work(&work);
schedule_delayed_work(&work, delay);

• 刷新操作

void flush_scheduled_work(void);

3种下半部机制的比较

机制	上下文	顺序执行保障
软中断	中断	没有
tasklet	中断	同类型不能同时执行
工作队列	进程	没有（和进程上下文一样被调度）

Q：我们要选择哪种机制？ {% endcq %}

如果有休眠的要求，选择工作队列；否则，最好使用 tasklet；要是必须专注性能的提高，选择软中断

II 在下半部之间加锁

• 如果进程上下文和一个下半部共享数据，在访问这些数据之前，你需要禁止下半部的处理并得到锁的使用权

• 如果中断上下文和一个下半部共享数据，在访问数据之前，需要禁止中断并得到锁的使用权

0x03 内核同步

I 临界区

临界区就是访问和操作共享资源的代码段，必须保证原子地执行才能保证安全

II 加锁

保证在临界区执行的县城只有一个

III 造成并发的原因

• 中断

• 软中断和 tasklet

• 内核抢占

• 睡眠及用户空间的同步

• 对称多处理

IV 死锁产生条件

• 要有一个或多个执行线程和一个或多个资源

• 每一个线程都在等待其中一个资源

• 所有的资源都被占用

• 所有县城都在互相等待，但他们永远不会释放已经占有的资源

V 内核同步方法

原子操作

• 原子整数操作（asm/atomic.h）

atomic_dec_and_test(atmoic_t, *v)

• 原子位操作（asm/bitops.h）

set_bit(0, &word)

自旋锁

• 自旋锁只能被一个可执行进程持有

• 若争用一个被占用的锁则进程忙等（旋转）

• 自旋锁不能长期被占用，否则效率低

---

本文的版权归作者 罗远航 所有，采用 Attribution-NonCommercial 3.0 License。任何人可以进行转载、分享，但不可在未经允许的情况下用于商业用途；转载请注明出处。感谢配合！