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