品味内核中断上下半部

在《初尝内核中断》里我们了解了Linux内核中断模块的实现，也体验了一番共享中断的魅力，但上面这样的中断程序把任务都交到了服务子程序里去处理，这在单片机等性能要求不高的嵌入式系统里还好，但在我们的Linux等多任务嵌入式系统里，那这可是大大的浪费效率的做法，故而在Linux的内核中断里有上下半部的处理模式，将响应中断后最紧要的内容处理完后，就把不慌不忙的内容交到下半部去处理，即上半部处理最急需处理的内容，这样不需要非得该中断处理完再去响应同一中断线上的另一个中断请求了，效率也提上来了。
总的来说，上半部（top half）主要是处理与硬件交互等急切的事件，然后就把交互来的数据或请求等需求处理交到下半部（bottom half）去处理，下半部就处理这些比较费时的内容，并且还有可能被打断，下半部的执行由Linux Kernel去安排其时机。
Linux下半部机制由softirq、tasklet和工作队列等来实现，由于softirq与tasklet很相似，在中断里也较少使用，我们这里就学习下tasklet和工作队列（workqueue，在嵌入式Linux较常使用）机制。
1.tasklet机制
tasklet（小任务）也是在软中断（softirq，不是软件中断）的基础上实现的，但两个相同的tasklet是不会同时执行的，就算在不同的处理器上也不行。多次被调试时，tasklet也只运行一次。
在Linux Kernel源码的include/linux/interrupt.h文件里，关于tasklet有如下的结构体：
struct tasklet_struct
{
        struct tasklet_struct *next;
        unsigned long state;
        atomic_t count;
        void (*func)(unsigned long);
        unsigned long data;
};
其中，next指向下一个tasklet，state表示当前tasklet的状态，有同一文件里有如下枚举：
enum
{
        TASKLET_STATE_SCHED, /* Tasklet is scheduled for execution */
        TASKLET_STATE_RUN /* Tasklet is running (SMP only) */
};
其后面的注释已说明其状态，count是个原子型量，一个引用计数，为0时该tasklet才被激活，否则禁止，不可执行。func函数指针指向该tasklet对应的处理函数，而这个处理函数的唯一参数是data。 在同一文件里有如下的宏来方便我们静态创建tasklet，以供使用：
#define DECLARE_TASKLET(name, func, data) \
        struct tasklet_struct name = { NULL, 0, ATOMIC_INIT(0), func, data }

#define DECLARE_TASKLET_DISABLED(name, func, data) \
        struct tasklet_struct name = { NULL, 0, ATOMIC_INIT(1), func, data }
两者区别就在名称和ATOMIC_INIT初始计数器的值，第一个宏创建的tasklet默认处于激活状态，而后一个宏则创建的tasklet默认处于禁止状态，这两种状态切换可由tasklet_disable和tasklet_enable两函数实现，只有处于激活状态的tasklet才可被调用运行。
在arch/x86/include/asm/atomic.h有如下宏：
#define ATOMIC_INIT(i) { (i) }
在include/linux/types.h有如下结构体：
typedef struct {
        int counter;
} atomic_t;
相信不需要怎么说就可以一目了解count量了，除了这静态创建，那么相对应的还有动态创建，在include/linux/interrupt.h文件有声明，但在kernel/softirq.c有如下定义：
void tasklet_init(struct tasklet_struct *t,
        void (*func)(unsigned long), unsigned long data)
{
        t->next = NULL;
        t->state = 0;
        atomic_set(&t->count, 0);
        t->func = func;
        t->data = data;
}
在arch/x86/include/asm/atomic.h文件中有如下内容：
static inline void atomic_set(atomic_t *v, int i)
{
        v->counter = i;
}
从上面这两种可知动态创建时需先声明一个tasklet_struct结构体变量再使用tasklet_init来初始化创建。而atomic_set设置t->count.counter=0。
通过静态或者动态创建tasklet，其最根本还是func函数指针，整个tasklet具体的处理就靠它了。创建好tasklet后，我们需要去调度让tasklet进入等待执行的状态，加入运行态（可运行，待调度其执行），相应的函数在include/linux/interrupt.h文件中：
static inline void tasklet_schedule(struct tasklet_struct *t) ;
好了，基本了解了tasklet，下面还是通过实践来感受下：
#include <linux/kernel.h>
#include <linux/module.h>
#include <linux/init.h>
#include <linux/interrupt.h>

