最近再看《深入理解LINUX内核》，简单的笔记下workqueue的实现吧。
workqueue.h
/*
* workqueue.h --- work queue handling for Linux.
*/
#ifndef _LINUX_WORKQUEUE_H
#define _LINUX_WORKQUEUE_H
#include <linux/timer.h>
#include <linux/linkage.h>
#include <linux/bitops.h>
struct workqueue_struct;
/*
插入到工作队列中的任务结构体
*/
struct work_struct {
unsigned long pending; /* 该任务是否已经要执行了 */
struct list_head entry; /* 任务在队列中是链表存储 */
void (*func)(void *); /* 回调函数 */
void *data; /* 传入参数 */
void *wq_data; /* 保存该任务所属的工作队列 */
struct timer_list timer; /* 任务可能要延时执行 */
};
/* 初始化一个任务结构体*/
#define __WORK_INITIALIZER(n, f, d) { \
.entry = { &(n).entry, &(n).entry }, \
.func = (f), \
.data = (d), \
.timer = TIMER_INITIALIZER(NULL, 0, 0), \
}
/* 初始化一个任务结构体*/
#define DECLARE_WORK(n, f, d) \
struct work_struct n = __WORK_INITIALIZER(n, f, d)
/*
* initialize a work-struct's func and data pointers:
*/
/* 初始化一个任务结构体指针 */
#define PREPARE_WORK(_work, _func, _data) \
do { \
(_work)->func = _func; \
(_work)->data = _data; \
} while (0)
/*
* initialize all of a work-struct:
*/
/* 初始化一个任务结构体指针 */
#define INIT_WORK(_work, _func, _data) \
do { \
INIT_LIST_HEAD(&(_work)->entry); \
(_work)->pending = 0; \
PREPARE_WORK((_work), (_func), (_data)); \
init_timer(&(_work)->timer); \
} while (0)
extern struct workqueue_struct *__create_workqueue(const char *name,
int singlethread);
#define create_workqueue(name) __create_workqueue((name), 0)
#define create_singlethread_workqueue(name) __create_workqueue((name), 1)
extern void destroy_workqueue(struct workqueue_struct *wq);
extern int FASTCALL(queue_work(struct workqueue_struct *wq, struct work_struct *work));
extern int FASTCALL(queue_delayed_work(struct workqueue_struct *wq, struct work_struct *work, unsigned long delay));
extern void FASTCALL(flush_workqueue(struct workqueue_struct *wq));
extern int FASTCALL(schedule_work(struct work_struct *work));
extern int FASTCALL(schedule_delayed_work(struct work_struct *work, unsigned long delay));
extern int schedule_delayed_work_on(int cpu, struct work_struct *work, unsigned long delay);
extern void flush_scheduled_work(void);
extern int current_is_keventd(void);
extern int keventd_up(void);
extern void init_workqueues(void);
/*
* Kill off a pending schedule_delayed_work(). Note that the work callback
* function may still be running on return from cancel_delayed_work(). Run
* flush_scheduled_work() to wait on it.
*/
/*
取消一个任务。
*/
static inline int cancel_delayed_work(struct work_struct *work)
{
int ret;
ret = del_timer_sync(&work->timer);
if (ret)
clear_bit(0, &work->pending);
return ret;
}
#endif
workqueue.c
/*
* linux/kernel/workqueue.c
*
* Generic mechanism for defining kernel helper threads for running
* arbitrary tasks in process context.
*
* Started by Ingo Molnar, Copyright (C) 2002
*
* Derived from the taskqueue/keventd code by:
*
* David Woodhouse <dwmw2@infradead.org>
* Andrew Morton <andrewm@uow.edu.au>
* Kai Petzke <wpp@marie.physik.tu-berlin.de>
* Theodore Ts'o <tytso@mit.edu>
*/
#include <linux/module.h>
#include <linux/kernel.h>
#include <linux/sched.h>
#include <linux/init.h>
#include <linux/signal.h>
#include <linux/completion.h>
#include <linux/workqueue.h>
#include <linux/slab.h>
#include <linux/cpu.h>
#include <linux/notifier.h>
#include <linux/kthread.h>
/*
* The per-CPU workqueue (if single thread, we always use cpu 0's).
*
* The sequence counters are for flush_scheduled_work(). It wants to wait
* until until all currently-scheduled works are completed, but it doesn't
* want to be livelocked by new, incoming ones. So it waits until
* remove_sequence is >= the insert_sequence which pertained when
* flush_scheduled_work() was called.
*/
/*
多CPU环境下，每个CPU上都维护一个工作队列
*/
struct cpu_workqueue_struct {
spinlock_t lock; /* 锁 */
long remove_sequence; /* Least-recently added (next to run) */
long insert_sequence; /* Next to add */
struct list_head worklist;/* 工作队列上挂着的任务链表 */
wait_queue_head_t more_work;
wait_queue_head_t work_done;
struct workqueue_struct *wq; /* 该CPU工作队列所属的总工作队列 */
task_t *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[NR_CPUS];/* 在每个CPU上都有对应的工作队列 */
const char *name;/* 该结构体的名称 */
struct list_head list; /* Empty if single thread *//* 内核中把所有申请的工作队列全部串起来，便于管理,
如果是单CPU队列，就置为NULL */
};
/* All the per-cpu workqueues on the system, for hotplug cpu to add/remove
threads to each one as cpus come/go. */
/*
内核中把所有申请的工作队列全部串起来，便于管理。
这里定义了一个锁和一个链表头。
*/
/*
注意这里串起来的队列全部是多CPU队列，原因为当CPU发生变更时可以
进行遍历更改。由于单CPU不会出现这个问题，因此单CPU队列不会放在这个
队列中
*/
static DEFINE_SPINLOCK(workqueue_lock);
static LIST_HEAD(workqueues);
/*
判断一个工作队列是否是单CPU型的。
*/
/* If it's single threaded, it isn't in the list of workqueues. */
static inline int is_single_threaded(struct workqueue_struct *wq)
{
return list_empty(&wq->list);
}
/*
将一个工作任务插入到一个工作队列中
*/
/* Preempt must be disabled. */
static void __queue_work(struct cpu_workqueue_struct *cwq,
struct work_struct *work)
{
unsigned long flags;
spin_lock_irqsave(&cwq->lock, flags);/* 先禁掉中断，并且锁上该工作队列*/
work->wq_data = cwq;/* 更新工作任务对应的工作队列 */
list_add_tail(&work->entry, &cwq->worklist);/* 讲工作任务添加到工作队列内部的链表上 */
cwq->insert_sequence++;/* 累加插入计数 */
wake_up(&cwq->more_work);/* 唤醒工作队列上的等待队列 */
spin_unlock_irqrestore(&cwq->lock, flags);
}
/*
* Queue work on a workqueue. Return non-zero if it was successfully
* added.
*
* We queue the work to the CPU it was submitted, but there is no
* guarantee that it will be processed by that CPU.
*/
/*
将一个工作任务插入到一个工作队列中
*/
int fastcall queue_work(struct workqueue_struct *wq, struct work_struct *work)
{
/*
获取当前CPU的锁定
*/
int ret = 0, cpu = get_cpu();
/* 如果pending为1，说明已经加到工作队列上了，
不能重复加入了
*/
if (!test_and_set_bit(0, &work->pending)) {
/* 如果当前的工作队列是单CPU类型的，就不用指定当前CPU 了*/
if (unlikely(is_single_threaded(wq)))
cpu = 0;
BUG_ON(!list_empty(&work->entry));
__queue_work(wq->cpu_wq + cpu, work);
ret = 1;
}
put_cpu();
return ret;
}
/*
内部工具函数，如果一个任务被延迟执行，那么它不会被立即
加入到工作队列中，而是会先放到一个定时器中，当定时器到期
时，会进入该函数，在该函数中，才会放到工作队列中。
*/
static void delayed_work_timer_fn(unsigned long __data)
{
/*从该任务中找出要加入的队列，加进去*/
struct work_struct *work = (struct work_struct *)__data;
struct workqueue_struct *wq = work->wq_data;
int cpu = smp_processor_id();
if (unlikely(is_single_threaded(wq)))
cpu = 0;
__queue_work(wq->cpu_wq + cpu, work);
}
/*
添加一个需要延迟的任务，其实是先加入一个定时
*/
int fastcall queue_delayed_work(struct workqueue_struct *wq,
struct work_struct *work, unsigned long delay)
{
int ret = 0;
struct timer_list *timer = &work->timer;
/*
同样，先看下该任务是否已经被加入了
*/
if (!test_and_set_bit(0, &work->pending)) {
BUG_ON(timer_pending(timer));
BUG_ON(!list_empty(&work->entry));
/* 先不加人工作队列，而是先定时 */
/* This stores wq for the moment, for the timer_fn */
work->wq_data = wq;
timer->expires = jiffies + delay;
timer->data = (unsigned long)work;
timer->function = delayed_work_timer_fn;
add_timer(timer);
ret = 1;
}
return ret;
}
/* 内部工具函数，工作队列的线程函数，用来不停的遍历任务并执行 */
static inline void run_workqueue(struct cpu_workqueue_struct *cwq)
{
unsigned long flags;
/*
* Keep taking off work from the queue until
* done.
*/
spin_lock_irqsave(&cwq->lock, flags);/* 禁掉中断，并且锁上该工作队列*/
cwq->run_depth++;/* 记下递归值，防止太深了 */
if (cwq->run_depth > 3) {
/* morton gets to eat his hat */
printk("%s: recursion depth exceeded: %d\n",
__FUNCTION__, cwq->run_depth);
dump_stack();
}
/* 如果工作队列上的任务链表不空，就一个一个执行 */
while (!list_empty(&cwq->worklist)) {
struct work_struct *work = list_entry(cwq->worklist.next,
struct work_struct, entry);
void (*f) (void *) = work->func;
void *data = work->data;
/* 将这个任务从工作队列链表中删除 */
list_del_init(cwq->worklist.next);
spin_unlock_irqrestore(&cwq->lock, flags);
/* 注意这里先解锁，然后再加锁。原因是因为
工作任务中的回调函数有可能会修改该工作队列，比如最简单的，把自己
重新加入到该队列中，如果不解锁，很可能造成死锁。
*/
BUG_ON(work->wq_data != cwq);
clear_bit(0, &work->pending);
f(data);/* 执行该工作任务中的回调函数 */
spin_lock_irqsave(&cwq->lock, flags);
cwq->remove_sequence++;/* 累加删除计数 */
wake_up(&cwq->work_done);/* 唤醒工作队列上的完成队列 */
}
cwq->run_depth--;
spin_unlock_irqrestore(&cwq->lock, flags);
}
/* 工作队列的线程函数 */
static int worker_thread(void *__cwq)
{
struct cpu_workqueue_struct *cwq = __cwq;
DECLARE_WAITQUEUE(wait, current);
struct k_sigaction sa;
sigset_t blocked;
current->flags |= PF_NOFREEZE;
set_user_nice(current, -5);
/* Block and flush all signals */
sigfillset(&blocked);
sigprocmask(SIG_BLOCK, &blocked, NULL);
flush_signals(current);
/* SIG_IGN makes children autoreap: see do_notify_parent(). */
sa.sa.sa_handler = SIG_IGN;
sa.sa.sa_flags = 0;
siginitset(&sa.sa.sa_mask, sigmask(SIGCHLD));
do_sigaction(SIGCHLD, &sa, (struct k_sigaction *)0);
set_current_state(TASK_INTERRUPTIBLE);
while (!kthread_should_stop()) {
add_wait_queue(&cwq->more_work, &wait);
if (list_empty(&cwq->worklist))
schedule();
else
__set_current_state(TASK_RUNNING);
remove_wait_queue(&cwq->more_work, &wait);
if (!list_empty(&cwq->worklist))
run_workqueue(cwq);
set_current_state(TASK_INTERRUPTIBLE);
}
__set_current_state(TASK_RUNNING);
return 0;
}
/*
刷新工作队列，就是让工作队列中的工作任务尽可能快的执行完
*/
static void flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
{
if (cwq->thread == current) {
/*
* Probably keventd trying to flush its own queue. So simply run
* it by hand rather than deadlocking.
*/
/*可以自己刷新自己*/
run_workqueue(cwq);
} else {
DEFINE_WAIT(wait);
long sequence_needed;
spin_lock_irq(&cwq->lock);
sequence_needed = cwq->insert_sequence;
while (sequence_needed - cwq->remove_sequence > 0) {
prepare_to_wait(&cwq->work_done, &wait,
TASK_UNINTERRUPTIBLE);
spin_unlock_irq(&cwq->lock);
schedule();/* 不停的让出CPU，好让其他线程可以工作 */
spin_lock_irq(&cwq->lock);
}
finish_wait(&cwq->work_done, &wait);
spin_unlock_irq(&cwq->lock);
}
}
/*
* flush_workqueue - ensure that any scheduled work has run to completion.
*
* Forces execution of the workqueue and blocks until its completion.
* This is typically used in driver shutdown handlers.
*
* This function will sample each workqueue's current insert_sequence number and
* will sleep until the head sequence is greater than or equal to that. This
* means that we sleep until all works which were queued on entry have been
* handled, but we are not livelocked by new incoming ones.
*
* This function used to run the workqueues itself. Now we just wait for the
* helper threads to do it.
*/
/*
刷新工作队列，就是让工作队列中的工作任务尽可能快的执行完
*/
void fastcall flush_workqueue(struct workqueue_struct *wq)
{
might_sleep();
if (is_single_threaded(wq)) {
/* Always use cpu 0's area. */
flush_cpu_workqueue(wq->cpu_wq + 0);
} else {
int cpu;
lock_cpu_hotplug();
for_each_online_cpu(cpu)
flush_cpu_workqueue(wq->cpu_wq + cpu);
unlock_cpu_hotplug();
}
}
/* 创建工作队列的线程 */
static struct task_struct *create_workqueue_thread(struct workqueue_struct *wq,
int cpu)
{
struct cpu_workqueue_struct *cwq = wq->cpu_wq + cpu;
struct task_struct *p;
spin_lock_init(&cwq->lock);
cwq->wq = wq;
cwq->thread = NULL;
cwq->insert_sequence = 0;
cwq->remove_sequence = 0;
INIT_LIST_HEAD(&cwq->worklist);
init_waitqueue_head(&cwq->more_work);
init_waitqueue_head(&cwq->work_done);
/* 设置下线程的名字 */
if (is_single_threaded(wq))
p = kthread_create(worker_thread, cwq, "%s", wq->name);
else
p = kthread_create(worker_thread, cwq, "%s/%d", wq->name, cpu);
if (IS_ERR(p))
return NULL;
cwq->thread = p;
return p;
}
/* 创建一个工作队列 */
struct workqueue_struct *__create_workqueue(const char *name,
int singlethread)
{
int cpu, destroy = 0;
struct workqueue_struct *wq;
struct task_struct *p;
BUG_ON(strlen(name) > 10);
wq = kmalloc(sizeof(*wq), GFP_KERNEL);
if (!wq)
return NULL;
memset(wq, 0, sizeof(*wq));
wq->name = name;
/* We don't need the distraction of CPUs appearing and vanishing. */
lock_cpu_hotplug();
if (singlethread) {
INIT_LIST_HEAD(&wq->list);
/* 如果是单CPU类型队列，就使用0 */
p = create_workqueue_thread(wq, 0);
if (!p)
destroy = 1;
else
wake_up_process(p);
} else {
spin_lock(&workqueue_lock);
list_add(&wq->list, &workqueues);
spin_unlock(&workqueue_lock);
for_each_online_cpu(cpu) {
p = create_workqueue_thread(wq, cpu);
if (p) {
kthread_bind(p, cpu);
wake_up_process(p);
} else
destroy = 1;
}
}
unlock_cpu_hotplug();
/*
* Was there any error during startup? If yes then clean up:
*/
if (destroy) {
destroy_workqueue(wq);
wq = NULL;
}
return wq;
}
static void cleanup_workqueue_thread(struct workqueue_struct *wq, int cpu)
{
struct cpu_workqueue_struct *cwq;
unsigned long flags;
struct task_struct *p;
cwq = wq->cpu_wq + cpu;
spin_lock_irqsave(&cwq->lock, flags);
p = cwq->thread;
cwq->thread = NULL;
spin_unlock_irqrestore(&cwq->lock, flags);
if (p)
kthread_stop(p);
}
/* 释放一个工作队列 */
void destroy_workqueue(struct workqueue_struct *wq)
{
int cpu;
flush_workqueue(wq);
/* We don't need the distraction of CPUs appearing and vanishing. */
lock_cpu_hotplug();
if (is_single_threaded(wq))
cleanup_workqueue_thread(wq, 0);
else {
for_each_online_cpu(cpu)
cleanup_workqueue_thread(wq, cpu);
spin_lock(&workqueue_lock);
list_del(&wq->list);
spin_unlock(&workqueue_lock);
}
unlock_cpu_hotplug();
kfree(wq);
}
/*
如果只是简单的一个任务，为了这个任务搞一个新的工作队列太麻烦了，
可以使用内部预配置的一个工作队列
*/
static struct workqueue_struct *keventd_wq;
int fastcall schedule_work(struct work_struct *work)
{
return queue_work(keventd_wq, work);
}
int fastcall schedule_delayed_work(struct work_struct *work, unsigned long delay)
{
return queue_delayed_work(keventd_wq, work, delay);
}
int schedule_delayed_work_on(int cpu,
struct work_struct *work, unsigned long delay)
{
int ret = 0;
struct timer_list *timer = &work->timer;
if (!test_and_set_bit(0, &work->pending)) {
BUG_ON(timer_pending(timer));
BUG_ON(!list_empty(&work->entry));
/* This stores keventd_wq for the moment, for the timer_fn */
work->wq_data = keventd_wq;
timer->expires = jiffies + delay;
timer->data = (unsigned long)work;
timer->function = delayed_work_timer_fn;
add_timer_on(timer, cpu);
ret = 1;
}
return ret;
}
void flush_scheduled_work(void)
{
flush_workqueue(keventd_wq);
}
int keventd_up(void)
{
return keventd_wq != NULL;
}
/*
判断当前上下文是不是在keventd_wq中
*/
int current_is_keventd(void)
{
struct cpu_workqueue_struct *cwq;
int cpu = smp_processor_id(); /* preempt-safe: keventd is per-cpu */
int ret = 0;
BUG_ON(!keventd_wq);
cwq = keventd_wq->cpu_wq + cpu;
if (current == cwq->thread)
ret = 1;
return ret;
}
/*
如果CPU支持热插拔，会使用下面的回调函数
*/
#ifdef CONFIG_HOTPLUG_CPU
/* Take the work from this (downed) CPU. */
static void take_over_work(struct workqueue_struct *wq, unsigned int cpu)
{
struct cpu_workqueue_struct *cwq = wq->cpu_wq + cpu;
LIST_HEAD(list);
struct work_struct *work;
spin_lock_irq(&cwq->lock);
list_splice_init(&cwq->worklist, &list);
while (!list_empty(&list)) {
printk("Taking work for %s\n", wq->name);
work = list_entry(list.next,struct work_struct,entry);
list_del(&work->entry);
__queue_work(wq->cpu_wq + smp_processor_id(), work);
}
spin_unlock_irq(&cwq->lock);
}
/* We're holding the cpucontrol mutex here */
static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
unsigned long action,
void *hcpu)
{
unsigned int hotcpu = (unsigned long)hcpu;
struct workqueue_struct *wq;
switch (action) {
case CPU_UP_PREPARE:
/* Create a new workqueue thread for it. */
list_for_each_entry(wq, &workqueues, list) {
if (create_workqueue_thread(wq, hotcpu) < 0) {
printk("workqueue for %i failed\n", hotcpu);
return NOTIFY_BAD;
}
}
break;
case CPU_ONLINE:
/* Kick off worker threads. */
list_for_each_entry(wq, &workqueues, list) {
kthread_bind(wq->cpu_wq[hotcpu].thread, hotcpu);
wake_up_process(wq->cpu_wq[hotcpu].thread);
}
break;
case CPU_UP_CANCELED:
list_for_each_entry(wq, &workqueues, list) {
/* Unbind so it can run. */
kthread_bind(wq->cpu_wq[hotcpu].thread,
smp_processor_id());
cleanup_workqueue_thread(wq, hotcpu);
}
break;
case CPU_DEAD:
list_for_each_entry(wq, &workqueues, list)
cleanup_workqueue_thread(wq, hotcpu);
list_for_each_entry(wq, &workqueues, list)
take_over_work(wq, hotcpu);
break;
}
return NOTIFY_OK;
}
#endif
void init_workqueues(void)
{
hotcpu_notifier(workqueue_cpu_callback, 0);
keventd_wq = create_workqueue("events");
BUG_ON(!keventd_wq);
}
EXPORT_SYMBOL_GPL(__create_workqueue);
EXPORT_SYMBOL_GPL(queue_work);
EXPORT_SYMBOL_GPL(queue_delayed_work);
EXPORT_SYMBOL_GPL(flush_workqueue);
EXPORT_SYMBOL_GPL(destroy_workqueue);
EXPORT_SYMBOL(schedule_work);
EXPORT_SYMBOL(schedule_delayed_work);
EXPORT_SYMBOL(schedule_delayed_work_on);
EXPORT_SYMBOL(flush_scheduled_work);
发表评论