SAFE public projects git trees. - safe/jmp/linux-2.6/blob - kernel/workqueue.c

   1 /*
   2  * linux/kernel/workqueue.c
   3  *
   4  * Generic mechanism for defining kernel helper threads for running
   5  * arbitrary tasks in process context.
   6  *
   7  * Started by Ingo Molnar, Copyright (C) 2002
   8  *
   9  * Derived from the taskqueue/keventd code by:
  10  *
  11  *   David Woodhouse <dwmw2@infradead.org>
  12  *   Andrew Morton <andrewm@uow.edu.au>
  13  *   Kai Petzke <wpp@marie.physik.tu-berlin.de>
  14  *   Theodore Ts'o <tytso@mit.edu>
  15  *
  16  * Made to use alloc_percpu by Christoph Lameter <clameter@sgi.com>.
  17  */
  18
  19 #include <linux/module.h>
  20 #include <linux/kernel.h>
  21 #include <linux/sched.h>
  22 #include <linux/init.h>
  23 #include <linux/signal.h>
  24 #include <linux/completion.h>
  25 #include <linux/workqueue.h>
  26 #include <linux/slab.h>
  27 #include <linux/cpu.h>
  28 #include <linux/notifier.h>
  29 #include <linux/kthread.h>
  30 #include <linux/hardirq.h>
  31 #include <linux/mempolicy.h>
  32 #include <linux/freezer.h>
  33 #include <linux/kallsyms.h>
  34 #include <linux/debug_locks.h>
  35
  36 /*
  37  * The per-CPU workqueue (if single thread, we always use the first
  38  * possible cpu).
  39  */
  40 struct cpu_workqueue_struct {
  41
  42         spinlock_t lock;
  43
  44         struct list_head worklist;
  45         wait_queue_head_t more_work;
  46         struct work_struct *current_work;
  47
  48         struct workqueue_struct *wq;
  49         struct task_struct *thread;
  50         int should_stop;
  51
  52         int run_depth;          /* Detect run_workqueue() recursion depth */
  53 } ____cacheline_aligned;
  54
  55 /*
  56  * The externally visible workqueue abstraction is an array of
  57  * per-CPU workqueues:
  58  */
  59 struct workqueue_struct {
  60         struct cpu_workqueue_struct *cpu_wq;
  61         const char *name;
  62         struct list_head list;  /* Empty if single thread */
  63         int freezeable;         /* Freeze threads during suspend */
  64 };
  65
  66 /* All the per-cpu workqueues on the system, for hotplug cpu to add/remove
  67    threads to each one as cpus come/go. */
  68 static DEFINE_MUTEX(workqueue_mutex);
  69 static LIST_HEAD(workqueues);
  70
  71 static int singlethread_cpu __read_mostly;
  72 static cpumask_t cpu_singlethread_map __read_mostly;
  73 /* optimization, we could use cpu_possible_map */
  74 static cpumask_t cpu_populated_map __read_mostly;
  75
  76 /* If it's single threaded, it isn't in the list of workqueues. */
  77 static inline int is_single_threaded(struct workqueue_struct *wq)
  78 {
  79         return list_empty(&wq->list);
  80 }
  81
  82 static const cpumask_t *wq_cpu_map(struct workqueue_struct *wq)
  83 {
  84         return is_single_threaded(wq)
  85                 ? &cpu_singlethread_map : &cpu_populated_map;
  86 }
  87
  88 /*
  89  * Set the workqueue on which a work item is to be run
  90  * - Must *only* be called if the pending flag is set
  91  */
  92 static inline void set_wq_data(struct work_struct *work, void *wq)
  93 {
  94         unsigned long new;
  95
  96         BUG_ON(!work_pending(work));
  97
  98         new = (unsigned long) wq | (1UL << WORK_STRUCT_PENDING);
  99         new |= WORK_STRUCT_FLAG_MASK & *work_data_bits(work);
 100         atomic_long_set(&work->data, new);
 101 }
 102
 103 static inline void *get_wq_data(struct work_struct *work)
 104 {
 105         return (void *) (atomic_long_read(&work->data) & WORK_STRUCT_WQ_DATA_MASK);
 106 }
 107
 108 static void insert_work(struct cpu_workqueue_struct *cwq,
 109                                 struct work_struct *work, int tail)
 110 {
 111         set_wq_data(work, cwq);
 112         if (tail)
 113                 list_add_tail(&work->entry, &cwq->worklist);
 114         else
 115                 list_add(&work->entry, &cwq->worklist);
 116         wake_up(&cwq->more_work);
 117 }
 118
 119 /* Preempt must be disabled. */
 120 static void __queue_work(struct cpu_workqueue_struct *cwq,
 121                          struct work_struct *work)
 122 {
 123         unsigned long flags;
 124
 125         spin_lock_irqsave(&cwq->lock, flags);
 126         insert_work(cwq, work, 1);
 127         spin_unlock_irqrestore(&cwq->lock, flags);
 128 }
 129
 130 /**
 131  * queue_work - queue work on a workqueue
 132  * @wq: workqueue to use
 133  * @work: work to queue
 134  *
 135  * Returns 0 if @work was already on a queue, non-zero otherwise.
 136  *
 137  * We queue the work to the CPU it was submitted, but there is no
 138  * guarantee that it will be processed by that CPU.
 139  */
 140 int fastcall queue_work(struct workqueue_struct *wq, struct work_struct *work)
 141 {
 142         int ret = 0, cpu = get_cpu();
 143
 144         if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
 145                 if (unlikely(is_single_threaded(wq)))
 146                         cpu = singlethread_cpu;
 147                 BUG_ON(!list_empty(&work->entry));
 148                 __queue_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
 149                 ret = 1;
 150         }
 151         put_cpu();
 152         return ret;
 153 }
 154 EXPORT_SYMBOL_GPL(queue_work);
 155
 156 void delayed_work_timer_fn(unsigned long __data)
 157 {
 158         struct delayed_work *dwork = (struct delayed_work *)__data;
 159         struct workqueue_struct *wq = get_wq_data(&dwork->work);
 160         int cpu = smp_processor_id();
 161
 162         if (unlikely(is_single_threaded(wq)))
 163                 cpu = singlethread_cpu;
 164
 165         __queue_work(per_cpu_ptr(wq->cpu_wq, cpu), &dwork->work);
 166 }
 167
 168 /**
 169  * queue_delayed_work - queue work on a workqueue after delay
 170  * @wq: workqueue to use
 171  * @dwork: delayable work to queue
 172  * @delay: number of jiffies to wait before queueing
 173  *
 174  * Returns 0 if @work was already on a queue, non-zero otherwise.
 175  */
 176 int fastcall queue_delayed_work(struct workqueue_struct *wq,
 177                         struct delayed_work *dwork, unsigned long delay)
 178 {
 179         int ret = 0;
 180         struct timer_list *timer = &dwork->timer;
 181         struct work_struct *work = &dwork->work;
 182
 183         timer_stats_timer_set_start_info(timer);
 184         if (delay == 0)
 185                 return queue_work(wq, work);
 186
 187         if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
 188                 BUG_ON(timer_pending(timer));
 189                 BUG_ON(!list_empty(&work->entry));
 190
 191                 /* This stores wq for the moment, for the timer_fn */
 192                 set_wq_data(work, wq);
 193                 timer->expires = jiffies + delay;
 194                 timer->data = (unsigned long)dwork;
 195                 timer->function = delayed_work_timer_fn;
 196                 add_timer(timer);
 197                 ret = 1;
 198         }
 199         return ret;
 200 }
 201 EXPORT_SYMBOL_GPL(queue_delayed_work);
 202
 203 /**
 204  * queue_delayed_work_on - queue work on specific CPU after delay
 205  * @cpu: CPU number to execute work on
 206  * @wq: workqueue to use
 207  * @dwork: work to queue
 208  * @delay: number of jiffies to wait before queueing
 209  *
 210  * Returns 0 if @work was already on a queue, non-zero otherwise.
 211  */
 212 int queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
 213                         struct delayed_work *dwork, unsigned long delay)
 214 {
 215         int ret = 0;
 216         struct timer_list *timer = &dwork->timer;
 217         struct work_struct *work = &dwork->work;
 218
 219         if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
 220                 BUG_ON(timer_pending(timer));
 221                 BUG_ON(!list_empty(&work->entry));
 222
 223                 /* This stores wq for the moment, for the timer_fn */
 224                 set_wq_data(work, wq);
 225                 timer->expires = jiffies + delay;
 226                 timer->data = (unsigned long)dwork;
 227                 timer->function = delayed_work_timer_fn;
 228                 add_timer_on(timer, cpu);
 229                 ret = 1;
 230         }
 231         return ret;
 232 }
 233 EXPORT_SYMBOL_GPL(queue_delayed_work_on);
 234
 235 static void run_workqueue(struct cpu_workqueue_struct *cwq)
 236 {
 237         spin_lock_irq(&cwq->lock);
 238         cwq->run_depth++;
 239         if (cwq->run_depth > 3) {
 240                 /* morton gets to eat his hat */
 241                 printk("%s: recursion depth exceeded: %d\n",
 242                         __FUNCTION__, cwq->run_depth);
 243                 dump_stack();
 244         }
 245         while (!list_empty(&cwq->worklist)) {
 246                 struct work_struct *work = list_entry(cwq->worklist.next,
 247                                                 struct work_struct, entry);
 248                 work_func_t f = work->func;
 249
 250                 cwq->current_work = work;
 251                 list_del_init(cwq->worklist.next);
 252                 spin_unlock_irq(&cwq->lock);
 253
 254                 BUG_ON(get_wq_data(work) != cwq);
 255                 if (!test_bit(WORK_STRUCT_NOAUTOREL, work_data_bits(work)))
 256                         work_release(work);
 257                 f(work);
 258
 259                 if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
 260                         printk(KERN_ERR "BUG: workqueue leaked lock or atomic: "
 261                                         "%s/0x%08x/%d\n",
 262                                         current->comm, preempt_count(),
 263                                         current->pid);
 264                         printk(KERN_ERR "    last function: ");
 265                         print_symbol("%s\n", (unsigned long)f);
 266                         debug_show_held_locks(current);
 267                         dump_stack();
 268                 }
 269
 270                 spin_lock_irq(&cwq->lock);
 271                 cwq->current_work = NULL;
 272         }
 273         cwq->run_depth--;
 274         spin_unlock_irq(&cwq->lock);
 275 }
 276
 277 /*
 278  * NOTE: the caller must not touch *cwq if this func returns true
 279  */
 280 static int cwq_should_stop(struct cpu_workqueue_struct *cwq)
 281 {
 282         int should_stop = cwq->should_stop;
 283
 284         if (unlikely(should_stop)) {
 285                 spin_lock_irq(&cwq->lock);
 286                 should_stop = cwq->should_stop && list_empty(&cwq->worklist);
 287                 if (should_stop)
 288                         cwq->thread = NULL;
 289                 spin_unlock_irq(&cwq->lock);
 290         }
 291
 292         return should_stop;
 293 }
 294
 295 static int worker_thread(void *__cwq)
 296 {
 297         struct cpu_workqueue_struct *cwq = __cwq;
 298         DEFINE_WAIT(wait);
 299         struct k_sigaction sa;
 300         sigset_t blocked;
 301
 302         if (!cwq->wq->freezeable)
 303                 current->flags |= PF_NOFREEZE;
 304
 305         set_user_nice(current, -5);
 306
 307         /* Block and flush all signals */
 308         sigfillset(&blocked);
 309         sigprocmask(SIG_BLOCK, &blocked, NULL);
 310         flush_signals(current);
 311
 312         /*
 313          * We inherited MPOL_INTERLEAVE from the booting kernel.
 314          * Set MPOL_DEFAULT to insure node local allocations.
 315          */
 316         numa_default_policy();
 317
 318         /* SIG_IGN makes children autoreap: see do_notify_parent(). */
 319         sa.sa.sa_handler = SIG_IGN;
 320         sa.sa.sa_flags = 0;
 321         siginitset(&sa.sa.sa_mask, sigmask(SIGCHLD));
 322         do_sigaction(SIGCHLD, &sa, (struct k_sigaction *)0);
 323
 324         for (;;) {
 325                 if (cwq->wq->freezeable)
 326                         try_to_freeze();
 327
 328                 prepare_to_wait(&cwq->more_work, &wait, TASK_INTERRUPTIBLE);
 329                 if (!cwq->should_stop && list_empty(&cwq->worklist))
 330                         schedule();
 331                 finish_wait(&cwq->more_work, &wait);
 332
 333                 if (cwq_should_stop(cwq))
 334                         break;
 335
 336                 run_workqueue(cwq);
 337         }
 338
 339         return 0;
 340 }
 341
 342 struct wq_barrier {
 343         struct work_struct      work;
 344         struct completion       done;
 345 };
 346
 347 static void wq_barrier_func(struct work_struct *work)
 348 {
 349         struct wq_barrier *barr = container_of(work, struct wq_barrier, work);
 350         complete(&barr->done);
 351 }
 352
 353 static void insert_wq_barrier(struct cpu_workqueue_struct *cwq,
 354                                         struct wq_barrier *barr, int tail)
 355 {
 356         INIT_WORK(&barr->work, wq_barrier_func);
 357         __set_bit(WORK_STRUCT_PENDING, work_data_bits(&barr->work));
 358
 359         init_completion(&barr->done);
 360
 361         insert_work(cwq, &barr->work, tail);
 362 }
 363
 364 static void flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
 365 {
 366         if (cwq->thread == current) {
 367                 /*
 368                  * Probably keventd trying to flush its own queue. So simply run
 369                  * it by hand rather than deadlocking.
 370                  */
 371                 run_workqueue(cwq);
 372         } else {
 373                 struct wq_barrier barr;
 374                 int active = 0;
 375
 376                 spin_lock_irq(&cwq->lock);
 377                 if (!list_empty(&cwq->worklist) || cwq->current_work != NULL) {
 378                         insert_wq_barrier(cwq, &barr, 1);
 379                         active = 1;
 380                 }
 381                 spin_unlock_irq(&cwq->lock);
 382
 383                 if (active)
 384                         wait_for_completion(&barr.done);
 385         }
 386 }
 387
 388 /**
 389  * flush_workqueue - ensure that any scheduled work has run to completion.
 390  * @wq: workqueue to flush
 391  *
 392  * Forces execution of the workqueue and blocks until its completion.
 393  * This is typically used in driver shutdown handlers.
 394  *
 395  * We sleep until all works which were queued on entry have been handled,
 396  * but we are not livelocked by new incoming ones.
 397  *
 398  * This function used to run the workqueues itself.  Now we just wait for the
 399  * helper threads to do it.
 400  */
 401 void fastcall flush_workqueue(struct workqueue_struct *wq)
 402 {
 403         const cpumask_t *cpu_map = wq_cpu_map(wq);
 404         int cpu
 405
 406         might_sleep();
 407         for_each_cpu_mask(cpu, *cpu_map)
 408                 flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, cpu));
 409 }
 410 EXPORT_SYMBOL_GPL(flush_workqueue);
 411
 412 static void wait_on_work(struct cpu_workqueue_struct *cwq,
 413                                 struct work_struct *work)
 414 {
 415         struct wq_barrier barr;
 416         int running = 0;
 417
 418         spin_lock_irq(&cwq->lock);
 419         if (unlikely(cwq->current_work == work)) {
 420                 insert_wq_barrier(cwq, &barr, 0);
 421                 running = 1;
 422         }
 423         spin_unlock_irq(&cwq->lock);
 424
 425         if (unlikely(running))
 426                 wait_for_completion(&barr.done);
 427 }
 428
 429 /**
 430  * flush_work - block until a work_struct's callback has terminated
 431  * @wq: the workqueue on which the work is queued
 432  * @work: the work which is to be flushed
 433  *
 434  * flush_work() will attempt to cancel the work if it is queued.  If the work's
 435  * callback appears to be running, flush_work() will block until it has
 436  * completed.
 437  *
 438  * flush_work() is designed to be used when the caller is tearing down data
 439  * structures which the callback function operates upon.  It is expected that,
 440  * prior to calling flush_work(), the caller has arranged for the work to not
 441  * be requeued.
 442  */
 443 void flush_work(struct workqueue_struct *wq, struct work_struct *work)
 444 {
 445         const cpumask_t *cpu_map = wq_cpu_map(wq);
 446         struct cpu_workqueue_struct *cwq;
 447         int cpu;
 448
 449         might_sleep();
 450
 451         cwq = get_wq_data(work);
 452         /* Was it ever queued ? */
 453         if (!cwq)
 454                 return;
 455
 456         /*
 457          * This work can't be re-queued, no need to re-check that
 458          * get_wq_data() is still the same when we take cwq->lock.
 459          */
 460         spin_lock_irq(&cwq->lock);
 461         list_del_init(&work->entry);
 462         work_release(work);
 463         spin_unlock_irq(&cwq->lock);
 464
 465         for_each_cpu_mask(cpu, *cpu_map)
 466                 wait_on_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
 467 }
 468 EXPORT_SYMBOL_GPL(flush_work);
 469
 470
 471 static struct workqueue_struct *keventd_wq;
 472
 473 /**
 474  * schedule_work - put work task in global workqueue
 475  * @work: job to be done
 476  *
 477  * This puts a job in the kernel-global workqueue.
 478  */
 479 int fastcall schedule_work(struct work_struct *work)
 480 {
 481         return queue_work(keventd_wq, work);
 482 }
 483 EXPORT_SYMBOL(schedule_work);
 484
 485 /**
 486  * schedule_delayed_work - put work task in global workqueue after delay
 487  * @dwork: job to be done
 488  * @delay: number of jiffies to wait or 0 for immediate execution
 489  *
 490  * After waiting for a given time this puts a job in the kernel-global
 491  * workqueue.
 492  */
 493 int fastcall schedule_delayed_work(struct delayed_work *dwork,
 494                                         unsigned long delay)
 495 {
 496         timer_stats_timer_set_start_info(&dwork->timer);
 497         return queue_delayed_work(keventd_wq, dwork, delay);
 498 }
 499 EXPORT_SYMBOL(schedule_delayed_work);
 500
 501 /**
 502  * schedule_delayed_work_on - queue work in global workqueue on CPU after delay
 503  * @cpu: cpu to use
 504  * @dwork: job to be done
 505  * @delay: number of jiffies to wait
 506  *
 507  * After waiting for a given time this puts a job in the kernel-global
 508  * workqueue on the specified CPU.
 509  */
 510 int schedule_delayed_work_on(int cpu,
 511                         struct delayed_work *dwork, unsigned long delay)
 512 {
 513         return queue_delayed_work_on(cpu, keventd_wq, dwork, delay);
 514 }
 515 EXPORT_SYMBOL(schedule_delayed_work_on);
 516
 517 /**
 518  * schedule_on_each_cpu - call a function on each online CPU from keventd
 519  * @func: the function to call
 520  *
 521  * Returns zero on success.
 522  * Returns -ve errno on failure.
 523  *
 524  * Appears to be racy against CPU hotplug.
 525  *
 526  * schedule_on_each_cpu() is very slow.
 527  */
 528 int schedule_on_each_cpu(work_func_t func)
 529 {
 530         int cpu;
 531         struct work_struct *works;
 532
 533         works = alloc_percpu(struct work_struct);
 534         if (!works)
 535                 return -ENOMEM;
 536
 537         preempt_disable();              /* CPU hotplug */
 538         for_each_online_cpu(cpu) {
 539                 struct work_struct *work = per_cpu_ptr(works, cpu);
 540
 541                 INIT_WORK(work, func);
 542                 set_bit(WORK_STRUCT_PENDING, work_data_bits(work));
 543                 __queue_work(per_cpu_ptr(keventd_wq->cpu_wq, cpu), work);
 544         }
 545         preempt_enable();
 546         flush_workqueue(keventd_wq);
 547         free_percpu(works);
 548         return 0;
 549 }
 550
 551 void flush_scheduled_work(void)
 552 {
 553         flush_workqueue(keventd_wq);
 554 }
 555 EXPORT_SYMBOL(flush_scheduled_work);
 556
 557 void flush_work_keventd(struct work_struct *work)
 558 {
 559         flush_work(keventd_wq, work);
 560 }
 561 EXPORT_SYMBOL(flush_work_keventd);
 562
 563 /**
 564  * cancel_rearming_delayed_workqueue - reliably kill off a delayed work whose handler rearms the delayed work.
 565  * @wq:   the controlling workqueue structure
 566  * @dwork: the delayed work struct
 567  */
 568 void cancel_rearming_delayed_workqueue(struct workqueue_struct *wq,
 569                                        struct delayed_work *dwork)
 570 {
 571         /* Was it ever queued ? */
 572         if (!get_wq_data(&dwork->work))
 573                 return;
 574
 575         while (!cancel_delayed_work(dwork))
 576                 flush_workqueue(wq);
 577 }
 578 EXPORT_SYMBOL(cancel_rearming_delayed_workqueue);
 579
 580 /**
 581  * cancel_rearming_delayed_work - reliably kill off a delayed keventd work whose handler rearms the delayed work.
 582  * @dwork: the delayed work struct
 583  */
 584 void cancel_rearming_delayed_work(struct delayed_work *dwork)
 585 {
 586         cancel_rearming_delayed_workqueue(keventd_wq, dwork);
 587 }
 588 EXPORT_SYMBOL(cancel_rearming_delayed_work);
 589
 590 /**
 591  * execute_in_process_context - reliably execute the routine with user context
 592  * @fn:         the function to execute
 593  * @ew:         guaranteed storage for the execute work structure (must
 594  *              be available when the work executes)
 595  *
 596  * Executes the function immediately if process context is available,
 597  * otherwise schedules the function for delayed execution.
 598  *
 599  * Returns:     0 - function was executed
 600  *              1 - function was scheduled for execution
 601  */
 602 int execute_in_process_context(work_func_t fn, struct execute_work *ew)
 603 {
 604         if (!in_interrupt()) {
 605                 fn(&ew->work);
 606                 return 0;
 607         }
 608
 609         INIT_WORK(&ew->work, fn);
 610         schedule_work(&ew->work);
 611
 612         return 1;
 613 }
 614 EXPORT_SYMBOL_GPL(execute_in_process_context);
 615
 616 int keventd_up(void)
 617 {
 618         return keventd_wq != NULL;
 619 }
 620
 621 int current_is_keventd(void)
 622 {
 623         struct cpu_workqueue_struct *cwq;
 624         int cpu = smp_processor_id();   /* preempt-safe: keventd is per-cpu */
 625         int ret = 0;
 626
 627         BUG_ON(!keventd_wq);
 628
 629         cwq = per_cpu_ptr(keventd_wq->cpu_wq, cpu);
 630         if (current == cwq->thread)
 631                 ret = 1;
 632
 633         return ret;
 634
 635 }
 636
 637 static struct cpu_workqueue_struct *
 638 init_cpu_workqueue(struct workqueue_struct *wq, int cpu)
 639 {
 640         struct cpu_workqueue_struct *cwq = per_cpu_ptr(wq->cpu_wq, cpu);
 641
 642         cwq->wq = wq;
 643         spin_lock_init(&cwq->lock);
 644         INIT_LIST_HEAD(&cwq->worklist);
 645         init_waitqueue_head(&cwq->more_work);
 646
 647         return cwq;
 648 }
 649
 650 static int create_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu)
 651 {
 652         struct workqueue_struct *wq = cwq->wq;
 653         const char *fmt = is_single_threaded(wq) ? "%s" : "%s/%d";
 654         struct task_struct *p;
 655
 656         p = kthread_create(worker_thread, cwq, fmt, wq->name, cpu);
 657         /*
 658          * Nobody can add the work_struct to this cwq,
 659          *      if (caller is __create_workqueue)
 660          *              nobody should see this wq
 661          *      else // caller is CPU_UP_PREPARE
 662          *              cpu is not on cpu_online_map
 663          * so we can abort safely.
 664          */
 665         if (IS_ERR(p))
 666                 return PTR_ERR(p);
 667
 668         cwq->thread = p;
 669         cwq->should_stop = 0;
 670         if (!is_single_threaded(wq))
 671                 kthread_bind(p, cpu);
 672
 673         if (is_single_threaded(wq) || cpu_online(cpu))
 674                 wake_up_process(p);
 675
 676         return 0;
 677 }
 678
 679 struct workqueue_struct *__create_workqueue(const char *name,
 680                                             int singlethread, int freezeable)
 681 {
 682         struct workqueue_struct *wq;
 683         struct cpu_workqueue_struct *cwq;
 684         int err = 0, cpu;
 685
 686         wq = kzalloc(sizeof(*wq), GFP_KERNEL);
 687         if (!wq)
 688                 return NULL;
 689
 690         wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct);
 691         if (!wq->cpu_wq) {
 692                 kfree(wq);
 693                 return NULL;
 694         }
 695
 696         wq->name = name;
 697         wq->freezeable = freezeable;
 698
 699         if (singlethread) {
 700                 INIT_LIST_HEAD(&wq->list);
 701                 cwq = init_cpu_workqueue(wq, singlethread_cpu);
 702                 err = create_workqueue_thread(cwq, singlethread_cpu);
 703         } else {
 704                 mutex_lock(&workqueue_mutex);
 705                 list_add(&wq->list, &workqueues);
 706
 707                 for_each_possible_cpu(cpu) {
 708                         cwq = init_cpu_workqueue(wq, cpu);
 709                         if (err || !cpu_online(cpu))
 710                                 continue;
 711                         err = create_workqueue_thread(cwq, cpu);
 712                 }
 713                 mutex_unlock(&workqueue_mutex);
 714         }
 715
 716         if (err) {
 717                 destroy_workqueue(wq);
 718                 wq = NULL;
 719         }
 720         return wq;
 721 }
 722 EXPORT_SYMBOL_GPL(__create_workqueue);
 723
 724 static void cleanup_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu)
 725 {
 726         struct wq_barrier barr;
 727         int alive = 0;
 728
 729         spin_lock_irq(&cwq->lock);
 730         if (cwq->thread != NULL) {
 731                 insert_wq_barrier(cwq, &barr, 1);
 732                 cwq->should_stop = 1;
 733                 alive = 1;
 734         }
 735         spin_unlock_irq(&cwq->lock);
 736
 737         if (alive) {
 738                 wait_for_completion(&barr.done);
 739
 740                 while (unlikely(cwq->thread != NULL))
 741                         cpu_relax();
 742                 /*
 743                  * Wait until cwq->thread unlocks cwq->lock,
 744                  * it won't touch *cwq after that.
 745                  */
 746                 smp_rmb();
 747                 spin_unlock_wait(&cwq->lock);
 748         }
 749 }
 750
 751 /**
 752  * destroy_workqueue - safely terminate a workqueue
 753  * @wq: target workqueue
 754  *
 755  * Safely destroy a workqueue. All work currently pending will be done first.
 756  */
 757 void destroy_workqueue(struct workqueue_struct *wq)
 758 {
 759         const cpumask_t *cpu_map = wq_cpu_map(wq);
 760         struct cpu_workqueue_struct *cwq;
 761         int cpu;
 762
 763         mutex_lock(&workqueue_mutex);
 764         list_del(&wq->list);
 765         mutex_unlock(&workqueue_mutex);
 766
 767         for_each_cpu_mask(cpu, *cpu_map) {
 768                 cwq = per_cpu_ptr(wq->cpu_wq, cpu);
 769                 cleanup_workqueue_thread(cwq, cpu);
 770         }
 771
 772         free_percpu(wq->cpu_wq);
 773         kfree(wq);
 774 }
 775 EXPORT_SYMBOL_GPL(destroy_workqueue);
 776
 777 static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
 778                                                 unsigned long action,
 779                                                 void *hcpu)
 780 {
 781         unsigned int cpu = (unsigned long)hcpu;
 782         struct cpu_workqueue_struct *cwq;
 783         struct workqueue_struct *wq;
 784
 785         switch (action) {
 786         case CPU_LOCK_ACQUIRE:
 787                 mutex_lock(&workqueue_mutex);
 788                 return NOTIFY_OK;
 789
 790         case CPU_LOCK_RELEASE:
 791                 mutex_unlock(&workqueue_mutex);
 792                 return NOTIFY_OK;
 793
 794         case CPU_UP_PREPARE:
 795                 cpu_set(cpu, cpu_populated_map);
 796         }
 797
 798         list_for_each_entry(wq, &workqueues, list) {
 799                 cwq = per_cpu_ptr(wq->cpu_wq, cpu);
 800
 801                 switch (action) {
 802                 case CPU_UP_PREPARE:
 803                         if (!create_workqueue_thread(cwq, cpu))
 804                                 break;
 805                         printk(KERN_ERR "workqueue for %i failed\n", cpu);
 806                         return NOTIFY_BAD;
 807
 808                 case CPU_ONLINE:
 809                         wake_up_process(cwq->thread);
 810                         break;
 811
 812                 case CPU_UP_CANCELED:
 813                         if (cwq->thread)
 814                                 wake_up_process(cwq->thread);
 815                 case CPU_DEAD:
 816                         cleanup_workqueue_thread(cwq, cpu);
 817                         break;
 818                 }
 819         }
 820
 821         return NOTIFY_OK;
 822 }
 823
 824 void init_workqueues(void)
 825 {
 826         cpu_populated_map = cpu_online_map;
 827         singlethread_cpu = first_cpu(cpu_possible_map);
 828         cpu_singlethread_map = cpumask_of_cpu(singlethread_cpu);
 829         hotcpu_notifier(workqueue_cpu_callback, 0);
 830         keventd_wq = create_workqueue("events");
 831         BUG_ON(!keventd_wq);
 832 }