[linux.git] / kernel / async.c

/*
 * async.c: Asynchronous function calls for boot performance
 *
 * (C) Copyright 2009 Intel Corporation
 * Author: Arjan van de Ven <[email protected]>
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; version 2
 * of the License.
 */


/*

Goals and Theory of Operation

The primary goal of this feature is to reduce the kernel boot time,
by doing various independent hardware delays and discovery operations
decoupled and not strictly serialized.

More specifically, the asynchronous function call concept allows
certain operations (primarily during system boot) to happen
asynchronously, out of order, while these operations still
have their externally visible parts happen sequentially and in-order.
(not unlike how out-of-order CPUs retire their instructions in order)

Key to the asynchronous function call implementation is the concept of
a "sequence cookie" (which, although it has an abstracted type, can be
thought of as a monotonically incrementing number).

The async core will assign each scheduled event such a sequence cookie and
pass this to the called functions.

The asynchronously called function should before doing a globally visible
operation, such as registering device numbers, call the
async_synchronize_cookie() function and pass in its own cookie. The
async_synchronize_cookie() function will make sure that all asynchronous
operations that were scheduled prior to the operation corresponding with the
cookie have completed.

Subsystem/driver initialization code that scheduled asynchronous probe
functions, but which shares global resources with other drivers/subsystems
that do not use the asynchronous call feature, need to do a full
synchronization with the async_synchronize_full() function, before returning
from their init function. This is to maintain strict ordering between the
asynchronous and synchronous parts of the kernel.

*/

#include <linux/async.h>
#include <linux/bug.h>
#include <linux/module.h>
#include <linux/wait.h>
#include <linux/sched.h>
#include <linux/init.h>
#include <linux/kthread.h>
#include <linux/delay.h>
#include <linux/slab.h>
#include <asm/atomic.h>

static async_cookie_t next_cookie = 1;

#define MAX_THREADS	256
#define MAX_WORK	32768

static LIST_HEAD(async_pending);
static LIST_HEAD(async_running);
static DEFINE_SPINLOCK(async_lock);

static int async_enabled = 0;

struct async_entry {
	struct list_head list;
	async_cookie_t   cookie;
	async_func_ptr	 *func;
	void             *data;
	struct list_head *running;
};

static DECLARE_WAIT_QUEUE_HEAD(async_done);
static DECLARE_WAIT_QUEUE_HEAD(async_new);

static atomic_t entry_count;
static atomic_t thread_count;

extern int initcall_debug;


/*
 * MUST be called with the lock held!
 */
static async_cookie_t  __lowest_in_progress(struct list_head *running)
{
	struct async_entry *entry;

	if (!list_empty(running)) {
		entry = list_first_entry(running,
			struct async_entry, list);
		return entry->cookie;
	}

	list_for_each_entry(entry, &async_pending, list)
		if (entry->running == running)
			return entry->cookie;

	return next_cookie;	/* "infinity" value */
}

static async_cookie_t  lowest_in_progress(struct list_head *running)
{
	unsigned long flags;
	async_cookie_t ret;

	spin_lock_irqsave(&async_lock, flags);
	ret = __lowest_in_progress(running);
	spin_unlock_irqrestore(&async_lock, flags);
	return ret;
}
/*
 * pick the first pending entry and run it
 */
