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/padata.txt for an api documentation.
   6  *
   7  * Copyright (C) 2008, 2009 secunet Security Networks AG
   8  * Copyright (C) 2008, 2009 Steffen Klassert <steffen.klassert@secunet.com>
   9  *
  10  * This program is free software; you can redistribute it and/or modify it
  11  * under the terms and conditions of the GNU General Public License,
  12  * version 2, as published by the Free Software Foundation.
  13  *
  14  * This program is distributed in the hope it will be useful, but WITHOUT
  15  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  16  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  17  * more details.
  18  *
  19  * You should have received a copy of the GNU General Public License along with
  20  * this program; if not, write to the Free Software Foundation, Inc.,
  21  * 51 Franklin St - Fifth Floor, Boston, MA 02110-1301 USA.
  22  */
  23
  24 #include <linux/export.h>
  25 #include <linux/cpumask.h>
  26 #include <linux/err.h>
  27 #include <linux/cpu.h>
  28 #include <linux/padata.h>
  29 #include <linux/mutex.h>
  30 #include <linux/sched.h>
  31 #include <linux/slab.h>
  32 #include <linux/sysfs.h>
  33 #include <linux/rcupdate.h>
  34 #include <linux/module.h>
  35
  36 #define MAX_OBJ_NUM 1000
  37
  38 static int padata_index_to_cpu(struct parallel_data *pd, int cpu_index)
  39 {
  40         int cpu, target_cpu;
  41
  42         target_cpu = cpumask_first(pd->cpumask.pcpu);
  43         for (cpu = 0; cpu < cpu_index; cpu++)
  44                 target_cpu = cpumask_next(target_cpu, pd->cpumask.pcpu);
  45
  46         return target_cpu;
  47 }
  48
  49 static int padata_cpu_hash(struct parallel_data *pd, unsigned int seq_nr)
  50 {
  51         /*
  52          * Hash the sequence numbers to the cpus by taking
  53          * seq_nr mod. number of cpus in use.
  54          */
  55         int cpu_index = seq_nr % cpumask_weight(pd->cpumask.pcpu);
  56
  57         return padata_index_to_cpu(pd, cpu_index);
  58 }
  59
  60 static void padata_parallel_worker(struct work_struct *parallel_work)
  61 {
  62         struct padata_parallel_queue *pqueue;
  63         LIST_HEAD(local_list);
  64
  65         local_bh_disable();
  66         pqueue = container_of(parallel_work,
  67                               struct padata_parallel_queue, work);
  68
  69         spin_lock(&pqueue->parallel.lock);
  70         list_replace_init(&pqueue->parallel.list, &local_list);
  71         spin_unlock(&pqueue->parallel.lock);
  72
  73         while (!list_empty(&local_list)) {
  74                 struct padata_priv *padata;
  75
  76                 padata = list_entry(local_list.next,
  77                                     struct padata_priv, list);
  78
  79                 list_del_init(&padata->list);
  80
  81                 padata->parallel(padata);
  82         }
  83
  84         local_bh_enable();
  85 }
  86
  87 /**
  88  * padata_do_parallel - padata parallelization function
  89  *
  90  * @pinst: padata instance
  91  * @padata: object to be parallelized
  92  * @cb_cpu: pointer to the CPU that the serialization callback function should
  93  *          run on.  If it's not in the serial cpumask of @pinst
  94  *          (i.e. cpumask.cbcpu), this function selects a fallback CPU and if
  95  *          none found, returns -EINVAL.
  96  *
  97  * The parallelization callback function will run with BHs off.
  98  * Note: Every object which is parallelized by padata_do_parallel
  99  * must be seen by padata_do_serial.
 100  */
 101 int padata_do_parallel(struct padata_instance *pinst,
 102                        struct padata_priv *padata, int *cb_cpu)
 103 {
 104         int i, cpu, cpu_index, target_cpu, err;
 105         struct padata_parallel_queue *queue;
 106         struct parallel_data *pd;
 107
 108         rcu_read_lock_bh();
 109
 110         pd = rcu_dereference_bh(pinst->pd);
 111
 112         err = -EINVAL;
 113         if (!(pinst->flags & PADATA_INIT) || pinst->flags & PADATA_INVALID)
 114                 goto out;
 115
 116         if (!cpumask_test_cpu(*cb_cpu, pd->cpumask.cbcpu)) {
 117                 if (!cpumask_weight(pd->cpumask.cbcpu))
 118                         goto out;
 119
 120                 /* Select an alternate fallback CPU and notify the caller. */
 121                 cpu_index = *cb_cpu % cpumask_weight(pd->cpumask.cbcpu);
 122
 123                 cpu = cpumask_first(pd->cpumask.cbcpu);
 124                 for (i = 0; i < cpu_index; i++)
 125                         cpu = cpumask_next(cpu, pd->cpumask.cbcpu);
 126
 127                 *cb_cpu = cpu;
 128         }
 129
 130         err =  -EBUSY;
 131         if ((pinst->flags & PADATA_RESET))
 132                 goto out;
 133
 134         if (atomic_read(&pd->refcnt) >= MAX_OBJ_NUM)
 135                 goto out;
 136
 137         err = 0;
 138         atomic_inc(&pd->refcnt);
 139         padata->pd = pd;
 140         padata->cb_cpu = *cb_cpu;
 141
 142         padata->seq_nr = atomic_inc_return(&pd->seq_nr);
 143         target_cpu = padata_cpu_hash(pd, padata->seq_nr);
 144         padata->cpu = target_cpu;
 145         queue = per_cpu_ptr(pd->pqueue, target_cpu);
 146
 147         spin_lock(&queue->parallel.lock);
 148         list_add_tail(&padata->list, &queue->parallel.list);
 149         spin_unlock(&queue->parallel.lock);
 150
 151         queue_work(pinst->parallel_wq, &queue->work);
 152
 153 out:
 154         rcu_read_unlock_bh();
 155
 156         return err;
 157 }
 158 EXPORT_SYMBOL(padata_do_parallel);
 159
 160 /*
 161  * padata_find_next - Find the next object that needs serialization.
 162  *
 163  * Return values are:
 164  *
 165  * A pointer to the control struct of the next object that needs
 166  * serialization, if present in one of the percpu reorder queues.
 167  *
 168  * NULL, if the next object that needs serialization will
 169  *  be parallel processed by another cpu and is not yet present in
 170  *  the cpu's reorder queue.
 171  */
 172 static struct padata_priv *padata_find_next(struct parallel_data *pd,
 173                                             bool remove_object)
 174 {
 175         struct padata_parallel_queue *next_queue;
 176         struct padata_priv *padata;
 177         struct padata_list *reorder;
 178         int cpu = pd->cpu;
 179
 180         next_queue = per_cpu_ptr(pd->pqueue, cpu);
 181         reorder = &next_queue->reorder;
 182
 183         spin_lock(&reorder->lock);
 184         if (list_empty(&reorder->list)) {
 185                 spin_unlock(&reorder->lock);
 186                 return NULL;
 187         }
 188
 189         padata = list_entry(reorder->list.next, struct padata_priv, list);
 190
 191         /*
 192          * Checks the rare case where two or more parallel jobs have hashed to
 193          * the same CPU and one of the later ones finishes first.
 194          */
 195         if (padata->seq_nr != pd->processed) {
 196                 spin_unlock(&reorder->lock);
 197                 return NULL;
 198         }
 199
 200         if (remove_object) {
 201                 list_del_init(&padata->list);
 202                 atomic_dec(&pd->reorder_objects);
 203                 ++pd->processed;
 204                 pd->cpu = cpumask_next_wrap(cpu, pd->cpumask.pcpu, -1, false);
 205         }
 206
 207         spin_unlock(&reorder->lock);
 208         return padata;
 209 }
 210
 211 static void padata_reorder(struct parallel_data *pd)
 212 {
 213         int cb_cpu;
 214         struct padata_priv *padata;
 215         struct padata_serial_queue *squeue;
 216         struct padata_instance *pinst = pd->pinst;
 217         struct padata_parallel_queue *next_queue;
 218
 219         /*
 220          * We need to ensure that only one cpu can work on dequeueing of
 221          * the reorder queue the time. Calculating in which percpu reorder
 222          * queue the next object will arrive takes some time. A spinlock
 223          * would be highly contended. Also it is not clear in which order
 224          * the objects arrive to the reorder queues. So a cpu could wait to
 225          * get the lock just to notice that there is nothing to do at the
 226          * moment. Therefore we use a trylock and let the holder of the lock
 227          * care for all the objects enqueued during the holdtime of the lock.
 228          */
 229         if (!spin_trylock_bh(&pd->lock))
 230                 return;
 231
 232         while (1) {
 233                 padata = padata_find_next(pd, true);
 234
 235                 /*
 236                  * If the next object that needs serialization is parallel
 237                  * processed by another cpu and is still on it's way to the
 238                  * cpu's reorder queue, nothing to do for now.
 239                  */
 240                 if (!padata)
 241                         break;
 242
 243                 cb_cpu = padata->cb_cpu;
 244                 squeue = per_cpu_ptr(pd->squeue, cb_cpu);
 245
 246                 spin_lock(&squeue->serial.lock);
 247                 list_add_tail(&padata->list, &squeue->serial.list);
 248                 spin_unlock(&squeue->serial.lock);
 249
 250                 queue_work_on(cb_cpu, pinst->serial_wq, &squeue->work);
 251         }
 252
 253         spin_unlock_bh(&pd->lock);
 254
 255         /*
 256          * The next object that needs serialization might have arrived to
 257          * the reorder queues in the meantime.
 258          *
 259          * Ensure reorder queue is read after pd->lock is dropped so we see
 260          * new objects from another task in padata_do_serial.  Pairs with
 261          * smp_mb__after_atomic in padata_do_serial.
 262          */
 263         smp_mb();
 264
 265         next_queue = per_cpu_ptr(pd->pqueue, pd->cpu);
 266         if (!list_empty(&next_queue->reorder.list) &&
 267             padata_find_next(pd, false))
 268                 queue_work(pinst->serial_wq, &pd->reorder_work);
 269 }
 270
 271 static void invoke_padata_reorder(struct work_struct *work)
 272 {
 273         struct parallel_data *pd;
 274
 275         local_bh_disable();
 276         pd = container_of(work, struct parallel_data, reorder_work);
 277         padata_reorder(pd);
 278         local_bh_enable();
 279 }
 280
 281 static void padata_serial_worker(struct work_struct *serial_work)
 282 {
 283         struct padata_serial_queue *squeue;
 284         struct parallel_data *pd;
 285         LIST_HEAD(local_list);
 286
 287         local_bh_disable();
 288         squeue = container_of(serial_work, struct padata_serial_queue, work);
 289         pd = squeue->pd;
 290
 291         spin_lock(&squeue->serial.lock);
 292         list_replace_init(&squeue->serial.list, &local_list);
 293         spin_unlock(&squeue->serial.lock);
 294
 295         while (!list_empty(&local_list)) {
 296                 struct padata_priv *padata;
 297
 298                 padata = list_entry(local_list.next,
 299                                     struct padata_priv, list);
 300
 301                 list_del_init(&padata->list);
 302
 303                 padata->serial(padata);
 304                 atomic_dec(&pd->refcnt);
 305         }
 306         local_bh_enable();
 307 }
 308
 309 /**
 310  * padata_do_serial - padata serialization function
 311  *
 312  * @padata: object to be serialized.
 313  *
 314  * padata_do_serial must be called for every parallelized object.
 315  * The serialization callback function will run with BHs off.
 316  */
 317 void padata_do_serial(struct padata_priv *padata)
 318 {
 319         struct parallel_data *pd = padata->pd;
 320         struct padata_parallel_queue *pqueue = per_cpu_ptr(pd->pqueue,
 321                                                            padata->cpu);
 322         struct padata_priv *cur;
 323
 324         spin_lock(&pqueue->reorder.lock);
 325         /* Sort in ascending order of sequence number. */
 326         list_for_each_entry_reverse(cur, &pqueue->reorder.list, list)
 327                 if (cur->seq_nr < padata->seq_nr)
 328                         break;
 329         list_add(&padata->list, &cur->list);
 330         atomic_inc(&pd->reorder_objects);
 331         spin_unlock(&pqueue->reorder.lock);
 332
 333         /*
 334          * Ensure the addition to the reorder list is ordered correctly
 335          * with the trylock of pd->lock in padata_reorder.  Pairs with smp_mb
 336          * in padata_reorder.
 337          */
 338         smp_mb__after_atomic();
 339
 340         padata_reorder(pd);
 341 }
 342 EXPORT_SYMBOL(padata_do_serial);
 343
 344 static int padata_setup_cpumasks(struct parallel_data *pd,
 345                                  const struct cpumask *pcpumask,
 346                                  const struct cpumask *cbcpumask)
 347 {
 348         struct workqueue_attrs *attrs;
 349         int err = -ENOMEM;
 350
 351         if (!alloc_cpumask_var(&pd->cpumask.pcpu, GFP_KERNEL))
 352                 goto out;
 353         cpumask_and(pd->cpumask.pcpu, pcpumask, cpu_online_mask);
 354
 355         if (!alloc_cpumask_var(&pd->cpumask.cbcpu, GFP_KERNEL))
 356                 goto free_pcpu_mask;
 357         cpumask_and(pd->cpumask.cbcpu, cbcpumask, cpu_online_mask);
 358
 359         attrs = alloc_workqueue_attrs();
 360         if (!attrs)
 361                 goto free_cbcpu_mask;
 362
 363         /* Restrict parallel_wq workers to pd->cpumask.pcpu. */
 364         cpumask_copy(attrs->cpumask, pd->cpumask.pcpu);
 365         err = apply_workqueue_attrs(pd->pinst->parallel_wq, attrs);
 366         free_workqueue_attrs(attrs);
 367         if (err < 0)
 368                 goto free_cbcpu_mask;
 369
 370         return 0;
 371
 372 free_cbcpu_mask:
 373         free_cpumask_var(pd->cpumask.cbcpu);
 374 free_pcpu_mask:
 375         free_cpumask_var(pd->cpumask.pcpu);
 376 out:
 377         return err;
 378 }
 379
 380 static void __padata_list_init(struct padata_list *pd_list)
 381 {
 382         INIT_LIST_HEAD(&pd_list->list);
 383         spin_lock_init(&pd_list->lock);
 384 }
 385
 386 /* Initialize all percpu queues used by serial workers */
 387 static void padata_init_squeues(struct parallel_data *pd)
 388 {
 389         int cpu;
 390         struct padata_serial_queue *squeue;
 391
 392         for_each_cpu(cpu, pd->cpumask.cbcpu) {
 393                 squeue = per_cpu_ptr(pd->squeue, cpu);
 394                 squeue->pd = pd;
 395                 __padata_list_init(&squeue->serial);
 396                 INIT_WORK(&squeue->work, padata_serial_worker);
 397         }
 398 }
 399
 400 /* Initialize all percpu queues used by parallel workers */
 401 static void padata_init_pqueues(struct parallel_data *pd)
 402 {
 403         int cpu;
 404         struct padata_parallel_queue *pqueue;
 405
 406         for_each_cpu(cpu, pd->cpumask.pcpu) {
 407                 pqueue = per_cpu_ptr(pd->pqueue, cpu);
 408
 409                 __padata_list_init(&pqueue->reorder);
 410                 __padata_list_init(&pqueue->parallel);
 411                 INIT_WORK(&pqueue->work, padata_parallel_worker);
 412                 atomic_set(&pqueue->num_obj, 0);
 413         }
 414 }
 415
 416 /* Allocate and initialize the internal cpumask dependend resources. */
 417 static struct parallel_data *padata_alloc_pd(struct padata_instance *pinst,
 418                                              const struct cpumask *pcpumask,
 419                                              const struct cpumask *cbcpumask)
 420 {
 421         struct parallel_data *pd;
 422
 423         pd = kzalloc(sizeof(struct parallel_data), GFP_KERNEL);
 424         if (!pd)
 425                 goto err;
 426
 427         pd->pqueue = alloc_percpu(struct padata_parallel_queue);
 428         if (!pd->pqueue)
 429                 goto err_free_pd;
 430
 431         pd->squeue = alloc_percpu(struct padata_serial_queue);
 432         if (!pd->squeue)
 433                 goto err_free_pqueue;
 434
 435         pd->pinst = pinst;
 436         if (padata_setup_cpumasks(pd, pcpumask, cbcpumask) < 0)
 437                 goto err_free_squeue;
 438
 439         padata_init_pqueues(pd);
 440         padata_init_squeues(pd);
 441         atomic_set(&pd->seq_nr, -1);
 442         atomic_set(&pd->reorder_objects, 0);
 443         atomic_set(&pd->refcnt, 0);
 444         spin_lock_init(&pd->lock);
 445         pd->cpu = cpumask_first(pd->cpumask.pcpu);
 446         INIT_WORK(&pd->reorder_work, invoke_padata_reorder);
 447
 448         return pd;
 449
 450 err_free_squeue:
 451         free_percpu(pd->squeue);
 452 err_free_pqueue:
 453         free_percpu(pd->pqueue);
 454 err_free_pd:
 455         kfree(pd);
 456 err:
 457         return NULL;
 458 }
 459
 460 static void padata_free_pd(struct parallel_data *pd)
 461 {
 462         free_cpumask_var(pd->cpumask.pcpu);
 463         free_cpumask_var(pd->cpumask.cbcpu);
 464         free_percpu(pd->pqueue);
 465         free_percpu(pd->squeue);
 466         kfree(pd);
 467 }
 468
 469 /* Flush all objects out of the padata queues. */
 470 static void padata_flush_queues(struct parallel_data *pd)
 471 {
 472         int cpu;
 473         struct padata_parallel_queue *pqueue;
 474         struct padata_serial_queue *squeue;
 475
 476         for_each_cpu(cpu, pd->cpumask.pcpu) {
 477                 pqueue = per_cpu_ptr(pd->pqueue, cpu);
 478                 flush_work(&pqueue->work);
 479         }
 480
 481         if (atomic_read(&pd->reorder_objects))
 482                 padata_reorder(pd);
 483
 484         for_each_cpu(cpu, pd->cpumask.cbcpu) {
 485                 squeue = per_cpu_ptr(pd->squeue, cpu);
 486                 flush_work(&squeue->work);
 487         }
 488
 489         BUG_ON(atomic_read(&pd->refcnt) != 0);
 490 }
 491
 492 static void __padata_start(struct padata_instance *pinst)
 493 {
 494         pinst->flags |= PADATA_INIT;
 495 }
 496
 497 static void __padata_stop(struct padata_instance *pinst)
 498 {
 499         if (!(pinst->flags & PADATA_INIT))
 500                 return;
 501
 502         pinst->flags &= ~PADATA_INIT;
 503
 504         synchronize_rcu();
 505
 506         get_online_cpus();
 507         padata_flush_queues(pinst->pd);
 508         put_online_cpus();
 509 }
 510
 511 /* Replace the internal control structure with a new one. */
 512 static void padata_replace(struct padata_instance *pinst,
 513                            struct parallel_data *pd_new)
 514 {
 515         struct parallel_data *pd_old = pinst->pd;
 516         int notification_mask = 0;
 517
 518         pinst->flags |= PADATA_RESET;
 519
 520         rcu_assign_pointer(pinst->pd, pd_new);
 521
 522         synchronize_rcu();
 523
 524         if (!cpumask_equal(pd_old->cpumask.pcpu, pd_new->cpumask.pcpu))
 525                 notification_mask |= PADATA_CPU_PARALLEL;
 526         if (!cpumask_equal(pd_old->cpumask.cbcpu, pd_new->cpumask.cbcpu))
 527                 notification_mask |= PADATA_CPU_SERIAL;
 528
 529         padata_flush_queues(pd_old);
 530         padata_free_pd(pd_old);
 531
 532         if (notification_mask)
 533                 blocking_notifier_call_chain(&pinst->cpumask_change_notifier,
 534                                              notification_mask,
 535                                              &pd_new->cpumask);
 536
 537         pinst->flags &= ~PADATA_RESET;
 538 }
 539
 540 /**
 541  * padata_register_cpumask_notifier - Registers a notifier that will be called
 542  *                             if either pcpu or cbcpu or both cpumasks change.
 543  *
 544  * @pinst: A poineter to padata instance
 545  * @nblock: A pointer to notifier block.
 546  */
 547 int padata_register_cpumask_notifier(struct padata_instance *pinst,
 548                                      struct notifier_block *nblock)
 549 {
 550         return blocking_notifier_chain_register(&pinst->cpumask_change_notifier,
 551                                                 nblock);
 552 }
 553 EXPORT_SYMBOL(padata_register_cpumask_notifier);
 554
 555 /**
 556  * padata_unregister_cpumask_notifier - Unregisters cpumask notifier
 557  *        registered earlier  using padata_register_cpumask_notifier
 558  *
 559  * @pinst: A pointer to data instance.
 560  * @nlock: A pointer to notifier block.
 561  */
 562 int padata_unregister_cpumask_notifier(struct padata_instance *pinst,
 563                                        struct notifier_block *nblock)
 564 {
 565         return blocking_notifier_chain_unregister(
 566                 &pinst->cpumask_change_notifier,
 567                 nblock);
 568 }
 569 EXPORT_SYMBOL(padata_unregister_cpumask_notifier);
 570
 571
 572 /* If cpumask contains no active cpu, we mark the instance as invalid. */
 573 static bool padata_validate_cpumask(struct padata_instance *pinst,
 574                                     const struct cpumask *cpumask)
 575 {
 576         if (!cpumask_intersects(cpumask, cpu_online_mask)) {
 577                 pinst->flags |= PADATA_INVALID;
 578                 return false;
 579         }
 580
 581         pinst->flags &= ~PADATA_INVALID;
 582         return true;
 583 }
 584
 585 static int __padata_set_cpumasks(struct padata_instance *pinst,
 586                                  cpumask_var_t pcpumask,
 587                                  cpumask_var_t cbcpumask)
 588 {
 589         int valid;
 590         struct parallel_data *pd;
 591
 592         valid = padata_validate_cpumask(pinst, pcpumask);
 593         if (!valid) {
 594                 __padata_stop(pinst);
 595                 goto out_replace;
 596         }
 597
 598         valid = padata_validate_cpumask(pinst, cbcpumask);
 599         if (!valid)
 600                 __padata_stop(pinst);
 601
 602 out_replace:
 603         pd = padata_alloc_pd(pinst, pcpumask, cbcpumask);
 604         if (!pd)
 605                 return -ENOMEM;
 606
 607         cpumask_copy(pinst->cpumask.pcpu, pcpumask);
 608         cpumask_copy(pinst->cpumask.cbcpu, cbcpumask);
 609
 610         padata_replace(pinst, pd);
 611
 612         if (valid)
 613                 __padata_start(pinst);
 614
 615         return 0;
 616 }
 617
 618 /**
 619  * padata_set_cpumask: Sets specified by @cpumask_type cpumask to the value
 620  *                     equivalent to @cpumask.
 621  *
 622  * @pinst: padata instance
 623  * @cpumask_type: PADATA_CPU_SERIAL or PADATA_CPU_PARALLEL corresponding
 624  *                to parallel and serial cpumasks respectively.
 625  * @cpumask: the cpumask to use
 626  */
 627 int padata_set_cpumask(struct padata_instance *pinst, int cpumask_type,
 628                        cpumask_var_t cpumask)
 629 {
 630         struct cpumask *serial_mask, *parallel_mask;
 631         int err = -EINVAL;
 632
 633         mutex_lock(&pinst->lock);
 634         get_online_cpus();
 635
 636         switch (cpumask_type) {
 637         case PADATA_CPU_PARALLEL:
 638                 serial_mask = pinst->cpumask.cbcpu;
 639                 parallel_mask = cpumask;
 640                 break;
 641         case PADATA_CPU_SERIAL:
 642                 parallel_mask = pinst->cpumask.pcpu;
 643                 serial_mask = cpumask;
 644                 break;
 645         default:
 646                  goto out;
 647         }
 648
 649         err =  __padata_set_cpumasks(pinst, parallel_mask, serial_mask);
 650
 651 out:
 652         put_online_cpus();
 653         mutex_unlock(&pinst->lock);
 654
 655         return err;
 656 }
 657 EXPORT_SYMBOL(padata_set_cpumask);
 658
 659 /**
 660  * padata_start - start the parallel processing
 661  *
 662  * @pinst: padata instance to start
 663  */
 664 int padata_start(struct padata_instance *pinst)
 665 {
 666         int err = 0;
 667
 668         mutex_lock(&pinst->lock);
 669
 670         if (pinst->flags & PADATA_INVALID)
 671                 err = -EINVAL;
 672
 673         __padata_start(pinst);
 674
 675         mutex_unlock(&pinst->lock);
 676
 677         return err;
 678 }
 679 EXPORT_SYMBOL(padata_start);
 680
 681 /**
 682  * padata_stop - stop the parallel processing
 683  *
 684  * @pinst: padata instance to stop
 685  */
 686 void padata_stop(struct padata_instance *pinst)
 687 {
 688         mutex_lock(&pinst->lock);
 689         __padata_stop(pinst);
 690         mutex_unlock(&pinst->lock);
 691 }
 692 EXPORT_SYMBOL(padata_stop);
 693
 694 #ifdef CONFIG_HOTPLUG_CPU
 695
 696 static int __padata_add_cpu(struct padata_instance *pinst, int cpu)
 697 {
 698         struct parallel_data *pd;
 699
 700         if (cpumask_test_cpu(cpu, cpu_online_mask)) {
 701                 pd = padata_alloc_pd(pinst, pinst->cpumask.pcpu,
 702                                      pinst->cpumask.cbcpu);
 703                 if (!pd)
 704                         return -ENOMEM;
 705
 706                 padata_replace(pinst, pd);
 707
 708                 if (padata_validate_cpumask(pinst, pinst->cpumask.pcpu) &&
 709                     padata_validate_cpumask(pinst, pinst->cpumask.cbcpu))
 710                         __padata_start(pinst);
 711         }
 712
 713         return 0;
 714 }
 715
 716 static int __padata_remove_cpu(struct padata_instance *pinst, int cpu)
 717 {
 718         struct parallel_data *pd = NULL;
 719
 720         if (cpumask_test_cpu(cpu, cpu_online_mask)) {
 721
 722                 if (!padata_validate_cpumask(pinst, pinst->cpumask.pcpu) ||
 723                     !padata_validate_cpumask(pinst, pinst->cpumask.cbcpu))
 724                         __padata_stop(pinst);
 725
 726                 pd = padata_alloc_pd(pinst, pinst->cpumask.pcpu,
 727                                      pinst->cpumask.cbcpu);
 728                 if (!pd)
 729                         return -ENOMEM;
 730
 731                 padata_replace(pinst, pd);
 732
 733                 cpumask_clear_cpu(cpu, pd->cpumask.cbcpu);
 734                 cpumask_clear_cpu(cpu, pd->cpumask.pcpu);
 735         }
 736
 737         return 0;
 738 }
 739
 740  /**
 741  * padata_remove_cpu - remove a cpu from the one or both(serial and parallel)
 742  *                     padata cpumasks.
 743  *
 744  * @pinst: padata instance
 745  * @cpu: cpu to remove
 746  * @mask: bitmask specifying from which cpumask @cpu should be removed
 747  *        The @mask may be any combination of the following flags:
 748  *          PADATA_CPU_SERIAL   - serial cpumask
 749  *          PADATA_CPU_PARALLEL - parallel cpumask
 750  */
 751 int padata_remove_cpu(struct padata_instance *pinst, int cpu, int mask)
 752 {
 753         int err;
 754
 755         if (!(mask & (PADATA_CPU_SERIAL | PADATA_CPU_PARALLEL)))
 756                 return -EINVAL;
 757
 758         mutex_lock(&pinst->lock);
 759
 760         get_online_cpus();
 761         if (mask & PADATA_CPU_SERIAL)
 762                 cpumask_clear_cpu(cpu, pinst->cpumask.cbcpu);
 763         if (mask & PADATA_CPU_PARALLEL)
 764                 cpumask_clear_cpu(cpu, pinst->cpumask.pcpu);
 765
 766         err = __padata_remove_cpu(pinst, cpu);
 767         put_online_cpus();
 768
 769         mutex_unlock(&pinst->lock);
 770
 771         return err;
 772 }
 773 EXPORT_SYMBOL(padata_remove_cpu);
 774
 775 static inline int pinst_has_cpu(struct padata_instance *pinst, int cpu)
 776 {
 777         return cpumask_test_cpu(cpu, pinst->cpumask.pcpu) ||
 778                 cpumask_test_cpu(cpu, pinst->cpumask.cbcpu);
 779 }
 780
 781 static int padata_cpu_online(unsigned int cpu, struct hlist_node *node)
 782 {
 783         struct padata_instance *pinst;
 784         int ret;
 785
 786         pinst = hlist_entry_safe(node, struct padata_instance, node);
 787         if (!pinst_has_cpu(pinst, cpu))
 788                 return 0;
 789
 790         mutex_lock(&pinst->lock);
 791         ret = __padata_add_cpu(pinst, cpu);
 792         mutex_unlock(&pinst->lock);
 793         return ret;
 794 }
 795
 796 static int padata_cpu_prep_down(unsigned int cpu, struct hlist_node *node)
 797 {
 798         struct padata_instance *pinst;
 799         int ret;
 800
 801         pinst = hlist_entry_safe(node, struct padata_instance, node);
 802         if (!pinst_has_cpu(pinst, cpu))
 803                 return 0;
 804
 805         mutex_lock(&pinst->lock);
 806         ret = __padata_remove_cpu(pinst, cpu);
 807         mutex_unlock(&pinst->lock);
 808         return ret;
 809 }
 810
 811 static enum cpuhp_state hp_online;
 812 #endif
 813
 814 static void __padata_free(struct padata_instance *pinst)
 815 {
 816 #ifdef CONFIG_HOTPLUG_CPU
 817         cpuhp_state_remove_instance_nocalls(hp_online, &pinst->node);
 818 #endif
 819
 820         padata_stop(pinst);
 821         padata_free_pd(pinst->pd);
 822         free_cpumask_var(pinst->cpumask.pcpu);
 823         free_cpumask_var(pinst->cpumask.cbcpu);
 824         destroy_workqueue(pinst->serial_wq);
 825         destroy_workqueue(pinst->parallel_wq);
 826         kfree(pinst);
 827 }
 828
 829 #define kobj2pinst(_kobj)                                       \
 830         container_of(_kobj, struct padata_instance, kobj)
 831 #define attr2pentry(_attr)                                      \
 832         container_of(_attr, struct padata_sysfs_entry, attr)
 833
 834 static void padata_sysfs_release(struct kobject *kobj)
 835 {
 836         struct padata_instance *pinst = kobj2pinst(kobj);
 837         __padata_free(pinst);
 838 }
 839
 840 struct padata_sysfs_entry {
 841         struct attribute attr;
 842         ssize_t (*show)(struct padata_instance *, struct attribute *, char *);
 843         ssize_t (*store)(struct padata_instance *, struct attribute *,
 844                          const char *, size_t);
 845 };
 846
 847 static ssize_t show_cpumask(struct padata_instance *pinst,
 848                             struct attribute *attr,  char *buf)
 849 {
 850         struct cpumask *cpumask;
 851         ssize_t len;
 852
 853         mutex_lock(&pinst->lock);
 854         if (!strcmp(attr->name, "serial_cpumask"))
 855                 cpumask = pinst->cpumask.cbcpu;
 856         else
 857                 cpumask = pinst->cpumask.pcpu;
 858
 859         len = snprintf(buf, PAGE_SIZE, "%*pb\n",
 860                        nr_cpu_ids, cpumask_bits(cpumask));
 861         mutex_unlock(&pinst->lock);
 862         return len < PAGE_SIZE ? len : -EINVAL;
 863 }
 864
 865 static ssize_t store_cpumask(struct padata_instance *pinst,
 866                              struct attribute *attr,
 867                              const char *buf, size_t count)
 868 {
 869         cpumask_var_t new_cpumask;
 870         ssize_t ret;
 871         int mask_type;
 872
 873         if (!alloc_cpumask_var(&new_cpumask, GFP_KERNEL))
 874                 return -ENOMEM;
 875
 876         ret = bitmap_parse(buf, count, cpumask_bits(new_cpumask),
 877                            nr_cpumask_bits);
 878         if (ret < 0)
 879                 goto out;
 880
 881         mask_type = !strcmp(attr->name, "serial_cpumask") ?
 882                 PADATA_CPU_SERIAL : PADATA_CPU_PARALLEL;
 883         ret = padata_set_cpumask(pinst, mask_type, new_cpumask);
 884         if (!ret)
 885                 ret = count;
 886
 887 out:
 888         free_cpumask_var(new_cpumask);
 889         return ret;
 890 }
 891
 892 #define PADATA_ATTR_RW(_name, _show_name, _store_name)          \
 893         static struct padata_sysfs_entry _name##_attr =         \
 894                 __ATTR(_name, 0644, _show_name, _store_name)
 895 #define PADATA_ATTR_RO(_name, _show_name)               \
 896         static struct padata_sysfs_entry _name##_attr = \
 897                 __ATTR(_name, 0400, _show_name, NULL)
 898
 899 PADATA_ATTR_RW(serial_cpumask, show_cpumask, store_cpumask);
 900 PADATA_ATTR_RW(parallel_cpumask, show_cpumask, store_cpumask);
 901
 902 /*
 903  * Padata sysfs provides the following objects:
 904  * serial_cpumask   [RW] - cpumask for serial workers
 905  * parallel_cpumask [RW] - cpumask for parallel workers
 906  */
 907 static struct attribute *padata_default_attrs[] = {
 908         &serial_cpumask_attr.attr,
 909         &parallel_cpumask_attr.attr,
 910         NULL,
 911 };
 912 ATTRIBUTE_GROUPS(padata_default);
 913
 914 static ssize_t padata_sysfs_show(struct kobject *kobj,
 915                                  struct attribute *attr, char *buf)
 916 {
 917         struct padata_instance *pinst;
 918         struct padata_sysfs_entry *pentry;
 919         ssize_t ret = -EIO;
 920
 921         pinst = kobj2pinst(kobj);
 922         pentry = attr2pentry(attr);
 923         if (pentry->show)
 924                 ret = pentry->show(pinst, attr, buf);
 925
 926         return ret;
 927 }
 928
 929 static ssize_t padata_sysfs_store(struct kobject *kobj, struct attribute *attr,
 930                                   const char *buf, size_t count)
 931 {
 932         struct padata_instance *pinst;
 933         struct padata_sysfs_entry *pentry;
 934         ssize_t ret = -EIO;
 935
 936         pinst = kobj2pinst(kobj);
 937         pentry = attr2pentry(attr);
 938         if (pentry->show)
 939                 ret = pentry->store(pinst, attr, buf, count);
 940
 941         return ret;
 942 }
 943
 944 static const struct sysfs_ops padata_sysfs_ops = {
 945         .show = padata_sysfs_show,
 946         .store = padata_sysfs_store,
 947 };
 948
 949 static struct kobj_type padata_attr_type = {
 950         .sysfs_ops = &padata_sysfs_ops,
 951         .default_groups = padata_default_groups,
 952         .release = padata_sysfs_release,
 953 };
 954
 955 /**
 956  * padata_alloc - allocate and initialize a padata instance and specify
 957  *                cpumasks for serial and parallel workers.
 958  *
 959  * @name: used to identify the instance
 960  * @pcpumask: cpumask that will be used for padata parallelization
 961  * @cbcpumask: cpumask that will be used for padata serialization
 962  */
 963 static struct padata_instance *padata_alloc(const char *name,
 964                                             const struct cpumask *pcpumask,
 965                                             const struct cpumask *cbcpumask)
 966 {
 967         struct padata_instance *pinst;
 968         struct parallel_data *pd = NULL;
 969
 970         pinst = kzalloc(sizeof(struct padata_instance), GFP_KERNEL);
 971         if (!pinst)
 972                 goto err;
 973
 974         pinst->parallel_wq = alloc_workqueue("%s_parallel", WQ_UNBOUND, 0,
 975                                              name);
 976         if (!pinst->parallel_wq)
 977                 goto err_free_inst;
 978
 979         get_online_cpus();
 980
 981         pinst->serial_wq = alloc_workqueue("%s_serial", WQ_MEM_RECLAIM |
 982                                            WQ_CPU_INTENSIVE, 1, name);
 983         if (!pinst->serial_wq)
 984                 goto err_put_cpus;
 985
 986         if (!alloc_cpumask_var(&pinst->cpumask.pcpu, GFP_KERNEL))
 987                 goto err_free_serial_wq;
 988         if (!alloc_cpumask_var(&pinst->cpumask.cbcpu, GFP_KERNEL)) {
 989                 free_cpumask_var(pinst->cpumask.pcpu);
 990                 goto err_free_serial_wq;
 991         }
 992         if (!padata_validate_cpumask(pinst, pcpumask) ||
 993             !padata_validate_cpumask(pinst, cbcpumask))
 994                 goto err_free_masks;
 995
 996         pd = padata_alloc_pd(pinst, pcpumask, cbcpumask);
 997         if (!pd)
 998                 goto err_free_masks;
 999
1000         rcu_assign_pointer(pinst->pd, pd);
1001
1002         cpumask_copy(pinst->cpumask.pcpu, pcpumask);
1003         cpumask_copy(pinst->cpumask.cbcpu, cbcpumask);
1004
1005         pinst->flags = 0;
1006
1007         BLOCKING_INIT_NOTIFIER_HEAD(&pinst->cpumask_change_notifier);
1008         kobject_init(&pinst->kobj, &padata_attr_type);
1009         mutex_init(&pinst->lock);
1010
1011 #ifdef CONFIG_HOTPLUG_CPU
1012         cpuhp_state_add_instance_nocalls_cpuslocked(hp_online, &pinst->node);
1013 #endif
1014
1015         put_online_cpus();
1016
1017         return pinst;
1018
1019 err_free_masks:
1020         free_cpumask_var(pinst->cpumask.pcpu);
1021         free_cpumask_var(pinst->cpumask.cbcpu);
1022 err_free_serial_wq:
1023         destroy_workqueue(pinst->serial_wq);
1024 err_put_cpus:
1025         put_online_cpus();
1026         destroy_workqueue(pinst->parallel_wq);
1027 err_free_inst:
1028         kfree(pinst);
1029 err:
1030         return NULL;
1031 }
1032
1033 /**
1034  * padata_alloc_possible - Allocate and initialize padata instance.
1035  *                         Use the cpu_possible_mask for serial and
1036  *                         parallel workers.
1037  *
1038  * @name: used to identify the instance
1039  */
1040 struct padata_instance *padata_alloc_possible(const char *name)
1041 {
1042         return padata_alloc(name, cpu_possible_mask, cpu_possible_mask);
1043 }
1044 EXPORT_SYMBOL(padata_alloc_possible);
1045
1046 /**
1047  * padata_free - free a padata instance
1048  *
1049  * @padata_inst: padata instance to free
1050  */
1051 void padata_free(struct padata_instance *pinst)
1052 {
1053         kobject_put(&pinst->kobj);
1054 }
1055 EXPORT_SYMBOL(padata_free);
1056
1057 #ifdef CONFIG_HOTPLUG_CPU
1058
1059 static __init int padata_driver_init(void)
1060 {
1061         int ret;
1062
1063         ret = cpuhp_setup_state_multi(CPUHP_AP_ONLINE_DYN, "padata:online",
1064                                       padata_cpu_online,
1065                                       padata_cpu_prep_down);
1066         if (ret < 0)
1067                 return ret;
1068         hp_online = ret;
1069         return 0;
1070 }
1071 module_init(padata_driver_init);
1072
1073 static __exit void padata_driver_exit(void)
1074 {
1075         cpuhp_remove_multi_state(hp_online);
1076 }
1077 module_exit(padata_driver_exit);
1078 #endif