[mirror_ubuntu-artful-kernel.git] / kernel / padata.c

/*
 * padata.c - generic interface to process data streams in parallel
 *
 * Copyright (C) 2008, 2009 secunet Security Networks AG
 * Copyright (C) 2008, 2009 Steffen Klassert <steffen.klassert@secunet.com>
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms and conditions of the GNU General Public License,
 * version 2, as published by the Free Software Foundation.
 *
 * This program is distributed in the hope it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License along with
 * this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin St - Fifth Floor, Boston, MA 02110-1301 USA.
 */

#include <linux/module.h>
#include <linux/cpumask.h>
#include <linux/err.h>
#include <linux/cpu.h>
#include <linux/padata.h>
#include <linux/mutex.h>
#include <linux/sched.h>
#include <linux/slab.h>
#include <linux/rcupdate.h>

#define MAX_SEQ_NR INT_MAX - NR_CPUS
#define MAX_OBJ_NUM 1000

static int padata_index_to_cpu(struct parallel_data *pd, int cpu_index)
{
	int cpu, target_cpu;

	target_cpu = cpumask_first(pd->cpumask);
	for (cpu = 0; cpu < cpu_index; cpu++)
		target_cpu = cpumask_next(target_cpu, pd->cpumask);

	return target_cpu;
}

static int padata_cpu_hash(struct padata_priv *padata)
{
	int cpu_index;
	struct parallel_data *pd;

	pd =  padata->pd;

	/*
	 * Hash the sequence numbers to the cpus by taking
	 * seq_nr mod. number of cpus in use.
	 */
	cpu_index =  padata->seq_nr % cpumask_weight(pd->cpumask);

	return padata_index_to_cpu(pd, cpu_index);
}

static void padata_parallel_worker(struct work_struct *work)
{
	struct padata_queue *queue;
	struct parallel_data *pd;
	struct padata_instance *pinst;
	LIST_HEAD(local_list);

	local_bh_disable();
	queue = container_of(work, struct padata_queue, pwork);
	pd = queue->pd;
	pinst = pd->pinst;

	spin_lock(&queue->parallel.lock);
	list_replace_init(&queue->parallel.list, &local_list);
	spin_unlock(&queue->parallel.lock);

	while (!list_empty(&local_list)) {
		struct padata_priv *padata;

		padata = list_entry(local_list.next,
				    struct padata_priv, list);

		list_del_init(&padata->list);

		padata->parallel(padata);
	}

	local_bh_enable();
}

/*
 * padata_do_parallel - padata parallelization function
 *
 * @pinst: padata instance
 * @padata: object to be parallelized
 * @cb_cpu: cpu the serialization callback function will run on,
 *          must be in the cpumask of padata.
 *
 * The parallelization callback function will run with BHs off.
 * Note: Every object which is parallelized by padata_do_parallel
 * must be seen by padata_do_serial.
 */
int padata_do_parallel(struct padata_instance *pinst,
		       struct padata_priv *padata, int cb_cpu)
{
	int target_cpu, err;
	struct padata_queue *queue;
	struct parallel_data *pd;

	rcu_read_lock_bh();

	pd = rcu_dereference(pinst->pd);

	err = 0;
	if (!(pinst->flags & PADATA_INIT))
		goto out;

	err =  -EBUSY;
	if ((pinst->flags & PADATA_RESET))
		goto out;

	if (atomic_read(&pd->refcnt) >= MAX_OBJ_NUM)
		goto out;

	err = -EINVAL;
	if (!cpumask_test_cpu(cb_cpu, pd->cpumask))
		goto out;

	err = -EINPROGRESS;
	atomic_inc(&pd->refcnt);
	padata->pd = pd;
	padata->cb_cpu = cb_cpu;

	if (unlikely(atomic_read(&pd->seq_nr) == pd->max_seq_nr))
		atomic_set(&pd->seq_nr, -1);

	padata->seq_nr = atomic_inc_return(&pd->seq_nr);

	target_cpu = padata_cpu_hash(padata);
	queue = per_cpu_ptr(pd->queue, target_cpu);

	spin_lock(&queue->parallel.lock);
	list_add_tail(&padata->list, &queue->parallel.list);
	spin_unlock(&queue->parallel.lock);

	queue_work_on(target_cpu, pinst->wq, &queue->pwork);

out:
	rcu_read_unlock_bh();

	return err;
}
EXPORT_SYMBOL(padata_do_parallel);

static struct padata_priv *padata_get_next(struct parallel_data *pd)
{
	int cpu, num_cpus, empty, calc_seq_nr;
	int seq_nr, next_nr, overrun, next_overrun;
	struct padata_queue *queue, *next_queue;
	struct padata_priv *padata;
	struct padata_list *reorder;

	empty = 0;
	next_nr = -1;
	next_overrun = 0;
	next_queue = NULL;

	num_cpus = cpumask_weight(pd->cpumask);

	for_each_cpu(cpu, pd->cpumask) {
		queue = per_cpu_ptr(pd->queue, cpu);
		reorder = &queue->reorder;

		/*
		 * Calculate the seq_nr of the object that should be
		 * next in this queue.
		 */
		overrun = 0;
		calc_seq_nr = (atomic_read(&queue->num_obj) * num_cpus)
			       + queue->cpu_index;

		if (unlikely(calc_seq_nr > pd->max_seq_nr)) {
			calc_seq_nr = calc_seq_nr - pd->max_seq_nr - 1;
			overrun = 1;
		}

		if (!list_empty(&reorder->list)) {
			padata = list_entry(reorder->list.next,
					    struct padata_priv, list);

			seq_nr  = padata->seq_nr;
			BUG_ON(calc_seq_nr != seq_nr);
		} else {
			seq_nr = calc_seq_nr;
			empty++;
		}

		if (next_nr < 0 || seq_nr < next_nr
		    || (next_overrun && !overrun)) {
			next_nr = seq_nr;
			next_overrun = overrun;
			next_queue = queue;
		}
	}

