net/sunrpc/xprtrdma/verbs.c

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