2 * Copyright (C) 2007 Oracle. All rights reserved.
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.
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.
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.
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"
25 #define WORK_QUEUED_BIT 0
26 #define WORK_DONE_BIT 1
27 #define WORK_ORDER_DONE_BIT 2
28 #define WORK_HIGH_PRIO_BIT 3
31 * container for the kthread task pointer and the list of pending work
32 * One of these is allocated per thread.
34 struct btrfs_worker_thread {
35 /* pool we belong to */
36 struct btrfs_workers *workers;
38 /* list of struct btrfs_work that are waiting for service */
39 struct list_head pending;
40 struct list_head prio_pending;
42 /* list of worker threads from struct btrfs_workers */
43 struct list_head worker_list;
46 struct task_struct *task;
48 /* number of things on the pending list */
51 /* reference counter for this struct */
54 unsigned long sequence;
56 /* protects the pending list. */
59 /* set to non-zero when this thread is already awake and kicking */
62 /* are we currently idle */
67 * helper function to move a thread onto the idle list after it
68 * has finished some requests.
70 static void check_idle_worker(struct btrfs_worker_thread *worker)
72 if (!worker->idle && atomic_read(&worker->num_pending) <
73 worker->workers->idle_thresh / 2) {
75 spin_lock_irqsave(&worker->workers->lock, flags);
78 /* the list may be empty if the worker is just starting */
79 if (!list_empty(&worker->worker_list)) {
80 list_move(&worker->worker_list,
81 &worker->workers->idle_list);
83 spin_unlock_irqrestore(&worker->workers->lock, flags);
88 * helper function to move a thread off the idle list after new
89 * pending work is added.
91 static void check_busy_worker(struct btrfs_worker_thread *worker)
93 if (worker->idle && atomic_read(&worker->num_pending) >=
94 worker->workers->idle_thresh) {
96 spin_lock_irqsave(&worker->workers->lock, flags);
99 if (!list_empty(&worker->worker_list)) {
100 list_move_tail(&worker->worker_list,
101 &worker->workers->worker_list);
103 spin_unlock_irqrestore(&worker->workers->lock, flags);
107 static void check_pending_worker_creates(struct btrfs_worker_thread *worker)
109 struct btrfs_workers *workers = worker->workers;
113 if (!workers->atomic_start_pending)
116 spin_lock_irqsave(&workers->lock, flags);
117 if (!workers->atomic_start_pending)
120 workers->atomic_start_pending = 0;
121 if (workers->num_workers >= workers->max_workers)
124 spin_unlock_irqrestore(&workers->lock, flags);
125 btrfs_start_workers(workers, 1);
129 spin_unlock_irqrestore(&workers->lock, flags);
132 static noinline int run_ordered_completions(struct btrfs_workers *workers,
133 struct btrfs_work *work)
135 if (!workers->ordered)
138 set_bit(WORK_DONE_BIT, &work->flags);
140 spin_lock(&workers->order_lock);
143 if (!list_empty(&workers->prio_order_list)) {
144 work = list_entry(workers->prio_order_list.next,
145 struct btrfs_work, order_list);
146 } else if (!list_empty(&workers->order_list)) {
147 work = list_entry(workers->order_list.next,
148 struct btrfs_work, order_list);
152 if (!test_bit(WORK_DONE_BIT, &work->flags))
155 /* we are going to call the ordered done function, but
156 * we leave the work item on the list as a barrier so
157 * that later work items that are done don't have their
158 * functions called before this one returns
160 if (test_and_set_bit(WORK_ORDER_DONE_BIT, &work->flags))
163 spin_unlock(&workers->order_lock);
165 work->ordered_func(work);
167 /* now take the lock again and call the freeing code */
168 spin_lock(&workers->order_lock);
169 list_del(&work->order_list);
170 work->ordered_free(work);
173 spin_unlock(&workers->order_lock);
177 static void put_worker(struct btrfs_worker_thread *worker)
179 if (atomic_dec_and_test(&worker->refs))
183 static int try_worker_shutdown(struct btrfs_worker_thread *worker)
187 spin_lock_irq(&worker->lock);
188 spin_lock(&worker->workers->lock);
189 if (worker->workers->num_workers > 1 &&
192 !list_empty(&worker->worker_list) &&
193 list_empty(&worker->prio_pending) &&
194 list_empty(&worker->pending) &&
195 atomic_read(&worker->num_pending) == 0) {
197 list_del_init(&worker->worker_list);
198 worker->workers->num_workers--;
200 spin_unlock(&worker->workers->lock);
201 spin_unlock_irq(&worker->lock);
208 static struct btrfs_work *get_next_work(struct btrfs_worker_thread *worker,
209 struct list_head *prio_head,
210 struct list_head *head)
212 struct btrfs_work *work = NULL;
213 struct list_head *cur = NULL;
215 if(!list_empty(prio_head))
216 cur = prio_head->next;
219 if (!list_empty(&worker->prio_pending))
222 if (!list_empty(head))
229 spin_lock_irq(&worker->lock);
230 list_splice_tail_init(&worker->prio_pending, prio_head);
231 list_splice_tail_init(&worker->pending, head);
233 if (!list_empty(prio_head))
234 cur = prio_head->next;
235 else if (!list_empty(head))
237 spin_unlock_irq(&worker->lock);
243 work = list_entry(cur, struct btrfs_work, list);
250 * main loop for servicing work items
252 static int worker_loop(void *arg)
254 struct btrfs_worker_thread *worker = arg;
255 struct list_head head;
256 struct list_head prio_head;
257 struct btrfs_work *work;
259 INIT_LIST_HEAD(&head);
260 INIT_LIST_HEAD(&prio_head);
267 work = get_next_work(worker, &prio_head, &head);
271 list_del(&work->list);
272 clear_bit(WORK_QUEUED_BIT, &work->flags);
274 work->worker = worker;
278 atomic_dec(&worker->num_pending);
280 * unless this is an ordered work queue,
281 * 'work' was probably freed by func above.
283 run_ordered_completions(worker->workers, work);
285 check_pending_worker_creates(worker);
289 spin_lock_irq(&worker->lock);
290 check_idle_worker(worker);
292 if (freezing(current)) {
294 spin_unlock_irq(&worker->lock);
297 spin_unlock_irq(&worker->lock);
298 if (!kthread_should_stop()) {
301 * we've dropped the lock, did someone else
305 if (!list_empty(&worker->pending) ||
306 !list_empty(&worker->prio_pending))
310 * this short schedule allows more work to
311 * come in without the queue functions
312 * needing to go through wake_up_process()
314 * worker->working is still 1, so nobody
315 * is going to try and wake us up
319 if (!list_empty(&worker->pending) ||
320 !list_empty(&worker->prio_pending))
323 if (kthread_should_stop())
326 /* still no more work?, sleep for real */
327 spin_lock_irq(&worker->lock);
328 set_current_state(TASK_INTERRUPTIBLE);
329 if (!list_empty(&worker->pending) ||
330 !list_empty(&worker->prio_pending)) {
331 spin_unlock_irq(&worker->lock);
336 * this makes sure we get a wakeup when someone
337 * adds something new to the queue
340 spin_unlock_irq(&worker->lock);
342 if (!kthread_should_stop()) {
343 schedule_timeout(HZ * 120);
344 if (!worker->working &&
345 try_worker_shutdown(worker)) {
350 __set_current_state(TASK_RUNNING);
352 } while (!kthread_should_stop());
357 * this will wait for all the worker threads to shutdown
359 int btrfs_stop_workers(struct btrfs_workers *workers)
361 struct list_head *cur;
362 struct btrfs_worker_thread *worker;
365 spin_lock_irq(&workers->lock);
366 list_splice_init(&workers->idle_list, &workers->worker_list);
367 while (!list_empty(&workers->worker_list)) {
368 cur = workers->worker_list.next;
369 worker = list_entry(cur, struct btrfs_worker_thread,
372 atomic_inc(&worker->refs);
373 workers->num_workers -= 1;
374 if (!list_empty(&worker->worker_list)) {
375 list_del_init(&worker->worker_list);
380 spin_unlock_irq(&workers->lock);
382 kthread_stop(worker->task);
383 spin_lock_irq(&workers->lock);
386 spin_unlock_irq(&workers->lock);
391 * simple init on struct btrfs_workers
393 void btrfs_init_workers(struct btrfs_workers *workers, char *name, int max)
395 workers->num_workers = 0;
396 INIT_LIST_HEAD(&workers->worker_list);
397 INIT_LIST_HEAD(&workers->idle_list);
398 INIT_LIST_HEAD(&workers->order_list);
399 INIT_LIST_HEAD(&workers->prio_order_list);
400 spin_lock_init(&workers->lock);
401 spin_lock_init(&workers->order_lock);
402 workers->max_workers = max;
403 workers->idle_thresh = 32;
404 workers->name = name;
405 workers->ordered = 0;
406 workers->atomic_start_pending = 0;
407 workers->atomic_worker_start = 0;
411 * starts new worker threads. This does not enforce the max worker
412 * count in case you need to temporarily go past it.
414 int btrfs_start_workers(struct btrfs_workers *workers, int num_workers)
416 struct btrfs_worker_thread *worker;
420 for (i = 0; i < num_workers; i++) {
421 worker = kzalloc(sizeof(*worker), GFP_NOFS);
427 INIT_LIST_HEAD(&worker->pending);
428 INIT_LIST_HEAD(&worker->prio_pending);
429 INIT_LIST_HEAD(&worker->worker_list);
430 spin_lock_init(&worker->lock);
432 atomic_set(&worker->num_pending, 0);
433 atomic_set(&worker->refs, 1);
434 worker->workers = workers;
435 worker->task = kthread_run(worker_loop, worker,
436 "btrfs-%s-%d", workers->name,
437 workers->num_workers + i);
438 if (IS_ERR(worker->task)) {
439 ret = PTR_ERR(worker->task);
443 spin_lock_irq(&workers->lock);
444 list_add_tail(&worker->worker_list, &workers->idle_list);
446 workers->num_workers++;
447 spin_unlock_irq(&workers->lock);
451 btrfs_stop_workers(workers);
456 * run through the list and find a worker thread that doesn't have a lot
457 * to do right now. This can return null if we aren't yet at the thread
458 * count limit and all of the threads are busy.
460 static struct btrfs_worker_thread *next_worker(struct btrfs_workers *workers)
462 struct btrfs_worker_thread *worker;
463 struct list_head *next;
464 int enforce_min = workers->num_workers < workers->max_workers;
467 * if we find an idle thread, don't move it to the end of the
468 * idle list. This improves the chance that the next submission
469 * will reuse the same thread, and maybe catch it while it is still
472 if (!list_empty(&workers->idle_list)) {
473 next = workers->idle_list.next;
474 worker = list_entry(next, struct btrfs_worker_thread,
478 if (enforce_min || list_empty(&workers->worker_list))
482 * if we pick a busy task, move the task to the end of the list.
483 * hopefully this will keep things somewhat evenly balanced.
484 * Do the move in batches based on the sequence number. This groups
485 * requests submitted at roughly the same time onto the same worker.
487 next = workers->worker_list.next;
488 worker = list_entry(next, struct btrfs_worker_thread, worker_list);
491 if (worker->sequence % workers->idle_thresh == 0)
492 list_move_tail(next, &workers->worker_list);
497 * selects a worker thread to take the next job. This will either find
498 * an idle worker, start a new worker up to the max count, or just return
499 * one of the existing busy workers.
501 static struct btrfs_worker_thread *find_worker(struct btrfs_workers *workers)
503 struct btrfs_worker_thread *worker;
505 struct list_head *fallback;
508 spin_lock_irqsave(&workers->lock, flags);
509 worker = next_worker(workers);
512 if (workers->num_workers >= workers->max_workers) {
514 } else if (workers->atomic_worker_start) {
515 workers->atomic_start_pending = 1;
518 spin_unlock_irqrestore(&workers->lock, flags);
519 /* we're below the limit, start another worker */
520 btrfs_start_workers(workers, 1);
529 * we have failed to find any workers, just
530 * return the first one we can find.
532 if (!list_empty(&workers->worker_list))
533 fallback = workers->worker_list.next;
534 if (!list_empty(&workers->idle_list))
535 fallback = workers->idle_list.next;
537 worker = list_entry(fallback,
538 struct btrfs_worker_thread, worker_list);
541 * this makes sure the worker doesn't exit before it is placed
542 * onto a busy/idle list
544 atomic_inc(&worker->num_pending);
545 spin_unlock_irqrestore(&workers->lock, flags);
550 * btrfs_requeue_work just puts the work item back on the tail of the list
551 * it was taken from. It is intended for use with long running work functions
552 * that make some progress and want to give the cpu up for others.
554 int btrfs_requeue_work(struct btrfs_work *work)
556 struct btrfs_worker_thread *worker = work->worker;
560 if (test_and_set_bit(WORK_QUEUED_BIT, &work->flags))
563 spin_lock_irqsave(&worker->lock, flags);
564 if (test_bit(WORK_HIGH_PRIO_BIT, &work->flags))
565 list_add_tail(&work->list, &worker->prio_pending);
567 list_add_tail(&work->list, &worker->pending);
568 atomic_inc(&worker->num_pending);
570 /* by definition we're busy, take ourselves off the idle
574 spin_lock(&worker->workers->lock);
576 list_move_tail(&worker->worker_list,
577 &worker->workers->worker_list);
578 spin_unlock(&worker->workers->lock);
580 if (!worker->working) {
586 wake_up_process(worker->task);
587 spin_unlock_irqrestore(&worker->lock, flags);
593 void btrfs_set_work_high_prio(struct btrfs_work *work)
595 set_bit(WORK_HIGH_PRIO_BIT, &work->flags);
599 * places a struct btrfs_work into the pending queue of one of the kthreads
601 int btrfs_queue_worker(struct btrfs_workers *workers, struct btrfs_work *work)
603 struct btrfs_worker_thread *worker;
607 /* don't requeue something already on a list */
608 if (test_and_set_bit(WORK_QUEUED_BIT, &work->flags))
611 worker = find_worker(workers);
612 if (workers->ordered) {
614 * you're not allowed to do ordered queues from an
617 spin_lock(&workers->order_lock);
618 if (test_bit(WORK_HIGH_PRIO_BIT, &work->flags)) {
619 list_add_tail(&work->order_list,
620 &workers->prio_order_list);
622 list_add_tail(&work->order_list, &workers->order_list);
624 spin_unlock(&workers->order_lock);
626 INIT_LIST_HEAD(&work->order_list);
629 spin_lock_irqsave(&worker->lock, flags);
631 if (test_bit(WORK_HIGH_PRIO_BIT, &work->flags))
632 list_add_tail(&work->list, &worker->prio_pending);
634 list_add_tail(&work->list, &worker->pending);
635 check_busy_worker(worker);
638 * avoid calling into wake_up_process if this thread has already
641 if (!worker->working)
646 wake_up_process(worker->task);
647 spin_unlock_irqrestore(&worker->lock, flags);