workqueue.c 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821
  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. #include <linux/module.h>
  19. #include <linux/kernel.h>
  20. #include <linux/sched.h>
  21. #include <linux/init.h>
  22. #include <linux/signal.h>
  23. #include <linux/completion.h>
  24. #include <linux/workqueue.h>
  25. #include <linux/slab.h>
  26. #include <linux/cpu.h>
  27. #include <linux/notifier.h>
  28. #include <linux/kthread.h>
  29. #include <linux/hardirq.h>
  30. #include <linux/mempolicy.h>
  31. #include <linux/freezer.h>
  32. #include <linux/kallsyms.h>
  33. #include <linux/debug_locks.h>
  34. /*
  35. * The per-CPU workqueue (if single thread, we always use the first
  36. * possible cpu).
  37. *
  38. * The sequence counters are for flush_scheduled_work(). It wants to wait
  39. * until all currently-scheduled works are completed, but it doesn't
  40. * want to be livelocked by new, incoming ones. So it waits until
  41. * remove_sequence is >= the insert_sequence which pertained when
  42. * flush_scheduled_work() was called.
  43. */
  44. struct cpu_workqueue_struct {
  45. spinlock_t lock;
  46. long remove_sequence; /* Least-recently added (next to run) */
  47. long insert_sequence; /* Next to add */
  48. struct list_head worklist;
  49. wait_queue_head_t more_work;
  50. wait_queue_head_t work_done;
  51. struct workqueue_struct *wq;
  52. struct task_struct *thread;
  53. int run_depth; /* Detect run_workqueue() recursion depth */
  54. int freezeable; /* Freeze the thread during suspend */
  55. } ____cacheline_aligned;
  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. /* All the per-cpu workqueues on the system, for hotplug cpu to add/remove
  66. threads to each one as cpus come/go. */
  67. static DEFINE_MUTEX(workqueue_mutex);
  68. static LIST_HEAD(workqueues);
  69. static int singlethread_cpu;
  70. /* If it's single threaded, it isn't in the list of workqueues. */
  71. static inline int is_single_threaded(struct workqueue_struct *wq)
  72. {
  73. return list_empty(&wq->list);
  74. }
  75. /*
  76. * Set the workqueue on which a work item is to be run
  77. * - Must *only* be called if the pending flag is set
  78. */
  79. static inline void set_wq_data(struct work_struct *work, void *wq)
  80. {
  81. unsigned long new;
  82. BUG_ON(!work_pending(work));
  83. new = (unsigned long) wq | (1UL << WORK_STRUCT_PENDING);
  84. new |= WORK_STRUCT_FLAG_MASK & *work_data_bits(work);
  85. atomic_long_set(&work->data, new);
  86. }
  87. static inline void *get_wq_data(struct work_struct *work)
  88. {
  89. return (void *) (atomic_long_read(&work->data) & WORK_STRUCT_WQ_DATA_MASK);
  90. }
  91. static int __run_work(struct cpu_workqueue_struct *cwq, struct work_struct *work)
  92. {
  93. int ret = 0;
  94. unsigned long flags;
  95. spin_lock_irqsave(&cwq->lock, flags);
  96. /*
  97. * We need to re-validate the work info after we've gotten
  98. * the cpu_workqueue lock. We can run the work now iff:
  99. *
  100. * - the wq_data still matches the cpu_workqueue_struct
  101. * - AND the work is still marked pending
  102. * - AND the work is still on a list (which will be this
  103. * workqueue_struct list)
  104. *
  105. * All these conditions are important, because we
  106. * need to protect against the work being run right
  107. * now on another CPU (all but the last one might be
  108. * true if it's currently running and has not been
  109. * released yet, for example).
  110. */
  111. if (get_wq_data(work) == cwq
  112. && work_pending(work)
  113. && !list_empty(&work->entry)) {
  114. work_func_t f = work->func;
  115. list_del_init(&work->entry);
  116. spin_unlock_irqrestore(&cwq->lock, flags);
  117. if (!test_bit(WORK_STRUCT_NOAUTOREL, work_data_bits(work)))
  118. work_release(work);
  119. f(work);
  120. spin_lock_irqsave(&cwq->lock, flags);
  121. cwq->remove_sequence++;
  122. wake_up(&cwq->work_done);
  123. ret = 1;
  124. }
  125. spin_unlock_irqrestore(&cwq->lock, flags);
  126. return ret;
  127. }
  128. /**
  129. * run_scheduled_work - run scheduled work synchronously
  130. * @work: work to run
  131. *
  132. * This checks if the work was pending, and runs it
  133. * synchronously if so. It returns a boolean to indicate
  134. * whether it had any scheduled work to run or not.
  135. *
  136. * NOTE! This _only_ works for normal work_structs. You
  137. * CANNOT use this for delayed work, because the wq data
  138. * for delayed work will not point properly to the per-
  139. * CPU workqueue struct, but will change!
  140. */
  141. int fastcall run_scheduled_work(struct work_struct *work)
  142. {
  143. for (;;) {
  144. struct cpu_workqueue_struct *cwq;
  145. if (!work_pending(work))
  146. return 0;
  147. if (list_empty(&work->entry))
  148. return 0;
  149. /* NOTE! This depends intimately on __queue_work! */
  150. cwq = get_wq_data(work);
  151. if (!cwq)
  152. return 0;
  153. if (__run_work(cwq, work))
  154. return 1;
  155. }
  156. }
  157. EXPORT_SYMBOL(run_scheduled_work);
  158. /* Preempt must be disabled. */
  159. static void __queue_work(struct cpu_workqueue_struct *cwq,
  160. struct work_struct *work)
  161. {
  162. unsigned long flags;
  163. spin_lock_irqsave(&cwq->lock, flags);
  164. set_wq_data(work, cwq);
  165. list_add_tail(&work->entry, &cwq->worklist);
  166. cwq->insert_sequence++;
  167. wake_up(&cwq->more_work);
  168. spin_unlock_irqrestore(&cwq->lock, flags);
  169. }
  170. /**
  171. * queue_work - queue work on a workqueue
  172. * @wq: workqueue to use
  173. * @work: work to queue
  174. *
  175. * Returns 0 if @work was already on a queue, non-zero otherwise.
  176. *
  177. * We queue the work to the CPU it was submitted, but there is no
  178. * guarantee that it will be processed by that CPU.
  179. */
  180. int fastcall queue_work(struct workqueue_struct *wq, struct work_struct *work)
  181. {
  182. int ret = 0, cpu = get_cpu();
  183. if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
  184. if (unlikely(is_single_threaded(wq)))
  185. cpu = singlethread_cpu;
  186. BUG_ON(!list_empty(&work->entry));
  187. __queue_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
  188. ret = 1;
  189. }
  190. put_cpu();
  191. return ret;
  192. }
  193. EXPORT_SYMBOL_GPL(queue_work);
  194. void delayed_work_timer_fn(unsigned long __data)
  195. {
  196. struct delayed_work *dwork = (struct delayed_work *)__data;
  197. struct workqueue_struct *wq = get_wq_data(&dwork->work);
  198. int cpu = smp_processor_id();
  199. if (unlikely(is_single_threaded(wq)))
  200. cpu = singlethread_cpu;
  201. __queue_work(per_cpu_ptr(wq->cpu_wq, cpu), &dwork->work);
  202. }
  203. /**
  204. * queue_delayed_work - queue work on a workqueue after delay
  205. * @wq: workqueue to use
  206. * @dwork: delayable work to queue
  207. * @delay: number of jiffies to wait before queueing
  208. *
  209. * Returns 0 if @work was already on a queue, non-zero otherwise.
  210. */
  211. int fastcall queue_delayed_work(struct workqueue_struct *wq,
  212. struct delayed_work *dwork, unsigned long delay)
  213. {
  214. int ret = 0;
  215. struct timer_list *timer = &dwork->timer;
  216. struct work_struct *work = &dwork->work;
  217. timer_stats_timer_set_start_info(timer);
  218. if (delay == 0)
  219. return queue_work(wq, work);
  220. if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
  221. BUG_ON(timer_pending(timer));
  222. BUG_ON(!list_empty(&work->entry));
  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(timer);
  229. ret = 1;
  230. }
  231. return ret;
  232. }
  233. EXPORT_SYMBOL_GPL(queue_delayed_work);
  234. /**
  235. * queue_delayed_work_on - queue work on specific CPU after delay
  236. * @cpu: CPU number to execute work on
  237. * @wq: workqueue to use
  238. * @dwork: work to queue
  239. * @delay: number of jiffies to wait before queueing
  240. *
  241. * Returns 0 if @work was already on a queue, non-zero otherwise.
  242. */
  243. int queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
  244. struct delayed_work *dwork, unsigned long delay)
  245. {
  246. int ret = 0;
  247. struct timer_list *timer = &dwork->timer;
  248. struct work_struct *work = &dwork->work;
  249. if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
  250. BUG_ON(timer_pending(timer));
  251. BUG_ON(!list_empty(&work->entry));
  252. /* This stores wq for the moment, for the timer_fn */
  253. set_wq_data(work, wq);
  254. timer->expires = jiffies + delay;
  255. timer->data = (unsigned long)dwork;
  256. timer->function = delayed_work_timer_fn;
  257. add_timer_on(timer, cpu);
  258. ret = 1;
  259. }
  260. return ret;
  261. }
  262. EXPORT_SYMBOL_GPL(queue_delayed_work_on);
  263. static void run_workqueue(struct cpu_workqueue_struct *cwq)
  264. {
  265. unsigned long flags;
  266. /*
  267. * Keep taking off work from the queue until
  268. * done.
  269. */
  270. spin_lock_irqsave(&cwq->lock, flags);
  271. cwq->run_depth++;
  272. if (cwq->run_depth > 3) {
  273. /* morton gets to eat his hat */
  274. printk("%s: recursion depth exceeded: %d\n",
  275. __FUNCTION__, cwq->run_depth);
  276. dump_stack();
  277. }
  278. while (!list_empty(&cwq->worklist)) {
  279. struct work_struct *work = list_entry(cwq->worklist.next,
  280. struct work_struct, entry);
  281. work_func_t f = work->func;
  282. list_del_init(cwq->worklist.next);
  283. spin_unlock_irqrestore(&cwq->lock, flags);
  284. BUG_ON(get_wq_data(work) != cwq);
  285. if (!test_bit(WORK_STRUCT_NOAUTOREL, work_data_bits(work)))
  286. work_release(work);
  287. f(work);
  288. if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
  289. printk(KERN_ERR "BUG: workqueue leaked lock or atomic: "
  290. "%s/0x%08x/%d\n",
  291. current->comm, preempt_count(),
  292. current->pid);
  293. printk(KERN_ERR " last function: ");
  294. print_symbol("%s\n", (unsigned long)f);
  295. debug_show_held_locks(current);
  296. dump_stack();
  297. }
  298. spin_lock_irqsave(&cwq->lock, flags);
  299. cwq->remove_sequence++;
  300. wake_up(&cwq->work_done);
  301. }
  302. cwq->run_depth--;
  303. spin_unlock_irqrestore(&cwq->lock, flags);
  304. }
  305. static int worker_thread(void *__cwq)
  306. {
  307. struct cpu_workqueue_struct *cwq = __cwq;
  308. DECLARE_WAITQUEUE(wait, current);
  309. struct k_sigaction sa;
  310. sigset_t blocked;
  311. if (!cwq->freezeable)
  312. current->flags |= PF_NOFREEZE;
  313. set_user_nice(current, -5);
  314. /* Block and flush all signals */
  315. sigfillset(&blocked);
  316. sigprocmask(SIG_BLOCK, &blocked, NULL);
  317. flush_signals(current);
  318. /*
  319. * We inherited MPOL_INTERLEAVE from the booting kernel.
  320. * Set MPOL_DEFAULT to insure node local allocations.
  321. */
  322. numa_default_policy();
  323. /* SIG_IGN makes children autoreap: see do_notify_parent(). */
  324. sa.sa.sa_handler = SIG_IGN;
  325. sa.sa.sa_flags = 0;
  326. siginitset(&sa.sa.sa_mask, sigmask(SIGCHLD));
  327. do_sigaction(SIGCHLD, &sa, (struct k_sigaction *)0);
  328. set_current_state(TASK_INTERRUPTIBLE);
  329. while (!kthread_should_stop()) {
  330. if (cwq->freezeable)
  331. try_to_freeze();
  332. add_wait_queue(&cwq->more_work, &wait);
  333. if (list_empty(&cwq->worklist))
  334. schedule();
  335. else
  336. __set_current_state(TASK_RUNNING);
  337. remove_wait_queue(&cwq->more_work, &wait);
  338. if (!list_empty(&cwq->worklist))
  339. run_workqueue(cwq);
  340. set_current_state(TASK_INTERRUPTIBLE);
  341. }
  342. __set_current_state(TASK_RUNNING);
  343. return 0;
  344. }
  345. static void flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
  346. {
  347. if (cwq->thread == current) {
  348. /*
  349. * Probably keventd trying to flush its own queue. So simply run
  350. * it by hand rather than deadlocking.
  351. */
  352. run_workqueue(cwq);
  353. } else {
  354. DEFINE_WAIT(wait);
  355. long sequence_needed;
  356. spin_lock_irq(&cwq->lock);
  357. sequence_needed = cwq->insert_sequence;
  358. while (sequence_needed - cwq->remove_sequence > 0) {
  359. prepare_to_wait(&cwq->work_done, &wait,
  360. TASK_UNINTERRUPTIBLE);
  361. spin_unlock_irq(&cwq->lock);
  362. schedule();
  363. spin_lock_irq(&cwq->lock);
  364. }
  365. finish_wait(&cwq->work_done, &wait);
  366. spin_unlock_irq(&cwq->lock);
  367. }
  368. }
  369. /**
  370. * flush_workqueue - ensure that any scheduled work has run to completion.
  371. * @wq: workqueue to flush
  372. *
  373. * Forces execution of the workqueue and blocks until its completion.
  374. * This is typically used in driver shutdown handlers.
  375. *
  376. * This function will sample each workqueue's current insert_sequence number and
  377. * will sleep until the head sequence is greater than or equal to that. This
  378. * means that we sleep until all works which were queued on entry have been
  379. * handled, but we are not livelocked by new incoming ones.
  380. *
  381. * This function used to run the workqueues itself. Now we just wait for the
  382. * helper threads to do it.
  383. */
  384. void fastcall flush_workqueue(struct workqueue_struct *wq)
  385. {
  386. might_sleep();
  387. if (is_single_threaded(wq)) {
  388. /* Always use first cpu's area. */
  389. flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, singlethread_cpu));
  390. } else {
  391. int cpu;
  392. mutex_lock(&workqueue_mutex);
  393. for_each_online_cpu(cpu)
  394. flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, cpu));
  395. mutex_unlock(&workqueue_mutex);
  396. }
  397. }
  398. EXPORT_SYMBOL_GPL(flush_workqueue);
  399. static struct task_struct *create_workqueue_thread(struct workqueue_struct *wq,
  400. int cpu, int freezeable)
  401. {
  402. struct cpu_workqueue_struct *cwq = per_cpu_ptr(wq->cpu_wq, cpu);
  403. struct task_struct *p;
  404. spin_lock_init(&cwq->lock);
  405. cwq->wq = wq;
  406. cwq->thread = NULL;
  407. cwq->insert_sequence = 0;
  408. cwq->remove_sequence = 0;
  409. cwq->freezeable = freezeable;
  410. INIT_LIST_HEAD(&cwq->worklist);
  411. init_waitqueue_head(&cwq->more_work);
  412. init_waitqueue_head(&cwq->work_done);
  413. if (is_single_threaded(wq))
  414. p = kthread_create(worker_thread, cwq, "%s", wq->name);
  415. else
  416. p = kthread_create(worker_thread, cwq, "%s/%d", wq->name, cpu);
  417. if (IS_ERR(p))
  418. return NULL;
  419. cwq->thread = p;
  420. return p;
  421. }
  422. struct workqueue_struct *__create_workqueue(const char *name,
  423. int singlethread, int freezeable)
  424. {
  425. int cpu, destroy = 0;
  426. struct workqueue_struct *wq;
  427. struct task_struct *p;
  428. wq = kzalloc(sizeof(*wq), GFP_KERNEL);
  429. if (!wq)
  430. return NULL;
  431. wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct);
  432. if (!wq->cpu_wq) {
  433. kfree(wq);
  434. return NULL;
  435. }
  436. wq->name = name;
  437. mutex_lock(&workqueue_mutex);
  438. if (singlethread) {
  439. INIT_LIST_HEAD(&wq->list);
  440. p = create_workqueue_thread(wq, singlethread_cpu, freezeable);
  441. if (!p)
  442. destroy = 1;
  443. else
  444. wake_up_process(p);
  445. } else {
  446. list_add(&wq->list, &workqueues);
  447. for_each_online_cpu(cpu) {
  448. p = create_workqueue_thread(wq, cpu, freezeable);
  449. if (p) {
  450. kthread_bind(p, cpu);
  451. wake_up_process(p);
  452. } else
  453. destroy = 1;
  454. }
  455. }
  456. mutex_unlock(&workqueue_mutex);
  457. /*
  458. * Was there any error during startup? If yes then clean up:
  459. */
  460. if (destroy) {
  461. destroy_workqueue(wq);
  462. wq = NULL;
  463. }
  464. return wq;
  465. }
  466. EXPORT_SYMBOL_GPL(__create_workqueue);
  467. static void cleanup_workqueue_thread(struct workqueue_struct *wq, int cpu)
  468. {
  469. struct cpu_workqueue_struct *cwq;
  470. unsigned long flags;
  471. struct task_struct *p;
  472. cwq = per_cpu_ptr(wq->cpu_wq, cpu);
  473. spin_lock_irqsave(&cwq->lock, flags);
  474. p = cwq->thread;
  475. cwq->thread = NULL;
  476. spin_unlock_irqrestore(&cwq->lock, flags);
  477. if (p)
  478. kthread_stop(p);
  479. }
  480. /**
  481. * destroy_workqueue - safely terminate a workqueue
  482. * @wq: target workqueue
  483. *
  484. * Safely destroy a workqueue. All work currently pending will be done first.
  485. */
  486. void destroy_workqueue(struct workqueue_struct *wq)
  487. {
  488. int cpu;
  489. flush_workqueue(wq);
  490. /* We don't need the distraction of CPUs appearing and vanishing. */
  491. mutex_lock(&workqueue_mutex);
  492. if (is_single_threaded(wq))
  493. cleanup_workqueue_thread(wq, singlethread_cpu);
  494. else {
  495. for_each_online_cpu(cpu)
  496. cleanup_workqueue_thread(wq, cpu);
  497. list_del(&wq->list);
  498. }
  499. mutex_unlock(&workqueue_mutex);
  500. free_percpu(wq->cpu_wq);
  501. kfree(wq);
  502. }
  503. EXPORT_SYMBOL_GPL(destroy_workqueue);
  504. static struct workqueue_struct *keventd_wq;
  505. /**
  506. * schedule_work - put work task in global workqueue
  507. * @work: job to be done
  508. *
  509. * This puts a job in the kernel-global workqueue.
  510. */
  511. int fastcall schedule_work(struct work_struct *work)
  512. {
  513. return queue_work(keventd_wq, work);
  514. }
  515. EXPORT_SYMBOL(schedule_work);
  516. /**
  517. * schedule_delayed_work - put work task in global workqueue after delay
  518. * @dwork: job to be done
  519. * @delay: number of jiffies to wait or 0 for immediate execution
  520. *
  521. * After waiting for a given time this puts a job in the kernel-global
  522. * workqueue.
  523. */
  524. int fastcall schedule_delayed_work(struct delayed_work *dwork,
  525. unsigned long delay)
  526. {
  527. timer_stats_timer_set_start_info(&dwork->timer);
  528. return queue_delayed_work(keventd_wq, dwork, delay);
  529. }
  530. EXPORT_SYMBOL(schedule_delayed_work);
  531. /**
  532. * schedule_delayed_work_on - queue work in global workqueue on CPU after delay
  533. * @cpu: cpu to use
  534. * @dwork: job to be done
  535. * @delay: number of jiffies to wait
  536. *
  537. * After waiting for a given time this puts a job in the kernel-global
  538. * workqueue on the specified CPU.
  539. */
  540. int schedule_delayed_work_on(int cpu,
  541. struct delayed_work *dwork, unsigned long delay)
  542. {
  543. return queue_delayed_work_on(cpu, keventd_wq, dwork, delay);
  544. }
  545. EXPORT_SYMBOL(schedule_delayed_work_on);
  546. /**
  547. * schedule_on_each_cpu - call a function on each online CPU from keventd
  548. * @func: the function to call
  549. *
  550. * Returns zero on success.
  551. * Returns -ve errno on failure.
  552. *
  553. * Appears to be racy against CPU hotplug.
  554. *
  555. * schedule_on_each_cpu() is very slow.
  556. */
  557. int schedule_on_each_cpu(work_func_t func)
  558. {
  559. int cpu;
  560. struct work_struct *works;
  561. works = alloc_percpu(struct work_struct);
  562. if (!works)
  563. return -ENOMEM;
  564. mutex_lock(&workqueue_mutex);
  565. for_each_online_cpu(cpu) {
  566. struct work_struct *work = per_cpu_ptr(works, cpu);
  567. INIT_WORK(work, func);
  568. set_bit(WORK_STRUCT_PENDING, work_data_bits(work));
  569. __queue_work(per_cpu_ptr(keventd_wq->cpu_wq, cpu), work);
  570. }
  571. mutex_unlock(&workqueue_mutex);
  572. flush_workqueue(keventd_wq);
  573. free_percpu(works);
  574. return 0;
  575. }
  576. void flush_scheduled_work(void)
  577. {
  578. flush_workqueue(keventd_wq);
  579. }
  580. EXPORT_SYMBOL(flush_scheduled_work);
  581. /**
  582. * cancel_rearming_delayed_workqueue - reliably kill off a delayed work whose handler rearms the delayed work.
  583. * @wq: the controlling workqueue structure
  584. * @dwork: the delayed work struct
  585. */
  586. void cancel_rearming_delayed_workqueue(struct workqueue_struct *wq,
  587. struct delayed_work *dwork)
  588. {
  589. while (!cancel_delayed_work(dwork))
  590. flush_workqueue(wq);
  591. }
  592. EXPORT_SYMBOL(cancel_rearming_delayed_workqueue);
  593. /**
  594. * cancel_rearming_delayed_work - reliably kill off a delayed keventd work whose handler rearms the delayed work.
  595. * @dwork: the delayed work struct
  596. */
  597. void cancel_rearming_delayed_work(struct delayed_work *dwork)
  598. {
  599. cancel_rearming_delayed_workqueue(keventd_wq, dwork);
  600. }
  601. EXPORT_SYMBOL(cancel_rearming_delayed_work);
  602. /**
  603. * execute_in_process_context - reliably execute the routine with user context
  604. * @fn: the function to execute
  605. * @ew: guaranteed storage for the execute work structure (must
  606. * be available when the work executes)
  607. *
  608. * Executes the function immediately if process context is available,
  609. * otherwise schedules the function for delayed execution.
  610. *
  611. * Returns: 0 - function was executed
  612. * 1 - function was scheduled for execution
  613. */
  614. int execute_in_process_context(work_func_t fn, struct execute_work *ew)
  615. {
  616. if (!in_interrupt()) {
  617. fn(&ew->work);
  618. return 0;
  619. }
  620. INIT_WORK(&ew->work, fn);
  621. schedule_work(&ew->work);
  622. return 1;
  623. }
  624. EXPORT_SYMBOL_GPL(execute_in_process_context);
  625. int keventd_up(void)
  626. {
  627. return keventd_wq != NULL;
  628. }
  629. int current_is_keventd(void)
  630. {
  631. struct cpu_workqueue_struct *cwq;
  632. int cpu = smp_processor_id(); /* preempt-safe: keventd is per-cpu */
  633. int ret = 0;
  634. BUG_ON(!keventd_wq);
  635. cwq = per_cpu_ptr(keventd_wq->cpu_wq, cpu);
  636. if (current == cwq->thread)
  637. ret = 1;
  638. return ret;
  639. }
  640. /* Take the work from this (downed) CPU. */
  641. static void take_over_work(struct workqueue_struct *wq, unsigned int cpu)
  642. {
  643. struct cpu_workqueue_struct *cwq = per_cpu_ptr(wq->cpu_wq, cpu);
  644. struct list_head list;
  645. struct work_struct *work;
  646. spin_lock_irq(&cwq->lock);
  647. list_replace_init(&cwq->worklist, &list);
  648. while (!list_empty(&list)) {
  649. printk("Taking work for %s\n", wq->name);
  650. work = list_entry(list.next,struct work_struct,entry);
  651. list_del(&work->entry);
  652. __queue_work(per_cpu_ptr(wq->cpu_wq, smp_processor_id()), work);
  653. }
  654. spin_unlock_irq(&cwq->lock);
  655. }
  656. /* We're holding the cpucontrol mutex here */
  657. static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
  658. unsigned long action,
  659. void *hcpu)
  660. {
  661. unsigned int hotcpu = (unsigned long)hcpu;
  662. struct workqueue_struct *wq;
  663. switch (action) {
  664. case CPU_UP_PREPARE:
  665. mutex_lock(&workqueue_mutex);
  666. /* Create a new workqueue thread for it. */
  667. list_for_each_entry(wq, &workqueues, list) {
  668. if (!create_workqueue_thread(wq, hotcpu, 0)) {
  669. printk("workqueue for %i failed\n", hotcpu);
  670. return NOTIFY_BAD;
  671. }
  672. }
  673. break;
  674. case CPU_ONLINE:
  675. /* Kick off worker threads. */
  676. list_for_each_entry(wq, &workqueues, list) {
  677. struct cpu_workqueue_struct *cwq;
  678. cwq = per_cpu_ptr(wq->cpu_wq, hotcpu);
  679. kthread_bind(cwq->thread, hotcpu);
  680. wake_up_process(cwq->thread);
  681. }
  682. mutex_unlock(&workqueue_mutex);
  683. break;
  684. case CPU_UP_CANCELED:
  685. list_for_each_entry(wq, &workqueues, list) {
  686. if (!per_cpu_ptr(wq->cpu_wq, hotcpu)->thread)
  687. continue;
  688. /* Unbind so it can run. */
  689. kthread_bind(per_cpu_ptr(wq->cpu_wq, hotcpu)->thread,
  690. any_online_cpu(cpu_online_map));
  691. cleanup_workqueue_thread(wq, hotcpu);
  692. }
  693. mutex_unlock(&workqueue_mutex);
  694. break;
  695. case CPU_DOWN_PREPARE:
  696. mutex_lock(&workqueue_mutex);
  697. break;
  698. case CPU_DOWN_FAILED:
  699. mutex_unlock(&workqueue_mutex);
  700. break;
  701. case CPU_DEAD:
  702. list_for_each_entry(wq, &workqueues, list)
  703. cleanup_workqueue_thread(wq, hotcpu);
  704. list_for_each_entry(wq, &workqueues, list)
  705. take_over_work(wq, hotcpu);
  706. mutex_unlock(&workqueue_mutex);
  707. break;
  708. }
  709. return NOTIFY_OK;
  710. }
  711. void init_workqueues(void)
  712. {
  713. singlethread_cpu = first_cpu(cpu_possible_map);
  714. hotcpu_notifier(workqueue_cpu_callback, 0);
  715. keventd_wq = create_workqueue("events");
  716. BUG_ON(!keventd_wq);
  717. }