sched.c 26.9 KB
Newer Older
L
Linus Torvalds 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
/*
 * linux/net/sunrpc/sched.c
 *
 * Scheduling for synchronous and asynchronous RPC requests.
 *
 * Copyright (C) 1996 Olaf Kirch, <okir@monad.swb.de>
 * 
 * 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/smp_lock.h>
#include <linux/spinlock.h>
A
Arjan van de Ven 已提交
21
#include <linux/mutex.h>
L
Linus Torvalds 已提交
22 23 24 25 26 27 28 29 30 31 32 33 34 35 36

#include <linux/sunrpc/clnt.h>

#ifdef RPC_DEBUG
#define RPCDBG_FACILITY		RPCDBG_SCHED
#define RPC_TASK_MAGIC_ID	0xf00baa
static int			rpc_task_id;
#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 43

static void			__rpc_default_timer(struct rpc_task *task);
static void			rpciod_killall(void);
44
static void			rpc_async_schedule(struct work_struct *);
L
Linus Torvalds 已提交
45 46 47 48 49 50 51 52 53 54 55 56 57 58

/*
 * RPC tasks sit here while waiting for conditions to improve.
 */
static RPC_WAITQ(delay_queue, "delayq");

/*
 * All RPC tasks are linked into this list
 */
static LIST_HEAD(all_tasks);

/*
 * rpciod-related stuff
 */
A
Arjan van de Ven 已提交
59
static DEFINE_MUTEX(rpciod_mutex);
L
Linus Torvalds 已提交
60
static unsigned int		rpciod_users;
61
struct workqueue_struct *rpciod_workqueue;
L
Linus Torvalds 已提交
62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 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 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177

/*
 * Spinlock for other critical sections of code.
 */
static DEFINE_SPINLOCK(rpc_sched_lock);

/*
 * 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().
 */
static inline void
__rpc_disable_timer(struct rpc_task *task)
{
	dprintk("RPC: %4d disabling timer\n", task->tk_pid);
	task->tk_timeout_fn = NULL;
	task->tk_timeout = 0;
}

/*
 * Run a timeout function.
 * We use the callback in order to allow __rpc_wake_up_task()
 * and friends to disable the timer synchronously on SMP systems
 * without calling del_timer_sync(). The latter could cause a
 * deadlock if called while we're holding spinlocks...
 */
static void rpc_run_timer(struct rpc_task *task)
{
	void (*callback)(struct rpc_task *);

	callback = task->tk_timeout_fn;
	task->tk_timeout_fn = NULL;
	if (callback && RPC_IS_QUEUED(task)) {
		dprintk("RPC: %4d running timer\n", task->tk_pid);
		callback(task);
	}
	smp_mb__before_clear_bit();
	clear_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate);
	smp_mb__after_clear_bit();
}

/*
 * Set up a timer for the current task.
 */
static inline void
__rpc_add_timer(struct rpc_task *task, rpc_action timer)
{
	if (!task->tk_timeout)
		return;

	dprintk("RPC: %4d setting alarm for %lu ms\n",
			task->tk_pid, task->tk_timeout * 1000 / HZ);

	if (timer)
		task->tk_timeout_fn = timer;
	else
		task->tk_timeout_fn = __rpc_default_timer;
	set_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate);
	mod_timer(&task->tk_timer, jiffies + task->tk_timeout);
}

/*
 * Delete any timer for the current task. Because we use del_timer_sync(),
 * this function should never be called while holding queue->lock.
 */
static void
rpc_delete_timer(struct rpc_task *task)
{
	if (RPC_IS_QUEUED(task))
		return;
	if (test_and_clear_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate)) {
		del_singleshot_timer_sync(&task->tk_timer);
		dprintk("RPC: %4d deleting timer\n", task->tk_pid);
	}
}

/*
 * 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) {
		if (t->tk_cookie == task->tk_cookie) {
			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]);
	task->u.tk_wait.rpc_waitq = queue;
178
	queue->qlen++;
L
Linus Torvalds 已提交
179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212
	rpc_set_queued(task);

	dprintk("RPC: %4d added to queue %p \"%s\"\n",
				task->tk_pid, queue, rpc_qname(queue));
}

/*
 * 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);
	}
	list_del(&task->u.tk_wait.list);
}

/*
 * Remove request from queue.
 * Note: must be called with spin lock held.
 */
