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