This source file includes following definitions.
- rpc_task_timeout
- __rpc_disable_timer
- rpc_set_queue_timer
- __rpc_add_timer
- rpc_set_waitqueue_priority
- rpc_reset_waitqueue_priority
- __rpc_list_enqueue_task
- __rpc_list_dequeue_task
- __rpc_add_wait_queue_priority
- __rpc_add_wait_queue
- __rpc_remove_wait_queue_priority
- __rpc_remove_wait_queue
- __rpc_init_priority_wait_queue
- rpc_init_priority_wait_queue
- rpc_init_wait_queue
- rpc_destroy_wait_queue
- rpc_wait_bit_killable
- rpc_task_set_debuginfo
- rpc_task_set_debuginfo
- rpc_set_active
- rpc_complete_task
- __rpc_wait_for_completion_task
- rpc_make_runnable
- __rpc_sleep_on_priority
- __rpc_sleep_on_priority_timeout
- rpc_set_tk_callback
- rpc_sleep_check_activated
- rpc_sleep_on_timeout
- rpc_sleep_on
- rpc_sleep_on_priority_timeout
- rpc_sleep_on_priority
- __rpc_do_wake_up_task_on_wq
- rpc_wake_up_task_on_wq_queue_action_locked
- rpc_wake_up_task_queue_locked
- rpc_wake_up_queued_task
- rpc_task_action_set_status
- rpc_wake_up_task_queue_set_status_locked
- rpc_wake_up_queued_task_set_status
- __rpc_find_next_queued_priority
- __rpc_find_next_queued
- rpc_wake_up_first_on_wq
- rpc_wake_up_first
- rpc_wake_up_next_func
- rpc_wake_up_next
- rpc_wake_up
- rpc_wake_up_status
- __rpc_queue_timer_fn
- __rpc_atrun
- rpc_delay
- rpc_prepare_task
- rpc_init_task_statistics
- rpc_reset_task_statistics
- rpc_exit_task
- rpc_signal_task
- rpc_exit
- rpc_release_calldata
- __rpc_execute
- rpc_execute
- rpc_async_schedule
- rpc_malloc
- rpc_free
- rpc_init_task
- rpc_alloc_task
- rpc_new_task
- rpc_free_task
- rpc_async_release
- rpc_release_resources_task
- rpc_final_put_task
- rpc_do_put_task
- rpc_put_task
- rpc_put_task_async
- rpc_release_task
- rpciod_up
- rpciod_down
- rpciod_start
- rpciod_stop
- rpc_destroy_mempool
- rpc_init_mempool
1
2
3
4
5
6
7
8
9
10
11
12
13 #include <linux/module.h>
14
15 #include <linux/sched.h>
16 #include <linux/interrupt.h>
17 #include <linux/slab.h>
18 #include <linux/mempool.h>
19 #include <linux/smp.h>
20 #include <linux/spinlock.h>
21 #include <linux/mutex.h>
22 #include <linux/freezer.h>
23 #include <linux/sched/mm.h>
24
25 #include <linux/sunrpc/clnt.h>
26 #include <linux/sunrpc/metrics.h>
27
28 #include "sunrpc.h"
29
30 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
31 #define RPCDBG_FACILITY RPCDBG_SCHED
32 #endif
33
34 #define CREATE_TRACE_POINTS
35 #include <trace/events/sunrpc.h>
36
37
38
39
40 #define RPC_BUFFER_MAXSIZE (2048)
41 #define RPC_BUFFER_POOLSIZE (8)
42 #define RPC_TASK_POOLSIZE (8)
43 static struct kmem_cache *rpc_task_slabp __read_mostly;
44 static struct kmem_cache *rpc_buffer_slabp __read_mostly;
45 static mempool_t *rpc_task_mempool __read_mostly;
46 static mempool_t *rpc_buffer_mempool __read_mostly;
47
48 static void rpc_async_schedule(struct work_struct *);
49 static void rpc_release_task(struct rpc_task *task);
50 static void __rpc_queue_timer_fn(struct work_struct *);
51
52
53
54
55 static struct rpc_wait_queue delay_queue;
56
57
58
59
60 struct workqueue_struct *rpciod_workqueue __read_mostly;
61 struct workqueue_struct *xprtiod_workqueue __read_mostly;
62 EXPORT_SYMBOL_GPL(xprtiod_workqueue);
63
64 unsigned long
65 rpc_task_timeout(const struct rpc_task *task)
66 {
67 unsigned long timeout = READ_ONCE(task->tk_timeout);
68
69 if (timeout != 0) {
70 unsigned long now = jiffies;
71 if (time_before(now, timeout))
72 return timeout - now;
73 }
74 return 0;
75 }
76 EXPORT_SYMBOL_GPL(rpc_task_timeout);
77
78
79
80
81
82
83 static void
84 __rpc_disable_timer(struct rpc_wait_queue *queue, struct rpc_task *task)
85 {
86 if (list_empty(&task->u.tk_wait.timer_list))
87 return;
88 dprintk("RPC: %5u disabling timer\n", task->tk_pid);
89 task->tk_timeout = 0;
90 list_del(&task->u.tk_wait.timer_list);
91 if (list_empty(&queue->timer_list.list))
92 cancel_delayed_work(&queue->timer_list.dwork);
93 }
94
95 static void
96 rpc_set_queue_timer(struct rpc_wait_queue *queue, unsigned long expires)
97 {
98 unsigned long now = jiffies;
99 queue->timer_list.expires = expires;
100 if (time_before_eq(expires, now))
101 expires = 0;
102 else
103 expires -= now;
104 mod_delayed_work(rpciod_workqueue, &queue->timer_list.dwork, expires);
105 }
106
107
108
109
110 static void
111 __rpc_add_timer(struct rpc_wait_queue *queue, struct rpc_task *task,
112 unsigned long timeout)
113 {
114 dprintk("RPC: %5u setting alarm for %u ms\n",
115 task->tk_pid, jiffies_to_msecs(timeout - jiffies));
116
117 task->tk_timeout = timeout;
118 if (list_empty(&queue->timer_list.list) || time_before(timeout, queue->timer_list.expires))
119 rpc_set_queue_timer(queue, timeout);
120 list_add(&task->u.tk_wait.timer_list, &queue->timer_list.list);
121 }
122
123 static void rpc_set_waitqueue_priority(struct rpc_wait_queue *queue, int priority)
124 {
125 if (queue->priority != priority) {
126 queue->priority = priority;
127 queue->nr = 1U << priority;
128 }
129 }
130
131 static void rpc_reset_waitqueue_priority(struct rpc_wait_queue *queue)
132 {
133 rpc_set_waitqueue_priority(queue, queue->maxpriority);
134 }
135
136
137
138
139 static void
140 __rpc_list_enqueue_task(struct list_head *q, struct rpc_task *task)
141 {
142 struct rpc_task *t;
143
144 list_for_each_entry(t, q, u.tk_wait.list) {
145 if (t->tk_owner == task->tk_owner) {
146 list_add_tail(&task->u.tk_wait.links,
147 &t->u.tk_wait.links);
148
149 task->u.tk_wait.list.next = q;
150 task->u.tk_wait.list.prev = NULL;
151 return;
152 }
153 }
154 INIT_LIST_HEAD(&task->u.tk_wait.links);
155 list_add_tail(&task->u.tk_wait.list, q);
156 }
157
158
159
160
161 static void
162 __rpc_list_dequeue_task(struct rpc_task *task)
163 {
164 struct list_head *q;
165 struct rpc_task *t;
166
167 if (task->u.tk_wait.list.prev == NULL) {
168 list_del(&task->u.tk_wait.links);
169 return;
170 }
171 if (!list_empty(&task->u.tk_wait.links)) {
172 t = list_first_entry(&task->u.tk_wait.links,
173 struct rpc_task,
174 u.tk_wait.links);
175
176 q = t->u.tk_wait.list.next;
177 list_add_tail(&t->u.tk_wait.list, q);
178 list_del(&task->u.tk_wait.links);
179 }
180 list_del(&task->u.tk_wait.list);
181 }
182
183
184
185
186 static void __rpc_add_wait_queue_priority(struct rpc_wait_queue *queue,
187 struct rpc_task *task,
188 unsigned char queue_priority)
189 {
190 if (unlikely(queue_priority > queue->maxpriority))
191 queue_priority = queue->maxpriority;
192 __rpc_list_enqueue_task(&queue->tasks[queue_priority], task);
193 }
194
195
196
197
198
199
200
201
202
203 static void __rpc_add_wait_queue(struct rpc_wait_queue *queue,
204 struct rpc_task *task,
205 unsigned char queue_priority)
206 {
207 WARN_ON_ONCE(RPC_IS_QUEUED(task));
208 if (RPC_IS_QUEUED(task))
209 return;
210
211 INIT_LIST_HEAD(&task->u.tk_wait.timer_list);
212 if (RPC_IS_PRIORITY(queue))
213 __rpc_add_wait_queue_priority(queue, task, queue_priority);
214 else if (RPC_IS_SWAPPER(task))
215 list_add(&task->u.tk_wait.list, &queue->tasks[0]);
216 else
217 list_add_tail(&task->u.tk_wait.list, &queue->tasks[0]);
218 task->tk_waitqueue = queue;
219 queue->qlen++;
220
221 smp_wmb();
222 rpc_set_queued(task);
223
224 dprintk("RPC: %5u added to queue %p \"%s\"\n",
225 task->tk_pid, queue, rpc_qname(queue));
226 }
227
228
229
230
231 static void __rpc_remove_wait_queue_priority(struct rpc_task *task)
232 {
233 __rpc_list_dequeue_task(task);
234 }
235
236
237
238
239
240 static void __rpc_remove_wait_queue(struct rpc_wait_queue *queue, struct rpc_task *task)
241 {
242 __rpc_disable_timer(queue, task);
243 if (RPC_IS_PRIORITY(queue))
244 __rpc_remove_wait_queue_priority(task);
245 else
246 list_del(&task->u.tk_wait.list);
247 queue->qlen--;
248 dprintk("RPC: %5u removed from queue %p \"%s\"\n",
249 task->tk_pid, queue, rpc_qname(queue));
250 }
251
252 static void __rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname, unsigned char nr_queues)
253 {
254 int i;
255
256 spin_lock_init(&queue->lock);
257 for (i = 0; i < ARRAY_SIZE(queue->tasks); i++)
258 INIT_LIST_HEAD(&queue->tasks[i]);
259 queue->maxpriority = nr_queues - 1;
260 rpc_reset_waitqueue_priority(queue);
261 queue->qlen = 0;
262 queue->timer_list.expires = 0;
263 INIT_DELAYED_WORK(&queue->timer_list.dwork, __rpc_queue_timer_fn);
264 INIT_LIST_HEAD(&queue->timer_list.list);
265 rpc_assign_waitqueue_name(queue, qname);
266 }
267
268 void rpc_init_priority_wait_queue(struct rpc_wait_queue *queue, const char *qname)
269 {
270 __rpc_init_priority_wait_queue(queue, qname, RPC_NR_PRIORITY);
271 }
272 EXPORT_SYMBOL_GPL(rpc_init_priority_wait_queue);
273
274 void rpc_init_wait_queue(struct rpc_wait_queue *queue, const char *qname)
275 {
276 __rpc_init_priority_wait_queue(queue, qname, 1);
277 }
278 EXPORT_SYMBOL_GPL(rpc_init_wait_queue);
279
280 void rpc_destroy_wait_queue(struct rpc_wait_queue *queue)
281 {
282 cancel_delayed_work_sync(&queue->timer_list.dwork);
283 }
284 EXPORT_SYMBOL_GPL(rpc_destroy_wait_queue);
285
286 static int rpc_wait_bit_killable(struct wait_bit_key *key, int mode)
287 {
288 freezable_schedule_unsafe();
289 if (signal_pending_state(mode, current))
290 return -ERESTARTSYS;
291 return 0;
292 }
293
294 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) || IS_ENABLED(CONFIG_TRACEPOINTS)
295 static void rpc_task_set_debuginfo(struct rpc_task *task)
296 {
297 static atomic_t rpc_pid;
298
299 task->tk_pid = atomic_inc_return(&rpc_pid);
300 }
301 #else
302 static inline void rpc_task_set_debuginfo(struct rpc_task *task)
303 {
304 }
305 #endif
306
307 static void rpc_set_active(struct rpc_task *task)
308 {
309 rpc_task_set_debuginfo(task);
310 set_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
311 trace_rpc_task_begin(task, NULL);
312 }
313
314
315
316
317
318 static int rpc_complete_task(struct rpc_task *task)
319 {
320 void *m = &task->tk_runstate;
321 wait_queue_head_t *wq = bit_waitqueue(m, RPC_TASK_ACTIVE);
322 struct wait_bit_key k = __WAIT_BIT_KEY_INITIALIZER(m, RPC_TASK_ACTIVE);
323 unsigned long flags;
324 int ret;
325
326 trace_rpc_task_complete(task, NULL);
327
328 spin_lock_irqsave(&wq->lock, flags);
329 clear_bit(RPC_TASK_ACTIVE, &task->tk_runstate);
330 ret = atomic_dec_and_test(&task->tk_count);
331 if (waitqueue_active(wq))
332 __wake_up_locked_key(wq, TASK_NORMAL, &k);
333 spin_unlock_irqrestore(&wq->lock, flags);
334 return ret;
335 }
336
337
338
339
340
341
342
343
344 int __rpc_wait_for_completion_task(struct rpc_task *task, wait_bit_action_f *action)
345 {
346 if (action == NULL)
347 action = rpc_wait_bit_killable;
348 return out_of_line_wait_on_bit(&task->tk_runstate, RPC_TASK_ACTIVE,
349 action, TASK_KILLABLE);
350 }
351 EXPORT_SYMBOL_GPL(__rpc_wait_for_completion_task);
352
353
354
355
356
357
358
359
360
361
362
363
364 static void rpc_make_runnable(struct workqueue_struct *wq,
365 struct rpc_task *task)
366 {
367 bool need_wakeup = !rpc_test_and_set_running(task);
368
369 rpc_clear_queued(task);
370 if (!need_wakeup)
371 return;
372 if (RPC_IS_ASYNC(task)) {
373 INIT_WORK(&task->u.tk_work, rpc_async_schedule);
374 queue_work(wq, &task->u.tk_work);
375 } else
376 wake_up_bit(&task->tk_runstate, RPC_TASK_QUEUED);
377 }
378
379
380
381
382
383
384
385 static void __rpc_sleep_on_priority(struct rpc_wait_queue *q,
386 struct rpc_task *task,
387 unsigned char queue_priority)
388 {
389 dprintk("RPC: %5u sleep_on(queue \"%s\" time %lu)\n",
390 task->tk_pid, rpc_qname(q), jiffies);
391
392 trace_rpc_task_sleep(task, q);
393
394 __rpc_add_wait_queue(q, task, queue_priority);
395
396 }
397
398 static void __rpc_sleep_on_priority_timeout(struct rpc_wait_queue *q,
399 struct rpc_task *task, unsigned long timeout,
400 unsigned char queue_priority)
401 {
402 if (time_is_after_jiffies(timeout)) {
403 __rpc_sleep_on_priority(q, task, queue_priority);
404 __rpc_add_timer(q, task, timeout);
405 } else
406 task->tk_status = -ETIMEDOUT;
407 }
408
409 static void rpc_set_tk_callback(struct rpc_task *task, rpc_action action)
410 {
411 if (action && !WARN_ON_ONCE(task->tk_callback != NULL))
412 task->tk_callback = action;
413 }
414
415 static bool rpc_sleep_check_activated(struct rpc_task *task)
416 {
417
418 if (WARN_ON_ONCE(!RPC_IS_ACTIVATED(task))) {
419 task->tk_status = -EIO;
420 rpc_put_task_async(task);
421 return false;
422 }
423 return true;
424 }
425
426 void rpc_sleep_on_timeout(struct rpc_wait_queue *q, struct rpc_task *task,
427 rpc_action action, unsigned long timeout)
428 {
429 if (!rpc_sleep_check_activated(task))
430 return;
431
432 rpc_set_tk_callback(task, action);
433
434
435
436
437 spin_lock(&q->lock);
438 __rpc_sleep_on_priority_timeout(q, task, timeout, task->tk_priority);
439 spin_unlock(&q->lock);
440 }
441 EXPORT_SYMBOL_GPL(rpc_sleep_on_timeout);
442
443 void rpc_sleep_on(struct rpc_wait_queue *q, struct rpc_task *task,
444 rpc_action action)
445 {
446 if (!rpc_sleep_check_activated(task))
447 return;
448
449 rpc_set_tk_callback(task, action);
450
451 WARN_ON_ONCE(task->tk_timeout != 0);
452
453
454
455 spin_lock(&q->lock);
456 __rpc_sleep_on_priority(q, task, task->tk_priority);
457 spin_unlock(&q->lock);
458 }
459 EXPORT_SYMBOL_GPL(rpc_sleep_on);
460
461 void rpc_sleep_on_priority_timeout(struct rpc_wait_queue *q,
462 struct rpc_task *task, unsigned long timeout, int priority)
463 {
464 if (!rpc_sleep_check_activated(task))
465 return;
466
467 priority -= RPC_PRIORITY_LOW;
468
469
470
471 spin_lock(&q->lock);
472 __rpc_sleep_on_priority_timeout(q, task, timeout, priority);
473 spin_unlock(&q->lock);
474 }
475 EXPORT_SYMBOL_GPL(rpc_sleep_on_priority_timeout);
476
477 void rpc_sleep_on_priority(struct rpc_wait_queue *q, struct rpc_task *task,
478 int priority)
479 {
480 if (!rpc_sleep_check_activated(task))
481 return;
482
483 WARN_ON_ONCE(task->tk_timeout != 0);
484 priority -= RPC_PRIORITY_LOW;
485
486
487
488 spin_lock(&q->lock);
489 __rpc_sleep_on_priority(q, task, priority);
490 spin_unlock(&q->lock);
491 }
492 EXPORT_SYMBOL_GPL(rpc_sleep_on_priority);
493
494
495
496
497
498
499
500
501
502 static void __rpc_do_wake_up_task_on_wq(struct workqueue_struct *wq,
503 struct rpc_wait_queue *queue,
504 struct rpc_task *task)
505 {
506 dprintk("RPC: %5u __rpc_wake_up_task (now %lu)\n",
507 task->tk_pid, jiffies);
508
509
510 if (!RPC_IS_ACTIVATED(task)) {
511 printk(KERN_ERR "RPC: Inactive task (%p) being woken up!\n", task);
512 return;
513 }
514
515 trace_rpc_task_wakeup(task, queue);
516
517 __rpc_remove_wait_queue(queue, task);
518
519 rpc_make_runnable(wq, task);
520
521 dprintk("RPC: __rpc_wake_up_task done\n");
522 }
523
524
525
526
527 static struct rpc_task *
528 rpc_wake_up_task_on_wq_queue_action_locked(struct workqueue_struct *wq,
529 struct rpc_wait_queue *queue, struct rpc_task *task,
530 bool (*action)(struct rpc_task *, void *), void *data)
531 {
532 if (RPC_IS_QUEUED(task)) {
533 smp_rmb();
534 if (task->tk_waitqueue == queue) {
535 if (action == NULL || action(task, data)) {
536 __rpc_do_wake_up_task_on_wq(wq, queue, task);
537 return task;
538 }
539 }
540 }
541 return NULL;
542 }
543
544
545
546
547 static void rpc_wake_up_task_queue_locked(struct rpc_wait_queue *queue,
548 struct rpc_task *task)
549 {
550 rpc_wake_up_task_on_wq_queue_action_locked(rpciod_workqueue, queue,
551 task, NULL, NULL);
552 }
553
554
555
556
557 void rpc_wake_up_queued_task(struct rpc_wait_queue *queue, struct rpc_task *task)
558 {
559 if (!RPC_IS_QUEUED(task))
560 return;
561 spin_lock(&queue->lock);
562 rpc_wake_up_task_queue_locked(queue, task);
563 spin_unlock(&queue->lock);
564 }
565 EXPORT_SYMBOL_GPL(rpc_wake_up_queued_task);
566
567 static bool rpc_task_action_set_status(struct rpc_task *task, void *status)
568 {
569 task->tk_status = *(int *)status;
570 return true;
571 }
572
573 static void
574 rpc_wake_up_task_queue_set_status_locked(struct rpc_wait_queue *queue,
575 struct rpc_task *task, int status)
576 {
577 rpc_wake_up_task_on_wq_queue_action_locked(rpciod_workqueue, queue,
578 task, rpc_task_action_set_status, &status);
579 }
580
581
582
583
584
585
586
587
588
589
590 void
591 rpc_wake_up_queued_task_set_status(struct rpc_wait_queue *queue,
592 struct rpc_task *task, int status)
593 {
594 if (!RPC_IS_QUEUED(task))
595 return;
596 spin_lock(&queue->lock);
597 rpc_wake_up_task_queue_set_status_locked(queue, task, status);
598 spin_unlock(&queue->lock);
599 }
600
601
602
603
604 static struct rpc_task *__rpc_find_next_queued_priority(struct rpc_wait_queue *queue)
605 {
606 struct list_head *q;
607 struct rpc_task *task;
608
609
610
611
612 q = &queue->tasks[queue->priority];
613 if (!list_empty(q) && --queue->nr) {
614 task = list_first_entry(q, struct rpc_task, u.tk_wait.list);
615 goto out;
616 }
617
618
619
620
621 do {
622 if (q == &queue->tasks[0])
623 q = &queue->tasks[queue->maxpriority];
624 else
625 q = q - 1;
626 if (!list_empty(q)) {
627 task = list_first_entry(q, struct rpc_task, u.tk_wait.list);
628 goto new_queue;
629 }
630 } while (q != &queue->tasks[queue->priority]);
631
632 rpc_reset_waitqueue_priority(queue);
633 return NULL;
634
635 new_queue:
636 rpc_set_waitqueue_priority(queue, (unsigned int)(q - &queue->tasks[0]));
637 out:
638 return task;
639 }
640
641 static struct rpc_task *__rpc_find_next_queued(struct rpc_wait_queue *queue)
642 {
643 if (RPC_IS_PRIORITY(queue))
644 return __rpc_find_next_queued_priority(queue);
645 if (!list_empty(&queue->tasks[0]))
646 return list_first_entry(&queue->tasks[0], struct rpc_task, u.tk_wait.list);
647 return NULL;
648 }
649
650
651
652
653 struct rpc_task *rpc_wake_up_first_on_wq(struct workqueue_struct *wq,
654 struct rpc_wait_queue *queue,
655 bool (*func)(struct rpc_task *, void *), void *data)
656 {
657 struct rpc_task *task = NULL;
658
659 dprintk("RPC: wake_up_first(%p \"%s\")\n",
660 queue, rpc_qname(queue));
661 spin_lock(&queue->lock);
662 task = __rpc_find_next_queued(queue);
663 if (task != NULL)
664 task = rpc_wake_up_task_on_wq_queue_action_locked(wq, queue,
665 task, func, data);
666 spin_unlock(&queue->lock);
667
668 return task;
669 }
670
671
672
673
674 struct rpc_task *rpc_wake_up_first(struct rpc_wait_queue *queue,
675 bool (*func)(struct rpc_task *, void *), void *data)
676 {
677 return rpc_wake_up_first_on_wq(rpciod_workqueue, queue, func, data);
678 }
679 EXPORT_SYMBOL_GPL(rpc_wake_up_first);
680
681 static bool rpc_wake_up_next_func(struct rpc_task *task, void *data)
682 {
683 return true;
684 }
685
686
687
688
689 struct rpc_task *rpc_wake_up_next(struct rpc_wait_queue *queue)
690 {
691 return rpc_wake_up_first(queue, rpc_wake_up_next_func, NULL);
692 }
693 EXPORT_SYMBOL_GPL(rpc_wake_up_next);
694
695
696
697
698
699
700
701 void rpc_wake_up(struct rpc_wait_queue *queue)
702 {
703 struct list_head *head;
704
705 spin_lock(&queue->lock);
706 head = &queue->tasks[queue->maxpriority];
707 for (;;) {
708 while (!list_empty(head)) {
709 struct rpc_task *task;
710 task = list_first_entry(head,
711 struct rpc_task,
712 u.tk_wait.list);
713 rpc_wake_up_task_queue_locked(queue, task);
714 }
715 if (head == &queue->tasks[0])
716 break;
717 head--;
718 }
719 spin_unlock(&queue->lock);
720 }
721 EXPORT_SYMBOL_GPL(rpc_wake_up);
722
723
724
725
726
727
728
729
730 void rpc_wake_up_status(struct rpc_wait_queue *queue, int status)
731 {
732 struct list_head *head;
733
734 spin_lock(&queue->lock);
735 head = &queue->tasks[queue->maxpriority];
736 for (;;) {
737 while (!list_empty(head)) {
738 struct rpc_task *task;
739 task = list_first_entry(head,
740 struct rpc_task,
741 u.tk_wait.list);
742 task->tk_status = status;
743 rpc_wake_up_task_queue_locked(queue, task);
744 }
745 if (head == &queue->tasks[0])
746 break;
747 head--;
748 }
749 spin_unlock(&queue->lock);
750 }
751 EXPORT_SYMBOL_GPL(rpc_wake_up_status);
752
753 static void __rpc_queue_timer_fn(struct work_struct *work)
754 {
755 struct rpc_wait_queue *queue = container_of(work,
756 struct rpc_wait_queue,
757 timer_list.dwork.work);
758 struct rpc_task *task, *n;
759 unsigned long expires, now, timeo;
760
761 spin_lock(&queue->lock);
762 expires = now = jiffies;
763 list_for_each_entry_safe(task, n, &queue->timer_list.list, u.tk_wait.timer_list) {
764 timeo = task->tk_timeout;
765 if (time_after_eq(now, timeo)) {
766 dprintk("RPC: %5u timeout\n", task->tk_pid);
767 task->tk_status = -ETIMEDOUT;
768 rpc_wake_up_task_queue_locked(queue, task);
769 continue;
770 }
771 if (expires == now || time_after(expires, timeo))
772 expires = timeo;
773 }
774 if (!list_empty(&queue->timer_list.list))
775 rpc_set_queue_timer(queue, expires);
776 spin_unlock(&queue->lock);
777 }
778
779 static void __rpc_atrun(struct rpc_task *task)
780 {
781 if (task->tk_status == -ETIMEDOUT)
782 task->tk_status = 0;
783 }
784
785
786
787
788 void rpc_delay(struct rpc_task *task, unsigned long delay)
789 {
790 rpc_sleep_on_timeout(&delay_queue, task, __rpc_atrun, jiffies + delay);
791 }
792 EXPORT_SYMBOL_GPL(rpc_delay);
793
794
795
796
797 void rpc_prepare_task(struct rpc_task *task)
798 {
799 task->tk_ops->rpc_call_prepare(task, task->tk_calldata);
800 }
801
802 static void
803 rpc_init_task_statistics(struct rpc_task *task)
804 {
805
806 task->tk_garb_retry = 2;
807 task->tk_cred_retry = 2;
808 task->tk_rebind_retry = 2;
809
810
811 task->tk_start = ktime_get();
812 }
813
814 static void
815 rpc_reset_task_statistics(struct rpc_task *task)
816 {
817 task->tk_timeouts = 0;
818 task->tk_flags &= ~(RPC_CALL_MAJORSEEN|RPC_TASK_SENT);
819 rpc_init_task_statistics(task);
820 }
821
822
823
824
825 void rpc_exit_task(struct rpc_task *task)
826 {
827 task->tk_action = NULL;
828 if (task->tk_ops->rpc_count_stats)
829 task->tk_ops->rpc_count_stats(task, task->tk_calldata);
830 else if (task->tk_client)
831 rpc_count_iostats(task, task->tk_client->cl_metrics);
832 if (task->tk_ops->rpc_call_done != NULL) {
833 task->tk_ops->rpc_call_done(task, task->tk_calldata);
834 if (task->tk_action != NULL) {
835
836 xprt_release(task);
837 rpc_reset_task_statistics(task);
838 }
839 }
840 }
841
842 void rpc_signal_task(struct rpc_task *task)
843 {
844 struct rpc_wait_queue *queue;
845
846 if (!RPC_IS_ACTIVATED(task))
847 return;
848 set_bit(RPC_TASK_SIGNALLED, &task->tk_runstate);
849 smp_mb__after_atomic();
850 queue = READ_ONCE(task->tk_waitqueue);
851 if (queue)
852 rpc_wake_up_queued_task_set_status(queue, task, -ERESTARTSYS);
853 }
854
855 void rpc_exit(struct rpc_task *task, int status)
856 {
857 task->tk_status = status;
858 task->tk_action = rpc_exit_task;
859 rpc_wake_up_queued_task(task->tk_waitqueue, task);
860 }
861 EXPORT_SYMBOL_GPL(rpc_exit);
862
863 void rpc_release_calldata(const struct rpc_call_ops *ops, void *calldata)
864 {
865 if (ops->rpc_release != NULL)
866 ops->rpc_release(calldata);
867 }
868
869
870
871
872 static void __rpc_execute(struct rpc_task *task)
873 {
874 struct rpc_wait_queue *queue;
875 int task_is_async = RPC_IS_ASYNC(task);
876 int status = 0;
877
878 dprintk("RPC: %5u __rpc_execute flags=0x%x\n",
879 task->tk_pid, task->tk_flags);
880
881 WARN_ON_ONCE(RPC_IS_QUEUED(task));
882 if (RPC_IS_QUEUED(task))
883 return;
884
885 for (;;) {
886 void (*do_action)(struct rpc_task *);
887
888
889
890
891
892
893
894
895 do_action = task->tk_action;
896 if (task->tk_callback) {
897 do_action = task->tk_callback;
898 task->tk_callback = NULL;
899 }
900 if (!do_action)
901 break;
902 trace_rpc_task_run_action(task, do_action);
903 do_action(task);
904
905
906
907
908 if (!RPC_IS_QUEUED(task))
909 continue;
910
911
912
913
914 if (RPC_SIGNALLED(task)) {
915 task->tk_rpc_status = -ERESTARTSYS;
916 rpc_exit(task, -ERESTARTSYS);
917 }
918
919
920
921
922
923
924
925
926
927
928 queue = task->tk_waitqueue;
929 spin_lock(&queue->lock);
930 if (!RPC_IS_QUEUED(task)) {
931 spin_unlock(&queue->lock);
932 continue;
933 }
934 rpc_clear_running(task);
935 spin_unlock(&queue->lock);
936 if (task_is_async)
937 return;
938
939
940 dprintk("RPC: %5u sync task going to sleep\n", task->tk_pid);
941 status = out_of_line_wait_on_bit(&task->tk_runstate,
942 RPC_TASK_QUEUED, rpc_wait_bit_killable,
943 TASK_KILLABLE);
944 if (status < 0) {
945
946
947
948
949
950
951 dprintk("RPC: %5u got signal\n", task->tk_pid);
952 set_bit(RPC_TASK_SIGNALLED, &task->tk_runstate);
953 task->tk_rpc_status = -ERESTARTSYS;
954 rpc_exit(task, -ERESTARTSYS);
955 }
956 dprintk("RPC: %5u sync task resuming\n", task->tk_pid);
957 }
958
959 dprintk("RPC: %5u return %d, status %d\n", task->tk_pid, status,
960 task->tk_status);
961
962 rpc_release_task(task);
963 }
964
965
966
967
968
969
970
971
972
973
974 void rpc_execute(struct rpc_task *task)
975 {
976 bool is_async = RPC_IS_ASYNC(task);
977
978 rpc_set_active(task);
979 rpc_make_runnable(rpciod_workqueue, task);
980 if (!is_async)
981 __rpc_execute(task);
982 }
983
984 static void rpc_async_schedule(struct work_struct *work)
985 {
986 unsigned int pflags = memalloc_nofs_save();
987
988 __rpc_execute(container_of(work, struct rpc_task, u.tk_work));
989 memalloc_nofs_restore(pflags);
990 }
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009 int rpc_malloc(struct rpc_task *task)
1010 {
1011 struct rpc_rqst *rqst = task->tk_rqstp;
1012 size_t size = rqst->rq_callsize + rqst->rq_rcvsize;
1013 struct rpc_buffer *buf;
1014 gfp_t gfp = GFP_NOFS;
1015
1016 if (RPC_IS_SWAPPER(task))
1017 gfp = __GFP_MEMALLOC | GFP_NOWAIT | __GFP_NOWARN;
1018
1019 size += sizeof(struct rpc_buffer);
1020 if (size <= RPC_BUFFER_MAXSIZE)
1021 buf = mempool_alloc(rpc_buffer_mempool, gfp);
1022 else
1023 buf = kmalloc(size, gfp);
1024
1025 if (!buf)
1026 return -ENOMEM;
1027
1028 buf->len = size;
1029 dprintk("RPC: %5u allocated buffer of size %zu at %p\n",
1030 task->tk_pid, size, buf);
1031 rqst->rq_buffer = buf->data;
1032 rqst->rq_rbuffer = (char *)rqst->rq_buffer + rqst->rq_callsize;
1033 return 0;
1034 }
1035 EXPORT_SYMBOL_GPL(rpc_malloc);
1036
1037
1038
1039
1040
1041
1042 void rpc_free(struct rpc_task *task)
1043 {
1044 void *buffer = task->tk_rqstp->rq_buffer;
1045 size_t size;
1046 struct rpc_buffer *buf;
1047
1048 buf = container_of(buffer, struct rpc_buffer, data);
1049 size = buf->len;
1050
1051 dprintk("RPC: freeing buffer of size %zu at %p\n",
1052 size, buf);
1053
1054 if (size <= RPC_BUFFER_MAXSIZE)
1055 mempool_free(buf, rpc_buffer_mempool);
1056 else
1057 kfree(buf);
1058 }
1059 EXPORT_SYMBOL_GPL(rpc_free);
1060
1061
1062
1063
1064 static void rpc_init_task(struct rpc_task *task, const struct rpc_task_setup *task_setup_data)
1065 {
1066 memset(task, 0, sizeof(*task));
1067 atomic_set(&task->tk_count, 1);
1068 task->tk_flags = task_setup_data->flags;
1069 task->tk_ops = task_setup_data->callback_ops;
1070 task->tk_calldata = task_setup_data->callback_data;
1071 INIT_LIST_HEAD(&task->tk_task);
1072
1073 task->tk_priority = task_setup_data->priority - RPC_PRIORITY_LOW;
1074 task->tk_owner = current->tgid;
1075
1076
1077 task->tk_workqueue = task_setup_data->workqueue;
1078
1079 task->tk_xprt = rpc_task_get_xprt(task_setup_data->rpc_client,
1080 xprt_get(task_setup_data->rpc_xprt));
1081
1082 task->tk_op_cred = get_rpccred(task_setup_data->rpc_op_cred);
1083
1084 if (task->tk_ops->rpc_call_prepare != NULL)
1085 task->tk_action = rpc_prepare_task;
1086
1087 rpc_init_task_statistics(task);
1088
1089 dprintk("RPC: new task initialized, procpid %u\n",
1090 task_pid_nr(current));
1091 }
1092
1093 static struct rpc_task *
1094 rpc_alloc_task(void)
1095 {
1096 return (struct rpc_task *)mempool_alloc(rpc_task_mempool, GFP_NOFS);
1097 }
1098
1099
1100
1101
1102 struct rpc_task *rpc_new_task(const struct rpc_task_setup *setup_data)
1103 {
1104 struct rpc_task *task = setup_data->task;
1105 unsigned short flags = 0;
1106
1107 if (task == NULL) {
1108 task = rpc_alloc_task();
1109 flags = RPC_TASK_DYNAMIC;
1110 }
1111
1112 rpc_init_task(task, setup_data);
1113 task->tk_flags |= flags;
1114 dprintk("RPC: allocated task %p\n", task);
1115 return task;
1116 }
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137 static void rpc_free_task(struct rpc_task *task)
1138 {
1139 unsigned short tk_flags = task->tk_flags;
1140
1141 put_rpccred(task->tk_op_cred);
1142 rpc_release_calldata(task->tk_ops, task->tk_calldata);
1143
1144 if (tk_flags & RPC_TASK_DYNAMIC) {
1145 dprintk("RPC: %5u freeing task\n", task->tk_pid);
1146 mempool_free(task, rpc_task_mempool);
1147 }
1148 }
1149
1150 static void rpc_async_release(struct work_struct *work)
1151 {
1152 unsigned int pflags = memalloc_nofs_save();
1153
1154 rpc_free_task(container_of(work, struct rpc_task, u.tk_work));
1155 memalloc_nofs_restore(pflags);
1156 }
1157
1158 static void rpc_release_resources_task(struct rpc_task *task)
1159 {
1160 xprt_release(task);
1161 if (task->tk_msg.rpc_cred) {
1162 put_cred(task->tk_msg.rpc_cred);
1163 task->tk_msg.rpc_cred = NULL;
1164 }
1165 rpc_task_release_client(task);
1166 }
1167
1168 static void rpc_final_put_task(struct rpc_task *task,
1169 struct workqueue_struct *q)
1170 {
1171 if (q != NULL) {
1172 INIT_WORK(&task->u.tk_work, rpc_async_release);
1173 queue_work(q, &task->u.tk_work);
1174 } else
1175 rpc_free_task(task);
1176 }
1177
1178 static void rpc_do_put_task(struct rpc_task *task, struct workqueue_struct *q)
1179 {
1180 if (atomic_dec_and_test(&task->tk_count)) {
1181 rpc_release_resources_task(task);
1182 rpc_final_put_task(task, q);
1183 }
1184 }
1185
1186 void rpc_put_task(struct rpc_task *task)
1187 {
1188 rpc_do_put_task(task, NULL);
1189 }
1190 EXPORT_SYMBOL_GPL(rpc_put_task);
1191
1192 void rpc_put_task_async(struct rpc_task *task)
1193 {
1194 rpc_do_put_task(task, task->tk_workqueue);
1195 }
1196 EXPORT_SYMBOL_GPL(rpc_put_task_async);
1197
1198 static void rpc_release_task(struct rpc_task *task)
1199 {
1200 dprintk("RPC: %5u release task\n", task->tk_pid);
1201
1202 WARN_ON_ONCE(RPC_IS_QUEUED(task));
1203
1204 rpc_release_resources_task(task);
1205
1206
1207
1208
1209
1210
1211 if (atomic_read(&task->tk_count) != 1 + !RPC_IS_ASYNC(task)) {
1212
1213 if (!rpc_complete_task(task))
1214 return;
1215 } else {
1216 if (!atomic_dec_and_test(&task->tk_count))
1217 return;
1218 }
1219 rpc_final_put_task(task, task->tk_workqueue);
1220 }
1221
1222 int rpciod_up(void)
1223 {
1224 return try_module_get(THIS_MODULE) ? 0 : -EINVAL;
1225 }
1226
1227 void rpciod_down(void)
1228 {
1229 module_put(THIS_MODULE);
1230 }
1231
1232
1233
1234
1235 static int rpciod_start(void)
1236 {
1237 struct workqueue_struct *wq;
1238
1239
1240
1241
1242 dprintk("RPC: creating workqueue rpciod\n");
1243 wq = alloc_workqueue("rpciod", WQ_MEM_RECLAIM | WQ_UNBOUND, 0);
1244 if (!wq)
1245 goto out_failed;
1246 rpciod_workqueue = wq;
1247
1248 wq = alloc_workqueue("xprtiod", WQ_UNBOUND|WQ_MEM_RECLAIM|WQ_HIGHPRI, 0);
1249 if (!wq)
1250 goto free_rpciod;
1251 xprtiod_workqueue = wq;
1252 return 1;
1253 free_rpciod:
1254 wq = rpciod_workqueue;
1255 rpciod_workqueue = NULL;
1256 destroy_workqueue(wq);
1257 out_failed:
1258 return 0;
1259 }
1260
1261 static void rpciod_stop(void)
1262 {
1263 struct workqueue_struct *wq = NULL;
1264
1265 if (rpciod_workqueue == NULL)
1266 return;
1267 dprintk("RPC: destroying workqueue rpciod\n");
1268
1269 wq = rpciod_workqueue;
1270 rpciod_workqueue = NULL;
1271 destroy_workqueue(wq);
1272 wq = xprtiod_workqueue;
1273 xprtiod_workqueue = NULL;
1274 destroy_workqueue(wq);
1275 }
1276
1277 void
1278 rpc_destroy_mempool(void)
1279 {
1280 rpciod_stop();
1281 mempool_destroy(rpc_buffer_mempool);
1282 mempool_destroy(rpc_task_mempool);
1283 kmem_cache_destroy(rpc_task_slabp);
1284 kmem_cache_destroy(rpc_buffer_slabp);
1285 rpc_destroy_wait_queue(&delay_queue);
1286 }
1287
1288 int
1289 rpc_init_mempool(void)
1290 {
1291
1292
1293
1294
1295 rpc_init_wait_queue(&delay_queue, "delayq");
1296 if (!rpciod_start())
1297 goto err_nomem;
1298
1299 rpc_task_slabp = kmem_cache_create("rpc_tasks",
1300 sizeof(struct rpc_task),
1301 0, SLAB_HWCACHE_ALIGN,
1302 NULL);
1303 if (!rpc_task_slabp)
1304 goto err_nomem;
1305 rpc_buffer_slabp = kmem_cache_create("rpc_buffers",
1306 RPC_BUFFER_MAXSIZE,
1307 0, SLAB_HWCACHE_ALIGN,
1308 NULL);
1309 if (!rpc_buffer_slabp)
1310 goto err_nomem;
1311 rpc_task_mempool = mempool_create_slab_pool(RPC_TASK_POOLSIZE,
1312 rpc_task_slabp);
1313 if (!rpc_task_mempool)
1314 goto err_nomem;
1315 rpc_buffer_mempool = mempool_create_slab_pool(RPC_BUFFER_POOLSIZE,
1316 rpc_buffer_slabp);
1317 if (!rpc_buffer_mempool)
1318 goto err_nomem;
1319 return 0;
1320 err_nomem:
1321 rpc_destroy_mempool();
1322 return -ENOMEM;
1323 }