	padata = NULL;

	if (empty == num_cpus)
		goto out;

	reorder = &next_queue->reorder;

	if (!list_empty(&reorder->list)) {
		padata = list_entry(reorder->list.next,
				    struct padata_priv, list);

		if (unlikely(next_overrun)) {
			for_each_cpu(cpu, pd->cpumask) {
				queue = per_cpu_ptr(pd->queue, cpu);
				atomic_set(&queue->num_obj, 0);
			}
		}

		spin_lock(&reorder->lock);
		list_del_init(&padata->list);
		atomic_dec(&pd->reorder_objects);
		spin_unlock(&reorder->lock);

		atomic_inc(&next_queue->num_obj);

		goto out;
	}

	queue = per_cpu_ptr(pd->queue, smp_processor_id());
	if (queue->cpu_index == next_queue->cpu_index) {
		padata = ERR_PTR(-ENODATA);
		goto out;
	}

	padata = ERR_PTR(-EINPROGRESS);
out:
	return padata;
}

static void padata_reorder(struct parallel_data *pd)
{
	struct padata_priv *padata;
	struct padata_queue *queue;
	struct padata_instance *pinst = pd->pinst;

	if (!spin_trylock_bh(&pd->lock))
		return;

	while (1) {
		padata = padata_get_next(pd);

		if (!padata || PTR_ERR(padata) == -EINPROGRESS)
			break;

		if (PTR_ERR(padata) == -ENODATA) {
			del_timer(&pd->timer);
			spin_unlock_bh(&pd->lock);
			return;
		}

		queue = per_cpu_ptr(pd->queue, padata->cb_cpu);

		spin_lock(&queue->serial.lock);
		list_add_tail(&padata->list, &queue->serial.list);
		spin_unlock(&queue->serial.lock);

		queue_work_on(padata->cb_cpu, pinst->wq, &queue->swork);
	}

	spin_unlock_bh(&pd->lock);

	if (atomic_read(&pd->reorder_objects)
			&& !(pinst->flags & PADATA_RESET))
		mod_timer(&pd->timer, jiffies + HZ);
	else
		del_timer(&pd->timer);

	return;
}

static void padata_reorder_timer(unsigned long arg)
{
	struct parallel_data *pd = (struct parallel_data *)arg;

	padata_reorder(pd);
}

static void padata_serial_worker(struct work_struct *work)
{
	struct padata_queue *queue;
	struct parallel_data *pd;
	LIST_HEAD(local_list);

	local_bh_disable();
	queue = container_of(work, struct padata_queue, swork);
	pd = queue->pd;

	spin_lock(&queue->serial.lock);
	list_replace_init(&queue->serial.list, &local_list);
	spin_unlock(&queue->serial.lock);

	while (!list_empty(&local_list)) {
		struct padata_priv *padata;

		padata = list_entry(local_list.next,
				    struct padata_priv, list);

		list_del_init(&padata->list);

		padata->serial(padata);
		atomic_dec(&pd->refcnt);
	}
	local_bh_enable();
}

/*
 * padata_do_serial - padata serialization function
 *
 * @padata: object to be serialized.
 *
 * padata_do_serial must be called for every parallelized object.
 * The serialization callback function will run with BHs off.
 */
void padata_do_serial(struct padata_priv *padata)
{
	int cpu;
	struct padata_queue *queue;
	struct parallel_data *pd;

	pd = padata->pd;

	cpu = get_cpu();
	queue = per_cpu_ptr(pd->queue, cpu);

	spin_lock(&queue->reorder.lock);
	atomic_inc(&pd->reorder_objects);
	list_add_tail(&padata->list, &queue->reorder.list);
	spin_unlock(&queue->reorder.lock);

	put_cpu();

	padata_reorder(pd);
}
EXPORT_SYMBOL(padata_do_serial);

static struct parallel_data *padata_alloc_pd(struct padata_instance *pinst,
					     const struct cpumask *cpumask)
{
	int cpu, cpu_index, num_cpus;
	struct padata_queue *queue;
	struct parallel_data *pd;

	cpu_index = 0;

	pd = kzalloc(sizeof(struct parallel_data), GFP_KERNEL);
	if (!pd)
		goto err;

	pd->queue = alloc_percpu(struct padata_queue);
	if (!pd->queue)
		goto err_free_pd;

	if (!alloc_cpumask_var(&pd->cpumask, GFP_KERNEL))
		goto err_free_queue;

	cpumask_and(pd->cpumask, cpumask, cpu_active_mask);

	for_each_cpu(cpu, pd->cpumask) {
		queue = per_cpu_ptr(pd->queue, cpu);

		queue->pd = pd;

		queue->cpu_index = cpu_index;
		cpu_index++;

		INIT_LIST_HEAD(&queue->reorder.list);
		INIT_LIST_HEAD(&queue->parallel.list);
		INIT_LIST_HEAD(&queue->serial.list);
		spin_lock_init(&queue->reorder.lock);
		spin_lock_init(&queue->parallel.lock);
		spin_lock_init(&queue->serial.lock);

		INIT_WORK(&queue->pwork, padata_parallel_worker);
		INIT_WORK(&queue->swork, padata_serial_worker);
		atomic_set(&queue->num_obj, 0);
	}

	num_cpus = cpumask_weight(pd->cpumask);
	pd->max_seq_nr = (MAX_SEQ_NR / num_cpus) * num_cpus - 1;

	setup_timer(&pd->timer, padata_reorder_timer, (unsigned long)pd);
	atomic_set(&pd->seq_nr, -1);
	atomic_set(&pd->reorder_objects, 0);
	atomic_set(&pd->refcnt, 0);
	pd->pinst = pinst;
	spin_lock_init(&pd->lock);

	return pd;

err_free_queue:
	free_percpu(pd->queue);
err_free_pd:
	kfree(pd);
err:
	return NULL;
}

