sched.c 26.0 KB
Newer Older
L
Linus Torvalds 已提交
1 2 3 4 5 6
/*
 * linux/net/sunrpc/sched.c
 *
 * Scheduling for synchronous and asynchronous RPC requests.
 *
 * Copyright (C) 1996 Olaf Kirch, <okir@monad.swb.de>
7
 *
L
Linus Torvalds 已提交
8 9 10 11 12 13 14 15 16 17 18 19
 * TCP NFS related read + write fixes
 * (C) 1999 Dave Airlie, University of Limerick, Ireland <airlied@linux.ie>
 */

#include <linux/module.h>

#include <linux/sched.h>
#include <linux/interrupt.h>
#include <linux/slab.h>
#include <linux/mempool.h>
#include <linux/smp.h>
#include <linux/spinlock.h>
A
Arjan van de Ven 已提交
20
#include <linux/mutex.h>
L
Linus Torvalds 已提交
21 22 23

#include <linux/sunrpc/clnt.h>

24 25
#include "sunrpc.h"

L
Linus Torvalds 已提交
26 27 28 29 30 31 32 33 34 35 36
#ifdef RPC_DEBUG
#define RPCDBG_FACILITY		RPCDBG_SCHED
#define RPC_TASK_MAGIC_ID	0xf00baa
#endif

/*
 * RPC slabs and memory pools
 */
#define RPC_BUFFER_MAXSIZE	(2048)
#define RPC_BUFFER_POOLSIZE	(8)
#define RPC_TASK_POOLSIZE	(8)
37 38
static struct kmem_cache	*rpc_task_slabp __read_mostly;
static struct kmem_cache	*rpc_buffer_slabp __read_mostly;
39 40
static mempool_t	*rpc_task_mempool __read_mostly;
static mempool_t	*rpc_buffer_mempool __read_mostly;
L
Linus Torvalds 已提交
41

42
static void			rpc_async_schedule(struct work_struct *);
43
static void			 rpc_release_task(struct rpc_task *task);
44
static void __rpc_queue_timer_fn(unsigned long ptr);
L
Linus Torvalds 已提交
45 46 47 48

/*
 * RPC tasks sit here while waiting for conditions to improve.
 */
49
static struct rpc_wait_queue delay_queue;
L
Linus Torvalds 已提交
50 51 52 53

/*
 * rpciod-related stuff
 */
54
struct workqueue_struct *rpciod_workqueue;
L
Linus Torvalds 已提交
55 56 57 58 59 60

/*
 * Disable the timer for a given RPC task. Should be called with
 * queue->lock and bh_disabled in order to avoid races within
 * rpc_run_timer().
 */
61
static void
62
__rpc_disable_timer(struct rpc_wait_queue *queue, struct rpc_task *task)
L
Linus Torvalds 已提交
63
{
64 65
	if (task->tk_timeout == 0)
		return;
66
	dprintk("RPC: %5u disabling timer\n", task->tk_pid);
L
Linus Torvalds 已提交
67
	task->tk_timeout = 0;
68
	list_del(&task->u.tk_wait.timer_list);
69 70
	if (list_empty(&queue->timer_list.list))
		del_timer(&queue->timer_list.timer);
71 72 73 74 75 76 77
}

static void
rpc_set_queue_timer(struct rpc_wait_queue *queue, unsigned long expires)
{
	queue->timer_list.expires = expires;
	mod_timer(&queue->timer_list.timer, expires);
L
Linus Torvalds 已提交
78 79 80 81 82
}

/*
 * Set up a timer for the current task.
 */
83
static void
84
__rpc_add_timer(struct rpc_wait_queue *queue, struct rpc_task *task)
L
Linus Torvalds 已提交
85 86 87 88
{
	if (!task->tk_timeout)
		return;

89
	dprintk("RPC: %5u setting alarm for %lu ms\n",
L
Linus Torvalds 已提交
90 91
			task->tk_pid, task->tk_timeout * 1000 / HZ);

92 93 94 95
	task->u.tk_wait.expires = jiffies + task->tk_timeout;
	if (list_empty(&queue->timer_list.list) || time_before(task->u.tk_wait.expires, queue->timer_list.expires))
		rpc_set_queue_timer(queue, task->u.tk_wait.expires);
	list_add(&task->u.tk_wait.timer_list, &queue->timer_list.list);
L
Linus Torvalds 已提交
96 97 98 99 100 101 102 103 104 105 106 107 108 109 110
}

/*
 * Add new request to a priority queue.
 */
static void __rpc_add_wait_queue_priority(struct rpc_wait_queue *queue, struct rpc_task *task)
{
	struct list_head *q;
	struct rpc_task *t;

	INIT_LIST_HEAD(&task->u.tk_wait.links);
	q = &queue->tasks[task->tk_priority];
	if (unlikely(task->tk_priority > queue->maxpriority))
		q = &queue->tasks[queue->maxpriority];
	list_for_each_entry(t, q, u.tk_wait.list) {
111
		if (t->tk_owner == task->tk_owner) {
L
Linus Torvalds 已提交
112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136
			list_add_tail(&task->u.tk_wait.list, &t->u.tk_wait.links);
			return;
		}
	}
	list_add_tail(&task->u.tk_wait.list, q);
}

/*
 * Add new request to wait queue.
 *
 * Swapper tasks always get inserted at the head of the queue.
 * This should avoid many nasty memory deadlocks and hopefully
 * improve overall performance.
 * Everyone else gets appended to the queue to ensure proper FIFO behavior.
 */