static void run_one_entry(void)
{
	unsigned long flags;
	struct async_entry *entry;
	ktime_t calltime, delta, rettime;

	/* 1) pick one task from the pending queue */

	spin_lock_irqsave(&async_lock, flags);
	if (list_empty(&async_pending))
		goto out;
	entry = list_first_entry(&async_pending, struct async_entry, list);

	/* 2) move it to the running queue */
	list_move_tail(&entry->list, entry->running);
	spin_unlock_irqrestore(&async_lock, flags);

	/* 3) run it (and print duration)*/
	if (initcall_debug && system_state == SYSTEM_BOOTING) {
		printk("calling  %lli_%pF @ %i\n", (long long)entry->cookie,
			entry->func, task_pid_nr(current));
		calltime = ktime_get();
	}
	entry->func(entry->data, entry->cookie);
	if (initcall_debug && system_state == SYSTEM_BOOTING) {
		rettime = ktime_get();
		delta = ktime_sub(rettime, calltime);
		printk("initcall %lli_%pF returned 0 after %lld usecs\n",
			(long long)entry->cookie,
			entry->func,
			(long long)ktime_to_ns(delta) >> 10);
	}

	/* 4) remove it from the running queue */
	spin_lock_irqsave(&async_lock, flags);
	list_del(&entry->list);

	/* 5) free the entry  */
	kfree(entry);
	atomic_dec(&entry_count);

	spin_unlock_irqrestore(&async_lock, flags);

	/* 6) wake up any waiters. */
	wake_up(&async_done);
	return;

out:
	spin_unlock_irqrestore(&async_lock, flags);
}


static async_cookie_t __async_schedule(async_func_ptr *ptr, void *data, struct list_head *running)
{
	struct async_entry *entry;
	unsigned long flags;
	async_cookie_t newcookie;
	

	/* allow irq-off callers */
	entry = kzalloc(sizeof(struct async_entry), GFP_ATOMIC);

	/*
	 * If we're out of memory or if there's too much work
	 * pending already, we execute synchronously.
	 */
	if (!async_enabled || !entry || atomic_read(&entry_count) > MAX_WORK) {
		kfree(entry);
		spin_lock_irqsave(&async_lock, flags);
		newcookie = next_cookie++;
		spin_unlock_irqrestore(&async_lock, flags);

		/* low on memory.. run synchronously */
		ptr(data, newcookie);
		return newcookie;
	}
	entry->func = ptr;
	entry->data = data;
	entry->running = running;

	spin_lock_irqsave(&async_lock, flags);
	newcookie = entry->cookie = next_cookie++;
	list_add_tail(&entry->list, &async_pending);
	atomic_inc(&entry_count);
	spin_unlock_irqrestore(&async_lock, flags);
	wake_up(&async_new);
	return newcookie;
}

/**
 * async_schedule - schedule a function for asynchronous execution
 * @ptr: function to execute asynchronously
 * @data: data pointer to pass to the function
 *
 * Returns an async_cookie_t that may be used for checkpointing later.
 * Note: This function may be called from atomic or non-atomic contexts.
 */
async_cookie_t async_schedule(async_func_ptr *ptr, void *data)
{
	return __async_schedule(ptr, data, &async_running);
}
EXPORT_SYMBOL_GPL(async_schedule);

/**
 * async_schedule_domain - schedule a function for asynchronous execution within a certain domain
 * @ptr: function to execute asynchronously
 * @data: data pointer to pass to the function
 * @running: running list for the domain
 *
 * Returns an async_cookie_t that may be used for checkpointing later.
 * @running may be used in the async_synchronize_*_domain() functions
 * to wait within a certain synchronization domain rather than globally.
 * A synchronization domain is specified via the running queue @running to use.
 * Note: This function may be called from atomic or non-atomic contexts.
 */
async_cookie_t async_schedule_domain(async_func_ptr *ptr, void *data,
				     struct list_head *running)
{
	return __async_schedule(ptr, data, running);
}
EXPORT_SYMBOL_GPL(async_schedule_domain);

/**
 * async_synchronize_full - synchronize all asynchronous function calls
 *
 * This function waits until all asynchronous function calls have been done.
 */
void async_synchronize_full(void)
{
	do {
		async_synchronize_cookie(next_cookie);
	} while (!list_empty(&async_running) || !list_empty(&async_pending));
}
EXPORT_SYMBOL_GPL(async_synchronize_full);

/**
 * async_synchronize_full_domain - synchronize all asynchronous function within a certain domain
 * @list: running list to synchronize on
 *
 * This function waits until all asynchronous function calls for the
 * synchronization domain specified by the running list @list have been done.
 */
