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