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 * verbs.c 44 * 45 * Encapsulates the major functions managing: 46 * o adapters 47 * o endpoints 48 * o connections 49 * o buffer memory 50 */ 51 52 #include <linux/interrupt.h> 53 #include <linux/slab.h> 54 #include <linux/sunrpc/addr.h> 55 #include <linux/sunrpc/svc_rdma.h> 56 57 #include <asm-generic/barrier.h> 58 #include <asm/bitops.h> 59 60 #include <rdma/ib_cm.h> 61 62 #include "xprt_rdma.h" 63 #include <trace/events/rpcrdma.h> 64 65 /* 66 * Globals/Macros 67 */ 68 69 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 70 # define RPCDBG_FACILITY RPCDBG_TRANS 71 #endif 72 73 /* 74 * internal functions 75 */ 76 static void rpcrdma_sendctx_put_locked(struct rpcrdma_sendctx *sc); 77 static void rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt); 78 static void rpcrdma_mrs_destroy(struct rpcrdma_buffer *buf); 79 static int rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt, bool temp); 80 static void rpcrdma_dma_unmap_regbuf(struct rpcrdma_regbuf *rb); 81 82 struct workqueue_struct *rpcrdma_receive_wq __read_mostly; 83 84 int 85 rpcrdma_alloc_wq(void) 86 { 87 struct workqueue_struct *recv_wq; 88 89 recv_wq = alloc_workqueue("xprtrdma_receive", 90 WQ_MEM_RECLAIM | WQ_HIGHPRI, 91 0); 92 if (!recv_wq) 93 return -ENOMEM; 94 95 rpcrdma_receive_wq = recv_wq; 96 return 0; 97 } 98 99 void 100 rpcrdma_destroy_wq(void) 101 { 102 struct workqueue_struct *wq; 103 104 if (rpcrdma_receive_wq) { 105 wq = rpcrdma_receive_wq; 106 rpcrdma_receive_wq = NULL; 107 destroy_workqueue(wq); 108 } 109 } 110 111 static void 112 rpcrdma_qp_async_error_upcall(struct ib_event *event, void *context) 113 { 114 struct rpcrdma_ep *ep = context; 115 struct rpcrdma_xprt *r_xprt = container_of(ep, struct rpcrdma_xprt, 116 rx_ep); 117 118 trace_xprtrdma_qp_error(r_xprt, event); 119 pr_err("rpcrdma: %s on device %s ep %p\n", 120 ib_event_msg(event->event), event->device->name, context); 121 122 if (ep->rep_connected == 1) { 123 ep->rep_connected = -EIO; 124 rpcrdma_conn_func(ep); 125 wake_up_all(&ep->rep_connect_wait); 126 } 127 } 128 129 /** 130 * rpcrdma_wc_send - Invoked by RDMA provider for each polled Send WC 131 * @cq: completion queue (ignored) 132 * @wc: completed WR 133 * 134 */ 135 static void 136 rpcrdma_wc_send(struct ib_cq *cq, struct ib_wc *wc) 137 { 138 struct ib_cqe *cqe = wc->wr_cqe; 139 struct rpcrdma_sendctx *sc = 140 container_of(cqe, struct rpcrdma_sendctx, sc_cqe); 141 142 /* WARNING: Only wr_cqe and status are reliable at this point */ 143 trace_xprtrdma_wc_send(sc, wc); 144 if (wc->status != IB_WC_SUCCESS && wc->status != IB_WC_WR_FLUSH_ERR) 145 pr_err("rpcrdma: Send: %s (%u/0x%x)\n", 146 ib_wc_status_msg(wc->status), 147 wc->status, wc->vendor_err); 148 149 rpcrdma_sendctx_put_locked(sc); 150 } 151 152 /** 153 * rpcrdma_wc_receive - Invoked by RDMA provider for each polled Receive WC 154 * @cq: completion queue (ignored) 155 * @wc: completed WR 156 * 157 */ 158 static void 159 rpcrdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc) 160 { 161 struct ib_cqe *cqe = wc->wr_cqe; 162 struct rpcrdma_rep *rep = container_of(cqe, struct rpcrdma_rep, 163 rr_cqe); 164 165 /* WARNING: Only wr_id and status are reliable at this point */ 166 trace_xprtrdma_wc_receive(wc); 167 if (wc->status != IB_WC_SUCCESS) 168 goto out_fail; 169 170 /* status == SUCCESS means all fields in wc are trustworthy */ 171 rpcrdma_set_xdrlen(&rep->rr_hdrbuf, wc->byte_len); 172 rep->rr_wc_flags = wc->wc_flags; 173 rep->rr_inv_rkey = wc->ex.invalidate_rkey; 174 175 ib_dma_sync_single_for_cpu(rdmab_device(rep->rr_rdmabuf), 176 rdmab_addr(rep->rr_rdmabuf), 177 wc->byte_len, DMA_FROM_DEVICE); 178 179 out_schedule: 180 rpcrdma_reply_handler(rep); 181 return; 182 183 out_fail: 184 if (wc->status != IB_WC_WR_FLUSH_ERR) 185 pr_err("rpcrdma: Recv: %s (%u/0x%x)\n", 186 ib_wc_status_msg(wc->status), 187 wc->status, wc->vendor_err); 188 rpcrdma_set_xdrlen(&rep->rr_hdrbuf, 0); 189 goto out_schedule; 190 } 191 192 static void 193 rpcrdma_update_connect_private(struct rpcrdma_xprt *r_xprt, 194 struct rdma_conn_param *param) 195 { 196 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 197 const struct rpcrdma_connect_private *pmsg = param->private_data; 198 unsigned int rsize, wsize; 199 200 /* Default settings for RPC-over-RDMA Version One */ 201 r_xprt->rx_ia.ri_implicit_roundup = xprt_rdma_pad_optimize; 202 rsize = RPCRDMA_V1_DEF_INLINE_SIZE; 203 wsize = RPCRDMA_V1_DEF_INLINE_SIZE; 204 205 if (pmsg && 206 pmsg->cp_magic == rpcrdma_cmp_magic && 207 pmsg->cp_version == RPCRDMA_CMP_VERSION) { 208 r_xprt->rx_ia.ri_implicit_roundup = true; 209 rsize = rpcrdma_decode_buffer_size(pmsg->cp_send_size); 210 wsize = rpcrdma_decode_buffer_size(pmsg->cp_recv_size); 211 } 212 213 if (rsize < cdata->inline_rsize) 214 cdata->inline_rsize = rsize; 215 if (wsize < cdata->inline_wsize) 216 cdata->inline_wsize = wsize; 217 dprintk("RPC: %s: max send %u, max recv %u\n", 218 __func__, cdata->inline_wsize, cdata->inline_rsize); 219 rpcrdma_set_max_header_sizes(r_xprt); 220 } 221 222 static int 223 rpcrdma_conn_upcall(struct rdma_cm_id *id, struct rdma_cm_event *event) 224 { 225 struct rpcrdma_xprt *xprt = id->context; 226 struct rpcrdma_ia *ia = &xprt->rx_ia; 227 struct rpcrdma_ep *ep = &xprt->rx_ep; 228 int connstate = 0; 229 230 trace_xprtrdma_conn_upcall(xprt, event); 231 switch (event->event) { 232 case RDMA_CM_EVENT_ADDR_RESOLVED: 233 case RDMA_CM_EVENT_ROUTE_RESOLVED: 234 ia->ri_async_rc = 0; 235 complete(&ia->ri_done); 236 break; 237 case RDMA_CM_EVENT_ADDR_ERROR: 238 ia->ri_async_rc = -EPROTO; 239 complete(&ia->ri_done); 240 break; 241 case RDMA_CM_EVENT_ROUTE_ERROR: 242 ia->ri_async_rc = -ENETUNREACH; 243 complete(&ia->ri_done); 244 break; 245 case RDMA_CM_EVENT_DEVICE_REMOVAL: 246 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 247 pr_info("rpcrdma: removing device %s for %s:%s\n", 248 ia->ri_device->name, 249 rpcrdma_addrstr(xprt), rpcrdma_portstr(xprt)); 250 #endif 251 set_bit(RPCRDMA_IAF_REMOVING, &ia->ri_flags); 252 ep->rep_connected = -ENODEV; 253 xprt_force_disconnect(&xprt->rx_xprt); 254 wait_for_completion(&ia->ri_remove_done); 255 256 ia->ri_id = NULL; 257 ia->ri_device = NULL; 258 /* Return 1 to ensure the core destroys the id. */ 259 return 1; 260 case RDMA_CM_EVENT_ESTABLISHED: 261 ++xprt->rx_xprt.connect_cookie; 262 connstate = 1; 263 rpcrdma_update_connect_private(xprt, &event->param.conn); 264 goto connected; 265 case RDMA_CM_EVENT_CONNECT_ERROR: 266 connstate = -ENOTCONN; 267 goto connected; 268 case RDMA_CM_EVENT_UNREACHABLE: 269 connstate = -ENETUNREACH; 270 goto connected; 271 case RDMA_CM_EVENT_REJECTED: 272 dprintk("rpcrdma: connection to %s:%s rejected: %s\n", 273 rpcrdma_addrstr(xprt), rpcrdma_portstr(xprt), 274 rdma_reject_msg(id, event->status)); 275 connstate = -ECONNREFUSED; 276 if (event->status == IB_CM_REJ_STALE_CONN) 277 connstate = -EAGAIN; 278 goto connected; 279 case RDMA_CM_EVENT_DISCONNECTED: 280 ++xprt->rx_xprt.connect_cookie; 281 connstate = -ECONNABORTED; 282 connected: 283 ep->rep_connected = connstate; 284 rpcrdma_conn_func(ep); 285 wake_up_all(&ep->rep_connect_wait); 286 /*FALLTHROUGH*/ 287 default: 288 dprintk("RPC: %s: %s:%s on %s/%s (ep 0x%p): %s\n", 289 __func__, 290 rpcrdma_addrstr(xprt), rpcrdma_portstr(xprt), 291 ia->ri_device->name, ia->ri_ops->ro_displayname, 292 ep, rdma_event_msg(event->event)); 293 break; 294 } 295 296 return 0; 297 } 298 299 static struct rdma_cm_id * 300 rpcrdma_create_id(struct rpcrdma_xprt *xprt, struct rpcrdma_ia *ia) 301 { 302 unsigned long wtimeout = msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1; 303 struct rdma_cm_id *id; 304 int rc; 305 306 trace_xprtrdma_conn_start(xprt); 307 308 init_completion(&ia->ri_done); 309 init_completion(&ia->ri_remove_done); 310 311 id = rdma_create_id(xprt->rx_xprt.xprt_net, rpcrdma_conn_upcall, 312 xprt, RDMA_PS_TCP, IB_QPT_RC); 313 if (IS_ERR(id)) { 314 rc = PTR_ERR(id); 315 dprintk("RPC: %s: rdma_create_id() failed %i\n", 316 __func__, rc); 317 return id; 318 } 319 320 ia->ri_async_rc = -ETIMEDOUT; 321 rc = rdma_resolve_addr(id, NULL, 322 (struct sockaddr *)&xprt->rx_xprt.addr, 323 RDMA_RESOLVE_TIMEOUT); 324 if (rc) { 325 dprintk("RPC: %s: rdma_resolve_addr() failed %i\n", 326 __func__, rc); 327 goto out; 328 } 329 rc = wait_for_completion_interruptible_timeout(&ia->ri_done, wtimeout); 330 if (rc < 0) { 331 trace_xprtrdma_conn_tout(xprt); 332 goto out; 333 } 334 335 rc = ia->ri_async_rc; 336 if (rc) 337 goto out; 338 339 ia->ri_async_rc = -ETIMEDOUT; 340 rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT); 341 if (rc) { 342 dprintk("RPC: %s: rdma_resolve_route() failed %i\n", 343 __func__, rc); 344 goto out; 345 } 346 rc = wait_for_completion_interruptible_timeout(&ia->ri_done, wtimeout); 347 if (rc < 0) { 348 trace_xprtrdma_conn_tout(xprt); 349 goto out; 350 } 351 rc = ia->ri_async_rc; 352 if (rc) 353 goto out; 354 355 return id; 356 357 out: 358 rdma_destroy_id(id); 359 return ERR_PTR(rc); 360 } 361 362 /* 363 * Exported functions. 364 */ 365 366 /** 367 * rpcrdma_ia_open - Open and initialize an Interface Adapter. 368 * @xprt: transport with IA to (re)initialize 369 * 370 * Returns 0 on success, negative errno if an appropriate 371 * Interface Adapter could not be found and opened. 372 */ 373 int 374 rpcrdma_ia_open(struct rpcrdma_xprt *xprt) 375 { 376 struct rpcrdma_ia *ia = &xprt->rx_ia; 377 int rc; 378 379 ia->ri_id = rpcrdma_create_id(xprt, ia); 380 if (IS_ERR(ia->ri_id)) { 381 rc = PTR_ERR(ia->ri_id); 382 goto out_err; 383 } 384 ia->ri_device = ia->ri_id->device; 385 386 ia->ri_pd = ib_alloc_pd(ia->ri_device, 0); 387 if (IS_ERR(ia->ri_pd)) { 388 rc = PTR_ERR(ia->ri_pd); 389 pr_err("rpcrdma: ib_alloc_pd() returned %d\n", rc); 390 goto out_err; 391 } 392 393 switch (xprt_rdma_memreg_strategy) { 394 case RPCRDMA_FRWR: 395 if (frwr_is_supported(ia)) { 396 ia->ri_ops = &rpcrdma_frwr_memreg_ops; 397 break; 398 } 399 /*FALLTHROUGH*/ 400 case RPCRDMA_MTHCAFMR: 401 if (fmr_is_supported(ia)) { 402 ia->ri_ops = &rpcrdma_fmr_memreg_ops; 403 break; 404 } 405 /*FALLTHROUGH*/ 406 default: 407 pr_err("rpcrdma: Device %s does not support memreg mode %d\n", 408 ia->ri_device->name, xprt_rdma_memreg_strategy); 409 rc = -EINVAL; 410 goto out_err; 411 } 412 413 return 0; 414 415 out_err: 416 rpcrdma_ia_close(ia); 417 return rc; 418 } 419 420 /** 421 * rpcrdma_ia_remove - Handle device driver unload 422 * @ia: interface adapter being removed 423 * 424 * Divest transport H/W resources associated with this adapter, 425 * but allow it to be restored later. 426 */ 427 void 428 rpcrdma_ia_remove(struct rpcrdma_ia *ia) 429 { 430 struct rpcrdma_xprt *r_xprt = container_of(ia, struct rpcrdma_xprt, 431 rx_ia); 432 struct rpcrdma_ep *ep = &r_xprt->rx_ep; 433 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 434 struct rpcrdma_req *req; 435 struct rpcrdma_rep *rep; 436 437 cancel_delayed_work_sync(&buf->rb_refresh_worker); 438 439 /* This is similar to rpcrdma_ep_destroy, but: 440 * - Don't cancel the connect worker. 441 * - Don't call rpcrdma_ep_disconnect, which waits 442 * for another conn upcall, which will deadlock. 443 * - rdma_disconnect is unneeded, the underlying 444 * connection is already gone. 445 */ 446 if (ia->ri_id->qp) { 447 ib_drain_qp(ia->ri_id->qp); 448 rdma_destroy_qp(ia->ri_id); 449 ia->ri_id->qp = NULL; 450 } 451 ib_free_cq(ep->rep_attr.recv_cq); 452 ep->rep_attr.recv_cq = NULL; 453 ib_free_cq(ep->rep_attr.send_cq); 454 ep->rep_attr.send_cq = NULL; 455 456 /* The ULP is responsible for ensuring all DMA 457 * mappings and MRs are gone. 458 */ 459 list_for_each_entry(rep, &buf->rb_recv_bufs, rr_list) 460 rpcrdma_dma_unmap_regbuf(rep->rr_rdmabuf); 461 list_for_each_entry(req, &buf->rb_allreqs, rl_all) { 462 rpcrdma_dma_unmap_regbuf(req->rl_rdmabuf); 463 rpcrdma_dma_unmap_regbuf(req->rl_sendbuf); 464 rpcrdma_dma_unmap_regbuf(req->rl_recvbuf); 465 } 466 rpcrdma_mrs_destroy(buf); 467 ib_dealloc_pd(ia->ri_pd); 468 ia->ri_pd = NULL; 469 470 /* Allow waiters to continue */ 471 complete(&ia->ri_remove_done); 472 473 trace_xprtrdma_remove(r_xprt); 474 } 475 476 /** 477 * rpcrdma_ia_close - Clean up/close an IA. 478 * @ia: interface adapter to close 479 * 480 */ 481 void 482 rpcrdma_ia_close(struct rpcrdma_ia *ia) 483 { 484 if (ia->ri_id != NULL && !IS_ERR(ia->ri_id)) { 485 if (ia->ri_id->qp) 486 rdma_destroy_qp(ia->ri_id); 487 rdma_destroy_id(ia->ri_id); 488 } 489 ia->ri_id = NULL; 490 ia->ri_device = NULL; 491 492 /* If the pd is still busy, xprtrdma missed freeing a resource */ 493 if (ia->ri_pd && !IS_ERR(ia->ri_pd)) 494 ib_dealloc_pd(ia->ri_pd); 495 ia->ri_pd = NULL; 496 } 497 498 /* 499 * Create unconnected endpoint. 500 */ 501 int 502 rpcrdma_ep_create(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia, 503 struct rpcrdma_create_data_internal *cdata) 504 { 505 struct rpcrdma_connect_private *pmsg = &ep->rep_cm_private; 506 struct ib_cq *sendcq, *recvcq; 507 unsigned int max_sge; 508 int rc; 509 510 max_sge = min_t(unsigned int, ia->ri_device->attrs.max_send_sge, 511 RPCRDMA_MAX_SEND_SGES); 512 if (max_sge < RPCRDMA_MIN_SEND_SGES) { 513 pr_warn("rpcrdma: HCA provides only %d send SGEs\n", max_sge); 514 return -ENOMEM; 515 } 516 ia->ri_max_send_sges = max_sge; 517 518 rc = ia->ri_ops->ro_open(ia, ep, cdata); 519 if (rc) 520 return rc; 521 522 ep->rep_attr.event_handler = rpcrdma_qp_async_error_upcall; 523 ep->rep_attr.qp_context = ep; 524 ep->rep_attr.srq = NULL; 525 ep->rep_attr.cap.max_send_sge = max_sge; 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_send_batch = min_t(unsigned int, RPCRDMA_MAX_SEND_BATCH, 542 cdata->max_requests >> 2); 543 ep->rep_send_count = ep->rep_send_batch; 544 init_waitqueue_head(&ep->rep_connect_wait); 545 INIT_DELAYED_WORK(&ep->rep_connect_worker, rpcrdma_connect_worker); 546 547 sendcq = ib_alloc_cq(ia->ri_device, NULL, 548 ep->rep_attr.cap.max_send_wr + 1, 549 1, IB_POLL_WORKQUEUE); 550 if (IS_ERR(sendcq)) { 551 rc = PTR_ERR(sendcq); 552 dprintk("RPC: %s: failed to create send CQ: %i\n", 553 __func__, rc); 554 goto out1; 555 } 556 557 recvcq = ib_alloc_cq(ia->ri_device, NULL, 558 ep->rep_attr.cap.max_recv_wr + 1, 559 0, IB_POLL_WORKQUEUE); 560 if (IS_ERR(recvcq)) { 561 rc = PTR_ERR(recvcq); 562 dprintk("RPC: %s: failed to create recv CQ: %i\n", 563 __func__, rc); 564 goto out2; 565 } 566 567 ep->rep_attr.send_cq = sendcq; 568 ep->rep_attr.recv_cq = recvcq; 569 570 /* Initialize cma parameters */ 571 memset(&ep->rep_remote_cma, 0, sizeof(ep->rep_remote_cma)); 572 573 /* Prepare RDMA-CM private message */ 574 pmsg->cp_magic = rpcrdma_cmp_magic; 575 pmsg->cp_version = RPCRDMA_CMP_VERSION; 576 pmsg->cp_flags |= ia->ri_ops->ro_send_w_inv_ok; 577 pmsg->cp_send_size = rpcrdma_encode_buffer_size(cdata->inline_wsize); 578 pmsg->cp_recv_size = rpcrdma_encode_buffer_size(cdata->inline_rsize); 579 ep->rep_remote_cma.private_data = pmsg; 580 ep->rep_remote_cma.private_data_len = sizeof(*pmsg); 581 582 /* Client offers RDMA Read but does not initiate */ 583 ep->rep_remote_cma.initiator_depth = 0; 584 ep->rep_remote_cma.responder_resources = 585 min_t(int, U8_MAX, ia->ri_device->attrs.max_qp_rd_atom); 586 587 /* Limit transport retries so client can detect server 588 * GID changes quickly. RPC layer handles re-establishing 589 * transport connection and retransmission. 590 */ 591 ep->rep_remote_cma.retry_count = 6; 592 593 /* RPC-over-RDMA handles its own flow control. In addition, 594 * make all RNR NAKs visible so we know that RPC-over-RDMA 595 * flow control is working correctly (no NAKs should be seen). 596 */ 597 ep->rep_remote_cma.flow_control = 0; 598 ep->rep_remote_cma.rnr_retry_count = 0; 599 600 return 0; 601 602 out2: 603 ib_free_cq(sendcq); 604 out1: 605 return rc; 606 } 607 608 /* 609 * rpcrdma_ep_destroy 610 * 611 * Disconnect and destroy endpoint. After this, the only 612 * valid operations on the ep are to free it (if dynamically 613 * allocated) or re-create it. 614 */ 615 void 616 rpcrdma_ep_destroy(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 617 { 618 cancel_delayed_work_sync(&ep->rep_connect_worker); 619 620 if (ia->ri_id && ia->ri_id->qp) { 621 rpcrdma_ep_disconnect(ep, ia); 622 rdma_destroy_qp(ia->ri_id); 623 ia->ri_id->qp = NULL; 624 } 625 626 if (ep->rep_attr.recv_cq) 627 ib_free_cq(ep->rep_attr.recv_cq); 628 if (ep->rep_attr.send_cq) 629 ib_free_cq(ep->rep_attr.send_cq); 630 } 631 632 /* Re-establish a connection after a device removal event. 633 * Unlike a normal reconnection, a fresh PD and a new set 634 * of MRs and buffers is needed. 635 */ 636 static int 637 rpcrdma_ep_recreate_xprt(struct rpcrdma_xprt *r_xprt, 638 struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 639 { 640 int rc, err; 641 642 trace_xprtrdma_reinsert(r_xprt); 643 644 rc = -EHOSTUNREACH; 645 if (rpcrdma_ia_open(r_xprt)) 646 goto out1; 647 648 rc = -ENOMEM; 649 err = rpcrdma_ep_create(ep, ia, &r_xprt->rx_data); 650 if (err) { 651 pr_err("rpcrdma: rpcrdma_ep_create returned %d\n", err); 652 goto out2; 653 } 654 655 rc = -ENETUNREACH; 656 err = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr); 657 if (err) { 658 pr_err("rpcrdma: rdma_create_qp returned %d\n", err); 659 goto out3; 660 } 661 662 rpcrdma_mrs_create(r_xprt); 663 return 0; 664 665 out3: 666 rpcrdma_ep_destroy(ep, ia); 667 out2: 668 rpcrdma_ia_close(ia); 669 out1: 670 return rc; 671 } 672 673 static int 674 rpcrdma_ep_reconnect(struct rpcrdma_xprt *r_xprt, struct rpcrdma_ep *ep, 675 struct rpcrdma_ia *ia) 676 { 677 struct rdma_cm_id *id, *old; 678 int err, rc; 679 680 trace_xprtrdma_reconnect(r_xprt); 681 682 rpcrdma_ep_disconnect(ep, ia); 683 684 rc = -EHOSTUNREACH; 685 id = rpcrdma_create_id(r_xprt, ia); 686 if (IS_ERR(id)) 687 goto out; 688 689 /* As long as the new ID points to the same device as the 690 * old ID, we can reuse the transport's existing PD and all 691 * previously allocated MRs. Also, the same device means 692 * the transport's previous DMA mappings are still valid. 693 * 694 * This is a sanity check only. There should be no way these 695 * point to two different devices here. 696 */ 697 old = id; 698 rc = -ENETUNREACH; 699 if (ia->ri_device != id->device) { 700 pr_err("rpcrdma: can't reconnect on different device!\n"); 701 goto out_destroy; 702 } 703 704 err = rdma_create_qp(id, ia->ri_pd, &ep->rep_attr); 705 if (err) { 706 dprintk("RPC: %s: rdma_create_qp returned %d\n", 707 __func__, err); 708 goto out_destroy; 709 } 710 711 /* Atomically replace the transport's ID and QP. */ 712 rc = 0; 713 old = ia->ri_id; 714 ia->ri_id = id; 715 rdma_destroy_qp(old); 716 717 out_destroy: 718 rdma_destroy_id(old); 719 out: 720 return rc; 721 } 722 723 /* 724 * Connect unconnected endpoint. 725 */ 726 int 727 rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 728 { 729 struct rpcrdma_xprt *r_xprt = container_of(ia, struct rpcrdma_xprt, 730 rx_ia); 731 int rc; 732 733 retry: 734 switch (ep->rep_connected) { 735 case 0: 736 dprintk("RPC: %s: connecting...\n", __func__); 737 rc = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr); 738 if (rc) { 739 dprintk("RPC: %s: rdma_create_qp failed %i\n", 740 __func__, rc); 741 rc = -ENETUNREACH; 742 goto out_noupdate; 743 } 744 break; 745 case -ENODEV: 746 rc = rpcrdma_ep_recreate_xprt(r_xprt, ep, ia); 747 if (rc) 748 goto out_noupdate; 749 break; 750 default: 751 rc = rpcrdma_ep_reconnect(r_xprt, ep, ia); 752 if (rc) 753 goto out; 754 } 755 756 ep->rep_connected = 0; 757 rpcrdma_post_recvs(r_xprt, true); 758 759 rc = rdma_connect(ia->ri_id, &ep->rep_remote_cma); 760 if (rc) { 761 dprintk("RPC: %s: rdma_connect() failed with %i\n", 762 __func__, rc); 763 goto out; 764 } 765 766 wait_event_interruptible(ep->rep_connect_wait, ep->rep_connected != 0); 767 if (ep->rep_connected <= 0) { 768 if (ep->rep_connected == -EAGAIN) 769 goto retry; 770 rc = ep->rep_connected; 771 goto out; 772 } 773 774 dprintk("RPC: %s: connected\n", __func__); 775 776 out: 777 if (rc) 778 ep->rep_connected = rc; 779 780 out_noupdate: 781 return rc; 782 } 783 784 /* 785 * rpcrdma_ep_disconnect 786 * 787 * This is separate from destroy to facilitate the ability 788 * to reconnect without recreating the endpoint. 789 * 790 * This call is not reentrant, and must not be made in parallel 791 * on the same endpoint. 792 */ 793 void 794 rpcrdma_ep_disconnect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 795 { 796 int rc; 797 798 rc = rdma_disconnect(ia->ri_id); 799 if (!rc) 800 /* returns without wait if not connected */ 801 wait_event_interruptible(ep->rep_connect_wait, 802 ep->rep_connected != 1); 803 else 804 ep->rep_connected = rc; 805 trace_xprtrdma_disconnect(container_of(ep, struct rpcrdma_xprt, 806 rx_ep), rc); 807 808 ib_drain_qp(ia->ri_id->qp); 809 } 810 811 /* Fixed-size circular FIFO queue. This implementation is wait-free and 812 * lock-free. 813 * 814 * Consumer is the code path that posts Sends. This path dequeues a 815 * sendctx for use by a Send operation. Multiple consumer threads 816 * are serialized by the RPC transport lock, which allows only one 817 * ->send_request call at a time. 818 * 819 * Producer is the code path that handles Send completions. This path 820 * enqueues a sendctx that has been completed. Multiple producer 821 * threads are serialized by the ib_poll_cq() function. 822 */ 823 824 /* rpcrdma_sendctxs_destroy() assumes caller has already quiesced 825 * queue activity, and ib_drain_qp has flushed all remaining Send 826 * requests. 827 */ 828 static void rpcrdma_sendctxs_destroy(struct rpcrdma_buffer *buf) 829 { 830 unsigned long i; 831 832 for (i = 0; i <= buf->rb_sc_last; i++) 833 kfree(buf->rb_sc_ctxs[i]); 834 kfree(buf->rb_sc_ctxs); 835 } 836 837 static struct rpcrdma_sendctx *rpcrdma_sendctx_create(struct rpcrdma_ia *ia) 838 { 839 struct rpcrdma_sendctx *sc; 840 841 sc = kzalloc(sizeof(*sc) + 842 ia->ri_max_send_sges * sizeof(struct ib_sge), 843 GFP_KERNEL); 844 if (!sc) 845 return NULL; 846 847 sc->sc_wr.wr_cqe = &sc->sc_cqe; 848 sc->sc_wr.sg_list = sc->sc_sges; 849 sc->sc_wr.opcode = IB_WR_SEND; 850 sc->sc_cqe.done = rpcrdma_wc_send; 851 return sc; 852 } 853 854 static int rpcrdma_sendctxs_create(struct rpcrdma_xprt *r_xprt) 855 { 856 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 857 struct rpcrdma_sendctx *sc; 858 unsigned long i; 859 860 /* Maximum number of concurrent outstanding Send WRs. Capping 861 * the circular queue size stops Send Queue overflow by causing 862 * the ->send_request call to fail temporarily before too many 863 * Sends are posted. 864 */ 865 i = buf->rb_max_requests + RPCRDMA_MAX_BC_REQUESTS; 866 dprintk("RPC: %s: allocating %lu send_ctxs\n", __func__, i); 867 buf->rb_sc_ctxs = kcalloc(i, sizeof(sc), GFP_KERNEL); 868 if (!buf->rb_sc_ctxs) 869 return -ENOMEM; 870 871 buf->rb_sc_last = i - 1; 872 for (i = 0; i <= buf->rb_sc_last; i++) { 873 sc = rpcrdma_sendctx_create(&r_xprt->rx_ia); 874 if (!sc) 875 goto out_destroy; 876 877 sc->sc_xprt = r_xprt; 878 buf->rb_sc_ctxs[i] = sc; 879 } 880 buf->rb_flags = 0; 881 882 return 0; 883 884 out_destroy: 885 rpcrdma_sendctxs_destroy(buf); 886 return -ENOMEM; 887 } 888 889 /* The sendctx queue is not guaranteed to have a size that is a 890 * power of two, thus the helpers in circ_buf.h cannot be used. 891 * The other option is to use modulus (%), which can be expensive. 892 */ 893 static unsigned long rpcrdma_sendctx_next(struct rpcrdma_buffer *buf, 894 unsigned long item) 895 { 896 return likely(item < buf->rb_sc_last) ? item + 1 : 0; 897 } 898 899 /** 900 * rpcrdma_sendctx_get_locked - Acquire a send context 901 * @buf: transport buffers from which to acquire an unused context 902 * 903 * Returns pointer to a free send completion context; or NULL if 904 * the queue is empty. 905 * 906 * Usage: Called to acquire an SGE array before preparing a Send WR. 907 * 908 * The caller serializes calls to this function (per rpcrdma_buffer), 909 * and provides an effective memory barrier that flushes the new value 910 * of rb_sc_head. 911 */ 912 struct rpcrdma_sendctx *rpcrdma_sendctx_get_locked(struct rpcrdma_buffer *buf) 913 { 914 struct rpcrdma_xprt *r_xprt; 915 struct rpcrdma_sendctx *sc; 916 unsigned long next_head; 917 918 next_head = rpcrdma_sendctx_next(buf, buf->rb_sc_head); 919 920 if (next_head == READ_ONCE(buf->rb_sc_tail)) 921 goto out_emptyq; 922 923 /* ORDER: item must be accessed _before_ head is updated */ 924 sc = buf->rb_sc_ctxs[next_head]; 925 926 /* Releasing the lock in the caller acts as a memory 927 * barrier that flushes rb_sc_head. 928 */ 929 buf->rb_sc_head = next_head; 930 931 return sc; 932 933 out_emptyq: 934 /* The queue is "empty" if there have not been enough Send 935 * completions recently. This is a sign the Send Queue is 936 * backing up. Cause the caller to pause and try again. 937 */ 938 set_bit(RPCRDMA_BUF_F_EMPTY_SCQ, &buf->rb_flags); 939 r_xprt = container_of(buf, struct rpcrdma_xprt, rx_buf); 940 r_xprt->rx_stats.empty_sendctx_q++; 941 return NULL; 942 } 943 944 /** 945 * rpcrdma_sendctx_put_locked - Release a send context 946 * @sc: send context to release 947 * 948 * Usage: Called from Send completion to return a sendctxt 949 * to the queue. 950 * 951 * The caller serializes calls to this function (per rpcrdma_buffer). 952 */ 953 static void 954 rpcrdma_sendctx_put_locked(struct rpcrdma_sendctx *sc) 955 { 956 struct rpcrdma_buffer *buf = &sc->sc_xprt->rx_buf; 957 unsigned long next_tail; 958 959 /* Unmap SGEs of previously completed by unsignaled 960 * Sends by walking up the queue until @sc is found. 961 */ 962 next_tail = buf->rb_sc_tail; 963 do { 964 next_tail = rpcrdma_sendctx_next(buf, next_tail); 965 966 /* ORDER: item must be accessed _before_ tail is updated */ 967 rpcrdma_unmap_sendctx(buf->rb_sc_ctxs[next_tail]); 968 969 } while (buf->rb_sc_ctxs[next_tail] != sc); 970 971 /* Paired with READ_ONCE */ 972 smp_store_release(&buf->rb_sc_tail, next_tail); 973 974 if (test_and_clear_bit(RPCRDMA_BUF_F_EMPTY_SCQ, &buf->rb_flags)) { 975 smp_mb__after_atomic(); 976 xprt_write_space(&sc->sc_xprt->rx_xprt); 977 } 978 } 979 980 static void 981 rpcrdma_mr_recovery_worker(struct work_struct *work) 982 { 983 struct rpcrdma_buffer *buf = container_of(work, struct rpcrdma_buffer, 984 rb_recovery_worker.work); 985 struct rpcrdma_mr *mr; 986 987 spin_lock(&buf->rb_recovery_lock); 988 while (!list_empty(&buf->rb_stale_mrs)) { 989 mr = rpcrdma_mr_pop(&buf->rb_stale_mrs); 990 spin_unlock(&buf->rb_recovery_lock); 991 992 trace_xprtrdma_recover_mr(mr); 993 mr->mr_xprt->rx_ia.ri_ops->ro_recover_mr(mr); 994 995 spin_lock(&buf->rb_recovery_lock); 996 } 997 spin_unlock(&buf->rb_recovery_lock); 998 } 999 1000 void 1001 rpcrdma_mr_defer_recovery(struct rpcrdma_mr *mr) 1002 { 1003 struct rpcrdma_xprt *r_xprt = mr->mr_xprt; 1004 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1005 1006 spin_lock(&buf->rb_recovery_lock); 1007 rpcrdma_mr_push(mr, &buf->rb_stale_mrs); 1008 spin_unlock(&buf->rb_recovery_lock); 1009 1010 schedule_delayed_work(&buf->rb_recovery_worker, 0); 1011 } 1012 1013 static void 1014 rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt) 1015 { 1016 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1017 struct rpcrdma_ia *ia = &r_xprt->rx_ia; 1018 unsigned int count; 1019 LIST_HEAD(free); 1020 LIST_HEAD(all); 1021 1022 for (count = 0; count < 3; count++) { 1023 struct rpcrdma_mr *mr; 1024 int rc; 1025 1026 mr = kzalloc(sizeof(*mr), GFP_KERNEL); 1027 if (!mr) 1028 break; 1029 1030 rc = ia->ri_ops->ro_init_mr(ia, mr); 1031 if (rc) { 1032 kfree(mr); 1033 break; 1034 } 1035 1036 mr->mr_xprt = r_xprt; 1037 1038 list_add(&mr->mr_list, &free); 1039 list_add(&mr->mr_all, &all); 1040 } 1041 1042 spin_lock(&buf->rb_mrlock); 1043 list_splice(&free, &buf->rb_mrs); 1044 list_splice(&all, &buf->rb_all); 1045 r_xprt->rx_stats.mrs_allocated += count; 1046 spin_unlock(&buf->rb_mrlock); 1047 trace_xprtrdma_createmrs(r_xprt, count); 1048 1049 xprt_write_space(&r_xprt->rx_xprt); 1050 } 1051 1052 static void 1053 rpcrdma_mr_refresh_worker(struct work_struct *work) 1054 { 1055 struct rpcrdma_buffer *buf = container_of(work, struct rpcrdma_buffer, 1056 rb_refresh_worker.work); 1057 struct rpcrdma_xprt *r_xprt = container_of(buf, struct rpcrdma_xprt, 1058 rx_buf); 1059 1060 rpcrdma_mrs_create(r_xprt); 1061 } 1062 1063 struct rpcrdma_req * 1064 rpcrdma_create_req(struct rpcrdma_xprt *r_xprt) 1065 { 1066 struct rpcrdma_buffer *buffer = &r_xprt->rx_buf; 1067 struct rpcrdma_regbuf *rb; 1068 struct rpcrdma_req *req; 1069 1070 req = kzalloc(sizeof(*req), GFP_KERNEL); 1071 if (req == NULL) 1072 return ERR_PTR(-ENOMEM); 1073 1074 rb = rpcrdma_alloc_regbuf(RPCRDMA_HDRBUF_SIZE, 1075 DMA_TO_DEVICE, GFP_KERNEL); 1076 if (IS_ERR(rb)) { 1077 kfree(req); 1078 return ERR_PTR(-ENOMEM); 1079 } 1080 req->rl_rdmabuf = rb; 1081 xdr_buf_init(&req->rl_hdrbuf, rb->rg_base, rdmab_length(rb)); 1082 req->rl_buffer = buffer; 1083 INIT_LIST_HEAD(&req->rl_registered); 1084 1085 spin_lock(&buffer->rb_reqslock); 1086 list_add(&req->rl_all, &buffer->rb_allreqs); 1087 spin_unlock(&buffer->rb_reqslock); 1088 return req; 1089 } 1090 1091 static int 1092 rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt, bool temp) 1093 { 1094 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 1095 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1096 struct rpcrdma_rep *rep; 1097 int rc; 1098 1099 rc = -ENOMEM; 1100 rep = kzalloc(sizeof(*rep), GFP_KERNEL); 1101 if (rep == NULL) 1102 goto out; 1103 1104 rep->rr_rdmabuf = rpcrdma_alloc_regbuf(cdata->inline_rsize, 1105 DMA_FROM_DEVICE, GFP_KERNEL); 1106 if (IS_ERR(rep->rr_rdmabuf)) { 1107 rc = PTR_ERR(rep->rr_rdmabuf); 1108 goto out_free; 1109 } 1110 xdr_buf_init(&rep->rr_hdrbuf, rep->rr_rdmabuf->rg_base, 1111 rdmab_length(rep->rr_rdmabuf)); 1112 1113 rep->rr_cqe.done = rpcrdma_wc_receive; 1114 rep->rr_rxprt = r_xprt; 1115 INIT_WORK(&rep->rr_work, rpcrdma_deferred_completion); 1116 rep->rr_recv_wr.next = NULL; 1117 rep->rr_recv_wr.wr_cqe = &rep->rr_cqe; 1118 rep->rr_recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov; 1119 rep->rr_recv_wr.num_sge = 1; 1120 rep->rr_temp = temp; 1121 1122 spin_lock(&buf->rb_lock); 1123 list_add(&rep->rr_list, &buf->rb_recv_bufs); 1124 spin_unlock(&buf->rb_lock); 1125 return 0; 1126 1127 out_free: 1128 kfree(rep); 1129 out: 1130 dprintk("RPC: %s: reply buffer %d alloc failed\n", 1131 __func__, rc); 1132 return rc; 1133 } 1134 1135 int 1136 rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt) 1137 { 1138 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1139 int i, rc; 1140 1141 buf->rb_max_requests = r_xprt->rx_data.max_requests; 1142 buf->rb_bc_srv_max_requests = 0; 1143 spin_lock_init(&buf->rb_mrlock); 1144 spin_lock_init(&buf->rb_lock); 1145 spin_lock_init(&buf->rb_recovery_lock); 1146 INIT_LIST_HEAD(&buf->rb_mrs); 1147 INIT_LIST_HEAD(&buf->rb_all); 1148 INIT_LIST_HEAD(&buf->rb_stale_mrs); 1149 INIT_DELAYED_WORK(&buf->rb_refresh_worker, 1150 rpcrdma_mr_refresh_worker); 1151 INIT_DELAYED_WORK(&buf->rb_recovery_worker, 1152 rpcrdma_mr_recovery_worker); 1153 1154 rpcrdma_mrs_create(r_xprt); 1155 1156 INIT_LIST_HEAD(&buf->rb_send_bufs); 1157 INIT_LIST_HEAD(&buf->rb_allreqs); 1158 spin_lock_init(&buf->rb_reqslock); 1159 for (i = 0; i < buf->rb_max_requests; i++) { 1160 struct rpcrdma_req *req; 1161 1162 req = rpcrdma_create_req(r_xprt); 1163 if (IS_ERR(req)) { 1164 dprintk("RPC: %s: request buffer %d alloc" 1165 " failed\n", __func__, i); 1166 rc = PTR_ERR(req); 1167 goto out; 1168 } 1169 list_add(&req->rl_list, &buf->rb_send_bufs); 1170 } 1171 1172 buf->rb_credits = 1; 1173 buf->rb_posted_receives = 0; 1174 INIT_LIST_HEAD(&buf->rb_recv_bufs); 1175 1176 rc = rpcrdma_sendctxs_create(r_xprt); 1177 if (rc) 1178 goto out; 1179 1180 return 0; 1181 out: 1182 rpcrdma_buffer_destroy(buf); 1183 return rc; 1184 } 1185 1186 static void 1187 rpcrdma_destroy_rep(struct rpcrdma_rep *rep) 1188 { 1189 rpcrdma_free_regbuf(rep->rr_rdmabuf); 1190 kfree(rep); 1191 } 1192 1193 void 1194 rpcrdma_destroy_req(struct rpcrdma_req *req) 1195 { 1196 rpcrdma_free_regbuf(req->rl_recvbuf); 1197 rpcrdma_free_regbuf(req->rl_sendbuf); 1198 rpcrdma_free_regbuf(req->rl_rdmabuf); 1199 kfree(req); 1200 } 1201 1202 static void 1203 rpcrdma_mrs_destroy(struct rpcrdma_buffer *buf) 1204 { 1205 struct rpcrdma_xprt *r_xprt = container_of(buf, struct rpcrdma_xprt, 1206 rx_buf); 1207 struct rpcrdma_ia *ia = rdmab_to_ia(buf); 1208 struct rpcrdma_mr *mr; 1209 unsigned int count; 1210 1211 count = 0; 1212 spin_lock(&buf->rb_mrlock); 1213 while (!list_empty(&buf->rb_all)) { 1214 mr = list_entry(buf->rb_all.next, struct rpcrdma_mr, mr_all); 1215 list_del(&mr->mr_all); 1216 1217 spin_unlock(&buf->rb_mrlock); 1218 1219 /* Ensure MW is not on any rl_registered list */ 1220 if (!list_empty(&mr->mr_list)) 1221 list_del(&mr->mr_list); 1222 1223 ia->ri_ops->ro_release_mr(mr); 1224 count++; 1225 spin_lock(&buf->rb_mrlock); 1226 } 1227 spin_unlock(&buf->rb_mrlock); 1228 r_xprt->rx_stats.mrs_allocated = 0; 1229 1230 dprintk("RPC: %s: released %u MRs\n", __func__, count); 1231 } 1232 1233 void 1234 rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf) 1235 { 1236 cancel_delayed_work_sync(&buf->rb_recovery_worker); 1237 cancel_delayed_work_sync(&buf->rb_refresh_worker); 1238 1239 rpcrdma_sendctxs_destroy(buf); 1240 1241 while (!list_empty(&buf->rb_recv_bufs)) { 1242 struct rpcrdma_rep *rep; 1243 1244 rep = list_first_entry(&buf->rb_recv_bufs, 1245 struct rpcrdma_rep, rr_list); 1246 list_del(&rep->rr_list); 1247 rpcrdma_destroy_rep(rep); 1248 } 1249 1250 spin_lock(&buf->rb_reqslock); 1251 while (!list_empty(&buf->rb_allreqs)) { 1252 struct rpcrdma_req *req; 1253 1254 req = list_first_entry(&buf->rb_allreqs, 1255 struct rpcrdma_req, rl_all); 1256 list_del(&req->rl_all); 1257 1258 spin_unlock(&buf->rb_reqslock); 1259 rpcrdma_destroy_req(req); 1260 spin_lock(&buf->rb_reqslock); 1261 } 1262 spin_unlock(&buf->rb_reqslock); 1263 1264 rpcrdma_mrs_destroy(buf); 1265 } 1266 1267 /** 1268 * rpcrdma_mr_get - Allocate an rpcrdma_mr object 1269 * @r_xprt: controlling transport 1270 * 1271 * Returns an initialized rpcrdma_mr or NULL if no free 1272 * rpcrdma_mr objects are available. 1273 */ 1274 struct rpcrdma_mr * 1275 rpcrdma_mr_get(struct rpcrdma_xprt *r_xprt) 1276 { 1277 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1278 struct rpcrdma_mr *mr = NULL; 1279 1280 spin_lock(&buf->rb_mrlock); 1281 if (!list_empty(&buf->rb_mrs)) 1282 mr = rpcrdma_mr_pop(&buf->rb_mrs); 1283 spin_unlock(&buf->rb_mrlock); 1284 1285 if (!mr) 1286 goto out_nomrs; 1287 return mr; 1288 1289 out_nomrs: 1290 trace_xprtrdma_nomrs(r_xprt); 1291 if (r_xprt->rx_ep.rep_connected != -ENODEV) 1292 schedule_delayed_work(&buf->rb_refresh_worker, 0); 1293 1294 /* Allow the reply handler and refresh worker to run */ 1295 cond_resched(); 1296 1297 return NULL; 1298 } 1299 1300 static void 1301 __rpcrdma_mr_put(struct rpcrdma_buffer *buf, struct rpcrdma_mr *mr) 1302 { 1303 spin_lock(&buf->rb_mrlock); 1304 rpcrdma_mr_push(mr, &buf->rb_mrs); 1305 spin_unlock(&buf->rb_mrlock); 1306 } 1307 1308 /** 1309 * rpcrdma_mr_put - Release an rpcrdma_mr object 1310 * @mr: object to release 1311 * 1312 */ 1313 void 1314 rpcrdma_mr_put(struct rpcrdma_mr *mr) 1315 { 1316 __rpcrdma_mr_put(&mr->mr_xprt->rx_buf, mr); 1317 } 1318 1319 /** 1320 * rpcrdma_mr_unmap_and_put - DMA unmap an MR and release it 1321 * @mr: object to release 1322 * 1323 */ 1324 void 1325 rpcrdma_mr_unmap_and_put(struct rpcrdma_mr *mr) 1326 { 1327 struct rpcrdma_xprt *r_xprt = mr->mr_xprt; 1328 1329 trace_xprtrdma_dma_unmap(mr); 1330 ib_dma_unmap_sg(r_xprt->rx_ia.ri_device, 1331 mr->mr_sg, mr->mr_nents, mr->mr_dir); 1332 __rpcrdma_mr_put(&r_xprt->rx_buf, mr); 1333 } 1334 1335 /** 1336 * rpcrdma_buffer_get - Get a request buffer 1337 * @buffers: Buffer pool from which to obtain a buffer 1338 * 1339 * Returns a fresh rpcrdma_req, or NULL if none are available. 1340 */ 1341 struct rpcrdma_req * 1342 rpcrdma_buffer_get(struct rpcrdma_buffer *buffers) 1343 { 1344 struct rpcrdma_req *req; 1345 1346 spin_lock(&buffers->rb_lock); 1347 req = list_first_entry_or_null(&buffers->rb_send_bufs, 1348 struct rpcrdma_req, rl_list); 1349 if (req) 1350 list_del_init(&req->rl_list); 1351 spin_unlock(&buffers->rb_lock); 1352 return req; 1353 } 1354 1355 /** 1356 * rpcrdma_buffer_put - Put request/reply buffers back into pool 1357 * @req: object to return 1358 * 1359 */ 1360 void 1361 rpcrdma_buffer_put(struct rpcrdma_req *req) 1362 { 1363 struct rpcrdma_buffer *buffers = req->rl_buffer; 1364 struct rpcrdma_rep *rep = req->rl_reply; 1365 1366 req->rl_reply = NULL; 1367 1368 spin_lock(&buffers->rb_lock); 1369 list_add(&req->rl_list, &buffers->rb_send_bufs); 1370 if (rep) { 1371 if (!rep->rr_temp) { 1372 list_add(&rep->rr_list, &buffers->rb_recv_bufs); 1373 rep = NULL; 1374 } 1375 } 1376 spin_unlock(&buffers->rb_lock); 1377 if (rep) 1378 rpcrdma_destroy_rep(rep); 1379 } 1380 1381 /* 1382 * Put reply buffers back into pool when not attached to 1383 * request. This happens in error conditions. 1384 */ 1385 void 1386 rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep) 1387 { 1388 struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf; 1389 1390 if (!rep->rr_temp) { 1391 spin_lock(&buffers->rb_lock); 1392 list_add(&rep->rr_list, &buffers->rb_recv_bufs); 1393 spin_unlock(&buffers->rb_lock); 1394 } else { 1395 rpcrdma_destroy_rep(rep); 1396 } 1397 } 1398 1399 /** 1400 * rpcrdma_alloc_regbuf - allocate and DMA-map memory for SEND/RECV buffers 1401 * @size: size of buffer to be allocated, in bytes 1402 * @direction: direction of data movement 1403 * @flags: GFP flags 1404 * 1405 * Returns an ERR_PTR, or a pointer to a regbuf, a buffer that 1406 * can be persistently DMA-mapped for I/O. 1407 * 1408 * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for 1409 * receiving the payload of RDMA RECV operations. During Long Calls 1410 * or Replies they may be registered externally via ro_map. 1411 */ 1412 struct rpcrdma_regbuf * 1413 rpcrdma_alloc_regbuf(size_t size, enum dma_data_direction direction, 1414 gfp_t flags) 1415 { 1416 struct rpcrdma_regbuf *rb; 1417 1418 rb = kmalloc(sizeof(*rb) + size, flags); 1419 if (rb == NULL) 1420 return ERR_PTR(-ENOMEM); 1421 1422 rb->rg_device = NULL; 1423 rb->rg_direction = direction; 1424 rb->rg_iov.length = size; 1425 1426 return rb; 1427 } 1428 1429 /** 1430 * __rpcrdma_map_regbuf - DMA-map a regbuf 1431 * @ia: controlling rpcrdma_ia 1432 * @rb: regbuf to be mapped 1433 */ 1434 bool 1435 __rpcrdma_dma_map_regbuf(struct rpcrdma_ia *ia, struct rpcrdma_regbuf *rb) 1436 { 1437 struct ib_device *device = ia->ri_device; 1438 1439 if (rb->rg_direction == DMA_NONE) 1440 return false; 1441 1442 rb->rg_iov.addr = ib_dma_map_single(device, 1443 (void *)rb->rg_base, 1444 rdmab_length(rb), 1445 rb->rg_direction); 1446 if (ib_dma_mapping_error(device, rdmab_addr(rb))) 1447 return false; 1448 1449 rb->rg_device = device; 1450 rb->rg_iov.lkey = ia->ri_pd->local_dma_lkey; 1451 return true; 1452 } 1453 1454 static void 1455 rpcrdma_dma_unmap_regbuf(struct rpcrdma_regbuf *rb) 1456 { 1457 if (!rb) 1458 return; 1459 1460 if (!rpcrdma_regbuf_is_mapped(rb)) 1461 return; 1462 1463 ib_dma_unmap_single(rb->rg_device, rdmab_addr(rb), 1464 rdmab_length(rb), rb->rg_direction); 1465 rb->rg_device = NULL; 1466 } 1467 1468 /** 1469 * rpcrdma_free_regbuf - deregister and free registered buffer 1470 * @rb: regbuf to be deregistered and freed 1471 */ 1472 void 1473 rpcrdma_free_regbuf(struct rpcrdma_regbuf *rb) 1474 { 1475 rpcrdma_dma_unmap_regbuf(rb); 1476 kfree(rb); 1477 } 1478 1479 /* 1480 * Prepost any receive buffer, then post send. 1481 * 1482 * Receive buffer is donated to hardware, reclaimed upon recv completion. 1483 */ 1484 int 1485 rpcrdma_ep_post(struct rpcrdma_ia *ia, 1486 struct rpcrdma_ep *ep, 1487 struct rpcrdma_req *req) 1488 { 1489 struct ib_send_wr *send_wr = &req->rl_sendctx->sc_wr; 1490 int rc; 1491 1492 if (!ep->rep_send_count || 1493 test_bit(RPCRDMA_REQ_F_TX_RESOURCES, &req->rl_flags)) { 1494 send_wr->send_flags |= IB_SEND_SIGNALED; 1495 ep->rep_send_count = ep->rep_send_batch; 1496 } else { 1497 send_wr->send_flags &= ~IB_SEND_SIGNALED; 1498 --ep->rep_send_count; 1499 } 1500 1501 rc = ia->ri_ops->ro_send(ia, req); 1502 trace_xprtrdma_post_send(req, rc); 1503 if (rc) 1504 return -ENOTCONN; 1505 return 0; 1506 } 1507 1508 /** 1509 * rpcrdma_post_recvs - Maybe post some Receive buffers 1510 * @r_xprt: controlling transport 1511 * @temp: when true, allocate temp rpcrdma_rep objects 1512 * 1513 */ 1514 void 1515 rpcrdma_post_recvs(struct rpcrdma_xprt *r_xprt, bool temp) 1516 { 1517 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1518 struct ib_recv_wr *wr, *bad_wr; 1519 int needed, count, rc; 1520 1521 needed = buf->rb_credits + (buf->rb_bc_srv_max_requests << 1); 1522 if (buf->rb_posted_receives > needed) 1523 return; 1524 needed -= buf->rb_posted_receives; 1525 1526 count = 0; 1527 wr = NULL; 1528 while (needed) { 1529 struct rpcrdma_regbuf *rb; 1530 struct rpcrdma_rep *rep; 1531 1532 spin_lock(&buf->rb_lock); 1533 rep = list_first_entry_or_null(&buf->rb_recv_bufs, 1534 struct rpcrdma_rep, rr_list); 1535 if (likely(rep)) 1536 list_del(&rep->rr_list); 1537 spin_unlock(&buf->rb_lock); 1538 if (!rep) { 1539 if (rpcrdma_create_rep(r_xprt, temp)) 1540 break; 1541 continue; 1542 } 1543 1544 rb = rep->rr_rdmabuf; 1545 if (!rpcrdma_regbuf_is_mapped(rb)) { 1546 if (!__rpcrdma_dma_map_regbuf(&r_xprt->rx_ia, rb)) { 1547 rpcrdma_recv_buffer_put(rep); 1548 break; 1549 } 1550 } 1551 1552 trace_xprtrdma_post_recv(rep->rr_recv_wr.wr_cqe); 1553 rep->rr_recv_wr.next = wr; 1554 wr = &rep->rr_recv_wr; 1555 ++count; 1556 --needed; 1557 } 1558 if (!count) 1559 return; 1560 1561 rc = ib_post_recv(r_xprt->rx_ia.ri_id->qp, wr, 1562 (const struct ib_recv_wr **)&bad_wr); 1563 if (rc) { 1564 for (wr = bad_wr; wr; wr = wr->next) { 1565 struct rpcrdma_rep *rep; 1566 1567 rep = container_of(wr, struct rpcrdma_rep, rr_recv_wr); 1568 rpcrdma_recv_buffer_put(rep); 1569 --count; 1570 } 1571 } 1572 buf->rb_posted_receives += count; 1573 trace_xprtrdma_post_recvs(r_xprt, count, rc); 1574 } 1575