Chinaunix首页 | 论坛 | 博客
  • 博客访问: 33942
  • 博文数量: 6
  • 博客积分: 511
  • 博客等级: 下士
  • 技术积分: 72
  • 用 户 组: 普通用户
  • 注册时间: 2010-10-19 08:49
文章分类
文章存档

2011年(3)

2010年(3)

分类: LINUX

2011-01-28 09:58:39

下半部机制与延迟执行的工作

 

前面曾经讨论过关于中断处理,内核处理硬件的机制。中断处理对任何系统来说都是非常重要的,它是操作系统必不可少的一部分。由于不同方面的限制,中断处理程序仅能形成中断处理的第一部分,以下为这些限制:

1,              中断处理是异步的,它中断其它正在执行的代码,而那些被中断的代码可能是非常重要的,本身又中断了其它代码。因此要避免中断代码的长期执行,中断应该尽可能快的运行完。

2,              中断最好在运行的运行的时候禁止了当前中断,最坏的情况下禁止了当前cpu上的所有中断。这样禁止了中断也就禁止了硬件与操作系统的通讯,因此中断处理应该尽可能快的执行。

3,              由于中断处理硬件,因此它对时间的要求是十分苛刻的。

4,              中断不运行在进程上下文,因此它不可能睡眠,这也限制了它们可以完成的事件。

 

显然,中断处理仅是管理硬件中断方法的一方面,操作系统明显需要一种快速,简单,异步,可以立即响应硬件,而且完成任何关键时间的任务。中断处理可以很好的完成这个任务,但其它对于时间要求不重要的工作,均会推迟至中断使能后的某个时间执行。

那么相应的中断管理就被分成两个部分。第一部分,中断处理程序,上半部,由内核异步的立即的响应硬件中断,这在以上章节中有所描述,以下对下半部机制进行描述。

 

下半部机制

       下半部机制是为了完成任何中断相关的工作,这些工作不是由中断处理程序所完成。理想的世界中,这基本上是所有的工作,因为我们尽可能的让中断处理程序执行最少的工作,其它的推给下半部分。

       然而,中断处理必须完成一部分工作。例如,中断必须应答硬件,确认中断的接收。它也可能拷贝数据至硬件,或从硬件拷贝数据。这项工作是时间相关的,它必须要在中断处理中完成。

       其它任何事情几乎均在下半部中完成。例如,你在上半部中拷贝数据至内存,那么在下半部中你就应该来处理这部分数据。不幸的是,没有任何关于如何分解上下半部分的依据,这一切都取决于驱动程序的写作者。我们仅需要记住中断处理是异步发生的,要尽量保持最少的被禁止的中断线,因此最小化他的持续时间是非常重要的。以下为如何分配上下部分的基本准则:

1,  如果工作对时间非常敏感,那么在中断中完成它;

2,  如果工作与硬件相关,那么在中断中完成它;

3,  如果工作要保证不被其它中断所中断,那么在中断中完成它;

4,  其它的工作均可以推迟至下半部中;

总之,中断执行的越快越好。

 

那么什么时下半部呢

       理解推迟的工作,及推迟到什么时候执行是至关重要的。因为中断处理运行的时候,当前的中断在所有处理器上是被禁止,你要限制中断处理中完成的工作。更为糟糕的是,当一个中断带有IRQF_DISABLE时,那么所有本地cpu上的中断均被禁止。最小化关中断时间,对系统的响应和系统的性能至关重要。再加上另一个实事,中断是异步发生的,它要中断其它未知的程序,那么显然你应该最小化中断执行时间。但何时是“稍后”呢?重要的是记住不是当前,下半部不是把工作推迟至到指定的将来的时间点,只是推迟工作至系统不是太忙,中断被打开的时候来执行。通常下半部在中断发生后立即运行,关键在于下半部运行的时候中断均是打开的。不只是linux处理硬件的时候把任务分为两部分;多数操作系统也是如此。上半部快速简单执行,它们执行时可能部分中断或全部中断均被禁止;下半部之后运行,运行时所有中断被打开。这种设计保持低的延迟,禁止中断尽可能的短。

 

下半部的环境

       不像上半部完全在中断中执行,在下半部中有多种可以使用的机制。这些机制有不同的接口及了系统,以方便我们使用下半部机制。相对于前章节只是单单描述中断处理,本节将会讲述几种下半部机制。在linux的历史中,有几种下半部机制。让人不解的是,这些机制有相似的名字。

 

工作队列

       内核中定义了一系统的队列,每个队列均包括所链接要调用的函数。队列的函数在一定的时间运行,这取决于它所在的队列。驱动开发者可以注册合适的下半部队列。

 

