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 761 rpcrdma_clean_cq(ep->rep_attr.recv_cq); 762 rpcrdma_clean_cq(ep->rep_attr.send_cq); 763 764 if (ia->ri_id->qp) { 765 rdma_destroy_qp(ia->ri_id); 766 ia->ri_id->qp = NULL; 767 } 768 769 rc = ib_destroy_cq(ep->rep_attr.recv_cq); 770 if (rc) 771 dprintk("RPC: %s: ib_destroy_cq returned %i\n", 772 __func__, rc); 773 774 rc = ib_destroy_cq(ep->rep_attr.send_cq); 775 if (rc) 776 dprintk("RPC: %s: ib_destroy_cq returned %i\n", 777 __func__, rc); 778 779 if (ia->ri_dma_mr) { 780 rc = ib_dereg_mr(ia->ri_dma_mr); 781 dprintk("RPC: %s: ib_dereg_mr returned %i\n", 782 __func__, rc); 783 } 784 } 785 786 /* 787 * Connect unconnected endpoint. 788 */ 789 int 790 rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 791 { 792 struct rdma_cm_id *id, *old; 793 int rc = 0; 794 int retry_count = 0; 795 796 if (ep->rep_connected != 0) { 797 struct rpcrdma_xprt *xprt; 798 retry: 799 dprintk("RPC: %s: reconnecting...\n", __func__); 800 801 rpcrdma_ep_disconnect(ep, ia); 802 rpcrdma_flush_cqs(ep); 803 804 xprt = container_of(ia, struct rpcrdma_xprt, rx_ia); 805 id = rpcrdma_create_id(xprt, ia, 806 (struct sockaddr *)&xprt->rx_data.addr); 807 if (IS_ERR(id)) { 808 rc = -EHOSTUNREACH; 809 goto out; 810 } 811 /* TEMP TEMP TEMP - fail if new device: 812 * Deregister/remarshal *all* requests! 813 * Close and recreate adapter, pd, etc! 814 * Re-determine all attributes still sane! 815 * More stuff I haven't thought of! 816 * Rrrgh! 817 */ 818 if (ia->ri_device != id->device) { 819 printk("RPC: %s: can't reconnect on " 820 "different device!\n", __func__); 821 rpcrdma_destroy_id(id); 822 rc = -ENETUNREACH; 823 goto out; 824 } 825 /* END TEMP */ 826 rc = rdma_create_qp(id, ia->ri_pd, &ep->rep_attr); 827 if (rc) { 828 dprintk("RPC: %s: rdma_create_qp failed %i\n", 829 __func__, rc); 830 rpcrdma_destroy_id(id); 831 rc = -ENETUNREACH; 832 goto out; 833 } 834 835 write_lock(&ia->ri_qplock); 836 old = ia->ri_id; 837 ia->ri_id = id; 838 write_unlock(&ia->ri_qplock); 839 840 rdma_destroy_qp(old); 841 rpcrdma_destroy_id(old); 842 } else { 843 dprintk("RPC: %s: connecting...\n", __func__); 844 rc = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr); 845 if (rc) { 846 dprintk("RPC: %s: rdma_create_qp failed %i\n", 847 __func__, rc); 848 /* do not update ep->rep_connected */ 849 return -ENETUNREACH; 850 } 851 } 852 853 ep->rep_connected = 0; 854 855 rc = rdma_connect(ia->ri_id, &ep->rep_remote_cma); 856 if (rc) { 857 dprintk("RPC: %s: rdma_connect() failed with %i\n", 858 __func__, rc); 859 goto out; 860 } 861 862 wait_event_interruptible(ep->rep_connect_wait, ep->rep_connected != 0); 863 864 /* 865 * Check state. A non-peer reject indicates no listener 866 * (ECONNREFUSED), which may be a transient state. All 867 * others indicate a transport condition which has already 868 * undergone a best-effort. 869 */ 870 if (ep->rep_connected == -ECONNREFUSED && 871 ++retry_count <= RDMA_CONNECT_RETRY_MAX) { 872 dprintk("RPC: %s: non-peer_reject, retry\n", __func__); 873 goto retry; 874 } 875 if (ep->rep_connected <= 0) { 876 /* Sometimes, the only way to reliably connect to remote 877 * CMs is to use same nonzero values for ORD and IRD. */ 878 if (retry_count++ <= RDMA_CONNECT_RETRY_MAX + 1 && 879 (ep->rep_remote_cma.responder_resources == 0 || 880 ep->rep_remote_cma.initiator_depth != 881 ep->rep_remote_cma.responder_resources)) { 882 if (ep->rep_remote_cma.responder_resources == 0) 883 ep->rep_remote_cma.responder_resources = 1; 884 ep->rep_remote_cma.initiator_depth = 885 ep->rep_remote_cma.responder_resources; 886 goto retry; 887 } 888 rc = ep->rep_connected; 889 } else { 890 dprintk("RPC: %s: connected\n", __func__); 891 } 892 893 out: 894 if (rc) 895 ep->rep_connected = rc; 896 return rc; 897 } 898 899 /* 900 * rpcrdma_ep_disconnect 901 * 902 * This is separate from destroy to facilitate the ability 903 * to reconnect without recreating the endpoint. 904 * 905 * This call is not reentrant, and must not be made in parallel 906 * on the same endpoint. 907 */ 908 void 909 rpcrdma_ep_disconnect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 910 { 911 int rc; 912 913 rpcrdma_flush_cqs(ep); 914 rc = rdma_disconnect(ia->ri_id); 915 if (!rc) { 916 /* returns without wait if not connected */ 917 wait_event_interruptible(ep->rep_connect_wait, 918 ep->rep_connected != 1); 919 dprintk("RPC: %s: after wait, %sconnected\n", __func__, 920 (ep->rep_connected == 1) ? "still " : "dis"); 921 } else { 922 dprintk("RPC: %s: rdma_disconnect %i\n", __func__, rc); 923 ep->rep_connected = rc; 924 } 925 } 926 927 static struct rpcrdma_req * 928 rpcrdma_create_req(struct rpcrdma_xprt *r_xprt) 929 { 930 struct rpcrdma_req *req; 931 932 req = kzalloc(sizeof(*req), GFP_KERNEL); 933 if (req == NULL) 934 return ERR_PTR(-ENOMEM); 935 936 req->rl_buffer = &r_xprt->rx_buf; 937 return req; 938 } 939 940 static struct rpcrdma_rep * 941 rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt) 942 { 943 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 944 struct rpcrdma_ia *ia = &r_xprt->rx_ia; 945 struct rpcrdma_rep *rep; 946 int rc; 947 948 rc = -ENOMEM; 949 rep = kzalloc(sizeof(*rep), GFP_KERNEL); 950 if (rep == NULL) 951 goto out; 952 953 rep->rr_rdmabuf = rpcrdma_alloc_regbuf(ia, cdata->inline_rsize, 954 GFP_KERNEL); 955 if (IS_ERR(rep->rr_rdmabuf)) { 956 rc = PTR_ERR(rep->rr_rdmabuf); 957 goto out_free; 958 } 959 960 rep->rr_device = ia->ri_device; 961 rep->rr_rxprt = r_xprt; 962 return rep; 963 964 out_free: 965 kfree(rep); 966 out: 967 return ERR_PTR(rc); 968 } 969 970 int 971 rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt) 972 { 973 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 974 struct rpcrdma_ia *ia = &r_xprt->rx_ia; 975 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 976 char *p; 977 size_t len; 978 int i, rc; 979 980 buf->rb_max_requests = cdata->max_requests; 981 spin_lock_init(&buf->rb_lock); 982 983 /* Need to allocate: 984 * 1. arrays for send and recv pointers 985 * 2. arrays of struct rpcrdma_req to fill in pointers 986 * 3. array of struct rpcrdma_rep for replies 987 * Send/recv buffers in req/rep need to be registered 988 */ 989 len = buf->rb_max_requests * 990 (sizeof(struct rpcrdma_req *) + sizeof(struct rpcrdma_rep *)); 991 992 p = kzalloc(len, GFP_KERNEL); 993 if (p == NULL) { 994 dprintk("RPC: %s: req_t/rep_t/pad kzalloc(%zd) failed\n", 995 __func__, len); 996 rc = -ENOMEM; 997 goto out; 998 } 999 buf->rb_pool = p; /* for freeing it later */ 1000 1001 buf->rb_send_bufs = (struct rpcrdma_req **) p; 1002 p = (char *) &buf->rb_send_bufs[buf->rb_max_requests]; 1003 buf->rb_recv_bufs = (struct rpcrdma_rep **) p; 1004 p = (char *) &buf->rb_recv_bufs[buf->rb_max_requests]; 1005 1006 rc = ia->ri_ops->ro_init(r_xprt); 1007 if (rc) 1008 goto out; 1009 1010 for (i = 0; i < buf->rb_max_requests; i++) { 1011 struct rpcrdma_req *req; 1012 struct rpcrdma_rep *rep; 1013 1014 req = rpcrdma_create_req(r_xprt); 1015 if (IS_ERR(req)) { 1016 dprintk("RPC: %s: request buffer %d alloc" 1017 " failed\n", __func__, i); 1018 rc = PTR_ERR(req); 1019 goto out; 1020 } 1021 buf->rb_send_bufs[i] = req; 1022 1023 rep = rpcrdma_create_rep(r_xprt); 1024 if (IS_ERR(rep)) { 1025 dprintk("RPC: %s: reply buffer %d alloc failed\n", 1026 __func__, i); 1027 rc = PTR_ERR(rep); 1028 goto out; 1029 } 1030 buf->rb_recv_bufs[i] = rep; 1031 } 1032 1033 return 0; 1034 out: 1035 rpcrdma_buffer_destroy(buf); 1036 return rc; 1037 } 1038 1039 static void 1040 rpcrdma_destroy_rep(struct rpcrdma_ia *ia, struct rpcrdma_rep *rep) 1041 { 1042 if (!rep) 1043 return; 1044 1045 rpcrdma_free_regbuf(ia, rep->rr_rdmabuf); 1046 kfree(rep); 1047 } 1048 1049 static void 1050 rpcrdma_destroy_req(struct rpcrdma_ia *ia, struct rpcrdma_req *req) 1051 { 1052 if (!req) 1053 return; 1054 1055 rpcrdma_free_regbuf(ia, req->rl_sendbuf); 1056 rpcrdma_free_regbuf(ia, req->rl_rdmabuf); 1057 kfree(req); 1058 } 1059 1060 void 1061 rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf) 1062 { 1063 struct rpcrdma_ia *ia = rdmab_to_ia(buf); 1064 int i; 1065 1066 /* clean up in reverse order from create 1067 * 1. recv mr memory (mr free, then kfree) 1068 * 2. send mr memory (mr free, then kfree) 1069 * 3. MWs 1070 */ 1071 dprintk("RPC: %s: entering\n", __func__); 1072 1073 for (i = 0; i < buf->rb_max_requests; i++) { 1074 if (buf->rb_recv_bufs) 1075 rpcrdma_destroy_rep(ia, buf->rb_recv_bufs[i]); 1076 if (buf->rb_send_bufs) 1077 rpcrdma_destroy_req(ia, buf->rb_send_bufs[i]); 1078 } 1079 1080 ia->ri_ops->ro_destroy(buf); 1081 1082 kfree(buf->rb_pool); 1083 } 1084 1085 struct rpcrdma_mw * 1086 rpcrdma_get_mw(struct rpcrdma_xprt *r_xprt) 1087 { 1088 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1089 struct rpcrdma_mw *mw = NULL; 1090 1091 spin_lock(&buf->rb_mwlock); 1092 if (!list_empty(&buf->rb_mws)) { 1093 mw = list_first_entry(&buf->rb_mws, 1094 struct rpcrdma_mw, mw_list); 1095 list_del_init(&mw->mw_list); 1096 } 1097 spin_unlock(&buf->rb_mwlock); 1098 1099 if (!mw) 1100 pr_err("RPC: %s: no MWs available\n", __func__); 1101 return mw; 1102 } 1103 1104 void 1105 rpcrdma_put_mw(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mw *mw) 1106 { 1107 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1108 1109 spin_lock(&buf->rb_mwlock); 1110 list_add_tail(&mw->mw_list, &buf->rb_mws); 1111 spin_unlock(&buf->rb_mwlock); 1112 } 1113 1114 static void 1115 rpcrdma_buffer_put_sendbuf(struct rpcrdma_req *req, struct rpcrdma_buffer *buf) 1116 { 1117 buf->rb_send_bufs[--buf->rb_send_index] = req; 1118 req->rl_niovs = 0; 1119 if (req->rl_reply) { 1120 buf->rb_recv_bufs[--buf->rb_recv_index] = req->rl_reply; 1121 req->rl_reply = NULL; 1122 } 1123 } 1124 1125 /* 1126 * Get a set of request/reply buffers. 1127 * 1128 * Reply buffer (if needed) is attached to send buffer upon return. 1129 * Rule: 1130 * rb_send_index and rb_recv_index MUST always be pointing to the 1131 * *next* available buffer (non-NULL). They are incremented after 1132 * removing buffers, and decremented *before* returning them. 1133 */ 1134 struct rpcrdma_req * 1135 rpcrdma_buffer_get(struct rpcrdma_buffer *buffers) 1136 { 1137 struct rpcrdma_req *req; 1138 unsigned long flags; 1139 1140 spin_lock_irqsave(&buffers->rb_lock, flags); 1141 1142 if (buffers->rb_send_index == buffers->rb_max_requests) { 1143 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1144 dprintk("RPC: %s: out of request buffers\n", __func__); 1145 return ((struct rpcrdma_req *)NULL); 1146 } 1147 1148 req = buffers->rb_send_bufs[buffers->rb_send_index]; 1149 if (buffers->rb_send_index < buffers->rb_recv_index) { 1150 dprintk("RPC: %s: %d extra receives outstanding (ok)\n", 1151 __func__, 1152 buffers->rb_recv_index - buffers->rb_send_index); 1153 req->rl_reply = NULL; 1154 } else { 1155 req->rl_reply = buffers->rb_recv_bufs[buffers->rb_recv_index]; 1156 buffers->rb_recv_bufs[buffers->rb_recv_index++] = NULL; 1157 } 1158 buffers->rb_send_bufs[buffers->rb_send_index++] = NULL; 1159 1160 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1161 return req; 1162 } 1163 1164 /* 1165 * Put request/reply buffers back into pool. 1166 * Pre-decrement counter/array index. 1167 */ 1168 void 1169 rpcrdma_buffer_put(struct rpcrdma_req *req) 1170 { 1171 struct rpcrdma_buffer *buffers = req->rl_buffer; 1172 unsigned long flags; 1173 1174 spin_lock_irqsave(&buffers->rb_lock, flags); 1175 rpcrdma_buffer_put_sendbuf(req, buffers); 1176 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1177 } 1178 1179 /* 1180 * Recover reply buffers from pool. 1181 * This happens when recovering from error conditions. 1182 * Post-increment counter/array index. 1183 */ 1184 void 1185 rpcrdma_recv_buffer_get(struct rpcrdma_req *req) 1186 { 1187 struct rpcrdma_buffer *buffers = req->rl_buffer; 1188 unsigned long flags; 1189 1190 spin_lock_irqsave(&buffers->rb_lock, flags); 1191 if (buffers->rb_recv_index < buffers->rb_max_requests) { 1192 req->rl_reply = buffers->rb_recv_bufs[buffers->rb_recv_index]; 1193 buffers->rb_recv_bufs[buffers->rb_recv_index++] = NULL; 1194 } 1195 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1196 } 1197 1198 /* 1199 * Put reply buffers back into pool when not attached to 1200 * request. This happens in error conditions. 1201 */ 1202 void 1203 rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep) 1204 { 1205 struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf; 1206 unsigned long flags; 1207 1208 spin_lock_irqsave(&buffers->rb_lock, flags); 1209 buffers->rb_recv_bufs[--buffers->rb_recv_index] = rep; 1210 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1211 } 1212 1213 /* 1214 * Wrappers for internal-use kmalloc memory registration, used by buffer code. 1215 */ 1216 1217 void 1218 rpcrdma_mapping_error(struct rpcrdma_mr_seg *seg) 1219 { 1220 dprintk("RPC: map_one: offset %p iova %llx len %zu\n", 1221 seg->mr_offset, 1222 (unsigned long long)seg->mr_dma, seg->mr_dmalen); 1223 } 1224 1225 /** 1226 * rpcrdma_alloc_regbuf - kmalloc and register memory for SEND/RECV buffers 1227 * @ia: controlling rpcrdma_ia 1228 * @size: size of buffer to be allocated, in bytes 1229 * @flags: GFP flags 1230 * 1231 * Returns pointer to private header of an area of internally 1232 * registered memory, or an ERR_PTR. The registered buffer follows 1233 * the end of the private header. 1234 * 1235 * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for 1236 * receiving the payload of RDMA RECV operations. regbufs are not 1237 * used for RDMA READ/WRITE operations, thus are registered only for 1238 * LOCAL access. 1239 */ 1240 struct rpcrdma_regbuf * 1241 rpcrdma_alloc_regbuf(struct rpcrdma_ia *ia, size_t size, gfp_t flags) 1242 { 1243 struct rpcrdma_regbuf *rb; 1244 struct ib_sge *iov; 1245 1246 rb = kmalloc(sizeof(*rb) + size, flags); 1247 if (rb == NULL) 1248 goto out; 1249 1250 iov = &rb->rg_iov; 1251 iov->addr = ib_dma_map_single(ia->ri_device, 1252 (void *)rb->rg_base, size, 1253 DMA_BIDIRECTIONAL); 1254 if (ib_dma_mapping_error(ia->ri_device, iov->addr)) 1255 goto out_free; 1256 1257 iov->length = size; 1258 iov->lkey = ia->ri_pd->local_dma_lkey; 1259 rb->rg_size = size; 1260 rb->rg_owner = NULL; 1261 return rb; 1262 1263 out_free: 1264 kfree(rb); 1265 out: 1266 return ERR_PTR(-ENOMEM); 1267 } 1268 1269 /** 1270 * rpcrdma_free_regbuf - deregister and free registered buffer 1271 * @ia: controlling rpcrdma_ia 1272 * @rb: regbuf to be deregistered and freed 1273 */ 1274 void 1275 rpcrdma_free_regbuf(struct rpcrdma_ia *ia, struct rpcrdma_regbuf *rb) 1276 { 1277 struct ib_sge *iov; 1278 1279 if (!rb) 1280 return; 1281 1282 iov = &rb->rg_iov; 1283 ib_dma_unmap_single(ia->ri_device, 1284 iov->addr, iov->length, DMA_BIDIRECTIONAL); 1285 kfree(rb); 1286 } 1287 1288 /* 1289 * Prepost any receive buffer, then post send. 1290 * 1291 * Receive buffer is donated to hardware, reclaimed upon recv completion. 1292 */ 1293 int 1294 rpcrdma_ep_post(struct rpcrdma_ia *ia, 1295 struct rpcrdma_ep *ep, 1296 struct rpcrdma_req *req) 1297 { 1298 struct ib_device *device = ia->ri_device; 1299 struct ib_send_wr send_wr, *send_wr_fail; 1300 struct rpcrdma_rep *rep = req->rl_reply; 1301 struct ib_sge *iov = req->rl_send_iov; 1302 int i, rc; 1303 1304 if (rep) { 1305 rc = rpcrdma_ep_post_recv(ia, ep, rep); 1306 if (rc) 1307 goto out; 1308 req->rl_reply = NULL; 1309 } 1310 1311 send_wr.next = NULL; 1312 send_wr.wr_id = RPCRDMA_IGNORE_COMPLETION; 1313 send_wr.sg_list = iov; 1314 send_wr.num_sge = req->rl_niovs; 1315 send_wr.opcode = IB_WR_SEND; 1316 1317 for (i = 0; i < send_wr.num_sge; i++) 1318 ib_dma_sync_single_for_device(device, iov[i].addr, 1319 iov[i].length, DMA_TO_DEVICE); 1320 dprintk("RPC: %s: posting %d s/g entries\n", 1321 __func__, send_wr.num_sge); 1322 1323 if (DECR_CQCOUNT(ep) > 0) 1324 send_wr.send_flags = 0; 1325 else { /* Provider must take a send completion every now and then */ 1326 INIT_CQCOUNT(ep); 1327 send_wr.send_flags = IB_SEND_SIGNALED; 1328 } 1329 1330 rc = ib_post_send(ia->ri_id->qp, &send_wr, &send_wr_fail); 1331 if (rc) 1332 dprintk("RPC: %s: ib_post_send returned %i\n", __func__, 1333 rc); 1334 out: 1335 return rc; 1336 } 1337 1338 /* 1339 * (Re)post a receive buffer. 1340 */ 1341 int 1342 rpcrdma_ep_post_recv(struct rpcrdma_ia *ia, 1343 struct rpcrdma_ep *ep, 1344 struct rpcrdma_rep *rep) 1345 { 1346 struct ib_recv_wr recv_wr, *recv_wr_fail; 1347 int rc; 1348 1349 recv_wr.next = NULL; 1350 recv_wr.wr_id = (u64) (unsigned long) rep; 1351 recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov; 1352 recv_wr.num_sge = 1; 1353 1354 ib_dma_sync_single_for_cpu(ia->ri_device, 1355 rdmab_addr(rep->rr_rdmabuf), 1356 rdmab_length(rep->rr_rdmabuf), 1357 DMA_BIDIRECTIONAL); 1358 1359 rc = ib_post_recv(ia->ri_id->qp, &recv_wr, &recv_wr_fail); 1360 1361 if (rc) 1362 dprintk("RPC: %s: ib_post_recv returned %i\n", __func__, 1363 rc); 1364 return rc; 1365 } 1366 1367 /* How many chunk list items fit within our inline buffers? 1368 */ 1369 unsigned int 1370 rpcrdma_max_segments(struct rpcrdma_xprt *r_xprt) 1371 { 1372 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 1373 int bytes, segments; 1374 1375 bytes = min_t(unsigned int, cdata->inline_wsize, cdata->inline_rsize); 1376 bytes -= RPCRDMA_HDRLEN_MIN; 1377 if (bytes < sizeof(struct rpcrdma_segment) * 2) { 1378 pr_warn("RPC: %s: inline threshold too small\n", 1379 __func__); 1380 return 0; 1381 } 1382 1383 segments = 1 << (fls(bytes / sizeof(struct rpcrdma_segment)) - 1); 1384 dprintk("RPC: %s: max chunk list size = %d segments\n", 1385 __func__, segments); 1386 return segments; 1387 } 1388