RDS: make sure we post recv buffers

If we get an ENOMEM during rds_ib_recv_refill, we might never come
back and refill again later.

This makes sure to kick krdsd into helping out.

Signed-off-by: Chris Mason <chris.mason@oracle.com>
Signed-off-by: Bang Nguyen <bang.nguyen@oracle.com>

diff --git a/net/rds/connection.c b/net/rds/connection.c
index fa4c3d3edbd78e7dca0aa2b7463c8045dfb2d885..6984ad2135ea9133d10e1cd9c9bbf3866bbad012 100644 (file)
--- a/net/rds/connection.c
+++ b/net/rds/connection.c
@@ -282,6 +282,8 @@ void rds_conn_shutdown(struct rds_connection *conn)
 
                wait_event(conn->c_waitq,
                           !test_bit(RDS_IN_XMIT, &conn->c_flags));
+               wait_event(conn->c_waitq,
+                          !test_bit(RDS_RECV_REFILL, &conn->c_flags));
 
                conn->c_trans->conn_shutdown(conn);
                rds_conn_reset(conn);

diff --git a/net/rds/ib.h b/net/rds/ib.h
index 0fa5cdcb61302b3f8996328dcdb63136e76f8327..a75f2ad011e9fa7af7736af5aee5465a58fb71c9 100644 (file)
--- a/net/rds/ib.h
+++ b/net/rds/ib.h
@@ -341,7 +341,7 @@ void rds_ib_recv_exit(void);
 int rds_ib_recv(struct rds_connection *conn);
 int rds_ib_recv_alloc_caches(struct rds_ib_connection *ic);
 void rds_ib_recv_free_caches(struct rds_ib_connection *ic);
-void rds_ib_recv_refill(struct rds_connection *conn, int prefill);
+void rds_ib_recv_refill(struct rds_connection *conn, int prefill, gfp_t gfp);
 void rds_ib_inc_free(struct rds_incoming *inc);
 int rds_ib_inc_copy_to_user(struct rds_incoming *inc, struct iovec *iov,
                             size_t size);

