1 // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause 2 /* 3 * Copyright (c) 2014-2017 Oracle. All rights reserved. 4 * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved. 5 * 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 10 * license below: 11 * 12 * Redistribution and use in source and binary forms, with or without 13 * modification, are permitted provided that the following conditions 14 * are met: 15 * 16 * Redistributions of source code must retain the above copyright 17 * notice, this list of conditions and the following disclaimer. 18 * 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. 23 * 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 27 * permission. 28 * 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. 40 */ 41 42 /* 43 * transport.c 44 * 45 * This file contains the top-level implementation of an RPC RDMA 46 * transport. 47 * 48 * Naming convention: functions beginning with xprt_ are part of the 49 * transport switch. All others are RPC RDMA internal. 50 */ 51 52 #include <linux/module.h> 53 #include <linux/slab.h> 54 #include <linux/seq_file.h> 55 #include <linux/smp.h> 56 57 #include <linux/sunrpc/addr.h> 58 #include <linux/sunrpc/svc_rdma.h> 59 60 #include "xprt_rdma.h" 61 #include <trace/events/rpcrdma.h> 62 63 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 64 # define RPCDBG_FACILITY RPCDBG_TRANS 65 #endif 66 67 /* 68 * tunables 69 */ 70 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 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; 76 static struct xprt_class xprt_rdma; 77 78 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 79 80 static unsigned int min_slot_table_size = RPCRDMA_MIN_SLOT_TABLE; 81 static unsigned int max_slot_table_size = RPCRDMA_MAX_SLOT_TABLE; 82 static unsigned int min_inline_size = RPCRDMA_MIN_INLINE; 83 static unsigned int max_inline_size = RPCRDMA_MAX_INLINE; 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; 88 89 static struct ctl_table_header *sunrpc_table_header; 90 91 static struct ctl_table xr_tunables_table[] = { 92 { 93 .procname = "rdma_slot_table_entries", 94 .data = &xprt_rdma_slot_table_entries, 95 .maxlen = sizeof(unsigned int), 96 .mode = 0644, 97 .proc_handler = proc_dointvec_minmax, 98 .extra1 = &min_slot_table_size, 99 .extra2 = &max_slot_table_size 100 }, 101 { 102 .procname = "rdma_max_inline_read", 103 .data = &xprt_rdma_max_inline_read, 104 .maxlen = sizeof(unsigned int), 105 .mode = 0644, 106 .proc_handler = proc_dointvec_minmax, 107 .extra1 = &min_inline_size, 108 .extra2 = &max_inline_size, 109 }, 110 { 111 .procname = "rdma_max_inline_write", 112 .data = &xprt_rdma_max_inline_write, 113 .maxlen = sizeof(unsigned int), 114 .mode = 0644, 115 .proc_handler = proc_dointvec_minmax, 116 .extra1 = &min_inline_size, 117 .extra2 = &max_inline_size, 118 }, 119 { 120 .procname = "rdma_inline_write_padding", 121 .data = &dummy, 122 .maxlen = sizeof(unsigned int), 123 .mode = 0644, 124 .proc_handler = proc_dointvec_minmax, 125 .extra1 = SYSCTL_ZERO, 126 .extra2 = &max_padding, 127 }, 128 { 129 .procname = "rdma_memreg_strategy", 130 .data = &xprt_rdma_memreg_strategy, 131 .maxlen = sizeof(unsigned int), 132 .mode = 0644, 133 .proc_handler = proc_dointvec_minmax, 134 .extra1 = &min_memreg, 135 .extra2 = &max_memreg, 136 }, 137 { 138 .procname = "rdma_pad_optimize", 139 .data = &xprt_rdma_pad_optimize, 140 .maxlen = sizeof(unsigned int), 141 .mode = 0644, 142 .proc_handler = proc_dointvec, 143 }, 144 { }, 145 }; 146 147 static struct ctl_table sunrpc_table[] = { 148 { 149 .procname = "sunrpc", 150 .mode = 0555, 151 .child = xr_tunables_table 152 }, 153 { }, 154 }; 155 156 #endif 157 158 static const struct rpc_xprt_ops xprt_rdma_procs; 159 160 static void 161 xprt_rdma_format_addresses4(struct rpc_xprt *xprt, struct sockaddr *sap) 162 { 163 struct sockaddr_in *sin = (struct sockaddr_in *)sap; 164 char buf[20]; 165 166 snprintf(buf, sizeof(buf), "%08x", ntohl(sin->sin_addr.s_addr)); 167 xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL); 168 169 xprt->address_strings[RPC_DISPLAY_NETID] = RPCBIND_NETID_RDMA; 170 } 171 172 static void 173 xprt_rdma_format_addresses6(struct rpc_xprt *xprt, struct sockaddr *sap) 174 { 175 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)sap; 176 char buf[40]; 177 178 snprintf(buf, sizeof(buf), "%pi6", &sin6->sin6_addr); 179 xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL); 180 181 xprt->address_strings[RPC_DISPLAY_NETID] = RPCBIND_NETID_RDMA6; 182 } 183 184 void 185 xprt_rdma_format_addresses(struct rpc_xprt *xprt, struct sockaddr *sap) 186 { 187 char buf[128]; 188 189 switch (sap->sa_family) { 190 case AF_INET: 191 xprt_rdma_format_addresses4(xprt, sap); 192 break; 193 case AF_INET6: 194 xprt_rdma_format_addresses6(xprt, sap); 195 break; 196 default: 197 pr_err("rpcrdma: Unrecognized address family\n"); 198 return; 199 } 200 201 (void)rpc_ntop(sap, buf, sizeof(buf)); 202 xprt->address_strings[RPC_DISPLAY_ADDR] = kstrdup(buf, GFP_KERNEL); 203 204 snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap)); 205 xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL); 206 207 snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap)); 208 xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL); 209 210 xprt->address_strings[RPC_DISPLAY_PROTO] = "rdma"; 211 } 212 213 void 214 xprt_rdma_free_addresses(struct rpc_xprt *xprt) 215 { 216 unsigned int i; 217 218 for (i = 0; i < RPC_DISPLAY_MAX; i++) 219 switch (i) { 220 case RPC_DISPLAY_PROTO: 221 case RPC_DISPLAY_NETID: 222 continue; 223 default: 224 kfree(xprt->address_strings[i]); 225 } 226 } 227 228 /** 229 * xprt_rdma_connect_worker - establish connection in the background 230 * @work: worker thread context 231 * 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. 235 */ 236 static void 237 xprt_rdma_connect_worker(struct work_struct *work) 238 { 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; 242 int rc; 243 244 rc = rpcrdma_xprt_connect(r_xprt); 245 xprt_clear_connecting(xprt); 246 if (!rc) { 247 xprt->connect_cookie++; 248 xprt->stat.connect_count++; 249 xprt->stat.connect_time += (long)jiffies - 250 xprt->stat.connect_start; 251 xprt_set_connected(xprt); 252 rc = -EAGAIN; 253 } else { 254 /* Force a call to xprt_rdma_close to clean up */ 255 spin_lock(&xprt->transport_lock); 256 set_bit(XPRT_CLOSE_WAIT, &xprt->state); 257 spin_unlock(&xprt->transport_lock); 258 } 259 xprt_wake_pending_tasks(xprt, rc); 260 } 261 262 /** 263 * xprt_rdma_inject_disconnect - inject a connection fault 264 * @xprt: transport context 265 * 266 * If @xprt is connected, disconnect it to simulate spurious 267 * connection loss. Caller must hold @xprt's send lock to 268 * ensure that data structures and hardware resources are 269 * stable during the rdma_disconnect() call. 270 */ 271 static void 272 xprt_rdma_inject_disconnect(struct rpc_xprt *xprt) 273 { 274 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 275 276 trace_xprtrdma_op_inject_dsc(r_xprt); 277 rdma_disconnect(r_xprt->rx_ep->re_id); 278 } 279 280 /** 281 * xprt_rdma_destroy - Full tear down of transport 282 * @xprt: doomed transport context 283 * 284 * Caller guarantees there will be no more calls to us with 285 * this @xprt. 286 */ 287 static void 288 xprt_rdma_destroy(struct rpc_xprt *xprt) 289 { 290 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 291 292 cancel_delayed_work_sync(&r_xprt->rx_connect_worker); 293 294 rpcrdma_xprt_disconnect(r_xprt); 295 rpcrdma_buffer_destroy(&r_xprt->rx_buf); 296 297 xprt_rdma_free_addresses(xprt); 298 xprt_free(xprt); 299 300 module_put(THIS_MODULE); 301 } 302 303 /* 60 second timeout, no retries */ 304 static const struct rpc_timeout xprt_rdma_default_timeout = { 305 .to_initval = 60 * HZ, 306 .to_maxval = 60 * HZ, 307 }; 308 309 /** 310 * xprt_setup_rdma - Set up transport to use RDMA 311 * 312 * @args: rpc transport arguments 313 */ 314 static struct rpc_xprt * 315 xprt_setup_rdma(struct xprt_create *args) 316 { 317 struct rpc_xprt *xprt; 318 struct rpcrdma_xprt *new_xprt; 319 struct sockaddr *sap; 320 int rc; 321 322 if (args->addrlen > sizeof(xprt->addr)) 323 return ERR_PTR(-EBADF); 324 325 if (!try_module_get(THIS_MODULE)) 326 return ERR_PTR(-EIO); 327 328 xprt = xprt_alloc(args->net, sizeof(struct rpcrdma_xprt), 0, 329 xprt_rdma_slot_table_entries); 330 if (!xprt) { 331 module_put(THIS_MODULE); 332 return ERR_PTR(-ENOMEM); 333 } 334 335 xprt->timeout = &xprt_rdma_default_timeout; 336 xprt->connect_timeout = xprt->timeout->to_initval; 337 xprt->max_reconnect_timeout = xprt->timeout->to_maxval; 338 xprt->bind_timeout = RPCRDMA_BIND_TO; 339 xprt->reestablish_timeout = RPCRDMA_INIT_REEST_TO; 340 xprt->idle_timeout = RPCRDMA_IDLE_DISC_TO; 341 342 xprt->resvport = 0; /* privileged port not needed */ 343 xprt->ops = &xprt_rdma_procs; 344 345 /* 346 * Set up RDMA-specific connect data. 347 */ 348 sap = args->dstaddr; 349 350 /* Ensure xprt->addr holds valid server TCP (not RDMA) 351 * address, for any side protocols which peek at it */ 352 xprt->prot = IPPROTO_TCP; 353 xprt->xprt_class = &xprt_rdma; 354 xprt->addrlen = args->addrlen; 355 memcpy(&xprt->addr, sap, xprt->addrlen); 356 357 if (rpc_get_port(sap)) 358 xprt_set_bound(xprt); 359 xprt_rdma_format_addresses(xprt, sap); 360 361 new_xprt = rpcx_to_rdmax(xprt); 362 rc = rpcrdma_buffer_create(new_xprt); 363 if (rc) { 364 xprt_rdma_free_addresses(xprt); 365 xprt_free(xprt); 366 module_put(THIS_MODULE); 367 return ERR_PTR(rc); 368 } 369 370 INIT_DELAYED_WORK(&new_xprt->rx_connect_worker, 371 xprt_rdma_connect_worker); 372 373 xprt->max_payload = RPCRDMA_MAX_DATA_SEGS << PAGE_SHIFT; 374 375 return xprt; 376 } 377 378 /** 379 * xprt_rdma_close - close a transport connection 380 * @xprt: transport context 381 * 382 * Called during autoclose or device removal. 383 * 384 * Caller holds @xprt's send lock to prevent activity on this 385 * transport while the connection is torn down. 386 */ 387 void xprt_rdma_close(struct rpc_xprt *xprt) 388 { 389 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 390 391 rpcrdma_xprt_disconnect(r_xprt); 392 393 xprt->reestablish_timeout = 0; 394 ++xprt->connect_cookie; 395 xprt_disconnect_done(xprt); 396 } 397 398 /** 399 * xprt_rdma_set_port - update server port with rpcbind result 400 * @xprt: controlling RPC transport 401 * @port: new port value 402 * 403 * Transport connect status is unchanged. 404 */ 405 static void 406 xprt_rdma_set_port(struct rpc_xprt *xprt, u16 port) 407 { 408 struct sockaddr *sap = (struct sockaddr *)&xprt->addr; 409 char buf[8]; 410 411 rpc_set_port(sap, port); 412 413 kfree(xprt->address_strings[RPC_DISPLAY_PORT]); 414 snprintf(buf, sizeof(buf), "%u", port); 415 xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL); 416 417 kfree(xprt->address_strings[RPC_DISPLAY_HEX_PORT]); 418 snprintf(buf, sizeof(buf), "%4hx", port); 419 xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL); 420 } 421 422 /** 423 * xprt_rdma_timer - invoked when an RPC times out 424 * @xprt: controlling RPC transport 425 * @task: RPC task that timed out 426 * 427 * Invoked when the transport is still connected, but an RPC 428 * retransmit timeout occurs. 429 * 430 * Since RDMA connections don't have a keep-alive, forcibly 431 * disconnect and retry to connect. This drives full 432 * detection of the network path, and retransmissions of 433 * all pending RPCs. 434 */ 435 static void 436 xprt_rdma_timer(struct rpc_xprt *xprt, struct rpc_task *task) 437 { 438 xprt_force_disconnect(xprt); 439 } 440 441 /** 442 * xprt_rdma_set_connect_timeout - set timeouts for establishing a connection 443 * @xprt: controlling transport instance 444 * @connect_timeout: reconnect timeout after client disconnects 445 * @reconnect_timeout: reconnect timeout after server disconnects 446 * 447 */ 448 static void xprt_rdma_set_connect_timeout(struct rpc_xprt *xprt, 449 unsigned long connect_timeout, 450 unsigned long reconnect_timeout) 451 { 452 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 453 454 trace_xprtrdma_op_set_cto(r_xprt, connect_timeout, reconnect_timeout); 455 456 spin_lock(&xprt->transport_lock); 457 458 if (connect_timeout < xprt->connect_timeout) { 459 struct rpc_timeout to; 460 unsigned long initval; 461 462 to = *xprt->timeout; 463 initval = connect_timeout; 464 if (initval < RPCRDMA_INIT_REEST_TO << 1) 465 initval = RPCRDMA_INIT_REEST_TO << 1; 466 to.to_initval = initval; 467 to.to_maxval = initval; 468 r_xprt->rx_timeout = to; 469 xprt->timeout = &r_xprt->rx_timeout; 470 xprt->connect_timeout = connect_timeout; 471 } 472 473 if (reconnect_timeout < xprt->max_reconnect_timeout) 474 xprt->max_reconnect_timeout = reconnect_timeout; 475 476 spin_unlock(&xprt->transport_lock); 477 } 478 479 /** 480 * xprt_rdma_connect - schedule an attempt to reconnect 481 * @xprt: transport state 482 * @task: RPC scheduler context (unused) 483 * 484 */ 485 static void 486 xprt_rdma_connect(struct rpc_xprt *xprt, struct rpc_task *task) 487 { 488 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 489 struct rpcrdma_ep *ep = r_xprt->rx_ep; 490 unsigned long delay; 491 492 delay = 0; 493 if (ep && ep->re_connect_status != 0) { 494 delay = xprt_reconnect_delay(xprt); 495 xprt_reconnect_backoff(xprt, RPCRDMA_INIT_REEST_TO); 496 } 497 trace_xprtrdma_op_connect(r_xprt, delay); 498 queue_delayed_work(xprtiod_workqueue, &r_xprt->rx_connect_worker, 499 delay); 500 } 501 502 /** 503 * xprt_rdma_alloc_slot - allocate an rpc_rqst 504 * @xprt: controlling RPC transport 505 * @task: RPC task requesting a fresh rpc_rqst 506 * 507 * tk_status values: 508 * %0 if task->tk_rqstp points to a fresh rpc_rqst 509 * %-EAGAIN if no rpc_rqst is available; queued on backlog 510 */ 511 static void 512 xprt_rdma_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task) 513 { 514 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 515 struct rpcrdma_req *req; 516 517 req = rpcrdma_buffer_get(&r_xprt->rx_buf); 518 if (!req) 519 goto out_sleep; 520 task->tk_rqstp = &req->rl_slot; 521 task->tk_status = 0; 522 return; 523 524 out_sleep: 525 task->tk_status = -EAGAIN; 526 xprt_add_backlog(xprt, task); 527 } 528 529 /** 530 * xprt_rdma_free_slot - release an rpc_rqst 531 * @xprt: controlling RPC transport 532 * @rqst: rpc_rqst to release 533 * 534 */ 535 static void 536 xprt_rdma_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *rqst) 537 { 538 struct rpcrdma_xprt *r_xprt = 539 container_of(xprt, struct rpcrdma_xprt, rx_xprt); 540 541 rpcrdma_reply_put(&r_xprt->rx_buf, rpcr_to_rdmar(rqst)); 542 if (!xprt_wake_up_backlog(xprt, rqst)) { 543 memset(rqst, 0, sizeof(*rqst)); 544 rpcrdma_buffer_put(&r_xprt->rx_buf, rpcr_to_rdmar(rqst)); 545 } 546 } 547 548 static bool rpcrdma_check_regbuf(struct rpcrdma_xprt *r_xprt, 549 struct rpcrdma_regbuf *rb, size_t size, 550 gfp_t flags) 551 { 552 if (unlikely(rdmab_length(rb) < size)) { 553 if (!rpcrdma_regbuf_realloc(rb, size, flags)) 554 return false; 555 r_xprt->rx_stats.hardway_register_count += size; 556 } 557 return true; 558 } 559 560 /** 561 * xprt_rdma_allocate - allocate transport resources for an RPC 562 * @task: RPC task 563 * 564 * Return values: 565 * 0: Success; rq_buffer points to RPC buffer to use 566 * ENOMEM: Out of memory, call again later 567 * EIO: A permanent error occurred, do not retry 568 */ 569 static int 570 xprt_rdma_allocate(struct rpc_task *task) 571 { 572 struct rpc_rqst *rqst = task->tk_rqstp; 573 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(rqst->rq_xprt); 574 struct rpcrdma_req *req = rpcr_to_rdmar(rqst); 575 gfp_t flags; 576 577 flags = RPCRDMA_DEF_GFP; 578 if (RPC_IS_SWAPPER(task)) 579 flags = __GFP_MEMALLOC | GFP_NOWAIT | __GFP_NOWARN; 580 581 if (!rpcrdma_check_regbuf(r_xprt, req->rl_sendbuf, rqst->rq_callsize, 582 flags)) 583 goto out_fail; 584 if (!rpcrdma_check_regbuf(r_xprt, req->rl_recvbuf, rqst->rq_rcvsize, 585 flags)) 586 goto out_fail; 587 588 rqst->rq_buffer = rdmab_data(req->rl_sendbuf); 589 rqst->rq_rbuffer = rdmab_data(req->rl_recvbuf); 590 return 0; 591 592 out_fail: 593 return -ENOMEM; 594 } 595 596 /** 597 * xprt_rdma_free - release resources allocated by xprt_rdma_allocate 598 * @task: RPC task 599 * 600 * Caller guarantees rqst->rq_buffer is non-NULL. 601 */ 602 static void 603 xprt_rdma_free(struct rpc_task *task) 604 { 605 struct rpc_rqst *rqst = task->tk_rqstp; 606 struct rpcrdma_req *req = rpcr_to_rdmar(rqst); 607 608 if (unlikely(!list_empty(&req->rl_registered))) { 609 trace_xprtrdma_mrs_zap(task); 610 frwr_unmap_sync(rpcx_to_rdmax(rqst->rq_xprt), req); 611 } 612 613 /* XXX: If the RPC is completing because of a signal and 614 * not because a reply was received, we ought to ensure 615 * that the Send completion has fired, so that memory 616 * involved with the Send is not still visible to the NIC. 617 */ 618 } 619 620 /** 621 * xprt_rdma_send_request - marshal and send an RPC request 622 * @rqst: RPC message in rq_snd_buf 623 * 624 * Caller holds the transport's write lock. 625 * 626 * Returns: 627 * %0 if the RPC message has been sent 628 * %-ENOTCONN if the caller should reconnect and call again 629 * %-EAGAIN if the caller should call again 630 * %-ENOBUFS if the caller should call again after a delay 631 * %-EMSGSIZE if encoding ran out of buffer space. The request 632 * was not sent. Do not try to send this message again. 633 * %-EIO if an I/O error occurred. The request was not sent. 634 * Do not try to send this message again. 635 */ 636 static int 637 xprt_rdma_send_request(struct rpc_rqst *rqst) 638 { 639 struct rpc_xprt *xprt = rqst->rq_xprt; 640 struct rpcrdma_req *req = rpcr_to_rdmar(rqst); 641 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 642 int rc = 0; 643 644 #if defined(CONFIG_SUNRPC_BACKCHANNEL) 645 if (unlikely(!rqst->rq_buffer)) 646 return xprt_rdma_bc_send_reply(rqst); 647 #endif /* CONFIG_SUNRPC_BACKCHANNEL */ 648 649 if (!xprt_connected(xprt)) 650 return -ENOTCONN; 651 652 if (!xprt_request_get_cong(xprt, rqst)) 653 return -EBADSLT; 654 655 rc = rpcrdma_marshal_req(r_xprt, rqst); 656 if (rc < 0) 657 goto failed_marshal; 658 659 /* Must suppress retransmit to maintain credits */ 660 if (rqst->rq_connect_cookie == xprt->connect_cookie) 661 goto drop_connection; 662 rqst->rq_xtime = ktime_get(); 663 664 if (rpcrdma_post_sends(r_xprt, req)) 665 goto drop_connection; 666 667 rqst->rq_xmit_bytes_sent += rqst->rq_snd_buf.len; 668 669 /* An RPC with no reply will throw off credit accounting, 670 * so drop the connection to reset the credit grant. 671 */ 672 if (!rpc_reply_expected(rqst->rq_task)) 673 goto drop_connection; 674 return 0; 675 676 failed_marshal: 677 if (rc != -ENOTCONN) 678 return rc; 679 drop_connection: 680 xprt_rdma_close(xprt); 681 return -ENOTCONN; 682 } 683 684 void xprt_rdma_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 685 { 686 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 687 long idle_time = 0; 688 689 if (xprt_connected(xprt)) 690 idle_time = (long)(jiffies - xprt->last_used) / HZ; 691 692 seq_puts(seq, "\txprt:\trdma "); 693 seq_printf(seq, "%u %lu %lu %lu %ld %lu %lu %lu %llu %llu ", 694 0, /* need a local port? */ 695 xprt->stat.bind_count, 696 xprt->stat.connect_count, 697 xprt->stat.connect_time / HZ, 698 idle_time, 699 xprt->stat.sends, 700 xprt->stat.recvs, 701 xprt->stat.bad_xids, 702 xprt->stat.req_u, 703 xprt->stat.bklog_u); 704 seq_printf(seq, "%lu %lu %lu %llu %llu %llu %llu %lu %lu %lu %lu ", 705 r_xprt->rx_stats.read_chunk_count, 706 r_xprt->rx_stats.write_chunk_count, 707 r_xprt->rx_stats.reply_chunk_count, 708 r_xprt->rx_stats.total_rdma_request, 709 r_xprt->rx_stats.total_rdma_reply, 710 r_xprt->rx_stats.pullup_copy_count, 711 r_xprt->rx_stats.fixup_copy_count, 712 r_xprt->rx_stats.hardway_register_count, 713 r_xprt->rx_stats.failed_marshal_count, 714 r_xprt->rx_stats.bad_reply_count, 715 r_xprt->rx_stats.nomsg_call_count); 716 seq_printf(seq, "%lu %lu %lu %lu %lu %lu\n", 717 r_xprt->rx_stats.mrs_recycled, 718 r_xprt->rx_stats.mrs_orphaned, 719 r_xprt->rx_stats.mrs_allocated, 720 r_xprt->rx_stats.local_inv_needed, 721 r_xprt->rx_stats.empty_sendctx_q, 722 r_xprt->rx_stats.reply_waits_for_send); 723 } 724 725 static int 726 xprt_rdma_enable_swap(struct rpc_xprt *xprt) 727 { 728 return 0; 729 } 730 731 static void 732 xprt_rdma_disable_swap(struct rpc_xprt *xprt) 733 { 734 } 735 736 /* 737 * Plumbing for rpc transport switch and kernel module 738 */ 739 740 static const struct rpc_xprt_ops xprt_rdma_procs = { 741 .reserve_xprt = xprt_reserve_xprt_cong, 742 .release_xprt = xprt_release_xprt_cong, /* sunrpc/xprt.c */ 743 .alloc_slot = xprt_rdma_alloc_slot, 744 .free_slot = xprt_rdma_free_slot, 745 .release_request = xprt_release_rqst_cong, /* ditto */ 746 .wait_for_reply_request = xprt_wait_for_reply_request_def, /* ditto */ 747 .timer = xprt_rdma_timer, 748 .rpcbind = rpcb_getport_async, /* sunrpc/rpcb_clnt.c */ 749 .set_port = xprt_rdma_set_port, 750 .connect = xprt_rdma_connect, 751 .buf_alloc = xprt_rdma_allocate, 752 .buf_free = xprt_rdma_free, 753 .send_request = xprt_rdma_send_request, 754 .close = xprt_rdma_close, 755 .destroy = xprt_rdma_destroy, 756 .set_connect_timeout = xprt_rdma_set_connect_timeout, 757 .print_stats = xprt_rdma_print_stats, 758 .enable_swap = xprt_rdma_enable_swap, 759 .disable_swap = xprt_rdma_disable_swap, 760 .inject_disconnect = xprt_rdma_inject_disconnect, 761 #if defined(CONFIG_SUNRPC_BACKCHANNEL) 762 .bc_setup = xprt_rdma_bc_setup, 763 .bc_maxpayload = xprt_rdma_bc_maxpayload, 764 .bc_num_slots = xprt_rdma_bc_max_slots, 765 .bc_free_rqst = xprt_rdma_bc_free_rqst, 766 .bc_destroy = xprt_rdma_bc_destroy, 767 #endif 768 }; 769 770 static struct xprt_class xprt_rdma = { 771 .list = LIST_HEAD_INIT(xprt_rdma.list), 772 .name = "rdma", 773 .owner = THIS_MODULE, 774 .ident = XPRT_TRANSPORT_RDMA, 775 .setup = xprt_setup_rdma, 776 .netid = { "rdma", "rdma6", "" }, 777 }; 778 779 void xprt_rdma_cleanup(void) 780 { 781 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 782 if (sunrpc_table_header) { 783 unregister_sysctl_table(sunrpc_table_header); 784 sunrpc_table_header = NULL; 785 } 786 #endif 787 788 xprt_unregister_transport(&xprt_rdma); 789 xprt_unregister_transport(&xprt_rdma_bc); 790 } 791 792 int xprt_rdma_init(void) 793 { 794 int rc; 795 796 rc = xprt_register_transport(&xprt_rdma); 797 if (rc) 798 return rc; 799 800 rc = xprt_register_transport(&xprt_rdma_bc); 801 if (rc) { 802 xprt_unregister_transport(&xprt_rdma); 803 return rc; 804 } 805 806 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 807 if (!sunrpc_table_header) 808 sunrpc_table_header = register_sysctl_table(sunrpc_table); 809 #endif 810 return 0; 811 } 812