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}
654EXPORT_SYMBOL_GPL(xprt_force_disconnect);
655
656
657
658
659
660
661
662
663
664
665
666
667void xprt_conditional_disconnect(struct rpc_xprt *xprt, unsigned int cookie)
668{
669
670 spin_lock_bh(&xprt->transport_lock);
671 if (cookie != xprt->connect_cookie)
672 goto out;
673 if (test_bit(XPRT_CLOSING, &xprt->state))
674 goto out;
675 set_bit(XPRT_CLOSE_WAIT, &xprt->state);
676
677 if (test_and_set_bit(XPRT_LOCKED, &xprt->state) == 0)
678 queue_work(xprtiod_workqueue, &xprt->task_cleanup);
679 xprt_wake_pending_tasks(xprt, -EAGAIN);
680out:
681 spin_unlock_bh(&xprt->transport_lock);
682}
683
684static bool
685xprt_has_timer(const struct rpc_xprt *xprt)
686{
687 return xprt->idle_timeout != 0;
688}
689
690static void
691xprt_schedule_autodisconnect(struct rpc_xprt *xprt)
692 __must_hold(&xprt->transport_lock)
693{
694 if (list_empty(&xprt->recv) && xprt_has_timer(xprt))
695 mod_timer(&xprt->timer, xprt->last_used + xprt->idle_timeout);
696}
697
698static void
699xprt_init_autodisconnect(unsigned long data)
700{
701 struct rpc_xprt *xprt = (struct rpc_xprt *)data;
702
703 spin_lock(&xprt->transport_lock);
704 if (!list_empty(&xprt->recv))
705 goto out_abort;
706
707 xprt->last_used = jiffies;
708 if (test_and_set_bit(XPRT_LOCKED, &xprt->state))
709 goto out_abort;
710 spin_unlock(&xprt->transport_lock);
711 queue_work(xprtiod_workqueue, &xprt->task_cleanup);
712 return;
713out_abort:
714 spin_unlock(&xprt->transport_lock);
715}
716
717bool xprt_lock_connect(struct rpc_xprt *xprt,
718 struct rpc_task *task,
719 void *cookie)
720{
721 bool ret = false;
722
723 spin_lock_bh(&xprt->transport_lock);
724 if (!test_bit(XPRT_LOCKED, &xprt->state))
725 goto out;
726 if (xprt->snd_task != task)
727 goto out;
728 xprt_task_clear_bytes_sent(task);
729 xprt->snd_task = cookie;
730 ret = true;
731out:
732 spin_unlock_bh(&xprt->transport_lock);
733 return ret;
734}
735
736void xprt_unlock_connect(struct rpc_xprt *xprt, void *cookie)
737{
738 spin_lock_bh(&xprt->transport_lock);
739 if (xprt->snd_task != cookie)
740 goto out;
741 if (!test_bit(XPRT_LOCKED, &xprt->state))
742 goto out;
743 xprt->snd_task =NULL;
744 xprt->ops->release_xprt(xprt, NULL);
745 xprt_schedule_autodisconnect(xprt);
746out:
747 spin_unlock_bh(&xprt->transport_lock);
748 wake_up_bit(&xprt->state, XPRT_LOCKED);
749}
750
751
752
753
754
755
756void xprt_connect(struct rpc_task *task)
757{
758 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
759
760 dprintk("RPC: %5u xprt_connect xprt %p %s connected\n", task->tk_pid,
761 xprt, (xprt_connected(xprt) ? "is" : "is not"));
762
763 if (!xprt_bound(xprt)) {
764 task->tk_status = -EAGAIN;
765 return;
766 }
767 if (!xprt_lock_write(xprt, task))
768 return;
769
770 if (test_and_clear_bit(XPRT_CLOSE_WAIT, &xprt->state))
771 xprt->ops->close(xprt);
772
773 if (!xprt_connected(xprt)) {
774 task->tk_rqstp->rq_bytes_sent = 0;
775 task->tk_timeout = task->tk_rqstp->rq_timeout;
776 task->tk_rqstp->rq_connect_cookie = xprt->connect_cookie;
777 rpc_sleep_on(&xprt->pending, task, xprt_connect_status);
778
779 if (test_bit(XPRT_CLOSING, &xprt->state))
780 return;
781 if (xprt_test_and_set_connecting(xprt))
782 return;
783 xprt->stat.connect_start = jiffies;
784 xprt->ops->connect(xprt, task);
785 }
786 xprt_release_write(xprt, task);
787}
788
789static void xprt_connect_status(struct rpc_task *task)
790{
791 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt;
792
793 if (task->tk_status == 0) {
794 xprt->stat.connect_count++;
795 xprt->stat.connect_time += (long)jiffies - xprt->stat.connect_start;
796 dprintk("RPC: %5u xprt_connect_status: connection established\n",
797 task->tk_pid);
798 return;
799 }
800
801 switch (task->tk_status) {
802 case -ECONNREFUSED:
803 case -ECONNRESET:
804 case -ECONNABORTED:
805 case -ENETUNREACH:
806 case -EHOSTUNREACH:
807 case -EPIPE:
808 case -EAGAIN:
809 dprintk("RPC: %5u xprt_connect_status: retrying\n", task->tk_pid);
810 break;
811 case -ETIMEDOUT:
812 dprintk("RPC: %5u xprt_connect_status: connect attempt timed "
813 "out\n", task->tk_pid);
814 break;
815 default:
816 dprintk("RPC: %5u xprt_connect_status: error %d connecting to "
817 "server %s\n", task->tk_pid, -task->tk_status,
818 xprt->servername);
819 task->tk_status = -EIO;
820 }
821}
822
823
824
825
826
827
828
829struct rpc_rqst *xprt_lookup_rqst(struct rpc_xprt *xprt, __be32 xid)
830{
831 struct rpc_rqst *entry;
832
833 list_for_each_entry(entry, &xprt->recv, rq_list)
834 if (entry->rq_xid == xid) {
835 trace_xprt_lookup_rqst(xprt, xid, 0);
836 return entry;
837 }
838
839 dprintk("RPC: xprt_lookup_rqst did not find xid %08x\n",
840 ntohl(xid));
841 trace_xprt_lookup_rqst(xprt, xid, -ENOENT);
842 xprt->stat.bad_xids++;
843 return NULL;
844}
845EXPORT_SYMBOL_GPL(xprt_lookup_rqst);
846
847static void xprt_update_rtt(struct rpc_task *task)
848{
849 struct rpc_rqst *req = task->tk_rqstp;
850 struct rpc_rtt *rtt = task->tk_client->cl_rtt;
851 unsigned int timer = task->tk_msg.rpc_proc->p_timer;
852 long m = usecs_to_jiffies(ktime_to_us(req->rq_rtt));
853
854 if (timer) {
855 if (req->rq_ntrans == 1)
856 rpc_update_rtt(rtt, timer, m);
857 rpc_set_timeo(rtt, timer, req->rq_ntrans - 1);
858 }
859}
860
861
862
863
864
865
866
867
868void xprt_complete_rqst(struct rpc_task *task, int copied)
869{
870 struct rpc_rqst *req = task->tk_rqstp;
871 struct rpc_xprt *xprt = req->rq_xprt;
872
873 dprintk("RPC: %5u xid %08x complete (%d bytes received)\n",
874 task->tk_pid, ntohl(req->rq_xid), copied);
875 trace_xprt_complete_rqst(xprt, req->rq_xid, copied);
876
877 xprt->stat.recvs++;
878 req->rq_rtt = ktime_sub(ktime_get(), req->rq_xtime);
879 if (xprt->ops->timer != NULL)
880 xprt_update_rtt(task);
881
882 list_del_init(&req->rq_list);
883 req->rq_private_buf.len = copied;
884
885
886 smp_wmb();
887 req->rq_reply_bytes_recvd = copied;
888 rpc_wake_up_queued_task(&xprt->pending, task);
889}
890EXPORT_SYMBOL_GPL(xprt_complete_rqst);
891
892static void xprt_timer(struct rpc_task *task)
893{
894 struct rpc_rqst *req = task->tk_rqstp;
895 struct rpc_xprt *xprt = req->rq_xprt;
896
897 if (task->tk_status != -ETIMEDOUT)
898 return;
899 dprintk("RPC: %5u xprt_timer\n", task->tk_pid);
900
901 if (!req->rq_reply_bytes_recvd) {
902 if (xprt->ops->timer)
903 xprt->ops->timer(xprt, task);
904 } else
905 task->tk_status = 0;
906}
907
908
909
910
911
912
913bool xprt_prepare_transmit(struct rpc_task *task)
914{
915 struct rpc_rqst *req = task->tk_rqstp;
916 struct rpc_xprt *xprt = req->rq_xprt;
917 bool ret = false;
918
919 dprintk("RPC: %5u xprt_prepare_transmit\n", task->tk_pid);
920
921 spin_lock_bh(&xprt->transport_lock);
922 if (!req->rq_bytes_sent) {
923 if (req->rq_reply_bytes_recvd) {
924 task->tk_status = req->rq_reply_bytes_recvd;
925 goto out_unlock;
926 }
927 if ((task->tk_flags & RPC_TASK_NO_RETRANS_TIMEOUT)
928 && xprt_connected(xprt)
929 && req->rq_connect_cookie == xprt->connect_cookie) {
930 xprt->ops->set_retrans_timeout(task);
931 rpc_sleep_on(&xprt->pending, task, xprt_timer);
932 goto out_unlock;
933 }
934 }
935 if (!xprt->ops->reserve_xprt(xprt, task)) {
936 task->tk_status = -EAGAIN;
937 goto out_unlock;
938 }
939 ret = true;
940out_unlock:
941 spin_unlock_bh(&xprt->transport_lock);
942 return ret;
943}
944
945void xprt_end_transmit(struct rpc_task *task)
946{
947 xprt_release_write(task->tk_rqstp->rq_xprt, task);
948}
949
950
951
952
953
954
955
956void xprt_transmit(struct rpc_task *task)
957{
958 struct rpc_rqst *req = task->tk_rqstp;
959 struct rpc_xprt *xprt = req->rq_xprt;
960 int status, numreqs;
961
962 dprintk("RPC: %5u xprt_transmit(%u)\n", task->tk_pid, req->rq_slen);
963
964 if (!req->rq_reply_bytes_recvd) {
965 if (list_empty(&req->rq_list) && rpc_reply_expected(task)) {
966
967
968
969 spin_lock_bh(&xprt->transport_lock);
970
971 memcpy(&req->rq_private_buf, &req->rq_rcv_buf,
972 sizeof(req->rq_private_buf));
973
974 list_add_tail(&req->rq_list, &xprt->recv);
975 spin_unlock_bh(&xprt->transport_lock);
976 xprt_reset_majortimeo(req);
977
978 del_singleshot_timer_sync(&xprt->timer);
979 }
980 } else if (!req->rq_bytes_sent)
981 return;
982
983 req->rq_xtime = ktime_get();
984 status = xprt->ops->send_request(task);
985 trace_xprt_transmit(xprt, req->rq_xid, status);
986 if (status != 0) {
987 task->tk_status = status;
988 return;
989 }
990 xprt_inject_disconnect(xprt);
991
992 dprintk("RPC: %5u xmit complete\n", task->tk_pid);
993 task->tk_flags |= RPC_TASK_SENT;
994 spin_lock_bh(&xprt->transport_lock);
995
996 xprt->ops->set_retrans_timeout(task);
997
998 numreqs = atomic_read(&xprt->num_reqs);
999 if (numreqs > xprt->stat.max_slots)
1000 xprt->stat.max_slots = numreqs;
1001 xprt->stat.sends++;
1002 xprt->stat.req_u += xprt->stat.sends - xprt->stat.recvs;
1003 xprt->stat.bklog_u += xprt->backlog.qlen;
1004 xprt->stat.sending_u += xprt->sending.qlen;
1005 xprt->stat.pending_u += xprt->pending.qlen;
1006
1007
1008 if (!xprt_connected(xprt))
1009 task->tk_status = -ENOTCONN;
1010 else {
1011
1012
1013
1014
1015 if (!req->rq_reply_bytes_recvd && rpc_reply_expected(task))
1016 rpc_sleep_on(&xprt->pending, task, xprt_timer);
1017 req->rq_connect_cookie = xprt->connect_cookie;
1018 }
1019 spin_unlock_bh(&xprt->transport_lock);
1020}
1021
1022static void xprt_add_backlog(struct rpc_xprt *xprt, struct rpc_task *task)
1023{
1024 set_bit(XPRT_CONGESTED, &xprt->state);
1025 rpc_sleep_on(&xprt->backlog, task, NULL);
1026}
1027
1028static void xprt_wake_up_backlog(struct rpc_xprt *xprt)
1029{
1030 if (rpc_wake_up_next(&xprt->backlog) == NULL)
1031 clear_bit(XPRT_CONGESTED, &xprt->state);
1032}
1033
1034static bool xprt_throttle_congested(struct rpc_xprt *xprt, struct rpc_task *task)
1035{
1036 bool ret = false;
1037
1038 if (!test_bit(XPRT_CONGESTED, &xprt->state))
1039 goto out;
1040 spin_lock(&xprt->reserve_lock);
1041 if (test_bit(XPRT_CONGESTED, &xprt->state)) {
1042 rpc_sleep_on(&xprt->backlog, task, NULL);
1043 ret = true;
1044 }
1045 spin_unlock(&xprt->reserve_lock);
1046out:
1047 return ret;
1048}
1049
1050static struct rpc_rqst *xprt_dynamic_alloc_slot(struct rpc_xprt *xprt)
1051{
1052 struct rpc_rqst *req = ERR_PTR(-EAGAIN);
1053
1054 if (!atomic_add_unless(&xprt->num_reqs, 1, xprt->max_reqs))
1055 goto out;
1056 spin_unlock(&xprt->reserve_lock);
1057 req = kzalloc(sizeof(struct rpc_rqst), GFP_NOFS);
1058 spin_lock(&xprt->reserve_lock);
1059 if (req != NULL)
1060 goto out;
1061 atomic_dec(&xprt->num_reqs);
1062 req = ERR_PTR(-ENOMEM);
1063out:
1064 return req;
1065}
1066
1067static bool xprt_dynamic_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req)
1068{
1069 if (atomic_add_unless(&xprt->num_reqs, -1, xprt->min_reqs)) {
1070 kfree(req);
1071 return true;
1072 }
1073 return false;
1074}
1075
1076void xprt_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task)
1077{
1078 struct rpc_rqst *req;
1079
1080 spin_lock(&xprt->reserve_lock);
1081 if (!list_empty(&xprt->free)) {
1082 req = list_entry(xprt->free.next, struct rpc_rqst, rq_list);
1083 list_del(&req->rq_list);
1084 goto out_init_req;
1085 }
1086 req = xprt_dynamic_alloc_slot(xprt);
1087 if (!IS_ERR(req))
1088 goto out_init_req;
1089 switch (PTR_ERR(req)) {
1090 case -ENOMEM:
1091 dprintk("RPC: dynamic allocation of request slot "
1092 "failed! Retrying\n");
1093 task->tk_status = -ENOMEM;
1094 break;
1095 case -EAGAIN:
1096 xprt_add_backlog(xprt, task);
1097 dprintk("RPC: waiting for request slot\n");
1098 default:
1099 task->tk_status = -EAGAIN;
1100 }
1101 spin_unlock(&xprt->reserve_lock);
1102 return;
1103out_init_req:
1104 task->tk_status = 0;
1105 task->tk_rqstp = req;
1106 xprt_request_init(task, xprt);
1107 spin_unlock(&xprt->reserve_lock);
1108}
1109EXPORT_SYMBOL_GPL(xprt_alloc_slot);
1110
1111void xprt_lock_and_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task)
1112{
1113
1114
1115
1116
1117
1118 if (xprt_lock_write(xprt, task)) {
1119 xprt_alloc_slot(xprt, task);
1120 xprt_release_write(xprt, task);
1121 }
1122}
1123EXPORT_SYMBOL_GPL(xprt_lock_and_alloc_slot);
1124
1125static void xprt_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *req)
1126{
1127 spin_lock(&xprt->reserve_lock);
1128 if (!xprt_dynamic_free_slot(xprt, req)) {
1129 memset(req, 0, sizeof(*req));
1130 list_add(&req->rq_list, &xprt->free);
1131 }
1132 xprt_wake_up_backlog(xprt);
1133 spin_unlock(&xprt->reserve_lock);
1134}
1135
1136static void xprt_free_all_slots(struct rpc_xprt *xprt)
1137{
1138 struct rpc_rqst *req;
1139 while (!list_empty(&xprt->free)) {
1140 req = list_first_entry(&xprt->free, struct rpc_rqst, rq_list);
1141 list_del(&req->rq_list);
1142 kfree(req);
1143 }
1144}
1145
1146struct rpc_xprt *xprt_alloc(struct net *net, size_t size,
1147 unsigned int num_prealloc,
1148 unsigned int max_alloc)
1149{
1150 struct rpc_xprt *xprt;
1151 struct rpc_rqst *req;
1152 int i;
1153
1154 xprt = kzalloc(size, GFP_KERNEL);
1155 if (xprt == NULL)
1156 goto out;
1157
1158 xprt_init(xprt, net);
1159
1160 for (i = 0; i < num_prealloc; i++) {
1161 req = kzalloc(sizeof(struct rpc_rqst), GFP_KERNEL);
1162 if (!req)
1163 goto out_free;
1164 list_add(&req->rq_list, &xprt->free);
1165 }
1166 if (max_alloc > num_prealloc)
1167 xprt->max_reqs = max_alloc;
1168 else
1169 xprt->max_reqs = num_prealloc;
1170 xprt->min_reqs = num_prealloc;
1171 atomic_set(&xprt->num_reqs, num_prealloc);
1172
1173 return xprt;
1174
1175out_free:
1176 xprt_free(xprt);
1177out:
1178 return NULL;
1179}
1180EXPORT_SYMBOL_GPL(xprt_alloc);
1181
1182void xprt_free(struct rpc_xprt *xprt)
1183{
1184 put_net(xprt->xprt_net);
1185 xprt_free_all_slots(xprt);
1186 kfree_rcu(xprt, rcu);
1187}
1188EXPORT_SYMBOL_GPL(xprt_free);
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198void xprt_reserve(struct rpc_task *task)
1199{
1200 struct rpc_xprt *xprt = task->tk_xprt;
1201
1202 task->tk_status = 0;
1203 if (task->tk_rqstp != NULL)
1204 return;
1205
1206 task->tk_timeout = 0;
1207 task->tk_status = -EAGAIN;
1208 if (!xprt_throttle_congested(xprt, task))
1209 xprt->ops->alloc_slot(xprt, task);
1210}
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221void xprt_retry_reserve(struct rpc_task *task)
1222{
1223 struct rpc_xprt *xprt = task->tk_xprt;
1224
1225 task->tk_status = 0;
1226 if (task->tk_rqstp != NULL)
1227 return;
1228
1229 task->tk_timeout = 0;
1230 task->tk_status = -EAGAIN;
1231 xprt->ops->alloc_slot(xprt, task);
1232}
1233
1234static inline __be32 xprt_alloc_xid(struct rpc_xprt *xprt)
1235{
1236 return (__force __be32)xprt->xid++;
1237}
1238
1239static inline void xprt_init_xid(struct rpc_xprt *xprt)
1240{
1241 xprt->xid = prandom_u32();
1242}
1243
1244static void xprt_request_init(struct rpc_task *task, struct rpc_xprt *xprt)
1245{
1246 struct rpc_rqst *req = task->tk_rqstp;
1247
1248 INIT_LIST_HEAD(&req->rq_list);
1249 req->rq_timeout = task->tk_client->cl_timeout->to_initval;
1250 req->rq_task = task;
1251 req->rq_xprt = xprt;
1252 req->rq_buffer = NULL;
1253 req->rq_xid = xprt_alloc_xid(xprt);
1254 req->rq_connect_cookie = xprt->connect_cookie - 1;
1255 req->rq_bytes_sent = 0;
1256 req->rq_snd_buf.len = 0;
1257 req->rq_snd_buf.buflen = 0;
1258 req->rq_rcv_buf.len = 0;
1259 req->rq_rcv_buf.buflen = 0;
1260 req->rq_release_snd_buf = NULL;
1261 xprt_reset_majortimeo(req);
1262 dprintk("RPC: %5u reserved req %p xid %08x\n", task->tk_pid,
1263 req, ntohl(req->rq_xid));
1264}
1265
1266
1267
1268
1269
1270
1271void xprt_release(struct rpc_task *task)
1272{
1273 struct rpc_xprt *xprt;
1274 struct rpc_rqst *req = task->tk_rqstp;
1275
1276 if (req == NULL) {
1277 if (task->tk_client) {
1278 xprt = task->tk_xprt;
1279 if (xprt->snd_task == task)
1280 xprt_release_write(xprt, task);
1281 }
1282 return;
1283 }
1284
1285 xprt = req->rq_xprt;
1286 if (task->tk_ops->rpc_count_stats != NULL)
1287 task->tk_ops->rpc_count_stats(task, task->tk_calldata);
1288 else if (task->tk_client)
1289 rpc_count_iostats(task, task->tk_client->cl_metrics);
1290 spin_lock_bh(&xprt->transport_lock);
1291 xprt->ops->release_xprt(xprt, task);
1292 if (xprt->ops->release_request)
1293 xprt->ops->release_request(task);
1294 if (!list_empty(&req->rq_list))
1295 list_del(&req->rq_list);
1296 xprt->last_used = jiffies;
1297 xprt_schedule_autodisconnect(xprt);
1298 spin_unlock_bh(&xprt->transport_lock);
1299 if (req->rq_buffer)
1300 xprt->ops->buf_free(task);
1301 xprt_inject_disconnect(xprt);
1302 if (req->rq_cred != NULL)
1303 put_rpccred(req->rq_cred);
1304 task->tk_rqstp = NULL;
1305 if (req->rq_release_snd_buf)
1306 req->rq_release_snd_buf(req);
1307
1308 dprintk("RPC: %5u release request %p\n", task->tk_pid, req);
1309 if (likely(!bc_prealloc(req)))
1310 xprt_free_slot(xprt, req);
1311 else
1312 xprt_free_bc_request(req);
1313}
1314
1315static void xprt_init(struct rpc_xprt *xprt, struct net *net)
1316{
1317 kref_init(&xprt->kref);
1318
1319 spin_lock_init(&xprt->transport_lock);
1320 spin_lock_init(&xprt->reserve_lock);
1321
1322 INIT_LIST_HEAD(&xprt->free);
1323 INIT_LIST_HEAD(&xprt->recv);
1324#if defined(CONFIG_SUNRPC_BACKCHANNEL)
1325 spin_lock_init(&xprt->bc_pa_lock);
1326 INIT_LIST_HEAD(&xprt->bc_pa_list);
1327#endif
1328 INIT_LIST_HEAD(&xprt->xprt_switch);
1329
1330 xprt->last_used = jiffies;
1331 xprt->cwnd = RPC_INITCWND;
1332 xprt->bind_index = 0;
1333
1334 rpc_init_wait_queue(&xprt->binding, "xprt_binding");
1335 rpc_init_wait_queue(&xprt->pending, "xprt_pending");
1336 rpc_init_priority_wait_queue(&xprt->sending, "xprt_sending");
1337 rpc_init_priority_wait_queue(&xprt->backlog, "xprt_backlog");
1338
1339 xprt_init_xid(xprt);
1340
1341 xprt->xprt_net = get_net(net);
1342}
1343
1344
1345
1346
1347
1348
1349struct rpc_xprt *xprt_create_transport(struct xprt_create *args)
1350{
1351 struct rpc_xprt *xprt;
1352 struct xprt_class *t;
1353
1354 spin_lock(&xprt_list_lock);
1355 list_for_each_entry(t, &xprt_list, list) {
1356 if (t->ident == args->ident) {
1357 spin_unlock(&xprt_list_lock);
1358 goto found;
1359 }
1360 }
1361 spin_unlock(&xprt_list_lock);
1362 dprintk("RPC: transport (%d) not supported\n", args->ident);
1363 return ERR_PTR(-EIO);
1364
1365found:
1366 xprt = t->setup(args);
1367 if (IS_ERR(xprt)) {
1368 dprintk("RPC: xprt_create_transport: failed, %ld\n",
1369 -PTR_ERR(xprt));
1370 goto out;
1371 }
1372 if (args->flags & XPRT_CREATE_NO_IDLE_TIMEOUT)
1373 xprt->idle_timeout = 0;
1374 INIT_WORK(&xprt->task_cleanup, xprt_autoclose);
1375 if (xprt_has_timer(xprt))
1376 setup_timer(&xprt->timer, xprt_init_autodisconnect,
1377 (unsigned long)xprt);
1378 else
1379 init_timer(&xprt->timer);
1380
1381 if (strlen(args->servername) > RPC_MAXNETNAMELEN) {
1382 xprt_destroy(xprt);
1383 return ERR_PTR(-EINVAL);
1384 }
1385 xprt->servername = kstrdup(args->servername, GFP_KERNEL);
1386 if (xprt->servername == NULL) {
1387 xprt_destroy(xprt);
1388 return ERR_PTR(-ENOMEM);
1389 }
1390
1391 rpc_xprt_debugfs_register(xprt);
1392
1393 dprintk("RPC: created transport %p with %u slots\n", xprt,
1394 xprt->max_reqs);
1395out:
1396 return xprt;
1397}
1398
1399
1400
1401
1402
1403
1404static void xprt_destroy(struct rpc_xprt *xprt)
1405{
1406 dprintk("RPC: destroying transport %p\n", xprt);
1407
1408
1409 wait_on_bit_lock(&xprt->state, XPRT_LOCKED, TASK_UNINTERRUPTIBLE);
1410
1411 del_timer_sync(&xprt->timer);
1412
1413 rpc_xprt_debugfs_unregister(xprt);
1414 rpc_destroy_wait_queue(&xprt->binding);
1415 rpc_destroy_wait_queue(&xprt->pending);
1416 rpc_destroy_wait_queue(&xprt->sending);
1417 rpc_destroy_wait_queue(&xprt->backlog);
1418 cancel_work_sync(&xprt->task_cleanup);
1419 kfree(xprt->servername);
1420
1421
1422
1423 xprt->ops->destroy(xprt);
1424}
1425
1426static void xprt_destroy_kref(struct kref *kref)
1427{
1428 xprt_destroy(container_of(kref, struct rpc_xprt, kref));
1429}
1430
1431
1432
1433
1434
1435
1436struct rpc_xprt *xprt_get(struct rpc_xprt *xprt)
1437{
1438 if (xprt != NULL && kref_get_unless_zero(&xprt->kref))
1439 return xprt;
1440 return NULL;
1441}
1442EXPORT_SYMBOL_GPL(xprt_get);
1443
1444
1445
1446
1447
1448
1449void xprt_put(struct rpc_xprt *xprt)
1450{
1451 if (xprt != NULL)
1452 kref_put(&xprt->kref, xprt_destroy_kref);
1453}
1454EXPORT_SYMBOL_GPL(xprt_put);
1455