static void padata_free_pd(struct parallel_data *pd)
{
	free_cpumask_var(pd->cpumask);
	free_percpu(pd->queue);
	kfree(pd);
}

static void padata_flush_queues(struct parallel_data *pd)
{
	int cpu;
	struct padata_queue *queue;

	for_each_cpu(cpu, pd->cpumask) {
		queue = per_cpu_ptr(pd->queue, cpu);
		flush_work(&queue->pwork);
	}

	del_timer_sync(&pd->timer);

	if (atomic_read(&pd->reorder_objects))
		padata_reorder(pd);

	for_each_cpu(cpu, pd->cpumask) {
		queue = per_cpu_ptr(pd->queue, cpu);
		flush_work(&queue->swork);
	}

	BUG_ON(atomic_read(&pd->refcnt) != 0);
}

static void padata_replace(struct padata_instance *pinst,
			   struct parallel_data *pd_new)
{
	struct parallel_data *pd_old = pinst->pd;

	pinst->flags |= PADATA_RESET;

	rcu_assign_pointer(pinst->pd, pd_new);

	synchronize_rcu();

	padata_flush_queues(pd_old);
	padata_free_pd(pd_old);

	pinst->flags &= ~PADATA_RESET;
}

/*
 * padata_set_cpumask - set the cpumask that padata should use
 *
 * @pinst: padata instance
 * @cpumask: the cpumask to use
 */
int padata_set_cpumask(struct padata_instance *pinst,
			cpumask_var_t cpumask)
{
	struct parallel_data *pd;
	int err = 0;

	mutex_lock(&pinst->lock);

	get_online_cpus();

	pd = padata_alloc_pd(pinst, cpumask);
	if (!pd) {
		err = -ENOMEM;
		goto out;
	}

	cpumask_copy(pinst->cpumask, cpumask);

	padata_replace(pinst, pd);

out:
	put_online_cpus();

	mutex_unlock(&pinst->lock);

	return err;
}
EXPORT_SYMBOL(padata_set_cpumask);

static int __padata_add_cpu(struct padata_instance *pinst, int cpu)
{
	struct parallel_data *pd;

	if (cpumask_test_cpu(cpu, cpu_active_mask)) {
		pd = padata_alloc_pd(pinst, pinst->cpumask);
		if (!pd)
			return -ENOMEM;

		padata_replace(pinst, pd);
	}

	return 0;
}

/*
 * padata_add_cpu - add a cpu to the padata cpumask
 *
 * @pinst: padata instance
 * @cpu: cpu to add
 */
int padata_add_cpu(struct padata_instance *pinst, int cpu)
{
	int err;

	mutex_lock(&pinst->lock);

	get_online_cpus();
	cpumask_set_cpu(cpu, pinst->cpumask);
	err = __padata_add_cpu(pinst, cpu);
	put_online_cpus();

	mutex_unlock(&pinst->lock);

	return err;
}
EXPORT_SYMBOL(padata_add_cpu);

static int __padata_remove_cpu(struct padata_instance *pinst, int cpu)
{
	struct parallel_data *pd;

	if (cpumask_test_cpu(cpu, cpu_online_mask)) {
		pd = padata_alloc_pd(pinst, pinst->cpumask);
		if (!pd)
			return -ENOMEM;

		padata_replace(pinst, pd);
	}

	return 0;
}

/*
 * padata_remove_cpu - remove a cpu from the padata cpumask
 *
 * @pinst: padata instance
 * @cpu: cpu to remove
 */
int padata_remove_cpu(struct padata_instance *pinst, int cpu)
{
	int err;

	mutex_lock(&pinst->lock);

	get_online_cpus();
	cpumask_clear_cpu(cpu, pinst->cpumask);
	err = __padata_remove_cpu(pinst, cpu);
	put_online_cpus();

	mutex_unlock(&pinst->lock);

	return err;
}
EXPORT_SYMBOL(padata_remove_cpu);

/*
 * padata_start - start the parallel processing
 *
 * @pinst: padata instance to start
 */
void padata_start(struct padata_instance *pinst)
{
	mutex_lock(&pinst->lock);
	pinst->flags |= PADATA_INIT;
	mutex_unlock(&pinst->lock);
}
EXPORT_SYMBOL(padata_start);

/*
 * padata_stop - stop the parallel processing
 *
 * @pinst: padata instance to stop
 */
void padata_stop(struct padata_instance *pinst)
{
	mutex_lock(&pinst->lock);
	pinst->flags &= ~PADATA_INIT;
	mutex_unlock(&pinst->lock);
}
EXPORT_SYMBOL(padata_stop);

