padata.c

   1// SPDX-License-Identifier: GPL-2.0
   2/*
   3 * padata.c - generic interface to process data streams in parallel
   4 *
   5 * See Documentation/core-api/padata.rst for more information.
   6 *
   7 * Copyright (C) 2008, 2009 secunet Security Networks AG
   8 * Copyright (C) 2008, 2009 Steffen Klassert <steffen.klassert@secunet.com>
   9 *
  10 * Copyright (c) 2020 Oracle and/or its affiliates.
  11 * Author: Daniel Jordan <daniel.m.jordan@oracle.com>
  12 */
  13
  14#include <linux/completion.h>
  15#include <linux/export.h>
  16#include <linux/cpumask.h>
  17#include <linux/err.h>
  18#include <linux/cpu.h>
  19#include <linux/padata.h>
  20#include <linux/mutex.h>
  21#include <linux/sched.h>
  22#include <linux/slab.h>
  23#include <linux/sysfs.h>
  24#include <linux/rcupdate.h>
  25
  26#define	PADATA_WORK_ONSTACK	1	/* Work's memory is on stack */
  27
  28struct padata_work {
  29	struct work_struct	pw_work;
  30	struct list_head	pw_list;  /* padata_free_works linkage */
  31	void			*pw_data;
  32};
  33
  34static DEFINE_SPINLOCK(padata_works_lock);
  35static struct padata_work *padata_works;
  36static LIST_HEAD(padata_free_works);
  37
  38struct padata_mt_job_state {
  39	spinlock_t		lock;
  40	struct completion	completion;
  41	struct padata_mt_job	*job;
  42	int			nworks;
  43	int			nworks_fini;
  44	unsigned long		chunk_size;
  45};
  46
  47static void padata_free_pd(struct parallel_data *pd);
  48static void __init padata_mt_helper(struct work_struct *work);
  49
  50static inline void padata_get_pd(struct parallel_data *pd)
  51{
  52	refcount_inc(&pd->refcnt);
  53}
  54
  55static inline void padata_put_pd_cnt(struct parallel_data *pd, int cnt)
  56{
  57	if (refcount_sub_and_test(cnt, &pd->refcnt))
  58		padata_free_pd(pd);
  59}
  60
  61static inline void padata_put_pd(struct parallel_data *pd)
  62{
  63	padata_put_pd_cnt(pd, 1);
  64}
  65
  66static int padata_cpu_hash(struct parallel_data *pd, unsigned int seq_nr)
  67{
  68	/*
  69	 * Hash the sequence numbers to the cpus by taking
  70	 * seq_nr mod. number of cpus in use.
  71	 */
  72	int cpu_index = seq_nr % cpumask_weight(pd->cpumask.pcpu);
  73
  74	return cpumask_nth(cpu_index, pd->cpumask.pcpu);
  75}
  76
  77static struct padata_work *padata_work_alloc(void)
  78{
  79	struct padata_work *pw;
  80
  81	lockdep_assert_held(&padata_works_lock);
  82
  83	if (list_empty(&padata_free_works))
  84		return NULL;	/* No more work items allowed to be queued. */
  85
  86	pw = list_first_entry(&padata_free_works, struct padata_work, pw_list);
  87	list_del(&pw->pw_list);
  88	return pw;
  89}
  90
  91/*
  92 * This function is marked __ref because this function may be optimized in such
  93 * a way that it directly refers to work_fn's address, which causes modpost to
  94 * complain when work_fn is marked __init. This scenario was observed with clang
  95 * LTO, where padata_work_init() was optimized to refer directly to
  96 * padata_mt_helper() because the calls to padata_work_init() with other work_fn
  97 * values were eliminated or inlined.
  98 */
  99static void __ref padata_work_init(struct padata_work *pw, work_func_t work_fn,
 100				   void *data, int flags)
 101{
 102	if (flags & PADATA_WORK_ONSTACK)
 103		INIT_WORK_ONSTACK(&pw->pw_work, work_fn);
 104	else
 105		INIT_WORK(&pw->pw_work, work_fn);
 106	pw->pw_data = data;
 107}
 108
 109static int __init padata_work_alloc_mt(int nworks, void *data,
 110				       struct list_head *head)
 111{
 112	int i;
 113
 114	spin_lock_bh(&padata_works_lock);
 115	/* Start at 1 because the current task participates in the job. */
 116	for (i = 1; i < nworks; ++i) {
 117		struct padata_work *pw = padata_work_alloc();
 118
 119		if (!pw)
 120			break;
 121		padata_work_init(pw, padata_mt_helper, data, 0);
 122		list_add(&pw->pw_list, head);
 123	}
 124	spin_unlock_bh(&padata_works_lock);
 125
 126	return i;
 127}
 128
 129static void padata_work_free(struct padata_work *pw)
 130{
 131	lockdep_assert_held(&padata_works_lock);
 132	list_add(&pw->pw_list, &padata_free_works);
 133}
 134
 135static void __init padata_works_free(struct list_head *works)
 136{
 137	struct padata_work *cur, *next;
 138
 139	if (list_empty(works))
 140		return;
 141
 142	spin_lock_bh(&padata_works_lock);
 143	list_for_each_entry_safe(cur, next, works, pw_list) {
 144		list_del(&cur->pw_list);
 145		padata_work_free(cur);
 146	}
 147	spin_unlock_bh(&padata_works_lock);
 148}
 149
 150static void padata_parallel_worker(struct work_struct *parallel_work)
 151{
 152	struct padata_work *pw = container_of(parallel_work, struct padata_work,
 153					      pw_work);
 154	struct padata_priv *padata = pw->pw_data;
 155
 156	local_bh_disable();
 157	padata->parallel(padata);
 158	spin_lock(&padata_works_lock);
 159	padata_work_free(pw);
 160	spin_unlock(&padata_works_lock);
 161	local_bh_enable();
 162}
 163
 164/**
 165 * padata_do_parallel - padata parallelization function
 166 *
 167 * @ps: padatashell
 168 * @padata: object to be parallelized
 169 * @cb_cpu: pointer to the CPU that the serialization callback function should
 170 *          run on.  If it's not in the serial cpumask of @pinst
 171 *          (i.e. cpumask.cbcpu), this function selects a fallback CPU and if
 172 *          none found, returns -EINVAL.
 173 *
 174 * The parallelization callback function will run with BHs off.
 175 * Note: Every object which is parallelized by padata_do_parallel
 176 * must be seen by padata_do_serial.
 177 *
 178 * Return: 0 on success or else negative error code.
 179 */
 180int padata_do_parallel(struct padata_shell *ps,
 181		       struct padata_priv *padata, int *cb_cpu)
 182{
 183	struct padata_instance *pinst = ps->pinst;
 184	struct parallel_data *pd;
 185	struct padata_work *pw;
 186	int cpu_index, err;
 187
 188	rcu_read_lock_bh();
 189
 190	pd = rcu_dereference_bh(ps->pd);
 191
 192	err = -EINVAL;
 193	if (!(pinst->flags & PADATA_INIT) || pinst->flags & PADATA_INVALID)
 194		goto out;
 195
 196	if (!cpumask_test_cpu(*cb_cpu, pd->cpumask.cbcpu)) {
 197		if (cpumask_empty(pd->cpumask.cbcpu))
 198			goto out;
 199
 200		/* Select an alternate fallback CPU and notify the caller. */
 201		cpu_index = *cb_cpu % cpumask_weight(pd->cpumask.cbcpu);
 202		*cb_cpu = cpumask_nth(cpu_index, pd->cpumask.cbcpu);
 203	}
 204
 205	err = -EBUSY;
 206	if ((pinst->flags & PADATA_RESET))
 207		goto out;
 208
 209	padata_get_pd(pd);
 210	padata->pd = pd;
 211	padata->cb_cpu = *cb_cpu;
 212
 213	spin_lock(&padata_works_lock);
 214	padata->seq_nr = ++pd->seq_nr;
 215	pw = padata_work_alloc();
 216	spin_unlock(&padata_works_lock);
 217
 218	if (!pw) {
 219		/* Maximum works limit exceeded, run in the current task. */
 220		padata->parallel(padata);
 221	}
 222
 223	rcu_read_unlock_bh();
 224
 225	if (pw) {
 226		padata_work_init(pw, padata_parallel_worker, padata, 0);
 227		queue_work(pinst->parallel_wq, &pw->pw_work);
 228	}
 229
 230	return 0;
 231out:
 232	rcu_read_unlock_bh();
 233
 234	return err;
 235}
 236EXPORT_SYMBOL(padata_do_parallel);
 237
 238/*
 239 * padata_find_next - Find the next object that needs serialization.
 240 *
 241 * Return:
 242 * * A pointer to the control struct of the next object that needs
 243 *   serialization, if present in one of the percpu reorder queues.
 244 * * NULL, if the next object that needs serialization will
 245 *   be parallel processed by another cpu and is not yet present in
 246 *   the cpu's reorder queue.
 247 */
 248static struct padata_priv *padata_find_next(struct parallel_data *pd, int cpu,
 249					    unsigned int processed)
 250{
 251	struct padata_priv *padata;
 252	struct padata_list *reorder;
 253
 254	reorder = per_cpu_ptr(pd->reorder_list, cpu);
 255
 256	spin_lock(&reorder->lock);
 257	if (list_empty(&reorder->list))
 258		goto notfound;
 259
 260	padata = list_entry(reorder->list.next, struct padata_priv, list);
 261
 262	/*
 263	 * Checks the rare case where two or more parallel jobs have hashed to
 264	 * the same CPU and one of the later ones finishes first.
 265	 */
 266	if (padata->seq_nr != processed)
 267		goto notfound;
 268
 269	list_del_init(&padata->list);
 270	spin_unlock(&reorder->lock);
 271	return padata;
 272
 273notfound:
 274	pd->processed = processed;
 275	pd->cpu = cpu;
 276	spin_unlock(&reorder->lock);
 277	return NULL;
 278}
 279
 280static void padata_reorder(struct padata_priv *padata)
 281{
 282	struct parallel_data *pd = padata->pd;
 283	struct padata_instance *pinst = pd->ps->pinst;
 284	unsigned int processed;
 285	int cpu;
 286
 287	processed = pd->processed;
 288	cpu = pd->cpu;
 289
 290	do {
 291		struct padata_serial_queue *squeue;
 292		int cb_cpu;
 293
 294		processed++;
 295		/* When sequence wraps around, reset to the first CPU. */
 296		if (unlikely(processed == 0))
 297			cpu = cpumask_first(pd->cpumask.pcpu);
 298		else
 299			cpu = cpumask_next_wrap(cpu, pd->cpumask.pcpu);
 300
 301		cb_cpu = padata->cb_cpu;
 302		squeue = per_cpu_ptr(pd->squeue, cb_cpu);
 303
 304		spin_lock(&squeue->serial.lock);
 305		list_add_tail(&padata->list, &squeue->serial.list);
 306		queue_work_on(cb_cpu, pinst->serial_wq, &squeue->work);
 307
 308		/*
 309		 * If the next object that needs serialization is parallel
 310		 * processed by another cpu and is still on it's way to the
 311		 * cpu's reorder queue, end the loop.
 312		 */
 313		padata = padata_find_next(pd, cpu, processed);
 314		spin_unlock(&squeue->serial.lock);
 315	} while (padata);
 316}
 317
 318static void padata_serial_worker(struct work_struct *serial_work)
 319{
 320	struct padata_serial_queue *squeue;
 321	struct parallel_data *pd;
 322	LIST_HEAD(local_list);
 323	int cnt;
 324
 325	local_bh_disable();
 326	squeue = container_of(serial_work, struct padata_serial_queue, work);
 327	pd = squeue->pd;
 328
 329	spin_lock(&squeue->serial.lock);
 330	list_replace_init(&squeue->serial.list, &local_list);
 331	spin_unlock(&squeue->serial.lock);
 332
 333	cnt = 0;
 334
 335	while (!list_empty(&local_list)) {
 336		struct padata_priv *padata;
 337
 338		padata = list_entry(local_list.next,
 339				    struct padata_priv, list);
 340
 341		list_del_init(&padata->list);
 342
 343		padata->serial(padata);
 344		cnt++;
 345	}
 346	local_bh_enable();
 347
 348	padata_put_pd_cnt(pd, cnt);
 349}
 350
 351/**
 352 * padata_do_serial - padata serialization function
 353 *
 354 * @padata: object to be serialized.
 355 *
 356 * padata_do_serial must be called for every parallelized object.
 357 * The serialization callback function will run with BHs off.
 358 */
 359void padata_do_serial(struct padata_priv *padata)
 360{
 361	struct parallel_data *pd = padata->pd;
 362	int hashed_cpu = padata_cpu_hash(pd, padata->seq_nr);
 363	struct padata_list *reorder = per_cpu_ptr(pd->reorder_list, hashed_cpu);
 364	struct padata_priv *cur;
 365	struct list_head *pos;
 366	bool gotit = true;
 367
 368	spin_lock(&reorder->lock);
 369	/* Sort in ascending order of sequence number. */
 370	list_for_each_prev(pos, &reorder->list) {
 371		cur = list_entry(pos, struct padata_priv, list);
 372		/* Compare by difference to consider integer wrap around */
 373		if ((signed int)(cur->seq_nr - padata->seq_nr) < 0)
 374			break;
 375	}
 376	if (padata->seq_nr != pd->processed) {
 377		gotit = false;
 378		list_add(&padata->list, pos);
 379	}
 380	spin_unlock(&reorder->lock);
 381
 382	if (gotit)
 383		padata_reorder(padata);
 384}
 385EXPORT_SYMBOL(padata_do_serial);
 386
 387static int padata_setup_cpumasks(struct padata_instance *pinst)
 388{
 389	struct workqueue_attrs *attrs;
 390	int err;
 391
 392	attrs = alloc_workqueue_attrs();
 393	if (!attrs)
 394		return -ENOMEM;
 395
 396	/* Restrict parallel_wq workers to pd->cpumask.pcpu. */
 397	cpumask_copy(attrs->cpumask, pinst->cpumask.pcpu);
 398	err = apply_workqueue_attrs(pinst->parallel_wq, attrs);
 399	free_workqueue_attrs(attrs);
 400
 401	return err;
 402}
 403
 404static void __init padata_mt_helper(struct work_struct *w)
 405{
 406	struct padata_work *pw = container_of(w, struct padata_work, pw_work);
 407	struct padata_mt_job_state *ps = pw->pw_data;
 408	struct padata_mt_job *job = ps->job;
 409	bool done;
 410
 411	spin_lock(&ps->lock);
 412
 413	while (job->size > 0) {
 414		unsigned long start, size, end;
 415
 416		start = job->start;
 417		/* So end is chunk size aligned if enough work remains. */
 418		size = roundup(start + 1, ps->chunk_size) - start;
 419		size = min(size, job->size);
 420		end = start + size;
 421
 422		job->start = end;
 423		job->size -= size;
 424
 425		spin_unlock(&ps->lock);
 426		job->thread_fn(start, end, job->fn_arg);
 427		spin_lock(&ps->lock);
 428	}
 429
 430	++ps->nworks_fini;
 431	done = (ps->nworks_fini == ps->nworks);
 432	spin_unlock(&ps->lock);
 433
 434	if (done)
 435		complete(&ps->completion);
 436}
 437
 438/**
 439 * padata_do_multithreaded - run a multithreaded job
 440 * @job: Description of the job.
 441 *
 442 * See the definition of struct padata_mt_job for more details.
 443 */
 444void __init padata_do_multithreaded(struct padata_mt_job *job)
 445{
 446	/* In case threads finish at different times. */
 447	static const unsigned long load_balance_factor = 4;
 448	struct padata_work my_work, *pw;
 449	struct padata_mt_job_state ps;
 450	LIST_HEAD(works);
 451	int nworks, nid;
 452	static atomic_t last_used_nid __initdata;
 453
 454	if (job->size == 0)
 455		return;
 456
 457	/* Ensure at least one thread when size < min_chunk. */
 458	nworks = max(job->size / max(job->min_chunk, job->align), 1ul);
 459	nworks = min(nworks, job->max_threads);
 460
 461	if (nworks == 1) {
 462		/* Single thread, no coordination needed, cut to the chase. */
 463		job->thread_fn(job->start, job->start + job->size, job->fn_arg);
 464		return;
 465	}
 466
 467	spin_lock_init(&ps.lock);
 468	init_completion(&ps.completion);
 469	ps.job	       = job;
 470	ps.nworks      = padata_work_alloc_mt(nworks, &ps, &works);
 471	ps.nworks_fini = 0;
 472
 473	/*
 474	 * Chunk size is the amount of work a helper does per call to the
 475	 * thread function.  Load balance large jobs between threads by
 476	 * increasing the number of chunks, guarantee at least the minimum
 477	 * chunk size from the caller, and honor the caller's alignment.
 478	 * Ensure chunk_size is at least 1 to prevent divide-by-0
 479	 * panic in padata_mt_helper().
 480	 */
 481	ps.chunk_size = job->size / (ps.nworks * load_balance_factor);
 482	ps.chunk_size = max(ps.chunk_size, job->min_chunk);
 483	ps.chunk_size = max(ps.chunk_size, 1ul);
 484	ps.chunk_size = roundup(ps.chunk_size, job->align);
 485
 486	list_for_each_entry(pw, &works, pw_list)
 487		if (job->numa_aware) {
 488			int old_node = atomic_read(&last_used_nid);
 489
 490			do {
 491				nid = next_node_in(old_node, node_states[N_CPU]);
 492			} while (!atomic_try_cmpxchg(&last_used_nid, &old_node, nid));
 493			queue_work_node(nid, system_dfl_wq, &pw->pw_work);
 494		} else {
 495			queue_work(system_dfl_wq, &pw->pw_work);
 496		}
 497
 498	/* Use the current thread, which saves starting a workqueue worker. */
 499	padata_work_init(&my_work, padata_mt_helper, &ps, PADATA_WORK_ONSTACK);
 500	padata_mt_helper(&my_work.pw_work);
 501
 502	/* Wait for all the helpers to finish. */
 503	wait_for_completion(&ps.completion);
 504
 505	destroy_work_on_stack(&my_work.pw_work);
 506	padata_works_free(&works);
 507}
 508
 509/* Initialize all percpu queues used by serial workers */
 510static void padata_init_squeues(struct parallel_data *pd)
 511{
 512	int cpu;
 513	struct padata_serial_queue *squeue;
 514
 515	for_each_cpu(cpu, pd->cpumask.cbcpu) {
 516		squeue = per_cpu_ptr(pd->squeue, cpu);
 517		squeue->pd = pd;
 518		INIT_LIST_HEAD(&squeue->serial.list);
 519		spin_lock_init(&squeue->serial.lock);
 520		INIT_WORK(&squeue->work, padata_serial_worker);
 521	}
 522}
 523
 524/* Initialize per-CPU reorder lists */
 525static void padata_init_reorder_list(struct parallel_data *pd)
 526{
 527	int cpu;
 528	struct padata_list *list;
 529
 530	for_each_cpu(cpu, pd->cpumask.pcpu) {
 531		list = per_cpu_ptr(pd->reorder_list, cpu);
 532		INIT_LIST_HEAD(&list->list);
 533		spin_lock_init(&list->lock);
 534	}
 535}
 536
 537/* Allocate and initialize the internal cpumask dependend resources. */
 538static struct parallel_data *padata_alloc_pd(struct padata_shell *ps)
 539{
 540	struct padata_instance *pinst = ps->pinst;
 541	struct parallel_data *pd;
 542
 543	pd = kzalloc(sizeof(struct parallel_data), GFP_KERNEL);
 544	if (!pd)
 545		goto err;
 546
 547	pd->reorder_list = alloc_percpu(struct padata_list);
 548	if (!pd->reorder_list)
 549		goto err_free_pd;
 550
 551	pd->squeue = alloc_percpu(struct padata_serial_queue);
 552	if (!pd->squeue)
 553		goto err_free_reorder_list;
 554
 555	pd->ps = ps;
 556
 557	if (!alloc_cpumask_var(&pd->cpumask.pcpu, GFP_KERNEL))
 558		goto err_free_squeue;
 559	if (!alloc_cpumask_var(&pd->cpumask.cbcpu, GFP_KERNEL))
 560		goto err_free_pcpu;
 561
 562	cpumask_and(pd->cpumask.pcpu, pinst->cpumask.pcpu, cpu_online_mask);
 563	cpumask_and(pd->cpumask.cbcpu, pinst->cpumask.cbcpu, cpu_online_mask);
 564
 565	padata_init_reorder_list(pd);
 566	padata_init_squeues(pd);
 567	pd->seq_nr = -1;
 568	refcount_set(&pd->refcnt, 1);
 569	pd->cpu = cpumask_first(pd->cpumask.pcpu);
 570
 571	return pd;
 572
 573err_free_pcpu:
 574	free_cpumask_var(pd->cpumask.pcpu);
 575err_free_squeue:
 576	free_percpu(pd->squeue);
 577err_free_reorder_list:
 578	free_percpu(pd->reorder_list);
 579err_free_pd:
 580	kfree(pd);
 581err:
 582	return NULL;
 583}
 584
 585static void padata_free_pd(struct parallel_data *pd)
 586{
 587	free_cpumask_var(pd->cpumask.pcpu);
 588	free_cpumask_var(pd->cpumask.cbcpu);
 589	free_percpu(pd->reorder_list);
 590	free_percpu(pd->squeue);
 591	kfree(pd);
 592}
 593
 594static void __padata_start(struct padata_instance *pinst)
 595{
 596	pinst->flags |= PADATA_INIT;
 597}
 598
 599static void __padata_stop(struct padata_instance *pinst)
 600{
 601	if (!(pinst->flags & PADATA_INIT))
 602		return;
 603
 604	pinst->flags &= ~PADATA_INIT;
 605
 606	synchronize_rcu();
 607}
 608
 609/* Replace the internal control structure with a new one. */
 610static int padata_replace_one(struct padata_shell *ps)
 611{
 612	struct parallel_data *pd_new;
 613
 614	pd_new = padata_alloc_pd(ps);
 615	if (!pd_new)
 616		return -ENOMEM;
 617
 618	ps->opd = rcu_dereference_protected(ps->pd, 1);
 619	rcu_assign_pointer(ps->pd, pd_new);
 620
 621	return 0;
 622}
 623
 624static int padata_replace(struct padata_instance *pinst)
 625{
 626	struct padata_shell *ps;
 627	int err = 0;
 628
 629	pinst->flags |= PADATA_RESET;
 630
 631	list_for_each_entry(ps, &pinst->pslist, list) {
 632		err = padata_replace_one(ps);
 633		if (err)
 634			break;
 635	}
 636
 637	synchronize_rcu();
 638
 639	list_for_each_entry_continue_reverse(ps, &pinst->pslist, list)
 640		padata_put_pd(ps->opd);
 641
 642	pinst->flags &= ~PADATA_RESET;
 643
 644	return err;
 645}
 646
 647/* If cpumask contains no active cpu, we mark the instance as invalid. */
 648static bool padata_validate_cpumask(struct padata_instance *pinst,
 649				    const struct cpumask *cpumask)
 650{
 651	if (!cpumask_intersects(cpumask, cpu_online_mask)) {
 652		pinst->flags |= PADATA_INVALID;
 653		return false;
 654	}
 655
 656	pinst->flags &= ~PADATA_INVALID;
 657	return true;
 658}
 659
 660static int __padata_set_cpumasks(struct padata_instance *pinst,
 661				 cpumask_var_t pcpumask,
 662				 cpumask_var_t cbcpumask)
 663{
 664	int valid;
 665	int err;
 666
 667	valid = padata_validate_cpumask(pinst, pcpumask);
 668	if (!valid) {
 669		__padata_stop(pinst);
 670		goto out_replace;
 671	}
 672
 673	valid = padata_validate_cpumask(pinst, cbcpumask);
 674	if (!valid)
 675		__padata_stop(pinst);
 676
 677out_replace:
 678	cpumask_copy(pinst->cpumask.pcpu, pcpumask);
 679	cpumask_copy(pinst->cpumask.cbcpu, cbcpumask);
 680
 681	err = padata_setup_cpumasks(pinst) ?: padata_replace(pinst);
 682
 683	if (valid)
 684		__padata_start(pinst);
 685
 686	return err;
 687}
 688
 689/**
 690 * padata_set_cpumask - Sets specified by @cpumask_type cpumask to the value
 691 *                      equivalent to @cpumask.
 692 * @pinst: padata instance
 693 * @cpumask_type: PADATA_CPU_SERIAL or PADATA_CPU_PARALLEL corresponding
 694 *                to parallel and serial cpumasks respectively.
 695 * @cpumask: the cpumask to use
 696 *
 697 * Return: 0 on success or negative error code
 698 */
 699int padata_set_cpumask(struct padata_instance *pinst, int cpumask_type,
 700		       cpumask_var_t cpumask)
 701{
 702	struct cpumask *serial_mask, *parallel_mask;
 703	int err = -EINVAL;
 704
 705	cpus_read_lock();
 706	mutex_lock(&pinst->lock);
 707
 708	switch (cpumask_type) {
 709	case PADATA_CPU_PARALLEL:
 710		serial_mask = pinst->cpumask.cbcpu;
 711		parallel_mask = cpumask;
 712		break;
 713	case PADATA_CPU_SERIAL:
 714		parallel_mask = pinst->cpumask.pcpu;
 715		serial_mask = cpumask;
 716		break;
 717	default:
 718		 goto out;
 719	}
 720
 721	err =  __padata_set_cpumasks(pinst, parallel_mask, serial_mask);
 722
 723out:
 724	mutex_unlock(&pinst->lock);
 725	cpus_read_unlock();
 726
 727	return err;
 728}
 729EXPORT_SYMBOL(padata_set_cpumask);
 730
 731#ifdef CONFIG_HOTPLUG_CPU
 732
 733static int __padata_add_cpu(struct padata_instance *pinst, int cpu)
 734{
 735	int err = 0;
 736
 737	if (cpumask_test_cpu(cpu, cpu_online_mask)) {
 738		err = padata_replace(pinst);
 739
 740		if (padata_validate_cpumask(pinst, pinst->cpumask.pcpu) &&
 741		    padata_validate_cpumask(pinst, pinst->cpumask.cbcpu))
 742			__padata_start(pinst);
 743	}
 744
 745	return err;
 746}
 747
 748static int __padata_remove_cpu(struct padata_instance *pinst, int cpu)
 749{
 750	int err = 0;
 751
 752	if (!cpumask_test_cpu(cpu, cpu_online_mask)) {
 753		if (!padata_validate_cpumask(pinst, pinst->cpumask.pcpu) ||
 754		    !padata_validate_cpumask(pinst, pinst->cpumask.cbcpu))
 755			__padata_stop(pinst);
 756
 757		err = padata_replace(pinst);
 758	}
 759
 760	return err;
 761}
 762
 763static inline int pinst_has_cpu(struct padata_instance *pinst, int cpu)
 764{
 765	return cpumask_test_cpu(cpu, pinst->cpumask.pcpu) ||
 766		cpumask_test_cpu(cpu, pinst->cpumask.cbcpu);
 767}
 768
 769static int padata_cpu_online(unsigned int cpu, struct hlist_node *node)
 770{
 771	struct padata_instance *pinst;
 772	int ret;
 773
 774	pinst = hlist_entry_safe(node, struct padata_instance, cpu_online_node);
 775	if (!pinst_has_cpu(pinst, cpu))
 776		return 0;
 777
 778	mutex_lock(&pinst->lock);
 779	ret = __padata_add_cpu(pinst, cpu);
 780	mutex_unlock(&pinst->lock);
 781	return ret;
 782}
 783
 784static int padata_cpu_dead(unsigned int cpu, struct hlist_node *node)
 785{
 786	struct padata_instance *pinst;
 787	int ret;
 788
 789	pinst = hlist_entry_safe(node, struct padata_instance, cpu_dead_node);
 790	if (!pinst_has_cpu(pinst, cpu))
 791		return 0;
 792
 793	mutex_lock(&pinst->lock);
 794	ret = __padata_remove_cpu(pinst, cpu);
 795	mutex_unlock(&pinst->lock);
 796	return ret;
 797}
 798
 799static enum cpuhp_state hp_online;
 800#endif
 801
 802static void __padata_free(struct padata_instance *pinst)
 803{
 804#ifdef CONFIG_HOTPLUG_CPU
 805	cpuhp_state_remove_instance_nocalls(CPUHP_PADATA_DEAD,
 806					    &pinst->cpu_dead_node);
 807	cpuhp_state_remove_instance_nocalls(hp_online, &pinst->cpu_online_node);
 808#endif
 809
 810	WARN_ON(!list_empty(&pinst->pslist));
 811
 812	free_cpumask_var(pinst->cpumask.pcpu);
 813	free_cpumask_var(pinst->cpumask.cbcpu);
 814	destroy_workqueue(pinst->serial_wq);
 815	destroy_workqueue(pinst->parallel_wq);
 816	kfree(pinst);
 817}
 818
 819#define kobj2pinst(_kobj)					\
 820	container_of(_kobj, struct padata_instance, kobj)
 821#define attr2pentry(_attr)					\
 822	container_of(_attr, struct padata_sysfs_entry, attr)
 823
 824static void padata_sysfs_release(struct kobject *kobj)
 825{
 826	struct padata_instance *pinst = kobj2pinst(kobj);
 827	__padata_free(pinst);
 828}
 829
 830struct padata_sysfs_entry {
 831	struct attribute attr;
 832	ssize_t (*show)(struct padata_instance *, struct attribute *, char *);
 833	ssize_t (*store)(struct padata_instance *, struct attribute *,
 834			 const char *, size_t);
 835};
 836
 837static ssize_t show_cpumask(struct padata_instance *pinst,
 838			    struct attribute *attr,  char *buf)
 839{
 840	struct cpumask *cpumask;
 841	ssize_t len;
 842
 843	mutex_lock(&pinst->lock);
 844	if (!strcmp(attr->name, "serial_cpumask"))
 845		cpumask = pinst->cpumask.cbcpu;
 846	else
 847		cpumask = pinst->cpumask.pcpu;
 848
 849	len = snprintf(buf, PAGE_SIZE, "%*pb\n",
 850		       nr_cpu_ids, cpumask_bits(cpumask));
 851	mutex_unlock(&pinst->lock);
 852	return len < PAGE_SIZE ? len : -EINVAL;
 853}
 854
 855static ssize_t store_cpumask(struct padata_instance *pinst,
 856			     struct attribute *attr,
 857			     const char *buf, size_t count)
 858{
 859	cpumask_var_t new_cpumask;
 860	ssize_t ret;
 861	int mask_type;
 862
 863	if (!alloc_cpumask_var(&new_cpumask, GFP_KERNEL))
 864		return -ENOMEM;
 865
 866	ret = bitmap_parse(buf, count, cpumask_bits(new_cpumask),
 867			   nr_cpumask_bits);
 868	if (ret < 0)
 869		goto out;
 870
 871	mask_type = !strcmp(attr->name, "serial_cpumask") ?
 872		PADATA_CPU_SERIAL : PADATA_CPU_PARALLEL;
 873	ret = padata_set_cpumask(pinst, mask_type, new_cpumask);
 874	if (!ret)
 875		ret = count;
 876
 877out:
 878	free_cpumask_var(new_cpumask);
 879	return ret;
 880}
 881
 882#define PADATA_ATTR_RW(_name, _show_name, _store_name)		\
 883	static struct padata_sysfs_entry _name##_attr =		\
 884		__ATTR(_name, 0644, _show_name, _store_name)
 885#define PADATA_ATTR_RO(_name, _show_name)		\
 886	static struct padata_sysfs_entry _name##_attr = \
 887		__ATTR(_name, 0400, _show_name, NULL)
 888
 889PADATA_ATTR_RW(serial_cpumask, show_cpumask, store_cpumask);
 890PADATA_ATTR_RW(parallel_cpumask, show_cpumask, store_cpumask);
 891
 892/*
 893 * Padata sysfs provides the following objects:
 894 * serial_cpumask   [RW] - cpumask for serial workers
 895 * parallel_cpumask [RW] - cpumask for parallel workers
 896 */
 897static struct attribute *padata_default_attrs[] = {
 898	&serial_cpumask_attr.attr,
 899	&parallel_cpumask_attr.attr,
 900	NULL,
 901};
 902ATTRIBUTE_GROUPS(padata_default);
 903
 904static ssize_t padata_sysfs_show(struct kobject *kobj,
 905				 struct attribute *attr, char *buf)
 906{
 907	struct padata_instance *pinst;
 908	struct padata_sysfs_entry *pentry;
 909	ssize_t ret = -EIO;
 910
 911	pinst = kobj2pinst(kobj);
 912	pentry = attr2pentry(attr);
 913	if (pentry->show)
 914		ret = pentry->show(pinst, attr, buf);
 915
 916	return ret;
 917}
 918
 919static ssize_t padata_sysfs_store(struct kobject *kobj, struct attribute *attr,
 920				  const char *buf, size_t count)
 921{
 922	struct padata_instance *pinst;
 923	struct padata_sysfs_entry *pentry;
 924	ssize_t ret = -EIO;
 925
 926	pinst = kobj2pinst(kobj);
 927	pentry = attr2pentry(attr);
 928	if (pentry->store)
 929		ret = pentry->store(pinst, attr, buf, count);
 930
 931	return ret;
 932}
 933
 934static const struct sysfs_ops padata_sysfs_ops = {
 935	.show = padata_sysfs_show,
 936	.store = padata_sysfs_store,
 937};
 938
 939static const struct kobj_type padata_attr_type = {
 940	.sysfs_ops = &padata_sysfs_ops,
 941	.default_groups = padata_default_groups,
 942	.release = padata_sysfs_release,
 943};
 944
 945/**
 946 * padata_alloc - allocate and initialize a padata instance
 947 * @name: used to identify the instance
 948 *
 949 * Return: new instance on success, NULL on error
 950 */
 951struct padata_instance *padata_alloc(const char *name)
 952{
 953	struct padata_instance *pinst;
 954
 955	pinst = kzalloc(sizeof(struct padata_instance), GFP_KERNEL);
 956	if (!pinst)
 957		goto err;
 958
 959	pinst->parallel_wq = alloc_workqueue("%s_parallel", WQ_UNBOUND, 0,
 960					     name);
 961	if (!pinst->parallel_wq)
 962		goto err_free_inst;
 963
 964	cpus_read_lock();
 965
 966	pinst->serial_wq = alloc_workqueue("%s_serial",
 967					   WQ_MEM_RECLAIM | WQ_CPU_INTENSIVE | WQ_PERCPU,
 968					   1, name);
 969	if (!pinst->serial_wq)
 970		goto err_put_cpus;
 971
 972	if (!alloc_cpumask_var(&pinst->cpumask.pcpu, GFP_KERNEL))
 973		goto err_free_serial_wq;
 974	if (!alloc_cpumask_var(&pinst->cpumask.cbcpu, GFP_KERNEL)) {
 975		free_cpumask_var(pinst->cpumask.pcpu);
 976		goto err_free_serial_wq;
 977	}
 978
 979	INIT_LIST_HEAD(&pinst->pslist);
 980
 981	cpumask_copy(pinst->cpumask.pcpu, cpu_possible_mask);
 982	cpumask_copy(pinst->cpumask.cbcpu, cpu_possible_mask);
 983
 984	if (padata_setup_cpumasks(pinst))
 985		goto err_free_masks;
 986
 987	__padata_start(pinst);
 988
 989	kobject_init(&pinst->kobj, &padata_attr_type);
 990	mutex_init(&pinst->lock);
 991
 992#ifdef CONFIG_HOTPLUG_CPU
 993	cpuhp_state_add_instance_nocalls_cpuslocked(hp_online,
 994						    &pinst->cpu_online_node);
 995	cpuhp_state_add_instance_nocalls_cpuslocked(CPUHP_PADATA_DEAD,
 996						    &pinst->cpu_dead_node);
 997#endif
 998
 999	cpus_read_unlock();
1000
1001	return pinst;
1002
1003err_free_masks:
1004	free_cpumask_var(pinst->cpumask.pcpu);
1005	free_cpumask_var(pinst->cpumask.cbcpu);
1006err_free_serial_wq:
1007	destroy_workqueue(pinst->serial_wq);
1008err_put_cpus:
1009	cpus_read_unlock();
1010	destroy_workqueue(pinst->parallel_wq);
1011err_free_inst:
1012	kfree(pinst);
1013err:
1014	return NULL;
1015}
1016EXPORT_SYMBOL(padata_alloc);
1017
1018/**
1019 * padata_free - free a padata instance
1020 *
1021 * @pinst: padata instance to free
1022 */
1023void padata_free(struct padata_instance *pinst)
1024{
1025	kobject_put(&pinst->kobj);
1026}
1027EXPORT_SYMBOL(padata_free);
1028
1029/**
1030 * padata_alloc_shell - Allocate and initialize padata shell.
1031 *
1032 * @pinst: Parent padata_instance object.
1033 *
1034 * Return: new shell on success, NULL on error
1035 */
1036struct padata_shell *padata_alloc_shell(struct padata_instance *pinst)
1037{
1038	struct parallel_data *pd;
1039	struct padata_shell *ps;
1040
1041	ps = kzalloc(sizeof(*ps), GFP_KERNEL);
1042	if (!ps)
1043		goto out;
1044
1045	ps->pinst = pinst;
1046
1047	cpus_read_lock();
1048	pd = padata_alloc_pd(ps);
1049	cpus_read_unlock();
1050
1051	if (!pd)
1052		goto out_free_ps;
1053
1054	mutex_lock(&pinst->lock);
1055	RCU_INIT_POINTER(ps->pd, pd);
1056	list_add(&ps->list, &pinst->pslist);
1057	mutex_unlock(&pinst->lock);
1058
1059	return ps;
1060
1061out_free_ps:
1062	kfree(ps);
1063out:
1064	return NULL;
1065}
1066EXPORT_SYMBOL(padata_alloc_shell);
1067
1068/**
1069 * padata_free_shell - free a padata shell
1070 *
1071 * @ps: padata shell to free
1072 */
1073void padata_free_shell(struct padata_shell *ps)
1074{
1075	struct parallel_data *pd;
1076
1077	if (!ps)
1078		return;
1079
1080	mutex_lock(&ps->pinst->lock);
1081	list_del(&ps->list);
1082	pd = rcu_dereference_protected(ps->pd, 1);
1083	padata_put_pd(pd);
1084	mutex_unlock(&ps->pinst->lock);
1085
1086	kfree(ps);
1087}
1088EXPORT_SYMBOL(padata_free_shell);
1089
1090void __init padata_init(void)
1091{
1092	unsigned int i, possible_cpus;
1093#ifdef CONFIG_HOTPLUG_CPU
1094	int ret;
1095
1096	ret = cpuhp_setup_state_multi(CPUHP_AP_ONLINE_DYN, "padata:online",
1097				      padata_cpu_online, NULL);
1098	if (ret < 0)
1099		goto err;
1100	hp_online = ret;
1101
1102	ret = cpuhp_setup_state_multi(CPUHP_PADATA_DEAD, "padata:dead",
1103				      NULL, padata_cpu_dead);
1104	if (ret < 0)
1105		goto remove_online_state;
1106#endif
1107
1108	possible_cpus = num_possible_cpus();
1109	padata_works = kmalloc_array(possible_cpus, sizeof(struct padata_work),
1110				     GFP_KERNEL);
1111	if (!padata_works)
1112		goto remove_dead_state;
1113
1114	for (i = 0; i < possible_cpus; ++i)
1115		list_add(&padata_works[i].pw_list, &padata_free_works);
1116
1117	return;
1118
1119remove_dead_state:
1120#ifdef CONFIG_HOTPLUG_CPU
1121	cpuhp_remove_multi_state(CPUHP_PADATA_DEAD);
1122remove_online_state:
1123	cpuhp_remove_multi_state(hp_online);
1124err:
1125#endif
1126	pr_warn("padata: initialization failed\n");
1127}