static void __rpc_remove_wait_queue(struct rpc_task *task)
{
	struct rpc_wait_queue *queue;
	queue = task->u.tk_wait.rpc_waitq;

	if (RPC_IS_PRIORITY(queue))
		__rpc_remove_wait_queue_priority(task);
	else
		list_del(&task->u.tk_wait.list);
213
	queue->qlen--;
L
Linus Torvalds 已提交
214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260
	dprintk("RPC: %4d removed from queue %p \"%s\"\n",
				task->tk_pid, queue, rpc_qname(queue));
}

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

static inline void rpc_set_waitqueue_cookie(struct rpc_wait_queue *queue, unsigned long cookie)
{
	queue->cookie = cookie;
	queue->nr = RPC_BATCH_COUNT;
}

static inline void rpc_reset_waitqueue_priority(struct rpc_wait_queue *queue)
{
	rpc_set_waitqueue_priority(queue, queue->maxpriority);
	rpc_set_waitqueue_cookie(queue, 0);
}

static void __rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname, int maxprio)
{
	int i;

	spin_lock_init(&queue->lock);
	for (i = 0; i < ARRAY_SIZE(queue->tasks); i++)
		INIT_LIST_HEAD(&queue->tasks[i]);
	queue->maxpriority = maxprio;
	rpc_reset_waitqueue_priority(queue);
#ifdef RPC_DEBUG
	queue->name = qname;
#endif
}

void rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname)
{
	__rpc_init_priority_wait_queue(queue, qname, RPC_PRIORITY_HIGH);
}

void rpc_init_wait_queue(struct rpc_wait_queue *queue, const char *qname)
{
	__rpc_init_priority_wait_queue(queue, qname, 0);
}
EXPORT_SYMBOL(rpc_init_wait_queue);

261 262 263 264 265 266 267 268
static int rpc_wait_bit_interruptible(void *word)
{
	if (signal_pending(current))
		return -ERESTARTSYS;
	schedule();
	return 0;
}

269 270 271 272 273 274 275 276 277 278 279 280 281 282
static void rpc_set_active(struct rpc_task *task)
{
	if (test_and_set_bit(RPC_TASK_ACTIVE, &task->tk_runstate) != 0)
		return;
	spin_lock(&rpc_sched_lock);
#ifdef RPC_DEBUG
	task->tk_magic = RPC_TASK_MAGIC_ID;
	task->tk_pid = rpc_task_id++;
#endif
	/* Add to global list of all tasks */
	list_add_tail(&task->tk_task, &all_tasks);
	spin_unlock(&rpc_sched_lock);
}

283 284 285
/*
 * Mark an RPC call as having completed by clearing the 'active' bit
 */
286
static void rpc_mark_complete_task(struct rpc_task *task)
287
{
288 289 290
	smp_mb__before_clear_bit();
	clear_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
	smp_mb__after_clear_bit();
291 292 293 294 295 296 297 298 299 300 301 302 303 304 305
	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)
		action = rpc_wait_bit_interruptible;
	return wait_on_bit(&task->tk_runstate, RPC_TASK_ACTIVE,
			action, TASK_INTERRUPTIBLE);
}
EXPORT_SYMBOL(__rpc_wait_for_completion_task);

L
Linus Torvalds 已提交
306 307 308 309 310 311 312 313 314 315
/*
 * Make an RPC task runnable.
 *
 * Note: If the task is ASYNC, this must be called with 
 * the spinlock held to protect the wait queue operation.
 */
