Linux内核工作队列(workqueue)详解

news2025/2/25 22:01:52

1、为什么需要工作队列？

在内核代码中，经常会遇到不能或不合适去马上调用某个处理过程，此时希望将该工作推送给某个内核线程执行，这样做的原因有很多，比如：

中断触发了某个过程的执行条件，而该过程执行时间较长或者会调用导致睡眠的函数，则该过程不应该在中断上下文中立即被调用。
类似于中断，一些紧急性的任务不希望执行比较耗时的非关键过程，则需要把该过程提交到低优先级线程执行。比如一个轮询的通信接收线程，它需要快速完成检测和接收数据，而对数据的解析则应该交由低优先级线程慢慢处理。
有时希望将一些工作集中起来以获取批处理的性能；或则合并缩减一些执行线程，减少资源消耗。
基于以上需求，人们开发出了工作队列这一机制。工作队列不光在操作系统内核中会用到，一些应用程序或协议栈也会实现自己的工作队列。

2、工作队列的概念

工作队列 ( workqueue )是将操作（或回调）延期异步执行的一种机制。工作队列可以把工作推后，交由一个内核线程去执行，并且工作队列是执行在线程上下文中，因此工作执行过程中可以被重新调度、抢占、睡眠。

工作项（work item）是工作队列中的元素，是一个回调函数和多个回调函数输入参数的集合，有时也会有额外的属性成员，总之通过一个结构体即可记录和描述一个工作项。

3、工作队列的特性

通过工作队列来执行一个工作相比于直接执行，会有一下特性：

异步，工作不是在本中断或线程中执行，而是交由其他线程执行。
延期，交由低优先级线程执行，执行前可能被抢占，也可能有其他工作排在前面执行，所以从提交工作队列到工作真正被执行之间会延时不定长时间。
排队，FIFO 模式先到先执行。也肯会有多个优先级的工作队列，低优先级的工作队列要等到高优先级的工作队列全部执行完成后才能执行。但同一个工作队列内部的各项都是按时间先后顺序执行的额，不会进行钱赞重排。
缓存，既然是队列它就能缓存多个项，需要异步执行丢进去就行，队列会逐个执行。虽然工作队列能缓存多个项，但也是有上限的当队列已满时，新的入队项就会被丢弃，丢弃的个数会被统计下来。

4、实现

Linux内核中的工作队列包括：共享工作队列和自定义工作队列。区别如下：

1）共享工作队列：将新创建的工作任务添加到Linux内核创建的全局工作队列system_wq中，无需自己创建工作队列；

2）自定义工作队列：将新创建的工作任务添加到自己创建的工作队列中；

1、共享工作队列举例（demol来源于网络）

#include <linux/kernel.h>
#include <linux/module.h>
#include <linux/slab.h>
#include <linux/kobject.h>
#include <linux/list.h>
#include <linux/kthread.h>
#include <asm/ptrace.h>
#include <linux/sched.h>
#include <linux/delay.h>
#include <linux/interrupt.h>

int data  = 10;
 
static struct work_struct       work1;
static struct work_struct       work2;
 
 
 
static void do_work1(struct work_struct *arg)
{
    printk(KERN_INFO "do_work1 .....");
}
 
static void do_work2(struct work_struct *arg)
{
     printk(KERN_INFO "do_work2 .....");
}
 
int threadfn(void *data)
{
     static int count =  0 ;
     int args = *(int *)data;
     printk(KERN_INFO "enter thead_fn");
     while(1)
     {
         msleep(2*1000);
         printk(KERN_INFO "threadfn data: %d, count: %d",args , ++count);
         schedule_work(&work1);
         schedule_work(&work2);
         
     }
} 
 
 
static int __init test_kobj_init(void)
{
 
    INIT_WORK(&work1,do_work1);
    INIT_WORK(&work2,do_work2);
 
    struct task_struct *  thread =  kthread_create( threadfn,(void * )&data,"mythread");
    if(thread != NULL)
    {
        printk(KERN_INFO "thread create success");
        wake_up_process(thread);
        
    }else
    {
        printk(KERN_ERR "thread create err");
    }
 
     return 0;
}
 
 
 
