1 // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
3 * Copyright (c) 2014-2017 Oracle. All rights reserved.
4 * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
6 * This software is available to you under a choice of one of two
7 * licenses. You may choose to be licensed under the terms of the GNU
8 * General Public License (GPL) Version 2, available from the file
9 * COPYING in the main directory of this source tree, or the BSD-type
12 * Redistribution and use in source and binary forms, with or without
13 * modification, are permitted provided that the following conditions
16 * Redistributions of source code must retain the above copyright
17 * notice, this list of conditions and the following disclaimer.
19 * Redistributions in binary form must reproduce the above
20 * copyright notice, this list of conditions and the following
21 * disclaimer in the documentation and/or other materials provided
22 * with the distribution.
24 * Neither the name of the Network Appliance, Inc. nor the names of
25 * its contributors may be used to endorse or promote products
26 * derived from this software without specific prior written
29 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
30 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
31 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
32 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
33 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
34 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
35 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
36 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
37 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
38 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
39 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
45 * This file contains the top-level implementation of an RPC RDMA
48 * Naming convention: functions beginning with xprt_ are part of the
49 * transport switch. All others are RPC RDMA internal.
52 #include <linux/module.h>
53 #include <linux/slab.h>
54 #include <linux/seq_file.h>
55 #include <linux/smp.h>
57 #include <linux/sunrpc/addr.h>
58 #include <linux/sunrpc/svc_rdma.h>
60 #include "xprt_rdma.h"
61 #include <trace/events/rpcrdma.h>
63 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
64 # define RPCDBG_FACILITY RPCDBG_TRANS
71 static unsigned int xprt_rdma_slot_table_entries
= RPCRDMA_DEF_SLOT_TABLE
;
72 unsigned int xprt_rdma_max_inline_read
= RPCRDMA_DEF_INLINE
;
73 static unsigned int xprt_rdma_max_inline_write
= RPCRDMA_DEF_INLINE
;
74 unsigned int xprt_rdma_memreg_strategy
= RPCRDMA_FRWR
;
75 int xprt_rdma_pad_optimize
;
77 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
79 static unsigned int min_slot_table_size
= RPCRDMA_MIN_SLOT_TABLE
;
80 static unsigned int max_slot_table_size
= RPCRDMA_MAX_SLOT_TABLE
;
81 static unsigned int min_inline_size
= RPCRDMA_MIN_INLINE
;
82 static unsigned int max_inline_size
= RPCRDMA_MAX_INLINE
;
83 static unsigned int zero
;
84 static unsigned int max_padding
= PAGE_SIZE
;
85 static unsigned int min_memreg
= RPCRDMA_BOUNCEBUFFERS
;
86 static unsigned int max_memreg
= RPCRDMA_LAST
- 1;
87 static unsigned int dummy
;
89 static struct ctl_table_header
*sunrpc_table_header
;
91 static struct ctl_table xr_tunables_table
[] = {
93 .procname
= "rdma_slot_table_entries",
94 .data
= &xprt_rdma_slot_table_entries
,
95 .maxlen
= sizeof(unsigned int),
97 .proc_handler
= proc_dointvec_minmax
,
98 .extra1
= &min_slot_table_size
,
99 .extra2
= &max_slot_table_size
102 .procname
= "rdma_max_inline_read",
103 .data
= &xprt_rdma_max_inline_read
,
104 .maxlen
= sizeof(unsigned int),
106 .proc_handler
= proc_dointvec_minmax
,
107 .extra1
= &min_inline_size
,
108 .extra2
= &max_inline_size
,
111 .procname
= "rdma_max_inline_write",
112 .data
= &xprt_rdma_max_inline_write
,
113 .maxlen
= sizeof(unsigned int),
115 .proc_handler
= proc_dointvec_minmax
,
116 .extra1
= &min_inline_size
,
117 .extra2
= &max_inline_size
,
120 .procname
= "rdma_inline_write_padding",
122 .maxlen
= sizeof(unsigned int),
124 .proc_handler
= proc_dointvec_minmax
,
126 .extra2
= &max_padding
,
129 .procname
= "rdma_memreg_strategy",
130 .data
= &xprt_rdma_memreg_strategy
,
131 .maxlen
= sizeof(unsigned int),
133 .proc_handler
= proc_dointvec_minmax
,
134 .extra1
= &min_memreg
,
135 .extra2
= &max_memreg
,
138 .procname
= "rdma_pad_optimize",
139 .data
= &xprt_rdma_pad_optimize
,
140 .maxlen
= sizeof(unsigned int),
142 .proc_handler
= proc_dointvec
,
147 static struct ctl_table sunrpc_table
[] = {
149 .procname
= "sunrpc",
151 .child
= xr_tunables_table
158 static const struct rpc_xprt_ops xprt_rdma_procs
;
161 xprt_rdma_format_addresses4(struct rpc_xprt
*xprt
, struct sockaddr
*sap
)
163 struct sockaddr_in
*sin
= (struct sockaddr_in
*)sap
;
166 snprintf(buf
, sizeof(buf
), "%08x", ntohl(sin
->sin_addr
.s_addr
));
167 xprt
->address_strings
[RPC_DISPLAY_HEX_ADDR
] = kstrdup(buf
, GFP_KERNEL
);
169 xprt
->address_strings
[RPC_DISPLAY_NETID
] = RPCBIND_NETID_RDMA
;
173 xprt_rdma_format_addresses6(struct rpc_xprt
*xprt
, struct sockaddr
*sap
)
175 struct sockaddr_in6
*sin6
= (struct sockaddr_in6
*)sap
;
178 snprintf(buf
, sizeof(buf
), "%pi6", &sin6
->sin6_addr
);
179 xprt
->address_strings
[RPC_DISPLAY_HEX_ADDR
] = kstrdup(buf
, GFP_KERNEL
);
181 xprt
->address_strings
[RPC_DISPLAY_NETID
] = RPCBIND_NETID_RDMA6
;
185 xprt_rdma_format_addresses(struct rpc_xprt
*xprt
, struct sockaddr
*sap
)
189 switch (sap
->sa_family
) {
191 xprt_rdma_format_addresses4(xprt
, sap
);
194 xprt_rdma_format_addresses6(xprt
, sap
);
197 pr_err("rpcrdma: Unrecognized address family\n");
201 (void)rpc_ntop(sap
, buf
, sizeof(buf
));
202 xprt
->address_strings
[RPC_DISPLAY_ADDR
] = kstrdup(buf
, GFP_KERNEL
);
204 snprintf(buf
, sizeof(buf
), "%u", rpc_get_port(sap
));
205 xprt
->address_strings
[RPC_DISPLAY_PORT
] = kstrdup(buf
, GFP_KERNEL
);
207 snprintf(buf
, sizeof(buf
), "%4hx", rpc_get_port(sap
));
208 xprt
->address_strings
[RPC_DISPLAY_HEX_PORT
] = kstrdup(buf
, GFP_KERNEL
);
210 xprt
->address_strings
[RPC_DISPLAY_PROTO
] = "rdma";
214 xprt_rdma_free_addresses(struct rpc_xprt
*xprt
)
218 for (i
= 0; i
< RPC_DISPLAY_MAX
; i
++)
220 case RPC_DISPLAY_PROTO
:
221 case RPC_DISPLAY_NETID
:
224 kfree(xprt
->address_strings
[i
]);
229 rpcrdma_conn_func(struct rpcrdma_ep
*ep
)
231 schedule_delayed_work(&ep
->rep_connect_worker
, 0);
235 rpcrdma_connect_worker(struct work_struct
*work
)
237 struct rpcrdma_ep
*ep
=
238 container_of(work
, struct rpcrdma_ep
, rep_connect_worker
.work
);
239 struct rpcrdma_xprt
*r_xprt
=
240 container_of(ep
, struct rpcrdma_xprt
, rx_ep
);
241 struct rpc_xprt
*xprt
= &r_xprt
->rx_xprt
;
243 spin_lock_bh(&xprt
->transport_lock
);
244 if (ep
->rep_connected
> 0) {
245 if (!xprt_test_and_set_connected(xprt
))
246 xprt_wake_pending_tasks(xprt
, 0);
248 if (xprt_test_and_clear_connected(xprt
))
249 xprt_wake_pending_tasks(xprt
, -ENOTCONN
);
251 spin_unlock_bh(&xprt
->transport_lock
);
255 xprt_rdma_connect_worker(struct work_struct
*work
)
257 struct rpcrdma_xprt
*r_xprt
= container_of(work
, struct rpcrdma_xprt
,
258 rx_connect_worker
.work
);
259 struct rpc_xprt
*xprt
= &r_xprt
->rx_xprt
;
262 xprt_clear_connected(xprt
);
264 rc
= rpcrdma_ep_connect(&r_xprt
->rx_ep
, &r_xprt
->rx_ia
);
266 xprt_wake_pending_tasks(xprt
, rc
);
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 trace_xprtrdma_inject_dsc(r_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 trace_xprtrdma_destroy(r_xprt
);
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
);
310 module_put(THIS_MODULE
);
313 static const struct rpc_timeout xprt_rdma_default_timeout
= {
314 .to_initval
= 60 * HZ
,
315 .to_maxval
= 60 * HZ
,
319 * xprt_setup_rdma - Set up transport to use RDMA
321 * @args: rpc transport arguments
323 static struct rpc_xprt
*
324 xprt_setup_rdma(struct xprt_create
*args
)
326 struct rpcrdma_create_data_internal cdata
;
327 struct rpc_xprt
*xprt
;
328 struct rpcrdma_xprt
*new_xprt
;
329 struct rpcrdma_ep
*new_ep
;
330 struct sockaddr
*sap
;
333 if (args
->addrlen
> sizeof(xprt
->addr
)) {
334 dprintk("RPC: %s: address too large\n", __func__
);
335 return ERR_PTR(-EBADF
);
338 xprt
= xprt_alloc(args
->net
, sizeof(struct rpcrdma_xprt
), 0, 0);
340 dprintk("RPC: %s: couldn't allocate rpcrdma_xprt\n",
342 return ERR_PTR(-ENOMEM
);
345 /* 60 second timeout, no retries */
346 xprt
->timeout
= &xprt_rdma_default_timeout
;
347 xprt
->bind_timeout
= RPCRDMA_BIND_TO
;
348 xprt
->reestablish_timeout
= RPCRDMA_INIT_REEST_TO
;
349 xprt
->idle_timeout
= RPCRDMA_IDLE_DISC_TO
;
351 xprt
->resvport
= 0; /* privileged port not needed */
352 xprt
->tsh_size
= 0; /* RPC-RDMA handles framing */
353 xprt
->ops
= &xprt_rdma_procs
;
356 * Set up RDMA-specific connect data.
360 /* Ensure xprt->addr holds valid server TCP (not RDMA)
361 * address, for any side protocols which peek at it */
362 xprt
->prot
= IPPROTO_TCP
;
363 xprt
->addrlen
= args
->addrlen
;
364 memcpy(&xprt
->addr
, sap
, xprt
->addrlen
);
366 if (rpc_get_port(sap
))
367 xprt_set_bound(xprt
);
368 xprt_rdma_format_addresses(xprt
, sap
);
370 cdata
.max_requests
= xprt_rdma_slot_table_entries
;
372 cdata
.rsize
= RPCRDMA_MAX_SEGS
* PAGE_SIZE
; /* RDMA write max */
373 cdata
.wsize
= RPCRDMA_MAX_SEGS
* PAGE_SIZE
; /* RDMA read max */
375 cdata
.inline_wsize
= xprt_rdma_max_inline_write
;
376 if (cdata
.inline_wsize
> cdata
.wsize
)
377 cdata
.inline_wsize
= cdata
.wsize
;
379 cdata
.inline_rsize
= xprt_rdma_max_inline_read
;
380 if (cdata
.inline_rsize
> cdata
.rsize
)
381 cdata
.inline_rsize
= cdata
.rsize
;
384 * Create new transport instance, which includes initialized
390 new_xprt
= rpcx_to_rdmax(xprt
);
392 rc
= rpcrdma_ia_open(new_xprt
);
397 * initialize and create ep
399 new_xprt
->rx_data
= cdata
;
400 new_ep
= &new_xprt
->rx_ep
;
402 rc
= rpcrdma_ep_create(&new_xprt
->rx_ep
,
403 &new_xprt
->rx_ia
, &new_xprt
->rx_data
);
407 rc
= rpcrdma_buffer_create(new_xprt
);
411 INIT_DELAYED_WORK(&new_xprt
->rx_connect_worker
,
412 xprt_rdma_connect_worker
);
414 xprt
->max_payload
= new_xprt
->rx_ia
.ri_ops
->ro_maxpages(new_xprt
);
415 if (xprt
->max_payload
== 0)
417 xprt
->max_payload
<<= PAGE_SHIFT
;
418 dprintk("RPC: %s: transport data payload maximum: %zu bytes\n",
419 __func__
, xprt
->max_payload
);
421 if (!try_module_get(THIS_MODULE
))
424 dprintk("RPC: %s: %s:%s\n", __func__
,
425 xprt
->address_strings
[RPC_DISPLAY_ADDR
],
426 xprt
->address_strings
[RPC_DISPLAY_PORT
]);
427 trace_xprtrdma_create(new_xprt
);
431 rpcrdma_buffer_destroy(&new_xprt
->rx_buf
);
434 rpcrdma_ep_destroy(new_ep
, &new_xprt
->rx_ia
);
436 rpcrdma_ia_close(&new_xprt
->rx_ia
);
438 trace_xprtrdma_destroy(new_xprt
);
439 xprt_rdma_free_addresses(xprt
);
445 * xprt_rdma_close - Close down RDMA connection
446 * @xprt: generic transport to be closed
448 * Called during transport shutdown reconnect, or device
449 * removal. Caller holds the transport's write lock.
452 xprt_rdma_close(struct rpc_xprt
*xprt
)
454 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
455 struct rpcrdma_ep
*ep
= &r_xprt
->rx_ep
;
456 struct rpcrdma_ia
*ia
= &r_xprt
->rx_ia
;
458 dprintk("RPC: %s: closing xprt %p\n", __func__
, xprt
);
460 if (test_and_clear_bit(RPCRDMA_IAF_REMOVING
, &ia
->ri_flags
)) {
461 xprt_clear_connected(xprt
);
462 rpcrdma_ia_remove(ia
);
465 if (ep
->rep_connected
== -ENODEV
)
467 if (ep
->rep_connected
> 0)
468 xprt
->reestablish_timeout
= 0;
469 xprt_disconnect_done(xprt
);
470 rpcrdma_ep_disconnect(ep
, ia
);
474 * xprt_rdma_set_port - update server port with rpcbind result
475 * @xprt: controlling RPC transport
476 * @port: new port value
478 * Transport connect status is unchanged.
481 xprt_rdma_set_port(struct rpc_xprt
*xprt
, u16 port
)
483 struct sockaddr
*sap
= (struct sockaddr
*)&xprt
->addr
;
486 dprintk("RPC: %s: setting port for xprt %p (%s:%s) to %u\n",
488 xprt
->address_strings
[RPC_DISPLAY_ADDR
],
489 xprt
->address_strings
[RPC_DISPLAY_PORT
],
492 rpc_set_port(sap
, port
);
494 kfree(xprt
->address_strings
[RPC_DISPLAY_PORT
]);
495 snprintf(buf
, sizeof(buf
), "%u", port
);
496 xprt
->address_strings
[RPC_DISPLAY_PORT
] = kstrdup(buf
, GFP_KERNEL
);
498 kfree(xprt
->address_strings
[RPC_DISPLAY_HEX_PORT
]);
499 snprintf(buf
, sizeof(buf
), "%4hx", port
);
500 xprt
->address_strings
[RPC_DISPLAY_HEX_PORT
] = kstrdup(buf
, GFP_KERNEL
);
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 xprt_force_disconnect(xprt
);
523 xprt_rdma_connect(struct rpc_xprt
*xprt
, struct rpc_task
*task
)
525 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
527 if (r_xprt
->rx_ep
.rep_connected
!= 0) {
529 schedule_delayed_work(&r_xprt
->rx_connect_worker
,
530 xprt
->reestablish_timeout
);
531 xprt
->reestablish_timeout
<<= 1;
532 if (xprt
->reestablish_timeout
> RPCRDMA_MAX_REEST_TO
)
533 xprt
->reestablish_timeout
= RPCRDMA_MAX_REEST_TO
;
534 else if (xprt
->reestablish_timeout
< RPCRDMA_INIT_REEST_TO
)
535 xprt
->reestablish_timeout
= RPCRDMA_INIT_REEST_TO
;
537 schedule_delayed_work(&r_xprt
->rx_connect_worker
, 0);
538 if (!RPC_IS_ASYNC(task
))
539 flush_delayed_work(&r_xprt
->rx_connect_worker
);
544 * xprt_rdma_alloc_slot - allocate an rpc_rqst
545 * @xprt: controlling RPC transport
546 * @task: RPC task requesting a fresh rpc_rqst
549 * %0 if task->tk_rqstp points to a fresh rpc_rqst
550 * %-EAGAIN if no rpc_rqst is available; queued on backlog
553 xprt_rdma_alloc_slot(struct rpc_xprt
*xprt
, struct rpc_task
*task
)
555 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
556 struct rpcrdma_req
*req
;
558 req
= rpcrdma_buffer_get(&r_xprt
->rx_buf
);
561 task
->tk_rqstp
= &req
->rl_slot
;
566 rpc_sleep_on(&xprt
->backlog
, task
, NULL
);
567 task
->tk_status
= -EAGAIN
;
571 * xprt_rdma_free_slot - release an rpc_rqst
572 * @xprt: controlling RPC transport
573 * @rqst: rpc_rqst to release
577 xprt_rdma_free_slot(struct rpc_xprt
*xprt
, struct rpc_rqst
*rqst
)
579 memset(rqst
, 0, sizeof(*rqst
));
580 rpcrdma_buffer_put(rpcr_to_rdmar(rqst
));
581 rpc_wake_up_next(&xprt
->backlog
);
585 rpcrdma_get_sendbuf(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
586 size_t size
, gfp_t flags
)
588 struct rpcrdma_regbuf
*rb
;
590 if (req
->rl_sendbuf
&& rdmab_length(req
->rl_sendbuf
) >= size
)
593 rb
= rpcrdma_alloc_regbuf(size
, DMA_TO_DEVICE
, flags
);
597 rpcrdma_free_regbuf(req
->rl_sendbuf
);
598 r_xprt
->rx_stats
.hardway_register_count
+= size
;
599 req
->rl_sendbuf
= rb
;
603 /* The rq_rcv_buf is used only if a Reply chunk is necessary.
604 * The decision to use a Reply chunk is made later in
605 * rpcrdma_marshal_req. This buffer is registered at that time.
607 * Otherwise, the associated RPC Reply arrives in a separate
608 * Receive buffer, arbitrarily chosen by the HCA. The buffer
609 * allocated here for the RPC Reply is not utilized in that
610 * case. See rpcrdma_inline_fixup.
612 * A regbuf is used here to remember the buffer size.
615 rpcrdma_get_recvbuf(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
616 size_t size
, gfp_t flags
)
618 struct rpcrdma_regbuf
*rb
;
620 if (req
->rl_recvbuf
&& rdmab_length(req
->rl_recvbuf
) >= size
)
623 rb
= rpcrdma_alloc_regbuf(size
, DMA_NONE
, flags
);
627 rpcrdma_free_regbuf(req
->rl_recvbuf
);
628 r_xprt
->rx_stats
.hardway_register_count
+= size
;
629 req
->rl_recvbuf
= rb
;
634 * xprt_rdma_allocate - allocate transport resources for an RPC
638 * 0: Success; rq_buffer points to RPC buffer to use
639 * ENOMEM: Out of memory, call again later
640 * EIO: A permanent error occurred, do not retry
642 * The RDMA allocate/free functions need the task structure as a place
643 * to hide the struct rpcrdma_req, which is necessary for the actual
644 * send/recv sequence.
646 * xprt_rdma_allocate provides buffers that are already mapped for
647 * DMA, and a local DMA lkey is provided for each.
650 xprt_rdma_allocate(struct rpc_task
*task
)
652 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
653 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(rqst
->rq_xprt
);
654 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
657 flags
= RPCRDMA_DEF_GFP
;
658 if (RPC_IS_SWAPPER(task
))
659 flags
= __GFP_MEMALLOC
| GFP_NOWAIT
| __GFP_NOWARN
;
661 if (!rpcrdma_get_sendbuf(r_xprt
, req
, rqst
->rq_callsize
, flags
))
663 if (!rpcrdma_get_recvbuf(r_xprt
, req
, rqst
->rq_rcvsize
, flags
))
666 rqst
->rq_buffer
= req
->rl_sendbuf
->rg_base
;
667 rqst
->rq_rbuffer
= req
->rl_recvbuf
->rg_base
;
668 trace_xprtrdma_allocate(task
, req
);
672 trace_xprtrdma_allocate(task
, NULL
);
677 * xprt_rdma_free - release resources allocated by xprt_rdma_allocate
680 * Caller guarantees rqst->rq_buffer is non-NULL.
683 xprt_rdma_free(struct rpc_task
*task
)
685 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
686 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(rqst
->rq_xprt
);
687 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
689 if (test_bit(RPCRDMA_REQ_F_PENDING
, &req
->rl_flags
))
690 rpcrdma_release_rqst(r_xprt
, req
);
691 trace_xprtrdma_rpc_done(task
, req
);
695 * xprt_rdma_send_request - marshal and send an RPC request
696 * @task: RPC task with an RPC message in rq_snd_buf
698 * Caller holds the transport's write lock.
701 * %0 if the RPC message has been sent
702 * %-ENOTCONN if the caller should reconnect and call again
703 * %-EAGAIN if the caller should call again
704 * %-ENOBUFS if the caller should call again after a delay
705 * %-EIO if a permanent error occurred and the request was not
706 * sent. Do not try to send this message again.
709 xprt_rdma_send_request(struct rpc_task
*task
)
711 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
712 struct rpc_xprt
*xprt
= rqst
->rq_xprt
;
713 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
714 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
717 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
718 if (unlikely(!rqst
->rq_buffer
))
719 return xprt_rdma_bc_send_reply(rqst
);
720 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
722 if (!xprt_connected(xprt
))
723 goto drop_connection
;
725 rc
= rpcrdma_marshal_req(r_xprt
, rqst
);
729 /* Must suppress retransmit to maintain credits */
730 if (rqst
->rq_connect_cookie
== xprt
->connect_cookie
)
731 goto drop_connection
;
732 rqst
->rq_xtime
= ktime_get();
734 __set_bit(RPCRDMA_REQ_F_PENDING
, &req
->rl_flags
);
735 if (rpcrdma_ep_post(&r_xprt
->rx_ia
, &r_xprt
->rx_ep
, req
))
736 goto drop_connection
;
738 rqst
->rq_xmit_bytes_sent
+= rqst
->rq_snd_buf
.len
;
739 rqst
->rq_bytes_sent
= 0;
741 /* An RPC with no reply will throw off credit accounting,
742 * so drop the connection to reset the credit grant.
744 if (!rpc_reply_expected(task
))
745 goto drop_connection
;
752 xprt_disconnect_done(xprt
);
753 return -ENOTCONN
; /* implies disconnect */
756 void xprt_rdma_print_stats(struct rpc_xprt
*xprt
, struct seq_file
*seq
)
758 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
761 if (xprt_connected(xprt
))
762 idle_time
= (long)(jiffies
- xprt
->last_used
) / HZ
;
764 seq_puts(seq
, "\txprt:\trdma ");
765 seq_printf(seq
, "%u %lu %lu %lu %ld %lu %lu %lu %llu %llu ",
766 0, /* need a local port? */
767 xprt
->stat
.bind_count
,
768 xprt
->stat
.connect_count
,
769 xprt
->stat
.connect_time
,
776 seq_printf(seq
, "%lu %lu %lu %llu %llu %llu %llu %lu %lu %lu %lu ",
777 r_xprt
->rx_stats
.read_chunk_count
,
778 r_xprt
->rx_stats
.write_chunk_count
,
779 r_xprt
->rx_stats
.reply_chunk_count
,
780 r_xprt
->rx_stats
.total_rdma_request
,
781 r_xprt
->rx_stats
.total_rdma_reply
,
782 r_xprt
->rx_stats
.pullup_copy_count
,
783 r_xprt
->rx_stats
.fixup_copy_count
,
784 r_xprt
->rx_stats
.hardway_register_count
,
785 r_xprt
->rx_stats
.failed_marshal_count
,
786 r_xprt
->rx_stats
.bad_reply_count
,
787 r_xprt
->rx_stats
.nomsg_call_count
);
788 seq_printf(seq
, "%lu %lu %lu %lu %lu %lu\n",
789 r_xprt
->rx_stats
.mrs_recovered
,
790 r_xprt
->rx_stats
.mrs_orphaned
,
791 r_xprt
->rx_stats
.mrs_allocated
,
792 r_xprt
->rx_stats
.local_inv_needed
,
793 r_xprt
->rx_stats
.empty_sendctx_q
,
794 r_xprt
->rx_stats
.reply_waits_for_send
);
798 xprt_rdma_enable_swap(struct rpc_xprt
*xprt
)
804 xprt_rdma_disable_swap(struct rpc_xprt
*xprt
)
809 * Plumbing for rpc transport switch and kernel module
812 static const struct rpc_xprt_ops xprt_rdma_procs
= {
813 .reserve_xprt
= xprt_reserve_xprt_cong
,
814 .release_xprt
= xprt_release_xprt_cong
, /* sunrpc/xprt.c */
815 .alloc_slot
= xprt_rdma_alloc_slot
,
816 .free_slot
= xprt_rdma_free_slot
,
817 .release_request
= xprt_release_rqst_cong
, /* ditto */
818 .set_retrans_timeout
= xprt_set_retrans_timeout_def
, /* ditto */
819 .timer
= xprt_rdma_timer
,
820 .rpcbind
= rpcb_getport_async
, /* sunrpc/rpcb_clnt.c */
821 .set_port
= xprt_rdma_set_port
,
822 .connect
= xprt_rdma_connect
,
823 .buf_alloc
= xprt_rdma_allocate
,
824 .buf_free
= xprt_rdma_free
,
825 .send_request
= xprt_rdma_send_request
,
826 .close
= xprt_rdma_close
,
827 .destroy
= xprt_rdma_destroy
,
828 .print_stats
= xprt_rdma_print_stats
,
829 .enable_swap
= xprt_rdma_enable_swap
,
830 .disable_swap
= xprt_rdma_disable_swap
,
831 .inject_disconnect
= xprt_rdma_inject_disconnect
,
832 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
833 .bc_setup
= xprt_rdma_bc_setup
,
834 .bc_up
= xprt_rdma_bc_up
,
835 .bc_maxpayload
= xprt_rdma_bc_maxpayload
,
836 .bc_free_rqst
= xprt_rdma_bc_free_rqst
,
837 .bc_destroy
= xprt_rdma_bc_destroy
,
841 static struct xprt_class xprt_rdma
= {
842 .list
= LIST_HEAD_INIT(xprt_rdma
.list
),
844 .owner
= THIS_MODULE
,
845 .ident
= XPRT_TRANSPORT_RDMA
,
846 .setup
= xprt_setup_rdma
,
849 void xprt_rdma_cleanup(void)
853 dprintk("RPCRDMA Module Removed, deregister RPC RDMA transport\n");
854 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
855 if (sunrpc_table_header
) {
856 unregister_sysctl_table(sunrpc_table_header
);
857 sunrpc_table_header
= NULL
;
860 rc
= xprt_unregister_transport(&xprt_rdma
);
862 dprintk("RPC: %s: xprt_unregister returned %i\n",
865 rpcrdma_destroy_wq();
867 rc
= xprt_unregister_transport(&xprt_rdma_bc
);
869 dprintk("RPC: %s: xprt_unregister(bc) returned %i\n",
873 int xprt_rdma_init(void)
877 rc
= rpcrdma_alloc_wq();
881 rc
= xprt_register_transport(&xprt_rdma
);
883 rpcrdma_destroy_wq();
887 rc
= xprt_register_transport(&xprt_rdma_bc
);
889 xprt_unregister_transport(&xprt_rdma
);
890 rpcrdma_destroy_wq();
894 dprintk("RPCRDMA Module Init, register RPC RDMA transport\n");
896 dprintk("Defaults:\n");
897 dprintk("\tSlots %d\n"
898 "\tMaxInlineRead %d\n\tMaxInlineWrite %d\n",
899 xprt_rdma_slot_table_entries
,
900 xprt_rdma_max_inline_read
, xprt_rdma_max_inline_write
);
901 dprintk("\tPadding 0\n\tMemreg %d\n", xprt_rdma_memreg_strategy
);
903 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
904 if (!sunrpc_table_header
)
905 sunrpc_table_header
= register_sysctl_table(sunrpc_table
);