net/sunrpc/xprtrdma/verbs.c

   1 /*
   2  * Copyright (c) 2003-2007 Network Appliance, Inc. All rights reserved.
   3  *
   4  * This software is available to you under a choice of one of two
   5  * licenses.  You may choose to be licensed under the terms of the GNU
   6  * General Public License (GPL) Version 2, available from the file
   7  * COPYING in the main directory of this source tree, or the BSD-type
   8  * license below:
   9  *
  10  * Redistribution and use in source and binary forms, with or without
  11  * modification, are permitted provided that the following conditions
  12  * are met:
  13  *
  14  *      Redistributions of source code must retain the above copyright
  15  *      notice, this list of conditions and the following disclaimer.
  16  *
  17  *      Redistributions in binary form must reproduce the above
  18  *      copyright notice, this list of conditions and the following
  19  *      disclaimer in the documentation and/or other materials provided
  20  *      with the distribution.
  21  *
  22  *      Neither the name of the Network Appliance, Inc. nor the names of
  23  *      its contributors may be used to endorse or promote products
  24  *      derived from this software without specific prior written
  25  *      permission.
  26  *
  27  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  28  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  29  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  30  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  31  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  32  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  33  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  34  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  35  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  36  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  37  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  38  */
  39
  40 /*
  41  * verbs.c
  42  *
  43  * Encapsulates the major functions managing:
  44  *  o adapters
  45  *  o endpoints
  46  *  o connections
  47  *  o buffer memory
  48  */
  49
  50 #include <linux/interrupt.h>
  51 #include <linux/slab.h>
  52 #include <linux/prefetch.h>
  53 #include <linux/sunrpc/addr.h>
  54 #include <asm/bitops.h>
  55 #include <linux/module.h> /* try_module_get()/module_put() */
  56
  57 #include "xprt_rdma.h"
  58
  59 /*
  60  * Globals/Macros
  61  */
  62
  63 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
  64 # define RPCDBG_FACILITY        RPCDBG_TRANS
  65 #endif
  66
  67 /*
  68  * internal functions
  69  */
  70
  71 /*
  72  * handle replies in tasklet context, using a single, global list
  73  * rdma tasklet function -- just turn around and call the func
  74  * for all replies on the list
  75  */
  76
  77 static DEFINE_SPINLOCK(rpcrdma_tk_lock_g);
  78 static LIST_HEAD(rpcrdma_tasklets_g);
  79
  80 static void
  81 rpcrdma_run_tasklet(unsigned long data)
  82 {
  83         struct rpcrdma_rep *rep;
  84         unsigned long flags;
  85
  86         data = data;
  87         spin_lock_irqsave(&rpcrdma_tk_lock_g, flags);
  88         while (!list_empty(&rpcrdma_tasklets_g)) {
  89                 rep = list_entry(rpcrdma_tasklets_g.next,
  90                                  struct rpcrdma_rep, rr_list);
  91                 list_del(&rep->rr_list);
  92                 spin_unlock_irqrestore(&rpcrdma_tk_lock_g, flags);
  93
  94                 rpcrdma_reply_handler(rep);
  95
  96                 spin_lock_irqsave(&rpcrdma_tk_lock_g, flags);
  97         }
  98         spin_unlock_irqrestore(&rpcrdma_tk_lock_g, flags);
  99 }
 100
 101 static DECLARE_TASKLET(rpcrdma_tasklet_g, rpcrdma_run_tasklet, 0UL);
 102
 103 static void
 104 rpcrdma_schedule_tasklet(struct list_head *sched_list)
 105 {
 106         unsigned long flags;
 107
 108         spin_lock_irqsave(&rpcrdma_tk_lock_g, flags);
 109         list_splice_tail(sched_list, &rpcrdma_tasklets_g);
 110         spin_unlock_irqrestore(&rpcrdma_tk_lock_g, flags);
 111         tasklet_schedule(&rpcrdma_tasklet_g);
 112 }
 113
 114 static void
 115 rpcrdma_qp_async_error_upcall(struct ib_event *event, void *context)
 116 {
 117         struct rpcrdma_ep *ep = context;
 118
 119         pr_err("RPC:       %s: %s on device %s ep %p\n",
 120                __func__, ib_event_msg(event->event),
 121                 event->device->name, context);
 122         if (ep->rep_connected == 1) {
 123                 ep->rep_connected = -EIO;
 124                 rpcrdma_conn_func(ep);
 125                 wake_up_all(&ep->rep_connect_wait);
 126         }
 127 }
 128
 129 static void
 130 rpcrdma_cq_async_error_upcall(struct ib_event *event, void *context)
 131 {
 132         struct rpcrdma_ep *ep = context;
 133
 134         pr_err("RPC:       %s: %s on device %s ep %p\n",
 135                __func__, ib_event_msg(event->event),
 136                 event->device->name, context);
 137         if (ep->rep_connected == 1) {
 138                 ep->rep_connected = -EIO;
 139                 rpcrdma_conn_func(ep);
 140                 wake_up_all(&ep->rep_connect_wait);
 141         }
 142 }
 143
 144 static void
 145 rpcrdma_sendcq_process_wc(struct ib_wc *wc)
 146 {
 147         /* WARNING: Only wr_id and status are reliable at this point */
 148         if (wc->wr_id == RPCRDMA_IGNORE_COMPLETION) {
 149                 if (wc->status != IB_WC_SUCCESS &&
 150                     wc->status != IB_WC_WR_FLUSH_ERR)
 151                         pr_err("RPC:       %s: SEND: %s\n",
 152                                __func__, ib_wc_status_msg(wc->status));
 153         } else {
 154                 struct rpcrdma_mw *r;
 155
 156                 r = (struct rpcrdma_mw *)(unsigned long)wc->wr_id;
 157                 r->mw_sendcompletion(wc);
 158         }
 159 }
 160
 161 static int
 162 rpcrdma_sendcq_poll(struct ib_cq *cq, struct rpcrdma_ep *ep)
 163 {
 164         struct ib_wc *wcs;
 165         int budget, count, rc;
 166
 167         budget = RPCRDMA_WC_BUDGET / RPCRDMA_POLLSIZE;
 168         do {
 169                 wcs = ep->rep_send_wcs;
 170
 171                 rc = ib_poll_cq(cq, RPCRDMA_POLLSIZE, wcs);
 172                 if (rc <= 0)
 173                         return rc;
 174
 175                 count = rc;
 176                 while (count-- > 0)
 177                         rpcrdma_sendcq_process_wc(wcs++);
 178         } while (rc == RPCRDMA_POLLSIZE && --budget);
 179         return 0;
 180 }
 181
 182 /*
 183  * Handle send, fast_reg_mr, and local_inv completions.
 184  *
 185  * Send events are typically suppressed and thus do not result
 186  * in an upcall. Occasionally one is signaled, however. This
 187  * prevents the provider's completion queue from wrapping and
 188  * losing a completion.
 189  */
 190 static void
 191 rpcrdma_sendcq_upcall(struct ib_cq *cq, void *cq_context)
 192 {
 193         struct rpcrdma_ep *ep = (struct rpcrdma_ep *)cq_context;
 194         int rc;
 195
 196         rc = rpcrdma_sendcq_poll(cq, ep);
 197         if (rc) {
 198                 dprintk("RPC:       %s: ib_poll_cq failed: %i\n",
 199                         __func__, rc);
 200                 return;
 201         }
 202
 203         rc = ib_req_notify_cq(cq,
 204                         IB_CQ_NEXT_COMP | IB_CQ_REPORT_MISSED_EVENTS);
 205         if (rc == 0)
 206                 return;
 207         if (rc < 0) {
 208                 dprintk("RPC:       %s: ib_req_notify_cq failed: %i\n",
 209                         __func__, rc);
 210                 return;
 211         }
 212
 213         rpcrdma_sendcq_poll(cq, ep);
 214 }
 215
 216 static void
 217 rpcrdma_recvcq_process_wc(struct ib_wc *wc, struct list_head *sched_list)
 218 {
 219         struct rpcrdma_rep *rep =
 220                         (struct rpcrdma_rep *)(unsigned long)wc->wr_id;
 221
 222         /* WARNING: Only wr_id and status are reliable at this point */
 223         if (wc->status != IB_WC_SUCCESS)
 224                 goto out_fail;
 225
 226         /* status == SUCCESS means all fields in wc are trustworthy */
 227         if (wc->opcode != IB_WC_RECV)
 228                 return;
 229
 230         dprintk("RPC:       %s: rep %p opcode 'recv', length %u: success\n",
 231                 __func__, rep, wc->byte_len);
 232
 233         rep->rr_len = wc->byte_len;
 234         ib_dma_sync_single_for_cpu(rep->rr_device,
 235                                    rdmab_addr(rep->rr_rdmabuf),
 236                                    rep->rr_len, DMA_FROM_DEVICE);
 237         prefetch(rdmab_to_msg(rep->rr_rdmabuf));
 238
 239 out_schedule:
 240         list_add_tail(&rep->rr_list, sched_list);
 241         return;
 242 out_fail:
 243         if (wc->status != IB_WC_WR_FLUSH_ERR)
 244                 pr_err("RPC:       %s: rep %p: %s\n",
 245                        __func__, rep, ib_wc_status_msg(wc->status));
 246         rep->rr_len = ~0U;
 247         goto out_schedule;
 248 }
 249
 250 static int
 251 rpcrdma_recvcq_poll(struct ib_cq *cq, struct rpcrdma_ep *ep)
 252 {
 253         struct list_head sched_list;
 254         struct ib_wc *wcs;
 255         int budget, count, rc;
 256
 257         INIT_LIST_HEAD(&sched_list);
 258         budget = RPCRDMA_WC_BUDGET / RPCRDMA_POLLSIZE;
 259         do {
 260                 wcs = ep->rep_recv_wcs;
 261
 262                 rc = ib_poll_cq(cq, RPCRDMA_POLLSIZE, wcs);
 263                 if (rc <= 0)
 264                         goto out_schedule;
 265
 266                 count = rc;
 267                 while (count-- > 0)
 268                         rpcrdma_recvcq_process_wc(wcs++, &sched_list);
 269         } while (rc == RPCRDMA_POLLSIZE && --budget);
 270         rc = 0;
 271
 272 out_schedule:
 273         rpcrdma_schedule_tasklet(&sched_list);
 274         return rc;
 275 }
 276
 277 /*
 278  * Handle receive completions.
 279  *
 280  * It is reentrant but processes single events in order to maintain
 281  * ordering of receives to keep server credits.
 282  *
 283  * It is the responsibility of the scheduled tasklet to return
 284  * recv buffers to the pool. NOTE: this affects synchronization of
 285  * connection shutdown. That is, the structures required for
 286  * the completion of the reply handler must remain intact until
 287  * all memory has been reclaimed.
 288  */
 289 static void
 290 rpcrdma_recvcq_upcall(struct ib_cq *cq, void *cq_context)
 291 {
 292         struct rpcrdma_ep *ep = (struct rpcrdma_ep *)cq_context;
 293         int rc;
 294
 295         rc = rpcrdma_recvcq_poll(cq, ep);
 296         if (rc) {
 297                 dprintk("RPC:       %s: ib_poll_cq failed: %i\n",
 298                         __func__, rc);
 299                 return;
 300         }
 301
 302         rc = ib_req_notify_cq(cq,
 303                         IB_CQ_NEXT_COMP | IB_CQ_REPORT_MISSED_EVENTS);
 304         if (rc == 0)
 305                 return;
 306         if (rc < 0) {
 307                 dprintk("RPC:       %s: ib_req_notify_cq failed: %i\n",
 308                         __func__, rc);
 309                 return;
 310         }
 311
 312         rpcrdma_recvcq_poll(cq, ep);
 313 }
 314
 315 static void
 316 rpcrdma_flush_cqs(struct rpcrdma_ep *ep)
 317 {
 318         struct ib_wc wc;
 319         LIST_HEAD(sched_list);
 320
 321         while (ib_poll_cq(ep->rep_attr.recv_cq, 1, &wc) > 0)
 322                 rpcrdma_recvcq_process_wc(&wc, &sched_list);
 323         if (!list_empty(&sched_list))
 324                 rpcrdma_schedule_tasklet(&sched_list);
 325         while (ib_poll_cq(ep->rep_attr.send_cq, 1, &wc) > 0)
 326                 rpcrdma_sendcq_process_wc(&wc);
 327 }
 328
 329 static int
 330 rpcrdma_conn_upcall(struct rdma_cm_id *id, struct rdma_cm_event *event)
 331 {
 332         struct rpcrdma_xprt *xprt = id->context;
 333         struct rpcrdma_ia *ia = &xprt->rx_ia;
 334         struct rpcrdma_ep *ep = &xprt->rx_ep;
 335 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
 336         struct sockaddr *sap = (struct sockaddr *)&ep->rep_remote_addr;
 337 #endif
 338         struct ib_qp_attr *attr = &ia->ri_qp_attr;
 339         struct ib_qp_init_attr *iattr = &ia->ri_qp_init_attr;
 340         int connstate = 0;
 341
 342         switch (event->event) {
 343         case RDMA_CM_EVENT_ADDR_RESOLVED:
 344         case RDMA_CM_EVENT_ROUTE_RESOLVED:
 345                 ia->ri_async_rc = 0;
 346                 complete(&ia->ri_done);
 347                 break;
 348         case RDMA_CM_EVENT_ADDR_ERROR:
 349                 ia->ri_async_rc = -EHOSTUNREACH;
 350                 dprintk("RPC:       %s: CM address resolution error, ep 0x%p\n",
 351                         __func__, ep);
 352                 complete(&ia->ri_done);
 353                 break;
 354         case RDMA_CM_EVENT_ROUTE_ERROR:
 355                 ia->ri_async_rc = -ENETUNREACH;
 356                 dprintk("RPC:       %s: CM route resolution error, ep 0x%p\n",
 357                         __func__, ep);
 358                 complete(&ia->ri_done);
 359                 break;
 360         case RDMA_CM_EVENT_ESTABLISHED:
 361                 connstate = 1;
 362                 ib_query_qp(ia->ri_id->qp, attr,
 363                             IB_QP_MAX_QP_RD_ATOMIC | IB_QP_MAX_DEST_RD_ATOMIC,
 364                             iattr);
 365                 dprintk("RPC:       %s: %d responder resources"
 366                         " (%d initiator)\n",
 367                         __func__, attr->max_dest_rd_atomic,
 368                         attr->max_rd_atomic);
 369                 goto connected;
 370         case RDMA_CM_EVENT_CONNECT_ERROR:
 371                 connstate = -ENOTCONN;
 372                 goto connected;
 373         case RDMA_CM_EVENT_UNREACHABLE:
 374                 connstate = -ENETDOWN;
 375                 goto connected;
 376         case RDMA_CM_EVENT_REJECTED:
 377                 connstate = -ECONNREFUSED;
 378                 goto connected;
 379         case RDMA_CM_EVENT_DISCONNECTED:
 380                 connstate = -ECONNABORTED;
 381                 goto connected;
 382         case RDMA_CM_EVENT_DEVICE_REMOVAL:
 383                 connstate = -ENODEV;
 384 connected:
 385                 dprintk("RPC:       %s: %sconnected\n",
 386                                         __func__, connstate > 0 ? "" : "dis");
 387                 ep->rep_connected = connstate;
 388                 rpcrdma_conn_func(ep);
 389                 wake_up_all(&ep->rep_connect_wait);
 390                 /*FALLTHROUGH*/
 391         default:
 392                 dprintk("RPC:       %s: %pIS:%u (ep 0x%p): %s\n",
 393                         __func__, sap, rpc_get_port(sap), ep,
 394                         rdma_event_msg(event->event));
 395                 break;
 396         }
 397
 398 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
 399         if (connstate == 1) {
 400                 int ird = attr->max_dest_rd_atomic;
 401                 int tird = ep->rep_remote_cma.responder_resources;
 402
 403                 pr_info("rpcrdma: connection to %pIS:%u on %s, memreg '%s', %d credits, %d responders%s\n",
 404                         sap, rpc_get_port(sap),
 405                         ia->ri_device->name,
 406                         ia->ri_ops->ro_displayname,
 407                         xprt->rx_buf.rb_max_requests,
 408                         ird, ird < 4 && ird < tird / 2 ? " (low!)" : "");
 409         } else if (connstate < 0) {
 410                 pr_info("rpcrdma: connection to %pIS:%u closed (%d)\n",
 411                         sap, rpc_get_port(sap), connstate);
 412         }
 413 #endif
 414
 415         return 0;
 416 }
 417
 418 static void rpcrdma_destroy_id(struct rdma_cm_id *id)
 419 {
 420         if (id) {
 421                 module_put(id->device->owner);
 422                 rdma_destroy_id(id);
 423         }
 424 }
 425
 426 static struct rdma_cm_id *
 427 rpcrdma_create_id(struct rpcrdma_xprt *xprt,
 428                         struct rpcrdma_ia *ia, struct sockaddr *addr)
 429 {
 430         struct rdma_cm_id *id;
 431         int rc;
 432
 433         init_completion(&ia->ri_done);
 434
 435         id = rdma_create_id(rpcrdma_conn_upcall, xprt, RDMA_PS_TCP, IB_QPT_RC);
 436         if (IS_ERR(id)) {
 437                 rc = PTR_ERR(id);
 438                 dprintk("RPC:       %s: rdma_create_id() failed %i\n",
 439                         __func__, rc);
 440                 return id;
 441         }
 442
 443         ia->ri_async_rc = -ETIMEDOUT;
 444         rc = rdma_resolve_addr(id, NULL, addr, RDMA_RESOLVE_TIMEOUT);
 445         if (rc) {
 446                 dprintk("RPC:       %s: rdma_resolve_addr() failed %i\n",
 447                         __func__, rc);
 448                 goto out;
 449         }
 450         wait_for_completion_interruptible_timeout(&ia->ri_done,
 451                                 msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1);
 452
 453         /* FIXME:
 454          * Until xprtrdma supports DEVICE_REMOVAL, the provider must
 455          * be pinned while there are active NFS/RDMA mounts to prevent
 456          * hangs and crashes at umount time.
 457          */
 458         if (!ia->ri_async_rc && !try_module_get(id->device->owner)) {
 459                 dprintk("RPC:       %s: Failed to get device module\n",
 460                         __func__);
 461                 ia->ri_async_rc = -ENODEV;
 462         }
 463         rc = ia->ri_async_rc;
 464         if (rc)
 465                 goto out;
 466
 467         ia->ri_async_rc = -ETIMEDOUT;
 468         rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT);
 469         if (rc) {
 470                 dprintk("RPC:       %s: rdma_resolve_route() failed %i\n",
 471                         __func__, rc);
 472                 goto put;
 473         }
 474         wait_for_completion_interruptible_timeout(&ia->ri_done,
 475                                 msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1);
 476         rc = ia->ri_async_rc;
 477         if (rc)
 478                 goto put;
 479
 480         return id;
 481 put:
 482         module_put(id->device->owner);
 483 out:
 484         rdma_destroy_id(id);
 485         return ERR_PTR(rc);
 486 }
 487
 488 /*
 489  * Drain any cq, prior to teardown.
 490  */
 491 static void
 492 rpcrdma_clean_cq(struct ib_cq *cq)
 493 {
 494         struct ib_wc wc;
 495         int count = 0;
 496
 497         while (1 == ib_poll_cq(cq, 1, &wc))
 498                 ++count;
 499
 500         if (count)
 501                 dprintk("RPC:       %s: flushed %d events (last 0x%x)\n",
 502                         __func__, count, wc.opcode);
 503 }
 504
 505 /*
 506  * Exported functions.
 507  */
 508
 509 /*
 510  * Open and initialize an Interface Adapter.
 511  *  o initializes fields of struct rpcrdma_ia, including
 512  *    interface and provider attributes and protection zone.
 513  */
 514 int
 515 rpcrdma_ia_open(struct rpcrdma_xprt *xprt, struct sockaddr *addr, int memreg)
 516 {
 517         struct rpcrdma_ia *ia = &xprt->rx_ia;
 518         struct ib_device_attr *devattr = &ia->ri_devattr;
 519         int rc;
 520
 521         ia->ri_dma_mr = NULL;
 522
 523         ia->ri_id = rpcrdma_create_id(xprt, ia, addr);
 524         if (IS_ERR(ia->ri_id)) {
 525                 rc = PTR_ERR(ia->ri_id);
 526                 goto out1;
 527         }
 528         ia->ri_device = ia->ri_id->device;
 529
 530         ia->ri_pd = ib_alloc_pd(ia->ri_device);
 531         if (IS_ERR(ia->ri_pd)) {
 532                 rc = PTR_ERR(ia->ri_pd);
 533                 dprintk("RPC:       %s: ib_alloc_pd() failed %i\n",
 534                         __func__, rc);
 535                 goto out2;
 536         }
 537
 538         rc = ib_query_device(ia->ri_device, devattr);
 539         if (rc) {
 540                 dprintk("RPC:       %s: ib_query_device failed %d\n",
 541                         __func__, rc);
 542                 goto out3;
 543         }
 544
 545         if (memreg == RPCRDMA_FRMR) {
 546                 if (!(devattr->device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS) ||
 547                     (devattr->max_fast_reg_page_list_len == 0)) {
 548                         dprintk("RPC:       %s: FRMR registration "
 549                                 "not supported by HCA\n", __func__);
 550                         memreg = RPCRDMA_MTHCAFMR;
 551                 }
 552         }
 553         if (memreg == RPCRDMA_MTHCAFMR) {
 554                 if (!ia->ri_device->alloc_fmr) {
 555                         dprintk("RPC:       %s: MTHCAFMR registration "
 556                                 "not supported by HCA\n", __func__);
 557                         rc = -EINVAL;
 558                         goto out3;
 559                 }
 560         }
 561
 562         switch (memreg) {
 563         case RPCRDMA_FRMR:
 564                 ia->ri_ops = &rpcrdma_frwr_memreg_ops;
 565                 break;
 566         case RPCRDMA_ALLPHYSICAL:
 567                 ia->ri_ops = &rpcrdma_physical_memreg_ops;
 568                 break;
 569         case RPCRDMA_MTHCAFMR:
 570                 ia->ri_ops = &rpcrdma_fmr_memreg_ops;
 571                 break;
 572         default:
 573                 printk(KERN_ERR "RPC: Unsupported memory "
 574                                 "registration mode: %d\n", memreg);
 575                 rc = -ENOMEM;
 576                 goto out3;
 577         }
 578         dprintk("RPC:       %s: memory registration strategy is '%s'\n",
 579                 __func__, ia->ri_ops->ro_displayname);
 580
 581         rwlock_init(&ia->ri_qplock);
 582         return 0;
 583
 584 out3:
 585         ib_dealloc_pd(ia->ri_pd);
 586         ia->ri_pd = NULL;
 587 out2:
 588         rpcrdma_destroy_id(ia->ri_id);
 589         ia->ri_id = NULL;
 590 out1:
 591         return rc;
 592 }
 593
 594 /*
 595  * Clean up/close an IA.
 596  *   o if event handles and PD have been initialized, free them.
 597  *   o close the IA
 598  */
 599 void
 600 rpcrdma_ia_close(struct rpcrdma_ia *ia)
 601 {
 602         dprintk("RPC:       %s: entering\n", __func__);
 603         if (ia->ri_id != NULL && !IS_ERR(ia->ri_id)) {
 604                 if (ia->ri_id->qp)
 605                         rdma_destroy_qp(ia->ri_id);
 606                 rpcrdma_destroy_id(ia->ri_id);
 607                 ia->ri_id = NULL;
 608         }
 609
 610         /* If the pd is still busy, xprtrdma missed freeing a resource */
 611         if (ia->ri_pd && !IS_ERR(ia->ri_pd))
 612                 ib_dealloc_pd(ia->ri_pd);
 613 }
 614
 615 /*
 616  * Create unconnected endpoint.
 617  */
 618 int
 619 rpcrdma_ep_create(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia,
 620                                 struct rpcrdma_create_data_internal *cdata)
 621 {
 622         struct ib_device_attr *devattr = &ia->ri_devattr;
 623         struct ib_cq *sendcq, *recvcq;
 624         struct ib_cq_init_attr cq_attr = {};
 625         int rc, err;
 626
 627         if (devattr->max_sge < RPCRDMA_MAX_IOVS) {
 628                 dprintk("RPC:       %s: insufficient sge's available\n",
 629                         __func__);
 630                 return -ENOMEM;
 631         }
 632
 633         /* check provider's send/recv wr limits */
 634         if (cdata->max_requests > devattr->max_qp_wr)
 635                 cdata->max_requests = devattr->max_qp_wr;
 636
 637         ep->rep_attr.event_handler = rpcrdma_qp_async_error_upcall;
 638         ep->rep_attr.qp_context = ep;
 639         ep->rep_attr.srq = NULL;
 640         ep->rep_attr.cap.max_send_wr = cdata->max_requests;
 641         rc = ia->ri_ops->ro_open(ia, ep, cdata);
 642         if (rc)
 643                 return rc;
 644         ep->rep_attr.cap.max_recv_wr = cdata->max_requests;
 645         ep->rep_attr.cap.max_send_sge = RPCRDMA_MAX_IOVS;
 646         ep->rep_attr.cap.max_recv_sge = 1;
 647         ep->rep_attr.cap.max_inline_data = 0;
 648         ep->rep_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
 649         ep->rep_attr.qp_type = IB_QPT_RC;
 650         ep->rep_attr.port_num = ~0;
 651
 652         dprintk("RPC:       %s: requested max: dtos: send %d recv %d; "
 653                 "iovs: send %d recv %d\n",
 654                 __func__,
 655                 ep->rep_attr.cap.max_send_wr,
 656                 ep->rep_attr.cap.max_recv_wr,
 657                 ep->rep_attr.cap.max_send_sge,
 658                 ep->rep_attr.cap.max_recv_sge);
 659
 660         /* set trigger for requesting send completion */
 661         ep->rep_cqinit = ep->rep_attr.cap.max_send_wr/2 - 1;
 662         if (ep->rep_cqinit > RPCRDMA_MAX_UNSIGNALED_SENDS)
 663                 ep->rep_cqinit = RPCRDMA_MAX_UNSIGNALED_SENDS;
 664         else if (ep->rep_cqinit <= 2)
 665                 ep->rep_cqinit = 0;
 666         INIT_CQCOUNT(ep);
 667         init_waitqueue_head(&ep->rep_connect_wait);
 668         INIT_DELAYED_WORK(&ep->rep_connect_worker, rpcrdma_connect_worker);
 669
 670         cq_attr.cqe = ep->rep_attr.cap.max_send_wr + 1;
 671         sendcq = ib_create_cq(ia->ri_device, rpcrdma_sendcq_upcall,
 672                               rpcrdma_cq_async_error_upcall, ep, &cq_attr);
 673         if (IS_ERR(sendcq)) {
 674                 rc = PTR_ERR(sendcq);
 675                 dprintk("RPC:       %s: failed to create send CQ: %i\n",
 676                         __func__, rc);
 677                 goto out1;
 678         }
 679
 680         rc = ib_req_notify_cq(sendcq, IB_CQ_NEXT_COMP);
 681         if (rc) {
 682                 dprintk("RPC:       %s: ib_req_notify_cq failed: %i\n",
 683                         __func__, rc);
 684                 goto out2;
 685         }
 686
 687         cq_attr.cqe = ep->rep_attr.cap.max_recv_wr + 1;
 688         recvcq = ib_create_cq(ia->ri_device, rpcrdma_recvcq_upcall,
 689                               rpcrdma_cq_async_error_upcall, ep, &cq_attr);
 690         if (IS_ERR(recvcq)) {
 691                 rc = PTR_ERR(recvcq);
 692                 dprintk("RPC:       %s: failed to create recv CQ: %i\n",
 693                         __func__, rc);
 694                 goto out2;
 695         }
 696
 697         rc = ib_req_notify_cq(recvcq, IB_CQ_NEXT_COMP);
 698         if (rc) {
 699                 dprintk("RPC:       %s: ib_req_notify_cq failed: %i\n",
 700                         __func__, rc);
 701                 ib_destroy_cq(recvcq);
 702                 goto out2;
 703         }
 704
 705         ep->rep_attr.send_cq = sendcq;
 706         ep->rep_attr.recv_cq = recvcq;
 707
 708         /* Initialize cma parameters */
 709
 710         /* RPC/RDMA does not use private data */
 711         ep->rep_remote_cma.private_data = NULL;
 712         ep->rep_remote_cma.private_data_len = 0;
 713
 714         /* Client offers RDMA Read but does not initiate */
 715         ep->rep_remote_cma.initiator_depth = 0;
 716         if (devattr->max_qp_rd_atom > 32)       /* arbitrary but <= 255 */
 717                 ep->rep_remote_cma.responder_resources = 32;
 718         else
 719                 ep->rep_remote_cma.responder_resources =
 720                                                 devattr->max_qp_rd_atom;
 721
 722         ep->rep_remote_cma.retry_count = 7;
 723         ep->rep_remote_cma.flow_control = 0;
 724         ep->rep_remote_cma.rnr_retry_count = 0;
 725
 726         return 0;
 727
 728 out2:
 729         err = ib_destroy_cq(sendcq);
 730         if (err)
 731                 dprintk("RPC:       %s: ib_destroy_cq returned %i\n",
 732                         __func__, err);
 733 out1:
 734         if (ia->ri_dma_mr)
 735                 ib_dereg_mr(ia->ri_dma_mr);
 736         return rc;
 737 }
 738
 739 /*
 740  * rpcrdma_ep_destroy
 741  *
 742  * Disconnect and destroy endpoint. After this, the only
 743  * valid operations on the ep are to free it (if dynamically
 744  * allocated) or re-create it.
 745  */
 746 void
 747 rpcrdma_ep_destroy(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
 748 {
 749         int rc;
 750
 751         dprintk("RPC:       %s: entering, connected is %d\n",
 752                 __func__, ep->rep_connected);
 753
 754         cancel_delayed_work_sync(&ep->rep_connect_worker);
 755
 756         if (ia->ri_id->qp)
 757                 rpcrdma_ep_disconnect(ep, ia);
 758
 759         rpcrdma_clean_cq(ep->rep_attr.recv_cq);
 760         rpcrdma_clean_cq(ep->rep_attr.send_cq);
 761
 762         if (ia->ri_id->qp) {
 763                 rdma_destroy_qp(ia->ri_id);
 764                 ia->ri_id->qp = NULL;
 765         }
 766
 767         rc = ib_destroy_cq(ep->rep_attr.recv_cq);
 768         if (rc)
 769                 dprintk("RPC:       %s: ib_destroy_cq returned %i\n",
 770                         __func__, rc);
 771
 772         rc = ib_destroy_cq(ep->rep_attr.send_cq);
 773         if (rc)
 774                 dprintk("RPC:       %s: ib_destroy_cq returned %i\n",
 775                         __func__, rc);
 776
 777         if (ia->ri_dma_mr) {
 778                 rc = ib_dereg_mr(ia->ri_dma_mr);
 779                 dprintk("RPC:       %s: ib_dereg_mr returned %i\n",
 780                         __func__, rc);
 781         }
 782 }
 783
 784 /*
 785  * Connect unconnected endpoint.
 786  */
 787 int
 788 rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
 789 {
 790         struct rdma_cm_id *id, *old;
 791         int rc = 0;
 792         int retry_count = 0;
 793
 794         if (ep->rep_connected != 0) {
 795                 struct rpcrdma_xprt *xprt;
 796 retry:
 797                 dprintk("RPC:       %s: reconnecting...\n", __func__);
 798
 799                 rpcrdma_ep_disconnect(ep, ia);
 800                 rpcrdma_flush_cqs(ep);
 801
 802                 xprt = container_of(ia, struct rpcrdma_xprt, rx_ia);
 803                 id = rpcrdma_create_id(xprt, ia,
 804                                 (struct sockaddr *)&xprt->rx_data.addr);
 805                 if (IS_ERR(id)) {
 806                         rc = -EHOSTUNREACH;
 807                         goto out;
 808                 }
 809                 /* TEMP TEMP TEMP - fail if new device:
 810                  * Deregister/remarshal *all* requests!
 811                  * Close and recreate adapter, pd, etc!
 812                  * Re-determine all attributes still sane!
 813                  * More stuff I haven't thought of!
 814                  * Rrrgh!
 815                  */
 816                 if (ia->ri_device != id->device) {
 817                         printk("RPC:       %s: can't reconnect on "
 818                                 "different device!\n", __func__);
 819                         rpcrdma_destroy_id(id);
 820                         rc = -ENETUNREACH;
 821                         goto out;
 822                 }
 823                 /* END TEMP */
 824                 rc = rdma_create_qp(id, ia->ri_pd, &ep->rep_attr);
 825                 if (rc) {
 826                         dprintk("RPC:       %s: rdma_create_qp failed %i\n",
 827                                 __func__, rc);
 828                         rpcrdma_destroy_id(id);
 829                         rc = -ENETUNREACH;
 830                         goto out;
 831                 }
 832
 833                 write_lock(&ia->ri_qplock);
 834                 old = ia->ri_id;
 835                 ia->ri_id = id;
 836                 write_unlock(&ia->ri_qplock);
 837
 838                 rdma_destroy_qp(old);
 839                 rpcrdma_destroy_id(old);
 840         } else {
 841                 dprintk("RPC:       %s: connecting...\n", __func__);
 842                 rc = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr);
 843                 if (rc) {
 844                         dprintk("RPC:       %s: rdma_create_qp failed %i\n",
 845                                 __func__, rc);
 846                         /* do not update ep->rep_connected */
 847                         return -ENETUNREACH;
 848                 }
 849         }
 850
 851         ep->rep_connected = 0;
 852
 853         rc = rdma_connect(ia->ri_id, &ep->rep_remote_cma);
 854         if (rc) {
 855                 dprintk("RPC:       %s: rdma_connect() failed with %i\n",
 856                                 __func__, rc);
 857                 goto out;
 858         }
 859
 860         wait_event_interruptible(ep->rep_connect_wait, ep->rep_connected != 0);
 861
 862         /*
 863          * Check state. A non-peer reject indicates no listener
 864          * (ECONNREFUSED), which may be a transient state. All
 865          * others indicate a transport condition which has already
 866          * undergone a best-effort.
 867          */
 868         if (ep->rep_connected == -ECONNREFUSED &&
 869             ++retry_count <= RDMA_CONNECT_RETRY_MAX) {
 870                 dprintk("RPC:       %s: non-peer_reject, retry\n", __func__);
 871                 goto retry;
 872         }
 873         if (ep->rep_connected <= 0) {
 874                 /* Sometimes, the only way to reliably connect to remote
 875                  * CMs is to use same nonzero values for ORD and IRD. */
 876                 if (retry_count++ <= RDMA_CONNECT_RETRY_MAX + 1 &&
 877                     (ep->rep_remote_cma.responder_resources == 0 ||
 878                      ep->rep_remote_cma.initiator_depth !=
 879                                 ep->rep_remote_cma.responder_resources)) {
 880                         if (ep->rep_remote_cma.responder_resources == 0)
 881                                 ep->rep_remote_cma.responder_resources = 1;
 882                         ep->rep_remote_cma.initiator_depth =
 883                                 ep->rep_remote_cma.responder_resources;
 884                         goto retry;
 885                 }
 886                 rc = ep->rep_connected;
 887         } else {
 888                 dprintk("RPC:       %s: connected\n", __func__);
 889         }
 890
 891 out:
 892         if (rc)
 893                 ep->rep_connected = rc;
 894         return rc;
 895 }
 896
 897 /*
 898  * rpcrdma_ep_disconnect
 899  *
 900  * This is separate from destroy to facilitate the ability
 901  * to reconnect without recreating the endpoint.
 902  *
 903  * This call is not reentrant, and must not be made in parallel
 904  * on the same endpoint.
 905  */
 906 void
 907 rpcrdma_ep_disconnect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
 908 {
 909         int rc;
 910
 911         rpcrdma_flush_cqs(ep);
 912         rc = rdma_disconnect(ia->ri_id);
 913         if (!rc) {
 914                 /* returns without wait if not connected */
 915                 wait_event_interruptible(ep->rep_connect_wait,
 916                                                         ep->rep_connected != 1);
 917                 dprintk("RPC:       %s: after wait, %sconnected\n", __func__,
 918                         (ep->rep_connected == 1) ? "still " : "dis");
 919         } else {
 920                 dprintk("RPC:       %s: rdma_disconnect %i\n", __func__, rc);
 921                 ep->rep_connected = rc;
 922         }
 923 }
 924
 925 static struct rpcrdma_req *
 926 rpcrdma_create_req(struct rpcrdma_xprt *r_xprt)
 927 {
 928         struct rpcrdma_req *req;
 929
 930         req = kzalloc(sizeof(*req), GFP_KERNEL);
 931         if (req == NULL)
 932                 return ERR_PTR(-ENOMEM);
 933
 934         req->rl_buffer = &r_xprt->rx_buf;
 935         return req;
 936 }
 937
 938 static struct rpcrdma_rep *
 939 rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt)
 940 {
 941         struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
 942         struct rpcrdma_ia *ia = &r_xprt->rx_ia;
 943         struct rpcrdma_rep *rep;
 944         int rc;
 945
 946         rc = -ENOMEM;
 947         rep = kzalloc(sizeof(*rep), GFP_KERNEL);
 948         if (rep == NULL)
 949                 goto out;
 950
 951         rep->rr_rdmabuf = rpcrdma_alloc_regbuf(ia, cdata->inline_rsize,
 952                                                GFP_KERNEL);
 953         if (IS_ERR(rep->rr_rdmabuf)) {
 954                 rc = PTR_ERR(rep->rr_rdmabuf);
 955                 goto out_free;
 956         }
 957
 958         rep->rr_device = ia->ri_device;
 959         rep->rr_rxprt = r_xprt;
 960         return rep;
 961
 962 out_free:
 963         kfree(rep);
 964 out:
 965         return ERR_PTR(rc);
 966 }
 967
 968 int
 969 rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt)
 970 {
 971         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
 972         struct rpcrdma_ia *ia = &r_xprt->rx_ia;
 973         struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
 974         char *p;
 975         size_t len;
 976         int i, rc;
 977
 978         buf->rb_max_requests = cdata->max_requests;
 979         spin_lock_init(&buf->rb_lock);
 980
 981         /* Need to allocate:
 982          *   1.  arrays for send and recv pointers
 983          *   2.  arrays of struct rpcrdma_req to fill in pointers
 984          *   3.  array of struct rpcrdma_rep for replies
 985          * Send/recv buffers in req/rep need to be registered
 986          */
 987         len = buf->rb_max_requests *
 988                 (sizeof(struct rpcrdma_req *) + sizeof(struct rpcrdma_rep *));
 989
 990         p = kzalloc(len, GFP_KERNEL);
 991         if (p == NULL) {
 992                 dprintk("RPC:       %s: req_t/rep_t/pad kzalloc(%zd) failed\n",
 993                         __func__, len);
 994                 rc = -ENOMEM;
 995                 goto out;
 996         }
 997         buf->rb_pool = p;       /* for freeing it later */
 998
 999         buf->rb_send_bufs = (struct rpcrdma_req **) p;
1000         p = (char *) &buf->rb_send_bufs[buf->rb_max_requests];
1001         buf->rb_recv_bufs = (struct rpcrdma_rep **) p;
1002         p = (char *) &buf->rb_recv_bufs[buf->rb_max_requests];
1003
1004         rc = ia->ri_ops->ro_init(r_xprt);
1005         if (rc)
1006                 goto out;
1007
1008         for (i = 0; i < buf->rb_max_requests; i++) {
1009                 struct rpcrdma_req *req;
1010                 struct rpcrdma_rep *rep;
1011
1012                 req = rpcrdma_create_req(r_xprt);
1013                 if (IS_ERR(req)) {
1014                         dprintk("RPC:       %s: request buffer %d alloc"
1015                                 " failed\n", __func__, i);
1016                         rc = PTR_ERR(req);
1017                         goto out;
1018                 }
1019                 buf->rb_send_bufs[i] = req;
1020
1021                 rep = rpcrdma_create_rep(r_xprt);
1022                 if (IS_ERR(rep)) {
1023                         dprintk("RPC:       %s: reply buffer %d alloc failed\n",
1024                                 __func__, i);
1025                         rc = PTR_ERR(rep);
1026                         goto out;
1027                 }
1028                 buf->rb_recv_bufs[i] = rep;
1029         }
1030
1031         return 0;
1032 out:
1033         rpcrdma_buffer_destroy(buf);
1034         return rc;
1035 }
1036
1037 static void
1038 rpcrdma_destroy_rep(struct rpcrdma_ia *ia, struct rpcrdma_rep *rep)
1039 {
1040         if (!rep)
1041                 return;
1042
1043         rpcrdma_free_regbuf(ia, rep->rr_rdmabuf);
1044         kfree(rep);
1045 }
1046
1047 static void
1048 rpcrdma_destroy_req(struct rpcrdma_ia *ia, struct rpcrdma_req *req)
1049 {
1050         if (!req)
1051                 return;
1052
1053         rpcrdma_free_regbuf(ia, req->rl_sendbuf);
1054         rpcrdma_free_regbuf(ia, req->rl_rdmabuf);
1055         kfree(req);
1056 }
1057
1058 void
1059 rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf)
1060 {
1061         struct rpcrdma_ia *ia = rdmab_to_ia(buf);
1062         int i;
1063
1064         /* clean up in reverse order from create
1065          *   1.  recv mr memory (mr free, then kfree)
1066          *   2.  send mr memory (mr free, then kfree)
1067          *   3.  MWs
1068          */
1069         dprintk("RPC:       %s: entering\n", __func__);
1070
1071         for (i = 0; i < buf->rb_max_requests; i++) {
1072                 if (buf->rb_recv_bufs)
1073                         rpcrdma_destroy_rep(ia, buf->rb_recv_bufs[i]);
1074                 if (buf->rb_send_bufs)
1075                         rpcrdma_destroy_req(ia, buf->rb_send_bufs[i]);
1076         }
1077
1078         ia->ri_ops->ro_destroy(buf);
1079
1080         kfree(buf->rb_pool);
1081 }
1082
1083 struct rpcrdma_mw *
1084 rpcrdma_get_mw(struct rpcrdma_xprt *r_xprt)
1085 {
1086         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1087         struct rpcrdma_mw *mw = NULL;
1088
1089         spin_lock(&buf->rb_mwlock);
1090         if (!list_empty(&buf->rb_mws)) {
1091                 mw = list_first_entry(&buf->rb_mws,
1092                                       struct rpcrdma_mw, mw_list);
1093                 list_del_init(&mw->mw_list);
1094         }
1095         spin_unlock(&buf->rb_mwlock);
1096
1097         if (!mw)
1098                 pr_err("RPC:       %s: no MWs available\n", __func__);
1099         return mw;
1100 }
1101
1102 void
1103 rpcrdma_put_mw(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mw *mw)
1104 {
1105         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1106
1107         spin_lock(&buf->rb_mwlock);
1108         list_add_tail(&mw->mw_list, &buf->rb_mws);
1109         spin_unlock(&buf->rb_mwlock);
1110 }
1111
1112 static void
1113 rpcrdma_buffer_put_sendbuf(struct rpcrdma_req *req, struct rpcrdma_buffer *buf)
1114 {
1115         buf->rb_send_bufs[--buf->rb_send_index] = req;
1116         req->rl_niovs = 0;
1117         if (req->rl_reply) {
1118                 buf->rb_recv_bufs[--buf->rb_recv_index] = req->rl_reply;
1119                 req->rl_reply = NULL;
1120         }
1121 }
1122
1123 /*
1124  * Get a set of request/reply buffers.
1125  *
1126  * Reply buffer (if needed) is attached to send buffer upon return.
1127  * Rule:
1128  *    rb_send_index and rb_recv_index MUST always be pointing to the
1129  *    *next* available buffer (non-NULL). They are incremented after
1130  *    removing buffers, and decremented *before* returning them.
1131  */
1132 struct rpcrdma_req *
1133 rpcrdma_buffer_get(struct rpcrdma_buffer *buffers)
1134 {
1135         struct rpcrdma_req *req;
1136         unsigned long flags;
1137
1138         spin_lock_irqsave(&buffers->rb_lock, flags);
1139
1140         if (buffers->rb_send_index == buffers->rb_max_requests) {
1141                 spin_unlock_irqrestore(&buffers->rb_lock, flags);
1142                 dprintk("RPC:       %s: out of request buffers\n", __func__);
1143                 return ((struct rpcrdma_req *)NULL);
1144         }
1145
1146         req = buffers->rb_send_bufs[buffers->rb_send_index];
1147         if (buffers->rb_send_index < buffers->rb_recv_index) {
1148                 dprintk("RPC:       %s: %d extra receives outstanding (ok)\n",
1149                         __func__,
1150                         buffers->rb_recv_index - buffers->rb_send_index);
1151                 req->rl_reply = NULL;
1152         } else {
1153                 req->rl_reply = buffers->rb_recv_bufs[buffers->rb_recv_index];
1154                 buffers->rb_recv_bufs[buffers->rb_recv_index++] = NULL;
1155         }
1156         buffers->rb_send_bufs[buffers->rb_send_index++] = NULL;
1157
1158         spin_unlock_irqrestore(&buffers->rb_lock, flags);
1159         return req;
1160 }
1161
1162 /*
1163  * Put request/reply buffers back into pool.
1164  * Pre-decrement counter/array index.
1165  */
1166 void
1167 rpcrdma_buffer_put(struct rpcrdma_req *req)
1168 {
1169         struct rpcrdma_buffer *buffers = req->rl_buffer;
1170         unsigned long flags;
1171
1172         spin_lock_irqsave(&buffers->rb_lock, flags);
1173         rpcrdma_buffer_put_sendbuf(req, buffers);
1174         spin_unlock_irqrestore(&buffers->rb_lock, flags);
1175 }
1176
1177 /*
1178  * Recover reply buffers from pool.
1179  * This happens when recovering from error conditions.
1180  * Post-increment counter/array index.
1181  */
1182 void
1183 rpcrdma_recv_buffer_get(struct rpcrdma_req *req)
1184 {
1185         struct rpcrdma_buffer *buffers = req->rl_buffer;
1186         unsigned long flags;
1187
1188         spin_lock_irqsave(&buffers->rb_lock, flags);
1189         if (buffers->rb_recv_index < buffers->rb_max_requests) {
1190                 req->rl_reply = buffers->rb_recv_bufs[buffers->rb_recv_index];
1191                 buffers->rb_recv_bufs[buffers->rb_recv_index++] = NULL;
1192         }
1193         spin_unlock_irqrestore(&buffers->rb_lock, flags);
1194 }
1195
1196 /*
1197  * Put reply buffers back into pool when not attached to
1198  * request. This happens in error conditions.
1199  */
1200 void
1201 rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep)
1202 {
1203         struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf;
1204         unsigned long flags;
1205
1206         spin_lock_irqsave(&buffers->rb_lock, flags);
1207         buffers->rb_recv_bufs[--buffers->rb_recv_index] = rep;
1208         spin_unlock_irqrestore(&buffers->rb_lock, flags);
1209 }
1210
1211 /*
1212  * Wrappers for internal-use kmalloc memory registration, used by buffer code.
1213  */
1214
1215 void
1216 rpcrdma_mapping_error(struct rpcrdma_mr_seg *seg)
1217 {
1218         dprintk("RPC:       map_one: offset %p iova %llx len %zu\n",
1219                 seg->mr_offset,
1220                 (unsigned long long)seg->mr_dma, seg->mr_dmalen);
1221 }
1222
1223 /**
1224  * rpcrdma_alloc_regbuf - kmalloc and register memory for SEND/RECV buffers
1225  * @ia: controlling rpcrdma_ia
1226  * @size: size of buffer to be allocated, in bytes
1227  * @flags: GFP flags
1228  *
1229  * Returns pointer to private header of an area of internally
1230  * registered memory, or an ERR_PTR. The registered buffer follows
1231  * the end of the private header.
1232  *
1233  * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for
1234  * receiving the payload of RDMA RECV operations. regbufs are not
1235  * used for RDMA READ/WRITE operations, thus are registered only for
1236  * LOCAL access.
1237  */
1238 struct rpcrdma_regbuf *
1239 rpcrdma_alloc_regbuf(struct rpcrdma_ia *ia, size_t size, gfp_t flags)
1240 {
1241         struct rpcrdma_regbuf *rb;
1242         struct ib_sge *iov;
1243
1244         rb = kmalloc(sizeof(*rb) + size, flags);
1245         if (rb == NULL)
1246                 goto out;
1247
1248         iov = &rb->rg_iov;
1249         iov->addr = ib_dma_map_single(ia->ri_device,
1250                                       (void *)rb->rg_base, size,
1251                                       DMA_BIDIRECTIONAL);
1252         if (ib_dma_mapping_error(ia->ri_device, iov->addr))
1253                 goto out_free;
1254
1255         iov->length = size;
1256         iov->lkey = ia->ri_pd->local_dma_lkey;
1257         rb->rg_size = size;
1258         rb->rg_owner = NULL;
1259         return rb;
1260
1261 out_free:
1262         kfree(rb);
1263 out:
1264         return ERR_PTR(-ENOMEM);
1265 }
1266
1267 /**
1268  * rpcrdma_free_regbuf - deregister and free registered buffer
1269  * @ia: controlling rpcrdma_ia
1270  * @rb: regbuf to be deregistered and freed
1271  */
1272 void
1273 rpcrdma_free_regbuf(struct rpcrdma_ia *ia, struct rpcrdma_regbuf *rb)
1274 {
1275         struct ib_sge *iov;
1276
1277         if (!rb)
1278                 return;
1279
1280         iov = &rb->rg_iov;
1281         ib_dma_unmap_single(ia->ri_device,
1282                             iov->addr, iov->length, DMA_BIDIRECTIONAL);
1283         kfree(rb);
1284 }
1285
1286 /*
1287  * Prepost any receive buffer, then post send.
1288  *
1289  * Receive buffer is donated to hardware, reclaimed upon recv completion.
1290  */
1291 int
1292 rpcrdma_ep_post(struct rpcrdma_ia *ia,
1293                 struct rpcrdma_ep *ep,
1294                 struct rpcrdma_req *req)
1295 {
1296         struct ib_device *device = ia->ri_device;
1297         struct ib_send_wr send_wr, *send_wr_fail;
1298         struct rpcrdma_rep *rep = req->rl_reply;
1299         struct ib_sge *iov = req->rl_send_iov;
1300         int i, rc;
1301
1302         if (rep) {
1303                 rc = rpcrdma_ep_post_recv(ia, ep, rep);
1304                 if (rc)
1305                         goto out;
1306                 req->rl_reply = NULL;
1307         }
1308
1309         send_wr.next = NULL;
1310         send_wr.wr_id = RPCRDMA_IGNORE_COMPLETION;
1311         send_wr.sg_list = iov;
1312         send_wr.num_sge = req->rl_niovs;
1313         send_wr.opcode = IB_WR_SEND;
1314
1315         for (i = 0; i < send_wr.num_sge; i++)
1316                 ib_dma_sync_single_for_device(device, iov[i].addr,
1317                                               iov[i].length, DMA_TO_DEVICE);
1318         dprintk("RPC:       %s: posting %d s/g entries\n",
1319                 __func__, send_wr.num_sge);
1320
1321         if (DECR_CQCOUNT(ep) > 0)
1322                 send_wr.send_flags = 0;
1323         else { /* Provider must take a send completion every now and then */
1324                 INIT_CQCOUNT(ep);
1325                 send_wr.send_flags = IB_SEND_SIGNALED;
1326         }
1327
1328         rc = ib_post_send(ia->ri_id->qp, &send_wr, &send_wr_fail);
1329         if (rc)
1330                 dprintk("RPC:       %s: ib_post_send returned %i\n", __func__,
1331                         rc);
1332 out:
1333         return rc;
1334 }
1335
1336 /*
1337  * (Re)post a receive buffer.
1338  */
1339 int
1340 rpcrdma_ep_post_recv(struct rpcrdma_ia *ia,
1341                      struct rpcrdma_ep *ep,
1342                      struct rpcrdma_rep *rep)
1343 {
1344         struct ib_recv_wr recv_wr, *recv_wr_fail;
1345         int rc;
1346
1347         recv_wr.next = NULL;
1348         recv_wr.wr_id = (u64) (unsigned long) rep;
1349         recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov;
1350         recv_wr.num_sge = 1;
1351
1352         ib_dma_sync_single_for_cpu(ia->ri_device,
1353                                    rdmab_addr(rep->rr_rdmabuf),
1354                                    rdmab_length(rep->rr_rdmabuf),
1355                                    DMA_BIDIRECTIONAL);
1356
1357         rc = ib_post_recv(ia->ri_id->qp, &recv_wr, &recv_wr_fail);
1358
1359         if (rc)
1360                 dprintk("RPC:       %s: ib_post_recv returned %i\n", __func__,
1361                         rc);
1362         return rc;
1363 }
1364
1365 /* How many chunk list items fit within our inline buffers?
1366  */
1367 unsigned int
1368 rpcrdma_max_segments(struct rpcrdma_xprt *r_xprt)
1369 {
1370         struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
1371         int bytes, segments;
1372
1373         bytes = min_t(unsigned int, cdata->inline_wsize, cdata->inline_rsize);
1374         bytes -= RPCRDMA_HDRLEN_MIN;
1375         if (bytes < sizeof(struct rpcrdma_segment) * 2) {
1376                 pr_warn("RPC:       %s: inline threshold too small\n",
1377                         __func__);
1378                 return 0;
1379         }
1380
1381         segments = 1 << (fls(bytes / sizeof(struct rpcrdma_segment)) - 1);
1382         dprintk("RPC:       %s: max chunk list size = %d segments\n",
1383                 __func__, segments);
1384         return segments;
1385 }