Skip to content

Commit

Permalink
RDS/IB+IW: Move recv processing to a tasklet
Browse files Browse the repository at this point in the history 
Move receive processing from event handler to a tasklet.
This should help prevent hangcheck timer from going off
when RDS is under heavy load.

Signed-off-by: Andy Grover <[email protected]>
Signed-off-by: David S. Miller <[email protected]>
  • Loading branch information
@davem330
Andy Grover authored and davem330 committed Oct 30, 2009
1 parent 0514f8a commit d521b63
Show file tree
Hide file tree
Showing 6 changed files with 52 additions and 12 deletions.
2 changes: 2 additions & 0 deletions net/rds/ib.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,7 @@ struct rds_ib_connection {
struct rds_ib_send_work *i_sends;

/* rx */
struct tasklet_struct i_recv_tasklet;
struct mutex i_recv_mutex;
struct rds_ib_work_ring i_recv_ring;
struct rds_ib_incoming *i_ibinc;
Expand Down Expand Up @@ -303,6 +304,7 @@ 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);
void rds_ib_recv_cq_comp_handler(struct ib_cq *cq, void *context);
void rds_ib_recv_tasklet_fn(unsigned long data);
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);
Expand Down
2 changes: 2 additions & 0 deletions net/rds/ib_cm.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -694,6 +694,8 @@ int rds_ib_conn_alloc(struct rds_connection *conn, gfp_t gfp)
return -ENOMEM;

INIT_LIST_HEAD(&ic->ib_node);
tasklet_init(&ic->i_recv_tasklet, rds_ib_recv_tasklet_fn,
(unsigned long) ic);
mutex_init(&ic->i_recv_mutex);
#ifndef KERNEL_HAS_ATOMIC64
spin_lock_init(&ic->i_ack_lock);
Expand Down
28 changes: 22 additions & 6 deletions net/rds/ib_recv.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -825,17 +825,22 @@ void rds_ib_recv_cq_comp_handler(struct ib_cq *cq, void *context)
{
struct rds_connection *conn = context;
struct rds_ib_connection *ic = conn->c_transport_data;
struct ib_wc wc;
struct rds_ib_ack_state state = { 0, };
struct rds_ib_recv_work *recv;

rdsdebug("conn %p cq %p\n", conn, cq);

rds_ib_stats_inc(s_ib_rx_cq_call);

ib_req_notify_cq(cq, IB_CQ_SOLICITED);
tasklet_schedule(&ic->i_recv_tasklet);
}

