你了解过Linux内核的的tasklet机制和工作队列？

1. Tasklet机制分析

上面我们介绍了软中断机制，linux内核为什么还要引入tasklet机制呢？主要原因是软中断的pending标志位也就32位，一般情况是不随意增加软中断处理的。而且内核也没有提供通用的增加软中断的接口。其次内，软中断处理函数要求可重入，需要考虑到竞争条件比较多，要求比较高的编程技巧。所以内核提供了tasklet这样的一种通用的机制。

其实每次写总结的文章，总是想把细节的东西说明白，所以越写越多。这样做的好处是能真正理解其中的机制。但是，内容太多的一个坏处就是难道记忆，所以，在讲清楚讲详细的同时，我还要把精髓总结出来。Tasklet的特点，也是tasklet的精髓就是：tasklet不能休眠，同一个tasklet不能在两个CPU上同时运行，但是不同tasklet可能在不同CPU上同时运行，则需要注意共享数据的保护。

主要的数据结构

static DEFINE_PER_CPU(struct tasklet_head, tasklet_vec);

static DEFINE_PER_CPU(struct tasklet_head, tasklet_hi_vec);

struct tasklet_struct{ struct tasklet_struct *next; unsigned long state; atomic_t count; void (*func)(unsigned long); unsigned long data;};

struct tasklet_struct

{

struct tasklet_struct *next;

unsigned long state;

atomic_t count;

void (*func)(unsigned long);

unsigned long data;

};

如何使用tasklet

使用tasklet比较简单，只需要初始化一个tasklet_struct结构体，然后调用tasklet_schedule,就能利用tasklet机制执行初始化的func函数。

static inline void tasklet_schedule(struct tasklet_struct *t){ if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state)) __tasklet_schedule(t);}

static inline void tasklet_schedule(struct tasklet_struct *t)

{

if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state))

__tasklet_schedule(t);

}

tasklet_schedule处理过程也比较简单，就是把tasklet_struct结构体挂到tasklet_vec链表或者挂接到tasklet_hi_vec链表上，并调度软中断TASKLET_SOFTIRQ或者HI_SOFTIRQ

void __tasklet_schedule(struct tasklet_struct *t){ unsigned long flags;local_irq_save(flags); t->next = NULL; *__get_cpu_var(tasklet_vec).tail = t; __get_cpu_var(tasklet_vec).tail = &(t->next); raise_softirq_irqoff(TASKLET_SOFTIRQ); local_irq_restore(flags);}EXPORT_SYMBOL(__tasklet_schedule);void __tasklet_hi_schedule(struct tasklet_struct *t){ unsigned long flags; local_irq_save(flags); t->next = NULL; *__get_cpu_var(tasklet_hi_vec).tail = t; __get_cpu_var(tasklet_hi_vec).tail = &(t->next); raise_softirq_irqoff(HI_SOFTIRQ); local_irq_restore(flags);}EXPORT_SYMBOL(__tasklet_hi_schedule);

void __tasklet_schedule(struct tasklet_struct *t)

{

unsigned long flags;local_irq_save(flags);

t->next = NULL;

*__get_cpu_var(tasklet_vec).tail = t;

__get_cpu_var(tasklet_vec).tail = &(t->next);

raise_softirq_irqoff(TASKLET_SOFTIRQ);

local_irq_restore(flags);

}

EXPORT_SYMBOL(__tasklet_schedule);

void __tasklet_hi_schedule(struct tasklet_struct *t)

{

unsigned long flags;

local_irq_save(flags);

t->next = NULL;

*__get_cpu_var(tasklet_hi_vec).tail = t;

__get_cpu_var(tasklet_hi_vec).tail = &(t->next);

raise_softirq_irqoff(HI_SOFTIRQ);

local_irq_restore(flags);

}

EXPORT_SYMBOL(__tasklet_hi_schedule);

Tasklet执行过程

