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