static void rpc_make_runnable(struct rpc_task *task)
{
	BUG_ON(task->tk_timeout_fn);
	rpc_clear_queued(task);
316 317 318 319 320
	if (rpc_test_and_set_running(task))
		return;
	/* We might have raced */
	if (RPC_IS_QUEUED(task)) {
		rpc_clear_running(task);
L
Linus Torvalds 已提交
321
		return;
322
	}
L
Linus Torvalds 已提交
323 324 325
	if (RPC_IS_ASYNC(task)) {
		int status;

326
		INIT_WORK(&task->u.tk_work, rpc_async_schedule);
L
Linus Torvalds 已提交
327 328 329 330 331 332 333
		status = queue_work(task->tk_workqueue, &task->u.tk_work);
		if (status < 0) {
			printk(KERN_WARNING "RPC: failed to add task to queue: error: %d!\n", status);
			task->tk_status = status;
			return;
		}
	} else
334
		wake_up_bit(&task->tk_runstate, RPC_TASK_QUEUED);
L
Linus Torvalds 已提交
335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363
}

/*
 * 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,
			rpc_action action, rpc_action timer)
{
	dprintk("RPC: %4d sleep_on(queue \"%s\" time %ld)\n", task->tk_pid,
				rpc_qname(q), jiffies);

	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;
	__rpc_add_timer(task, timer);
}

void rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
				rpc_action action, rpc_action timer)
{
364 365 366
	/* Mark the task as being activated if so needed */
	rpc_set_active(task);

L
Linus Torvalds 已提交
367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429
	/*
	 * Protect the queue operations.
	 */
	spin_lock_bh(&q->lock);
	__rpc_sleep_on(q, task, action, timer);
	spin_unlock_bh(&q->lock);
}

/**
 * __rpc_do_wake_up_task - wake up a single rpc_task
 * @task: task to be woken up
 *
 * Caller must hold queue->lock, and have cleared the task queued flag.
 */
static void __rpc_do_wake_up_task(struct rpc_task *task)
{
	dprintk("RPC: %4d __rpc_wake_up_task (now %ld)\n", task->tk_pid, jiffies);

#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;
	}

	__rpc_disable_timer(task);
	__rpc_remove_wait_queue(task);

	rpc_make_runnable(task);

	dprintk("RPC:      __rpc_wake_up_task done\n");
}

/*
 * Wake up the specified task
 */
static void __rpc_wake_up_task(struct rpc_task *task)
{
	if (rpc_start_wakeup(task)) {
		if (RPC_IS_QUEUED(task))
			__rpc_do_wake_up_task(task);
		rpc_finish_wakeup(task);
	}
}

/*
 * Default timeout handler if none specified by user
 */
static void
__rpc_default_timer(struct rpc_task *task)
{
	dprintk("RPC: %d timeout (default timer)\n", task->tk_pid);
	task->tk_status = -ETIMEDOUT;
	rpc_wake_up_task(task);
}

/*
 * Wake up the specified task
 */
void rpc_wake_up_task(struct rpc_task *task)
{
430
	rcu_read_lock_bh();
L
Linus Torvalds 已提交
431 432 433 434
	if (rpc_start_wakeup(task)) {
		if (RPC_IS_QUEUED(task)) {
			struct rpc_wait_queue *queue = task->u.tk_wait.rpc_waitq;

435 436
			/* Note: we're already in a bh-safe context */
			spin_lock(&queue->lock);
L
Linus Torvalds 已提交
437
			__rpc_do_wake_up_task(task);
438
			spin_unlock(&queue->lock);
L
Linus Torvalds 已提交
439 440 441
		}
		rpc_finish_wakeup(task);
	}
442
	rcu_read_unlock_bh();
L
Linus Torvalds 已提交
443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504
}