static void __rpc_add_wait_queue(struct rpc_wait_queue *queue, struct rpc_task *task)
{
	BUG_ON (RPC_IS_QUEUED(task));

	if (RPC_IS_PRIORITY(queue))
		__rpc_add_wait_queue_priority(queue, task);
	else if (RPC_IS_SWAPPER(task))
		list_add(&task->u.tk_wait.list, &queue->tasks[0]);
	else
		list_add_tail(&task->u.tk_wait.list, &queue->tasks[0]);
137
	task->tk_waitqueue = queue;
138
	queue->qlen++;
L
Linus Torvalds 已提交
139 140
	rpc_set_queued(task);

141 142
	dprintk("RPC: %5u added to queue %p \"%s\"\n",
			task->tk_pid, queue, rpc_qname(queue));
L
Linus Torvalds 已提交
143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162
}

/*
 * Remove request from a priority queue.
 */
static void __rpc_remove_wait_queue_priority(struct rpc_task *task)
{
	struct rpc_task *t;

	if (!list_empty(&task->u.tk_wait.links)) {
		t = list_entry(task->u.tk_wait.links.next, struct rpc_task, u.tk_wait.list);
		list_move(&t->u.tk_wait.list, &task->u.tk_wait.list);
		list_splice_init(&task->u.tk_wait.links, &t->u.tk_wait.links);
	}
}

/*
 * Remove request from queue.
 * Note: must be called with spin lock held.
 */
163
static void __rpc_remove_wait_queue(struct rpc_wait_queue *queue, struct rpc_task *task)
L
Linus Torvalds 已提交
164
{
165
	__rpc_disable_timer(queue, task);
L
Linus Torvalds 已提交
166 167
	if (RPC_IS_PRIORITY(queue))
		__rpc_remove_wait_queue_priority(task);
168
	list_del(&task->u.tk_wait.list);
169
	queue->qlen--;
170 171
	dprintk("RPC: %5u removed from queue %p \"%s\"\n",
			task->tk_pid, queue, rpc_qname(queue));
L
Linus Torvalds 已提交
172 173 174 175 176 177 178 179
}

static inline void rpc_set_waitqueue_priority(struct rpc_wait_queue *queue, int priority)
{
	queue->priority = priority;
	queue->count = 1 << (priority * 2);
}

180
static inline void rpc_set_waitqueue_owner(struct rpc_wait_queue *queue, pid_t pid)
L
Linus Torvalds 已提交
181
{
182
	queue->owner = pid;
L
Linus Torvalds 已提交
183 184 185 186 187 188
	queue->nr = RPC_BATCH_COUNT;
}

static inline void rpc_reset_waitqueue_priority(struct rpc_wait_queue *queue)
{
	rpc_set_waitqueue_priority(queue, queue->maxpriority);
189
	rpc_set_waitqueue_owner(queue, 0);
L
Linus Torvalds 已提交
190 191
}

192
static void __rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname, unsigned char nr_queues)
L
Linus Torvalds 已提交
193 194 195 196 197 198
{
	int i;

	spin_lock_init(&queue->lock);
	for (i = 0; i < ARRAY_SIZE(queue->tasks); i++)
		INIT_LIST_HEAD(&queue->tasks[i]);
199
	queue->maxpriority = nr_queues - 1;
L
Linus Torvalds 已提交
200
	rpc_reset_waitqueue_priority(queue);
201 202 203
	queue->qlen = 0;
	setup_timer(&queue->timer_list.timer, __rpc_queue_timer_fn, (unsigned long)queue);
	INIT_LIST_HEAD(&queue->timer_list.list);
L
Linus Torvalds 已提交
204 205 206 207 208 209 210
#ifdef RPC_DEBUG
	queue->name = qname;
#endif
}

void rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname)
{
211
	__rpc_init_priority_wait_queue(queue, qname, RPC_NR_PRIORITY);
L
Linus Torvalds 已提交
212
}
213
EXPORT_SYMBOL_GPL(rpc_init_priority_wait_queue);
L
Linus Torvalds 已提交
214 215 216

void rpc_init_wait_queue(struct rpc_wait_queue *queue, const char *qname)
{
217
	__rpc_init_priority_wait_queue(queue, qname, 1);
L
Linus Torvalds 已提交
218
}
219
EXPORT_SYMBOL_GPL(rpc_init_wait_queue);
L
Linus Torvalds 已提交
220

221 222
void rpc_destroy_wait_queue(struct rpc_wait_queue *queue)
{
223
	del_timer_sync(&queue->timer_list.timer);
224 225 226
}
EXPORT_SYMBOL_GPL(rpc_destroy_wait_queue);

227
static int rpc_wait_bit_killable(void *word)
228
{
229
	if (fatal_signal_pending(current))
230 231 232 233 234
		return -ERESTARTSYS;
	schedule();
	return 0;
}

235 236 237 238 239 240 241 242 243 244 245 246 247 248
#ifdef RPC_DEBUG
static void rpc_task_set_debuginfo(struct rpc_task *task)
{
	static atomic_t rpc_pid;

	task->tk_magic = RPC_TASK_MAGIC_ID;
	task->tk_pid = atomic_inc_return(&rpc_pid);
}
#else
static inline void rpc_task_set_debuginfo(struct rpc_task *task)
{
}
#endif

249 250
static void rpc_set_active(struct rpc_task *task)
{
251
	struct rpc_clnt *clnt;
252 253
	if (test_and_set_bit(RPC_TASK_ACTIVE, &task->tk_runstate) != 0)
		return;
254
	rpc_task_set_debuginfo(task);
255
	/* Add to global list of all tasks */
256 257 258 259 260 261
	clnt = task->tk_client;
	if (clnt != NULL) {
		spin_lock(&clnt->cl_lock);
		list_add_tail(&task->tk_task, &clnt->cl_tasks);
		spin_unlock(&clnt->cl_lock);
	}
262 263
}

