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>
56 #include "xprt_rdma.h"
58 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
59 # define RPCDBG_FACILITY RPCDBG_TRANS
66 static unsigned int xprt_rdma_slot_table_entries
= RPCRDMA_DEF_SLOT_TABLE
;
67 unsigned int xprt_rdma_max_inline_read
= RPCRDMA_DEF_INLINE
;
68 static unsigned int xprt_rdma_max_inline_write
= RPCRDMA_DEF_INLINE
;
69 static unsigned int xprt_rdma_inline_write_padding
;
70 unsigned int xprt_rdma_memreg_strategy
= RPCRDMA_FRMR
;
71 int xprt_rdma_pad_optimize
;
73 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
75 static unsigned int min_slot_table_size
= RPCRDMA_MIN_SLOT_TABLE
;
76 static unsigned int max_slot_table_size
= RPCRDMA_MAX_SLOT_TABLE
;
77 static unsigned int min_inline_size
= RPCRDMA_MIN_INLINE
;
78 static unsigned int max_inline_size
= RPCRDMA_MAX_INLINE
;
79 static unsigned int zero
;
80 static unsigned int max_padding
= PAGE_SIZE
;
81 static unsigned int min_memreg
= RPCRDMA_BOUNCEBUFFERS
;
82 static unsigned int max_memreg
= RPCRDMA_LAST
- 1;
84 static struct ctl_table_header
*sunrpc_table_header
;
86 static struct ctl_table xr_tunables_table
[] = {
88 .procname
= "rdma_slot_table_entries",
89 .data
= &xprt_rdma_slot_table_entries
,
90 .maxlen
= sizeof(unsigned int),
92 .proc_handler
= proc_dointvec_minmax
,
93 .extra1
= &min_slot_table_size
,
94 .extra2
= &max_slot_table_size
97 .procname
= "rdma_max_inline_read",
98 .data
= &xprt_rdma_max_inline_read
,
99 .maxlen
= sizeof(unsigned int),
101 .proc_handler
= proc_dointvec_minmax
,
102 .extra1
= &min_inline_size
,
103 .extra2
= &max_inline_size
,
106 .procname
= "rdma_max_inline_write",
107 .data
= &xprt_rdma_max_inline_write
,
108 .maxlen
= sizeof(unsigned int),
110 .proc_handler
= proc_dointvec_minmax
,
111 .extra1
= &min_inline_size
,
112 .extra2
= &max_inline_size
,
115 .procname
= "rdma_inline_write_padding",
116 .data
= &xprt_rdma_inline_write_padding
,
117 .maxlen
= sizeof(unsigned int),
119 .proc_handler
= proc_dointvec_minmax
,
121 .extra2
= &max_padding
,
124 .procname
= "rdma_memreg_strategy",
125 .data
= &xprt_rdma_memreg_strategy
,
126 .maxlen
= sizeof(unsigned int),
128 .proc_handler
= proc_dointvec_minmax
,
129 .extra1
= &min_memreg
,
130 .extra2
= &max_memreg
,
133 .procname
= "rdma_pad_optimize",
134 .data
= &xprt_rdma_pad_optimize
,
135 .maxlen
= sizeof(unsigned int),
137 .proc_handler
= proc_dointvec
,
142 static struct ctl_table sunrpc_table
[] = {
144 .procname
= "sunrpc",
146 .child
= xr_tunables_table
153 static const struct rpc_xprt_ops xprt_rdma_procs
;
156 xprt_rdma_format_addresses4(struct rpc_xprt
*xprt
, struct sockaddr
*sap
)
158 struct sockaddr_in
*sin
= (struct sockaddr_in
*)sap
;
161 snprintf(buf
, sizeof(buf
), "%08x", ntohl(sin
->sin_addr
.s_addr
));
162 xprt
->address_strings
[RPC_DISPLAY_HEX_ADDR
] = kstrdup(buf
, GFP_KERNEL
);
164 xprt
->address_strings
[RPC_DISPLAY_NETID
] = RPCBIND_NETID_RDMA
;
168 xprt_rdma_format_addresses6(struct rpc_xprt
*xprt
, struct sockaddr
*sap
)
170 struct sockaddr_in6
*sin6
= (struct sockaddr_in6
*)sap
;
173 snprintf(buf
, sizeof(buf
), "%pi6", &sin6
->sin6_addr
);
174 xprt
->address_strings
[RPC_DISPLAY_HEX_ADDR
] = kstrdup(buf
, GFP_KERNEL
);
176 xprt
->address_strings
[RPC_DISPLAY_NETID
] = RPCBIND_NETID_RDMA6
;
180 xprt_rdma_format_addresses(struct rpc_xprt
*xprt
, struct sockaddr
*sap
)
184 switch (sap
->sa_family
) {
186 xprt_rdma_format_addresses4(xprt
, sap
);
189 xprt_rdma_format_addresses6(xprt
, sap
);
192 pr_err("rpcrdma: Unrecognized address family\n");
196 (void)rpc_ntop(sap
, buf
, sizeof(buf
));
197 xprt
->address_strings
[RPC_DISPLAY_ADDR
] = kstrdup(buf
, GFP_KERNEL
);
199 snprintf(buf
, sizeof(buf
), "%u", rpc_get_port(sap
));
200 xprt
->address_strings
[RPC_DISPLAY_PORT
] = kstrdup(buf
, GFP_KERNEL
);
202 snprintf(buf
, sizeof(buf
), "%4hx", rpc_get_port(sap
));
203 xprt
->address_strings
[RPC_DISPLAY_HEX_PORT
] = kstrdup(buf
, GFP_KERNEL
);
205 xprt
->address_strings
[RPC_DISPLAY_PROTO
] = "rdma";
209 xprt_rdma_free_addresses(struct rpc_xprt
*xprt
)
213 for (i
= 0; i
< RPC_DISPLAY_MAX
; i
++)
215 case RPC_DISPLAY_PROTO
:
216 case RPC_DISPLAY_NETID
:
219 kfree(xprt
->address_strings
[i
]);
224 rpcrdma_conn_func(struct rpcrdma_ep
*ep
)
226 schedule_delayed_work(&ep
->rep_connect_worker
, 0);
230 rpcrdma_connect_worker(struct work_struct
*work
)
232 struct rpcrdma_ep
*ep
=
233 container_of(work
, struct rpcrdma_ep
, rep_connect_worker
.work
);
234 struct rpcrdma_xprt
*r_xprt
=
235 container_of(ep
, struct rpcrdma_xprt
, rx_ep
);
236 struct rpc_xprt
*xprt
= &r_xprt
->rx_xprt
;
238 spin_lock_bh(&xprt
->transport_lock
);
239 if (++xprt
->connect_cookie
== 0) /* maintain a reserved value */
240 ++xprt
->connect_cookie
;
241 if (ep
->rep_connected
> 0) {
242 if (!xprt_test_and_set_connected(xprt
))
243 xprt_wake_pending_tasks(xprt
, 0);
245 if (xprt_test_and_clear_connected(xprt
))
246 xprt_wake_pending_tasks(xprt
, -ENOTCONN
);
248 spin_unlock_bh(&xprt
->transport_lock
);
252 xprt_rdma_connect_worker(struct work_struct
*work
)
254 struct rpcrdma_xprt
*r_xprt
= container_of(work
, struct rpcrdma_xprt
,
255 rx_connect_worker
.work
);
256 struct rpc_xprt
*xprt
= &r_xprt
->rx_xprt
;
259 xprt_clear_connected(xprt
);
261 dprintk("RPC: %s: %sconnect\n", __func__
,
262 r_xprt
->rx_ep
.rep_connected
!= 0 ? "re" : "");
263 rc
= rpcrdma_ep_connect(&r_xprt
->rx_ep
, &r_xprt
->rx_ia
);
265 xprt_wake_pending_tasks(xprt
, rc
);
267 dprintk("RPC: %s: exit\n", __func__
);
268 xprt_clear_connecting(xprt
);
272 xprt_rdma_inject_disconnect(struct rpc_xprt
*xprt
)
274 struct rpcrdma_xprt
*r_xprt
= container_of(xprt
, struct rpcrdma_xprt
,
277 pr_info("rpcrdma: injecting transport disconnect on xprt=%p\n", xprt
);
278 rdma_disconnect(r_xprt
->rx_ia
.ri_id
);
285 * Free all memory associated with the object, including its own.
286 * NOTE: none of the *destroy methods free memory for their top-level
287 * objects, even though they may have allocated it (they do free
288 * private memory). It's up to the caller to handle it. In this
289 * case (RDMA transport), all structure memory is inlined with the
290 * struct rpcrdma_xprt.
293 xprt_rdma_destroy(struct rpc_xprt
*xprt
)
295 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
297 dprintk("RPC: %s: called\n", __func__
);
299 cancel_delayed_work_sync(&r_xprt
->rx_connect_worker
);
301 xprt_clear_connected(xprt
);
303 rpcrdma_ep_destroy(&r_xprt
->rx_ep
, &r_xprt
->rx_ia
);
304 rpcrdma_buffer_destroy(&r_xprt
->rx_buf
);
305 rpcrdma_ia_close(&r_xprt
->rx_ia
);
307 xprt_rdma_free_addresses(xprt
);
311 dprintk("RPC: %s: returning\n", __func__
);
313 module_put(THIS_MODULE
);
316 static const struct rpc_timeout xprt_rdma_default_timeout
= {
317 .to_initval
= 60 * HZ
,
318 .to_maxval
= 60 * HZ
,
322 * xprt_setup_rdma - Set up transport to use RDMA
324 * @args: rpc transport arguments
326 static struct rpc_xprt
*
327 xprt_setup_rdma(struct xprt_create
*args
)
329 struct rpcrdma_create_data_internal cdata
;
330 struct rpc_xprt
*xprt
;
331 struct rpcrdma_xprt
*new_xprt
;
332 struct rpcrdma_ep
*new_ep
;
333 struct sockaddr
*sap
;
336 if (args
->addrlen
> sizeof(xprt
->addr
)) {
337 dprintk("RPC: %s: address too large\n", __func__
);
338 return ERR_PTR(-EBADF
);
341 xprt
= xprt_alloc(args
->net
, sizeof(struct rpcrdma_xprt
),
342 xprt_rdma_slot_table_entries
,
343 xprt_rdma_slot_table_entries
);
345 dprintk("RPC: %s: couldn't allocate rpcrdma_xprt\n",
347 return ERR_PTR(-ENOMEM
);
350 /* 60 second timeout, no retries */
351 xprt
->timeout
= &xprt_rdma_default_timeout
;
352 xprt
->bind_timeout
= RPCRDMA_BIND_TO
;
353 xprt
->reestablish_timeout
= RPCRDMA_INIT_REEST_TO
;
354 xprt
->idle_timeout
= RPCRDMA_IDLE_DISC_TO
;
356 xprt
->resvport
= 0; /* privileged port not needed */
357 xprt
->tsh_size
= 0; /* RPC-RDMA handles framing */
358 xprt
->ops
= &xprt_rdma_procs
;
361 * Set up RDMA-specific connect data.
364 sap
= (struct sockaddr
*)&cdata
.addr
;
365 memcpy(sap
, args
->dstaddr
, args
->addrlen
);
367 /* Ensure xprt->addr holds valid server TCP (not RDMA)
368 * address, for any side protocols which peek at it */
369 xprt
->prot
= IPPROTO_TCP
;
370 xprt
->addrlen
= args
->addrlen
;
371 memcpy(&xprt
->addr
, sap
, xprt
->addrlen
);
373 if (rpc_get_port(sap
))
374 xprt_set_bound(xprt
);
376 cdata
.max_requests
= xprt
->max_reqs
;
378 cdata
.rsize
= RPCRDMA_MAX_SEGS
* PAGE_SIZE
; /* RDMA write max */
379 cdata
.wsize
= RPCRDMA_MAX_SEGS
* PAGE_SIZE
; /* RDMA read max */
381 cdata
.inline_wsize
= xprt_rdma_max_inline_write
;
382 if (cdata
.inline_wsize
> cdata
.wsize
)
383 cdata
.inline_wsize
= cdata
.wsize
;
385 cdata
.inline_rsize
= xprt_rdma_max_inline_read
;
386 if (cdata
.inline_rsize
> cdata
.rsize
)
387 cdata
.inline_rsize
= cdata
.rsize
;
389 cdata
.padding
= xprt_rdma_inline_write_padding
;
392 * Create new transport instance, which includes initialized
398 new_xprt
= rpcx_to_rdmax(xprt
);
400 rc
= rpcrdma_ia_open(new_xprt
, sap
);
405 * initialize and create ep
407 new_xprt
->rx_data
= cdata
;
408 new_ep
= &new_xprt
->rx_ep
;
409 new_ep
->rep_remote_addr
= cdata
.addr
;
411 rc
= rpcrdma_ep_create(&new_xprt
->rx_ep
,
412 &new_xprt
->rx_ia
, &new_xprt
->rx_data
);
417 * Allocate pre-registered send and receive buffers for headers and
418 * any inline data. Also specify any padding which will be provided
419 * from a preregistered zero buffer.
421 rc
= rpcrdma_buffer_create(new_xprt
);
426 * Register a callback for connection events. This is necessary because
427 * connection loss notification is async. We also catch connection loss
428 * when reaping receives.
430 INIT_DELAYED_WORK(&new_xprt
->rx_connect_worker
,
431 xprt_rdma_connect_worker
);
433 xprt_rdma_format_addresses(xprt
, sap
);
434 xprt
->max_payload
= new_xprt
->rx_ia
.ri_ops
->ro_maxpages(new_xprt
);
435 if (xprt
->max_payload
== 0)
437 xprt
->max_payload
<<= PAGE_SHIFT
;
438 dprintk("RPC: %s: transport data payload maximum: %zu bytes\n",
439 __func__
, xprt
->max_payload
);
441 if (!try_module_get(THIS_MODULE
))
444 dprintk("RPC: %s: %s:%s\n", __func__
,
445 xprt
->address_strings
[RPC_DISPLAY_ADDR
],
446 xprt
->address_strings
[RPC_DISPLAY_PORT
]);
450 xprt_rdma_free_addresses(xprt
);
453 rpcrdma_ep_destroy(new_ep
, &new_xprt
->rx_ia
);
455 rpcrdma_ia_close(&new_xprt
->rx_ia
);
462 * xprt_rdma_close - Close down RDMA connection
463 * @xprt: generic transport to be closed
465 * Called during transport shutdown reconnect, or device
466 * removal. Caller holds the transport's write lock.
469 xprt_rdma_close(struct rpc_xprt
*xprt
)
471 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
472 struct rpcrdma_ep
*ep
= &r_xprt
->rx_ep
;
473 struct rpcrdma_ia
*ia
= &r_xprt
->rx_ia
;
475 dprintk("RPC: %s: closing xprt %p\n", __func__
, xprt
);
477 if (test_and_clear_bit(RPCRDMA_IAF_REMOVING
, &ia
->ri_flags
)) {
478 xprt_clear_connected(xprt
);
479 rpcrdma_ia_remove(ia
);
482 if (ep
->rep_connected
== -ENODEV
)
484 if (ep
->rep_connected
> 0)
485 xprt
->reestablish_timeout
= 0;
486 xprt_disconnect_done(xprt
);
487 rpcrdma_ep_disconnect(ep
, ia
);
491 xprt_rdma_set_port(struct rpc_xprt
*xprt
, u16 port
)
493 struct sockaddr_in
*sap
;
495 sap
= (struct sockaddr_in
*)&xprt
->addr
;
496 sap
->sin_port
= htons(port
);
497 sap
= (struct sockaddr_in
*)&rpcx_to_rdmad(xprt
).addr
;
498 sap
->sin_port
= htons(port
);
499 dprintk("RPC: %s: %u\n", __func__
, port
);
503 * xprt_rdma_timer - invoked when an RPC times out
504 * @xprt: controlling RPC transport
505 * @task: RPC task that timed out
507 * Invoked when the transport is still connected, but an RPC
508 * retransmit timeout occurs.
510 * Since RDMA connections don't have a keep-alive, forcibly
511 * disconnect and retry to connect. This drives full
512 * detection of the network path, and retransmissions of
516 xprt_rdma_timer(struct rpc_xprt
*xprt
, struct rpc_task
*task
)
518 dprintk("RPC: %5u %s: xprt = %p\n", task
->tk_pid
, __func__
, xprt
);
520 xprt_force_disconnect(xprt
);
524 xprt_rdma_connect(struct rpc_xprt
*xprt
, struct rpc_task
*task
)
526 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
528 if (r_xprt
->rx_ep
.rep_connected
!= 0) {
530 schedule_delayed_work(&r_xprt
->rx_connect_worker
,
531 xprt
->reestablish_timeout
);
532 xprt
->reestablish_timeout
<<= 1;
533 if (xprt
->reestablish_timeout
> RPCRDMA_MAX_REEST_TO
)
534 xprt
->reestablish_timeout
= RPCRDMA_MAX_REEST_TO
;
535 else if (xprt
->reestablish_timeout
< RPCRDMA_INIT_REEST_TO
)
536 xprt
->reestablish_timeout
= RPCRDMA_INIT_REEST_TO
;
538 schedule_delayed_work(&r_xprt
->rx_connect_worker
, 0);
539 if (!RPC_IS_ASYNC(task
))
540 flush_delayed_work(&r_xprt
->rx_connect_worker
);
544 /* Allocate a fixed-size buffer in which to construct and send the
545 * RPC-over-RDMA header for this request.
548 rpcrdma_get_rdmabuf(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
551 size_t size
= RPCRDMA_HDRBUF_SIZE
;
552 struct rpcrdma_regbuf
*rb
;
557 rb
= rpcrdma_alloc_regbuf(size
, DMA_TO_DEVICE
, flags
);
561 r_xprt
->rx_stats
.hardway_register_count
+= size
;
562 req
->rl_rdmabuf
= rb
;
563 xdr_buf_init(&req
->rl_hdrbuf
, rb
->rg_base
, rdmab_length(rb
));
568 rpcrdma_get_sendbuf(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
569 size_t size
, gfp_t flags
)
571 struct rpcrdma_regbuf
*rb
;
573 if (req
->rl_sendbuf
&& rdmab_length(req
->rl_sendbuf
) >= size
)
576 rb
= rpcrdma_alloc_regbuf(size
, DMA_TO_DEVICE
, flags
);
580 rpcrdma_free_regbuf(req
->rl_sendbuf
);
581 r_xprt
->rx_stats
.hardway_register_count
+= size
;
582 req
->rl_sendbuf
= rb
;
586 /* The rq_rcv_buf is used only if a Reply chunk is necessary.
587 * The decision to use a Reply chunk is made later in
588 * rpcrdma_marshal_req. This buffer is registered at that time.
590 * Otherwise, the associated RPC Reply arrives in a separate
591 * Receive buffer, arbitrarily chosen by the HCA. The buffer
592 * allocated here for the RPC Reply is not utilized in that
593 * case. See rpcrdma_inline_fixup.
595 * A regbuf is used here to remember the buffer size.
598 rpcrdma_get_recvbuf(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
599 size_t size
, gfp_t flags
)
601 struct rpcrdma_regbuf
*rb
;
603 if (req
->rl_recvbuf
&& rdmab_length(req
->rl_recvbuf
) >= size
)
606 rb
= rpcrdma_alloc_regbuf(size
, DMA_NONE
, flags
);
610 rpcrdma_free_regbuf(req
->rl_recvbuf
);
611 r_xprt
->rx_stats
.hardway_register_count
+= size
;
612 req
->rl_recvbuf
= rb
;
617 * xprt_rdma_allocate - allocate transport resources for an RPC
621 * 0: Success; rq_buffer points to RPC buffer to use
622 * ENOMEM: Out of memory, call again later
623 * EIO: A permanent error occurred, do not retry
625 * The RDMA allocate/free functions need the task structure as a place
626 * to hide the struct rpcrdma_req, which is necessary for the actual
627 * send/recv sequence.
629 * xprt_rdma_allocate provides buffers that are already mapped for
630 * DMA, and a local DMA lkey is provided for each.
633 xprt_rdma_allocate(struct rpc_task
*task
)
635 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
636 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(rqst
->rq_xprt
);
637 struct rpcrdma_req
*req
;
640 req
= rpcrdma_buffer_get(&r_xprt
->rx_buf
);
644 flags
= RPCRDMA_DEF_GFP
;
645 if (RPC_IS_SWAPPER(task
))
646 flags
= __GFP_MEMALLOC
| GFP_NOWAIT
| __GFP_NOWARN
;
648 if (!rpcrdma_get_rdmabuf(r_xprt
, req
, flags
))
650 if (!rpcrdma_get_sendbuf(r_xprt
, req
, rqst
->rq_callsize
, flags
))
652 if (!rpcrdma_get_recvbuf(r_xprt
, req
, rqst
->rq_rcvsize
, flags
))
655 dprintk("RPC: %5u %s: send size = %zd, recv size = %zd, req = %p\n",
656 task
->tk_pid
, __func__
, rqst
->rq_callsize
,
657 rqst
->rq_rcvsize
, req
);
659 req
->rl_connect_cookie
= 0; /* our reserved value */
660 rpcrdma_set_xprtdata(rqst
, req
);
661 rqst
->rq_buffer
= req
->rl_sendbuf
->rg_base
;
662 rqst
->rq_rbuffer
= req
->rl_recvbuf
->rg_base
;
666 rpcrdma_buffer_put(req
);
671 * xprt_rdma_free - release resources allocated by xprt_rdma_allocate
674 * Caller guarantees rqst->rq_buffer is non-NULL.
677 xprt_rdma_free(struct rpc_task
*task
)
679 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
680 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(rqst
->rq_xprt
);
681 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
683 if (test_bit(RPCRDMA_REQ_F_BACKCHANNEL
, &req
->rl_flags
))
686 dprintk("RPC: %s: called on 0x%p\n", __func__
, req
->rl_reply
);
688 if (test_bit(RPCRDMA_REQ_F_PENDING
, &req
->rl_flags
))
689 rpcrdma_release_rqst(r_xprt
, req
);
690 rpcrdma_buffer_put(req
);
694 * xprt_rdma_send_request - marshal and send an RPC request
695 * @task: RPC task with an RPC message in rq_snd_buf
697 * Caller holds the transport's write lock.
700 * 0: The request has been sent
701 * ENOTCONN: Caller needs to invoke connect logic then call again
702 * ENOBUFS: Call again later to send the request
703 * EIO: A permanent error occurred. The request was not sent,
704 * and don't try it again
706 * send_request invokes the meat of RPC RDMA. It must do the following:
708 * 1. Marshal the RPC request into an RPC RDMA request, which means
709 * putting a header in front of data, and creating IOVs for RDMA
710 * from those in the request.
711 * 2. In marshaling, detect opportunities for RDMA, and use them.
712 * 3. Post a recv message to set up asynch completion, then send
713 * the request (rpcrdma_ep_post).
714 * 4. No partial sends are possible in the RPC-RDMA protocol (as in UDP).
717 xprt_rdma_send_request(struct rpc_task
*task
)
719 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
720 struct rpc_xprt
*xprt
= rqst
->rq_xprt
;
721 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
722 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
725 if (!xprt_connected(xprt
))
726 goto drop_connection
;
728 /* On retransmit, remove any previously registered chunks */
729 if (unlikely(!list_empty(&req
->rl_registered
)))
730 r_xprt
->rx_ia
.ri_ops
->ro_unmap_sync(r_xprt
,
731 &req
->rl_registered
);
733 rc
= rpcrdma_marshal_req(r_xprt
, rqst
);
737 if (req
->rl_reply
== NULL
) /* e.g. reconnection */
738 rpcrdma_recv_buffer_get(req
);
740 /* Must suppress retransmit to maintain credits */
741 if (req
->rl_connect_cookie
== xprt
->connect_cookie
)
742 goto drop_connection
;
743 req
->rl_connect_cookie
= xprt
->connect_cookie
;
745 set_bit(RPCRDMA_REQ_F_PENDING
, &req
->rl_flags
);
746 if (rpcrdma_ep_post(&r_xprt
->rx_ia
, &r_xprt
->rx_ep
, req
))
747 goto drop_connection
;
749 rqst
->rq_xmit_bytes_sent
+= rqst
->rq_snd_buf
.len
;
750 rqst
->rq_bytes_sent
= 0;
757 xprt_disconnect_done(xprt
);
758 return -ENOTCONN
; /* implies disconnect */
761 void xprt_rdma_print_stats(struct rpc_xprt
*xprt
, struct seq_file
*seq
)
763 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
766 if (xprt_connected(xprt
))
767 idle_time
= (long)(jiffies
- xprt
->last_used
) / HZ
;
769 seq_puts(seq
, "\txprt:\trdma ");
770 seq_printf(seq
, "%u %lu %lu %lu %ld %lu %lu %lu %llu %llu ",
771 0, /* need a local port? */
772 xprt
->stat
.bind_count
,
773 xprt
->stat
.connect_count
,
774 xprt
->stat
.connect_time
,
781 seq_printf(seq
, "%lu %lu %lu %llu %llu %llu %llu %lu %lu %lu %lu ",
782 r_xprt
->rx_stats
.read_chunk_count
,
783 r_xprt
->rx_stats
.write_chunk_count
,
784 r_xprt
->rx_stats
.reply_chunk_count
,
785 r_xprt
->rx_stats
.total_rdma_request
,
786 r_xprt
->rx_stats
.total_rdma_reply
,
787 r_xprt
->rx_stats
.pullup_copy_count
,
788 r_xprt
->rx_stats
.fixup_copy_count
,
789 r_xprt
->rx_stats
.hardway_register_count
,
790 r_xprt
->rx_stats
.failed_marshal_count
,
791 r_xprt
->rx_stats
.bad_reply_count
,
792 r_xprt
->rx_stats
.nomsg_call_count
);
793 seq_printf(seq
, "%lu %lu %lu %lu %lu %lu\n",
794 r_xprt
->rx_stats
.mrs_recovered
,
795 r_xprt
->rx_stats
.mrs_orphaned
,
796 r_xprt
->rx_stats
.mrs_allocated
,
797 r_xprt
->rx_stats
.local_inv_needed
,
798 r_xprt
->rx_stats
.empty_sendctx_q
,
799 r_xprt
->rx_stats
.reply_waits_for_send
);
803 xprt_rdma_enable_swap(struct rpc_xprt
*xprt
)
809 xprt_rdma_disable_swap(struct rpc_xprt
*xprt
)
814 * Plumbing for rpc transport switch and kernel module
817 static const struct rpc_xprt_ops xprt_rdma_procs
= {
818 .reserve_xprt
= xprt_reserve_xprt_cong
,
819 .release_xprt
= xprt_release_xprt_cong
, /* sunrpc/xprt.c */
820 .alloc_slot
= xprt_alloc_slot
,
821 .release_request
= xprt_release_rqst_cong
, /* ditto */
822 .set_retrans_timeout
= xprt_set_retrans_timeout_def
, /* ditto */
823 .timer
= xprt_rdma_timer
,
824 .rpcbind
= rpcb_getport_async
, /* sunrpc/rpcb_clnt.c */
825 .set_port
= xprt_rdma_set_port
,
826 .connect
= xprt_rdma_connect
,
827 .buf_alloc
= xprt_rdma_allocate
,
828 .buf_free
= xprt_rdma_free
,
829 .send_request
= xprt_rdma_send_request
,
830 .close
= xprt_rdma_close
,
831 .destroy
= xprt_rdma_destroy
,
832 .print_stats
= xprt_rdma_print_stats
,
833 .enable_swap
= xprt_rdma_enable_swap
,
834 .disable_swap
= xprt_rdma_disable_swap
,
835 .inject_disconnect
= xprt_rdma_inject_disconnect
,
836 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
837 .bc_setup
= xprt_rdma_bc_setup
,
838 .bc_up
= xprt_rdma_bc_up
,
839 .bc_maxpayload
= xprt_rdma_bc_maxpayload
,
840 .bc_free_rqst
= xprt_rdma_bc_free_rqst
,
841 .bc_destroy
= xprt_rdma_bc_destroy
,
845 static struct xprt_class xprt_rdma
= {
846 .list
= LIST_HEAD_INIT(xprt_rdma
.list
),
848 .owner
= THIS_MODULE
,
849 .ident
= XPRT_TRANSPORT_RDMA
,
850 .setup
= xprt_setup_rdma
,
853 void xprt_rdma_cleanup(void)
857 dprintk("RPCRDMA Module Removed, deregister RPC RDMA transport\n");
858 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
859 if (sunrpc_table_header
) {
860 unregister_sysctl_table(sunrpc_table_header
);
861 sunrpc_table_header
= NULL
;
864 rc
= xprt_unregister_transport(&xprt_rdma
);
866 dprintk("RPC: %s: xprt_unregister returned %i\n",
869 rpcrdma_destroy_wq();
871 rc
= xprt_unregister_transport(&xprt_rdma_bc
);
873 dprintk("RPC: %s: xprt_unregister(bc) returned %i\n",
877 int xprt_rdma_init(void)
881 rc
= rpcrdma_alloc_wq();
885 rc
= xprt_register_transport(&xprt_rdma
);
887 rpcrdma_destroy_wq();
891 rc
= xprt_register_transport(&xprt_rdma_bc
);
893 xprt_unregister_transport(&xprt_rdma
);
894 rpcrdma_destroy_wq();
898 dprintk("RPCRDMA Module Init, register RPC RDMA transport\n");
900 dprintk("Defaults:\n");
901 dprintk("\tSlots %d\n"
902 "\tMaxInlineRead %d\n\tMaxInlineWrite %d\n",
903 xprt_rdma_slot_table_entries
,
904 xprt_rdma_max_inline_read
, xprt_rdma_max_inline_write
);
905 dprintk("\tPadding %d\n\tMemreg %d\n",
906 xprt_rdma_inline_write_padding
, xprt_rdma_memreg_strategy
);
908 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
909 if (!sunrpc_table_header
)
910 sunrpc_table_header
= register_sysctl_table(sunrpc_table
);