linux内核分析--中断中的下半部

1.软中断   实际上软中断使用的并不多,反而是后面的tasklet比较多,但tasklet是通过软中断实现的,软中断的代码位于/kernel/softirq.c中。软中断是在编译期间静态分配的,由softirq_action结构表示,它定义在linux/interrupt.h中:
1
2
3
4
structsoftirq_action{
    void(*action)(structsoftirq_action *); //待执行的函数
    void*data;//传给函数的参数
};

kernel/softirq.c中定义了一个包含有32个结构体的数组:

1
staticstructsoftirq_action softirq_vec[32]

       每个被注册的软中断都占据该数组的一项,因此最多可能有32个软中断,这是没法动态改变的。由于大部分驱动程序都使用tasklet来实现它们的下半部,所以现在的内核中,只用到了6个。上面的软中断结构中,第一项是软中断处理程序,原型如下:

1
void softirq_handler(struct softirq_action *)

       当内核运行一个软中断处理程序时,它会执行这个action函数,其唯一的参数为指向相应softirq_action结构体的指针。例如,如果my_softirq指向softirq_vec数组的实现,那么内核会用如下的方式调用软中断处理程序中的函数:

1
my_softirq->action(my_softirq)

       一个软中断不会抢占另外一个软中断,实际上,唯一可以抢占软中断的是中断处理程序。不过,其他的软中断----甚至是相同类型的软中断-----可以在其他类型的机器上同时执行。一个注册的软中断必须在被标记后才能执行----触发软中断(rasing the softirq).通常,中断处理程序会在返回前标记它的软中断,使其在稍后执行。在下列地方,待处理的软中断会被检查和执行:

1.处理完一个硬件中断以后                   2.在ksoftirqd内核线程中                 3.在那些显示检查和执行待处理的软中断的代码中,如网络子系统中。

       软中断会在do_softirq()中执行,如果有待处理的软中断,do_softirq会循环遍历每一个,调用他们的处理程序。核心部分如下:

1
2
3
4
5
6
7
8
9
10
11
u32 pending = softirq_pending(cpu);
if(pending){
    structsoftirq_action *h = softirq_vec;
    softirq_pending(cpu) = 0;
    do{
      if(pending &1)
        h->action(h);
      h++;
      pending >>=1;
    }while(pending);
}

       上述代码会检查并执行所有待处理的软中断,softirq_pending(),用它来获得待处理的软中断的32位位图-----如果第n位被设置为1,那么第n位对应类型的软中断等待处理,一旦待处理的软中断位图被保存,就可以将实际的软中断位图清零了。pending &1是判断pending的第一位是否被置为1.一旦pending为0,就表示没有待处理的中断了,因为pending最多可能设置32位,循环最多也只能执行32位。软中断保留给系统中对时间要求最严格以及最重要的下半部使用。所以使用之前还是要想清楚的。下面简要的说明如何使用软中断:

1.分配索引:在编译期间,要通过<linux/interrupt.h>中建立的一个枚举类型来静态地声明软中断。内核用这些从0开始的索引来表示一种相对优先级,
  索引号越小就越先执行。所以,可以根据自己的需要将自己的索引号放在合适的位置。
