// SPDX-License-Identifier: GPL-2.0/* * 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>#include <linux/module.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){	/*	 * Hash the sequence numbers to the cpus by taking	 * seq_nr mod. number of cpus in use.	 */	int 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;	LIST_HEAD(local_list); 	local_bh_disable();	pqueue = container_of(parallel_work,			      struct padata_parallel_queue, work); 	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: pointer to the CPU that the serialization callback function should *          run on.  If it's not in the serial cpumask of @pinst *          (i.e. cpumask.cbcpu), this function selects a fallback CPU and if *          none found, returns -EINVAL. * * 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 i, cpu, cpu_index, 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)) {		if (!cpumask_weight(pd->cpumask.cbcpu))			goto out; 		/* Select an alternate fallback CPU and notify the caller. */		cpu_index = *cb_cpu % cpumask_weight(pd->cpumask.cbcpu); 		cpu = cpumask_first(pd->cpumask.cbcpu);		for (i = 0; i < cpu_index; i++)			cpu = cpumask_next(cpu, pd->cpumask.cbcpu); 		*cb_cpu = cpu;	} 	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; 	padata->seq_nr = atomic_inc_return(&pd->seq_nr);	target_cpu = padata_cpu_hash(pd, padata->seq_nr);	padata->cpu = target_cpu;	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(pinst->parallel_wq, &queue->work); out:	rcu_read_unlock_bh(); 	return err;}EXPORT_SYMBOL(padata_do_parallel); /* * padata_find_next - Find 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 the next object that needs serialization will *  be parallel processed by another cpu and is not yet present in *  the cpu's reorder queue. */static struct padata_priv *padata_find_next(struct parallel_data *pd,					    bool remove_object){	struct padata_parallel_queue *next_queue;	struct padata_priv *padata;	struct padata_list *reorder;	int cpu = pd->cpu; 	next_queue = per_cpu_ptr(pd->pqueue, cpu);	reorder = &next_queue->reorder; 	spin_lock(&reorder->lock);	if (list_empty(&reorder->list)) {		spin_unlock(&reorder->lock);		return NULL;	} 	padata = list_entry(reorder->list.next, struct padata_priv, list); 	/*	 * Checks the rare case where two or more parallel jobs have hashed to	 * the same CPU and one of the later ones finishes first.	 */	if (padata->seq_nr != pd->processed) {		spin_unlock(&reorder->lock);		return NULL;	} 	if (remove_object) {		list_del_init(&padata->list);		atomic_dec(&pd->reorder_objects);		++pd->processed;		pd->cpu = cpumask_next_wrap(cpu, pd->cpumask.pcpu, -1, false);	} 	spin_unlock(&reorder->lock);	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;	struct padata_parallel_queue *next_queue; 	/*	 * 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_find_next(pd, true); 		/*		 * If 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)			break; 		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->serial_wq, &squeue->work);	} 	spin_unlock_bh(&pd->lock); 	/*	 * The next object that needs serialization might have arrived to	 * the reorder queues in the meantime.	 *	 * Ensure reorder queue is read after pd->lock is dropped so we see	 * new objects from another task in padata_do_serial.  Pairs with	 * smp_mb__after_atomic in padata_do_serial.	 */	smp_mb(); 	next_queue = per_cpu_ptr(pd->pqueue, pd->cpu);	if (!list_empty(&next_queue->reorder.list) &&	    padata_find_next(pd, false))		queue_work(pinst->serial_wq, &pd->reorder_work);} static void invoke_padata_reorder(struct work_struct *work){	struct parallel_data *pd; 	local_bh_disable();	pd = container_of(work, struct parallel_data, reorder_work);	padata_reorder(pd);	local_bh_enable();} 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){	struct parallel_data *pd = padata->pd;	struct padata_parallel_queue *pqueue = per_cpu_ptr(pd->pqueue,							   padata->cpu);	struct padata_priv *cur; 	spin_lock(&pqueue->reorder.lock);	/* Sort in ascending order of sequence number. */	list_for_each_entry_reverse(cur, &pqueue->reorder.list, list)		if (cur->seq_nr < padata->seq_nr)			break;	list_add(&padata->list, &cur->list);	atomic_inc(&pd->reorder_objects);	spin_unlock(&pqueue->reorder.lock); 	/*	 * Ensure the addition to the reorder list is ordered correctly	 * with the trylock of pd->lock in padata_reorder.  Pairs with smp_mb	 * in padata_reorder.	 */	smp_mb__after_atomic(); 	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){	struct workqueue_attrs *attrs;	int err = -ENOMEM; 	if (!alloc_cpumask_var(&pd->cpumask.pcpu, GFP_KERNEL))		goto out;	cpumask_and(pd->cpumask.pcpu, pcpumask, cpu_online_mask); 	if (!alloc_cpumask_var(&pd->cpumask.cbcpu, GFP_KERNEL))		goto free_pcpu_mask;	cpumask_and(pd->cpumask.cbcpu, cbcpumask, cpu_online_mask); 	attrs = alloc_workqueue_attrs();	if (!attrs)		goto free_cbcpu_mask; 	/* Restrict parallel_wq workers to pd->cpumask.pcpu. */	cpumask_copy(attrs->cpumask, pd->cpumask.pcpu);	err = apply_workqueue_attrs(pd->pinst->parallel_wq, attrs);	free_workqueue_attrs(attrs);	if (err < 0)		goto free_cbcpu_mask; 	return 0; free_cbcpu_mask:	free_cpumask_var(pd->cpumask.cbcpu);free_pcpu_mask:	free_cpumask_var(pd->cpumask.pcpu);out:	return err;} 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;	struct padata_parallel_queue *pqueue; 	for_each_cpu(cpu, pd->cpumask.pcpu) {		pqueue = per_cpu_ptr(pd->pqueue, cpu); 		__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; 	pd->pinst = pinst;	if (padata_setup_cpumasks(pd, pcpumask, cbcpumask) < 0)		goto err_free_squeue; 	padata_init_pqueues(pd);	padata_init_squeues(pd);	atomic_set(&pd->seq_nr, -1);	atomic_set(&pd->reorder_objects, 0);	atomic_set(&pd->refcnt, 0);	spin_lock_init(&pd->lock);	pd->cpu = cpumask_first(pd->cpumask.pcpu);	INIT_WORK(&pd->reorder_work, invoke_padata_reorder); 	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);	} 	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;

Lines 1–500 of 1078. Load more