Tasklet_action在软中断TASKLET_SOFTIRQ被调度到后会被执行，它从tasklet_vec链表中把tasklet_struct结构体都取下来，然后逐个执行。如果t->count的值等于0，说明这个tasklet在调度之后，被disable掉了，所以会将tasklet结构体重新放回到tasklet_vec链表，并重新调度TASKLET_SOFTIRQ软中断，在之后enable这个tasklet之后重新再执行它。

static void tasklet_action(struct softirq_action *a){ struct tasklet_struct *list;local_irq_disable(); list = __get_cpu_var(tasklet_vec).head; __get_cpu_var(tasklet_vec).head = NULL; __get_cpu_var(tasklet_vec).tail = &__get_cpu_var(tasklet_vec).head; local_irq_enable(); while (list) { struct tasklet_struct *t = list; list = list->next; if (tasklet_trylock(t)) { if (!atomic_read(&t->count)) { if (!test_and_clear_bit(TASKLET_STATE_SCHED, &t->state)) BUG(); t->func(t->data); tasklet_unlock(t); continue; } tasklet_unlock(t); } local_irq_disable(); t->next = NULL; *__get_cpu_var(tasklet_vec).tail = t; __get_cpu_var(tasklet_vec).tail = &(t->next); __raise_softirq_irqoff(TASKLET_SOFTIRQ); local_irq_enable(); }}

static void tasklet_action(struct softirq_action *a)

