1 /* 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> 21 #include <linux/mutex.h> 22 23 #include <linux/sunrpc/clnt.h> 24 25 #ifdef RPC_DEBUG 26 #define RPCDBG_FACILITY RPCDBG_SCHED 27 #define RPC_TASK_MAGIC_ID 0xf00baa 28 #endif 29 30 /* 31 * RPC slabs and memory pools 32 */ 33 #define RPC_BUFFER_MAXSIZE (2048) 34 #define RPC_BUFFER_POOLSIZE (8) 35 #define RPC_TASK_POOLSIZE (8) 36 static struct kmem_cache *rpc_task_slabp __read_mostly; 37 static struct kmem_cache *rpc_buffer_slabp __read_mostly; 38 static mempool_t *rpc_task_mempool __read_mostly; 39 static mempool_t *rpc_buffer_mempool __read_mostly; 40 41 static void __rpc_default_timer(struct rpc_task *task); 42 static void rpc_async_schedule(struct work_struct *); 43 static void rpc_release_task(struct rpc_task *task); 44 45 /* 46 * RPC tasks sit here while waiting for conditions to improve. 47 */ 48 static RPC_WAITQ(delay_queue, "delayq"); 49 50 /* 51 * rpciod-related stuff 52 */ 53 static DEFINE_MUTEX(rpciod_mutex); 54 static atomic_t rpciod_users = ATOMIC_INIT(0); 55 struct workqueue_struct *rpciod_workqueue; 56 57 /* 58 * Disable the timer for a given RPC task. Should be called with 59 * queue->lock and bh_disabled in order to avoid races within 60 * rpc_run_timer(). 61 */ 62 static inline void 63 __rpc_disable_timer(struct rpc_task *task) 64 { 65 dprintk("RPC: %5u disabling timer\n", task->tk_pid); 66 task->tk_timeout_fn = NULL; 67 task->tk_timeout = 0; 68 } 69 70 /* 71 * Run a timeout function. 72 * We use the callback in order to allow __rpc_wake_up_task() 73 * and friends to disable the timer synchronously on SMP systems 74 * without calling del_timer_sync(). The latter could cause a 75 * deadlock if called while we're holding spinlocks... 76 */ 77 static void rpc_run_timer(struct rpc_task *task) 78 { 79 void (*callback)(struct rpc_task *); 80 81 callback = task->tk_timeout_fn; 82 task->tk_timeout_fn = NULL; 83 if (callback && RPC_IS_QUEUED(task)) { 84 dprintk("RPC: %5u running timer\n", task->tk_pid); 85 callback(task); 86 } 87 smp_mb__before_clear_bit(); 88 clear_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate); 89 smp_mb__after_clear_bit(); 90 } 91 92 /* 93 * Set up a timer for the current task. 94 */ 95 static inline void 96 __rpc_add_timer(struct rpc_task *task, rpc_action timer) 97 { 98 if (!task->tk_timeout) 99 return; 100 101 dprintk("RPC: %5u setting alarm for %lu ms\n", 102 task->tk_pid, task->tk_timeout * 1000 / HZ); 103 104 if (timer) 105 task->tk_timeout_fn = timer; 106 else 107 task->tk_timeout_fn = __rpc_default_timer; 108 set_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate); 109 mod_timer(&task->tk_timer, jiffies + task->tk_timeout); 110 } 111 112 /* 113 * Delete any timer for the current task. Because we use del_timer_sync(), 114 * this function should never be called while holding queue->lock. 115 */ 116 static void 117 rpc_delete_timer(struct rpc_task *task) 118 { 119 if (RPC_IS_QUEUED(task)) 120 return; 121 if (test_and_clear_bit(RPC_TASK_HAS_TIMER, &task->tk_runstate)) { 122 del_singleshot_timer_sync(&task->tk_timer); 123 dprintk("RPC: %5u deleting timer\n", task->tk_pid); 124 } 125 } 126 127 /* 128 * Add new request to a priority queue. 129 */ 130 static void __rpc_add_wait_queue_priority(struct rpc_wait_queue *queue, struct rpc_task *task) 131 { 132 struct list_head *q; 133 struct rpc_task *t; 134 135 INIT_LIST_HEAD(&task->u.tk_wait.links); 136 q = &queue->tasks[task->tk_priority]; 137 if (unlikely(task->tk_priority > queue->maxpriority)) 138 q = &queue->tasks[queue->maxpriority]; 139 list_for_each_entry(t, q, u.tk_wait.list) { 140 if (t->tk_cookie == task->tk_cookie) { 141 list_add_tail(&task->u.tk_wait.list, &t->u.tk_wait.links); 142 return; 143 } 144 } 145 list_add_tail(&task->u.tk_wait.list, q); 146 } 147 148 /* 149 * Add new request to wait queue. 150 * 151 * Swapper tasks always get inserted at the head of the queue. 152 * This should avoid many nasty memory deadlocks and hopefully 153 * improve overall performance. 154 * Everyone else gets appended to the queue to ensure proper FIFO behavior. 155 */ 156 static void __rpc_add_wait_queue(struct rpc_wait_queue *queue, struct rpc_task *task) 157 { 158 BUG_ON (RPC_IS_QUEUED(task)); 159 160 if (RPC_IS_PRIORITY(queue)) 161 __rpc_add_wait_queue_priority(queue, task); 162 else if (RPC_IS_SWAPPER(task)) 163 list_add(&task->u.tk_wait.list, &queue->tasks[0]); 164 else 165 list_add_tail(&task->u.tk_wait.list, &queue->tasks[0]); 166 task->u.tk_wait.rpc_waitq = queue; 167 queue->qlen++; 168 rpc_set_queued(task); 169 170 dprintk("RPC: %5u added to queue %p \"%s\"\n", 171 task->tk_pid, queue, rpc_qname(queue)); 172 } 173 174 /* 175 * Remove request from a priority queue. 176 */ 177 static void __rpc_remove_wait_queue_priority(struct rpc_task *task) 178 { 179 struct rpc_task *t; 180 181 if (!list_empty(&task->u.tk_wait.links)) { 182 t = list_entry(task->u.tk_wait.links.next, struct rpc_task, u.tk_wait.list); 183 list_move(&t->u.tk_wait.list, &task->u.tk_wait.list); 184 list_splice_init(&task->u.tk_wait.links, &t->u.tk_wait.links); 185 } 186 list_del(&task->u.tk_wait.list); 187 } 188 189 /* 190 * Remove request from queue. 191 * Note: must be called with spin lock held. 192 */ 193 static void __rpc_remove_wait_queue(struct rpc_task *task) 194 { 195 struct rpc_wait_queue *queue; 196 queue = task->u.tk_wait.rpc_waitq; 197 198 if (RPC_IS_PRIORITY(queue)) 199 __rpc_remove_wait_queue_priority(task); 200 else 201 list_del(&task->u.tk_wait.list); 202 queue->qlen--; 203 dprintk("RPC: %5u removed from queue %p \"%s\"\n", 204 task->tk_pid, queue, rpc_qname(queue)); 205 } 206 207 static inline void rpc_set_waitqueue_priority(struct rpc_wait_queue *queue, int priority) 208 { 209 queue->priority = priority; 210 queue->count = 1 << (priority * 2); 211 } 212 213 static inline void rpc_set_waitqueue_cookie(struct rpc_wait_queue *queue, unsigned long cookie) 214 { 215 queue->cookie = cookie; 216 queue->nr = RPC_BATCH_COUNT; 217 } 218 219 static inline void rpc_reset_waitqueue_priority(struct rpc_wait_queue *queue) 220 { 221 rpc_set_waitqueue_priority(queue, queue->maxpriority); 222 rpc_set_waitqueue_cookie(queue, 0); 223 } 224 225 static void __rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname, int maxprio) 226 { 227 int i; 228 229 spin_lock_init(&queue->lock); 230 for (i = 0; i < ARRAY_SIZE(queue->tasks); i++) 231 INIT_LIST_HEAD(&queue->tasks[i]); 232 queue->maxpriority = maxprio; 233 rpc_reset_waitqueue_priority(queue); 234 #ifdef RPC_DEBUG 235 queue->name = qname; 236 #endif 237 } 238 239 void rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname) 240 { 241 __rpc_init_priority_wait_queue(queue, qname, RPC_PRIORITY_HIGH); 242 } 243 244 void rpc_init_wait_queue(struct rpc_wait_queue *queue, const char *qname) 245 { 246 __rpc_init_priority_wait_queue(queue, qname, 0); 247 } 248 EXPORT_SYMBOL(rpc_init_wait_queue); 249 250 static int rpc_wait_bit_interruptible(void *word) 251 { 252 if (signal_pending(current)) 253 return -ERESTARTSYS; 254 schedule(); 255 return 0; 256 } 257 258 #ifdef RPC_DEBUG 259 static void rpc_task_set_debuginfo(struct rpc_task *task) 260 { 261 static atomic_t rpc_pid; 262 263 task->tk_magic = RPC_TASK_MAGIC_ID; 264 task->tk_pid = atomic_inc_return(&rpc_pid); 265 } 266 #else 267 static inline void rpc_task_set_debuginfo(struct rpc_task *task) 268 { 269 } 270 #endif 271 272 static void rpc_set_active(struct rpc_task *task) 273 { 274 struct rpc_clnt *clnt; 275 if (test_and_set_bit(RPC_TASK_ACTIVE, &task->tk_runstate) != 0) 276 return; 277 rpc_task_set_debuginfo(task); 278 /* Add to global list of all tasks */ 279 clnt = task->tk_client; 280 if (clnt != NULL) { 281 spin_lock(&clnt->cl_lock); 282 list_add_tail(&task->tk_task, &clnt->cl_tasks); 283 spin_unlock(&clnt->cl_lock); 284 } 285 } 286 287 /* 288 * Mark an RPC call as having completed by clearing the 'active' bit 289 */ 290 static void rpc_mark_complete_task(struct rpc_task *task) 291 { 292 smp_mb__before_clear_bit(); 293 clear_bit(RPC_TASK_ACTIVE, &task->tk_runstate); 294 smp_mb__after_clear_bit(); 295 wake_up_bit(&task->tk_runstate, RPC_TASK_ACTIVE); 296 } 297 298 /* 299 * Allow callers to wait for completion of an RPC call 300 */ 301 int __rpc_wait_for_completion_task(struct rpc_task *task, int (*action)(void *)) 302 { 303 if (action == NULL) 304 action = rpc_wait_bit_interruptible; 305 return wait_on_bit(&task->tk_runstate, RPC_TASK_ACTIVE, 306 action, TASK_INTERRUPTIBLE); 307 } 308 EXPORT_SYMBOL(__rpc_wait_for_completion_task); 309 310 /* 311 * Make an RPC task runnable. 312 * 313 * Note: If the task is ASYNC, this must be called with 314 * the spinlock held to protect the wait queue operation. 315 */ 316 static void rpc_make_runnable(struct rpc_task *task) 317 { 318 BUG_ON(task->tk_timeout_fn); 319 rpc_clear_queued(task); 320 if (rpc_test_and_set_running(task)) 321 return; 322 /* We might have raced */ 323 if (RPC_IS_QUEUED(task)) { 324 rpc_clear_running(task); 325 return; 326 } 327 if (RPC_IS_ASYNC(task)) { 328 int status; 329 330 INIT_WORK(&task->u.tk_work, rpc_async_schedule); 331 status = queue_work(task->tk_workqueue, &task->u.tk_work); 332 if (status < 0) { 333 printk(KERN_WARNING "RPC: failed to add task to queue: error: %d!\n", status); 334 task->tk_status = status; 335 return; 336 } 337 } else 338 wake_up_bit(&task->tk_runstate, RPC_TASK_QUEUED); 339 } 340 341 /* 342 * Prepare for sleeping on a wait queue. 343 * By always appending tasks to the list we ensure FIFO behavior. 344 * NB: An RPC task will only receive interrupt-driven events as long 345 * as it's on a wait queue. 346 */ 347 static void __rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task, 348 rpc_action action, rpc_action timer) 349 { 350 dprintk("RPC: %5u sleep_on(queue \"%s\" time %lu)\n", 351 task->tk_pid, rpc_qname(q), jiffies); 352 353 if (!RPC_IS_ASYNC(task) && !RPC_IS_ACTIVATED(task)) { 354 printk(KERN_ERR "RPC: Inactive synchronous task put to sleep!\n"); 355 return; 356 } 357 358 __rpc_add_wait_queue(q, task); 359 360 BUG_ON(task->tk_callback != NULL); 361 task->tk_callback = action; 362 __rpc_add_timer(task, timer); 363 } 364 365 void rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task, 366 rpc_action action, rpc_action timer) 367 { 368 /* Mark the task as being activated if so needed */ 369 rpc_set_active(task); 370 371 /* 372 * Protect the queue operations. 373 */ 374 spin_lock_bh(&q->lock); 375 __rpc_sleep_on(q, task, action, timer); 376 spin_unlock_bh(&q->lock); 377 } 378 379 /** 380 * __rpc_do_wake_up_task - wake up a single rpc_task 381 * @task: task to be woken up 382 * 383 * Caller must hold queue->lock, and have cleared the task queued flag. 384 */ 385 static void __rpc_do_wake_up_task(struct rpc_task *task) 386 { 387 dprintk("RPC: %5u __rpc_wake_up_task (now %lu)\n", 388 task->tk_pid, jiffies); 389 390 #ifdef RPC_DEBUG 391 BUG_ON(task->tk_magic != RPC_TASK_MAGIC_ID); 392 #endif 393 /* Has the task been executed yet? If not, we cannot wake it up! */ 394 if (!RPC_IS_ACTIVATED(task)) { 395 printk(KERN_ERR "RPC: Inactive task (%p) being woken up!\n", task); 396 return; 397 } 398 399 __rpc_disable_timer(task); 400 __rpc_remove_wait_queue(task); 401 402 rpc_make_runnable(task); 403 404 dprintk("RPC: __rpc_wake_up_task done\n"); 405 } 406 407 /* 408 * Wake up the specified task 409 */ 410 static void __rpc_wake_up_task(struct rpc_task *task) 411 { 412 if (rpc_start_wakeup(task)) { 413 if (RPC_IS_QUEUED(task)) 414 __rpc_do_wake_up_task(task); 415 rpc_finish_wakeup(task); 416 } 417 } 418 419 /* 420 * Default timeout handler if none specified by user 421 */ 422 static void 423 __rpc_default_timer(struct rpc_task *task) 424 { 425 dprintk("RPC: %5u timeout (default timer)\n", task->tk_pid); 426 task->tk_status = -ETIMEDOUT; 427 rpc_wake_up_task(task); 428 } 429 430 /* 431 * Wake up the specified task 432 */ 433 void rpc_wake_up_task(struct rpc_task *task) 434 { 435 rcu_read_lock_bh(); 436 if (rpc_start_wakeup(task)) { 437 if (RPC_IS_QUEUED(task)) { 438 struct rpc_wait_queue *queue = task->u.tk_wait.rpc_waitq; 439 440 /* Note: we're already in a bh-safe context */ 441 spin_lock(&queue->lock); 442 __rpc_do_wake_up_task(task); 443 spin_unlock(&queue->lock); 444 } 445 rpc_finish_wakeup(task); 446 } 447 rcu_read_unlock_bh(); 448 } 449 450 /* 451 * Wake up the next task on a priority queue. 452 */ 453 static struct rpc_task * __rpc_wake_up_next_priority(struct rpc_wait_queue *queue) 454 { 455 struct list_head *q; 456 struct rpc_task *task; 457 458 /* 459 * Service a batch of tasks from a single cookie. 460 */ 461 q = &queue->tasks[queue->priority]; 462 if (!list_empty(q)) { 463 task = list_entry(q->next, struct rpc_task, u.tk_wait.list); 464 if (queue->cookie == task->tk_cookie) { 465 if (--queue->nr) 466 goto out; 467 list_move_tail(&task->u.tk_wait.list, q); 468 } 469 /* 470 * Check if we need to switch queues. 471 */ 472 if (--queue->count) 473 goto new_cookie; 474 } 475 476 /* 477 * Service the next queue. 478 */ 479 do { 480 if (q == &queue->tasks[0]) 481 q = &queue->tasks[queue->maxpriority]; 482 else 483 q = q - 1; 484 if (!list_empty(q)) { 485 task = list_entry(q->next, struct rpc_task, u.tk_wait.list); 486 goto new_queue; 487 } 488 } while (q != &queue->tasks[queue->priority]); 489 490 rpc_reset_waitqueue_priority(queue); 491 return NULL; 492 493 new_queue: 494 rpc_set_waitqueue_priority(queue, (unsigned int)(q - &queue->tasks[0])); 495 new_cookie: 496 rpc_set_waitqueue_cookie(queue, task->tk_cookie); 497 out: 498 __rpc_wake_up_task(task); 499 return task; 500 } 501 502 /* 503 * Wake up the next task on the wait queue. 504 */ 505 struct rpc_task * rpc_wake_up_next(struct rpc_wait_queue *queue) 506 { 507 struct rpc_task *task = NULL; 508 509 dprintk("RPC: wake_up_next(%p \"%s\")\n", 510 queue, rpc_qname(queue)); 511 rcu_read_lock_bh(); 512 spin_lock(&queue->lock); 513 if (RPC_IS_PRIORITY(queue)) 514 task = __rpc_wake_up_next_priority(queue); 515 else { 516 task_for_first(task, &queue->tasks[0]) 517 __rpc_wake_up_task(task); 518 } 519 spin_unlock(&queue->lock); 520 rcu_read_unlock_bh(); 521 522 return task; 523 } 524 525 /** 526 * rpc_wake_up - wake up all rpc_tasks 527 * @queue: rpc_wait_queue on which the tasks are sleeping 528 * 529 * Grabs queue->lock 530 */ 531 void rpc_wake_up(struct rpc_wait_queue *queue) 532 { 533 struct rpc_task *task, *next; 534 struct list_head *head; 535 536 rcu_read_lock_bh(); 537 spin_lock(&queue->lock); 538 head = &queue->tasks[queue->maxpriority]; 539 for (;;) { 540 list_for_each_entry_safe(task, next, head, u.tk_wait.list) 541 __rpc_wake_up_task(task); 542 if (head == &queue->tasks[0]) 543 break; 544 head--; 545 } 546 spin_unlock(&queue->lock); 547 rcu_read_unlock_bh(); 548 } 549 550 /** 551 * rpc_wake_up_status - wake up all rpc_tasks and set their status value. 552 * @queue: rpc_wait_queue on which the tasks are sleeping 553 * @status: status value to set 554 * 555 * Grabs queue->lock 556 */ 557 void rpc_wake_up_status(struct rpc_wait_queue *queue, int status) 558 { 559 struct rpc_task *task, *next; 560 struct list_head *head; 561 562 rcu_read_lock_bh(); 563 spin_lock(&queue->lock); 564 head = &queue->tasks[queue->maxpriority]; 565 for (;;) { 566 list_for_each_entry_safe(task, next, head, u.tk_wait.list) { 567 task->tk_status = status; 568 __rpc_wake_up_task(task); 569 } 570 if (head == &queue->tasks[0]) 571 break; 572 head--; 573 } 574 spin_unlock(&queue->lock); 575 rcu_read_unlock_bh(); 576 } 577 578 static void __rpc_atrun(struct rpc_task *task) 579 { 580 rpc_wake_up_task(task); 581 } 582 583 /* 584 * Run a task at a later time 585 */ 586 void rpc_delay(struct rpc_task *task, unsigned long delay) 587 { 588 task->tk_timeout = delay; 589 rpc_sleep_on(&delay_queue, task, NULL, __rpc_atrun); 590 } 591 592 /* 593 * Helper to call task->tk_ops->rpc_call_prepare 594 */ 595 static void rpc_prepare_task(struct rpc_task *task) 596 { 597 lock_kernel(); 598 task->tk_ops->rpc_call_prepare(task, task->tk_calldata); 599 unlock_kernel(); 600 } 601 602 /* 603 * Helper that calls task->tk_ops->rpc_call_done if it exists 604 */ 605 void rpc_exit_task(struct rpc_task *task) 606 { 607 task->tk_action = NULL; 608 if (task->tk_ops->rpc_call_done != NULL) { 609 lock_kernel(); 610 task->tk_ops->rpc_call_done(task, task->tk_calldata); 611 unlock_kernel(); 612 if (task->tk_action != NULL) { 613 WARN_ON(RPC_ASSASSINATED(task)); 614 /* Always release the RPC slot and buffer memory */ 615 xprt_release(task); 616 } 617 } 618 } 619 EXPORT_SYMBOL(rpc_exit_task); 620 621 void rpc_release_calldata(const struct rpc_call_ops *ops, void *calldata) 622 { 623 if (ops->rpc_release != NULL) { 624 lock_kernel(); 625 ops->rpc_release(calldata); 626 unlock_kernel(); 627 } 628 } 629 630 /* 631 * This is the RPC `scheduler' (or rather, the finite state machine). 632 */ 633 static void __rpc_execute(struct rpc_task *task) 634 { 635 int status = 0; 636 637 dprintk("RPC: %5u __rpc_execute flags=0x%x\n", 638 task->tk_pid, task->tk_flags); 639 640 BUG_ON(RPC_IS_QUEUED(task)); 641 642 for (;;) { 643 /* 644 * Garbage collection of pending timers... 645 */ 646 rpc_delete_timer(task); 647 648 /* 649 * Execute any pending callback. 650 */ 651 if (RPC_DO_CALLBACK(task)) { 652 /* Define a callback save pointer */ 653 void (*save_callback)(struct rpc_task *); 654 655 /* 656 * If a callback exists, save it, reset it, 657 * call it. 658 * The save is needed to stop from resetting 659 * another callback set within the callback handler 660 * - Dave 661 */ 662 save_callback=task->tk_callback; 663 task->tk_callback=NULL; 664 save_callback(task); 665 } 666 667 /* 668 * Perform the next FSM step. 669 * tk_action may be NULL when the task has been killed 670 * by someone else. 671 */ 672 if (!RPC_IS_QUEUED(task)) { 673 if (task->tk_action == NULL) 674 break; 675 task->tk_action(task); 676 } 677 678 /* 679 * Lockless check for whether task is sleeping or not. 680 */ 681 if (!RPC_IS_QUEUED(task)) 682 continue; 683 rpc_clear_running(task); 684 if (RPC_IS_ASYNC(task)) { 685 /* Careful! we may have raced... */ 686 if (RPC_IS_QUEUED(task)) 687 return; 688 if (rpc_test_and_set_running(task)) 689 return; 690 continue; 691 } 692 693 /* sync task: sleep here */ 694 dprintk("RPC: %5u sync task going to sleep\n", task->tk_pid); 695 /* Note: Caller should be using rpc_clnt_sigmask() */ 696 status = out_of_line_wait_on_bit(&task->tk_runstate, 697 RPC_TASK_QUEUED, rpc_wait_bit_interruptible, 698 TASK_INTERRUPTIBLE); 699 if (status == -ERESTARTSYS) { 700 /* 701 * When a sync task receives a signal, it exits with 702 * -ERESTARTSYS. In order to catch any callbacks that 703 * clean up after sleeping on some queue, we don't 704 * break the loop here, but go around once more. 705 */ 706 dprintk("RPC: %5u got signal\n", task->tk_pid); 707 task->tk_flags |= RPC_TASK_KILLED; 708 rpc_exit(task, -ERESTARTSYS); 709 rpc_wake_up_task(task); 710 } 711 rpc_set_running(task); 712 dprintk("RPC: %5u sync task resuming\n", task->tk_pid); 713 } 714 715 dprintk("RPC: %5u return %d, status %d\n", task->tk_pid, status, 716 task->tk_status); 717 /* Release all resources associated with the task */ 718 rpc_release_task(task); 719 } 720 721 /* 722 * User-visible entry point to the scheduler. 723 * 724 * This may be called recursively if e.g. an async NFS task updates 725 * the attributes and finds that dirty pages must be flushed. 726 * NOTE: Upon exit of this function the task is guaranteed to be 727 * released. In particular note that tk_release() will have 728 * been called, so your task memory may have been freed. 729 */ 730 void rpc_execute(struct rpc_task *task) 731 { 732 rpc_set_active(task); 733 rpc_set_running(task); 734 __rpc_execute(task); 735 } 736 737 static void rpc_async_schedule(struct work_struct *work) 738 { 739 __rpc_execute(container_of(work, struct rpc_task, u.tk_work)); 740 } 741 742 struct rpc_buffer { 743 size_t len; 744 char data[]; 745 }; 746 747 /** 748 * rpc_malloc - allocate an RPC buffer 749 * @task: RPC task that will use this buffer 750 * @size: requested byte size 751 * 752 * To prevent rpciod from hanging, this allocator never sleeps, 753 * returning NULL if the request cannot be serviced immediately. 754 * The caller can arrange to sleep in a way that is safe for rpciod. 755 * 756 * Most requests are 'small' (under 2KiB) and can be serviced from a 757 * mempool, ensuring that NFS reads and writes can always proceed, 758 * and that there is good locality of reference for these buffers. 759 * 760 * In order to avoid memory starvation triggering more writebacks of 761 * NFS requests, we avoid using GFP_KERNEL. 762 */ 763 void *rpc_malloc(struct rpc_task *task, size_t size) 764 { 765 struct rpc_buffer *buf; 766 gfp_t gfp = RPC_IS_SWAPPER(task) ? GFP_ATOMIC : GFP_NOWAIT; 767 768 size += sizeof(struct rpc_buffer); 769 if (size <= RPC_BUFFER_MAXSIZE) 770 buf = mempool_alloc(rpc_buffer_mempool, gfp); 771 else 772 buf = kmalloc(size, gfp); 773 774 if (!buf) 775 return NULL; 776 777 buf->len = size; 778 dprintk("RPC: %5u allocated buffer of size %zu at %p\n", 779 task->tk_pid, size, buf); 780 return &buf->data; 781 } 782 783 /** 784 * rpc_free - free buffer allocated via rpc_malloc 785 * @buffer: buffer to free 786 * 787 */ 788 void rpc_free(void *buffer) 789 { 790 size_t size; 791 struct rpc_buffer *buf; 792 793 if (!buffer) 794 return; 795 796 buf = container_of(buffer, struct rpc_buffer, data); 797 size = buf->len; 798 799 dprintk("RPC: freeing buffer of size %zu at %p\n", 800 size, buf); 801 802 if (size <= RPC_BUFFER_MAXSIZE) 803 mempool_free(buf, rpc_buffer_mempool); 804 else 805 kfree(buf); 806 } 807 808 /* 809 * Creation and deletion of RPC task structures 810 */ 811 void rpc_init_task(struct rpc_task *task, struct rpc_clnt *clnt, int flags, const struct rpc_call_ops *tk_ops, void *calldata) 812 { 813 memset(task, 0, sizeof(*task)); 814 init_timer(&task->tk_timer); 815 task->tk_timer.data = (unsigned long) task; 816 task->tk_timer.function = (void (*)(unsigned long)) rpc_run_timer; 817 atomic_set(&task->tk_count, 1); 818 task->tk_client = clnt; 819 task->tk_flags = flags; 820 task->tk_ops = tk_ops; 821 if (tk_ops->rpc_call_prepare != NULL) 822 task->tk_action = rpc_prepare_task; 823 task->tk_calldata = calldata; 824 INIT_LIST_HEAD(&task->tk_task); 825 826 /* Initialize retry counters */ 827 task->tk_garb_retry = 2; 828 task->tk_cred_retry = 2; 829 830 task->tk_priority = RPC_PRIORITY_NORMAL; 831 task->tk_cookie = (unsigned long)current; 832 833 /* Initialize workqueue for async tasks */ 834 task->tk_workqueue = rpciod_workqueue; 835 836 if (clnt) { 837 kref_get(&clnt->cl_kref); 838 if (clnt->cl_softrtry) 839 task->tk_flags |= RPC_TASK_SOFT; 840 if (!clnt->cl_intr) 841 task->tk_flags |= RPC_TASK_NOINTR; 842 } 843 844 BUG_ON(task->tk_ops == NULL); 845 846 /* starting timestamp */ 847 task->tk_start = jiffies; 848 849 dprintk("RPC: new task initialized, procpid %u\n", 850 current->pid); 851 } 852 853 static struct rpc_task * 854 rpc_alloc_task(void) 855 { 856 return (struct rpc_task *)mempool_alloc(rpc_task_mempool, GFP_NOFS); 857 } 858 859 static void rpc_free_task(struct rcu_head *rcu) 860 { 861 struct rpc_task *task = container_of(rcu, struct rpc_task, u.tk_rcu); 862 dprintk("RPC: %5u freeing task\n", task->tk_pid); 863 mempool_free(task, rpc_task_mempool); 864 } 865 866 /* 867 * Create a new task for the specified client. 868 */ 869 struct rpc_task *rpc_new_task(struct rpc_clnt *clnt, int flags, const struct rpc_call_ops *tk_ops, void *calldata) 870 { 871 struct rpc_task *task; 872 873 task = rpc_alloc_task(); 874 if (!task) 875 goto out; 876 877 rpc_init_task(task, clnt, flags, tk_ops, calldata); 878 879 dprintk("RPC: allocated task %p\n", task); 880 task->tk_flags |= RPC_TASK_DYNAMIC; 881 out: 882 return task; 883 } 884 885 886 void rpc_put_task(struct rpc_task *task) 887 { 888 const struct rpc_call_ops *tk_ops = task->tk_ops; 889 void *calldata = task->tk_calldata; 890 891 if (!atomic_dec_and_test(&task->tk_count)) 892 return; 893 /* Release resources */ 894 if (task->tk_rqstp) 895 xprt_release(task); 896 if (task->tk_msg.rpc_cred) 897 rpcauth_unbindcred(task); 898 if (task->tk_client) { 899 rpc_release_client(task->tk_client); 900 task->tk_client = NULL; 901 } 902 if (task->tk_flags & RPC_TASK_DYNAMIC) 903 call_rcu_bh(&task->u.tk_rcu, rpc_free_task); 904 rpc_release_calldata(tk_ops, calldata); 905 } 906 EXPORT_SYMBOL(rpc_put_task); 907 908 static void rpc_release_task(struct rpc_task *task) 909 { 910 #ifdef RPC_DEBUG 911 BUG_ON(task->tk_magic != RPC_TASK_MAGIC_ID); 912 #endif 913 dprintk("RPC: %5u release task\n", task->tk_pid); 914 915 if (!list_empty(&task->tk_task)) { 916 struct rpc_clnt *clnt = task->tk_client; 917 /* Remove from client task list */ 918 spin_lock(&clnt->cl_lock); 919 list_del(&task->tk_task); 920 spin_unlock(&clnt->cl_lock); 921 } 922 BUG_ON (RPC_IS_QUEUED(task)); 923 924 /* Synchronously delete any running timer */ 925 rpc_delete_timer(task); 926 927 #ifdef RPC_DEBUG 928 task->tk_magic = 0; 929 #endif 930 /* Wake up anyone who is waiting for task completion */ 931 rpc_mark_complete_task(task); 932 933 rpc_put_task(task); 934 } 935 936 /* 937 * Kill all tasks for the given client. 938 * XXX: kill their descendants as well? 939 */ 940 void rpc_killall_tasks(struct rpc_clnt *clnt) 941 { 942 struct rpc_task *rovr; 943 944 945 if (list_empty(&clnt->cl_tasks)) 946 return; 947 dprintk("RPC: killing all tasks for client %p\n", clnt); 948 /* 949 * Spin lock all_tasks to prevent changes... 950 */ 951 spin_lock(&clnt->cl_lock); 952 list_for_each_entry(rovr, &clnt->cl_tasks, tk_task) { 953 if (! RPC_IS_ACTIVATED(rovr)) 954 continue; 955 if (!(rovr->tk_flags & RPC_TASK_KILLED)) { 956 rovr->tk_flags |= RPC_TASK_KILLED; 957 rpc_exit(rovr, -EIO); 958 rpc_wake_up_task(rovr); 959 } 960 } 961 spin_unlock(&clnt->cl_lock); 962 } 963 964 /* 965 * Start up the rpciod process if it's not already running. 966 */ 967 int 968 rpciod_up(void) 969 { 970 struct workqueue_struct *wq; 971 int error = 0; 972 973 if (atomic_inc_not_zero(&rpciod_users)) 974 return 0; 975 976 mutex_lock(&rpciod_mutex); 977 978 /* Guard against races with rpciod_down() */ 979 if (rpciod_workqueue != NULL) 980 goto out_ok; 981 /* 982 * Create the rpciod thread and wait for it to start. 983 */ 984 dprintk("RPC: creating workqueue rpciod\n"); 985 error = -ENOMEM; 986 wq = create_workqueue("rpciod"); 987 if (wq == NULL) 988 goto out; 989 990 rpciod_workqueue = wq; 991 error = 0; 992 out_ok: 993 atomic_inc(&rpciod_users); 994 out: 995 mutex_unlock(&rpciod_mutex); 996 return error; 997 } 998 999 void 1000 rpciod_down(void) 1001 { 1002 if (!atomic_dec_and_test(&rpciod_users)) 1003 return; 1004 1005 mutex_lock(&rpciod_mutex); 1006 dprintk("RPC: destroying workqueue rpciod\n"); 1007 1008 if (atomic_read(&rpciod_users) == 0 && rpciod_workqueue != NULL) { 1009 destroy_workqueue(rpciod_workqueue); 1010 rpciod_workqueue = NULL; 1011 } 1012 mutex_unlock(&rpciod_mutex); 1013 } 1014 1015 void 1016 rpc_destroy_mempool(void) 1017 { 1018 if (rpc_buffer_mempool) 1019 mempool_destroy(rpc_buffer_mempool); 1020 if (rpc_task_mempool) 1021 mempool_destroy(rpc_task_mempool); 1022 if (rpc_task_slabp) 1023 kmem_cache_destroy(rpc_task_slabp); 1024 if (rpc_buffer_slabp) 1025 kmem_cache_destroy(rpc_buffer_slabp); 1026 } 1027 1028 int 1029 rpc_init_mempool(void) 1030 { 1031 rpc_task_slabp = kmem_cache_create("rpc_tasks", 1032 sizeof(struct rpc_task), 1033 0, SLAB_HWCACHE_ALIGN, 1034 NULL); 1035 if (!rpc_task_slabp) 1036 goto err_nomem; 1037 rpc_buffer_slabp = kmem_cache_create("rpc_buffers", 1038 RPC_BUFFER_MAXSIZE, 1039 0, SLAB_HWCACHE_ALIGN, 1040 NULL); 1041 if (!rpc_buffer_slabp) 1042 goto err_nomem; 1043 rpc_task_mempool = mempool_create_slab_pool(RPC_TASK_POOLSIZE, 1044 rpc_task_slabp); 1045 if (!rpc_task_mempool) 1046 goto err_nomem; 1047 rpc_buffer_mempool = mempool_create_slab_pool(RPC_BUFFER_POOLSIZE, 1048 rpc_buffer_slabp); 1049 if (!rpc_buffer_mempool) 1050 goto err_nomem; 1051 return 0; 1052 err_nomem: 1053 rpc_destroy_mempool(); 1054 return -ENOMEM; 1055 } 1056