static int irqn;
static char* devname;
static struct tasklet_struct slam_tl;

module_param(irqn,int,0644);
module_param(devname,charp,0644);

static void slam_tl_handler(unsigned long data)
{
        printk("Hello slam tasklet handler!\n");
}

static irqreturn_t tasklet_irq_handler(int irq,void * devid)
{
        static int count=0;

        printk("Enter tasklet_irq_handler!\n");
        printk("ISR devid:%d\n",(int)(*(int *)devid));
        count++;
        printk("count=%d\n",count);
        if(count > 65536)
                count=0;
        if(!(count%10)){
                printk("Every 10 twices enter tasklet bottom half!\n");
                tasklet_schedule(&slam_tl);
        }
        printk("Leaving IRQ handler!\n");
        return IRQ_HANDLED;
}

static int __init tasklet_irq_init(void)
{
        printk("Enter tasklet_irq_init...\n");
        if(request_irq(irqn,tasklet_irq_handler,IRQF_SHARED,devname,&irqn) != 0)
        {
                printk("Request IRQ failed!\ndevname=%s,IRQ:%d\n",devname,irqn);
                return -1;
        }
        tasklet_init(&slam_tl,slam_tl_handler,0);

        printk("Request IRQ success!\ndevname=%s,IRQ:%d\n",devname,irqn);
        printk("Exit tasklet_irq_init...\n");
        return 0;
}

static void __exit tasklet_irq_exit(void)
{
        printk("Enter tasklet_irq_exit...\n");
        tasklet_kill(&slam_tl);
        free_irq(irqn,&irqn);
        printk("Release irq sucess!\ndevname=%s,IRQ:%d\n",devname,irqn);
        printk("Exit tasklet_irq_exit...\n");
}

module_init(tasklet_irq_init);
module_exit(tasklet_irq_exit);

MODULE_LICENSE("GPL");
其对应Makefile内容如下：
obj-m += tasklet_irq.o

CUR_PATH:=$(shell pwd)
LINUX_KERNEL_PATH:=/home/xinu/linux-3.13.6

all:
        make -C $(LINUX_KERNEL_PATH) M=$(CUR_PATH) modules

clean:
        make -C $(LINUX_KERNEL_PATH) M=$(CUR_PATH) clean
相应的文件目录树如下：
/home/xinu/xinu/linux_kernel_driver_l1/tasklet_irq/
├── Makefile
└── tasklet_irq.c
执行命令可参考《初尝内核中断》，此处不提供参考的打印信息。
2.工作队列机制
在品味完上面基于软中断的tasklet后，我们再来品味嵌入式Linux里经常用到的work queue吧。 工作队列能实现tasklet无法做到的功能，比如睡眠，因为tasklet运行于中断上下文（interrupt context），而工作队列是运行于进程上下文（process context），接下来我们会说到在Linux Kernel里工作队列的两种使用：共享队列和自创队列。
无论哪种都需要先创建工作项：
在include/linux/workqueue.h文件里有如下内容：
#define DECLARE_WORK(n, f) \
        struct work_struct n = __WORK_INITIALIZER(n, f)
和
#define INIT_WORK(_work, _func) \
do { \
        __INIT_WORK((_work), (_func), 0); \
} while (0)
这两个宏分别以静态和动态方式创建work，两宏的参数均为工作work及对应的处理函数。现在有了work之后，那么work queue呢？在同一文件里有如下内容：
#define create_workqueue(name) \
        alloc_workqueue("%s", WQ_MEM_RECLAIM, 1, (name))
#define create_singlethread_workqueue(name) \
        alloc_workqueue("%s", WQ_UNBOUND | WQ_MEM_RECLAIM, 1, (name))
