Linux内核中的软中断tasklet和工作队列详解（超详细~）

Posted 2022-08-09 贺二公子

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Linux内核中的软中断tasklet和工作队列详解（超详细~）相关的知识，希望对你有一定的参考价值。

原文地址：https://www.bilibili.com/read/cv17094615

本文基于Linux2.6.32内核版本。

引言

软中断、tasklet和工作队列并不是Linux内核中一直存在的机制，而是由更早版本的内核中的“下半部”（bottom half）演变而来。下半部的机制实际上包括五种，但2.6版本的内核中，下半部和任务队列的函数都消失了，只剩下了前三者。
介绍这三种下半部实现之前，有必要说一下上半部与下半部的区别。
- 上半部指的是中断处理程序，下半部则指的是一些虽然与中断有相关性但是可以延后执行的任务。举个例子：在网络传输中，网卡接收到数据包这个事件不一定需要马上被处理，适合用下半部去实现；但是用户敲击键盘这样的事件就必须马上被响应，应该用中断实现。
- 两者的主要区别在于：中断不能被相同类型的中断打断，而下半部依然可以被中断打断；中断对于时间非常敏感，而下半部基本上都是一些可以延迟的工作。由于二者的这种区别，所以对于一个工作是放在上半部还是放在下半部去执行，可以参考下面4条：
  1. 如果一个任务对时间非常敏感，将其放在中断处理程序中执行。
  2. 如果一个任务和硬件相关，将其放在中断处理程序中执行。
  3. 如果一个任务要保证不被其他中断（特别是相同的中断）打断，将其放在中断处理程序中执行。
  4. 其他所有任务，考虑放在下半部去执行。
  5. 有写内核任务需要延后执行，因此才有的下半部，进而实现了三种实现下半部的方法。这就是本文要讨论的软中断、tasklet和工作队列。
下表可以更直观的看到它们之间的关系。

软中断

软中断作为下半部机制的代表，是随着SMP（share memory processor）的出现应运而生的，它也是tasklet实现的基础（tasklet实际上只是在软中断的基础上添加了一定的机制）。软中断一般是“可延迟函数”的总称，有时候也包括了tasklet（请读者在遇到的时候根据上下文推断是否包含tasklet）。它的出现就是因为要满足上面所提出的上半部和下半部的区别，使得对时间不敏感的任务延后执行，而且可以在多个CPU上并行执行，使得总的系统效率可以更高。它的特性包括：
- 产生后并不是马上可以执行，必须要等待内核的调度才能执行。软中断不能被自己打断(即单个cpu上软中断不能嵌套执行)，只能被硬件中断打断（上半部）。
- 可以并发运行在多个CPU上（即使同一类型的也可以）。所以软中断必须设计为可重入的函数（允许多个CPU同时操作），因此也需要使用自旋锁来保其数据结构。

实现原理和实例

软中断的调度时机:

do_irq完成I/O中断时调用irq_exit。
系统使用I/O APIC,在处理完本地时钟中断时。
local_bh_enable，即开启本地软中断时。
SMP系统中，cpu处理完被CALL_FUNCTION_VECTOR处理器间中断所触发的函数时。
ksoftirqd/n线程被唤醒时。

下面以从中断处理返回函数irq_exit中调用软中断为例详细说明。
触发和初始化的的流程如图所示:

软中断处理流程

asmlinkage void __do_softirq(void)

    struct softirq_action *h;
    __u32 pending;
    int max_restart = MAX_SOFTIRQ_RESTART;
    int cpu;

    pending = local_softirq_pending();
    account_system_vtime(current);

    __local_bh_disable((unsigned long)__builtin_return_address(0));
    lockdep_softirq_enter();

    cpu = smp_processor_id();
restart:
    /* Reset the pending bitmask before enabling irqs */
    set_softirq_pending(0);

    local_irq_enable();

    h = softirq_vec;

    do 
        if (pending & 1) 
            int prev_count = preempt_count();
            kstat_incr_softirqs_this_cpu(h - softirq_vec);

            trace_softirq_entry(h, softirq_vec);
            h->action(h);
            trace_softirq_exit(h, softirq_vec);
            if (unlikely(prev_count != preempt_count())) 
                printk(KERN_ERR "huh, entered softirq %td %s %p"
                       "with preempt_count %08x,"
                       " exited with %08x?\\n", h - softirq_vec,
                       softirq_to_name[h - softirq_vec],
                       h->action, prev_count, preempt_count());
                preempt_count() = prev_count;
            

            rcu_bh_qs(cpu);
        
        h++;
        pending >>= 1;
     while (pending);

    local_irq_disable();

    pending = local_softirq_pending();
    if (pending && --max_restart)
        goto restart;

    if (pending)
        wakeup_softirqd();

    lockdep_softirq_exit();

    account_system_vtime(current);
    _local_bh_enable();

