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