264 265 266
/*
 * Mark an RPC call as having completed by clearing the 'active' bit
 */
267
static void rpc_mark_complete_task(struct rpc_task *task)
268
{
269 270 271
	smp_mb__before_clear_bit();
	clear_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
	smp_mb__after_clear_bit();
272 273 274 275 276 277 278 279 280
	wake_up_bit(&task->tk_runstate, RPC_TASK_ACTIVE);
}

/*
 * Allow callers to wait for completion of an RPC call
 */
int __rpc_wait_for_completion_task(struct rpc_task *task, int (*action)(void *))
{
	if (action == NULL)
281
		action = rpc_wait_bit_killable;
282
	return wait_on_bit(&task->tk_runstate, RPC_TASK_ACTIVE,
283
			action, TASK_KILLABLE);
284
}
285
EXPORT_SYMBOL_GPL(__rpc_wait_for_completion_task);
286

L
Linus Torvalds 已提交
287 288 289
/*
 * Make an RPC task runnable.
 *
290
 * Note: If the task is ASYNC, this must be called with
L
Linus Torvalds 已提交
291 292 293 294 295
 * the spinlock held to protect the wait queue operation.
 */
static void rpc_make_runnable(struct rpc_task *task)
{
	rpc_clear_queued(task);
296 297
	if (rpc_test_and_set_running(task))
		return;
L
Linus Torvalds 已提交
298 299 300
	if (RPC_IS_ASYNC(task)) {
		int status;

301
		INIT_WORK(&task->u.tk_work, rpc_async_schedule);
302
		status = queue_work(rpciod_workqueue, &task->u.tk_work);
L
Linus Torvalds 已提交
303 304 305 306 307 308
		if (status < 0) {
			printk(KERN_WARNING "RPC: failed to add task to queue: error: %d!\n", status);
			task->tk_status = status;
			return;
		}
	} else
309
		wake_up_bit(&task->tk_runstate, RPC_TASK_QUEUED);
L
Linus Torvalds 已提交
310 311 312 313 314 315 316 317 318
}

/*
 * Prepare for sleeping on a wait queue.
 * By always appending tasks to the list we ensure FIFO behavior.
 * NB: An RPC task will only receive interrupt-driven events as long
 * as it's on a wait queue.
 */
static void __rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
319
			rpc_action action)
L
Linus Torvalds 已提交
320
{
321 322
	dprintk("RPC: %5u sleep_on(queue \"%s\" time %lu)\n",
			task->tk_pid, rpc_qname(q), jiffies);
L
Linus Torvalds 已提交
323 324 325 326 327 328 329 330 331 332

	if (!RPC_IS_ASYNC(task) && !RPC_IS_ACTIVATED(task)) {
		printk(KERN_ERR "RPC: Inactive synchronous task put to sleep!\n");
		return;
	}

	__rpc_add_wait_queue(q, task);

	BUG_ON(task->tk_callback != NULL);
	task->tk_callback = action;
333
	__rpc_add_timer(q, task);
L
Linus Torvalds 已提交
334 335 336
}

void rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
337
				rpc_action action)
L
Linus Torvalds 已提交
338
{
339 340 341
	/* Mark the task as being activated if so needed */
	rpc_set_active(task);

L
Linus Torvalds 已提交
342 343 344 345
	/*
	 * Protect the queue operations.
	 */
	spin_lock_bh(&q->lock);
346
	__rpc_sleep_on(q, task, action);
L
Linus Torvalds 已提交
347 348
	spin_unlock_bh(&q->lock);
}
349
EXPORT_SYMBOL_GPL(rpc_sleep_on);
L
Linus Torvalds 已提交
350 351 352

/**
 * __rpc_do_wake_up_task - wake up a single rpc_task
353
 * @queue: wait queue
L
Linus Torvalds 已提交
354 355 356 357
 * @task: task to be woken up
 *
 * Caller must hold queue->lock, and have cleared the task queued flag.
 */
358
static void __rpc_do_wake_up_task(struct rpc_wait_queue *queue, struct rpc_task *task)
L
Linus Torvalds 已提交
359
{
360 361
	dprintk("RPC: %5u __rpc_wake_up_task (now %lu)\n",
			task->tk_pid, jiffies);
L
Linus Torvalds 已提交
362 363 364 365 366 367 368 369 370 371

#ifdef RPC_DEBUG
	BUG_ON(task->tk_magic != RPC_TASK_MAGIC_ID);
#endif
	/* Has the task been executed yet? If not, we cannot wake it up! */
	if (!RPC_IS_ACTIVATED(task)) {
		printk(KERN_ERR "RPC: Inactive task (%p) being woken up!\n", task);
		return;
	}

372
	__rpc_remove_wait_queue(queue, task);
L
Linus Torvalds 已提交
373 374 375

	rpc_make_runnable(task);

376
	dprintk("RPC:       __rpc_wake_up_task done\n");
L
Linus Torvalds 已提交
377 378 379
}

/*
380
 * Wake up a queued task while the queue lock is being held
L
Linus Torvalds 已提交
381
 */
382
static void rpc_wake_up_task_queue_locked(struct rpc_wait_queue *queue, struct rpc_task *task)
L
Linus Torvalds 已提交
383
{
384 385
	if (RPC_IS_QUEUED(task) && task->tk_waitqueue == queue)
		__rpc_do_wake_up_task(queue, task);
L
Linus Torvalds 已提交
386 387
}

388 389 390 391 392 393 394 395 396 397 398 399 400 401
/*
 * Tests whether rpc queue is empty
 */
int rpc_queue_empty(struct rpc_wait_queue *queue)
{
	int res;

	spin_lock_bh(&queue->lock);
	res = queue->qlen;
	spin_unlock_bh(&queue->lock);
	return (res == 0);
}
EXPORT_SYMBOL_GPL(rpc_queue_empty);

