workqueue.c 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852
  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. struct cpu_workqueue_struct {
  39. spinlock_t lock;
  40. struct list_head worklist;
  41. wait_queue_head_t more_work;
  42. struct work_struct *current_work;
  43. struct workqueue_struct *wq;
  44. struct task_struct *thread;
  45. int run_depth; /* Detect run_workqueue() recursion depth */
  46. } ____cacheline_aligned;
  47. /*
  48. * The externally visible workqueue abstraction is an array of
  49. * per-CPU workqueues:
  50. */
  51. struct workqueue_struct {
  52. struct cpu_workqueue_struct *cpu_wq;
  53. struct list_head list;
  54. const char *name;
  55. int singlethread;
  56. int freezeable; /* Freeze threads during suspend */
  57. };
  58. /* All the per-cpu workqueues on the system, for hotplug cpu to add/remove
  59. threads to each one as cpus come/go. */
  60. static DEFINE_MUTEX(workqueue_mutex);
  61. static LIST_HEAD(workqueues);
  62. static int singlethread_cpu __read_mostly;
  63. static cpumask_t cpu_singlethread_map __read_mostly;
  64. /*
  65. * _cpu_down() first removes CPU from cpu_online_map, then CPU_DEAD
  66. * flushes cwq->worklist. This means that flush_workqueue/wait_on_work
  67. * which comes in between can't use for_each_online_cpu(). We could
  68. * use cpu_possible_map, the cpumask below is more a documentation
  69. * than optimization.
  70. */
  71. static cpumask_t cpu_populated_map __read_mostly;
  72. /* If it's single threaded, it isn't in the list of workqueues. */
  73. static inline int is_single_threaded(struct workqueue_struct *wq)
  74. {
  75. return wq->singlethread;
  76. }
  77. static const cpumask_t *wq_cpu_map(struct workqueue_struct *wq)
  78. {
  79. return is_single_threaded(wq)
  80. ? &cpu_singlethread_map : &cpu_populated_map;
  81. }
  82. static
  83. struct cpu_workqueue_struct *wq_per_cpu(struct workqueue_struct *wq, int cpu)
  84. {
  85. if (unlikely(is_single_threaded(wq)))
  86. cpu = singlethread_cpu;
  87. return per_cpu_ptr(wq->cpu_wq, cpu);
  88. }
  89. /*
  90. * Set the workqueue on which a work item is to be run
  91. * - Must *only* be called if the pending flag is set
  92. */
  93. static inline void set_wq_data(struct work_struct *work,
  94. struct cpu_workqueue_struct *cwq)
  95. {
  96. unsigned long new;
  97. BUG_ON(!work_pending(work));
  98. new = (unsigned long) cwq | (1UL << WORK_STRUCT_PENDING);
  99. new |= WORK_STRUCT_FLAG_MASK & *work_data_bits(work);
  100. atomic_long_set(&work->data, new);
  101. }
  102. static inline
  103. struct cpu_workqueue_struct *get_wq_data(struct work_struct *work)
  104. {
  105. return (void *) (atomic_long_read(&work->data) & WORK_STRUCT_WQ_DATA_MASK);
  106. }
  107. static void insert_work(struct cpu_workqueue_struct *cwq,
  108. struct work_struct *work, int tail)
  109. {
  110. set_wq_data(work, cwq);
  111. /*
  112. * Ensure that we get the right work->data if we see the
  113. * result of list_add() below, see try_to_grab_pending().
  114. */
  115. smp_wmb();
  116. if (tail)
  117. list_add_tail(&work->entry, &cwq->worklist);
  118. else
  119. list_add(&work->entry, &cwq->worklist);
  120. wake_up(&cwq->more_work);
  121. }
  122. /* Preempt must be disabled. */
  123. static void __queue_work(struct cpu_workqueue_struct *cwq,
  124. struct work_struct *work)
  125. {
  126. unsigned long flags;
  127. spin_lock_irqsave(&cwq->lock, flags);
  128. insert_work(cwq, work, 1);
  129. spin_unlock_irqrestore(&cwq->lock, flags);
  130. }
  131. /**
  132. * queue_work - queue work on a workqueue
  133. * @wq: workqueue to use
  134. * @work: work to queue
  135. *
  136. * Returns 0 if @work was already on a queue, non-zero otherwise.
  137. *
  138. * We queue the work to the CPU it was submitted, but there is no
  139. * guarantee that it will be processed by that CPU.
  140. */
  141. int fastcall queue_work(struct workqueue_struct *wq, struct work_struct *work)
  142. {
  143. int ret = 0;
  144. if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
  145. BUG_ON(!list_empty(&work->entry));
  146. __queue_work(wq_per_cpu(wq, get_cpu()), work);
  147. put_cpu();
  148. ret = 1;
  149. }
  150. return ret;
  151. }
  152. EXPORT_SYMBOL_GPL(queue_work);
  153. void delayed_work_timer_fn(unsigned long __data)
  154. {
  155. struct delayed_work *dwork = (struct delayed_work *)__data;
  156. struct cpu_workqueue_struct *cwq = get_wq_data(&dwork->work);
  157. struct workqueue_struct *wq = cwq->wq;
  158. __queue_work(wq_per_cpu(wq, smp_processor_id()), &dwork->work);
  159. }
  160. /**
  161. * queue_delayed_work - queue work on a workqueue after delay
  162. * @wq: workqueue to use
  163. * @dwork: delayable work to queue
  164. * @delay: number of jiffies to wait before queueing
  165. *
  166. * Returns 0 if @work was already on a queue, non-zero otherwise.
  167. */
  168. int fastcall queue_delayed_work(struct workqueue_struct *wq,
  169. struct delayed_work *dwork, unsigned long delay)
  170. {
  171. timer_stats_timer_set_start_info(&dwork->timer);
  172. if (delay == 0)
  173. return queue_work(wq, &dwork->work);
  174. return queue_delayed_work_on(-1, wq, dwork, delay);
  175. }
  176. EXPORT_SYMBOL_GPL(queue_delayed_work);
  177. /**
  178. * queue_delayed_work_on - queue work on specific CPU after delay
  179. * @cpu: CPU number to execute work on
  180. * @wq: workqueue to use
  181. * @dwork: work to queue
  182. * @delay: number of jiffies to wait before queueing
  183. *
  184. * Returns 0 if @work was already on a queue, non-zero otherwise.
  185. */
  186. int queue_delayed_work_on(int cpu, struct workqueue_struct *wq,
  187. struct delayed_work *dwork, unsigned long delay)
  188. {
  189. int ret = 0;
  190. struct timer_list *timer = &dwork->timer;
  191. struct work_struct *work = &dwork->work;
  192. if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work))) {
  193. BUG_ON(timer_pending(timer));
  194. BUG_ON(!list_empty(&work->entry));
  195. /* This stores cwq for the moment, for the timer_fn */
  196. set_wq_data(work, wq_per_cpu(wq, raw_smp_processor_id()));
  197. timer->expires = jiffies + delay;
  198. timer->data = (unsigned long)dwork;
  199. timer->function = delayed_work_timer_fn;
  200. if (unlikely(cpu >= 0))
  201. add_timer_on(timer, cpu);
  202. else
  203. add_timer(timer);
  204. ret = 1;
  205. }
  206. return ret;
  207. }
  208. EXPORT_SYMBOL_GPL(queue_delayed_work_on);
  209. static void run_workqueue(struct cpu_workqueue_struct *cwq)
  210. {
  211. spin_lock_irq(&cwq->lock);
  212. cwq->run_depth++;
  213. if (cwq->run_depth > 3) {
  214. /* morton gets to eat his hat */
  215. printk("%s: recursion depth exceeded: %d\n",
  216. __FUNCTION__, cwq->run_depth);
  217. dump_stack();
  218. }
  219. while (!list_empty(&cwq->worklist)) {
  220. struct work_struct *work = list_entry(cwq->worklist.next,
  221. struct work_struct, entry);
  222. work_func_t f = work->func;
  223. cwq->current_work = work;
  224. list_del_init(cwq->worklist.next);
  225. spin_unlock_irq(&cwq->lock);
  226. BUG_ON(get_wq_data(work) != cwq);
  227. work_clear_pending(work);
  228. f(work);
  229. if (unlikely(in_atomic() || lockdep_depth(current) > 0)) {
  230. printk(KERN_ERR "BUG: workqueue leaked lock or atomic: "
  231. "%s/0x%08x/%d\n",
  232. current->comm, preempt_count(),
  233. current->pid);
  234. printk(KERN_ERR " last function: ");
  235. print_symbol("%s\n", (unsigned long)f);
  236. debug_show_held_locks(current);
  237. dump_stack();
  238. }
  239. spin_lock_irq(&cwq->lock);
  240. cwq->current_work = NULL;
  241. }
  242. cwq->run_depth--;
  243. spin_unlock_irq(&cwq->lock);
  244. }
  245. static int worker_thread(void *__cwq)
  246. {
  247. struct cpu_workqueue_struct *cwq = __cwq;
  248. DEFINE_WAIT(wait);
  249. if (cwq->wq->freezeable)
  250. set_freezable();
  251. set_user_nice(current, -5);
  252. for (;;) {
  253. prepare_to_wait(&cwq->more_work, &wait, TASK_INTERRUPTIBLE);
  254. if (!freezing(current) &&
  255. !kthread_should_stop() &&
  256. list_empty(&cwq->worklist))
  257. schedule();
  258. finish_wait(&cwq->more_work, &wait);
  259. try_to_freeze();
  260. if (kthread_should_stop())
  261. break;
  262. run_workqueue(cwq);
  263. }
  264. return 0;
  265. }
  266. struct wq_barrier {
  267. struct work_struct work;
  268. struct completion done;
  269. };
  270. static void wq_barrier_func(struct work_struct *work)
  271. {
  272. struct wq_barrier *barr = container_of(work, struct wq_barrier, work);
  273. complete(&barr->done);
  274. }
  275. static void insert_wq_barrier(struct cpu_workqueue_struct *cwq,
  276. struct wq_barrier *barr, int tail)
  277. {
  278. INIT_WORK(&barr->work, wq_barrier_func);
  279. __set_bit(WORK_STRUCT_PENDING, work_data_bits(&barr->work));
  280. init_completion(&barr->done);
  281. insert_work(cwq, &barr->work, tail);
  282. }
  283. static int flush_cpu_workqueue(struct cpu_workqueue_struct *cwq)
  284. {
  285. int active;
  286. if (cwq->thread == current) {
  287. /*
  288. * Probably keventd trying to flush its own queue. So simply run
  289. * it by hand rather than deadlocking.
  290. */
  291. run_workqueue(cwq);
  292. active = 1;
  293. } else {
  294. struct wq_barrier barr;
  295. active = 0;
  296. spin_lock_irq(&cwq->lock);
  297. if (!list_empty(&cwq->worklist) || cwq->current_work != NULL) {
  298. insert_wq_barrier(cwq, &barr, 1);
  299. active = 1;
  300. }
  301. spin_unlock_irq(&cwq->lock);
  302. if (active)
  303. wait_for_completion(&barr.done);
  304. }
  305. return active;
  306. }
  307. /**
  308. * flush_workqueue - ensure that any scheduled work has run to completion.
  309. * @wq: workqueue to flush
  310. *
  311. * Forces execution of the workqueue and blocks until its completion.
  312. * This is typically used in driver shutdown handlers.
  313. *
  314. * We sleep until all works which were queued on entry have been handled,
  315. * but we are not livelocked by new incoming ones.
  316. *
  317. * This function used to run the workqueues itself. Now we just wait for the
  318. * helper threads to do it.
  319. */
  320. void fastcall flush_workqueue(struct workqueue_struct *wq)
  321. {
  322. const cpumask_t *cpu_map = wq_cpu_map(wq);
  323. int cpu;
  324. might_sleep();
  325. for_each_cpu_mask(cpu, *cpu_map)
  326. flush_cpu_workqueue(per_cpu_ptr(wq->cpu_wq, cpu));
  327. }
  328. EXPORT_SYMBOL_GPL(flush_workqueue);
  329. /*
  330. * Upon a successful return (>= 0), the caller "owns" WORK_STRUCT_PENDING bit,
  331. * so this work can't be re-armed in any way.
  332. */
  333. static int try_to_grab_pending(struct work_struct *work)
  334. {
  335. struct cpu_workqueue_struct *cwq;
  336. int ret = -1;
  337. if (!test_and_set_bit(WORK_STRUCT_PENDING, work_data_bits(work)))
  338. return 0;
  339. /*
  340. * The queueing is in progress, or it is already queued. Try to
  341. * steal it from ->worklist without clearing WORK_STRUCT_PENDING.
  342. */
  343. cwq = get_wq_data(work);
  344. if (!cwq)
  345. return ret;
  346. spin_lock_irq(&cwq->lock);
  347. if (!list_empty(&work->entry)) {
  348. /*
  349. * This work is queued, but perhaps we locked the wrong cwq.
  350. * In that case we must see the new value after rmb(), see
  351. * insert_work()->wmb().
  352. */
  353. smp_rmb();
  354. if (cwq == get_wq_data(work)) {
  355. list_del_init(&work->entry);
  356. ret = 1;
  357. }
  358. }
  359. spin_unlock_irq(&cwq->lock);
  360. return ret;
  361. }
  362. static void wait_on_cpu_work(struct cpu_workqueue_struct *cwq,
  363. struct work_struct *work)
  364. {
  365. struct wq_barrier barr;
  366. int running = 0;
  367. spin_lock_irq(&cwq->lock);
  368. if (unlikely(cwq->current_work == work)) {
  369. insert_wq_barrier(cwq, &barr, 0);
  370. running = 1;
  371. }
  372. spin_unlock_irq(&cwq->lock);
  373. if (unlikely(running))
  374. wait_for_completion(&barr.done);
  375. }
  376. static void wait_on_work(struct work_struct *work)
  377. {
  378. struct cpu_workqueue_struct *cwq;
  379. struct workqueue_struct *wq;
  380. const cpumask_t *cpu_map;
  381. int cpu;
  382. might_sleep();
  383. cwq = get_wq_data(work);
  384. if (!cwq)
  385. return;
  386. wq = cwq->wq;
  387. cpu_map = wq_cpu_map(wq);
  388. for_each_cpu_mask(cpu, *cpu_map)
  389. wait_on_cpu_work(per_cpu_ptr(wq->cpu_wq, cpu), work);
  390. }
  391. static int __cancel_work_timer(struct work_struct *work,
  392. struct timer_list* timer)
  393. {
  394. int ret;
  395. do {
  396. ret = (timer && likely(del_timer(timer)));
  397. if (!ret)
  398. ret = try_to_grab_pending(work);
  399. wait_on_work(work);
  400. } while (unlikely(ret < 0));
  401. work_clear_pending(work);
  402. return ret;
  403. }
  404. /**
  405. * cancel_work_sync - block until a work_struct's callback has terminated
  406. * @work: the work which is to be flushed
  407. *
  408. * Returns true if @work was pending.
  409. *
  410. * cancel_work_sync() will cancel the work if it is queued. If the work's
  411. * callback appears to be running, cancel_work_sync() will block until it
  412. * has completed.
  413. *
  414. * It is possible to use this function if the work re-queues itself. It can
  415. * cancel the work even if it migrates to another workqueue, however in that
  416. * case it only guarantees that work->func() has completed on the last queued
  417. * workqueue.
  418. *
  419. * cancel_work_sync(&delayed_work->work) should be used only if ->timer is not
  420. * pending, otherwise it goes into a busy-wait loop until the timer expires.
  421. *
  422. * The caller must ensure that workqueue_struct on which this work was last
  423. * queued can't be destroyed before this function returns.
  424. */
  425. int cancel_work_sync(struct work_struct *work)
  426. {
  427. return __cancel_work_timer(work, NULL);
  428. }
  429. EXPORT_SYMBOL_GPL(cancel_work_sync);
  430. /**
  431. * cancel_delayed_work_sync - reliably kill off a delayed work.
  432. * @dwork: the delayed work struct
  433. *
  434. * Returns true if @dwork was pending.
  435. *
  436. * It is possible to use this function if @dwork rearms itself via queue_work()
  437. * or queue_delayed_work(). See also the comment for cancel_work_sync().
  438. */
  439. int cancel_delayed_work_sync(struct delayed_work *dwork)
  440. {
  441. return __cancel_work_timer(&dwork->work, &dwork->timer);
  442. }
  443. EXPORT_SYMBOL(cancel_delayed_work_sync);
  444. static struct workqueue_struct *keventd_wq __read_mostly;
  445. /**
  446. * schedule_work - put work task in global workqueue
  447. * @work: job to be done
  448. *
  449. * This puts a job in the kernel-global workqueue.
  450. */
  451. int fastcall schedule_work(struct work_struct *work)
  452. {
  453. return queue_work(keventd_wq, work);
  454. }
  455. EXPORT_SYMBOL(schedule_work);
  456. /**
  457. * schedule_delayed_work - put work task in global workqueue after delay
  458. * @dwork: job to be done
  459. * @delay: number of jiffies to wait or 0 for immediate execution
  460. *
  461. * After waiting for a given time this puts a job in the kernel-global
  462. * workqueue.
  463. */
  464. int fastcall schedule_delayed_work(struct delayed_work *dwork,
  465. unsigned long delay)
  466. {
  467. timer_stats_timer_set_start_info(&dwork->timer);
  468. return queue_delayed_work(keventd_wq, dwork, delay);
  469. }
  470. EXPORT_SYMBOL(schedule_delayed_work);
  471. /**
  472. * schedule_delayed_work_on - queue work in global workqueue on CPU after delay
  473. * @cpu: cpu to use
  474. * @dwork: job to be done
  475. * @delay: number of jiffies to wait
  476. *
  477. * After waiting for a given time this puts a job in the kernel-global
  478. * workqueue on the specified CPU.
  479. */
  480. int schedule_delayed_work_on(int cpu,
  481. struct delayed_work *dwork, unsigned long delay)
  482. {
  483. return queue_delayed_work_on(cpu, keventd_wq, dwork, delay);
  484. }
  485. EXPORT_SYMBOL(schedule_delayed_work_on);
  486. /**
  487. * schedule_on_each_cpu - call a function on each online CPU from keventd
  488. * @func: the function to call
  489. *
  490. * Returns zero on success.
  491. * Returns -ve errno on failure.
  492. *
  493. * Appears to be racy against CPU hotplug.
  494. *
  495. * schedule_on_each_cpu() is very slow.
  496. */
  497. int schedule_on_each_cpu(work_func_t func)
  498. {
  499. int cpu;
  500. struct work_struct *works;
  501. works = alloc_percpu(struct work_struct);
  502. if (!works)
  503. return -ENOMEM;
  504. preempt_disable(); /* CPU hotplug */
  505. for_each_online_cpu(cpu) {
  506. struct work_struct *work = per_cpu_ptr(works, cpu);
  507. INIT_WORK(work, func);
  508. set_bit(WORK_STRUCT_PENDING, work_data_bits(work));
  509. __queue_work(per_cpu_ptr(keventd_wq->cpu_wq, cpu), work);
  510. }
  511. preempt_enable();
  512. flush_workqueue(keventd_wq);
  513. free_percpu(works);
  514. return 0;
  515. }
  516. void flush_scheduled_work(void)
  517. {
  518. flush_workqueue(keventd_wq);
  519. }
  520. EXPORT_SYMBOL(flush_scheduled_work);
  521. /**
  522. * execute_in_process_context - reliably execute the routine with user context
  523. * @fn: the function to execute
  524. * @ew: guaranteed storage for the execute work structure (must
  525. * be available when the work executes)
  526. *
  527. * Executes the function immediately if process context is available,
  528. * otherwise schedules the function for delayed execution.
  529. *
  530. * Returns: 0 - function was executed
  531. * 1 - function was scheduled for execution
  532. */
  533. int execute_in_process_context(work_func_t fn, struct execute_work *ew)
  534. {
  535. if (!in_interrupt()) {
  536. fn(&ew->work);
  537. return 0;
  538. }
  539. INIT_WORK(&ew->work, fn);
  540. schedule_work(&ew->work);
  541. return 1;
  542. }
  543. EXPORT_SYMBOL_GPL(execute_in_process_context);
  544. int keventd_up(void)
  545. {
  546. return keventd_wq != NULL;
  547. }
  548. int current_is_keventd(void)
  549. {
  550. struct cpu_workqueue_struct *cwq;
  551. int cpu = smp_processor_id(); /* preempt-safe: keventd is per-cpu */
  552. int ret = 0;
  553. BUG_ON(!keventd_wq);
  554. cwq = per_cpu_ptr(keventd_wq->cpu_wq, cpu);
  555. if (current == cwq->thread)
  556. ret = 1;
  557. return ret;
  558. }
  559. static struct cpu_workqueue_struct *
  560. init_cpu_workqueue(struct workqueue_struct *wq, int cpu)
  561. {
  562. struct cpu_workqueue_struct *cwq = per_cpu_ptr(wq->cpu_wq, cpu);
  563. cwq->wq = wq;
  564. spin_lock_init(&cwq->lock);
  565. INIT_LIST_HEAD(&cwq->worklist);
  566. init_waitqueue_head(&cwq->more_work);
  567. return cwq;
  568. }
  569. static int create_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu)
  570. {
  571. struct workqueue_struct *wq = cwq->wq;
  572. const char *fmt = is_single_threaded(wq) ? "%s" : "%s/%d";
  573. struct task_struct *p;
  574. p = kthread_create(worker_thread, cwq, fmt, wq->name, cpu);
  575. /*
  576. * Nobody can add the work_struct to this cwq,
  577. * if (caller is __create_workqueue)
  578. * nobody should see this wq
  579. * else // caller is CPU_UP_PREPARE
  580. * cpu is not on cpu_online_map
  581. * so we can abort safely.
  582. */
  583. if (IS_ERR(p))
  584. return PTR_ERR(p);
  585. cwq->thread = p;
  586. return 0;
  587. }
  588. static void start_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu)
  589. {
  590. struct task_struct *p = cwq->thread;
  591. if (p != NULL) {
  592. if (cpu >= 0)
  593. kthread_bind(p, cpu);
  594. wake_up_process(p);
  595. }
  596. }
  597. struct workqueue_struct *__create_workqueue(const char *name,
  598. int singlethread, int freezeable)
  599. {
  600. struct workqueue_struct *wq;
  601. struct cpu_workqueue_struct *cwq;
  602. int err = 0, cpu;
  603. wq = kzalloc(sizeof(*wq), GFP_KERNEL);
  604. if (!wq)
  605. return NULL;
  606. wq->cpu_wq = alloc_percpu(struct cpu_workqueue_struct);
  607. if (!wq->cpu_wq) {
  608. kfree(wq);
  609. return NULL;
  610. }
  611. wq->name = name;
  612. wq->singlethread = singlethread;
  613. wq->freezeable = freezeable;
  614. INIT_LIST_HEAD(&wq->list);
  615. if (singlethread) {
  616. cwq = init_cpu_workqueue(wq, singlethread_cpu);
  617. err = create_workqueue_thread(cwq, singlethread_cpu);
  618. start_workqueue_thread(cwq, -1);
  619. } else {
  620. mutex_lock(&workqueue_mutex);
  621. list_add(&wq->list, &workqueues);
  622. for_each_possible_cpu(cpu) {
  623. cwq = init_cpu_workqueue(wq, cpu);
  624. if (err || !cpu_online(cpu))
  625. continue;
  626. err = create_workqueue_thread(cwq, cpu);
  627. start_workqueue_thread(cwq, cpu);
  628. }
  629. mutex_unlock(&workqueue_mutex);
  630. }
  631. if (err) {
  632. destroy_workqueue(wq);
  633. wq = NULL;
  634. }
  635. return wq;
  636. }
  637. EXPORT_SYMBOL_GPL(__create_workqueue);
  638. static void cleanup_workqueue_thread(struct cpu_workqueue_struct *cwq, int cpu)
  639. {
  640. /*
  641. * Our caller is either destroy_workqueue() or CPU_DEAD,
  642. * workqueue_mutex protects cwq->thread
  643. */
  644. if (cwq->thread == NULL)
  645. return;
  646. flush_cpu_workqueue(cwq);
  647. /*
  648. * If the caller is CPU_DEAD and cwq->worklist was not empty,
  649. * a concurrent flush_workqueue() can insert a barrier after us.
  650. * However, in that case run_workqueue() won't return and check
  651. * kthread_should_stop() until it flushes all work_struct's.
  652. * When ->worklist becomes empty it is safe to exit because no
  653. * more work_structs can be queued on this cwq: flush_workqueue
  654. * checks list_empty(), and a "normal" queue_work() can't use
  655. * a dead CPU.
  656. */
  657. kthread_stop(cwq->thread);
  658. cwq->thread = NULL;
  659. }
  660. /**
  661. * destroy_workqueue - safely terminate a workqueue
  662. * @wq: target workqueue
  663. *
  664. * Safely destroy a workqueue. All work currently pending will be done first.
  665. */
  666. void destroy_workqueue(struct workqueue_struct *wq)
  667. {
  668. const cpumask_t *cpu_map = wq_cpu_map(wq);
  669. struct cpu_workqueue_struct *cwq;
  670. int cpu;
  671. mutex_lock(&workqueue_mutex);
  672. list_del(&wq->list);
  673. mutex_unlock(&workqueue_mutex);
  674. for_each_cpu_mask(cpu, *cpu_map) {
  675. cwq = per_cpu_ptr(wq->cpu_wq, cpu);
  676. cleanup_workqueue_thread(cwq, cpu);
  677. }
  678. free_percpu(wq->cpu_wq);
  679. kfree(wq);
  680. }
  681. EXPORT_SYMBOL_GPL(destroy_workqueue);
  682. static int __devinit workqueue_cpu_callback(struct notifier_block *nfb,
  683. unsigned long action,
  684. void *hcpu)
  685. {
  686. unsigned int cpu = (unsigned long)hcpu;
  687. struct cpu_workqueue_struct *cwq;
  688. struct workqueue_struct *wq;
  689. action &= ~CPU_TASKS_FROZEN;
  690. switch (action) {
  691. case CPU_LOCK_ACQUIRE:
  692. mutex_lock(&workqueue_mutex);
  693. return NOTIFY_OK;
  694. case CPU_LOCK_RELEASE:
  695. mutex_unlock(&workqueue_mutex);
  696. return NOTIFY_OK;
  697. case CPU_UP_PREPARE:
  698. cpu_set(cpu, cpu_populated_map);
  699. }
  700. list_for_each_entry(wq, &workqueues, list) {
  701. cwq = per_cpu_ptr(wq->cpu_wq, cpu);
  702. switch (action) {
  703. case CPU_UP_PREPARE:
  704. if (!create_workqueue_thread(cwq, cpu))
  705. break;
  706. printk(KERN_ERR "workqueue for %i failed\n", cpu);
  707. return NOTIFY_BAD;
  708. case CPU_ONLINE:
  709. start_workqueue_thread(cwq, cpu);
  710. break;
  711. case CPU_UP_CANCELED:
  712. start_workqueue_thread(cwq, -1);
  713. case CPU_DEAD:
  714. cleanup_workqueue_thread(cwq, cpu);
  715. break;
  716. }
  717. }
  718. return NOTIFY_OK;
  719. }
  720. void __init init_workqueues(void)
  721. {
  722. cpu_populated_map = cpu_online_map;
  723. singlethread_cpu = first_cpu(cpu_possible_map);
  724. cpu_singlethread_map = cpumask_of_cpu(singlethread_cpu);
  725. hotcpu_notifier(workqueue_cpu_callback, 0);
  726. keventd_wq = create_workqueue("events");
  727. BUG_ON(!keventd_wq);
  728. }