net/sunrpc/xprtrdma/svc_rdma_backchannel.c

   1 // SPDX-License-Identifier: GPL-2.0
   2 /*
   3  * Copyright (c) 2015-2018 Oracle.  All rights reserved.
   4  *
   5  * Support for backward direction RPCs on RPC/RDMA (server-side).
   6  */
   7
   8 #include <linux/sunrpc/svc_rdma.h>
   9
  10 #include "xprt_rdma.h"
  11 #include <trace/events/rpcrdma.h>
  12
  13 /**
  14  * svc_rdma_handle_bc_reply - Process incoming backchannel Reply
  15  * @rqstp: resources for handling the Reply
  16  * @rctxt: Received message
  17  *
  18  */
  19 void svc_rdma_handle_bc_reply(struct svc_rqst *rqstp,
  20                               struct svc_rdma_recv_ctxt *rctxt)
  21 {
  22         struct svc_xprt *sxprt = rqstp->rq_xprt;
  23         struct rpc_xprt *xprt = sxprt->xpt_bc_xprt;
  24         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
  25         struct xdr_buf *rcvbuf = &rqstp->rq_arg;
  26         struct kvec *dst, *src = &rcvbuf->head[0];
  27         __be32 *rdma_resp = rctxt->rc_recv_buf;
  28         struct rpc_rqst *req;
  29         u32 credits;
  30
  31         spin_lock(&xprt->queue_lock);
  32         req = xprt_lookup_rqst(xprt, *rdma_resp);
  33         if (!req)
  34                 goto out_unlock;
  35
  36         dst = &req->rq_private_buf.head[0];
  37         memcpy(&req->rq_private_buf, &req->rq_rcv_buf, sizeof(struct xdr_buf));
  38         if (dst->iov_len < src->iov_len)
  39                 goto out_unlock;
  40         memcpy(dst->iov_base, src->iov_base, src->iov_len);
  41         xprt_pin_rqst(req);
  42         spin_unlock(&xprt->queue_lock);
  43
  44         credits = be32_to_cpup(rdma_resp + 2);
  45         if (credits == 0)
  46                 credits = 1;    /* don't deadlock */
  47         else if (credits > r_xprt->rx_buf.rb_bc_max_requests)
  48                 credits = r_xprt->rx_buf.rb_bc_max_requests;
  49         spin_lock(&xprt->transport_lock);
  50         xprt->cwnd = credits << RPC_CWNDSHIFT;
  51         spin_unlock(&xprt->transport_lock);
  52
  53         spin_lock(&xprt->queue_lock);
  54         xprt_complete_rqst(req->rq_task, rcvbuf->len);
  55         xprt_unpin_rqst(req);
  56         rcvbuf->len = 0;
  57
  58 out_unlock:
  59         spin_unlock(&xprt->queue_lock);
  60 }
  61
  62 /* Send a backwards direction RPC call.
  63  *
  64  * Caller holds the connection's mutex and has already marshaled
  65  * the RPC/RDMA request.
  66  *
  67  * This is similar to svc_rdma_send_reply_msg, but takes a struct
  68  * rpc_rqst instead, does not support chunks, and avoids blocking
  69  * memory allocation.
  70  *
  71  * XXX: There is still an opportunity to block in svc_rdma_send()
  72  * if there are no SQ entries to post the Send. This may occur if
  73  * the adapter has a small maximum SQ depth.
  74  */
  75 static int svc_rdma_bc_sendto(struct svcxprt_rdma *rdma,
  76                               struct rpc_rqst *rqst,
  77                               struct svc_rdma_send_ctxt *ctxt)
  78 {
  79         int ret;
  80
  81         ret = svc_rdma_map_reply_msg(rdma, ctxt, NULL, &rqst->rq_snd_buf);
  82         if (ret < 0)
  83                 return -EIO;
  84
  85         /* Bump page refcnt so Send completion doesn't release
  86          * the rq_buffer before all retransmits are complete.
  87          */
  88         get_page(virt_to_page(rqst->rq_buffer));
  89         ctxt->sc_send_wr.opcode = IB_WR_SEND;
  90         return svc_rdma_send(rdma, ctxt);
  91 }
  92
  93 /* Server-side transport endpoint wants a whole page for its send
  94  * buffer. The client RPC code constructs the RPC header in this
  95  * buffer before it invokes ->send_request.
  96  */
  97 static int
  98 xprt_rdma_bc_allocate(struct rpc_task *task)
  99 {
 100         struct rpc_rqst *rqst = task->tk_rqstp;
 101         size_t size = rqst->rq_callsize;
 102         struct page *page;
 103
 104         if (size > PAGE_SIZE) {
 105                 WARN_ONCE(1, "svcrdma: large bc buffer request (size %zu)\n",
 106                           size);
 107                 return -EINVAL;
 108         }
 109
 110         page = alloc_page(RPCRDMA_DEF_GFP);
 111         if (!page)
 112                 return -ENOMEM;
 113         rqst->rq_buffer = page_address(page);
 114
 115         rqst->rq_rbuffer = kmalloc(rqst->rq_rcvsize, RPCRDMA_DEF_GFP);
 116         if (!rqst->rq_rbuffer) {
 117                 put_page(page);
 118                 return -ENOMEM;
 119         }
 120         return 0;
 121 }
 122
 123 static void
 124 xprt_rdma_bc_free(struct rpc_task *task)
 125 {
 126         struct rpc_rqst *rqst = task->tk_rqstp;
 127
 128         put_page(virt_to_page(rqst->rq_buffer));
 129         kfree(rqst->rq_rbuffer);
 130 }
 131
 132 static int
 133 rpcrdma_bc_send_request(struct svcxprt_rdma *rdma, struct rpc_rqst *rqst)
 134 {
 135         struct rpc_xprt *xprt = rqst->rq_xprt;
 136         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
 137         struct svc_rdma_send_ctxt *ctxt;
 138         __be32 *p;
 139         int rc;
 140
 141         ctxt = svc_rdma_send_ctxt_get(rdma);
 142         if (!ctxt)
 143                 goto drop_connection;
 144
 145         p = xdr_reserve_space(&ctxt->sc_stream, RPCRDMA_HDRLEN_MIN);
 146         if (!p)
 147                 goto put_ctxt;
 148         *p++ = rqst->rq_xid;
 149         *p++ = rpcrdma_version;
 150         *p++ = cpu_to_be32(r_xprt->rx_buf.rb_bc_max_requests);
 151         *p++ = rdma_msg;
 152         *p++ = xdr_zero;
 153         *p++ = xdr_zero;
 154         *p   = xdr_zero;
 155
 156         rqst->rq_xtime = ktime_get();
 157         rc = svc_rdma_bc_sendto(rdma, rqst, ctxt);
 158         if (rc)
 159                 goto put_ctxt;
 160         return 0;
 161
 162 put_ctxt:
 163         svc_rdma_send_ctxt_put(rdma, ctxt);
 164
 165 drop_connection:
 166         return -ENOTCONN;
 167 }
 168
 169 /**
 170  * xprt_rdma_bc_send_request - Send a reverse-direction Call
 171  * @rqst: rpc_rqst containing Call message to be sent
 172  *
 173  * Return values:
 174  *   %0 if the message was sent successfully
 175  *   %ENOTCONN if the message was not sent
 176  */
 177 static int xprt_rdma_bc_send_request(struct rpc_rqst *rqst)
 178 {
 179         struct svc_xprt *sxprt = rqst->rq_xprt->bc_xprt;
 180         struct svcxprt_rdma *rdma =
 181                 container_of(sxprt, struct svcxprt_rdma, sc_xprt);
 182         int ret;
 183
 184         if (test_bit(XPT_DEAD, &sxprt->xpt_flags))
 185                 return -ENOTCONN;
 186
 187         ret = rpcrdma_bc_send_request(rdma, rqst);
 188         if (ret == -ENOTCONN)
 189                 svc_close_xprt(sxprt);
 190         return ret;
 191 }
 192
 193 static void
 194 xprt_rdma_bc_close(struct rpc_xprt *xprt)
 195 {
 196         xprt_disconnect_done(xprt);
 197         xprt->cwnd = RPC_CWNDSHIFT;
 198 }
 199
 200 static void
 201 xprt_rdma_bc_put(struct rpc_xprt *xprt)
 202 {
 203         xprt_rdma_free_addresses(xprt);
 204         xprt_free(xprt);
 205 }
 206
 207 static const struct rpc_xprt_ops xprt_rdma_bc_procs = {
 208         .reserve_xprt           = xprt_reserve_xprt_cong,
 209         .release_xprt           = xprt_release_xprt_cong,
 210         .alloc_slot             = xprt_alloc_slot,
 211         .free_slot              = xprt_free_slot,
 212         .release_request        = xprt_release_rqst_cong,
 213         .buf_alloc              = xprt_rdma_bc_allocate,
 214         .buf_free               = xprt_rdma_bc_free,
 215         .send_request           = xprt_rdma_bc_send_request,
 216         .wait_for_reply_request = xprt_wait_for_reply_request_def,
 217         .close                  = xprt_rdma_bc_close,
 218         .destroy                = xprt_rdma_bc_put,
 219         .print_stats            = xprt_rdma_print_stats
 220 };
 221
 222 static const struct rpc_timeout xprt_rdma_bc_timeout = {
 223         .to_initval = 60 * HZ,
 224         .to_maxval = 60 * HZ,
 225 };
 226
 227 /* It shouldn't matter if the number of backchannel session slots
 228  * doesn't match the number of RPC/RDMA credits. That just means
 229  * one or the other will have extra slots that aren't used.
 230  */
 231 static struct rpc_xprt *
 232 xprt_setup_rdma_bc(struct xprt_create *args)
 233 {
 234         struct rpc_xprt *xprt;
 235         struct rpcrdma_xprt *new_xprt;
 236
 237         if (args->addrlen > sizeof(xprt->addr))
 238                 return ERR_PTR(-EBADF);
 239
 240         xprt = xprt_alloc(args->net, sizeof(*new_xprt),
 241                           RPCRDMA_MAX_BC_REQUESTS,
 242                           RPCRDMA_MAX_BC_REQUESTS);
 243         if (!xprt)
 244                 return ERR_PTR(-ENOMEM);
 245
 246         xprt->timeout = &xprt_rdma_bc_timeout;
 247         xprt_set_bound(xprt);
 248         xprt_set_connected(xprt);
 249         xprt->bind_timeout = RPCRDMA_BIND_TO;
 250         xprt->reestablish_timeout = RPCRDMA_INIT_REEST_TO;
 251         xprt->idle_timeout = RPCRDMA_IDLE_DISC_TO;
 252
 253         xprt->prot = XPRT_TRANSPORT_BC_RDMA;
 254         xprt->ops = &xprt_rdma_bc_procs;
 255
 256         memcpy(&xprt->addr, args->dstaddr, args->addrlen);
 257         xprt->addrlen = args->addrlen;
 258         xprt_rdma_format_addresses(xprt, (struct sockaddr *)&xprt->addr);
 259         xprt->resvport = 0;
 260
 261         xprt->max_payload = xprt_rdma_max_inline_read;
 262
 263         new_xprt = rpcx_to_rdmax(xprt);
 264         new_xprt->rx_buf.rb_bc_max_requests = xprt->max_reqs;
 265
 266         xprt_get(xprt);
 267         args->bc_xprt->xpt_bc_xprt = xprt;
 268         xprt->bc_xprt = args->bc_xprt;
 269
 270         /* Final put for backchannel xprt is in __svc_rdma_free */
 271         xprt_get(xprt);
 272         return xprt;
 273 }
 274
 275 struct xprt_class xprt_rdma_bc = {
 276         .list                   = LIST_HEAD_INIT(xprt_rdma_bc.list),
 277         .name                   = "rdma backchannel",
 278         .owner                  = THIS_MODULE,
 279         .ident                  = XPRT_TRANSPORT_BC_RDMA,
 280         .setup                  = xprt_setup_rdma_bc,
 281 };