diff --git a/net/rds/ib_cm.c b/net/rds/ib_cm.c
index 306a42928edecafeff9fb7086dfcfc339531de5f..4b444f31d4de0f31a267018327fb12dd4ad2dd86 100644 (file)
--- a/net/rds/ib_cm.c
+++ b/net/rds/ib_cm.c
@@ -165,7 +165,7 @@ void rds_ib_cm_connect_complete(struct rds_connection *conn, struct rdma_cm_even
        rds_ib_recv_init_ring(ic);
        /* Post receive buffers - as a side effect, this will update
         * the posted credit count. */
-       rds_ib_recv_refill(conn, 1);
+       rds_ib_recv_refill(conn, 1, GFP_KERNEL);
 
        /* Tune RNR behavior */
        rds_ib_tune_rnr(ic, &qp_attr);

diff --git a/net/rds/ib_recv.c b/net/rds/ib_recv.c
index 2b4dee090988ecbdbbb0cc23b48c48e1093e34ac..8502b8c3fc670c3143838867703b244900a5e178 100644 (file)
--- a/net/rds/ib_recv.c
+++ b/net/rds/ib_recv.c
@@ -296,7 +296,7 @@ static struct rds_page_frag *rds_ib_refill_one_frag(struct rds_ib_connection *ic
 }
 
 static int rds_ib_recv_refill_one(struct rds_connection *conn,
-                                 struct rds_ib_recv_work *recv, int prefill)
+                                 struct rds_ib_recv_work *recv, gfp_t gfp)
 {
        struct rds_ib_connection *ic = conn->c_transport_data;
        struct ib_sge *sge;
@@ -304,7 +304,7 @@ static int rds_ib_recv_refill_one(struct rds_connection *conn,
        gfp_t slab_mask = GFP_NOWAIT;
        gfp_t page_mask = GFP_NOWAIT;
 
-       if (prefill) {
+       if (gfp & __GFP_WAIT) {
                slab_mask = GFP_KERNEL;
                page_mask = GFP_HIGHUSER;
        }
@@ -346,6 +346,26 @@ out:
        return ret;
 }
 
+static int acquire_refill(struct rds_connection *conn)
+{
+       return test_and_set_bit(RDS_RECV_REFILL, &conn->c_flags) == 0;
+}
+
+static void release_refill(struct rds_connection *conn)
+{
+       clear_bit(RDS_RECV_REFILL, &conn->c_flags);
+       smp_mb__after_clear_bit();
+       /*
+        * We don't use wait_on_bit()/wake_up_bit() because our waking is in a
+        * hot path and finding waiters is very rare.  We don't want to walk
+        * the system-wide hashed waitqueue buckets in the fast path only to
+        * almost never find waiters.
+        */
+       if (waitqueue_active(&conn->c_waitq))
+               wake_up_all(&conn->c_waitq);
+}
+
+
 /*
  * This tries to allocate and post unused work requests after making sure that
  * they have all the allocations they need to queue received fragments into
@@ -353,15 +373,24 @@ out:
  *
  * -1 is returned if posting fails due to temporary resource exhaustion.
  */
-void rds_ib_recv_refill(struct rds_connection *conn, int prefill)
+void rds_ib_recv_refill(struct rds_connection *conn, int prefill, gfp_t gfp)
 {
        struct rds_ib_connection *ic = conn->c_transport_data;
        struct rds_ib_recv_work *recv;
        struct ib_recv_wr *failed_wr;
        unsigned int posted = 0;
        int ret = 0;
+       int can_wait = gfp & __GFP_WAIT;
        u32 pos;
 
+       /*
+        * the goal here is to just make sure that someone, somewhere
+        * is posting buffers.  If we can't get the refill lock,
+        * let them do their thing
+        */
+       if (!acquire_refill(conn))
+               return;
+
        while ((prefill || rds_conn_up(conn))
                        && rds_ib_ring_alloc(&ic->i_recv_ring, 1, &pos)) {
                if (pos >= ic->i_recv_ring.w_nr) {
@@ -371,7 +400,7 @@ void rds_ib_recv_refill(struct rds_connection *conn, int prefill)
                }
 
                recv = &ic->i_recvs[pos];
-               ret = rds_ib_recv_refill_one(conn, recv, prefill);
+               ret = rds_ib_recv_refill_one(conn, recv, gfp);
                if (ret) {
                        break;
                }
@@ -398,6 +427,24 @@ void rds_ib_recv_refill(struct rds_connection *conn, int prefill)
 
        if (ret)
                rds_ib_ring_unalloc(&ic->i_recv_ring, 1);
+
+       release_refill(conn);
+
+       /* if we're called from the softirq handler, we'll be GFP_NOWAIT.
+        * in this case the ring being low is going to lead to more interrupts
+        * and we can safely let the softirq code take care of it unless the
+        * ring is completely empty.
+        *
+        * if we're called from krdsd, we'll be GFP_KERNEL.  In this case
+        * we might have raced with the softirq code while we had the refill
+        * lock held.  Use rds_ib_ring_low() instead of ring_empty to decide
+        * if we should requeue.
+        */
+       if (rds_conn_up(conn) &&
+          ((can_wait && rds_ib_ring_low(&ic->i_recv_ring)) ||
+          rds_ib_ring_empty(&ic->i_recv_ring))) {
+               queue_delayed_work(rds_wq, &conn->c_recv_w, 1);
+       }
 }
 
 /*
@@ -971,7 +1018,7 @@ void rds_ib_recv_cqe_handler(struct rds_ib_connection *ic,
                rds_ib_stats_inc(s_ib_rx_ring_empty);
 
        if (rds_ib_ring_low(&ic->i_recv_ring))
-               rds_ib_recv_refill(conn, 0);
+               rds_ib_recv_refill(conn, 0, GFP_NOWAIT);
 }
 
 int rds_ib_recv(struct rds_connection *conn)
@@ -980,8 +1027,10 @@ int rds_ib_recv(struct rds_connection *conn)
        int ret = 0;
 
        rdsdebug("conn %p\n", conn);
-       if (rds_conn_up(conn))
+       if (rds_conn_up(conn)) {
                rds_ib_attempt_ack(ic);
+               rds_ib_recv_refill(conn, 0, GFP_KERNEL);
+       }
 
        return ret;
 }

diff --git a/net/rds/rds.h b/net/rds/rds.h
index c033c63609225652f90cb29800b65eb84928b46e..5590164fbceac1b3ffdefda0e048ae7593c2bbbd 100644 (file)
--- a/net/rds/rds.h
+++ b/net/rds/rds.h
@@ -81,6 +81,7 @@ enum {
 #define RDS_LL_SEND_FULL       0
 #define RDS_RECONNECT_PENDING  1
 #define RDS_IN_XMIT            2
+#define RDS_RECV_REFILL                3
 
 struct rds_connection {
        struct hlist_node       c_hash_node;