// SPDX-License-Identifier: GPL-2.0/* * padata.c - generic interface to process data streams in parallel * * See Documentation/core-api/padata.rst for more information. * * Copyright (C) 2008, 2009 secunet Security Networks AG * Copyright (C) 2008, 2009 Steffen Klassert <steffen.klassert@secunet.com> * * Copyright (c) 2020 Oracle and/or its affiliates. * Author: Daniel Jordan <daniel.m.jordan@oracle.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/completion.h>#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	PADATA_WORK_ONSTACK	1	/* Work's memory is on stack */ struct padata_work {	struct work_struct	pw_work;	struct list_head	pw_list;  /* padata_free_works linkage */	void			*pw_data;}; static DEFINE_SPINLOCK(padata_works_lock);static struct padata_work *padata_works;static LIST_HEAD(padata_free_works); struct padata_mt_job_state {	spinlock_t		lock;	struct completion	completion;	struct padata_mt_job	*job;	int			nworks;	int			nworks_fini;	unsigned long		chunk_size;}; static void padata_free_pd(struct parallel_data *pd);static void __init padata_mt_helper(struct work_struct *work); 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 struct padata_work *padata_work_alloc(void){	struct padata_work *pw; 	lockdep_assert_held(&padata_works_lock); 	if (list_empty(&padata_free_works))		return NULL;	/* No more work items allowed to be queued. */ 	pw = list_first_entry(&padata_free_works, struct padata_work, pw_list);	list_del(&pw->pw_list);	return pw;} static void padata_work_init(struct padata_work *pw, work_func_t work_fn,			     void *data, int flags){	if (flags & PADATA_WORK_ONSTACK)		INIT_WORK_ONSTACK(&pw->pw_work, work_fn);	else		INIT_WORK(&pw->pw_work, work_fn);	pw->pw_data = data;} static int __init padata_work_alloc_mt(int nworks, void *data,				       struct list_head *head){	int i; 	spin_lock(&padata_works_lock);	/* Start at 1 because the current task participates in the job. */	for (i = 1; i < nworks; ++i) {		struct padata_work *pw = padata_work_alloc(); 		if (!pw)			break;		padata_work_init(pw, padata_mt_helper, data, 0);		list_add(&pw->pw_list, head);	}	spin_unlock(&padata_works_lock); 	return i;} static void padata_work_free(struct padata_work *pw){	lockdep_assert_held(&padata_works_lock);	list_add(&pw->pw_list, &padata_free_works);} static void __init padata_works_free(struct list_head *works){	struct padata_work *cur, *next; 	if (list_empty(works))		return; 	spin_lock(&padata_works_lock);	list_for_each_entry_safe(cur, next, works, pw_list) {		list_del(&cur->pw_list);		padata_work_free(cur);	}	spin_unlock(&padata_works_lock);} static void padata_parallel_worker(struct work_struct *parallel_work){	struct padata_work *pw = container_of(parallel_work, struct padata_work,					      pw_work);	struct padata_priv *padata = pw->pw_data; 	local_bh_disable();	padata->parallel(padata);	spin_lock(&padata_works_lock);	padata_work_free(pw);	spin_unlock(&padata_works_lock);	local_bh_enable();} /** * padata_do_parallel - padata parallelization function * * @ps: padatashell * @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. * * Return: 0 on success or else negative error code. */int padata_do_parallel(struct padata_shell *ps,		       struct padata_priv *padata, int *cb_cpu){	struct padata_instance *pinst = ps->pinst;	int i, cpu, cpu_index, err;	struct parallel_data *pd;	struct padata_work *pw; 	rcu_read_lock_bh(); 	pd = rcu_dereference_bh(ps->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; 	atomic_inc(&pd->refcnt);	padata->pd = pd;	padata->cb_cpu = *cb_cpu; 	rcu_read_unlock_bh(); 	spin_lock(&padata_works_lock);	padata->seq_nr = ++pd->seq_nr;	pw = padata_work_alloc();	spin_unlock(&padata_works_lock);	if (pw) {		padata_work_init(pw, padata_parallel_worker, padata, 0);		queue_work(pinst->parallel_wq, &pw->pw_work);	} else {		/* Maximum works limit exceeded, run in the current task. */		padata->parallel(padata);	} 	return 0;out:	rcu_read_unlock_bh(); 	return err;}EXPORT_SYMBOL(padata_do_parallel); /* * padata_find_next - Find the next object that needs serialization. * * Return: * * 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_priv *padata;	struct padata_list *reorder;	int cpu = pd->cpu; 	reorder = per_cpu_ptr(pd->reorder_list, cpu); 	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);		++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){	struct padata_instance *pinst = pd->ps->pinst;	int cb_cpu;	struct padata_priv *padata;	struct padata_serial_queue *squeue;	struct padata_list *reorder; 	/*	 * 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 in padata_do_serial.	 */	smp_mb(); 	reorder = per_cpu_ptr(pd->reorder_list, pd->cpu);	if (!list_empty(&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);	int cnt; 	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); 	cnt = 0; 	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);		cnt++;	}	local_bh_enable(); 	if (atomic_sub_and_test(cnt, &pd->refcnt))		padata_free_pd(pd);} /** * 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;	int hashed_cpu = padata_cpu_hash(pd, padata->seq_nr);	struct padata_list *reorder = per_cpu_ptr(pd->reorder_list, hashed_cpu);	struct padata_priv *cur; 	spin_lock(&reorder->lock);	/* Sort in ascending order of sequence number. */	list_for_each_entry_reverse(cur, &reorder->list, list)		if (cur->seq_nr < padata->seq_nr)			break;	list_add(&padata->list, &cur->list);	spin_unlock(&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(); 	padata_reorder(pd);}EXPORT_SYMBOL(padata_do_serial); static int padata_setup_cpumasks(struct padata_instance *pinst){	struct workqueue_attrs *attrs;	int err; 	attrs = alloc_workqueue_attrs();	if (!attrs)		return -ENOMEM; 	/* Restrict parallel_wq workers to pd->cpumask.pcpu. */	cpumask_copy(attrs->cpumask, pinst->cpumask.pcpu);	err = apply_workqueue_attrs(pinst->parallel_wq, attrs);	free_workqueue_attrs(attrs); 	return err;} static void __init padata_mt_helper(struct work_struct *w){	struct padata_work *pw = container_of(w, struct padata_work, pw_work);	struct padata_mt_job_state *ps = pw->pw_data;	struct padata_mt_job *job = ps->job;	bool done; 	spin_lock(&ps->lock); 	while (job->size > 0) {		unsigned long start, size, end; 		start = job->start;		/* So end is chunk size aligned if enough work remains. */		size = roundup(start + 1, ps->chunk_size) - start;		size = min(size, job->size);		end = start + size; 		job->start = end;		job->size -= size; 		spin_unlock(&ps->lock);		job->thread_fn(start, end, job->fn_arg);		spin_lock(&ps->lock);	} 	++ps->nworks_fini;	done = (ps->nworks_fini == ps->nworks);	spin_unlock(&ps->lock); 	if (done)		complete(&ps->completion);} /** * padata_do_multithreaded - run a multithreaded job * @job: Description of the job. * * See the definition of struct padata_mt_job for more details. */void __init padata_do_multithreaded(struct padata_mt_job *job){	/* In case threads finish at different times. */	static const unsigned long load_balance_factor = 4;	struct padata_work my_work, *pw;	struct padata_mt_job_state ps;	LIST_HEAD(works);	int nworks; 	if (job->size == 0)		return; 	/* Ensure at least one thread when size < min_chunk. */	nworks = max(job->size / job->min_chunk, 1ul);	nworks = min(nworks, job->max_threads); 	if (nworks == 1) {		/* Single thread, no coordination needed, cut to the chase. */		job->thread_fn(job->start, job->start + job->size, job->fn_arg);		return;	}

Lines 1–500 of 1153. Load more