void async_synchronize_full_domain(struct list_head *list)
{
	async_synchronize_cookie_domain(next_cookie, list);
}
EXPORT_SYMBOL_GPL(async_synchronize_full_domain);

/**
 * async_synchronize_cookie_domain - synchronize asynchronous function calls within a certain domain with cookie checkpointing
 * @cookie: async_cookie_t to use as checkpoint
 * @running: running list to synchronize on
 *
 * This function waits until all asynchronous function calls for the
 * synchronization domain specified by the running list @list submitted
 * prior to @cookie have been done.
 */
void async_synchronize_cookie_domain(async_cookie_t cookie,
				     struct list_head *running)
{
	ktime_t starttime, delta, endtime;

	if (initcall_debug && system_state == SYSTEM_BOOTING) {
		printk("async_waiting @ %i\n", task_pid_nr(current));
		starttime = ktime_get();
	}

	wait_event(async_done, lowest_in_progress(running) >= cookie);

	if (initcall_debug && system_state == SYSTEM_BOOTING) {
		endtime = ktime_get();
		delta = ktime_sub(endtime, starttime);

		printk("async_continuing @ %i after %lli usec\n",
			task_pid_nr(current),
			(long long)ktime_to_ns(delta) >> 10);
	}
}
EXPORT_SYMBOL_GPL(async_synchronize_cookie_domain);

/**
 * async_synchronize_cookie - synchronize asynchronous function calls with cookie checkpointing
 * @cookie: async_cookie_t to use as checkpoint
 *
 * This function waits until all asynchronous function calls prior to @cookie
 * have been done.
 */
void async_synchronize_cookie(async_cookie_t cookie)
{
	async_synchronize_cookie_domain(cookie, &async_running);
}
EXPORT_SYMBOL_GPL(async_synchronize_cookie);


static int async_thread(void *unused)
{
	DECLARE_WAITQUEUE(wq, current);
	add_wait_queue(&async_new, &wq);

	while (!kthread_should_stop()) {
		int ret = HZ;
		set_current_state(TASK_INTERRUPTIBLE);
		/*
		 * check the list head without lock.. false positives
		 * are dealt with inside run_one_entry() while holding
		 * the lock.
		 */
		rmb();
		if (!list_empty(&async_pending))
			run_one_entry();
		else
			ret = schedule_timeout(HZ);

		if (ret == 0) {
			/*
			 * we timed out, this means we as thread are redundant.
			 * we sign off and die, but we to avoid any races there
			 * is a last-straw check to see if work snuck in.
			 */
			atomic_dec(&thread_count);
			wmb(); /* manager must see our departure first */
			if (list_empty(&async_pending))
				break;
			/*
			 * woops work came in between us timing out and us
			 * signing off; we need to stay alive and keep working.
			 */
			atomic_inc(&thread_count);
		}
	}
	remove_wait_queue(&async_new, &wq);

	return 0;
}

static int async_manager_thread(void *unused)
{
	DECLARE_WAITQUEUE(wq, current);
	add_wait_queue(&async_new, &wq);

	while (!kthread_should_stop()) {
		int tc, ec;

		set_current_state(TASK_INTERRUPTIBLE);

		tc = atomic_read(&thread_count);
		rmb();
		ec = atomic_read(&entry_count);

		while (tc < ec && tc < MAX_THREADS) {
			if (IS_ERR(kthread_run(async_thread, NULL, "async/%i",
					       tc))) {
				msleep(100);
				continue;
			}
			atomic_inc(&thread_count);
			tc++;
		}

		schedule();
	}
	remove_wait_queue(&async_new, &wq);

	return 0;
}

static int __init async_init(void)
{
	async_enabled =
		!IS_ERR(kthread_run(async_manager_thread, NULL, "async/mgr"));

	WARN_ON(!async_enabled);
	return 0;
}