/*
 * 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;

	/*
	 * Service a batch of tasks from a single cookie.
	 */
	q = &queue->tasks[queue->priority];
	if (!list_empty(q)) {
		task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
		if (queue->cookie == task->tk_cookie) {
			if (--queue->nr)
				goto out;
			list_move_tail(&task->u.tk_wait.list, q);
		}
		/*
		 * Check if we need to switch queues.
		 */
		if (--queue->count)
			goto new_cookie;
	}

	/*
	 * 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]));
new_cookie:
	rpc_set_waitqueue_cookie(queue, task->tk_cookie);
out:
	__rpc_wake_up_task(task);
	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;

	dprintk("RPC:      wake_up_next(%p \"%s\")\n", queue, rpc_qname(queue));
505 506
	rcu_read_lock_bh();
	spin_lock(&queue->lock);
L
Linus Torvalds 已提交
507 508 509 510 511 512
	if (RPC_IS_PRIORITY(queue))
		task = __rpc_wake_up_next_priority(queue);
	else {
		task_for_first(task, &queue->tasks[0])
			__rpc_wake_up_task(task);
	}
513 514
	spin_unlock(&queue->lock);
	rcu_read_unlock_bh();
L
Linus Torvalds 已提交
515 516 517 518 519 520 521 522 523 524 525 526

	return task;
}

/**
 * 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)
{
527
	struct rpc_task *task, *next;
L
Linus Torvalds 已提交
528
	struct list_head *head;
529

530 531
	rcu_read_lock_bh();
	spin_lock(&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 536 537 538 539
			__rpc_wake_up_task(task);
		if (head == &queue->tasks[0])
			break;
		head--;
	}
540 541
	spin_unlock(&queue->lock);
	rcu_read_unlock_bh();
L
Linus Torvalds 已提交
542 543 544 545 546 547 548 549 550 551 552
}

/**
 * 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)
{
553
	struct rpc_task *task, *next;
L
Linus Torvalds 已提交
554 555
	struct list_head *head;

556 557
	rcu_read_lock_bh();
	spin_lock(&queue->lock);
L
Linus Torvalds 已提交
558 559
	head = &queue->tasks[queue->maxpriority];
	for (;;) {
560
		list_for_each_entry_safe(task, next, head, u.tk_wait.list) {
L
Linus Torvalds 已提交
561 562 563 564 565 566 567
			task->tk_status = status;
			__rpc_wake_up_task(task);
		}
		if (head == &queue->tasks[0])
			break;
		head--;
	}
568 569
	spin_unlock(&queue->lock);
	rcu_read_unlock_bh();
L
Linus Torvalds 已提交
570 571
}

572 573 574 575 576
static void __rpc_atrun(struct rpc_task *task)
{
	rpc_wake_up_task(task);
}

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

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

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

615 616 617 618 619 620 621 622 623
void rpc_release_calldata(const struct rpc_call_ops *ops, void *calldata)
{
	if (ops->rpc_release != NULL) {
		lock_kernel();
		ops->rpc_release(calldata);
		unlock_kernel();
	}
}

L
Linus Torvalds 已提交
624 625 626 627 628 629 630 631 632 633 634 635
/*
 * This is the RPC `scheduler' (or rather, the finite state machine).
 */
static int __rpc_execute(struct rpc_task *task)
{
	int		status = 0;

	dprintk("RPC: %4d rpc_execute flgs %x\n",
				task->tk_pid, task->tk_flags);

	BUG_ON(RPC_IS_QUEUED(task));

636
	for (;;) {
L
Linus Torvalds 已提交
637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666
		/*
		 * Garbage collection of pending timers...
		 */
		rpc_delete_timer(task);

		/*
		 * Execute any pending callback.
		 */
		if (RPC_DO_CALLBACK(task)) {
			/* Define a callback save pointer */
			void (*save_callback)(struct rpc_task *);
	
			/* 
			 * If a callback exists, save it, reset it,
			 * call it.
			 * The save is needed to stop from resetting
			 * another callback set within the callback handler
			 * - Dave
			 */
			save_callback=task->tk_callback;
			task->tk_callback=NULL;
			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 已提交
667
			if (task->tk_action == NULL)
L
Linus Torvalds 已提交
668
				break;
T
Trond Myklebust 已提交
669
			task->tk_action(task);
L
Linus Torvalds 已提交
670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688
		}

		/*
		 * Lockless check for whether task is sleeping or not.
		 */
		if (!RPC_IS_QUEUED(task))
			continue;
		rpc_clear_running(task);
		if (RPC_IS_ASYNC(task)) {
			/* Careful! we may have raced... */
			if (RPC_IS_QUEUED(task))
				return 0;
			if (rpc_test_and_set_running(task))
				return 0;
			continue;
		}

		/* sync task: sleep here */
		dprintk("RPC: %4d sync task going to sleep\n", task->tk_pid);
689 690 691 692 693
		/* Note: Caller should be using rpc_clnt_sigmask() */
		status = out_of_line_wait_on_bit(&task->tk_runstate,
				RPC_TASK_QUEUED, rpc_wait_bit_interruptible,
				TASK_INTERRUPTIBLE);
		if (status == -ERESTARTSYS) {
L
Linus Torvalds 已提交
694 695 696 697 698 699
			/*
			 * 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.
			 */
700 701 702 703
			dprintk("RPC: %4d got signal\n", task->tk_pid);
			task->tk_flags |= RPC_TASK_KILLED;
			rpc_exit(task, -ERESTARTSYS);
			rpc_wake_up_task(task);
L
Linus Torvalds 已提交
704 705 706 707 708
		}
		rpc_set_running(task);
		dprintk("RPC: %4d sync task resuming\n", task->tk_pid);
	}

709
	dprintk("RPC: %4d, return %d, status %d\n", task->tk_pid, status, task->tk_status);
L
Linus Torvalds 已提交
710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726
	/* Release all resources associated with the task */
	rpc_release_task(task);
	return status;
}

