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 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; 76 77 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 78 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; 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 = &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 void 229 rpcrdma_conn_func(struct rpcrdma_ep *ep) 230 { 231 schedule_delayed_work(&ep->rep_connect_worker, 0); 232 } 233 234 void 235 rpcrdma_connect_worker(struct work_struct *work) 236 { 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; 242 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); 247 } else { 248 if (xprt_test_and_clear_connected(xprt)) 249 xprt_wake_pending_tasks(xprt, -ENOTCONN); 250 } 251 spin_unlock_bh(&xprt->transport_lock); 252 } 253 254 static void 255 xprt_rdma_connect_worker(struct work_struct *work) 256 { 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; 260 int rc = 0; 261 262 xprt_clear_connected(xprt); 263 264 rc = rpcrdma_ep_connect(&r_xprt->rx_ep, &r_xprt->rx_ia); 265 if (rc) 266 xprt_wake_pending_tasks(xprt, rc); 267 268 xprt_clear_connecting(xprt); 269 } 270 271 static void 272 xprt_rdma_inject_disconnect(struct rpc_xprt *xprt) 273 { 274 struct rpcrdma_xprt *r_xprt = container_of(xprt, struct rpcrdma_xprt, 275 rx_xprt); 276 277 trace_xprtrdma_inject_dsc(r_xprt); 278 rdma_disconnect(r_xprt->rx_ia.ri_id); 279 } 280 281 /* 282 * xprt_rdma_destroy 283 * 284 * Destroy the xprt. 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. 291 */ 292 static void 293 xprt_rdma_destroy(struct rpc_xprt *xprt) 294 { 295 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 296 297 trace_xprtrdma_destroy(r_xprt); 298 299 cancel_delayed_work_sync(&r_xprt->rx_connect_worker); 300 301 xprt_clear_connected(xprt); 302 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); 306 307 xprt_rdma_free_addresses(xprt); 308 xprt_free(xprt); 309 310 module_put(THIS_MODULE); 311 } 312 313 static const struct rpc_timeout xprt_rdma_default_timeout = { 314 .to_initval = 60 * HZ, 315 .to_maxval = 60 * HZ, 316 }; 317 318 /** 319 * xprt_setup_rdma - Set up transport to use RDMA 320 * 321 * @args: rpc transport arguments 322 */ 323 static struct rpc_xprt * 324 xprt_setup_rdma(struct xprt_create *args) 325 { 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; 331 int rc; 332 333 if (args->addrlen > sizeof(xprt->addr)) { 334 dprintk("RPC: %s: address too large\n", __func__); 335 return ERR_PTR(-EBADF); 336 } 337 338 xprt = xprt_alloc(args->net, sizeof(struct rpcrdma_xprt), 0, 0); 339 if (xprt == NULL) { 340 dprintk("RPC: %s: couldn't allocate rpcrdma_xprt\n", 341 __func__); 342 return ERR_PTR(-ENOMEM); 343 } 344 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; 350 351 xprt->resvport = 0; /* privileged port not needed */ 352 xprt->tsh_size = 0; /* RPC-RDMA handles framing */ 353 xprt->ops = &xprt_rdma_procs; 354 355 /* 356 * Set up RDMA-specific connect data. 357 */ 358 sap = args->dstaddr; 359 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); 365 366 if (rpc_get_port(sap)) 367 xprt_set_bound(xprt); 368 xprt_rdma_format_addresses(xprt, sap); 369 370 cdata.max_requests = xprt_rdma_slot_table_entries; 371 372 cdata.rsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA write max */ 373 cdata.wsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA read max */ 374 375 cdata.inline_wsize = xprt_rdma_max_inline_write; 376 if (cdata.inline_wsize > cdata.wsize) 377 cdata.inline_wsize = cdata.wsize; 378 379 cdata.inline_rsize = xprt_rdma_max_inline_read; 380 if (cdata.inline_rsize > cdata.rsize) 381 cdata.inline_rsize = cdata.rsize; 382 383 /* 384 * Create new transport instance, which includes initialized 385 * o ia 386 * o endpoint 387 * o buffers 388 */ 389 390 new_xprt = rpcx_to_rdmax(xprt); 391 392 rc = rpcrdma_ia_open(new_xprt); 393 if (rc) 394 goto out1; 395 396 /* 397 * initialize and create ep 398 */ 399 new_xprt->rx_data = cdata; 400 new_ep = &new_xprt->rx_ep; 401 402 rc = rpcrdma_ep_create(&new_xprt->rx_ep, 403 &new_xprt->rx_ia, &new_xprt->rx_data); 404 if (rc) 405 goto out2; 406 407 rc = rpcrdma_buffer_create(new_xprt); 408 if (rc) 409 goto out3; 410 411 INIT_DELAYED_WORK(&new_xprt->rx_connect_worker, 412 xprt_rdma_connect_worker); 413 414 xprt->max_payload = new_xprt->rx_ia.ri_ops->ro_maxpages(new_xprt); 415 if (xprt->max_payload == 0) 416 goto out4; 417 xprt->max_payload <<= PAGE_SHIFT; 418 dprintk("RPC: %s: transport data payload maximum: %zu bytes\n", 419 __func__, xprt->max_payload); 420 421 if (!try_module_get(THIS_MODULE)) 422 goto out4; 423 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); 428 return xprt; 429 430 out4: 431 rpcrdma_buffer_destroy(&new_xprt->rx_buf); 432 rc = -ENODEV; 433 out3: 434 rpcrdma_ep_destroy(new_ep, &new_xprt->rx_ia); 435 out2: 436 rpcrdma_ia_close(&new_xprt->rx_ia); 437 out1: 438 trace_xprtrdma_destroy(new_xprt); 439 xprt_rdma_free_addresses(xprt); 440 xprt_free(xprt); 441 return ERR_PTR(rc); 442 } 443 444 /** 445 * xprt_rdma_close - Close down RDMA connection 446 * @xprt: generic transport to be closed 447 * 448 * Called during transport shutdown reconnect, or device 449 * removal. Caller holds the transport's write lock. 450 */ 451 static void 452 xprt_rdma_close(struct rpc_xprt *xprt) 453 { 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; 457 458 dprintk("RPC: %s: closing xprt %p\n", __func__, xprt); 459 460 if (test_and_clear_bit(RPCRDMA_IAF_REMOVING, &ia->ri_flags)) { 461 xprt_clear_connected(xprt); 462 rpcrdma_ia_remove(ia); 463 return; 464 } 465 if (ep->rep_connected == -ENODEV) 466 return; 467 if (ep->rep_connected > 0) 468 xprt->reestablish_timeout = 0; 469 xprt_disconnect_done(xprt); 470 rpcrdma_ep_disconnect(ep, ia); 471 } 472 473 /** 474 * xprt_rdma_set_port - update server port with rpcbind result 475 * @xprt: controlling RPC transport 476 * @port: new port value 477 * 478 * Transport connect status is unchanged. 479 */ 480 static void 481 xprt_rdma_set_port(struct rpc_xprt *xprt, u16 port) 482 { 483 struct sockaddr *sap = (struct sockaddr *)&xprt->addr; 484 char buf[8]; 485 486 dprintk("RPC: %s: setting port for xprt %p (%s:%s) to %u\n", 487 __func__, xprt, 488 xprt->address_strings[RPC_DISPLAY_ADDR], 489 xprt->address_strings[RPC_DISPLAY_PORT], 490 port); 491 492 rpc_set_port(sap, port); 493 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); 497 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); 501 } 502 503 /** 504 * xprt_rdma_timer - invoked when an RPC times out 505 * @xprt: controlling RPC transport 506 * @task: RPC task that timed out 507 * 508 * Invoked when the transport is still connected, but an RPC 509 * retransmit timeout occurs. 510 * 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 514 * all pending RPCs. 515 */ 516 static void 517 xprt_rdma_timer(struct rpc_xprt *xprt, struct rpc_task *task) 518 { 519 xprt_force_disconnect(xprt); 520 } 521 522 static void 523 xprt_rdma_connect(struct rpc_xprt *xprt, struct rpc_task *task) 524 { 525 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 526 527 if (r_xprt->rx_ep.rep_connected != 0) { 528 /* Reconnect */ 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; 536 } else { 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); 540 } 541 } 542 543 /** 544 * xprt_rdma_alloc_slot - allocate an rpc_rqst 545 * @xprt: controlling RPC transport 546 * @task: RPC task requesting a fresh rpc_rqst 547 * 548 * tk_status values: 549 * %0 if task->tk_rqstp points to a fresh rpc_rqst 550 * %-EAGAIN if no rpc_rqst is available; queued on backlog 551 */ 552 static void 553 xprt_rdma_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task) 554 { 555 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 556 struct rpcrdma_req *req; 557 558 req = rpcrdma_buffer_get(&r_xprt->rx_buf); 559 if (!req) 560 goto out_sleep; 561 task->tk_rqstp = &req->rl_slot; 562 task->tk_status = 0; 563 return; 564 565 out_sleep: 566 rpc_sleep_on(&xprt->backlog, task, NULL); 567 task->tk_status = -EAGAIN; 568 } 569 570 /** 571 * xprt_rdma_free_slot - release an rpc_rqst 572 * @xprt: controlling RPC transport 573 * @rqst: rpc_rqst to release 574 * 575 */ 576 static void 577 xprt_rdma_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *rqst) 578 { 579 memset(rqst, 0, sizeof(*rqst)); 580 rpcrdma_buffer_put(rpcr_to_rdmar(rqst)); 581 rpc_wake_up_next(&xprt->backlog); 582 } 583 584 static bool 585 rpcrdma_get_sendbuf(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req, 586 size_t size, gfp_t flags) 587 { 588 struct rpcrdma_regbuf *rb; 589 590 if (req->rl_sendbuf && rdmab_length(req->rl_sendbuf) >= size) 591 return true; 592 593 rb = rpcrdma_alloc_regbuf(size, DMA_TO_DEVICE, flags); 594 if (IS_ERR(rb)) 595 return false; 596 597 rpcrdma_free_regbuf(req->rl_sendbuf); 598 r_xprt->rx_stats.hardway_register_count += size; 599 req->rl_sendbuf = rb; 600 return true; 601 } 602 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. 606 * 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. 611 * 612 * A regbuf is used here to remember the buffer size. 613 */ 614 static bool 615 rpcrdma_get_recvbuf(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req, 616 size_t size, gfp_t flags) 617 { 618 struct rpcrdma_regbuf *rb; 619 620 if (req->rl_recvbuf && rdmab_length(req->rl_recvbuf) >= size) 621 return true; 622 623 rb = rpcrdma_alloc_regbuf(size, DMA_NONE, flags); 624 if (IS_ERR(rb)) 625 return false; 626 627 rpcrdma_free_regbuf(req->rl_recvbuf); 628 r_xprt->rx_stats.hardway_register_count += size; 629 req->rl_recvbuf = rb; 630 return true; 631 } 632 633 /** 634 * xprt_rdma_allocate - allocate transport resources for an RPC 635 * @task: RPC task 636 * 637 * Return values: 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 641 * 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. 645 * 646 * xprt_rdma_allocate provides buffers that are already mapped for 647 * DMA, and a local DMA lkey is provided for each. 648 */ 649 static int 650 xprt_rdma_allocate(struct rpc_task *task) 651 { 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); 655 gfp_t flags; 656 657 flags = RPCRDMA_DEF_GFP; 658 if (RPC_IS_SWAPPER(task)) 659 flags = __GFP_MEMALLOC | GFP_NOWAIT | __GFP_NOWARN; 660 661 if (!rpcrdma_get_sendbuf(r_xprt, req, rqst->rq_callsize, flags)) 662 goto out_fail; 663 if (!rpcrdma_get_recvbuf(r_xprt, req, rqst->rq_rcvsize, flags)) 664 goto out_fail; 665 666 rqst->rq_buffer = req->rl_sendbuf->rg_base; 667 rqst->rq_rbuffer = req->rl_recvbuf->rg_base; 668 trace_xprtrdma_allocate(task, req); 669 return 0; 670 671 out_fail: 672 trace_xprtrdma_allocate(task, NULL); 673 return -ENOMEM; 674 } 675 676 /** 677 * xprt_rdma_free - release resources allocated by xprt_rdma_allocate 678 * @task: RPC task 679 * 680 * Caller guarantees rqst->rq_buffer is non-NULL. 681 */ 682 static void 683 xprt_rdma_free(struct rpc_task *task) 684 { 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); 688 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); 692 } 693 694 /** 695 * xprt_rdma_send_request - marshal and send an RPC request 696 * @task: RPC task with an RPC message in rq_snd_buf 697 * 698 * Caller holds the transport's write lock. 699 * 700 * Returns: 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. 707 */ 708 static int 709 xprt_rdma_send_request(struct rpc_task *task) 710 { 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); 715 int rc = 0; 716 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 */ 721 722 if (!xprt_connected(xprt)) 723 goto drop_connection; 724 725 rc = rpcrdma_marshal_req(r_xprt, rqst); 726 if (rc < 0) 727 goto failed_marshal; 728 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(); 733 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; 737 738 rqst->rq_xmit_bytes_sent += rqst->rq_snd_buf.len; 739 rqst->rq_bytes_sent = 0; 740 741 /* An RPC with no reply will throw off credit accounting, 742 * so drop the connection to reset the credit grant. 743 */ 744 if (!rpc_reply_expected(task)) 745 goto drop_connection; 746 return 0; 747 748 failed_marshal: 749 if (rc != -ENOTCONN) 750 return rc; 751 drop_connection: 752 xprt_disconnect_done(xprt); 753 return -ENOTCONN; /* implies disconnect */ 754 } 755 756 void xprt_rdma_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 757 { 758 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 759 long idle_time = 0; 760 761 if (xprt_connected(xprt)) 762 idle_time = (long)(jiffies - xprt->last_used) / HZ; 763 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, 770 idle_time, 771 xprt->stat.sends, 772 xprt->stat.recvs, 773 xprt->stat.bad_xids, 774 xprt->stat.req_u, 775 xprt->stat.bklog_u); 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); 795 } 796 797 static int 798 xprt_rdma_enable_swap(struct rpc_xprt *xprt) 799 { 800 return 0; 801 } 802 803 static void 804 xprt_rdma_disable_swap(struct rpc_xprt *xprt) 805 { 806 } 807 808 /* 809 * Plumbing for rpc transport switch and kernel module 810 */ 811 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, 838 #endif 839 }; 840 841 static struct xprt_class xprt_rdma = { 842 .list = LIST_HEAD_INIT(xprt_rdma.list), 843 .name = "rdma", 844 .owner = THIS_MODULE, 845 .ident = XPRT_TRANSPORT_RDMA, 846 .setup = xprt_setup_rdma, 847 }; 848 849 void xprt_rdma_cleanup(void) 850 { 851 int rc; 852 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; 858 } 859 #endif 860 rc = xprt_unregister_transport(&xprt_rdma); 861 if (rc) 862 dprintk("RPC: %s: xprt_unregister returned %i\n", 863 __func__, rc); 864 865 rpcrdma_destroy_wq(); 866 867 rc = xprt_unregister_transport(&xprt_rdma_bc); 868 if (rc) 869 dprintk("RPC: %s: xprt_unregister(bc) returned %i\n", 870 __func__, rc); 871 } 872 873 int xprt_rdma_init(void) 874 { 875 int rc; 876 877 rc = rpcrdma_alloc_wq(); 878 if (rc) 879 return rc; 880 881 rc = xprt_register_transport(&xprt_rdma); 882 if (rc) { 883 rpcrdma_destroy_wq(); 884 return rc; 885 } 886 887 rc = xprt_register_transport(&xprt_rdma_bc); 888 if (rc) { 889 xprt_unregister_transport(&xprt_rdma); 890 rpcrdma_destroy_wq(); 891 return rc; 892 } 893 894 dprintk("RPCRDMA Module Init, register RPC RDMA transport\n"); 895 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); 902 903 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 904 if (!sunrpc_table_header) 905 sunrpc_table_header = register_sysctl_table(sunrpc_table); 906 #endif 907 return 0; 908 } 909