[ofa-general] RDS flow control
Jon Mason
jon at opengridcomputing.com
Tue May 20 13:45:22 PDT 2008
On Mon, May 19, 2008 at 10:05:59AM +0200, Olaf Kirch wrote:
>
> > However, I'm still seeing performance degradation of ~5% with some packet
> > sizes. And that is *just* the overhead from exchanging the credit information
> > and checking it - at some point we need to take a spinlock, and that seems
> > to delay things just enough to make a dent in my throughput graph.
>
> Here's an updated version of the flow control patch - which is now completely
> lockless, and uses a single atomic_t to hold both credit counters. This has
> given me back close to full performance in my testing (throughput seems to be
> down less than 1%, which is almost within the noise range).
>
> I'll push it to my git tree a little later today, so folks can test it if
> they like.
Works well on my setup.
With proper flow control, there should no longer be a need for rnr_retry (as there
should always be a posted recv buffer waiting for the incoming data). I did a
quick test and removed it and everything seemed to be happy on my rds-stress run.
Thanks for pulling this in.
Jon
>
> Olaf
> --
> Olaf Kirch | --- o --- Nous sommes du soleil we love when we play
> okir at lst.de | / | \ sol.dhoop.naytheet.ah kin.ir.samse.qurax
> ----
> From: Olaf Kirch <olaf.kirch at oracle.com>
> Subject: RDS: Implement IB flow control
>
> Here it is - flow control for RDS/IB.
>
> This patch is still very much experimental. Here's the essentials
>
> - The approach chosen here uses a credit-based flow control
> mechanism. Every SEND WR (including ACKs) consumes one credit,
> and if the sender runs out of credits, it stalls.
>
> - As new receive buffers are posted, credits are transferred to the
> remote node (using yet another RDS header byte for this).
>
> - Flow control is negotiated during connection setup. Initial credits
> are exchanged in the rds_ib_connect_private sruct - sending a value
> of zero (which is also the default for older protocol versions)
> means no flow control.
>
> - We avoid deadlock (both nodes depleting their credits, and being
> unable to inform the peer of newly posted buffers) by requiring
> that the last credit can only be used if we're posting new credits
> to the peer.
>
> The approach implemented here is lock-free; preliminary tests show
> the impact on throughput to be less than 1%, and the impact on RTT,
> CPU, TX delay and other metrics to be below the noise threshold.
>
> Flow control is configurable via sysctl. It only affects newly created
> connections however - so your best bet is to set this right after loading
> the RDS module.
>
> Signed-off-by: Olaf Kirch <olaf.kirch at oracle.com>
> ---
> net/rds/ib.c | 1
> net/rds/ib.h | 30 ++++++++
> net/rds/ib_cm.c | 49 ++++++++++++-
> net/rds/ib_recv.c | 48 +++++++++---
> net/rds/ib_send.c | 194 ++++++++++++++++++++++++++++++++++++++++++++++++++++
> net/rds/ib_stats.c | 3
> net/rds/ib_sysctl.c | 10 ++
> net/rds/rds.h | 4 -
> 8 files changed, 325 insertions(+), 14 deletions(-)
>
> Index: ofa_kernel-1.3/net/rds/ib.h
> ===================================================================
> --- ofa_kernel-1.3.orig/net/rds/ib.h
> +++ ofa_kernel-1.3/net/rds/ib.h
> @@ -46,6 +46,7 @@ struct rds_ib_connect_private {
> __be16 dp_protocol_minor_mask; /* bitmask */
> __be32 dp_reserved1;
> __be64 dp_ack_seq;
> + __be32 dp_credit; /* non-zero enables flow ctl */
> };
>
> struct rds_ib_send_work {
> @@ -110,15 +111,32 @@ struct rds_ib_connection {
> struct ib_sge i_ack_sge;
> u64 i_ack_dma;
> unsigned long i_ack_queued;
> +
> + /* Flow control related information
> + *
> + * Our algorithm uses a pair variables that we need to access
> + * atomically - one for the send credits, and one posted
> + * recv credits we need to transfer to remote.
> + * Rather than protect them using a slow spinlock, we put both into
> + * a single atomic_t and update it using cmpxchg
> + */
> + atomic_t i_credits;
>
> /* Protocol version specific information */
> unsigned int i_hdr_idx; /* 1 (old) or 0 (3.1 or later) */
> + unsigned int i_flowctl : 1; /* enable/disable flow ctl */
>
> /* Batched completions */
> unsigned int i_unsignaled_wrs;
> long i_unsignaled_bytes;
> };
>
> +/* This assumes that atomic_t is at least 32 bits */
> +#define IB_GET_SEND_CREDITS(v) ((v) & 0xffff)
> +#define IB_GET_POST_CREDITS(v) ((v) >> 16)
> +#define IB_SET_SEND_CREDITS(v) ((v) & 0xffff)
> +#define IB_SET_POST_CREDITS(v) ((v) << 16)
> +
> struct rds_ib_ipaddr {
> struct list_head list;
> __be32 ipaddr;
> @@ -153,14 +171,17 @@ struct rds_ib_statistics {
> unsigned long s_ib_tx_cq_call;
> unsigned long s_ib_tx_cq_event;
> unsigned long s_ib_tx_ring_full;
> + unsigned long s_ib_tx_throttle;
> unsigned long s_ib_tx_sg_mapping_failure;
> unsigned long s_ib_tx_stalled;
> + unsigned long s_ib_tx_credit_updates;
> unsigned long s_ib_rx_cq_call;
> unsigned long s_ib_rx_cq_event;
> unsigned long s_ib_rx_ring_empty;
> unsigned long s_ib_rx_refill_from_cq;
> unsigned long s_ib_rx_refill_from_thread;
> unsigned long s_ib_rx_alloc_limit;
> + unsigned long s_ib_rx_credit_updates;
> unsigned long s_ib_ack_sent;
> unsigned long s_ib_ack_send_failure;
> unsigned long s_ib_ack_send_delayed;
> @@ -244,6 +265,8 @@ void rds_ib_flush_mrs(void);
> int __init rds_ib_recv_init(void);
> void rds_ib_recv_exit(void);
> int rds_ib_recv(struct rds_connection *conn);
> +int rds_ib_recv_refill(struct rds_connection *conn, gfp_t kptr_gfp,
> + gfp_t page_gfp, int prefill);
> void rds_ib_inc_purge(struct rds_incoming *inc);
> void rds_ib_inc_free(struct rds_incoming *inc);
> int rds_ib_inc_copy_to_user(struct rds_incoming *inc, struct iovec *iov,
> @@ -252,6 +275,7 @@ void rds_ib_recv_cq_comp_handler(struct
> void rds_ib_recv_init_ring(struct rds_ib_connection *ic);
> void rds_ib_recv_clear_ring(struct rds_ib_connection *ic);
> void rds_ib_recv_init_ack(struct rds_ib_connection *ic);
> +void rds_ib_attempt_ack(struct rds_ib_connection *ic);
> void rds_ib_ack_send_complete(struct rds_ib_connection *ic);
> u64 rds_ib_piggyb_ack(struct rds_ib_connection *ic);
>
> @@ -266,12 +290,17 @@ u32 rds_ib_ring_completed(struct rds_ib_
> extern wait_queue_head_t rds_ib_ring_empty_wait;
>
> /* ib_send.c */
> +void rds_ib_xmit_complete(struct rds_connection *conn);
> int rds_ib_xmit(struct rds_connection *conn, struct rds_message *rm,
> unsigned int hdr_off, unsigned int sg, unsigned int off);
> void rds_ib_send_cq_comp_handler(struct ib_cq *cq, void *context);
> void rds_ib_send_init_ring(struct rds_ib_connection *ic);
> void rds_ib_send_clear_ring(struct rds_ib_connection *ic);
> int rds_ib_xmit_rdma(struct rds_connection *conn, struct rds_rdma_op *op);
> +void rds_ib_send_add_credits(struct rds_connection *conn, unsigned int credits);
> +void rds_ib_advertise_credits(struct rds_connection *conn, unsigned int posted);
> +int rds_ib_send_grab_credits(struct rds_ib_connection *ic, u32 wanted,
> + u32 *adv_credits);
>
> /* ib_stats.c */
> RDS_DECLARE_PER_CPU(struct rds_ib_statistics, rds_ib_stats);
> @@ -287,6 +316,7 @@ extern unsigned long rds_ib_sysctl_max_r
> extern unsigned long rds_ib_sysctl_max_unsig_wrs;
> extern unsigned long rds_ib_sysctl_max_unsig_bytes;
> extern unsigned long rds_ib_sysctl_max_recv_allocation;
> +extern unsigned int rds_ib_sysctl_flow_control;
> extern ctl_table rds_ib_sysctl_table[];
>
> /*
> Index: ofa_kernel-1.3/net/rds/ib_cm.c
> ===================================================================
> --- ofa_kernel-1.3.orig/net/rds/ib_cm.c
> +++ ofa_kernel-1.3/net/rds/ib_cm.c
> @@ -55,6 +55,22 @@ static void rds_ib_set_protocol(struct r
> }
>
> /*
> + * Set up flow control
> + */
> +static void rds_ib_set_flow_control(struct rds_connection *conn, u32 credits)
> +{
> + struct rds_ib_connection *ic = conn->c_transport_data;
> +
> + if (rds_ib_sysctl_flow_control && credits != 0) {
> + /* We're doing flow control */
> + ic->i_flowctl = 1;
> + rds_ib_send_add_credits(conn, credits);
> + } else {
> + ic->i_flowctl = 0;
> + }
> +}
> +
> +/*
> * Connection established.
> * We get here for both outgoing and incoming connection.
> */
> @@ -72,12 +88,16 @@ static void rds_ib_connect_complete(stru
> rds_ib_set_protocol(conn,
> RDS_PROTOCOL(dp->dp_protocol_major,
> dp->dp_protocol_minor));
> + rds_ib_set_flow_control(conn, be32_to_cpu(dp->dp_credit));
> }
>
> - rdsdebug("RDS/IB: ib conn complete on %u.%u.%u.%u version %u.%u\n",
> + printk(KERN_NOTICE "RDS/IB: connected to %u.%u.%u.%u version %u.%u%s\n",
> NIPQUAD(conn->c_laddr),
> RDS_PROTOCOL_MAJOR(conn->c_version),
> - RDS_PROTOCOL_MINOR(conn->c_version));
> + RDS_PROTOCOL_MINOR(conn->c_version),
> + ic->i_flowctl? ", flow control" : "");
> +
> + rds_ib_recv_refill(conn, GFP_KERNEL, GFP_HIGHUSER, 1);
>
> /* Tune the RNR timeout. We use a rather low timeout, but
> * not the absolute minimum - this should be tunable.
> @@ -129,6 +149,24 @@ static void rds_ib_cm_fill_conn_param(st
> dp->dp_protocol_minor_mask = cpu_to_be16(RDS_IB_SUPPORTED_PROTOCOLS);
> dp->dp_ack_seq = rds_ib_piggyb_ack(ic);
>
> + /* Advertise flow control.
> + *
> + * Major chicken and egg alert!
> + * We would like to post receive buffers before we get here (eg.
> + * in rds_ib_setup_qp), so that we can give the peer an accurate
> + * credit value.
> + * Unfortunately we can't post receive buffers until we've finished
> + * protocol negotiation, and know in which order data and payload
> + * are arranged.
> + *
> + * What we do here is we give the peer a small initial credit, and
> + * initialize the number of posted buffers to a negative value.
> + */
> + if (ic->i_flowctl) {
> + atomic_set(&ic->i_credits, IB_SET_POST_CREDITS(-4));
> + dp->dp_credit = cpu_to_be32(4);
> + }
> +
> conn_param->private_data = dp;
> conn_param->private_data_len = sizeof(*dp);
> }
> @@ -363,6 +401,7 @@ static int rds_ib_cm_handle_connect(stru
> ic = conn->c_transport_data;
>
> rds_ib_set_protocol(conn, version);
> + rds_ib_set_flow_control(conn, be32_to_cpu(dp->dp_credit));
>
> /* If the peer gave us the last packet it saw, process this as if
> * we had received a regular ACK. */
> @@ -428,6 +467,7 @@ out:
> static int rds_ib_cm_initiate_connect(struct rdma_cm_id *cm_id)
> {
> struct rds_connection *conn = cm_id->context;
> + struct rds_ib_connection *ic = conn->c_transport_data;
> struct rdma_conn_param conn_param;
> struct rds_ib_connect_private dp;
> int ret;
> @@ -435,6 +475,7 @@ static int rds_ib_cm_initiate_connect(st
> /* If the peer doesn't do protocol negotiation, we must
> * default to RDSv3.0 */
> rds_ib_set_protocol(conn, RDS_PROTOCOL_3_0);
> + ic->i_flowctl = rds_ib_sysctl_flow_control; /* advertise flow control */
>
> ret = rds_ib_setup_qp(conn);
> if (ret) {
> @@ -688,6 +729,10 @@ void rds_ib_conn_shutdown(struct rds_con
> #endif
> ic->i_ack_recv = 0;
>
> + /* Clear flow control state */
> + ic->i_flowctl = 0;
> + atomic_set(&ic->i_credits, 0);
> +
> if (ic->i_ibinc) {
> rds_inc_put(&ic->i_ibinc->ii_inc);
> ic->i_ibinc = NULL;
> Index: ofa_kernel-1.3/net/rds/ib_recv.c
> ===================================================================
> --- ofa_kernel-1.3.orig/net/rds/ib_recv.c
> +++ ofa_kernel-1.3/net/rds/ib_recv.c
> @@ -220,16 +220,17 @@ out:
> * -1 is returned if posting fails due to temporary resource exhaustion.
> */
> int rds_ib_recv_refill(struct rds_connection *conn, gfp_t kptr_gfp,
> - gfp_t page_gfp)
> + gfp_t page_gfp, int prefill)
> {
> 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;
> u32 pos;
>
> - while (rds_conn_up(conn) && rds_ib_ring_alloc(&ic->i_recv_ring, 1, &pos)) {
> -
> + while ((prefill || rds_conn_up(conn))
> + && rds_ib_ring_alloc(&ic->i_recv_ring, 1, &pos)) {
> if (pos >= ic->i_recv_ring.w_nr) {
> printk(KERN_NOTICE "Argh - ring alloc returned pos=%u\n",
> pos);
> @@ -257,8 +258,14 @@ int rds_ib_recv_refill(struct rds_connec
> ret = -1;
> break;
> }
> +
> + posted++;
> }
>
> + /* We're doing flow control - update the window. */
> + if (ic->i_flowctl && posted)
> + rds_ib_advertise_credits(conn, posted);
> +
> if (ret)
> rds_ib_ring_unalloc(&ic->i_recv_ring, 1);
> return ret;
> @@ -436,7 +443,7 @@ static u64 rds_ib_get_ack(struct rds_ib_
> #endif
>
>
> -static void rds_ib_send_ack(struct rds_ib_connection *ic)
> +static void rds_ib_send_ack(struct rds_ib_connection *ic, unsigned int adv_credits)
> {
> struct rds_header *hdr = ic->i_ack;
> struct ib_send_wr *failed_wr;
> @@ -448,6 +455,7 @@ static void rds_ib_send_ack(struct rds_i
> rdsdebug("send_ack: ic %p ack %llu\n", ic, (unsigned long long) seq);
> rds_message_populate_header(hdr, 0, 0, 0);
> hdr->h_ack = cpu_to_be64(seq);
> + hdr->h_credit = adv_credits;
> rds_message_make_checksum(hdr);
> ic->i_ack_queued = jiffies;
>
> @@ -460,6 +468,8 @@ static void rds_ib_send_ack(struct rds_i
> set_bit(IB_ACK_REQUESTED, &ic->i_ack_flags);
>
> rds_ib_stats_inc(s_ib_ack_send_failure);
> + /* Need to finesse this later. */
> + BUG();
> } else
> rds_ib_stats_inc(s_ib_ack_sent);
> }
> @@ -502,15 +512,27 @@ static void rds_ib_send_ack(struct rds_i
> * When we get here, we're called from the recv queue handler.
> * Check whether we ought to transmit an ACK.
> */
> -static void rds_ib_attempt_ack(struct rds_ib_connection *ic)
> +void rds_ib_attempt_ack(struct rds_ib_connection *ic)
> {
> + unsigned int adv_credits;
> +
> if (!test_bit(IB_ACK_REQUESTED, &ic->i_ack_flags))
> return;
> - if (!test_and_set_bit(IB_ACK_IN_FLIGHT, &ic->i_ack_flags)) {
> - clear_bit(IB_ACK_REQUESTED, &ic->i_ack_flags);
> - rds_ib_send_ack(ic);
> - } else
> +
> + if (test_and_set_bit(IB_ACK_IN_FLIGHT, &ic->i_ack_flags)) {
> rds_ib_stats_inc(s_ib_ack_send_delayed);
> + return;
> + }
> +
> + /* Can we get a send credit? */
> + if (!rds_ib_send_grab_credits(ic, 1, &adv_credits)) {
> + rds_ib_stats_inc(s_ib_tx_throttle);
> + clear_bit(IB_ACK_IN_FLIGHT, &ic->i_ack_flags);
> + return;
> + }
> +
> + clear_bit(IB_ACK_REQUESTED, &ic->i_ack_flags);
> + rds_ib_send_ack(ic, adv_credits);
> }
>
> /*
> @@ -706,6 +728,10 @@ void rds_ib_process_recv(struct rds_conn
> state->ack_recv = be64_to_cpu(ihdr->h_ack);
> state->ack_recv_valid = 1;
>
> + /* Process the credits update if there was one */
> + if (ihdr->h_credit)
> + rds_ib_send_add_credits(conn, ihdr->h_credit);
> +
> if (ihdr->h_sport == 0 && ihdr->h_dport == 0 && byte_len == 0) {
> /* This is an ACK-only packet. The fact that it gets
> * special treatment here is that historically, ACKs
> @@ -877,7 +903,7 @@ void rds_ib_recv_cq_comp_handler(struct
>
> if (mutex_trylock(&ic->i_recv_mutex)) {
> if (rds_ib_recv_refill(conn, GFP_ATOMIC,
> - GFP_ATOMIC | __GFP_HIGHMEM))
> + GFP_ATOMIC | __GFP_HIGHMEM, 0))
> ret = -EAGAIN;
> else
> rds_ib_stats_inc(s_ib_rx_refill_from_cq);
> @@ -901,7 +927,7 @@ int rds_ib_recv(struct rds_connection *c
> * we're really low and we want the caller to back off for a bit.
> */
> mutex_lock(&ic->i_recv_mutex);
> - if (rds_ib_recv_refill(conn, GFP_KERNEL, GFP_HIGHUSER))
> + if (rds_ib_recv_refill(conn, GFP_KERNEL, GFP_HIGHUSER, 0))
> ret = -ENOMEM;
> else
> rds_ib_stats_inc(s_ib_rx_refill_from_thread);
> Index: ofa_kernel-1.3/net/rds/ib.c
> ===================================================================
> --- ofa_kernel-1.3.orig/net/rds/ib.c
> +++ ofa_kernel-1.3/net/rds/ib.c
> @@ -187,6 +187,7 @@ static void rds_ib_exit(void)
>
> struct rds_transport rds_ib_transport = {
> .laddr_check = rds_ib_laddr_check,
> + .xmit_complete = rds_ib_xmit_complete,
> .xmit = rds_ib_xmit,
> .xmit_cong_map = NULL,
> .xmit_rdma = rds_ib_xmit_rdma,
> Index: ofa_kernel-1.3/net/rds/ib_send.c
> ===================================================================
> --- ofa_kernel-1.3.orig/net/rds/ib_send.c
> +++ ofa_kernel-1.3/net/rds/ib_send.c
> @@ -245,6 +245,144 @@ void rds_ib_send_cq_comp_handler(struct
> }
> }
>
> +/*
> + * This is the main function for allocating credits when sending
> + * messages.
> + *
> + * Conceptually, we have two counters:
> + * - send credits: this tells us how many WRs we're allowed
> + * to submit without overruning the reciever's queue. For
> + * each SEND WR we post, we decrement this by one.
> + *
> + * - posted credits: this tells us how many WRs we recently
> + * posted to the receive queue. This value is transferred
> + * to the peer as a "credit update" in a RDS header field.
> + * Every time we transmit credits to the peer, we subtract
> + * the amount of transferred credits from this counter.
> + *
> + * It is essential that we avoid situations where both sides have
> + * exhausted their send credits, and are unable to send new credits
> + * to the peer. We achieve this by requiring that we send at least
> + * one credit update to the peer before exhausting our credits.
> + * When new credits arrive, we subtract one credit that is withheld
> + * until we've posted new buffers and are ready to transmit these
> + * credits (see rds_ib_send_add_credits below).
> + *
> + * The RDS send code is essentially single-threaded; rds_send_xmit
> + * grabs c_send_sem to ensure exclusive access to the send ring.
> + * However, the ACK sending code is independent and can race with
> + * message SENDs.
> + *
> + * In the send path, we need to update the counters for send credits
> + * and the counter of posted buffers atomically - when we use the
> + * last available credit, we cannot allow another thread to race us
> + * and grab the posted credits counter. Hence, we have to use a
> + * spinlock to protect the credit counter, or use atomics.
> + *
> + * Spinlocks shared between the send and the receive path are bad,
> + * because they create unnecessary delays. An early implementation
> + * using a spinlock showed a 5% degradation in throughput at some
> + * loads.
> + *
> + * This implementation avoids spinlocks completely, putting both
> + * counters into a single atomic, and updating that atomic using
> + * atomic_add (in the receive path, when receiving fresh credits),
> + * and using atomic_cmpxchg when updating the two counters.
> + */
> +int rds_ib_send_grab_credits(struct rds_ib_connection *ic,
> + u32 wanted, u32 *adv_credits)
> +{
> + unsigned int avail, posted, got = 0, advertise;
> + long oldval, newval;
> +
> + *adv_credits = 0;
> + if (!ic->i_flowctl)
> + return wanted;
> +
> +try_again:
> + advertise = 0;
> + oldval = newval = atomic_read(&ic->i_credits);
> + posted = IB_GET_POST_CREDITS(oldval);
> + avail = IB_GET_SEND_CREDITS(oldval);
> +
> + rdsdebug("rds_ib_send_grab_credits(%u): credits=%u posted=%u\n",
> + wanted, avail, posted);
> +
> + /* The last credit must be used to send a credit updated. */
> + if (avail && !posted)
> + avail--;
> +
> + if (avail < wanted) {
> + struct rds_connection *conn = ic->i_cm_id->context;
> +
> + /* Oops, there aren't that many credits left! */
> + set_bit(RDS_LL_SEND_FULL, &conn->c_flags);
> + got = avail;
> + } else {
> + /* Sometimes you get what you want, lalala. */
> + got = wanted;
> + }
> + newval -= IB_SET_SEND_CREDITS(got);
> +
> + if (got && posted) {
> + advertise = min_t(unsigned int, posted, RDS_MAX_ADV_CREDIT);
> + newval -= IB_SET_POST_CREDITS(advertise);
> + }
> +
> + /* Finally bill everything */
> + if (atomic_cmpxchg(&ic->i_credits, oldval, newval) != oldval)
> + goto try_again;
> +
> + *adv_credits = advertise;
> + return got;
> +}
> +
> +void rds_ib_send_add_credits(struct rds_connection *conn, unsigned int credits)
> +{
> + struct rds_ib_connection *ic = conn->c_transport_data;
> +
> + if (credits == 0)
> + return;
> +
> + rdsdebug("rds_ib_send_add_credits(%u): current=%u%s\n",
> + credits,
> + IB_GET_SEND_CREDITS(atomic_read(&ic->i_credits)),
> + test_bit(RDS_LL_SEND_FULL, &conn->c_flags)? ", ll_send_full" : "");
> +
> + atomic_add(IB_SET_SEND_CREDITS(credits), &ic->i_credits);
> + if (test_and_clear_bit(RDS_LL_SEND_FULL, &conn->c_flags))
> + queue_delayed_work(rds_wq, &conn->c_send_w, 0);
> +
> + WARN_ON(IB_GET_SEND_CREDITS(credits) >= 16384);
> +
> + rds_ib_stats_inc(s_ib_rx_credit_updates);
> +}
> +
> +void rds_ib_advertise_credits(struct rds_connection *conn, unsigned int posted)
> +{
> + struct rds_ib_connection *ic = conn->c_transport_data;
> +
> + if (posted == 0)
> + return;
> +
> + atomic_add(IB_SET_POST_CREDITS(posted), &ic->i_credits);
> +
> + /* Decide whether to send an update to the peer now.
> + * If we would send a credit update for every single buffer we
> + * post, we would end up with an ACK storm (ACK arrives,
> + * consumes buffer, we refill the ring, send ACK to remote
> + * advertising the newly posted buffer... ad inf)
> + *
> + * Performance pretty much depends on how often we send
> + * credit updates - too frequent updates mean lots of ACKs.
> + * Too infrequent updates, and the peer will run out of
> + * credits and has to throttle.
> + * For the time being, 16 seems to be a good compromise.
> + */
> + if (IB_GET_POST_CREDITS(atomic_read(&ic->i_credits)) >= 16)
> + set_bit(IB_ACK_REQUESTED, &ic->i_ack_flags);
> +}
> +
> static inline void
> rds_ib_xmit_populate_wr(struct rds_ib_connection *ic,
> struct rds_ib_send_work *send, unsigned int pos,
> @@ -307,6 +445,8 @@ int rds_ib_xmit(struct rds_connection *c
> u32 pos;
> u32 i;
> u32 work_alloc;
> + u32 credit_alloc;
> + u32 adv_credits = 0;
> int send_flags = 0;
> int sent;
> int ret;
> @@ -314,6 +454,7 @@ int rds_ib_xmit(struct rds_connection *c
> BUG_ON(off % RDS_FRAG_SIZE);
> BUG_ON(hdr_off != 0 && hdr_off != sizeof(struct rds_header));
>
> + /* FIXME we may overallocate here */
> if (be32_to_cpu(rm->m_inc.i_hdr.h_len) == 0)
> i = 1;
> else
> @@ -327,8 +468,29 @@ int rds_ib_xmit(struct rds_connection *c
> goto out;
> }
>
> + credit_alloc = work_alloc;
> + if (ic->i_flowctl) {
> + credit_alloc = rds_ib_send_grab_credits(ic, work_alloc, &adv_credits);
> + if (credit_alloc < work_alloc) {
> + rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc - credit_alloc);
> + work_alloc = credit_alloc;
> + }
> + if (work_alloc == 0) {
> + rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc);
> + rds_ib_stats_inc(s_ib_tx_throttle);
> + ret = -ENOMEM;
> + goto out;
> + }
> + }
> +
> /* map the message the first time we see it */
> if (ic->i_rm == NULL) {
> + /*
> + printk(KERN_NOTICE "rds_ib_xmit prep msg dport=%u flags=0x%x len=%d\n",
> + be16_to_cpu(rm->m_inc.i_hdr.h_dport),
> + rm->m_inc.i_hdr.h_flags,
> + be32_to_cpu(rm->m_inc.i_hdr.h_len));
> + */
> if (rm->m_nents) {
> rm->m_count = ib_dma_map_sg(dev,
> rm->m_sg, rm->m_nents, DMA_TO_DEVICE);
> @@ -449,6 +611,24 @@ add_header:
> * have been set up to point to the right header buffer. */
> memcpy(&ic->i_send_hdrs[pos], &rm->m_inc.i_hdr, sizeof(struct rds_header));
>
> + if (0) {
> + struct rds_header *hdr = &ic->i_send_hdrs[pos];
> +
> + printk(KERN_NOTICE "send WR dport=%u flags=0x%x len=%d\n",
> + be16_to_cpu(hdr->h_dport),
> + hdr->h_flags,
> + be32_to_cpu(hdr->h_len));
> + }
> + if (adv_credits) {
> + struct rds_header *hdr = &ic->i_send_hdrs[pos];
> +
> + /* add credit and redo the header checksum */
> + hdr->h_credit = adv_credits;
> + rds_message_make_checksum(hdr);
> + adv_credits = 0;
> + rds_ib_stats_inc(s_ib_tx_credit_updates);
> + }
> +
> if (prev)
> prev->s_wr.next = &send->s_wr;
> prev = send;
> @@ -472,6 +652,8 @@ add_header:
> rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc - i);
> work_alloc = i;
> }
> + if (ic->i_flowctl && i < credit_alloc)
> + rds_ib_send_add_credits(conn, credit_alloc - i);
>
> /* XXX need to worry about failed_wr and partial sends. */
> failed_wr = &first->s_wr;
> @@ -487,11 +669,14 @@ add_header:
> ic->i_rm = prev->s_rm;
> prev->s_rm = NULL;
> }
> + /* Finesse this later */
> + BUG();
> goto out;
> }
>
> ret = sent;
> out:
> + BUG_ON(adv_credits);
> return ret;
> }
>
> @@ -630,3 +815,12 @@ int rds_ib_xmit_rdma(struct rds_connecti
> out:
> return ret;
> }
> +
> +void rds_ib_xmit_complete(struct rds_connection *conn)
> +{
> + struct rds_ib_connection *ic = conn->c_transport_data;
> +
> + /* We may have a pending ACK or window update we were unable
> + * to send previously (due to flow control). Try again. */
> + rds_ib_attempt_ack(ic);
> +}
> Index: ofa_kernel-1.3/net/rds/ib_stats.c
> ===================================================================
> --- ofa_kernel-1.3.orig/net/rds/ib_stats.c
> +++ ofa_kernel-1.3/net/rds/ib_stats.c
> @@ -46,14 +46,17 @@ static char *rds_ib_stat_names[] = {
> "ib_tx_cq_call",
> "ib_tx_cq_event",
> "ib_tx_ring_full",
> + "ib_tx_throttle",
> "ib_tx_sg_mapping_failure",
> "ib_tx_stalled",
> + "ib_tx_credit_updates",
> "ib_rx_cq_call",
> "ib_rx_cq_event",
> "ib_rx_ring_empty",
> "ib_rx_refill_from_cq",
> "ib_rx_refill_from_thread",
> "ib_rx_alloc_limit",
> + "ib_rx_credit_updates",
> "ib_ack_sent",
> "ib_ack_send_failure",
> "ib_ack_send_delayed",
> Index: ofa_kernel-1.3/net/rds/rds.h
> ===================================================================
> --- ofa_kernel-1.3.orig/net/rds/rds.h
> +++ ofa_kernel-1.3/net/rds/rds.h
> @@ -170,6 +170,7 @@ struct rds_connection {
> #define RDS_FLAG_CONG_BITMAP 0x01
> #define RDS_FLAG_ACK_REQUIRED 0x02
> #define RDS_FLAG_RETRANSMITTED 0x04
> +#define RDS_MAX_ADV_CREDIT 255
>
> /*
> * Maximum space available for extension headers.
> @@ -183,7 +184,8 @@ struct rds_header {
> __be16 h_sport;
> __be16 h_dport;
> u8 h_flags;
> - u8 h_padding[5];
> + u8 h_credit;
> + u8 h_padding[4];
> __sum16 h_csum;
>
> u8 h_exthdr[RDS_HEADER_EXT_SPACE];
> Index: ofa_kernel-1.3/net/rds/ib_sysctl.c
> ===================================================================
> --- ofa_kernel-1.3.orig/net/rds/ib_sysctl.c
> +++ ofa_kernel-1.3/net/rds/ib_sysctl.c
> @@ -53,6 +53,8 @@ unsigned long rds_ib_sysctl_max_unsig_by
> static unsigned long rds_ib_sysctl_max_unsig_bytes_min = 1;
> static unsigned long rds_ib_sysctl_max_unsig_bytes_max = ~0UL;
>
> +unsigned int rds_ib_sysctl_flow_control = 1;
> +
> ctl_table rds_ib_sysctl_table[] = {
> {
> .ctl_name = 1,
> @@ -102,6 +104,14 @@ ctl_table rds_ib_sysctl_table[] = {
> .mode = 0644,
> .proc_handler = &proc_doulongvec_minmax,
> },
> + {
> + .ctl_name = 6,
> + .procname = "flow_control",
> + .data = &rds_ib_sysctl_flow_control,
> + .maxlen = sizeof(rds_ib_sysctl_flow_control),
> + .mode = 0644,
> + .proc_handler = &proc_dointvec,
> + },
> { .ctl_name = 0}
> };
>
> _______________________________________________
> general mailing list
> general at lists.openfabrics.org
> http://lists.openfabrics.org/cgi-bin/mailman/listinfo/general
>
> To unsubscribe, please visit http://openib.org/mailman/listinfo/openib-general
More information about the general
mailing list