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 /** 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_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); 252 } 253 } else { 254 if (xprt_test_and_clear_connected(xprt)) 255 xprt_wake_pending_tasks(xprt, rc); 256 } 257 } 258 259 /** 260 * xprt_rdma_inject_disconnect - inject a connection fault 261 * @xprt: transport context 262 * 263 * If @xprt is connected, disconnect it to simulate spurious connection 264 * loss. 265 */ 266 static void 267 xprt_rdma_inject_disconnect(struct rpc_xprt *xprt) 268 { 269 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 270 271 trace_xprtrdma_inject_dsc(r_xprt); 272 rdma_disconnect(r_xprt->rx_ia.ri_id); 273 } 274 275 /** 276 * xprt_rdma_destroy - Full tear down of transport 277 * @xprt: doomed transport context 278 * 279 * Caller guarantees there will be no more calls to us with 280 * this @xprt. 281 */ 282 static void 283 xprt_rdma_destroy(struct rpc_xprt *xprt) 284 { 285 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 286 287 trace_xprtrdma_destroy(r_xprt); 288 289 cancel_delayed_work_sync(&r_xprt->rx_connect_worker); 290 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); 294 295 xprt_rdma_free_addresses(xprt); 296 xprt_free(xprt); 297 298 module_put(THIS_MODULE); 299 } 300 301 static const struct rpc_timeout xprt_rdma_default_timeout = { 302 .to_initval = 60 * HZ, 303 .to_maxval = 60 * HZ, 304 }; 305 306 /** 307 * xprt_setup_rdma - Set up transport to use RDMA 308 * 309 * @args: rpc transport arguments 310 */ 311 static struct rpc_xprt * 312 xprt_setup_rdma(struct xprt_create *args) 313 { 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; 319 int rc; 320 321 if (args->addrlen > sizeof(xprt->addr)) { 322 dprintk("RPC: %s: address too large\n", __func__); 323 return ERR_PTR(-EBADF); 324 } 325 326 xprt = xprt_alloc(args->net, sizeof(struct rpcrdma_xprt), 0, 0); 327 if (xprt == NULL) { 328 dprintk("RPC: %s: couldn't allocate rpcrdma_xprt\n", 329 __func__); 330 return ERR_PTR(-ENOMEM); 331 } 332 333 /* 60 second timeout, no retries */ 334 xprt->timeout = &xprt_rdma_default_timeout; 335 xprt->bind_timeout = RPCRDMA_BIND_TO; 336 xprt->reestablish_timeout = RPCRDMA_INIT_REEST_TO; 337 xprt->idle_timeout = RPCRDMA_IDLE_DISC_TO; 338 339 xprt->resvport = 0; /* privileged port not needed */ 340 xprt->tsh_size = 0; /* RPC-RDMA handles framing */ 341 xprt->ops = &xprt_rdma_procs; 342 343 /* 344 * Set up RDMA-specific connect data. 345 */ 346 sap = args->dstaddr; 347 348 /* Ensure xprt->addr holds valid server TCP (not RDMA) 349 * address, for any side protocols which peek at it */ 350 xprt->prot = IPPROTO_TCP; 351 xprt->addrlen = args->addrlen; 352 memcpy(&xprt->addr, sap, xprt->addrlen); 353 354 if (rpc_get_port(sap)) 355 xprt_set_bound(xprt); 356 xprt_rdma_format_addresses(xprt, sap); 357 358 cdata.max_requests = xprt_rdma_slot_table_entries; 359 360 cdata.rsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA write max */ 361 cdata.wsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA read max */ 362 363 cdata.inline_wsize = xprt_rdma_max_inline_write; 364 if (cdata.inline_wsize > cdata.wsize) 365 cdata.inline_wsize = cdata.wsize; 366 367 cdata.inline_rsize = xprt_rdma_max_inline_read; 368 if (cdata.inline_rsize > cdata.rsize) 369 cdata.inline_rsize = cdata.rsize; 370 371 /* 372 * Create new transport instance, which includes initialized 373 * o ia 374 * o endpoint 375 * o buffers 376 */ 377 378 new_xprt = rpcx_to_rdmax(xprt); 379 380 rc = rpcrdma_ia_open(new_xprt); 381 if (rc) 382 goto out1; 383 384 /* 385 * initialize and create ep 386 */ 387 new_xprt->rx_data = cdata; 388 new_ep = &new_xprt->rx_ep; 389 390 rc = rpcrdma_ep_create(&new_xprt->rx_ep, 391 &new_xprt->rx_ia, &new_xprt->rx_data); 392 if (rc) 393 goto out2; 394 395 rc = rpcrdma_buffer_create(new_xprt); 396 if (rc) 397 goto out3; 398 399 INIT_DELAYED_WORK(&new_xprt->rx_connect_worker, 400 xprt_rdma_connect_worker); 401 402 xprt->max_payload = new_xprt->rx_ia.ri_ops->ro_maxpages(new_xprt); 403 if (xprt->max_payload == 0) 404 goto out4; 405 xprt->max_payload <<= PAGE_SHIFT; 406 dprintk("RPC: %s: transport data payload maximum: %zu bytes\n", 407 __func__, xprt->max_payload); 408 409 if (!try_module_get(THIS_MODULE)) 410 goto out4; 411 412 dprintk("RPC: %s: %s:%s\n", __func__, 413 xprt->address_strings[RPC_DISPLAY_ADDR], 414 xprt->address_strings[RPC_DISPLAY_PORT]); 415 trace_xprtrdma_create(new_xprt); 416 return xprt; 417 418 out4: 419 rpcrdma_buffer_destroy(&new_xprt->rx_buf); 420 rc = -ENODEV; 421 out3: 422 rpcrdma_ep_destroy(new_ep, &new_xprt->rx_ia); 423 out2: 424 rpcrdma_ia_close(&new_xprt->rx_ia); 425 out1: 426 trace_xprtrdma_destroy(new_xprt); 427 xprt_rdma_free_addresses(xprt); 428 xprt_free(xprt); 429 return ERR_PTR(rc); 430 } 431 432 /** 433 * xprt_rdma_close - close a transport connection 434 * @xprt: transport context 435 * 436 * Called during transport shutdown, reconnect, or device removal. 437 * Caller holds @xprt's send lock to prevent activity on this 438 * transport while the connection is torn down. 439 */ 440 static void 441 xprt_rdma_close(struct rpc_xprt *xprt) 442 { 443 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 444 struct rpcrdma_ep *ep = &r_xprt->rx_ep; 445 struct rpcrdma_ia *ia = &r_xprt->rx_ia; 446 447 dprintk("RPC: %s: closing xprt %p\n", __func__, xprt); 448 449 if (test_and_clear_bit(RPCRDMA_IAF_REMOVING, &ia->ri_flags)) { 450 xprt_clear_connected(xprt); 451 rpcrdma_ia_remove(ia); 452 return; 453 } 454 if (ep->rep_connected == -ENODEV) 455 return; 456 if (ep->rep_connected > 0) 457 xprt->reestablish_timeout = 0; 458 xprt_disconnect_done(xprt); 459 rpcrdma_ep_disconnect(ep, ia); 460 461 /* Prepare @xprt for the next connection by reinitializing 462 * its credit grant to one (see RFC 8166, Section 3.3.3). 463 */ 464 r_xprt->rx_buf.rb_credits = 1; 465 xprt->cwnd = RPC_CWNDSHIFT; 466 } 467 468 /** 469 * xprt_rdma_set_port - update server port with rpcbind result 470 * @xprt: controlling RPC transport 471 * @port: new port value 472 * 473 * Transport connect status is unchanged. 474 */ 475 static void 476 xprt_rdma_set_port(struct rpc_xprt *xprt, u16 port) 477 { 478 struct sockaddr *sap = (struct sockaddr *)&xprt->addr; 479 char buf[8]; 480 481 dprintk("RPC: %s: setting port for xprt %p (%s:%s) to %u\n", 482 __func__, xprt, 483 xprt->address_strings[RPC_DISPLAY_ADDR], 484 xprt->address_strings[RPC_DISPLAY_PORT], 485 port); 486 487 rpc_set_port(sap, port); 488 489 kfree(xprt->address_strings[RPC_DISPLAY_PORT]); 490 snprintf(buf, sizeof(buf), "%u", port); 491 xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL); 492 493 kfree(xprt->address_strings[RPC_DISPLAY_HEX_PORT]); 494 snprintf(buf, sizeof(buf), "%4hx", port); 495 xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL); 496 } 497 498 /** 499 * xprt_rdma_timer - invoked when an RPC times out 500 * @xprt: controlling RPC transport 501 * @task: RPC task that timed out 502 * 503 * Invoked when the transport is still connected, but an RPC 504 * retransmit timeout occurs. 505 * 506 * Since RDMA connections don't have a keep-alive, forcibly 507 * disconnect and retry to connect. This drives full 508 * detection of the network path, and retransmissions of 509 * all pending RPCs. 510 */ 511 static void 512 xprt_rdma_timer(struct rpc_xprt *xprt, struct rpc_task *task) 513 { 514 xprt_force_disconnect(xprt); 515 } 516 517 /** 518 * xprt_rdma_connect - try to establish a transport connection 519 * @xprt: transport state 520 * @task: RPC scheduler context 521 * 522 */ 523 static void 524 xprt_rdma_connect(struct rpc_xprt *xprt, struct rpc_task *task) 525 { 526 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 527 528 if (r_xprt->rx_ep.rep_connected != 0) { 529 /* Reconnect */ 530 schedule_delayed_work(&r_xprt->rx_connect_worker, 531 xprt->reestablish_timeout); 532 xprt->reestablish_timeout <<= 1; 533 if (xprt->reestablish_timeout > RPCRDMA_MAX_REEST_TO) 534 xprt->reestablish_timeout = RPCRDMA_MAX_REEST_TO; 535 else if (xprt->reestablish_timeout < RPCRDMA_INIT_REEST_TO) 536 xprt->reestablish_timeout = RPCRDMA_INIT_REEST_TO; 537 } else { 538 schedule_delayed_work(&r_xprt->rx_connect_worker, 0); 539 if (!RPC_IS_ASYNC(task)) 540 flush_delayed_work(&r_xprt->rx_connect_worker); 541 } 542 } 543 544 /** 545 * xprt_rdma_alloc_slot - allocate an rpc_rqst 546 * @xprt: controlling RPC transport 547 * @task: RPC task requesting a fresh rpc_rqst 548 * 549 * tk_status values: 550 * %0 if task->tk_rqstp points to a fresh rpc_rqst 551 * %-EAGAIN if no rpc_rqst is available; queued on backlog 552 */ 553 static void 554 xprt_rdma_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task) 555 { 556 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 557 struct rpcrdma_req *req; 558 559 req = rpcrdma_buffer_get(&r_xprt->rx_buf); 560 if (!req) 561 goto out_sleep; 562 task->tk_rqstp = &req->rl_slot; 563 task->tk_status = 0; 564 return; 565 566 out_sleep: 567 rpc_sleep_on(&xprt->backlog, task, NULL); 568 task->tk_status = -EAGAIN; 569 } 570 571 /** 572 * xprt_rdma_free_slot - release an rpc_rqst 573 * @xprt: controlling RPC transport 574 * @rqst: rpc_rqst to release 575 * 576 */ 577 static void 578 xprt_rdma_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *rqst) 579 { 580 memset(rqst, 0, sizeof(*rqst)); 581 rpcrdma_buffer_put(rpcr_to_rdmar(rqst)); 582 rpc_wake_up_next(&xprt->backlog); 583 } 584 585 static bool 586 rpcrdma_get_sendbuf(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req, 587 size_t size, gfp_t flags) 588 { 589 struct rpcrdma_regbuf *rb; 590 591 if (req->rl_sendbuf && rdmab_length(req->rl_sendbuf) >= size) 592 return true; 593 594 rb = rpcrdma_alloc_regbuf(size, DMA_TO_DEVICE, flags); 595 if (IS_ERR(rb)) 596 return false; 597 598 rpcrdma_free_regbuf(req->rl_sendbuf); 599 r_xprt->rx_stats.hardway_register_count += size; 600 req->rl_sendbuf = rb; 601 return true; 602 } 603 604 /* The rq_rcv_buf is used only if a Reply chunk is necessary. 605 * The decision to use a Reply chunk is made later in 606 * rpcrdma_marshal_req. This buffer is registered at that time. 607 * 608 * Otherwise, the associated RPC Reply arrives in a separate 609 * Receive buffer, arbitrarily chosen by the HCA. The buffer 610 * allocated here for the RPC Reply is not utilized in that 611 * case. See rpcrdma_inline_fixup. 612 * 613 * A regbuf is used here to remember the buffer size. 614 */ 615 static bool 616 rpcrdma_get_recvbuf(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req, 617 size_t size, gfp_t flags) 618 { 619 struct rpcrdma_regbuf *rb; 620 621 if (req->rl_recvbuf && rdmab_length(req->rl_recvbuf) >= size) 622 return true; 623 624 rb = rpcrdma_alloc_regbuf(size, DMA_NONE, flags); 625 if (IS_ERR(rb)) 626 return false; 627 628 rpcrdma_free_regbuf(req->rl_recvbuf); 629 r_xprt->rx_stats.hardway_register_count += size; 630 req->rl_recvbuf = rb; 631 return true; 632 } 633 634 /** 635 * xprt_rdma_allocate - allocate transport resources for an RPC 636 * @task: RPC task 637 * 638 * Return values: 639 * 0: Success; rq_buffer points to RPC buffer to use 640 * ENOMEM: Out of memory, call again later 641 * EIO: A permanent error occurred, do not retry 642 */ 643 static int 644 xprt_rdma_allocate(struct rpc_task *task) 645 { 646 struct rpc_rqst *rqst = task->tk_rqstp; 647 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(rqst->rq_xprt); 648 struct rpcrdma_req *req = rpcr_to_rdmar(rqst); 649 gfp_t flags; 650 651 flags = RPCRDMA_DEF_GFP; 652 if (RPC_IS_SWAPPER(task)) 653 flags = __GFP_MEMALLOC | GFP_NOWAIT | __GFP_NOWARN; 654 655 if (!rpcrdma_get_sendbuf(r_xprt, req, rqst->rq_callsize, flags)) 656 goto out_fail; 657 if (!rpcrdma_get_recvbuf(r_xprt, req, rqst->rq_rcvsize, flags)) 658 goto out_fail; 659 660 rqst->rq_buffer = req->rl_sendbuf->rg_base; 661 rqst->rq_rbuffer = req->rl_recvbuf->rg_base; 662 trace_xprtrdma_allocate(task, req); 663 return 0; 664 665 out_fail: 666 trace_xprtrdma_allocate(task, NULL); 667 return -ENOMEM; 668 } 669 670 /** 671 * xprt_rdma_free - release resources allocated by xprt_rdma_allocate 672 * @task: RPC task 673 * 674 * Caller guarantees rqst->rq_buffer is non-NULL. 675 */ 676 static void 677 xprt_rdma_free(struct rpc_task *task) 678 { 679 struct rpc_rqst *rqst = task->tk_rqstp; 680 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(rqst->rq_xprt); 681 struct rpcrdma_req *req = rpcr_to_rdmar(rqst); 682 683 if (test_bit(RPCRDMA_REQ_F_PENDING, &req->rl_flags)) 684 rpcrdma_release_rqst(r_xprt, req); 685 trace_xprtrdma_rpc_done(task, req); 686 } 687 688 /** 689 * xprt_rdma_send_request - marshal and send an RPC request 690 * @rqst: RPC message in rq_snd_buf 691 * 692 * Caller holds the transport's write lock. 693 * 694 * Returns: 695 * %0 if the RPC message has been sent 696 * %-ENOTCONN if the caller should reconnect and call again 697 * %-EAGAIN if the caller should call again 698 * %-ENOBUFS if the caller should call again after a delay 699 * %-EIO if a permanent error occurred and the request was not 700 * sent. Do not try to send this message again. 701 */ 702 static int 703 xprt_rdma_send_request(struct rpc_rqst *rqst) 704 { 705 struct rpc_xprt *xprt = rqst->rq_xprt; 706 struct rpcrdma_req *req = rpcr_to_rdmar(rqst); 707 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 708 int rc = 0; 709 710 #if defined(CONFIG_SUNRPC_BACKCHANNEL) 711 if (unlikely(!rqst->rq_buffer)) 712 return xprt_rdma_bc_send_reply(rqst); 713 #endif /* CONFIG_SUNRPC_BACKCHANNEL */ 714 715 if (!xprt_connected(xprt)) 716 goto drop_connection; 717 718 if (!xprt_request_get_cong(xprt, rqst)) 719 return -EBADSLT; 720 721 rc = rpcrdma_marshal_req(r_xprt, rqst); 722 if (rc < 0) 723 goto failed_marshal; 724 725 /* Must suppress retransmit to maintain credits */ 726 if (rqst->rq_connect_cookie == xprt->connect_cookie) 727 goto drop_connection; 728 rqst->rq_xtime = ktime_get(); 729 730 __set_bit(RPCRDMA_REQ_F_PENDING, &req->rl_flags); 731 if (rpcrdma_ep_post(&r_xprt->rx_ia, &r_xprt->rx_ep, req)) 732 goto drop_connection; 733 734 rqst->rq_xmit_bytes_sent += rqst->rq_snd_buf.len; 735 rqst->rq_bytes_sent = 0; 736 737 /* An RPC with no reply will throw off credit accounting, 738 * so drop the connection to reset the credit grant. 739 */ 740 if (!rpc_reply_expected(rqst->rq_task)) 741 goto drop_connection; 742 return 0; 743 744 failed_marshal: 745 if (rc != -ENOTCONN) 746 return rc; 747 drop_connection: 748 xprt_disconnect_done(xprt); 749 return -ENOTCONN; /* implies disconnect */ 750 } 751 752 void xprt_rdma_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 753 { 754 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 755 long idle_time = 0; 756 757 if (xprt_connected(xprt)) 758 idle_time = (long)(jiffies - xprt->last_used) / HZ; 759 760 seq_puts(seq, "\txprt:\trdma "); 761 seq_printf(seq, "%u %lu %lu %lu %ld %lu %lu %lu %llu %llu ", 762 0, /* need a local port? */ 763 xprt->stat.bind_count, 764 xprt->stat.connect_count, 765 xprt->stat.connect_time / HZ, 766 idle_time, 767 xprt->stat.sends, 768 xprt->stat.recvs, 769 xprt->stat.bad_xids, 770 xprt->stat.req_u, 771 xprt->stat.bklog_u); 772 seq_printf(seq, "%lu %lu %lu %llu %llu %llu %llu %lu %lu %lu %lu ", 773 r_xprt->rx_stats.read_chunk_count, 774 r_xprt->rx_stats.write_chunk_count, 775 r_xprt->rx_stats.reply_chunk_count, 776 r_xprt->rx_stats.total_rdma_request, 777 r_xprt->rx_stats.total_rdma_reply, 778 r_xprt->rx_stats.pullup_copy_count, 779 r_xprt->rx_stats.fixup_copy_count, 780 r_xprt->rx_stats.hardway_register_count, 781 r_xprt->rx_stats.failed_marshal_count, 782 r_xprt->rx_stats.bad_reply_count, 783 r_xprt->rx_stats.nomsg_call_count); 784 seq_printf(seq, "%lu %lu %lu %lu %lu %lu\n", 785 r_xprt->rx_stats.mrs_recycled, 786 r_xprt->rx_stats.mrs_orphaned, 787 r_xprt->rx_stats.mrs_allocated, 788 r_xprt->rx_stats.local_inv_needed, 789 r_xprt->rx_stats.empty_sendctx_q, 790 r_xprt->rx_stats.reply_waits_for_send); 791 } 792 793 static int 794 xprt_rdma_enable_swap(struct rpc_xprt *xprt) 795 { 796 return 0; 797 } 798 799 static void 800 xprt_rdma_disable_swap(struct rpc_xprt *xprt) 801 { 802 } 803 804 /* 805 * Plumbing for rpc transport switch and kernel module 806 */ 807 808 static const struct rpc_xprt_ops xprt_rdma_procs = { 809 .reserve_xprt = xprt_reserve_xprt_cong, 810 .release_xprt = xprt_release_xprt_cong, /* sunrpc/xprt.c */ 811 .alloc_slot = xprt_rdma_alloc_slot, 812 .free_slot = xprt_rdma_free_slot, 813 .release_request = xprt_release_rqst_cong, /* ditto */ 814 .set_retrans_timeout = xprt_set_retrans_timeout_def, /* ditto */ 815 .timer = xprt_rdma_timer, 816 .rpcbind = rpcb_getport_async, /* sunrpc/rpcb_clnt.c */ 817 .set_port = xprt_rdma_set_port, 818 .connect = xprt_rdma_connect, 819 .buf_alloc = xprt_rdma_allocate, 820 .buf_free = xprt_rdma_free, 821 .send_request = xprt_rdma_send_request, 822 .close = xprt_rdma_close, 823 .destroy = xprt_rdma_destroy, 824 .print_stats = xprt_rdma_print_stats, 825 .enable_swap = xprt_rdma_enable_swap, 826 .disable_swap = xprt_rdma_disable_swap, 827 .inject_disconnect = xprt_rdma_inject_disconnect, 828 #if defined(CONFIG_SUNRPC_BACKCHANNEL) 829 .bc_setup = xprt_rdma_bc_setup, 830 .bc_up = xprt_rdma_bc_up, 831 .bc_maxpayload = xprt_rdma_bc_maxpayload, 832 .bc_free_rqst = xprt_rdma_bc_free_rqst, 833 .bc_destroy = xprt_rdma_bc_destroy, 834 #endif 835 }; 836 837 static struct xprt_class xprt_rdma = { 838 .list = LIST_HEAD_INIT(xprt_rdma.list), 839 .name = "rdma", 840 .owner = THIS_MODULE, 841 .ident = XPRT_TRANSPORT_RDMA, 842 .setup = xprt_setup_rdma, 843 }; 844 845 void xprt_rdma_cleanup(void) 846 { 847 int rc; 848 849 dprintk("RPCRDMA Module Removed, deregister RPC RDMA transport\n"); 850 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 851 if (sunrpc_table_header) { 852 unregister_sysctl_table(sunrpc_table_header); 853 sunrpc_table_header = NULL; 854 } 855 #endif 856 rc = xprt_unregister_transport(&xprt_rdma); 857 if (rc) 858 dprintk("RPC: %s: xprt_unregister returned %i\n", 859 __func__, rc); 860 861 rpcrdma_destroy_wq(); 862 863 rc = xprt_unregister_transport(&xprt_rdma_bc); 864 if (rc) 865 dprintk("RPC: %s: xprt_unregister(bc) returned %i\n", 866 __func__, rc); 867 } 868 869 int xprt_rdma_init(void) 870 { 871 int rc; 872 873 rc = rpcrdma_alloc_wq(); 874 if (rc) 875 return rc; 876 877 rc = xprt_register_transport(&xprt_rdma); 878 if (rc) { 879 rpcrdma_destroy_wq(); 880 return rc; 881 } 882 883 rc = xprt_register_transport(&xprt_rdma_bc); 884 if (rc) { 885 xprt_unregister_transport(&xprt_rdma); 886 rpcrdma_destroy_wq(); 887 return rc; 888 } 889 890 dprintk("RPCRDMA Module Init, register RPC RDMA transport\n"); 891 892 dprintk("Defaults:\n"); 893 dprintk("\tSlots %d\n" 894 "\tMaxInlineRead %d\n\tMaxInlineWrite %d\n", 895 xprt_rdma_slot_table_entries, 896 xprt_rdma_max_inline_read, xprt_rdma_max_inline_write); 897 dprintk("\tPadding 0\n\tMemreg %d\n", xprt_rdma_memreg_strategy); 898 899 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 900 if (!sunrpc_table_header) 901 sunrpc_table_header = register_sysctl_table(sunrpc_table); 902 #endif 903 return 0; 904 } 905