1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40#include <linux/module.h>
41
42#include <linux/types.h>
43#include <linux/interrupt.h>
44#include <linux/workqueue.h>
45#include <linux/net.h>
46#include <linux/ktime.h>
47
48#include <linux/sunrpc/clnt.h>
49#include <linux/sunrpc/metrics.h>
50#include <linux/sunrpc/bc_xprt.h>
51#include <linux/rcupdate.h>
52
53#include <trace/events/sunrpc.h>
54
55#include "sunrpc.h"
56
57
58
59
60
61#if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
62# define RPCDBG_FACILITY RPCDBG_XPRT
63#endif
64
65
66
67
68static void xprt_init(struct rpc_xprt *xprt, struct net *net);
69static void xprt_request_init(struct rpc_task *, struct rpc_xprt *);
70static void xprt_connect_status(struct rpc_task *task);
71static int __xprt_get_cong(struct rpc_xprt *, struct rpc_task *);
72static void __xprt_put_cong(struct rpc_xprt *, struct rpc_rqst *);
73static void xprt_destroy(struct rpc_xprt *xprt);
74
75static DEFINE_SPINLOCK(xprt_list_lock);
76static LIST_HEAD(xprt_list);
77
78
79
80
81
82
83
84
85
86
87
88
89
90int xprt_register_transport(struct xprt_class *transport)
91{
92 struct xprt_class *t;
93 int result;
94
95 result = -EEXIST;
96 spin_lock(&xprt_list_lock);
97 list_for_each_entry(t, &xprt_list, list) {
98
99 if (t->ident == transport->ident)
100 goto out;
101 }
102
103 list_add_tail(&transport->list, &xprt_list);
104 printk(KERN_INFO "RPC: Registered %s transport module.\n",
105 transport->name);
106 result = 0;
107
108out:
109 spin_unlock(&xprt_list_lock);
110 return result;
111}
112EXPORT_SYMBOL_GPL(xprt_register_transport);
113
114
115
116
117
118
119
120
121
122int xprt_unregister_transport(struct xprt_class *transport)
123{
124 struct xprt_class *t;
125 int result;
126
127 result = 0;
128 spin_lock(&xprt_list_lock);
129 list_for_each_entry(t, &xprt_list, list) {
130 if (t == transport) {
131 printk(KERN_INFO
132 "RPC: Unregistered %s transport module.\n",
133 transport->name);
134 list_del_init(&transport->list);
135 goto out;
136 }
137 }
138 result = -ENOENT;
139
140out:
141 spin_unlock(&xprt_list_lock);
142 return result;
143}
144EXPORT_SYMBOL_GPL(xprt_unregister_transport);
145
146
147
148
149
150
151
152
153
154int xprt_load_transport(const char *transport_name)
155{
156 struct xprt_class *t;
157 int result;
158
159 result = 0;
160 spin_lock(&xprt_list_lock);
161 list_for_each_entry(t, &xprt_list, list) {
162 if (strcmp(t->name, transport_name) == 0) {
163 spin_unlock(&xprt_list_lock);
164 goto out;
165 }
166 }
167 spin_unlock(&xprt_list_lock);
168 result = request_module("xprt%s", transport_name);
169out:
170 return result;
171}
172EXPORT_SYMBOL_GPL(xprt_load_transport);
173
174
175
176
177
178
179
180
181
182
183int xprt_reserve_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
184{
185 struct rpc_rqst *req = task->tk_rqstp;
186 int priority;
187
188 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
189 if (task == xprt->snd_task)
190 return 1;
191 goto out_sleep;
192 }
193 xprt->snd_task = task;
194 if (req != NULL)
195 req->rq_ntrans++;
196
197 return 1;
198
199out_sleep:
200 dprintk("RPC: %5u failed to lock transport %p\n",
201 task->tk_pid, xprt);
202 task->tk_timeout = 0;
203 task->tk_status = -EAGAIN;
204 if (req == NULL)
205 priority = RPC_PRIORITY_LOW;
206 else if (!req->rq_ntrans)
207 priority = RPC_PRIORITY_NORMAL;
208 else
209 priority = RPC_PRIORITY_HIGH;
210 rpc_sleep_on_priority(&xprt->sending, task, NULL, priority);
211 return 0;
212}
213EXPORT_SYMBOL_GPL(xprt_reserve_xprt);
214
215static void xprt_clear_locked(struct rpc_xprt *xprt)
216{
217 xprt->snd_task = NULL;
218 if (!test_bit(XPRT_CLOSE_WAIT, &xprt->state)) {
219 smp_mb__before_atomic();
220 clear_bit(XPRT_LOCKED, &xprt->state);
221 smp_mb__after_atomic();
222 } else
223 queue_work(xprtiod_workqueue, &xprt->task_cleanup);
224}
225
226
227
228
229
230
231
232
233
234int xprt_reserve_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task)
235{
236 struct rpc_rqst *req = task->tk_rqstp;
237 int priority;
238
239 if (test_and_set_bit(XPRT_LOCKED, &xprt->state)) {
240 if (task == xprt->snd_task)
241 return 1;
242 goto out_sleep;
243 }
244 if (req == NULL) {
245 xprt->snd_task = task;
246 return 1;
247 }
248 if (__xprt_get_cong(xprt, task)) {
249 xprt->snd_task = task;
250 req->rq_ntrans++;
251 return 1;
252 }
253 xprt_clear_locked(xprt);
254out_sleep:
255 if (req)
256 __xprt_put_cong(xprt, req);
257 dprintk("RPC: %5u failed to lock transport %p\n", task->tk_pid, xprt);
258 task->tk_timeout = 0;
259 task->tk_status = -EAGAIN;
260 if (req == NULL)
261 priority = RPC_PRIORITY_LOW;
262 else if (!req->rq_ntrans)
263 priority = RPC_PRIORITY_NORMAL;
264 else
265 priority = RPC_PRIORITY_HIGH;
266 rpc_sleep_on_priority(&xprt->sending, task, NULL, priority);
267 return 0;
268}
269EXPORT_SYMBOL_GPL(xprt_reserve_xprt_cong);
270
271static inline int xprt_lock_write(struct rpc_xprt *xprt, struct rpc_task *task)
272{
273 int retval;
274
275 spin_lock_bh(&xprt->transport_lock);
276 retval = xprt->ops->reserve_xprt(xprt, task);
277 spin_unlock_bh(&xprt->transport_lock);
278 return retval;
279}
280
281static bool __xprt_lock_write_func(struct rpc_task *task, void *data)
282{
283 struct rpc_xprt *xprt = data;
284 struct rpc_rqst *req;
285
286 req = task->tk_rqstp;
287 xprt->snd_task = task;
288 if (req)
289 req->rq_ntrans++;
290 return true;
291}
292
293static void __xprt_lock_write_next(struct rpc_xprt *xprt)
294{
295 if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
296 return;
297
298 if (rpc_wake_up_first_on_wq(xprtiod_workqueue, &xprt->sending,
299 __xprt_lock_write_func, xprt))
300 return;
301 xprt_clear_locked(xprt);
302}
303
304static bool __xprt_lock_write_cong_func(struct rpc_task *task, void *data)
305{
306 struct rpc_xprt *xprt = data;
307 struct rpc_rqst *req;
308
309 req = task->tk_rqstp;
310 if (req == NULL) {
311 xprt->snd_task = task;
312 return true;
313 }
314 if (__xprt_get_cong(xprt, task)) {
315 xprt->snd_task = task;
316 req->rq_ntrans++;
317 return true;
318 }
319 return false;
320}
321
322static void __xprt_lock_write_next_cong(struct rpc_xprt *xprt)
323{
324 if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
325 return;
326 if (RPCXPRT_CONGESTED(xprt))
327 goto out_unlock;
328 if (rpc_wake_up_first_on_wq(xprtiod_workqueue, &xprt->sending,
329 __xprt_lock_write_cong_func, xprt))
330 return;
331out_unlock:
332 xprt_clear_locked(xprt);
333}
334
335static void xprt_task_clear_bytes_sent(struct rpc_task *task)
336{
337 if (task != NULL) {
338 struct rpc_rqst *req = task->tk_rqstp;
339 if (req != NULL)
340 req->rq_bytes_sent = 0;
341 }
342}
343
344
345
346
347
348
349
350
351void xprt_release_xprt(struct rpc_xprt *xprt, struct rpc_task *task)
352{
353 if (xprt->snd_task == task) {
354 xprt_task_clear_bytes_sent(task);
355 xprt_clear_locked(xprt);
356 __xprt_lock_write_next(xprt);
357 }
358}
359EXPORT_SYMBOL_GPL(xprt_release_xprt);
360
361
362
363
364
365
366
367
368
369void xprt_release_xprt_cong(struct rpc_xprt *xprt, struct rpc_task *task)
370{
371 if (xprt->snd_task == task) {
372 xprt_task_clear_bytes_sent(task);
373 xprt_clear_locked(xprt);
374 __xprt_lock_write_next_cong(xprt);
375 }
376}
377EXPORT_SYMBOL_GPL(xprt_release_xprt_cong);
378
379static inline void xprt_release_write(struct rpc_xprt *xprt, struct rpc_task *task)
380{
381 spin_lock_bh(&xprt->transport_lock);
382 xprt->ops->release_xprt(xprt, task);
383 spin_unlock_bh(&xprt->transport_lock);
384}
385
386
387
388
389
390static int
391__xprt_get_cong(struct rpc_xprt *xprt, struct rpc_task *task)
392{
393 struct rpc_rqst *req = task->tk_rqstp;
394
395 if (req->rq_cong)
396 return 1;
397 dprintk("RPC: %5u xprt_cwnd_limited cong = %lu cwnd = %lu\n",
398 task->tk_pid, xprt->cong, xprt->cwnd);
399 if (RPCXPRT_CONGESTED(xprt))
400 return 0;
401 req->rq_cong = 1;
402 xprt->cong += RPC_CWNDSCALE;
403 return 1;
404}
405
406
407
408
409
410static void
411__xprt_put_cong(struct rpc_xprt *xprt, struct rpc_rqst *req)
412{
413 if (!req->rq_cong)
414 return;
415 req->rq_cong = 0;
416 xprt->cong -= RPC_CWNDSCALE;
417 __xprt_lock_write_next_cong(xprt);
418}
419
420
421
422
423
424
425
426void xprt_release_rqst_cong(struct rpc_task *task)
427{
428 struct rpc_rqst *req = task->tk_rqstp;
429
430 __xprt_put_cong(req->rq_xprt, req);
431}
432EXPORT_SYMBOL_GPL(xprt_release_rqst_cong);
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450void xprt_adjust_cwnd(struct rpc_xprt *xprt, struct rpc_task *task, int result)
451{
452 struct rpc_rqst *req = task->tk_rqstp;
453 unsigned long cwnd = xprt->cwnd;
454
455 if (result >= 0 && cwnd <= xprt->cong) {
456
457
458 cwnd += (RPC_CWNDSCALE * RPC_CWNDSCALE + (cwnd >> 1)) / cwnd;
459 if (cwnd > RPC_MAXCWND(xprt))
460 cwnd = RPC_MAXCWND(xprt);
461 __xprt_lock_write_next_cong(xprt);
462 } else if (result == -ETIMEDOUT) {
463 cwnd >>= 1;
464 if (cwnd < RPC_CWNDSCALE)
465 cwnd = RPC_CWNDSCALE;
466 }
467 dprintk("RPC: cong %ld, cwnd was %ld, now %ld\n",
468 xprt->cong, xprt->cwnd, cwnd);
469 xprt->cwnd = cwnd;
470 __xprt_put_cong(xprt, req);
471}
472EXPORT_SYMBOL_GPL(xprt_adjust_cwnd);
473
474
475
476
477
478
479
480void xprt_wake_pending_tasks(struct rpc_xprt *xprt, int status)
481{
482 if (status < 0)
483 rpc_wake_up_status(&xprt->pending, status);
484 else
485 rpc_wake_up(&xprt->pending);
486}
487EXPORT_SYMBOL_GPL(xprt_wake_pending_tasks);
488
489
490
491
492
493
494
495
496
497
498void xprt_wait_for_buffer_space(struct rpc_task *task, rpc_action action)
499{
500 struct rpc_rqst *req = task->tk_rqstp;
501 struct rpc_xprt *xprt = req->rq_xprt;
502
503 task->tk_timeout = RPC_IS_SOFT(task) ? req->rq_timeout : 0;
504 rpc_sleep_on(&xprt->pending, task, action);
505}
506EXPORT_SYMBOL_GPL(xprt_wait_for_buffer_space);
507
508
509
510
511
512
513
514void xprt_write_space(struct rpc_xprt *xprt)
515{
516 spin_lock_bh(&xprt->transport_lock);
517 if (xprt->snd_task) {
518 dprintk("RPC: write space: waking waiting task on "
519 "xprt %p\n", xprt);
520 rpc_wake_up_queued_task(&xprt->pending, xprt->snd_task);
521 }
522 spin_unlock_bh(&xprt->transport_lock);
523}
524EXPORT_SYMBOL_GPL(xprt_write_space);
525
526
527
528
529
530
531
532
533
534void xprt_set_retrans_timeout_def(struct rpc_task *task)
535{
536 task->tk_timeout = task->tk_rqstp->rq_timeout;
537}
538EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_def);
539
540
541
542
543
544
545
546void xprt_set_retrans_timeout_rtt(struct rpc_task *task)
547{
548 int timer = task->tk_msg.rpc_proc->p_timer;
549 struct rpc_clnt *clnt = task->tk_client;
550 struct rpc_rtt *rtt = clnt->cl_rtt;
551 struct rpc_rqst *req = task->tk_rqstp;
552 unsigned long max_timeout = clnt->cl_timeout->to_maxval;
553
554 task->tk_timeout = rpc_calc_rto(rtt, timer);
555 task->tk_timeout <<= rpc_ntimeo(rtt, timer) + req->rq_retries;
556 if (task->tk_timeout > max_timeout || task->tk_timeout == 0)
557 task->tk_timeout = max_timeout;
558}
559EXPORT_SYMBOL_GPL(xprt_set_retrans_timeout_rtt);
560
561static void xprt_reset_majortimeo(struct rpc_rqst *req)
562{
563 const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout;
564
565 req->rq_majortimeo = req->rq_timeout;
566 if (to->to_exponential)
567 req->rq_majortimeo <<= to->to_retries;
568 else
569 req->rq_majortimeo += to->to_increment * to->to_retries;
570 if (req->rq_majortimeo > to->to_maxval || req->rq_majortimeo == 0)
571 req->rq_majortimeo = to->to_maxval;
572 req->rq_majortimeo += jiffies;
573}
574
575
576
577
578
579
580int xprt_adjust_timeout(struct rpc_rqst *req)
581{
582 struct rpc_xprt *xprt = req->rq_xprt;
583 const struct rpc_timeout *to = req->rq_task->tk_client->cl_timeout;
584 int status = 0;
585
586 if (time_before(jiffies, req->rq_majortimeo)) {
587 if (to->to_exponential)
588 req->rq_timeout <<= 1;
589 else
590 req->rq_timeout += to->to_increment;
591 if (to->to_maxval && req->rq_timeout >= to->to_maxval)
592 req->rq_timeout = to->to_maxval;
593 req->rq_retries++;
594 } else {
595 req->rq_timeout = to->to_initval;
596 req->rq_retries = 0;
597 xprt_reset_majortimeo(req);
598
599 spin_lock_bh(&xprt->transport_lock);
600 rpc_init_rtt(req->rq_task->tk_client->cl_rtt, to->to_initval);
601 spin_unlock_bh(&xprt->transport_lock);
602 status = -ETIMEDOUT;
603 }
604
605 if (req->rq_timeout == 0) {
606 printk(KERN_WARNING "xprt_adjust_timeout: rq_timeout = 0!\n");
607 req->rq_timeout = 5 * HZ;
608 }
609 return status;
610}
611
612static void xprt_autoclose(struct work_struct *work)
613{
614 struct rpc_xprt *xprt =
615 container_of(work, struct rpc_xprt, task_cleanup);
616
617 clear_bit(XPRT_CLOSE_WAIT, &xprt->state);
618 xprt->ops->close(xprt);
619 xprt_release_write(xprt, NULL);
620 wake_up_bit(&xprt->state, XPRT_LOCKED);
621}
622
623
624
625
626
627
628void xprt_disconnect_done(struct rpc_xprt *xprt)
629{
630 dprintk("RPC: disconnected transport %p\n", xprt);
631 spin_lock_bh(&xprt->transport_lock);
632 xprt_clear_connected(xprt);
633 xprt_wake_pending_tasks(xprt, -EAGAIN);
634 spin_unlock_bh(&xprt->transport_lock);
635}
636EXPORT_SYMBOL_GPL(xprt_disconnect_done);
637
638
639
640
641
642
643void xprt_force_disconnect(struct rpc_xprt *xprt)
644{
645
646 spin_lock_bh(&xprt->transport_lock);
647 set_bit(XPRT_CLOSE_WAIT, &xprt->state);
648
649 if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
650 queue_work(xprtiod_workqueue, &xprt->task_cleanup);
651 xprt_wake_pending_tasks(xprt, -EAGAIN);
652 spin_unlock_bh(&xprt->transport_lock);
653}
654
655
656
657
658
659
660
661
662
663
664
665
666void xprt_conditional_disconnect(struct rpc_xprt *xprt, unsigned int cookie)
667{
668
669 spin_lock_bh(&xprt->transport_lock);
670 if (cookie != xprt->connect_cookie)
671 goto out;
672 if (test_bit(XPRT_CLOSING, &xprt->state))
673 goto out;
674 set_bit(XPRT_CLOSE_WAIT, &xprt->state);
675
676 if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
677 queue_work(xprtiod_workqueue, &xprt->task_cleanup);
678 xprt_wake_pending_tasks(xprt, -EAGAIN);
679out:
680 spin_unlock_bh(&xprt->transport_lock);
681}
682
683static bool
684xprt_has_timer(const struct rpc_xprt *xprt)
685{
686 return xprt->idle_timeout != 0;
687}
688
689static void
690xprt_schedule_autodisconnect(struct rpc_xprt *xprt)
691 __must_hold(&xprt->transport_lock)
692{
693 if (list_empty(&xprt->recv) && xprt_has_timer(xprt))
694 mod_timer(&xprt->timer, xprt->last_used + xprt->idle_timeout);
695}
696
697static void
698xprt_init_autodisconnect(unsigned long data)
699{
700 struct rpc_xprt *xprt = (struct rpc_xprt *)data;
701
702 spin_lock(&xprt->transport_lock);
703 if (!list_empty(&xprt->recv))
704 goto out_abort;
705
706 xprt->last_used = jiffies;
707 if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
708 goto out_abort;
709 spin_unlock(&xprt->transport_lock);
710 queue_work(xprtiod_workqueue, &xprt->task_cleanup);
711 return;
712out_abort:
713 spin_unlock(&xprt->transport_lock);
714}
715
716bool xprt_lock_connect(struct rpc_xprt *xprt,
717 struct rpc_task *task,
718 void *cookie)
719{
720 bool ret = false;
721
722 spin_lock_bh(&xprt->transport_lock);
723 if (!test_bit(XPRT_LOCKED, &xprt->state))
724 goto out;
725 if (xprt->snd_task != task)
726 goto out;
727 xprt_task_clear_bytes_sent(task);
728 xprt->snd_task = cookie;
729 ret = true;
730out:
731 spin_unlock_bh(&xprt->transport_lock);
732 return ret;
733}
734
735void xprt_unlock_connect(struct rpc_xprt *xprt, void *cookie)
736{
737 spin_lock_bh(&xprt->transport_lock);
738 if (xprt->snd_task != cookie)
739 goto out;
740 if (!test_bit(XPRT_LOCKED, &xprt->state))
741 goto out;
742 xprt->snd_task =NULL;
743 xprt->ops->release_xprt(xprt, NULL);
744 xprt_schedule_autodisconnect(xprt);
745out:
746 spin_unlock_bh(&xprt->transport_lock);
747 wake_up_bit(&xprt->state, XPRT_LOCKED);
748}
749
750
751
752
753
754
755void xprt_connect(struct rpc_task *task)
756{
757 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
758
759 dprintk("RPC: %5u xprt_connect xprt %p %s connected\n", task->tk_pid,
760 xprt, (xprt_connected(xprt) ? "is" : "is not"));
761
762 if (!xprt_bound(xprt)) {
763 task->tk_status = -EAGAIN;
764 return;
765 }
766 if (!xprt_lock_write(xprt, task))
767 return;
768
769 if (test_and_clear_bit(XPRT_CLOSE_WAIT, &xprt->state))
770 xprt->ops->close(xprt);
771
772 if (!xprt_connected(xprt)) {
773 task->tk_rqstp->rq_bytes_sent = 0;
774 task->tk_timeout = task->tk_rqstp->rq_timeout;
775 task->tk_rqstp->rq_connect_cookie = xprt->connect_cookie;
776 rpc_sleep_on(&xprt->pending, task, xprt_connect_status);
777
778 if (test_bit(XPRT_CLOSING, &xprt->state))
779 return;
780 if (xprt_test_and_set_connecting(xprt))
781 return;
782 xprt->stat.connect_start = jiffies;
783 xprt->ops->connect(xprt, task);
784 }
785 xprt_release_write(xprt, task);
786}
787
788static void xprt_connect_status(struct rpc_task *task)
789{
790 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
791
792 if (task->tk_status == 0) {
793 xprt->stat.connect_count++;
794 xprt->stat.connect_time += (long)jiffies - xprt->stat.connect_start;
795 dprintk("RPC: %5u xprt_connect_status: connection established\n",
796 task->tk_pid);
797 return;
798 }
799
800 switch (task->tk_status) {
801 case -ECONNREFUSED:
802 case -ECONNRESET:
803 case -ECONNABORTED:
804 case -ENETUNREACH:
805 case -EHOSTUNREACH:
806 case -EPIPE:
807 case -EAGAIN:
808 dprintk("RPC: %5u xprt_connect_status: retrying\n", task->tk_pid);
809 break;
810 case -ETIMEDOUT:
811 dprintk("RPC: %5u xprt_connect_status: connect attempt timed "
812 "out\n", task->tk_pid);
813 break;
814 default:
815 dprintk("RPC: %5u xprt_connect_status: error %d connecting to "
816 "server %s\n", task->tk_pid, -task->tk_status,
817 xprt->servername);
818 task->tk_status = -EIO;
819 }
820}
821
822
823
824
825
826
827
828struct rpc_rqst *xprt_lookup_rqst(struct rpc_xprt *xprt, __be32 xid)
829{
830 struct rpc_rqst *entry;
831
832 list_for_each_entry(entry, &xprt->recv, rq_list)
833 if (entry->rq_xid == xid) {
834 trace_xprt_lookup_rqst(xprt, xid, 0);
835 return entry;
836 }
837
838 dprintk("RPC: xprt_lookup_rqst did not find xid %08x\n",
839 ntohl(xid));
840 trace_xprt_lookup_rqst(xprt, xid, -ENOENT);
841 xprt->stat.bad_xids++;
842 return NULL;
843}
844EXPORT_SYMBOL_GPL(xprt_lookup_rqst);
845
846static void xprt_update_rtt(struct rpc_task *task)
847{
848 struct rpc_rqst *req = task->tk_rqstp;
849 struct rpc_rtt *rtt = task->tk_client->cl_rtt;
850 unsigned int timer = task->tk_msg.rpc_proc->p_timer;
851 long m = usecs_to_jiffies(ktime_to_us(req->rq_rtt));
852
853 if (timer) {
854 if (req->rq_ntrans == 1)
855 rpc_update_rtt(rtt, timer, m);
856 rpc_set_timeo(rtt, timer, req->rq_ntrans - 1);
857 }
858}
859
860
861
862
863
864
865
866
867void xprt_complete_rqst(struct rpc_task *task, int copied)
868{
869 struct rpc_rqst *req = task->tk_rqstp;
870 struct rpc_xprt *xprt = req->rq_xprt;
871
872 dprintk("RPC: %5u xid %08x complete (%d bytes received)\n",
873 task->tk_pid, ntohl(req->rq_xid), copied);
874 trace_xprt_complete_rqst(xprt, req->rq_xid, copied);
875
876 xprt->stat.recvs++;
877 req->rq_rtt = ktime_sub(ktime_get(), req->rq_xtime);
878 if (xprt->ops->timer != NULL)
879 xprt_update_rtt(task);
880
881 list_del_init(&req->rq_list);
882 req->rq_private_buf.len = copied;
883
884
885 smp_wmb();
886 req->rq_reply_bytes_recvd = copied;
887 rpc_wake_up_queued_task(&xprt->pending, task);
888}
889EXPORT_SYMBOL_GPL(xprt_complete_rqst);
890
891static void xprt_timer(struct rpc_task *task)
892{
893 struct rpc_rqst *req = task->tk_rqstp;
894 struct rpc_xprt *xprt = req->rq_xprt;
895
896 if (task->tk_status != -ETIMEDOUT)
897 return;
898 dprintk("RPC: %5u xprt_timer\n", task->tk_pid);
899
900 if (!req->rq_reply_bytes_recvd) {
901 if (xprt->ops->timer)
902 xprt->ops->timer(xprt, task);
903 } else
904 task->tk_status = 0;
905}
906
907
908
909
910
911
912bool xprt_prepare_transmit(struct rpc_task *task)
913{
914 struct rpc_rqst *req = task->tk_rqstp;
915 struct rpc_xprt *xprt = req->rq_xprt;
916 bool ret = false;
917
918 dprintk("RPC: %5u xprt_prepare_transmit\n", task->tk_pid);
919
920 spin_lock_bh(&xprt->transport_lock);
921 if (!req->rq_bytes_sent) {
922 if (req->rq_reply_bytes_recvd) {
923 task->tk_status = req->rq_reply_bytes_recvd;
924 goto out_unlock;
925 }
926 if ((task->tk_flags & RPC_TASK_NO_RETRANS_TIMEOUT)
927 && xprt_connected(xprt)
928 && req->rq_connect_cookie == xprt->connect_cookie) {
929 xprt->ops->set_retrans_timeout(task);
930 rpc_sleep_on(&xprt->pending, task, xprt_timer);
931 goto out_unlock;
932 }
933 }
934 if (!xprt->ops->reserve_xprt(xprt, task)) {
935 task->tk_status = -EAGAIN;
936 goto out_unlock;
937 }
938 ret = true;
939out_unlock:
940 spin_unlock_bh(&xprt->transport_lock);
941 return ret;
942}
943
944void xprt_end_transmit(struct rpc_task *task)
945{
946 xprt_release_write(task->tk_rqstp->rq_xprt, task);
947}
948
949
950
951
952
953
954
955void xprt_transmit(struct rpc_task *task)
956{
957 struct rpc_rqst *req = task->tk_rqstp;
958 struct rpc_xprt *xprt = req->rq_xprt;
959 int status, numreqs;
960
961 dprintk("RPC: %5u xprt_transmit(%u)\n", task->tk_pid, req->rq_slen);
962
963 if (!req->rq_reply_bytes_recvd) {
964 if (list_empty(&req->rq_list) && rpc_reply_expected(task)) {
965
966
967
968 spin_lock_bh(&xprt->transport_lock);
969
970 memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
971 sizeof(req->rq_private_buf));
972
973 list_add_tail(&req->rq_list, &xprt->recv);
974 spin_unlock_bh(&xprt->transport_lock);
975 xprt_reset_majortimeo(req);
976
977 del_singleshot_timer_sync(&xprt->timer);
978 }
979 } else if (!req->rq_bytes_sent)
980 return;
981
982 req->rq_xtime = ktime_get();
983 status = xprt->ops->send_request(task);
984 trace_xprt_transmit(xprt, req->rq_xid, status);
985 if (status != 0) {
986 task->tk_status = status;
987 return;
988 }
989 xprt_inject_disconnect(xprt);
990
991 dprintk("RPC: %5u xmit complete\n", task->tk_pid);
992 task->tk_flags |= RPC_TASK_SENT;
993 spin_lock_bh(&xprt->transport_lock);
994
995 xprt->ops->set_retrans_timeout(task);
996
997 numreqs = atomic_read(&xprt->num_reqs);
998 if (numreqs > xprt->stat.max_slots)
999 xprt->stat.max_slots = numreqs;
1000 xprt->stat.sends++;
1001 xprt->stat.req_u += xprt->stat.sends - xprt->stat.recvs;
1002 xprt->stat.bklog_u += xprt->backlog.qlen;
1003 xprt->stat.sending_u += xprt->sending.qlen;
1004 xprt->stat.pending_u += xprt->pending.qlen;
1005
1006
1007 if (!xprt_connected(xprt))
1008 task->tk_status = -ENOTCONN;
1009 else {
1010
1011
1012
1013
1014 if (!req->rq_reply_bytes_recvd && rpc_reply_expected(task))
1015 rpc_sleep_on(&xprt->pending, task, xprt_timer);
1016 req->rq_connect_cookie = xprt->connect_cookie;
1017 }
1018 spin_unlock_bh(&xprt->transport_lock);
1019}
1020
1021static void xprt_add_backlog(struct rpc_xprt *xprt, struct rpc_task *task)
1022{
1023 set_bit(XPRT_CONGESTED, &xprt->state);
1024 rpc_sleep_on(&xprt->backlog, task, NULL);
1025}
1026
1027static void xprt_wake_up_backlog(struct rpc_xprt *xprt)
1028{
1029 if (rpc_wake_up_next(&xprt->backlog) == NULL)
1030 clear_bit(XPRT_CONGESTED, &xprt->state);
1031}
1032
1033static bool xprt_throttle_congested(struct rpc_xprt *xprt, struct rpc_task *task)
1034{
1035 bool ret = false;
1036
1037 if (!test_bit(XPRT_CONGESTED, &xprt->state))
1038 goto out;
1039 spin_lock(&xprt->reserve_lock);
1040 if (test_bit(XPRT_CONGESTED, &xprt->state)) {
1041 rpc_sleep_on(&xprt->backlog, task, NULL);
1042 ret = true;
1043 }
1044 spin_unlock(&xprt->reserve_lock);
1045out:
1046 return ret;
1047}
1048
1049static struct rpc_rqst *xprt_dynamic_alloc_slot(struct rpc_xprt *xprt, gfp_t gfp_flags)
1050{
1051 struct rpc_rqst *req = ERR_PTR(-EAGAIN);
1052
1053 if (!atomic_add_unless(&xprt->num_reqs, 1, xprt->max_reqs))
1054 goto out;
1055 req = kzalloc(sizeof(struct rpc_rqst), gfp_flags);
1056 if (req != NULL)
1057 goto out;
1058 atomic_dec(&xprt->num_reqs);
1059 req = ERR_PTR(-ENOMEM);
1060out:
1061 return req;
1062}
1063
1064static bool xprt_dynamic_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req)
1065{
1066 if (atomic_add_unless(&xprt->num_reqs, -1, xprt->min_reqs)) {
1067 kfree(req);
1068 return true;
1069 }
1070 return false;
1071}
1072
1073void xprt_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task)
1074{
1075 struct rpc_rqst *req;
1076
1077 spin_lock(&xprt->reserve_lock);
1078 if (!list_empty(&xprt->free)) {
1079 req = list_entry(xprt->free.next, struct rpc_rqst, rq_list);
1080 list_del(&req->rq_list);
1081 goto out_init_req;
1082 }
1083 req = xprt_dynamic_alloc_slot(xprt, GFP_NOWAIT|__GFP_NOWARN);
1084 if (!IS_ERR(req))
1085 goto out_init_req;
1086 switch (PTR_ERR(req)) {
1087 case -ENOMEM:
1088 dprintk("RPC: dynamic allocation of request slot "
1089 "failed! Retrying\n");
1090 task->tk_status = -ENOMEM;
1091 break;
1092 case -EAGAIN:
1093 xprt_add_backlog(xprt, task);
1094 dprintk("RPC: waiting for request slot\n");
1095 default:
1096 task->tk_status = -EAGAIN;
1097 }
1098 spin_unlock(&xprt->reserve_lock);
1099 return;
1100out_init_req:
1101 task->tk_status = 0;
1102 task->tk_rqstp = req;
1103 xprt_request_init(task, xprt);
1104 spin_unlock(&xprt->reserve_lock);
1105}
1106EXPORT_SYMBOL_GPL(xprt_alloc_slot);
1107
1108void xprt_lock_and_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task)
1109{
1110
1111
1112
1113
1114
1115 if (xprt_lock_write(xprt, task)) {
1116 xprt_alloc_slot(xprt, task);
1117 xprt_release_write(xprt, task);
1118 }
1119}
1120EXPORT_SYMBOL_GPL(xprt_lock_and_alloc_slot);
1121
1122static void xprt_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req)
1123{
1124 spin_lock(&xprt->reserve_lock);
1125 if (!xprt_dynamic_free_slot(xprt, req)) {
1126 memset(req, 0, sizeof(*req));
1127 list_add(&req->rq_list, &xprt->free);
1128 }
1129 xprt_wake_up_backlog(xprt);
1130 spin_unlock(&xprt->reserve_lock);
1131}
1132
1133static void xprt_free_all_slots(struct rpc_xprt *xprt)
1134{
1135 struct rpc_rqst *req;
1136 while (!list_empty(&xprt->free)) {
1137 req = list_first_entry(&xprt->free, struct rpc_rqst, rq_list);
1138 list_del(&req->rq_list);
1139 kfree(req);
1140 }
1141}
1142
1143struct rpc_xprt *xprt_alloc(struct net *net, size_t size,
1144 unsigned int num_prealloc,
1145 unsigned int max_alloc)
1146{
1147 struct rpc_xprt *xprt;
1148 struct rpc_rqst *req;
1149 int i;
1150
1151 xprt = kzalloc(size, GFP_KERNEL);
1152 if (xprt == NULL)
1153 goto out;
1154
1155 xprt_init(xprt, net);
1156
1157 for (i = 0; i < num_prealloc; i++) {
1158 req = kzalloc(sizeof(struct rpc_rqst), GFP_KERNEL);
1159 if (!req)
1160 goto out_free;
1161 list_add(&req->rq_list, &xprt->free);
1162 }
1163 if (max_alloc > num_prealloc)
1164 xprt->max_reqs = max_alloc;
1165 else
1166 xprt->max_reqs = num_prealloc;
1167 xprt->min_reqs = num_prealloc;
1168 atomic_set(&xprt->num_reqs, num_prealloc);
1169
1170 return xprt;
1171
1172out_free:
1173 xprt_free(xprt);
1174out:
1175 return NULL;
1176}
1177EXPORT_SYMBOL_GPL(xprt_alloc);
1178
1179void xprt_free(struct rpc_xprt *xprt)
1180{
1181 put_net(xprt->xprt_net);
1182 xprt_free_all_slots(xprt);
1183 kfree_rcu(xprt, rcu);
1184}
1185EXPORT_SYMBOL_GPL(xprt_free);
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195void xprt_reserve(struct rpc_task *task)
1196{
1197 struct rpc_xprt *xprt = task->tk_xprt;
1198
1199 task->tk_status = 0;
1200 if (task->tk_rqstp != NULL)
1201 return;
1202
1203 task->tk_timeout = 0;
1204 task->tk_status = -EAGAIN;
1205 if (!xprt_throttle_congested(xprt, task))
1206 xprt->ops->alloc_slot(xprt, task);
1207}
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218void xprt_retry_reserve(struct rpc_task *task)
1219{
1220 struct rpc_xprt *xprt = task->tk_xprt;
1221
1222 task->tk_status = 0;
1223 if (task->tk_rqstp != NULL)
1224 return;
1225
1226 task->tk_timeout = 0;
1227 task->tk_status = -EAGAIN;
1228 xprt->ops->alloc_slot(xprt, task);
1229}
1230
1231static inline __be32 xprt_alloc_xid(struct rpc_xprt *xprt)
1232{
1233 return (__force __be32)xprt->xid++;
1234}
1235
1236static inline void xprt_init_xid(struct rpc_xprt *xprt)
1237{
1238 xprt->xid = prandom_u32();
1239}
1240
1241static void xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
1242{
1243 struct rpc_rqst *req = task->tk_rqstp;
1244
1245 INIT_LIST_HEAD(&req->rq_list);
1246 req->rq_timeout = task->tk_client->cl_timeout->to_initval;
1247 req->rq_task = task;
1248 req->rq_xprt = xprt;
1249 req->rq_buffer = NULL;
1250 req->rq_xid = xprt_alloc_xid(xprt);
1251 req->rq_connect_cookie = xprt->connect_cookie - 1;
1252 req->rq_bytes_sent = 0;
1253 req->rq_snd_buf.len = 0;
1254 req->rq_snd_buf.buflen = 0;
1255 req->rq_rcv_buf.len = 0;
1256 req->rq_rcv_buf.buflen = 0;
1257 req->rq_release_snd_buf = NULL;
1258 xprt_reset_majortimeo(req);
1259 dprintk("RPC: %5u reserved req %p xid %08x\n", task->tk_pid,
1260 req, ntohl(req->rq_xid));
1261}
1262
1263
1264
1265
1266
1267
1268void xprt_release(struct rpc_task *task)
1269{
1270 struct rpc_xprt *xprt;
1271 struct rpc_rqst *req = task->tk_rqstp;
1272
1273 if (req == NULL) {
1274 if (task->tk_client) {
1275 xprt = task->tk_xprt;
1276 if (xprt->snd_task == task)
1277 xprt_release_write(xprt, task);
1278 }
1279 return;
1280 }
1281
1282 xprt = req->rq_xprt;
1283 if (task->tk_ops->rpc_count_stats != NULL)
1284 task->tk_ops->rpc_count_stats(task, task->tk_calldata);
1285 else if (task->tk_client)
1286 rpc_count_iostats(task, task->tk_client->cl_metrics);
1287 spin_lock_bh(&xprt->transport_lock);
1288 xprt->ops->release_xprt(xprt, task);
1289 if (xprt->ops->release_request)
1290 xprt->ops->release_request(task);
1291 if (!list_empty(&req->rq_list))
1292 list_del(&req->rq_list);
1293 xprt->last_used = jiffies;
1294 xprt_schedule_autodisconnect(xprt);
1295 spin_unlock_bh(&xprt->transport_lock);
1296 if (req->rq_buffer)
1297 xprt->ops->buf_free(task);
1298 xprt_inject_disconnect(xprt);
1299 if (req->rq_cred != NULL)
1300 put_rpccred(req->rq_cred);
1301 task->tk_rqstp = NULL;
1302 if (req->rq_release_snd_buf)
1303 req->rq_release_snd_buf(req);
1304
1305 dprintk("RPC: %5u release request %p\n", task->tk_pid, req);
1306 if (likely(!bc_prealloc(req)))
1307 xprt_free_slot(xprt, req);
1308 else
1309 xprt_free_bc_request(req);
1310}
1311
1312static void xprt_init(struct rpc_xprt *xprt, struct net *net)
1313{
1314 kref_init(&xprt->kref);
1315
1316 spin_lock_init(&xprt->transport_lock);
1317 spin_lock_init(&xprt->reserve_lock);
1318
1319 INIT_LIST_HEAD(&xprt->free);
1320 INIT_LIST_HEAD(&xprt->recv);
1321#if defined(CONFIG_SUNRPC_BACKCHANNEL)
1322 spin_lock_init(&xprt->bc_pa_lock);
1323 INIT_LIST_HEAD(&xprt->bc_pa_list);
1324#endif
1325 INIT_LIST_HEAD(&xprt->xprt_switch);
1326
1327 xprt->last_used = jiffies;
1328 xprt->cwnd = RPC_INITCWND;
1329 xprt->bind_index = 0;
1330
1331 rpc_init_wait_queue(&xprt->binding, "xprt_binding");
1332 rpc_init_wait_queue(&xprt->pending, "xprt_pending");
1333 rpc_init_priority_wait_queue(&xprt->sending, "xprt_sending");
1334 rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog");
1335
1336 xprt_init_xid(xprt);
1337
1338 xprt->xprt_net = get_net(net);
1339}
1340
1341
1342
1343
1344
1345
1346struct rpc_xprt *xprt_create_transport(struct xprt_create *args)
1347{
1348 struct rpc_xprt *xprt;
1349 struct xprt_class *t;
1350
1351 spin_lock(&xprt_list_lock);
1352 list_for_each_entry(t, &xprt_list, list) {
1353 if (t->ident == args->ident) {
1354 spin_unlock(&xprt_list_lock);
1355 goto found;
1356 }
1357 }
1358 spin_unlock(&xprt_list_lock);
1359 dprintk("RPC: transport (%d) not supported\n", args->ident);
1360 return ERR_PTR(-EIO);
1361
1362found:
1363 xprt = t->setup(args);
1364 if (IS_ERR(xprt)) {
1365 dprintk("RPC: xprt_create_transport: failed, %ld\n",
1366 -PTR_ERR(xprt));
1367 goto out;
1368 }
1369 if (args->flags & XPRT_CREATE_NO_IDLE_TIMEOUT)
1370 xprt->idle_timeout = 0;
1371 INIT_WORK(&xprt->task_cleanup, xprt_autoclose);
1372 if (xprt_has_timer(xprt))
1373 setup_timer(&xprt->timer, xprt_init_autodisconnect,
1374 (unsigned long)xprt);
1375 else
1376 init_timer(&xprt->timer);
1377
1378 if (strlen(args->servername) > RPC_MAXNETNAMELEN) {
1379 xprt_destroy(xprt);
1380 return ERR_PTR(-EINVAL);
1381 }
1382 xprt->servername = kstrdup(args->servername, GFP_KERNEL);
1383 if (xprt->servername == NULL) {
1384 xprt_destroy(xprt);
1385 return ERR_PTR(-ENOMEM);
1386 }
1387
1388 rpc_xprt_debugfs_register(xprt);
1389
1390 dprintk("RPC: created transport %p with %u slots\n", xprt,
1391 xprt->max_reqs);
1392out:
1393 return xprt;
1394}
1395
1396
1397
1398
1399
1400
1401static void xprt_destroy(struct rpc_xprt *xprt)
1402{
1403 dprintk("RPC: destroying transport %p\n", xprt);
1404
1405
1406 wait_on_bit_lock(&xprt->state, XPRT_LOCKED, TASK_UNINTERRUPTIBLE);
1407
1408 del_timer_sync(&xprt->timer);
1409
1410 rpc_xprt_debugfs_unregister(xprt);
1411 rpc_destroy_wait_queue(&xprt->binding);
1412 rpc_destroy_wait_queue(&xprt->pending);
1413 rpc_destroy_wait_queue(&xprt->sending);
1414 rpc_destroy_wait_queue(&xprt->backlog);
1415 cancel_work_sync(&xprt->task_cleanup);
1416 kfree(xprt->servername);
1417
1418
1419
1420 xprt->ops->destroy(xprt);
1421}
1422
1423static void xprt_destroy_kref(struct kref *kref)
1424{
1425 xprt_destroy(container_of(kref, struct rpc_xprt, kref));
1426}
1427
1428
1429
1430
1431
1432
1433struct rpc_xprt *xprt_get(struct rpc_xprt *xprt)
1434{
1435 if (xprt != NULL && kref_get_unless_zero(&xprt->kref))
1436 return xprt;
1437 return NULL;
1438}
1439EXPORT_SYMBOL_GPL(xprt_get);
1440
1441
1442
1443
1444
1445
1446void xprt_put(struct rpc_xprt *xprt)
1447{
1448 if (xprt != NULL)
1449 kref_put(&xprt->kref, xprt_destroy_kref);
1450}
1451EXPORT_SYMBOL_GPL(xprt_put);
1452