首先调用local_softirq_pending函数取得目前有哪些位存在软件中断。
调用__local_bh_disable关闭软中断，其实就是设置正在处理软件中断标记，在同一个CPU上使得不能重入__do_softirq函数。
重新设置软中断标记为0，set_softirq_pending重新设置软中断标记为0，这样在之后重新开启中断之后硬件中断中又可以设置软件中断位。
调用local_irq_enable，开启硬件中断。
之后在一个循环中，遍历pending标志的每一位，如果这一位设置就会调用软件中断的处理函数。在这个过程中硬件中断是开启的，随时可以打断软件中断。这样保证硬件中断不会丢失。
之后关闭硬件中断(local_irq_disable)，查看是否又有软件中断处于pending状态，如果是，并且在本次调用__do_softirq函数过程中没有累计重复进入软件中断处理的次数超过max_restart=10次，就可以重新调用软件中断处理。如果超过了10次，就调用wakeup_softirqd()唤醒内核的一个进程来处理软件中断。设立10次的限制，也是为了避免影响系统响应时间。
调用_local_bh_enable开启软中断。

软中断内核线程

之前我们分析的触发软件中断的位置其实是中断上下文中，而在软中断的内核线程中实际已经是进程的上下文。
这里说的软中断上下文指的就是系统为每个CPU建立的ksoftirqd进程。
软中断的内核进程中主要有两个大循环，外层的循环处理有软件中断就处理，没有软件中断就休眠。内层的循环处理软件中断，每循环一次都试探一次是否过长时间占据了CPU，需要调度就释放CPU给其它进程。具体的操作在注释中做了解释。

set_current_state(TASK_INTERRUPTIBLE);
    //外层大循环。
    while (!kthread_should_stop()) 
        preempt_disable();//禁止内核抢占，自己掌握cpu
        if (!local_softirq_pending()) 
            preempt_enable_no_resched();
            //如果没有软中断在pending中就让出cpu
            schedule();
            //调度之后重新掌握cpu
            preempt_disable();
        

        __set_current_state(TASK_RUNNING);

        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))
                goto wait_to_die;
            //有软中断则开始软中断调度
            do_softirq();
            //查看是否需要调度，避免一直占用cpu
            preempt_enable_no_resched();
            cond_resched();
            preempt_disable();
            rcu_sched_qs((long)__bind_cpu);
        
        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);
    while (!kthread_should_stop()) 
        schedule();
        set_current_state(TASK_INTERRUPTIBLE);
    
    __set_current_state(TASK_RUNNING);
    return 0;

tasklet

由于软中断必须使用可重入函数，这就导致设计上的复杂度变高，作为设备驱动程序的开发者来说，增加了负担。而如果某种应用并不需要在多个CPU上并行执行，那么软中断其实是没有必要的。因此诞生了弥补以上两个要求的tasklet。它具有以下特性：
1. 一种特定类型的tasklet只能运行在一个CPU上，不能并行，只能串行执行。
2. 多个不同类型的tasklet可以并行在多个CPU上。
3. 软中断是静态分配的，在内核编译好之后，就不能改变。但tasklet就灵活许多，可以在运行时改变（比如添加模块时）。
tasklet是在两种软中断类型的基础上实现的，因此如果不需要软中断的并行特性，tasklet就是最好的选择。也就是说tasklet是软中断的一种特殊用法，即延迟情况下的串行执行。

实现原理

调度原理

static inline void tasklet_schedule(struct tasklet_struct *t)

    if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state))
        __tasklet_schedule(t);

void __tasklet_schedule(struct tasklet_struct *t)

    unsigned long flags;

    local_irq_save(flags);
    t->next = NULL;
    *__get_cpu_var(tasklet_vec).tail = t;
    __get_cpu_var(tasklet_vec).tail = &(t->next);//加入低优先级列表
    raise_softirq_irqoff(TASKLET_SOFTIRQ);//触发软中断
    local_irq_restore(flags);

tasklet执行过程 TASKLET_SOFTIRQ对应执行函数为tasklet_action，HI_SOFTIRQ为tasklet_hi_action，以tasklet_action为例说明，tasklet_hi_action大同小异。

