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 <asm/bitops.h> 55 #include <linux/module.h> /* try_module_get()/module_put() */ 56 57 #include "xprt_rdma.h" 58 59 /* 60 * Globals/Macros 61 */ 62 63 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 64 # define RPCDBG_FACILITY RPCDBG_TRANS 65 #endif 66 67 /* 68 * internal functions 69 */ 70 71 /* 72 * handle replies in tasklet context, using a single, global list 73 * rdma tasklet function -- just turn around and call the func 74 * for all replies on the list 75 */ 76 77 static DEFINE_SPINLOCK(rpcrdma_tk_lock_g); 78 static LIST_HEAD(rpcrdma_tasklets_g); 79 80 static void 81 rpcrdma_run_tasklet(unsigned long data) 82 { 83 struct rpcrdma_rep *rep; 84 unsigned long flags; 85 86 data = data; 87 spin_lock_irqsave(&rpcrdma_tk_lock_g, flags); 88 while (!list_empty(&rpcrdma_tasklets_g)) { 89 rep = list_entry(rpcrdma_tasklets_g.next, 90 struct rpcrdma_rep, rr_list); 91 list_del(&rep->rr_list); 92 spin_unlock_irqrestore(&rpcrdma_tk_lock_g, flags); 93 94 rpcrdma_reply_handler(rep); 95 96 spin_lock_irqsave(&rpcrdma_tk_lock_g, flags); 97 } 98 spin_unlock_irqrestore(&rpcrdma_tk_lock_g, flags); 99 } 100 101 static DECLARE_TASKLET(rpcrdma_tasklet_g, rpcrdma_run_tasklet, 0UL); 102 103 static void 104 rpcrdma_schedule_tasklet(struct list_head *sched_list) 105 { 106 unsigned long flags; 107 108 spin_lock_irqsave(&rpcrdma_tk_lock_g, flags); 109 list_splice_tail(sched_list, &rpcrdma_tasklets_g); 110 spin_unlock_irqrestore(&rpcrdma_tk_lock_g, flags); 111 tasklet_schedule(&rpcrdma_tasklet_g); 112 } 113 114 static void 115 rpcrdma_qp_async_error_upcall(struct ib_event *event, void *context) 116 { 117 struct rpcrdma_ep *ep = context; 118 119 pr_err("RPC: %s: %s on device %s ep %p\n", 120 __func__, ib_event_msg(event->event), 121 event->device->name, context); 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 static void 130 rpcrdma_cq_async_error_upcall(struct ib_event *event, void *context) 131 { 132 struct rpcrdma_ep *ep = context; 133 134 pr_err("RPC: %s: %s on device %s ep %p\n", 135 __func__, ib_event_msg(event->event), 136 event->device->name, context); 137 if (ep->rep_connected == 1) { 138 ep->rep_connected = -EIO; 139 rpcrdma_conn_func(ep); 140 wake_up_all(&ep->rep_connect_wait); 141 } 142 } 143 144 static void 145 rpcrdma_sendcq_process_wc(struct ib_wc *wc) 146 { 147 /* WARNING: Only wr_id and status are reliable at this point */ 148 if (wc->wr_id == RPCRDMA_IGNORE_COMPLETION) { 149 if (wc->status != IB_WC_SUCCESS && 150 wc->status != IB_WC_WR_FLUSH_ERR) 151 pr_err("RPC: %s: SEND: %s\n", 152 __func__, ib_wc_status_msg(wc->status)); 153 } else { 154 struct rpcrdma_mw *r; 155 156 r = (struct rpcrdma_mw *)(unsigned long)wc->wr_id; 157 r->mw_sendcompletion(wc); 158 } 159 } 160 161 static int 162 rpcrdma_sendcq_poll(struct ib_cq *cq, struct rpcrdma_ep *ep) 163 { 164 struct ib_wc *wcs; 165 int budget, count, rc; 166 167 budget = RPCRDMA_WC_BUDGET / RPCRDMA_POLLSIZE; 168 do { 169 wcs = ep->rep_send_wcs; 170 171 rc = ib_poll_cq(cq, RPCRDMA_POLLSIZE, wcs); 172 if (rc <= 0) 173 return rc; 174 175 count = rc; 176 while (count-- > 0) 177 rpcrdma_sendcq_process_wc(wcs++); 178 } while (rc == RPCRDMA_POLLSIZE && --budget); 179 return 0; 180 } 181 182 /* 183 * Handle send, fast_reg_mr, and local_inv completions. 184 * 185 * Send events are typically suppressed and thus do not result 186 * in an upcall. Occasionally one is signaled, however. This 187 * prevents the provider's completion queue from wrapping and 188 * losing a completion. 189 */ 190 static void 191 rpcrdma_sendcq_upcall(struct ib_cq *cq, void *cq_context) 192 { 193 struct rpcrdma_ep *ep = (struct rpcrdma_ep *)cq_context; 194 int rc; 195 196 rc = rpcrdma_sendcq_poll(cq, ep); 197 if (rc) { 198 dprintk("RPC: %s: ib_poll_cq failed: %i\n", 199 __func__, rc); 200 return; 201 } 202 203 rc = ib_req_notify_cq(cq, 204 IB_CQ_NEXT_COMP | IB_CQ_REPORT_MISSED_EVENTS); 205 if (rc == 0) 206 return; 207 if (rc < 0) { 208 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n", 209 __func__, rc); 210 return; 211 } 212 213 rpcrdma_sendcq_poll(cq, ep); 214 } 215 216 static void 217 rpcrdma_recvcq_process_wc(struct ib_wc *wc, struct list_head *sched_list) 218 { 219 struct rpcrdma_rep *rep = 220 (struct rpcrdma_rep *)(unsigned long)wc->wr_id; 221 222 /* WARNING: Only wr_id and status are reliable at this point */ 223 if (wc->status != IB_WC_SUCCESS) 224 goto out_fail; 225 226 /* status == SUCCESS means all fields in wc are trustworthy */ 227 if (wc->opcode != IB_WC_RECV) 228 return; 229 230 dprintk("RPC: %s: rep %p opcode 'recv', length %u: success\n", 231 __func__, rep, wc->byte_len); 232 233 rep->rr_len = wc->byte_len; 234 ib_dma_sync_single_for_cpu(rep->rr_device, 235 rdmab_addr(rep->rr_rdmabuf), 236 rep->rr_len, DMA_FROM_DEVICE); 237 prefetch(rdmab_to_msg(rep->rr_rdmabuf)); 238 239 out_schedule: 240 list_add_tail(&rep->rr_list, sched_list); 241 return; 242 out_fail: 243 if (wc->status != IB_WC_WR_FLUSH_ERR) 244 pr_err("RPC: %s: rep %p: %s\n", 245 __func__, rep, ib_wc_status_msg(wc->status)); 246 rep->rr_len = ~0U; 247 goto out_schedule; 248 } 249 250 static int 251 rpcrdma_recvcq_poll(struct ib_cq *cq, struct rpcrdma_ep *ep) 252 { 253 struct list_head sched_list; 254 struct ib_wc *wcs; 255 int budget, count, rc; 256 257 INIT_LIST_HEAD(&sched_list); 258 budget = RPCRDMA_WC_BUDGET / RPCRDMA_POLLSIZE; 259 do { 260 wcs = ep->rep_recv_wcs; 261 262 rc = ib_poll_cq(cq, RPCRDMA_POLLSIZE, wcs); 263 if (rc <= 0) 264 goto out_schedule; 265 266 count = rc; 267 while (count-- > 0) 268 rpcrdma_recvcq_process_wc(wcs++, &sched_list); 269 } while (rc == RPCRDMA_POLLSIZE && --budget); 270 rc = 0; 271 272 out_schedule: 273 rpcrdma_schedule_tasklet(&sched_list); 274 return rc; 275 } 276 277 /* 278 * Handle receive completions. 279 * 280 * It is reentrant but processes single events in order to maintain 281 * ordering of receives to keep server credits. 282 * 283 * It is the responsibility of the scheduled tasklet to return 284 * recv buffers to the pool. NOTE: this affects synchronization of 285 * connection shutdown. That is, the structures required for 286 * the completion of the reply handler must remain intact until 287 * all memory has been reclaimed. 288 */ 289 static void 290 rpcrdma_recvcq_upcall(struct ib_cq *cq, void *cq_context) 291 { 292 struct rpcrdma_ep *ep = (struct rpcrdma_ep *)cq_context; 293 int rc; 294 295 rc = rpcrdma_recvcq_poll(cq, ep); 296 if (rc) { 297 dprintk("RPC: %s: ib_poll_cq failed: %i\n", 298 __func__, rc); 299 return; 300 } 301 302 rc = ib_req_notify_cq(cq, 303 IB_CQ_NEXT_COMP | IB_CQ_REPORT_MISSED_EVENTS); 304 if (rc == 0) 305 return; 306 if (rc < 0) { 307 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n", 308 __func__, rc); 309 return; 310 } 311 312 rpcrdma_recvcq_poll(cq, ep); 313 } 314 315 static void 316 rpcrdma_flush_cqs(struct rpcrdma_ep *ep) 317 { 318 struct ib_wc wc; 319 LIST_HEAD(sched_list); 320 321 while (ib_poll_cq(ep->rep_attr.recv_cq, 1, &wc) > 0) 322 rpcrdma_recvcq_process_wc(&wc, &sched_list); 323 if (!list_empty(&sched_list)) 324 rpcrdma_schedule_tasklet(&sched_list); 325 while (ib_poll_cq(ep->rep_attr.send_cq, 1, &wc) > 0) 326 rpcrdma_sendcq_process_wc(&wc); 327 } 328 329 static int 330 rpcrdma_conn_upcall(struct rdma_cm_id *id, struct rdma_cm_event *event) 331 { 332 struct rpcrdma_xprt *xprt = id->context; 333 struct rpcrdma_ia *ia = &xprt->rx_ia; 334 struct rpcrdma_ep *ep = &xprt->rx_ep; 335 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 336 struct sockaddr *sap = (struct sockaddr *)&ep->rep_remote_addr; 337 #endif 338 struct ib_qp_attr *attr = &ia->ri_qp_attr; 339 struct ib_qp_init_attr *iattr = &ia->ri_qp_init_attr; 340 int connstate = 0; 341 342 switch (event->event) { 343 case RDMA_CM_EVENT_ADDR_RESOLVED: 344 case RDMA_CM_EVENT_ROUTE_RESOLVED: 345 ia->ri_async_rc = 0; 346 complete(&ia->ri_done); 347 break; 348 case RDMA_CM_EVENT_ADDR_ERROR: 349 ia->ri_async_rc = -EHOSTUNREACH; 350 dprintk("RPC: %s: CM address resolution error, ep 0x%p\n", 351 __func__, ep); 352 complete(&ia->ri_done); 353 break; 354 case RDMA_CM_EVENT_ROUTE_ERROR: 355 ia->ri_async_rc = -ENETUNREACH; 356 dprintk("RPC: %s: CM route resolution error, ep 0x%p\n", 357 __func__, ep); 358 complete(&ia->ri_done); 359 break; 360 case RDMA_CM_EVENT_ESTABLISHED: 361 connstate = 1; 362 ib_query_qp(ia->ri_id->qp, attr, 363 IB_QP_MAX_QP_RD_ATOMIC | IB_QP_MAX_DEST_RD_ATOMIC, 364 iattr); 365 dprintk("RPC: %s: %d responder resources" 366 " (%d initiator)\n", 367 __func__, attr->max_dest_rd_atomic, 368 attr->max_rd_atomic); 369 goto connected; 370 case RDMA_CM_EVENT_CONNECT_ERROR: 371 connstate = -ENOTCONN; 372 goto connected; 373 case RDMA_CM_EVENT_UNREACHABLE: 374 connstate = -ENETDOWN; 375 goto connected; 376 case RDMA_CM_EVENT_REJECTED: 377 connstate = -ECONNREFUSED; 378 goto connected; 379 case RDMA_CM_EVENT_DISCONNECTED: 380 connstate = -ECONNABORTED; 381 goto connected; 382 case RDMA_CM_EVENT_DEVICE_REMOVAL: 383 connstate = -ENODEV; 384 connected: 385 dprintk("RPC: %s: %sconnected\n", 386 __func__, connstate > 0 ? "" : "dis"); 387 ep->rep_connected = connstate; 388 rpcrdma_conn_func(ep); 389 wake_up_all(&ep->rep_connect_wait); 390 /*FALLTHROUGH*/ 391 default: 392 dprintk("RPC: %s: %pIS:%u (ep 0x%p): %s\n", 393 __func__, sap, rpc_get_port(sap), ep, 394 rdma_event_msg(event->event)); 395 break; 396 } 397 398 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG) 399 if (connstate == 1) { 400 int ird = attr->max_dest_rd_atomic; 401 int tird = ep->rep_remote_cma.responder_resources; 402 403 pr_info("rpcrdma: connection to %pIS:%u on %s, memreg '%s', %d credits, %d responders%s\n", 404 sap, rpc_get_port(sap), 405 ia->ri_device->name, 406 ia->ri_ops->ro_displayname, 407 xprt->rx_buf.rb_max_requests, 408 ird, ird < 4 && ird < tird / 2 ? " (low!)" : ""); 409 } else if (connstate < 0) { 410 pr_info("rpcrdma: connection to %pIS:%u closed (%d)\n", 411 sap, rpc_get_port(sap), connstate); 412 } 413 #endif 414 415 return 0; 416 } 417 418 static void rpcrdma_destroy_id(struct rdma_cm_id *id) 419 { 420 if (id) { 421 module_put(id->device->owner); 422 rdma_destroy_id(id); 423 } 424 } 425 426 static struct rdma_cm_id * 427 rpcrdma_create_id(struct rpcrdma_xprt *xprt, 428 struct rpcrdma_ia *ia, struct sockaddr *addr) 429 { 430 struct rdma_cm_id *id; 431 int rc; 432 433 init_completion(&ia->ri_done); 434 435 id = rdma_create_id(rpcrdma_conn_upcall, xprt, RDMA_PS_TCP, IB_QPT_RC); 436 if (IS_ERR(id)) { 437 rc = PTR_ERR(id); 438 dprintk("RPC: %s: rdma_create_id() failed %i\n", 439 __func__, rc); 440 return id; 441 } 442 443 ia->ri_async_rc = -ETIMEDOUT; 444 rc = rdma_resolve_addr(id, NULL, addr, RDMA_RESOLVE_TIMEOUT); 445 if (rc) { 446 dprintk("RPC: %s: rdma_resolve_addr() failed %i\n", 447 __func__, rc); 448 goto out; 449 } 450 wait_for_completion_interruptible_timeout(&ia->ri_done, 451 msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1); 452 453 /* FIXME: 454 * Until xprtrdma supports DEVICE_REMOVAL, the provider must 455 * be pinned while there are active NFS/RDMA mounts to prevent 456 * hangs and crashes at umount time. 457 */ 458 if (!ia->ri_async_rc && !try_module_get(id->device->owner)) { 459 dprintk("RPC: %s: Failed to get device module\n", 460 __func__); 461 ia->ri_async_rc = -ENODEV; 462 } 463 rc = ia->ri_async_rc; 464 if (rc) 465 goto out; 466 467 ia->ri_async_rc = -ETIMEDOUT; 468 rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT); 469 if (rc) { 470 dprintk("RPC: %s: rdma_resolve_route() failed %i\n", 471 __func__, rc); 472 goto put; 473 } 474 wait_for_completion_interruptible_timeout(&ia->ri_done, 475 msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1); 476 rc = ia->ri_async_rc; 477 if (rc) 478 goto put; 479 480 return id; 481 put: 482 module_put(id->device->owner); 483 out: 484 rdma_destroy_id(id); 485 return ERR_PTR(rc); 486 } 487 488 /* 489 * Drain any cq, prior to teardown. 490 */ 491 static void 492 rpcrdma_clean_cq(struct ib_cq *cq) 493 { 494 struct ib_wc wc; 495 int count = 0; 496 497 while (1 == ib_poll_cq(cq, 1, &wc)) 498 ++count; 499 500 if (count) 501 dprintk("RPC: %s: flushed %d events (last 0x%x)\n", 502 __func__, count, wc.opcode); 503 } 504 505 /* 506 * Exported functions. 507 */ 508 509 /* 510 * Open and initialize an Interface Adapter. 511 * o initializes fields of struct rpcrdma_ia, including 512 * interface and provider attributes and protection zone. 513 */ 514 int 515 rpcrdma_ia_open(struct rpcrdma_xprt *xprt, struct sockaddr *addr, int memreg) 516 { 517 struct rpcrdma_ia *ia = &xprt->rx_ia; 518 struct ib_device_attr *devattr = &ia->ri_devattr; 519 int rc; 520 521 ia->ri_dma_mr = NULL; 522 523 ia->ri_id = rpcrdma_create_id(xprt, ia, addr); 524 if (IS_ERR(ia->ri_id)) { 525 rc = PTR_ERR(ia->ri_id); 526 goto out1; 527 } 528 ia->ri_device = ia->ri_id->device; 529 530 ia->ri_pd = ib_alloc_pd(ia->ri_device); 531 if (IS_ERR(ia->ri_pd)) { 532 rc = PTR_ERR(ia->ri_pd); 533 dprintk("RPC: %s: ib_alloc_pd() failed %i\n", 534 __func__, rc); 535 goto out2; 536 } 537 538 rc = ib_query_device(ia->ri_device, devattr); 539 if (rc) { 540 dprintk("RPC: %s: ib_query_device failed %d\n", 541 __func__, rc); 542 goto out3; 543 } 544 545 if (memreg == RPCRDMA_FRMR) { 546 /* Requires both frmr reg and local dma lkey */ 547 if (((devattr->device_cap_flags & 548 (IB_DEVICE_MEM_MGT_EXTENSIONS|IB_DEVICE_LOCAL_DMA_LKEY)) != 549 (IB_DEVICE_MEM_MGT_EXTENSIONS|IB_DEVICE_LOCAL_DMA_LKEY)) || 550 (devattr->max_fast_reg_page_list_len == 0)) { 551 dprintk("RPC: %s: FRMR registration " 552 "not supported by HCA\n", __func__); 553 memreg = RPCRDMA_MTHCAFMR; 554 } 555 } 556 if (memreg == RPCRDMA_MTHCAFMR) { 557 if (!ia->ri_device->alloc_fmr) { 558 dprintk("RPC: %s: MTHCAFMR registration " 559 "not supported by HCA\n", __func__); 560 goto out3; 561 } 562 } 563 564 switch (memreg) { 565 case RPCRDMA_FRMR: 566 ia->ri_ops = &rpcrdma_frwr_memreg_ops; 567 break; 568 case RPCRDMA_ALLPHYSICAL: 569 ia->ri_ops = &rpcrdma_physical_memreg_ops; 570 break; 571 case RPCRDMA_MTHCAFMR: 572 ia->ri_ops = &rpcrdma_fmr_memreg_ops; 573 break; 574 default: 575 printk(KERN_ERR "RPC: Unsupported memory " 576 "registration mode: %d\n", memreg); 577 rc = -ENOMEM; 578 goto out3; 579 } 580 dprintk("RPC: %s: memory registration strategy is '%s'\n", 581 __func__, ia->ri_ops->ro_displayname); 582 583 rwlock_init(&ia->ri_qplock); 584 return 0; 585 586 out3: 587 ib_dealloc_pd(ia->ri_pd); 588 ia->ri_pd = NULL; 589 out2: 590 rpcrdma_destroy_id(ia->ri_id); 591 ia->ri_id = NULL; 592 out1: 593 return rc; 594 } 595 596 /* 597 * Clean up/close an IA. 598 * o if event handles and PD have been initialized, free them. 599 * o close the IA 600 */ 601 void 602 rpcrdma_ia_close(struct rpcrdma_ia *ia) 603 { 604 dprintk("RPC: %s: entering\n", __func__); 605 if (ia->ri_id != NULL && !IS_ERR(ia->ri_id)) { 606 if (ia->ri_id->qp) 607 rdma_destroy_qp(ia->ri_id); 608 rpcrdma_destroy_id(ia->ri_id); 609 ia->ri_id = NULL; 610 } 611 612 /* If the pd is still busy, xprtrdma missed freeing a resource */ 613 if (ia->ri_pd && !IS_ERR(ia->ri_pd)) 614 ib_dealloc_pd(ia->ri_pd); 615 } 616 617 /* 618 * Create unconnected endpoint. 619 */ 620 int 621 rpcrdma_ep_create(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia, 622 struct rpcrdma_create_data_internal *cdata) 623 { 624 struct ib_device_attr *devattr = &ia->ri_devattr; 625 struct ib_cq *sendcq, *recvcq; 626 struct ib_cq_init_attr cq_attr = {}; 627 int rc, err; 628 629 if (devattr->max_sge < RPCRDMA_MAX_IOVS) { 630 dprintk("RPC: %s: insufficient sge's available\n", 631 __func__); 632 return -ENOMEM; 633 } 634 635 /* check provider's send/recv wr limits */ 636 if (cdata->max_requests > devattr->max_qp_wr) 637 cdata->max_requests = devattr->max_qp_wr; 638 639 ep->rep_attr.event_handler = rpcrdma_qp_async_error_upcall; 640 ep->rep_attr.qp_context = ep; 641 ep->rep_attr.srq = NULL; 642 ep->rep_attr.cap.max_send_wr = cdata->max_requests; 643 rc = ia->ri_ops->ro_open(ia, ep, cdata); 644 if (rc) 645 return rc; 646 ep->rep_attr.cap.max_recv_wr = cdata->max_requests; 647 ep->rep_attr.cap.max_send_sge = RPCRDMA_MAX_IOVS; 648 ep->rep_attr.cap.max_recv_sge = 1; 649 ep->rep_attr.cap.max_inline_data = 0; 650 ep->rep_attr.sq_sig_type = IB_SIGNAL_REQ_WR; 651 ep->rep_attr.qp_type = IB_QPT_RC; 652 ep->rep_attr.port_num = ~0; 653 654 dprintk("RPC: %s: requested max: dtos: send %d recv %d; " 655 "iovs: send %d recv %d\n", 656 __func__, 657 ep->rep_attr.cap.max_send_wr, 658 ep->rep_attr.cap.max_recv_wr, 659 ep->rep_attr.cap.max_send_sge, 660 ep->rep_attr.cap.max_recv_sge); 661 662 /* set trigger for requesting send completion */ 663 ep->rep_cqinit = ep->rep_attr.cap.max_send_wr/2 - 1; 664 if (ep->rep_cqinit > RPCRDMA_MAX_UNSIGNALED_SENDS) 665 ep->rep_cqinit = RPCRDMA_MAX_UNSIGNALED_SENDS; 666 else if (ep->rep_cqinit <= 2) 667 ep->rep_cqinit = 0; 668 INIT_CQCOUNT(ep); 669 init_waitqueue_head(&ep->rep_connect_wait); 670 INIT_DELAYED_WORK(&ep->rep_connect_worker, rpcrdma_connect_worker); 671 672 cq_attr.cqe = ep->rep_attr.cap.max_send_wr + 1; 673 sendcq = ib_create_cq(ia->ri_device, rpcrdma_sendcq_upcall, 674 rpcrdma_cq_async_error_upcall, ep, &cq_attr); 675 if (IS_ERR(sendcq)) { 676 rc = PTR_ERR(sendcq); 677 dprintk("RPC: %s: failed to create send CQ: %i\n", 678 __func__, rc); 679 goto out1; 680 } 681 682 rc = ib_req_notify_cq(sendcq, IB_CQ_NEXT_COMP); 683 if (rc) { 684 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n", 685 __func__, rc); 686 goto out2; 687 } 688 689 cq_attr.cqe = ep->rep_attr.cap.max_recv_wr + 1; 690 recvcq = ib_create_cq(ia->ri_device, rpcrdma_recvcq_upcall, 691 rpcrdma_cq_async_error_upcall, ep, &cq_attr); 692 if (IS_ERR(recvcq)) { 693 rc = PTR_ERR(recvcq); 694 dprintk("RPC: %s: failed to create recv CQ: %i\n", 695 __func__, rc); 696 goto out2; 697 } 698 699 rc = ib_req_notify_cq(recvcq, IB_CQ_NEXT_COMP); 700 if (rc) { 701 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n", 702 __func__, rc); 703 ib_destroy_cq(recvcq); 704 goto out2; 705 } 706 707 ep->rep_attr.send_cq = sendcq; 708 ep->rep_attr.recv_cq = recvcq; 709 710 /* Initialize cma parameters */ 711 712 /* RPC/RDMA does not use private data */ 713 ep->rep_remote_cma.private_data = NULL; 714 ep->rep_remote_cma.private_data_len = 0; 715 716 /* Client offers RDMA Read but does not initiate */ 717 ep->rep_remote_cma.initiator_depth = 0; 718 if (devattr->max_qp_rd_atom > 32) /* arbitrary but <= 255 */ 719 ep->rep_remote_cma.responder_resources = 32; 720 else 721 ep->rep_remote_cma.responder_resources = 722 devattr->max_qp_rd_atom; 723 724 ep->rep_remote_cma.retry_count = 7; 725 ep->rep_remote_cma.flow_control = 0; 726 ep->rep_remote_cma.rnr_retry_count = 0; 727 728 return 0; 729 730 out2: 731 err = ib_destroy_cq(sendcq); 732 if (err) 733 dprintk("RPC: %s: ib_destroy_cq returned %i\n", 734 __func__, err); 735 out1: 736 if (ia->ri_dma_mr) 737 ib_dereg_mr(ia->ri_dma_mr); 738 return rc; 739 } 740 741 /* 742 * rpcrdma_ep_destroy 743 * 744 * Disconnect and destroy endpoint. After this, the only 745 * valid operations on the ep are to free it (if dynamically 746 * allocated) or re-create it. 747 */ 748 void 749 rpcrdma_ep_destroy(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 750 { 751 int rc; 752 753 dprintk("RPC: %s: entering, connected is %d\n", 754 __func__, ep->rep_connected); 755 756 cancel_delayed_work_sync(&ep->rep_connect_worker); 757 758 if (ia->ri_id->qp) { 759 rpcrdma_ep_disconnect(ep, ia); 760 rdma_destroy_qp(ia->ri_id); 761 ia->ri_id->qp = NULL; 762 } 763 764 rpcrdma_clean_cq(ep->rep_attr.recv_cq); 765 rc = ib_destroy_cq(ep->rep_attr.recv_cq); 766 if (rc) 767 dprintk("RPC: %s: ib_destroy_cq returned %i\n", 768 __func__, rc); 769 770 rpcrdma_clean_cq(ep->rep_attr.send_cq); 771 rc = ib_destroy_cq(ep->rep_attr.send_cq); 772 if (rc) 773 dprintk("RPC: %s: ib_destroy_cq returned %i\n", 774 __func__, rc); 775 776 if (ia->ri_dma_mr) { 777 rc = ib_dereg_mr(ia->ri_dma_mr); 778 dprintk("RPC: %s: ib_dereg_mr returned %i\n", 779 __func__, rc); 780 } 781 } 782 783 /* 784 * Connect unconnected endpoint. 785 */ 786 int 787 rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 788 { 789 struct rdma_cm_id *id, *old; 790 int rc = 0; 791 int retry_count = 0; 792 793 if (ep->rep_connected != 0) { 794 struct rpcrdma_xprt *xprt; 795 retry: 796 dprintk("RPC: %s: reconnecting...\n", __func__); 797 798 rpcrdma_ep_disconnect(ep, ia); 799 rpcrdma_flush_cqs(ep); 800 801 xprt = container_of(ia, struct rpcrdma_xprt, rx_ia); 802 id = rpcrdma_create_id(xprt, ia, 803 (struct sockaddr *)&xprt->rx_data.addr); 804 if (IS_ERR(id)) { 805 rc = -EHOSTUNREACH; 806 goto out; 807 } 808 /* TEMP TEMP TEMP - fail if new device: 809 * Deregister/remarshal *all* requests! 810 * Close and recreate adapter, pd, etc! 811 * Re-determine all attributes still sane! 812 * More stuff I haven't thought of! 813 * Rrrgh! 814 */ 815 if (ia->ri_device != id->device) { 816 printk("RPC: %s: can't reconnect on " 817 "different device!\n", __func__); 818 rpcrdma_destroy_id(id); 819 rc = -ENETUNREACH; 820 goto out; 821 } 822 /* END TEMP */ 823 rc = rdma_create_qp(id, ia->ri_pd, &ep->rep_attr); 824 if (rc) { 825 dprintk("RPC: %s: rdma_create_qp failed %i\n", 826 __func__, rc); 827 rpcrdma_destroy_id(id); 828 rc = -ENETUNREACH; 829 goto out; 830 } 831 832 write_lock(&ia->ri_qplock); 833 old = ia->ri_id; 834 ia->ri_id = id; 835 write_unlock(&ia->ri_qplock); 836 837 rdma_destroy_qp(old); 838 rpcrdma_destroy_id(old); 839 } else { 840 dprintk("RPC: %s: connecting...\n", __func__); 841 rc = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr); 842 if (rc) { 843 dprintk("RPC: %s: rdma_create_qp failed %i\n", 844 __func__, rc); 845 /* do not update ep->rep_connected */ 846 return -ENETUNREACH; 847 } 848 } 849 850 ep->rep_connected = 0; 851 852 rc = rdma_connect(ia->ri_id, &ep->rep_remote_cma); 853 if (rc) { 854 dprintk("RPC: %s: rdma_connect() failed with %i\n", 855 __func__, rc); 856 goto out; 857 } 858 859 wait_event_interruptible(ep->rep_connect_wait, ep->rep_connected != 0); 860 861 /* 862 * Check state. A non-peer reject indicates no listener 863 * (ECONNREFUSED), which may be a transient state. All 864 * others indicate a transport condition which has already 865 * undergone a best-effort. 866 */ 867 if (ep->rep_connected == -ECONNREFUSED && 868 ++retry_count <= RDMA_CONNECT_RETRY_MAX) { 869 dprintk("RPC: %s: non-peer_reject, retry\n", __func__); 870 goto retry; 871 } 872 if (ep->rep_connected <= 0) { 873 /* Sometimes, the only way to reliably connect to remote 874 * CMs is to use same nonzero values for ORD and IRD. */ 875 if (retry_count++ <= RDMA_CONNECT_RETRY_MAX + 1 && 876 (ep->rep_remote_cma.responder_resources == 0 || 877 ep->rep_remote_cma.initiator_depth != 878 ep->rep_remote_cma.responder_resources)) { 879 if (ep->rep_remote_cma.responder_resources == 0) 880 ep->rep_remote_cma.responder_resources = 1; 881 ep->rep_remote_cma.initiator_depth = 882 ep->rep_remote_cma.responder_resources; 883 goto retry; 884 } 885 rc = ep->rep_connected; 886 } else { 887 dprintk("RPC: %s: connected\n", __func__); 888 } 889 890 out: 891 if (rc) 892 ep->rep_connected = rc; 893 return rc; 894 } 895 896 /* 897 * rpcrdma_ep_disconnect 898 * 899 * This is separate from destroy to facilitate the ability 900 * to reconnect without recreating the endpoint. 901 * 902 * This call is not reentrant, and must not be made in parallel 903 * on the same endpoint. 904 */ 905 void 906 rpcrdma_ep_disconnect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 907 { 908 int rc; 909 910 rpcrdma_flush_cqs(ep); 911 rc = rdma_disconnect(ia->ri_id); 912 if (!rc) { 913 /* returns without wait if not connected */ 914 wait_event_interruptible(ep->rep_connect_wait, 915 ep->rep_connected != 1); 916 dprintk("RPC: %s: after wait, %sconnected\n", __func__, 917 (ep->rep_connected == 1) ? "still " : "dis"); 918 } else { 919 dprintk("RPC: %s: rdma_disconnect %i\n", __func__, rc); 920 ep->rep_connected = rc; 921 } 922 } 923 924 static struct rpcrdma_req * 925 rpcrdma_create_req(struct rpcrdma_xprt *r_xprt) 926 { 927 struct rpcrdma_req *req; 928 929 req = kzalloc(sizeof(*req), GFP_KERNEL); 930 if (req == NULL) 931 return ERR_PTR(-ENOMEM); 932 933 req->rl_buffer = &r_xprt->rx_buf; 934 return req; 935 } 936 937 static struct rpcrdma_rep * 938 rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt) 939 { 940 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 941 struct rpcrdma_ia *ia = &r_xprt->rx_ia; 942 struct rpcrdma_rep *rep; 943 int rc; 944 945 rc = -ENOMEM; 946 rep = kzalloc(sizeof(*rep), GFP_KERNEL); 947 if (rep == NULL) 948 goto out; 949 950 rep->rr_rdmabuf = rpcrdma_alloc_regbuf(ia, cdata->inline_rsize, 951 GFP_KERNEL); 952 if (IS_ERR(rep->rr_rdmabuf)) { 953 rc = PTR_ERR(rep->rr_rdmabuf); 954 goto out_free; 955 } 956 957 rep->rr_device = ia->ri_device; 958 rep->rr_rxprt = r_xprt; 959 return rep; 960 961 out_free: 962 kfree(rep); 963 out: 964 return ERR_PTR(rc); 965 } 966 967 int 968 rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt) 969 { 970 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 971 struct rpcrdma_ia *ia = &r_xprt->rx_ia; 972 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 973 char *p; 974 size_t len; 975 int i, rc; 976 977 buf->rb_max_requests = cdata->max_requests; 978 spin_lock_init(&buf->rb_lock); 979 980 /* Need to allocate: 981 * 1. arrays for send and recv pointers 982 * 2. arrays of struct rpcrdma_req to fill in pointers 983 * 3. array of struct rpcrdma_rep for replies 984 * Send/recv buffers in req/rep need to be registered 985 */ 986 len = buf->rb_max_requests * 987 (sizeof(struct rpcrdma_req *) + sizeof(struct rpcrdma_rep *)); 988 989 p = kzalloc(len, GFP_KERNEL); 990 if (p == NULL) { 991 dprintk("RPC: %s: req_t/rep_t/pad kzalloc(%zd) failed\n", 992 __func__, len); 993 rc = -ENOMEM; 994 goto out; 995 } 996 buf->rb_pool = p; /* for freeing it later */ 997 998 buf->rb_send_bufs = (struct rpcrdma_req **) p; 999 p = (char *) &buf->rb_send_bufs[buf->rb_max_requests]; 1000 buf->rb_recv_bufs = (struct rpcrdma_rep **) p; 1001 p = (char *) &buf->rb_recv_bufs[buf->rb_max_requests]; 1002 1003 rc = ia->ri_ops->ro_init(r_xprt); 1004 if (rc) 1005 goto out; 1006 1007 for (i = 0; i < buf->rb_max_requests; i++) { 1008 struct rpcrdma_req *req; 1009 struct rpcrdma_rep *rep; 1010 1011 req = rpcrdma_create_req(r_xprt); 1012 if (IS_ERR(req)) { 1013 dprintk("RPC: %s: request buffer %d alloc" 1014 " failed\n", __func__, i); 1015 rc = PTR_ERR(req); 1016 goto out; 1017 } 1018 buf->rb_send_bufs[i] = req; 1019 1020 rep = rpcrdma_create_rep(r_xprt); 1021 if (IS_ERR(rep)) { 1022 dprintk("RPC: %s: reply buffer %d alloc failed\n", 1023 __func__, i); 1024 rc = PTR_ERR(rep); 1025 goto out; 1026 } 1027 buf->rb_recv_bufs[i] = rep; 1028 } 1029 1030 return 0; 1031 out: 1032 rpcrdma_buffer_destroy(buf); 1033 return rc; 1034 } 1035 1036 static void 1037 rpcrdma_destroy_rep(struct rpcrdma_ia *ia, struct rpcrdma_rep *rep) 1038 { 1039 if (!rep) 1040 return; 1041 1042 rpcrdma_free_regbuf(ia, rep->rr_rdmabuf); 1043 kfree(rep); 1044 } 1045 1046 static void 1047 rpcrdma_destroy_req(struct rpcrdma_ia *ia, struct rpcrdma_req *req) 1048 { 1049 if (!req) 1050 return; 1051 1052 rpcrdma_free_regbuf(ia, req->rl_sendbuf); 1053 rpcrdma_free_regbuf(ia, req->rl_rdmabuf); 1054 kfree(req); 1055 } 1056 1057 void 1058 rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf) 1059 { 1060 struct rpcrdma_ia *ia = rdmab_to_ia(buf); 1061 int i; 1062 1063 /* clean up in reverse order from create 1064 * 1. recv mr memory (mr free, then kfree) 1065 * 2. send mr memory (mr free, then kfree) 1066 * 3. MWs 1067 */ 1068 dprintk("RPC: %s: entering\n", __func__); 1069 1070 for (i = 0; i < buf->rb_max_requests; i++) { 1071 if (buf->rb_recv_bufs) 1072 rpcrdma_destroy_rep(ia, buf->rb_recv_bufs[i]); 1073 if (buf->rb_send_bufs) 1074 rpcrdma_destroy_req(ia, buf->rb_send_bufs[i]); 1075 } 1076 1077 ia->ri_ops->ro_destroy(buf); 1078 1079 kfree(buf->rb_pool); 1080 } 1081 1082 struct rpcrdma_mw * 1083 rpcrdma_get_mw(struct rpcrdma_xprt *r_xprt) 1084 { 1085 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1086 struct rpcrdma_mw *mw = NULL; 1087 1088 spin_lock(&buf->rb_mwlock); 1089 if (!list_empty(&buf->rb_mws)) { 1090 mw = list_first_entry(&buf->rb_mws, 1091 struct rpcrdma_mw, mw_list); 1092 list_del_init(&mw->mw_list); 1093 } 1094 spin_unlock(&buf->rb_mwlock); 1095 1096 if (!mw) 1097 pr_err("RPC: %s: no MWs available\n", __func__); 1098 return mw; 1099 } 1100 1101 void 1102 rpcrdma_put_mw(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mw *mw) 1103 { 1104 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1105 1106 spin_lock(&buf->rb_mwlock); 1107 list_add_tail(&mw->mw_list, &buf->rb_mws); 1108 spin_unlock(&buf->rb_mwlock); 1109 } 1110 1111 static void 1112 rpcrdma_buffer_put_sendbuf(struct rpcrdma_req *req, struct rpcrdma_buffer *buf) 1113 { 1114 buf->rb_send_bufs[--buf->rb_send_index] = req; 1115 req->rl_niovs = 0; 1116 if (req->rl_reply) { 1117 buf->rb_recv_bufs[--buf->rb_recv_index] = req->rl_reply; 1118 req->rl_reply = NULL; 1119 } 1120 } 1121 1122 /* 1123 * Get a set of request/reply buffers. 1124 * 1125 * Reply buffer (if needed) is attached to send buffer upon return. 1126 * Rule: 1127 * rb_send_index and rb_recv_index MUST always be pointing to the 1128 * *next* available buffer (non-NULL). They are incremented after 1129 * removing buffers, and decremented *before* returning them. 1130 */ 1131 struct rpcrdma_req * 1132 rpcrdma_buffer_get(struct rpcrdma_buffer *buffers) 1133 { 1134 struct rpcrdma_req *req; 1135 unsigned long flags; 1136 1137 spin_lock_irqsave(&buffers->rb_lock, flags); 1138 1139 if (buffers->rb_send_index == buffers->rb_max_requests) { 1140 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1141 dprintk("RPC: %s: out of request buffers\n", __func__); 1142 return ((struct rpcrdma_req *)NULL); 1143 } 1144 1145 req = buffers->rb_send_bufs[buffers->rb_send_index]; 1146 if (buffers->rb_send_index < buffers->rb_recv_index) { 1147 dprintk("RPC: %s: %d extra receives outstanding (ok)\n", 1148 __func__, 1149 buffers->rb_recv_index - buffers->rb_send_index); 1150 req->rl_reply = NULL; 1151 } else { 1152 req->rl_reply = buffers->rb_recv_bufs[buffers->rb_recv_index]; 1153 buffers->rb_recv_bufs[buffers->rb_recv_index++] = NULL; 1154 } 1155 buffers->rb_send_bufs[buffers->rb_send_index++] = NULL; 1156 1157 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1158 return req; 1159 } 1160 1161 /* 1162 * Put request/reply buffers back into pool. 1163 * Pre-decrement counter/array index. 1164 */ 1165 void 1166 rpcrdma_buffer_put(struct rpcrdma_req *req) 1167 { 1168 struct rpcrdma_buffer *buffers = req->rl_buffer; 1169 unsigned long flags; 1170 1171 spin_lock_irqsave(&buffers->rb_lock, flags); 1172 rpcrdma_buffer_put_sendbuf(req, buffers); 1173 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1174 } 1175 1176 /* 1177 * Recover reply buffers from pool. 1178 * This happens when recovering from error conditions. 1179 * Post-increment counter/array index. 1180 */ 1181 void 1182 rpcrdma_recv_buffer_get(struct rpcrdma_req *req) 1183 { 1184 struct rpcrdma_buffer *buffers = req->rl_buffer; 1185 unsigned long flags; 1186 1187 spin_lock_irqsave(&buffers->rb_lock, flags); 1188 if (buffers->rb_recv_index < buffers->rb_max_requests) { 1189 req->rl_reply = buffers->rb_recv_bufs[buffers->rb_recv_index]; 1190 buffers->rb_recv_bufs[buffers->rb_recv_index++] = NULL; 1191 } 1192 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1193 } 1194 1195 /* 1196 * Put reply buffers back into pool when not attached to 1197 * request. This happens in error conditions. 1198 */ 1199 void 1200 rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep) 1201 { 1202 struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf; 1203 unsigned long flags; 1204 1205 spin_lock_irqsave(&buffers->rb_lock, flags); 1206 buffers->rb_recv_bufs[--buffers->rb_recv_index] = rep; 1207 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1208 } 1209 1210 /* 1211 * Wrappers for internal-use kmalloc memory registration, used by buffer code. 1212 */ 1213 1214 void 1215 rpcrdma_mapping_error(struct rpcrdma_mr_seg *seg) 1216 { 1217 dprintk("RPC: map_one: offset %p iova %llx len %zu\n", 1218 seg->mr_offset, 1219 (unsigned long long)seg->mr_dma, seg->mr_dmalen); 1220 } 1221 1222 /** 1223 * rpcrdma_alloc_regbuf - kmalloc and register memory for SEND/RECV buffers 1224 * @ia: controlling rpcrdma_ia 1225 * @size: size of buffer to be allocated, in bytes 1226 * @flags: GFP flags 1227 * 1228 * Returns pointer to private header of an area of internally 1229 * registered memory, or an ERR_PTR. The registered buffer follows 1230 * the end of the private header. 1231 * 1232 * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for 1233 * receiving the payload of RDMA RECV operations. regbufs are not 1234 * used for RDMA READ/WRITE operations, thus are registered only for 1235 * LOCAL access. 1236 */ 1237 struct rpcrdma_regbuf * 1238 rpcrdma_alloc_regbuf(struct rpcrdma_ia *ia, size_t size, gfp_t flags) 1239 { 1240 struct rpcrdma_regbuf *rb; 1241 struct ib_sge *iov; 1242 1243 rb = kmalloc(sizeof(*rb) + size, flags); 1244 if (rb == NULL) 1245 goto out; 1246 1247 iov = &rb->rg_iov; 1248 iov->addr = ib_dma_map_single(ia->ri_device, 1249 (void *)rb->rg_base, size, 1250 DMA_BIDIRECTIONAL); 1251 if (ib_dma_mapping_error(ia->ri_device, iov->addr)) 1252 goto out_free; 1253 1254 iov->length = size; 1255 iov->lkey = ia->ri_pd->local_dma_lkey; 1256 rb->rg_size = size; 1257 rb->rg_owner = NULL; 1258 return rb; 1259 1260 out_free: 1261 kfree(rb); 1262 out: 1263 return ERR_PTR(-ENOMEM); 1264 } 1265 1266 /** 1267 * rpcrdma_free_regbuf - deregister and free registered buffer 1268 * @ia: controlling rpcrdma_ia 1269 * @rb: regbuf to be deregistered and freed 1270 */ 1271 void 1272 rpcrdma_free_regbuf(struct rpcrdma_ia *ia, struct rpcrdma_regbuf *rb) 1273 { 1274 struct ib_sge *iov; 1275 1276 if (!rb) 1277 return; 1278 1279 iov = &rb->rg_iov; 1280 ib_dma_unmap_single(ia->ri_device, 1281 iov->addr, iov->length, DMA_BIDIRECTIONAL); 1282 kfree(rb); 1283 } 1284 1285 /* 1286 * Prepost any receive buffer, then post send. 1287 * 1288 * Receive buffer is donated to hardware, reclaimed upon recv completion. 1289 */ 1290 int 1291 rpcrdma_ep_post(struct rpcrdma_ia *ia, 1292 struct rpcrdma_ep *ep, 1293 struct rpcrdma_req *req) 1294 { 1295 struct ib_device *device = ia->ri_device; 1296 struct ib_send_wr send_wr, *send_wr_fail; 1297 struct rpcrdma_rep *rep = req->rl_reply; 1298 struct ib_sge *iov = req->rl_send_iov; 1299 int i, rc; 1300 1301 if (rep) { 1302 rc = rpcrdma_ep_post_recv(ia, ep, rep); 1303 if (rc) 1304 goto out; 1305 req->rl_reply = NULL; 1306 } 1307 1308 send_wr.next = NULL; 1309 send_wr.wr_id = RPCRDMA_IGNORE_COMPLETION; 1310 send_wr.sg_list = iov; 1311 send_wr.num_sge = req->rl_niovs; 1312 send_wr.opcode = IB_WR_SEND; 1313 1314 for (i = 0; i < send_wr.num_sge; i++) 1315 ib_dma_sync_single_for_device(device, iov[i].addr, 1316 iov[i].length, DMA_TO_DEVICE); 1317 dprintk("RPC: %s: posting %d s/g entries\n", 1318 __func__, send_wr.num_sge); 1319 1320 if (DECR_CQCOUNT(ep) > 0) 1321 send_wr.send_flags = 0; 1322 else { /* Provider must take a send completion every now and then */ 1323 INIT_CQCOUNT(ep); 1324 send_wr.send_flags = IB_SEND_SIGNALED; 1325 } 1326 1327 rc = ib_post_send(ia->ri_id->qp, &send_wr, &send_wr_fail); 1328 if (rc) 1329 dprintk("RPC: %s: ib_post_send returned %i\n", __func__, 1330 rc); 1331 out: 1332 return rc; 1333 } 1334 1335 /* 1336 * (Re)post a receive buffer. 1337 */ 1338 int 1339 rpcrdma_ep_post_recv(struct rpcrdma_ia *ia, 1340 struct rpcrdma_ep *ep, 1341 struct rpcrdma_rep *rep) 1342 { 1343 struct ib_recv_wr recv_wr, *recv_wr_fail; 1344 int rc; 1345 1346 recv_wr.next = NULL; 1347 recv_wr.wr_id = (u64) (unsigned long) rep; 1348 recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov; 1349 recv_wr.num_sge = 1; 1350 1351 ib_dma_sync_single_for_cpu(ia->ri_device, 1352 rdmab_addr(rep->rr_rdmabuf), 1353 rdmab_length(rep->rr_rdmabuf), 1354 DMA_BIDIRECTIONAL); 1355 1356 rc = ib_post_recv(ia->ri_id->qp, &recv_wr, &recv_wr_fail); 1357 1358 if (rc) 1359 dprintk("RPC: %s: ib_post_recv returned %i\n", __func__, 1360 rc); 1361 return rc; 1362 } 1363 1364 /* How many chunk list items fit within our inline buffers? 1365 */ 1366 unsigned int 1367 rpcrdma_max_segments(struct rpcrdma_xprt *r_xprt) 1368 { 1369 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 1370 int bytes, segments; 1371 1372 bytes = min_t(unsigned int, cdata->inline_wsize, cdata->inline_rsize); 1373 bytes -= RPCRDMA_HDRLEN_MIN; 1374 if (bytes < sizeof(struct rpcrdma_segment) * 2) { 1375 pr_warn("RPC: %s: inline threshold too small\n", 1376 __func__); 1377 return 0; 1378 } 1379 1380 segments = 1 << (fls(bytes / sizeof(struct rpcrdma_segment)) - 1); 1381 dprintk("RPC: %s: max chunk list size = %d segments\n", 1382 __func__, segments); 1383 return segments; 1384 } 1385