2.注册处理程序:接着,在运行时通过调用open_softirq()注册中断处理程序,例如网络子系统,如下:
1
2
open_softirq(NET_TX_SOFTIRQ,net_tx_action,NULL);
open_softirq(NET_RX_SOFTIRQ,net_rx_action,NULL);
  函数有三个参数,软中断索引号,处理函数和data域存放的数组。软中断处理程序执行的时候,允许响应中断,但它自己不能休眠。在一个处理程序运
  行的时候,当前处理器的软中断被禁止,但其他的处理器仍可以执行别的软中断。实际上,如果一个软中断在它被执行的时候同时再次被触发了,那么
  另外一个处理器可以同时运行其处理程序。这意味着对共享数据都需要严格的锁保护。大部分软中断处理程序都通过采取单处理数据(仅属于某一个处
  理器的数据)或者其他一些技巧来避免显示的加锁,从而提供更出色的性能。(这么说来,不要求中断处理程序为可重入的,但要求软中断可重入的。
3.触发软中断:经过上面两项,新的软中断处理程序就能够运行,raist_softirq(中断索引号)函数可以将一个软中断设置为挂起状态,从而让它在下次调
  用do_softirq()函数时投入运行。该函数在触发一个软中断之前先要禁止中断,触发后再恢复回原来的状态,如果中断本来就已经被禁止了,那么可以调用另一函数raise_softirq_irqoff(),这会带来一些优化效果。

       在中断处理程序中触发软中断是最常见的形式,中断处理程序执行硬件设备的相关操作,然后触发相应的软中断,最后退出。内核在执行完中断处理程序后,马上就会调用do_softirq()函数。于是,软中断就开始执行中断处理程序留给它去完成的剩下任务。

       2.Tasklets:tasklet是通过软中断实现的,所以它们本身也是软中断。它由两类软中断代表:HI_SOFTIRQ和TASKLET_SOFTIRQ.区别在于前者会先于后者执行。Tasklets由tasklet_struct结构表示,每个结构体单独代表一个tasklet,在linux/interrupt.h中定义:

1
2
3
4
5
6
7
structtasklet_struct{
    structtasklet_struct *next;
    unsignedlongstate;
    atomic_t count;
    void(*func)(unsignedlong);
    unsignedlongdata;
};

       结构体中func成员是tasklet的处理程序,data是它唯一的参数。state的取值只能是0,TASKLET_STATE_SCHED(表明tasklet已经被调度,正准备投入运行)和TASKLET_STATE_RUN(表明该tasklet正在运行,它只有在多处理器的系统上才会作为一种优化来使用,单处理器系统任何时候都清楚单个tasklet是不是正在运行)之间取值。count是tasklet的引用计数器,如果不为0,则tasklet被禁止,不允许执行;只有当它为0时,tasklet才被激活,并且被设置为挂起状态时,该tasklet才能够执行。已调度的tasklet(等同于被触发的软中断)存放在两个单处理器数据结构:tasklet_vec(普通tasklet)和task_hi_vec高优先级的tasklet)中,这两个数据结构都是由tasklet_struct结构体构成的链表,链表中的每个tasklet_struct代表一个不同的tasklet。Tasklets由tasklet_schedule()和tasklet_hi_schedule()进行调度,它们接收一个指向tasklet_struct结构的指针作为参数。两个函数非常相似(区别在于一个使用TASKLET_SOFTIRQ而另外一个使用HI_SOFTIRQ).有关tasklet_schedule()的操作细节:

1.检查tasklet的状态是否为TASKLET_STATE_SCHED.如果是,说明tasklet已经被调度过了,函数返回。
2.保存中断状态,然后禁止本地中断。在执行tasklet代码时,这么做能够保证处理器上的数据不会弄乱。
3.把需要调度的tasklet加到每个处理器一个的tasklet_vec链表或task_hi_vec链表的表头上去。
4.唤起TASKLET_SOFTIRQ或HI_SOFTIRQ软中断,这样在下一次调用do_softirq()时就会执行该tasklet。
5.恢复中断到原状态并返回。

       那么作为tasklet处理的核心tasklet_action()和tasklet_hi_action(),具体做了些什么呢:

1.禁止中断,并为当前处理器检索tasklet_vec或tasklet_hi_vec链表。
2.将当前处理器上的该链表设置为NULL,达到清空的效果。
3.运行相应中断。
4.循环遍历获得链表上的每一个待处理的tasklet。
5.如果是多处理器系统,通过检查TASKLET_STATE_RUN来判断这个tasklet是否正在其他处理器上运行。如果它正在运行,那么现在就不要执行,跳
   到下一个待处理的tasklet去。
6.如果当前这个tasklet没有执行,将其状态设置为TASKLETLET_STATE_RUN,这样别的处理器就不会再去执行它了。
7.检查count值是否为0,确保tasklet没有被禁止。如果tasklet被禁止,则跳到下一个挂起的tasklet去。
8.现在可以确定这个tasklet没有在其他地方执行,并且被我们设置为执行状态,这样它在其他部分就不会被执行,并且引用计数器为0,现在可以执行
   tasklet的处理程序了。
9.重复执行下一个tasklet,直至没有剩余的等待处理的tasklets。

       说了这么多,我们该怎样使用这个tasklet呢,这个我在linux设备驱动理论帖讲的太多了。但别急,下边为了博客完整,我仍然会大致讲讲:

       1.声明自己的tasklet:投其所好,既可以静态也可以动态创建,这取决于选择是想有一个对tasklet的直接引用还是间接引用。静态创建方法(直接引用),可以使用下列两个宏的一个(在linux/interrupt.h中定义):

1
2
DECLARE_TASKLET(name,func,data)
DECLARE_TASKLET_DISABLED(name,func,data)

       这两个宏之间的区别在于引用计数器的初始值不同,前面一个把创建的tasklet的引用计数器设置为0,使其处于激活状态,另外一个将其设置为1,处于禁止状态。而动态创建(间接引用)的方式如下:

