net/sunrpc/xprtrdma/backchannel.c

   1 /*
   2  * Copyright (c) 2015 Oracle.  All rights reserved.
   3  *
   4  * Support for backward direction RPCs on RPC/RDMA.
   5  */
   6
   7 #include <linux/module.h>
   8 #include <linux/sunrpc/xprt.h>
   9 #include <linux/sunrpc/svc.h>
  10 #include <linux/sunrpc/svc_xprt.h>
  11
  12 #include "xprt_rdma.h"
  13
  14 #if IS_ENABLED(CONFIG_SUNRPC_DEBUG)
  15 # define RPCDBG_FACILITY        RPCDBG_TRANS
  16 #endif
  17
  18 #undef RPCRDMA_BACKCHANNEL_DEBUG
  19
  20 static void rpcrdma_bc_free_rqst(struct rpcrdma_xprt *r_xprt,
  21                                  struct rpc_rqst *rqst)
  22 {
  23         struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
  24         struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
  25
  26         spin_lock(&buf->rb_reqslock);
  27         list_del(&req->rl_all);
  28         spin_unlock(&buf->rb_reqslock);
  29
  30         rpcrdma_destroy_req(req);
  31
  32         kfree(rqst);
  33 }
  34
  35 static int rpcrdma_bc_setup_rqst(struct rpcrdma_xprt *r_xprt,
  36                                  struct rpc_rqst *rqst)
  37 {
  38         struct rpcrdma_regbuf *rb;
  39         struct rpcrdma_req *req;
  40         size_t size;
  41
  42         req = rpcrdma_create_req(r_xprt);
  43         if (IS_ERR(req))
  44                 return PTR_ERR(req);
  45         req->rl_backchannel = true;
  46
  47         rb = rpcrdma_alloc_regbuf(RPCRDMA_HDRBUF_SIZE,
  48                                   DMA_TO_DEVICE, GFP_KERNEL);
  49         if (IS_ERR(rb))
  50                 goto out_fail;
  51         req->rl_rdmabuf = rb;
  52         xdr_buf_init(&req->rl_hdrbuf, rb->rg_base, rdmab_length(rb));
  53
  54         size = r_xprt->rx_data.inline_rsize;
  55         rb = rpcrdma_alloc_regbuf(size, DMA_TO_DEVICE, GFP_KERNEL);
  56         if (IS_ERR(rb))
  57                 goto out_fail;
  58         req->rl_sendbuf = rb;
  59         xdr_buf_init(&rqst->rq_snd_buf, rb->rg_base,
  60                      min_t(size_t, size, PAGE_SIZE));
  61         rpcrdma_set_xprtdata(rqst, req);
  62         return 0;
  63
  64 out_fail:
  65         rpcrdma_bc_free_rqst(r_xprt, rqst);
  66         return -ENOMEM;
  67 }
  68
  69 /* Allocate and add receive buffers to the rpcrdma_buffer's
  70  * existing list of rep's. These are released when the
  71  * transport is destroyed.
  72  */
  73 static int rpcrdma_bc_setup_reps(struct rpcrdma_xprt *r_xprt,
  74                                  unsigned int count)
  75 {
  76         struct rpcrdma_rep *rep;
  77         int rc = 0;
  78
  79         while (count--) {
  80                 rep = rpcrdma_create_rep(r_xprt);
  81                 if (IS_ERR(rep)) {
  82                         pr_err("RPC:       %s: reply buffer alloc failed\n",
  83                                __func__);
  84                         rc = PTR_ERR(rep);
  85                         break;
  86                 }
  87
  88                 rpcrdma_recv_buffer_put(rep);
  89         }
  90
  91         return rc;
  92 }
  93
  94 /**
  95  * xprt_rdma_bc_setup - Pre-allocate resources for handling backchannel requests
  96  * @xprt: transport associated with these backchannel resources
  97  * @reqs: number of concurrent incoming requests to expect
  98  *
  99  * Returns 0 on success; otherwise a negative errno
 100  */
 101 int xprt_rdma_bc_setup(struct rpc_xprt *xprt, unsigned int reqs)
 102 {
 103         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
 104         struct rpcrdma_buffer *buffer = &r_xprt->rx_buf;
 105         struct rpc_rqst *rqst;
 106         unsigned int i;
 107         int rc;
 108
 109         /* The backchannel reply path returns each rpc_rqst to the
 110          * bc_pa_list _after_ the reply is sent. If the server is
 111          * faster than the client, it can send another backward
 112          * direction request before the rpc_rqst is returned to the
 113          * list. The client rejects the request in this case.
 114          *
 115          * Twice as many rpc_rqsts are prepared to ensure there is
 116          * always an rpc_rqst available as soon as a reply is sent.
 117          */
 118         if (reqs > RPCRDMA_BACKWARD_WRS >> 1)
 119                 goto out_err;
 120
 121         for (i = 0; i < (reqs << 1); i++) {
 122                 rqst = kzalloc(sizeof(*rqst), GFP_KERNEL);
 123                 if (!rqst)
 124                         goto out_free;
 125
 126                 dprintk("RPC:       %s: new rqst %p\n", __func__, rqst);
 127
 128                 rqst->rq_xprt = &r_xprt->rx_xprt;
 129                 INIT_LIST_HEAD(&rqst->rq_list);
 130                 INIT_LIST_HEAD(&rqst->rq_bc_list);
 131
 132                 if (rpcrdma_bc_setup_rqst(r_xprt, rqst))
 133                         goto out_free;
 134
 135                 spin_lock_bh(&xprt->bc_pa_lock);
 136                 list_add(&rqst->rq_bc_pa_list, &xprt->bc_pa_list);
 137                 spin_unlock_bh(&xprt->bc_pa_lock);
 138         }
 139
 140         rc = rpcrdma_bc_setup_reps(r_xprt, reqs);
 141         if (rc)
 142                 goto out_free;
 143
 144         rc = rpcrdma_ep_post_extra_recv(r_xprt, reqs);
 145         if (rc)
 146                 goto out_free;
 147
 148         buffer->rb_bc_srv_max_requests = reqs;
 149         request_module("svcrdma");
 150
 151         return 0;
 152
 153 out_free:
 154         xprt_rdma_bc_destroy(xprt, reqs);
 155
 156 out_err:
 157         pr_err("RPC:       %s: setup backchannel transport failed\n", __func__);
 158         return -ENOMEM;
 159 }
 160
 161 /**
 162  * xprt_rdma_bc_up - Create transport endpoint for backchannel service
 163  * @serv: server endpoint
 164  * @net: network namespace
 165  *
 166  * The "xprt" is an implied argument: it supplies the name of the
 167  * backchannel transport class.
 168  *
 169  * Returns zero on success, negative errno on failure
 170  */
 171 int xprt_rdma_bc_up(struct svc_serv *serv, struct net *net)
 172 {
 173         int ret;
 174
 175         ret = svc_create_xprt(serv, "rdma-bc", net, PF_INET, 0, 0);
 176         if (ret < 0)
 177                 return ret;
 178         return 0;
 179 }
 180
 181 /**
 182  * xprt_rdma_bc_maxpayload - Return maximum backchannel message size
 183  * @xprt: transport
 184  *
 185  * Returns maximum size, in bytes, of a backchannel message
 186  */
 187 size_t xprt_rdma_bc_maxpayload(struct rpc_xprt *xprt)
 188 {
 189         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
 190         struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
 191         size_t maxmsg;
 192
 193         maxmsg = min_t(unsigned int, cdata->inline_rsize, cdata->inline_wsize);
 194         maxmsg = min_t(unsigned int, maxmsg, PAGE_SIZE);
 195         return maxmsg - RPCRDMA_HDRLEN_MIN;
 196 }
 197
 198 /**
 199  * rpcrdma_bc_marshal_reply - Send backwards direction reply
 200  * @rqst: buffer containing RPC reply data
 201  *
 202  * Returns zero on success.
 203  */
 204 int rpcrdma_bc_marshal_reply(struct rpc_rqst *rqst)
 205 {
 206         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(rqst->rq_xprt);
 207         struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
 208         __be32 *p;
 209
 210         rpcrdma_set_xdrlen(&req->rl_hdrbuf, 0);
 211         xdr_init_encode(&req->rl_stream, &req->rl_hdrbuf,
 212                         req->rl_rdmabuf->rg_base);
 213
 214         p = xdr_reserve_space(&req->rl_stream, 28);
 215         if (unlikely(!p))
 216                 return -EIO;
 217         *p++ = rqst->rq_xid;
 218         *p++ = rpcrdma_version;
 219         *p++ = cpu_to_be32(r_xprt->rx_buf.rb_bc_srv_max_requests);
 220         *p++ = rdma_msg;
 221         *p++ = xdr_zero;
 222         *p++ = xdr_zero;
 223         *p = xdr_zero;
 224
 225         if (!rpcrdma_prepare_send_sges(&r_xprt->rx_ia, req, RPCRDMA_HDRLEN_MIN,
 226                                        &rqst->rq_snd_buf, rpcrdma_noch))
 227                 return -EIO;
 228         return 0;
 229 }
 230
 231 /**
 232  * xprt_rdma_bc_destroy - Release resources for handling backchannel requests
 233  * @xprt: transport associated with these backchannel resources
 234  * @reqs: number of incoming requests to destroy; ignored
 235  */
 236 void xprt_rdma_bc_destroy(struct rpc_xprt *xprt, unsigned int reqs)
 237 {
 238         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
 239         struct rpc_rqst *rqst, *tmp;
 240
 241         spin_lock_bh(&xprt->bc_pa_lock);
 242         list_for_each_entry_safe(rqst, tmp, &xprt->bc_pa_list, rq_bc_pa_list) {
 243                 list_del(&rqst->rq_bc_pa_list);
 244                 spin_unlock_bh(&xprt->bc_pa_lock);
 245
 246                 rpcrdma_bc_free_rqst(r_xprt, rqst);
 247
 248                 spin_lock_bh(&xprt->bc_pa_lock);
 249         }
 250         spin_unlock_bh(&xprt->bc_pa_lock);
 251 }
 252
 253 /**
 254  * xprt_rdma_bc_free_rqst - Release a backchannel rqst
 255  * @rqst: request to release
 256  */
 257 void xprt_rdma_bc_free_rqst(struct rpc_rqst *rqst)
 258 {
 259         struct rpc_xprt *xprt = rqst->rq_xprt;
 260
 261         dprintk("RPC:       %s: freeing rqst %p (req %p)\n",
 262                 __func__, rqst, rpcr_to_rdmar(rqst));
 263
 264         smp_mb__before_atomic();
 265         WARN_ON_ONCE(!test_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state));
 266         clear_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state);
 267         smp_mb__after_atomic();
 268
 269         spin_lock_bh(&xprt->bc_pa_lock);
 270         list_add_tail(&rqst->rq_bc_pa_list, &xprt->bc_pa_list);
 271         spin_unlock_bh(&xprt->bc_pa_lock);
 272 }
 273
 274 /**
 275  * rpcrdma_bc_receive_call - Handle a backward direction call
 276  * @xprt: transport receiving the call
 277  * @rep: receive buffer containing the call
 278  *
 279  * Operational assumptions:
 280  *    o Backchannel credits are ignored, just as the NFS server
 281  *      forechannel currently does
 282  *    o The ULP manages a replay cache (eg, NFSv4.1 sessions).
 283  *      No replay detection is done at the transport level
 284  */
 285 void rpcrdma_bc_receive_call(struct rpcrdma_xprt *r_xprt,
 286                              struct rpcrdma_rep *rep)
 287 {
 288         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
 289         struct svc_serv *bc_serv;
 290         struct rpcrdma_req *req;
 291         struct rpc_rqst *rqst;
 292         struct xdr_buf *buf;
 293         size_t size;
 294         __be32 *p;
 295
 296         p = xdr_inline_decode(&rep->rr_stream, 0);
 297         size = xdr_stream_remaining(&rep->rr_stream);
 298
 299 #ifdef RPCRDMA_BACKCHANNEL_DEBUG
 300         pr_info("RPC:       %s: callback XID %08x, length=%u\n",
 301                 __func__, be32_to_cpup(p), size);
 302         pr_info("RPC:       %s: %*ph\n", __func__, size, p);
 303 #endif
 304
 305         /* Grab a free bc rqst */
 306         spin_lock(&xprt->bc_pa_lock);
 307         if (list_empty(&xprt->bc_pa_list)) {
 308                 spin_unlock(&xprt->bc_pa_lock);
 309                 goto out_overflow;
 310         }
 311         rqst = list_first_entry(&xprt->bc_pa_list,
 312                                 struct rpc_rqst, rq_bc_pa_list);
 313         list_del(&rqst->rq_bc_pa_list);
 314         spin_unlock(&xprt->bc_pa_lock);
 315         dprintk("RPC:       %s: using rqst %p\n", __func__, rqst);
 316
 317         /* Prepare rqst */
 318         rqst->rq_reply_bytes_recvd = 0;
 319         rqst->rq_bytes_sent = 0;
 320         rqst->rq_xid = *p;
 321
 322         rqst->rq_private_buf.len = size;
 323         set_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state);
 324
 325         buf = &rqst->rq_rcv_buf;
 326         memset(buf, 0, sizeof(*buf));
 327         buf->head[0].iov_base = p;
 328         buf->head[0].iov_len = size;
 329         buf->len = size;
 330
 331         /* The receive buffer has to be hooked to the rpcrdma_req
 332          * so that it is not released while the req is pointing
 333          * to its buffer, and so that it can be reposted after
 334          * the Upper Layer is done decoding it.
 335          */
 336         req = rpcr_to_rdmar(rqst);
 337         dprintk("RPC:       %s: attaching rep %p to req %p\n",
 338                 __func__, rep, req);
 339         req->rl_reply = rep;
 340
 341         /* Defeat the retransmit detection logic in send_request */
 342         req->rl_connect_cookie = 0;
 343
 344         /* Queue rqst for ULP's callback service */
 345         bc_serv = xprt->bc_serv;
 346         spin_lock(&bc_serv->sv_cb_lock);
 347         list_add(&rqst->rq_bc_list, &bc_serv->sv_cb_list);
 348         spin_unlock(&bc_serv->sv_cb_lock);
 349
 350         wake_up(&bc_serv->sv_cb_waitq);
 351
 352         r_xprt->rx_stats.bcall_count++;
 353         return;
 354
 355 out_overflow:
 356         pr_warn("RPC/RDMA backchannel overflow\n");
 357         xprt_disconnect_done(xprt);
 358         /* This receive buffer gets reposted automatically
 359          * when the connection is re-established.
 360          */
 361         return;
 362 }