Softirqstasklets

       Softirqs是一系列的静态定义的下半部机制,它们可以同时运行于任何一个cpu;即使两个同样的softirqs也可以同时执行。Tasklets,有着可怕及让人疑惑的名字,使用更方便,它在softirqs上动态的创建下半部。两个不同的的talsklets可以在不同的cpu上同时运行,但两个同类型的tasklets不可能同时运行。因此,tasklets是性能与易用性的两者较好的折中。对于多数下半部处理,tasklet是非常有效的。Softirqs对于性能非常重要的使用,也是非常有用的,比如网络。Softirq在使用时需要更仔细,因为两个softirqs可以同时运行。另外softirq必须要静态的在编译的时候注册。相反的,tasklet可以动态的注册。

       在现在的2.6内核中有三种下半部机制:softirqstasklets,工作队列。

 

Softirqs

我们从softirq开始来讨论实际的下半部机制,在实际中softirqs很少直接的被使用;tasklets则是一种更为常见下半部机制。然而,因为tasklets是建立在softirqs上,我们首先来看softirqs,它位于内核代码下kernel/softirq.c文件中。

 

使用softirqs

       Softirqs在编译的时候被静态的分配,这与tasklets完全不同。Softirqs由结构体softirq_action描述:

struct softirq_action {

void (*action)(struct softirq_action *);

};

一个32个这样的数组描述了内核中的softirqs

static struct softirq_action softirq_vec[NR_SOFTIRQS];

每个注册的softirq使用了这个数组中的一个数据,注册的softirqs在编译时被静态的决定,在当前的内核中强制限制了32个注册的softirqs

 

 

Softirq 处理程序

    Softirq的处理程序原型如下:

           void softirq_handler(struct softirq_action *)

当内核运行一个softirq处理程序时,它执行这个函数时,带有一个参数,这个参数指向相应的softirq_action。例如:如果my_softirq指向softirq_vec中的一个入口时,内核将会如下调用处理函数

       Mysoftirq->actionmy_softirq;

    看起来这个有点复杂,内核传递整个结构给softirq处理程序。这个技巧使得未来在结构体中的修改无需改变每个softirq处理程序。

    一个softirq永远不会抢占另外一个softirq。仅有一个事件可以抢占softirq,那就是中断处理程序。其它的softirq,即便是同类型的softirq可以同时运行于另外一个处理器。

 

Softirqs的执行

    一个注册的softirq必须被标记后才能被执行,这被称为打开softirq,通常在中断处理程序中标记一个softirq,然后才可能去执行这个softirq。在合适的时间,softirq被执行。挂起(标记)的被检查且执行的条件如下:

1,  硬件中断返回时

2,  在内核线程ksoftirqd

3,  在任何明确检查并执行挂起的softirqs中,比如网络子系统中

无论它被调用的方法,softirq__do_softirq()中执行,它又是被do_softirq所调用。这个函数本身非常简单,如果有挂起的softirqs__do_softirq()循环每一个,调用其处理程序,如下:

u32 pending;

pending = local_softirq_pending();

if (pending) {

struct softirq_action *h;

/* reset the pending bitmask */

set_softirq_pending(0);

h = softirq_vec;

do {

if (pending & 1)

h->action(h);

h++;

pending >>= 1;

} while (pending);

}

以上代码为softirq处理的核心,它检查,执行挂起的softirqs,特别的

1,  设置本地变量penging,如果第n位被设置,则表明第nsoftirq挂起

2,  既然挂起位被保存,那么它清除所有挂起的softirq

3,  指针h指向softirq_vec的第一个入口

4,  如果第一个位被设置,那么将会调用h->action(h)将会被调用

5,  然后h递加,指向第二个入口,penging右移一位,本来第二位的变成了第一位。

6,  如果本地变量pending不为零,则循环以上步骤。

7,  直到penging为零,则表明没有挂起的softirq

使用softirqs

    Softirqs被保留,以为了对时间要求更为严格的下半部处理过程。当前,仅有两个子系统网络及块设备直接使用了softirqs。另外,内核定时器及tasklets也是建立在softirq上。如果你要添加一个新的softirq,正常的应该首先问你自己为什么使用一个tasklet是不足的。Tasklet可以动态的创建,且容易使用,因为它们对锁的要求非常弱,且它们的性能也非常好。然而,对于时间要求严格的应用,它们本身使用锁是非常高效的,那么softirq或许是一个正确的办法。

 

分配一个索引号

    在文件linux/interrupt.h中在编译时静态的声明一个枚举类型的softirqs。内核使用这个索引,作为优先级,最高级别为从0开始。越小的数字优先级越高。创建一个softirq包括在枚举类型中添加一项,你可能不会想简单的把你的项加入尾部,你所加入的项目取决于它要执行的优先级。常规条件下,HI_SOFTIRQ总是第一个项目,RCU_SOFTIRQ总是最后一项。

 

注册你的处理程序

    接下来,softirq处理程序在运行时由open_softirq()注册,它只需要两个参数:索引号及它的处理程序。如网络了系统,注册softirqs,在net/core/dev.c文件中,如下:

open_softirq(NET_TX_SOFTIRQ, net_tx_action);

           open_softirq(NET_RX_SOFTIRQ, net_rx_action);