core_initcall(async_init);
Commit	Line	Data
22a9d645 AV	1	/*
	2	* async.c: Asynchronous function calls for boot performance
	3	*
	4	* (C) Copyright 2009 Intel Corporation
	5	* Author: Arjan van de Ven <[email protected]>
	6	*
	7	* This program is free software; you can redistribute it and/or
	8	* modify it under the terms of the GNU General Public License
	9	* as published by the Free Software Foundation; version 2
	10	* of the License.
	11	*/
	12
	13
	14	/*
	15
	16	Goals and Theory of Operation
	17
	18	The primary goal of this feature is to reduce the kernel boot time,
	19	by doing various independent hardware delays and discovery operations
	20	decoupled and not strictly serialized.
	21
	22	More specifically, the asynchronous function call concept allows
	23	certain operations (primarily during system boot) to happen
	24	asynchronously, out of order, while these operations still
	25	have their externally visible parts happen sequentially and in-order.
	26	(not unlike how out-of-order CPUs retire their instructions in order)
	27
	28	Key to the asynchronous function call implementation is the concept of
	29	a "sequence cookie" (which, although it has an abstracted type, can be
	30	thought of as a monotonically incrementing number).
	31
	32	The async core will assign each scheduled event such a sequence cookie and
	33	pass this to the called functions.
	34
	35	The asynchronously called function should before doing a globally visible
	36	operation, such as registering device numbers, call the
	37	async_synchronize_cookie() function and pass in its own cookie. The
	38	async_synchronize_cookie() function will make sure that all asynchronous
	39	operations that were scheduled prior to the operation corresponding with the
	40	cookie have completed.
	41
	42	Subsystem/driver initialization code that scheduled asynchronous probe
	43	functions, but which shares global resources with other drivers/subsystems
	44	that do not use the asynchronous call feature, need to do a full
	45	synchronization with the async_synchronize_full() function, before returning
	46	from their init function. This is to maintain strict ordering between the
	47	asynchronous and synchronous parts of the kernel.
	48
	49	*/
	50
	51	#include <linux/async.h>
97107943	52	#include <linux/bug.h>
22a9d645 AV	53	#include <linux/module.h>
	54	#include <linux/wait.h>
	55	#include <linux/sched.h>
	56	#include <linux/init.h>
	57	#include <linux/kthread.h>
86532d8b	58	#include <linux/delay.h>
5a0e3ad6	59	#include <linux/slab.h>
22a9d645 AV	60	#include <asm/atomic.h>
	61
	62	static async_cookie_t next_cookie = 1;
	63
	64	#define MAX_THREADS 256
	65	#define MAX_WORK 32768
	66
	67	static LIST_HEAD(async_pending);
	68	static LIST_HEAD(async_running);
	69	static DEFINE_SPINLOCK(async_lock);
	70
cdb80f63 AV	71	static int async_enabled = 0;
cdb80f63 AV	72
22a9d645 AV	73	struct async_entry {
	74	struct list_head list;
	75	async_cookie_t cookie;
	76	async_func_ptr *func;
	77	void *data;
	78	struct list_head *running;
	79	};
	80
	81	static DECLARE_WAIT_QUEUE_HEAD(async_done);
	82	static DECLARE_WAIT_QUEUE_HEAD(async_new);
	83
	84	static atomic_t entry_count;
	85	static atomic_t thread_count;
	86
	87	extern int initcall_debug;
	88
	89
	90	/*
	91	* MUST be called with the lock held!
	92	*/
	93	static async_cookie_t __lowest_in_progress(struct list_head *running)
	94	{
	95	struct async_entry *entry;
d5a877e8	96
37a76bd4 AV	97	if (!list_empty(running)) {
37a76bd4 AV	98	entry = list_first_entry(running,
22a9d645	99	struct async_entry, list);
3af968e0	100	return entry->cookie;
22a9d645 AV	101	}
22a9d645 AV	102
3af968e0 LT	103	list_for_each_entry(entry, &async_pending, list)
	104	if (entry->running == running)
	105	return entry->cookie;
d5a877e8	106
3af968e0	107	return next_cookie; /* "infinity" value */
22a9d645	108	}
37a76bd4 AV	109
	110	static async_cookie_t lowest_in_progress(struct list_head *running)
	111	{
	112	unsigned long flags;
	113	async_cookie_t ret;
	114
	115	spin_lock_irqsave(&async_lock, flags);
	116	ret = __lowest_in_progress(running);
	117	spin_unlock_irqrestore(&async_lock, flags);
	118	return ret;
	119	}
