git.oblomov.eu Git - 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
  47         struct workqueue_struct *wq;
  48         struct task_struct *thread;
  49         struct work_struct *current_work;
  50
  51         int run_depth;          /* Detect run_workqueue() recursion depth */
  52
  53         int freezeable;         /* Freeze the thread during suspend */
  54 } ____cacheline_aligned;
  55
  56 /*
  57  * The externally visible workqueue abstraction is an array of
  58  * per-CPU workqueues:
  59  */
  60 struct workqueue_struct {
  61         struct cpu_workqueue_struct *cpu_wq;
  62         const char *name;
  63         struct list_head list;  /* Empty if single thread */
  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;
  72
  73 /* If it's single threaded, it isn't in the list of workqueues. */
  74 static inline int is_single_threaded(struct workqueue_struct *wq)
  75 {
  76         return list_empty(&wq->list);
  77 }
  78
  79 /*
  80  * Set the workqueue on which a work item is to be run
  81  * - Must *only* be called if the pending flag is set
  82  */
  83 static inline void set_wq_data(struct work_struct *work, void *wq)
  84 {
  85         unsigned long new;
  86
  87         BUG_ON(!work_pending(work));
  88
  89         new = (unsigned long) wq | (1UL << WORK_STRUCT_PENDING);
  90         new |= WORK_STRUCT_FLAG_MASK & *work_data_bits(work);
  91         atomic_long_set(&work->data, new);
  92 }
  93
  94 static inline void *get_wq_data(struct work_struct *work)
  95 {
  96         return (void *) (atomic_long_read(&work->data) & WORK_STRUCT_WQ_DATA_MASK);
  97 }
  98
  99 static int __run_work(struct cpu_workqueue_struct *cwq, struct work_struct *work)
 100 {
 101         int ret = 0;
 102         unsigned long flags;
 103
 104         spin_lock_irqsave(&cwq->lock, flags);
 105         /*
 106          * We need to re-validate the work info after we've gotten
 107          * the cpu_workqueue lock. We can run the work now iff:
 108          *
 109          *  - the wq_data still matches the cpu_workqueue_struct
 110          *  - AND the work is still marked pending
 111          *  - AND the work is still on a list (which will be this
 112          *    workqueue_struct list)
 113          *
 114          * All these conditions are important, because we
 115          * need to protect against the work being run right
 116          * now on another CPU (all but the last one might be
 117          * true if it's currently running and has not been
 118          * released yet, for example).
 119          */
 120         if (get_wq_data(work) == cwq
 121             && work_pending(work)
 122             && !list_empty(&work->entry)) {
 123                 work_func_t f = work->func;
 124                 cwq->current_work = work;
 125                 list_del_init(&work->entry);
 126                 spin_unlock_irqrestore(&cwq->lock, flags);
 127
 128                 if (!test_bit(WORK_STRUCT_NOAUTOREL, work_data_bits(work)))
 129                         work_release(work);
 130                 f(work);
 131
 132                 spin_lock_irqsave(&cwq->lock, flags);
 133                 cwq->current_work = NULL;
 134                 ret = 1;
 135         }
 136         spin_unlock_irqrestore(&cwq->lock, flags);
 137         return ret;
 138 }
 139
 140 /**
 141  * run_scheduled_work - run scheduled work synchronously
 142  * @work: work to run
 143  *
 144  * This checks if the work was pending, and runs it
 145  * synchronously if so. It returns a boolean to indicate
 146  * whether it had any scheduled work to run or not.
 147  *
 148  * NOTE! This _only_ works for normal work_structs. You
 149  * CANNOT use this for delayed work, because the wq data
 150  * for delayed work will not point properly to the per-
 151  * CPU workqueue struct, but will change!
 152  */
 153 int fastcall run_scheduled_work(struct work_struct *work)
 154 {
 155         for (;;) {
 156                 struct cpu_workqueue_struct *cwq;
 157
 158                 if (!work_pending(work))
 159                         return 0;
 160                 if (list_empty(&work->entry))
 161                         return 0;
 162                 /* NOTE! This depends intimately on __queue_work! */
 163                 cwq = get_wq_data(work);
 164                 if (!cwq)
 165                         return 0;
 166                 if (__run_work(cwq, work))
 167                         return 1;
 168         }
 169 }
 170 EXPORT_SYMBOL(run_scheduled_work);
 171
 172 static void insert_work(struct cpu_workqueue_struct *cwq,
 173                                 struct work_struct *work, int tail)
 174 {
 175         set_wq_data(work, cwq);
 176         if (tail)
 177                 list_add_tail(&work->entry, &cwq->worklist);
 178         else
 179                 list_add(&work->entry, &cwq->worklist);
 180         wake_up(&cwq->more_work);
 181 }
 182
 183 /* Preempt must be disabled. */
 184 static void __queue_work(struct cpu_workqueue_struct *cwq,
 185                          struct work_struct *work)
 186 {
 187         unsigned long flags;
 188
 189         spin_lock_irqsave(&cwq->lock, flags);
 190         insert_work(cwq, work, 1);
 191         spin_unlock_irqrestore(&cwq->lock, flags);
 192 }
 193
 194 /**
 195  * queue_work - queue work on a workqueue
 196  * @wq: workqueue to use
 197  * @work: work to queue
 198  *
 199  * Returns 0 if @work was already on a queue, non-zero otherwise.
 200  *
 201  * We queue the work to the CPU it was submitted, but there is no
 202  * guarantee that it will be processed by that CPU.
 203  */
 204 int fastcall queue_work(struct workqueue_struct *wq, struct work_struct *work)
 205 {
 206         int ret = 0, cpu = get_cpu();
 207
 208         if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
 209                 if (unlikely(is_single_threaded(wq)))
 210                         cpu = singlethread_cpu;
 211                 BUG_ON(!list_empty(&work->entry));
 212                 __queue_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
 213                 ret = 1;
 214         }
 215         put_cpu();
 216         return ret;
 217 }
 218 EXPORT_SYMBOL_GPL(queue_work);
 219
 220 void delayed_work_timer_fn(unsigned long __data)
 221 {
 222         struct delayed_work *dwork = (struct delayed_work *)__data;
 223         struct workqueue_struct *wq = get_wq_data(&dwork->work);
 224         int cpu = smp_processor_id();
 225
 226         if (unlikely(is_single_threaded(wq)))
 227                 cpu = singlethread_cpu;
 228
 229         __queue_work(per_cpu_ptr(wq->cpu_wq, cpu), &dwork->work);
 230 }
 231
 232 /**
 233  * queue_delayed_work - queue work on a workqueue after delay
 234  * @wq: workqueue to use
 235  * @dwork: delayable work to queue
 236  * @delay: number of jiffies to wait before queueing
 237  *
 238  * Returns 0 if @work was already on a queue, non-zero otherwise.
 239  */
 240 int fastcall queue_delayed_work(struct workqueue_struct *wq,
 241                         struct delayed_work *dwork, unsigned long delay)
 242 {
 243         int ret = 0;
 244         struct timer_list *timer = &dwork->timer;
 245         struct work_struct *work = &dwork->work;
 246
 247         timer_stats_timer_set_start_info(timer);
 248         if (delay == 0)
 249                 return queue_work(wq, work);
 250
 251         if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
 252                 BUG_ON(timer_pending(timer));
 253                 BUG_ON(!list_empty(&work->entry));
 254
 255                 /* This stores wq for the moment, for the timer_fn */
 256                 set_wq_data(work, wq);
 257                 timer->expires = jiffies + delay;
 258                 timer->data = (unsigned long)dwork;
 259                 timer->function = delayed_work_timer_fn;
 260                 add_timer(timer);
 261                 ret = 1;
 262         }
 263         return ret;
 264 }
 265 EXPORT_SYMBOL_GPL(queue_delayed_work);
 266
 267 /**
 268  * queue_delayed_work_on - queue work on specific CPU after delay
 269  * @cpu: CPU number to execute work on
 270  * @wq: workqueue to use
 271  * @dwork: work to queue
 272  * @delay: number of jiffies to wait before queueing
 273  *
 274  * Returns 0 if @work was already on a queue, non-zero otherwise.
 275  */
 276 int queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
 277                         struct delayed_work *dwork, unsigned long delay)
 278 {
 279         int ret = 0;
 280         struct timer_list *timer = &dwork->timer;
 281         struct work_struct *work = &dwork->work;
 282
 283         if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
 284                 BUG_ON(timer_pending(timer));
 285                 BUG_ON(!list_empty(&work->entry));
 286
 287                 /* This stores wq for the moment, for the timer_fn */
 288                 set_wq_data(work, wq);
 289                 timer->expires = jiffies + delay;
 290                 timer->data = (unsigned long)dwork;
 291                 timer->function = delayed_work_timer_fn;
 292                 add_timer_on(timer, cpu);
 293                 ret = 1;
 294         }
 295         return ret;
 296 }
 297 EXPORT_SYMBOL_GPL(queue_delayed_work_on);
 298
 299 static void run_workqueue(struct cpu_workqueue_struct *cwq)
 300 {
 301         unsigned long flags;
 302
 303         /*
 304          * Keep taking off work from the queue until
 305          * done.
 306          */
 307         spin_lock_irqsave(&cwq->lock, flags);
 308         cwq->run_depth++;
 309         if (cwq->run_depth > 3) {
 310                 /* morton gets to eat his hat */
 311                 printk("%s: recursion depth exceeded: %d\n",
 312                         __FUNCTION__, cwq->run_depth);
 313                 dump_stack();
 314         }
 315         while (!list_empty(&cwq->worklist)) {
 316                 struct work_struct *work = list_entry(cwq->worklist.next,
 317                                                 struct work_struct, entry);
 318                 work_func_t f = work->func;
 319
 320                 cwq->current_work = work;
 321                 list_del_init(cwq->worklist.next);
 322                 spin_unlock_irqrestore(&cwq->lock, flags);
 323
 324                 BUG_ON(get_wq_data(work) != cwq);
 325                 if (!test_bit(WORK_STRUCT_NOAUTOREL, work_data_bits(work)))
 326                         work_release(work);
 327                 f(work);
 328
 329                 if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
 330                         printk(KERN_ERR "BUG: workqueue leaked lock or atomic: "
 331                                         "%s/0x%08x/%d\n",
 332                                         current->comm, preempt_count(),
 333                                         current->pid);
 334                         printk(KERN_ERR "    last function: ");
 335                         print_symbol("%s\n", (unsigned long)f);
 336                         debug_show_held_locks(current);
 337                         dump_stack();
 338                 }
 339
 340                 spin_lock_irqsave(&cwq->lock, flags);
 341                 cwq->current_work = NULL;
 342         }
 343         cwq->run_depth--;
 344         spin_unlock_irqrestore(&cwq->lock, flags);
 345 }
 346
 347 static int worker_thread(void *__cwq)
 348 {
 349         struct cpu_workqueue_struct *cwq = __cwq;
 350         DECLARE_WAITQUEUE(wait, current);
 351         struct k_sigaction sa;
 352         sigset_t blocked;
 353
 354         if (!cwq->freezeable)
 355                 current->flags |= PF_NOFREEZE;
 356
 357         set_user_nice(current, -5);
 358
 359         /* Block and flush all signals */
 360         sigfillset(&blocked);
 361         sigprocmask(SIG_BLOCK, &blocked, NULL);
 362         flush_signals(current);
 363
 364         /*
 365          * We inherited MPOL_INTERLEAVE from the booting kernel.
 366          * Set MPOL_DEFAULT to insure node local allocations.
 367          */
 368         numa_default_policy();
 369
 370         /* SIG_IGN makes children autoreap: see do_notify_parent(). */
 371         sa.sa.sa_handler = SIG_IGN;
 372         sa.sa.sa_flags = 0;
 373         siginitset(&sa.sa.sa_mask, sigmask(SIGCHLD));
 374         do_sigaction(SIGCHLD, &sa, (struct k_sigaction *)0);
 375
 376         set_current_state(TASK_INTERRUPTIBLE);
 377         while (!kthread_should_stop()) {
 378                 if (cwq->freezeable)
 379                         try_to_freeze();
 380
 381                 add_wait_queue(&cwq->more_work, &wait);
 382                 if (list_empty(&cwq->worklist))
 383                         schedule();
 384                 else
 385                         __set_current_state(TASK_RUNNING);
 386                 remove_wait_queue(&cwq->more_work, &wait);
 387
 388                 if (!list_empty(&cwq->worklist))
 389                         run_workqueue(cwq);
 390                 set_current_state(TASK_INTERRUPTIBLE);
 391         }
 392         __set_current_state(TASK_RUNNING);
 393         return 0;
 394 }
 395
 396 struct wq_barrier {
 397         struct work_struct      work;
 398         struct completion       done;
 399 };
 400
 401 static void wq_barrier_func(struct work_struct *work)
 402 {
 403         struct wq_barrier *barr = container_of(work, struct wq_barrier, work);
 404         complete(&barr->done);
 405 }
 406
 407 static inline void init_wq_barrier(struct wq_barrier *barr)
 408 {
 409         INIT_WORK(&barr->work, wq_barrier_func);
 410         __set_bit(WORK_STRUCT_PENDING, work_data_bits(&barr->work));
 411
 412         init_completion(&barr->done);
 413 }
 414
 415 static void flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
 416 {
 417         if (cwq->thread == current) {
 418                 /*
 419                  * Probably keventd trying to flush its own queue. So simply run
 420                  * it by hand rather than deadlocking.
 421                  */
 422                 preempt_enable();
 423                 /*
 424                  * We can still touch *cwq here because we are keventd, and
 425                  * hot-unplug will be waiting us to exit.
 426                  */
 427                 run_workqueue(cwq);
 428                 preempt_disable();
 429         } else {
 430                 struct wq_barrier barr;
 431
 432                 init_wq_barrier(&barr);
 433                 __queue_work(cwq, &barr.work);
 434
 435                 preempt_enable();       /* Can no longer touch *cwq */
 436                 wait_for_completion(&barr.done);
 437                 preempt_disable();
 438         }
 439 }
 440
 441 /**
 442  * flush_workqueue - ensure that any scheduled work has run to completion.
 443  * @wq: workqueue to flush
 444  *
 445  * Forces execution of the workqueue and blocks until its completion.
 446  * This is typically used in driver shutdown handlers.
 447  *
 448  * We sleep until all works which were queued on entry have been handled,
 449  * but we are not livelocked by new incoming ones.
 450  *
 451  * This function used to run the workqueues itself.  Now we just wait for the
 452  * helper threads to do it.
 453  */
 454 void fastcall flush_workqueue(struct workqueue_struct *wq)
 455 {
 456         preempt_disable();              /* CPU hotplug */
 457         if (is_single_threaded(wq)) {
 458                 /* Always use first cpu's area. */
 459                 flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, singlethread_cpu));
 460         } else {
 461                 int cpu;
 462
 463                 for_each_online_cpu(cpu)
 464                         flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, cpu));
 465         }
 466         preempt_enable();
 467 }
 468 EXPORT_SYMBOL_GPL(flush_workqueue);
 469
 470 static void wait_on_work(struct cpu_workqueue_struct *cwq,
 471                                 struct work_struct *work)
 472 {
 473         struct wq_barrier barr;
 474         int running = 0;
 475
 476         spin_lock_irq(&cwq->lock);
 477         if (unlikely(cwq->current_work == work)) {
 478                 init_wq_barrier(&barr);
 479                 insert_work(cwq, &barr.work, 0);
 480                 running = 1;
 481         }
 482         spin_unlock_irq(&cwq->lock);
 483
 484         if (unlikely(running)) {
 485                 mutex_unlock(&workqueue_mutex);
 486                 wait_for_completion(&barr.done);
 487                 mutex_lock(&workqueue_mutex);
 488         }
 489 }
 490
 491 /**
 492  * flush_work - block until a work_struct's callback has terminated
 493  * @wq: the workqueue on which the work is queued
 494  * @work: the work which is to be flushed
 495  *
 496  * flush_work() will attempt to cancel the work if it is queued.  If the work's
 497  * callback appears to be running, flush_work() will block until it has
 498  * completed.
 499  *
 500  * flush_work() is designed to be used when the caller is tearing down data
 501  * structures which the callback function operates upon.  It is expected that,
 502  * prior to calling flush_work(), the caller has arranged for the work to not
 503  * be requeued.
 504  */
 505 void flush_work(struct workqueue_struct *wq, struct work_struct *work)
 506 {
 507         struct cpu_workqueue_struct *cwq;
 508
 509         mutex_lock(&workqueue_mutex);
 510         cwq = get_wq_data(work);
 511         /* Was it ever queued ? */
 512         if (!cwq)
 513                 goto out;
 514
 515         /*
 516          * This work can't be re-queued, and the lock above protects us
 517          * from take_over_work(), no need to re-check that get_wq_data()
 518          * is still the same when we take cwq->lock.
 519          */
 520         spin_lock_irq(&cwq->lock);
 521         list_del_init(&work->entry);
 522         work_release(work);
 523         spin_unlock_irq(&cwq->lock);
 524
 525         if (is_single_threaded(wq)) {
 526                 /* Always use first cpu's area. */
 527                 wait_on_work(per_cpu_ptr(wq->cpu_wq, singlethread_cpu), work);
 528         } else {
 529                 int cpu;
 530
 531                 for_each_online_cpu(cpu)
 532                         wait_on_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
 533         }
 534 out:
 535         mutex_unlock(&workqueue_mutex);
 536 }
 537 EXPORT_SYMBOL_GPL(flush_work);
 538
 539 static struct task_struct *create_workqueue_thread(struct workqueue_struct *wq,
 540                                                    int cpu, int freezeable)
 541 {
 542         struct cpu_workqueue_struct *cwq = per_cpu_ptr(wq->cpu_wq, cpu);
 543         struct task_struct *p;
 544
 545         spin_lock_init(&cwq->lock);
 546         cwq->wq = wq;
 547         cwq->thread = NULL;
 548         cwq->freezeable = freezeable;
 549         INIT_LIST_HEAD(&cwq->worklist);
 550         init_waitqueue_head(&cwq->more_work);
 551
 552         if (is_single_threaded(wq))
 553                 p = kthread_create(worker_thread, cwq, "%s", wq->name);
 554         else
 555                 p = kthread_create(worker_thread, cwq, "%s/%d", wq->name, cpu);
 556         if (IS_ERR(p))
 557                 return NULL;
 558         cwq->thread = p;
 559         return p;
 560 }
 561
 562 struct workqueue_struct *__create_workqueue(const char *name,
 563                                             int singlethread, int freezeable)
 564 {
 565         int cpu, destroy = 0;
 566         struct workqueue_struct *wq;
 567         struct task_struct *p;
 568
 569         wq = kzalloc(sizeof(*wq), GFP_KERNEL);
 570         if (!wq)
 571                 return NULL;
 572
 573         wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct);
 574         if (!wq->cpu_wq) {
 575                 kfree(wq);
 576                 return NULL;
 577         }
 578
 579         wq->name = name;
 580         mutex_lock(&workqueue_mutex);
 581         if (singlethread) {
 582                 INIT_LIST_HEAD(&wq->list);
 583                 p = create_workqueue_thread(wq, singlethread_cpu, freezeable);
 584                 if (!p)
 585                         destroy = 1;
 586                 else
 587                         wake_up_process(p);
 588         } else {
 589                 list_add(&wq->list, &workqueues);
 590                 for_each_online_cpu(cpu) {
 591                         p = create_workqueue_thread(wq, cpu, freezeable);
 592                         if (p) {
 593                                 kthread_bind(p, cpu);
 594                                 wake_up_process(p);
 595                         } else
 596                                 destroy = 1;
 597                 }
 598         }
 599         mutex_unlock(&workqueue_mutex);
 600
 601         /*
 602          * Was there any error during startup? If yes then clean up:
 603          */
 604         if (destroy) {
 605                 destroy_workqueue(wq);
 606                 wq = NULL;
 607         }
 608         return wq;
 609 }
 610 EXPORT_SYMBOL_GPL(__create_workqueue);
 611
 612 static void cleanup_workqueue_thread(struct workqueue_struct *wq, int cpu)
 613 {
 614         struct cpu_workqueue_struct *cwq;
 615         unsigned long flags;
 616         struct task_struct *p;
 617
 618         cwq = per_cpu_ptr(wq->cpu_wq, cpu);
 619         spin_lock_irqsave(&cwq->lock, flags);
 620         p = cwq->thread;
 621         cwq->thread = NULL;
 622         spin_unlock_irqrestore(&cwq->lock, flags);
 623         if (p)
 624                 kthread_stop(p);
 625 }
 626
 627 /**
 628  * destroy_workqueue - safely terminate a workqueue
 629  * @wq: target workqueue
 630  *
 631  * Safely destroy a workqueue. All work currently pending will be done first.
 632  */
 633 void destroy_workqueue(struct workqueue_struct *wq)
 634 {
 635         int cpu;
 636
 637         flush_workqueue(wq);
 638
 639         /* We don't need the distraction of CPUs appearing and vanishing. */
 640         mutex_lock(&workqueue_mutex);
 641         if (is_single_threaded(wq))
 642                 cleanup_workqueue_thread(wq, singlethread_cpu);
 643         else {
 644                 for_each_online_cpu(cpu)
 645                         cleanup_workqueue_thread(wq, cpu);
 646                 list_del(&wq->list);
 647         }
 648         mutex_unlock(&workqueue_mutex);
 649         free_percpu(wq->cpu_wq);
 650         kfree(wq);
 651 }
 652 EXPORT_SYMBOL_GPL(destroy_workqueue);
 653
 654 static struct workqueue_struct *keventd_wq;
 655
 656 /**
 657  * schedule_work - put work task in global workqueue
 658  * @work: job to be done
 659  *
 660  * This puts a job in the kernel-global workqueue.
 661  */
 662 int fastcall schedule_work(struct work_struct *work)
 663 {
 664         return queue_work(keventd_wq, work);
 665 }
 666 EXPORT_SYMBOL(schedule_work);
 667
 668 /**
 669  * schedule_delayed_work - put work task in global workqueue after delay
 670  * @dwork: job to be done
 671  * @delay: number of jiffies to wait or 0 for immediate execution
 672  *
 673  * After waiting for a given time this puts a job in the kernel-global
 674  * workqueue.
 675  */
 676 int fastcall schedule_delayed_work(struct delayed_work *dwork,
 677                                         unsigned long delay)
 678 {
 679         timer_stats_timer_set_start_info(&dwork->timer);
 680         return queue_delayed_work(keventd_wq, dwork, delay);
 681 }
 682 EXPORT_SYMBOL(schedule_delayed_work);
 683
 684 /**
 685  * schedule_delayed_work_on - queue work in global workqueue on CPU after delay
 686  * @cpu: cpu to use
 687  * @dwork: job to be done
 688  * @delay: number of jiffies to wait
 689  *
 690  * After waiting for a given time this puts a job in the kernel-global
 691  * workqueue on the specified CPU.
 692  */
 693 int schedule_delayed_work_on(int cpu,
 694                         struct delayed_work *dwork, unsigned long delay)
 695 {
 696         return queue_delayed_work_on(cpu, keventd_wq, dwork, delay);
 697 }
 698 EXPORT_SYMBOL(schedule_delayed_work_on);
 699
 700 /**
 701  * schedule_on_each_cpu - call a function on each online CPU from keventd
 702  * @func: the function to call
 703  *
 704  * Returns zero on success.
 705  * Returns -ve errno on failure.
 706  *
 707  * Appears to be racy against CPU hotplug.
 708  *
 709  * schedule_on_each_cpu() is very slow.
 710  */
 711 int schedule_on_each_cpu(work_func_t func)
 712 {
 713         int cpu;
 714         struct work_struct *works;
 715
 716         works = alloc_percpu(struct work_struct);
 717         if (!works)
 718                 return -ENOMEM;
 719
 720         preempt_disable();              /* CPU hotplug */
 721         for_each_online_cpu(cpu) {
 722                 struct work_struct *work = per_cpu_ptr(works, cpu);
 723
 724                 INIT_WORK(work, func);
 725                 set_bit(WORK_STRUCT_PENDING, work_data_bits(work));
 726                 __queue_work(per_cpu_ptr(keventd_wq->cpu_wq, cpu), work);
 727         }
 728         preempt_enable();
 729         flush_workqueue(keventd_wq);
 730         free_percpu(works);
 731         return 0;
 732 }
 733
 734 void flush_scheduled_work(void)
 735 {
 736         flush_workqueue(keventd_wq);
 737 }
 738 EXPORT_SYMBOL(flush_scheduled_work);
 739
 740 void flush_work_keventd(struct work_struct *work)
 741 {
 742         flush_work(keventd_wq, work);
 743 }
 744 EXPORT_SYMBOL(flush_work_keventd);
 745
 746 /**
 747  * cancel_rearming_delayed_workqueue - reliably kill off a delayed work whose handler rearms the delayed work.
 748  * @wq:   the controlling workqueue structure
 749  * @dwork: the delayed work struct
 750  */
 751 void cancel_rearming_delayed_workqueue(struct workqueue_struct *wq,
 752                                        struct delayed_work *dwork)
 753 {
 754         while (!cancel_delayed_work(dwork))
 755                 flush_workqueue(wq);
 756 }
 757 EXPORT_SYMBOL(cancel_rearming_delayed_workqueue);
 758
 759 /**
 760  * cancel_rearming_delayed_work - reliably kill off a delayed keventd work whose handler rearms the delayed work.
 761  * @dwork: the delayed work struct
 762  */
 763 void cancel_rearming_delayed_work(struct delayed_work *dwork)
 764 {
 765         cancel_rearming_delayed_workqueue(keventd_wq, dwork);
 766 }
 767 EXPORT_SYMBOL(cancel_rearming_delayed_work);
 768
 769 /**
 770  * execute_in_process_context - reliably execute the routine with user context
 771  * @fn:         the function to execute
 772  * @ew:         guaranteed storage for the execute work structure (must
 773  *              be available when the work executes)
 774  *
 775  * Executes the function immediately if process context is available,
 776  * otherwise schedules the function for delayed execution.
 777  *
 778  * Returns:     0 - function was executed
 779  *              1 - function was scheduled for execution
 780  */
 781 int execute_in_process_context(work_func_t fn, struct execute_work *ew)
 782 {
 783         if (!in_interrupt()) {
 784                 fn(&ew->work);
 785                 return 0;
 786         }
 787
 788         INIT_WORK(&ew->work, fn);
 789         schedule_work(&ew->work);
 790
 791         return 1;
 792 }
 793 EXPORT_SYMBOL_GPL(execute_in_process_context);
 794
 795 int keventd_up(void)
 796 {
 797         return keventd_wq != NULL;
 798 }
 799
 800 int current_is_keventd(void)
 801 {
 802         struct cpu_workqueue_struct *cwq;
 803         int cpu = smp_processor_id();   /* preempt-safe: keventd is per-cpu */
 804         int ret = 0;
 805
 806         BUG_ON(!keventd_wq);
 807
 808         cwq = per_cpu_ptr(keventd_wq->cpu_wq, cpu);
 809         if (current == cwq->thread)
 810                 ret = 1;
 811
 812         return ret;
 813
 814 }
 815
 816 /* Take the work from this (downed) CPU. */
 817 static void take_over_work(struct workqueue_struct *wq, unsigned int cpu)
 818 {
 819         struct cpu_workqueue_struct *cwq = per_cpu_ptr(wq->cpu_wq, cpu);
 820         struct list_head list;
 821         struct work_struct *work;
 822
 823         spin_lock_irq(&cwq->lock);
 824         list_replace_init(&cwq->worklist, &list);
 825
 826         while (!list_empty(&list)) {
 827                 printk("Taking work for %s\n", wq->name);
 828                 work = list_entry(list.next,struct work_struct,entry);
 829                 list_del(&work->entry);
 830                 __queue_work(per_cpu_ptr(wq->cpu_wq, smp_processor_id()), work);
 831         }
 832         spin_unlock_irq(&cwq->lock);
 833 }
 834
 835 /* We're holding the cpucontrol mutex here */
 836 static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
 837                                   unsigned long action,
 838                                   void *hcpu)
 839 {
 840         unsigned int hotcpu = (unsigned long)hcpu;
 841         struct workqueue_struct *wq;
 842
 843         switch (action) {
 844         case CPU_UP_PREPARE:
 845                 mutex_lock(&workqueue_mutex);
 846                 /* Create a new workqueue thread for it. */
 847                 list_for_each_entry(wq, &workqueues, list) {
 848                         if (!create_workqueue_thread(wq, hotcpu, 0)) {
 849                                 printk("workqueue for %i failed\n", hotcpu);
 850                                 return NOTIFY_BAD;
 851                         }
 852                 }
 853                 break;
 854
 855         case CPU_ONLINE:
 856                 /* Kick off worker threads. */
 857                 list_for_each_entry(wq, &workqueues, list) {
 858                         struct cpu_workqueue_struct *cwq;
 859
 860                         cwq = per_cpu_ptr(wq->cpu_wq, hotcpu);
 861                         kthread_bind(cwq->thread, hotcpu);
 862                         wake_up_process(cwq->thread);
 863                 }
 864                 mutex_unlock(&workqueue_mutex);
 865                 break;
 866
 867         case CPU_UP_CANCELED:
 868                 list_for_each_entry(wq, &workqueues, list) {
 869                         if (!per_cpu_ptr(wq->cpu_wq, hotcpu)->thread)
 870                                 continue;
 871                         /* Unbind so it can run. */
 872                         kthread_bind(per_cpu_ptr(wq->cpu_wq, hotcpu)->thread,
 873                                      any_online_cpu(cpu_online_map));
 874                         cleanup_workqueue_thread(wq, hotcpu);
 875                 }
 876                 mutex_unlock(&workqueue_mutex);
 877                 break;
 878
 879         case CPU_DOWN_PREPARE:
 880                 mutex_lock(&workqueue_mutex);
 881                 break;
 882
 883         case CPU_DOWN_FAILED:
 884                 mutex_unlock(&workqueue_mutex);
 885                 break;
 886
 887         case CPU_DEAD:
 888                 list_for_each_entry(wq, &workqueues, list)
 889                         cleanup_workqueue_thread(wq, hotcpu);
 890                 list_for_each_entry(wq, &workqueues, list)
 891                         take_over_work(wq, hotcpu);
 892                 mutex_unlock(&workqueue_mutex);
 893                 break;
 894         }
 895
 896         return NOTIFY_OK;
 897 }
 898
 899 void init_workqueues(void)
 900 {
 901         singlethread_cpu = first_cpu(cpu_possible_map);
 902         hotcpu_notifier(workqueue_cpu_callback, 0);
 903         keventd_wq = create_workqueue("events");
 904         BUG_ON(!keventd_wq);
 905 }
 906