{

struct tasklet_struct *list;local_irq_disable();

list = __get_cpu_var(tasklet_vec).head;

__get_cpu_var(tasklet_vec).head = NULL;

__get_cpu_var(tasklet_vec).tail = &__get_cpu_var(tasklet_vec).head;

local_irq_enable();

while (list)

{

struct tasklet_struct *t = list;

list = list->next;

if (tasklet_trylock(t))

{

if (!atomic_read(&t->count))

{

if (!test_and_clear_bit(TASKLET_STATE_SCHED, &t->state))

BUG();

t->func(t->data);

tasklet_unlock(t);

continue;

}

tasklet_unlock(t);

}

local_irq_disable();

t->next = NULL;

*__get_cpu_var(tasklet_vec).tail = t;

__get_cpu_var(tasklet_vec).tail = &(t->next);

__raise_softirq_irqoff(TASKLET_SOFTIRQ);

local_irq_enable();

}

2. Linux工作队列

前面已经介绍了tasklet机制，有了tasklet机制为什么还要增加工作队列机制呢？我的理解是由于tasklet机制的限制，变形tasklet中的回调函数有很多的限制，比如不能有休眠的操作等等。而是用工作队列机制，需要处理的函数在进程上下文中调用，休眠操作都是允许的。但是工作队列的实时性不如tasklet，采用工作队列的例程可能不能在短时间内被调用执行。

数据结构说明

首先需要说明的是workqueue_struct和cpu_workqueue_struct这两个数据结构，创建一个工作队列首先需要创建workqueue_struct，然后可以在每个CPU上创建一个cpu_workqueue_struct管理结构体。

struct cpu_workqueue_struct{ spinlock_t lock; struct list_head worklist; wait_queue_head_t more_work; struct work_struct *current_work; struct workqueue_struct *wq; struct task_struct *thread; int run_depth; /* Detect run_workqueue() recursion depth */} ____cacheline_aligned;/* * The externally visible workqueue abstraction is an array of * per-CPU workqueues: */struct workqueue_struct{ struct cpu_workqueue_struct *cpu_wq; struct list_head list; const char *name; int singlethread; int freezeable; /* Freeze threads during suspend */ int rt;#ifdef CONFIG_LOCKDEP struct lockdep_map lockdep_map;#endif};

struct cpu_workqueue_struct

{

spinlock_t lock;

struct list_head worklist;

wait_queue_head_t more_work;

struct work_struct *current_work;

struct workqueue_struct *wq;

struct task_struct *thread;

int run_depth; /* Detect run_workqueue() recursion depth */

} ____cacheline_aligned;

* The externally visible workqueue abstraction is an array of

* per-CPU workqueues:

struct workqueue_struct

{

struct cpu_workqueue_struct *cpu_wq;

struct list_head list;

const char *name;

int singlethread;

int freezeable; /* Freeze threads during suspend */

int rt;

#ifdef CONFIG_LOCKDEP

struct lockdep_map lockdep_map;

#endif

};

Work_struct表示将要提交的处理的工作。

struct work_struct{ atomic_long_t data;#define WORK_STRUCT_PENDING 0 /* T if work item pending execution */#define WORK_STRUCT_FLAG_MASK (3UL)#define WORK_STRUCT_WQ_DATA_MASK (~WORK_STRUCT_FLAG_MASK) struct list_head entry; work_func_t func;#ifdef CONFIG_LOCKDEP struct lockdep_map lockdep_map;#endif};

struct work_struct

{

atomic_long_t data;

#define WORK_STRUCT_PENDING 0 /* T if work item pending execution */

#define WORK_STRUCT_FLAG_MASK (3UL)

#define WORK_STRUCT_WQ_DATA_MASK (~WORK_STRUCT_FLAG_MASK)

struct list_head entry;

work_func_t func;

#ifdef CONFIG_LOCKDEP

struct lockdep_map lockdep_map;

#endif

};

上面三个数据结构的关系如下图所示

介绍主要数据结构的目的并不是想要把工作队列具体的细节说明白，主要的目的是给大家一个总的架构的轮廓。具体的分析在下面展开。从上面的该模块主要数据结构的关系来看，主要需要分析如下几个问题：

1. Workqueque是怎样创建的，包括event/0内核进程的创建

2. Work_queue是如何提交到工作队列的

3. Event/0内核进程如何处理提交到队列上的工作

Workqueque的创建

首先申请了workqueue_struct结构体内存，cpu_workqueue_struct结构体的内存。然后在init_cpu_workqueue函数中对cpu_workqueue_struct结构体进行初始化。同时调用create_workqueue_thread函数创建处理工作队列的内核进程。

create_workqueue_thread中创建了如下的内核进程

p = kthread_create(worker_thread, cwq, fmt, wq->name, cpu);

最后调用start_workqueue_thread启动新创建的进程。

struct workqueue_struct *__create_workqueue_key(const char *name, int singlethread, int freezeable, int rt, struct lock_class_key *key, const char *lock_name){ struct workqueue_struct *wq; struct cpu_workqueue_struct *cwq; int err = 0, cpu;wq = kzalloc(sizeof(*wq), GFP_KERNEL); if (!wq) return NULL; wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct); if (!wq->cpu_wq) { kfree(wq); return NULL; } wq->name = name; lockdep_init_map(&wq->lockdep_map, lock_name, key, 0); wq->singlethread = singlethread; wq->freezeable = freezeable; wq->rt = rt; INIT_LIST_HEAD(&wq->list); if (singlethread) { cwq = init_cpu_workqueue(wq, singlethread_cpu); err = create_workqueue_thread(cwq, singlethread_cpu); start_workqueue_thread(cwq, -1); } else { cpu_maps_update_begin(); /* * We must place this wq on list even if the code below fails. * cpu_down(cpu) can remove cpu from cpu_populated_map before * destroy_workqueue() takes the lock, in that case we leak * cwq[cpu]->thread. */ spin_lock(&workqueue_lock); list_add(&wq->list, &workqueues); spin_unlock(&workqueue_lock); /* * We must initialize cwqs for each possible cpu even if we * are going to call destroy_workqueue() finally. Otherwise * cpu_up() can hit the uninitialized cwq once we drop the * lock. */ for_each_possible_cpu(cpu) { cwq = init_cpu_workqueue(wq, cpu); if (err || !cpu_online(cpu)) continue; err = create_workqueue_thread(cwq, cpu); start_workqueue_thread(cwq, cpu); } cpu_maps_update_done(); } if (err) { destroy_workqueue(wq); wq = NULL; } return wq;}EXPORT_SYMBOL_GPL(__create_workqueue_key);

struct workqueue_struct *__create_workqueue_key(const char *name,

int singlethread,

int freezeable,

int rt,

struct lock_class_key *key,

const char *lock_name)

{

struct workqueue_struct *wq;

struct cpu_workqueue_struct *cwq;

int err = 0, cpu;wq = kzalloc(sizeof(*wq), GFP_KERNEL);

if (!wq)

return NULL;

wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct);

if (!wq->cpu_wq)

{

kfree(wq);

return NULL;

}

wq->name = name;

lockdep_init_map(&wq->lockdep_map, lock_name, key, 0);

wq->singlethread = singlethread;

wq->freezeable = freezeable;

wq->rt = rt;

INIT_LIST_HEAD(&wq->list);

if (singlethread)

{

cwq = init_cpu_workqueue(wq, singlethread_cpu);

err = create_workqueue_thread(cwq, singlethread_cpu);

start_workqueue_thread(cwq, -1);

}

else

{

cpu_maps_update_begin();

* We must place this wq on list even if the code below fails.

* cpu_down(cpu) can remove cpu from cpu_populated_map before

* destroy_workqueue() takes the lock, in that case we leak

* cwq[cpu]->thread.

spin_lock(&workqueue_lock);

list_add(&wq->list, &workqueues);

spin_unlock(&workqueue_lock);

* We must initialize cwqs for each possible cpu even if we

* are going to call destroy_workqueue() finally. Otherwise

* cpu_up() can hit the uninitialized cwq once we drop the

* lock.

for_each_possible_cpu(cpu)

{

cwq = init_cpu_workqueue(wq, cpu);

if (err || !cpu_online(cpu))

continue;

err = create_workqueue_thread(cwq, cpu);

start_workqueue_thread(cwq, cpu);

}

cpu_maps_update_done();

}

if (err)

{

destroy_workqueue(wq);

wq = NULL;

}

return wq;

}

EXPORT_SYMBOL_GPL(__create_workqueue_key);

向工作队列中添加工作

Shedule_work 函数向工作队列中添加任务。这个接口比较简单，无非是一些队列操作，不再叙述。

/** * schedule_work - put work task in global workqueue * @work: job to be done * * This puts a job in the kernel-global workqueue. */int schedule_work(struct work_struct *work){ return queue_work(keventd_wq, work);}EXPORT_SYMBOL(schedule_work);

/**

* schedule_work - put work task in global workqueue

* @work: job to be done

* This puts a job in the kernel-global workqueue.

int schedule_work(struct work_struct *work)

{

return queue_work(keventd_wq, work);

}

EXPORT_SYMBOL(schedule_work);

工作队列内核进程的处理过程

在创建工作队列的时候，我们创建了一个或者多个进程来处理挂到队列上的工作。这个内核进程的主要函数体为worker_thread，这个函数比较有意思的地方就是，自己降低的优先级，说明worker_thread调度的优先级比较低。在系统负载大大时候，采用工作队列执行的操作可能存在较大的延迟。

就函数的执行流程来说是真心的简单，只是从队列中取出work，从队列中删除掉，清除掉pending标记，并执行work设置的回调函数。

static int worker_thread(void *__cwq){ struct cpu_workqueue_struct *cwq = __cwq; DEFINE_WAIT(wait);if (cwq->wq->freezeable) set_freezable(); set_user_nice(current, -5); for (;;) { prepare_to_wait(&cwq->more_work, &wait, TASK_INTERRUPTIBLE); if (!freezing(current) && !kthread_should_stop() && list_empty(&cwq->worklist)) schedule(); finish_wait(&cwq->more_work, &wait); try_to_freeze(); if (kthread_should_stop()) break; run_workqueue(cwq); } return 0;}static void run_workqueue(struct cpu_workqueue_struct *cwq){ spin_lock_irq(&cwq->lock); cwq->run_depth++; if (cwq->run_depth > 3) { /* morton gets to eat his hat */ printk("%s: recursion depth exceeded: %dn", __func__, cwq->run_depth); dump_stack(); } while (!list_empty(&cwq->worklist)) { struct work_struct *work = list_entry(cwq->worklist.next, struct work_struct, entry); work_func_t f = work->func;#ifdef CONFIG_LOCKDEP /* * It is permissible to free the struct work_struct * from inside the function that is called from it, * this we need to take into account for lockdep too. * To avoid bogus "held lock freed" warnings as well * as problems when looking into work->lockdep_map, * make a copy and use that here. */ struct lockdep_map lockdep_map = work->lockdep_map;#endifcwq->current_work = work; list_del_init(cwq->worklist.next); spin_unlock_irq(&cwq->lock); BUG_ON(get_wq_data(work) != cwq); work_clear_pending(work); lock_map_acquire(&cwq->wq->lockdep_map); lock_map_acquire(&lockdep_map); f(work); lock_map_release(&lockdep_map); lock_map_release(&cwq->wq->lockdep_map); if (unlikely(in_atomic() || lockdep_depth(current) > 0)) { printk(KERN_ERR "BUG: workqueue leaked lock or atomic: " "%s/0x%08x/%dn", current->comm, preempt_count(), task_pid_nr(current)); printk(KERN_ERR " last function: "); print_symbol("%sn", (unsigned long)f); debug_show_held_locks(current); dump_stack(); } spin_lock_irq(&cwq->lock); cwq->current_work = NULL; } cwq->run_depth--; spin_unlock_irq(&cwq->lock);}

static int worker_thread(void *__cwq)

{

struct cpu_workqueue_struct *cwq = __cwq;

DEFINE_WAIT(wait);if (cwq->wq->freezeable)

set_freezable();

set_user_nice(current, -5);

for (;;)

{

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

if (!freezing(current) &&

!kthread_should_stop() &&

list_empty(&cwq->worklist))

schedule();

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

try_to_freeze();

if (kthread_should_stop())

break;

run_workqueue(cwq);

}

return 0;

}

static void run_workqueue(struct cpu_workqueue_struct *cwq)

{

spin_lock_irq(&cwq->lock);

cwq->run_depth++;

if (cwq->run_depth > 3)

{

/* morton gets to eat his hat */

printk("%s: recursion depth exceeded: %dn",

__func__, cwq->run_depth);

dump_stack();

}

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

{

struct work_struct *work = list_entry(cwq->worklist.next,

struct work_struct, entry);

work_func_t f = work->func;

#ifdef CONFIG_LOCKDEP

* It is permissible to free the struct work_struct

* from inside the function that is called from it,

* this we need to take into account for lockdep too.

* To avoid bogus "held lock freed" warnings as well

* as problems when looking into work->lockdep_map,

* make a copy and use that here.

struct lockdep_map lockdep_map = work->lockdep_map;

#endifcwq->current_work = work;

list_del_init(cwq->worklist.next);

spin_unlock_irq(&cwq->lock);

BUG_ON(get_wq_data(work) != cwq);

work_clear_pending(work);

lock_map_acquire(&cwq->wq->lockdep_map);

lock_map_acquire(&lockdep_map);

f(work);

lock_map_release(&lockdep_map);

lock_map_release(&cwq->wq->lockdep_map);

if (unlikely(in_atomic() || lockdep_depth(current) > 0))

{

printk(KERN_ERR "BUG: workqueue leaked lock or atomic: "

"%s/0x%08x/%dn",

current->comm, preempt_count(),

task_pid_nr(current));

printk(KERN_ERR " last function: ");

print_symbol("%sn", (unsigned long)f);

debug_show_held_locks(current);

dump_stack();

}

spin_lock_irq(&cwq->lock);

cwq->current_work = NULL;

}

cwq->run_depth--;

spin_unlock_irq(&cwq->lock);

}

阅读全文

内核(39814) 内核(39814)
Linux(206513) Linux(206513)
队列(10839) 队列(10839)
机制(9696) 机制(9696)



请按住滑块，拖动到最右边



了解新功能

linux内核信号是如何处理的？看完全懂了……

本文简单介绍下Linux信号处理机制，为介绍二进制翻译下信号处理机制做一个铺垫。本文主要参考书目《Linux内核源代码情景分析》《独辟蹊径品内核:Linux内核源代码导读》首先，先说一下

2017-11-16 05:11:00

13549

详解linux内核中的mutex同步机制

在linux内核中，互斥量（mutex，即mutual exclusion）是一种保证串行化的睡眠锁机制。和spinlock的语义类似，都是允许一个执行线索进入临界区，不同的是当无法获得锁的时候

2022-05-13 08:56:26

6271

Linux驱动开发-内核共享工作队列

在工作队列里，我们把推后执行的任务叫做工作（work），描述它的数据结构为work_struct，这些工作以队列结构组织成工作队列（workqueue），其数据结构为workqueue_struct，而工作线程就是负责执行工作队列中的工作。系统有默认的工作者线程,自己也可以创建自己的工作者线程。

2022-09-17 15:03:50

1025

Linux内核同步机制mutex详解

在linux内核中，互斥量mutex是一种保证CPU串行运行的睡眠锁机制。和spinlock类似，都是同一个时刻只有一个线程进入临界资源，不同的是，当无法获取锁的时候，spinlock原地自旋，而mutex则是选择挂起当前线程，进入阻塞状态。所以，mutex无法在中断上下文中使用。

2023-06-26 16:05:58

498

Linux内核的作用

Linux操作系统是当今世界上最为广泛使用的开源操作系统之一，内核则是一个操作系统的核心和灵魂所在。对于一名Linux驱动开发者来说，了解Linux内核的运行机制和Linux内核提供的一些关键功能（如虚拟内存管理、进程管理、文件系统、网络协议栈等）都是我们日常工作和学习的重点。

2023-07-06 11:46:41

1046

Linux内核地址映射模型与Linux内核高端内存详解

Linux 操作系统和驱动程序运行在内核空间，应用程序运行在用户空间，两者不能简单地使用指针传递数据，因为Linux使用的虚拟内存机制，用户空间的数据可能被换出，当内核空间使用用户空间指针时，对应

2018-05-08 10:33:19

3299

正在加载...

林超文PCB设计：PADS教程，PADS视频教程	郑振宇老师：Altium Designer教程，Altium Designer视频教程
张飞实战电子视频教程	朱有鹏老师：海思HI3518e教程，HI3518e视频教程
李增老师：信号完整性教程，高速电路仿真教程	华为鸿蒙系统教程，HarmonyOS视频教程
赛盛：EMC设计教程，EMC视频教程	杜洋老师：STM32教程，STM32视频教程
唐佐林：c语言基础教程，c语言基础视频教程	张飞：BUCK电源教程，BUCK电源视频教程
正点原子：FPGA教程，FPGA视频教程	韦东山老师：嵌入式教程，嵌入式视频教程
张先凤老师：C语言基础视频教程	许孝刚老师：Modbus通讯视频教程
王振涛老师：NB-IoT开发视频教程	Mill老师：FPGA教程，Zynq视频教程
C语言视频教程	RK3566芯片资料合集
朱有鹏老师：U-Boot源码分析视频教程	开源硬件专题

搜索历史

你了解过Linux内核的的tasklet机制和工作队列？

评论