/*
 * 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.
 */
int
rpc_execute(struct rpc_task *task)
{
727
	rpc_set_active(task);
L
Linus Torvalds 已提交
728 729 730 731
	rpc_set_running(task);
	return __rpc_execute(task);
}

732
static void rpc_async_schedule(struct work_struct *work)
L
Linus Torvalds 已提交
733
{
734
	__rpc_execute(container_of(work, struct rpc_task, u.tk_work));
L
Linus Torvalds 已提交
735 736
}

737 738 739 740
/**
 * rpc_malloc - allocate an RPC buffer
 * @task: RPC task that will use this buffer
 * @size: requested byte size
L
Linus Torvalds 已提交
741 742 743 744 745 746
 *
 * We try to ensure that some NFS reads and writes can always proceed
 * by using a mempool when allocating 'small' buffers.
 * In order to avoid memory starvation triggering more writebacks of
 * NFS requests, we use GFP_NOFS rather than GFP_KERNEL.
 */
747
void * rpc_malloc(struct rpc_task *task, size_t size)
L
Linus Torvalds 已提交
748
{
749
	struct rpc_rqst *req = task->tk_rqstp;
A
Al Viro 已提交
750
	gfp_t	gfp;
L
Linus Torvalds 已提交
751 752 753 754 755 756 757

	if (task->tk_flags & RPC_TASK_SWAPPER)
		gfp = GFP_ATOMIC;
	else
		gfp = GFP_NOFS;

	if (size > RPC_BUFFER_MAXSIZE) {
758 759 760
		req->rq_buffer = kmalloc(size, gfp);
		if (req->rq_buffer)
			req->rq_bufsize = size;
L
Linus Torvalds 已提交
761
	} else {
762 763 764
		req->rq_buffer = mempool_alloc(rpc_buffer_mempool, gfp);
		if (req->rq_buffer)
			req->rq_bufsize = RPC_BUFFER_MAXSIZE;
L
Linus Torvalds 已提交
765
	}
766
	return req->rq_buffer;
L
Linus Torvalds 已提交
767 768
}

769 770 771 772 773 774
/**
 * rpc_free - free buffer allocated via rpc_malloc
 * @task: RPC task with a buffer to be freed
 *
 */
void rpc_free(struct rpc_task *task)
L
Linus Torvalds 已提交
775
{
776 777 778 779 780
	struct rpc_rqst *req = task->tk_rqstp;

	if (req->rq_buffer) {
		if (req->rq_bufsize == RPC_BUFFER_MAXSIZE)
			mempool_free(req->rq_buffer, rpc_buffer_mempool);
L
Linus Torvalds 已提交
781
		else
782 783 784
			kfree(req->rq_buffer);
		req->rq_buffer = NULL;
		req->rq_bufsize = 0;
L
Linus Torvalds 已提交
785 786 787 788 789 790
	}
}