1
tasklet_init(t,tasklet_handler,dev);

       2.编写tasklet处理程序:函数类型是void tasklet_handler(unsigned long data).因为是靠软中断实现,所以tasklet不能休眠,也就是说不能在tasklet中使用信号量或者其他什么阻塞式的函数。由于tasklet运行时允许响应中断,所以必须做好预防工作,如果新加入的tasklet和中断处理程序之间共享了某些数据额的话。两个相同的tasklet绝不能同时执行,如果新加入的tasklet和其他的tasklet或者软中断共享了数据,就必须要进行适当地锁保护。

       3.调度自己的tasklet:前边的工作一做完,接下来就剩下调度了。通过一下方法实现:tasklet_schedule(&my_tasklet).在tasklet被调度以后,只要有合适的机会就会得到运行。如果在还没有得到运行机会之前,如果有一个相同的tasklet又被调度了,那么它仍然只会运行一次。如果这时已经开始运行,那么这个新的tasklet会被重新调度并再次运行。一种优化策略是一个tasklet总在调度它的处理器上执行。调用tasklet_disable()来禁止某个指定的tasklet,如果该tasklet当前正在执行,这个函数会等到它执行完毕再返回。调用tasklet_disable_nosync()也是来禁止的,只是不用在返回前等待tasklet执行完毕,这么做安全性就不咋嘀了(因为没法估计该tasklet是否仍在执行)。tasklet_enable()激活一个tasklet。可以使用tasklet_kill()函数从挂起的对列中去掉一个tasklet。这个函数会首先等待该tasklet执行完毕,然后再将其移去。当然,没有什么可以阻止其他地方的代码重新调度该tasklet。由于该函数可能会引起休眠,所以禁止在中断上下文中使用它。

       接下来的问题,我在前边说过,对于软中断,内核会选择几个特殊的实际进行处理(常见的是中断处理程序返回时)。软中断被触发的频率有时会很好,而且还可能会自行重复触发,这带来的结果就是用户空间的进程无法获得足够的处理器时间,因为处于饥饿状态。同时,如果单纯的对重复触发的软中断采取不立即处理的策略也是无法接受的。两种极端但完美的情况是什么样的呢:

1.只要还有被触发并等待处理的软中断,本次执行就要负责处理,重新触发的软中断也在本次执行返回前被处理。问题在于,用户进程可能被忽略而使其
   处于饥饿状态。
2.选择不处理重新触发的软中断。在从中断返回的时候,内核和平常一样,也会检查所有挂起的软中断并处理它们,但是,任何自行重新触发的软中断都
   不会马上处理,它们被放到下一个软中断执行时机去处理。问题在于新的或重新触发的软中断必要要等一定的时间才能被执行。

       我现在想的是来个折衷方案吧,那多好,内核开发者门还真是想到了。内核选中的方案是不会立即处理重新触发的软中断,作为改进,当大量软中断出现的时候,内核会唤醒一组内核线程来处理这些负载。这些线程在最低优先级上运行(nice值为19)。这种这种方案能够保证在软中断负担很重的时候用户程序不会因为得不到处理时间而处理饥饿状态。相应的,也能保证“过量”的软中断终究会得到处理。最后,在空闲系统上,这个方案同样表现良好,软中断处理得非常迅速(因为仅存的内存线程肯定会马上调度)。为了保证只要有空闲的处理器,它们就会处理软中断,所以给每个处理器都分配一个这样的线程。所有线程的名字都叫做ksoftirad/n,区别在于n,它对应的是处理器的编号。一旦该线程被初始化,它就会执行类似下面这样的死循环:

1
2
3
4
5
6
7
8
9
10
11
for(;;){
    if(!softirq_pending(cpu))
        schedule();
    set_current_state(TASK_RUNNING);
    while(softirq_pending(cpu)){
        do_softirq();
        if(need_resched())
            schedule();
    }
    set_current_state(TASK_INTERRUPTIBLE);
}

       softirq_pending()负责发现是否有待处理的软中断。当所有需要执行的操作都完成以后,该内核线程将自己设置为TASK_INTERRUPTIBLE状态,唤起调度程序选择其他可执行进程投入运行。最后,只要do_softirq()函数发现已经执行过的内核线程重新触发了自己,软中断内核线程就会被唤醒。