L
Linus Torvalds 已提交
402
/*
403
 * Wake up a task on a specific queue
L
Linus Torvalds 已提交
404
 */
405
void rpc_wake_up_queued_task(struct rpc_wait_queue *queue, struct rpc_task *task)
L
Linus Torvalds 已提交
406
{
407
	spin_lock_bh(&queue->lock);
408
	rpc_wake_up_task_queue_locked(queue, task);
409
	spin_unlock_bh(&queue->lock);
L
Linus Torvalds 已提交
410
}
411 412 413 414 415
EXPORT_SYMBOL_GPL(rpc_wake_up_queued_task);

/*
 * Wake up the specified task
 */
416
static void rpc_wake_up_task(struct rpc_task *task)
417 418 419
{
	rpc_wake_up_queued_task(task->tk_waitqueue, task);
}
L
Linus Torvalds 已提交
420 421 422 423 424 425 426 427 428 429

/*
 * Wake up the next task on a priority queue.
 */
static struct rpc_task * __rpc_wake_up_next_priority(struct rpc_wait_queue *queue)
{
	struct list_head *q;
	struct rpc_task *task;

	/*
430
	 * Service a batch of tasks from a single owner.
L
Linus Torvalds 已提交
431 432 433 434
	 */
	q = &queue->tasks[queue->priority];
	if (!list_empty(q)) {
		task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
435
		if (queue->owner == task->tk_owner) {
L
Linus Torvalds 已提交
436 437 438 439 440 441 442 443
			if (--queue->nr)
				goto out;
			list_move_tail(&task->u.tk_wait.list, q);
		}
		/*
		 * Check if we need to switch queues.
		 */
		if (--queue->count)
444
			goto new_owner;
L
Linus Torvalds 已提交
445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465
	}

	/*
	 * Service the next queue.
	 */
	do {
		if (q == &queue->tasks[0])
			q = &queue->tasks[queue->maxpriority];
		else
			q = q - 1;
		if (!list_empty(q)) {
			task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
			goto new_queue;
		}
	} while (q != &queue->tasks[queue->priority]);

	rpc_reset_waitqueue_priority(queue);
	return NULL;

new_queue:
	rpc_set_waitqueue_priority(queue, (unsigned int)(q - &queue->tasks[0]));
466 467
new_owner:
	rpc_set_waitqueue_owner(queue, task->tk_owner);
L
Linus Torvalds 已提交
468
out:
469
	rpc_wake_up_task_queue_locked(queue, task);
L
Linus Torvalds 已提交
470 471 472 473 474 475 476 477 478 479
	return task;
}

/*
 * Wake up the next task on the wait queue.
 */
struct rpc_task * rpc_wake_up_next(struct rpc_wait_queue *queue)
{
	struct rpc_task	*task = NULL;

480 481
	dprintk("RPC:       wake_up_next(%p \"%s\")\n",
			queue, rpc_qname(queue));
482
	spin_lock_bh(&queue->lock);
L
Linus Torvalds 已提交
483 484 485 486
	if (RPC_IS_PRIORITY(queue))
		task = __rpc_wake_up_next_priority(queue);
	else {
		task_for_first(task, &queue->tasks[0])
487
			rpc_wake_up_task_queue_locked(queue, task);
L
Linus Torvalds 已提交
488
	}
489
	spin_unlock_bh(&queue->lock);
L
Linus Torvalds 已提交
490 491 492

	return task;
}
493
EXPORT_SYMBOL_GPL(rpc_wake_up_next);
L
Linus Torvalds 已提交
494 495 496 497 498 499 500 501 502

/**
 * rpc_wake_up - wake up all rpc_tasks
 * @queue: rpc_wait_queue on which the tasks are sleeping
 *
 * Grabs queue->lock
 */
void rpc_wake_up(struct rpc_wait_queue *queue)
{
503
	struct rpc_task *task, *next;
L
Linus Torvalds 已提交
504
	struct list_head *head;
505

506
	spin_lock_bh(&queue->lock);
L
Linus Torvalds 已提交
507 508
	head = &queue->tasks[queue->maxpriority];
	for (;;) {
509
		list_for_each_entry_safe(task, next, head, u.tk_wait.list)
510
			rpc_wake_up_task_queue_locked(queue, task);
L
Linus Torvalds 已提交
511 512 513 514
		if (head == &queue->tasks[0])
			break;
		head--;
	}
515
	spin_unlock_bh(&queue->lock);
L
Linus Torvalds 已提交
516
}
517
EXPORT_SYMBOL_GPL(rpc_wake_up);
L
Linus Torvalds 已提交
518 519 520 521 522 523 524 525 526 527

/**
 * rpc_wake_up_status - wake up all rpc_tasks and set their status value.
 * @queue: rpc_wait_queue on which the tasks are sleeping
 * @status: status value to set
 *
 * Grabs queue->lock
 */
void rpc_wake_up_status(struct rpc_wait_queue *queue, int status)
{
528
	struct rpc_task *task, *next;
L
Linus Torvalds 已提交
529 530
	struct list_head *head;

531
	spin_lock_bh(&queue->lock);
L
Linus Torvalds 已提交
532 533
	head = &queue->tasks[queue->maxpriority];
	for (;;) {
534
		list_for_each_entry_safe(task, next, head, u.tk_wait.list) {
L
Linus Torvalds 已提交
535
			task->tk_status = status;
536
			rpc_wake_up_task_queue_locked(queue, task);
L
Linus Torvalds 已提交
537 538 539 540 541
		}
		if (head == &queue->tasks[0])
			break;
		head--;
	}
542
	spin_unlock_bh(&queue->lock);
L
Linus Torvalds 已提交
543
}
544
EXPORT_SYMBOL_GPL(rpc_wake_up_status);
L
Linus Torvalds 已提交
545

