~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

TOMOYO Linux Cross Reference
Linux/net/sunrpc/sched.c

Version: ~ [ linux-5.4-rc3 ] ~ [ linux-5.3.6 ] ~ [ linux-5.2.21 ] ~ [ linux-5.1.21 ] ~ [ linux-5.0.21 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.79 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.149 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.196 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.196 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.19.8 ] ~ [ linux-3.18.140 ] ~ [ linux-3.17.8 ] ~ [ linux-3.16.75 ] ~ [ linux-3.15.10 ] ~ [ linux-3.14.79 ] ~ [ linux-3.13.11 ] ~ [ linux-3.12.74 ] ~ [ linux-3.11.10 ] ~ [ linux-3.10.108 ] ~ [ linux-3.9.11 ] ~ [ linux-3.8.13 ] ~ [ linux-3.7.10 ] ~ [ linux-3.6.11 ] ~ [ linux-3.5.7 ] ~ [ linux-3.4.113 ] ~ [ linux-3.3.8 ] ~ [ linux-3.2.102 ] ~ [ linux-3.1.10 ] ~ [ linux-3.0.101 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

  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/spinlock.h>
 20 #include <linux/mutex.h>
 21 #include <linux/freezer.h>
 22 
 23 #include <linux/sunrpc/clnt.h>
 24 
 25 #include "sunrpc.h"
 26 
 27 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
 28 #define RPCDBG_FACILITY         RPCDBG_SCHED
 29 #endif
 30 
 31 #define CREATE_TRACE_POINTS
 32 #include <trace/events/sunrpc.h>
 33 
 34 /*
 35  * RPC slabs and memory pools
 36  */
 37 #define RPC_BUFFER_MAXSIZE      (2048)
 38 #define RPC_BUFFER_POOLSIZE     (8)
 39 #define RPC_TASK_POOLSIZE       (8)
 40 static struct kmem_cache        *rpc_task_slabp __read_mostly;
 41 static struct kmem_cache        *rpc_buffer_slabp __read_mostly;
 42 static mempool_t        *rpc_task_mempool __read_mostly;
 43 static mempool_t        *rpc_buffer_mempool __read_mostly;
 44 
 45 static void                     rpc_async_schedule(struct work_struct *);
 46 static void                      rpc_release_task(struct rpc_task *task);
 47 static void __rpc_queue_timer_fn(unsigned long ptr);
 48 
 49 /*
 50  * RPC tasks sit here while waiting for conditions to improve.
 51  */
 52 static struct rpc_wait_queue delay_queue;
 53 
 54 /*
 55  * rpciod-related stuff
 56  */
 57 struct workqueue_struct *rpciod_workqueue __read_mostly;
 58 struct workqueue_struct *xprtiod_workqueue __read_mostly;
 59 
 60 /*
 61  * Disable the timer for a given RPC task. Should be called with
 62  * queue->lock and bh_disabled in order to avoid races within
 63  * rpc_run_timer().
 64  */
 65 static void
 66 __rpc_disable_timer(struct rpc_wait_queue *queue, struct rpc_task *task)
 67 {
 68         if (task->tk_timeout == 0)
 69                 return;
 70         dprintk("RPC: %5u disabling timer\n", task->tk_pid);
 71         task->tk_timeout = 0;
 72         list_del(&task->u.tk_wait.timer_list);
 73         if (list_empty(&queue->timer_list.list))
 74                 del_timer(&queue->timer_list.timer);
 75 }
 76 
 77 static void
 78 rpc_set_queue_timer(struct rpc_wait_queue *queue, unsigned long expires)
 79 {
 80         queue->timer_list.expires = expires;
 81         mod_timer(&queue->timer_list.timer, expires);
 82 }
 83 
 84 /*
 85  * Set up a timer for the current task.
 86  */
 87 static void
 88 __rpc_add_timer(struct rpc_wait_queue *queue, struct rpc_task *task)
 89 {
 90         if (!task->tk_timeout)
 91                 return;
 92 
 93         dprintk("RPC: %5u setting alarm for %u ms\n",
 94                 task->tk_pid, jiffies_to_msecs(task->tk_timeout));
 95 
 96         task->u.tk_wait.expires = jiffies + task->tk_timeout;
 97         if (list_empty(&queue->timer_list.list) || time_before(task->u.tk_wait.expires, queue->timer_list.expires))
 98                 rpc_set_queue_timer(queue, task->u.tk_wait.expires);
 99         list_add(&task->u.tk_wait.timer_list, &queue->timer_list.list);
100 }
101 
102 static void rpc_rotate_queue_owner(struct rpc_wait_queue *queue)
103 {
104         struct list_head *q = &queue->tasks[queue->priority];
105         struct rpc_task *task;
106 
107         if (!list_empty(q)) {
108                 task = list_first_entry(q, struct rpc_task, u.tk_wait.list);
109                 if (task->tk_owner == queue->owner)
110                         list_move_tail(&task->u.tk_wait.list, q);
111         }
112 }
113 
114 static void rpc_set_waitqueue_priority(struct rpc_wait_queue *queue, int priority)
115 {
116         if (queue->priority != priority) {
117                 /* Fairness: rotate the list when changing priority */
118                 rpc_rotate_queue_owner(queue);
119                 queue->priority = priority;
120         }
121 }
122 
123 static void rpc_set_waitqueue_owner(struct rpc_wait_queue *queue, pid_t pid)
124 {
125         queue->owner = pid;
126         queue->nr = RPC_BATCH_COUNT;
127 }
128 
129 static void rpc_reset_waitqueue_priority(struct rpc_wait_queue *queue)
130 {
131         rpc_set_waitqueue_priority(queue, queue->maxpriority);
132         rpc_set_waitqueue_owner(queue, 0);
133 }
134 
135 /*
136  * Add new request to a priority queue.
137  */
138 static void __rpc_add_wait_queue_priority(struct rpc_wait_queue *queue,
139                 struct rpc_task *task,
140                 unsigned char queue_priority)
141 {
142         struct list_head *q;
143         struct rpc_task *t;
144 
145         INIT_LIST_HEAD(&task->u.tk_wait.links);
146         if (unlikely(queue_priority > queue->maxpriority))
147                 queue_priority = queue->maxpriority;
148         if (queue_priority > queue->priority)
149                 rpc_set_waitqueue_priority(queue, queue_priority);
150         q = &queue->tasks[queue_priority];
151         list_for_each_entry(t, q, u.tk_wait.list) {
152                 if (t->tk_owner == task->tk_owner) {
153                         list_add_tail(&task->u.tk_wait.list, &t->u.tk_wait.links);
154                         return;
155                 }
156         }
157         list_add_tail(&task->u.tk_wait.list, q);
158 }
159 
160 /*
161  * Add new request to wait queue.
162  *
163  * Swapper tasks always get inserted at the head of the queue.
164  * This should avoid many nasty memory deadlocks and hopefully
165  * improve overall performance.
166  * Everyone else gets appended to the queue to ensure proper FIFO behavior.
167  */
168 static void __rpc_add_wait_queue(struct rpc_wait_queue *queue,
169                 struct rpc_task *task,
170                 unsigned char queue_priority)
171 {
172         WARN_ON_ONCE(RPC_IS_QUEUED(task));
173         if (RPC_IS_QUEUED(task))
174                 return;
175 
176         if (RPC_IS_PRIORITY(queue))
177                 __rpc_add_wait_queue_priority(queue, task, queue_priority);
178         else if (RPC_IS_SWAPPER(task))
179                 list_add(&task->u.tk_wait.list, &queue->tasks[0]);
180         else
181                 list_add_tail(&task->u.tk_wait.list, &queue->tasks[0]);
182         task->tk_waitqueue = queue;
183         queue->qlen++;
184         /* barrier matches the read in rpc_wake_up_task_queue_locked() */
185         smp_wmb();
186         rpc_set_queued(task);
187 
188         dprintk("RPC: %5u added to queue %p \"%s\"\n",
189                         task->tk_pid, queue, rpc_qname(queue));
190 }
191 
192 /*
193  * Remove request from a priority queue.
194  */
195 static void __rpc_remove_wait_queue_priority(struct rpc_task *task)
196 {
197         struct rpc_task *t;
198 
199         if (!list_empty(&task->u.tk_wait.links)) {
200                 t = list_entry(task->u.tk_wait.links.next, struct rpc_task, u.tk_wait.list);
201                 list_move(&t->u.tk_wait.list, &task->u.tk_wait.list);
202                 list_splice_init(&task->u.tk_wait.links, &t->u.tk_wait.links);
203         }
204 }
205 
206 /*
207  * Remove request from queue.
208  * Note: must be called with spin lock held.
209  */
210 static void __rpc_remove_wait_queue(struct rpc_wait_queue *queue, struct rpc_task *task)
211 {
212         __rpc_disable_timer(queue, task);
213         if (RPC_IS_PRIORITY(queue))
214                 __rpc_remove_wait_queue_priority(task);
215         list_del(&task->u.tk_wait.list);
216         queue->qlen--;
217         dprintk("RPC: %5u removed from queue %p \"%s\"\n",
218                         task->tk_pid, queue, rpc_qname(queue));
219 }
220 
221 static void __rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname, unsigned char nr_queues)
222 {
223         int i;
224 
225         spin_lock_init(&queue->lock);
226         for (i = 0; i < ARRAY_SIZE(queue->tasks); i++)
227                 INIT_LIST_HEAD(&queue->tasks[i]);
228         queue->maxpriority = nr_queues - 1;
229         rpc_reset_waitqueue_priority(queue);
230         queue->qlen = 0;
231         setup_timer(&queue->timer_list.timer, __rpc_queue_timer_fn, (unsigned long)queue);
232         INIT_LIST_HEAD(&queue->timer_list.list);
233         rpc_assign_waitqueue_name(queue, qname);
234 }
235 
236 void rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname)
237 {
238         __rpc_init_priority_wait_queue(queue, qname, RPC_NR_PRIORITY);
239 }
240 EXPORT_SYMBOL_GPL(rpc_init_priority_wait_queue);
241 
242 void rpc_init_wait_queue(struct rpc_wait_queue *queue, const char *qname)
243 {
244         __rpc_init_priority_wait_queue(queue, qname, 1);
245 }
246 EXPORT_SYMBOL_GPL(rpc_init_wait_queue);
247 
248 void rpc_destroy_wait_queue(struct rpc_wait_queue *queue)
249 {
250         del_timer_sync(&queue->timer_list.timer);
251 }
252 EXPORT_SYMBOL_GPL(rpc_destroy_wait_queue);
253 
254 static int rpc_wait_bit_killable(struct wait_bit_key *key, int mode)
255 {
256         freezable_schedule_unsafe();
257         if (signal_pending_state(mode, current))
258                 return -ERESTARTSYS;
259         return 0;
260 }
261 
262 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) || IS_ENABLED(CONFIG_TRACEPOINTS)
263 static void rpc_task_set_debuginfo(struct rpc_task *task)
264 {
265         static atomic_t rpc_pid;
266 
267         task->tk_pid = atomic_inc_return(&rpc_pid);
268 }
269 #else
270 static inline void rpc_task_set_debuginfo(struct rpc_task *task)
271 {
272 }
273 #endif
274 
275 static void rpc_set_active(struct rpc_task *task)
276 {
277         trace_rpc_task_begin(task->tk_client, task, NULL);
278 
279         rpc_task_set_debuginfo(task);
280         set_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
281 }
282 
283 /*
284  * Mark an RPC call as having completed by clearing the 'active' bit
285  * and then waking up all tasks that were sleeping.
286  */
287 static int rpc_complete_task(struct rpc_task *task)
288 {
289         void *m = &task->tk_runstate;
290         wait_queue_head_t *wq = bit_waitqueue(m, RPC_TASK_ACTIVE);
291         struct wait_bit_key k = __WAIT_BIT_KEY_INITIALIZER(m, RPC_TASK_ACTIVE);
292         unsigned long flags;
293         int ret;
294 
295         trace_rpc_task_complete(task->tk_client, task, NULL);
296 
297         spin_lock_irqsave(&wq->lock, flags);
298         clear_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
299         ret = atomic_dec_and_test(&task->tk_count);
300         if (waitqueue_active(wq))
301                 __wake_up_locked_key(wq, TASK_NORMAL, &k);
302         spin_unlock_irqrestore(&wq->lock, flags);
303         return ret;
304 }
305 
306 /*
307  * Allow callers to wait for completion of an RPC call
308  *
309  * Note the use of out_of_line_wait_on_bit() rather than wait_on_bit()
310  * to enforce taking of the wq->lock and hence avoid races with
311  * rpc_complete_task().
312  */
313 int __rpc_wait_for_completion_task(struct rpc_task *task, wait_bit_action_f *action)
314 {
315         if (action == NULL)
316                 action = rpc_wait_bit_killable;
317         return out_of_line_wait_on_bit(&task->tk_runstate, RPC_TASK_ACTIVE,
318                         action, TASK_KILLABLE);
319 }
320 EXPORT_SYMBOL_GPL(__rpc_wait_for_completion_task);
321 
322 /*
323  * Make an RPC task runnable.
324  *
325  * Note: If the task is ASYNC, and is being made runnable after sitting on an
326  * rpc_wait_queue, this must be called with the queue spinlock held to protect
327  * the wait queue operation.
328  * Note the ordering of rpc_test_and_set_running() and rpc_clear_queued(),
329  * which is needed to ensure that __rpc_execute() doesn't loop (due to the
330  * lockless RPC_IS_QUEUED() test) before we've had a chance to test
331  * the RPC_TASK_RUNNING flag.
332  */
333 static void rpc_make_runnable(struct workqueue_struct *wq,
334                 struct rpc_task *task)
335 {
336         bool need_wakeup = !rpc_test_and_set_running(task);
337 
338         rpc_clear_queued(task);
339         if (!need_wakeup)
340                 return;
341         if (RPC_IS_ASYNC(task)) {
342                 INIT_WORK(&task->u.tk_work, rpc_async_schedule);
343                 queue_work(wq, &task->u.tk_work);
344         } else
345                 wake_up_bit(&task->tk_runstate, RPC_TASK_QUEUED);
346 }
347 
348 /*
349  * Prepare for sleeping on a wait queue.
350  * By always appending tasks to the list we ensure FIFO behavior.
351  * NB: An RPC task will only receive interrupt-driven events as long
352  * as it's on a wait queue.
353  */
354 static void __rpc_sleep_on_priority(struct rpc_wait_queue *q,
355                 struct rpc_task *task,
356                 rpc_action action,
357                 unsigned char queue_priority)
358 {
359         dprintk("RPC: %5u sleep_on(queue \"%s\" time %lu)\n",
360                         task->tk_pid, rpc_qname(q), jiffies);
361 
362         trace_rpc_task_sleep(task->tk_client, task, q);
363 
364         __rpc_add_wait_queue(q, task, queue_priority);
365 
366         WARN_ON_ONCE(task->tk_callback != NULL);
367         task->tk_callback = action;
368         __rpc_add_timer(q, task);
369 }
370 
371 void rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
372                                 rpc_action action)
373 {
374         /* We shouldn't ever put an inactive task to sleep */
375         WARN_ON_ONCE(!RPC_IS_ACTIVATED(task));
376         if (!RPC_IS_ACTIVATED(task)) {
377                 task->tk_status = -EIO;
378                 rpc_put_task_async(task);
379                 return;
380         }
381 
382         /*
383          * Protect the queue operations.
384          */
385         spin_lock_bh(&q->lock);
386         __rpc_sleep_on_priority(q, task, action, task->tk_priority);
387         spin_unlock_bh(&q->lock);
388 }
389 EXPORT_SYMBOL_GPL(rpc_sleep_on);
390 
391 void rpc_sleep_on_priority(struct rpc_wait_queue *q, struct rpc_task *task,
392                 rpc_action action, int priority)
393 {
394         /* We shouldn't ever put an inactive task to sleep */
395         WARN_ON_ONCE(!RPC_IS_ACTIVATED(task));
396         if (!RPC_IS_ACTIVATED(task)) {
397                 task->tk_status = -EIO;
398                 rpc_put_task_async(task);
399                 return;
400         }
401 
402         /*
403          * Protect the queue operations.
404          */
405         spin_lock_bh(&q->lock);
406         __rpc_sleep_on_priority(q, task, action, priority - RPC_PRIORITY_LOW);
407         spin_unlock_bh(&q->lock);
408 }
409 EXPORT_SYMBOL_GPL(rpc_sleep_on_priority);
410 
411 /**
412  * __rpc_do_wake_up_task_on_wq - wake up a single rpc_task
413  * @wq: workqueue on which to run task
414  * @queue: wait queue
415  * @task: task to be woken up
416  *
417  * Caller must hold queue->lock, and have cleared the task queued flag.
418  */
419 static void __rpc_do_wake_up_task_on_wq(struct workqueue_struct *wq,
420                 struct rpc_wait_queue *queue,
421                 struct rpc_task *task)
422 {
423         dprintk("RPC: %5u __rpc_wake_up_task (now %lu)\n",
424                         task->tk_pid, jiffies);
425 
426         /* Has the task been executed yet? If not, we cannot wake it up! */
427         if (!RPC_IS_ACTIVATED(task)) {
428                 printk(KERN_ERR "RPC: Inactive task (%p) being woken up!\n", task);
429                 return;
430         }
431 
432         trace_rpc_task_wakeup(task->tk_client, task, queue);
433 
434         __rpc_remove_wait_queue(queue, task);
435 
436         rpc_make_runnable(wq, task);
437 
438         dprintk("RPC:       __rpc_wake_up_task done\n");
439 }
440 
441 /*
442  * Wake up a queued task while the queue lock is being held
443  */
444 static void rpc_wake_up_task_on_wq_queue_locked(struct workqueue_struct *wq,
445                 struct rpc_wait_queue *queue, struct rpc_task *task)
446 {
447         if (RPC_IS_QUEUED(task)) {
448                 smp_rmb();
449                 if (task->tk_waitqueue == queue)
450                         __rpc_do_wake_up_task_on_wq(wq, queue, task);
451         }
452 }
453 
454 /*
455  * Wake up a queued task while the queue lock is being held
456  */
457 static void rpc_wake_up_task_queue_locked(struct rpc_wait_queue *queue, struct rpc_task *task)
458 {
459         rpc_wake_up_task_on_wq_queue_locked(rpciod_workqueue, queue, task);
460 }
461 
462 /*
463  * Wake up a task on a specific queue
464  */
465 void rpc_wake_up_queued_task(struct rpc_wait_queue *queue, struct rpc_task *task)
466 {
467         spin_lock_bh(&queue->lock);
468         rpc_wake_up_task_queue_locked(queue, task);
469         spin_unlock_bh(&queue->lock);
470 }
471 EXPORT_SYMBOL_GPL(rpc_wake_up_queued_task);
472 
473 /*
474  * Wake up the next task on a priority queue.
475  */
476 static struct rpc_task *__rpc_find_next_queued_priority(struct rpc_wait_queue *queue)
477 {
478         struct list_head *q;
479         struct rpc_task *task;
480 
481         /*
482          * Service a batch of tasks from a single owner.
483          */
484         q = &queue->tasks[queue->priority];
485         if (!list_empty(q)) {
486                 task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
487                 if (queue->owner == task->tk_owner) {
488                         if (--queue->nr)
489                                 goto out;
490                         list_move_tail(&task->u.tk_wait.list, q);
491                 }
492                 /*
493                  * Check if we need to switch queues.
494                  */
495                 goto new_owner;
496         }
497 
498         /*
499          * Service the next queue.
500          */
501         do {
502                 if (q == &queue->tasks[0])
503                         q = &queue->tasks[queue->maxpriority];
504                 else
505                         q = q - 1;
506                 if (!list_empty(q)) {
507                         task = list_entry(q->next, struct rpc_task, u.tk_wait.list);
508                         goto new_queue;
509                 }
510         } while (q != &queue->tasks[queue->priority]);
511 
512         rpc_reset_waitqueue_priority(queue);
513         return NULL;
514 
515 new_queue:
516         rpc_set_waitqueue_priority(queue, (unsigned int)(q - &queue->tasks[0]));
517 new_owner:
518         rpc_set_waitqueue_owner(queue, task->tk_owner);
519 out:
520         return task;
521 }
522 
523 static struct rpc_task *__rpc_find_next_queued(struct rpc_wait_queue *queue)
524 {
525         if (RPC_IS_PRIORITY(queue))
526                 return __rpc_find_next_queued_priority(queue);
527         if (!list_empty(&queue->tasks[0]))
528                 return list_first_entry(&queue->tasks[0], struct rpc_task, u.tk_wait.list);
529         return NULL;
530 }
531 
532 /*
533  * Wake up the first task on the wait queue.
534  */
535 struct rpc_task *rpc_wake_up_first_on_wq(struct workqueue_struct *wq,
536                 struct rpc_wait_queue *queue,
537                 bool (*func)(struct rpc_task *, void *), void *data)
538 {
539         struct rpc_task *task = NULL;
540 
541         dprintk("RPC:       wake_up_first(%p \"%s\")\n",
542                         queue, rpc_qname(queue));
543         spin_lock_bh(&queue->lock);
544         task = __rpc_find_next_queued(queue);
545         if (task != NULL) {
546                 if (func(task, data))
547                         rpc_wake_up_task_on_wq_queue_locked(wq, queue, task);
548                 else
549                         task = NULL;
550         }
551         spin_unlock_bh(&queue->lock);
552 
553         return task;
554 }
555 
556 /*
557  * Wake up the first task on the wait queue.
558  */
559 struct rpc_task *rpc_wake_up_first(struct rpc_wait_queue *queue,
560                 bool (*func)(struct rpc_task *, void *), void *data)
561 {
562         return rpc_wake_up_first_on_wq(rpciod_workqueue, queue, func, data);
563 }
564 EXPORT_SYMBOL_GPL(rpc_wake_up_first);
565 
566 static bool rpc_wake_up_next_func(struct rpc_task *task, void *data)
567 {
568         return true;
569 }
570 
571 /*
572  * Wake up the next task on the wait queue.
573 */
574 struct rpc_task *rpc_wake_up_next(struct rpc_wait_queue *queue)
575 {
576         return rpc_wake_up_first(queue, rpc_wake_up_next_func, NULL);
577 }
578 EXPORT_SYMBOL_GPL(rpc_wake_up_next);
579 
580 /**
581  * rpc_wake_up - wake up all rpc_tasks
582  * @queue: rpc_wait_queue on which the tasks are sleeping
583  *
584  * Grabs queue->lock
585  */
586 void rpc_wake_up(struct rpc_wait_queue *queue)
587 {
588         struct list_head *head;
589 
590         spin_lock_bh(&queue->lock);
591         head = &queue->tasks[queue->maxpriority];
592         for (;;) {
593                 while (!list_empty(head)) {
594                         struct rpc_task *task;
595                         task = list_first_entry(head,
596                                         struct rpc_task,
597                                         u.tk_wait.list);
598                         rpc_wake_up_task_queue_locked(queue, task);
599                 }
600                 if (head == &queue->tasks[0])
601                         break;
602                 head--;
603         }
604         spin_unlock_bh(&queue->lock);
605 }
606 EXPORT_SYMBOL_GPL(rpc_wake_up);
607 
608 /**
609  * rpc_wake_up_status - wake up all rpc_tasks and set their status value.
610  * @queue: rpc_wait_queue on which the tasks are sleeping
611  * @status: status value to set
612  *
613  * Grabs queue->lock
614  */
615 void rpc_wake_up_status(struct rpc_wait_queue *queue, int status)
616 {
617         struct list_head *head;
618 
619         spin_lock_bh(&queue->lock);
620         head = &queue->tasks[queue->maxpriority];
621         for (;;) {
622                 while (!list_empty(head)) {
623                         struct rpc_task *task;
624                         task = list_first_entry(head,
625                                         struct rpc_task,
626                                         u.tk_wait.list);
627                         task->tk_status = status;
628                         rpc_wake_up_task_queue_locked(queue, task);
629                 }
630                 if (head == &queue->tasks[0])
631                         break;
632                 head--;
633         }
634         spin_unlock_bh(&queue->lock);
635 }
636 EXPORT_SYMBOL_GPL(rpc_wake_up_status);
637 
638 static void __rpc_queue_timer_fn(unsigned long ptr)
639 {
640         struct rpc_wait_queue *queue = (struct rpc_wait_queue *)ptr;
641         struct rpc_task *task, *n;
642         unsigned long expires, now, timeo;
643 
644         spin_lock(&queue->lock);
645         expires = now = jiffies;
646         list_for_each_entry_safe(task, n, &queue->timer_list.list, u.tk_wait.timer_list) {
647                 timeo = task->u.tk_wait.expires;
648                 if (time_after_eq(now, timeo)) {
649                         dprintk("RPC: %5u timeout\n", task->tk_pid);
650                         task->tk_status = -ETIMEDOUT;
651                         rpc_wake_up_task_queue_locked(queue, task);
652                         continue;
653                 }
654                 if (expires == now || time_after(expires, timeo))
655                         expires = timeo;
656         }
657         if (!list_empty(&queue->timer_list.list))
658                 rpc_set_queue_timer(queue, expires);
659         spin_unlock(&queue->lock);
660 }
661 
662 static void __rpc_atrun(struct rpc_task *task)
663 {
664         if (task->tk_status == -ETIMEDOUT)
665                 task->tk_status = 0;
666 }
667 
668 /*
669  * Run a task at a later time
670  */
671 void rpc_delay(struct rpc_task *task, unsigned long delay)
672 {
673         task->tk_timeout = delay;
674         rpc_sleep_on(&delay_queue, task, __rpc_atrun);
675 }
676 EXPORT_SYMBOL_GPL(rpc_delay);
677 
678 /*
679  * Helper to call task->tk_ops->rpc_call_prepare
680  */
681 void rpc_prepare_task(struct rpc_task *task)
682 {
683         task->tk_ops->rpc_call_prepare(task, task->tk_calldata);
684 }
685 
686 static void
687 rpc_init_task_statistics(struct rpc_task *task)
688 {
689         /* Initialize retry counters */
690         task->tk_garb_retry = 2;
691         task->tk_cred_retry = 2;
692         task->tk_rebind_retry = 2;
693 
694         /* starting timestamp */
695         task->tk_start = ktime_get();
696 }
697 
698 static void
699 rpc_reset_task_statistics(struct rpc_task *task)
700 {
701         task->tk_timeouts = 0;
702         task->tk_flags &= ~(RPC_CALL_MAJORSEEN|RPC_TASK_KILLED|RPC_TASK_SENT);
703 
704         rpc_init_task_statistics(task);
705 }
706 
707 /*
708  * Helper that calls task->tk_ops->rpc_call_done if it exists
709  */
710 void rpc_exit_task(struct rpc_task *task)
711 {
712         task->tk_action = NULL;
713         if (task->tk_ops->rpc_call_done != NULL) {
714                 task->tk_ops->rpc_call_done(task, task->tk_calldata);
715                 if (task->tk_action != NULL) {
716                         WARN_ON(RPC_ASSASSINATED(task));
717                         /* Always release the RPC slot and buffer memory */
718                         xprt_release(task);
719                         rpc_reset_task_statistics(task);
720                 }
721         }
722 }
723 
724 void rpc_exit(struct rpc_task *task, int status)
725 {
726         task->tk_status = status;
727         task->tk_action = rpc_exit_task;
728         if (RPC_IS_QUEUED(task))
729                 rpc_wake_up_queued_task(task->tk_waitqueue, task);
730 }
731 EXPORT_SYMBOL_GPL(rpc_exit);
732 
733 void rpc_release_calldata(const struct rpc_call_ops *ops, void *calldata)
734 {
735         if (ops->rpc_release != NULL)
736                 ops->rpc_release(calldata);
737 }
738 
739 /*
740  * This is the RPC `scheduler' (or rather, the finite state machine).
741  */
742 static void __rpc_execute(struct rpc_task *task)
743 {
744         struct rpc_wait_queue *queue;
745         int task_is_async = RPC_IS_ASYNC(task);
746         int status = 0;
747 
748         dprintk("RPC: %5u __rpc_execute flags=0x%x\n",
749                         task->tk_pid, task->tk_flags);
750 
751         WARN_ON_ONCE(RPC_IS_QUEUED(task));
752         if (RPC_IS_QUEUED(task))
753                 return;
754 
755         for (;;) {
756                 void (*do_action)(struct rpc_task *);
757 
758                 /*
759                  * Execute any pending callback first.
760                  */
761                 do_action = task->tk_callback;
762                 task->tk_callback = NULL;
763                 if (do_action == NULL) {
764                         /*
765                          * Perform the next FSM step.
766                          * tk_action may be NULL if the task has been killed.
767                          * In particular, note that rpc_killall_tasks may
768                          * do this at any time, so beware when dereferencing.
769                          */
770                         do_action = task->tk_action;
771                         if (do_action == NULL)
772                                 break;
773                 }
774                 trace_rpc_task_run_action(task->tk_client, task, task->tk_action);
775                 do_action(task);
776 
777                 /*
778                  * Lockless check for whether task is sleeping or not.
779                  */
780                 if (!RPC_IS_QUEUED(task))
781                         continue;
782                 /*
783                  * The queue->lock protects against races with
784                  * rpc_make_runnable().
785                  *
786                  * Note that once we clear RPC_TASK_RUNNING on an asynchronous
787                  * rpc_task, rpc_make_runnable() can assign it to a
788                  * different workqueue. We therefore cannot assume that the
789                  * rpc_task pointer may still be dereferenced.
790                  */
791                 queue = task->tk_waitqueue;
792                 spin_lock_bh(&queue->lock);
793                 if (!RPC_IS_QUEUED(task)) {
794                         spin_unlock_bh(&queue->lock);
795                         continue;
796                 }
797                 rpc_clear_running(task);
798                 spin_unlock_bh(&queue->lock);
799                 if (task_is_async)
800                         return;
801 
802                 /* sync task: sleep here */
803                 dprintk("RPC: %5u sync task going to sleep\n", task->tk_pid);
804                 status = out_of_line_wait_on_bit(&task->tk_runstate,
805                                 RPC_TASK_QUEUED, rpc_wait_bit_killable,
806                                 TASK_KILLABLE);
807                 if (status == -ERESTARTSYS) {
808                         /*
809                          * When a sync task receives a signal, it exits with
810                          * -ERESTARTSYS. In order to catch any callbacks that
811                          * clean up after sleeping on some queue, we don't
812                          * break the loop here, but go around once more.
813                          */
814                         dprintk("RPC: %5u got signal\n", task->tk_pid);
815                         task->tk_flags |= RPC_TASK_KILLED;
816                         rpc_exit(task, -ERESTARTSYS);
817                 }
818                 dprintk("RPC: %5u sync task resuming\n", task->tk_pid);
819         }
820 
821         dprintk("RPC: %5u return %d, status %d\n", task->tk_pid, status,
822                         task->tk_status);
823         /* Release all resources associated with the task */
824         rpc_release_task(task);
825 }
826 
827 /*
828  * User-visible entry point to the scheduler.
829  *
830  * This may be called recursively if e.g. an async NFS task updates
831  * the attributes and finds that dirty pages must be flushed.
832  * NOTE: Upon exit of this function the task is guaranteed to be
833  *       released. In particular note that tk_release() will have
834  *       been called, so your task memory may have been freed.
835  */
836 void rpc_execute(struct rpc_task *task)
837 {
838         bool is_async = RPC_IS_ASYNC(task);
839 
840         rpc_set_active(task);
841         rpc_make_runnable(rpciod_workqueue, task);
842         if (!is_async)
843                 __rpc_execute(task);
844 }
845 
846 static void rpc_async_schedule(struct work_struct *work)
847 {
848         __rpc_execute(container_of(work, struct rpc_task, u.tk_work));
849 }
850 
851 /**
852  * rpc_malloc - allocate an RPC buffer
853  * @task: RPC task that will use this buffer
854  * @size: requested byte size
855  *
856  * To prevent rpciod from hanging, this allocator never sleeps,
857  * returning NULL and suppressing warning if the request cannot be serviced
858  * immediately.
859  * The caller can arrange to sleep in a way that is safe for rpciod.
860  *
861  * Most requests are 'small' (under 2KiB) and can be serviced from a
862  * mempool, ensuring that NFS reads and writes can always proceed,
863  * and that there is good locality of reference for these buffers.
864  *
865  * In order to avoid memory starvation triggering more writebacks of
866  * NFS requests, we avoid using GFP_KERNEL.
867  */
868 void *rpc_malloc(struct rpc_task *task, size_t size)
869 {
870         struct rpc_buffer *buf;
871         gfp_t gfp = GFP_NOIO | __GFP_NOWARN;
872 
873         if (RPC_IS_SWAPPER(task))
874                 gfp = __GFP_MEMALLOC | GFP_NOWAIT | __GFP_NOWARN;
875 
876         size += sizeof(struct rpc_buffer);
877         if (size <= RPC_BUFFER_MAXSIZE)
878                 buf = mempool_alloc(rpc_buffer_mempool, gfp);
879         else
880                 buf = kmalloc(size, gfp);
881 
882         if (!buf)
883                 return NULL;
884 
885         buf->len = size;
886         dprintk("RPC: %5u allocated buffer of size %zu at %p\n",
887                         task->tk_pid, size, buf);
888         return &buf->data;
889 }
890 EXPORT_SYMBOL_GPL(rpc_malloc);
891 
892 /**
893  * rpc_free - free buffer allocated via rpc_malloc
894  * @buffer: buffer to free
895  *
896  */
897 void rpc_free(void *buffer)
898 {
899         size_t size;
900         struct rpc_buffer *buf;
901 
902         if (!buffer)
903                 return;
904 
905         buf = container_of(buffer, struct rpc_buffer, data);
906         size = buf->len;
907 
908         dprintk("RPC:       freeing buffer of size %zu at %p\n",
909                         size, buf);
910 
911         if (size <= RPC_BUFFER_MAXSIZE)
912                 mempool_free(buf, rpc_buffer_mempool);
913         else
914                 kfree(buf);
915 }
916 EXPORT_SYMBOL_GPL(rpc_free);
917 
918 /*
919  * Creation and deletion of RPC task structures
920  */
921 static void rpc_init_task(struct rpc_task *task, const struct rpc_task_setup *task_setup_data)
922 {
923         memset(task, 0, sizeof(*task));
924         atomic_set(&task->tk_count, 1);
925         task->tk_flags  = task_setup_data->flags;
926         task->tk_ops = task_setup_data->callback_ops;
927         task->tk_calldata = task_setup_data->callback_data;
928         INIT_LIST_HEAD(&task->tk_task);
929 
930         task->tk_priority = task_setup_data->priority - RPC_PRIORITY_LOW;
931         task->tk_owner = current->tgid;
932 
933         /* Initialize workqueue for async tasks */
934         task->tk_workqueue = task_setup_data->workqueue;
935 
936         task->tk_xprt = xprt_get(task_setup_data->rpc_xprt);
937 
938         if (task->tk_ops->rpc_call_prepare != NULL)
939                 task->tk_action = rpc_prepare_task;
940 
941         rpc_init_task_statistics(task);
942 
943         dprintk("RPC:       new task initialized, procpid %u\n",
944                                 task_pid_nr(current));
945 }
946 
947 static struct rpc_task *
948 rpc_alloc_task(void)
949 {
950         return (struct rpc_task *)mempool_alloc(rpc_task_mempool, GFP_NOIO);
951 }
952 
953 /*
954  * Create a new task for the specified client.
955  */
956 struct rpc_task *rpc_new_task(const struct rpc_task_setup *setup_data)
957 {
958         struct rpc_task *task = setup_data->task;
959         unsigned short flags = 0;
960 
961         if (task == NULL) {
962                 task = rpc_alloc_task();
963                 if (task == NULL) {
964                         rpc_release_calldata(setup_data->callback_ops,
965                                         setup_data->callback_data);
966                         return ERR_PTR(-ENOMEM);
967                 }
968                 flags = RPC_TASK_DYNAMIC;
969         }
970 
971         rpc_init_task(task, setup_data);
972         task->tk_flags |= flags;
973         dprintk("RPC:       allocated task %p\n", task);
974         return task;
975 }
976 
977 /*
978  * rpc_free_task - release rpc task and perform cleanups
979  *
980  * Note that we free up the rpc_task _after_ rpc_release_calldata()
981  * in order to work around a workqueue dependency issue.
982  *
983  * Tejun Heo states:
984  * "Workqueue currently considers two work items to be the same if they're
985  * on the same address and won't execute them concurrently - ie. it
986  * makes a work item which is queued again while being executed wait
987  * for the previous execution to complete.
988  *
989  * If a work function frees the work item, and then waits for an event
990  * which should be performed by another work item and *that* work item
991  * recycles the freed work item, it can create a false dependency loop.
992  * There really is no reliable way to detect this short of verifying
993  * every memory free."
994  *
995  */
996 static void rpc_free_task(struct rpc_task *task)
997 {
998         unsigned short tk_flags = task->tk_flags;
999 
1000         rpc_release_calldata(task->tk_ops, task->tk_calldata);
1001 
1002         if (tk_flags & RPC_TASK_DYNAMIC) {
1003                 dprintk("RPC: %5u freeing task\n", task->tk_pid);
1004                 mempool_free(task, rpc_task_mempool);
1005         }
1006 }
1007 
1008 static void rpc_async_release(struct work_struct *work)
1009 {
1010         rpc_free_task(container_of(work, struct rpc_task, u.tk_work));
1011 }
1012 
1013 static void rpc_release_resources_task(struct rpc_task *task)
1014 {
1015         xprt_release(task);
1016         if (task->tk_msg.rpc_cred) {
1017                 put_rpccred(task->tk_msg.rpc_cred);
1018                 task->tk_msg.rpc_cred = NULL;
1019         }
1020         rpc_task_release_client(task);
1021 }
1022 
1023 static void rpc_final_put_task(struct rpc_task *task,
1024                 struct workqueue_struct *q)
1025 {
1026         if (q != NULL) {
1027                 INIT_WORK(&task->u.tk_work, rpc_async_release);
1028                 queue_work(q, &task->u.tk_work);
1029         } else
1030                 rpc_free_task(task);
1031 }
1032 
1033 static void rpc_do_put_task(struct rpc_task *task, struct workqueue_struct *q)
1034 {
1035         if (atomic_dec_and_test(&task->tk_count)) {
1036                 rpc_release_resources_task(task);
1037                 rpc_final_put_task(task, q);
1038         }
1039 }
1040 
1041 void rpc_put_task(struct rpc_task *task)
1042 {
1043         rpc_do_put_task(task, NULL);
1044 }
1045 EXPORT_SYMBOL_GPL(rpc_put_task);
1046 
1047 void rpc_put_task_async(struct rpc_task *task)
1048 {
1049         rpc_do_put_task(task, task->tk_workqueue);
1050 }
1051 EXPORT_SYMBOL_GPL(rpc_put_task_async);
1052 
1053 static void rpc_release_task(struct rpc_task *task)
1054 {
1055         dprintk("RPC: %5u release task\n", task->tk_pid);
1056 
1057         WARN_ON_ONCE(RPC_IS_QUEUED(task));
1058 
1059         rpc_release_resources_task(task);
1060 
1061         /*
1062          * Note: at this point we have been removed from rpc_clnt->cl_tasks,
1063          * so it should be safe to use task->tk_count as a test for whether
1064          * or not any other processes still hold references to our rpc_task.
1065          */
1066         if (atomic_read(&task->tk_count) != 1 + !RPC_IS_ASYNC(task)) {
1067                 /* Wake up anyone who may be waiting for task completion */
1068                 if (!rpc_complete_task(task))
1069                         return;
1070         } else {
1071                 if (!atomic_dec_and_test(&task->tk_count))
1072                         return;
1073         }
1074         rpc_final_put_task(task, task->tk_workqueue);
1075 }
1076 
1077 int rpciod_up(void)
1078 {
1079         return try_module_get(THIS_MODULE) ? 0 : -EINVAL;
1080 }
1081 
1082 void rpciod_down(void)
1083 {
1084         module_put(THIS_MODULE);
1085 }
1086 
1087 /*
1088  * Start up the rpciod workqueue.
1089  */
1090 static int rpciod_start(void)
1091 {
1092         struct workqueue_struct *wq;
1093 
1094         /*
1095          * Create the rpciod thread and wait for it to start.
1096          */
1097         dprintk("RPC:       creating workqueue rpciod\n");
1098         wq = alloc_workqueue("rpciod", WQ_MEM_RECLAIM, 0);
1099         if (!wq)
1100                 goto out_failed;
1101         rpciod_workqueue = wq;
1102         /* Note: highpri because network receive is latency sensitive */
1103         wq = alloc_workqueue("xprtiod", WQ_MEM_RECLAIM | WQ_HIGHPRI, 0);
1104         if (!wq)
1105                 goto free_rpciod;
1106         xprtiod_workqueue = wq;
1107         return 1;
1108 free_rpciod:
1109         wq = rpciod_workqueue;
1110         rpciod_workqueue = NULL;
1111         destroy_workqueue(wq);
1112 out_failed:
1113         return 0;
1114 }
1115 
1116 static void rpciod_stop(void)
1117 {
1118         struct workqueue_struct *wq = NULL;
1119 
1120         if (rpciod_workqueue == NULL)
1121                 return;
1122         dprintk("RPC:       destroying workqueue rpciod\n");
1123 
1124         wq = rpciod_workqueue;
1125         rpciod_workqueue = NULL;
1126         destroy_workqueue(wq);
1127         wq = xprtiod_workqueue;
1128         xprtiod_workqueue = NULL;
1129         destroy_workqueue(wq);
1130 }
1131 
1132 void
1133 rpc_destroy_mempool(void)
1134 {
1135         rpciod_stop();
1136         mempool_destroy(rpc_buffer_mempool);
1137         mempool_destroy(rpc_task_mempool);
1138         kmem_cache_destroy(rpc_task_slabp);
1139         kmem_cache_destroy(rpc_buffer_slabp);
1140         rpc_destroy_wait_queue(&delay_queue);
1141 }
1142 
1143 int
1144 rpc_init_mempool(void)
1145 {
1146         /*
1147          * The following is not strictly a mempool initialisation,
1148          * but there is no harm in doing it here
1149          */
1150         rpc_init_wait_queue(&delay_queue, "delayq");
1151         if (!rpciod_start())
1152                 goto err_nomem;
1153 
1154         rpc_task_slabp = kmem_cache_create("rpc_tasks",
1155                                              sizeof(struct rpc_task),
1156                                              0, SLAB_HWCACHE_ALIGN,
1157                                              NULL);
1158         if (!rpc_task_slabp)
1159                 goto err_nomem;
1160         rpc_buffer_slabp = kmem_cache_create("rpc_buffers",
1161                                              RPC_BUFFER_MAXSIZE,
1162                                              0, SLAB_HWCACHE_ALIGN,
1163                                              NULL);
1164         if (!rpc_buffer_slabp)
1165                 goto err_nomem;
1166         rpc_task_mempool = mempool_create_slab_pool(RPC_TASK_POOLSIZE,
1167                                                     rpc_task_slabp);
1168         if (!rpc_task_mempool)
1169                 goto err_nomem;
1170         rpc_buffer_mempool = mempool_create_slab_pool(RPC_BUFFER_POOLSIZE,
1171                                                       rpc_buffer_slabp);
1172         if (!rpc_buffer_mempool)
1173                 goto err_nomem;
1174         return 0;
1175 err_nomem:
1176         rpc_destroy_mempool();
1177         return -ENOMEM;
1178 }
1179 

~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

kernel.org | git.kernel.org | LWN.net | Project Home | Wiki (Japanese) | Wiki (English) | SVN repository | Mail admin

Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.

osdn.jp