svcrdma: Remove transport reference counting

Jason tells me that a ULP cannot rely on getting an ESTABLISHED
and DISCONNECTED event pair for each connection, so transport
reference counting in the CM event handler will never be reliable.

Now that we have ib_drain_qp(), svcrdma should no longer need to
hold transport references while Sends and Receives are posted. So
remove the get/put call sites in the CM event handlers.

This eliminates a significant source of locked memory bus traffic.

Signed-off-by: Chuck Lever <chuck.lever@oracle.com>

diff --git a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
index 5bb97b5f4606d33928908f47b2003ef227b8269d..c6ea2903c21a47bf549d0dbdac6cde1f9390b3b2 100644 (file)
--- a/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_recvfrom.c
@@ -257,7 +257,6 @@ static int __svc_rdma_post_recv(struct svcxprt_rdma *rdma,
 {
        int ret;
 
-       svc_xprt_get(&rdma->sc_xprt);
        trace_svcrdma_post_recv(ctxt);
        ret = ib_post_recv(rdma->sc_qp, &ctxt->rc_recv_wr, NULL);
        if (ret)
@@ -267,7 +266,6 @@ static int __svc_rdma_post_recv(struct svcxprt_rdma *rdma,
 err_post:
        trace_svcrdma_rq_post_err(rdma, ret);
        svc_rdma_recv_ctxt_put(rdma, ctxt);
-       svc_xprt_put(&rdma->sc_xprt);
        return ret;
 }
 
@@ -344,15 +342,13 @@ static void svc_rdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
        spin_unlock(&rdma->sc_rq_dto_lock);
        if (!test_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags))
                svc_xprt_enqueue(&rdma->sc_xprt);
-       goto out;
+       return;
 
 flushed:
 post_err:
        svc_rdma_recv_ctxt_put(rdma, ctxt);
        set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags);
        svc_xprt_enqueue(&rdma->sc_xprt);
-out:
-       svc_xprt_put(&rdma->sc_xprt);
 }
 
 /**

diff --git a/net/sunrpc/xprtrdma/svc_rdma_rw.c b/net/sunrpc/xprtrdma/svc_rdma_rw.c
index c16d10601d6591bf303085f704011d5758abae33..fe54cbe97a46066ae1cef13920117370e1c3968d 100644 (file)
--- a/net/sunrpc/xprtrdma/svc_rdma_rw.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_rw.c
@@ -164,7 +164,6 @@ static void svc_rdma_cc_init(struct svcxprt_rdma *rdma,
 {
        svc_rdma_cc_cid_init(rdma, &cc->cc_cid);
        cc->cc_rdma = rdma;
-       svc_xprt_get(&rdma->sc_xprt);
 
        INIT_LIST_HEAD(&cc->cc_rwctxts);
        cc->cc_sqecount = 0;
@@ -184,7 +183,6 @@ static void svc_rdma_cc_release(struct svc_rdma_chunk_ctxt *cc,
                                    ctxt->rw_nents, dir);
                svc_rdma_put_rw_ctxt(rdma, ctxt);
        }
-       svc_xprt_put(&rdma->sc_xprt);
 }
 
 /* State for sending a Write or Reply chunk.

diff --git a/net/sunrpc/xprtrdma/svc_rdma_sendto.c b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
index 73d46e8cdc16374c11b6f1b28cb0d8a872d108a8..7b94d971feb3b7359bef0588be0804de96c96392 100644 (file)
--- a/net/sunrpc/xprtrdma/svc_rdma_sendto.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_sendto.c
@@ -291,8 +291,6 @@ static void svc_rdma_wc_send(struct ib_cq *cq, struct ib_wc *wc)
                set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags);
                svc_xprt_enqueue(&rdma->sc_xprt);
        }
-
-       svc_xprt_put(&rdma->sc_xprt);
 }
 
 /**
@@ -330,7 +328,6 @@ int svc_rdma_send(struct svcxprt_rdma *rdma, struct svc_rdma_send_ctxt *ctxt)
                        continue;
                }
 
-               svc_xprt_get(&rdma->sc_xprt);
                trace_svcrdma_post_send(ctxt);
                ret = ib_post_send(rdma->sc_qp, wr, NULL);
                if (ret)
@@ -340,7 +337,6 @@ int svc_rdma_send(struct svcxprt_rdma *rdma, struct svc_rdma_send_ctxt *ctxt)
 
        trace_svcrdma_sq_post_err(rdma, ret);
        set_bit(XPT_CLOSE, &rdma->sc_xprt.xpt_flags);
-       svc_xprt_put(&rdma->sc_xprt);
        wake_up(&rdma->sc_send_wait);
        return ret;
 }

diff --git a/net/sunrpc/xprtrdma/svc_rdma_transport.c b/net/sunrpc/xprtrdma/svc_rdma_transport.c
index 3da7901a49e63f3ce919cafe73a8c8f53026587c..aa60f75c8c1d53ac8f4ff4aa1c9e95d2ee47d6f3 100644 (file)
--- a/net/sunrpc/xprtrdma/svc_rdma_transport.c
+++ b/net/sunrpc/xprtrdma/svc_rdma_transport.c
@@ -271,7 +271,6 @@ static int rdma_cma_handler(struct rdma_cm_id *cma_id,
        switch (event->event) {
        case RDMA_CM_EVENT_ESTABLISHED:
                /* Accept complete */
