async-thread.c 8.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343
  1. /*
  2. * Copyright (C) 2007 Oracle. All rights reserved.
  3. *
  4. * This program is free software; you can redistribute it and/or
  5. * modify it under the terms of the GNU General Public
  6. * License v2 as published by the Free Software Foundation.
  7. *
  8. * This program is distributed in the hope that it will be useful,
  9. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  11. * General Public License for more details.
  12. *
  13. * You should have received a copy of the GNU General Public
  14. * License along with this program; if not, write to the
  15. * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  16. * Boston, MA 021110-1307, USA.
  17. */
  18. #include <linux/version.h>
  19. #include <linux/kthread.h>
  20. #include <linux/list.h>
  21. #include <linux/spinlock.h>
  22. #if LINUX_VERSION_CODE >= KERNEL_VERSION(2,6,20)
  23. # include <linux/freezer.h>
  24. #else
  25. # include <linux/sched.h>
  26. #endif
  27. #include "async-thread.h"
  28. /*
  29. * container for the kthread task pointer and the list of pending work
  30. * One of these is allocated per thread.
  31. */
  32. struct btrfs_worker_thread {
  33. /* pool we belong to */
  34. struct btrfs_workers *workers;
  35. /* list of struct btrfs_work that are waiting for service */
  36. struct list_head pending;
  37. /* list of worker threads from struct btrfs_workers */
  38. struct list_head worker_list;
  39. /* kthread */
  40. struct task_struct *task;
  41. /* number of things on the pending list */
  42. atomic_t num_pending;
  43. unsigned long sequence;
  44. /* protects the pending list. */
  45. spinlock_t lock;
  46. /* set to non-zero when this thread is already awake and kicking */
  47. int working;
  48. /* are we currently idle */
  49. int idle;
  50. };
  51. /*
  52. * helper function to move a thread onto the idle list after it
  53. * has finished some requests.
  54. */
  55. static void check_idle_worker(struct btrfs_worker_thread *worker)
  56. {
  57. if (!worker->idle && atomic_read(&worker->num_pending) <
  58. worker->workers->idle_thresh / 2) {
  59. unsigned long flags;
  60. spin_lock_irqsave(&worker->workers->lock, flags);
  61. worker->idle = 1;
  62. list_move(&worker->worker_list, &worker->workers->idle_list);
  63. spin_unlock_irqrestore(&worker->workers->lock, flags);
  64. }
  65. }
  66. /*
  67. * helper function to move a thread off the idle list after new
  68. * pending work is added.
  69. */
  70. static void check_busy_worker(struct btrfs_worker_thread *worker)
  71. {
  72. if (worker->idle && atomic_read(&worker->num_pending) >=
  73. worker->workers->idle_thresh) {
  74. unsigned long flags;
  75. spin_lock_irqsave(&worker->workers->lock, flags);
  76. worker->idle = 0;
  77. list_move_tail(&worker->worker_list,
  78. &worker->workers->worker_list);
  79. spin_unlock_irqrestore(&worker->workers->lock, flags);
  80. }
  81. }
  82. /*
  83. * main loop for servicing work items
  84. */
  85. static int worker_loop(void *arg)
  86. {
  87. struct btrfs_worker_thread *worker = arg;
  88. struct list_head *cur;
  89. struct btrfs_work *work;
  90. do {
  91. spin_lock_irq(&worker->lock);
  92. while(!list_empty(&worker->pending)) {
  93. cur = worker->pending.next;
  94. work = list_entry(cur, struct btrfs_work, list);
  95. list_del(&work->list);
  96. clear_bit(0, &work->flags);
  97. work->worker = worker;
  98. spin_unlock_irq(&worker->lock);
  99. work->func(work);
  100. atomic_dec(&worker->num_pending);
  101. spin_lock_irq(&worker->lock);
  102. check_idle_worker(worker);
  103. }
  104. worker->working = 0;
  105. if (freezing(current)) {
  106. refrigerator();
  107. } else {
  108. set_current_state(TASK_INTERRUPTIBLE);
  109. spin_unlock_irq(&worker->lock);
  110. schedule();
  111. __set_current_state(TASK_RUNNING);
  112. }
  113. } while (!kthread_should_stop());
  114. return 0;
  115. }
  116. /*
  117. * this will wait for all the worker threads to shutdown
  118. */
  119. int btrfs_stop_workers(struct btrfs_workers *workers)
  120. {
  121. struct list_head *cur;
  122. struct btrfs_worker_thread *worker;
  123. list_splice_init(&workers->idle_list, &workers->worker_list);
  124. while(!list_empty(&workers->worker_list)) {
  125. cur = workers->worker_list.next;
  126. worker = list_entry(cur, struct btrfs_worker_thread,
  127. worker_list);
  128. kthread_stop(worker->task);
  129. list_del(&worker->worker_list);
  130. kfree(worker);
  131. }
  132. return 0;
  133. }
  134. /*
  135. * simple init on struct btrfs_workers
  136. */
  137. void btrfs_init_workers(struct btrfs_workers *workers, char *name, int max)
  138. {
  139. workers->num_workers = 0;
  140. INIT_LIST_HEAD(&workers->worker_list);
  141. INIT_LIST_HEAD(&workers->idle_list);
  142. spin_lock_init(&workers->lock);
  143. workers->max_workers = max;
  144. workers->idle_thresh = 32;
  145. workers->name = name;
  146. }
  147. /*
  148. * starts new worker threads. This does not enforce the max worker
  149. * count in case you need to temporarily go past it.
  150. */
  151. int btrfs_start_workers(struct btrfs_workers *workers, int num_workers)
  152. {
  153. struct btrfs_worker_thread *worker;
  154. int ret = 0;
  155. int i;
  156. for (i = 0; i < num_workers; i++) {
  157. worker = kzalloc(sizeof(*worker), GFP_NOFS);
  158. if (!worker) {
  159. ret = -ENOMEM;
  160. goto fail;
  161. }
  162. INIT_LIST_HEAD(&worker->pending);
  163. INIT_LIST_HEAD(&worker->worker_list);
  164. spin_lock_init(&worker->lock);
  165. atomic_set(&worker->num_pending, 0);
  166. worker->task = kthread_run(worker_loop, worker,
  167. "btrfs-%s-%d", workers->name,
  168. workers->num_workers + i);
  169. worker->workers = workers;
  170. if (IS_ERR(worker->task)) {
  171. kfree(worker);
  172. ret = PTR_ERR(worker->task);
  173. goto fail;
  174. }
  175. spin_lock_irq(&workers->lock);
  176. list_add_tail(&worker->worker_list, &workers->idle_list);
  177. worker->idle = 1;
  178. workers->num_workers++;
  179. spin_unlock_irq(&workers->lock);
  180. }
  181. return 0;
  182. fail:
  183. btrfs_stop_workers(workers);
  184. return ret;
  185. }
  186. /*
  187. * run through the list and find a worker thread that doesn't have a lot
  188. * to do right now. This can return null if we aren't yet at the thread
  189. * count limit and all of the threads are busy.
  190. */
  191. static struct btrfs_worker_thread *next_worker(struct btrfs_workers *workers)
  192. {
  193. struct btrfs_worker_thread *worker;
  194. struct list_head *next;
  195. int enforce_min = workers->num_workers < workers->max_workers;
  196. /*
  197. * if we find an idle thread, don't move it to the end of the
  198. * idle list. This improves the chance that the next submission
  199. * will reuse the same thread, and maybe catch it while it is still
  200. * working
  201. */
  202. if (!list_empty(&workers->idle_list)) {
  203. next = workers->idle_list.next;
  204. worker = list_entry(next, struct btrfs_worker_thread,
  205. worker_list);
  206. return worker;
  207. }
  208. if (enforce_min || list_empty(&workers->worker_list))
  209. return NULL;
  210. /*
  211. * if we pick a busy task, move the task to the end of the list.
  212. * hopefully this will keep things somewhat evenly balanced
  213. */
  214. next = workers->worker_list.next;
  215. worker = list_entry(next, struct btrfs_worker_thread, worker_list);
  216. atomic_inc(&worker->num_pending);
  217. worker->sequence++;
  218. if (worker->sequence % workers->idle_thresh == 0)
  219. list_move_tail(next, &workers->worker_list);
  220. return worker;
  221. }
  222. static struct btrfs_worker_thread *find_worker(struct btrfs_workers *workers)
  223. {
  224. struct btrfs_worker_thread *worker;
  225. unsigned long flags;
  226. again:
  227. spin_lock_irqsave(&workers->lock, flags);
  228. worker = next_worker(workers);
  229. spin_unlock_irqrestore(&workers->lock, flags);
  230. if (!worker) {
  231. spin_lock_irqsave(&workers->lock, flags);
  232. if (workers->num_workers >= workers->max_workers) {
  233. struct list_head *fallback = NULL;
  234. /*
  235. * we have failed to find any workers, just
  236. * return the force one
  237. */
  238. if (!list_empty(&workers->worker_list))
  239. fallback = workers->worker_list.next;
  240. if (!list_empty(&workers->idle_list))
  241. fallback = workers->idle_list.next;
  242. BUG_ON(!fallback);
  243. worker = list_entry(fallback,
  244. struct btrfs_worker_thread, worker_list);
  245. spin_unlock_irqrestore(&workers->lock, flags);
  246. } else {
  247. spin_unlock_irqrestore(&workers->lock, flags);
  248. /* we're below the limit, start another worker */
  249. btrfs_start_workers(workers, 1);
  250. goto again;
  251. }
  252. }
  253. return worker;
  254. }
  255. /*
  256. * btrfs_requeue_work just puts the work item back on the tail of the list
  257. * it was taken from. It is intended for use with long running work functions
  258. * that make some progress and want to give the cpu up for others.
  259. */
  260. int btrfs_requeue_work(struct btrfs_work *work)
  261. {
  262. struct btrfs_worker_thread *worker = work->worker;
  263. unsigned long flags;
  264. if (test_and_set_bit(0, &work->flags))
  265. goto out;
  266. spin_lock_irqsave(&worker->lock, flags);
  267. atomic_inc(&worker->num_pending);
  268. list_add_tail(&work->list, &worker->pending);
  269. check_busy_worker(worker);
  270. spin_unlock_irqrestore(&worker->lock, flags);
  271. out:
  272. return 0;
  273. }
  274. /*
  275. * places a struct btrfs_work into the pending queue of one of the kthreads
  276. */
  277. int btrfs_queue_worker(struct btrfs_workers *workers, struct btrfs_work *work)
  278. {
  279. struct btrfs_worker_thread *worker;
  280. unsigned long flags;
  281. int wake = 0;
  282. /* don't requeue something already on a list */
  283. if (test_and_set_bit(0, &work->flags))
  284. goto out;
  285. worker = find_worker(workers);
  286. spin_lock_irqsave(&worker->lock, flags);
  287. atomic_inc(&worker->num_pending);
  288. check_busy_worker(worker);
  289. list_add_tail(&work->list, &worker->pending);
  290. /*
  291. * avoid calling into wake_up_process if this thread has already
  292. * been kicked
  293. */
  294. if (!worker->working)
  295. wake = 1;
  296. worker->working = 1;
  297. spin_unlock_irqrestore(&worker->lock, flags);
  298. if (wake)
  299. wake_up_process(worker->task);
  300. out:
  301. return 0;
  302. }