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 * verbs.c 42 * 43 * Encapsulates the major functions managing: 44 * o adapters 45 * o endpoints 46 * o connections 47 * o buffer memory 48 */ 49 50 #include <linux/interrupt.h> 51 #include <linux/slab.h> 52 #include <linux/prefetch.h> 53 #include <linux/sunrpc/addr.h> 54 #include <linux/sunrpc/svc_rdma.h> 55 #include <asm/bitops.h> 56 #include <linux/module.h> /* try_module_get()/module_put() */ 57 58 #include "xprt_rdma.h" 59 60 /* 61 * Globals/Macros 62 */ 63 64 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 65 # define RPCDBG_FACILITY RPCDBG_TRANS 66 #endif 67 68 /* 69 * internal functions 70 */ 71 72 static struct workqueue_struct *rpcrdma_receive_wq; 73 74 int 75 rpcrdma_alloc_wq(void) 76 { 77 struct workqueue_struct *recv_wq; 78 79 recv_wq = alloc_workqueue("xprtrdma_receive", 80 WQ_MEM_RECLAIM | WQ_UNBOUND | WQ_HIGHPRI, 81 0); 82 if (!recv_wq) 83 return -ENOMEM; 84 85 rpcrdma_receive_wq = recv_wq; 86 return 0; 87 } 88 89 void 90 rpcrdma_destroy_wq(void) 91 { 92 struct workqueue_struct *wq; 93 94 if (rpcrdma_receive_wq) { 95 wq = rpcrdma_receive_wq; 96 rpcrdma_receive_wq = NULL; 97 destroy_workqueue(wq); 98 } 99 } 100 101 static void 102 rpcrdma_qp_async_error_upcall(struct ib_event *event, void *context) 103 { 104 struct rpcrdma_ep *ep = context; 105 106 pr_err("rpcrdma: %s on device %s ep %p\n", 107 ib_event_msg(event->event), event->device->name, context); 108 109 if (ep->rep_connected == 1) { 110 ep->rep_connected = -EIO; 111 rpcrdma_conn_func(ep); 112 wake_up_all(&ep->rep_connect_wait); 113 } 114 } 115 116 /** 117 * rpcrdma_wc_send - Invoked by RDMA provider for each polled Send WC 118 * @cq: completion queue (ignored) 119 * @wc: completed WR 120 * 121 */ 122 static void 123 rpcrdma_wc_send(struct ib_cq *cq, struct ib_wc *wc) 124 { 125 /* WARNING: Only wr_cqe and status are reliable at this point */ 126 if (wc->status != IB_WC_SUCCESS && wc->status != IB_WC_WR_FLUSH_ERR) 127 pr_err("rpcrdma: Send: %s (%u/0x%x)\n", 128 ib_wc_status_msg(wc->status), 129 wc->status, wc->vendor_err); 130 } 131 132 /* Perform basic sanity checking to avoid using garbage 133 * to update the credit grant value. 134 */ 135 static void 136 rpcrdma_update_granted_credits(struct rpcrdma_rep *rep) 137 { 138 struct rpcrdma_msg *rmsgp = rdmab_to_msg(rep->rr_rdmabuf); 139 struct rpcrdma_buffer *buffer = &rep->rr_rxprt->rx_buf; 140 u32 credits; 141 142 if (rep->rr_len < RPCRDMA_HDRLEN_ERR) 143 return; 144 145 credits = be32_to_cpu(rmsgp->rm_credit); 146 if (credits == 0) 147 credits = 1; /* don't deadlock */ 148 else if (credits > buffer->rb_max_requests) 149 credits = buffer->rb_max_requests; 150 151 atomic_set(&buffer->rb_credits, credits); 152 } 153 154 /** 155 * rpcrdma_wc_receive - Invoked by RDMA provider for each polled Receive WC 156 * @cq: completion queue (ignored) 157 * @wc: completed WR 158 * 159 */ 160 static void 161 rpcrdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc) 162 { 163 struct ib_cqe *cqe = wc->wr_cqe; 164 struct rpcrdma_rep *rep = container_of(cqe, struct rpcrdma_rep, 165 rr_cqe); 166 167 /* WARNING: Only wr_id and status are reliable at this point */ 168 if (wc->status != IB_WC_SUCCESS) 169 goto out_fail; 170 171 /* status == SUCCESS means all fields in wc are trustworthy */ 172 if (wc->opcode != IB_WC_RECV) 173 return; 174 175 dprintk("RPC: %s: rep %p opcode 'recv', length %u: success\n", 176 __func__, rep, wc->byte_len); 177 178 rep->rr_len = wc->byte_len; 179 rep->rr_wc_flags = wc->wc_flags; 180 rep->rr_inv_rkey = wc->ex.invalidate_rkey; 181 182 ib_dma_sync_single_for_cpu(rep->rr_device, 183 rdmab_addr(rep->rr_rdmabuf), 184 rep->rr_len, DMA_FROM_DEVICE); 185 186 rpcrdma_update_granted_credits(rep); 187 188 out_schedule: 189 queue_work(rpcrdma_receive_wq, &rep->rr_work); 190 return; 191 192 out_fail: 193 if (wc->status != IB_WC_WR_FLUSH_ERR) 194 pr_err("rpcrdma: Recv: %s (%u/0x%x)\n", 195 ib_wc_status_msg(wc->status), 196 wc->status, wc->vendor_err); 197 rep->rr_len = RPCRDMA_BAD_LEN; 198 goto out_schedule; 199 } 200 201 static void 202 rpcrdma_update_connect_private(struct rpcrdma_xprt *r_xprt, 203 struct rdma_conn_param *param) 204 { 205 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 206 const struct rpcrdma_connect_private *pmsg = param->private_data; 207 unsigned int rsize, wsize; 208 209 /* Default settings for RPC-over-RDMA Version One */ 210 r_xprt->rx_ia.ri_reminv_expected = false; 211 rsize = RPCRDMA_V1_DEF_INLINE_SIZE; 212 wsize = RPCRDMA_V1_DEF_INLINE_SIZE; 213 214 if (pmsg && 215 pmsg->cp_magic == rpcrdma_cmp_magic && 216 pmsg->cp_version == RPCRDMA_CMP_VERSION) { 217 r_xprt->rx_ia.ri_reminv_expected = true; 218 rsize = rpcrdma_decode_buffer_size(pmsg->cp_send_size); 219 wsize = rpcrdma_decode_buffer_size(pmsg->cp_recv_size); 220 } 221 222 if (rsize < cdata->inline_rsize) 223 cdata->inline_rsize = rsize; 224 if (wsize < cdata->inline_wsize) 225 cdata->inline_wsize = wsize; 226 dprintk("RPC: %s: max send %u, max recv %u\n", 227 __func__, cdata->inline_wsize, cdata->inline_rsize); 228 rpcrdma_set_max_header_sizes(r_xprt); 229 } 230 231 static int 232 rpcrdma_conn_upcall(struct rdma_cm_id *id, struct rdma_cm_event *event) 233 { 234 struct rpcrdma_xprt *xprt = id->context; 235 struct rpcrdma_ia *ia = &xprt->rx_ia; 236 struct rpcrdma_ep *ep = &xprt->rx_ep; 237 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 238 struct sockaddr *sap = (struct sockaddr *)&ep->rep_remote_addr; 239 #endif 240 struct ib_qp_attr *attr = &ia->ri_qp_attr; 241 struct ib_qp_init_attr *iattr = &ia->ri_qp_init_attr; 242 int connstate = 0; 243 244 switch (event->event) { 245 case RDMA_CM_EVENT_ADDR_RESOLVED: 246 case RDMA_CM_EVENT_ROUTE_RESOLVED: 247 ia->ri_async_rc = 0; 248 complete(&ia->ri_done); 249 break; 250 case RDMA_CM_EVENT_ADDR_ERROR: 251 ia->ri_async_rc = -EHOSTUNREACH; 252 dprintk("RPC: %s: CM address resolution error, ep 0x%p\n", 253 __func__, ep); 254 complete(&ia->ri_done); 255 break; 256 case RDMA_CM_EVENT_ROUTE_ERROR: 257 ia->ri_async_rc = -ENETUNREACH; 258 dprintk("RPC: %s: CM route resolution error, ep 0x%p\n", 259 __func__, ep); 260 complete(&ia->ri_done); 261 break; 262 case RDMA_CM_EVENT_ESTABLISHED: 263 connstate = 1; 264 ib_query_qp(ia->ri_id->qp, attr, 265 IB_QP_MAX_QP_RD_ATOMIC | IB_QP_MAX_DEST_RD_ATOMIC, 266 iattr); 267 dprintk("RPC: %s: %d responder resources" 268 " (%d initiator)\n", 269 __func__, attr->max_dest_rd_atomic, 270 attr->max_rd_atomic); 271 rpcrdma_update_connect_private(xprt, &event->param.conn); 272 goto connected; 273 case RDMA_CM_EVENT_CONNECT_ERROR: 274 connstate = -ENOTCONN; 275 goto connected; 276 case RDMA_CM_EVENT_UNREACHABLE: 277 connstate = -ENETDOWN; 278 goto connected; 279 case RDMA_CM_EVENT_REJECTED: 280 connstate = -ECONNREFUSED; 281 goto connected; 282 case RDMA_CM_EVENT_DISCONNECTED: 283 connstate = -ECONNABORTED; 284 goto connected; 285 case RDMA_CM_EVENT_DEVICE_REMOVAL: 286 connstate = -ENODEV; 287 connected: 288 dprintk("RPC: %s: %sconnected\n", 289 __func__, connstate > 0 ? "" : "dis"); 290 atomic_set(&xprt->rx_buf.rb_credits, 1); 291 ep->rep_connected = connstate; 292 rpcrdma_conn_func(ep); 293 wake_up_all(&ep->rep_connect_wait); 294 /*FALLTHROUGH*/ 295 default: 296 dprintk("RPC: %s: %pIS:%u (ep 0x%p): %s\n", 297 __func__, sap, rpc_get_port(sap), ep, 298 rdma_event_msg(event->event)); 299 break; 300 } 301 302 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 303 if (connstate == 1) { 304 int ird = attr->max_dest_rd_atomic; 305 int tird = ep->rep_remote_cma.responder_resources; 306 307 pr_info("rpcrdma: connection to %pIS:%u on %s, memreg '%s', %d credits, %d responders%s\n", 308 sap, rpc_get_port(sap), 309 ia->ri_device->name, 310 ia->ri_ops->ro_displayname, 311 xprt->rx_buf.rb_max_requests, 312 ird, ird < 4 && ird < tird / 2 ? " (low!)" : ""); 313 } else if (connstate < 0) { 314 pr_info("rpcrdma: connection to %pIS:%u closed (%d)\n", 315 sap, rpc_get_port(sap), connstate); 316 } 317 #endif 318 319 return 0; 320 } 321 322 static void rpcrdma_destroy_id(struct rdma_cm_id *id) 323 { 324 if (id) { 325 module_put(id->device->owner); 326 rdma_destroy_id(id); 327 } 328 } 329 330 static struct rdma_cm_id * 331 rpcrdma_create_id(struct rpcrdma_xprt *xprt, 332 struct rpcrdma_ia *ia, struct sockaddr *addr) 333 { 334 unsigned long wtimeout = msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1; 335 struct rdma_cm_id *id; 336 int rc; 337 338 init_completion(&ia->ri_done); 339 340 id = rdma_create_id(&init_net, rpcrdma_conn_upcall, xprt, RDMA_PS_TCP, 341 IB_QPT_RC); 342 if (IS_ERR(id)) { 343 rc = PTR_ERR(id); 344 dprintk("RPC: %s: rdma_create_id() failed %i\n", 345 __func__, rc); 346 return id; 347 } 348 349 ia->ri_async_rc = -ETIMEDOUT; 350 rc = rdma_resolve_addr(id, NULL, addr, RDMA_RESOLVE_TIMEOUT); 351 if (rc) { 352 dprintk("RPC: %s: rdma_resolve_addr() failed %i\n", 353 __func__, rc); 354 goto out; 355 } 356 rc = wait_for_completion_interruptible_timeout(&ia->ri_done, wtimeout); 357 if (rc < 0) { 358 dprintk("RPC: %s: wait() exited: %i\n", 359 __func__, rc); 360 goto out; 361 } 362 363 /* FIXME: 364 * Until xprtrdma supports DEVICE_REMOVAL, the provider must 365 * be pinned while there are active NFS/RDMA mounts to prevent 366 * hangs and crashes at umount time. 367 */ 368 if (!ia->ri_async_rc && !try_module_get(id->device->owner)) { 369 dprintk("RPC: %s: Failed to get device module\n", 370 __func__); 371 ia->ri_async_rc = -ENODEV; 372 } 373 rc = ia->ri_async_rc; 374 if (rc) 375 goto out; 376 377 ia->ri_async_rc = -ETIMEDOUT; 378 rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT); 379 if (rc) { 380 dprintk("RPC: %s: rdma_resolve_route() failed %i\n", 381 __func__, rc); 382 goto put; 383 } 384 rc = wait_for_completion_interruptible_timeout(&ia->ri_done, wtimeout); 385 if (rc < 0) { 386 dprintk("RPC: %s: wait() exited: %i\n", 387 __func__, rc); 388 goto put; 389 } 390 rc = ia->ri_async_rc; 391 if (rc) 392 goto put; 393 394 return id; 395 put: 396 module_put(id->device->owner); 397 out: 398 rdma_destroy_id(id); 399 return ERR_PTR(rc); 400 } 401 402 /* 403 * Exported functions. 404 */ 405 406 /* 407 * Open and initialize an Interface Adapter. 408 * o initializes fields of struct rpcrdma_ia, including 409 * interface and provider attributes and protection zone. 410 */ 411 int 412 rpcrdma_ia_open(struct rpcrdma_xprt *xprt, struct sockaddr *addr, int memreg) 413 { 414 struct rpcrdma_ia *ia = &xprt->rx_ia; 415 int rc; 416 417 ia->ri_id = rpcrdma_create_id(xprt, ia, addr); 418 if (IS_ERR(ia->ri_id)) { 419 rc = PTR_ERR(ia->ri_id); 420 goto out1; 421 } 422 ia->ri_device = ia->ri_id->device; 423 424 ia->ri_pd = ib_alloc_pd(ia->ri_device, 0); 425 if (IS_ERR(ia->ri_pd)) { 426 rc = PTR_ERR(ia->ri_pd); 427 pr_err("rpcrdma: ib_alloc_pd() returned %d\n", rc); 428 goto out2; 429 } 430 431 switch (memreg) { 432 case RPCRDMA_FRMR: 433 if (frwr_is_supported(ia)) { 434 ia->ri_ops = &rpcrdma_frwr_memreg_ops; 435 break; 436 } 437 /*FALLTHROUGH*/ 438 case RPCRDMA_MTHCAFMR: 439 if (fmr_is_supported(ia)) { 440 ia->ri_ops = &rpcrdma_fmr_memreg_ops; 441 break; 442 } 443 /*FALLTHROUGH*/ 444 default: 445 pr_err("rpcrdma: Unsupported memory registration mode: %d\n", 446 memreg); 447 rc = -EINVAL; 448 goto out3; 449 } 450 451 return 0; 452 453 out3: 454 ib_dealloc_pd(ia->ri_pd); 455 ia->ri_pd = NULL; 456 out2: 457 rpcrdma_destroy_id(ia->ri_id); 458 ia->ri_id = NULL; 459 out1: 460 return rc; 461 } 462 463 /* 464 * Clean up/close an IA. 465 * o if event handles and PD have been initialized, free them. 466 * o close the IA 467 */ 468 void 469 rpcrdma_ia_close(struct rpcrdma_ia *ia) 470 { 471 dprintk("RPC: %s: entering\n", __func__); 472 if (ia->ri_id != NULL && !IS_ERR(ia->ri_id)) { 473 if (ia->ri_id->qp) 474 rdma_destroy_qp(ia->ri_id); 475 rpcrdma_destroy_id(ia->ri_id); 476 ia->ri_id = NULL; 477 } 478 479 /* If the pd is still busy, xprtrdma missed freeing a resource */ 480 if (ia->ri_pd && !IS_ERR(ia->ri_pd)) 481 ib_dealloc_pd(ia->ri_pd); 482 } 483 484 /* 485 * Create unconnected endpoint. 486 */ 487 int 488 rpcrdma_ep_create(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia, 489 struct rpcrdma_create_data_internal *cdata) 490 { 491 struct rpcrdma_connect_private *pmsg = &ep->rep_cm_private; 492 struct ib_cq *sendcq, *recvcq; 493 unsigned int max_qp_wr; 494 int rc; 495 496 if (ia->ri_device->attrs.max_sge < RPCRDMA_MAX_SEND_SGES) { 497 dprintk("RPC: %s: insufficient sge's available\n", 498 __func__); 499 return -ENOMEM; 500 } 501 502 if (ia->ri_device->attrs.max_qp_wr <= RPCRDMA_BACKWARD_WRS) { 503 dprintk("RPC: %s: insufficient wqe's available\n", 504 __func__); 505 return -ENOMEM; 506 } 507 max_qp_wr = ia->ri_device->attrs.max_qp_wr - RPCRDMA_BACKWARD_WRS - 1; 508 509 /* check provider's send/recv wr limits */ 510 if (cdata->max_requests > max_qp_wr) 511 cdata->max_requests = max_qp_wr; 512 513 ep->rep_attr.event_handler = rpcrdma_qp_async_error_upcall; 514 ep->rep_attr.qp_context = ep; 515 ep->rep_attr.srq = NULL; 516 ep->rep_attr.cap.max_send_wr = cdata->max_requests; 517 ep->rep_attr.cap.max_send_wr += RPCRDMA_BACKWARD_WRS; 518 ep->rep_attr.cap.max_send_wr += 1; /* drain cqe */ 519 rc = ia->ri_ops->ro_open(ia, ep, cdata); 520 if (rc) 521 return rc; 522 ep->rep_attr.cap.max_recv_wr = cdata->max_requests; 523 ep->rep_attr.cap.max_recv_wr += RPCRDMA_BACKWARD_WRS; 524 ep->rep_attr.cap.max_recv_wr += 1; /* drain cqe */ 525 ep->rep_attr.cap.max_send_sge = RPCRDMA_MAX_SEND_SGES; 526 ep->rep_attr.cap.max_recv_sge = 1; 527 ep->rep_attr.cap.max_inline_data = 0; 528 ep->rep_attr.sq_sig_type = IB_SIGNAL_REQ_WR; 529 ep->rep_attr.qp_type = IB_QPT_RC; 530 ep->rep_attr.port_num = ~0; 531 532 dprintk("RPC: %s: requested max: dtos: send %d recv %d; " 533 "iovs: send %d recv %d\n", 534 __func__, 535 ep->rep_attr.cap.max_send_wr, 536 ep->rep_attr.cap.max_recv_wr, 537 ep->rep_attr.cap.max_send_sge, 538 ep->rep_attr.cap.max_recv_sge); 539 540 /* set trigger for requesting send completion */ 541 ep->rep_cqinit = ep->rep_attr.cap.max_send_wr/2 - 1; 542 if (ep->rep_cqinit <= 2) 543 ep->rep_cqinit = 0; /* always signal? */ 544 rpcrdma_init_cqcount(ep, 0); 545 init_waitqueue_head(&ep->rep_connect_wait); 546 INIT_DELAYED_WORK(&ep->rep_connect_worker, rpcrdma_connect_worker); 547 548 sendcq = ib_alloc_cq(ia->ri_device, NULL, 549 ep->rep_attr.cap.max_send_wr + 1, 550 0, IB_POLL_SOFTIRQ); 551 if (IS_ERR(sendcq)) { 552 rc = PTR_ERR(sendcq); 553 dprintk("RPC: %s: failed to create send CQ: %i\n", 554 __func__, rc); 555 goto out1; 556 } 557 558 recvcq = ib_alloc_cq(ia->ri_device, NULL, 559 ep->rep_attr.cap.max_recv_wr + 1, 560 0, IB_POLL_SOFTIRQ); 561 if (IS_ERR(recvcq)) { 562 rc = PTR_ERR(recvcq); 563 dprintk("RPC: %s: failed to create recv CQ: %i\n", 564 __func__, rc); 565 goto out2; 566 } 567 568 ep->rep_attr.send_cq = sendcq; 569 ep->rep_attr.recv_cq = recvcq; 570 571 /* Initialize cma parameters */ 572 memset(&ep->rep_remote_cma, 0, sizeof(ep->rep_remote_cma)); 573 574 /* Prepare RDMA-CM private message */ 575 pmsg->cp_magic = rpcrdma_cmp_magic; 576 pmsg->cp_version = RPCRDMA_CMP_VERSION; 577 pmsg->cp_flags |= ia->ri_ops->ro_send_w_inv_ok; 578 pmsg->cp_send_size = rpcrdma_encode_buffer_size(cdata->inline_wsize); 579 pmsg->cp_recv_size = rpcrdma_encode_buffer_size(cdata->inline_rsize); 580 ep->rep_remote_cma.private_data = pmsg; 581 ep->rep_remote_cma.private_data_len = sizeof(*pmsg); 582 583 /* Client offers RDMA Read but does not initiate */ 584 ep->rep_remote_cma.initiator_depth = 0; 585 if (ia->ri_device->attrs.max_qp_rd_atom > 32) /* arbitrary but <= 255 */ 586 ep->rep_remote_cma.responder_resources = 32; 587 else 588 ep->rep_remote_cma.responder_resources = 589 ia->ri_device->attrs.max_qp_rd_atom; 590 591 /* Limit transport retries so client can detect server 592 * GID changes quickly. RPC layer handles re-establishing 593 * transport connection and retransmission. 594 */ 595 ep->rep_remote_cma.retry_count = 6; 596 597 /* RPC-over-RDMA handles its own flow control. In addition, 598 * make all RNR NAKs visible so we know that RPC-over-RDMA 599 * flow control is working correctly (no NAKs should be seen). 600 */ 601 ep->rep_remote_cma.flow_control = 0; 602 ep->rep_remote_cma.rnr_retry_count = 0; 603 604 return 0; 605 606 out2: 607 ib_free_cq(sendcq); 608 out1: 609 return rc; 610 } 611 612 /* 613 * rpcrdma_ep_destroy 614 * 615 * Disconnect and destroy endpoint. After this, the only 616 * valid operations on the ep are to free it (if dynamically 617 * allocated) or re-create it. 618 */ 619 void 620 rpcrdma_ep_destroy(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 621 { 622 dprintk("RPC: %s: entering, connected is %d\n", 623 __func__, ep->rep_connected); 624 625 cancel_delayed_work_sync(&ep->rep_connect_worker); 626 627 if (ia->ri_id->qp) { 628 rpcrdma_ep_disconnect(ep, ia); 629 rdma_destroy_qp(ia->ri_id); 630 ia->ri_id->qp = NULL; 631 } 632 633 ib_free_cq(ep->rep_attr.recv_cq); 634 ib_free_cq(ep->rep_attr.send_cq); 635 } 636 637 /* 638 * Connect unconnected endpoint. 639 */ 640 int 641 rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 642 { 643 struct rdma_cm_id *id, *old; 644 int rc = 0; 645 int retry_count = 0; 646 647 if (ep->rep_connected != 0) { 648 struct rpcrdma_xprt *xprt; 649 retry: 650 dprintk("RPC: %s: reconnecting...\n", __func__); 651 652 rpcrdma_ep_disconnect(ep, ia); 653 654 xprt = container_of(ia, struct rpcrdma_xprt, rx_ia); 655 id = rpcrdma_create_id(xprt, ia, 656 (struct sockaddr *)&xprt->rx_data.addr); 657 if (IS_ERR(id)) { 658 rc = -EHOSTUNREACH; 659 goto out; 660 } 661 /* TEMP TEMP TEMP - fail if new device: 662 * Deregister/remarshal *all* requests! 663 * Close and recreate adapter, pd, etc! 664 * Re-determine all attributes still sane! 665 * More stuff I haven't thought of! 666 * Rrrgh! 667 */ 668 if (ia->ri_device != id->device) { 669 printk("RPC: %s: can't reconnect on " 670 "different device!\n", __func__); 671 rpcrdma_destroy_id(id); 672 rc = -ENETUNREACH; 673 goto out; 674 } 675 /* END TEMP */ 676 rc = rdma_create_qp(id, ia->ri_pd, &ep->rep_attr); 677 if (rc) { 678 dprintk("RPC: %s: rdma_create_qp failed %i\n", 679 __func__, rc); 680 rpcrdma_destroy_id(id); 681 rc = -ENETUNREACH; 682 goto out; 683 } 684 685 old = ia->ri_id; 686 ia->ri_id = id; 687 688 rdma_destroy_qp(old); 689 rpcrdma_destroy_id(old); 690 } else { 691 dprintk("RPC: %s: connecting...\n", __func__); 692 rc = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr); 693 if (rc) { 694 dprintk("RPC: %s: rdma_create_qp failed %i\n", 695 __func__, rc); 696 /* do not update ep->rep_connected */ 697 return -ENETUNREACH; 698 } 699 } 700 701 ep->rep_connected = 0; 702 703 rc = rdma_connect(ia->ri_id, &ep->rep_remote_cma); 704 if (rc) { 705 dprintk("RPC: %s: rdma_connect() failed with %i\n", 706 __func__, rc); 707 goto out; 708 } 709 710 wait_event_interruptible(ep->rep_connect_wait, ep->rep_connected != 0); 711 712 /* 713 * Check state. A non-peer reject indicates no listener 714 * (ECONNREFUSED), which may be a transient state. All 715 * others indicate a transport condition which has already 716 * undergone a best-effort. 717 */ 718 if (ep->rep_connected == -ECONNREFUSED && 719 ++retry_count <= RDMA_CONNECT_RETRY_MAX) { 720 dprintk("RPC: %s: non-peer_reject, retry\n", __func__); 721 goto retry; 722 } 723 if (ep->rep_connected <= 0) { 724 /* Sometimes, the only way to reliably connect to remote 725 * CMs is to use same nonzero values for ORD and IRD. */ 726 if (retry_count++ <= RDMA_CONNECT_RETRY_MAX + 1 && 727 (ep->rep_remote_cma.responder_resources == 0 || 728 ep->rep_remote_cma.initiator_depth != 729 ep->rep_remote_cma.responder_resources)) { 730 if (ep->rep_remote_cma.responder_resources == 0) 731 ep->rep_remote_cma.responder_resources = 1; 732 ep->rep_remote_cma.initiator_depth = 733 ep->rep_remote_cma.responder_resources; 734 goto retry; 735 } 736 rc = ep->rep_connected; 737 } else { 738 struct rpcrdma_xprt *r_xprt; 739 unsigned int extras; 740 741 dprintk("RPC: %s: connected\n", __func__); 742 743 r_xprt = container_of(ia, struct rpcrdma_xprt, rx_ia); 744 extras = r_xprt->rx_buf.rb_bc_srv_max_requests; 745 746 if (extras) { 747 rc = rpcrdma_ep_post_extra_recv(r_xprt, extras); 748 if (rc) { 749 pr_warn("%s: rpcrdma_ep_post_extra_recv: %i\n", 750 __func__, rc); 751 rc = 0; 752 } 753 } 754 } 755 756 out: 757 if (rc) 758 ep->rep_connected = rc; 759 return rc; 760 } 761 762 /* 763 * rpcrdma_ep_disconnect 764 * 765 * This is separate from destroy to facilitate the ability 766 * to reconnect without recreating the endpoint. 767 * 768 * This call is not reentrant, and must not be made in parallel 769 * on the same endpoint. 770 */ 771 void 772 rpcrdma_ep_disconnect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 773 { 774 int rc; 775 776 rc = rdma_disconnect(ia->ri_id); 777 if (!rc) { 778 /* returns without wait if not connected */ 779 wait_event_interruptible(ep->rep_connect_wait, 780 ep->rep_connected != 1); 781 dprintk("RPC: %s: after wait, %sconnected\n", __func__, 782 (ep->rep_connected == 1) ? "still " : "dis"); 783 } else { 784 dprintk("RPC: %s: rdma_disconnect %i\n", __func__, rc); 785 ep->rep_connected = rc; 786 } 787 788 ib_drain_qp(ia->ri_id->qp); 789 } 790 791 static void 792 rpcrdma_mr_recovery_worker(struct work_struct *work) 793 { 794 struct rpcrdma_buffer *buf = container_of(work, struct rpcrdma_buffer, 795 rb_recovery_worker.work); 796 struct rpcrdma_mw *mw; 797 798 spin_lock(&buf->rb_recovery_lock); 799 while (!list_empty(&buf->rb_stale_mrs)) { 800 mw = list_first_entry(&buf->rb_stale_mrs, 801 struct rpcrdma_mw, mw_list); 802 list_del_init(&mw->mw_list); 803 spin_unlock(&buf->rb_recovery_lock); 804 805 dprintk("RPC: %s: recovering MR %p\n", __func__, mw); 806 mw->mw_xprt->rx_ia.ri_ops->ro_recover_mr(mw); 807 808 spin_lock(&buf->rb_recovery_lock); 809 } 810 spin_unlock(&buf->rb_recovery_lock); 811 } 812 813 void 814 rpcrdma_defer_mr_recovery(struct rpcrdma_mw *mw) 815 { 816 struct rpcrdma_xprt *r_xprt = mw->mw_xprt; 817 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 818 819 spin_lock(&buf->rb_recovery_lock); 820 list_add(&mw->mw_list, &buf->rb_stale_mrs); 821 spin_unlock(&buf->rb_recovery_lock); 822 823 schedule_delayed_work(&buf->rb_recovery_worker, 0); 824 } 825 826 static void 827 rpcrdma_create_mrs(struct rpcrdma_xprt *r_xprt) 828 { 829 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 830 struct rpcrdma_ia *ia = &r_xprt->rx_ia; 831 unsigned int count; 832 LIST_HEAD(free); 833 LIST_HEAD(all); 834 835 for (count = 0; count < 32; count++) { 836 struct rpcrdma_mw *mw; 837 int rc; 838 839 mw = kzalloc(sizeof(*mw), GFP_KERNEL); 840 if (!mw) 841 break; 842 843 rc = ia->ri_ops->ro_init_mr(ia, mw); 844 if (rc) { 845 kfree(mw); 846 break; 847 } 848 849 mw->mw_xprt = r_xprt; 850 851 list_add(&mw->mw_list, &free); 852 list_add(&mw->mw_all, &all); 853 } 854 855 spin_lock(&buf->rb_mwlock); 856 list_splice(&free, &buf->rb_mws); 857 list_splice(&all, &buf->rb_all); 858 r_xprt->rx_stats.mrs_allocated += count; 859 spin_unlock(&buf->rb_mwlock); 860 861 dprintk("RPC: %s: created %u MRs\n", __func__, count); 862 } 863 864 static void 865 rpcrdma_mr_refresh_worker(struct work_struct *work) 866 { 867 struct rpcrdma_buffer *buf = container_of(work, struct rpcrdma_buffer, 868 rb_refresh_worker.work); 869 struct rpcrdma_xprt *r_xprt = container_of(buf, struct rpcrdma_xprt, 870 rx_buf); 871 872 rpcrdma_create_mrs(r_xprt); 873 } 874 875 struct rpcrdma_req * 876 rpcrdma_create_req(struct rpcrdma_xprt *r_xprt) 877 { 878 struct rpcrdma_buffer *buffer = &r_xprt->rx_buf; 879 struct rpcrdma_req *req; 880 881 req = kzalloc(sizeof(*req), GFP_KERNEL); 882 if (req == NULL) 883 return ERR_PTR(-ENOMEM); 884 885 INIT_LIST_HEAD(&req->rl_free); 886 spin_lock(&buffer->rb_reqslock); 887 list_add(&req->rl_all, &buffer->rb_allreqs); 888 spin_unlock(&buffer->rb_reqslock); 889 req->rl_cqe.done = rpcrdma_wc_send; 890 req->rl_buffer = &r_xprt->rx_buf; 891 INIT_LIST_HEAD(&req->rl_registered); 892 req->rl_send_wr.next = NULL; 893 req->rl_send_wr.wr_cqe = &req->rl_cqe; 894 req->rl_send_wr.sg_list = req->rl_send_sge; 895 req->rl_send_wr.opcode = IB_WR_SEND; 896 return req; 897 } 898 899 struct rpcrdma_rep * 900 rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt) 901 { 902 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 903 struct rpcrdma_ia *ia = &r_xprt->rx_ia; 904 struct rpcrdma_rep *rep; 905 int rc; 906 907 rc = -ENOMEM; 908 rep = kzalloc(sizeof(*rep), GFP_KERNEL); 909 if (rep == NULL) 910 goto out; 911 912 rep->rr_rdmabuf = rpcrdma_alloc_regbuf(cdata->inline_rsize, 913 DMA_FROM_DEVICE, GFP_KERNEL); 914 if (IS_ERR(rep->rr_rdmabuf)) { 915 rc = PTR_ERR(rep->rr_rdmabuf); 916 goto out_free; 917 } 918 919 rep->rr_device = ia->ri_device; 920 rep->rr_cqe.done = rpcrdma_wc_receive; 921 rep->rr_rxprt = r_xprt; 922 INIT_WORK(&rep->rr_work, rpcrdma_reply_handler); 923 rep->rr_recv_wr.next = NULL; 924 rep->rr_recv_wr.wr_cqe = &rep->rr_cqe; 925 rep->rr_recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov; 926 rep->rr_recv_wr.num_sge = 1; 927 return rep; 928 929 out_free: 930 kfree(rep); 931 out: 932 return ERR_PTR(rc); 933 } 934 935 int 936 rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt) 937 { 938 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 939 int i, rc; 940 941 buf->rb_max_requests = r_xprt->rx_data.max_requests; 942 buf->rb_bc_srv_max_requests = 0; 943 atomic_set(&buf->rb_credits, 1); 944 spin_lock_init(&buf->rb_mwlock); 945 spin_lock_init(&buf->rb_lock); 946 spin_lock_init(&buf->rb_recovery_lock); 947 INIT_LIST_HEAD(&buf->rb_mws); 948 INIT_LIST_HEAD(&buf->rb_all); 949 INIT_LIST_HEAD(&buf->rb_stale_mrs); 950 INIT_DELAYED_WORK(&buf->rb_refresh_worker, 951 rpcrdma_mr_refresh_worker); 952 INIT_DELAYED_WORK(&buf->rb_recovery_worker, 953 rpcrdma_mr_recovery_worker); 954 955 rpcrdma_create_mrs(r_xprt); 956 957 INIT_LIST_HEAD(&buf->rb_send_bufs); 958 INIT_LIST_HEAD(&buf->rb_allreqs); 959 spin_lock_init(&buf->rb_reqslock); 960 for (i = 0; i < buf->rb_max_requests; i++) { 961 struct rpcrdma_req *req; 962 963 req = rpcrdma_create_req(r_xprt); 964 if (IS_ERR(req)) { 965 dprintk("RPC: %s: request buffer %d alloc" 966 " failed\n", __func__, i); 967 rc = PTR_ERR(req); 968 goto out; 969 } 970 req->rl_backchannel = false; 971 list_add(&req->rl_free, &buf->rb_send_bufs); 972 } 973 974 INIT_LIST_HEAD(&buf->rb_recv_bufs); 975 for (i = 0; i < buf->rb_max_requests + RPCRDMA_MAX_BC_REQUESTS; i++) { 976 struct rpcrdma_rep *rep; 977 978 rep = rpcrdma_create_rep(r_xprt); 979 if (IS_ERR(rep)) { 980 dprintk("RPC: %s: reply buffer %d alloc failed\n", 981 __func__, i); 982 rc = PTR_ERR(rep); 983 goto out; 984 } 985 list_add(&rep->rr_list, &buf->rb_recv_bufs); 986 } 987 988 return 0; 989 out: 990 rpcrdma_buffer_destroy(buf); 991 return rc; 992 } 993 994 static struct rpcrdma_req * 995 rpcrdma_buffer_get_req_locked(struct rpcrdma_buffer *buf) 996 { 997 struct rpcrdma_req *req; 998 999 req = list_first_entry(&buf->rb_send_bufs, 1000 struct rpcrdma_req, rl_free); 1001 list_del(&req->rl_free); 1002 return req; 1003 } 1004 1005 static struct rpcrdma_rep * 1006 rpcrdma_buffer_get_rep_locked(struct rpcrdma_buffer *buf) 1007 { 1008 struct rpcrdma_rep *rep; 1009 1010 rep = list_first_entry(&buf->rb_recv_bufs, 1011 struct rpcrdma_rep, rr_list); 1012 list_del(&rep->rr_list); 1013 return rep; 1014 } 1015 1016 static void 1017 rpcrdma_destroy_rep(struct rpcrdma_rep *rep) 1018 { 1019 rpcrdma_free_regbuf(rep->rr_rdmabuf); 1020 kfree(rep); 1021 } 1022 1023 void 1024 rpcrdma_destroy_req(struct rpcrdma_req *req) 1025 { 1026 rpcrdma_free_regbuf(req->rl_recvbuf); 1027 rpcrdma_free_regbuf(req->rl_sendbuf); 1028 rpcrdma_free_regbuf(req->rl_rdmabuf); 1029 kfree(req); 1030 } 1031 1032 static void 1033 rpcrdma_destroy_mrs(struct rpcrdma_buffer *buf) 1034 { 1035 struct rpcrdma_xprt *r_xprt = container_of(buf, struct rpcrdma_xprt, 1036 rx_buf); 1037 struct rpcrdma_ia *ia = rdmab_to_ia(buf); 1038 struct rpcrdma_mw *mw; 1039 unsigned int count; 1040 1041 count = 0; 1042 spin_lock(&buf->rb_mwlock); 1043 while (!list_empty(&buf->rb_all)) { 1044 mw = list_entry(buf->rb_all.next, struct rpcrdma_mw, mw_all); 1045 list_del(&mw->mw_all); 1046 1047 spin_unlock(&buf->rb_mwlock); 1048 ia->ri_ops->ro_release_mr(mw); 1049 count++; 1050 spin_lock(&buf->rb_mwlock); 1051 } 1052 spin_unlock(&buf->rb_mwlock); 1053 r_xprt->rx_stats.mrs_allocated = 0; 1054 1055 dprintk("RPC: %s: released %u MRs\n", __func__, count); 1056 } 1057 1058 void 1059 rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf) 1060 { 1061 cancel_delayed_work_sync(&buf->rb_recovery_worker); 1062 1063 while (!list_empty(&buf->rb_recv_bufs)) { 1064 struct rpcrdma_rep *rep; 1065 1066 rep = rpcrdma_buffer_get_rep_locked(buf); 1067 rpcrdma_destroy_rep(rep); 1068 } 1069 buf->rb_send_count = 0; 1070 1071 spin_lock(&buf->rb_reqslock); 1072 while (!list_empty(&buf->rb_allreqs)) { 1073 struct rpcrdma_req *req; 1074 1075 req = list_first_entry(&buf->rb_allreqs, 1076 struct rpcrdma_req, rl_all); 1077 list_del(&req->rl_all); 1078 1079 spin_unlock(&buf->rb_reqslock); 1080 rpcrdma_destroy_req(req); 1081 spin_lock(&buf->rb_reqslock); 1082 } 1083 spin_unlock(&buf->rb_reqslock); 1084 buf->rb_recv_count = 0; 1085 1086 rpcrdma_destroy_mrs(buf); 1087 } 1088 1089 struct rpcrdma_mw * 1090 rpcrdma_get_mw(struct rpcrdma_xprt *r_xprt) 1091 { 1092 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1093 struct rpcrdma_mw *mw = NULL; 1094 1095 spin_lock(&buf->rb_mwlock); 1096 if (!list_empty(&buf->rb_mws)) { 1097 mw = list_first_entry(&buf->rb_mws, 1098 struct rpcrdma_mw, mw_list); 1099 list_del_init(&mw->mw_list); 1100 } 1101 spin_unlock(&buf->rb_mwlock); 1102 1103 if (!mw) 1104 goto out_nomws; 1105 return mw; 1106 1107 out_nomws: 1108 dprintk("RPC: %s: no MWs available\n", __func__); 1109 schedule_delayed_work(&buf->rb_refresh_worker, 0); 1110 1111 /* Allow the reply handler and refresh worker to run */ 1112 cond_resched(); 1113 1114 return NULL; 1115 } 1116 1117 void 1118 rpcrdma_put_mw(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mw *mw) 1119 { 1120 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1121 1122 spin_lock(&buf->rb_mwlock); 1123 list_add_tail(&mw->mw_list, &buf->rb_mws); 1124 spin_unlock(&buf->rb_mwlock); 1125 } 1126 1127 static struct rpcrdma_rep * 1128 rpcrdma_buffer_get_rep(struct rpcrdma_buffer *buffers) 1129 { 1130 /* If an RPC previously completed without a reply (say, a 1131 * credential problem or a soft timeout occurs) then hold off 1132 * on supplying more Receive buffers until the number of new 1133 * pending RPCs catches up to the number of posted Receives. 1134 */ 1135 if (unlikely(buffers->rb_send_count < buffers->rb_recv_count)) 1136 return NULL; 1137 1138 if (unlikely(list_empty(&buffers->rb_recv_bufs))) 1139 return NULL; 1140 buffers->rb_recv_count++; 1141 return rpcrdma_buffer_get_rep_locked(buffers); 1142 } 1143 1144 /* 1145 * Get a set of request/reply buffers. 1146 * 1147 * Reply buffer (if available) is attached to send buffer upon return. 1148 */ 1149 struct rpcrdma_req * 1150 rpcrdma_buffer_get(struct rpcrdma_buffer *buffers) 1151 { 1152 struct rpcrdma_req *req; 1153 1154 spin_lock(&buffers->rb_lock); 1155 if (list_empty(&buffers->rb_send_bufs)) 1156 goto out_reqbuf; 1157 buffers->rb_send_count++; 1158 req = rpcrdma_buffer_get_req_locked(buffers); 1159 req->rl_reply = rpcrdma_buffer_get_rep(buffers); 1160 spin_unlock(&buffers->rb_lock); 1161 return req; 1162 1163 out_reqbuf: 1164 spin_unlock(&buffers->rb_lock); 1165 pr_warn("RPC: %s: out of request buffers\n", __func__); 1166 return NULL; 1167 } 1168 1169 /* 1170 * Put request/reply buffers back into pool. 1171 * Pre-decrement counter/array index. 1172 */ 1173 void 1174 rpcrdma_buffer_put(struct rpcrdma_req *req) 1175 { 1176 struct rpcrdma_buffer *buffers = req->rl_buffer; 1177 struct rpcrdma_rep *rep = req->rl_reply; 1178 1179 req->rl_send_wr.num_sge = 0; 1180 req->rl_reply = NULL; 1181 1182 spin_lock(&buffers->rb_lock); 1183 buffers->rb_send_count--; 1184 list_add_tail(&req->rl_free, &buffers->rb_send_bufs); 1185 if (rep) { 1186 buffers->rb_recv_count--; 1187 list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs); 1188 } 1189 spin_unlock(&buffers->rb_lock); 1190 } 1191 1192 /* 1193 * Recover reply buffers from pool. 1194 * This happens when recovering from disconnect. 1195 */ 1196 void 1197 rpcrdma_recv_buffer_get(struct rpcrdma_req *req) 1198 { 1199 struct rpcrdma_buffer *buffers = req->rl_buffer; 1200 1201 spin_lock(&buffers->rb_lock); 1202 req->rl_reply = rpcrdma_buffer_get_rep(buffers); 1203 spin_unlock(&buffers->rb_lock); 1204 } 1205 1206 /* 1207 * Put reply buffers back into pool when not attached to 1208 * request. This happens in error conditions. 1209 */ 1210 void 1211 rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep) 1212 { 1213 struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf; 1214 1215 spin_lock(&buffers->rb_lock); 1216 buffers->rb_recv_count--; 1217 list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs); 1218 spin_unlock(&buffers->rb_lock); 1219 } 1220 1221 /** 1222 * rpcrdma_alloc_regbuf - allocate and DMA-map memory for SEND/RECV buffers 1223 * @size: size of buffer to be allocated, in bytes 1224 * @direction: direction of data movement 1225 * @flags: GFP flags 1226 * 1227 * Returns an ERR_PTR, or a pointer to a regbuf, a buffer that 1228 * can be persistently DMA-mapped for I/O. 1229 * 1230 * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for 1231 * receiving the payload of RDMA RECV operations. During Long Calls 1232 * or Replies they may be registered externally via ro_map. 1233 */ 1234 struct rpcrdma_regbuf * 1235 rpcrdma_alloc_regbuf(size_t size, enum dma_data_direction direction, 1236 gfp_t flags) 1237 { 1238 struct rpcrdma_regbuf *rb; 1239 1240 rb = kmalloc(sizeof(*rb) + size, flags); 1241 if (rb == NULL) 1242 return ERR_PTR(-ENOMEM); 1243 1244 rb->rg_device = NULL; 1245 rb->rg_direction = direction; 1246 rb->rg_iov.length = size; 1247 1248 return rb; 1249 } 1250 1251 /** 1252 * __rpcrdma_map_regbuf - DMA-map a regbuf 1253 * @ia: controlling rpcrdma_ia 1254 * @rb: regbuf to be mapped 1255 */ 1256 bool 1257 __rpcrdma_dma_map_regbuf(struct rpcrdma_ia *ia, struct rpcrdma_regbuf *rb) 1258 { 1259 if (rb->rg_direction == DMA_NONE) 1260 return false; 1261 1262 rb->rg_iov.addr = ib_dma_map_single(ia->ri_device, 1263 (void *)rb->rg_base, 1264 rdmab_length(rb), 1265 rb->rg_direction); 1266 if (ib_dma_mapping_error(ia->ri_device, rdmab_addr(rb))) 1267 return false; 1268 1269 rb->rg_device = ia->ri_device; 1270 rb->rg_iov.lkey = ia->ri_pd->local_dma_lkey; 1271 return true; 1272 } 1273 1274 static void 1275 rpcrdma_dma_unmap_regbuf(struct rpcrdma_regbuf *rb) 1276 { 1277 if (!rpcrdma_regbuf_is_mapped(rb)) 1278 return; 1279 1280 ib_dma_unmap_single(rb->rg_device, rdmab_addr(rb), 1281 rdmab_length(rb), rb->rg_direction); 1282 rb->rg_device = NULL; 1283 } 1284 1285 /** 1286 * rpcrdma_free_regbuf - deregister and free registered buffer 1287 * @rb: regbuf to be deregistered and freed 1288 */ 1289 void 1290 rpcrdma_free_regbuf(struct rpcrdma_regbuf *rb) 1291 { 1292 if (!rb) 1293 return; 1294 1295 rpcrdma_dma_unmap_regbuf(rb); 1296 kfree(rb); 1297 } 1298 1299 /* 1300 * Prepost any receive buffer, then post send. 1301 * 1302 * Receive buffer is donated to hardware, reclaimed upon recv completion. 1303 */ 1304 int 1305 rpcrdma_ep_post(struct rpcrdma_ia *ia, 1306 struct rpcrdma_ep *ep, 1307 struct rpcrdma_req *req) 1308 { 1309 struct ib_send_wr *send_wr = &req->rl_send_wr; 1310 struct ib_send_wr *send_wr_fail; 1311 int rc; 1312 1313 if (req->rl_reply) { 1314 rc = rpcrdma_ep_post_recv(ia, req->rl_reply); 1315 if (rc) 1316 return rc; 1317 req->rl_reply = NULL; 1318 } 1319 1320 dprintk("RPC: %s: posting %d s/g entries\n", 1321 __func__, send_wr->num_sge); 1322 1323 rpcrdma_set_signaled(ep, send_wr); 1324 rc = ib_post_send(ia->ri_id->qp, send_wr, &send_wr_fail); 1325 if (rc) 1326 goto out_postsend_err; 1327 return 0; 1328 1329 out_postsend_err: 1330 pr_err("rpcrdma: RDMA Send ib_post_send returned %i\n", rc); 1331 return -ENOTCONN; 1332 } 1333 1334 int 1335 rpcrdma_ep_post_recv(struct rpcrdma_ia *ia, 1336 struct rpcrdma_rep *rep) 1337 { 1338 struct ib_recv_wr *recv_wr_fail; 1339 int rc; 1340 1341 if (!rpcrdma_dma_map_regbuf(ia, rep->rr_rdmabuf)) 1342 goto out_map; 1343 rc = ib_post_recv(ia->ri_id->qp, &rep->rr_recv_wr, &recv_wr_fail); 1344 if (rc) 1345 goto out_postrecv; 1346 return 0; 1347 1348 out_map: 1349 pr_err("rpcrdma: failed to DMA map the Receive buffer\n"); 1350 return -EIO; 1351 1352 out_postrecv: 1353 pr_err("rpcrdma: ib_post_recv returned %i\n", rc); 1354 return -ENOTCONN; 1355 } 1356 1357 /** 1358 * rpcrdma_ep_post_extra_recv - Post buffers for incoming backchannel requests 1359 * @r_xprt: transport associated with these backchannel resources 1360 * @min_reqs: minimum number of incoming requests expected 1361 * 1362 * Returns zero if all requested buffers were posted, or a negative errno. 1363 */ 1364 int 1365 rpcrdma_ep_post_extra_recv(struct rpcrdma_xprt *r_xprt, unsigned int count) 1366 { 1367 struct rpcrdma_buffer *buffers = &r_xprt->rx_buf; 1368 struct rpcrdma_ia *ia = &r_xprt->rx_ia; 1369 struct rpcrdma_rep *rep; 1370 int rc; 1371 1372 while (count--) { 1373 spin_lock(&buffers->rb_lock); 1374 if (list_empty(&buffers->rb_recv_bufs)) 1375 goto out_reqbuf; 1376 rep = rpcrdma_buffer_get_rep_locked(buffers); 1377 spin_unlock(&buffers->rb_lock); 1378 1379 rc = rpcrdma_ep_post_recv(ia, rep); 1380 if (rc) 1381 goto out_rc; 1382 } 1383 1384 return 0; 1385 1386 out_reqbuf: 1387 spin_unlock(&buffers->rb_lock); 1388 pr_warn("%s: no extra receive buffers\n", __func__); 1389 return -ENOMEM; 1390 1391 out_rc: 1392 rpcrdma_recv_buffer_put(rep); 1393 return rc; 1394 } 1395