#ifdef CONFIG_HOTPLUG_CPU
static int padata_cpu_callback(struct notifier_block *nfb,
			       unsigned long action, void *hcpu)
{
	int err;
	struct padata_instance *pinst;
	int cpu = (unsigned long)hcpu;

	pinst = container_of(nfb, struct padata_instance, cpu_notifier);

	switch (action) {
	case CPU_ONLINE:
	case CPU_ONLINE_FROZEN:
		if (!cpumask_test_cpu(cpu, pinst->cpumask))
			break;
		mutex_lock(&pinst->lock);
		err = __padata_add_cpu(pinst, cpu);
		mutex_unlock(&pinst->lock);
		if (err)
			return NOTIFY_BAD;
		break;

	case CPU_DOWN_PREPARE:
	case CPU_DOWN_PREPARE_FROZEN:
		if (!cpumask_test_cpu(cpu, pinst->cpumask))
			break;
		mutex_lock(&pinst->lock);
		err = __padata_remove_cpu(pinst, cpu);
		mutex_unlock(&pinst->lock);
		if (err)
			return NOTIFY_BAD;
		break;

	case CPU_UP_CANCELED:
	case CPU_UP_CANCELED_FROZEN:
		if (!cpumask_test_cpu(cpu, pinst->cpumask))
			break;
		mutex_lock(&pinst->lock);
		__padata_remove_cpu(pinst, cpu);
		mutex_unlock(&pinst->lock);

	case CPU_DOWN_FAILED:
	case CPU_DOWN_FAILED_FROZEN:
		if (!cpumask_test_cpu(cpu, pinst->cpumask))
			break;
		mutex_lock(&pinst->lock);
		__padata_add_cpu(pinst, cpu);
		mutex_unlock(&pinst->lock);
	}

	return NOTIFY_OK;
}
#endif

/*
 * padata_alloc - allocate and initialize a padata instance
 *
 * @cpumask: cpumask that padata uses for parallelization
 * @wq: workqueue to use for the allocated padata instance
 */
struct padata_instance *padata_alloc(const struct cpumask *cpumask,
				     struct workqueue_struct *wq)
{
	struct padata_instance *pinst;
	struct parallel_data *pd;

	pinst = kzalloc(sizeof(struct padata_instance), GFP_KERNEL);
	if (!pinst)
		goto err;

	get_online_cpus();

	pd = padata_alloc_pd(pinst, cpumask);
	if (!pd)
		goto err_free_inst;

	if (!alloc_cpumask_var(&pinst->cpumask, GFP_KERNEL))
		goto err_free_pd;

	rcu_assign_pointer(pinst->pd, pd);

	pinst->wq = wq;

	cpumask_copy(pinst->cpumask, cpumask);

	pinst->flags = 0;

#ifdef CONFIG_HOTPLUG_CPU
	pinst->cpu_notifier.notifier_call = padata_cpu_callback;
	pinst->cpu_notifier.priority = 0;
	register_hotcpu_notifier(&pinst->cpu_notifier);
#endif

	put_online_cpus();

	mutex_init(&pinst->lock);

	return pinst;

err_free_pd:
	padata_free_pd(pd);
err_free_inst:
	kfree(pinst);
	put_online_cpus();
err:
	return NULL;
}
EXPORT_SYMBOL(padata_alloc);

/*
 * padata_free - free a padata instance
 *
 * @ padata_inst: padata instance to free
 */
void padata_free(struct padata_instance *pinst)
{
	padata_stop(pinst);

	synchronize_rcu();

#ifdef CONFIG_HOTPLUG_CPU
	unregister_hotcpu_notifier(&pinst->cpu_notifier);
#endif
	padata_flush_queues(pinst->pd);
	padata_free_pd(pinst->pd);
	free_cpumask_var(pinst->cpumask);
	kfree(pinst);
}
EXPORT_SYMBOL(padata_free);
Commit	Line	Data
16295bec SK	1	/*
	2	* padata.c - generic interface to process data streams in parallel
	3	*
	4	* Copyright (C) 2008, 2009 secunet Security Networks AG
	5	* Copyright (C) 2008, 2009 Steffen Klassert <steffen.klassert@secunet.com>
	6	*
	7	* This program is free software; you can redistribute it and/or modify it
	8	* under the terms and conditions of the GNU General Public License,
	9	* version 2, as published by the Free Software Foundation.
	10	*
	11	* This program is distributed in the hope it will be useful, but WITHOUT
	12	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	13	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
	14	* more details.
	15	*
	16	* You should have received a copy of the GNU General Public License along with
	17	* this program; if not, write to the Free Software Foundation, Inc.,
	18	* 51 Franklin St - Fifth Floor, Boston, MA 02110-1301 USA.
	19	*/
	20
	21	#include <linux/module.h>
	22	#include <linux/cpumask.h>
	23	#include <linux/err.h>
	24	#include <linux/cpu.h>
	25	#include <linux/padata.h>
	26	#include <linux/mutex.h>
	27	#include <linux/sched.h>
5a0e3ad6	28	#include <linux/slab.h>
16295bec SK	29	#include <linux/rcupdate.h>
	30
	31	#define MAX_SEQ_NR INT_MAX - NR_CPUS