softirq运行的时候,中断是打开的,且不能睡眠。一旦一个处理程序运行的时候,softirqs在当前处理器上被禁止,然而,在另一个处理器上,可以执行其它的softirq如果在执行中,同样的softirq被标记,另一个处理器可以同时运行它。这就意味着共享数据,要使用一种合适的保护机制。简单的避免你的softirq不同时运行,并不是一个好的主意。如果一个softirq获得锁,从而不允许另外一个softirq同时运行,那么使用softirq就没有任何意义。这是重点,也是为什么人们更偏向于使用taskletstasklets本质上是softirq,但对于同一个处理程序无法同时运行在多个处理器上。

 

打开softirq

    在一个处理程序被加入至枚举链表,并通过open_softirq()函数注册,然后等待运行的机会。为了标记挂起,以在下一次调用do_softirq()时运行,调用raise_soft().例如,网络子系列中,将会调用:

       raise_softirq(NET_TX_SOFTIRQ);

这就打开了NET_TX_SOFTIRQ。它的处理函数,net_tx_action(),在下次内核执行softirq时将会运行。Raise_softirq这个函数首先禁止中断,然后标记softirq,最后再恢复至原来的状态。如果中断已经关闭,函数raise_softirq_irqoff()将会是一个比较优化的函数。例如:

              raise_softirq_irqoff(NET_TX_SOFTIRQ);

softirq经常在中断处理中raised,在中断处理中,中断处理程序完成基本的硬件相关的工作,并打开softirq,然后退出中断。处理中断时,内核调用do_softirq().在中断处理停止执行时,softirq开始执行。

 

Tasklets

    Tasklets也是下半部机制,它是建立在softirq基础之上的。如上所提,它与进程无任何关系。Tasklets很自然的与softirq相似,但是它有个简单的接口,对于锁的要求也不是那么严格。作为驱动开发者,使用tasklets或是softirq有个简单的准则:多数情况下你应该使用taskletsSoftirq仅是那些执行频率高性能要求严格的线程在使用。Tasklets则不是,它被广泛使用,且可以很好的完成工作,而且很容易使用。

 

使用tasklets

    因为tasklets的执行是建立在softirq基础之上的,所以他们也是softirqs。如上所述,tasklet是由两个类型的softirq所表示:HI_SOFTIRQTASKLET_SOFTIRQ,它们的不同在于前者总是先于后者执行。

 

Tasklet的数据结构

    Tasklettasklet结构体表示,如下:

struct tasklet_struct {

struct tasklet_struct *next; /* next tasklet in the list */

unsigned long state; /* state of the tasklet */

atomic_t count; /* reference counter */

void (*func)(unsigned long); /* tasklet handler function */

unsigned long data; /* argument to the tasklet function */

};

函数成员是tasklet的处理程序,接收唯一的data作为其输入参数。State成员准确的说为零,TASKLET_STATE_SCHED, TASKLET_STATE_RUNTASKLET_STATE_SCHED表示一个tasklet准备被调度运行,TASKLET_STATE_RUN表明tasklet正在运行。作为一个优化TASKLET_STATE_RUN仅在SMP上被使用。

Count域是用来表明tasklet被引用的次数。如果非零,那么tasklet被禁止且不能运行;如果为零,则tasklet可以被使能,且可以被标志挂起以运行。

 

Tasklets的调度

    调度tasklets(相对于softirqraised)被存储于每个处理器的两个数据结构中:tasklet_vec(常规tasklets)task_hi_vec(高优先级的tasklets)。所有两种结构均被链接在task_struct结构中。每种被链接在表中的tasklet代表一个不同的tasklet

    Tasklets是通过tasklet_schedule()tasklet_hi_schedule()函数进行调度的,它们接受一个指向tasklets的指针作为唯一的参数。每个函数保证所提供的tasklet仍然没有被调用,且调用相应合适的__tasklet_schedule__tasklet_hi_schedule。现在观察一下tasklet_schedule完成的步骤:

1,  核对tasklet的状态是不是TASKLET_STATE_SCHED。如果是,那么tasklet已经被调用准备运行,函数将会立即返回。

2,  调用__tasklet_schedule().

3,  保存中断系统的状态,然后禁止本地中断。这就保证了本地处理器不会打乱tasklet的代码,当tasklet_schedule掌控tasklets时。

4,  tasklet加入至tasklet_vectasklet_hi_vec链表中,每个处理器有各自的链表。

5,  打开TASKLET_SOFTIRQHI_SOFTIRQ,然后do_softirq()在不久的将来可以执行这段代码

6,  将中断恢复至先前的状态,然后返回。

在下一个方便的时间,do_softirq如前章所述开始运行。因为多数taskletssoftirq是在中断处理中被标志挂起。Do_softirq多数情况下运行于最后一个中断处理程序结束时,因为标志位TASKLET_SOFTIRQTHI_SOFTIRQ现在被打开,do_softirq执行相应的中断处理程序。这些处理程序,tasklet_action,tasklet_hi_action,是tasklet处理的核心。以下为处理程序完成的步骤:

1,  禁止本地中断,获取本地cputasklet_vectasklet_hi_vec矢量链表

2,  清除本地cpu中相应的链表

3,  使能中断

4,  循环执行链表中的每个tasklet

5,  如果为多处理器机器,检查tasklet是否运行在另一个处理器上,通过检查TASKLET_STATE_RUN标志,如果当前运行,现在不执行它跳至下一个tasklet

6,  如果tasklet当前未运行,设置stateTASKLET_RUN,这样其它cpu无法运行它

7,  检查count的值是否为0,以保证tasklet没有被禁止,如果tasklet被禁止,跳过它去执行下一个

8,  现在我们了解tasklet没有在其它地方运行,它被标记为运行态,所以不会在其它地方执行,且count为零,现在可以运行tasklet处理程序。

9,  tasklet运行后,清除stateTASKLET_STATE_RUN标志位。

10,              重复下一个挂起的tasklet,直到所有的等待执行的tasklet被执行完。

Tasklet的执行非常简单,但却是明智的。可以看出,所有的tasklets均是建立在两种类型的softirq基础之上的,HI_SOFTIRQTASKLET_SOFTIRQ。当tasklet被调度时,内核打开其中的一个softirq。这些softirq,又是由特别的函数处理,然后运行。这些特别的函数保证仅有一个给定类型的tasklet在运行。

 

Tasklet的使用

    在多数情况下,tasklets对于处理硬件下半部的是一个比较好的机制。它动态创建,方便使用,且可以快速执行。

1,  定义一个tasklet

我们可以动态的或静态的创建tasklets,使用哪个取决于你间接或直接的使用tasklet。如果你静态的创建tasklet(直接的引用),使用以下两个宏中的一个:

DECLARE_TASKLET(name, func, data)

DECLARE_TASKLET_DISABLED(name, func, data);

以上宏均静态的创建的名称为nametasklet_struct结构。当tasklet被调度时,指定的函数被执行,参数被传递。以上两个宏的不同在于初始引用记数的不同。第一个宏的引用记数为0tasklet被使能,第二个宏的引用记数为1tasklet被禁止。以下为例子:

DECLARE_TASKLET(my_tasklet, my_tasklet_handler, dev);

等同的形式如下:

struct tasklet_struct my_tasklet = { NULL, 0, ATOMIC_INIT(0),

my_tasklet_handler, dev };

可以使用tasklet_init()动态的创建一个tasklet_struct结构,如下:

tasklet_init(t, tasklet_handler, dev); /* dynamically as opposed to statically */

2,  tasklet的处理程序

Tasklet处理程序有其原型:

           void tasklet_handler(unsigned long data)

由于tasklet本身为softirq,所以他不可以睡眠。这就意味着,你无法使用信号量或其它可能导致阻塞的函数。Tasklet本身运行的时候中断都是打开的,所以在你的tasklet与中断中有共识的数据时,一定要小心。与softirq不同的是,两个相同的tasklet不可能同时运行-----尽管不同的tasklet可以同时的运行于两个不同的处理器上。如果在你的tasklet之间或与softirq之间有共享数据的时候,你应该使用合适的锁机制。

3,  调度你的tasklet

为了调度你的tasklet执行,tasklet_schedule函数被调用,且传递一个相关的tasklet_struct结构体

        Tasklet_schedule(&my_tasklet);

一旦tasklet被调度,它在未来的某个时间会运行一次。如果相同的tasklet再次被调度,且上次的并没有机会执行,它只会运行一次。如果它正在执行,例如在另一个处理器上运行,这个tasklet将会被再次调度且运行。作为一个优化,一个tasklet总是运行那个调度它运行的处理器上-----为了更好的使用处理器的缓存。

    通过调用tasklet_disable函数,你可以禁止一个tasklet的执行。如果相应的tasklet已经正在执行,它直到执行完才会返回。你也可以使用另外一个函数再禁止给定的tasklet的执行,tasklet_disable_nosync(),它并不等待tasklet的执行结果,直接返回,当然它并不十分安全。调用tasklet_enable将会使能给定的tasklet,这个函数几乎总是在一个使用DECLARE_TASKLET_DISABLE被创建之前,被调用。如:

        Tasklet_disable(&my_tasklet)

 

        Tasklet_enable(&my_tasklet);

你可以使用tasklet_kill来移去一个已经被挂起的tasklet,这个函数接受一个指向tasklet_struct的指针。从队列中移除一个调度的tasklet是非常有用的,当我们处理一个可以进行自我调度的tasklet时。这个函数首先等待指定的tasklet执行完,然后从队列中移除它。没有其它的代码可以阻止一个tasklet再次调度tasklet,这个函数不能在中断上下文中使用,因为它要睡眠。

 

 