其中第一个宏创建的工作队列是每个CPU一个线程的方式，而第二个宏则是无论多个少CPU，只创建一个线程的方式（等会演示第二个）。
到此，共享与自创的工作队列已漫漫有些许体会了，共享的话肯定不用这两个创建workqueue的操作，是的，共享会直接提交到内核的工作者线程（system_wq队列）去处理，两者的区别还在于调度工作队列的工作时处理函数的不同，即工作与工作队列的关联方式：
共享队列用到的调度函数：
static inline bool schedule_work(struct work_struct *work) ;
自创队列用到的调度函数：
static inline bool queue_work(struct workqueue_struct *wq, struct work_struct *work);
等会演示会用到如上的函数，关于delay延时处理的不演示，后期会专题讲述，更详细的内容可先查看include/linux/workqueue.h和kernel/workqueue.c了解。
好了，基本内容了解了，接下来还是演示一例：
#include <linux/kernel.h>
#include <linux/module.h>
#include <linux/init.h>
#include <linux/interrupt.h>
#include <linux/sched.h>
#include <linux/slab.h>

static int irqn;
static char* devname;

module_param(irqn,int,0644);
MODULE_PARM_DESC(irqn,"The share irq number.");
module_param(devname,charp,0644);
MODULE_PARM_DESC(devname,"The share irq name.");

static void shr_work_routine(struct work_struct *ws)
{
        printk("shr_work_routine of system_wq!\n");
}

static void slam_work_routine(struct work_struct *ws)
{
        printk("slam_work_routine of slam_wq!\n");
}

static struct work_struct * shr_work;
static DECLARE_WORK(slam_work,slam_work_routine);
static struct workqueue_struct * slam_wq;

static irqreturn_t workqueue_irq_handler(int irq,void * devid)
{
        printk("Enter workqueue_irq_handler!\n");
        printk("ISR devid:%d\n",(int)(*(int *)devid));

        schedule_work(shr_work);
        queue_work(slam_wq,&slam_work);

        printk("Leaving IRQ handler!\n");
        return IRQ_HANDLED;
}

static int __init workqueue_irq_init(void)
{
        printk("Enter workqueue_irq_init...\n");
        if(request_irq(irqn,workqueue_irq_handler,IRQF_SHARED,devname,&irqn) != 0)
        {
                printk("Request IRQ failed!\ndevname=%s,IRQ:%d\n",devname,irqn);
                return -1;
        }

        shr_work = kzalloc(sizeof(typeof(*shr_work)),GFP_KERNEL);
        INIT_WORK(shr_work,shr_work_routine);

        slam_wq = create_workqueue("slam_wq");

        printk("Exit workqueue_irq_init...\n");
        return 0;
}

static void __exit workqueue_irq_exit(void)
{
        printk("Enter workqueue_irq_exit...\n");
        flush_workqueue(slam_wq);
        free_irq(irqn,&irqn);
        destroy_workqueue(slam_wq);
        printk("Exit workqueue_irq_exit...\n");
}

module_init(workqueue_irq_init);
module_exit(workqueue_irq_exit);

MODULE_LICENSE("GPL");
相应的Makefile只修改了生成的目标名称，其他未更改，故不给出其内容，最后相应的文件目录树如下：
/home/xinu/xinu/linux_kernel_driver_l1/workqueue_irq/
 ├── Makefile
 └── workqueue_irq.c
加载KO后，可使用ps命令，可查看到创建的slam_wq工作队列线程。 至此，我们了解了中断的上下半部的机制，品味了最常用到的tasklet和work queue机制，本次仅是浅尝，后期将二度深尝。 
参考网址：
http://edsionte.com/techblog/archives/1539
http://edsionte.com/techblog/archives/1547
http://edsionte.com/techblog/archives/1582
http://edsionte.com/techblog/archives/1618
http://uliweb.clkg.org/tutorial/view_chapter/109
http://hacklu.com/blog/内核抢占和中断-75 http://home.ustc.edu.cn/~boj/courses/linux_kernel/2_int.html
http://blog.chinaunix.net/uid-27411029-id-4013570.html
http://wangcong.org/blog/archives/49 http://bgutech.blog.163.com/blog/static/18261124320116181119889/
http://oss.org.cn/kernel-book/ldd3/ch07s06.html
http://blog.csdn.net/lanmanck/article/details/4770030
http://nano-chicken.blogspot.com/2010/12/linux-modules73-work-queue.html https://gist.github.com/yagihiro/309746
http://vmlinz.is-programmer.com/posts/26436.html http://www.ibm.com/developerworks/linux/library/l-tasklets/ https://www.kernel.org/doc/htmldocs/kernel-hacking/index.html
地点: 中国广东省深圳市南山区科技园 邮政编码: 518057

评论

发表评论

此博客中的热门博文

I/O映射之I/O端口