97e3d94a	32	#define MAX_OBJ_NUM 1000
16295bec SK	33
	34	static int padata_index_to_cpu(struct parallel_data *pd, int cpu_index)
	35	{
	36	int cpu, target_cpu;
	37
	38	target_cpu = cpumask_first(pd->cpumask);
	39	for (cpu = 0; cpu < cpu_index; cpu++)
	40	target_cpu = cpumask_next(target_cpu, pd->cpumask);
	41
	42	return target_cpu;
	43	}
	44
	45	static int padata_cpu_hash(struct padata_priv *padata)
	46	{
	47	int cpu_index;
	48	struct parallel_data *pd;
	49
	50	pd = padata->pd;
	51
	52	/*
	53	* Hash the sequence numbers to the cpus by taking
	54	* seq_nr mod. number of cpus in use.
	55	*/
	56	cpu_index = padata->seq_nr % cpumask_weight(pd->cpumask);
	57
	58	return padata_index_to_cpu(pd, cpu_index);
	59	}
	60
	61	static void padata_parallel_worker(struct work_struct *work)
	62	{
	63	struct padata_queue *queue;
	64	struct parallel_data *pd;
	65	struct padata_instance *pinst;
	66	LIST_HEAD(local_list);
	67
	68	local_bh_disable();
	69	queue = container_of(work, struct padata_queue, pwork);
	70	pd = queue->pd;
	71	pinst = pd->pinst;
	72
	73	spin_lock(&queue->parallel.lock);
	74	list_replace_init(&queue->parallel.list, &local_list);
	75	spin_unlock(&queue->parallel.lock);
	76
	77	while (!list_empty(&local_list)) {
	78	struct padata_priv *padata;
	79
	80	padata = list_entry(local_list.next,
	81	struct padata_priv, list);
	82
	83	list_del_init(&padata->list);
	84
	85	padata->parallel(padata);
	86	}
	87
	88	local_bh_enable();
	89	}
	90
	91	/*
	92	* padata_do_parallel - padata parallelization function
	93	*
	94	* @pinst: padata instance
	95	* @padata: object to be parallelized
	96	* @cb_cpu: cpu the serialization callback function will run on,
97	* must be in the cpumask of padata.
98	*
99	* The parallelization callback function will run with BHs off.
100	* Note: Every object which is parallelized by padata_do_parallel
101	* must be seen by padata_do_serial.
102	*/
103	int padata_do_parallel(struct padata_instance *pinst,
104	struct padata_priv *padata, int cb_cpu)
105	{
106	int target_cpu, err;
107	struct padata_queue *queue;
108	struct parallel_data *pd;
109
110	rcu_read_lock_bh();
111
112	pd = rcu_dereference(pinst->pd);
113
114	err = 0;
115	if (!(pinst->flags & PADATA_INIT))
116	goto out;
117
118	err = -EBUSY;
119	if ((pinst->flags & PADATA_RESET))
120	goto out;
121
122	if (atomic_read(&pd->refcnt) >= MAX_OBJ_NUM)
123	goto out;
124
125	err = -EINVAL;
126	if (!cpumask_test_cpu(cb_cpu, pd->cpumask))
127	goto out;
128
129	err = -EINPROGRESS;
130	atomic_inc(&pd->refcnt);
131	padata->pd = pd;
132	padata->cb_cpu = cb_cpu;
133
134	if (unlikely(atomic_read(&pd->seq_nr) == pd->max_seq_nr))
135	atomic_set(&pd->seq_nr, -1);
136
137	padata->seq_nr = atomic_inc_return(&pd->seq_nr);
138
139	target_cpu = padata_cpu_hash(padata);
140	queue = per_cpu_ptr(pd->queue, target_cpu);
141
142	spin_lock(&queue->parallel.lock);
143	list_add_tail(&padata->list, &queue->parallel.list);
144	spin_unlock(&queue->parallel.lock);
145
146	queue_work_on(target_cpu, pinst->wq, &queue->pwork);
147
148	out:
149	rcu_read_unlock_bh();
150
151	return err;
152	}
153	EXPORT_SYMBOL(padata_do_parallel);
154
155	static struct padata_priv padata_get_next(struct parallel_data pd)
156	{
157	int cpu, num_cpus, empty, calc_seq_nr;
158	int seq_nr, next_nr, overrun, next_overrun;
159	struct padata_queue queue, next_queue;
160	struct padata_priv *padata;
161	struct padata_list *reorder;
162
163	empty = 0;
164	next_nr = -1;
165	next_overrun = 0;
166	next_queue = NULL;
167
168	num_cpus = cpumask_weight(pd->cpumask);
169
170	for_each_cpu(cpu, pd->cpumask) {
171	queue = per_cpu_ptr(pd->queue, cpu);
172	reorder = &queue->reorder;
173
174	/*
175	* Calculate the seq_nr of the object that should be
176	* next in this queue.
177	*/
178	overrun = 0;
179	calc_seq_nr = (atomic_read(&queue->num_obj) * num_cpus)
180	+ queue->cpu_index;
181
182	if (unlikely(calc_seq_nr > pd->max_seq_nr)) {
183	calc_seq_nr = calc_seq_nr - pd->max_seq_nr - 1;
184	overrun = 1;
185	}
186
187	if (!list_empty(&reorder->list)) {
188	padata = list_entry(reorder->list.next,
189	struct padata_priv, list);
190
191	seq_nr = padata->seq_nr;
192	BUG_ON(calc_seq_nr != seq_nr);
193	} else {
194	seq_nr = calc_seq_nr;
195	empty++;
196	}
197
198	if (next_nr < 0 \|\| seq_nr < next_nr
199	\|\| (next_overrun && !overrun)) {
200	next_nr = seq_nr;
201	next_overrun = overrun;
202	next_queue = queue;
203	}
204	}
205
206	padata = NULL;
207
208	if (empty == num_cpus)
209	goto out;
210
211	reorder = &next_queue->reorder;
212
213	if (!list_empty(&reorder->list)) {
214	padata = list_entry(reorder->list.next,
215	struct padata_priv, list);
216
217	if (unlikely(next_overrun)) {
218	for_each_cpu(cpu, pd->cpumask) {
219	queue = per_cpu_ptr(pd->queue, cpu);
220	atomic_set(&queue->num_obj, 0);
221	}
222	}
223
224	spin_lock(&reorder->lock);
225	list_del_init(&padata->list);
226	atomic_dec(&pd->reorder_objects);
227	spin_unlock(&reorder->lock);
228
229	atomic_inc(&next_queue->num_obj);
230
231	goto out;
232	}
233
d46a5ac7 SK	234	queue = per_cpu_ptr(pd->queue, smp_processor_id());
d46a5ac7 SK	235	if (queue->cpu_index == next_queue->cpu_index) {
16295bec SK	236	padata = ERR_PTR(-ENODATA);
	237	goto out;
	238	}
	239
	240	padata = ERR_PTR(-EINPROGRESS);
	241	out:
	242	return padata;
	243	}
	244
	245	static void padata_reorder(struct parallel_data *pd)
	246	{
	247	struct padata_priv *padata;
	248	struct padata_queue *queue;
	249	struct padata_instance *pinst = pd->pinst;
	250
16295bec	251	if (!spin_trylock_bh(&pd->lock))
d46a5ac7	252	return;
16295bec SK	253
	254	while (1) {
	255	padata = padata_get_next(pd);
	256
	257	if (!padata \|\| PTR_ERR(padata) == -EINPROGRESS)
	258	break;
	259
	260	if (PTR_ERR(padata) == -ENODATA) {
d46a5ac7	261	del_timer(&pd->timer);
16295bec	262	spin_unlock_bh(&pd->lock);
d46a5ac7	263	return;
16295bec SK	264	}
	265
	266	queue = per_cpu_ptr(pd->queue, padata->cb_cpu);
	267
	268	spin_lock(&queue->serial.lock);
	269	list_add_tail(&padata->list, &queue->serial.list);
	270	spin_unlock(&queue->serial.lock);
	271
	272	queue_work_on(padata->cb_cpu, pinst->wq, &queue->swork);
	273	}
	274
	275	spin_unlock_bh(&pd->lock);
	276
d46a5ac7 SK	277	if (atomic_read(&pd->reorder_objects)
	278	&& !(pinst->flags & PADATA_RESET))
	279	mod_timer(&pd->timer, jiffies + HZ);
	280	else
	281	del_timer(&pd->timer);
16295bec	282
16295bec SK	283	return;
	284	}
	285
