2 * Copyright (c) 2014-2017 Oracle. All rights reserved.
3 * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
5 * This software is available to you under a choice of one of two
6 * licenses. You may choose to be licensed under the terms of the GNU
7 * General Public License (GPL) Version 2, available from the file
8 * COPYING in the main directory of this source tree, or the BSD-type
11 * Redistribution and use in source and binary forms, with or without
12 * modification, are permitted provided that the following conditions
15 * Redistributions of source code must retain the above copyright
16 * notice, this list of conditions and the following disclaimer.
18 * Redistributions in binary form must reproduce the above
19 * copyright notice, this list of conditions and the following
20 * disclaimer in the documentation and/or other materials provided
21 * with the distribution.
23 * Neither the name of the Network Appliance, Inc. nor the names of
24 * its contributors may be used to endorse or promote products
25 * derived from this software without specific prior written
28 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
29 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
30 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
31 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
32 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
33 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
34 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
35 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
36 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
37 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
38 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
44 * This file contains the top-level implementation of an RPC RDMA
47 * Naming convention: functions beginning with xprt_ are part of the
48 * transport switch. All others are RPC RDMA internal.
51 #include <linux/module.h>
52 #include <linux/slab.h>
53 #include <linux/seq_file.h>
54 #include <linux/sunrpc/addr.h>
55 #include <linux/smp.h>
57 #include "xprt_rdma.h"
59 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
60 # define RPCDBG_FACILITY RPCDBG_TRANS
67 static unsigned int xprt_rdma_slot_table_entries
= RPCRDMA_DEF_SLOT_TABLE
;
68 unsigned int xprt_rdma_max_inline_read
= RPCRDMA_DEF_INLINE
;
69 static unsigned int xprt_rdma_max_inline_write
= RPCRDMA_DEF_INLINE
;
70 static unsigned int xprt_rdma_inline_write_padding
;
71 unsigned int xprt_rdma_memreg_strategy
= RPCRDMA_FRMR
;
72 int xprt_rdma_pad_optimize
;
74 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
76 static unsigned int min_slot_table_size
= RPCRDMA_MIN_SLOT_TABLE
;
77 static unsigned int max_slot_table_size
= RPCRDMA_MAX_SLOT_TABLE
;
78 static unsigned int min_inline_size
= RPCRDMA_MIN_INLINE
;
79 static unsigned int max_inline_size
= RPCRDMA_MAX_INLINE
;
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 struct ctl_table xr_tunables_table
[] = {
89 .procname
= "rdma_slot_table_entries",
90 .data
= &xprt_rdma_slot_table_entries
,
91 .maxlen
= sizeof(unsigned int),
93 .proc_handler
= proc_dointvec_minmax
,
94 .extra1
= &min_slot_table_size
,
95 .extra2
= &max_slot_table_size
98 .procname
= "rdma_max_inline_read",
99 .data
= &xprt_rdma_max_inline_read
,
100 .maxlen
= sizeof(unsigned int),
102 .proc_handler
= proc_dointvec_minmax
,
103 .extra1
= &min_inline_size
,
104 .extra2
= &max_inline_size
,
107 .procname
= "rdma_max_inline_write",
108 .data
= &xprt_rdma_max_inline_write
,
109 .maxlen
= sizeof(unsigned int),
111 .proc_handler
= proc_dointvec_minmax
,
112 .extra1
= &min_inline_size
,
113 .extra2
= &max_inline_size
,
116 .procname
= "rdma_inline_write_padding",
117 .data
= &xprt_rdma_inline_write_padding
,
118 .maxlen
= sizeof(unsigned int),
120 .proc_handler
= proc_dointvec_minmax
,
122 .extra2
= &max_padding
,
125 .procname
= "rdma_memreg_strategy",
126 .data
= &xprt_rdma_memreg_strategy
,
127 .maxlen
= sizeof(unsigned int),
129 .proc_handler
= proc_dointvec_minmax
,
130 .extra1
= &min_memreg
,
131 .extra2
= &max_memreg
,
134 .procname
= "rdma_pad_optimize",
135 .data
= &xprt_rdma_pad_optimize
,
136 .maxlen
= sizeof(unsigned int),
138 .proc_handler
= proc_dointvec
,
143 static struct ctl_table sunrpc_table
[] = {
145 .procname
= "sunrpc",
147 .child
= xr_tunables_table
154 static const struct rpc_xprt_ops xprt_rdma_procs
;
157 xprt_rdma_format_addresses4(struct rpc_xprt
*xprt
, struct sockaddr
*sap
)
159 struct sockaddr_in
*sin
= (struct sockaddr_in
*)sap
;
162 snprintf(buf
, sizeof(buf
), "%08x", ntohl(sin
->sin_addr
.s_addr
));
163 xprt
->address_strings
[RPC_DISPLAY_HEX_ADDR
] = kstrdup(buf
, GFP_KERNEL
);
165 xprt
->address_strings
[RPC_DISPLAY_NETID
] = RPCBIND_NETID_RDMA
;
169 xprt_rdma_format_addresses6(struct rpc_xprt
*xprt
, struct sockaddr
*sap
)
171 struct sockaddr_in6
*sin6
= (struct sockaddr_in6
*)sap
;
174 snprintf(buf
, sizeof(buf
), "%pi6", &sin6
->sin6_addr
);
175 xprt
->address_strings
[RPC_DISPLAY_HEX_ADDR
] = kstrdup(buf
, GFP_KERNEL
);
177 xprt
->address_strings
[RPC_DISPLAY_NETID
] = RPCBIND_NETID_RDMA6
;
181 xprt_rdma_format_addresses(struct rpc_xprt
*xprt
, struct sockaddr
*sap
)
185 switch (sap
->sa_family
) {
187 xprt_rdma_format_addresses4(xprt
, sap
);
190 xprt_rdma_format_addresses6(xprt
, sap
);
193 pr_err("rpcrdma: Unrecognized address family\n");
197 (void)rpc_ntop(sap
, buf
, sizeof(buf
));
198 xprt
->address_strings
[RPC_DISPLAY_ADDR
] = kstrdup(buf
, GFP_KERNEL
);
200 snprintf(buf
, sizeof(buf
), "%u", rpc_get_port(sap
));
201 xprt
->address_strings
[RPC_DISPLAY_PORT
] = kstrdup(buf
, GFP_KERNEL
);
203 snprintf(buf
, sizeof(buf
), "%4hx", rpc_get_port(sap
));
204 xprt
->address_strings
[RPC_DISPLAY_HEX_PORT
] = kstrdup(buf
, GFP_KERNEL
);
206 xprt
->address_strings
[RPC_DISPLAY_PROTO
] = "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 rpcrdma_conn_func(struct rpcrdma_ep
*ep
)
227 schedule_delayed_work(&ep
->rep_connect_worker
, 0);
231 rpcrdma_connect_worker(struct work_struct
*work
)
233 struct rpcrdma_ep
*ep
=
234 container_of(work
, struct rpcrdma_ep
, rep_connect_worker
.work
);
235 struct rpcrdma_xprt
*r_xprt
=
236 container_of(ep
, struct rpcrdma_xprt
, rx_ep
);
237 struct rpc_xprt
*xprt
= &r_xprt
->rx_xprt
;
239 spin_lock_bh(&xprt
->transport_lock
);
240 if (++xprt
->connect_cookie
== 0) /* maintain a reserved value */
241 ++xprt
->connect_cookie
;
242 if (ep
->rep_connected
> 0) {
243 if (!xprt_test_and_set_connected(xprt
))
244 xprt_wake_pending_tasks(xprt
, 0);
246 if (xprt_test_and_clear_connected(xprt
))
247 xprt_wake_pending_tasks(xprt
, -ENOTCONN
);
249 spin_unlock_bh(&xprt
->transport_lock
);
253 xprt_rdma_connect_worker(struct work_struct
*work
)
255 struct rpcrdma_xprt
*r_xprt
= container_of(work
, struct rpcrdma_xprt
,
256 rx_connect_worker
.work
);
257 struct rpc_xprt
*xprt
= &r_xprt
->rx_xprt
;
260 xprt_clear_connected(xprt
);
262 dprintk("RPC: %s: %sconnect\n", __func__
,
263 r_xprt
->rx_ep
.rep_connected
!= 0 ? "re" : "");
264 rc
= rpcrdma_ep_connect(&r_xprt
->rx_ep
, &r_xprt
->rx_ia
);
266 xprt_wake_pending_tasks(xprt
, rc
);
268 dprintk("RPC: %s: exit\n", __func__
);
269 xprt_clear_connecting(xprt
);
273 xprt_rdma_inject_disconnect(struct rpc_xprt
*xprt
)
275 struct rpcrdma_xprt
*r_xprt
= container_of(xprt
, struct rpcrdma_xprt
,
278 pr_info("rpcrdma: injecting transport disconnect on xprt=%p\n", xprt
);
279 rdma_disconnect(r_xprt
->rx_ia
.ri_id
);
286 * Free all memory associated with the object, including its own.
287 * NOTE: none of the *destroy methods free memory for their top-level
288 * objects, even though they may have allocated it (they do free
289 * private memory). It's up to the caller to handle it. In this
290 * case (RDMA transport), all structure memory is inlined with the
291 * struct rpcrdma_xprt.
294 xprt_rdma_destroy(struct rpc_xprt
*xprt
)
296 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
298 dprintk("RPC: %s: called\n", __func__
);
300 cancel_delayed_work_sync(&r_xprt
->rx_connect_worker
);
302 xprt_clear_connected(xprt
);
304 rpcrdma_ep_destroy(&r_xprt
->rx_ep
, &r_xprt
->rx_ia
);
305 rpcrdma_buffer_destroy(&r_xprt
->rx_buf
);
306 rpcrdma_ia_close(&r_xprt
->rx_ia
);
308 xprt_rdma_free_addresses(xprt
);
312 dprintk("RPC: %s: returning\n", __func__
);
314 module_put(THIS_MODULE
);
317 static const struct rpc_timeout xprt_rdma_default_timeout
= {
318 .to_initval
= 60 * HZ
,
319 .to_maxval
= 60 * HZ
,
323 * xprt_setup_rdma - Set up transport to use RDMA
325 * @args: rpc transport arguments
327 static struct rpc_xprt
*
328 xprt_setup_rdma(struct xprt_create
*args
)
330 struct rpcrdma_create_data_internal cdata
;
331 struct rpc_xprt
*xprt
;
332 struct rpcrdma_xprt
*new_xprt
;
333 struct rpcrdma_ep
*new_ep
;
334 struct sockaddr
*sap
;
337 if (args
->addrlen
> sizeof(xprt
->addr
)) {
338 dprintk("RPC: %s: address too large\n", __func__
);
339 return ERR_PTR(-EBADF
);
342 xprt
= xprt_alloc(args
->net
, sizeof(struct rpcrdma_xprt
),
343 xprt_rdma_slot_table_entries
,
344 xprt_rdma_slot_table_entries
);
346 dprintk("RPC: %s: couldn't allocate rpcrdma_xprt\n",
348 return ERR_PTR(-ENOMEM
);
351 /* 60 second timeout, no retries */
352 xprt
->timeout
= &xprt_rdma_default_timeout
;
353 xprt
->bind_timeout
= RPCRDMA_BIND_TO
;
354 xprt
->reestablish_timeout
= RPCRDMA_INIT_REEST_TO
;
355 xprt
->idle_timeout
= RPCRDMA_IDLE_DISC_TO
;
357 xprt
->resvport
= 0; /* privileged port not needed */
358 xprt
->tsh_size
= 0; /* RPC-RDMA handles framing */
359 xprt
->ops
= &xprt_rdma_procs
;
362 * Set up RDMA-specific connect data.
365 sap
= (struct sockaddr
*)&cdata
.addr
;
366 memcpy(sap
, args
->dstaddr
, args
->addrlen
);
368 /* Ensure xprt->addr holds valid server TCP (not RDMA)
369 * address, for any side protocols which peek at it */
370 xprt
->prot
= IPPROTO_TCP
;
371 xprt
->addrlen
= args
->addrlen
;
372 memcpy(&xprt
->addr
, sap
, xprt
->addrlen
);
374 if (rpc_get_port(sap
))
375 xprt_set_bound(xprt
);
377 cdata
.max_requests
= xprt
->max_reqs
;
379 cdata
.rsize
= RPCRDMA_MAX_SEGS
* PAGE_SIZE
; /* RDMA write max */
380 cdata
.wsize
= RPCRDMA_MAX_SEGS
* PAGE_SIZE
; /* RDMA read max */
382 cdata
.inline_wsize
= xprt_rdma_max_inline_write
;
383 if (cdata
.inline_wsize
> cdata
.wsize
)
384 cdata
.inline_wsize
= cdata
.wsize
;
386 cdata
.inline_rsize
= xprt_rdma_max_inline_read
;
387 if (cdata
.inline_rsize
> cdata
.rsize
)
388 cdata
.inline_rsize
= cdata
.rsize
;
390 cdata
.padding
= xprt_rdma_inline_write_padding
;
393 * Create new transport instance, which includes initialized
399 new_xprt
= rpcx_to_rdmax(xprt
);
401 rc
= rpcrdma_ia_open(new_xprt
, sap
);
406 * initialize and create ep
408 new_xprt
->rx_data
= cdata
;
409 new_ep
= &new_xprt
->rx_ep
;
410 new_ep
->rep_remote_addr
= cdata
.addr
;
412 rc
= rpcrdma_ep_create(&new_xprt
->rx_ep
,
413 &new_xprt
->rx_ia
, &new_xprt
->rx_data
);
418 * Allocate pre-registered send and receive buffers for headers and
419 * any inline data. Also specify any padding which will be provided
420 * from a preregistered zero buffer.
422 rc
= rpcrdma_buffer_create(new_xprt
);
427 * Register a callback for connection events. This is necessary because
428 * connection loss notification is async. We also catch connection loss
429 * when reaping receives.
431 INIT_DELAYED_WORK(&new_xprt
->rx_connect_worker
,
432 xprt_rdma_connect_worker
);
434 xprt_rdma_format_addresses(xprt
, sap
);
435 xprt
->max_payload
= new_xprt
->rx_ia
.ri_ops
->ro_maxpages(new_xprt
);
436 if (xprt
->max_payload
== 0)
438 xprt
->max_payload
<<= PAGE_SHIFT
;
439 dprintk("RPC: %s: transport data payload maximum: %zu bytes\n",
440 __func__
, xprt
->max_payload
);
442 if (!try_module_get(THIS_MODULE
))
445 dprintk("RPC: %s: %s:%s\n", __func__
,
446 xprt
->address_strings
[RPC_DISPLAY_ADDR
],
447 xprt
->address_strings
[RPC_DISPLAY_PORT
]);
451 xprt_rdma_free_addresses(xprt
);
454 rpcrdma_ep_destroy(new_ep
, &new_xprt
->rx_ia
);
456 rpcrdma_ia_close(&new_xprt
->rx_ia
);
463 * xprt_rdma_close - Close down RDMA connection
464 * @xprt: generic transport to be closed
466 * Called during transport shutdown reconnect, or device
467 * removal. Caller holds the transport's write lock.
470 xprt_rdma_close(struct rpc_xprt
*xprt
)
472 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
473 struct rpcrdma_ep
*ep
= &r_xprt
->rx_ep
;
474 struct rpcrdma_ia
*ia
= &r_xprt
->rx_ia
;
476 dprintk("RPC: %s: closing xprt %p\n", __func__
, xprt
);
478 if (test_and_clear_bit(RPCRDMA_IAF_REMOVING
, &ia
->ri_flags
)) {
479 xprt_clear_connected(xprt
);
480 rpcrdma_ia_remove(ia
);
483 if (ep
->rep_connected
== -ENODEV
)
485 if (ep
->rep_connected
> 0)
486 xprt
->reestablish_timeout
= 0;
487 xprt_disconnect_done(xprt
);
488 rpcrdma_ep_disconnect(ep
, ia
);
492 xprt_rdma_set_port(struct rpc_xprt
*xprt
, u16 port
)
494 struct sockaddr_in
*sap
;
496 sap
= (struct sockaddr_in
*)&xprt
->addr
;
497 sap
->sin_port
= htons(port
);
498 sap
= (struct sockaddr_in
*)&rpcx_to_rdmad(xprt
).addr
;
499 sap
->sin_port
= htons(port
);
500 dprintk("RPC: %s: %u\n", __func__
, port
);
504 * xprt_rdma_timer - invoked when an RPC times out
505 * @xprt: controlling RPC transport
506 * @task: RPC task that timed out
508 * Invoked when the transport is still connected, but an RPC
509 * retransmit timeout occurs.
511 * Since RDMA connections don't have a keep-alive, forcibly
512 * disconnect and retry to connect. This drives full
513 * detection of the network path, and retransmissions of
517 xprt_rdma_timer(struct rpc_xprt
*xprt
, struct rpc_task
*task
)
519 dprintk("RPC: %5u %s: xprt = %p\n", task
->tk_pid
, __func__
, xprt
);
521 xprt_force_disconnect(xprt
);
525 xprt_rdma_connect(struct rpc_xprt
*xprt
, struct rpc_task
*task
)
527 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
529 if (r_xprt
->rx_ep
.rep_connected
!= 0) {
531 schedule_delayed_work(&r_xprt
->rx_connect_worker
,
532 xprt
->reestablish_timeout
);
533 xprt
->reestablish_timeout
<<= 1;
534 if (xprt
->reestablish_timeout
> RPCRDMA_MAX_REEST_TO
)
535 xprt
->reestablish_timeout
= RPCRDMA_MAX_REEST_TO
;
536 else if (xprt
->reestablish_timeout
< RPCRDMA_INIT_REEST_TO
)
537 xprt
->reestablish_timeout
= RPCRDMA_INIT_REEST_TO
;
539 schedule_delayed_work(&r_xprt
->rx_connect_worker
, 0);
540 if (!RPC_IS_ASYNC(task
))
541 flush_delayed_work(&r_xprt
->rx_connect_worker
);
545 /* Allocate a fixed-size buffer in which to construct and send the
546 * RPC-over-RDMA header for this request.
549 rpcrdma_get_rdmabuf(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
552 size_t size
= RPCRDMA_HDRBUF_SIZE
;
553 struct rpcrdma_regbuf
*rb
;
558 rb
= rpcrdma_alloc_regbuf(size
, DMA_TO_DEVICE
, flags
);
562 r_xprt
->rx_stats
.hardway_register_count
+= size
;
563 req
->rl_rdmabuf
= rb
;
564 xdr_buf_init(&req
->rl_hdrbuf
, rb
->rg_base
, rdmab_length(rb
));
569 rpcrdma_get_sendbuf(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
570 size_t size
, gfp_t flags
)
572 struct rpcrdma_regbuf
*rb
;
574 if (req
->rl_sendbuf
&& rdmab_length(req
->rl_sendbuf
) >= size
)
577 rb
= rpcrdma_alloc_regbuf(size
, DMA_TO_DEVICE
, flags
);
581 rpcrdma_free_regbuf(req
->rl_sendbuf
);
582 r_xprt
->rx_stats
.hardway_register_count
+= size
;
583 req
->rl_sendbuf
= rb
;
587 /* The rq_rcv_buf is used only if a Reply chunk is necessary.
588 * The decision to use a Reply chunk is made later in
589 * rpcrdma_marshal_req. This buffer is registered at that time.
591 * Otherwise, the associated RPC Reply arrives in a separate
592 * Receive buffer, arbitrarily chosen by the HCA. The buffer
593 * allocated here for the RPC Reply is not utilized in that
594 * case. See rpcrdma_inline_fixup.
596 * A regbuf is used here to remember the buffer size.
599 rpcrdma_get_recvbuf(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
600 size_t size
, gfp_t flags
)
602 struct rpcrdma_regbuf
*rb
;
604 if (req
->rl_recvbuf
&& rdmab_length(req
->rl_recvbuf
) >= size
)
607 rb
= rpcrdma_alloc_regbuf(size
, DMA_NONE
, flags
);
611 rpcrdma_free_regbuf(req
->rl_recvbuf
);
612 r_xprt
->rx_stats
.hardway_register_count
+= size
;
613 req
->rl_recvbuf
= rb
;
618 * xprt_rdma_allocate - allocate transport resources for an RPC
622 * 0: Success; rq_buffer points to RPC buffer to use
623 * ENOMEM: Out of memory, call again later
624 * EIO: A permanent error occurred, do not retry
626 * The RDMA allocate/free functions need the task structure as a place
627 * to hide the struct rpcrdma_req, which is necessary for the actual
628 * send/recv sequence.
630 * xprt_rdma_allocate provides buffers that are already mapped for
631 * DMA, and a local DMA lkey is provided for each.
634 xprt_rdma_allocate(struct rpc_task
*task
)
636 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
637 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(rqst
->rq_xprt
);
638 struct rpcrdma_req
*req
;
641 req
= rpcrdma_buffer_get(&r_xprt
->rx_buf
);
645 flags
= RPCRDMA_DEF_GFP
;
646 if (RPC_IS_SWAPPER(task
))
647 flags
= __GFP_MEMALLOC
| GFP_NOWAIT
| __GFP_NOWARN
;
649 if (!rpcrdma_get_rdmabuf(r_xprt
, req
, flags
))
651 if (!rpcrdma_get_sendbuf(r_xprt
, req
, rqst
->rq_callsize
, flags
))
653 if (!rpcrdma_get_recvbuf(r_xprt
, req
, rqst
->rq_rcvsize
, flags
))
656 dprintk("RPC: %5u %s: send size = %zd, recv size = %zd, req = %p\n",
657 task
->tk_pid
, __func__
, rqst
->rq_callsize
,
658 rqst
->rq_rcvsize
, req
);
660 req
->rl_cpu
= smp_processor_id();
661 req
->rl_connect_cookie
= 0; /* our reserved value */
662 rpcrdma_set_xprtdata(rqst
, req
);
663 rqst
->rq_buffer
= req
->rl_sendbuf
->rg_base
;
664 rqst
->rq_rbuffer
= req
->rl_recvbuf
->rg_base
;
668 rpcrdma_buffer_put(req
);
673 * xprt_rdma_free - release resources allocated by xprt_rdma_allocate
676 * Caller guarantees rqst->rq_buffer is non-NULL.
679 xprt_rdma_free(struct rpc_task
*task
)
681 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
682 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(rqst
->rq_xprt
);
683 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
685 if (test_bit(RPCRDMA_REQ_F_BACKCHANNEL
, &req
->rl_flags
))
688 dprintk("RPC: %s: called on 0x%p\n", __func__
, req
->rl_reply
);
690 if (test_bit(RPCRDMA_REQ_F_PENDING
, &req
->rl_flags
))
691 rpcrdma_release_rqst(r_xprt
, req
);
692 rpcrdma_buffer_put(req
);
696 * xprt_rdma_send_request - marshal and send an RPC request
697 * @task: RPC task with an RPC message in rq_snd_buf
699 * Caller holds the transport's write lock.
702 * 0: The request has been sent
703 * ENOTCONN: Caller needs to invoke connect logic then call again
704 * ENOBUFS: Call again later to send the request
705 * EIO: A permanent error occurred. The request was not sent,
706 * and don't try it again
708 * send_request invokes the meat of RPC RDMA. It must do the following:
710 * 1. Marshal the RPC request into an RPC RDMA request, which means
711 * putting a header in front of data, and creating IOVs for RDMA
712 * from those in the request.
713 * 2. In marshaling, detect opportunities for RDMA, and use them.
714 * 3. Post a recv message to set up asynch completion, then send
715 * the request (rpcrdma_ep_post).
716 * 4. No partial sends are possible in the RPC-RDMA protocol (as in UDP).
719 xprt_rdma_send_request(struct rpc_task
*task
)
721 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
722 struct rpc_xprt
*xprt
= rqst
->rq_xprt
;
723 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
724 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
727 if (!xprt_connected(xprt
))
728 goto drop_connection
;
730 /* On retransmit, remove any previously registered chunks */
731 if (unlikely(!list_empty(&req
->rl_registered
)))
732 r_xprt
->rx_ia
.ri_ops
->ro_unmap_sync(r_xprt
,
733 &req
->rl_registered
);
735 rc
= rpcrdma_marshal_req(r_xprt
, rqst
);
739 if (req
->rl_reply
== NULL
) /* e.g. reconnection */
740 rpcrdma_recv_buffer_get(req
);
742 /* Must suppress retransmit to maintain credits */
743 if (req
->rl_connect_cookie
== xprt
->connect_cookie
)
744 goto drop_connection
;
745 req
->rl_connect_cookie
= xprt
->connect_cookie
;
747 set_bit(RPCRDMA_REQ_F_PENDING
, &req
->rl_flags
);
748 if (rpcrdma_ep_post(&r_xprt
->rx_ia
, &r_xprt
->rx_ep
, req
))
749 goto drop_connection
;
751 rqst
->rq_xmit_bytes_sent
+= rqst
->rq_snd_buf
.len
;
752 rqst
->rq_bytes_sent
= 0;
759 xprt_disconnect_done(xprt
);
760 return -ENOTCONN
; /* implies disconnect */
763 void xprt_rdma_print_stats(struct rpc_xprt
*xprt
, struct seq_file
*seq
)
765 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
768 if (xprt_connected(xprt
))
769 idle_time
= (long)(jiffies
- xprt
->last_used
) / HZ
;
771 seq_puts(seq
, "\txprt:\trdma ");
772 seq_printf(seq
, "%u %lu %lu %lu %ld %lu %lu %lu %llu %llu ",
773 0, /* need a local port? */
774 xprt
->stat
.bind_count
,
775 xprt
->stat
.connect_count
,
776 xprt
->stat
.connect_time
,
783 seq_printf(seq
, "%lu %lu %lu %llu %llu %llu %llu %lu %lu %lu %lu ",
784 r_xprt
->rx_stats
.read_chunk_count
,
785 r_xprt
->rx_stats
.write_chunk_count
,
786 r_xprt
->rx_stats
.reply_chunk_count
,
787 r_xprt
->rx_stats
.total_rdma_request
,
788 r_xprt
->rx_stats
.total_rdma_reply
,
789 r_xprt
->rx_stats
.pullup_copy_count
,
790 r_xprt
->rx_stats
.fixup_copy_count
,
791 r_xprt
->rx_stats
.hardway_register_count
,
792 r_xprt
->rx_stats
.failed_marshal_count
,
793 r_xprt
->rx_stats
.bad_reply_count
,
794 r_xprt
->rx_stats
.nomsg_call_count
);
795 seq_printf(seq
, "%lu %lu %lu %lu %lu %lu\n",
796 r_xprt
->rx_stats
.mrs_recovered
,
797 r_xprt
->rx_stats
.mrs_orphaned
,
798 r_xprt
->rx_stats
.mrs_allocated
,
799 r_xprt
->rx_stats
.local_inv_needed
,
800 r_xprt
->rx_stats
.empty_sendctx_q
,
801 r_xprt
->rx_stats
.reply_waits_for_send
);
805 xprt_rdma_enable_swap(struct rpc_xprt
*xprt
)
811 xprt_rdma_disable_swap(struct rpc_xprt
*xprt
)
816 * Plumbing for rpc transport switch and kernel module
819 static const struct rpc_xprt_ops xprt_rdma_procs
= {
820 .reserve_xprt
= xprt_reserve_xprt_cong
,
821 .release_xprt
= xprt_release_xprt_cong
, /* sunrpc/xprt.c */
822 .alloc_slot
= xprt_alloc_slot
,
823 .release_request
= xprt_release_rqst_cong
, /* ditto */
824 .set_retrans_timeout
= xprt_set_retrans_timeout_def
, /* ditto */
825 .timer
= xprt_rdma_timer
,
826 .rpcbind
= rpcb_getport_async
, /* sunrpc/rpcb_clnt.c */
827 .set_port
= xprt_rdma_set_port
,
828 .connect
= xprt_rdma_connect
,
829 .buf_alloc
= xprt_rdma_allocate
,
830 .buf_free
= xprt_rdma_free
,
831 .send_request
= xprt_rdma_send_request
,
832 .close
= xprt_rdma_close
,
833 .destroy
= xprt_rdma_destroy
,
834 .print_stats
= xprt_rdma_print_stats
,
835 .enable_swap
= xprt_rdma_enable_swap
,
836 .disable_swap
= xprt_rdma_disable_swap
,
837 .inject_disconnect
= xprt_rdma_inject_disconnect
,
838 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
839 .bc_setup
= xprt_rdma_bc_setup
,
840 .bc_up
= xprt_rdma_bc_up
,
841 .bc_maxpayload
= xprt_rdma_bc_maxpayload
,
842 .bc_free_rqst
= xprt_rdma_bc_free_rqst
,
843 .bc_destroy
= xprt_rdma_bc_destroy
,
847 static struct xprt_class xprt_rdma
= {
848 .list
= LIST_HEAD_INIT(xprt_rdma
.list
),
850 .owner
= THIS_MODULE
,
851 .ident
= XPRT_TRANSPORT_RDMA
,
852 .setup
= xprt_setup_rdma
,
855 void xprt_rdma_cleanup(void)
859 dprintk("RPCRDMA Module Removed, deregister RPC RDMA transport\n");
860 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
861 if (sunrpc_table_header
) {
862 unregister_sysctl_table(sunrpc_table_header
);
863 sunrpc_table_header
= NULL
;
866 rc
= xprt_unregister_transport(&xprt_rdma
);
868 dprintk("RPC: %s: xprt_unregister returned %i\n",
871 rpcrdma_destroy_wq();
873 rc
= xprt_unregister_transport(&xprt_rdma_bc
);
875 dprintk("RPC: %s: xprt_unregister(bc) returned %i\n",
879 int xprt_rdma_init(void)
883 rc
= rpcrdma_alloc_wq();
887 rc
= xprt_register_transport(&xprt_rdma
);
889 rpcrdma_destroy_wq();
893 rc
= xprt_register_transport(&xprt_rdma_bc
);
895 xprt_unregister_transport(&xprt_rdma
);
896 rpcrdma_destroy_wq();
900 dprintk("RPCRDMA Module Init, register RPC RDMA transport\n");
902 dprintk("Defaults:\n");
903 dprintk("\tSlots %d\n"
904 "\tMaxInlineRead %d\n\tMaxInlineWrite %d\n",
905 xprt_rdma_slot_table_entries
,
906 xprt_rdma_max_inline_read
, xprt_rdma_max_inline_write
);
907 dprintk("\tPadding %d\n\tMemreg %d\n",
908 xprt_rdma_inline_write_padding
, xprt_rdma_memreg_strategy
);
910 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
911 if (!sunrpc_table_header
)
912 sunrpc_table_header
= register_sysctl_table(sunrpc_table
);