22a9d645 AV	120	/*
	121	* pick the first pending entry and run it
	122	*/
	123	static void run_one_entry(void)
	124	{
	125	unsigned long flags;
	126	struct async_entry *entry;
	127	ktime_t calltime, delta, rettime;
	128
	129	/* 1) pick one task from the pending queue */
	130
	131	spin_lock_irqsave(&async_lock, flags);
	132	if (list_empty(&async_pending))
	133	goto out;
	134	entry = list_first_entry(&async_pending, struct async_entry, list);
	135
	136	/* 2) move it to the running queue */
f7de7621	137	list_move_tail(&entry->list, entry->running);
22a9d645 AV	138	spin_unlock_irqrestore(&async_lock, flags);
	139
	140	/* 3) run it (and print duration)*/
ad160d23	141	if (initcall_debug && system_state == SYSTEM_BOOTING) {
58763a29 AM	142	printk("calling %lli_%pF @ %i\n", (long long)entry->cookie,
58763a29 AM	143	entry->func, task_pid_nr(current));
22a9d645 AV	144	calltime = ktime_get();
	145	}
	146	entry->func(entry->data, entry->cookie);
ad160d23	147	if (initcall_debug && system_state == SYSTEM_BOOTING) {
22a9d645 AV	148	rettime = ktime_get();
22a9d645 AV	149	delta = ktime_sub(rettime, calltime);
58763a29 AM	150	printk("initcall %lli_%pF returned 0 after %lld usecs\n",
	151	(long long)entry->cookie,
	152	entry->func,
	153	(long long)ktime_to_ns(delta) >> 10);
22a9d645 AV	154	}
	155
	156	/* 4) remove it from the running queue */
	157	spin_lock_irqsave(&async_lock, flags);
	158	list_del(&entry->list);
	159
	160	/* 5) free the entry */
	161	kfree(entry);
	162	atomic_dec(&entry_count);
	163
	164	spin_unlock_irqrestore(&async_lock, flags);
	165
	166	/* 6) wake up any waiters. */
	167	wake_up(&async_done);
	168	return;
	169
	170	out:
	171	spin_unlock_irqrestore(&async_lock, flags);
	172	}
	173
	174
	175	static async_cookie_t __async_schedule(async_func_ptr ptr, void data, struct list_head *running)
	176	{
	177	struct async_entry *entry;
	178	unsigned long flags;
	179	async_cookie_t newcookie;
	180
	181
	182	/* allow irq-off callers */
	183	entry = kzalloc(sizeof(struct async_entry), GFP_ATOMIC);
	184
	185	/*
	186	* If we're out of memory or if there's too much work
	187	* pending already, we execute synchronously.
	188	*/
cdb80f63	189	if (!async_enabled \|\| !entry \|\| atomic_read(&entry_count) > MAX_WORK) {
22a9d645 AV	190	kfree(entry);
	191	spin_lock_irqsave(&async_lock, flags);
	192	newcookie = next_cookie++;
	193	spin_unlock_irqrestore(&async_lock, flags);
	194
	195	/* low on memory.. run synchronously */
	196	ptr(data, newcookie);
	197	return newcookie;
	198	}
	199	entry->func = ptr;
	200	entry->data = data;
	201	entry->running = running;
	202
	203	spin_lock_irqsave(&async_lock, flags);
	204	newcookie = entry->cookie = next_cookie++;
	205	list_add_tail(&entry->list, &async_pending);
	206	atomic_inc(&entry_count);
	207	spin_unlock_irqrestore(&async_lock, flags);
	208	wake_up(&async_new);
	209	return newcookie;
	210	}
	211