546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569
static void __rpc_queue_timer_fn(unsigned long ptr)
{
	struct rpc_wait_queue *queue = (struct rpc_wait_queue *)ptr;
	struct rpc_task *task, *n;
	unsigned long expires, now, timeo;

	spin_lock(&queue->lock);
	expires = now = jiffies;
	list_for_each_entry_safe(task, n, &queue->timer_list.list, u.tk_wait.timer_list) {
		timeo = task->u.tk_wait.expires;
		if (time_after_eq(now, timeo)) {
			dprintk("RPC: %5u timeout\n", task->tk_pid);
			task->tk_status = -ETIMEDOUT;
			rpc_wake_up_task_queue_locked(queue, task);
			continue;
		}
		if (expires == now || time_after(expires, timeo))
			expires = timeo;
	}
	if (!list_empty(&queue->timer_list.list))
		rpc_set_queue_timer(queue, expires);
	spin_unlock(&queue->lock);
}

570 571
static void __rpc_atrun(struct rpc_task *task)
{
572
	task->tk_status = 0;
573 574
}

L
Linus Torvalds 已提交
575 576 577
/*
 * Run a task at a later time
 */
578
void rpc_delay(struct rpc_task *task, unsigned long delay)
L
Linus Torvalds 已提交
579 580
{
	task->tk_timeout = delay;
581
	rpc_sleep_on(&delay_queue, task, __rpc_atrun);
L
Linus Torvalds 已提交
582
}
583
EXPORT_SYMBOL_GPL(rpc_delay);
L
Linus Torvalds 已提交
584

T
Trond Myklebust 已提交
585 586 587
/*
 * Helper to call task->tk_ops->rpc_call_prepare
 */
588
void rpc_prepare_task(struct rpc_task *task)
T
Trond Myklebust 已提交
589 590 591 592
{
	task->tk_ops->rpc_call_prepare(task, task->tk_calldata);
}

593
/*
594
 * Helper that calls task->tk_ops->rpc_call_done if it exists
595
 */
T
Trond Myklebust 已提交
596
void rpc_exit_task(struct rpc_task *task)
597
{
T
Trond Myklebust 已提交
598
	task->tk_action = NULL;
599 600
	if (task->tk_ops->rpc_call_done != NULL) {
		task->tk_ops->rpc_call_done(task, task->tk_calldata);
601
		if (task->tk_action != NULL) {
T
Trond Myklebust 已提交
602 603 604
			WARN_ON(RPC_ASSASSINATED(task));
			/* Always release the RPC slot and buffer memory */
			xprt_release(task);
605 606 607
		}
	}
}
608
EXPORT_SYMBOL_GPL(rpc_exit_task);
609

610 611
void rpc_release_calldata(const struct rpc_call_ops *ops, void *calldata)
{
612
	if (ops->rpc_release != NULL)
613 614 615
		ops->rpc_release(calldata);
}

L
Linus Torvalds 已提交
616 617 618
/*
 * This is the RPC `scheduler' (or rather, the finite state machine).
 */
T
Trond Myklebust 已提交
619
static void __rpc_execute(struct rpc_task *task)
L
Linus Torvalds 已提交
620
{
621 622 623
	struct rpc_wait_queue *queue;
	int task_is_async = RPC_IS_ASYNC(task);
	int status = 0;
L
Linus Torvalds 已提交
624

625 626
	dprintk("RPC: %5u __rpc_execute flags=0x%x\n",
			task->tk_pid, task->tk_flags);
L
Linus Torvalds 已提交
627 628 629

	BUG_ON(RPC_IS_QUEUED(task));

630
	for (;;) {
L
Linus Torvalds 已提交
631 632 633 634

		/*
		 * Execute any pending callback.
		 */
635
		if (task->tk_callback) {
L
Linus Torvalds 已提交
636
			void (*save_callback)(struct rpc_task *);
637 638

			/*
639 640
			 * We set tk_callback to NULL before calling it,
			 * in case it sets the tk_callback field itself:
L
Linus Torvalds 已提交
641
			 */
642 643
			save_callback = task->tk_callback;
			task->tk_callback = NULL;
L
Linus Torvalds 已提交
644 645 646 647 648 649 650 651 652
			save_callback(task);
		}

		/*
		 * Perform the next FSM step.
		 * tk_action may be NULL when the task has been killed
		 * by someone else.
		 */
		if (!RPC_IS_QUEUED(task)) {
T
Trond Myklebust 已提交
653
			if (task->tk_action == NULL)
L
Linus Torvalds 已提交
654
				break;
T
Trond Myklebust 已提交
655
			task->tk_action(task);
L
Linus Torvalds 已提交
656 657 658 659 660 661 662
		}

		/*
		 * Lockless check for whether task is sleeping or not.
		 */
		if (!RPC_IS_QUEUED(task))
			continue;
663 664 665 666 667 668 669 670 671 672 673 674 675
		/*
		 * The queue->lock protects against races with
		 * rpc_make_runnable().
		 *
		 * Note that once we clear RPC_TASK_RUNNING on an asynchronous
		 * rpc_task, rpc_make_runnable() can assign it to a
		 * different workqueue. We therefore cannot assume that the
		 * rpc_task pointer may still be dereferenced.
		 */
		queue = task->tk_waitqueue;
		spin_lock_bh(&queue->lock);
		if (!RPC_IS_QUEUED(task)) {
			spin_unlock_bh(&queue->lock);
L
Linus Torvalds 已提交
676 677
			continue;
		}
678 679 680 681
		rpc_clear_running(task);
		spin_unlock_bh(&queue->lock);
		if (task_is_async)
			return;
L
Linus Torvalds 已提交
682 683

		/* sync task: sleep here */
684
		dprintk("RPC: %5u sync task going to sleep\n", task->tk_pid);
685
		status = out_of_line_wait_on_bit(&task->tk_runstate,
686 687
				RPC_TASK_QUEUED, rpc_wait_bit_killable,
				TASK_KILLABLE);
688
		if (status == -ERESTARTSYS) {
L
Linus Torvalds 已提交
689 690 691 692 693 694
			/*
			 * When a sync task receives a signal, it exits with
			 * -ERESTARTSYS. In order to catch any callbacks that
			 * clean up after sleeping on some queue, we don't
			 * break the loop here, but go around once more.
			 */
695
			dprintk("RPC: %5u got signal\n", task->tk_pid);
696 697 698
			task->tk_flags |= RPC_TASK_KILLED;
			rpc_exit(task, -ERESTARTSYS);
			rpc_wake_up_task(task);
L
Linus Torvalds 已提交
699 700
		}
		rpc_set_running(task);
701
		dprintk("RPC: %5u sync task resuming\n", task->tk_pid);
L
Linus Torvalds 已提交
702 703
	}

704 705
	dprintk("RPC: %5u return %d, status %d\n", task->tk_pid, status,
			task->tk_status);
L
Linus Torvalds 已提交
706 707 708 709 710 711 712 713 714 715 716 717 718
	/* Release all resources associated with the task */
	rpc_release_task(task);
}

