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