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 55 #include "xprt_rdma.h" 56 57 #ifdef RPC_DEBUG 58 # define RPCDBG_FACILITY RPCDBG_TRANS 59 #endif 60 61 MODULE_LICENSE("Dual BSD/GPL"); 62 63 MODULE_DESCRIPTION("RPC/RDMA Transport for Linux kernel NFS"); 64 MODULE_AUTHOR("Network Appliance, Inc."); 65 66 /* 67 * tunables 68 */ 69 70 static unsigned int xprt_rdma_slot_table_entries = RPCRDMA_DEF_SLOT_TABLE; 71 static unsigned int xprt_rdma_max_inline_read = RPCRDMA_DEF_INLINE; 72 static unsigned int xprt_rdma_max_inline_write = RPCRDMA_DEF_INLINE; 73 static unsigned int xprt_rdma_inline_write_padding; 74 static unsigned int xprt_rdma_memreg_strategy = RPCRDMA_FRMR; 75 int xprt_rdma_pad_optimize = 0; 76 77 #ifdef RPC_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 zero; 82 static unsigned int max_padding = PAGE_SIZE; 83 static unsigned int min_memreg = RPCRDMA_BOUNCEBUFFERS; 84 static unsigned int max_memreg = RPCRDMA_LAST - 1; 85 86 static struct ctl_table_header *sunrpc_table_header; 87 88 static ctl_table xr_tunables_table[] = { 89 { 90 .procname = "rdma_slot_table_entries", 91 .data = &xprt_rdma_slot_table_entries, 92 .maxlen = sizeof(unsigned int), 93 .mode = 0644, 94 .proc_handler = proc_dointvec_minmax, 95 .extra1 = &min_slot_table_size, 96 .extra2 = &max_slot_table_size 97 }, 98 { 99 .procname = "rdma_max_inline_read", 100 .data = &xprt_rdma_max_inline_read, 101 .maxlen = sizeof(unsigned int), 102 .mode = 0644, 103 .proc_handler = proc_dointvec, 104 }, 105 { 106 .procname = "rdma_max_inline_write", 107 .data = &xprt_rdma_max_inline_write, 108 .maxlen = sizeof(unsigned int), 109 .mode = 0644, 110 .proc_handler = proc_dointvec, 111 }, 112 { 113 .procname = "rdma_inline_write_padding", 114 .data = &xprt_rdma_inline_write_padding, 115 .maxlen = sizeof(unsigned int), 116 .mode = 0644, 117 .proc_handler = proc_dointvec_minmax, 118 .extra1 = &zero, 119 .extra2 = &max_padding, 120 }, 121 { 122 .procname = "rdma_memreg_strategy", 123 .data = &xprt_rdma_memreg_strategy, 124 .maxlen = sizeof(unsigned int), 125 .mode = 0644, 126 .proc_handler = proc_dointvec_minmax, 127 .extra1 = &min_memreg, 128 .extra2 = &max_memreg, 129 }, 130 { 131 .procname = "rdma_pad_optimize", 132 .data = &xprt_rdma_pad_optimize, 133 .maxlen = sizeof(unsigned int), 134 .mode = 0644, 135 .proc_handler = proc_dointvec, 136 }, 137 { }, 138 }; 139 140 static ctl_table sunrpc_table[] = { 141 { 142 .procname = "sunrpc", 143 .mode = 0555, 144 .child = xr_tunables_table 145 }, 146 { }, 147 }; 148 149 #endif 150 151 static struct rpc_xprt_ops xprt_rdma_procs; /* forward reference */ 152 153 static void 154 xprt_rdma_format_addresses(struct rpc_xprt *xprt) 155 { 156 struct sockaddr *sap = (struct sockaddr *) 157 &rpcx_to_rdmad(xprt).addr; 158 struct sockaddr_in *sin = (struct sockaddr_in *)sap; 159 char buf[64]; 160 161 (void)rpc_ntop(sap, buf, sizeof(buf)); 162 xprt->address_strings[RPC_DISPLAY_ADDR] = kstrdup(buf, GFP_KERNEL); 163 164 snprintf(buf, sizeof(buf), "%u", rpc_get_port(sap)); 165 xprt->address_strings[RPC_DISPLAY_PORT] = kstrdup(buf, GFP_KERNEL); 166 167 xprt->address_strings[RPC_DISPLAY_PROTO] = "rdma"; 168 169 snprintf(buf, sizeof(buf), "%08x", ntohl(sin->sin_addr.s_addr)); 170 xprt->address_strings[RPC_DISPLAY_HEX_ADDR] = kstrdup(buf, GFP_KERNEL); 171 172 snprintf(buf, sizeof(buf), "%4hx", rpc_get_port(sap)); 173 xprt->address_strings[RPC_DISPLAY_HEX_PORT] = kstrdup(buf, GFP_KERNEL); 174 175 /* netid */ 176 xprt->address_strings[RPC_DISPLAY_NETID] = "rdma"; 177 } 178 179 static void 180 xprt_rdma_free_addresses(struct rpc_xprt *xprt) 181 { 182 unsigned int i; 183 184 for (i = 0; i < RPC_DISPLAY_MAX; i++) 185 switch (i) { 186 case RPC_DISPLAY_PROTO: 187 case RPC_DISPLAY_NETID: 188 continue; 189 default: 190 kfree(xprt->address_strings[i]); 191 } 192 } 193 194 static void 195 xprt_rdma_connect_worker(struct work_struct *work) 196 { 197 struct rpcrdma_xprt *r_xprt = 198 container_of(work, struct rpcrdma_xprt, rdma_connect.work); 199 struct rpc_xprt *xprt = &r_xprt->xprt; 200 int rc = 0; 201 202 current->flags |= PF_FSTRANS; 203 xprt_clear_connected(xprt); 204 205 dprintk("RPC: %s: %sconnect\n", __func__, 206 r_xprt->rx_ep.rep_connected != 0 ? "re" : ""); 207 rc = rpcrdma_ep_connect(&r_xprt->rx_ep, &r_xprt->rx_ia); 208 if (rc) 209 xprt_wake_pending_tasks(xprt, rc); 210 211 dprintk("RPC: %s: exit\n", __func__); 212 xprt_clear_connecting(xprt); 213 current->flags &= ~PF_FSTRANS; 214 } 215 216 /* 217 * xprt_rdma_destroy 218 * 219 * Destroy the xprt. 220 * Free all memory associated with the object, including its own. 221 * NOTE: none of the *destroy methods free memory for their top-level 222 * objects, even though they may have allocated it (they do free 223 * private memory). It's up to the caller to handle it. In this 224 * case (RDMA transport), all structure memory is inlined with the 225 * struct rpcrdma_xprt. 226 */ 227 static void 228 xprt_rdma_destroy(struct rpc_xprt *xprt) 229 { 230 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 231 int rc; 232 233 dprintk("RPC: %s: called\n", __func__); 234 235 cancel_delayed_work_sync(&r_xprt->rdma_connect); 236 237 xprt_clear_connected(xprt); 238 239 rpcrdma_buffer_destroy(&r_xprt->rx_buf); 240 rc = rpcrdma_ep_destroy(&r_xprt->rx_ep, &r_xprt->rx_ia); 241 if (rc) 242 dprintk("RPC: %s: rpcrdma_ep_destroy returned %i\n", 243 __func__, rc); 244 rpcrdma_ia_close(&r_xprt->rx_ia); 245 246 xprt_rdma_free_addresses(xprt); 247 248 xprt_free(xprt); 249 250 dprintk("RPC: %s: returning\n", __func__); 251 252 module_put(THIS_MODULE); 253 } 254 255 static const struct rpc_timeout xprt_rdma_default_timeout = { 256 .to_initval = 60 * HZ, 257 .to_maxval = 60 * HZ, 258 }; 259 260 /** 261 * xprt_setup_rdma - Set up transport to use RDMA 262 * 263 * @args: rpc transport arguments 264 */ 265 static struct rpc_xprt * 266 xprt_setup_rdma(struct xprt_create *args) 267 { 268 struct rpcrdma_create_data_internal cdata; 269 struct rpc_xprt *xprt; 270 struct rpcrdma_xprt *new_xprt; 271 struct rpcrdma_ep *new_ep; 272 struct sockaddr_in *sin; 273 int rc; 274 275 if (args->addrlen > sizeof(xprt->addr)) { 276 dprintk("RPC: %s: address too large\n", __func__); 277 return ERR_PTR(-EBADF); 278 } 279 280 xprt = xprt_alloc(args->net, sizeof(struct rpcrdma_xprt), 281 xprt_rdma_slot_table_entries, 282 xprt_rdma_slot_table_entries); 283 if (xprt == NULL) { 284 dprintk("RPC: %s: couldn't allocate rpcrdma_xprt\n", 285 __func__); 286 return ERR_PTR(-ENOMEM); 287 } 288 289 /* 60 second timeout, no retries */ 290 xprt->timeout = &xprt_rdma_default_timeout; 291 xprt->bind_timeout = (60U * HZ); 292 xprt->reestablish_timeout = (5U * HZ); 293 xprt->idle_timeout = (5U * 60 * HZ); 294 295 xprt->resvport = 0; /* privileged port not needed */ 296 xprt->tsh_size = 0; /* RPC-RDMA handles framing */ 297 xprt->max_payload = RPCRDMA_MAX_DATA_SEGS * PAGE_SIZE; 298 xprt->ops = &xprt_rdma_procs; 299 300 /* 301 * Set up RDMA-specific connect data. 302 */ 303 304 /* Put server RDMA address in local cdata */ 305 memcpy(&cdata.addr, args->dstaddr, args->addrlen); 306 307 /* Ensure xprt->addr holds valid server TCP (not RDMA) 308 * address, for any side protocols which peek at it */ 309 xprt->prot = IPPROTO_TCP; 310 xprt->addrlen = args->addrlen; 311 memcpy(&xprt->addr, &cdata.addr, xprt->addrlen); 312 313 sin = (struct sockaddr_in *)&cdata.addr; 314 if (ntohs(sin->sin_port) != 0) 315 xprt_set_bound(xprt); 316 317 dprintk("RPC: %s: %pI4:%u\n", 318 __func__, &sin->sin_addr.s_addr, ntohs(sin->sin_port)); 319 320 /* Set max requests */ 321 cdata.max_requests = xprt->max_reqs; 322 323 /* Set some length limits */ 324 cdata.rsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA write max */ 325 cdata.wsize = RPCRDMA_MAX_SEGS * PAGE_SIZE; /* RDMA read max */ 326 327 cdata.inline_wsize = xprt_rdma_max_inline_write; 328 if (cdata.inline_wsize > cdata.wsize) 329 cdata.inline_wsize = cdata.wsize; 330 331 cdata.inline_rsize = xprt_rdma_max_inline_read; 332 if (cdata.inline_rsize > cdata.rsize) 333 cdata.inline_rsize = cdata.rsize; 334 335 cdata.padding = xprt_rdma_inline_write_padding; 336 337 /* 338 * Create new transport instance, which includes initialized 339 * o ia 340 * o endpoint 341 * o buffers 342 */ 343 344 new_xprt = rpcx_to_rdmax(xprt); 345 346 rc = rpcrdma_ia_open(new_xprt, (struct sockaddr *) &cdata.addr, 347 xprt_rdma_memreg_strategy); 348 if (rc) 349 goto out1; 350 351 /* 352 * initialize and create ep 353 */ 354 new_xprt->rx_data = cdata; 355 new_ep = &new_xprt->rx_ep; 356 new_ep->rep_remote_addr = cdata.addr; 357 358 rc = rpcrdma_ep_create(&new_xprt->rx_ep, 359 &new_xprt->rx_ia, &new_xprt->rx_data); 360 if (rc) 361 goto out2; 362 363 /* 364 * Allocate pre-registered send and receive buffers for headers and 365 * any inline data. Also specify any padding which will be provided 366 * from a preregistered zero buffer. 367 */ 368 rc = rpcrdma_buffer_create(&new_xprt->rx_buf, new_ep, &new_xprt->rx_ia, 369 &new_xprt->rx_data); 370 if (rc) 371 goto out3; 372 373 /* 374 * Register a callback for connection events. This is necessary because 375 * connection loss notification is async. We also catch connection loss 376 * when reaping receives. 377 */ 378 INIT_DELAYED_WORK(&new_xprt->rdma_connect, xprt_rdma_connect_worker); 379 new_ep->rep_func = rpcrdma_conn_func; 380 new_ep->rep_xprt = xprt; 381 382 xprt_rdma_format_addresses(xprt); 383 384 if (!try_module_get(THIS_MODULE)) 385 goto out4; 386 387 return xprt; 388 389 out4: 390 xprt_rdma_free_addresses(xprt); 391 rc = -EINVAL; 392 out3: 393 (void) rpcrdma_ep_destroy(new_ep, &new_xprt->rx_ia); 394 out2: 395 rpcrdma_ia_close(&new_xprt->rx_ia); 396 out1: 397 xprt_free(xprt); 398 return ERR_PTR(rc); 399 } 400 401 /* 402 * Close a connection, during shutdown or timeout/reconnect 403 */ 404 static void 405 xprt_rdma_close(struct rpc_xprt *xprt) 406 { 407 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 408 409 dprintk("RPC: %s: closing\n", __func__); 410 if (r_xprt->rx_ep.rep_connected > 0) 411 xprt->reestablish_timeout = 0; 412 xprt_disconnect_done(xprt); 413 (void) rpcrdma_ep_disconnect(&r_xprt->rx_ep, &r_xprt->rx_ia); 414 } 415 416 static void 417 xprt_rdma_set_port(struct rpc_xprt *xprt, u16 port) 418 { 419 struct sockaddr_in *sap; 420 421 sap = (struct sockaddr_in *)&xprt->addr; 422 sap->sin_port = htons(port); 423 sap = (struct sockaddr_in *)&rpcx_to_rdmad(xprt).addr; 424 sap->sin_port = htons(port); 425 dprintk("RPC: %s: %u\n", __func__, port); 426 } 427 428 static void 429 xprt_rdma_connect(struct rpc_task *task) 430 { 431 struct rpc_xprt *xprt = (struct rpc_xprt *)task->tk_xprt; 432 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 433 434 if (r_xprt->rx_ep.rep_connected != 0) { 435 /* Reconnect */ 436 schedule_delayed_work(&r_xprt->rdma_connect, 437 xprt->reestablish_timeout); 438 xprt->reestablish_timeout <<= 1; 439 if (xprt->reestablish_timeout > (30 * HZ)) 440 xprt->reestablish_timeout = (30 * HZ); 441 else if (xprt->reestablish_timeout < (5 * HZ)) 442 xprt->reestablish_timeout = (5 * HZ); 443 } else { 444 schedule_delayed_work(&r_xprt->rdma_connect, 0); 445 if (!RPC_IS_ASYNC(task)) 446 flush_delayed_work(&r_xprt->rdma_connect); 447 } 448 } 449 450 static int 451 xprt_rdma_reserve_xprt(struct rpc_xprt *xprt, struct rpc_task *task) 452 { 453 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 454 int credits = atomic_read(&r_xprt->rx_buf.rb_credits); 455 456 /* == RPC_CWNDSCALE @ init, but *after* setup */ 457 if (r_xprt->rx_buf.rb_cwndscale == 0UL) { 458 r_xprt->rx_buf.rb_cwndscale = xprt->cwnd; 459 dprintk("RPC: %s: cwndscale %lu\n", __func__, 460 r_xprt->rx_buf.rb_cwndscale); 461 BUG_ON(r_xprt->rx_buf.rb_cwndscale <= 0); 462 } 463 xprt->cwnd = credits * r_xprt->rx_buf.rb_cwndscale; 464 return xprt_reserve_xprt_cong(xprt, task); 465 } 466 467 /* 468 * The RDMA allocate/free functions need the task structure as a place 469 * to hide the struct rpcrdma_req, which is necessary for the actual send/recv 470 * sequence. For this reason, the recv buffers are attached to send 471 * buffers for portions of the RPC. Note that the RPC layer allocates 472 * both send and receive buffers in the same call. We may register 473 * the receive buffer portion when using reply chunks. 474 */ 475 static void * 476 xprt_rdma_allocate(struct rpc_task *task, size_t size) 477 { 478 struct rpc_xprt *xprt = task->tk_xprt; 479 struct rpcrdma_req *req, *nreq; 480 481 req = rpcrdma_buffer_get(&rpcx_to_rdmax(xprt)->rx_buf); 482 BUG_ON(NULL == req); 483 484 if (size > req->rl_size) { 485 dprintk("RPC: %s: size %zd too large for buffer[%zd]: " 486 "prog %d vers %d proc %d\n", 487 __func__, size, req->rl_size, 488 task->tk_client->cl_prog, task->tk_client->cl_vers, 489 task->tk_msg.rpc_proc->p_proc); 490 /* 491 * Outgoing length shortage. Our inline write max must have 492 * been configured to perform direct i/o. 493 * 494 * This is therefore a large metadata operation, and the 495 * allocate call was made on the maximum possible message, 496 * e.g. containing long filename(s) or symlink data. In 497 * fact, while these metadata operations *might* carry 498 * large outgoing payloads, they rarely *do*. However, we 499 * have to commit to the request here, so reallocate and 500 * register it now. The data path will never require this 501 * reallocation. 502 * 503 * If the allocation or registration fails, the RPC framework 504 * will (doggedly) retry. 505 */ 506 if (rpcx_to_rdmax(xprt)->rx_ia.ri_memreg_strategy == 507 RPCRDMA_BOUNCEBUFFERS) { 508 /* forced to "pure inline" */ 509 dprintk("RPC: %s: too much data (%zd) for inline " 510 "(r/w max %d/%d)\n", __func__, size, 511 rpcx_to_rdmad(xprt).inline_rsize, 512 rpcx_to_rdmad(xprt).inline_wsize); 513 size = req->rl_size; 514 rpc_exit(task, -EIO); /* fail the operation */ 515 rpcx_to_rdmax(xprt)->rx_stats.failed_marshal_count++; 516 goto out; 517 } 518 if (task->tk_flags & RPC_TASK_SWAPPER) 519 nreq = kmalloc(sizeof *req + size, GFP_ATOMIC); 520 else 521 nreq = kmalloc(sizeof *req + size, GFP_NOFS); 522 if (nreq == NULL) 523 goto outfail; 524 525 if (rpcrdma_register_internal(&rpcx_to_rdmax(xprt)->rx_ia, 526 nreq->rl_base, size + sizeof(struct rpcrdma_req) 527 - offsetof(struct rpcrdma_req, rl_base), 528 &nreq->rl_handle, &nreq->rl_iov)) { 529 kfree(nreq); 530 goto outfail; 531 } 532 rpcx_to_rdmax(xprt)->rx_stats.hardway_register_count += size; 533 nreq->rl_size = size; 534 nreq->rl_niovs = 0; 535 nreq->rl_nchunks = 0; 536 nreq->rl_buffer = (struct rpcrdma_buffer *)req; 537 nreq->rl_reply = req->rl_reply; 538 memcpy(nreq->rl_segments, 539 req->rl_segments, sizeof nreq->rl_segments); 540 /* flag the swap with an unused field */ 541 nreq->rl_iov.length = 0; 542 req->rl_reply = NULL; 543 req = nreq; 544 } 545 dprintk("RPC: %s: size %zd, request 0x%p\n", __func__, size, req); 546 out: 547 req->rl_connect_cookie = 0; /* our reserved value */ 548 return req->rl_xdr_buf; 549 550 outfail: 551 rpcrdma_buffer_put(req); 552 rpcx_to_rdmax(xprt)->rx_stats.failed_marshal_count++; 553 return NULL; 554 } 555 556 /* 557 * This function returns all RDMA resources to the pool. 558 */ 559 static void 560 xprt_rdma_free(void *buffer) 561 { 562 struct rpcrdma_req *req; 563 struct rpcrdma_xprt *r_xprt; 564 struct rpcrdma_rep *rep; 565 int i; 566 567 if (buffer == NULL) 568 return; 569 570 req = container_of(buffer, struct rpcrdma_req, rl_xdr_buf[0]); 571 if (req->rl_iov.length == 0) { /* see allocate above */ 572 r_xprt = container_of(((struct rpcrdma_req *) req->rl_buffer)->rl_buffer, 573 struct rpcrdma_xprt, rx_buf); 574 } else 575 r_xprt = container_of(req->rl_buffer, struct rpcrdma_xprt, rx_buf); 576 rep = req->rl_reply; 577 578 dprintk("RPC: %s: called on 0x%p%s\n", 579 __func__, rep, (rep && rep->rr_func) ? " (with waiter)" : ""); 580 581 /* 582 * Finish the deregistration. When using mw bind, this was 583 * begun in rpcrdma_reply_handler(). In all other modes, we 584 * do it here, in thread context. The process is considered 585 * complete when the rr_func vector becomes NULL - this 586 * was put in place during rpcrdma_reply_handler() - the wait 587 * call below will not block if the dereg is "done". If 588 * interrupted, our framework will clean up. 589 */ 590 for (i = 0; req->rl_nchunks;) { 591 --req->rl_nchunks; 592 i += rpcrdma_deregister_external( 593 &req->rl_segments[i], r_xprt, NULL); 594 } 595 596 if (rep && wait_event_interruptible(rep->rr_unbind, !rep->rr_func)) { 597 rep->rr_func = NULL; /* abandon the callback */ 598 req->rl_reply = NULL; 599 } 600 601 if (req->rl_iov.length == 0) { /* see allocate above */ 602 struct rpcrdma_req *oreq = (struct rpcrdma_req *)req->rl_buffer; 603 oreq->rl_reply = req->rl_reply; 604 (void) rpcrdma_deregister_internal(&r_xprt->rx_ia, 605 req->rl_handle, 606 &req->rl_iov); 607 kfree(req); 608 req = oreq; 609 } 610 611 /* Put back request+reply buffers */ 612 rpcrdma_buffer_put(req); 613 } 614 615 /* 616 * send_request invokes the meat of RPC RDMA. It must do the following: 617 * 1. Marshal the RPC request into an RPC RDMA request, which means 618 * putting a header in front of data, and creating IOVs for RDMA 619 * from those in the request. 620 * 2. In marshaling, detect opportunities for RDMA, and use them. 621 * 3. Post a recv message to set up asynch completion, then send 622 * the request (rpcrdma_ep_post). 623 * 4. No partial sends are possible in the RPC-RDMA protocol (as in UDP). 624 */ 625 626 static int 627 xprt_rdma_send_request(struct rpc_task *task) 628 { 629 struct rpc_rqst *rqst = task->tk_rqstp; 630 struct rpc_xprt *xprt = task->tk_xprt; 631 struct rpcrdma_req *req = rpcr_to_rdmar(rqst); 632 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 633 634 /* marshal the send itself */ 635 if (req->rl_niovs == 0 && rpcrdma_marshal_req(rqst) != 0) { 636 r_xprt->rx_stats.failed_marshal_count++; 637 dprintk("RPC: %s: rpcrdma_marshal_req failed\n", 638 __func__); 639 return -EIO; 640 } 641 642 if (req->rl_reply == NULL) /* e.g. reconnection */ 643 rpcrdma_recv_buffer_get(req); 644 645 if (req->rl_reply) { 646 req->rl_reply->rr_func = rpcrdma_reply_handler; 647 /* this need only be done once, but... */ 648 req->rl_reply->rr_xprt = xprt; 649 } 650 651 /* Must suppress retransmit to maintain credits */ 652 if (req->rl_connect_cookie == xprt->connect_cookie) 653 goto drop_connection; 654 req->rl_connect_cookie = xprt->connect_cookie; 655 656 if (rpcrdma_ep_post(&r_xprt->rx_ia, &r_xprt->rx_ep, req)) 657 goto drop_connection; 658 659 rqst->rq_xmit_bytes_sent += rqst->rq_snd_buf.len; 660 rqst->rq_bytes_sent = 0; 661 return 0; 662 663 drop_connection: 664 xprt_disconnect_done(xprt); 665 return -ENOTCONN; /* implies disconnect */ 666 } 667 668 static void xprt_rdma_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) 669 { 670 struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt); 671 long idle_time = 0; 672 673 if (xprt_connected(xprt)) 674 idle_time = (long)(jiffies - xprt->last_used) / HZ; 675 676 seq_printf(seq, 677 "\txprt:\trdma %u %lu %lu %lu %ld %lu %lu %lu %Lu %Lu " 678 "%lu %lu %lu %Lu %Lu %Lu %Lu %lu %lu %lu\n", 679 680 0, /* need a local port? */ 681 xprt->stat.bind_count, 682 xprt->stat.connect_count, 683 xprt->stat.connect_time, 684 idle_time, 685 xprt->stat.sends, 686 xprt->stat.recvs, 687 xprt->stat.bad_xids, 688 xprt->stat.req_u, 689 xprt->stat.bklog_u, 690 691 r_xprt->rx_stats.read_chunk_count, 692 r_xprt->rx_stats.write_chunk_count, 693 r_xprt->rx_stats.reply_chunk_count, 694 r_xprt->rx_stats.total_rdma_request, 695 r_xprt->rx_stats.total_rdma_reply, 696 r_xprt->rx_stats.pullup_copy_count, 697 r_xprt->rx_stats.fixup_copy_count, 698 r_xprt->rx_stats.hardway_register_count, 699 r_xprt->rx_stats.failed_marshal_count, 700 r_xprt->rx_stats.bad_reply_count); 701 } 702 703 /* 704 * Plumbing for rpc transport switch and kernel module 705 */ 706 707 static struct rpc_xprt_ops xprt_rdma_procs = { 708 .reserve_xprt = xprt_rdma_reserve_xprt, 709 .release_xprt = xprt_release_xprt_cong, /* sunrpc/xprt.c */ 710 .alloc_slot = xprt_alloc_slot, 711 .release_request = xprt_release_rqst_cong, /* ditto */ 712 .set_retrans_timeout = xprt_set_retrans_timeout_def, /* ditto */ 713 .rpcbind = rpcb_getport_async, /* sunrpc/rpcb_clnt.c */ 714 .set_port = xprt_rdma_set_port, 715 .connect = xprt_rdma_connect, 716 .buf_alloc = xprt_rdma_allocate, 717 .buf_free = xprt_rdma_free, 718 .send_request = xprt_rdma_send_request, 719 .close = xprt_rdma_close, 720 .destroy = xprt_rdma_destroy, 721 .print_stats = xprt_rdma_print_stats 722 }; 723 724 static struct xprt_class xprt_rdma = { 725 .list = LIST_HEAD_INIT(xprt_rdma.list), 726 .name = "rdma", 727 .owner = THIS_MODULE, 728 .ident = XPRT_TRANSPORT_RDMA, 729 .setup = xprt_setup_rdma, 730 }; 731 732 static void __exit xprt_rdma_cleanup(void) 733 { 734 int rc; 735 736 dprintk(KERN_INFO "RPCRDMA Module Removed, deregister RPC RDMA transport\n"); 737 #ifdef RPC_DEBUG 738 if (sunrpc_table_header) { 739 unregister_sysctl_table(sunrpc_table_header); 740 sunrpc_table_header = NULL; 741 } 742 #endif 743 rc = xprt_unregister_transport(&xprt_rdma); 744 if (rc) 745 dprintk("RPC: %s: xprt_unregister returned %i\n", 746 __func__, rc); 747 } 748 749 static int __init xprt_rdma_init(void) 750 { 751 int rc; 752 753 rc = xprt_register_transport(&xprt_rdma); 754 755 if (rc) 756 return rc; 757 758 dprintk(KERN_INFO "RPCRDMA Module Init, register RPC RDMA transport\n"); 759 760 dprintk(KERN_INFO "Defaults:\n"); 761 dprintk(KERN_INFO "\tSlots %d\n" 762 "\tMaxInlineRead %d\n\tMaxInlineWrite %d\n", 763 xprt_rdma_slot_table_entries, 764 xprt_rdma_max_inline_read, xprt_rdma_max_inline_write); 765 dprintk(KERN_INFO "\tPadding %d\n\tMemreg %d\n", 766 xprt_rdma_inline_write_padding, xprt_rdma_memreg_strategy); 767 768 #ifdef RPC_DEBUG 769 if (!sunrpc_table_header) 770 sunrpc_table_header = register_sysctl_table(sunrpc_table); 771 #endif 772 return 0; 773 } 774 775 module_init(xprt_rdma_init); 776 module_exit(xprt_rdma_cleanup); 777