Skip to content

Commit

Permalink
RDS: rewrite rds_ib_xmit
Browse files Browse the repository at this point in the history 
Now that the header always goes first, it is possible to
simplify rds_ib_xmit. Instead of having a path to handle 0-byte
dgrams and another path to handle >0, these can both be handled
in one path. This lets us eliminate xmit_populate_wr().

Rename sent to bytes_sent, to differentiate better from other
variable named "send".

Signed-off-by: Andy Grover <andy.grover@oracle.com>
  • Loading branch information
Andy Grover committed Sep 9, 2010
1 parent 919ced4 commit da5a06c
Showing 1 changed file with 45 additions and 78 deletions.
123 changes: 45 additions & 78 deletions net/rds/ib_send.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -425,38 +425,6 @@ void rds_ib_advertise_credits(struct rds_connection *conn, unsigned int posted)
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,
unsigned long buffer, unsigned int length,
int send_flags)
{
struct ib_sge *sge;

WARN_ON(pos != send - ic->i_sends);

send->s_wr.send_flags = send_flags;
send->s_wr.opcode = IB_WR_SEND;
send->s_wr.num_sge = 1;
send->s_wr.next = NULL;
send->s_queued = jiffies;
send->s_op = NULL;

sge = &send->s_sge[0];
sge->addr = ic->i_send_hdrs_dma + (pos * sizeof(struct rds_header));
sge->length = sizeof(struct rds_header);
sge->lkey = ic->i_mr->lkey;

if (length != 0) {
send->s_wr.num_sge = 2;

sge = &send->s_sge[1];
sge->addr = buffer;
sge->length = length;
sge->lkey = ic->i_mr->lkey;
}
}

/*
* This can be called multiple times for a given message. The first time
* we see a message we map its scatterlist into the IB device so that
Expand All @@ -483,11 +451,11 @@ int rds_ib_xmit(struct rds_connection *conn, struct rds_message *rm,
u32 pos;
u32 i;
u32 work_alloc;
u32 credit_alloc;
u32 credit_alloc = 0;
u32 posted;
u32 adv_credits = 0;
int send_flags = 0;
int sent;
int bytes_sent = 0;
int ret;
int flow_controlled = 0;

Expand Down Expand Up @@ -515,7 +483,6 @@ int rds_ib_xmit(struct rds_connection *conn, struct rds_message *rm,
goto out;
}

credit_alloc = work_alloc;
if (ic->i_flowctl) {
credit_alloc = rds_ib_send_grab_credits(ic, work_alloc, &posted, 0, RDS_MAX_ADV_CREDIT);
adv_credits += posted;
Expand Down Expand Up @@ -591,13 +558,6 @@ int rds_ib_xmit(struct rds_connection *conn, struct rds_message *rm,
BUG_ON(adv_credits > 255);
}

send = &ic->i_sends[pos];
first = send;
prev = NULL;
scat = &rm->data.m_sg[sg];
sent = 0;
i = 0;

/* Sometimes you want to put a fence between an RDMA
* READ and the following SEND.
* We could either do this all the time
Expand All @@ -607,31 +567,45 @@ int rds_ib_xmit(struct rds_connection *conn, struct rds_message *rm,
if (rm->rdma.m_rdma_op.r_active && rm->rdma.m_rdma_op.r_fence)
send_flags = IB_SEND_FENCE;

/*
* We could be copying the header into the unused tail of the page.
* That would need to be changed in the future when those pages might
* be mapped userspace pages or page cache pages. So instead we always
* use a second sge and our long-lived ring of mapped headers. We send
* the header after the data so that the data payload can be aligned on
* the receiver.
*/
/* Each frag gets a header. Msgs may be 0 bytes */
send = &ic->i_sends[pos];
first = send;
prev = NULL;
scat = &rm->data.m_sg[sg];
i = 0;
do {
unsigned int len = 0;

/* handle a 0-len message */
if (be32_to_cpu(rm->m_inc.i_hdr.h_len) == 0) {
rds_ib_xmit_populate_wr(ic, send, pos, 0, 0, send_flags);
goto add_header;
}
/* Set up the header */
send->s_wr.send_flags = send_flags;
send->s_wr.opcode = IB_WR_SEND;
send->s_wr.num_sge = 1;
send->s_wr.next = NULL;
send->s_queued = jiffies;
send->s_op = NULL;

/* if there's data reference it with a chain of work reqs */
for (; i < work_alloc && scat != &rm->data.m_sg[rm->data.m_count]; i++) {
unsigned int len;
send->s_sge[0].addr = ic->i_send_hdrs_dma
+ (pos * sizeof(struct rds_header));
send->s_sge[0].length = sizeof(struct rds_header);

send = &ic->i_sends[pos];
memcpy(&ic->i_send_hdrs[pos], &rm->m_inc.i_hdr, sizeof(struct rds_header));

/* Set up the data, if present */
if (i < work_alloc
&& scat != &rm->data.m_sg[rm->data.m_count]) {
len = min(RDS_FRAG_SIZE, ib_sg_dma_len(dev, scat) - off);
send->s_wr.num_sge = 2;

len = min(RDS_FRAG_SIZE, ib_sg_dma_len(dev, scat) - off);
rds_ib_xmit_populate_wr(ic, send, pos,
ib_sg_dma_address(dev, scat) + off, len,
send_flags);
send->s_sge[1].addr = ib_sg_dma_address(dev, scat) + off;
send->s_sge[1].length = len;

bytes_sent += len;
off += len;
if (off == ib_sg_dma_len(dev, scat)) {
scat++;
off = 0;
}
}

/*
* We want to delay signaling completions just enough to get
Expand All @@ -658,18 +632,6 @@ int rds_ib_xmit(struct rds_connection *conn, struct rds_message *rm,
rdsdebug("send %p wr %p num_sge %u next %p\n", send,
&send->s_wr, send->s_wr.num_sge, send->s_wr.next);

sent += len;
off += len;
if (off == ib_sg_dma_len(dev, scat)) {
scat++;
off = 0;
}

add_header:
/* Tack on the header after the data. The header SGE should already
* 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 (adv_credits) {
struct rds_header *hdr = &ic->i_send_hdrs[pos];

Expand All @@ -685,12 +647,16 @@ int rds_ib_xmit(struct rds_connection *conn, struct rds_message *rm,
prev = send;

pos = (pos + 1) % ic->i_send_ring.w_nr;
}
send = &ic->i_sends[pos];
i++;

} while (i < work_alloc
&& scat != &rm->data.m_sg[rm->data.m_count]);

/* Account the RDS header in the number of bytes we sent, but just once.
* The caller has no concept of fragmentation. */
if (hdr_off == 0)
sent += sizeof(struct rds_header);
bytes_sent += sizeof(struct rds_header);

/* if we finished the message then send completion owns it */
if (scat == &rm->data.m_sg[rm->data.m_count]) {
Expand All @@ -699,6 +665,7 @@ int rds_ib_xmit(struct rds_connection *conn, struct rds_message *rm,
ic->i_rm = NULL;
}

/* Put back wrs & credits we didn't use */
if (i < work_alloc) {
rds_ib_ring_unalloc(&ic->i_send_ring, work_alloc - i);
work_alloc = i;
Expand All @@ -725,7 +692,7 @@ int rds_ib_xmit(struct rds_connection *conn, struct rds_message *rm,
goto out;
}

ret = sent;
ret = bytes_sent;
out:
BUG_ON(adv_credits);
return ret;
Expand Down

0 comments on commit da5a06c

Please sign in to comment.