2 * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
4 * This software is available to you under a choice of one of two
5 * licenses. You may choose to be licensed under the terms of the GNU
6 * General Public License (GPL) Version 2, available from the file
7 * COPYING in the main directory of this source tree, or the BSD-type
10 * Redistribution and use in source and binary forms, with or without
11 * modification, are permitted provided that the following conditions
14 * Redistributions of source code must retain the above copyright
15 * notice, this list of conditions and the following disclaimer.
17 * Redistributions in binary form must reproduce the above
18 * copyright notice, this list of conditions and the following
19 * disclaimer in the documentation and/or other materials provided
20 * with the distribution.
22 * Neither the name of the Network Appliance, Inc. nor the names of
23 * its contributors may be used to endorse or promote products
24 * derived from this software without specific prior written
27 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
28 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
29 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
30 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
31 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
32 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
33 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
34 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
35 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
36 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
37 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
43 * This file contains the top-level implementation of an RPC RDMA
46 * Naming convention: functions beginning with xprt_ are part of the
47 * transport switch. All others are RPC RDMA internal.
50 #include <linux/module.h>
51 #include <linux/init.h>
52 #include <linux/seq_file.h>
54 #include "xprt_rdma.h"
57 # define RPCDBG_FACILITY RPCDBG_TRANS
60 MODULE_LICENSE("Dual BSD/GPL");
62 MODULE_DESCRIPTION("RPC/RDMA Transport for Linux kernel NFS");
63 MODULE_AUTHOR("Network Appliance, Inc.");
69 static unsigned int xprt_rdma_slot_table_entries
= RPCRDMA_DEF_SLOT_TABLE
;
70 static unsigned int xprt_rdma_max_inline_read
= RPCRDMA_DEF_INLINE
;
71 static unsigned int xprt_rdma_max_inline_write
= RPCRDMA_DEF_INLINE
;
72 static unsigned int xprt_rdma_inline_write_padding
;
73 static unsigned int xprt_rdma_memreg_strategy
= RPCRDMA_FRMR
;
74 int xprt_rdma_pad_optimize
= 0;
78 static unsigned int min_slot_table_size
= RPCRDMA_MIN_SLOT_TABLE
;
79 static unsigned int max_slot_table_size
= RPCRDMA_MAX_SLOT_TABLE
;
80 static unsigned int zero
;
81 static unsigned int max_padding
= PAGE_SIZE
;
82 static unsigned int min_memreg
= RPCRDMA_BOUNCEBUFFERS
;
83 static unsigned int max_memreg
= RPCRDMA_LAST
- 1;
85 static struct ctl_table_header
*sunrpc_table_header
;
87 static ctl_table xr_tunables_table
[] = {
89 .ctl_name
= CTL_UNNUMBERED
,
90 .procname
= "rdma_slot_table_entries",
91 .data
= &xprt_rdma_slot_table_entries
,
92 .maxlen
= sizeof(unsigned int),
94 .proc_handler
= &proc_dointvec_minmax
,
95 .strategy
= &sysctl_intvec
,
96 .extra1
= &min_slot_table_size
,
97 .extra2
= &max_slot_table_size
100 .ctl_name
= CTL_UNNUMBERED
,
101 .procname
= "rdma_max_inline_read",
102 .data
= &xprt_rdma_max_inline_read
,
103 .maxlen
= sizeof(unsigned int),
105 .proc_handler
= &proc_dointvec
,
106 .strategy
= &sysctl_intvec
,
109 .ctl_name
= CTL_UNNUMBERED
,
110 .procname
= "rdma_max_inline_write",
111 .data
= &xprt_rdma_max_inline_write
,
112 .maxlen
= sizeof(unsigned int),
114 .proc_handler
= &proc_dointvec
,
115 .strategy
= &sysctl_intvec
,
118 .ctl_name
= CTL_UNNUMBERED
,
119 .procname
= "rdma_inline_write_padding",
120 .data
= &xprt_rdma_inline_write_padding
,
121 .maxlen
= sizeof(unsigned int),
123 .proc_handler
= &proc_dointvec_minmax
,
124 .strategy
= &sysctl_intvec
,
126 .extra2
= &max_padding
,
129 .ctl_name
= CTL_UNNUMBERED
,
130 .procname
= "rdma_memreg_strategy",
131 .data
= &xprt_rdma_memreg_strategy
,
132 .maxlen
= sizeof(unsigned int),
134 .proc_handler
= &proc_dointvec_minmax
,
135 .strategy
= &sysctl_intvec
,
136 .extra1
= &min_memreg
,
137 .extra2
= &max_memreg
,
140 .ctl_name
= CTL_UNNUMBERED
,
141 .procname
= "rdma_pad_optimize",
142 .data
= &xprt_rdma_pad_optimize
,
143 .maxlen
= sizeof(unsigned int),
145 .proc_handler
= &proc_dointvec
,
152 static ctl_table sunrpc_table
[] = {
154 .ctl_name
= CTL_SUNRPC
,
155 .procname
= "sunrpc",
157 .child
= xr_tunables_table
166 static struct rpc_xprt_ops xprt_rdma_procs
; /* forward reference */
169 xprt_rdma_format_addresses(struct rpc_xprt
*xprt
)
171 struct sockaddr_in
*addr
= (struct sockaddr_in
*)
172 &rpcx_to_rdmad(xprt
).addr
;
175 buf
= kzalloc(20, GFP_KERNEL
);
177 snprintf(buf
, 20, "%pI4", &addr
->sin_addr
.s_addr
);
178 xprt
->address_strings
[RPC_DISPLAY_ADDR
] = buf
;
180 buf
= kzalloc(8, GFP_KERNEL
);
182 snprintf(buf
, 8, "%u", ntohs(addr
->sin_port
));
183 xprt
->address_strings
[RPC_DISPLAY_PORT
] = buf
;
185 xprt
->address_strings
[RPC_DISPLAY_PROTO
] = "rdma";
187 buf
= kzalloc(48, GFP_KERNEL
);
189 snprintf(buf
, 48, "addr=%pI4 port=%u proto=%s",
190 &addr
->sin_addr
.s_addr
,
191 ntohs(addr
->sin_port
), "rdma");
192 xprt
->address_strings
[RPC_DISPLAY_ALL
] = buf
;
194 buf
= kzalloc(10, GFP_KERNEL
);
196 snprintf(buf
, 10, "%02x%02x%02x%02x",
197 NIPQUAD(addr
->sin_addr
.s_addr
));
198 xprt
->address_strings
[RPC_DISPLAY_HEX_ADDR
] = buf
;
200 buf
= kzalloc(8, GFP_KERNEL
);
202 snprintf(buf
, 8, "%4hx", ntohs(addr
->sin_port
));
203 xprt
->address_strings
[RPC_DISPLAY_HEX_PORT
] = buf
;
206 xprt
->address_strings
[RPC_DISPLAY_NETID
] = "rdma";
210 xprt_rdma_free_addresses(struct rpc_xprt
*xprt
)
214 for (i
= 0; i
< RPC_DISPLAY_MAX
; i
++)
216 case RPC_DISPLAY_PROTO
:
217 case RPC_DISPLAY_NETID
:
220 kfree(xprt
->address_strings
[i
]);
225 xprt_rdma_connect_worker(struct work_struct
*work
)
227 struct rpcrdma_xprt
*r_xprt
=
228 container_of(work
, struct rpcrdma_xprt
, rdma_connect
.work
);
229 struct rpc_xprt
*xprt
= &r_xprt
->xprt
;
232 if (!xprt
->shutdown
) {
233 xprt_clear_connected(xprt
);
235 dprintk("RPC: %s: %sconnect\n", __func__
,
236 r_xprt
->rx_ep
.rep_connected
!= 0 ? "re" : "");
237 rc
= rpcrdma_ep_connect(&r_xprt
->rx_ep
, &r_xprt
->rx_ia
);
244 xprt_wake_pending_tasks(xprt
, rc
);
247 dprintk("RPC: %s: exit\n", __func__
);
248 xprt_clear_connecting(xprt
);
255 * Free all memory associated with the object, including its own.
256 * NOTE: none of the *destroy methods free memory for their top-level
257 * objects, even though they may have allocated it (they do free
258 * private memory). It's up to the caller to handle it. In this
259 * case (RDMA transport), all structure memory is inlined with the
260 * struct rpcrdma_xprt.
263 xprt_rdma_destroy(struct rpc_xprt
*xprt
)
265 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
268 dprintk("RPC: %s: called\n", __func__
);
270 cancel_delayed_work(&r_xprt
->rdma_connect
);
271 flush_scheduled_work();
273 xprt_clear_connected(xprt
);
275 rpcrdma_buffer_destroy(&r_xprt
->rx_buf
);
276 rc
= rpcrdma_ep_destroy(&r_xprt
->rx_ep
, &r_xprt
->rx_ia
);
278 dprintk("RPC: %s: rpcrdma_ep_destroy returned %i\n",
280 rpcrdma_ia_close(&r_xprt
->rx_ia
);
282 xprt_rdma_free_addresses(xprt
);
288 dprintk("RPC: %s: returning\n", __func__
);
290 module_put(THIS_MODULE
);
293 static const struct rpc_timeout xprt_rdma_default_timeout
= {
294 .to_initval
= 60 * HZ
,
295 .to_maxval
= 60 * HZ
,
299 * xprt_setup_rdma - Set up transport to use RDMA
301 * @args: rpc transport arguments
303 static struct rpc_xprt
*
304 xprt_setup_rdma(struct xprt_create
*args
)
306 struct rpcrdma_create_data_internal cdata
;
307 struct rpc_xprt
*xprt
;
308 struct rpcrdma_xprt
*new_xprt
;
309 struct rpcrdma_ep
*new_ep
;
310 struct sockaddr_in
*sin
;
313 if (args
->addrlen
> sizeof(xprt
->addr
)) {
314 dprintk("RPC: %s: address too large\n", __func__
);
315 return ERR_PTR(-EBADF
);
318 xprt
= kzalloc(sizeof(struct rpcrdma_xprt
), GFP_KERNEL
);
320 dprintk("RPC: %s: couldn't allocate rpcrdma_xprt\n",
322 return ERR_PTR(-ENOMEM
);
325 xprt
->max_reqs
= xprt_rdma_slot_table_entries
;
326 xprt
->slot
= kcalloc(xprt
->max_reqs
,
327 sizeof(struct rpc_rqst
), GFP_KERNEL
);
328 if (xprt
->slot
== NULL
) {
329 dprintk("RPC: %s: couldn't allocate %d slots\n",
330 __func__
, xprt
->max_reqs
);
332 return ERR_PTR(-ENOMEM
);
335 /* 60 second timeout, no retries */
336 xprt
->timeout
= &xprt_rdma_default_timeout
;
337 xprt
->bind_timeout
= (60U * HZ
);
338 xprt
->connect_timeout
= (60U * HZ
);
339 xprt
->reestablish_timeout
= (5U * HZ
);
340 xprt
->idle_timeout
= (5U * 60 * HZ
);
342 xprt
->resvport
= 0; /* privileged port not needed */
343 xprt
->tsh_size
= 0; /* RPC-RDMA handles framing */
344 xprt
->max_payload
= RPCRDMA_MAX_DATA_SEGS
* PAGE_SIZE
;
345 xprt
->ops
= &xprt_rdma_procs
;
348 * Set up RDMA-specific connect data.
351 /* Put server RDMA address in local cdata */
352 memcpy(&cdata
.addr
, args
->dstaddr
, args
->addrlen
);
354 /* Ensure xprt->addr holds valid server TCP (not RDMA)
355 * address, for any side protocols which peek at it */
356 xprt
->prot
= IPPROTO_TCP
;
357 xprt
->addrlen
= args
->addrlen
;
358 memcpy(&xprt
->addr
, &cdata
.addr
, xprt
->addrlen
);
360 sin
= (struct sockaddr_in
*)&cdata
.addr
;
361 if (ntohs(sin
->sin_port
) != 0)
362 xprt_set_bound(xprt
);
364 dprintk("RPC: %s: %pI4:%u\n",
365 __func__
, &sin
->sin_addr
.s_addr
, ntohs(sin
->sin_port
));
367 /* Set max requests */
368 cdata
.max_requests
= xprt
->max_reqs
;
370 /* Set some length limits */
371 cdata
.rsize
= RPCRDMA_MAX_SEGS
* PAGE_SIZE
; /* RDMA write max */
372 cdata
.wsize
= RPCRDMA_MAX_SEGS
* PAGE_SIZE
; /* RDMA read max */
374 cdata
.inline_wsize
= xprt_rdma_max_inline_write
;
375 if (cdata
.inline_wsize
> cdata
.wsize
)
376 cdata
.inline_wsize
= cdata
.wsize
;
378 cdata
.inline_rsize
= xprt_rdma_max_inline_read
;
379 if (cdata
.inline_rsize
> cdata
.rsize
)
380 cdata
.inline_rsize
= cdata
.rsize
;
382 cdata
.padding
= xprt_rdma_inline_write_padding
;
385 * Create new transport instance, which includes initialized
391 new_xprt
= rpcx_to_rdmax(xprt
);
393 rc
= rpcrdma_ia_open(new_xprt
, (struct sockaddr
*) &cdata
.addr
,
394 xprt_rdma_memreg_strategy
);
399 * initialize and create ep
401 new_xprt
->rx_data
= cdata
;
402 new_ep
= &new_xprt
->rx_ep
;
403 new_ep
->rep_remote_addr
= cdata
.addr
;
405 rc
= rpcrdma_ep_create(&new_xprt
->rx_ep
,
406 &new_xprt
->rx_ia
, &new_xprt
->rx_data
);
411 * Allocate pre-registered send and receive buffers for headers and
412 * any inline data. Also specify any padding which will be provided
413 * from a preregistered zero buffer.
415 rc
= rpcrdma_buffer_create(&new_xprt
->rx_buf
, new_ep
, &new_xprt
->rx_ia
,
421 * Register a callback for connection events. This is necessary because
422 * connection loss notification is async. We also catch connection loss
423 * when reaping receives.
425 INIT_DELAYED_WORK(&new_xprt
->rdma_connect
, xprt_rdma_connect_worker
);
426 new_ep
->rep_func
= rpcrdma_conn_func
;
427 new_ep
->rep_xprt
= xprt
;
429 xprt_rdma_format_addresses(xprt
);
431 if (!try_module_get(THIS_MODULE
))
437 xprt_rdma_free_addresses(xprt
);
440 (void) rpcrdma_ep_destroy(new_ep
, &new_xprt
->rx_ia
);
442 rpcrdma_ia_close(&new_xprt
->rx_ia
);
450 * Close a connection, during shutdown or timeout/reconnect
453 xprt_rdma_close(struct rpc_xprt
*xprt
)
455 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
457 dprintk("RPC: %s: closing\n", __func__
);
458 if (r_xprt
->rx_ep
.rep_connected
> 0)
459 xprt
->reestablish_timeout
= 0;
460 xprt_disconnect_done(xprt
);
461 (void) rpcrdma_ep_disconnect(&r_xprt
->rx_ep
, &r_xprt
->rx_ia
);
465 xprt_rdma_set_port(struct rpc_xprt
*xprt
, u16 port
)
467 struct sockaddr_in
*sap
;
469 sap
= (struct sockaddr_in
*)&xprt
->addr
;
470 sap
->sin_port
= htons(port
);
471 sap
= (struct sockaddr_in
*)&rpcx_to_rdmad(xprt
).addr
;
472 sap
->sin_port
= htons(port
);
473 dprintk("RPC: %s: %u\n", __func__
, port
);
477 xprt_rdma_connect(struct rpc_task
*task
)
479 struct rpc_xprt
*xprt
= (struct rpc_xprt
*)task
->tk_xprt
;
480 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
482 if (!xprt_test_and_set_connecting(xprt
)) {
483 if (r_xprt
->rx_ep
.rep_connected
!= 0) {
485 schedule_delayed_work(&r_xprt
->rdma_connect
,
486 xprt
->reestablish_timeout
);
487 xprt
->reestablish_timeout
<<= 1;
488 if (xprt
->reestablish_timeout
> (30 * HZ
))
489 xprt
->reestablish_timeout
= (30 * HZ
);
490 else if (xprt
->reestablish_timeout
< (5 * HZ
))
491 xprt
->reestablish_timeout
= (5 * HZ
);
493 schedule_delayed_work(&r_xprt
->rdma_connect
, 0);
494 if (!RPC_IS_ASYNC(task
))
495 flush_scheduled_work();
501 xprt_rdma_reserve_xprt(struct rpc_task
*task
)
503 struct rpc_xprt
*xprt
= task
->tk_xprt
;
504 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
505 int credits
= atomic_read(&r_xprt
->rx_buf
.rb_credits
);
507 /* == RPC_CWNDSCALE @ init, but *after* setup */
508 if (r_xprt
->rx_buf
.rb_cwndscale
== 0UL) {
509 r_xprt
->rx_buf
.rb_cwndscale
= xprt
->cwnd
;
510 dprintk("RPC: %s: cwndscale %lu\n", __func__
,
511 r_xprt
->rx_buf
.rb_cwndscale
);
512 BUG_ON(r_xprt
->rx_buf
.rb_cwndscale
<= 0);
514 xprt
->cwnd
= credits
* r_xprt
->rx_buf
.rb_cwndscale
;
515 return xprt_reserve_xprt_cong(task
);
519 * The RDMA allocate/free functions need the task structure as a place
520 * to hide the struct rpcrdma_req, which is necessary for the actual send/recv
521 * sequence. For this reason, the recv buffers are attached to send
522 * buffers for portions of the RPC. Note that the RPC layer allocates
523 * both send and receive buffers in the same call. We may register
524 * the receive buffer portion when using reply chunks.
527 xprt_rdma_allocate(struct rpc_task
*task
, size_t size
)
529 struct rpc_xprt
*xprt
= task
->tk_xprt
;
530 struct rpcrdma_req
*req
, *nreq
;
532 req
= rpcrdma_buffer_get(&rpcx_to_rdmax(xprt
)->rx_buf
);
535 if (size
> req
->rl_size
) {
536 dprintk("RPC: %s: size %zd too large for buffer[%zd]: "
537 "prog %d vers %d proc %d\n",
538 __func__
, size
, req
->rl_size
,
539 task
->tk_client
->cl_prog
, task
->tk_client
->cl_vers
,
540 task
->tk_msg
.rpc_proc
->p_proc
);
542 * Outgoing length shortage. Our inline write max must have
543 * been configured to perform direct i/o.
545 * This is therefore a large metadata operation, and the
546 * allocate call was made on the maximum possible message,
547 * e.g. containing long filename(s) or symlink data. In
548 * fact, while these metadata operations *might* carry
549 * large outgoing payloads, they rarely *do*. However, we
550 * have to commit to the request here, so reallocate and
551 * register it now. The data path will never require this
554 * If the allocation or registration fails, the RPC framework
555 * will (doggedly) retry.
557 if (rpcx_to_rdmax(xprt
)->rx_ia
.ri_memreg_strategy
==
558 RPCRDMA_BOUNCEBUFFERS
) {
559 /* forced to "pure inline" */
560 dprintk("RPC: %s: too much data (%zd) for inline "
561 "(r/w max %d/%d)\n", __func__
, size
,
562 rpcx_to_rdmad(xprt
).inline_rsize
,
563 rpcx_to_rdmad(xprt
).inline_wsize
);
565 rpc_exit(task
, -EIO
); /* fail the operation */
566 rpcx_to_rdmax(xprt
)->rx_stats
.failed_marshal_count
++;
569 if (task
->tk_flags
& RPC_TASK_SWAPPER
)
570 nreq
= kmalloc(sizeof *req
+ size
, GFP_ATOMIC
);
572 nreq
= kmalloc(sizeof *req
+ size
, GFP_NOFS
);
576 if (rpcrdma_register_internal(&rpcx_to_rdmax(xprt
)->rx_ia
,
577 nreq
->rl_base
, size
+ sizeof(struct rpcrdma_req
)
578 - offsetof(struct rpcrdma_req
, rl_base
),
579 &nreq
->rl_handle
, &nreq
->rl_iov
)) {
583 rpcx_to_rdmax(xprt
)->rx_stats
.hardway_register_count
+= size
;
584 nreq
->rl_size
= size
;
586 nreq
->rl_nchunks
= 0;
587 nreq
->rl_buffer
= (struct rpcrdma_buffer
*)req
;
588 nreq
->rl_reply
= req
->rl_reply
;
589 memcpy(nreq
->rl_segments
,
590 req
->rl_segments
, sizeof nreq
->rl_segments
);
591 /* flag the swap with an unused field */
592 nreq
->rl_iov
.length
= 0;
593 req
->rl_reply
= NULL
;
596 dprintk("RPC: %s: size %zd, request 0x%p\n", __func__
, size
, req
);
598 req
->rl_connect_cookie
= 0; /* our reserved value */
599 return req
->rl_xdr_buf
;
602 rpcrdma_buffer_put(req
);
603 rpcx_to_rdmax(xprt
)->rx_stats
.failed_marshal_count
++;
608 * This function returns all RDMA resources to the pool.
611 xprt_rdma_free(void *buffer
)
613 struct rpcrdma_req
*req
;
614 struct rpcrdma_xprt
*r_xprt
;
615 struct rpcrdma_rep
*rep
;
621 req
= container_of(buffer
, struct rpcrdma_req
, rl_xdr_buf
[0]);
622 if (req
->rl_iov
.length
== 0) { /* see allocate above */
623 r_xprt
= container_of(((struct rpcrdma_req
*) req
->rl_buffer
)->rl_buffer
,
624 struct rpcrdma_xprt
, rx_buf
);
626 r_xprt
= container_of(req
->rl_buffer
, struct rpcrdma_xprt
, rx_buf
);
629 dprintk("RPC: %s: called on 0x%p%s\n",
630 __func__
, rep
, (rep
&& rep
->rr_func
) ? " (with waiter)" : "");
633 * Finish the deregistration. When using mw bind, this was
634 * begun in rpcrdma_reply_handler(). In all other modes, we
635 * do it here, in thread context. The process is considered
636 * complete when the rr_func vector becomes NULL - this
637 * was put in place during rpcrdma_reply_handler() - the wait
638 * call below will not block if the dereg is "done". If
639 * interrupted, our framework will clean up.
641 for (i
= 0; req
->rl_nchunks
;) {
643 i
+= rpcrdma_deregister_external(
644 &req
->rl_segments
[i
], r_xprt
, NULL
);
647 if (rep
&& wait_event_interruptible(rep
->rr_unbind
, !rep
->rr_func
)) {
648 rep
->rr_func
= NULL
; /* abandon the callback */
649 req
->rl_reply
= NULL
;
652 if (req
->rl_iov
.length
== 0) { /* see allocate above */
653 struct rpcrdma_req
*oreq
= (struct rpcrdma_req
*)req
->rl_buffer
;
654 oreq
->rl_reply
= req
->rl_reply
;
655 (void) rpcrdma_deregister_internal(&r_xprt
->rx_ia
,
662 /* Put back request+reply buffers */
663 rpcrdma_buffer_put(req
);
667 * send_request invokes the meat of RPC RDMA. It must do the following:
668 * 1. Marshal the RPC request into an RPC RDMA request, which means
669 * putting a header in front of data, and creating IOVs for RDMA
670 * from those in the request.
671 * 2. In marshaling, detect opportunities for RDMA, and use them.
672 * 3. Post a recv message to set up asynch completion, then send
673 * the request (rpcrdma_ep_post).
674 * 4. No partial sends are possible in the RPC-RDMA protocol (as in UDP).
678 xprt_rdma_send_request(struct rpc_task
*task
)
680 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
681 struct rpc_xprt
*xprt
= task
->tk_xprt
;
682 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
683 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
685 /* marshal the send itself */
686 if (req
->rl_niovs
== 0 && rpcrdma_marshal_req(rqst
) != 0) {
687 r_xprt
->rx_stats
.failed_marshal_count
++;
688 dprintk("RPC: %s: rpcrdma_marshal_req failed\n",
693 if (req
->rl_reply
== NULL
) /* e.g. reconnection */
694 rpcrdma_recv_buffer_get(req
);
697 req
->rl_reply
->rr_func
= rpcrdma_reply_handler
;
698 /* this need only be done once, but... */
699 req
->rl_reply
->rr_xprt
= xprt
;
702 /* Must suppress retransmit to maintain credits */
703 if (req
->rl_connect_cookie
== xprt
->connect_cookie
)
704 goto drop_connection
;
705 req
->rl_connect_cookie
= xprt
->connect_cookie
;
707 if (rpcrdma_ep_post(&r_xprt
->rx_ia
, &r_xprt
->rx_ep
, req
))
708 goto drop_connection
;
710 task
->tk_bytes_sent
+= rqst
->rq_snd_buf
.len
;
711 rqst
->rq_bytes_sent
= 0;
715 xprt_disconnect_done(xprt
);
716 return -ENOTCONN
; /* implies disconnect */
719 static void xprt_rdma_print_stats(struct rpc_xprt
*xprt
, struct seq_file
*seq
)
721 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
724 if (xprt_connected(xprt
))
725 idle_time
= (long)(jiffies
- xprt
->last_used
) / HZ
;
728 "\txprt:\trdma %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu "
729 "%lu %lu %lu %Lu %Lu %Lu %Lu %lu %lu %lu\n",
731 0, /* need a local port? */
732 xprt
->stat
.bind_count
,
733 xprt
->stat
.connect_count
,
734 xprt
->stat
.connect_time
,
742 r_xprt
->rx_stats
.read_chunk_count
,
743 r_xprt
->rx_stats
.write_chunk_count
,
744 r_xprt
->rx_stats
.reply_chunk_count
,
745 r_xprt
->rx_stats
.total_rdma_request
,
746 r_xprt
->rx_stats
.total_rdma_reply
,
747 r_xprt
->rx_stats
.pullup_copy_count
,
748 r_xprt
->rx_stats
.fixup_copy_count
,
749 r_xprt
->rx_stats
.hardway_register_count
,
750 r_xprt
->rx_stats
.failed_marshal_count
,
751 r_xprt
->rx_stats
.bad_reply_count
);
755 * Plumbing for rpc transport switch and kernel module
758 static struct rpc_xprt_ops xprt_rdma_procs
= {
759 .reserve_xprt
= xprt_rdma_reserve_xprt
,
760 .release_xprt
= xprt_release_xprt_cong
, /* sunrpc/xprt.c */
761 .release_request
= xprt_release_rqst_cong
, /* ditto */
762 .set_retrans_timeout
= xprt_set_retrans_timeout_def
, /* ditto */
763 .rpcbind
= rpcb_getport_async
, /* sunrpc/rpcb_clnt.c */
764 .set_port
= xprt_rdma_set_port
,
765 .connect
= xprt_rdma_connect
,
766 .buf_alloc
= xprt_rdma_allocate
,
767 .buf_free
= xprt_rdma_free
,
768 .send_request
= xprt_rdma_send_request
,
769 .close
= xprt_rdma_close
,
770 .destroy
= xprt_rdma_destroy
,
771 .print_stats
= xprt_rdma_print_stats
774 static struct xprt_class xprt_rdma
= {
775 .list
= LIST_HEAD_INIT(xprt_rdma
.list
),
777 .owner
= THIS_MODULE
,
778 .ident
= XPRT_TRANSPORT_RDMA
,
779 .setup
= xprt_setup_rdma
,
782 static void __exit
xprt_rdma_cleanup(void)
786 dprintk(KERN_INFO
"RPCRDMA Module Removed, deregister RPC RDMA transport\n");
788 if (sunrpc_table_header
) {
789 unregister_sysctl_table(sunrpc_table_header
);
790 sunrpc_table_header
= NULL
;
793 rc
= xprt_unregister_transport(&xprt_rdma
);
795 dprintk("RPC: %s: xprt_unregister returned %i\n",
799 static int __init
xprt_rdma_init(void)
803 rc
= xprt_register_transport(&xprt_rdma
);
808 dprintk(KERN_INFO
"RPCRDMA Module Init, register RPC RDMA transport\n");
810 dprintk(KERN_INFO
"Defaults:\n");
811 dprintk(KERN_INFO
"\tSlots %d\n"
812 "\tMaxInlineRead %d\n\tMaxInlineWrite %d\n",
813 xprt_rdma_slot_table_entries
,
814 xprt_rdma_max_inline_read
, xprt_rdma_max_inline_write
);
815 dprintk(KERN_INFO
"\tPadding %d\n\tMemreg %d\n",
816 xprt_rdma_inline_write_padding
, xprt_rdma_memreg_strategy
);
819 if (!sunrpc_table_header
)
820 sunrpc_table_header
= register_sysctl_table(sunrpc_table
);
825 module_init(xprt_rdma_init
);
826 module_exit(xprt_rdma_cleanup
);