/*
 * Creation and deletion of RPC task structures
 */
791
void rpc_init_task(struct rpc_task *task, struct rpc_clnt *clnt, int flags, const struct rpc_call_ops *tk_ops, void *calldata)
L
Linus Torvalds 已提交
792 793 794 795 796
{
	memset(task, 0, sizeof(*task));
	init_timer(&task->tk_timer);
	task->tk_timer.data     = (unsigned long) task;
	task->tk_timer.function = (void (*)(unsigned long)) rpc_run_timer;
797
	atomic_set(&task->tk_count, 1);
L
Linus Torvalds 已提交
798 799
	task->tk_client = clnt;
	task->tk_flags  = flags;
800
	task->tk_ops = tk_ops;
T
Trond Myklebust 已提交
801 802
	if (tk_ops->rpc_call_prepare != NULL)
		task->tk_action = rpc_prepare_task;
803
	task->tk_calldata = calldata;
L
Linus Torvalds 已提交
804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822

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

	task->tk_priority = RPC_PRIORITY_NORMAL;
	task->tk_cookie = (unsigned long)current;

	/* Initialize workqueue for async tasks */
	task->tk_workqueue = rpciod_workqueue;

	if (clnt) {
		atomic_inc(&clnt->cl_users);
		if (clnt->cl_softrtry)
			task->tk_flags |= RPC_TASK_SOFT;
		if (!clnt->cl_intr)
			task->tk_flags |= RPC_TASK_NOINTR;
	}

823 824
	BUG_ON(task->tk_ops == NULL);

825 826 827
	/* starting timestamp */
	task->tk_start = jiffies;

L
Linus Torvalds 已提交
828 829 830 831 832 833 834 835 836 837
	dprintk("RPC: %4d new task procpid %d\n", task->tk_pid,
				current->pid);
}

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

838
static void rpc_free_task(struct rcu_head *rcu)
L
Linus Torvalds 已提交
839
{
840
	struct rpc_task *task = container_of(rcu, struct rpc_task, u.tk_rcu);
L
Linus Torvalds 已提交
841 842 843 844 845 846 847 848 849
	dprintk("RPC: %4d freeing task\n", task->tk_pid);
	mempool_free(task, rpc_task_mempool);
}

/*
 * Create a new task for the specified client.  We have to
 * clean up after an allocation failure, as the client may
 * have specified "oneshot".
 */
850
struct rpc_task *rpc_new_task(struct rpc_clnt *clnt, int flags, const struct rpc_call_ops *tk_ops, void *calldata)
L
Linus Torvalds 已提交
851 852 853 854 855 856 857
{
	struct rpc_task	*task;

	task = rpc_alloc_task();
	if (!task)
		goto cleanup;

858
	rpc_init_task(task, clnt, flags, tk_ops, calldata);
L
Linus Torvalds 已提交
859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875

	dprintk("RPC: %4d allocated task\n", task->tk_pid);
	task->tk_flags |= RPC_TASK_DYNAMIC;
out:
	return task;

cleanup:
	/* Check whether to release the client */
	if (clnt) {
		printk("rpc_new_task: failed, users=%d, oneshot=%d\n",
			atomic_read(&clnt->cl_users), clnt->cl_oneshot);
		atomic_inc(&clnt->cl_users); /* pretend we were used ... */
		rpc_release_client(clnt);
	}
	goto out;
}

876 877

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

882 883 884 885 886 887 888 889 890 891 892 893
	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;
	}
	if (task->tk_flags & RPC_TASK_DYNAMIC)
894
		call_rcu_bh(&task->u.tk_rcu, rpc_free_task);
895
	rpc_release_calldata(tk_ops, calldata);
896 897 898 899 900
}
EXPORT_SYMBOL(rpc_put_task);

