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/rcupdate.h>
  30
  31 #define MAX_SEQ_NR INT_MAX - NR_CPUS
  32 #define MAX_OBJ_NUM 1000
  33
  34 static int padata_index_to_cpu(struct parallel_data *pd, int cpu_index)
  35 {
  36         int cpu, target_cpu;
  37
  38         target_cpu = cpumask_first(pd->cpumask);
  39         for (cpu = 0; cpu < cpu_index; cpu++)
  40                 target_cpu = cpumask_next(target_cpu, pd->cpumask);
  41
  42         return target_cpu;
  43 }
  44
  45 static int padata_cpu_hash(struct padata_priv *padata)
  46 {
  47         int cpu_index;
  48         struct parallel_data *pd;
  49
  50         pd =  padata->pd;
  51
  52         /*
  53          * Hash the sequence numbers to the cpus by taking
  54          * seq_nr mod. number of cpus in use.
  55          */
  56         cpu_index =  padata->seq_nr % cpumask_weight(pd->cpumask);
  57
  58         return padata_index_to_cpu(pd, cpu_index);
  59 }
  60
  61 static void padata_parallel_worker(struct work_struct *work)
  62 {
  63         struct padata_queue *queue;
  64         struct parallel_data *pd;
  65         struct padata_instance *pinst;
  66         LIST_HEAD(local_list);
  67
  68         local_bh_disable();
  69         queue = container_of(work, struct padata_queue, pwork);
  70         pd = queue->pd;
  71         pinst = pd->pinst;
  72
  73         spin_lock(&queue->parallel.lock);
  74         list_replace_init(&queue->parallel.list, &local_list);
  75         spin_unlock(&queue->parallel.lock);
  76
  77         while (!list_empty(&local_list)) {
  78                 struct padata_priv *padata;
  79
  80                 padata = list_entry(local_list.next,
  81                                     struct padata_priv, list);
  82
  83                 list_del_init(&padata->list);
  84
  85                 padata->parallel(padata);
  86         }
  87
  88         local_bh_enable();
  89 }
  90
  91 /*
  92  * padata_do_parallel - padata parallelization function
  93  *
  94  * @pinst: padata instance
  95  * @padata: object to be parallelized
  96  * @cb_cpu: cpu the serialization callback function will run on,
  97  *          must be in the cpumask of padata.
  98  *
  99  * The parallelization callback function will run with BHs off.
 100  * Note: Every object which is parallelized by padata_do_parallel
 101  * must be seen by padata_do_serial.
 102  */
 103 int padata_do_parallel(struct padata_instance *pinst,
 104                        struct padata_priv *padata, int cb_cpu)
 105 {
 106         int target_cpu, err;
 107         struct padata_queue *queue;
 108         struct parallel_data *pd;
 109
 110         rcu_read_lock_bh();
 111
 112         pd = rcu_dereference(pinst->pd);
 113
 114         err = 0;
 115         if (!(pinst->flags & PADATA_INIT))
 116                 goto out;
 117
 118         err =  -EBUSY;
 119         if ((pinst->flags & PADATA_RESET))
 120                 goto out;
 121
 122         if (atomic_read(&pd->refcnt) >= MAX_OBJ_NUM)
 123                 goto out;
 124
 125         err = -EINVAL;
 126         if (!cpumask_test_cpu(cb_cpu, pd->cpumask))
 127                 goto out;
 128
 129         err = -EINPROGRESS;
 130         atomic_inc(&pd->refcnt);
 131         padata->pd = pd;
 132         padata->cb_cpu = cb_cpu;
 133
 134         if (unlikely(atomic_read(&pd->seq_nr) == pd->max_seq_nr))
 135                 atomic_set(&pd->seq_nr, -1);
 136
 137         padata->seq_nr = atomic_inc_return(&pd->seq_nr);
 138
 139         target_cpu = padata_cpu_hash(padata);
 140         queue = per_cpu_ptr(pd->queue, target_cpu);
 141
 142         spin_lock(&queue->parallel.lock);
 143         list_add_tail(&padata->list, &queue->parallel.list);
 144         spin_unlock(&queue->parallel.lock);
 145
 146         queue_work_on(target_cpu, pinst->wq, &queue->pwork);
 147
 148 out:
 149         rcu_read_unlock_bh();
 150
 151         return err;
 152 }
 153 EXPORT_SYMBOL(padata_do_parallel);
 154
 155 static struct padata_priv *padata_get_next(struct parallel_data *pd)
 156 {
 157         int cpu, num_cpus, empty, calc_seq_nr;
 158         int seq_nr, next_nr, overrun, next_overrun;
 159         struct padata_queue *queue, *next_queue;
 160         struct padata_priv *padata;
 161         struct padata_list *reorder;
 162
 163         empty = 0;
 164         next_nr = -1;
 165         next_overrun = 0;
 166         next_queue = NULL;
 167
 168         num_cpus = cpumask_weight(pd->cpumask);
 169
 170         for_each_cpu(cpu, pd->cpumask) {
 171                 queue = per_cpu_ptr(pd->queue, cpu);
 172                 reorder = &queue->reorder;
 173
 174                 /*
 175                  * Calculate the seq_nr of the object that should be
 176                  * next in this queue.
 177                  */
 178                 overrun = 0;
 179                 calc_seq_nr = (atomic_read(&queue->num_obj) * num_cpus)
 180                                + queue->cpu_index;
 181
 182                 if (unlikely(calc_seq_nr > pd->max_seq_nr)) {
 183                         calc_seq_nr = calc_seq_nr - pd->max_seq_nr - 1;
 184                         overrun = 1;
 185                 }
 186
 187                 if (!list_empty(&reorder->list)) {
 188                         padata = list_entry(reorder->list.next,
 189                                             struct padata_priv, list);
 190
 191                         seq_nr  = padata->seq_nr;
 192                         BUG_ON(calc_seq_nr != seq_nr);
 193                 } else {
 194                         seq_nr = calc_seq_nr;
 195                         empty++;
 196                 }
 197
 198                 if (next_nr < 0 || seq_nr < next_nr
 199                     || (next_overrun && !overrun)) {
 200                         next_nr = seq_nr;
 201                         next_overrun = overrun;
 202                         next_queue = queue;
 203                 }
 204         }
 205
 206         padata = NULL;
 207
 208         if (empty == num_cpus)
 209                 goto out;
 210
 211         reorder = &next_queue->reorder;
 212
 213         if (!list_empty(&reorder->list)) {
 214                 padata = list_entry(reorder->list.next,
 215                                     struct padata_priv, list);
 216
 217                 if (unlikely(next_overrun)) {
 218                         for_each_cpu(cpu, pd->cpumask) {
 219                                 queue = per_cpu_ptr(pd->queue, cpu);
 220                                 atomic_set(&queue->num_obj, 0);
 221                         }
 222                 }
 223
 224                 spin_lock(&reorder->lock);
 225                 list_del_init(&padata->list);
 226                 atomic_dec(&pd->reorder_objects);
 227                 spin_unlock(&reorder->lock);
 228
 229                 atomic_inc(&next_queue->num_obj);
 230
 231                 goto out;
 232         }
 233
 234         queue = per_cpu_ptr(pd->queue, smp_processor_id());
 235         if (queue->cpu_index == next_queue->cpu_index) {
 236                 padata = ERR_PTR(-ENODATA);
 237                 goto out;
 238         }
 239
 240         padata = ERR_PTR(-EINPROGRESS);
 241 out:
 242         return padata;
 243 }
 244
 245 static void padata_reorder(struct parallel_data *pd)
 246 {
 247         struct padata_priv *padata;
 248         struct padata_queue *queue;
 249         struct padata_instance *pinst = pd->pinst;
 250
 251         if (!spin_trylock_bh(&pd->lock))
 252                 return;
 253
 254         while (1) {
 255                 padata = padata_get_next(pd);
 256
 257                 if (!padata || PTR_ERR(padata) == -EINPROGRESS)
 258                         break;
 259
 260                 if (PTR_ERR(padata) == -ENODATA) {
 261                         del_timer(&pd->timer);
 262                         spin_unlock_bh(&pd->lock);
 263                         return;
 264                 }
 265
 266                 queue = per_cpu_ptr(pd->queue, padata->cb_cpu);
 267
 268                 spin_lock(&queue->serial.lock);
 269                 list_add_tail(&padata->list, &queue->serial.list);
 270                 spin_unlock(&queue->serial.lock);
 271
 272                 queue_work_on(padata->cb_cpu, pinst->wq, &queue->swork);
 273         }
 274
 275         spin_unlock_bh(&pd->lock);
 276
 277         if (atomic_read(&pd->reorder_objects)
 278                         && !(pinst->flags & PADATA_RESET))
 279                 mod_timer(&pd->timer, jiffies + HZ);
 280         else
 281                 del_timer(&pd->timer);
 282
 283         return;
 284 }
 285
 286 static void padata_reorder_timer(unsigned long arg)
 287 {
 288         struct parallel_data *pd = (struct parallel_data *)arg;
 289
 290         padata_reorder(pd);
 291 }
 292
 293 static void padata_serial_worker(struct work_struct *work)
 294 {
 295         struct padata_queue *queue;
 296         struct parallel_data *pd;
 297         LIST_HEAD(local_list);
 298
 299         local_bh_disable();
 300         queue = container_of(work, struct padata_queue, swork);
 301         pd = queue->pd;
 302
 303         spin_lock(&queue->serial.lock);
 304         list_replace_init(&queue->serial.list, &local_list);
 305         spin_unlock(&queue->serial.lock);
 306
 307         while (!list_empty(&local_list)) {
 308                 struct padata_priv *padata;
 309
 310                 padata = list_entry(local_list.next,
 311                                     struct padata_priv, list);
 312
 313                 list_del_init(&padata->list);
 314
 315                 padata->serial(padata);
 316                 atomic_dec(&pd->refcnt);
 317         }
 318         local_bh_enable();
 319 }
 320
 321 /*
 322  * padata_do_serial - padata serialization function
 323  *
 324  * @padata: object to be serialized.
 325  *
 326  * padata_do_serial must be called for every parallelized object.
 327  * The serialization callback function will run with BHs off.
 328  */
 329 void padata_do_serial(struct padata_priv *padata)
 330 {
 331         int cpu;
 332         struct padata_queue *queue;
 333         struct parallel_data *pd;
 334
 335         pd = padata->pd;
 336
 337         cpu = get_cpu();
 338         queue = per_cpu_ptr(pd->queue, cpu);
 339
 340         spin_lock(&queue->reorder.lock);
 341         atomic_inc(&pd->reorder_objects);
 342         list_add_tail(&padata->list, &queue->reorder.list);
 343         spin_unlock(&queue->reorder.lock);
 344
 345         put_cpu();
 346
 347         padata_reorder(pd);
 348 }
 349 EXPORT_SYMBOL(padata_do_serial);
 350
 351 static struct parallel_data *padata_alloc_pd(struct padata_instance *pinst,
 352                                              const struct cpumask *cpumask)
 353 {
 354         int cpu, cpu_index, num_cpus;
 355         struct padata_queue *queue;
 356         struct parallel_data *pd;
 357
 358         cpu_index = 0;
 359
 360         pd = kzalloc(sizeof(struct parallel_data), GFP_KERNEL);
 361         if (!pd)
 362                 goto err;
 363
 364         pd->queue = alloc_percpu(struct padata_queue);
 365         if (!pd->queue)
 366                 goto err_free_pd;
 367
 368         if (!alloc_cpumask_var(&pd->cpumask, GFP_KERNEL))
 369                 goto err_free_queue;
 370
 371         cpumask_and(pd->cpumask, cpumask, cpu_active_mask);
 372
 373         for_each_cpu(cpu, pd->cpumask) {
 374                 queue = per_cpu_ptr(pd->queue, cpu);
 375
 376                 queue->pd = pd;
 377
 378                 queue->cpu_index = cpu_index;
 379                 cpu_index++;
 380
 381                 INIT_LIST_HEAD(&queue->reorder.list);
 382                 INIT_LIST_HEAD(&queue->parallel.list);
 383                 INIT_LIST_HEAD(&queue->serial.list);
 384                 spin_lock_init(&queue->reorder.lock);
 385                 spin_lock_init(&queue->parallel.lock);
 386                 spin_lock_init(&queue->serial.lock);
 387
 388                 INIT_WORK(&queue->pwork, padata_parallel_worker);
 389                 INIT_WORK(&queue->swork, padata_serial_worker);
 390                 atomic_set(&queue->num_obj, 0);
 391         }
 392
 393         num_cpus = cpumask_weight(pd->cpumask);
 394         pd->max_seq_nr = (MAX_SEQ_NR / num_cpus) * num_cpus - 1;
 395
 396         setup_timer(&pd->timer, padata_reorder_timer, (unsigned long)pd);
 397         atomic_set(&pd->seq_nr, -1);
 398         atomic_set(&pd->reorder_objects, 0);
 399         atomic_set(&pd->refcnt, 0);
 400         pd->pinst = pinst;
 401         spin_lock_init(&pd->lock);
 402
 403         return pd;
 404
 405 err_free_queue:
 406         free_percpu(pd->queue);
 407 err_free_pd:
 408         kfree(pd);
 409 err:
 410         return NULL;
 411 }
 412
 413 static void padata_free_pd(struct parallel_data *pd)
 414 {
 415         free_cpumask_var(pd->cpumask);
 416         free_percpu(pd->queue);
 417         kfree(pd);
 418 }
 419
 420 static void padata_flush_queues(struct parallel_data *pd)
 421 {
 422         int cpu;
 423         struct padata_queue *queue;
 424
 425         for_each_cpu(cpu, pd->cpumask) {
 426                 queue = per_cpu_ptr(pd->queue, cpu);
 427                 flush_work(&queue->pwork);
 428         }
 429
 430         del_timer_sync(&pd->timer);
 431
 432         if (atomic_read(&pd->reorder_objects))
 433                 padata_reorder(pd);
 434
 435         for_each_cpu(cpu, pd->cpumask) {
 436                 queue = per_cpu_ptr(pd->queue, cpu);
 437                 flush_work(&queue->swork);
 438         }
 439
 440         BUG_ON(atomic_read(&pd->refcnt) != 0);
 441 }
 442
 443 static void padata_replace(struct padata_instance *pinst,
 444                            struct parallel_data *pd_new)
 445 {
 446         struct parallel_data *pd_old = pinst->pd;
 447
 448         pinst->flags |= PADATA_RESET;
 449
 450         rcu_assign_pointer(pinst->pd, pd_new);
 451
 452         synchronize_rcu();
 453
 454         padata_flush_queues(pd_old);
 455         padata_free_pd(pd_old);
 456
 457         pinst->flags &= ~PADATA_RESET;
 458 }
 459
 460 /*
 461  * padata_set_cpumask - set the cpumask that padata should use
 462  *
 463  * @pinst: padata instance
 464  * @cpumask: the cpumask to use
 465  */
 466 int padata_set_cpumask(struct padata_instance *pinst,
 467                         cpumask_var_t cpumask)
 468 {
 469         struct parallel_data *pd;
 470         int err = 0;
 471
 472         mutex_lock(&pinst->lock);
 473
 474         get_online_cpus();
 475
 476         pd = padata_alloc_pd(pinst, cpumask);
 477         if (!pd) {
 478                 err = -ENOMEM;
 479                 goto out;
 480         }
 481
 482         cpumask_copy(pinst->cpumask, cpumask);
 483
 484         padata_replace(pinst, pd);
 485
 486 out:
 487         put_online_cpus();
 488
 489         mutex_unlock(&pinst->lock);
 490
 491         return err;
 492 }
 493 EXPORT_SYMBOL(padata_set_cpumask);
 494
 495 static int __padata_add_cpu(struct padata_instance *pinst, int cpu)
 496 {
 497         struct parallel_data *pd;
 498
 499         if (cpumask_test_cpu(cpu, cpu_active_mask)) {
 500                 pd = padata_alloc_pd(pinst, pinst->cpumask);
 501                 if (!pd)
 502                         return -ENOMEM;
 503
 504                 padata_replace(pinst, pd);
 505         }
 506
 507         return 0;
 508 }
 509
 510 /*
 511  * padata_add_cpu - add a cpu to the padata cpumask
 512  *
 513  * @pinst: padata instance
 514  * @cpu: cpu to add
 515  */
 516 int padata_add_cpu(struct padata_instance *pinst, int cpu)
 517 {
 518         int err;
 519
 520         mutex_lock(&pinst->lock);
 521
 522         get_online_cpus();
 523         cpumask_set_cpu(cpu, pinst->cpumask);
 524         err = __padata_add_cpu(pinst, cpu);
 525         put_online_cpus();
 526
 527         mutex_unlock(&pinst->lock);
 528
 529         return err;
 530 }
 531 EXPORT_SYMBOL(padata_add_cpu);
 532
 533 static int __padata_remove_cpu(struct padata_instance *pinst, int cpu)
 534 {
 535         struct parallel_data *pd;
 536
 537         if (cpumask_test_cpu(cpu, cpu_online_mask)) {
 538                 pd = padata_alloc_pd(pinst, pinst->cpumask);
 539                 if (!pd)
 540                         return -ENOMEM;
 541
 542                 padata_replace(pinst, pd);
 543         }
 544
 545         return 0;
 546 }
 547
 548 /*
 549  * padata_remove_cpu - remove a cpu from the padata cpumask
 550  *
 551  * @pinst: padata instance
 552  * @cpu: cpu to remove
 553  */
 554 int padata_remove_cpu(struct padata_instance *pinst, int cpu)
 555 {
 556         int err;
 557
 558         mutex_lock(&pinst->lock);
 559
 560         get_online_cpus();
 561         cpumask_clear_cpu(cpu, pinst->cpumask);
 562         err = __padata_remove_cpu(pinst, cpu);
 563         put_online_cpus();
 564
 565         mutex_unlock(&pinst->lock);
 566
 567         return err;
 568 }
 569 EXPORT_SYMBOL(padata_remove_cpu);
 570
 571 /*
 572  * padata_start - start the parallel processing
 573  *
 574  * @pinst: padata instance to start
 575  */
 576 void padata_start(struct padata_instance *pinst)
 577 {
 578         mutex_lock(&pinst->lock);
 579         pinst->flags |= PADATA_INIT;
 580         mutex_unlock(&pinst->lock);
 581 }
 582 EXPORT_SYMBOL(padata_start);
 583
 584 /*
 585  * padata_stop - stop the parallel processing
 586  *
 587  * @pinst: padata instance to stop
 588  */
 589 void padata_stop(struct padata_instance *pinst)
 590 {
 591         mutex_lock(&pinst->lock);
 592         pinst->flags &= ~PADATA_INIT;
 593         mutex_unlock(&pinst->lock);
 594 }
 595 EXPORT_SYMBOL(padata_stop);
 596
 597 #ifdef CONFIG_HOTPLUG_CPU
 598 static int padata_cpu_callback(struct notifier_block *nfb,
 599                                unsigned long action, void *hcpu)
 600 {
 601         int err;
 602         struct padata_instance *pinst;
 603         int cpu = (unsigned long)hcpu;
 604
 605         pinst = container_of(nfb, struct padata_instance, cpu_notifier);
 606
 607         switch (action) {
 608         case CPU_ONLINE:
 609         case CPU_ONLINE_FROZEN:
 610                 if (!cpumask_test_cpu(cpu, pinst->cpumask))
 611                         break;
 612                 mutex_lock(&pinst->lock);
 613                 err = __padata_add_cpu(pinst, cpu);
 614                 mutex_unlock(&pinst->lock);
 615                 if (err)
 616                         return NOTIFY_BAD;
 617                 break;
 618
 619         case CPU_DOWN_PREPARE:
 620         case CPU_DOWN_PREPARE_FROZEN:
 621                 if (!cpumask_test_cpu(cpu, pinst->cpumask))
 622                         break;
 623                 mutex_lock(&pinst->lock);
 624                 err = __padata_remove_cpu(pinst, cpu);
 625                 mutex_unlock(&pinst->lock);
 626                 if (err)
 627                         return NOTIFY_BAD;
 628                 break;
 629
 630         case CPU_UP_CANCELED:
 631         case CPU_UP_CANCELED_FROZEN:
 632                 if (!cpumask_test_cpu(cpu, pinst->cpumask))
 633                         break;
 634                 mutex_lock(&pinst->lock);
 635                 __padata_remove_cpu(pinst, cpu);
 636                 mutex_unlock(&pinst->lock);
 637
 638         case CPU_DOWN_FAILED:
 639         case CPU_DOWN_FAILED_FROZEN:
 640                 if (!cpumask_test_cpu(cpu, pinst->cpumask))
 641                         break;
 642                 mutex_lock(&pinst->lock);
 643                 __padata_add_cpu(pinst, cpu);
 644                 mutex_unlock(&pinst->lock);
 645         }
 646
 647         return NOTIFY_OK;
 648 }
 649 #endif
 650
 651 /*
 652  * padata_alloc - allocate and initialize a padata instance
 653  *
 654  * @cpumask: cpumask that padata uses for parallelization
 655  * @wq: workqueue to use for the allocated padata instance
 656  */
 657 struct padata_instance *padata_alloc(const struct cpumask *cpumask,
 658                                      struct workqueue_struct *wq)
 659 {
 660         struct padata_instance *pinst;
 661         struct parallel_data *pd;
 662
 663         pinst = kzalloc(sizeof(struct padata_instance), GFP_KERNEL);
 664         if (!pinst)
 665                 goto err;
 666
 667         get_online_cpus();
 668
 669         pd = padata_alloc_pd(pinst, cpumask);
 670         if (!pd)
 671                 goto err_free_inst;
 672
 673         if (!alloc_cpumask_var(&pinst->cpumask, GFP_KERNEL))
 674                 goto err_free_pd;
 675
 676         rcu_assign_pointer(pinst->pd, pd);
 677
 678         pinst->wq = wq;
 679
 680         cpumask_copy(pinst->cpumask, cpumask);
 681
 682         pinst->flags = 0;
 683
 684 #ifdef CONFIG_HOTPLUG_CPU
 685         pinst->cpu_notifier.notifier_call = padata_cpu_callback;
 686         pinst->cpu_notifier.priority = 0;
 687         register_hotcpu_notifier(&pinst->cpu_notifier);
 688 #endif
 689
 690         put_online_cpus();
 691
 692         mutex_init(&pinst->lock);
 693
 694         return pinst;
 695
 696 err_free_pd:
 697         padata_free_pd(pd);
 698 err_free_inst:
 699         kfree(pinst);
 700         put_online_cpus();
 701 err:
 702         return NULL;
 703 }
 704 EXPORT_SYMBOL(padata_alloc);
 705
 706 /*
 707  * padata_free - free a padata instance
 708  *
 709  * @ padata_inst: padata instance to free
 710  */
 711 void padata_free(struct padata_instance *pinst)
 712 {
 713         padata_stop(pinst);
 714
 715         synchronize_rcu();
 716
 717 #ifdef CONFIG_HOTPLUG_CPU
 718         unregister_hotcpu_notifier(&pinst->cpu_notifier);
 719 #endif
 720         padata_flush_queues(pinst->pd);
 721         padata_free_pd(pinst->pd);
 722         free_cpumask_var(pinst->cpumask);
 723         kfree(pinst);
 724 }
 725 EXPORT_SYMBOL(padata_free);