Ksoftirqd

    Softirq的处理是由每个处理器的内核线程来协助的,在系统的softirq过多时,这些内核线程来处理softirq。因为tasklet是倚靠softirq来实现的,所以以下讨论同时适用于softirqtasklet。为了简便,我们主要介绍softirqs

    我们已经讲过,内核在很多地方处理softirq,多数是在中断处理程序返回时Softirq可能以很高的速度被打开,如在网络通讯时,数据量较多时,而且,它们可以自激活,也就是,在运行一个softirq时,它可以再次打开一个softirq(如网络子系统在高数据流量的时候softirq再次打开自己)。高速率的softirq以及自激活的特性可能导致用户空间程序的过度饥饿。然而不适时处理再次激活的softirq,是不可容忍的。在softirq设计之初,这就是一个比较为难的问题,而且必须要确定,一直一来没有一个比较好的解决方法。让我们来看看两个比较明显的解决办法。

    第一个方法是简单的保持处理softirq,如果它们到来,再次核查,处理挂起的softirq,然后返回。这保证了内核处理softirq的时间要求,更重要的是任何刚产生的softirq可以立即得到执行。问题出现于在负载比较高的情况下,softirq不停的再次激活自身,内核可能持续的服务于softirq而不能完成其它工作。用户空间被忽略,实际上只有softirq及中断处理程序可以运行,明显用户对此是不能容忍的。这种处理方法在系统负载不重的情况下,可能会很好的工作;如果系统经历可观的中断时,这种处理方法是行不通的。

    另一种处理方法是不处理再次产生的softirq。在中断处理程序返回时,内核仅仅查看挂起的softirq并且常规的去执行它们。如果任何softirq再次激活自身,它们不会被立即执行,直到下一次内核处理挂起的softirq这看起来,直到下一次中断发生时,任何新的softirq才有可能执行,要经历一定长的时间段。糟糕的是,在一个空闲的系统中,应该立即处理softirq,不幸的是这种方法明显忘记哪个进程是可运行的。因此,这种方法避免了用户空间的饥饿,但是它却会导致softirq的饥饿,而且不能很好的利用空闲的系统。

    在设计softirq时,内核开发者意识到有必要进行一定的折中处理。内核中所使用的方法是不立即处理自我激活的softirq。取而代之的是,如果softirq增长过量的时候,内核唤醒softirq处理的线程,以处理softirq,以降低负载。内核线程运行时有着最低的优先级,这保证它们不会代替其它重要的进程执行。这种设计避免了,较多的softirq,完全使用户空间饥饿。相应的,保证了softirq最终的执行。最后,这种方法保证了,在空闲系统上,softirq可以更快的运行,因为内核线程会立即调用它。

    每个处理器均有一个这种线程。它们被命名为ksoftirqd/n,此处n是处理器号。在两处理器机器上,你将会看到ksoftirq/0ksoftirq/1.在每个处理器上保证如果一个空闲的处理器存在,它可以一直服务于softirq。在这些线程被初始化后,它们运行小的循环,如下:

 

for (;;) {

if (!softirq_pending(cpu))

schedule();

set_current_state(TASK_RUNNING);

while (softirq_pending(cpu)) {

do_softirq();

if (need_resched())

schedulee();

}

set_current_state(TASK_INTERRUPTIBLE);

}

如果任何softirq挂起(由函数softirq_pening()报告),ksoftirqd调用do_softirq来处理它们。注意,它确实会再次处理任何自激活的softirq。在每次循环如果有必要,它将会调用schedule(),以使能更重要的进程运行。在所有的进程被处理完以后,内核线程设置自身为TASK_INTERRUPT然后调用scheduler选择一个新的进程运行。

    Softirq进程由do_softirq唤醒,如果do_softirq检测到过多(10次)的内核线程自激活。

 

工作队列

    工作队列与以上推迟工作至以后执行是不同的。工作队列推迟工作至一个内核线程---这个下半部经工作常在进程上下文完成因此,代码推迟至工作队列中,可以使用进程上下文所有的优点,更重要的是:工作队列可以调度,因此可以睡眠。

    通常很容易选择是否使用工作队列或softirq/tasklets。如果推迟的工作需要睡眠,应用使用工作队列;如果推迟的工作不需要睡眠,可以使用softirqstasklets实际上,我们可以创建一个内核线程,取代工作队列。因为内核开发者不赞成创建一个新的内核线程,因此我们推荐使用工作队列,它们确实非常容易使用!

    如果你需要调度一个实体来完成你的下半部分的处理,你需要工作队列。它们是仅有的可以运行于进程上下文的下半部机制,只有它们才可以睡眠。这意味着,将你需要分配大量内存,获取信号量,或进行阻塞I/O操作时,它们是非常有用的。如果你不需要内核线程来处理你的延迟工作,那么考虑使用tasklet

 

