By judicious@newsmth
4 Softirq、下半部(Bottom Halves)以及Tasklets的设计思想
4.1 设计思想:内核在进行中断处理时,要求相应的中断处理程序迅速处理完,这就要求中断处理程序处理的尽量快,而把尽可能多的处理向后推,为了实现这个目标,内核提供三种实现方法Softirq、下半部(Bottom Halves)、Tasklets。
这三者之间的关系:a)Softirq是最底层机制,Tasklets工作在它的基础之上,下半部是一种特殊的Tasklets。b)在UP(unique processor)系统中,不能同时处理两个Softirq,也就说不能为了执行一个Softirq而去中断另外一个Softirq。下半部和Tasklets
4.2 Softirq是一种延时机制,不要同可编程异常(Programmable Exception,在Intel手册中称之为软中断,主要是因为这种中断是有CPU发出的指令实现的,所以称之为软中断)弄混淆。
Softirq要求是可重入的,而Tasklets并不要求是可重入的,所以在大多数的情况下,使用Tasklets就可以了,由于下半部是全局锁定(globally locked)的,所以使用的也不是很多,在内核中将会逐渐消失。内核中实现了四种Softirqs,包括:
0,HI_SOFTIRQ,Handles high-priority tasklets and bottom halves
1,NET_TX_SOFTIRQ,Transmits packets to network cards
2,NET_RX_SOFTIRQ,Receives packets from network cards
3,TASKLET_SOFTIRQ,Handles tasklets
上面的index(0、1、2、3)代表了Softirq的优先级的不同,index值越小,其优先级越高。
4.3 Tasklets是建立在Softirq之上的,前面已经说过,它与Softirq之间的主要区别在于:’Tasklet is running only on one CPU simultaneously.’(linux/interrupt.h)。由于Tasklets不必要求是可重入的,所以在内核中它是最优先考虑使用的方法,更由于它是动态加载的,在模块尤其是模块驱动程序中也是唯一的选择,所以特别重要。
4.3.1Tasklets主要的数据结构:
struct tasklet_struct
{
struct tasklet_struct *next;
unsigned long state;
atomic_t count;
void (*func)(unsigned long);
unsigned long data;
};
如果要在一个驱动程序模块中使用Tasklets,我们需要如何做呢?(以《Linux device driver(2nd edition)》的short程序为例)
1, 声明和初始化:
#define DECLARE_TASKLET(name, func, data) \
struct tasklet_struct name = { NULL, 0, ATOMIC_INIT(0), func, data }
例如:
void short_do_tasklet (unsigned long);
DECLARE_TASKLET (short_tasklet, short_do_tasklet, 0);
2, 激活Tasklets:
tasklet_schedule();
例如:
void short_tl_interrupt(int irq, void *dev_id, struct pt_regs *regs)
{
do_gettimeofday((struct timeval *) tv_head); /* cast to stop
’volatile’ warning */
short_incr_tv(&tv_head);
tasklet_schedule(&short_tasklet);
short_bh_count++; /* record that an interrupt arrived */
}
实际的下半部routine short_do_tasklet()如下,它将会在系统合适的时间里运行a) schedule() and b) 中断或者系统调用返回
void short_do_tasklet (unsigned long unused)
{
int savecount = short_bh_count, written;
short_bh_count = 0; /* we have already been removed from queue */
/*
* The bottom half reads the tv array, filled by the top half,
* and prints it to the circular text buffer, which is then consumed
* by reading processes
*/
/* First write the number of interrupts that occurred before
this bh */
written = sprintf((char *)short_head,"bh after %6i\n",savecount);
short_incr_bp(&short_head, written);
/*
* Then, write the time values. Write exactly 16 bytes at a time,
* so it aligns with PAGE_SIZE
*/
do {
written = sprintf((char *)short_head,"%08u.%06u\n",
(int)(tv_tail->tv_sec % 100000000),
(int)(tv_tail->tv_usec));
short_incr_bp(&short_head, written);
short_incr_tv(&tv_tail);
} while (tv_tail != tv_head);
wake_up_interruptible(&short_queue); /* wake any reading process */
}
4.4 下半部的设计思想是把中断处理程序分成两个部分:上半部分(Top Half)和下半部分(Bottom Half)。其中上半部分被立即处理,下半部分根据其类型在不同的时候处理,但是下半部总是在开中断的环境下被处理的,所以它是运行在一个”安全的时间”里的。
内核对Bottom Halves加锁(spinlock,就是说一个cpu只能同时处理一个Bottom Half,同时也就说明Bottom Halves不能被阻塞),并且一个系统最多只能有32个,同时一个Bottom Half只能处理一个Handler(以上就是BottomHalf的局限性所在)。
Bottom Half常用到的函数包括:
初始化部分:void init_bh(int nr, void (*routine)(void)),
移出下半部:void remove_bh(int nr),
执行下半部:void mark_bh(int nr),
下半部的执行位置:下半部是一个被全局锁定的asklets(globally locked Tasklets ),所以当问下半部的执行位置也就是问Tasklets的执行位置,通常在两种环境下执行a) schedule() and b) on each interrupt/syscall return path.
内核定义的常用的下半部:
IMMEDIATE_BH:相当于任务队列中的立即队列(tq_immediate)
TQUEUE_BH:相当于任务队列中的时钟队列(tq_timer),每个timer interrupt的时候都会被执行一次;
其它的下半部一般不在模块中使用,而是用于内核中一些特殊的驱动,并且内核中下半部的得使用在不断减少,越来越多的被Tasklets替代。
下半部的使用示例(以《Linux device driver(2nd edition)》的short程序为例):
void short_bh_interrupt(int irq, void *dev_id, struct pt_regs *regs)
{
/* cast to stop 'volatile' warning */
do_gettimeofday((struct timeval *) tv_head);
short_incr_tv(&tv_head);
/* Queue the bh. Don't care for multiple enqueueing */
queue_task(&short_task, &tq_immediate);
mark_bh(IMMEDIATE_BH);
short_bh_count++; /* record that an interrupt arrived */
}
void short_do_tasklet (unsigned long unused)
{
int savecount = short_bh_count, written;
short_bh_count = 0; /* we have already been removed from the queue */
/*
* The bottom half reads the tv array, filled by the top half,
* and prints it to the circular text buffer, which is then consumed
* by reading processes
*/
/* First write the number of interrupts that occurred before this bh */
written = sprintf((char *)short_head,"bh after %6i\n",savecount);
short_incr_bp(&short_head, written);
/*
* Then, write the time values. Write exactly 16 bytes at a time,
* so it aligns with PAGE_SIZE
*/
do {
written = sprintf((char *)short_head,"%08u.%06u\n",
(int)(tv_tail->tv_sec % 100000000),
(int)(tv_tail->tv_usec));
short_incr_bp(&short_head, written);
short_incr_tv(&tv_tail);
} while (tv_tail != tv_head);
wake_up_interruptible(&short_queue); /* awake any reading process */
}
4.5 任务队列(Task Queues):
由于下半部是静态连接入内核的,所以在模块中不能使用,但是我们可以使用相应的任务队列( task queues)来代替。
任务队列的组织方式和等待队列的组织方式想同,一般的步骤也是
1,首先申明一个任务队列元素:
static struct tq_struct short_task;
short_task.routine = (void (*)(void *)) short_do_tasklet;
short_task.data = NULL; /* unused */
2,然后将这个队列元素插入某个队列中:
queue_task()
例如:queue_task(&short_task, &tq_immediate);将任务队列元素short_task插入立即队列tq_immediate中。
3,最后:mark_bh(IMMEDIATE_BH);
这样内核会在合适的位置调度short_do_tasklet();
或者也可以按照如下的方法是现在的任务队列:(form linux/tqueue.h)
/*
* To implement your own list of active bottom halfs, use the following
* two definitions:
*
* DECLARE_TASK_QUEUE(my_tqueue);
* struct tq_struct my_task = {
* routine: (void (*)(void *)) my_routine,
* data: &my_data
* };
*
* To activate a bottom half on a list, use:
*
* queue_task(&my_task, &my_tqueue);
*
* To later run the queued tasks use
*
* run_task_queue(&my_tqueue);
*
* This allows you to do deferred processing. For example, you could
* have a task queue called tq_timer, which is executed within the timer
* interrupt.
*/
在2.6内核中已经去掉了任务队列这种机制。
你可以使用这个链接引用该篇文章 http://publishblog.blogchina.com/blog/tb.b?diaryID=5864176