void rpc_release_task(struct rpc_task *task)
{
L
Linus Torvalds 已提交
901 902 903
#ifdef RPC_DEBUG
	BUG_ON(task->tk_magic != RPC_TASK_MAGIC_ID);
#endif
904
	dprintk("RPC: %4d release task\n", task->tk_pid);
L
Linus Torvalds 已提交
905 906 907 908 909 910 911 912 913 914 915 916 917 918

	/* Remove from global task list */
	spin_lock(&rpc_sched_lock);
	list_del(&task->tk_task);
	spin_unlock(&rpc_sched_lock);

	BUG_ON (RPC_IS_QUEUED(task));

	/* Synchronously delete any running timer */
	rpc_delete_timer(task);

#ifdef RPC_DEBUG
	task->tk_magic = 0;
#endif
919 920 921 922
	/* Wake up anyone who is waiting for task completion */
	rpc_mark_complete_task(task);

	rpc_put_task(task);
L
Linus Torvalds 已提交
923 924
}

925 926
/**
 * rpc_run_task - Allocate a new RPC task, then run rpc_execute against it
927 928 929 930
 * @clnt: pointer to RPC client
 * @flags: RPC flags
 * @ops: RPC call ops
 * @data: user call data
931 932 933 934 935 936 937
 */
struct rpc_task *rpc_run_task(struct rpc_clnt *clnt, int flags,
					const struct rpc_call_ops *ops,
					void *data)
{
	struct rpc_task *task;
	task = rpc_new_task(clnt, flags, ops, data);
938
	if (task == NULL) {
939
		rpc_release_calldata(ops, data);
940
		return ERR_PTR(-ENOMEM);
941
	}
942 943 944 945 946 947
	atomic_inc(&task->tk_count);
	rpc_execute(task);
	return task;
}
EXPORT_SYMBOL(rpc_run_task);

L
Linus Torvalds 已提交
948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004
/*
 * Kill all tasks for the given client.
 * XXX: kill their descendants as well?
 */
void rpc_killall_tasks(struct rpc_clnt *clnt)
{
	struct rpc_task	*rovr;
	struct list_head *le;

	dprintk("RPC:      killing all tasks for client %p\n", clnt);

	/*
	 * Spin lock all_tasks to prevent changes...
	 */
	spin_lock(&rpc_sched_lock);
	alltask_for_each(rovr, le, &all_tasks) {
		if (! RPC_IS_ACTIVATED(rovr))
			continue;
		if (!clnt || rovr->tk_client == clnt) {
			rovr->tk_flags |= RPC_TASK_KILLED;
			rpc_exit(rovr, -EIO);
			rpc_wake_up_task(rovr);
		}
	}
	spin_unlock(&rpc_sched_lock);
}

static DECLARE_MUTEX_LOCKED(rpciod_running);

static void rpciod_killall(void)
{
	unsigned long flags;

	while (!list_empty(&all_tasks)) {
		clear_thread_flag(TIF_SIGPENDING);
		rpc_killall_tasks(NULL);
		flush_workqueue(rpciod_workqueue);
		if (!list_empty(&all_tasks)) {
			dprintk("rpciod_killall: waiting for tasks to exit\n");
			yield();
		}
	}

	spin_lock_irqsave(&current->sighand->siglock, flags);
	recalc_sigpending();
	spin_unlock_irqrestore(&current->sighand->siglock, flags);
}

/*
 * Start up the rpciod process if it's not already running.
 */
int
rpciod_up(void)
{
	struct workqueue_struct *wq;
	int error = 0;

A
Arjan van de Ven 已提交
1005
	mutex_lock(&rpciod_mutex);
L
Linus Torvalds 已提交
1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027
	dprintk("rpciod_up: users %d\n", rpciod_users);
	rpciod_users++;
	if (rpciod_workqueue)
		goto out;
	/*
	 * If there's no pid, we should be the first user.
	 */
	if (rpciod_users > 1)
		printk(KERN_WARNING "rpciod_up: no workqueue, %d users??\n", rpciod_users);
	/*
	 * Create the rpciod thread and wait for it to start.
	 */
	error = -ENOMEM;
	wq = create_workqueue("rpciod");
	if (wq == NULL) {
		printk(KERN_WARNING "rpciod_up: create workqueue failed, error=%d\n", error);
		rpciod_users--;
		goto out;
	}
	rpciod_workqueue = wq;
	error = 0;
out:
A
Arjan van de Ven 已提交
1028
	mutex_unlock(&rpciod_mutex);
L
Linus Torvalds 已提交
1029 1030 1031 1032 1033 1034
	return error;
}