/*
 * User-visible entry point to the scheduler.
 *
 * This may be called recursively if e.g. an async NFS task updates
 * the attributes and finds that dirty pages must be flushed.
 * NOTE: Upon exit of this function the task is guaranteed to be
 *	 released. In particular note that tk_release() will have
 *	 been called, so your task memory may have been freed.
 */
T
Trond Myklebust 已提交
719
void rpc_execute(struct rpc_task *task)
L
Linus Torvalds 已提交
720
{
721
	rpc_set_active(task);
L
Linus Torvalds 已提交
722
	rpc_set_running(task);
T
Trond Myklebust 已提交
723
	__rpc_execute(task);
L
Linus Torvalds 已提交
724 725
}

726
static void rpc_async_schedule(struct work_struct *work)
L
Linus Torvalds 已提交
727
{
728
	__rpc_execute(container_of(work, struct rpc_task, u.tk_work));
L
Linus Torvalds 已提交
729 730
}

731 732 733 734
/**
 * rpc_malloc - allocate an RPC buffer
 * @task: RPC task that will use this buffer
 * @size: requested byte size
L
Linus Torvalds 已提交
735
 *
736 737 738 739 740 741 742 743
 * To prevent rpciod from hanging, this allocator never sleeps,
 * returning NULL if the request cannot be serviced immediately.
 * The caller can arrange to sleep in a way that is safe for rpciod.
 *
 * Most requests are 'small' (under 2KiB) and can be serviced from a
 * mempool, ensuring that NFS reads and writes can always proceed,
 * and that there is good locality of reference for these buffers.
 *
L
Linus Torvalds 已提交
744
 * In order to avoid memory starvation triggering more writebacks of
745
 * NFS requests, we avoid using GFP_KERNEL.
L
Linus Torvalds 已提交
746
 */
747
void *rpc_malloc(struct rpc_task *task, size_t size)
L
Linus Torvalds 已提交
748
{
749
	struct rpc_buffer *buf;
750
	gfp_t gfp = RPC_IS_SWAPPER(task) ? GFP_ATOMIC : GFP_NOWAIT;
L
Linus Torvalds 已提交
751

752
	size += sizeof(struct rpc_buffer);
753 754
	if (size <= RPC_BUFFER_MAXSIZE)
		buf = mempool_alloc(rpc_buffer_mempool, gfp);
L
Linus Torvalds 已提交
755
	else
756
		buf = kmalloc(size, gfp);
757 758 759 760

	if (!buf)
		return NULL;

761
	buf->len = size;
G
Geert Uytterhoeven 已提交
762
	dprintk("RPC: %5u allocated buffer of size %zu at %p\n",
763
			task->tk_pid, size, buf);
764
	return &buf->data;
L
Linus Torvalds 已提交
765
}
766
EXPORT_SYMBOL_GPL(rpc_malloc);
L
Linus Torvalds 已提交
767

768 769
/**
 * rpc_free - free buffer allocated via rpc_malloc
770
 * @buffer: buffer to free
771 772
 *
 */
773
void rpc_free(void *buffer)
L
Linus Torvalds 已提交
774
{
775 776
	size_t size;
	struct rpc_buffer *buf;
777

778 779
	if (!buffer)
		return;
780 781 782

	buf = container_of(buffer, struct rpc_buffer, data);
	size = buf->len;
783

G
Geert Uytterhoeven 已提交
784
	dprintk("RPC:       freeing buffer of size %zu at %p\n",
785
			size, buf);
786

787 788 789 790
	if (size <= RPC_BUFFER_MAXSIZE)
		mempool_free(buf, rpc_buffer_mempool);
	else
		kfree(buf);
L
Linus Torvalds 已提交
791
}
792
EXPORT_SYMBOL_GPL(rpc_free);
L
Linus Torvalds 已提交
793 794 795 796

/*
 * Creation and deletion of RPC task structures
 */