d46a5ac7 SK	286	static void padata_reorder_timer(unsigned long arg)
	287	{
	288	struct parallel_data pd = (struct parallel_data )arg;
	289
	290	padata_reorder(pd);
	291	}
	292
16295bec SK	293	static void padata_serial_worker(struct work_struct *work)
	294	{
	295	struct padata_queue *queue;
	296	struct parallel_data *pd;
	297	LIST_HEAD(local_list);
	298
	299	local_bh_disable();
	300	queue = container_of(work, struct padata_queue, swork);
	301	pd = queue->pd;
	302
	303	spin_lock(&queue->serial.lock);
	304	list_replace_init(&queue->serial.list, &local_list);
	305	spin_unlock(&queue->serial.lock);
	306
	307	while (!list_empty(&local_list)) {
	308	struct padata_priv *padata;
	309
	310	padata = list_entry(local_list.next,
	311	struct padata_priv, list);
	312
	313	list_del_init(&padata->list);
	314
	315	padata->serial(padata);
	316	atomic_dec(&pd->refcnt);
	317	}
	318	local_bh_enable();
	319	}
	320
	321	/*
	322	* padata_do_serial - padata serialization function
	323	*
	324	* @padata: object to be serialized.
	325	*
	326	* padata_do_serial must be called for every parallelized object.
	327	* The serialization callback function will run with BHs off.
	328	*/
	329	void padata_do_serial(struct padata_priv *padata)
	330	{
	331	int cpu;
	332	struct padata_queue *queue;
	333	struct parallel_data *pd;
	334
	335	pd = padata->pd;
	336
	337	cpu = get_cpu();
	338	queue = per_cpu_ptr(pd->queue, cpu);
	339
	340	spin_lock(&queue->reorder.lock);
	341	atomic_inc(&pd->reorder_objects);
	342	list_add_tail(&padata->list, &queue->reorder.list);
	343	spin_unlock(&queue->reorder.lock);
	344
	345	put_cpu();
	346
	347	padata_reorder(pd);
	348	}
	349	EXPORT_SYMBOL(padata_do_serial);
	350
	351	static struct parallel_data padata_alloc_pd(struct padata_instance pinst,
	352	const struct cpumask *cpumask)
	353	{
	354	int cpu, cpu_index, num_cpus;
	355	struct padata_queue *queue;
	356	struct parallel_data *pd;
357
358	cpu_index = 0;
359
360	pd = kzalloc(sizeof(struct parallel_data), GFP_KERNEL);
361	if (!pd)
362	goto err;
363
364	pd->queue = alloc_percpu(struct padata_queue);
365	if (!pd->queue)
366	goto err_free_pd;
367
368	if (!alloc_cpumask_var(&pd->cpumask, GFP_KERNEL))
369	goto err_free_queue;
370
7b389b2c SK	371	cpumask_and(pd->cpumask, cpumask, cpu_active_mask);
	372
	373	for_each_cpu(cpu, pd->cpumask) {
16295bec SK	374	queue = per_cpu_ptr(pd->queue, cpu);
	375
	376	queue->pd = pd;
	377
7b389b2c SK	378	queue->cpu_index = cpu_index;
7b389b2c SK	379	cpu_index++;
16295bec SK	380
	381	INIT_LIST_HEAD(&queue->reorder.list);
	382	INIT_LIST_HEAD(&queue->parallel.list);
	383	INIT_LIST_HEAD(&queue->serial.list);
	384	spin_lock_init(&queue->reorder.lock);
	385	spin_lock_init(&queue->parallel.lock);
	386	spin_lock_init(&queue->serial.lock);
	387
	388	INIT_WORK(&queue->pwork, padata_parallel_worker);
	389	INIT_WORK(&queue->swork, padata_serial_worker);
	390	atomic_set(&queue->num_obj, 0);
	391	}
	392
16295bec SK	393	num_cpus = cpumask_weight(pd->cpumask);
	394	pd->max_seq_nr = (MAX_SEQ_NR / num_cpus) * num_cpus - 1;
	395
d46a5ac7	396	setup_timer(&pd->timer, padata_reorder_timer, (unsigned long)pd);
16295bec SK	397	atomic_set(&pd->seq_nr, -1);
	398	atomic_set(&pd->reorder_objects, 0);
	399	atomic_set(&pd->refcnt, 0);
	400	pd->pinst = pinst;
	401	spin_lock_init(&pd->lock);
	402
	403	return pd;
	404
	405	err_free_queue:
	406	free_percpu(pd->queue);
	407	err_free_pd:
	408	kfree(pd);
	409	err:
	410	return NULL;
	411	}
	412
	413	static void padata_free_pd(struct parallel_data *pd)
	414	{
	415	free_cpumask_var(pd->cpumask);
	416	free_percpu(pd->queue);
	417	kfree(pd);
	418	}
	419