while (ib_poll_cq(cq, 1, &wc) > 0) {
static inline void rds_poll_cq(struct rds_ib_connection *ic,
struct rds_ib_ack_state *state)
{
struct rds_connection *conn = ic->conn;
struct ib_wc wc;
struct rds_ib_recv_work *recv;

while (ib_poll_cq(ic->i_recv_cq, 1, &wc) > 0) {
rdsdebug("wc wr_id 0x%llx status %u byte_len %u imm_data %u\n",
(unsigned long long)wc.wr_id, wc.status, wc.byte_len,
be32_to_cpu(wc.ex.imm_data));
Expand All @@ -853,7 +858,7 @@ void rds_ib_recv_cq_comp_handler(struct ib_cq *cq, void *context)
if (rds_conn_up(conn) || rds_conn_connecting(conn)) {
/* We expect errors as the qp is drained during shutdown */
if (wc.status == IB_WC_SUCCESS) {
rds_ib_process_recv(conn, recv, wc.byte_len, &state);
rds_ib_process_recv(conn, recv, wc.byte_len, state);
} else {
rds_ib_conn_error(conn, "recv completion on "
"%pI4 had status %u, disconnecting and "
Expand All @@ -864,6 +869,17 @@ void rds_ib_recv_cq_comp_handler(struct ib_cq *cq, void *context)

rds_ib_ring_free(&ic->i_recv_ring, 1);
}
}

void rds_ib_recv_tasklet_fn(unsigned long data)
{
struct rds_ib_connection *ic = (struct rds_ib_connection *) data;
struct rds_connection *conn = ic->conn;
struct rds_ib_ack_state state = { 0, };

rds_poll_cq(ic, &state);
ib_req_notify_cq(ic->i_recv_cq, IB_CQ_SOLICITED);
rds_poll_cq(ic, &state);

if (state.ack_next_valid)
rds_ib_set_ack(ic, state.ack_next, state.ack_required);
Expand Down
2 changes: 2 additions & 0 deletions net/rds/iw.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,7 @@ struct rds_iw_connection {
struct rds_iw_send_work *i_sends;

/* rx */
struct tasklet_struct i_recv_tasklet;
struct mutex i_recv_mutex;
struct rds_iw_work_ring i_recv_ring;
struct rds_iw_incoming *i_iwinc;
Expand Down Expand Up @@ -330,6 +331,7 @@ void rds_iw_inc_free(struct rds_incoming *inc);
int rds_iw_inc_copy_to_user(struct rds_incoming *inc, struct iovec *iov,
size_t size);
void rds_iw_recv_cq_comp_handler(struct ib_cq *cq, void *context);
void rds_iw_recv_tasklet_fn(unsigned long data);
void rds_iw_recv_init_ring(struct rds_iw_connection *ic);
void rds_iw_recv_clear_ring(struct rds_iw_connection *ic);
void rds_iw_recv_init_ack(struct rds_iw_connection *ic);
Expand Down
2 changes: 2 additions & 0 deletions net/rds/iw_cm.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -696,6 +696,8 @@ int rds_iw_conn_alloc(struct rds_connection *conn, gfp_t gfp)
return -ENOMEM;

INIT_LIST_HEAD(&ic->iw_node);
tasklet_init(&ic->i_recv_tasklet, rds_iw_recv_tasklet_fn,
(unsigned long) ic);
mutex_init(&ic->i_recv_mutex);
#ifndef KERNEL_HAS_ATOMIC64
spin_lock_init(&ic->i_ack_lock);
Expand Down
28 changes: 22 additions & 6 deletions net/rds/iw_recv.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -784,17 +784,22 @@ void rds_iw_recv_cq_comp_handler(struct ib_cq *cq, void *context)
{
struct rds_connection *conn = context;
struct rds_iw_connection *ic = conn->c_transport_data;
struct ib_wc wc;
struct rds_iw_ack_state state = { 0, };
struct rds_iw_recv_work *recv;

rdsdebug("conn %p cq %p\n", conn, cq);

rds_iw_stats_inc(s_iw_rx_cq_call);

ib_req_notify_cq(cq, IB_CQ_SOLICITED);
tasklet_schedule(&ic->i_recv_tasklet);
}

while (ib_poll_cq(cq, 1, &wc) > 0) {
static inline void rds_poll_cq(struct rds_iw_connection *ic,
struct rds_iw_ack_state *state)
{
struct rds_connection *conn = ic->conn;
struct ib_wc wc;
struct rds_iw_recv_work *recv;

while (ib_poll_cq(ic->i_recv_cq, 1, &wc) > 0) {
rdsdebug("wc wr_id 0x%llx status %u byte_len %u imm_data %u\n",
(unsigned long long)wc.wr_id, wc.status, wc.byte_len,
be32_to_cpu(wc.ex.imm_data));
Expand All @@ -812,7 +817,7 @@ void rds_iw_recv_cq_comp_handler(struct ib_cq *cq, void *context)
if (rds_conn_up(conn) || rds_conn_connecting(conn)) {
/* We expect errors as the qp is drained during shutdown */
if (wc.status == IB_WC_SUCCESS) {
rds_iw_process_recv(conn, recv, wc.byte_len, &state);
rds_iw_process_recv(conn, recv, wc.byte_len, state);
} else {
rds_iw_conn_error(conn, "recv completion on "
"%pI4 had status %u, disconnecting and "
Expand All @@ -823,6 +828,17 @@ void rds_iw_recv_cq_comp_handler(struct ib_cq *cq, void *context)

rds_iw_ring_free(&ic->i_recv_ring, 1);
}
}

void rds_iw_recv_tasklet_fn(unsigned long data)
{
struct rds_iw_connection *ic = (struct rds_iw_connection *) data;
struct rds_connection *conn = ic->conn;
struct rds_iw_ack_state state = { 0, };

rds_poll_cq(ic, &state);
ib_req_notify_cq(ic->i_recv_cq, IB_CQ_SOLICITED);
rds_poll_cq(ic, &state);

if (state.ack_next_valid)
rds_iw_set_ack(ic, state.ack_next, state.ack_required);
Expand Down

0 comments on commit d521b63

Please sign in to comment.