797
static void rpc_init_task(struct rpc_task *task, const struct rpc_task_setup *task_setup_data)
L
Linus Torvalds 已提交
798 799
{
	memset(task, 0, sizeof(*task));
800
	atomic_set(&task->tk_count, 1);
801 802 803
	task->tk_flags  = task_setup_data->flags;
	task->tk_ops = task_setup_data->callback_ops;
	task->tk_calldata = task_setup_data->callback_data;
804
	INIT_LIST_HEAD(&task->tk_task);
L
Linus Torvalds 已提交
805 806 807 808 809

	/* Initialize retry counters */
	task->tk_garb_retry = 2;
	task->tk_cred_retry = 2;

810 811
	task->tk_priority = task_setup_data->priority - RPC_PRIORITY_LOW;
	task->tk_owner = current->tgid;
L
Linus Torvalds 已提交
812 813

	/* Initialize workqueue for async tasks */
814
	task->tk_workqueue = task_setup_data->workqueue;
L
Linus Torvalds 已提交
815

816 817 818 819
	task->tk_client = task_setup_data->rpc_client;
	if (task->tk_client != NULL) {
		kref_get(&task->tk_client->cl_kref);
		if (task->tk_client->cl_softrtry)
L
Linus Torvalds 已提交
820 821 822
			task->tk_flags |= RPC_TASK_SOFT;
	}

823 824
	if (task->tk_ops->rpc_call_prepare != NULL)
		task->tk_action = rpc_prepare_task;
825

826
	if (task_setup_data->rpc_message != NULL) {
827 828 829
		task->tk_msg.rpc_proc = task_setup_data->rpc_message->rpc_proc;
		task->tk_msg.rpc_argp = task_setup_data->rpc_message->rpc_argp;
		task->tk_msg.rpc_resp = task_setup_data->rpc_message->rpc_resp;
830
		/* Bind the user cred */
831
		rpcauth_bindcred(task, task_setup_data->rpc_message->rpc_cred, task_setup_data->flags);
832 833 834 835
		if (task->tk_action == NULL)
			rpc_call_start(task);
	}

836 837 838
	/* starting timestamp */
	task->tk_start = jiffies;

839
	dprintk("RPC:       new task initialized, procpid %u\n",
840
				task_pid_nr(current));
L
Linus Torvalds 已提交
841 842 843 844 845 846 847 848 849
}

static struct rpc_task *
rpc_alloc_task(void)
{
	return (struct rpc_task *)mempool_alloc(rpc_task_mempool, GFP_NOFS);
}

/*
850
 * Create a new task for the specified client.
L
Linus Torvalds 已提交
851
 */
852
struct rpc_task *rpc_new_task(const struct rpc_task_setup *setup_data)
L
Linus Torvalds 已提交
853
{
854 855 856 857 858
	struct rpc_task	*task = setup_data->task;
	unsigned short flags = 0;

	if (task == NULL) {
		task = rpc_alloc_task();
859 860 861 862 863
		if (task == NULL) {
			rpc_release_calldata(setup_data->callback_ops,
					setup_data->callback_data);
			return ERR_PTR(-ENOMEM);
		}
864 865
		flags = RPC_TASK_DYNAMIC;
	}
L
Linus Torvalds 已提交
866

867
	rpc_init_task(task, setup_data);
868 869 870 871 872
	if (task->tk_status < 0) {
		int err = task->tk_status;
		rpc_put_task(task);
		return ERR_PTR(err);
	}
L
Linus Torvalds 已提交
873

874
	task->tk_flags |= flags;
875
	dprintk("RPC:       allocated task %p\n", task);
L
Linus Torvalds 已提交
876 877 878
	return task;
}

879
static void rpc_free_task(struct rpc_task *task)
L
Linus Torvalds 已提交
880
{
881 882
	const struct rpc_call_ops *tk_ops = task->tk_ops;
	void *calldata = task->tk_calldata;
L
Linus Torvalds 已提交
883

884 885 886 887
	if (task->tk_flags & RPC_TASK_DYNAMIC) {
		dprintk("RPC: %5u freeing task\n", task->tk_pid);
		mempool_free(task, rpc_task_mempool);
	}
888 889 890 891 892 893 894 895 896 897
	rpc_release_calldata(tk_ops, calldata);
}

static void rpc_async_release(struct work_struct *work)
{
	rpc_free_task(container_of(work, struct rpc_task, u.tk_work));
}

void rpc_put_task(struct rpc_task *task)
{
898 899 900 901 902 903 904 905 906 907 908
	if (!atomic_dec_and_test(&task->tk_count))
		return;
	/* Release resources */
	if (task->tk_rqstp)
		xprt_release(task);
	if (task->tk_msg.rpc_cred)
		rpcauth_unbindcred(task);
	if (task->tk_client) {
		rpc_release_client(task->tk_client);
		task->tk_client = NULL;
	}
909 910 911 912 913
	if (task->tk_workqueue != NULL) {
		INIT_WORK(&task->u.tk_work, rpc_async_release);
		queue_work(task->tk_workqueue, &task->u.tk_work);
	} else
		rpc_free_task(task);
914
}
915
EXPORT_SYMBOL_GPL(rpc_put_task);
916

917
static void rpc_release_task(struct rpc_task *task)
918
{
L
Linus Torvalds 已提交
919 920 921
#ifdef RPC_DEBUG
	BUG_ON(task->tk_magic != RPC_TASK_MAGIC_ID);
#endif
922
	dprintk("RPC: %5u release task\n", task->tk_pid);
L
Linus Torvalds 已提交
923

924
	if (!list_empty(&task->tk_task)) {
925
		struct rpc_clnt *clnt = task->tk_client;
926
		/* Remove from client task list */
927
		spin_lock(&clnt->cl_lock);
928
		list_del(&task->tk_task);
929
		spin_unlock(&clnt->cl_lock);
930
	}
L
Linus Torvalds 已提交
931 932 933 934 935
	BUG_ON (RPC_IS_QUEUED(task));

#ifdef RPC_DEBUG
	task->tk_magic = 0;
#endif
936 937 938 939
	/* Wake up anyone who is waiting for task completion */
	rpc_mark_complete_task(task);

	rpc_put_task(task);
L
Linus Torvalds 已提交
940 941 942 943 944 945
}