static void tasklet_action(struct softirq_action *a)

    struct tasklet_struct *list;

    local_irq_disable();
    list = __get_cpu_var(tasklet_vec).head;
    __get_cpu_var(tasklet_vec).head = NULL;
    __get_cpu_var(tasklet_vec).tail = &__get_cpu_var(tasklet_vec).head;//取得tasklet链表
    local_irq_enable();

    while (list) 
        struct tasklet_struct *t = list;

        list = list->next;

        if (tasklet_trylock(t)) 
            if (!atomic_read(&t->count)) 
                //执行tasklet
                if (!test_and_clear_bit(TASKLET_STATE_SCHED, &t->state))
                    BUG();
                t->func(t->data);
                tasklet_unlock(t);
                continue;
            
            tasklet_unlock(t);
        
        //如果t->count的值不等于0，说明这个tasklet在调度之后，被disable掉了，所以会将tasklet结构体重新放回到tasklet_vec链表，并重新调度TASKLET_SOFTIRQ软中断，在之后enable这个tasklet之后重新再执行它
        local_irq_disable();
        t->next = NULL;
        *__get_cpu_var(tasklet_vec).tail = t;
        __get_cpu_var(tasklet_vec).tail = &(t->next);
        __raise_softirq_irqoff(TASKLET_SOFTIRQ);
        local_irq_enable();

工作队列

从上面的介绍看以看出，软中断运行在中断上下文中，因此不能阻塞和睡眠，而tasklet使用软中断实现，当然也不能阻塞和睡眠。但如果某延迟处理函数需要睡眠或者阻塞呢？没关系工作队列就可以如您所愿了。
把推后执行的任务叫做工作（work），描述它的数据结构为work_struct ，这些工作以队列结构组织成工作队列（workqueue），其数据结构为workqueue_struct ，而工作线程就是负责执行工作队列中的工作。系统默认的工作者线程为events。
工作队列(work queue)是另外一种将工作推后执行的形式。工作队列可以把工作推后，交由一个内核线程去执行—这个下半部分总是会在进程上下文执行，但由于是内核线程，其不能访问用户空间。最重要特点的就是工作队列允许重新调度甚至是睡眠。
通常，在工作队列和软中断/tasklet中作出选择非常容易。可使用以下规则：
1. 如果推后执行的任务需要睡眠，那么只能选择工作队列。
2. 如果推后执行的任务需要延时指定的时间再触发，那么使用工作队列，因为其可以利用timer延时(内核定时器实现)。
3. 如果推后执行的任务需要在一个tick之内处理，则使用软中断或tasklet，因为其可以抢占普通进程和内核线程，同时不可睡眠。
4. 如果推后执行的任务对延迟的时间没有任何要求，则使用工作队列，此时通常为无关紧要的任务。
实际上，工作队列的本质就是将工作交给内核线程处理，因此其可以用内核线程替换。但是内核线程的创建和销毁对编程者的要求较高，而工作队列实现了内核线程的封装，不易出错，所以我们也推荐使用工作队列。

实现原理

1.工作队列的组织结构

即workqueue_struct、cpu_workqueue_struct与work_struct的关系。
一个工作队列对应一个work_queue_struct，工作队列中每cpu的工作队列由cpu_workqueue_struct表示，而work_struct为其上的具体工作。
关系如下图所示:

2. 工作队列的工作过程

应用实例

linux各个接口的状态(up/down)的消息需要通知netdev_chain上感兴趣的模块同时上报用户空间消息。这里使用的就是工作队列。
具体流程图如下所示：
是否处于中断中在Linux中是通过preempt_count来判断的,具体如下：在linux系统的进程数据结构里，有这么一个数据结构:

#define preempt_count() (current_thread_info()->preempt_count) 
//利用preempt_count可以表示是否处于中断处理或者软件中断处理过程中,如下所示： 
# define hardirq_count() (preempt_count() & HARDIRQ_MASK) 
#define softirq_count() 以上是关于Linux内核中的软中断tasklet和工作队列详解（超详细~）的主要内容，如果未能解决你的问题，请参考以下文章 
 Linux内核中的软中断tasklet和工作队列详解（超详细~）
 [Linux内核]软中断tasklet工作队列
 《深入理解Linux内核》软中断/tasklet/工作队列
 Linux 内核 tasklet 机制和工作队列zz
 Linux内核中的下半部机制之tasklet
 Linux内核中的下半部机制之tasklet

Linux内核中的软中断tasklet和工作队列详解（超详细~）

引言

软中断

相关数据结构

相关API

实现原理和实例

软中断的调度时机:

软中断处理流程

软中断内核线程

tasklet

相关数据结构

相关API

实现原理

工作队列

相关数据结构

相关API

实现原理

1.工作队列的组织结构

2. 工作队列的工作过程

应用实例