f30d5b30 CH	212	/**
	213	* async_schedule - schedule a function for asynchronous execution
	214	* @ptr: function to execute asynchronously
	215	* @data: data pointer to pass to the function
	216	*
	217	* Returns an async_cookie_t that may be used for checkpointing later.
	218	* Note: This function may be called from atomic or non-atomic contexts.
	219	*/
22a9d645 AV	220	async_cookie_t async_schedule(async_func_ptr ptr, void data)
22a9d645 AV	221	{
7a89bbc7	222	return __async_schedule(ptr, data, &async_running);
22a9d645 AV	223	}
	224	EXPORT_SYMBOL_GPL(async_schedule);
	225
f30d5b30	226	/**
766ccb9e	227	* async_schedule_domain - schedule a function for asynchronous execution within a certain domain
f30d5b30 CH	228	* @ptr: function to execute asynchronously
f30d5b30 CH	229	* @data: data pointer to pass to the function
766ccb9e	230	* @running: running list for the domain
f30d5b30 CH	231	*
f30d5b30 CH	232	* Returns an async_cookie_t that may be used for checkpointing later.
766ccb9e CH	233	* @running may be used in the async_synchronize_*_domain() functions
	234	* to wait within a certain synchronization domain rather than globally.
	235	* A synchronization domain is specified via the running queue @running to use.
f30d5b30 CH	236	* Note: This function may be called from atomic or non-atomic contexts.
f30d5b30 CH	237	*/
766ccb9e CH	238	async_cookie_t async_schedule_domain(async_func_ptr ptr, void data,
766ccb9e CH	239	struct list_head *running)
22a9d645 AV	240	{
	241	return __async_schedule(ptr, data, running);
	242	}
766ccb9e	243	EXPORT_SYMBOL_GPL(async_schedule_domain);
22a9d645	244
f30d5b30 CH	245	/**
	246	* async_synchronize_full - synchronize all asynchronous function calls
	247	*
	248	* This function waits until all asynchronous function calls have been done.
	249	*/
22a9d645 AV	250	void async_synchronize_full(void)
22a9d645 AV	251	{
33b04b93 AV	252	do {
	253	async_synchronize_cookie(next_cookie);
	254	} while (!list_empty(&async_running) \|\| !list_empty(&async_pending));
22a9d645 AV	255	}
	256	EXPORT_SYMBOL_GPL(async_synchronize_full);
	257
f30d5b30	258	/**
766ccb9e	259	* async_synchronize_full_domain - synchronize all asynchronous function within a certain domain
f30d5b30 CH	260	* @list: running list to synchronize on
f30d5b30 CH	261	*
766ccb9e CH	262	* This function waits until all asynchronous function calls for the
766ccb9e CH	263	* synchronization domain specified by the running list @list have been done.
f30d5b30	264	*/
766ccb9e	265	void async_synchronize_full_domain(struct list_head *list)
22a9d645	266	{
766ccb9e	267	async_synchronize_cookie_domain(next_cookie, list);
22a9d645	268	}
766ccb9e	269	EXPORT_SYMBOL_GPL(async_synchronize_full_domain);
22a9d645	270
f30d5b30	271	/**
766ccb9e	272	* async_synchronize_cookie_domain - synchronize asynchronous function calls within a certain domain with cookie checkpointing
f30d5b30 CH	273	* @cookie: async_cookie_t to use as checkpoint
	274	* @running: running list to synchronize on
	275	*
766ccb9e CH	276	* This function waits until all asynchronous function calls for the
	277	* synchronization domain specified by the running list @list submitted
	278	* prior to @cookie have been done.
f30d5b30	279	*/
766ccb9e CH	280	void async_synchronize_cookie_domain(async_cookie_t cookie,
766ccb9e CH	281	struct list_head *running)
22a9d645 AV	282	{
	283	ktime_t starttime, delta, endtime;
	284
ad160d23	285	if (initcall_debug && system_state == SYSTEM_BOOTING) {
22a9d645 AV	286	printk("async_waiting @ %i\n", task_pid_nr(current));
	287	starttime = ktime_get();
	288	}
	289
37a76bd4	290	wait_event(async_done, lowest_in_progress(running) >= cookie);
22a9d645	291
ad160d23	292	if (initcall_debug && system_state == SYSTEM_BOOTING) {
22a9d645 AV	293	endtime = ktime_get();
	294	delta = ktime_sub(endtime, starttime);
	295
	296	printk("async_continuing @ %i after %lli usec\n",
58763a29 AM	297	task_pid_nr(current),
58763a29 AM	298	(long long)ktime_to_ns(delta) >> 10);
22a9d645 AV	299	}
22a9d645 AV	300	}
766ccb9e	301	EXPORT_SYMBOL_GPL(async_synchronize_cookie_domain);
22a9d645	302
f30d5b30 CH	303	/**
	304	* async_synchronize_cookie - synchronize asynchronous function calls with cookie checkpointing
	305	* @cookie: async_cookie_t to use as checkpoint
	306	*
	307	* This function waits until all asynchronous function calls prior to @cookie
	308	* have been done.
	309	*/
