workqueue.c 20 KB

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