1/* 2 * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved. 3 * 4 * This software is available to you under a choice of one of two 5 * licenses. You may choose to be licensed under the terms of the GNU 6 * General Public License (GPL) Version 2, available from the file 7 * COPYING in the main directory of this source tree, or the BSD-type 8 * license below: 9 * 10 * Redistribution and use in source and binary forms, with or without 11 * modification, are permitted provided that the following conditions 12 * are met: 13 * 14 * Redistributions of source code must retain the above copyright 15 * notice, this list of conditions and the following disclaimer. 16 * 17 * Redistributions in binary form must reproduce the above 18 * copyright notice, this list of conditions and the following 19 * disclaimer in the documentation and/or other materials provided 20 * with the distribution. 21 * 22 * Neither the name of the Network Appliance, Inc. nor the names of 23 * its contributors may be used to endorse or promote products 24 * derived from this software without specific prior written 25 * permission. 26 * 27 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 28 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 29 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR 30 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT 31 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 32 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT 33 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, 34 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 35 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 36 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 37 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 38 */ 39 40/* 41 * transport.c 42 * 43 * This file contains the top-level implementation of an RPC RDMA 44 * transport. 45 * 46 * Naming convention: functions beginning with xprt_ are part of the 47 * transport switch. All others are RPC RDMA internal. 48 */ 49 50#include <linux/module.h> 51#include <linux/init.h> 52#include <linux/slab.h> 53#include <linux/seq_file.h> 54#include <linux/sunrpc/addr.h> 55 56#include "xprt_rdma.h" 57 58#if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 59# define RPCDBG_FACILITY RPCDBG_TRANS 60#endif 61 62MODULE_LICENSE("Dual BSD/GPL"); 63 64MODULE_DESCRIPTION("RPC/RDMA Transport for Linux kernel NFS"); 65MODULE_AUTHOR("Network Appliance, Inc."); 66 67/* 68 * tunables 69 */ 70 71static unsigned int xprt_rdma_slot_table_entries = RPCRDMA_DEF_SLOT_TABLE; 72static unsigned int xprt_rdma_max_inline_read = RPCRDMA_DEF_INLINE; 73static unsigned int xprt_rdma_max_inline_write = RPCRDMA_DEF_INLINE; 74static unsigned int xprt_rdma_inline_write_padding; 75static unsigned int xprt_rdma_memreg_strategy = RPCRDMA_FRMR; 76 int xprt_rdma_pad_optimize = 1; 77 78#if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 79 80static unsigned int min_slot_table_size = RPCRDMA_MIN_SLOT_TABLE; 81static unsigned int max_slot_table_size = RPCRDMA_MAX_SLOT_TABLE; 82static unsigned int zero; 83static unsigned int max_padding = PAGE_SIZE; 84static unsigned int min_memreg = RPCRDMA_BOUNCEBUFFERS; 85static unsigned int max_memreg = RPCRDMA_LAST - 1; 86 87static struct ctl_table_header *sunrpc_table_header; 88 89static struct ctl_table xr_tunables_table[] = { 90 { 91 .procname = "rdma_slot_table_entries", 92 .data = &xprt_rdma_slot_table_entries, 93 .maxlen = sizeof(unsigned int), 94 .mode = 0644, 95 .proc_handler = proc_dointvec_minmax, 96 .extra1 = &min_slot_table_size, 97 .extra2 = &max_slot_table_size 98 }, 99 { 100 .procname = "rdma_max_inline_read", 101 .data = &xprt_rdma_max_inline_read, 102 .maxlen = sizeof(unsigned int), 103 .mode = 0644, 104 .proc_handler = proc_dointvec, 105 }, 106 { 107 .procname = "rdma_max_inline_write", 108 .data = &xprt_rdma_max_inline_write, 109 .maxlen = sizeof(unsigned int), 110 .mode = 0644, 111 .proc_handler = proc_dointvec, 112 }, 113 { 114 .procname = "rdma_inline_write_padding", 115 .data = &xprt_rdma_inline_write_padding, 116 .maxlen = sizeof(unsigned int), 117 .mode = 0644, 118 .proc_handler = proc_dointvec_minmax, 119 .extra1 = &zero, 120 .extra2 = &max_padding, 121 }, 122 { 123 .procname = "rdma_memreg_strategy", 124 .data = &xprt_rdma_memreg_strategy, 125 .maxlen = sizeof(unsigned int), 126 .mode = 0644, 127 .proc_handler = proc_dointvec_minmax, 128 .extra1 = &min_memreg, 129 .extra2 = &max_memreg, 130 }, 131 { 132 .procname = "rdma_pad_optimize", 133 .data = &xprt_rdma_pad_optimize, 134 .maxlen = sizeof(unsigned int), 135 .mode = 0644, 136 .proc_handler = proc_dointvec, 137 }, 138 { }, 139}; 140 141static struct ctl_table sunrpc_table[] = { 142 { 143 .procname = "sunrpc", 144 .mode = 0555, 145 .child = xr_tunables_table 146 }, 147 { }, 148}; 149 150#endif 151 152#define RPCRDMA_BIND_TO (60U * HZ) 153#define RPCRDMA_INIT_REEST_TO (5U * HZ) 154#define RPCRDMA_MAX_REEST_TO (30U * HZ) 155#define RPCRDMA_IDLE_DISC_TO (5U * 60 * HZ) 156 157static struct rpc_xprt_ops xprt_rdma_procs; /* forward reference */ 158 159static void 160xprt_rdma_format_addresses4(struct rpc_xprt *xprt, struct sockaddr *sap) 161{ 162 struct sockaddr_in *sin = (struct sockaddr_in *)sap; 163 char buf[20]; 164 165 snprintf(buf, sizeof(buf), "%08x", ntohl(sin->sin_addr.s_addr)); 166 xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL); 167 168 xprt->address_strings[RPC_DISPLAY_NETID] = RPCBIND_NETID_RDMA; 169} 170 171static void 172xprt_rdma_format_addresses6(struct rpc_xprt *xprt, struct sockaddr *sap) 173{ 174 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)sap; 175 char buf[40]; 176 177 snprintf(buf, sizeof(buf), "%pi6", &sin6->sin6_addr); 178 xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL); 179 180 xprt->address_strings[RPC_DISPLAY_NETID] = RPCBIND_NETID_RDMA6; 181} 182 183static void 184xprt_rdma_format_addresses(struct rpc_xprt *xprt) 185{ 186 struct sockaddr *sap = (struct sockaddr *) 187 &rpcx_to_rdmad(xprt).addr; 188 char buf[128]; 189 190 switch (sap->sa_family) { 191 case AF_INET: 192 xprt_rdma_format_addresses4(xprt, sap); 193 break; 194 case AF_INET6: 195 xprt_rdma_format_addresses6(xprt, sap); 196 break; 197 default: 198 pr_err("rpcrdma: Unrecognized address family\n"); 199 return; 200 } 201 202 (void)rpc_ntop(sap, buf, sizeof(buf)); 203 xprt->address_strings[RPC_DISPLAY_ADDR] = kstrdup(buf, GFP_KERNEL); 204 205 snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap)); 206 xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL); 207 208 snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap)); 209 xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL); 210 211 xprt->address_strings[RPC_DISPLAY_PROTO] = "rdma"; 212} 213 214static void 215xprt_rdma_free_addresses(struct rpc_xprt *xprt) 216{ 217 unsigned int i; 218 219 for (i = 0; i < RPC_DISPLAY_MAX; i++) 220 switch (i) { 221 case RPC_DISPLAY_PROTO: 222 case RPC_DISPLAY_NETID: 223 continue; 224 default: 225 kfree(xprt->address_strings[i]); 226 } 227} 228 229static void 230xprt_rdma_connect_worker(struct work_struct *work) 231{ 232 struct rpcrdma_xprt *r_xprt = container_of(work, struct rpcrdma_xprt, 233 rx_connect_worker.work); 234 struct rpc_xprt *xprt = &r_xprt->rx_xprt; 235 int rc = 0; 236 237 xprt_clear_connected(xprt); 238 239 dprintk("RPC: %s: %sconnect\n", __func__, 240 r_xprt->rx_ep.rep_connected != 0 ? "re" : ""); 241 rc = rpcrdma_ep_connect(&r_xprt->rx_ep, &r_xprt->rx_ia); 242 if (rc) 243 xprt_wake_pending_tasks(xprt, rc); 244 245 dprintk("RPC: %s: exit\n", __func__); 246 xprt_clear_connecting(xprt); 247} 248 249/* 250 * xprt_rdma_destroy 251 * 252 * Destroy the xprt. 253 * Free all memory associated with the object, including its own. 254 * NOTE: none of the *destroy methods free memory for their top-level 255 * objects, even though they may have allocated it (they do free 256 * private memory). It's up to the caller to handle it. In this 257 * case (RDMA transport), all structure memory is inlined with the 258 * struct rpcrdma_xprt. 259 */ 260static void 261xprt_rdma_destroy(struct rpc_xprt *xprt) 262{ 263 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 264 265 dprintk("RPC: %s: called\n", __func__); 266 267 cancel_delayed_work_sync(&r_xprt->rx_connect_worker); 268 269 xprt_clear_connected(xprt); 270 271 rpcrdma_buffer_destroy(&r_xprt->rx_buf); 272 rpcrdma_ep_destroy(&r_xprt->rx_ep, &r_xprt->rx_ia); 273 rpcrdma_ia_close(&r_xprt->rx_ia); 274 275 xprt_rdma_free_addresses(xprt); 276 277 xprt_free(xprt); 278 279 dprintk("RPC: %s: returning\n", __func__); 280 281 module_put(THIS_MODULE); 282} 283 284static const struct rpc_timeout xprt_rdma_default_timeout = { 285 .to_initval = 60 * HZ, 286 .to_maxval = 60 * HZ, 287}; 288 289/** 290 * xprt_setup_rdma - Set up transport to use RDMA 291 * 292 * @args: rpc transport arguments 293 */ 294static struct rpc_xprt * 295xprt_setup_rdma(struct xprt_create *args) 296{ 297 struct rpcrdma_create_data_internal cdata; 298 struct rpc_xprt *xprt; 299 struct rpcrdma_xprt *new_xprt; 300 struct rpcrdma_ep *new_ep; 301 struct sockaddr_in *sin; 302 int rc; 303 304 if (args->addrlen > sizeof(xprt->addr)) { 305 dprintk("RPC: %s: address too large\n", __func__); 306 return ERR_PTR(-EBADF); 307 } 308 309 xprt = xprt_alloc(args->net, sizeof(struct rpcrdma_xprt), 310 xprt_rdma_slot_table_entries, 311 xprt_rdma_slot_table_entries); 312 if (xprt == NULL) { 313 dprintk("RPC: %s: couldn't allocate rpcrdma_xprt\n", 314 __func__); 315 return ERR_PTR(-ENOMEM); 316 } 317 318 /* 60 second timeout, no retries */ 319 xprt->timeout = &xprt_rdma_default_timeout; 320 xprt->bind_timeout = RPCRDMA_BIND_TO; 321 xprt->reestablish_timeout = RPCRDMA_INIT_REEST_TO; 322 xprt->idle_timeout = RPCRDMA_IDLE_DISC_TO; 323 324 xprt->resvport = 0; /* privileged port not needed */ 325 xprt->tsh_size = 0; /* RPC-RDMA handles framing */ 326 xprt->ops = &xprt_rdma_procs; 327 328 /* 329 * Set up RDMA-specific connect data. 330 */ 331 332 /* Put server RDMA address in local cdata */ 333 memcpy(&cdata.addr, args->dstaddr, args->addrlen); 334 335 /* Ensure xprt->addr holds valid server TCP (not RDMA) 336 * address, for any side protocols which peek at it */ 337 xprt->prot = IPPROTO_TCP; 338 xprt->addrlen = args->addrlen; 339 memcpy(&xprt->addr, &cdata.addr, xprt->addrlen); 340 341 sin = (struct sockaddr_in *)&cdata.addr; 342 if (ntohs(sin->sin_port) != 0) 343 xprt_set_bound(xprt); 344 345 dprintk("RPC: %s: %pI4:%u\n", 346 __func__, &sin->sin_addr.s_addr, ntohs(sin->sin_port)); 347 348 /* Set max requests */ 349 cdata.max_requests = xprt->max_reqs; 350 351 /* Set some length limits */ 352 cdata.rsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA write max */ 353 cdata.wsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA read max */ 354 355 cdata.inline_wsize = xprt_rdma_max_inline_write; 356 if (cdata.inline_wsize > cdata.wsize) 357 cdata.inline_wsize = cdata.wsize; 358 359 cdata.inline_rsize = xprt_rdma_max_inline_read; 360 if (cdata.inline_rsize > cdata.rsize) 361 cdata.inline_rsize = cdata.rsize; 362 363 cdata.padding = xprt_rdma_inline_write_padding; 364 365 /* 366 * Create new transport instance, which includes initialized 367 * o ia 368 * o endpoint 369 * o buffers 370 */ 371 372 new_xprt = rpcx_to_rdmax(xprt); 373 374 rc = rpcrdma_ia_open(new_xprt, (struct sockaddr *) &cdata.addr, 375 xprt_rdma_memreg_strategy); 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 new_ep->rep_remote_addr = cdata.addr; 385 386 rc = rpcrdma_ep_create(&new_xprt->rx_ep, 387 &new_xprt->rx_ia, &new_xprt->rx_data); 388 if (rc) 389 goto out2; 390 391 /* 392 * Allocate pre-registered send and receive buffers for headers and 393 * any inline data. Also specify any padding which will be provided 394 * from a preregistered zero buffer. 395 */ 396 rc = rpcrdma_buffer_create(new_xprt); 397 if (rc) 398 goto out3; 399 400 /* 401 * Register a callback for connection events. This is necessary because 402 * connection loss notification is async. We also catch connection loss 403 * when reaping receives. 404 */ 405 INIT_DELAYED_WORK(&new_xprt->rx_connect_worker, 406 xprt_rdma_connect_worker); 407 408 xprt_rdma_format_addresses(xprt); 409 xprt->max_payload = new_xprt->rx_ia.ri_ops->ro_maxpages(new_xprt); 410 if (xprt->max_payload == 0) 411 goto out4; 412 xprt->max_payload <<= PAGE_SHIFT; 413 dprintk("RPC: %s: transport data payload maximum: %zu bytes\n", 414 __func__, xprt->max_payload); 415 416 if (!try_module_get(THIS_MODULE)) 417 goto out4; 418 419 return xprt; 420 421out4: 422 xprt_rdma_free_addresses(xprt); 423 rc = -EINVAL; 424out3: 425 rpcrdma_ep_destroy(new_ep, &new_xprt->rx_ia); 426out2: 427 rpcrdma_ia_close(&new_xprt->rx_ia); 428out1: 429 xprt_free(xprt); 430 return ERR_PTR(rc); 431} 432 433/* 434 * Close a connection, during shutdown or timeout/reconnect 435 */ 436static void 437xprt_rdma_close(struct rpc_xprt *xprt) 438{ 439 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 440 441 dprintk("RPC: %s: closing\n", __func__); 442 if (r_xprt->rx_ep.rep_connected > 0) 443 xprt->reestablish_timeout = 0; 444 xprt_disconnect_done(xprt); 445 rpcrdma_ep_disconnect(&r_xprt->rx_ep, &r_xprt->rx_ia); 446} 447 448static void 449xprt_rdma_set_port(struct rpc_xprt *xprt, u16 port) 450{ 451 struct sockaddr_in *sap; 452 453 sap = (struct sockaddr_in *)&xprt->addr; 454 sap->sin_port = htons(port); 455 sap = (struct sockaddr_in *)&rpcx_to_rdmad(xprt).addr; 456 sap->sin_port = htons(port); 457 dprintk("RPC: %s: %u\n", __func__, port); 458} 459 460static void 461xprt_rdma_connect(struct rpc_xprt *xprt, struct rpc_task *task) 462{ 463 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 464 465 if (r_xprt->rx_ep.rep_connected != 0) { 466 /* Reconnect */ 467 schedule_delayed_work(&r_xprt->rx_connect_worker, 468 xprt->reestablish_timeout); 469 xprt->reestablish_timeout <<= 1; 470 if (xprt->reestablish_timeout > RPCRDMA_MAX_REEST_TO) 471 xprt->reestablish_timeout = RPCRDMA_MAX_REEST_TO; 472 else if (xprt->reestablish_timeout < RPCRDMA_INIT_REEST_TO) 473 xprt->reestablish_timeout = RPCRDMA_INIT_REEST_TO; 474 } else { 475 schedule_delayed_work(&r_xprt->rx_connect_worker, 0); 476 if (!RPC_IS_ASYNC(task)) 477 flush_delayed_work(&r_xprt->rx_connect_worker); 478 } 479} 480 481/* 482 * The RDMA allocate/free functions need the task structure as a place 483 * to hide the struct rpcrdma_req, which is necessary for the actual send/recv 484 * sequence. 485 * 486 * The RPC layer allocates both send and receive buffers in the same call 487 * (rq_send_buf and rq_rcv_buf are both part of a single contiguous buffer). 488 * We may register rq_rcv_buf when using reply chunks. 489 */ 490static void * 491xprt_rdma_allocate(struct rpc_task *task, size_t size) 492{ 493 struct rpc_xprt *xprt = task->tk_rqstp->rq_xprt; 494 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 495 struct rpcrdma_regbuf *rb; 496 struct rpcrdma_req *req; 497 size_t min_size; 498 gfp_t flags; 499 500 req = rpcrdma_buffer_get(&r_xprt->rx_buf); 501 if (req == NULL) 502 return NULL; 503 504 flags = GFP_NOIO | __GFP_NOWARN; 505 if (RPC_IS_SWAPPER(task)) 506 flags = __GFP_MEMALLOC | GFP_NOWAIT | __GFP_NOWARN; 507 508 if (req->rl_rdmabuf == NULL) 509 goto out_rdmabuf; 510 if (req->rl_sendbuf == NULL) 511 goto out_sendbuf; 512 if (size > req->rl_sendbuf->rg_size) 513 goto out_sendbuf; 514 515out: 516 dprintk("RPC: %s: size %zd, request 0x%p\n", __func__, size, req); 517 req->rl_connect_cookie = 0; /* our reserved value */ 518 return req->rl_sendbuf->rg_base; 519 520out_rdmabuf: 521 min_size = RPCRDMA_INLINE_WRITE_THRESHOLD(task->tk_rqstp); 522 rb = rpcrdma_alloc_regbuf(&r_xprt->rx_ia, min_size, flags); 523 if (IS_ERR(rb)) 524 goto out_fail; 525 req->rl_rdmabuf = rb; 526 527out_sendbuf: 528 /* XDR encoding and RPC/RDMA marshaling of this request has not 529 * yet occurred. Thus a lower bound is needed to prevent buffer 530 * overrun during marshaling. 531 * 532 * RPC/RDMA marshaling may choose to send payload bearing ops 533 * inline, if the result is smaller than the inline threshold. 534 * The value of the "size" argument accounts for header 535 * requirements but not for the payload in these cases. 536 * 537 * Likewise, allocate enough space to receive a reply up to the 538 * size of the inline threshold. 539 * 540 * It's unlikely that both the send header and the received 541 * reply will be large, but slush is provided here to allow 542 * flexibility when marshaling. 543 */ 544 min_size = RPCRDMA_INLINE_READ_THRESHOLD(task->tk_rqstp); 545 min_size += RPCRDMA_INLINE_WRITE_THRESHOLD(task->tk_rqstp); 546 if (size < min_size) 547 size = min_size; 548 549 rb = rpcrdma_alloc_regbuf(&r_xprt->rx_ia, size, flags); 550 if (IS_ERR(rb)) 551 goto out_fail; 552 rb->rg_owner = req; 553 554 r_xprt->rx_stats.hardway_register_count += size; 555 rpcrdma_free_regbuf(&r_xprt->rx_ia, req->rl_sendbuf); 556 req->rl_sendbuf = rb; 557 goto out; 558 559out_fail: 560 rpcrdma_buffer_put(req); 561 r_xprt->rx_stats.failed_marshal_count++; 562 return NULL; 563} 564 565/* 566 * This function returns all RDMA resources to the pool. 567 */ 568static void 569xprt_rdma_free(void *buffer) 570{ 571 struct rpcrdma_req *req; 572 struct rpcrdma_xprt *r_xprt; 573 struct rpcrdma_regbuf *rb; 574 int i; 575 576 if (buffer == NULL) 577 return; 578 579 rb = container_of(buffer, struct rpcrdma_regbuf, rg_base[0]); 580 req = rb->rg_owner; 581 r_xprt = container_of(req->rl_buffer, struct rpcrdma_xprt, rx_buf); 582 583 dprintk("RPC: %s: called on 0x%p\n", __func__, req->rl_reply); 584 585 for (i = 0; req->rl_nchunks;) { 586 --req->rl_nchunks; 587 i += r_xprt->rx_ia.ri_ops->ro_unmap(r_xprt, 588 &req->rl_segments[i]); 589 } 590 591 rpcrdma_buffer_put(req); 592} 593 594/* 595 * send_request invokes the meat of RPC RDMA. It must do the following: 596 * 1. Marshal the RPC request into an RPC RDMA request, which means 597 * putting a header in front of data, and creating IOVs for RDMA 598 * from those in the request. 599 * 2. In marshaling, detect opportunities for RDMA, and use them. 600 * 3. Post a recv message to set up asynch completion, then send 601 * the request (rpcrdma_ep_post). 602 * 4. No partial sends are possible in the RPC-RDMA protocol (as in UDP). 603 */ 604 605static int 606xprt_rdma_send_request(struct rpc_task *task) 607{ 608 struct rpc_rqst *rqst = task->tk_rqstp; 609 struct rpc_xprt *xprt = rqst->rq_xprt; 610 struct rpcrdma_req *req = rpcr_to_rdmar(rqst); 611 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 612 int rc = 0; 613 614 rc = rpcrdma_marshal_req(rqst); 615 if (rc < 0) 616 goto failed_marshal; 617 618 if (req->rl_reply == NULL) /* e.g. reconnection */ 619 rpcrdma_recv_buffer_get(req); 620 621 if (req->rl_reply) { 622 req->rl_reply->rr_func = rpcrdma_reply_handler; 623 /* this need only be done once, but... */ 624 req->rl_reply->rr_xprt = xprt; 625 } 626 627 /* Must suppress retransmit to maintain credits */ 628 if (req->rl_connect_cookie == xprt->connect_cookie) 629 goto drop_connection; 630 req->rl_connect_cookie = xprt->connect_cookie; 631 632 if (rpcrdma_ep_post(&r_xprt->rx_ia, &r_xprt->rx_ep, req)) 633 goto drop_connection; 634 635 rqst->rq_xmit_bytes_sent += rqst->rq_snd_buf.len; 636 rqst->rq_bytes_sent = 0; 637 return 0; 638 639failed_marshal: 640 r_xprt->rx_stats.failed_marshal_count++; 641 dprintk("RPC: %s: rpcrdma_marshal_req failed, status %i\n", 642 __func__, rc); 643 if (rc == -EIO) 644 return -EIO; 645drop_connection: 646 xprt_disconnect_done(xprt); 647 return -ENOTCONN; /* implies disconnect */ 648} 649 650static void xprt_rdma_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 651{ 652 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 653 long idle_time = 0; 654 655 if (xprt_connected(xprt)) 656 idle_time = (long)(jiffies - xprt->last_used) / HZ; 657 658 seq_printf(seq, 659 "\txprt:\trdma %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu " 660 "%lu %lu %lu %Lu %Lu %Lu %Lu %lu %lu %lu\n", 661 662 0, /* need a local port? */ 663 xprt->stat.bind_count, 664 xprt->stat.connect_count, 665 xprt->stat.connect_time, 666 idle_time, 667 xprt->stat.sends, 668 xprt->stat.recvs, 669 xprt->stat.bad_xids, 670 xprt->stat.req_u, 671 xprt->stat.bklog_u, 672 673 r_xprt->rx_stats.read_chunk_count, 674 r_xprt->rx_stats.write_chunk_count, 675 r_xprt->rx_stats.reply_chunk_count, 676 r_xprt->rx_stats.total_rdma_request, 677 r_xprt->rx_stats.total_rdma_reply, 678 r_xprt->rx_stats.pullup_copy_count, 679 r_xprt->rx_stats.fixup_copy_count, 680 r_xprt->rx_stats.hardway_register_count, 681 r_xprt->rx_stats.failed_marshal_count, 682 r_xprt->rx_stats.bad_reply_count); 683} 684 685/* 686 * Plumbing for rpc transport switch and kernel module 687 */ 688 689static struct rpc_xprt_ops xprt_rdma_procs = { 690 .reserve_xprt = xprt_reserve_xprt_cong, 691 .release_xprt = xprt_release_xprt_cong, /* sunrpc/xprt.c */ 692 .alloc_slot = xprt_alloc_slot, 693 .release_request = xprt_release_rqst_cong, /* ditto */ 694 .set_retrans_timeout = xprt_set_retrans_timeout_def, /* ditto */ 695 .rpcbind = rpcb_getport_async, /* sunrpc/rpcb_clnt.c */ 696 .set_port = xprt_rdma_set_port, 697 .connect = xprt_rdma_connect, 698 .buf_alloc = xprt_rdma_allocate, 699 .buf_free = xprt_rdma_free, 700 .send_request = xprt_rdma_send_request, 701 .close = xprt_rdma_close, 702 .destroy = xprt_rdma_destroy, 703 .print_stats = xprt_rdma_print_stats 704}; 705 706static struct xprt_class xprt_rdma = { 707 .list = LIST_HEAD_INIT(xprt_rdma.list), 708 .name = "rdma", 709 .owner = THIS_MODULE, 710 .ident = XPRT_TRANSPORT_RDMA, 711 .setup = xprt_setup_rdma, 712}; 713 714static void __exit xprt_rdma_cleanup(void) 715{ 716 int rc; 717 718 dprintk("RPCRDMA Module Removed, deregister RPC RDMA transport\n"); 719#if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 720 if (sunrpc_table_header) { 721 unregister_sysctl_table(sunrpc_table_header); 722 sunrpc_table_header = NULL; 723 } 724#endif 725 rc = xprt_unregister_transport(&xprt_rdma); 726 if (rc) 727 dprintk("RPC: %s: xprt_unregister returned %i\n", 728 __func__, rc); 729} 730 731static int __init xprt_rdma_init(void) 732{ 733 int rc; 734 735 rc = xprt_register_transport(&xprt_rdma); 736 737 if (rc) 738 return rc; 739 740 dprintk("RPCRDMA Module Init, register RPC RDMA transport\n"); 741 742 dprintk("Defaults:\n"); 743 dprintk("\tSlots %d\n" 744 "\tMaxInlineRead %d\n\tMaxInlineWrite %d\n", 745 xprt_rdma_slot_table_entries, 746 xprt_rdma_max_inline_read, xprt_rdma_max_inline_write); 747 dprintk("\tPadding %d\n\tMemreg %d\n", 748 xprt_rdma_inline_write_padding, xprt_rdma_memreg_strategy); 749 750#if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 751 if (!sunrpc_table_header) 752 sunrpc_table_header = register_sysctl_table(sunrpc_table); 753#endif 754 return 0; 755} 756 757module_init(xprt_rdma_init); 758module_exit(xprt_rdma_cleanup); 759