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 <linux/sunrpc/svc_rdma.h>
  55 #include <asm/bitops.h>
  56 #include <linux/module.h> /* try_module_get()/module_put() */
  57 #include <rdma/ib_cm.h>
  58
  59 #include "xprt_rdma.h"
  60
  61 /*
  62  * Globals/Macros
  63  */
  64
  65 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
  66 # define RPCDBG_FACILITY        RPCDBG_TRANS
  67 #endif
  68
  69 /*
  70  * internal functions
  71  */
  72 static void rpcrdma_create_mrs(struct rpcrdma_xprt *r_xprt);
  73 static void rpcrdma_destroy_mrs(struct rpcrdma_buffer *buf);
  74 static void rpcrdma_dma_unmap_regbuf(struct rpcrdma_regbuf *rb);
  75
  76 static struct workqueue_struct *rpcrdma_receive_wq;
  77
  78 int
  79 rpcrdma_alloc_wq(void)
  80 {
  81         struct workqueue_struct *recv_wq;
  82
  83         recv_wq = alloc_workqueue("xprtrdma_receive",
  84                                   WQ_MEM_RECLAIM | WQ_UNBOUND | WQ_HIGHPRI,
  85                                   0);
  86         if (!recv_wq)
  87                 return -ENOMEM;
  88
  89         rpcrdma_receive_wq = recv_wq;
  90         return 0;
  91 }
  92
  93 void
  94 rpcrdma_destroy_wq(void)
  95 {
  96         struct workqueue_struct *wq;
  97
  98         if (rpcrdma_receive_wq) {
  99                 wq = rpcrdma_receive_wq;
 100                 rpcrdma_receive_wq = NULL;
 101                 destroy_workqueue(wq);
 102         }
 103 }
 104
 105 static void
 106 rpcrdma_qp_async_error_upcall(struct ib_event *event, void *context)
 107 {
 108         struct rpcrdma_ep *ep = context;
 109
 110         pr_err("rpcrdma: %s on device %s ep %p\n",
 111                ib_event_msg(event->event), event->device->name, context);
 112
 113         if (ep->rep_connected == 1) {
 114                 ep->rep_connected = -EIO;
 115                 rpcrdma_conn_func(ep);
 116                 wake_up_all(&ep->rep_connect_wait);
 117         }
 118 }
 119
 120 /**
 121  * rpcrdma_wc_send - Invoked by RDMA provider for each polled Send WC
 122  * @cq: completion queue (ignored)
 123  * @wc: completed WR
 124  *
 125  */
 126 static void
 127 rpcrdma_wc_send(struct ib_cq *cq, struct ib_wc *wc)
 128 {
 129         /* WARNING: Only wr_cqe and status are reliable at this point */
 130         if (wc->status != IB_WC_SUCCESS && wc->status != IB_WC_WR_FLUSH_ERR)
 131                 pr_err("rpcrdma: Send: %s (%u/0x%x)\n",
 132                        ib_wc_status_msg(wc->status),
 133                        wc->status, wc->vendor_err);
 134 }
 135
 136 /* Perform basic sanity checking to avoid using garbage
 137  * to update the credit grant value.
 138  */
 139 static void
 140 rpcrdma_update_granted_credits(struct rpcrdma_rep *rep)
 141 {
 142         struct rpcrdma_msg *rmsgp = rdmab_to_msg(rep->rr_rdmabuf);
 143         struct rpcrdma_buffer *buffer = &rep->rr_rxprt->rx_buf;
 144         u32 credits;
 145
 146         if (rep->rr_len < RPCRDMA_HDRLEN_ERR)
 147                 return;
 148
 149         credits = be32_to_cpu(rmsgp->rm_credit);
 150         if (credits == 0)
 151                 credits = 1;    /* don't deadlock */
 152         else if (credits > buffer->rb_max_requests)
 153                 credits = buffer->rb_max_requests;
 154
 155         atomic_set(&buffer->rb_credits, credits);
 156 }
 157
 158 /**
 159  * rpcrdma_wc_receive - Invoked by RDMA provider for each polled Receive WC
 160  * @cq: completion queue (ignored)
 161  * @wc: completed WR
 162  *
 163  */
 164 static void
 165 rpcrdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
 166 {
 167         struct ib_cqe *cqe = wc->wr_cqe;
 168         struct rpcrdma_rep *rep = container_of(cqe, struct rpcrdma_rep,
 169                                                rr_cqe);
 170
 171         /* WARNING: Only wr_id and status are reliable at this point */
 172         if (wc->status != IB_WC_SUCCESS)
 173                 goto out_fail;
 174
 175         /* status == SUCCESS means all fields in wc are trustworthy */
 176         if (wc->opcode != IB_WC_RECV)
 177                 return;
 178
 179         dprintk("RPC:       %s: rep %p opcode 'recv', length %u: success\n",
 180                 __func__, rep, wc->byte_len);
 181
 182         rep->rr_len = wc->byte_len;
 183         rep->rr_wc_flags = wc->wc_flags;
 184         rep->rr_inv_rkey = wc->ex.invalidate_rkey;
 185
 186         ib_dma_sync_single_for_cpu(rdmab_device(rep->rr_rdmabuf),
 187                                    rdmab_addr(rep->rr_rdmabuf),
 188                                    rep->rr_len, DMA_FROM_DEVICE);
 189
 190         rpcrdma_update_granted_credits(rep);
 191
 192 out_schedule:
 193         queue_work(rpcrdma_receive_wq, &rep->rr_work);
 194         return;
 195
 196 out_fail:
 197         if (wc->status != IB_WC_WR_FLUSH_ERR)
 198                 pr_err("rpcrdma: Recv: %s (%u/0x%x)\n",
 199                        ib_wc_status_msg(wc->status),
 200                        wc->status, wc->vendor_err);
 201         rep->rr_len = RPCRDMA_BAD_LEN;
 202         goto out_schedule;
 203 }
 204
 205 static void
 206 rpcrdma_update_connect_private(struct rpcrdma_xprt *r_xprt,
 207                                struct rdma_conn_param *param)
 208 {
 209         struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
 210         const struct rpcrdma_connect_private *pmsg = param->private_data;
 211         unsigned int rsize, wsize;
 212
 213         /* Default settings for RPC-over-RDMA Version One */
 214         r_xprt->rx_ia.ri_reminv_expected = false;
 215         r_xprt->rx_ia.ri_implicit_roundup = xprt_rdma_pad_optimize;
 216         rsize = RPCRDMA_V1_DEF_INLINE_SIZE;
 217         wsize = RPCRDMA_V1_DEF_INLINE_SIZE;
 218
 219         if (pmsg &&
 220             pmsg->cp_magic == rpcrdma_cmp_magic &&
 221             pmsg->cp_version == RPCRDMA_CMP_VERSION) {
 222                 r_xprt->rx_ia.ri_reminv_expected = true;
 223                 r_xprt->rx_ia.ri_implicit_roundup = true;
 224                 rsize = rpcrdma_decode_buffer_size(pmsg->cp_send_size);
 225                 wsize = rpcrdma_decode_buffer_size(pmsg->cp_recv_size);
 226         }
 227
 228         if (rsize < cdata->inline_rsize)
 229                 cdata->inline_rsize = rsize;
 230         if (wsize < cdata->inline_wsize)
 231                 cdata->inline_wsize = wsize;
 232         dprintk("RPC:       %s: max send %u, max recv %u\n",
 233                 __func__, cdata->inline_wsize, cdata->inline_rsize);
 234         rpcrdma_set_max_header_sizes(r_xprt);
 235 }
 236
 237 static int
 238 rpcrdma_conn_upcall(struct rdma_cm_id *id, struct rdma_cm_event *event)
 239 {
 240         struct rpcrdma_xprt *xprt = id->context;
 241         struct rpcrdma_ia *ia = &xprt->rx_ia;
 242         struct rpcrdma_ep *ep = &xprt->rx_ep;
 243 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
 244         struct sockaddr *sap = (struct sockaddr *)&ep->rep_remote_addr;
 245 #endif
 246         struct ib_qp_attr *attr = &ia->ri_qp_attr;
 247         struct ib_qp_init_attr *iattr = &ia->ri_qp_init_attr;
 248         int connstate = 0;
 249
 250         switch (event->event) {
 251         case RDMA_CM_EVENT_ADDR_RESOLVED:
 252         case RDMA_CM_EVENT_ROUTE_RESOLVED:
 253                 ia->ri_async_rc = 0;
 254                 complete(&ia->ri_done);
 255                 break;
 256         case RDMA_CM_EVENT_ADDR_ERROR:
 257                 ia->ri_async_rc = -EHOSTUNREACH;
 258                 dprintk("RPC:       %s: CM address resolution error, ep 0x%p\n",
 259                         __func__, ep);
 260                 complete(&ia->ri_done);
 261                 break;
 262         case RDMA_CM_EVENT_ROUTE_ERROR:
 263                 ia->ri_async_rc = -ENETUNREACH;
 264                 dprintk("RPC:       %s: CM route resolution error, ep 0x%p\n",
 265                         __func__, ep);
 266                 complete(&ia->ri_done);
 267                 break;
 268         case RDMA_CM_EVENT_DEVICE_REMOVAL:
 269 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
 270                 pr_info("rpcrdma: removing device for %pIS:%u\n",
 271                         sap, rpc_get_port(sap));
 272 #endif
 273                 set_bit(RPCRDMA_IAF_REMOVING, &ia->ri_flags);
 274                 ep->rep_connected = -ENODEV;
 275                 xprt_force_disconnect(&xprt->rx_xprt);
 276                 wait_for_completion(&ia->ri_remove_done);
 277
 278                 ia->ri_id = NULL;
 279                 ia->ri_pd = NULL;
 280                 ia->ri_device = NULL;
 281                 /* Return 1 to ensure the core destroys the id. */
 282                 return 1;
 283         case RDMA_CM_EVENT_ESTABLISHED:
 284                 connstate = 1;
 285                 ib_query_qp(ia->ri_id->qp, attr,
 286                             IB_QP_MAX_QP_RD_ATOMIC | IB_QP_MAX_DEST_RD_ATOMIC,
 287                             iattr);
 288                 dprintk("RPC:       %s: %d responder resources"
 289                         " (%d initiator)\n",
 290                         __func__, attr->max_dest_rd_atomic,
 291                         attr->max_rd_atomic);
 292                 rpcrdma_update_connect_private(xprt, &event->param.conn);
 293                 goto connected;
 294         case RDMA_CM_EVENT_CONNECT_ERROR:
 295                 connstate = -ENOTCONN;
 296                 goto connected;
 297         case RDMA_CM_EVENT_UNREACHABLE:
 298                 connstate = -ENETDOWN;
 299                 goto connected;
 300         case RDMA_CM_EVENT_REJECTED:
 301 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
 302                 pr_info("rpcrdma: connection to %pIS:%u on %s rejected: %s\n",
 303                         sap, rpc_get_port(sap), ia->ri_device->name,
 304                         rdma_reject_msg(id, event->status));
 305 #endif
 306                 connstate = -ECONNREFUSED;
 307                 if (event->status == IB_CM_REJ_STALE_CONN)
 308                         connstate = -EAGAIN;
 309                 goto connected;
 310         case RDMA_CM_EVENT_DISCONNECTED:
 311                 connstate = -ECONNABORTED;
 312 connected:
 313                 dprintk("RPC:       %s: %sconnected\n",
 314                                         __func__, connstate > 0 ? "" : "dis");
 315                 atomic_set(&xprt->rx_buf.rb_credits, 1);
 316                 ep->rep_connected = connstate;
 317                 rpcrdma_conn_func(ep);
 318                 wake_up_all(&ep->rep_connect_wait);
 319                 /*FALLTHROUGH*/
 320         default:
 321                 dprintk("RPC:       %s: %pIS:%u (ep 0x%p): %s\n",
 322                         __func__, sap, rpc_get_port(sap), ep,
 323                         rdma_event_msg(event->event));
 324                 break;
 325         }
 326
 327 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
 328         if (connstate == 1) {
 329                 int ird = attr->max_dest_rd_atomic;
 330                 int tird = ep->rep_remote_cma.responder_resources;
 331
 332                 pr_info("rpcrdma: connection to %pIS:%u on %s, memreg '%s', %d credits, %d responders%s\n",
 333                         sap, rpc_get_port(sap),
 334                         ia->ri_device->name,
 335                         ia->ri_ops->ro_displayname,
 336                         xprt->rx_buf.rb_max_requests,
 337                         ird, ird < 4 && ird < tird / 2 ? " (low!)" : "");
 338         } else if (connstate < 0) {
 339                 pr_info("rpcrdma: connection to %pIS:%u closed (%d)\n",
 340                         sap, rpc_get_port(sap), connstate);
 341         }
 342 #endif
 343
 344         return 0;
 345 }
 346
 347 static void rpcrdma_destroy_id(struct rdma_cm_id *id)
 348 {
 349         if (id) {
 350                 module_put(id->device->owner);
 351                 rdma_destroy_id(id);
 352         }
 353 }
 354
 355 static struct rdma_cm_id *
 356 rpcrdma_create_id(struct rpcrdma_xprt *xprt,
 357                         struct rpcrdma_ia *ia, struct sockaddr *addr)
 358 {
 359         unsigned long wtimeout = msecs_to_jiffies(RDMA_RESOLVE_TIMEOUT) + 1;
 360         struct rdma_cm_id *id;
 361         int rc;
 362
 363         init_completion(&ia->ri_done);
 364         init_completion(&ia->ri_remove_done);
 365
 366         id = rdma_create_id(&init_net, rpcrdma_conn_upcall, xprt, RDMA_PS_TCP,
 367                             IB_QPT_RC);
 368         if (IS_ERR(id)) {
 369                 rc = PTR_ERR(id);
 370                 dprintk("RPC:       %s: rdma_create_id() failed %i\n",
 371                         __func__, rc);
 372                 return id;
 373         }
 374
 375         ia->ri_async_rc = -ETIMEDOUT;
 376         rc = rdma_resolve_addr(id, NULL, addr, RDMA_RESOLVE_TIMEOUT);
 377         if (rc) {
 378                 dprintk("RPC:       %s: rdma_resolve_addr() failed %i\n",
 379                         __func__, rc);
 380                 goto out;
 381         }
 382         rc = wait_for_completion_interruptible_timeout(&ia->ri_done, wtimeout);
 383         if (rc < 0) {
 384                 dprintk("RPC:       %s: wait() exited: %i\n",
 385                         __func__, rc);
 386                 goto out;
 387         }
 388
 389         /* FIXME:
 390          * Until xprtrdma supports DEVICE_REMOVAL, the provider must
 391          * be pinned while there are active NFS/RDMA mounts to prevent
 392          * hangs and crashes at umount time.
 393          */
 394         if (!ia->ri_async_rc && !try_module_get(id->device->owner)) {
 395                 dprintk("RPC:       %s: Failed to get device module\n",
 396                         __func__);
 397                 ia->ri_async_rc = -ENODEV;
 398         }
 399         rc = ia->ri_async_rc;
 400         if (rc)
 401                 goto out;
 402
 403         ia->ri_async_rc = -ETIMEDOUT;
 404         rc = rdma_resolve_route(id, RDMA_RESOLVE_TIMEOUT);
 405         if (rc) {
 406                 dprintk("RPC:       %s: rdma_resolve_route() failed %i\n",
 407                         __func__, rc);
 408                 goto put;
 409         }
 410         rc = wait_for_completion_interruptible_timeout(&ia->ri_done, wtimeout);
 411         if (rc < 0) {
 412                 dprintk("RPC:       %s: wait() exited: %i\n",
 413                         __func__, rc);
 414                 goto put;
 415         }
 416         rc = ia->ri_async_rc;
 417         if (rc)
 418                 goto put;
 419
 420         return id;
 421 put:
 422         module_put(id->device->owner);
 423 out:
 424         rdma_destroy_id(id);
 425         return ERR_PTR(rc);
 426 }
 427
 428 /*
 429  * Exported functions.
 430  */
 431
 432 /**
 433  * rpcrdma_ia_open - Open and initialize an Interface Adapter.
 434  * @xprt: controlling transport
 435  * @addr: IP address of remote peer
 436  *
 437  * Returns 0 on success, negative errno if an appropriate
 438  * Interface Adapter could not be found and opened.
 439  */
 440 int
 441 rpcrdma_ia_open(struct rpcrdma_xprt *xprt, struct sockaddr *addr)
 442 {
 443         struct rpcrdma_ia *ia = &xprt->rx_ia;
 444         int rc;
 445
 446         ia->ri_id = rpcrdma_create_id(xprt, ia, addr);
 447         if (IS_ERR(ia->ri_id)) {
 448                 rc = PTR_ERR(ia->ri_id);
 449                 goto out_err;
 450         }
 451         ia->ri_device = ia->ri_id->device;
 452
 453         ia->ri_pd = ib_alloc_pd(ia->ri_device, 0);
 454         if (IS_ERR(ia->ri_pd)) {
 455                 rc = PTR_ERR(ia->ri_pd);
 456                 pr_err("rpcrdma: ib_alloc_pd() returned %d\n", rc);
 457                 goto out_err;
 458         }
 459
 460         switch (xprt_rdma_memreg_strategy) {
 461         case RPCRDMA_FRMR:
 462                 if (frwr_is_supported(ia)) {
 463                         ia->ri_ops = &rpcrdma_frwr_memreg_ops;
 464                         break;
 465                 }
 466                 /*FALLTHROUGH*/
 467         case RPCRDMA_MTHCAFMR:
 468                 if (fmr_is_supported(ia)) {
 469                         ia->ri_ops = &rpcrdma_fmr_memreg_ops;
 470                         break;
 471                 }
 472                 /*FALLTHROUGH*/
 473         default:
 474                 pr_err("rpcrdma: Device %s does not support memreg mode %d\n",
 475                        ia->ri_device->name, xprt_rdma_memreg_strategy);
 476                 rc = -EINVAL;
 477                 goto out_err;
 478         }
 479
 480         return 0;
 481
 482 out_err:
 483         rpcrdma_ia_close(ia);
 484         return rc;
 485 }
 486
 487 /**
 488  * rpcrdma_ia_remove - Handle device driver unload
 489  * @ia: interface adapter being removed
 490  *
 491  * Divest transport H/W resources associated with this adapter,
 492  * but allow it to be restored later.
 493  */
 494 void
 495 rpcrdma_ia_remove(struct rpcrdma_ia *ia)
 496 {
 497         struct rpcrdma_xprt *r_xprt = container_of(ia, struct rpcrdma_xprt,
 498                                                    rx_ia);
 499         struct rpcrdma_ep *ep = &r_xprt->rx_ep;
 500         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
 501         struct rpcrdma_req *req;
 502         struct rpcrdma_rep *rep;
 503
 504         cancel_delayed_work_sync(&buf->rb_refresh_worker);
 505
 506         /* This is similar to rpcrdma_ep_destroy, but:
 507          * - Don't cancel the connect worker.
 508          * - Don't call rpcrdma_ep_disconnect, which waits
 509          *   for another conn upcall, which will deadlock.
 510          * - rdma_disconnect is unneeded, the underlying
 511          *   connection is already gone.
 512          */
 513         if (ia->ri_id->qp) {
 514                 ib_drain_qp(ia->ri_id->qp);
 515                 rdma_destroy_qp(ia->ri_id);
 516                 ia->ri_id->qp = NULL;
 517         }
 518         ib_free_cq(ep->rep_attr.recv_cq);
 519         ib_free_cq(ep->rep_attr.send_cq);
 520
 521         /* The ULP is responsible for ensuring all DMA
 522          * mappings and MRs are gone.
 523          */
 524         list_for_each_entry(rep, &buf->rb_recv_bufs, rr_list)
 525                 rpcrdma_dma_unmap_regbuf(rep->rr_rdmabuf);
 526         list_for_each_entry(req, &buf->rb_allreqs, rl_all) {
 527                 rpcrdma_dma_unmap_regbuf(req->rl_rdmabuf);
 528                 rpcrdma_dma_unmap_regbuf(req->rl_sendbuf);
 529                 rpcrdma_dma_unmap_regbuf(req->rl_recvbuf);
 530         }
 531         rpcrdma_destroy_mrs(buf);
 532
 533         /* Allow waiters to continue */
 534         complete(&ia->ri_remove_done);
 535 }
 536
 537 /**
 538  * rpcrdma_ia_close - Clean up/close an IA.
 539  * @ia: interface adapter to close
 540  *
 541  */
 542 void
 543 rpcrdma_ia_close(struct rpcrdma_ia *ia)
 544 {
 545         dprintk("RPC:       %s: entering\n", __func__);
 546         if (ia->ri_id != NULL && !IS_ERR(ia->ri_id)) {
 547                 if (ia->ri_id->qp)
 548                         rdma_destroy_qp(ia->ri_id);
 549                 rpcrdma_destroy_id(ia->ri_id);
 550         }
 551         ia->ri_id = NULL;
 552         ia->ri_device = NULL;
 553
 554         /* If the pd is still busy, xprtrdma missed freeing a resource */
 555         if (ia->ri_pd && !IS_ERR(ia->ri_pd))
 556                 ib_dealloc_pd(ia->ri_pd);
 557         ia->ri_pd = NULL;
 558 }
 559
 560 /*
 561  * Create unconnected endpoint.
 562  */
 563 int
 564 rpcrdma_ep_create(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia,
 565                   struct rpcrdma_create_data_internal *cdata)
 566 {
 567         struct rpcrdma_connect_private *pmsg = &ep->rep_cm_private;
 568         unsigned int max_qp_wr, max_sge;
 569         struct ib_cq *sendcq, *recvcq;
 570         int rc;
 571
 572         max_sge = min_t(unsigned int, ia->ri_device->attrs.max_sge,
 573                         RPCRDMA_MAX_SEND_SGES);
 574         if (max_sge < RPCRDMA_MIN_SEND_SGES) {
 575                 pr_warn("rpcrdma: HCA provides only %d send SGEs\n", max_sge);
 576                 return -ENOMEM;
 577         }
 578         ia->ri_max_send_sges = max_sge - RPCRDMA_MIN_SEND_SGES;
 579
 580         if (ia->ri_device->attrs.max_qp_wr <= RPCRDMA_BACKWARD_WRS) {
 581                 dprintk("RPC:       %s: insufficient wqe's available\n",
 582                         __func__);
 583                 return -ENOMEM;
 584         }
 585         max_qp_wr = ia->ri_device->attrs.max_qp_wr - RPCRDMA_BACKWARD_WRS - 1;
 586
 587         /* check provider's send/recv wr limits */
 588         if (cdata->max_requests > max_qp_wr)
 589                 cdata->max_requests = max_qp_wr;
 590
 591         ep->rep_attr.event_handler = rpcrdma_qp_async_error_upcall;
 592         ep->rep_attr.qp_context = ep;
 593         ep->rep_attr.srq = NULL;
 594         ep->rep_attr.cap.max_send_wr = cdata->max_requests;
 595         ep->rep_attr.cap.max_send_wr += RPCRDMA_BACKWARD_WRS;
 596         ep->rep_attr.cap.max_send_wr += 1;      /* drain cqe */
 597         rc = ia->ri_ops->ro_open(ia, ep, cdata);
 598         if (rc)
 599                 return rc;
 600         ep->rep_attr.cap.max_recv_wr = cdata->max_requests;
 601         ep->rep_attr.cap.max_recv_wr += RPCRDMA_BACKWARD_WRS;
 602         ep->rep_attr.cap.max_recv_wr += 1;      /* drain cqe */
 603         ep->rep_attr.cap.max_send_sge = max_sge;
 604         ep->rep_attr.cap.max_recv_sge = 1;
 605         ep->rep_attr.cap.max_inline_data = 0;
 606         ep->rep_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
 607         ep->rep_attr.qp_type = IB_QPT_RC;
 608         ep->rep_attr.port_num = ~0;
 609
 610         dprintk("RPC:       %s: requested max: dtos: send %d recv %d; "
 611                 "iovs: send %d recv %d\n",
 612                 __func__,
 613                 ep->rep_attr.cap.max_send_wr,
 614                 ep->rep_attr.cap.max_recv_wr,
 615                 ep->rep_attr.cap.max_send_sge,
 616                 ep->rep_attr.cap.max_recv_sge);
 617
 618         /* set trigger for requesting send completion */
 619         ep->rep_cqinit = ep->rep_attr.cap.max_send_wr/2 - 1;
 620         if (ep->rep_cqinit <= 2)
 621                 ep->rep_cqinit = 0;     /* always signal? */
 622         rpcrdma_init_cqcount(ep, 0);
 623         init_waitqueue_head(&ep->rep_connect_wait);
 624         INIT_DELAYED_WORK(&ep->rep_connect_worker, rpcrdma_connect_worker);
 625
 626         sendcq = ib_alloc_cq(ia->ri_device, NULL,
 627                              ep->rep_attr.cap.max_send_wr + 1,
 628                              0, IB_POLL_SOFTIRQ);
 629         if (IS_ERR(sendcq)) {
 630                 rc = PTR_ERR(sendcq);
 631                 dprintk("RPC:       %s: failed to create send CQ: %i\n",
 632                         __func__, rc);
 633                 goto out1;
 634         }
 635
 636         recvcq = ib_alloc_cq(ia->ri_device, NULL,
 637                              ep->rep_attr.cap.max_recv_wr + 1,
 638                              0, IB_POLL_SOFTIRQ);
 639         if (IS_ERR(recvcq)) {
 640                 rc = PTR_ERR(recvcq);
 641                 dprintk("RPC:       %s: failed to create recv CQ: %i\n",
 642                         __func__, rc);
 643                 goto out2;
 644         }
 645
 646         ep->rep_attr.send_cq = sendcq;
 647         ep->rep_attr.recv_cq = recvcq;
 648
 649         /* Initialize cma parameters */
 650         memset(&ep->rep_remote_cma, 0, sizeof(ep->rep_remote_cma));
 651
 652         /* Prepare RDMA-CM private message */
 653         pmsg->cp_magic = rpcrdma_cmp_magic;
 654         pmsg->cp_version = RPCRDMA_CMP_VERSION;
 655         pmsg->cp_flags |= ia->ri_ops->ro_send_w_inv_ok;
 656         pmsg->cp_send_size = rpcrdma_encode_buffer_size(cdata->inline_wsize);
 657         pmsg->cp_recv_size = rpcrdma_encode_buffer_size(cdata->inline_rsize);
 658         ep->rep_remote_cma.private_data = pmsg;
 659         ep->rep_remote_cma.private_data_len = sizeof(*pmsg);
 660
 661         /* Client offers RDMA Read but does not initiate */
 662         ep->rep_remote_cma.initiator_depth = 0;
 663         if (ia->ri_device->attrs.max_qp_rd_atom > 32)   /* arbitrary but <= 255 */
 664                 ep->rep_remote_cma.responder_resources = 32;
 665         else
 666                 ep->rep_remote_cma.responder_resources =
 667                                                 ia->ri_device->attrs.max_qp_rd_atom;
 668
 669         /* Limit transport retries so client can detect server
 670          * GID changes quickly. RPC layer handles re-establishing
 671          * transport connection and retransmission.
 672          */
 673         ep->rep_remote_cma.retry_count = 6;
 674
 675         /* RPC-over-RDMA handles its own flow control. In addition,
 676          * make all RNR NAKs visible so we know that RPC-over-RDMA
 677          * flow control is working correctly (no NAKs should be seen).
 678          */
 679         ep->rep_remote_cma.flow_control = 0;
 680         ep->rep_remote_cma.rnr_retry_count = 0;
 681
 682         return 0;
 683
 684 out2:
 685         ib_free_cq(sendcq);
 686 out1:
 687         return rc;
 688 }
 689
 690 /*
 691  * rpcrdma_ep_destroy
 692  *
 693  * Disconnect and destroy endpoint. After this, the only
 694  * valid operations on the ep are to free it (if dynamically
 695  * allocated) or re-create it.
 696  */
 697 void
 698 rpcrdma_ep_destroy(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
 699 {
 700         dprintk("RPC:       %s: entering, connected is %d\n",
 701                 __func__, ep->rep_connected);
 702
 703         cancel_delayed_work_sync(&ep->rep_connect_worker);
 704
 705         if (ia->ri_id->qp) {
 706                 rpcrdma_ep_disconnect(ep, ia);
 707                 rdma_destroy_qp(ia->ri_id);
 708                 ia->ri_id->qp = NULL;
 709         }
 710
 711         ib_free_cq(ep->rep_attr.recv_cq);
 712         ib_free_cq(ep->rep_attr.send_cq);
 713 }
 714
 715 /* Re-establish a connection after a device removal event.
 716  * Unlike a normal reconnection, a fresh PD and a new set
 717  * of MRs and buffers is needed.
 718  */
 719 static int
 720 rpcrdma_ep_recreate_xprt(struct rpcrdma_xprt *r_xprt,
 721                          struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
 722 {
 723         struct sockaddr *sap = (struct sockaddr *)&r_xprt->rx_data.addr;
 724         int rc, err;
 725
 726         pr_info("%s: r_xprt = %p\n", __func__, r_xprt);
 727
 728         rc = -EHOSTUNREACH;
 729         if (rpcrdma_ia_open(r_xprt, sap))
 730                 goto out1;
 731
 732         rc = -ENOMEM;
 733         err = rpcrdma_ep_create(ep, ia, &r_xprt->rx_data);
 734         if (err) {
 735                 pr_err("rpcrdma: rpcrdma_ep_create returned %d\n", err);
 736                 goto out2;
 737         }
 738
 739         rc = -ENETUNREACH;
 740         err = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr);
 741         if (err) {
 742                 pr_err("rpcrdma: rdma_create_qp returned %d\n", err);
 743                 goto out3;
 744         }
 745
 746         rpcrdma_create_mrs(r_xprt);
 747         return 0;
 748
 749 out3:
 750         rpcrdma_ep_destroy(ep, ia);
 751 out2:
 752         rpcrdma_ia_close(ia);
 753 out1:
 754         return rc;
 755 }
 756
 757 static int
 758 rpcrdma_ep_reconnect(struct rpcrdma_xprt *r_xprt, struct rpcrdma_ep *ep,
 759                      struct rpcrdma_ia *ia)
 760 {
 761         struct sockaddr *sap = (struct sockaddr *)&r_xprt->rx_data.addr;
 762         struct rdma_cm_id *id, *old;
 763         int err, rc;
 764
 765         dprintk("RPC:       %s: reconnecting...\n", __func__);
 766
 767         rpcrdma_ep_disconnect(ep, ia);
 768
 769         rc = -EHOSTUNREACH;
 770         id = rpcrdma_create_id(r_xprt, ia, sap);
 771         if (IS_ERR(id))
 772                 goto out;
 773
 774         /* As long as the new ID points to the same device as the
 775          * old ID, we can reuse the transport's existing PD and all
 776          * previously allocated MRs. Also, the same device means
 777          * the transport's previous DMA mappings are still valid.
 778          *
 779          * This is a sanity check only. There should be no way these
 780          * point to two different devices here.
 781          */
 782         old = id;
 783         rc = -ENETUNREACH;
 784         if (ia->ri_device != id->device) {
 785                 pr_err("rpcrdma: can't reconnect on different device!\n");
 786                 goto out_destroy;
 787         }
 788
 789         err = rdma_create_qp(id, ia->ri_pd, &ep->rep_attr);
 790         if (err) {
 791                 dprintk("RPC:       %s: rdma_create_qp returned %d\n",
 792                         __func__, err);
 793                 goto out_destroy;
 794         }
 795
 796         /* Atomically replace the transport's ID and QP. */
 797         rc = 0;
 798         old = ia->ri_id;
 799         ia->ri_id = id;
 800         rdma_destroy_qp(old);
 801
 802 out_destroy:
 803         rpcrdma_destroy_id(old);
 804 out:
 805         return rc;
 806 }
 807
 808 /*
 809  * Connect unconnected endpoint.
 810  */
 811 int
 812 rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
 813 {
 814         struct rpcrdma_xprt *r_xprt = container_of(ia, struct rpcrdma_xprt,
 815                                                    rx_ia);
 816         unsigned int extras;
 817         int rc;
 818
 819 retry:
 820         switch (ep->rep_connected) {
 821         case 0:
 822                 dprintk("RPC:       %s: connecting...\n", __func__);
 823                 rc = rdma_create_qp(ia->ri_id, ia->ri_pd, &ep->rep_attr);
 824                 if (rc) {
 825                         dprintk("RPC:       %s: rdma_create_qp failed %i\n",
 826                                 __func__, rc);
 827                         rc = -ENETUNREACH;
 828                         goto out_noupdate;
 829                 }
 830                 break;
 831         case -ENODEV:
 832                 rc = rpcrdma_ep_recreate_xprt(r_xprt, ep, ia);
 833                 if (rc)
 834                         goto out_noupdate;
 835                 break;
 836         default:
 837                 rc = rpcrdma_ep_reconnect(r_xprt, ep, ia);
 838                 if (rc)
 839                         goto out;
 840         }
 841
 842         ep->rep_connected = 0;
 843
 844         rc = rdma_connect(ia->ri_id, &ep->rep_remote_cma);
 845         if (rc) {
 846                 dprintk("RPC:       %s: rdma_connect() failed with %i\n",
 847                                 __func__, rc);
 848                 goto out;
 849         }
 850
 851         wait_event_interruptible(ep->rep_connect_wait, ep->rep_connected != 0);
 852         if (ep->rep_connected <= 0) {
 853                 if (ep->rep_connected == -EAGAIN)
 854                         goto retry;
 855                 rc = ep->rep_connected;
 856                 goto out;
 857         }
 858
 859         dprintk("RPC:       %s: connected\n", __func__);
 860         extras = r_xprt->rx_buf.rb_bc_srv_max_requests;
 861         if (extras)
 862                 rpcrdma_ep_post_extra_recv(r_xprt, extras);
 863
 864 out:
 865         if (rc)
 866                 ep->rep_connected = rc;
 867
 868 out_noupdate:
 869         return rc;
 870 }
 871
 872 /*
 873  * rpcrdma_ep_disconnect
 874  *
 875  * This is separate from destroy to facilitate the ability
 876  * to reconnect without recreating the endpoint.
 877  *
 878  * This call is not reentrant, and must not be made in parallel
 879  * on the same endpoint.
 880  */
 881 void
 882 rpcrdma_ep_disconnect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
 883 {
 884         int rc;
 885
 886         rc = rdma_disconnect(ia->ri_id);
 887         if (!rc) {
 888                 /* returns without wait if not connected */
 889                 wait_event_interruptible(ep->rep_connect_wait,
 890                                                         ep->rep_connected != 1);
 891                 dprintk("RPC:       %s: after wait, %sconnected\n", __func__,
 892                         (ep->rep_connected == 1) ? "still " : "dis");
 893         } else {
 894                 dprintk("RPC:       %s: rdma_disconnect %i\n", __func__, rc);
 895                 ep->rep_connected = rc;
 896         }
 897
 898         ib_drain_qp(ia->ri_id->qp);
 899 }
 900
 901 static void
 902 rpcrdma_mr_recovery_worker(struct work_struct *work)
 903 {
 904         struct rpcrdma_buffer *buf = container_of(work, struct rpcrdma_buffer,
 905                                                   rb_recovery_worker.work);
 906         struct rpcrdma_mw *mw;
 907
 908         spin_lock(&buf->rb_recovery_lock);
 909         while (!list_empty(&buf->rb_stale_mrs)) {
 910                 mw = rpcrdma_pop_mw(&buf->rb_stale_mrs);
 911                 spin_unlock(&buf->rb_recovery_lock);
 912
 913                 dprintk("RPC:       %s: recovering MR %p\n", __func__, mw);
 914                 mw->mw_xprt->rx_ia.ri_ops->ro_recover_mr(mw);
 915
 916                 spin_lock(&buf->rb_recovery_lock);
 917         }
 918         spin_unlock(&buf->rb_recovery_lock);
 919 }
 920
 921 void
 922 rpcrdma_defer_mr_recovery(struct rpcrdma_mw *mw)
 923 {
 924         struct rpcrdma_xprt *r_xprt = mw->mw_xprt;
 925         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
 926
 927         spin_lock(&buf->rb_recovery_lock);
 928         rpcrdma_push_mw(mw, &buf->rb_stale_mrs);
 929         spin_unlock(&buf->rb_recovery_lock);
 930
 931         schedule_delayed_work(&buf->rb_recovery_worker, 0);
 932 }
 933
 934 static void
 935 rpcrdma_create_mrs(struct rpcrdma_xprt *r_xprt)
 936 {
 937         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
 938         struct rpcrdma_ia *ia = &r_xprt->rx_ia;
 939         unsigned int count;
 940         LIST_HEAD(free);
 941         LIST_HEAD(all);
 942
 943         for (count = 0; count < 32; count++) {
 944                 struct rpcrdma_mw *mw;
 945                 int rc;
 946
 947                 mw = kzalloc(sizeof(*mw), GFP_KERNEL);
 948                 if (!mw)
 949                         break;
 950
 951                 rc = ia->ri_ops->ro_init_mr(ia, mw);
 952                 if (rc) {
 953                         kfree(mw);
 954                         break;
 955                 }
 956
 957                 mw->mw_xprt = r_xprt;
 958
 959                 list_add(&mw->mw_list, &free);
 960                 list_add(&mw->mw_all, &all);
 961         }
 962
 963         spin_lock(&buf->rb_mwlock);
 964         list_splice(&free, &buf->rb_mws);
 965         list_splice(&all, &buf->rb_all);
 966         r_xprt->rx_stats.mrs_allocated += count;
 967         spin_unlock(&buf->rb_mwlock);
 968
 969         dprintk("RPC:       %s: created %u MRs\n", __func__, count);
 970 }
 971
 972 static void
 973 rpcrdma_mr_refresh_worker(struct work_struct *work)
 974 {
 975         struct rpcrdma_buffer *buf = container_of(work, struct rpcrdma_buffer,
 976                                                   rb_refresh_worker.work);
 977         struct rpcrdma_xprt *r_xprt = container_of(buf, struct rpcrdma_xprt,
 978                                                    rx_buf);
 979
 980         rpcrdma_create_mrs(r_xprt);
 981 }
 982
 983 struct rpcrdma_req *
 984 rpcrdma_create_req(struct rpcrdma_xprt *r_xprt)
 985 {
 986         struct rpcrdma_buffer *buffer = &r_xprt->rx_buf;
 987         struct rpcrdma_req *req;
 988
 989         req = kzalloc(sizeof(*req), GFP_KERNEL);
 990         if (req == NULL)
 991                 return ERR_PTR(-ENOMEM);
 992
 993         INIT_LIST_HEAD(&req->rl_free);
 994         spin_lock(&buffer->rb_reqslock);
 995         list_add(&req->rl_all, &buffer->rb_allreqs);
 996         spin_unlock(&buffer->rb_reqslock);
 997         req->rl_cqe.done = rpcrdma_wc_send;
 998         req->rl_buffer = &r_xprt->rx_buf;
 999         INIT_LIST_HEAD(&req->rl_registered);
1000         req->rl_send_wr.next = NULL;
1001         req->rl_send_wr.wr_cqe = &req->rl_cqe;
1002         req->rl_send_wr.sg_list = req->rl_send_sge;
1003         req->rl_send_wr.opcode = IB_WR_SEND;
1004         return req;
1005 }
1006
1007 struct rpcrdma_rep *
1008 rpcrdma_create_rep(struct rpcrdma_xprt *r_xprt)
1009 {
1010         struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
1011         struct rpcrdma_rep *rep;
1012         int rc;
1013
1014         rc = -ENOMEM;
1015         rep = kzalloc(sizeof(*rep), GFP_KERNEL);
1016         if (rep == NULL)
1017                 goto out;
1018
1019         rep->rr_rdmabuf = rpcrdma_alloc_regbuf(cdata->inline_rsize,
1020                                                DMA_FROM_DEVICE, GFP_KERNEL);
1021         if (IS_ERR(rep->rr_rdmabuf)) {
1022                 rc = PTR_ERR(rep->rr_rdmabuf);
1023                 goto out_free;
1024         }
1025
1026         rep->rr_cqe.done = rpcrdma_wc_receive;
1027         rep->rr_rxprt = r_xprt;
1028         INIT_WORK(&rep->rr_work, rpcrdma_reply_handler);
1029         rep->rr_recv_wr.next = NULL;
1030         rep->rr_recv_wr.wr_cqe = &rep->rr_cqe;
1031         rep->rr_recv_wr.sg_list = &rep->rr_rdmabuf->rg_iov;
1032         rep->rr_recv_wr.num_sge = 1;
1033         return rep;
1034
1035 out_free:
1036         kfree(rep);
1037 out:
1038         return ERR_PTR(rc);
1039 }
1040
1041 int
1042 rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt)
1043 {
1044         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1045         int i, rc;
1046
1047         buf->rb_max_requests = r_xprt->rx_data.max_requests;
1048         buf->rb_bc_srv_max_requests = 0;
1049         atomic_set(&buf->rb_credits, 1);
1050         spin_lock_init(&buf->rb_mwlock);
1051         spin_lock_init(&buf->rb_lock);
1052         spin_lock_init(&buf->rb_recovery_lock);
1053         INIT_LIST_HEAD(&buf->rb_mws);
1054         INIT_LIST_HEAD(&buf->rb_all);
1055         INIT_LIST_HEAD(&buf->rb_stale_mrs);
1056         INIT_DELAYED_WORK(&buf->rb_refresh_worker,
1057                           rpcrdma_mr_refresh_worker);
1058         INIT_DELAYED_WORK(&buf->rb_recovery_worker,
1059                           rpcrdma_mr_recovery_worker);
1060
1061         rpcrdma_create_mrs(r_xprt);
1062
1063         INIT_LIST_HEAD(&buf->rb_send_bufs);
1064         INIT_LIST_HEAD(&buf->rb_allreqs);
1065         spin_lock_init(&buf->rb_reqslock);
1066         for (i = 0; i < buf->rb_max_requests; i++) {
1067                 struct rpcrdma_req *req;
1068
1069                 req = rpcrdma_create_req(r_xprt);
1070                 if (IS_ERR(req)) {
1071                         dprintk("RPC:       %s: request buffer %d alloc"
1072                                 " failed\n", __func__, i);
1073                         rc = PTR_ERR(req);
1074                         goto out;
1075                 }
1076                 req->rl_backchannel = false;
1077                 list_add(&req->rl_free, &buf->rb_send_bufs);
1078         }
1079
1080         INIT_LIST_HEAD(&buf->rb_recv_bufs);
1081         for (i = 0; i < buf->rb_max_requests + RPCRDMA_MAX_BC_REQUESTS; i++) {
1082                 struct rpcrdma_rep *rep;
1083
1084                 rep = rpcrdma_create_rep(r_xprt);
1085                 if (IS_ERR(rep)) {
1086                         dprintk("RPC:       %s: reply buffer %d alloc failed\n",
1087                                 __func__, i);
1088                         rc = PTR_ERR(rep);
1089                         goto out;
1090                 }
1091                 list_add(&rep->rr_list, &buf->rb_recv_bufs);
1092         }
1093
1094         return 0;
1095 out:
1096         rpcrdma_buffer_destroy(buf);
1097         return rc;
1098 }
1099
1100 static struct rpcrdma_req *
1101 rpcrdma_buffer_get_req_locked(struct rpcrdma_buffer *buf)
1102 {
1103         struct rpcrdma_req *req;
1104
1105         req = list_first_entry(&buf->rb_send_bufs,
1106                                struct rpcrdma_req, rl_free);
1107         list_del(&req->rl_free);
1108         return req;
1109 }
1110
1111 static struct rpcrdma_rep *
1112 rpcrdma_buffer_get_rep_locked(struct rpcrdma_buffer *buf)
1113 {
1114         struct rpcrdma_rep *rep;
1115
1116         rep = list_first_entry(&buf->rb_recv_bufs,
1117                                struct rpcrdma_rep, rr_list);
1118         list_del(&rep->rr_list);
1119         return rep;
1120 }
1121
1122 static void
1123 rpcrdma_destroy_rep(struct rpcrdma_rep *rep)
1124 {
1125         rpcrdma_free_regbuf(rep->rr_rdmabuf);
1126         kfree(rep);
1127 }
1128
1129 void
1130 rpcrdma_destroy_req(struct rpcrdma_req *req)
1131 {
1132         rpcrdma_free_regbuf(req->rl_recvbuf);
1133         rpcrdma_free_regbuf(req->rl_sendbuf);
1134         rpcrdma_free_regbuf(req->rl_rdmabuf);
1135         kfree(req);
1136 }
1137
1138 static void
1139 rpcrdma_destroy_mrs(struct rpcrdma_buffer *buf)
1140 {
1141         struct rpcrdma_xprt *r_xprt = container_of(buf, struct rpcrdma_xprt,
1142                                                    rx_buf);
1143         struct rpcrdma_ia *ia = rdmab_to_ia(buf);
1144         struct rpcrdma_mw *mw;
1145         unsigned int count;
1146
1147         count = 0;
1148         spin_lock(&buf->rb_mwlock);
1149         while (!list_empty(&buf->rb_all)) {
1150                 mw = list_entry(buf->rb_all.next, struct rpcrdma_mw, mw_all);
1151                 list_del(&mw->mw_all);
1152
1153                 spin_unlock(&buf->rb_mwlock);
1154                 ia->ri_ops->ro_release_mr(mw);
1155                 count++;
1156                 spin_lock(&buf->rb_mwlock);
1157         }
1158         spin_unlock(&buf->rb_mwlock);
1159         r_xprt->rx_stats.mrs_allocated = 0;
1160
1161         dprintk("RPC:       %s: released %u MRs\n", __func__, count);
1162 }
1163
1164 void
1165 rpcrdma_buffer_destroy(struct rpcrdma_buffer *buf)
1166 {
1167         cancel_delayed_work_sync(&buf->rb_recovery_worker);
1168         cancel_delayed_work_sync(&buf->rb_refresh_worker);
1169
1170         while (!list_empty(&buf->rb_recv_bufs)) {
1171                 struct rpcrdma_rep *rep;
1172
1173                 rep = rpcrdma_buffer_get_rep_locked(buf);
1174                 rpcrdma_destroy_rep(rep);
1175         }
1176         buf->rb_send_count = 0;
1177
1178         spin_lock(&buf->rb_reqslock);
1179         while (!list_empty(&buf->rb_allreqs)) {
1180                 struct rpcrdma_req *req;
1181
1182                 req = list_first_entry(&buf->rb_allreqs,
1183                                        struct rpcrdma_req, rl_all);
1184                 list_del(&req->rl_all);
1185
1186                 spin_unlock(&buf->rb_reqslock);
1187                 rpcrdma_destroy_req(req);
1188                 spin_lock(&buf->rb_reqslock);
1189         }
1190         spin_unlock(&buf->rb_reqslock);
1191         buf->rb_recv_count = 0;
1192
1193         rpcrdma_destroy_mrs(buf);
1194 }
1195
1196 struct rpcrdma_mw *
1197 rpcrdma_get_mw(struct rpcrdma_xprt *r_xprt)
1198 {
1199         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1200         struct rpcrdma_mw *mw = NULL;
1201
1202         spin_lock(&buf->rb_mwlock);
1203         if (!list_empty(&buf->rb_mws))
1204                 mw = rpcrdma_pop_mw(&buf->rb_mws);
1205         spin_unlock(&buf->rb_mwlock);
1206
1207         if (!mw)
1208                 goto out_nomws;
1209         return mw;
1210
1211 out_nomws:
1212         dprintk("RPC:       %s: no MWs available\n", __func__);
1213         if (r_xprt->rx_ep.rep_connected != -ENODEV)
1214                 schedule_delayed_work(&buf->rb_refresh_worker, 0);
1215
1216         /* Allow the reply handler and refresh worker to run */
1217         cond_resched();
1218
1219         return NULL;
1220 }
1221
1222 void
1223 rpcrdma_put_mw(struct rpcrdma_xprt *r_xprt, struct rpcrdma_mw *mw)
1224 {
1225         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
1226
1227         spin_lock(&buf->rb_mwlock);
1228         rpcrdma_push_mw(mw, &buf->rb_mws);
1229         spin_unlock(&buf->rb_mwlock);
1230 }
1231
1232 static struct rpcrdma_rep *
1233 rpcrdma_buffer_get_rep(struct rpcrdma_buffer *buffers)
1234 {
1235         /* If an RPC previously completed without a reply (say, a
1236          * credential problem or a soft timeout occurs) then hold off
1237          * on supplying more Receive buffers until the number of new
1238          * pending RPCs catches up to the number of posted Receives.
1239          */
1240         if (unlikely(buffers->rb_send_count < buffers->rb_recv_count))
1241                 return NULL;
1242
1243         if (unlikely(list_empty(&buffers->rb_recv_bufs)))
1244                 return NULL;
1245         buffers->rb_recv_count++;
1246         return rpcrdma_buffer_get_rep_locked(buffers);
1247 }
1248
1249 /*
1250  * Get a set of request/reply buffers.
1251  *
1252  * Reply buffer (if available) is attached to send buffer upon return.
1253  */
1254 struct rpcrdma_req *
1255 rpcrdma_buffer_get(struct rpcrdma_buffer *buffers)
1256 {
1257         struct rpcrdma_req *req;
1258
1259         spin_lock(&buffers->rb_lock);
1260         if (list_empty(&buffers->rb_send_bufs))
1261                 goto out_reqbuf;
1262         buffers->rb_send_count++;
1263         req = rpcrdma_buffer_get_req_locked(buffers);
1264         req->rl_reply = rpcrdma_buffer_get_rep(buffers);
1265         spin_unlock(&buffers->rb_lock);
1266         return req;
1267
1268 out_reqbuf:
1269         spin_unlock(&buffers->rb_lock);
1270         pr_warn("RPC:       %s: out of request buffers\n", __func__);
1271         return NULL;
1272 }
1273
1274 /*
1275  * Put request/reply buffers back into pool.
1276  * Pre-decrement counter/array index.
1277  */
1278 void
1279 rpcrdma_buffer_put(struct rpcrdma_req *req)
1280 {
1281         struct rpcrdma_buffer *buffers = req->rl_buffer;
1282         struct rpcrdma_rep *rep = req->rl_reply;
1283
1284         req->rl_send_wr.num_sge = 0;
1285         req->rl_reply = NULL;
1286
1287         spin_lock(&buffers->rb_lock);
1288         buffers->rb_send_count--;
1289         list_add_tail(&req->rl_free, &buffers->rb_send_bufs);
1290         if (rep) {
1291                 buffers->rb_recv_count--;
1292                 list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs);
1293         }
1294         spin_unlock(&buffers->rb_lock);
1295 }
1296
1297 /*
1298  * Recover reply buffers from pool.
1299  * This happens when recovering from disconnect.
1300  */
1301 void
1302 rpcrdma_recv_buffer_get(struct rpcrdma_req *req)
1303 {
1304         struct rpcrdma_buffer *buffers = req->rl_buffer;
1305
1306         spin_lock(&buffers->rb_lock);
1307         req->rl_reply = rpcrdma_buffer_get_rep(buffers);
1308         spin_unlock(&buffers->rb_lock);
1309 }
1310
1311 /*
1312  * Put reply buffers back into pool when not attached to
1313  * request. This happens in error conditions.
1314  */
1315 void
1316 rpcrdma_recv_buffer_put(struct rpcrdma_rep *rep)
1317 {
1318         struct rpcrdma_buffer *buffers = &rep->rr_rxprt->rx_buf;
1319
1320         spin_lock(&buffers->rb_lock);
1321         buffers->rb_recv_count--;
1322         list_add_tail(&rep->rr_list, &buffers->rb_recv_bufs);
1323         spin_unlock(&buffers->rb_lock);
1324 }
1325
1326 /**
1327  * rpcrdma_alloc_regbuf - allocate and DMA-map memory for SEND/RECV buffers
1328  * @size: size of buffer to be allocated, in bytes
1329  * @direction: direction of data movement
1330  * @flags: GFP flags
1331  *
1332  * Returns an ERR_PTR, or a pointer to a regbuf, a buffer that
1333  * can be persistently DMA-mapped for I/O.
1334  *
1335  * xprtrdma uses a regbuf for posting an outgoing RDMA SEND, or for
1336  * receiving the payload of RDMA RECV operations. During Long Calls
1337  * or Replies they may be registered externally via ro_map.
1338  */
1339 struct rpcrdma_regbuf *
1340 rpcrdma_alloc_regbuf(size_t size, enum dma_data_direction direction,
1341                      gfp_t flags)
1342 {
1343         struct rpcrdma_regbuf *rb;
1344
1345         rb = kmalloc(sizeof(*rb) + size, flags);
1346         if (rb == NULL)
1347                 return ERR_PTR(-ENOMEM);
1348
1349         rb->rg_device = NULL;
1350         rb->rg_direction = direction;
1351         rb->rg_iov.length = size;
1352
1353         return rb;
1354 }
1355
1356 /**
1357  * __rpcrdma_map_regbuf - DMA-map a regbuf
1358  * @ia: controlling rpcrdma_ia
1359  * @rb: regbuf to be mapped
1360  */
1361 bool
1362 __rpcrdma_dma_map_regbuf(struct rpcrdma_ia *ia, struct rpcrdma_regbuf *rb)
1363 {
1364         struct ib_device *device = ia->ri_device;
1365
1366         if (rb->rg_direction == DMA_NONE)
1367                 return false;
1368
1369         rb->rg_iov.addr = ib_dma_map_single(device,
1370                                             (void *)rb->rg_base,
1371                                             rdmab_length(rb),
1372                                             rb->rg_direction);
1373         if (ib_dma_mapping_error(device, rdmab_addr(rb)))
1374                 return false;
1375
1376         rb->rg_device = device;
1377         rb->rg_iov.lkey = ia->ri_pd->local_dma_lkey;
1378         return true;
1379 }
1380
1381 static void
1382 rpcrdma_dma_unmap_regbuf(struct rpcrdma_regbuf *rb)
1383 {
1384         if (!rpcrdma_regbuf_is_mapped(rb))
1385                 return;
1386
1387         ib_dma_unmap_single(rb->rg_device, rdmab_addr(rb),
1388                             rdmab_length(rb), rb->rg_direction);
1389         rb->rg_device = NULL;
1390 }
1391
1392 /**
1393  * rpcrdma_free_regbuf - deregister and free registered buffer
1394  * @rb: regbuf to be deregistered and freed
1395  */
1396 void
1397 rpcrdma_free_regbuf(struct rpcrdma_regbuf *rb)
1398 {
1399         if (!rb)
1400                 return;
1401
1402         rpcrdma_dma_unmap_regbuf(rb);
1403         kfree(rb);
1404 }
1405
1406 /*
1407  * Prepost any receive buffer, then post send.
1408  *
1409  * Receive buffer is donated to hardware, reclaimed upon recv completion.
1410  */
1411 int
1412 rpcrdma_ep_post(struct rpcrdma_ia *ia,
1413                 struct rpcrdma_ep *ep,
1414                 struct rpcrdma_req *req)
1415 {
1416         struct ib_send_wr *send_wr = &req->rl_send_wr;
1417         struct ib_send_wr *send_wr_fail;
1418         int rc;
1419
1420         if (req->rl_reply) {
1421                 rc = rpcrdma_ep_post_recv(ia, req->rl_reply);
1422                 if (rc)
1423                         return rc;
1424                 req->rl_reply = NULL;
1425         }
1426
1427         dprintk("RPC:       %s: posting %d s/g entries\n",
1428                 __func__, send_wr->num_sge);
1429
1430         rpcrdma_set_signaled(ep, send_wr);
1431         rc = ib_post_send(ia->ri_id->qp, send_wr, &send_wr_fail);
1432         if (rc)
1433                 goto out_postsend_err;
1434         return 0;
1435
1436 out_postsend_err:
1437         pr_err("rpcrdma: RDMA Send ib_post_send returned %i\n", rc);
1438         return -ENOTCONN;
1439 }
1440
1441 int
1442 rpcrdma_ep_post_recv(struct rpcrdma_ia *ia,
1443                      struct rpcrdma_rep *rep)
1444 {
1445         struct ib_recv_wr *recv_wr_fail;
1446         int rc;
1447
1448         if (!rpcrdma_dma_map_regbuf(ia, rep->rr_rdmabuf))
1449                 goto out_map;
1450         rc = ib_post_recv(ia->ri_id->qp, &rep->rr_recv_wr, &recv_wr_fail);
1451         if (rc)
1452                 goto out_postrecv;
1453         return 0;
1454
1455 out_map:
1456         pr_err("rpcrdma: failed to DMA map the Receive buffer\n");
1457         return -EIO;
1458
1459 out_postrecv:
1460         pr_err("rpcrdma: ib_post_recv returned %i\n", rc);
1461         return -ENOTCONN;
1462 }
1463
1464 /**
1465  * rpcrdma_ep_post_extra_recv - Post buffers for incoming backchannel requests
1466  * @r_xprt: transport associated with these backchannel resources
1467  * @min_reqs: minimum number of incoming requests expected
1468  *
1469  * Returns zero if all requested buffers were posted, or a negative errno.
1470  */
1471 int
1472 rpcrdma_ep_post_extra_recv(struct rpcrdma_xprt *r_xprt, unsigned int count)
1473 {
1474         struct rpcrdma_buffer *buffers = &r_xprt->rx_buf;
1475         struct rpcrdma_ia *ia = &r_xprt->rx_ia;
1476         struct rpcrdma_rep *rep;
1477         int rc;
1478
1479         while (count--) {
1480                 spin_lock(&buffers->rb_lock);
1481                 if (list_empty(&buffers->rb_recv_bufs))
1482                         goto out_reqbuf;
1483                 rep = rpcrdma_buffer_get_rep_locked(buffers);
1484                 spin_unlock(&buffers->rb_lock);
1485
1486                 rc = rpcrdma_ep_post_recv(ia, rep);
1487                 if (rc)
1488                         goto out_rc;
1489         }
1490
1491         return 0;
1492
1493 out_reqbuf:
1494         spin_unlock(&buffers->rb_lock);
1495         pr_warn("%s: no extra receive buffers\n", __func__);
1496         return -ENOMEM;
1497
1498 out_rc:
1499         rpcrdma_recv_buffer_put(rep);
1500         return rc;
1501 }