工作队列和前面讨论的其他形式都不相同,它可以把工作推后,交由一个内核线程去执行----该工作总是会在进程上下文执行。这样,通过工作队列执行代码能占尽进程上下文的所有优势,最重要的就是工作队列允许重新调度甚至是睡眠。相比较前边两个,这个选择起来就很容易了。我说过,前边两个是不允许休眠的,这个是允许休眠的,这就很明白了是不?这意味着在你需要获得大量内存的时候,在你需要获取信号量时,在你需要执行阻塞式的I/O操作时,它都会非常有用(先说话, 这个不是我说的,是书上这么说的哦)。

       工作队列子系统是一个用于创建内核线程的接口,通过它创建的进程负责执行由内核其他部分排到队列里的任务。它创建的这些内核线程被称作工作者线程(worker threads).工作队列可以让你的驱动程序创建一个专门的工作者线程来处理需要推后的工作。不过,工作队列子系统提供了一个缺省的工作者线程来处理这些工作。因此,工作队列最基本的表现形式就转变成一个把需要推后执行的任务交给特定的通用线程这样一种接口。缺省的工作线程叫做event/n.每个处理器对应一个线程,这里的n代表了处理器编号。除非一个驱动程序或者子系统必须建立一个属于自己的内核线程,否则最好还是使用缺省线程。

       1.工作这线程结构用下面的结构表示:

1
2
3
structworkqueue_struct{
    structcpu_workqueue_struct cpu_wq[NR_CPUS];
}

       结构中数组的每一项对应系统的一个CPU.接下来,在看看在kernel/workqueue.c中的核心数据结构cpu_workqueue_struct:

1
2
3
4
5
6
7
8
9
10
structcpu_workqueue_struct{
    spinlock_t lock;
    atomic_t nr_queued;
    structlist_head worklist;
    wait_queue_head_t more_work;
    wait_queue_head_t work_done;
    structworkqueue_struct *wq;
    task_t *thread;
    structcompletion exti;
}

       2.表示工作的数据结构:所有的工作者线程都是用普通的内核线程来实现的,它们都要执行worker_thread()函数。在它初始化完以后,这个函数执行一个死循环执行一个循环并开始休眠,当有操作被插入到队列的时候,线程就会被唤醒,以便执行这些操作。当没有剩余的时候,它又会继续休眠。工作有work_struct(linux/workqueue)结构表示:

1
2
3
4
5
6
7
8
structwork_struct{
    unsignedlongpending;
    structlist_head entry;      //连接所有工作的链表
    void(*func)(void*);        //处理函数
    void*data;          //传递给处理函数的参数
    void*wq_data;
    structtimer_list timer;     //y延迟工作队列所用到的定时器
}

       当一个工作线程被唤醒时,它会执行它的链表上的所有工作。工作一旦执行完毕,它就将相应的work_struct对象从链表上移去,当链表不再有对象时,它就继续休眠。woker_thread函数的核心流程如下:

1
2
3
4
5
6
7
8
9
10
11
for(;;){
    set_task_state(current,TASK_INTERRUPTIBLE);
    add_wait_queue(&cwq->more_work,&wait);
    if(list_empty(&cwq->worklist))
        schedule();
    else
        set_task_state(current,TASK_RUNNING);
    remove_wait_queue(&cwq->more_work,&wait);
    if(!list_empty(&cwq->worklist))
        run_workqueue(cwq);
}

       分析一下上面的代码。首先线程将自己设置为休眠状态并把自己加入等待队列。如果工作对列是空的,线程调用schedule()函数进入睡眠状态。如果链表有对象,线程就将自己设为运行态,脱离等待队列。然后,再次调用run_workqueue()执行推后的工作。好了,接下来,问题就纠结在run_workqueue(),它完成实际推后到此的工作:

1
2
3
4
5
6
7
8
while(!list_empty(&cwq->worklist)){
    structwork_struct *work = list_entry(cwq->worklist.next,structwork_struct,entry);
    void(*f)(void*) = work->func;
    void*data = work->data;
    list_del_init(cwq->worklist.next);
    clear_bit(0,&work->pending);
    f(data);
}

       该函数循环遍历链表上每个待处理的工作,执行链表上每个结点上的work_struct的func成员函数:

1.当链表不为空时,选取下一个节点对象。
2.获取我们希望执行的函数func及其参数data。
3.把该结点从链表上接下来,将待处理标志位pending清0。
4.调用函数。
5.重复执行。

       老师说的好:光说不练,不是好汉。现在我们继续来看看怎么用吧:

       1.首先,实际创建一些需要推后完成的工作,可以在编译时静态地创建该数据结构:

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

当然了,如果愿意,我们当然可以在运行时通过指针动态创建一个工作:

1
INIT_WORK(structwork_struct *work,void(*func)(void*),void*data);

       2.工作队列处理函数,会由一个工作者线程执行,因此,函数会运行在进程上下文中,默认情况下,允许相应中断,并且不持有锁。如果需要,函数可以睡眠。需要注意的是,尽管处理函数运行在进程上下文中,但它不能访问用户空间,因为内核线程在用户空间没有相应的内存映射。函数原型如下:

1
void work_hander(void *data);

       3.对工作进行调度。前面的准备工作做完以后,下面就可以开始调度了,只需调用schedule_work(&work).这样work马上就会被调度,一旦其所在的处理器上的工作者线程被唤醒,它就会被执行。当然如果不想快速执行,而是想延迟一段时间执行,按就用schedule_delay_work(&work,delay);delay是要延迟的时间节拍,后面讲.

       4.刷新操作。插入队列的工作会在工作者线程下一次被唤醒的时候执行。有时,在继续下一步工作之前,你必须保证一些操作已经执行完毕等等。由于这些原因,内核提供了一个用于刷新指定工作队列的函数:void flush_scheduled_work(void); 这个函数会一直等待,直到队列中所有的对象都被执行后才返回。在等待所有待处理的工作执行的时候,该函数会进入休眠状态,所以只能在进程上下文中使用它。需要说明的是,该函数并不取消任何延迟执行的工作。取消延迟执行的工作应该调用:int cancel_delayed_work(struct work_struct *work);这个函数可以取消任何与work_struct 相关挂起的工作。

       5.创建新的工作队列。前边说过最好使用缺省线程,可如果你坚持要使用自己创建的线程,咋办?这时你就应该创建一个新的工作队列和与之相应的工作者线程,方法很简单,用下面的函数:struct workqueue_struct *create_workqueue(const char *name);name是新内核线程的名字。这样就会创建所有的工作者线程(系统中的每个处理器都有一个)并且做好所有开始处理工作之前的准备工作。在创建之后,就调用下面的函数吧:

1
2
int queue_work(struct workqueue_struct *wq, struct work_struct *work);
int queue_delayed_work(struct workqueue_struct *wq,struct work_struct *work,unsigned long delay);

       这两个函数和schedule_work()和schedule_delayed_work()相近,唯一的区别在于它们可以针对特定的工作队列而不是缺省的event队列进行操作。

       好了,工作队列也说完了,我还是结合前边一篇,把这三个地板不实现的策略比较一下,方便以后选择.

       首先,tasklet是基于软中断实现的,两者相近,工作队列机制与它们完全不同,靠内核线程来实现。软中断提供的序列化的保障最少,这就要求中断处理函数必须格外小心地采取一些步骤确保共享数据的安全,两个甚至更多相同类别的软中断有可能在不同的处理器上同时执行。如果被考察的代码本身多线索化的工作做得非常好,它完全使用单处理器变量,那么软中断就是非常好的选择。对于时间要求严格和执行效率很高的应用来说,它执行的也最快。否则选择tasklets意义更大。tasklet接口简单,而且两种同种类型的tasklet不能同时执行,所以实现起来也会简单一些。如果需要把任务推迟到进程上下文中完成,那你只能选择工作队列了。如果不需要休眠,那软中断和tasklet可能更合适。另外就是工作队列造成的开销最大,当然这是相对的,针对大部分情况,工作队列都能提供足够的支持。从方便度上考虑就是:工作队列,tasklets,最后才是软中断。我们在做驱动的时候,关于这三个下半部实现,需要考虑两点:首先,是不是需要一个可调度的实体来执行需要推后完成的工作(即休眠的需要),如果有,工作队列就是唯一的选择,否则最好用tasklet。性能如果是最重要的,那还是软中断吧。

       最后,就是一些禁止下半部的相关部分了,给一个表:

函数

描述

void local_bh_disable() 禁止本地处理器的软中断和tasklet的处理
void local_bh_enable() 激活本地处理器的软中断和tasklet的处理

       这些函数有可能被嵌套使用----最后被调用的local_bh_enable()最终激活下半部。函数通过preempt_count为每个进程维护一个计数器。当计数器变为0时,下半部才能够被处理。因为下半部的处理已经被禁止了,所以local_bh_enable()还需要检查所有现存的待处理的下半部并执行它们。

参考:http://www.cnblogs.com/hanyan225/archive/2011/07/20/2111522.html

linux内核分析--中断中的下半部,古老的榕树,5-wow.com

郑重声明:本站内容如果来自互联网及其他传播媒体,其版权均属原媒体及文章作者所有。转载目的在于传递更多信息及用于网络分享,并不代表本站赞同其观点和对其真实性负责,也不构成任何其他建议。