async-thread.c 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419
  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. # include <linux/freezer.h>
  23. #include "async-thread.h"
  24. #define WORK_QUEUED_BIT 0
  25. #define WORK_DONE_BIT 1
  26. #define WORK_ORDER_DONE_BIT 2
  27. /*
  28. * container for the kthread task pointer and the list of pending work
  29. * One of these is allocated per thread.
  30. */
  31. struct btrfs_worker_thread {
  32. /* pool we belong to */
  33. struct btrfs_workers *workers;
  34. /* list of struct btrfs_work that are waiting for service */
  35. struct list_head pending;
  36. /* list of worker threads from struct btrfs_workers */
  37. struct list_head worker_list;
  38. /* kthread */
  39. struct task_struct *task;
  40. /* number of things on the pending list */
  41. atomic_t num_pending;
  42. unsigned long sequence;
  43. /* protects the pending list. */
  44. spinlock_t lock;
  45. /* set to non-zero when this thread is already awake and kicking */
  46. int working;
  47. /* are we currently idle */
  48. int idle;
  49. };
  50. /*
  51. * helper function to move a thread onto the idle list after it
  52. * has finished some requests.
  53. */
  54. static void check_idle_worker(struct btrfs_worker_thread *worker)
  55. {
  56. if (!worker->idle && atomic_read(&worker->num_pending) <
  57. worker->workers->idle_thresh / 2) {
  58. unsigned long flags;
  59. spin_lock_irqsave(&worker->workers->lock, flags);
  60. worker->idle = 1;
  61. list_move(&worker->worker_list, &worker->workers->idle_list);
  62. spin_unlock_irqrestore(&worker->workers->lock, flags);
  63. }
  64. }
  65. /*
  66. * helper function to move a thread off the idle list after new
  67. * pending work is added.
  68. */
  69. static void check_busy_worker(struct btrfs_worker_thread *worker)
  70. {
  71. if (worker->idle && atomic_read(&worker->num_pending) >=
  72. worker->workers->idle_thresh) {
  73. unsigned long flags;
  74. spin_lock_irqsave(&worker->workers->lock, flags);
  75. worker->idle = 0;
  76. list_move_tail(&worker->worker_list,
  77. &worker->workers->worker_list);
  78. spin_unlock_irqrestore(&worker->workers->lock, flags);
  79. }
  80. }
  81. static noinline int run_ordered_completions(struct btrfs_workers *workers,
  82. struct btrfs_work *work)
  83. {
  84. unsigned long flags;
  85. if (!workers->ordered)
  86. return 0;
  87. set_bit(WORK_DONE_BIT, &work->flags);
  88. spin_lock_irqsave(&workers->lock, flags);
  89. while(!list_empty(&workers->order_list)) {
  90. work = list_entry(workers->order_list.next,
  91. struct btrfs_work, order_list);
  92. if (!test_bit(WORK_DONE_BIT, &work->flags))
  93. break;
  94. /* we are going to call the ordered done function, but
  95. * we leave the work item on the list as a barrier so
  96. * that later work items that are done don't have their
  97. * functions called before this one returns
  98. */
  99. if (test_and_set_bit(WORK_ORDER_DONE_BIT, &work->flags))
  100. break;
  101. spin_unlock_irqrestore(&workers->lock, flags);
  102. work->ordered_func(work);
  103. /* now take the lock again and call the freeing code */
  104. spin_lock_irqsave(&workers->lock, flags);
  105. list_del(&work->order_list);
  106. work->ordered_free(work);
  107. }
  108. spin_unlock_irqrestore(&workers->lock, flags);
  109. return 0;
  110. }
  111. /*
  112. * main loop for servicing work items
  113. */
  114. static int worker_loop(void *arg)
  115. {
  116. struct btrfs_worker_thread *worker = arg;
  117. struct list_head *cur;
  118. struct btrfs_work *work;
  119. do {
  120. spin_lock_irq(&worker->lock);
  121. while(!list_empty(&worker->pending)) {
  122. cur = worker->pending.next;
  123. work = list_entry(cur, struct btrfs_work, list);
  124. list_del(&work->list);
  125. clear_bit(WORK_QUEUED_BIT, &work->flags);
  126. work->worker = worker;
  127. spin_unlock_irq(&worker->lock);
  128. work->func(work);
  129. atomic_dec(&worker->num_pending);
  130. /*
  131. * unless this is an ordered work queue,
  132. * 'work' was probably freed by func above.
  133. */
  134. run_ordered_completions(worker->workers, work);
  135. spin_lock_irq(&worker->lock);
  136. check_idle_worker(worker);
  137. }
  138. worker->working = 0;
  139. if (freezing(current)) {
  140. refrigerator();
  141. } else {
  142. set_current_state(TASK_INTERRUPTIBLE);
  143. spin_unlock_irq(&worker->lock);
  144. if (!kthread_should_stop())
  145. schedule();
  146. __set_current_state(TASK_RUNNING);
  147. }
  148. } while (!kthread_should_stop());
  149. return 0;
  150. }
  151. /*
  152. * this will wait for all the worker threads to shutdown
  153. */
  154. int btrfs_stop_workers(struct btrfs_workers *workers)
  155. {
  156. struct list_head *cur;
  157. struct btrfs_worker_thread *worker;
  158. list_splice_init(&workers->idle_list, &workers->worker_list);
  159. while(!list_empty(&workers->worker_list)) {
  160. cur = workers->worker_list.next;
  161. worker = list_entry(cur, struct btrfs_worker_thread,
  162. worker_list);
  163. kthread_stop(worker->task);
  164. list_del(&worker->worker_list);
  165. kfree(worker);
  166. }
  167. return 0;
  168. }
  169. /*
  170. * simple init on struct btrfs_workers
  171. */
  172. void btrfs_init_workers(struct btrfs_workers *workers, char *name, int max)
  173. {
  174. workers->num_workers = 0;
  175. INIT_LIST_HEAD(&workers->worker_list);
  176. INIT_LIST_HEAD(&workers->idle_list);
  177. INIT_LIST_HEAD(&workers->order_list);
  178. spin_lock_init(&workers->lock);
  179. workers->max_workers = max;
  180. workers->idle_thresh = 32;
  181. workers->name = name;
  182. workers->ordered = 0;
  183. }
  184. /*
  185. * starts new worker threads. This does not enforce the max worker
  186. * count in case you need to temporarily go past it.
  187. */
  188. int btrfs_start_workers(struct btrfs_workers *workers, int num_workers)
  189. {
  190. struct btrfs_worker_thread *worker;
  191. int ret = 0;
  192. int i;
  193. for (i = 0; i < num_workers; i++) {
  194. worker = kzalloc(sizeof(*worker), GFP_NOFS);
  195. if (!worker) {
  196. ret = -ENOMEM;
  197. goto fail;
  198. }
  199. INIT_LIST_HEAD(&worker->pending);
  200. INIT_LIST_HEAD(&worker->worker_list);
  201. spin_lock_init(&worker->lock);
  202. atomic_set(&worker->num_pending, 0);
  203. worker->task = kthread_run(worker_loop, worker,
  204. "btrfs-%s-%d", workers->name,
  205. workers->num_workers + i);
  206. worker->workers = workers;
  207. if (IS_ERR(worker->task)) {
  208. kfree(worker);
  209. ret = PTR_ERR(worker->task);
  210. goto fail;
  211. }
  212. spin_lock_irq(&workers->lock);
  213. list_add_tail(&worker->worker_list, &workers->idle_list);
  214. worker->idle = 1;
  215. workers->num_workers++;
  216. spin_unlock_irq(&workers->lock);
  217. }
  218. return 0;
  219. fail:
  220. btrfs_stop_workers(workers);
  221. return ret;
  222. }
  223. /*
  224. * run through the list and find a worker thread that doesn't have a lot
  225. * to do right now. This can return null if we aren't yet at the thread
  226. * count limit and all of the threads are busy.
  227. */
  228. static struct btrfs_worker_thread *next_worker(struct btrfs_workers *workers)
  229. {
  230. struct btrfs_worker_thread *worker;
  231. struct list_head *next;
  232. int enforce_min = workers->num_workers < workers->max_workers;
  233. /*
  234. * if we find an idle thread, don't move it to the end of the
  235. * idle list. This improves the chance that the next submission
  236. * will reuse the same thread, and maybe catch it while it is still
  237. * working
  238. */
  239. if (!list_empty(&workers->idle_list)) {
  240. next = workers->idle_list.next;
  241. worker = list_entry(next, struct btrfs_worker_thread,
  242. worker_list);
  243. return worker;
  244. }
  245. if (enforce_min || list_empty(&workers->worker_list))
  246. return NULL;
  247. /*
  248. * if we pick a busy task, move the task to the end of the list.
  249. * hopefully this will keep things somewhat evenly balanced.
  250. * Do the move in batches based on the sequence number. This groups
  251. * requests submitted at roughly the same time onto the same worker.
  252. */
  253. next = workers->worker_list.next;
  254. worker = list_entry(next, struct btrfs_worker_thread, worker_list);
  255. atomic_inc(&worker->num_pending);
  256. worker->sequence++;
  257. if (worker->sequence % workers->idle_thresh == 0)
  258. list_move_tail(next, &workers->worker_list);
  259. return worker;
  260. }
  261. /*
  262. * selects a worker thread to take the next job. This will either find
  263. * an idle worker, start a new worker up to the max count, or just return
  264. * one of the existing busy workers.
  265. */
  266. static struct btrfs_worker_thread *find_worker(struct btrfs_workers *workers)
  267. {
  268. struct btrfs_worker_thread *worker;
  269. unsigned long flags;
  270. again:
  271. spin_lock_irqsave(&workers->lock, flags);
  272. worker = next_worker(workers);
  273. spin_unlock_irqrestore(&workers->lock, flags);
  274. if (!worker) {
  275. spin_lock_irqsave(&workers->lock, flags);
  276. if (workers->num_workers >= workers->max_workers) {
  277. struct list_head *fallback = NULL;
  278. /*
  279. * we have failed to find any workers, just
  280. * return the force one
  281. */
  282. if (!list_empty(&workers->worker_list))
  283. fallback = workers->worker_list.next;
  284. if (!list_empty(&workers->idle_list))
  285. fallback = workers->idle_list.next;
  286. BUG_ON(!fallback);
  287. worker = list_entry(fallback,
  288. struct btrfs_worker_thread, worker_list);
  289. spin_unlock_irqrestore(&workers->lock, flags);
  290. } else {
  291. spin_unlock_irqrestore(&workers->lock, flags);
  292. /* we're below the limit, start another worker */
  293. btrfs_start_workers(workers, 1);
  294. goto again;
  295. }
  296. }
  297. return worker;
  298. }
  299. /*
  300. * btrfs_requeue_work just puts the work item back on the tail of the list
  301. * it was taken from. It is intended for use with long running work functions
  302. * that make some progress and want to give the cpu up for others.
  303. */
  304. int btrfs_requeue_work(struct btrfs_work *work)
  305. {
  306. struct btrfs_worker_thread *worker = work->worker;
  307. unsigned long flags;
  308. if (test_and_set_bit(WORK_QUEUED_BIT, &work->flags))
  309. goto out;
  310. spin_lock_irqsave(&worker->lock, flags);
  311. atomic_inc(&worker->num_pending);
  312. list_add_tail(&work->list, &worker->pending);
  313. /* by definition we're busy, take ourselves off the idle
  314. * list
  315. */
  316. if (worker->idle) {
  317. spin_lock_irqsave(&worker->workers->lock, flags);
  318. worker->idle = 0;
  319. list_move_tail(&worker->worker_list,
  320. &worker->workers->worker_list);
  321. spin_unlock_irqrestore(&worker->workers->lock, flags);
  322. }
  323. spin_unlock_irqrestore(&worker->lock, flags);
  324. out:
  325. return 0;
  326. }
  327. /*
  328. * places a struct btrfs_work into the pending queue of one of the kthreads
  329. */
  330. int btrfs_queue_worker(struct btrfs_workers *workers, struct btrfs_work *work)
  331. {
  332. struct btrfs_worker_thread *worker;
  333. unsigned long flags;
  334. int wake = 0;
  335. /* don't requeue something already on a list */
  336. if (test_and_set_bit(WORK_QUEUED_BIT, &work->flags))
  337. goto out;
  338. worker = find_worker(workers);
  339. if (workers->ordered) {
  340. spin_lock_irqsave(&workers->lock, flags);
  341. list_add_tail(&work->order_list, &workers->order_list);
  342. spin_unlock_irqrestore(&workers->lock, flags);
  343. } else {
  344. INIT_LIST_HEAD(&work->order_list);
  345. }
  346. spin_lock_irqsave(&worker->lock, flags);
  347. atomic_inc(&worker->num_pending);
  348. check_busy_worker(worker);
  349. list_add_tail(&work->list, &worker->pending);
  350. /*
  351. * avoid calling into wake_up_process if this thread has already
  352. * been kicked
  353. */
  354. if (!worker->working)
  355. wake = 1;
  356. worker->working = 1;
  357. spin_unlock_irqrestore(&worker->lock, flags);
  358. if (wake)
  359. wake_up_process(worker->task);
  360. out:
  361. return 0;
  362. }