xprtrdma: Fix MR list handling

[linux.git] / net / sunrpc / xprtrdma / verbs.c

diff --git a/net/sunrpc/xprtrdma/verbs.c b/net/sunrpc/xprtrdma/verbs.c
index 3a907537e2cf31cf471d7cb92dfbe85ebcc9d706..82361e7bbb51ca4512e07ec606be0ffef6179c2c 100644 (file)
--- a/net/sunrpc/xprtrdma/verbs.c
+++ b/net/sunrpc/xprtrdma/verbs.c
@@ -75,16 +75,15 @@
  * internal functions
  */
 static void rpcrdma_sendctx_put_locked(struct rpcrdma_sendctx *sc);
+static void rpcrdma_reqs_reset(struct rpcrdma_xprt *r_xprt);
 static void rpcrdma_reps_destroy(struct rpcrdma_buffer *buf);
 static void rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt);
 static void rpcrdma_mrs_destroy(struct rpcrdma_buffer *buf);
-static void rpcrdma_mr_free(struct rpcrdma_mr *mr);
 static struct rpcrdma_regbuf *
 rpcrdma_regbuf_alloc(size_t size, enum dma_data_direction direction,
                     gfp_t flags);
 static void rpcrdma_regbuf_dma_unmap(struct rpcrdma_regbuf *rb);
 static void rpcrdma_regbuf_free(struct rpcrdma_regbuf *rb);
-static void rpcrdma_post_recvs(struct rpcrdma_xprt *r_xprt, bool temp);
 
 /* Wait for outstanding transport work to finish. ib_drain_qp
  * handles the drains in the wrong order for us, so open code
@@ -170,7 +169,6 @@ rpcrdma_wc_receive(struct ib_cq *cq, struct ib_wc *wc)
                                   rdmab_addr(rep->rr_rdmabuf),
                                   wc->byte_len, DMA_FROM_DEVICE);
 
-       rpcrdma_post_recvs(r_xprt, false);
        rpcrdma_reply_handler(rep);
        return;
 
@@ -726,6 +724,7 @@ rpcrdma_ep_connect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
        ep->rep_connected = 0;
        xprt_clear_connected(xprt);
 
+       rpcrdma_reset_cwnd(r_xprt);
        rpcrdma_post_recvs(r_xprt, true);
 
        rc = rdma_connect(ia->ri_id, &ep->rep_remote_cma);
@@ -780,6 +779,7 @@ rpcrdma_ep_disconnect(struct rpcrdma_ep *ep, struct rpcrdma_ia *ia)
        trace_xprtrdma_disconnect(r_xprt, rc);
 
        rpcrdma_xprt_drain(r_xprt);
+       rpcrdma_reqs_reset(r_xprt);
 }
 
 /* Fixed-size circular FIFO queue. This implementation is wait-free and
@@ -965,7 +965,7 @@ rpcrdma_mrs_create(struct rpcrdma_xprt *r_xprt)
                mr->mr_xprt = r_xprt;
 
                spin_lock(&buf->rb_lock);
-               list_add(&mr->mr_list, &buf->rb_mrs);
+               rpcrdma_mr_push(mr, &buf->rb_mrs);
                list_add(&mr->mr_all, &buf->rb_all_mrs);
                spin_unlock(&buf->rb_lock);
        }
@@ -1042,6 +1042,26 @@ struct rpcrdma_req *rpcrdma_req_create(struct rpcrdma_xprt *r_xprt, size_t size,
        return NULL;
 }
 
+/**
+ * rpcrdma_reqs_reset - Reset all reqs owned by a transport
+ * @r_xprt: controlling transport instance
+ *
+ * ASSUMPTION: the rb_allreqs list is stable for the duration,
+ * and thus can be walked without holding rb_lock. Eg. the
+ * caller is holding the transport send lock to exclude
+ * device removal or disconnection.
+ */
+static void rpcrdma_reqs_reset(struct rpcrdma_xprt *r_xprt)
+{
+       struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
+       struct rpcrdma_req *req;
+
+       list_for_each_entry(req, &buf->rb_allreqs, rl_all) {
+               /* Credits are valid only for one connection */
+               req->rl_slot.rq_cong = 0;
+       }
+}
+
 static struct rpcrdma_rep *rpcrdma_rep_create(struct rpcrdma_xprt *r_xprt,
                                              bool temp)
 {
@@ -1141,7 +1161,6 @@ int rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt)
                list_add(&req->rl_list, &buf->rb_send_bufs);
        }
 