-               svc_xprt_get(xprt);
                dprintk("svcrdma: Connection completed on DTO xprt=%p, "
                        "cm_id=%p\n", xprt, cma_id);
                clear_bit(RDMAXPRT_CONN_PENDING, &rdma->sc_flags);
@@ -282,7 +281,6 @@ static int rdma_cma_handler(struct rdma_cm_id *cma_id,
                        xprt, cma_id);
                set_bit(XPT_CLOSE, &xprt->xpt_flags);
                svc_xprt_enqueue(xprt);
-               svc_xprt_put(xprt);
                break;
        case RDMA_CM_EVENT_DEVICE_REMOVAL:
                dprintk("svcrdma: Device removal cma_id=%p, xprt = %p, "
@@ -290,7 +288,6 @@ static int rdma_cma_handler(struct rdma_cm_id *cma_id,
                        rdma_event_msg(event->event), event->event);
                set_bit(XPT_CLOSE, &xprt->xpt_flags);
                svc_xprt_enqueue(xprt);
-               svc_xprt_put(xprt);
                break;
        default:
                dprintk("svcrdma: Unexpected event on DTO endpoint %p, "
@@ -539,24 +536,11 @@ static struct svc_xprt *svc_rdma_accept(struct svc_xprt *xprt)
        return NULL;
 }
 
-/*
- * When connected, an svc_xprt has at least two references:
- *
- * - A reference held by the cm_id between the ESTABLISHED and
- *   DISCONNECTED events. If the remote peer disconnected first, this
- *   reference could be gone.
- *
- * - A reference held by the svc_recv code that called this function
- *   as part of close processing.
- *
- * At a minimum one references should still be held.
- */
 static void svc_rdma_detach(struct svc_xprt *xprt)
 {
        struct svcxprt_rdma *rdma =
                container_of(xprt, struct svcxprt_rdma, sc_xprt);
 
-       /* Disconnect and flush posted WQE */
        rdma_disconnect(rdma->sc_cm_id);
 }
 
@@ -566,6 +550,7 @@ static void __svc_rdma_free(struct work_struct *work)
                container_of(work, struct svcxprt_rdma, sc_work);
        struct svc_xprt *xprt = &rdma->sc_xprt;
 
+       /* This blocks until the Completion Queues are empty */
        if (rdma->sc_qp && !IS_ERR(rdma->sc_qp))
                ib_drain_qp(rdma->sc_qp);