工作队列的执行

    工作队列最基本的形式,是创建一个内核线程来从其它地方处理工作,它提花这个接口。这些内核线程被称为工作者线程,工作队列允许你的驱动创建一个特殊的工作者线程以处理延迟的工作。工作队列子系统提供默认的工作者线程来处理你的工作。因此它最为常见的形式,工作队列是为了推迟一个要执行工作至通用内核线程执行的接口。

   默认的工作者线程被称为events/n,此处n为处理器序号,每个处理器一个。例如在单处理器上仅有一个这样的线程events/0.默认的工作者线程可以从多个地方处理延迟的工作,多数驱动把它们下半部工作推迟至默认的内核工作者线程。除非驱动或子系统有着较为严格的要求,需要创建自己的内核线程,否则最好使用默认的内核工作者线程。

    没有什么会阻止代码创建它自己的工作者线程,如果你需要完成大量的处理工作,这可能是一个优点。处理器紧张或性能重要的工作或许会从中获益,而且会降低默认线程的负载,避免其它队列中的工作过度饥饿。

   

线程的数据结构

    工作者线程由workqueue_struct结构体描述:

struct workqueue_struct {

struct cpu_workqueue_struct cpu_wq[NR_CPUS];

struct list_head list;

const char *name;

int singlethread;

int freezeable;

int rt;

};

这个数据结构定义于kernel/workqueue.c中包括了cpu_workqueue_struct结构体数组,每个可能的处理器占用数组中的一个元素。因为工作者线程存在于系统中的每个处理器上,因此每个处理器上有一个这样的工作者线程。Cpu_workqueue_struct是核心结构数据,也被定义于相同的文件中:

struct cpu_workqueue_struct {

spinlock_t lock; /* lock protecting this structure */

struct list_head worklist; /* list of work */

wait_queue_head_t more_work;

struct work_struct *current_struct;

struct workqueue_struct *wq; /* associated workqueue_struct */

task_t *thread; /* associated thread */

};

注意每个类型的工作者线程有相应的workqueue_struct与这相关。在内部,一个cpu_workqueue_struct代表了每个线程,每个处理器上的执行实例,因为每个工作者线程均运行于相应的处理器上。

 

工作的数据结构

所有工作者线程的执行如正常的内核线程运行worker_thread.worker_thread函数被调用初始化后,此函数进入无限循环,然后睡眠。当有工作入队的时候,这个线程被唤醒,然后处理工作;完成工作后,若没有其它工作,则去睡眠。

工作是由work_struct结构表示,定义于

Struct work_struct{

           Atomic_long_t data;

           Struct list_head entry;

           Work_func_t func;

}

这个数据结构链入链表,每个处理器上有相应的一这种类型,例如,内核中每个处理器有一个通用的延迟工作线程。当工作者线程被唤醒时,它运行它链表中的任何工作。它完成相应的工作后,先移除它,当链表空时,它进行睡眠。工作者线程的核心如下:

for (;;) {

prepare_to_wait(&cwq->more_work, &wait, TASK_INTERRUPTIBLE);

if (list_empty(&cwq->worklist))

schedule();

finish_wait(&cwq->more_work, &wait);

run_workqueue(cwq);

}

这个函数在无限循环中进行以下任务:

1,  线程标志自身睡眠(设置自身为TASK_INTERRUPTIBLE)且把自己加入等待队列。

2,  如果链表为空,则线程调度schedule去睡眠。

3,  如果链表非空,工作者线程并不去睡眠。它标记自己为运行态,且从等待队列中移除工作者线程。

4,  如果链表非空,工作者线程调用run_workqueue完成延迟执行的工作。

函数run_workqueue,实际上完成了那些被延迟执行的工作:

while (!list_empty(&cwq->worklist)) {

struct work_struct *work;

work_func_t f;

void *data;

work = list_entry(cwq->worklist.next, struct work_struct, entry);

f = work->func;

list_del_init(cwq->worklist.next);

work_clear_pending(work);

f(work);

}

这个函数循环链表中的每个入口上挂起的工作,然后执行它:

1,  如果链表非空,获取链表中下一入口

2,  获得要执行的函数,参数

3,  移除链表中这个工作的数据结构清除挂起状态

4,  调用工作函数

5,  重复

工作队列执行概要

    在不同的数据结构之间的关系有些复杂,如下:

 

在最高层面上,它们是工作者线程。它们可能是多种类型的工作者线程;每个处理器有一个给定的工作者线程。内核部分可以创建工作者线程,如果有必要。默认情况下,有一个events工作者线程。每个工作者线程由cpu_workqueue_struct结构表示。Workqueue_struct结构表示了给定工作者线程的某个工作者线程类型。

例如,如果除了通用的工作者线程,你需要创建额外的工作者线程,你可以创建自己指定类型的工作者线程。假定你的机器有四个处理器,那么将会有四个默认的线程及四个你所创建的线程,每个处理器上有一个events类型及你所创建的类型的工作者线程。

现在让我们从底层看起,从工作看起。你的驱动创建工作,这些工作需要推迟执行。用结构体work_struct表示这份工作,这个结构体的实例中包括了指向实际延迟要执行的工作的处理函数。这个工作将会提交至你所创建线程中,然后你所创建的线程被唤醒,并完成队列中的工作。