/*
 * Kill all tasks for the given client.
 * XXX: kill their descendants as well?
 */
946
void rpc_killall_tasks(struct rpc_clnt *clnt)
L
Linus Torvalds 已提交
947 948 949 950
{
	struct rpc_task	*rovr;


951 952 953 954 955 956 957 958
	if (list_empty(&clnt->cl_tasks))
		return;
	dprintk("RPC:       killing all tasks for client %p\n", clnt);
	/*
	 * Spin lock all_tasks to prevent changes...
	 */
	spin_lock(&clnt->cl_lock);
	list_for_each_entry(rovr, &clnt->cl_tasks, tk_task) {
L
Linus Torvalds 已提交
959 960
		if (! RPC_IS_ACTIVATED(rovr))
			continue;
961
		if (!(rovr->tk_flags & RPC_TASK_KILLED)) {
L
Linus Torvalds 已提交
962 963 964 965 966
			rovr->tk_flags |= RPC_TASK_KILLED;
			rpc_exit(rovr, -EIO);
			rpc_wake_up_task(rovr);
		}
	}
967
	spin_unlock(&clnt->cl_lock);
L
Linus Torvalds 已提交
968
}
969
EXPORT_SYMBOL_GPL(rpc_killall_tasks);
L
Linus Torvalds 已提交
970

971 972 973 974 975 976 977 978 979 980
int rpciod_up(void)
{
	return try_module_get(THIS_MODULE) ? 0 : -EINVAL;
}

void rpciod_down(void)
{
	module_put(THIS_MODULE);
}

L
Linus Torvalds 已提交
981
/*
982
 * Start up the rpciod workqueue.
L
Linus Torvalds 已提交
983
 */
984
static int rpciod_start(void)
L
Linus Torvalds 已提交
985 986
{
	struct workqueue_struct *wq;
T
Trond Myklebust 已提交
987

L
Linus Torvalds 已提交
988 989 990
	/*
	 * Create the rpciod thread and wait for it to start.
	 */
T
Trond Myklebust 已提交
991
	dprintk("RPC:       creating workqueue rpciod\n");
L
Linus Torvalds 已提交
992 993
	wq = create_workqueue("rpciod");
	rpciod_workqueue = wq;
994
	return rpciod_workqueue != NULL;
L
Linus Torvalds 已提交
995 996
}

997
static void rpciod_stop(void)
L
Linus Torvalds 已提交
998
{
999
	struct workqueue_struct *wq = NULL;
T
Trond Myklebust 已提交
1000

1001 1002
	if (rpciod_workqueue == NULL)
		return;
T
Trond Myklebust 已提交
1003
	dprintk("RPC:       destroying workqueue rpciod\n");
L
Linus Torvalds 已提交
1004

1005 1006 1007
	wq = rpciod_workqueue;
	rpciod_workqueue = NULL;
	destroy_workqueue(wq);
L
Linus Torvalds 已提交
1008 1009 1010 1011 1012
}

void
rpc_destroy_mempool(void)
{
1013
	rpciod_stop();
L
Linus Torvalds 已提交
1014 1015 1016 1017
	if (rpc_buffer_mempool)
		mempool_destroy(rpc_buffer_mempool);
	if (rpc_task_mempool)
		mempool_destroy(rpc_task_mempool);
1018 1019 1020 1021
	if (rpc_task_slabp)
		kmem_cache_destroy(rpc_task_slabp);
	if (rpc_buffer_slabp)
		kmem_cache_destroy(rpc_buffer_slabp);
1022
	rpc_destroy_wait_queue(&delay_queue);
L
Linus Torvalds 已提交
1023 1024 1025 1026 1027
}

int
rpc_init_mempool(void)
{
1028 1029 1030 1031 1032 1033 1034 1035
	/*
	 * The following is not strictly a mempool initialisation,
	 * but there is no harm in doing it here
	 */
	rpc_init_wait_queue(&delay_queue, "delayq");
	if (!rpciod_start())
		goto err_nomem;

L
Linus Torvalds 已提交
1036 1037 1038
	rpc_task_slabp = kmem_cache_create("rpc_tasks",
					     sizeof(struct rpc_task),
					     0, SLAB_HWCACHE_ALIGN,
1039
					     NULL);
L
Linus Torvalds 已提交
1040 1041 1042 1043 1044
	if (!rpc_task_slabp)
		goto err_nomem;
	rpc_buffer_slabp = kmem_cache_create("rpc_buffers",
					     RPC_BUFFER_MAXSIZE,
					     0, SLAB_HWCACHE_ALIGN,
1045
					     NULL);
L
Linus Torvalds 已提交
1046 1047
	if (!rpc_buffer_slabp)
		goto err_nomem;
1048 1049
	rpc_task_mempool = mempool_create_slab_pool(RPC_TASK_POOLSIZE,
						    rpc_task_slabp);
L
Linus Torvalds 已提交
1050 1051
	if (!rpc_task_mempool)
		goto err_nomem;
1052 1053
	rpc_buffer_mempool = mempool_create_slab_pool(RPC_BUFFER_POOLSIZE,
						      rpc_buffer_slabp);
L
Linus Torvalds 已提交
1054 1055 1056 1057 1058 1059 1060
	if (!rpc_buffer_mempool)
		goto err_nomem;
	return 0;
err_nomem:
	rpc_destroy_mempool();
	return -ENOMEM;
}