void
rpciod_down(void)
{
A
Arjan van de Ven 已提交
1035
	mutex_lock(&rpciod_mutex);
L
Linus Torvalds 已提交
1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051
	dprintk("rpciod_down sema %d\n", rpciod_users);
	if (rpciod_users) {
		if (--rpciod_users)
			goto out;
	} else
		printk(KERN_WARNING "rpciod_down: no users??\n");

	if (!rpciod_workqueue) {
		dprintk("rpciod_down: Nothing to do!\n");
		goto out;
	}
	rpciod_killall();

	destroy_workqueue(rpciod_workqueue);
	rpciod_workqueue = NULL;
 out:
A
Arjan van de Ven 已提交
1052
	mutex_unlock(&rpciod_mutex);
L
Linus Torvalds 已提交
1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066
}

#ifdef RPC_DEBUG
void rpc_show_tasks(void)
{
	struct list_head *le;
	struct rpc_task *t;

	spin_lock(&rpc_sched_lock);
	if (list_empty(&all_tasks)) {
		spin_unlock(&rpc_sched_lock);
		return;
	}
	printk("-pid- proc flgs status -client- -prog- --rqstp- -timeout "
1067
		"-rpcwait -action- ---ops--\n");
L
Linus Torvalds 已提交
1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081
	alltask_for_each(t, le, &all_tasks) {
		const char *rpc_waitq = "none";

		if (RPC_IS_QUEUED(t))
			rpc_waitq = rpc_qname(t->u.tk_wait.rpc_waitq);

		printk("%05d %04d %04x %06d %8p %6d %8p %08ld %8s %8p %8p\n",
			t->tk_pid,
			(t->tk_msg.rpc_proc ? t->tk_msg.rpc_proc->p_proc : -1),
			t->tk_flags, t->tk_status,
			t->tk_client,
			(t->tk_client ? t->tk_client->cl_prog : 0),
			t->tk_rqstp, t->tk_timeout,
			rpc_waitq,
1082
			t->tk_action, t->tk_ops);
L
Linus Torvalds 已提交
1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094
	}
	spin_unlock(&rpc_sched_lock);
}
#endif

void
rpc_destroy_mempool(void)
{
	if (rpc_buffer_mempool)
		mempool_destroy(rpc_buffer_mempool);
	if (rpc_task_mempool)
		mempool_destroy(rpc_task_mempool);
1095 1096 1097 1098
	if (rpc_task_slabp)
		kmem_cache_destroy(rpc_task_slabp);
	if (rpc_buffer_slabp)
		kmem_cache_destroy(rpc_buffer_slabp);
L
Linus Torvalds 已提交
1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115
}

int
rpc_init_mempool(void)
{
	rpc_task_slabp = kmem_cache_create("rpc_tasks",
					     sizeof(struct rpc_task),
					     0, SLAB_HWCACHE_ALIGN,
					     NULL, NULL);
	if (!rpc_task_slabp)
		goto err_nomem;
	rpc_buffer_slabp = kmem_cache_create("rpc_buffers",
					     RPC_BUFFER_MAXSIZE,
					     0, SLAB_HWCACHE_ALIGN,
					     NULL, NULL);
	if (!rpc_buffer_slabp)
		goto err_nomem;
1116 1117
	rpc_task_mempool = mempool_create_slab_pool(RPC_TASK_POOLSIZE,
						    rpc_task_slabp);
L
Linus Torvalds 已提交
1118 1119
	if (!rpc_task_mempool)
		goto err_nomem;
1120 1121
	rpc_buffer_mempool = mempool_create_slab_pool(RPC_BUFFER_POOLSIZE,
						      rpc_buffer_slabp);
L
Linus Torvalds 已提交
1122 1123 1124 1125 1126 1127 1128
	if (!rpc_buffer_mempool)
		goto err_nomem;
	return 0;
err_nomem:
	rpc_destroy_mempool();
	return -ENOMEM;
}