async-thread.c 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715
  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/kthread.h>
  19. #include <linux/slab.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. #define WORK_HIGH_PRIO_BIT 3
  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. struct list_head prio_pending;
  38. /* list of worker threads from struct btrfs_workers */
  39. struct list_head worker_list;
  40. /* kthread */
  41. struct task_struct *task;
  42. /* number of things on the pending list */
  43. atomic_t num_pending;
  44. /* reference counter for this struct */
  45. atomic_t refs;
  46. unsigned long sequence;
  47. /* protects the pending list. */
  48. spinlock_t lock;
  49. /* set to non-zero when this thread is already awake and kicking */
  50. int working;
  51. /* are we currently idle */
  52. int idle;
  53. };
  54. static int __btrfs_start_workers(struct btrfs_workers *workers);
  55. /*
  56. * btrfs_start_workers uses kthread_run, which can block waiting for memory
  57. * for a very long time. It will actually throttle on page writeback,
  58. * and so it may not make progress until after our btrfs worker threads
  59. * process all of the pending work structs in their queue
  60. *
  61. * This means we can't use btrfs_start_workers from inside a btrfs worker
  62. * thread that is used as part of cleaning dirty memory, which pretty much
  63. * involves all of the worker threads.
  64. *
  65. * Instead we have a helper queue who never has more than one thread
  66. * where we scheduler thread start operations. This worker_start struct
  67. * is used to contain the work and hold a pointer to the queue that needs
  68. * another worker.
  69. */
  70. struct worker_start {
  71. struct btrfs_work work;
  72. struct btrfs_workers *queue;
  73. };
  74. static void start_new_worker_func(struct btrfs_work *work)
  75. {
  76. struct worker_start *start;
  77. start = container_of(work, struct worker_start, work);
  78. __btrfs_start_workers(start->queue);
  79. kfree(start);
  80. }
  81. /*
  82. * helper function to move a thread onto the idle list after it
  83. * has finished some requests.
  84. */
  85. static void check_idle_worker(struct btrfs_worker_thread *worker)
  86. {
  87. if (!worker->idle && atomic_read(&worker->num_pending) <
  88. worker->workers->idle_thresh / 2) {
  89. unsigned long flags;
  90. spin_lock_irqsave(&worker->workers->lock, flags);
  91. worker->idle = 1;
  92. /* the list may be empty if the worker is just starting */
  93. if (!list_empty(&worker->worker_list)) {
  94. list_move(&worker->worker_list,
  95. &worker->workers->idle_list);
  96. }
  97. spin_unlock_irqrestore(&worker->workers->lock, flags);
  98. }
  99. }
  100. /*
  101. * helper function to move a thread off the idle list after new
  102. * pending work is added.
  103. */
  104. static void check_busy_worker(struct btrfs_worker_thread *worker)
  105. {
  106. if (worker->idle && atomic_read(&worker->num_pending) >=
  107. worker->workers->idle_thresh) {
  108. unsigned long flags;
  109. spin_lock_irqsave(&worker->workers->lock, flags);
  110. worker->idle = 0;
  111. if (!list_empty(&worker->worker_list)) {
  112. list_move_tail(&worker->worker_list,
  113. &worker->workers->worker_list);
  114. }
  115. spin_unlock_irqrestore(&worker->workers->lock, flags);
  116. }
  117. }
  118. static void check_pending_worker_creates(struct btrfs_worker_thread *worker)
  119. {
  120. struct btrfs_workers *workers = worker->workers;
  121. struct worker_start *start;
  122. unsigned long flags;
  123. rmb();
  124. if (!workers->atomic_start_pending)
  125. return;
  126. start = kzalloc(sizeof(*start), GFP_NOFS);
  127. if (!start)
  128. return;
  129. start->work.func = start_new_worker_func;
  130. start->queue = workers;
  131. spin_lock_irqsave(&workers->lock, flags);
  132. if (!workers->atomic_start_pending)
  133. goto out;
  134. workers->atomic_start_pending = 0;
  135. if (workers->num_workers + workers->num_workers_starting >=
  136. workers->max_workers)
  137. goto out;
  138. workers->num_workers_starting += 1;
  139. spin_unlock_irqrestore(&workers->lock, flags);
  140. btrfs_queue_worker(workers->atomic_worker_start, &start->work);
  141. return;
  142. out:
  143. kfree(start);
  144. spin_unlock_irqrestore(&workers->lock, flags);
  145. }
  146. static noinline void run_ordered_completions(struct btrfs_workers *workers,
  147. struct btrfs_work *work)
  148. {
  149. if (!workers->ordered)
  150. return;
  151. set_bit(WORK_DONE_BIT, &work->flags);
  152. spin_lock(&workers->order_lock);
  153. while (1) {
  154. if (!list_empty(&workers->prio_order_list)) {
  155. work = list_entry(workers->prio_order_list.next,
  156. struct btrfs_work, order_list);
  157. } else if (!list_empty(&workers->order_list)) {
  158. work = list_entry(workers->order_list.next,
  159. struct btrfs_work, order_list);
  160. } else {
  161. break;
  162. }
  163. if (!test_bit(WORK_DONE_BIT, &work->flags))
  164. break;
  165. /* we are going to call the ordered done function, but
  166. * we leave the work item on the list as a barrier so
  167. * that later work items that are done don't have their
  168. * functions called before this one returns
  169. */
  170. if (test_and_set_bit(WORK_ORDER_DONE_BIT, &work->flags))
  171. break;
  172. spin_unlock(&workers->order_lock);
  173. work->ordered_func(work);
  174. /* now take the lock again and drop our item from the list */
  175. spin_lock(&workers->order_lock);
  176. list_del(&work->order_list);
  177. spin_unlock(&workers->order_lock);
  178. /*
  179. * we don't want to call the ordered free functions
  180. * with the lock held though
  181. */
  182. work->ordered_free(work);
  183. spin_lock(&workers->order_lock);
  184. }
  185. spin_unlock(&workers->order_lock);
  186. }
  187. static void put_worker(struct btrfs_worker_thread *worker)
  188. {
  189. if (atomic_dec_and_test(&worker->refs))
  190. kfree(worker);
  191. }
  192. static int try_worker_shutdown(struct btrfs_worker_thread *worker)
  193. {
  194. int freeit = 0;
  195. spin_lock_irq(&worker->lock);
  196. spin_lock(&worker->workers->lock);
  197. if (worker->workers->num_workers > 1 &&
  198. worker->idle &&
  199. !worker->working &&
  200. !list_empty(&worker->worker_list) &&
  201. list_empty(&worker->prio_pending) &&
  202. list_empty(&worker->pending) &&
  203. atomic_read(&worker->num_pending) == 0) {
  204. freeit = 1;
  205. list_del_init(&worker->worker_list);
  206. worker->workers->num_workers--;
  207. }
  208. spin_unlock(&worker->workers->lock);
  209. spin_unlock_irq(&worker->lock);
  210. if (freeit)
  211. put_worker(worker);
  212. return freeit;
  213. }
  214. static struct btrfs_work *get_next_work(struct btrfs_worker_thread *worker,
  215. struct list_head *prio_head,
  216. struct list_head *head)
  217. {
  218. struct btrfs_work *work = NULL;
  219. struct list_head *cur = NULL;
  220. if(!list_empty(prio_head))
  221. cur = prio_head->next;
  222. smp_mb();
  223. if (!list_empty(&worker->prio_pending))
  224. goto refill;
  225. if (!list_empty(head))
  226. cur = head->next;
  227. if (cur)
  228. goto out;
  229. refill:
  230. spin_lock_irq(&worker->lock);
  231. list_splice_tail_init(&worker->prio_pending, prio_head);
  232. list_splice_tail_init(&worker->pending, head);
  233. if (!list_empty(prio_head))
  234. cur = prio_head->next;
  235. else if (!list_empty(head))
  236. cur = head->next;
  237. spin_unlock_irq(&worker->lock);
  238. if (!cur)
  239. goto out_fail;
  240. out:
  241. work = list_entry(cur, struct btrfs_work, list);
  242. out_fail:
  243. return work;
  244. }
  245. /*
  246. * main loop for servicing work items
  247. */
  248. static int worker_loop(void *arg)
  249. {
  250. struct btrfs_worker_thread *worker = arg;
  251. struct list_head head;
  252. struct list_head prio_head;
  253. struct btrfs_work *work;
  254. INIT_LIST_HEAD(&head);
  255. INIT_LIST_HEAD(&prio_head);
  256. do {
  257. again:
  258. while (1) {
  259. work = get_next_work(worker, &prio_head, &head);
  260. if (!work)
  261. break;
  262. list_del(&work->list);
  263. clear_bit(WORK_QUEUED_BIT, &work->flags);
  264. work->worker = worker;
  265. work->func(work);
  266. atomic_dec(&worker->num_pending);
  267. /*
  268. * unless this is an ordered work queue,
  269. * 'work' was probably freed by func above.
  270. */
  271. run_ordered_completions(worker->workers, work);
  272. check_pending_worker_creates(worker);
  273. cond_resched();
  274. }
  275. spin_lock_irq(&worker->lock);
  276. check_idle_worker(worker);
  277. if (freezing(current)) {
  278. worker->working = 0;
  279. spin_unlock_irq(&worker->lock);
  280. try_to_freeze();
  281. } else {
  282. spin_unlock_irq(&worker->lock);
  283. if (!kthread_should_stop()) {
  284. cpu_relax();
  285. /*
  286. * we've dropped the lock, did someone else
  287. * jump_in?
  288. */
  289. smp_mb();
  290. if (!list_empty(&worker->pending) ||
  291. !list_empty(&worker->prio_pending))
  292. continue;
  293. /*
  294. * this short schedule allows more work to
  295. * come in without the queue functions
  296. * needing to go through wake_up_process()
  297. *
  298. * worker->working is still 1, so nobody
  299. * is going to try and wake us up
  300. */
  301. schedule_timeout(1);
  302. smp_mb();
  303. if (!list_empty(&worker->pending) ||
  304. !list_empty(&worker->prio_pending))
  305. continue;
  306. if (kthread_should_stop())
  307. break;
  308. /* still no more work?, sleep for real */
  309. spin_lock_irq(&worker->lock);
  310. set_current_state(TASK_INTERRUPTIBLE);
  311. if (!list_empty(&worker->pending) ||
  312. !list_empty(&worker->prio_pending)) {
  313. spin_unlock_irq(&worker->lock);
  314. set_current_state(TASK_RUNNING);
  315. goto again;
  316. }
  317. /*
  318. * this makes sure we get a wakeup when someone
  319. * adds something new to the queue
  320. */
  321. worker->working = 0;
  322. spin_unlock_irq(&worker->lock);
  323. if (!kthread_should_stop()) {
  324. schedule_timeout(HZ * 120);
  325. if (!worker->working &&
  326. try_worker_shutdown(worker)) {
  327. return 0;
  328. }
  329. }
  330. }
  331. __set_current_state(TASK_RUNNING);
  332. }
  333. } while (!kthread_should_stop());
  334. return 0;
  335. }
  336. /*
  337. * this will wait for all the worker threads to shutdown
  338. */
  339. void btrfs_stop_workers(struct btrfs_workers *workers)
  340. {
  341. struct list_head *cur;
  342. struct btrfs_worker_thread *worker;
  343. int can_stop;
  344. spin_lock_irq(&workers->lock);
  345. list_splice_init(&workers->idle_list, &workers->worker_list);
  346. while (!list_empty(&workers->worker_list)) {
  347. cur = workers->worker_list.next;
  348. worker = list_entry(cur, struct btrfs_worker_thread,
  349. worker_list);
  350. atomic_inc(&worker->refs);
  351. workers->num_workers -= 1;
  352. if (!list_empty(&worker->worker_list)) {
  353. list_del_init(&worker->worker_list);
  354. put_worker(worker);
  355. can_stop = 1;
  356. } else
  357. can_stop = 0;
  358. spin_unlock_irq(&workers->lock);
  359. if (can_stop)
  360. kthread_stop(worker->task);
  361. spin_lock_irq(&workers->lock);
  362. put_worker(worker);
  363. }
  364. spin_unlock_irq(&workers->lock);
  365. }
  366. /*
  367. * simple init on struct btrfs_workers
  368. */
  369. void btrfs_init_workers(struct btrfs_workers *workers, char *name, int max,
  370. struct btrfs_workers *async_helper)
  371. {
  372. workers->num_workers = 0;
  373. workers->num_workers_starting = 0;
  374. INIT_LIST_HEAD(&workers->worker_list);
  375. INIT_LIST_HEAD(&workers->idle_list);
  376. INIT_LIST_HEAD(&workers->order_list);
  377. INIT_LIST_HEAD(&workers->prio_order_list);
  378. spin_lock_init(&workers->lock);
  379. spin_lock_init(&workers->order_lock);
  380. workers->max_workers = max;
  381. workers->idle_thresh = 32;
  382. workers->name = name;
  383. workers->ordered = 0;
  384. workers->atomic_start_pending = 0;
  385. workers->atomic_worker_start = async_helper;
  386. }
  387. /*
  388. * starts new worker threads. This does not enforce the max worker
  389. * count in case you need to temporarily go past it.
  390. */
  391. static int __btrfs_start_workers(struct btrfs_workers *workers)
  392. {
  393. struct btrfs_worker_thread *worker;
  394. int ret = 0;
  395. worker = kzalloc(sizeof(*worker), GFP_NOFS);
  396. if (!worker) {
  397. ret = -ENOMEM;
  398. goto fail;
  399. }
  400. INIT_LIST_HEAD(&worker->pending);
  401. INIT_LIST_HEAD(&worker->prio_pending);
  402. INIT_LIST_HEAD(&worker->worker_list);
  403. spin_lock_init(&worker->lock);
  404. atomic_set(&worker->num_pending, 0);
  405. atomic_set(&worker->refs, 1);
  406. worker->workers = workers;
  407. worker->task = kthread_run(worker_loop, worker,
  408. "btrfs-%s-%d", workers->name,
  409. workers->num_workers + 1);
  410. if (IS_ERR(worker->task)) {
  411. ret = PTR_ERR(worker->task);
  412. kfree(worker);
  413. goto fail;
  414. }
  415. spin_lock_irq(&workers->lock);
  416. list_add_tail(&worker->worker_list, &workers->idle_list);
  417. worker->idle = 1;
  418. workers->num_workers++;
  419. workers->num_workers_starting--;
  420. WARN_ON(workers->num_workers_starting < 0);
  421. spin_unlock_irq(&workers->lock);
  422. return 0;
  423. fail:
  424. spin_lock_irq(&workers->lock);
  425. workers->num_workers_starting--;
  426. spin_unlock_irq(&workers->lock);
  427. return ret;
  428. }
  429. int btrfs_start_workers(struct btrfs_workers *workers)
  430. {
  431. spin_lock_irq(&workers->lock);
  432. workers->num_workers_starting++;
  433. spin_unlock_irq(&workers->lock);
  434. return __btrfs_start_workers(workers);
  435. }
  436. /*
  437. * run through the list and find a worker thread that doesn't have a lot
  438. * to do right now. This can return null if we aren't yet at the thread
  439. * count limit and all of the threads are busy.
  440. */
  441. static struct btrfs_worker_thread *next_worker(struct btrfs_workers *workers)
  442. {
  443. struct btrfs_worker_thread *worker;
  444. struct list_head *next;
  445. int enforce_min;
  446. enforce_min = (workers->num_workers + workers->num_workers_starting) <
  447. workers->max_workers;
  448. /*
  449. * if we find an idle thread, don't move it to the end of the
  450. * idle list. This improves the chance that the next submission
  451. * will reuse the same thread, and maybe catch it while it is still
  452. * working
  453. */
  454. if (!list_empty(&workers->idle_list)) {
  455. next = workers->idle_list.next;
  456. worker = list_entry(next, struct btrfs_worker_thread,
  457. worker_list);
  458. return worker;
  459. }
  460. if (enforce_min || list_empty(&workers->worker_list))
  461. return NULL;
  462. /*
  463. * if we pick a busy task, move the task to the end of the list.
  464. * hopefully this will keep things somewhat evenly balanced.
  465. * Do the move in batches based on the sequence number. This groups
  466. * requests submitted at roughly the same time onto the same worker.
  467. */
  468. next = workers->worker_list.next;
  469. worker = list_entry(next, struct btrfs_worker_thread, worker_list);
  470. worker->sequence++;
  471. if (worker->sequence % workers->idle_thresh == 0)
  472. list_move_tail(next, &workers->worker_list);
  473. return worker;
  474. }
  475. /*
  476. * selects a worker thread to take the next job. This will either find
  477. * an idle worker, start a new worker up to the max count, or just return
  478. * one of the existing busy workers.
  479. */
  480. static struct btrfs_worker_thread *find_worker(struct btrfs_workers *workers)
  481. {
  482. struct btrfs_worker_thread *worker;
  483. unsigned long flags;
  484. struct list_head *fallback;
  485. int ret;
  486. spin_lock_irqsave(&workers->lock, flags);
  487. again:
  488. worker = next_worker(workers);
  489. if (!worker) {
  490. if (workers->num_workers + workers->num_workers_starting >=
  491. workers->max_workers) {
  492. goto fallback;
  493. } else if (workers->atomic_worker_start) {
  494. workers->atomic_start_pending = 1;
  495. goto fallback;
  496. } else {
  497. workers->num_workers_starting++;
  498. spin_unlock_irqrestore(&workers->lock, flags);
  499. /* we're below the limit, start another worker */
  500. ret = __btrfs_start_workers(workers);
  501. spin_lock_irqsave(&workers->lock, flags);
  502. if (ret)
  503. goto fallback;
  504. goto again;
  505. }
  506. }
  507. goto found;
  508. fallback:
  509. fallback = NULL;
  510. /*
  511. * we have failed to find any workers, just
  512. * return the first one we can find.
  513. */
  514. if (!list_empty(&workers->worker_list))
  515. fallback = workers->worker_list.next;
  516. if (!list_empty(&workers->idle_list))
  517. fallback = workers->idle_list.next;
  518. BUG_ON(!fallback);
  519. worker = list_entry(fallback,
  520. struct btrfs_worker_thread, worker_list);
  521. found:
  522. /*
  523. * this makes sure the worker doesn't exit before it is placed
  524. * onto a busy/idle list
  525. */
  526. atomic_inc(&worker->num_pending);
  527. spin_unlock_irqrestore(&workers->lock, flags);
  528. return worker;
  529. }
  530. /*
  531. * btrfs_requeue_work just puts the work item back on the tail of the list
  532. * it was taken from. It is intended for use with long running work functions
  533. * that make some progress and want to give the cpu up for others.
  534. */
  535. void btrfs_requeue_work(struct btrfs_work *work)
  536. {
  537. struct btrfs_worker_thread *worker = work->worker;
  538. unsigned long flags;
  539. int wake = 0;
  540. if (test_and_set_bit(WORK_QUEUED_BIT, &work->flags))
  541. return;
  542. spin_lock_irqsave(&worker->lock, flags);
  543. if (test_bit(WORK_HIGH_PRIO_BIT, &work->flags))
  544. list_add_tail(&work->list, &worker->prio_pending);
  545. else
  546. list_add_tail(&work->list, &worker->pending);
  547. atomic_inc(&worker->num_pending);
  548. /* by definition we're busy, take ourselves off the idle
  549. * list
  550. */
  551. if (worker->idle) {
  552. spin_lock(&worker->workers->lock);
  553. worker->idle = 0;
  554. list_move_tail(&worker->worker_list,
  555. &worker->workers->worker_list);
  556. spin_unlock(&worker->workers->lock);
  557. }
  558. if (!worker->working) {
  559. wake = 1;
  560. worker->working = 1;
  561. }
  562. if (wake)
  563. wake_up_process(worker->task);
  564. spin_unlock_irqrestore(&worker->lock, flags);
  565. }
  566. void btrfs_set_work_high_prio(struct btrfs_work *work)
  567. {
  568. set_bit(WORK_HIGH_PRIO_BIT, &work->flags);
  569. }
  570. /*
  571. * places a struct btrfs_work into the pending queue of one of the kthreads
  572. */
  573. void btrfs_queue_worker(struct btrfs_workers *workers, struct btrfs_work *work)
  574. {
  575. struct btrfs_worker_thread *worker;
  576. unsigned long flags;
  577. int wake = 0;
  578. /* don't requeue something already on a list */
  579. if (test_and_set_bit(WORK_QUEUED_BIT, &work->flags))
  580. return;
  581. worker = find_worker(workers);
  582. if (workers->ordered) {
  583. /*
  584. * you're not allowed to do ordered queues from an
  585. * interrupt handler
  586. */
  587. spin_lock(&workers->order_lock);
  588. if (test_bit(WORK_HIGH_PRIO_BIT, &work->flags)) {
  589. list_add_tail(&work->order_list,
  590. &workers->prio_order_list);
  591. } else {
  592. list_add_tail(&work->order_list, &workers->order_list);
  593. }
  594. spin_unlock(&workers->order_lock);
  595. } else {
  596. INIT_LIST_HEAD(&work->order_list);
  597. }
  598. spin_lock_irqsave(&worker->lock, flags);
  599. if (test_bit(WORK_HIGH_PRIO_BIT, &work->flags))
  600. list_add_tail(&work->list, &worker->prio_pending);
  601. else
  602. list_add_tail(&work->list, &worker->pending);
  603. check_busy_worker(worker);
  604. /*
  605. * avoid calling into wake_up_process if this thread has already
  606. * been kicked
  607. */
  608. if (!worker->working)
  609. wake = 1;
  610. worker->working = 1;
  611. if (wake)
  612. wake_up_process(worker->task);
  613. spin_unlock_irqrestore(&worker->lock, flags);
  614. }