多数驱动使用内核中默认的工作者线程,被称为events的线程,它们是非常简单而且容易使用的。一些比较严格的情况下,要求我们需要自己创建自己的工作者线程。如XFS文件系统,创建了两个新的工作者线程。在同一工作者线程中的工作会因一个工作的睡眠而导致后面的工作被延迟更多后执行。

 

工作队列的使用

    工作队列非常简单,我们首先讨论默认的events工作队列,然后再看看创建一个新的工作者线程。

1,  创建工作

第一步是要首先创建一个要推迟执行的工作,为了静态的创建工作,可以调用

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

这个宏静态的创建了work_struct结构,其名称为name,处理函数为func,且参数为data,相应的你可动态的创建这种数据结构,使用以下宏:

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

       这个函数动态的初始化了work工作,且其处理函数为func,参数为data

2,  工作处理函数

工作队列中工作的处理函数原型为:

    Void work_handler(void *data)

工作者线程执行这个函数,因此这个函数运行于进程上下文,默认情况下,中断是使能的,且没有持有锁。如果需要,这个函数是可以睡眠的。注意,尽管是运行于进程上下文的,但工作者线程无法访问用户空间的内存,因为并没有与之相关的用户空间内存映射至内核线程。内核仅在代表用户空间运行时,也既通过系统调用时进入内核空间时,才可以访问用户空间内存,也只有这时用户空间的存储才被映射至内核线程中。这也是内核线程与用户进程最大之不同处。

3,  调度工作队列

现在工作已经被创建,我们可以调度它,为了把指定的工作处理,加入至默认events工作者线程,简单的调用:

    Schedule_work(&work);

这个函数会被立即调用,只有当前工作者线程被唤醒。有时你可能并不想立即执行,而是延迟指定的时间再执行,这时你可以使用另一种调度方式再调度它在指定的某个时间执行:

    Schedule_delayed_work(&work,delay);

Delay的单位为tick

4,  刷新工作

工作队列中的工作在工作队列被唤醒时才会被执行。有时你需要保证指定的函数执行完,然后才能进行新的工作。这个对于模块来说是非常重要的,模块在卸载前一般都会调用这个函数。内核的其它部分可能需要做一定的工作以保证没有工作挂起,以保证避免竞争的发生。为了完成这个功能,可以调用以下函数:

        Void flush_schedule_work(void);

这个函数等待所有队列中相应函数执行完返回,在等待期间这个函数要进行睡眠。因此你仅能从进程上下文中调用它。注意这个函数并不取消那些被延迟的任何工作,这就意味着那么延迟的工作,未被刷新。为了取消延迟的工作可以调用以下函数:

        Int cancel_delayed_work(struct work_struct *work);

这个函数取消挂起的工作,如果有,任何给定的相关work_strut均被取消。

5,  创建一个新的工作队列

如果默认的工作队列,不能满足你的需要,你可能需要创建一个新工作队列及相应的工作者线程。你可以创建一个工作队列及相应的工作者线程,能过以下函数:

struct workqueue_struct *create_workqueue(const char *name);

        参数name为工作者线程的名称,例如,默认的工作者线程的创建:

struct workqueue_struct *keventd_wq;

keventd_wq = create_workqueue(“events”);

这个函数创建了所有的默认的工作者线程,并且准备让它们处理工作。创建工作的处理是相同的,与工作队列的类型无关。在工作被创建之后,接下的函数就如schedule_workschedule_delayed_work(),除非它们工作在给定的工作队列而不是默认的events队列,可以使用以下函数:

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)

       最终,你可以通过以下函数刷新一个等待的工作队列,强制它们执行

           flush_workqueue(struct workqueue_struct *wq)

如前所述,这个函数与flush_shceduled_work()相同,不过它会等待直到给定的工作例为空才返回。

 

哪个下半部机制我应该使用?

    决定使用哪个下半部机制是非常重要的,在当前2.6内核中,你可以有三个选择:softirqtasklet,及工作队列。Tasklets建立在softirq之上,因此它们相似。工作队列的机制是完全不同的,它建立在内核线程之上。

    Softirq在设计的时候,提供了最少的串行化处理机制。这就要求,在处理softirq的处理函数时,要有额外的步骤来保证共识数据的安全,因为两个或多个同类型的softirq可能同时运行于多个不同的处理器上。

    如果不是线程化,tasklet可能会更有意义。它们有简单的接口,而且两个相同类型的tasklet不会同时运行,更容易使用。Tasklet是高效的不能并行的softirqs。一个驱动的开发者应该更倾向于使用tasklet,而非softirq,除非准备利用每个处理器变量或相似的约数保证softirq能安全的运行于多处理器上,才使用softirq

    如果你的延迟工作要求运行于进程上下文,你只能选择工作队列这种下半部机制。如果进程上下文,并不是必须的,特别是情况,如果你不需要睡眠,或许softirqtasklet会是更好的选择。工作队列涉及到很多内核方面的东西,比如内核线程,进程上下文切换等,但这并不是说它们效率低,但是在每秒钟有上千次网络中断的系统中,其它方法或许更为合适。在多数情况下工作队列均是有效的。

    下半部机制比较

