blob: dff07795bd16af0dbbb5ff1cb51e4729a28ef3ab [file] [log] [blame]
Linus Torvalds1da177e2005-04-16 15:20:36 -07001/*
2 * linux/net/sunrpc/sched.c
3 *
4 * Scheduling for synchronous and asynchronous RPC requests.
5 *
6 * Copyright (C) 1996 Olaf Kirch, <okir@monad.swb.de>
7 *
8 * TCP NFS related read + write fixes
9 * (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
10 */
11
12#include <linux/module.h>
13
14#include <linux/sched.h>
15#include <linux/interrupt.h>
16#include <linux/slab.h>
17#include <linux/mempool.h>
18#include <linux/smp.h>
19#include <linux/smp_lock.h>
20#include <linux/spinlock.h>
Arjan van de Ven4a3e2f72006-03-20 22:33:17 -080021#include <linux/mutex.h>
Linus Torvalds1da177e2005-04-16 15:20:36 -070022
23#include <linux/sunrpc/clnt.h>
24#include <linux/sunrpc/xprt.h>
25
26#ifdef RPC_DEBUG
27#define RPCDBG_FACILITY RPCDBG_SCHED
28#define RPC_TASK_MAGIC_ID 0xf00baa
29static int rpc_task_id;
30#endif
31
32/*
33 * RPC slabs and memory pools
34 */
35#define RPC_BUFFER_MAXSIZE (2048)
36#define RPC_BUFFER_POOLSIZE (8)
37#define RPC_TASK_POOLSIZE (8)
Eric Dumazetba899662005-08-26 12:05:31 -070038static kmem_cache_t *rpc_task_slabp __read_mostly;
39static kmem_cache_t *rpc_buffer_slabp __read_mostly;
40static mempool_t *rpc_task_mempool __read_mostly;
41static mempool_t *rpc_buffer_mempool __read_mostly;
Linus Torvalds1da177e2005-04-16 15:20:36 -070042
43static void __rpc_default_timer(struct rpc_task *task);
44static void rpciod_killall(void);
Linus Torvalds1da177e2005-04-16 15:20:36 -070045static void rpc_async_schedule(void *);
46
47/*
48 * RPC tasks that create another task (e.g. for contacting the portmapper)
49 * will wait on this queue for their child's completion
50 */
51static RPC_WAITQ(childq, "childq");
52
53/*
54 * RPC tasks sit here while waiting for conditions to improve.
55 */
56static RPC_WAITQ(delay_queue, "delayq");
57
58/*
59 * All RPC tasks are linked into this list
60 */
61static LIST_HEAD(all_tasks);
62
63/*
64 * rpciod-related stuff
65 */
Arjan van de Ven4a3e2f72006-03-20 22:33:17 -080066static DEFINE_MUTEX(rpciod_mutex);
Linus Torvalds1da177e2005-04-16 15:20:36 -070067static unsigned int rpciod_users;
68static struct workqueue_struct *rpciod_workqueue;
69
70/*
71 * Spinlock for other critical sections of code.
72 */
73static DEFINE_SPINLOCK(rpc_sched_lock);
74
75/*
76 * Disable the timer for a given RPC task. Should be called with
77 * queue->lock and bh_disabled in order to avoid races within
78 * rpc_run_timer().
79 */
80static inline void
81__rpc_disable_timer(struct rpc_task *task)
82{
83 dprintk("RPC: %4d disabling timer\n", task->tk_pid);
84 task->tk_timeout_fn = NULL;
85 task->tk_timeout = 0;
86}
87
88/*
89 * Run a timeout function.
90 * We use the callback in order to allow __rpc_wake_up_task()
91 * and friends to disable the timer synchronously on SMP systems
92 * without calling del_timer_sync(). The latter could cause a
93 * deadlock if called while we're holding spinlocks...
94 */
95static void rpc_run_timer(struct rpc_task *task)
96{
97 void (*callback)(struct rpc_task *);
98
99 callback = task->tk_timeout_fn;
100 task->tk_timeout_fn = NULL;
101 if (callback && RPC_IS_QUEUED(task)) {
102 dprintk("RPC: %4d running timer\n", task->tk_pid);
103 callback(task);
104 }
105 smp_mb__before_clear_bit();
106 clear_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate);
107 smp_mb__after_clear_bit();
108}
109
110/*
111 * Set up a timer for the current task.
112 */
113static inline void
114__rpc_add_timer(struct rpc_task *task, rpc_action timer)
115{
116 if (!task->tk_timeout)
117 return;
118
119 dprintk("RPC: %4d setting alarm for %lu ms\n",
120 task->tk_pid, task->tk_timeout * 1000 / HZ);
121
122 if (timer)
123 task->tk_timeout_fn = timer;
124 else
125 task->tk_timeout_fn = __rpc_default_timer;
126 set_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate);
127 mod_timer(&task->tk_timer, jiffies + task->tk_timeout);
128}
129
130/*
131 * Delete any timer for the current task. Because we use del_timer_sync(),
132 * this function should never be called while holding queue->lock.
133 */
134static void
135rpc_delete_timer(struct rpc_task *task)
136{
137 if (RPC_IS_QUEUED(task))
138 return;
139 if (test_and_clear_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate)) {
140 del_singleshot_timer_sync(&task->tk_timer);
141 dprintk("RPC: %4d deleting timer\n", task->tk_pid);
142 }
143}
144
145/*
146 * Add new request to a priority queue.
147 */
148static void __rpc_add_wait_queue_priority(struct rpc_wait_queue *queue, struct rpc_task *task)
149{
150 struct list_head *q;
151 struct rpc_task *t;
152
153 INIT_LIST_HEAD(&task->u.tk_wait.links);
154 q = &queue->tasks[task->tk_priority];
155 if (unlikely(task->tk_priority > queue->maxpriority))
156 q = &queue->tasks[queue->maxpriority];
157 list_for_each_entry(t, q, u.tk_wait.list) {
158 if (t->tk_cookie == task->tk_cookie) {
159 list_add_tail(&task->u.tk_wait.list, &t->u.tk_wait.links);
160 return;
161 }
162 }
163 list_add_tail(&task->u.tk_wait.list, q);
164}
165
166/*
167 * Add new request to wait queue.
168 *
169 * Swapper tasks always get inserted at the head of the queue.
170 * This should avoid many nasty memory deadlocks and hopefully
171 * improve overall performance.
172 * Everyone else gets appended to the queue to ensure proper FIFO behavior.
173 */
174static void __rpc_add_wait_queue(struct rpc_wait_queue *queue, struct rpc_task *task)
175{
176 BUG_ON (RPC_IS_QUEUED(task));
177
178 if (RPC_IS_PRIORITY(queue))
179 __rpc_add_wait_queue_priority(queue, task);
180 else if (RPC_IS_SWAPPER(task))
181 list_add(&task->u.tk_wait.list, &queue->tasks[0]);
182 else
183 list_add_tail(&task->u.tk_wait.list, &queue->tasks[0]);
184 task->u.tk_wait.rpc_waitq = queue;
185 rpc_set_queued(task);
186
187 dprintk("RPC: %4d added to queue %p \"%s\"\n",
188 task->tk_pid, queue, rpc_qname(queue));
189}
190
191/*
192 * Remove request from a priority queue.
193 */
194static void __rpc_remove_wait_queue_priority(struct rpc_task *task)
195{
196 struct rpc_task *t;
197
198 if (!list_empty(&task->u.tk_wait.links)) {
199 t = list_entry(task->u.tk_wait.links.next, struct rpc_task, u.tk_wait.list);
200 list_move(&t->u.tk_wait.list, &task->u.tk_wait.list);
201 list_splice_init(&task->u.tk_wait.links, &t->u.tk_wait.links);
202 }
203 list_del(&task->u.tk_wait.list);
204}
205
206/*
207 * Remove request from queue.
208 * Note: must be called with spin lock held.
209 */
210static void __rpc_remove_wait_queue(struct rpc_task *task)
211{
212 struct rpc_wait_queue *queue;
213 queue = task->u.tk_wait.rpc_waitq;
214
215 if (RPC_IS_PRIORITY(queue))
216 __rpc_remove_wait_queue_priority(task);
217 else
218 list_del(&task->u.tk_wait.list);
219 dprintk("RPC: %4d removed from queue %p \"%s\"\n",
220 task->tk_pid, queue, rpc_qname(queue));
221}
222
223static inline void rpc_set_waitqueue_priority(struct rpc_wait_queue *queue, int priority)
224{
225 queue->priority = priority;
226 queue->count = 1 << (priority * 2);
227}
228
229static inline void rpc_set_waitqueue_cookie(struct rpc_wait_queue *queue, unsigned long cookie)
230{
231 queue->cookie = cookie;
232 queue->nr = RPC_BATCH_COUNT;
233}
234
235static inline void rpc_reset_waitqueue_priority(struct rpc_wait_queue *queue)
236{
237 rpc_set_waitqueue_priority(queue, queue->maxpriority);
238 rpc_set_waitqueue_cookie(queue, 0);
239}
240
241static void __rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname, int maxprio)
242{
243 int i;
244
245 spin_lock_init(&queue->lock);
246 for (i = 0; i < ARRAY_SIZE(queue->tasks); i++)
247 INIT_LIST_HEAD(&queue->tasks[i]);
248 queue->maxpriority = maxprio;
249 rpc_reset_waitqueue_priority(queue);
250#ifdef RPC_DEBUG
251 queue->name = qname;
252#endif
253}
254
255void rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname)
256{
257 __rpc_init_priority_wait_queue(queue, qname, RPC_PRIORITY_HIGH);
258}
259
260void rpc_init_wait_queue(struct rpc_wait_queue *queue, const char *qname)
261{
262 __rpc_init_priority_wait_queue(queue, qname, 0);
263}
264EXPORT_SYMBOL(rpc_init_wait_queue);
265
Trond Myklebust44c28872006-01-03 09:55:06 +0100266static int rpc_wait_bit_interruptible(void *word)
267{
268 if (signal_pending(current))
269 return -ERESTARTSYS;
270 schedule();
271 return 0;
272}
273
274/*
275 * Mark an RPC call as having completed by clearing the 'active' bit
276 */
277static inline void rpc_mark_complete_task(struct rpc_task *task)
278{
279 rpc_clear_active(task);
280 wake_up_bit(&task->tk_runstate, RPC_TASK_ACTIVE);
281}
282
283/*
284 * Allow callers to wait for completion of an RPC call
285 */
286int __rpc_wait_for_completion_task(struct rpc_task *task, int (*action)(void *))
287{
288 if (action == NULL)
289 action = rpc_wait_bit_interruptible;
290 return wait_on_bit(&task->tk_runstate, RPC_TASK_ACTIVE,
291 action, TASK_INTERRUPTIBLE);
292}
293EXPORT_SYMBOL(__rpc_wait_for_completion_task);
294
Linus Torvalds1da177e2005-04-16 15:20:36 -0700295/*
296 * Make an RPC task runnable.
297 *
298 * Note: If the task is ASYNC, this must be called with
299 * the spinlock held to protect the wait queue operation.
300 */
301static void rpc_make_runnable(struct rpc_task *task)
302{
303 int do_ret;
304
305 BUG_ON(task->tk_timeout_fn);
306 do_ret = rpc_test_and_set_running(task);
307 rpc_clear_queued(task);
308 if (do_ret)
309 return;
310 if (RPC_IS_ASYNC(task)) {
311 int status;
312
313 INIT_WORK(&task->u.tk_work, rpc_async_schedule, (void *)task);
314 status = queue_work(task->tk_workqueue, &task->u.tk_work);
315 if (status < 0) {
316 printk(KERN_WARNING "RPC: failed to add task to queue: error: %d!\n", status);
317 task->tk_status = status;
318 return;
319 }
320 } else
Trond Myklebust96651ab2005-06-22 17:16:21 +0000321 wake_up_bit(&task->tk_runstate, RPC_TASK_QUEUED);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700322}
323
324/*
325 * Place a newly initialized task on the workqueue.
326 */
327static inline void
328rpc_schedule_run(struct rpc_task *task)
329{
Trond Myklebust44c28872006-01-03 09:55:06 +0100330 rpc_set_active(task);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700331 rpc_make_runnable(task);
332}
333
334/*
335 * Prepare for sleeping on a wait queue.
336 * By always appending tasks to the list we ensure FIFO behavior.
337 * NB: An RPC task will only receive interrupt-driven events as long
338 * as it's on a wait queue.
339 */
340static void __rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
341 rpc_action action, rpc_action timer)
342{
343 dprintk("RPC: %4d sleep_on(queue \"%s\" time %ld)\n", task->tk_pid,
344 rpc_qname(q), jiffies);
345
346 if (!RPC_IS_ASYNC(task) && !RPC_IS_ACTIVATED(task)) {
347 printk(KERN_ERR "RPC: Inactive synchronous task put to sleep!\n");
348 return;
349 }
350
351 /* Mark the task as being activated if so needed */
Trond Myklebust44c28872006-01-03 09:55:06 +0100352 rpc_set_active(task);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700353
354 __rpc_add_wait_queue(q, task);
355
356 BUG_ON(task->tk_callback != NULL);
357 task->tk_callback = action;
358 __rpc_add_timer(task, timer);
359}
360
361void rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
362 rpc_action action, rpc_action timer)
363{
364 /*
365 * Protect the queue operations.
366 */
367 spin_lock_bh(&q->lock);
368 __rpc_sleep_on(q, task, action, timer);
369 spin_unlock_bh(&q->lock);
370}
371
372/**
373 * __rpc_do_wake_up_task - wake up a single rpc_task
374 * @task: task to be woken up
375 *
376 * Caller must hold queue->lock, and have cleared the task queued flag.
377 */
378static void __rpc_do_wake_up_task(struct rpc_task *task)
379{
380 dprintk("RPC: %4d __rpc_wake_up_task (now %ld)\n", task->tk_pid, jiffies);
381
382#ifdef RPC_DEBUG
383 BUG_ON(task->tk_magic != RPC_TASK_MAGIC_ID);
384#endif
385 /* Has the task been executed yet? If not, we cannot wake it up! */
386 if (!RPC_IS_ACTIVATED(task)) {
387 printk(KERN_ERR "RPC: Inactive task (%p) being woken up!\n", task);
388 return;
389 }
390
391 __rpc_disable_timer(task);
392 __rpc_remove_wait_queue(task);
393
394 rpc_make_runnable(task);
395
396 dprintk("RPC: __rpc_wake_up_task done\n");
397}
398
399/*
400 * Wake up the specified task
401 */
402static void __rpc_wake_up_task(struct rpc_task *task)
403{
404 if (rpc_start_wakeup(task)) {
405 if (RPC_IS_QUEUED(task))
406 __rpc_do_wake_up_task(task);
407 rpc_finish_wakeup(task);
408 }
409}
410
411/*
412 * Default timeout handler if none specified by user
413 */
414static void
415__rpc_default_timer(struct rpc_task *task)
416{
417 dprintk("RPC: %d timeout (default timer)\n", task->tk_pid);
418 task->tk_status = -ETIMEDOUT;
419 rpc_wake_up_task(task);
420}
421
422/*
423 * Wake up the specified task
424 */
425void rpc_wake_up_task(struct rpc_task *task)
426{
427 if (rpc_start_wakeup(task)) {
428 if (RPC_IS_QUEUED(task)) {
429 struct rpc_wait_queue *queue = task->u.tk_wait.rpc_waitq;
430
431 spin_lock_bh(&queue->lock);
432 __rpc_do_wake_up_task(task);
433 spin_unlock_bh(&queue->lock);
434 }
435 rpc_finish_wakeup(task);
436 }
437}
438
439/*
440 * Wake up the next task on a priority queue.
441 */
442static struct rpc_task * __rpc_wake_up_next_priority(struct rpc_wait_queue *queue)
443{
444 struct list_head *q;
445 struct rpc_task *task;
446
447 /*
448 * Service a batch of tasks from a single cookie.
449 */
450 q = &queue->tasks[queue->priority];
451 if (!list_empty(q)) {
452 task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
453 if (queue->cookie == task->tk_cookie) {
454 if (--queue->nr)
455 goto out;
456 list_move_tail(&task->u.tk_wait.list, q);
457 }
458 /*
459 * Check if we need to switch queues.
460 */
461 if (--queue->count)
462 goto new_cookie;
463 }
464
465 /*
466 * Service the next queue.
467 */
468 do {
469 if (q == &queue->tasks[0])
470 q = &queue->tasks[queue->maxpriority];
471 else
472 q = q - 1;
473 if (!list_empty(q)) {
474 task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
475 goto new_queue;
476 }
477 } while (q != &queue->tasks[queue->priority]);
478
479 rpc_reset_waitqueue_priority(queue);
480 return NULL;
481
482new_queue:
483 rpc_set_waitqueue_priority(queue, (unsigned int)(q - &queue->tasks[0]));
484new_cookie:
485 rpc_set_waitqueue_cookie(queue, task->tk_cookie);
486out:
487 __rpc_wake_up_task(task);
488 return task;
489}
490
491/*
492 * Wake up the next task on the wait queue.
493 */
494struct rpc_task * rpc_wake_up_next(struct rpc_wait_queue *queue)
495{
496 struct rpc_task *task = NULL;
497
498 dprintk("RPC: wake_up_next(%p \"%s\")\n", queue, rpc_qname(queue));
499 spin_lock_bh(&queue->lock);
500 if (RPC_IS_PRIORITY(queue))
501 task = __rpc_wake_up_next_priority(queue);
502 else {
503 task_for_first(task, &queue->tasks[0])
504 __rpc_wake_up_task(task);
505 }
506 spin_unlock_bh(&queue->lock);
507
508 return task;
509}
510
511/**
512 * rpc_wake_up - wake up all rpc_tasks
513 * @queue: rpc_wait_queue on which the tasks are sleeping
514 *
515 * Grabs queue->lock
516 */
517void rpc_wake_up(struct rpc_wait_queue *queue)
518{
Trond Myklebuste6d83d52006-03-13 21:20:48 -0800519 struct rpc_task *task, *next;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700520 struct list_head *head;
Trond Myklebuste6d83d52006-03-13 21:20:48 -0800521
Linus Torvalds1da177e2005-04-16 15:20:36 -0700522 spin_lock_bh(&queue->lock);
523 head = &queue->tasks[queue->maxpriority];
524 for (;;) {
Trond Myklebuste6d83d52006-03-13 21:20:48 -0800525 list_for_each_entry_safe(task, next, head, u.tk_wait.list)
Linus Torvalds1da177e2005-04-16 15:20:36 -0700526 __rpc_wake_up_task(task);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700527 if (head == &queue->tasks[0])
528 break;
529 head--;
530 }
531 spin_unlock_bh(&queue->lock);
532}
533
534/**
535 * rpc_wake_up_status - wake up all rpc_tasks and set their status value.
536 * @queue: rpc_wait_queue on which the tasks are sleeping
537 * @status: status value to set
538 *
539 * Grabs queue->lock
540 */
541void rpc_wake_up_status(struct rpc_wait_queue *queue, int status)
542{
Trond Myklebuste6d83d52006-03-13 21:20:48 -0800543 struct rpc_task *task, *next;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700544 struct list_head *head;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700545
546 spin_lock_bh(&queue->lock);
547 head = &queue->tasks[queue->maxpriority];
548 for (;;) {
Trond Myklebuste6d83d52006-03-13 21:20:48 -0800549 list_for_each_entry_safe(task, next, head, u.tk_wait.list) {
Linus Torvalds1da177e2005-04-16 15:20:36 -0700550 task->tk_status = status;
551 __rpc_wake_up_task(task);
552 }
553 if (head == &queue->tasks[0])
554 break;
555 head--;
556 }
557 spin_unlock_bh(&queue->lock);
558}
559
560/*
561 * Run a task at a later time
562 */
563static void __rpc_atrun(struct rpc_task *);
564void
565rpc_delay(struct rpc_task *task, unsigned long delay)
566{
567 task->tk_timeout = delay;
568 rpc_sleep_on(&delay_queue, task, NULL, __rpc_atrun);
569}
570
571static void
572__rpc_atrun(struct rpc_task *task)
573{
574 task->tk_status = 0;
575 rpc_wake_up_task(task);
576}
577
578/*
Trond Myklebust4ce70ad2006-01-03 09:55:05 +0100579 * Helper to call task->tk_ops->rpc_call_prepare
580 */
581static void rpc_prepare_task(struct rpc_task *task)
582{
583 task->tk_ops->rpc_call_prepare(task, task->tk_calldata);
584}
585
586/*
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100587 * Helper that calls task->tk_ops->rpc_call_done if it exists
Trond Myklebustd05fdb02005-06-22 17:16:19 +0000588 */
Trond Myklebustabbcf282006-01-03 09:55:03 +0100589void rpc_exit_task(struct rpc_task *task)
Trond Myklebustd05fdb02005-06-22 17:16:19 +0000590{
Trond Myklebustabbcf282006-01-03 09:55:03 +0100591 task->tk_action = NULL;
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100592 if (task->tk_ops->rpc_call_done != NULL) {
593 task->tk_ops->rpc_call_done(task, task->tk_calldata);
Trond Myklebustd05fdb02005-06-22 17:16:19 +0000594 if (task->tk_action != NULL) {
Trond Myklebustabbcf282006-01-03 09:55:03 +0100595 WARN_ON(RPC_ASSASSINATED(task));
596 /* Always release the RPC slot and buffer memory */
597 xprt_release(task);
Trond Myklebustd05fdb02005-06-22 17:16:19 +0000598 }
599 }
Trond Myklebustd05fdb02005-06-22 17:16:19 +0000600}
Trond Myklebustabbcf282006-01-03 09:55:03 +0100601EXPORT_SYMBOL(rpc_exit_task);
Trond Myklebustd05fdb02005-06-22 17:16:19 +0000602
603/*
Linus Torvalds1da177e2005-04-16 15:20:36 -0700604 * This is the RPC `scheduler' (or rather, the finite state machine).
605 */
606static int __rpc_execute(struct rpc_task *task)
607{
608 int status = 0;
609
610 dprintk("RPC: %4d rpc_execute flgs %x\n",
611 task->tk_pid, task->tk_flags);
612
613 BUG_ON(RPC_IS_QUEUED(task));
614
Trond Myklebustd05fdb02005-06-22 17:16:19 +0000615 for (;;) {
Linus Torvalds1da177e2005-04-16 15:20:36 -0700616 /*
617 * Garbage collection of pending timers...
618 */
619 rpc_delete_timer(task);
620
621 /*
622 * Execute any pending callback.
623 */
624 if (RPC_DO_CALLBACK(task)) {
625 /* Define a callback save pointer */
626 void (*save_callback)(struct rpc_task *);
627
628 /*
629 * If a callback exists, save it, reset it,
630 * call it.
631 * The save is needed to stop from resetting
632 * another callback set within the callback handler
633 * - Dave
634 */
635 save_callback=task->tk_callback;
636 task->tk_callback=NULL;
637 lock_kernel();
638 save_callback(task);
639 unlock_kernel();
640 }
641
642 /*
643 * Perform the next FSM step.
644 * tk_action may be NULL when the task has been killed
645 * by someone else.
646 */
647 if (!RPC_IS_QUEUED(task)) {
Trond Myklebustabbcf282006-01-03 09:55:03 +0100648 if (task->tk_action == NULL)
Linus Torvalds1da177e2005-04-16 15:20:36 -0700649 break;
Trond Myklebustabbcf282006-01-03 09:55:03 +0100650 lock_kernel();
651 task->tk_action(task);
652 unlock_kernel();
Linus Torvalds1da177e2005-04-16 15:20:36 -0700653 }
654
655 /*
656 * Lockless check for whether task is sleeping or not.
657 */
658 if (!RPC_IS_QUEUED(task))
659 continue;
660 rpc_clear_running(task);
661 if (RPC_IS_ASYNC(task)) {
662 /* Careful! we may have raced... */
663 if (RPC_IS_QUEUED(task))
664 return 0;
665 if (rpc_test_and_set_running(task))
666 return 0;
667 continue;
668 }
669
670 /* sync task: sleep here */
671 dprintk("RPC: %4d sync task going to sleep\n", task->tk_pid);
Trond Myklebust96651ab2005-06-22 17:16:21 +0000672 /* Note: Caller should be using rpc_clnt_sigmask() */
673 status = out_of_line_wait_on_bit(&task->tk_runstate,
674 RPC_TASK_QUEUED, rpc_wait_bit_interruptible,
675 TASK_INTERRUPTIBLE);
676 if (status == -ERESTARTSYS) {
Linus Torvalds1da177e2005-04-16 15:20:36 -0700677 /*
678 * When a sync task receives a signal, it exits with
679 * -ERESTARTSYS. In order to catch any callbacks that
680 * clean up after sleeping on some queue, we don't
681 * break the loop here, but go around once more.
682 */
Trond Myklebust96651ab2005-06-22 17:16:21 +0000683 dprintk("RPC: %4d got signal\n", task->tk_pid);
684 task->tk_flags |= RPC_TASK_KILLED;
685 rpc_exit(task, -ERESTARTSYS);
686 rpc_wake_up_task(task);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700687 }
688 rpc_set_running(task);
689 dprintk("RPC: %4d sync task resuming\n", task->tk_pid);
690 }
691
Trond Myklebuste60859a2006-01-03 09:55:10 +0100692 dprintk("RPC: %4d, return %d, status %d\n", task->tk_pid, status, task->tk_status);
Trond Myklebust44c28872006-01-03 09:55:06 +0100693 /* Wake up anyone who is waiting for task completion */
694 rpc_mark_complete_task(task);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700695 /* Release all resources associated with the task */
696 rpc_release_task(task);
697 return status;
698}
699
700/*
701 * User-visible entry point to the scheduler.
702 *
703 * This may be called recursively if e.g. an async NFS task updates
704 * the attributes and finds that dirty pages must be flushed.
705 * NOTE: Upon exit of this function the task is guaranteed to be
706 * released. In particular note that tk_release() will have
707 * been called, so your task memory may have been freed.
708 */
709int
710rpc_execute(struct rpc_task *task)
711{
Trond Myklebust44c28872006-01-03 09:55:06 +0100712 rpc_set_active(task);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700713 rpc_set_running(task);
714 return __rpc_execute(task);
715}
716
717static void rpc_async_schedule(void *arg)
718{
719 __rpc_execute((struct rpc_task *)arg);
720}
721
Chuck Lever02107142006-01-03 09:55:49 +0100722/**
723 * rpc_malloc - allocate an RPC buffer
724 * @task: RPC task that will use this buffer
725 * @size: requested byte size
Linus Torvalds1da177e2005-04-16 15:20:36 -0700726 *
727 * We try to ensure that some NFS reads and writes can always proceed
728 * by using a mempool when allocating 'small' buffers.
729 * In order to avoid memory starvation triggering more writebacks of
730 * NFS requests, we use GFP_NOFS rather than GFP_KERNEL.
731 */
Chuck Lever02107142006-01-03 09:55:49 +0100732void * rpc_malloc(struct rpc_task *task, size_t size)
Linus Torvalds1da177e2005-04-16 15:20:36 -0700733{
Chuck Lever02107142006-01-03 09:55:49 +0100734 struct rpc_rqst *req = task->tk_rqstp;
Al Virodd0fc662005-10-07 07:46:04 +0100735 gfp_t gfp;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700736
737 if (task->tk_flags & RPC_TASK_SWAPPER)
738 gfp = GFP_ATOMIC;
739 else
740 gfp = GFP_NOFS;
741
742 if (size > RPC_BUFFER_MAXSIZE) {
Chuck Lever02107142006-01-03 09:55:49 +0100743 req->rq_buffer = kmalloc(size, gfp);
744 if (req->rq_buffer)
745 req->rq_bufsize = size;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700746 } else {
Chuck Lever02107142006-01-03 09:55:49 +0100747 req->rq_buffer = mempool_alloc(rpc_buffer_mempool, gfp);
748 if (req->rq_buffer)
749 req->rq_bufsize = RPC_BUFFER_MAXSIZE;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700750 }
Chuck Lever02107142006-01-03 09:55:49 +0100751 return req->rq_buffer;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700752}
753
Chuck Lever02107142006-01-03 09:55:49 +0100754/**
755 * rpc_free - free buffer allocated via rpc_malloc
756 * @task: RPC task with a buffer to be freed
757 *
758 */
759void rpc_free(struct rpc_task *task)
Linus Torvalds1da177e2005-04-16 15:20:36 -0700760{
Chuck Lever02107142006-01-03 09:55:49 +0100761 struct rpc_rqst *req = task->tk_rqstp;
762
763 if (req->rq_buffer) {
764 if (req->rq_bufsize == RPC_BUFFER_MAXSIZE)
765 mempool_free(req->rq_buffer, rpc_buffer_mempool);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700766 else
Chuck Lever02107142006-01-03 09:55:49 +0100767 kfree(req->rq_buffer);
768 req->rq_buffer = NULL;
769 req->rq_bufsize = 0;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700770 }
771}
772
773/*
774 * Creation and deletion of RPC task structures
775 */
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100776void rpc_init_task(struct rpc_task *task, struct rpc_clnt *clnt, int flags, const struct rpc_call_ops *tk_ops, void *calldata)
Linus Torvalds1da177e2005-04-16 15:20:36 -0700777{
778 memset(task, 0, sizeof(*task));
779 init_timer(&task->tk_timer);
780 task->tk_timer.data = (unsigned long) task;
781 task->tk_timer.function = (void (*)(unsigned long)) rpc_run_timer;
Trond Myklebust44c28872006-01-03 09:55:06 +0100782 atomic_set(&task->tk_count, 1);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700783 task->tk_client = clnt;
784 task->tk_flags = flags;
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100785 task->tk_ops = tk_ops;
Trond Myklebust4ce70ad2006-01-03 09:55:05 +0100786 if (tk_ops->rpc_call_prepare != NULL)
787 task->tk_action = rpc_prepare_task;
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100788 task->tk_calldata = calldata;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700789
790 /* Initialize retry counters */
791 task->tk_garb_retry = 2;
792 task->tk_cred_retry = 2;
793
794 task->tk_priority = RPC_PRIORITY_NORMAL;
795 task->tk_cookie = (unsigned long)current;
796
797 /* Initialize workqueue for async tasks */
798 task->tk_workqueue = rpciod_workqueue;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700799
800 if (clnt) {
801 atomic_inc(&clnt->cl_users);
802 if (clnt->cl_softrtry)
803 task->tk_flags |= RPC_TASK_SOFT;
804 if (!clnt->cl_intr)
805 task->tk_flags |= RPC_TASK_NOINTR;
806 }
807
808#ifdef RPC_DEBUG
809 task->tk_magic = RPC_TASK_MAGIC_ID;
810 task->tk_pid = rpc_task_id++;
811#endif
812 /* Add to global list of all tasks */
813 spin_lock(&rpc_sched_lock);
814 list_add_tail(&task->tk_task, &all_tasks);
815 spin_unlock(&rpc_sched_lock);
816
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100817 BUG_ON(task->tk_ops == NULL);
818
Linus Torvalds1da177e2005-04-16 15:20:36 -0700819 dprintk("RPC: %4d new task procpid %d\n", task->tk_pid,
820 current->pid);
821}
822
823static struct rpc_task *
824rpc_alloc_task(void)
825{
826 return (struct rpc_task *)mempool_alloc(rpc_task_mempool, GFP_NOFS);
827}
828
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100829static void rpc_free_task(struct rpc_task *task)
Linus Torvalds1da177e2005-04-16 15:20:36 -0700830{
831 dprintk("RPC: %4d freeing task\n", task->tk_pid);
832 mempool_free(task, rpc_task_mempool);
833}
834
835/*
836 * Create a new task for the specified client. We have to
837 * clean up after an allocation failure, as the client may
838 * have specified "oneshot".
839 */
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100840struct rpc_task *rpc_new_task(struct rpc_clnt *clnt, int flags, const struct rpc_call_ops *tk_ops, void *calldata)
Linus Torvalds1da177e2005-04-16 15:20:36 -0700841{
842 struct rpc_task *task;
843
844 task = rpc_alloc_task();
845 if (!task)
846 goto cleanup;
847
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100848 rpc_init_task(task, clnt, flags, tk_ops, calldata);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700849
850 dprintk("RPC: %4d allocated task\n", task->tk_pid);
851 task->tk_flags |= RPC_TASK_DYNAMIC;
852out:
853 return task;
854
855cleanup:
856 /* Check whether to release the client */
857 if (clnt) {
858 printk("rpc_new_task: failed, users=%d, oneshot=%d\n",
859 atomic_read(&clnt->cl_users), clnt->cl_oneshot);
860 atomic_inc(&clnt->cl_users); /* pretend we were used ... */
861 rpc_release_client(clnt);
862 }
863 goto out;
864}
865
866void rpc_release_task(struct rpc_task *task)
867{
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100868 const struct rpc_call_ops *tk_ops = task->tk_ops;
869 void *calldata = task->tk_calldata;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700870
871#ifdef RPC_DEBUG
872 BUG_ON(task->tk_magic != RPC_TASK_MAGIC_ID);
873#endif
Trond Myklebust44c28872006-01-03 09:55:06 +0100874 if (!atomic_dec_and_test(&task->tk_count))
875 return;
876 dprintk("RPC: %4d release task\n", task->tk_pid);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700877
878 /* Remove from global task list */
879 spin_lock(&rpc_sched_lock);
880 list_del(&task->tk_task);
881 spin_unlock(&rpc_sched_lock);
882
883 BUG_ON (RPC_IS_QUEUED(task));
Linus Torvalds1da177e2005-04-16 15:20:36 -0700884
885 /* Synchronously delete any running timer */
886 rpc_delete_timer(task);
887
888 /* Release resources */
889 if (task->tk_rqstp)
890 xprt_release(task);
891 if (task->tk_msg.rpc_cred)
892 rpcauth_unbindcred(task);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700893 if (task->tk_client) {
894 rpc_release_client(task->tk_client);
895 task->tk_client = NULL;
896 }
897
898#ifdef RPC_DEBUG
899 task->tk_magic = 0;
900#endif
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100901 if (task->tk_flags & RPC_TASK_DYNAMIC)
902 rpc_free_task(task);
903 if (tk_ops->rpc_release)
904 tk_ops->rpc_release(calldata);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700905}
906
907/**
Trond Myklebust44c28872006-01-03 09:55:06 +0100908 * rpc_run_task - Allocate a new RPC task, then run rpc_execute against it
Martin Waitz99acf042006-02-01 03:06:56 -0800909 * @clnt: pointer to RPC client
910 * @flags: RPC flags
911 * @ops: RPC call ops
912 * @data: user call data
Trond Myklebust44c28872006-01-03 09:55:06 +0100913 */
914struct rpc_task *rpc_run_task(struct rpc_clnt *clnt, int flags,
915 const struct rpc_call_ops *ops,
916 void *data)
917{
918 struct rpc_task *task;
919 task = rpc_new_task(clnt, flags, ops, data);
920 if (task == NULL)
921 return ERR_PTR(-ENOMEM);
922 atomic_inc(&task->tk_count);
923 rpc_execute(task);
924 return task;
925}
926EXPORT_SYMBOL(rpc_run_task);
927
928/**
Linus Torvalds1da177e2005-04-16 15:20:36 -0700929 * rpc_find_parent - find the parent of a child task.
930 * @child: child task
Martin Waitz99acf042006-02-01 03:06:56 -0800931 * @parent: parent task
Linus Torvalds1da177e2005-04-16 15:20:36 -0700932 *
933 * Checks that the parent task is still sleeping on the
934 * queue 'childq'. If so returns a pointer to the parent.
935 * Upon failure returns NULL.
936 *
937 * Caller must hold childq.lock
938 */
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100939static inline struct rpc_task *rpc_find_parent(struct rpc_task *child, struct rpc_task *parent)
Linus Torvalds1da177e2005-04-16 15:20:36 -0700940{
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100941 struct rpc_task *task;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700942 struct list_head *le;
943
Linus Torvalds1da177e2005-04-16 15:20:36 -0700944 task_for_each(task, le, &childq.tasks[0])
945 if (task == parent)
946 return parent;
947
948 return NULL;
949}
950
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100951static void rpc_child_exit(struct rpc_task *child, void *calldata)
Linus Torvalds1da177e2005-04-16 15:20:36 -0700952{
953 struct rpc_task *parent;
954
955 spin_lock_bh(&childq.lock);
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100956 if ((parent = rpc_find_parent(child, calldata)) != NULL) {
Linus Torvalds1da177e2005-04-16 15:20:36 -0700957 parent->tk_status = child->tk_status;
958 __rpc_wake_up_task(parent);
959 }
960 spin_unlock_bh(&childq.lock);
961}
962
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100963static const struct rpc_call_ops rpc_child_ops = {
964 .rpc_call_done = rpc_child_exit,
965};
966
Linus Torvalds1da177e2005-04-16 15:20:36 -0700967/*
968 * Note: rpc_new_task releases the client after a failure.
969 */
970struct rpc_task *
971rpc_new_child(struct rpc_clnt *clnt, struct rpc_task *parent)
972{
973 struct rpc_task *task;
974
Trond Myklebust963d8fe2006-01-03 09:55:04 +0100975 task = rpc_new_task(clnt, RPC_TASK_ASYNC | RPC_TASK_CHILD, &rpc_child_ops, parent);
Linus Torvalds1da177e2005-04-16 15:20:36 -0700976 if (!task)
977 goto fail;
Linus Torvalds1da177e2005-04-16 15:20:36 -0700978 return task;
979
980fail:
981 parent->tk_status = -ENOMEM;
982 return NULL;
983}
984
985void rpc_run_child(struct rpc_task *task, struct rpc_task *child, rpc_action func)
986{
987 spin_lock_bh(&childq.lock);
988 /* N.B. Is it possible for the child to have already finished? */
989 __rpc_sleep_on(&childq, task, func, NULL);
990 rpc_schedule_run(child);
991 spin_unlock_bh(&childq.lock);
992}
993
994/*
995 * Kill all tasks for the given client.
996 * XXX: kill their descendants as well?
997 */
998void rpc_killall_tasks(struct rpc_clnt *clnt)
999{
1000 struct rpc_task *rovr;
1001 struct list_head *le;
1002
1003 dprintk("RPC: killing all tasks for client %p\n", clnt);
1004
1005 /*
1006 * Spin lock all_tasks to prevent changes...
1007 */
1008 spin_lock(&rpc_sched_lock);
1009 alltask_for_each(rovr, le, &all_tasks) {
1010 if (! RPC_IS_ACTIVATED(rovr))
1011 continue;
1012 if (!clnt || rovr->tk_client == clnt) {
1013 rovr->tk_flags |= RPC_TASK_KILLED;
1014 rpc_exit(rovr, -EIO);
1015 rpc_wake_up_task(rovr);
1016 }
1017 }
1018 spin_unlock(&rpc_sched_lock);
1019}
1020
1021static DECLARE_MUTEX_LOCKED(rpciod_running);
1022
1023static void rpciod_killall(void)
1024{
1025 unsigned long flags;
1026
1027 while (!list_empty(&all_tasks)) {
1028 clear_thread_flag(TIF_SIGPENDING);
1029 rpc_killall_tasks(NULL);
1030 flush_workqueue(rpciod_workqueue);
1031 if (!list_empty(&all_tasks)) {
1032 dprintk("rpciod_killall: waiting for tasks to exit\n");
1033 yield();
1034 }
1035 }
1036
1037 spin_lock_irqsave(&current->sighand->siglock, flags);
1038 recalc_sigpending();
1039 spin_unlock_irqrestore(&current->sighand->siglock, flags);
1040}
1041
1042/*
1043 * Start up the rpciod process if it's not already running.
1044 */
1045int
1046rpciod_up(void)
1047{
1048 struct workqueue_struct *wq;
1049 int error = 0;
1050
Arjan van de Ven4a3e2f72006-03-20 22:33:17 -08001051 mutex_lock(&rpciod_mutex);
Linus Torvalds1da177e2005-04-16 15:20:36 -07001052 dprintk("rpciod_up: users %d\n", rpciod_users);
1053 rpciod_users++;
1054 if (rpciod_workqueue)
1055 goto out;
1056 /*
1057 * If there's no pid, we should be the first user.
1058 */
1059 if (rpciod_users > 1)
1060 printk(KERN_WARNING "rpciod_up: no workqueue, %d users??\n", rpciod_users);
1061 /*
1062 * Create the rpciod thread and wait for it to start.
1063 */
1064 error = -ENOMEM;
1065 wq = create_workqueue("rpciod");
1066 if (wq == NULL) {
1067 printk(KERN_WARNING "rpciod_up: create workqueue failed, error=%d\n", error);
1068 rpciod_users--;
1069 goto out;
1070 }
1071 rpciod_workqueue = wq;
1072 error = 0;
1073out:
Arjan van de Ven4a3e2f72006-03-20 22:33:17 -08001074 mutex_unlock(&rpciod_mutex);
Linus Torvalds1da177e2005-04-16 15:20:36 -07001075 return error;
1076}
1077
1078void
1079rpciod_down(void)
1080{
Arjan van de Ven4a3e2f72006-03-20 22:33:17 -08001081 mutex_lock(&rpciod_mutex);
Linus Torvalds1da177e2005-04-16 15:20:36 -07001082 dprintk("rpciod_down sema %d\n", rpciod_users);
1083 if (rpciod_users) {
1084 if (--rpciod_users)
1085 goto out;
1086 } else
1087 printk(KERN_WARNING "rpciod_down: no users??\n");
1088
1089 if (!rpciod_workqueue) {
1090 dprintk("rpciod_down: Nothing to do!\n");
1091 goto out;
1092 }
1093 rpciod_killall();
1094
1095 destroy_workqueue(rpciod_workqueue);
1096 rpciod_workqueue = NULL;
1097 out:
Arjan van de Ven4a3e2f72006-03-20 22:33:17 -08001098 mutex_unlock(&rpciod_mutex);
Linus Torvalds1da177e2005-04-16 15:20:36 -07001099}
1100
1101#ifdef RPC_DEBUG
1102void rpc_show_tasks(void)
1103{
1104 struct list_head *le;
1105 struct rpc_task *t;
1106
1107 spin_lock(&rpc_sched_lock);
1108 if (list_empty(&all_tasks)) {
1109 spin_unlock(&rpc_sched_lock);
1110 return;
1111 }
1112 printk("-pid- proc flgs status -client- -prog- --rqstp- -timeout "
Trond Myklebust963d8fe2006-01-03 09:55:04 +01001113 "-rpcwait -action- ---ops--\n");
Linus Torvalds1da177e2005-04-16 15:20:36 -07001114 alltask_for_each(t, le, &all_tasks) {
1115 const char *rpc_waitq = "none";
1116
1117 if (RPC_IS_QUEUED(t))
1118 rpc_waitq = rpc_qname(t->u.tk_wait.rpc_waitq);
1119
1120 printk("%05d %04d %04x %06d %8p %6d %8p %08ld %8s %8p %8p\n",
1121 t->tk_pid,
1122 (t->tk_msg.rpc_proc ? t->tk_msg.rpc_proc->p_proc : -1),
1123 t->tk_flags, t->tk_status,
1124 t->tk_client,
1125 (t->tk_client ? t->tk_client->cl_prog : 0),
1126 t->tk_rqstp, t->tk_timeout,
1127 rpc_waitq,
Trond Myklebust963d8fe2006-01-03 09:55:04 +01001128 t->tk_action, t->tk_ops);
Linus Torvalds1da177e2005-04-16 15:20:36 -07001129 }
1130 spin_unlock(&rpc_sched_lock);
1131}
1132#endif
1133
1134void
1135rpc_destroy_mempool(void)
1136{
1137 if (rpc_buffer_mempool)
1138 mempool_destroy(rpc_buffer_mempool);
1139 if (rpc_task_mempool)
1140 mempool_destroy(rpc_task_mempool);
1141 if (rpc_task_slabp && kmem_cache_destroy(rpc_task_slabp))
1142 printk(KERN_INFO "rpc_task: not all structures were freed\n");
1143 if (rpc_buffer_slabp && kmem_cache_destroy(rpc_buffer_slabp))
1144 printk(KERN_INFO "rpc_buffers: not all structures were freed\n");
1145}
1146
1147int
1148rpc_init_mempool(void)
1149{
1150 rpc_task_slabp = kmem_cache_create("rpc_tasks",
1151 sizeof(struct rpc_task),
1152 0, SLAB_HWCACHE_ALIGN,
1153 NULL, NULL);
1154 if (!rpc_task_slabp)
1155 goto err_nomem;
1156 rpc_buffer_slabp = kmem_cache_create("rpc_buffers",
1157 RPC_BUFFER_MAXSIZE,
1158 0, SLAB_HWCACHE_ALIGN,
1159 NULL, NULL);
1160 if (!rpc_buffer_slabp)
1161 goto err_nomem;
1162 rpc_task_mempool = mempool_create(RPC_TASK_POOLSIZE,
1163 mempool_alloc_slab,
1164 mempool_free_slab,
1165 rpc_task_slabp);
1166 if (!rpc_task_mempool)
1167 goto err_nomem;
1168 rpc_buffer_mempool = mempool_create(RPC_BUFFER_POOLSIZE,
1169 mempool_alloc_slab,
1170 mempool_free_slab,
1171 rpc_buffer_slabp);
1172 if (!rpc_buffer_mempool)
1173 goto err_nomem;
1174 return 0;
1175err_nomem:
1176 rpc_destroy_mempool();
1177 return -ENOMEM;
1178}