709 lines · 17.7 KB · C
RawDownload
/* * 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> * * Made to use alloc_percpu by Christoph Lameter <clameter@sgi.com>. */ #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>#include <linux/hardirq.h>#include <linux/mempolicy.h> /* * The per-CPU workqueue (if single thread, we always use the first * possible cpu). * * The sequence counters are for flush_scheduled_work().  It wants to wait * 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. */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;	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;	const char *name;	struct list_head list; 	/* Empty if single thread */}; /* All the per-cpu workqueues on the system, for hotplug cpu to add/remove   threads to each one as cpus come/go. */static DEFINE_MUTEX(workqueue_mutex);static LIST_HEAD(workqueues); static int singlethread_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 - queue work on a workqueue * @wq: workqueue to use * @work: work to queue * * Returns 0 if @work was already on a queue, non-zero otherwise. * * 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){	int ret = 0, cpu = get_cpu(); 	if (!test_and_set_bit(0, &work->pending)) {		if (unlikely(is_single_threaded(wq)))			cpu = singlethread_cpu;		BUG_ON(!list_empty(&work->entry));		__queue_work(per_cpu_ptr(wq->cpu_wq, cpu), work);		ret = 1;	}	put_cpu();	return ret;}EXPORT_SYMBOL_GPL(queue_work); static void delayed_work_timer_fn(unsigned long __data){	struct delayed_work *dwork = (struct delayed_work *)__data;	struct workqueue_struct *wq = dwork->work.wq_data;	int cpu = smp_processor_id(); 	if (unlikely(is_single_threaded(wq)))		cpu = singlethread_cpu; 	__queue_work(per_cpu_ptr(wq->cpu_wq, cpu), &dwork->work);} /** * queue_delayed_work - queue work on a workqueue after delay * @wq: workqueue to use * @work: delayable work to queue * @delay: number of jiffies to wait before queueing * * Returns 0 if @work was already on a queue, non-zero otherwise. */int fastcall queue_delayed_work(struct workqueue_struct *wq,			struct delayed_work *dwork, unsigned long delay){	int ret = 0;	struct timer_list *timer = &dwork->timer;	struct work_struct *work = &dwork->work; 	if (delay == 0)		return queue_work(wq, work); 	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)dwork;		timer->function = delayed_work_timer_fn;		add_timer(timer);		ret = 1;	}	return ret;}EXPORT_SYMBOL_GPL(queue_delayed_work); /** * queue_delayed_work_on - queue work on specific CPU after delay * @cpu: CPU number to execute work on * @wq: workqueue to use * @work: work to queue * @delay: number of jiffies to wait before queueing * * Returns 0 if @work was already on a queue, non-zero otherwise. */int queue_delayed_work_on(int cpu, struct workqueue_struct *wq,			struct delayed_work *dwork, unsigned long delay){	int ret = 0;	struct timer_list *timer = &dwork->timer;	struct work_struct *work = &dwork->work; 	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)dwork;		timer->function = delayed_work_timer_fn;		add_timer_on(timer, cpu);		ret = 1;	}	return ret;}EXPORT_SYMBOL_GPL(queue_delayed_work_on); static 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);		work_func_t f = 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); 	/*	 * We inherited MPOL_INTERLEAVE from the booting kernel.	 * Set MPOL_DEFAULT to insure node local allocations.	 */	numa_default_policy(); 	/* 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();			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. * @wq: workqueue to flush * * 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 first cpu's area. */		flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, singlethread_cpu));	} else {		int cpu; 		mutex_lock(&workqueue_mutex);		for_each_online_cpu(cpu)			flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, cpu));		mutex_unlock(&workqueue_mutex);	}}EXPORT_SYMBOL_GPL(flush_workqueue); static struct task_struct *create_workqueue_thread(struct workqueue_struct *wq,						   int cpu){	struct cpu_workqueue_struct *cwq = per_cpu_ptr(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; 	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;	mutex_lock(&workqueue_mutex);	if (singlethread) {		INIT_LIST_HEAD(&wq->list);		p = create_workqueue_thread(wq, singlethread_cpu);		if (!p)			destroy = 1;		else			wake_up_process(p);	} else {		list_add(&wq->list, &workqueues);		for_each_online_cpu(cpu) {			p = create_workqueue_thread(wq, cpu);			if (p) {				kthread_bind(p, cpu);				wake_up_process(p);			} else				destroy = 1;		}	}	mutex_unlock(&workqueue_mutex); 	/*	 * Was there any error during startup? If yes then clean up:	 */	if (destroy) {		destroy_workqueue(wq);		wq = NULL;	}	return wq;}EXPORT_SYMBOL_GPL(__create_workqueue); static void cleanup_workqueue_thread(struct workqueue_struct *wq, int cpu){	struct cpu_workqueue_struct *cwq;	unsigned long flags;	struct task_struct *p; 	cwq = per_cpu_ptr(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);} /** * destroy_workqueue - safely terminate a workqueue * @wq: target workqueue * * Safely destroy a workqueue. All work currently pending will be done first. */void destroy_workqueue(struct workqueue_struct *wq){	int cpu; 	flush_workqueue(wq); 	/* We don't need the distraction of CPUs appearing and vanishing. */	mutex_lock(&workqueue_mutex);	if (is_single_threaded(wq))		cleanup_workqueue_thread(wq, singlethread_cpu);	else {		for_each_online_cpu(cpu)			cleanup_workqueue_thread(wq, cpu);		list_del(&wq->list);	}	mutex_unlock(&workqueue_mutex);	free_percpu(wq->cpu_wq);	kfree(wq);}EXPORT_SYMBOL_GPL(destroy_workqueue); static struct workqueue_struct *keventd_wq; /** * schedule_work - put work task in global workqueue * @work: job to be done * * This puts a job in the kernel-global workqueue. */int fastcall schedule_work(struct work_struct *work){	return queue_work(keventd_wq, work);}EXPORT_SYMBOL(schedule_work); /** * schedule_delayed_work - put work task in global workqueue after delay * @dwork: job to be done * @delay: number of jiffies to wait or 0 for immediate execution * * After waiting for a given time this puts a job in the kernel-global * workqueue. */int fastcall schedule_delayed_work(struct delayed_work *dwork, unsigned long delay){	return queue_delayed_work(keventd_wq, dwork, delay);}EXPORT_SYMBOL(schedule_delayed_work); /** * schedule_delayed_work_on - queue work in global workqueue on CPU after delay * @cpu: cpu to use * @dwork: job to be done * @delay: number of jiffies to wait * * After waiting for a given time this puts a job in the kernel-global * workqueue on the specified CPU. */int schedule_delayed_work_on(int cpu,			struct delayed_work *dwork, unsigned long delay){	return queue_delayed_work_on(cpu, keventd_wq, dwork, delay);

Lines 1–500 of 709.