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