SAFE public projects git trees. - safe/jmp/linux-2.6/blob - net/sunrpc/sched.c

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