net/sunrpc/xprtrdma/svc_rdma_transport.c

   1 // SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
   2 /*
   3  * Copyright (c) 2015-2018 Oracle. All rights reserved.
   4  * Copyright (c) 2014 Open Grid Computing, Inc. All rights reserved.
   5  * Copyright (c) 2005-2007 Network Appliance, Inc. All rights reserved.
   6  *
   7  * This software is available to you under a choice of one of two
   8  * licenses.  You may choose to be licensed under the terms of the GNU
   9  * General Public License (GPL) Version 2, available from the file
  10  * COPYING in the main directory of this source tree, or the BSD-type
  11  * license below:
  12  *
  13  * Redistribution and use in source and binary forms, with or without
  14  * modification, are permitted provided that the following conditions
  15  * are met:
  16  *
  17  *      Redistributions of source code must retain the above copyright
  18  *      notice, this list of conditions and the following disclaimer.
  19  *
  20  *      Redistributions in binary form must reproduce the above
  21  *      copyright notice, this list of conditions and the following
  22  *      disclaimer in the documentation and/or other materials provided
  23  *      with the distribution.
  24  *
  25  *      Neither the name of the Network Appliance, Inc. nor the names of
  26  *      its contributors may be used to endorse or promote products
  27  *      derived from this software without specific prior written
  28  *      permission.
  29  *
  30  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  31  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  32  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  33  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  34  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  35  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  36  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  37  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  38  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  39  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  40  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  41  *
  42  * Author: Tom Tucker <tom@opengridcomputing.com>
  43  */
  44
  45 #include <linux/interrupt.h>
  46 #include <linux/sched.h>
  47 #include <linux/slab.h>
  48 #include <linux/spinlock.h>
  49 #include <linux/workqueue.h>
  50 #include <linux/export.h>
  51
  52 #include <rdma/ib_verbs.h>
  53 #include <rdma/rdma_cm.h>
  54 #include <rdma/rw.h>
  55
  56 #include <linux/sunrpc/addr.h>
  57 #include <linux/sunrpc/debug.h>
  58 #include <linux/sunrpc/svc_xprt.h>
  59 #include <linux/sunrpc/svc_rdma.h>
  60
  61 #include "xprt_rdma.h"
  62 #include <trace/events/rpcrdma.h>
  63
  64 #define RPCDBG_FACILITY RPCDBG_SVCXPRT
  65
  66 static struct svcxprt_rdma *svc_rdma_create_xprt(struct svc_serv *serv,
  67                                                  struct net *net, int node);
  68 static struct svc_xprt *svc_rdma_create(struct svc_serv *serv,
  69                                         struct net *net,
  70                                         struct sockaddr *sa, int salen,
  71                                         int flags);
  72 static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt);
  73 static void svc_rdma_detach(struct svc_xprt *xprt);
  74 static void svc_rdma_free(struct svc_xprt *xprt);
  75 static int svc_rdma_has_wspace(struct svc_xprt *xprt);
  76 static void svc_rdma_kill_temp_xprt(struct svc_xprt *);
  77
  78 static const struct svc_xprt_ops svc_rdma_ops = {
  79         .xpo_create = svc_rdma_create,
  80         .xpo_recvfrom = svc_rdma_recvfrom,
  81         .xpo_sendto = svc_rdma_sendto,
  82         .xpo_result_payload = svc_rdma_result_payload,
  83         .xpo_release_ctxt = svc_rdma_release_ctxt,
  84         .xpo_detach = svc_rdma_detach,
  85         .xpo_free = svc_rdma_free,
  86         .xpo_has_wspace = svc_rdma_has_wspace,
  87         .xpo_accept = svc_rdma_accept,
  88         .xpo_kill_temp_xprt = svc_rdma_kill_temp_xprt,
  89 };
  90
  91 struct svc_xprt_class svc_rdma_class = {
  92         .xcl_name = "rdma",
  93         .xcl_owner = THIS_MODULE,
  94         .xcl_ops = &svc_rdma_ops,
  95         .xcl_max_payload = RPCSVC_MAXPAYLOAD_RDMA,
  96         .xcl_ident = XPRT_TRANSPORT_RDMA,
  97 };
  98
  99 /* QP event handler */
 100 static void qp_event_handler(struct ib_event *event, void *context)
 101 {
 102         struct svc_xprt *xprt = context;
 103
 104         trace_svcrdma_qp_error(event, (struct sockaddr *)&xprt->xpt_remote);
 105         switch (event->event) {
 106         /* These are considered benign events */
 107         case IB_EVENT_PATH_MIG:
 108         case IB_EVENT_COMM_EST:
 109         case IB_EVENT_SQ_DRAINED:
 110         case IB_EVENT_QP_LAST_WQE_REACHED:
 111                 break;
 112
 113         /* These are considered fatal events */
 114         case IB_EVENT_PATH_MIG_ERR:
 115         case IB_EVENT_QP_FATAL:
 116         case IB_EVENT_QP_REQ_ERR:
 117         case IB_EVENT_QP_ACCESS_ERR:
 118         case IB_EVENT_DEVICE_FATAL:
 119         default:
 120                 svc_xprt_deferred_close(xprt);
 121                 break;
 122         }
 123 }
 124
 125 static struct svcxprt_rdma *svc_rdma_create_xprt(struct svc_serv *serv,
 126                                                  struct net *net, int node)
 127 {
 128         struct svcxprt_rdma *cma_xprt;
 129
 130         cma_xprt = kzalloc_node(sizeof(*cma_xprt), GFP_KERNEL, node);
 131         if (!cma_xprt)
 132                 return NULL;
 133
 134         svc_xprt_init(net, &svc_rdma_class, &cma_xprt->sc_xprt, serv);
 135         INIT_LIST_HEAD(&cma_xprt->sc_accept_q);
 136         INIT_LIST_HEAD(&cma_xprt->sc_rq_dto_q);
 137         init_llist_head(&cma_xprt->sc_send_ctxts);
 138         init_llist_head(&cma_xprt->sc_recv_ctxts);
 139         init_llist_head(&cma_xprt->sc_rw_ctxts);
 140         init_waitqueue_head(&cma_xprt->sc_send_wait);
 141
 142         spin_lock_init(&cma_xprt->sc_lock);
 143         spin_lock_init(&cma_xprt->sc_rq_dto_lock);
 144         spin_lock_init(&cma_xprt->sc_send_lock);
 145         spin_lock_init(&cma_xprt->sc_rw_ctxt_lock);
 146
 147         /*
 148          * Note that this implies that the underlying transport support
 149          * has some form of congestion control (see RFC 7530 section 3.1
 150          * paragraph 2). For now, we assume that all supported RDMA
 151          * transports are suitable here.
 152          */
 153         set_bit(XPT_CONG_CTRL, &cma_xprt->sc_xprt.xpt_flags);
 154
 155         return cma_xprt;
 156 }
 157
 158 static void
 159 svc_rdma_parse_connect_private(struct svcxprt_rdma *newxprt,
 160                                struct rdma_conn_param *param)
 161 {
 162         const struct rpcrdma_connect_private *pmsg = param->private_data;
 163
 164         if (pmsg &&
 165             pmsg->cp_magic == rpcrdma_cmp_magic &&
 166             pmsg->cp_version == RPCRDMA_CMP_VERSION) {
 167                 newxprt->sc_snd_w_inv = pmsg->cp_flags &
 168                                         RPCRDMA_CMP_F_SND_W_INV_OK;
 169
 170                 dprintk("svcrdma: client send_size %u, recv_size %u "
 171                         "remote inv %ssupported\n",
 172                         rpcrdma_decode_buffer_size(pmsg->cp_send_size),
 173                         rpcrdma_decode_buffer_size(pmsg->cp_recv_size),
 174                         newxprt->sc_snd_w_inv ? "" : "un");
 175         }
 176 }
 177
 178 /*
 179  * This function handles the CONNECT_REQUEST event on a listening
 180  * endpoint. It is passed the cma_id for the _new_ connection. The context in
 181  * this cma_id is inherited from the listening cma_id and is the svc_xprt
 182  * structure for the listening endpoint.
 183  *
 184  * This function creates a new xprt for the new connection and enqueues it on
 185  * the accept queue for the listent xprt. When the listen thread is kicked, it
 186  * will call the recvfrom method on the listen xprt which will accept the new
 187  * connection.
 188  */
 189 static void handle_connect_req(struct rdma_cm_id *new_cma_id,
 190                                struct rdma_conn_param *param)
 191 {
 192         struct svcxprt_rdma *listen_xprt = new_cma_id->context;
 193         struct svcxprt_rdma *newxprt;
 194         struct sockaddr *sa;
 195
 196         newxprt = svc_rdma_create_xprt(listen_xprt->sc_xprt.xpt_server,
 197                                        listen_xprt->sc_xprt.xpt_net,
 198                                        ibdev_to_node(new_cma_id->device));
 199         if (!newxprt)
 200                 return;
 201         newxprt->sc_cm_id = new_cma_id;
 202         new_cma_id->context = newxprt;
 203         svc_rdma_parse_connect_private(newxprt, param);
 204
 205         /* Save client advertised inbound read limit for use later in accept. */
 206         newxprt->sc_ord = param->initiator_depth;
 207
 208         sa = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.dst_addr;
 209         newxprt->sc_xprt.xpt_remotelen = svc_addr_len(sa);
 210         memcpy(&newxprt->sc_xprt.xpt_remote, sa,
 211                newxprt->sc_xprt.xpt_remotelen);
 212         snprintf(newxprt->sc_xprt.xpt_remotebuf,
 213                  sizeof(newxprt->sc_xprt.xpt_remotebuf) - 1, "%pISc", sa);
 214
 215         /* The remote port is arbitrary and not under the control of the
 216          * client ULP. Set it to a fixed value so that the DRC continues
 217          * to be effective after a reconnect.
 218          */
 219         rpc_set_port((struct sockaddr *)&newxprt->sc_xprt.xpt_remote, 0);
 220
 221         sa = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.src_addr;
 222         svc_xprt_set_local(&newxprt->sc_xprt, sa, svc_addr_len(sa));
 223
 224         /*
 225          * Enqueue the new transport on the accept queue of the listening
 226          * transport
 227          */
 228         spin_lock(&listen_xprt->sc_lock);
 229         list_add_tail(&newxprt->sc_accept_q, &listen_xprt->sc_accept_q);
 230         spin_unlock(&listen_xprt->sc_lock);
 231
 232         set_bit(XPT_CONN, &listen_xprt->sc_xprt.xpt_flags);
 233         svc_xprt_enqueue(&listen_xprt->sc_xprt);
 234 }
 235
 236 /**
 237  * svc_rdma_listen_handler - Handle CM events generated on a listening endpoint
 238  * @cma_id: the server's listener rdma_cm_id
 239  * @event: details of the event
 240  *
 241  * Return values:
 242  *     %0: Do not destroy @cma_id
 243  *     %1: Destroy @cma_id (never returned here)
 244  *
 245  * NB: There is never a DEVICE_REMOVAL event for INADDR_ANY listeners.
 246  */
 247 static int svc_rdma_listen_handler(struct rdma_cm_id *cma_id,
 248                                    struct rdma_cm_event *event)
 249 {
 250         switch (event->event) {
 251         case RDMA_CM_EVENT_CONNECT_REQUEST:
 252                 handle_connect_req(cma_id, &event->param.conn);
 253                 break;
 254         default:
 255                 break;
 256         }
 257         return 0;
 258 }
 259
 260 /**
 261  * svc_rdma_cma_handler - Handle CM events on client connections
 262  * @cma_id: the server's listener rdma_cm_id
 263  * @event: details of the event
 264  *
 265  * Return values:
 266  *     %0: Do not destroy @cma_id
 267  *     %1: Destroy @cma_id (never returned here)
 268  */
 269 static int svc_rdma_cma_handler(struct rdma_cm_id *cma_id,
 270                                 struct rdma_cm_event *event)
 271 {
 272         struct svcxprt_rdma *rdma = cma_id->context;
 273         struct svc_xprt *xprt = &rdma->sc_xprt;
 274
 275         switch (event->event) {
 276         case RDMA_CM_EVENT_ESTABLISHED:
 277                 clear_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags);
 278
 279                 /* Handle any requests that were received while
 280                  * CONN_PENDING was set. */
 281                 svc_xprt_enqueue(xprt);
 282                 break;
 283         case RDMA_CM_EVENT_DISCONNECTED:
 284         case RDMA_CM_EVENT_DEVICE_REMOVAL:
 285                 svc_xprt_deferred_close(xprt);
 286                 break;
 287         default:
 288                 break;
 289         }
 290         return 0;
 291 }
 292
 293 /*
 294  * Create a listening RDMA service endpoint.
 295  */
 296 static struct svc_xprt *svc_rdma_create(struct svc_serv *serv,
 297                                         struct net *net,
 298                                         struct sockaddr *sa, int salen,
 299                                         int flags)
 300 {
 301         struct rdma_cm_id *listen_id;
 302         struct svcxprt_rdma *cma_xprt;
 303         int ret;
 304
 305         if (sa->sa_family != AF_INET && sa->sa_family != AF_INET6)
 306                 return ERR_PTR(-EAFNOSUPPORT);
 307         cma_xprt = svc_rdma_create_xprt(serv, net, NUMA_NO_NODE);
 308         if (!cma_xprt)
 309                 return ERR_PTR(-ENOMEM);
 310         set_bit(XPT_LISTENER, &cma_xprt->sc_xprt.xpt_flags);
 311         strcpy(cma_xprt->sc_xprt.xpt_remotebuf, "listener");
 312
 313         listen_id = rdma_create_id(net, svc_rdma_listen_handler, cma_xprt,
 314                                    RDMA_PS_TCP, IB_QPT_RC);
 315         if (IS_ERR(listen_id)) {
 316                 ret = PTR_ERR(listen_id);
 317                 goto err0;
 318         }
 319
 320         /* Allow both IPv4 and IPv6 sockets to bind a single port
 321          * at the same time.
 322          */
 323 #if IS_ENABLED(CONFIG_IPV6)
 324         ret = rdma_set_afonly(listen_id, 1);
 325         if (ret)
 326                 goto err1;
 327 #endif
 328         ret = rdma_bind_addr(listen_id, sa);
 329         if (ret)
 330                 goto err1;
 331         cma_xprt->sc_cm_id = listen_id;
 332
 333         ret = rdma_listen(listen_id, RPCRDMA_LISTEN_BACKLOG);
 334         if (ret)
 335                 goto err1;
 336
 337         /*
 338          * We need to use the address from the cm_id in case the
 339          * caller specified 0 for the port number.
 340          */
 341         sa = (struct sockaddr *)&cma_xprt->sc_cm_id->route.addr.src_addr;
 342         svc_xprt_set_local(&cma_xprt->sc_xprt, sa, salen);
 343
 344         return &cma_xprt->sc_xprt;
 345
 346  err1:
 347         rdma_destroy_id(listen_id);
 348  err0:
 349         kfree(cma_xprt);
 350         return ERR_PTR(ret);
 351 }
 352
 353 /*
 354  * This is the xpo_recvfrom function for listening endpoints. Its
 355  * purpose is to accept incoming connections. The CMA callback handler
 356  * has already created a new transport and attached it to the new CMA
 357  * ID.
 358  *
 359  * There is a queue of pending connections hung on the listening
 360  * transport. This queue contains the new svc_xprt structure. This
 361  * function takes svc_xprt structures off the accept_q and completes
 362  * the connection.
 363  */
 364 static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
 365 {
 366         struct svcxprt_rdma *listen_rdma;
 367         struct svcxprt_rdma *newxprt = NULL;
 368         struct rdma_conn_param conn_param;
 369         struct rpcrdma_connect_private pmsg;
 370         struct ib_qp_init_attr qp_attr;
 371         unsigned int ctxts, rq_depth;
 372         struct ib_device *dev;
 373         int ret = 0;
 374         RPC_IFDEBUG(struct sockaddr *sap);
 375
 376         listen_rdma = container_of(xprt, struct svcxprt_rdma, sc_xprt);
 377         clear_bit(XPT_CONN, &xprt->xpt_flags);
 378         /* Get the next entry off the accept list */
 379         spin_lock(&listen_rdma->sc_lock);
 380         if (!list_empty(&listen_rdma->sc_accept_q)) {
 381                 newxprt = list_entry(listen_rdma->sc_accept_q.next,
 382                                      struct svcxprt_rdma, sc_accept_q);
 383                 list_del_init(&newxprt->sc_accept_q);
 384         }
 385         if (!list_empty(&listen_rdma->sc_accept_q))
 386                 set_bit(XPT_CONN, &listen_rdma->sc_xprt.xpt_flags);
 387         spin_unlock(&listen_rdma->sc_lock);
 388         if (!newxprt)
 389                 return NULL;
 390
 391         dev = newxprt->sc_cm_id->device;
 392         newxprt->sc_port_num = newxprt->sc_cm_id->port_num;
 393
 394         /* Qualify the transport resource defaults with the
 395          * capabilities of this particular device */
 396         /* Transport header, head iovec, tail iovec */
 397         newxprt->sc_max_send_sges = 3;
 398         /* Add one SGE per page list entry */
 399         newxprt->sc_max_send_sges += (svcrdma_max_req_size / PAGE_SIZE) + 1;
 400         if (newxprt->sc_max_send_sges > dev->attrs.max_send_sge)
 401                 newxprt->sc_max_send_sges = dev->attrs.max_send_sge;
 402         newxprt->sc_max_req_size = svcrdma_max_req_size;
 403         newxprt->sc_max_requests = svcrdma_max_requests;
 404         newxprt->sc_max_bc_requests = svcrdma_max_bc_requests;
 405         newxprt->sc_recv_batch = RPCRDMA_MAX_RECV_BATCH;
 406         rq_depth = newxprt->sc_max_requests + newxprt->sc_max_bc_requests +
 407                    newxprt->sc_recv_batch;
 408         if (rq_depth > dev->attrs.max_qp_wr) {
 409                 pr_warn("svcrdma: reducing receive depth to %d\n",
 410                         dev->attrs.max_qp_wr);
 411                 rq_depth = dev->attrs.max_qp_wr;
 412                 newxprt->sc_recv_batch = 1;
 413                 newxprt->sc_max_requests = rq_depth - 2;
 414                 newxprt->sc_max_bc_requests = 2;
 415         }
 416         newxprt->sc_fc_credits = cpu_to_be32(newxprt->sc_max_requests);
 417         ctxts = rdma_rw_mr_factor(dev, newxprt->sc_port_num, RPCSVC_MAXPAGES);
 418         ctxts *= newxprt->sc_max_requests;
 419         newxprt->sc_sq_depth = rq_depth + ctxts;
 420         if (newxprt->sc_sq_depth > dev->attrs.max_qp_wr) {
 421                 pr_warn("svcrdma: reducing send depth to %d\n",
 422                         dev->attrs.max_qp_wr);
 423                 newxprt->sc_sq_depth = dev->attrs.max_qp_wr;
 424         }
 425         atomic_set(&newxprt->sc_sq_avail, newxprt->sc_sq_depth);
 426
 427         newxprt->sc_pd = ib_alloc_pd(dev, 0);
 428         if (IS_ERR(newxprt->sc_pd)) {
 429                 trace_svcrdma_pd_err(newxprt, PTR_ERR(newxprt->sc_pd));
 430                 goto errout;
 431         }
 432         newxprt->sc_sq_cq = ib_alloc_cq_any(dev, newxprt, newxprt->sc_sq_depth,
 433                                             IB_POLL_WORKQUEUE);
 434         if (IS_ERR(newxprt->sc_sq_cq))
 435                 goto errout;
 436         newxprt->sc_rq_cq =
 437                 ib_alloc_cq_any(dev, newxprt, rq_depth, IB_POLL_WORKQUEUE);
 438         if (IS_ERR(newxprt->sc_rq_cq))
 439                 goto errout;
 440
 441         memset(&qp_attr, 0, sizeof qp_attr);
 442         qp_attr.event_handler = qp_event_handler;
 443         qp_attr.qp_context = &newxprt->sc_xprt;
 444         qp_attr.port_num = newxprt->sc_port_num;
 445         qp_attr.cap.max_rdma_ctxs = ctxts;
 446         qp_attr.cap.max_send_wr = newxprt->sc_sq_depth - ctxts;
 447         qp_attr.cap.max_recv_wr = rq_depth;
 448         qp_attr.cap.max_send_sge = newxprt->sc_max_send_sges;
 449         qp_attr.cap.max_recv_sge = 1;
 450         qp_attr.sq_sig_type = IB_SIGNAL_REQ_WR;
 451         qp_attr.qp_type = IB_QPT_RC;
 452         qp_attr.send_cq = newxprt->sc_sq_cq;
 453         qp_attr.recv_cq = newxprt->sc_rq_cq;
 454         dprintk("svcrdma: newxprt->sc_cm_id=%p, newxprt->sc_pd=%p\n",
 455                 newxprt->sc_cm_id, newxprt->sc_pd);
 456         dprintk("    cap.max_send_wr = %d, cap.max_recv_wr = %d\n",
 457                 qp_attr.cap.max_send_wr, qp_attr.cap.max_recv_wr);
 458         dprintk("    cap.max_send_sge = %d, cap.max_recv_sge = %d\n",
 459                 qp_attr.cap.max_send_sge, qp_attr.cap.max_recv_sge);
 460
 461         ret = rdma_create_qp(newxprt->sc_cm_id, newxprt->sc_pd, &qp_attr);
 462         if (ret) {
 463                 trace_svcrdma_qp_err(newxprt, ret);
 464                 goto errout;
 465         }
 466         newxprt->sc_qp = newxprt->sc_cm_id->qp;
 467
 468         if (!(dev->attrs.device_cap_flags & IB_DEVICE_MEM_MGT_EXTENSIONS))
 469                 newxprt->sc_snd_w_inv = false;
 470         if (!rdma_protocol_iwarp(dev, newxprt->sc_port_num) &&
 471             !rdma_ib_or_roce(dev, newxprt->sc_port_num)) {
 472                 trace_svcrdma_fabric_err(newxprt, -EINVAL);
 473                 goto errout;
 474         }
 475
 476         if (!svc_rdma_post_recvs(newxprt))
 477                 goto errout;
 478
 479         /* Construct RDMA-CM private message */
 480         pmsg.cp_magic = rpcrdma_cmp_magic;
 481         pmsg.cp_version = RPCRDMA_CMP_VERSION;
 482         pmsg.cp_flags = 0;
 483         pmsg.cp_send_size = pmsg.cp_recv_size =
 484                 rpcrdma_encode_buffer_size(newxprt->sc_max_req_size);
 485
 486         /* Accept Connection */
 487         set_bit(RDMAXPRT_CONN_PENDING, &newxprt->sc_flags);
 488         memset(&conn_param, 0, sizeof conn_param);
 489         conn_param.responder_resources = 0;
 490         conn_param.initiator_depth = min_t(int, newxprt->sc_ord,
 491                                            dev->attrs.max_qp_init_rd_atom);
 492         if (!conn_param.initiator_depth) {
 493                 ret = -EINVAL;
 494                 trace_svcrdma_initdepth_err(newxprt, ret);
 495                 goto errout;
 496         }
 497         conn_param.private_data = &pmsg;
 498         conn_param.private_data_len = sizeof(pmsg);
 499         rdma_lock_handler(newxprt->sc_cm_id);
 500         newxprt->sc_cm_id->event_handler = svc_rdma_cma_handler;
 501         ret = rdma_accept(newxprt->sc_cm_id, &conn_param);
 502         rdma_unlock_handler(newxprt->sc_cm_id);
 503         if (ret) {
 504                 trace_svcrdma_accept_err(newxprt, ret);
 505                 goto errout;
 506         }
 507
 508 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
 509         dprintk("svcrdma: new connection %p accepted:\n", newxprt);
 510         sap = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.src_addr;
 511         dprintk("    local address   : %pIS:%u\n", sap, rpc_get_port(sap));
 512         sap = (struct sockaddr *)&newxprt->sc_cm_id->route.addr.dst_addr;
 513         dprintk("    remote address  : %pIS:%u\n", sap, rpc_get_port(sap));
 514         dprintk("    max_sge         : %d\n", newxprt->sc_max_send_sges);
 515         dprintk("    sq_depth        : %d\n", newxprt->sc_sq_depth);
 516         dprintk("    rdma_rw_ctxs    : %d\n", ctxts);
 517         dprintk("    max_requests    : %d\n", newxprt->sc_max_requests);
 518         dprintk("    ord             : %d\n", conn_param.initiator_depth);
 519 #endif
 520
 521         return &newxprt->sc_xprt;
 522
 523  errout:
 524         /* Take a reference in case the DTO handler runs */
 525         svc_xprt_get(&newxprt->sc_xprt);
 526         if (newxprt->sc_qp && !IS_ERR(newxprt->sc_qp))
 527                 ib_destroy_qp(newxprt->sc_qp);
 528         rdma_destroy_id(newxprt->sc_cm_id);
 529         /* This call to put will destroy the transport */
 530         svc_xprt_put(&newxprt->sc_xprt);
 531         return NULL;
 532 }
 533
 534 static void svc_rdma_detach(struct svc_xprt *xprt)
 535 {
 536         struct svcxprt_rdma *rdma =
 537                 container_of(xprt, struct svcxprt_rdma, sc_xprt);
 538
 539         rdma_disconnect(rdma->sc_cm_id);
 540 }
 541
 542 static void __svc_rdma_free(struct work_struct *work)
 543 {
 544         struct svcxprt_rdma *rdma =
 545                 container_of(work, struct svcxprt_rdma, sc_work);
 546
 547         /* This blocks until the Completion Queues are empty */
 548         if (rdma->sc_qp && !IS_ERR(rdma->sc_qp))
 549                 ib_drain_qp(rdma->sc_qp);
 550
 551         svc_rdma_flush_recv_queues(rdma);
 552
 553         svc_rdma_destroy_rw_ctxts(rdma);
 554         svc_rdma_send_ctxts_destroy(rdma);
 555         svc_rdma_recv_ctxts_destroy(rdma);
 556
 557         /* Destroy the QP if present (not a listener) */
 558         if (rdma->sc_qp && !IS_ERR(rdma->sc_qp))
 559                 ib_destroy_qp(rdma->sc_qp);
 560
 561         if (rdma->sc_sq_cq && !IS_ERR(rdma->sc_sq_cq))
 562                 ib_free_cq(rdma->sc_sq_cq);
 563
 564         if (rdma->sc_rq_cq && !IS_ERR(rdma->sc_rq_cq))
 565                 ib_free_cq(rdma->sc_rq_cq);
 566
 567         if (rdma->sc_pd && !IS_ERR(rdma->sc_pd))
 568                 ib_dealloc_pd(rdma->sc_pd);
 569
 570         /* Destroy the CM ID */
 571         rdma_destroy_id(rdma->sc_cm_id);
 572
 573         kfree(rdma);
 574 }
 575
 576 static void svc_rdma_free(struct svc_xprt *xprt)
 577 {
 578         struct svcxprt_rdma *rdma =
 579                 container_of(xprt, struct svcxprt_rdma, sc_xprt);
 580
 581         INIT_WORK(&rdma->sc_work, __svc_rdma_free);
 582         schedule_work(&rdma->sc_work);
 583 }
 584
 585 static int svc_rdma_has_wspace(struct svc_xprt *xprt)
 586 {
 587         struct svcxprt_rdma *rdma =
 588                 container_of(xprt, struct svcxprt_rdma, sc_xprt);
 589
 590         /*
 591          * If there are already waiters on the SQ,
 592          * return false.
 593          */
 594         if (waitqueue_active(&rdma->sc_send_wait))
 595                 return 0;
 596
 597         /* Otherwise return true. */
 598         return 1;
 599 }
 600
 601 static void svc_rdma_kill_temp_xprt(struct svc_xprt *xprt)
 602 {
 603 }