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 * xprt_rdma_connect_worker - establish connection in the background
230 * @work: worker thread context
232 * Requester holds the xprt's send lock to prevent activity on this
233 * transport while a fresh connection is being established. RPC tasks
234 * sleep on the xprt's pending queue waiting for connect to complete.
237 xprt_rdma_connect_worker(struct work_struct
*work
)
239 struct rpcrdma_xprt
*r_xprt
= container_of(work
, struct rpcrdma_xprt
,
240 rx_connect_worker
.work
);
241 struct rpc_xprt
*xprt
= &r_xprt
->rx_xprt
;
244 rc
= rpcrdma_ep_connect(&r_xprt
->rx_ep
, &r_xprt
->rx_ia
);
245 xprt_clear_connecting(xprt
);
246 if (r_xprt
->rx_ep
.rep_connected
> 0) {
247 if (!xprt_test_and_set_connected(xprt
)) {
248 xprt
->stat
.connect_count
++;
249 xprt
->stat
.connect_time
+= (long)jiffies
-
250 xprt
->stat
.connect_start
;
251 xprt_wake_pending_tasks(xprt
, -EAGAIN
);
254 if (xprt_test_and_clear_connected(xprt
))
255 xprt_wake_pending_tasks(xprt
, rc
);
260 * xprt_rdma_inject_disconnect - inject a connection fault
261 * @xprt: transport context
263 * If @xprt is connected, disconnect it to simulate spurious connection
267 xprt_rdma_inject_disconnect(struct rpc_xprt
*xprt
)
269 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
271 trace_xprtrdma_op_inject_dsc(r_xprt
);
272 rdma_disconnect(r_xprt
->rx_ia
.ri_id
);
276 * xprt_rdma_destroy - Full tear down of transport
277 * @xprt: doomed transport context
279 * Caller guarantees there will be no more calls to us with
283 xprt_rdma_destroy(struct rpc_xprt
*xprt
)
285 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
287 trace_xprtrdma_op_destroy(r_xprt
);
289 cancel_delayed_work_sync(&r_xprt
->rx_connect_worker
);
291 rpcrdma_ep_destroy(&r_xprt
->rx_ep
, &r_xprt
->rx_ia
);
292 rpcrdma_buffer_destroy(&r_xprt
->rx_buf
);
293 rpcrdma_ia_close(&r_xprt
->rx_ia
);
295 xprt_rdma_free_addresses(xprt
);
298 module_put(THIS_MODULE
);
301 static const struct rpc_timeout xprt_rdma_default_timeout
= {
302 .to_initval
= 60 * HZ
,
303 .to_maxval
= 60 * HZ
,
307 * xprt_setup_rdma - Set up transport to use RDMA
309 * @args: rpc transport arguments
311 static struct rpc_xprt
*
312 xprt_setup_rdma(struct xprt_create
*args
)
314 struct rpcrdma_create_data_internal cdata
;
315 struct rpc_xprt
*xprt
;
316 struct rpcrdma_xprt
*new_xprt
;
317 struct rpcrdma_ep
*new_ep
;
318 struct sockaddr
*sap
;
321 if (args
->addrlen
> sizeof(xprt
->addr
))
322 return ERR_PTR(-EBADF
);
324 xprt
= xprt_alloc(args
->net
, sizeof(struct rpcrdma_xprt
), 0, 0);
326 return ERR_PTR(-ENOMEM
);
328 /* 60 second timeout, no retries */
329 xprt
->timeout
= &xprt_rdma_default_timeout
;
330 xprt
->bind_timeout
= RPCRDMA_BIND_TO
;
331 xprt
->reestablish_timeout
= RPCRDMA_INIT_REEST_TO
;
332 xprt
->idle_timeout
= RPCRDMA_IDLE_DISC_TO
;
334 xprt
->resvport
= 0; /* privileged port not needed */
335 xprt
->ops
= &xprt_rdma_procs
;
338 * Set up RDMA-specific connect data.
342 /* Ensure xprt->addr holds valid server TCP (not RDMA)
343 * address, for any side protocols which peek at it */
344 xprt
->prot
= IPPROTO_TCP
;
345 xprt
->addrlen
= args
->addrlen
;
346 memcpy(&xprt
->addr
, sap
, xprt
->addrlen
);
348 if (rpc_get_port(sap
))
349 xprt_set_bound(xprt
);
350 xprt_rdma_format_addresses(xprt
, sap
);
352 cdata
.max_requests
= xprt_rdma_slot_table_entries
;
354 cdata
.rsize
= RPCRDMA_MAX_SEGS
* PAGE_SIZE
; /* RDMA write max */
355 cdata
.wsize
= RPCRDMA_MAX_SEGS
* PAGE_SIZE
; /* RDMA read max */
357 cdata
.inline_wsize
= xprt_rdma_max_inline_write
;
358 if (cdata
.inline_wsize
> cdata
.wsize
)
359 cdata
.inline_wsize
= cdata
.wsize
;
361 cdata
.inline_rsize
= xprt_rdma_max_inline_read
;
362 if (cdata
.inline_rsize
> cdata
.rsize
)
363 cdata
.inline_rsize
= cdata
.rsize
;
366 * Create new transport instance, which includes initialized
372 new_xprt
= rpcx_to_rdmax(xprt
);
374 rc
= rpcrdma_ia_open(new_xprt
);
379 * initialize and create ep
381 new_xprt
->rx_data
= cdata
;
382 new_ep
= &new_xprt
->rx_ep
;
384 rc
= rpcrdma_ep_create(&new_xprt
->rx_ep
,
385 &new_xprt
->rx_ia
, &new_xprt
->rx_data
);
389 rc
= rpcrdma_buffer_create(new_xprt
);
393 INIT_DELAYED_WORK(&new_xprt
->rx_connect_worker
,
394 xprt_rdma_connect_worker
);
396 xprt
->max_payload
= frwr_maxpages(new_xprt
);
397 if (xprt
->max_payload
== 0)
399 xprt
->max_payload
<<= PAGE_SHIFT
;
400 dprintk("RPC: %s: transport data payload maximum: %zu bytes\n",
401 __func__
, xprt
->max_payload
);
403 if (!try_module_get(THIS_MODULE
))
406 dprintk("RPC: %s: %s:%s\n", __func__
,
407 xprt
->address_strings
[RPC_DISPLAY_ADDR
],
408 xprt
->address_strings
[RPC_DISPLAY_PORT
]);
409 trace_xprtrdma_create(new_xprt
);
413 rpcrdma_buffer_destroy(&new_xprt
->rx_buf
);
416 rpcrdma_ep_destroy(new_ep
, &new_xprt
->rx_ia
);
418 rpcrdma_ia_close(&new_xprt
->rx_ia
);
420 trace_xprtrdma_op_destroy(new_xprt
);
421 xprt_rdma_free_addresses(xprt
);
427 * xprt_rdma_close - close a transport connection
428 * @xprt: transport context
430 * Called during autoclose or device removal.
432 * Caller holds @xprt's send lock to prevent activity on this
433 * transport while the connection is torn down.
435 void xprt_rdma_close(struct rpc_xprt
*xprt
)
437 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
438 struct rpcrdma_ep
*ep
= &r_xprt
->rx_ep
;
439 struct rpcrdma_ia
*ia
= &r_xprt
->rx_ia
;
443 trace_xprtrdma_op_close(r_xprt
);
445 /* Prevent marshaling and sending of new requests */
446 xprt_clear_connected(xprt
);
448 if (test_and_clear_bit(RPCRDMA_IAF_REMOVING
, &ia
->ri_flags
)) {
449 rpcrdma_ia_remove(ia
);
453 if (ep
->rep_connected
== -ENODEV
)
455 if (ep
->rep_connected
> 0)
456 xprt
->reestablish_timeout
= 0;
457 rpcrdma_ep_disconnect(ep
, ia
);
459 /* Prepare @xprt for the next connection by reinitializing
460 * its credit grant to one (see RFC 8166, Section 3.3.3).
462 r_xprt
->rx_buf
.rb_credits
= 1;
463 xprt
->cwnd
= RPC_CWNDSHIFT
;
466 ++xprt
->connect_cookie
;
467 xprt_disconnect_done(xprt
);
471 * xprt_rdma_set_port - update server port with rpcbind result
472 * @xprt: controlling RPC transport
473 * @port: new port value
475 * Transport connect status is unchanged.
478 xprt_rdma_set_port(struct rpc_xprt
*xprt
, u16 port
)
480 struct sockaddr
*sap
= (struct sockaddr
*)&xprt
->addr
;
483 dprintk("RPC: %s: setting port for xprt %p (%s:%s) to %u\n",
485 xprt
->address_strings
[RPC_DISPLAY_ADDR
],
486 xprt
->address_strings
[RPC_DISPLAY_PORT
],
489 rpc_set_port(sap
, port
);
491 kfree(xprt
->address_strings
[RPC_DISPLAY_PORT
]);
492 snprintf(buf
, sizeof(buf
), "%u", port
);
493 xprt
->address_strings
[RPC_DISPLAY_PORT
] = kstrdup(buf
, GFP_KERNEL
);
495 kfree(xprt
->address_strings
[RPC_DISPLAY_HEX_PORT
]);
496 snprintf(buf
, sizeof(buf
), "%4hx", port
);
497 xprt
->address_strings
[RPC_DISPLAY_HEX_PORT
] = kstrdup(buf
, GFP_KERNEL
);
501 * xprt_rdma_timer - invoked when an RPC times out
502 * @xprt: controlling RPC transport
503 * @task: RPC task that timed out
505 * Invoked when the transport is still connected, but an RPC
506 * retransmit timeout occurs.
508 * Since RDMA connections don't have a keep-alive, forcibly
509 * disconnect and retry to connect. This drives full
510 * detection of the network path, and retransmissions of
514 xprt_rdma_timer(struct rpc_xprt
*xprt
, struct rpc_task
*task
)
516 xprt_force_disconnect(xprt
);
520 * xprt_rdma_connect - try to establish a transport connection
521 * @xprt: transport state
522 * @task: RPC scheduler context
526 xprt_rdma_connect(struct rpc_xprt
*xprt
, struct rpc_task
*task
)
528 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
530 trace_xprtrdma_op_connect(r_xprt
);
531 if (r_xprt
->rx_ep
.rep_connected
!= 0) {
533 schedule_delayed_work(&r_xprt
->rx_connect_worker
,
534 xprt
->reestablish_timeout
);
535 xprt
->reestablish_timeout
<<= 1;
536 if (xprt
->reestablish_timeout
> RPCRDMA_MAX_REEST_TO
)
537 xprt
->reestablish_timeout
= RPCRDMA_MAX_REEST_TO
;
538 else if (xprt
->reestablish_timeout
< RPCRDMA_INIT_REEST_TO
)
539 xprt
->reestablish_timeout
= RPCRDMA_INIT_REEST_TO
;
541 schedule_delayed_work(&r_xprt
->rx_connect_worker
, 0);
542 if (!RPC_IS_ASYNC(task
))
543 flush_delayed_work(&r_xprt
->rx_connect_worker
);
548 * xprt_rdma_alloc_slot - allocate an rpc_rqst
549 * @xprt: controlling RPC transport
550 * @task: RPC task requesting a fresh rpc_rqst
553 * %0 if task->tk_rqstp points to a fresh rpc_rqst
554 * %-EAGAIN if no rpc_rqst is available; queued on backlog
557 xprt_rdma_alloc_slot(struct rpc_xprt
*xprt
, struct rpc_task
*task
)
559 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
560 struct rpcrdma_req
*req
;
562 req
= rpcrdma_buffer_get(&r_xprt
->rx_buf
);
565 task
->tk_rqstp
= &req
->rl_slot
;
570 rpc_sleep_on(&xprt
->backlog
, task
, NULL
);
571 task
->tk_status
= -EAGAIN
;
575 * xprt_rdma_free_slot - release an rpc_rqst
576 * @xprt: controlling RPC transport
577 * @rqst: rpc_rqst to release
581 xprt_rdma_free_slot(struct rpc_xprt
*xprt
, struct rpc_rqst
*rqst
)
583 memset(rqst
, 0, sizeof(*rqst
));
584 rpcrdma_buffer_put(rpcr_to_rdmar(rqst
));
585 rpc_wake_up_next(&xprt
->backlog
);
589 rpcrdma_get_sendbuf(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
590 size_t size
, gfp_t flags
)
592 struct rpcrdma_regbuf
*rb
;
594 if (req
->rl_sendbuf
&& rdmab_length(req
->rl_sendbuf
) >= size
)
597 rb
= rpcrdma_alloc_regbuf(size
, DMA_TO_DEVICE
, flags
);
601 rpcrdma_free_regbuf(req
->rl_sendbuf
);
602 r_xprt
->rx_stats
.hardway_register_count
+= size
;
603 req
->rl_sendbuf
= rb
;
607 /* The rq_rcv_buf is used only if a Reply chunk is necessary.
608 * The decision to use a Reply chunk is made later in
609 * rpcrdma_marshal_req. This buffer is registered at that time.
611 * Otherwise, the associated RPC Reply arrives in a separate
612 * Receive buffer, arbitrarily chosen by the HCA. The buffer
613 * allocated here for the RPC Reply is not utilized in that
614 * case. See rpcrdma_inline_fixup.
616 * A regbuf is used here to remember the buffer size.
619 rpcrdma_get_recvbuf(struct rpcrdma_xprt
*r_xprt
, struct rpcrdma_req
*req
,
620 size_t size
, gfp_t flags
)
622 struct rpcrdma_regbuf
*rb
;
624 if (req
->rl_recvbuf
&& rdmab_length(req
->rl_recvbuf
) >= size
)
627 rb
= rpcrdma_alloc_regbuf(size
, DMA_NONE
, flags
);
631 rpcrdma_free_regbuf(req
->rl_recvbuf
);
632 r_xprt
->rx_stats
.hardway_register_count
+= size
;
633 req
->rl_recvbuf
= rb
;
638 * xprt_rdma_allocate - allocate transport resources for an RPC
642 * 0: Success; rq_buffer points to RPC buffer to use
643 * ENOMEM: Out of memory, call again later
644 * EIO: A permanent error occurred, do not retry
647 xprt_rdma_allocate(struct rpc_task
*task
)
649 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
650 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(rqst
->rq_xprt
);
651 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
654 flags
= RPCRDMA_DEF_GFP
;
655 if (RPC_IS_SWAPPER(task
))
656 flags
= __GFP_MEMALLOC
| GFP_NOWAIT
| __GFP_NOWARN
;
658 if (!rpcrdma_get_sendbuf(r_xprt
, req
, rqst
->rq_callsize
, flags
))
660 if (!rpcrdma_get_recvbuf(r_xprt
, req
, rqst
->rq_rcvsize
, flags
))
663 rqst
->rq_buffer
= req
->rl_sendbuf
->rg_base
;
664 rqst
->rq_rbuffer
= req
->rl_recvbuf
->rg_base
;
665 trace_xprtrdma_op_allocate(task
, req
);
669 trace_xprtrdma_op_allocate(task
, NULL
);
674 * xprt_rdma_free - release resources allocated by xprt_rdma_allocate
677 * Caller guarantees rqst->rq_buffer is non-NULL.
680 xprt_rdma_free(struct rpc_task
*task
)
682 struct rpc_rqst
*rqst
= task
->tk_rqstp
;
683 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(rqst
->rq_xprt
);
684 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
686 if (test_bit(RPCRDMA_REQ_F_PENDING
, &req
->rl_flags
))
687 rpcrdma_release_rqst(r_xprt
, req
);
688 trace_xprtrdma_op_free(task
, req
);
692 * xprt_rdma_send_request - marshal and send an RPC request
693 * @rqst: RPC message in rq_snd_buf
695 * Caller holds the transport's write lock.
698 * %0 if the RPC message has been sent
699 * %-ENOTCONN if the caller should reconnect and call again
700 * %-EAGAIN if the caller should call again
701 * %-ENOBUFS if the caller should call again after a delay
702 * %-EMSGSIZE if encoding ran out of buffer space. The request
703 * was not sent. Do not try to send this message again.
704 * %-EIO if an I/O error occurred. The request was not sent.
705 * Do not try to send this message again.
708 xprt_rdma_send_request(struct rpc_rqst
*rqst
)
710 struct rpc_xprt
*xprt
= rqst
->rq_xprt
;
711 struct rpcrdma_req
*req
= rpcr_to_rdmar(rqst
);
712 struct rpcrdma_xprt
*r_xprt
= rpcx_to_rdmax(xprt
);
715 #if defined(CONFIG_SUNRPC_BACKCHANNEL)
716 if (unlikely(!rqst
->rq_buffer
))
717 return xprt_rdma_bc_send_reply(rqst
);
718 #endif /* CONFIG_SUNRPC_BACKCHANNEL */
720 if (!xprt_connected(xprt
))
723 if (!xprt_request_get_cong(xprt
, rqst
))
726 rc
= rpcrdma_marshal_req(r_xprt
, rqst
);
730 /* Must suppress retransmit to maintain credits */
731 if (rqst
->rq_connect_cookie
== xprt
->connect_cookie
)
732 goto drop_connection
;
733 rqst
->rq_xtime
= ktime_get();
735 __set_bit(RPCRDMA_REQ_F_PENDING
, &req
->rl_flags
);
736 if (rpcrdma_ep_post(&r_xprt
->rx_ia
, &r_xprt
->rx_ep
, req
))
737 goto drop_connection
;
739 rqst
->rq_xmit_bytes_sent
+= rqst
->rq_snd_buf
.len
;
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(rqst
->rq_task
))
745 goto drop_connection
;
752 xprt_rdma_close(xprt
);
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
/ HZ
,
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_recycled
,
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_maxpayload
= xprt_rdma_bc_maxpayload
,
835 .bc_free_rqst
= xprt_rdma_bc_free_rqst
,
836 .bc_destroy
= xprt_rdma_bc_destroy
,
840 static struct xprt_class xprt_rdma
= {
841 .list
= LIST_HEAD_INIT(xprt_rdma
.list
),
843 .owner
= THIS_MODULE
,
844 .ident
= XPRT_TRANSPORT_RDMA
,
845 .setup
= xprt_setup_rdma
,
848 void xprt_rdma_cleanup(void)
850 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
851 if (sunrpc_table_header
) {
852 unregister_sysctl_table(sunrpc_table_header
);
853 sunrpc_table_header
= NULL
;
857 xprt_unregister_transport(&xprt_rdma
);
858 xprt_unregister_transport(&xprt_rdma_bc
);
861 int xprt_rdma_init(void)
865 rc
= xprt_register_transport(&xprt_rdma
);
869 rc
= xprt_register_transport(&xprt_rdma_bc
);
871 xprt_unregister_transport(&xprt_rdma
);
875 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
876 if (!sunrpc_table_header
)
877 sunrpc_table_header
= register_sysctl_table(sunrpc_table
);