-       buf->rb_credits = 1;
        init_llist_head(&buf->rb_free_reps);
 
        rc = rpcrdma_sendctxs_create(r_xprt);
@@ -1163,10 +1182,19 @@ int rpcrdma_buffer_create(struct rpcrdma_xprt *r_xprt)
  */
 void rpcrdma_req_destroy(struct rpcrdma_req *req)
 {
+       struct rpcrdma_mr *mr;
+
        list_del(&req->rl_all);
 
-       while (!list_empty(&req->rl_free_mrs))
-               rpcrdma_mr_free(rpcrdma_mr_pop(&req->rl_free_mrs));
+       while ((mr = rpcrdma_mr_pop(&req->rl_free_mrs))) {
+               struct rpcrdma_buffer *buf = &mr->mr_xprt->rx_buf;
+
+               spin_lock(&buf->rb_lock);
+               list_del(&mr->mr_all);
+               spin_unlock(&buf->rb_lock);
+
+               frwr_release_mr(mr);
+       }
 
        rpcrdma_regbuf_free(req->rl_recvbuf);
        rpcrdma_regbuf_free(req->rl_sendbuf);
@@ -1174,24 +1202,28 @@ void rpcrdma_req_destroy(struct rpcrdma_req *req)
        kfree(req);
 }
 
-static void
-rpcrdma_mrs_destroy(struct rpcrdma_buffer *buf)
+/**
+ * rpcrdma_mrs_destroy - Release all of a transport's MRs
+ * @buf: controlling buffer instance
+ *
+ * Relies on caller holding the transport send lock to protect
+ * removing mr->mr_list from req->rl_free_mrs safely.
+ */
+static void rpcrdma_mrs_destroy(struct rpcrdma_buffer *buf)
 {
        struct rpcrdma_xprt *r_xprt = container_of(buf, struct rpcrdma_xprt,
                                                   rx_buf);
        struct rpcrdma_mr *mr;
-       unsigned int count;
 
-       count = 0;
        spin_lock(&buf->rb_lock);
        while ((mr = list_first_entry_or_null(&buf->rb_all_mrs,
                                              struct rpcrdma_mr,
                                              mr_all)) != NULL) {
+               list_del(&mr->mr_list);
                list_del(&mr->mr_all);
                spin_unlock(&buf->rb_lock);
 
                frwr_release_mr(mr);
-               count++;
                spin_lock(&buf->rb_lock);
        }
        spin_unlock(&buf->rb_lock);
@@ -1264,17 +1296,6 @@ void rpcrdma_mr_put(struct rpcrdma_mr *mr)
        rpcrdma_mr_push(mr, &mr->mr_req->rl_free_mrs);
 }
 
-static void rpcrdma_mr_free(struct rpcrdma_mr *mr)
-{
-       struct rpcrdma_xprt *r_xprt = mr->mr_xprt;
-       struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
-
-       mr->mr_req = NULL;
-       spin_lock(&buf->rb_lock);
-       rpcrdma_mr_push(mr, &buf->rb_mrs);
-       spin_unlock(&buf->rb_lock);
-}
-
 /**
  * rpcrdma_buffer_get - Get a request buffer
  * @buffers: Buffer pool from which to obtain a buffer
@@ -1455,8 +1476,13 @@ rpcrdma_ep_post(struct rpcrdma_ia *ia,
        return 0;
 }
 
-static void
-rpcrdma_post_recvs(struct rpcrdma_xprt *r_xprt, bool temp)
+/**
+ * rpcrdma_post_recvs - Refill the Receive Queue
+ * @r_xprt: controlling transport instance
+ * @temp: mark Receive buffers to be deleted after use
+ *
+ */
+void rpcrdma_post_recvs(struct rpcrdma_xprt *r_xprt, bool temp)
 {
        struct rpcrdma_buffer *buf = &r_xprt->rx_buf;
        struct rpcrdma_ep *ep = &r_xprt->rx_ep;