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 if (!(devattr->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS) || 547 (devattr->max_fast_reg_page_list_len == 0)) { 548 dprintk("RPC: %s: FRMR registration " 549 "not supported by HCA\n", __func__); 550 memreg = RPCRDMA_MTHCAFMR; 551 } 552 } 553 if (memreg == RPCRDMA_MTHCAFMR) { 554 if (!ia->ri_device->alloc_fmr) { 555 dprintk("RPC: %s: MTHCAFMR registration " 556 "not supported by HCA\n", __func__); 557 rc = -EINVAL; 558 goto out3; 559 } 560 } 561 562 switch (memreg) { 563 case RPCRDMA_FRMR: 564 ia->ri_ops = &rpcrdma_frwr_memreg_ops; 565 break; 566 case RPCRDMA_ALLPHYSICAL: 567 ia->ri_ops = &rpcrdma_physical_memreg_ops; 568 break; 569 case RPCRDMA_MTHCAFMR: 570 ia->ri_ops = &rpcrdma_fmr_memreg_ops; 571 break; 572 default: 573 printk(KERN_ERR "RPC: Unsupported memory " 574 "registration mode: %d\n", memreg); 575 rc = -ENOMEM; 576 goto out3; 577 } 578 dprintk("RPC: %s: memory registration strategy is '%s'\n", 579 __func__, ia->ri_ops->ro_displayname); 580 581 rwlock_init(&ia->ri_qplock); 582 return 0; 583 584 out3: 585 ib_dealloc_pd(ia->ri_pd); 586 ia->ri_pd = NULL; 587 out2: 588 rpcrdma_destroy_id(ia->ri_id); 589 ia->ri_id = NULL; 590 out1: 591 return rc; 592 } 593 594 /* 595 * Clean up/close an IA. 596 * o if event handles and PD have been initialized, free them. 597 * o close the IA 598 */ 599 void 600 rpcrdma_ia_close(struct rpcrdma_ia *ia) 601 { 602 dprintk("RPC: %s: entering\n", __func__); 603 if (ia->ri_id != NULL && !IS_ERR(ia->ri_id)) { 604 if (ia->ri_id->qp) 605 rdma_destroy_qp(ia->ri_id); 606 rpcrdma_destroy_id(ia->ri_id); 607 ia->ri_id = NULL; 608 } 609 610 /* If the pd is still busy, xprtrdma missed freeing a resource */ 611 if (ia->ri_pd && !IS_ERR(ia->ri_pd)) 612 ib_dealloc_pd(ia->ri_pd); 613 } 614 615 /* 616 * Create unconnected endpoint. 617 */ 618 int 619 rpcrdma_ep_create(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia, 620 struct rpcrdma_create_data_internal *cdata) 621 { 622 struct ib_device_attr *devattr = &ia->ri_devattr; 623 struct ib_cq *sendcq, *recvcq; 624 struct ib_cq_init_attr cq_attr = {}; 625 int rc, err; 626 627 if (devattr->max_sge < RPCRDMA_MAX_IOVS) { 628 dprintk("RPC: %s: insufficient sge's available\n", 629 __func__); 630 return -ENOMEM; 631 } 632 633 /* check provider's send/recv wr limits */ 634 if (cdata->max_requests > devattr->max_qp_wr) 635 cdata->max_requests = devattr->max_qp_wr; 636 637 ep->rep_attr.event_handler = rpcrdma_qp_async_error_upcall; 638 ep->rep_attr.qp_context = ep; 639 ep->rep_attr.srq = NULL; 640 ep->rep_attr.cap.max_send_wr = cdata->max_requests; 641 rc = ia->ri_ops->ro_open(ia, ep, cdata); 642 if (rc) 643 return rc; 644 ep->rep_attr.cap.max_recv_wr = cdata->max_requests; 645 ep->rep_attr.cap.max_send_sge = RPCRDMA_MAX_IOVS; 646 ep->rep_attr.cap.max_recv_sge = 1; 647 ep->rep_attr.cap.max_inline_data = 0; 648 ep->rep_attr.sq_sig_type = IB_SIGNAL_REQ_WR; 649 ep->rep_attr.qp_type = IB_QPT_RC; 650 ep->rep_attr.port_num = ~0; 651 652 dprintk("RPC: %s: requested max: dtos: send %d recv %d; " 653 "iovs: send %d recv %d\n", 654 __func__, 655 ep->rep_attr.cap.max_send_wr, 656 ep->rep_attr.cap.max_recv_wr, 657 ep->rep_attr.cap.max_send_sge, 658 ep->rep_attr.cap.max_recv_sge); 659 660 /* set trigger for requesting send completion */ 661 ep->rep_cqinit = ep->rep_attr.cap.max_send_wr/2 - 1; 662 if (ep->rep_cqinit > RPCRDMA_MAX_UNSIGNALED_SENDS) 663 ep->rep_cqinit = RPCRDMA_MAX_UNSIGNALED_SENDS; 664 else if (ep->rep_cqinit <= 2) 665 ep->rep_cqinit = 0; 666 INIT_CQCOUNT(ep); 667 init_waitqueue_head(&ep->rep_connect_wait); 668 INIT_DELAYED_WORK(&ep->rep_connect_worker, rpcrdma_connect_worker); 669 670 cq_attr.cqe = ep->rep_attr.cap.max_send_wr + 1; 671 sendcq = ib_create_cq(ia->ri_device, rpcrdma_sendcq_upcall, 672 rpcrdma_cq_async_error_upcall, ep, &cq_attr); 673 if (IS_ERR(sendcq)) { 674 rc = PTR_ERR(sendcq); 675 dprintk("RPC: %s: failed to create send CQ: %i\n", 676 __func__, rc); 677 goto out1; 678 } 679 680 rc = ib_req_notify_cq(sendcq, IB_CQ_NEXT_COMP); 681 if (rc) { 682 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n", 683 __func__, rc); 684 goto out2; 685 } 686 687 cq_attr.cqe = ep->rep_attr.cap.max_recv_wr + 1; 688 recvcq = ib_create_cq(ia->ri_device, rpcrdma_recvcq_upcall, 689 rpcrdma_cq_async_error_upcall, ep, &cq_attr); 690 if (IS_ERR(recvcq)) { 691 rc = PTR_ERR(recvcq); 692 dprintk("RPC: %s: failed to create recv CQ: %i\n", 693 __func__, rc); 694 goto out2; 695 } 696 697 rc = ib_req_notify_cq(recvcq, IB_CQ_NEXT_COMP); 698 if (rc) { 699 dprintk("RPC: %s: ib_req_notify_cq failed: %i\n", 700 __func__, rc); 701 ib_destroy_cq(recvcq); 702 goto out2; 703 } 704 705 ep->rep_attr.send_cq = sendcq; 706 ep->rep_attr.recv_cq = recvcq; 707 708 /* Initialize cma parameters */ 709 710 /* RPC/RDMA does not use private data */ 711 ep->rep_remote_cma.private_data = NULL; 712 ep->rep_remote_cma.private_data_len = 0; 713 714 /* Client offers RDMA Read but does not initiate */ 715 ep->rep_remote_cma.initiator_depth = 0; 716 if (devattr->max_qp_rd_atom > 32) /* arbitrary but <= 255 */ 717 ep->rep_remote_cma.responder_resources = 32; 718 else 719 ep->rep_remote_cma.responder_resources = 720 devattr->max_qp_rd_atom; 721 722 ep->rep_remote_cma.retry_count = 7; 723 ep->rep_remote_cma.flow_control = 0; 724 ep->rep_remote_cma.rnr_retry_count = 0; 725 726 return 0; 727 728 out2: 729 err = ib_destroy_cq(sendcq); 730 if (err) 731 dprintk("RPC: %s: ib_destroy_cq returned %i\n", 732 __func__, err); 733 out1: 734 if (ia->ri_dma_mr) 735 ib_dereg_mr(ia->ri_dma_mr); 736 return rc; 737 } 738 739 /* 740 * rpcrdma_ep_destroy 741 * 742 * Disconnect and destroy endpoint. After this, the only 743 * valid operations on the ep are to free it (if dynamically 744 * allocated) or re-create it. 745 */ 746 void 747 rpcrdma_ep_destroy(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 748 { 749 int rc; 750 751 dprintk("RPC: %s: entering, connected is %d\n", 752 __func__, ep->rep_connected); 753 754 cancel_delayed_work_sync(&ep->rep_connect_worker); 755 756 if (ia->ri_id->qp) 757 rpcrdma_ep_disconnect(ep, ia); 758 759 rpcrdma_clean_cq(ep->rep_attr.recv_cq); 760 rpcrdma_clean_cq(ep->rep_attr.send_cq); 761 762 if (ia->ri_id->qp) { 763 rdma_destroy_qp(ia->ri_id); 764 ia->ri_id->qp = NULL; 765 } 766 767 rc = ib_destroy_cq(ep->rep_attr.recv_cq); 768 if (rc) 769 dprintk("RPC: %s: ib_destroy_cq returned %i\n", 770 __func__, rc); 771 772 rc = ib_destroy_cq(ep->rep_attr.send_cq); 773 if (rc) 774 dprintk("RPC: %s: ib_destroy_cq returned %i\n", 775 __func__, rc); 776 777 if (ia->ri_dma_mr) { 778 rc = ib_dereg_mr(ia->ri_dma_mr); 779 dprintk("RPC: %s: ib_dereg_mr returned %i\n", 780 __func__, rc); 781 } 782 } 783 784 /* 785 * Connect unconnected endpoint. 786 */ 787 int 788 rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 789 { 790 struct rdma_cm_id *id, *old; 791 int rc = 0; 792 int retry_count = 0; 793 794 if (ep->rep_connected != 0) { 795 struct rpcrdma_xprt *xprt; 796 retry: 797 dprintk("RPC: %s: reconnecting...\n", __func__); 798 799 rpcrdma_ep_disconnect(ep, ia); 800 rpcrdma_flush_cqs(ep); 801 802 xprt = container_of(ia, struct rpcrdma_xprt, rx_ia); 803 id = rpcrdma_create_id(xprt, ia, 804 (struct sockaddr *)&xprt->rx_data.addr); 805 if (IS_ERR(id)) { 806 rc = -EHOSTUNREACH; 807 goto out; 808 } 809 /* TEMP TEMP TEMP - fail if new device: 810 * Deregister/remarshal *all* requests! 811 * Close and recreate adapter, pd, etc! 812 * Re-determine all attributes still sane! 813 * More stuff I haven't thought of! 814 * Rrrgh! 815 */ 816 if (ia->ri_device != id->device) { 817 printk("RPC: %s: can't reconnect on " 818 "different device!\n", __func__); 819 rpcrdma_destroy_id(id); 820 rc = -ENETUNREACH; 821 goto out; 822 } 823 /* END TEMP */ 824 rc = rdma_create_qp(id, ia->ri_pd, &ep->rep_attr); 825 if (rc) { 826 dprintk("RPC: %s: rdma_create_qp failed %i\n", 827 __func__, rc); 828 rpcrdma_destroy_id(id); 829 rc = -ENETUNREACH; 830 goto out; 831 } 832 833 write_lock(&ia->ri_qplock); 834 old = ia->ri_id; 835 ia->ri_id = id; 836 write_unlock(&ia->ri_qplock); 837 838 rdma_destroy_qp(old); 839 rpcrdma_destroy_id(old); 840 } else { 841 dprintk("RPC: %s: connecting...\n", __func__); 842 rc = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr); 843 if (rc) { 844 dprintk("RPC: %s: rdma_create_qp failed %i\n", 845 __func__, rc); 846 /* do not update ep->rep_connected */ 847 return -ENETUNREACH; 848 } 849 } 850 851 ep->rep_connected = 0; 852 853 rc = rdma_connect(ia->ri_id, &ep->rep_remote_cma); 854 if (rc) { 855 dprintk("RPC: %s: rdma_connect() failed with %i\n", 856 __func__, rc); 857 goto out; 858 } 859 860 wait_event_interruptible(ep->rep_connect_wait, ep->rep_connected != 0); 861 862 /* 863 * Check state. A non-peer reject indicates no listener 864 * (ECONNREFUSED), which may be a transient state. All 865 * others indicate a transport condition which has already 866 * undergone a best-effort. 867 */ 868 if (ep->rep_connected == -ECONNREFUSED && 869 ++retry_count <= RDMA_CONNECT_RETRY_MAX) { 870 dprintk("RPC: %s: non-peer_reject, retry\n", __func__); 871 goto retry; 872 } 873 if (ep->rep_connected <= 0) { 874 /* Sometimes, the only way to reliably connect to remote 875 * CMs is to use same nonzero values for ORD and IRD. */ 876 if (retry_count++ <= RDMA_CONNECT_RETRY_MAX + 1 && 877 (ep->rep_remote_cma.responder_resources == 0 || 878 ep->rep_remote_cma.initiator_depth != 879 ep->rep_remote_cma.responder_resources)) { 880 if (ep->rep_remote_cma.responder_resources == 0) 881 ep->rep_remote_cma.responder_resources = 1; 882 ep->rep_remote_cma.initiator_depth = 883 ep->rep_remote_cma.responder_resources; 884 goto retry; 885 } 886 rc = ep->rep_connected; 887 } else { 888 dprintk("RPC: %s: connected\n", __func__); 889 } 890 891 out: 892 if (rc) 893 ep->rep_connected = rc; 894 return rc; 895 } 896 897 /* 898 * rpcrdma_ep_disconnect 899 * 900 * This is separate from destroy to facilitate the ability 901 * to reconnect without recreating the endpoint. 902 * 903 * This call is not reentrant, and must not be made in parallel 904 * on the same endpoint. 905 */ 906 void 907 rpcrdma_ep_disconnect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia) 908 { 909 int rc; 910 911 rpcrdma_flush_cqs(ep); 912 rc = rdma_disconnect(ia->ri_id); 913 if (!rc) { 914 /* returns without wait if not connected */ 915 wait_event_interruptible(ep->rep_connect_wait, 916 ep->rep_connected != 1); 917 dprintk("RPC: %s: after wait, %sconnected\n", __func__, 918 (ep->rep_connected == 1) ? "still " : "dis"); 919 } else { 920 dprintk("RPC: %s: rdma_disconnect %i\n", __func__, rc); 921 ep->rep_connected = rc; 922 } 923 } 924 925 static struct rpcrdma_req * 926 rpcrdma_create_req(struct rpcrdma_xprt *r_xprt) 927 { 928 struct rpcrdma_req *req; 929 930 req = kzalloc(sizeof(*req), GFP_KERNEL); 931 if (req == NULL) 932 return ERR_PTR(-ENOMEM); 933 934 req->rl_buffer = &r_xprt->rx_buf; 935 return req; 936 } 937 938 static struct rpcrdma_rep * 939 rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt) 940 { 941 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 942 struct rpcrdma_ia *ia = &r_xprt->rx_ia; 943 struct rpcrdma_rep *rep; 944 int rc; 945 946 rc = -ENOMEM; 947 rep = kzalloc(sizeof(*rep), GFP_KERNEL); 948 if (rep == NULL) 949 goto out; 950 951 rep->rr_rdmabuf = rpcrdma_alloc_regbuf(ia, cdata->inline_rsize, 952 GFP_KERNEL); 953 if (IS_ERR(rep->rr_rdmabuf)) { 954 rc = PTR_ERR(rep->rr_rdmabuf); 955 goto out_free; 956 } 957 958 rep->rr_device = ia->ri_device; 959 rep->rr_rxprt = r_xprt; 960 return rep; 961 962 out_free: 963 kfree(rep); 964 out: 965 return ERR_PTR(rc); 966 } 967 968 int 969 rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt) 970 { 971 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 972 struct rpcrdma_ia *ia = &r_xprt->rx_ia; 973 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 974 char *p; 975 size_t len; 976 int i, rc; 977 978 buf->rb_max_requests = cdata->max_requests; 979 spin_lock_init(&buf->rb_lock); 980 981 /* Need to allocate: 982 * 1. arrays for send and recv pointers 983 * 2. arrays of struct rpcrdma_req to fill in pointers 984 * 3. array of struct rpcrdma_rep for replies 985 * Send/recv buffers in req/rep need to be registered 986 */ 987 len = buf->rb_max_requests * 988 (sizeof(struct rpcrdma_req *) + sizeof(struct rpcrdma_rep *)); 989 990 p = kzalloc(len, GFP_KERNEL); 991 if (p == NULL) { 992 dprintk("RPC: %s: req_t/rep_t/pad kzalloc(%zd) failed\n", 993 __func__, len); 994 rc = -ENOMEM; 995 goto out; 996 } 997 buf->rb_pool = p; /* for freeing it later */ 998 999 buf->rb_send_bufs = (struct rpcrdma_req **) p; 1000 p = (char *) &buf->rb_send_bufs[buf->rb_max_requests]; 1001 buf->rb_recv_bufs = (struct rpcrdma_rep **) p; 1002 p = (char *) &buf->rb_recv_bufs[buf->rb_max_requests]; 1003 1004 rc = ia->ri_ops->ro_init(r_xprt); 1005 if (rc) 1006 goto out; 1007 1008 for (i = 0; i < buf->rb_max_requests; i++) { 1009 struct rpcrdma_req *req; 1010 struct rpcrdma_rep *rep; 1011 1012 req = rpcrdma_create_req(r_xprt); 1013 if (IS_ERR(req)) { 1014 dprintk("RPC: %s: request buffer %d alloc" 1015 " failed\n", __func__, i); 1016 rc = PTR_ERR(req); 1017 goto out; 1018 } 1019 buf->rb_send_bufs[i] = req; 1020 1021 rep = rpcrdma_create_rep(r_xprt); 1022 if (IS_ERR(rep)) { 1023 dprintk("RPC: %s: reply buffer %d alloc failed\n", 1024 __func__, i); 1025 rc = PTR_ERR(rep); 1026 goto out; 1027 } 1028 buf->rb_recv_bufs[i] = rep; 1029 } 1030 1031 return 0; 1032 out: 1033 rpcrdma_buffer_destroy(buf); 1034 return rc; 1035 } 1036 1037 static void 1038 rpcrdma_destroy_rep(struct rpcrdma_ia *ia, struct rpcrdma_rep *rep) 1039 { 1040 if (!rep) 1041 return; 1042 1043 rpcrdma_free_regbuf(ia, rep->rr_rdmabuf); 1044 kfree(rep); 1045 } 1046 1047 static void 1048 rpcrdma_destroy_req(struct rpcrdma_ia *ia, struct rpcrdma_req *req) 1049 { 1050 if (!req) 1051 return; 1052 1053 rpcrdma_free_regbuf(ia, req->rl_sendbuf); 1054 rpcrdma_free_regbuf(ia, req->rl_rdmabuf); 1055 kfree(req); 1056 } 1057 1058 void 1059 rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf) 1060 { 1061 struct rpcrdma_ia *ia = rdmab_to_ia(buf); 1062 int i; 1063 1064 /* clean up in reverse order from create 1065 * 1. recv mr memory (mr free, then kfree) 1066 * 2. send mr memory (mr free, then kfree) 1067 * 3. MWs 1068 */ 1069 dprintk("RPC: %s: entering\n", __func__); 1070 1071 for (i = 0; i < buf->rb_max_requests; i++) { 1072 if (buf->rb_recv_bufs) 1073 rpcrdma_destroy_rep(ia, buf->rb_recv_bufs[i]); 1074 if (buf->rb_send_bufs) 1075 rpcrdma_destroy_req(ia, buf->rb_send_bufs[i]); 1076 } 1077 1078 ia->ri_ops->ro_destroy(buf); 1079 1080 kfree(buf->rb_pool); 1081 } 1082 1083 struct rpcrdma_mw * 1084 rpcrdma_get_mw(struct rpcrdma_xprt *r_xprt) 1085 { 1086 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1087 struct rpcrdma_mw *mw = NULL; 1088 1089 spin_lock(&buf->rb_mwlock); 1090 if (!list_empty(&buf->rb_mws)) { 1091 mw = list_first_entry(&buf->rb_mws, 1092 struct rpcrdma_mw, mw_list); 1093 list_del_init(&mw->mw_list); 1094 } 1095 spin_unlock(&buf->rb_mwlock); 1096 1097 if (!mw) 1098 pr_err("RPC: %s: no MWs available\n", __func__); 1099 return mw; 1100 } 1101 1102 void 1103 rpcrdma_put_mw(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mw *mw) 1104 { 1105 struct rpcrdma_buffer *buf = &r_xprt->rx_buf; 1106 1107 spin_lock(&buf->rb_mwlock); 1108 list_add_tail(&mw->mw_list, &buf->rb_mws); 1109 spin_unlock(&buf->rb_mwlock); 1110 } 1111 1112 static void 1113 rpcrdma_buffer_put_sendbuf(struct rpcrdma_req *req, struct rpcrdma_buffer *buf) 1114 { 1115 buf->rb_send_bufs[--buf->rb_send_index] = req; 1116 req->rl_niovs = 0; 1117 if (req->rl_reply) { 1118 buf->rb_recv_bufs[--buf->rb_recv_index] = req->rl_reply; 1119 req->rl_reply = NULL; 1120 } 1121 } 1122 1123 /* 1124 * Get a set of request/reply buffers. 1125 * 1126 * Reply buffer (if needed) is attached to send buffer upon return. 1127 * Rule: 1128 * rb_send_index and rb_recv_index MUST always be pointing to the 1129 * *next* available buffer (non-NULL). They are incremented after 1130 * removing buffers, and decremented *before* returning them. 1131 */ 1132 struct rpcrdma_req * 1133 rpcrdma_buffer_get(struct rpcrdma_buffer *buffers) 1134 { 1135 struct rpcrdma_req *req; 1136 unsigned long flags; 1137 1138 spin_lock_irqsave(&buffers->rb_lock, flags); 1139 1140 if (buffers->rb_send_index == buffers->rb_max_requests) { 1141 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1142 dprintk("RPC: %s: out of request buffers\n", __func__); 1143 return ((struct rpcrdma_req *)NULL); 1144 } 1145 1146 req = buffers->rb_send_bufs[buffers->rb_send_index]; 1147 if (buffers->rb_send_index < buffers->rb_recv_index) { 1148 dprintk("RPC: %s: %d extra receives outstanding (ok)\n", 1149 __func__, 1150 buffers->rb_recv_index - buffers->rb_send_index); 1151 req->rl_reply = NULL; 1152 } else { 1153 req->rl_reply = buffers->rb_recv_bufs[buffers->rb_recv_index]; 1154 buffers->rb_recv_bufs[buffers->rb_recv_index++] = NULL; 1155 } 1156 buffers->rb_send_bufs[buffers->rb_send_index++] = NULL; 1157 1158 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1159 return req; 1160 } 1161 1162 /* 1163 * Put request/reply buffers back into pool. 1164 * Pre-decrement counter/array index. 1165 */ 1166 void 1167 rpcrdma_buffer_put(struct rpcrdma_req *req) 1168 { 1169 struct rpcrdma_buffer *buffers = req->rl_buffer; 1170 unsigned long flags; 1171 1172 spin_lock_irqsave(&buffers->rb_lock, flags); 1173 rpcrdma_buffer_put_sendbuf(req, buffers); 1174 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1175 } 1176 1177 /* 1178 * Recover reply buffers from pool. 1179 * This happens when recovering from error conditions. 1180 * Post-increment counter/array index. 1181 */ 1182 void 1183 rpcrdma_recv_buffer_get(struct rpcrdma_req *req) 1184 { 1185 struct rpcrdma_buffer *buffers = req->rl_buffer; 1186 unsigned long flags; 1187 1188 spin_lock_irqsave(&buffers->rb_lock, flags); 1189 if (buffers->rb_recv_index < buffers->rb_max_requests) { 1190 req->rl_reply = buffers->rb_recv_bufs[buffers->rb_recv_index]; 1191 buffers->rb_recv_bufs[buffers->rb_recv_index++] = NULL; 1192 } 1193 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1194 } 1195 1196 /* 1197 * Put reply buffers back into pool when not attached to 1198 * request. This happens in error conditions. 1199 */ 1200 void 1201 rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep) 1202 { 1203 struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf; 1204 unsigned long flags; 1205 1206 spin_lock_irqsave(&buffers->rb_lock, flags); 1207 buffers->rb_recv_bufs[--buffers->rb_recv_index] = rep; 1208 spin_unlock_irqrestore(&buffers->rb_lock, flags); 1209 } 1210 1211 /* 1212 * Wrappers for internal-use kmalloc memory registration, used by buffer code. 1213 */ 1214 1215 void 1216 rpcrdma_mapping_error(struct rpcrdma_mr_seg *seg) 1217 { 1218 dprintk("RPC: map_one: offset %p iova %llx len %zu\n", 1219 seg->mr_offset, 1220 (unsigned long long)seg->mr_dma, seg->mr_dmalen); 1221 } 1222 1223 /** 1224 * rpcrdma_alloc_regbuf - kmalloc and register memory for SEND/RECV buffers 1225 * @ia: controlling rpcrdma_ia 1226 * @size: size of buffer to be allocated, in bytes 1227 * @flags: GFP flags 1228 * 1229 * Returns pointer to private header of an area of internally 1230 * registered memory, or an ERR_PTR. The registered buffer follows 1231 * the end of the private header. 1232 * 1233 * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for 1234 * receiving the payload of RDMA RECV operations. regbufs are not 1235 * used for RDMA READ/WRITE operations, thus are registered only for 1236 * LOCAL access. 1237 */ 1238 struct rpcrdma_regbuf * 1239 rpcrdma_alloc_regbuf(struct rpcrdma_ia *ia, size_t size, gfp_t flags) 1240 { 1241 struct rpcrdma_regbuf *rb; 1242 struct ib_sge *iov; 1243 1244 rb = kmalloc(sizeof(*rb) + size, flags); 1245 if (rb == NULL) 1246 goto out; 1247 1248 iov = &rb->rg_iov; 1249 iov->addr = ib_dma_map_single(ia->ri_device, 1250 (void *)rb->rg_base, size, 1251 DMA_BIDIRECTIONAL); 1252 if (ib_dma_mapping_error(ia->ri_device, iov->addr)) 1253 goto out_free; 1254 1255 iov->length = size; 1256 iov->lkey = ia->ri_pd->local_dma_lkey; 1257 rb->rg_size = size; 1258 rb->rg_owner = NULL; 1259 return rb; 1260 1261 out_free: 1262 kfree(rb); 1263 out: 1264 return ERR_PTR(-ENOMEM); 1265 } 1266 1267 /** 1268 * rpcrdma_free_regbuf - deregister and free registered buffer 1269 * @ia: controlling rpcrdma_ia 1270 * @rb: regbuf to be deregistered and freed 1271 */ 1272 void 1273 rpcrdma_free_regbuf(struct rpcrdma_ia *ia, struct rpcrdma_regbuf *rb) 1274 { 1275 struct ib_sge *iov; 1276 1277 if (!rb) 1278 return; 1279 1280 iov = &rb->rg_iov; 1281 ib_dma_unmap_single(ia->ri_device, 1282 iov->addr, iov->length, DMA_BIDIRECTIONAL); 1283 kfree(rb); 1284 } 1285 1286 /* 1287 * Prepost any receive buffer, then post send. 1288 * 1289 * Receive buffer is donated to hardware, reclaimed upon recv completion. 1290 */ 1291 int 1292 rpcrdma_ep_post(struct rpcrdma_ia *ia, 1293 struct rpcrdma_ep *ep, 1294 struct rpcrdma_req *req) 1295 { 1296 struct ib_device *device = ia->ri_device; 1297 struct ib_send_wr send_wr, *send_wr_fail; 1298 struct rpcrdma_rep *rep = req->rl_reply; 1299 struct ib_sge *iov = req->rl_send_iov; 1300 int i, rc; 1301 1302 if (rep) { 1303 rc = rpcrdma_ep_post_recv(ia, ep, rep); 1304 if (rc) 1305 goto out; 1306 req->rl_reply = NULL; 1307 } 1308 1309 send_wr.next = NULL; 1310 send_wr.wr_id = RPCRDMA_IGNORE_COMPLETION; 1311 send_wr.sg_list = iov; 1312 send_wr.num_sge = req->rl_niovs; 1313 send_wr.opcode = IB_WR_SEND; 1314 1315 for (i = 0; i < send_wr.num_sge; i++) 1316 ib_dma_sync_single_for_device(device, iov[i].addr, 1317 iov[i].length, DMA_TO_DEVICE); 1318 dprintk("RPC: %s: posting %d s/g entries\n", 1319 __func__, send_wr.num_sge); 1320 1321 if (DECR_CQCOUNT(ep) > 0) 1322 send_wr.send_flags = 0; 1323 else { /* Provider must take a send completion every now and then */ 1324 INIT_CQCOUNT(ep); 1325 send_wr.send_flags = IB_SEND_SIGNALED; 1326 } 1327 1328 rc = ib_post_send(ia->ri_id->qp, &send_wr, &send_wr_fail); 1329 if (rc) 1330 dprintk("RPC: %s: ib_post_send returned %i\n", __func__, 1331 rc); 1332 out: 1333 return rc; 1334 } 1335 1336 /* 1337 * (Re)post a receive buffer. 1338 */ 1339 int 1340 rpcrdma_ep_post_recv(struct rpcrdma_ia *ia, 1341 struct rpcrdma_ep *ep, 1342 struct rpcrdma_rep *rep) 1343 { 1344 struct ib_recv_wr recv_wr, *recv_wr_fail; 1345 int rc; 1346 1347 recv_wr.next = NULL; 1348 recv_wr.wr_id = (u64) (unsigned long) rep; 1349 recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov; 1350 recv_wr.num_sge = 1; 1351 1352 ib_dma_sync_single_for_cpu(ia->ri_device, 1353 rdmab_addr(rep->rr_rdmabuf), 1354 rdmab_length(rep->rr_rdmabuf), 1355 DMA_BIDIRECTIONAL); 1356 1357 rc = ib_post_recv(ia->ri_id->qp, &recv_wr, &recv_wr_fail); 1358 1359 if (rc) 1360 dprintk("RPC: %s: ib_post_recv returned %i\n", __func__, 1361 rc); 1362 return rc; 1363 } 1364 1365 /* How many chunk list items fit within our inline buffers? 1366 */ 1367 unsigned int 1368 rpcrdma_max_segments(struct rpcrdma_xprt *r_xprt) 1369 { 1370 struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data; 1371 int bytes, segments; 1372 1373 bytes = min_t(unsigned int, cdata->inline_wsize, cdata->inline_rsize); 1374 bytes -= RPCRDMA_HDRLEN_MIN; 1375 if (bytes < sizeof(struct rpcrdma_segment) * 2) { 1376 pr_warn("RPC: %s: inline threshold too small\n", 1377 __func__); 1378 return 0; 1379 } 1380 1381 segments = 1 << (fls(bytes / sizeof(struct rpcrdma_segment)) - 1); 1382 dprintk("RPC: %s: max chunk list size = %d segments\n", 1383 __func__, segments); 1384 return segments; 1385 } 1386