/* * padata.c - generic interface to process data streams in parallel * * See Documentation/padata.txt for an api documentation. * * 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/export.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/sysfs.h>#include <linux/rcupdate.h> #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.pcpu);	for (cpu = 0; cpu < cpu_index; cpu++)		target_cpu = cpumask_next(target_cpu, pd->cpumask.pcpu); 	return target_cpu;} static int padata_cpu_hash(struct parallel_data *pd){	unsigned int seq_nr;	int cpu_index; 	/*	 * Hash the sequence numbers to the cpus by taking	 * seq_nr mod. number of cpus in use.	 */ 	seq_nr = atomic_inc_return(&pd->seq_nr);	cpu_index = seq_nr % cpumask_weight(pd->cpumask.pcpu); 	return padata_index_to_cpu(pd, cpu_index);} static void padata_parallel_worker(struct work_struct *parallel_work){	struct padata_parallel_queue *pqueue;	struct parallel_data *pd;	struct padata_instance *pinst;	LIST_HEAD(local_list); 	local_bh_disable();	pqueue = container_of(parallel_work,			      struct padata_parallel_queue, work);	pd = pqueue->pd;	pinst = pd->pinst; 	spin_lock(&pqueue->parallel.lock);	list_replace_init(&pqueue->parallel.list, &local_list);	spin_unlock(&pqueue->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 serial cpumask of padata(i.e. cpumask.cbcpu). * * 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_parallel_queue *queue;	struct parallel_data *pd; 	rcu_read_lock_bh(); 	pd = rcu_dereference_bh(pinst->pd); 	err = -EINVAL;	if (!(pinst->flags & PADATA_INIT) || pinst->flags & PADATA_INVALID)		goto out; 	if (!cpumask_test_cpu(cb_cpu, pd->cpumask.cbcpu))		goto out; 	err =  -EBUSY;	if ((pinst->flags & PADATA_RESET))		goto out; 	if (atomic_read(&pd->refcnt) >= MAX_OBJ_NUM)		goto out; 	err = 0;	atomic_inc(&pd->refcnt);	padata->pd = pd;	padata->cb_cpu = cb_cpu; 	target_cpu = padata_cpu_hash(pd);	queue = per_cpu_ptr(pd->pqueue, 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->work); out:	rcu_read_unlock_bh(); 	return err;}EXPORT_SYMBOL(padata_do_parallel); /* * padata_get_next - Get the next object that needs serialization. * * Return values are: * * A pointer to the control struct of the next object that needs * serialization, if present in one of the percpu reorder queues. * * NULL, if all percpu reorder queues are empty. * * -EINPROGRESS, if the next object that needs serialization will *  be parallel processed by another cpu and is not yet present in *  the cpu's reorder queue. * * -ENODATA, if this cpu has to do the parallel processing for *  the next object. */static struct padata_priv *padata_get_next(struct parallel_data *pd){	int cpu, num_cpus;	unsigned int next_nr, next_index;	struct padata_parallel_queue *next_queue;	struct padata_priv *padata;	struct padata_list *reorder; 	num_cpus = cpumask_weight(pd->cpumask.pcpu); 	/*	 * Calculate the percpu reorder queue and the sequence	 * number of the next object.	 */	next_nr = pd->processed;	next_index = next_nr % num_cpus;	cpu = padata_index_to_cpu(pd, next_index);	next_queue = per_cpu_ptr(pd->pqueue, cpu); 	padata = NULL; 	reorder = &next_queue->reorder; 	if (!list_empty(&reorder->list)) {		padata = list_entry(reorder->list.next,				    struct padata_priv, list); 		spin_lock(&reorder->lock);		list_del_init(&padata->list);		atomic_dec(&pd->reorder_objects);		spin_unlock(&reorder->lock); 		pd->processed++; 		goto out;	} 	if (__this_cpu_read(pd->pqueue->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){	int cb_cpu;	struct padata_priv *padata;	struct padata_serial_queue *squeue;	struct padata_instance *pinst = pd->pinst; 	/*	 * We need to ensure that only one cpu can work on dequeueing of	 * the reorder queue the time. Calculating in which percpu reorder	 * queue the next object will arrive takes some time. A spinlock	 * would be highly contended. Also it is not clear in which order	 * the objects arrive to the reorder queues. So a cpu could wait to	 * get the lock just to notice that there is nothing to do at the	 * moment. Therefore we use a trylock and let the holder of the lock	 * care for all the objects enqueued during the holdtime of the lock.	 */	if (!spin_trylock_bh(&pd->lock))		return; 	while (1) {		padata = padata_get_next(pd); 		/*		 * All reorder queues are empty, or the next object that needs		 * serialization is parallel processed by another cpu and is		 * still on it's way to the cpu's reorder queue, nothing to		 * do for now.		 */		if (!padata || PTR_ERR(padata) == -EINPROGRESS)			break; 		/*		 * This cpu has to do the parallel processing of the next		 * object. It's waiting in the cpu's parallelization queue,		 * so exit immediately.		 */		if (PTR_ERR(padata) == -ENODATA) {			del_timer(&pd->timer);			spin_unlock_bh(&pd->lock);			return;		} 		cb_cpu = padata->cb_cpu;		squeue = per_cpu_ptr(pd->squeue, cb_cpu); 		spin_lock(&squeue->serial.lock);		list_add_tail(&padata->list, &squeue->serial.list);		spin_unlock(&squeue->serial.lock); 		queue_work_on(cb_cpu, pinst->wq, &squeue->work);	} 	spin_unlock_bh(&pd->lock); 	/*	 * The next object that needs serialization might have arrived to	 * the reorder queues in the meantime, we will be called again	 * from the timer function if no one else cares for it.	 */	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 *serial_work){	struct padata_serial_queue *squeue;	struct parallel_data *pd;	LIST_HEAD(local_list); 	local_bh_disable();	squeue = container_of(serial_work, struct padata_serial_queue, work);	pd = squeue->pd; 	spin_lock(&squeue->serial.lock);	list_replace_init(&squeue->serial.list, &local_list);	spin_unlock(&squeue->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_parallel_queue *pqueue;	struct parallel_data *pd; 	pd = padata->pd; 	cpu = get_cpu();	pqueue = per_cpu_ptr(pd->pqueue, cpu); 	spin_lock(&pqueue->reorder.lock);	atomic_inc(&pd->reorder_objects);	list_add_tail(&padata->list, &pqueue->reorder.list);	spin_unlock(&pqueue->reorder.lock); 	put_cpu(); 	padata_reorder(pd);}EXPORT_SYMBOL(padata_do_serial); static int padata_setup_cpumasks(struct parallel_data *pd,				 const struct cpumask *pcpumask,				 const struct cpumask *cbcpumask){	if (!alloc_cpumask_var(&pd->cpumask.pcpu, GFP_KERNEL))		return -ENOMEM; 	cpumask_and(pd->cpumask.pcpu, pcpumask, cpu_online_mask);	if (!alloc_cpumask_var(&pd->cpumask.cbcpu, GFP_KERNEL)) {		free_cpumask_var(pd->cpumask.cbcpu);		return -ENOMEM;	} 	cpumask_and(pd->cpumask.cbcpu, cbcpumask, cpu_online_mask);	return 0;} static void __padata_list_init(struct padata_list *pd_list){	INIT_LIST_HEAD(&pd_list->list);	spin_lock_init(&pd_list->lock);} /* Initialize all percpu queues used by serial workers */static void padata_init_squeues(struct parallel_data *pd){	int cpu;	struct padata_serial_queue *squeue; 	for_each_cpu(cpu, pd->cpumask.cbcpu) {		squeue = per_cpu_ptr(pd->squeue, cpu);		squeue->pd = pd;		__padata_list_init(&squeue->serial);		INIT_WORK(&squeue->work, padata_serial_worker);	}} /* Initialize all percpu queues used by parallel workers */static void padata_init_pqueues(struct parallel_data *pd){	int cpu_index, cpu;	struct padata_parallel_queue *pqueue; 	cpu_index = 0;	for_each_cpu(cpu, pd->cpumask.pcpu) {		pqueue = per_cpu_ptr(pd->pqueue, cpu);		pqueue->pd = pd;		pqueue->cpu_index = cpu_index;		cpu_index++; 		__padata_list_init(&pqueue->reorder);		__padata_list_init(&pqueue->parallel);		INIT_WORK(&pqueue->work, padata_parallel_worker);		atomic_set(&pqueue->num_obj, 0);	}} /* Allocate and initialize the internal cpumask dependend resources. */static struct parallel_data *padata_alloc_pd(struct padata_instance *pinst,					     const struct cpumask *pcpumask,					     const struct cpumask *cbcpumask){	struct parallel_data *pd; 	pd = kzalloc(sizeof(struct parallel_data), GFP_KERNEL);	if (!pd)		goto err; 	pd->pqueue = alloc_percpu(struct padata_parallel_queue);	if (!pd->pqueue)		goto err_free_pd; 	pd->squeue = alloc_percpu(struct padata_serial_queue);	if (!pd->squeue)		goto err_free_pqueue;	if (padata_setup_cpumasks(pd, pcpumask, cbcpumask) < 0)		goto err_free_squeue; 	padata_init_pqueues(pd);	padata_init_squeues(pd);	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_squeue:	free_percpu(pd->squeue);err_free_pqueue:	free_percpu(pd->pqueue);err_free_pd:	kfree(pd);err:	return NULL;} static void padata_free_pd(struct parallel_data *pd){	free_cpumask_var(pd->cpumask.pcpu);	free_cpumask_var(pd->cpumask.cbcpu);	free_percpu(pd->pqueue);	free_percpu(pd->squeue);	kfree(pd);} /* Flush all objects out of the padata queues. */static void padata_flush_queues(struct parallel_data *pd){	int cpu;	struct padata_parallel_queue *pqueue;	struct padata_serial_queue *squeue; 	for_each_cpu(cpu, pd->cpumask.pcpu) {		pqueue = per_cpu_ptr(pd->pqueue, cpu);		flush_work(&pqueue->work);	} 	del_timer_sync(&pd->timer); 	if (atomic_read(&pd->reorder_objects))		padata_reorder(pd); 	for_each_cpu(cpu, pd->cpumask.cbcpu) {		squeue = per_cpu_ptr(pd->squeue, cpu);		flush_work(&squeue->work);	} 	BUG_ON(atomic_read(&pd->refcnt) != 0);} static void __padata_start(struct padata_instance *pinst){	pinst->flags |= PADATA_INIT;} static void __padata_stop(struct padata_instance *pinst){	if (!(pinst->flags & PADATA_INIT))		return; 	pinst->flags &= ~PADATA_INIT; 	synchronize_rcu(); 	get_online_cpus();	padata_flush_queues(pinst->pd);	put_online_cpus();}

Lines 1–500 of 1105. Load more