图片
对于外设，操作系统会采用端口映射和内存映射两种方式来对其进行控制，其中端口映射就是操作系统规定好一段地址给指定换上设，其与外设的寄存器按顺序一一对应上。 在Linux内核源码include/asm-generic/io.h头文件里有如下我们对I/O port进行操作的函数： static inline u8 inb(unsigned long addr) { return readb(addr + PCI_IOBASE); } static inline u16 inw(unsigned long addr) { return readw(addr + PCI_IOBASE); } static inline u32 inl(unsigned long addr) { return readl(addr + PCI_IOBASE); } static inline void outb(u8 b, unsigned long addr) { writeb(b, addr + PCI_IOBASE); } static inline void outw(u16 b, unsigned long addr) { writew(b, addr + PCI_IOBASE); } static inline void outl(u32 b, unsigned long addr) { writel(b, addr + PCI_IOBASE); } #define inb_p(addr) inb(addr) #define inw_p(addr) inw(addr) #define inl_p(addr) inl(addr) #define outb_p(x, addr) outb((x), (addr)) #define outw_p(x, addr) outw((x), (addr)) #define outl_p(x, addr) outl((x), (addr)) 分别对应对端口进行读取或写入字节、字、双字大小数据的操作。 该头文件中还有其他的操作函数及宏定义，请自行深入了解。 我们可通过读取/proc/ioports文件来了解Linux里I/O端口的映射
阅读全文

通过Netlink检测网线插拔

  最近有个需求需要检测RJ45网卡的网线有没有接上，而最近正在了解Netlink相关资料，刚好也看下通过Netlink可以进行检测，故在此做下粗略笔记：   1.首先要创建一个Netlink Socket，在用户层使用如下参数来调用socket()函数：    fd = socket(AF_NETLINK, SOCK_RAW, NETLINK_ROUTE);    上面这个函数第一个参数必须是AF_NETLINK或PF_NETLINK，这两个标志在Linux下是一样的，第二个参数可以是SOCK_RAW或SOCK_DGRAM（对应用到TCP或UDP协议），而最后一个参数NETLINK_ROUTE为“路由守护进程”，用于接收来自内核的路由通知事件。   2.将上面创建的Socket绑定    addr.nl_family = AF_NETLINK;    addr.nl_groups = RTNLGRP_LINK; //指定接收路由多播组消息    bind(fd, (struct sockaddr*)&addr, sizeof(addr));    上面将创建的socket与相应的协议族和组进行绑定，接下来通过读fd这个socket来获得相应的消息数据struct nlmsghdr，再对该结构体数据进行判断来获得网线是接上或是拔掉，相应的源码如下： #include <sys/types.h>  #include <sys/socket.h>  #include <asm/types.h>  #include <linux/netlink.h>  #include <linux/rtnetlink.h>  #include <stdlib.h>  #include <stdio.h>  #include <sys/ioctl.h>  #include <linux/if.h>  #include <string.h>  #define BUFLEN 20480  int main(int argc, char *argv[])  {      int fd, retval;      char buf[BUFLEN] = {0};
阅读全文

使用seq_file

在《使用procfs》一文的源码示例中有说到proc文件系统每次读取的数据只能是1个页，如果超过则需多次读取，这样的话会增加读取次数，增多系统调用次数，影响了整体的效率，故而才有seq file序列文件的出现，该项功能使得内核对于大文件的读取更加容易。 对于seq file，其结构体定义在include/linux/seq_file.h文件中，内容如下： struct seq_file { char *buf; size_t size; size_t from; size_t count; size_t pad_until; loff_t index; loff_t read_pos; u64 version; struct mutex lock; const struct seq_operations *op; int poll_event; #ifdef CONFIG_USER_NS struct user_namespace *user_ns; #endif void *private; }; 这里不详细说明该结构体，仅供感受下。 对于seq_file，相应的文件操作需要实现seq_operations结构体的成员，该结构体在同一头文件中有如下定义： struct seq_operations { void * (*start) (struct seq_file *m, loff_t *pos); void (*stop) (struct seq_file *m, void *v); void * (*next) (struct seq_file *m, void *v, loff_t *pos); int (*show) (struct seq_file *m, void *v); }; 实现该结构体成员后，需要使用seq_open打开，让文件与该seq_operations关联起来，最终这个open函数要作为file_oper
阅读全文