下半部                             内在串行化

Softirq           软                

Tasklet                             不能运行相同的tasklet

Work queues                         

  上下文中是不能睡眠的,原因在于Linux的软中断实现上下文有可能是中断上下文(也有可能为进程上下文,如最后在ksoftirqd内核线程中执行),如果在中断上下文中睡眠,那么会导致Linux无法调度,直接的反应是系统Kernel Panic,并且提示dequeue_task出错。__do_IRQ完成之后,返回到do_IRQ函数,在该函数中调用了一个非常重要的函数irq_exit(),在该函数中调用invoke_softirq(),invoke_softirq调用do_softirq()函数,执行软中断的操作。此时,程序的执行环境还是中断上下文,但是与中断上半部不同的是,软中断执行过程中是开中断的,能够被硬中断而中断。所以,如果用户的程序在软中断中睡眠,操作系统该如何调度呢?只有kernel panic了。

 

在下半部之间的锁机制

    锁用来保护共享数据,防止在下半部之间同时访问数据引起的问题,下半部可以在任何时间运行。

    Tasklet由于本身的串行化处理,带来了一个好处,同一个tasklet不会同时运行,即使在两个cpu上。这意味着你不需要担心内部同时发生的事情。在不同的tasklet内部同时发生的事情要求使用锁的保护机制,如果他们之间有共享的数据。

    由于softirq并无串行化的处理,既使两个相同的softirq可能同时运行于不同的cpu上,所以所有需要共享的数据要有合理的锁机制。

    如果进程上下文代码的处理中与下半部中共享数据,那么你应该禁止下半部处理过程且在访问数据前要获取锁。完成这些主要是为了对SMP的保护和避免死锁。

    如果在中断上下文代码及下半部机制中共识数据,你需要禁止中断且在访问数据前获取锁。这也保证了本地及SMP保护且避免了死锁的发生。

    所有在工作队列中共享的数据需要使用锁,使用锁的机制与内核线程一致,因为它本身就是一个内核线程,且运行在进程上下文。

 

禁止下半部的执行

    一般情况下仅仅禁止下半部的执行,并不是很有效。更为常见是的,为了保护共享数据,你需要获得锁且禁止下半部的执行。为了禁止所有下半部的执行,调用local_bh_disable,使用函数local_bh_enable可以再次使能下半部的执行。

    仅有最后一个配对的local_bh_enable才真正使能下半部的执行。例如,不我们第一次去调用local_bh_disable时,下半部被禁止,如果再有三次调用local_bh_disable,当前的下半部仍然被禁止。除非你调用4local_bh_enable函数,否则下半部不会被再次使能。

 

    此函数借助它所保持的每个任务的计数preempt_count来完成这个功能的。当这个计数为0时,下半部才有机会执行,因为下半部被禁止,local_bh_enable也检查任何挂起的下半部,并执行它。

   

    这个函数对于支持的体系且有点特别,且通常均是使用复杂的宏来完成,位于asm/softirq.h,以下:

 

/*

* disable local bottom halves by incrementing the preempt_count

*/

void local_bh_disable(void)

{

struct thread_info *t = current_thread_info();

t->preempt_count += SOFTIRQ_OFFSET;

}

 

/*

* decrement the preempt_count - this will ‘automatically’ enable

* bottom halves if the count returns to zero

* optionally run any bottom halves that are pending

*/

void local_bh_enable(void)

{

struct thread_info *t = current_thread_info();

t->preempt_count -= SOFTIRQ_OFFSET;

/*

* is preempt_count zero and are any bottom halves pending?

* if so, run them

*/

if (unlikely(!t->preempt_count && softirq_pending(smp_processor_id())))

do_softirq();

}

    这些函数的调用,并不禁止工作队列的执行,因为工作队列运行于进程上下文,它们与异步的执行无任何关系,也没有必要去禁止它们。因为softirqtasklet可以异步的执行(它们执行始于中断处理程序的返回),因此内核代码需要禁止它们。在工作队列中为了保护共享的数据,所使用的机制就如进程上下文中所使用的一致。

 

 

 

参考文献

Lkd 3rd        Chapter 8 Bottom Halves and Deferring Work

更为详细的内容

http://blog.csdn.net/sailor_8318/archive/2008/07/16/2657294.aspx     工作队列

http://blog.csdn.net/sailor_8318/archive/2008/07/13/2645186.aspx     tasklet

http://blog.csdn.net/sailor_8318/archive/2008/07/13/2645180.aspx       softirq

阅读(2881) | 评论(0) | 转发(0) |
0

上一篇:定时器及时间管理

下一篇:没有了

给主人留下些什么吧!~~