22a9d645 AV	310	void async_synchronize_cookie(async_cookie_t cookie)
22a9d645 AV	311	{
766ccb9e	312	async_synchronize_cookie_domain(cookie, &async_running);
22a9d645 AV	313	}
	314	EXPORT_SYMBOL_GPL(async_synchronize_cookie);
	315
	316
	317	static int async_thread(void *unused)
	318	{
	319	DECLARE_WAITQUEUE(wq, current);
	320	add_wait_queue(&async_new, &wq);
	321
	322	while (!kthread_should_stop()) {
	323	int ret = HZ;
	324	set_current_state(TASK_INTERRUPTIBLE);
	325	/*
	326	* check the list head without lock.. false positives
	327	* are dealt with inside run_one_entry() while holding
	328	* the lock.
	329	*/
	330	rmb();
	331	if (!list_empty(&async_pending))
	332	run_one_entry();
	333	else
	334	ret = schedule_timeout(HZ);
	335
	336	if (ret == 0) {
	337	/*
	338	* we timed out, this means we as thread are redundant.
	339	* we sign off and die, but we to avoid any races there
	340	* is a last-straw check to see if work snuck in.
	341	*/
	342	atomic_dec(&thread_count);
	343	wmb(); /* manager must see our departure first */
	344	if (list_empty(&async_pending))
	345	break;
	346	/*
	347	* woops work came in between us timing out and us
	348	* signing off; we need to stay alive and keep working.
	349	*/
	350	atomic_inc(&thread_count);
	351	}
	352	}
	353	remove_wait_queue(&async_new, &wq);
	354
	355	return 0;
	356	}
	357
	358	static int async_manager_thread(void *unused)
	359	{
	360	DECLARE_WAITQUEUE(wq, current);
	361	add_wait_queue(&async_new, &wq);
	362
	363	while (!kthread_should_stop()) {
	364	int tc, ec;
	365
	366	set_current_state(TASK_INTERRUPTIBLE);
	367
	368	tc = atomic_read(&thread_count);
	369	rmb();
	370	ec = atomic_read(&entry_count);
	371
	372	while (tc < ec && tc < MAX_THREADS) {
86532d8b CH	373	if (IS_ERR(kthread_run(async_thread, NULL, "async/%i",
	374	tc))) {
	375	msleep(100);
	376	continue;
	377	}
22a9d645 AV	378	atomic_inc(&thread_count);
	379	tc++;
	380	}
	381
	382	schedule();
	383	}
	384	remove_wait_queue(&async_new, &wq);
	385
	386	return 0;
	387	}
	388
	389	static int __init async_init(void)
	390	{
97107943 AV	391	async_enabled =
97107943 AV	392	!IS_ERR(kthread_run(async_manager_thread, NULL, "async/mgr"));
22a9d645	393
97107943 AV	394	WARN_ON(!async_enabled);
97107943 AV	395	return 0;
cdb80f63 AV	396	}
cdb80f63 AV	397
22a9d645	398	core_initcall(async_init);