2b73b07a SK	420	static void padata_flush_queues(struct parallel_data *pd)
	421	{
	422	int cpu;
	423	struct padata_queue *queue;
	424
	425	for_each_cpu(cpu, pd->cpumask) {
	426	queue = per_cpu_ptr(pd->queue, cpu);
	427	flush_work(&queue->pwork);
	428	}
	429
	430	del_timer_sync(&pd->timer);
	431
	432	if (atomic_read(&pd->reorder_objects))
	433	padata_reorder(pd);
	434
	435	for_each_cpu(cpu, pd->cpumask) {
	436	queue = per_cpu_ptr(pd->queue, cpu);
	437	flush_work(&queue->swork);
	438	}
	439
	440	BUG_ON(atomic_read(&pd->refcnt) != 0);
	441	}
	442
16295bec SK	443	static void padata_replace(struct padata_instance *pinst,
	444	struct parallel_data *pd_new)
	445	{
	446	struct parallel_data *pd_old = pinst->pd;
	447
	448	pinst->flags \|= PADATA_RESET;
	449
	450	rcu_assign_pointer(pinst->pd, pd_new);
	451
	452	synchronize_rcu();
	453
2b73b07a	454	padata_flush_queues(pd_old);
16295bec SK	455	padata_free_pd(pd_old);
	456
	457	pinst->flags &= ~PADATA_RESET;
	458	}
	459
	460	/*
	461	* padata_set_cpumask - set the cpumask that padata should use
	462	*
	463	* @pinst: padata instance
	464	* @cpumask: the cpumask to use
	465	*/
	466	int padata_set_cpumask(struct padata_instance *pinst,
	467	cpumask_var_t cpumask)
	468	{
	469	struct parallel_data *pd;
	470	int err = 0;
	471
16295bec SK	472	mutex_lock(&pinst->lock);
16295bec SK	473
6751fb3c SK	474	get_online_cpus();
6751fb3c SK	475
16295bec SK	476	pd = padata_alloc_pd(pinst, cpumask);
	477	if (!pd) {
	478	err = -ENOMEM;
	479	goto out;
	480	}
	481
	482	cpumask_copy(pinst->cpumask, cpumask);
	483
	484	padata_replace(pinst, pd);
	485
	486	out:
6751fb3c SK	487	put_online_cpus();
6751fb3c SK	488
16295bec SK	489	mutex_unlock(&pinst->lock);
	490
	491	return err;
	492	}
	493	EXPORT_SYMBOL(padata_set_cpumask);
	494
	495	static int __padata_add_cpu(struct padata_instance *pinst, int cpu)
	496	{
	497	struct parallel_data *pd;
	498
	499	if (cpumask_test_cpu(cpu, cpu_active_mask)) {
	500	pd = padata_alloc_pd(pinst, pinst->cpumask);
	501	if (!pd)
	502	return -ENOMEM;
	503
	504	padata_replace(pinst, pd);
	505	}
	506
	507	return 0;
	508	}
	509
	510	/*
	511	* padata_add_cpu - add a cpu to the padata cpumask
	512	*
	513	* @pinst: padata instance
	514	* @cpu: cpu to add
	515	*/
	516	int padata_add_cpu(struct padata_instance *pinst, int cpu)
	517	{
	518	int err;
	519
16295bec SK	520	mutex_lock(&pinst->lock);
16295bec SK	521
6751fb3c	522	get_online_cpus();
16295bec SK	523	cpumask_set_cpu(cpu, pinst->cpumask);
16295bec SK	524	err = __padata_add_cpu(pinst, cpu);
6751fb3c	525	put_online_cpus();
16295bec SK	526
	527	mutex_unlock(&pinst->lock);
	528
	529	return err;
	530	}
	531	EXPORT_SYMBOL(padata_add_cpu);
	532
	533	static int __padata_remove_cpu(struct padata_instance *pinst, int cpu)
	534	{
	535	struct parallel_data *pd;
	536
	537	if (cpumask_test_cpu(cpu, cpu_online_mask)) {
	538	pd = padata_alloc_pd(pinst, pinst->cpumask);
	539	if (!pd)
	540	return -ENOMEM;
	541
	542	padata_replace(pinst, pd);
	543	}
	544
	545	return 0;
	546	}
	547
	548	/*
	549	* padata_remove_cpu - remove a cpu from the padata cpumask
	550	*
	551	* @pinst: padata instance
	552	* @cpu: cpu to remove
	553	*/
	554	int padata_remove_cpu(struct padata_instance *pinst, int cpu)
	555	{
	556	int err;
	557
16295bec SK	558	mutex_lock(&pinst->lock);
16295bec SK	559
6751fb3c	560	get_online_cpus();
16295bec SK	561	cpumask_clear_cpu(cpu, pinst->cpumask);
16295bec SK	562	err = __padata_remove_cpu(pinst, cpu);
6751fb3c	563	put_online_cpus();
16295bec SK	564
	565	mutex_unlock(&pinst->lock);
	566
	567	return err;
	568	}
	569	EXPORT_SYMBOL(padata_remove_cpu);
	570
	571	/*
	572	* padata_start - start the parallel processing
	573	*
	574	* @pinst: padata instance to start
	575	*/
	576	void padata_start(struct padata_instance *pinst)
	577	{
16295bec SK	578	mutex_lock(&pinst->lock);
	579	pinst->flags \|= PADATA_INIT;
	580	mutex_unlock(&pinst->lock);
	581	}
	582	EXPORT_SYMBOL(padata_start);
	583
	584	/*
	585	* padata_stop - stop the parallel processing
	586	*
	587	* @pinst: padata instance to stop
	588	*/
	589	void padata_stop(struct padata_instance *pinst)
	590	{
16295bec SK	591	mutex_lock(&pinst->lock);
	592	pinst->flags &= ~PADATA_INIT;
	593	mutex_unlock(&pinst->lock);
	594	}
	595	EXPORT_SYMBOL(padata_stop);
	596
