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(&r_xprt->rx_ia, 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_ia *ia = &r_xprt->rx_ia;
  39         struct rpcrdma_regbuf *rb;
  40         struct rpcrdma_req *req;
  41         size_t size;
  42
  43         req = rpcrdma_create_req(r_xprt);
  44         if (IS_ERR(req))
  45                 return PTR_ERR(req);
  46         req->rl_backchannel = true;
  47
  48         size = r_xprt->rx_data.inline_wsize;
  49         rb = rpcrdma_alloc_regbuf(ia, size, GFP_KERNEL);
  50         if (IS_ERR(rb))
  51                 goto out_fail;
  52         req->rl_rdmabuf = rb;
  53
  54         size += r_xprt->rx_data.inline_rsize;
  55         rb = rpcrdma_alloc_regbuf(ia, size, GFP_KERNEL);
  56         if (IS_ERR(rb))
  57                 goto out_fail;
  58         rb->rg_owner = req;
  59         req->rl_sendbuf = rb;
  60         /* so that rpcr_to_rdmar works when receiving a request */
  61         rqst->rq_buffer = (void *)req->rl_sendbuf->rg_base;
  62         xdr_buf_init(&rqst->rq_snd_buf, rqst->rq_buffer, size);
  63         return 0;
  64
  65 out_fail:
  66         rpcrdma_bc_free_rqst(r_xprt, rqst);
  67         return -ENOMEM;
  68 }
  69
  70 /* Allocate and add receive buffers to the rpcrdma_buffer's
  71  * existing list of rep's. These are released when the
  72  * transport is destroyed.
  73  */
  74 static int rpcrdma_bc_setup_reps(struct rpcrdma_xprt *r_xprt,
  75                                  unsigned int count)
  76 {
  77         struct rpcrdma_rep *rep;
  78         int rc = 0;
  79
  80         while (count--) {
  81                 rep = rpcrdma_create_rep(r_xprt);
  82                 if (IS_ERR(rep)) {
  83                         pr_err("RPC:       %s: reply buffer alloc failed\n",
  84                                __func__);
  85                         rc = PTR_ERR(rep);
  86                         break;
  87                 }
  88
  89                 rpcrdma_recv_buffer_put(rep);
  90         }
  91
  92         return rc;
  93 }
  94
  95 /**
  96  * xprt_rdma_bc_setup - Pre-allocate resources for handling backchannel requests
  97  * @xprt: transport associated with these backchannel resources
  98  * @reqs: number of concurrent incoming requests to expect
  99  *
 100  * Returns 0 on success; otherwise a negative errno
 101  */
 102 int xprt_rdma_bc_setup(struct rpc_xprt *xprt, unsigned int reqs)
 103 {
 104         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
 105         struct rpcrdma_buffer *buffer = &r_xprt->rx_buf;
 106         struct rpc_rqst *rqst;
 107         unsigned int i;
 108         int rc;
 109
 110         /* The backchannel reply path returns each rpc_rqst to the
 111          * bc_pa_list _after_ the reply is sent. If the server is
 112          * faster than the client, it can send another backward
 113          * direction request before the rpc_rqst is returned to the
 114          * list. The client rejects the request in this case.
 115          *
 116          * Twice as many rpc_rqsts are prepared to ensure there is
 117          * always an rpc_rqst available as soon as a reply is sent.
 118          */
 119         if (reqs > RPCRDMA_BACKWARD_WRS >> 1)
 120                 goto out_err;
 121
 122         for (i = 0; i < (reqs << 1); i++) {
 123                 rqst = kzalloc(sizeof(*rqst), GFP_KERNEL);
 124                 if (!rqst) {
 125                         pr_err("RPC:       %s: Failed to create bc rpc_rqst\n",
 126                                __func__);
 127                         goto out_free;
 128                 }
 129                 dprintk("RPC:       %s: new rqst %p\n", __func__, rqst);
 130
 131                 rqst->rq_xprt = &r_xprt->rx_xprt;
 132                 INIT_LIST_HEAD(&rqst->rq_list);
 133                 INIT_LIST_HEAD(&rqst->rq_bc_list);
 134
 135                 if (rpcrdma_bc_setup_rqst(r_xprt, rqst))
 136                         goto out_free;
 137
 138                 spin_lock_bh(&xprt->bc_pa_lock);
 139                 list_add(&rqst->rq_bc_pa_list, &xprt->bc_pa_list);
 140                 spin_unlock_bh(&xprt->bc_pa_lock);
 141         }
 142
 143         rc = rpcrdma_bc_setup_reps(r_xprt, reqs);
 144         if (rc)
 145                 goto out_free;
 146
 147         rc = rpcrdma_ep_post_extra_recv(r_xprt, reqs);
 148         if (rc)
 149                 goto out_free;
 150
 151         buffer->rb_bc_srv_max_requests = reqs;
 152         request_module("svcrdma");
 153
 154         return 0;
 155
 156 out_free:
 157         xprt_rdma_bc_destroy(xprt, reqs);
 158
 159 out_err:
 160         pr_err("RPC:       %s: setup backchannel transport failed\n", __func__);
 161         return -ENOMEM;
 162 }
 163
 164 /**
 165  * xprt_rdma_bc_up - Create transport endpoint for backchannel service
 166  * @serv: server endpoint
 167  * @net: network namespace
 168  *
 169  * The "xprt" is an implied argument: it supplies the name of the
 170  * backchannel transport class.
 171  *
 172  * Returns zero on success, negative errno on failure
 173  */
 174 int xprt_rdma_bc_up(struct svc_serv *serv, struct net *net)
 175 {
 176         int ret;
 177
 178         ret = svc_create_xprt(serv, "rdma-bc", net, PF_INET, 0, 0);
 179         if (ret < 0)
 180                 return ret;
 181         return 0;
 182 }
 183
 184 /**
 185  * xprt_rdma_bc_maxpayload - Return maximum backchannel message size
 186  * @xprt: transport
 187  *
 188  * Returns maximum size, in bytes, of a backchannel message
 189  */
 190 size_t xprt_rdma_bc_maxpayload(struct rpc_xprt *xprt)
 191 {
 192         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
 193         struct rpcrdma_create_data_internal *cdata = &r_xprt->rx_data;
 194         size_t maxmsg;
 195
 196         maxmsg = min_t(unsigned int, cdata->inline_rsize, cdata->inline_wsize);
 197         return maxmsg - RPCRDMA_HDRLEN_MIN;
 198 }
 199
 200 /**
 201  * rpcrdma_bc_marshal_reply - Send backwards direction reply
 202  * @rqst: buffer containing RPC reply data
 203  *
 204  * Returns zero on success.
 205  */
 206 int rpcrdma_bc_marshal_reply(struct rpc_rqst *rqst)
 207 {
 208         struct rpc_xprt *xprt = rqst->rq_xprt;
 209         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
 210         struct rpcrdma_req *req = rpcr_to_rdmar(rqst);
 211         struct rpcrdma_msg *headerp;
 212         size_t rpclen;
 213
 214         headerp = rdmab_to_msg(req->rl_rdmabuf);
 215         headerp->rm_xid = rqst->rq_xid;
 216         headerp->rm_vers = rpcrdma_version;
 217         headerp->rm_credit =
 218                         cpu_to_be32(r_xprt->rx_buf.rb_bc_srv_max_requests);
 219         headerp->rm_type = rdma_msg;
 220         headerp->rm_body.rm_chunks[0] = xdr_zero;
 221         headerp->rm_body.rm_chunks[1] = xdr_zero;
 222         headerp->rm_body.rm_chunks[2] = xdr_zero;
 223
 224         rpclen = rqst->rq_svec[0].iov_len;
 225
 226 #ifdef RPCRDMA_BACKCHANNEL_DEBUG
 227         pr_info("RPC:       %s: rpclen %zd headerp 0x%p lkey 0x%x\n",
 228                 __func__, rpclen, headerp, rdmab_lkey(req->rl_rdmabuf));
 229         pr_info("RPC:       %s: RPC/RDMA: %*ph\n",
 230                 __func__, (int)RPCRDMA_HDRLEN_MIN, headerp);
 231         pr_info("RPC:       %s:      RPC: %*ph\n",
 232                 __func__, (int)rpclen, rqst->rq_svec[0].iov_base);
 233 #endif
 234
 235         req->rl_send_iov[0].addr = rdmab_addr(req->rl_rdmabuf);
 236         req->rl_send_iov[0].length = RPCRDMA_HDRLEN_MIN;
 237         req->rl_send_iov[0].lkey = rdmab_lkey(req->rl_rdmabuf);
 238
 239         req->rl_send_iov[1].addr = rdmab_addr(req->rl_sendbuf);
 240         req->rl_send_iov[1].length = rpclen;
 241         req->rl_send_iov[1].lkey = rdmab_lkey(req->rl_sendbuf);
 242
 243         req->rl_niovs = 2;
 244         return 0;
 245 }
 246
 247 /**
 248  * xprt_rdma_bc_destroy - Release resources for handling backchannel requests
 249  * @xprt: transport associated with these backchannel resources
 250  * @reqs: number of incoming requests to destroy; ignored
 251  */
 252 void xprt_rdma_bc_destroy(struct rpc_xprt *xprt, unsigned int reqs)
 253 {
 254         struct rpcrdma_xprt *r_xprt = rpcx_to_rdmax(xprt);
 255         struct rpc_rqst *rqst, *tmp;
 256
 257         spin_lock_bh(&xprt->bc_pa_lock);
 258         list_for_each_entry_safe(rqst, tmp, &xprt->bc_pa_list, rq_bc_pa_list) {
 259                 list_del(&rqst->rq_bc_pa_list);
 260                 spin_unlock_bh(&xprt->bc_pa_lock);
 261
 262                 rpcrdma_bc_free_rqst(r_xprt, rqst);
 263
 264                 spin_lock_bh(&xprt->bc_pa_lock);
 265         }
 266         spin_unlock_bh(&xprt->bc_pa_lock);
 267 }
 268
 269 /**
 270  * xprt_rdma_bc_free_rqst - Release a backchannel rqst
 271  * @rqst: request to release
 272  */
 273 void xprt_rdma_bc_free_rqst(struct rpc_rqst *rqst)
 274 {
 275         struct rpc_xprt *xprt = rqst->rq_xprt;
 276
 277         dprintk("RPC:       %s: freeing rqst %p (req %p)\n",
 278                 __func__, rqst, rpcr_to_rdmar(rqst));
 279
 280         smp_mb__before_atomic();
 281         WARN_ON_ONCE(!test_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state));
 282         clear_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state);
 283         smp_mb__after_atomic();
 284
 285         spin_lock_bh(&xprt->bc_pa_lock);
 286         list_add_tail(&rqst->rq_bc_pa_list, &xprt->bc_pa_list);
 287         spin_unlock_bh(&xprt->bc_pa_lock);
 288 }
 289
 290 /**
 291  * rpcrdma_bc_receive_call - Handle a backward direction call
 292  * @xprt: transport receiving the call
 293  * @rep: receive buffer containing the call
 294  *
 295  * Called in the RPC reply handler, which runs in a tasklet.
 296  * Be quick about it.
 297  *
 298  * Operational assumptions:
 299  *    o Backchannel credits are ignored, just as the NFS server
 300  *      forechannel currently does
 301  *    o The ULP manages a replay cache (eg, NFSv4.1 sessions).
 302  *      No replay detection is done at the transport level
 303  */
 304 void rpcrdma_bc_receive_call(struct rpcrdma_xprt *r_xprt,
 305                              struct rpcrdma_rep *rep)
 306 {
 307         struct rpc_xprt *xprt = &r_xprt->rx_xprt;
 308         struct rpcrdma_msg *headerp;
 309         struct svc_serv *bc_serv;
 310         struct rpcrdma_req *req;
 311         struct rpc_rqst *rqst;
 312         struct xdr_buf *buf;
 313         size_t size;
 314         __be32 *p;
 315
 316         headerp = rdmab_to_msg(rep->rr_rdmabuf);
 317 #ifdef RPCRDMA_BACKCHANNEL_DEBUG
 318         pr_info("RPC:       %s: callback XID %08x, length=%u\n",
 319                 __func__, be32_to_cpu(headerp->rm_xid), rep->rr_len);
 320         pr_info("RPC:       %s: %*ph\n", __func__, rep->rr_len, headerp);
 321 #endif
 322
 323         /* Sanity check:
 324          * Need at least enough bytes for RPC/RDMA header, as code
 325          * here references the header fields by array offset. Also,
 326          * backward calls are always inline, so ensure there
 327          * are some bytes beyond the RPC/RDMA header.
 328          */
 329         if (rep->rr_len < RPCRDMA_HDRLEN_MIN + 24)
 330                 goto out_short;
 331         p = (__be32 *)((unsigned char *)headerp + RPCRDMA_HDRLEN_MIN);
 332         size = rep->rr_len - RPCRDMA_HDRLEN_MIN;
 333
 334         /* Grab a free bc rqst */
 335         spin_lock(&xprt->bc_pa_lock);
 336         if (list_empty(&xprt->bc_pa_list)) {
 337                 spin_unlock(&xprt->bc_pa_lock);
 338                 goto out_overflow;
 339         }
 340         rqst = list_first_entry(&xprt->bc_pa_list,
 341                                 struct rpc_rqst, rq_bc_pa_list);
 342         list_del(&rqst->rq_bc_pa_list);
 343         spin_unlock(&xprt->bc_pa_lock);
 344         dprintk("RPC:       %s: using rqst %p\n", __func__, rqst);
 345
 346         /* Prepare rqst */
 347         rqst->rq_reply_bytes_recvd = 0;
 348         rqst->rq_bytes_sent = 0;
 349         rqst->rq_xid = headerp->rm_xid;
 350
 351         rqst->rq_private_buf.len = size;
 352         set_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state);
 353
 354         buf = &rqst->rq_rcv_buf;
 355         memset(buf, 0, sizeof(*buf));
 356         buf->head[0].iov_base = p;
 357         buf->head[0].iov_len = size;
 358         buf->len = size;
 359
 360         /* The receive buffer has to be hooked to the rpcrdma_req
 361          * so that it can be reposted after the server is done
 362          * parsing it but just before sending the backward
 363          * direction reply.
 364          */
 365         req = rpcr_to_rdmar(rqst);
 366         dprintk("RPC:       %s: attaching rep %p to req %p\n",
 367                 __func__, rep, req);
 368         req->rl_reply = rep;
 369
 370         /* Defeat the retransmit detection logic in send_request */
 371         req->rl_connect_cookie = 0;
 372
 373         /* Queue rqst for ULP's callback service */
 374         bc_serv = xprt->bc_serv;
 375         spin_lock(&bc_serv->sv_cb_lock);
 376         list_add(&rqst->rq_bc_list, &bc_serv->sv_cb_list);
 377         spin_unlock(&bc_serv->sv_cb_lock);
 378
 379         wake_up(&bc_serv->sv_cb_waitq);
 380
 381         r_xprt->rx_stats.bcall_count++;
 382         return;
 383
 384 out_overflow:
 385         pr_warn("RPC/RDMA backchannel overflow\n");
 386         xprt_disconnect_done(xprt);
 387         /* This receive buffer gets reposted automatically
 388          * when the connection is re-established.
 389          */
 390         return;
 391
 392 out_short:
 393         pr_warn("RPC/RDMA short backward direction call\n");
 394
 395         if (rpcrdma_ep_post_recv(&r_xprt->rx_ia, &r_xprt->rx_ep, rep))
 396                 xprt_disconnect_done(xprt);
 397         else
 398                 pr_warn("RPC:       %s: reposting rep %p\n",
 399                         __func__, rep);
 400 }