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