diff options
Diffstat (limited to 'net/sunrpc')
| -rw-r--r-- | net/sunrpc/clnt.c | 19 | ||||
| -rw-r--r-- | net/sunrpc/svc.c | 10 | ||||
| -rw-r--r-- | net/sunrpc/svcsock.c | 10 | ||||
| -rw-r--r-- | net/sunrpc/sysfs.c | 4 | ||||
| -rw-r--r-- | net/sunrpc/xdr.c | 14 | ||||
| -rw-r--r-- | net/sunrpc/xprtrdma/backchannel.c | 5 | ||||
| -rw-r--r-- | net/sunrpc/xprtrdma/frwr_ops.c | 2 | ||||
| -rw-r--r-- | net/sunrpc/xprtrdma/rpc_rdma.c | 157 | ||||
| -rw-r--r-- | net/sunrpc/xprtrdma/transport.c | 68 | ||||
| -rw-r--r-- | net/sunrpc/xprtrdma/verbs.c | 126 | ||||
| -rw-r--r-- | net/sunrpc/xprtrdma/xprt_rdma.h | 4 | ||||
| -rw-r--r-- | net/sunrpc/xprtsock.c | 16 |
12 files changed, 359 insertions, 76 deletions
diff --git a/net/sunrpc/clnt.c b/net/sunrpc/clnt.c index bc8ca470718b..efa26899bc7d 100644 --- a/net/sunrpc/clnt.c +++ b/net/sunrpc/clnt.c @@ -1026,8 +1026,23 @@ rpc_free_auth(struct rpc_clnt *clnt) return NULL; } -/* - * Release reference to the RPC client +/** + * rpc_hold_client - acquire a reference on an rpc_clnt + * @clnt: rpc_clnt to pin + * + * Pairs with rpc_release_client(). + */ +void rpc_hold_client(struct rpc_clnt *clnt) +{ + refcount_inc(&clnt->cl_count); +} + +/** + * rpc_release_client - release a reference on an rpc_clnt + * @clnt: rpc_clnt to release + * + * Pairs with rpc_hold_client(). The rpc_clnt's resources are + * freed once its reference count drops to zero. */ void rpc_release_client(struct rpc_clnt *clnt) diff --git a/net/sunrpc/svc.c b/net/sunrpc/svc.c index 576fa42e7abf..ae9ec4bf34f7 100644 --- a/net/sunrpc/svc.c +++ b/net/sunrpc/svc.c @@ -1598,6 +1598,12 @@ static void svc_release_rqst(struct svc_rqst *rqstp) if (procp && procp->pc_release) procp->pc_release(rqstp); + + /* + * A subsequent svc_release_rqst() on this rqstp must not + * re-invoke pc_release against released state. + */ + rqstp->rq_procinfo = NULL; } /** @@ -1616,6 +1622,9 @@ void svc_process(struct svc_rqst *rqstp) svc_xprt_deferred_close(rqstp->rq_xprt); #endif + /* Discard a stale release hook from a previous RPC. */ + rqstp->rq_procinfo = NULL; + /* * Setup response xdr_buf. * Initially it has just one page @@ -1672,6 +1681,7 @@ void svc_process_bc(struct rpc_rqst *req, struct svc_rqst *rqstp) int proc_error; /* Build the svc_rqst used by the common processing routine */ + rqstp->rq_procinfo = NULL; rqstp->rq_xid = req->rq_xid; rqstp->rq_prot = req->rq_xprt->prot; rqstp->rq_bc_net = req->rq_xprt->xprt_net; diff --git a/net/sunrpc/svcsock.c b/net/sunrpc/svcsock.c index 7be3de1a1aed..eb747493db82 100644 --- a/net/sunrpc/svcsock.c +++ b/net/sunrpc/svcsock.c @@ -471,6 +471,7 @@ static void svc_tcp_handshake_done(void *data, int status, key_serial_t peerid) } clear_bit(XPT_HANDSHAKE, &xprt->xpt_flags); complete_all(&svsk->sk_handshake_done); + svc_xprt_put(xprt); } /** @@ -494,9 +495,13 @@ static void svc_tcp_handshake(struct svc_xprt *xprt) clear_bit(XPT_TLS_SESSION, &xprt->xpt_flags); init_completion(&svsk->sk_handshake_done); + /* Pin the transport across the asynchronous handshake callback. */ + svc_xprt_get(xprt); + ret = tls_server_hello_x509(&args, GFP_KERNEL); if (ret) { trace_svc_tls_not_started(xprt); + svc_xprt_put(xprt); goto out_failed; } @@ -505,8 +510,13 @@ static void svc_tcp_handshake(struct svc_xprt *xprt) if (ret <= 0) { if (tls_handshake_cancel(sk)) { trace_svc_tls_timed_out(xprt); + svc_xprt_put(xprt); goto out_close; } + /* Cancellation lost to handshake_complete(): the + * callback is in flight and should finish quickly. + */ + wait_for_completion(&svsk->sk_handshake_done); } if (!test_bit(XPT_TLS_SESSION, &xprt->xpt_flags)) { diff --git a/net/sunrpc/sysfs.c b/net/sunrpc/sysfs.c index a90480f80154..e638b92b7ad1 100644 --- a/net/sunrpc/sysfs.c +++ b/net/sunrpc/sysfs.c @@ -327,7 +327,7 @@ static ssize_t rpc_sysfs_xprt_switch_add_xprt_store(struct kobject *kobj, { struct rpc_xprt_switch *xprt_switch = rpc_sysfs_xprt_switch_kobj_get_xprt(kobj); - struct xprt_create xprt_create_args; + struct xprt_create xprt_create_args = {}; struct rpc_xprt *xprt, *new; if (!xprt_switch) @@ -348,7 +348,7 @@ static ssize_t rpc_sysfs_xprt_switch_add_xprt_store(struct kobject *kobj, xprt_create_args.reconnect_timeout = xprt->max_reconnect_timeout; new = xprt_create_transport(&xprt_create_args); - if (IS_ERR_OR_NULL(new)) { + if (IS_ERR(new)) { count = PTR_ERR(new); goto out_put_xprt; } diff --git a/net/sunrpc/xdr.c b/net/sunrpc/xdr.c index e83d5d0be78b..4c41e83b7ca9 100644 --- a/net/sunrpc/xdr.c +++ b/net/sunrpc/xdr.c @@ -152,6 +152,8 @@ unsigned int xdr_buf_to_bvec(struct bio_vec *bvec, unsigned int bvec_size, unsigned int count = 0; if (head->iov_len) { + if (unlikely(count >= bvec_size)) + goto bvec_overflow; bvec_set_virt(bvec++, head->iov_base, head->iov_len); ++count; } @@ -165,25 +167,27 @@ unsigned int xdr_buf_to_bvec(struct bio_vec *bvec, unsigned int bvec_size, while (remaining > 0) { len = min_t(unsigned int, remaining, PAGE_SIZE - offset); + if (unlikely(count >= bvec_size)) + goto bvec_overflow; bvec_set_page(bvec++, *pages++, len, offset); remaining -= len; offset = 0; - if (unlikely(++count > bvec_size)) - goto bvec_overflow; + ++count; } } if (tail->iov_len) { - bvec_set_virt(bvec, tail->iov_base, tail->iov_len); - if (unlikely(++count > bvec_size)) + if (unlikely(count >= bvec_size)) goto bvec_overflow; + bvec_set_virt(bvec, tail->iov_base, tail->iov_len); + ++count; } return count; bvec_overflow: pr_warn_once("%s: bio_vec array overflow\n", __func__); - return count - 1; + return count; } EXPORT_SYMBOL_GPL(xdr_buf_to_bvec); diff --git a/net/sunrpc/xprtrdma/backchannel.c b/net/sunrpc/xprtrdma/backchannel.c index 2f0f9618dd05..e5b3463da25f 100644 --- a/net/sunrpc/xprtrdma/backchannel.c +++ b/net/sunrpc/xprtrdma/backchannel.c @@ -159,9 +159,7 @@ void xprt_rdma_bc_free_rqst(struct rpc_rqst *rqst) rpcrdma_rep_put(&r_xprt->rx_buf, rep); req->rl_reply = NULL; - spin_lock(&xprt->bc_pa_lock); - list_add_tail(&rqst->rq_bc_pa_list, &xprt->bc_pa_list); - spin_unlock(&xprt->bc_pa_lock); + rpcrdma_req_put(req); xprt_put(xprt); } @@ -203,6 +201,7 @@ create_req: rqst->rq_xprt = xprt; __set_bit(RPC_BC_PA_IN_USE, &rqst->rq_bc_pa_state); xdr_buf_init(&rqst->rq_snd_buf, rdmab_data(req->rl_sendbuf), size); + kref_init(&req->rl_kref); return rqst; } diff --git a/net/sunrpc/xprtrdma/frwr_ops.c b/net/sunrpc/xprtrdma/frwr_ops.c index 7f79a0a2601e..e5c71cf705a3 100644 --- a/net/sunrpc/xprtrdma/frwr_ops.c +++ b/net/sunrpc/xprtrdma/frwr_ops.c @@ -474,7 +474,7 @@ int frwr_send(struct rpcrdma_xprt *r_xprt, struct rpcrdma_req *req) ++num_wrs; } - if ((kref_read(&req->rl_kref) > 1) || num_wrs > ep->re_send_count) { + if (req->rl_sendctx->sc_unmap_count || num_wrs > ep->re_send_count) { send_wr->send_flags |= IB_SEND_SIGNALED; ep->re_send_count = min_t(unsigned int, ep->re_send_batch, num_wrs - ep->re_send_count); diff --git a/net/sunrpc/xprtrdma/rpc_rdma.c b/net/sunrpc/xprtrdma/rpc_rdma.c index 0e0f21974710..e8b5fc86a01b 100644 --- a/net/sunrpc/xprtrdma/rpc_rdma.c +++ b/net/sunrpc/xprtrdma/rpc_rdma.c @@ -467,29 +467,11 @@ static int rpcrdma_encode_reply_chunk(struct rpcrdma_xprt *r_xprt, return 0; } -static void rpcrdma_sendctx_done(struct kref *kref) -{ - struct rpcrdma_req *req = - container_of(kref, struct rpcrdma_req, rl_kref); - struct rpcrdma_rep *rep = req->rl_reply; - - rpcrdma_complete_rqst(rep); - rep->rr_rxprt->rx_stats.reply_waits_for_send++; -} - -/** - * rpcrdma_sendctx_unmap - DMA-unmap Send buffer - * @sc: sendctx containing SGEs to unmap - * - */ -void rpcrdma_sendctx_unmap(struct rpcrdma_sendctx *sc) +static void rpcrdma_sendctx_dma_unmap(struct rpcrdma_sendctx *sc) { struct rpcrdma_regbuf *rb = sc->sc_req->rl_sendbuf; struct ib_sge *sge; - if (!sc->sc_unmap_count) - return; - /* The first two SGEs contain the transport header and * the inline buffer. These are always left mapped so * they can be cheaply re-used. @@ -498,8 +480,29 @@ void rpcrdma_sendctx_unmap(struct rpcrdma_sendctx *sc) ++sge, --sc->sc_unmap_count) ib_dma_unmap_page(rdmab_device(rb), sge->addr, sge->length, DMA_TO_DEVICE); +} - kref_put(&sc->sc_req->rl_kref, rpcrdma_sendctx_done); +/** + * rpcrdma_sendctx_unmap - DMA-unmap Send buffer and release Send owner + * @sc: sendctx containing SGEs to unmap + * + */ +void rpcrdma_sendctx_unmap(struct rpcrdma_sendctx *sc) +{ + struct rpcrdma_req *req = sc->sc_req; + + rpcrdma_sendctx_dma_unmap(sc); + sc->sc_req = NULL; + rpcrdma_req_put(req); +} + +/* No Send was posted. Release DMA mappings prepared for this + * sendctx, but leave the request reference count alone. + */ +static void rpcrdma_sendctx_cancel(struct rpcrdma_sendctx *sc) +{ + rpcrdma_sendctx_dma_unmap(sc); + sc->sc_req = NULL; } /* Prepare an SGE for the RPC-over-RDMA transport header. @@ -691,8 +694,6 @@ static bool rpcrdma_prepare_noch_mapped(struct rpcrdma_xprt *r_xprt, tail->iov_len)) return false; - if (req->rl_sendctx->sc_unmap_count) - kref_get(&req->rl_kref); return true; } @@ -722,7 +723,6 @@ static bool rpcrdma_prepare_readch(struct rpcrdma_xprt *r_xprt, len -= len & 3; if (!rpcrdma_prepare_tail_iov(req, xdr, page_base, len)) return false; - kref_get(&req->rl_kref); } return true; @@ -743,6 +743,7 @@ inline int rpcrdma_prepare_send_sges(struct rpcrdma_xprt *r_xprt, struct xdr_buf *xdr, enum rpcrdma_chunktype rtype) { + struct rpcrdma_sendctx *sc; int ret; ret = -EAGAIN; @@ -751,7 +752,6 @@ inline int rpcrdma_prepare_send_sges(struct rpcrdma_xprt *r_xprt, goto out_nosc; req->rl_sendctx->sc_unmap_count = 0; req->rl_sendctx->sc_req = req; - kref_init(&req->rl_kref); req->rl_wr.wr_cqe = &req->rl_sendctx->sc_cqe; req->rl_wr.sg_list = req->rl_sendctx->sc_sges; req->rl_wr.num_sge = 0; @@ -779,10 +779,16 @@ inline int rpcrdma_prepare_send_sges(struct rpcrdma_xprt *r_xprt, goto out_unmap; } + /* The Send-side owner releases this reference when the + * Send has completed. + */ + kref_get(&req->rl_kref); return 0; out_unmap: - rpcrdma_sendctx_unmap(req->rl_sendctx); + sc = req->rl_sendctx; + rpcrdma_sendctx_cancel(sc); + rpcrdma_sendctx_unget_locked(r_xprt, sc); out_nosc: trace_xprtrdma_prepsend_failed(&req->rl_slot, ret); return ret; @@ -1081,6 +1087,8 @@ rpcrdma_is_bcall(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep) /* Peek at stream contents without advancing. */ p = xdr_inline_decode(xdr, 0); + if ((char *)xdr->end - (char *)p < 5 * XDR_UNIT) + return false; /* Chunk lists */ if (xdr_item_is_present(p++)) @@ -1105,7 +1113,7 @@ rpcrdma_is_bcall(struct rpcrdma_xprt *r_xprt, struct rpcrdma_rep *rep) */ p = xdr_inline_decode(xdr, 3 * sizeof(*p)); if (unlikely(!p)) - return true; + return false; rpcrdma_bc_receive_call(r_xprt, rep); return true; @@ -1329,6 +1337,11 @@ void rpcrdma_complete_rqst(struct rpcrdma_rep *rep) struct rpc_rqst *rqst = rep->rr_rqst; int status; + /* I3: every registered MR has been invalidated and + * ib_dma_unmap_sg()'d before complete_rqst runs. + */ + WARN_ON_ONCE(!list_empty(&rpcr_to_rdmar(rqst)->rl_registered)); + switch (rep->rr_proc) { case rdma_msg: status = rpcrdma_decode_msg(r_xprt, rep, rqst); @@ -1360,13 +1373,69 @@ out_badheader: goto out; } -static void rpcrdma_reply_done(struct kref *kref) -{ - struct rpcrdma_req *req = - container_of(kref, struct rpcrdma_req, rl_kref); - - rpcrdma_complete_rqst(req->rl_reply); -} +/* Reply-side ownership invariants + * + * I1 (Receive WR ownership). A struct rpcrdma_rep is owned by the + * HCA between ib_post_recv() and the matching Receive completion. + * After ib_dma_sync_single_for_cpu() in rpcrdma_wc_receive() it is + * owned by the CPU until rpcrdma_rep_put() returns it to + * rb_free_reps; a rep on rb_free_reps is not re-posted until + * rpcrdma_post_recvs() pulls it off. Asserted: rpcrdma_post_recvs() + * WARNs that a pulled rep has rr_rqst == NULL. + * + * I2 (rep attachment). While req->rl_reply == rep, the rep cannot be + * re-posted. rpcrdma_reply_put() NULLs req->rl_reply before handing + * the rep to rpcrdma_rep_put(). Asserted: rpcrdma_reply_put() WARNs + * that rl_reply is NULL after the put. + * + * I3 (Registered-MR fence). On entry to rpcrdma_complete_rqst() every + * MR that was on req->rl_registered has had its rkey invalidated + * (remotely via IB_WC_WITH_INVALIDATE or locally via IB_WR_LOCAL_INV) + * and its pages ib_dma_unmap_sg()'d. The LocalInv chain is posted + * on a single QP; strong send-queue ordering makes the last + * completion (frwr_wc_localinv_done) observe the + * ib_dma_unmap_sg() that ran from each earlier completion's + * frwr_mr_put() before complete_rqst is called. The inline + * frwr_reminv() path unmaps its one MR synchronously before + * rpcrdma_reply_handler() reaches complete_rqst. Asserted: + * rpcrdma_complete_rqst() WARNs that rl_registered is empty. + * + * I4 (Send-buffer release). req->rl_kref carries two unconditional + * owners while a Send is outstanding: the RPC-layer reference (set + * at xprt_rdma_alloc_slot / xprt_rdma_bc_rqst_get / rpcrdma_req_release + * pool-entry) and the Send-side reference (kref_get() in + * rpcrdma_prepare_send_sges()). rpcrdma_req_release() runs only + * after both have dropped, so the req does not return to its free + * pool until rpcrdma_sendctx_unmap() has fired -- the HCA has + * released the send buffer before the req can be reused. Asserted: + * rpcrdma_req_release() WARNs that rl_sendctx is NULL. + * + * I5 (req lifecycle). A req is owned by the RPC layer between slot + * acquisition and the matching xprt_rdma_free_slot() (or, for the + * backchannel, xprt_rdma_bc_free_rqst()). While owned, rl_kref >= 1. + * The pools (rb_send_bufs, bc_pa_list, backlog wake target) never + * contain a req with outstanding Send-side or Reply-side work. + * + * Non-hazards. The following claims have been raised by adversarial + * review and are each closed by the invariants above: + * + * * "Reply completes the RPC while the HCA still holds the send + * buffer" -- excluded by I4. The Send-side kref reference is held + * until rpcrdma_sendctx_unmap() runs from Send completion. + * + * * "Signal-driven release races the in-flight Send" -- same + * resolution. xprt_rdma_free() does not touch rl_kref; the + * Send-side reference keeps the req out of its pool until Send + * completion fires. + * + * * "Receive completion races rep reuse" -- excluded by I1. A rep + * is on rb_free_reps only after rpcrdma_rep_put() has been called + * and rpcrdma_post_recvs() owns the next transition back to the HCA. + * + * * "Pages still DMA-mapped when call_decode reads them" -- excluded + * by I3. The matching ib_dma_unmap_sg() for every MR has run on + * the same CPU thread that calls rpcrdma_complete_rqst(). + */ /** * rpcrdma_reply_handler - Process received RPC/RDMA messages @@ -1402,6 +1471,14 @@ void rpcrdma_reply_handler(struct rpcrdma_rep *rep) credits = be32_to_cpu(*p++); rep->rr_proc = *p++; + /* The credit grant from the wire is not trustworthy; + * sanitize it before any code path consumes it. + */ + if (credits == 0) + credits = 1; /* don't deadlock */ + else if (credits > r_xprt->rx_ep->re_max_requests) + credits = r_xprt->rx_ep->re_max_requests; + if (rep->rr_vers != rpcrdma_version) goto out_badversion; @@ -1418,10 +1495,6 @@ void rpcrdma_reply_handler(struct rpcrdma_rep *rep) xprt_pin_rqst(rqst); spin_unlock(&xprt->queue_lock); - if (credits == 0) - credits = 1; /* don't deadlock */ - else if (credits > r_xprt->rx_ep->re_max_requests) - credits = r_xprt->rx_ep->re_max_requests; if (buf->rb_credits != credits) rpcrdma_update_cwnd(r_xprt, credits); @@ -1439,7 +1512,7 @@ void rpcrdma_reply_handler(struct rpcrdma_rep *rep) frwr_unmap_async(r_xprt, req); /* LocalInv completion will complete the RPC */ else - kref_put(&req->rl_kref, rpcrdma_reply_done); + rpcrdma_complete_rqst(rep); out_post: rpcrdma_post_recvs(r_xprt, @@ -1454,11 +1527,13 @@ out_norqst: out_badversion: trace_xprtrdma_reply_vers_err(rep); - goto out; + rpcrdma_rep_put(buf, rep); + credits = buf->rb_credits; + goto out_post; out_shortreply: trace_xprtrdma_reply_short_err(rep); - -out: rpcrdma_rep_put(buf, rep); + credits = buf->rb_credits; + goto out_post; } diff --git a/net/sunrpc/xprtrdma/transport.c b/net/sunrpc/xprtrdma/transport.c index 61706df5e485..5ff8e5126a6c 100644 --- a/net/sunrpc/xprtrdma/transport.c +++ b/net/sunrpc/xprtrdma/transport.c @@ -279,6 +279,13 @@ xprt_rdma_destroy(struct rpc_xprt *xprt) cancel_delayed_work_sync(&r_xprt->rx_connect_worker); rpcrdma_xprt_disconnect(r_xprt); + + /* The disconnect's sendctx drain can return bc_prealloc reqs + * to bc_pa_list after xprt_destroy_backchannel() emptied it. + */ +#if defined(CONFIG_SUNRPC_BACKCHANNEL) + xprt_rdma_bc_destroy(xprt, 0); +#endif rpcrdma_buffer_destroy(&r_xprt->rx_buf); xprt_rdma_free_addresses(xprt); @@ -487,6 +494,50 @@ xprt_rdma_connect(struct rpc_xprt *xprt, struct rpc_task *task) queue_delayed_work(system_long_wq, &r_xprt->rx_connect_worker, delay); } +/* rl_kref has two owners while a Send is outstanding: the rpc_rqst + * owner and the sendctx. Replies complete the RPC but do not drop + * either reference. The req returns to its free pool only after + * xprt_rdma_free_slot() or xprt_rdma_bc_free_rqst() has dropped the + * RPC-layer reference and rpcrdma_sendctx_unmap() has dropped the + * Send-side reference. + */ +static void rpcrdma_req_release(struct kref *kref) +{ + struct rpcrdma_req *req = + container_of(kref, struct rpcrdma_req, rl_kref); + struct rpc_rqst *rqst = &req->rl_slot; + struct rpc_xprt *xprt = rqst->rq_xprt; + struct rpcrdma_xprt *r_xprt; + + /* I4: both the RPC-layer and Send-side owners have dropped, + * so rpcrdma_sendctx_unmap() has cleared rl_sendctx. + */ + WARN_ON_ONCE(req->rl_sendctx); + + kref_init(&req->rl_kref); + +#if defined(CONFIG_SUNRPC_BACKCHANNEL) + if (bc_prealloc(rqst)) { + spin_lock(&xprt->bc_pa_lock); + list_add_tail(&rqst->rq_bc_pa_list, &xprt->bc_pa_list); + spin_unlock(&xprt->bc_pa_lock); + return; + } +#endif + + if (xprt_wake_up_backlog(xprt, rqst)) + return; + + r_xprt = rpcx_to_rdmax(xprt); + memset(rqst, 0, sizeof(*rqst)); + rpcrdma_buffer_put(&r_xprt->rx_buf, req); +} + +void rpcrdma_req_put(struct rpcrdma_req *req) +{ + kref_put(&req->rl_kref, rpcrdma_req_release); +} + /** * xprt_rdma_alloc_slot - allocate an rpc_rqst * @xprt: controlling RPC transport @@ -505,6 +556,7 @@ xprt_rdma_alloc_slot(struct rpc_xprt *xprt, struct rpc_task *task) req = rpcrdma_buffer_get(&r_xprt->rx_buf); if (!req) goto out_sleep; + kref_init(&req->rl_kref); task->tk_rqstp = &req->rl_slot; task->tk_status = 0; return; @@ -520,6 +572,7 @@ out_sleep: if (req) { struct rpc_rqst *rqst = &req->rl_slot; + kref_init(&req->rl_kref); if (!xprt_wake_up_backlog(xprt, rqst)) { memset(rqst, 0, sizeof(*rqst)); rpcrdma_buffer_put(&r_xprt->rx_buf, req); @@ -540,10 +593,7 @@ xprt_rdma_free_slot(struct rpc_xprt *xprt, struct rpc_rqst *rqst) container_of(xprt, struct rpcrdma_xprt, rx_xprt); rpcrdma_reply_put(&r_xprt->rx_buf, rpcr_to_rdmar(rqst)); - if (!xprt_wake_up_backlog(xprt, rqst)) { - memset(rqst, 0, sizeof(*rqst)); - rpcrdma_buffer_put(&r_xprt->rx_buf, rpcr_to_rdmar(rqst)); - } + rpcrdma_req_put(rpcr_to_rdmar(rqst)); } static bool rpcrdma_check_regbuf(struct rpcrdma_xprt *r_xprt, @@ -607,10 +657,10 @@ xprt_rdma_free(struct rpc_task *task) frwr_unmap_sync(rpcx_to_rdmax(rqst->rq_xprt), req); } - /* XXX: If the RPC is completing because of a signal and - * not because a reply was received, we ought to ensure - * that the Send completion has fired, so that memory - * involved with the Send is not still visible to the NIC. + /* The Send-side rl_kref owner keeps req out of its free pool + * until rpcrdma_sendctx_unmap() has fired -- see I4 above + * rpcrdma_reply_handler() -- so signal-driven release here + * does not let the HCA touch a recycled send buffer. */ } @@ -716,7 +766,7 @@ void xprt_rdma_print_stats(struct rpc_xprt *xprt, struct seq_file *seq) r_xprt->rx_stats.mrs_allocated, r_xprt->rx_stats.local_inv_needed, r_xprt->rx_stats.empty_sendctx_q, - r_xprt->rx_stats.reply_waits_for_send); + 0LU); /* was reply_waits_for_send; column preserved */ } static int diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c index aecf9c0a153f..0ac82f26bf6e 100644 --- a/net/sunrpc/xprtrdma/verbs.c +++ b/net/sunrpc/xprtrdma/verbs.c @@ -65,6 +65,8 @@ static int rpcrdma_sendctxs_create(struct rpcrdma_xprt *r_xprt); static void rpcrdma_sendctxs_destroy(struct rpcrdma_xprt *r_xprt); +static unsigned long rpcrdma_sendctx_next(struct rpcrdma_buffer *buf, + unsigned long item); static void rpcrdma_sendctx_put_locked(struct rpcrdma_xprt *r_xprt, struct rpcrdma_sendctx *sc); static int rpcrdma_reqs_setup(struct rpcrdma_xprt *r_xprt); @@ -79,6 +81,8 @@ rpcrdma_regbuf_alloc_node(size_t size, enum dma_data_direction direction, int node); static struct rpcrdma_regbuf * rpcrdma_regbuf_alloc(size_t size, enum dma_data_direction direction); +static bool rpcrdma_regbuf_realloc_node(struct rpcrdma_regbuf *rb, + size_t size, gfp_t flags, int node); static void rpcrdma_regbuf_dma_unmap(struct rpcrdma_regbuf *rb); static void rpcrdma_regbuf_free(struct rpcrdma_regbuf *rb); @@ -243,8 +247,17 @@ rpcrdma_cm_event_handler(struct rdma_cm_id *id, struct rdma_cm_event *event) complete(&ep->re_done); return 0; case RDMA_CM_EVENT_ADDR_CHANGE: - ep->re_connect_status = -ENODEV; - goto disconnected; + switch (xchg(&ep->re_connect_status, -ENODEV)) { + case 0: + goto wake_connect_worker; + case 1: + /* The later DISCONNECTED event balances the + * ESTABLISHED get; do not put here. + */ + rpcrdma_force_disconnect(ep); + return 0; + } + return 0; case RDMA_CM_EVENT_ESTABLISHED: rpcrdma_ep_get(ep); ep->re_connect_status = 1; @@ -267,7 +280,6 @@ wake_connect_worker: return 0; case RDMA_CM_EVENT_DISCONNECTED: ep->re_connect_status = -ECONNABORTED; -disconnected: rpcrdma_force_disconnect(ep); return rpcrdma_ep_put(ep); default: @@ -324,6 +336,7 @@ static struct rdma_cm_id *rpcrdma_create_id(struct rpcrdma_xprt *r_xprt, if (rc) goto out; + ep->re_id = id; rc = rpcrdma_rn_register(id->device, &ep->re_rn, rpcrdma_ep_removal_done); if (rc) goto out; @@ -396,7 +409,6 @@ static int rpcrdma_ep_create(struct rpcrdma_xprt *r_xprt) } __module_get(THIS_MODULE); device = id->device; - ep->re_id = id; reinit_completion(&ep->re_done); ep->re_max_requests = r_xprt->rx_xprt.max_reqs; @@ -539,7 +551,17 @@ int rpcrdma_xprt_connect(struct rpcrdma_xprt *r_xprt) goto out; } rpcrdma_mrs_create(r_xprt); - frwr_wp_create(r_xprt); + + /* + * rpcrdma_encode_write_list() dereferences the write-pad + * MR with no NULL check, so fail the connect rather than + * publish a transport whose write-pad MR is NULL. + */ + rc = frwr_wp_create(r_xprt); + if (rc) { + rc = -ENOTCONN; + goto out; + } out: trace_xprtrdma_connect(r_xprt, rc); @@ -571,9 +593,9 @@ void rpcrdma_xprt_disconnect(struct rpcrdma_xprt *r_xprt) rpcrdma_xprt_drain(r_xprt); rpcrdma_reps_unmap(r_xprt); + rpcrdma_sendctxs_destroy(r_xprt); rpcrdma_reqs_reset(r_xprt); rpcrdma_mrs_destroy(r_xprt); - rpcrdma_sendctxs_destroy(r_xprt); if (rpcrdma_ep_put(ep)) rdma_destroy_id(id); @@ -605,6 +627,25 @@ static void rpcrdma_sendctxs_destroy(struct rpcrdma_xprt *r_xprt) if (!buf->rb_sc_ctxs) return; + + /* The QP is drained, but the final unsignaled Sends might not + * have been walked by a signaled Send completion. Release those + * Send owners before request buffers are reset. + * + * Unlike the completion sweep, this walk can visit slots with + * no Send posted: after a partial rpcrdma_sendctxs_create() + * failure on reconnect, rb_sc_head and rb_sc_tail are stale, + * and slots between them can be NULL or have sc_req clear. + */ + for (i = rpcrdma_sendctx_next(buf, buf->rb_sc_tail); + i != rpcrdma_sendctx_next(buf, buf->rb_sc_head); + i = rpcrdma_sendctx_next(buf, i)) { + struct rpcrdma_sendctx *sc = buf->rb_sc_ctxs[i]; + + if (sc && sc->sc_req) + rpcrdma_sendctx_unmap(sc); + } + for (i = 0; i <= buf->rb_sc_last; i++) kfree(buf->rb_sc_ctxs[i]); kfree(buf->rb_sc_ctxs); @@ -667,6 +708,12 @@ static unsigned long rpcrdma_sendctx_next(struct rpcrdma_buffer *buf, return likely(item < buf->rb_sc_last) ? item + 1 : 0; } +static unsigned long rpcrdma_sendctx_prev(struct rpcrdma_buffer *buf, + unsigned long item) +{ + return item > 0 ? item - 1 : buf->rb_sc_last; +} + /** * rpcrdma_sendctx_get_locked - Acquire a send context * @r_xprt: controlling transport instance @@ -724,6 +771,29 @@ out_emptyq: } /** + * rpcrdma_sendctx_unget_locked - Release an unposted send context + * @r_xprt: controlling transport instance + * @sc: send context to release + * + * Usage: Called when no Send is posted for the sendctx most + * recently returned by rpcrdma_sendctx_get_locked(). + * + * The caller serializes calls to this function and to + * rpcrdma_sendctx_get_locked() (per transport). + */ +void rpcrdma_sendctx_unget_locked(struct rpcrdma_xprt *r_xprt, + struct rpcrdma_sendctx *sc) +{ + struct rpcrdma_buffer *buf = &r_xprt->rx_buf; + + if (WARN_ON_ONCE(buf->rb_sc_ctxs[buf->rb_sc_head] != sc)) + return; + + buf->rb_sc_head = rpcrdma_sendctx_prev(buf, buf->rb_sc_head); + xprt_write_space(&r_xprt->rx_xprt); +} + +/** * rpcrdma_sendctx_put_locked - Release a send context * @r_xprt: controlling transport instance * @sc: send context to release @@ -739,15 +809,18 @@ static void rpcrdma_sendctx_put_locked(struct rpcrdma_xprt *r_xprt, struct rpcrdma_buffer *buf = &r_xprt->rx_buf; unsigned long next_tail; - /* Unmap SGEs of previously completed but unsignaled - * Sends by walking up the queue until @sc is found. + /* Release previously completed but unsignaled Sends by walking + * up the queue until @sc is found. */ next_tail = buf->rb_sc_tail; do { + struct rpcrdma_sendctx *cur; + next_tail = rpcrdma_sendctx_next(buf, next_tail); /* ORDER: item must be accessed _before_ tail is updated */ - rpcrdma_sendctx_unmap(buf->rb_sc_ctxs[next_tail]); + cur = buf->rb_sc_ctxs[next_tail]; + rpcrdma_sendctx_unmap(cur); } while (buf->rb_sc_ctxs[next_tail] != sc); @@ -1211,6 +1284,10 @@ void rpcrdma_reply_put(struct rpcrdma_buffer *buffers, struct rpcrdma_req *req) rpcrdma_rep_put(buffers, req->rl_reply); req->rl_reply = NULL; } + /* I2: rl_reply NULL after the put closes the + * 'rep on rb_free_reps still referenced by req' window. + */ + WARN_ON_ONCE(req->rl_reply); } /** @@ -1292,9 +1369,15 @@ rpcrdma_regbuf_alloc(size_t size, enum dma_data_direction direction) */ bool rpcrdma_regbuf_realloc(struct rpcrdma_regbuf *rb, size_t size, gfp_t flags) { + return rpcrdma_regbuf_realloc_node(rb, size, flags, NUMA_NO_NODE); +} + +static bool rpcrdma_regbuf_realloc_node(struct rpcrdma_regbuf *rb, + size_t size, gfp_t flags, int node) +{ void *buf; - buf = kmalloc(size, flags); + buf = kmalloc_node(size, flags, node); if (!buf) return false; @@ -1306,6 +1389,23 @@ bool rpcrdma_regbuf_realloc(struct rpcrdma_regbuf *rb, size_t size, gfp_t flags) return true; } +static bool rpcrdma_rep_resize(struct rpcrdma_xprt *r_xprt, + struct rpcrdma_rep *rep) +{ + struct rpcrdma_regbuf *rb = rep->rr_rdmabuf; + struct rpcrdma_ep *ep = r_xprt->rx_ep; + size_t size = ep->re_inline_recv; + + if (likely(rdmab_length(rb) >= size)) + return true; + if (!rpcrdma_regbuf_realloc_node(rb, size, XPRTRDMA_GFP_FLAGS, + ibdev_to_node(ep->re_id->device))) + return false; + + xdr_buf_init(&rep->rr_hdrbuf, rdmab_data(rb), rdmab_length(rb)); + return true; +} + /** * __rpcrdma_regbuf_dma_map - DMA-map a regbuf * @r_xprt: controlling transport instance @@ -1387,6 +1487,12 @@ void rpcrdma_post_recvs(struct rpcrdma_xprt *r_xprt, int needed) rep = rpcrdma_rep_create(r_xprt); if (!rep) break; + /* I1: a rep on rb_free_reps must carry no rqst pointer. */ + WARN_ON_ONCE(rep->rr_rqst); + if (!rpcrdma_rep_resize(r_xprt, rep)) { + rpcrdma_rep_put(buf, rep); + break; + } if (!rpcrdma_regbuf_dma_map(r_xprt, rep->rr_rdmabuf)) { rpcrdma_rep_put(buf, rep); break; diff --git a/net/sunrpc/xprtrdma/xprt_rdma.h b/net/sunrpc/xprtrdma/xprt_rdma.h index f53a77472724..57be5776aaff 100644 --- a/net/sunrpc/xprtrdma/xprt_rdma.h +++ b/net/sunrpc/xprtrdma/xprt_rdma.h @@ -427,7 +427,6 @@ struct rpcrdma_stats { /* accessed when receiving a reply */ unsigned long long total_rdma_reply; unsigned long long fixup_copy_count; - unsigned long reply_waits_for_send; unsigned long local_inv_needed; unsigned long nomsg_call_count; unsigned long bcall_count; @@ -496,6 +495,8 @@ void rpcrdma_req_destroy(struct rpcrdma_req *req); int rpcrdma_buffer_create(struct rpcrdma_xprt *); void rpcrdma_buffer_destroy(struct rpcrdma_buffer *); struct rpcrdma_sendctx *rpcrdma_sendctx_get_locked(struct rpcrdma_xprt *r_xprt); +void rpcrdma_sendctx_unget_locked(struct rpcrdma_xprt *r_xprt, + struct rpcrdma_sendctx *sc); struct rpcrdma_mr *rpcrdma_mr_get(struct rpcrdma_xprt *r_xprt); void rpcrdma_mrs_refresh(struct rpcrdma_xprt *r_xprt); @@ -505,6 +506,7 @@ void rpcrdma_buffer_put(struct rpcrdma_buffer *buffers, struct rpcrdma_req *req); void rpcrdma_rep_put(struct rpcrdma_buffer *buf, struct rpcrdma_rep *rep); void rpcrdma_reply_put(struct rpcrdma_buffer *buffers, struct rpcrdma_req *req); +void rpcrdma_req_put(struct rpcrdma_req *req); bool rpcrdma_regbuf_realloc(struct rpcrdma_regbuf *rb, size_t size, gfp_t flags); diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c index 2e1fe6013361..359407aae03e 100644 --- a/net/sunrpc/xprtsock.c +++ b/net/sunrpc/xprtsock.c @@ -2734,8 +2734,11 @@ static void xs_tcp_tls_setup_socket(struct work_struct *work) lower_xprt = rcu_dereference(lower_clnt->cl_xprt); rcu_read_unlock(); - if (wait_on_bit_lock(&lower_xprt->state, XPRT_LOCKED, TASK_KILLABLE)) + if (wait_on_bit_lock(&lower_xprt->state, XPRT_LOCKED, TASK_KILLABLE)) { + /* XPRT_LOCKED was never acquired. */ + rpc_shutdown_client(lower_clnt); goto out_unlock; + } status = xs_tls_handshake_sync(lower_xprt, &upper_xprt->xprtsec); if (status) { @@ -2758,6 +2761,7 @@ static void xs_tcp_tls_setup_socket(struct work_struct *work) out_unlock: current_restore_flags(pflags, PF_MEMALLOC); upper_transport->clnt = NULL; + rpc_release_client(upper_clnt); xprt_unlock_connect(upper_xprt, upper_transport); return; @@ -2805,7 +2809,15 @@ static void xs_connect(struct rpc_xprt *xprt, struct rpc_task *task) } else dprintk("RPC: xs_connect scheduled xprt %p\n", xprt); - transport->clnt = task->tk_client; + /* + * Only the TLS connect_worker reads transport->clnt; pinning + * the upper rpc_clnt unconditionally would form a cycle with + * cl_xprt and prevent xprt destruction. + */ + if (xprt->xprtsec.policy != RPC_XPRTSEC_NONE) { + rpc_hold_client(task->tk_client); + transport->clnt = task->tk_client; + } queue_delayed_work(xprtiod_workqueue, &transport->connect_worker, delay); |