static void __exit test_kobj_exit(void)
{
     printk(KERN_INFO "test_kobj_exit ");
     return;
}
 
module_init(test_kobj_init);
module_exit(test_kobj_exit);
 
MODULE_AUTHOR("weilaikeji");
MODULE_LICENSE("GPL");

输出结果

2、自定义工作队列举例（demol来源于网络）

#include <linux/kernel.h>
#include <linux/module.h>
#include <linux/slab.h>
#include <linux/kobject.h>
#include <linux/list.h>
#include <linux/kthread.h>
#include <asm/ptrace.h>
#include <linux/sched.h>
#include <linux/delay.h>
#include <linux/interrupt.h>
int data  = 10;

static struct workqueue_struct *workqueue;
static struct work_struct       work1;
static struct work_struct       work2;



static void do_work1(struct work_struct *arg)
{
    printk(KERN_INFO "do_work1 .....");
}

static void do_work2(struct work_struct *arg)
{
    printk(KERN_INFO "do_work2 .....");
}


int threadfn(void *data)
{
    static int count =  0 ;
    int args = *(int *)data;
    printk(KERN_INFO "enter thead_fn");
    while(1)
    {
        msleep(2*1000);
        printk(KERN_INFO "threadfn data: %d, count: %d",args , ++count);
        queue_work(workqueue,&work1);//将任务放到自己创建的工作队列上去执行
        queue_work(workqueue,&work2);
        
    }
} 


static int __init test_kobj_init(void)
{

    
   workqueue = create_workqueue("wanghb_queue");
   INIT_WORK(&work1,do_work1);
   INIT_WORK(&work2,do_work2);

   struct task_struct *  thread =  kthread_create( threadfn,(void * )&data,"mythread");
   if(thread != NULL)
   {
       printk(KERN_INFO "thread create success");
       wake_up_process(thread);
       
   }else
   {
       printk(KERN_ERR "thread create err");
   }

    return 0;
}



static void __exit test_kobj_exit(void)
{
    printk(KERN_INFO "test_kobj_exit ");
    destroy_workqueue(workqueue);
    return;
}

module_init(test_kobj_init);
module_exit(test_kobj_exit);

MODULE_AUTHOR("weilaikeji");
MODULE_LICENSE("GPL");

输出结果

3、执行函数的入参如何传递的？

看到这里，会有个疑问，如何把用户的数据作为参数传递给执行函数呢？

2.6.20版本之后使用工作队列需要把work_struct定义在用户的数据结构中，然后通过container_of来得到用户数据。具体用法如下：

struct my_struct_t { 
    char *name; 
    struct work_struct my_work; 
};

void my_func(struct work_struct *work) 
{ 
    struct my_struct_t *my_name = container_of(work, struct my_struct_t, my_work);         
    printk(KERN_INFO “Hello world, my name is %s!\n”, my_name->name);
} 

struct workqueue_struct *my_wq = create_workqueue(“my wq”); 
struct my_struct_t my_name; my_name.name = “Jack”; 
INIT_WORK(&(my_name.my_work), my_func); 
queue_work(my_wq, &my_work);
destroy_workqueue(my_wq);

在使用如下函数时注意事项

1、flush_work()：堵塞工作任务，直到工作任务完成

2、flush_delayed_work()：等待延时工作任务完成

3、cancel_work_sync()：取消工作任务并等待它完成

4、cancel_delayed_work()：取消延时工作任务

5、cancel_delayed_work_sync()：取消延时工作任务并等待它完成

6、create_workqueue()：对于多CPU系统，内核会在每个CPU上创建一个工作队列，使线程处理并行化

7、create_singlethread_workqueue()：内核只在一个CPU上创建一个工作队列

8、queue_work_on()：在指定CPU上添加工作任务，queue_work()调用queue_work_on()在所有CPU上添加工作任务