kernel/padata.c

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