e2cb2f1c	597	#ifdef CONFIG_HOTPLUG_CPU
975d2603 HK	598	static int padata_cpu_callback(struct notifier_block *nfb,
975d2603 HK	599	unsigned long action, void *hcpu)
16295bec SK	600	{
	601	int err;
	602	struct padata_instance *pinst;
	603	int cpu = (unsigned long)hcpu;
	604
	605	pinst = container_of(nfb, struct padata_instance, cpu_notifier);
	606
	607	switch (action) {
	608	case CPU_ONLINE:
	609	case CPU_ONLINE_FROZEN:
	610	if (!cpumask_test_cpu(cpu, pinst->cpumask))
	611	break;
	612	mutex_lock(&pinst->lock);
	613	err = __padata_add_cpu(pinst, cpu);
	614	mutex_unlock(&pinst->lock);
	615	if (err)
	616	return NOTIFY_BAD;
	617	break;
	618
	619	case CPU_DOWN_PREPARE:
	620	case CPU_DOWN_PREPARE_FROZEN:
	621	if (!cpumask_test_cpu(cpu, pinst->cpumask))
	622	break;
	623	mutex_lock(&pinst->lock);
	624	err = __padata_remove_cpu(pinst, cpu);
	625	mutex_unlock(&pinst->lock);
	626	if (err)
	627	return NOTIFY_BAD;
	628	break;
	629
	630	case CPU_UP_CANCELED:
	631	case CPU_UP_CANCELED_FROZEN:
	632	if (!cpumask_test_cpu(cpu, pinst->cpumask))
	633	break;
	634	mutex_lock(&pinst->lock);
	635	__padata_remove_cpu(pinst, cpu);
	636	mutex_unlock(&pinst->lock);
	637
	638	case CPU_DOWN_FAILED:
	639	case CPU_DOWN_FAILED_FROZEN:
	640	if (!cpumask_test_cpu(cpu, pinst->cpumask))
	641	break;
	642	mutex_lock(&pinst->lock);
	643	__padata_add_cpu(pinst, cpu);
	644	mutex_unlock(&pinst->lock);
	645	}
	646
	647	return NOTIFY_OK;
	648	}
e2cb2f1c	649	#endif
16295bec SK	650
	651	/*
	652	* padata_alloc - allocate and initialize a padata instance
	653	*
	654	* @cpumask: cpumask that padata uses for parallelization
	655	* @wq: workqueue to use for the allocated padata instance
	656	*/
	657	struct padata_instance padata_alloc(const struct cpumask cpumask,
	658	struct workqueue_struct *wq)
	659	{
16295bec SK	660	struct padata_instance *pinst;
	661	struct parallel_data *pd;
	662
	663	pinst = kzalloc(sizeof(struct padata_instance), GFP_KERNEL);
	664	if (!pinst)
	665	goto err;
	666
6751fb3c SK	667	get_online_cpus();
6751fb3c SK	668
16295bec SK	669	pd = padata_alloc_pd(pinst, cpumask);
	670	if (!pd)
	671	goto err_free_inst;
	672
74781387 SK	673	if (!alloc_cpumask_var(&pinst->cpumask, GFP_KERNEL))
	674	goto err_free_pd;
	675
16295bec SK	676	rcu_assign_pointer(pinst->pd, pd);
	677
	678	pinst->wq = wq;
	679
	680	cpumask_copy(pinst->cpumask, cpumask);
	681
	682	pinst->flags = 0;
	683
e2cb2f1c	684	#ifdef CONFIG_HOTPLUG_CPU
16295bec SK	685	pinst->cpu_notifier.notifier_call = padata_cpu_callback;
16295bec SK	686	pinst->cpu_notifier.priority = 0;
e2cb2f1c SK	687	register_hotcpu_notifier(&pinst->cpu_notifier);
e2cb2f1c SK	688	#endif
16295bec	689
6751fb3c SK	690	put_online_cpus();
6751fb3c SK	691
16295bec SK	692	mutex_init(&pinst->lock);
	693
	694	return pinst;
	695
	696	err_free_pd:
	697	padata_free_pd(pd);
	698	err_free_inst:
	699	kfree(pinst);
6751fb3c	700	put_online_cpus();
16295bec SK	701	err:
	702	return NULL;
	703	}
	704	EXPORT_SYMBOL(padata_alloc);
	705
	706	/*
	707	* padata_free - free a padata instance
	708	*
	709	* @ padata_inst: padata instance to free
	710	*/
	711	void padata_free(struct padata_instance *pinst)
	712	{
	713	padata_stop(pinst);
	714
	715	synchronize_rcu();
	716
e2cb2f1c	717	#ifdef CONFIG_HOTPLUG_CPU
16295bec	718	unregister_hotcpu_notifier(&pinst->cpu_notifier);
e2cb2f1c	719	#endif
2b73b07a	720	padata_flush_queues(pinst->pd);
16295bec	721	padata_free_pd(pinst->pd);
74781387	722	free_cpumask_var(pinst->cpumask);
16295bec SK	723	kfree(pinst);
	724	}
	725	EXPORT_SYMBOL(padata_free);